commit 3dda5d70ff8d22080a1e807946c961ef412ae045 Author: david.sorber Date: Fri Jul 7 15:46:11 2023 -0400 Additional read_test uring adjustments. diff --git a/software/read_test/cmake/liburing.cmake b/software/read_test/cmake/liburing.cmake index f52b84c..3b3c5bc 100644 --- a/software/read_test/cmake/liburing.cmake +++ b/software/read_test/cmake/liburing.cmake @@ -38,7 +38,7 @@ ExternalProject_Add(liburing_local BUILD_BYPRODUCTS ${liburing_static_lib} ) -INCLUDE_DIRECTORIES(${LIBURING_PREFIX}/src/liburing-build/include) +INCLUDE_DIRECTORIES(${LIBURING_PREFIX}/include) SET_SOURCE_FILES_PROPERTIES( ${liburing_static_lib} PROPERTIES diff --git a/software/read_test/src/URingExecDriver.cc b/software/read_test/src/URingExecDriver.cc index aa2cf09..3800ae3 100644 --- a/software/read_test/src/URingExecDriver.cc +++ b/software/read_test/src/URingExecDriver.cc @@ -28,31 +28,20 @@ BlackLynx::SearchLynx::URingExecDriver::URingExecDriver( 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); @@ -60,7 +49,7 @@ void BlackLynx::SearchLynx::URingExecDriver::start() 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); @@ -68,7 +57,7 @@ void BlackLynx::SearchLynx::URingExecDriver::start() } // 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); @@ -80,14 +69,24 @@ void BlackLynx::SearchLynx::URingExecDriver::readerThreadBody( { BlnxLog(info) << "URING READER starting"; + struct io_uring m_ring; + std::atomic 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"); @@ -123,7 +122,7 @@ void BlackLynx::SearchLynx::URingExecDriver::readerThreadBody( sqe = io_uring_get_sqe(&m_ring); if (! sqe) { - waitForCompletion(outQueue); + waitForCompletion(&m_ring, m_ioCounter, outQueue); } } @@ -139,14 +138,15 @@ void BlackLynx::SearchLynx::URingExecDriver::readerThreadBody( io_uring_sqe_set_data(sqe, reinterpret_cast(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; @@ -154,6 +154,40 @@ void BlackLynx::SearchLynx::URingExecDriver::readerThreadBody( 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(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* @@ -177,17 +211,12 @@ void BlackLynx::SearchLynx::URingExecDriver::readerThreadBody( 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(); @@ -197,9 +226,14 @@ void BlackLynx::SearchLynx::URingExecDriver::readerThreadBody( // 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& ioCounter, chunk_queue_t* outQueue) { int rc = 0; @@ -207,17 +241,17 @@ void BlackLynx::SearchLynx::URingExecDriver::waitForCompletion( 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; @@ -244,7 +278,7 @@ void BlackLynx::SearchLynx::URingExecDriver::waitForCompletion( 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); } } \ No newline at end of file diff --git a/software/read_test/src/URingExecDriver.h b/software/read_test/src/URingExecDriver.h index 5b0c330..12e3cf2 100644 --- a/software/read_test/src/URingExecDriver.h +++ b/software/read_test/src/URingExecDriver.h @@ -54,12 +54,10 @@ private: chunk_queue_t* inQueue, chunk_queue_t* outQueue) override; - - void waitForCompletion(chunk_queue_t* outQueue); - - struct io_uring m_ring; - - std::atomic m_ioCounter; + void waitForCompletion( + struct io_uring* ring, + std::atomic& ioCounter, + chunk_queue_t* outQueue); }; diff --git a/software/read_test/src/main.cc b/software/read_test/src/main.cc index 0c83394..44c5027 100644 --- a/software/read_test/src/main.cc +++ b/software/read_test/src/main.cc @@ -296,8 +296,8 @@ int main(int argc, char** argv) break; case ReadType::URING: - numThreads = 1; - driver = new bsl::URingExecDriver(filePath, 1, chunkSize); +// numThreads = 1; + driver = new bsl::URingExecDriver(filePath, numThreads, chunkSize); break; } std::cout << "Num threads: " << numThreads << "\n";