Pipeline.hpp

Go to the documentation of this file.

 
#ifndef H_POTTU_PIPELINE
#define H_POTTU_PIPELINE
 
#include "DataSource.hpp"
#include "StageRawToStampedRaw.hpp"
#include "StageStampedRawToDetectorEvent.hpp"
#include "StageDetectorEvent.hpp"
#include "StageDetectorEventFinal.hpp"
#include "StageDetectorEventToDetectorEvent.hpp"
#include "ContextBase.hpp"
 
#if POTTU_ENABLE_MULTITHREADING == 1
#include <taskflow/taskflow.hpp>
#include <taskflow/algorithm/pipeline.hpp>
#endif
 
#if POTTU_ROOT_SUPPORT == 1
#include <TROOT.h>
#endif
 
#include <vector>
#include <memory>
#include <variant>
#include <thread>
 
//#include <iostream>
 
 
namespace pottu {
 
    class Pipeline {
    public:
    
 
    void setSource( DataSource *stage ) noexcept {
        _source = stage;
        _builtPipeline.reset();
    }
 
    
    void setTimeStamper( StageRawToStampedRaw *stage ) noexcept {
        _stamper = stage;
        _builtPipeline.reset();
    }
 
    
    void setDetectorEventGenerator( StageStampedRawToDetectorEvent *stage ) noexcept {
        _deteventgen = stage;
        _builtPipeline.reset();
    }
 
    
    void add( StageDetectorEvent *stage ) noexcept {
        _stagesRest.emplace_back( stage );
        _builtPipeline.reset();
    }
 
    
    void add( StageDetectorEventToDetectorEvent *stage ) noexcept {
        _stagesRest.emplace_back( stage );
        _builtPipeline.reset();
    }
 
 
    void add( StageDetectorEventFinal *stage ) noexcept {
        _stagesRest.emplace_back( stage );
        _builtPipeline.reset();
    }
 
 
 
