#include <src/AvalancheSchedulerSvc.h>

Inheritance diagram for AvalancheSchedulerSvc:

[legend]

Collaboration diagram for AvalancheSchedulerSvc:

[legend]

Classes
struct	AlgQueueEntry
	Struct to hold entries in the alg queues. More...

struct	AlgQueueSort
	Comparison operator to sort the queues. More...

Public Member Functions
	~AvalancheSchedulerSvc () override=default
	Destructor. More...

StatusCode	initialize () override
	Initialise. More...

StatusCode	finalize () override
	Finalise. More...

StatusCode	pushNewEvent (EventContext *eventContext) override
	Make an event available to the scheduler. More...

StatusCode	pushNewEvents (std::vector< EventContext * > &eventContexts) override

StatusCode	popFinishedEvent (EventContext *&eventContext) override
	Blocks until an event is available. More...

StatusCode	tryPopFinishedEvent (EventContext *&eventContext) override
	Try to fetch an event from the scheduler. More...

unsigned int	freeSlots () override
	Get free slots number. More...

virtual StatusCode	scheduleEventView (const EventContext *sourceContext, const std::string &nodeName, std::unique_ptr< EventContext > viewContext) override
	Method to inform the scheduler about event views. More...

Public Member Functions inherited from extends< Service, IScheduler >
void *	i_cast (const InterfaceID &tid) const override
	Implementation of IInterface::i_cast. More...

StatusCode	queryInterface (const InterfaceID &ti, void **pp) override
	Implementation of IInterface::queryInterface. More...

std::vector< std::string >	getInterfaceNames () const override
	Implementation of IInterface::getInterfaceNames. More...

Public Member Functions inherited from Service
const std::string &	name () const override
	Retrieve name of the service. More...

StatusCode	configure () override

StatusCode	initialize () override

StatusCode	start () override

StatusCode	stop () override

StatusCode	finalize () override

StatusCode	terminate () override

Gaudi::StateMachine::State	FSMState () const override

Gaudi::StateMachine::State	targetFSMState () const override

StatusCode	reinitialize () override

StatusCode	restart () override

StatusCode	sysInitialize () override
	Initialize Service. More...

StatusCode	sysStart () override
	Initialize Service. More...

StatusCode	sysStop () override
	Initialize Service. More...

StatusCode	sysFinalize () override
	Finalize Service. More...

StatusCode	sysReinitialize () override
	Re-initialize the Service. More...

StatusCode	sysRestart () override
	Re-initialize the Service. More...

	Service (std::string name, ISvcLocator *svcloc)
	Standard Constructor. More...

SmartIF< ISvcLocator > &	serviceLocator () const override
	Retrieve pointer to service locator. More...

StatusCode	setProperties ()
	Method for setting declared properties to the values specified for the job. More...

template<class T >
StatusCode	service (const std::string &name, const T *&psvc, bool createIf=true) const
	Access a service by name, creating it if it doesn't already exist. More...

template<class T >
StatusCode	service (const std::string &name, T *&psvc, bool createIf=true) const

template<typename IFace = IService>
SmartIF< IFace >	service (const std::string &name, bool createIf=true) const

template<class T >
StatusCode	service (const std::string &svcType, const std::string &svcName, T *&psvc) const
	Access a service by name and type, creating it if it doesn't already exist. More...

template<class T >
StatusCode	declareTool (ToolHandle< T > &handle, std::string toolTypeAndName, bool createIf=true)
	Declare used tool. More...

SmartIF< IAuditorSvc > &	auditorSvc () const
	The standard auditor service.May not be invoked before sysInitialize() has been invoked. More...

Public Member Functions inherited from PropertyHolder< CommonMessaging< implements< IService, IProperty, IStateful > > >
	PropertyHolder ()=default

Gaudi::Details::PropertyBase &	declareProperty (Gaudi::Details::PropertyBase &prop)
	Declare a property. More...

Gaudi::Details::PropertyBase *	declareProperty (const std::string &name, TYPE &value, const std::string &doc="none")
	Helper to wrap a regular data member and use it as a regular property. More...

Gaudi::Details::PropertyBase *	declareProperty (const std::string &name, Gaudi::Property< TYPE, VERIFIER, HANDLERS > &prop, const std::string &doc="none")
	Declare a PropertyBase instance setting name and documentation. More...

Gaudi::Details::PropertyBase *	declareRemoteProperty (const std::string &name, IProperty *rsvc, const std::string &rname="")
	Declare a remote property. More...

StatusCode	setProperty (const Gaudi::Details::PropertyBase &p) override
	set the property form another property More...

StatusCode	setProperty (const std::string &s) override
	set the property from the formatted string More...

StatusCode	setProperty (const std::string &n, const std::string &v) override
	set the property from name and the value More...

StatusCode	setProperty (const std::string &name, const TYPE &value)
	set the property form the value More...

StatusCode	getProperty (Gaudi::Details::PropertyBase *p) const override
	get the property More...

const Gaudi::Details::PropertyBase &	getProperty (const std::string &name) const override
	get the property by name More...

StatusCode	getProperty (const std::string &n, std::string &v) const override
	convert the property to the string More...

const std::vector< Gaudi::Details::PropertyBase * > &	getProperties () const override
	get all properties More...

bool	hasProperty (const std::string &name) const override
	Return true if we have a property with the given name. More...

	PropertyHolder (const PropertyHolder &)=delete

PropertyHolder &	operator= (const PropertyHolder &)=delete

Public Member Functions inherited from CommonMessaging< implements< IService, IProperty, IStateful > >
MSG::Level	msgLevel () const
	get the cached level (originally extracted from the embedded MsgStream) More...

bool	msgLevel (MSG::Level lvl) const
	get the output level from the embedded MsgStream More...

MSG::Level	outputLevel () const
	Backward compatibility function for getting the output level. More...

Public Member Functions inherited from CommonMessagingBase
virtual	~CommonMessagingBase ()=default
	Virtual destructor. More...

const SmartIF< IMessageSvc > &	msgSvc () const
	The standard message service. More...

MsgStream &	msgStream () const
	Return an uninitialized MsgStream. More...

MsgStream &	msgStream (const MSG::Level level) const
	Predefined configurable message stream for the efficient printouts. More...

MsgStream &	always () const
	shortcut for the method msgStream(MSG::ALWAYS) More...

MsgStream &	fatal () const
	shortcut for the method msgStream(MSG::FATAL) More...

MsgStream &	err () const
	shortcut for the method msgStream(MSG::ERROR) More...

MsgStream &	error () const
	shortcut for the method msgStream(MSG::ERROR) More...

MsgStream &	warning () const
	shortcut for the method msgStream(MSG::WARNING) More...

MsgStream &	info () const
	shortcut for the method msgStream(MSG::INFO) More...

MsgStream &	debug () const
	shortcut for the method msgStream(MSG::DEBUG) More...

MsgStream &	verbose () const
	shortcut for the method msgStream(MSG::VERBOSE) More...

MsgStream &	msg () const
	shortcut for the method msgStream(MSG::INFO) More...

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

using	AState = AlgsExecutionStates::State

using	action = std::function< StatusCode()>

Private Member Functions
void	activate ()
	Activate scheduler. More...

StatusCode	deactivate ()
	Deactivate scheduler. More...

unsigned int	algname2index (const std::string &algoname)
	Convert a name to an integer. More...

const std::string &	index2algname (unsigned int index)
	Convert an integer to a name. More...

StatusCode	updateStates ()
	Loop on algorithm in the slots and promote them to successive states. More...

StatusCode	setAlgState (unsigned int iAlgo, EventContext *contextPtr, AState state, bool iterate=false)

StatusCode	enqueue (unsigned int iAlgo, int si, EventContext *)
	Algorithm promotion. More...

StatusCode	promoteToAsyncScheduled (unsigned int iAlgo, int si, EventContext *)

StatusCode	promoteToExecuted (unsigned int iAlgo, int si, EventContext *)
	The call to this method is triggered only from within the AlgoExecutionTask. More...

StatusCode	promoteToAsyncExecuted (unsigned int iAlgo, int si, IAlgorithm algo, EventContext )
	The call to this method is triggered only from within the IOBoundAlgTask. More...

StatusCode	promoteToFinished (unsigned int iAlgo, int si)

bool	isStalled (const EventSlot &) const
	Check if scheduling in a particular slot is in a stall. More...

void	eventFailed (EventContext *eventContext)
	Method to execute if an event failed. More...

void	dumpSchedulerState (int iSlot)
	Dump the state of the scheduler. More...

Private Attributes
Gaudi::Property< int >	m_threadPoolSize

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

Gaudi::Property< bool >	m_simulateExecution

Gaudi::Property< std::string >	m_optimizationMode

Gaudi::Property< bool >	m_dumpIntraEventDynamics

Gaudi::Property< bool >	m_useIOBoundAlgScheduler

Gaudi::Property< bool >	m_checkDeps {this, "CheckDependencies", false, "Runtime check of Algorithm Data Dependencies"}

Gaudi::Property< std::string >	m_useDataLoader

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

Gaudi::Property< bool >	m_showDataDeps

Gaudi::Property< bool >	m_showDataFlow

Gaudi::Property< bool >	m_showControlFlow

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

std::atomic< ActivationState >	m_isActive {INACTIVE}
	Flag to track if the scheduler is active or not. More...

std::thread	m_thread
	The thread in which the activate function runs. More...

std::unordered_map< std::string, unsigned int >	m_algname_index_map
	Map to bookkeep the information necessary to the name2index conversion. More...

std::vector< std::string >	m_algname_vect
	Vector to bookkeep the information necessary to the index2name conversion. More...

