Project

General

Profile

Download (8.04 KB) Statistics
| Branch: | Revision:

root/src/CopyManager.cc @ eee13fa3

2c54c4c3 david.sorber
/********************************************************************
* 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. *
********************************************************************/
eee13fa3 david.sorber
#include <algorithm>
#include <cerrno>
2c54c4c3 david.sorber
#include <chrono>
eee13fa3 david.sorber
#include <cstring>
#include <string_view>
#include <vector>
2c54c4c3 david.sorber
eee13fa3 david.sorber
#include <sys/types.h>
2c54c4c3 david.sorber
#include <sys/stat.h>
eee13fa3 david.sorber
#include <fcntl.h>
#include <sys/sendfile.h>
2c54c4c3 david.sorber
#include "CopyManager.h"
eee13fa3 david.sorber
#include "Logger.h"
#include "Utilities.h"
2c54c4c3 david.sorber

DBS::CopyManager::CopyManager(CopyManagerOptions& options)
: r_options(options),
m_terminate(false),
p_mainThread(nullptr)
{
}

DBS::CopyManager::~CopyManager()
{
// Add processor thread terminators
eee13fa3 david.sorber
for (uint32_t idx = 0; idx < r_options.getNumCopyThreads(); ++idx)
2c54c4c3 david.sorber
{
m_entryQueue.push_back(ENTRY_TERMINATOR);
}

// Join and delete thread terminators
eee13fa3 david.sorber
for (auto& thread : m_copyThreads)
2c54c4c3 david.sorber
{
thread->join();
delete thread;
}
eee13fa3 david.sorber
m_copyThreads.clear();
2c54c4c3 david.sorber
// 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()
{
eee13fa3 david.sorber
// Copy top level source directory itself to destination directory rather
// than just the contents of the source directory. To support this option
// modify the destination directory path and pre-create the top level
// directory.
if (r_options.getCopyTopLevel())
{
// Determine source top level directory
std::string srcPath{r_options.getSrcPath()};
if (srcPath.back() == '/')
{
srcPath.erase(srcPath.size() - 1);
}

std::vector<std::string> tokens;
tokenize(srcPath, "/", tokens);

sfs::path newDestPath{r_options.getDestPath()};
newDestPath /= tokens.back();

std::error_code errorCode;
if (! sfs::create_directories(newDestPath, errorCode))
{
// Error code 0 means the directory already existed
if (errorCode.value() != 0)
{
LOG(error) << "create_directories(" << newDestPath << ") -- "
<< errorCode.value() << " -- " << errorCode.message();
return -1;
}
}
// TODO: handle top level directory metadata

// Reassign destination directory
r_options.setDestPath(newDestPath);
}

2c54c4c3 david.sorber
// Spawn processor threads
eee13fa3 david.sorber
for (uint32_t idx = 0; idx < r_options.getNumCopyThreads(); ++idx)
2c54c4c3 david.sorber
{
eee13fa3 david.sorber
m_copyThreads.push_back(
new std::thread(&CopyManager::copyThreadBody, this));
2c54c4c3 david.sorber
}

// 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))
{
eee13fa3 david.sorber
// Flow control -- delay until the entry queue is bewlo the specified
// threshold size before proceeding
2c54c4c3 david.sorber
while (m_entryQueue.size() >= r_options.getFlowControlThreshold())
{
eee13fa3 david.sorber
std::this_thread::sleep_for(std::chrono::milliseconds(2));
2c54c4c3 david.sorber
}

// std::cout << "PATH: " << dirEntry << std::endl;
m_entryQueue.push_back(dirEntry);
}

// Add processor thread terminators
eee13fa3 david.sorber
for (uint32_t idx = 0; idx < r_options.getNumCopyThreads(); ++idx)
2c54c4c3 david.sorber
{
m_entryQueue.push_back(ENTRY_TERMINATOR);
}
}

eee13fa3 david.sorber
void DBS::CopyManager::copyThreadBody()
2c54c4c3 david.sorber
{
sfs::directory_entry entry;
struct stat statBuf{};
int rc = 0;
std::error_code errorCode;
eee13fa3 david.sorber
int fromFd = 0;
int toFd = 0;
2c54c4c3 david.sorber
while (true)
{
// Grab entry from queue
entry = m_entryQueue.pop_front();
if (entry == ENTRY_TERMINATOR)
{
break;
}

if (entry.is_directory())
{
// Directory entry
eee13fa3 david.sorber
// const char* destDirPath = entry.path().string().c_str();
// rc = stat(destDirPath, &statBuf);

// Create destination path
auto destDirPath = sfs::path{r_options.getDestPath()} /
sfs::relative(entry.path(),
r_options.getSrcPath());
2c54c4c3 david.sorber
eee13fa3 david.sorber
if (! sfs::create_directories(destDirPath, errorCode))
2c54c4c3 david.sorber
{
eee13fa3 david.sorber
// Error code 0 means the directory already existed
if (errorCode.value() != 0)
{
LOG(error) << "create_directories(" << entry.path() << ") -- "
<< errorCode.value() << " -- " << errorCode.message();
}
2c54c4c3 david.sorber
}
eee13fa3 david.sorber
LOG(debug) << "Created directory: " << entry.path();
2c54c4c3 david.sorber
}
else if (entry.is_regular_file())
{
// Regular file entry
eee13fa3 david.sorber
// Remove filename from file path
auto fileDirPath = entry.path();
fileDirPath = fileDirPath.remove_filename();
// LOG(error) << "FILEDIRPATH: " << fileDirPath;

// Create destination directory path (sans filename)
auto destDirPath = sfs::path{r_options.getDestPath()} /
sfs::relative(fileDirPath,
r_options.getSrcPath());
// LOG(error) << "DIRPATH: " << destDirPath;

// Create destination directory path if it doesn't already exist
if (! sfs::create_directories(destDirPath, errorCode))
{
// Error code 0 means the directory already existed
if (errorCode.value() != 0)
{
LOG(error) << "create_directories(" << entry.path() << ") -- "
<< errorCode.value() << " -- " << errorCode.message();
}
}

// Create destination file path including filename
auto destFilePath = destDirPath / entry.path().filename();

fromFd = open(entry.path().string().c_str(), O_RDONLY);
if (fromFd == -1)
{
LOG(error) << "Unable to open: " << entry.path() << " -- "
<< errno << " -- " << std::strerror(errno);
continue;
}

fstat(fromFd, &statBuf);
// TODO: check return code

toFd = open(destFilePath.string().c_str(),
O_WRONLY | O_CREAT, statBuf.st_mode);
if (toFd == -1)
{
LOG(error) << "Unable to open: " << entry.path() << " -- "
<< errno << " -- " << std::strerror(errno);
continue;
}

// Determine sendfile() "count" parameter which must be no more than
// MAX_COPY_SIZE per sendfile() documentation.
ssize_t count = std::min(sfs::file_size(entry.path()), MAX_COPY_SIZE);

// Iterate as needed calling sendfile() to copy data from the source file
// to the destination file
rc = 1;
while (rc > 0)
{
rc = sendfile(toFd, fromFd, 0, count);
}
if (rc < 0)
{
LOG(error) << "Error while writing: " << destFilePath << " -- "
<< errno << " -- " << std::strerror(errno);
continue;
}

LOG(debug) << "Copy: " << entry.path().string()
<< " to: " << destFilePath.string();
2c54c4c3 david.sorber
}
else
{
// Unsupported entry type
eee13fa3 david.sorber
LOG(error) << "Unsupported entry type: " << entry.path();
2c54c4c3 david.sorber
}

}
}