dpmClient.cpp

Go to the documentation of this file.

/**
 * @file
 * @ingroup daq_ocm_libdaq_test
 * @copyright 2022 ESO - European Southern Observatory
 *
 * @brief `daq::DpmClient`
 */
#include <daq/dpmClient.hpp>
 
#include <boost/assert.hpp>
 
#include <Daqif.hpp>
#include <log4cplus/loggingmacros.h>
#include <mal/Mal.hpp>
#include <mal/rr/qos/ReplyTime.hpp>
 
#include <daq/conversion.hpp>
#include <daqif/subscription.hpp>
 
namespace daq {
 
using DaqReplyPtr = std::shared_ptr<daqif::DaqReply>;
 
struct DpmClientImpl::Subscriptions {
    std::shared_ptr<bool> alive = std::make_shared<bool>(true);
    daqif::Subscription<daqif::DaqStatus> status;
    daqif::Subscription<daqif::StorageStatus> storage;
};
 
DpmClientImpl::DpmClientImpl(boost::asio::io_context& io_ctx, mal::Mal& mal, DpmClientParams params)
    : m_io_ctx(io_ctx)
    , m_executor(m_io_ctx)
    , m_mal(mal)
    , m_params(std::move(params))
    , m_logger(log4cplus::Logger::getInstance("daq.ocm")) {
    m_dpmif = m_mal.getClient<daqif::DpmDaqControlAsync>(
        daqif::MakeServiceUri(m_params.rr_uri, "daq"),
        {std::make_shared<elt::mal::rr::qos::ReplyTime>(params.timeout)},
        {});
    assert(m_dpmif);
 
    m_subs = std::make_unique<Subscriptions>();
    // Add subscription to status notifications, which will invoked from unknown threads,
    // so we dispatch to asio instead and emit signal safely there.
    m_subs->status = daqif::MakeSubscription<daqif::DaqStatus>(
        m_mal,
        daqif::MakeServiceUri(m_params.ps_uri, "daq/status"),
        [weak = std::weak_ptr<bool>(m_subs->alive), this](auto& sub, auto const& event) {
            auto shared = weak.lock();
            if (!shared) {
                // Abandoned
                return;
            }
            auto const& sample = event.getData();
            Status status(sample->getId(), sample->getFileId());
            status << *sample;
            LOG4CPLUS_TRACE(m_logger, "Received DaqStatus update: " << status);
            // Dispatch to main thread
            m_io_ctx.post([status = std::move(status), weak, this] {
                auto shared = weak.lock();
                if (!shared) {
                    // Abandoned
                    return;
                }
                UpdateStatus(status);
            });
        });
    // Add subscription to storage notifications, which will invoked from unknown threads,
    // so we dispatch to asio instead and emit signal safely there.
    m_subs->storage = daqif::MakeSubscription<daqif::StorageStatus>(
        m_mal,
        daqif::MakeServiceUri(m_params.ps_uri, "dpm/storage"),
        [weak = std::weak_ptr<bool>(m_subs->alive), this](auto& sub, auto const& event) {
            auto shared = weak.lock();
            if (!shared) {
                // Abandoned
                return;
            }
            auto const& sample = event.getData();
            std::filesystem::space_info space = {};
            space << *sample;
            // Dispatch to main thread
            m_io_ctx.post([space, weak, this] {
                auto shared = weak.lock();
                if (!shared) {
                    // Abandoned
                    return;
                }
                m_storage_signal(space);
            });
        });
}
 
DpmClientImpl::~DpmClientImpl() = default;
 
auto DpmClientImpl::ScheduleAsync(std::string const& spec) -> boost::future<State> {
    return m_dpmif->QueueDaq(spec).then(m_executor, [&](boost::future<DaqReplyPtr> f) {
        auto reply = f.get();
        // If it doesn't throw it was successful, until status change is published we can assume
        // state is Scheduled.
        return State::Scheduled;
    });
}
 
auto DpmClientImpl::AbortAsync(std::string const& id) -> boost::future<State> {
    return m_dpmif->AbortDaq(id).then(m_executor, [&](boost::future<DaqReplyPtr> f) {
        auto reply = f.get();
        // If it doesn't throw it was successful, until status change is published we can assume
        // state is Aborted.
        return State::Aborted;
    });
}
 
void DpmClientImpl::StartMonitorStatus(std::string const& id) {
    LOG4CPLUS_TRACE(m_logger, "DpmClientImpl::StartMonitorStatus: " << id);
    RequestStatus(id);
}
 
void DpmClientImpl::RequestStatus(std::string const& id) {
    LOG4CPLUS_TRACE(m_logger, "RequestStatus:" << id);
    m_res_status.AsyncAcquire().then(
        m_executor, [&, id, alive = std::weak_ptr(m_liveness)](boost::future<ResourceToken> f_token) {
            auto is_alive = alive.lock();
            if (!is_alive) {
                LOG4CPLUS_TRACE("daq.ocm", "DpmClient: Client abandoned.");
                return;
            }
            LOG4CPLUS_TRACE(m_logger, "AsyncAcquire: Resource available returning ready token");
            auto token = f_token.get();
            m_dpmif->GetDaqStatus(id).then(
                m_executor,
                [&, id, alive = std::weak_ptr(m_liveness), token = std::move(token)](
                    boost::future<std::shared_ptr<daqif::DaqStatus>> f) {
                    auto is_alive = alive.lock();
                    if (!is_alive) {
                        LOG4CPLUS_TRACE("daq.ocm", "DpmClient: Client abandoned.");
                        return;
                    }
                    try {
                        auto reply = f.get();
                        BOOST_ASSERT_MSG(reply, "MAL returned empty shared pointer!");
                        Status status(reply->getId(), reply->getFileId());
                        status << *reply;
 
                        UpdateStatus(status);
                        // To query indefinitely we can schedule a
                        // new request here. For now we rely on pub/sub
                        // to keep us updated after a successful
                        // alignment.
                    } catch (elt::mal::TimeoutException const&) {
                        // We only retry on timeouts (DPM not running?)
                        LOG4CPLUS_DEBUG(m_logger,
                                        "Timeout from DPM GetDaqStatus() -> reschedule attempt");
                        ScheduleRequestStatus(id);
                    } catch (std::exception const& err) {
                        LOG4CPLUS_ERROR(
                            m_logger,
                            "Failure from GetDaqStatus() -> will not reschedule attempt. Error: "
                                << err.what());
                    }
                });
        });
}
 
void DpmClientImpl::UpdateStatus(Status const& status) {
    m_status_signal(status);
}
 
void DpmClientImpl::ScheduleRequestStatus(std::string const& id) {
    auto it = m_timers.find(id);
    if (it != m_timers.end()) {
        it->second.cancel();
    } else {
        it = m_timers.emplace(id, m_io_ctx).first;
    }
    // Set timer
    it->second.expires_from_now(m_params.status_retry_interval);
    // note: async_wait never invoked immedately but as if by post()
    it->second.async_wait(
        [&, id, alive = std::weak_ptr(m_liveness), this](boost::system::error_code ec) {
            if (ec == boost::asio::error::operation_aborted) {
                return;
            }
            auto is_alive = alive.lock();
            if (!is_alive) {
                LOG4CPLUS_TRACE("daq.ocm", "DpmClient: Client abandoned.");
                return;
            }
            RequestStatus(id);
        });
}
 
void DpmClientImpl::StopMonitorStatus(std::string const& id) {
    m_timers.erase(id);
}
 
auto DpmClientImpl::ConnectStatusSignal(StatusSignal::slot_type const& slot)
    -> boost::signals2::connection {
    return m_status_signal.connect(slot);
}
 
auto DpmClientImpl::ConnectStorageSignal(StorageSignal::slot_type const& slot)
    -> boost::signals2::connection {
    return m_storage_signal.connect(slot);
}
 
}  // namespace daq