eventRecordingUnit.hpp

Go to the documentation of this file.

 
#ifndef RTCTK_COMPONENTFRAMEWORK_EVENTRECORDINGUNIT_HPP
#define RTCTK_COMPONENTFRAMEWORK_EVENTRECORDINGUNIT_HPP
 
#include "rtctk/componentFramework/dataRecorder.hpp"
#include "rtctk/componentFramework/exceptions.hpp"
#include "rtctk/componentFramework/recordingInfo.hpp"
#include "rtctk/componentFramework/recordingUnit.hpp"
#include "rtctk/componentFramework/eventServiceIf.hpp"
#include "rtctk/componentFramework/typedEventService.hpp"
#include "rtctk/componentFramework/eventDefinitions.hpp"
#include "rtctk/componentFramework/fitsDataRecorder.hpp"
#include "ecs/taiClock.hpp"
 
#include <numapp/numapolicies.hpp>
#include <numapp/thread.hpp>
 
#include <cstdint>
#include <string>
#include <exception>
#include <mutex>
#include <thread>
#include <string_view>
 
namespace rtctk::componentFramework {
 
using namespace std::string_view_literals;
 
template <>
struct RecordingInfo<ComputationFinishedEvent> {
    using DataType = ComputationFinishedEvent;
    using Output = std::tuple<ecs::TaiClock::time_point::rep, std::string, std::string, std::string, std::string>;
    using Recorder = FitsRecorder<ecs::TaiClock::time_point::rep, std::string, std::string, std::string, std::string>;
 
    static constexpr Recorder::ColumnDescription COLUMNS = Recorder::ColumnDescription {
        { {"timestamp"sv, ""sv},
          {"source_timestamp"sv, ""sv},
          {"origin"sv, ""sv},
          {"item"sv, ""sv},
          {"result"sv, ""sv} }
    };
 
    static Output AsTuple(const DataType& data) {
        auto ts = ecs::TaiClock::now().time_since_epoch().count();
        return std::make_tuple(ts, std::to_string(data.time), data.origin, data.item, data.result.dump());
    }
};
 
template <>
struct RecordingInfo<ComputationStartedEvent> {
    using DataType = ComputationStartedEvent;
    using Output = std::tuple<ecs::TaiClock::time_point::rep, std::string, std::string, std::string>;
    using Recorder = FitsRecorder<ecs::TaiClock::time_point::rep, std::string, std::string, std::string>;
 
    static constexpr Recorder::ColumnDescription COLUMNS = Recorder::ColumnDescription {
        { {"timestamp"sv, ""sv},
          {"source_timestamp"sv, ""sv},
          {"origin"sv, ""sv},
          {"item"sv, ""sv} }
    };
 
    static Output AsTuple(const DataType& data) {
        auto ts = ecs::TaiClock::now().time_since_epoch().count();
        return std::make_tuple(ts, std::to_string(data.time), data.origin, data.item);
    }
};
 
template <>
struct RecordingInfo<ComputationEvent> {
    using DataType = ComputationEvent;
    using Output = std::tuple<ecs::TaiClock::time_point::rep, std::string, std::string, std::string, std::string>;
    using Recorder = FitsRecorder<ecs::TaiClock::time_point::rep, std::string, std::string, std::string, std::string>;
 
    static constexpr Recorder::ColumnDescription COLUMNS = Recorder::ColumnDescription {
        { {"timestamp"sv, ""sv},
          {"source_timestamp"sv, ""sv},
          {"origin"sv, ""sv},
          {"item"sv, ""sv},
          {"state"sv, ""sv} }
    };
 
    static Output AsTuple(const DataType& data) {
        auto ts = ecs::TaiClock::now().time_since_epoch().count();
        return std::make_tuple(ts, std::to_string(data.time), data.origin, data.item, data.state);
    }
};
 
template <>
struct RecordingInfo<ConfigurationEvent> {
    using DataType = ConfigurationEvent;
    using Output = std::tuple<ecs::TaiClock::time_point::rep, std::string, std::string, std::string, std::string>;
    using Recorder = FitsRecorder<ecs::TaiClock::time_point::rep, std::string, std::string, std::string, std::string>;
 