SmartIF< IPrecedenceSvc >	m_precSvc
	A shortcut to the Precedence Service. More...

SmartIF< IHiveWhiteBoard >	m_whiteboard
	A shortcut to the whiteboard. More...

SmartIF< IAccelerator >	m_IOBoundAlgScheduler
	A shortcut to IO-bound algorithm scheduler. More...

std::vector< EventSlot >	m_eventSlots
	Vector of events slots. More...

std::atomic_int	m_freeSlots
	Atomic to account for asyncronous updates by the scheduler wrt the rest. More...

tbb::concurrent_bounded_queue< EventContext * >	m_finishedEvents
	Queue of finished events. More...

SmartIF< IAlgExecStateSvc >	m_algExecStateSvc
	Algorithm execution state manager. More...

SmartIF< ICondSvc >	m_condSvc
	A shortcut to service for Conditions handling. More...

unsigned int	m_algosInFlight = 0
	Number of algorithms presently in flight. More...

unsigned int	m_IOBoundAlgosInFlight = 0
	Number of algorithms presently in flight. More...

SmartIF< IAlgResourcePool >	m_algResourcePool
	Cache for the algorithm resource pool. More...

tbb::concurrent_bounded_queue< action >	m_actionsQueue
	Queue where closures are stored and picked for execution. More...

tbb::concurrent_priority_queue< AlgQueueEntry, AlgQueueSort >	m_scheduledQueue
	Queues for scheduled algorithms. More...

std::queue< AlgQueueEntry >	m_retryQueue

std::atomic< bool >	m_needsUpdate {true}

SmartIF< IThreadPoolSvc >	m_threadPoolSvc

size_t	m_maxEventsInFlight {0}

size_t	m_maxAlgosInFlight {1}

Friends
class	AlgoExecutionTask

Additional Inherited Members
Public Types inherited from extends< Service, IScheduler >
using	base_class = extends
	Typedef to this class. More...

using	extend_interfaces_base = extend_interfaces< Interfaces... >
	Typedef to the base of this class. More...

Public Types inherited from Service
using	Factory = Gaudi::PluginService::Factory< IService (const std::string &, ISvcLocator )>

Public Types inherited from PropertyHolder< CommonMessaging< implements< IService, IProperty, IStateful > > >
using	PropertyHolderImpl = PropertyHolder< CommonMessaging< implements< IService, IProperty, IStateful > > >
	Typedef used to refer to this class from derived classes, as in. More...

Public Types inherited from CommonMessaging< implements< IService, IProperty, IStateful > >
using	base_class = CommonMessaging

Public Types inherited from extend_interfaces< Interfaces... >
using	ext_iids = typename Gaudi::interface_list_cat< typename Interfaces::ext_iids... >::type
	take union of the ext_iids of all Interfaces... More...

Protected Member Functions inherited from Service
	~Service () override
	Standard Destructor. More...

int	outputLevel () const
	get the `Service's` output level More...

Protected Member Functions inherited from PropertyHolder< CommonMessaging< implements< IService, IProperty, IStateful > > >
Gaudi::Details::PropertyBase *	property (const std::string &name) const

Protected Member Functions inherited from CommonMessaging< implements< IService, IProperty, IStateful > >
MSG::Level	setUpMessaging () const
	Set up local caches. More...

MSG::Level	resetMessaging ()
	Reinitialize internal states. More...

void	updateMsgStreamOutputLevel (int level)
	Update the output level of the cached MsgStream. More...

Protected Attributes inherited from Service
Gaudi::StateMachine::State	m_state = Gaudi::StateMachine::OFFLINE
	Service state. More...

Gaudi::StateMachine::State	m_targetState = Gaudi::StateMachine::OFFLINE
	Service state. More...

Gaudi::Property< int >	m_outputLevel {this, "OutputLevel", MSG::NIL, "output level"}

Gaudi::Property< bool >	m_auditInit {this, "AuditServices", false, " unused"}

Gaudi::Property< bool >	m_auditorInitialize {this, "AuditInitialize", false, "trigger auditor on initialize()"}

Gaudi::Property< bool >	m_auditorStart {this, "AuditStart", false, "trigger auditor on start()"}

Gaudi::Property< bool >	m_auditorStop {this, "AuditStop", false, "trigger auditor on stop()"}

Gaudi::Property< bool >	m_auditorFinalize {this, "AuditFinalize", false, "trigger auditor on finalize()"}

Gaudi::Property< bool >	m_auditorReinitialize {this, "AuditReinitialize", false, "trigger auditor on reinitialize()"}

Gaudi::Property< bool >	m_auditorRestart {this, "AuditRestart", false, "trigger auditor on restart()"}

SmartIF< IAuditorSvc >	m_pAuditorSvc
	Auditor Service. More...

Detailed Description

Introduction

The scheduler is named after its ability to generically maximize the average intra-event task occupancy by inducing avalanche-like concurrency disclosure waves in conditions of arbitrary intra-event task precedence constraints (see section 3.2 of http://cern.ch/go/7Jn7).

Task precedence management

The scheduler is driven by graph-based task precedence management. When compared to approach used in the ForwardSchedulerSvc, the following advantages can be emphasized:

(1) Faster decision making (thus lower concurrency disclosure downtime); (2) Capacity for proactive task scheduling decision making.

Point (2) allowed to implement a number of generic, non-intrusive intra-event throughput maximization scheduling strategies.

Scheduling principles

o Task scheduling prerequisites

A task is scheduled ASA all following conditions are met:

if a control flow (CF) graph traversal reaches the task;
when all data flow (DF) dependencies of the task are satisfied;
when the DF-ready task pool parsing mechanism (*) considers it, and:
- a free (or re-entrant) algorithm instance to run within the task is available;
- there is a free computational resource to run the task.

o (*) Avalanche induction strategies

The scheduler is able to maximize the intra-event throughput by applying several search strategies within the pool, prioritizing tasks according to the following types of precedence rules graph asymmetries:

(A) Local task-to-data asymmetry; (B) Local task-to-task asymmetry; (C) Global task-to-task asymmetry.

o Other mechanisms of throughput maximization

The scheduler is able to maximize the overall throughput of data processing by scheduling the CPU-blocking tasks efficiently. The mechanism can be applied to the following types of tasks:

I/O-bound tasks;
tasks with computation offloading (accelerators, GPGPUs, clouds, quantum computing devices..joke);
synchronization-bound tasks.

Credits

Historically, the AvalancheSchedulerSvc branched off the ForwardSchedulerSvc and in many ways built its success on ideas and code of the latter.

Author: Illya Shapoval

Version: 1.0

Definition at line 112 of file AvalancheSchedulerSvc.h.

Member Typedef Documentation

◆ action

using AvalancheSchedulerSvc::action = std::function<StatusCode()>

private

Definition at line 149 of file AvalancheSchedulerSvc.h.

◆ AState

using AvalancheSchedulerSvc::AState = AlgsExecutionStates::State

private

Definition at line 148 of file AvalancheSchedulerSvc.h.

Member Enumeration Documentation

◆ ActivationState

enum AvalancheSchedulerSvc::ActivationState

private

Enumerator
INACTIVE
ACTIVE
FAILURE

Definition at line 151 of file AvalancheSchedulerSvc.h.

151 { INACTIVE = 0, ACTIVE = 1, FAILURE = 2 };

AvalancheSchedulerSvc::INACTIVE

Definition: AvalancheSchedulerSvc.h:151

AvalancheSchedulerSvc::ACTIVE

Definition: AvalancheSchedulerSvc.h:151

AvalancheSchedulerSvc::FAILURE

Definition: AvalancheSchedulerSvc.h:151

Constructor & Destructor Documentation

◆ ~AvalancheSchedulerSvc()

AvalancheSchedulerSvc::~AvalancheSchedulerSvc ( )

overridedefault

Destructor.

Member Function Documentation

◆ activate()

void AvalancheSchedulerSvc::activate ( )

private

Activate scheduler.

Activate the scheduler.

From this moment on the queue of actions is checked. The checking will stop when the m_isActive flag is false and the queue is not empty. This will guarantee that all actions are executed and a stall is not created. The TBB pool must be initialised in the thread from where the tasks are launched (http://threadingbuildingblocks.org/docs/doxygen/a00342.html) The scheduler is initialised here since this method runs in a separate thread and spawns the tasks (through the execution of the lambdas)

Definition at line 373 of file AvalancheSchedulerSvc.cpp.

                                      {
 
   ON_DEBUG debug() << "AvalancheSchedulerSvc::activate()" << endmsg;
 
   if ( m_threadPoolSvc->initPool( m_threadPoolSize ).isFailure() ) {
     error() << "problems initializing ThreadPoolSvc" << endmsg;
     m_isActive = FAILURE;
     return;
   }
 
   // Wait for actions pushed into the queue by finishing tasks.
   action     thisAction;
   StatusCode sc( StatusCode::SUCCESS );
 
   m_isActive = ACTIVE;
 
   // Continue to wait if the scheduler is running or there is something to do
   ON_DEBUG debug() << "Start checking the actionsQueue" << endmsg;
   while ( m_isActive == ACTIVE || m_actionsQueue.size() != 0 ) {
     m_actionsQueue.pop( thisAction );
     sc = thisAction();
     ON_VERBOSE {
       if ( sc.isFailure() )
         verbose() << "Action did not succeed (which is not bad per se)." << endmsg;
       else
         verbose() << "Action succeeded." << endmsg;
     }
     else sc.ignore();
 
     // If all queued actions have been processed, update the slot states
     if ( m_needsUpdate.load() && m_actionsQueue.empty() ) {
       sc = updateStates();
       ON_VERBOSE {
         if ( sc.isFailure() )
           verbose() << "updateStates did not succeed (which is not bad per se)." << endmsg;
         else
           verbose() << "updateStates succeeded." << endmsg;
       }
       else sc.ignore();
     }
   }
 
   ON_DEBUG debug() << "Terminating thread-pool resources" << endmsg;
   if ( m_threadPoolSvc->terminatePool().isFailure() ) {
     error() << "Problems terminating thread pool" << endmsg;
     m_isActive = FAILURE;
   }
 }

◆ algname2index()

unsigned int AvalancheSchedulerSvc::algname2index ( const std::string & algoname )

inlineprivate

Convert a name to an integer.

Definition at line 203 of file AvalancheSchedulerSvc.h.

203 { return m_algname_index_map[algoname]; };

AvalancheSchedulerSvc::m_algname_index_map

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

Map to bookkeep the information necessary to the name2index conversion.

Definition: AvalancheSchedulerSvc.h:203

◆ deactivate()

StatusCode AvalancheSchedulerSvc::deactivate ( )

private

Deactivate scheduler.

Deactivates the scheduler.

Two actions are pushed into the queue: 1) Drain the scheduler until all events are finished. 2) Flip the status flag m_isActive to false This second action is the last one to be executed by the scheduler.

