manager.cpp

Go to the documentation of this file.

/**
 * @file
 * @ingroup daq_ocm_libdaq
 * @copyright 2021 ESO - European Southern Observatory
 *
 * @brief Definition of `daq::ManagerImpl` and related utilities.
 */
#include <daq/manager.hpp>
 
#include <time.h>
#include <algorithm>
#include <stdexcept>
 
#include <log4cplus/loggingmacros.h>
#include <fmt/format.h>
#include <fmt/ostream.h>
#include <mal/Mal.hpp>
 
#include <daq/daqController.hpp>
#include <daq/status.hpp>
 
#include <daq/op/initiate.hpp>
#include <daq/op/awaitState.hpp>
 
namespace daq {
 
std::string MakeDaqIdCandidate(char const* instrument_id, unsigned jitter) {
    // '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 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 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);
    return std::string(time_str, n + 23);
}
 
ManagerImpl::ManagerImpl(rad::IoExecutor& executor, std::string instrument_id)
    : m_alive_token(std::make_shared<bool>())
    , m_executor(executor)
    , m_instrument_id(std::move(instrument_id))
    , m_logger(log4cplus::Logger::getInstance("daq.manager")) {
}
 
 
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()));
        }
    }
}
 
 
std::string ManagerImpl::MakeDaqId() const {
    for (unsigned jitter=0; ; ++jitter) {
        auto id_candidate = daq::MakeDaqIdCandidate(m_instrument_id.c_str(), jitter);
        if (!HaveDaq(id_candidate)) {
            return id_candidate;
        }
    }
}
 
bool ManagerImpl::HaveDaq(std::string_view id) const noexcept {
    return FindDaq(id);
}
 
void ManagerImpl::AddDaq(std::shared_ptr<DaqController> daq) {
    if (HaveDaq(daq->GetId())) {
        throw std::invalid_argument("DaqController with same id already exists");
    }
    m_daq_controllers.emplace_back(daq->GetId(),
                                   daq,
                                   daq->GetStatus()->ConnectObserver(
                                   [this](ObservableStatus const& status) {
                                      this->m_status_signal.Signal(status);
                                   }));
    // Notify observers that DAQ was added.
    m_status_signal.Signal(*daq->GetStatus());
}
 
Status ManagerImpl::GetStatus(std::string_view id) const {
    auto& daq = FindDaqOrThrow(id);
    return daq.GetStatus()->GetStatus();
}
 
ManagerImpl::Daq::Daq(std::string id_arg,
                      std::shared_ptr<DaqController> controller_arg,
                      boost::signals2::connection connection_arg) noexcept
    : id(std::move(id_arg))
    , controller(std::move(controller_arg))
    , connection(std::move(connection_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);
}
 
boost::future<State> ManagerImpl::StartDaqAsync(std::string_view id) {
    try {
        return FindDaqOrThrow(id).StartAsync();
    } 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& daq = FindDaqOrThrow(id);
        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 {
            auto [fut, abort] = op::InitiateAbortableOperation<op::AwaitStateAsync>(
                m_executor.get_io_context(), status, state, timeout);
            // 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++;
}
 
}  // namespace daq