computation.hpp

Go to the documentation of this file.

#ifndef RTCTK_EXAMPLEDATATASK_COMPUTATION_HPP
#define RTCTK_EXAMPLEDATATASK_COMPUTATION_HPP
 
#include "rtctk/componentFramework/logger.hpp"
#include "rtctk/componentFramework/runtimeRepoApiIf.hpp"
#include "rtctk/componentFramework/oldbApiIf.hpp"
#include "rtctk/componentFramework/matrixBuffer.hpp"
 
#include <string>
#include <vector>
#include <numeric>
#include <chrono>
// #include <cblas.h>
 
using namespace rtctk::componentFramework;
using rtctk::componentFramework::MatrixBuffer;
using rtctk::componentFramework::GetLogger;
 
namespace rtctk::exampleDataTask {
 
template<class TopicType>
class Computation
{
    public:
        Computation(rtctk::componentFramework::RuntimeRepoApiIf& rtr, rtctk::componentFramework::OldbApiIf& oldb, std::string const& name)
        : m_rtr(rtr)
        , m_oldb(oldb)
        , m_name(name)
        , m_dp_slopes("/common/static/wfs_1/slopes")
        , m_dp_modes("/" + m_name + "/" + "static/computation/modes")
        , m_dp_iterations("/" + m_name + "/" + "static/common/samples_to_read")
        , m_dp_s2m("/" + m_name + "/" + "dynamic/computation/slopes2modes")
        , m_dp_avg_slopes("/common/dynamic/wfs_1/avg_slopes")
        , m_dp_avg_modes("/common/dynamic/wfs_1/avg_modes")
        , m_dp_stats_iteration("/" + m_name + "/" + "statistics/computation/iteration")
        , m_dp_stats_time("/" + m_name + "/" + "statistics/computation/time")
        , m_dp_stats_last_id("/" + m_name + "/" + "statistics/computation/last_sample_id")
        {
            using rtctk::componentFramework::GetLogger;
            m_sample_id = 0;
            m_callback_counter = 0;
            m_iteration = 0;
 
            // create some oldb points
            m_oldb.CreateDataPoint(m_dp_stats_iteration, "RtcInt32");
            m_oldb.CreateDataPoint(m_dp_stats_time, "RtcDouble");
            m_oldb.CreateDataPoint(m_dp_stats_last_id, "RtcInt32");
            m_oldb.SetDataPoint<int32_t>(m_dp_stats_iteration, 0);
            m_oldb.SetDataPoint<double>(m_dp_stats_time, 0.0);
            m_oldb.SetDataPoint<int32_t>(m_dp_stats_last_id, 0);
            // load static config for the first and only time after this onlt DynamicLoad can be called.
            StaticLoad();
        }
 
        void StaticLoad()
        {
            // read anything that is unlikely to change.
            m_n_slopes = m_rtr.GetDataPoint<int32_t>(m_dp_slopes);
            m_n_modes = m_rtr.GetDataPoint<int32_t>(m_dp_modes);
            m_n_iterations = m_rtr.GetDataPoint<int32_t>(m_dp_iterations);
            
            // set vectors and matrix to correct size.
            m_buffer_matrix.resize(m_n_iterations, m_n_slopes);
            m_avg_slopes.resize(m_n_slopes);
            m_s2m_matrix.resize(m_n_modes, m_n_slopes);
            m_avg_modes.resize(m_n_modes);
 
            // fill with zeros where needed.
            std::fill (m_avg_slopes.begin(), m_avg_slopes.end(), 0.0); 
            std::fill (m_avg_modes.begin(), m_avg_modes.end(), 0.0);
 
            // some simple debug output
            LOG4CPLUS_DEBUG(GetLogger(), "m_n_slopes: " << m_n_slopes);
            LOG4CPLUS_DEBUG(GetLogger(), "m_n_modes: " << m_n_modes);
            LOG4CPLUS_DEBUG(GetLogger(), "m_n_iterations: " << m_n_iterations);
 
            // cause the DynamicLoad
            DynamicLoad();
        }
 
        void DynamicLoad()
        {
             // read things that will change
            m_s2m_matrix = m_rtr.GetDataPoint<MatrixBuffer<float>>(m_dp_s2m);
            if(m_n_modes != (int) m_s2m_matrix.get_nrows() || m_n_slopes != (int) m_s2m_matrix.get_ncols())
            {
                LOG4CPLUS_INFO(GetLogger(), "Computation::DynamicLoad() - s2m wrong shape ");
                LOG4CPLUS_INFO(GetLogger(), "Computation::DynamicLoad() - Expected: " << m_n_modes << " x " << m_n_slopes);
                LOG4CPLUS_INFO(GetLogger(), "Computation::DynamicLoad() - recieved: " << m_s2m_matrix.get_nrows() << " x " << m_s2m_matrix.get_ncols());
                throw std::runtime_error("s2m wrong shape");
            }
            assert((int)m_s2m_matrix.get_nrows() == m_n_modes);
            assert((int)m_s2m_matrix.get_ncols() == m_n_slopes);
        }
 