    static constexpr Recorder::ColumnDescription COLUMNS = Recorder::ColumnDescription {
        { {"timestamp"sv, ""sv},
          {"source_timestamp"sv, ""sv},
          {"origin"sv, ""sv},
          {"items"sv, ""sv},
          {"state"sv, ""sv} }
    };
 
    static Output AsTuple(const DataType& data) {
        auto ts = ecs::TaiClock::now().time_since_epoch().count();
        std::string items;
        bool is_first = true;
        for(const auto& item : data.items) {
            if(is_first==false) {
                items += ";";
            }
            items += item;
            is_first = false;
        }
        return std::make_tuple(ts, std::to_string(data.time), data.origin, items, data.state);
    }
};
 
// TODO: add trait classes for other event types
 
template <typename EventType, typename OutputStage = typename RecordingInfo<EventType>::Recorder>
class TypedEventRecordingUnit : public RecordingUnit {
public:
  using FilterMethod = std::function<bool(EventType const&)>;
 
    TypedEventRecordingUnit(std::string const& comp_id,
                            std::string const& unit_id,
                            RepositoryIf& rtr,
                            RepositoryIf& oldb,
                            EventServiceIf& es,
                            FilterMethod input_filter = nullptr)
        : RecordingUnit(comp_id, unit_id, rtr, oldb)
        , m_event_service(es)
        , m_subscriber(m_event_service.MakeSubscriber<EventType>())
        , m_input_filter(input_filter)
        , m_output(RecordingInfo<EventType>::COLUMNS)
        {
            Update();
        }
 
    virtual ~TypedEventRecordingUnit() {
        m_stop = true;
        if(m_process_thread.joinable()) {
            m_process_thread.join();
        }
        SetStopped();
    }
 
    void Prepare(const std::filesystem::path& file_path) override {
 
        SetState(State::PREPARING, State::STOPPED, "tried to prepare a non-stopped EventRecordingUnit");
 
        m_file_path = file_path / (GetId() + ".fits");
 
        auto policies = numapp::NumaPolicies();
        m_process_thread = numapp::MakeThread(m_unit_id, policies, [&]() { return Process(); });
    }
    void Start() override {
        m_start = true;
    }
    std::optional<std::filesystem::path> Stop() override {
        m_stop = true;
        if(m_process_thread.joinable()) {
            m_process_thread.join();
        }
 
        auto file = m_file_path;
        m_file_path.reset();
        SetStopped();
        return file;
    }
 
private:
 
    void Process() {
 
        using namespace std::chrono_literals;
 
        m_output.Open(*m_file_path);
 
        m_subscriber->Subscribe([&](EventType const& ev) {
            if(GetState() == State::RUNNING) {
                try{
                    if(m_input_filter) {
                        if(m_input_filter(ev)) {
                            m_output.Write(RecordingInfo<EventType>::AsTuple(ev));
                        }
                    }else{
                        m_output.Write(RecordingInfo<EventType>::AsTuple(ev));
                    }
                }catch (...) {
                    SetFailed(std::current_exception());
                    return;
                }
            }
        });
 
        SetState(State::IDLE, State::PREPARING, "EventRecordingUnit tried to go to IDLE from another state than PREPARING");
 
        while(m_stop==false) {
 
            switch(GetState()) {
                case State::IDLE: {
                    if(m_start) {
                        SetState(State::WAITING, State::IDLE, "EventRecorder illegal state transition, tried going to WAITING from another state than IDLE");
                        break;
                    }
                    std::this_thread::sleep_for(1ms);
                    break;
                }
                case State::WAITING: {
                    if(not HasLeaders() or (HasLeaders() and HasFirstLeaderStarted())) {
                        SetState(State::RUNNING, State::WAITING, "EventRecorder illegal state transition, tried going to RUNNING from another state than WAITING");
                        break;
                    }
                    std::this_thread::sleep_for(1ms);
                    break;
                }
                case State::RUNNING: {
                    if(HasLeaders() and HasLastLeaderFinished()) {
                        SetState(State::FINISHED, State::RUNNING, "EventRecorder illegal state transition, tried going to FINISHED from another state than RUNNING");
                        break;
                    }
                    std::this_thread::sleep_for(1ms);
                }
                default: {
                    std::this_thread::sleep_for(1ms);
                }
            }
        }
 
        m_subscriber->Unsubscribe();
        m_output.Close();
        // Stopped state will be set by Stop function after join
        m_start = false;
        m_stop = false;
        ResetLeaderStates();
    }
 
