manager.cpp

Go to the documentation of this file.

/**
 * @file
 * @ingroup daq_common_libdaq
 * @copyright 2022 ESO - European Southern Observatory
 *
 * @brief Definition of `daq::ManagerImpl` and related utilities.
 */
#include <daq/manager.hpp>
 
#include <algorithm>
#include <stdexcept>
#include <time.h>
 
#include <fmt/format.h>
#include <fmt/ostream.h>
#include <log4cplus/loggingmacros.h>
#include <mal/Mal.hpp>
 
#include <daq/conversion.hpp>
#include <daq/daqController.hpp>
#include <daq/error/report.hpp>
#include <daq/status.hpp>
#include <daq/workspace.hpp>
 
#include <daq/op/awaitState.hpp>
#include <daq/op/initiate.hpp>
 
namespace daq {
bool IsStale(ManagerParams const& params,
             State state,
             std::chrono::system_clock::time_point creation_time) {
    auto now = std::chrono::system_clock::now();
    if (IsFinalState(state)) {
        return true;
    }
    auto full = MakeState(state);
    if (full.state == daqif::StateAcquiring) {
        return now > creation_time + params.acquiring_stale_age;
    }
    if (full.state == daqif::StateMerging) {
        return now > creation_time + params.merging_stale_age;
    }
    return false;
}
 
std::string MakeIdCandidate(char const* instrument_id,
                            unsigned jitter,
                            std::chrono::system_clock::time_point* tp) {
    using time_point = std::chrono::system_clock::time_point;
    using duration = std::chrono::system_clock::duration;
    using seconds = std::chrono::seconds;
    using microseconds = std::chrono::microseconds;
    // 'KMOS.2017-06-02T16:45:55.701'
    // olas-id must be <= 5 characters
    // Format: <olas>-<strfmtime>.<frac>
    // Size:   1-5    1 20 1 3 = 30
    // chrono formatting is not provided until C++20
    struct timeval tv;
    struct timeval jitter_tv {
        0, jitter * 1000
    };  // 1 ms
    if (gettimeofday(&tv, nullptr) != 0) {
        // GCOVR_EXCL_START
        throw boost::enable_current_exception(std::system_error(errno, std::generic_category()));
        // GCOVR_EXCL_STOP
    }
    // Add jitter
    struct timeval res;
    timeradd(&tv, &jitter_tv, &res);
    tv = res;
 
    struct tm tm_time;
    time_t time = tv.tv_sec;
    if (gmtime_r(&time, &tm_time) == nullptr) {
        // GCOVR_EXCL_START
        throw boost::enable_current_exception(std::system_error(errno, std::generic_category()));
        // GCOVR_EXCL_STOP
    }
    char time_str[31] = {0};
    int n = snprintf(&time_str[0], 7, "%.5s.", instrument_id);
    // This part is always 20 characters long
    strftime(&time_str[n], 20, "%Y-%m-%dT%H:%M:%S", &tm_time);
    char frac_str[5];
    snprintf(&frac_str[0], 5, ".%.3d", static_cast<int>(tv.tv_usec / 1000.0));
    // Append the fractional part
    strncpy(&time_str[n + 19], &frac_str[0], 4);
    if (tp != nullptr) {
        // Store resulting time in out optional out param
        *tp = time_point(
            std::chrono::duration_cast<duration>(seconds(tv.tv_sec) + microseconds(tv.tv_usec)));
    }
    return std::string(time_str, n + 23);
}
 
ManagerImpl::ManagerImpl(rad::IoExecutor& executor,
                         ManagerParams params,
                         Workspace& workspace,
                         std::shared_ptr<ObservableEventLog> event_log,
                         DaqControllerFactory& daq_factory,
                         std::shared_ptr<DpmClient> dpm_client,
                         log4cplus::Logger const& logger)
    : m_alive_token(std::make_shared<bool>())
    , m_executor(executor)
    , m_params(std::move(params))
    , m_workspace(workspace)
    , m_event_log(std::move(event_log))
    , m_daq_factory(daq_factory)
    , m_dpm_client(std::move(dpm_client))
    , m_logger(logger) {
}
 
ManagerImpl::~ManagerImpl() noexcept {
    // Abort any ongoing operations
    for (auto& op : m_abort_funcs) {
        try {
            op.Abort();
        } catch (std::exception const& e) {
            LOG4CPLUS_WARN(m_logger,
                           fmt::format("ManagerImpl::~ManagerImpl: Error when aborting "
                                       "operation {}: {}",
                                       op.GetId(),
                                       e.what()));
        }
    }
}
 
void ManagerImpl::RestoreFromWorkspace() try {
    LOG4CPLUS_INFO(m_logger, "RestoreFromWorkspace: Starting");
    auto ids = m_workspace.LoadList();
    // New list of pruned DAQs
    decltype(ids) pruned;
 
    for (auto const& id : ids) {
        try {
            LOG4CPLUS_INFO(m_logger, "RestoreFromWorkspace: Loading state for DAQ " << id);
            auto context = m_workspace.LoadContext(id);
            auto status = m_workspace.LoadStatus(id);
 
            // Ignore stale DAQs
            if (IsStale(m_params, status.state, context.creation_time)) {
                LOG4CPLUS_INFO(m_logger,
                               "RestoreFromWorkspace: DAQ " << status << " is stale -> archiving");
                m_workspace.ArchiveDaq(id);
                continue;
            }
 
            auto full = MakeState(status.state);
            // DAQ should be loaded.
            if (full.state == daqif::StateAcquiring) {
                LOG4CPLUS_INFO(m_logger,
                               "RestoreFromWorkspace: Restoring OCM phase DAQ: " << status);
                auto daq = m_daq_factory.MakeOcmPhase(
                    std::move(context), std::make_shared<ObservableStatus>(status), m_event_log);
                assert(daq);
                AddDaq(daq, Store::No);
                // We keep this
                pruned.push_back(id);
            } else if (full.state == daqif::StateMerging) {
                LOG4CPLUS_INFO(m_logger,
                               "RestoreFromWorkspace: Restoring DPM phase DAQ: " << status);
                auto daq = m_daq_factory.MakeDpmPhase(
                    std::move(context), std::make_shared<ObservableStatus>(status), m_event_log);
                assert(daq);
                AddDaq(daq, Store::No);
 
                // We keep this
                pruned.push_back(id);
            } else {
                LOG4CPLUS_INFO(
                    m_logger,
                    "RestoreFromWorkspace: Skipping loading DAQ in unsupported state: " << status);
            }
        } catch (...) {
            error::NestedExceptionReporter r(std::current_exception());
            LOG4CPLUS_ERROR(m_logger,
                            "RestoreFromWorkspace: Loading state for DAQ "
                                << id << " failed (ignoring): " << r);
            try {
                m_workspace.ArchiveDaq(id);
            } catch (...) {
                error::NestedExceptionReporter r(std::current_exception());
                LOG4CPLUS_ERROR(m_logger,
                                "RestoreFromWorkspace: Failed to archive DAQ " << id
                                                                               << "(ignoring): \n"
                                                                               << r);
            }
            LOG4CPLUS_INFO(m_logger, "RestoreFromWorkspace: Skipping " << id);
        }
    }
 
    // Write back pruned list of DAQs
    m_workspace.StoreList(pruned);
 
    LOG4CPLUS_INFO(m_logger, "RestoreFromWorkspace: Successfully completed");
} catch (...) {
    LOG4CPLUS_INFO(m_logger, "RestoreFromWorkspace: Failed");
    std::throw_with_nested(std::runtime_error("Failed to restore from workspace"));
}
 
std::string ManagerImpl::MakeDaqId(std::chrono::system_clock::time_point* time) const {
    for (unsigned jitter = 0;; ++jitter) {
        auto id_candidate = daq::MakeIdCandidate(m_params.instrument_id.c_str(), jitter, time);
        if (!HaveDaq(id_candidate, id_candidate)) {
            return id_candidate;
        }
    }
}
 
bool ManagerImpl::HaveDaq(std::string_view id, std::string_view file_id) const noexcept {
    assert(!id.empty());
    auto it =
        std::find_if(m_daq_controllers.begin(), m_daq_controllers.end(), [=](auto const& daq) {
            // Return true if daq is equal
            // Return true if file_id is non-empty and equal
            return daq.id == id ||
                   (!file_id.empty() && file_id == daq.controller->GetContext().file_id);
        });
    if (it != m_daq_controllers.end()) {
        LOG4CPLUS_DEBUG(m_logger,
                        "Manager: Found conflicting DAQ: id="
                            << id << ", file_id=" << file_id << " with existing: id=" << it->id
                            << ", file_id=" << it->controller->GetContext().file_id);
        return true;
    }
    return false;
}
 
void ManagerImpl::AddInitialKeywords(DaqContext& ctx) {
    // note: ARCFILE and ORIGFILE is added by daqDpmMerge
    fits::KeywordVector kws;
    kws.emplace_back(std::in_place_type<fits::ValueKeyword>, "ORIGIN", m_params.origin);
    kws.emplace_back(std::in_place_type<fits::ValueKeyword>, "INSTRUME", m_params.instrument_id);
    // @todo:
    // - TELESCOP
    // - DATE?
 
    // Update so that OCM keywords are added first (DaqContext may already contain keywords from
    // request).
    fits::UpdateKeywords(kws, ctx.keywords, fits::ConflictPolicy::Skip);
    ctx.keywords.swap(kws);
}
 
void ManagerImpl::AddDaq(std::shared_ptr<DaqController> const& daq, Store store) {
    assert(daq);
    LOG4CPLUS_INFO(m_logger, "Manager: AddDaq: Attempting to add DAQ " << daq->GetId());
    if (daq->GetId().empty()) {
        throw boost::enable_current_exception(std::invalid_argument("DaqController has empty id!"));
    }
    if (daq->GetContext().file_id.empty()) {
        throw boost::enable_current_exception(
            std::invalid_argument("DaqController has empty file_id"));
    }
    if (HaveDaq(daq->GetId(), daq->GetContext().file_id)) {
        throw boost::enable_current_exception(
            std::invalid_argument("DaqController with same id already exists"));
    }
 
    if (store == Store::Yes) {
        // Requested to store DAQ (i.e. it was not loaded from workspace)
        m_workspace.StoreContext(daq->GetContext());
        m_workspace.StoreStatus(*daq->GetStatus());
    }
 
    if (IsActiveDpmState(daq->GetState())) {
        // Start monitoring to recover from e.g. state deviation due to OCM being offline
        // and not receiving published status updates from DPM.
        m_dpm_client->StartMonitorStatus(daq->GetId());
    }
 
    m_daq_controllers.emplace_back(
        daq->GetId(),
        daq,
        daq->GetStatus()->ConnectObserver([alive = std::weak_ptr<bool>(m_alive_token),
                                           prev_state = daq->GetState(),
                                           this](ObservableStatus const& status) mutable {
            if (alive.expired()) {
                LOG4CPLUS_INFO("daq", "Manager has expired");
                return;
            }
            if (IsFinalState(status.GetState())) {
                if (!IsFinalState(prev_state)) {
                    // Transition to final state -> Archive DAQ
                    LOG4CPLUS_INFO(
                        m_logger,
                        fmt::format("DAQ transitioned to a final state -> archiving: {} (prev {})",
                                    status,
                                    prev_state));
                    m_workspace.StoreStatus(status);
 
                    m_executor.submit([alive = alive, id = status.GetId(), this] {
                        if (alive.expired()) {
                            LOG4CPLUS_INFO("daq", "Manager has expired");
                            return;
                        }
                        ArchiveDaq(id);
                    });
                }
            } else {  // Not any final state
                m_workspace.StoreStatus(status);
 
                // Handle handover
                if (prev_state != State::Stopped && status.GetState() == State::Stopped) {
                    // If DAQ is stopped we need to move it to the merging phase
                    // To be safe we defer the execution.
                    // Manager lives as long as executor so we don't have to
                    // check liveness.
                    m_executor.submit([alive = alive, id = status.GetId(), this] {
                        if (alive.expired()) {
                            LOG4CPLUS_INFO("daq", "Manager has expired");
                            return;
                        }
                        MoveToMergePhase(id);
                    });
                } else if (prev_state == State::NotScheduled &&
                           IsActiveDpmState(status.GetState())) {
                    // Transition from NotScheduled -> Scheduled+ transfers responsibility to DPM.
                    // As such we want to monitor status in case status updates are not received.
                    m_dpm_client->StartMonitorStatus(status.GetId());
                }
            }
 
            // Signal other observers.
            this->m_status_signal.Signal(status);
 
            prev_state = status.GetState();
        }),
        daq->ConnectContext(
            [alive = std::weak_ptr<bool>(m_alive_token), this](DaqContext const& ctx) {
                if (alive.expired()) {
                    LOG4CPLUS_INFO("daq", "Manager has expired");
                    return;
                }
                m_workspace.StoreContext(ctx);
            }));
 
    if (store == Store::Yes) {
        // Store daq list
        StoreActiveDaqs();
    }
 
    // Notify observers that DAQ was added.
    m_status_signal.Signal(*daq->GetStatus());
 
    if (daq->GetState() == State::NotScheduled) {
        ScheduleDaqsAsync();
    }
}
 
void ManagerImpl::RemoveDaq(std::string_view id) {
    auto it = std::find_if(m_daq_controllers.begin(),
                           m_daq_controllers.end(),
                           [id](auto const& daq) { return daq.id == id; });
    if (it == m_daq_controllers.end()) {
        throw boost::enable_current_exception(
            std::invalid_argument(fmt::format("Remove DAQ failed - no id found: {}", id)));
    }
    m_daq_controllers.erase(it);
}
 
void ManagerImpl::ArchiveDaq(std::string const& id) {
    LOG4CPLUS_INFO(m_logger, fmt::format("Moving DAQ to archive: id={}", id));
 
    // Since DAQ has been completed we can stop monitoring for changes.
    m_dpm_client->StopMonitorStatus(id);
 
    // Archive persistent storage
    m_workspace.ArchiveDaq(id);
 
    // Remove daq controller
    RemoveDaq(id);
 
    StoreActiveDaqs();
    LOG4CPLUS_INFO(m_logger, fmt::format("Moving DAQ to archive done: id={}", id));
}
 
void ManagerImpl::StoreActiveDaqs() const {
    LOG4CPLUS_INFO(m_logger, "StoreActiveDaqs()");
    // And remove from
    std::vector<std::string> daqs;
    for (Daq const& daq : m_daq_controllers) {
        assert(daq.controller);
        if (!IsFinalState(daq.controller->GetState())) {
            daqs.push_back(daq.id);
        }
    }
    m_workspace.StoreList(daqs);
}
 
Status ManagerImpl::GetStatus(std::string_view id) const {
    auto* daq = FindDaq(id);
    if (daq) {
        return daq->GetStatus()->GetStatus();
    }
    // Try to find archived status
    auto maybe_status = m_workspace.LoadArchivedStatus(std::string(id));
    if (maybe_status) {
        return *maybe_status;
    } else {
        throw boost::enable_current_exception(
            std::invalid_argument(fmt::format("DaqController with id '{}' does not exist", id)));
    }
}
 
ManagerImpl::Daq::Daq(std::string id_arg,
                      std::shared_ptr<DaqController> controller_arg,
                      boost::signals2::connection conn_status_arg,
                      boost::signals2::connection conn_context_arg) noexcept
    : id(std::move(id_arg))
    , controller(std::move(controller_arg))
    , conn_status(std::move(conn_status_arg))
    , conn_context(std::move(conn_context_arg)) {
}
 
DaqController const* ManagerImpl::FindDaq(std::string_view id) const noexcept {
    return const_cast<ManagerImpl*>(this)->FindDaq(id);
}
 
DaqController* ManagerImpl::FindDaq(std::string_view id) noexcept {
    auto it = std::find_if(m_daq_controllers.begin(),
                           m_daq_controllers.end(),
                           [id](auto const& daq) { return daq.id == id; });
    if (it != m_daq_controllers.end()) {
        return it->controller.get();
    }
    return nullptr;
}
 
DaqController& ManagerImpl::FindDaqOrThrow(std::string_view id) {
    auto daq_ptr = FindDaq(id);
 
    if (!daq_ptr) {
        throw boost::enable_current_exception(std::invalid_argument(
            fmt::format("DaqController with id '{}' does not exist", std::string(id))));
    }
    return *daq_ptr;
}
 
DaqController const& ManagerImpl::FindDaqOrThrow(std::string_view id) const {
    return const_cast<ManagerImpl*>(this)->FindDaqOrThrow(id);
}
 
void ManagerImpl::MoveToMergePhase(std::string_view id) {
    auto* daq = FindDaq(id);
    if (!daq) {
        LOG4CPLUS_WARN(m_logger, fmt::format("Daq requested to move does not exist: id={}", id));
        return;
    }
    LOG4CPLUS_INFO(m_logger, fmt::format("Moving DAQ to merge-phase: id={}", id));
    // Copy state we want to keep.
    auto ctx = daq->GetContext();
    auto status = daq->GetStatus();
    auto event_log = daq->GetEventLog();
    // Delete old DAQ before creating new
    // note: this invalidates "daq"
    RemoveDaq(id);
 
    // Manually transition to first state in Merging
    status->SetState(State::NotScheduled);
 
    auto new_daq =
        m_daq_factory.MakeDpmPhase(std::move(ctx), std::move(status), std::move(event_log));
    AddDaq(new_daq);
}
 
boost::future<State> ManagerImpl::StartDaqAsync(DaqContext ctx) {
    try {
        AddInitialKeywords(ctx);
 
        auto id = ctx.id;
        auto file_id = ctx.file_id;
        auto daq = m_daq_factory.MakeOcmPhase(
            std::move(ctx),
            std::make_shared<ObservableStatus>(std::move(id), std::move(file_id)),
            m_event_log);
        assert(daq);
        AddDaq(daq);
        return daq->StartAsync()
            .then(m_executor,
                  [&, daq](boost::future<State> f) -> boost::future<State> {
                      if (f.has_exception()) {
                          // Any error during start may lead to partially started acquisition that
                          // we need to abort
                          return daq->AbortAsync(ErrorPolicy::Tolerant)
                              .then(m_executor,
                                    [f = std::move(f)](boost::future<Status>) mutable -> State {
                                        // We ignore errors from AbortAsync as we can't do anything
                                        // about it Then we return original error
                                        f.get();  // throws
                                        __builtin_unreachable();
                                    });
                      } else {
                          return boost::make_ready_future<State>(f.get());
                      }
                  })
            .unwrap();
    } catch (...) {
        return boost::make_exceptional_future<State>();
    }
}
 
boost::future<Status> ManagerImpl::StopDaqAsync(std::string_view id, ErrorPolicy policy) {
    try {
        return FindDaqOrThrow(id).StopAsync(policy);
    } catch (...) {
        return boost::make_exceptional_future<Status>();
    }
}
 
boost::future<Status> ManagerImpl::AbortDaqAsync(std::string_view id, ErrorPolicy policy) {
    try {
        return FindDaqOrThrow(id).AbortAsync(policy);
    } catch (...) {
        return boost::make_exceptional_future<Status>();
    }
}
 
boost::future<Result<Status>> ManagerImpl::AwaitDaqStateAsync(std::string_view id,
                                                              State state,
                                                              std::chrono::milliseconds timeout) {
    try {
        auto* maybe_daq = FindDaq(id);
        if (!maybe_daq) {
            // It could be the DAQ just completed and await should be released because of that.
            auto status = GetStatus(id);
            if (!IsSubsequentState(state, status.state)) {
                LOG4CPLUS_INFO(m_logger,
                               fmt::format("{}: Await condition already fulfilled (archived).", status));
                // Condition already fulfilled.
                return boost::make_ready_future<Result<Status>>({false, status});
            }
            // Since DAQ was archived we cannot wait anyway so we return exception
            throw boost::enable_current_exception(std::invalid_argument(
                fmt::format("DaqController with id '{}' is archived and cannot be awaited", id)));
        }
 
        // DAQ is active
        auto& daq = *maybe_daq;
        auto status = daq.GetStatus();
        if (!IsSubsequentState(state, daq.GetState())) {
            LOG4CPLUS_INFO(m_logger,
                           fmt::format("{}: Await condition already fulfilled.", *status));
            // Condition already fulfilled.
            return boost::make_ready_future<Result<Status>>({false, *status});
        } else {
            // Create child logger for await state.
            auto logger = log4cplus::Logger::getInstance(m_logger.getName() + ".awaitstate");
            auto [fut, abort] = op::InitiateAbortableOperation<op::AwaitStateAsync>(
                m_executor.get_io_context(), status, state, timeout, logger);
            // Store abort function so when Manager is deleted it will
            auto& ref = m_abort_funcs.emplace_back(std::move(abort));
            LOG4CPLUS_DEBUG("daq.manager",
                            fmt::format("op::AwaitStateAsync initiated. id={}", ref.GetId()));
            return fut.then(
                m_executor,
                [this, id = ref.GetId(), alive = std::weak_ptr<bool>(m_alive_token)](auto res) {
                    LOG4CPLUS_DEBUG("daq.manager",
                                    fmt::format("op::AwaitStateAsync completed. id={}", id));
                    // Remove abort function since operation completed, but only if
                    // object is alive.
                    auto is_alive = !alive.expired();
                    if (is_alive) {
                        // Manager is still alive, so we remove abort function
                        RemoveAbortFunc(id);
                    }
                    return res.get();
                });
        }
    } catch (...) {
        return boost::make_exceptional_future<Result<Status>>();
    }
}
 
void ManagerImpl::UpdateKeywords(std::string_view id, fits::KeywordVector const& keywords) {
    return FindDaqOrThrow(id).UpdateKeywords(keywords);
}
 
StatusSignal& ManagerImpl::GetStatusSignal() {
    return m_status_signal;
}
 
std::vector<std::shared_ptr<DaqController const>> ManagerImpl::GetDaqControllers() {
    std::vector<std::shared_ptr<DaqController const>> controllers;
    controllers.reserve(m_daq_controllers.size());
    std::transform(m_daq_controllers.begin(),
                   m_daq_controllers.end(),
                   std::back_inserter(controllers),
                   [](auto const& daq) { return daq.controller; });
    return controllers;
}
 
void ManagerImpl::RemoveAbortFunc(uint64_t id) noexcept {
    try {
        m_abort_funcs.erase(std::remove_if(m_abort_funcs.begin(),
                                           m_abort_funcs.end(),
                                           [id](auto const& obj) { return id == obj.GetId(); }),
                            m_abort_funcs.end());
    } catch (...) {
    }
}
 
ManagerImpl::OpAbortFunc::OpAbortFunc(OpAbortFunc::Func&& func)
    : m_id(NextId()), m_func(std::move(func)) {
    assert(m_func);
}
 
uint64_t ManagerImpl::OpAbortFunc::GetId() const noexcept {
    return m_id;
}
 
bool ManagerImpl::OpAbortFunc::Abort() noexcept {
    return m_func();
}
 
uint64_t ManagerImpl::OpAbortFunc::NextId() {
    static uint64_t next_id = 0;
    return next_id++;
}
 
void ManagerImpl::ScheduleDaqsAsync() {
    LOG4CPLUS_TRACE(m_logger, "ScheduleDaqAsync()");
    // Regardless if caller was invoked from timer or manually we reset the deadline timer
    // so that timer is restarted when ScheduleMergeAsync completes as necessary.
    m_schedule_retry.reset();
 
    for (auto& daq : m_daq_controllers) {
        if (daq.controller->GetState() != State::NotScheduled) {
            continue;
        }
        daq.controller->ScheduleMergeAsync().then(
            m_executor, [id = daq.id, this](boost::future<State> reply) {
                if (!reply.has_exception()) {
                    LOG4CPLUS_INFO(
                        m_logger,
                        fmt::format("ScheduleDaqAsyn: Successfully scheduled DAQ '{}'", id));
                    // Success
                    return;
                }
                LOG4CPLUS_WARN(
                    m_logger,
                    fmt::format("ScheduleDaqAsyn: Failed to schedule DAQ '{}', will retry in 60s",
                                id));
                // Some kind of error happened, at this point we don't care what it is
                // but simply schedule a new attempt using deadline timer, unless already scheduled.
                if (m_schedule_retry) {
                    // Already scheduled..
                    return;
                }
                m_schedule_retry.emplace(m_executor.get_io_context(),
                                         boost::posix_time::seconds(60));
                m_schedule_retry->async_wait([this](boost::system::error_code const& error) {
                    if (error) {
                        return;
                    }
                    ScheduleDaqsAsync();
                });
            });
    }
}
 
}  // namespace daq