| The Gaudi Framework v32r1 (f65d50dc)

IOBoundAlgSchedulerSvc.cpp
Go to the documentation of this file.
1 #include <algorithm>
2 #include <map>
3 #include <queue>
4 #include <sstream>
5 #include <thread>
6 #include <unordered_set>
7
8 // Framework includes
9 #include "GaudiAlg/GaudiAlgorithm.h"
10 #include "GaudiKernel/IAlgorithm.h"
11 #include "GaudiKernel/IDataManagerSvc.h"
12 #include <Gaudi/Algorithm.h> // will be IAlgorithm if context getter promoted to interface
13
14 // Local
15 #include "IOBoundAlgSchedulerSvc.h"
16
17 // Instantiation of a static factory class used by clients to create instances of this service
18 DECLARE_COMPONENT( IOBoundAlgSchedulerSvc )
19
20
24 StatusCode IOBoundAlgSchedulerSvc::initialize() {
25
26 // Initialise mother class (read properties, ...)
27 StatusCode sc( Service::initialize() );
28 if ( !sc.isSuccess() ) warning() << "Base class could not be initialized" << endmsg;
29
30 // Activate the scheduler in another thread.
31 info() << "Activating scheduler in a separate thread" << endmsg;
32 m_thread = std::thread( std::bind( &IOBoundAlgSchedulerSvc::activate, this ) );
33
34 return sc;
35 }
36 //---------------------------------------------------------------------------
37
41 StatusCode IOBoundAlgSchedulerSvc::finalize() {
42
43 StatusCode sc( Service::finalize() );
44 if ( !sc.isSuccess() ) warning() << "Base class could not be finalized" << endmsg;
45
46 sc = deactivate();
47 if ( !sc.isSuccess() ) warning() << "Scheduler could not be deactivated" << endmsg;
48
49 info() << "Joining preemptive scheduler's thread" << endmsg;
50 m_thread.join();
51
52 return sc;
53 }
54 //---------------------------------------------------------------------------
61 void IOBoundAlgSchedulerSvc::activate() {
62
63 // Now it's running
64 m_isActive = true;
65
66 // Wait for actions pushed into the queue by finishing tasks.
67 action thisAction;
68 StatusCode sc( StatusCode::SUCCESS );
69
70 // Continue to wait if the scheduler is running or there is something to do
71 info() << "Start checking the queue of I/O-bound algorithm tasks.." << endmsg;
72 while ( m_isActive or m_actionsQueue.size() > 0 ) {
73 m_actionsQueue.pop( thisAction );
74 std::thread th( thisAction );
75 th.detach();
76 }
77 }
78
79 //---------------------------------------------------------------------------
80
87 StatusCode IOBoundAlgSchedulerSvc::deactivate() {
88
89 if ( m_isActive ) {
90 // Drain the scheduler
91 // m_actionsQueue.push(std::bind(&IOBoundSchedulerSvc::m_drain,
92 // this));
93 // we set the flag in this thread, not in the last action, to avoid stall,
94 // since we execute tasks asynchronously, in a detached thread, and it's possible that
95 // the task doesn't complete by the time the last while-iteration is entered
96 m_isActive = false;
97 // This would be the last (empty) action, just to trigger one last while-iteration
98 m_actionsQueue.push( []() -> StatusCode { return StatusCode::SUCCESS; } );
99 }
100
101 return StatusCode::SUCCESS;
102 }
103
104 StatusCode IOBoundAlgSchedulerSvc::push( IAlgTask& task ) {
105
106 // the temporary lambda should be moved into the queue in here
107 auto actionn = [&]() {
108 debug() << " .. launching I/O-bound algo-closure .. " << endmsg;
109 return task.execute();
110 };
111
112 m_actionsQueue.push( actionn );
113
114 // until the queue to accelerator becomes limited
115 return StatusCode::SUCCESS;
116 }
117
118 //===========================================================================
Service::initialize
StatusCode initialize() override
Definition: Service.cpp:60
Service::finalize
StatusCode finalize() override
Definition: Service.cpp:164
std::function< StatusCode()>
CommonMessagingBase::warning
MsgStream & warning() const
shortcut for the method msgStream(MSG::WARNING)
Definition: CommonMessaging.h:101
IAlgorithm.h
IOBoundAlgSchedulerSvc::m_isActive
bool m_isActive
Flag to track if the scheduler is active or not.
Definition: IOBoundAlgSchedulerSvc.h:48
GaudiAlgorithm.h
Header file for class GaudiAlgorithm.
IOBoundAlgSchedulerSvc::m_thread
std::thread m_thread
The thread in which the activate function runs.
Definition: IOBoundAlgSchedulerSvc.h:51
StatusCode::SUCCESS
constexpr static const auto SUCCESS
Definition: StatusCode.h:85
IAlgTask::execute
virtual StatusCode execute()=0
CommonMessagingBase::info
MsgStream & info() const
shortcut for the method msgStream(MSG::INFO)
Definition: CommonMessaging.h:104
IOBoundAlgSchedulerSvc.h
DECLARE_COMPONENT
#define DECLARE_COMPONENT(type)
Definition: PluginServiceV1.h:46
IAlgTask
General interface for a wrapper around Gaudi algorithm.
Definition: IAlgTask.h:15
IDataManagerSvc.h
std::thread::join
T join(T... args)
Algorithm.h
StatusCode
This class is used for returning status codes from appropriate routines.
Definition: StatusCode.h:50
IOBoundAlgSchedulerSvc::deactivate
StatusCode deactivate()
Deactivate scheduler.
Definition: IOBoundAlgSchedulerSvc.cpp:87
std::bind
T bind(T... args)
CommonMessagingBase::debug
MsgStream & debug() const
shortcut for the method msgStream(MSG::DEBUG)
Definition: CommonMessaging.h:107
StatusCode::isSuccess
bool isSuccess() const
Definition: StatusCode.h:267
IOBoundAlgSchedulerSvc::activate
void activate()
Activate scheduler.
Definition: IOBoundAlgSchedulerSvc.cpp:61
IOBoundAlgSchedulerSvc::push
StatusCode push(IAlgTask &task) override
Add an algorithm to local queue to run on accelerator.
Definition: IOBoundAlgSchedulerSvc.cpp:104
std::thread::detach
T detach(T... args)
IOBoundAlgSchedulerSvc::finalize
StatusCode finalize() override
Finalise.
Definition: IOBoundAlgSchedulerSvc.cpp:41
IOBoundAlgSchedulerSvc
Please refer to the full documentation of the methods for more details.
Definition: IOBoundAlgSchedulerSvc.h:25
IOBoundAlgSchedulerSvc::m_actionsQueue
tbb::concurrent_bounded_queue< action > m_actionsQueue
This is done since the copy of the lambda storage is too expensive.
Definition: IOBoundAlgSchedulerSvc.h:59
std::thread
STL class.
endmsg
MsgStream & endmsg(MsgStream &s)
MsgStream Modifier: endmsg. Calls the output method of the MsgStream.
Definition: MsgStream.h:192