status.hpp

Go to the documentation of this file.

/**
 * @file
 * @ingroup daq_common_libdaq
 * @copyright 2022 ESO - European Southern Observatory
 *
 * @brief Contains declaration for Status and ObservableStatus
 */
#ifndef OCM_DAQ_DAQ_STATUS_HPP_
#define OCM_DAQ_DAQ_STATUS_HPP_
 
#include <chrono>
#include <iosfwd>
#include <map>
#include <string>
#include <vector>
 
#include <boost/signals2/signal.hpp>
 
#include "dpPart.hpp"
#include "state.hpp"
#include <daq/fits/keyword.hpp>
 
namespace daq {
 
namespace alert {
 
/**
 * Request
 *
 * Key: request-name + source-name
 */
constexpr std::string_view REQUEST = "request";
 
/**
 * Failure during rsync source copy.
 * Key: local file path
 */
constexpr std::string_view COLLECTING_RSYNC = "collecting.rsync";
 
/**
 * Failure during rsync source copy.
 * Key: receiver index in DP Spec
 */
constexpr std::string_view RELEASING_RSYNC = "releasing.rsync";
 
/**
 * Merging failed.
 * Key: "" (nothing)
 *
 * Warning/alert during merging.
 * Key: warning-id (e.g. "primary_hdu_resize")
 */
constexpr std::string_view MERGING_MERGE = "merging.merge";
}  // namespace alert
 
/**
 * Uniquely identfies an alert.
 */
struct AlertId {
    /**
     * Standardized category
     */
    std::string category;
    /**
     * Unique key for each alert
     */
    std::string key;
};
 
bool operator==(AlertId const& lhs, AlertId const& rhs) noexcept;
bool operator!=(AlertId const& lhs, AlertId const& rhs) noexcept;
std::ostream& operator<<(std::ostream& os, AlertId const& s);
 
struct PutTimeStruct {
    std::chrono::system_clock::time_point time;
};
std::ostream& operator<<(std::ostream& os, PutTimeStruct const& t);
 
PutTimeStruct PutTime(std::chrono::system_clock::time_point const& time);
 
/**
 * Describes an active Data Acquisition alert.
 *
 * category and key make up the unique identifier of an alert.
 */
struct Alert {
    using Clock = std::chrono::system_clock;
    using TimePoint = Clock::time_point;
 
    AlertId id;
    std::string description;
    TimePoint timestamp;
};
 
/**
 * Construct alert
 */
Alert MakeAlert(std::string_view category, std::string key, std::string description);
Alert MakeAlert(AlertId id, std::string description);
AlertId MakeAlertId(std::string_view category, std::string key);
 
/**
 * Comparison operator for Alert.
 * @returns true if category and key both compares equal between lhs and rhs (description .
 */
bool operator==(Alert const& lhs, Alert const& rhs) noexcept;
bool operator==(Alert const& lhs, AlertId const& rhs) noexcept;
bool operator==(AlertId const& lhs, Alert const& rhs) noexcept;
bool operator!=(Alert const& lhs, Alert const& rhs) noexcept;
 
std::ostream& operator<<(std::ostream& os, Alert const& s);
std::ostream& operator<<(std::ostream& os, std::vector<Alert> const& s);
 
/**
 * Set alert.
 * @param alerts Container of alert to modify.
 * @param alert to set.
 * @relatesalso Status
 */
void SetAlert(std::vector<Alert>& alerts, Alert alert);
 
/**
 * Clear alert.
 * @param alerts Container of alert to modify.
 * @param alert to set.
 * @returns true if there was an alert to clear, false otherwise.
 * @relatesalso Status
 */
bool ClearAlert(std::vector<Alert>& alerts, AlertId const& alert);
 
/**
 * Persistent status for receiver delivery.
 */
struct ReceiverStatus {
    enum class State { NotStarted = 0, Started, Success, Failure };
 
