commit 864b608ce1d6f627f17f7c4bcff9cf67b5fa7b39 Author: David Sorber Date: Fri Jul 8 11:04:09 2022 -0400 read_test: add new uring read method to read_test. It uses (requires) liburing and a kernel >= 5.6. diff --git a/software/read_test/CMakeLists.txt b/software/read_test/CMakeLists.txt index 7f36221..ee40328 100644 --- a/software/read_test/CMakeLists.txt +++ b/software/read_test/CMakeLists.txt @@ -108,11 +108,14 @@ ENDFUNCTION() CHECK_LIB(config++) CHECK_LIB(aio) +SET(READTEST_EXTERNAL_LIBS ${READTEST_EXTERNAL_LIBS} /home/dsorber/Documents/dev/liburing/src/liburing.a) + ################################################################################ # Include paths ################################################################################ INCLUDE_DIRECTORIES(${CMAKE_CURRENT_SOURCE_DIR}/src) +INCLUDE_DIRECTORIES(/home/dsorber/Documents/dev/liburing/src/include) ################################################################################ @@ -136,6 +139,7 @@ SET(local_sources ${CMAKE_CURRENT_SOURCE_DIR}/src/AIO_ExecDriver.cc ${CMAKE_CURRENT_SOURCE_DIR}/src/Chunk.cc ${CMAKE_CURRENT_SOURCE_DIR}/src/ExecDriver.cc + ${CMAKE_CURRENT_SOURCE_DIR}/src/URingExecDriver.cc ) ################################################################################ diff --git a/software/read_test/src/AIO_ExecDriver.cc b/software/read_test/src/AIO_ExecDriver.cc index bd66b83..22d88b1 100644 --- a/software/read_test/src/AIO_ExecDriver.cc +++ b/software/read_test/src/AIO_ExecDriver.cc @@ -64,7 +64,7 @@ void BlackLynx::SearchLynx::AIO_ExecDriver::readerThreadBody( chunk_queue_t* inQueue, chunk_queue_t* outQueue) { - BlnxLog(error) << "AIO READER"; + BlnxLog(info) << "AIO READER starting"; const uint32_t QUEUE_DEPTH = 32; const uint32_t IO_SIZE = 4194304; @@ -97,15 +97,7 @@ void BlackLynx::SearchLynx::AIO_ExecDriver::readerThreadBody( return; } - // 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(""); @@ -229,7 +221,8 @@ void BlackLynx::SearchLynx::AIO_ExecDriver::readerThreadBody( if (leftOverSize > 0) { - BlnxLog(error) << "TAIL CHUNK: " << chunk->m_chunkId << " -- " << leftOverSize; + 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) @@ -239,7 +232,7 @@ void BlackLynx::SearchLynx::AIO_ExecDriver::readerThreadBody( } inFileStream.seekg(chunk->m_fileOffset + (totalSize - leftOverSize)); - inFileStream.read(chunk->p_chunkData + (totalSize - leftOverSize), + inFileStream.read(chunk->p_chunkData + (totalSize - leftOverSize), leftOverSize); inFileStream.close(); diff --git a/software/read_test/src/Logger/SearchLynxLogger.h b/software/read_test/src/Logger/SearchLynxLogger.h index c9777bf..4ca54f3 100644 --- a/software/read_test/src/Logger/SearchLynxLogger.h +++ b/software/read_test/src/Logger/SearchLynxLogger.h @@ -28,7 +28,8 @@ namespace SearchLynx #ifdef CYBERLYNX #define LOG_FILENAME "/var/log/blacklynx/cyberlynx.log" #else -#define LOG_FILENAME "/var/log/blacklynx/searchlynx.log" +//#define LOG_FILENAME "/var/log/blacklynx/searchlynx.log" + #define LOG_FILENAME "read_test.log" #endif // Enumeration representing the log severity levels diff --git a/software/read_test/src/URingExecDriver.cc b/software/read_test/src/URingExecDriver.cc index 604ac75..10a1f39 100644 --- a/software/read_test/src/URingExecDriver.cc +++ b/software/read_test/src/URingExecDriver.cc @@ -10,12 +10,14 @@ * Unpublished -- all rights reserved under the copyright laws * * of the United States. * ********************************************************************/ +#include +#include +#include #include #include #include - #include "URingExecDriver.h" #include "Utilities/ChunkConstants.h" #include "Utilities/FileUtils.h" @@ -26,12 +28,25 @@ BlackLynx::SearchLynx::URingExecDriver::URingExecDriver( const std::string& inputFile, uint32_t numReaderThreads) - : ExecDriver(inputFile, numReaderThreads) + : ExecDriver(inputFile, numReaderThreads), + m_ioCounter(0) { + // 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() @@ -63,24 +78,16 @@ void BlackLynx::SearchLynx::URingExecDriver::readerThreadBody( chunk_queue_t* inQueue, chunk_queue_t* outQueue) { - BlnxLog(error) << "URING READER"; + BlnxLog(info) << "URING READER starting"; - 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; -// } - + struct io_uring_sqe* sqe = nullptr; Chunk* chunk = nullptr; std::string prevFilename(""); + int rc = 0; // std::ofstream output("chunk.bin"); @@ -108,13 +115,78 @@ void BlackLynx::SearchLynx::URingExecDriver::readerThreadBody( // 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; + // 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; -// output.write(chunk->p_chunkData, chunk->m_chunkSize); - - outQueue->push_back(chunk); + // 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(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); } // output.close(); @@ -122,9 +194,56 @@ void BlackLynx::SearchLynx::URingExecDriver::readerThreadBody( // Release file descriptor ::close(fd); - // Destroy the IO context - io_destroy(context); - // Send terminator to clean up thread m_cleanupQueue.push_back(nullptr); +} + +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(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); + } } \ No newline at end of file diff --git a/software/read_test/src/URingExecDriver.h b/software/read_test/src/URingExecDriver.h index e4eea10..b17b4de 100644 --- a/software/read_test/src/URingExecDriver.h +++ b/software/read_test/src/URingExecDriver.h @@ -20,6 +20,8 @@ #include #include +#include "liburing.h" + #include "Chunk.h" #include "ExecDriver.h" #include "Utilities/ProtectedAndSynchronizedQueue.h" @@ -34,6 +36,8 @@ class URingExecDriver : public ExecDriver { public: + static const uint32_t QUEUE_DEPTH = 256; + URingExecDriver( const std::string& inputFile, uint32_t numReaderThreads); @@ -48,6 +52,13 @@ private: chunk_queue_t* inQueue, chunk_queue_t* outQueue) override; + + void waitForCompletion(chunk_queue_t* outQueue); + + struct io_uring m_ring; + + uint32_t m_ioCounter; + }; } diff --git a/software/read_test/src/main.cc b/software/read_test/src/main.cc index 5504723..f37ca82 100644 --- a/software/read_test/src/main.cc +++ b/software/read_test/src/main.cc @@ -9,6 +9,7 @@ #include "AIO_ExecDriver.h" #include "ExecDriver.h" +#include "URingExecDriver.h" #include "Utilities/BufferAllocator.h" #include "Utilities/FileUtils.h" #include "Utilities/SearchLynxConfiguration.h" @@ -60,7 +61,8 @@ enum class ReadType DEFAULT_INVALID, AUTO, NORMAL, - AIO + AIO, + URING }; struct ReadTypeValue @@ -88,6 +90,10 @@ struct ReadTypeValue { return ReadType::AIO; } + else if (value == "uring") + { + return ReadType::URING; + } return ReadType::DEFAULT_INVALID; } @@ -108,7 +114,7 @@ void validate( // BlackLynx::SearchLynx::CharacterUtilities::lowercaseInPlace(val); lowercase(val); // Check for valid values - if (val == "auto" || val == "normal" || val == "aio") + if (val == "auto" || val == "normal" || val == "aio" || val == "uring") { v = boost::any(ReadTypeValue(val)); } @@ -123,6 +129,7 @@ int main(int argc, char** argv) bpo::variables_map optionsMap; ReadTypeValue readType("default"); std::string filename; + uint32_t numThreads = 0; try { @@ -137,6 +144,9 @@ int main(int argc, char** argv) ("read-type", bpo::value(&readType), "read type") + ("num-threads", bpo::value(&numThreads), + "number reader threads") + ("input-file", bpo::value(&filename)->required(), "input file") ; @@ -194,13 +204,29 @@ int main(int argc, char** argv) case ReadType::DEFAULT_INVALID: case ReadType::AUTO: case ReadType::AIO: - driver = new bsl::AIO_ExecDriver(filename, 2); + // Set default number of threads + if (numThreads == 0) + { + numThreads = 3; + } + driver = new bsl::AIO_ExecDriver(filename, numThreads); break; case ReadType::NORMAL: - driver = new bsl::ExecDriver(filename, 22); + // Set default number of threads + if (numThreads == 0) + { + numThreads = 22; + } + driver = new bsl::ExecDriver(filename, numThreads); + break; + + case ReadType::URING: + numThreads = 1; + driver = new bsl::URingExecDriver(filename, 1); break; } + std::cout << "Num threads: " << numThreads << "\n"; driver->start(); driver->waitUntilComplete();