Definition at line 430 of file AvalancheSchedulerSvc.cpp.

                                              {
 
   if ( m_isActive == ACTIVE ) {
 
     // Set the number of slots available to an error code
     m_freeSlots.store( 0 );
 
     // Empty queue
     action thisAction;
     while ( m_actionsQueue.try_pop( thisAction ) ) {};
 
     // This would be the last action
     m_actionsQueue.push( [this]() -> StatusCode {
       ON_VERBOSE verbose() << "Deactivating scheduler" << endmsg;
       m_isActive = INACTIVE;
       return StatusCode::SUCCESS;
     } );
   }
 
   return StatusCode::SUCCESS;
 }

◆ dumpSchedulerState()

void AvalancheSchedulerSvc::dumpSchedulerState ( int iSlot )

private

Dump the state of the scheduler.

Used for debugging purposes, the state of the scheduler is dumped on screen in order to be inspected.

Definition at line 756 of file AvalancheSchedulerSvc.cpp.

                                                           {
 
   // To have just one big message
   std::ostringstream outputMS;
 
   outputMS << "Dumping scheduler state\n"
            << "=========================================================================================\n"
            << "++++++++++++++++++++++++++++++++++++ SCHEDULER STATE ++++++++++++++++++++++++++++++++++++\n"
            << "=========================================================================================\n\n";
 
   //===========================================================================
 
   outputMS << "------------------ Last schedule: Task/Event/Slot/Thread/State Mapping "
            << "------------------\n\n";
 
   // Figure if TimelineSvc is available (used below to detect threads IDs)
   auto timelineSvc = serviceLocator()->service<ITimelineSvc>( "TimelineSvc", false );
   if ( !timelineSvc.isValid() || !timelineSvc->isEnabled() ) {
     outputMS << "WARNING Enable TimelineSvc in record mode (RecordTimeline = True) to trace the mapping\n";
   } else {
 
     // Figure optimal printout layout
     size_t indt( 0 );
     for ( auto& slot : m_eventSlots )
       for ( auto it = slot.algsStates.begin( AState::SCHEDULED ); it != slot.algsStates.end( AState::SCHEDULED ); ++it )
         if ( index2algname( *it ).length() > indt ) indt = index2algname( *it ).length();
 
     // Figure the last running schedule across all slots
     for ( auto& slot : m_eventSlots ) {
       for ( auto it = slot.algsStates.begin( AState::SCHEDULED ); it != slot.algsStates.end( AState::SCHEDULED );
             ++it ) {
 
         const std::string algoName{index2algname( *it )};
 
         outputMS << "  task: " << std::setw( indt ) << algoName << " evt/slot: " << slot.eventContext->evt() << "/"
                  << slot.eventContext->slot();
 
         // Try to get POSIX threads IDs the currently running tasks are scheduled to
         if ( timelineSvc.isValid() ) {
           TimelineEvent te{};
           te.algorithm = algoName;
           te.slot      = slot.eventContext->slot();
           te.event     = slot.eventContext->evt();
 
           if ( timelineSvc->getTimelineEvent( te ) )
             outputMS << " thread.id: 0x" << std::hex << te.thread << std::dec;
           else
             outputMS << " thread.id: [unknown]"; // this means a task has just
                                                  // been signed off as SCHEDULED,
                                                  // but has not been assigned to a thread yet
                                                  // (i.e., not running yet)
         }
         outputMS << " state: [" << m_algExecStateSvc->algExecState( algoName, *( slot.eventContext ) ) << "]\n";
       }
     }
   }
 
   //===========================================================================
 
   outputMS << "\n---------------------------- Task/CF/FSM Mapping "
            << ( 0 > iSlot ? "[all slots] --" : "[target slot] " ) << "--------------------------\n\n";
 
   int slotCount = -1;
   for ( auto& slot : m_eventSlots ) {
     ++slotCount;
     if ( slot.complete ) continue;
 
     outputMS << "[ slot: "
              << ( slot.eventContext->valid() ? std::to_string( slot.eventContext->slot() ) : "[ctx invalid]" )
              << "  event: "
              << ( slot.eventContext->valid() ? std::to_string( slot.eventContext->evt() ) : "[ctx invalid]" )
              << " ]:\n\n";
 
     if ( 0 > iSlot || iSlot == slotCount ) {
 
       // Snapshot of the Control Flow and FSM states
       outputMS << m_precSvc->printState( slot ) << "\n";
 
       // Mention sub slots (this is expensive if the number of sub-slots is high)
       if ( m_verboseSubSlots && !slot.allSubSlots.empty() ) {
         outputMS << "\nNumber of sub-slots: " << slot.allSubSlots.size() << "\n\n";
         auto slotID = slot.eventContext->valid() ? std::to_string( slot.eventContext->slot() ) : "[ctx invalid]";
         for ( auto& ss : slot.allSubSlots ) {
           outputMS << "[ slot: " << slotID << ", sub-slot: "
                    << ( ss.eventContext->valid() ? std::to_string( ss.eventContext->subSlot() ) : "[ctx invalid]" )
                    << ", entry: " << ss.entryPoint << ", event: "
                    << ( ss.eventContext->valid() ? std::to_string( ss.eventContext->evt() ) : "[ctx invalid]" )
                    << " ]:\n\n";
           outputMS << m_precSvc->printState( ss ) << "\n";
         }
       }
     }
   }
 
   //===========================================================================
 
   if ( 0 <= iSlot ) {
     outputMS << "\n------------------------------ Algorithm Execution States -----------------------------\n\n";
     m_algExecStateSvc->dump( outputMS, *( m_eventSlots[iSlot].eventContext ) );
   }
 
   outputMS << "\n=========================================================================================\n"
            << "++++++++++++++++++++++++++++++++++++++ END OF DUMP ++++++++++++++++++++++++++++++++++++++\n"
            << "=========================================================================================\n\n";
 
   info() << outputMS.str() << endmsg;
 }

◆ enqueue()

StatusCode AvalancheSchedulerSvc::enqueue	(	unsigned int	iAlgo,
		int	si,
		EventContext *	eventContext
	)

private

Algorithm promotion.

Definition at line 866 of file AvalancheSchedulerSvc.cpp.

                                                                                                   {
 
   // Use the algorithm rank to sort the queue
   const std::string& algName( index2algname( iAlgo ) );
   unsigned int       rank = 0;
   if ( !m_optimizationMode.empty() ) { rank = m_precSvc->getPriority( algName ); }
 
   // Get algorithm pointer
   IAlgorithm* iAlgoPtr = nullptr;
   StatusCode  getAlgSC( m_algResourcePool->acquireAlgorithm( algName, iAlgoPtr ) );
 
   // Check if the algorithm is available
   AState state;
   if ( getAlgSC.isSuccess() ) {
 
     // Add the algorithm to the scheduled queue
     m_scheduledQueue.push( {iAlgo, si, eventContext, rank, iAlgoPtr} );
     ++m_algosInFlight;
 
     // Avoid to use tbb if the pool size is 1 and run in this thread
     if ( -100 != m_threadPoolSize ) {
 
       // the child task that executes an Algorithm
       tbb::task* algoTask =
           new ( tbb::task::allocate_root() ) AlgoExecutionTask( this, serviceLocator(), m_algExecStateSvc );
       // schedule the algoTask
       tbb::task::enqueue( *algoTask );
     } else {
 
       AlgoExecutionTask theTask( this, serviceLocator(), m_algExecStateSvc );
       theTask.execute();
     }
 
     ON_DEBUG debug() << "Algorithm " << index2algname( iAlgo ) << " was submitted on event " << eventContext->evt()
                      << " in slot " << si << ". Algorithms scheduled are " << m_algosInFlight << endmsg;
 
     state = AState::SCHEDULED;
   } else {
 
     // Add the algorithm to the retry queue
     m_retryQueue.push( {iAlgo, si, eventContext, rank, nullptr} );
 
     state = AState::RESOURCELESS;
   }
 
   // Update alg state
   StatusCode updateSc = setAlgState( iAlgo, eventContext, state );
 
   ON_VERBOSE dumpSchedulerState( -1 );
 
   if ( updateSc.isSuccess() )
     ON_VERBOSE verbose() << "Promoting " << index2algname( iAlgo ) << " to " << state << " on slot " << si << endmsg;
   return updateSc;
 }

