scheduler.hpp

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 and related class declarations.
 */
#ifndef DAQ_DPM_SCHEDULER_HPP
#define DAQ_DPM_SCHEDULER_HPP
#include <daq/config.hpp>
 
#include <functional>
#include <vector>
 
#include <boost/signals2.hpp>
#include <fmt/ostream.h>
#include <log4cplus/logger.h>
#include <rad/ioExecutor.hpp>
 
#include <daq/dpm/sourceResolver.hpp>
#include <daq/json/dpSpec.hpp>
#include <daq/process/asyncProcess.hpp>
#include <daq/process/rsyncAsyncProcess.hpp>
#include <daq/resourceToken.hpp>
#include <daq/status.hpp>
 
namespace daq::dpm {
 
class Workspace;
class DaqWorkspace;
struct Resources;
 
/**
 * Controller for specific DAQ.
 *
 * Responsibilities:
 *
 * State::Scheduled for each state:
 *  1. Establish list of missing sources and create and store source mapping.
 *  2. Transition to Tranferring when source mapping has been stored successfully.
 *
 * State::Collecting
 *  1. Use source mapping from previous state and query workspace if source has been downloaded.
 *  2. Transfer missing files.
 *  3. Transition to Merging when all files have been successfully transferred.
 *
 * State::Merging
 *  1. ...
 *  N. Transition to Releasing
 *
 * State::Releasing
 *  1. ...
 *
 * State::Complete
 *  - Nothing
 *
 * DaqController cannot assume that all activities for each state is complete.
 */
class DaqController {
public:
    virtual ~DaqController() {
    }
    /**
     * @returns data acquisition identifier.
     */
    virtual auto GetId() const noexcept -> std::string const& = 0;
 
    /**
     * @return Error flag.
     */
    virtual auto GetErrorFlag() const noexcept -> bool = 0;
 
    /**
     * @returns state of data acquisition.
     */
    virtual auto GetState() const noexcept -> State = 0;
 
    /**
     * @returns Data product filename.
     */
    virtual auto GetResult() const noexcept -> std::filesystem::path const& = 0;
 
    /**
     * @returns status object associated with this daq.
     */
    virtual auto GetStatus() noexcept -> ObservableStatus& = 0;
    virtual auto GetStatus() const noexcept -> ObservableStatus const& = 0;
 
    /**
     * Start/stop operations.
     *
     * This assumes the implementation has an internal scheduler that it uses.
     */
    virtual void Start() = 0;
    virtual void Stop() = 0;
    virtual void Poll() = 0;
    virtual bool IsStopped() const noexcept = 0;
};
 
std::ostream& operator<<(std::ostream& os, DaqController const& daq);
 
/**
 * Imposes limits on how many concurrent operations are allowed.
 */
struct ConcurrencyLimits {
    /**
     * Limits how many DAQs overall can be scheduled concurrently.
     */
    unsigned short daq = 1;
 
    /**
     * Maximum number of concurrent output transfers.
     */
    unsigned short net_send = 0;
 
    /**
     * Maximum number of concurrent input transfers = 0.
     */
    unsigned short net_receive = 0;
 
    /**
     * Maximum number of concurrent merge processes.
     */
    unsigned short merge = 1;
};
 
/**
 * Options controlling scheduler operations.
 *
 * Limits with value `0` is unlimited.
 */
struct SchedulerOptions {
    /**
     * Imposes limits on how many concurrent operations are allowed.
     */
    struct ConcurrencyLimits {
        /**
         * Limits how many DAQs overall can be scheduled concurrently.
         */
        unsigned short daq = 1;
 
        /**
         * Maximum number of concurrent output transfers.
         */
        unsigned short net_send = 0;
 
        /**
         * Maximum number of concurrent input transfers = 0.
         */
        unsigned short net_receive = 0;
 
