daqController.hpp

Go to the documentation of this file.

/**
 * @file
 * @ingroup daq_common_libdaq
 * @copyright 2022 ESO - European Southern Observatory
 *
 * @brief Contains declaration for for DaqController
 */
#ifndef OCF_DAQ_DAQ_CONTROLLER_HPP_
#define OCF_DAQ_DAQ_CONTROLLER_HPP_
#include "config.hpp"
 
#include <chrono>
#include <iostream>
#include <string>
#include <type_traits>
#include <utility>
#include <variant>
#include <vector>
 
#include <Metadaqif.hpp>
#include <boost/asio/io_context.hpp>
#include <boost/asio/steady_timer.hpp>
#include <boost/thread/future.hpp>
#include <fmt/ostream.h>
#include <log4cplus/logger.h>
#include <nlohmann/json.hpp>
#include <rad/ioExecutor.hpp>
 
#include <daq/daqContext.hpp>
#include <daq/dpPart.hpp>
#include <daq/error.hpp>
#include <daq/eventLog.hpp>
#include <daq/op/asyncOpParams.hpp>
#include <daq/op/initiate.hpp>
#include <daq/pendingReplies.hpp>
#include <daq/source.hpp>
#include <daq/state.hpp>
#include <daq/status.hpp>
#include <daq/utility.hpp>
 
namespace daq {
 
class DpmClient;
class DaqController;
 
/**
 * OCM Async operations
 *
 * @ingroup daq_common_libdaq
 */
struct OcmAsyncOperations {
    // Await returns a pair of future return value and an abort function.
    using AwaitReturnType = std::pair<boost::future<Result<DpParts>>, std::function<bool()>>;
 
    /**
     * Default constructs object with standard async operations.
     */
    OcmAsyncOperations(rad::IoExecutor& executor);
    OcmAsyncOperations(OcmAsyncOperations&&) = default;
    OcmAsyncOperations(OcmAsyncOperations const&) = default;
    OcmAsyncOperations& operator=(OcmAsyncOperations&&) = default;
    OcmAsyncOperations& operator=(OcmAsyncOperations const&) = default;
 
    bool IsValid() const noexcept;
 
    std::reference_wrapper<rad::IoExecutor> executor;
 
