businessLogicGpu.hpp

Go to the documentation of this file.

#ifndef RTCTK_EXAMPLEDATATASK_BUSINESSLOGIC_GPU_HPP
#define RTCTK_EXAMPLEDATATASK_BUSINESSLOGIC_GPU_HPP
 
#include "rtctk/componentFramework/logger.hpp"
#include "rtctk/componentFramework/runnableStateMachineLogic.hpp"
#include "rtctk/componentFramework/serviceContainer.hpp"
#include "rtctk/componentFramework/runtimeRepoApiIf.hpp"
#include "rtctk/componentFramework/oldbApiIf.hpp"
#include "rtctk/dataTask/readerThread.hpp"
 
#include <ipcq/reader.hpp>
 
#include <chrono>
#include <iostream>
#include <vector>
#include <thread>
 
using namespace rtctk::componentFramework;
using namespace rtctk::dataTask;
using namespace std::chrono_literals;
 
namespace rtctk::exampleDataTask {
 
// TODO: consider removing template arguments here, since users need to edit BusinessLogicGpu anyway
template <class ComputationType, class TopicType>
class BusinessLogicGpu : public RunnableStateMachineLogic
{
    using ReaderType = ipcq::Reader<TopicType>;
 
    public:
        BusinessLogicGpu(const std::string& name, rtctk::componentFramework::ServiceContainer& services)
            : RunnableStateMachineLogic(name, services)
            , m_rtr(m_services.Get<RuntimeRepoApiIf>())
            , m_oldb(m_services.Get<OldbApiIf>())
        {
            LOG4CPLUS_DEBUG(GetLogger(), "ExampleDataTask c'tor");
        }
 
        virtual ~BusinessLogicGpu()
        {
            LOG4CPLUS_INFO(GetLogger(), "ExampleDataTask d'tor");
        }
 
        // activities
        void Starting(StopToken st) override
        {
            m_reader.reset();
            m_computation.reset();
        }
 
        void Recovering(StopToken st) override
        {
        }
 
        void Initialising(StopToken st) override
        {
            // create the user computation object
            m_computation = std::make_unique<ComputationType>(m_rtr, m_oldb, m_name);
            m_reader = std::make_unique<ReaderThread<TopicType,ReaderType>>();
 
            // register callbacks
            m_reader->RegisterOnDataCallback( [this] (const TopicType& sample) {m_computation->OnDataAvailable(sample);});
            m_reader->RegisterInitThreadCallback([this](){m_computation->ThreadInit();});
 
            // static
            m_reader->SetQueueName(GetParam<std::string>("/common/static/loop_1/queue_name"));
            m_reader->SetCpuAffinity(GetParam<int32_t>("/" + m_name + "/static/reader_thread/cpu_affinity"));
            m_reader->SetLoopFrequency(GetParam<int32_t>("/common/static/loop_1/frequency"));
 
            // dynamic
            m_reader->SetSamplesToRead(GetParam<int32_t>("/" + m_name + "/static/common/samples_to_read"));
            m_reader->SetSamplesToSkip(GetParam<int32_t>("/" + m_name + "/dynamic/reader_thread/samples_to_skip"));
        }
 
        void Enabling(StopToken st) override
        {
            m_reader->Spawn();
        }
 
        void Disabling(StopToken st) override
        {
            m_reader->Join();
        }
 
        void GoingRunning(StopToken st) override
        {
            m_computation->Reset();
            m_reader->Run();
        }
 
        void GoingIdle(StopToken st) override
        {
            m_reader->Idle();
        }
 
        void Running(StopToken st) override
        {
            // wait for signal from read thread to trigger computation
            while(true)
            {
                if(st.StopRequested()){break;}
 
                m_reader->WaitUntilComputationAllowed();
 
                if(st.StopRequested()){break;} // TODO: do we need this?
 
                m_computation->Compute();
 
                m_reader->SignalComputationDone();
            }
 
            m_reader->SignalComputationDone();
        }
 
        void Updating(StopToken st, Payload args) override
        {
            m_reader->SetSamplesToSkip(GetParam<int32_t>("/" + m_name + "/dynamic/reader_thread/samples_to_skip"));
            m_computation->DynamicLoad();
        }
 
        // guards
        bool IsUpdatingAllowed(Payload args) override
        {
            if(m_computation->isComputing())
            {
                return false;
            }
            else 
            {
                return true;
            }
        }
 
    protected:
 
        template<typename T>
        T GetParam(std::string const& path)
        {
            DataPointPath dp_uri(path);
            auto value = m_rtr.GetDataPoint<T>(dp_uri);
            return value;
        }
 
        RuntimeRepoApiIf& m_rtr;
        OldbApiIf& m_oldb;
 
        std::unique_ptr<ComputationType> m_computation;
        std::unique_ptr<ReaderThread<TopicType,ReaderType>> m_reader;
};
 
} //namespace
 
#endif