commit 708828177bb07b3bc72b23802db2acef955ee2f1 Author: David Sorber Date: Sun Jul 9 11:07:37 2023 -0400 Adding simple single threaded write_test utility. diff --git a/software/read_test/CMakeLists.txt b/software/read_test/CMakeLists.txt index 9b838e2..85e20c2 100644 --- a/software/read_test/CMakeLists.txt +++ b/software/read_test/CMakeLists.txt @@ -123,7 +123,6 @@ SET(READTEST_EXTERNAL_LIBS ${READTEST_EXTERNAL_LIBS} ${liburing_static_lib}) # Include paths ################################################################################ INCLUDE_DIRECTORIES(${CMAKE_CURRENT_SOURCE_DIR}/src) -#INCLUDE_DIRECTORIES(/home/dsorber/Documents/dev/liburing/src/include) ################################################################################ @@ -149,7 +148,7 @@ SET(local_sources ################################################################################ -# TARGET: SearchLynx command line executable +# TARGET: read_test command line executable ################################################################################ ADD_EXECUTABLE(read_test ${CMAKE_CURRENT_SOURCE_DIR}/src/main.cc ${logger_sources} @@ -159,8 +158,20 @@ ADD_EXECUTABLE(read_test ${CMAKE_CURRENT_SOURCE_DIR}/src/main.cc ADD_DEPENDENCIES(read_test liburing_local) TARGET_COMPILE_DEFINITIONS(read_test PRIVATE -DCOMPONENT_ID="ReadTest") -#SET_TARGET_PROPERTIES(searchlynx_exe PROPERTIES OUTPUT_NAME searchlynx) -#TARGET_COMPILE_DEFINITIONS(searchlynx_exe PRIVATE) TARGET_LINK_LIBRARIES(read_test ${READTEST_EXTERNAL_LIBS} ${CMAKE_THREAD_LIBS_INIT}) + +################################################################################ +# TARGET: write_test command line executable +################################################################################ +ADD_EXECUTABLE(write_test ${CMAKE_CURRENT_SOURCE_DIR}/src/write_test.cc + ${logger_sources} + ${utilities_sources} + ${local_sources}) + +ADD_DEPENDENCIES(write_test liburing_local) + +TARGET_COMPILE_DEFINITIONS(write_test PRIVATE -DCOMPONENT_ID="ReadTest") +TARGET_LINK_LIBRARIES(write_test ${READTEST_EXTERNAL_LIBS} + ${CMAKE_THREAD_LIBS_INIT}) diff --git a/software/read_test/scripts/make_data.py b/software/read_test/scripts/make_data.py new file mode 100755 index 0000000..a81deb4 --- /dev/null +++ b/software/read_test/scripts/make_data.py @@ -0,0 +1,14 @@ +#!/usr/bin/python3 +import random +import sys + +def main(): + print('Make Data\n') + + for ctr in range(97, 123): + char = chr(ctr) + num = random.randint(80, 100) + print(f'{{"{char * num}\\n"}},') + +if __name__ == '__main__': + sys.exit(main()) diff --git a/software/read_test/src/URingExecDriver.cc b/software/read_test/src/URingExecDriver.cc index 3800ae3..34e887a 100644 --- a/software/read_test/src/URingExecDriver.cc +++ b/software/read_test/src/URingExecDriver.cc @@ -70,7 +70,7 @@ void BlackLynx::SearchLynx::URingExecDriver::readerThreadBody( BlnxLog(info) << "URING READER starting"; struct io_uring m_ring; - std::atomic m_ioCounter; + std::atomic m_ioCounter = 0; const uint32_t MIN_SIZE_ALIGN = 512; // Init the ring @@ -170,6 +170,7 @@ void BlackLynx::SearchLynx::URingExecDriver::readerThreadBody( // Grab the data (chunk) associated with this request and then pass // it along + Chunk* compChunk = reinterpret_cast(io_uring_cqe_get_data(cqe)); if (cqe->res < 0) { BlnxLog(error) << "Chunk ID: " << chunk->m_chunkId @@ -178,10 +179,9 @@ void BlackLynx::SearchLynx::URingExecDriver::readerThreadBody( 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; } + --m_ioCounter; io_uring_cqe_seen(&m_ring, cqe); } else @@ -281,4 +281,4 @@ void BlackLynx::SearchLynx::URingExecDriver::waitForCompletion( --ioCounter; io_uring_cqe_seen(ring, cqe); } -} \ No newline at end of file +} diff --git a/software/read_test/src/write_test.cc b/software/read_test/src/write_test.cc new file mode 100644 index 0000000..0b77d23 --- /dev/null +++ b/software/read_test/src/write_test.cc @@ -0,0 +1,202 @@ +#include +#include +#include +#include +#include +#include + +#include "liburing.h" + +static const uint32_t QUEUE_DEPTH = 256; + + +void waitForCompletion( + struct io_uring* ring, + std::atomic& ioCounter) +{ + int rc = 0; + struct io_uring_cqe *cqe = nullptr; + bool gotOne = false; + + while (ioCounter > 0) + { + // Wait for at least one completion but handle any that are ready + if (! gotOne) + { + rc = io_uring_wait_cqe(ring, &cqe); + gotOne = true; + } + else + { + rc = io_uring_peek_cqe(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 + uint32_t index = io_uring_cqe_get_data64(cqe); + if (cqe->res < 0) + { + std::cerr << "Index: " << index + << " -- completion returned: " << cqe->res << std::endl; + } + else + { + std::cout << "Got completion for index: " << index << std::endl; + } + + --ioCounter; + io_uring_cqe_seen(ring, cqe); + } +} + +int main(int argc, char** argv) +{ + std::cout << "Write Util\n" << std::endl; + + std::vector DATA = { + {"aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa\n"}, + {"bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb\n"}, + {"ccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccc\n"}, + {"ddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddd\n"}, + {"eeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeee\n"}, + {"fffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffff\n"}, + {"ggggggggggggggggggggggggggggggggggggggggggggggggggggggggggggggggggggggggggggggggggggggggg\n"}, + {"hhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhh\n"}, + {"iiiiiiiiiiiiiiiiiiiiiiiiiiiiiiiiiiiiiiiiiiiiiiiiiiiiiiiiiiiiiiiiiiiiiiiiiiiiiiiiiiiiiiiiiiiiiiii\n"}, + {"jjjjjjjjjjjjjjjjjjjjjjjjjjjjjjjjjjjjjjjjjjjjjjjjjjjjjjjjjjjjjjjjjjjjjjjjjjjjjjjjjjjjjjjjjjjjjj\n"}, + {"kkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkk\n"}, + {"llllllllllllllllllllllllllllllllllllllllllllllllllllllllllllllllllllllllllllllllllllllllllllll\n"}, + {"mmmmmmmmmmmmmmmmmmmmmmmmmmmmmmmmmmmmmmmmmmmmmmmmmmmmmmmmmmmmmmmmmmmmmmmmmmmmmmmmmm\n"}, + {"nnnnnnnnnnnnnnnnnnnnnnnnnnnnnnnnnnnnnnnnnnnnnnnnnnnnnnnnnnnnnnnnnnnnnnnnnnnnnnnnnnnnnnnnnnn\n"}, + {"oooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooo\n"}, + {"pppppppppppppppppppppppppppppppppppppppppppppppppppppppppppppppppppppppppppppppppppppppppppppppppppp\n"}, + {"qqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqq\n"}, + {"rrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrr\n"}, + {"ssssssssssssssssssssssssssssssssssssssssssssssssssssssssssssssssssssssssssssssssssss\n"}, + {"tttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttt\n"}, + {"uuuuuuuuuuuuuuuuuuuuuuuuuuuuuuuuuuuuuuuuuuuuuuuuuuuuuuuuuuuuuuuuuuuuuuuuuuuuuuuuuuuuuuuuuu\n"}, + {"vvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvv\n"}, + {"wwwwwwwwwwwwwwwwwwwwwwwwwwwwwwwwwwwwwwwwwwwwwwwwwwwwwwwwwwwwwwwwwwwwwwwwwwwwwwwwwwwwww\n"}, + {"xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx\n"}, + {"yyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyy\n"}, + {"zzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzz\n"} + }; + + std::string outFilePath{"output.bin"}; + + struct io_uring m_ring; + std::atomic m_ioCounter = 0; + //~ const uint32_t MIN_SIZE_ALIGN = 512; + + // 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; + + // Open the file + //~ fd = ::open(outFilePath.c_str(), O_WRONLY | O_DIRECT); + fd = ::open(outFilePath.c_str(), O_CREAT | O_WRONLY, 0644); + if (fd < 0) + { + std::cerr << "Unable to open file: " << outFilePath << std::endl; + return -1; + } + + std::atomic offset = 0; + for (uint32_t itemIdx = 0; itemIdx < DATA.size(); ++itemIdx) + { + // Wait until a submission queue entry is ready + sqe = nullptr; + while (sqe == nullptr) + { + sqe = io_uring_get_sqe(&m_ring); + if (! sqe) + { + waitForCompletion(&m_ring, m_ioCounter); + } + } + + // Prep the write request and set the chunk object as its associated + // data + std::string& item(DATA[itemIdx]); + io_uring_prep_write(sqe, fd, item.data(), item.size(), offset); + offset += item.size(); + + //~ io_uring_sqe_set_data(sqe, reinterpret_cast(chunk)); + io_uring_sqe_set_data64(sqe, itemIdx); + + // Update the counter and submit the request + ++m_ioCounter; + rc = io_uring_submit(&m_ring); + if (rc < 0) + { + std::cerr << "io_uring_submit returned: " << rc << std::endl; + --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 + uint32_t index = io_uring_cqe_get_data64(cqe); + if (cqe->res < 0) + { + std::cerr << "Index: " << index << " completion returned: " + << cqe->res << std::endl; + } + else + { + std::cout << "Got completion for index: " << index << std::endl; + } + --m_ioCounter; + io_uring_cqe_seen(&m_ring, cqe); + } + else + { + break; + } + } + } + + // Wait for any lingering IO completions + while (m_ioCounter > 0) + { + waitForCompletion(&m_ring, m_ioCounter); + } + + // Release file descriptor + ::close(fd); + + // Clean up the ring + io_uring_queue_exit(&m_ring); + + return 0; +}