daqController.hpp

Go to the documentation of this file.

/**
 * @file
 * @ingroup daq_ocm_libdaq
 * @copyright 2021 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 <utility>
#include <variant>
#include <vector>
 
#include <boost/thread/future.hpp>
#include <boost/asio/io_context.hpp>
#include <boost/asio/steady_timer.hpp>
#include <log4cplus/logger.h>
#include <gsl/pointers>
#include <rad/ioExecutor.hpp>
#include <Metadaqif.hpp>
 
#include "fitsController.hpp"
#include "error.hpp"
#include "source.hpp"
#include "state.hpp"
#include "status.hpp"
#include "eventLog.hpp"
#include "pendingReplies.hpp"
#include "daqProperties.hpp"
#include "dpPart.hpp"
#include "op/initiate.hpp"
#include "op/asyncOpParams.hpp"
 
namespace daq {
 
/**
 * Async operations
 *
 * @ingroup daq_ocm_libdaq
 */
struct AsyncOperations {
    // 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.
     */
    AsyncOperations();
    AsyncOperations(AsyncOperations&&) = default;
    AsyncOperations(AsyncOperations const&) = default;
    AsyncOperations& operator=(AsyncOperations&&) = default;
    AsyncOperations& operator=(AsyncOperations const&) = default;
 
    bool IsValid() const noexcept;
 
    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;
};
 
/**
 * 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_ocm_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;
 
    /**
     * 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;
};
 
/**
 * Implements `daq::DaqController`
 *
 *
 * @ingroup daq_ocm_libdaq
 */
class DaqControllerImpl : public DaqController {
  public:
    /**
     * Construct object.
     *
     * @param io_context Executor used for continuations and timer.
     * @param properties General properties 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<DaqControllerImpl> Create(boost::asio::io_context& io_context,
                                                     DaqProperties properties,
                                                     std::shared_ptr<ObservableStatus> status,
                                                     std::shared_ptr<ObservableEventLog> event_log,
                                                     AsyncOperations operations);
 
    boost::future<State> StartAsync() override;
    boost::future<Status> StopAsync(ErrorPolicy policy) override;
    boost::future<Status> AbortAsync(ErrorPolicy policy) 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 ;
    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;
 
    /**
     * @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.
    DaqControllerImpl(boost::asio::io_context& io_context,
                  DaqProperties properties,
                  std::unique_ptr<FitsController> fits_controller,
                  std::shared_ptr<ObservableStatus> status,
                  std::shared_ptr<ObservableEventLog> event_log,
                  AsyncOperations ops);
#ifdef UNIT_TEST
private:
#endif
    template<class T, class... Args>
    void AddEvent(Args&&... args) {
        m_event_log->EmplaceEvent<T>(std::forward<Args>(args)...);
    }
    /**
     * Constructs the parameters used for asynchronous operations.
     *
     * @note AsyncOpParams 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::AwaitOpParams MakeAwaitParams();
 
    /**
     * Await completion of primary sources.
     *
     * If there are no sources this method does nothing.
     */
    void InitiateAwaitPrimarySources();
 
    void SetErrorFlag(bool error) noexcept;
    void SetState(StateVariant&& s) noexcept;
 
    std::optional<std::variant<gsl::not_null<Source<PrimSource>*>,
                               gsl::not_null<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);
 
    StateVariant m_state;
    boost::asio::io_context& m_io_ctx;
    rad::IoExecutor m_executor;
    DaqProperties m_properties;
 
    std::unique_ptr<FitsController> m_fits_ctl;
    std::shared_ptr<ObservableStatus> m_status;
    std::shared_ptr<ObservableEventLog> m_event_log;
 
    std::vector<Source<PrimSource>> m_prim_sources;  ///< Note: Consider vector immutable!
    std::vector<Source<MetaSource>> m_meta_sources;  ///< Note: Consider vector immutable!
 
    AsyncOperations 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;
};
 
std::ostream& operator<<(std::ostream& os, DaqController const& daq);
 
}  // namespace daq
 
#endif // OCF_DAQ_DAQ_CONTROLLER_HPP_