        /**
         * Maximum number of concurrent transfers
         */
        unsigned short merge = 0;
    } concurrency_limits;
};
 
/**
 * Options for DaqController
 */
struct DaqControllerOptions {
    std::string merge_bin = "daqDpmMerge";
    std::string rsync_bin = "rsync";
};
 
/**
 * Limited resources
 */
struct Resources {
    Resource daqs;
    Resource net_send;
    Resource net_receive;
    Resource merge;
};
 
struct ResourcesConnections {
    boost::signals2::scoped_connection daqs;
    boost::signals2::scoped_connection net_send;
    boost::signals2::scoped_connection net_receive;
    boost::signals2::scoped_connection merge;
};
 
/**
 * Schedules asynchronous activities that results in merged Data Product and delivery.
 *
 * Internally it maintains a prioritized queue of DAQs that should be merged.
 *
 * Main responsibilities:
 *
 * - Maintain DAQ queue in priority order.
 * - Process each queued DAQ in order (possibly with configurable parallelization):
 *   - Ensure* input source files are transferred.
 *   - Ensure* DAQ is eventually merged to create DP.
 * - Update DAQ state to reflect the state it is in.
 * - State persistence
 *   - Store state to workspace and
 *   - Load state to recover at startup.
 *
 * Outside the scope of Scheduler:
 *
 * - Prerequisite for DAQ is that the Data Product Specification is final and can be considered
 *   "read-only" within the Scheduler. This means that e.g. the input file names have been
 *   determined.
 *
 * TBD/notes:
 * - Workspace management (should be same component responsible for):
 *   - Who finalizes DP Specification (local filenames)?
 *     Local file names are missing and needs to be determined (ensuring that files do not collide).
 *   - Who initializes workspace with new DAQs? It should be the entity that decides about file
 *     names (how about daq::dpm::Workspace?).
 *   - Who archives DAQs?
 * - Only DAQs that are actively being worked on needs to be represented in Scheduler. The remaining
 *   backlog could just be entries in a file `<workspace>/queue.json`.
 *   - This would imply that Scheduler is at least responsible for:
 *     - Maintaining `queue.json`
 *     - Interface use strings rather than fat objects.
 *     - It creates fat objects
 */
class Scheduler {
public:
    virtual ~Scheduler() {
    }
 
    /**
     * Start/stop operations.
     *
     * This assumes the implementation has an internal scheduler that it uses.
     */
    virtual void Start() = 0;
    virtual void Stop() = 0;
 
    /**
     * Queues DAQ for processing.
     *
     * @note Scheduler is responsible for creating unique local names for input files.
     *
     * @param dp_spec JSON encoded Data Product Specification. If parsing fails a
     * std::invalid_argument will be thrown.
     * @param status JSON encoded daq::Status. May be an empty string in which case a default
     * initial status will be used.
     * @returns DAQ id on success.
     * @throw std::invalid_argument if dp_spec is invalid or is already queued.
     */
    virtual std::string QueueDaq(std::string const& dp_spec, std::string const& status) = 0;
 
    /**
     * Abort merging DAQ identified by @c id.
     *
     * @note Workspace related to this will be purged.
     *
     * @param id DAQ id.
     *
     * @throw std::invalid_argument if DAQ is unknown.
     */
    virtual void AbortDaq(std::string const& id) = 0;
 
    /**
     * Queries if DAQ with ID has been queued before in the current workspace.
     *
     * @param id DAQ id.
     * @return true if DAQ is in the merge queue, false otherwise.
     * @throw std::invalid_argument if DAQ is not known.
     */
    virtual bool IsQueued(std::string const& id) const noexcept = 0;
 
    /**
     * Queries current DAQ status, possibly from last recorded status in workspace.
     *
     * @param id DAQ id.
     * @return Merge status of @a id.
     * @throw std::invalid_argument if DAQ is not known.
     */
    virtual Status GetDaqStatus(std::string const& id) const = 0;
 
    /**
     * Queries current DAQ queue.
     *
     * @return list of DAQs pending or already started to be merged.
     */
    virtual std::vector<std::string> GetQueue() const noexcept = 0;
 
    /**
     * Signals
     */
    /// @{
    using StatusSignal = boost::signals2::signal<void(Status const&)>;
    virtual boost::signals2::connection ConnectStatus(StatusSignal::slot_type const& slot) = 0;
 
    /// @}
};
 
class SchedulerImpl : public Scheduler {
public:
    using DaqControllerFactory =
        std::function<std::unique_ptr<DaqController>(std::unique_ptr<DaqWorkspace>, Resources&)>;
 
    /**
     * Constructs a scheduler loading information from workspace @a ws.
     *
     * The scheduler will load the stored list of queued DAQs and schedule operations to be
     * performed in priority order.
     */
    SchedulerImpl(rad::IoExecutor& executor,
                  Workspace& workspace,
                  DaqControllerFactory daq_controller_factory,
                  SchedulerOptions const& options);
 
    std::string QueueDaq(std::string const& dp_spec, std::string const& status) override;
    void AbortDaq(std::string const&) override;
    bool IsQueued(std::string const& id) const noexcept override;
    Status GetDaqStatus(std::string const& id) const override;
    std::vector<std::string> GetQueue() const noexcept override;
 
    /**
     * @name Signals.
     *
     * - DAQ Status
     */
    /// @{
    boost::signals2::connection ConnectStatus(StatusSignal::slot_type const& slot) override;
    /// @}
    void Start() override;
    void Stop() override;
 
private:
    /**
     * Polls for possible activitites to start.
     *
     * This is the core of the scheduler which will initiate asynchronus activities.
     *
     * This should be invoked as external events happen:
     * - Previously started activities complete (which may allow new activities to start).
     * - Modifiers invoked (QueueDaq/AbortDaq)
     * - Limits changed.
     *
     * Activities are only started if limits allow them, in DAQ priority order.
     *
     * List of activities:
     *
     * - Initiate active DAQ.
     * - State::Collecting
     *   - Transfer all files (identify missing sources and start transfer).
     * - State::Merge:
     *   - Merge to create Data Product
     * - State::Releasing:
     *   - Release finished DP to receivers.
     *
     * Updates m_active with new DAQs if limits and queue allows.
     * If active daq is Scheduled it will first be transitioned to Tranferring.
     */
    void Poll();
 
