awaitPrim.cpp

Go to the documentation of this file.

/**
 * @file
 * @ingroup daq_op
 * @copyright 2022 ESO - European Southern Observatory
 *
 * @brief Contains definition for the AwaitPrimAsync operation
 */
#include <daq/op/awaitPrim.hpp>
 
#include <cstddef>
 
#include <fmt/format.h>
#include <fmt/ostream.h>
#include <log4cplus/loggingmacros.h>
 
#include <daq/error/report.hpp>
#include <daq/op/util.hpp>
#include <daq/state.hpp>
 
namespace {
/**
 * Process all replies together to decide if more awaits are necessary.
 *
 * @note Errors from sources are logged, but otherwise ignored.
 *
 * @returns true if all awaited-on sources have completed.
 * @returns false if one or more sources are not completed.
 */
bool AwaitUnwrapReplies(boost::future<std::vector<boost::future<bool>>>&& futures) {
    std::vector<boost::future<bool>> values = futures.get();
    if (values.empty()) {
        // No requests were sent (because they were all completed)
        return true;
    }
    std::vector<std::exception_ptr> exceptions;
    std::size_t num_ok = 0;
    std::size_t num_exceptions = 0;
    for (auto& f : values) {
        try {
            if (f.get()) {
                num_ok++;
            }
        } catch (std::exception const& e) {
            num_exceptions++;
            LOG4CPLUS_INFO(
                "daq",
                fmt::format("daq::op::AwaitUnwrapReplies: Source replied with exception: {}",
                            e.what()));
        }
    }
    LOG4CPLUS_DEBUG(
        "daq",
        fmt::format("daq::op::AwaitUnwrapReplies: {}/{} replies contained exceptions. {}/{} "
                    "replies report success",
                    num_exceptions,
                    values.size(),
                    num_ok,
                    values.size()));
 
    return num_ok == values.size();
}
 
class RecWaitSpec : public recif::RecWaitSpec {
public:
    RecWaitSpec(float timeout) : m_timeout{timeout}, m_info{} {
    }
    std::string getInfo() const override {
        return m_info;
    }
    void setInfo(const std::string& info) override {
        m_info = info;
    }
 
    float getTimeout() const override {
        return m_timeout;
    }
    void setTimeout(float timeout) override {
        m_timeout = timeout;
    }
    bool hasKey() const override {
        return false;
    }
    std::unique_ptr<recif::RecWaitSpec> cloneKey() const override {
        throw std::runtime_error("not clonable");
    }
    std::unique_ptr<recif::RecWaitSpec> clone() const override {
        throw std::runtime_error("not clonable");
    }
    bool keyEquals(const recif::RecWaitSpec& other) const override {
        return false;
    }
 
private:
    float m_timeout;
    std::string m_info;
};
}  // namespace
 