◆ eventFailed()

void AvalancheSchedulerSvc::eventFailed ( EventContext * eventContext )

private

Method to execute if an event failed.

It can be possible that an event fails.

In this case this method is called. It dumps the state of the scheduler and marks the event as finished.

Definition at line 735 of file AvalancheSchedulerSvc.cpp.

                                                                     {
   const uint slotIdx = eventContext->slot();
 
   error() << "Event " << eventContext->evt() << " on slot " << slotIdx << " failed" << endmsg;
 
   dumpSchedulerState( msgLevel( MSG::VERBOSE ) ? -1 : slotIdx );
 
   // dump temporal and topological precedence analysis (if enabled in the PrecedenceSvc)
   m_precSvc->dumpPrecedenceRules( m_eventSlots[slotIdx] );
 
   // Push into the finished events queue the failed context
   m_eventSlots[slotIdx].complete = true;
   m_finishedEvents.push( m_eventSlots[slotIdx].eventContext.release() );
 }

◆ finalize()

StatusCode AvalancheSchedulerSvc::finalize ( )

override

Finalise.

Here the scheduler is deactivated and the thread joined.

Definition at line 342 of file AvalancheSchedulerSvc.cpp.

                                            {
 
   StatusCode sc( Service::finalize() );
   if ( sc.isFailure() ) warning() << "Base class could not be finalized" << endmsg;
 
   sc = deactivate();
   if ( sc.isFailure() ) warning() << "Scheduler could not be deactivated" << endmsg;
 
   info() << "Joining Scheduler thread" << endmsg;
   m_thread.join();
 
   // Final error check after thread pool termination
   if ( m_isActive == FAILURE ) {
     error() << "problems in scheduler thread" << endmsg;
     return StatusCode::FAILURE;
   }
 
   return sc;
 }

◆ freeSlots()

unsigned int AvalancheSchedulerSvc::freeSlots ( )

override

Get free slots number.

Definition at line 530 of file AvalancheSchedulerSvc.cpp.

530 { return std::max( m_freeSlots.load(), 0 ); }

AvalancheSchedulerSvc::m_freeSlots

std::atomic_int m_freeSlots

Atomic to account for asyncronous updates by the scheduler wrt the rest.

Definition: AvalancheSchedulerSvc.h:227

std::max

T max(T... args)

◆ index2algname()

const std::string& AvalancheSchedulerSvc::index2algname ( unsigned int index )

inlineprivate

Convert an integer to a name.

Definition at line 209 of file AvalancheSchedulerSvc.h.

209 { return m_algname_vect[index]; };

AvalancheSchedulerSvc::m_algname_vect

std::vector< std::string > m_algname_vect

Vector to bookkeep the information necessary to the index2name conversion.

Definition: AvalancheSchedulerSvc.h:209

◆ initialize()

StatusCode AvalancheSchedulerSvc::initialize ( )

override

Initialise.

Here, among some "bureaucracy" operations, the scheduler is activated, executing the activate() function in a new thread.

In addition the algorithms list is acquired from the algResourcePool.

Definition at line 77 of file AvalancheSchedulerSvc.cpp.

                                              {
 
   // Initialise mother class (read properties, ...)
   StatusCode sc( Service::initialize() );
   if ( sc.isFailure() ) warning() << "Base class could not be initialized" << endmsg;
 
   // Get hold of the TBBSvc. This should initialize the thread pool
   m_threadPoolSvc = serviceLocator()->service( "ThreadPoolSvc" );
   if ( !m_threadPoolSvc.isValid() ) {
     fatal() << "Error retrieving ThreadPoolSvc" << endmsg;
     return StatusCode::FAILURE;
   }
 
   // Activate the scheduler in another thread.
   info() << "Activating scheduler in a separate thread" << endmsg;
   m_thread = std::thread( [this]() { this->activate(); } );
 
   while ( m_isActive != ACTIVE ) {
     if ( m_isActive == FAILURE ) {
       fatal() << "Terminating initialization" << endmsg;
       return StatusCode::FAILURE;
     } else {
       ON_DEBUG debug() << "Waiting for AvalancheSchedulerSvc to activate" << endmsg;
       sleep( 1 );
     }
   }
 
   if ( m_enableCondSvc ) {
     // Get hold of the CondSvc
     m_condSvc = serviceLocator()->service( "CondSvc" );
     if ( !m_condSvc.isValid() ) {
       warning() << "No CondSvc found, or not enabled. "
                 << "Will not manage CondAlgorithms" << endmsg;
       m_enableCondSvc = false;
     }
   }
 
   // Get the algo resource pool
   m_algResourcePool = serviceLocator()->service( "AlgResourcePool" );
   if ( !m_algResourcePool.isValid() ) {
     fatal() << "Error retrieving AlgoResourcePool" << endmsg;
     return StatusCode::FAILURE;
   }
 
   m_algExecStateSvc = serviceLocator()->service( "AlgExecStateSvc" );
   if ( !m_algExecStateSvc.isValid() ) {
     fatal() << "Error retrieving AlgExecStateSvc" << endmsg;
     return StatusCode::FAILURE;
   }
 
   // Get Whiteboard
   m_whiteboard = serviceLocator()->service( m_whiteboardSvcName );
   if ( !m_whiteboard.isValid() ) {
     fatal() << "Error retrieving EventDataSvc interface IHiveWhiteBoard." << endmsg;
     return StatusCode::FAILURE;
   }
 
   // Get dedicated scheduler for I/O-bound algorithms
   if ( m_useIOBoundAlgScheduler ) {
     m_IOBoundAlgScheduler = serviceLocator()->service( m_IOBoundAlgSchedulerSvcName );
     if ( !m_IOBoundAlgScheduler.isValid() )
       fatal() << "Error retrieving IOBoundSchedulerAlgSvc interface IAccelerator." << endmsg;
   }
 
   // Set the MaxEventsInFlight parameters from the number of WB stores
   m_maxEventsInFlight = m_whiteboard->getNumberOfStores();
 
   // Set the number of free slots
   m_freeSlots = m_maxEventsInFlight;
 
   // Get the list of algorithms
   const std::list<IAlgorithm*>& algos      = m_algResourcePool->getFlatAlgList();
   const unsigned int            algsNumber = algos.size();
   if ( algsNumber != 0 ) {
     info() << "Found " << algsNumber << " algorithms" << endmsg;
   } else {
     error() << "No algorithms found" << endmsg;
     return StatusCode::FAILURE;
   }
 
   /* Dependencies
    1) Look for handles in algo, if none
    2) Assume none are required
   */
 
   DataObjIDColl globalInp, globalOutp;
 
   // figure out all outputs
   for ( IAlgorithm* ialgoPtr : algos ) {
     Gaudi::Algorithm* algoPtr = dynamic_cast<Gaudi::Algorithm*>( ialgoPtr );
     if ( !algoPtr ) {
       fatal() << "Could not convert IAlgorithm into Gaudi::Algorithm: this will result in a crash." << endmsg;
       return StatusCode::FAILURE;
     }
     for ( auto id : algoPtr->outputDataObjs() ) globalOutp.insert( id );
   }
 
   std::ostringstream ostdd;
   ostdd << "Data Dependencies for Algorithms:";
 
   std::map<std::string, DataObjIDColl> algosDependenciesMap;
   for ( IAlgorithm* ialgoPtr : algos ) {
     Gaudi::Algorithm* algoPtr = dynamic_cast<Gaudi::Algorithm*>( ialgoPtr );
     if ( nullptr == algoPtr ) {
       fatal() << "Could not convert IAlgorithm into Gaudi::Algorithm for " << ialgoPtr->name()
               << ": this will result in a crash." << endmsg;
       return StatusCode::FAILURE;
     }
 
     ostdd << "\n  " << algoPtr->name();
 
     DataObjIDColl algoDependencies;
     if ( !algoPtr->inputDataObjs().empty() || !algoPtr->outputDataObjs().empty() ) {
       for ( const DataObjID* idp : sortedDataObjIDColl( algoPtr->inputDataObjs() ) ) {
         DataObjID id = *idp;
         ostdd << "\n    o INPUT  " << id;
         if ( id.key().find( ":" ) != std::string::npos ) {
           ostdd << " contains alternatives which require resolution...\n";
           auto tokens = boost::tokenizer<boost::char_separator<char>>{id.key(), boost::char_separator<char>{":"}};
           auto itok   = std::find_if( tokens.begin(), tokens.end(), [&]( const std::string& t ) {
             return globalOutp.find( DataObjID{t} ) != globalOutp.end();
           } );
           if ( itok != tokens.end() ) {
             ostdd << "found matching output for " << *itok << " -- updating scheduler info\n";
             id.updateKey( *itok );
           } else {
             error() << "failed to find alternate in global output list"
                     << " for id: " << id << " in Alg " << algoPtr->name() << endmsg;
             m_showDataDeps = true;
           }
         }
         algoDependencies.insert( id );
         globalInp.insert( id );
       }
       for ( const DataObjID* id : sortedDataObjIDColl( algoPtr->outputDataObjs() ) ) {
         ostdd << "\n    o OUTPUT " << *id;
         if ( id->key().find( ":" ) != std::string::npos ) {
           error() << " in Alg " << algoPtr->name() << " alternatives are NOT allowed for outputs! id: " << *id
                   << endmsg;
           m_showDataDeps = true;
         }
       }
     } else {
       ostdd << "\n      none";
     }
     algosDependenciesMap[algoPtr->name()] = algoDependencies;
   }
 
   if ( m_showDataDeps ) { info() << ostdd.str() << endmsg; }
 
   // Check if we have unmet global input dependencies, and, optionally, heal them
   // WARNING: this step must be done BEFORE the Precedence Service is initialized
   if ( m_checkDeps ) {
     DataObjIDColl unmetDep;
     for ( auto o : globalInp )
       if ( globalOutp.find( o ) == globalOutp.end() ) unmetDep.insert( o );
 
     if ( unmetDep.size() > 0 ) {
 
       std::ostringstream ost;
       for ( const DataObjID* o : sortedDataObjIDColl( unmetDep ) ) {
         ost << "\n   o " << *o << "    required by Algorithm: ";
 
         for ( const auto& p : algosDependenciesMap )
           if ( p.second.find( *o ) != p.second.end() ) ost << "\n       * " << p.first;
       }
 
       if ( !m_useDataLoader.empty() ) {
 
         // Find the DataLoader Alg
         IAlgorithm* dataLoaderAlg( nullptr );
         for ( IAlgorithm* algo : algos )
           if ( algo->name() == m_useDataLoader ) {
             dataLoaderAlg = algo;
             break;
           }
 
         if ( dataLoaderAlg == nullptr ) {
           fatal() << "No DataLoader Algorithm \"" << m_useDataLoader.value()
                   << "\" found, and unmet INPUT dependencies "
                   << "detected:\n"
                   << ost.str() << endmsg;
           return StatusCode::FAILURE;
         }
 
         info() << "Will attribute the following unmet INPUT dependencies to \"" << dataLoaderAlg->type() << "/"
                << dataLoaderAlg->name() << "\" Algorithm" << ost.str() << endmsg;
 
         // Set the property Load of DataLoader Alg
         Gaudi::Algorithm* dataAlg = dynamic_cast<Gaudi::Algorithm*>( dataLoaderAlg );
         if ( !dataAlg ) {
           fatal() << "Unable to dcast DataLoader \"" << m_useDataLoader.value() << "\" IAlg to Gaudi::Algorithm"
                   << endmsg;
           return StatusCode::FAILURE;
         }
 
         for ( auto& id : unmetDep ) {
           ON_DEBUG debug() << "adding OUTPUT dep \"" << id << "\" to " << dataLoaderAlg->type() << "/"
                            << dataLoaderAlg->name() << endmsg;
           dataAlg->addDependency( id, Gaudi::DataHandle::Writer );
         }
 
       } else {
         fatal() << "Auto DataLoading not requested, "
                 << "and the following unmet INPUT dependencies were found:" << ost.str() << endmsg;
         return StatusCode::FAILURE;
       }
 
     } else {
       info() << "No unmet INPUT data dependencies were found" << endmsg;
     }
   }
 
   // Get the precedence service
   m_precSvc = serviceLocator()->service( "PrecedenceSvc" );
   if ( !m_precSvc.isValid() ) {
     fatal() << "Error retrieving PrecedenceSvc" << endmsg;
     return StatusCode::FAILURE;
   }
   const PrecedenceSvc* precSvc = dynamic_cast<const PrecedenceSvc*>( m_precSvc.get() );
   if ( !precSvc ) {
     fatal() << "Unable to dcast PrecedenceSvc" << endmsg;
     return StatusCode::FAILURE;
   }
 
   // Fill the containers to convert algo names to index
   m_algname_vect.resize( algsNumber );
   for ( IAlgorithm* algo : algos ) {
     const std::string& name    = algo->name();
     auto               index   = precSvc->getRules()->getAlgorithmNode( name )->getAlgoIndex();
     m_algname_index_map[name]  = index;
     m_algname_vect.at( index ) = name;
   }
 
   // Shortcut for the message service
   SmartIF<IMessageSvc> messageSvc( serviceLocator() );
   if ( !messageSvc.isValid() ) error() << "Error retrieving MessageSvc interface IMessageSvc." << endmsg;
 
   m_eventSlots.reserve( m_maxEventsInFlight );
   for ( size_t i = 0; i < m_maxEventsInFlight; ++i ) {
     m_eventSlots.emplace_back( algsNumber, precSvc->getRules()->getControlFlowNodeCounter(), messageSvc );
     m_eventSlots.back().complete = true;
   }
 
   if ( m_threadPoolSize > 1 ) { m_maxAlgosInFlight = (size_t)m_threadPoolSize; }
 
   // Clearly inform about the level of concurrency
   info() << "Concurrency level information:" << endmsg;
   info() << " o Number of events in flight: " << m_maxEventsInFlight << endmsg;
   info() << " o TBB thread pool size: " << m_threadPoolSize << endmsg;
 
   if ( m_showControlFlow ) m_precSvc->dumpControlFlow();
 
   if ( m_showDataFlow ) m_precSvc->dumpDataFlow();
 
   // Simulate execution flow
   if ( m_simulateExecution ) m_precSvc->simulate( m_eventSlots[0] );
 
   return sc;
 }

