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 <fmt/format.h>
#include <fmt/ostream.h>
#include <log4cplus/loggingmacros.h>
 
#include <daq/dpm/workspace.hpp>
#include <daq/error/report.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());
}
}  // namespace
 
ResourceToken::ResourceToken(Resource* resource) noexcept : m_resource(resource) {
}
 
ResourceToken::~ResourceToken() noexcept {
    m_resource->Release();
}
 
void Resource::Release() noexcept {
    m_used--;
    // If limit is 0 (infinite resouces) then there's no point in signalling as nothing will ever be
    // blocked by missing resource.
    if (m_limit > 0 && m_used < m_limit) {
        m_signal();
    }
}
 
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("dpm.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();
        }
    };
    m_resources.daqs.Connect(slot);
    m_resources.net_receive.Connect(slot);
    m_resources.merge.Connect(slot);
    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) {
    LOG4CPLUS_TRACE(m_logger, "QueueDaq()");
    try {
        auto json = nlohmann::json::parse(dp_spec_serialized);
        DpSpec dp_spec = ParseDpSpec(json);
        std::string const& id = dp_spec.id;
        std::string const& file_id = dp_spec.target.file_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);
                Status initial_status;
                initial_status.id = id;
                initial_status.file_id = file_id;
                // Initial state in DPM is Scheduled as that's where DPM takes over.
                initial_status.state = State::Scheduled;
                initial_status.error = false;
                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 (DpSpecError const&) {
        LOG4CPLUS_DEBUG(m_logger,
                        "Invalid Data Product Specification provided: \n"
                            << dp_spec_serialized);
 
        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&) {
        std::throw_with_nested(
            std::invalid_argument("Scheduler: Could not queue DAQ for merging because provided "
                                  "data product specification is invalid JSON"));
    }
}
 
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)) {
        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 (IsFinalState(state) || state == State::Releasing) {
            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("dpm.daqcontroller")) {
    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");
    }
 
    // 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.
    auto error = m_status.GetError();
    switch (m_status.GetState()) {
    case State::Scheduled:
        SetState(Scheduled{}, error);
        break;
    case State::Transferring:
        SetState(Transferring{SourceResolver(m_workspace->LoadSourceLookup())}, error);
        break;
    case State::Merging:
        SetState(Merging(), error);
        break;
    case State::Completed:
        SetState(Completed{}, error);
        break;
    default:
        throw std::runtime_error("Not implemented");
    };
 
    m_status_connection = m_status.ConnectStatus(
        [prev = m_status.GetStatus(), this](ObservableStatus const& status) mutable {
            if (prev == status.GetStatus()) {
                // No changes -> don't write
                return;
            }
            prev = status.GetStatus();
            LOG4CPLUS_TRACE(m_logger,
                            fmt::format("DaqController: Updating workspace status file: {}", prev));
            m_workspace->StoreStatus(prev);
        });
}
 
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.GetError();
}
 
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<DpSpec::SourceFitsFile> sources;
        if (m_dpspec.target.source) {
            sources.push_back(*m_dpspec.target.source);
        }
        for (auto const& s : m_dpspec.sources) {
            if (!std::holds_alternative<DpSpec::SourceFitsFile>(s)) {
                continue;
            }
            sources.push_back(std::get<DpSpec::SourceFitsFile>(s));
        }
 
        auto sources_path = m_workspace->GetSourcesPath();
 
        SourceResolver resolver;
        unsigned index = 0;
        for (auto const& s : sources) {
            Origin origin = ParseSourceOrigin(s.origin);
            // Use <index>_<source>_<filename> as local filename.
            auto local_path =
                sources_path /
                fmt::format("{}_{}_{}", index, s.source_name, origin.path.filename().native());
            LOG4CPLUS_INFO(m_logger,
                           fmt::format("Poll(Scheduled): Source file '{}' from source \"{}\" "
                                       "on host \"{}\" will be stored in {}",
                                       origin.path,
                                       s.source_name,
                                       !origin.host.empty() ? origin.host.c_str() : "<n/a>",
                                       local_path));
 
            resolver.Add({s.source_name, s.origin}, local_path);
        }
 
        // Store source resolution mapping.
        m_workspace->StoreSourceLookup(resolver.GetMapping());
 
        SetState(Transferring(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(DaqControllerImpl::Transferring& ctx) {
    LOG4CPLUS_TRACE(m_logger, "Poll(Transferring)");
    try {
        // Check if any source is missing and start transfers if possible
        // If all transfers are done transition to State::Merging.
 
        auto const& sources = ctx.resolver.GetMapping();
        auto missing = sources.size();
        auto root = m_workspace->GetPath();
        for (auto const& source : sources) {
            if (ctx.HasTransfer(source.first)) {
                // Transfer already in progress
                continue;
            }
            // [recovery]
            if (m_workspace->Exists(source.second)) {
                missing--;
                // File already exist
                continue;
            }
            if (m_soft_stop) {
                // If there are errors we don't initiate new transfers
                LOG4CPLUS_TRACE(m_logger,
                                "Poll(Transferring): 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
                Origin origin = ParseSourceOrigin(source.first.origin);
                using Proc = std::shared_ptr<RsyncAsyncProcessIf>;
                RsyncOptions opts;
                opts.rsync = m_options.rsync_bin;
                Proc proc = m_rsync_factory(m_executor.get_io_context(),
                                            origin.RsyncPath(),
                                            root / source.second,
                                            opts,
                                            RsyncAsyncProcessIf::DryRun::Disabled);
 
                // Connect signals (stdout is ignored for now as it merely contains the progress
                // which will be handled differently later on).
                proc->ConnectStderr([logger = log4cplus::Logger::getInstance("dpm.transfer")](
                                        pid_t pid, std::string const& line) {
                    LOG4CPLUS_INFO(logger, pid << ": " << Trim(line));
                });
                // Start transfer and set up completion handler
                proc->Initiate().then(
                    m_executor,
                    [liveness = std::weak_ptr<bool>(m_liveness),
                     source = source.first,
                     dest = source.second,
                     proc = proc,
                     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));
                    });
                ctx.transfers.emplace_back(Transferring::Transfer{
                    source.first, source.second, std::move(proc), std::move(*token)});
            } else {
                // Out of tokens -> bail out
                LOG4CPLUS_TRACE(m_logger,
                                fmt::format("Poll(Transferring): 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(), false);
            return;
        }
    } catch (...) {
        SetError(true);
        // 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(Transferring): 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) noexcept {
    LOG4CPLUS_TRACE(m_logger,
                    fmt::format("{}: TransferComplete: {} -> {}", *this, source, local_path));
    auto* ctx = std::get_if<Transferring>(&m_state_ctx);
    // Multiple changes can be made here so we defer signalling until we've done them all.
    auto defer = ObservableStatus::DeferSignal(&m_status);
    try {
        if (ctx) {
            ctx->EraseTransfer(source);
        }
 
        int return_code = result.get();
        auto alert_id = MakeAlertId(alert::TRANSFERRING_RSYNC, local_path);
        if (return_code != 0) {
            m_status.SetAlert(MakeAlert(
                alert_id, fmt::format("rsync file transfer failed for remote file {}", source)));
            LOG4CPLUS_ERROR(m_logger, fmt::format("rync file transfer failed: {}", source));
            throw std::runtime_error("rsync failed to transfer file");
        }
        // Opportunistically clear alert in case it as set from previous attempt
        m_status.ClearAlert(alert_id);
        LOG4CPLUS_INFO(m_logger,
                       fmt::format("{}: Poll(Transferring): Successfully transfer file: {} -> {}",
                                   *this,
                                   source,
                                   local_path));
 
        // Transfer ok
        DeferredPoll();
        return;
    } catch (...) {
        SetError(true);
        LOG4CPLUS_ERROR(m_logger,
                        fmt::format("{}: Poll(Transferring): 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(Transferring): All pending transfers have completed "
                            "(with or without error) so we stop DAQ",
                            *this));
            Stop();
        }
    }
}
 
bool DaqControllerImpl::Transferring::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::Transferring::Transfer*
DaqControllerImpl::Transferring::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::Transferring::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::Transferring::~Transferring() noexcept {
    for (auto& transfer : transfers) {
        if (transfer.proc && transfer.proc->IsRunning()) {
            LOG4CPLUS_DEBUG(
                "dpm",
                "DaqController::Transferring::~Transferring: Aborting running transfer process");
            transfer.proc->Abort();
        }
    }
}
 
DaqControllerImpl::Merging::~Merging() noexcept {
    if (merger && merger->IsRunning()) {
        LOG4CPLUS_DEBUG("dpm", "DaqController::Merging::~Merging: Aborting running merge process");
        merger->Abort();
    }
}
 
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)) {
            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("--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);
 
        ctx.merger = m_proc_factory(m_executor.get_io_context(), std::move(args));
 
        ctx.merger->ConnectStdout([logger = log4cplus::Logger::getInstance("dpm.merger"), this](
                                      pid_t pid, std::string const& line) {
            LOG4CPLUS_DEBUG(logger, pid << ": " << Trim(line));
            HandleMergeMessage(line);
        });
        ctx.merger->ConnectStderr([logger = log4cplus::Logger::getInstance("dpm.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("dpm.daqcontroller",
                                    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()");
    // Defer signal until all changes have been made
    auto defer = ObservableStatus::DeferSignal(&m_status);
    try {
        auto alert_id = MakeAlertId(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);
            m_status.SetAlert(MakeAlert(alert_id, 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);
            m_status.SetAlert(MakeAlert(alert_id, msg));
            LOG4CPLUS_ERROR(m_logger, msg);
            throw std::runtime_error(msg);
        }
        m_status.ClearAlert(alert_id);
 
        // 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
        SetState(Completed{});
        LOG4CPLUS_INFO(m_logger, fmt::format("{}: Completed successfully!", *this));
    } catch (...) {
        SetError(true);
        Stop();
        LOG4CPLUS_ERROR(
            m_logger,
            fmt::format(
                "{}: PollMergeComplete: Failed to create data product: {}", *this, m_result));
    }
}
 
void DaqControllerImpl::Poll(Completed&) {
    LOG4CPLUS_TRACE(m_logger, "Poll(Completed)");
    // We're done
    Stop();
}
 
void DaqControllerImpl::SetState(StateVariant s, bool error) {
    m_state_ctx = std::move(s);
    auto new_state = MakeState(m_state_ctx);
    if (new_state == m_status.GetState() && error == m_status.GetError()) {
        // No change
        return;
    }
 
    m_status.SetState(new_state, error);
    DeferredPoll();
}
 
void DaqControllerImpl::SetError(bool error) {
    if (error == m_status.GetError()) {
        // No change
        return;
    }
 
    m_status.SetError(error);
    m_soft_stop = true;
    DeferredPoll();
}
 
State DaqControllerImpl::MakeState(StateVariant const& s) {
    if (std::holds_alternative<Scheduled>(s)) {
        return State::Scheduled;
    }
    if (std::holds_alternative<Transferring>(s)) {
        return State::Transferring;
    }
    if (std::holds_alternative<Merging>(s)) {
        return State::Merging;
    }
    if (std::holds_alternative<Completed>(s)) {
        return State::Completed;
    }
 
    assert(false);  // NOLINT  Not implemented yet
}
 
DaqControllerImpl::Transferring::Transferring(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");
}
 
}  // namespace daq::dpm