| The Gaudi Framework v28r2p1 (f1a77ff4)

- 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 | declarePrivateTool (ToolHandle< T > &handle, std::string toolTypeAndName="", bool createIf=true)
| Declare used Private tool. More...
template<class T >
StatusCode | declarePublicTool (ToolHandle< T > &handle, std::string toolTypeAndName="", bool createIf=true)
| Declare used Public 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...
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 output level 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_maxEventsInFlight
Gaudi::Property< int > | m_threadPoolSize
Gaudi::Property< std::string > | m_whiteboardSvcName
Gaudi::Property< std::string > | m_IOBoundAlgSchedulerSvcName
Gaudi::Property< unsigned int > | m_maxAlgosInFlight
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< std::vector< std::vector< std::string > > > | m_algosDependencies
Gaudi::Property< bool > | m_checkDeps {this, "CheckDependencies", false, "[[deprecated]]"}
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< 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...
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...
concurrency::ExecutionFlowManager | m_efManager
| Member to take care of the control flow. More...
SmartIF< IThreadPoolSvc > | m_threadPoolSvc
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 > >
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 100 of file AvalancheSchedulerSvc.h.

## Member Enumeration Documentation

enum AvalancheSchedulerSvc::ActivationState
| private

Enumerator
INACTIVE |
ACTIVE |
FAILURE |

Definition at line 131 of file AvalancheSchedulerSvc.h.

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

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

305 {
306
307 if (msgLevel(MSG::DEBUG))
308 debug() << "AvalancheSchedulerSvc::activate()" << endmsg;
309
310 if ( m_threadPoolSvc->initPool( m_threadPoolSize ).isFailure() ) {
311 error() << "problems initializing ThreadPoolSvc" << endmsg;
312 m_isActive = FAILURE;
313 return;
314 }
315
316 // Wait for actions pushed into the queue by finishing tasks.
317 action thisAction;
318 StatusCode sc( StatusCode::SUCCESS );
319
320 m_isActive = ACTIVE;
321
322 // Continue to wait if the scheduler is running or there is something to do
323 info() << "Start checking the actionsQueue" << endmsg;
324 while ( m_isActive == ACTIVE or m_actionsQueue.size() != 0 ) {
325 m_actionsQueue.pop( thisAction );
326 sc = thisAction();
327 if ( sc != StatusCode::SUCCESS )
328 verbose() << "Action did not succeed (which is not bad per se)." << endmsg;
329 else
330 verbose() << "Action succeeded." << endmsg;
331 }
332
333 info() << "Terminating thread-pool resources" << endmsg;
334 if ( m_threadPoolSvc->terminatePool().isFailure() ) {
335 error() << "Problems terminating thread pool" << endmsg;
336 m_isActive = FAILURE;
337 }
338 }
IThreadPoolSvc::initPool
virtual StatusCode initPool(const int &poolSize)=0
Initializes the thread pool.
StatusCode::SUCCESS
Definition: StatusCode.h:30
std::function< StatusCode()>
CommonMessagingBase::info
MsgStream & info() const
shortcut for the method msgStream(MSG::INFO)
Definition: CommonMessaging.h:123
CommonMessagingBase::verbose
MsgStream & verbose() const
shortcut for the method msgStream(MSG::VERBOSE)
Definition: CommonMessaging.h:129
MSG::DEBUG
Definition: IMessageSvc.h:18
AvalancheSchedulerSvc::ACTIVE
Definition: AvalancheSchedulerSvc.h:131
StatusCode::isFailure
bool isFailure() const
Test for a status code of FAILURE.
Definition: StatusCode.h:84
AvalancheSchedulerSvc::m_isActive
std::atomic< ActivationState > m_isActive
Flag to track if the scheduler is active or not.
Definition: AvalancheSchedulerSvc.h:166
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:117
StatusCode
This class is used for returning status codes from appropriate routines.
Definition: StatusCode.h:26
AvalancheSchedulerSvc::m_threadPoolSize
Gaudi::Property< int > m_threadPoolSize
Definition: AvalancheSchedulerSvc.h:135
AvalancheSchedulerSvc::m_threadPoolSvc
SmartIF< IThreadPoolSvc > m_threadPoolSvc
Definition: AvalancheSchedulerSvc.h:266
CommonMessagingBase::debug
MsgStream & debug() const
shortcut for the method msgStream(MSG::DEBUG)
Definition: CommonMessaging.h:126
AvalancheSchedulerSvc::FAILURE
Definition: AvalancheSchedulerSvc.h:131
CommonMessagingBase::msgLevel
MSG::Level msgLevel() const
get the output level from the embedded MsgStream
Definition: CommonMessaging.h:135
endmsg
MsgStream & endmsg(MsgStream &s)
MsgStream Modifier: endmsg. Calls the output method of the MsgStream.
Definition: MsgStream.h:244
AvalancheSchedulerSvc::m_actionsQueue
tbb::concurrent_bounded_queue< action > m_actionsQueue
Queue where closures are stored and picked for execution.
Definition: AvalancheSchedulerSvc.h:243

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

Definition at line 1045 of file AvalancheSchedulerSvc.cpp.

1046 {
1047
1048 std::lock_guard<std::mutex> lock( m_ssMut );
1049 m_sState.push_back( SchedulerState( a, e, t ) );
1050 }
AvalancheSchedulerSvc::m_sState
static std::list< SchedulerState > m_sState
Definition: AvalancheSchedulerSvc.h:299
std::lock
T lock(T...args)
AvalancheSchedulerSvc::m_ssMut
static std::mutex m_ssMut
Definition: AvalancheSchedulerSvc.h:300
std::lock_guard

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

Convert a name to an integer.

Definition at line 374 of file AvalancheSchedulerSvc.cpp.

374 {
375 unsigned int index = m_algname_index_map[algoname];
376 return index;
377 }
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:175

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

348 {
349
350 if ( m_isActive == ACTIVE ) {
351 // Drain the scheduler
352 m_actionsQueue.push( std::bind( &AvalancheSchedulerSvc::m_drain, this ) );
353 // This would be the last action
354 m_actionsQueue.push( [this]() -> StatusCode {
355 m_isActive = INACTIVE;
356 return StatusCode::SUCCESS;
357 } );
358 }
359
360 return StatusCode::SUCCESS;
361 }
StatusCode::SUCCESS
Definition: StatusCode.h:30
AvalancheSchedulerSvc::INACTIVE
Definition: AvalancheSchedulerSvc.h:131
AvalancheSchedulerSvc::ACTIVE
Definition: AvalancheSchedulerSvc.h:131
AvalancheSchedulerSvc::m_isActive
std::atomic< ActivationState > m_isActive
Flag to track if the scheduler is active or not.
Definition: AvalancheSchedulerSvc.h:166
StatusCode
This class is used for returning status codes from appropriate routines.
Definition: StatusCode.h:26
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:243
AvalancheSchedulerSvc::m_drain
StatusCode m_drain()
Drain the actions present in the queue.
Definition: AvalancheSchedulerSvc.cpp:454

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

Definition at line 1053 of file AvalancheSchedulerSvc.cpp.

1054 {
1055
1056 std::lock_guard<std::mutex> lock( m_ssMut );
1057
1058 for ( std::list<SchedulerState>::iterator itr = m_sState.begin(); itr != m_sState.end(); ++itr ) {
1059 if ( *itr == a ) {
1060 m_sState.erase( itr );
1061 return true;
1062 }
1063 }
1064
1065 error() << "could not find Alg " << a->name() << " in Scheduler!" << endmsg;
1066 return false;
1067 }
Algorithm::name
const std::string & name() const override
The identifying name of the algorithm object.
Definition: Algorithm.cpp:715
AvalancheSchedulerSvc::m_sState
static std::list< SchedulerState > m_sState
Definition: AvalancheSchedulerSvc.h:299
CommonMessagingBase::error
MsgStream & error() const
shortcut for the method msgStream(MSG::ERROR)
Definition: CommonMessaging.h:117
std::lock
T lock(T...args)
AvalancheSchedulerSvc::m_ssMut
static std::mutex m_ssMut
Definition: AvalancheSchedulerSvc.h:300
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:244

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

755 {
756
757 // To have just one big message
758 std::ostringstream outputMessageStream;
759
760 outputMessageStream << "============================== Execution Task State ============================="
761 << std::endl;
762 dumpState( outputMessageStream );
763
764 outputMessageStream << std::endl
765 << "============================== Scheduler State ================================="
766 << std::endl;
767
768 int slotCount = -1;
769 for ( auto thisSlot : m_eventSlots ) {
770 slotCount++;
771 if ( thisSlot.complete ) continue;
772
773 outputMessageStream << "----------- slot: " << thisSlot.eventContext->slot()
774 << " event: " << thisSlot.eventContext->evt() << " -----------" << std::endl;
775
776 if ( 0 > iSlot or iSlot == slotCount ) {
777 outputMessageStream << "Algorithms states:" << std::endl;
778
779 const DataObjIDColl& wbSlotContent( thisSlot.dataFlowMgr.content() );
780 for ( unsigned int algoIdx = 0; algoIdx < thisSlot.algsStates.size(); ++algoIdx ) {
781 outputMessageStream << " o " << index2algname( algoIdx ) << " ["
782 << AlgsExecutionStates::stateNames[thisSlot.algsStates[algoIdx]] << "] Data deps: ";
783 DataObjIDColl deps( thisSlot.dataFlowMgr.dataDependencies( algoIdx ) );
784 const int depsSize = deps.size();
785 if ( depsSize == 0 ) outputMessageStream << " none";
786
787 DataObjIDColl missing;
788 for ( auto d : deps ) {
789 outputMessageStream << d << " ";
790 if ( wbSlotContent.find( d ) == wbSlotContent.end() ) {
791 // outputMessageStream << "[missing] ";
792 missing.insert( d );
793 }
794 }
795
796 if ( !missing.empty() ) {
797 outputMessageStream << ". The following are missing: ";
798 for ( auto d : missing ) {
799 outputMessageStream << d << " ";
800 }
801 }
802
803 outputMessageStream << std::endl;
804 }
805
806 // Snapshot of the WhiteBoard
807 outputMessageStream << "\nWhiteboard contents: " << std::endl;
808 for ( auto& product : wbSlotContent ) outputMessageStream << " o " << product << std::endl;
809
810 // Snapshot of the ControlFlow
811 outputMessageStream << "\nControl Flow:" << std::endl;
812 std::stringstream cFlowStateStringStream;
813 m_efManager.printEventState( cFlowStateStringStream, thisSlot.algsStates, thisSlot.controlFlowState, 0 );
814
815 outputMessageStream << cFlowStateStringStream.str() << std::endl;
816 }
817 }
818
819 outputMessageStream << "=================================== END ======================================" << std::endl;
820
821 info() << "Dumping Scheduler State " << std::endl << outputMessageStream.str() << endmsg;
822 }
std::unordered_set::empty
T empty(T...args)
concurrency::ExecutionFlowManager::printEventState
void printEventState(std::stringstream &ss, AlgsExecutionStates &states, const std::vector< int > &node_decisions, const unsigned int &recursionLevel) const
Print the state of the control flow for a given event.
Definition: ExecutionFlowManager.h:60
CommonMessagingBase::info
MsgStream & info() const
shortcut for the method msgStream(MSG::INFO)
Definition: CommonMessaging.h:123
std::endl
T endl(T...args)
std::stringstream
STL class.
std::ostringstream
STL class.
std::stringstream::str
T str(T...args)
std::unordered_set< DataObjID, DataObjID_Hasher >
AvalancheSchedulerSvc::m_efManager
concurrency::ExecutionFlowManager m_efManager
Member to take care of the control flow.
Definition: AvalancheSchedulerSvc.h:263
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:368
AvalancheSchedulerSvc::m_eventSlots
std::vector< EventSlot > m_eventSlots
Vector of events slots.
Definition: AvalancheSchedulerSvc.h:190
AvalancheSchedulerSvc::dumpState
void dumpState() override
Definition: AvalancheSchedulerSvc.cpp:1081
endmsg
MsgStream & endmsg(MsgStream &s)
MsgStream Modifier: endmsg. Calls the output method of the MsgStream.
Definition: MsgStream.h:244
AlgsExecutionStates::stateNames
static std::map< State, std::string > stateNames
Definition: AlgsExecutionStates.h:42

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

