commit a0d1d164e641d1c7c4dc023ea1dbeb6bd808e6e9 Author: David Sorber Date: Tue Jul 5 21:06:07 2022 -0400 Adding initial version of read_test. diff --git a/software/read_test/CMakeLists.txt b/software/read_test/CMakeLists.txt new file mode 100644 index 0000000..d3b912b --- /dev/null +++ b/software/read_test/CMakeLists.txt @@ -0,0 +1,155 @@ +PROJECT("read_test") +cmake_minimum_required(VERSION 3.5) + +IF (NOT CMAKE_BUILD_TYPE) + SET(CMAKE_BUILD_TYPE ReleaseWithDebug CACHE STRING + "Choose the type of build, options are: None Debug SearchLynxRelease ReleaseWithDebug." + FORCE ) +ENDIF() + +MESSAGE(STATUS "Build Type: " ${CMAKE_BUILD_TYPE}) + +ADD_CUSTOM_TARGET(Debug + COMMAND ${CMAKE_COMMAND} -DCMAKE_BUILD_TYPE=Debug ${CMAKE_SOURCE_DIR} + COMMAND ${CMAKE_COMMAND} --build ${CMAKE_BINARY_DIR} --target all + COMMENT "Switch CMAKE_BUILD_TYPE to Debug" +) + +ADD_CUSTOM_TARGET(SearchLynxRelease + COMMAND ${CMAKE_COMMAND} -DCMAKE_BUILD_TYPE=SearchLynxRelease ${CMAKE_SOURCE_DIR} + COMMAND ${CMAKE_COMMAND} --build ${CMAKE_BINARY_DIR} --target all + COMMENT "Switch CMAKE_BUILD_TYPE to SearchLynxRelease" +) + +ADD_CUSTOM_TARGET(ReleaseWithDebug + COMMAND ${CMAKE_COMMAND} -DCMAKE_BUILD_TYPE=ReleaseWithDebug ${CMAKE_SOURCE_DIR} + COMMAND ${CMAKE_COMMAND} --build ${CMAKE_BINARY_DIR} --target all + COMMENT "Switch CMAKE_BUILD_TYPE to ReleaseWithDebug" +) + +# Uncomment this line to enable AddressSanitizer; remember to also disable +# tcmalloc +#SET(CMAKE_CXX_FLAGS "-fsanitize=address -fno-omit-frame-pointer -fuse-ld=gold") + +# +# Set the compile flags +# +SET(CMAKE_CXX_STANDARD 14) +SET(CMAKE_CXX_STANDARD_REQUIRED ON) +SET(CMAKE_CXX_EXTENSIONS OFF) + +ADD_DEFINITIONS("-Wall -Wextra -Wno-unused-parameter -mavx2") +#ADD_DEFINITIONS("-fno-builtin-malloc -fno-builtin-calloc -fno-builtin-realloc -fno-builtin-free") + +# This define is used along with the __FILE__ macro by the Logger to remove +# the root of the build path from log messages. +STRING(LENGTH ${CMAKE_CURRENT_SOURCE_DIR} BP_LEN) +ADD_DEFINITIONS("-DBUILD_PATH_LEN=${BP_LEN}") + +IF (CMAKE_BUILD_TYPE STREQUAL "SearchLynxRelease") + + ADD_DEFINITIONS("-O3 -flto") + + # Tune for Haswell minimum + ADD_DEFINITIONS("-march=haswell -mtune=intel") + +ELSEIF (CMAKE_BUILD_TYPE STREQUAL "ReleaseWithDebug") + + # NOTE: remove "-flto" to temporarily improve debug compile time but don't + # forget to re-add it! -flto + ADD_DEFINITIONS("-g -O3") + + # Tune for Haswell minimum + ADD_DEFINITIONS("-march=haswell -mtune=intel") + +ELSEIF (CMAKE_BUILD_TYPE STREQUAL "Debug") + + ADD_DEFINITIONS("-g -O0") + +ENDIF() + +SET(READTEST_EXTERNAL_LIBS ) + +################################################################################ +# Find the threads package +################################################################################ +SET(THREADS_PREFER_PTHREAD_FLAG ON) +FIND_PACKAGE(Threads REQUIRED) + + +################################################################################ +# Check for Boost, set include paths, and add to SEARCHLYNX_EXTERNAL_LIBS +# add any additional Boost library needs here +################################################################################ +FIND_PACKAGE(Boost 1.53.0 COMPONENTS system filesystem date_time program_options + regex iostreams serialization REQUIRED) +IF (Boost_FOUND) + INCLUDE_DIRECTORIES(${Boost_INCLUDE_DIRS}) + SET(READTEST_EXTERNAL_LIBS ${READTEST_EXTERNAL_LIBS} ${Boost_LIBRARIES}) +ENDIF() + + +################################################################################ +# Function to check for a library and, if found, add it to CYBERLYNX_EXTERNAL_LIBS +################################################################################ +FUNCTION(check_lib lib_name) + FIND_LIBRARY(LIB_${lib_name} ${lib_name}) + IF (NOT LIB_${lib_name}) + MESSAGE(FATAL_ERROR "Can't find ${lib_name}!") + ELSE() + SET(READTEST_EXTERNAL_LIBS ${READTEST_EXTERNAL_LIBS} ${LIB_${lib_name}} PARENT_SCOPE) + ENDIF() +ENDFUNCTION() + + +################################################################################ +# Find the libraries we need that aren't built-in to cmake +################################################################################ +CHECK_LIB(config++) +CHECK_LIB(aio) + + +################################################################################ +# Include paths +################################################################################ +INCLUDE_DIRECTORIES(${CMAKE_CURRENT_SOURCE_DIR}/src) + + +################################################################################ +# Source +################################################################################ +SET(logger_sources + ${CMAKE_CURRENT_SOURCE_DIR}/src/Logger/SearchLynxLogger.cc + ${CMAKE_CURRENT_SOURCE_DIR}/src/Logger/SearchLynxCLogger.cc +) + +SET(utilities_sources + ${CMAKE_CURRENT_SOURCE_DIR}/src/Utilities/SearchLynxConfiguration.cc +# ${CMAKE_CURRENT_SOURCE_DIR}/src/lib/Utilities/TranslatedChunkManager.cc +# ${CMAKE_CURRENT_SOURCE_DIR}/src/lib/Utilities/ChunkData.cc + ${CMAKE_CURRENT_SOURCE_DIR}/src/Utilities/BufferAllocator.cc + ${CMAKE_CURRENT_SOURCE_DIR}/src/Utilities/FileUtils.cc + ${CMAKE_CURRENT_SOURCE_DIR}/src/Utilities/SystemArchInfo.cc +) + +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 +) + +################################################################################ +# TARGET: SearchLynx command line executable +################################################################################ +ADD_EXECUTABLE(read_test ${CMAKE_CURRENT_SOURCE_DIR}/src/main.cc + ${logger_sources} + ${utilities_sources} + ${local_sources}) + + +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}) + diff --git a/software/read_test/src/AIO_ExecDriver.cc b/software/read_test/src/AIO_ExecDriver.cc new file mode 100644 index 0000000..bd66b83 --- /dev/null +++ b/software/read_test/src/AIO_ExecDriver.cc @@ -0,0 +1,266 @@ +/******************************************************************** +* This software is "commercial computer software" as defined in the * +* Federal Acquisition Regulations and is subject to BlackLynx * +* Inc's standard End User License Agreement. * +* * +* CONFIDENTIAL - All source code is the propriety and confidential * +* information of BlackLynx Inc. * +* * +* Copyright 2014-2020 BlackLynx Inc. * +* Unpublished -- all rights reserved under the copyright laws * +* of the United States. * +********************************************************************/ + +#include +#include +#include + +#include + +#include "AIO_ExecDriver.h" +#include "Utilities/ChunkConstants.h" +#include "Utilities/FileUtils.h" +#include "Utilities/SearchLynxConfiguration.h" +#include "Utilities/BufferAllocator.h" + + +BlackLynx::SearchLynx::AIO_ExecDriver::AIO_ExecDriver( + const std::string& inputFile, + uint32_t numReaderThreads) + : ExecDriver(inputFile, numReaderThreads) +{ +} + +BlackLynx::SearchLynx::AIO_ExecDriver::~AIO_ExecDriver() +{ +} + +void BlackLynx::SearchLynx::AIO_ExecDriver::start() +{ + // Start threads in reverse order staring with cleanup thread + p_cleanupThread = new std::thread(&BlackLynx::SearchLynx::ExecDriver::cleanupThreadBody, + this, + &m_cleanupQueue); + + // Start specified number of reader threads + for (uint32_t idx = 0; idx < m_numReaderThreads; ++idx) + { + std::thread* thisThread = + new std::thread(&BlackLynx::SearchLynx::AIO_ExecDriver::readerThreadBody, + this, + &m_readerQueue, + &m_cleanupQueue); + m_readerThreads.push_back(thisThread); + } + + // Start setup thread body + p_setupThread = new std::thread(&BlackLynx::SearchLynx::ExecDriver::setupThreadBody, + this, + &m_readerQueue, + m_numReaderThreads); +} + +void BlackLynx::SearchLynx::AIO_ExecDriver::readerThreadBody( + chunk_queue_t* inQueue, + chunk_queue_t* outQueue) +{ + BlnxLog(error) << "AIO READER"; + + const uint32_t QUEUE_DEPTH = 32; + const uint32_t IO_SIZE = 4194304; + const uint32_t MIN_SIZE_ALIGN = 512; + std::vector chunkIoReqs; + std::vector chunkIoReqPtrs; + std::vector events{QUEUE_DEPTH}; + + uint32_t maxSize = DEFAULT_CHUNK_SIZE + EXTRA_BYTES_TO_READ; + uint32_t maxIOs = maxSize / IO_SIZE; + if (maxSize % IO_SIZE > 0) + { + ++maxIOs; + } + + // Resize the chunk IO list and ptr list + chunkIoReqs.resize(maxIOs); + chunkIoReqPtrs.resize(maxIOs); + for (uint32_t idx = 0; idx < maxIOs; ++idx) + { + chunkIoReqPtrs[idx] = &chunkIoReqs[idx]; + } + + io_context_t context = 0; + + // Setup IO context + if (io_setup(QUEUE_DEPTH, &context) < 0) + { + BlnxLog(error) << "Unable to setup IO context"; + 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(""); + +// std::ofstream output("chunk.bin"); + + while (true) + { + chunk = inQueue->pop_front(); + + // Check for terminator + if (chunk == nullptr) + { + break; + } + + if (prevFilename != chunk->m_filename) + { + // Open the file + fd = ::open(r_inputFile.c_str(), O_RDONLY | O_DIRECT); + if (fd < 0) + { + BlnxLog(error) << "Unable to open file: " << r_inputFile; + return; + } + } + + // Grab a buffer to hold the chunk data + chunk->allocateBuffer(); + + // Determine how many IOs we need + uint32_t totalSize = chunk->m_chunkSize + chunk->m_extraSize; + uint32_t totalIOs = totalSize / IO_SIZE; + if (totalSize % IO_SIZE > 0) + { + ++totalIOs; + } + + 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 + << " Total IOs: " << totalIOs; + + // Build all of the IO reqs + off_t bufferOffset = 0; + off_t fileOffset = chunk->m_fileOffset; + uint32_t leftToRead = totalSize; + uint32_t readSize = 0; + + uint32_t leftOverSize = 0; + + for (uint32_t idx = 0; idx < totalIOs; ++idx) + { + // Determine how much to data to read in this IO request + if (leftToRead >= IO_SIZE) + { + readSize = IO_SIZE; + leftToRead -= IO_SIZE; + } + else + { + readSize = leftToRead; + leftToRead = 0; + + // Adjust "tail" size so it is at least a multiple of 512 + leftOverSize = (readSize & (MIN_SIZE_ALIGN - 1)); + readSize -= leftOverSize; + } + + io_prep_pread(&chunkIoReqs[idx], fd, chunk->p_chunkData + bufferOffset, + readSize, fileOffset); + + bufferOffset += readSize; + fileOffset += readSize; + } + + uint32_t inflight = 0; + uint32_t submitted = 0; + uint32_t completed = 0; + uint32_t reqIdx = 0; + uint32_t toSubmit = 0; + int rc = 0; + + while (completed < totalIOs) + { + if (submitted < totalIOs) + { + toSubmit = QUEUE_DEPTH - inflight; + + rc = io_submit(context, toSubmit, &chunkIoReqPtrs[reqIdx]); + if (rc < 0) + { + BlnxLog(error) << "io_submit returned: " << rc << " -- " << reqIdx; + // WHAT NOW? + } + else + { +// BlnxLog(debug) << "io_submit returned: " << rc; + submitted += rc; + inflight += rc; + reqIdx += rc; + } + } + + int rc = io_getevents(context, 1, QUEUE_DEPTH, events.data(), nullptr); + if (rc < 0) + { + BlnxLog(error) << "io_getevents returned: " << rc; + // WHAT NOW? + } + else + { +// BlnxLog(debug) << "io_getevents returned: " << rc; + + inflight -= rc; + completed += rc; + } + } + + if (leftOverSize > 0) + { + BlnxLog(error) << "TAIL CHUNK: " << chunk->m_chunkId << " -- " << leftOverSize; + + 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(); + } + +// std::string test(chunk->p_chunkData, 30); +// BlnxLog(debug) << "CHUNK " << chunk->m_chunkId << " DATA: " << test; + +// output.write(chunk->p_chunkData, chunk->m_chunkSize); + + outQueue->push_back(chunk); + } + +// 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); +} \ No newline at end of file diff --git a/software/read_test/src/AIO_ExecDriver.h b/software/read_test/src/AIO_ExecDriver.h new file mode 100644 index 0000000..870d6fc --- /dev/null +++ b/software/read_test/src/AIO_ExecDriver.h @@ -0,0 +1,58 @@ +/******************************************************************** +* This software is "commercial computer software" as defined in the * +* Federal Acquisition Regulations and is subject to BlackLynx * +* Inc's standard End User License Agreement. * +* * +* CONFIDENTIAL - All source code is the propriety and confidential * +* information of BlackLynx Inc. * +* * +* Copyright 2014-2020 BlackLynx Inc. * +* Unpublished -- all rights reserved under the copyright laws * +* of the United States. * +********************************************************************/ + +#ifndef __BLACKLYNX_AIO_EXECDRIVER_H__ +#define __BLACKLYNX_AIO_EXECDRIVER_H__ + +#include +#include +#include +#include +#include + +#include "Chunk.h" +#include "ExecDriver.h" +#include "Utilities/ProtectedAndSynchronizedQueue.h" + + +namespace BlackLynx +{ +namespace SearchLynx +{ + +class AIO_ExecDriver : public ExecDriver +{ +public: + + AIO_ExecDriver( + const std::string& inputFile, + uint32_t numReaderThreads); + + virtual ~AIO_ExecDriver(); + + void start(); + +// void waitUntilComplete(); + +private: + + virtual void readerThreadBody( + chunk_queue_t* inQueue, + chunk_queue_t* outQueue); + +}; + +} +} + +#endif // __BLACKLYNX_AIO_EXECDRIVER_H__ \ No newline at end of file diff --git a/software/read_test/src/Chunk.cc b/software/read_test/src/Chunk.cc new file mode 100644 index 0000000..6dad828 --- /dev/null +++ b/software/read_test/src/Chunk.cc @@ -0,0 +1,41 @@ +/******************************************************************** +* This software is "commercial computer software" as defined in the * +* Federal Acquisition Regulations and is subject to BlackLynx * +* Inc's standard End User License Agreement. * +* * +* CONFIDENTIAL - All source code is the propriety and confidential * +* information of BlackLynx Inc. * +* * +* Copyright 2014-2020 BlackLynx Inc. * +* Unpublished -- all rights reserved under the copyright laws * +* of the United States. * +********************************************************************/ + +#include "Chunk.h" +#include "Utilities/BufferAllocator.h" + +BlackLynx::SearchLynx::Chunk::Chunk( + const std::string& filename, + uint64_t chunkId, + uint32_t chunkSize, + uint32_t extraSize, + uint64_t fileOffset) + : m_filename(filename), + m_chunkId(chunkId), + m_chunkSize(chunkSize), + m_extraSize(extraSize), + m_fileOffset(fileOffset), + p_chunkData(nullptr) +{ +} + +BlackLynx::SearchLynx::Chunk::~Chunk() +{ +} + +void BlackLynx::SearchLynx::Chunk::allocateBuffer() +{ + auto bufAllocator = BufferAllocator::getInstance(); + p_chunkData = bufAllocator->request(m_chunkSize + m_extraSize); +} + diff --git a/software/read_test/src/Chunk.h b/software/read_test/src/Chunk.h new file mode 100644 index 0000000..3b3911c --- /dev/null +++ b/software/read_test/src/Chunk.h @@ -0,0 +1,57 @@ +/******************************************************************** +* This software is "commercial computer software" as defined in the * +* Federal Acquisition Regulations and is subject to BlackLynx * +* Inc's standard End User License Agreement. * +* * +* CONFIDENTIAL - All source code is the propriety and confidential * +* information of BlackLynx Inc. * +* * +* Copyright 2014-2020 BlackLynx Inc. * +* Unpublished -- all rights reserved under the copyright laws * +* of the United States. * +********************************************************************/ + +#ifndef __BLACKLYNX_CHUNK_H__ +#define __BLACKLYNX_CHUNK_H__ + +#include +#include +#include + +namespace BlackLynx +{ +namespace SearchLynx +{ + +class Chunk +{ +public: + + Chunk( + const std::string& filename, + uint64_t chunkId, + uint32_t chunkSize, + uint32_t extraSize, + uint64_t fileOffset); + + virtual ~Chunk(); + + // NOTE: These are public (as they would be if this were a struct) so we + // don't have to deal with unnecessary getter/setter boilerplate + std::string m_filename; //!< name of file + uint64_t m_chunkId; //!< ID number of this chunk + uint32_t m_chunkSize; //!< number of bytes of data in this chunk + uint32_t m_extraSize; //!< how much extra data is tacked on to this chunk + uint64_t m_fileOffset; //!< absolute file offset of chunk's data + char* p_chunkData; //!< wrapped buffer for the chunk data + + void allocateBuffer(); + +private: + +}; + +} +} + +#endif // __BLACKLYNX_EXECDRIVER_H__ \ No newline at end of file diff --git a/software/read_test/src/ExecDriver.cc b/software/read_test/src/ExecDriver.cc new file mode 100644 index 0000000..3a3099b --- /dev/null +++ b/software/read_test/src/ExecDriver.cc @@ -0,0 +1,268 @@ +/******************************************************************** +* This software is "commercial computer software" as defined in the * +* Federal Acquisition Regulations and is subject to BlackLynx * +* Inc's standard End User License Agreement. * +* * +* CONFIDENTIAL - All source code is the propriety and confidential * +* information of BlackLynx Inc. * +* * +* Copyright 2014-2020 BlackLynx Inc. * +* Unpublished -- all rights reserved under the copyright laws * +* of the United States. * +********************************************************************/ + +#include "ExecDriver.h" +#include "Utilities/ChunkConstants.h" +#include "Utilities/FileUtils.h" +#include "Utilities/SearchLynxConfiguration.h" +#include "Utilities/BufferAllocator.h" + + +BlackLynx::SearchLynx::ExecDriver::ExecDriver( + const std::string& inputFile, + uint32_t numReaderThreads) + : r_inputFile(inputFile), + m_numReaderThreads(numReaderThreads), + DEFAULT_CHUNK_SIZE(SearchLynxConfiguration::getInstance()->getChunkSizeInKb() * 1024), + EXTRA_BYTES_TO_READ(RAW_TEXT_EXTRA_BYTES) +{ +} + +BlackLynx::SearchLynx::ExecDriver::~ExecDriver() +{ +} + +void BlackLynx::SearchLynx::ExecDriver::start() +{ + // Start threads in reverse order staring with cleanup thread + p_cleanupThread = new std::thread(&BlackLynx::SearchLynx::ExecDriver::cleanupThreadBody, + this, + &m_cleanupQueue); + + // Start specified number of reader threads + for (uint32_t idx = 0; idx < m_numReaderThreads; ++idx) + { + std::thread* thisThread = + new std::thread(&BlackLynx::SearchLynx::ExecDriver::readerThreadBody, + this, + &m_readerQueue, + &m_cleanupQueue); + m_readerThreads.push_back(thisThread); + } + + // Start setup thread body + p_setupThread = new std::thread(&BlackLynx::SearchLynx::ExecDriver::setupThreadBody, + this, + &m_readerQueue, + m_numReaderThreads); +} + +void BlackLynx::SearchLynx::ExecDriver::waitUntilComplete() +{ + // Join threads in normal order + + // Join setup thread + if (p_setupThread != nullptr) + { + if (p_setupThread->joinable()) + { + p_setupThread->join(); + } + } + delete p_setupThread; + p_setupThread = nullptr; + + // Join reader threads + for (auto readerThread : m_readerThreads) + { + if (readerThread->joinable()) + { + readerThread->join(); + } + delete readerThread; + } + m_readerThreads.clear(); + + // Join cleanup thread + if (p_cleanupThread != nullptr) + { + if (p_cleanupThread->joinable()) + { + p_cleanupThread->join(); + } + } + delete p_cleanupThread; + p_cleanupThread = nullptr; +} + +void BlackLynx::SearchLynx::ExecDriver::setupThreadBody( + chunk_queue_t* outQueue, + uint32_t numTerminators) +{ + uint64_t chunkIdAssigner = 1; + uint64_t fileOffset = 0; + uint32_t chunkSize = 0; + uint32_t extraSize = 0; + uint64_t fileSize = FileUtils::getFileSize(r_inputFile); + + while (fileOffset < fileSize) + { + // Determine chunk size + if ((fileSize - fileOffset) >= DEFAULT_CHUNK_SIZE) + { + chunkSize = DEFAULT_CHUNK_SIZE; + } + else + { + chunkSize = fileSize - fileOffset; + } + + // Determine extra size + if ((fileOffset + chunkSize + EXTRA_BYTES_TO_READ) <= fileSize) + { + extraSize = EXTRA_BYTES_TO_READ; + } + else + { + extraSize = fileSize - (fileOffset + chunkSize); + } + +// bool firstChunkInFile = (fileOffset == 0); +// bool lastChunkInFile = ((fileOffset + chunkSize) == fileSize); + + // Make chunk to process + auto chunk = new Chunk(r_inputFile, chunkIdAssigner, chunkSize, + extraSize, fileOffset); + + // Increment the chunk ID by 2 so that we match the record version + // of the execution driver + chunkIdAssigner += 2; + + fileOffset += chunkSize; + + // Send the chunk to the reader + outQueue->push_back(chunk); + } + + // Send the specified number of terminators + for (uint32_t idx = 0; idx < numTerminators; ++idx) + { + outQueue->push_back(nullptr); + } +} + +void BlackLynx::SearchLynx::ExecDriver::readerThreadBody( + chunk_queue_t* inQueue, + chunk_queue_t* outQueue) +{ + Chunk* chunk = nullptr; + + std::ifstream inFileStream; + std::string prevFilename(""); + + while (true) + { + chunk = inQueue->pop_front(); + + // Check for terminator + if (chunk == nullptr) + { + break; + } + + // Open file for reading if the previous filename is different than the + // current filename + if (chunk->m_filename != prevFilename) + { + if (inFileStream.is_open()) + { + inFileStream.close(); + } + + inFileStream.open(chunk->m_filename, std::ios::in | std::ios::binary); + if (! inFileStream) + { + BlnxLog(error) << "Unable to open file: " << chunk->m_filename; + break; + } + + // Tell the kernel this file will be read sequentially + int rc = posix_fadvise(FileUtils::getFdHack(inFileStream), 0, 0, + POSIX_FADV_SEQUENTIAL); + if (rc) + { + BlnxLog(error) << "posix_fadvise() returned: " << rc; + } + } + + // Store filename to compare against next chunk + prevFilename = chunk->m_filename; + + + 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; + + // Grab a buffer to hold the chunk data + chunk->allocateBuffer(); + + inFileStream.seekg(chunk->m_fileOffset); + inFileStream.read(chunk->p_chunkData, + chunk->m_chunkSize + chunk->m_extraSize); + + if (inFileStream.gcount() != (chunk->m_chunkSize + chunk->m_extraSize)) + { + // Log a runtime error + std::ostringstream msg; + msg << "Unable to read " + << (chunk->m_chunkSize + chunk->m_extraSize) + << " bytes at offset " << chunk->m_fileOffset << " from " + << chunk->m_filename << " (chunk ID: " << chunk->m_chunkId + << ")"; + BlnxLog(error) << msg.str(); + continue; + } + + outQueue->push_back(chunk); + } + + inFileStream.close(); + + // Send terminator to clean up thread. Note the cleanup thread counts the + // number of terminators and does not terminate until it receives all + // expected terminators. + m_cleanupQueue.push_back(nullptr); +} + +void BlackLynx::SearchLynx::ExecDriver::cleanupThreadBody( + chunk_queue_t* inQueue) +{ + uint32_t terminatorCount = 0; + Chunk* chunk = nullptr; + auto bufAllocator = BufferAllocator::getInstance(); + + while (true) + { + chunk = inQueue->pop_front(); + + // Check for terminator + if (chunk == nullptr) + { + // Check if we've received one terminator from each reader + if (++terminatorCount == m_numReaderThreads) + { + break; + } + else + { + continue; + } + } + + bufAllocator->release(chunk->p_chunkData); + delete chunk; + } +} \ No newline at end of file diff --git a/software/read_test/src/ExecDriver.h b/software/read_test/src/ExecDriver.h new file mode 100644 index 0000000..2df5a23 --- /dev/null +++ b/software/read_test/src/ExecDriver.h @@ -0,0 +1,74 @@ +/******************************************************************** +* This software is "commercial computer software" as defined in the * +* Federal Acquisition Regulations and is subject to BlackLynx * +* Inc's standard End User License Agreement. * +* * +* CONFIDENTIAL - All source code is the propriety and confidential * +* information of BlackLynx Inc. * +* * +* Copyright 2014-2020 BlackLynx Inc. * +* Unpublished -- all rights reserved under the copyright laws * +* of the United States. * +********************************************************************/ + +#ifndef __BLACKLYNX_EXECDRIVER_H__ +#define __BLACKLYNX_EXECDRIVER_H__ + +#include +#include +#include +#include + +#include "Chunk.h" +#include "Utilities/ProtectedAndSynchronizedQueue.h" + + +namespace BlackLynx +{ +namespace SearchLynx +{ + +class ExecDriver +{ +public: + + typedef ProtectedAndSynchronizedQueue chunk_queue_t; + + ExecDriver( + const std::string& inputFile, + uint32_t numReaderThreads); + + virtual ~ExecDriver(); + + virtual void start(); + + virtual void waitUntilComplete(); + + void setupThreadBody(chunk_queue_t* outQueue, uint32_t numTerminators); + + void cleanupThreadBody(chunk_queue_t* inQueue); + +protected: + + virtual void readerThreadBody( + chunk_queue_t* inQueue, + chunk_queue_t* outQueue); + + const std::string& r_inputFile; + uint32_t m_numReaderThreads; + + std::thread* p_setupThread; + std::vector m_readerThreads; + std::thread* p_cleanupThread; + + uint32_t DEFAULT_CHUNK_SIZE; + uint32_t EXTRA_BYTES_TO_READ; + + chunk_queue_t m_readerQueue; + chunk_queue_t m_cleanupQueue; +}; + +} +} + +#endif // __BLACKLYNX_EXECDRIVER_H__ \ No newline at end of file diff --git a/software/read_test/src/Logger/SearchLynxCLogger.cc b/software/read_test/src/Logger/SearchLynxCLogger.cc new file mode 100644 index 0000000..a4d4aa4 --- /dev/null +++ b/software/read_test/src/Logger/SearchLynxCLogger.cc @@ -0,0 +1,32 @@ +/******************************************************************** +* This software is "commercial computer software" as defined in the * +* Federal Acquisition Regulations and is subject to BlackLynx * +* Inc's standard End User License Agreement. * +* * +* CONFIDENTIAL - All source code is the propriety and confidential * +* information of BlackLynx Inc. * +* * +* Copyright 2014-2020 BlackLynx Inc. * +* Unpublished -- all rights reserved under the copyright laws * +* of the United States. * +********************************************************************/ + +#include + +#include "SearchLynxCLogger.h" +#include "SearchLynxLogger.h" + +void searchlynx_logError( + int logLevel, + const char* formatString, + ...) +{ + // used for reading in the variable message contents and formatting + va_list args; + va_start(args, formatString); + + char msg[2048]; + vsnprintf(&msg[0], 2048, formatString, args); + + BlnxLog(static_cast(logLevel)) << msg; +} diff --git a/software/read_test/src/Logger/SearchLynxCLogger.h b/software/read_test/src/Logger/SearchLynxCLogger.h new file mode 100644 index 0000000..c545848 --- /dev/null +++ b/software/read_test/src/Logger/SearchLynxCLogger.h @@ -0,0 +1,71 @@ +/******************************************************************** +* This software is "commercial computer software" as defined in the * +* Federal Acquisition Regulations and is subject to BlackLynx * +* Inc's standard End User License Agreement. * +* * +* CONFIDENTIAL - All source code is the propriety and confidential * +* information of BlackLynx Inc. * +* * +* Copyright 2014-2020 BlackLynx Inc. * +* Unpublished -- all rights reserved under the copyright laws * +* of the United States. * +********************************************************************/ + +#ifndef __BLACKLYNX_C_LOGGER__ +#define __BLACKLYNX_C_LOGGER__ + +#ifdef __cplusplus +extern "C" { +#endif + +enum SEARCHLYNX_LOG_LEVEL_TYPE +{ + LOG_LEVEL_ERROR = 0, + LOG_LEVEL_WARNING = 1, + LOG_LEVEL_INFO = 2, + LOG_LEVEL_DEBUG = 3, + LOG_LEVEL_TRACE = 4 +}; + +// NOTE: the ## __VA_ARGS__ is a gcc trick; see: +// https://gcc.gnu.org/onlinedocs/gcc/Variadic-Macros.html + +#define searchlynx_c_LOG_ERROR(format, ...) { \ + searchlynx_logError(LOG_LEVEL_ERROR, format, ## __VA_ARGS__); \ +} + +#define searchlynx_c_LOG_WARNING(format, ...) { \ + searchlynx_logError(LOG_LEVEL_WARNING, format, ## __VA_ARGS__); \ +} + +#define searchlynx_c_LOG_INFO(format, ...) { \ + searchlynx_logError(LOG_LEVEL_INFO, format, ## __VA_ARGS__); \ +} + +#define searchlynx_c_LOG_DEBUG(format, ...) { \ + searchlynx_logError(LOG_LEVEL_DEBUG, format, ## __VA_ARGS__); \ +} + +#define searchlynx_c_LOG_TRACE(format, ...) { \ + searchlynx_logError(LOG_LEVEL_TRACE, format, ## __VA_ARGS__); \ +} + +/** + * NOTE: the c_LOG_ERROR, c_LOG_WARNING, c_LOG_DEBUG, and c_LOG_INFO + * macros should be used instead of using this function directly! + * + * @param level the severity of the message + * @param fmt format string for the log message [variable arguments] + * + */ + +void searchlynx_logError( + int logLevel, + const char* formatString, + ...); + +#ifdef __cplusplus +} +#endif + +#endif diff --git a/software/read_test/src/Logger/SearchLynxLogger.cc b/software/read_test/src/Logger/SearchLynxLogger.cc new file mode 100644 index 0000000..38062fc --- /dev/null +++ b/software/read_test/src/Logger/SearchLynxLogger.cc @@ -0,0 +1,89 @@ +/******************************************************************** +* This software is "commercial computer software" as defined in the * +* Federal Acquisition Regulations and is subject to BlackLynx * +* Inc's standard End User License Agreement. * +* * +* CONFIDENTIAL - All source code is the propriety and confidential * +* information of BlackLynx Inc. * +* * +* Copyright 2014-2020 BlackLynx Inc. * +* Unpublished -- all rights reserved under the copyright laws * +* of the United States. * +********************************************************************/ +#include +#include +#include + +#include "SearchLynxLogger.h" +//#include "SearchLynxConfiguration.h" + +enum BlackLynx::SearchLynx::SEARCHLYNX_LOG_LEVEL default_level = BlackLynx::SearchLynx::debug; + +BlackLynx::SearchLynx::SearchLynxLogger::SearchLynxLogger() +{ + m_outputStream.str(""); + m_logFile.open(LOG_FILENAME, std::ios::out | std::ios::app); +} + +BlackLynx::SearchLynx::SearchLynxLogger::~SearchLynxLogger() +{ + m_logFile.close(); +} + +void BlackLynx::SearchLynx::SearchLynxLogger::log_msg( + enum SEARCHLYNX_LOG_LEVEL level, + const std::string& msg) +{ + if (level > default_level) + { + return; + } + + struct timeval tv; + struct tm *tm; + gettimeofday(&tv, NULL); + tm = localtime(&tv.tv_sec); + + char time_format[64], time_str[64]; + + const bool LOG_WITH_MILLISECONDS = true; + if (LOG_WITH_MILLISECONDS == true) + { + // Set format to include milliseconds + strftime(time_format, sizeof(time_format), "%a %b %d %H:%M:%S.%%03u %Y", + tm); + std::snprintf(time_str, sizeof(time_str), time_format, (tv.tv_usec / 1000)); + } + else + { + // Set format to not include milliseconds + strftime(time_str, sizeof(time_str), "%a %b %d %H:%M:%S %Y", tm); + } + + // NOTE: "COMPONENT_ID" must be defined somewhere, ideally at compile time + switch (level) + { + case error: + m_logFile << time_str << "|ERROR|" << COMPONENT_ID << "|" << msg << std::endl; + break; + + case warn: + m_logFile << time_str << "|WARNING|" << COMPONENT_ID << "|" << msg << std::endl; + break; + + case info: + m_logFile << time_str << "|INFO|" << COMPONENT_ID << "|" << msg << std::endl; + break; + + case debug: + m_logFile << time_str << "|DEBUG|" << COMPONENT_ID << "|" << msg << std::endl; + break; + + case trace: + m_logFile << time_str << "|TRACE|" << COMPONENT_ID << "|" << msg << std::endl; + break; + + default: + break; + } +} diff --git a/software/read_test/src/Logger/SearchLynxLogger.h b/software/read_test/src/Logger/SearchLynxLogger.h new file mode 100644 index 0000000..c9777bf --- /dev/null +++ b/software/read_test/src/Logger/SearchLynxLogger.h @@ -0,0 +1,178 @@ +/******************************************************************** +* This software is "commercial computer software" as defined in the * +* Federal Acquisition Regulations and is subject to BlackLynx * +* Inc's standard End User License Agreement. * +* * +* CONFIDENTIAL - All source code is the propriety and confidential * +* information of BlackLynx Inc. * +* * +* Copyright 2014-2020 BlackLynx Inc. * +* Unpublished -- all rights reserved under the copyright laws * +* of the United States. * +*********************************************************************/ + +#ifndef __BLACKLYNX_LOGGER__ +#define __BLACKLYNX_LOGGER__ + +#include +#include +#include +#include +#include + +namespace BlackLynx +{ +namespace SearchLynx +{ + +#ifdef CYBERLYNX +#define LOG_FILENAME "/var/log/blacklynx/cyberlynx.log" +#else +#define LOG_FILENAME "/var/log/blacklynx/searchlynx.log" +#endif + +// Enumeration representing the log severity levels +enum SEARCHLYNX_LOG_LEVEL {noop=-1, error=0, warn=1, info=2, debug=3, trace=4}; + +#define BlnxLog(sev) BlackLynx::SearchLynx::SearchLynxLogger::getInstance(sev).internalLog() \ + << &(__FILE__[BUILD_PATH_LEN + 1]) << ":" << __LINE__ << "|" + +class SearchLynxLogger : public std::ostringstream +{ +public: + + /** + * SearchLynxLogger Destructor + */ + ~SearchLynxLogger(); + + /** + * Set the log severity level + */ + void setSeverity(enum SEARCHLYNX_LOG_LEVEL sev) + { + m_severity = sev; + } + + /** + * lock mutex + */ + void lockMutex() + { + m_rloggerMutex.lock(); + } + + /** + * unlock mutex + */ + void unlockMutex() + { + m_rloggerMutex.unlock(); + } + + /** + * Send a log message to tlogd. + * + * @param level - the severity of the message being logged + * @param message - the message to send + */ + void log_msg(enum SEARCHLYNX_LOG_LEVEL level, const std::string& msg); + + /** + * Get the singleton instance + */ + static SearchLynxLogger& getInstance(enum SEARCHLYNX_LOG_LEVEL sev) + { + static SearchLynxLogger instance; + if (sev >= 0) + { + // Lock our mutex and set the severity level + instance.lockMutex(); + instance.setSeverity(sev); + } + return instance; + }; + + /** + * Nested class used to get around the "endl" tag, it automatically + * handles the "cleanup" at the end of the log message stream in the + * destructor + */ + class InternalLogger + { + public: + InternalLogger(enum SEARCHLYNX_LOG_LEVEL severity) + { + internal_sev = severity; + getInstance(noop); + } + + InternalLogger(const InternalLogger &other) + { + } + + ~InternalLogger() + { + std::string str = internal_msg_stream.str(); + if(! str.empty()) + { + getInstance(noop).log_msg(internal_sev, str); + + // Clean the output stream for future logging + internal_msg_stream.str(""); + + // Make sure we unlock! + getInstance(noop).unlockMutex(); + } + } + + /** + * Overriding the insertion operator to build m_outputStream + */ + template + InternalLogger& operator<< (const T& msg) + { + internal_msg_stream << msg; + return *this; + } + + private: + std::stringstream internal_msg_stream; + enum SEARCHLYNX_LOG_LEVEL internal_sev; + }; + + + InternalLogger internalLog() + { + return InternalLogger(m_severity); + } + +private: + + /** + * Private constructor (singleton) + */ + SearchLynxLogger(); + + /** + * The log message to write + */ + std::ostringstream m_outputStream; + + /** + * Severity level of the log message + */ + enum SEARCHLYNX_LOG_LEVEL m_severity; + + /** + * Mutex for making this singleton thread-safe + */ + std::mutex m_rloggerMutex; + + std::ofstream m_logFile; +}; + +} // namespace SearchLynx +} // namespace BlackLynx + +#endif // __BLACKLYNX_RYFX_LOGGER__ diff --git a/software/read_test/src/Utilities/BufferAllocator.cc b/software/read_test/src/Utilities/BufferAllocator.cc new file mode 100644 index 0000000..c6b3eea --- /dev/null +++ b/software/read_test/src/Utilities/BufferAllocator.cc @@ -0,0 +1,386 @@ +/******************************************************************** +* This software is "commercial computer software" as defined in the * +* Federal Acquisition Regulations and is subject to BlackLynx * +* Inc's standard End User License Agreement. * +* * +* CONFIDENTIAL - All source code is the propriety and confidential * +* information of BlackLynx Inc. * +* * +* Copyright 2014-2020 BlackLynx Inc. * +* Unpublished -- all rights reserved under the copyright laws * +* of the United States. * +********************************************************************/ +#include +#include +#include +#include + +#include "Utilities/SearchLynxConfiguration.h" +#include "Utilities/BufferAllocator.h" +#include "Utilities/ChunkConstants.h" +#include "Utilities/NumberUtils.h" +//#include "Utilities/ResourceUtils.h" +#include "Utilities/SystemArchInfo.h" +//#include "Utilities/TranslatedChunkManager.h" + + +BlackLynx::SearchLynx::BufferAllocator* BlackLynx::SearchLynx::BufferAllocator::getInstance() +{ + static BlackLynx::SearchLynx::BufferAllocator theInstance; + return &theInstance; +} + +BlackLynx::SearchLynx::BufferAllocator::BufferAllocator() + : m_allocationSize(0), + p_allocationThread(nullptr), + p_deallocationThread(nullptr), + m_allocationRequested(false), + m_deallocationInited(false), + m_deallocationRequested(false), + m_buffersAllocated(false), + m_memorySizeBytes(get_phys_pages() * getpagesize()) +{ +} + +BlackLynx::SearchLynx::BufferAllocator::~BufferAllocator() +{ + // Before we can terminate we need to make sure that the allocation thread + // has actually finished. Otherwise we terminate with a thread still + // running and trying to do allocations + if (m_allocationRequested) + { + bool terminated = false; + while (! m_buffersAllocated.getValue(terminated)); + + std::lock_guard lock(m_mutex); + for (auto& bufferSizeIterator : m_extraAllocatedBuffers) + { + BlnxLog(debug) << "Number of additional buffers allocated: " + << "[" << bufferSizeIterator.first << "] " + << bufferSizeIterator.second; + } + } + + if (p_allocationThread != nullptr) + { + if (p_allocationThread->joinable()) + { + p_allocationThread->join(); + } + delete p_allocationThread; + p_allocationThread = nullptr; + } + + if (p_deallocationThread != nullptr) + { + if (p_deallocationThread->joinable()) + { + p_deallocationThread->join(); + } + delete p_deallocationThread; + p_deallocationThread = nullptr; + } + + // Make sure any last deallocation that's needed happens + m_deallocationRequested = true; + deallocateBuffers(); +} + +void BlackLynx::SearchLynx::BufferAllocator::allocateChunkBuffers() +{ + std::lock_guard lock(m_mutex); + + // Clear flag + m_deallocationRequested = false; + + // Allocate the chunk buffers that we need + uint32_t numberOfBuffers = 0; + uint32_t bufferSize = 0; + + auto SLC = BlackLynx::SearchLynx::SearchLynxConfiguration::getInstance(); + uint32_t chunkSizeInKB = SLC->getChunkSizeInKb() + + (BlackLynx::SearchLynx::RAW_TEXT_EXTRA_BYTES / 1024) + 1; + + // For the initial buffer allocation use the minimum of the fixed size + // (16 GB) one quarter of the system's RAM. + auto SysArch = SystemArchInfo::getInstance(); + uint64_t memSizeKb = std::min((SysArch->getRamSizeBytes() / 4) / 1024, + CHUNK_BUFFER_MEM_SIZE_KB); + + // Divide this space by the chunk size to determine the initial number of + // buffers to allocate. + numberOfBuffers = memSizeKb / chunkSizeInKB; + bufferSize = chunkSizeInKB * 1024; + + BlnxLog(debug) << "Allocating full chunk buffers: " << numberOfBuffers + << " of size " << bufferSize; + + allocateBuffers(numberOfBuffers, bufferSize); + + // Now allocate the same number of smaller buffers. These are really + // designed for use when in record/field mode and we need to stitch together + // a new chunk from two existing ones + allocateBuffers(numberOfBuffers, getpagesize()); + + // Indicate that the buffers have been allocated + m_buffersAllocated.setValue(true); +} + +void BlackLynx::SearchLynx::BufferAllocator::allocateChunkBuffersWithSize( + uint64_t numberOfBuffers, + uint32_t bufferSize) +{ + std::lock_guard lock(m_mutex); + + // Clear flag + m_deallocationRequested = false; + + BlnxLog(debug) << "Allocating full chunk buffers: " << numberOfBuffers + << " of size " << bufferSize; + + allocateBuffers(numberOfBuffers, bufferSize); + + // Indicate that the buffers have been allocated + m_buffersAllocated.setValue(true); +} + +void BlackLynx::SearchLynx::BufferAllocator::allocateBuffers( + uint32_t numberToAllocate, + uint32_t allocationSize) +{ + uint32_t allocSizeOrig = allocationSize; + uint32_t remainder = allocationSize % getpagesize(); + if (remainder != 0) + { + allocationSize += (getpagesize() - remainder); + BlnxLog(debug) << "Rounding allocation size " << allocSizeOrig + << " up to " << allocationSize; + } + + uint32_t actualNumberToAllocate = numberToAllocate; + auto iter = m_availableBuffers.find(allocationSize); + if (iter != m_availableBuffers.end()) + { + auto& bufferSet(iter->second); + if (bufferSet.size() >= actualNumberToAllocate) + { + // Bail out if we already have enough buffers of this size + return; + } + + actualNumberToAllocate -= bufferSet.size(); + } + + // Allocate new buffers + for (uint32_t idx = 0; idx < actualNumberToAllocate; ++idx) + { + char* thisBuffer = (char*) aligned_alloc(getpagesize(), allocationSize); + m_allocatedBuffers[thisBuffer] = allocationSize; + m_availableBuffers[allocationSize].insert(thisBuffer); + } + + m_extraAllocatedBuffers[allocationSize] = 0; +} + +void BlackLynx::SearchLynx::BufferAllocator::deallocateBuffers() +{ + std::lock_guard lock(m_mutex); + + // Check if the m_deallocationRequested flag has been cleared (because an + // allocation started before deallocation could happen) + if (! m_deallocationRequested) + { + return; + } + + m_deallocationInited = true; + + for (auto& bufferSizeIterator : m_availableBuffers) + { + for (auto& bufferIterator : bufferSizeIterator.second) + { + delete bufferIterator; + } + } + + m_allocatedBuffers.clear(); + m_availableBuffers.clear(); + + m_deallocationRequested = false; + m_deallocationInited = false; +} + +char* BlackLynx::SearchLynx::BufferAllocator::request( + uint32_t bufferSize, + bool waitIfNoneAvailable) +{ + // Wait until the buffers have been allocated + bool terminated = false; + while (! m_buffersAllocated.getValue(terminated)); + + std::lock_guard lock(m_mutex); + + // Search through the list of available buffers and find the smallest buffer + // size that will satisfy the request + char* buffer = nullptr; + for (auto& bufferSizeIterator : m_availableBuffers) + { + if (bufferSize <= bufferSizeIterator.first) + { + // This is the smallest buffer size that can be used. If a buffer + // is available, return it. If not, allocate a new one. + if (bufferSizeIterator.second.empty() && waitIfNoneAvailable) + { + while (bufferSizeIterator.second.empty()) + { + m_mutex.unlock(); + std::this_thread::sleep_for(std::chrono::milliseconds(3)); + m_mutex.lock(); + } + } + + if (bufferSizeIterator.second.empty()) + { + // We couldn't find a buffer in the bin so we need to allocate + // one. Round the size up to the nearest megabyte and allocate + // a new buffer of that size. + uint32_t roundedSize = NumberUtils::roundUp1M(bufferSize); + buffer = (char*) aligned_alloc(getpagesize(), roundedSize); + if (buffer == nullptr) + { + BlnxLog(error) << "aligned_alloc returned null -- size: " + << bufferSize << " roundedSize: " + << roundedSize << " -- errno: " << errno; + } + + m_allocatedBuffers[buffer] = roundedSize; + + // Make sure there is a new "bin" to hold buffers of this size + if (m_availableBuffers.find(roundedSize) == m_availableBuffers.end()) + { + m_availableBuffers[roundedSize]; + } + + // Keep track of how many times we do this for debug + m_extraAllocatedBuffers[roundedSize]++; + + BlnxLog(debug) << "Allocated new buffer of size: " << roundedSize + << " -- total: " << m_extraAllocatedBuffers[roundedSize]; + } + else + { + // We've got a buffer, remove it from the list and return it + auto it = --bufferSizeIterator.second.end(); + buffer = *it; + bufferSizeIterator.second.erase(it); + } + break; + } + } + + if (buffer == nullptr) + { + BlnxLog(error) << "Couldn't allocate buffer of size " << bufferSize + << " (waitIfNoneAvailable: " << int(waitIfNoneAvailable) + << ")"; + } + return buffer; +} + +void BlackLynx::SearchLynx::BufferAllocator::release(char* buffer) +{ + // Deallocation has already been inited so don't bother trying to release + if (m_deallocationInited || buffer == nullptr) + { + return; + } + + std::lock_guard lock(m_mutex); + + // Find the buffer in the list of allocated buffers to find the size of the + // allocated buffer + bool bufferReturned = false; + if (m_allocatedBuffers.find(buffer) != m_allocatedBuffers.end()) + { + uint64_t bufferSize = m_allocatedBuffers[buffer]; + m_availableBuffers[bufferSize].insert(buffer); + bufferReturned = true; + } + + if (! bufferReturned) + { + BlnxLog(error) << "Buffer " << (uint64_t) buffer << " not returned"; + } +} + +void BlackLynx::SearchLynx::BufferAllocator::initAllocation() +{ + { + std::lock_guard lock(m_mutex); + m_allocationRequested = true; + } + + // Clean up any prior allocation thread; this can happen if a process + // runs multiple searches + if (p_allocationThread != nullptr) + { + if (p_allocationThread->joinable()) + { + p_allocationThread->join(); + } + delete p_allocationThread; + p_allocationThread = nullptr; + } + + p_allocationThread = new std::thread( + &BlackLynx::SearchLynx::BufferAllocator::allocateChunkBuffers, this); +} + +void BlackLynx::SearchLynx::BufferAllocator::initAllocation( + uint64_t numberOfBuffers, + uint32_t bufferSize) +{ + if (! m_allocationRequested) + { + m_allocationRequested = true; + + // Clean up any prior allocation thread; this can happen if a process + // runs multiple searches + if (p_allocationThread != nullptr) + { + if (p_allocationThread->joinable()) + { + p_allocationThread->join(); + } + delete p_allocationThread; + p_allocationThread = nullptr; + } + + p_allocationThread = new std::thread( + &BlackLynx::SearchLynx::BufferAllocator::allocateChunkBuffersWithSize, + this, numberOfBuffers, bufferSize); + } +} + +void BlackLynx::SearchLynx::BufferAllocator::initDeallocation() +{ + { + std::lock_guard lock(m_mutex); + m_deallocationRequested = true; + } + + // Clean up any prior deallocation thread; this can happen if a process + // runs multiple searches + if (p_deallocationThread != nullptr) + { + if (p_deallocationThread->joinable()) + { + p_deallocationThread->join(); + } + delete p_deallocationThread; + p_deallocationThread = nullptr; + } + + p_deallocationThread = new std::thread( + &BlackLynx::SearchLynx::BufferAllocator::deallocateBuffers, this); +} diff --git a/software/read_test/src/Utilities/BufferAllocator.h b/software/read_test/src/Utilities/BufferAllocator.h new file mode 100644 index 0000000..2985354 --- /dev/null +++ b/software/read_test/src/Utilities/BufferAllocator.h @@ -0,0 +1,148 @@ +/******************************************************************** +* This software is "commercial computer software" as defined in the * +* Federal Acquisition Regulations and is subject to BlackLynx * +* Inc's standard End User License Agreement. * +* * +* CONFIDENTIAL - All source code is the propriety and confidential * +* information of BlackLynx Inc. * +* * +* Copyright 2014-2020 BlackLynx Inc. * +* Unpublished -- all rights reserved under the copyright laws * +* of the United States. * +********************************************************************/ + +#ifndef __BLACKLYNX_BUFFERALLOCATOR_H__ +#define __BLACKLYNX_BUFFERALLOCATOR_H__ + +#include +#include +#include +#include +#include +#include +#include + +#include "Utilities/ProtectedAndSynchronizedQueue.h" +#include "Utilities/SynchronizedValue.h" + +namespace BlackLynx +{ +namespace SearchLynx +{ + +/** + * This class exists to work around the new/delete "slowness" issue that got + * introduced in recent Linux kernels to combat the "meltdown" attack. + * + * Instead of allocating and deleting buffers as needed in the traditional way, + * we allocate a set of buffers at program startup and "request" and "return" + * them instead of "new" and "delete. That way the "returned" buffers can be + * reused as "new" buffers without having to pay the extra overhead penalty of + * a new/delete operation. + */ +class BufferAllocator +{ +public: + + // Free memory threshold 1 GB + static const uint64_t FREE_MEM_THRESHOLD_BYTES{0x80000000}; + + // 16 GB chunk buffer memory is the default for initial allocation + const uint64_t CHUNK_BUFFER_MEM_SIZE_KB = 16 * 1024 * 1024; + + /** + * Get the instance handle + * + * @return the instance handle + */ + static BufferAllocator* getInstance(); + + /** + * Deallocate any previously allocated buffers. This routine should be + * called prior to program exit. + */ + void deallocateBuffers(); + + /** + * Request a buffer; this is the equivalent of new(). + * + * @param bufferSize the size of the requested buffer + * + * @return the allocated buffer + */ + char* request(uint32_t bufferSize, bool waitIfNoneAvailable=false); + + /** + * Release a buffer; this is the equivalent of delete(). + * + * @param buffer the buffer to be released + */ + void release(char* buffer); + + /** + * Initialize allocation and a background thread and then return; calls to + * request will block until the allocation thread had terminated. + */ + void initAllocation(); + void initAllocation(uint64_t numberOfBuffers, uint32_t bufferSize); + + void initDeallocation(); + +private: + + /* + * Class constructor and destructor + */ + BufferAllocator(); + ~BufferAllocator(); + + void allocateChunkBuffers(); + void allocateChunkBuffersWithSize(uint64_t numberOfBuffers, uint32_t bufferSize); + + /** + * Allocate a set of buffers. This method must be called prior to + * any requests being made. + * + * @param numberToAllocate the number of buffers + * @param allocation size the size of the buffers + */ + void allocateBuffers(uint32_t numberToAllocate, uint32_t allocationSize); + + /** + * Queue of available buffers + */ + std::map< + char*, // list of buffers + uint32_t // buffer size + > m_allocatedBuffers; + + std::map< + uint32_t, // buffer size + std::set // list of buffers + > m_availableBuffers; + + std::mutex m_mutex; + + std::map< + uint32_t, // buffer size + uint32_t // newly allocated buffer count + > m_extraAllocatedBuffers; + + uint32_t m_allocationSize; + std::thread* p_allocationThread; + + std::thread* p_deallocationThread; + + std::atomic m_allocationRequested; + std::atomic m_deallocationInited; + std::atomic m_deallocationRequested; + SynchronizedValue m_buffersAllocated; + + const uint64_t m_memorySizeBytes; +}; + + +} // namespace SearchLynx +} // namespace BlackLynx + +#endif diff --git a/software/read_test/src/Utilities/ChunkConstants.h b/software/read_test/src/Utilities/ChunkConstants.h new file mode 100644 index 0000000..4584127 --- /dev/null +++ b/software/read_test/src/Utilities/ChunkConstants.h @@ -0,0 +1,30 @@ +/******************************************************************** +* This software is "commercial computer software" as defined in the * +* Federal Acquisition Regulations and is subject to BlackLynx * +* Inc's standard End User License Agreement. * +* * +* CONFIDENTIAL - All source code is the propriety and confidential * +* information of BlackLynx Inc. * +* * +* Copyright 2014-2020 BlackLynx Inc. * +* Unpublished -- all rights reserved under the copyright laws * +* of the United States. * +********************************************************************/ + +#ifndef __BLACKLYNX_CHUNKCONSTANTS_H__ +#define __BLACKLYNX_CHUNKCONSTANTS_H__ + +#include + + +namespace BlackLynx +{ +namespace SearchLynx +{ + +const uint32_t RAW_TEXT_EXTRA_BYTES = 64 * 1024; // 64 KB + +} +} + +#endif diff --git a/software/read_test/src/Utilities/FileUtils.cc b/software/read_test/src/Utilities/FileUtils.cc new file mode 100644 index 0000000..41e73b6 --- /dev/null +++ b/software/read_test/src/Utilities/FileUtils.cc @@ -0,0 +1,542 @@ +/******************************************************************** +* This software is "commercial computer software" as defined in the * +* Federal Acquisition Regulations and is subject to BlackLynx * +* Inc's standard End User License Agreement. * +* * +* CONFIDENTIAL - All source code is the propriety and confidential * +* information of BlackLynx Inc. * +* * +* Copyright 2014-2020 BlackLynx Inc. * +* Unpublished -- all rights reserved under the copyright laws * +* of the United States. * +********************************************************************/ + +#include +#include +#include +#include +#include + +#include +#include +#include +#include + +#include "Logger/SearchLynxLogger.h" +#include "Utilities/FileUtils.h" + +#include "boost/filesystem.hpp" +#include + +// Alias namespace to improve readability +namespace bfs = boost::filesystem; + +static const int WHITESPACE_CMPISTRI_FLAGS = (_SIDD_UBYTE_OPS | + _SIDD_CMP_EQUAL_ANY | + _SIDD_LEAST_SIGNIFICANT | + _SIDD_NEGATIVE_POLARITY); + +uint64_t BlackLynx::SearchLynx::FileUtils::getAverageInputFileSize( + std::vector& fileList) +{ + uint64_t totalFileSize = 0; + for (auto thisFilename : fileList) + { + totalFileSize += getFileSize(thisFilename); + } + + return totalFileSize / fileList.size(); +} + +bool BlackLynx::SearchLynx::FileUtils::isFileReadable(const std::string& filepath) +{ + return (access(filepath.c_str(), R_OK) == 0); +} + +bool BlackLynx::SearchLynx::FileUtils::areFilesReadable( + std::vector& fileList) +{ + bool filesAreReadable = true; + + for (auto& thisFilename : fileList) + { + int rval = access(thisFilename.c_str(), R_OK); + filesAreReadable &= (rval == 0); + } + + return filesAreReadable; +} + +uint64_t BlackLynx::SearchLynx::FileUtils::getFileSize(const std::string& filename) +{ + uint64_t fileSize = 0; + struct stat statBuffer; + + int returnValue = stat(filename.c_str(), &statBuffer); + if (returnValue == 0) + { + fileSize = statBuffer.st_size; + } + + return fileSize; +} + +uint64_t BlackLynx::SearchLynx::FileUtils::getTotalBytesInFiles( + std::vector& fileList, + bool removeZeroLengthFilenames) +{ + uint64_t totalSizeInBytes = 0; + uint32_t fileIndex = 0; + while (fileIndex < fileList.size()) + { + struct stat statBuffer; + int returnValue = stat(fileList[fileIndex].c_str(), &statBuffer); + if (returnValue == 0) + { + totalSizeInBytes += statBuffer.st_size; + if (removeZeroLengthFilenames && (statBuffer.st_size == 0)) + { + fileList.erase(fileList.begin() + fileIndex); + } + else + { + fileIndex++; + } + } + else + { + totalSizeInBytes = 0; + fileIndex++; + } + } + + return totalSizeInBytes; +} + +char* BlackLynx::SearchLynx::FileUtils::readBytesFromFileOffset( + std::string filename, + uint64_t startOffset, + uint64_t lengthInBytes, + uint64_t& actualBytesRead) +{ + char* buffer = new char[lengthInBytes + 1]; + + if (buffer != nullptr) + { + std::ifstream inputFile(filename, std::ios::in | std::ios::binary); + if (inputFile.is_open()) + { + inputFile.seekg(startOffset, std::ios::beg); + inputFile.read(buffer, lengthInBytes); + actualBytesRead = inputFile.gcount(); + inputFile.close(); + } + } + else + { + BlnxLog(error) << "Unable to allocate buffer!"; + actualBytesRead = 0; + } + buffer[lengthInBytes] = '\0'; + + return buffer; +} + +uint64_t BlackLynx::SearchLynx::FileUtils::readBytesFromFileOffset( + std::string filename, + uint64_t startOffset, + uint64_t lengthInBytes, + void* destinationBuffer) +{ + uint64_t actualBytesRead = 0; + + std::ifstream inputFile(filename, std::ios::in | std::ios::binary); + if (inputFile.is_open()) + { + inputFile.seekg(startOffset, std::ios::beg); + inputFile.read((char*) destinationBuffer, lengthInBytes); + actualBytesRead = inputFile.gcount(); + inputFile.close(); + } + + return actualBytesRead; +} + +uint64_t BlackLynx::SearchLynx::FileUtils::readBytesFromFileOffset( + std::ifstream& inputFile, + uint64_t startOffset, + uint64_t lengthInBytes, + void* destinationBuffer) +{ + uint64_t actualBytesRead = 0; + + if (inputFile.is_open()) + { + inputFile.seekg(startOffset, std::ios::beg); + inputFile.read((char*) destinationBuffer, lengthInBytes); + actualBytesRead = inputFile.gcount(); + } + + return actualBytesRead; +} + +void BlackLynx::SearchLynx::FileUtils::readFileListFromFile( + const std::string& filename, + std::vector* fileList) +{ + uint64_t fileSize = BlackLynx::SearchLynx::FileUtils::getFileSize(filename); + std::ifstream inputStream(filename, std::ios::binary); + if (!inputStream) + { + std::string errorText = "File " + filename + " cannot be opened"; + BlnxLog(warn) << errorText; + return; + } + + uint64_t extraData = 16 - (fileSize % 16); + char* fileBuffer = new char [fileSize + extraData]; + inputStream.read(fileBuffer, fileSize); + inputStream.close(); + std::memset(fileBuffer + fileSize, '\0', extraData); + + std::string delimiterList("\r\n"); + char* rawFilename = fileBuffer; + char* delimiter = findDelimiter(rawFilename, fileBuffer + fileSize - rawFilename, + delimiterList); + while (delimiter != nullptr) + { + std::string filename(rawFilename, delimiter - rawFilename); + boost::algorithm::trim(filename); + + if (filename.length() > 0) + { + fileList->push_back(filename); // push everything into the file list + // NOTE: file glob expansion now happens in the prepareInputFiles() + // function + } + rawFilename = delimiter + 1; + delimiter = findDelimiter(rawFilename, fileBuffer + fileSize - rawFilename, + delimiterList); + } + + delete[] fileBuffer; +} + +uint32_t BlackLynx::SearchLynx::FileUtils::estimateNumberChunks( + const std::vector& fileList, + uint32_t chunkSize, + bool recordBased) +{ + uint32_t numChunks = 0; + uint32_t localChunks = 0; + struct stat statBuffer; + + for (auto& filePath : fileList) + { + if (stat(filePath.c_str(), &statBuffer) == 0) + { + localChunks = (statBuffer.st_size / chunkSize); + numChunks += (recordBased ? (2 * localChunks) : localChunks); + + if (statBuffer.st_size % chunkSize) + { + ++numChunks; + } + } + } + + return numChunks; +} + +int BlackLynx::SearchLynx::FileUtils::readDictionary( + const std::string& filename, + std::vector& dictionaryEntries) +{ + uint64_t fileSize = BlackLynx::SearchLynx::FileUtils::getFileSize(filename); + std::ifstream inputStream(filename, std::ios::binary); + if (! inputStream) + { + BlnxLog(error) << "Dictionary file " << filename << " cannot be opened"; + return -1; + } + + uint64_t extraData = 16 - (fileSize % 16); + char* fileBuffer = new char [fileSize + extraData]; + inputStream.read(fileBuffer, fileSize); + fileBuffer[fileSize] = '\0'; + inputStream.close(); + + for (uint64_t offset = fileSize; offset < fileSize + extraData; offset++) + { + fileBuffer[offset] = '\0'; + } + + std::string delimiterList("\r\n"); + char* thisEntry = fileBuffer; + char* delimiter = findDelimiter(thisEntry, + fileBuffer + fileSize - thisEntry, + delimiterList); + while (delimiter != nullptr) + { + std::string entryText(thisEntry, delimiter - thisEntry); + boost::algorithm::trim(entryText); + + if (entryText.length() > 0) + { + dictionaryEntries.push_back(entryText); + } + thisEntry = delimiter + 1; + delimiter = findDelimiter(thisEntry, + fileBuffer + fileSize - thisEntry, + delimiterList); + } + + if (thisEntry < (fileBuffer + fileSize)) + { + // The file didn't end in a line feed, so grab the last entry + std::string entryText(thisEntry, fileBuffer + fileSize - thisEntry); + boost::algorithm::trim(entryText); + + if (entryText.length() > 0) + { + dictionaryEntries.push_back(entryText); + } + } + + delete [] fileBuffer; + return 0; +} + +void BlackLynx::SearchLynx::FileUtils::readFilesRecursively( + const std::string& filepathStr, + std::vector& fileList) +{ + if (filepathStr.empty()) + { + return; + } + + bfs::path filepath(filepathStr); + bfs::path filename = filepath.filename(); + bfs::path basepath = filepath.parent_path(); + bool hasWildcards = (filename.string().find_first_of("*?") != std::string::npos); + + // If no actual filename was provided then there is nothing to do + if (filename.string() == ".") + { + return; + } + + // Incorporate the same logic as below for the input file path + if (hasWildcards || bfs::exists(filepathStr)) + { + fileList.push_back(filepathStr); + } + + // Before we attempt to merrily iterate along, make sure that the basepath + // is in fact a valid directory + if (! bfs::is_directory(basepath)) + { + return; + } + + for (bfs::recursive_directory_iterator endIter, dirIter(basepath); + dirIter != endIter; ++dirIter) + { + if (bfs::is_directory(*dirIter)) + { + if (hasWildcards) + { + // The filename had wildcards so we're just going to append it + // to the directory path and call it a day. Note the idea here + // is that prepareInputFiles() will expand wildcards so we don't + // need to duplicate that logic here. + fileList.push_back((*dirIter / filename).string()); + } + else + { + // The filename did not contain wildcards so we need to check if + // the file exists before adding it to the output. + if (bfs::exists(*dirIter / filename)) + { + fileList.push_back((*dirIter / filename).string()); + } + } + } + } +} + +char* BlackLynx::SearchLynx::FileUtils::findUTF8Codepoint( + char* buffer, + uint64_t bufferSize) +{ + if ((bufferSize == 0) || (buffer == nullptr)) + { + return nullptr; + } + + static const int CMPISTRI_FLAGS = (_SIDD_UBYTE_OPS | + _SIDD_CMP_RANGES | + _SIDD_LEAST_SIGNIFICANT); + + static const unsigned char UTF8_CPMSB_CHAR_RANGE[16] = + {1, 1, 1, 1, 1, 1, 1, 1, 1, 1, + 0xC0, 0xDF, // 2 byte + 0xE0, 0xEF, // 3 byte + 0xF0, 0xF7}; // 4 byte + + const __m128i UTF8_MSB_Chars = + _mm_lddqu_si128(reinterpret_cast(UTF8_CPMSB_CHAR_RANGE)); + + uint64_t offset = 0; + if (bufferSize >= 16) + { + // We can only use the _mm_cmpistri() ("fast search") on 16 byte aligned + // data so calculate the size we can search in this manner and the + // remaining portion will be searched manually below. Note the integer + // division is intended and both operations should be converted into + // the appropriate shifts by the compiler. + uint32_t fastSearchSize = (bufferSize / 16) * 16; + + while (offset < fastSearchSize) + { + const __m128i data = _mm_lddqu_si128(reinterpret_cast(buffer + offset)); + const int index = _mm_cmpistri(UTF8_MSB_Chars, data, CMPISTRI_FLAGS); + if (index < 16) + { + // we found a delimiter + return buffer + offset + index; + } + + offset += 16; + } + } + + // If the input isn't a multiple of 16 bytes, we can't use the _mm + // instructions. Instead do a basic search of the last 16 or fewer + // characters. + while (offset < bufferSize) + { + if ((buffer[offset] >= UTF8_CPMSB_CHAR_RANGE[10]) && + (buffer[offset] <= UTF8_CPMSB_CHAR_RANGE[11])) + { + return buffer + offset; + } + else if ((buffer[offset] >= UTF8_CPMSB_CHAR_RANGE[12]) && + (buffer[offset] <= UTF8_CPMSB_CHAR_RANGE[13])) + { + return buffer + offset; + } + else if ((buffer[offset] >= UTF8_CPMSB_CHAR_RANGE[14]) && + (buffer[offset] <= UTF8_CPMSB_CHAR_RANGE[15])) + { + return buffer + offset; + } + + ++offset; + } + + return nullptr; +} + +unsigned char BlackLynx::SearchLynx::FileUtils::getFirstCharacterInInputFile( + const std::string& inputFile) +{ + const char RAW_WHITESPACE_CHARS[16] = {' ', '\t', '\n', '\r', '\f', '\v'}; + const __m128i m_whitespaceChars = + _mm_lddqu_si128(reinterpret_cast(RAW_WHITESPACE_CHARS)); + // First we need to find the first non-whitespace character in the input + // file to guess at what the file type is + unsigned char firstCharacterInFile = ' '; + + // Get input file size + const uint64_t fileSize = BlackLynx::SearchLynx::FileUtils::getFileSize(inputFile); + + std::ifstream inputFileStream(inputFile, std::ios::in | std::ios::binary); + if (inputFileStream.is_open()) + { + uint64_t offset = 0; + unsigned char buffer[16]; + + // Iterate over the file + while (offset < fileSize) + { + inputFileStream.seekg(offset, std::ios::beg); + inputFileStream.read((char*) buffer, sizeof(buffer)); + + // Search for the first non whitespace character + const __m128i data = _mm_lddqu_si128(reinterpret_cast(buffer)); + const int index = _mm_cmpistri(m_whitespaceChars, data, + WHITESPACE_CMPISTRI_FLAGS); + + if (index < 16) + { + firstCharacterInFile = buffer[index]; + break; + } + + offset += 16; + } + + inputFileStream.close(); + } + + return firstCharacterInFile; +} + +unsigned char BlackLynx::SearchLynx::FileUtils::getLastCharacterInInputFile( + const std::string& inputFile) +{ + const char RAW_WHITESPACE_CHARS[16] = {' ', '\t', '\n', '\r', '\f', '\v'}; + const __m128i m_whitespaceChars = + _mm_lddqu_si128(reinterpret_cast(RAW_WHITESPACE_CHARS)); + // First we need to find the first non-whitespace character in the input + // file to guess at what the file type is + unsigned char lastChar = ' '; + + // Get input file size + const uint64_t fileSize = BlackLynx::SearchLynx::FileUtils::getFileSize(inputFile); + + std::ifstream inputFileStream(inputFile, std::ios::in | std::ios::binary); + if (inputFileStream.is_open()) + { + int64_t offset = 0; + if (fileSize > 16) + { + offset = fileSize - 16; + } + unsigned char buffer[16]; + + // Iterate over the file backwards in chunks of 16 bytes + while (offset >= 0) + { + inputFileStream.seekg(offset, std::ios::beg); + inputFileStream.read((char*) buffer, sizeof(buffer)); + + // If the input file is tiny back fill the last character so we can + // use the string compare intrinsic on a full 16 bytes + if (fileSize < 16) + { + std::memset(buffer + fileSize, buffer[fileSize - 1], 16 - fileSize); + } + + // Search for the first non whitespace character + const __m128i data = _mm_lddqu_si128(reinterpret_cast(buffer)); + const int index = _mm_cmpistri(m_whitespaceChars, data, + WHITESPACE_CMPISTRI_FLAGS | + _SIDD_MOST_SIGNIFICANT); + + if (index < 16) + { + lastChar = buffer[index]; + break; + } + + offset -= 16; + } + + inputFileStream.close(); + } + + return lastChar; +} diff --git a/software/read_test/src/Utilities/FileUtils.h b/software/read_test/src/Utilities/FileUtils.h new file mode 100644 index 0000000..26b61dc --- /dev/null +++ b/software/read_test/src/Utilities/FileUtils.h @@ -0,0 +1,357 @@ +/******************************************************************** +* This software is "commercial computer software" as defined in the * +* Federal Acquisition Regulations and is subject to BlackLynx * +* Inc's standard End User License Agreement. * +* * +* CONFIDENTIAL - All source code is the propriety and confidential * +* information of BlackLynx Inc. * +* * +* Copyright 2014-2020 BlackLynx Inc. * +* Unpublished -- all rights reserved under the copyright laws * +* of the United States. * +********************************************************************/ + +#ifndef __BLACKLYNX_FILEUTILS_H__ +#define __BLACKLYNX_FILEUTILS_H__ + +#include +#include +#include +#include + +#include +#include +#include +#include + +#include + + +namespace BlackLynx +{ +namespace SearchLynx +{ + +class FileUtils +{ +public: + /** + * Check whether a file exists + * + * @param name the filename to check + * + * @return TRUE is the file exists, FALSE is not + */ + static inline bool exists(const std::string& name) + { + struct stat buffer; + return (stat (name.c_str(), &buffer) == 0); + } + + /** + * Get the file size + * + * @param fileList the list of files + * + * @return the file size + */ + static uint64_t getFileSize(const std::string& filename); + + /** + * Get average file size + * + * @param fileList the list of files + * + * @return the average file size + */ + static uint64_t getAverageInputFileSize(std::vector& fileList); + + /** + * Get the total number of bytes that exist in the list of files + * + * @param fileList the list of files + * + * @return the total number of bytes + */ + static uint64_t getTotalBytesInFiles( + std::vector& fileList, + bool removeZeroLengthFilenames); + + /** + * Check if file is readable + * @param filepath + * @return True if the file is readable; false otherwise + */ + static bool isFileReadable(const std::string& filepath); + + /** + * Check if the list of files is readable + * + * @param fileList the list of files + * + * @return TRUE if all files are readable, FALSE otherwise + */ + static bool areFilesReadable(std::vector& fileList); + + /** + * Read the specified bytes from the specified file + * + * @param filename the file to read from + * @param startOffset the starting offset to read from + * @param lengthInBytes the number of bytes to read + * @param actualBytesRead the number of bytes actually read + * + * @return an allocated buffer holding the bytes, or NULL on ERROR. this buffer + * must be deallocated by the caller + */ + static char* readBytesFromFileOffset( + std::string filename, + uint64_t startOffset, + uint64_t lengthInBytes, + uint64_t& actualBytesRead); + + /** + * Read the specified bytes from the specified file + * + * @param filename the file to read from + * @param startOffset the starting offset to read from + * @param lengthInBytes the number of bytes to read + * @param destinationBuffer the buffer to copy the data into + * + * @return the number of bytes read + */ + static uint64_t readBytesFromFileOffset( + std::string filename, + uint64_t startOffset, + uint64_t lengthInBytes, + void* destinationBuffer); + + /** + * Read the specified bytes from the specified file + * + * @param inputFile the file to read from + * @param startOffset the starting offset to read from + * @param lengthInBytes the number of bytes to read + * @param destinationBuffer the buffer to copy the data into + * + * @return the number of bytes read + */ + static uint64_t readBytesFromFileOffset( + std::ifstream& inputFile, + uint64_t startOffset, + uint64_t lengthInBytes, + void* destinationBuffer); + + /** + * Read the specified input file and turn it into a list of filenames + * + * @param filename the file to read + * @param fileList the generated list of files + */ + static void readFileListFromFile( + const std::string& filename, + std::vector* fileList); + + /** + * Find the specified delimiters in the passed buffer + * + * @param buffer the buffer to search + * @param bufferSize the size of the buffer + * @param tokens the list of delimiters + */ + static inline char* findDelimiter( + char* buffer, + uint32_t bufferSize, + const std::string& tokens) + { + if ((bufferSize == 0) || (buffer == nullptr)) + { + return nullptr; + } + + static const int CMPESTRI_FLAGS = (_SIDD_UBYTE_OPS | + _SIDD_CMP_EQUAL_ANY | + _SIDD_LEAST_SIGNIFICANT); + + const __m128i delimiterCharacters = + _mm_lddqu_si128(reinterpret_cast(tokens.c_str())); + + // Calculate the number of iterations to call _mm_cmpestri. + uint32_t numIterations = bufferSize / 16; + uint32_t offset = 0; + + for (uint32_t iterIdx = 0; iterIdx < numIterations; ++iterIdx) + { + const __m128i data = _mm_lddqu_si128(reinterpret_cast(buffer + offset)); + const int index = _mm_cmpestri(delimiterCharacters, tokens.size(), + data, 16, CMPESTRI_FLAGS); + if (index < 16) + { + // Found a delimiter + return buffer + offset + index; + } + + offset += 16; + } + + // If the input isn't a multiple of 16 bytes, we can't use the _mm + // instructions. Instead resort to the basic search for the last 16 + // or fewer characters + while (offset < bufferSize) + { + for (uint32_t tokenIndex = 0; tokenIndex < tokens.length(); tokenIndex++) + { + if (buffer[offset] == tokens[tokenIndex]) + { + return buffer + offset; + } + } + + ++offset; + } + + return nullptr; + } + + /** + * Divide the passed buffer based on the delimiters + * + * @param buffer the buffer to search + * @param bufferSize the size of the buffer + * @param tokens the list of delimiters + * @param dividedComponents the resulting divided buffer components + */ + static inline void divideBufferByDelimiter( + char* buffer, + uint64_t bufferSize, + const std::string& tokens, + std::vector>& dividedComponents) + { + char* currentChar = buffer; + char* lastAvailableChar = buffer + bufferSize - 1; + + while (currentChar <= lastAvailableChar) + { + char* delimiter = findDelimiter(currentChar, + lastAvailableChar - currentChar + 1, + tokens); + if (delimiter == nullptr) + { + break; + } + + dividedComponents.push_back(std::make_pair(currentChar, + delimiter - currentChar)); + currentChar = delimiter + 1; + } + + if (currentChar <= lastAvailableChar) + { + dividedComponents.push_back(std::make_pair(currentChar, + lastAvailableChar - currentChar + 1)); + } + } + + /** + * Given a limited amount of information use a simple heuristic to estimate + * the number of chunks that will be created. This function is used to feed + * the progress reporting feature which needs an estimate of the number of + * chunks that will be created up front in order to give a reasonable + * estimate of percentage complete. + * + * @param fileList the prepared list of input files + * @param chunkSize size of chunks to be created in bytes + * @param recordBased is the search record-based? + * @return the estimated number of chunks that will be created + */ + static uint32_t estimateNumberChunks( + const std::vector& fileList, + uint32_t chunkSize, + bool recordBased=false); + + /** + * Read entries from a dictionary + * + * @param filename the file to read from + * @param dictionaryEntries the entries read from the file + */ + static int readDictionary( + const std::string& filename, + std::vector& dictionaryEntries); + + /** + * Read files recursively from the path provided. + * + * NOTE: If the filename portion of the path contains wildcards this + * function will not expand them as that is the job of + * prepareInputFiles. + * + * @param filepath input file path from which to read recursively + * @param fileList output vector in which to store recursive paths + */ + static void readFilesRecursively( + const std::string& filepathStr, + std::vector& fileList); + + /** + * Find start of UTF-8 code point in a buffer. Note this finds the start of + * 2, 3, or 4 byte code points as all 1 byte code points are backwards + * compatible with ASCII. + * + * @param buffer the buffer to search + * @param bufferSize the size of the buffer + * @return pointer to start of code point (i.e. MSByte) or NULL if none are + * found + */ + static char* findUTF8Codepoint( + char* buffer, + uint64_t bufferSize); + + /** + * Get file descriptor from a C++ stream. Note: this is a hack that works + * on Linux. + * + * All credit goes to: https://www.ginac.de/~kreckel/fileno/ + * + * @param stream + * @return file descriptor + */ + template + static inline int getFdHack(const std::basic_ios& stream) + { + typedef std::basic_filebuf filebuf_t; + filebuf_t* bbuf = dynamic_cast(stream.rdbuf()); + if (bbuf != nullptr) { + // This subclass is only there for accessing the FILE*. Ouuwww, sucks! + struct my_filebuf : public std::basic_filebuf { + int fd() { return this->_M_file.fd(); } + }; + return static_cast(bbuf)->fd(); + } + return -1; + } + + /** + * Helper function to return the first non-whitespace character in the file. + * @param inputFile the file to search + * @return the first non-whitespace character in the file + */ + static unsigned char getFirstCharacterInInputFile( + const std::string& inputFile); + + /** + * Helper function to return the last non-whitespace character in the file. + * @param inputFile the file to search + * @return the last non-whitespace character in the file + */ + static unsigned char getLastCharacterInInputFile( + const std::string& inputFile); + +private: + +}; + +} // namespace SearchLynx +} // namespace BlackLynx + +#endif /* __BLACKLYNX_FILEUTILS_H__ */ diff --git a/software/read_test/src/Utilities/NumberUtils.h b/software/read_test/src/Utilities/NumberUtils.h new file mode 100644 index 0000000..2afcfa2 --- /dev/null +++ b/software/read_test/src/Utilities/NumberUtils.h @@ -0,0 +1,69 @@ +/******************************************************************** +* This software is "commercial computer software" as defined in the * +* Federal Acquisition Regulations and is subject to BlackLynx * +* Inc's standard End User License Agreement. * +* * +* CONFIDENTIAL - All source code is the propriety and confidential * +* information of BlackLynx Inc. * +* * +* Copyright 2014-2020 BlackLynx Inc. * +* Unpublished -- all rights reserved under the copyright laws * +* of the United States. * +********************************************************************/ + +#ifndef __BLACKLYNX_NUMBERUTILS_H__ +#define __BLACKLYNX_NUMBERUTILS_H__ + +#include + +namespace BlackLynx +{ +namespace SearchLynx +{ + +class NumberUtils +{ +public: + + static inline uint64_t simplePair(uint32_t input1, uint32_t input2) + { + return (static_cast(input2) << 32) | input1; + } + + static inline void simpleDepair( + uint64_t input, + uint32_t& output1, + uint32_t& output2) + { + output1 = static_cast(input); + output2 = static_cast(input >> 32); + } + + /** + * Round the input number up to the nearest 1 megabyte (2^20) + * @param input + * @return the input number rounded up to the nearest 1 MB + */ + static inline uint32_t roundUp1M(uint32_t input) + { + return (input % (1 << 20) ? input + (1 << 20) - (input % (1 << 20)) : input); + } + + /** + * Check if a given number if a power of 2 + * @param num the positive integer to check + * @return true if the number if a power of 2; false otherwise + */ + static inline bool isPowerOfTwo(uint64_t num) + { + // First num in the below expression is for the case when num is 0 + return num && (! (num & (num - 1))); + } + +}; + + +} +} + +#endif diff --git a/software/read_test/src/Utilities/ProtectedAndSynchronizedQueue.h b/software/read_test/src/Utilities/ProtectedAndSynchronizedQueue.h new file mode 100644 index 0000000..ceb6a37 --- /dev/null +++ b/software/read_test/src/Utilities/ProtectedAndSynchronizedQueue.h @@ -0,0 +1,159 @@ +/******************************************************************** +* This software is "commercial computer software" as defined in the * +* Federal Acquisition Regulations and is subject to BlackLynx * +* Inc's standard End User License Agreement. * +* * +* CONFIDENTIAL - All source code is the propriety and confidential * +* information of BlackLynx Inc. * +* * +* Copyright 2014-2020 BlackLynx Inc. * +* Unpublished -- all rights reserved under the copyright laws * +* of the United States. * +********************************************************************/ + +#ifndef __BLACKLYNX_PROTECTEDANDSYNCHRONIZEDQUEUE_H__ +#define __BLACKLYNX_PROTECTEDANDSYNCHRONIZEDQUEUE_H__ + +#include +#include +#include +#include + + +namespace BlackLynx +{ +namespace SearchLynx +{ + +template +class ProtectedAndSynchronizedQueue +{ +public: + ProtectedAndSynchronizedQueue() + { + m_terminateFlag = false; + } + + virtual ~ProtectedAndSynchronizedQueue() {}; + + /* + * Add a new element to the back of the queue + */ + inline void push_back(T newElement) + { + if (m_terminateFlag) + { + return; + } + + { + std::lock_guard lock(m_mutex); + m_queue.push(newElement); + } + m_CV.notify_one(); + } + + /* + * Retrieve the first element of the queue and remove it from the queue + */ + inline T pop_front() + { + // Wait on queue blocking + std::unique_lock muxLock(m_mutex); + while (m_queue.empty()) + { + m_CV.wait_for(muxLock, std::chrono::milliseconds(100)); + + // If we've been asked to terminate, break out and return + // NOTE: the return value is garbage if terminate has been + // called prior this call returning + if (m_terminateFlag) + { + #pragma GCC diagnostic ignored "-Wmaybe-uninitialized" + T elementToReturn; + return elementToReturn; + } + } + + T elementToReturn = m_queue.front(); + m_queue.pop(); + muxLock.unlock(); + + return elementToReturn; + } + + /* + * Retrieve the first element of the queue and remove it from the queue + */ + inline T pop_front(bool waitOnEmptyQueue, bool& elementReturned) + { + // Wait on queue blocking + std::unique_lock muxLock(m_mutex); + if (waitOnEmptyQueue) + { + while (m_queue.empty()) + { + m_CV.wait_for(muxLock, std::chrono::milliseconds(100)); + + // If we've been asked to terminate, break out and return + // NOTE: the return value is garbage if terminate has been + // called prior this call returning + if (m_terminateFlag) + { + break; + } + } + } + + #pragma GCC diagnostic ignored "-Wmaybe-uninitialized" + T elementToReturn; + if (m_queue.size() > 0) + { + elementReturned = true; + elementToReturn = m_queue.front(); + m_queue.pop(); + } + else + { + elementReturned = false; + } + muxLock.unlock(); + + return elementToReturn; + } + + /* + * Terminate any waiting pop_front() requests + */ + void terminate() + { + m_terminateFlag = true; + m_CV.notify_all(); + } + + inline bool empty() + { + std::lock_guard lock(m_mutex); + return m_queue.empty(); + } + + inline uint32_t size() + { + std::lock_guard lock(m_mutex); + return m_queue.size(); + } + +private: + bool m_terminateFlag; + + std::queue m_queue; + std::mutex m_mutex; + std::condition_variable m_CV; + +}; + +} +} + + +#endif diff --git a/software/read_test/src/Utilities/SearchLynxConfiguration.cc b/software/read_test/src/Utilities/SearchLynxConfiguration.cc new file mode 100644 index 0000000..9d6ec09 --- /dev/null +++ b/software/read_test/src/Utilities/SearchLynxConfiguration.cc @@ -0,0 +1,399 @@ +/******************************************************************** +* This software is "commercial computer software" as defined in the * +* Federal Acquisition Regulations and is subject to BlackLynx * +* Inc's standard End User License Agreement. * +* * +* CONFIDENTIAL - All source code is the propriety and confidential * +* information of BlackLynx Inc. * +* * +* Copyright 2014-2020 BlackLynx Inc. * +* Unpublished -- all rights reserved under the copyright laws * +* of the United States. * +********************************************************************/ + +#include +#include +#include + +#include + +#include "Logger/SearchLynxLogger.h" +#include "Utilities/SearchLynxConfiguration.h" +#include "Utilities/SystemArchInfo.h" + + +// NOTE: This *must* remain a #define because its value may be needed very +// early on before any global or static variables have been initialized. +#define CONFIGURATION_FILE "/etc/blacklynx/searchlynx.conf" + +BlackLynx::SearchLynx::SearchLynxConfiguration* BlackLynx::SearchLynx::SearchLynxConfiguration::getInstance() +{ + static BlackLynx::SearchLynx::SearchLynxConfiguration theInstance; + theInstance.load(); + return &theInstance; +} + +BlackLynx::SearchLynx::SearchLynxConfiguration::SearchLynxConfiguration() + : m_configLoaded(false), + m_defaultLogLevel(info), + m_numberOfReaderThreads(22), + m_numberOfReaderThreadsOverridenByUser(false), + m_numberOfProcessorThreads(40), + m_chunkSizeInKb(65536), + m_chunkSizeOverridenByUser(false), + m_numberOfFieldFinderThreads(40), + m_maxGeneratedMatches(0), + m_licenseFilePath(""), + m_forceSoftwareExecution(false), + m_enableMatchQualifiers(false), + m_inputIsFilelist(false), + m_findInputRecursively(false), + m_followConversationEnabled(false), + m_followConversationValue(""), + m_followPayloadEnabled(false), + m_followPayloadValue(""), + m_forceSingleSocket(false), + m_isStreamingModeEnabled(false), + m_streamingInputTerminateFileIntervalInMsec(5000), + m_streamingInputProcessingIntervalInMsec(1000), + m_disableIndexResultBuffering(false), + m_disableResultBuffering(false), + m_udpTimeoutInMsec(1000), + m_tcpTimeoutInMsec(1000), + m_noSetAffinity(false) +{ +} + +BlackLynx::SearchLynx::SearchLynxConfiguration::~SearchLynxConfiguration() +{ +} + +void BlackLynx::SearchLynx::SearchLynxConfiguration::load() +{ + if (m_configLoaded) + { + return; + } + + try + { + configuration.readFile(CONFIGURATION_FILE); + } + catch (const libconfig::FileIOException &fioex) + { +// std::cerr << "Error : " << fioex.getError(); + throw fioex; + return; + } + catch(const libconfig::ParseException &pex) + { + BlnxLog(error) << "Parse error at " << pex.getFile() << ":" + << pex.getLine() << " - " << pex.getError(); + return; + } + + const libconfig::Setting& root = configuration.getRoot(); + + int intValue; + root.lookupValue("default-log-level", intValue); + m_defaultLogLevel = static_cast(intValue); + + if ((root.lookupValue("max-reader-threads", intValue) == 1) || + (root.lookupValue("max-readers", intValue) == 1)) + { + m_numberOfReaderThreadsOverridenByUser = true; + m_numberOfReaderThreads = intValue; + } + if ((root.lookupValue("max-proc-threads", intValue) == 1) || + (root.lookupValue("max-spawns", intValue) == 1)) + { + m_numberOfProcessorThreads = intValue; + } + if ((root.lookupValue("chunk-size", intValue) == 1) || + (root.lookupValue("shard-size", intValue) == 1)) + { + m_chunkSizeOverridenByUser = true; + m_chunkSizeInKb = intValue; + } + if (root.lookupValue("max-field-finders", intValue) == 1) + { + m_numberOfFieldFinderThreads = intValue; + } + if ((root.lookupValue("force-software-execution", intValue) == 1) || + (root.lookupValue("force-sw-only", intValue) == 1)) + { + m_forceSoftwareExecution = (intValue == 1); + } + if (root.lookupValue("enable-match-qualifiers", intValue) == 1) + { + m_enableMatchQualifiers = (intValue == 1); + } + if (root.lookupValue("force-single-socket", intValue) == 1) + { + m_forceSingleSocket = (intValue == 1); + } + + std::string strValue; + if (root.lookupValue("license-file-path", strValue)) + { + m_licenseFilePath = strValue; + } + + if (root.lookupValue("follow-conversation", strValue)) + { + m_followConversationValue = strValue; + m_followConversationEnabled = true; + } + + if (root.lookupValue("follow-payload", strValue)) + { + m_followPayloadValue = strValue; + m_followPayloadEnabled = true; + } + + if (root.lookupValue("streaming-input-terminate-file-interval-msec", intValue) == 1) + { + m_streamingInputTerminateFileIntervalInMsec = intValue; + } + + if (root.lookupValue("streaming-input-processing-interval-msec", intValue) == 1) + { + m_streamingInputProcessingIntervalInMsec = intValue; + } + + if (root.lookupValue("disable-index-result-buffering", intValue) == 1) + { + m_disableIndexResultBuffering = (intValue == 1); + } + + if (root.lookupValue("disable-result-buffering", intValue) == 1) + { + m_disableResultBuffering = (intValue == 1); + } + + if (root.lookupValue("udp-timeout-msec", intValue) == 1) + { + m_udpTimeoutInMsec = intValue; + } + + if (root.lookupValue("tcp-timeout-msec", intValue) == 1) + { + m_tcpTimeoutInMsec = intValue; + } + + if (root.lookupValue("no-setaffinity", intValue) == 1) + { + m_noSetAffinity = (intValue == 1); + } + + // This option must stay the last one + if (root.lookupValue("enable-streaming-input", intValue) == 1) + { + m_isStreamingModeEnabled = (intValue == 1); + m_chunkSizeInKb = 65536; + m_chunkSizeOverridenByUser = false; + } + + // Mark that the configuration has been loaded + m_configLoaded = true; +} + +void BlackLynx::SearchLynx::SearchLynxConfiguration::print() +{ + BlnxLog(debug) << "SearchLynx current configuration:"; + BlnxLog(info) << " default-log-level = " << m_defaultLogLevel; + BlnxLog(info) << " license-file-path = " << m_licenseFilePath; + BlnxLog(info) << " force-software-execution = " << m_forceSoftwareExecution; + BlnxLog(info) << " max-reader-threads = " << m_numberOfReaderThreads; + BlnxLog(info) << " max-proc-threads = " << m_numberOfProcessorThreads; + BlnxLog(info) << " chunk-size (KB) = " << m_chunkSizeInKb; + BlnxLog(info) << " max-field-finders = " << m_numberOfFieldFinderThreads; + BlnxLog(info) << " max-generated-matches = " << m_maxGeneratedMatches; + BlnxLog(info) << " enable-match-qualifiers = " << m_enableMatchQualifiers; + BlnxLog(info) << " input-is-filelist = " << m_inputIsFilelist; + BlnxLog(info) << " find-input-recursively = " << m_findInputRecursively; + BlnxLog(info) << " follow-conversation = " << m_followConversationValue; + BlnxLog(info) << " follow-payload = " << m_followPayloadValue; + BlnxLog(info) << " force-single-socket = " << m_forceSingleSocket; + BlnxLog(info) << " no-setaffinity = " << m_noSetAffinity; + + // DBS - 8/24/2020 + // NOTE: until the status of the streaming feature is determined these + // streaming related options will be hidden from the user. +// BlnxLog(info) << " enable-streaming-input = " << m_isStreamingModeEnabled; +// BlnxLog(info) << " streaming-input-terminate-file-interval-msec = " << m_streamingInputTerminateFileIntervalInMsec; +// BlnxLog(info) << " streaming-input-processing-interval-msec = " << m_streamingInputProcessingIntervalInMsec; +// BlnxLog(info) << " disable-index-result-buffering = " << m_disableIndexResultBuffering; +// BlnxLog(info) << " disable-result-buffering = " << m_disableResultBuffering; +// BlnxLog(info) << " tcp-timeout-msec = " << m_tcpTimeoutInMsec; +// BlnxLog(info) << " udp-timeout-msec = " << m_udpTimeoutInMsec; + +} + +int BlackLynx::SearchLynx::SearchLynxConfiguration::setOptions( + std::string& optionString) +{ + if (! m_configLoaded) + { + load(); + m_configLoaded = true; + } + + // Save the global options + m_globalOptionString.append(optionString); + m_globalOptionString.append(" "); + + BlnxLog(info) << "Parsing global option string: " << optionString; + if (optionString.empty()) + { + return 0; + } + + std::vector optionVector; + boost::split(optionVector, optionString, boost::is_any_of(",")); + + for (auto pairIterator : optionVector) + { + std::vector optionPairList; + boost::split(optionPairList, pairIterator, boost::is_any_of("=")); + + // Strip any spaces from the front or rear of the string + for (uint32_t idx = 0; idx < optionPairList[0].length(); ++idx) + { + if (optionPairList[0][0] == ' ') + { + optionPairList[0].erase(0, 1); + } + else + { + break; + } + } + for (int32_t idx = optionPairList[0].length() - 1; idx >= 0; ++idx) + { + if (optionPairList[0][optionPairList[0].length() - 1] == ' ') + { + optionPairList[0].erase(optionPairList[0].length() - 1, 1); + } + else + { + break; + } + } + + if ((optionPairList[0] == "max-reader-threads") || + (optionPairList[0] == "max-readers")) + { + m_numberOfReaderThreadsOverridenByUser = true; + m_numberOfReaderThreads = std::atoi(optionPairList[1].c_str()); + } + else if (optionPairList[0] == "enable-streaming-input") + { + int intValue = std::atoi(optionPairList[1].c_str()); + m_isStreamingModeEnabled = (intValue == 1); + } + else if (optionPairList[0] == "tcp-timeout-msec") + { + int intValue = std::atoi(optionPairList[1].c_str()); + m_tcpTimeoutInMsec = intValue; + } + else if (optionPairList[0] == "udp-timeout-msec") + { + int intValue = std::atoi(optionPairList[1].c_str()); + m_udpTimeoutInMsec = intValue; + } + else if (optionPairList[0] == "disable-index-result-buffering") + { + int intValue = std::atoi(optionPairList[1].c_str()); + m_disableIndexResultBuffering = (intValue == 1); + } + else if (optionPairList[0] == "disable-result-buffering") + { + int intValue = std::atoi(optionPairList[1].c_str()); + m_disableResultBuffering = (intValue == 1); + } + else if (optionPairList[0] == "streaming-input-terminate-file-interval-msec") + { + int intValue = std::atoi(optionPairList[1].c_str()); + m_streamingInputTerminateFileIntervalInMsec = intValue; + } + else if (optionPairList[0] == "streaming-input-processing-interval-msec") + { + int intValue = std::atoi(optionPairList[1].c_str()); + m_streamingInputProcessingIntervalInMsec = intValue; + } + else if (optionPairList[0] == "force-single-socket") + { + int intValue = std::atoi(optionPairList[1].c_str()); + m_forceSingleSocket = (intValue == 1); + } + else if ((optionPairList[0] == "force-software-execution") || + (optionPairList[0] == "force-sw-only")) + { + int intValue = std::atoi(optionPairList[1].c_str()); + m_forceSoftwareExecution = (intValue == 1); + } + else if (optionPairList[0] == "enable-match-qualifiers") + { + int intValue = std::atoi(optionPairList[1].c_str()); + m_enableMatchQualifiers = (intValue == 1); + } + else if ((optionPairList[0] == "max-proc-threads") || + (optionPairList[0] == "max-spawns")) + { + m_numberOfProcessorThreads = std::atoi(optionPairList[1].c_str()); + } + else if ((optionPairList[0] == "chunk-size") || + (optionPairList[0] == "shard-size")) + { + m_chunkSizeInKb = std::atoi(optionPairList[1].c_str()); + m_chunkSizeOverridenByUser = true; + } + else if (optionPairList[0] == "max-field-finders") + { + m_numberOfFieldFinderThreads = std::atoi(optionPairList[1].c_str()); + } + else if (optionPairList[0] == "max-generated-matches") + { + m_maxGeneratedMatches = std::atoi(optionPairList[1].c_str()); + } + else if (optionPairList[0] == "input-is-filelist") + { + int intValue = std::atoi(optionPairList[1].c_str()); + m_inputIsFilelist = (intValue == 1); + } + else if (optionPairList[0] == "find-input-recursively") + { + int intValue = std::atoi(optionPairList[1].c_str()); + m_findInputRecursively = (intValue == 1); + } + else if (optionPairList[0] == "follow-conversation") + { + m_followConversationValue = optionPairList[1]; + m_followConversationEnabled = true; + } + else if (optionPairList[0] == "follow-payload") + { + m_followPayloadValue = optionPairList[1]; + m_followPayloadEnabled = true; + } + else if (optionPairList[0] == "no-setaffinity") + { + int intValue = std::atoi(optionPairList[1].c_str()); + m_noSetAffinity = (intValue == 1); + } + else + { + BlnxLog(error) << "Unknown option: " << optionPairList[0] << "\n"; + } + } + + // When in streaming mode we ignore any chunk size + if (m_isStreamingModeEnabled) + { + m_chunkSizeInKb = 65536; + m_chunkSizeOverridenByUser = false; + } + + return 0; +} diff --git a/software/read_test/src/Utilities/SearchLynxConfiguration.h b/software/read_test/src/Utilities/SearchLynxConfiguration.h new file mode 100644 index 0000000..93687fb --- /dev/null +++ b/software/read_test/src/Utilities/SearchLynxConfiguration.h @@ -0,0 +1,342 @@ +/******************************************************************** +* This software is "commercial computer software" as defined in the * +* Federal Acquisition Regulations and is subject to BlackLynx * +* Inc's standard End User License Agreement. * +* * +* CONFIDENTIAL - All source code is the propriety and confidential * +* information of BlackLynx Inc. * +* * +* Copyright 2014-2020 BlackLynx Inc. * +* Unpublished -- all rights reserved under the copyright laws * +* of the United States. * +********************************************************************/ + +#ifndef __BLACKLYNX_SEARCHLYNXCONFIGURATION_H__ +#define __BLACKLYNX_SEARCHLYNXCONFIGURATION_H__ + +#include +#include + +#include + +#include "Logger/SearchLynxLogger.h" + +namespace BlackLynx +{ +namespace SearchLynx +{ + +class SearchLynxConfiguration +{ +public: + + /** + * Get the singleton instance + * + * @return the singleton instance + */ + static SearchLynxConfiguration* getInstance(); + + /* + * This method will set the global options in option_string + * + * @param option_string the option string + * the format of this parameter is a comma separated list + * of NAME=VALUE pairs. for example: + * + * shard-size=32, max-spawns=40 + * + * @return 0 on success, -1 otherwise + */ + int setOptions(std::string& optionString); + + /** + * Get the default log level + * + * @return the default log level + */ + inline enum SEARCHLYNX_LOG_LEVEL getLogLevel() + { + return m_defaultLogLevel; + } + + /** + * Get the number of reader threads to use + * + * @return the number of reader threads + */ + inline uint32_t getNumberOfReaderThreads() + { + return m_numberOfReaderThreads; + } + + void setNumberOfReaderThreads(uint32_t numberOfThreads) + { + m_numberOfReaderThreads = numberOfThreads; + } + + bool getIsNumberOfReaderThreadsOverriddenByUser() + { + return m_numberOfReaderThreadsOverridenByUser; + } + + /** + * Get the number of processor threads to use + * + * @return the number of processor threads + */ + inline uint32_t getNumberOfProcessorThreads() + { + return m_numberOfProcessorThreads; + } + + void setNumberOfProcessorThreads(uint32_t numberOfThreads) + { + m_numberOfProcessorThreads = numberOfThreads; + } + + /** + * Get the license file path + * + * @return the license file path + */ + inline const std::string& getLicenseFilePath() + { + return m_licenseFilePath; + } + + /** + * Get the chunk size to use + * + * @return the chunk size + */ + inline uint32_t getChunkSizeInKb() + { + return m_chunkSizeInKb; + } + + void setChunkSizeInKb(uint32_t chunkSize) + { + m_chunkSizeInKb = chunkSize; + } + + bool isChunkSizeOverridenByUser() + { + return m_chunkSizeOverridenByUser; + } + + /** + * Get the number of field finder threads to use + * + * @return the number of threads + */ + inline uint32_t getNumberOfFieldFinderThreads() + { + return m_numberOfFieldFinderThreads; + } + + /** + * Get the max number of generated matches + * + * @return the number of threads + */ + inline uint32_t getMaxNumberOfGeneratedMatches() + { + return m_maxGeneratedMatches; + } + + /** + * Get the software execution state + * + * @return the software execution state + */ + inline bool getForceSoftwareExecution() + { + return m_forceSoftwareExecution; + } + + /** + * Get the match qualifier state + * + * @return the match qualifier state + */ + inline bool getEnableMatchQualifiers() + { + return m_enableMatchQualifiers; + } + + /** + * + * @return input is filelist flag state + */ + inline bool getInputIsFileList() + { + return m_inputIsFilelist; + } + + /** + * + * @return find input recursively flag state + */ + inline bool getFindInputRecursively() + { + return m_findInputRecursively; + } + +#ifdef CYBERLYNX + inline bool getIsFollowConversationEnabled() + { + return m_followConversationEnabled; + } + + inline const std::string& getFollowConversationValue() + { + return m_followConversationValue; + } + + inline bool getIsFollowPayloadEnabled() + { + return m_followPayloadEnabled; + } + + inline const std::string& getFollowPayloadValue() + { + return m_followPayloadValue; + } +#endif + + /* + * @return is streaming mode enabled ? + */ + inline bool getIsStreamingModeEnabled() + { + return m_isStreamingModeEnabled; + } + + /* + * @return force using single socket configuration + */ + inline bool getForceSingleSocket() + { + return m_forceSingleSocket; + } + + /* + * @return the input timeouts + */ + inline uint64_t getStreamingInputTerminateFileIntervalInMsec() + { + return m_streamingInputTerminateFileIntervalInMsec; + } + + inline uint64_t getStreamingInputProcessingIntervalInMsec() + { + return m_streamingInputProcessingIntervalInMsec; + } + + /* + * @return the index/result buffering disable + */ + inline bool getDisableIndexResultBuffering() + { + return m_disableIndexResultBuffering; + } + + inline bool getDisableResultBuffering() + { + return m_disableResultBuffering; + } + + /* + * @return the tcp/udp timeouts + */ + inline uint64_t getUdpTimeoutInMsec() + { + return m_udpTimeoutInMsec; + } + + inline uint64_t getTcpTimeoutInMsec() + { + return m_tcpTimeoutInMsec; + } + + inline bool getNoSetAffinity() + { + return m_noSetAffinity; + } + + /** + * Get the global option string in use + * + * @return the global option string + */ + inline const std::string& getGlobalOptionString() + { + return m_globalOptionString; + } + + /** + * Print configuration to the log. + */ + void print(); + + /** + * Load the configuration file data. + */ + void load(); + +private: + + SearchLynxConfiguration(); + ~SearchLynxConfiguration(); + + bool m_configLoaded; + + libconfig::Config configuration; + enum SEARCHLYNX_LOG_LEVEL m_defaultLogLevel; + + uint32_t m_numberOfReaderThreads; + bool m_numberOfReaderThreadsOverridenByUser; + uint32_t m_numberOfProcessorThreads; + + uint32_t m_chunkSizeInKb; + bool m_chunkSizeOverridenByUser; + + uint32_t m_numberOfFieldFinderThreads; + uint32_t m_maxGeneratedMatches; + + std::string m_licenseFilePath; + + bool m_forceSoftwareExecution; + bool m_enableMatchQualifiers; + + bool m_inputIsFilelist; + bool m_findInputRecursively; + + std::string m_globalOptionString; + + bool m_followConversationEnabled; + std::string m_followConversationValue; + + bool m_followPayloadEnabled; + std::string m_followPayloadValue; + + bool m_forceSingleSocket; + + bool m_isStreamingModeEnabled; + uint64_t m_streamingInputTerminateFileIntervalInMsec; + uint64_t m_streamingInputProcessingIntervalInMsec; + + bool m_disableIndexResultBuffering; + bool m_disableResultBuffering; + + uint64_t m_udpTimeoutInMsec; + uint64_t m_tcpTimeoutInMsec; + + bool m_noSetAffinity; +}; + +} // namespace SearchLynx +} // namespace BlackLynx + +#endif diff --git a/software/read_test/src/Utilities/SynchronizedValue.h b/software/read_test/src/Utilities/SynchronizedValue.h new file mode 100644 index 0000000..c4143f0 --- /dev/null +++ b/software/read_test/src/Utilities/SynchronizedValue.h @@ -0,0 +1,137 @@ +/******************************************************************** +* This software is "commercial computer software" as defined in the * +* Federal Acquisition Regulations and is subject to BlackLynx * +* Inc's standard End User License Agreement. * +* * +* CONFIDENTIAL - All source code is the propriety and confidential * +* information of BlackLynx Inc. * +* * +* Copyright 2018-2020 BlackLynx Inc. * +* Unpublished -- all rights reserved under the copyright laws * +* of the United States. * +********************************************************************/ +#ifndef __BLNX_SYNCHRONIZEDVALUE_H__ +#define __BLNX_SYNCHRONIZEDVALUE_H__ + +#include +#include +#include +#include + +namespace BlackLynx +{ +namespace SearchLynx +{ + +template +class SynchronizedValue +{ +public: + SynchronizedValue() + : m_terminateFlag(false), + m_flag(false) + { + } + + SynchronizedValue(T initValue) + : m_terminateFlag(false), + m_flag(false), + m_value(initValue) + { + } + + virtual ~SynchronizedValue() {} + + /** + * Terminate any pending getValue() calls. + */ + void terminate() + { + m_terminateFlag = true; + m_CV.notify_all(); + } + + /** + * Set the value and notify any blocked getValue() calls. + */ + void setValue(T value) + { + m_value = value; + { + std::lock_guard lock(m_mutex); + m_flag = true; + } + m_CV.notify_all(); + } + + /** + * Get the value, potentially blocking until it becomes available. Note + * that the "terminated" parameter should always be checked before using + * the returned value as a call to terminate() might cause this function + * to return a default initialized value. + * + * @param terminated - indicates if a blocking wait was terminated + * @param block - (default: true) block while waiting for value to be set + * + * @return the stored value + */ + T getValue(bool& terminated, bool block=true) + { + terminated = false; + + // If not blocking; return the current value without waiting if it + // hasn't yet been set + if (! block) + { + std::lock_guard lock(m_mutex); + return m_value; + } + + std::unique_lock muxLock(m_mutex); + while (m_flag == false) + { + m_CV.wait_for(muxLock, std::chrono::milliseconds(1)); + + // If we've been asked to terminate, break out of the loop and abort + if (m_terminateFlag) + { + terminated = true; + break; + } + } + muxLock.unlock(); + + return m_value; + } + + /** + * Reset the internal flag to allow the value to be set again. Note this + * function must be used with care! + */ + void reset() + { + std::lock_guard lock(m_mutex); + m_flag = false; + } + + void reset(T newVal) + { + std::lock_guard lock(m_mutex); + m_flag = false; + m_value = newVal; + } + +private: + + bool m_terminateFlag; + bool m_flag; + T m_value; + std::mutex m_mutex; + std::condition_variable m_CV; + +}; + +} +} + +#endif diff --git a/software/read_test/src/Utilities/SystemArchInfo.cc b/software/read_test/src/Utilities/SystemArchInfo.cc new file mode 100644 index 0000000..3e48c2d --- /dev/null +++ b/software/read_test/src/Utilities/SystemArchInfo.cc @@ -0,0 +1,168 @@ +/******************************************************************** +* This software is "commercial computer software" as defined in the * +* Federal Acquisition Regulations and is subject to BlackLynx * +* Inc's standard End User License Agreement. * +* * +* CONFIDENTIAL - All source code is the propriety and confidential * +* information of BlackLynx Inc. * +* * +* Copyright 2014-2020 BlackLynx Inc. * +* Unpublished -- all rights reserved under the copyright laws * +* of the United States. * +********************************************************************/ + +#include +#include +#include +#include + +#include +#include +#include + +#include +#include + +#include "Logger/SearchLynxLogger.h" +#include "SearchLynxConfiguration.h" +#include "Utilities/SystemArchInfo.h" + + +BlackLynx::SearchLynx::SystemArchInfo* BlackLynx::SearchLynx::SystemArchInfo::getInstance() +{ + static BlackLynx::SearchLynx::SystemArchInfo theInstance; + return &theInstance; +} + +BlackLynx::SearchLynx::SystemArchInfo::SystemArchInfo() + : m_cpuSupportsSSE4_2(__builtin_cpu_supports("sse4.2")), + m_cpuSupportsAVX2(__builtin_cpu_supports("avx2")), + m_cpuSupportsAVX512F(__builtin_cpu_supports("avx512f")) + +{ + boost::filesystem::path nodeDirPath{"/sys/devices/system/node/"}; + std::vector nodeStrs; + + // Iterate over the contents of the node dir path directory looking for nodes + for (boost::filesystem::directory_iterator nodeDirIter{nodeDirPath}; + nodeDirIter != boost::filesystem::directory_iterator{}; ++nodeDirIter) + { + // Find any entries that look like "nodeX" where X is an integer + std::string filename = nodeDirIter->path().filename().string(); + if ((filename.substr(0, 4) == "node") && (filename.size() == 5)) + { + nodeStrs.push_back(filename); + } + } + + // Now iterate over the nodes looking for attached CPUs + for (auto& nodeStr : nodeStrs) + { + boost::filesystem::path nodePath(nodeDirPath); + nodePath += nodeStr; + uint32_t nodeIndex = std::stoi(nodeStr.substr(4)); + + // Iterate over the contents of the node path directory looking for nodes + for (boost::filesystem::directory_iterator nodeIter{nodePath}; + nodeIter != boost::filesystem::directory_iterator{}; ++nodeIter) + { + // Look for "cpuXX" entries + std::string entry = nodeIter->path().filename().string(); + if (entry.substr(0, 3) == "cpu") + { + const std::string& indexStr(entry.substr(3)); + if (std::all_of(indexStr.begin(), indexStr.end(), ::isdigit)) + { + m_nodeToCpuNums[nodeIndex].push_back(std::stoi(indexStr)); + } + } + } + } + + CPU_ZERO(&m_allCPUs); + + // Finally build the corresponding cpu_set_t's for later use + for (auto& nodeIter : m_nodeToCpuNums) + { + CPU_ZERO(&m_nodeToCpuSet[nodeIter.first]); + + for (auto cpuIdx : nodeIter.second) + { + CPU_SET(cpuIdx, &m_nodeToCpuSet[nodeIter.first]); + + // Also set the all CPU set + CPU_SET(cpuIdx,&m_allCPUs); + } + } +} + +BlackLynx::SearchLynx::SystemArchInfo::~SystemArchInfo() +{ +} + +bool BlackLynx::SearchLynx::SystemArchInfo::isDualSocket() +{ + return (getNumNodes() == 2); +} + +uint32_t BlackLynx::SearchLynx::SystemArchInfo::getNumNodes() +{ + return m_nodeToCpuNums.size(); +} + +uint32_t BlackLynx::SearchLynx::SystemArchInfo::getNumCores() +{ + uint32_t numCores = 0; + for (auto& iter : m_nodeToCpuNums) + { + numCores += iter.second.size(); + } + + return numCores; +} + +const std::vector& BlackLynx::SearchLynx::SystemArchInfo::getCPUsForNode( + uint32_t nodeIdx) +{ + return m_nodeToCpuNums[nodeIdx]; +} + +cpu_set_t* BlackLynx::SearchLynx::SystemArchInfo::getCPU_Set(uint32_t nodeIdx) +{ + return &m_nodeToCpuSet[nodeIdx]; +} + +int BlackLynx::SearchLynx::SystemArchInfo::pinThisThread(uint32_t nodeIdx) +{ + return BlackLynx::SearchLynx::SystemArchInfo::pinThisThread(getCPU_Set(nodeIdx)); +} + +int BlackLynx::SearchLynx::SystemArchInfo::pinToAllNodes() +{ + return BlackLynx::SearchLynx::SystemArchInfo::pinThisThread(&m_allCPUs); +} + +int BlackLynx::SearchLynx::SystemArchInfo::pinThisThread(cpu_set_t* pinToCpuSet) +{ + auto slConfig = SearchLynxConfiguration::getInstance(); + if (slConfig->getNoSetAffinity() == false) + { + return pthread_setaffinity_np(pthread_self(), sizeof(cpu_set_t), + pinToCpuSet); + } + return 0; +} + + +uint64_t BlackLynx::SearchLynx::SystemArchInfo::getRamSizeBytes() +{ + struct sysinfo info; + int rc = sysinfo(&info); + if (rc != 0) + { + BlnxLog(error) << "Unable to read system info: " << errno; + return 0; + } + + return info.totalram; +} \ No newline at end of file diff --git a/software/read_test/src/Utilities/SystemArchInfo.h b/software/read_test/src/Utilities/SystemArchInfo.h new file mode 100644 index 0000000..4ec13be --- /dev/null +++ b/software/read_test/src/Utilities/SystemArchInfo.h @@ -0,0 +1,134 @@ +/******************************************************************** +* This software is "commercial computer software" as defined in the * +* Federal Acquisition Regulations and is subject to BlackLynx * +* Inc's standard End User License Agreement. * +* * +* CONFIDENTIAL - All source code is the propriety and confidential * +* information of BlackLynx Inc. * +* * +* Copyright 2014-2020 BlackLynx Inc. * +* Unpublished -- all rights reserved under the copyright laws * +* of the United States. * +********************************************************************/ + +#ifndef __BLACKLYNX_SYSTEMARCHINFO_H__ +#define __BLACKLYNX_SYSTEMARCHINFO_H__ + +#include +#include +#include + +#include + +#include +#include + + +namespace BlackLynx +{ +namespace SearchLynx +{ + +class SystemArchInfo +{ +public: + + /** + * Get a pointer to the SystemArchInfo singleton + * @return a pointer to the SystemArchInfo singleton + */ + static SystemArchInfo* getInstance(); + + /** + * @return true if the system has 2 NUMA nodes (i.e. is dual socket) + */ + bool isDualSocket(); + + /** + * Get the number of NUMA nodes (i.e. physical sockets) in the system + * @return the number of NUMA nodes in the system + */ + uint32_t getNumNodes(); + + /** + * Get the total number of cores across all NUMA nodes in the system + * @return the total number of cores across all NUMA nodes in the system + */ + uint32_t getNumCores(); + + /** + * Get the list of CPU numbers corresponding to the passed NUMA node index. + * @param nodeIdx the NUMA node index (i.e. CPU socket) + * @return the list of CPU numbers corresponding to the passed NUMA node + * index + */ + const std::vector& getCPUsForNode(uint32_t nodeIdx); + + /** + * Get the cpu_set_t corresponding to the passed NUMA node index. + * @param nodeIdx the NUMA node index (i.e. CPU socket) + * @return the cpu_set_t corresponding to the passed NUMA node index + */ + cpu_set_t* getCPU_Set(uint32_t nodeIdx); + + /** + * Pin the current thread to the specified NUMA node + * @param nodeIdx NUMA node index (i.e. CPU socket) + * @return zero on success; non-zero for various errors + */ + int pinThisThread(uint32_t nodeIdx); + + /** + * Pin the current thread to the all NUMA nodes; NOTE: this effectively + * "un does" all previous pinning thereby allowing the thread to run + * anywhere + * @return zero on success; non-zero for various errors + */ + int pinToAllNodes(); + + /** + * Pin the current thread to the specified CPU set + * @param pinToCpuSet the CPU set to which to pin this thread + * @return zero on success; non-zero for various errors + */ + static int pinThisThread(cpu_set_t* pinToCpuSet); + + /** + * CPU feature check functions + */ + inline bool cpuSupportsSSE4_2() { return m_cpuSupportsSSE4_2; }; + inline bool cpuSupportsAVX2() { return m_cpuSupportsAVX2; }; + inline bool cpuSupportsAVX512F() { return m_cpuSupportsAVX512F; }; + + /** + * Return the total amount of RAM in the current system in bytes. + * @return + */ + uint64_t getRamSizeBytes(); + +private: + + // Constructor + SystemArchInfo(); + + // Destructor + virtual ~SystemArchInfo(); + + // NUMA node (i.e. physical socket) to CPU (i.e. core) number + std::map> m_nodeToCpuNums; + + // NUMA node (i.e. physical socket) to CPU set + std::map m_nodeToCpuSet; + + // CPU set for all CPUs + cpu_set_t m_allCPUs; + + bool m_cpuSupportsSSE4_2; //!< CPU supports SSE 4.2 (required minimum) + bool m_cpuSupportsAVX2; //!< CPU supports AVX2 + bool m_cpuSupportsAVX512F; //!< CPU supports AVX512F +}; + +} +} + +#endif /* __BLACKLYNX_SYSTEMARCHINFO_H__ */ diff --git a/software/read_test/src/main.cc b/software/read_test/src/main.cc new file mode 100644 index 0000000..a8281bc --- /dev/null +++ b/software/read_test/src/main.cc @@ -0,0 +1,78 @@ +#include +#include +#include +#include +#include + +#include "AIO_ExecDriver.h" +#include "ExecDriver.h" +#include "Utilities/BufferAllocator.h" +#include "Utilities/FileUtils.h" +#include "Utilities/SearchLynxConfiguration.h" + +std::vector UNITS{"b/sec", "KB/s", "MB/s", "GB/s", "TB/s"}; + +double printRate(double rawRate, uint32_t& unitIdx) +{ + uint32_t idx = 0; + double rate = rawRate; + while (rate > 1024) + { + rate /= 1024; + ++idx; + } + + unitIdx = idx; + return rate; +} + + +int main(int argc, char** argv) +{ + std::cout << "Read Test Utility\n" << std::endl; + + if (argc < 2) + { + std::cerr << "ERROR: please provide an input file" << std::endl; + return -1; + } + + // Start time measurement + auto start = std::chrono::system_clock::now(); + + // Prime the config object + auto slConfig = BlackLynx::SearchLynx::SearchLynxConfiguration::getInstance(); + + // Allocate buffers + auto bufAllocator = BlackLynx::SearchLynx::BufferAllocator::getInstance(); + bufAllocator->initAllocation(); + + // Read in the file + std::string filename(argv[1]); +// BlackLynx::SearchLynx::ExecDriver driver(filename, 22); + BlackLynx::SearchLynx::AIO_ExecDriver driver(filename, 3); + driver.start(); + driver.waitUntilComplete(); + + // Deallocate buffers + bufAllocator->deallocateBuffers(); + + // End time measurement + auto end = std::chrono::system_clock::now(); + auto diff = end - start; + double duration = std::chrono::duration (diff).count() ; + std::cout << "Duration: " << std::fixed << std::setprecision(3) + << duration << " ms" << std::endl; + + uint64_t fileSize = BlackLynx::SearchLynx::FileUtils::getFileSize(filename); + + double rawRate = fileSize / duration; + std::cout << "Raw rate: " << rawRate << " b/sec" << std::endl; + + uint32_t idx = 0; + double rate = printRate(rawRate, idx); + std::cout << "Rate: " << std::fixed << std::setprecision(3) << rate + << " " << UNITS[idx] << std::endl; + + return 0; +}