Project

General

Profile

« Previous | Next » 

Revision 864b608c

Added by David Sorber about 4 years ago

read_test: add new uring read method to read_test. It uses (requires)
liburing and a kernel >= 5.6.

View differences:

software/read_test/CMakeLists.txt
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)
################################################################################
......
${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
)
################################################################################
software/read_test/src/AIO_ExecDriver.cc
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;
......
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("");
......
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)
......
}
inFileStream.seekg(chunk->m_fileOffset + (totalSize - leftOverSize));
inFileStream.read(chunk->p_chunkData + (totalSize - leftOverSize),
inFileStream.read(chunk->p_chunkData + (totalSize - leftOverSize),
leftOverSize);
inFileStream.close();
software/read_test/src/Logger/SearchLynxLogger.h
#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
software/read_test/src/URingExecDriver.cc
* Unpublished -- all rights reserved under the copyright laws *
* of the United States. *
********************************************************************/
#include <cerrno>
#include <sstream>
#include <stdexcept>
#include <sys/types.h>
#include <sys/stat.h>
#include <fcntl.h>
#include "URingExecDriver.h"
#include "Utilities/ChunkConstants.h"
#include "Utilities/FileUtils.h"
......
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()
......
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");
......
// 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<void*>(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();
......
// 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<Chunk*>(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);
}
}
software/read_test/src/URingExecDriver.h
#include <thread>
#include <vector>
#include "liburing.h"
#include "Chunk.h"
#include "ExecDriver.h"
#include "Utilities/ProtectedAndSynchronizedQueue.h"
......
{
public:
static const uint32_t QUEUE_DEPTH = 256;
URingExecDriver(
const std::string& inputFile,
uint32_t numReaderThreads);
......
chunk_queue_t* inQueue,
chunk_queue_t* outQueue) override;
void waitForCompletion(chunk_queue_t* outQueue);
struct io_uring m_ring;
uint32_t m_ioCounter;
};
}
software/read_test/src/main.cc
#include "AIO_ExecDriver.h"
#include "ExecDriver.h"
#include "URingExecDriver.h"
#include "Utilities/BufferAllocator.h"
#include "Utilities/FileUtils.h"
#include "Utilities/SearchLynxConfiguration.h"
......
DEFAULT_INVALID,
AUTO,
NORMAL,
AIO
AIO,
URING
};
struct ReadTypeValue
......
{
return ReadType::AIO;
}
else if (value == "uring")
{
return ReadType::URING;
}
return ReadType::DEFAULT_INVALID;
}
......
// 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));
}
......
bpo::variables_map optionsMap;
ReadTypeValue readType("default");
std::string filename;
uint32_t numThreads = 0;
try
{
......
("read-type", bpo::value<ReadTypeValue>(&readType),
"read type")
("num-threads", bpo::value<uint32_t>(&numThreads),
"number reader threads")
("input-file", bpo::value<std::string>(&filename)->required(),
"input file")
;
......
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();

Also available in: Unified diff