commit 2c54c4c31fd414b2d22549e047a36db7522f9d7b Author: david.sorber Date: Thu Jan 18 15:23:03 2024 -0500 Work in progress. diff --git a/CMakeLists.txt b/CMakeLists.txt index 94d2b4b..cf0a146 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -2,6 +2,7 @@ cmake_minimum_required(VERSION 3.22) project("copytool") cmake_policy(SET CMP0054 NEW) + ################################################################################ # Check GCC version ################################################################################ @@ -36,7 +37,7 @@ MESSAGE(STATUS "DISTRIB_RELEASE: " ${DISTRIB_RELEASE}) ################################################################################ -# Configuration +# Build Configuration ################################################################################ # Do not let CMake add the build directory path to RPATH. This means that CL/SL @@ -87,11 +88,23 @@ ELSE() 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 diff --git a/src/CopyManager.cc b/src/CopyManager.cc new file mode 100644 index 0000000..27345ad --- /dev/null +++ b/src/CopyManager.cc @@ -0,0 +1,133 @@ +/******************************************************************** +* 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 +#include + +#include + +#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; + } +} \ No newline at end of file diff --git a/src/CopyManager.h b/src/CopyManager.h new file mode 100644 index 0000000..032d3f7 --- /dev/null +++ b/src/CopyManager.h @@ -0,0 +1,62 @@ +/******************************************************************** +* 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 +#include +#include +#include +#include +#include +#include + +#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 m_entryQueue; + std::atomic_bool m_terminate; + + std::thread* p_mainThread; + std::vector m_procThreads; + +}; + +} // namespace DBS + + +#endif //__COPYTOOL_COPYMANAGER_H_ diff --git a/src/CopyManagerOptions.cc b/src/CopyManagerOptions.cc new file mode 100644 index 0000000..55fdf4b --- /dev/null +++ b/src/CopyManagerOptions.cc @@ -0,0 +1,39 @@ +/******************************************************************** +* 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() +{ + +} \ No newline at end of file diff --git a/src/CopyManagerOptions.h b/src/CopyManagerOptions.h new file mode 100644 index 0000000..5dccc9f --- /dev/null +++ b/src/CopyManagerOptions.h @@ -0,0 +1,84 @@ +/******************************************************************** +* 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 +#include +#include + +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_ diff --git a/src/Logger.cc b/src/Logger.cc new file mode 100644 index 0000000..2508d9e --- /dev/null +++ b/src/Logger.cc @@ -0,0 +1,89 @@ +/******************************************************************** +* 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 +#include +#include + +#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; + } +} \ No newline at end of file diff --git a/src/Logger.h b/src/Logger.h new file mode 100644 index 0000000..545bf6b --- /dev/null +++ b/src/Logger.h @@ -0,0 +1,199 @@ +/******************************************************************** +* 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 +#include +#include +#include +#include +#include +#include + +#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 + 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__ \ No newline at end of file diff --git a/src/ProtectedQueue.h b/src/ProtectedQueue.h new file mode 100644 index 0000000..6324ac6 --- /dev/null +++ b/src/ProtectedQueue.h @@ -0,0 +1,244 @@ +/******************************************************************** +* 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 +#include +#include +#include +#include + + +namespace DBS +{ + +template +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 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 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 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& callback) + { + // Call the callback on all remaining items in the queue purging + // each afterwards + { + std::lock_guard 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 lock(m_mutex); + return m_deque.empty(); + } + + inline uint32_t size() + { + std::lock_guard lock(m_mutex); + return m_deque.size(); + } + + inline void setItemAddedCallback(std::function& callback) + { + p_itemAddedCallback = &callback; + } + + inline void setItemRemovedCallback(std::function& 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 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 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 m_deque; + std::mutex m_mutex; + std::condition_variable m_CV; + + std::function* p_itemAddedCallback; + std::function* p_itemRemovedCallback; +}; + +} + +#endif diff --git a/src/copytool_main.cc b/src/copytool_main.cc index 21de4e5..1f038d7 100644 --- a/src/copytool_main.cc +++ b/src/copytool_main.cc @@ -4,25 +4,27 @@ #include #include +#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; }