    std::function<boost::future<void>(op::AsyncOpParams)> start;
    std::function<boost::future<Result<void>>(ErrorPolicy, op::AsyncOpParams)> abort;
    std::function<boost::future<Result<DpParts>>(ErrorPolicy, op::AsyncOpParams)> stop;
    std::function<AwaitReturnType(op::AwaitOpParams)> await_prim;
};
 
/**
 * Abstract factory for DaqControllers.
 *
 * Main purpose is to allow daq::Manager to use a factory that creates mocks rather than real
 * implementations for testing.
 */
class DaqControllerFactory {
public:
    /**
     * Create instance for the OCM phase of the DAQ process.
     */
    virtual auto
    MakeOcmPhase(DaqContext daq_ctx,
                 std::shared_ptr<ObservableStatus> status,
                 std::shared_ptr<ObservableEventLog> event_log,
                 fits::KeywordFormatter const& kw_formatter) -> std::shared_ptr<DaqController> = 0;
    /**
     * Create instance for the DPM phase of the DAQ process.
     */
    virtual auto MakeDpmPhase(DaqContext daq_ctx,
                              std::shared_ptr<ObservableStatus> status,
                              std::shared_ptr<ObservableEventLog> event_log)
        -> std::shared_ptr<DaqController> = 0;
};
 
/**
 * Default factory producing "real" implementations.
 */
class DaqControllerFactoryDefault : public DaqControllerFactory {
public:
    DaqControllerFactoryDefault(boost::asio::io_context& io_ctx,
                                elt::mal::Mal& m_mal,
                                std::shared_ptr<DpmClient> dpm_client);
    auto MakeOcmPhase(DaqContext daq_ctx,
                      std::shared_ptr<ObservableStatus> status,
                      std::shared_ptr<ObservableEventLog> event_log,
                      fits::KeywordFormatter const& kw_formatter)
        -> std::shared_ptr<DaqController> override;
    auto MakeDpmPhase(DaqContext daq_ctx,
                      std::shared_ptr<ObservableStatus> status,
                      std::shared_ptr<ObservableEventLog> event_log)
        -> std::shared_ptr<DaqController> override;
 
private:
    boost::asio::io_context& m_io_ctx;
    rad::IoExecutor m_executor;
    elt::mal::Mal& m_mal;
    OcmAsyncOperations m_async_ops;
    std::shared_ptr<DpmClient> m_dpm_client;
};
 
/**
 * Controls the execution of single data acquisition that ultimately result in a set of FITS
 * keywords and/or FITS files.
 *
 * Important
 * ---------
 *
 * Due to the dynamic nature of data acquisitions (mostly w.r.t. fatal errors from data soures and
 * commands from user) the following assumptions are made:
 *
 * - Each data source completes only *once*.
 *   - This means that any command that deals with potential completion (mainly the stop and await
 *     operations) *must* record the outcome (DpParts + Keywords) and mark the source state as
 *     completed.
 *   - This ensures that no outcome is lost and that subsequent commands does not include completed
 *     sources.
 * - Conflicting commands are resolved in the order they are received and w.r.t. to state.
 *   - If data sources are awaited on, some of which has aleady completed, and DaqController is
 *     asked to abort, this means that the DAQ state is `Aborting` and even if the await operation
 *     completes successfully, the result will be discarded because user asked to abort.
 *   - If user asks to stop and primary soures are awaited on, there's a race between which request
 *     to data source completes first.
 *     - Commands are only sent to sources that are in-progress (not completed/aborted/stopped).
 *     - Whichever reply is received first (from await or stop) will determine the result from
 *       that data souce.
 * - Async operations must be able to deal with partial failures.
 *   - This is required to allow user to retry operation.
 * - Asynchronous errors are published and user response is requied.
 *  - If e.g. the await operation fails partially it means user must intervene to decide if it
 *    should try to stop or abort the DAQ (in the future a DAQ-wide policy could be set to decide
 *    what to do automatically).
 *
 * Usage
 * -----
 *
 * 1. Starting
 *
 *    Start with `StartAsync`
 *
 * 2. Stopping
 *
 *    Manual option
 *    -------------
 *
 *    Stop/abort with `StopAsync` or `AbortAsync`.
 *
 *    Automatic option
 *    ----------------
 *
 *    Automatically initiate stop if all primary sources are automatically stopped as
 *    indicated by `ObservableStatus` status update topic.
 *
 *    If asynchronous operations have been started, it is still possible to abort.
 *    This command will then supersede ongoing commands, and ongoing commands will
 *    be aborted at the next synchronization point (typically when reply is received).
 *
 * It is not possible to retry or go backwards by executing `StartAsync` after
 * `StopAsync` or `AbortAsync`. The reasoning behind this is to avoid the risk
 * of duplicate data acquisition id being "in flight" at the same time..
 *
 * It is possible to abort on the other hand, even though the data acquisition might not have
 * been started yet.
 *
 * For StartAsync, StopAsync, AbortAsync:
 *
 * The error handling strategy is to not Set result until we have a result
 * from each source either by reply or internal error when Sending/receiving
 * (MAL issues e.g.).
 *
 * This implies:
 *
 * - That way we can communicate the full result using the promise.
 *   On errors it contains the exception(s) on success the new state.
 * - We can have a mixed result of partial success and partial failure,
 *   although the future will contain exception(s).
 * - We let caller decide what to do (e.g. abort).
 * - We use an error flag to indicate that error occurred, which is
 *   possible in any state.
 * - We stay in the state that had the error to e.g. allow retries.
 *   So if an error occurred in `Starting` then DaqController remains
 *   in `Starting` with the error flag Set.
 *
 * Pending decision decisions:
 * - Whether to fail fast or be robust,
 *   or even whether to be configurable.
 *   - This was addressed with the ErrorPolicy option when stopping/aborting.
 * - Support aborting while there are still replies pending for startDaq?
 *   This would mean that one can be in state Starting and Stopping at the same time.
 *   - This was addressed to ignore errors when forcibly aborting.
 * - Who connects the clients + configure QoS? Since connect is done implicitly there's
 *   no need actually. Simply specify timeout should be sufficient. And that should
 *   be done by the caller.
 *
 * @ingroup daq_common_libdaq
 */
class DaqController : public std::enable_shared_from_this<DaqController> {
public:
    DaqController() = default;
    virtual ~DaqController() = default;
    /**
     * Starts the data acquisition.
     *
     * @returns A future that will be Set once data acquisition has started
     * or or fails.
     *
     * @throws std::exception-derived exception if internal error occurs.
     *
     * @return Future that is ready once all sources are started or failed.
     *
     * @pre `GetState() == State::Notstarted`
     * @post `GetState() == State::Starting` on success
     *       `GetState() == State::Error` on error
     */
    virtual boost::future<State> StartAsync() = 0;
 
