|
/********************************************************************
|
|
* This software is "commercial computer software" as defined in the *
|
|
* Federal Acquisition Regulations and is subject to BlackLynx *
|
|
* Inc's standard End User License Agreement. *
|
|
* *
|
|
* CONFIDENTIAL - All source code is the propriety and confidential *
|
|
* information of BlackLynx Inc. *
|
|
* *
|
|
* Copyright 2014-2020 BlackLynx Inc. *
|
|
* Unpublished -- all rights reserved under the copyright laws *
|
|
* of the United States. *
|
|
********************************************************************/
|
|
#include <iomanip>
|
|
|
|
#include <sys/stat.h>
|
|
#include <sys/types.h>
|
|
|
|
#include "ExecDriver.h"
|
|
#include "Utilities/ChunkConstants.h"
|
|
#include "Utilities/FileUtils.h"
|
|
#include "Utilities/BufferAllocator.h"
|
|
|
|
|
|
BlackLynx::SearchLynx::ExecDriver::ExecDriver(
|
|
const std::string& inputFile,
|
|
uint32_t numReaderThreads,
|
|
uint32_t chunkSize)
|
|
: r_inputFile(inputFile),
|
|
m_numReaderThreads(numReaderThreads),
|
|
DEFAULT_CHUNK_SIZE(chunkSize * 1024),
|
|
EXTRA_BYTES_TO_READ(RAW_TEXT_EXTRA_BYTES)
|
|
{
|
|
}
|
|
|
|
BlackLynx::SearchLynx::ExecDriver::~ExecDriver()
|
|
{
|
|
}
|
|
|
|
void BlackLynx::SearchLynx::ExecDriver::start()
|
|
{
|
|
// Start threads in reverse order staring with cleanup thread
|
|
p_cleanupThread = new std::thread(&BlackLynx::SearchLynx::ExecDriver::cleanupThreadBody,
|
|
this,
|
|
&m_cleanupQueue);
|
|
|
|
// Start specified number of reader threads
|
|
for (uint32_t idx = 0; idx < m_numReaderThreads; ++idx)
|
|
{
|
|
std::thread* thisThread =
|
|
new std::thread(&BlackLynx::SearchLynx::ExecDriver::readerThreadBody,
|
|
this,
|
|
&m_readerQueue,
|
|
&m_cleanupQueue);
|
|
m_readerThreads.push_back(thisThread);
|
|
}
|
|
|
|
// Start setup thread body
|
|
p_setupThread = new std::thread(&BlackLynx::SearchLynx::ExecDriver::setupThreadBody,
|
|
this,
|
|
&m_readerQueue,
|
|
m_numReaderThreads);
|
|
}
|
|
|
|
void BlackLynx::SearchLynx::ExecDriver::waitUntilComplete()
|
|
{
|
|
// Join threads in normal order
|
|
|
|
// Join setup thread
|
|
if (p_setupThread != nullptr)
|
|
{
|
|
if (p_setupThread->joinable())
|
|
{
|
|
p_setupThread->join();
|
|
}
|
|
}
|
|
delete p_setupThread;
|
|
p_setupThread = nullptr;
|
|
|
|
// Join reader threads
|
|
for (auto readerThread : m_readerThreads)
|
|
{
|
|
if (readerThread->joinable())
|
|
{
|
|
readerThread->join();
|
|
}
|
|
delete readerThread;
|
|
}
|
|
m_readerThreads.clear();
|
|
|
|
// Join cleanup thread
|
|
if (p_cleanupThread != nullptr)
|
|
{
|
|
if (p_cleanupThread->joinable())
|
|
{
|
|
p_cleanupThread->join();
|
|
}
|
|
}
|
|
delete p_cleanupThread;
|
|
p_cleanupThread = nullptr;
|
|
}
|
|
|
|
void BlackLynx::SearchLynx::ExecDriver::setupThreadBody(
|
|
chunk_queue_t* outQueue,
|
|
uint32_t numTerminators)
|
|
{
|
|
uint64_t chunkIdAssigner = 1;
|
|
uint64_t fileOffset = 0;
|
|
uint32_t chunkSize = 0;
|
|
uint32_t extraSize = 0;
|
|
uint64_t fileSize = FileUtils::getFileSize(r_inputFile);
|
|
|
|
while (fileOffset < fileSize)
|
|
{
|
|
// Determine chunk size
|
|
if ((fileSize - fileOffset) >= DEFAULT_CHUNK_SIZE)
|
|
{
|
|
chunkSize = DEFAULT_CHUNK_SIZE;
|
|
}
|
|
else
|
|
{
|
|
chunkSize = fileSize - fileOffset;
|
|
}
|
|
|
|
// Determine extra size
|
|
if ((fileOffset + chunkSize + EXTRA_BYTES_TO_READ) <= fileSize)
|
|
{
|
|
extraSize = EXTRA_BYTES_TO_READ;
|
|
}
|
|
else
|
|
{
|
|
extraSize = fileSize - (fileOffset + chunkSize);
|
|
}
|
|
|
|
// bool firstChunkInFile = (fileOffset == 0);
|
|
// bool lastChunkInFile = ((fileOffset + chunkSize) == fileSize);
|
|
|
|
// Make chunk to process
|
|
auto chunk = new Chunk(r_inputFile, chunkIdAssigner, chunkSize,
|
|
extraSize, fileOffset);
|
|
|
|
// Increment the chunk ID by 2 so that we match the record version
|
|
// of the execution driver
|
|
chunkIdAssigner += 2;
|
|
|
|
fileOffset += chunkSize;
|
|
|
|
// Send the chunk to the reader
|
|
outQueue->push_back(chunk);
|
|
}
|
|
|
|
// Send the specified number of terminators
|
|
for (uint32_t idx = 0; idx < numTerminators; ++idx)
|
|
{
|
|
outQueue->push_back(nullptr);
|
|
}
|
|
}
|
|
|
|
void BlackLynx::SearchLynx::ExecDriver::readerThreadBody(
|
|
chunk_queue_t* inQueue,
|
|
chunk_queue_t* outQueue)
|
|
{
|
|
Chunk* chunk = nullptr;
|
|
|
|
std::ifstream inFileStream;
|
|
std::string prevFilename("");
|
|
|
|
while (true)
|
|
{
|
|
chunk = inQueue->pop_front();
|
|
|
|
// Check for terminator
|
|
if (chunk == nullptr)
|
|
{
|
|
break;
|
|
}
|
|
|
|
// Open file for reading if the previous filename is different than the
|
|
// current filename
|
|
if (chunk->m_filename != prevFilename)
|
|
{
|
|
if (inFileStream.is_open())
|
|
{
|
|
inFileStream.close();
|
|
}
|
|
|
|
inFileStream.open(chunk->m_filename, std::ios::in | std::ios::binary);
|
|
if (! inFileStream)
|
|
{
|
|
BlnxLog(error) << "Unable to open file: " << chunk->m_filename;
|
|
break;
|
|
}
|
|
|
|
// Tell the kernel this file will be read sequentially
|
|
int rc = posix_fadvise(FileUtils::getFdHack(inFileStream), 0, 0,
|
|
POSIX_FADV_SEQUENTIAL);
|
|
if (rc)
|
|
{
|
|
BlnxLog(error) << "posix_fadvise() returned: " << rc;
|
|
}
|
|
}
|
|
|
|
// Store filename to compare against next chunk
|
|
prevFilename = chunk->m_filename;
|
|
|
|
|
|
BlnxLog(debug) << "Reader Thread reading:"
|
|
<< " Filename: " << chunk->m_filename
|
|
<< " ChunkID: " << chunk->m_chunkId
|
|
<< " Chunk Size: " << chunk->m_chunkSize
|
|
<< " Extra Size: " << chunk->m_extraSize
|
|
<< " File Offset: " << chunk->m_fileOffset;
|
|
|
|
// Grab a buffer to hold the chunk data
|
|
chunk->allocateBuffer();
|
|
|
|
inFileStream.seekg(chunk->m_fileOffset);
|
|
inFileStream.read(chunk->p_chunkData,
|
|
chunk->m_chunkSize + chunk->m_extraSize);
|
|
|
|
if (inFileStream.gcount() != (chunk->m_chunkSize + chunk->m_extraSize))
|
|
{
|
|
// Log a runtime error
|
|
std::ostringstream msg;
|
|
msg << "Unable to read "
|
|
<< (chunk->m_chunkSize + chunk->m_extraSize)
|
|
<< " bytes at offset " << chunk->m_fileOffset << " from "
|
|
<< chunk->m_filename << " (chunk ID: " << chunk->m_chunkId
|
|
<< ")";
|
|
BlnxLog(error) << msg.str();
|
|
continue;
|
|
}
|
|
|
|
outQueue->push_back(chunk);
|
|
}
|
|
|
|
inFileStream.close();
|
|
|
|
// Send terminator to clean up thread. Note the cleanup thread counts the
|
|
// number of terminators and does not terminate until it receives all
|
|
// expected terminators.
|
|
m_cleanupQueue.push_back(nullptr);
|
|
}
|
|
|
|
void BlackLynx::SearchLynx::ExecDriver::cleanupThreadBody(
|
|
chunk_queue_t* inQueue)
|
|
{
|
|
uint32_t terminatorCount = 0;
|
|
Chunk* chunk = nullptr;
|
|
auto bufAllocator = BufferAllocator::getInstance();
|
|
|
|
while (true)
|
|
{
|
|
chunk = inQueue->pop_front();
|
|
|
|
// Check for terminator
|
|
if (chunk == nullptr)
|
|
{
|
|
// Check if we've received one terminator from each reader
|
|
if (++terminatorCount == m_numReaderThreads)
|
|
{
|
|
break;
|
|
}
|
|
else
|
|
{
|
|
continue;
|
|
}
|
|
}
|
|
|
|
// Uncomment this to output each chunk's buffer to a file
|
|
#if 0
|
|
// Create directory to hold chunk files
|
|
mkdir("chunks", 0777);
|
|
|
|
// Output each chunk with a filename corresponding to its chunk ID.
|
|
// E.g. chunk_235.bin
|
|
std::ostringstream fname;
|
|
fname << "chunks/chunk_" << std::setw(3) << std::setfill('0')
|
|
<< chunk->m_chunkId << ".bin";
|
|
BlnxLog(error) << "FNAME: " << fname.str();
|
|
std::ofstream output(fname.str());
|
|
output.write(chunk->p_chunkData, chunk->m_chunkSize);
|
|
output.close();
|
|
#endif
|
|
|
|
bufAllocator->release(chunk->p_chunkData);
|
|
delete chunk;
|
|
}
|
|
}
|