Definition at line 1081 of file AvalancheSchedulerSvc.cpp.

1082 {
1083
1084 std::lock_guard<std::mutex> lock( m_ssMut );
1085
1086 std::ostringstream ost;
1087 ost << "dumping Executing Threads: [" << m_sState.size() << "]" << std::endl;
1088 dumpState( ost );
1089
1090 info() << ost.str() << endmsg;
1091 }
CommonMessagingBase::info
MsgStream & info() const
shortcut for the method msgStream(MSG::INFO)
Definition: CommonMessaging.h:123
std::endl
T endl(T...args)
AvalancheSchedulerSvc::m_sState
static std::list< SchedulerState > m_sState
Definition: AvalancheSchedulerSvc.h:299
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:300
std::lock_guard
AvalancheSchedulerSvc::dumpState
void dumpState() override
Definition: AvalancheSchedulerSvc.cpp:1081
endmsg
MsgStream & endmsg(MsgStream &s)
MsgStream Modifier: endmsg. Calls the output method of the MsgStream.
Definition: MsgStream.h:244

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

Definition at line 1070 of file AvalancheSchedulerSvc.cpp.

1071 {
1072
1073 std::lock_guard<std::mutex> lock( m_ssMut );
1074
1075 for ( auto it : m_sState ) {
1076 ost << " " << it << std::endl;
1077 }
1078 }
std::endl
T endl(T...args)
AvalancheSchedulerSvc::m_sState
static std::list< SchedulerState > m_sState
Definition: AvalancheSchedulerSvc.h:299
std::lock
T lock(T...args)
AvalancheSchedulerSvc::m_ssMut
static std::mutex m_ssMut
Definition: AvalancheSchedulerSvc.h:300
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 509 of file AvalancheSchedulerSvc.cpp.

509 {
510
511 // Set the number of slots available to an error code
512 m_freeSlots.store( 0 );
513
514 fatal() << "*** Event " << eventContext->evt() << " on slot "
515 << eventContext->slot() << " failed! ***" << endmsg;
516
517 std::ostringstream ost;
518 m_algExecStateSvc->dump(ost, *eventContext);
519
520 info() << "Dumping Alg Exec State for slot " << eventContext->slot()
521 << ":\n" << ost.str() << endmsg;
522
523 dumpSchedulerState(-1);
524
525 // Empty queue and deactivate the service
526 action thisAction;
527 while ( m_actionsQueue.try_pop( thisAction ) ) {
528 };
529 deactivate();
530
531 // Push into the finished events queue the failed context
532 EventContext* thisEvtContext;
533 while ( m_finishedEvents.try_pop( thisEvtContext ) ) {
534 m_finishedEvents.push( thisEvtContext );
535 };
536 m_finishedEvents.push( eventContext );
537
538 return StatusCode::FAILURE;
539 }
StatusCode::FAILURE
Definition: StatusCode.h:29
std::function< StatusCode()>
EventContext::slot
ContextID_t slot() const
Definition: EventContext.h:41
CommonMessagingBase::info
MsgStream & info() const
shortcut for the method msgStream(MSG::INFO)
Definition: CommonMessaging.h:123
AvalancheSchedulerSvc::dumpSchedulerState
void dumpSchedulerState(int iSlot)
Dump the state of the scheduler.
Definition: AvalancheSchedulerSvc.cpp:755
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:25
EventContext::evt
ContextEvt_t evt() const
Definition: EventContext.h:40
AvalancheSchedulerSvc::m_freeSlots
std::atomic_int m_freeSlots
Atomic to account for asyncronous updates by the scheduler wrt the rest.
Definition: AvalancheSchedulerSvc.h:193
std::ostringstream
STL class.
AvalancheSchedulerSvc::m_finishedEvents
tbb::concurrent_bounded_queue< EventContext * > m_finishedEvents
Queue of finished events.
Definition: AvalancheSchedulerSvc.h:196
std::ostringstream::str
T str(T...args)
AvalancheSchedulerSvc::deactivate
StatusCode deactivate()
Deactivate scheduler.
Definition: AvalancheSchedulerSvc.cpp:348
CommonMessagingBase::fatal
MsgStream & fatal() const
shortcut for the method msgStream(MSG::FATAL)
Definition: CommonMessaging.h:111
AvalancheSchedulerSvc::m_algExecStateSvc
SmartIF< IAlgExecStateSvc > m_algExecStateSvc
Algorithm execution state manager.
Definition: AvalancheSchedulerSvc.h:202
endmsg
MsgStream & endmsg(MsgStream &s)
MsgStream Modifier: endmsg. Calls the output method of the MsgStream.
Definition: MsgStream.h:244
AvalancheSchedulerSvc::m_actionsQueue
tbb::concurrent_bounded_queue< action > m_actionsQueue
Queue where closures are stored and picked for execution.
Definition: AvalancheSchedulerSvc.h:243

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

Finalise.

Here the scheduler is deactivated and the thread joined.

Definition at line 273 of file AvalancheSchedulerSvc.cpp.

273 {
274
275 StatusCode sc( Service::finalize() );
276 if ( !sc.isSuccess() ) warning() << "Base class could not be finalized" << endmsg;
277
278 sc = deactivate();
279 if ( !sc.isSuccess() ) warning() << "Scheduler could not be deactivated" << endmsg;
280
281 info() << "Joining Scheduler thread" << endmsg;
282 m_thread.join();
283
284 // Final error check after thread pool termination
285 if ( m_isActive == FAILURE ) {
286 error() << "problems in scheduler thread" << endmsg;
287 return StatusCode::FAILURE;
288 }
289
290 // m_efManager.getPrecedenceRulesGraph()->dumpExecutionPlan();
291
292 return sc;
293 }
StatusCode::FAILURE
Definition: StatusCode.h:29
Service::finalize
StatusCode finalize() override
Definition: Service.cpp:174
CommonMessagingBase::info
MsgStream & info() const
shortcut for the method msgStream(MSG::INFO)
Definition: CommonMessaging.h:123
AvalancheSchedulerSvc::m_isActive
std::atomic< ActivationState > m_isActive
Flag to track if the scheduler is active or not.
Definition: AvalancheSchedulerSvc.h:166
CommonMessagingBase::error
MsgStream & error() const
shortcut for the method msgStream(MSG::ERROR)
Definition: CommonMessaging.h:117
CommonMessagingBase::warning
MsgStream & warning() const
shortcut for the method msgStream(MSG::WARNING)
Definition: CommonMessaging.h:120
std::thread::join
T join(T...args)
StatusCode
This class is used for returning status codes from appropriate routines.
Definition: StatusCode.h:26
AvalancheSchedulerSvc::deactivate
StatusCode deactivate()
Deactivate scheduler.
Definition: AvalancheSchedulerSvc.cpp:348
AvalancheSchedulerSvc::FAILURE
Definition: AvalancheSchedulerSvc.h:131
endmsg
MsgStream & endmsg(MsgStream &s)
MsgStream Modifier: endmsg. Calls the output method of the MsgStream.
Definition: MsgStream.h:244
AvalancheSchedulerSvc::m_thread
std::thread m_thread
The thread in which the activate function runs.
Definition: AvalancheSchedulerSvc.h:169

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

Get free slots number.

Definition at line 446 of file AvalancheSchedulerSvc.cpp.

446 {
447 return std::max( m_freeSlots.load(), 0 );
448 }
AvalancheSchedulerSvc::m_freeSlots
std::atomic_int m_freeSlots
Atomic to account for asyncronous updates by the scheduler wrt the rest.
Definition: AvalancheSchedulerSvc.h:193
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 368 of file AvalancheSchedulerSvc.cpp.

368 {
369 return m_algname_vect[index];
370 }
AvalancheSchedulerSvc::m_algname_vect
std::vector< std::string > m_algname_vect
Vector to bookkeep the information necessary to the index2name conversion.
Definition: AvalancheSchedulerSvc.h:181

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

