| The Gaudi Framework v30r0 (c919700c)

- 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 "PrecedenceSvc.h"
8
9 // Framework include files
10 #include "GaudiKernel/Algorithm.h"
11 #include "GaudiKernel/IAccelerator.h"
12 #include "GaudiKernel/IAlgExecStateSvc.h"
13 #include "GaudiKernel/IAlgResourcePool.h"
14 #include "GaudiKernel/ICondSvc.h"
15 #include "GaudiKernel/IHiveWhiteBoard.h"
16 #include "GaudiKernel/IRunable.h"
17 #include "GaudiKernel/IScheduler.h"
18 #include "GaudiKernel/IThreadPoolSvc.h"
19 #include "GaudiKernel/Service.h"
20
21 // C++ include files
22 #include <functional>
23 #include <string>
24 #include <thread>
25 #include <unordered_map>
26 #include <vector>
27
28 // External libs
29 #include "tbb/concurrent_queue.h"
30 #include "tbb/task.h"
31
32 typedef AlgsExecutionStates::State State;
33 typedef std::function<StatusCode()> action;
34
35 //---------------------------------------------------------------------------
36
102 class AvalancheSchedulerSvc : public extends<Service, IScheduler>
103 {
104 public:
106 using extends::extends;
107
109 ~AvalancheSchedulerSvc() override = default;
110
112 StatusCode initialize() override;
113
115 StatusCode finalize() override;
116
118 StatusCode pushNewEvent( EventContext* eventContext ) override;
119
120 // Make multiple events available to the scheduler
121 StatusCode pushNewEvents( std::vector<EventContext*>& eventContexts ) override;
122
124 StatusCode popFinishedEvent( EventContext*& eventContext ) override;
125
127 StatusCode tryPopFinishedEvent( EventContext*& eventContext ) override;
128
130 unsigned int freeSlots() override;
131
132 private:
133 enum ActivationState { INACTIVE = 0, ACTIVE = 1, FAILURE = 2 };
134
135 Gaudi::Property<int> m_threadPoolSize{
136 this, "ThreadPoolSize", -1,
137 "Size of the threadpool initialised by TBB; a value of -1 gives TBB the freedom to choose"};
138 Gaudi::Property<std::string> m_whiteboardSvcName{this, "WhiteboardSvc", "EventDataSvc", "The whiteboard name"};
139 Gaudi::Property<std::string> m_IOBoundAlgSchedulerSvcName{this, "IOBoundAlgSchedulerSvc", "IOBoundAlgSchedulerSvc"};
140 Gaudi::Property<unsigned int> m_maxIOBoundAlgosInFlight{this, "MaxIOBoundAlgosInFlight", 0,
141 "Maximum number of simultaneous I/O-bound algorithms"};
142 Gaudi::Property<bool> m_simulateExecution{
143 this, "SimulateExecution", false,
144 "Flag to perform single-pass simulation of execution flow before the actual execution"};
145 Gaudi::Property<std::string> m_optimizationMode{this, "Optimizer", "",
146 "The following modes are currently available: PCE, COD, DRE, E"};
147 Gaudi::Property<bool> m_dumpIntraEventDynamics{this, "DumpIntraEventDynamics", false,
148 "Dump intra-event concurrency dynamics to csv file"};
149 Gaudi::Property<bool> m_useIOBoundAlgScheduler{this, "PreemptiveIOBoundTasks", false,
150 "Turn on preemptive way of scheduling of I/O-bound algorithms"};
151
152 Gaudi::Property<bool> m_checkDeps{this, "CheckDependencies", false, "Runtime check of Algorithm Data Dependencies"};
153
154 Gaudi::Property<std::string> m_useDataLoader{this, "DataLoaderAlg", "",
155 "Attribute unmet input dependencies to this DataLoader Algorithm"};
156
157 Gaudi::Property<bool> m_enableCondSvc{this, "EnableConditions", false, "Enable ConditionsSvc"};
158
159 Gaudi::Property<bool> m_showDataDeps{this, "ShowDataDependencies", true,
160 "Show the INPUT and OUTPUT data dependencies of Algorithms"};
161
162 Gaudi::Property<bool> m_showDataFlow{this, "ShowDataFlow", false,
163 "Show the configuration of DataFlow between Algorithms"};
164
165 Gaudi::Property<bool> m_showControlFlow{this, "ShowControlFlow", false,
166 "Show the configuration of all Algorithms and Sequences"};
167
168 // Utils and shortcuts ----------------------------------------------------
169
171 void activate();
172
174 StatusCode deactivate();
175
177 std::atomic<ActivationState> m_isActive{INACTIVE};
178
180 std::thread m_thread;
181
183 inline unsigned int algname2index( const std::string& algoname );
184
186 std::unordered_map<std::string, unsigned int> m_algname_index_map;
187
189 inline const std::string& index2algname( unsigned int index );
190
192 std::vector<std::string> m_algname_vect;
193
195 SmartIF<IPrecedenceSvc> m_precSvc;
196
198 SmartIF<IHiveWhiteBoard> m_whiteboard;
199
201 SmartIF<IAccelerator> m_IOBoundAlgScheduler;
202
204 std::vector<EventSlot> m_eventSlots;
205
207 std::atomic_int m_freeSlots;
208
210 tbb::concurrent_bounded_queue<EventContext*> m_finishedEvents;
211
213 StatusCode eventFailed( EventContext* eventContext );
214
216 SmartIF<IAlgExecStateSvc> m_algExecStateSvc;
217
219 SmartIF<ICondSvc> m_condSvc;
220
221 // States management ------------------------------------------------------
222
224 unsigned int m_algosInFlight = 0;
225
227 unsigned int m_IOBoundAlgosInFlight = 0;
228
231 StatusCode updateStates( int si = -1, int algo_index = -1 );
232
234 StatusCode promoteToScheduled( unsigned int iAlgo, int si );
235 StatusCode promoteToAsyncScheduled( unsigned int iAlgo, int si ); // tests of an asynchronous scheduler
236 StatusCode promoteToExecuted( unsigned int iAlgo, int si, IAlgorithm* algo, EventContext* );
237 StatusCode promoteToAsyncExecuted( unsigned int iAlgo, int si, IAlgorithm* algo,
238 EventContext* ); // tests of an asynchronous scheduler
239 StatusCode promoteToFinished( unsigned int iAlgo, int si );
240
242 StatusCode isStalled( int si );
243
245 void dumpSchedulerState( int iSlot );
246
247 // Algos Management -------------------------------------------------------
249 SmartIF<IAlgResourcePool> m_algResourcePool;
250
252 StatusCode m_drain();
253
254 // Actions management -----------------------------------------------------
255
257 tbb::concurrent_bounded_queue<action> m_actionsQueue;
258
259 // ------------------------------------------------------------------------
260
261 // Service for thread pool initialization
262 SmartIF<IThreadPoolSvc> m_threadPoolSvc;
263 size_t m_maxEventsInFlight{0};
264 size_t m_maxAlgosInFlight{1};
265 bool m_first = true;
266
267 class SchedulerState
268 {
269
270 public:
271 SchedulerState( Algorithm* a, EventContext* e, pthread_t t ) : m_a( a ), m_e( *e ), m_t( t ) {}
272
273 Algorithm* alg() const { return m_a; }
274 EventContext ctx() const { return m_e; }
275 pthread_t thread() const { return m_t; }
276
277 friend std::ostream& operator<<( std::ostream& os, const SchedulerState& ss )
278 {
279 os << ss.ctx() << " a: " << ss.alg()->name() << " [" << std::hex << ss.alg() << std::dec << "] t: 0x"
280 << std::hex << ss.thread() << std::dec;
281 return os;
282 }
283
284 bool operator==( const SchedulerState& ss ) const { return ( m_a == ss.alg() ); }
285
286 bool operator==( Algorithm* a ) const { return ( m_a == a ); }
287
288 bool operator<( const SchedulerState& rhs ) const { return ( m_a < rhs.alg() ); }
289
290 private:
291 Algorithm* m_a;
292 EventContext m_e;
293 pthread_t m_t;
294 };
295
296 static std::list<SchedulerState> m_sState;
297 static std::mutex m_ssMut;
298
299 public:
300 void addAlg( Algorithm*, EventContext*, pthread_t );
301 bool delAlg( Algorithm* );
302 void dumpState() override;
303
304 private:
305 void dumpState( std::ostringstream& );
306 };
307
308 #endif // GAUDIHIVE_AVALANCHESCHEDULERSVC_H
AvalancheSchedulerSvc::m_showDataFlow
Gaudi::Property< bool > m_showDataFlow
Definition: AvalancheSchedulerSvc.h:162
IThreadPoolSvc.h
AvalancheSchedulerSvc::SchedulerState::m_e
EventContext m_e
Definition: AvalancheSchedulerSvc.h:292
AvalancheSchedulerSvc::tryPopFinishedEvent
StatusCode tryPopFinishedEvent(EventContext *&eventContext) override
Try to fetch an event from the scheduler.
Definition: AvalancheSchedulerSvc.cpp:562
Algorithm.h
SmartIF< IPrecedenceSvc >
AvalancheSchedulerSvc::INACTIVE
Definition: AvalancheSchedulerSvc.h:133
AvalancheSchedulerSvc::m_whiteboardSvcName
Gaudi::Property< std::string > m_whiteboardSvcName
Definition: AvalancheSchedulerSvc.h:138
AvalancheSchedulerSvc::m_IOBoundAlgosInFlight
unsigned int m_IOBoundAlgosInFlight
Number of algoritms presently in flight.
Definition: AvalancheSchedulerSvc.h:227
AvalancheSchedulerSvc::m_precSvc
SmartIF< IPrecedenceSvc > m_precSvc
A shortcut to the Precedence Service.
Definition: AvalancheSchedulerSvc.h:195
Algorithm::name
const std::string & name() const override
The identifying name of the algorithm object.
Definition: Algorithm.cpp:737
std::atomic< ActivationState >
IAlgExecStateSvc.h
std::function< StatusCode()>
Gaudi::Property
Implementation of property with value of concrete type.
Definition: Property.h:319
AvalancheSchedulerSvc::m_dumpIntraEventDynamics
Gaudi::Property< bool > m_dumpIntraEventDynamics
Definition: AvalancheSchedulerSvc.h:147
AvalancheSchedulerSvc::m_showDataDeps
Gaudi::Property< bool > m_showDataDeps
Definition: AvalancheSchedulerSvc.h:159
AvalancheSchedulerSvc::SchedulerState::m_t
pthread_t m_t
Definition: AvalancheSchedulerSvc.h:293
AvalancheSchedulerSvc::SchedulerState::operator==
bool operator==(const SchedulerState &ss) const
Definition: AvalancheSchedulerSvc.h:284
AvalancheSchedulerSvc::initialize
StatusCode initialize() override
Initialise.
Definition: AvalancheSchedulerSvc.cpp:61
AvalancheSchedulerSvc::dumpSchedulerState
void dumpSchedulerState(int iSlot)
Dump the state of the scheduler.
Definition: AvalancheSchedulerSvc.cpp:796
AvalancheSchedulerSvc::promoteToScheduled
StatusCode promoteToScheduled(unsigned int iAlgo, int si)
Algorithm promotion.
Definition: AvalancheSchedulerSvc.cpp:833
IRunable.h
AvalancheSchedulerSvc::isStalled
StatusCode isStalled(int si)
Check if the scheduling is in a stall.
Definition: AvalancheSchedulerSvc.cpp:771
AvalancheSchedulerSvc::delAlg
bool delAlg(Algorithm *)
Definition: AvalancheSchedulerSvc.cpp:1040
IAlgResourcePool.h
AvalancheSchedulerSvc::ACTIVE
Definition: AvalancheSchedulerSvc.h:133
AvalancheSchedulerSvc::activate
void activate()
Activate scheduler.
Definition: AvalancheSchedulerSvc.cpp:363
AvalancheSchedulerSvc::m_useDataLoader
Gaudi::Property< std::string > m_useDataLoader
Definition: AvalancheSchedulerSvc.h:154
AvalancheSchedulerSvc::m_optimizationMode
Gaudi::Property< std::string > m_optimizationMode
Definition: AvalancheSchedulerSvc.h:145
IScheduler.h
AvalancheSchedulerSvc::SchedulerState::operator<<
friend std::ostream & operator<<(std::ostream &os, const SchedulerState &ss)
Definition: AvalancheSchedulerSvc.h:277
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:24
AvalancheSchedulerSvc::algname2index
unsigned int algname2index(const std::string &algoname)
Convert a name to an integer.
Definition: AvalancheSchedulerSvc.cpp:431
std::mutex
State
AlgsExecutionStates::State State
Definition: AvalancheSchedulerSvc.h:32
AvalancheSchedulerSvc::addAlg
void addAlg(Algorithm *, EventContext *, pthread_t)
Definition: AvalancheSchedulerSvc.cpp:1032
AvalancheSchedulerSvc::m_isActive
std::atomic< ActivationState > m_isActive
Flag to track if the scheduler is active or not.
Definition: AvalancheSchedulerSvc.h:177
AvalancheSchedulerSvc::m_algname_index_map
std::unordered_map< std::string, unsigned int > m_algname_index_map
Map to bookkeep the information necessary to the name2index conversion.
Definition: AvalancheSchedulerSvc.h:186
AvalancheSchedulerSvc::m_checkDeps
Gaudi::Property< bool > m_checkDeps
Definition: AvalancheSchedulerSvc.h:152
std::string
STL class.
AvalancheSchedulerSvc::m_useIOBoundAlgScheduler
Gaudi::Property< bool > m_useIOBoundAlgScheduler
Definition: AvalancheSchedulerSvc.h:149
AvalancheSchedulerSvc::SchedulerState::operator<
bool operator<(const SchedulerState &rhs) const
Definition: AvalancheSchedulerSvc.h:288
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:207
AvalancheSchedulerSvc::SchedulerState::alg
Algorithm * alg() const
Definition: AvalancheSchedulerSvc.h:273
AvalancheSchedulerSvc::pushNewEvents
StatusCode pushNewEvents(std::vector< EventContext * > &eventContexts) override
Definition: AvalancheSchedulerSvc.cpp:506
AvalancheSchedulerSvc::m_algosInFlight
unsigned int m_algosInFlight
Number of algoritms presently in flight.
Definition: AvalancheSchedulerSvc.h:224
AvalancheSchedulerSvc::m_sState
static std::list< SchedulerState > m_sState
Definition: AvalancheSchedulerSvc.h:296
AvalancheSchedulerSvc::m_algResourcePool
SmartIF< IAlgResourcePool > m_algResourcePool
Cache for the algorithm resource pool.
Definition: AvalancheSchedulerSvc.h:249
AvalancheSchedulerSvc::popFinishedEvent
StatusCode popFinishedEvent(EventContext *&eventContext) override
Blocks until an event is availble.
Definition: AvalancheSchedulerSvc.cpp:540
AvalancheSchedulerSvc::ActivationState
ActivationState
Definition: AvalancheSchedulerSvc.h:133
AvalancheSchedulerSvc::m_showControlFlow
Gaudi::Property< bool > m_showControlFlow
Definition: AvalancheSchedulerSvc.h:165
StatusCode
This class is used for returning status codes from appropriate routines.
Definition: StatusCode.h:26
AvalancheSchedulerSvc::m_whiteboard
SmartIF< IHiveWhiteBoard > m_whiteboard
A shortcut to the whiteboard.
Definition: AvalancheSchedulerSvc.h:198
std::ostringstream
STL class.
AvalancheSchedulerSvc::m_finishedEvents
tbb::concurrent_bounded_queue< EventContext * > m_finishedEvents
Queue of finished events.
Definition: AvalancheSchedulerSvc.h:210
AvalancheSchedulerSvc::m_algname_vect
std::vector< std::string > m_algname_vect
Vector to bookkeep the information necessary to the index2name conversion.
Definition: AvalancheSchedulerSvc.h:192
AvalancheSchedulerSvc::SchedulerState::SchedulerState
SchedulerState(Algorithm *a, EventContext *e, pthread_t t)
Definition: AvalancheSchedulerSvc.h:271
AvalancheSchedulerSvc::m_IOBoundAlgSchedulerSvcName
Gaudi::Property< std::string > m_IOBoundAlgSchedulerSvcName
Definition: AvalancheSchedulerSvc.h:139
AvalancheSchedulerSvc::SchedulerState::ctx
EventContext ctx() const
Definition: AvalancheSchedulerSvc.h:274
action
std::function< StatusCode()> action
Definition: AvalancheSchedulerSvc.h:33
AvalancheSchedulerSvc::finalize
StatusCode finalize() override
Finalise.
Definition: AvalancheSchedulerSvc.cpp:332
AvalancheSchedulerSvc::m_ssMut
static std::mutex m_ssMut
Definition: AvalancheSchedulerSvc.h:297
AvalancheSchedulerSvc::~AvalancheSchedulerSvc
~AvalancheSchedulerSvc() override=default
Destructor.
AvalancheSchedulerSvc::m_threadPoolSize
Gaudi::Property< int > m_threadPoolSize
Definition: AvalancheSchedulerSvc.h:135
AvalancheSchedulerSvc::m_threadPoolSvc
SmartIF< IThreadPoolSvc > m_threadPoolSvc
Definition: AvalancheSchedulerSvc.h:262
std::list
STL class.
AvalancheSchedulerSvc::m_IOBoundAlgScheduler
SmartIF< IAccelerator > m_IOBoundAlgScheduler
A shortcut to IO-bound algorithm scheduler.
Definition: AvalancheSchedulerSvc.h:201
IAlgorithm
The IAlgorithm is the interface implemented by the Algorithm base class.
Definition: IAlgorithm.h:28
AvalancheSchedulerSvc
Introduction
Definition: AvalancheSchedulerSvc.h:102
Algorithm
Base class from which all concrete algorithm classes should be derived.
Definition: Algorithm.h:79
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:445
std::vector
STL class.
AvalancheSchedulerSvc::m_simulateExecution
Gaudi::Property< bool > m_simulateExecution
Definition: AvalancheSchedulerSvc.h:142
AvalancheSchedulerSvc::SchedulerState::m_a
Algorithm * m_a
Definition: AvalancheSchedulerSvc.h:291
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:580
AvalancheSchedulerSvc::m_enableCondSvc
Gaudi::Property< bool > m_enableCondSvc
Definition: AvalancheSchedulerSvc.h:157
AvalancheSchedulerSvc::index2algname
const std::string & index2algname(unsigned int index)
Convert an integer to a name.
Definition: AvalancheSchedulerSvc.cpp:427
PrecedenceSvc.h
AvalancheSchedulerSvc::m_maxAlgosInFlight
size_t m_maxAlgosInFlight
Definition: AvalancheSchedulerSvc.h:264
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:938
std::hex
T hex(T...args)
AvalancheSchedulerSvc::m_condSvc
SmartIF< ICondSvc > m_condSvc
A shortcut to service for Conditions handling.
Definition: AvalancheSchedulerSvc.h:219
AvalancheSchedulerSvc::freeSlots
unsigned int freeSlots() override
Get free slots number.
Definition: AvalancheSchedulerSvc.cpp:517
AvalancheSchedulerSvc::m_eventSlots
std::vector< EventSlot > m_eventSlots
Vector of events slots.
Definition: AvalancheSchedulerSvc.h:204
AvalancheSchedulerSvc::m_first
bool m_first
Definition: AvalancheSchedulerSvc.h:265
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:987
AvalancheSchedulerSvc::SchedulerState::operator==
bool operator==(Algorithm *a) const
Definition: AvalancheSchedulerSvc.h:286
AvalancheSchedulerSvc::deactivate
StatusCode deactivate()
Deactivate scheduler.
Definition: AvalancheSchedulerSvc.cpp:406
AvalancheSchedulerSvc::dumpState
void dumpState() override
Definition: AvalancheSchedulerSvc.cpp:1068
AvalancheSchedulerSvc::updateStates
StatusCode updateStates(int si=-1, int algo_index=-1)
Loop on algorithm in the slots and promote them to successive states (-1 for algo_index means skippin...
Definition: AvalancheSchedulerSvc.cpp:627
AlgsExecutionStates.h
AvalancheSchedulerSvc::FAILURE
Definition: AvalancheSchedulerSvc.h:133
AlgsExecutionStates::State
State
Execution states of the algorithms.
Definition: AlgsExecutionStates.h:32
ICondSvc.h
IAccelerator.h
std::ostream
STL class.
std::thread
STL class.
AvalancheSchedulerSvc::m_algExecStateSvc
SmartIF< IAlgExecStateSvc > m_algExecStateSvc
Algorithm execution state manager.
Definition: AvalancheSchedulerSvc.h:216
AvalancheSchedulerSvc::m_maxIOBoundAlgosInFlight
Gaudi::Property< unsigned int > m_maxIOBoundAlgosInFlight
Definition: AvalancheSchedulerSvc.h:140
AvalancheSchedulerSvc::SchedulerState::thread
pthread_t thread() const
Definition: AvalancheSchedulerSvc.h:275
AvalancheSchedulerSvc::m_maxEventsInFlight
size_t m_maxEventsInFlight
Definition: AvalancheSchedulerSvc.h:263
EventSlot.h
AvalancheSchedulerSvc::m_actionsQueue
tbb::concurrent_bounded_queue< action > m_actionsQueue
Queue where closures are stored and picked for execution.
Definition: AvalancheSchedulerSvc.h:257
AvalancheSchedulerSvc::SchedulerState
Definition: AvalancheSchedulerSvc.h:267
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:180
AvalancheSchedulerSvc::m_drain
StatusCode m_drain()
Drain the actions present in the queue.
Definition: AvalancheSchedulerSvc.cpp:523