ForwardSchedulerSvc.h

Go to the documentation of this file.

#ifndef GAUDIHIVE_FORWARDSCHEDULERSVC_H
#define GAUDIHIVE_FORWARDSCHEDULERSVC_H

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

// Local includes
#include "AlgsExecutionStates.h"
#include "EventSlot.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"

typedef AlgsExecutionStates::State State;

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

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

  ~ForwardSchedulerSvc();

  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:

  // Utils and shortcuts ----------------------------------------------------

  void activate();

  StatusCode deactivate();

  bool m_isActive;

  std::thread m_thread;

  inline unsigned int algname2index(const std::string& algoname);

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

  inline const std::string& index2algname (unsigned int index);

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

  SmartIF<IHiveWhiteBoard> m_whiteboard;

  std::string m_whiteboardSvcName;

  std::vector<EventSlot> m_eventSlots;

  int m_maxEventsInFlight;

  std::atomic_int m_freeSlots;

  tbb::concurrent_bounded_queue<EventContext*> m_finishedEvents;

  StatusCode eventFailed(EventContext* eventContext);


  // States management ------------------------------------------------------

  unsigned int m_maxAlgosInFlight;

  unsigned int m_algosInFlight;

  StatusCode updateStates(int si=-1, const std::string& algo_name=std::string());

  StatusCode promoteToControlReady(unsigned int iAlgo, int si);
  StatusCode promoteToDataReady(unsigned int iAlgo, int si);
  StatusCode promoteToScheduled(unsigned int iAlgo, int si);
  StatusCode promoteToExecuted(unsigned int iAlgo, int si, IAlgorithm* algo);
  StatusCode promoteToFinished(unsigned int iAlgo, int si);

  StatusCode isStalled(int si);

  void dumpSchedulerState(int iSlot);

  bool m_updateNeeded;

  // Algos Management -------------------------------------------------------
  SmartIF<IAlgResourcePool>  m_algResourcePool;

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

  StatusCode m_drain();

  int m_threadPoolSize;

  // Actions management -----------------------------------------------------

  typedef std::function<StatusCode ()> action;

  tbb::concurrent_bounded_queue<action> m_actionsQueue;

  concurrency::ExecutionFlowManager m_efManager;
  // XXX: CF tests. Temporary property to switch between ControlFlow implementations
  bool m_CFNext;
  // XXX: CF tests. Temporary property to switch between DataFlow implementations
  bool m_DFNext;
  // Flag to perform single-pass simulation of execution flow before the actual execution
  bool m_simulateExecution;
  // Optimization mode in which algorithms, ready for execution, are prioritized in special way
  std::string m_optimizationMode;
  // Dump intra-event concurrency dynamics to csv file
  bool m_dumpIntraEventDynamics;

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

};

#endif // GAUDIHIVE_FORWARDSCHEDULERSVC_H