ParallelSequentialSchedulerSvc.h

Go to the documentation of this file.

#ifndef GAUDIHIVE_PARALLELSEQUENTIALSCHEDULERSVC_H
#define GAUDIHIVE_PARALLELSEQUENTIALSCHEDULERSVC_H

// Framework include files
#include "GaudiKernel/IScheduler.h"
#include "GaudiKernel/IRunable.h"
#include "GaudiKernel/Service.h"
#include "GaudiKernel/IAlgResourcePool.h"
#include "GaudiKernel/CommonMessaging.h"
#include "GaudiKernel/EventContext.h"

#include "AlgResourcePool.h"
#include "ExecutionFlowManager.h"
#include "DataFlowManager.h"

// C++ include files
#include <vector>
#include <string>
#include <unordered_map>
#include <functional>
#include <thread>

// External libs
#include "tbb/concurrent_queue.h"
#include "tbb/task.h"
#include "tbb/task_scheduler_init.h"

class SequentialTask; //forward declaration


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

class ParallelSequentialSchedulerSvc: public extends1<Service, IScheduler> {
public:
  ParallelSequentialSchedulerSvc( const std::string& name, ISvcLocator* svc );

  ~ParallelSequentialSchedulerSvc();

  virtual StatusCode initialize();

  virtual StatusCode finalize();

  virtual StatusCode pushNewEvent(EventContext* eventContext);

  // Make multiple events available to the scheduler
  virtual StatusCode pushNewEvents(std::vector<EventContext*>& eventContexts);

  virtual StatusCode popFinishedEvent(EventContext*& eventContext);

  virtual StatusCode tryPopFinishedEvent(EventContext*& eventContext);

  virtual unsigned int freeSlots();


private:

  bool m_useTopAlgList;

  std::list<IAlgorithm*> m_algList;

  tbb::concurrent_bounded_queue<EventContext*> m_finishedEvents;

  // number of events in flight
  std::atomic_int m_freeSlots;

  SmartIF<IHiveWhiteBoard> m_whiteboard;

  std::string m_whiteboardSvcName;

  SmartIF<IAlgResourcePool>  m_algResourcePool;

  int m_threadPoolSize;

  //TBB scheduler
  std::unique_ptr<tbb::task_scheduler_init> m_tbb_sched;

  //control flow manager
  concurrency::ExecutionFlowManager m_controlFlow;

  std::vector<std::string> m_algname_vect;

  std::unordered_map<std::string,unsigned int> m_algname_index_map;

  std::vector<std::vector<std::string>> m_algosDependencies;

  // Needed to queue actions on algorithm finishing and decrement algos in flight
  friend class SequentialTask;

};

class SequentialTask: public tbb::task {
public:
  SequentialTask(ISvcLocator* svcLocator,
                 EventContext* eventContext,
                 ParallelSequentialSchedulerSvc* scheduler,
                 IAlgResourcePool* algPool):

                   m_serviceLocator(svcLocator),
                   m_eventContext(eventContext),
                   m_scheduler(scheduler),
                   m_algPool(algPool){

  };
  virtual tbb::task* execute();
private:
  SmartIF<ISvcLocator> m_serviceLocator;
  EventContext* m_eventContext;
  SmartIF<ParallelSequentialSchedulerSvc> m_scheduler;
  SmartIF<IAlgResourcePool> m_algPool;
};

#endif // GAUDIHIVE_PARALLELSEQUENTIALSCHEDULERSVC_H