| The Gaudi Framework v29r1 (5cd95fa2)

- Main Page
- Related Pages
- Modules
- Namespaces
- Classes
- Files

- Class List
- Class Index
- Class Hierarchy
- Class Members

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

# Introduction

More...

`#include <src/AvalancheSchedulerSvc.h>`

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

## Classes
struct | enqueueSchedulerActionTask
class | SchedulerState

## 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 availble. More...
StatusCode | tryPopFinishedEvent (EventContext *&eventContext) override
| Try to fetch an event from the scheduler. More...
unsigned int | freeSlots () override
| Get free slots number. More...
void | addAlg (Algorithm *, EventContext *, pthread_t)
bool | delAlg (Algorithm *)
void | dumpState () override
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...
| ~extends () override=default
| Virtual destructor. 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
| ~PropertyHolder () override=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, TYPE &value, const std::string &doc="none") const
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
Gaudi::Details::PropertyBase * | declareProperty (const std::string &name, GaudiHandleBase &ref, const std::string &doc="none")
| Specializations for various GaudiHandles. More...
Gaudi::Details::PropertyBase * | declareProperty (const std::string &name, GaudiHandleArrayBase &ref, const std::string &doc="none")
Gaudi::Details::PropertyBase * | declareProperty (const std::string &name, DataObjectHandleBase &ref, const std::string &doc="none")
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...
MSG::Level | msgLevel () const
| get the cached level (originally extracted from the embedded MsgStream) More...
MSG::Level | outputLevel () const __attribute__((deprecated))
| Backward compatibility function for getting the output level. More...
bool | msgLevel (MSG::Level lvl) const
| get the output level from the embedded MsgStream More...
Public Member Functions inherited from extend_interfaces< Interfaces... >
| ~extend_interfaces () override=default
| Virtual destructor. More...

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

## 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 | eventFailed (EventContext *eventContext)
| Method to check if an event failed and take appropriate actions. More...
StatusCode | updateStates (int si=-1, const std::string &algo_name=std::string())
| Loop on algorithm in the slots and promote them to successive states (-1 means all slots, while empty string means skipping an update of the Control Flow state) More...
StatusCode | promoteToScheduled (unsigned int iAlgo, int si)
| Algorithm promotion. More...
StatusCode | promoteToAsyncScheduled (unsigned int iAlgo, int si)
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)
StatusCode | isStalled (int si)
| Check if the scheduling is in a stall. More...
void | dumpSchedulerState (int iSlot)
| Dump the state of the scheduler. More...
StatusCode | m_drain ()
| Drain the actions present in the queue. More...
void | dumpState (std::ostringstream &)

## 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
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 algoritms presently in flight. More...
unsigned int | m_IOBoundAlgosInFlight = 0
| Number of algoritms presently in flight. More...
bool | m_updateNeeded = true
| Keep track of update actions scheduled. 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...
SmartIF< IThreadPoolSvc > | m_threadPoolSvc
size_t | m_maxEventsInFlight {0}
size_t | m_maxAlgosInFlight {1}
bool | m_first = true

## Static Private Attributes
static std::list< SchedulerState > | m_sState
static std::mutex | m_ssMut

## 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
typedef Gaudi::PluginService::Factory< IService *, const std::string &, ISvcLocator * > | Factory
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 ()
| 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, "[[deprecated]] 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 102 of file AvalancheSchedulerSvc.h.

## Member Enumeration Documentation

enum AvalancheSchedulerSvc::ActivationState
| private

Enumerator
INACTIVE |
ACTIVE |
FAILURE |

Definition at line 133 of file AvalancheSchedulerSvc.h.

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

## Constructor & Destructor Documentation

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

Destructor.

## Member Function Documentation

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

364 {
365
366 if ( msgLevel( MSG::DEBUG ) ) debug() << "AvalancheSchedulerSvc::activate()" << endmsg;
367
368 if ( m_threadPoolSvc->initPool( m_threadPoolSize ).isFailure() ) {
369 error() << "problems initializing ThreadPoolSvc" << endmsg;
370 m_isActive = FAILURE;
371 return;
372 }
373
374 // Wait for actions pushed into the queue by finishing tasks.
375 action thisAction;
376 StatusCode sc( StatusCode::SUCCESS );
377
378 m_isActive = ACTIVE;
379
380 // Continue to wait if the scheduler is running or there is something to do
381 info() << "Start checking the actionsQueue" << endmsg;
382 while ( m_isActive == ACTIVE or m_actionsQueue.size() != 0 ) {
383 m_actionsQueue.pop( thisAction );
384 sc = thisAction();
385 if ( sc != StatusCode::SUCCESS )
386 verbose() << "Action did not succeed (which is not bad per se)." << endmsg;
387 else
388 verbose() << "Action succeeded." << endmsg;
389 }
390
391 info() << "Terminating thread-pool resources" << endmsg;
392 if ( m_threadPoolSvc->terminatePool().isFailure() ) {
393 error() << "Problems terminating thread pool" << endmsg;
394 m_isActive = FAILURE;
395 }
396 }
IThreadPoolSvc::initPool
virtual StatusCode initPool(const int &poolSize)=0
Initializes the thread pool.
StatusCode::SUCCESS
Definition: StatusCode.h:31
std::function< StatusCode()>
CommonMessagingBase::info
MsgStream & info() const
shortcut for the method msgStream(MSG::INFO)
Definition: CommonMessaging.h:112
CommonMessagingBase::verbose
MsgStream & verbose() const
shortcut for the method msgStream(MSG::VERBOSE)
Definition: CommonMessaging.h:118
MSG::DEBUG
Definition: IMessageSvc.h:16
AvalancheSchedulerSvc::ACTIVE
Definition: AvalancheSchedulerSvc.h:133
StatusCode::isFailure
bool isFailure() const
Test for a status code of FAILURE.
Definition: StatusCode.h:86
AvalancheSchedulerSvc::m_isActive
std::atomic< ActivationState > m_isActive
Flag to track if the scheduler is active or not.
Definition: AvalancheSchedulerSvc.h:177
IThreadPoolSvc::terminatePool
virtual StatusCode terminatePool()=0
Finalize the thread pool.
CommonMessagingBase::error
MsgStream & error() const
shortcut for the method msgStream(MSG::ERROR)
Definition: CommonMessaging.h:106
StatusCode
This class is used for returning status codes from appropriate routines.
Definition: StatusCode.h:28
AvalancheSchedulerSvc::m_threadPoolSize
Gaudi::Property< int > m_threadPoolSize
Definition: AvalancheSchedulerSvc.h:135
AvalancheSchedulerSvc::m_threadPoolSvc
SmartIF< IThreadPoolSvc > m_threadPoolSvc
Definition: AvalancheSchedulerSvc.h:283
CommonMessagingBase::debug
MsgStream & debug() const
shortcut for the method msgStream(MSG::DEBUG)
Definition: CommonMessaging.h:115
AvalancheSchedulerSvc::FAILURE
Definition: AvalancheSchedulerSvc.h:133
CommonMessagingBase::msgLevel
MSG::Level msgLevel() const
get the cached level (originally extracted from the embedded MsgStream)
Definition: CommonMessaging.h:124
endmsg
MsgStream & endmsg(MsgStream &s)
MsgStream Modifier: endmsg. Calls the output method of the MsgStream.
Definition: MsgStream.h:209
AvalancheSchedulerSvc::m_actionsQueue
tbb::concurrent_bounded_queue< action > m_actionsQueue
Queue where closures are stored and picked for execution.
Definition: AvalancheSchedulerSvc.h:260

void AvalancheSchedulerSvc::addAlg | ( | Algorithm * | a,
|  | EventContext * | e,
|  | pthread_t | t
| ) |  |

Definition at line 1093 of file AvalancheSchedulerSvc.cpp.

1094 {
1095
1096 std::lock_guard<std::mutex> lock( m_ssMut );
1097 m_sState.push_back( SchedulerState( a, e, t ) );
1098 }
AvalancheSchedulerSvc::m_sState
static std::list< SchedulerState > m_sState
Definition: AvalancheSchedulerSvc.h:317
std::lock
T lock(T...args)
AvalancheSchedulerSvc::m_ssMut
static std::mutex m_ssMut
Definition: AvalancheSchedulerSvc.h:318
std::lock_guard

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

Convert a name to an integer.

Definition at line 431 of file AvalancheSchedulerSvc.cpp.

432 {
433 unsigned int index = m_algname_index_map[algoname];
434 return index;
435 }
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:186

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

407 {
408
409 if ( m_isActive == ACTIVE ) {
410 // Drain the scheduler
411 m_actionsQueue.push( std::bind( &AvalancheSchedulerSvc::m_drain, this ) );
412 // This would be the last action
413 m_actionsQueue.push( [this]() -> StatusCode {
414 m_isActive = INACTIVE;
415 return StatusCode::SUCCESS;
416 } );
417 }
418
419 return StatusCode::SUCCESS;
420 }
StatusCode::SUCCESS
Definition: StatusCode.h:31
AvalancheSchedulerSvc::INACTIVE
Definition: AvalancheSchedulerSvc.h:133
AvalancheSchedulerSvc::ACTIVE
Definition: AvalancheSchedulerSvc.h:133
AvalancheSchedulerSvc::m_isActive
std::atomic< ActivationState > m_isActive
Flag to track if the scheduler is active or not.
Definition: AvalancheSchedulerSvc.h:177
StatusCode
This class is used for returning status codes from appropriate routines.
Definition: StatusCode.h:28
std::bind
T bind(T...args)
AvalancheSchedulerSvc::m_actionsQueue
tbb::concurrent_bounded_queue< action > m_actionsQueue
Queue where closures are stored and picked for execution.
Definition: AvalancheSchedulerSvc.h:260
AvalancheSchedulerSvc::m_drain
StatusCode m_drain()
Drain the actions present in the queue.
Definition: AvalancheSchedulerSvc.cpp:512

bool AvalancheSchedulerSvc::delAlg | ( | Algorithm * | a | ) |

Definition at line 1101 of file AvalancheSchedulerSvc.cpp.

1102 {
1103
1104 std::lock_guard<std::mutex> lock( m_ssMut );
1105
1106 for ( std::list<SchedulerState>::iterator itr = m_sState.begin(); itr != m_sState.end(); ++itr ) {
1107 if ( *itr == a ) {
1108 m_sState.erase( itr );
1109 return true;
1110 }
1111 }
1112
1113 error() << "could not find Alg " << a->name() << " in Scheduler!" << endmsg;
1114 return false;
1115 }
Algorithm::name
const std::string & name() const override
The identifying name of the algorithm object.
Definition: Algorithm.cpp:731
AvalancheSchedulerSvc::m_sState
static std::list< SchedulerState > m_sState
Definition: AvalancheSchedulerSvc.h:317
CommonMessagingBase::error
MsgStream & error() const
shortcut for the method msgStream(MSG::ERROR)
Definition: CommonMessaging.h:106
std::lock
T lock(T...args)
AvalancheSchedulerSvc::m_ssMut
static std::mutex m_ssMut
Definition: AvalancheSchedulerSvc.h:318
std::list
STL class.
std::lock_guard
endmsg
MsgStream & endmsg(MsgStream &s)
MsgStream Modifier: endmsg. Calls the output method of the MsgStream.
Definition: MsgStream.h:209

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.

