awaitPrim.hpp

Go to the documentation of this file.

/**
 * @file
 * @ingroup daq_common_libdaq
 * @copyright 2022 ESO - European Southern Observatory
 *
 * @brief Contains declaration for the AwaitPrimAsync operation
 */
#ifndef OCM_DAQ_OP_AWAIT_HPP_
#define OCM_DAQ_OP_AWAIT_HPP_
#include "../config.hpp"
 
#include <vector>
 
#include <boost/thread/future.hpp>
#include <boost/asio/steady_timer.hpp>
 
#include "../dpPart.hpp"
#include "../error.hpp"
#include "asyncOpParams.hpp"
 
namespace daq::op {
 
/**
 * A composite async operation that awaits primary data sources.
 *
 * This is used by OCM to await primary data sources completion.
 * Since sources may acquire data for a long time the operation is implemented by
 * sending await commands to client with a smaller timeout, and then resending request
 * on timeout. This is done to be able to have a reasonable MAL timeout to detect network related
 * issues.
 *
 * Notes
 * -----
 *
 * The await operation returns files produced from awaited-on sources (using DpParts result).
 *
 * Await requests are ony sent to source if observed state requires it (i.e. if source is not
 * already observed to be stopped, from a previous AwaitPrimAsync operation).
 *
 * Await operation is completed when any of the following conditions are fulfilled:
 *
 * - All sources reply that recording is complete.
 * - Any source reply with fatal error.
 * - Operation is aborted.
 *
 * The operation does not allow configurable error policy but behaves as if ErrorPolicy::Robust
 * is set. The following are condidered fatal errors (causing result of operation to be an
 * exception):
 * - Internal errors.
 *
 * @todo Do not assume commands are honoring the timeout parameter. Instead keep track of the send
 * interval. Otherwise a malfunctioning subsystem that immediately returns exception will be spammed
 * with requests.
 * @ingroup daq_common_libdaq
 */
struct AwaitPrimAsync {
public:
    /**
     * Constructs operation with the privided parameters.
     *
     * @param params parameters.
     */
    explicit AwaitPrimAsync(AwaitOpParams params) noexcept;
 
    /**
     * Initiates operation that await acquisition completion.
     *
     * @note Caller is responsible for keeping object alive until
     * result is set.
     */
    [[nodiscard]] boost::future<Result<DpParts>> Initiate();
 
    /**
     * Aborts the operation.
     */
    void Abort() noexcept;
 
private:
    /**
     * Initiates the await operation that will eventually set the m_promise value or exception
     * and returns.
     */
    void InitiateAwait();
    /**
     * Send request to all incomplete primary sources and wait for reply.
     *
     * @returns future set when all sources reply.
     * - True means all awaited-on sources have completed.
     * - False if one or more sources are not completed.
     * - Contains exception on fatal error.
     *
     * Source state are updated using status from reply, including produced files.
     */
    boost::future<bool> AwaitOnceAsync();
    /**
     * Callback for each reply that
     * - updates source state,
     * - adds data products
     *
     * @returns true if RecWait completed successfully (data acquisition is complete)
     * @returns false if RecWait timed out (data acquisition is still ongoing)
     * @throws DaqSourceError if RecWait contains exception.
     */
    bool HandleRecWaitReply(Source<PrimSource>& source,
                            boost::future<std::shared_ptr<recif::RecWaitStatus>>&& fut);
 
    /**
     * Returns future with value set when requests should be sent.
     */
    boost::future<void> MakeInterval();
    AwaitOpParams m_params;
 
    struct IntervalTimer {
        IntervalTimer(boost::asio::io_context& io_ctx, std::chrono::steady_clock::time_point next)
            : timer(io_ctx, next), promise() {
        }
        ~IntervalTimer() {
            timer.cancel();
        }
        boost::asio::steady_timer timer;
        boost::promise<void> promise;
    };
    /**
     * Indicates if it completed with error.
     */
    bool m_error;
 
    /**
     * Holds result from data sources.
     */
    DpParts m_parts;
 
    /**
     * Indicates whether operation was aborted or not.
     *
     * If this is true it means that m_promise has already been fulfilled.
     */
    bool m_aborted;
 
    /**
     * Promise for future returned from `Initiate()`
     */
    boost::promise<Result<DpParts>> m_promise;
    /**
     * Time of last InitiateAwait
     */
    std::optional<IntervalTimer> m_interval;
    std::optional<std::chrono::steady_clock::time_point> m_last_start;
};
 
}  // namespace daq::op
#endif  // #ifndef OCM_DAQ_OP_AWAIT_HPP_