| The Gaudi Framework v28r2p1 (f1a77ff4)

- 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{this, "ThreadPoolSize", -1,
119 "Size of the threadpool initialised by TBB; a value of -1 gives TBB the freedom to choose"};
120 Gaudi::Property<std::string> m_whiteboardSvcName{this, "WhiteboardSvc", "EventDataSvc", "The whiteboard name"};
121 Gaudi::Property<unsigned int> m_maxAlgosInFlight{this, "MaxAlgosInFlight", 1,
122 "[[deprecated]] Taken from the whiteboard"};
123 Gaudi::Property<std::vector<std::vector<std::string>>> m_algosDependencies{
124 this, "AlgosDependencies", {}, "[[deprecated]]"};
125 Gaudi::Property<bool> m_checkDeps{this, "CheckDependencies", false,
126 "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 // Utils and shortcuts ----------------------------------------------------
131
133 void activate();
134
136 StatusCode deactivate();
137
139 std::atomic<ActivationState> m_isActive{INACTIVE};
140
142 std::thread m_thread;
143
145 inline unsigned int algname2index( const std::string& algoname );
146
148 std::unordered_map<std::string, unsigned int> m_algname_index_map;
149
151 inline const std::string& index2algname( unsigned int index );
152
154 std::vector<std::string> m_algname_vect;
155
157 SmartIF<IHiveWhiteBoard> m_whiteboard;
158
160 std::vector<EventSlot> m_eventSlots;
161
163 std::atomic_int m_freeSlots;
164
166 tbb::concurrent_bounded_queue<EventContext*> m_finishedEvents;
167
169 StatusCode eventFailed( EventContext* eventContext );
170
172 SmartIF<IAlgExecStateSvc> m_algExecStateSvc;
173
174 // States management ------------------------------------------------------
175
177 unsigned int m_algosInFlight = 0;
178
181 StatusCode updateStates( int si = -1 );
182
184 StatusCode promoteToControlReady( unsigned int iAlgo, int si );
185 StatusCode promoteToDataReady( unsigned int iAlgo, int si );
186 StatusCode promoteToScheduled( unsigned int iAlgo, int si );
187 StatusCode promoteToExecuted( unsigned int iAlgo, int si, IAlgorithm* algo, EventContext* );
188 StatusCode promoteToFinished( unsigned int iAlgo, int si );
189
191 StatusCode isStalled( int si );
192
194 void dumpSchedulerState( int iSlot );
195
197 bool m_updateNeeded = true;
198
199 // Algos Management -------------------------------------------------------
201 SmartIF<IAlgResourcePool> m_algResourcePool;
202
204 StatusCode m_drain();
205
206 // Actions management -----------------------------------------------------
207
209 tbb::concurrent_bounded_queue<action> m_actionsQueue;
210
211 // helper task to enqueue the scheduler's actions (closures)
212 struct enqueueSchedulerActionTask: public tbb::task {
213
214 std::function<StatusCode()> m_closure;
215 SmartIF<ForwardSchedulerSvc> m_scheduler;
216
217 enqueueSchedulerActionTask(ForwardSchedulerSvc* scheduler, std::function<StatusCode()> _closure) :
218 m_closure(_closure), m_scheduler(scheduler) {}
219
220 tbb::task* execute() override {
221 m_scheduler->m_actionsQueue.push(m_closure);
222 return nullptr;
223 }
224 };
225
226 // ------------------------------------------------------------------------
227
229 concurrency::ExecutionFlowManager m_efManager;
230
231 // Service for thread pool initialization
232 SmartIF<IThreadPoolSvc> m_threadPoolSvc;
233
234 bool m_first = true;
235
236 class SchedulerState
237 {
238
239 public:
240 SchedulerState( Algorithm* a, EventContext* e, pthread_t t ) : m_a( a ), m_e( *e ), m_t( t ) {}
241
242 Algorithm* alg() const { return m_a; }
243 EventContext ctx() const { return m_e; }
244 pthread_t thread() const { return m_t; }
245
246 friend std::ostream& operator<<( std::ostream& os, const SchedulerState& ss )
247 {
248 os << ss.ctx() << " a: " << ss.alg()->name() << " [" << std::hex << ss.alg() << std::dec << "] t: 0x"
249 << std::hex << ss.thread() << std::dec;
250 return os;
251 }
252
253 bool operator==( const SchedulerState& ss ) const { return ( m_a == ss.alg() ); }
254
255 bool operator==( Algorithm* a ) const { return ( m_a == a ); }
256
257 bool operator<( const SchedulerState& rhs ) const { return ( m_a < rhs.alg() ); }
258
259 private:
260 Algorithm* m_a;
261 EventContext m_e;
262 pthread_t m_t;
263 };
264
265 static std::list<SchedulerState> m_sState;
266 static std::mutex m_ssMut;
267
268 public:
269 void addAlg( Algorithm*, EventContext*, pthread_t );
270 bool delAlg( Algorithm* );
271 void dumpState() override;
272
273 private:
274 void dumpState( std::ostringstream& );
275 };
276
277 #endif // GAUDIHIVE_FORWARDSCHEDULERSVC_H
ForwardSchedulerSvc::deactivate
StatusCode deactivate()
Deactivate scheduler.
Definition: ForwardSchedulerSvc.cpp:373
IThreadPoolSvc.h
ForwardSchedulerSvc::SchedulerState::SchedulerState
SchedulerState(Algorithm *a, EventContext *e, pthread_t t)
Definition: ForwardSchedulerSvc.h:240
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:529
ForwardSchedulerSvc::m_algResourcePool
SmartIF< IAlgResourcePool > m_algResourcePool
Cache for the algorithm resource pool.
Definition: ForwardSchedulerSvc.h:201
Algorithm::name
const std::string & name() const override
The identifying name of the algorithm object.
Definition: Algorithm.cpp:715
std::atomic< ActivationState >
IAlgExecStateSvc.h
ExecutionFlowManager.h
std::function< StatusCode()>
Gaudi::Property
Implementation of property with value of concrete type.
Definition: Property.h:314
IRunable.h
ForwardSchedulerSvc::isStalled
StatusCode isStalled(int si)
Check if the scheduling is in a stall.
Definition: ForwardSchedulerSvc.cpp:717
ForwardSchedulerSvc::SchedulerState::m_e
EventContext m_e
Definition: ForwardSchedulerSvc.h:261
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:575
ForwardSchedulerSvc::finalize
StatusCode finalize() override
Finalise.
Definition: ForwardSchedulerSvc.cpp:298
IAlgResourcePool.h
ForwardSchedulerSvc::FAILURE
Definition: ForwardSchedulerSvc.h:114
ForwardSchedulerSvc::SchedulerState
Definition: ForwardSchedulerSvc.h:236
ForwardSchedulerSvc::m_threadPoolSvc
SmartIF< IThreadPoolSvc > m_threadPoolSvc
Definition: ForwardSchedulerSvc.h:232
ForwardSchedulerSvc::enqueueSchedulerActionTask
Definition: ForwardSchedulerSvc.h:212
State
AlgsExecutionStates::State State
Definition: ForwardSchedulerSvc.h:30
ForwardSchedulerSvc::m_checkDeps
Gaudi::Property< bool > m_checkDeps
Definition: ForwardSchedulerSvc.h:125
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:412
ForwardSchedulerSvc::m_algosDependencies
Gaudi::Property< std::vector< std::vector< std::string > > > m_algosDependencies
Definition: ForwardSchedulerSvc.h:123
EventContext
This class represents an entry point to all the event specific data.
Definition: EventContext.h:25
ForwardSchedulerSvc::m_whiteboard
SmartIF< IHiveWhiteBoard > m_whiteboard
A shortcut to the whiteboard.
Definition: ForwardSchedulerSvc.h:157
std::mutex
ForwardSchedulerSvc::m_drain
StatusCode m_drain()
Drain the actions present in the queue.
Definition: ForwardSchedulerSvc.cpp:475
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:214
ForwardSchedulerSvc::m_maxAlgosInFlight
Gaudi::Property< unsigned int > m_maxAlgosInFlight
Definition: ForwardSchedulerSvc.h:121
std::string
STL class.
Service.h
ForwardSchedulerSvc::enqueueSchedulerActionTask::enqueueSchedulerActionTask
enqueueSchedulerActionTask(ForwardSchedulerSvc *scheduler, std::function< StatusCode()> _closure)
Definition: ForwardSchedulerSvc.h:217
ForwardSchedulerSvc::SchedulerState::thread
pthread_t thread() const
Definition: ForwardSchedulerSvc.h:244
ForwardSchedulerSvc::enqueueSchedulerActionTask::m_scheduler
SmartIF< ForwardSchedulerSvc > m_scheduler
Definition: ForwardSchedulerSvc.h:215
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:163
ForwardSchedulerSvc::addAlg
void addAlg(Algorithm *, EventContext *, pthread_t)
Definition: ForwardSchedulerSvc.cpp:977
ForwardSchedulerSvc::m_algosInFlight
unsigned int m_algosInFlight
Number of algoritms presently in flight.
Definition: ForwardSchedulerSvc.h:177
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
ForwardSchedulerSvc::m_finishedEvents
tbb::concurrent_bounded_queue< EventContext * > m_finishedEvents
Queue of finished events.
Definition: ForwardSchedulerSvc.h:166
ForwardSchedulerSvc::m_thread
std::thread m_thread
The thread in which the activate function runs.
Definition: ForwardSchedulerSvc.h:142
ForwardSchedulerSvc::SchedulerState::operator<
bool operator<(const SchedulerState &rhs) const
Definition: ForwardSchedulerSvc.h:257
ForwardSchedulerSvc::m_algname_vect
std::vector< std::string > m_algname_vect
Vector to bookkeep the information necessary to the index2name conversion.
Definition: ForwardSchedulerSvc.h:154
ForwardSchedulerSvc::m_sState
static std::list< SchedulerState > m_sState
Definition: ForwardSchedulerSvc.h:265
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:160
ForwardSchedulerSvc::promoteToScheduled
StatusCode promoteToScheduled(unsigned int iAlgo, int si)
Definition: ForwardSchedulerSvc.cpp:844
ForwardSchedulerSvc::promoteToFinished
StatusCode promoteToFinished(unsigned int iAlgo, int si)
StatusCode
This class is used for returning status codes from appropriate routines.
Definition: StatusCode.h:26
std::ostringstream
STL class.
ForwardSchedulerSvc::SchedulerState::operator==
bool operator==(Algorithm *a) const
Definition: ForwardSchedulerSvc.h:255
ForwardSchedulerSvc::m_whiteboardSvcName
Gaudi::Property< std::string > m_whiteboardSvcName
Definition: ForwardSchedulerSvc.h:120
ForwardSchedulerSvc::algname2index
unsigned int algname2index(const std::string &algoname)
Convert a name to an integer.
Definition: ForwardSchedulerSvc.cpp:399
ForwardSchedulerSvc::enqueueSchedulerActionTask::execute
tbb::task * execute() override
Definition: ForwardSchedulerSvc.h:220
ForwardSchedulerSvc::promoteToExecuted
StatusCode promoteToExecuted(unsigned int iAlgo, int si, IAlgorithm *algo, EventContext *)
Definition: ForwardSchedulerSvc.cpp:909
ForwardSchedulerSvc::SchedulerState::operator==
bool operator==(const SchedulerState &ss) const
Definition: ForwardSchedulerSvc.h:253
ForwardSchedulerSvc::~ForwardSchedulerSvc
~ForwardSchedulerSvc() override=default
Destructor.
ForwardSchedulerSvc::freeSlots
unsigned int freeSlots() override
Get free slots number.
Definition: ForwardSchedulerSvc.cpp:467
ForwardSchedulerSvc::pushNewEvents
StatusCode pushNewEvents(std::vector< EventContext * > &eventContexts) override
Definition: ForwardSchedulerSvc.cpp:457
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:27
ForwardSchedulerSvc::m_actionsQueue
tbb::concurrent_bounded_queue< action > m_actionsQueue
Queue where closures are stored and picked for execution.
Definition: ForwardSchedulerSvc.h:209
ForwardSchedulerSvc::m_updateNeeded
bool m_updateNeeded
Keep track of update actions scheduled.
Definition: ForwardSchedulerSvc.h:197
ForwardSchedulerSvc::tryPopFinishedEvent
StatusCode tryPopFinishedEvent(EventContext *&eventContext) override
Try to fetch an event from the scheduler.
Definition: ForwardSchedulerSvc.cpp:512
Algorithm
Base class from which all concrete algorithm classes should be derived.
Definition: Algorithm.h:78
ForwardSchedulerSvc::SchedulerState::m_t
pthread_t m_t
Definition: ForwardSchedulerSvc.h:262
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:148
ForwardSchedulerSvc::activate
void activate()
Activate scheduler.
Definition: ForwardSchedulerSvc.cpp:330
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:172
ForwardSchedulerSvc::promoteToControlReady
StatusCode promoteToControlReady(unsigned int iAlgo, int si)
Algorithm promotion: Accepted by the control flow.
Definition: ForwardSchedulerSvc.cpp:812
ForwardSchedulerSvc::m_efManager
concurrency::ExecutionFlowManager m_efManager
Member to take care of the control flow.
Definition: ForwardSchedulerSvc.h:229
ForwardSchedulerSvc::popFinishedEvent
StatusCode popFinishedEvent(EventContext *&eventContext) override
Blocks until an event is availble.
Definition: ForwardSchedulerSvc.cpp:490
ForwardSchedulerSvc::dumpSchedulerState
void dumpSchedulerState(int iSlot)
Dump the state of the scheduler.
Definition: ForwardSchedulerSvc.cpp:741
ForwardSchedulerSvc::SchedulerState::m_a
Algorithm * m_a
Definition: ForwardSchedulerSvc.h:260
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:266
ForwardSchedulerSvc::index2algname
const std::string & index2algname(unsigned int index)
Convert an integer to a name.
Definition: ForwardSchedulerSvc.cpp:393
ForwardSchedulerSvc::delAlg
bool delAlg(Algorithm *)
Definition: ForwardSchedulerSvc.cpp:984
ForwardSchedulerSvc::SchedulerState::ctx
EventContext ctx() const
Definition: ForwardSchedulerSvc.h:243
ForwardSchedulerSvc::SchedulerState::operator<<
friend std::ostream & operator<<(std::ostream &os, const SchedulerState &ss)
Definition: ForwardSchedulerSvc.h:246
ForwardSchedulerSvc::m_isActive
std::atomic< ActivationState > m_isActive
Flag to track if the scheduler is active or not.
Definition: ForwardSchedulerSvc.h:139
AlgsExecutionStates.h
ForwardSchedulerSvc::SchedulerState::alg
Algorithm * alg() const
Definition: ForwardSchedulerSvc.h:242
ForwardSchedulerSvc::m_first
bool m_first
Definition: ForwardSchedulerSvc.h:234
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:1010
std::thread
STL class.
EventSlot.h
ForwardSchedulerSvc::promoteToDataReady
StatusCode promoteToDataReady(unsigned int iAlgo, int si)
Definition: ForwardSchedulerSvc.cpp:826
AtlasMCRecoScenario.scheduler
scheduler
Definition: AtlasMCRecoScenario.py:23
std::unordered_map< std::string, unsigned int >