awaitPrim.cpp

Go to the documentation of this file.

/**
 * @file
 * @ingroup daq_op
 * @copyright 2021 ESO - European Southern Observatory
 *
 * @brief Contains definition for the AwaitPrimAsync operation
 */
#include <daq/op/awaitPrim.hpp>
 
#include <fmt/format.h>
#include <fmt/ostream.h>
#include <log4cplus/loggingmacros.h>
 
#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;
    size_t num_ok = 0;
    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_aborted(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 condition is fulfilled, set reply value or exception.
            if (m_aborted) {
                LOG4CPLUS_DEBUG(m_params.common.logger,
                                fmt::format("AwaitPrimAsync::InitiateAwait: Operation aborted"));
                return;
            }
            try {
                auto is_ok = fut.get();
                if (is_ok) {
                    // All done, set promise
                    this->m_promise.set_value({m_error, std::move(m_parts)});
                } else {
                    // If condition is not fulfilled we do it again.
                    this->InitiateAwait();
                }
            } catch (...) {
                // Fatal error. Report to operation invoker.
                this->m_promise.set_exception(boost::current_exception());
            }
        });
}
 
boost::future<void> AwaitPrimAsync::MakeInterval() {
    using std::chrono::duration_cast;
    using std::chrono::steady_clock;
    using std::chrono::milliseconds;
 
    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 {
    LOG4CPLUS_INFO(m_params.common.logger,
                   fmt::format("AwaitPrimAsync::Abort: Requested to abort! "
                               "Number of files received so far: {}",
                               m_parts.size()));
    if (m_aborted) {
        return;
    }
    m_aborted = true;
    m_interval.reset();
    m_promise.set_value({m_error, std::move(m_parts)});
}
 
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_aborted) {
        LOG4CPLUS_DEBUG(m_params.common.logger,
                        fmt::format("AwaitPrimAsync::AwaitOnceAsync: Operation aborted"));
        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 !IsFinalState(source.GetState());
               },
               m_params.common,
               // Sender
               [this](Source<PrimSource>& s) -> boost::future<Reply> {
                   auto spec = std::make_shared<RecWaitSpec>(
                       duration_cast<Seconds>(this->m_params.wait_interval).count());
                   return s.GetSource().GetRrClient().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) {
    try {
        auto reply = fut.get();
        auto status = reply->getStatus();
        if (status == recif::Success) {
            // Await returned
            source.SetState(State::Stopped);
 
            auto rec_status = reply->getRecStatus();
            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_error = true;
        throw boost::enable_current_exception(
            DaqSourceError("RecWait", std::string(source.GetSource().GetName()), e.what()));
    } catch (...) {
        m_error = true;
        throw boost::enable_current_exception(DaqSourceError(
            "RecWait", std::string(source.GetSource().GetName()), "unknown exception"));
    }
}
}  // namespace daq::op