Project

General

Profile

« Previous | Next » 

Revision 3dda5d70

Added by david.sorber about 3 years ago

Additional read_test uring adjustments.

View differences:

software/read_test/src/URingExecDriver.cc
const std::string& inputFile,
uint32_t numReaderThreads,
uint32_t chunkSize)
: ExecDriver(inputFile, numReaderThreads, chunkSize),
m_ioCounter(0)
: ExecDriver(inputFile, numReaderThreads, chunkSize)
{
// 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);
}
BlackLynx::SearchLynx::URingExecDriver::~URingExecDriver()
{
// Clean up the ring
io_uring_queue_exit(&m_ring);
}
void BlackLynx::SearchLynx::URingExecDriver::start()
{
// Start threads in reverse order staring with cleanup thread
p_cleanupThread = new std::thread(&BlackLynx::SearchLynx::ExecDriver::cleanupThreadBody,
p_cleanupThread = new std::thread(&ExecDriver::cleanupThreadBody,
this,
&m_cleanupQueue);
......
for (uint32_t idx = 0; idx < m_numReaderThreads; ++idx)
{
std::thread* thisThread =
new std::thread(&BlackLynx::SearchLynx::URingExecDriver::readerThreadBody,
new std::thread(&URingExecDriver::readerThreadBody,
this,
&m_readerQueue,
&m_cleanupQueue);
......
}
// Start setup thread body
p_setupThread = new std::thread(&BlackLynx::SearchLynx::ExecDriver::setupThreadBody,
p_setupThread = new std::thread(&ExecDriver::setupThreadBody,
this,
&m_readerQueue,
m_numReaderThreads);
......
{
BlnxLog(info) << "URING READER starting";
struct io_uring m_ring;
std::atomic<uint32_t> m_ioCounter;
const uint32_t MIN_SIZE_ALIGN = 512;
// Setup IO context
// 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());
}
int fd = -1;
struct io_uring_sqe* sqe = nullptr;
struct io_uring_cqe *cqe = nullptr;
Chunk* chunk = nullptr;
std::string prevFilename("");
int rc = 0;
// std::ofstream output("chunk.bin");
......
sqe = io_uring_get_sqe(&m_ring);
if (! sqe)
{
waitForCompletion(outQueue);
waitForCompletion(&m_ring, m_ioCounter, outQueue);
}
}
......
io_uring_sqe_set_data(sqe, reinterpret_cast<void*>(chunk));
BlnxLog(debug) << "Reader Thread reading:"
<< " Filename: " << chunk->m_filename
// << " 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);
// << " 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;
......
if (rc < 0)
{
BlnxLog(error) << "io_uring_submit returned: " << rc;
--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
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;
Chunk* compChunk = reinterpret_cast<Chunk*>(io_uring_cqe_get_data(cqe));
outQueue->push_back(compChunk);
--m_ioCounter;
}
io_uring_cqe_seen(&m_ring, cqe);
}
else
{
break;
}
}
// NOTE: when using O_DIRECT for high performance the read size *must*
......
inFileStream.close();
}
if (m_ioCounter >= QUEUE_DEPTH)
{
waitForCompletion(outQueue);
}
}
// Wait for any lingering IO completions
while (m_ioCounter > 0)
{
waitForCompletion(outQueue);
waitForCompletion(&m_ring, m_ioCounter, outQueue);
}
// output.close();
......
// Send terminator to clean up thread
m_cleanupQueue.push_back(nullptr);
// Clean up the ring
io_uring_queue_exit(&m_ring);
}
void BlackLynx::SearchLynx::URingExecDriver::waitForCompletion(
struct io_uring* ring,
std::atomic<uint32_t>& ioCounter,
chunk_queue_t* outQueue)
{
int rc = 0;
......
bool gotOne = false;
Chunk* chunk = nullptr;
while (m_ioCounter > 0)
while (ioCounter > 0)
{
// Wait for at least one completion but handle any that are ready
if (! gotOne)
{
rc = io_uring_wait_cqe(&m_ring, &cqe);
rc = io_uring_wait_cqe(ring, &cqe);
gotOne = true;
}
else
{
rc = io_uring_peek_cqe(&m_ring, &cqe);
rc = io_uring_peek_cqe(ring, &cqe);
if (rc == -EAGAIN)
{
cqe = nullptr;
......
BlnxLog(debug) << "Got completion for chunk ID: " << chunk->m_chunkId;
outQueue->push_back(chunk);
--m_ioCounter;
io_uring_cqe_seen(&m_ring, cqe);
--ioCounter;
io_uring_cqe_seen(ring, cqe);
}
}

Also available in: Unified diff