scheduler.cpp

Go to the documentation of this file.

/**
 * @file
 * @ingroup daq_dpm
 * @copyright (c) Copyright ESO 2022
 * All Rights Reserved
 * ESO (eso.org) is an Intergovernmental Organisation, and therefore special legal conditions apply.
 *
 * @brief daq::dpm::Scheduler implementation
 */
#include <daq/dpm/scheduler.hpp>
 
#include <algorithm>
 
#include <boost/asio/post.hpp>
#include <boost/range/adaptor/indexed.hpp>
#include <fmt/format.h>
#include <fmt/std.h>
#include <log4cplus/loggingmacros.h>
 
#include <daq/dpm/config.hpp>
#include <daq/dpm/workspace.hpp>
#include <daq/error/report.hpp>
#include <daq/json.hpp>
#include <daq/log4cplus.hpp>
 
namespace daq::dpm {
 
namespace {
void PrintArgs(log4cplus::Logger& logger, std::vector<std::string> const& args) {
    std::stringstream ss;
    ss << "{";
    bool first = true;
    for (auto const& token : args) {
        if (!first) {
            ss << ", ";
        }
        ss << token;
        first = false;
    }
    ss << "}";
    LOG4CPLUS_DEBUG(logger, "Executing merger with args: " << ss.str());
}
 
void CheckConsistency(Status const& status, json::DpSpec const& spec) {
    if (status.id != spec.id) {
        throw std::invalid_argument(
            fmt::format("Scheduler: Could not queue DAQ for merging as "
                        "provided status does not match provided data product specification "
                        "w.r.t DAQ Id: {} vs {}",
                        status.id,
                        spec.id));
    }
    if (status.file_id != spec.target.file_id) {
        throw std::invalid_argument(
            fmt::format("Scheduler: Could not queue DAQ for merging as "
                        "provided status does not match provided data product specification "
                        "w.r.t DAQ File Id: {} vs {}",
                        status.file_id,
                        spec.target.file_id));
    }
}
 
auto ParseDpSpec(log4cplus::Logger const& logger, std::string const& spec) -> json::DpSpec try {
    return json::ParseDpSpec(nlohmann::json::parse(spec));
} catch (json::DpSpecError const&) {
    LOG4CPLUS_ERROR(logger, "Invalid Data Product Specification provided: \n" << spec);
 
    std::throw_with_nested(
        std::invalid_argument("Scheduler: Could not queue DAQ for merging because "
                              "Data Product Specification is invalid"));
} catch (nlohmann::json::parse_error const&) {
    LOG4CPLUS_ERROR(logger, "Invalid Data Product Specification provided: \n" << spec);
    std::throw_with_nested(
        std::invalid_argument("Scheduler: Could not queue DAQ for merging because provided "
                              "data product specification is invalid JSON"));
}
 
auto ParseStatus(log4cplus::Logger const& logger, std::string const& status)
    -> std::optional<Status> try {
    if (status.empty()) {
        return std::nullopt;
    }
    auto json = nlohmann::json::parse(status);
    return json.get<Status>();
} catch (nlohmann::json::parse_error const&) {
    std::throw_with_nested(
        std::invalid_argument("Scheduler: Could not queue DAQ for merging because provided "
                              "DAQ status is invalid JSON"));
}
 
}  // namespace
 
std::ostream& operator<<(std::ostream& os, DaqController const& daq) {
    os << "DAQ{" << daq.GetStatus() << "}";
    return os;
}
 
SchedulerImpl::SchedulerImpl(rad::IoExecutor& executor,
                             Workspace& workspace,
                             DaqControllerFactory daq_controller_factory,
                             SchedulerOptions const& options)
    : m_executor(executor)
    , m_workspace(workspace)
    , m_daq_controller_factory(std::move(daq_controller_factory))
    , m_options(options)
    , m_logger(log4cplus::Logger::getInstance(LOGGER_NAME_SCHEDULER))
    , m_liveness(std::make_shared<bool>(false)) {
    m_queue = m_workspace.LoadQueue();
    auto slot = [liveness = std::weak_ptr<bool>(m_liveness), this]() {
        if (auto ptr = liveness.lock(); ptr) {
            // Scheduler still alive
            DeferredPoll();
        }
    };
    auto& conns = m_resources_connections;
    conns.daqs = m_resources.daqs.Connect(slot);
    conns.net_receive = m_resources.net_receive.Connect(slot);
    conns.merge = m_resources.merge.Connect(slot);
    conns.net_send = m_resources.net_send.Connect(slot);
}
 
void SchedulerImpl::Start() {
    m_stopped = false;
    DeferredPoll();
}
 
void SchedulerImpl::Stop() {
    m_stopped = true;
}
 
std::string SchedulerImpl::QueueDaq(std::string const& dp_spec_serialized,
                                    std::string const& status_serialized) {
    LOG4CPLUS_TRACE(m_logger, "QueueDaq()");
    try {
        json::DpSpec dp_spec = ParseDpSpec(m_logger, dp_spec_serialized);
        Status initial_status = ParseStatus(m_logger, status_serialized)
                                    .value_or(Status(dp_spec.id, dp_spec.target.file_id));
        CheckConsistency(initial_status, dp_spec);
        std::string const& id = dp_spec.id;
        if (IsQueued(id)) {
            LOG4CPLUS_ERROR(m_logger, "QueueDaq(): DAQ conflict detected -> aborting");
            throw std::invalid_argument(
                fmt::format("Scheduler: Could not queue DAQ for merging as "
                            "a Data Acquisition with same id has been queued before: '{}'",
                            id));
        }
        // New daq for this workspace -> Initialize persistent state.
        try {
            LOG4CPLUS_INFO(m_logger,
                           fmt::format("QueueDaq(): Initializing new workspace for DAQ {}", id));
            auto daq_ws = m_workspace.InitializeDaq(id);
            assert(daq_ws);
            try {
                assert(daq_ws);
                // Initial state in DPM is Scheduled as that's where DPM takes over.
                initial_status.state = State::Scheduled;
                initial_status.timestamp = Status::TimePoint::clock::now();
 
                daq_ws->StoreStatus(initial_status);
                daq_ws->StoreSpecification(dp_spec_serialized);
 
                try {
                    m_queue.push_back(id);
                    // Finally update the backlog
                    m_workspace.StoreQueue(m_queue);
 
                    DeferredPoll();
                    return id;
                } catch (...) {
                    // Undo push
                    m_queue.pop_back();
                    std::throw_with_nested(std::runtime_error("Failed to store DAQ queue"));
                }
            } catch (...) {
                std::throw_with_nested(
                    std::runtime_error("Failed to write status to DAQ workspace"));
            }
        } catch (...) {
            // Undo creation of DAQ workspace
            m_workspace.RemoveDaq(id);
            std::throw_with_nested(std::runtime_error(fmt::format(
                "Failed to initialize DAQ workspace in {}", m_workspace.GetPath().native())));
        }
    } catch (std::invalid_argument const& e) {
        LOG4CPLUS_ERROR(m_logger, "Failed to queue DAQ: " << e.what());
        std::throw_with_nested(std::invalid_argument("Scheduler: Failed to queue DAQ"));
    } catch (std::exception const& e) {
        LOG4CPLUS_ERROR(m_logger, "Failed to queue DAQ: " << e.what());
        std::throw_with_nested(std::runtime_error("Scheduler: Failed to queue DAQ"));
    }
}
 
bool SchedulerImpl::IsQueued(std::string const& id) const noexcept {
    return std::find(m_queue.begin(), m_queue.end(), id) != m_queue.end();
}
 
Status SchedulerImpl::GetDaqStatus(std::string const& id) const try {
    LOG4CPLUS_TRACE(m_logger, fmt::format("GetDaqStatus({})", id));
    if (!IsQueued(id)) {
        // Try to find it in the archive.
        auto maybe_status = m_workspace.LoadArchivedStatus(id);
        if (maybe_status) {
            return *maybe_status;
        } else {
            LOG4CPLUS_INFO(m_logger, fmt::format("GetDaqStatus({}): No such ID", id));
            throw std::invalid_argument(fmt::format("DAQ with id='{}' not found", id));
        }
    }
    return m_workspace.LoadDaq(id)->LoadStatus();
} catch (...) {
    std::throw_with_nested(std::runtime_error("Scheduler: GetDaqStatus() failed"));
}
 
std::vector<std::string> SchedulerImpl::GetQueue() const noexcept {
    return m_queue;
}
 
void SchedulerImpl::AbortDaq(std::string const& id) try {
    LOG4CPLUS_TRACE(m_logger, fmt::format("AbortDaq({})", id));
    // - If DAQ is either being released or is already in final state it cannot be aborted.
    // - If DAQ is active must first be made inactive by destroying DaqController which will also
    //   terminate started processes.
    // - If DAQ is not active aborting simply means to erase DAQ from filesystem and queue.
    if (!IsQueued(id)) {
        LOG4CPLUS_INFO(m_logger, fmt::format("AbortDaq({}): No such ID", id));
        throw std::invalid_argument(fmt::format("DAQ with id='{}' not found", id));
    }
 
    // If DAQ is active/in-progress we must first delete it
    auto it = std::find_if(m_active.begin(), m_active.end(), [&id](Active const& active) {
        assert(active.daq);
        return id == active.daq->GetId();
    });
 
    // Status to update with final aborted state
    auto pub = [&, status = m_workspace.LoadDaq(id)->LoadStatus()](State new_state) mutable {
        status.state = State::AbortingMerging;
        m_status_signal(status);
    };
 
    if (it != m_active.end()) {
        auto state = it->daq->GetState();
        if (IsAbortableState(state)) {
            LOG4CPLUS_DEBUG(m_logger,
                            fmt::format("AbortDaq({}): Cannot abort in state {}", id, state));
        }
        // Erase if active
        LOG4CPLUS_DEBUG(
            m_logger,
            fmt::format("AbortDaq({}): Erasing active DAQ currently in state {}", id, state));
        m_active.erase(it);
    }
 
    // Publish that we're aborting
    pub(State::AbortingMerging);
 
    // Erase id from queue
    LOG4CPLUS_DEBUG(m_logger, fmt::format("AbortDaq({}): Removing DAQ from merge queue", id));
    m_queue.erase(std::find(m_queue.begin(), m_queue.end(), id));
    m_workspace.StoreQueue(m_queue);
 
    // Delete workspace
    LOG4CPLUS_DEBUG(m_logger, fmt::format("AbortDaq({}): Removing DAQ workspace", id));
    m_workspace.RemoveDaq(id);
 
    // Publish that DAQ is aborted
    pub(State::Aborted);
 
    // Queue has changed -> Poll
    DeferredPoll();
    LOG4CPLUS_INFO(m_logger, fmt::format("AbortDaq({}): Aborted and removed DAQ", id));
} catch (...) {
    std::throw_with_nested(std::runtime_error("Scheduler: AbortDaq() failed"));
}
 
boost::signals2::connection SchedulerImpl::ConnectStatus(StatusSignal::slot_type const& slot) {
    return m_status_signal.connect(slot);
}
 
void SchedulerImpl::Poll() {
    LOG4CPLUS_TRACE(m_logger, "Poll()");
    ArchiveCompleted();
    ActivateFromQueue();
}
 
void SchedulerImpl::ArchiveCompleted() {
    // Archive completed DAQs
    for (auto& active : m_active) {  // auto it = remove_it; it != m_active.end(); ++it) {
        try {
            if (!IsFinalState(active.daq->GetState())) {
                continue;
            }
            auto id = active.daq->GetId();
 
            auto archive_path = m_workspace.ArchiveDaq(id);
            LOG4CPLUS_INFO(
                m_logger,
                fmt::format("DAQ {} is in final state {} -> moved workspace to archive: {}",
                            id,
                            active.daq->GetState(),
                            archive_path.native()));
 
            // Delete controller
            active.daq.reset();
 
            // Remove ID from queue
            m_queue.erase(std::find(m_queue.begin(), m_queue.end(), id));
            m_workspace.StoreQueue(m_queue);
        } catch (...) {
            // @todo: Decide what to do here...
            LOG4CPLUS_ERROR(m_logger,
                            "Failed to archive DAQ workspace:\n"
                                << error::NestedExceptionReporter(std::current_exception()));
        }
    }
 
    // Erase removed daqs
    auto remove_it = std::remove_if(
        m_active.begin(), m_active.end(), [&](auto const& active) { return !active.daq; });
    if (remove_it != m_active.end()) {
        m_active.erase(remove_it, m_active.end());
        // Since this freed resources we schedule poll() to be run again.
        DeferredPoll();
    }
}
 
void SchedulerImpl::ActivateFromQueue() {
    // Update active DAQs
    // See if a new DAQ can be active.
    if (m_queue.empty()) {
        LOG4CPLUS_INFO(m_logger, "Merge queue empty - all done!");
        return;
    }
 
    auto candidates = GetCandidates();
    if (candidates.empty()) {
        LOG4CPLUS_INFO(m_logger, "All DAQ merge candidates are active/in-progress");
        return;
    }
    // Schedule as many candidates as allowed by resource limits
    try {
        for (auto const& id : candidates) {
            LOG4CPLUS_TRACE(m_logger, fmt::format("{}: Attempting to schedule DAQ", id));
            auto maybe_token = m_resources.daqs.Acquire();
            if (!maybe_token) {
                LOG4CPLUS_INFO(m_logger,
                               fmt::format("Limit reached: Cannot schedule '{}' "
                                           "Currently active DAQs at/exceed limit. "
                                           "current: {}, limit: {}, queue size: {}",
                                           id,
                                           m_resources.daqs.GetUsed(),
                                           m_resources.daqs.GetLimit(),
                                           m_queue.size()));
                return;
            }
 
            try {
                // @todo: Store token with DAQ?
                auto ws = m_workspace.LoadDaq(id);
                auto daq_controller = m_daq_controller_factory(std::move(ws), m_resources);
 
                // Aggregate signal
                // both DaqController and slot are owned by Scheduler so we don't need
                // to manage connection lifetime.
                daq_controller->GetStatus().ConnectStatus([&](ObservableStatus const& s) {
                    // Completed DAQs should be archived so we schedule a poll in that case.
                    if (IsFinalState(s.GetState())) {
                        DeferredPoll();
                    }
                    // Just forward status
                    m_status_signal(s.GetStatus());
                });
                // Tell DaqController to start
                daq_controller->Start();
 
                m_active.emplace_back(std::move(daq_controller), std::move(*maybe_token));
 
                LOG4CPLUS_DEBUG(m_logger, fmt::format("{}: DAQ scheduled for merging", id));
                // Schedule new Poll() as queue has been modified.
                DeferredPoll();
            } catch (...) {
                std::throw_with_nested(
                    std::runtime_error(fmt::format("{}: Failed to activate DAQ for merging.", id)));
            }
        }
    } catch (...) {
        std::throw_with_nested(std::runtime_error("Failed to load "));
    }
}
 
void SchedulerImpl::DeferredPoll() {
    if (m_stopped || *m_liveness) {
        // Stopped or already scheduled
        return;
    }
    boost::asio::post(m_executor.get_io_context().get_executor(),
                      [liveness = std::weak_ptr<bool>(m_liveness), this]() {
                          if (auto ptr = liveness.lock(); ptr) {
                              // Scheduler still alive
                              // Clear flag before invoking Poll so that another
                              // poll can be scheduled.
                              *ptr = false;
                              Poll();
                          }
                      });
}
 
std::vector<std::string> SchedulerImpl::GetCandidates() const {
    // Return all DAQs in m_queue that is not in m_active.
    std::vector<std::string> candidates;
    for (auto const& id : m_queue) {
        auto it = std::find_if(m_active.begin(), m_active.end(), [&](auto const& active) {
            assert(active.daq);
            return active.daq->GetId() == id;
        });
        if (it == m_active.end()) {
            candidates.push_back(id);
        }
    }
    return candidates;
}
 
DaqControllerImpl::DaqControllerImpl(rad::IoExecutor& executor,
                                     std::unique_ptr<DaqWorkspace> workspace,
                                     Resources& resources,
                                     RsyncFactory rsync_factory,
                                     ProcFactory proc_factory,
                                     DaqControllerOptions options)
    : m_executor(executor)
    , m_workspace(std::move(workspace))
    , m_resources(resources)
    , m_rsync_factory(std::move(rsync_factory))
    , m_proc_factory(std::move(proc_factory))
    , m_options(std::move(options))
    , m_dpspec(m_workspace->LoadSpecification())
    , m_result()
    , m_state_ctx(Scheduled{})  // m_state_ctx is updated in body with actual state.
    , m_status(m_workspace->LoadStatus())
    , m_status_connection()
    , m_liveness(std::make_shared<bool>(false))
    , m_stopped(true)
    , m_logger(log4cplus::Logger::getInstance(LOGGER_NAME_CONTROLLER)) {
    if (m_options.merge_bin.empty()) {
        throw std::invalid_argument("Specified merge application name is empty");
    }
    if (m_options.rsync_bin.empty()) {
        throw std::invalid_argument("Specified rsync application name is empty");
    }
 
    LOG4CPLUS_DEBUG(m_logger,
                    "Loaded status: " << m_status << "\nAlerts: " << m_status.GetAlerts());
 
    // If result has already been created it is contained in the stored status.
    // Otherwise we set to-be-used absolute path
    if (m_status.GetStatus().result.empty()) {
        m_result = m_workspace->GetResultPath() /
                   (m_dpspec.target.file_prefix + m_dpspec.target.file_id + ".fits");
    } else {
        m_result = m_status.GetStatus().result;
    }
 
    // Update m_state from loaded status.
    switch (m_status.GetState()) {
    case State::Scheduled:
        SetState(Scheduled{});
        break;
    case State::Collecting:
        SetState(Collecting{SourceResolver(m_workspace->LoadSourceLookup())});
        break;
    case State::Merging:
        SetState(Merging());
        break;
    case State::Releasing:
        SetState(Releasing{});
        break;
    case State::Completed:
        SetState(Completed{});
        break;
    default:
        throw std::runtime_error("Not implemented");
    };
 
    m_status_connection = m_status.ConnectStatus([this](ObservableStatus const& status) mutable {
        if (status.ChangesSinceLastSignal() == 0u) {
            // No changes -> don't write
            LOG4CPLUS_TRACE(
                m_logger,
                fmt::format(
                    "DaqController: No change -> skipping write of workspace status file: {}",
                    status));
            return;
        }
        LOG4CPLUS_TRACE(m_logger,
                        fmt::format("DaqController: Updating workspace status file: {}", status));
        m_workspace->StoreStatus(status);
    });
}
 
void DaqControllerImpl::Start() {
    m_stopped = false;
    DeferredPoll();
}
 
void DaqControllerImpl::Stop() {
    m_stopped = true;
}
 
auto DaqControllerImpl::IsStopped() const noexcept -> bool {
    return m_stopped;
}
 
auto DaqControllerImpl::GetId() const noexcept -> std::string const& {
    return m_status.GetId();
}
 
auto DaqControllerImpl::GetErrorFlag() const noexcept -> bool {
    return m_status.HasError();
}
 
auto DaqControllerImpl::GetState() const noexcept -> State {
    return m_status.GetState();
}
 
auto DaqControllerImpl::GetStatus() noexcept -> ObservableStatus& {
    return m_status;
}
 
auto DaqControllerImpl::GetStatus() const noexcept -> ObservableStatus const& {
    return m_status;
}
 
void DaqControllerImpl::DeferredPoll() {
    if (m_stopped || *m_liveness) {
        // Stopped or already scheduled
        return;
    }
    boost::asio::post(m_executor.get_io_context().get_executor(),
                      [liveness = std::weak_ptr<bool>(m_liveness), this]() {
                          if (auto ptr = liveness.lock(); ptr) {
                              // Scheduler still alive
                              // Clear flag before invoking Poll so that another
                              // poll can be scheduled.
                              *ptr = false;
                              Poll();
                          }
                      });
}
 
void DaqControllerImpl::Poll() {
    LOG4CPLUS_TRACE(m_logger, "Poll()");
    if (m_stopped) {
        LOG4CPLUS_DEBUG(
            m_logger,
            fmt::format("{}: Poll() - DaqController is stopped so nothing will be done", *this));
        // Stopped or already scheduled
        return;
    }
    std::visit([this](auto& state) { Poll(state); }, m_state_ctx);
}
 
void DaqControllerImpl::Poll(DaqControllerImpl::Scheduled&) {
    LOG4CPLUS_TRACE(m_logger, "Poll(Scheduled)");
    try {
        // Create list of remote sources and build mapping to local files.
        std::vector<json::FitsFileSource> sources;
        if (m_dpspec.target.source) {
            sources.push_back(*m_dpspec.target.source);
        }
        for (auto const& s : m_dpspec.sources) {
            if (!std::holds_alternative<json::FitsFileSource>(s)) {
                continue;
            }
            sources.push_back(std::get<json::FitsFileSource>(s));
        }
 
        auto sources_path = m_workspace->GetSourcesPath();
 
        SourceResolver resolver;
        using boost::adaptors::indexed;
        for (auto const& it : sources | indexed(0u)) {
            auto index = it.index();
            auto const& s = it.value();
            json::Location location = json::ParseSourceLocation(s.location);
            // Use <index>_<source>_<filename> as local filename.
            auto local_path =
                sources_path /
                fmt::format("{}_{}_{}", index, s.source_name, location.path.filename().native());
            LOG4CPLUS_INFO(m_logger,
                           fmt::format("Poll(Scheduled): Source file '{}' from source \"{}\" "
                                       "on host \"{}\" will be stored in {}",
                                       location.path,
                                       s.source_name,
                                       !location.host.empty() ? location.host.c_str() : "<n/a>",
                                       local_path));
 
            resolver.Add({s.source_name, s.location}, local_path);
        }
 
        // Store source resolution mapping.
        m_workspace->StoreSourceLookup(resolver.GetMapping());
 
        SetState(Collecting(std::move(resolver)));
    } catch (...) {
        auto msg =
            fmt::format("{}: Failed to collect and store list of required sources", m_status);
        LOG4CPLUS_ERROR(m_logger, fmt::format("Poll(Scheduled): Failed to process DAQ: {}", msg));
        // No point in retrying these kinds of errors.
        Stop();
        std::throw_with_nested(std::runtime_error(msg));
    }
}
 
void DaqControllerImpl::Poll(Collecting& ctx) {
    LOG4CPLUS_TRACE(m_logger, "Poll(Collecting)");
    try {
        // Check if any source is missing and start transfers if possible
        // If all transfers are done transition to State::Merging.
        auto defer = ObservableStatus::DeferSignal(&m_status);
 
        auto const& sources = ctx.resolver.GetMapping();
        auto missing = sources.size();
        auto root = m_workspace->GetPath();
        for (auto const& source : sources) {
            auto dest = source.second;
            if (ctx.HasTransfer(source.first)) {
                // Transfer already in progress
                continue;
            }
            // [recovery]
            if (m_workspace->Exists(dest)) {
                LOG4CPLUS_DEBUG(
                    m_logger,
                    fmt::format("Poll(Collecting): File exist -> won't start new transfer: {}",
                                dest));
                // Clear alert if there was a previous issue.
                m_status.ClearAlert(MakeAlertId(alert::COLLECTING_RSYNC, dest));
                missing--;
                // File already exist
                continue;
            }
            if (m_soft_stop) {
                // If there are errors we don't initiate new transfers
                LOG4CPLUS_DEBUG(m_logger,
                                "Poll(Collecting): Soft stop is enabled"
                                "-> won't start new transfer");
                continue;
            }
            // We need to transfer the file, try to start.
            auto token = m_resources.net_receive.Acquire();
            if (token) {
                // We got the token, so we can start transfer
                json::Location location = json::ParseSourceLocation(source.first.location);
                using Proc = std::shared_ptr<RsyncAsyncProcessIf>;
                RsyncOptions opts;
                opts.rsync = m_options.rsync_bin;
                auto log_file =
                    std::filesystem::path(dest).filename().replace_extension("rsync.log");
                opts.logfile = m_workspace->GetPath() / m_workspace->GetLogsPath() / log_file;
                Proc proc = m_rsync_factory(m_executor.get_io_context(),
                                            location.RsyncPath(),
                                            root / dest,
                                            opts,
                                            RsyncAsyncProcessIf::DryRun::Disabled);
 
                auto last_lines = std::make_unique<LogCaptureLast>(
                    log4cplus::Logger::getInstance(LOGGER_NAME_TRANSFER), 10);
                // Connect signals (stdout is ignored for now as it merely contains the progress
                // which will be handled differently later on).
                proc->ConnectStderr(std::ref(*last_lines));
                // Start transfer and set up completion handler
                proc->Initiate().then(
                    m_executor,
                    [liveness = std::weak_ptr<bool>(m_liveness),
                     source = source.first,
                     dest = dest,
                     proc = proc,
                     last_lines = std::move(last_lines),
                     this](boost::future<int> f) {
                        if (liveness.expired()) {
                            LOG4CPLUS_ERROR("dpm.daqcontroller",
                                            fmt::format("DaqController abandoned -> ignoring "
                                                        "result from rsync for transfer of {}",
                                                        source));
                            return;
                        }
                        TransferComplete(source, dest, std::move(f), *last_lines);
                    });
                LOG4CPLUS_DEBUG(
                    m_logger,
                    fmt::format(
                        "{}: Started transfer: {} -> {}: {}", *this, source.first, dest, *proc));
                ctx.transfers.emplace_back(Collecting::Transfer{
                    source.first, source.second, std::move(proc), std::move(*token)});
            } else {
                // Out of tokens -> bail out
                LOG4CPLUS_TRACE(m_logger,
                                fmt::format("Poll(Collecting): Could not start transfer due to "
                                            "resource limit reached: {}",
                                            source.first));
                return;
            }
        }
 
        if (missing == 0) {
            // If all transfers are complete we can transition to next state
            SetState(Merging());
            return;
        }
    } catch (...) {
        // No point in retrying these kinds of errors.
        auto msg = fmt::format("{}: Failed to transfer required sources", m_status);
        LOG4CPLUS_ERROR(m_logger, fmt::format("Poll(Collecting): Failed to process DAQ: {}", msg));
        Stop();
        std::throw_with_nested(std::runtime_error(msg));
    }
}
 
void DaqControllerImpl::TransferComplete(SourceResolver::SourceFile const& source,
                                         std::filesystem::path const& local_path,
                                         boost::future<int> result,
                                         LogCaptureLast const& log) noexcept {
    LOG4CPLUS_TRACE(m_logger,
                    fmt::format("{}: TransferComplete: {} -> {}", *this, source, local_path));
    auto* ctx = std::get_if<Collecting>(&m_state_ctx);
    try {
        if (ctx) {
            ctx->EraseTransfer(source);
        }
 
        int return_code = result.get();
        auto alert =
            ObservableStatus::AlertActivator(&m_status, alert::COLLECTING_RSYNC, local_path);
        if (return_code != 0) {
            auto msg = fmt::format(
                "rsync file transfer failed for remote file {}. Last lines: \n{}", source, log);
            alert.Set(msg);
            LOG4CPLUS_ERROR(m_logger, msg);
            throw std::runtime_error(msg);
        }
        // Opportunistically clear alert in case it as set from previous attempt
        alert.Clear();
        LOG4CPLUS_INFO(m_logger,
                       fmt::format("{}: Poll(Collecting): Successfully transferred file: {} -> {}",
                                   *this,
                                   source,
                                   local_path));
 
        // Transfer ok
        DeferredPoll();
        return;
    } catch (...) {
        LOG4CPLUS_ERROR(m_logger,
                        fmt::format("{}: Poll(Collecting): Failed to transfer file: {} -> {}",
                                    *this,
                                    source,
                                    local_path));
        // Stopping DAQ can be done when there's no more pending file transfers (we let them
        // complete).
        if (ctx && ctx->transfers.empty()) {
            LOG4CPLUS_ERROR(
                m_logger,
                fmt::format("{}: Poll(Collecting): All pending transfers have completed "
                            "(with or without error) so we stop DAQ",
                            *this));
            Stop();
        }
    }
}
 
bool DaqControllerImpl::Collecting::HasTransfer(
    SourceResolver::SourceFile const& source) const noexcept {
    auto it = std::find_if(transfers.cbegin(), transfers.cend(), [&](Transfer const& t) {
        return t.source == source;
    });
    return it != transfers.cend();
}
 
DaqControllerImpl::Collecting::Transfer*
DaqControllerImpl::Collecting::GetTransfer(SourceResolver::SourceFile const& source) noexcept {
    auto it = std::find_if(
        transfers.begin(), transfers.end(), [&](Transfer const& t) { return t.source == source; });
 
    if (it != transfers.end()) {
        return &(*it);
    }
    return nullptr;
}
 
void DaqControllerImpl::Collecting::EraseTransfer(SourceResolver::SourceFile const& source) {
    auto it = std::find_if(transfers.cbegin(), transfers.cend(), [&](Transfer const& t) {
        return t.source == source;
    });
    if (it != transfers.end()) {
        transfers.erase(it);
    }
}
 
DaqControllerImpl::Collecting::~Collecting() noexcept {
    for (auto& transfer : transfers) {
        if (transfer.proc && transfer.proc->IsRunning()) {
            LOG4CPLUS_INFO(
                "dpm",
                "DaqController::Collecting::~Collecting: Terminating running transfer process");
            transfer.proc->Signal(SIGTERM);
        }
    }
}
 
DaqControllerImpl::Merging::~Merging() noexcept {
    if (merger && merger->IsRunning()) {
        LOG4CPLUS_DEBUG("dpm", "DaqController::Merging::~Merging: Aborting running merge process");
        merger->Signal(SIGTERM);
    }
}
 
DaqControllerImpl::Releasing::~Releasing() noexcept {
    for (auto& p : transfers) {
        if (p.second.proc && p.second.proc->IsRunning()) {
            LOG4CPLUS_INFO(
                "dpm",
                "DaqController::Releasing::~Releasing: Terminating running transfer process");
            p.second.proc->Signal(SIGTERM);
        }
    }
}
 
void DaqControllerImpl::Poll(DaqControllerImpl::Merging& ctx) {
    LOG4CPLUS_TRACE(m_logger, "Poll(Merging)");
    auto defer = ObservableStatus::DeferSignal(&m_status);
    try {
        if (ctx.merger) {
            // Merge is already in progress
            return;
        }
        // [recovery]
        if (m_workspace->Exists(m_result)) {
            m_status.ClearAlert(MakeAlertId(alert::MERGING_MERGE, ""));
            SetState(Completed{});
            LOG4CPLUS_TRACE(m_logger,
                            fmt::format("{}: Poll(Merging): "
                                        "Recovered from error automatically (manual merge)",
                                        *this));
            return;
        }
        auto token = m_resources.merge.Acquire();
        if (!token) {
            LOG4CPLUS_TRACE(m_logger,
                            fmt::format("{}: Poll(Merging): Could not start merging due to "
                                        "resource limit reached: {}",
                                        *this,
                                        m_resources.merge.GetLimit()));
 
            return;
        }
 
        // Launch merge
        // note: daqDpmMerge will atomically move result into output so we don't have
        // to do that ourselves.
        std::vector<std::string> args{m_options.merge_bin};
        args.emplace_back("--json");
        args.emplace_back("--root");
        args.emplace_back(m_workspace->GetPath());
        args.emplace_back("--logfile");
        args.emplace_back(m_workspace->GetLogsPath() / "merge.log");
        args.emplace_back("--resolver");
        args.emplace_back(m_workspace->GetSourceLookupPath());
        args.emplace_back("-o");
        args.emplace_back(m_result.native());
        // Positional argument for the specification which is passed as absolute path.
        args.emplace_back(m_workspace->GetPath() / m_workspace->GetSpecificationPath());
 
        PrintArgs(m_logger, args);
 
        // Clear all merger alerts before HandleMergeMessage() sets them again.
        m_status.ClearAlert(MakeAlertId(alert::MERGING_MERGE, "*"));
 
        ctx.merger = m_proc_factory(m_executor.get_io_context(), args);
 
        ctx.merger->ConnectStdout([logger = log4cplus::Logger::getInstance(LOGGER_NAME_MERGER),
                                   this](pid_t pid, std::string const& line) {
            LOG4CPLUS_DEBUG(logger, pid << ": " << Trim(line));
            HandleMergeMessage(line);
        });
        ctx.merger->ConnectStderr([logger = log4cplus::Logger::getInstance(LOGGER_NAME_MERGER)](
                                      pid_t pid, std::string const& line) {
            LOG4CPLUS_INFO(logger, pid << ": " << Trim(line));
        });
 
        ctx.merger->Initiate().then(
            m_executor,
            [liveness = std::weak_ptr<bool>(m_liveness),
             id = m_status.GetId(),
             proc = ctx.merger,
             this](boost::future<int> f) {
                if (liveness.expired()) {
                    LOG4CPLUS_ERROR(LOGGER_NAME_SCHEDULER,
                                    fmt::format("{}: DaqController abandoned -> ignoring "
                                                "result from merger",
                                                id));
                    return;
                }
                MergeComplete(std::move(f));
            });
        ctx.token = std::move(*token);
    } catch (...) {
        Stop();
        auto msg = fmt::format("{}: Failed to initiate merging", *this);
        LOG4CPLUS_ERROR(m_logger, msg);
        std::throw_with_nested(std::runtime_error(msg));
    }
}
 
void DaqControllerImpl::MergeComplete(boost::future<int> result) noexcept {
    LOG4CPLUS_TRACE(m_logger, "MergeComplete()");
    try {
        auto alert = ObservableStatus::AlertActivator(&m_status, alert::MERGING_MERGE, "");
        auto* ctx = std::get_if<Merging>(&m_state_ctx);
        if (ctx) {
            ctx->Reset();
        }
        auto exit_code = result.get();
        if (exit_code != 0) {
            auto msg = fmt::format("Merging failed with code {}", exit_code);
            alert.Set(msg);
            throw std::runtime_error(fmt::format("Merging failed with code {}", exit_code));
        }
        if (!m_workspace->Exists(m_result)) {
            auto abs_path = m_workspace->GetPath() / m_result;
            auto msg =
                fmt::format("Merging reported with success but file is not found: {}", abs_path);
            alert.Set(msg);
            LOG4CPLUS_ERROR(m_logger, msg);
            throw std::runtime_error(msg);
        }
        alert.Clear();
 
        // Create symlink and set result
        m_workspace->MakeResultSymlink(m_result);
        // @todo: Include host?
        m_status.SetResult(m_result.native());
 
        // All good -> transition to State::Completed
        // @todo: Transition to Releasing rather than completed
        LOG4CPLUS_INFO(m_logger, fmt::format("{}: Merge completed successfully!", *this));
        SetState(Releasing{});
    } catch (...) {
        Stop();
        LOG4CPLUS_ERROR(
            m_logger,
            fmt::format(
                "{}: PollMergeComplete: Failed to create data product: {}", *this, m_result));
    }
}
 
bool DaqControllerImpl::TryStartRelease(Releasing& ctx,
                                        json::ReceiverTypes const& receiver,
                                        std::size_t index) {
    // Start release transfer according to configuration
 
    if (auto* cfg = std::get_if<json::OlasReceiver>(&receiver); cfg) {
        return TryStartRelease(ctx, *cfg, index);
    } else {
        LOG4CPLUS_ERROR(
            m_logger,
            fmt::format("Receiver[{}]: TryStartRelease: Unknown receiver type encountered.",
                        index));
    }
 
    return true;
}
 
bool DaqControllerImpl::TryStartRelease(Releasing& ctx,
                                        json::OlasReceiver const& receiver,
                                        std::size_t index) {
    LOG4CPLUS_TRACE(
        m_logger,
        fmt::format("Receiver[{}]: TryStartRelease: Will try to start transfer to receiver",
                    index));
    namespace fs = std::filesystem;
    if (receiver.host.empty()) {
        // Local host case
        //
        // Try first to create a hardlink, then symlink, which if it works we don't need to copy
        // the file (in the case of symlink OLAS is responsible for copying).
        //
        // Failure to crate hardlink may include:
        // - Filesystem doesn't support it.
        // - Different filesystem between link and target.
        // - Link is in a directory that does not exist (rsync might create it but we don't)
 
        // note: We cannot simply use result filename as this may contain custom prefixes.
        auto link = receiver.path / GetArchiveFilename();
        if (std::error_code ec = m_workspace->MakeHardLink(m_result, link); !ec) {
            // Mark as completed
            m_status.SetReceiverStatus(index, ReceiverStatus{ReceiverStatus::State::Success});
            DeferredPoll();
            LOG4CPLUS_INFO(
                m_logger,
                fmt::format("Receiver[{}]: Hardlink created from {} -> {}", index, m_result, link));
            return true;
        } else if (receiver.options.allow_symlink && !m_workspace->MakeSymlink(m_result, link)) {
            // Mark as completed
            m_status.SetReceiverStatus(index, ReceiverStatus{ReceiverStatus::State::Success});
            DeferredPoll();
            LOG4CPLUS_INFO(
                m_logger,
                fmt::format("Receiver[{}]: Symlink created from {} -> {}", index, m_result, link));
            return true;
        } else {
            // Failure to create hardlink is not an error.
            LOG4CPLUS_TRACE(
                m_logger,
                fmt::format("Receiver[{}]: Hard or symlink could not be created from {} -> {}",
                            index,
                            m_result,
                            link));
        }
    }
 
    auto maybe_token = m_resources.net_send.Acquire();
    if (!maybe_token) {
        // Out of tokens -> bail out
        LOG4CPLUS_TRACE(m_logger,
                        fmt::format("Receiver[{}]: Could not start rsync transfer due to "
                                    "resource limit reached.",
                                    index));
        return false;
    }
    using Proc = std::shared_ptr<RsyncAsyncProcessIf>;
    RsyncOptions opts;
    opts.rsync = m_options.rsync_bin;
 
    auto const& path_from = m_result.native();
    auto path_to = receiver.path / GetArchiveFilename();
    auto location = receiver.host.empty() ? path_to.native()
                                          : fmt::format("{}:{}", receiver.host, path_to.native());
    opts.logfile = m_workspace->GetPath() / m_workspace->GetLogsPath() /
                   fmt::format("{}_receiver.rsync.log", index);
    Proc proc = m_rsync_factory(m_executor.get_io_context(),
                                path_from,
                                location,
                                opts,
                                RsyncAsyncProcessIf::DryRun::Disabled);
    auto last_lines =
        std::make_unique<LogCaptureLast>(log4cplus::Logger::getInstance(LOGGER_NAME_TRANSFER), 10);
 
    // Connect signals (stdout is ignored for now as it merely contains the progress
    // which will be handled differently later on).
    proc->ConnectStderr(std::ref(*last_lines));
    // Start transfer and set up completion handler
    proc->Initiate().then(
        m_executor,
        [this,
         proc,
         index,
         last_lines = std::move(last_lines),
         liveness = std::weak_ptr<bool>(m_liveness)](boost::future<int> f) {
            if (liveness.expired()) {
                LOG4CPLUS_ERROR("dpm.daqcontroller",
                                fmt::format("Receiver[{}]: DaqController abandoned -> ignoring "
                                            "result from rsync for receiver transfer "
                                            "of {}",
                                            index,
                                            *proc));
                return;
            }
            ReleaseComplete(index, *proc, std::move(f), *last_lines);
        });
    LOG4CPLUS_DEBUG(m_logger,
                    fmt::format("{}: Receiver[{}] Started release: {}", *this, index, *proc));
    // Update DAQ status
    m_status.SetReceiverStatus(index, ReceiverStatus{ReceiverStatus::State::Started});
    // Add transfer
    ctx.transfers.emplace(index, Releasing::Transfer{std::move(proc), std::move(*maybe_token)});
    return true;
}
 
void DaqControllerImpl::Poll(Releasing& ctx) {
    LOG4CPLUS_TRACE(m_logger, "Poll(Releasing)");
 
    if (m_dpspec.receivers.empty()) {
        LOG4CPLUS_INFO(m_logger,
                       "Poll(Releasing): No receivers configured -> transition to Completed");
        SetState(Completed{});
        return;
    }
 
    // Multiple changes can be made here so we defer signalling until we've done them all.
    auto defer = ObservableStatus::DeferSignal(&m_status);
    try {
        // For every non-started, non-completed receiver we initiate the process, limited by
        // available resources.
        // Once releasing is done (no more pending, transition to Completed.
 
        // Check status of each receiver as configured in dpspec
        // - Try to start non-started receiver transfer
        auto index = 0u;
        auto pending = 0;
        LOG4CPLUS_TRACE(m_logger,
                        "Poll(Releasing): Number of receivers: " << m_dpspec.receivers.size());
 
        for (auto it = m_dpspec.receivers.begin(); it != m_dpspec.receivers.end(); ++it, ++index) {
            // Persistent status of receivers contain status of previously started transfers.
            auto receiver_status = m_status.GetReceiverStatus(index);
            if (receiver_status.state == ReceiverStatus::State::NotStarted) {
                LOG4CPLUS_TRACE(
                    m_logger,
                    "Poll(Releasing): Receiver[" << index << "] Will try to start receiver.");
                // No status for receiver -> Try to start it
                TryStartRelease(ctx, *it, index);
                pending++;
            } else if (!receiver_status.IsFinalState()) {
                LOG4CPLUS_TRACE(m_logger,
                                "Poll(Releasing): Receiver["
                                    << index << "] already in progress: " << receiver_status.state);
                pending++;
            } else {
                LOG4CPLUS_TRACE(m_logger,
                                "Poll(Releasing): Receiver[" << index << "] already completed with "
                                                             << receiver_status.state);
            }
        }
 
        if (pending == 0 && ctx.transfers.empty()) {
            // Since we have no pending (to be started) and no active transfers we're done!
            LOG4CPLUS_INFO(
                m_logger,
                "Poll(Releasing): No release transfers remains -> transition to Completed");
            SetState(Completed{});
        }
    } catch (...) {
        // Alerts should already be set.
        Stop();
 
        auto msg = fmt::format("{}: Failed to release ", m_status);
        LOG4CPLUS_ERROR(m_logger, fmt::format("Poll(Releasing): Failed to process DAQ: {}", msg));
        std::throw_with_nested(std::runtime_error(msg));
    }
}
 
void DaqControllerImpl::ReleaseComplete(std::size_t index,
                                        AsyncProcessIf const& proc,
                                        boost::future<int> result,
                                        LogCaptureLast const& lines) noexcept {
    auto& ctx = std::get<Releasing>(m_state_ctx);
    // Remove transfer
    ctx.transfers.erase(index);
    auto alert = ObservableStatus::AlertActivator(
        &m_status, alert::RELEASING_RSYNC, fmt::format("{}", index));
    try {
        auto code = result.get();
 
        LOG4CPLUS_DEBUG(
            m_logger,
            fmt::format(
                "{}: Receiver[{}]: Process completed with rc={}: {}", *this, index, code, proc));
 
        if (code == 0) {
            alert.Clear();
            m_status.SetReceiverStatus(index, ReceiverStatus{ReceiverStatus::State::Success});
        } else {
            m_status.SetReceiverStatus(index, ReceiverStatus{ReceiverStatus::State::Failure});
            auto msg = fmt::format(
                "Receiver[{}]: Failed to release file using command: {}. Last lines:\n{}",
                index,
                proc,
                lines);
            alert.Set(msg);
            LOG4CPLUS_ERROR(m_logger, *this << ' ' << msg);
        }
    } catch (std::exception const& e) {
        auto msg = fmt::format(
            "Receiver[{}]: Failed to release file using command: {}. Last "
            "lines:\n{}\nException:\n{}",
            index,
            proc,
            lines,
            e.what());
        LOG4CPLUS_ERROR(m_logger, *this << ' ' << msg);
        alert.Set(msg);
        Stop();
    }
 
    DeferredPoll();
}
 
void DaqControllerImpl::Poll(Completed&) {
    LOG4CPLUS_TRACE(m_logger, "Poll(Completed)");
    // We're done
    Stop();
}
 
void DaqControllerImpl::SetState(StateVariant s) {
    m_state_ctx = std::move(s);
    auto new_state = MakeState(m_state_ctx);
    if (new_state == m_status.GetState()) {
        // No change
        return;
    }
    LOG4CPLUS_DEBUG(m_logger, "New state: " << new_state << ", Alerts: " << m_status.GetAlerts());
 
    m_status.SetState(new_state);
    DeferredPoll();
}
 
State DaqControllerImpl::MakeState(StateVariant const& s) {
    if (std::holds_alternative<Scheduled>(s)) {
        return State::Scheduled;
    }
    if (std::holds_alternative<Collecting>(s)) {
        return State::Collecting;
    }
    if (std::holds_alternative<Merging>(s)) {
        return State::Merging;
    }
    if (std::holds_alternative<Releasing>(s)) {
        return State::Releasing;
    }
    if (std::holds_alternative<Completed>(s)) {
        return State::Completed;
    }
 
    assert(false);  // NOLINT  Not implemented yet
}
 
DaqControllerImpl::Collecting::Collecting(SourceResolver resolver_arg)
    : resolver(std::move(resolver_arg)) {
}
 
void DaqControllerImpl::Merging::Reset() {
    merger.reset();
    token.reset();
}
 
void DaqControllerImpl::HandleMergeMessage(std::string const& line) noexcept try {
    // A valid JSON message on stdout has the basic form:
    // {"type": "<content-type>",
    //  "timestamp": <int64 nanonseconds since epoch>,
    //  "content": <content>}
    auto json = nlohmann::json::parse(line);
    auto const& type = json.at("type").get<std::string>();
    auto const& content = json.at("content");
    if (type == "alert") {
        // Content schema:
        // {"id", "ID",
        //  "message", "MESSAGE"}
        std::string id;
        std::string message;
        content.at("id").get_to(id);
        content.at("message").get_to(message);
        auto alert_id = MakeAlertId(alert::MERGING_MERGE, id);
        m_status.SetAlert(MakeAlert(alert_id, message));
    }
} catch (...) {
    LOG4CPLUS_DEBUG(m_logger, "Failed to parse JSON message from merger");
}
 
std::string DaqControllerImpl::GetArchiveFilename() const {
    if (m_status.GetFileId().empty()) {
        throw std::runtime_error(fmt::format("{}: DAQ status has no file_id!", m_status));
    }
    return fmt::format("{}.fits", m_status.GetFileId());
}
 
}  // namespace daq::dpm