Project

General

Profile

« Previous | Next » 

Revision 2c54c4c3

Added by david.sorber over 2 years ago

Work in progress.

View differences:

CMakeLists.txt
project("copytool")
cmake_policy(SET CMP0054 NEW)
################################################################################
# Check GCC version
################################################################################
......
################################################################################
# Configuration
# Build Configuration
################################################################################
# Do not let CMake add the build directory path to RPATH. This means that CL/SL
......
SET(LIBRARY_INSTALL_PATH "/usr/lib64/")
ENDIF()
################################################################################
# Target
# Target: copytool
################################################################################
ADD_EXECUTABLE(copytool ${CMAKE_CURRENT_SOURCE_DIR}/src/copytool_main.cc)
INCLUDE_DIRECTORIES(${CMAKE_CURRENT_SOURCE_DIR}/src)
SET(copytool_sources
${CMAKE_CURRENT_SOURCE_DIR}/src/CopyManager.cc
${CMAKE_CURRENT_SOURCE_DIR}/src/CopyManagerOptions.cc
${CMAKE_CURRENT_SOURCE_DIR}/src/Logger.cc
)
ADD_EXECUTABLE(copytool
${copytool_sources}
${CMAKE_CURRENT_SOURCE_DIR}/src/copytool_main.cc)
################################################################################
# Create uninstall target
src/CopyManager.cc
/********************************************************************
* CONFIDENTIAL - All source code is the propriety and confidential *
* information of David Sorber. *
* *
* Copyright 2024 David Sorber *
* Unpublished -- all rights reserved under the copyright laws *
* of the United States. *
********************************************************************/
#include <chrono>
#include <system_error>
#include <sys/stat.h>
#include "CopyManager.h"
DBS::CopyManager::CopyManager(CopyManagerOptions& options)
: r_options(options),
m_terminate(false),
p_mainThread(nullptr)
{
}
DBS::CopyManager::~CopyManager()
{
// Add processor thread terminators
for (uint32_t idx = 0; idx < r_options.getNumProcThreads(); ++idx)
{
m_entryQueue.push_back(ENTRY_TERMINATOR);
}
// Join and delete thread terminators
for (auto& thread : m_procThreads)
{
thread->join();
delete thread;
}
m_procThreads.clear();
// Cleanup the main thread
if (p_mainThread != nullptr)
{
if (p_mainThread->joinable())
{
p_mainThread->join();
}
delete p_mainThread;
p_mainThread = nullptr;
}
}
int DBS::CopyManager::run()
{
// Spawn processor threads
for (uint32_t idx = 0; idx < r_options.getNumProcThreads(); ++idx)
{
m_procThreads.push_back(
new std::thread(&CopyManager::processThreadBody, this));
}
// Spawn the main thread
p_mainThread = new std::thread(&CopyManager::mainThreadBody, this);
p_mainThread->join();
return 0;
}
void DBS::CopyManager::mainThreadBody()
{
const std::string& srcPath(r_options.getSrcPath());
for (const auto& dirEntry : sfs::recursive_directory_iterator(srcPath))
{
// Flow control
while (m_entryQueue.size() >= r_options.getFlowControlThreshold())
{
std::this_thread::sleep_for(std::chrono::milliseconds(5));
}
// std::cout << "PATH: " << dirEntry << std::endl;
m_entryQueue.push_back(dirEntry);
}
// Add processor thread terminators
for (uint32_t idx = 0; idx < r_options.getNumProcThreads(); ++idx)
{
m_entryQueue.push_back(ENTRY_TERMINATOR);
}
}
void DBS::CopyManager::processThreadBody()
{
sfs::directory_entry entry;
struct stat statBuf{};
int rc = 0;
std::error_code errorCode;
while (true)
{
// Grab entry from queue
entry = m_entryQueue.pop_front();
if (entry == ENTRY_TERMINATOR)
{
break;
}
if (entry.is_directory())
{
// Directory entry
const char* dirPath = entry.path().string().c_str();
rc = stat(dirPath, &statBuf);
if (! sfs::create_directories(entry.path(), errorCode))
{
// TODO: log it
}
}
else if (entry.is_regular_file())
{
// Regular file entry
}
else
{
// Unsupported entry type
// TODO: log it?
}
std::cout << "PATH: " << entry << std::endl;
}
}
src/CopyManager.h
/********************************************************************
* CONFIDENTIAL - All source code is the propriety and confidential *
* information of David Sorber. *
* *
* Copyright 2024 David Sorber *
* Unpublished -- all rights reserved under the copyright laws *
* of the United States. *
********************************************************************/
#ifndef __COPYTOOL_COPYMANAGER_H_
#define __COPYTOOL_COPYMANAGER_H_
#include <atomic>
#include <cstdint>
#include <filesystem>
#include <map>
#include <string>
#include <thread>
#include <vector>
#include "CopyManagerOptions.h"
#include "ProtectedQueue.h"
namespace sfs = std::filesystem;
namespace DBS
{
class CopyManager
{
public:
static inline const sfs::directory_entry ENTRY_TERMINATOR{};
CopyManager(CopyManagerOptions& options);
~CopyManager();
int run();
private:
void mainThreadBody();
void processThreadBody();
CopyManagerOptions& r_options;
ProtectedQueue<sfs::directory_entry> m_entryQueue;
std::atomic_bool m_terminate;
std::thread* p_mainThread;
std::vector<std::thread*> m_procThreads;
};
} // namespace DBS
#endif //__COPYTOOL_COPYMANAGER_H_
src/CopyManagerOptions.cc
/********************************************************************
* CONFIDENTIAL - All source code is the propriety and confidential *
* information of David Sorber. *
* *
* Copyright 2024 David Sorber *
* Unpublished -- all rights reserved under the copyright laws *
* of the United States. *
********************************************************************/
#include "CopyManagerOptions.h"
DBS::CopyManagerOptions::CopyManagerOptions(
const char* srcPath,
const char* destPath)
: m_srcPath(srcPath),
m_destPath(destPath),
m_logLevel(LOG_LEVEL_DEFAULT),
m_flowControlThreshold(FC_THRESHOLD_DEFAULT),
m_numProcThreads(NUM_PROC_THREADS_DEFAULT)
{
}
DBS::CopyManagerOptions::CopyManagerOptions(
const std::string srcPath,
const std::string destPath)
: m_srcPath(srcPath),
m_destPath(destPath),
m_logLevel(LOG_LEVEL_DEFAULT),
m_flowControlThreshold(FC_THRESHOLD_DEFAULT),
m_numProcThreads(NUM_PROC_THREADS_DEFAULT)
{
}
DBS::CopyManagerOptions::~CopyManagerOptions()
{
}
src/CopyManagerOptions.h
/********************************************************************
* CONFIDENTIAL - All source code is the propriety and confidential *
* information of David Sorber. *
* *
* Copyright 2024 David Sorber *
* Unpublished -- all rights reserved under the copyright laws *
* of the United States. *
********************************************************************/
#ifndef __COPYTOOL_COPYMANAGEROPTIONS_H_
#define __COPYTOOL_COPYMANAGEROPTIONS_H_
#include <cstdint>
#include <map>
#include <string>
namespace DBS
{
class CopyManagerOptions
{
public:
static const uint32_t LOG_LEVEL_DEFAULT{3};
static const uint32_t FC_THRESHOLD_DEFAULT{1000};
static const uint32_t NUM_PROC_THREADS_DEFAULT{4};
CopyManagerOptions(
const char* srcPath,
const char* destPath);
CopyManagerOptions(
const std::string srcPath,
const std::string destPath);
~CopyManagerOptions();
//--- Getters -------------------------------------------------------------
inline const std::string& getSrcPath() const
{
return m_srcPath;
}
inline const std::string& getDestPath() const
{
return m_destPath;
}
inline uint32_t getLogLevel() const
{
return m_logLevel;
}
inline uint32_t getFlowControlThreshold() const
{
return m_flowControlThreshold;
}
inline uint32_t getNumProcThreads() const
{
return m_numProcThreads;
}
//--- Setters -------------------------------------------------------------
inline void setFlowControlThreshold(uint32_t threshold)
{
m_flowControlThreshold = threshold;
}
private:
std::string m_srcPath; // source path; "copy from"
std::string m_destPath; // destination path; "copy to"
uint32_t m_logLevel;
uint32_t m_flowControlThreshold;
uint32_t m_numProcThreads;
};
} // namespace DBS
#endif //__COPYTOOL_COPYMANAGEROPTIONS_H_
src/Logger.cc
/********************************************************************
* CONFIDENTIAL - All source code is the propriety and confidential *
* information of David Sorber. *
* *
* Copyright 2024 David Sorber *
* Unpublished -- all rights reserved under the copyright laws *
* of the United States. *
********************************************************************/
#include <cstdio>
#include <chrono>
#include <stdexcept>
#include "Logger.h"
DBS::Logger::Logger(CopyManagerOptions* configObj)
: r_options(*configObj)
{
m_outputStream.str("");
m_logFile.open("loggy"/*r_config.getLogPath()*/, std::ios::out | std::ios::app);
if (! m_logFile)
{
throw std::runtime_error("Unable to open log file: " /*+ r_config.getLogPath()*/);
}
}
DBS::Logger::~Logger()
{
m_logFile.close();
}
void DBS::Logger::log_msg(
enum LOG_LEVEL level,
const std::string& msg)
{
if (level > 3)//r_config.getLogLevel())
{
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|copytool|" << msg << std::endl;
break;
case warn:
m_logFile << time_str << "|WARNING|copytool|" << msg << std::endl;
break;
case info:
m_logFile << time_str << "|INFO|copytool|" << msg << std::endl;
break;
case debug:
m_logFile << time_str << "|DEBUG|copytool|" << msg << std::endl;
break;
case trace:
m_logFile << time_str << "|TRACE|copytool|" << msg << std::endl;
break;
default:
break;
}
}
src/Logger.h
/********************************************************************
* CONFIDENTIAL - All source code is the propriety and confidential *
* information of David Sorber. *
* *
* Copyright 2024 David Sorber *
* Unpublished -- all rights reserved under the copyright laws *
* of the United States. *
********************************************************************/
#ifndef __LOGGER__
#define __LOGGER__
#include <cstring>
#include <iostream>
#include <fstream>
#include <string>
#include <sstream>
#include <mutex>
#include <sys/time.h>
#include "CopyManagerOptions.h"
// Defined outside of namespace just for convenience.
// NOTE: when using >= C++14 and >= -O1 the following function will be executed
// at compile time. This avoids the need for the previous CMake path length trick.
inline constexpr size_t PATH_OFFSET(const char* input)
{
// OTHER NOTE: this assumes truncating the file path after the "lib/"
// portion which is currently valid for LH source tree organization.
const char* output = std::strstr(input, "src/");
if (output == nullptr)
{
return 0;
}
return (output - input) + 4;
}
namespace DBS
{
// Enumeration representing the log severity levels
enum LOG_LEVEL {off=-1, error=0, warn=1, info=2, debug=3, trace=4};
// Regular BlnxLog definition, goes to the log file
#define BlnxLog(sev) BlackLynx::Lighthouse::Logger::getInstance(sev).internalLog() \
<< &(__FILE__[PATH_OFFSET(__FILE__)]) << ":" << __LINE__ << "|"
class Logger
{
public:
/**
* Logger Destructor
*/
~Logger();
/**
* Get the singleton instance
*/
static Logger& getInstance(
enum LOG_LEVEL sev,
CopyManagerOptions* configObj=nullptr)
{
static Logger instance(configObj);
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 LOG_LEVEL severity, Logger &outer)
: m_internalSev(severity),
r_outerLogger(outer)
{
}
InternalLogger(const InternalLogger &other)
: m_internalSev(other.m_internalSev),
r_outerLogger(other.r_outerLogger)
{
}
~InternalLogger()
{
std::string str = m_internalMsgStream.str();
if (! str.empty())
{
r_outerLogger.log_msg(m_internalSev, str);
// Clean the output stream for future logging
m_internalMsgStream.str("");
}
// Make sure we unlock!
r_outerLogger.unlockMutex();
}
/**
* Overriding the insertion operator to build log message
*/
template<typename T>
InternalLogger& operator<< (const T& msg)
{
m_internalMsgStream << msg;
return *this;
}
private:
enum LOG_LEVEL m_internalSev;
Logger& r_outerLogger;
std::stringstream m_internalMsgStream;
};
InternalLogger internalLog()
{
return InternalLogger(m_severity, *this);
}
private:
/**
* Private constructor (singleton)
*/
Logger(CopyManagerOptions* configObj);
/**
* Set the log severity level
*/
void setSeverity(enum LOG_LEVEL sev)
{
m_severity = sev;
}
/**
* Lock the internal mutex
*/
void lockMutex()
{
m_rloggerMutex.lock();
}
/**
* Unlock the internal mutex
*/
void unlockMutex()
{
m_rloggerMutex.unlock();
}
/**
* Write a log message to the log file.
*
* @param level - the severity of the message being logged
* @param message - the message to send
*/
void log_msg(enum LOG_LEVEL level, const std::string& msg);
//--------------------------------------------------------------------------
/**
* The log message to write
*/
std::ostringstream m_outputStream;
/**
* Severity level of the log message
*/
enum LOG_LEVEL m_severity;
/**
* Mutex for making this singleton thread-safe
*/
std::mutex m_rloggerMutex;
std::ofstream m_logFile;
CopyManagerOptions& r_options;
};
} // DBS
#endif // __LOGGER__
src/ProtectedQueue.h
/********************************************************************
* CONFIDENTIAL - All source code is the propriety and confidential *
* information of David Sorber. *
* *
* Copyright 2024 David Sorber *
* Unpublished -- all rights reserved under the copyright laws *
* of the United States. *
********************************************************************/
#ifndef __DBS_PROTECTEDQUEUE_H__
#define __DBS_PROTECTEDQUEUE_H__
#include <condition_variable>
#include <functional>
#include <iostream>
#include <queue>
#include <mutex>
namespace DBS
{
template<class T>
class ProtectedQueue
{
public:
ProtectedQueue()
: m_terminateFlag(false),
p_itemAddedCallback(nullptr),
p_itemRemovedCallback(nullptr)
{}
virtual ~ProtectedQueue() {}
/**
* Add a new element to the back of the queue
*/
inline void push_back(T newElement)
{
{
std::lock_guard<std::mutex> lock(m_mutex);
m_deque.push_back(newElement);
// Call item added callback if one exists
if (p_itemAddedCallback != nullptr)
{
(*p_itemAddedCallback)(m_deque.size());
}
}
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<std::mutex> muxLock(m_mutex);
while (m_deque.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)
{
T elementToReturn{};
return elementToReturn;
}
}
T elementToReturn = m_deque.front();
m_deque.pop_front();
// Call item removed callback if one exists
if (p_itemRemovedCallback != nullptr)
{
(*p_itemRemovedCallback)(m_deque.size());
}
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<std::mutex> muxLock(m_mutex);
if (waitOnEmptyQueue)
{
while (m_deque.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;
}
}
}
T elementToReturn{};
if (m_deque.size() > 0)
{
elementReturned = true;
elementToReturn = m_deque.front();
m_deque.pop_front();
// Call item removed callback if one exists
if (p_itemRemovedCallback != nullptr)
{
(*p_itemRemovedCallback)(m_deque.size());
}
}
else
{
elementReturned = false;
}
return elementToReturn;
}
/**
* Terminate any waiting pop_front() requests
*/
void terminate()
{
m_terminateFlag = true;
m_CV.notify_all();
}
/*
* Resets the terminate flag to allow the queue to function again after
* being terminated previously
*/
void unTerminate()
{
m_terminateFlag = false;
}
void cleanup(std::function<void(T)>& callback)
{
// Call the callback on all remaining items in the queue purging
// each afterwards
{
std::lock_guard<std::mutex> lock(m_mutex);
while (! m_deque.empty())
{
callback(m_deque.front());
m_deque.pop_front();
}
}
m_CV.notify_all();
}
inline bool empty()
{
std::lock_guard<std::mutex> lock(m_mutex);
return m_deque.empty();
}
inline uint32_t size()
{
std::lock_guard<std::mutex> lock(m_mutex);
return m_deque.size();
}
inline void setItemAddedCallback(std::function<void(uint32_t)>& callback)
{
p_itemAddedCallback = &callback;
}
inline void setItemRemovedCallback(std::function<void(uint32_t)>& callback)
{
p_itemRemovedCallback = &callback;
}
/**
* Get the position of item in the queue
*
* @return position of item in queue if it exists, else UINT32_MAX
*/
uint32_t getPositionOf(T item)
{
std::lock_guard<std::mutex> lock(m_mutex);
uint32_t position = 0;
for (auto queueItem : m_deque)
{
if (item == queueItem)
{
return position;
}
position ++;
}
return UINT32_MAX;
}
/**
* Removes item from queue. If item does not exist in queue, nothing
* will happen
*
* @param item Item to remove
*/
void removeItem(T item)
{
std::lock_guard<std::mutex> lock(m_mutex);
// Erase item from deque
auto iter = m_deque.begin();
while (iter != m_deque.end())
{
if (*iter == item)
{
m_deque.erase(iter);
return;
}
++iter;
}
}
private:
bool m_terminateFlag;
std::deque<T> m_deque;
std::mutex m_mutex;
std::condition_variable m_CV;
std::function<void(uint32_t)>* p_itemAddedCallback;
std::function<void(uint32_t)>* p_itemRemovedCallback;
};
}
#endif
src/copytool_main.cc
#include <string>
#include <thread>
#include "CopyManagerOptions.h"
#include "CopyManager.h"
namespace sfs = std::filesystem;
int main(int argc, char** argv)
{
std::cout << "Sendfile Copy Util\n" << std::endl;
if (argc < 2)
if (argc < 3)
{
std::cerr << "ERROR: please specify input path\n" << std::endl;
std::cerr << "ERROR: please specify input path and output path\n" << std::endl;
return -1;
}
const std::string inputPath(argv[1]);
DBS::CopyManagerOptions options(argv[1], argv[2]);
for (const auto& dirEntry : sfs::recursive_directory_iterator(inputPath))
{
std::cout << "PATH: " << dirEntry << std::endl;
}
DBS::CopyManager manager(options);
int rc = manager.run();
return 0;
}

Also available in: Unified diff