Project

General

Profile

« Previous | Next » 

Revision a0d1d164

Added by David Sorber about 4 years ago

Adding initial version of read_test.

View differences:

software/read_test/CMakeLists.txt
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})
software/read_test/src/AIO_ExecDriver.cc
/********************************************************************
* 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 <sys/types.h>
#include <sys/stat.h>
#include <fcntl.h>
#include <libaio.h>
#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<struct iocb> chunkIoReqs;
std::vector<struct iocb*> chunkIoReqPtrs;
std::vector<struct io_event> 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);
}
software/read_test/src/AIO_ExecDriver.h
/********************************************************************
* 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 <cstdint>
#include <mutex>
#include <string>
#include <thread>
#include <vector>
#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__
software/read_test/src/Chunk.cc
/********************************************************************
* 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);
}
software/read_test/src/Chunk.h
/********************************************************************
* 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 <cstdint>
#include <string>
#include <vector>
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__
software/read_test/src/ExecDriver.cc
/********************************************************************
* 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;
}
}
software/read_test/src/ExecDriver.h
/********************************************************************
* 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 <cstdint>
#include <string>
#include <thread>
#include <vector>
#include "Chunk.h"
#include "Utilities/ProtectedAndSynchronizedQueue.h"
namespace BlackLynx
{
namespace SearchLynx
{
class ExecDriver
{
public:
typedef ProtectedAndSynchronizedQueue<Chunk*> 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<std::thread*> 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__
software/read_test/src/Logger/SearchLynxCLogger.cc
/********************************************************************
* 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 <stdarg.h>
#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<enum BlackLynx::SearchLynx::SEARCHLYNX_LOG_LEVEL>(logLevel)) << msg;
}
software/read_test/src/Logger/SearchLynxCLogger.h
/********************************************************************
* 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
software/read_test/src/Logger/SearchLynxLogger.cc
/********************************************************************
* 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 <cstdio>
#include <chrono>
#include <thread>
#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;
}
}
software/read_test/src/Logger/SearchLynxLogger.h
/********************************************************************
* 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 <iostream>
#include <fstream>
#include <sstream>
#include <mutex>
#include <sys/time.h>
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<typename T>
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__
software/read_test/src/Utilities/BufferAllocator.cc
/********************************************************************
* 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 <algorithm>
#include <cstdint>
#include <unistd.h>
#include <sys/sysinfo.h>
#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<std::mutex> 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<std::mutex> 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(
... This diff was truncated because it exceeds the maximum size that can be displayed.

Also available in: Unified diff