rsyncAsyncProcess.cpp

Go to the documentation of this file.

/**
 * @file
 * @ingroup daq_libdaq
 * @copyright (c) Copyright ESO 2022
 * All Rights Reserved
 * ESO (eso.org) is an Intergovernmental Organisation, and therefore special legal conditions apply.
 *
 * @brief daq::RsyncAsyncProcess class definition
 */
#include <daq/process/rsyncAsyncProcess.hpp>
 
#include <array>
#include <string_view>
 
#include <fmt/format.h>
 
namespace daq {
namespace {
 
/**
 * Literal speed units used by rsync.
 */
static constexpr std::array<std::string_view, 4> RSYNC_UNITS = {"B/s", "kB/s", "MB/s", "GB/s"};
 
static std::vector<std::string>
MakeAsyncProcessArgs(std::string source,
                     std::string dest,
                     RsyncOptions const& opts,
                     RsyncAsyncProcess::DryRun flag) {
    std::vector<std::string> args;
    args.emplace_back(opts.rsync.value_or("rsync"));
    if (opts.bw_limit) {
        args.emplace_back(fmt::format("--bwlimit={}", *opts.bw_limit));
    }
    if (opts.whole_file) {
        args.emplace_back(*opts.whole_file ? "--whole-file" : "--no-whole-file");
    }
    if (opts.inplace) {
        args.emplace_back(*opts.whole_file ? "--inplace" : "--no-inplace");
    }
    if (opts.timeout) {
        args.emplace_back(fmt::format("--timeout={}", opts.timeout->count()));
    }
    if (opts.logfile) {
        args.emplace_back(fmt::format("--log-file={}", opts.logfile->native()));
    }
    if (flag == RsyncAsyncProcess::DryRun::Enabled) {
        args.emplace_back("--dry-run");
    }
    // preseve modification times
    args.emplace_back("--times");
    // Mandatory to have functioning progress updates.
    args.emplace_back("--info=progress2");
    // No human readable gets rid of thousand separators in number of transferred bytes.
    args.emplace_back("--no-human-readable");
    args.emplace_back(std::move(source));
    args.emplace_back(std::move(dest));
    return args;
}
 
}  // namespace
 
std::optional<RsyncProgress> ParseRsyncProgress(std::string const& line) noexcept {
    using std::chrono::hours;
    using std::chrono::minutes;
    using std::chrono::seconds;
 
    // Format in the form:
    //           0   0%    0.00kB/s    0:00:00
    //   367656960  34%  350.62MB/s    0:00:01
    //   735313920  68%  350.80MB/s    0:00:00
    //  1073741824 100%  351.29MB/s    0:00:02 (xfr#1 to-chk=0/1)
    RsyncProgress progress;
    unsigned percent;
    float speed;
    char unit[5];
    unsigned hrs;
    unsigned mins;
    unsigned secs;
 
    auto ret = std::sscanf(line.c_str(),
                           " %lu %u%% %f%4s %u:%u:%u",
                           &progress.transferred,
                           &percent,
                           &speed,
                           &unit[0],
                           &hrs,
                           &mins,
                           &secs);
    if (ret != 7) {
        return {};
    }
    progress.progress = percent / 100.0f;
    bool found = false;
    for (auto u : RSYNC_UNITS) {
        if (u == unit) {
            found = true;
            break;
        }
        speed *= 1024.0f;
    }
    if (!found) {
        return {};
    }
    progress.speed = speed;
    progress.remaining = hours(hrs) + minutes(mins) + seconds(secs);
    return progress;
}
 
RsyncAsyncProcess::RsyncAsyncProcess(boost::asio::io_context& ctx,
                                     std::string source,
                                     std::string dest,
                                     RsyncOptions const& opts,
                                     DryRun flag)
    : AsyncProcess(ctx, MakeAsyncProcessArgs(std::move(source), std::move(dest), opts, flag)) {
}
 
RsyncAsyncProcess::~RsyncAsyncProcess() {
}
 
boost::future<int> RsyncAsyncProcess::Initiate() {
    AsyncProcess::ConnectStdout([this](pid_t pid, std::string const& line) mutable {
        // Note: parse failures are ignored
        auto maybe_progress = ParseRsyncProgress(line);
        if (maybe_progress) {
            m_progress(pid, *maybe_progress);
        }
    });
    return AsyncProcess::Initiate();;
}
 
boost::signals2::connection RsyncAsyncProcess::ConnectProgress(SigProgress::slot_type const& slot) {
    return m_progress.connect(slot);
}
 
}  // namespace daq