The dependencies of each algo are printed and the missing ones specified.

Definition at line 809 of file AvalancheSchedulerSvc.cpp.

810 {
811
812 // To have just one big message
813 std::ostringstream outputMessageStream;
814
815 outputMessageStream << "============================== Execution Task State ============================="
816 << std::endl;
817 dumpState( outputMessageStream );
818
819 outputMessageStream << std::endl
820 << "============================== Scheduler State ================================="
821 << std::endl;
822
823 int slotCount = -1;
824 for ( auto& thisSlot : m_eventSlots ) {
825 slotCount++;
826 if ( thisSlot.complete ) continue;
827
828 // dump temporal and topological precedence analysis (if enabled in the PrecedenceSvc)
829 if ( msgLevel( MSG::DEBUG ) ) m_precSvc->dumpPrecedenceRules( thisSlot );
830
831 outputMessageStream << "----------- slot: " << thisSlot.eventContext->slot()
832 << " event: " << thisSlot.eventContext->evt() << " -----------" << std::endl;
833
834 if ( 0 > iSlot or iSlot == slotCount ) {
835 outputMessageStream << "Algorithms states:" << std::endl;
836
837 const DataObjIDColl& wbSlotContent( thisSlot.dataFlowMgr.content() );
838 for ( unsigned int algoIdx = 0; algoIdx < thisSlot.algsStates.size(); ++algoIdx ) {
839 outputMessageStream << " o " << index2algname( algoIdx ) << " ["
840 << AlgsExecutionStates::stateNames[thisSlot.algsStates[algoIdx]] << "] Data deps: ";
841 DataObjIDColl deps( thisSlot.dataFlowMgr.dataDependencies( algoIdx ) );
842 const int depsSize = deps.size();
843 if ( depsSize == 0 ) outputMessageStream << " none";
844
845 DataObjIDColl missing;
846 for ( auto d : deps ) {
847 outputMessageStream << d << " ";
848 if ( wbSlotContent.find( d ) == wbSlotContent.end() ) {
849 // outputMessageStream << "[missing] ";
850 missing.insert( d );
851 }
852 }
853
854 outputMessageStream << std::endl;
855 }
856
857 // Snapshot of the WhiteBoard
858 outputMessageStream << "\nWhiteboard contents: " << std::endl;
859 for ( auto& product : wbSlotContent ) outputMessageStream << " o " << product << std::endl;
860
861 // Snapshot of the ControlFlow
862 outputMessageStream << "\nControl Flow:" << std::endl;
863 outputMessageStream << m_precSvc->printState( thisSlot ) << std::endl;
864 }
865 }
866
867 outputMessageStream << "=================================== END ======================================" << std::endl;
868
869 info() << "Dumping Scheduler State " << std::endl << outputMessageStream.str() << endmsg;
870 }
AvalancheSchedulerSvc::m_precSvc
SmartIF< IPrecedenceSvc > m_precSvc
A shortcut to the Precedence Service.
Definition: AvalancheSchedulerSvc.h:195
CommonMessagingBase::info
MsgStream & info() const
shortcut for the method msgStream(MSG::INFO)
Definition: CommonMessaging.h:112
MSG::DEBUG
Definition: IMessageSvc.h:16
std::endl
T endl(T...args)
IPrecedenceSvc::printState
virtual const std::string printState(EventSlot &) const =0
IPrecedenceSvc::dumpPrecedenceRules
virtual void dumpPrecedenceRules(EventSlot &)=0
Dump precedence rules.
std::ostringstream
STL class.
std::ostringstream::str
T str(T...args)
std::unordered_set< DataObjID, DataObjID_Hasher >
std::unordered_set::insert
T insert(T...args)
std::unordered_set::size
T size(T...args)
AvalancheSchedulerSvc::index2algname
const std::string & index2algname(unsigned int index)
Convert an integer to a name.
Definition: AvalancheSchedulerSvc.cpp:427
AvalancheSchedulerSvc::m_eventSlots
std::vector< EventSlot > m_eventSlots
Vector of events slots.
Definition: AvalancheSchedulerSvc.h:204
AvalancheSchedulerSvc::dumpState
void dumpState() override
Definition: AvalancheSchedulerSvc.cpp:1129
CommonMessagingBase::msgLevel
MSG::Level msgLevel() const
get the cached level (originally extracted from the embedded MsgStream)
Definition: CommonMessaging.h:124
endmsg
MsgStream & endmsg(MsgStream &s)
MsgStream Modifier: endmsg. Calls the output method of the MsgStream.
Definition: MsgStream.h:209
AlgsExecutionStates::stateNames
static std::map< State, std::string > stateNames
Definition: AlgsExecutionStates.h:42

void AvalancheSchedulerSvc::dumpState | ( |  | ) |
| override

Definition at line 1129 of file AvalancheSchedulerSvc.cpp.

1130 {
1131
1132 std::lock_guard<std::mutex> lock( m_ssMut );
1133
1134 std::ostringstream ost;
1135 ost << "dumping Executing Threads: [" << m_sState.size() << "]" << std::endl;
1136 dumpState( ost );
1137
1138 info() << ost.str() << endmsg;
1139 }
CommonMessagingBase::info
MsgStream & info() const
shortcut for the method msgStream(MSG::INFO)
Definition: CommonMessaging.h:112
std::endl
T endl(T...args)
AvalancheSchedulerSvc::m_sState
static std::list< SchedulerState > m_sState
Definition: AvalancheSchedulerSvc.h:317
std::ostringstream
STL class.
std::lock
T lock(T...args)
std::ostringstream::str
T str(T...args)
AvalancheSchedulerSvc::m_ssMut
static std::mutex m_ssMut
Definition: AvalancheSchedulerSvc.h:318
std::lock_guard
AvalancheSchedulerSvc::dumpState
void dumpState() override
Definition: AvalancheSchedulerSvc.cpp:1129
endmsg
MsgStream & endmsg(MsgStream &s)
MsgStream Modifier: endmsg. Calls the output method of the MsgStream.
Definition: MsgStream.h:209

void AvalancheSchedulerSvc::dumpState | ( | std::ostringstream & | ost | ) |
| private

Definition at line 1118 of file AvalancheSchedulerSvc.cpp.

1119 {
1120
1121 std::lock_guard<std::mutex> lock( m_ssMut );
1122
1123 for ( auto it : m_sState ) {
1124 ost << " " << it << std::endl;
1125 }
1126 }
std::endl
T endl(T...args)
AvalancheSchedulerSvc::m_sState
static std::list< SchedulerState > m_sState
Definition: AvalancheSchedulerSvc.h:317
std::lock
T lock(T...args)
AvalancheSchedulerSvc::m_ssMut
static std::mutex m_ssMut
Definition: AvalancheSchedulerSvc.h:318
std::lock_guard

StatusCode AvalancheSchedulerSvc::eventFailed | ( | EventContext * | eventContext | ) |
| private

Method to check if an event failed and take appropriate actions.

It can be possible that an event fails.

In this case this method is called. It dumps the state of the scheduler, drains the actions (without executing them) and events in the queues and returns a failure.

Definition at line 569 of file AvalancheSchedulerSvc.cpp.

570 {
571
572 // Set the number of slots available to an error code
573 m_freeSlots.store( 0 );
574
575 fatal() << "*** Event " << eventContext->evt() << " on slot " << eventContext->slot() << " failed! ***" << endmsg;
576
577 std::ostringstream ost;
578 m_algExecStateSvc->dump( ost, *eventContext );
579
580 info() << "Dumping Alg Exec State for slot " << eventContext->slot() << ":\n" << ost.str() << endmsg;
581
582 dumpSchedulerState( -1 );
583
584 // Empty queue and deactivate the service
585 action thisAction;
586 while ( m_actionsQueue.try_pop( thisAction ) ) {
587 };
588 deactivate();
589
590 // Push into the finished events queue the failed context
591 EventContext* thisEvtContext;
592 while ( m_finishedEvents.try_pop( thisEvtContext ) ) {
593 m_finishedEvents.push( thisEvtContext );
594 };
595 m_finishedEvents.push( eventContext );
596
597 return StatusCode::FAILURE;
598 }
StatusCode::FAILURE
Definition: StatusCode.h:31
std::function< StatusCode()>
EventContext::slot
ContextID_t slot() const
Definition: EventContext.h:40
CommonMessagingBase::info
MsgStream & info() const
shortcut for the method msgStream(MSG::INFO)
Definition: CommonMessaging.h:112
AvalancheSchedulerSvc::dumpSchedulerState
void dumpSchedulerState(int iSlot)
Dump the state of the scheduler.
Definition: AvalancheSchedulerSvc.cpp:809
IAlgExecStateSvc::dump
virtual void dump(std::ostringstream &ost, const EventContext &ctx) const =0
EventContext
This class represents an entry point to all the event specific data.
Definition: EventContext.h:24
EventContext::evt
ContextEvt_t evt() const
Definition: EventContext.h:39
AvalancheSchedulerSvc::m_freeSlots
std::atomic_int m_freeSlots
Atomic to account for asyncronous updates by the scheduler wrt the rest.
Definition: AvalancheSchedulerSvc.h:207
std::ostringstream
STL class.
AvalancheSchedulerSvc::m_finishedEvents
tbb::concurrent_bounded_queue< EventContext * > m_finishedEvents
Queue of finished events.
Definition: AvalancheSchedulerSvc.h:210
std::ostringstream::str
T str(T...args)
AvalancheSchedulerSvc::deactivate
StatusCode deactivate()
Deactivate scheduler.
Definition: AvalancheSchedulerSvc.cpp:406
CommonMessagingBase::fatal
MsgStream & fatal() const
shortcut for the method msgStream(MSG::FATAL)
Definition: CommonMessaging.h:100
AvalancheSchedulerSvc::m_algExecStateSvc
SmartIF< IAlgExecStateSvc > m_algExecStateSvc
Algorithm execution state manager.
Definition: AvalancheSchedulerSvc.h:216
endmsg
MsgStream & endmsg(MsgStream &s)
MsgStream Modifier: endmsg. Calls the output method of the MsgStream.
Definition: MsgStream.h:209
AvalancheSchedulerSvc::m_actionsQueue
tbb::concurrent_bounded_queue< action > m_actionsQueue
Queue where closures are stored and picked for execution.
Definition: AvalancheSchedulerSvc.h:260

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