45 {
46
47 // Initialise mother class (read properties, ...)
48 StatusCode sc( Service::initialize() );
49 if ( !sc.isSuccess() ) warning() << "Base class could not be initialized" << endmsg;
50
51 // Get hold of the TBBSvc. This should initialize the thread pool
52 m_threadPoolSvc = serviceLocator()->service( "ThreadPoolSvc" );
53 if ( !m_threadPoolSvc.isValid() ) {
54 fatal() << "Error retrieving ThreadPoolSvc" << endmsg;
55 return StatusCode::FAILURE;
56 }
57
58 // Activate the scheduler in another thread.
59 info() << "Activating scheduler in a separate thread" << endmsg;
60 m_thread = std::thread( std::bind( &AvalancheSchedulerSvc::activate, this ) );
61
62 while ( m_isActive != ACTIVE ) {
63 if ( m_isActive == FAILURE ) {
64 fatal() << "Terminating initialization" << endmsg;
65 return StatusCode::FAILURE;
66 } else {
67 info() << "Waiting for AvalancheSchedulerSvc to activate" << endmsg;
68 sleep( 1 );
69 }
70 }
71
72 // Get the algo resource pool
73 m_algResourcePool = serviceLocator()->service( "AlgResourcePool" );
74 if ( !m_algResourcePool.isValid() ) {
75 fatal() << "Error retrieving AlgoResourcePool" << endmsg;
76 return StatusCode::FAILURE;
77 }
78
79 m_algExecStateSvc = serviceLocator()->service("AlgExecStateSvc");
80 if (!m_algExecStateSvc.isValid()) {
81 fatal() << "Error retrieving AlgExecStateSvc" << endmsg;
82 return StatusCode::FAILURE;
83 }
84
85 // Get Whiteboard
86 m_whiteboard = serviceLocator()->service( m_whiteboardSvcName );
87 if ( !m_whiteboard.isValid() ) {
88 fatal() << "Error retrieving EventDataSvc interface IHiveWhiteBoard." << endmsg;
89 return StatusCode::FAILURE;
90 }
91
92 // Get dedicated scheduler for I/O-bound algorithms
93 if ( m_useIOBoundAlgScheduler ) {
94 m_IOBoundAlgScheduler = serviceLocator()->service( m_IOBoundAlgSchedulerSvcName );
95 if ( !m_IOBoundAlgScheduler.isValid() )
96 fatal() << "Error retrieving IOBoundSchedulerAlgSvc interface IAccelerator." << endmsg;
97 }
98
99 // Check the MaxEventsInFlight parameters and react
100 // Deprecated for the moment
101 size_t numberOfWBSlots = m_whiteboard->getNumberOfStores();
102 if ( m_maxEventsInFlight != 0 ) {
103 warning() << "Property MaxEventsInFlight was set. This works but it's deprecated. "
104 << "Please migrate your code options files." << endmsg;
105
106 if ( m_maxEventsInFlight != (int)numberOfWBSlots ) {
107 warning() << "In addition, the number of events in flight (" << m_maxEventsInFlight
108 << ") differs from the slots in the whiteboard (" << numberOfWBSlots
109 << "). Setting the number of events in flight to " << numberOfWBSlots << endmsg;
110 }
111 }
112
113 // Align the two quantities
114 m_maxEventsInFlight = numberOfWBSlots;
115
116 // Set the number of free slots
117 m_freeSlots = m_maxEventsInFlight;
118
119 // set global concurrency flags
120 Gaudi::Concurrency::ConcurrencyFlags::setNumConcEvents( numberOfWBSlots );
121
122 if ( m_algosDependencies.size() != 0 ) {
123 warning() << " ##### Property AlgosDependencies is deprecated and ignored."
124 << " FIX your job options #####" << endmsg;
125 }
126
127 // Get the list of algorithms
128 const std::list<IAlgorithm*>& algos = m_algResourcePool->getFlatAlgList();
129 const unsigned int algsNumber = algos.size();
130 info() << "Found " << algsNumber << " algorithms" << endmsg;
131
132 /* Dependencies
133 1) Look for handles in algo, if none
134 2) Assume none are required
135 */
136
137 DataObjIDColl globalInp, globalOutp;
138
139 // figure out all outputs
140 for (IAlgorithm* ialgoPtr : algos) {
141 Algorithm* algoPtr = dynamic_cast<Algorithm*>(ialgoPtr);
142 if (!algoPtr) {
143 fatal() << "Could not convert IAlgorithm into Algorithm: this will result in a crash." << endmsg;
144 }
145 for (auto id : algoPtr->outputDataObjs()) {
146 auto r = globalOutp.insert(id);
147 if (!r.second) {
148 warning() << "multiple algorithms declare " << id << " as output! could be a single instance in multiple paths though, or control flow may guarantee only one runs...!" << endmsg;
149 }
150 }
151 }
152 info() << "outputs:\n" ;
153 for (const auto& i : globalOutp ) {
154 info() << i << '\n' ;
155 }
156 info() << endmsg;
157
158
159
160 info() << "Data Dependencies for Algorithms:";
161
162 std::vector<DataObjIDColl> m_algosDependencies;
163 for ( IAlgorithm* ialgoPtr : algos ) {
164 Algorithm* algoPtr = dynamic_cast<Algorithm*>( ialgoPtr );
165 if ( nullptr == algoPtr )
166 fatal() << "Could not convert IAlgorithm into Algorithm: this will result in a crash." << endmsg;
167
168 info() << "\n " << algoPtr->name();
169
170 // FIXME
171 DataObjIDColl algoDependencies;
172 if ( !algoPtr->inputDataObjs().empty() || !algoPtr->outputDataObjs().empty() ) {
173 for ( auto id : algoPtr->inputDataObjs() ) {
174 info() << "\n o INPUT " << id;
175 if (id.key().find(":")!=std::string::npos) {
176 info() << " contains alternatives which require resolution... " << endmsg;
177 auto tokens = boost::tokenizer<boost::char_separator<char>>{id.key(),boost::char_separator<char>{":"}};
178 auto itok = std::find_if( tokens.begin(), tokens.end(),
179 [&](const std::string& t) {
180 return globalOutp.find( DataObjID{t} ) != globalOutp.end();
181 } );
182 if (itok!=tokens.end()) {
183 info() << "found matching output for " << *itok << " -- updating scheduler info" << endmsg;
184 id.updateKey(*itok);
185 } else {
186 error() << "failed to find alternate in global output list" << endmsg;
187 }
188 }
189 algoDependencies.insert( id );
190 globalInp.insert( id );
191 }
192 for ( auto id : algoPtr->outputDataObjs() ) {
193 info() << "\n o OUTPUT " << id;
194 if (id.key().find(":")!=std::string::npos) {
195 info() << " alternatives are NOT allowed for outputs..." << endmsg;
196 }
197 }
198 } else {
199 info() << "\n none";
200 }
201 m_algosDependencies.emplace_back( algoDependencies );
202 }
203 info() << endmsg;
204
205 // Fill the containers to convert algo names to index
206 m_algname_vect.reserve( algsNumber );
207 unsigned int index = 0;
208 for ( IAlgorithm* algo : algos ) {
209 const std::string& name = algo->name();
210 m_algname_index_map[name] = index;
211 m_algname_vect.emplace_back( name );
212 index++;
213 }
214
215 // Check if we have unmet global input dependencies
216 if ( m_checkDeps ) {
217 DataObjIDColl unmetDep;
218 for ( auto o : globalInp ) {
219 if ( globalOutp.find( o ) == globalOutp.end() ) {
220 unmetDep.insert( o );
221 }
222 }
223
224 if ( unmetDep.size() > 0 ) {
225 fatal() << "The following unmet INPUT data dependencies were found: ";
226 for ( auto& o : unmetDep ) {
227 fatal() << "\n o " << o << " required by Algorithm: ";
228 for ( size_t i = 0; i < m_algosDependencies.size(); ++i ) {
229 if ( m_algosDependencies[i].find( o ) != m_algosDependencies[i].end() ) {
230 fatal() << "\n * " << m_algname_vect[i];
231 }
232 }
233 }
234 fatal() << endmsg;
235 return StatusCode::FAILURE;
236 } else {
237 info() << "No unmet INPUT data dependencies were found" << endmsg;
238 }
239 }
240
241 // prepare the control flow part
242 const AlgResourcePool* algPool = dynamic_cast<const AlgResourcePool*>( m_algResourcePool.get() );
243 sc = m_efManager.initialize( algPool->getPRGraph(), m_algname_index_map, m_eventSlots, m_optimizationMode );
244 unsigned int controlFlowNodeNumber = m_efManager.getPrecedenceRulesGraph()->getControlFlowNodeCounter();
245
246 // Shortcut for the message service
247 SmartIF<IMessageSvc> messageSvc( serviceLocator() );
248 if ( !messageSvc.isValid() ) error() << "Error retrieving MessageSvc interface IMessageSvc." << endmsg;
249
250 m_eventSlots.assign( m_maxEventsInFlight,
251 EventSlot( m_algosDependencies, algsNumber, controlFlowNodeNumber, messageSvc ) );
252 std::for_each( m_eventSlots.begin(), m_eventSlots.end(), []( EventSlot& slot ) { slot.complete = true; } );
253
254 // Clearly inform about the level of concurrency
255 info() << "Concurrency level information:" << endmsg;
256 info() << " o Number of events in flight: " << m_maxEventsInFlight << endmsg;
257 info() << " o Number of algorithms in flight: " << m_maxAlgosInFlight << endmsg;
258 info() << " o TBB thread pool size: " << m_threadPoolSize << endmsg;
259
260 // Simulating execution flow by only analyzing the graph topology and logic
261 if ( m_simulateExecution ) {
262 auto vis = concurrency::RunSimulator( 0 );
263 m_efManager.simulateExecutionFlow( vis );
264 }
265
266 return sc;
267 }
concurrency::ExecutionFlowManager::simulateExecutionFlow
void simulateExecutionFlow(IGraphVisitor &visitor) const
Definition: ExecutionFlowManager.cpp:53
Service::initialize
StatusCode initialize() override
Definition: Service.cpp:64
SmartIF< IMessageSvc >
std::unordered_set::empty
T empty(T...args)
AvalancheSchedulerSvc::m_whiteboardSvcName
Gaudi::Property< std::string > m_whiteboardSvcName
Definition: AvalancheSchedulerSvc.h:137
Algorithm::name
const std::string & name() const override
The identifying name of the algorithm object.
Definition: Algorithm.cpp:715
Service::name
const std::string & name() const override
Retrieve name of the service.
Definition: Service.cpp:289
AlgResourcePool::getPRGraph
virtual concurrency::PrecedenceRulesGraph * getPRGraph() const
Definition: AlgResourcePool.h:58
StatusCode::FAILURE
Definition: StatusCode.h:29
DataObjID
Definition: DataObjID.h:38
CommonMessagingBase::info
MsgStream & info() const
shortcut for the method msgStream(MSG::INFO)
Definition: CommonMessaging.h:123
concurrency::RunSimulator
Definition: PRGraphVisitors.h:9
DataHandleHolderBase::outputDataObjs
const DataObjIDColl & outputDataObjs() const override
Definition: DataHandleHolderBase.h:63
AvalancheSchedulerSvc::ACTIVE
Definition: AvalancheSchedulerSvc.h:131
concurrency::ExecutionFlowManager::initialize
StatusCode initialize(PrecedenceRulesGraph *graph, const std::unordered_map< std::string, unsigned int > &algname_index_map)
Initialize the control flow manager It greps the topalg list and the index map for the algo names...
Definition: ExecutionFlowManager.cpp:6
AvalancheSchedulerSvc::activate
void activate()
Activate scheduler.
Definition: AvalancheSchedulerSvc.cpp:305
std::vector::end
T end(T...args)
AvalancheSchedulerSvc::m_algosDependencies
Gaudi::Property< std::vector< std::vector< std::string > > > m_algosDependencies
Definition: AvalancheSchedulerSvc.h:153
AvalancheSchedulerSvc::m_optimizationMode
Gaudi::Property< std::string > m_optimizationMode
Definition: AvalancheSchedulerSvc.h:147
IAlgResourcePool::getFlatAlgList
virtual std::list< IAlgorithm * > getFlatAlgList()=0
Get the flat list of algorithms.
AlgResourcePool
The AlgResourcePool is a concrete implementation of the IAlgResourcePool interface.
Definition: AlgResourcePool.h:31
AvalancheSchedulerSvc::m_isActive
std::atomic< ActivationState > m_isActive
Flag to track if the scheduler is active or not.
Definition: AvalancheSchedulerSvc.h:166
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:175
AvalancheSchedulerSvc::m_checkDeps
Gaudi::Property< bool > m_checkDeps
Definition: AvalancheSchedulerSvc.h:155
std::string
STL class.
AvalancheSchedulerSvc::m_useIOBoundAlgScheduler
Gaudi::Property< bool > m_useIOBoundAlgScheduler
Definition: AvalancheSchedulerSvc.h:151
SmartIF::get
TYPE * get() const
Get interface pointer.
Definition: SmartIF.h:76
AvalancheSchedulerSvc::m_freeSlots
std::atomic_int m_freeSlots
Atomic to account for asyncronous updates by the scheduler wrt the rest.
Definition: AvalancheSchedulerSvc.h:193
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:78
CommonMessagingBase::error
MsgStream & error() const
shortcut for the method msgStream(MSG::ERROR)
Definition: CommonMessaging.h:117
AvalancheSchedulerSvc::m_algResourcePool
SmartIF< IAlgResourcePool > m_algResourcePool
Cache for the algorithm resource pool.
Definition: AvalancheSchedulerSvc.h:235
CommonMessagingBase::warning
MsgStream & warning() const
shortcut for the method msgStream(MSG::WARNING)
Definition: CommonMessaging.h:120
StatusCode
This class is used for returning status codes from appropriate routines.
Definition: StatusCode.h:26
AvalancheSchedulerSvc::m_whiteboard
SmartIF< IHiveWhiteBoard > m_whiteboard
A shortcut to the whiteboard.
Definition: AvalancheSchedulerSvc.h:184
DataHandleHolderBase::inputDataObjs
const DataObjIDColl & inputDataObjs() const override
Definition: DataHandleHolderBase.h:62
AvalancheSchedulerSvc::m_algname_vect
std::vector< std::string > m_algname_vect
Vector to bookkeep the information necessary to the index2name conversion.
Definition: AvalancheSchedulerSvc.h:181
AvalancheSchedulerSvc::m_IOBoundAlgSchedulerSvcName
Gaudi::Property< std::string > m_IOBoundAlgSchedulerSvcName
Definition: AvalancheSchedulerSvc.h:139
std::bind
T bind(T...args)
std::unordered_set< DataObjID, DataObjID_Hasher >
AvalancheSchedulerSvc::m_efManager
concurrency::ExecutionFlowManager m_efManager
Member to take care of the control flow.
Definition: AvalancheSchedulerSvc.h:263
AvalancheSchedulerSvc::m_threadPoolSize
Gaudi::Property< int > m_threadPoolSize
Definition: AvalancheSchedulerSvc.h:135
AvalancheSchedulerSvc::m_threadPoolSvc
SmartIF< IThreadPoolSvc > m_threadPoolSvc
Definition: AvalancheSchedulerSvc.h:266
std::list< IAlgorithm * >
AvalancheSchedulerSvc::m_IOBoundAlgScheduler
SmartIF< IAccelerator > m_IOBoundAlgScheduler
A shortcut to IO-bound algorithm scheduler.
Definition: AvalancheSchedulerSvc.h:187
IAlgorithm
The IAlgorithm is the interface implemented by the Algorithm base class.
Definition: IAlgorithm.h:27
std::unordered_set::insert
T insert(T...args)
AvalancheSchedulerSvc::m_maxEventsInFlight
Gaudi::Property< int > m_maxEventsInFlight
Definition: AvalancheSchedulerSvc.h:133
Algorithm
Base class from which all concrete algorithm classes should be derived.
Definition: Algorithm.h:78
std::find
T find(T...args)
std::list::size
T size(T...args)
AvalancheSchedulerSvc::m_maxAlgosInFlight
Gaudi::Property< unsigned int > m_maxAlgosInFlight
Definition: AvalancheSchedulerSvc.h:141
std::vector
STL class.
AvalancheSchedulerSvc::m_simulateExecution
Gaudi::Property< bool > m_simulateExecution
Definition: AvalancheSchedulerSvc.h:145
SmartIF::isValid
bool isValid() const
Allow for check if smart pointer is valid.
Definition: SmartIF.h:62
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:190
AvalancheSchedulerSvc::FAILURE
Definition: AvalancheSchedulerSvc.h:131
CommonMessagingBase::fatal
MsgStream & fatal() const
shortcut for the method msgStream(MSG::FATAL)
Definition: CommonMessaging.h:111
IHiveWhiteBoard::getNumberOfStores
virtual size_t getNumberOfStores() const =0
Get the number of &#39;slots&#39;.
concurrency::ExecutionFlowManager::getPrecedenceRulesGraph
PrecedenceRulesGraph * getPrecedenceRulesGraph() const
Get the flow graph instance.
Definition: ExecutionFlowManager.h:38
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:202
endmsg
MsgStream & endmsg(MsgStream &s)
MsgStream Modifier: endmsg. Calls the output method of the MsgStream.
Definition: MsgStream.h:244
Gaudi::Concurrency::ConcurrencyFlags::setNumConcEvents
static GAUDI_API void setNumConcEvents(const std::size_t &nE)
Definition: ConcurrencyFlags.h:55
concurrency::PrecedenceRulesGraph::getControlFlowNodeCounter
unsigned int getControlFlowNodeCounter() const
Get total number of graph nodes.
Definition: PrecedenceRulesGraph.h:321
std::vector::reserve
T reserve(T...args)
std::vector::emplace_back
T emplace_back(T...args)
AvalancheSchedulerSvc::m_thread
std::thread m_thread
The thread in which the activate function runs.
Definition: AvalancheSchedulerSvc.h:169

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

