|
/********************************************************************
|
|
* 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);
|
|
}
|