ipcqDdtForwarder.hpp

Go to the documentation of this file.

 
#ifndef IPCQ_DDT_FORWARDER_HPP
#define IPCQ_DDT_FORWARDER_HPP
 
#include <rtctk/ddtServer/ddtForwarder.hpp>
 
#include <rtctk/componentFramework/bufferMonitor.hpp>
#include <rtctk/componentFramework/componentMetricsIf.hpp>
#include <rtctk/componentFramework/durationMonitor.hpp>
#include <rtctk/componentFramework/exceptions.hpp>
#include <rtctk/componentFramework/frequencyEstimator.hpp>
#include <rtctk/componentFramework/oldbIf.hpp>
#include <rtctk/componentFramework/repositoryIfUtils.hpp>
#include <rtctk/componentFramework/runtimeRepoIf.hpp>
 
#include <ipcq/adapter.hpp>
#include <ipcq/reader.hpp>
 
#include <numapp/numapolicies.hpp>
#include <numapp/thread.hpp>
 
#include <fmt/format.h>
#include <gsl/span>
 
#include <atomic>
#include <memory>
#include <shared_mutex>
#include <string_view>
 
namespace rtctk::ddtServer {
 
namespace {
 
// Helper type used to construct the tuple of DDT Publishers
 
template <typename Tuple>
struct Wrapper;
 
template <typename... PublisherType>
struct Wrapper<std::tuple<PublisherType...>> {
    using ServiceContainer = rtctk::componentFramework::ServiceContainer;
 
    static std::tuple<PublisherType...>
    MakePublishers(const std::string& db_prefix, ServiceContainer& services) {
        return std::make_tuple(PublisherType(db_prefix, services)...);
    }
};
 
}  // namespace

template <typename FwdInfo, typename ReaderType = ipcq::Reader<typename FwdInfo::Topic>>
class IpcqDdtForwarder : public DdtForwarder {
public:
    using Topic = typename FwdInfo::Topic;
    using ServiceContainer = rtctk::componentFramework::ServiceContainer;
    using RuntimeRepoIf = rtctk::componentFramework::RuntimeRepoIf;
    using OldbIf = rtctk::componentFramework::OldbIf;
    using ComponentMetricsIf = rtctk::componentFramework::ComponentMetricsIf;
    using FrequencyEstimator = rtctk::componentFramework::FrequencyEstimator<>;
    using DurationMonitor = rtctk::componentFramework::DurationMonitor<>;
    using BufferMonitor = rtctk::componentFramework::BufferMonitor<>;
 
    static_assert(FwdInfo::ID.find_first_not_of("abcdefghijklmnopqrstuvwxyz_0123456789") ==
                      std::string_view::npos,
                  "DDT Forwarder ID contains illegal characters!");
 
    IpcqDdtForwarder(const std::string& comp_id, ServiceContainer& services)
        : DdtForwarder(comp_id, "IpcqDdtForwarder", std::string(FwdInfo::ID), services)
        , m_rtr(services.Get<RuntimeRepoIf>())
        , m_oldb(services.Get<OldbIf>())
        , m_metrics(services.Get<ComponentMetricsIf>())
        , m_comp_id(comp_id)
        , m_oldb_prefix(fmt::format("/forwarders/{}", GetId()))  // counters prepend comp name
        , m_rtr_prefix_static(fmt::format("/{}/static/forwarders/{}", comp_id, GetId()))
        , m_rtr_prefix_dynamic(fmt::format("/{}/dynamic/forwarders/{}", comp_id, GetId()))
        , m_publishers(Wrapper<typename FwdInfo::DdtPublisherTupleType>::MakePublishers(
              fmt::format("{}/forwarders/{}/publishers", comp_id, GetId()), services))
        , m_requested_state(State::STOPPED) {
        using namespace rtctk::componentFramework;
 
        // reader monitoring
        const InfluxTagMap base_tags = {
            {"forwarder_id", GetId()},
        };
 
        m_samples_forwarded_reg = m_metrics.AddCounter(
            &m_samples_forwarded, CounterMetricInfo(
                m_oldb_prefix + "/samples_forwarded",
                "Number of samples forwarded",
                base_tags,
                "samples_forwarded"
            )
        );
 
        m_last_sample_id_forwarded_reg = m_metrics.AddCounter(
            &m_last_sample_id_forwarded, CounterMetricInfo(
                m_oldb_prefix + "/last_sample_id_forwarded",
                "Last sample id forwarded",
                base_tags,
                "last_sample_id_forwarded"
            )
        );
 
        m_freq_estimator = std::make_unique<FrequencyEstimator>(
            m_metrics, "Estimated frequency of the data forwarder", m_oldb_prefix);
 
        m_dur_monitor = std::make_unique<DurationMonitor>(
            m_metrics, "Duration of publishing data to DDT", m_oldb_prefix);
 
        m_buffer_monitor =
            std::make_unique<BufferMonitor>(m_metrics, "SHM read buffer occupancy", m_oldb_prefix);
 
        // reader configuration
 
        auto queue_name_path = DataPointPath{m_rtr_prefix_static + "/shm_queue_name"};
        m_queue_name = m_rtr.GetDataPoint<std::string>(queue_name_path);
 
        auto thread_policies_path = DataPointPath{m_rtr_prefix_static + "/thread_policies"};
        m_thread_policies = GetNumaPolicies(m_rtr, thread_policies_path);
 
        auto subsample_factor_path = DataPointPath{m_rtr_prefix_dynamic + "/subsample_factor"};
        m_subsample_factor = m_rtr.GetDataPoint<int64_t>(subsample_factor_path);
    }
 