731 {
732 // Get the slot
733 EventSlot& thisSlot = m_eventSlots[iSlot];
734
735 if ( m_actionsQueue.empty() && m_algosInFlight == 0 && m_IOBoundAlgosInFlight == 0 &&
736 ( !thisSlot.algsStates.algsPresent( AlgsExecutionStates::DATAREADY ) ) ) {
737
738 info() << "About to declare a stall" << endmsg;
739 fatal() << "*** Stall detected! ***\n" << endmsg;
740 dumpSchedulerState( iSlot );
741 // throw GaudiException ("Stall detected",name(),StatusCode::FAILURE);
742
743 return StatusCode::FAILURE;
744 }
745 return StatusCode::SUCCESS;
746 }
AlgsExecutionStates::algsPresent
bool algsPresent(State state) const
Definition: AlgsExecutionStates.h:54
StatusCode::SUCCESS
Definition: StatusCode.h:30
AvalancheSchedulerSvc::m_IOBoundAlgosInFlight
unsigned int m_IOBoundAlgosInFlight
Number of algoritms presently in flight.
Definition: AvalancheSchedulerSvc.h:210
StatusCode::FAILURE
Definition: StatusCode.h:29
CommonMessagingBase::info
MsgStream & info() const
shortcut for the method msgStream(MSG::INFO)
Definition: CommonMessaging.h:123
AvalancheSchedulerSvc::dumpSchedulerState
void dumpSchedulerState(int iSlot)
Dump the state of the scheduler.
Definition: AvalancheSchedulerSvc.cpp:755
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:207
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:190
CommonMessagingBase::fatal
MsgStream & fatal() const
shortcut for the method msgStream(MSG::FATAL)
Definition: CommonMessaging.h:111
endmsg
MsgStream & endmsg(MsgStream &s)
MsgStream Modifier: endmsg. Calls the output method of the MsgStream.
Definition: MsgStream.h:244
AvalancheSchedulerSvc::m_actionsQueue
tbb::concurrent_bounded_queue< action > m_actionsQueue
Queue where closures are stored and picked for execution.
Definition: AvalancheSchedulerSvc.h:243

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

454 {
455
456 unsigned int slotNum = 0;
457 for ( auto& thisSlot : m_eventSlots ) {
458 if ( not thisSlot.algsStates.allAlgsExecuted() and not thisSlot.complete ) {
459 updateStates( slotNum );
460 }
461 slotNum++;
462 }
463 return StatusCode::SUCCESS;
464 }
StatusCode::SUCCESS
Definition: StatusCode.h:30
AvalancheSchedulerSvc::m_eventSlots
std::vector< EventSlot > m_eventSlots
Vector of events slots.
Definition: AvalancheSchedulerSvc.h:190
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:555

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

470 {
471 // debug() << "popFinishedEvent: queue size: " << m_finishedEvents.size() << endmsg;
472 if ( m_freeSlots.load() == m_maxEventsInFlight or m_isActive == INACTIVE ) {
473 // debug() << "freeslots: " << m_freeSlots << "/" << m_maxEventsInFlight
474 // << " active: " << m_isActive << endmsg;
475 return StatusCode::FAILURE;
476 } else {
477 // debug() << "freeslots: " << m_freeSlots << "/" << m_maxEventsInFlight
478 // << " active: " << m_isActive << endmsg;
479 m_finishedEvents.pop( eventContext );
480 m_freeSlots++;
481 if (msgLevel(MSG::DEBUG))
482 debug() << "Popped slot " << eventContext->slot() << "(event "
483 << eventContext->evt() << ")" << endmsg;
484 return StatusCode::SUCCESS;
485 }
486 }
StatusCode::SUCCESS
Definition: StatusCode.h:30
AvalancheSchedulerSvc::INACTIVE
Definition: AvalancheSchedulerSvc.h:131
StatusCode::FAILURE
Definition: StatusCode.h:29
EventContext::slot
ContextID_t slot() const
Definition: EventContext.h:41
MSG::DEBUG
Definition: IMessageSvc.h:18
AvalancheSchedulerSvc::m_isActive
std::atomic< ActivationState > m_isActive
Flag to track if the scheduler is active or not.
Definition: AvalancheSchedulerSvc.h:166
EventContext::evt
ContextEvt_t evt() const
Definition: EventContext.h:40
AvalancheSchedulerSvc::m_freeSlots
std::atomic_int m_freeSlots
Atomic to account for asyncronous updates by the scheduler wrt the rest.
Definition: AvalancheSchedulerSvc.h:193
AvalancheSchedulerSvc::m_finishedEvents
tbb::concurrent_bounded_queue< EventContext * > m_finishedEvents
Queue of finished events.
Definition: AvalancheSchedulerSvc.h:196
AvalancheSchedulerSvc::m_maxEventsInFlight
Gaudi::Property< int > m_maxEventsInFlight
Definition: AvalancheSchedulerSvc.h:133
CommonMessagingBase::debug
MsgStream & debug() const
shortcut for the method msgStream(MSG::DEBUG)
Definition: CommonMessaging.h:126
CommonMessagingBase::msgLevel
MSG::Level msgLevel() const
get the output level from the embedded MsgStream
Definition: CommonMessaging.h:135
endmsg
MsgStream & endmsg(MsgStream &s)
MsgStream Modifier: endmsg. Calls the output method of the MsgStream.
Definition: MsgStream.h:244

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

