ipcqRecordingUnit.ipp

Go to the documentation of this file.

 
// Note this is a template implementation file and should not be included directly.
// The typical header protection macro is not added to avoid it showing up in Doxygen API
// documentation.
#pragma once
 
#include <rtctk/componentFramework/ipcqRecordingUnit.hpp>
 
#include <rtctk/componentFramework/fitsDataRecorder.hpp>
 
#include <boost/algorithm/string.hpp>
#include <ipcq/adapter.hpp>
#include <ipcq/error.hpp>
#include <ipcq/reader.hpp>
 
#include <numapp/numapolicies.hpp>
#include <numapp/thread.hpp>
 
namespace rtctk::componentFramework {
 
template <typename RecInfoType>
IpcqRecordingUnit<RecInfoType>::IpcqRecordingUnit(const std::string& comp_id,
                                                  const std::string& unit_id,
                                                  ServiceContainer& services)
    : RecordingUnit(comp_id, unit_id, "IPCQ", services)
    , m_queue_name(m_rtr.GetDataPoint<std::string>(
          DataPointPath(fmt::format(RTR_PATH_QUEUE_NAME, comp_id, unit_id))))
    , m_output(RecInfoType::COLUMNS) {
    auto queue_name_path = DataPointPath(fmt::format(OLDB_PATH_QUEUE_NAME, comp_id, unit_id));
    m_oldb.CreateDataPoint<std::string>(queue_name_path);
    m_oldb.SetDataPoint<std::string>(queue_name_path, m_queue_name);
 
    auto cpu_affinity_path = DataPointPath(fmt::format(RTR_PATH_CPU_AFFINITY, comp_id, unit_id));
    if (m_rtr.DataPointExists(cpu_affinity_path)) {
        m_cpu_affinity = m_rtr.GetDataPoint<int32_t>(cpu_affinity_path);
    }
 
    const InfluxTagMap base_tags = {{"unit_id", unit_id}};
 
    m_samples_written_reg =
        m_metrics.AddCounter(&m_samples_written,
                             CounterMetricInfo(unit_id + "/samples_written",
                                               "Number of samples successfully written to file",
                                               base_tags,
                                               "samples_written"));
 
    m_last_observed_sample_id_reg =
        m_metrics.AddCounter(&m_last_observed_sample_id,
                             CounterMetricInfo(unit_id + "/last_sample_id_written",
                                               "Last sample id successfully written to file",
                                               base_tags,
                                               "last_sample_id_written"));
 
    m_freq_estimator = std::make_unique<FrequencyEstimator<>>(
        m_metrics, "Estimated frequency of the data writer", unit_id);
 
    m_dur_monitor =
        std::make_unique<DurationMonitor<>>(m_metrics, "Duration of writing data to file", unit_id);
 
    m_buffer_monitor =
        std::make_unique<BufferMonitor<>>(m_metrics, "SHM read buffer occupancy", unit_id);
 
    LoadDynamicConfig();
}
 
template <typename RecInfoType>
IpcqRecordingUnit<RecInfoType>::~IpcqRecordingUnit() {
    m_stop = true;
    if (m_process_thread.joinable()) {
        m_process_thread.join();
    }
}
 
template <typename RecInfoType>
void IpcqRecordingUnit<RecInfoType>::Prepare(const std::filesystem::path& file_path) {
    if (not IsEnabled()) {
        return;
    }
 
    SetState(State::PREPARING, State::STOPPED, "tried to prepare a non-stopped ipcqRecordingUnit");
 
    m_file_path = file_path / (GetId() + ".fits");
 
    m_stop = false;
    auto policies = numapp::NumaPolicies();
    if (m_cpu_affinity.has_value()) {
        auto cpu_mask =
            numapp::Cpumask::MakeFromCpuStringAll(std::to_string(m_cpu_affinity.value()).c_str());
        policies.SetCpuAffinity(numapp::CpuAffinity(cpu_mask));
    }
 
    m_process_thread =
        numapp::MakeThread(m_unit_id.substr(0, 15), policies, [&]() { return Process(); });
}
 
template <typename RecInfoType>
void IpcqRecordingUnit<RecInfoType>::Start() {
    m_start = true;
}
 
template <typename RecInfoType>
std::vector<std::filesystem::path> IpcqRecordingUnit<RecInfoType>::Stop() {
    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;
}
 
template <typename RecInfoType>
void IpcqRecordingUnit<RecInfoType>::Update() {
    if (GetState() == State::RUNNING) {
        CII_THROW(InvalidStateChange, "tried to update a running IpcqRecordingUnit");
    }
 
    RecordingUnit::Update();
    LoadDynamicConfig();
}
 
template <typename RecInfoType>
void IpcqRecordingUnit<RecInfoType>::LoadDynamicConfig() {
    auto subsample_factor_path =
        DataPointPath(fmt::format(RTR_PATH_SUBSAMPLE_FACTOR, m_comp_id, m_unit_id));
    m_subsample_factor = m_rtr.GetDataPoint<int64_t>(subsample_factor_path);
 
    auto disabled_fields = GetDisabled(
        m_rtr, DataPointPath(fmt::format(RTR_PATH_TELEMETRY_SUBSET, m_comp_id, m_unit_id)));
    m_output.SetDisabledFields(disabled_fields);
 
    auto start_sample_id_path =
        DataPointPath(fmt::format(RTR_PATH_START_SAMPLE_ID, m_comp_id, m_unit_id));
    m_start_sample_id = m_rtr.GetDataPoint<int64_t>(start_sample_id_path);
 
    auto stop_after_samples_path =
        DataPointPath(fmt::format(RTR_PATH_STOP_AFTER_NUM_SAMPLES, m_comp_id, m_unit_id));
    m_stop_after_num_samples = m_rtr.GetDataPoint<int64_t>(stop_after_samples_path);
}
 
template <typename RecInfoType>
void IpcqRecordingUnit<RecInfoType>::Process() {
    using namespace std::chrono_literals;
 
    try {
        std::vector<typename RecInfoType::Topic> buffer;
        buffer.reserve(MAX_SAMPLES_READ);
 
        const std::error_code ok{};
        std::pair<std::error_code, size_t> result;
 
        bool files_open = false;
        m_samples_written.Store(0);
        m_last_observed_sample_id.Store(0);
 
        auto reader = ipcq::Reader<typename RecInfoType::Topic>(m_queue_name.c_str());
 
        size_t reader_capacity = reader.Capacity();
 
        if (not SetState(State::IDLE, State::PREPARING)) {
            try {
                CII_THROW(InvalidStateChange,
                          "IpcqRecordingUnit not in preparing before starting thread");
            } catch (...) {
                SetFailed(std::current_exception());
                return;
            }
        }
 
        while (m_stop == false) {
            switch (GetState()) {
            case State::IDLE: {
                if (m_start) {
                    if (auto ret = reader.Reset();
                        ret == ok || ret == ipcq::Error::Closed || ret == ipcq::Error::WouldBlock) {
                        SetState(State::WAITING,
                                 State::IDLE,
                                 "IpcqRecordingUnit not in IDLE before entering WAITING");
                    } else {
                        CII_THROW(IpcqError,
                                  fmt::format("Error resetting ipcq Reader: {}", ret.message()));
                    }
                    break;
                }
                std::this_thread::sleep_for(1ms);
                break;
            }
            case State::WAITING: {
                result = reader.Read(ipcq::BackInserter(buffer), 1, 100ms);
                if (not(result.first == ok or result.first == ipcq::Error::Timeout)) {
                    std::string error = "Error reading from ipcq: " + result.first.message();
                    CII_THROW(IpcqError, error);
                }
                for (const auto& element : buffer) {
                    m_last_observed_sample_id.Store(element.sample_id);
                }
                buffer.clear();
 
                if ((m_last_observed_sample_id.Load() >= m_start_sample_id - 1) and
                    (not HasLeaders() or (HasLeaders() and HasFirstLeaderStarted()))) {
                    m_sampling_counter = 0;
                    SetState(State::RUNNING,
                             State::WAITING,
                             "Expected to be in WAITING, before going RUNNING");
                }
                break;
            }
            case State::RUNNING: {
                if (HasLeaders() and HasLastLeaderFinished()) {
                    SetState(State::FINISHED,
                             State::RUNNING,
                             "Expected to be in RUNNING, before going FINISHED");
                    break;
                }
 
                size_t to_skip = 0;
                size_t to_read = reader.NumAvailable();
                if (to_read == 0) {  // try to always read at least one
                    to_read = 1;
                }
                to_read = std::min(to_read, MAX_SAMPLES_READ);
 
                if (m_subsample_factor > 1) {
                    m_sampling_counter = m_sampling_counter % m_subsample_factor;
                    if (m_sampling_counter == 0) {
                        to_read = 1;
                    } else {
                        to_skip = std::min(to_read, m_subsample_factor - m_sampling_counter);
                    }
                }
 
                if (to_skip == 0) {
                    m_buffer_monitor->Tick(reader.NumAvailable(), reader_capacity);
                    result = reader.Read(ipcq::BackInserter(buffer), to_read, 100ms);
                    for (const auto& element : buffer) {
                        m_last_observed_sample_id.Store(element.sample_id);
                        auto t1 = std::chrono::steady_clock::now();
                        auto sample = RecInfoType::AsTuple(element);
                        if (not files_open) {
                            // defer opening of files until we get first sample and can set sizes
                            m_output.SetColumnLength(sample);
                            m_output.Open(*m_file_path);
                            files_open = true;
                        }
                        m_output.Write(sample);
                        auto t2 = std::chrono::steady_clock::now();
                        m_dur_monitor->Tick(t2 - t1);
                        m_freq_estimator->Tick();
                        m_samples_written++;
                        if (m_stop_after_num_samples > 0 and
                            m_samples_written.Load() >= m_stop_after_num_samples) {
                            SetState(State::FINISHED,
                                     State::RUNNING,
                                     "Expected to be in RUNNING, before going FINISHED");
                            break;
                        }
                    }
                    buffer.clear();
                } else {
                    result = reader.Skip(to_skip, 100ms);
                }
                m_sampling_counter += result.second;
                // check for error
                if (not(result.first == ok or result.first == ipcq::Error::Timeout)) {
                    std::string error = "Error reading from ipcq: " + result.first.message();
                    CII_THROW(IpcqError, error);
                }
                break;
            }
            default: {
                std::this_thread::sleep_for(1ms);
            }
            }
        }
 
        // cleanup
        m_output.Close();
        m_start = false;
        m_stop = false;
        ResetLeaderStates();
        // the actual STOPPED state will be set by the Stop function
 
    } catch (...) {
        // try closing the file even if it might fail
        try {
            m_output.Close();
        } catch (const FitsDataRecorderFitsError& e) {
            // ignore this error we will keep the original exception
        }
        m_start = false;
        m_stop = false;
        SetFailed(std::current_exception());
        ResetLeaderStates();
    }
}
 
template <typename RecInfoType>
typename RecInfoType::Recorder::DisabledFields
IpcqRecordingUnit<RecInfoType>::GetDisabled(RepositoryIf& rtr, const DataPointPath& path) {
    typename RecInfoType::Recorder::DisabledFields result{};
 
    if (not rtr.DataPointExists(path)) {
        return result;  // no fields disabled
    }
    auto fields = rtr.GetDataPoint<std::vector<std::string>>(path);
 
    if (fields.empty()) {
        return result;  // no fields disabled
    }
 
    if (fields.size() > std::tuple_size_v<typename RecInfoType::Recorder::DisabledFields>) {
        CII_THROW(InvalidSetting, "Invalid disabled fields setting. Too many fields");
    }
 
    // set them all to disabled an rean
    for (size_t i = 0; i < std::tuple_size_v<typename RecInfoType::Recorder::TupleType>; i++) {
        result[i] = true;
    }
 
    for (const auto& name : fields) {
        bool found = false;
        for (size_t i = 0; i < std::tuple_size_v<typename RecInfoType::Recorder::TupleType>; i++) {
            if (name == RecInfoType::COLUMNS[i].name) {
                result[i] = false;
                found = true;
                break;
            }
        }
        if (!found) {
            CII_THROW(InvalidSetting,
                      std::string{"Invalid disabled fields setting: Could not find field '" + name +
                                  "'"});
        }
    }
    return result;
}
 
}  // namespace rtctk::componentFramework