    /**
     * Checks queue for DAQs to start merging, limited on available resources.
     */
    void ActivateFromQueue();
 
    /**
     * Checks m_active for completed daqs and if completed:
     * - Erase from m_active
     * - Erase from queue
     * - Archive workspace
     */
    void ArchiveCompleted();
 
    /**
     * Schedules Poll() to be executed at a later time.
     *
     * @thread_safe
     */
    void DeferredPoll();
 
    /**
     * Get candidates for merging (set of DAQs from backlog/queue not in m_active.
     */
    std::vector<std::string> GetCandidates() const;
 
    struct Active {
        Active(std::unique_ptr<DaqController> daq_arg, ResourceToken token_arg)
            : daq(std::move(daq_arg)), token(std::move(token_arg)) {
            assert(daq);
        }
        std::unique_ptr<DaqController> daq;
        ResourceToken token;
    };
 
    rad::IoExecutor& m_executor;
    Workspace& m_workspace;
    DaqControllerFactory m_daq_controller_factory;
    SchedulerOptions m_options;
 
    // Must be ordered before m_active
    Resources m_resources;
    ResourcesConnections m_resources_connections;
 
    /**
     * Active (transferring files, merging or delivering result) DAQs in priority order.
     *
     * Operations for active DAQs are concurrent. There's no synchronization between DAQs.
     *
     * Number of active DAQs is limited by m_resources.daq.
     */
    std::vector<Active> m_active;
 
    /**
     * Container of all DAQs that are currently being processed or is simply in the backlog/queue.
     * This must be kept synchronized with file system so that it can be relied on for answering
     * questions like "have this DAQ been queued before".
     */
    std::vector<std::string> m_queue;
 
    StatusSignal m_status_signal;
    log4cplus::Logger m_logger;
    /**
     * A shared pointer value which if valid indicates that Scheduler is alive.
     * The value itself indicates whether a deferred Poll() has been scheduled but not yet executed.
     */
    std::shared_ptr<bool> m_liveness;
 
    /**
     * Indicates if Scheduler is stopped -> shouldn't poll.
     */
    bool m_stopped = true;
};
 
/**
 * Internal data structure to SchedulerImpl
 */
class DaqControllerImpl : public DaqController {
public:
    using RsyncFactory =
        std::function<std::unique_ptr<RsyncAsyncProcessIf>(boost::asio::io_context&,
                                                           std::string const&,  // source
                                                           std::string const&,  // dest
                                                           RsyncOptions const&,
                                                           RsyncAsyncProcess::DryRun)>;
 
    using ProcFactory = std::function<std::unique_ptr<AsyncProcessIf>(
        boost::asio::io_context&, std::vector<std::string> const&)>;
    /**
     * Construct controller for existing workspace.
     * @post State is loaded from @a workspace.
     *
     * @throws std-exception (possibly nested) containing error.
     */
    DaqControllerImpl(rad::IoExecutor& executor,
                      std::unique_ptr<DaqWorkspace> workspace,
                      Resources& resources,
                      RsyncFactory rsync_factory,
                      ProcFactory proc_factory,
                      DaqControllerOptions opts);
 
    void Start() override;
    void Stop() override;
    auto IsStopped() const noexcept -> bool override;
 
    /**
     * @returns data acquisition identifier.
     */
    auto GetId() const noexcept -> std::string const& override;
 
    /**
     * @return Error flag.
     */
    auto GetErrorFlag() const noexcept -> bool override;
 
    /**
     * @returns state of data acquisition.
     */
    auto GetState() const noexcept -> State override;
 
    auto GetResult() const noexcept -> std::filesystem::path const& override {
        return m_result;
    }
 
    /**
     * @returns status object associated with this daq.
     */
    auto GetStatus() noexcept -> ObservableStatus& override;
    auto GetStatus() const noexcept -> ObservableStatus const& override;
    void Poll() override;
 
private:
    using SourceFile = SourceResolver::SourceFile;
    void DeferredPoll();
    struct Scheduled {};
    struct Collecting {
        struct Transfer {
            SourceFile source;
            std::filesystem::path local_path;
            std::shared_ptr<RsyncAsyncProcessIf> proc;
            ResourceToken token;
        };
        Collecting(SourceResolver resolver);
        Collecting(Collecting&&) = default;
        Collecting& operator=(Collecting&&) = default;
        /**
         * Aborts any in-progress transfers.
         */
        ~Collecting() noexcept;
        bool HasTransfer(SourceFile const&) const noexcept;
        Transfer* GetTransfer(SourceFile const&) noexcept;
        void EraseTransfer(SourceFile const&);
 