993 {
994 // Put back the instance
995 Algorithm* castedAlgo = dynamic_cast<Algorithm*>( algo ); // DP: expose context getter in IAlgo?
996 if ( !castedAlgo ) fatal() << "[Asynchronous] The casting did not succeed!" << endmsg;
997 // EventContext* eventContext = castedAlgo->getContext();
998
999 // Check if the execution failed
1000 if (m_algExecStateSvc->eventStatus(*eventContext) != EventStatus::Success)
1001 eventFailed(eventContext).ignore();
1002
1003 StatusCode sc = m_algResourcePool->releaseAlgorithm( algo->name(), algo );
1004
1005 if ( !sc.isSuccess() ) {
1006 error() << "[Asynchronous] [Event " << eventContext->evt() << ", Slot " << eventContext->slot() << "] "
1007 << "Instance of algorithm " << algo->name() << " could not be properly put back." << endmsg;
1008 return StatusCode::FAILURE;
1009 }
1010
1011 m_IOBoundAlgosInFlight--;
1012
1013 EventSlot& thisSlot = m_eventSlots[si];
1014
1015 if (msgLevel(MSG::DEBUG))
1016 debug() << "[Asynchronous] Algorithm " << algo->name() << " executed in slot " << si
1017 << ". Algorithms scheduled are " << m_IOBoundAlgosInFlight << endmsg;
1018
1019 // Schedule an update of the status of the algorithms
1020 auto updateAction = std::bind( &AvalancheSchedulerSvc::updateStates, this, -1, algo->name() );
1021 m_actionsQueue.push( updateAction );
1022 m_updateNeeded = false;
1023
1024 if (msgLevel(MSG::DEBUG))
1025 debug() << "[Asynchronous] Trying to handle execution result of "
1026 << index2algname(iAlgo) << " on slot " << si << endmsg;
1027 State state;
1028 if ( algo->filterPassed() ) {
1029 state = State::EVTACCEPTED;
1030 } else {
1031 state = State::EVTREJECTED;
1032 }
1033
1034 sc = thisSlot.algsStates.updateState( iAlgo, state );
1035
1036 if (sc.isSuccess())
1037 if (msgLevel(MSG::VERBOSE))
1038 verbose() << "[Asynchronous] Promoting " << index2algname(iAlgo) << " on slot "
1039 << si << " to " << AlgsExecutionStates::stateNames[state] << endmsg;
1040
1041 return sc;
1042 }
AvalancheSchedulerSvc::m_IOBoundAlgosInFlight
unsigned int m_IOBoundAlgosInFlight
Number of algoritms presently in flight.
Definition: AvalancheSchedulerSvc.h:210
StatusCode::FAILURE
Definition: StatusCode.h:29
compareRootHistos.state
state
Definition: compareRootHistos.py:348
EventContext::slot
ContextID_t slot() const
Definition: EventContext.h:41
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:74
CommonMessagingBase::verbose
MsgStream & verbose() const
shortcut for the method msgStream(MSG::VERBOSE)
Definition: CommonMessaging.h:129
MSG::DEBUG
Definition: IMessageSvc.h:18
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:17
EventContext::evt
ContextEvt_t evt() const
Definition: EventContext.h:40
CommonMessagingBase::error
MsgStream & error() const
shortcut for the method msgStream(MSG::ERROR)
Definition: CommonMessaging.h:117
AvalancheSchedulerSvc::m_algResourcePool
SmartIF< IAlgResourcePool > m_algResourcePool
Cache for the algorithm resource pool.
Definition: AvalancheSchedulerSvc.h:235
StatusCode
This class is used for returning status codes from appropriate routines.
Definition: StatusCode.h:26
EventStatus::Success
Definition: IAlgExecStateSvc.h:64
std::bind
T bind(T...args)
AvalancheSchedulerSvc::m_updateNeeded
bool m_updateNeeded
Keep track of update actions scheduled.
Definition: AvalancheSchedulerSvc.h:231
Algorithm
Base class from which all concrete algorithm classes should be derived.
Definition: Algorithm.h:78
CommonMessagingBase::debug
MsgStream & debug() const
shortcut for the method msgStream(MSG::DEBUG)
Definition: CommonMessaging.h:126
AvalancheSchedulerSvc::eventFailed
StatusCode eventFailed(EventContext *eventContext)
Method to check if an event failed and take appropriate actions.
Definition: AvalancheSchedulerSvc.cpp:509
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:368
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:190
StatusCode::ignore
void ignore() const
Definition: StatusCode.h:106
CommonMessagingBase::fatal
MsgStream & fatal() const
shortcut for the method msgStream(MSG::FATAL)
Definition: CommonMessaging.h:111
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:555
CommonMessagingBase::msgLevel
MSG::Level msgLevel() const
get the output level from the embedded MsgStream
Definition: CommonMessaging.h:135
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:202
endmsg
MsgStream & endmsg(MsgStream &s)
MsgStream Modifier: endmsg. Calls the output method of the MsgStream.
Definition: MsgStream.h:244
AvalancheSchedulerSvc::m_actionsQueue
tbb::concurrent_bounded_queue< action > m_actionsQueue
Queue where closures are stored and picked for execution.
Definition: AvalancheSchedulerSvc.h:243
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:14

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

Definition at line 890 of file AvalancheSchedulerSvc.cpp.

890 {
891
892 if ( m_IOBoundAlgosInFlight == m_maxIOBoundAlgosInFlight ) return StatusCode::FAILURE;
893
894 // bool IOBound = m_efManager.getPrecedenceRulesGraph()->getAlgorithmNode(algName)->isIOBound();
895
896 const std::string& algName( index2algname( iAlgo ) );
897 IAlgorithm* ialgoPtr = nullptr;
898 StatusCode sc( m_algResourcePool->acquireAlgorithm( algName, ialgoPtr ) );
899
900 if ( sc.isSuccess() ) { // if we managed to get an algorithm instance try to schedule it
901 EventContext* eventContext( m_eventSlots[si].eventContext );
902 if ( !eventContext )
903 fatal() << "[Asynchronous] Event context for algorithm " << algName << " is a nullptr (slot " << si << ")"
904 << endmsg;
905
906 ++m_IOBoundAlgosInFlight;
907 // Can we use tbb-based overloaded new-operator for a "custom" task (an algorithm wrapper, not derived from tbb::task)? it seems it works..
908 IOBoundAlgTask* theTask = new( tbb::task::allocate_root() )
909 IOBoundAlgTask(ialgoPtr, iAlgo, eventContext, serviceLocator(), m_algExecStateSvc);
910 m_IOBoundAlgScheduler->push(*theTask);
911
912 if (msgLevel(MSG::DEBUG))
913 debug() << "[Asynchronous] Algorithm " << algName << " was submitted on event "
914 << eventContext->evt() << " in slot " << si
915 << ". algorithms scheduled are " << m_IOBoundAlgosInFlight << endmsg;
916
917 StatusCode updateSc( m_eventSlots[si].algsStates.updateState( iAlgo, AlgsExecutionStates::SCHEDULED ) );
918
919 if (updateSc.isSuccess())
920 if (msgLevel(MSG::VERBOSE))
921 verbose() << "[Asynchronous] Promoting " << index2algname(iAlgo)
922 << " to SCHEDULED on slot " << si << endmsg;
923 return updateSc;
924 } else {
925 if ( msgLevel( MSG::DEBUG ) )
926 debug() << "[Asynchronous] Could not acquire instance for algorithm " << index2algname( iAlgo ) << " on slot "
927 << si << endmsg;
928 return sc;
929 }
930 }
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:210
StatusCode::FAILURE
Definition: StatusCode.h:29
CommonMessagingBase::verbose
MsgStream & verbose() const
shortcut for the method msgStream(MSG::VERBOSE)
Definition: CommonMessaging.h:129
MSG::DEBUG
Definition: IMessageSvc.h:18
MSG::VERBOSE
Definition: IMessageSvc.h:17
EventContext
This class represents an entry point to all the event specific data.
Definition: EventContext.h:25
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:235
StatusCode
This class is used for returning status codes from appropriate routines.
Definition: StatusCode.h:26
AvalancheSchedulerSvc::m_IOBoundAlgScheduler
SmartIF< IAccelerator > m_IOBoundAlgScheduler
A shortcut to IO-bound algorithm scheduler.
Definition: AvalancheSchedulerSvc.h:187
IAlgorithm
The IAlgorithm is the interface implemented by the Algorithm base class.
Definition: IAlgorithm.h:27
CommonMessagingBase::debug
MsgStream & debug() const
shortcut for the method msgStream(MSG::DEBUG)
Definition: CommonMessaging.h:126
AvalancheSchedulerSvc::index2algname
const std::string & index2algname(unsigned int index)
Convert an integer to a name.
Definition: AvalancheSchedulerSvc.cpp:368
AvalancheSchedulerSvc::m_eventSlots
std::vector< EventSlot > m_eventSlots
Vector of events slots.
Definition: AvalancheSchedulerSvc.h:190
IAccelerator::push
virtual StatusCode push(IAlgTask &task)=0
CommonMessagingBase::fatal
MsgStream & fatal() const
shortcut for the method msgStream(MSG::FATAL)
Definition: CommonMessaging.h:111
CommonMessagingBase::msgLevel
MSG::Level msgLevel() const
get the output level from the embedded MsgStream
Definition: CommonMessaging.h:135
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:202
AvalancheSchedulerSvc::m_maxIOBoundAlgosInFlight
Gaudi::Property< unsigned int > m_maxIOBoundAlgosInFlight
Definition: AvalancheSchedulerSvc.h:143
endmsg
MsgStream & endmsg(MsgStream &s)
MsgStream Modifier: endmsg. Calls the output method of the MsgStream.
Definition: MsgStream.h:244

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

937 {
938 // Put back the instance
939 Algorithm* castedAlgo = dynamic_cast<Algorithm*>( algo ); // DP: expose context getter in IAlgo?
940 if ( !castedAlgo ) fatal() << "The casting did not succeed!" << endmsg;
941 // EventContext* eventContext = castedAlgo->getContext();
942
943 // Check if the execution failed
944 if (m_algExecStateSvc->eventStatus(*eventContext) != EventStatus::Success)
945 eventFailed(eventContext).ignore();
946
947 Gaudi::Hive::setCurrentContext(eventContext);
948 StatusCode sc = m_algResourcePool->releaseAlgorithm( algo->name(), algo );
949
950 if ( !sc.isSuccess() ) {
951 error() << "[Event " << eventContext->evt() << ", Slot " << eventContext->slot() << "] "
952 << "Instance of algorithm " << algo->name() << " could not be properly put back." << endmsg;
953 return StatusCode::FAILURE;
954 }
955
956 m_algosInFlight--;
957
958 EventSlot& thisSlot = m_eventSlots[si];
959
960 if ( msgLevel( MSG::DEBUG ) )
961 debug() << "Algorithm " << algo->name() << " executed in slot " << si << ". Algorithms scheduled are "
962 << m_algosInFlight << endmsg;
963
964 // Schedule an update of the status of the algorithms
965 auto updateAction = std::bind( &AvalancheSchedulerSvc::updateStates, this, -1, algo->name() );
966 m_actionsQueue.push( updateAction );
967 m_updateNeeded = false;
968
969 if ( msgLevel( MSG::DEBUG ) )
970 debug() << "Trying to handle execution result of " << index2algname( iAlgo ) << " on slot " << si << endmsg;
971 State state;
972 if ( algo->filterPassed() ) {
973 state = State::EVTACCEPTED;
974 } else {
975 state = State::EVTREJECTED;
976 }
977
978 sc = thisSlot.algsStates.updateState( iAlgo, state );
979
980 if (sc.isSuccess())
981 if (msgLevel(MSG::VERBOSE))
982 verbose() << "Promoting " << index2algname(iAlgo) << " on slot " << si << " to "
983 << AlgsExecutionStates::stateNames[state] << endmsg;
984
985 return sc;
986 }
StatusCode::FAILURE
Definition: StatusCode.h:29
compareRootHistos.state
state
Definition: compareRootHistos.py:348
EventContext::slot
ContextID_t slot() const
Definition: EventContext.h:41
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:74
CommonMessagingBase::verbose
MsgStream & verbose() const
shortcut for the method msgStream(MSG::VERBOSE)
Definition: CommonMessaging.h:129
MSG::DEBUG
Definition: IMessageSvc.h:18
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:17
EventContext::evt
ContextEvt_t evt() const
Definition: EventContext.h:40
AvalancheSchedulerSvc::m_algosInFlight
unsigned int m_algosInFlight
Number of algoritms presently in flight.
Definition: AvalancheSchedulerSvc.h:207
CommonMessagingBase::error
MsgStream & error() const
shortcut for the method msgStream(MSG::ERROR)
Definition: CommonMessaging.h:117
AvalancheSchedulerSvc::m_algResourcePool
SmartIF< IAlgResourcePool > m_algResourcePool
Cache for the algorithm resource pool.
Definition: AvalancheSchedulerSvc.h:235
StatusCode
This class is used for returning status codes from appropriate routines.
Definition: StatusCode.h:26
EventStatus::Success
Definition: IAlgExecStateSvc.h:64
std::bind
T bind(T...args)
Gaudi::Hive::setCurrentContext
GAUDI_API void setCurrentContext(const EventContext *ctx)
Definition: ThreadLocalContext.cpp:45
AvalancheSchedulerSvc::m_updateNeeded
bool m_updateNeeded
Keep track of update actions scheduled.
Definition: AvalancheSchedulerSvc.h:231
Algorithm
Base class from which all concrete algorithm classes should be derived.
Definition: Algorithm.h:78
CommonMessagingBase::debug
MsgStream & debug() const
shortcut for the method msgStream(MSG::DEBUG)
Definition: CommonMessaging.h:126
AvalancheSchedulerSvc::eventFailed
StatusCode eventFailed(EventContext *eventContext)
Method to check if an event failed and take appropriate actions.
Definition: AvalancheSchedulerSvc.cpp:509
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:368
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:190
StatusCode::ignore
void ignore() const
Definition: StatusCode.h:106
CommonMessagingBase::fatal
MsgStream & fatal() const
shortcut for the method msgStream(MSG::FATAL)
Definition: CommonMessaging.h:111
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:555
CommonMessagingBase::msgLevel
MSG::Level msgLevel() const
get the output level from the embedded MsgStream
Definition: CommonMessaging.h:135
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:202
endmsg
MsgStream & endmsg(MsgStream &s)
MsgStream Modifier: endmsg. Calls the output method of the MsgStream.
Definition: MsgStream.h:244
AvalancheSchedulerSvc::m_actionsQueue
tbb::concurrent_bounded_queue< action > m_actionsQueue
Queue where closures are stored and picked for execution.
Definition: AvalancheSchedulerSvc.h:243
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:14

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

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