namespace daq::op {
 
AwaitPrimAsync::AwaitPrimAsync(AwaitOpParams m_params) noexcept
    : m_params(m_params), m_error(false), m_parts(), m_abort_requested(false) {
}
 
[[nodiscard]] boost::future<Result<DpParts>> AwaitPrimAsync::Initiate() {
    LOG4CPLUS_DEBUG(m_params.common.logger,
                    fmt::format("AwaitPrimAsync::Initiate: Operation initiating"));
    InitiateAwait();
    return m_promise.get_future();
}
 
void AwaitPrimAsync::InitiateAwait() {
    // If the last await was processed before the configured interval expired wait until that time
    // has elapsed.
    MakeInterval()
        .then(m_params.common.executor, [this](auto) { return AwaitOnceAsync(); })
        .unwrap()
        .then(m_params.common.executor, [this](boost::future<bool> fut) -> void {
            if (m_abort_requested) {
                LOG4CPLUS_DEBUG(
                    m_params.common.logger,
                    "AwaitPrimAsync::InitiateAwait: Operation aborted -> setting promise");
                try {
                    m_promise.set_value({m_error, std::move(m_parts)});
                } catch (boost::promise_already_satisfied const&) {
                    LOG4CPLUS_DEBUG(m_params.common.logger,
                                    "AwaitPrimAsync::InitiateAwait: Promise already satisfied");
                }
                return;
            }
            // If condition is fulfilled, set reply value or exception.
            try {
                // This
                auto is_ok = fut.get();
                if (is_ok) {
                    // All done, set promise
                    m_promise.set_value({m_error, std::move(m_parts)});
                } else {
                    // If condition is not fulfilled we do it again.
                    InitiateAwait();
                }
            } catch (boost::promise_already_satisfied const&) {
                LOG4CPLUS_DEBUG(m_params.common.logger,
                                "AwaitPrimAsync::InitiateAwait: Promise already satisfied");
            } catch (...) {
                // Fatal error. Report to operation invoker.
                m_promise.set_exception(boost::current_exception());
            }
        });
}
 
boost::future<void> AwaitPrimAsync::MakeInterval() {
    using std::chrono::duration_cast;
    using std::chrono::milliseconds;
    using std::chrono::steady_clock;
 
    if (!m_last_start) {
        // First time -> return ready future
        return boost::make_ready_future();
    }
 
    auto now = steady_clock::now();
    auto next_start = *m_last_start + duration_cast<steady_clock::duration>(m_params.wait_interval);
    if (now >= next_start) {
        // Interval already expired -> return ready future
        return boost::make_ready_future();
    }
    LOG4CPLUS_DEBUG(m_params.common.logger,
                    fmt::format("AwaitPrimAsync::MakeInterval: Waiting {}ms until sending RecWait",
                                duration_cast<milliseconds>(next_start - now).count()));
    // Wait until
    m_interval.reset();
    m_interval.emplace(m_params.common.executor.get_io_context(), next_start);
    m_interval->timer.async_wait([this](boost::system::error_code const& ec) {
        // We don't care if we're cancelled, we have to set promise anyway
        // to avoid broken promise
        m_interval->promise.set_value();
    });
    return m_interval->promise.get_future();
}
 
void AwaitPrimAsync::Abort() noexcept {
    if (m_abort_requested) {
        // Already requested
        return;
    }
    LOG4CPLUS_INFO(m_params.common.logger,
                   fmt::format("AwaitPrimAsync::Abort: Requested to abort! "
                               "Number of files received so far: {}",
                               m_parts.size()));
    m_abort_requested = true;
    if (m_interval) {
        m_interval->timer.cancel();
    }
}
 
boost::future<bool> AwaitPrimAsync::AwaitOnceAsync() {
    using Reply = std::shared_ptr<recif::RecWaitStatus>;
    using Seconds = std::chrono::duration<float>;
    using std::chrono::duration_cast;
    if (m_abort_requested) {
        LOG4CPLUS_DEBUG(
            m_params.common.logger,
            fmt::format("AwaitPrimAsync::AwaitOnceAsync: Operation requested to abort"));
        return boost::make_ready_future<bool>(true);
    }
    LOG4CPLUS_DEBUG(m_params.common.logger,
                    fmt::format("AwaitPrimAsync::AwaitOnceAsync: Sending requests..."));
 
    m_last_start = std::chrono::steady_clock::now();
 
    return SendRequestAndCollectReplies<bool>(
               m_params.common.prim_sources.begin(),
               m_params.common.prim_sources.end(),
               [](Source<PrimSource> const& source) -> bool {
                   /* only send to sources that are not already stopped */
                   return IsSubsequentState(State::Stopped, source.GetState());
               },
               m_params.common,
               // Sender
               [this](Source<PrimSource>& s) -> boost::future<Reply> {
                   auto& client = s.GetSource().GetRrClient();
                   auto seconds = duration_cast<Seconds>(this->m_params.wait_interval).count();
#if defined(UNIT_TEST)
                   // ECII-783: MAL suddenly requires that interface *must* be implemented by
                   // matching middleware otherwise it fails.
                   // To avoid complicated mocking we still use our local implementation for unit
                   // testing.
                   auto spec = std::make_shared<RecWaitSpec>(seconds);
#else
                   auto mal = client.getMal();
                   BOOST_ASSERT_MSG(mal, "MAL RR client returned invalid MAL pointer");
                   auto spec = mal->createDataEntity<recif::RecWaitSpec>();
                   // Initialize all members by assigning from local implementation
                   *spec = RecWaitSpec(seconds);
#endif
                   return client.RecWait(spec);
               },
               // reply handler
               [this](AsyncOpParams params, Source<PrimSource>& source, boost::future<Reply>&& fut)
                   -> bool { return HandleRecWaitReply(source, std::move(fut)); },
               std::string_view("AwaitPrimAsync: await primary data acquisition"))
        .then(m_params.common.executor, AwaitUnwrapReplies);
}
 
bool AwaitPrimAsync::HandleRecWaitReply(
    Source<PrimSource>& source, boost::future<std::shared_ptr<recif::RecWaitStatus>>&& fut) {
    // Note: Alerts are set and cleared immedately rather than deferring to
    // a later point as the await operation can take a very long time.
    auto alert_id =
        MakeAlertId(alert::REQUEST, std::string("RecWait") + source.GetSource().GetName());
    try {
        auto reply = fut.get();
        // Request did not throw -> clear any alert
        m_params.common.status.ClearAlert(alert_id);
 
        auto status = reply->getStatus();
        if (status == recif::Success) {
            // Await returned
            source.SetState(State::Stopped);
 
            auto rec_status = reply->getRecStat();
            LOG4CPLUS_INFO(m_params.common.logger,
                           fmt::format("Data source '{}' replied successfully and provides {} "
                                       "number of files",
                                       source.GetSource(),
                                       rec_status->getDpFiles().size()));
            if (rec_status->getDpFiles().empty()) {
                LOG4CPLUS_WARN(m_params.common.logger,
                               fmt::format("Data source '{}' replied successfully for "
                                           "RecWait but did not produce any files!",
                                           source.GetSource()));
            }
            for (auto const& file : rec_status->getDpFiles()) {
                m_parts.emplace_back(std::string(source.GetSource().GetName()), file);
            }
            // All ok, return true
            return true;
        } else {
            // Wait timed out, return false to resend
            return false;
        }
    } catch (recif::ExceptionErr const& e) {
        m_params.common.status.SetAlert(
            MakeAlert(alert_id,
                      fmt::format("Primary source '{}' request 'RecWaitStatus' replied "
                                  "with ICD error: ({}) {}",
                                  source.GetSource().GetName(),
                                  e.getCode(),
                                  e.getDesc())));
        m_error = true;
        throw boost::enable_current_exception(
            DaqSourceError("RecWait", std::string(source.GetSource().GetName()), e.what()));
    } catch (...) {
        auto what = error::FormatException(std::current_exception());
        m_params.common.status.SetAlert(
            MakeAlert(alert_id,
                      fmt::format("Primary source '{}' request 'RecWaitStatus' replied "
                                  "with non-ICD error: {}",
                                  source.GetSource().GetName(),
                                  what)));
        m_error = true;
        throw boost::enable_current_exception(DaqSourceError(
            "RecWait", std::string(source.GetSource().GetName()), "unknown exception"));
    }
}
}  // namespace daq::op