manager.hpp

Go to the documentation of this file.

/**
 * @file
 * @ingroup daq_common_libdaq
 * @copyright 2022 ESO - European Southern Observatory
 *
 * @brief Declaration of `daq::Manager`
 */
#ifndef OCM_DAQ_MANAGER_HPP_
#define OCM_DAQ_MANAGER_HPP_
#include "config.hpp"
 
#include <chrono>
#include <cstdint>
#include <string_view>
#include <vector>
 
#include <boost/asio/deadline_timer.hpp>
#include <boost/thread/future.hpp>
#include <log4cplus/logger.h>
#include <rad/ioExecutor.hpp>
 
#include <daq/daqContext.hpp>
#include <daq/dpmClient.hpp>
#include <daq/error.hpp>
#include <daq/eventLog.hpp>
#include <daq/fits/keyword.hpp>
#include <daq/state.hpp>
#include <daq/status.hpp>
#include <daq/utility.hpp>
 
namespace daq {
 
class Workspace;
 
/**
 * Exception indicating the DAQ id was invalid.
 */
struct InvalidDaqId : std::invalid_argument {
    explicit InvalidDaqId(std::string_view id);
    explicit InvalidDaqId(std::string_view id, std::string_view reason);
};
 
/**
 * Configurations parameters directly related to manager.
 */
struct ManagerParams {
    /**
     * Instrument identifier.
     */
    std::string instrument_id;
    std::string origin = "ESO-PARANAL";
 
    /**
     * Age of DAQ in acquiring state after which it is automatically considered abandoned and will
     * be archived without further action at startup.
     */
    std::chrono::hours acquiring_stale_age = std::chrono::hours(14);
 
    /**
     * Age of DAQ in merging state, after which it is automatically considered abandoned and will be
     * archived without further action at startup.
     */
    std::chrono::hours merging_stale_age = std::chrono::hours(2 * 24);
};
 
/**
 * Creates a DAQ id candidate that may or may not be unique.
 *
 * @param instrument_id The instrument ID to use for the id. Only the 5 first characters will be
 * used if the name is longer than 5.
 * @param jitter Jitter is a millisecond component added to the current time. This is meant to be
 * used to find a unique id by adding jitter until a unique id is found.
 *
 * @note The function cannot guarantee unique ID and is the responsibility of the caller.
 * @ingroup daq_common_libdaq
 */
std::string MakeIdCandidate(char const* instrument_id,
                            unsigned jitter = 0,
                            std::chrono::system_clock::time_point* out = nullptr);
 
class DaqController;
class DaqControllerFactory;
class FitsController;
 
/**
 * Observes any status.
 */
class StatusSignal {
public:
    using SignalType = boost::signals2::signal<void(ObservableStatus const&)>;
 
    template <class Observer>
    boost::signals2::connection ConnectObserver(Observer o) {
        return m_signal.connect(std::move(o));
    }
 
    void Signal(ObservableStatus const& status) {
        m_signal(status);
    }
 
private:
    SignalType m_signal;
};
 
/**
 * Manager owns DaqController and FitsController (active data acquisitions) instances and
 * multiplexes requests to them.
 *
 *
 * Important responsibilities:
 *
 * - Provide interface that allows multiplexing of multiple DaqControllers and FitsControllers
 *   (identified by unique id). This interface is close to what the MAL ICD interface looks like to
 *   control  DaqControllers and FitsControllers.
 * - Creation of `DaqController` instance as part of a new data acquisition (using factory)
 * - Replace `DaqController` with another implementation to perform the merging (triggered when DAQ
 *   reaches Stopped state).
 * - Monitor primary data sources to stop DaqController if a primary source completes (TODO).
 * - (TBD): On startup query previous execution for incomplete DAQ (only consider DAQs with a
 *   maximum age to discard abandoned DAQs)
 *   - Load from persistant storage.
 *   - Query DPM for in-progress merges.
 *
 * FitsController also act as a data source.
 *
 * Out of scope:
 *
 * - Creation of `FitsController` instances.
 *
 * @note: It's a template only to facilitate mocking of DaqController, without requiring it to be
 * an interface.
 *
 * @ingroup daq_common_libdaq
 */
class Manager {
public:
    virtual ~Manager() {
    }
    using Signal = boost::signals2::signal<void(ObservableStatus const&)>;
 
    /**
     * Restore from state stored in workspace.
     *
     * This should typically only be done after construction and not when running.
     */
    virtual void RestoreFromWorkspace() = 0;
 
