util.hpp

Go to the documentation of this file.

/**
 * @file
 * @ingroup daq_common_libdaq
 * @copyright 2022 ESO - European Southern Observatory
 *
 * @brief Contains declaration for the async op utilities
 */
#ifndef OCF_DAQ_OP_UTIL_HPP_
#define OCF_DAQ_OP_UTIL_HPP_
#include "../config.hpp"
 
#include <algorithm>
#include <functional>
#include <optional>
#include <vector>
 
#include <Metadaqif.hpp>
#include <boost/thread/future.hpp>
#include <fmt/format.h>
#include <fmt/ostream.h>
#include <log4cplus/logger.h>
#include <log4cplus/loggingmacros.h>
 
#include "../error.hpp"
#include "../state.hpp"
#include "asyncOpParams.hpp"
 
namespace daq::op {
namespace eventlog {
template <class SourceType>
void Reply(AsyncOpParams& params, SourceType const& source, std::string description, bool error) {
    params.event_log.AddEvent(GenericEvent(
        params.id,
        fmt::format(!error ? "{}: Got reply from '{}'" : "{}: Got error or timeout from '{}'",
                    description,
                    source),
        params.status.GetStatus()));
}
template <class SourceType>
void Request(AsyncOpParams& params, SourceType const& source, std::string description) {
    params.event_log.AddEvent(
        GenericEvent(params.id,
                     fmt::format("{}: Sending request to'{}'", description, source),
                     params.status.GetStatus()));
}
 
template <class F>
size_t CountExceptions(std::vector<F> const& futures) noexcept {
    return std::count_if(futures.begin(), futures.end(), [](F const& future) noexcept -> bool {
        return future.has_exception();
    });
}
}  // namespace eventlog
 
/**
 * Utility function to Send requests and collect replies.
 *
 * @todo: Need to generalize based on requestor reply type.
 *
 * @param begin Beginning of sequence of sources to send request.
 * @param end End of sequence of sources to send request.
 * @param filter_pred Unary predicate in the form `(Source const&) -> bool` returning true for those
 * source a request should be sent.
 * @param params Parameters for this async operation.
 * @param sender Functor that sends request and returns future.
 * @param reply_handler Handler invoked once for each reply received. The handler may transform the
 * reply and return a value containing a different type than the input. It should not return a
 * future.
 * @param replies_handler Handler invoked once when all replies are received. It will be passed the
 * vector of replies returned from reply_handler. To communicate failure it shall throw an
 * exception.
 *
 * @return Future of R, as returned by replies handler
 *
 * Type requirement:
 *
 *  with `T` being the requestor reply type.
 *
 * `Sender`: boost::future<T> (Source<S>&)
 * `ReplyHandler`: R (AsyncOpParams, Source<S>, boost::future<T>)
 *
 * @ingroup daq_common_libdaq
 */
template <class R, class Iterator, class Pred, class Sender, class ReplyHandler>
boost::future<std::vector<boost::future<R>>>
SendRequestAndCollectReplies(Iterator begin,
                             Iterator end,
                             Pred filter_pred,
                             AsyncOpParams params,
                             Sender sender,
                             ReplyHandler reply_handler,
                             std::string_view logging_description);
 
template <class T>
std::vector<std::exception_ptr> ExtractExceptions(std::vector<boost::future<T>>& futures) {
    std::vector<std::exception_ptr> exceptions;
    for (auto& f : futures) {
        try {
            f.get();
        } catch (std::exception const& e) {
            LOG4CPLUS_DEBUG("TEST", "E:" << e.what());
            exceptions.push_back(std::current_exception());
        }
    }
    return exceptions;
}
 
/**
 * Unwrap futures to extract errors.
 *
 * @throws DaqSourceErrors containing all exceptions.
 *
 * @ingroup daq_common_libdaq
 */
void UnwrapVoidReplies(boost::future<std::vector<boost::future<void>>> futures);
 
/**
 * Unwrap replies
 *
 * @ingroup daq_common_libdaq
 */
template <class R>
std::vector<R> UnwrapReplies(boost::future<std::vector<boost::future<R>>>&& futures) {
    // @todo: Implement
    return {};
}
 
/**
 * Reply handler that checks for exceptions in reply
 *
 * @throws Exception contained in reply.
 *
 * @return fut
 *
 * @param expected_state The state the source is expected to be in when receving the reply.
 * @param success_state The state the source is set to on success.
 * @param daq Data acquisition the operation belongs to.
 * @param source The source to handle reply from.
 * @param fut Reply.
 * @param func Callback invoked with reply which if it throws exception will generate alert.
 *
 * @ingroup daq_common_libdaq
 */
template <class ReplyType>
std::optional<ReplyType> HandleMetaDaqReply(char const* request,
                                            std::optional<State> expected_state,
                                            State success_state,
                                            std::optional<State> error_state,
                                            AsyncOpParams params,
                                            Source<MetaSource>& source,
                                            boost::future<ReplyType>&& fut,
                                            std::function<void(ReplyType const&)> func = {});
 
/**
 * Reply handler that checks for exceptions in reply
 *
 * @throws Exception contained in reply.
 *
 * @return fut contained value or std::nullopt of expected state is already satisifed
 *
 * @param expected_state The state the source is expected to be in when receving the reply.
 * @param success_state The state the source is set to on success.
 * @param daq Data acquisition the operation belongs to.
 * @param source The source to handle reply from.
 * @param fut Reply.
 *
 * @ingroup daq_common_libdaq
 */
template <class ReplyType>
std::optional<ReplyType> HandlePrimDaqReply(char const* request,
                                            std::optional<State> expected_state,
                                            State success_state,
                                            std::optional<State> error_state,
                                            AsyncOpParams params,
                                            Source<PrimSource>& source,
                                            boost::future<ReplyType>&& fut);
 
// Implementation
template <class R, class Iterator, class Pred, class Sender, class ReplyHandler>
boost::future<std::vector<boost::future<R>>>
SendRequestAndCollectReplies(Iterator begin,
                             Iterator end,
                             Pred filter_pred,
                             AsyncOpParams params,
                             Sender sender,
                             ReplyHandler reply_handler,
                             std::string_view logging_description) {
    LOG4CPLUS_INFO(params.logger,
                   fmt::format("{}: {}: Sending requests.", params.status, logging_description));
    using Sourceype = typename Iterator::value_type;
 
    std::vector<boost::future<R>> replies;
    bool stop = false;
 
    // Function that sends request and returns a processed (by reply_handler) reply
    // This transforms the type from the raw MAL future to whatever reply_handler returns.
    auto send_func = [&, sender = std::move(sender), descr = std::string(logging_description)](
                         Sourceype& source) -> boost::future<R> {
        try {
            // continuation
            auto cont =
                [=,
                     // The source vector should be considered immutable so it's ok to
                     // take ref.
                     & source = source,
                 reply_token = params.pending_replies.Acquire(
                     std::string(source.GetSource().GetName()), std::string(logging_description)),
                 reply_handler = std::move(reply_handler)](auto fut) mutable -> R {
                eventlog::Reply(params, source, descr, fut.has_exception());
                // Got reply, release reply_token.
                reply_token.Release();
                return std::invoke(reply_handler, params, source, std::move(fut));
            };
 
            eventlog::Request(params, source, descr);
 
            // Send request
            auto fut =
                std::invoke(std::move(sender), source).then(params.executor, std::move(cont));
            return fut;
        } catch (std::exception const& e) {
            // @todo: Do I need the error flag?
            source.SetErrorFlag();
            stop = true;  // just an optimiazation to not Send futher requests
 
            eventlog::Request(params, source, descr);
 
            LOG4CPLUS_ERROR(params.logger,
                            fmt::format("{}: {}: "
                                        "Error sending request to source '{}': {}",
                                        params.status,
                                        descr,
                                        source,
                                        e.what()));
            return boost::make_exceptional_future<R>();
        }
    };
 
    // Send
    for (auto it = begin; it != end; ++it) {
        if (stop) {
            break;
        }
        if (!filter_pred(*it)) {
            continue;
        }
 
        replies.emplace_back(send_func(*it));
    }
 
    LOG4CPLUS_INFO(
        params.logger,
        fmt::format(
            "{}: {}: Awaiting {} replies.", params.status, logging_description, replies.size()));
 
    // Join replies
    // note: when_all returns boost::future<vector<boost::future<R>>>
    return boost::when_all(replies.begin(), replies.end())
        .then(params.executor,
              [params, descr = std::string(logging_description)](
                  boost::future<std::vector<boost::future<R>>>&& replies_fut)
                  -> std::vector<boost::future<R>> {
                  auto replies = replies_fut.get();  // should never throw
                  LOG4CPLUS_INFO(
                      params.logger,
                      fmt::format("{}: {}: All replies received or timed out (num errors {}/{})",
                                  params.status,
                                  descr,
                                  eventlog::CountExceptions(replies),
                                  replies.size()));
                  return replies;
              });
}
 
}  // namespace daq::op
#endif  // #ifndef OCF_DAQ_OP_UTIL_HPP_