◆ isStalled()

bool AvalancheSchedulerSvc::isStalled ( const EventSlot & slot ) const

private

Check if scheduling in a particular slot is in a stall.

Check if we are in present of a stall condition for a particular slot.

This is the case when a slot has no actions queued in the actionsQueue, has no scheduled algorithms and has no algorithms with all of its dependencies satisfied.

Definition at line 717 of file AvalancheSchedulerSvc.cpp.

                                                                    {
 
   if ( !slot.algsStates.containsAny( {AState::DATAREADY, AState::SCHEDULED, AState::RESOURCELESS} ) &&
        !subSlotAlgsInStates( slot, {AState::DATAREADY, AState::SCHEDULED, AState::RESOURCELESS} ) ) {
 
     error() << "*** Stall detected in slot " << slot.eventContext->slot() << "! ***" << endmsg;
 
     return true;
   }
   return false;
 }

◆ popFinishedEvent()

StatusCode AvalancheSchedulerSvc::popFinishedEvent ( EventContext *& eventContext )

override

Blocks until an event is available.

Get a finished event or block until one becomes available.

Definition at line 536 of file AvalancheSchedulerSvc.cpp.

                                                                                 {
 
   // ON_DEBUG debug() << "popFinishedEvent: queue size: " << m_finishedEvents.size() << endmsg;
   if ( m_freeSlots.load() == (int)m_maxEventsInFlight || m_isActive == INACTIVE ) {
     // ON_DEBUG debug() << "freeslots: " << m_freeSlots << "/" << m_maxEventsInFlight
     //      << " active: " << m_isActive << endmsg;
     return StatusCode::FAILURE;
   } else {
     // ON_DEBUG debug() << "freeslots: " << m_freeSlots << "/" << m_maxEventsInFlight
     //      << " active: " << m_isActive << endmsg;
     m_finishedEvents.pop( eventContext );
     ++m_freeSlots;
     ON_DEBUG debug() << "Popped slot " << eventContext->slot() << " (event " << eventContext->evt() << ")" << endmsg;
     return StatusCode::SUCCESS;
   }
 }

◆ promoteToAsyncExecuted()

StatusCode AvalancheSchedulerSvc::promoteToAsyncExecuted	(	unsigned int	iAlgo,
		int	si,
		IAlgorithm *	algo,
		EventContext *	eventContext
	)

private

The call to this method is triggered only from within the IOBoundAlgTask.

Definition at line 1019 of file AvalancheSchedulerSvc.cpp.

                                                                                        {
   Gaudi::Hive::setCurrentContext( eventContext );
   StatusCode sc = m_algResourcePool->releaseAlgorithm( algo->name(), algo );
 
   if ( sc.isFailure() ) {
     error() << "[Asynchronous]  [Event " << eventContext->evt() << ", Slot " << eventContext->slot() << "] "
             << "Instance of algorithm " << algo->name() << " could not be properly put back." << endmsg;
     return StatusCode::FAILURE;
   }
 
   --m_IOBoundAlgosInFlight;
 
   ON_DEBUG debug() << "[Asynchronous] Trying to handle execution result of " << algo->name() << " on slot " << si
                    << endmsg;
 
   const AlgExecState& algstate = m_algExecStateSvc->algExecState( algo, *eventContext );
   AState              state    = algstate.execStatus().isSuccess()
                      ? ( algstate.filterPassed() ? AState::EVTACCEPTED : AState::EVTREJECTED )
                      : AState::ERROR;
 
   // Update alg state and iterate PrecedenceSvc
   sc = setAlgState( iAlgo, eventContext, state, true );
 
   ON_VERBOSE if ( sc.isSuccess() ) verbose()
       << "[Asynchronous] Promoting " << algo->name() << " on slot " << si << " to " << state << endmsg;
 
   ON_DEBUG debug() << "[Asynchronous] Algorithm " << algo->name() << " executed in slot " << si
                    << ". Algorithms scheduled are " << m_IOBoundAlgosInFlight << endmsg;
 
   // Prompt a call to updateStates
   m_needsUpdate.store( true );
   return sc;
 }

◆ promoteToAsyncScheduled()

StatusCode AvalancheSchedulerSvc::promoteToAsyncScheduled	(	unsigned int	iAlgo,
		int	si,
		EventContext *	eventContext
	)

private