    ~IpcqDdtForwarder() override {
        m_requested_state = State::STOPPED;
        if (m_processing_thread.joinable()) {
            m_processing_thread.join();
        }
    }
 
    void Start() override {
        AssertState({State::STOPPED});
        m_requested_state = State::STARTING;
        SetState(State::STARTING);
 
        // we trim the thread name to not cause an exception from numapp
        m_processing_thread = numapp::MakeThread(GetId().substr(0, 15),
                                                 m_thread_policies.value_or(numapp::NumaPolicies()),
                                                 [&]() { Process(); });
 
        while (true) {
            using namespace std::chrono_literals;
            std::this_thread::sleep_for(10ms);
 
            CheckErrors();
 
            if (GetState() != State::STARTING) {
                break;
            }
        }
    }
 
    void Run() override {
        AssertState({State::STARTING, State::IDLE});
        m_requested_state = State::RUNNING;
    }
 
    void Idle() override {
        AssertState({State::RUNNING});
        m_requested_state = State::IDLE;
    }
 
    void Stop() override {
        m_requested_state = State::STOPPED;
        if (m_processing_thread.joinable()) {
            m_processing_thread.join();
        }
    }
 
    void Recover() override {
        Stop();
 
        // clear stored exception
        {
            std::scoped_lock lock(m_exception_mutex);
            m_exception = std::current_exception();
        }
 
        // make sure that we get out of error state
        SetState(State::STOPPED);
    }
 
    void Update() override {
        using namespace rtctk::componentFramework;
 
        LOG4CPLUS_INFO(m_logger, fmt::format("Updating IpcqDdtForwarder '{}'", GetId()));
 
        auto subsample_factor_path = DataPointPath{m_rtr_prefix_dynamic + "/subsample_factor"};
        m_subsample_factor = m_rtr.GetDataPoint<int64_t>(subsample_factor_path);
 
        std::apply([&](auto&&... pub) { ((pub.Update()), ...); }, m_publishers);
    }
 