    /**
     * Stops the data acquisition.
     *
     * @pre `GetState() not in (State::Stopped or State::Aborted)`
     * @post `GetState() == State::Stopping`
     */
    virtual boost::future<Status> StopAsync(ErrorPolicy policy) = 0;
 
    /**
     * Aborts the data acquisition.
     *
     * @param policy Error policy determining if errors are tolerated or not.
     *
     * It is possible to issue this request more than once, to e.g. retry a failed abort attempt.
     *
     * @pre `GetState() not in (State::Aborted, State::Stopped)`
     * @post `GetState() == State::Aborting` if a data acquisition was ongoing otherwise `GetState()
     * == State::Aborted`.
     */
    virtual boost::future<Status> AbortAsync(ErrorPolicy policy) = 0;
 
    /**
     * Schedules DAQ for merging by sending request to DPM.
     *
     * @returns future containing exception:
     * - std::logic_error if DAQ has already been scheduled
     * - mal::TimeoutException if DPM is offline (timeout).
     * - unspecified DPM fails
     */
    virtual boost::future<State> ScheduleMergeAsync() = 0;
 
    /**
     * Updates (replace or add) list of keywords.
     *
     * @param keywords Keywords to add.
     */
    virtual void UpdateKeywords(fits::KeywordVector const& keywords) = 0;
 
    /** Awaits that data acquisition stops or aborts.
     *
     * It is possible to await only only a subset of data sources by specifying their ids in
     * `sources`.
     *
     * @param sources An optional vector of source-ids to await, if  empty all primary sources are
     * awaited on.
     *
     * @returns future set with std::invalid_argument if source-id is not recognized.
     * @returns future set with boost::broken_promise if DaqController is destroyed before
     * operation completes.
     */
    virtual boost::future<State>
    AwaitAsync(std::vector<std::string> sources, std::chrono::milliseconds timeout) = 0;
 
    /**
     * @returns state of data acquisition.
     */
    virtual State GetState() const DAQ_NOEXCEPT = 0;
 
    /**
     * @returns status object associated with this daq.
     */
    virtual std::shared_ptr<ObservableStatus> GetStatus() DAQ_NOEXCEPT = 0;
    virtual std::shared_ptr<ObservableStatus const> GetStatus() const DAQ_NOEXCEPT = 0;
 
    /**
     * @returns event log associated with this daq.
     */
    virtual std::shared_ptr<ObservableEventLog> GetEventLog() DAQ_NOEXCEPT = 0;
 
    /**
     * @returns data acquisition identifier.
     */
    virtual std::string const& GetId() const DAQ_NOEXCEPT = 0;
 
    /**
     * @return Error flag.
     */
    virtual bool GetErrorFlag() const DAQ_NOEXCEPT = 0;
 