    TypedEventService m_event_service;
    std::unique_ptr<TypedEventSubscriber<EventType>> m_subscriber;
 
    FilterMethod m_input_filter;
 
    OutputStage m_output;
 
    std::atomic<bool> m_start = false;
    std::atomic<bool> m_stop = false;
    std::thread m_process_thread;
};
 
 
template <typename OutputStage = FitsRecorder<ecs::TaiClock::time_point::rep, std::string>>
class JsonEventRecordingUnit : public RecordingUnit {
public:
  using FilterMethod = std::function<bool(JsonPayload const&)>;
 
    JsonEventRecordingUnit(std::string const& comp_id,
                           std::string const& unit_id,
                           RepositoryIf& rtr,
                           RepositoryIf& oldb,
                           EventServiceIf& es,
                           std::string const& topic_name,
                           FilterMethod input_filter = nullptr)
        : RecordingUnit(comp_id, unit_id, rtr, oldb)
        , m_event_service(es)
        , m_subscriber(m_event_service.MakeSubscriber(topic_name))
        , m_input_filter(input_filter)
        , m_output(COLUMNS)
        {
            Update();
        }
 
    virtual ~JsonEventRecordingUnit() {
        m_stop = true;
        if(m_process_thread.joinable()) {
            m_process_thread.join();
        }
        SetStopped();
    }
 
    void Prepare(const std::filesystem::path& file_path) override {
 
        SetState(State::PREPARING, State::STOPPED, "EventRecorder tried to go from non-STOPPED State to PREPARING");
 
        m_file_path = file_path / (GetId() + ".fits");
 
        auto policies = numapp::NumaPolicies();
        m_process_thread = numapp::MakeThread(m_unit_id, policies, [&]() { return Process(); });
    }
    void Start() override {
        m_start = true;
    }
    std::optional<std::filesystem::path> Stop() override {
        m_stop = true;
        if(m_process_thread.joinable()) {
            m_process_thread.join();
        }
        SetStopped();
        return m_file_path;
    }
 
private:
 
    void Process() {
 
        using namespace std::chrono_literals;
 
        m_output.Open(*m_file_path);
 
        m_subscriber->Subscribe([&](JsonPayload const& ev) {
 
            if(GetState() == State::RUNNING) {
                try{
                    if(m_input_filter) {
                        if(m_input_filter(ev)) {
                            m_output.Write(AsTuple(ev));
                        }
                    }else{
                        m_output.Write(AsTuple(ev));
                    }
                }catch (...) {
                    SetFailed(std::current_exception());
                    return;
                }
            }
        });
 
        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())) {
                        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);
                }
                default: {
                    std::this_thread::sleep_for(1ms);
                }
            }
        }
 
        m_subscriber->Unsubscribe();
        m_output.Close();
        // the actual STOPPED state will be set by the Stop function
        m_start = false;
        m_stop = false;
        ResetLeaderStates();
    }
 
    static std::tuple<ecs::TaiClock::time_point::rep, std::string> AsTuple(JsonPayload const& data) {
        auto ts = ecs::TaiClock::now().time_since_epoch().count();
        return std::make_tuple(ts, data.dump());
    }
 
    static constexpr typename OutputStage::ColumnDescription COLUMNS = typename OutputStage::ColumnDescription {
        { {"timestamp"sv, ""sv},
          {"payload"sv, ""sv} }
    };
 
    EventServiceIf& m_event_service;
    std::unique_ptr<EventSubscriberIf> m_subscriber;
 
    FilterMethod m_input_filter;
 
    OutputStage m_output;
 
    std::atomic<bool> m_start = false;
    std::atomic<bool> m_stop = false;
    std::thread m_process_thread;
};
 
}  // namespace rtctk::componentFramework
 
#endif  // RTCTK_COMPONENTFRAMEWORK_EVENTRECORDINGUNIT_HPP