Project

General

Profile

Download (8.21 KB) Statistics
| Branch: | Tag: | Revision:

root/software/read_test/src/ExecDriver.cc @ a5734307

a0d1d164 David Sorber
/********************************************************************
* 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 "ExecDriver.h"
#include "Utilities/ChunkConstants.h"
#include "Utilities/FileUtils.h"
#include "Utilities/SearchLynxConfiguration.h"
#include "Utilities/BufferAllocator.h"

BlackLynx::SearchLynx::ExecDriver::ExecDriver(
const std::string& inputFile,
uint32_t numReaderThreads)
: r_inputFile(inputFile),
m_numReaderThreads(numReaderThreads),
DEFAULT_CHUNK_SIZE(SearchLynxConfiguration::getInstance()->getChunkSizeInKb() * 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;
}
}
bufAllocator->release(chunk->p_chunkData);
delete chunk;
}
}