dpmService.cpp

Go to the documentation of this file.

/**
 * @file
 * @ingroup daq_dpm
 * @copyright (c) Copyright ESO 2022
 * All Rights Reserved
 * ESO (eso.org) is an Intergovernmental Organisation, and therefore special legal conditions apply.
 *
 * @brief daq::dpm::DpmService definition.
 */
#include <daq/dpm/dpmService.hpp>
 
#include <exception>
 
#include <boost/asio/io_context.hpp>
#include <boost/throw_exception.hpp>
 
#include <fmt/format.h>
#include <log4cplus/loggingmacros.h>
 
#include <daq/conversion.hpp>
#include <daq/dpm/config.hpp>
#include <daq/error/report.hpp>
#include <daq/json.hpp>
 
namespace daq::dpm {
 
DpmService::DpmService(rad::IoExecutor& executor,
                       elt::mal::Mal& mal,
                       Workspace& workspace,
                       Scheduler& scheduler)
    : m_executor(executor)
    , m_mal(mal)
    , m_workspace(workspace)
    , m_scheduler(scheduler)
    , m_work_guard(m_executor.get_io_context().get_executor())
    , m_sigset(executor.get_io_context(), SIGINT, SIGTERM)
    , m_logger(log4cplus::Logger::getInstance(LOGGER_NAME)) {
    InitiateSignalHandler();
}
 
boost::future<std::shared_ptr<::daqif::StorageStatus>> DpmService::QueryStorageStatus() {
    try {
        // Note: QueryStorageStatus() is thread-safe
        auto status = m_workspace.QueryStorageStatus();
 
        std::shared_ptr<daqif::StorageStatus> rep = m_mal.createDataEntity<daqif::StorageStatus>();
        rep->setCapacity(status.capacity);
        rep->setFree(status.free);
        rep->setAvailable(status.available);
 
        return boost::make_ready_future<std::shared_ptr<daqif::StorageStatus>>(rep);
    } catch (...) {
        error::NestedExceptionReporter r(std::current_exception());
        return boost::make_exceptional_future<std::shared_ptr<daqif::StorageStatus>>(
            daqif::RuntimeError(r.Str()));
    }
}
 
boost::future<std::string> DpmService::Exit() {
    // Removing guard which will cause asio::io_context::run to finish when it runs out of pending
    // work.
    // Note: If e.g. timers are added continously then there will always be an outstanding operation
    // so it won't complete.
    LOG4CPLUS_INFO(m_logger, "Application termination requested with Exit()");
    return boost::async(m_executor, [this]() mutable -> std::string {
        HandleExit();
        return "OK";
    });
}
 
boost::future<std::string> DpmService::GetVersion() {
    return boost::make_ready_future(std::string(VERSION));
}
 
boost::future<std::shared_ptr<::daqif::DaqReply>>
DpmService::QueueDaq(std::string const& status, std::string const& specification) {
    LOG4CPLUS_DEBUG(
        m_logger,
        "QueueDaq() Queuing new DAQ:\nStatus: " << status << "\nSpecification: " << specification);
    using V = ::daqif::DaqReply;
    using R = std::shared_ptr<V>;
    return boost::async(m_executor, [status, specification, this]() mutable -> R {
        try {
            auto id = m_scheduler.QueueDaq(status, specification);
            try {
                auto rep = m_mal.createDataEntity<V>();
                rep->setId(id);
                rep->setError(false);
                return rep;
            } catch (...) {
                error::NestedExceptionReporter r(std::current_exception());
                BOOST_THROW_EXCEPTION(daqif::DaqException(id, r.Str()));
            }
        } catch (daqif::DaqException const& e) {
            error::NestedExceptionReporter r(e);
            LOG4CPLUS_ERROR(m_logger, "QueueDaq() failed with ICD error:\n" << r.Str());
            throw;
        } catch (...) {
            error::NestedExceptionReporter r(std::current_exception());
            LOG4CPLUS_ERROR(m_logger, "QueueDaq() failed with following error(s):\n" << r);
            BOOST_THROW_EXCEPTION(daqif::DaqException("", r.Str()));
        }
    });
}
 
boost::future<std::shared_ptr<::daqif::DaqReply>> DpmService::AbortDaq(const std::string& id) {
    LOG4CPLUS_INFO(m_logger, fmt::format("AbortDaq({}): Handling request", id));
    using V = ::daqif::DaqReply;
    using R = std::shared_ptr<V>;
    return boost::async(m_executor, [id, this]() mutable -> R {
        try {
            m_scheduler.AbortDaq(id);
            try {
                auto rep = m_mal.createDataEntity<V>();
                rep->setId(id);
                rep->setError(false);
                return rep;
            } catch (...) {
                error::NestedExceptionReporter r(std::current_exception());
                throw daqif::DaqException(id, r.Str());
            }
        } catch (daqif::DaqException const& e) {
            error::NestedExceptionReporter r(e);
            LOG4CPLUS_ERROR(m_logger, "AbortDaq() failed with following error(s):\n" << r);
            throw;
        } catch (...) {
            error::NestedExceptionReporter r(std::current_exception());
            LOG4CPLUS_ERROR(m_logger, "AbortDaq() failed with following error(s):\n" << r);
            BOOST_THROW_EXCEPTION(daqif::DaqException("", r.Str()));
        }
    });
}
 
boost::future<std::shared_ptr<::daqif::DaqStatus>> DpmService::GetDaqStatus(const std::string& id) {
    LOG4CPLUS_INFO(m_logger, fmt::format("GetDaqStatus({}): Handling request", id));
    using V = ::daqif::DaqStatus;
    using R = std::shared_ptr<V>;
    return boost::async(m_executor, [id, this]() mutable -> R {
        try {
            return GetDaqStatusSync(id);
        } catch (daqif::DaqException const& e) {
            error::NestedExceptionReporter r(e);
            LOG4CPLUS_ERROR(m_logger, "GetDaqStatus() failed with following error(s):\n" << r);
            throw;
        } catch (...) {
            error::NestedExceptionReporter r(std::current_exception());
            LOG4CPLUS_ERROR(m_logger, "GetDaqStatus() failed with following error(s):\n" << r);
            BOOST_THROW_EXCEPTION(daqif::DaqException(id, r.Str()));
        }
    });
}
 
boost::future<std::string> DpmService::GetInternalDaqStatus(const std::string& id) {
    LOG4CPLUS_INFO(m_logger, fmt::format("GetInternalDaqStatus({}): Handling request", id));
    return boost::async(m_executor, [id, this]() mutable -> std::string {
        try {
            auto status = m_scheduler.GetDaqStatus(id);
            nlohmann::json j = status;
            return j.dump(2);
        } catch (daqif::DaqException const& e) {
            error::NestedExceptionReporter r(e);
            LOG4CPLUS_ERROR(m_logger,
                            "GetInternalDaqStatus() failed with following error(s):\n"
                                << r);
            throw;
        } catch (...) {
            error::NestedExceptionReporter r(std::current_exception());
            LOG4CPLUS_ERROR(m_logger,
                            "GetInternalDaqStatus() failed with following error(s):\n"
                                << r);
            BOOST_THROW_EXCEPTION(daqif::DaqException(id, r.Str()));
        }
    });
}
 
boost::future<std::vector<std::shared_ptr<::daqif::DaqStatus>>> DpmService::GetActiveDaqs() {
    LOG4CPLUS_INFO(m_logger, "GetActiveDaqs(): Handling request");
    using V = std::shared_ptr<::daqif::DaqStatus>;
    using R = std::vector<V>;
    return boost::async(m_executor, [this]() mutable -> R {
        try {
            auto ids = m_scheduler.GetQueue();
            R statuses;
            for (auto const& id : ids) {
                statuses.emplace_back(GetDaqStatusSync(id));
            }
            return statuses;
        } catch (daqif::DaqException const& e) {
            error::NestedExceptionReporter r(e);
            LOG4CPLUS_ERROR(m_logger, "GetActiveDaqs() failed with following error(s):\n" << r);
            throw;
        } catch (...) {
            error::NestedExceptionReporter r(std::current_exception());
            LOG4CPLUS_ERROR(m_logger, "GetActiveDaqs() failed with following error(s):\n" << r);
            BOOST_THROW_EXCEPTION(daqif::DaqException("", r.Str()));
        }
    });
}
 
void DpmService::InitiateSignalHandler() {
    m_sigset.async_wait([&](boost::system::error_code const& ec, int sig_num) {
        if (ec) {
            return;
        }
        LOG4CPLUS_INFO(m_logger,
                       fmt::format("Application termination requested with signal {}", sig_num));
        HandleExit();
    });
}
 
void DpmService::HandleExit() {
    LOG4CPLUS_TRACE(m_logger, "HandleExit(): Application termination has been requested");
    // If we still have a work-guard we try a nice shutdown, if there's no work guard
    if (m_work_guard.owns_work()) {
        // Reset work guard and cancel pending operations
        m_work_guard.reset();
        m_scheduler.Stop();
        m_sigset.cancel();
        m_sigset.clear();
    } else {
        LOG4CPLUS_INFO(m_logger,
                       "Application termination already requested before. Stopping immediately!");
        m_executor.get_io_context().stop();
    }
}
 
std::shared_ptr<daqif::DaqStatus> DpmService::GetDaqStatusSync(const std::string& id) {
    LOG4CPLUS_INFO(m_logger, fmt::format("GetDaqStatusSync({}): Handling request", id));
    using V = ::daqif::DaqStatus;
    try {
        auto status = m_scheduler.GetDaqStatus(id);
        auto rep = m_mal.createDataEntity<V>();
        *rep << status;
        return rep;
    } catch (...) {
        error::NestedExceptionReporter r(std::current_exception());
        BOOST_THROW_EXCEPTION(daqif::DaqException(id, r.Str()));
    }
}
 
}  // namespace daq::dpm