Finalise.

Here the scheduler is deactivated and the thread joined.

Definition at line 332 of file AvalancheSchedulerSvc.cpp.

333 {
334
335 StatusCode sc( Service::finalize() );
336 if ( !sc.isSuccess() ) warning() << "Base class could not be finalized" << endmsg;
337
338 sc = deactivate();
339 if ( !sc.isSuccess() ) warning() << "Scheduler could not be deactivated" << endmsg;
340
341 info() << "Joining Scheduler thread" << endmsg;
342 m_thread.join();
343
344 // Final error check after thread pool termination
345 if ( m_isActive == FAILURE ) {
346 error() << "problems in scheduler thread" << endmsg;
347 return StatusCode::FAILURE;
348 }
349
350 return sc;
351 }
StatusCode::FAILURE
Definition: StatusCode.h:31
Service::finalize
StatusCode finalize() override
Definition: Service.cpp:174
CommonMessagingBase::info
MsgStream & info() const
shortcut for the method msgStream(MSG::INFO)
Definition: CommonMessaging.h:112
AvalancheSchedulerSvc::m_isActive
std::atomic< ActivationState > m_isActive
Flag to track if the scheduler is active or not.
Definition: AvalancheSchedulerSvc.h:177
CommonMessagingBase::error
MsgStream & error() const
shortcut for the method msgStream(MSG::ERROR)
Definition: CommonMessaging.h:106
CommonMessagingBase::warning
MsgStream & warning() const
shortcut for the method msgStream(MSG::WARNING)
Definition: CommonMessaging.h:109
std::thread::join
T join(T...args)
StatusCode
This class is used for returning status codes from appropriate routines.
Definition: StatusCode.h:28
AvalancheSchedulerSvc::deactivate
StatusCode deactivate()
Deactivate scheduler.
Definition: AvalancheSchedulerSvc.cpp:406
AvalancheSchedulerSvc::FAILURE
Definition: AvalancheSchedulerSvc.h:133
endmsg
MsgStream & endmsg(MsgStream &s)
MsgStream Modifier: endmsg. Calls the output method of the MsgStream.
Definition: MsgStream.h:209
AvalancheSchedulerSvc::m_thread
std::thread m_thread
The thread in which the activate function runs.
Definition: AvalancheSchedulerSvc.h:180

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

Get free slots number.

Definition at line 506 of file AvalancheSchedulerSvc.cpp.

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

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

Convert an integer to a name.

Definition at line 427 of file AvalancheSchedulerSvc.cpp.

