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 <atomic>
#include <cstdint>
#include <filesystem>
#include <optional>
#include <string>
#include <thread>
#include <type_traits>
 
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_t<std::is_same_v<uint32_t, decltype(std::declval<T>().sample_id)>,
                         std::true_type,
                         std::false_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);

    ~IpcqRecordingUnit() override;

    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 LoadDynamicConfig();
    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 = -1;
    int64_t m_start_sample_id = -1;
    int64_t m_stop_after_num_samples = -1;
    uint64_t m_sampling_counter = 0;
    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