Project

General

Profile

« Previous | Next » 

Revision 70882817

Added by David Sorber about 3 years ago

Adding simple single threaded write_test utility.

View differences:

software/read_test/CMakeLists.txt
# Include paths
################################################################################
INCLUDE_DIRECTORIES(${CMAKE_CURRENT_SOURCE_DIR}/src)
#INCLUDE_DIRECTORIES(/home/dsorber/Documents/dev/liburing/src/include)
################################################################################
......
################################################################################
# TARGET: SearchLynx command line executable
# TARGET: read_test command line executable
################################################################################
ADD_EXECUTABLE(read_test ${CMAKE_CURRENT_SOURCE_DIR}/src/main.cc
${logger_sources}
......
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})
software/read_test/scripts/make_data.py
#!/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())
software/read_test/src/URingExecDriver.cc
BlnxLog(info) << "URING READER starting";
struct io_uring m_ring;
std::atomic<uint32_t> m_ioCounter;
std::atomic<uint32_t> m_ioCounter = 0;
const uint32_t MIN_SIZE_ALIGN = 512;
// Init the ring
......
// Grab the data (chunk) associated with this request and then pass
// it along
Chunk* compChunk = reinterpret_cast<Chunk*>(io_uring_cqe_get_data(cqe));
if (cqe->res < 0)
{
BlnxLog(error) << "Chunk ID: " << chunk->m_chunkId
......
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;
}
--m_ioCounter;
io_uring_cqe_seen(&m_ring, cqe);
}
else
......
--ioCounter;
io_uring_cqe_seen(ring, cqe);
}
}
}
software/read_test/src/write_test.cc
#include <atomic>
#include <cstdint>
#include <iostream>
#include <sstream>
#include <string>
#include <vector>
#include "liburing.h"
static const uint32_t QUEUE_DEPTH = 256;
void waitForCompletion(
struct io_uring* ring,
std::atomic<uint32_t>& 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<std::string> 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<uint32_t> 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<uint32_t> 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<void*>(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;
}

Also available in: Unified diff