| The Gaudi Framework v28r2p1 (f1a77ff4)

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

- File List
- File Members

AvalancheSchedulerSvc.h
Go to the documentation of this file.
1 #ifndef GAUDIHIVE_AVALANCHESCHEDULERSVC_H
2 #define GAUDIHIVE_AVALANCHESCHEDULERSVC_H
3
4 // Local includes
5 #include "AlgsExecutionStates.h"
6 #include "EventSlot.h"
7 #include "ExecutionFlowManager.h"
8
9 // Framework include files
10 #include "GaudiKernel/IAccelerator.h"
11 #include "GaudiKernel/IAlgExecStateSvc.h"
12 #include "GaudiKernel/IAlgResourcePool.h"
13 #include "GaudiKernel/IHiveWhiteBoard.h"
14 #include "GaudiKernel/IRunable.h"
15 #include "GaudiKernel/IScheduler.h"
16 #include "GaudiKernel/IThreadPoolSvc.h"
17 #include "GaudiKernel/Service.h"
18
19 // C++ include files
20 #include <functional>
21 #include <string>
22 #include <thread>
23 #include <unordered_map>
24 #include <vector>
25
26 // External libs
27 #include "tbb/concurrent_queue.h"
28 #include "tbb/task.h"
29
30 typedef AlgsExecutionStates::State State;
31 typedef std::function<StatusCode()> action;
32
33 //---------------------------------------------------------------------------
34
100 class AvalancheSchedulerSvc : public extends<Service, IScheduler>
101 {
102 public:
104 using extends::extends;
105
107 ~AvalancheSchedulerSvc() override = default;
108
110 StatusCode initialize() override;
111
113 StatusCode finalize() override;
114
116 StatusCode pushNewEvent( EventContext* eventContext ) override;
117
118 // Make multiple events available to the scheduler
119 StatusCode pushNewEvents( std::vector<EventContext*>& eventContexts ) override;
120
122 StatusCode popFinishedEvent( EventContext*& eventContext ) override;
123
125 StatusCode tryPopFinishedEvent( EventContext*& eventContext ) override;
126
128 unsigned int freeSlots() override;
129
130 private:
131 enum ActivationState { INACTIVE = 0, ACTIVE = 1, FAILURE = 2 };
132
133 Gaudi::Property<int> m_maxEventsInFlight{this, "MaxEventsInFlight", 0,
134 "Maximum number of event processed simultaneously"};
135 Gaudi::Property<int> m_threadPoolSize{this, "ThreadPoolSize", -1,
136 "Size of the threadpool initialised by TBB; a value of -1 gives TBB the freedom to choose"};
137 Gaudi::Property<std::string> m_whiteboardSvcName{this, "WhiteboardSvc", "EventDataSvc",
138 "The whiteboard name"};
139 Gaudi::Property<std::string> m_IOBoundAlgSchedulerSvcName{this, "IOBoundAlgSchedulerSvc",
140 "IOBoundAlgSchedulerSvc"};
141 Gaudi::Property<unsigned int> m_maxAlgosInFlight{this, "MaxAlgosInFlight", 1,
142 "[[deprecated]] Taken from the whiteboard"};
143 Gaudi::Property<unsigned int> m_maxIOBoundAlgosInFlight{this, "MaxIOBoundAlgosInFlight", 0,
144 "Maximum number of simultaneous I/O-bound algorithms"};
145 Gaudi::Property<bool> m_simulateExecution{this, "SimulateExecution", false,
146 "Flag to perform single-pass simulation of execution flow before the actual execution"};
147 Gaudi::Property<std::string> m_optimizationMode{this, "Optimizer", "",
148 "The following modes are currently available: PCE, COD, DRE, E"};
149 Gaudi::Property<bool> m_dumpIntraEventDynamics{this, "DumpIntraEventDynamics", false,
150 "Dump intra-event concurrency dynamics to csv file"};
151 Gaudi::Property<bool> m_useIOBoundAlgScheduler{this, "PreemptiveIOBoundTasks", false,
152 "Turn on preemptive way of scheduling of I/O-bound algorithms"};
153 Gaudi::Property<std::vector<std::vector<std::string>>> m_algosDependencies{this, "AlgosDependencies", {},
154 "[[deprecated]]"};
155 Gaudi::Property<bool> m_checkDeps{this, "CheckDependencies", false, "[[deprecated]]"};
156
157 // Utils and shortcuts ----------------------------------------------------
158
160 void activate();
161
163 StatusCode deactivate();
164
166 std::atomic<ActivationState> m_isActive{INACTIVE};
167
169 std::thread m_thread;
170
172 inline unsigned int algname2index( const std::string& algoname );
173
175 std::unordered_map<std::string, unsigned int> m_algname_index_map;
176
178 inline const std::string& index2algname( unsigned int index );
179
181 std::vector<std::string> m_algname_vect;
182
184 SmartIF<IHiveWhiteBoard> m_whiteboard;
185
187 SmartIF<IAccelerator> m_IOBoundAlgScheduler;
188
190 std::vector<EventSlot> m_eventSlots;
191
193 std::atomic_int m_freeSlots;
194
196 tbb::concurrent_bounded_queue<EventContext*> m_finishedEvents;
197
199 StatusCode eventFailed( EventContext* eventContext );
200
202 SmartIF<IAlgExecStateSvc> m_algExecStateSvc;
203
204 // States management ------------------------------------------------------
205
207 unsigned int m_algosInFlight = 0;
208
210 unsigned int m_IOBoundAlgosInFlight = 0;
211
214 StatusCode updateStates( int si = -1, const std::string& algo_name = std::string() );
215
217 StatusCode promoteToScheduled( unsigned int iAlgo, int si );
218 StatusCode promoteToAsyncScheduled( unsigned int iAlgo, int si ); // tests of an asynchronous scheduler
219 StatusCode promoteToExecuted( unsigned int iAlgo, int si, IAlgorithm* algo, EventContext* );
220 StatusCode promoteToAsyncExecuted( unsigned int iAlgo, int si, IAlgorithm* algo,
221 EventContext* ); // tests of an asynchronous scheduler
222 StatusCode promoteToFinished( unsigned int iAlgo, int si );
223
225 StatusCode isStalled( int si );
226
228 void dumpSchedulerState( int iSlot );
229
231 bool m_updateNeeded = true;
232
233 // Algos Management -------------------------------------------------------
235 SmartIF<IAlgResourcePool> m_algResourcePool;
236
238 StatusCode m_drain();
239
240 // Actions management -----------------------------------------------------
241
243 tbb::concurrent_bounded_queue<action> m_actionsQueue;
244
245 // helper task to enqueue the scheduler's actions (closures)
246 struct enqueueSchedulerActionTask: public tbb::task {
247
248 std::function<StatusCode()> m_closure;
249 SmartIF<AvalancheSchedulerSvc> m_scheduler;
250
251 enqueueSchedulerActionTask(AvalancheSchedulerSvc* scheduler, std::function<StatusCode()> _closure) :
252 m_closure(_closure), m_scheduler(scheduler) {}
253
254 tbb::task* execute() override {
255 m_scheduler->m_actionsQueue.push(m_closure);
256 return nullptr;
257 }
258 };
259
260 // ------------------------------------------------------------------------
261
263 concurrency::ExecutionFlowManager m_efManager;
264
265 // Service for thread pool initialization
266 SmartIF<IThreadPoolSvc> m_threadPoolSvc;
267
268 bool m_first = true;
269
270 class SchedulerState
271 {
272
273 public:
274 SchedulerState( Algorithm* a, EventContext* e, pthread_t t ) : m_a( a ), m_e( *e ), m_t( t ) {}
275
276 Algorithm* alg() const { return m_a; }
277 EventContext ctx() const { return m_e; }
278 pthread_t thread() const { return m_t; }
279
280 friend std::ostream& operator<<( std::ostream& os, const SchedulerState& ss )
281 {
282 os << ss.ctx() << " a: " << ss.alg()->name() << " [" << std::hex << ss.alg() << std::dec << "] t: 0x"
283 << std::hex << ss.thread() << std::dec;
284 return os;
285 }
286
287 bool operator==( const SchedulerState& ss ) const { return ( m_a == ss.alg() ); }
288
289 bool operator==( Algorithm* a ) const { return ( m_a == a ); }
290
291 bool operator<( const SchedulerState& rhs ) const { return ( m_a < rhs.alg() ); }
292
293 private:
294 Algorithm* m_a;
295 EventContext m_e;
296 pthread_t m_t;
297 };
298
299 static std::list<SchedulerState> m_sState;
300 static std::mutex m_ssMut;
301
302 public:
303 void addAlg( Algorithm*, EventContext*, pthread_t );
304 bool delAlg( Algorithm* );
305 void dumpState() override;
306
307 private:
308 void dumpState( std::ostringstream& );
309 };
310
311 #endif // GAUDIHIVE_AVALANCHESCHEDULERSVC_H
IThreadPoolSvc.h
AvalancheSchedulerSvc::SchedulerState::m_e
EventContext m_e
Definition: AvalancheSchedulerSvc.h:295
AvalancheSchedulerSvc::tryPopFinishedEvent
StatusCode tryPopFinishedEvent(EventContext *&eventContext) override
Try to fetch an event from the scheduler.
Definition: AvalancheSchedulerSvc.cpp:492
SmartIF< IHiveWhiteBoard >
AvalancheSchedulerSvc::INACTIVE
Definition: AvalancheSchedulerSvc.h:131
AvalancheSchedulerSvc::m_whiteboardSvcName
Gaudi::Property< std::string > m_whiteboardSvcName
Definition: AvalancheSchedulerSvc.h:137
AvalancheSchedulerSvc::m_IOBoundAlgosInFlight
unsigned int m_IOBoundAlgosInFlight
Number of algoritms presently in flight.
Definition: AvalancheSchedulerSvc.h:210
Algorithm::name
const std::string & name() const override
The identifying name of the algorithm object.
Definition: Algorithm.cpp:715
std::atomic< ActivationState >
AvalancheSchedulerSvc::enqueueSchedulerActionTask::m_scheduler
SmartIF< AvalancheSchedulerSvc > m_scheduler
Definition: AvalancheSchedulerSvc.h:249
IAlgExecStateSvc.h
ExecutionFlowManager.h
std::function< StatusCode()>
Gaudi::Property
Implementation of property with value of concrete type.
Definition: Property.h:314
AvalancheSchedulerSvc::m_dumpIntraEventDynamics
Gaudi::Property< bool > m_dumpIntraEventDynamics
Definition: AvalancheSchedulerSvc.h:149
AvalancheSchedulerSvc::SchedulerState::m_t
pthread_t m_t
Definition: AvalancheSchedulerSvc.h:296
AvalancheSchedulerSvc::SchedulerState::operator==
bool operator==(const SchedulerState &ss) const
Definition: AvalancheSchedulerSvc.h:287
AvalancheSchedulerSvc::initialize
StatusCode initialize() override
Initialise.
Definition: AvalancheSchedulerSvc.cpp:45
AvalancheSchedulerSvc::dumpSchedulerState
void dumpSchedulerState(int iSlot)
Dump the state of the scheduler.
Definition: AvalancheSchedulerSvc.cpp:755
AvalancheSchedulerSvc::promoteToScheduled
StatusCode promoteToScheduled(unsigned int iAlgo, int si)
Algorithm promotion.
Definition: AvalancheSchedulerSvc.cpp:826
IRunable.h
AvalancheSchedulerSvc::enqueueSchedulerActionTask::enqueueSchedulerActionTask
enqueueSchedulerActionTask(AvalancheSchedulerSvc *scheduler, std::function< StatusCode()> _closure)
Definition: AvalancheSchedulerSvc.h:251
AvalancheSchedulerSvc::isStalled
StatusCode isStalled(int si)
Check if the scheduling is in a stall.
Definition: AvalancheSchedulerSvc.cpp:731
AvalancheSchedulerSvc::delAlg
bool delAlg(Algorithm *)
Definition: AvalancheSchedulerSvc.cpp:1053
IAlgResourcePool.h
AvalancheSchedulerSvc::ACTIVE
Definition: AvalancheSchedulerSvc.h:131
AvalancheSchedulerSvc::activate
void activate()
Activate scheduler.
Definition: AvalancheSchedulerSvc.cpp:305
AvalancheSchedulerSvc::m_algosDependencies
Gaudi::Property< std::vector< std::vector< std::string > > > m_algosDependencies
Definition: AvalancheSchedulerSvc.h:153
AvalancheSchedulerSvc::m_optimizationMode
Gaudi::Property< std::string > m_optimizationMode
Definition: AvalancheSchedulerSvc.h:147
IScheduler.h
AvalancheSchedulerSvc::SchedulerState::operator<<
friend std::ostream & operator<<(std::ostream &os, const SchedulerState &ss)
Definition: AvalancheSchedulerSvc.h:280
AvalancheSchedulerSvc::promoteToAsyncScheduled
StatusCode promoteToAsyncScheduled(unsigned int iAlgo, int si)
Definition: AvalancheSchedulerSvc.cpp:890
EventContext
This class represents an entry point to all the event specific data.
Definition: EventContext.h:25
AvalancheSchedulerSvc::algname2index
unsigned int algname2index(const std::string &algoname)
Convert a name to an integer.
Definition: AvalancheSchedulerSvc.cpp:374
std::mutex
State
AlgsExecutionStates::State State
Definition: AvalancheSchedulerSvc.h:30
AvalancheSchedulerSvc::addAlg
void addAlg(Algorithm *, EventContext *, pthread_t)
Definition: AvalancheSchedulerSvc.cpp:1045
AvalancheSchedulerSvc::m_isActive
std::atomic< ActivationState > m_isActive
Flag to track if the scheduler is active or not.
Definition: AvalancheSchedulerSvc.h:166
AvalancheSchedulerSvc::m_algname_index_map
std::unordered_map< std::string, unsigned int > m_algname_index_map
Map to bookkeep the information necessary to the name2index conversion.
Definition: AvalancheSchedulerSvc.h:175
AvalancheSchedulerSvc::m_checkDeps
Gaudi::Property< bool > m_checkDeps
Definition: AvalancheSchedulerSvc.h:155
std::string
STL class.
AvalancheSchedulerSvc::m_useIOBoundAlgScheduler
Gaudi::Property< bool > m_useIOBoundAlgScheduler
Definition: AvalancheSchedulerSvc.h:151
AvalancheSchedulerSvc::SchedulerState::operator<
bool operator<(const SchedulerState &rhs) const
Definition: AvalancheSchedulerSvc.h:291
Service.h
AvalancheSchedulerSvc::m_freeSlots
std::atomic_int m_freeSlots
Atomic to account for asyncronous updates by the scheduler wrt the rest.
Definition: AvalancheSchedulerSvc.h:193
AvalancheSchedulerSvc::SchedulerState::alg
Algorithm * alg() const
Definition: AvalancheSchedulerSvc.h:276
AvalancheSchedulerSvc::pushNewEvents
StatusCode pushNewEvents(std::vector< EventContext * > &eventContexts) override
Definition: AvalancheSchedulerSvc.cpp:436
AvalancheSchedulerSvc::m_algosInFlight
unsigned int m_algosInFlight
Number of algoritms presently in flight.
Definition: AvalancheSchedulerSvc.h:207
AvalancheSchedulerSvc::m_sState
static std::list< SchedulerState > m_sState
Definition: AvalancheSchedulerSvc.h:299
AvalancheSchedulerSvc::m_algResourcePool
SmartIF< IAlgResourcePool > m_algResourcePool
Cache for the algorithm resource pool.
Definition: AvalancheSchedulerSvc.h:235
concurrency::ExecutionFlowManager
Manage the execution flow using a graph of task precedence rules Once initialized, the graph is const and can be shared across events.
Definition: ExecutionFlowManager.h:21
AvalancheSchedulerSvc::enqueueSchedulerActionTask::execute
tbb::task * execute() override
Definition: AvalancheSchedulerSvc.h:254
AvalancheSchedulerSvc::popFinishedEvent
StatusCode popFinishedEvent(EventContext *&eventContext) override
Blocks until an event is availble.
Definition: AvalancheSchedulerSvc.cpp:470
AvalancheSchedulerSvc::ActivationState
ActivationState
Definition: AvalancheSchedulerSvc.h:131
StatusCode
This class is used for returning status codes from appropriate routines.
Definition: StatusCode.h:26
AvalancheSchedulerSvc::m_whiteboard
SmartIF< IHiveWhiteBoard > m_whiteboard
A shortcut to the whiteboard.
Definition: AvalancheSchedulerSvc.h:184
std::ostringstream
STL class.
AvalancheSchedulerSvc::m_finishedEvents
tbb::concurrent_bounded_queue< EventContext * > m_finishedEvents
Queue of finished events.
Definition: AvalancheSchedulerSvc.h:196
AvalancheSchedulerSvc::m_algname_vect
std::vector< std::string > m_algname_vect
Vector to bookkeep the information necessary to the index2name conversion.
Definition: AvalancheSchedulerSvc.h:181
AvalancheSchedulerSvc::SchedulerState::SchedulerState
SchedulerState(Algorithm *a, EventContext *e, pthread_t t)
Definition: AvalancheSchedulerSvc.h:274
AvalancheSchedulerSvc::m_IOBoundAlgSchedulerSvcName
Gaudi::Property< std::string > m_IOBoundAlgSchedulerSvcName
Definition: AvalancheSchedulerSvc.h:139
AvalancheSchedulerSvc::SchedulerState::ctx
EventContext ctx() const
Definition: AvalancheSchedulerSvc.h:277
action
std::function< StatusCode()> action
Definition: AvalancheSchedulerSvc.h:31
AvalancheSchedulerSvc::finalize
StatusCode finalize() override
Finalise.
Definition: AvalancheSchedulerSvc.cpp:273
AvalancheSchedulerSvc::m_ssMut
static std::mutex m_ssMut
Definition: AvalancheSchedulerSvc.h:300
AvalancheSchedulerSvc::~AvalancheSchedulerSvc
~AvalancheSchedulerSvc() override=default
Destructor.
AvalancheSchedulerSvc::m_efManager
concurrency::ExecutionFlowManager m_efManager
Member to take care of the control flow.
Definition: AvalancheSchedulerSvc.h:263
AvalancheSchedulerSvc::m_threadPoolSize
Gaudi::Property< int > m_threadPoolSize
Definition: AvalancheSchedulerSvc.h:135
AvalancheSchedulerSvc::m_threadPoolSvc
SmartIF< IThreadPoolSvc > m_threadPoolSvc
Definition: AvalancheSchedulerSvc.h:266
std::list
STL class.
AvalancheSchedulerSvc::m_IOBoundAlgScheduler
SmartIF< IAccelerator > m_IOBoundAlgScheduler
A shortcut to IO-bound algorithm scheduler.
Definition: AvalancheSchedulerSvc.h:187
IAlgorithm
The IAlgorithm is the interface implemented by the Algorithm base class.
Definition: IAlgorithm.h:27
AvalancheSchedulerSvc::m_updateNeeded
bool m_updateNeeded
Keep track of update actions scheduled.
Definition: AvalancheSchedulerSvc.h:231
AvalancheSchedulerSvc
Introduction
Definition: AvalancheSchedulerSvc.h:100
AvalancheSchedulerSvc::m_maxEventsInFlight
Gaudi::Property< int > m_maxEventsInFlight
Definition: AvalancheSchedulerSvc.h:133
Algorithm
Base class from which all concrete algorithm classes should be derived.
Definition: Algorithm.h:78
AvalancheSchedulerSvc::promoteToFinished
StatusCode promoteToFinished(unsigned int iAlgo, int si)
AvalancheSchedulerSvc::pushNewEvent
StatusCode pushNewEvent(EventContext *eventContext) override
Make an event available to the scheduler.
Definition: AvalancheSchedulerSvc.cpp:387
AvalancheSchedulerSvc::m_maxAlgosInFlight
Gaudi::Property< unsigned int > m_maxAlgosInFlight
Definition: AvalancheSchedulerSvc.h:141
std::vector
STL class.
AvalancheSchedulerSvc::m_simulateExecution
Gaudi::Property< bool > m_simulateExecution
Definition: AvalancheSchedulerSvc.h:145
AvalancheSchedulerSvc::SchedulerState::m_a
Algorithm * m_a
Definition: AvalancheSchedulerSvc.h:294
IHiveWhiteBoard.h
extends
Base class used to extend a class implementing other interfaces.
Definition: extends.h:10
AvalancheSchedulerSvc::eventFailed
StatusCode eventFailed(EventContext *eventContext)
Method to check if an event failed and take appropriate actions.
Definition: AvalancheSchedulerSvc.cpp:509
AvalancheSchedulerSvc::index2algname
const std::string & index2algname(unsigned int index)
Convert an integer to a name.
Definition: AvalancheSchedulerSvc.cpp:368
AvalancheSchedulerSvc::promoteToExecuted
StatusCode promoteToExecuted(unsigned int iAlgo, int si, IAlgorithm *algo, EventContext *)
The call to this method is triggered only from within the AlgoExecutionTask.
Definition: AvalancheSchedulerSvc.cpp:936
AvalancheSchedulerSvc::enqueueSchedulerActionTask::m_closure
std::function< StatusCode()> m_closure
Definition: AvalancheSchedulerSvc.h:248
std::hex
T hex(T...args)
AvalancheSchedulerSvc::freeSlots
unsigned int freeSlots() override
Get free slots number.
Definition: AvalancheSchedulerSvc.cpp:446
AvalancheSchedulerSvc::m_eventSlots
std::vector< EventSlot > m_eventSlots
Vector of events slots.
Definition: AvalancheSchedulerSvc.h:190
AvalancheSchedulerSvc::m_first
bool m_first
Definition: AvalancheSchedulerSvc.h:268
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:992
AvalancheSchedulerSvc::SchedulerState::operator==
bool operator==(Algorithm *a) const
Definition: AvalancheSchedulerSvc.h:289
AvalancheSchedulerSvc::deactivate
StatusCode deactivate()
Deactivate scheduler.
Definition: AvalancheSchedulerSvc.cpp:348
AvalancheSchedulerSvc::dumpState
void dumpState() override
Definition: AvalancheSchedulerSvc.cpp:1081
AlgsExecutionStates.h
AvalancheSchedulerSvc::FAILURE
Definition: AvalancheSchedulerSvc.h:131
AvalancheSchedulerSvc::updateStates
StatusCode updateStates(int si=-1, const std::string &algo_name=std::string())
Loop on algorithm in the slots and promote them to successive states (-1 means all slots...
Definition: AvalancheSchedulerSvc.cpp:555
AlgsExecutionStates::State
State
Execution states of the algorithms.
Definition: AlgsExecutionStates.h:32
IAccelerator.h
std::ostream
STL class.
std::thread
STL class.
AvalancheSchedulerSvc::m_algExecStateSvc
SmartIF< IAlgExecStateSvc > m_algExecStateSvc
Algorithm execution state manager.
Definition: AvalancheSchedulerSvc.h:202
AvalancheSchedulerSvc::m_maxIOBoundAlgosInFlight
Gaudi::Property< unsigned int > m_maxIOBoundAlgosInFlight
Definition: AvalancheSchedulerSvc.h:143
AvalancheSchedulerSvc::SchedulerState::thread
pthread_t thread() const
Definition: AvalancheSchedulerSvc.h:278
EventSlot.h
AvalancheSchedulerSvc::m_actionsQueue
tbb::concurrent_bounded_queue< action > m_actionsQueue
Queue where closures are stored and picked for execution.
Definition: AvalancheSchedulerSvc.h:243
AtlasMCRecoScenario.scheduler
scheduler
Definition: AtlasMCRecoScenario.py:23
AvalancheSchedulerSvc::SchedulerState
Definition: AvalancheSchedulerSvc.h:270
std::unordered_map< std::string, unsigned int >
AvalancheSchedulerSvc::m_thread
std::thread m_thread
The thread in which the activate function runs.
Definition: AvalancheSchedulerSvc.h:169
AvalancheSchedulerSvc::enqueueSchedulerActionTask
Definition: AvalancheSchedulerSvc.h:246
AvalancheSchedulerSvc::m_drain
StatusCode m_drain()
Drain the actions present in the queue.
Definition: AvalancheSchedulerSvc.cpp:454