ipcqRecordingUnit.hpp

Go to the documentation of this file.

 
#ifndef RTCTK_COMPONENTFRAMEWORK_IPCQRECORDINGUNIT_HPP
#define RTCTK_COMPONENTFRAMEWORK_IPCQRECORDINGUNIT_HPP
 
#include "rtctk/componentFramework/dataRecorder.hpp"
#include "rtctk/componentFramework/exceptions.hpp"
#include "rtctk/componentFramework/recordingInfo.hpp"
#include "rtctk/componentFramework/recordingUnit.hpp"
#include <cstdint>
#include <string>
#include <exception>
#include <mutex>
#include <thread>
 
namespace rtctk::componentFramework {
 
class IpcqError : public RtctkException {
    public:
        using RtctkException::RtctkException;
        explicit IpcqError(const std::string& message) :
            RtctkException("Error in IPCQ: " + message) {}
};
 
template <typename Topic, typename OutputStage = typename RecordingInfo<Topic>::Recorder>
class IpcqRecordingUnit : public RecordingUnit {
public:
    IpcqRecordingUnit(std::string const& comp_id,
                      std::string const& unit_id,
                      RepositoryIf& rtr,
                      RepositoryIf& oldb)
        : RecordingUnit(comp_id, unit_id, rtr, oldb),
            m_queue_name{rtr.GetDataPoint<std::string>(DataPointPath(comp_id) /
                                                       DataPointPath("static") /
                                                       DataPointPath("rec_units") /
                                                       DataPointPath(unit_id) /
                                                       SHM_QUEUE_NAME_PATH)},
            m_output(RecordingInfo<Topic>::COLUMNS, GetDisabled(rtr, DataPointPath(comp_id) /
                                                                     DataPointPath("static") /
                                                                     DataPointPath("rec_units") /
                                                                     DataPointPath(unit_id) /
                                                                     TELEMETRY_SUBSET_PATH))
        {
 
            DataPointPath path = DataPointPath(comp_id) /
                                 DataPointPath("static") /
                                 DataPointPath("rec_units") /
                                 DataPointPath(unit_id) /
                                 CPU_AFFINITY_PATH;
            if(rtr.DataPointExists(path)) {
                m_cpu_affinity = rtr.GetDataPoint<int32_t>(path);
            }
 
            Update();
        }
 
    virtual ~IpcqRecordingUnit() {
        m_stop = true;
        if(m_process_thread.joinable()) {
            m_process_thread.join();
        }
    }
 
    void Prepare(const std::filesystem::path& file_path) override;
    void Start() override;
    std::optional<std::filesystem::path> Stop() override;
    void Update() override;
 
protected:
    static typename OutputStage::DisabledFields GetDisabled(RepositoryIf& rtr, const DataPointPath& sub_path);
 
private:
    void Process();
    std::thread m_process_thread;
    std::atomic<bool> m_start = false;
    std::atomic<bool> m_stop = false;
    std::string m_queue_name;
    std::optional<size_t> m_cpu_affinity;
    OutputStage m_output;
    int64_t m_subsample_factor;
    int64_t m_start_sample_id;
    int64_t m_stop_after_num_samples;
    int64_t m_last_observed_sample_id;
    uint64_t m_sampling_counter;
    inline static const DataPointPath SHM_QUEUE_NAME_PATH = DataPointPath("shm_queue_name");
    inline static const DataPointPath CPU_AFFINITY_PATH = DataPointPath("cpu_affinity");
    inline static const DataPointPath TELEMETRY_SUBSET_PATH = DataPointPath("telemetry_subset");
    inline static const DataPointPath SUBSAMPLE_FACTOR_PATH = DataPointPath("subsample_factor");
    inline static const DataPointPath START_SAMPLE_ID_PATH = DataPointPath("start_at_sample_id");
    inline static const DataPointPath STOP_AFTER_NUM_SAMPLES = DataPointPath("stop_after_num_samples");
};
 
}  // namespace rtctk::componentFramework
 
#include "rtctk/componentFramework/ipcqRecordingUnit.ipp"
 
#endif  // RTCTK_COMPONENTFRAMEWORK_IPCQRECORDINGUNIT_HPP