daqController.cpp

Go to the documentation of this file.

#include <daq/daqController.hpp>
 
#include <iostream>
#include <stdexcept>
#include <type_traits>
 
#include <fmt/format.h>
#include <fmt/ostream.h>
#include <log4cplus/loggingmacros.h>
 
#include <daq/error.hpp>
#include <daq/op/abort.hpp>
#include <daq/op/initiate.hpp>
#include <daq/op/start.hpp>
#include <daq/op/stop.hpp>
#include <daq/op/awaitPrim.hpp>
 
namespace daq {
namespace {
 
template <class>
inline constexpr bool always_false_v = false;  // NOLINT
 
template <class Sources>
bool AllInState(Sources sources, State state) {
    return std::all_of(
        sources.begin(), sources.end(), [=](auto const& s) -> bool { return s.state == state; });
}
 
}  // namespace
 
AsyncOperations::AsyncOperations()
    : start([](auto par) { return daq::op::InitiateOperation<daq::op::StartAsync>(par); })
    , abort([](ErrorPolicy policy, auto par) {
        return daq::op::InitiateOperation<op::AbortAsync>(policy, std::move(par));
    })
    , stop([](ErrorPolicy policy, auto par) {
        return daq::op::InitiateOperation<op::StopAsync>(policy, std::move(par));
    })
    , await_prim([](auto par) {
        return daq::op::InitiateAbortableOperation<op::AwaitPrimAsync>(std::move(par));
    }) {
}
 
bool AsyncOperations::IsValid() const noexcept {
    return start && stop && abort && await_prim;
}
 
std::ostream& operator<<(std::ostream& os, DaqController const& daq) {
    os << "DaqController(id='" << daq.GetId() << "', state=" << daq.GetState() << ")";
    return os;
}
 
std::ostream& operator<<(std::ostream& os, State state) {
    switch (state) {
    case State::NotStarted:
        os << "NotStarted";
        return os;
    case State::Starting:
        os << "Starting";
        return os;
    case State::Acquiring:
        os << "Acquiring";
        return os;
    case State::Stopping:
        os << "Stopping";
        return os;
    case State::Stopped:
        os << "Stopped";
        return os;
    case State::Aborting:
        os << "Aborting";
        return os;
    case State::Aborted:
        os << "Aborted";
        return os;
    // GCOVR_EXCL_START
    default:
        os << "Unknown";
        return os;
        // GCOVR_EXCL_STOP
    }
}
 
std::shared_ptr<DaqControllerImpl>
DaqControllerImpl::Create(boost::asio::io_context& io_context,
                          DaqProperties properties,
                          std::shared_ptr<ObservableStatus> status,
                          std::shared_ptr<ObservableEventLog> event_log,
                          AsyncOperations ops) {
    // note: make_shared doesn't work since constructor is protected,
    // to protect against non-shared ownership.
    LOG4CPLUS_TRACE("daq", fmt::format("DaqControllerImpl::Create"));
    auto fits_ctl = std::make_unique<FitsControllerImpl>(properties, event_log);
    return std::shared_ptr<DaqControllerImpl>(new DaqControllerImpl(io_context,
                                                                    std::move(properties),
                                                                    std::move(fits_ctl),
                                                                    std::move(status),
                                                                    std::move(event_log),
                                                                    std::move(ops)));
}
 
DaqControllerImpl::DaqControllerImpl(boost::asio::io_context& io_context,
                                     DaqProperties properties,
                                     std::unique_ptr<FitsController> fits_ctl,
                                     std::shared_ptr<ObservableStatus> status,
                                     std::shared_ptr<ObservableEventLog> event_log,
                                     AsyncOperations ops)
    : m_state(MakeState(status->GetState()))
    , m_io_ctx(io_context)
    , m_executor(io_context)
    , m_properties(std::move(properties))
    , m_fits_ctl(std::move(fits_ctl))
    , m_status(std::move(status))
    , m_event_log(std::move(event_log))
    , m_prim_sources(MakeSources<PrimSource>(m_properties.prim_sources))
    , m_meta_sources(MakeSources<MetaSource>(m_properties.meta_sources))
    , m_async_ops(std::move(ops))
    , m_pending_replies(PendingReplies::Create())
    , m_logger(log4cplus::Logger::getInstance("daq")) {
    assert(m_status);
    assert(m_event_log);
 
    if (m_properties.id != m_status->GetId()) {
        throw std::invalid_argument(
            "Data acquisition id mismatch between DaqProperties and ObservableStatus");
    }
 
    if (m_prim_sources.empty() && m_meta_sources.empty()) {
        throw std::invalid_argument("No data sources provided");
    }
 
    if (!m_async_ops.IsValid()) {
        throw std::invalid_argument("AsyncOperations is invalid");
    }
}
 
State DaqControllerImpl::GetState() const DAQ_NOEXCEPT {
    State s;
    std::visit(
        [&](auto const& var) {
            using T = std::decay_t<decltype(var)>;
            if constexpr (std::is_same_v<T, NotStarted>) {
                s = State::NotStarted;
            } else if constexpr (std::is_same_v<T, Starting>) {
                s = State::Starting;
            } else if constexpr (std::is_same_v<T, Acquiring>) {
                s = State::Acquiring;
            } else if constexpr (std::is_same_v<T, Stopping>) {
                s = State::Stopping;
            } else if constexpr (std::is_same_v<T, Stopped>) {
                s = State::Stopped;
            } else if constexpr (std::is_same_v<T, Aborting>) {
                s = State::Aborting;
            } else if constexpr (std::is_same_v<T, Aborted>) {
                s = State::Aborted;
            } else {
                static_assert(always_false_v<T>, "non-exhaustive visitor!");
            }
        },
        m_state);
    return s;
}
 
std::shared_ptr<ObservableStatus> DaqControllerImpl::GetStatus() DAQ_NOEXCEPT {
    return m_status;
}
 
std::shared_ptr<ObservableStatus const> DaqControllerImpl::GetStatus() const DAQ_NOEXCEPT {
    return m_status;
}
 
std::shared_ptr<ObservableEventLog> DaqControllerImpl::GetEventLog() DAQ_NOEXCEPT {
    return m_event_log;
}
 
std::string const& DaqControllerImpl::GetId() const DAQ_NOEXCEPT {
    return m_status->GetId();
}
 
bool DaqControllerImpl::GetErrorFlag() const DAQ_NOEXCEPT {
    return m_status->GetError();
}
 
constexpr log4cplus::Logger const& DaqControllerImpl::GetLogger() const noexcept {
    return m_logger;
}
 
DaqControllerImpl::StateVariant DaqControllerImpl::MakeState(State s) const noexcept {
    switch (s) {
        case State::NotStarted:
            return StateVariant(NotStarted());
        case State::Starting:
            return StateVariant(Starting());
        case State::Acquiring:
            return StateVariant(Acquiring());
        case State::Stopping:
            return StateVariant(Stopping());
        case State::Stopped:
            return StateVariant(Stopped());
        case State::Aborting:
            return StateVariant(Aborting());
        case State::Aborted:
            return StateVariant(Aborted());
    };
    LOG4CPLUS_FATAL(m_logger, fmt::format("Invalid state provided: '{}'", s));
    std::terminate();
}
 
void DaqControllerImpl::SetErrorFlag(bool error) noexcept {
    m_status->SetError(error);
}
 
void DaqControllerImpl::SetState(DaqControllerImpl::StateVariant&& s) noexcept {
    m_state = s;
    // Publish changes
    m_status->SetState(GetState());
}
 
op::AsyncOpParams DaqControllerImpl::MakeParams() {
    return op::AsyncOpParams(*m_status.get(),
                             *m_event_log.get(),
                             m_executor,
                             m_logger,
                             GetId(),
                             *m_pending_replies.get(),
                             m_prim_sources,
                             m_meta_sources);
}
 
op::AwaitOpParams DaqControllerImpl::MakeAwaitParams() {
    return op::AwaitOpParams(op::AsyncOpParams(*m_status.get(),
                                               *m_event_log.get(),
                                               m_executor,
                                               m_logger,
                                               GetId(),
                                               *m_pending_replies.get(),
                                               m_prim_sources,
                                               m_meta_sources),
                             m_properties.await_interval);
}
 
boost::future<State> DaqControllerImpl::StartAsync() {
    m_event_log->AddEvent(
        ActionEvent(GetId(), "DaqController::StartAsync()", m_status->GetStatus()));
 
    // Make sure we're not already started.
    if (!std::holds_alternative<NotStarted>(m_state)) {
        return boost::make_exceptional_future<State>(
            std::runtime_error("Data acquisition is already started"));
    }
 
    SetState(Starting{});
 
    // Special handling for the FITS file produced by OCM
    try {
        m_fits_ctl->Start();
    } catch (std::exception const& e) {
        LOG4CPLUS_ERROR(
            GetLogger(),
            fmt::format("{}: StartAsync: FitsController::Start failed (aborting start): {}",
                        *this,
                        e.what()));
        SetErrorFlag(true);
        return boost::make_exceptional_future<State>();
    }
 
    return m_async_ops.start(MakeParams())
        .then(m_executor,
              [daq = std::static_pointer_cast<DaqControllerImpl>(shared_from_this())](
                  boost::future<void> f) {
                  LOG4CPLUS_INFO(daq->GetLogger(),
                                 fmt::format("{}: StartAsync: Completed {}",
                                             *daq,
                                             f.has_value() ? "successfully" : "with error"));
                  // Re-raise any exception
                  if (f.has_exception()) {
                      daq->SetErrorFlag(true);
                      (void)f.get();
                  }
                  // Await completion if there are primary data sources
                  daq->InitiateAwaitPrimarySources();
 
                  // No exception -> all done! update and return Acquiring state
                  daq->SetErrorFlag(false);
                  daq->SetState(Acquiring{});
                  return daq->GetState();
              });
}
 
boost::future<Status> DaqControllerImpl::StopAsync(ErrorPolicy policy) {
    m_event_log->AddEvent(
        ActionEvent(GetId(), "DaqController::StopAsync()", m_status->GetStatus()));
 
    if (std::holds_alternative<Stopped>(m_state)) {
        return boost::make_exceptional_future<Status>(
            std::runtime_error("Data acquisition already stopped"));
    }
 
    if (std::holds_alternative<Aborted>(m_state)) {
        return boost::make_exceptional_future<Status>(
            std::runtime_error("Data acquisition already aborted"));
    }
 
    if (!std::holds_alternative<Acquiring>(m_state)) {
        return boost::make_exceptional_future<Status>(
            std::runtime_error("Cannot stop a data acquisition that is not Acquiring"));
    }
 
    // @todo If we're in Starting this will potentially mess up assumption that we're
    // in Starting. Revisit to remove this assumption?
    // @todo: Store produced files
    // m_status.AddFiles(reply.getFiles());
    SetState(Stopping{});
 
    return m_async_ops.stop(policy, MakeParams())
        .then(m_executor,
              [policy, daq = std::static_pointer_cast<DaqControllerImpl>(shared_from_this())](
                  boost::future<Result<DpParts>> f) -> Status {
                  LOG4CPLUS_INFO(daq->GetLogger(),
                                 fmt::format("{}: StopAsync: Completed {}",
                                             *daq,
                                             f.has_value() ? "successfully" : "with error"));
                  if (!f.has_exception() || policy == ErrorPolicy::Tolerant) {
                      // Stop local FITS controller on success or if forced
                      try {
                          // Update keywords in FITS controller from current status, then save
                          // stop the controller to save the state.
                          daq->m_fits_ctl->UpdateKeywords(daq->m_status->GetKeywords());
                          auto res = daq->m_fits_ctl->Stop(policy);
                          if (res) {
                              daq->m_status->AddFiles({*res});
                          } else {
                              // If policy prevented an error to be thrown, but there is no
                              // resulting file we mark it as an error
                              daq->SetErrorFlag(true);
                          }
                      } catch (std::exception const& e) {
                          LOG4CPLUS_ERROR(daq->GetLogger(),
                                         fmt::format("{}: StopAsync: Failed to create OCM FITS "
                                                     "file: {}",
                                                     *daq,
                                                     e.what()));
                          daq->SetErrorFlag(true);
                          throw;
                      } catch (...) {
                          daq->SetErrorFlag(true);
                          throw;
                      }
                  }
 
                  if (f.has_exception()) {
                      daq->SetErrorFlag(true);
                      (void)f.get();  // Throw to propagate any error
                  }
 
                  auto result = f.get();
                  daq->m_status->AddFiles(result.result);
 
                  // If there were no exceptions we're all done.
                  // Update and return Stopped state
                  daq->SetErrorFlag(result.error);
                  daq->SetState(Stopped{});
                  return daq->GetStatus()->GetStatus();
              });
}
 
boost::future<Status> DaqControllerImpl::AbortAsync(ErrorPolicy policy) {
    m_event_log->AddEvent(ActionEvent(
        GetId(), fmt::format("DaqController::AbortAsync({})", policy), m_status->GetStatus()));
 
    if (std::holds_alternative<NotStarted>(m_state)) {
        LOG4CPLUS_INFO(m_logger, fmt::format("{}: Aborting not started data acquisition", *this));
        SetState(Aborted{});
        return boost::make_ready_future<Status>(GetStatus()->GetStatus());
    }
 
    if (std::holds_alternative<Stopped>(m_state) || std::holds_alternative<Aborted>(m_state)) {
        return boost::make_exceptional_future<Status>(
            std::runtime_error("Data acquisition already stopped or aborted"));
    }
 
    SetState(Aborting{});
 
    return m_async_ops.abort(policy, MakeParams())
        .then(m_executor,
              [policy, daq = std::static_pointer_cast<DaqControllerImpl>(shared_from_this())](
                  boost::future<Result<void>> f) -> Status {
                  if (daq->GetState() == State::Aborted) {
                      // It can happen that a request to abort was superseded and finalized before
                      // reply was received.
                      LOG4CPLUS_INFO(
                          daq->GetLogger(),
                          fmt::format("{}: AbortAsync: Data acquisition already aborted. "
                                      "Do nothing else (errors are ignored). ",
                                      *daq));
                      // @todo: Should throw instead as the command was superseeded by other
                      // command.
                      return daq->GetStatus()->GetStatus();
                  }
                  LOG4CPLUS_DEBUG(daq->GetLogger(),
                                  fmt::format("{}: AbortAsync: Completed, updating DAQ status and "
                                              "set reply remaining. Has fatal error={}",
                                              *daq,
                                              f.has_exception()));
 
                  if (!f.has_exception() || policy == ErrorPolicy::Tolerant) {
                      // Stop local FITS controller on success or if forced
                      try {
                          daq->m_fits_ctl->Abort(policy);
                      } catch (...) {
                          daq->SetErrorFlag(true);
                          if (policy != ErrorPolicy::Tolerant) {
                              throw;
                          }
                      }
                  }
 
                  if (f.has_exception()) {
                      LOG4CPLUS_INFO(
                          daq->GetLogger(),
                          fmt::format("{}: AbortAsync: Completed with fatal error.", *daq));
 
                      daq->SetErrorFlag(true);
                      f.get();  // throw to propagate
                  }
                  auto result = f.get();
                  daq->SetErrorFlag(result.error);
 
                  // Success
                  daq->SetState(Aborted{});
                  LOG4CPLUS_INFO(daq->GetLogger(),
                                 fmt::format("{}: AbortAsync: Completed successfully.", *daq));
                  return daq->GetStatus()->GetStatus();
              });
}
 
void DaqControllerImpl::UpdateKeywords(fits::KeywordVector const& keywords) {
    m_event_log->AddEvent(ActionEvent(
        GetId(), fmt::format("DaqController::UpdateKeywords(<omitted>)"), m_status->GetStatus()));
 
    if (std::holds_alternative<Stopped>(m_state) || std::holds_alternative<Aborted>(m_state)) {
        throw boost::enable_current_exception(
            std::runtime_error("Data acquisition already stopped or aborted"));
    }
 
    m_status->UpdateKeywords(keywords);
}
 
boost::future<State>
DaqControllerImpl::AwaitAsync(std::vector<std::string> sources, std::chrono::milliseconds timeout) {
    m_event_log->AddEvent(ActionEvent(
        GetId(),
        fmt::format("DaqController::AwaitAsync({}, {} ms)",
                    sources.empty() ? "all primary sources" : "a user defined list of sources",
                    timeout.count()),
        m_status->GetStatus()));
 
    std::vector<
        std::variant<gsl::not_null<Source<PrimSource>*>, gsl::not_null<Source<MetaSource>*>>>
        await_on;
 
    if (!sources.empty()) {
        for (auto const& source_id : sources) {
            auto source_var = FindSource(source_id);
            if (source_var) {
                await_on.emplace_back(*source_var);
            } else {
                return boost::make_exceptional_future<State>(
                    std::invalid_argument(fmt::format("Source with id='{}' not found", source_id)));
            }
        }
    } else {
        // Use all primary sources by default
        for (auto& source : m_prim_sources) {
            await_on.emplace_back(&source);
        }
    }
 
    // note that condition references sources in DaqControllerImpl and should not be invoked
    // unless DaqControllerImpl is alive.
    auto condition = [sources = await_on]() {
        return std::all_of(sources.begin(), sources.end(), [](auto var) {
            return std::visit(
                [](auto v) {
                    return v->GetState() == State::Aborted || v->GetState() == State::Stopped;
                },
                var);
        });
    };
 
    // Test if condition is already satified
    if (condition()) {
        return boost::make_ready_future<State>(GetState());
    }
 
    // Wait is not already satisfied, attach state listeners to all sources.
    // daq is captured with a weak ptr to avoid keeping DaqControllerImpl alive if no state changes
    // occur on monitored components.
    auto promise = std::make_shared<boost::promise<State>>();
 
    if (timeout > std::chrono::milliseconds(0)) {
        auto timer = std::make_unique<boost::asio::steady_timer>(m_io_ctx, timeout);
        timer->async_wait([promise,
                           timer_ptr = timer.get(),
                           daq_weak = std::weak_ptr<DaqControllerImpl>(
                               std::static_pointer_cast<DaqControllerImpl>(shared_from_this()))](
                              boost::system::error_code ec) {
            // Promise might already be fulfilled, there's no way to check if it is though.
            try {
                auto daq = daq_weak.lock();
                if (ec) {
                    if (daq) {
                        LOG4CPLUS_DEBUG(
                            daq->m_logger,
                            fmt::format("{}: AsyncWait: Operation abandoned before completing.",
                                        *daq));
                    }
                    // Timer deleted or was cancelled, set exception in promise
                    promise->set_exception(DaqOperationAborted(""));
                } else {
                    // Normal timeout
                    // For this case we also want to delete the timer itself
                    if (daq) {
                        LOG4CPLUS_DEBUG(
                            daq->m_logger,
                            fmt::format("{}: AsyncWait: Operation timed out before completing.",
                                        *daq));
                        daq->m_timers.erase(
                            std::remove_if(
                                daq->m_timers.begin(),
                                daq->m_timers.end(),
                                [timer_ptr](std::unique_ptr<boost::asio::steady_timer>& val) {
                                    return val.get() == timer_ptr;
                                }),
                            daq->m_timers.end());
                    }
 
                    promise->set_exception(DaqOperationTimeout(""));
                }
            } catch (...) {
            }
        });
        m_timers.emplace_back(std::move(timer));
    }
 
    auto listener = [condition,
                     promise,
                     daq_weak = std::weak_ptr<DaqControllerImpl>(
                         std::static_pointer_cast<DaqControllerImpl>(shared_from_this()))](State,
                                                                                           bool) {
        auto daq = daq_weak.lock();
        if (!daq) {
            LOG4CPLUS_WARN("daq", "DaqControllerImpl deleted before await condition was fulfulled");
            // this async op was abandoned. Do nothing.
            return;
        }
 
        if (condition()) {
            LOG4CPLUS_INFO(daq->m_logger,
                           fmt::format("{}: AwaitAsync: Await condition fulfilled", *daq));
            try {
                promise->set_value(daq->GetState());
            } catch (...) {
                // Exception might be thrown because promise is already fulfilled, which is
                // expected to happen.
            }
        }
    };
 
    // Connect listeners to sources that should be awaited.
    for (auto& source : await_on) {
        std::visit(
            [daq = std::static_pointer_cast<DaqControllerImpl>(shared_from_this()),
             listener](auto s) {
                LOG4CPLUS_DEBUG(
                    daq->m_logger,
                    fmt::format("{}: AsyncWait: Attaching listener on source '{}'.", *daq, *s));
                // Use automatic connection disconnect if daq is destroyed. (i.e. track_foreign).
                using SlotType = typename std::remove_reference<decltype(
                    *s.get())>::type::StateSignal::slot_type;
                s->ConnectStateListener(SlotType(std::move(listener)).track_foreign(daq));
            },
            source);
    }
    return promise->get_future();
}
 
std::optional<std::variant<gsl::not_null<Source<PrimSource>*>, gsl::not_null<Source<MetaSource>*>>>
DaqControllerImpl::FindSource(std::string_view source_id) {
    {
        auto it =
            std::find_if(m_prim_sources.begin(), m_prim_sources.end(), [source_id](auto& source) {
                return source.GetSource().GetName() == source_id;
            });
        if (it != m_prim_sources.end()) {
            return gsl::not_null<Source<PrimSource>*>(&(*it));
        }
    }
    {
        auto it =
            std::find_if(m_meta_sources.begin(), m_meta_sources.end(), [source_id](auto& source) {
                return source.GetSource().GetName() == source_id;
            });
        if (it != m_meta_sources.end()) {
            return gsl::not_null<Source<MetaSource>*>(&(*it));
        }
    }
 
    // Not found
    return {};
}
 
template <class SourceType>
std::vector<Source<SourceType>> DaqControllerImpl::MakeSources(std::vector<SourceType> sources) {
    //@todo: Check for duplicates
    std::vector<Source<SourceType>> dest;
    dest.reserve(sources.size());
    std::transform(
        std::make_move_iterator(sources.begin()),
        std::make_move_iterator(sources.end()),
        std::back_inserter(dest),
        [](SourceType&& s) -> Source<SourceType> { return Source<SourceType>{std::move(s)}; });
    return dest;
}
 
void DaqControllerImpl::InitiateAwaitPrimarySources() {
    AddEvent<ActionEvent>(GetId(),
                          "DaqController::InitiateAwaitPrimarySources(): Initiating",
                          m_status->GetStatus());
    if (m_prim_sources.empty()) {
        LOG4CPLUS_DEBUG(
            m_logger,
            fmt::format("{}: InitiateAwaitPrimarySources: No primary sources to monitor.", *this));
        return;
    }
    LOG4CPLUS_DEBUG(m_logger,
                    fmt::format("{}: InitiateAwaitPrimarySources: Starting operation.", *this));
    auto [future, abort] = m_async_ops.await_prim(MakeAwaitParams());
    m_abort_await_primary_sources = abort;
    // Set up continuation that stops data acquisition automatically
    future.then(
        m_executor,
        [daq = std::static_pointer_cast<DaqControllerImpl>(shared_from_this())](boost::future<Result<DpParts>> fut) {
            // Add FITS files from the stopped sources
            auto result = fut.get();
            LOG4CPLUS_DEBUG(daq->m_logger,
                    fmt::format("{}: InitiateAwaitPrimarySources: Adding {} files from "
                                "primary sources", *daq,
                                result.result.size()));
            daq->m_status->AddFiles({result.result});
            // @todo: How to treat errors from await?
 
            // If daq is still acquiring we stop using strict error policy, otherwise do nothing
            if (!std::holds_alternative<Acquiring>(daq->m_state)) {
                LOG4CPLUS_DEBUG(
                    daq->m_logger,
                    fmt::format(
                        "{}: InitiateAwaitPrimarySources: "
                        "AwaitAsync completed but another operation has already transitioned "
                        "DAQ from Acquiring so automatic stop will not be performed.",
                        *daq));
                return;
            }
            daq->AddEvent<ActionEvent>(
                daq->GetId(),
                "DaqController::InitiateAwaitPrimarySources(): "
                "Primary sources completed. Performing automatic stop of metadata sources",
                daq->m_status->GetStatus());
            // No continuation is necessary. If an error occurs that information is published
            // and user needs to intervene.
            // If a separate topic is created for errors only it may be published with a
            // continuation attached to the result of StopAsync (or more likely implemented inside
            // StopAsync).
            daq->StopAsync(ErrorPolicy::Strict);
        });
}
 
}  // namespace daq