    bool IsFinalState() const noexcept;
    State state;
};
 
bool operator==(ReceiverStatus const& lhs, ReceiverStatus const& rhs) noexcept;
bool operator!=(ReceiverStatus const& lhs, ReceiverStatus const& rhs) noexcept;
 
std::ostream& operator<<(std::ostream& os, ReceiverStatus::State state);
 
/**
 * Non observable status object that keeps stores status of data acquisition.
 *
 * It is also planned to be serializable to allow crash recovery from persistent storage.
 *
 * @ingroup daq_common_libdaq
 */
struct Status {
    using TimePoint = std::chrono::time_point<std::chrono::system_clock>;
    using Clock = TimePoint::clock;
 
    Status() = default;
    /**
     * @param id DAQ id
     * @param file_id File identifier (OLAS compatible).
     */
    explicit Status(std::string id, std::string file_id) noexcept;
    Status(
        std::string id, std::string file_id, State state, bool error, TimePoint timestamp) noexcept;
 
    Status(Status&&) = default;
    Status(Status const&) = default;
    Status& operator=(Status&&) = default;
    Status& operator=(Status const&) = default;
 
    bool operator==(Status const& rhs) const noexcept;
    bool operator!=(Status const& rhs) const noexcept;
 
    std::string id;
    std::string file_id;
    State state = State::NotStarted;
    bool error = false;
    /**
     * Active alerts
     */
    std::vector<Alert> alerts;
 
    /**
     * Receiver processing (e.g. olas or rsync) status.
     *
     * Key represents corresponding DpSpec::receivers index.
     */
    std::map<std::size_t, ReceiverStatus> receivers;
 
    /**
     * Path to resulting data product.
     */
    std::string result;
 
    /**
     * Timestamp of last update.
     */
    TimePoint timestamp;
};
 
std::ostream& operator<<(std::ostream& os, Status const& s);
 
/**
 * Stores data acquisition status and allows subscription to status changes.
 *
 * DaqController instances will update ObservableStatus as changes occur.
 *
 * @ingroup daq_common_libdaq
 */
class ObservableStatus {
public:
    /**
     * Defer signal changes until later time.
     *
     * This is mainly usedful to allow making multiple changes with only one signal being emitted.
     *
     * Effects:
     * - Blocks sending signals until Deferrer expires (destroyed or manually reset).
     * - When a deferrer is expired a signal will be emitted (if there are multiple deferrers each
     *   will emit a signal).
     */
    class DeferSignal {
    public:
        DeferSignal() = default;
        DeferSignal(ObservableStatus* status, bool force = true);
        ~DeferSignal() noexcept;
        DeferSignal(DeferSignal&&) noexcept;
        DeferSignal& operator=(DeferSignal&&) noexcept;
 
        /**
         * If object is valid this will unblock and signal changes and then remove reference to
         * ObservableStatus.
         * @post IsValid() == false
         */
        void Reset() noexcept;
 
        /**
         * @returns true if object is valid.
         */
        bool IsValid() const noexcept {
            return m_status != nullptr;
        }
 
    private:
        ObservableStatus* m_status = nullptr;
        bool m_force;
    };
 
    /**
     * Construct a new object
     *
     * @param id Data acquisition identifier.
     */
    explicit ObservableStatus(std::string id, std::string file_id) noexcept;
    explicit ObservableStatus(Status status);
    ObservableStatus(ObservableStatus&&) = default;
    ObservableStatus(ObservableStatus const&) = delete;
    ObservableStatus& operator=(ObservableStatus&&) = default;
    ObservableStatus& operator=(ObservableStatus const&) = delete;
 
    bool operator==(ObservableStatus const& rhs) const noexcept;
    bool operator!=(ObservableStatus const& rhs) const noexcept;
 
    bool operator==(Status const& rhs) const noexcept;
    bool operator!=(Status const& rhs) const noexcept;
 
    /**
     * Assign new status where the new status refers to the same DAQ.
     *
     * @param status New status to assign from.
     * @throws std::invalid_argument if status.id is not equal to this->id.
     */
    ObservableStatus& operator=(Status const& status);
 
    ///@name Accessors
    ///@{
    /**
     * @return Data acquisition identifier
     */
    std::string const& GetId() const noexcept;
 
    /**
     * @return OLAS file id
     */
    std::string const& GetFileId() const noexcept;
 
    /**
     * @return Data acquisition state
     */
    State GetState() const noexcept;
 