        void OnDataAvailable(TopicType const& sample) // TODO: should this be nothrow?
        {
            m_sample_id = sample.sample_id;
            // copy vector into matrix position
            std::copy(sample.wfs.slopes.begin(), sample.wfs.slopes.end(), m_buffer_matrix.begin() + m_callback_counter*m_n_slopes);
            m_callback_counter++;
        }
 
        void Reset()
        {
            // This is invoked going to idle to clean up incase of stopped during buffering of data.
            m_iteration = 0;
            ResetComputation();
        }
 
        void Compute()
        {
            using rtctk::componentFramework::GetLogger;
            m_computation_running = true;
            LOG4CPLUS_INFO(GetLogger(), "Computation::Compute() - samples: " << m_callback_counter);
            m_iteration++;
            // perform computation on the data buffers
            auto time_start = std::chrono::system_clock::now();
            // average matrix to avg_array this is overly simplistic and not optimised
            for(int i = 0 ; i < m_n_iterations; i++)
            {
                for(int j = 0; j < m_n_slopes; j++)
                {
                    m_avg_slopes[j] += m_buffer_matrix(i,j);
                }
            }
 
            for(int j = 0; j < m_n_slopes; j++)
            {
               m_avg_slopes[j] =  m_avg_slopes[j]/static_cast<float>(m_n_iterations);
            }
 
            // multiply avg_array with S2M matrix to project the average slopes onto modes.
            //TODO: openblas not available on Centos 7 and no point in making it a dependancy.
            //      for this example I have added a very basic matrix vector multiplication of my own
            //      It is single threaded and not optimised. It is here only for demonstration purposes.
            // cblas_sgemv(CblasRowMajor, CblasNoTrans, m_n_modes, m_n_slopes, 1.0, m_s2m_matrix.data(), m_n_slopes, m_avg_slopes.data(), 1, 0.0, m_avg_modes.data(), 1);
            my_sgemv(m_s2m_matrix.data(), m_avg_slopes.data(), m_avg_modes.data(), m_n_modes, m_n_slopes);
            auto time_stop = std::chrono::system_clock::now();
            std::chrono::duration<double> elapsed = time_stop - time_start;
 
            // write output back to runtime repo
            m_rtr.SetDataPoint<std::vector<float>>(m_dp_avg_slopes, m_avg_slopes);
            m_rtr.SetDataPoint<std::vector<float>>(m_dp_avg_modes, m_avg_modes);
 
            // write statistics:
            m_oldb.SetDataPoint<int32_t>(m_dp_stats_iteration, m_iteration);
            m_oldb.SetDataPoint<double>(m_dp_stats_time, elapsed.count());
            m_oldb.SetDataPoint<int32_t>(m_dp_stats_last_id, m_sample_id);
 
            ResetComputation(); // clean up computation and reset some values
            m_computation_running = false;
        }
 
        bool isComputing(){return m_computation_running;}
 
    protected:
 
        void ResetComputation()
        {
            // called at end of computation to or duing idle to put
            m_callback_counter = 0;
        }
 
 
        // single threaded sgemv for row major matrix
        void my_sgemv(const float * mat, const float * in_vec, float * out_vec, uint32_t rows, uint32_t cols)
        {
            for (uint32_t i = 0; i < rows; ++i)
            {
                out_vec[i] = my_dot_product(mat + i*cols, in_vec, cols);
            }
        }
        // single threaded dot product
        float my_dot_product(const float *x, const float *y, uint32_t n)
        {
            double res = 0.0;
            for (uint32_t i = 0; i < n; ++i)
            {
                res += x[i] * y[i];
            }
            return res;
        }
 
        rtctk::componentFramework::RuntimeRepoApiIf& m_rtr;
        rtctk::componentFramework::OldbApiIf& m_oldb;
        std::string m_name;
 
        // inputs
        DataPointPath m_dp_slopes;
        DataPointPath m_dp_modes;
        DataPointPath m_dp_iterations;
        DataPointPath m_dp_s2m;
        
        //outputs to write
        DataPointPath m_dp_avg_slopes;
        DataPointPath m_dp_avg_modes;
 
        // write statistics:
        DataPointPath m_dp_stats_iteration;
        DataPointPath m_dp_stats_time;
        DataPointPath m_dp_stats_last_id;
 
        unsigned int m_sample_id;
        unsigned int m_iteration;
        std::atomic<unsigned int> m_callback_counter;
        std::atomic<bool> m_computation_running;
        
        int m_n_slopes;
        int m_n_modes;
        int m_n_iterations;
 
        std::vector<float> m_avg_slopes;
        std::vector<float> m_avg_modes;
 
        MatrixBuffer<float> m_buffer_matrix;
        MatrixBuffer<float> m_s2m_matrix;
 
};
 
} // namespace
 
#endif