rtctkGenRtrWriter.h

Go to the documentation of this file.

 
#ifndef RTCTK_GEN_RTR_WRITER_
#define RTCTK_GEN_RTR_WRITER_
 
#include "rtctk/componentFramework/logger.hpp"
#include "rtctk/componentFramework/runtimeRepoIf.hpp"
#include <rtctk/componentFramework/matrixSpan.hpp>
 
#include <boost/accumulators/accumulators.hpp>
#include <boost/accumulators/statistics/count.hpp>
#include <boost/accumulators/statistics/max.hpp>
#include <boost/accumulators/statistics/mean.hpp>
#include <boost/accumulators/statistics/min.hpp>
#include <boost/accumulators/statistics/stats.hpp>
#include <boost/accumulators/statistics/variance.hpp>
 
namespace rtctk::standaloneTools {
 
namespace rtctk_cfw = rtctk::componentFramework;
using namespace std::chrono;
namespace ba = boost::accumulators;
 
class GenRtrWriterBase {
protected:
    log4cplus::Logger& m_logger;
    uint16_t m_idx;
    std::shared_ptr<rtctk_cfw::RuntimeRepoIf> m_rtr;
    rtctk_cfw::DataPointPath m_dp;
    std::string m_server_alias;
    bool m_dp_created;
    uint32_t m_payload_bytes;
 
    std::chrono::microseconds m_sleep_period;
    uint32_t m_iteration_num;
 
    std::thread m_thread;
    std::atomic<bool> m_thread_running = true;
 
    ba::accumulator_set<
        unsigned,
        ba::stats<ba::tag::mean, ba::tag::variance, ba::tag::min, ba::tag::max, ba::tag::count>>
        m_accumulator;
 
public:
    explicit GenRtrWriterBase(std::shared_ptr<rtctk_cfw::RuntimeRepoIf>& rtr,
                              const rtctk_cfw::DataPointPath& dp,
                              const std::string& server_alias = "")
        : m_logger(rtctk_cfw::GetLogger("rtctk"))
        , m_rtr(rtr)
        , m_dp(dp)
        , m_server_alias(server_alias) {
        s_idx++;
    }
 
    virtual ~GenRtrWriterBase() noexcept {
        --s_idx;
    }
 
    GenRtrWriterBase() = delete;
    GenRtrWriterBase(const GenRtrWriterBase&) = delete;
    GenRtrWriterBase(GenRtrWriterBase&&) = delete;
 
    virtual void Initialize(std::size_t dim_x,
                            std::size_t dim_y,
                            std::chrono::microseconds period,
                            uint32_t iterations) = 0;
 
    void SignalThreadExit() {
        m_thread_running = false;
    }  // SignalThreadExit
 
    void Finalize() {
        LOG4CPLUS_INFO_FMT(m_logger, "[%s] Finalize", m_dp.ToString().c_str());
 
        SignalThreadExit();
 
        if (m_thread.joinable()) {
            m_thread.join();
        }
        //? should we check if dp exists?
        if (m_dp_created) {
            LOG4CPLUS_DEBUG_FMT(m_logger, "[%s] Going to delete DP.", m_dp.ToString().c_str());
            m_rtr->DeleteDataPoint(m_dp);
        }
    }  // Finalize
 
    void Join() {
        m_thread.join();
    }  // Join
 
    // index of Generic RTR Writer
    static std::atomic<uint16_t> s_idx;
};  // GenRtrWriterBase
 
template <typename T>
class GenRtrWriter : public GenRtrWriterBase {
    std::unique_ptr<rtctk_cfw::MatrixBuffer<T>> m_value;
 
