Project

General

Profile

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

root/software/read_test/src/URingExecDriver.cc @ 70882817

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/BufferAllocator.h"

BlackLynx::SearchLynx::URingExecDriver::URingExecDriver(
const std::string& inputFile,
f12339cf david.sorber
uint32_t numReaderThreads,
uint32_t chunkSize)
3dda5d70 david.sorber
: ExecDriver(inputFile, numReaderThreads, chunkSize)
ef18eb33 David Sorber
{
864b608c David Sorber
// BlnxLog(info) << "RING SPACE: " << io_uring_sq_space_left(&m_ring);
3dda5d70 david.sorber
ef18eb33 David Sorber
}
BlackLynx::SearchLynx::URingExecDriver::~URingExecDriver()
{
}

void BlackLynx::SearchLynx::URingExecDriver::start()
{
// Start threads in reverse order staring with cleanup thread
3dda5d70 david.sorber
p_cleanupThread = new std::thread(&ExecDriver::cleanupThreadBody,
ef18eb33 David Sorber
this,
&m_cleanupQueue);
// Start specified number of reader threads
for (uint32_t idx = 0; idx < m_numReaderThreads; ++idx)
{
std::thread* thisThread =
3dda5d70 david.sorber
new std::thread(&URingExecDriver::readerThreadBody,
ef18eb33 David Sorber
this,
&m_readerQueue,
&m_cleanupQueue);
m_readerThreads.push_back(thisThread);
}
// Start setup thread body
3dda5d70 david.sorber
p_setupThread = new std::thread(&ExecDriver::setupThreadBody,
ef18eb33 David Sorber
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
3dda5d70 david.sorber
struct io_uring m_ring;
70882817 David Sorber
std::atomic<uint32_t> m_ioCounter = 0;
ef18eb33 David Sorber
const uint32_t MIN_SIZE_ALIGN = 512;
3dda5d70 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());
}
ef18eb33 David Sorber
int fd = -1;
864b608c David Sorber
struct io_uring_sqe* sqe = nullptr;
3dda5d70 david.sorber
struct io_uring_cqe *cqe = nullptr;
ef18eb33 David Sorber
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);
f12339cf david.sorber
// fd = ::open(r_inputFile.c_str(), O_RDONLY);
ef18eb33 David Sorber
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)
{
3dda5d70 david.sorber
waitForCompletion(&m_ring, m_ioCounter, outQueue);
864b608c David Sorber
}
}
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:"
3dda5d70 david.sorber
// << " Filename: " << chunk->m_filename
864b608c David Sorber
<< " ChunkID: " << chunk->m_chunkId
<< " Chunk Size: " << chunk->m_chunkSize
<< " Extra Size: " << chunk->m_extraSize
<< " File Offset: " << chunk->m_fileOffset
<< " IOPs in flight: " << m_ioCounter
3dda5d70 david.sorber
// << " SPACE LEFT: " << io_uring_sq_space_left(&m_ring)
// << " READY: " << io_uring_cq_ready(&m_ring)
;
864b608c David Sorber
// 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;
3dda5d70 david.sorber
--m_ioCounter;
}
// Non-blocking reap
while (true)
{
rc = io_uring_peek_cqe(&m_ring, &cqe);
if (rc == 0)
{
if (cqe == nullptr)
{
break;
}

// Grab the data (chunk) associated with this request and then pass
// it along
70882817 David Sorber
Chunk* compChunk = reinterpret_cast<Chunk*>(io_uring_cqe_get_data(cqe));
3dda5d70 david.sorber
if (cqe->res < 0)
{
BlnxLog(error) << "Chunk ID: " << chunk->m_chunkId
<< " -- completion returned: " << cqe->res;
}
else
{
BlnxLog(debug) << "Got completion for chunk ID: " << chunk->m_chunkId;
outQueue->push_back(compChunk);
}
70882817 David Sorber
--m_ioCounter;
3dda5d70 david.sorber
io_uring_cqe_seen(&m_ring, cqe);
}
else
{
break;
}
864b608c David Sorber
}
// 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();
}
}
// Wait for any lingering IO completions
while (m_ioCounter > 0)
{
3dda5d70 david.sorber
waitForCompletion(&m_ring, m_ioCounter, outQueue);
ef18eb33 David Sorber
}
// output.close();
// Release file descriptor
::close(fd);
// Send terminator to clean up thread
m_cleanupQueue.push_back(nullptr);
3dda5d70 david.sorber
// Clean up the ring
io_uring_queue_exit(&m_ring);
864b608c David Sorber
}

void BlackLynx::SearchLynx::URingExecDriver::waitForCompletion(
3dda5d70 david.sorber
struct io_uring* ring,
std::atomic<uint32_t>& ioCounter,
864b608c David Sorber
chunk_queue_t* outQueue)
{
int rc = 0;
struct io_uring_cqe *cqe = nullptr;
bool gotOne = false;
Chunk* chunk = nullptr;
3dda5d70 david.sorber
while (ioCounter > 0)
864b608c David Sorber
{
// Wait for at least one completion but handle any that are ready
if (! gotOne)
{
3dda5d70 david.sorber
rc = io_uring_wait_cqe(ring, &cqe);
864b608c David Sorber
gotOne = true;
}
else
{
3dda5d70 david.sorber
rc = io_uring_peek_cqe(ring, &cqe);
864b608c David Sorber
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);
3dda5d70 david.sorber
--ioCounter;
io_uring_cqe_seen(ring, cqe);
864b608c David Sorber
}
70882817 David Sorber
}