repositoryRecordingUnit.hpp

Go to the documentation of this file.

#ifndef RTCTK_COMPONENTFRAMEWORK_REPOSITORYRECORDINGUNIT_HPP
#define RTCTK_COMPONENTFRAMEWORK_REPOSITORYRECORDINGUNIT_HPP
 
#include <algorithm>
#include <rtctk/componentFramework/componentMetricsIf.hpp>
#include <rtctk/componentFramework/dataPointPath.hpp>
#include <rtctk/componentFramework/dataRecorder.hpp>
#include <rtctk/componentFramework/exceptions.hpp>
#include <rtctk/componentFramework/fitsDataRecorder.hpp>
#include <rtctk/componentFramework/logger.hpp>
#include <rtctk/componentFramework/recordingInfo.hpp>
#include <rtctk/componentFramework/recordingUnit.hpp>
#include <rtctk/componentFramework/serviceContainer.hpp>
#include <rtctk/componentFramework/timeString.hpp>
 
#include <numapp/numapolicies.hpp>
#include <numapp/thread.hpp>
#include <taiclock/taiClock.hpp>
 
#include <boost/asio.hpp>
#include <fmt/format.h>
 
#include <chrono>
#include <cstdint>
#include <fstream>
#include <queue>
#include <semaphore>
#include <stop_token>
#include <string>
#include <thread>
 
namespace rtctk::componentFramework {
 
using namespace std::string_view_literals;
using namespace std::chrono_literals;

class RepositoryRecordingUnit : public RecordingUnit {
public:
    RepositoryRecordingUnit(const std::string& comp_id,
                            const std::string& unit_id,
                            ServiceContainer& services);

    ~RepositoryRecordingUnit() override;

    void Prepare(const std::filesystem::path& file_path) override;

    void Start() override;

    std::vector<std::filesystem::path> Stop() override;
 
private:
    struct DataPointCfg {
        struct MetaDataCfg {
            bool enabled = false;
            std::vector<std::string> filter = {};
        };
 
        struct CaptureCfg {
            bool on_start = true;
            bool on_stop = true;
            bool on_change = true;
        };
 
        enum class DataSource : uint8_t {
            RTR,
            OLDB,
        };

        bool enabled = true;

        DataPointPath name;

        DataSource source = DataSource::RTR;
 
        // settings related to metadata
        MetaDataCfg metadata;
 
        // settings related to when to record the data
        CaptureCfg capture;
    };

 
    class MetaDataQueue {
    public:
        /*
         * Pushes an item to the queue
         */
        void Push(const RepositoryIf::MetaData& data);
        /*
         * Retrieves an item from the queue, returns std::nullopt if empty
         */
        std::optional<RepositoryIf::MetaData> TryPop();
        /*
         * Gets number of items currently in queue
         */
        size_t Size();
 
    private:
        std::mutex m_mutex;
        std::queue<RepositoryIf::MetaData> m_queue;
    };


    class DataPointRecorderIf {
    public:
        virtual ~DataPointRecorderIf() = default;
        virtual void Prepare(std::filesystem::path file_path) = 0;
        virtual void Start() = 0;
        virtual std::vector<std::filesystem::path> Stop() = 0;
        virtual size_t HasWork() = 0;
        virtual size_t Work() = 0;
    };


    template <typename DpType>
    class DataPointRecorder : public DataPointRecorderIf {
    public:
        using TimePointType = taiclock::TaiClock::time_point::rep;
        using SequenceIdType = uint64_t;
        using ValueType = std::conditional_t<IS_SPAN_CONVERTIBLE<DpType>, AsSpan<DpType>, DpType>;
        using FitsRecorderType = FitsRecorder<TimePointType, SequenceIdType, ValueType>;
 
        DataPointRecorder(RepositorySubscriberIf& repo, const DataPointCfg& cfg)
            : m_logger(GetLogger("rtctk"))
            , m_repo(repo)
            , m_cfg(cfg)
            , m_fits_writer(COLUMNS)
            , m_subscription({}) {
        }
 
        void Prepare(std::filesystem::path file_path) override {
            // set the column length to avoid dynamic-sized fits binary table cells
            m_fits_writer.SetColumnLength(2, m_repo.GetDataPointSize(m_cfg.name));
 
            // set the disabled field depending on the data source
            if (m_cfg.source == DataPointCfg::DataSource::RTR) {
                m_fits_writer.SetDisabledFields({false, false, false});
            } else {
                // no sequence_id field for oldb recording
                m_fits_writer.SetDisabledFields({false, true, false});
            }
 
            // compute the file names
            std::string filename = m_cfg.name.ToString();
            if (filename.starts_with('/')) {
                filename.erase(0, 1);
            }
            std::ranges::replace(filename, '/', '_');
 
            m_pathname_fits = file_path / (filename + m_fits_writer.DefaultFileExtension());
            m_pathname_jsonl = file_path / (filename + ".jsonl");
 
            // open the files
            m_fits_writer.Open(m_pathname_fits);
            if (m_cfg.metadata.enabled) {
                m_md_writer.open(m_pathname_jsonl, std::ios::out);
            }
        }
 
        void Start() override {
            // subscribe if necessary
            if (m_cfg.capture.on_change) {
                m_subscription = std::move(m_repo.Subscribe(
                    m_cfg.name,
                    [this](const auto& path, const auto& metadata) { m_queue.Push(metadata); },
                    nullptr));
            }
 
            // capture data on_start
            if (m_cfg.capture.on_start) {
                RetrieveAndRecord(std::nullopt);
            }
        }
 
