ddsPubThread.hpp

Go to the documentation of this file.

#ifndef RTCTK_REUSABLECOMPONENT_TELREPUB_DDS_PUB_THREAD_HPP
#define RTCTK_REUSABLECOMPONENT_TELREPUB_DDS_PUB_THREAD_HPP
 
#include "agnostictopicif.hpp"
#include "mudpiProcessor.hpp"
#include "queue.hpp"
#include <cstdint>
#include <fastdds/dds/publisher/DataWriter.hpp>
#include <fmt/core.h>
#include <fmt/printf.h>
#include <memory>
#include <rtctk/componentFramework/bufferManager.hpp>
#include <rtctk/componentFramework/componentMetricsIf.hpp>
#include <rtctk/componentFramework/ddsCommon.hpp>
#include <rtctk/componentFramework/durationMonitor.hpp>
#include <rtctk/componentFramework/logger.hpp>
#include <rtctk/telRepub/wrangler.hpp>
#include <tbb/concurrent_queue.h>
 
#include <numapp/thread.hpp>
 
#include <chrono>
#include <iostream>
#include <utility>
 
namespace rtctk::telRepub {
 
using namespace rtctk::componentFramework;

struct CfgPubThread {
    numapp::NumaPolicies thread_policies;
};

struct CfgPubThreadMudpi : CfgPubThread, CfgMudpiProc {
    std::uint32_t queue_size = 200u;
};

class PubThread {
public:
    PubThread(DataWriter& data_writer,
              componentFramework::ServiceContainer& service,
              const numapp::NumaPolicies& thread_policies,
              std::uint32_t sample_id_increment);

    virtual ~PubThread();

    virtual void RunAsync();

    void IdleAsync();
 
    std::string GetTopicName();
 
    virtual bool Monitor();
 
protected:
    enum class State { Idle, Run, Exit };
 
    std::string m_topic_name;
 
    log4cplus::Logger m_logger;
 
    std::atomic<State> m_requested_state;
    std::thread m_thread;
    std::string m_thread_name;
 
    DataWriter& m_data_writer;
 
    ComponentMetricsIf& m_metrics;

    componentFramework::AlertSource m_topic_publish_alert;
 
    componentFramework::AlertSource m_wrangler_alert;

    // @note: OLDB does not support uint64_t which we really want, so we have to use signed 64.
    perfc::CounterI64 m_pc_published_samples;
    perfc::ScopedRegistration m_pc_published_samples_reg;
 
    perfc::CounterI64 m_pc_publication_errors;
    perfc::ScopedRegistration m_pc_publication_errors_reg;
 
    perfc::CounterI64 m_pc_last_sample_id_published;
    perfc::ScopedRegistration m_pc_last_sample_id_published_reg;
 
    std::unique_ptr<DurationMonitor<>> m_dur_mon;
 
    std::uint32_t m_expected_sample_id_increment;
 
    static std::atomic<std::uint16_t> s_count;  // instance index
};  // PubThread

class PubThreadMudpi : public PubThread {
public:
    PubThreadMudpi(DataWriter& data_writer,
                   componentFramework::ServiceContainer& service,
                   CfgPubThreadMudpi& cfg,
                   WranglerFunction&& wrangler);

    virtual ~PubThreadMudpi();

    void RunAsync() override;

    MudpiProcessor& GetMudpiProcessor();
 
    std::shared_ptr<Queue> GetQueue();
 
    bool Monitor() override;
 
private:
    void ProcessingLoop();
 
    std::shared_ptr<Queue> m_frame_queue;
    llnetio::mudpi::TopicId m_mudpi_topic_id;
    MudpiProcessor m_mudpi_processor;
    WranglerFunction m_wrangler;
};

class PubThreadSim : public PubThread {
public:
    PubThreadSim(DataWriter& data_writer,
                 componentFramework::ServiceContainer& service,
                 const numapp::NumaPolicies& thread_policies,
                 std::uint32_t sample_id_increment,
                 std::uint16_t sim_freq,
                 std::uint32_t sim_payload_size);

    virtual ~PubThreadSim();
 
private:
    void SimProcessingLoop();
 
    AgnosticTopic m_sim_msg;
 
    std::uint16_t m_sim_freq;
    std::uint32_t m_sim_payload_size;  // bytes
};
 
}  // namespace rtctk::telRepub
 
#endif  // RTCTK_REUSABLECOMPONENT_TELREPUB_DDS_PUB_THREAD_HPP