    /**
     * Creates a new unique identifier based on the instrument id and current time.
     *
     * If there is a id collision when using current time a millisecond jitter component
     * is added until a unique id is found.
     *
     * The format is the same as ARCFILE - the file extension: "<OLAS_ID>-2020-08-19T09:33:11.951"
     *
     * @note OLAS_ID is the first 5 characters of instrument id.
     * @note The returned ID is guaranteed to be unique for all IDs known by Manager (there is
     * currently no global repository of previous IDs) and is suitable for both DAQ id and OLAS
     * FileId.
     *
     * @param [out] time Optional time used to produce the ID.
     * @throws std::system_error on failure.
     */
    virtual std::string MakeDaqId(std::chrono::system_clock::time_point* time = nullptr) const = 0;
 
    /**
     * Query existing data acquisition by @c id and optional @c file_id.
     *
     * @param id DAQ id to look up.
     * @param file_id Optional file_id to look up.
     *
     * @returns true if there is a already a DAQ with the same ID (or same file_id if argument is
     * provided).
     * @returns false otherwise.
     */
    virtual bool HaveDaq(std::string_view id, std::string_view file_id = {}) const DAQ_NOEXCEPT = 0;
 
    /**
     * Get status
     *
     * @throw std::invalid_argument if no data acquisition exist with provided `id`.
     */
    virtual Status GetStatus(std::string_view id) const = 0;
 
    /**
     * Start DaqController identified by `id`.
     *
     * @param id Data acquisition id.
     *
     * @returns Exceptional future containing std::invalid_argument, if data acquisition with `id`
     * does not exist.
     * @returns Future that will eventually be ready when data acquisition has started, or failed to
     * start.
     */
    virtual boost::future<State> StartDaqAsync(DaqContext ctx) = 0;
 
    /**
     * Stop DaqController identified by `id`.
     *
     * @param id Data acquisition id.
     * @param policy Error policy determining if errors are tolerated or not.
     *
     * @returns Exceptional future containing std::invalid_argument, if data acquisition with `id`
     * does not exist.
     * @returns Future that will eventually be ready when data acquisition has stopped, or failed to
     * stop.
     */
    virtual boost::future<Status> StopDaqAsync(std::string_view id, ErrorPolicy policy) = 0;
 
    /**
     * Abort DaqController identified by `id`.
     *
     * @param id Data acquisition id.
     * @param policy Error policy determining if errors are tolerated or not.
     *
     * @returns Exceptional future containing std::invalid_argument, if data acquisition with `id`
     * does not exist.
     * @returns Future that will eventually be ready when data acquisition has aborted, or failed to
     * abort.
     */
    virtual boost::future<Status> AbortDaqAsync(std::string_view id, ErrorPolicy policy) = 0;
 
    /**
     * Await DAQ state
     *
     * @param id Data acquisition id.
     * @param state target state to await.
     * @param timeout How long to wait for state to be reached.
     * @returns Exceptional future containing std::invalid_argument, if data acquisition with `id`
     * does not exist or timeout is negative.
     * @returns Future with a value set when condition is fulfilled or times out.
     */
    virtual boost::future<Result<Status>>
    AwaitDaqStateAsync(std::string_view id, State state, std::chrono::milliseconds timeout) = 0;
    /**
     * Update FITS keywords for DaqController identified by `id`.
     * @param id Data acquisition id.
     * @param keywords FITS keywords to update.
     *
     * @throw std::invalid_argument if no data acquisition exist with provided `id`.
     * @throw std::runtime_error If DaqController state does not allow updating of keywords (it's
     * e.g. already been submitted for merging).
     */
    virtual void UpdateKeywords(std::string_view id, fits::KeywordVector const& keywords) = 0;
 
    /**
     * @returns status observer object.
     *
     * The status observer observes any DAQ status changes.
     */
    virtual StatusSignal& GetStatusSignal() = 0;
 
    /**
     * @returns current Daq Controllers.
     */
    virtual std::vector<std::shared_ptr<DaqController const>> GetDaqControllers() = 0;
};
 
/**
 * Implements `daq::Manager`.
 *
 * @ingroup daq_common_libdaq
 */
class ManagerImpl : public Manager {
public:
    /**
     * @param instrument_id Instrument id.
     */
    explicit ManagerImpl(rad::IoExecutor& executor,
                         ManagerParams params,
                         Workspace& workspace,
                         fits::KeywordFormatter const& formatter,
                         std::shared_ptr<ObservableEventLog> event_log,
                         DaqControllerFactory& daq_factory,
                         std::shared_ptr<DpmClient> dpm_client,
                         log4cplus::Logger const& logger);
    ~ManagerImpl() noexcept;
    /**
     * Loads status and constructs DaqControllers corresponding to stored state.
     */
    void RestoreFromWorkspace() override;
 