Algorithm promotion.

Definition at line 826 of file AvalancheSchedulerSvc.cpp.

826 {
827
828 if ( m_algosInFlight == m_maxAlgosInFlight ) return StatusCode::FAILURE;
829
830 const std::string& algName( index2algname( iAlgo ) );
831 IAlgorithm* ialgoPtr = nullptr;
832 StatusCode sc( m_algResourcePool->acquireAlgorithm( algName, ialgoPtr ) );
833
834 if ( sc.isSuccess() ) { // if we managed to get an algorithm instance try to schedule it
835 EventContext* eventContext( m_eventSlots[si].eventContext );
836 if ( !eventContext )
837 fatal() << "Event context for algorithm " << algName << " is a nullptr (slot " << si << ")" << endmsg;
838
839 ++m_algosInFlight;
840 auto promote2ExecutedClosure = std::bind(&AvalancheSchedulerSvc::promoteToExecuted,
841 this,
842 iAlgo,
843 eventContext->slot(),
844 ialgoPtr,
845 eventContext);
846 // Avoid to use tbb if the pool size is 1 and run in this thread
847 if (-100 != m_threadPoolSize) {
848
849 // this parent task is needed to promote an Algorithm as EXECUTED,
850 // it will be started as soon as the child task (see below) is completed
851 tbb::task* triggerAlgoStateUpdate = new(tbb::task::allocate_root())
852 enqueueSchedulerActionTask(this, promote2ExecutedClosure);
853 // setting parent's refcount to 1 is made here only for consistency
854 // (in this case since it is not scheduled explicitly and there it has only one child task)
855 triggerAlgoStateUpdate->set_ref_count(1);
856 // the child task that executes an Algorithm
857 tbb::task* algoTask = new(triggerAlgoStateUpdate->allocate_child())
858 AlgoExecutionTask(ialgoPtr, iAlgo, eventContext, serviceLocator(), m_algExecStateSvc);
859 // schedule the algoTask
860 tbb::task::enqueue( *algoTask);
861
862 } else {
863 AlgoExecutionTask theTask(ialgoPtr, iAlgo, eventContext, serviceLocator(), m_algExecStateSvc);
864 theTask.execute();
865 promote2ExecutedClosure();
866 }
867
868 if ( msgLevel( MSG::DEBUG ) )
869 debug() << "Algorithm " << algName << " was submitted on event " << eventContext->evt() << " in slot " << si
870 << ". Algorithms scheduled are " << m_algosInFlight << endmsg;
871
872 StatusCode updateSc( m_eventSlots[si].algsStates.updateState( iAlgo, AlgsExecutionStates::SCHEDULED ) );
873
874 if ( msgLevel( MSG::VERBOSE ) ) dumpSchedulerState( -1 );
875
876 if (updateSc.isSuccess())
877 if (msgLevel(MSG::VERBOSE))
878 verbose() << "Promoting " << index2algname(iAlgo) << " to SCHEDULED on slot "
879 << si << endmsg;
880 return updateSc;
881 } else {
882 if ( msgLevel( MSG::DEBUG ) )
883 debug() << "Could not acquire instance for algorithm " << index2algname( iAlgo ) << " on slot " << si << endmsg;
884 return sc;
885 }
886 }
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:29
AvalancheSchedulerSvc::dumpSchedulerState
void dumpSchedulerState(int iSlot)
Dump the state of the scheduler.
Definition: AvalancheSchedulerSvc.cpp:755
AlgoExecutionTask
Definition: AlgoExecutionTask.h:14
CommonMessagingBase::verbose
MsgStream & verbose() const
shortcut for the method msgStream(MSG::VERBOSE)
Definition: CommonMessaging.h:129
MSG::DEBUG
Definition: IMessageSvc.h:18
MSG::VERBOSE
Definition: IMessageSvc.h:17
EventContext
This class represents an entry point to all the event specific data.
Definition: EventContext.h:25
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:207
AvalancheSchedulerSvc::m_algResourcePool
SmartIF< IAlgResourcePool > m_algResourcePool
Cache for the algorithm resource pool.
Definition: AvalancheSchedulerSvc.h:235
StatusCode
This class is used for returning status codes from appropriate routines.
Definition: StatusCode.h:26
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:27
AvalancheSchedulerSvc::m_maxAlgosInFlight
Gaudi::Property< unsigned int > m_maxAlgosInFlight
Definition: AvalancheSchedulerSvc.h:141
CommonMessagingBase::debug
MsgStream & debug() const
shortcut for the method msgStream(MSG::DEBUG)
Definition: CommonMessaging.h:126
AvalancheSchedulerSvc::index2algname
const std::string & index2algname(unsigned int index)
Convert an integer to a name.
Definition: AvalancheSchedulerSvc.cpp:368
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:936
AvalancheSchedulerSvc::m_eventSlots
std::vector< EventSlot > m_eventSlots
Vector of events slots.
Definition: AvalancheSchedulerSvc.h:190
CommonMessagingBase::fatal
MsgStream & fatal() const
shortcut for the method msgStream(MSG::FATAL)
Definition: CommonMessaging.h:111
CommonMessagingBase::msgLevel
MSG::Level msgLevel() const
get the output level from the embedded MsgStream
Definition: CommonMessaging.h:135
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:202
endmsg
MsgStream & endmsg(MsgStream &s)
MsgStream Modifier: endmsg. Calls the output method of the MsgStream.
Definition: MsgStream.h:244

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

387 {
388
389 if ( m_first ) {
390 m_first = false;
391 }
392
393 if ( !eventContext ) {
394 fatal() << "Event context is nullptr" << endmsg;
395 return StatusCode::FAILURE;
396 }
397
398 if ( m_freeSlots.load() == 0 ) {
399 if ( msgLevel( MSG::DEBUG ) ) debug() << "A free processing slot could not be found." << endmsg;
400 return StatusCode::FAILURE;
401 }
402
403 // no problem as push new event is only called from one thread (event loop manager)
404 m_freeSlots--;
405
406 auto action = [this, eventContext]() -> StatusCode {
407 // Event processing slot forced to be the same as the wb slot
408 const unsigned int thisSlotNum = eventContext->slot();
409 EventSlot& thisSlot = m_eventSlots[thisSlotNum];
410 if ( !thisSlot.complete ) {
411 fatal() << "The slot " << thisSlotNum << " is supposed to be a finished event but it's not" << endmsg;
412 return StatusCode::FAILURE;
413 }
414
415 info() << "Executing event " << eventContext->evt() << " on slot " << thisSlotNum << endmsg;
416 thisSlot.reset( eventContext );
417
418 // promote to CR and DR the initial set of algorithms
419 auto vis = concurrency::Trigger( thisSlotNum );
420 m_efManager.touchReadyAlgorithms( vis );
421
422 return this->updateStates( thisSlotNum );
423 }; // end of lambda
424
425 // Kick off the scheduling!
426 if ( msgLevel( MSG::VERBOSE ) ) {
427 verbose() << "Pushing the action to update the scheduler for slot " << eventContext->slot() << endmsg;
428 verbose() << "Free slots available " << m_freeSlots.load() << endmsg;
429 }
430 m_actionsQueue.push( action );
431
432 return StatusCode::SUCCESS;
433 }
StatusCode::SUCCESS
Definition: StatusCode.h:30
StatusCode::FAILURE
Definition: StatusCode.h:29
std::function< StatusCode()>
EventContext::slot
ContextID_t slot() const
Definition: EventContext.h:41
CommonMessagingBase::info
MsgStream & info() const
shortcut for the method msgStream(MSG::INFO)
Definition: CommonMessaging.h:123
CommonMessagingBase::verbose
MsgStream & verbose() const
shortcut for the method msgStream(MSG::VERBOSE)
Definition: CommonMessaging.h:129
MSG::DEBUG
Definition: IMessageSvc.h:18
concurrency::ExecutionFlowManager::touchReadyAlgorithms
void touchReadyAlgorithms(IGraphVisitor &visitor) const
Promote all algorithms, ready to be executed, to DataReady state.
Definition: ExecutionFlowManager.cpp:127
concurrency::Trigger
A visitor, performing full top-down traversals of a graph.
Definition: PRGraphVisitors.h:38
MSG::VERBOSE
Definition: IMessageSvc.h:17
EventContext::evt
ContextEvt_t evt() const
Definition: EventContext.h:40
AvalancheSchedulerSvc::m_freeSlots
std::atomic_int m_freeSlots
Atomic to account for asyncronous updates by the scheduler wrt the rest.
Definition: AvalancheSchedulerSvc.h:193
StatusCode
This class is used for returning status codes from appropriate routines.
Definition: StatusCode.h:26
AvalancheSchedulerSvc::m_efManager
concurrency::ExecutionFlowManager m_efManager
Member to take care of the control flow.
Definition: AvalancheSchedulerSvc.h:263
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:26
CommonMessagingBase::debug
MsgStream & debug() const
shortcut for the method msgStream(MSG::DEBUG)
Definition: CommonMessaging.h:126
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:190
AvalancheSchedulerSvc::m_first
bool m_first
Definition: AvalancheSchedulerSvc.h:268
CommonMessagingBase::fatal
MsgStream & fatal() const
shortcut for the method msgStream(MSG::FATAL)
Definition: CommonMessaging.h:111
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:555
CommonMessagingBase::msgLevel
MSG::Level msgLevel() const
get the output level from the embedded MsgStream
Definition: CommonMessaging.h:135
endmsg
MsgStream & endmsg(MsgStream &s)
MsgStream Modifier: endmsg. Calls the output method of the MsgStream.
Definition: MsgStream.h:244
AvalancheSchedulerSvc::m_actionsQueue
tbb::concurrent_bounded_queue< action > m_actionsQueue
Queue where closures are stored and picked for execution.
Definition: AvalancheSchedulerSvc.h:243

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