Definition at line 923 of file AvalancheSchedulerSvc.cpp.

                                                                                                                   {
 
   if ( m_IOBoundAlgosInFlight == m_maxIOBoundAlgosInFlight ) return StatusCode::FAILURE;
 
   // bool IOBound = m_precSvc->isBlocking(algName);
 
   const std::string& algName( index2algname( iAlgo ) );
   IAlgorithm*        ialgoPtr = nullptr;
   StatusCode         sc( m_algResourcePool->acquireAlgorithm( algName, ialgoPtr ) );
 
   if ( sc.isSuccess() ) { // if we managed to get an algorithm instance try to schedule it
 
     ++m_IOBoundAlgosInFlight;
     auto promote2ExecutedClosure = [this, iAlgo, ialgoPtr, eventContext]() {
       this->m_actionsQueue.push( [this, iAlgo, ialgoPtr, eventContext]() {
         return this->AvalancheSchedulerSvc::promoteToAsyncExecuted( iAlgo, eventContext->slot(), ialgoPtr,
                                                                     eventContext );
       } );
       return StatusCode::SUCCESS;
     };
     // Can we use tbb-based overloaded new-operator for a "custom" task (an algorithm wrapper, not derived from
     // tbb::task)? it seems it works..
 
     // FIXME - The memory allocation here is causing memory leaks as detected by the gcc leak sanitizer
     //
     // clang-format off
     // Direct leak of 224 byte(s) in 7 object(s) allocated from:
     //   #0 0x7fc0cb524da8 in operator new(unsigned long) /afs/cern.ch/cms/CAF/CMSCOMM/COMM_ECAL/dkonst/GCC/build/contrib/gcc-8.2.0/src/gcc/8.2.0/libsanitizer/lsan/lsan_interceptors.cc:229
     //   #1 0x7fc0ba979f7b in function<AvalancheSchedulerSvc::promoteToAsyncScheduled(unsigned int, int, EventContext*)::<lambda()> > /cvmfs/lhcb.cern.ch/lib/lcg/releases/gcc/8.2.0-3fa06/x86_64-centos7/include/c++/8.2.0/bits/std_function.h:249
     //   #2 0x7fc0ba97d181 in AvalancheSchedulerSvc::promoteToAsyncScheduled(unsigned int, int, EventContext*) ../GaudiHive/src/AvalancheSchedulerSvc.cpp:969
     //   #3 0x7fc0ba98354d in AvalancheSchedulerSvc::updateStates(int, int, int, int) ../GaudiHive/src/AvalancheSchedulerSvc.cpp:660
     // clang-format on
     //
     // These leaks are currently suppressed in Gaudi/job/Gaudi-LSan.supp - remove entry there to reactivate
     //
     IOBoundAlgTask* theTask = new ( tbb::task::allocate_root() )
         IOBoundAlgTask( ialgoPtr, *eventContext, serviceLocator(), m_algExecStateSvc, promote2ExecutedClosure );
     m_IOBoundAlgScheduler->push( *theTask );
     //
     // FIXME
 
     ON_DEBUG debug() << "[Asynchronous] Algorithm " << algName << " was submitted on event " << eventContext->evt()
                      << " in slot " << si << ". algorithms scheduled are " << m_IOBoundAlgosInFlight << endmsg;
 
     // Update alg state
     StatusCode updateSc = setAlgState( iAlgo, eventContext, AState::SCHEDULED );
 
     ON_VERBOSE if ( updateSc.isSuccess() ) verbose()
         << "[Asynchronous] Promoting " << algName << " to SCHEDULED on slot " << si << endmsg;
     return updateSc;
   } else {
     ON_DEBUG debug() << "[Asynchronous] Could not acquire instance for algorithm " << index2algname( iAlgo )
                      << " on slot " << si << endmsg;
     return sc;
   }
 }

◆ promoteToExecuted()

StatusCode AvalancheSchedulerSvc::promoteToExecuted	(	unsigned int	iAlgo,
		int	si,
		EventContext *	eventContext
	)

private

The call to this method is triggered only from within the AlgoExecutionTask.

Definition at line 985 of file AvalancheSchedulerSvc.cpp.

                                                                                                             {
 
   const std::string& algName( index2algname( iAlgo ) );
 
   Gaudi::Hive::setCurrentContext( eventContext );
 
   --m_algosInFlight;
 
   ON_DEBUG debug() << "Trying to handle execution result of " << algName << " on slot " << si << endmsg;
 
   const AlgExecState& algstate = m_algExecStateSvc->algExecState( algName, *eventContext );
   AState              state    = algstate.execStatus().isSuccess()
                      ? ( algstate.filterPassed() ? AState::EVTACCEPTED : AState::EVTREJECTED )
                      : AState::ERROR;
 
   // Update alg state and iterate PrecedenceSvc
   StatusCode sc = setAlgState( iAlgo, eventContext, state, true );
 
   ON_VERBOSE if ( sc.isSuccess() ) verbose()
       << "Promoting " << algName << " on slot " << si << " to " << state << endmsg;
 
   ON_DEBUG debug() << "Algorithm " << algName << " executed in slot " << si << ". Algorithms scheduled are "
                    << m_algosInFlight << endmsg;
 
   // Prompt a call to updateStates
   m_needsUpdate.store( true );
   return sc;
 }

◆ promoteToFinished()

StatusCode AvalancheSchedulerSvc::promoteToFinished	(	unsigned int	iAlgo,
		int	si
	)

private

◆ pushNewEvent()

StatusCode AvalancheSchedulerSvc::pushNewEvent ( EventContext * eventContext )

override

Make an event available to the scheduler.

Add event to the scheduler.

There are two cases possible: 1) No slot is free. A StatusCode::FAILURE is returned. 2) At least one slot is free. An action which resets the slot and kicks off its update is queued.

Definition at line 461 of file AvalancheSchedulerSvc.cpp.

                                                                            {
 
   if ( !eventContext ) {
     fatal() << "Event context is nullptr" << endmsg;
     return StatusCode::FAILURE;
   }
 
   if ( m_freeSlots.load() == 0 ) {
     ON_DEBUG debug() << "A free processing slot could not be found." << endmsg;
     return StatusCode::FAILURE;
   }
 
   // no problem as push new event is only called from one thread (event loop manager)
   --m_freeSlots;
 
   auto action = [this, eventContext]() -> StatusCode {
     // Event processing slot forced to be the same as the wb slot
     const unsigned int thisSlotNum = eventContext->slot();
     EventSlot&         thisSlot    = m_eventSlots[thisSlotNum];
     if ( !thisSlot.complete ) {
       fatal() << "The slot " << thisSlotNum << " is supposed to be a finished event but it's not" << endmsg;
       return StatusCode::FAILURE;
     }
 
     ON_DEBUG debug() << "Executing event " << eventContext->evt() << " on slot " << thisSlotNum << endmsg;
     thisSlot.reset( eventContext );
 
     // Result status code:
     StatusCode result = StatusCode::SUCCESS;
 
     // promote to CR and DR the initial set of algorithms
     Cause cs = {Cause::source::Root, "RootDecisionHub"};
     if ( m_precSvc->iterate( thisSlot, cs ).isFailure() ) {
       error() << "Failed to call IPrecedenceSvc::iterate for slot " << thisSlotNum << endmsg;
       result = StatusCode::FAILURE;
     }
 
     if ( this->updateStates().isFailure() ) {
       error() << "Failed to call AvalancheSchedulerSvc::updateStates for slot " << thisSlotNum << endmsg;
       result = StatusCode::FAILURE;
     }
 
     return result;
   }; // end of lambda
 
   // Kick off the scheduling!
   ON_VERBOSE {
     verbose() << "Pushing the action to update the scheduler for slot " << eventContext->slot() << endmsg;
     verbose() << "Free slots available " << m_freeSlots.load() << endmsg;
   }
 
   m_actionsQueue.push( action );
 
   return StatusCode::SUCCESS;
 }

◆ pushNewEvents()

StatusCode AvalancheSchedulerSvc::pushNewEvents ( std::vector< EventContext * > & eventContexts )

override

Definition at line 519 of file AvalancheSchedulerSvc.cpp.

                                                                                          {
   StatusCode sc;
   for ( auto context : eventContexts ) {
     sc = pushNewEvent( context );
     if ( sc != StatusCode::SUCCESS ) return sc;
   }
   return sc;
 }

◆ scheduleEventView()

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

overridevirtual

Method to inform the scheduler about event views.

Definition at line 1058 of file AvalancheSchedulerSvc.cpp.

                                                                                                {
   //  Prevent view nesting
   if ( sourceContext->usesSubSlot() ) {
     fatal() << "Attempted to nest EventViews at node " << nodeName << ": this is not supported" << endmsg;
     return StatusCode::FAILURE;
   }
 
   ON_VERBOSE verbose() << "Queuing a view for [" << viewContext.get() << "]" << endmsg;
 
   // It's not possible to create an std::functional from a move-capturing lambda
   // So, we have to release the unique pointer
   auto action = [this, slotIndex = sourceContext->slot(), viewContextPtr = viewContext.release(),
                  &nodeName]() -> StatusCode {
     // Attach the sub-slot to the top-level slot
     EventSlot& topSlot = this->m_eventSlots[slotIndex];
 
     if ( viewContextPtr ) {
       // Re-create the unique pointer
       auto viewContext = std::unique_ptr<EventContext>( viewContextPtr );
       topSlot.addSubSlot( std::move( viewContext ), nodeName );
       return StatusCode::SUCCESS;
     } else {
       // Disable the view node if there are no views
       topSlot.disableSubSlots( nodeName );
       return StatusCode::SUCCESS;
     }
   };
 
   m_actionsQueue.push( std::move( action ) );
 
   return StatusCode::SUCCESS;
 }

◆ setAlgState()

StatusCode AvalancheSchedulerSvc::setAlgState	(	unsigned int	iAlgo,
		EventContext *	contextPtr,
		AState	state,
		bool	iterate = `false`
	)

private

