rtmsLlnetioPublisher.hpp

Go to the documentation of this file.

 
#ifndef FGF_ANDOR_RTMS_LLNETIO_PUBLISHER_HPP_H_
#define FGF_ANDOR_RTMS_LLNETIO_PUBLISHER_HPP_H_
 
#include <cstdint>
#include <list>
#include <memory>
#include <string>
#include <thread>
#include <condition_variable>
#include <mutex>
#include <vector>
 
#include <boost/atomic.hpp>
#include <boost/lockfree/spsc_queue.hpp>
 
#include <ifw/rtmslib_llnetio/rtmsSender.hpp>
#include <ifw/rtmslib_llnetio/sampleExtInfo.hpp>
#include <ifw/fgf/common_llnetio/logger.hpp>
 
namespace ifw::fgf::andor {
 
    constexpr auto PUBLISHER_STAT_WINDOW_SIZE = 1000;
    constexpr auto PUBLISHER_QUEUE_CAPACITY = 8;

    struct RtmsNetworkConfig {
        std::string                     src_ip;
        std::string                     dest_ip;
        int                             dest_port{0};
        ifw::rtmslib::SenderConfig      sender_cfg{};
    };

    struct PublishFramePacket {
        std::vector<uint8_t> data;
        double timestamp{0};
        int width{0};
        int height{0};
        int bpp{0};
    };

    struct PublisherThreadConfig {
        int cpu_affinity{-1};
        int scheduler_priority{0};
    };

    class RtmsLlnetioPublisher {
    public:
 
        RtmsLlnetioPublisher(const RtmsNetworkConfig& rtms_cfg,
                             bool compute_latency = false,
                             const PublisherThreadConfig& thread_cfg = PublisherThreadConfig());
 
        virtual ~RtmsLlnetioPublisher();

        bool Push(std::shared_ptr<PublishFramePacket> pkt);

        void PushLatencyStart(double ts);

        void Notify();

        void Stop();

        bool IsStopped() const;

        void SetOffsets(int offset_x, int offset_y);
 
        // ---- Latency statistics (runtime toggle)
        void SetComputeLatency(bool enable);
        bool GetComputeLatency() const;
        void ResetLatency();
 
        void GetFirstPacketSentLatency(double& latency, double& stddev,
                                       double& variance, int& nb_of_samples);
        void GetFrameSentLatency(double& latency, double& stddev,
                                 double& variance, int& nb_of_samples);
 
    private:
        void Run();
 
        void SetThreadAffinity(int cpu_id);
        void SetThreadPriority(int priority);
 
        void ComputeStat(const std::list<double>& samples,
                         double& mean, double& stddev,
                         double& variance, int& nb_of_samples) const;

        void LogStatsIfNeeded();
 
        // ---- SPSC queue ----
        boost::lockfree::spsc_queue<std::shared_ptr<PublishFramePacket>,
            boost::lockfree::capacity<PUBLISHER_QUEUE_CAPACITY>> m_queue;
 
        // ---- Publisher thread ----
        std::thread                    m_pub_thread;
        std::condition_variable        m_pub_condvar;
        std::mutex                     m_pub_mtx;
        static std::atomic<bool>      m_stopped;
 
        // ---- Sender ----
        RtmsNetworkConfig              m_rtms_cfg;
        std::unique_ptr<ifw::rtmslib::RtmsSender> m_sender;
        int                            m_active_sender_bpp{0};
        size_t                         m_active_sender_size{0};
 
        // ---- Offsets (set from CommAdapter) ----
        int                            m_offset_x{0};
        int                            m_offset_y{0};
 
        // ---- Logging ----
        log4cplus::Logger              m_logger;
 
        // ---- Latency tracking ----
        bool                           m_compute_latency{false};
        std::mutex                     m_start_times_mtx;
        std::list<double>              m_start_times;
 
        std::mutex                     m_first_pkt_lat_mtx;
        std::list<double>              m_first_pkt_latencies;
 
        std::mutex                     m_frame_lat_mtx;
        std::list<double>              m_frame_latencies;
 
        // ---- Periodic stats logging ----
        std::mutex                     m_stats_mtx;
        int                            m_frames_sent{0};
        int                            m_frames_dropped{0};
        double                         m_last_stats_time{0};
 
        // ---- Thread tuning ----
        PublisherThreadConfig          m_thread_cfg;
    };
 
} // namespace ifw::fgf::andor
 
#endif // FGF_ANDOR_RTMS_LLNETIO_PUBLISHER_HPP_H_