        std::vector<std::filesystem::path> Stop() override {
            if (m_cfg.capture.on_change) {
                m_subscription.Unsubscribe();
            }
 
            if (m_cfg.capture.on_stop) {
                RetrieveAndRecord(std::nullopt);
            }
 
            m_fits_writer.Close();
            if (m_cfg.metadata.enabled) {
                m_md_writer.close();
            }
 
            std::vector<std::filesystem::path> files;
            files.push_back(m_pathname_fits);
            if (m_cfg.metadata.enabled) {
                files.push_back(m_pathname_jsonl);
            }
            return files;
        }
 
        size_t HasWork() override {
            return m_queue.Size();
        }
 
        size_t Work() override {
            size_t work_counter = 0;
            for (;;) {
                if (auto md = m_queue.TryPop(); md.has_value()) {
                    RetrieveAndRecord(md);
                    work_counter++;
                } else {
                    break;
                }
            }
            return work_counter;
        }
 
    private:
        void RetrieveAndRecord(const std::optional<RepositoryIf::MetaData>& md_cb) {
            DpType value;
 
            RepositoryIf::MetaData md_read;
            RepositoryIf::BatchRequest req;
            req.ReadDataPoint(m_cfg.name, value, md_read);
            m_repo.SendRequest(req).Wait();
 
            auto ts = md_read["timestamp"].Cast<RepositoryIf::Timestamp>();
            SequenceIdType sequence_id = 0;
 
            bool write_sample = true;
 
            if (m_cfg.source == DataPointCfg::DataSource::RTR) {
                sequence_id = md_read["sequence_id"].Cast<RtcUInt64>();
                if (md_cb.has_value()) {
                    auto sequence_id_cb = md_cb.value()["sequence_id"].Cast<RtcUInt64>();
                    if (sequence_id_cb != sequence_id) {
                        write_sample = false;
                        LOG4CPLUS_WARN(
                            m_logger,
                            fmt::format(
                                "DataPointRecorder: pathname: '{}' missed sequence_id: '{}'.",
                                m_cfg.name,
                                sequence_id_cb));
                    }
                }
            }
 
            if (write_sample) {
                m_fits_writer.Write(AsTuple(ts, sequence_id, value));
 
                if (m_cfg.metadata.enabled) {
                    JsonPayload md_in = md_read;
                    if (m_cfg.metadata.filter.empty()) {
                        m_md_writer << md_in << "\n";
                    } else {
                        JsonPayload md_out;
                        for (const auto& filter : m_cfg.metadata.filter) {
                            if (md_in.contains(filter)) {
                                md_out[filter] = md_in.at(filter);
                            }
                        }
                        m_md_writer << md_out << "\n";
                    }
                }
            }
        }
 
        static FitsRecorderType::TupleType AsTuple(const taiclock::TaiClock::time_point& ts,
                                                   uint64_t sequence_id,
                                                   const DpType& data) {
            auto t = ts.time_since_epoch().count();
            if constexpr (IS_SPAN_CONVERTIBLE<DpType>) {
                return std::make_tuple(t, sequence_id, ToSpan(data));
            } else {
                return std::make_tuple(t, sequence_id, data);
            }
        }
 
        static constexpr typename FitsRecorderType::ColumnDescription COLUMNS =
            typename FitsRecorderType::ColumnDescription{
                {{"timestamp"sv, ""sv}, {"sequence_id"sv, ""sv}, {"payload"sv, ""sv}}};
 
        log4cplus::Logger& m_logger;
        RepositorySubscriberIf& m_repo;
        DataPointCfg m_cfg;
        MetaDataQueue m_queue;
        // TODO: consider adding a mutex to protect the file writers
        FitsRecorderType m_fits_writer;
        std::ofstream m_md_writer;
        RepositorySubscriberIf::Subscription m_subscription;
        std::filesystem::path m_pathname_fits;
        std::filesystem::path m_pathname_jsonl;
    };

 
    void ReadConfigAndCreateDataPointRecorders();
 
    std::unique_ptr<DataPointRecorderIf>
    MakeDataPointRecorder(RepositorySubscriberIf& repo, const DataPointCfg& dp_cfg);
 
    void Process(const std::stop_token& st, const std::filesystem::path& base_path);
 
    using TypeMapType = std::map<std::type_index,
                                 std::function<std::unique_ptr<DataPointRecorderIf>(
                                     RepositorySubscriberIf&, const DataPointCfg&)>>;
 
    template <typename T>
    static TypeMapType::value_type MakeRecordingUnitTypeMapEntry() {
        return std::make_pair(std::type_index{typeid(T)},
                              [](RepositorySubscriberIf& repo, const DataPointCfg& cfg) {
                                  return std::make_unique<DataPointRecorder<T>>(repo, cfg);
                              });
    }
 
    static const TypeMapType TYPE_MAP;
 
    inline static constexpr std::string_view RTR_PATH_NUM_WORKERS =
        "/{}/dynamic/rec_units/{}/num_workers";
 
    inline static constexpr std::string_view RTR_PATH_DATA_POINT_LIST =
        "/{}/dynamic/rec_units/{}/datapoint_list";
 
    inline static constexpr size_t DEFAULT_NUM_WORKERS = 3;
 
    log4cplus::Logger& m_logger;
    size_t m_num_workers;
    std::vector<std::unique_ptr<DataPointRecorderIf>> m_recorders;
    perfc::CounterI64 m_samples_queued;
    perfc::ScopedRegistration m_samples_queued_reg;
    perfc::CounterI64 m_samples_written;
    perfc::ScopedRegistration m_samples_written_reg;
    std::mutex m_mutex_files;
    std::vector<std::filesystem::path> m_files;
    std::binary_semaphore m_sem_start;
    std::jthread m_process_thread;
};
 
}  // namespace rtctk::componentFramework
 
#endif  // RTCTK_COMPONENTFRAMEWORK_DATAPOINTRECORDINGUNIT_HPP