427 { 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:192

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

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

StatusCode AvalancheSchedulerSvc::isStalled | ( | int | iSlot | ) |
| private

Check if the scheduling is in a stall.

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

This is the case when no actions are present in the actionsQueue, no algorithm is in flight and no algorithm has all of its dependencies satisfied.

Definition at line 784 of file AvalancheSchedulerSvc.cpp.

785 {
786 // Get the slot
787 EventSlot& thisSlot = m_eventSlots[iSlot];
788
789 if ( m_actionsQueue.empty() && m_algosInFlight == 0 && m_IOBoundAlgosInFlight == 0 &&
790 ( !thisSlot.algsStates.algsPresent( AlgsExecutionStates::DATAREADY ) ) ) {
791
792 info() << "About to declare a stall" << endmsg;
793 fatal() << "*** Stall detected! ***\n" << endmsg;
794 dumpSchedulerState( iSlot );
795 // throw GaudiException ("Stall detected",name(),StatusCode::FAILURE);
796
797 return StatusCode::FAILURE;
798 }
799 return StatusCode::SUCCESS;
800 }
AlgsExecutionStates::algsPresent
bool algsPresent(State state) const
Definition: AlgsExecutionStates.h:53
StatusCode::SUCCESS
Definition: StatusCode.h:31
AvalancheSchedulerSvc::m_IOBoundAlgosInFlight
unsigned int m_IOBoundAlgosInFlight
Number of algoritms presently in flight.
Definition: AvalancheSchedulerSvc.h:227
StatusCode::FAILURE
Definition: StatusCode.h:31
CommonMessagingBase::info
MsgStream & info() const
shortcut for the method msgStream(MSG::INFO)
Definition: CommonMessaging.h:112
AvalancheSchedulerSvc::dumpSchedulerState
void dumpSchedulerState(int iSlot)
Dump the state of the scheduler.
Definition: AvalancheSchedulerSvc.cpp:809
EventSlot::algsStates
AlgsExecutionStates algsStates
Vector of algorithms states.
Definition: EventSlot.h:37
AvalancheSchedulerSvc::m_algosInFlight
unsigned int m_algosInFlight
Number of algoritms presently in flight.
Definition: AvalancheSchedulerSvc.h:224
AlgsExecutionStates::DATAREADY
Definition: AlgsExecutionStates.h:35
EventSlot
Class representing the event slot.
Definition: EventSlot.h:11
AvalancheSchedulerSvc::m_eventSlots
std::vector< EventSlot > m_eventSlots
Vector of events slots.
Definition: AvalancheSchedulerSvc.h:204
CommonMessagingBase::fatal
MsgStream & fatal() const
shortcut for the method msgStream(MSG::FATAL)
Definition: CommonMessaging.h:100
endmsg
MsgStream & endmsg(MsgStream &s)
MsgStream Modifier: endmsg. Calls the output method of the MsgStream.
Definition: MsgStream.h:209
AvalancheSchedulerSvc::m_actionsQueue
tbb::concurrent_bounded_queue< action > m_actionsQueue
Queue where closures are stored and picked for execution.
Definition: AvalancheSchedulerSvc.h:260

StatusCode AvalancheSchedulerSvc::m_drain | ( |  | ) |
| private

Drain the actions present in the queue.

Update the states for all slots until nothing is left to do.

Definition at line 512 of file AvalancheSchedulerSvc.cpp.

513 {
514
515 unsigned int slotNum = 0;
516 for ( auto& thisSlot : m_eventSlots ) {
517 if ( not thisSlot.algsStates.allAlgsExecuted() and not thisSlot.complete ) {
518 updateStates( slotNum );
519 }
520 slotNum++;
521 }
522 return StatusCode::SUCCESS;
523 }
StatusCode::SUCCESS
Definition: StatusCode.h:31
AvalancheSchedulerSvc::m_eventSlots
std::vector< EventSlot > m_eventSlots
Vector of events slots.
Definition: AvalancheSchedulerSvc.h:204
AvalancheSchedulerSvc::updateStates
StatusCode updateStates(int si=-1, const std::string &algo_name=std::string())
Loop on algorithm in the slots and promote them to successive states (-1 means all slots...
Definition: AvalancheSchedulerSvc.cpp:614

StatusCode AvalancheSchedulerSvc::popFinishedEvent | ( | EventContext *& | eventContext | ) |
| override

Blocks until an event is availble.

Get a finished event or block until one becomes available.

Definition at line 529 of file AvalancheSchedulerSvc.cpp.

530 {
531 // debug() << "popFinishedEvent: queue size: " << m_finishedEvents.size() << endmsg;
532 if ( m_freeSlots.load() == (int)m_maxEventsInFlight or m_isActive == INACTIVE ) {
533 // debug() << "freeslots: " << m_freeSlots << "/" << m_maxEventsInFlight
534 // << " active: " << m_isActive << endmsg;
535 return StatusCode::FAILURE;
536 } else {
537 // debug() << "freeslots: " << m_freeSlots << "/" << m_maxEventsInFlight
538 // << " active: " << m_isActive << endmsg;
539 m_finishedEvents.pop( eventContext );
540 m_freeSlots++;
541 if ( msgLevel( MSG::DEBUG ) )
542 debug() << "Popped slot " << eventContext->slot() << "(event " << eventContext->evt() << ")" << endmsg;
543 return StatusCode::SUCCESS;
544 }
545 }
StatusCode::SUCCESS
Definition: StatusCode.h:31
AvalancheSchedulerSvc::INACTIVE
Definition: AvalancheSchedulerSvc.h:133
StatusCode::FAILURE
Definition: StatusCode.h:31
EventContext::slot
ContextID_t slot() const
Definition: EventContext.h:40
MSG::DEBUG
Definition: IMessageSvc.h:16
AvalancheSchedulerSvc::m_isActive
std::atomic< ActivationState > m_isActive
Flag to track if the scheduler is active or not.
Definition: AvalancheSchedulerSvc.h:177
EventContext::evt
ContextEvt_t evt() const
Definition: EventContext.h:39
AvalancheSchedulerSvc::m_freeSlots
std::atomic_int m_freeSlots
Atomic to account for asyncronous updates by the scheduler wrt the rest.
Definition: AvalancheSchedulerSvc.h:207
AvalancheSchedulerSvc::m_finishedEvents
tbb::concurrent_bounded_queue< EventContext * > m_finishedEvents
Queue of finished events.
Definition: AvalancheSchedulerSvc.h:210
CommonMessagingBase::debug
MsgStream & debug() const
shortcut for the method msgStream(MSG::DEBUG)
Definition: CommonMessaging.h:115
CommonMessagingBase::msgLevel
MSG::Level msgLevel() const
get the cached level (originally extracted from the embedded MsgStream)
Definition: CommonMessaging.h:124
AvalancheSchedulerSvc::m_maxEventsInFlight
size_t m_maxEventsInFlight
Definition: AvalancheSchedulerSvc.h:284
endmsg
MsgStream & endmsg(MsgStream &s)
MsgStream Modifier: endmsg. Calls the output method of the MsgStream.
Definition: MsgStream.h:209

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

1042 {
1043 // Put back the instance
1044 Algorithm* castedAlgo = dynamic_cast<Algorithm*>( algo ); // DP: expose context getter in IAlgo?
1045 if ( !castedAlgo ) fatal() << "[Asynchronous] The casting did not succeed!" << endmsg;
1046 // EventContext* eventContext = castedAlgo->getContext();
1047
1048 // Check if the execution failed
1049 if ( m_algExecStateSvc->eventStatus( *eventContext ) != EventStatus::Success ) eventFailed( eventContext ).ignore();
1050
1051 StatusCode sc = m_algResourcePool->releaseAlgorithm( algo->name(), algo );
1052
1053 if ( !sc.isSuccess() ) {
1054 error() << "[Asynchronous] [Event " << eventContext->evt() << ", Slot " << eventContext->slot() << "] "
1055 << "Instance of algorithm " << algo->name() << " could not be properly put back." << endmsg;
1056 return StatusCode::FAILURE;
1057 }
1058
1059 m_IOBoundAlgosInFlight--;
1060
1061 EventSlot& thisSlot = m_eventSlots[si];
1062
1063 if ( msgLevel( MSG::DEBUG ) )
1064 debug() << "[Asynchronous] Algorithm " << algo->name() << " executed in slot " << si
1065 << ". Algorithms scheduled are " << m_IOBoundAlgosInFlight << endmsg;
1066
1067 // Schedule an update of the status of the algorithms
1068 auto updateAction = std::bind( &AvalancheSchedulerSvc::updateStates, this, -1, algo->name() );
1069 m_actionsQueue.push( updateAction );
1070 m_updateNeeded = false;
1071
1072 if ( msgLevel( MSG::DEBUG ) )
1073 debug() << "[Asynchronous] Trying to handle execution result of " << index2algname( iAlgo ) << " on slot " << si
1074 << endmsg;
1075 State state;
1076 if ( algo->filterPassed() ) {
1077 state = State::EVTACCEPTED;
1078 } else {
1079 state = State::EVTREJECTED;
1080 }
1081
1082 sc = thisSlot.algsStates.updateState( iAlgo, state );
1083
1084 if ( sc.isSuccess() )
1085 if ( msgLevel( MSG::VERBOSE ) )
1086 verbose() << "[Asynchronous] Promoting " << index2algname( iAlgo ) << " on slot " << si << " to "
1087 << AlgsExecutionStates::stateNames[state] << endmsg;
1088
1089 return sc;
1090 }
AvalancheSchedulerSvc::m_IOBoundAlgosInFlight
unsigned int m_IOBoundAlgosInFlight
Number of algoritms presently in flight.
Definition: AvalancheSchedulerSvc.h:227
StatusCode::FAILURE
Definition: StatusCode.h:31
compareRootHistos.state
state
Definition: compareRootHistos.py:424
EventContext::slot
ContextID_t slot() const
Definition: EventContext.h:40
EventSlot::algsStates
AlgsExecutionStates algsStates
Vector of algorithms states.
Definition: EventSlot.h:37
StatusCode::isSuccess
bool isSuccess() const
Test for a status code of SUCCESS.
Definition: StatusCode.h:75
CommonMessagingBase::verbose
MsgStream & verbose() const
shortcut for the method msgStream(MSG::VERBOSE)
Definition: CommonMessaging.h:118
MSG::DEBUG
Definition: IMessageSvc.h:16
IAlgorithm::filterPassed
virtual bool filterPassed() const =0
Did this algorithm pass or fail its filter criterion for the last event?
MSG::VERBOSE
Definition: IMessageSvc.h:16
EventContext::evt
ContextEvt_t evt() const
Definition: EventContext.h:39
CommonMessagingBase::error
MsgStream & error() const
shortcut for the method msgStream(MSG::ERROR)
Definition: CommonMessaging.h:106
AvalancheSchedulerSvc::m_algResourcePool
SmartIF< IAlgResourcePool > m_algResourcePool
Cache for the algorithm resource pool.
Definition: AvalancheSchedulerSvc.h:252
StatusCode
This class is used for returning status codes from appropriate routines.
Definition: StatusCode.h:28
EventStatus::Success
Definition: IAlgExecStateSvc.h:67
std::bind
T bind(T...args)
AvalancheSchedulerSvc::m_updateNeeded
bool m_updateNeeded
Keep track of update actions scheduled.
Definition: AvalancheSchedulerSvc.h:248
Algorithm
Base class from which all concrete algorithm classes should be derived.
Definition: Algorithm.h:79
CommonMessagingBase::debug
MsgStream & debug() const
shortcut for the method msgStream(MSG::DEBUG)
Definition: CommonMessaging.h:115
AvalancheSchedulerSvc::eventFailed
StatusCode eventFailed(EventContext *eventContext)
Method to check if an event failed and take appropriate actions.
Definition: AvalancheSchedulerSvc.cpp:569
IAlgExecStateSvc::eventStatus
virtual const EventStatus::Status & eventStatus(const EventContext &ctx) const =0
AvalancheSchedulerSvc::index2algname
const std::string & index2algname(unsigned int index)
Convert an integer to a name.
Definition: AvalancheSchedulerSvc.cpp:427
EventSlot
Class representing the event slot.
Definition: EventSlot.h:11
AvalancheSchedulerSvc::m_eventSlots
std::vector< EventSlot > m_eventSlots
Vector of events slots.
Definition: AvalancheSchedulerSvc.h:204
StatusCode::ignore
void ignore() const
Definition: StatusCode.h:109
CommonMessagingBase::fatal
MsgStream & fatal() const
shortcut for the method msgStream(MSG::FATAL)
Definition: CommonMessaging.h:100
AvalancheSchedulerSvc::updateStates
StatusCode updateStates(int si=-1, const std::string &algo_name=std::string())
Loop on algorithm in the slots and promote them to successive states (-1 means all slots...
Definition: AvalancheSchedulerSvc.cpp:614
CommonMessagingBase::msgLevel
MSG::Level msgLevel() const
get the cached level (originally extracted from the embedded MsgStream)
Definition: CommonMessaging.h:124
AlgsExecutionStates::State
State
Execution states of the algorithms.
Definition: AlgsExecutionStates.h:32
IAlgResourcePool::releaseAlgorithm
virtual StatusCode releaseAlgorithm(const std::string &name, IAlgorithm *&algo)=0
Release a certain algorithm.
AvalancheSchedulerSvc::m_algExecStateSvc
SmartIF< IAlgExecStateSvc > m_algExecStateSvc
Algorithm execution state manager.
Definition: AvalancheSchedulerSvc.h:216
endmsg
MsgStream & endmsg(MsgStream &s)
MsgStream Modifier: endmsg. Calls the output method of the MsgStream.
Definition: MsgStream.h:209
AvalancheSchedulerSvc::m_actionsQueue
tbb::concurrent_bounded_queue< action > m_actionsQueue
Queue where closures are stored and picked for execution.
Definition: AvalancheSchedulerSvc.h:260
AlgsExecutionStates::stateNames
static std::map< State, std::string > stateNames
Definition: AlgsExecutionStates.h:42
AlgsExecutionStates::updateState
StatusCode updateState(unsigned int iAlgo, State newState)
Definition: AlgsExecutionStates.cpp:8

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

Definition at line 936 of file AvalancheSchedulerSvc.cpp.

937 {
938
939 if ( m_IOBoundAlgosInFlight == m_maxIOBoundAlgosInFlight ) return StatusCode::FAILURE;
940
941 // bool IOBound = m_precSvc->isBlocking(algName);
942
943 const std::string& algName( index2algname( iAlgo ) );
944 IAlgorithm* ialgoPtr = nullptr;
945 StatusCode sc( m_algResourcePool->acquireAlgorithm( algName, ialgoPtr ) );
946
947 if ( sc.isSuccess() ) { // if we managed to get an algorithm instance try to schedule it
948 EventContext* eventContext( m_eventSlots[si].eventContext );
949 if ( !eventContext ) {
950 fatal() << "[Asynchronous] Event context for algorithm " << algName << " is a nullptr (slot " << si << ")"
951 << endmsg;
952 return StatusCode::FAILURE;
953 }
954
955 ++m_IOBoundAlgosInFlight;
956 // Can we use tbb-based overloaded new-operator for a "custom" task (an algorithm wrapper, not derived from
957 // tbb::task)? it seems it works..
958 IOBoundAlgTask* theTask = new ( tbb::task::allocate_root() )
959 IOBoundAlgTask( ialgoPtr, eventContext, serviceLocator(), m_algExecStateSvc );
960 m_IOBoundAlgScheduler->push( *theTask );
961
962 if ( msgLevel( MSG::DEBUG ) )
963 debug() << "[Asynchronous] Algorithm " << algName << " was submitted on event " << eventContext->evt()
964 << " in slot " << si << ". algorithms scheduled are " << m_IOBoundAlgosInFlight << endmsg;
965
966 StatusCode updateSc( m_eventSlots[si].algsStates.updateState( iAlgo, AlgsExecutionStates::SCHEDULED ) );
967
968 if ( updateSc.isSuccess() )
969 if ( msgLevel( MSG::VERBOSE ) )
970 verbose() << "[Asynchronous] Promoting " << index2algname( iAlgo ) << " to SCHEDULED on slot " << si << endmsg;
971 return updateSc;
972 } else {
973 if ( msgLevel( MSG::DEBUG ) )
974 debug() << "[Asynchronous] Could not acquire instance for algorithm " << index2algname( iAlgo ) << " on slot "
975 << si << endmsg;
976 return sc;
977 }
978 }
IOBoundAlgTask
Wrapper around I/O-bound Gaudi-algorithms.
Definition: IOBoundAlgTask.h:21
IAlgResourcePool::acquireAlgorithm
virtual StatusCode acquireAlgorithm(const std::string &name, IAlgorithm *&algo, bool blocking=false)=0
Acquire a certain algorithm using its name.
AvalancheSchedulerSvc::m_IOBoundAlgosInFlight
unsigned int m_IOBoundAlgosInFlight
Number of algoritms presently in flight.
Definition: AvalancheSchedulerSvc.h:227
StatusCode::FAILURE
Definition: StatusCode.h:31
CommonMessagingBase::verbose
MsgStream & verbose() const
shortcut for the method msgStream(MSG::VERBOSE)
Definition: CommonMessaging.h:118
MSG::DEBUG
Definition: IMessageSvc.h:16
MSG::VERBOSE
Definition: IMessageSvc.h:16
EventContext
This class represents an entry point to all the event specific data.
Definition: EventContext.h:24
AlgsExecutionStates::SCHEDULED
Definition: AlgsExecutionStates.h:36
std::string
STL class.
AvalancheSchedulerSvc::m_algResourcePool
SmartIF< IAlgResourcePool > m_algResourcePool
Cache for the algorithm resource pool.
Definition: AvalancheSchedulerSvc.h:252
StatusCode
This class is used for returning status codes from appropriate routines.
Definition: StatusCode.h:28
AvalancheSchedulerSvc::m_IOBoundAlgScheduler
SmartIF< IAccelerator > m_IOBoundAlgScheduler
A shortcut to IO-bound algorithm scheduler.
Definition: AvalancheSchedulerSvc.h:201
IAlgorithm
The IAlgorithm is the interface implemented by the Algorithm base class.
Definition: IAlgorithm.h:28
CommonMessagingBase::debug
MsgStream & debug() const
shortcut for the method msgStream(MSG::DEBUG)
Definition: CommonMessaging.h:115
AvalancheSchedulerSvc::index2algname
const std::string & index2algname(unsigned int index)
Convert an integer to a name.
Definition: AvalancheSchedulerSvc.cpp:427
AvalancheSchedulerSvc::m_eventSlots
std::vector< EventSlot > m_eventSlots
Vector of events slots.
Definition: AvalancheSchedulerSvc.h:204
IAccelerator::push
virtual StatusCode push(IAlgTask &task)=0
CommonMessagingBase::fatal
MsgStream & fatal() const
shortcut for the method msgStream(MSG::FATAL)
Definition: CommonMessaging.h:100
CommonMessagingBase::msgLevel
MSG::Level msgLevel() const
get the cached level (originally extracted from the embedded MsgStream)
Definition: CommonMessaging.h:124
Service::serviceLocator
SmartIF< ISvcLocator > & serviceLocator() const override
Retrieve pointer to service locator.
Definition: Service.cpp:292
AvalancheSchedulerSvc::m_algExecStateSvc
SmartIF< IAlgExecStateSvc > m_algExecStateSvc
Algorithm execution state manager.
Definition: AvalancheSchedulerSvc.h:216
AvalancheSchedulerSvc::m_maxIOBoundAlgosInFlight
Gaudi::Property< unsigned int > m_maxIOBoundAlgosInFlight
Definition: AvalancheSchedulerSvc.h:140
endmsg
MsgStream & endmsg(MsgStream &s)
MsgStream Modifier: endmsg. Calls the output method of the MsgStream.
Definition: MsgStream.h:209

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

986 {
987 // Put back the instance
988 Algorithm* castedAlgo = dynamic_cast<Algorithm*>( algo ); // DP: expose context getter in IAlgo?
989 if ( !castedAlgo ) fatal() << "The casting did not succeed!" << endmsg;
990 // EventContext* eventContext = castedAlgo->getContext();
991
992 // Check if the execution failed
993 if ( m_algExecStateSvc->eventStatus( *eventContext ) != EventStatus::Success ) eventFailed( eventContext ).ignore();
994
995 Gaudi::Hive::setCurrentContext( eventContext );
996 StatusCode sc = m_algResourcePool->releaseAlgorithm( algo->name(), algo );
997
998 if ( !sc.isSuccess() ) {
999 error() << "[Event " << eventContext->evt() << ", Slot " << eventContext->slot() << "] "
1000 << "Instance of algorithm " << algo->name() << " could not be properly put back." << endmsg;
1001 return StatusCode::FAILURE;
1002 }
1003
1004 m_algosInFlight--;
1005
1006 EventSlot& thisSlot = m_eventSlots[si];
1007
1008 if ( msgLevel( MSG::DEBUG ) )
1009 debug() << "Algorithm " << algo->name() << " executed in slot " << si << ". Algorithms scheduled are "
1010 << m_algosInFlight << endmsg;
1011
1012 // Schedule an update of the status of the algorithms
1013 auto updateAction = std::bind( &AvalancheSchedulerSvc::updateStates, this, -1, algo->name() );
1014 m_actionsQueue.push( updateAction );
1015 m_updateNeeded = false;
1016
1017 if ( msgLevel( MSG::DEBUG ) )
1018 debug() << "Trying to handle execution result of " << index2algname( iAlgo ) << " on slot " << si << endmsg;
1019 State state;
1020 if ( algo->filterPassed() ) {
1021 state = State::EVTACCEPTED;
1022 } else {
1023 state = State::EVTREJECTED;
1024 }
1025
1026 sc = thisSlot.algsStates.updateState( iAlgo, state );
1027
1028 if ( sc.isSuccess() )
1029 if ( msgLevel( MSG::VERBOSE ) )
1030 verbose() << "Promoting " << index2algname( iAlgo ) << " on slot " << si << " to "
1031 << AlgsExecutionStates::stateNames[state] << endmsg;
1032
1033 return sc;
1034 }
StatusCode::FAILURE
Definition: StatusCode.h:31
compareRootHistos.state
state
Definition: compareRootHistos.py:424
EventContext::slot
ContextID_t slot() const
Definition: EventContext.h:40
EventSlot::algsStates
AlgsExecutionStates algsStates
Vector of algorithms states.
Definition: EventSlot.h:37
StatusCode::isSuccess
bool isSuccess() const
Test for a status code of SUCCESS.
Definition: StatusCode.h:75
CommonMessagingBase::verbose
MsgStream & verbose() const
shortcut for the method msgStream(MSG::VERBOSE)
Definition: CommonMessaging.h:118
MSG::DEBUG
Definition: IMessageSvc.h:16
IAlgorithm::filterPassed
virtual bool filterPassed() const =0
Did this algorithm pass or fail its filter criterion for the last event?
MSG::VERBOSE
Definition: IMessageSvc.h:16
EventContext::evt
ContextEvt_t evt() const
Definition: EventContext.h:39
AvalancheSchedulerSvc::m_algosInFlight
unsigned int m_algosInFlight
Number of algoritms presently in flight.
Definition: AvalancheSchedulerSvc.h:224
CommonMessagingBase::error
MsgStream & error() const
shortcut for the method msgStream(MSG::ERROR)
Definition: CommonMessaging.h:106
AvalancheSchedulerSvc::m_algResourcePool
SmartIF< IAlgResourcePool > m_algResourcePool
Cache for the algorithm resource pool.
Definition: AvalancheSchedulerSvc.h:252
StatusCode
This class is used for returning status codes from appropriate routines.
Definition: StatusCode.h:28
EventStatus::Success
Definition: IAlgExecStateSvc.h:67
std::bind
T bind(T...args)
Gaudi::Hive::setCurrentContext
GAUDI_API void setCurrentContext(const EventContext *ctx)
Definition: ThreadLocalContext.cpp:35
AvalancheSchedulerSvc::m_updateNeeded
bool m_updateNeeded
Keep track of update actions scheduled.
Definition: AvalancheSchedulerSvc.h:248
Algorithm
Base class from which all concrete algorithm classes should be derived.
Definition: Algorithm.h:79
CommonMessagingBase::debug
MsgStream & debug() const
shortcut for the method msgStream(MSG::DEBUG)
Definition: CommonMessaging.h:115
AvalancheSchedulerSvc::eventFailed
StatusCode eventFailed(EventContext *eventContext)
Method to check if an event failed and take appropriate actions.
Definition: AvalancheSchedulerSvc.cpp:569
IAlgExecStateSvc::eventStatus
virtual const EventStatus::Status & eventStatus(const EventContext &ctx) const =0
AvalancheSchedulerSvc::index2algname
const std::string & index2algname(unsigned int index)
Convert an integer to a name.
Definition: AvalancheSchedulerSvc.cpp:427
EventSlot
Class representing the event slot.
Definition: EventSlot.h:11
AvalancheSchedulerSvc::m_eventSlots
std::vector< EventSlot > m_eventSlots
Vector of events slots.
Definition: AvalancheSchedulerSvc.h:204
StatusCode::ignore
void ignore() const
Definition: StatusCode.h:109
CommonMessagingBase::fatal
MsgStream & fatal() const
shortcut for the method msgStream(MSG::FATAL)
Definition: CommonMessaging.h:100
AvalancheSchedulerSvc::updateStates
StatusCode updateStates(int si=-1, const std::string &algo_name=std::string())
Loop on algorithm in the slots and promote them to successive states (-1 means all slots...
Definition: AvalancheSchedulerSvc.cpp:614
CommonMessagingBase::msgLevel
MSG::Level msgLevel() const
get the cached level (originally extracted from the embedded MsgStream)
Definition: CommonMessaging.h:124
AlgsExecutionStates::State
State
Execution states of the algorithms.
Definition: AlgsExecutionStates.h:32
IAlgResourcePool::releaseAlgorithm
virtual StatusCode releaseAlgorithm(const std::string &name, IAlgorithm *&algo)=0
Release a certain algorithm.
AvalancheSchedulerSvc::m_algExecStateSvc
SmartIF< IAlgExecStateSvc > m_algExecStateSvc
Algorithm execution state manager.
Definition: AvalancheSchedulerSvc.h:216
endmsg
MsgStream & endmsg(MsgStream &s)
MsgStream Modifier: endmsg. Calls the output method of the MsgStream.
Definition: MsgStream.h:209
AvalancheSchedulerSvc::m_actionsQueue
tbb::concurrent_bounded_queue< action > m_actionsQueue
Queue where closures are stored and picked for execution.
Definition: AvalancheSchedulerSvc.h:260
AlgsExecutionStates::stateNames
static std::map< State, std::string > stateNames
Definition: AlgsExecutionStates.h:42
AlgsExecutionStates::updateState
StatusCode updateState(unsigned int iAlgo, State newState)
Definition: AlgsExecutionStates.cpp:8

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

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

Algorithm promotion.

Definition at line 874 of file AvalancheSchedulerSvc.cpp.

875 {
876
877 if ( m_algosInFlight == m_maxAlgosInFlight ) return StatusCode::FAILURE;
878
879 const std::string& algName( index2algname( iAlgo ) );
880 IAlgorithm* ialgoPtr = nullptr;
881 StatusCode sc( m_algResourcePool->acquireAlgorithm( algName, ialgoPtr ) );
882
883 if ( sc.isSuccess() ) { // if we managed to get an algorithm instance try to schedule it
884 EventContext* eventContext( m_eventSlots[si].eventContext );
885 if ( !eventContext ) {
886 fatal() << "Event context for algorithm " << algName << " is a nullptr (slot " << si << ")" << endmsg;
887 return StatusCode::FAILURE;
888 }
889
890 ++m_algosInFlight;
891 auto promote2ExecutedClosure = std::bind( &AvalancheSchedulerSvc::promoteToExecuted, this, iAlgo,
892 eventContext->slot(), ialgoPtr, eventContext );
893 // Avoid to use tbb if the pool size is 1 and run in this thread
894 if ( -100 != m_threadPoolSize ) {
895
896 // this parent task is needed to promote an Algorithm as EXECUTED,
897 // it will be started as soon as the child task (see below) is completed
898 tbb::task* triggerAlgoStateUpdate =
899 new ( tbb::task::allocate_root() ) enqueueSchedulerActionTask( this, promote2ExecutedClosure );
900 // setting parent's refcount to 1 is made here only for consistency
901 // (in this case since it is not scheduled explicitly and there it has only one child task)
902 triggerAlgoStateUpdate->set_ref_count( 1 );
903 // the child task that executes an Algorithm
904 tbb::task* algoTask = new ( triggerAlgoStateUpdate->allocate_child() )
905 AlgoExecutionTask( ialgoPtr, eventContext, serviceLocator(), m_algExecStateSvc );
906 // schedule the algoTask
907 tbb::task::enqueue( *algoTask );
908
909 } else {
910 AlgoExecutionTask theTask( ialgoPtr, eventContext, serviceLocator(), m_algExecStateSvc );
911 theTask.execute();
912 promote2ExecutedClosure();
913 }
914
915 if ( msgLevel( MSG::DEBUG ) )
916 debug() << "Algorithm " << algName << " was submitted on event " << eventContext->evt() << " in slot " << si
917 << ". Algorithms scheduled are " << m_algosInFlight << endmsg;
918
919 StatusCode updateSc( m_eventSlots[si].algsStates.updateState( iAlgo, AlgsExecutionStates::SCHEDULED ) );
920
921 if ( msgLevel( MSG::VERBOSE ) ) dumpSchedulerState( -1 );
922
923 if ( updateSc.isSuccess() )
924 if ( msgLevel( MSG::VERBOSE ) )
925 verbose() << "Promoting " << index2algname( iAlgo ) << " to SCHEDULED on slot " << si << endmsg;
926 return updateSc;
927 } else {
928 if ( msgLevel( MSG::DEBUG ) )
929 debug() << "Could not acquire instance for algorithm " << index2algname( iAlgo ) << " on slot " << si << endmsg;
930 return sc;
931 }
932 }
IAlgResourcePool::acquireAlgorithm
virtual StatusCode acquireAlgorithm(const std::string &name, IAlgorithm *&algo, bool blocking=false)=0
Acquire a certain algorithm using its name.
StatusCode::FAILURE
Definition: StatusCode.h:31
AvalancheSchedulerSvc::dumpSchedulerState
void dumpSchedulerState(int iSlot)
Dump the state of the scheduler.
Definition: AvalancheSchedulerSvc.cpp:809
AlgoExecutionTask
Definition: AlgoExecutionTask.h:14
CommonMessagingBase::verbose
MsgStream & verbose() const
shortcut for the method msgStream(MSG::VERBOSE)
Definition: CommonMessaging.h:118
MSG::DEBUG
Definition: IMessageSvc.h:16
MSG::VERBOSE
Definition: IMessageSvc.h:16
EventContext
This class represents an entry point to all the event specific data.
Definition: EventContext.h:24
AlgsExecutionStates::SCHEDULED
Definition: AlgsExecutionStates.h:36
std::string
STL class.
AvalancheSchedulerSvc::m_algosInFlight
unsigned int m_algosInFlight
Number of algoritms presently in flight.
Definition: AvalancheSchedulerSvc.h:224
AvalancheSchedulerSvc::m_algResourcePool
SmartIF< IAlgResourcePool > m_algResourcePool
Cache for the algorithm resource pool.
Definition: AvalancheSchedulerSvc.h:252
StatusCode
This class is used for returning status codes from appropriate routines.
Definition: StatusCode.h:28
std::bind
T bind(T...args)
AvalancheSchedulerSvc::m_threadPoolSize
Gaudi::Property< int > m_threadPoolSize
Definition: AvalancheSchedulerSvc.h:135
IAlgorithm
The IAlgorithm is the interface implemented by the Algorithm base class.
Definition: IAlgorithm.h:28
CommonMessagingBase::debug
MsgStream & debug() const
shortcut for the method msgStream(MSG::DEBUG)
Definition: CommonMessaging.h:115
AvalancheSchedulerSvc::index2algname
const std::string & index2algname(unsigned int index)
Convert an integer to a name.
Definition: AvalancheSchedulerSvc.cpp:427
AvalancheSchedulerSvc::m_maxAlgosInFlight
size_t m_maxAlgosInFlight
Definition: AvalancheSchedulerSvc.h:285
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:984
AvalancheSchedulerSvc::m_eventSlots
std::vector< EventSlot > m_eventSlots
Vector of events slots.
Definition: AvalancheSchedulerSvc.h:204
CommonMessagingBase::fatal
MsgStream & fatal() const
shortcut for the method msgStream(MSG::FATAL)
Definition: CommonMessaging.h:100
CommonMessagingBase::msgLevel
MSG::Level msgLevel() const
get the cached level (originally extracted from the embedded MsgStream)
Definition: CommonMessaging.h:124
Service::serviceLocator
SmartIF< ISvcLocator > & serviceLocator() const override
Retrieve pointer to service locator.
Definition: Service.cpp:292
AvalancheSchedulerSvc::m_algExecStateSvc
SmartIF< IAlgExecStateSvc > m_algExecStateSvc
Algorithm execution state manager.
Definition: AvalancheSchedulerSvc.h:216
endmsg
MsgStream & endmsg(MsgStream &s)
MsgStream Modifier: endmsg. Calls the output method of the MsgStream.
Definition: MsgStream.h:209

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

446 {
447
448 if ( m_first ) {
449 m_first = false;
450 }
451
452 if ( !eventContext ) {
453 fatal() << "Event context is nullptr" << endmsg;
454 return StatusCode::FAILURE;
455 }
456
457 if ( m_freeSlots.load() == 0 ) {
458 if ( msgLevel( MSG::DEBUG ) ) debug() << "A free processing slot could not be found." << endmsg;
459 return StatusCode::FAILURE;
460 }
461
462 // no problem as push new event is only called from one thread (event loop manager)
463 m_freeSlots--;
464
465 auto action = [this, eventContext]() -> StatusCode {
466 // Event processing slot forced to be the same as the wb slot
467 const unsigned int thisSlotNum = eventContext->slot();
468 EventSlot& thisSlot = m_eventSlots[thisSlotNum];
469 if ( !thisSlot.complete ) {
470 fatal() << "The slot " << thisSlotNum << " is supposed to be a finished event but it's not" << endmsg;
471 return StatusCode::FAILURE;
472 }
473
474 debug() << "Executing event " << eventContext->evt() << " on slot " << thisSlotNum << endmsg;
475 thisSlot.reset( eventContext );
476
477 // promote to CR and DR the initial set of algorithms
478 Cause cs = {Cause::source::Root, "RootDecisionHub"};
479 m_precSvc->iterate( thisSlot, cs );
480
481 return this->updateStates( thisSlotNum );
482 }; // end of lambda
483
484 // Kick off the scheduling!
485 if ( msgLevel( MSG::VERBOSE ) ) {
486 verbose() << "Pushing the action to update the scheduler for slot " << eventContext->slot() << endmsg;
487 verbose() << "Free slots available " << m_freeSlots.load() << endmsg;
488 }
489 m_actionsQueue.push( action );
490
491 return StatusCode::SUCCESS;
492 }
StatusCode::SUCCESS
Definition: StatusCode.h:31
AvalancheSchedulerSvc::m_precSvc
SmartIF< IPrecedenceSvc > m_precSvc
A shortcut to the Precedence Service.
Definition: AvalancheSchedulerSvc.h:195
IPrecedenceSvc::iterate
virtual StatusCode iterate(EventSlot &, const Cause &)=0
Infer the precedence effect caused by an execution flow event.
StatusCode::FAILURE
Definition: StatusCode.h:31
std::function< StatusCode()>
EventContext::slot
ContextID_t slot() const
Definition: EventContext.h:40
CommonMessagingBase::verbose
MsgStream & verbose() const
shortcut for the method msgStream(MSG::VERBOSE)
Definition: CommonMessaging.h:118
MSG::DEBUG
Definition: IMessageSvc.h:16
MSG::VERBOSE
Definition: IMessageSvc.h:16
EventContext::evt
ContextEvt_t evt() const
Definition: EventContext.h:39
AvalancheSchedulerSvc::m_freeSlots
std::atomic_int m_freeSlots
Atomic to account for asyncronous updates by the scheduler wrt the rest.
Definition: AvalancheSchedulerSvc.h:207
Cause
Definition: PrecedenceRulesGraph.h:313
StatusCode
This class is used for returning status codes from appropriate routines.
Definition: StatusCode.h:28
Cause::source::Root
EventSlot::complete
bool complete
Flags completion of the event.
Definition: EventSlot.h:39
EventSlot::reset
void reset(EventContext *theeventContext)
Reset all resources in order to reuse the slot.
Definition: EventSlot.h:25
CommonMessagingBase::debug
MsgStream & debug() const
shortcut for the method msgStream(MSG::DEBUG)
Definition: CommonMessaging.h:115
EventSlot
Class representing the event slot.
Definition: EventSlot.h:11
AvalancheSchedulerSvc::m_eventSlots
std::vector< EventSlot > m_eventSlots
Vector of events slots.
Definition: AvalancheSchedulerSvc.h:204
AvalancheSchedulerSvc::m_first
bool m_first
Definition: AvalancheSchedulerSvc.h:286
CommonMessagingBase::fatal
MsgStream & fatal() const
shortcut for the method msgStream(MSG::FATAL)
Definition: CommonMessaging.h:100
AvalancheSchedulerSvc::updateStates
StatusCode updateStates(int si=-1, const std::string &algo_name=std::string())
Loop on algorithm in the slots and promote them to successive states (-1 means all slots...
Definition: AvalancheSchedulerSvc.cpp:614
CommonMessagingBase::msgLevel
MSG::Level msgLevel() const
get the cached level (originally extracted from the embedded MsgStream)
Definition: CommonMessaging.h:124
endmsg
MsgStream & endmsg(MsgStream &s)
MsgStream Modifier: endmsg. Calls the output method of the MsgStream.
Definition: MsgStream.h:209
AvalancheSchedulerSvc::m_actionsQueue
tbb::concurrent_bounded_queue< action > m_actionsQueue
Queue where closures are stored and picked for execution.
Definition: AvalancheSchedulerSvc.h:260

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

Definition at line 495 of file AvalancheSchedulerSvc.cpp.

496 {
497 StatusCode sc;
498 for ( auto context : eventContexts ) {
499 sc = pushNewEvent( context );
500 if ( sc != StatusCode::SUCCESS ) return sc;
501 }
502 return sc;
503 }
StatusCode::SUCCESS
Definition: StatusCode.h:31
StatusCode
This class is used for returning status codes from appropriate routines.
Definition: StatusCode.h:28
AvalancheSchedulerSvc::pushNewEvent
StatusCode pushNewEvent(EventContext *eventContext) override
Make an event available to the scheduler.
Definition: AvalancheSchedulerSvc.cpp:445

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

552 {
553 if ( m_finishedEvents.try_pop( eventContext ) ) {
554 if ( msgLevel( MSG::DEBUG ) )
555 debug() << "Try Pop successful slot " << eventContext->slot() << "(event " << eventContext->evt() << ")"
556 << endmsg;
557 m_freeSlots++;
558 return StatusCode::SUCCESS;
559 }
560 return StatusCode::FAILURE;
561 }
StatusCode::SUCCESS
Definition: StatusCode.h:31
StatusCode::FAILURE
Definition: StatusCode.h:31
EventContext::slot
ContextID_t slot() const
Definition: EventContext.h:40
MSG::DEBUG
Definition: IMessageSvc.h:16
EventContext::evt
ContextEvt_t evt() const
Definition: EventContext.h:39
AvalancheSchedulerSvc::m_freeSlots
std::atomic_int m_freeSlots
Atomic to account for asyncronous updates by the scheduler wrt the rest.
Definition: AvalancheSchedulerSvc.h:207
AvalancheSchedulerSvc::m_finishedEvents
tbb::concurrent_bounded_queue< EventContext * > m_finishedEvents
Queue of finished events.
Definition: AvalancheSchedulerSvc.h:210
CommonMessagingBase::debug
MsgStream & debug() const
shortcut for the method msgStream(MSG::DEBUG)
Definition: CommonMessaging.h:115
CommonMessagingBase::msgLevel
MSG::Level msgLevel() const
get the cached level (originally extracted from the embedded MsgStream)
Definition: CommonMessaging.h:124
endmsg
MsgStream & endmsg(MsgStream &s)
MsgStream Modifier: endmsg. Calls the output method of the MsgStream.
Definition: MsgStream.h:209

StatusCode AvalancheSchedulerSvc::updateStates | ( | int | si = `-1`,
|  | const std::string & | algo_name = `std::string()`
| ) |  |
| private

Loop on algorithm in the slots and promote them to successive states (-1 means all slots, while empty string 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 614 of file AvalancheSchedulerSvc.cpp.

615 {
616
617 m_updateNeeded = true;
618
619 StatusCode global_sc( StatusCode::FAILURE, true );
620
621 // Sort from the oldest to the newest event
622 // Prepare a vector of pointers to the slots to avoid copies
623 std::vector<EventSlot*> eventSlotsPtrs;
624
625 // Consider all slots if si <0 or just one otherwise
626 if ( si < 0 ) {
627 const int eventsSlotsSize( m_eventSlots.size() );
628 eventSlotsPtrs.reserve( eventsSlotsSize );
629 for ( auto slotIt = m_eventSlots.begin(); slotIt != m_eventSlots.end(); slotIt++ ) {
630 if ( !slotIt->complete ) eventSlotsPtrs.push_back( &( *slotIt ) );
631 }
632 std::sort( eventSlotsPtrs.begin(), eventSlotsPtrs.end(),
633 []( EventSlot* a, EventSlot* b ) { return a->eventContext->evt() < b->eventContext->evt(); } );
634 } else {
635 eventSlotsPtrs.push_back( &m_eventSlots[si] );
636 }
637
638 for ( EventSlot* thisSlotPtr : eventSlotsPtrs ) {
639 int iSlot = thisSlotPtr->eventContext->slot();
640
641 // Cache the states of the algos to improve readability and performance
642 auto& thisSlot = m_eventSlots[iSlot];
643 AlgsExecutionStates& thisAlgsStates = thisSlot.algsStates;
644
645 // Perform the I->CR->DR transitions
646 if ( !algo_name.empty() ) {
647 Cause cs = {Cause::source::Task, algo_name};
648 m_precSvc->iterate( thisSlot, cs );
649 }
650
651 StatusCode partial_sc( StatusCode::FAILURE, true );
652
653 // Perform DR->SCHEDULED
654 if ( !m_optimizationMode.empty() ) {
655 auto comp_nodes = [this]( const uint& i, const uint& j ) {
656 return ( m_precSvc->getPriority( index2algname( i ) ) < m_precSvc->getPriority( index2algname( j ) ) );
657 };
658 std::priority_queue<uint, std::vector<uint>, std::function<bool( const uint&, const uint& )>> buffer(
659 comp_nodes, std::vector<uint>() );
660 for ( auto it = thisAlgsStates.begin( AlgsExecutionStates::State::DATAREADY );
661 it != thisAlgsStates.end( AlgsExecutionStates::State::DATAREADY ); ++it )
662 buffer.push( *it );
663 /*std::stringstream s;
664 auto buffer2 = buffer;
665 while (!buffer2.empty()) {
666 s << m_precSvc->getPriority(index2algname(buffer2.top())) << ", ";
667 buffer2.pop();
668 }
669 info() << "DRBuffer is: [ " << s.str() << " ] <--" << algo_name << " executed" << endmsg;*/
670
671 /*while (!buffer.empty()) {
672 partial_sc = promoteToScheduled(buffer.top(), iSlot);
673 if (partial_sc.isFailure()) {
674 if (msgLevel(MSG::VERBOSE))
675 verbose() << "Could not apply transition from "
676 << AlgsExecutionStates::stateNames[AlgsExecutionStates::State::DATAREADY]
677 << " for algorithm " << index2algname(buffer.top()) << " on processing slot " << iSlot << endmsg;
678 if (m_useIOBoundAlgScheduler) {
679 partial_sc = promoteToAsyncScheduled(buffer.top(), iSlot);
680 if (msgLevel(MSG::VERBOSE))
681 if (partial_sc.isFailure())
682 verbose() << "[Asynchronous] Could not apply transition from "
683 << AlgsExecutionStates::stateNames[AlgsExecutionStates::State::DATAREADY]
684 << " for algorithm " << index2algname(buffer.top()) << " on processing slot " << iSlot <<
685 endmsg;
686 }
687 }
688 buffer.pop();
689 }*/
690 while ( !buffer.empty() ) {
691 bool IOBound = false;
692 if ( m_useIOBoundAlgScheduler ) IOBound = m_precSvc->isBlocking( index2algname( buffer.top() ) );
693
694 if ( !IOBound )
695 partial_sc = promoteToScheduled( buffer.top(), iSlot );
696 else
697 partial_sc = promoteToAsyncScheduled( buffer.top(), iSlot );
698
699 if ( msgLevel( MSG::VERBOSE ) )
700 if ( partial_sc.isFailure() )
701 verbose() << "Could not apply transition from "
702 << AlgsExecutionStates::stateNames[AlgsExecutionStates::State::DATAREADY] << " for algorithm "
703 << index2algname( buffer.top() ) << " on processing slot " << iSlot << endmsg;
704
705 buffer.pop();
706 }
707
708 } else {
709 for ( auto it = thisAlgsStates.begin( AlgsExecutionStates::State::DATAREADY );
710 it != thisAlgsStates.end( AlgsExecutionStates::State::DATAREADY ); ++it ) {
711 uint algIndex = *it;
712
713 bool IOBound = false;
714 if ( m_useIOBoundAlgScheduler ) IOBound = m_precSvc->isBlocking( index2algname( algIndex ) );
715
716 if ( !IOBound )
717 partial_sc = promoteToScheduled( algIndex, iSlot );
718 else
719 partial_sc = promoteToAsyncScheduled( algIndex, iSlot );
720
721 if ( msgLevel( MSG::VERBOSE ) )
722 if ( partial_sc.isFailure() )
723 verbose() << "Could not apply transition from "
724 << AlgsExecutionStates::stateNames[AlgsExecutionStates::State::DATAREADY] << " for algorithm "
725 << index2algname( algIndex ) << " on processing slot " << iSlot << endmsg;
726 }
727 }
728
729 if ( m_dumpIntraEventDynamics ) {
730 std::stringstream s;
731 s << algo_name << ", " << thisAlgsStates.sizeOfSubset( State::CONTROLREADY ) << ", "
732 << thisAlgsStates.sizeOfSubset( State::DATAREADY ) << ", " << thisAlgsStates.sizeOfSubset( State::SCHEDULED )
733 << ", " << std::chrono::high_resolution_clock::now().time_since_epoch().count() << "\n";
734 auto threads = ( m_threadPoolSize != -1 ) ? std::to_string( m_threadPoolSize )
735 : std::to_string( tbb::task_scheduler_init::default_num_threads() );
736 std::ofstream myfile;
737 myfile.open( "IntraEventConcurrencyDynamics_" + threads + "T.csv", std::ios::app );
738 myfile << s.str();
739 myfile.close();
740 }
741
742 // Not complete because this would mean that the slot is already free!
743 if ( !thisSlot.complete && m_precSvc->CFRulesResolved( thisSlot ) &&
744 !thisSlot.algsStates.algsPresent( AlgsExecutionStates::CONTROLREADY ) &&
745 !thisSlot.algsStates.algsPresent( AlgsExecutionStates::DATAREADY ) &&
746 !thisSlot.algsStates.algsPresent( AlgsExecutionStates::SCHEDULED ) ) {
747
748 thisSlot.complete = true;
749 // if the event did not fail, add it to the finished events
750 // otherwise it is taken care of in the error handling already
751 if ( m_algExecStateSvc->eventStatus( *thisSlot.eventContext ) == EventStatus::Success ) {
752 m_finishedEvents.push( thisSlot.eventContext );
753 if ( msgLevel( MSG::DEBUG ) )
754 debug() << "Event " << thisSlot.eventContext->evt() << " finished (slot " << thisSlot.eventContext->slot()
755 << ")." << endmsg;
756 }
757
758 // now let's return the fully evaluated result of the control flow
759 if ( msgLevel( MSG::DEBUG ) ) debug() << m_precSvc->printState( thisSlot ) << endmsg;
760
761 thisSlot.eventContext = nullptr;
762 } else {
763 StatusCode eventStalledSC = isStalled( iSlot );
764 if ( !eventStalledSC.isSuccess() ) {
765 m_algExecStateSvc->setEventStatus( EventStatus::AlgStall, *thisSlot.eventContext );
766 eventFailed( thisSlot.eventContext ).ignore();
767 }
768 }
769 } // end loop on slots
770
771 verbose() << "States Updated." << endmsg;
772
773 return global_sc;
774 }
Cause::source::Task
std::string::empty
T empty(T...args)
std::ofstream::open
T open(T...args)
AvalancheSchedulerSvc::m_precSvc
SmartIF< IPrecedenceSvc > m_precSvc
A shortcut to the Precedence Service.
Definition: AvalancheSchedulerSvc.h:195
IPrecedenceSvc::iterate
virtual StatusCode iterate(EventSlot &, const Cause &)=0
Infer the precedence effect caused by an execution flow event.
StatusCode::FAILURE
Definition: StatusCode.h:31
std::function
EventContext::slot
ContextID_t slot() const
Definition: EventContext.h:40
AvalancheSchedulerSvc::m_dumpIntraEventDynamics
Gaudi::Property< bool > m_dumpIntraEventDynamics
Definition: AvalancheSchedulerSvc.h:147
AvalancheSchedulerSvc::promoteToScheduled
StatusCode promoteToScheduled(unsigned int iAlgo, int si)
Algorithm promotion.
Definition: AvalancheSchedulerSvc.cpp:874
StatusCode::isSuccess
bool isSuccess() const
Test for a status code of SUCCESS.
Definition: StatusCode.h:75
EventSlot::eventContext
EventContext * eventContext
Cache for the eventContext.
Definition: EventSlot.h:32
AvalancheSchedulerSvc::isStalled
StatusCode isStalled(int si)
Check if the scheduling is in a stall.
Definition: AvalancheSchedulerSvc.cpp:784
std::to_string
T to_string(T...args)
CommonMessagingBase::verbose
MsgStream & verbose() const
shortcut for the method msgStream(MSG::VERBOSE)
Definition: CommonMessaging.h:118
MSG::DEBUG
Definition: IMessageSvc.h:16
std::vector::end
T end(T...args)
AvalancheSchedulerSvc::m_optimizationMode
Gaudi::Property< std::string > m_optimizationMode
Definition: AvalancheSchedulerSvc.h:145
AlgsExecutionStates::sizeOfSubset
size_t sizeOfSubset(State state) const
Definition: AlgsExecutionStates.h:69
AvalancheSchedulerSvc::promoteToAsyncScheduled
StatusCode promoteToAsyncScheduled(unsigned int iAlgo, int si)
Definition: AvalancheSchedulerSvc.cpp:936
MSG::VERBOSE
Definition: IMessageSvc.h:16
AlgsExecutionStates::SCHEDULED
Definition: AlgsExecutionStates.h:36
std::stringstream
STL class.
AvalancheSchedulerSimpleTest.threads
int threads
Definition: AvalancheSchedulerSimpleTest.py:41
IPrecedenceSvc::printState
virtual const std::string printState(EventSlot &) const =0
EventContext::evt
ContextEvt_t evt() const
Definition: EventContext.h:39
AvalancheSchedulerSvc::m_useIOBoundAlgScheduler
Gaudi::Property< bool > m_useIOBoundAlgScheduler
Definition: AvalancheSchedulerSvc.h:149
std::vector::push_back
T push_back(T...args)
std::ofstream
STL class.
AlgsExecutionStates
The AlgsExecutionStates encodes the state machine for the execution of algorithms within a single eve...
Definition: AlgsExecutionStates.h:28
Cause
Definition: PrecedenceRulesGraph.h:313
std::priority_queue
STL class.
StatusCode
This class is used for returning status codes from appropriate routines.
Definition: StatusCode.h:28
IPrecedenceSvc::getPriority
virtual uint getPriority(const std::string &) const =0
Get task priority.
AvalancheSchedulerSvc::m_finishedEvents
tbb::concurrent_bounded_queue< EventContext * > m_finishedEvents
Queue of finished events.
Definition: AvalancheSchedulerSvc.h:210
std::ofstream::close
T close(T...args)
AlgsExecutionStates::CONTROLREADY
Definition: AlgsExecutionStates.h:34
EventStatus::AlgStall
Definition: IAlgExecStateSvc.h:67
std::stringstream::str
T str(T...args)
EventStatus::Success
Definition: IAlgExecStateSvc.h:67
IAlgExecStateSvc::setEventStatus
virtual void setEventStatus(const EventStatus::Status &sc, const EventContext &ctx)=0
AvalancheSchedulerSvc::m_threadPoolSize
Gaudi::Property< int > m_threadPoolSize
Definition: AvalancheSchedulerSvc.h:135
IPrecedenceSvc::CFRulesResolved
virtual bool CFRulesResolved(EventSlot &) const =0
Check if control flow rules are resolved.
AvalancheSchedulerSvc::m_updateNeeded
bool m_updateNeeded
Keep track of update actions scheduled.
Definition: AvalancheSchedulerSvc.h:248
std::vector::size
T size(T...args)
std::vector
STL class.
CommonMessagingBase::debug
MsgStream & debug() const
shortcut for the method msgStream(MSG::DEBUG)
Definition: CommonMessaging.h:115
AvalancheSchedulerSvc::eventFailed
StatusCode eventFailed(EventContext *eventContext)
Method to check if an event failed and take appropriate actions.
Definition: AvalancheSchedulerSvc.cpp:569
std::vector::begin
T begin(T...args)
AlgsExecutionStates::begin
Iterator begin(State kind)
Definition: AlgsExecutionStates.h:125
std::chrono::high_resolution_clock::now
T now(T...args)
IAlgExecStateSvc::eventStatus
virtual const EventStatus::Status & eventStatus(const EventContext &ctx) const =0
AvalancheSchedulerSvc::index2algname
const std::string & index2algname(unsigned int index)
Convert an integer to a name.
Definition: AvalancheSchedulerSvc.cpp:427
AlgsExecutionStates::DATAREADY
Definition: AlgsExecutionStates.h:35
EventSlot
Class representing the event slot.
Definition: EventSlot.h:11
BrunelScenarioForwardScheduler.app
app
Definition: BrunelScenarioForwardScheduler.py:147
gaudirun.s
string s
Definition: gaudirun.py:253
AvalancheSchedulerSvc::m_eventSlots
std::vector< EventSlot > m_eventSlots
Vector of events slots.
Definition: AvalancheSchedulerSvc.h:204
std::sort
T sort(T...args)
StatusCode::ignore
void ignore() const
Definition: StatusCode.h:109
CommonMessagingBase::msgLevel
MSG::Level msgLevel() const
get the cached level (originally extracted from the embedded MsgStream)
Definition: CommonMessaging.h:124
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:216
endmsg
MsgStream & endmsg(MsgStream &s)
MsgStream Modifier: endmsg. Calls the output method of the MsgStream.
Definition: MsgStream.h:209
std::vector::reserve
T reserve(T...args)
AlgsExecutionStates::stateNames
static std::map< State, std::string > stateNames
Definition: AlgsExecutionStates.h:42
AlgsExecutionStates::end
Iterator end(State kind)
Definition: AlgsExecutionStates.h:127

## Member Data Documentation

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.

SmartIF<IAlgExecStateSvc> AvalancheSchedulerSvc::m_algExecStateSvc
| private

Algorithm execution state manager.

Definition at line 216 of file AvalancheSchedulerSvc.h.

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

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

Vector to bookkeep the information necessary to the index2name conversion.

Definition at line 192 of file AvalancheSchedulerSvc.h.

unsigned int AvalancheSchedulerSvc::m_algosInFlight = 0
| private

Number of algoritms presently in flight.

Definition at line 224 of file AvalancheSchedulerSvc.h.

SmartIF<IAlgResourcePool> AvalancheSchedulerSvc::m_algResourcePool
| private

Cache for the algorithm resource pool.

Definition at line 252 of file AvalancheSchedulerSvc.h.

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

Definition at line 152 of file AvalancheSchedulerSvc.h.

SmartIF<ICondSvc> AvalancheSchedulerSvc::m_condSvc
| private

A shortcut to service for Conditions handling.

Definition at line 219 of file AvalancheSchedulerSvc.h.

Gaudi::Property<bool> AvalancheSchedulerSvc::m_dumpIntraEventDynamics
| private

Initial value:
{this, "DumpIntraEventDynamics", false,
"Dump intra-event concurrency dynamics to csv file"}

Definition at line 147 of file AvalancheSchedulerSvc.h.

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

Definition at line 157 of file AvalancheSchedulerSvc.h.

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

Vector of events slots.

Definition at line 204 of file AvalancheSchedulerSvc.h.

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

Queue of finished events.

Definition at line 210 of file AvalancheSchedulerSvc.h.

bool AvalancheSchedulerSvc::m_first = true
| private

Definition at line 286 of file AvalancheSchedulerSvc.h.

std::atomic_int AvalancheSchedulerSvc::m_freeSlots
| private

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

Definition at line 207 of file AvalancheSchedulerSvc.h.

unsigned int AvalancheSchedulerSvc::m_IOBoundAlgosInFlight = 0
| private

Number of algoritms presently in flight.

Definition at line 227 of file AvalancheSchedulerSvc.h.

SmartIF<IAccelerator> AvalancheSchedulerSvc::m_IOBoundAlgScheduler
| private

A shortcut to IO-bound algorithm scheduler.

Definition at line 201 of file AvalancheSchedulerSvc.h.

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

Definition at line 139 of file AvalancheSchedulerSvc.h.

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

Flag to track if the scheduler is active or not.

Definition at line 177 of file AvalancheSchedulerSvc.h.

size_t AvalancheSchedulerSvc::m_maxAlgosInFlight {1}
| private

Definition at line 285 of file AvalancheSchedulerSvc.h.

size_t AvalancheSchedulerSvc::m_maxEventsInFlight {0}
| private

Definition at line 284 of file AvalancheSchedulerSvc.h.

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

Initial value:
{this, "MaxIOBoundAlgosInFlight", 0,
"Maximum number of simultaneous I/O-bound algorithms"}

Definition at line 140 of file AvalancheSchedulerSvc.h.

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

SmartIF<IPrecedenceSvc> AvalancheSchedulerSvc::m_precSvc
| private

A shortcut to the Precedence Service.

Definition at line 195 of file AvalancheSchedulerSvc.h.

Gaudi::Property<bool> AvalancheSchedulerSvc::m_showControlFlow
| private

Initial value:
{this, "ShowControlFlow", false,
"Show the configuration of all Algorithms and Sequences"}

Definition at line 165 of file AvalancheSchedulerSvc.h.

Gaudi::Property<bool> AvalancheSchedulerSvc::m_showDataDeps
| private

Initial value:
{this, "ShowDataDependencies", true,
"Show the INPUT and OUTPUT data dependencies of Algorithms"}

Definition at line 159 of file AvalancheSchedulerSvc.h.

Gaudi::Property<bool> AvalancheSchedulerSvc::m_showDataFlow
| private

Initial value:
{this, "ShowDataFlow", false,
"Show the configuration of DataFlow between Algorithms"}

Definition at line 162 of file AvalancheSchedulerSvc.h.

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

std::mutex AvalancheSchedulerSvc::m_ssMut
| static private

Definition at line 318 of file AvalancheSchedulerSvc.h.

std::list< AvalancheSchedulerSvc::SchedulerState > AvalancheSchedulerSvc::m_sState
| static private

Definition at line 317 of file AvalancheSchedulerSvc.h.

std::thread AvalancheSchedulerSvc::m_thread
| private

The thread in which the activate function runs.

Definition at line 180 of file AvalancheSchedulerSvc.h.

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

SmartIF<IThreadPoolSvc> AvalancheSchedulerSvc::m_threadPoolSvc
| private

Definition at line 283 of file AvalancheSchedulerSvc.h.

bool AvalancheSchedulerSvc::m_updateNeeded = true
| private

Keep track of update actions scheduled.

Definition at line 248 of file AvalancheSchedulerSvc.h.

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

Initial value:
{this, "DataLoaderAlg", "",
"Attribute unmet input dependencies to this DataLoader Algorithm"}

Definition at line 154 of file AvalancheSchedulerSvc.h.

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

SmartIF<IHiveWhiteBoard> AvalancheSchedulerSvc::m_whiteboard
| private

A shortcut to the whiteboard.

Definition at line 198 of file AvalancheSchedulerSvc.h.

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

Definition at line 138 of file AvalancheSchedulerSvc.h.

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

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