    void run( std::size_t num_threads = 1 ) {
        if( !_source || !_stamper || !_deteventgen )
        throw std::runtime_error( "Pipeline: Source, stamper or detector event generator is not set!" );
 
        if( num_threads == 0 ) {
        num_threads = std::thread::hardware_concurrency();
        if( num_threads == 0 )
            num_threads = 4;
        }
 
#if POTTU_ENABLE_MULTITHREADING != 1
        if( num_threads > 1 )
        throw std::runtime_error( "Asking more than one thread but no multithreading enabled."
                      "Add: target_compile_definitions( yoursort PRIVATE POTTU_ENABLE_MULTITHREADING=1 )"
                      "into the CMakeLists.txt file.");
#endif
 
        if( !_builtPipeline || _builtPipeline->lineBuffers.size() != num_threads ) {
        if( num_threads == 1 )
            _buildSingleThreadedPipeline();
        else 
            _buildMultiThreadedPipeline( num_threads );
        }
 
        if( num_threads == 1 ) {
        _runSingleThreaded();
        } else {
#if POTTU_ENABLE_MULTITHREADING == 1
        ContextBase::getActive().logInfo( "!!! Running now multithreaded pipeline!" );
 
        //_builtPipeline->taskflow.dump(std::cout);
 
        _builtPipeline->executor.run(_builtPipeline->taskflow).wait();
#endif
        }
    }
 
 
    void _buildSingleThreadedPipeline() {
        _builtPipeline.reset();
        _builtPipeline = std::move( std::make_unique<BuiltPipeline>() );
        _builtPipeline->lineBuffers.resize(1);
 
        auto &detevents = _builtPipeline->lineBuffers[0].detevents;
        // Counting the number of stages that has separate output
        std::size_t sepOutN{0};
        for( auto &vstage : _stagesRest ) {
        if( vstage.index() == 1 || vstage.index() == 2 )
            ++sepOutN;
        }
        detevents.resize( sepOutN+1 );
 
        ContextBase::getActive().logInfo(
        fmt::format( "Singlethreaded pipeline created with {} stages and {} buffers.",
                 3+_stagesRest.size(), 3+detevents.size() ) );
    }
    
    
    void _buildMultiThreadedPipeline( std::size_t num_lines ) {
 
#if POTTU_ROOT_SUPPORT == 1
        ROOT::EnableThreadSafety();
#endif
        
#if POTTU_ENABLE_MULTITHREADING == 1
        _builtPipeline.reset();
        _builtPipeline = std::move( std::make_unique<BuiltPipeline>() );
        _builtPipeline->lineBuffers.reserve(num_lines);
        _builtPipeline->lineBuffers.resize(1);
 
        // Creating the buffers for the rest of the stages for _the first line_.
        // Counting the number of stages that has separate output
        std::size_t sepOutN{0};
        for( auto &vstage : _stagesRest ) {
        if( vstage.index() == 1 || vstage.index() == 2 )
            ++sepOutN;
        }
        _builtPipeline->lineBuffers[0].detevents.resize( sepOutN+1 );
 
        // Copying the above buffers for the rest of the lines
        for( std::size_t line = 1; line < num_lines; ++line )
        _builtPipeline->lineBuffers.push_back( _builtPipeline->lineBuffers[0] );
 
 
        // Creating tf::pipe's. 
        _builtPipeline->pipes.emplace_back(
        tf::PipeType::SERIAL,
        [this](tf::Pipeflow &pf)
            {
            if( _source->allDone() )
                pf.stop();
            else
                _source->process( _builtPipeline->lineBuffers[pf.line()].rawitems );
            } );
        _builtPipeline->pipes.emplace_back(
        tf::PipeType::SERIAL,
        [this](tf::Pipeflow &pf)
            {
            _stamper->process( _builtPipeline->lineBuffers[pf.line()].rawitems,
                       _builtPipeline->lineBuffers[pf.line()].rawitems_ts );
            } );
        _builtPipeline->pipes.emplace_back(
        tf::PipeType::SERIAL,
        [this](tf::Pipeflow &pf)
            {
            _deteventgen->process( _builtPipeline->lineBuffers[pf.line()].rawitems_ts,
                           _builtPipeline->lineBuffers[pf.line()].detevents[0] );
            } );
        
        std::size_t bufc{0};
        for( auto &vstage : _stagesRest ) {
        switch( vstage.index() ) {
        case 0: {
            auto p = std::get<StageDetectorEvent *>( vstage );
            
            _builtPipeline->pipes.emplace_back(
            tf::PipeType::SERIAL,
            [this,p,bufc](tf::Pipeflow &pf)
                {
                p->process( _builtPipeline->lineBuffers[pf.line()].detevents[bufc] );
                } );
            break;
        }
        case 1: {
            auto p = std::get<StageDetectorEventToDetectorEvent *>( vstage );
            
            _builtPipeline->pipes.emplace_back(
            tf::PipeType::SERIAL,
            [this,p,bufc](tf::Pipeflow &pf)
                {
                p->process( _builtPipeline->lineBuffers[pf.line()].detevents[bufc],
                               _builtPipeline->lineBuffers[pf.line()].detevents[bufc+1] );
                } );
            bufc++;
            break;
        }
        case 2: {
            auto p = std::get<StageDetectorEventFinal *>( vstage );
            
            _builtPipeline->pipes.emplace_back(
            tf::PipeType::SERIAL,
            [this,p,bufc](tf::Pipeflow &pf)
                {
                // Copying the input data to next buffer before consuming it
                _builtPipeline->lineBuffers[pf.line()].detevents[bufc+1] = _builtPipeline->lineBuffers[pf.line()].detevents[bufc];
                p->process( _builtPipeline->lineBuffers[pf.line()].detevents[bufc] );
                // Zeroing the input data so that it cannot be used anymore
                _builtPipeline->lineBuffers[pf.line()].detevents[bufc].clear();
                } );
            bufc++;
            break;
        }
        }
        
        }
        _builtPipeline->pl.reset( num_lines, _builtPipeline->pipes.begin(), _builtPipeline->pipes.end() );
        _builtPipeline->taskflow.composed_of(_builtPipeline->pl);
        ContextBase::getActive().logInfo(
        fmt::format( "Multithreaded pipeline created with {} stages, {} lines and {} buffers/line.",
                 3+_stagesRest.size(), _builtPipeline->lineBuffers.size(),
                 3+_builtPipeline->lineBuffers[0].detevents.size() ) );
#endif      
    }
 
 
    void _runSingleThreaded() {
        auto &buffers = _builtPipeline->lineBuffers[0];
        while( !_source->allDone() ) {
        _source->process( buffers.rawitems );
        _stamper->process( buffers.rawitems, buffers.rawitems_ts );
        _deteventgen->process( buffers.rawitems_ts, buffers.detevents[0] );
        std::size_t bufc{0};
        for( auto &vstage : _stagesRest ) {
            switch( vstage.index() ) {
            case 0:
            std::get<StageDetectorEvent *>( vstage )
                ->process( buffers.detevents[bufc] );
            break;
            case 1:
            std::get<StageDetectorEventToDetectorEvent *>( vstage )
                ->process( buffers.detevents[bufc], buffers.detevents[bufc+1] );
            bufc++;
            break;
            case 2:
            // Copying the input data to next buffer before consuming it
            buffers.detevents[bufc+1] = buffers.detevents[bufc];
            std::get<StageDetectorEventFinal *>( vstage )
                ->process( buffers.detevents[bufc] );
            bufc++;
            break;
            }
        }
 
        }
    }
 
 
    
    struct BuiltPipeline {
        
        struct LineBuffers {
        std::vector<dataitem_t>                   rawitems;
        std::vector<dataitem_ts_t>                rawitems_ts;
        std::vector< std::vector<DetectorEvent> > detevents;
        };
 
        std::vector<LineBuffers> lineBuffers;
#if POTTU_ENABLE_MULTITHREADING == 1
        using pipe_type = tf::Pipe<std::function<void(tf::Pipeflow&)>>;
        tf::Taskflow                        taskflow;
        tf::Executor                        executor;
        std::vector< pipe_type >            pipes;
        tf::ScalablePipeline<std::vector<pipe_type>::iterator> pl;
#endif      
    };
 
    
 
    DataSource                     *_source{nullptr};
    StageRawToStampedRaw           *_stamper{nullptr};
    StageStampedRawToDetectorEvent *_deteventgen{nullptr};
    std::vector<
        std::variant< StageDetectorEvent *,
              StageDetectorEventToDetectorEvent *,
              StageDetectorEventFinal *>>  _stagesRest;
    
    std::unique_ptr<BuiltPipeline>  _builtPipeline;
    
    };
 
    
}
 
#endif