eventRecordingUnit.hpp

Go to the documentation of this file.

#ifndef RTCTK_COMPONENTFRAMEWORK_EVENTRECORDINGUNIT_HPP
#define RTCTK_COMPONENTFRAMEWORK_EVENTRECORDINGUNIT_HPP
 
#include <rtctk/componentFramework/componentMetricsIf.hpp>
#include <rtctk/componentFramework/eventDefinitions.hpp>
#include <rtctk/componentFramework/eventServiceIf.hpp>
#include <rtctk/componentFramework/exceptions.hpp>
#include <rtctk/componentFramework/fitsDataRecorder.hpp>
#include <rtctk/componentFramework/recordingInfo.hpp>
#include <rtctk/componentFramework/recordingUnit.hpp>
 
#include <fmt/format.h>
#include <numapp/numapolicies.hpp>
#include <numapp/thread.hpp>
#include <taiclock/taiClock.hpp>
 
#include <cstdint>
#include <exception>
#include <string>
#include <string_view>
#include <thread>
 
namespace rtctk::componentFramework {

template <class OutputStage = std::ofstream>
class EventRecordingUnit : public RecordingUnit {
public:
    using FilterMethod = std::function<bool(const JsonPayload&)>;

    EventRecordingUnit(const std::string& comp_id,
                       const std::string& unit_id,
                       ServiceContainer& services,
                       FilterMethod input_filter = nullptr)
        : RecordingUnit(comp_id, unit_id, "Event", services)
        , m_event_service(services.Get<EventServiceIf>()) {
        if (input_filter) {
            m_input_filter = std::move(input_filter);
        } else {
            m_input_filter = [](const JsonPayload&) { return true; };
        }
 
        auto rtr_topic_path = DataPointPath(fmt::format(RTR_PATH_TOPIC_NAME, comp_id, unit_id));
        m_topic_name = m_rtr.GetDataPoint<std::string>(rtr_topic_path);
 
        auto oldb_topic_path = DataPointPath(fmt::format(OLDB_PATH_TOPIC_NAME, comp_id, unit_id));
        m_oldb.CreateDataPoint<std::string>(oldb_topic_path);
        m_oldb.SetDataPoint<std::string>(oldb_topic_path, m_topic_name);
 
        const InfluxTagMap base_tags = {
            {"unit_id", unit_id},
            {"topic_name", m_topic_name},
        };
 
        m_samples_written_reg = m_metrics.AddCounter(
            &m_samples_written, CounterMetricInfo(
                unit_id + "/samples_written",
                "Samples written",
                base_tags,
                "samples_written"
            )
        );
    }
 
    ~EventRecordingUnit() 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,
                 "EventRecorder tried to go from non-STOPPED State to PREPARING");
 
        m_subscriber.reset();
        m_subscriber = m_event_service.MakeSubscriber(m_topic_name);
 
        m_file_path = file_path / (GetId() + ".jsonl");
 
        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();
        }
        SetStopped();
        std::vector<std::filesystem::path> files;
        if (m_file_path) {
            files.push_back(*m_file_path);
        }
        return files;
    }
 
private:
    void Process() {
        using namespace std::chrono_literals;
 
        OutputStage output;
 
        try {
            output.open(*m_file_path, std::ios::out);
            m_samples_written.Store(0);
 
            SetState(State::IDLE,
                     State::PREPARING,
                     "Event recorder was not in PREPARING when trying to go to IDLE");
 
            while (m_stop == false) {
                switch (GetState()) {
                case State::IDLE:
                    if (m_start) {
                        SetState(State::WAITING,
                                 State::IDLE,
                                 "Event recorder was not in IDLE when trying to go to WAITING");
                        break;
                    }
                    std::this_thread::sleep_for(1ms);
                    break;
                case State::WAITING:
                    if (not HasLeaders() or (HasLeaders() and HasFirstLeaderStarted())) {
 
                        m_subscriber->Subscribe([&](const JsonPayload& ev) {
                            if (GetState() == State::RUNNING) {
                                try {
                                    if (m_input_filter(ev)) {
                                        output << ev << "\n";
                                        m_samples_written++;
                                    }
                                } catch (...) {
                                    SetFailed(std::current_exception());
                                    return;
                                }
                            }
                        });
 
                        SetState(State::RUNNING,
                                 State::WAITING,
                                 "Event recorder was not in WAITING when trying to go to RUNNING");
                        break;
                    }
                    std::this_thread::sleep_for(1ms);
                    break;
                case State::RUNNING:
                    if (HasLeaders() and HasLastLeaderFinished()) {
                        SetState(State::FINISHED,
                                 State::RUNNING,
                                 "Event recorder was not running when trying to go to finish");
                        break;
                    }
                    std::this_thread::sleep_for(1ms);
                    break;
                default:
                    std::this_thread::sleep_for(1ms);
                }
            }
 
            m_subscriber->Unsubscribe();
            // the actual STOPPED state will be set by the Stop function
            m_start = false;
            m_stop = false;
            ResetLeaderStates();
        } catch (...) {
            m_subscriber->Unsubscribe();
            m_start = false;
            m_stop = false;
            SetFailed(std::current_exception());
            ResetLeaderStates();
        }
    }
 
    EventServiceIf& m_event_service;
    std::unique_ptr<EventSubscriberIf> m_subscriber;
 
    FilterMethod m_input_filter;
 
    std::string m_topic_name;
 
    perfc::CounterI64 m_samples_written;
    perfc::ScopedRegistration m_samples_written_reg;
 
    std::atomic<bool> m_start = false;
    std::atomic<bool> m_stop = false;
    std::thread m_process_thread;

    inline static constexpr std::string_view RTR_PATH_TOPIC_NAME =
        "/{}/static/rec_units/{}/topic_name";

    inline static constexpr std::string_view OLDB_PATH_TOPIC_NAME = "/{}/rec_units/{}/topic_name";
};
 
}  // namespace rtctk::componentFramework
 
#endif  // RTCTK_COMPONENTFRAMEWORK_EVENTRECORDINGUNIT_HPP