| The Gaudi Framework v32r1 (f65d50dc)

AvalancheSchedulerSvc.cpp
Go to the documentation of this file.
1 #include "AvalancheSchedulerSvc.h"
2
3 #include "AlgoExecutionTask.h"
4 #include "IOBoundAlgTask.h"
5
6 // Framework includes
7 #include "GaudiKernel/ConcurrencyFlags.h"
8 #include "GaudiKernel/DataHandleHolderVisitor.h"
9 #include "GaudiKernel/IAlgorithm.h"
10 #include "GaudiKernel/IDataManagerSvc.h"
11 #include "GaudiKernel/ThreadLocalContext.h"
12 #include <Gaudi/Algorithm.h> // can be removed ASA dynamic casts to Algorithm are removed
13
14 // C++
15 #include <algorithm>
16 #include <map>
17 #include <queue>
18 #include <sstream>
19 #include <unordered_set>
20
21 // External libs
22 #include "boost/algorithm/string.hpp"
23 #include "boost/thread.hpp"
24 #include "boost/tokenizer.hpp"
25 // DP waiting for the TBB service
26 #include "tbb/task_scheduler_init.h"
27
28 // Instantiation of a static factory class used by clients to create instances of this service
29 DECLARE_COMPONENT( AvalancheSchedulerSvc )
30
31 #define ON_DEBUG if ( msgLevel( MSG::DEBUG ) )
32 #define ON_VERBOSE if ( msgLevel( MSG::VERBOSE ) )
33
34 namespace {
35 struct DataObjIDSorter {
36 bool operator()( const DataObjID* a, const DataObjID* b ) { return a->fullKey() < b->fullKey(); }
37 };
38
39 // Sort a DataObjIDColl in a well-defined, reproducible manner.
40 // Used for making debugging dumps.
41 std::vector<const DataObjID*> sortedDataObjIDColl( const DataObjIDColl& coll ) {
42 std::vector<const DataObjID*> v;
43 v.reserve( coll.size() );
44 for ( const DataObjID& id : coll ) v.push_back( &id );
45 std::sort( v.begin(), v.end(), DataObjIDSorter() );
46 return v;
47 }
48
49 bool subSlotAlgsInStates( const EventSlot& slot, std::initializer_list<AlgsExecutionStates::State> testStates ) {
50 return std::any_of( slot.allSubSlots.begin(), slot.allSubSlots.end(),
51 [testStates]( const EventSlot& ss ) { return ss.algsStates.containsAny( testStates ); } );
52 }
53 } // namespace
54
55 //---------------------------------------------------------------------------
56
63 StatusCode AvalancheSchedulerSvc::initialize() {
64
65 // Initialise mother class (read properties, ...)
66 StatusCode sc( Service::initialize() );
67 if ( sc.isFailure() ) warning() << "Base class could not be initialized" << endmsg;
68
69 // Get hold of the TBBSvc. This should initialize the thread pool
70 m_threadPoolSvc = serviceLocator()->service( "ThreadPoolSvc" );
71 if ( !m_threadPoolSvc.isValid() ) {
72 fatal() << "Error retrieving ThreadPoolSvc" << endmsg;
73 return StatusCode::FAILURE;
74 }
75
76 // Activate the scheduler in another thread.
77 info() << "Activating scheduler in a separate thread" << endmsg;
78 m_thread = std::thread( [this]() { this->activate(); } );
79
80 while ( m_isActive != ACTIVE ) {
81 if ( m_isActive == FAILURE ) {
82 fatal() << "Terminating initialization" << endmsg;
83 return StatusCode::FAILURE;
84 } else {
85 ON_DEBUG debug() << "Waiting for AvalancheSchedulerSvc to activate" << endmsg;
86 sleep( 1 );
87 }
88 }
89
90 if ( m_enableCondSvc ) {
91 // Get hold of the CondSvc
92 m_condSvc = serviceLocator()->service( "CondSvc" );
93 if ( !m_condSvc.isValid() ) {
94 warning() << "No CondSvc found, or not enabled. "
95 << "Will not manage CondAlgorithms" << endmsg;
96 m_enableCondSvc = false;
97 }
98 }
99
100 // Get the algo resource pool
101 m_algResourcePool = serviceLocator()->service( "AlgResourcePool" );
102 if ( !m_algResourcePool.isValid() ) {
103 fatal() << "Error retrieving AlgoResourcePool" << endmsg;
104 return StatusCode::FAILURE;
105 }
106
107 m_algExecStateSvc = serviceLocator()->service( "AlgExecStateSvc" );
108 if ( !m_algExecStateSvc.isValid() ) {
109 fatal() << "Error retrieving AlgExecStateSvc" << endmsg;
110 return StatusCode::FAILURE;
111 }
112
113 // Get Whiteboard
114 m_whiteboard = serviceLocator()->service( m_whiteboardSvcName );
115 if ( !m_whiteboard.isValid() ) {
116 fatal() << "Error retrieving EventDataSvc interface IHiveWhiteBoard." << endmsg;
117 return StatusCode::FAILURE;
118 }
119
120 // Get dedicated scheduler for I/O-bound algorithms
121 if ( m_useIOBoundAlgScheduler ) {
122 m_IOBoundAlgScheduler = serviceLocator()->service( m_IOBoundAlgSchedulerSvcName );
123 if ( !m_IOBoundAlgScheduler.isValid() )
124 fatal() << "Error retrieving IOBoundSchedulerAlgSvc interface IAccelerator." << endmsg;
125 }
126
127 // Set the MaxEventsInFlight parameters from the number of WB stores
128 m_maxEventsInFlight = m_whiteboard->getNumberOfStores();
129
130 // Set the number of free slots
131 m_freeSlots = m_maxEventsInFlight;
132
133 // Get the list of algorithms
134 const std::list<IAlgorithm*>& algos = m_algResourcePool->getFlatAlgList();
135 const unsigned int algsNumber = algos.size();
136 if ( algsNumber != 0 ) {
137 info() << "Found " << algsNumber << " algorithms" << endmsg;
138 } else {
139 error() << "No algorithms found" << endmsg;
140 return StatusCode::FAILURE;
141 }
142
143 /* Dependencies
144 1) Look for handles in algo, if none
145 2) Assume none are required
146 */
147
148 DataObjIDColl globalInp, globalOutp;
149
150 // figure out all outputs
151 for ( IAlgorithm* ialgoPtr : algos ) {
152 Gaudi::Algorithm* algoPtr = dynamic_cast<Gaudi::Algorithm*>( ialgoPtr );
153 if ( !algoPtr ) {
154 fatal() << "Could not convert IAlgorithm into Gaudi::Algorithm: this will result in a crash." << endmsg;
155 return StatusCode::FAILURE;
156 }
157 for ( auto id : algoPtr->outputDataObjs() ) {
158 auto r = globalOutp.insert( id );
159 if ( !r.second ) {
160 warning() << "multiple algorithms declare " << id
161 << " as output! could be a single instance in multiple paths "
162 "though, or control flow may guarantee only one runs...!"
163 << endmsg;
164 }
165 }
166 }
167
168 std::ostringstream ostdd;
169 ostdd << "Data Dependencies for Algorithms:";
170
171 std::map<std::string, DataObjIDColl> algosDependenciesMap;
172 for ( IAlgorithm* ialgoPtr : algos ) {
173 Gaudi::Algorithm* algoPtr = dynamic_cast<Gaudi::Algorithm*>( ialgoPtr );
174 if ( nullptr == algoPtr ) {
175 fatal() << "Could not convert IAlgorithm into Gaudi::Algorithm for " << ialgoPtr->name()
176 << ": this will result in a crash." << endmsg;
177 return StatusCode::FAILURE;
178 }
179
180 ostdd << "\n " << algoPtr->name();
181
182 DataObjIDColl algoDependencies;
183 if ( !algoPtr->inputDataObjs().empty() || !algoPtr->outputDataObjs().empty() ) {
184 for ( const DataObjID* idp : sortedDataObjIDColl( algoPtr->inputDataObjs() ) ) {
185 DataObjID id = *idp;
186 ostdd << "\n o INPUT " << id;
187 if ( id.key().find( ":" ) != std::string::npos ) {
188 ostdd << " contains alternatives which require resolution...\n";
189 auto tokens = boost::tokenizer<boost::char_separator<char>>{id.key(), boost::char_separator<char>{":"}};
190 auto itok = std::find_if( tokens.begin(), tokens.end(), [&]( const std::string& t ) {
191 return globalOutp.find( DataObjID{t} ) != globalOutp.end();
192 } );
193 if ( itok != tokens.end() ) {
194 ostdd << "found matching output for " << *itok << " -- updating scheduler info\n";
195 id.updateKey( *itok );
196 } else {
197 error() << "failed to find alternate in global output list"
198 << " for id: " << id << " in Alg " << algoPtr->name() << endmsg;
199 m_showDataDeps = true;
200 }
201 }
202 algoDependencies.insert( id );
203 globalInp.insert( id );
204 }
205 for ( const DataObjID* id : sortedDataObjIDColl( algoPtr->outputDataObjs() ) ) {
206 ostdd << "\n o OUTPUT " << *id;
207 if ( id->key().find( ":" ) != std::string::npos ) {
208 error() << " in Alg " << algoPtr->name() << " alternatives are NOT allowed for outputs! id: " << *id
209 << endmsg;
210 m_showDataDeps = true;
211 }
212 }
213 } else {
214 ostdd << "\n none";
215 }
216 algosDependenciesMap[algoPtr->name()] = algoDependencies;
217 }
218
219 if ( m_showDataDeps ) { info() << ostdd.str() << endmsg; }
220
221 // Check if we have unmet global input dependencies, and, optionally, heal them
222 // WARNING: this step must be done BEFORE the Precedence Service is initialized
223 if ( m_checkDeps ) {
224 DataObjIDColl unmetDep;
225 for ( auto o : globalInp )
226 if ( globalOutp.find( o ) == globalOutp.end() ) unmetDep.insert( o );
227
228 if ( unmetDep.size() > 0 ) {
229
230 std::ostringstream ost;
231 for ( const DataObjID* o : sortedDataObjIDColl( unmetDep ) ) {
232 ost << "\n o " << *o << " required by Algorithm: ";
233
234 for ( const auto& p : algosDependenciesMap )
235 if ( p.second.find( *o ) != p.second.end() ) ost << "\n * " << p.first;
236 }
237
238 if ( !m_useDataLoader.empty() ) {
239
240 // Find the DataLoader Alg
241 IAlgorithm* dataLoaderAlg( nullptr );
242 for ( IAlgorithm* algo : algos )
243 if ( algo->name() == m_useDataLoader ) {
244 dataLoaderAlg = algo;
245 break;
246 }
247
248 if ( dataLoaderAlg == nullptr ) {
249 fatal() << "No DataLoader Algorithm \"" << m_useDataLoader.value()
250 << "\" found, and unmet INPUT dependencies "
251 << "detected:\n"
252 << ost.str() << endmsg;
253 return StatusCode::FAILURE;
254 }
255
256 info() << "Will attribute the following unmet INPUT dependencies to \"" << dataLoaderAlg->type() << "/"
257 << dataLoaderAlg->name() << "\" Algorithm" << ost.str() << endmsg;
258
259 // Set the property Load of DataLoader Alg
260 Gaudi::Algorithm* dataAlg = dynamic_cast<Gaudi::Algorithm*>( dataLoaderAlg );
261 if ( !dataAlg ) {
262 fatal() << "Unable to dcast DataLoader \"" << m_useDataLoader.value() << "\" IAlg to Gaudi::Algorithm"
263 << endmsg;
264 return StatusCode::FAILURE;
265 }
266
267 for ( auto& id : unmetDep ) {
268 ON_DEBUG debug() << "adding OUTPUT dep \"" << id << "\" to " << dataLoaderAlg->type() << "/"
269 << dataLoaderAlg->name() << endmsg;
270 dataAlg->addDependency( id, Gaudi::DataHandle::Writer );
271 }
272
273 } else {
274 fatal() << "Auto DataLoading not requested, "
275 << "and the following unmet INPUT dependencies were found:" << ost.str() << endmsg;
276 return StatusCode::FAILURE;
277 }
278
279 } else {
280 info() << "No unmet INPUT data dependencies were found" << endmsg;
281 }
282 }
283
284 // Get the precedence service
285 m_precSvc = serviceLocator()->service( "PrecedenceSvc" );
286 if ( !m_precSvc.isValid() ) {
287 fatal() << "Error retrieving PrecedenceSvc" << endmsg;
288 return StatusCode::FAILURE;
289 }
290 const PrecedenceSvc* precSvc = dynamic_cast<const PrecedenceSvc*>( m_precSvc.get() );
291 if ( !precSvc ) {
292 fatal() << "Unable to dcast PrecedenceSvc" << endmsg;
293 return StatusCode::FAILURE;
294 }
295
296 // Fill the containers to convert algo names to index
297 m_algname_vect.resize( algsNumber );
298 for ( IAlgorithm* algo : algos ) {
299 const std::string& name = algo->name();
300 auto index = precSvc->getRules()->getAlgorithmNode( name )->getAlgoIndex();
301 m_algname_index_map[name] = index;
302 m_algname_vect.at( index ) = name;
303 }
304
305 // Shortcut for the message service
306 SmartIF<IMessageSvc> messageSvc( serviceLocator() );
307 if ( !messageSvc.isValid() ) error() << "Error retrieving MessageSvc interface IMessageSvc." << endmsg;
308
309 m_eventSlots.reserve( m_maxEventsInFlight );
310 for ( size_t i = 0; i < m_maxEventsInFlight; ++i ) {
311 m_eventSlots.emplace_back( algsNumber, precSvc->getRules()->getControlFlowNodeCounter(), messageSvc );
312 m_eventSlots.back().complete = true;
313 }
314 m_actionsCounts.assign( m_maxEventsInFlight, 0 );
315
316 if ( m_threadPoolSize > 1 ) { m_maxAlgosInFlight = (size_t)m_threadPoolSize; }
317
318 // Clearly inform about the level of concurrency
319 info() << "Concurrency level information:" << endmsg;
320 info() << " o Number of events in flight: " << m_maxEventsInFlight << endmsg;
321 info() << " o TBB thread pool size: " << m_threadPoolSize << endmsg;
322
323 if ( m_showControlFlow ) m_precSvc->dumpControlFlow();
324
325 if ( m_showDataFlow ) m_precSvc->dumpDataFlow();
326
327 // Simulate execution flow
328 if ( m_simulateExecution ) m_precSvc->simulate( m_eventSlots[0] );
329
330 return sc;
331 }
332 //---------------------------------------------------------------------------
333
337 StatusCode AvalancheSchedulerSvc::finalize() {
338
339 StatusCode sc( Service::finalize() );
340 if ( sc.isFailure() ) warning() << "Base class could not be finalized" << endmsg;
341
342 sc = deactivate();
343 if ( sc.isFailure() ) warning() << "Scheduler could not be deactivated" << endmsg;
344
345 info() << "Joining Scheduler thread" << endmsg;
346 m_thread.join();
347
348 // Final error check after thread pool termination
349 if ( m_isActive == FAILURE ) {
350 error() << "problems in scheduler thread" << endmsg;
351 return StatusCode::FAILURE;
352 }
353
354 return sc;
355 }
356 //---------------------------------------------------------------------------
357
368 void AvalancheSchedulerSvc::activate() {
369
370 ON_DEBUG debug() << "AvalancheSchedulerSvc::activate()" << endmsg;
371
372 if ( m_threadPoolSvc->initPool( m_threadPoolSize ).isFailure() ) {
373 error() << "problems initializing ThreadPoolSvc" << endmsg;
374 m_isActive = FAILURE;
375 return;
376 }
377
378 // Wait for actions pushed into the queue by finishing tasks.
379 action thisAction;
380 StatusCode sc( StatusCode::SUCCESS );
381
382 m_isActive = ACTIVE;
383
384 // Continue to wait if the scheduler is running or there is something to do
385 ON_DEBUG debug() << "Start checking the actionsQueue" << endmsg;
386 while ( m_isActive == ACTIVE || m_actionsQueue.size() != 0 ) {
387 m_actionsQueue.pop( thisAction );
388 sc = thisAction();
389 ON_VERBOSE {
390 if ( sc.isFailure() )
391 verbose() << "Action did not succeed (which is not bad per se)." << endmsg;
392 else
393 verbose() << "Action succeeded." << endmsg;
394 }
395 else sc.ignore();
396 }
397
398 ON_DEBUG debug() << "Terminating thread-pool resources" << endmsg;
399 if ( m_threadPoolSvc->terminatePool().isFailure() ) {
400 error() << "Problems terminating thread pool" << endmsg;
401 m_isActive = FAILURE;
402 }
403 }
404
405 //---------------------------------------------------------------------------
406
413 StatusCode AvalancheSchedulerSvc::deactivate() {
414
415 if ( m_isActive == ACTIVE ) {
416
417 // Set the number of slots available to an error code
418 m_freeSlots.store( 0 );
419
420 // Empty queue
421 action thisAction;
422 while ( m_actionsQueue.try_pop( thisAction ) ) {};
423
424 // This would be the last action
425 m_actionsQueue.push( [this]() -> StatusCode {
426 ON_VERBOSE verbose() << "Deactivating scheduler" << endmsg;
427 m_isActive = INACTIVE;
428 return StatusCode::SUCCESS;
429 } );
430 }
431
432 return StatusCode::SUCCESS;
433 }
434
435 //---------------------------------------------------------------------------
436
437 // Utils and shortcuts
438
439 inline const std::string& AvalancheSchedulerSvc::index2algname( unsigned int index ) { return m_algname_vect[index]; }
440
441 //---------------------------------------------------------------------------
442
443 inline unsigned int AvalancheSchedulerSvc::algname2index( const std::string& algoname ) {
444 unsigned int index = m_algname_index_map[algoname];
445 return index;
446 }
447
448 //---------------------------------------------------------------------------
449
450 // EventSlot management
457 StatusCode AvalancheSchedulerSvc::pushNewEvent( EventContext* eventContext ) {
458
459 if ( !eventContext ) {
460 fatal() << "Event context is nullptr" << endmsg;
461 return StatusCode::FAILURE;
462 }
463
464 if ( m_freeSlots.load() == 0 ) {
465 ON_DEBUG debug() << "A free processing slot could not be found." << endmsg;
466 return StatusCode::FAILURE;
467 }
468
469 // no problem as push new event is only called from one thread (event loop manager)
470 --m_freeSlots;
471
472 auto action = [this, eventContext]() -> StatusCode {
473 // Event processing slot forced to be the same as the wb slot
474 const unsigned int thisSlotNum = eventContext->slot();
475 EventSlot& thisSlot = m_eventSlots[thisSlotNum];
476 if ( !thisSlot.complete ) {
477 fatal() << "The slot " << thisSlotNum << " is supposed to be a finished event but it's not" << endmsg;
478 return StatusCode::FAILURE;
479 }
480
481 ON_DEBUG debug() << "Executing event " << eventContext->evt() << " on slot " << thisSlotNum << endmsg;
482 thisSlot.reset( eventContext );
483
484 // Result status code:
485 StatusCode result = StatusCode::SUCCESS;
486
487 // promote to CR and DR the initial set of algorithms
488 Cause cs = {Cause::source::Root, "RootDecisionHub"};
489 if ( m_precSvc->iterate( thisSlot, cs ).isFailure() ) {
490 error() << "Failed to call IPrecedenceSvc::iterate for slot " << thisSlotNum << endmsg;
491 result = StatusCode::FAILURE;
492 }
493
494 if ( this->updateStates( thisSlotNum ).isFailure() ) {
495 error() << "Failed to call AvalancheSchedulerSvc::updateStates for slot " << thisSlotNum << endmsg;
496 result = StatusCode::FAILURE;
497 }
498
499 return result;
500 }; // end of lambda
501
502 // Kick off the scheduling!
503 ON_VERBOSE {
504 verbose() << "Pushing the action to update the scheduler for slot " << eventContext->slot() << endmsg;
505 verbose() << "Free slots available " << m_freeSlots.load() << endmsg;
506 }
507
508 m_actionsQueue.push( action );
509
510 return StatusCode::SUCCESS;
511 }
512
513 //---------------------------------------------------------------------------
514
515 StatusCode AvalancheSchedulerSvc::pushNewEvents( std::vector<EventContext*>& eventContexts ) {
516 StatusCode sc;
517 for ( auto context : eventContexts ) {
518 sc = pushNewEvent( context );
519 if ( sc != StatusCode::SUCCESS ) return sc;
520 }
521 return sc;
522 }
523
524 //---------------------------------------------------------------------------
525
526 unsigned int AvalancheSchedulerSvc::freeSlots() { return std::max( m_freeSlots.load(), 0 ); }
527
528 //---------------------------------------------------------------------------
532 StatusCode AvalancheSchedulerSvc::popFinishedEvent( EventContext*& eventContext ) {
533
534 // ON_DEBUG debug() << "popFinishedEvent: queue size: " << m_finishedEvents.size() << endmsg;
535 if ( m_freeSlots.load() == (int)m_maxEventsInFlight || m_isActive == INACTIVE ) {
536 // ON_DEBUG debug() << "freeslots: " << m_freeSlots << "/" << m_maxEventsInFlight
537 // << " active: " << m_isActive << endmsg;
538 return StatusCode::FAILURE;
539 } else {
540 // ON_DEBUG debug() << "freeslots: " << m_freeSlots << "/" << m_maxEventsInFlight
541 // << " active: " << m_isActive << endmsg;
542 m_finishedEvents.pop( eventContext );
543 ++m_freeSlots;
544 ON_DEBUG debug() << "Popped slot " << eventContext->slot() << " (event " << eventContext->evt() << ")" << endmsg;
545 return StatusCode::SUCCESS;
546 }
547 }
548
549 //---------------------------------------------------------------------------
553 StatusCode AvalancheSchedulerSvc::tryPopFinishedEvent( EventContext*& eventContext ) {
554
555 if ( m_finishedEvents.try_pop( eventContext ) ) {
556 ON_DEBUG debug() << "Try Pop successful slot " << eventContext->slot() << "(event " << eventContext->evt() << ")"
557 << endmsg;
558 ++m_freeSlots;
559 return StatusCode::SUCCESS;
560 }
561 return StatusCode::FAILURE;
562 }
563
564 //--------------------------------------------------------------------------
565 // States Management
566
576 StatusCode AvalancheSchedulerSvc::updateStates( int si, const int algo_index, const int sub_slot,
577 const int source_slot ) {
578
579 StatusCode global_sc( StatusCode::SUCCESS );
580
581 // Sort from the oldest to the newest event
582 // Prepare a vector of pointers to the slots to avoid copies
583 std::vector<EventSlot*> eventSlotsPtrs;
584
585 // Consider all slots if si <0 or just one otherwise
586 if ( si < 0 ) {
587 const int eventsSlotsSize( m_eventSlots.size() );
588 eventSlotsPtrs.reserve( eventsSlotsSize );
589 for ( auto slotIt = m_eventSlots.begin(); slotIt != m_eventSlots.end(); ++slotIt ) {
590 if ( !slotIt->complete ) eventSlotsPtrs.push_back( &( *slotIt ) );
591 }
592 std::sort( eventSlotsPtrs.begin(), eventSlotsPtrs.end(),
593 []( EventSlot* a, EventSlot* b ) { return a->eventContext->evt() < b->eventContext->evt(); } );
594 } else {
595 eventSlotsPtrs.push_back( &m_eventSlots[si] );
596 }
597
598 for ( EventSlot* thisSlotPtr : eventSlotsPtrs ) {
599 int iSlot = thisSlotPtr->eventContext->slot();
600
601 // Cache the states of the algos to improve readability and performance
602 auto& thisSlot = m_eventSlots[iSlot];
603 AlgsExecutionStates& thisAlgsStates = thisSlot.algsStates;
604
605 // Perform the I->CR->DR transitions
606 if ( algo_index >= 0 ) {
607 Cause cs = {Cause::source::Task, index2algname( algo_index )};
608
609 // Run in whole-event context if there's no sub-slot index, or the sub-slot has a different parent
610 if ( sub_slot == -1 || iSlot != source_slot ) {
611 if ( m_precSvc->iterate( thisSlot, cs ).isFailure() ) {
612 error() << "Failed to call IPrecedenceSvc::iterate for slot " << iSlot << endmsg;
613 global_sc = StatusCode::FAILURE;
614 }
615 } else {
616 if ( m_precSvc->iterate( thisSlot.allSubSlots[sub_slot], cs ).isFailure() ) {
617 error() << "Failed to call IPrecedenceSvc::iterate for sub-slot " << sub_slot << " of " << iSlot << endmsg;
618 global_sc = StatusCode::FAILURE;
619 }
620 }
621 }
622
623 StatusCode partial_sc( StatusCode::FAILURE, true );
624
625 // Perform DR->SCHEDULED
626 if ( !m_optimizationMode.empty() ) {
627 auto comp_nodes = [this]( const uint& i, const uint& j ) {
628 return ( m_precSvc->getPriority( index2algname( i ) ) < m_precSvc->getPriority( index2algname( j ) ) );
629 };
630 std::priority_queue<uint, std::vector<uint>, std::function<bool( const uint&, const uint& )>> buffer(
631 comp_nodes, std::vector<uint>() );
632 for ( auto it = thisAlgsStates.begin( AState::DATAREADY ); it != thisAlgsStates.end( AState::DATAREADY ); ++it )
633 buffer.push( *it );
634 while ( !buffer.empty() ) {
635 bool IOBound = false;
636 if ( m_useIOBoundAlgScheduler ) IOBound = m_precSvc->isBlocking( index2algname( buffer.top() ) );
637
638 if ( !IOBound )
639 partial_sc = promoteToScheduled( buffer.top(), iSlot, thisSlotPtr->eventContext.get() );
640 else
641 partial_sc = promoteToAsyncScheduled( buffer.top(), iSlot, thisSlotPtr->eventContext.get() );
642
643 ON_VERBOSE if ( partial_sc.isFailure() ) verbose()
644 << "Could not apply transition from " << AState::DATAREADY << " for algorithm "
645 << index2algname( buffer.top() ) << " on processing slot " << iSlot << endmsg;
646
647 buffer.pop();
648 }
649
650 } else {
651 for ( auto it = thisAlgsStates.begin( AState::DATAREADY ); it != thisAlgsStates.end( AState::DATAREADY ); ++it ) {
652 uint algIndex = *it;
653
654 bool IOBound = false;
655 if ( m_useIOBoundAlgScheduler ) IOBound = m_precSvc->isBlocking( index2algname( algIndex ) );
656
657 if ( !IOBound )
658 partial_sc = promoteToScheduled( algIndex, iSlot, thisSlotPtr->eventContext.get() );
659 else
660 partial_sc = promoteToAsyncScheduled( algIndex, iSlot, thisSlotPtr->eventContext.get() );
661
662 ON_VERBOSE if ( partial_sc.isFailure() ) verbose()
663 << "Could not apply transition from " << AState::DATAREADY << " for algorithm " << index2algname( algIndex )
664 << " on processing slot " << iSlot << endmsg;
665 }
666 }
667
668 // Check for algorithms ready in sub-slots
669 for ( auto& subslot : thisSlot.allSubSlots ) {
670 auto& subslotStates = subslot.algsStates;
671 for ( auto it = subslotStates.begin( AState::DATAREADY ); it != subslotStates.end( AState::DATAREADY ); ++it ) {
672 uint algIndex{*it};
673 partial_sc = promoteToScheduled( algIndex, iSlot, subslot.eventContext.get() );
674 // The following verbosity is expensive when the number of sub-slots is high
675 /*ON_VERBOSE if ( partial_sc.isFailure() ) verbose()
676 << "Could not apply transition from " << AState::DATAREADY << " for algorithm " << index2algname( algIndex )
677 << " on processing subslot " << subslot.eventContext->slot() << endmsg;*/
678 }
679 }
680
681 if ( m_dumpIntraEventDynamics ) {
682 std::stringstream s;
683 s << ( algo_index != -1 ? index2algname( algo_index ) : "START" ) << ", "
684 << thisAlgsStates.sizeOfSubset( AState::CONTROLREADY ) << ", "
685 << thisAlgsStates.sizeOfSubset( AState::DATAREADY ) << ", " << thisAlgsStates.sizeOfSubset( AState::SCHEDULED )
686 << ", " << std::chrono::high_resolution_clock::now().time_since_epoch().count() << "\n";
687 auto threads = ( m_threadPoolSize != -1 ) ? std::to_string( m_threadPoolSize )
688 : std::to_string( tbb::task_scheduler_init::default_num_threads() );
689 std::ofstream myfile;
690 myfile.open( "IntraEventFSMOccupancy_" + threads + "T.csv", std::ios::app );
691 myfile << s.str();
692 myfile.close();
693 }
694
695 // Not complete because this would mean that the slot is already free!
696 if ( m_precSvc->CFRulesResolved( thisSlot ) &&
697 !thisSlot.algsStates.containsAny( {AState::CONTROLREADY, AState::DATAREADY, AState::SCHEDULED} ) &&
698 !subSlotAlgsInStates( thisSlot, {AState::CONTROLREADY, AState::DATAREADY, AState::SCHEDULED} ) &&
699 !thisSlot.complete ) {
700
701 thisSlot.complete = true;
702 // if the event did not fail, add it to the finished events
703 // otherwise it is taken care of in the error handling
704 if ( m_algExecStateSvc->eventStatus( *thisSlot.eventContext ) == EventStatus::Success ) {
705 ON_DEBUG debug() << "Event " << thisSlot.eventContext->evt() << " finished (slot "
706 << thisSlot.eventContext->slot() << ")." << endmsg;
707 m_finishedEvents.push( thisSlot.eventContext.release() );
708 }
709
710 // now let's return the fully evaluated result of the control flow
711 ON_DEBUG debug() << m_precSvc->printState( thisSlot ) << endmsg;
712
713 thisSlot.eventContext.reset( nullptr );
714
715 } else if ( isStalled( thisSlot ) ) {
716 m_algExecStateSvc->setEventStatus( EventStatus::AlgStall, *thisSlot.eventContext );
717 eventFailed( thisSlot.eventContext.get() ); // can't release yet
718 }
719 partial_sc.ignore();
720 } // end loop on slots
721
722 ON_VERBOSE verbose() << "States Updated." << endmsg;
723
724 return global_sc;
725 }
726
727 //---------------------------------------------------------------------------
728
735 bool AvalancheSchedulerSvc::isStalled( const EventSlot& slot ) const {
736
737 if ( m_actionsCounts[slot.eventContext->slot()] == 0 &&
738 !slot.algsStates.containsAny( {AState::DATAREADY, AState::SCHEDULED} ) &&
739 !subSlotAlgsInStates( slot, {AState::DATAREADY, AState::SCHEDULED} ) ) {
740
741 error() << "*** Stall detected in slot " << slot.eventContext->slot() << "! ***" << endmsg;
742
743 return true;
744 }
745 return false;
746 }
747
748 //---------------------------------------------------------------------------
749
754 void AvalancheSchedulerSvc::eventFailed( EventContext* eventContext ) {
755 const uint slotIdx = eventContext->slot();
756
757 error() << "Event " << eventContext->evt() << " on slot " << slotIdx << " failed" << endmsg;
758
759 dumpSchedulerState( msgLevel( MSG::VERBOSE ) ? -1 : slotIdx );
760
761 // dump temporal and topological precedence analysis (if enabled in the PrecedenceSvc)
762 m_precSvc->dumpPrecedenceRules( m_eventSlots[slotIdx] );
763
764 // Push into the finished events queue the failed context
765 m_eventSlots[slotIdx].complete = true;
766 m_finishedEvents.push( m_eventSlots[slotIdx].eventContext.release() );
767 }
768
769 //---------------------------------------------------------------------------
770
775 void AvalancheSchedulerSvc::dumpSchedulerState( int iSlot ) {
776
777 // To have just one big message
778 std::ostringstream outputMS;
779
780 outputMS << "Dumping scheduler state\n"
781 << "=========================================================================================\n"
782 << "++++++++++++++++++++++++++++++++++++ SCHEDULER STATE ++++++++++++++++++++++++++++++++++++\n"
783 << "=========================================================================================\n\n";
784
785 //===========================================================================
786
787 outputMS << "------------------ Last schedule: Task/Event/Slot/Thread/State Mapping "
788 << "------------------\n\n";
789
790 // Figure if TimelineSvc is available (used below to detect threads IDs)
791 auto timelineSvc = serviceLocator()->service<ITimelineSvc>( "TimelineSvc", false );
792 if ( !timelineSvc.isValid() || !timelineSvc->isEnabled() ) {
793 outputMS << "WARNING Enable TimelineSvc in record mode (RecordTimeline = True) to trace the mapping\n";
794 } else {
795
796 // Figure optimal printout layout
797 size_t indt( 0 );
798 for ( auto& slot : m_eventSlots )
799 for ( auto it = slot.algsStates.begin( AState::SCHEDULED ); it != slot.algsStates.end( AState::SCHEDULED ); ++it )
800 if ( index2algname( *it ).length() > indt ) indt = index2algname( *it ).length();
801
802 // Figure the last running schedule across all slots
803 for ( auto& slot : m_eventSlots ) {
804 for ( auto it = slot.algsStates.begin( AState::SCHEDULED ); it != slot.algsStates.end( AState::SCHEDULED );
805 ++it ) {
806
807 const std::string algoName{index2algname( *it )};
808
809 outputMS << " task: " << std::setw( indt ) << algoName << " evt/slot: " << slot.eventContext->evt() << "/"
810 << slot.eventContext->slot();
811
812 // Try to get POSIX threads IDs the currently running tasks are scheduled to
813 if ( timelineSvc.isValid() ) {
814 TimelineEvent te{};
815 te.algorithm = algoName;
816 te.slot = slot.eventContext->slot();
817 te.event = slot.eventContext->evt();
818
819 if ( timelineSvc->getTimelineEvent( te ) )
820 outputMS << " thread.id: 0x" << std::hex << te.thread << std::dec;
821 else
822 outputMS << " thread.id: [unknown]"; // this means a task has just
823 // been signed off as SCHEDULED,
824 // but has not been assigned to a thread yet
825 // (i.e., not running yet)
826 }
827 outputMS << " state: [" << m_algExecStateSvc->algExecState( algoName, *( slot.eventContext ) ) << "]\n";
828 }
829 }
830 }
831
832 //===========================================================================
833
834 outputMS << "\n---------------------------- Task/CF/FSM Mapping "
835 << ( 0 > iSlot ? "[all slots] --" : "[target slot] " ) << "--------------------------\n\n";
836
837 int slotCount = -1;
838 for ( auto& slot : m_eventSlots ) {
839 ++slotCount;
840 if ( slot.complete ) continue;
841
842 outputMS << "[ slot: "
843 << ( slot.eventContext->valid() ? std::to_string( slot.eventContext->slot() ) : "[ctx invalid]" )
844 << " event: "
845 << ( slot.eventContext->valid() ? std::to_string( slot.eventContext->evt() ) : "[ctx invalid]" )
846 << " ]:\n\n";
847
848 if ( 0 > iSlot || iSlot == slotCount ) {
849
850 // Snapshot of the Control Flow and FSM states
851 outputMS << m_precSvc->printState( slot ) << "\n";
852
853 // Mention sub slots (this is expensive if the number of sub-slots is high)
854 if ( m_verboseSubSlots && !slot.allSubSlots.empty() ) {
855 outputMS << "\nNumber of sub-slots: " << slot.allSubSlots.size() << "\n\n";
856 auto slotID = slot.eventContext->valid() ? std::to_string( slot.eventContext->slot() ) : "[ctx invalid]";
857 for ( auto& ss : slot.allSubSlots ) {
858 outputMS << "[ slot: " << slotID << ", sub-slot: "
859 << ( ss.eventContext->valid() ? std::to_string( ss.eventContext->subSlot() ) : "[ctx invalid]" )
860 << ", entry: " << ss.entryPoint << ", event: "
861 << ( ss.eventContext->valid() ? std::to_string( ss.eventContext->evt() ) : "[ctx invalid]" )
862 << " ]:\n\n";
863 outputMS << m_precSvc->printState( ss ) << "\n";
864 }
865 }
866 }
867 }
868
869 //===========================================================================
870
871 if ( 0 <= iSlot ) {
872 outputMS << "\n------------------------------ Algorithm Execution States -----------------------------\n\n";
873 m_algExecStateSvc->dump( outputMS, *( m_eventSlots[iSlot].eventContext ) );
874 }
875
876 outputMS << "\n=========================================================================================\n"
877 << "++++++++++++++++++++++++++++++++++++++ END OF DUMP ++++++++++++++++++++++++++++++++++++++\n"
878 << "=========================================================================================\n\n";
879
880 info() << outputMS.str() << endmsg;
881 }
882
883 //---------------------------------------------------------------------------
884
885 StatusCode AvalancheSchedulerSvc::promoteToScheduled( unsigned int iAlgo, int si, EventContext* eventContext ) {
886
887 if ( m_algosInFlight == m_maxAlgosInFlight ) return StatusCode::FAILURE;
888
889 const std::string& algName( index2algname( iAlgo ) );
890 IAlgorithm* ialgoPtr = nullptr;
891 StatusCode sc( m_algResourcePool->acquireAlgorithm( algName, ialgoPtr ) );
892
893 if ( sc.isSuccess() ) { // if we managed to get an algorithm instance try to schedule it
894
895 ++m_algosInFlight;
896 auto promote2ExecutedClosure = [this, iAlgo, ialgoPtr, eventContext]() {
897 this->m_actionsQueue.push( [this, iAlgo, ialgoPtr, eventContext]() {
898 return this->AvalancheSchedulerSvc::promoteToExecuted( iAlgo, eventContext->slot(), ialgoPtr, eventContext );
899 } );
900 return StatusCode::SUCCESS;
901 };
902
903 // Avoid to use tbb if the pool size is 1 and run in this thread
904 if ( -100 != m_threadPoolSize ) {
905 // the child task that executes an Algorithm
906 tbb::task* algoTask = new ( tbb::task::allocate_root() )
907 AlgoExecutionTask( ialgoPtr, *eventContext, serviceLocator(), m_algExecStateSvc, promote2ExecutedClosure );
908 // schedule the algoTask
909 tbb::task::enqueue( *algoTask );
910
911 } else {
912 AlgoExecutionTask theTask( ialgoPtr, *eventContext, serviceLocator(), m_algExecStateSvc,
913 promote2ExecutedClosure );
914 theTask.execute();
915 }
916
917 ON_DEBUG debug() << "Algorithm " << algName << " was submitted on event " << eventContext->evt() << " in slot "
918 << si << ". Algorithms scheduled are " << m_algosInFlight << endmsg;
919
920 // Update states in the appropriate event slot
921 StatusCode updateSc;
922 EventSlot& thisSlot = m_eventSlots[si];
923 if ( eventContext->usesSubSlot() ) {
924 // Sub-slot
925 size_t const subSlotIndex = eventContext->subSlot();
926 updateSc = thisSlot.allSubSlots[subSlotIndex].algsStates.set( iAlgo, AState::SCHEDULED );
927 } else {
928 // Event level (standard behaviour)
929 updateSc = thisSlot.algsStates.set( iAlgo, AState::SCHEDULED );
930 }
931
932 ON_VERBOSE dumpSchedulerState( -1 );
933
934 if ( updateSc.isSuccess() )
935 ON_VERBOSE verbose() << "Promoting " << algName << " to SCHEDULED on slot " << si << endmsg;
936 return updateSc;
937 } else {
938 ON_DEBUG debug() << "Could not acquire instance for algorithm " << index2algname( iAlgo ) << " on slot " << si
939 << endmsg;
940 return sc;
941 }
942 }
943
944 //---------------------------------------------------------------------------
945
946 StatusCode AvalancheSchedulerSvc::promoteToAsyncScheduled( unsigned int iAlgo, int si, EventContext* eventContext ) {
947
948 if ( m_IOBoundAlgosInFlight == m_maxIOBoundAlgosInFlight ) return StatusCode::FAILURE;
949
950 // bool IOBound = m_precSvc->isBlocking(algName);
951
952 const std::string& algName( index2algname( iAlgo ) );
953 IAlgorithm* ialgoPtr = nullptr;
954 StatusCode sc( m_algResourcePool->acquireAlgorithm( algName, ialgoPtr ) );
955
956 if ( sc.isSuccess() ) { // if we managed to get an algorithm instance try to schedule it
957
958 ++m_IOBoundAlgosInFlight;
959 auto promote2ExecutedClosure = [this, iAlgo, ialgoPtr, eventContext]() {
960 this->m_actionsQueue.push( [this, iAlgo, ialgoPtr, eventContext]() {
961 return this->AvalancheSchedulerSvc::promoteToAsyncExecuted( iAlgo, eventContext->slot(), ialgoPtr,
962 eventContext );
963 } );
964 return StatusCode::SUCCESS;
965 };
966 // Can we use tbb-based overloaded new-operator for a "custom" task (an algorithm wrapper, not derived from
967 // tbb::task)? it seems it works..
968 IOBoundAlgTask* theTask = new ( tbb::task::allocate_root() )
969 IOBoundAlgTask( ialgoPtr, *eventContext, serviceLocator(), m_algExecStateSvc, promote2ExecutedClosure );
970 m_IOBoundAlgScheduler->push( *theTask );
971
972 ON_DEBUG debug() << "[Asynchronous] Algorithm " << algName << " was submitted on event " << eventContext->evt()
973 << " in slot " << si << ". algorithms scheduled are " << m_IOBoundAlgosInFlight << endmsg;
974
975 // Update states in the appropriate event slot
976 StatusCode updateSc;
977 EventSlot& thisSlot = m_eventSlots[si];
978 if ( eventContext->usesSubSlot() ) {
979 // Sub-slot
980 size_t const subSlotIndex = eventContext->subSlot();
981 updateSc = thisSlot.allSubSlots[subSlotIndex].algsStates.set( iAlgo, AState::SCHEDULED );
982 } else {
983 // Event level (standard behaviour)
984 updateSc = thisSlot.algsStates.set( iAlgo, AState::SCHEDULED );
985 }
986
987 ON_VERBOSE if ( updateSc.isSuccess() ) verbose()
988 << "[Asynchronous] Promoting " << algName << " to SCHEDULED on slot " << si << endmsg;
989 return updateSc;
990 } else {
991 ON_DEBUG debug() << "[Asynchronous] Could not acquire instance for algorithm " << index2algname( iAlgo )
992 << " on slot " << si << endmsg;
993 return sc;
994 }
995 }
996
997 //---------------------------------------------------------------------------
998
1002 StatusCode AvalancheSchedulerSvc::promoteToExecuted( unsigned int iAlgo, int si, IAlgorithm* algo,
1003 EventContext* eventContext ) {
1004 Gaudi::Hive::setCurrentContext( eventContext );
1005 StatusCode sc = m_algResourcePool->releaseAlgorithm( algo->name(), algo );
1006
1007 if ( sc.isFailure() ) {
1008 error() << "[Event " << eventContext->evt() << ", Slot " << eventContext->slot() << "] "
1009 << "Instance of algorithm " << algo->name() << " could not be properly put back." << endmsg;
1010 return StatusCode::FAILURE;
1011 }
1012
1013 --m_algosInFlight;
1014
1015 EventSlot& thisSlot = m_eventSlots[si];
1016
1017 ON_DEBUG debug() << "Trying to handle execution result of " << algo->name() << " on slot " << si << endmsg;
1018
1019 const AlgExecState& algstate = m_algExecStateSvc->algExecState( algo, *eventContext );
1020 AState state = algstate.execStatus().isSuccess()
1021 ? ( algstate.filterPassed() ? AState::EVTACCEPTED : AState::EVTREJECTED )
1022 : AState::ERROR;
1023
1024 // Update states in the appropriate slot
1025 int subSlotIndex = -1;
1026 if ( eventContext->usesSubSlot() ) {
1027 // Sub-slot
1028 subSlotIndex = eventContext->subSlot();
1029 sc = thisSlot.allSubSlots[subSlotIndex].algsStates.set( iAlgo, state );
1030 } else {
1031 // Event level (standard behaviour)
1032 sc = thisSlot.algsStates.set( iAlgo, state );
1033 }
1034
1035 ON_VERBOSE if ( sc.isSuccess() ) verbose()
1036 << "Promoting " << algo->name() << " on slot " << si << " to " << state << endmsg;
1037
1038 ON_DEBUG debug() << "Algorithm " << algo->name() << " executed in slot " << si << ". Algorithms scheduled are "
1039 << m_algosInFlight << endmsg;
1040
1041 // Schedule an update of the status of the algorithms
1042 ++m_actionsCounts[si];
1043 m_actionsQueue.push( [this, si, iAlgo, subSlotIndex]() {
1044 --this->m_actionsCounts[si]; // no bound check needed as decrements/increments are balanced in the current setup
1045 return this->updateStates( -1, iAlgo, subSlotIndex, si );
1046 } );
1047
1048 return sc;
1049 }
1050
1051 //---------------------------------------------------------------------------
1052
1056 StatusCode AvalancheSchedulerSvc::promoteToAsyncExecuted( unsigned int iAlgo, int si, IAlgorithm* algo,
1057 EventContext* eventContext ) {
1058 Gaudi::Hive::setCurrentContext( eventContext );
1059 StatusCode sc = m_algResourcePool->releaseAlgorithm( algo->name(), algo );
1060
1061 if ( sc.isFailure() ) {
1062 error() << "[Asynchronous] [Event " << eventContext->evt() << ", Slot " << eventContext->slot() << "] "
1063 << "Instance of algorithm " << algo->name() << " could not be properly put back." << endmsg;
1064 return StatusCode::FAILURE;
1065 }
1066
1067 --m_IOBoundAlgosInFlight;
1068
1069 EventSlot& thisSlot = m_eventSlots[si];
1070
1071 ON_DEBUG debug() << "[Asynchronous] Trying to handle execution result of " << algo->name() << " on slot " << si
1072 << endmsg;
1073
1074 const AlgExecState& algstate = m_algExecStateSvc->algExecState( algo, *eventContext );
1075 AState state = algstate.execStatus().isSuccess()
1076 ? ( algstate.filterPassed() ? AState::EVTACCEPTED : AState::EVTREJECTED )
1077 : AState::ERROR;
1078
1079 // Update states in the appropriate slot
1080 int subSlotIndex = -1;
1081 if ( eventContext->usesSubSlot() ) {
1082 // Sub-slot
1083 subSlotIndex = eventContext->subSlot();
1084 sc = thisSlot.allSubSlots[subSlotIndex].algsStates.set( iAlgo, state );
1085 } else {
1086 // Event level (standard behaviour)
1087 sc = thisSlot.algsStates.set( iAlgo, state );
1088 }
1089
1090 ON_VERBOSE if ( sc.isSuccess() ) verbose()
1091 << "[Asynchronous] Promoting " << algo->name() << " on slot " << si << " to " << state << endmsg;
1092
1093 ON_DEBUG debug() << "[Asynchronous] Algorithm " << algo->name() << " executed in slot " << si
1094 << ". Algorithms scheduled are " << m_IOBoundAlgosInFlight << endmsg;
1095
1096 // Schedule an update of the status of the algorithms
1097 ++m_actionsCounts[si];
1098 m_actionsQueue.push( [this, si, iAlgo, subSlotIndex]() {
1099 --this->m_actionsCounts[si]; // no bound check needed as decrements/increments are balanced in the current setup
1100 return this->updateStates( -1, iAlgo, subSlotIndex, si );
1101 } );
1102
1103 return sc;
1104 }
1105
1106 //---------------------------------------------------------------------------
1107
1108 // Method to inform the scheduler about event views
1109
1110 StatusCode AvalancheSchedulerSvc::scheduleEventView( const EventContext* sourceContext, const std::string& nodeName,
1111 std::unique_ptr<EventContext> viewContext ) {
1112 // Prevent view nesting
1113 if ( sourceContext->usesSubSlot() ) {
1114 fatal() << "Attempted to nest EventViews at node " << nodeName << ": this is not supported" << endmsg;
1115 return StatusCode::FAILURE;
1116 }
1117
1118 ON_VERBOSE verbose() << "Queuing a view for [" << viewContext.get() << "]" << endmsg;
1119
1120 // It's not possible to create an std::functional from a move-capturing lambda
1121 // So, we have to release the unique pointer
1122 auto action = [this, slotIndex = sourceContext->slot(), viewContextPtr = viewContext.release(),
1123 &nodeName]() -> StatusCode {
1124 // Attach the sub-slot to the top-level slot
1125 EventSlot& topSlot = this->m_eventSlots[slotIndex];
1126
1127 if ( viewContextPtr ) {
1128 // Re-create the unique pointer
1129 auto viewContext = std::unique_ptr<EventContext>( viewContextPtr );
1130 topSlot.addSubSlot( std::move( viewContext ), nodeName );
1131 return StatusCode::SUCCESS;
1132 } else {
1133 // Disable the view node if there are no views
1134 topSlot.disableSubSlots( nodeName );
1135 return StatusCode::SUCCESS;
1136 }
1137 };
1138
1139 m_actionsQueue.push( std::move( action ) );
1140
1141 return StatusCode::SUCCESS;
1142 }
IThreadPoolSvc::initPool
virtual StatusCode initPool(const int &poolSize)=0
Initializes the thread pool.
AvalancheSchedulerSvc::m_showDataFlow
Gaudi::Property< bool > m_showDataFlow
Definition: AvalancheSchedulerSvc.h:166
ON_DEBUG
#define ON_DEBUG
Definition: AvalancheSchedulerSvc.cpp:31
AvalancheSchedulerSvc::isStalled
bool isStalled(const EventSlot &) const
Check if scheduling in a particular slot is in a stall.
Definition: AvalancheSchedulerSvc.cpp:735
IOBoundAlgTask
Wrapper around I/O-bound Gaudi-algorithms.
Definition: IOBoundAlgTask.h:23
AvalancheSchedulerSvc::tryPopFinishedEvent
StatusCode tryPopFinishedEvent(EventContext *&eventContext) override
Try to fetch an event from the scheduler.
Definition: AvalancheSchedulerSvc.cpp:553
AvalancheSchedulerSvc::scheduleEventView
virtual StatusCode scheduleEventView(const EventContext *sourceContext, const std::string &nodeName, std::unique_ptr< EventContext > viewContext) override
Method to inform the scheduler about event views.
Definition: AvalancheSchedulerSvc.cpp:1110
Service::initialize
StatusCode initialize() override
Definition: Service.cpp:60
SmartIF< IMessageSvc >
Service::serviceLocator
SmartIF< ISvcLocator > & serviceLocator() const override
Retrieve pointer to service locator.
Definition: Service.cpp:277
EventSlot
Class representing an event slot.
Definition: EventSlot.h:14
IAlgResourcePool::acquireAlgorithm
virtual StatusCode acquireAlgorithm(const std::string &name, IAlgorithm *&algo, bool blocking=false)=0
Acquire a certain algorithm using its name.
Cause::source::Task
EventContext::slot
ContextID_t slot() const
Definition: EventContext.h:41
std::unordered_set::empty
T empty(T... args)
AvalancheSchedulerSvc::INACTIVE
Definition: AvalancheSchedulerSvc.h:137
AvalancheSchedulerSvc::m_whiteboardSvcName
Gaudi::Property< std::string > m_whiteboardSvcName
Definition: AvalancheSchedulerSvc.h:142
AvalancheSchedulerSvc::m_IOBoundAlgosInFlight
unsigned int m_IOBoundAlgosInFlight
Number of algorithms presently in flight.
Definition: AvalancheSchedulerSvc.h:228
std::ofstream::open
T open(T... args)
concurrency::PrecedenceRulesGraph::getAlgorithmNode
AlgorithmNode * getAlgorithmNode(const std::string &algoName) const
Get the AlgorithmNode from by algorithm name using graph index.
Definition: PrecedenceRulesGraph.h:649
AvalancheSchedulerSvc::m_precSvc
SmartIF< IPrecedenceSvc > m_precSvc
A shortcut to the Precedence Service.
Definition: AvalancheSchedulerSvc.h:201
IPrecedenceSvc::iterate
virtual StatusCode iterate(EventSlot &, const Cause &)=0
Infer the precedence effect caused by an execution flow event.
AvalancheSchedulerSvc::m_actionsCounts
std::vector< unsigned int > m_actionsCounts
Bookkeeping of the number of actions in flight per slot.
Definition: AvalancheSchedulerSvc.h:263
AlgsExecutionStates::containsAny
bool containsAny(std::initializer_list< State > l) const
check if the collection contains at least one state of any listed types
Definition: AlgsExecutionStates.h:50
EventContext::evt
ContextEvt_t evt() const
Definition: EventContext.h:40
EventSlot::disableSubSlots
void disableSubSlots(const std::string &nodeName)
Disable event views for a given CF view node by registering an empty container Contact B.
Definition: EventSlot.h:68
Service::finalize
StatusCode finalize() override
Definition: Service.cpp:164
std::function< StatusCode()>
IAlgorithm::type
virtual const std::string & type() const =0
The type of the algorithm.
DataObjID
Definition: DataObjID.h:37
AvalancheSchedulerSvc::m_dumpIntraEventDynamics
Gaudi::Property< bool > m_dumpIntraEventDynamics
Definition: AvalancheSchedulerSvc.h:151
AvalancheSchedulerSvc::m_showDataDeps
Gaudi::Property< bool > m_showDataDeps
Definition: AvalancheSchedulerSvc.h:163
CommonMessagingBase::warning
MsgStream & warning() const
shortcut for the method msgStream(MSG::WARNING)
Definition: CommonMessaging.h:101
EventSlot::addSubSlot
void addSubSlot(std::unique_ptr< EventContext > viewContext, const std::string &nodeName)
Add a subslot to the slot (this constructs a new slot and registers it with the parent one)
Definition: EventSlot.h:51
IAlgExecStateSvc::eventStatus
virtual const EventStatus::Status & eventStatus(const EventContext &ctx) const =0
AvalancheSchedulerSvc::initialize
StatusCode initialize() override
Initialise.
Definition: AvalancheSchedulerSvc.cpp:63
AvalancheSchedulerSvc::dumpSchedulerState
void dumpSchedulerState(int iSlot)
Dump the state of the scheduler.
Definition: AvalancheSchedulerSvc.cpp:775
IAlgorithm.h
AlgoExecutionTask
Definition: AlgoExecutionTask.h:14
ThreadLocalContext.h
PrecedenceSvc
A service to resolve the task execution precedence.
Definition: PrecedenceSvc.h:21
std::to_string
T to_string(T... args)
IOBoundAlgTask.h
SmartIF::isValid
bool isValid() const
Allow for check if smart pointer is valid.
Definition: SmartIF.h:62
StatusCode::SUCCESS
constexpr static const auto SUCCESS
Definition: StatusCode.h:85
std::priority_queue::pop
T pop(T... args)
std::priority_queue::top
T top(T... args)
TimelineEvent::algorithm
std::string algorithm
Definition: ITimelineSvc.h:21
AlgExecState::filterPassed
bool filterPassed() const
Definition: IAlgExecStateSvc.h:31
AvalancheSchedulerSvc::ACTIVE
Definition: AvalancheSchedulerSvc.h:137
AvalancheSchedulerSvc::activate
void activate()
Activate scheduler.
Definition: AvalancheSchedulerSvc.cpp:368
std::vector::end
T end(T... args)
AvalancheSchedulerSvc::m_useDataLoader
Gaudi::Property< std::string > m_useDataLoader
Definition: AvalancheSchedulerSvc.h:158
EventSlot::allSubSlots
std::vector< EventSlot > allSubSlots
Actual sub-slot instances.
Definition: EventSlot.h:90
AvalancheSchedulerSvc::m_optimizationMode
Gaudi::Property< std::string > m_optimizationMode
Definition: AvalancheSchedulerSvc.h:149
IPrecedenceSvc::simulate
virtual StatusCode simulate(EventSlot &) const =0
Simulate execution flow.
CommonMessagingBase::info
MsgStream & info() const
shortcut for the method msgStream(MSG::INFO)
Definition: CommonMessaging.h:104
IPrecedenceSvc::CFRulesResolved
virtual bool CFRulesResolved(EventSlot &) const =0
Check if control flow rules are resolved.
TimelineEvent
Definition: ITimelineSvc.h:13
IAlgResourcePool::getFlatAlgList
virtual std::list< IAlgorithm * > getFlatAlgList()=0
Get the flat list of algorithms.
std::priority_queue::push
T push(T... args)
MSG::VERBOSE
Definition: IMessageSvc.h:15
SmartIF::get
TYPE * get() const
Get interface pointer.
Definition: SmartIF.h:76
EventContext
This class represents an entry point to all the event specific data.
Definition: EventContext.h:24
std::map
STL class.
AvalancheSchedulerSvc::algname2index
unsigned int algname2index(const std::string &algoname)
Convert a name to an integer.
Definition: AvalancheSchedulerSvc.cpp:443
std::stringstream
STL class.
DataObjID::fullKey
std::string fullKey() const
Definition: DataObjID.cpp:88
std::unique_ptr::release
T release(T... args)
AvalancheSchedulerSvc::m_isActive
std::atomic< ActivationState > m_isActive
Flag to track if the scheduler is active or not.
Definition: AvalancheSchedulerSvc.h:183
std::setw
T setw(T... args)
CommonMessaging< implements< IService, IProperty, IStateful > >::msgLevel
MSG::Level msgLevel() const
get the cached level (originally extracted from the embedded MsgStream)
Definition: CommonMessaging.h:138
AvalancheSchedulerSvc::m_algname_index_map
std::unordered_map< std::string, unsigned int > m_algname_index_map
Map to bookkeep the information necessary to the name2index conversion.
Definition: AvalancheSchedulerSvc.h:192
std::vector::resize
T resize(T... args)
AlgoExecutionTask::execute
tbb::task * execute() override
Definition: AlgoExecutionTask.cpp:21
AvalancheSchedulerSvc::m_checkDeps
Gaudi::Property< bool > m_checkDeps
Definition: AvalancheSchedulerSvc.h:156
std::string
STL class.
AvalancheSchedulerSvc::m_useIOBoundAlgScheduler
Gaudi::Property< bool > m_useIOBoundAlgScheduler
Definition: AvalancheSchedulerSvc.h:153
IAlgExecStateSvc::algExecState
virtual const AlgExecState & algExecState(const Gaudi::StringKey &algName, const EventContext &ctx) const =0
DECLARE_COMPONENT
#define DECLARE_COMPONENT(type)
Definition: PluginServiceV1.h:46
concurrency::AlgorithmNode::getAlgoIndex
const unsigned int & getAlgoIndex() const
Get algorithm index.
Definition: PrecedenceRulesGraph.h:513
AvalancheSchedulerSvc::m_freeSlots
std::atomic_int m_freeSlots
Atomic to account for asyncronous updates by the scheduler wrt the rest.
Definition: AvalancheSchedulerSvc.h:213
std::vector::at
T at(T... args)
IThreadPoolSvc::terminatePool
virtual StatusCode terminatePool()=0
Finalize the thread pool.
AvalancheSchedulerSvc::pushNewEvents
StatusCode pushNewEvents(std::vector< EventContext * > &eventContexts) override
Definition: AvalancheSchedulerSvc.cpp:515
ISvcLocator::service
StatusCode service(const Gaudi::Utils::TypeNameString &name, T *&svc, bool createIf=true)
Templated method to access a service by name.
Definition: ISvcLocator.h:76
Service::name
const std::string & name() const override
Retrieve name of the service.
Definition: Service.cpp:274
AvalancheSchedulerSvc::m_algosInFlight
unsigned int m_algosInFlight
Number of algorithms presently in flight.
Definition: AvalancheSchedulerSvc.h:225
std::vector::push_back
T push_back(T... args)
std::ofstream
STL class.
CommonMessagingBase::error
MsgStream & error() const
shortcut for the method msgStream(MSG::ERROR)
Definition: CommonMessaging.h:98
AvalancheSchedulerSvc::m_verboseSubSlots
Gaudi::Property< bool > m_verboseSubSlots
Definition: AvalancheSchedulerSvc.h:172
AvalancheSchedulerSvc::m_algResourcePool
SmartIF< IAlgResourcePool > m_algResourcePool
Cache for the algorithm resource pool.
Definition: AvalancheSchedulerSvc.h:255
IDataManagerSvc.h
AlgsExecutionStates
The AlgsExecutionStates encodes the state machine for the execution of algorithms within a single eve...
Definition: AlgsExecutionStates.h:26
IAlgExecStateSvc::dump
virtual void dump(std::ostringstream &ost, const EventContext &ctx) const =0
AvalancheSchedulerSvc::promoteToScheduled
StatusCode promoteToScheduled(unsigned int iAlgo, int si, EventContext *)
Algorithm promotion.
Definition: AvalancheSchedulerSvc.cpp:885
AvalancheSchedulerSvc::popFinishedEvent
StatusCode popFinishedEvent(EventContext *&eventContext) override
Blocks until an event is available.
Definition: AvalancheSchedulerSvc.cpp:532
std::thread::join
T join(T... args)
Cause
Definition: PrecedenceRulesGraph.h:387
IPrecedenceSvc::dumpPrecedenceRules
virtual void dumpPrecedenceRules(EventSlot &)=0
Dump precedence rules.
Algorithm.h
std::priority_queue
STL class.
AvalancheSchedulerSvc::m_showControlFlow
Gaudi::Property< bool > m_showControlFlow
Definition: AvalancheSchedulerSvc.h:169
StatusCode
This class is used for returning status codes from appropriate routines.
Definition: StatusCode.h:50
AvalancheSchedulerSvc::m_whiteboard
SmartIF< IHiveWhiteBoard > m_whiteboard
A shortcut to the whiteboard.
Definition: AvalancheSchedulerSvc.h:204
std::ostringstream
STL class.
AvalancheSchedulerSvc::m_finishedEvents
tbb::concurrent_bounded_queue< EventContext * > m_finishedEvents
Queue of finished events.
Definition: AvalancheSchedulerSvc.h:216
AvalancheSchedulerSvc::m_algname_vect
std::vector< std::string > m_algname_vect
Vector to bookkeep the information necessary to the index2name conversion.
Definition: AvalancheSchedulerSvc.h:198
std::ofstream::close
T close(T... args)
AlgExecState::execStatus
const StatusCode & execStatus() const
Definition: IAlgExecStateSvc.h:33
AlgsExecutionStates::sizeOfSubset
size_t sizeOfSubset(State state) const
Definition: AlgsExecutionStates.h:64
AlgsExecutionStates::set
StatusCode set(unsigned int iAlgo, State newState)
Definition: AlgsExecutionStates.cpp:13
CommonMessagingBase::verbose
MsgStream & verbose() const
shortcut for the method msgStream(MSG::VERBOSE)
Definition: CommonMessaging.h:110
compareRootHistos.state
def state
Definition: compareRootHistos.py:455
EventStatus::AlgStall
Definition: IAlgExecStateSvc.h:63
AvalancheSchedulerSvc::m_IOBoundAlgSchedulerSvcName
Gaudi::Property< std::string > m_IOBoundAlgSchedulerSvcName
Definition: AvalancheSchedulerSvc.h:143
std::ostringstream::str
T str(T... args)
EventStatus::Success
Definition: IAlgExecStateSvc.h:63
IAlgExecStateSvc::setEventStatus
virtual void setEventStatus(const EventStatus::Status &sc, const EventContext &ctx)=0
Cause::source::Root
CommonMessagingBase::debug
MsgStream & debug() const
shortcut for the method msgStream(MSG::DEBUG)
Definition: CommonMessaging.h:107
concurrency::PrecedenceRulesGraph::getControlFlowNodeCounter
unsigned int getControlFlowNodeCounter() const
Get total number of control flow graph nodes.
Definition: PrecedenceRulesGraph.h:657
AvalancheSchedulerSvc::finalize
StatusCode finalize() override
Finalise.
Definition: AvalancheSchedulerSvc.cpp:337
std::unordered_set< DataObjID, DataObjID_Hasher >
IHiveWhiteBoard::getNumberOfStores
virtual size_t getNumberOfStores() const =0
Get the number of 'slots'.
DataHandleHolderVisitor.h
AvalancheSchedulerSvc::m_threadPoolSize
Gaudi::Property< int > m_threadPoolSize
Definition: AvalancheSchedulerSvc.h:139
IPrecedenceSvc::dumpDataFlow
virtual void dumpDataFlow() const =0
ITimelineSvc
Definition: ITimelineSvc.h:27
StatusCode::isSuccess
bool isSuccess() const
Definition: StatusCode.h:267
AvalancheSchedulerSvc::m_threadPoolSvc
SmartIF< IThreadPoolSvc > m_threadPoolSvc
Definition: AvalancheSchedulerSvc.h:268
std::list< IAlgorithm * >
AvalancheSchedulerSvc::m_IOBoundAlgScheduler
SmartIF< IAccelerator > m_IOBoundAlgScheduler
A shortcut to IO-bound algorithm scheduler.
Definition: AvalancheSchedulerSvc.h:207
std::max
T max(T... args)
IAlgorithm
The IAlgorithm is the interface implemented by the Algorithm base class.
Definition: IAlgorithm.h:28
AlgsExecutionStates::State
State
Execution states of the algorithms.
Definition: AlgsExecutionStates.h:29
Gaudi::Hive::setCurrentContext
GAUDI_API void setCurrentContext(const EventContext *ctx)
Definition: ThreadLocalContext.cpp:31
std::move
T move(T... args)
AvalancheSchedulerSvc
Definition: AvalancheSchedulerSvc.h:100
AvalancheSchedulerSvc::promoteToAsyncScheduled
StatusCode promoteToAsyncScheduled(unsigned int iAlgo, int si, EventContext *)
Definition: AvalancheSchedulerSvc.cpp:946
EventContext::subSlot
ContextID_t subSlot() const
Definition: EventContext.h:42
std::unique_ptr::get
T get(T... args)
std::unordered_set::insert
T insert(T... args)
DataHandleHolderBase::addDependency
void addDependency(const DataObjID &id, const Gaudi::DataHandle::Mode &mode) override
Definition: DataHandleHolderBase.h:58
DataHandleHolderBase::outputDataObjs
const DataObjIDColl & outputDataObjs() const override
Definition: DataHandleHolderBase.h:56
std::find_if
T find_if(T... args)
std::unordered_set::size
T size(T... args)
StatusCode::ignore
const StatusCode & ignore() const
Ignore/check StatusCode.
Definition: StatusCode.h:153
AvalancheSchedulerSvc::pushNewEvent
StatusCode pushNewEvent(EventContext *eventContext) override
Make an event available to the scheduler.
Definition: AvalancheSchedulerSvc.cpp:457
std::vector::assign
T assign(T... args)
std::vector
STL class.
std::unique_ptr< EventContext >
AvalancheSchedulerSvc::m_simulateExecution
Gaudi::Property< bool > m_simulateExecution
Definition: AvalancheSchedulerSvc.h:146
std::vector::begin
T begin(T... args)
AlgsExecutionStates::begin
Iterator begin(State kind)
Definition: AlgsExecutionStates.h:103
std::chrono::high_resolution_clock::now
T now(T... args)
std::any_of
T any_of(T... args)
AvalancheSchedulerSvc::m_enableCondSvc
Gaudi::Property< bool > m_enableCondSvc
Definition: AvalancheSchedulerSvc.h:161
AvalancheSchedulerSvc::index2algname
const std::string & index2algname(unsigned int index)
Convert an integer to a name.
Definition: AvalancheSchedulerSvc.cpp:439
ConcurrencyFlags.h
Gaudi::Algorithm
Base class from which all concrete algorithm classes should be derived.
Definition: Algorithm.h:79
AvalancheSchedulerSvc::m_maxAlgosInFlight
size_t m_maxAlgosInFlight
Definition: AvalancheSchedulerSvc.h:270
IPrecedenceSvc::getPriority
virtual uint getPriority(const std::string &) const =0
Get task priority.
std::vector::back
T back(T... args)
gaudirun.s
string s
Definition: gaudirun.py:316
StatusCode::FAILURE
constexpr static const auto FAILURE
Definition: StatusCode.h:86
AvalancheSchedulerSvc::promoteToExecuted
StatusCode promoteToExecuted(unsigned int iAlgo, int si, IAlgorithm *algo, EventContext *)
The call to this method is triggered only from within the AlgoExecutionTask.
Definition: AvalancheSchedulerSvc.cpp:1002
Gaudi::DataHandle::Writer
Definition: DataHandle.h:29
std::hex
T hex(T... args)
MSG::ERROR
Definition: IMessageSvc.h:15
AvalancheSchedulerSvc::m_condSvc
SmartIF< ICondSvc > m_condSvc
A shortcut to service for Conditions handling.
Definition: AvalancheSchedulerSvc.h:222
AvalancheSchedulerSvc::eventFailed
void eventFailed(EventContext *eventContext)
Method to execute if an event failed.
Definition: AvalancheSchedulerSvc.cpp:754
AvalancheSchedulerSvc::freeSlots
unsigned int freeSlots() override
Get free slots number.
Definition: AvalancheSchedulerSvc.cpp:526
AvalancheSchedulerSvc::m_eventSlots
std::vector< EventSlot > m_eventSlots
Vector of events slots.
Definition: AvalancheSchedulerSvc.h:210
EventSlot::complete
bool complete
Flags completion of the event.
Definition: EventSlot.h:79
std::sort
T sort(T... args)
AvalancheSchedulerSvc::promoteToAsyncExecuted
StatusCode promoteToAsyncExecuted(unsigned int iAlgo, int si, IAlgorithm *algo, EventContext *)
The call to this method is triggered only from within the IOBoundAlgTask.
Definition: AvalancheSchedulerSvc.cpp:1056
AvalancheSchedulerSvc::deactivate
StatusCode deactivate()
Deactivate scheduler.
Definition: AvalancheSchedulerSvc.cpp:413
StatusCode::isFailure
bool isFailure() const
Definition: StatusCode.h:130
IAccelerator::push
virtual StatusCode push(IAlgTask &task)=0
AvalancheSchedulerSvc::FAILURE
Definition: AvalancheSchedulerSvc.h:137
DataHandleHolderBase::inputDataObjs
const DataObjIDColl & inputDataObjs() const override
Definition: DataHandleHolderBase.h:55
EventSlot::reset
void reset(EventContext *theeventContext)
Reset all resources in order to reuse the slot (thread-unsafe)
Definition: EventSlot.h:39
AvalancheSchedulerSvc::updateStates
StatusCode updateStates(int si=-1, int algo_index=-1, int sub_slot=-1, int source_slot=-1)
Loop on algorithm in the slots and promote them to successive states (-1 for algo_index means skippin...
Definition: AvalancheSchedulerSvc.cpp:576
IPrecedenceSvc::dumpControlFlow
virtual void dumpControlFlow() const =0
Dump precedence rules.
IAlgResourcePool::releaseAlgorithm
virtual StatusCode releaseAlgorithm(const std::string &name, IAlgorithm *&algo)=0
Release a certain algorithm.
AlgExecState
Definition: IAlgExecStateSvc.h:27
IPrecedenceSvc::isBlocking
virtual bool isBlocking(const std::string &) const =0
Check if a task is CPU-blocking.
std::thread
STL class.
AvalancheSchedulerSvc::m_algExecStateSvc
SmartIF< IAlgExecStateSvc > m_algExecStateSvc
Algorithm execution state manager.
Definition: AvalancheSchedulerSvc.h:219
AvalancheSchedulerSvc::m_maxIOBoundAlgosInFlight
Gaudi::Property< unsigned int > m_maxIOBoundAlgosInFlight
Definition: AvalancheSchedulerSvc.h:144
CommonMessagingBase::fatal
MsgStream & fatal() const
shortcut for the method msgStream(MSG::FATAL)
Definition: CommonMessaging.h:92
IPrecedenceSvc::printState
virtual const std::string printState(EventSlot &) const =0
AvalancheSchedulerSvc::m_maxEventsInFlight
size_t m_maxEventsInFlight
Definition: AvalancheSchedulerSvc.h:269
endmsg
MsgStream & endmsg(MsgStream &s)
MsgStream Modifier: endmsg. Calls the output method of the MsgStream.
Definition: MsgStream.h:192
std::initializer_list
AvalancheSchedulerSvc.h
AvalancheSchedulerErrorTest.threads
int threads
Definition: AvalancheSchedulerErrorTest.py:17
EventSlot::eventContext
std::unique_ptr< EventContext > eventContext
Cache for the eventContext.
Definition: EventSlot.h:73
AvalancheSchedulerSvc::m_actionsQueue
tbb::concurrent_bounded_queue< action > m_actionsQueue
Queue where closures are stored and picked for execution.
Definition: AvalancheSchedulerSvc.h:260
AlgoExecutionTask.h
PrecedenceSvc::getRules
const concurrency::PrecedenceRulesGraph * getRules() const
Precedence rules accessor.
Definition: PrecedenceSvc.h:63
Gaudi::Algorithm::name
const std::string & name() const override
The identifying name of the algorithm object.
Definition: Algorithm.cpp:645
EventSlot::algsStates
AlgsExecutionStates algsStates
Vector of algorithms states.
Definition: EventSlot.h:75
std::vector::reserve
T reserve(T... args)
std::vector::emplace_back
T emplace_back(T... args)
AvalancheSchedulerSvc::m_thread
std::thread m_thread
The thread in which the activate function runs.
Definition: AvalancheSchedulerSvc.h:186
EventContext::usesSubSlot
bool usesSubSlot() const
Definition: EventContext.h:43
ON_VERBOSE
#define ON_VERBOSE
Definition: AvalancheSchedulerSvc.cpp:32
AlgsExecutionStates::end
Iterator end(State kind)
Definition: AlgsExecutionStates.h:104