| The Gaudi Framework v29r1 (5cd95fa2)

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

- File List
- File Members

ForwardSchedulerSvc.h
Go to the documentation of this file.
1 #ifndef GAUDIHIVE_FORWARDSCHEDULERSVC_H
2 #define GAUDIHIVE_FORWARDSCHEDULERSVC_H
3
4 // Local includes
5 #include "AlgsExecutionStates.h"
6 #include "DataFlowManager.h"
7 #include "EventSlot.h"
8 #include "ExecutionFlowManager.h"
9
10 // Framework include files
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
83 class ForwardSchedulerSvc : public extends<Service, IScheduler>
84 {
85 public:
87 using extends::extends;
88
90 ~ForwardSchedulerSvc() override = default;
91
93 StatusCode initialize() override;
94
96 StatusCode finalize() override;
97
99 StatusCode pushNewEvent( EventContext* eventContext ) override;
100
101 // Make multiple events available to the scheduler
102 StatusCode pushNewEvents( std::vector<EventContext*>& eventContexts ) override;
103
105 StatusCode popFinishedEvent( EventContext*& eventContext ) override;
106
108 StatusCode tryPopFinishedEvent( EventContext*& eventContext ) override;
109
111 unsigned int freeSlots() override;
112
113 private:
114 enum ActivationState { INACTIVE = 0, ACTIVE = 1, FAILURE = 2 };
115
116 Gaudi::Property<int> m_maxEventsInFlight{this, "MaxEventsInFlight", 0,
117 "Maximum number of event processed simultaneously"};
118 Gaudi::Property<int> m_threadPoolSize{
119 this, "ThreadPoolSize", -1,
120 "Size of the threadpool initialised by TBB; a value of -1 gives TBB the freedom to choose"};
121 Gaudi::Property<std::string> m_whiteboardSvcName{this, "WhiteboardSvc", "EventDataSvc", "The whiteboard name"};
122 Gaudi::Property<unsigned int> m_maxAlgosInFlight{this, "MaxAlgosInFlight", 1,
123 "[[deprecated]] Taken from the whiteboard"};
124 Gaudi::Property<std::vector<std::vector<std::string>>> m_algosDependencies{
125 this, "AlgosDependencies", {}, "[[deprecated]]"};
126 Gaudi::Property<bool> m_checkDeps{this, "CheckDependencies", false, "Runtime check of Algorithm Data Dependencies"};
127 Gaudi::Property<std::string> m_useDataLoader{this, "DataLoaderAlg", "",
128 "Attribute unmet input dependencies to this DataLoader Algorithm"};
129
130 Gaudi::Property<bool> m_showDataDeps{this, "ShowDataDependencies", true,
131 "Show the INPUT and OUTPUT data dependencies of Algorithms"};
132 Gaudi::Property<bool> m_showDataFlow{this, "ShowDataFlow", false,
133 "Show the configuration of DataFlow between Algorithms"};
134 Gaudi::Property<bool> m_showControlFlow{this, "ShowControlFlow", false,
135 "Show the configuration of all Algorithms and Sequences"};
136
137 // Utils and shortcuts ----------------------------------------------------
138
140 void activate();
141
143 StatusCode deactivate();
144
146 std::atomic<ActivationState> m_isActive{INACTIVE};
147
149 std::thread m_thread;
150
152 inline unsigned int algname2index( const std::string& algoname );
153
155 std::unordered_map<std::string, unsigned int> m_algname_index_map;
156
158 inline const std::string& index2algname( unsigned int index );
159
161 std::vector<std::string> m_algname_vect;
162
164 SmartIF<IHiveWhiteBoard> m_whiteboard;
165
167 std::vector<EventSlot> m_eventSlots;
168
170 std::atomic_int m_freeSlots;
171
173 tbb::concurrent_bounded_queue<EventContext*> m_finishedEvents;
174
176 StatusCode eventFailed( EventContext* eventContext );
177
179 SmartIF<IAlgExecStateSvc> m_algExecStateSvc;
180
181 // States management ------------------------------------------------------
182
184 unsigned int m_algosInFlight = 0;
185
188 StatusCode updateStates( int si = -1 );
189
191 StatusCode promoteToControlReady( unsigned int iAlgo, int si );
192 StatusCode promoteToDataReady( unsigned int iAlgo, int si );
193 StatusCode promoteToScheduled( unsigned int iAlgo, int si );
194 StatusCode promoteToExecuted( unsigned int iAlgo, int si, IAlgorithm* algo, EventContext* );
195 StatusCode promoteToFinished( unsigned int iAlgo, int si );
196
198 StatusCode isStalled( int si );
199
201 void dumpSchedulerState( int iSlot );
202
204 bool m_updateNeeded = true;
205
206 // Algos Management -------------------------------------------------------
208 SmartIF<IAlgResourcePool> m_algResourcePool;
209
211 StatusCode m_drain();
212
213 // Actions management -----------------------------------------------------
214
216 tbb::concurrent_bounded_queue<action> m_actionsQueue;
217
218 // helper task to enqueue the scheduler's actions (closures)
219 struct enqueueSchedulerActionTask : public tbb::task {
220
221 std::function<StatusCode()> m_closure;
222 SmartIF<ForwardSchedulerSvc> m_scheduler;
223
224 enqueueSchedulerActionTask( ForwardSchedulerSvc* scheduler, std::function<StatusCode()> _closure )
225 : m_closure( _closure ), m_scheduler( scheduler )
226 {
227 }
228
229 tbb::task* execute() override
230 {
231 m_scheduler->m_actionsQueue.push( m_closure );
232 return nullptr;
233 }
234 };
235
236 // ------------------------------------------------------------------------
237
239 concurrency::recursive_CF::ExecutionFlowManager m_efManager;
240
241 // Service for thread pool initialization
242 SmartIF<IThreadPoolSvc> m_threadPoolSvc;
243
244 bool m_first = true;
245
246 class SchedulerState
247 {
248
249 public:
250 SchedulerState( Algorithm* a, EventContext* e, pthread_t t ) : m_a( a ), m_e( *e ), m_t( t ) {}
251
252 Algorithm* alg() const { return m_a; }
253 EventContext ctx() const { return m_e; }
254 pthread_t thread() const { return m_t; }
255
256 friend std::ostream& operator<<( std::ostream& os, const SchedulerState& ss )
257 {
258 os << ss.ctx() << " a: " << ss.alg()->name() << " [" << std::hex << ss.alg() << std::dec << "] t: 0x"
259 << std::hex << ss.thread() << std::dec;
260 return os;
261 }
262
263 bool operator==( const SchedulerState& ss ) const { return ( m_a == ss.alg() ); }
264
265 bool operator==( Algorithm* a ) const { return ( m_a == a ); }
266
267 bool operator<( const SchedulerState& rhs ) const { return ( m_a < rhs.alg() ); }
268
269 private:
270 Algorithm* m_a;
271 EventContext m_e;
272 pthread_t m_t;
273 };
274
275 static std::list<SchedulerState> m_sState;
276 static std::mutex m_ssMut;
277
278 public:
279 void addAlg( Algorithm*, EventContext*, pthread_t );
280 bool delAlg( Algorithm* );
281 void dumpState() override;
282
283 private:
284 void dumpState( std::ostringstream& );
285 concurrency::recursive_CF::ControlFlowGraph* m_efg;
286 };
287
288 #endif // GAUDIHIVE_FORWARDSCHEDULERSVC_H
ForwardSchedulerSvc::deactivate
StatusCode deactivate()
Deactivate scheduler.
Definition: ForwardSchedulerSvc.cpp:385
IThreadPoolSvc.h
ForwardSchedulerSvc::SchedulerState::SchedulerState
SchedulerState(Algorithm *a, EventContext *e, pthread_t t)
Definition: ForwardSchedulerSvc.h:250
ForwardSchedulerSvc::m_showControlFlow
Gaudi::Property< bool > m_showControlFlow
Definition: ForwardSchedulerSvc.h:134
ForwardSchedulerSvc::initialize
StatusCode initialize() override
Initialise.
Definition: ForwardSchedulerSvc.cpp:44
SmartIF< IHiveWhiteBoard >
ForwardSchedulerSvc::eventFailed
StatusCode eventFailed(EventContext *eventContext)
Method to check if an event failed and take appropriate actions.
Definition: ForwardSchedulerSvc.cpp:543
ForwardSchedulerSvc::m_algResourcePool
SmartIF< IAlgResourcePool > m_algResourcePool
Cache for the algorithm resource pool.
Definition: ForwardSchedulerSvc.h:208
Algorithm::name
const std::string & name() const override
The identifying name of the algorithm object.
Definition: Algorithm.cpp:731
std::atomic< ActivationState >
IAlgExecStateSvc.h
ExecutionFlowManager.h
std::function< StatusCode()>
Gaudi::Property
Implementation of property with value of concrete type.
Definition: Property.h:319
ForwardSchedulerSvc::m_efManager
concurrency::recursive_CF::ExecutionFlowManager m_efManager
Member to take care of the control flow.
Definition: ForwardSchedulerSvc.h:239
IRunable.h
ForwardSchedulerSvc::isStalled
StatusCode isStalled(int si)
Check if the scheduling is in a stall.
Definition: ForwardSchedulerSvc.cpp:731
ForwardSchedulerSvc::SchedulerState::m_e
EventContext m_e
Definition: ForwardSchedulerSvc.h:271
ForwardSchedulerSvc::updateStates
StatusCode updateStates(int si=-1)
Loop on algorithm in the slots and promote them to successive states (-1 means all slots...
Definition: ForwardSchedulerSvc.cpp:588
ForwardSchedulerSvc::finalize
StatusCode finalize() override
Finalise.
Definition: ForwardSchedulerSvc.cpp:311
IAlgResourcePool.h
ForwardSchedulerSvc::FAILURE
Definition: ForwardSchedulerSvc.h:114
ForwardSchedulerSvc::SchedulerState
Definition: ForwardSchedulerSvc.h:246
ForwardSchedulerSvc::m_threadPoolSvc
SmartIF< IThreadPoolSvc > m_threadPoolSvc
Definition: ForwardSchedulerSvc.h:242
ForwardSchedulerSvc::enqueueSchedulerActionTask
Definition: ForwardSchedulerSvc.h:219
State
AlgsExecutionStates::State State
Definition: ForwardSchedulerSvc.h:30
ForwardSchedulerSvc::m_checkDeps
Gaudi::Property< bool > m_checkDeps
Definition: ForwardSchedulerSvc.h:126
IScheduler.h
ForwardSchedulerSvc
The SchedulerSvc implements the IScheduler interface.
Definition: ForwardSchedulerSvc.h:83
ForwardSchedulerSvc::pushNewEvent
StatusCode pushNewEvent(EventContext *eventContext) override
Make an event available to the scheduler.
Definition: ForwardSchedulerSvc.cpp:424
ForwardSchedulerSvc::m_algosDependencies
Gaudi::Property< std::vector< std::vector< std::string > > > m_algosDependencies
Definition: ForwardSchedulerSvc.h:124
EventContext
This class represents an entry point to all the event specific data.
Definition: EventContext.h:24
ForwardSchedulerSvc::m_whiteboard
SmartIF< IHiveWhiteBoard > m_whiteboard
A shortcut to the whiteboard.
Definition: ForwardSchedulerSvc.h:164
std::mutex
ForwardSchedulerSvc::m_drain
StatusCode m_drain()
Drain the actions present in the queue.
Definition: ForwardSchedulerSvc.cpp:487
ForwardSchedulerSvc::ACTIVE
Definition: ForwardSchedulerSvc.h:114
ForwardSchedulerSvc::ActivationState
ActivationState
Definition: ForwardSchedulerSvc.h:114
ForwardSchedulerSvc::enqueueSchedulerActionTask::m_closure
std::function< StatusCode()> m_closure
Definition: ForwardSchedulerSvc.h:221
ForwardSchedulerSvc::m_maxAlgosInFlight
Gaudi::Property< unsigned int > m_maxAlgosInFlight
Definition: ForwardSchedulerSvc.h:122
std::string
STL class.
Service.h
ForwardSchedulerSvc::enqueueSchedulerActionTask::enqueueSchedulerActionTask
enqueueSchedulerActionTask(ForwardSchedulerSvc *scheduler, std::function< StatusCode()> _closure)
Definition: ForwardSchedulerSvc.h:224
ForwardSchedulerSvc::SchedulerState::thread
pthread_t thread() const
Definition: ForwardSchedulerSvc.h:254
ForwardSchedulerSvc::enqueueSchedulerActionTask::m_scheduler
SmartIF< ForwardSchedulerSvc > m_scheduler
Definition: ForwardSchedulerSvc.h:222
ForwardSchedulerSvc::INACTIVE
Definition: ForwardSchedulerSvc.h:114
ForwardSchedulerSvc::m_freeSlots
std::atomic_int m_freeSlots
Atomic to account for asyncronous updates by the scheduler wrt the rest.
Definition: ForwardSchedulerSvc.h:170
ForwardSchedulerSvc::addAlg
void addAlg(Algorithm *, EventContext *, pthread_t)
Definition: ForwardSchedulerSvc.cpp:989
ForwardSchedulerSvc::m_showDataFlow
Gaudi::Property< bool > m_showDataFlow
Definition: ForwardSchedulerSvc.h:132
concurrency::recursive_CF::ControlFlowGraph
Definition: ControlFlowGraph.h:144
ForwardSchedulerSvc::m_algosInFlight
unsigned int m_algosInFlight
Number of algoritms presently in flight.
Definition: ForwardSchedulerSvc.h:184
ForwardSchedulerSvc::m_finishedEvents
tbb::concurrent_bounded_queue< EventContext * > m_finishedEvents
Queue of finished events.
Definition: ForwardSchedulerSvc.h:173
ForwardSchedulerSvc::m_thread
std::thread m_thread
The thread in which the activate function runs.
Definition: ForwardSchedulerSvc.h:149
ForwardSchedulerSvc::SchedulerState::operator<
bool operator<(const SchedulerState &rhs) const
Definition: ForwardSchedulerSvc.h:267
ForwardSchedulerSvc::m_algname_vect
std::vector< std::string > m_algname_vect
Vector to bookkeep the information necessary to the index2name conversion.
Definition: ForwardSchedulerSvc.h:161
ForwardSchedulerSvc::m_sState
static std::list< SchedulerState > m_sState
Definition: ForwardSchedulerSvc.h:275
ForwardSchedulerSvc::m_useDataLoader
Gaudi::Property< std::string > m_useDataLoader
Definition: ForwardSchedulerSvc.h:127
ForwardSchedulerSvc::m_eventSlots
std::vector< EventSlot > m_eventSlots
Vector of events slots.
Definition: ForwardSchedulerSvc.h:167
ForwardSchedulerSvc::promoteToScheduled
StatusCode promoteToScheduled(unsigned int iAlgo, int si)
Definition: ForwardSchedulerSvc.cpp:860
ForwardSchedulerSvc::promoteToFinished
StatusCode promoteToFinished(unsigned int iAlgo, int si)
StatusCode
This class is used for returning status codes from appropriate routines.
Definition: StatusCode.h:28
std::ostringstream
STL class.
ForwardSchedulerSvc::SchedulerState::operator==
bool operator==(Algorithm *a) const
Definition: ForwardSchedulerSvc.h:265
ForwardSchedulerSvc::m_whiteboardSvcName
Gaudi::Property< std::string > m_whiteboardSvcName
Definition: ForwardSchedulerSvc.h:121
ForwardSchedulerSvc::algname2index
unsigned int algname2index(const std::string &algoname)
Convert a name to an integer.
Definition: ForwardSchedulerSvc.cpp:410
ForwardSchedulerSvc::enqueueSchedulerActionTask::execute
tbb::task * execute() override
Definition: ForwardSchedulerSvc.h:229
ForwardSchedulerSvc::promoteToExecuted
StatusCode promoteToExecuted(unsigned int iAlgo, int si, IAlgorithm *algo, EventContext *)
Definition: ForwardSchedulerSvc.cpp:921
ForwardSchedulerSvc::SchedulerState::operator==
bool operator==(const SchedulerState &ss) const
Definition: ForwardSchedulerSvc.h:263
ForwardSchedulerSvc::~ForwardSchedulerSvc
~ForwardSchedulerSvc() override=default
Destructor.
ForwardSchedulerSvc::freeSlots
unsigned int freeSlots() override
Get free slots number.
Definition: ForwardSchedulerSvc.cpp:481
ForwardSchedulerSvc::pushNewEvents
StatusCode pushNewEvents(std::vector< EventContext * > &eventContexts) override
Definition: ForwardSchedulerSvc.cpp:470
ForwardSchedulerSvc::m_maxEventsInFlight
Gaudi::Property< int > m_maxEventsInFlight
Definition: ForwardSchedulerSvc.h:116
std::list
STL class.
IAlgorithm
The IAlgorithm is the interface implemented by the Algorithm base class.
Definition: IAlgorithm.h:28
ForwardSchedulerSvc::m_actionsQueue
tbb::concurrent_bounded_queue< action > m_actionsQueue
Queue where closures are stored and picked for execution.
Definition: ForwardSchedulerSvc.h:216
ForwardSchedulerSvc::m_updateNeeded
bool m_updateNeeded
Keep track of update actions scheduled.
Definition: ForwardSchedulerSvc.h:204
ForwardSchedulerSvc::m_efg
concurrency::recursive_CF::ControlFlowGraph * m_efg
Definition: ForwardSchedulerSvc.h:285
ForwardSchedulerSvc::tryPopFinishedEvent
StatusCode tryPopFinishedEvent(EventContext *&eventContext) override
Try to fetch an event from the scheduler.
Definition: ForwardSchedulerSvc.cpp:525
Algorithm
Base class from which all concrete algorithm classes should be derived.
Definition: Algorithm.h:79
ForwardSchedulerSvc::SchedulerState::m_t
pthread_t m_t
Definition: ForwardSchedulerSvc.h:272
std::vector
STL class.
ForwardSchedulerSvc::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: ForwardSchedulerSvc.h:155
ForwardSchedulerSvc::activate
void activate()
Activate scheduler.
Definition: ForwardSchedulerSvc.cpp:342
AtlasMCRecoFullPrecedenceDump.scheduler
scheduler
Definition: AtlasMCRecoFullPrecedenceDump.py:37
IHiveWhiteBoard.h
extends
Base class used to extend a class implementing other interfaces.
Definition: extends.h:10
ForwardSchedulerSvc::m_algExecStateSvc
SmartIF< IAlgExecStateSvc > m_algExecStateSvc
Algorithm execution state manager.
Definition: ForwardSchedulerSvc.h:179
ForwardSchedulerSvc::promoteToControlReady
StatusCode promoteToControlReady(unsigned int iAlgo, int si)
Algorithm promotion: Accepted by the control flow.
Definition: ForwardSchedulerSvc.cpp:828
ForwardSchedulerSvc::popFinishedEvent
StatusCode popFinishedEvent(EventContext *&eventContext) override
Blocks until an event is availble.
Definition: ForwardSchedulerSvc.cpp:503
ForwardSchedulerSvc::dumpSchedulerState
void dumpSchedulerState(int iSlot)
Dump the state of the scheduler.
Definition: ForwardSchedulerSvc.cpp:756
ForwardSchedulerSvc::SchedulerState::m_a
Algorithm * m_a
Definition: ForwardSchedulerSvc.h:270
std::hex
T hex(T...args)
action
std::function< StatusCode()> action
Definition: ForwardSchedulerSvc.h:31
ForwardSchedulerSvc::m_ssMut
static std::mutex m_ssMut
Definition: ForwardSchedulerSvc.h:276
ForwardSchedulerSvc::index2algname
const std::string & index2algname(unsigned int index)
Convert an integer to a name.
Definition: ForwardSchedulerSvc.cpp:406
ForwardSchedulerSvc::delAlg
bool delAlg(Algorithm *)
Definition: ForwardSchedulerSvc.cpp:997
ForwardSchedulerSvc::SchedulerState::ctx
EventContext ctx() const
Definition: ForwardSchedulerSvc.h:253
ForwardSchedulerSvc::SchedulerState::operator<<
friend std::ostream & operator<<(std::ostream &os, const SchedulerState &ss)
Definition: ForwardSchedulerSvc.h:256
ForwardSchedulerSvc::m_showDataDeps
Gaudi::Property< bool > m_showDataDeps
Definition: ForwardSchedulerSvc.h:130
ForwardSchedulerSvc::m_isActive
std::atomic< ActivationState > m_isActive
Flag to track if the scheduler is active or not.
Definition: ForwardSchedulerSvc.h:146
AlgsExecutionStates.h
ForwardSchedulerSvc::SchedulerState::alg
Algorithm * alg() const
Definition: ForwardSchedulerSvc.h:252
concurrency::recursive_CF::ExecutionFlowManager
Manage control flow part of the execution flow.
Definition: ExecutionFlowManager.h:24
ForwardSchedulerSvc::m_first
bool m_first
Definition: ForwardSchedulerSvc.h:244
AlgsExecutionStates::State
State
Execution states of the algorithms.
Definition: AlgsExecutionStates.h:32
ForwardSchedulerSvc::m_threadPoolSize
Gaudi::Property< int > m_threadPoolSize
Definition: ForwardSchedulerSvc.h:118
DataFlowManager.h
std::ostream
STL class.
ForwardSchedulerSvc::dumpState
void dumpState() override
Definition: ForwardSchedulerSvc.cpp:1025
std::thread
STL class.
EventSlot.h
ForwardSchedulerSvc::promoteToDataReady
StatusCode promoteToDataReady(unsigned int iAlgo, int si)
Definition: ForwardSchedulerSvc.cpp:842
std::unordered_map< std::string, unsigned int >