abort.cpp

Go to the documentation of this file.

/**
 * @file
 * @ingroup daq_op
 * @copyright 2021 ESO - European Southern Observatory
 *
 * @brief Contains definition for the AbortAsync operation
 */
#include <daq/op/abort.hpp>
 
#include <Metadaqif.hpp>
#include <fmt/format.h>
#include <fmt/ostream.h>
#include <log4cplus/loggingmacros.h>
 
#include <daq/op/util.hpp>
#include <daq/state.hpp>
 
namespace daq::op {
 
AbortAsync::AbortAsync(ErrorPolicy policy, AsyncOpParams m_params) noexcept
    : m_policy(policy), m_params(m_params) {
}
 
[[nodiscard]] boost::future<AbortAsync::ResultType> AbortAsync::Initiate() {
    using boost::future;
 
    // Sequencing is not important to abort. So we send request to both primary and metadata
    // sources immediately instead of synchronizing them.
    return boost::when_all(AbortMeta(), AbortPrim())
        .then(m_params.executor,
              [this](future<std::tuple<future<void>, future<void>>> futs) -> ResultType {
                  LOG4CPLUS_DEBUG(
                      m_params.logger,
                      fmt::format("{}: AbortAsync: AbortMeta() and AbortPrim() completed.", m_params.status));
                  auto tup = futs.get();
                  auto& prim = std::get<0>(tup);
                  auto& meta = std::get<1>(tup);
 
                  if (prim.has_value() && meta.has_value()) {
                      LOG4CPLUS_INFO(
                          m_params.logger,
                          fmt::format("{}: AbortAsync: DAQ aborted successfully.", m_params.status));
                      return { false };
                  }
                  if (m_policy == ErrorPolicy::Tolerant) {
                      // Errors are non-fatal
                      LOG4CPLUS_INFO(
                          m_params.logger,
                          fmt::format("{}: AbortAsync: Data Acquisition abort has errors that will "
                                      "be ignored due to error policy.", m_params.status));
                    return { true };
                  } else {
                      // Errors are fatal
                      LOG4CPLUS_INFO(
                          m_params.logger,
                          fmt::format("{}: AbortAsync: Data Acquisition abort failed.", m_params.status));
 
                      // Either prim or meta has error, collect them:
                      std::vector<std::variant<DaqSourceError, std::exception_ptr>> errors;
                      try {
                          prim.get();  // throws
                      } catch (DaqSourceError const& e) {
                          errors.emplace_back(e);
                      } catch (...) {
                          errors.emplace_back(std::current_exception());
                      }
                      try {
                          meta.get();  // throws
                      } catch (DaqSourceError const& e) {
                          errors.emplace_back(e);
                      } catch (...) {
                          errors.emplace_back(std::current_exception());
                      }
                      LOG4CPLUS_DEBUG(
                          m_params.logger,
                          fmt::format("{}: Throwing exception.", m_params.status));
                      throw boost::enable_current_exception(DaqSourceErrors(std::move(errors)));
                  }
              });
}
 
 
[[nodiscard]] boost::future<void> AbortAsync::AbortMeta() {
    return SendRequestAndCollectReplies<void>(
               m_params.meta_sources.begin(),
               m_params.meta_sources.end(),
               [&](Source<MetaSource>& s) {
                   return s.GetState() != State::Aborted && s.GetState() != State::Stopped;
               },
               m_params,
               // Sender
               [id = m_params.id](Source<MetaSource>& s) {
                   s.SetState(State::Aborting);
                   return s.GetSource().GetRrClient().AbortDaq(id);
               },
               // reply handler
               [](AsyncOpParams m_params,
                  Source<MetaSource>& source,
                  boost::future<std::shared_ptr<metadaqif::DaqReply>>&& fut) -> void {
                   HandleMetaDaqReply(State::Aborting, State::Aborted, {}, m_params, source, std::move(fut));
               },
               std::string_view("AbortAsync: abort metadata acquisition"))
        .then(m_params.executor, UnwrapVoidReplies);  // @note: UnwrapVoidReplies is thread safe
}
 
 
[[nodiscard]] boost::future<void> AbortAsync::AbortPrim() {
    return SendRequestAndCollectReplies<void>(
               m_params.prim_sources.begin(),
               m_params.prim_sources.end(),
               [&](Source<PrimSource>& s) {
                   return !IsFinalState(s.GetState());
               },
               m_params,
               // Sender
               [](Source<PrimSource>& s) {
                   s.SetState(State::Aborting);
                    // id is not supportd by interface
                   return s.GetSource().GetRrClient().RecAbort();
               },
               // reply handler
               [](AsyncOpParams m_params,
                  Source<PrimSource>& source,
                  boost::future<std::string>&& fut) -> void {
                   HandlePrimDaqReply(State::Aborting, State::Aborted, {}, m_params, source, std::move(fut));
               },
               std::string_view("AbortAsync: abort primary data acquisition"))
        .then(m_params.executor, UnwrapVoidReplies);  // @note: UnwrapVoidReplies is thread safe
}
 
}  // namespace daq::op