dataPointRecordingUnit.hpp

Go to the documentation of this file.

#ifndef RTCTK_COMPONENTFRAMEWORK_DATAPOINTRECORDINGUNIT_HPP
#define RTCTK_COMPONENTFRAMEWORK_DATAPOINTRECORDINGUNIT_HPP
 
#include <memory>
#include <rtctk/componentFramework/componentMetricsIf.hpp>
#include <rtctk/componentFramework/dataRecorder.hpp>
#include <rtctk/componentFramework/exceptions.hpp>
#include <rtctk/componentFramework/fitsDataRecorder.hpp>
#include <rtctk/componentFramework/recordingInfo.hpp>
#include <rtctk/componentFramework/recordingUnit.hpp>
#include <rtctk/componentFramework/serviceContainer.hpp>
#include <taiclock/taiClock.hpp>
 
#include <fmt/format.h>
#include <numapp/numapolicies.hpp>
#include <numapp/thread.hpp>
 
#include <cstdint>
#include <exception>
#include <string>
#include <string_view>
#include <thread>
#include <typeinfo>
#include <vector>
 
namespace rtctk::componentFramework {
 
using namespace std::string_view_literals;
 
enum class DataPointRecordingSource : uint8_t { OLDB, RUNTIMEREPO };

template <typename DpType>
class [[deprecated("Use RepositoryRecordingUnit instead.")]]DataPointRecordingUnit : public RecordingUnit {
public:
    // we are using a bit-mask to configure when to capture data
    using CaptureMask = uint32_t;
    static constexpr uint32_t CAPTURE_ON_START = 1;
    static constexpr uint32_t CAPTURE_ON_STOP = 2;
    static constexpr uint32_t CAPTURE_ON_CHANGE = 4;
 
    using TimepointType = taiclock::TaiClock::time_point::rep;
 
    using OutputDpType = std::conditional_t<IS_SPAN_CONVERTIBLE<DpType>, AsSpan<DpType>, DpType>;
    using OutputStageType = DataRecorder<TimepointType, OutputDpType>;

    DataPointRecordingUnit(const std::string& comp_id,
                           const std::string& unit_id,
                           ServiceContainer& services,
                           const DataPointPath& dp_path,
                           DataPointRecordingSource source = DataPointRecordingSource::RUNTIMEREPO,
                           std::unique_ptr<OutputStageType>&& output_stage = {},
                           std::optional<size_t> fixed_recording_length = std::nullopt)
        : RecordingUnit(comp_id, unit_id, "DataPoint", services)
        , m_dp_path{dp_path}
        , m_recording_source{source}
        , m_dp_buffer()
        , m_output{std::move(output_stage)}
        , m_recording_length{fixed_recording_length} {
        if (m_output == nullptr) {
            m_output = std::make_unique<FitsRecorder<TimepointType, OutputDpType>>(COLUMNS);
        }
 
        if (m_recording_source == DataPointRecordingSource::OLDB) {
            m_repository = static_cast<RepositoryIf*>(&m_oldb);
            m_repository_subscriber = static_cast<RepositorySubscriberIf*>(&m_oldb);
        } else if (m_recording_source == DataPointRecordingSource::RUNTIMEREPO) {
            m_repository = static_cast<RepositoryIf*>(&m_rtr);
            m_repository_subscriber = static_cast<RepositorySubscriberIf*>(&m_rtr);
        } else {
            CII_THROW(InvalidSetting, "Invalid recording source");
        }
        if (not m_repository or not m_repository_subscriber) {
            CII_THROW(InvalidSetting, "Invalid recording source");
        }
 
        auto input_dp_path = DataPointPath(fmt::format(OLDB_PATH_DP_NAME, comp_id, unit_id));
        if (not m_oldb.DataPointExists(input_dp_path)) {
            m_oldb.CreateDataPoint<std::string>(input_dp_path);
        }
        m_oldb.SetDataPoint<std::string>(input_dp_path, dp_path);
 
        auto capture_mask_path =
            DataPointPath(fmt::format(RTR_PATH_CAPTURE_MASK, comp_id, unit_id));
        if (m_rtr.DataPointExists(capture_mask_path)) {
            m_capture_mask = m_rtr.GetDataPoint<int32_t>(capture_mask_path);
        } else {
            m_capture_mask = CAPTURE_ON_START + CAPTURE_ON_CHANGE + CAPTURE_ON_STOP;
        }
 
        const InfluxTagMap base_tags = {
            {"unit_id", unit_id}
        };
 
        m_samples_written_reg = m_metrics.AddCounter(
            &m_samples_written, CounterMetricInfo(
                unit_id + "/samples_written",
                "Samples written",
                base_tags,
                "samples_written"
            )
        );
    }
 
    ~DataPointRecordingUnit() override {
        m_stop = true;
        if (m_process_thread.joinable()) {
            m_process_thread.join();
        }
        SetStopped();
    }

    void Prepare(const std::filesystem::path& file_path) override {
        if(not IsEnabled()) {
            return;
        }
 
        SetState(State::PREPARING,
                 State::STOPPED,
                 "Error in transition to PREPARING, DataPointRecordingUnit not in STOPPED State");
 
        m_file_path = file_path / (GetId() + m_output->DefaultFileExtension());
 
        if (not m_repository->DataPointExists(m_dp_path)) {
            SetFailed(nullptr);
            CII_THROW(InvalidSetting, "DataPoint does not exist");
        }
 
        if (m_capture_mask == 0) {
            SetFailed(nullptr);
            CII_THROW(InvalidSetting, "Invalid capture mask 0");
        }
 
        using FitsRecorderType = FitsRecorder<TimepointType, OutputDpType>;
        FitsRecorderType* fits_output = dynamic_cast<FitsRecorderType*>(m_output.get());
        if (fits_output != nullptr) {
            auto rec_length =
                m_recording_length.value_or(m_repository->GetDataPointSize(m_dp_path));
            fits_output->SetColumnLength(1, rec_length);
        }
 
        auto policies = numapp::NumaPolicies();
        m_process_thread =
            numapp::MakeThread(m_unit_id.substr(0, 15), policies, [&]() { return Process(); });
    }

