AvalancheSchedulerSvc.h

Go to the documentation of this file.

/***********************************************************************************\
* (c) Copyright 1998-2025 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.                                             *
\***********************************************************************************/
#pragma once
 
// Local includes
#include "AlgsExecutionStates.h"
#include "EventSlot.h"
#include "FiberManager.h"
#include "PrecedenceSvc.h"
 
// Framework include files
#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 <memory>
#include <queue>
#include <string>
#include <string_view>
#include <thread>
#include <unordered_map>
#include <vector>
 
// External libs
#include <tbb/concurrent_priority_queue.h>
#include <tbb/concurrent_queue.h>
#include <tbb/task_arena.h>
 
class IAlgorithm;
 
//---------------------------------------------------------------------------

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

  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;

  void dumpState() override;

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

  virtual void recordOccupancy( int samplePeriod, std::function<void( OccupancySnapshot )> callback ) override;
 
private:
  StatusCode dumpGraphFile( const std::map<std::string, DataObjIDColl>& inDeps,
                            const std::map<std::string, DataObjIDColl>& outDeps ) const;
 
private:
  using AState = AlgsExecutionStates::State;
  using action = std::function<StatusCode()>;
 
  enum ActivationState { INACTIVE = 0, ACTIVE = 1, FAILURE = 2 };
 
  // Occupancy snapshot data
  std::chrono::duration<int64_t, std::milli> m_snapshotInterval = std::chrono::duration<int64_t, std::milli>::min();
  std::chrono::system_clock::time_point      m_lastSnapshot     = std::chrono::system_clock::now();
  std::function<void( OccupancySnapshot )>   m_snapshotCallback;
 
  Gaudi::Property<int> m_threadPoolSize{
      this, "ThreadPoolSize", -1,
      "Size of the global thread pool initialised by TBB; a value of -1 requests to use"
      "all available hardware threads; -100 requests to bypass TBB executing "
      "all algorithms in the scheduler's thread." };
  Gaudi::Property<int> m_maxParallelismExtra{
      this, "maxParallelismExtra", 0,
      "Allows to add some extra threads to the maximum parallelism set in TBB"
      "The TBB max parallelism is set as: ThreadPoolSize + maxParallelismExtra + 1" };
  Gaudi::Property<std::string>  m_whiteboardSvcName{ this, "WhiteboardSvc", "EventDataSvc", "The whiteboard name" };
  Gaudi::Property<unsigned int> m_maxBlockingAlgosInFlight{
      this, "MaxBlockingAlgosInFlight", 0, "Maximum allowed number of simultaneously running CPU-blocking 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_enablePreemptiveBlockingTasks{
      this, "PreemptiveBlockingTasks", false,
      "Enable preemptive scheduling of CPU-blocking algorithms. Blocking algorithms must be flagged accordingly." };
  Gaudi::Property<int> m_numOffloadThreads{
      this, "NumOffloadThreads", 2,
      "Number of threads to use for CPU portion of asynchronous algorithms. Asynchronous algorithms must be flagged "
      "and use Boost Fiber functionality to suspend while waiting for offloaded work." };
  Gaudi::Property<bool>                     m_checkDeps{ this, "CheckDependencies", false,
                                     "Runtime check of Algorithm Input Data Dependencies" };
  Gaudi::Property<bool>                     m_checkOutput{ this, "CheckOutputUsage", false,
                                       "Runtime check of Algorithm Output Data usage" };
  Gaudi::Property<std::vector<std::string>> m_checkOutputIgnoreList{
      this,
      "CheckOutputUsageIgnoreList",
      {},
      "Ignore outputs of the Algorithms of this name when doing the check",
      "OrderedSet<std::string>" };
 
  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" };
 
  Gaudi::Property<std::string> m_dataDepsGraphFile{
      this, "DataDepsGraphFile", "",
      "Name of the output file (.dot or .md extensions allowed) containing the data dependency graph for some selected "
      "Algorithms" };
 
  Gaudi::Property<std::string> m_dataDepsGraphAlgoPattern{
      this, "DataDepsGraphAlgPattern", ".*",
      "Regex pattern for selecting desired Algorithms by name, whose data dependency has to be included in the data "
      "deps graph" };
 
  Gaudi::Property<std::string> m_dataDepsGraphObjectPattern{
      this, "DataDepsGraphObjectPattern", ".*",
      "Regex pattern for selecting desired input or output by their full key" };
 
  // 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;

  std::vector<EventSlot> m_eventSlots;

  std::atomic_int m_freeSlots{ 0 };

  tbb::concurrent_bounded_queue<EventContext*> m_finishedEvents;

  SmartIF<IAlgExecStateSvc> m_algExecStateSvc;

  SmartIF<ICondSvc> m_condSvc;

  unsigned int m_algosInFlight = 0;

  unsigned int m_blockingAlgosInFlight = 0;
 
  // States management ------------------------------------------------------

  StatusCode iterate();
 
  // Update algorithm state and, optionally, revise states of other downstream algorithms
  StatusCode revise( unsigned int iAlgo, EventContext* contextPtr, AState state, bool iterate = false );

  struct TaskSpec;
  StatusCode schedule( TaskSpec&& );
  StatusCode signoff( const TaskSpec& );

  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 TaskSpec {
    TaskSpec() {}
    TaskSpec( IAlgorithm* algPtr, unsigned int algIndex, const std::string& algName, unsigned int algRank,
              bool asynchronous, int slotIndex, EventContext* eventContext )
        : algPtr( algPtr )
        , algIndex( algIndex )
        , algName( algName )
        , algRank( algRank )
        , asynchronous( asynchronous )
        , slotIndex( slotIndex )
        , contextPtr( eventContext ) {}

    TaskSpec( const TaskSpec& ) = default;
    TaskSpec& operator=( const TaskSpec& ) = delete;
    TaskSpec( TaskSpec&& ) = default;
    TaskSpec& operator=( TaskSpec&& ) = default;
 
    IAlgorithm*      algPtr{ nullptr };
    unsigned int     algIndex{ 0 };
    std::string_view algName;
    unsigned int     algRank{ 0 };
    bool             asynchronous{ false };
    int              slotIndex{ 0 };
    EventContext*    contextPtr{ nullptr };
  };

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

  tbb::concurrent_priority_queue<TaskSpec, AlgQueueSort> m_scheduledQueue;
  tbb::concurrent_priority_queue<TaskSpec, AlgQueueSort> m_scheduledAsynchronousQueue;
  std::queue<TaskSpec>                                   m_retryQueue;
 
  // Prompt the scheduler to call updateStates
  std::atomic<bool> m_needsUpdate{ true };
 
  // ------------------------------------------------------------------------
 
  // Service for thread pool initialization
  SmartIF<IThreadPoolSvc>       m_threadPoolSvc;
  tbb::task_arena*              m_arena{ nullptr };
  std::unique_ptr<FiberManager> m_fiberManager{ nullptr };
 
  size_t m_maxEventsInFlight{ 0 };
  size_t m_maxAlgosInFlight{ 1 };
 
public:
  // get next schedule-able TaskSpec
  bool next( TaskSpec& ts, bool asynchronous ) {
    if ( asynchronous ) { return m_scheduledAsynchronousQueue.try_pop( ts ); }
    return m_scheduledQueue.try_pop( ts );
  }
};