    virtual DaqContext const& GetContext() const DAQ_NOEXCEPT = 0;
 
    using ContextSignal = boost::signals2::signal<void(DaqContext const&)>;
    /**
     * Connect observer that is invoked when context is modified.
     *
     * @param o Observer callable invoked on context changes.
     *
     * @return signal connection object that can be used to disconnect observer:
     */
    virtual boost::signals2::connection ConnectContext(ContextSignal::slot_type const& slot) = 0;
};
 
std::ostream& operator<<(std::ostream& os, DaqController const& daq);
 
/**
 * Implements common behaviour of OcmDaqController and DpmDaqController.
 *
 */
class CommonDaqController : public DaqController {
public:
    CommonDaqController(boost::asio::io_context& io_context,
                        DaqContext context,
                        std::shared_ptr<ObservableStatus> status,
                        std::shared_ptr<ObservableEventLog> event_log);
 
    std::shared_ptr<ObservableStatus> GetStatus() DAQ_NOEXCEPT override;
    std::shared_ptr<ObservableStatus const> GetStatus() const DAQ_NOEXCEPT override;
    std::shared_ptr<ObservableEventLog> GetEventLog() DAQ_NOEXCEPT override;
    std::string const& GetId() const DAQ_NOEXCEPT override;
    bool GetErrorFlag() const DAQ_NOEXCEPT override;
    DaqContext const& GetContext() const DAQ_NOEXCEPT override;
    boost::signals2::connection ConnectContext(ContextSignal::slot_type const& slot) override;
 
protected:
    template <class T, class... Args>
    void AddEvent(Args&&... args) {
        m_event_log->EmplaceEvent<T>(std::forward<Args>(args)...);
    }
    boost::asio::io_context& GetIoCtx() noexcept {
        return m_io_ctx;
    }
    rad::IoExecutor& GetIoExecutor() noexcept {
        return m_executor;
    }
    DaqContext& GetContextMut() noexcept {
        return m_context;
    }
    ObservableEventLog& GetEventLogRef() noexcept {
        return *m_event_log.get();
    }
    ObservableStatus& GetStatusRef() noexcept {
        return *m_status.get();
    }
    ObservableStatus const& GetStatusRef() const noexcept {
        return *m_status.get();
    }
    void EmitContextSignal() {
        m_sig_context(m_context);
    }
 
private:
    boost::asio::io_context& m_io_ctx;
    rad::IoExecutor m_executor;
    DaqContext m_context;
    ContextSignal m_sig_context;
    std::shared_ptr<ObservableStatus> m_status;
    std::shared_ptr<ObservableEventLog> m_event_log;
};
 
/**
 * Implements `daq::DaqController` for states responsible to be executed by OCM.
 *
 * The states executed by DPM are implemented by DpmDaqController.
 *
 * @ingroup daq_common_libdaq
 */
class OcmDaqController : public CommonDaqController {
public:
    /**
     * Construct object.
     *
     * @param io_context Executor used for continuations and timer.
     * @param context General context used to control DAQ execution.
     * @param status Data acquisition status object, also contains identifier (may not be empty).
     *           Caller is responsible for making the id unique.
     * @param prim Primary data sources
     * @param meta Metadata sources
     *
     * @pre `status == true`.
     * @pre `event_log == true`.
     * @throws std::invalid_argument if arguments are invalid.
     */
    static std::shared_ptr<OcmDaqController> Create(boost::asio::io_context& io_context,
                                                    fits::KeywordFormatter const& kw_formatter,
                                                    DaqContext context,
                                                    DaqSources const& sources,
                                                    std::shared_ptr<ObservableStatus> status,
                                                    std::shared_ptr<ObservableEventLog> event_log,
                                                    OcmAsyncOperations operations);
 
    boost::future<State> StartAsync() override;
    boost::future<Status> StopAsync(ErrorPolicy policy) override;
    boost::future<Status> AbortAsync(ErrorPolicy policy) override;
    boost::future<State> ScheduleMergeAsync() override;
    void UpdateKeywords(fits::KeywordVector const& keywords) override;
    boost::future<State>
    AwaitAsync(std::vector<std::string> sources, std::chrono::milliseconds timeout) override;
 
