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/fitsDataRecorder.hpp>
 
#include <boost/algorithm/string.hpp>
#include <ipcq/adapter.hpp>
#include <ipcq/reader.hpp>
 
#include <numapp/numapolicies.hpp>
#include <numapp/thread.hpp>
 
namespace rtctk::componentFramework {
 
template <typename RecInfoType>
void IpcqRecordingUnit<RecInfoType>::Prepare(const std::filesystem::path& file_path) {
    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");
    }
 
    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) {
                        SetState(State::WAITING,
                                 State::IDLE,
                                 "IpcqRecordingUnit not in IDLE before entering WAITING");
                    } else {
                        CII_THROW(IpcqError, "Error resetting ipcq Reader");
                    }
                    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 WAITING, before going RUNNING");
                    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<typename RecInfoType::Recorder::DisabledFields>::value) {
        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<typename RecInfoType::Recorder::TupleType>::value; i++) {
        result[i] = true;
    }
 
    for (const auto& name : fields) {
        bool found = false;
        for (size_t i = 0; i < std::tuple_size<typename RecInfoType::Recorder::TupleType>::value;
             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