    std::string MakeDaqId(std::chrono::system_clock::time_point* time = nullptr) const override;
 
    bool HaveDaq(std::string_view id, std::string_view file_id = {}) const noexcept override;
 
    Status GetStatus(std::string_view id) const override;
 
    boost::future<State> StartDaqAsync(DaqContext ctx) override;
 
    boost::future<Status> StopDaqAsync(std::string_view id, ErrorPolicy policy) override;
 
    boost::future<Status> AbortDaqAsync(std::string_view id, ErrorPolicy policy) override;
 
    boost::future<Result<Status>>
    AwaitDaqStateAsync(std::string_view id, State, std::chrono::milliseconds timeout) override;
 
    void UpdateKeywords(std::string_view id, fits::KeywordVector const& keywords) override;
 
    StatusSignal& GetStatusSignal() override;
 
    std::vector<std::shared_ptr<DaqController const>> GetDaqControllers() override;
 
private:
    struct Daq {
        Daq(std::string id_arg,
            std::shared_ptr<DaqController> controller_arg,
            boost::signals2::connection conn_status_arg,
            boost::signals2::connection conn_context_arg) noexcept;
 
        std::string id;
        std::shared_ptr<DaqController> controller;
        // connection for observer connected to controller.
        boost::signals2::scoped_connection conn_status;
        boost::signals2::scoped_connection conn_context;
    };
 
    /// Adds unique identifier (unrelated to DAQ id) to op abort functions
    class OpAbortFunc {
    public:
        using Func = std::function<bool()>;
        OpAbortFunc(Func&& f);
        OpAbortFunc(OpAbortFunc&&) = default;
        OpAbortFunc& operator=(OpAbortFunc&&) = default;
 
        std::uint64_t GetId() const noexcept;
        bool Abort() noexcept;
 
    private:
        static std::uint64_t NextId();
        std::uint64_t m_id;
        std::function<bool()> m_func;
    };
 
    enum class Store { Yes, No };
    /**
     * Adds initial keywords that can be provided by manager.
     *
     * - `ORIGIN`
     * - `INSTRUME`
     * - `ARCFILE`
     */
    void AddInitialKeywords(DaqContext& ctx);
    void FormatKeywordSources(DaqContext& ctx);
 
    /**
     * Adds DaqController to active set and starts monitoring status changes.
     *
     * @param daq Data acquisition to add.
     * @param store Determines if workspace should be updated or not (normally yes but if DAQ is
     * added from workspace it does not have to be updated immediately again).
     */
    void AddDaq(std::shared_ptr<DaqController> const& daq, Store store = Store::Yes);
 
    /** @name State persistence functions */
    /// @{
    /**
     * Remove daq
     * @param id DAQ to remove.
     */
    void RemoveDaq(std::string_view id);
    void ArchiveDaq(std::string const& id);
    /**
     * Store list of active daqs in workspace.
     */
    void StoreActiveDaqs() const;
    /// @}
 
    void RemoveAbortFunc(std::uint64_t id) noexcept;
    DaqController const* FindDaq(std::string_view id) const noexcept;
    DaqController* FindDaq(std::string_view id) noexcept;
    DaqController& FindDaqOrThrow(std::string_view id);
    DaqController const& FindDaqOrThrow(std::string_view id) const;
    /**
     * Invoked by Manager when a DAQ has entered Stopped state and DaqController implementation is
     * changed to the DPM phase version.
     */
    void MoveToMergePhase(std::string_view id);
 
    /**
     * Iterates data acquisitions and for those in State::NotStarted
     *
     * It sends pending requests and returns immediately. Upon failing the completion handler will
     * schedule a 60s deadline timer to retry.
     */
    void ScheduleDaqsAsync();
 
    std::shared_ptr<bool> m_alive_token;
    rad::IoExecutor& m_executor;
    ManagerParams m_params;
    Workspace& m_workspace;
    fits::KeywordFormatter const& m_kw_formatter;
    std::shared_ptr<ObservableEventLog> m_event_log;
    DaqControllerFactory& m_daq_factory;
    std::shared_ptr<DpmClient> m_dpm_client;
 
    StatusSignal m_status_signal;
    /// Data acquisitions
    std::vector<Daq> m_daq_controllers;
 
    /// Pair of DAQ-id and abort function
    std::vector<OpAbortFunc> m_abort_funcs;
    log4cplus::Logger m_logger;
 
    // Used by ScheduleDaqsAsync to retry schedule merges when DPM is offline
    std::optional<boost::asio::deadline_timer> m_schedule_retry;
};
 
}  // namespace daq
#endif  // #ifndef OCM_DAQ_MANAGER_HPP_