root/software/read_test/src/URingExecDriver.cc @ 1136d406
| 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. *
|
||||
********************************************************************/
|
||||
| 864b608c | David Sorber | #include <cerrno>
|
||
#include <sstream>
|
||||
#include <stdexcept>
|
||||
| ef18eb33 | David Sorber | |||
#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)
|
||||
| 864b608c | David Sorber | : ExecDriver(inputFile, numReaderThreads),
|
||
m_ioCounter(0)
|
||||
| ef18eb33 | David Sorber | {
|
||
| 864b608c | David Sorber | // Init the ring
|
||
int rc = io_uring_queue_init(QUEUE_DEPTH, &m_ring, 0);
|
||||
if (rc < 0)
|
||||
{
|
||||
std::ostringstream msg;
|
||||
msg << "FATAL - io_uring_queue_init returned: " << rc;
|
||||
throw std::runtime_error(msg.str());
|
||||
}
|
||||
// BlnxLog(info) << "RING SPACE: " << io_uring_sq_space_left(&m_ring);
|
||||
| ef18eb33 | David Sorber | }
|
||
BlackLynx::SearchLynx::URingExecDriver::~URingExecDriver()
|
||||
{
|
||||
| 864b608c | David Sorber | // Clean up the ring
|
||
io_uring_queue_exit(&m_ring);
|
||||
| ef18eb33 | David Sorber | }
|
||
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)
|
||||
{
|
||||
| 864b608c | David Sorber | BlnxLog(info) << "URING READER starting";
|
||
| ef18eb33 | David Sorber | |||
const uint32_t MIN_SIZE_ALIGN = 512;
|
||||
// Setup IO context
|
||||
int fd = -1;
|
||||
| 864b608c | David Sorber | struct io_uring_sqe* sqe = nullptr;
|
||
| ef18eb33 | David Sorber | Chunk* chunk = nullptr;
|
||
std::string prevFilename("");
|
||||
| 864b608c | David Sorber | int rc = 0;
|
||
| ef18eb33 | David Sorber | |||
// 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();
|
||||
| 864b608c | David Sorber | // Wait until a submission queue entry is ready
|
||
sqe = nullptr;
|
||||
while (sqe == nullptr)
|
||||
{
|
||||
sqe = io_uring_get_sqe(&m_ring);
|
||||
if (! sqe)
|
||||
{
|
||||
waitForCompletion(outQueue);
|
||||
}
|
||||
}
|
||||
uint32_t totalSize = chunk->m_chunkSize + chunk->m_extraSize;
|
||||
uint32_t leftOverSize = (totalSize & (MIN_SIZE_ALIGN - 1));
|
||||
uint32_t readSize = totalSize - leftOverSize;
|
||||
| ef18eb33 | David Sorber | |||
| 864b608c | David Sorber | // Prep the read request and set the chunk object as its associated
|
||
// data
|
||||
io_uring_prep_read(sqe, fd, chunk->p_chunkData, readSize,
|
||||
chunk->m_fileOffset);
|
||||
io_uring_sqe_set_data(sqe, reinterpret_cast<void*>(chunk));
|
||||
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
|
||||
<< " IOPs in flight: " << m_ioCounter
|
||||
<< " SPACE LEFT: " << io_uring_sq_space_left(&m_ring)
|
||||
<< " READY: " << io_uring_cq_ready(&m_ring);
|
||||
// Update the counter and submit the request
|
||||
++m_ioCounter;
|
||||
rc = io_uring_submit(&m_ring);
|
||||
if (rc < 0)
|
||||
{
|
||||
BlnxLog(error) << "io_uring_submit returned: " << rc;
|
||||
}
|
||||
// NOTE: when using O_DIRECT for high performance the read size *must*
|
||||
// be 512 byte aligned. Manually read any unaligned bytes, which should
|
||||
// only happen for the last "tail" chunk.
|
||||
if (leftOverSize > 0)
|
||||
{
|
||||
BlnxLog(info) << "CHUNK: " << chunk->m_chunkId << " -- "
|
||||
<< leftOverSize << " unaligned bytes to read";
|
||||
std::ifstream inFileStream(chunk->m_filename, std::ios::in | std::ios::binary);
|
||||
if (! inFileStream)
|
||||
{
|
||||
BlnxLog(error) << "Unable to open file: " << chunk->m_filename;
|
||||
break;
|
||||
}
|
||||
inFileStream.seekg(chunk->m_fileOffset + (totalSize - leftOverSize));
|
||||
inFileStream.read(chunk->p_chunkData + (totalSize - leftOverSize),
|
||||
leftOverSize);
|
||||
inFileStream.close();
|
||||
}
|
||||
if (m_ioCounter >= QUEUE_DEPTH)
|
||||
{
|
||||
waitForCompletion(outQueue);
|
||||
}
|
||||
}
|
||||
// Wait for any lingering IO completions
|
||||
while (m_ioCounter > 0)
|
||||
{
|
||||
waitForCompletion(outQueue);
|
||||
| ef18eb33 | David Sorber | }
|
||
// output.close();
|
||||
// Release file descriptor
|
||||
::close(fd);
|
||||
// Send terminator to clean up thread
|
||||
m_cleanupQueue.push_back(nullptr);
|
||||
| 864b608c | David Sorber | }
|
||
void BlackLynx::SearchLynx::URingExecDriver::waitForCompletion(
|
||||
chunk_queue_t* outQueue)
|
||||
{
|
||||
int rc = 0;
|
||||
struct io_uring_cqe *cqe = nullptr;
|
||||
bool gotOne = false;
|
||||
Chunk* chunk = nullptr;
|
||||
while (m_ioCounter > 0)
|
||||
{
|
||||
// Wait for at least one completion but handle any that are ready
|
||||
if (! gotOne)
|
||||
{
|
||||
rc = io_uring_wait_cqe(&m_ring, &cqe);
|
||||
gotOne = true;
|
||||
}
|
||||
else
|
||||
{
|
||||
rc = io_uring_peek_cqe(&m_ring, &cqe);
|
||||
if (rc == -EAGAIN)
|
||||
{
|
||||
cqe = nullptr;
|
||||
rc = 0;
|
||||
}
|
||||
}
|
||||
// Break out if no cqe was present
|
||||
if (cqe == nullptr)
|
||||
{
|
||||
break;
|
||||
}
|
||||
// Grab the data (chunk) associated with this request and then pass it
|
||||
// along
|
||||
chunk = reinterpret_cast<Chunk*>(io_uring_cqe_get_data(cqe));
|
||||
if (cqe->res < 0)
|
||||
{
|
||||
BlnxLog(error) << "Chunk ID: " << chunk->m_chunkId
|
||||
<< " -- completion returned: " << cqe->res;
|
||||
}
|
||||
BlnxLog(debug) << "Got completion for chunk ID: " << chunk->m_chunkId;
|
||||
outQueue->push_back(chunk);
|
||||
--m_ioCounter;
|
||||
io_uring_cqe_seen(&m_ring, cqe);
|
||||
}
|
||||
| ef18eb33 | David Sorber | }
|