    State GetState() const DAQ_NOEXCEPT override;
 
    /**
     * @returns Logger associated with this DaqController.
     */
    constexpr log4cplus::Logger const& GetLogger() const noexcept;
 
protected:
    struct NotStarted {};
    struct Starting {};
    struct Acquiring {};
    struct Stopping {};
    struct Stopped {};
    struct Aborting {};
    struct Aborted {};
 
    using StateVariant =
        std::variant<NotStarted, Starting, Acquiring, Stopping, Stopped, Aborting, Aborted>;
    StateVariant MakeState(State s) const noexcept;
#ifdef UNIT_TEST
public:
#endif
    // Protected constructor so user is forced to use shared pointer semantics
    // which is required because of future continuation style programming.
    OcmDaqController(boost::asio::io_context& io_context,
                     fits::KeywordFormatter const& kw_formatter,
                     DaqContext context,
                     DaqSources const& sources,
                     std::shared_ptr<ObservableStatus> status,
                     std::shared_ptr<ObservableEventLog> event_log,
                     OcmAsyncOperations ops);
#ifdef UNIT_TEST
private:
#endif
    /**
     * Constructs the parameters used for asynchronous operations.
     *
     * @note OcmAsyncOpParams will bind references to member variables so caller must
     * guarantee that DaqController outlives the async operation.
     * This is normally done by holding a shared copy in the last .then continuation
     * for the async operation, thus guaranteeing that all intermediate continuations
     * will access a valid object.
     */
    op::AsyncOpParams MakeParams(op::AlertState&);
    op::AwaitOpParams MakeAwaitParams(op::AlertState&);
 
    /**
     * Initiate async operation that triggers DAQ stop automatically.
     *
     * If there is no condition for automatic stop (e.g. meta data sources only) this method does
     * nothing.
     */
    void InitiateStopCondition();
    /**
     * Cancels the awaiting of primary sources.
     *
     * This does nothing if no await process is ongoing.
     */
    void CancelAwaitPrimarySources();
 
    void AddInitialKeywords();
 
    /**
     * Queries if ctx has stateful sources (primary or metadata source).
     *
     * @return true if ctx has 1 or more of primary or metadata sources.
     * @returns false otherwise.
     */
    auto HasStatefulSources() noexcept -> bool;
 
    /**
     * Set alert with category alert::DAQ_CONTROLLER.
     */
    void SetAlert(std::string key, std::string description) noexcept;
    void SetState(StateVariant&& s) noexcept;
 
    std::optional<std::variant<Source<PrimSource>*, Source<MetaSource>*>>
    FindSource(std::string_view source_id);
 
    /// Helper to build source vector
    template <class SourceType>
    std::vector<Source<SourceType>> MakeSources(std::vector<SourceType> sources);
 
    fits::KeywordFormatter const& m_kw_formatter;
    StateVariant m_state;
 
    std::vector<Source<PrimSource>> m_prim_sources;  ///< Note: Consider vector immutable!
    std::vector<Source<MetaSource>> m_meta_sources;  ///< Note: Consider vector immutable!
 
    OcmAsyncOperations m_async_ops;
    std::shared_ptr<PendingReplies> m_pending_replies;
    std::vector<std::unique_ptr<boost::asio::steady_timer>> m_timers;
 