        SourceResolver resolver;
        std::vector<Transfer> transfers;
    };
    struct Merging {
        Merging() = default;
        Merging(Merging&&) = default;
        Merging& operator=(Merging&&) = default;
        ~Merging() noexcept;
        void Reset();
        std::shared_ptr<AsyncProcessIf> merger = {};
        std::optional<ResourceToken> token = std::nullopt;
    };
    /**
     * In state releasing we transfer the data product result (m_result) to configured receivers.
     *
     * Failures are not treated as fatal.
     */
    struct Releasing {
        Releasing(Releasing&&) = default;
        Releasing& operator=(Releasing&&) = default;
        /**
         * Aborts any in-progress transfers.
         */
        ~Releasing() noexcept;
        /**
         * Rsync transfer of Data Product to a receiver
         */
        struct Transfer {
            /**
             * Transfer
             */
            std::shared_ptr<RsyncAsyncProcessIf> proc;
            /**
             * Token held to implement resource limits.
             */
            ResourceToken token;
        };
 
        /**
         * Active rsync receiver transfers.
         *
         * @note that this state is not stored persistently but rather daq::Status should contain
         * whether a transfer was completed successfully so it can be skipped if DPM is restarted.
         */
        std::map<std::size_t, Transfer> transfers;
    };
    struct Completed {};
 
    // Poll methods are allowed to perform state transitions.
    void Poll(Scheduled&);
    void Poll(Collecting&);
    void TransferComplete(SourceFile const& source,
                          std::filesystem::path const& local_path,
                          boost::future<int> result,
                          LogCaptureLast const& log) noexcept;
 
    void Poll(Merging&);
    void MergeComplete(boost::future<int> result) noexcept;
 
    void Poll(Releasing&);
    /**
     * Try to start a new release transfer, constrained by resources and soft-stop condition.
     * @return true if started.
     * @return false if resource limits were hit.
     */
    bool TryStartRelease(Releasing& ctx, json::ReceiverTypes const& receiver, std::size_t index);
    bool TryStartRelease(Releasing& ctx, json::OlasReceiver const& receiver, std::size_t index);
 
    /**
     * Completion handler for a Release Transfer.
     *
     * On Success: Update status with success.
     * On Failure: Update status with failure and increment try count.
     *
     * Finally make a deferred poll to schedule new transfers or transition to Complete.
     */
    void ReleaseComplete(std::size_t index,
                         AsyncProcessIf const& proc,
                         boost::future<int> result,
                         LogCaptureLast const& log) noexcept;
 
    void Poll(Completed&);
 
    /**
     * Handle JSON message from merger
     */
    void HandleMergeMessage(std::string const& line) noexcept;
 
    using StateVariant = std::variant<Scheduled, Collecting, Merging, Releasing, Completed>;
    static State MakeState(StateVariant const&);
    /**
     * Updates m_state and m_status and is the normal way m_state/m_status is updated (except
     * during construction).
     */
    void SetState(StateVariant s);
 
    /**
     * Get archive filename (the filename stem only).
     */
    std::string GetArchiveFilename() const;
 
    /**
     * State persistence
     */
    rad::IoExecutor& m_executor;
    std::unique_ptr<DaqWorkspace> m_workspace;
    Resources& m_resources;
    RsyncFactory m_rsync_factory;
    ProcFactory m_proc_factory;
    DaqControllerOptions m_options;
 
    json::DpSpec const m_dpspec;
    /**
     * Path to resulting data product (this may contain custom prefixes and is not compatible with
     * OLAS)
     */
    std::filesystem::path m_result;
 
    StateVariant m_state_ctx;
 
    /**
     * Current in-memory status.
     */
    ObservableStatus m_status;
    boost::signals2::scoped_connection m_status_connection;
 
    /**
     * A shared pointer value which if valid indicates that Scheduler is alive.
     * The value itself indicates whether a deferred Poll() has been scheduled but not yet
     * executed.
     */
    std::shared_ptr<bool> m_liveness;
    bool m_stopped;
    /**
     * If true no new async activities should be started. This is mainly used to avoid stopping
     * abruptly when an error occurs and let already started activities complete.
     */
    bool m_soft_stop = false;
    log4cplus::Logger m_logger;
};
 
}  // namespace daq::dpm
 
template <typename T, typename Char>
struct fmt::
    formatter<T, Char, std::enable_if_t<std::is_convertible_v<T*, daq::dpm::DaqController*>>>
    : ostream_formatter {};
 
#endif  // #ifndef DAQ_DPM_SCHEDULER_HPP