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"
#include "GaudiKernel/IThreadPoolSvc.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 extends<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:

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

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

  void activate();

  StatusCode deactivate();

  std::atomic<ActivationState> 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;

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

  bool m_first;

  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();

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

  bool m_checkDeps;


};

#endif // GAUDIHIVE_FORWARDSCHEDULERSVC_H