    /**
     * @return Data acquisition error flag.
     */
    bool GetError() const noexcept;
 
    /**
     * @return list of alerts.
     */
    std::vector<Alert> const& GetAlerts() const noexcept;
 
    /**
     * @return Timestamp.
     */
    Status::Clock::time_point GetTimestamp() const noexcept;
    ///@}
 
    ///@name Modifiers
    ///@{
    /**
     * Set state of data acquisition.
     *
     * @param s New state
     * @param error Optional new error flag value.
     *
     * @post Connected observers have been signalled.
     */
    void SetState(State s, std::optional<bool> error = std::nullopt) noexcept;
 
    /**
     * Set receiver status.
     * @param index Index of receiver in DpSpec.
     * @param status Status structure to write.
     */
    void SetReceiverStatus(std::size_t index, ReceiverStatus status);
 
    /**
     * Get receiver status.
     *
     * @param index Index of receiver in DpSpec.
     * @return Status structure.
     */
    ReceiverStatus GetReceiverStatus(std::size_t index) const noexcept;
 
    /**
     * Set error flag for data acquisition.
     *
     * @param error New error flag status.
     *
     * @post Connected observers have been signalled.
     */
    void SetError(bool error) noexcept;
 
    /**
     * Set resulting data product path.
     *
     * @param result
     */
    void SetResult(std::string result);
 
    /**
     * Set alert
     *
     * @note An existing alert with same ID will be overwritten.
     *
     * @param alert to set.
     */
    void SetAlert(Alert alert);
 
    /**
     * Clear alert
     *
     * @param alert to clear.
     */
    void ClearAlert(AlertId const& alert);
 
    /**
     * Connect observer that is invoked when state is modified.
     *
     * @param o Observer callable invoked on status changes (state or file changes)
     * Observer must be invocable with signature `void(ObservableStatus const&)`.
     *
     * @return signal connection object that can be used to disconnect observer:
     *
     * @code
     * auto c = status.ConnectObserver([](ObservableStatus const& s){});
     * // later the connection object can be used to disconnect
     * c.disconnect();
     * @endcode
     */
    using SignalType = boost::signals2::signal<void(ObservableStatus const&)>;
    template <class Observer>
    boost::signals2::connection ConnectObserver(Observer o) {
        return m_signal.connect(std::move(o));
    }
    boost::signals2::connection ConnectStatus(SignalType::slot_type const& slot) {
        return m_signal.connect(slot);
    }
    /**
     * Emit signal, which unless @a force == true, will only emit signal unless blocked.
     */
    void EmitSignal(bool forced = false) noexcept;
 
    /**
     * Query number of changes made since last signal.
     */
    unsigned ChangesSinceLastSignal() const noexcept {
        return m_changes_since_signal;
    }
 
    /**
     * Allow implicit conversion to non-observable status.
     */
    operator Status() const;
    Status const& GetStatus() const noexcept;
    ///@}
protected:
    friend class DeferSignal;
 
    /**
     * Decrement signal blocker and if number of blocks reaches zero emit signal if generations are
     * differnt, unless forced in which case signal is always emitted.
     *
     * @param force Force signal to be emitted even though
     * @returns generation counter which can be used to detect if Status has been modified.
     */
    void EnableSignals(bool force) noexcept;
 
    /**
     * Disable signals from being sent (unless forced).
     * @returns generation counter which can be used to detect if Status has been modified.
     */
    void DisableSignals() noexcept;
 
private:
    enum class UpdateTimestamp { No = 0, Yes = 1 };
    /**
     * Call to report that changes have been made, update timestamp and emit signal
     * if not blocked.
     */
    void RecordChanges(bool forced = false, UpdateTimestamp update = UpdateTimestamp::Yes) noexcept;
 
    Status m_status;
    // const Status should also be observable.
    mutable SignalType m_signal;
    /**
     * If positive there are instances of DeferSignal blocking signals
     */
    unsigned m_deferred_signals = 0u;
    unsigned m_changes_since_signal = 0u;
};
 
std::ostream& operator<<(std::ostream& os, ObservableStatus const& s);
 
}  // namespace daq
#endif  // #ifndef OCM_DAQ_DAQ_STATUS_HPP_