Definition at line 436 of file AvalancheSchedulerSvc.cpp.

436 {
437 StatusCode sc;
438 for ( auto context : eventContexts ) {
439 sc = pushNewEvent( context );
440 if ( sc != StatusCode::SUCCESS ) return sc;
441 }
442 return sc;
443 }
StatusCode::SUCCESS
Definition: StatusCode.h:30
StatusCode
This class is used for returning status codes from appropriate routines.
Definition: StatusCode.h:26
AvalancheSchedulerSvc::pushNewEvent
StatusCode pushNewEvent(EventContext *eventContext) override
Make an event available to the scheduler.
Definition: AvalancheSchedulerSvc.cpp:387

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

492 {
493 if ( m_finishedEvents.try_pop( eventContext ) ) {
494 if ( msgLevel( MSG::DEBUG ) )
495 debug() << "Try Pop successful slot " << eventContext->slot() << "(event " << eventContext->evt() << ")"
496 << endmsg;
497 m_freeSlots++;
498 return StatusCode::SUCCESS;
499 }
500 return StatusCode::FAILURE;
501 }
StatusCode::SUCCESS
Definition: StatusCode.h:30
StatusCode::FAILURE
Definition: StatusCode.h:29
EventContext::slot
ContextID_t slot() const
Definition: EventContext.h:41
MSG::DEBUG
Definition: IMessageSvc.h:18
EventContext::evt
ContextEvt_t evt() const
Definition: EventContext.h:40
AvalancheSchedulerSvc::m_freeSlots
std::atomic_int m_freeSlots
Atomic to account for asyncronous updates by the scheduler wrt the rest.
Definition: AvalancheSchedulerSvc.h:193
AvalancheSchedulerSvc::m_finishedEvents
tbb::concurrent_bounded_queue< EventContext * > m_finishedEvents
Queue of finished events.
Definition: AvalancheSchedulerSvc.h:196
CommonMessagingBase::debug
MsgStream & debug() const
shortcut for the method msgStream(MSG::DEBUG)
Definition: CommonMessaging.h:126
CommonMessagingBase::msgLevel
MSG::Level msgLevel() const
get the output level from the embedded MsgStream
Definition: CommonMessaging.h:135
endmsg
MsgStream & endmsg(MsgStream &s)
MsgStream Modifier: endmsg. Calls the output method of the MsgStream.
Definition: MsgStream.h:244

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

555 {
556
557 m_updateNeeded = true;
558
559 StatusCode global_sc( StatusCode::FAILURE, true );
560
561 // Sort from the oldest to the newest event
562 // Prepare a vector of pointers to the slots to avoid copies
563 std::vector<EventSlot*> eventSlotsPtrs;
564
565 // Consider all slots if si <0 or just one otherwise
566 if ( si < 0 ) {
567 const int eventsSlotsSize( m_eventSlots.size() );
568 eventSlotsPtrs.reserve( eventsSlotsSize );
569 for ( auto slotIt = m_eventSlots.begin(); slotIt != m_eventSlots.end(); slotIt++ ) {
570 if ( !slotIt->complete ) eventSlotsPtrs.push_back( &( *slotIt ) );
571 }
572 std::sort( eventSlotsPtrs.begin(), eventSlotsPtrs.end(),
573 []( EventSlot* a, EventSlot* b ) { return a->eventContext->evt() < b->eventContext->evt(); } );
574 } else {
575 eventSlotsPtrs.push_back( &m_eventSlots[si] );
576 }
577
578 for ( EventSlot* thisSlotPtr : eventSlotsPtrs ) {
579 int iSlot = thisSlotPtr->eventContext->slot();
580
581 // Cache the states of the algos to improve readability and performance
582 auto& thisSlot = m_eventSlots[iSlot];
583 AlgsExecutionStates& thisAlgsStates = thisSlot.algsStates;
584
585 // Take care of the control ready update
586 if ( !algo_name.empty() )
587 m_efManager.updateDecision( algo_name, iSlot, thisAlgsStates, thisSlot.controlFlowState );
588
589 StatusCode partial_sc( StatusCode::FAILURE, true );
590 // first update CONTROLREADY to DATAREADY
591
592 // now update DATAREADY to SCHEDULED
593 if ( !m_optimizationMode.empty() ) {
594 auto comp_nodes = [this]( const uint& i, const uint& j ) {
595 return ( m_efManager.getPrecedenceRulesGraph()->getAlgorithmNode( index2algname( i ) )->getRank() <
596 m_efManager.getPrecedenceRulesGraph()->getAlgorithmNode( index2algname( j ) )->getRank() );
597 };
598 std::priority_queue<uint, std::vector<uint>, std::function<bool( const uint&, const uint& )>> buffer(
599 comp_nodes, std::vector<uint>() );
600 for ( auto it = thisAlgsStates.begin( AlgsExecutionStates::State::DATAREADY );
601 it != thisAlgsStates.end( AlgsExecutionStates::State::DATAREADY ); ++it )
602 buffer.push( *it );
603 /*std::stringstream s;
604 auto buffer2 = buffer;
605 while (!buffer2.empty()) {
606 s << m_efManager.getPrecedenceRulesGraph()->getAlgorithmNode(index2algname(buffer2.top()))->getRank() << ", ";
607 buffer2.pop();
608 }
609 info() << "DRBuffer is: [ " << s.str() << " ] <--" << algo_name << " executed" << endmsg;*/
610
611 /*while (!buffer.empty()) {
612 partial_sc = promoteToScheduled(buffer.top(), iSlot);
613 if (partial_sc.isFailure()) {
614 if (msgLevel(MSG::VERBOSE))
615 verbose() << "Could not apply transition from "
616 << AlgsExecutionStates::stateNames[AlgsExecutionStates::State::DATAREADY]
617 << " for algorithm " << index2algname(buffer.top()) << " on processing slot " << iSlot << endmsg;
618 if (m_useIOBoundAlgScheduler) {
619 partial_sc = promoteToAsyncScheduled(buffer.top(), iSlot);
620 if (msgLevel(MSG::VERBOSE))
621 if (partial_sc.isFailure())
622 verbose() << "[Asynchronous] Could not apply transition from "
623 << AlgsExecutionStates::stateNames[AlgsExecutionStates::State::DATAREADY]
624 << " for algorithm " << index2algname(buffer.top()) << " on processing slot " << iSlot << endmsg;
625 }
626 }
627 buffer.pop();
628 }*/
629 while ( !buffer.empty() ) {
630 bool IOBound = false;
631 if ( m_useIOBoundAlgScheduler )
632 IOBound = m_efManager.getPrecedenceRulesGraph()->getAlgorithmNode( index2algname( buffer.top() ) )->isIOBound();
633
634 if ( !IOBound )
635 partial_sc = promoteToScheduled( buffer.top(), iSlot );
636 else
637 partial_sc = promoteToAsyncScheduled( buffer.top(), iSlot );
638
639 if (msgLevel(MSG::VERBOSE))
640 if (partial_sc.isFailure())
641 verbose() << "Could not apply transition from "
642 << AlgsExecutionStates::stateNames[AlgsExecutionStates::State::DATAREADY]
643 << " for algorithm " << index2algname(buffer.top()) << " on processing slot " << iSlot << endmsg;
644
645 buffer.pop();
646 }
647
648 } else {
649 for ( auto it = thisAlgsStates.begin( AlgsExecutionStates::State::DATAREADY );
650 it != thisAlgsStates.end( AlgsExecutionStates::State::DATAREADY ); ++it ) {
651 uint algIndex = *it;
652
653 bool IOBound = false;
654 if ( m_useIOBoundAlgScheduler )
655 IOBound = m_efManager.getPrecedenceRulesGraph()->getAlgorithmNode( index2algname( algIndex ) )->isIOBound();
656
657 if ( !IOBound )
658 partial_sc = promoteToScheduled( algIndex, iSlot );
659 else
660 partial_sc = promoteToAsyncScheduled( algIndex, iSlot );
661
662 if (msgLevel(MSG::VERBOSE))
663 if (partial_sc.isFailure())
664 verbose() << "Could not apply transition from "
665 << AlgsExecutionStates::stateNames[AlgsExecutionStates::State::DATAREADY]
666 << " for algorithm " << index2algname(algIndex) << " on processing slot " << iSlot << endmsg;
667 }
668 }
669
670 if (m_dumpIntraEventDynamics) {
671 auto now = std::chrono::system_clock::now();
672 std::stringstream s;
673 s << algo_name << ", " << thisAlgsStates.sizeOfSubset(State::CONTROLREADY) << ", "
674 << thisAlgsStates.sizeOfSubset(State::DATAREADY) << ", "
675 << thisAlgsStates.sizeOfSubset(State::SCHEDULED) << ", "
676 << std::chrono::duration_cast<std::chrono::nanoseconds> (now - m_efManager.getPrecedenceRulesGraph()->getInitTime()).count()
677 << "\n";
678 auto threads = (m_threadPoolSize != -1) ? std::to_string(m_threadPoolSize)
679 : std::to_string(tbb::task_scheduler_init::default_num_threads());
680 std::ofstream myfile;
681 myfile.open( "IntraEventConcurrencyDynamics_" + threads + "T.csv", std::ios::app );
682 myfile << s.str();
683 myfile.close();
684 }
685
686 // Not complete because this would mean that the slot is already free!
687 if ( !thisSlot.complete && m_efManager.rootDecisionResolved( thisSlot.controlFlowState ) &&
688 !thisSlot.algsStates.algsPresent( AlgsExecutionStates::CONTROLREADY ) &&
689 !thisSlot.algsStates.algsPresent( AlgsExecutionStates::DATAREADY ) &&
690 !thisSlot.algsStates.algsPresent( AlgsExecutionStates::SCHEDULED ) ) {
691
692 thisSlot.complete = true;
693 // if the event did not fail, add it to the finished events
694 // otherwise it is taken care of in the error handling already
695 if(m_algExecStateSvc->eventStatus(*thisSlot.eventContext) == EventStatus::Success) {
696 m_finishedEvents.push(thisSlot.eventContext);
697 if (msgLevel(MSG::DEBUG))
698 debug() << "Event " << thisSlot.eventContext->evt() << " finished (slot "
699 << thisSlot.eventContext->slot() << ")." << endmsg;
700 }
701 // now let's return the fully evaluated result of the control flow
702 if ( msgLevel( MSG::DEBUG ) ) {
703 std::stringstream ss;
704 m_efManager.printEventState( ss, thisSlot.algsStates, thisSlot.controlFlowState, 0 );
705 debug() << ss.str() << endmsg;
706 }
707
708 thisSlot.eventContext = nullptr;
709 } else {
710 StatusCode eventStalledSC = isStalled(iSlot);
711 if (! eventStalledSC.isSuccess()) {
712 m_algExecStateSvc->setEventStatus(EventStatus::AlgStall, *thisSlot.eventContext);
713 eventFailed(thisSlot.eventContext).ignore();
714 }
715 }
716 } // end loop on slots
717
718 verbose() << "States Updated." << endmsg;
719
720 return global_sc;
721 }
std::chrono::nanoseconds
std::string::empty
T empty(T...args)
concurrency::PrecedenceRulesGraph::getInitTime
const std::chrono::system_clock::time_point getInitTime() const
Definition: PrecedenceRulesGraph.h:344
std::ofstream::open
T open(T...args)
concurrency::ExecutionFlowManager::printEventState
void printEventState(std::stringstream &ss, AlgsExecutionStates &states, const std::vector< int > &node_decisions, const unsigned int &recursionLevel) const
Print the state of the control flow for a given event.
Definition: ExecutionFlowManager.h:60
StatusCode::FAILURE
Definition: StatusCode.h:29
std::function
EventContext::slot
ContextID_t slot() const
Definition: EventContext.h:41
AvalancheSchedulerSvc::m_dumpIntraEventDynamics
Gaudi::Property< bool > m_dumpIntraEventDynamics
Definition: AvalancheSchedulerSvc.h:149
AvalancheSchedulerSvc::promoteToScheduled
StatusCode promoteToScheduled(unsigned int iAlgo, int si)
Algorithm promotion.
Definition: AvalancheSchedulerSvc.cpp:826
StatusCode::isSuccess
bool isSuccess() const
Test for a status code of SUCCESS.
Definition: StatusCode.h:74
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:731
std::to_string
T to_string(T...args)
CommonMessagingBase::verbose
MsgStream & verbose() const
shortcut for the method msgStream(MSG::VERBOSE)
Definition: CommonMessaging.h:129
MSG::DEBUG
Definition: IMessageSvc.h:18
std::chrono::duration_cast
T duration_cast(T...args)
std::vector::end
T end(T...args)
AvalancheSchedulerSvc::m_optimizationMode
Gaudi::Property< std::string > m_optimizationMode
Definition: AvalancheSchedulerSvc.h:147
concurrency::PrecedenceRulesGraph::getAlgorithmNode
AlgorithmNode * getAlgorithmNode(const std::string &algoName) const
Get the AlgorithmNode from by algorithm name using graph index.
Definition: PrecedenceRulesGraph.cpp:645
AlgsExecutionStates::sizeOfSubset
size_t sizeOfSubset(State state) const
Definition: AlgsExecutionStates.h:73
AvalancheSchedulerSvc::promoteToAsyncScheduled
StatusCode promoteToAsyncScheduled(unsigned int iAlgo, int si)
Definition: AvalancheSchedulerSvc.cpp:890
MSG::VERBOSE
Definition: IMessageSvc.h:17
AlgsExecutionStates::SCHEDULED
Definition: AlgsExecutionStates.h:36
concurrency::AlgorithmNode::isIOBound
bool isIOBound() const
Check if algorithm is I/O-bound.
Definition: PrecedenceRulesGraph.h:187
std::stringstream
STL class.
EventContext::evt
ContextEvt_t evt() const
Definition: EventContext.h:40
AvalancheSchedulerSvc::m_useIOBoundAlgScheduler
Gaudi::Property< bool > m_useIOBoundAlgScheduler
Definition: AvalancheSchedulerSvc.h:151
std::vector::push_back
T push_back(T...args)
concurrency::ExecutionFlowManager::updateDecision
void updateDecision(const std::string &algo_name, const int &slotNum, AlgsExecutionStates &states, std::vector< int > &node_decisions) const
Definition: ExecutionFlowManager.cpp:101
std::ofstream
STL class.
concurrency::ExecutionFlowManager::rootDecisionResolved
bool rootDecisionResolved(const std::vector< int > &node_decisions) const
Check whether root decision was resolved.
Definition: ExecutionFlowManager.cpp:121
concurrency::AlgorithmNode::getRank
const float & getRank() const
Get Algorithm rank.
Definition: PrecedenceRulesGraph.h:180
AlgsExecutionStates
The AlgsExecutionStates encodes the state machine for the execution of algorithms within a single eve...
Definition: AlgsExecutionStates.h:28
std::priority_queue
STL class.
StatusCode
This class is used for returning status codes from appropriate routines.
Definition: StatusCode.h:26
AvalancheSchedulerSvc::m_finishedEvents
tbb::concurrent_bounded_queue< EventContext * > m_finishedEvents
Queue of finished events.
Definition: AvalancheSchedulerSvc.h:196
std::ofstream::close
T close(T...args)
AlgsExecutionStates::CONTROLREADY
Definition: AlgsExecutionStates.h:34
EventStatus::AlgStall
Definition: IAlgExecStateSvc.h:66
std::stringstream::str
T str(T...args)
EventStatus::Success
Definition: IAlgExecStateSvc.h:64
IAlgExecStateSvc::setEventStatus
virtual void setEventStatus(const EventStatus::Status &sc, const EventContext &ctx)=0
AvalancheSchedulerSvc::m_efManager
concurrency::ExecutionFlowManager m_efManager
Member to take care of the control flow.
Definition: AvalancheSchedulerSvc.h:263
AvalancheSchedulerSvc::m_threadPoolSize
Gaudi::Property< int > m_threadPoolSize
Definition: AvalancheSchedulerSvc.h:135
AvalancheSchedulerSvc::m_updateNeeded
bool m_updateNeeded
Keep track of update actions scheduled.
Definition: AvalancheSchedulerSvc.h:231
std::count
T count(T...args)
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:126
AvalancheSchedulerSvc::eventFailed
StatusCode eventFailed(EventContext *eventContext)
Method to check if an event failed and take appropriate actions.
Definition: AvalancheSchedulerSvc.cpp:509
std::vector::begin
T begin(T...args)
AlgsExecutionStates::begin
Iterator begin(State kind)
Definition: AlgsExecutionStates.h:135
std::chrono::system_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:368
AlgsExecutionStates::DATAREADY
Definition: AlgsExecutionStates.h:35
EventSlot
Class representing the event slot.
Definition: EventSlot.h:11
BrunelScenarioForwardScheduler.app
app
Definition: BrunelScenarioForwardScheduler.py:137
gaudirun.s
string s
Definition: gaudirun.py:245
AvalancheSchedulerSvc::m_eventSlots
std::vector< EventSlot > m_eventSlots
Vector of events slots.
Definition: AvalancheSchedulerSvc.h:190
std::sort
T sort(T...args)
StatusCode::ignore
void ignore() const
Definition: StatusCode.h:106
concurrency::ExecutionFlowManager::getPrecedenceRulesGraph
PrecedenceRulesGraph * getPrecedenceRulesGraph() const
Get the flow graph instance.
Definition: ExecutionFlowManager.h:38
CommonMessagingBase::msgLevel
MSG::Level msgLevel() const
get the output level from the embedded MsgStream
Definition: CommonMessaging.h:135
AvalancheSchedulerSvc::m_algExecStateSvc
SmartIF< IAlgExecStateSvc > m_algExecStateSvc
Algorithm execution state manager.
Definition: AvalancheSchedulerSvc.h:202
endmsg
MsgStream & endmsg(MsgStream &s)
MsgStream Modifier: endmsg. Calls the output method of the MsgStream.
Definition: MsgStream.h:244
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:140

