Revision eee13fa3
Added by david.sorber over 2 years ago
| src/CopyManager.cc | ||
|---|---|---|
|
* Unpublished -- all rights reserved under the copyright laws *
|
||
|
* of the United States. *
|
||
|
********************************************************************/
|
||
|
#include <algorithm>
|
||
|
#include <cerrno>
|
||
|
#include <chrono>
|
||
|
#include <system_error>
|
||
|
#include <cstring>
|
||
|
#include <string_view>
|
||
|
#include <vector>
|
||
|
|
||
|
#include <sys/types.h>
|
||
|
#include <sys/stat.h>
|
||
|
#include <fcntl.h>
|
||
|
#include <sys/sendfile.h>
|
||
|
|
||
|
#include "CopyManager.h"
|
||
|
#include "Logger.h"
|
||
|
#include "Utilities.h"
|
||
|
|
||
|
|
||
|
DBS::CopyManager::CopyManager(CopyManagerOptions& options)
|
||
| ... | ... | |
|
DBS::CopyManager::~CopyManager()
|
||
|
{
|
||
|
// Add processor thread terminators
|
||
|
for (uint32_t idx = 0; idx < r_options.getNumProcThreads(); ++idx)
|
||
|
for (uint32_t idx = 0; idx < r_options.getNumCopyThreads(); ++idx)
|
||
|
{
|
||
|
m_entryQueue.push_back(ENTRY_TERMINATOR);
|
||
|
}
|
||
|
|
||
|
// Join and delete thread terminators
|
||
|
for (auto& thread : m_procThreads)
|
||
|
for (auto& thread : m_copyThreads)
|
||
|
{
|
||
|
thread->join();
|
||
|
delete thread;
|
||
|
}
|
||
|
m_procThreads.clear();
|
||
|
m_copyThreads.clear();
|
||
|
|
||
|
// Cleanup the main thread
|
||
|
if (p_mainThread != nullptr)
|
||
| ... | ... | |
|
|
||
|
int DBS::CopyManager::run()
|
||
|
{
|
||
|
// 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);
|
||
|
}
|
||
|
|
||
|
// Spawn processor threads
|
||
|
for (uint32_t idx = 0; idx < r_options.getNumProcThreads(); ++idx)
|
||
|
for (uint32_t idx = 0; idx < r_options.getNumCopyThreads(); ++idx)
|
||
|
{
|
||
|
m_procThreads.push_back(
|
||
|
new std::thread(&CopyManager::processThreadBody, this));
|
||
|
m_copyThreads.push_back(
|
||
|
new std::thread(&CopyManager::copyThreadBody, this));
|
||
|
}
|
||
|
|
||
|
// Spawn the main thread
|
||
|
p_mainThread = new std::thread(&CopyManager::mainThreadBody, this);
|
||
|
|
||
|
|
||
|
p_mainThread->join();
|
||
|
|
||
|
return 0;
|
||
| ... | ... | |
|
const std::string& srcPath(r_options.getSrcPath());
|
||
|
for (const auto& dirEntry : sfs::recursive_directory_iterator(srcPath))
|
||
|
{
|
||
|
// Flow control
|
||
|
// Flow control -- delay until the entry queue is bewlo the specified
|
||
|
// threshold size before proceeding
|
||
|
while (m_entryQueue.size() >= r_options.getFlowControlThreshold())
|
||
|
{
|
||
|
std::this_thread::sleep_for(std::chrono::milliseconds(5));
|
||
|
std::this_thread::sleep_for(std::chrono::milliseconds(2));
|
||
|
}
|
||
|
|
||
|
// std::cout << "PATH: " << dirEntry << std::endl;
|
||
| ... | ... | |
|
}
|
||
|
|
||
|
// Add processor thread terminators
|
||
|
for (uint32_t idx = 0; idx < r_options.getNumProcThreads(); ++idx)
|
||
|
for (uint32_t idx = 0; idx < r_options.getNumCopyThreads(); ++idx)
|
||
|
{
|
||
|
m_entryQueue.push_back(ENTRY_TERMINATOR);
|
||
|
}
|
||
|
}
|
||
|
|
||
|
|
||
|
void DBS::CopyManager::processThreadBody()
|
||
|
void DBS::CopyManager::copyThreadBody()
|
||
|
{
|
||
|
sfs::directory_entry entry;
|
||
|
struct stat statBuf{};
|
||
|
int rc = 0;
|
||
|
std::error_code errorCode;
|
||
|
int fromFd = 0;
|
||
|
int toFd = 0;
|
||
|
|
||
|
while (true)
|
||
|
{
|
||
| ... | ... | |
|
if (entry.is_directory())
|
||
|
{
|
||
|
// Directory entry
|
||
|
const char* dirPath = entry.path().string().c_str();
|
||
|
rc = stat(dirPath, &statBuf);
|
||
|
// 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());
|
||
|
|
||
|
if (! sfs::create_directories(entry.path(), errorCode))
|
||
|
if (! sfs::create_directories(destDirPath, errorCode))
|
||
|
{
|
||
|
// TODO: log it
|
||
|
// Error code 0 means the directory already existed
|
||
|
if (errorCode.value() != 0)
|
||
|
{
|
||
|
LOG(error) << "create_directories(" << entry.path() << ") -- "
|
||
|
<< errorCode.value() << " -- " << errorCode.message();
|
||
|
}
|
||
|
}
|
||
|
|
||
|
LOG(debug) << "Created directory: " << entry.path();
|
||
|
}
|
||
|
else if (entry.is_regular_file())
|
||
|
{
|
||
|
// Regular file entry
|
||
|
|
||
|
// 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();
|
||
|
}
|
||
|
else
|
||
|
{
|
||
|
// Unsupported entry type
|
||
|
// TODO: log it?
|
||
|
LOG(error) << "Unsupported entry type: " << entry.path();
|
||
|
}
|
||
|
|
||
|
|
||
|
std::cout << "PATH: " << entry << std::endl;
|
||
|
}
|
||
|
}
|
||
First mostly functional version; let's call it v0.1.0.