#include </builds/gaudi/Gaudi/GaudiHive/src/AvalancheSchedulerSvc.h>
|
| StatusCode | initialize () override |
| | Initialise. More...
|
| |
| StatusCode | finalize () override |
| | Finalise. More...
|
| |
| StatusCode | pushNewEvent (EventContext *eventContext) override |
| | Make an event available to the scheduler. More...
|
| |
| StatusCode | pushNewEvents (std::vector< EventContext * > &eventContexts) override |
| |
| StatusCode | popFinishedEvent (EventContext *&eventContext) override |
| | Blocks until an event is available. More...
|
| |
| StatusCode | tryPopFinishedEvent (EventContext *&eventContext) override |
| | Try to fetch an event from the scheduler. More...
|
| |
| unsigned int | freeSlots () override |
| | Get free slots number. More...
|
| |
| void | dumpState () override |
| | Dump scheduler state for all slots. More...
|
| |
| virtual StatusCode | scheduleEventView (const EventContext *sourceContext, const std::string &nodeName, std::unique_ptr< EventContext > viewContext) override |
| | Method to inform the scheduler about event views. More...
|
| |
| virtual void | recordOccupancy (int samplePeriod, std::function< void(OccupancySnapshot)> callback) override |
| | Sample occupancy at fixed interval (ms) Negative value to deactivate, 0 to snapshot every change Each sample, apply the callback function to the result. More...
|
| |
| bool | next (TaskSpec &ts, bool asynchronous) |
| |
| 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...
|
| |
| 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...
|
| |
| 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 > |
| Gaudi::Details::PropertyBase * | declareProperty (const std::string &name, ToolHandle< T > &hndl, const std::string &doc="none") |
| |
| template<class T > |
| StatusCode | declareTool (ToolHandle< T > &handle, bool createIf=true) |
| |
| template<class T > |
| StatusCode | declareTool (ToolHandle< T > &handle, std::string toolTypeAndName, bool createIf=true) |
| | Declare used tool. More...
|
| |
| template<class T > |
| Gaudi::Details::PropertyBase * | declareProperty (const std::string &name, ToolHandleArray< T > &hndlArr, const std::string &doc="none") |
| |
| template<class T > |
| void | addToolsArray (ToolHandleArray< T > &hndlArr) |
| |
| const std::vector< IAlgTool * > & | tools () const |
| |
| SmartIF< IAuditorSvc > & | auditorSvc () const |
| | The standard auditor service.May not be invoked before sysInitialize() has been invoked. More...
|
| |
| | PropertyHolder ()=default |
| |
| Gaudi::Details::PropertyBase & | declareProperty (Gaudi::Details::PropertyBase &prop) |
| | Declare a property. More...
|
| |
| Gaudi::Details::PropertyBase * | declareProperty (const std::string &name, TYPE &value, const std::string &doc="none") |
| | Helper to wrap a regular data member and use it as a regular property. More...
|
| |
| Gaudi::Details::PropertyBase * | declareProperty (const std::string &name, Gaudi::Property< TYPE, VERIFIER, HANDLERS > &prop, const std::string &doc="none") |
| | Declare a PropertyBase instance setting name and documentation. More...
|
| |
| Gaudi::Details::PropertyBase * | declareRemoteProperty (const std::string &name, IProperty *rsvc, const std::string &rname="") |
| | Declare a remote property. More...
|
| |
| StatusCode | setProperty (const std::string &name, const Gaudi::Details::PropertyBase &p) override |
| | set the property from another property with a different name More...
|
| |
| StatusCode | setProperty (const std::string &s) override |
| | set the property from the formatted string More...
|
| |
| StatusCode | setProperty (const Gaudi::Details::PropertyBase &p) |
| | Set the property from a property. More...
|
| |
| virtual StatusCode | setProperty (const std::string &name, const Gaudi::Details::PropertyBase &p)=0 |
| | Set the property from a property with a different name. More...
|
| |
| virtual StatusCode | setProperty (const std::string &s)=0 |
| | Set the property by string. More...
|
| |
| StatusCode | setProperty (const std::string &name, const char *v) |
| | Special case for string literals. More...
|
| |
| StatusCode | setProperty (const std::string &name, const std::string &v) |
| | Special case for std::string. More...
|
| |
| StatusCode | setProperty (const std::string &name, const TYPE &value) |
| | set the property form the value More...
|
| |
| StatusCode | setPropertyRepr (const std::string &n, const std::string &r) override |
| | set the property from name and value string representation More...
|
| |
| StatusCode | getProperty (Gaudi::Details::PropertyBase *p) const override |
| | get the property More...
|
| |
| const Gaudi::Details::PropertyBase & | getProperty (std::string_view name) const override |
| | get the property by name More...
|
| |
| StatusCode | getProperty (std::string_view 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 (std::string_view name) const override |
| | Return true if we have a property with the given name. More...
|
| |
| Gaudi::Details::PropertyBase * | property (std::string_view name) const |
| | \fixme property and bindPropertiesTo should be protected More...
|
| |
| void | bindPropertiesTo (Gaudi::Interfaces::IOptionsSvc &optsSvc) |
| |
| | PropertyHolder (const PropertyHolder &)=delete |
| |
| PropertyHolder & | operator= (const PropertyHolder &)=delete |
| |
| MSG::Level | msgLevel () const |
| | get the cached level (originally extracted from the embedded MsgStream) More...
|
| |
| bool | msgLevel (MSG::Level lvl) const |
| | get the output level from the embedded MsgStream More...
|
| |
|
| std::chrono::duration< int64_t, std::milli > | m_snapshotInterval = std::chrono::duration<int64_t, std::milli>::min() |
| |
| std::chrono::system_clock::time_point | m_lastSnapshot = std::chrono::system_clock::now() |
| |
| std::function< void(OccupancySnapshot)> | m_snapshotCallback |
| |
| Gaudi::Property< int > | m_threadPoolSize |
| |
| Gaudi::Property< int > | m_maxParallelismExtra |
| |
| Gaudi::Property< std::string > | m_whiteboardSvcName { this, "WhiteboardSvc", "EventDataSvc", "The whiteboard name" } |
| |
| Gaudi::Property< unsigned int > | m_maxBlockingAlgosInFlight |
| |
| Gaudi::Property< bool > | m_simulateExecution |
| |
| Gaudi::Property< std::string > | m_optimizationMode |
| |
| Gaudi::Property< bool > | m_dumpIntraEventDynamics |
| |
| Gaudi::Property< bool > | m_enablePreemptiveBlockingTasks |
| |
| Gaudi::Property< int > | m_numOffloadThreads |
| |
| Gaudi::Property< bool > | m_checkDeps |
| |
| Gaudi::Property< bool > | m_checkOutput |
| |
| Gaudi::Property< std::vector< std::string > > | m_checkOutputIgnoreList |
| |
| Gaudi::Property< std::string > | m_useDataLoader |
| |
| Gaudi::Property< bool > | m_enableCondSvc { this, "EnableConditions", false, "Enable ConditionsSvc" } |
| |
| Gaudi::Property< bool > | m_showDataDeps |
| |
| Gaudi::Property< bool > | m_showDataFlow |
| |
| Gaudi::Property< bool > | m_showControlFlow |
| |
| Gaudi::Property< bool > | m_verboseSubSlots { this, "VerboseSubSlots", false, "Dump algorithm states for all sub-slots" } |
| |
| Gaudi::Property< std::string > | m_dataDepsGraphFile |
| |
| Gaudi::Property< std::string > | m_dataDepsGraphAlgoPattern |
| |
| Gaudi::Property< std::string > | m_dataDepsGraphObjectPattern |
| |
| std::atomic< ActivationState > | m_isActive { INACTIVE } |
| | Flag to track if the scheduler is active or not. More...
|
| |
| std::thread | m_thread |
| | The thread in which the activate function runs. More...
|
| |
| std::unordered_map< std::string, unsigned int > | m_algname_index_map |
| | Map to bookkeep the information necessary to the name2index conversion. More...
|
| |
| std::vector< std::string > | m_algname_vect |
| | Vector to bookkeep the information necessary to the index2name conversion. More...
|
| |
| SmartIF< IPrecedenceSvc > | m_precSvc |
| | A shortcut to the Precedence Service. More...
|
| |
| SmartIF< IHiveWhiteBoard > | m_whiteboard |
| | A shortcut to the whiteboard. More...
|
| |
| std::vector< EventSlot > | m_eventSlots |
| | Vector of events slots. More...
|
| |
| std::atomic_int | m_freeSlots { 0 } |
| | Atomic to account for asyncronous updates by the scheduler wrt the rest. More...
|
| |
| tbb::concurrent_bounded_queue< EventContext * > | m_finishedEvents |
| | Queue of finished events. More...
|
| |
| SmartIF< IAlgExecStateSvc > | m_algExecStateSvc |
| | Algorithm execution state manager. More...
|
| |
| SmartIF< ICondSvc > | m_condSvc |
| | A shortcut to service for Conditions handling. More...
|
| |
| unsigned int | m_algosInFlight = 0 |
| | Number of algorithms presently in flight. More...
|
| |
| unsigned int | m_blockingAlgosInFlight = 0 |
| | Number of algorithms presently in flight. More...
|
| |
| SmartIF< IAlgResourcePool > | m_algResourcePool |
| | Cache for the algorithm resource pool. More...
|
| |
| tbb::concurrent_bounded_queue< action > | m_actionsQueue |
| | Queue where closures are stored and picked for execution. More...
|
| |
| tbb::concurrent_priority_queue< TaskSpec, AlgQueueSort > | m_scheduledQueue |
| | Queues for scheduled algorithms. More...
|
| |
| tbb::concurrent_priority_queue< TaskSpec, AlgQueueSort > | m_scheduledAsynchronousQueue |
| |
| std::queue< TaskSpec > | m_retryQueue |
| |
| std::atomic< bool > | m_needsUpdate { true } |
| |
| SmartIF< IThreadPoolSvc > | m_threadPoolSvc |
| |
| tbb::task_arena * | m_arena { nullptr } |
| |
| std::unique_ptr< FiberManager > | m_fiberManager { nullptr } |
| |
| size_t | m_maxEventsInFlight { 0 } |
| |
| size_t | m_maxAlgosInFlight { 1 } |
| |
|
| using | base_class = extends |
| | Typedef to this class. More...
|
| |
| using | extend_interfaces_base = extend_interfaces< Interfaces... > |
| | Typedef to the base of this class. More...
|
| |
| using | Factory = Gaudi::PluginService::Factory< IService *(const std::string &, ISvcLocator *)> |
| |
| using | PropertyHolderImpl = PropertyHolder< CommonMessaging< implements< IService, IProperty, IStateful > > > |
| | Typedef used to refer to this class from derived classes, as in. More...
|
| |
| using | base_class = CommonMessaging |
| |
| using | ext_iids = typename Gaudi::interface_list_cat< typename Interfaces::ext_iids... >::type |
| | take union of the ext_iids of all Interfaces... More...
|
| |
| std::vector< IAlgTool * > & | tools () |
| |
| | ~Service () override |
| | Standard Destructor
More...
|
| |
| int | outputLevel () const |
| | get the Service's output level More...
|
| |
| MSG::Level | setUpMessaging () const |
| | Set up local caches. More...
|
| |
| MSG::Level | resetMessaging () |
| | Reinitialize internal states. More...
|
| |
| void | updateMsgStreamOutputLevel (int level) |
| | Update the output level of the cached MsgStream. More...
|
| |
| 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" } |
| | flag indicating whether ToolHandle tools have been added to m_tools More...
|
| |
| 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()" } |
| |
| Gaudi::Property< bool > | m_autoRetrieveTools |
| |
| Gaudi::Property< bool > | m_checkToolDeps |
| |
| SmartIF< IAuditorSvc > | m_pAuditorSvc |
| | Auditor Service
More...
|
| |
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 preemptive scheduling CPU-blocking tasks. 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 114 of file AvalancheSchedulerSvc.h.
◆ action
◆ AState
◆ ActivationState
◆ activate()
| void AvalancheSchedulerSvc::activate |
( |
| ) |
|
|
private |
Activate scheduler.
Activate the scheduler.
From this moment on the queue of actions is checked. The checking will stop when the m_isActive flag is false and the queue is not empty. This will guarantee that all actions are executed and a stall is not created. The TBB pool must be initialised in the thread from where the tasks are launched (http://threadingbuildingblocks.org/docs/doxygen/a00342.html) The scheduler is initialised here since this method runs in a separate thread and spawns the tasks (through the execution of the lambdas)
Definition at line 450 of file AvalancheSchedulerSvc.cpp.
452 ON_DEBUG debug() <<
"AvalancheSchedulerSvc::activate()" <<
endmsg;
455 error() <<
"problems initializing ThreadPoolSvc" <<
endmsg;
472 if ( sc.isFailure() )
473 verbose() <<
"Action did not succeed (which is not bad per se)." <<
endmsg;
483 if ( sc.isFailure() )
484 verbose() <<
"Iteration did not succeed (which is not bad per se)." <<
endmsg;
492 ON_DEBUG debug() <<
"Terminating thread-pool resources" <<
endmsg;
494 error() <<
"Problems terminating thread pool" <<
endmsg;
◆ algname2index()
| unsigned int AvalancheSchedulerSvc::algname2index |
( |
const std::string & |
algoname | ) |
|
|
inlineprivate |
◆ deactivate()
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 507 of file AvalancheSchedulerSvc.cpp.
◆ dumpGraphFile()
Definition at line 1170 of file AvalancheSchedulerSvc.cpp.
1173 assert( inDeps.
size() == outDeps.
size() );
1176 enum class FileType : short {
UNKNOWN, DOT, MD };
1183 fileExtension = FileType::DOT;
1185 fileExtension = FileType::MD;
1187 fileExtension = FileType::DOT;
1188 fileName = fileName +
".dot";
1190 info() <<
"Dumping data dependencies graph to file: " << fileName <<
endmsg;
1200 if ( fileExtension == FileType::DOT ) {
1202 startGraph =
"digraph datadeps {\nrankdir=\"LR\";\n\n";
1203 stopGraph =
"\n}\n";
1206 return "Alg_" + idx +
" [label=\"" +
alg +
"\";shape=box];\n";
1210 return "obj_" +
std::to_string( obj.hash() ) +
" [label=\"" + obj.key() +
"\"];\n";
1222 startGraph =
"```mermaid\ngraph LR;\n\n";
1223 stopGraph =
"\n```\n";
1226 return "Alg_" + idx +
"{{" +
alg +
"}}\n";
1230 return "obj_" +
std::to_string( obj.hash() ) +
">" + obj.key() +
"]\n";
1243 dataDepthGraphFile << startGraph;
1254 for (
const auto& [
name, ideps] : inDeps ) {
1259 for (
const auto& dep : ideps ) {
1262 if ( definedObjects.
find( dep.hash() ) == definedObjects.
end() ) {
1263 definedObjects.
insert( dep.hash() );
1264 dataDepthGraphFile << defineObj( dep );
1266 dataDepthGraphFile << defineInput( dep,
std::to_string( algoIndex ) );
1269 const auto& odeps = outDeps.
at(
name );
1270 for (
const auto& dep : odeps ) {
1273 if ( definedObjects.
find( dep.hash() ) == definedObjects.
end() ) {
1274 definedObjects.
insert( dep.hash() );
1275 dataDepthGraphFile << defineObj( dep );
1277 dataDepthGraphFile << defineOutput(
std::to_string( algoIndex ), dep );
1284 dataDepthGraphFile << stopGraph;
1285 dataDepthGraphFile.close();
◆ dumpSchedulerState()
| void AvalancheSchedulerSvc::dumpSchedulerState |
( |
int |
iSlot | ) |
|
|
private |
Dump the state of the scheduler.
Used for debugging purposes, the state of the scheduler is dumped on screen in order to be inspected.
Definition at line 872 of file AvalancheSchedulerSvc.cpp.
877 outputMS <<
"Dumping scheduler state\n"
878 <<
"=========================================================================================\n"
879 <<
"++++++++++++++++++++++++++++++++++++ SCHEDULER STATE ++++++++++++++++++++++++++++++++++++\n"
880 <<
"=========================================================================================\n\n";
884 outputMS <<
"------------------ Last schedule: Task/Event/Slot/Thread/State Mapping "
885 <<
"------------------\n\n";
889 if ( !timelineSvc.isValid() || !timelineSvc->isEnabled() ) {
890 outputMS <<
"WARNING Enable TimelineSvc in record mode (RecordTimeline = True) to trace the mapping\n";
897 const auto& schedAlgs = slot.algsStates.algsInState( AState::SCHEDULED );
898 for ( uint algIndex : schedAlgs ) {
906 const auto& schedAlgs = slot.algsStates.algsInState( AState::SCHEDULED );
907 for ( uint algIndex : schedAlgs ) {
911 outputMS <<
" task: " <<
std::setw( indt ) << algoName <<
" evt/slot: " << slot.eventContext->evt() <<
"/"
912 << slot.eventContext->slot();
915 if ( timelineSvc.isValid() ) {
918 te.slot = slot.eventContext->slot();
919 te.event = slot.eventContext->evt();
921 if ( timelineSvc->getTimelineEvent( te ) )
924 outputMS <<
" thread.id: [unknown]";
929 outputMS <<
" state: [" <<
m_algExecStateSvc->algExecState( algoName, *( slot.eventContext ) ) <<
"]\n";
936 outputMS <<
"\n---------------------------- Task/CF/FSM Mapping "
937 << ( 0 > iSlot ?
"[all slots] --" :
"[target slot] " ) <<
"--------------------------\n\n";
946 if ( slot.complete )
continue;
948 outputMS <<
"[ slot: "
949 << ( slot.eventContext->valid() ?
std::to_string( slot.eventContext->slot() ) :
"[ctx invalid]" )
951 << ( slot.eventContext->
valid() ?
std::
to_string( slot.eventContext->
evt() ) :
"[ctx invalid]" );
953 if ( slot.eventContext->eventID().isValid() ) { outputMS <<
", eventID: " << slot.eventContext->eventID(); }
954 outputMS <<
" ]:\n\n";
956 if ( 0 > iSlot || iSlot == slotCount ) {
960 outputMS <<
"ERROR alg(s):";
962 const auto& errorAlgs = slot.algsStates.algsInState(
AState::ERROR );
963 for ( uint algIndex : errorAlgs ) {
967 if ( errorCount == 0 ) outputMS <<
" in subslot(s)";
971 outputMS <<
m_precSvc->printState( slot ) <<
"\n";
976 outputMS <<
"\nNumber of sub-slots: " << slot.allSubSlots.size() <<
"\n\n";
977 auto slotID = slot.eventContext->valid() ?
std::to_string( slot.eventContext->slot() ) :
"[ctx invalid]";
978 for (
auto& ss : slot.allSubSlots ) {
979 outputMS <<
"[ slot: " << slotID <<
", sub-slot: "
980 << ( ss.eventContext->valid() ?
std::to_string( ss.eventContext->subSlot() ) :
"[ctx invalid]" )
981 <<
", entry: " << ss.entryPoint <<
", event: "
985 outputMS <<
"ERROR alg(s):";
986 const auto& errorAlgs = ss.algsStates.algsInState(
AState::ERROR );
987 for ( uint algIndex : errorAlgs ) { outputMS <<
" " <<
index2algname( algIndex ); }
991 outputMS <<
m_precSvc->printState( ss ) <<
"\n";
1000 if ( 0 <= iSlot && !wasAlgError ) {
1001 outputMS <<
"\n------------------------------ Algorithm Execution States -----------------------------\n\n";
1005 outputMS <<
"\n=========================================================================================\n"
1006 <<
"++++++++++++++++++++++++++++++++++++++ END OF DUMP ++++++++++++++++++++++++++++++++++++++\n"
1007 <<
"=========================================================================================\n\n";
◆ dumpState()
| void AvalancheSchedulerSvc::dumpState |
( |
| ) |
|
|
override |
◆ eventFailed()
| void AvalancheSchedulerSvc::eventFailed |
( |
EventContext * |
eventContext | ) |
|
|
private |
Method to execute if an event failed.
It can be possible that an event fails.
In this case this method is called. It dumps the state of the scheduler and marks the event as finished.
Definition at line 851 of file AvalancheSchedulerSvc.cpp.
852 const uint slotIdx = eventContext->
slot();
854 error() <<
"Event " << eventContext->
evt() <<
" on slot " << slotIdx <<
" failed" <<
endmsg;
◆ finalize()
Finalise.
Here the scheduler is deactivated and the thread joined.
Definition at line 416 of file AvalancheSchedulerSvc.cpp.
419 if ( sc.isFailure() ) warning() <<
"Base class could not be finalized" <<
endmsg;
422 if ( sc.isFailure() ) warning() <<
"Scheduler could not be deactivated" <<
endmsg;
424 debug() <<
"Deleting FiberManager" <<
endmsg;
427 info() <<
"Joining Scheduler thread" <<
endmsg;
432 error() <<
"problems in scheduler thread" <<
endmsg;
◆ freeSlots()
| unsigned int AvalancheSchedulerSvc::freeSlots |
( |
| ) |
|
|
override |
◆ index2algname()
| const std::string& AvalancheSchedulerSvc::index2algname |
( |
unsigned int |
index | ) |
|
|
inlineprivate |
◆ initialize()
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 76 of file AvalancheSchedulerSvc.cpp.
80 if ( sc.isFailure() ) warning() <<
"Base class could not be initialized" <<
endmsg;
85 fatal() <<
"Error retrieving ThreadPoolSvc" <<
endmsg;
90 fatal() <<
"Cannot cast ThreadPoolSvc" <<
endmsg;
95 fatal() <<
"Cannot find valid TBB task_arena" <<
endmsg;
103 info() <<
"Activating scheduler in a separate thread" <<
endmsg;
108 fatal() <<
"Terminating initialization" <<
endmsg;
111 ON_DEBUG debug() <<
"Waiting for AvalancheSchedulerSvc to activate" <<
endmsg;
120 warning() <<
"No CondSvc found, or not enabled. "
121 <<
"Will not manage CondAlgorithms" <<
endmsg;
129 fatal() <<
"Error retrieving AlgoResourcePool" <<
endmsg;
135 fatal() <<
"Error retrieving AlgExecStateSvc" <<
endmsg;
142 fatal() <<
"Error retrieving EventDataSvc interface IHiveWhiteBoard." <<
endmsg;
154 const unsigned int algsNumber = algos.
size();
155 if ( algsNumber != 0 ) {
156 info() <<
"Found " << algsNumber <<
" algorithms" <<
endmsg;
158 error() <<
"No algorithms found" <<
endmsg;
174 fatal() <<
"Could not convert IAlgorithm into Gaudi::Algorithm: this will result in a crash." <<
endmsg;
183 algosOutputDependenciesMap[algoPtr->
name()] = algoOutputs;
187 ostdd <<
"Data Dependencies for Algorithms:";
192 if (
nullptr == algoPtr ) {
193 fatal() <<
"Could not convert IAlgorithm into Gaudi::Algorithm for " << ialgoPtr->
name()
194 <<
": this will result in a crash." <<
endmsg;
202 ostdd <<
"\n " << algoPtr->
name();
204 auto write_owners = [&avis, &ostdd](
const DataObjID id ) {
205 auto owners = avis.owners_names_of(
id );
213 ostdd <<
"\n o INPUT " <<
id;
215 algoDependencies.
insert(
id );
219 ostdd <<
"\n o OUTPUT " << *
id;
221 if (
id->key().find(
":" ) != std::string::npos ) {
222 error() <<
" in Alg " << algoPtr->
name() <<
" alternatives are NOT allowed for outputs! id: " << *
id
230 algosInputDependenciesMap[algoPtr->
name()] = algoDependencies;
237 if (
dumpGraphFile( algosInputDependenciesMap, algosOutputDependenciesMap ).isFailure() ) {
247 for (
auto o : globalInp ) {
250 requiredInputKeys.
insert( o.key() );
251 if ( globalOutp.
find( o ) == globalOutp.
end() ) unmetDepInp.
insert( o );
254 for (
auto o : globalOutp ) {
255 if ( globalInp.find( o ) == globalInp.end() && requiredInputKeys.
find( o.key() ) == requiredInputKeys.
end() ) {
259 auto it = algosOutputDependenciesMap.
find( algoName );
260 if ( it != algosOutputDependenciesMap.
end() ) {
261 if ( it->second.find( o ) != it->second.end() ) {
267 if ( !ignored ) { unusedOutp.
insert( o ); }
274 if ( unmetDepInp.
size() > 0 ) {
276 auto printUnmet = [&](
auto msg ) {
277 for (
const DataObjID* o : sortedDataObjIDColl( unmetDepInp ) ) {
278 msg <<
" o " << *o <<
" required by Algorithm: " <<
endmsg;
280 for (
const auto& p : algosInputDependenciesMap )
281 if ( p.second.find( *o ) != p.second.end() )
msg <<
" * " << p.first <<
endmsg;
291 dataLoaderAlg = algo;
295 if ( dataLoaderAlg ==
nullptr ) {
297 <<
"\" found, and unmet INPUT dependencies "
299 printUnmet( fatal() );
303 info() <<
"Will attribute the following unmet INPUT dependencies to \"" << dataLoaderAlg->type() <<
"/"
304 << dataLoaderAlg->name() <<
"\" Algorithm" <<
endmsg;
305 printUnmet( info() );
310 fatal() <<
"Unable to dcast DataLoader \"" <<
m_useDataLoader.
value() <<
"\" IAlg to Gaudi::Algorithm"
315 for (
auto&
id : unmetDepInp ) {
316 ON_DEBUG debug() <<
"adding OUTPUT dep \"" <<
id <<
"\" to " << dataLoaderAlg->type() <<
"/"
317 << dataLoaderAlg->name() <<
endmsg;
322 fatal() <<
"Auto DataLoading not requested, "
323 <<
"and the following unmet INPUT dependencies were found:" <<
endmsg;
324 printUnmet( fatal() );
329 info() <<
"No unmet INPUT data dependencies were found" <<
endmsg;
334 if ( unusedOutp.
size() > 0 ) {
336 auto printUnusedOutp = [&](
auto msg ) {
337 for (
const DataObjID* o : sortedDataObjIDColl( unusedOutp ) ) {
338 msg <<
" o " << *o <<
" produced by Algorithm: " <<
endmsg;
340 for (
const auto& p : algosOutputDependenciesMap )
341 if ( p.second.find( *o ) != p.second.end() )
msg <<
" * " << p.first <<
endmsg;
345 fatal() <<
"The following unused OUTPUT items were found:" <<
endmsg;
346 printUnusedOutp( fatal() );
349 info() <<
"No unused OUTPUT items were found" <<
endmsg;
356 fatal() <<
"Error retrieving PrecedenceSvc" <<
endmsg;
361 fatal() <<
"Unable to dcast PrecedenceSvc" <<
endmsg;
376 if ( !messageSvc.isValid() ) error() <<
"Error retrieving MessageSvc interface IMessageSvc." <<
endmsg;
387 info() <<
"Concurrency level information:" <<
endmsg;
392 info() <<
"Task scheduling settings:" <<
endmsg;
393 info() <<
" o Avalanche generation mode: "
395 info() <<
" o Preemptive scheduling of CPU-blocking tasks: "
400 info() <<
" o Scheduling of condition tasks: " << (
m_enableCondSvc ?
"enabled" :
"disabled" ) <<
endmsg;
◆ isStalled()
| bool AvalancheSchedulerSvc::isStalled |
( |
const EventSlot & |
slot | ) |
const |
|
private |
Check if scheduling in a particular slot is in a stall.
Check if we are in present of a stall condition for a particular slot.
This is the case when a slot has no actions queued in the actionsQueue, has no scheduled algorithms and has no algorithms with all of its dependencies satisfied.
Definition at line 833 of file AvalancheSchedulerSvc.cpp.
835 if ( !slot.
algsStates.
containsAny( { AState::DATAREADY, AState::SCHEDULED, AState::RESOURCELESS } ) &&
836 !subSlotAlgsInStates( slot, { AState::DATAREADY, AState::SCHEDULED, AState::RESOURCELESS } ) ) {
◆ iterate()
Loop on all slots to schedule DATAREADY algorithms and sign off ready events.
Loop on all slots to schedule DATAREADY algorithms, sign off ready ones or detect execution stalls.
To check if an event is finished the method verifies that the root control flow decision of the task precedence graph is resolved and there are no algorithms moving in-between INITIAL and EVTACCEPTED FSM states.
Definition at line 658 of file AvalancheSchedulerSvc.cpp.
664 for (
unsigned int retryIndex = 0; retryIndex < retries; ++retryIndex ) {
671 OccupancySnapshot nextSnap;
676 if ( !thisSlot.eventContext )
continue;
678 int iSlot = thisSlot.eventContext->slot();
690 if ( nextSnap.states.empty() ) {
697 slotStateTotals.
resize( AState::MAXVALUE );
699 slotStateTotals[
state] = thisSlot.algsStates.sizeOfSubset(
AState(
state ) );
703 for (
auto& subslot : thisSlot.allSubSlots ) {
705 slotStateTotals[
state] += subslot.algsStates.sizeOfSubset(
AState(
state ) );
711 const auto& drAlgs = thisAlgsStates.
algsInState( AState::DATAREADY );
712 for ( uint algIndex : drAlgs ) {
715 bool asynchronous{
m_precSvc->isAsynchronous( algName ) };
718 schedule( TaskSpec(
nullptr, algIndex, algName, rank, asynchronous, iSlot, thisSlot.eventContext.get() ) );
721 <<
"Could not apply transition from " << AState::DATAREADY <<
" for algorithm " << algName
722 <<
" on processing slot " << iSlot <<
endmsg;
726 for (
auto& subslot : thisSlot.allSubSlots ) {
727 const auto& drAlgsSubSlot = subslot.algsStates.algsInState( AState::DATAREADY );
728 for ( uint algIndex : drAlgsSubSlot ) {
731 bool asynchronous{
m_precSvc->isAsynchronous( algName ) };
733 schedule( TaskSpec(
nullptr, algIndex, algName, rank, asynchronous, iSlot, subslot.eventContext.get() ) );
739 s <<
"START, " << thisAlgsStates.
sizeOfSubset( AState::CONTROLREADY ) <<
", "
751 if (
m_precSvc->CFRulesResolved( thisSlot ) &&
752 !thisSlot.algsStates.containsAny(
753 { AState::CONTROLREADY, AState::DATAREADY, AState::SCHEDULED, AState::RESOURCELESS } ) &&
754 !subSlotAlgsInStates( thisSlot,
755 { AState::CONTROLREADY, AState::DATAREADY, AState::SCHEDULED, AState::RESOURCELESS } ) &&
756 !thisSlot.complete ) {
758 thisSlot.complete =
true;
762 ON_DEBUG debug() <<
"Event " << thisSlot.eventContext->evt() <<
" finished (slot "
763 << thisSlot.eventContext->slot() <<
")." <<
endmsg;
770 thisSlot.eventContext.reset(
nullptr );
780 if ( !nextSnap.states.empty() ) {
◆ next()
| bool AvalancheSchedulerSvc::next |
( |
TaskSpec & |
ts, |
|
|
bool |
asynchronous |
|
) |
| |
|
inline |
◆ popFinishedEvent()
Blocks until an event is available.
Get a finished event or block until one becomes available.
Definition at line 617 of file AvalancheSchedulerSvc.cpp.
629 ON_DEBUG debug() <<
"Popped slot " << eventContext->
slot() <<
" (event " << eventContext->
evt() <<
")" <<
endmsg;
◆ pushNewEvent()
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 538 of file AvalancheSchedulerSvc.cpp.
540 if ( !eventContext ) {
541 fatal() <<
"Event context is nullptr" <<
endmsg;
546 ON_DEBUG debug() <<
"A free processing slot could not be found." <<
endmsg;
555 const unsigned int thisSlotNum = eventContext->
slot();
558 fatal() <<
"The slot " << thisSlotNum <<
" is supposed to be a finished event but it's not" <<
endmsg;
562 ON_DEBUG debug() <<
"Executing event " << eventContext->
evt() <<
" on slot " << thisSlotNum <<
endmsg;
563 thisSlot.
reset( eventContext );
570 if (
m_precSvc->iterate( thisSlot, cs ).isFailure() ) {
571 error() <<
"Failed to call IPrecedenceSvc::iterate for slot " << thisSlotNum <<
endmsg;
575 if ( this->
iterate().isFailure() ) {
576 error() <<
"Failed to call AvalancheSchedulerSvc::updateStates for slot " << thisSlotNum <<
endmsg;
585 verbose() <<
"Pushing the action to update the scheduler for slot " << eventContext->
slot() <<
endmsg;
◆ pushNewEvents()
◆ recordOccupancy()
| void AvalancheSchedulerSvc::recordOccupancy |
( |
int |
samplePeriod, |
|
|
std::function< void(OccupancySnapshot)> |
callback |
|
) |
| |
|
overridevirtual |
Sample occupancy at fixed interval (ms) Negative value to deactivate, 0 to snapshot every change Each sample, apply the callback function to the result.
Definition at line 1155 of file AvalancheSchedulerSvc.cpp.
1158 if ( samplePeriod < 0 ) {
◆ revise()
Definition at line 792 of file AvalancheSchedulerSvc.cpp.
794 auto slotIndex = contextPtr->
slot();
800 auto subSlotIndex = contextPtr->
subSlot();
807 <<
", subslot:" << subSlotIndex <<
", event:" << contextPtr->
evt() <<
"]" <<
endmsg;
817 <<
", event:" << contextPtr->
evt() <<
"]" <<
endmsg;
◆ schedule()
Definition at line 1014 of file AvalancheSchedulerSvc.cpp.
1021 if ( getAlgSC.isSuccess() ) {
1027 unsigned int algIndex{
ts.algIndex };
1028 std::string_view algName(
ts.algName );
1029 unsigned int algRank{
ts.algRank };
1030 bool asynchronous{
ts.asynchronous };
1031 int slotIndex{
ts.slotIndex };
1034 if ( asynchronous ) {
1042 if ( !asynchronous ) {
1050 sc =
revise( algIndex, contextPtr, AState::SCHEDULED );
1052 ON_DEBUG debug() <<
"Scheduled " << algName <<
" [slot:" << slotIndex <<
", event:" << contextPtr->evt()
1053 <<
", rank:" << algRank <<
", asynchronous:" << ( asynchronous ?
"yes" :
"no" )
1063 sc =
revise(
ts.algIndex,
ts.contextPtr, AState::SCHEDULED );
1069 sc =
revise(
ts.algIndex,
ts.contextPtr, AState::RESOURCELESS );
◆ scheduleEventView()
Method to inform the scheduler about event views.
Definition at line 1115 of file AvalancheSchedulerSvc.cpp.
1119 fatal() <<
"Attempted to nest EventViews at node " << nodeName <<
": this is not supported" <<
endmsg;
1127 auto action = [
this, slotIndex = sourceContext->
slot(), viewContextPtr = viewContext.
release(),
1132 if ( viewContextPtr ) {
◆ signoff()
The call to this method is triggered only from within the AlgTask.
Definition at line 1084 of file AvalancheSchedulerSvc.cpp.
1092 ? ( algstate.
filterPassed() ? AState::EVTACCEPTED : AState::EVTREJECTED )
1098 ON_DEBUG debug() <<
"Executed " <<
ts.algName <<
" [slot:" <<
ts.slotIndex <<
", event:" <<
ts.contextPtr->evt()
1099 <<
", rank:" <<
ts.algRank <<
", asynchronous:" << (
ts.asynchronous ?
"yes" :
"no" )
◆ tryPopFinishedEvent()
Try to fetch an event from the scheduler.
Try to get a finished event, if not available just return a failure.
Definition at line 638 of file AvalancheSchedulerSvc.cpp.
641 ON_DEBUG debug() <<
"Try Pop successful slot " << eventContext->
slot() <<
"(event " << eventContext->
evt() <<
")"
◆ AlgTask
◆ m_actionsQueue
| tbb::concurrent_bounded_queue<action> AvalancheSchedulerSvc::m_actionsQueue |
|
private |
◆ m_algExecStateSvc
◆ m_algname_index_map
◆ m_algname_vect
Vector to bookkeep the information necessary to the index2name conversion.
Definition at line 262 of file AvalancheSchedulerSvc.h.
◆ m_algosInFlight
| unsigned int AvalancheSchedulerSvc::m_algosInFlight = 0 |
|
private |
◆ m_algResourcePool
◆ m_arena
| tbb::task_arena* AvalancheSchedulerSvc::m_arena { nullptr } |
|
private |
◆ m_blockingAlgosInFlight
| unsigned int AvalancheSchedulerSvc::m_blockingAlgosInFlight = 0 |
|
private |
◆ m_checkDeps
Initial value:{ this, "CheckDependencies", false,
"Runtime check of Algorithm Input Data Dependencies" }
Definition at line 197 of file AvalancheSchedulerSvc.h.
◆ m_checkOutput
Initial value:{ this, "CheckOutputUsage", false,
"Runtime check of Algorithm Output Data usage" }
Definition at line 199 of file AvalancheSchedulerSvc.h.
◆ m_checkOutputIgnoreList
Initial value:{
this,
"CheckOutputUsageIgnoreList",
{},
"Ignore outputs of the Algorithms of this name when doing the check",
"OrderedSet<std::string>" }
Definition at line 201 of file AvalancheSchedulerSvc.h.
◆ m_condSvc
◆ m_dataDepsGraphAlgoPattern
Initial value:{
this, "DataDepsGraphAlgPattern", ".*",
"Regex pattern for selecting desired Algorithms by name, whose data dependency has to be included in the data "
"deps graph" }
Definition at line 229 of file AvalancheSchedulerSvc.h.
◆ m_dataDepsGraphFile
Initial value:{
this, "DataDepsGraphFile", "",
"Name of the output file (.dot or .md extensions allowed) containing the data dependency graph for some selected "
"Algorithms" }
Definition at line 224 of file AvalancheSchedulerSvc.h.
◆ m_dataDepsGraphObjectPattern
Initial value:{
this, "DataDepsGraphObjectPattern", ".*",
"Regex pattern for selecting desired input or output by their full key" }
Definition at line 234 of file AvalancheSchedulerSvc.h.
◆ m_dumpIntraEventDynamics
Initial value:{ this, "DumpIntraEventDynamics", false,
"Dump intra-event concurrency dynamics to csv file" }
Definition at line 187 of file AvalancheSchedulerSvc.h.
◆ m_enableCondSvc
| Gaudi::Property<bool> AvalancheSchedulerSvc::m_enableCondSvc { this, "EnableConditions", false, "Enable ConditionsSvc" } |
|
private |
◆ m_enablePreemptiveBlockingTasks
| Gaudi::Property<bool> AvalancheSchedulerSvc::m_enablePreemptiveBlockingTasks |
|
private |
Initial value:{
this, "PreemptiveBlockingTasks", false,
"Enable preemptive scheduling of CPU-blocking algorithms. Blocking algorithms must be flagged accordingly." }
Definition at line 189 of file AvalancheSchedulerSvc.h.
◆ m_eventSlots
◆ m_fiberManager
◆ m_finishedEvents
| tbb::concurrent_bounded_queue<EventContext*> AvalancheSchedulerSvc::m_finishedEvents |
|
private |
◆ m_freeSlots
| std::atomic_int AvalancheSchedulerSvc::m_freeSlots { 0 } |
|
private |
◆ m_isActive
◆ m_lastSnapshot
◆ m_maxAlgosInFlight
| size_t AvalancheSchedulerSvc::m_maxAlgosInFlight { 1 } |
|
private |
◆ m_maxBlockingAlgosInFlight
| Gaudi::Property<unsigned int> AvalancheSchedulerSvc::m_maxBlockingAlgosInFlight |
|
private |
Initial value:{
this, "MaxBlockingAlgosInFlight", 0, "Maximum allowed number of simultaneously running CPU-blocking algorithms" }
Definition at line 180 of file AvalancheSchedulerSvc.h.
◆ m_maxEventsInFlight
| size_t AvalancheSchedulerSvc::m_maxEventsInFlight { 0 } |
|
private |
◆ m_maxParallelismExtra
Initial value:{
this, "maxParallelismExtra", 0,
"Allows to add some extra threads to the maximum parallelism set in TBB"
"The TBB max parallelism is set as: ThreadPoolSize + maxParallelismExtra + 1" }
Definition at line 175 of file AvalancheSchedulerSvc.h.
◆ m_needsUpdate
| std::atomic<bool> AvalancheSchedulerSvc::m_needsUpdate { true } |
|
private |
◆ m_numOffloadThreads
Initial value:{
this, "NumOffloadThreads", 2,
"Number of threads to use for CPU portion of asynchronous algorithms. Asynchronous algorithms must be flagged "
"and "
"use Boost Fiber functionality to suspend while waiting for offloaded work." }
Definition at line 192 of file AvalancheSchedulerSvc.h.
◆ m_optimizationMode
Initial value:{ this, "Optimizer", "",
"The following modes are currently available: PCE, COD, DRE, E" }
Definition at line 185 of file AvalancheSchedulerSvc.h.
◆ m_precSvc
◆ m_retryQueue
◆ m_scheduledAsynchronousQueue
| tbb::concurrent_priority_queue<TaskSpec, AlgQueueSort> AvalancheSchedulerSvc::m_scheduledAsynchronousQueue |
|
private |
◆ m_scheduledQueue
◆ m_showControlFlow
Initial value:{ this, "ShowControlFlow", false,
"Show the configuration of all Algorithms and Sequences" }
Definition at line 219 of file AvalancheSchedulerSvc.h.
◆ m_showDataDeps
Initial value:{ this, "ShowDataDependencies", true,
"Show the INPUT and OUTPUT data dependencies of Algorithms" }
Definition at line 213 of file AvalancheSchedulerSvc.h.
◆ m_showDataFlow
Initial value:{ this, "ShowDataFlow", false,
"Show the configuration of DataFlow between Algorithms" }
Definition at line 216 of file AvalancheSchedulerSvc.h.
◆ m_simulateExecution
Initial value:{
this, "SimulateExecution", false,
"Flag to perform single-pass simulation of execution flow before the actual execution" }
Definition at line 182 of file AvalancheSchedulerSvc.h.
◆ m_snapshotCallback
| std::function<void( OccupancySnapshot )> AvalancheSchedulerSvc::m_snapshotCallback |
|
private |
◆ m_snapshotInterval
◆ m_thread
◆ m_threadPoolSize
Initial value:{
this, "ThreadPoolSize", -1,
"Size of the global thread pool initialised by TBB; a value of -1 requests to use"
"all available hardware threads; -100 requests to bypass TBB executing "
"all algorithms in the scheduler's thread." }
Definition at line 170 of file AvalancheSchedulerSvc.h.
◆ m_threadPoolSvc
◆ m_useDataLoader
Initial value:{ this, "DataLoaderAlg", "",
"Attribute unmet input dependencies to this DataLoader Algorithm" }
Definition at line 208 of file AvalancheSchedulerSvc.h.
◆ m_verboseSubSlots
| Gaudi::Property<bool> AvalancheSchedulerSvc::m_verboseSubSlots { this, "VerboseSubSlots", false, "Dump algorithm states for all sub-slots" } |
|
private |
◆ m_whiteboard
◆ m_whiteboardSvcName
The documentation for this class was generated from the following files:
std::unique_ptr< EventContext > eventContext
Cache for the eventContext.
SmartIF< IHiveWhiteBoard > m_whiteboard
A shortcut to the whiteboard.
GAUDI_API void setCurrentContext(const EventContext *ctx)
A service to resolve the task execution precedence.
const std::string name() const
property name
StatusCode initialize() override
Gaudi::Property< std::string > m_useDataLoader
void acceptDHVisitor(IDataHandleVisitor *) const override
const std::string & name() const override
The identifying name of the algorithm object.
Gaudi::Property< std::string > m_optimizationMode
unsigned int getControlFlowNodeCounter() const
Get total number of control flow graph nodes.
StatusCode iterate()
Loop on all slots to schedule DATAREADY algorithms and sign off ready events.
Class representing an event slot.
void addDependency(const DataObjID &id, const Gaudi::DataHandle::Mode &mode) override
tbb::concurrent_priority_queue< TaskSpec, AlgQueueSort > m_scheduledAsynchronousQueue
std::chrono::system_clock::time_point m_lastSnapshot
const concurrency::PrecedenceRulesGraph * getRules() const
Precedence rules accessor.
Gaudi::Property< std::string > m_dataDepsGraphAlgoPattern
tbb::concurrent_priority_queue< TaskSpec, AlgQueueSort > m_scheduledQueue
Queues for scheduled algorithms.
std::unique_ptr< FiberManager > m_fiberManager
StatusCode schedule(TaskSpec &&)
Gaudi::Property< bool > m_showControlFlow
std::atomic< bool > m_needsUpdate
Gaudi::Property< bool > m_enableCondSvc
StatusCode deactivate()
Deactivate scheduler.
MSG::Level msgLevel() const
get the cached level (originally extracted from the embedded MsgStream)
StatusCode finalize() override
std::vector< EventSlot > m_eventSlots
Vector of events slots.
unsigned int getAlgoIndex() const
Get algorithm index.
Gaudi::Property< int > m_numOffloadThreads
tbb::task_arena * m_arena
SmartIF< IAlgExecStateSvc > m_algExecStateSvc
Algorithm execution state manager.
bool complete
Flags completion of the event.
SmartIF< ICondSvc > m_condSvc
A shortcut to service for Conditions handling.
void eventFailed(EventContext *eventContext)
Method to execute if an event failed.
Gaudi::Property< int > m_threadPoolSize
void addSubSlot(std::unique_ptr< EventContext > viewContext, const std::string &nodeName)
Add a subslot to the slot (this constructs a new slot and registers it with the parent one)
Gaudi::Property< std::string > m_dataDepsGraphObjectPattern
size_t m_maxEventsInFlight
bool isValid() const
Allow for check if smart pointer is valid.
Gaudi::Property< unsigned int > m_maxBlockingAlgosInFlight
std::ostream & operator<<(std::ostream &s, const std::pair< T1, T2 > &p)
Serialize an std::pair in a python like format. E.g. "(1, 2)".
const std::string & name() const override
Retrieve name of the service
T hardware_concurrency(T... args)
Gaudi::Property< int > m_maxParallelismExtra
Gaudi::Property< bool > m_enablePreemptiveBlockingTasks
void schedule(F &&func)
Schedule work to run on the asynchronous pool.
Base class from which all concrete algorithm classes should be derived.
Gaudi::Property< std::string > m_whiteboardSvcName
Gaudi::Property< bool > m_checkOutput
void reset(EventContext *theeventContext)
Reset all resources in order to reuse the slot (thread-unsafe)
const ValueType & value() const
void disableSubSlots(const std::string &nodeName)
Disable event views for a given CF view node by registering an empty container Contact B.
const StatusCode & execStatus() const
Gaudi::Property< bool > m_simulateExecution
const std::string & index2algname(unsigned int index)
Convert an integer to a name.
std::vector< EventSlot > allSubSlots
Actual sub-slot instances.
AlgsExecutionStates::State AState
unsigned int m_algosInFlight
Number of algorithms presently in flight.
MsgStream & endmsg(MsgStream &s)
MsgStream Modifier: endmsg. Calls the output method of the MsgStream.
SmartIF< IAlgResourcePool > m_algResourcePool
Cache for the algorithm resource pool.
Gaudi::Property< bool > m_showDataDeps
size_t m_maxAlgosInFlight
T regex_search(T... args)
bool containsAny(std::initializer_list< State > l) const
check if the collection contains at least one state of any listed types
const StatusCode & ignore() const
Allow discarding a StatusCode without warning.
T emplace_back(T... args)
AlgorithmNode * getAlgorithmNode(const std::string &algoName) const
Get the AlgorithmNode from by algorithm name using graph index.
Gaudi::Property< bool > m_dumpIntraEventDynamics
StatusCode set(unsigned int iAlgo, State newState)
std::queue< TaskSpec > m_retryQueue
constexpr static const auto SUCCESS
ContextID_t subSlot() const
TYPE * get() const
Get interface pointer.
const DataObjIDColl & outputDataObjs() const override
std::chrono::duration< int64_t, std::milli > m_snapshotInterval
bool valid(Iterator begin, Iterator end)
check the validness of the trees or nodes
SmartIF< IThreadPoolSvc > m_threadPoolSvc
Gaudi::Property< std::string > m_dataDepsGraphFile
std::string toString() const override
value -> string
StatusCode revise(unsigned int iAlgo, EventContext *contextPtr, AState state, bool iterate=false)
bool filterPassed() const
void activate()
Activate scheduler.
tbb::concurrent_bounded_queue< action > m_actionsQueue
Queue where closures are stored and picked for execution.
std::unordered_map< std::string, unsigned int > m_algname_index_map
Map to bookkeep the information necessary to the name2index conversion.
Gaudi::Property< bool > m_checkDeps
bool isStalled(const EventSlot &) const
Check if scheduling in a particular slot is in a stall.
const DataObjIDColl & inputDataObjs() const override
std::thread m_thread
The thread in which the activate function runs.
Gaudi::Property< bool > m_showDataFlow
Gaudi::Property< std::vector< std::string > > m_checkOutputIgnoreList
constexpr static const auto FAILURE
size_t sizeOfSubset(State state) const
std::atomic_int m_freeSlots
Atomic to account for asyncronous updates by the scheduler wrt the rest.
unsigned int m_blockingAlgosInFlight
Number of algorithms presently in flight.
std::function< void(OccupancySnapshot)> m_snapshotCallback
StatusCode pushNewEvent(EventContext *eventContext) override
Make an event available to the scheduler.
std::function< StatusCode()> action
StatusCode dumpGraphFile(const std::map< std::string, DataObjIDColl > &inDeps, const std::map< std::string, DataObjIDColl > &outDeps) const
const boost::container::flat_set< int > algsInState(State state) const
AlgsExecutionStates algsStates
Vector of algorithms states.
SmartIF< IPrecedenceSvc > m_precSvc
A shortcut to the Precedence Service.
std::atomic< ActivationState > m_isActive
Flag to track if the scheduler is active or not.
tbb::concurrent_bounded_queue< EventContext * > m_finishedEvents
Queue of finished events.
std::vector< std::string > m_algname_vect
Vector to bookkeep the information necessary to the index2name conversion.
void dumpSchedulerState(int iSlot)
Dump the state of the scheduler.
Gaudi::Property< bool > m_verboseSubSlots
size_t index(const Gaudi::ParticleProperty *property, const Gaudi::Interfaces::IParticlePropertySvc *service)
helper utility for mapping of Gaudi::ParticleProperty object into non-negative integral sequential id...
SmartIF< ISvcLocator > & serviceLocator() const override
Retrieve pointer to service locator
A service which initializes a TBB thread pool.
Public Member Functions inherited from extends< Service, IScheduler >
1.8.18