PEPScheduler.h

Go to the documentation of this file.

#ifndef MARLINMT_CONCURRENCY_PEPSCHEDULER_h
#define MARLINMT_CONCURRENCY_PEPSCHEDULER_h 1

// -- marlinmt headers
#include <marlinmt/IScheduler.h>
#include <marlinmt/Logging.h>
#include <marlinmt/Utils.h>
#include <marlinmt/concurrency/ThreadPool.h>

// -- std headers
#include <unordered_set>

namespace marlinmt {

  class SuperSequence ;
  class Sequence ;

  namespace concurrency {

    struct WorkerOutput {
      std::shared_ptr<EventStore>         _event {nullptr} ;
      std::exception_ptr                  _exception {nullptr} ;
    };

    //--------------------------------------------------------------------------
    //--------------------------------------------------------------------------

    class PEPScheduler : public IScheduler {
    public:
      using ConditionsMap = std::map<std::string, std::string> ;
      using InputType = std::shared_ptr<EventStore> ;
      using OutputType = WorkerOutput ;
      using WorkerPool = ThreadPool<InputType,OutputType> ;
      using ProcessorSequence = std::shared_ptr<SuperSequence> ;
      using PushResultList = std::vector<WorkerPool::PushResult> ;
      using EventList = std::vector<std::shared_ptr<EventStore>> ;
      using Clock = std::chrono::steady_clock ;
      using TimePoint = std::chrono::steady_clock::time_point ;

    public:
      PEPScheduler() ;

      // from IScheduler interface
      void initialize() override ;
      void end() override ;
      void processRunHeader( std::shared_ptr<RunHeader> rhdr ) override ;
      void pushEvent( std::shared_ptr<EventStore> event ) override ;
      void popFinishedEvents( std::vector<std::shared_ptr<EventStore>> &events ) override ;
      std::size_t freeSlots() const override ;

    private:
      void preConfigure() ;
      void configureProcessors() ;
      void configurePool() ;

    private:
      WorkerPool                       _pool {} ;
      ProcessorSequence                _superSequence {nullptr} ;
      PushResultList                   _pushResults {} ;
      clock::time_point                _startTime {} ;
      clock::time_point                _endTime {} ;
      clock::duration_rep              _runHeaderTime {0} ;
      clock::duration_rep              _lockingTime {0} ;
      clock::duration_rep              _popTime {0} ;
      UIntParameter                    _queueSize {*this, "EventQueueSize", "The input event queue size (default 2*nthreads)"} ;
    };

  }

} // end namespace marlinmt

#endif