AvalancheSchedulerSvc.h

Go to the documentation of this file.

/***********************************************************************************\
* (c) Copyright 1998-2019 CERN for the benefit of the LHCb and ATLAS collaborations *
*                                                                                   *
* This software is distributed under the terms of the Apache version 2 licence,     *
* copied verbatim in the file "LICENSE".                                            *
*                                                                                   *
* In applying this licence, CERN does not waive the privileges and immunities       *
* granted to it by virtue of its status as an Intergovernmental Organization        *
* or submit itself to any jurisdiction.                                             *
\***********************************************************************************/
#ifndef GAUDIHIVE_AVALANCHESCHEDULERSVC_H
#define GAUDIHIVE_AVALANCHESCHEDULERSVC_H

// Local includes
#include "AlgsExecutionStates.h"
#include "EventSlot.h"
#include "PrecedenceSvc.h"

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

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

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

class IAlgorithm;

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

class AvalancheSchedulerSvc : public extends<Service, IScheduler> {
  friend class AlgoExecutionTask;

public:
  using extends::extends;

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

  virtual StatusCode scheduleEventView( const EventContext* sourceContext, const std::string& nodeName,
                                        std::unique_ptr<EventContext> viewContext ) override;

private:
  using AState = AlgsExecutionStates::State;
  using action = std::function<StatusCode()>;

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

  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_maxIOBoundAlgosInFlight{this, "MaxIOBoundAlgosInFlight", 0,
                                                          "Maximum number of simultaneous I/O-bound algorithms"};
  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<bool> m_checkDeps{this, "CheckDependencies", false, "Runtime check of Algorithm Data Dependencies"};

  Gaudi::Property<std::string> m_useDataLoader{this, "DataLoaderAlg", "",
                                               "Attribute unmet input dependencies to this DataLoader Algorithm"};

  Gaudi::Property<bool> m_enableCondSvc{this, "EnableConditions", false, "Enable ConditionsSvc"};

  Gaudi::Property<bool> m_showDataDeps{this, "ShowDataDependencies", true,
                                       "Show the INPUT and OUTPUT data dependencies of Algorithms"};

  Gaudi::Property<bool> m_showDataFlow{this, "ShowDataFlow", false,
                                       "Show the configuration of DataFlow between Algorithms"};

  Gaudi::Property<bool> m_showControlFlow{this, "ShowControlFlow", false,
                                          "Show the configuration of all Algorithms and Sequences"};

  Gaudi::Property<bool> m_verboseSubSlots{this, "VerboseSubSlots", false, "Dump algorithm states for all sub-slots"};

  // 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 ) { return m_algname_index_map[algoname]; };

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

  inline const std::string& index2algname( unsigned int index ) { return m_algname_vect[index]; };

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

  SmartIF<IPrecedenceSvc> m_precSvc;

  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;

  SmartIF<IAlgExecStateSvc> m_algExecStateSvc;

  SmartIF<ICondSvc> m_condSvc;

  unsigned int m_algosInFlight = 0;

  unsigned int m_IOBoundAlgosInFlight = 0;

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

  StatusCode updateStates();

  // Update algorithm state in the appropriate event slot
  StatusCode setAlgState( unsigned int iAlgo, EventContext* contextPtr, AState state, bool iterate = false );

  StatusCode enqueue( unsigned int iAlgo, int si, EventContext* );
  StatusCode promoteToAsyncScheduled( unsigned int iAlgo, int si, EventContext* ); // tests of an asynchronous scheduler
  StatusCode promoteToExecuted( unsigned int iAlgo, int si, EventContext* );
  StatusCode promoteToAsyncExecuted( unsigned int iAlgo, int si, IAlgorithm* algo,
                                     EventContext* ); // tests of an asynchronous scheduler
  StatusCode promoteToFinished( unsigned int iAlgo, int si );

  bool isStalled( const EventSlot& ) const;
  void eventFailed( EventContext* eventContext );

  void dumpSchedulerState( int iSlot );

  // Algos Management -------------------------------------------------------

  SmartIF<IAlgResourcePool> m_algResourcePool;

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

  tbb::concurrent_bounded_queue<action> m_actionsQueue;

  struct AlgQueueEntry {
    unsigned int  algIndex;
    int           slotIndex;
    EventContext* contextPtr;
    unsigned int  rank;
    IAlgorithm*   algPtr;
  };

  struct AlgQueueSort {
    bool operator()( const AlgQueueEntry& i, const AlgQueueEntry& j ) const { return ( i.rank < j.rank ); }
  };

  tbb::concurrent_priority_queue<AlgQueueEntry, AlgQueueSort> m_scheduledQueue;
  std::queue<AlgQueueEntry>                                   m_retryQueue;

  // Prompt the scheduler to call updateStates
  std::atomic<bool> m_needsUpdate{true};

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

  // Service for thread pool initialization
  SmartIF<IThreadPoolSvc> m_threadPoolSvc;
  size_t                  m_maxEventsInFlight{0};
  size_t                  m_maxAlgosInFlight{1};
};

#endif // GAUDIHIVE_AVALANCHESCHEDULERSVC_H