root/software/read_test/src/URingExecDriver.cc @ ef18eb33
| ef18eb33 | 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 2022 BlackLynx Inc. *
|
|||
* Unpublished -- all rights reserved under the copyright laws *
|
|||
* of the United States. *
|
|||
********************************************************************/
|
|||
#include <sys/types.h>
|
|||
#include <sys/stat.h>
|
|||
#include <fcntl.h>
|
|||
#include "URingExecDriver.h"
|
|||
#include "Utilities/ChunkConstants.h"
|
|||
#include "Utilities/FileUtils.h"
|
|||
#include "Utilities/SearchLynxConfiguration.h"
|
|||
#include "Utilities/BufferAllocator.h"
|
|||
BlackLynx::SearchLynx::URingExecDriver::URingExecDriver(
|
|||
const std::string& inputFile,
|
|||
uint32_t numReaderThreads)
|
|||
: ExecDriver(inputFile, numReaderThreads)
|
|||
{
|
|||
}
|
|||
BlackLynx::SearchLynx::URingExecDriver::~URingExecDriver()
|
|||
{
|
|||
}
|
|||
void BlackLynx::SearchLynx::URingExecDriver::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::URingExecDriver::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::URingExecDriver::readerThreadBody(
|
|||
chunk_queue_t* inQueue,
|
|||
chunk_queue_t* outQueue)
|
|||
{
|
|||
BlnxLog(error) << "URING READER";
|
|||
const uint32_t QUEUE_DEPTH = 32;
|
|||
const uint32_t IO_SIZE = 4194304;
|
|||
const uint32_t MIN_SIZE_ALIGN = 512;
|
|||
// Setup IO context
|
|||
// Open the file
|
|||
int fd = -1;
|
|||
// int fd = ::open(r_inputFile.c_str(), O_RDONLY | O_DIRECT);
|
|||
// if (fd < 0)
|
|||
// {
|
|||
// BlnxLog(error) << "Unable to open file: " << r_inputFile;
|
|||
// return;
|
|||
// }
|
|||
Chunk* chunk = nullptr;
|
|||
std::string prevFilename("");
|
|||
// std::ofstream output("chunk.bin");
|
|||
while (true)
|
|||
{
|
|||
chunk = inQueue->pop_front();
|
|||
// Check for terminator
|
|||
if (chunk == nullptr)
|
|||
{
|
|||
break;
|
|||
}
|
|||
if (prevFilename != chunk->m_filename)
|
|||
{
|
|||
// Open the file
|
|||
fd = ::open(r_inputFile.c_str(), O_RDONLY | O_DIRECT);
|
|||
if (fd < 0)
|
|||
{
|
|||
BlnxLog(error) << "Unable to open file: " << r_inputFile;
|
|||
return;
|
|||
}
|
|||
}
|
|||
// Grab a buffer to hold the chunk data
|
|||
chunk->allocateBuffer();
|
|||
// Determine how many IOs we need
|
|||
// std::string test(chunk->p_chunkData, 30);
|
|||
// BlnxLog(debug) << "CHUNK " << chunk->m_chunkId << " DATA: " << test;
|
|||
// output.write(chunk->p_chunkData, chunk->m_chunkSize);
|
|||
outQueue->push_back(chunk);
|
|||
}
|
|||
// output.close();
|
|||
// Release file descriptor
|
|||
::close(fd);
|
|||
// Destroy the IO context
|
|||
io_destroy(context);
|
|||
// Send terminator to clean up thread
|
|||
m_cleanupQueue.push_back(nullptr);
|
|||
}
|