## Member Data Documentation

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

Queue where closures are stored and picked for execution.

Definition at line 243 of file AvalancheSchedulerSvc.h.

SmartIF<IAlgExecStateSvc> AvalancheSchedulerSvc::m_algExecStateSvc
| private

Algorithm execution state manager.

Definition at line 202 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 175 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 181 of file AvalancheSchedulerSvc.h.

Gaudi::Property<std::vector<std::vector<std::string> > > AvalancheSchedulerSvc::m_algosDependencies
| private

Initial value:
{this, "AlgosDependencies", {},
"[[deprecated]]"}

Definition at line 153 of file AvalancheSchedulerSvc.h.

unsigned int AvalancheSchedulerSvc::m_algosInFlight = 0
| private

Number of algoritms presently in flight.

Definition at line 207 of file AvalancheSchedulerSvc.h.

SmartIF<IAlgResourcePool> AvalancheSchedulerSvc::m_algResourcePool
| private

Cache for the algorithm resource pool.

Definition at line 235 of file AvalancheSchedulerSvc.h.

Gaudi::Property<bool> AvalancheSchedulerSvc::m_checkDeps {this, "CheckDependencies", false, "[[deprecated]]"}
| private

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

concurrency::ExecutionFlowManager AvalancheSchedulerSvc::m_efManager
| private

Member to take care of the control flow.

Definition at line 263 of file AvalancheSchedulerSvc.h.

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

Vector of events slots.

Definition at line 190 of file AvalancheSchedulerSvc.h.

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

Queue of finished events.

Definition at line 196 of file AvalancheSchedulerSvc.h.

bool AvalancheSchedulerSvc::m_first = true
| private

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

unsigned int AvalancheSchedulerSvc::m_IOBoundAlgosInFlight = 0
| private

Number of algoritms presently in flight.

Definition at line 210 of file AvalancheSchedulerSvc.h.

SmartIF<IAccelerator> AvalancheSchedulerSvc::m_IOBoundAlgScheduler
| private

A shortcut to IO-bound algorithm scheduler.

Definition at line 187 of file AvalancheSchedulerSvc.h.

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

Initial value:
{this, "IOBoundAlgSchedulerSvc",
"IOBoundAlgSchedulerSvc"}

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

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

Initial value:
{this, "MaxAlgosInFlight", 1,
"[[deprecated]] Taken from the whiteboard"}

Definition at line 141 of file AvalancheSchedulerSvc.h.

Gaudi::Property<int> AvalancheSchedulerSvc::m_maxEventsInFlight
| private

Initial value:
{this, "MaxEventsInFlight", 0,
"Maximum number of event processed simultaneously"}

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

std::mutex AvalancheSchedulerSvc::m_ssMut
| static private

Definition at line 300 of file AvalancheSchedulerSvc.h.

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

Definition at line 299 of file AvalancheSchedulerSvc.h.

std::thread AvalancheSchedulerSvc::m_thread
| private

The thread in which the activate function runs.

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

bool AvalancheSchedulerSvc::m_updateNeeded = true
| private

Keep track of update actions scheduled.

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

SmartIF<IHiveWhiteBoard> AvalancheSchedulerSvc::m_whiteboard
| private

A shortcut to the whiteboard.

Definition at line 184 of file AvalancheSchedulerSvc.h.

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

Initial value:
{this, "WhiteboardSvc", "EventDataSvc",
"The whiteboard name"}

Definition at line 137 of file AvalancheSchedulerSvc.h.

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

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