ForwardSchedulerSvc.h

Go to the documentation of this file.

#ifndef GAUDIHIVE_FORWARDSCHEDULERSVC_H
#define GAUDIHIVE_FORWARDSCHEDULERSVC_H

#include <functional>
#include <string>
#include <thread>
#include <unordered_map>
#include <vector>

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

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

// Local includes
#include "AlgsExecutionStates.h"
#include "DataFlowManager.h"
#include "EventSlot.h"
#include "ExecutionFlowManager.h"

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

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

typedef AlgsExecutionStates::State State;

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

class ForwardSchedulerSvc : public extends<Service, IScheduler>
{
public:
  using extends::extends;

  ~ForwardSchedulerSvc() override = default;

  StatusCode initialize() override;

  StatusCode finalize() override;

  StatusCode pushNewEvent( EventContext* eventContext ) override;

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

  StatusCode popFinishedEvent( EventContext*& eventContext ) override;

  StatusCode tryPopFinishedEvent( EventContext*& eventContext ) override;

  unsigned int freeSlots() override;

private:
  enum ActivationState { INACTIVE = 0, ACTIVE = 1, FAILURE = 2 };

  Gaudi::Property<int> m_maxEventsInFlight{this, "MaxEventsInFlight", 0,
                                           "Maximum number of event processed simultaneously"};
  Gaudi::Property<int> m_threadPoolSize{
      this, "ThreadPoolSize", -1,
      "Size of the threadpool initialised by TBB; a value of -1 gives TBB the freedom to choose"};
  Gaudi::Property<std::string> m_whiteboardSvcName{this, "WhiteboardSvc", "EventDataSvc", "The whiteboard name"};
  Gaudi::Property<std::string> m_IOBoundAlgSchedulerSvcName{this, "IOBoundAlgSchedulerSvc", "IOBoundAlgSchedulerSvc"};
  Gaudi::Property<unsigned int> m_maxAlgosInFlight{this, "MaxAlgosInFlight", 1,
                                                   "[[deprecated]] Taken from the whiteboard"};

  Gaudi::Property<unsigned int> m_maxIOBoundAlgosInFlight{this, "MaxIOBoundAlgosInFlight", 0,
                                                          "Maximum number of simultaneous I/O-bound algorithms"};
  // XXX: CF tests. Temporary property to switch between ControlFlow implementations
  Gaudi::Property<bool> m_CFNext{this, "useGraphFlowManagement", false,
                                 "Temporary property to switch between ControlFlow implementations"};
  // XXX: CF tests. Temporary property to switch between DataFlow implementations
  Gaudi::Property<bool> m_DFNext{this, "DataFlowManagerNext", false,
                                 "Temporary property to switch between DataFlow implementations"};
  Gaudi::Property<bool> m_simulateExecution{
      this, "SimulateExecution", false,
      "Flag to perform single-pass simulation of execution flow before the actual execution"};
  Gaudi::Property<std::string> m_optimizationMode{this, "Optimizer", "",
                                                  "The following modes are currently available: PCE, COD, DRE,  E"};
  Gaudi::Property<bool> m_dumpIntraEventDynamics{this, "DumpIntraEventDynamics", false,
                                                 "Dump intra-event concurrency dynamics to csv file"};
  Gaudi::Property<bool> m_useIOBoundAlgScheduler{this, "PreemptiveIOBoundTasks", false,
                                                 "Turn on preemptive way of scheduling of I/O-bound algorithms"};
  Gaudi::Property<std::vector<std::vector<std::string>>> m_algosDependencies{
      this, "AlgosDependencies", {}, "[[deprecated]]"};
  Gaudi::Property<bool> m_checkDeps{this, "CheckDependencies", false, "[[deprecated]]"};

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

  void activate();

  StatusCode deactivate();

  std::atomic<ActivationState> m_isActive{INACTIVE};

  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;

  SmartIF<IAccelerator> m_IOBoundAlgScheduler;

  std::vector<EventSlot> m_eventSlots;

  std::atomic_int m_freeSlots;

  tbb::concurrent_bounded_queue<EventContext*> m_finishedEvents;

  StatusCode eventFailed( EventContext* eventContext );

  SmartIF<IAlgExecStateSvc> m_algExecStateSvc;

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

  unsigned int m_algosInFlight = 0;

  unsigned int m_IOBoundAlgosInFlight = 0;

  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 promoteToAsyncScheduled( unsigned int iAlgo, int si ); // tests of an asynchronous scheduler
  StatusCode promoteToExecuted( unsigned int iAlgo, int si, IAlgorithm* algo, EventContext* );
  StatusCode promoteToAsyncExecuted( unsigned int iAlgo, int si, IAlgorithm* algo,
                                     EventContext* ); // tests of an asynchronous scheduler
  StatusCode promoteToFinished( unsigned int iAlgo, int si );

  StatusCode isStalled( int si );

  void dumpSchedulerState( int iSlot );

  bool m_updateNeeded = true;

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

  StatusCode m_drain();

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

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

  tbb::concurrent_bounded_queue<action> m_actionsQueue;

  concurrency::ExecutionFlowManager m_efManager;

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

  // Service for thread pool initialization
  SmartIF<IThreadPoolSvc> m_threadPoolSvc;

  bool m_first = true;

  class SchedulerState
  {

  public:
    SchedulerState( Algorithm* a, EventContext* e, pthread_t t ) : m_a( a ), m_e( *e ), m_t( t ) {}

    Algorithm* alg() const { return m_a; }
    EventContext ctx() const { return m_e; }
    pthread_t thread() const { return m_t; }

    friend std::ostream& operator<<( std::ostream& os, const SchedulerState& ss )
    {
      os << ss.ctx() << "  a: " << ss.alg()->name() << " [" << std::hex << ss.alg() << std::dec << "]  t: 0x"
         << std::hex << ss.thread() << std::dec;
      return os;
    }

    bool operator==( const SchedulerState& ss ) const { return ( m_a == ss.alg() ); }

    bool operator==( Algorithm* a ) const { return ( m_a == a ); }

    bool operator<( const SchedulerState& rhs ) const { return ( m_a < rhs.alg() ); }

  private:
    Algorithm* m_a;
    EventContext m_e;
    pthread_t m_t;
  };

  static std::list<SchedulerState> m_sState;
  static std::mutex m_ssMut;

public:
  void addAlg( Algorithm*, EventContext*, pthread_t );
  bool delAlg( Algorithm* );
  void dumpState() override;

private:
  void dumpState( std::ostringstream& );
};

#endif // GAUDIHIVE_FORWARDSCHEDULERSVC_H