recordingThreadDetail.hpp

Go to the documentation of this file.

 
#ifndef RTCTK_REUSABLECOMPONENT_TELEMETRYRECORDER_RECORDINGTHREAD_DETAIL_HPP
#define RTCTK_REUSABLECOMPONENT_TELEMETRYRECORDER_RECORDINGTHREAD_DETAIL_HPP
 
#include <rtctk/telemetryRecorder/recordingThread.hpp>
#include <exception>
// include the numapp for threading
#include <numapp/mempolicy.hpp>
#include <numapp/numapolicies.hpp>
#include <numapp/thread.hpp>
 
namespace rtctk::telemetryRecorder {
 
template<typename Writer, typename Topic, typename Reader>
RecordingThread<Writer,Topic,Reader>::RecordingThread() :
    m_state(STOPPED),
    m_reader(nullptr),
    m_writer(std::make_unique<Writer>()) {}
 
template<typename Writer, typename Topic, typename Reader>
RecordingThread<Writer,Topic,Reader>::RecordingThread(std::unique_ptr<Writer>&& writer, const std::string& topic) :
    m_state(STOPPED),
    m_reader(std::make_unique<Reader>(topic.c_str())),
    m_writer(std::move(writer)) {}
 
template<typename Writer, typename Topic, typename Reader>
RecordingThread<Writer,Topic,Reader>::RecordingThread(std::unique_ptr<Writer>&& writer, std::unique_ptr<Reader>&& reader) :
    m_state(STOPPED),
    m_reader(std::move(reader)),
    m_writer(std::move(writer)) {}
 
template<typename Writer, typename Topic, typename Reader>
RecordingThread<Writer,Topic,Reader>::~RecordingThread()
{
    {
        std::lock_guard<std::mutex> state_guard(m_state_lock);
        m_state = STOPPED;
    }
    if(m_thread.joinable()) {
        m_thread.join();
    }
}
 
template<typename Writer, typename Topic, typename Reader>
void RecordingThread<Writer,Topic,Reader>::Start() {
    std::lock_guard<std::mutex> state_guard(m_state_lock);
    if (m_state != STOPPED)
        CII_THROW(InvalidRecordingThreadOperation, "Tried to Start() on running ReaderThread");
    m_state = PAUSED;
    m_thread = std::thread(&RecordingThread<Writer,Topic,Reader>::Process, this);
}
 
template<typename Writer, typename Topic, typename Reader>
void RecordingThread<Writer,Topic,Reader>::Stop() {
    {
        std::lock_guard<std::mutex> state_guard(m_state_lock);
        if (m_state == STOPPED)
            CII_THROW(InvalidRecordingThreadOperation, "Tried to Stop() already stopped ReaderThread");
        m_state = STOPPED;
    }
    if(m_thread.joinable()) {
        m_thread.join();
    }
}
 
template<typename Writer, typename Topic, typename Reader>
void RecordingThread<Writer,Topic,Reader>::Pause() {
    std::lock_guard<std::mutex> state_guard(m_state_lock);
    if (m_state != RUNNING)
        CII_THROW(InvalidRecordingThreadOperation, "Tried to Pause() on non-running ReaderThread");
    m_state = PAUSED;
}
 
template<typename Writer, typename Topic, typename Reader>
void RecordingThread<Writer,Topic,Reader>::Unpause() {
    std::lock_guard<std::mutex> state_guard(m_state_lock);
    if (m_state != PAUSED)
        CII_THROW(InvalidRecordingThreadOperation, "Tried to Unpause() on not paused ReaderThread");
    m_state = RUNNING;
}
 
template<typename Writer, typename Topic, typename Reader>
void RecordingThread<Writer,Topic,Reader>::Process() {
    std::error_code const ok{};
    while(true) {
        enum ThreadState state_copy;
        {
            std::lock_guard<std::mutex> state_guard(m_state_lock);
            state_copy = m_state;
        }
        while (state_copy == PAUSED) {
            std::this_thread::sleep_for(std::chrono::milliseconds(100));
            m_state_lock.lock();
            state_copy = m_state;
            m_state_lock.unlock();
        }
        if (state_copy == STOPPED) {
            break;
        }
        size_t skip = 0;
        size_t number = m_reader->NumAvailable();
        if (number == 0) { // try to always read at least one
            number = 1;
        }
        if (m_subsample_rate > 1) {
            m_sampling_counter = m_sampling_counter % m_subsample_rate;
            if (m_sampling_counter == 0) {
                number = 1;
            } else {
                skip = std::min(number, m_subsample_rate-m_sampling_counter);
            }
        }
        std::pair<std::error_code, size_t> result;
        if(skip == 0) {
            result = m_reader->Read([&] (const Topic& data) {this->ReadDataWrapper(data);}, number, std::chrono::milliseconds(100));
        } else {
            // skip
            result = m_reader->Read([] (const Topic& data) {}, skip, std::chrono::milliseconds(100));
        }
        m_sampling_counter += result.second;
        // check for error
        if (not (result.first == ok or result.first == ipcq::Error::Timeout)) {
            // TODO better error handling?
            // We stop in case of error
            m_state_lock.lock();
            m_state = STOPPED;
            m_state_lock.unlock();
            std::string error = "Error reading from ipcq: " + result.first.message();
            CII_THROW(RecordingThreadIPCQError, error);
        }
    }
}
 
template<typename Writer, typename Topic, typename Reader>
void RecordingThread<Writer,Topic,Reader>::Open() {
    std::lock_guard<std::mutex> state_guard(m_state_lock);
    if (m_state != PAUSED && m_state != STOPPED) {
        CII_THROW(InvalidRecordingThreadOperation, "Tried to open new file, while RecorderThread was running");
    }
    auto time = std::chrono::system_clock::now();
    time_t now = std::chrono::system_clock::to_time_t(time);
    std::stringstream format;
    format << m_output_dir << "/" << std::put_time(localtime(&now), "%F %T %Z") << ".fits";
    m_writer->Open(format.str());
    if (not m_reader) {
        m_reader = std::make_unique<Reader>(m_topic_name.c_str());
    }
}
 
template<typename Writer, typename Topic, typename Reader>
void RecordingThread<Writer,Topic,Reader>::Close() {
    std::lock_guard<std::mutex> state_guard(m_state_lock);
    if (m_state != PAUSED && m_state != STOPPED) {
        CII_THROW(InvalidRecordingThreadOperation, "Tried to close file, while RecorderThread was running");
    }
    m_writer->Close();
}
 
template<typename Writer, typename Topic, typename Reader>
void RecordingThread<Writer,Topic,Reader>::SetOutputUri(const ::elt::mal::Uri& storage_uri) {
    if(storage_uri.scheme() != network::uri::string_view("file")) {
        CII_THROW(InvalidStorageURLException, "Invalid scheme for TelemetryRecorder");
    }
    m_output_dir = storage_uri.path().to_string();
}
 
template<typename Writer, typename Topic, typename Reader>
void RecordingThread<Writer,Topic,Reader>::SetSubSampleFactor(uint64_t rate) {
    m_subsample_rate = rate;
}
 
template<typename Writer, typename Topic, typename Reader>
void RecordingThread<Writer,Topic,Reader>::SetTelemetrySubSet(const std::string& subset) {
    std::lock_guard<std::mutex> state_guard(m_state_lock);
    if (m_state != PAUSED && m_state != STOPPED) {
        CII_THROW(InvalidRecordingThreadOperation, "Tried to change TelemetrySubset, while RecorderThread was running");
    }
    m_writer->SetTelemetrySubSet(subset);
}
 
template<typename Writer, typename Topic, typename Reader>
void RecordingThread<Writer,Topic,Reader>::SetTopic(const std::string& topic) {
    m_topic_name = topic;
}
 
template<typename Writer, typename Topic, typename Reader>
void RecordingThread<Writer,Topic,Reader>::SetStartSampleId(const uint64_t sample_id) {
    if(m_last_observed_sample_id != 0 and sample_id <= m_last_observed_sample_id) {
        CII_THROW(RecordingThreadInvalidStartSample, sample_id);
    }
    m_start_sample = sample_id;
}
 
template<typename Writer, typename Topic, typename Reader>
void RecordingThread<Writer,Topic,Reader>::ReadDataWrapper(const Topic& telemetry_data) {
    m_last_observed_sample_id = telemetry_data.sample_id;
    if(m_last_observed_sample_id < m_start_sample) {
        return;
    }
    m_writer->WriteData(telemetry_data);
}
 
} // namespace rtctk::telemetryRecorder
 
#endif //RTCTK_REUSABLECOMPONENT_TELEMETRYRECORDER_RECORDINGTHREAD_DETAIL_HPP