    /**
     * If DaqController is awaiting the completion of primary data sources
     * this function will hold the abort function.
     *
     * @important Users must check if it is valid before invoking it.
     */
    std::function<bool()> m_abort_await_primary_sources;
    log4cplus::Logger m_logger;
};
 
/**
 * Implements behaviour from the state NotScheduled to Completed.
 *
 * Once successfully scheduled to be merged by DPM (NotScheduled -> Scheduled) the "location" for
 * the *true* DAQ state is also transferred. This means:
 *
 * - If DPM is offline:
 *   - OCM cannot know the true state of DAQ.
 *   - Aborting with strict error policy cannot be done.
 *   - Aborting with tolerant policy will forcefully abort DAQ locally but the DAQ may anyway be
 *     completed by DPM.
 *
 * DpmDaqController is responsible for requesting last known status from DPM (in state Scheduled and
 * onwards). This is done with the DPM client which aggregates all instances and manages retries.
 */
class DpmDaqController : public CommonDaqController {
public:
    /**
     * Constructor
     *
     * @important DpmDaqController must only be used with shared_ptr.
     *
     * @param io_context Used for async operations.
     * @param context DAQ context.
     * @param status DAQ status.
     * @param dpm_client Interface to DPM. During construction DpmDaqController will register
     * signal slot to receive DAQ status updates and mirror that in status accordingly.
     *
     * @throws std::invalid_argument if state of status is not NotScheduled or subsequent.
     */
    DpmDaqController(boost::asio::io_context& io_context,
                     DaqContext context,
                     std::shared_ptr<ObservableStatus> status,
                     std::shared_ptr<ObservableEventLog> event_log,
                     std::shared_ptr<DpmClient> dpm_client);
 
    boost::future<State> ScheduleMergeAsync() override;
 
    /**
     * @returns future containing std::runtime_error because operation is always invalid.
     */
    boost::future<State> StartAsync() override;
    /**
     * @returns future containing std::runtime_error because operation is always invalid.
     */
    boost::future<Status> StopAsync(ErrorPolicy policy) override;
 
    /**
     * @throws std::runtime_error because operation is always invalid (keywords cannot be modified
     * once delivered to DPM).
     */
    void UpdateKeywords(fits::KeywordVector const& keywords) override;
 
    /**
     * @returns future containing std::runtime_error because operation is always invalid.
     */
    boost::future<State>
    AwaitAsync(std::vector<std::string> sources, std::chrono::milliseconds timeout) override;
 
    /**
     * Attempts to abort Data Acquisition.
     *
     * - If Data Acquisition is not-yet-scheduled it will be aborted immediately.
     * - If Data Acquisition *may* have been scheduled (request has been sent to DPM) the request is
     *   forwarded to DPM.
     *
     * - If DPM is not running or otherwise fails the DAQ is still marked as aborted if policy is
     *   ErrorPolicy::Tolerant.
     */
    boost::future<Status> AbortAsync(ErrorPolicy policy) override;
 
    State GetState() const DAQ_NOEXCEPT override;
 
private:
    struct NotScheduled {
        /**
         * We impose limit so that at most a single request to schedule merge can be outstanding.
         * If a reply is pending this is indicated by the optional containing a future.
         *
         * This is valid in state NotScheduled.
         */
        bool schedule_reply_pending;
    };
 
    /**
     * Changes m_state_ctx as necessary based on current state.
     */
    void UpdateStateContext();
    void SetState(State state);
 
    using StateVariant = std::variant<std::monostate, NotScheduled>;
 
    /**
     * An extra liveness object.
     */
    std::shared_ptr<bool> m_liveness;
    std::shared_ptr<DpmClient> m_dpm_client;
    /**
     * Connection to disconnect slot from DpmClient::StatusSignal
     */
    boost::signals2::scoped_connection m_status_connection;
 
    // "Cached" representation of the data product specification created from local DaqContect
    // It is created on the fly in ScheduleMergeAsync as necessary.
    std::optional<std::string> m_dp_spec;
    StateVariant m_state_ctx;
    log4cplus::Logger m_logger;
};
 
}  // namespace daq
 
template <>
struct fmt::formatter<daq::DaqController> : ostream_formatter {};
 
template <typename T, typename Char>
struct fmt::formatter<T, Char, std::enable_if_t<std::is_convertible_v<T*, daq::DaqController*>>>
    : ostream_formatter {};
 
#endif  // OCF_DAQ_DAQ_CONTROLLER_HPP_