Definition at line 691 of file AvalancheSchedulerSvc.cpp.

                                                               {
   StatusCode updateSc;
   EventSlot& thisSlot = m_eventSlots[contextPtr->slot()];
   Cause      cs       = {Cause::source::Task, index2algname( iAlgo )};
   if ( contextPtr->usesSubSlot() ) {
     // Sub-slot
     size_t const subSlotIndex = contextPtr->subSlot();
     updateSc                  = thisSlot.allSubSlots[subSlotIndex].algsStates.set( iAlgo, state );
     if ( updateSc.isSuccess() && iterate ) updateSc = m_precSvc->iterate( thisSlot.allSubSlots[subSlotIndex], cs );
   } else {
     // Event level (standard behaviour)
     updateSc = thisSlot.algsStates.set( iAlgo, state );
     if ( updateSc.isSuccess() && iterate ) updateSc = m_precSvc->iterate( thisSlot, cs );
   }
   return updateSc;
 }

◆ tryPopFinishedEvent()

StatusCode AvalancheSchedulerSvc::tryPopFinishedEvent ( EventContext *& eventContext )

override

Try to fetch an event from the scheduler.

Try to get a finished event, if not available just return a failure.

Definition at line 557 of file AvalancheSchedulerSvc.cpp.

                                                                                    {
 
   if ( m_finishedEvents.try_pop( eventContext ) ) {
     ON_DEBUG debug() << "Try Pop successful slot " << eventContext->slot() << "(event " << eventContext->evt() << ")"
                      << endmsg;
     ++m_freeSlots;
     return StatusCode::SUCCESS;
   }
   return StatusCode::FAILURE;
 }

◆ updateStates()

StatusCode AvalancheSchedulerSvc::updateStates ( )

private

Loop on algorithm in the slots and promote them to successive states.

Update the state of the algorithms.

The oldest events are checked before the newest, in order to reduce the event backlog. To check if the event is finished the algorithm checks if:

No algorithms have been signed off by the control flow
No algorithms have been signed off by the data flow
No algorithms have been scheduled

Definition at line 580 of file AvalancheSchedulerSvc.cpp.

                                                {
 
   StatusCode global_sc( StatusCode::SUCCESS );
 
   // Retry algs
   AlgQueueEntry queuePop;
   const size_t  retries = m_retryQueue.size();
   for ( unsigned int retryIndex = 0; retryIndex < retries; ++retryIndex ) {
 
     queuePop = m_retryQueue.front();
     m_retryQueue.pop();
 
     global_sc = enqueue( queuePop.algIndex, queuePop.slotIndex, queuePop.contextPtr );
   }
 
   // Loop over all slots
   for ( EventSlot& thisSlot : m_eventSlots ) {
 
     // Ignore slots without a valid context (relevant when populating scheduler for first time)
     if ( !thisSlot.eventContext ) continue;
 
     int iSlot = thisSlot.eventContext->slot();
 
     // Cache the states of the algos to improve readability and performance
     AlgsExecutionStates& thisAlgsStates = thisSlot.algsStates;
 
     StatusCode partial_sc( StatusCode::FAILURE, true );
 
     // Perform DR->SCHEDULED
     for ( auto it = thisAlgsStates.begin( AState::DATAREADY ); it != thisAlgsStates.end( AState::DATAREADY ); ++it ) {
       uint algIndex = *it;
 
       bool IOBound = false;
       if ( m_useIOBoundAlgScheduler ) IOBound = m_precSvc->isBlocking( index2algname( algIndex ) );
 
       if ( !IOBound )
         partial_sc = enqueue( algIndex, iSlot, thisSlot.eventContext.get() );
       else
         partial_sc = promoteToAsyncScheduled( algIndex, iSlot, thisSlot.eventContext.get() );
 
       ON_VERBOSE if ( partial_sc.isFailure() ) verbose()
           << "Could not apply transition from " << AState::DATAREADY << " for algorithm " << index2algname( algIndex )
           << " on processing slot " << iSlot << endmsg;
     }
 
     // Check for algorithms ready in sub-slots
     for ( auto& subslot : thisSlot.allSubSlots ) {
       auto& subslotStates = subslot.algsStates;
       for ( auto it = subslotStates.begin( AState::DATAREADY ); it != subslotStates.end( AState::DATAREADY ); ++it ) {
         uint algIndex{*it};
         partial_sc = enqueue( algIndex, iSlot, subslot.eventContext.get() );
         // The following verbosity is expensive when the number of sub-slots is high
         /*ON_VERBOSE if ( partial_sc.isFailure() ) verbose()
             << "Could not apply transition from " << AState::DATAREADY << " for algorithm " << index2algname( algIndex )
             << " on processing subslot " << subslot.eventContext->slot() << endmsg;*/
       }
     }
 
     if ( m_dumpIntraEventDynamics ) {
       std::stringstream s;
       s << "START, " << thisAlgsStates.sizeOfSubset( AState::CONTROLREADY ) << ", "
         << thisAlgsStates.sizeOfSubset( AState::DATAREADY ) << ", " << thisAlgsStates.sizeOfSubset( AState::SCHEDULED )
         << ", " << std::chrono::high_resolution_clock::now().time_since_epoch().count() << "\n";
       auto threads = ( m_threadPoolSize != -1 ) ? std::to_string( m_threadPoolSize )
 #if TBB_INTERFACE_VERSION_MAJOR < 12
                                                 : std::to_string( tbb::task_scheduler_init::default_num_threads() );
 #else
                                                 : std::to_string( std::thread::hardware_concurrency() );
 #endif // TBB_INTERFACE_VERSION_MAJOR < 12
       std::ofstream myfile;
       myfile.open( "IntraEventFSMOccupancy_" + threads + "T.csv", std::ios::app );
       myfile << s.str();
       myfile.close();
     }
 
     // Not complete because this would mean that the slot is already free!
     if ( m_precSvc->CFRulesResolved( thisSlot ) &&
          !thisSlot.algsStates.containsAny(
              {AState::CONTROLREADY, AState::DATAREADY, AState::SCHEDULED, AState::RESOURCELESS} ) &&
          !subSlotAlgsInStates( thisSlot,
                                {AState::CONTROLREADY, AState::DATAREADY, AState::SCHEDULED, AState::RESOURCELESS} ) &&
          !thisSlot.complete ) {
 
       thisSlot.complete = true;
       // if the event did not fail, add it to the finished events
       // otherwise it is taken care of in the error handling
       if ( m_algExecStateSvc->eventStatus( *thisSlot.eventContext ) == EventStatus::Success ) {
         ON_DEBUG debug() << "Event " << thisSlot.eventContext->evt() << " finished (slot "
                          << thisSlot.eventContext->slot() << ")." << endmsg;
         m_finishedEvents.push( thisSlot.eventContext.release() );
       }
 
       // now let's return the fully evaluated result of the control flow
       ON_DEBUG debug() << m_precSvc->printState( thisSlot ) << endmsg;
 
       thisSlot.eventContext.reset( nullptr );
 
     } else if ( isStalled( thisSlot ) ) {
       m_algExecStateSvc->setEventStatus( EventStatus::AlgStall, *thisSlot.eventContext );
       eventFailed( thisSlot.eventContext.get() ); // can't release yet
     }
     partial_sc.ignore();
   } // end loop on slots
 
   ON_VERBOSE verbose() << "States Updated." << endmsg;
   m_needsUpdate.store( false );
   return global_sc;
 }

Friends And Related Function Documentation

◆ AlgoExecutionTask

friend class AlgoExecutionTask

friend

Definition at line 113 of file AvalancheSchedulerSvc.h.

Member Data Documentation

◆ m_actionsQueue

tbb::concurrent_bounded_queue<action> AvalancheSchedulerSvc::m_actionsQueue

private

Queue where closures are stored and picked for execution.

Definition at line 276 of file AvalancheSchedulerSvc.h.

◆ m_algExecStateSvc

SmartIF<IAlgExecStateSvc> AvalancheSchedulerSvc::m_algExecStateSvc

private

Algorithm execution state manager.

Definition at line 233 of file AvalancheSchedulerSvc.h.

◆ m_algname_index_map

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

private

Map to bookkeep the information necessary to the name2index conversion.

Definition at line 203 of file AvalancheSchedulerSvc.h.

◆ m_algname_vect

std::vector<std::string> AvalancheSchedulerSvc::m_algname_vect

private

Vector to bookkeep the information necessary to the index2name conversion.

Definition at line 209 of file AvalancheSchedulerSvc.h.

◆ m_algosInFlight

unsigned int AvalancheSchedulerSvc::m_algosInFlight = 0

private

Number of algorithms presently in flight.

Definition at line 239 of file AvalancheSchedulerSvc.h.

◆ m_algResourcePool

SmartIF<IAlgResourcePool> AvalancheSchedulerSvc::m_algResourcePool

private

Cache for the algorithm resource pool.

Definition at line 271 of file AvalancheSchedulerSvc.h.

◆ m_checkDeps

Gaudi::Property<bool> AvalancheSchedulerSvc::m_checkDeps {this, "CheckDependencies", false, "Runtime check of Algorithm Data Dependencies"}

private

Definition at line 170 of file AvalancheSchedulerSvc.h.

◆ m_condSvc

SmartIF<ICondSvc> AvalancheSchedulerSvc::m_condSvc

private

A shortcut to service for Conditions handling.

Definition at line 236 of file AvalancheSchedulerSvc.h.

◆ m_dumpIntraEventDynamics

Gaudi::Property<bool> AvalancheSchedulerSvc::m_dumpIntraEventDynamics

private

Initial value:

{this, "DumpIntraEventDynamics", false,

"Dump intra-event concurrency dynamics to csv file"}

Definition at line 165 of file AvalancheSchedulerSvc.h.

◆ m_enableCondSvc

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

private

Definition at line 175 of file AvalancheSchedulerSvc.h.

◆ m_eventSlots

std::vector<EventSlot> AvalancheSchedulerSvc::m_eventSlots

private

Vector of events slots.

Definition at line 224 of file AvalancheSchedulerSvc.h.

◆ m_finishedEvents

tbb::concurrent_bounded_queue<EventContext*> AvalancheSchedulerSvc::m_finishedEvents

private

Queue of finished events.

Definition at line 230 of file AvalancheSchedulerSvc.h.

◆ m_freeSlots

std::atomic_int AvalancheSchedulerSvc::m_freeSlots

private

Atomic to account for asyncronous updates by the scheduler wrt the rest.

Definition at line 227 of file AvalancheSchedulerSvc.h.

◆ m_IOBoundAlgosInFlight

unsigned int AvalancheSchedulerSvc::m_IOBoundAlgosInFlight = 0

private

Number of algorithms presently in flight.

Definition at line 242 of file AvalancheSchedulerSvc.h.

◆ m_IOBoundAlgScheduler

SmartIF<IAccelerator> AvalancheSchedulerSvc::m_IOBoundAlgScheduler

private

A shortcut to IO-bound algorithm scheduler.

Definition at line 221 of file AvalancheSchedulerSvc.h.

◆ m_IOBoundAlgSchedulerSvcName

Gaudi::Property<std::string> AvalancheSchedulerSvc::m_IOBoundAlgSchedulerSvcName {this, "IOBoundAlgSchedulerSvc", "IOBoundAlgSchedulerSvc"}

private

Definition at line 157 of file AvalancheSchedulerSvc.h.

◆ m_isActive

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

private

Flag to track if the scheduler is active or not.

Definition at line 197 of file AvalancheSchedulerSvc.h.

◆ m_maxAlgosInFlight

size_t AvalancheSchedulerSvc::m_maxAlgosInFlight {1}

private

Definition at line 304 of file AvalancheSchedulerSvc.h.

◆ m_maxEventsInFlight

size_t AvalancheSchedulerSvc::m_maxEventsInFlight {0}

private

Definition at line 303 of file AvalancheSchedulerSvc.h.

◆ m_maxIOBoundAlgosInFlight

Gaudi::Property<unsigned int> AvalancheSchedulerSvc::m_maxIOBoundAlgosInFlight

private

Initial value:

{this, "MaxIOBoundAlgosInFlight", 0,

"Maximum number of simultaneous I/O-bound algorithms"}

Definition at line 158 of file AvalancheSchedulerSvc.h.

◆ m_needsUpdate

std::atomic<bool> AvalancheSchedulerSvc::m_needsUpdate {true}

private

Definition at line 297 of file AvalancheSchedulerSvc.h.

◆ m_optimizationMode

Gaudi::Property<std::string> AvalancheSchedulerSvc::m_optimizationMode

private

Initial value:

{this, "Optimizer", "",

"The following modes are currently available: PCE, COD, DRE, E"}

Definition at line 163 of file AvalancheSchedulerSvc.h.

◆ m_precSvc

SmartIF<IPrecedenceSvc> AvalancheSchedulerSvc::m_precSvc

private

A shortcut to the Precedence Service.

Definition at line 215 of file AvalancheSchedulerSvc.h.

◆ m_retryQueue

std::queue<AlgQueueEntry> AvalancheSchedulerSvc::m_retryQueue

private

Definition at line 294 of file AvalancheSchedulerSvc.h.

◆ m_scheduledQueue

tbb::concurrent_priority_queue<AlgQueueEntry, AlgQueueSort> AvalancheSchedulerSvc::m_scheduledQueue

private

Queues for scheduled algorithms.

Definition at line 293 of file AvalancheSchedulerSvc.h.

◆ m_showControlFlow

Gaudi::Property<bool> AvalancheSchedulerSvc::m_showControlFlow

private

Initial value:

{this, "ShowControlFlow", false,

"Show the configuration of all Algorithms and Sequences"}

Definition at line 183 of file AvalancheSchedulerSvc.h.

◆ m_showDataDeps

Gaudi::Property<bool> AvalancheSchedulerSvc::m_showDataDeps

private

Initial value:

{this, "ShowDataDependencies", true,

"Show the INPUT and OUTPUT data dependencies of Algorithms"}

Definition at line 177 of file AvalancheSchedulerSvc.h.

◆ m_showDataFlow

Gaudi::Property<bool> AvalancheSchedulerSvc::m_showDataFlow

private

Initial value:

{this, "ShowDataFlow", false,

"Show the configuration of DataFlow between Algorithms"}

Definition at line 180 of file AvalancheSchedulerSvc.h.

◆ m_simulateExecution

Gaudi::Property<bool> AvalancheSchedulerSvc::m_simulateExecution

private

Initial value:

{
      this, "SimulateExecution", false,
      "Flag to perform single-pass simulation of execution flow before the actual execution"}

Definition at line 160 of file AvalancheSchedulerSvc.h.

◆ m_thread

std::thread AvalancheSchedulerSvc::m_thread

private

The thread in which the activate function runs.

Definition at line 200 of file AvalancheSchedulerSvc.h.

◆ m_threadPoolSize

Gaudi::Property<int> AvalancheSchedulerSvc::m_threadPoolSize

private

Initial value:

{
      this, "ThreadPoolSize", -1,
      "Size of the threadpool initialised by TBB; a value of -1 gives TBB the freedom to choose"}

Definition at line 153 of file AvalancheSchedulerSvc.h.

◆ m_threadPoolSvc

SmartIF<IThreadPoolSvc> AvalancheSchedulerSvc::m_threadPoolSvc

private

Definition at line 302 of file AvalancheSchedulerSvc.h.

◆ m_useDataLoader

Gaudi::Property<std::string> AvalancheSchedulerSvc::m_useDataLoader

private

Initial value:

{this, "DataLoaderAlg", "",

"Attribute unmet input dependencies to this DataLoader Algorithm"}

Definition at line 172 of file AvalancheSchedulerSvc.h.

◆ m_useIOBoundAlgScheduler

Gaudi::Property<bool> AvalancheSchedulerSvc::m_useIOBoundAlgScheduler

private

Initial value:

{this, "PreemptiveIOBoundTasks", false,

"Turn on preemptive way of scheduling of I/O-bound algorithms"}

Definition at line 167 of file AvalancheSchedulerSvc.h.

◆ m_verboseSubSlots

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

private

Definition at line 186 of file AvalancheSchedulerSvc.h.

◆ m_whiteboard

SmartIF<IHiveWhiteBoard> AvalancheSchedulerSvc::m_whiteboard

private

A shortcut to the whiteboard.

Definition at line 218 of file AvalancheSchedulerSvc.h.

◆ m_whiteboardSvcName

Gaudi::Property<std::string> AvalancheSchedulerSvc::m_whiteboardSvcName {this, "WhiteboardSvc", "EventDataSvc", "The whiteboard name"}

private

Definition at line 156 of file AvalancheSchedulerSvc.h.

The documentation for this class was generated from the following files:

GaudiHive/src/AvalancheSchedulerSvc.h
GaudiHive/src/AvalancheSchedulerSvc.cpp

Classes

Public Member Functions

Private Types

Private Member Functions

Private Attributes

Friends

Additional Inherited Members

Detailed Description

Introduction

Task precedence management

Scheduling principles

Credits

Member Typedef Documentation

◆ action

◆ AState

Member Enumeration Documentation

◆ ActivationState

Constructor & Destructor Documentation

◆ ~AvalancheSchedulerSvc()

Member Function Documentation

◆ activate()

◆ algname2index()

◆ deactivate()

◆ dumpSchedulerState()

◆ enqueue()

◆ eventFailed()

◆ finalize()

◆ freeSlots()

◆ index2algname()

◆ initialize()

◆ isStalled()

◆ popFinishedEvent()

◆ promoteToAsyncExecuted()

◆ promoteToAsyncScheduled()

◆ promoteToExecuted()

◆ promoteToFinished()

◆ pushNewEvent()

◆ pushNewEvents()

◆ scheduleEventView()

◆ setAlgState()

◆ tryPopFinishedEvent()

◆ updateStates()

Friends And Related Function Documentation

◆ AlgoExecutionTask

Member Data Documentation

◆ m_actionsQueue

◆ m_algExecStateSvc

◆ m_algname_index_map

◆ m_algname_vect

◆ m_algosInFlight

◆ m_algResourcePool

◆ m_checkDeps

◆ m_condSvc

◆ m_dumpIntraEventDynamics

◆ m_enableCondSvc

◆ m_eventSlots

◆ m_finishedEvents

◆ m_freeSlots

◆ m_IOBoundAlgosInFlight

◆ m_IOBoundAlgScheduler

◆ m_IOBoundAlgSchedulerSvcName

◆ m_isActive

◆ m_maxAlgosInFlight

◆ m_maxEventsInFlight

◆ m_maxIOBoundAlgosInFlight

◆ m_needsUpdate

◆ m_optimizationMode

◆ m_precSvc

◆ m_retryQueue

◆ m_scheduledQueue

◆ m_showControlFlow

◆ m_showDataDeps

◆ m_showDataFlow

◆ m_simulateExecution

◆ m_thread

◆ m_threadPoolSize

◆ m_threadPoolSvc

◆ m_useDataLoader

◆ m_useIOBoundAlgScheduler

◆ m_verboseSubSlots