    void WriteDataPoint() {
        uint16_t error_counter = 0;
        if (m_iteration_num > 0) {
            LOG4CPLUS_INFO_FMT(m_logger,
                               "[%s] Going to write value of size: %ld x %ld (%d bytes) %d times.",
                               m_dp.ToString().c_str(),
                               m_value->GetNrows(),
                               m_value->GetNcols(),
                               m_payload_bytes,
                               m_iteration_num);
        } else {
            LOG4CPLUS_INFO_FMT(
                m_logger,
                "[%s] Going to write value of size: %ld x %ld (%d bytes) until Ctrl-C.",
                m_dp.ToString().c_str(),
                m_value->GetNrows(),
                m_value->GetNcols(),
                m_payload_bytes);
        }
 
        // Enter the work loop: incrementing iteration counter
        // and write value to RTR's DP. We continue until the thread is running.
        uint32_t iteration_counter = 0;
        while (m_thread_running && (m_iteration_num == 0 ||
                                    (m_iteration_num > 0 && iteration_counter < m_iteration_num))) {
            try {
                if (iteration_counter % 100 == 0) {
                    LOG4CPLUS_DEBUG_FMT(m_logger,
                                        "[%s] Going to write DP. Iteration: %d.",
                                        m_dp.ToString().c_str(),
                                        iteration_counter);
                }
                auto timestamp = high_resolution_clock::now();  // timestamp
                //? set timestamp to value to be written to RTR
                auto t1 = std::chrono::steady_clock::now();
                // write DP
                m_rtr->WriteDataPoint(m_dp, *m_value);
                //? EH
 
                auto t2 = std::chrono::steady_clock::now();
                unsigned t_diff =
                    std::chrono::duration_cast<std::chrono::microseconds>(t2 - t1).count();
                m_accumulator(t_diff);
                error_counter = 0;
                if (iteration_counter % 100 == 0) {
                    auto timestamp_nanos =
                        duration_cast<nanoseconds>(timestamp.time_since_epoch()).count();
                    double timestamp_s = static_cast<double>(timestamp_nanos) / 1e9;
                    LOG4CPLUS_DEBUG_FMT(
                        m_logger,
                        "[%s]\tWrote to DP at time (epoch): %f. Iteration: %d.\tWrite "
                        "time [us]: min: %u max: %u  mean: %.2f variance: %.2f",
                        m_dp.ToString().c_str(),
                        timestamp_s,
                        iteration_counter,
                        ba::min(m_accumulator),
                        ba::max(m_accumulator),
                        ba::mean(m_accumulator),
                        ba::variance(m_accumulator));
                    m_accumulator = {};
                }
                iteration_counter++;
            } catch (const std::exception& ex) {
                LOG4CPLUS_WARN_FMT(m_logger,
                                   "[%s ] Iteration: %d exception caught: %s!",
                                   m_dp.ToString().c_str(),
                                   iteration_counter,
                                   ex.what());
                if (error_counter++ > 5) {
                    LOG4CPLUS_ERROR_FMT(
                        m_logger,
                        "[%s] More than 5 consecutive exception caught. Writing to RTR stoped!",
                        m_dp.ToString().c_str());
                    break;
                }  // if
            } catch (...) {
                LOG4CPLUS_WARN_FMT(m_logger,
                                   "[%s ] Iteration: %d unknown exception caught!",
                                   m_dp.ToString().c_str(),
                                   iteration_counter);
                if (error_counter++ > 5) {
                    LOG4CPLUS_ERROR_FMT(
                        m_logger,
                        "[%s] More than 5 consecutive exception caught. Writing to RTR stoped!",
                        m_dp.ToString().c_str());
                    break;
                }  // if
            }
 
            std::this_thread::sleep_for(m_sleep_period);
        }  // while
 
        LOG4CPLUS_INFO_FMT(m_logger,
                           "[%s] %u / %u values have been written. DONE!",
                           m_dp.ToString().c_str(),
                           iteration_counter,
                           m_iteration_num);
        LOG4CPLUS_DEBUG_FMT(m_logger,
                            "[%s]\tWrite "
                            "time [us]: min: %u max: %u med: %.2f variance: %.2f",
                            m_dp.ToString().c_str(),
                            ba::min(m_accumulator),
                            ba::max(m_accumulator),
                            ba::mean(m_accumulator),
                            ba::variance(m_accumulator));
    }  // WriteDataPoint
 
public:
    explicit GenRtrWriter(std::shared_ptr<rtctk_cfw::RuntimeRepoIf>& rtr,
                          const rtctk_cfw::DataPointPath& dp,
                          const std::string& server_alias = "")
        : GenRtrWriterBase(rtr, dp, server_alias) {
    }
 
    virtual ~GenRtrWriter() noexcept {
    }
 
    GenRtrWriter() = delete;
    GenRtrWriter(const GenRtrWriter&) = delete;
    GenRtrWriter(GenRtrWriter&&) = delete;
 
    void Initialize(std::size_t dim_x,
                    std::size_t dim_y,
                    std::chrono::microseconds period,
                    uint32_t iterations) {
        m_sleep_period = period;
        m_iteration_num = iterations;
        m_value = std::make_unique<rtctk_cfw::MatrixBuffer<T>>();
        m_value->resize(dim_x, dim_y);
        LOG4CPLUS_INFO_FMT(m_logger,
                           "[%s] Created matrix of size: %ld x %ld",
                           m_dp.ToString().c_str(),
                           dim_x,
                           dim_y);
        m_payload_bytes = m_value->GetNrows() * m_value->GetNcols() * sizeof(T);
 
        if (!m_rtr->DataPointExists(m_dp)) {
            try {
                LOG4CPLUS_DEBUG_FMT(m_logger,
                                    "[%s] DP does not exist going to create one (server alias=%s).",
                                    m_dp.ToString().c_str(),
                                    m_server_alias.c_str());
                // if server_alias is not provided it means it is "" what means to write to the
                // central OLDB
                m_rtr->CreateDataPoint<rtctk_cfw::MatrixBuffer<T>>(m_dp, *m_value, m_server_alias);
                m_dp_created = true;
            } catch (const std::exception& ex) {
                // we need to check if the DP has been created, and thus the problem probably just
                // to write a value ..
                // ... in such a case we need to delete it
                if (m_rtr->DataPointExists(m_dp)) {
                    LOG4CPLUS_WARN_FMT(m_logger,
                                       "[%s] Going to delete DP after an error.",
                                       m_dp.ToString().c_str());
                    m_rtr->DeleteDataPoint(m_dp);
                }
                CII_THROW_WITH_NESTED(
                    rtctk_cfw::RtctkException,
                    ex,
                    "[" + m_dp.ToString() + "] Problem creating DP and setting inital value");
            }  // try-catch
        } else {
            LOG4CPLUS_DEBUG_FMT(m_logger, "[%s] DP exists.", m_dp.ToString().c_str());
            m_dp_created = false;
        }  // if-else
 
        m_thread = std::thread(&GenRtrWriter::WriteDataPoint, this);
    }  // Initialize
 
};  // GenRtrWriter
 
}  // namespace rtctk::standaloneTools
 
#endif  // RTCTK_GEN_RTR_WRITER_