    void Start() override {
        m_start = true;
    }

    std::vector<std::filesystem::path> Stop() override {
        m_stop = true;
        if (m_process_thread.joinable()) {
            m_process_thread.join();
        }
        std::vector<std::filesystem::path> files;
        if (m_file_path) {
            files.push_back(*m_file_path);
        }
        m_file_path.reset();
        SetStopped();
        return files;
    }
 
    static constexpr typename OutputStageType::ColumnDescription COLUMNS =
        typename OutputStageType::ColumnDescription{{{"timestamp"sv, ""sv}, {"payload"sv, ""sv}}};
 
private:
    void Process() {
        using namespace std::chrono_literals;
 
        try {
            m_output->Open(*m_file_path);
            m_samples_written.Store(0);
            RepositorySubscriberIf::Subscription subscription{{}};
 
            if (m_capture_mask & CAPTURE_ON_CHANGE) {
                subscription = std::move(m_repository_subscriber->Subscribe<DpType>(
                    m_dp_path,
                    [this](auto& path, auto& value, auto& metadata) {
                        if (GetState() == State::RUNNING) {
                            try {
                                m_output->Write(AsTuple(value));
                                m_samples_written++;
                            } catch (...) {
                                SetFailed(std::current_exception());
                                return;
                            }
                        }
                    },
                    // NOLINTNEXTLINE(performance-unnecessary-value-param)
                    [this](std::exception_ptr error) {
                        if (error) {
                            std::rethrow_exception(error);
                        }
                    }));
            }
 
            SetState(State::IDLE,
                     State::PREPARING,
                     "Error in transition to IDLE, DataPointRecordingUnit not in PREPARING State");
 
            while (m_stop == false) {
                switch (GetState()) {
                case State::IDLE:
                    if (m_start) {
                        SetState(State::WAITING,
                                 State::IDLE,
                                 "Error in transition to WAITING, DataPointRecordingUnit not in "
                                 "IDLE State");
                        break;
                    }
                    std::this_thread::sleep_for(1ms);
                    break;
                case State::WAITING:
                    if (not HasLeaders() or (HasLeaders() and HasFirstLeaderStarted())) {
                        if (m_capture_mask & CAPTURE_ON_START) {
                            m_repository->ReadDataPoint(m_dp_path, m_dp_buffer);
                            m_output->Write(AsTuple(m_dp_buffer));
                            m_samples_written++;
                        }
 
                        SetState(State::RUNNING,
                                 State::WAITING,
                                 "Error in transition to RUNNING, DataPointRecordingUnit not in "
                                 "WAITING State");
                        break;
                    }
                    std::this_thread::sleep_for(1ms);
                    break;
                case State::RUNNING:
                    if (HasLeaders() and HasLastLeaderFinished()) {
                        SetState(State::FINISHED,
                                 State::RUNNING,
                                 "Error in transition to FINISHED, DataPointRecordingUnit not in "
                                 "RUNNING State");
                        break;
                    }
                    std::this_thread::sleep_for(1ms);
                    break;
                default:
                    std::this_thread::sleep_for(1ms);
                }
            }
 
            if (m_capture_mask & CAPTURE_ON_CHANGE) {
                subscription.Unsubscribe();
            }
 
            if (m_capture_mask & CAPTURE_ON_STOP) {
                m_repository->ReadDataPoint(m_dp_path, m_dp_buffer);
                m_output->Write(AsTuple(m_dp_buffer));
                m_samples_written++;
            }
 
            m_output->Close();
            // State is set to stopped after join in Stop function
            m_start = false;
            m_stop = false;
            ResetLeaderStates();
 
        } catch (...) {
            m_output->Close();
            m_start = false;
            m_stop = false;
            SetFailed(std::current_exception());
            ResetLeaderStates();
        }
    }
 
    static std::tuple<TimepointType, OutputDpType> AsTuple(const DpType& data) {
        auto ts = taiclock::TaiClock::now().time_since_epoch().count();
        if constexpr (IS_SPAN_CONVERTIBLE<DpType>) {
            return std::make_tuple(ts, ToSpan(data));
        } else {
            return std::make_tuple(ts, data);
        }
    }
 
    std::optional<std::filesystem::path> m_file_path;
 
    DataPointPath m_dp_path;
 
    DataPointRecordingSource m_recording_source;
 
    RepositoryIf* m_repository;
    RepositorySubscriberIf* m_repository_subscriber;
 
    DpType m_dp_buffer;
 
    CaptureMask m_capture_mask;
 
    std::unique_ptr<OutputStageType> m_output;
 
    std::optional<size_t> m_recording_length;
 
    std::atomic<bool> m_start = false;
    std::atomic<bool> m_stop = false;
    std::thread m_process_thread;
 
    perfc::CounterI64 m_samples_written;
    perfc::ScopedRegistration m_samples_written_reg;

    inline static constexpr std::string_view RTR_PATH_CAPTURE_MASK =
        "/{}/static/rec_units/{}/capture_mask";
    inline static constexpr std::string_view OLDB_PATH_DP_NAME = "/{}/rec_units/{}/dp_name";
};
 
}  // namespace rtctk::componentFramework
 
#endif  // RTCTK_COMPONENTFRAMEWORK_DATAPOINTRECORDINGUNIT_HPP