    void CheckErrors() override {
        // cppcheck-suppress unreadVariable
        auto lock = std::shared_lock{m_exception_mutex};
        if (m_exception) {
            std::rethrow_exception(m_exception);
        }
    }
 
private:
    void Process() {
        using namespace std::chrono_literals;
        using namespace rtctk::componentFramework;
 
        try {
            const std::error_code ok{};
            std::pair<std::error_code, size_t> result;
 
            std::vector<Topic> read_buffer;
            read_buffer.reserve(MAX_SAMPLES_READ);
 
            m_samples_forwarded.Store(0);
            m_last_sample_id_forwarded.Store(0);
 
            ReaderType reader(m_queue_name.c_str());
 
            size_t reader_capacity = reader.Capacity();
            size_t to_skip = 0;
            size_t to_read = 0;
 
            SetState(State::IDLE);
 
            while (m_requested_state != State::STOPPED) {
                switch (GetState()) {
                case State::IDLE:
                    if (m_requested_state == State::RUNNING) {
                        if (auto ret = reader.Reset(); ret == ok) {
                            SetState(State::RUNNING);
                        } else {
                            CII_THROW(IpcqError, "Error resetting ipcq Reader");
                        }
                        break;
                    }
                    std::this_thread::sleep_for(1ms);
                    break;
 
                case State::RUNNING:
                    if (m_requested_state == State::IDLE) {
                        SetState(State::IDLE);
                        break;
                    }
                    to_skip = 0;
                    to_read = reader.NumAvailable();
                    if (to_read == 0) {  // try to always read at least one
                        to_read = 1;
                    }
                    to_read = std::min(to_read, MAX_SAMPLES_READ);
 
                    if (m_subsample_factor > 1) {
                        m_sampling_counter = m_sampling_counter % m_subsample_factor;
                        if (m_sampling_counter == 0) {
                            to_read = 1;
                        } else {
                            to_skip = std::min(to_read, m_subsample_factor - m_sampling_counter);
                        }
                    }
 
                    if (to_skip == 0) {
                        m_buffer_monitor->Tick(reader.NumAvailable(), reader_capacity);
                        result = reader.Read(ipcq::BackInserter(read_buffer), to_read, 100ms);
                        for (const auto& sample : read_buffer) {
                            m_last_sample_id_forwarded.Store(sample.sample_id);
                            auto t1 = std::chrono::steady_clock::now();
                            ExtractAndPublish(sample);
                            auto t2 = std::chrono::steady_clock::now();
                            m_dur_monitor->Tick(t2 - t1);
                            m_freq_estimator->Tick();
                            m_samples_forwarded++;
                        }
                        read_buffer.clear();
                    } else {
                        result = reader.Skip(to_skip, 100ms);
                    }
                    m_sampling_counter += result.second;
                    // check for error
                    if (not(result.first == ok or result.first == ipcq::Error::Timeout)) {
                        std::string error = "Error reading from ipcq: " + result.first.message();
                        CII_THROW(IpcqError, error);
                    }
                    break;
 
                default:
                    std::this_thread::sleep_for(1ms);
                }
            }
 
            SetState(State::STOPPED);
 
        } catch (...) {
            {
                std::scoped_lock lock(m_exception_mutex);
                m_exception = std::current_exception();
            }
            SetState(State::ERROR);
        }
    }
 
    void ExtractAndPublish(const Topic& sample) {
        auto func = [&](auto& pub) {
            if (pub.IsEnabled()) {
                // invoke user function that extracts data from SHM Topic sample
                auto e = std::remove_reference_t<decltype(pub)>::StreamInfo::Extract(sample);
                // publish extracted data to DDT
                pub.Publish(std::get<0>(e),
                            reinterpret_cast<const uint8_t*>(std::get<1>(e).data()),
                            std::get<1>(e).size());
            }
        };
 
        std::apply([&](auto&&... pub) { (func(pub), ...); }, m_publishers);
    }
 
    RuntimeRepoIf& m_rtr;
    OldbIf& m_oldb;
    ComponentMetricsIf& m_metrics;
 
    std::string m_comp_id;
    std::string m_oldb_prefix;
    std::string m_rtr_prefix_static;
    std::string m_rtr_prefix_dynamic;
 
    typename FwdInfo::DdtPublisherTupleType m_publishers;
 
    std::atomic<State> m_requested_state;
    std::exception_ptr m_exception = nullptr;
    std::shared_mutex m_exception_mutex;
 
    std::thread m_processing_thread;

    std::string m_queue_name;

    std::optional<numapp::NumaPolicies> m_thread_policies;

    int64_t m_subsample_factor;

    uint64_t m_sampling_counter;

    perfc::CounterI64 m_samples_forwarded;
    perfc::ScopedRegistration m_samples_forwarded_reg;

    perfc::CounterI64 m_last_sample_id_forwarded;
    perfc::ScopedRegistration m_last_sample_id_forwarded_reg;
 
    std::unique_ptr<FrequencyEstimator> m_freq_estimator;
    std::unique_ptr<DurationMonitor> m_dur_monitor;
    std::unique_ptr<BufferMonitor> m_buffer_monitor;

    inline static constexpr size_t MAX_SAMPLES_READ = 16;
};
 
}  // namespace rtctk::ddtServer
 
#endif  // IPCQ_DDT_FORWARDER