ipcqRecordingUnit.hpp

Go to the documentation of this file.

 
#ifndef RTCTK_COMPONENTFRAMEWORK_IPCQRECORDINGUNIT_HPP
#define RTCTK_COMPONENTFRAMEWORK_IPCQRECORDINGUNIT_HPP
 
#include <rtctk/componentFramework/bufferMonitor.hpp>
#include <rtctk/componentFramework/componentMetricsIf.hpp>
#include <rtctk/componentFramework/dataRecorder.hpp>
#include <rtctk/componentFramework/durationMonitor.hpp>
#include <rtctk/componentFramework/exceptions.hpp>
#include <rtctk/componentFramework/frequencyEstimator.hpp>
#include <rtctk/componentFramework/recordingInfo.hpp>
#include <rtctk/componentFramework/recordingUnit.hpp>
 
#include <fmt/format.h>
 
#include <cstdint>
#include <string>
#include <thread>
 
namespace rtctk::componentFramework {
 
template <typename T, typename = int>
struct HasSampleIdWithCorrectType : std::false_type {};
 
template <typename T>
struct HasSampleIdWithCorrectType<T, decltype(std::declval<T>().sample_id, 0)>
    : std::conditional<std::is_same<uint32_t, decltype(std::declval<T>().sample_id)>::value,
                       std::true_type,
                       std::false_type>::type {};
// first check if the member exists. If yes, then check if it has the right type
 
template <typename T>
inline constexpr bool HAS_MEMBER_SAMPLE_ID_UINT32T = HasSampleIdWithCorrectType<T>::value;
 
template <typename RecInfoType>
class IpcqRecordingUnit : public RecordingUnit {
public:
    static_assert(HAS_MEMBER_SAMPLE_ID_UINT32T<typename RecInfoType::Topic>,
                  "Topic struct must contain member 'sample_id' with type 'uint32_t'!");
 
    IpcqRecordingUnit(const std::string& comp_id,
                      const std::string& unit_id,
                      ServiceContainer& services)
        : RecordingUnit(comp_id, unit_id, "IPCQ", services)
        , m_queue_name(m_rtr.GetDataPoint<std::string>(
              DataPointPath(fmt::format(RTR_PATH_QUEUE_NAME, comp_id, unit_id))))
        , m_output(RecInfoType::COLUMNS) {
        auto queue_name_path = DataPointPath(fmt::format(OLDB_PATH_QUEUE_NAME, comp_id, unit_id));
        m_oldb.CreateDataPoint<std::string>(queue_name_path);
        m_oldb.SetDataPoint<std::string>(queue_name_path, m_queue_name);
 
        auto cpu_affinity_path =
            DataPointPath(fmt::format(RTR_PATH_CPU_AFFINITY, comp_id, unit_id));
        if (m_rtr.DataPointExists(cpu_affinity_path)) {
            m_cpu_affinity = m_rtr.GetDataPoint<int32_t>(cpu_affinity_path);
        }
 
        m_samples_written_reg = m_metrics.AddCounter(
            &m_samples_written,
            CounterMetricInfo(unit_id + "/samples_written",
                              "Number of samples successfully written to file"));
        m_last_observed_sample_id_reg =
            m_metrics.AddCounter(&m_last_observed_sample_id,
                                 CounterMetricInfo(unit_id + "/last_sample_id_written",
                                                   "Last sample id successfully written to file"));
 
        m_freq_estimator = std::make_unique<FrequencyEstimator<>>(
            m_metrics, "Estimated frequency of the data writer", unit_id);
 
        m_dur_monitor = std::make_unique<DurationMonitor<>>(
            m_metrics, "Duration of writing data to file", unit_id);
 
        m_buffer_monitor =
            std::make_unique<BufferMonitor<>>(m_metrics, "SHM read buffer occupancy", unit_id);
 
        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::vector<std::filesystem::path> Stop() override;
    void Update() override;
 
protected:
    static typename RecInfoType::Recorder::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;
    typename RecInfoType::Recorder m_output;
    int64_t m_subsample_factor;
    int64_t m_start_sample_id;
    int64_t m_stop_after_num_samples;
    uint64_t m_sampling_counter;
    perfc::CounterI64 m_samples_written;
    perfc::ScopedRegistration m_samples_written_reg;
    perfc::CounterI64 m_last_observed_sample_id;
    perfc::ScopedRegistration m_last_observed_sample_id_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 std::string_view RTR_PATH_QUEUE_NAME =
        "/{}/static/rec_units/{}/shm_queue_name";
    inline static constexpr std::string_view RTR_PATH_CPU_AFFINITY =
        "/{}/static/rec_units/{}/cpu_affinity";
    inline static constexpr std::string_view RTR_PATH_TELEMETRY_SUBSET =
        "/{}/dynamic/rec_units/{}/telemetry_subset";
    inline static constexpr std::string_view RTR_PATH_SUBSAMPLE_FACTOR =
        "/{}/dynamic/rec_units/{}/subsample_factor";
    inline static constexpr std::string_view RTR_PATH_START_SAMPLE_ID =
        "/{}/dynamic/rec_units/{}/start_at_sample_id";
    inline static constexpr std::string_view RTR_PATH_STOP_AFTER_NUM_SAMPLES =
        "/{}/dynamic/rec_units/{}/stop_after_num_samples";
    inline static constexpr std::string_view OLDB_PATH_QUEUE_NAME = "/{}/rec_units/{}/queue_name";
 
    inline static constexpr size_t MAX_SAMPLES_READ = 16;
};
 
}  // namespace rtctk::componentFramework
 
#include <rtctk/componentFramework/ipcqRecordingUnit.ipp>
 
#endif  // RTCTK_COMPONENTFRAMEWORK_IPCQRECORDINGUNIT_HPP