| The Gaudi Framework v32r1 (f65d50dc)

Public Member Functions | Private Types | Private Member Functions | Private Attributes | List of all members
AvalancheSchedulerSvc Class Reference

`#include <src/AvalancheSchedulerSvc.h>`

Inheritance diagram for AvalancheSchedulerSvc:
[legend]
Collaboration diagram for AvalancheSchedulerSvc:
[legend]

## 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 (int si=-1, int algo_index=-1, int sub_slot=-1, int source_slot=-1)
| Loop on algorithm in the slots and promote them to successive states (-1 for algo_index means skipping an update of the Control Flow state) More...
StatusCode | promoteToScheduled (unsigned int iAlgo, int si, EventContext *)
| Algorithm promotion. More...
StatusCode | promoteToAsyncScheduled (unsigned int iAlgo, int si, EventContext *)
StatusCode | promoteToExecuted (unsigned int iAlgo, int si, IAlgorithm *algo, 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...
std::vector< unsigned int > | m_actionsCounts
| Bookkeeping of the number of actions in flight per slot. More...
SmartIF< IThreadPoolSvc > | m_threadPoolSvc
size_t | m_maxEventsInFlight {0}
size_t | m_maxAlgosInFlight {1}

## 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 100 of file AvalancheSchedulerSvc.h.

## Member Typedef Documentation

## ◆ action

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

Definition at line 135 of file AvalancheSchedulerSvc.h.

## ◆ AState

using AvalancheSchedulerSvc::AState = AlgsExecutionStates::State
| private

Definition at line 134 of file AvalancheSchedulerSvc.h.

## Member Enumeration Documentation

## ◆ ActivationState

enum AvalancheSchedulerSvc::ActivationState
| private

Enumerator
INACTIVE |
ACTIVE |
FAILURE |

Definition at line 137 of file AvalancheSchedulerSvc.h.

137 { INACTIVE = 0, ACTIVE = 1, FAILURE = 2 };
AvalancheSchedulerSvc::INACTIVE
Definition: AvalancheSchedulerSvc.h:137
AvalancheSchedulerSvc::ACTIVE
Definition: AvalancheSchedulerSvc.h:137
AvalancheSchedulerSvc::FAILURE
Definition: AvalancheSchedulerSvc.h:137

## Constructor & Destructor Documentation

## ◆ ~AvalancheSchedulerSvc()

AvalancheSchedulerSvc::~AvalancheSchedulerSvc | ( |  | ) |
| override default

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 368 of file AvalancheSchedulerSvc.cpp.

368 {
369
370 ON_DEBUG debug() << "AvalancheSchedulerSvc::activate()" << endmsg;
371
372 if ( m_threadPoolSvc->initPool( m_threadPoolSize ).isFailure() ) {
373 error() << "problems initializing ThreadPoolSvc" << endmsg;
374 m_isActive = FAILURE;
375 return;
376 }
377
378 // Wait for actions pushed into the queue by finishing tasks.
379 action thisAction;
380 StatusCode sc( StatusCode::SUCCESS );
381
382 m_isActive = ACTIVE;
383
384 // Continue to wait if the scheduler is running or there is something to do
385 ON_DEBUG debug() << "Start checking the actionsQueue" << endmsg;
386 while ( m_isActive == ACTIVE || m_actionsQueue.size() != 0 ) {
387 m_actionsQueue.pop( thisAction );
388 sc = thisAction();
389 ON_VERBOSE {
390 if ( sc.isFailure() )
391 verbose() << "Action did not succeed (which is not bad per se)." << endmsg;
392 else
393 verbose() << "Action succeeded." << endmsg;
394 }
395 else sc.ignore();
396 }
397
398 ON_DEBUG debug() << "Terminating thread-pool resources" << endmsg;
399 if ( m_threadPoolSvc->terminatePool().isFailure() ) {
400 error() << "Problems terminating thread pool" << endmsg;
401 m_isActive = FAILURE;
402 }
403 }
IThreadPoolSvc::initPool
virtual StatusCode initPool(const int &poolSize)=0
Initializes the thread pool.
ON_DEBUG
#define ON_DEBUG
Definition: AvalancheSchedulerSvc.cpp:31
StatusCode::SUCCESS
constexpr static const auto SUCCESS
Definition: StatusCode.h:85
AvalancheSchedulerSvc::ACTIVE
Definition: AvalancheSchedulerSvc.h:137
AvalancheSchedulerSvc::m_isActive
std::atomic< ActivationState > m_isActive
Flag to track if the scheduler is active or not.
Definition: AvalancheSchedulerSvc.h:183
IThreadPoolSvc::terminatePool
virtual StatusCode terminatePool()=0
Finalize the thread pool.
AvalancheSchedulerSvc::action
std::function< StatusCode()> action
Definition: AvalancheSchedulerSvc.h:135
CommonMessagingBase::error
MsgStream & error() const
shortcut for the method msgStream(MSG::ERROR)
Definition: CommonMessaging.h:98
StatusCode
This class is used for returning status codes from appropriate routines.
Definition: StatusCode.h:50
CommonMessagingBase::verbose
MsgStream & verbose() const
shortcut for the method msgStream(MSG::VERBOSE)
Definition: CommonMessaging.h:110
CommonMessagingBase::debug
MsgStream & debug() const
shortcut for the method msgStream(MSG::DEBUG)
Definition: CommonMessaging.h:107
AvalancheSchedulerSvc::m_threadPoolSize
Gaudi::Property< int > m_threadPoolSize
Definition: AvalancheSchedulerSvc.h:139
AvalancheSchedulerSvc::m_threadPoolSvc
SmartIF< IThreadPoolSvc > m_threadPoolSvc
Definition: AvalancheSchedulerSvc.h:268
StatusCode::isFailure
bool isFailure() const
Definition: StatusCode.h:130
AvalancheSchedulerSvc::FAILURE
Definition: AvalancheSchedulerSvc.h:137
endmsg
MsgStream & endmsg(MsgStream &s)
MsgStream Modifier: endmsg. Calls the output method of the MsgStream.
Definition: MsgStream.h:192
AvalancheSchedulerSvc::m_actionsQueue
tbb::concurrent_bounded_queue< action > m_actionsQueue
Queue where closures are stored and picked for execution.
Definition: AvalancheSchedulerSvc.h:260
ON_VERBOSE
#define ON_VERBOSE
Definition: AvalancheSchedulerSvc.cpp:32

## ◆ algname2index()

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

Convert a name to an integer.

Definition at line 443 of file AvalancheSchedulerSvc.cpp.

443 {
444 unsigned int index = m_algname_index_map[algoname];
445 return index;
446 }
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:192

## ◆ 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 413 of file AvalancheSchedulerSvc.cpp.

413 {
414
415 if ( m_isActive == ACTIVE ) {
416
417 // Set the number of slots available to an error code
418 m_freeSlots.store( 0 );
419
420 // Empty queue
421 action thisAction;
422 while ( m_actionsQueue.try_pop( thisAction ) ) {};
423
424 // This would be the last action
425 m_actionsQueue.push( [this]() -> StatusCode {
426 ON_VERBOSE verbose() << "Deactivating scheduler" << endmsg;
427 m_isActive = INACTIVE;
428 return StatusCode::SUCCESS;
429 } );
430 }
431
432 return StatusCode::SUCCESS;
433 }
AvalancheSchedulerSvc::INACTIVE
Definition: AvalancheSchedulerSvc.h:137
StatusCode::SUCCESS
constexpr static const auto SUCCESS
Definition: StatusCode.h:85
AvalancheSchedulerSvc::ACTIVE
Definition: AvalancheSchedulerSvc.h:137
AvalancheSchedulerSvc::m_isActive
std::atomic< ActivationState > m_isActive
Flag to track if the scheduler is active or not.
Definition: AvalancheSchedulerSvc.h:183
AvalancheSchedulerSvc::m_freeSlots
std::atomic_int m_freeSlots
Atomic to account for asyncronous updates by the scheduler wrt the rest.
Definition: AvalancheSchedulerSvc.h:213
AvalancheSchedulerSvc::action
std::function< StatusCode()> action
Definition: AvalancheSchedulerSvc.h:135
StatusCode
This class is used for returning status codes from appropriate routines.
Definition: StatusCode.h:50
CommonMessagingBase::verbose
MsgStream & verbose() const
shortcut for the method msgStream(MSG::VERBOSE)
Definition: CommonMessaging.h:110
endmsg
MsgStream & endmsg(MsgStream &s)
MsgStream Modifier: endmsg. Calls the output method of the MsgStream.
Definition: MsgStream.h:192
AvalancheSchedulerSvc::m_actionsQueue
tbb::concurrent_bounded_queue< action > m_actionsQueue
Queue where closures are stored and picked for execution.
Definition: AvalancheSchedulerSvc.h:260
ON_VERBOSE
#define ON_VERBOSE
Definition: AvalancheSchedulerSvc.cpp:32

## ◆ 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 775 of file AvalancheSchedulerSvc.cpp.

775 {
776
777 // To have just one big message
778 std::ostringstream outputMS;
779
780 outputMS << "Dumping scheduler state\n"
781 << "=========================================================================================\n"
782 << "++++++++++++++++++++++++++++++++++++ SCHEDULER STATE ++++++++++++++++++++++++++++++++++++\n"
783 << "=========================================================================================\n\n";
784
785 //===========================================================================
786
787 outputMS << "------------------ Last schedule: Task/Event/Slot/Thread/State Mapping "
788 << "------------------\n\n";
789
790 // Figure if TimelineSvc is available (used below to detect threads IDs)
791 auto timelineSvc = serviceLocator()->service<ITimelineSvc>( "TimelineSvc", false );
792 if ( !timelineSvc.isValid() || !timelineSvc->isEnabled() ) {
793 outputMS << "WARNING Enable TimelineSvc in record mode (RecordTimeline = True) to trace the mapping\n";
794 } else {
795
796 // Figure optimal printout layout
797 size_t indt( 0 );
798 for ( auto& slot : m_eventSlots )
799 for ( auto it = slot.algsStates.begin( AState::SCHEDULED ); it != slot.algsStates.end( AState::SCHEDULED ); ++it )
800 if ( index2algname( *it ).length() > indt ) indt = index2algname( *it ).length();
801
802 // Figure the last running schedule across all slots
803 for ( auto& slot : m_eventSlots ) {
804 for ( auto it = slot.algsStates.begin( AState::SCHEDULED ); it != slot.algsStates.end( AState::SCHEDULED );
805 ++it ) {
806
807 const std::string algoName{index2algname( *it )};
808
809 outputMS << " task: " << std::setw( indt ) << algoName << " evt/slot: " << slot.eventContext->evt() << "/"
810 << slot.eventContext->slot();
811
812 // Try to get POSIX threads IDs the currently running tasks are scheduled to
813 if ( timelineSvc.isValid() ) {
814 TimelineEvent te{};
815 te.algorithm = algoName;
816 te.slot = slot.eventContext->slot();
817 te.event = slot.eventContext->evt();
818
819 if ( timelineSvc->getTimelineEvent( te ) )
820 outputMS << " thread.id: 0x" << std::hex << te.thread << std::dec;
821 else
822 outputMS << " thread.id: [unknown]"; // this means a task has just
823 // been signed off as SCHEDULED,
824 // but has not been assigned to a thread yet
825 // (i.e., not running yet)
826 }
827 outputMS << " state: [" << m_algExecStateSvc->algExecState( algoName, *( slot.eventContext ) ) << "]\n";
828 }
829 }
830 }
831
832 //===========================================================================
833
834 outputMS << "\n---------------------------- Task/CF/FSM Mapping "
835 << ( 0 > iSlot ? "[all slots] --" : "[target slot] " ) << "--------------------------\n\n";
836
837 int slotCount = -1;
838 for ( auto& slot : m_eventSlots ) {
839 ++slotCount;
840 if ( slot.complete ) continue;
841
842 outputMS << "[ slot: "
843 << ( slot.eventContext->valid() ? std::to_string( slot.eventContext->slot() ) : "[ctx invalid]" )
844 << " event: "
845 << ( slot.eventContext->valid() ? std::to_string( slot.eventContext->evt() ) : "[ctx invalid]" )
846 << " ]:\n\n";
847
848 if ( 0 > iSlot || iSlot == slotCount ) {
849
850 // Snapshot of the Control Flow and FSM states
851 outputMS << m_precSvc->printState( slot ) << "\n";
852
853 // Mention sub slots (this is expensive if the number of sub-slots is high)
854 if ( m_verboseSubSlots && !slot.allSubSlots.empty() ) {
855 outputMS << "\nNumber of sub-slots: " << slot.allSubSlots.size() << "\n\n";
856 auto slotID = slot.eventContext->valid() ? std::to_string( slot.eventContext->slot() ) : "[ctx invalid]";
857 for ( auto& ss : slot.allSubSlots ) {
858 outputMS << "[ slot: " << slotID << ", sub-slot: "
859 << ( ss.eventContext->valid() ? std::to_string( ss.eventContext->subSlot() ) : "[ctx invalid]" )
860 << ", entry: " << ss.entryPoint << ", event: "
861 << ( ss.eventContext->valid() ? std::to_string( ss.eventContext->evt() ) : "[ctx invalid]" )
862 << " ]:\n\n";
863 outputMS << m_precSvc->printState( ss ) << "\n";
864 }
865 }
866 }
867 }
868
869 //===========================================================================
870
871 if ( 0 <= iSlot ) {
872 outputMS << "\n------------------------------ Algorithm Execution States -----------------------------\n\n";
873 m_algExecStateSvc->dump( outputMS, *( m_eventSlots[iSlot].eventContext ) );
874 }
875
876 outputMS << "\n=========================================================================================\n"
877 << "++++++++++++++++++++++++++++++++++++++ END OF DUMP ++++++++++++++++++++++++++++++++++++++\n"
878 << "=========================================================================================\n\n";
879
880 info() << outputMS.str() << endmsg;
881 }
Service::serviceLocator
SmartIF< ISvcLocator > & serviceLocator() const override
Retrieve pointer to service locator.
Definition: Service.cpp:277
AvalancheSchedulerSvc::m_precSvc
SmartIF< IPrecedenceSvc > m_precSvc
A shortcut to the Precedence Service.
Definition: AvalancheSchedulerSvc.h:201
std::to_string
T to_string(T... args)
TimelineEvent::algorithm
std::string algorithm
Definition: ITimelineSvc.h:21
CommonMessagingBase::info
MsgStream & info() const
shortcut for the method msgStream(MSG::INFO)
Definition: CommonMessaging.h:104
TimelineEvent
Definition: ITimelineSvc.h:13
std::setw
T setw(T... args)
std::string
STL class.
IAlgExecStateSvc::algExecState
virtual const AlgExecState & algExecState(const Gaudi::StringKey &algName, const EventContext &ctx) const =0
ISvcLocator::service
StatusCode service(const Gaudi::Utils::TypeNameString &name, T *&svc, bool createIf=true)
Templated method to access a service by name.
Definition: ISvcLocator.h:76
AvalancheSchedulerSvc::m_verboseSubSlots
Gaudi::Property< bool > m_verboseSubSlots
Definition: AvalancheSchedulerSvc.h:172
IAlgExecStateSvc::dump
virtual void dump(std::ostringstream &ost, const EventContext &ctx) const =0
std::ostringstream
STL class.
std::ostringstream::str
T str(T... args)
ITimelineSvc
Definition: ITimelineSvc.h:27
std::string::length
T length(T... args)
AvalancheSchedulerSvc::index2algname
const std::string & index2algname(unsigned int index)
Convert an integer to a name.
Definition: AvalancheSchedulerSvc.cpp:439
std::hex
T hex(T... args)
AvalancheSchedulerSvc::m_eventSlots
std::vector< EventSlot > m_eventSlots
Vector of events slots.
Definition: AvalancheSchedulerSvc.h:210
AvalancheSchedulerSvc::m_algExecStateSvc
SmartIF< IAlgExecStateSvc > m_algExecStateSvc
Algorithm execution state manager.
Definition: AvalancheSchedulerSvc.h:219
IPrecedenceSvc::printState
virtual const std::string printState(EventSlot &) const =0
endmsg
MsgStream & endmsg(MsgStream &s)
MsgStream Modifier: endmsg. Calls the output method of the MsgStream.
Definition: MsgStream.h:192

## ◆ 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 754 of file AvalancheSchedulerSvc.cpp.

754 {
755 const uint slotIdx = eventContext->slot();
756
757 error() << "Event " << eventContext->evt() << " on slot " << slotIdx << " failed" << endmsg;
758
759 dumpSchedulerState( msgLevel( MSG::VERBOSE ) ? -1 : slotIdx );
760
761 // dump temporal and topological precedence analysis (if enabled in the PrecedenceSvc)
762 m_precSvc->dumpPrecedenceRules( m_eventSlots[slotIdx] );
763
764 // Push into the finished events queue the failed context
765 m_eventSlots[slotIdx].complete = true;
766 m_finishedEvents.push( m_eventSlots[slotIdx].eventContext.release() );
767 }
EventContext::slot
ContextID_t slot() const
Definition: EventContext.h:41
AvalancheSchedulerSvc::m_precSvc
SmartIF< IPrecedenceSvc > m_precSvc
A shortcut to the Precedence Service.
Definition: AvalancheSchedulerSvc.h:201
EventContext::evt
ContextEvt_t evt() const
Definition: EventContext.h:40
AvalancheSchedulerSvc::dumpSchedulerState
void dumpSchedulerState(int iSlot)
Dump the state of the scheduler.
Definition: AvalancheSchedulerSvc.cpp:775
MSG::VERBOSE
Definition: IMessageSvc.h:15
CommonMessaging< implements< IService, IProperty, IStateful > >::msgLevel
MSG::Level msgLevel() const
get the cached level (originally extracted from the embedded MsgStream)
Definition: CommonMessaging.h:138
CommonMessagingBase::error
MsgStream & error() const
shortcut for the method msgStream(MSG::ERROR)
Definition: CommonMessaging.h:98
IPrecedenceSvc::dumpPrecedenceRules
virtual void dumpPrecedenceRules(EventSlot &)=0
Dump precedence rules.
AvalancheSchedulerSvc::m_finishedEvents
tbb::concurrent_bounded_queue< EventContext * > m_finishedEvents
Queue of finished events.
Definition: AvalancheSchedulerSvc.h:216
AvalancheSchedulerSvc::m_eventSlots
std::vector< EventSlot > m_eventSlots
Vector of events slots.
Definition: AvalancheSchedulerSvc.h:210
endmsg
MsgStream & endmsg(MsgStream &s)
MsgStream Modifier: endmsg. Calls the output method of the MsgStream.
Definition: MsgStream.h:192

## ◆ finalize()

StatusCode AvalancheSchedulerSvc::finalize | ( |  | ) |
| override

Finalise.

Here the scheduler is deactivated and the thread joined.

Definition at line 337 of file AvalancheSchedulerSvc.cpp.

337 {
338
339 StatusCode sc( Service::finalize() );
340 if ( sc.isFailure() ) warning() << "Base class could not be finalized" << endmsg;
341
342 sc = deactivate();
343 if ( sc.isFailure() ) warning() << "Scheduler could not be deactivated" << endmsg;
344
345 info() << "Joining Scheduler thread" << endmsg;
346 m_thread.join();
347
348 // Final error check after thread pool termination
349 if ( m_isActive == FAILURE ) {
350 error() << "problems in scheduler thread" << endmsg;
351 return StatusCode::FAILURE;
352 }
353
354 return sc;
355 }
Service::finalize
StatusCode finalize() override
Definition: Service.cpp:164
CommonMessagingBase::warning
MsgStream & warning() const
shortcut for the method msgStream(MSG::WARNING)
Definition: CommonMessaging.h:101
CommonMessagingBase::info
MsgStream & info() const
shortcut for the method msgStream(MSG::INFO)
Definition: CommonMessaging.h:104
AvalancheSchedulerSvc::m_isActive
std::atomic< ActivationState > m_isActive
Flag to track if the scheduler is active or not.
Definition: AvalancheSchedulerSvc.h:183
CommonMessagingBase::error
MsgStream & error() const
shortcut for the method msgStream(MSG::ERROR)
Definition: CommonMessaging.h:98
std::thread::join
T join(T... args)
StatusCode
This class is used for returning status codes from appropriate routines.
Definition: StatusCode.h:50
StatusCode::FAILURE
constexpr static const auto FAILURE
Definition: StatusCode.h:86
AvalancheSchedulerSvc::deactivate
StatusCode deactivate()
Deactivate scheduler.
Definition: AvalancheSchedulerSvc.cpp:413
AvalancheSchedulerSvc::FAILURE
Definition: AvalancheSchedulerSvc.h:137
endmsg
MsgStream & endmsg(MsgStream &s)
MsgStream Modifier: endmsg. Calls the output method of the MsgStream.
Definition: MsgStream.h:192
AvalancheSchedulerSvc::m_thread
std::thread m_thread
The thread in which the activate function runs.
Definition: AvalancheSchedulerSvc.h:186

## ◆ freeSlots()

unsigned int AvalancheSchedulerSvc::freeSlots | ( |  | ) |
| override

Get free slots number.

Definition at line 526 of file AvalancheSchedulerSvc.cpp.

526 { 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:213
std::max
T max(T... args)

## ◆ index2algname()

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

Convert an integer to a name.

Definition at line 439 of file AvalancheSchedulerSvc.cpp.

439 { 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:198

## ◆ 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 63 of file AvalancheSchedulerSvc.cpp.

63 {
64
65 // Initialise mother class (read properties, ...)
66 StatusCode sc( Service::initialize() );
67 if ( sc.isFailure() ) warning() << "Base class could not be initialized" << endmsg;
68
69 // Get hold of the TBBSvc. This should initialize the thread pool
70 m_threadPoolSvc = serviceLocator()->service( "ThreadPoolSvc" );
71 if ( !m_threadPoolSvc.isValid() ) {
72 fatal() << "Error retrieving ThreadPoolSvc" << endmsg;
73 return StatusCode::FAILURE;
74 }
75
76 // Activate the scheduler in another thread.
77 info() << "Activating scheduler in a separate thread" << endmsg;
78 m_thread = std::thread( [this]() { this->activate(); } );
79
80 while ( m_isActive != ACTIVE ) {
81 if ( m_isActive == FAILURE ) {
82 fatal() << "Terminating initialization" << endmsg;
83 return StatusCode::FAILURE;
84 } else {
85 ON_DEBUG debug() << "Waiting for AvalancheSchedulerSvc to activate" << endmsg;
86 sleep( 1 );
87 }
88 }
89
90 if ( m_enableCondSvc ) {
91 // Get hold of the CondSvc
92 m_condSvc = serviceLocator()->service( "CondSvc" );
93 if ( !m_condSvc.isValid() ) {
94 warning() << "No CondSvc found, or not enabled. "
95 << "Will not manage CondAlgorithms" << endmsg;
96 m_enableCondSvc = false;
97 }
98 }
99
100 // Get the algo resource pool
101 m_algResourcePool = serviceLocator()->service( "AlgResourcePool" );
102 if ( !m_algResourcePool.isValid() ) {
103 fatal() << "Error retrieving AlgoResourcePool" << endmsg;
104 return StatusCode::FAILURE;
105 }
106
107 m_algExecStateSvc = serviceLocator()->service( "AlgExecStateSvc" );
108 if ( !m_algExecStateSvc.isValid() ) {
109 fatal() << "Error retrieving AlgExecStateSvc" << endmsg;
110 return StatusCode::FAILURE;
111 }
112
113 // Get Whiteboard
114 m_whiteboard = serviceLocator()->service( m_whiteboardSvcName );
115 if ( !m_whiteboard.isValid() ) {
116 fatal() << "Error retrieving EventDataSvc interface IHiveWhiteBoard." << endmsg;
117 return StatusCode::FAILURE;
118 }
119
120 // Get dedicated scheduler for I/O-bound algorithms
121 if ( m_useIOBoundAlgScheduler ) {
122 m_IOBoundAlgScheduler = serviceLocator()->service( m_IOBoundAlgSchedulerSvcName );
123 if ( !m_IOBoundAlgScheduler.isValid() )
124 fatal() << "Error retrieving IOBoundSchedulerAlgSvc interface IAccelerator." << endmsg;
125 }
126
127 // Set the MaxEventsInFlight parameters from the number of WB stores
128 m_maxEventsInFlight = m_whiteboard->getNumberOfStores();
129
130 // Set the number of free slots
131 m_freeSlots = m_maxEventsInFlight;
132
133 // Get the list of algorithms
134 const std::list<IAlgorithm*>& algos = m_algResourcePool->getFlatAlgList();
135 const unsigned int algsNumber = algos.size();
136 if ( algsNumber != 0 ) {
137 info() << "Found " << algsNumber << " algorithms" << endmsg;
138 } else {
139 error() << "No algorithms found" << endmsg;
140 return StatusCode::FAILURE;
141 }
142
143 /* Dependencies
144 1) Look for handles in algo, if none
145 2) Assume none are required
146 */
147
148 DataObjIDColl globalInp, globalOutp;
149
150 // figure out all outputs
151 for ( IAlgorithm* ialgoPtr : algos ) {
152 Gaudi::Algorithm* algoPtr = dynamic_cast<Gaudi::Algorithm*>( ialgoPtr );
153 if ( !algoPtr ) {
154 fatal() << "Could not convert IAlgorithm into Gaudi::Algorithm: this will result in a crash." << endmsg;
155 return StatusCode::FAILURE;
156 }
157 for ( auto id : algoPtr->outputDataObjs() ) {
158 auto r = globalOutp.insert( id );
159 if ( !r.second ) {
160 warning() << "multiple algorithms declare " << id
161 << " as output! could be a single instance in multiple paths "
162 "though, or control flow may guarantee only one runs...!"
163 << endmsg;
164 }
165 }
166 }
167
168 std::ostringstream ostdd;
169 ostdd << "Data Dependencies for Algorithms:";
170
171 std::map<std::string, DataObjIDColl> algosDependenciesMap;
172 for ( IAlgorithm* ialgoPtr : algos ) {
173 Gaudi::Algorithm* algoPtr = dynamic_cast<Gaudi::Algorithm*>( ialgoPtr );
174 if ( nullptr == algoPtr ) {
175 fatal() << "Could not convert IAlgorithm into Gaudi::Algorithm for " << ialgoPtr->name()
176 << ": this will result in a crash." << endmsg;
177 return StatusCode::FAILURE;
178 }
179
180 ostdd << "\n " << algoPtr->name();
181
182 DataObjIDColl algoDependencies;
183 if ( !algoPtr->inputDataObjs().empty() || !algoPtr->outputDataObjs().empty() ) {
184 for ( const DataObjID* idp : sortedDataObjIDColl( algoPtr->inputDataObjs() ) ) {
185 DataObjID id = *idp;
186 ostdd << "\n o INPUT " << id;
187 if ( id.key().find( ":" ) != std::string::npos ) {
188 ostdd << " contains alternatives which require resolution...\n";
189 auto tokens = boost::tokenizer<boost::char_separator<char>>{id.key(), boost::char_separator<char>{":"}};
190 auto itok = std::find_if( tokens.begin(), tokens.end(), [&]( const std::string& t ) {
191 return globalOutp.find( DataObjID{t} ) != globalOutp.end();
192 } );
193 if ( itok != tokens.end() ) {
194 ostdd << "found matching output for " << *itok << " -- updating scheduler info\n";
195 id.updateKey( *itok );
196 } else {
197 error() << "failed to find alternate in global output list"
198 << " for id: " << id << " in Alg " << algoPtr->name() << endmsg;
199 m_showDataDeps = true;
200 }
201 }
202 algoDependencies.insert( id );
203 globalInp.insert( id );
204 }
205 for ( const DataObjID* id : sortedDataObjIDColl( algoPtr->outputDataObjs() ) ) {
206 ostdd << "\n o OUTPUT " << *id;
207 if ( id->key().find( ":" ) != std::string::npos ) {
208 error() << " in Alg " << algoPtr->name() << " alternatives are NOT allowed for outputs! id: " << *id
209 << endmsg;
210 m_showDataDeps = true;
211 }
212 }
213 } else {
214 ostdd << "\n none";
215 }
216 algosDependenciesMap[algoPtr->name()] = algoDependencies;
217 }
218
219 if ( m_showDataDeps ) { info() << ostdd.str() << endmsg; }
220
221 // Check if we have unmet global input dependencies, and, optionally, heal them
222 // WARNING: this step must be done BEFORE the Precedence Service is initialized
223 if ( m_checkDeps ) {
224 DataObjIDColl unmetDep;
225 for ( auto o : globalInp )
226 if ( globalOutp.find( o ) == globalOutp.end() ) unmetDep.insert( o );
227
228 if ( unmetDep.size() > 0 ) {
229
230 std::ostringstream ost;
231 for ( const DataObjID* o : sortedDataObjIDColl( unmetDep ) ) {
232 ost << "\n o " << *o << " required by Algorithm: ";
233
234 for ( const auto& p : algosDependenciesMap )
235 if ( p.second.find( *o ) != p.second.end() ) ost << "\n * " << p.first;
236 }
237
238 if ( !m_useDataLoader.empty() ) {
239
240 // Find the DataLoader Alg
241 IAlgorithm* dataLoaderAlg( nullptr );
242 for ( IAlgorithm* algo : algos )
243 if ( algo->name() == m_useDataLoader ) {
244 dataLoaderAlg = algo;
245 break;
246 }
247
248 if ( dataLoaderAlg == nullptr ) {
249 fatal() << "No DataLoader Algorithm \"" << m_useDataLoader.value()
250 << "\" found, and unmet INPUT dependencies "
251 << "detected:\n"
252 << ost.str() << endmsg;
253 return StatusCode::FAILURE;
254 }
255
256 info() << "Will attribute the following unmet INPUT dependencies to \"" << dataLoaderAlg->type() << "/"
257 << dataLoaderAlg->name() << "\" Algorithm" << ost.str() << endmsg;
258
259 // Set the property Load of DataLoader Alg
260 Gaudi::Algorithm* dataAlg = dynamic_cast<Gaudi::Algorithm*>( dataLoaderAlg );
261 if ( !dataAlg ) {
262 fatal() << "Unable to dcast DataLoader \"" << m_useDataLoader.value() << "\" IAlg to Gaudi::Algorithm"
263 << endmsg;
264 return StatusCode::FAILURE;
265 }
266
267 for ( auto& id : unmetDep ) {
268 ON_DEBUG debug() << "adding OUTPUT dep \"" << id << "\" to " << dataLoaderAlg->type() << "/"
269 << dataLoaderAlg->name() << endmsg;
270 dataAlg->addDependency( id, Gaudi::DataHandle::Writer );
271 }
272
273 } else {
274 fatal() << "Auto DataLoading not requested, "
275 << "and the following unmet INPUT dependencies were found:" << ost.str() << endmsg;
276 return StatusCode::FAILURE;
277 }
278
279 } else {
280 info() << "No unmet INPUT data dependencies were found" << endmsg;
281 }
282 }
283
284 // Get the precedence service
285 m_precSvc = serviceLocator()->service( "PrecedenceSvc" );
286 if ( !m_precSvc.isValid() ) {
287 fatal() << "Error retrieving PrecedenceSvc" << endmsg;
288 return StatusCode::FAILURE;
289 }
290 const PrecedenceSvc* precSvc = dynamic_cast<const PrecedenceSvc*>( m_precSvc.get() );
291 if ( !precSvc ) {
292 fatal() << "Unable to dcast PrecedenceSvc" << endmsg;
293 return StatusCode::FAILURE;
294 }
295
296 // Fill the containers to convert algo names to index
297 m_algname_vect.resize( algsNumber );
298 for ( IAlgorithm* algo : algos ) {
299 const std::string& name = algo->name();
300 auto index = precSvc->getRules()->getAlgorithmNode( name )->getAlgoIndex();
301 m_algname_index_map[name] = index;
302 m_algname_vect.at( index ) = name;
303 }
304
305 // Shortcut for the message service
306 SmartIF<IMessageSvc> messageSvc( serviceLocator() );
307 if ( !messageSvc.isValid() ) error() << "Error retrieving MessageSvc interface IMessageSvc." << endmsg;
308
309 m_eventSlots.reserve( m_maxEventsInFlight );
310 for ( size_t i = 0; i < m_maxEventsInFlight; ++i ) {
311 m_eventSlots.emplace_back( algsNumber, precSvc->getRules()->getControlFlowNodeCounter(), messageSvc );
312 m_eventSlots.back().complete = true;
313 }
314 m_actionsCounts.assign( m_maxEventsInFlight, 0 );
315
316 if ( m_threadPoolSize > 1 ) { m_maxAlgosInFlight = (size_t)m_threadPoolSize; }
317
318 // Clearly inform about the level of concurrency
319 info() << "Concurrency level information:" << endmsg;
320 info() << " o Number of events in flight: " << m_maxEventsInFlight << endmsg;
321 info() << " o TBB thread pool size: " << m_threadPoolSize << endmsg;
322
323 if ( m_showControlFlow ) m_precSvc->dumpControlFlow();
324
325 if ( m_showDataFlow ) m_precSvc->dumpDataFlow();
326
327 // Simulate execution flow
328 if ( m_simulateExecution ) m_precSvc->simulate( m_eventSlots[0] );
329
330 return sc;
331 }
AvalancheSchedulerSvc::m_showDataFlow
Gaudi::Property< bool > m_showDataFlow
Definition: AvalancheSchedulerSvc.h:166
ON_DEBUG
#define ON_DEBUG
Definition: AvalancheSchedulerSvc.cpp:31
Service::initialize
StatusCode initialize() override
Definition: Service.cpp:60
SmartIF< IMessageSvc >
Service::serviceLocator
SmartIF< ISvcLocator > & serviceLocator() const override
Retrieve pointer to service locator.
Definition: Service.cpp:277
std::unordered_set::empty
T empty(T... args)
AvalancheSchedulerSvc::m_whiteboardSvcName
Gaudi::Property< std::string > m_whiteboardSvcName
Definition: AvalancheSchedulerSvc.h:142
concurrency::PrecedenceRulesGraph::getAlgorithmNode
AlgorithmNode * getAlgorithmNode(const std::string &algoName) const
Get the AlgorithmNode from by algorithm name using graph index.
Definition: PrecedenceRulesGraph.h:649
AvalancheSchedulerSvc::m_precSvc
SmartIF< IPrecedenceSvc > m_precSvc
A shortcut to the Precedence Service.
Definition: AvalancheSchedulerSvc.h:201
AvalancheSchedulerSvc::m_actionsCounts
std::vector< unsigned int > m_actionsCounts
Bookkeeping of the number of actions in flight per slot.
Definition: AvalancheSchedulerSvc.h:263
DataObjID
Definition: DataObjID.h:37
AvalancheSchedulerSvc::m_showDataDeps
Gaudi::Property< bool > m_showDataDeps
Definition: AvalancheSchedulerSvc.h:163
CommonMessagingBase::warning
MsgStream & warning() const
shortcut for the method msgStream(MSG::WARNING)
Definition: CommonMessaging.h:101
PrecedenceSvc
A service to resolve the task execution precedence.
Definition: PrecedenceSvc.h:21
SmartIF::isValid
bool isValid() const
Allow for check if smart pointer is valid.
Definition: SmartIF.h:62
AvalancheSchedulerSvc::ACTIVE
Definition: AvalancheSchedulerSvc.h:137
AvalancheSchedulerSvc::activate
void activate()
Activate scheduler.
Definition: AvalancheSchedulerSvc.cpp:368
std::unordered_set::end
T end(T... args)
AvalancheSchedulerSvc::m_useDataLoader
Gaudi::Property< std::string > m_useDataLoader
Definition: AvalancheSchedulerSvc.h:158
IPrecedenceSvc::simulate
virtual StatusCode simulate(EventSlot &) const =0
Simulate execution flow.
CommonMessagingBase::info
MsgStream & info() const
shortcut for the method msgStream(MSG::INFO)
Definition: CommonMessaging.h:104
IAlgResourcePool::getFlatAlgList
virtual std::list< IAlgorithm * > getFlatAlgList()=0
Get the flat list of algorithms.
SmartIF::get
TYPE * get() const
Get interface pointer.
Definition: SmartIF.h:76
std::map
STL class.
AvalancheSchedulerSvc::m_isActive
std::atomic< ActivationState > m_isActive
Flag to track if the scheduler is active or not.
Definition: AvalancheSchedulerSvc.h:183
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:192
std::vector::resize
T resize(T... args)
AvalancheSchedulerSvc::m_checkDeps
Gaudi::Property< bool > m_checkDeps
Definition: AvalancheSchedulerSvc.h:156
std::string
STL class.
AvalancheSchedulerSvc::m_useIOBoundAlgScheduler
Gaudi::Property< bool > m_useIOBoundAlgScheduler
Definition: AvalancheSchedulerSvc.h:153
concurrency::AlgorithmNode::getAlgoIndex
const unsigned int & getAlgoIndex() const
Get algorithm index.
Definition: PrecedenceRulesGraph.h:513
AvalancheSchedulerSvc::m_freeSlots
std::atomic_int m_freeSlots
Atomic to account for asyncronous updates by the scheduler wrt the rest.
Definition: AvalancheSchedulerSvc.h:213
std::vector::at
T at(T... args)
ISvcLocator::service
StatusCode service(const Gaudi::Utils::TypeNameString &name, T *&svc, bool createIf=true)
Templated method to access a service by name.
Definition: ISvcLocator.h:76
Service::name
const std::string & name() const override
Retrieve name of the service.
Definition: Service.cpp:274
CommonMessagingBase::error
MsgStream & error() const
shortcut for the method msgStream(MSG::ERROR)
Definition: CommonMessaging.h:98
AvalancheSchedulerSvc::m_algResourcePool
SmartIF< IAlgResourcePool > m_algResourcePool
Cache for the algorithm resource pool.
Definition: AvalancheSchedulerSvc.h:255
AvalancheSchedulerSvc::m_showControlFlow
Gaudi::Property< bool > m_showControlFlow
Definition: AvalancheSchedulerSvc.h:169
StatusCode
This class is used for returning status codes from appropriate routines.
Definition: StatusCode.h:50
AvalancheSchedulerSvc::m_whiteboard
SmartIF< IHiveWhiteBoard > m_whiteboard
A shortcut to the whiteboard.
Definition: AvalancheSchedulerSvc.h:204
std::ostringstream
STL class.
AvalancheSchedulerSvc::m_algname_vect
std::vector< std::string > m_algname_vect
Vector to bookkeep the information necessary to the index2name conversion.
Definition: AvalancheSchedulerSvc.h:198
AvalancheSchedulerSvc::m_IOBoundAlgSchedulerSvcName
Gaudi::Property< std::string > m_IOBoundAlgSchedulerSvcName
Definition: AvalancheSchedulerSvc.h:143
std::ostringstream::str
T str(T... args)
CommonMessagingBase::debug
MsgStream & debug() const
shortcut for the method msgStream(MSG::DEBUG)
Definition: CommonMessaging.h:107
concurrency::PrecedenceRulesGraph::getControlFlowNodeCounter
unsigned int getControlFlowNodeCounter() const
Get total number of control flow graph nodes.
Definition: PrecedenceRulesGraph.h:657
std::unordered_set< DataObjID, DataObjID_Hasher >
IHiveWhiteBoard::getNumberOfStores
virtual size_t getNumberOfStores() const =0
Get the number of 'slots'.
AvalancheSchedulerSvc::m_threadPoolSize
Gaudi::Property< int > m_threadPoolSize
Definition: AvalancheSchedulerSvc.h:139
IPrecedenceSvc::dumpDataFlow
virtual void dumpDataFlow() const =0
AvalancheSchedulerSvc::m_threadPoolSvc
SmartIF< IThreadPoolSvc > m_threadPoolSvc
Definition: AvalancheSchedulerSvc.h:268
std::list< IAlgorithm * >
AvalancheSchedulerSvc::m_IOBoundAlgScheduler
SmartIF< IAccelerator > m_IOBoundAlgScheduler
A shortcut to IO-bound algorithm scheduler.
Definition: AvalancheSchedulerSvc.h:207
IAlgorithm
The IAlgorithm is the interface implemented by the Algorithm base class.
Definition: IAlgorithm.h:28
std::unordered_set::insert
T insert(T... args)
DataHandleHolderBase::addDependency
void addDependency(const DataObjID &id, const Gaudi::DataHandle::Mode &mode) override
Definition: DataHandleHolderBase.h:58
DataHandleHolderBase::outputDataObjs
const DataObjIDColl & outputDataObjs() const override
Definition: DataHandleHolderBase.h:56
std::find_if
T find_if(T... args)
std::list::size
T size(T... args)
std::vector::assign
T assign(T... args)
AvalancheSchedulerSvc::m_simulateExecution
Gaudi::Property< bool > m_simulateExecution
Definition: AvalancheSchedulerSvc.h:146
AvalancheSchedulerSvc::m_enableCondSvc
Gaudi::Property< bool > m_enableCondSvc
Definition: AvalancheSchedulerSvc.h:161
Gaudi::Algorithm
Base class from which all concrete algorithm classes should be derived.
Definition: Algorithm.h:79
AvalancheSchedulerSvc::m_maxAlgosInFlight
size_t m_maxAlgosInFlight
Definition: AvalancheSchedulerSvc.h:270
std::vector::back
T back(T... args)
StatusCode::FAILURE
constexpr static const auto FAILURE
Definition: StatusCode.h:86
Gaudi::DataHandle::Writer
Definition: DataHandle.h:29
AvalancheSchedulerSvc::m_condSvc
SmartIF< ICondSvc > m_condSvc
A shortcut to service for Conditions handling.
Definition: AvalancheSchedulerSvc.h:222
AvalancheSchedulerSvc::m_eventSlots
std::vector< EventSlot > m_eventSlots
Vector of events slots.
Definition: AvalancheSchedulerSvc.h:210
EventSlot::complete
bool complete
Flags completion of the event.
Definition: EventSlot.h:79
AvalancheSchedulerSvc::FAILURE
Definition: AvalancheSchedulerSvc.h:137
DataHandleHolderBase::inputDataObjs
const DataObjIDColl & inputDataObjs() const override
Definition: DataHandleHolderBase.h:55
IPrecedenceSvc::dumpControlFlow
virtual void dumpControlFlow() const =0
Dump precedence rules.
std::thread
STL class.
AvalancheSchedulerSvc::m_algExecStateSvc
SmartIF< IAlgExecStateSvc > m_algExecStateSvc
Algorithm execution state manager.
Definition: AvalancheSchedulerSvc.h:219
CommonMessagingBase::fatal
MsgStream & fatal() const
shortcut for the method msgStream(MSG::FATAL)
Definition: CommonMessaging.h:92
AvalancheSchedulerSvc::m_maxEventsInFlight
size_t m_maxEventsInFlight
Definition: AvalancheSchedulerSvc.h:269
endmsg
MsgStream & endmsg(MsgStream &s)
MsgStream Modifier: endmsg. Calls the output method of the MsgStream.
Definition: MsgStream.h:192
PrecedenceSvc::getRules
const concurrency::PrecedenceRulesGraph * getRules() const
Precedence rules accessor.
Definition: PrecedenceSvc.h:63
Gaudi::Algorithm::name
const std::string & name() const override
The identifying name of the algorithm object.
Definition: Algorithm.cpp:645
std::vector::reserve
T reserve(T... args)
std::vector::emplace_back
T emplace_back(T... args)
AvalancheSchedulerSvc::m_thread
std::thread m_thread
The thread in which the activate function runs.
Definition: AvalancheSchedulerSvc.h:186

## ◆ 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 735 of file AvalancheSchedulerSvc.cpp.

735 {
736
737 if ( m_actionsCounts[slot.eventContext->slot()] == 0 &&
738 !slot.algsStates.containsAny( {AState::DATAREADY, AState::SCHEDULED} ) &&
739 !subSlotAlgsInStates( slot, {AState::DATAREADY, AState::SCHEDULED} ) ) {
740
741 error() << "*** Stall detected in slot " << slot.eventContext->slot() << "! ***" << endmsg;
742
743 return true;
744 }
745 return false;
746 }
EventContext::slot
ContextID_t slot() const
Definition: EventContext.h:41
AvalancheSchedulerSvc::m_actionsCounts
std::vector< unsigned int > m_actionsCounts
Bookkeeping of the number of actions in flight per slot.
Definition: AvalancheSchedulerSvc.h:263
AlgsExecutionStates::containsAny
bool containsAny(std::initializer_list< State > l) const
check if the collection contains at least one state of any listed types
Definition: AlgsExecutionStates.h:50
CommonMessagingBase::error
MsgStream & error() const
shortcut for the method msgStream(MSG::ERROR)
Definition: CommonMessaging.h:98
endmsg
MsgStream & endmsg(MsgStream &s)
MsgStream Modifier: endmsg. Calls the output method of the MsgStream.
Definition: MsgStream.h:192
EventSlot::eventContext
std::unique_ptr< EventContext > eventContext
Cache for the eventContext.
Definition: EventSlot.h:73
EventSlot::algsStates
AlgsExecutionStates algsStates
Vector of algorithms states.
Definition: EventSlot.h:75

## ◆ 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 532 of file AvalancheSchedulerSvc.cpp.

532 {
533
534 // ON_DEBUG debug() << "popFinishedEvent: queue size: " << m_finishedEvents.size() << endmsg;
535 if ( m_freeSlots.load() == (int)m_maxEventsInFlight || m_isActive == INACTIVE ) {
536 // ON_DEBUG debug() << "freeslots: " << m_freeSlots << "/" << m_maxEventsInFlight
537 // << " active: " << m_isActive << endmsg;
538 return StatusCode::FAILURE;
539 } else {
540 // ON_DEBUG debug() << "freeslots: " << m_freeSlots << "/" << m_maxEventsInFlight
541 // << " active: " << m_isActive << endmsg;
542 m_finishedEvents.pop( eventContext );
543 ++m_freeSlots;
544 ON_DEBUG debug() << "Popped slot " << eventContext->slot() << " (event " << eventContext->evt() << ")" << endmsg;
545 return StatusCode::SUCCESS;
546 }
547 }
ON_DEBUG
#define ON_DEBUG
Definition: AvalancheSchedulerSvc.cpp:31
EventContext::slot
ContextID_t slot() const
Definition: EventContext.h:41
AvalancheSchedulerSvc::INACTIVE
Definition: AvalancheSchedulerSvc.h:137
EventContext::evt
ContextEvt_t evt() const
Definition: EventContext.h:40
StatusCode::SUCCESS
constexpr static const auto SUCCESS
Definition: StatusCode.h:85
AvalancheSchedulerSvc::m_isActive
std::atomic< ActivationState > m_isActive
Flag to track if the scheduler is active or not.
Definition: AvalancheSchedulerSvc.h:183
AvalancheSchedulerSvc::m_freeSlots
std::atomic_int m_freeSlots
Atomic to account for asyncronous updates by the scheduler wrt the rest.
Definition: AvalancheSchedulerSvc.h:213
AvalancheSchedulerSvc::m_finishedEvents
tbb::concurrent_bounded_queue< EventContext * > m_finishedEvents
Queue of finished events.
Definition: AvalancheSchedulerSvc.h:216
CommonMessagingBase::debug
MsgStream & debug() const
shortcut for the method msgStream(MSG::DEBUG)
Definition: CommonMessaging.h:107
StatusCode::FAILURE
constexpr static const auto FAILURE
Definition: StatusCode.h:86
AvalancheSchedulerSvc::m_maxEventsInFlight
size_t m_maxEventsInFlight
Definition: AvalancheSchedulerSvc.h:269
endmsg
MsgStream & endmsg(MsgStream &s)
MsgStream Modifier: endmsg. Calls the output method of the MsgStream.
Definition: MsgStream.h:192

## ◆ 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 1056 of file AvalancheSchedulerSvc.cpp.

1057 {
1058 Gaudi::Hive::setCurrentContext( eventContext );
1059 StatusCode sc = m_algResourcePool->releaseAlgorithm( algo->name(), algo );
1060
1061 if ( sc.isFailure() ) {
1062 error() << "[Asynchronous] [Event " << eventContext->evt() << ", Slot " << eventContext->slot() << "] "
1063 << "Instance of algorithm " << algo->name() << " could not be properly put back." << endmsg;
1064 return StatusCode::FAILURE;
1065 }
1066
1067 --m_IOBoundAlgosInFlight;
1068
1069 EventSlot& thisSlot = m_eventSlots[si];
1070
1071 ON_DEBUG debug() << "[Asynchronous] Trying to handle execution result of " << algo->name() << " on slot " << si
1072 << endmsg;
1073
1074 const AlgExecState& algstate = m_algExecStateSvc->algExecState( algo, *eventContext );
1075 AState state = algstate.execStatus().isSuccess()
1076 ? ( algstate.filterPassed() ? AState::EVTACCEPTED : AState::EVTREJECTED )
1077 : AState::ERROR;
1078
1079 // Update states in the appropriate slot
1080 int subSlotIndex = -1;
1081 if ( eventContext->usesSubSlot() ) {
1082 // Sub-slot
1083 subSlotIndex = eventContext->subSlot();
1084 sc = thisSlot.allSubSlots[subSlotIndex].algsStates.set( iAlgo, state );
1085 } else {
1086 // Event level (standard behaviour)
1087 sc = thisSlot.algsStates.set( iAlgo, state );
1088 }
1089
1090 ON_VERBOSE if ( sc.isSuccess() ) verbose()
1091 << "[Asynchronous] Promoting " << algo->name() << " on slot " << si << " to " << state << endmsg;
1092
1093 ON_DEBUG debug() << "[Asynchronous] Algorithm " << algo->name() << " executed in slot " << si
1094 << ". Algorithms scheduled are " << m_IOBoundAlgosInFlight << endmsg;
1095
1096 // Schedule an update of the status of the algorithms
1097 ++m_actionsCounts[si];
1098 m_actionsQueue.push( [this, si, iAlgo, subSlotIndex]() {
1099 --this->m_actionsCounts[si]; // no bound check needed as decrements/increments are balanced in the current setup
1100 return this->updateStates( -1, iAlgo, subSlotIndex, si );
1101 } );
1102
1103 return sc;
1104 }
ON_DEBUG
#define ON_DEBUG
Definition: AvalancheSchedulerSvc.cpp:31
AvalancheSchedulerSvc::AState
AlgsExecutionStates::State AState
Definition: AvalancheSchedulerSvc.h:134
EventSlot
Class representing an event slot.
Definition: EventSlot.h:14
EventContext::slot
ContextID_t slot() const
Definition: EventContext.h:41
AvalancheSchedulerSvc::m_IOBoundAlgosInFlight
unsigned int m_IOBoundAlgosInFlight
Number of algorithms presently in flight.
Definition: AvalancheSchedulerSvc.h:228
AvalancheSchedulerSvc::m_actionsCounts
std::vector< unsigned int > m_actionsCounts
Bookkeeping of the number of actions in flight per slot.
Definition: AvalancheSchedulerSvc.h:263
EventContext::evt
ContextEvt_t evt() const
Definition: EventContext.h:40
AlgExecState::filterPassed
bool filterPassed() const
Definition: IAlgExecStateSvc.h:31
EventSlot::allSubSlots
std::vector< EventSlot > allSubSlots
Actual sub-slot instances.
Definition: EventSlot.h:90
IAlgExecStateSvc::algExecState
virtual const AlgExecState & algExecState(const Gaudi::StringKey &algName, const EventContext &ctx) const =0
CommonMessagingBase::error
MsgStream & error() const
shortcut for the method msgStream(MSG::ERROR)
Definition: CommonMessaging.h:98
AvalancheSchedulerSvc::m_algResourcePool
SmartIF< IAlgResourcePool > m_algResourcePool
Cache for the algorithm resource pool.
Definition: AvalancheSchedulerSvc.h:255
StatusCode
This class is used for returning status codes from appropriate routines.
Definition: StatusCode.h:50
AlgExecState::execStatus
const StatusCode & execStatus() const
Definition: IAlgExecStateSvc.h:33
AlgsExecutionStates::set
StatusCode set(unsigned int iAlgo, State newState)
Definition: AlgsExecutionStates.cpp:13
CommonMessagingBase::verbose
MsgStream & verbose() const
shortcut for the method msgStream(MSG::VERBOSE)
Definition: CommonMessaging.h:110
compareRootHistos.state
def state
Definition: compareRootHistos.py:455
CommonMessagingBase::debug
MsgStream & debug() const
shortcut for the method msgStream(MSG::DEBUG)
Definition: CommonMessaging.h:107
StatusCode::isSuccess
bool isSuccess() const
Definition: StatusCode.h:267
Gaudi::Hive::setCurrentContext
GAUDI_API void setCurrentContext(const EventContext *ctx)
Definition: ThreadLocalContext.cpp:31
EventContext::subSlot
ContextID_t subSlot() const
Definition: EventContext.h:42
StatusCode::FAILURE
constexpr static const auto FAILURE
Definition: StatusCode.h:86
MSG::ERROR
Definition: IMessageSvc.h:15
AvalancheSchedulerSvc::m_eventSlots
std::vector< EventSlot > m_eventSlots
Vector of events slots.
Definition: AvalancheSchedulerSvc.h:210
StatusCode::isFailure
bool isFailure() const
Definition: StatusCode.h:130
AvalancheSchedulerSvc::updateStates
StatusCode updateStates(int si=-1, int algo_index=-1, int sub_slot=-1, int source_slot=-1)
Loop on algorithm in the slots and promote them to successive states (-1 for algo_index means skippin...
Definition: AvalancheSchedulerSvc.cpp:576
IAlgResourcePool::releaseAlgorithm
virtual StatusCode releaseAlgorithm(const std::string &name, IAlgorithm *&algo)=0
Release a certain algorithm.
AlgExecState
Definition: IAlgExecStateSvc.h:27
AvalancheSchedulerSvc::m_algExecStateSvc
SmartIF< IAlgExecStateSvc > m_algExecStateSvc
Algorithm execution state manager.
Definition: AvalancheSchedulerSvc.h:219
endmsg
MsgStream & endmsg(MsgStream &s)
MsgStream Modifier: endmsg. Calls the output method of the MsgStream.
Definition: MsgStream.h:192
AvalancheSchedulerSvc::m_actionsQueue
tbb::concurrent_bounded_queue< action > m_actionsQueue
Queue where closures are stored and picked for execution.
Definition: AvalancheSchedulerSvc.h:260
EventSlot::algsStates
AlgsExecutionStates algsStates
Vector of algorithms states.
Definition: EventSlot.h:75
EventContext::usesSubSlot
bool usesSubSlot() const
Definition: EventContext.h:43
ON_VERBOSE
#define ON_VERBOSE
Definition: AvalancheSchedulerSvc.cpp:32

## ◆ promoteToAsyncScheduled()

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

Definition at line 946 of file AvalancheSchedulerSvc.cpp.

946 {
947
948 if ( m_IOBoundAlgosInFlight == m_maxIOBoundAlgosInFlight ) return StatusCode::FAILURE;
949
950 // bool IOBound = m_precSvc->isBlocking(algName);
951
952 const std::string& algName( index2algname( iAlgo ) );
953 IAlgorithm* ialgoPtr = nullptr;
954 StatusCode sc( m_algResourcePool->acquireAlgorithm( algName, ialgoPtr ) );
955
956 if ( sc.isSuccess() ) { // if we managed to get an algorithm instance try to schedule it
957
958 ++m_IOBoundAlgosInFlight;
959 auto promote2ExecutedClosure = [this, iAlgo, ialgoPtr, eventContext]() {
960 this->m_actionsQueue.push( [this, iAlgo, ialgoPtr, eventContext]() {
961 return this->AvalancheSchedulerSvc::promoteToAsyncExecuted( iAlgo, eventContext->slot(), ialgoPtr,
962 eventContext );
963 } );
964 return StatusCode::SUCCESS;
965 };
966 // Can we use tbb-based overloaded new-operator for a "custom" task (an algorithm wrapper, not derived from
967 // tbb::task)? it seems it works..
968 IOBoundAlgTask* theTask = new ( tbb::task::allocate_root() )
969 IOBoundAlgTask( ialgoPtr, *eventContext, serviceLocator(), m_algExecStateSvc, promote2ExecutedClosure );
970 m_IOBoundAlgScheduler->push( *theTask );
971
972 ON_DEBUG debug() << "[Asynchronous] Algorithm " << algName << " was submitted on event " << eventContext->evt()
973 << " in slot " << si << ". algorithms scheduled are " << m_IOBoundAlgosInFlight << endmsg;
974
975 // Update states in the appropriate event slot
976 StatusCode updateSc;
977 EventSlot& thisSlot = m_eventSlots[si];
978 if ( eventContext->usesSubSlot() ) {
979 // Sub-slot
980 size_t const subSlotIndex = eventContext->subSlot();
981 updateSc = thisSlot.allSubSlots[subSlotIndex].algsStates.set( iAlgo, AState::SCHEDULED );
982 } else {
983 // Event level (standard behaviour)
984 updateSc = thisSlot.algsStates.set( iAlgo, AState::SCHEDULED );
985 }
986
987 ON_VERBOSE if ( updateSc.isSuccess() ) verbose()
988 << "[Asynchronous] Promoting " << algName << " to SCHEDULED on slot " << si << endmsg;
989 return updateSc;
990 } else {
991 ON_DEBUG debug() << "[Asynchronous] Could not acquire instance for algorithm " << index2algname( iAlgo )
992 << " on slot " << si << endmsg;
993 return sc;
994 }
995 }
ON_DEBUG
#define ON_DEBUG
Definition: AvalancheSchedulerSvc.cpp:31
IOBoundAlgTask
Wrapper around I/O-bound Gaudi-algorithms.
Definition: IOBoundAlgTask.h:23
EventSlot
Class representing an event slot.
Definition: EventSlot.h:14
Service::serviceLocator
SmartIF< ISvcLocator > & serviceLocator() const override
Retrieve pointer to service locator.
Definition: Service.cpp:277
IAlgResourcePool::acquireAlgorithm
virtual StatusCode acquireAlgorithm(const std::string &name, IAlgorithm *&algo, bool blocking=false)=0
Acquire a certain algorithm using its name.
EventContext::slot
ContextID_t slot() const
Definition: EventContext.h:41
AvalancheSchedulerSvc::m_IOBoundAlgosInFlight
unsigned int m_IOBoundAlgosInFlight
Number of algorithms presently in flight.
Definition: AvalancheSchedulerSvc.h:228
EventContext::evt
ContextEvt_t evt() const
Definition: EventContext.h:40
StatusCode::SUCCESS
constexpr static const auto SUCCESS
Definition: StatusCode.h:85
EventSlot::allSubSlots
std::vector< EventSlot > allSubSlots
Actual sub-slot instances.
Definition: EventSlot.h:90
std::string
STL class.
AvalancheSchedulerSvc::m_algResourcePool
SmartIF< IAlgResourcePool > m_algResourcePool
Cache for the algorithm resource pool.
Definition: AvalancheSchedulerSvc.h:255
StatusCode
This class is used for returning status codes from appropriate routines.
Definition: StatusCode.h:50
AlgsExecutionStates::set
StatusCode set(unsigned int iAlgo, State newState)
Definition: AlgsExecutionStates.cpp:13
CommonMessagingBase::verbose
MsgStream & verbose() const
shortcut for the method msgStream(MSG::VERBOSE)
Definition: CommonMessaging.h:110
CommonMessagingBase::debug
MsgStream & debug() const
shortcut for the method msgStream(MSG::DEBUG)
Definition: CommonMessaging.h:107
StatusCode::isSuccess
bool isSuccess() const
Definition: StatusCode.h:267
AvalancheSchedulerSvc::m_IOBoundAlgScheduler
SmartIF< IAccelerator > m_IOBoundAlgScheduler
A shortcut to IO-bound algorithm scheduler.
Definition: AvalancheSchedulerSvc.h:207
IAlgorithm
The IAlgorithm is the interface implemented by the Algorithm base class.
Definition: IAlgorithm.h:28
EventContext::subSlot
ContextID_t subSlot() const
Definition: EventContext.h:42
AvalancheSchedulerSvc::index2algname
const std::string & index2algname(unsigned int index)
Convert an integer to a name.
Definition: AvalancheSchedulerSvc.cpp:439
StatusCode::FAILURE
constexpr static const auto FAILURE
Definition: StatusCode.h:86
AvalancheSchedulerSvc::m_eventSlots
std::vector< EventSlot > m_eventSlots
Vector of events slots.
Definition: AvalancheSchedulerSvc.h:210
AvalancheSchedulerSvc::promoteToAsyncExecuted
StatusCode promoteToAsyncExecuted(unsigned int iAlgo, int si, IAlgorithm *algo, EventContext *)
The call to this method is triggered only from within the IOBoundAlgTask.
Definition: AvalancheSchedulerSvc.cpp:1056
IAccelerator::push
virtual StatusCode push(IAlgTask &task)=0
AvalancheSchedulerSvc::m_algExecStateSvc
SmartIF< IAlgExecStateSvc > m_algExecStateSvc
Algorithm execution state manager.
Definition: AvalancheSchedulerSvc.h:219
AvalancheSchedulerSvc::m_maxIOBoundAlgosInFlight
Gaudi::Property< unsigned int > m_maxIOBoundAlgosInFlight
Definition: AvalancheSchedulerSvc.h:144
endmsg
MsgStream & endmsg(MsgStream &s)
MsgStream Modifier: endmsg. Calls the output method of the MsgStream.
Definition: MsgStream.h:192
AvalancheSchedulerSvc::m_actionsQueue
tbb::concurrent_bounded_queue< action > m_actionsQueue
Queue where closures are stored and picked for execution.
Definition: AvalancheSchedulerSvc.h:260
EventSlot::algsStates
AlgsExecutionStates algsStates
Vector of algorithms states.
Definition: EventSlot.h:75
EventContext::usesSubSlot
bool usesSubSlot() const
Definition: EventContext.h:43
ON_VERBOSE
#define ON_VERBOSE
Definition: AvalancheSchedulerSvc.cpp:32

## ◆ promoteToExecuted()

StatusCode AvalancheSchedulerSvc::promoteToExecuted | ( | unsigned int | iAlgo,
|  | int | si,
|  | IAlgorithm * | algo,
|  | EventContext * | eventContext
| ) |  |
| private

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

Definition at line 1002 of file AvalancheSchedulerSvc.cpp.

1003 {
1004 Gaudi::Hive::setCurrentContext( eventContext );
1005 StatusCode sc = m_algResourcePool->releaseAlgorithm( algo->name(), algo );
1006
1007 if ( sc.isFailure() ) {
1008 error() << "[Event " << eventContext->evt() << ", Slot " << eventContext->slot() << "] "
1009 << "Instance of algorithm " << algo->name() << " could not be properly put back." << endmsg;
1010 return StatusCode::FAILURE;
1011 }
1012
1013 --m_algosInFlight;
1014
1015 EventSlot& thisSlot = m_eventSlots[si];
1016
1017 ON_DEBUG debug() << "Trying to handle execution result of " << algo->name() << " on slot " << si << endmsg;
1018
1019 const AlgExecState& algstate = m_algExecStateSvc->algExecState( algo, *eventContext );
1020 AState state = algstate.execStatus().isSuccess()
1021 ? ( algstate.filterPassed() ? AState::EVTACCEPTED : AState::EVTREJECTED )
1022 : AState::ERROR;
1023
1024 // Update states in the appropriate slot
1025 int subSlotIndex = -1;
1026 if ( eventContext->usesSubSlot() ) {
1027 // Sub-slot
1028 subSlotIndex = eventContext->subSlot();
1029 sc = thisSlot.allSubSlots[subSlotIndex].algsStates.set( iAlgo, state );
1030 } else {
1031 // Event level (standard behaviour)
1032 sc = thisSlot.algsStates.set( iAlgo, state );
1033 }
1034
1035 ON_VERBOSE if ( sc.isSuccess() ) verbose()
1036 << "Promoting " << algo->name() << " on slot " << si << " to " << state << endmsg;
1037
1038 ON_DEBUG debug() << "Algorithm " << algo->name() << " executed in slot " << si << ". Algorithms scheduled are "
1039 << m_algosInFlight << endmsg;
1040
1041 // Schedule an update of the status of the algorithms
1042 ++m_actionsCounts[si];
1043 m_actionsQueue.push( [this, si, iAlgo, subSlotIndex]() {
1044 --this->m_actionsCounts[si]; // no bound check needed as decrements/increments are balanced in the current setup
1045 return this->updateStates( -1, iAlgo, subSlotIndex, si );
1046 } );
1047
1048 return sc;
1049 }
ON_DEBUG
#define ON_DEBUG
Definition: AvalancheSchedulerSvc.cpp:31
AvalancheSchedulerSvc::AState
AlgsExecutionStates::State AState
Definition: AvalancheSchedulerSvc.h:134
EventSlot
Class representing an event slot.
Definition: EventSlot.h:14
EventContext::slot
ContextID_t slot() const
Definition: EventContext.h:41
AvalancheSchedulerSvc::m_actionsCounts
std::vector< unsigned int > m_actionsCounts
Bookkeeping of the number of actions in flight per slot.
Definition: AvalancheSchedulerSvc.h:263
EventContext::evt
ContextEvt_t evt() const
Definition: EventContext.h:40
AlgExecState::filterPassed
bool filterPassed() const
Definition: IAlgExecStateSvc.h:31
EventSlot::allSubSlots
std::vector< EventSlot > allSubSlots
Actual sub-slot instances.
Definition: EventSlot.h:90
IAlgExecStateSvc::algExecState
virtual const AlgExecState & algExecState(const Gaudi::StringKey &algName, const EventContext &ctx) const =0
AvalancheSchedulerSvc::m_algosInFlight
unsigned int m_algosInFlight
Number of algorithms presently in flight.
Definition: AvalancheSchedulerSvc.h:225
CommonMessagingBase::error
MsgStream & error() const
shortcut for the method msgStream(MSG::ERROR)
Definition: CommonMessaging.h:98
AvalancheSchedulerSvc::m_algResourcePool
SmartIF< IAlgResourcePool > m_algResourcePool
Cache for the algorithm resource pool.
Definition: AvalancheSchedulerSvc.h:255
StatusCode
This class is used for returning status codes from appropriate routines.
Definition: StatusCode.h:50
AlgExecState::execStatus
const StatusCode & execStatus() const
Definition: IAlgExecStateSvc.h:33
AlgsExecutionStates::set
StatusCode set(unsigned int iAlgo, State newState)
Definition: AlgsExecutionStates.cpp:13
CommonMessagingBase::verbose
MsgStream & verbose() const
shortcut for the method msgStream(MSG::VERBOSE)
Definition: CommonMessaging.h:110
compareRootHistos.state
def state
Definition: compareRootHistos.py:455
CommonMessagingBase::debug
MsgStream & debug() const
shortcut for the method msgStream(MSG::DEBUG)
Definition: CommonMessaging.h:107
StatusCode::isSuccess
bool isSuccess() const
Definition: StatusCode.h:267
Gaudi::Hive::setCurrentContext
GAUDI_API void setCurrentContext(const EventContext *ctx)
Definition: ThreadLocalContext.cpp:31
EventContext::subSlot
ContextID_t subSlot() const
Definition: EventContext.h:42
StatusCode::FAILURE
constexpr static const auto FAILURE
Definition: StatusCode.h:86
MSG::ERROR
Definition: IMessageSvc.h:15
AvalancheSchedulerSvc::m_eventSlots
std::vector< EventSlot > m_eventSlots
Vector of events slots.
Definition: AvalancheSchedulerSvc.h:210
StatusCode::isFailure
bool isFailure() const
Definition: StatusCode.h:130
AvalancheSchedulerSvc::updateStates
StatusCode updateStates(int si=-1, int algo_index=-1, int sub_slot=-1, int source_slot=-1)
Loop on algorithm in the slots and promote them to successive states (-1 for algo_index means skippin...
Definition: AvalancheSchedulerSvc.cpp:576
IAlgResourcePool::releaseAlgorithm
virtual StatusCode releaseAlgorithm(const std::string &name, IAlgorithm *&algo)=0
Release a certain algorithm.
AlgExecState
Definition: IAlgExecStateSvc.h:27
AvalancheSchedulerSvc::m_algExecStateSvc
SmartIF< IAlgExecStateSvc > m_algExecStateSvc
Algorithm execution state manager.
Definition: AvalancheSchedulerSvc.h:219
endmsg
MsgStream & endmsg(MsgStream &s)
MsgStream Modifier: endmsg. Calls the output method of the MsgStream.
Definition: MsgStream.h:192
AvalancheSchedulerSvc::m_actionsQueue
tbb::concurrent_bounded_queue< action > m_actionsQueue
Queue where closures are stored and picked for execution.
Definition: AvalancheSchedulerSvc.h:260
EventSlot::algsStates
AlgsExecutionStates algsStates
Vector of algorithms states.
Definition: EventSlot.h:75
EventContext::usesSubSlot
bool usesSubSlot() const
Definition: EventContext.h:43
ON_VERBOSE
#define ON_VERBOSE
Definition: AvalancheSchedulerSvc.cpp:32

## ◆ promoteToFinished()

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

## ◆ promoteToScheduled()

StatusCode AvalancheSchedulerSvc::promoteToScheduled | ( | unsigned int | iAlgo,
|  | int | si,
|  | EventContext * | eventContext
| ) |  |
| private

Algorithm promotion.

Definition at line 885 of file AvalancheSchedulerSvc.cpp.

885 {
886
887 if ( m_algosInFlight == m_maxAlgosInFlight ) return StatusCode::FAILURE;
888
889 const std::string& algName( index2algname( iAlgo ) );
890 IAlgorithm* ialgoPtr = nullptr;
891 StatusCode sc( m_algResourcePool->acquireAlgorithm( algName, ialgoPtr ) );
892
893 if ( sc.isSuccess() ) { // if we managed to get an algorithm instance try to schedule it
894
895 ++m_algosInFlight;
896 auto promote2ExecutedClosure = [this, iAlgo, ialgoPtr, eventContext]() {
897 this->m_actionsQueue.push( [this, iAlgo, ialgoPtr, eventContext]() {
898 return this->AvalancheSchedulerSvc::promoteToExecuted( iAlgo, eventContext->slot(), ialgoPtr, eventContext );
899 } );
900 return StatusCode::SUCCESS;
901 };
902
903 // Avoid to use tbb if the pool size is 1 and run in this thread
904 if ( -100 != m_threadPoolSize ) {
905 // the child task that executes an Algorithm
906 tbb::task* algoTask = new ( tbb::task::allocate_root() )
907 AlgoExecutionTask( ialgoPtr, *eventContext, serviceLocator(), m_algExecStateSvc, promote2ExecutedClosure );
908 // schedule the algoTask
909 tbb::task::enqueue( *algoTask );
910
911 } else {
912 AlgoExecutionTask theTask( ialgoPtr, *eventContext, serviceLocator(), m_algExecStateSvc,
913 promote2ExecutedClosure );
914 theTask.execute();
915 }
916
917 ON_DEBUG debug() << "Algorithm " << algName << " was submitted on event " << eventContext->evt() << " in slot "
918 << si << ". Algorithms scheduled are " << m_algosInFlight << endmsg;
919
920 // Update states in the appropriate event slot
921 StatusCode updateSc;
922 EventSlot& thisSlot = m_eventSlots[si];
923 if ( eventContext->usesSubSlot() ) {
924 // Sub-slot
925 size_t const subSlotIndex = eventContext->subSlot();
926 updateSc = thisSlot.allSubSlots[subSlotIndex].algsStates.set( iAlgo, AState::SCHEDULED );
927 } else {
928 // Event level (standard behaviour)
929 updateSc = thisSlot.algsStates.set( iAlgo, AState::SCHEDULED );
930 }
931
932 ON_VERBOSE dumpSchedulerState( -1 );
933
934 if ( updateSc.isSuccess() )
935 ON_VERBOSE verbose() << "Promoting " << algName << " to SCHEDULED on slot " << si << endmsg;
936 return updateSc;
937 } else {
938 ON_DEBUG debug() << "Could not acquire instance for algorithm " << index2algname( iAlgo ) << " on slot " << si
939 << endmsg;
940 return sc;
941 }
942 }
ON_DEBUG
#define ON_DEBUG
Definition: AvalancheSchedulerSvc.cpp:31
EventSlot
Class representing an event slot.
Definition: EventSlot.h:14
Service::serviceLocator
SmartIF< ISvcLocator > & serviceLocator() const override
Retrieve pointer to service locator.
Definition: Service.cpp:277
IAlgResourcePool::acquireAlgorithm
virtual StatusCode acquireAlgorithm(const std::string &name, IAlgorithm *&algo, bool blocking=false)=0
Acquire a certain algorithm using its name.
EventContext::slot
ContextID_t slot() const
Definition: EventContext.h:41
EventContext::evt
ContextEvt_t evt() const
Definition: EventContext.h:40
AvalancheSchedulerSvc::dumpSchedulerState
void dumpSchedulerState(int iSlot)
Dump the state of the scheduler.
Definition: AvalancheSchedulerSvc.cpp:775
AlgoExecutionTask
Definition: AlgoExecutionTask.h:14
StatusCode::SUCCESS
constexpr static const auto SUCCESS
Definition: StatusCode.h:85
EventSlot::allSubSlots
std::vector< EventSlot > allSubSlots
Actual sub-slot instances.
Definition: EventSlot.h:90
std::string
STL class.
AvalancheSchedulerSvc::m_algosInFlight
unsigned int m_algosInFlight
Number of algorithms presently in flight.
Definition: AvalancheSchedulerSvc.h:225
AvalancheSchedulerSvc::m_algResourcePool
SmartIF< IAlgResourcePool > m_algResourcePool
Cache for the algorithm resource pool.
Definition: AvalancheSchedulerSvc.h:255
StatusCode
This class is used for returning status codes from appropriate routines.
Definition: StatusCode.h:50
AlgsExecutionStates::set
StatusCode set(unsigned int iAlgo, State newState)
Definition: AlgsExecutionStates.cpp:13
CommonMessagingBase::verbose
MsgStream & verbose() const
shortcut for the method msgStream(MSG::VERBOSE)
Definition: CommonMessaging.h:110
CommonMessagingBase::debug
MsgStream & debug() const
shortcut for the method msgStream(MSG::DEBUG)
Definition: CommonMessaging.h:107
AvalancheSchedulerSvc::m_threadPoolSize
Gaudi::Property< int > m_threadPoolSize
Definition: AvalancheSchedulerSvc.h:139
StatusCode::isSuccess
bool isSuccess() const
Definition: StatusCode.h:267
IAlgorithm
The IAlgorithm is the interface implemented by the Algorithm base class.
Definition: IAlgorithm.h:28
EventContext::subSlot
ContextID_t subSlot() const
Definition: EventContext.h:42
AvalancheSchedulerSvc::index2algname
const std::string & index2algname(unsigned int index)
Convert an integer to a name.
Definition: AvalancheSchedulerSvc.cpp:439
AvalancheSchedulerSvc::m_maxAlgosInFlight
size_t m_maxAlgosInFlight
Definition: AvalancheSchedulerSvc.h:270
StatusCode::FAILURE
constexpr static const auto FAILURE
Definition: StatusCode.h:86
AvalancheSchedulerSvc::promoteToExecuted
StatusCode promoteToExecuted(unsigned int iAlgo, int si, IAlgorithm *algo, EventContext *)
The call to this method is triggered only from within the AlgoExecutionTask.
Definition: AvalancheSchedulerSvc.cpp:1002
AvalancheSchedulerSvc::m_eventSlots
std::vector< EventSlot > m_eventSlots
Vector of events slots.
Definition: AvalancheSchedulerSvc.h:210
AvalancheSchedulerSvc::m_algExecStateSvc
SmartIF< IAlgExecStateSvc > m_algExecStateSvc
Algorithm execution state manager.
Definition: AvalancheSchedulerSvc.h:219
endmsg
MsgStream & endmsg(MsgStream &s)
MsgStream Modifier: endmsg. Calls the output method of the MsgStream.
Definition: MsgStream.h:192
AvalancheSchedulerSvc::m_actionsQueue
tbb::concurrent_bounded_queue< action > m_actionsQueue
Queue where closures are stored and picked for execution.
Definition: AvalancheSchedulerSvc.h:260
EventSlot::algsStates
AlgsExecutionStates algsStates
Vector of algorithms states.
Definition: EventSlot.h:75
EventContext::usesSubSlot
bool usesSubSlot() const
Definition: EventContext.h:43
ON_VERBOSE
#define ON_VERBOSE
Definition: AvalancheSchedulerSvc.cpp:32

## ◆ 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 457 of file AvalancheSchedulerSvc.cpp.

457 {
458
459 if ( !eventContext ) {
460 fatal() << "Event context is nullptr" << endmsg;
461 return StatusCode::FAILURE;
462 }
463
464 if ( m_freeSlots.load() == 0 ) {
465 ON_DEBUG debug() << "A free processing slot could not be found." << endmsg;
466 return StatusCode::FAILURE;
467 }
468
469 // no problem as push new event is only called from one thread (event loop manager)
470 --m_freeSlots;
471
472 auto action = [this, eventContext]() -> StatusCode {
473 // Event processing slot forced to be the same as the wb slot
474 const unsigned int thisSlotNum = eventContext->slot();
475 EventSlot& thisSlot = m_eventSlots[thisSlotNum];
476 if ( !thisSlot.complete ) {
477 fatal() << "The slot " << thisSlotNum << " is supposed to be a finished event but it's not" << endmsg;
478 return StatusCode::FAILURE;
479 }
480
481 ON_DEBUG debug() << "Executing event " << eventContext->evt() << " on slot " << thisSlotNum << endmsg;
482 thisSlot.reset( eventContext );
483
484 // Result status code:
485 StatusCode result = StatusCode::SUCCESS;
486
487 // promote to CR and DR the initial set of algorithms
488 Cause cs = {Cause::source::Root, "RootDecisionHub"};
489 if ( m_precSvc->iterate( thisSlot, cs ).isFailure() ) {
490 error() << "Failed to call IPrecedenceSvc::iterate for slot " << thisSlotNum << endmsg;
491 result = StatusCode::FAILURE;
492 }
493
494 if ( this->updateStates( thisSlotNum ).isFailure() ) {
495 error() << "Failed to call AvalancheSchedulerSvc::updateStates for slot " << thisSlotNum << endmsg;
496 result = StatusCode::FAILURE;
497 }
498
499 return result;
500 }; // end of lambda
501
502 // Kick off the scheduling!
503 ON_VERBOSE {
504 verbose() << "Pushing the action to update the scheduler for slot " << eventContext->slot() << endmsg;
505 verbose() << "Free slots available " << m_freeSlots.load() << endmsg;
506 }
507
508 m_actionsQueue.push( action );
509
510 return StatusCode::SUCCESS;
511 }
ON_DEBUG
#define ON_DEBUG
Definition: AvalancheSchedulerSvc.cpp:31
EventSlot
Class representing an event slot.
Definition: EventSlot.h:14
EventContext::slot
ContextID_t slot() const
Definition: EventContext.h:41
AvalancheSchedulerSvc::m_precSvc
SmartIF< IPrecedenceSvc > m_precSvc
A shortcut to the Precedence Service.
Definition: AvalancheSchedulerSvc.h:201
IPrecedenceSvc::iterate
virtual StatusCode iterate(EventSlot &, const Cause &)=0
Infer the precedence effect caused by an execution flow event.
EventContext::evt
ContextEvt_t evt() const
Definition: EventContext.h:40
StatusCode::SUCCESS
constexpr static const auto SUCCESS
Definition: StatusCode.h:85
AvalancheSchedulerSvc::m_freeSlots
std::atomic_int m_freeSlots
Atomic to account for asyncronous updates by the scheduler wrt the rest.
Definition: AvalancheSchedulerSvc.h:213
AvalancheSchedulerSvc::action
std::function< StatusCode()> action
Definition: AvalancheSchedulerSvc.h:135
CommonMessagingBase::error
MsgStream & error() const
shortcut for the method msgStream(MSG::ERROR)
Definition: CommonMessaging.h:98
Cause
Definition: PrecedenceRulesGraph.h:387
StatusCode
This class is used for returning status codes from appropriate routines.
Definition: StatusCode.h:50
CommonMessagingBase::verbose
MsgStream & verbose() const
shortcut for the method msgStream(MSG::VERBOSE)
Definition: CommonMessaging.h:110
Cause::source::Root
CommonMessagingBase::debug
MsgStream & debug() const
shortcut for the method msgStream(MSG::DEBUG)
Definition: CommonMessaging.h:107
StatusCode::FAILURE
constexpr static const auto FAILURE
Definition: StatusCode.h:86
AvalancheSchedulerSvc::m_eventSlots
std::vector< EventSlot > m_eventSlots
Vector of events slots.
Definition: AvalancheSchedulerSvc.h:210
EventSlot::complete
bool complete
Flags completion of the event.
Definition: EventSlot.h:79
StatusCode::isFailure
bool isFailure() const
Definition: StatusCode.h:130
EventSlot::reset
void reset(EventContext *theeventContext)
Reset all resources in order to reuse the slot (thread-unsafe)
Definition: EventSlot.h:39
AvalancheSchedulerSvc::updateStates
StatusCode updateStates(int si=-1, int algo_index=-1, int sub_slot=-1, int source_slot=-1)
Loop on algorithm in the slots and promote them to successive states (-1 for algo_index means skippin...
Definition: AvalancheSchedulerSvc.cpp:576
CommonMessagingBase::fatal
MsgStream & fatal() const
shortcut for the method msgStream(MSG::FATAL)
Definition: CommonMessaging.h:92
endmsg
MsgStream & endmsg(MsgStream &s)
MsgStream Modifier: endmsg. Calls the output method of the MsgStream.
Definition: MsgStream.h:192
AvalancheSchedulerSvc::m_actionsQueue
tbb::concurrent_bounded_queue< action > m_actionsQueue
Queue where closures are stored and picked for execution.
Definition: AvalancheSchedulerSvc.h:260
ON_VERBOSE
#define ON_VERBOSE
Definition: AvalancheSchedulerSvc.cpp:32

## ◆ pushNewEvents()

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

Definition at line 515 of file AvalancheSchedulerSvc.cpp.

515 {
516 StatusCode sc;
517 for ( auto context : eventContexts ) {
518 sc = pushNewEvent( context );
519 if ( sc != StatusCode::SUCCESS ) return sc;
520 }
521 return sc;
522 }
StatusCode::SUCCESS
constexpr static const auto SUCCESS
Definition: StatusCode.h:85
StatusCode
This class is used for returning status codes from appropriate routines.
Definition: StatusCode.h:50
AvalancheSchedulerSvc::pushNewEvent
StatusCode pushNewEvent(EventContext *eventContext) override
Make an event available to the scheduler.
Definition: AvalancheSchedulerSvc.cpp:457

## ◆ scheduleEventView()

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

Method to inform the scheduler about event views.

Definition at line 1110 of file AvalancheSchedulerSvc.cpp.

1111 {
1112 // Prevent view nesting
1113 if ( sourceContext->usesSubSlot() ) {
1114 fatal() << "Attempted to nest EventViews at node " << nodeName << ": this is not supported" << endmsg;
1115 return StatusCode::FAILURE;
1116 }
1117
1118 ON_VERBOSE verbose() << "Queuing a view for [" << viewContext.get() << "]" << endmsg;
1119
1120 // It's not possible to create an std::functional from a move-capturing lambda
1121 // So, we have to release the unique pointer
1122 auto action = [this, slotIndex = sourceContext->slot(), viewContextPtr = viewContext.release(),
1123 &nodeName]() -> StatusCode {
1124 // Attach the sub-slot to the top-level slot
1125 EventSlot& topSlot = this->m_eventSlots[slotIndex];
1126
1127 if ( viewContextPtr ) {
1128 // Re-create the unique pointer
1129 auto viewContext = std::unique_ptr<EventContext>( viewContextPtr );
1130 topSlot.addSubSlot( std::move( viewContext ), nodeName );
1131 return StatusCode::SUCCESS;
1132 } else {
1133 // Disable the view node if there are no views
1134 topSlot.disableSubSlots( nodeName );
1135 return StatusCode::SUCCESS;
1136 }
1137 };
1138
1139 m_actionsQueue.push( std::move( action ) );
1140
1141 return StatusCode::SUCCESS;
1142 }
EventSlot
Class representing an event slot.
Definition: EventSlot.h:14
EventContext::slot
ContextID_t slot() const
Definition: EventContext.h:41
EventSlot::disableSubSlots
void disableSubSlots(const std::string &nodeName)
Disable event views for a given CF view node by registering an empty container Contact B.
Definition: EventSlot.h:68
EventSlot::addSubSlot
void addSubSlot(std::unique_ptr< EventContext > viewContext, const std::string &nodeName)
Add a subslot to the slot (this constructs a new slot and registers it with the parent one)
Definition: EventSlot.h:51
StatusCode::SUCCESS
constexpr static const auto SUCCESS
Definition: StatusCode.h:85
std::unique_ptr::release
T release(T... args)
AvalancheSchedulerSvc::action
std::function< StatusCode()> action
Definition: AvalancheSchedulerSvc.h:135
StatusCode
This class is used for returning status codes from appropriate routines.
Definition: StatusCode.h:50
CommonMessagingBase::verbose
MsgStream & verbose() const
shortcut for the method msgStream(MSG::VERBOSE)
Definition: CommonMessaging.h:110
std::move
T move(T... args)
std::unique_ptr::get
T get(T... args)
std::unique_ptr< EventContext >
StatusCode::FAILURE
constexpr static const auto FAILURE
Definition: StatusCode.h:86
AvalancheSchedulerSvc::m_eventSlots
std::vector< EventSlot > m_eventSlots
Vector of events slots.
Definition: AvalancheSchedulerSvc.h:210
CommonMessagingBase::fatal
MsgStream & fatal() const
shortcut for the method msgStream(MSG::FATAL)
Definition: CommonMessaging.h:92
endmsg
MsgStream & endmsg(MsgStream &s)
MsgStream Modifier: endmsg. Calls the output method of the MsgStream.
Definition: MsgStream.h:192
AvalancheSchedulerSvc::m_actionsQueue
tbb::concurrent_bounded_queue< action > m_actionsQueue
Queue where closures are stored and picked for execution.
Definition: AvalancheSchedulerSvc.h:260
EventContext::usesSubSlot
bool usesSubSlot() const
Definition: EventContext.h:43
ON_VERBOSE
#define ON_VERBOSE
Definition: AvalancheSchedulerSvc.cpp:32

## ◆ 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 553 of file AvalancheSchedulerSvc.cpp.

553 {
554
555 if ( m_finishedEvents.try_pop( eventContext ) ) {
556 ON_DEBUG debug() << "Try Pop successful slot " << eventContext->slot() << "(event " << eventContext->evt() << ")"
557 << endmsg;
558 ++m_freeSlots;
559 return StatusCode::SUCCESS;
560 }
561 return StatusCode::FAILURE;
562 }
ON_DEBUG
#define ON_DEBUG
Definition: AvalancheSchedulerSvc.cpp:31
EventContext::slot
ContextID_t slot() const
Definition: EventContext.h:41
EventContext::evt
ContextEvt_t evt() const
Definition: EventContext.h:40
StatusCode::SUCCESS
constexpr static const auto SUCCESS
Definition: StatusCode.h:85
AvalancheSchedulerSvc::m_freeSlots
std::atomic_int m_freeSlots
Atomic to account for asyncronous updates by the scheduler wrt the rest.
Definition: AvalancheSchedulerSvc.h:213
AvalancheSchedulerSvc::m_finishedEvents
tbb::concurrent_bounded_queue< EventContext * > m_finishedEvents
Queue of finished events.
Definition: AvalancheSchedulerSvc.h:216
CommonMessagingBase::debug
MsgStream & debug() const
shortcut for the method msgStream(MSG::DEBUG)
Definition: CommonMessaging.h:107
StatusCode::FAILURE
constexpr static const auto FAILURE
Definition: StatusCode.h:86
endmsg
MsgStream & endmsg(MsgStream &s)
MsgStream Modifier: endmsg. Calls the output method of the MsgStream.
Definition: MsgStream.h:192

## ◆ updateStates()

StatusCode AvalancheSchedulerSvc::updateStates | ( | int | si = `-1`,
|  | int | algo_index = `-1`,
|  | int | sub_slot = `-1`,
|  | int | source_slot = `-1`
| ) |  |
| private

Loop on algorithm in the slots and promote them to successive states (-1 for algo_index means skipping an update of the Control Flow state)

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 576 of file AvalancheSchedulerSvc.cpp.

577 {
578
579 StatusCode global_sc( StatusCode::SUCCESS );
580
581 // Sort from the oldest to the newest event
582 // Prepare a vector of pointers to the slots to avoid copies
583 std::vector<EventSlot*> eventSlotsPtrs;
584
585 // Consider all slots if si <0 or just one otherwise
586 if ( si < 0 ) {
587 const int eventsSlotsSize( m_eventSlots.size() );
588 eventSlotsPtrs.reserve( eventsSlotsSize );
589 for ( auto slotIt = m_eventSlots.begin(); slotIt != m_eventSlots.end(); ++slotIt ) {
590 if ( !slotIt->complete ) eventSlotsPtrs.push_back( &( *slotIt ) );
591 }
592 std::sort( eventSlotsPtrs.begin(), eventSlotsPtrs.end(),
593 []( EventSlot* a, EventSlot* b ) { return a->eventContext->evt() < b->eventContext->evt(); } );
594 } else {
595 eventSlotsPtrs.push_back( &m_eventSlots[si] );
596 }
597
598 for ( EventSlot* thisSlotPtr : eventSlotsPtrs ) {
599 int iSlot = thisSlotPtr->eventContext->slot();
600
601 // Cache the states of the algos to improve readability and performance
602 auto& thisSlot = m_eventSlots[iSlot];
603 AlgsExecutionStates& thisAlgsStates = thisSlot.algsStates;
604
605 // Perform the I->CR->DR transitions
606 if ( algo_index >= 0 ) {
607 Cause cs = {Cause::source::Task, index2algname( algo_index )};
608
609 // Run in whole-event context if there's no sub-slot index, or the sub-slot has a different parent
610 if ( sub_slot == -1 || iSlot != source_slot ) {
611 if ( m_precSvc->iterate( thisSlot, cs ).isFailure() ) {
612 error() << "Failed to call IPrecedenceSvc::iterate for slot " << iSlot << endmsg;
613 global_sc = StatusCode::FAILURE;
614 }
615 } else {
616 if ( m_precSvc->iterate( thisSlot.allSubSlots[sub_slot], cs ).isFailure() ) {
617 error() << "Failed to call IPrecedenceSvc::iterate for sub-slot " << sub_slot << " of " << iSlot << endmsg;
618 global_sc = StatusCode::FAILURE;
619 }
620 }
621 }
622
623 StatusCode partial_sc( StatusCode::FAILURE, true );
624
625 // Perform DR->SCHEDULED
626 if ( !m_optimizationMode.empty() ) {
627 auto comp_nodes = [this]( const uint& i, const uint& j ) {
628 return ( m_precSvc->getPriority( index2algname( i ) ) < m_precSvc->getPriority( index2algname( j ) ) );
629 };
630 std::priority_queue<uint, std::vector<uint>, std::function<bool( const uint&, const uint& )>> buffer(
631 comp_nodes, std::vector<uint>() );
632 for ( auto it = thisAlgsStates.begin( AState::DATAREADY ); it != thisAlgsStates.end( AState::DATAREADY ); ++it )
633 buffer.push( *it );
634 while ( !buffer.empty() ) {
635 bool IOBound = false;
636 if ( m_useIOBoundAlgScheduler ) IOBound = m_precSvc->isBlocking( index2algname( buffer.top() ) );
637
638 if ( !IOBound )
639 partial_sc = promoteToScheduled( buffer.top(), iSlot, thisSlotPtr->eventContext.get() );
640 else
641 partial_sc = promoteToAsyncScheduled( buffer.top(), iSlot, thisSlotPtr->eventContext.get() );
642
643 ON_VERBOSE if ( partial_sc.isFailure() ) verbose()
644 << "Could not apply transition from " << AState::DATAREADY << " for algorithm "
645 << index2algname( buffer.top() ) << " on processing slot " << iSlot << endmsg;
646
647 buffer.pop();
648 }
649
650 } else {
651 for ( auto it = thisAlgsStates.begin( AState::DATAREADY ); it != thisAlgsStates.end( AState::DATAREADY ); ++it ) {
652 uint algIndex = *it;
653
654 bool IOBound = false;
655 if ( m_useIOBoundAlgScheduler ) IOBound = m_precSvc->isBlocking( index2algname( algIndex ) );
656
657 if ( !IOBound )
658 partial_sc = promoteToScheduled( algIndex, iSlot, thisSlotPtr->eventContext.get() );
659 else
660 partial_sc = promoteToAsyncScheduled( algIndex, iSlot, thisSlotPtr->eventContext.get() );
661
662 ON_VERBOSE if ( partial_sc.isFailure() ) verbose()
663 << "Could not apply transition from " << AState::DATAREADY << " for algorithm " << index2algname( algIndex )
664 << " on processing slot " << iSlot << endmsg;
665 }
666 }
667
668 // Check for algorithms ready in sub-slots
669 for ( auto& subslot : thisSlot.allSubSlots ) {
670 auto& subslotStates = subslot.algsStates;
671 for ( auto it = subslotStates.begin( AState::DATAREADY ); it != subslotStates.end( AState::DATAREADY ); ++it ) {
672 uint algIndex{*it};
673 partial_sc = promoteToScheduled( algIndex, iSlot, subslot.eventContext.get() );
674 // The following verbosity is expensive when the number of sub-slots is high
675 /*ON_VERBOSE if ( partial_sc.isFailure() ) verbose()
676 << "Could not apply transition from " << AState::DATAREADY << " for algorithm " << index2algname( algIndex )
677 << " on processing subslot " << subslot.eventContext->slot() << endmsg;*/
678 }
679 }
680
681 if ( m_dumpIntraEventDynamics ) {
682 std::stringstream s;
683 s << ( algo_index != -1 ? index2algname( algo_index ) : "START" ) << ", "
684 << thisAlgsStates.sizeOfSubset( AState::CONTROLREADY ) << ", "
685 << thisAlgsStates.sizeOfSubset( AState::DATAREADY ) << ", " << thisAlgsStates.sizeOfSubset( AState::SCHEDULED )
686 << ", " << std::chrono::high_resolution_clock::now().time_since_epoch().count() << "\n";
687 auto threads = ( m_threadPoolSize != -1 ) ? std::to_string( m_threadPoolSize )
688 : std::to_string( tbb::task_scheduler_init::default_num_threads() );
689 std::ofstream myfile;
690 myfile.open( "IntraEventFSMOccupancy_" + threads + "T.csv", std::ios::app );
691 myfile << s.str();
692 myfile.close();
693 }
694
695 // Not complete because this would mean that the slot is already free!
696 if ( m_precSvc->CFRulesResolved( thisSlot ) &&
697 !thisSlot.algsStates.containsAny( {AState::CONTROLREADY, AState::DATAREADY, AState::SCHEDULED} ) &&
698 !subSlotAlgsInStates( thisSlot, {AState::CONTROLREADY, AState::DATAREADY, AState::SCHEDULED} ) &&
699 !thisSlot.complete ) {
700
701 thisSlot.complete = true;
702 // if the event did not fail, add it to the finished events
703 // otherwise it is taken care of in the error handling
704 if ( m_algExecStateSvc->eventStatus( *thisSlot.eventContext ) == EventStatus::Success ) {
705 ON_DEBUG debug() << "Event " << thisSlot.eventContext->evt() << " finished (slot "
706 << thisSlot.eventContext->slot() << ")." << endmsg;
707 m_finishedEvents.push( thisSlot.eventContext.release() );
708 }
709
710 // now let's return the fully evaluated result of the control flow
711 ON_DEBUG debug() << m_precSvc->printState( thisSlot ) << endmsg;
712
713 thisSlot.eventContext.reset( nullptr );
714
715 } else if ( isStalled( thisSlot ) ) {
716 m_algExecStateSvc->setEventStatus( EventStatus::AlgStall, *thisSlot.eventContext );
717 eventFailed( thisSlot.eventContext.get() ); // can't release yet
718 }
719 partial_sc.ignore();
720 } // end loop on slots
721
722 ON_VERBOSE verbose() << "States Updated." << endmsg;
723
724 return global_sc;
725 }
ON_DEBUG
#define ON_DEBUG
Definition: AvalancheSchedulerSvc.cpp:31
AvalancheSchedulerSvc::isStalled
bool isStalled(const EventSlot &) const
Check if scheduling in a particular slot is in a stall.
Definition: AvalancheSchedulerSvc.cpp:735
AvalancheSchedulerSvc::AState
AlgsExecutionStates::State AState
Definition: AvalancheSchedulerSvc.h:134
EventSlot
Class representing an event slot.
Definition: EventSlot.h:14
Cause::source::Task
EventContext::slot
ContextID_t slot() const
Definition: EventContext.h:41
std::ofstream::open
T open(T... args)
AvalancheSchedulerSvc::m_precSvc
SmartIF< IPrecedenceSvc > m_precSvc
A shortcut to the Precedence Service.
Definition: AvalancheSchedulerSvc.h:201
IPrecedenceSvc::iterate
virtual StatusCode iterate(EventSlot &, const Cause &)=0
Infer the precedence effect caused by an execution flow event.
EventContext::evt
ContextEvt_t evt() const
Definition: EventContext.h:40
std::function
AvalancheSchedulerSvc::m_dumpIntraEventDynamics
Gaudi::Property< bool > m_dumpIntraEventDynamics
Definition: AvalancheSchedulerSvc.h:151
IAlgExecStateSvc::eventStatus
virtual const EventStatus::Status & eventStatus(const EventContext &ctx) const =0
std::to_string
T to_string(T... args)
StatusCode::SUCCESS
constexpr static const auto SUCCESS
Definition: StatusCode.h:85
std
STL namespace.
std::vector::end
T end(T... args)
AvalancheSchedulerSvc::m_optimizationMode
Gaudi::Property< std::string > m_optimizationMode
Definition: AvalancheSchedulerSvc.h:149
IPrecedenceSvc::CFRulesResolved
virtual bool CFRulesResolved(EventSlot &) const =0
Check if control flow rules are resolved.
std::stringstream
STL class.
AvalancheSchedulerSvc::m_useIOBoundAlgScheduler
Gaudi::Property< bool > m_useIOBoundAlgScheduler
Definition: AvalancheSchedulerSvc.h:153
std::vector::push_back
T push_back(T... args)
std::ofstream
STL class.
CommonMessagingBase::error
MsgStream & error() const
shortcut for the method msgStream(MSG::ERROR)
Definition: CommonMessaging.h:98
AlgsExecutionStates
The AlgsExecutionStates encodes the state machine for the execution of algorithms within a single eve...
Definition: AlgsExecutionStates.h:26
AvalancheSchedulerSvc::promoteToScheduled
StatusCode promoteToScheduled(unsigned int iAlgo, int si, EventContext *)
Algorithm promotion.
Definition: AvalancheSchedulerSvc.cpp:885
Cause
Definition: PrecedenceRulesGraph.h:387
std::priority_queue
STL class.
StatusCode
This class is used for returning status codes from appropriate routines.
Definition: StatusCode.h:50
AvalancheSchedulerSvc::m_finishedEvents
tbb::concurrent_bounded_queue< EventContext * > m_finishedEvents
Queue of finished events.
Definition: AvalancheSchedulerSvc.h:216
std::ofstream::close
T close(T... args)
CommonMessagingBase::verbose
MsgStream & verbose() const
shortcut for the method msgStream(MSG::VERBOSE)
Definition: CommonMessaging.h:110
EventStatus::AlgStall
Definition: IAlgExecStateSvc.h:63
EventStatus::Success
Definition: IAlgExecStateSvc.h:63
IAlgExecStateSvc::setEventStatus
virtual void setEventStatus(const EventStatus::Status &sc, const EventContext &ctx)=0
CommonMessagingBase::debug
MsgStream & debug() const
shortcut for the method msgStream(MSG::DEBUG)
Definition: CommonMessaging.h:107
AvalancheSchedulerSvc::m_threadPoolSize
Gaudi::Property< int > m_threadPoolSize
Definition: AvalancheSchedulerSvc.h:139
AvalancheSchedulerSvc::promoteToAsyncScheduled
StatusCode promoteToAsyncScheduled(unsigned int iAlgo, int si, EventContext *)
Definition: AvalancheSchedulerSvc.cpp:946
std::count
T count(T... args)
std::vector::size
T size(T... args)
std::vector
STL class.
std::vector::begin
T begin(T... args)
AlgsExecutionStates::begin
Iterator begin(State kind)
Definition: AlgsExecutionStates.h:103
AvalancheSchedulerSvc::index2algname
const std::string & index2algname(unsigned int index)
Convert an integer to a name.
Definition: AvalancheSchedulerSvc.cpp:439
IPrecedenceSvc::getPriority
virtual uint getPriority(const std::string &) const =0
Get task priority.
gaudirun.s
string s
Definition: gaudirun.py:316
StatusCode::FAILURE
constexpr static const auto FAILURE
Definition: StatusCode.h:86
AvalancheSchedulerSvc::eventFailed
void eventFailed(EventContext *eventContext)
Method to execute if an event failed.
Definition: AvalancheSchedulerSvc.cpp:754
AvalancheSchedulerSvc::m_eventSlots
std::vector< EventSlot > m_eventSlots
Vector of events slots.
Definition: AvalancheSchedulerSvc.h:210
std::sort
T sort(T... args)
StatusCode::isFailure
bool isFailure() const
Definition: StatusCode.h:130
IPrecedenceSvc::isBlocking
virtual bool isBlocking(const std::string &) const =0
Check if a task is CPU-blocking.
AvalancheSchedulerSvc::m_algExecStateSvc
SmartIF< IAlgExecStateSvc > m_algExecStateSvc
Algorithm execution state manager.
Definition: AvalancheSchedulerSvc.h:219
IPrecedenceSvc::printState
virtual const std::string printState(EventSlot &) const =0
endmsg
MsgStream & endmsg(MsgStream &s)
MsgStream Modifier: endmsg. Calls the output method of the MsgStream.
Definition: MsgStream.h:192
AvalancheSchedulerErrorTest.threads
int threads
Definition: AvalancheSchedulerErrorTest.py:17
EventSlot::eventContext
std::unique_ptr< EventContext > eventContext
Cache for the eventContext.
Definition: EventSlot.h:73
std::vector::reserve
T reserve(T... args)
ON_VERBOSE
#define ON_VERBOSE
Definition: AvalancheSchedulerSvc.cpp:32
AlgsExecutionStates::end
Iterator end(State kind)
Definition: AlgsExecutionStates.h:104

## Member Data Documentation

## ◆ m_actionsCounts

std::vector<unsigned int> AvalancheSchedulerSvc::m_actionsCounts
| private

Bookkeeping of the number of actions in flight per slot.

Definition at line 263 of file AvalancheSchedulerSvc.h.

## ◆ m_actionsQueue

tbb::concurrent_bounded_queue<action> AvalancheSchedulerSvc::m_actionsQueue
| private

Queue where closures are stored and picked for execution.

Definition at line 260 of file AvalancheSchedulerSvc.h.

## ◆ m_algExecStateSvc

SmartIF<IAlgExecStateSvc> AvalancheSchedulerSvc::m_algExecStateSvc
| private

Algorithm execution state manager.

Definition at line 219 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 192 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 198 of file AvalancheSchedulerSvc.h.

## ◆ m_algosInFlight

unsigned int AvalancheSchedulerSvc::m_algosInFlight = 0
| private

Number of algorithms presently in flight.

Definition at line 225 of file AvalancheSchedulerSvc.h.

## ◆ m_algResourcePool

SmartIF<IAlgResourcePool> AvalancheSchedulerSvc::m_algResourcePool
| private

Cache for the algorithm resource pool.

Definition at line 255 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 156 of file AvalancheSchedulerSvc.h.

## ◆ m_condSvc

SmartIF<ICondSvc> AvalancheSchedulerSvc::m_condSvc
| private

A shortcut to service for Conditions handling.

Definition at line 222 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 151 of file AvalancheSchedulerSvc.h.

## ◆ m_enableCondSvc

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

Definition at line 161 of file AvalancheSchedulerSvc.h.

## ◆ m_eventSlots

std::vector<EventSlot> AvalancheSchedulerSvc::m_eventSlots
| private

Vector of events slots.

Definition at line 210 of file AvalancheSchedulerSvc.h.

## ◆ m_finishedEvents

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

Queue of finished events.

Definition at line 216 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 213 of file AvalancheSchedulerSvc.h.

## ◆ m_IOBoundAlgosInFlight

unsigned int AvalancheSchedulerSvc::m_IOBoundAlgosInFlight = 0
| private

Number of algorithms presently in flight.

Definition at line 228 of file AvalancheSchedulerSvc.h.

## ◆ m_IOBoundAlgScheduler

SmartIF<IAccelerator> AvalancheSchedulerSvc::m_IOBoundAlgScheduler
| private

A shortcut to IO-bound algorithm scheduler.

Definition at line 207 of file AvalancheSchedulerSvc.h.

## ◆ m_IOBoundAlgSchedulerSvcName

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

Definition at line 143 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 183 of file AvalancheSchedulerSvc.h.

## ◆ m_maxAlgosInFlight

size_t AvalancheSchedulerSvc::m_maxAlgosInFlight {1}
| private

Definition at line 270 of file AvalancheSchedulerSvc.h.

## ◆ m_maxEventsInFlight

size_t AvalancheSchedulerSvc::m_maxEventsInFlight {0}
| private

Definition at line 269 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 144 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 149 of file AvalancheSchedulerSvc.h.

## ◆ m_precSvc

SmartIF<IPrecedenceSvc> AvalancheSchedulerSvc::m_precSvc
| private

A shortcut to the Precedence Service.

Definition at line 201 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 169 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 163 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 166 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 146 of file AvalancheSchedulerSvc.h.

## ◆ m_thread

std::thread AvalancheSchedulerSvc::m_thread
| private

The thread in which the activate function runs.

Definition at line 186 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 139 of file AvalancheSchedulerSvc.h.

## ◆ m_threadPoolSvc

SmartIF<IThreadPoolSvc> AvalancheSchedulerSvc::m_threadPoolSvc
| private

Definition at line 268 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 158 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 153 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 172 of file AvalancheSchedulerSvc.h.

## ◆ m_whiteboard

SmartIF<IHiveWhiteBoard> AvalancheSchedulerSvc::m_whiteboard
| private

A shortcut to the whiteboard.

Definition at line 204 of file AvalancheSchedulerSvc.h.

## ◆ m_whiteboardSvcName

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

Definition at line 142 of file AvalancheSchedulerSvc.h.

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

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