1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
|
// -*- C++ -*-
//=============================================================================
/**
* @file TP_Reactor.h
*
* The ACE_TP_Reactor (aka, Thread Pool Reactor) uses the
* Leader/Followers pattern to demultiplex events among a pool of
* threads. When using a thread pool reactor, an application
* pre-spawns a fixed number of threads. When these threads
* invoke the ACE_TP_Reactor's handle_events() method, one thread
* will become the leader and wait for an event. The other
* follower threads will queue up waiting for their turn to become
* the leader. When an event occurs, the leader will pick a
* follower to become the leader and go on to handle the event.
* The consequence of using ACE_TP_Reactor is the amortization of
* the costs used to create threads. The context switching cost
* will also reduce. Moreover, the total resources used by
* threads are bounded because there are a fixed number of threads.
*
* @author Irfan Pyarali <irfan@cs.wustl.edu>
* @author Nanbor Wang <nanbor@cs.wustl.edu>
*/
//=============================================================================
#ifndef ACE_TP_REACTOR_H
#define ACE_TP_REACTOR_H
#include /**/ "ace/pre.h"
#include "ace/Select_Reactor.h"
#include "ace/Timer_Queue.h" /* Simple forward decl won't work... */
#if !defined (ACE_LACKS_PRAGMA_ONCE)
# pragma once
#endif /* ACE_LACKS_PRAGMA_ONCE */
ACE_BEGIN_VERSIONED_NAMESPACE_DECL
/**
* @class ACE_EH_Dispatch_Info
*
* @brief This structure contains information of the activated event
* handler.
*/
class ACE_EH_Dispatch_Info
{
public:
ACE_EH_Dispatch_Info (void);
void set (ACE_HANDLE handle,
ACE_Event_Handler *event_handler,
ACE_Reactor_Mask mask,
ACE_EH_PTMF callback);
bool dispatch (void) const;
ACE_HANDLE handle_;
ACE_Event_Handler *event_handler_;
ACE_Reactor_Mask mask_;
ACE_EH_PTMF callback_;
int resume_flag_;
bool reference_counting_required_;
private:
bool dispatch_;
// Disallow copying and assignment.
ACE_EH_Dispatch_Info (const ACE_EH_Dispatch_Info &);
ACE_EH_Dispatch_Info &operator= (const ACE_EH_Dispatch_Info &);
};
/**
* @class ACE_TP_Token_Guard
*
* @brief A helper class that helps grabbing, releasing and waiting
* on tokens for a thread that tries calling handle_events ().
*
* In short, this class will be owned by one thread by creating on the
* stack. This class gives the status of the ownership of the token
* and manages the ownership
*/
class ACE_TP_Token_Guard
{
public:
/// Constructor that will grab the token for us
ACE_TP_Token_Guard (ACE_Select_Reactor_Token &token);
/// Destructor. This will release the token if it hasnt been
/// released till this point
~ACE_TP_Token_Guard (void);
/// Release the token ..
void release_token (void);
/// Returns whether the thread that created this object ownes the
/// token or not.
bool is_owner (void);
/// A helper method that grabs the token for us, after which the
/// thread that owns that can do some actual work.
int acquire_read_token (ACE_Time_Value *max_wait_time = 0);
/**
* A helper method that grabs the token for us, after which the
* thread that owns that can do some actual work. This differs from
* acquire_read_token() as it uses acquire () to get the token instead of
* acquire_read ()
*/
int acquire_token (ACE_Time_Value *max_wait_time = 0);
private:
// Disallow default construction.
ACE_TP_Token_Guard (void);
// Disallow copying and assignment.
ACE_TP_Token_Guard (const ACE_TP_Token_Guard &);
ACE_TP_Token_Guard &operator= (const ACE_TP_Token_Guard &);
private:
/// The Select Reactor token.
ACE_Select_Reactor_Token &token_;
/// Flag that indicate whether the thread that created this object
/// owns the token or not. A value of false indicates that this class
/// hasn't got the token (and hence the thread) and a value of true
/// vice-versa.
bool owner_;
};
/**
* @class ACE_TP_Reactor
*
* @brief Specialization of ACE_Select_Reactor to support thread-pool
* based event dispatching.
*
* One of the shortcomings of the ACE_Select_Reactor is that it
* does not support a thread pool-based event dispatching model,
* similar to the one in ACE_WFMO_Reactor. In ACE_Select_Reactor, only
* thread can call handle_events() at any given time. ACE_TP_Reactor
* removes this short-coming.
*
* ACE_TP_Reactor is a specialization of ACE_Select_Reactor to support
* thread pool-based event dispatching. This reactor takes advantage
* of the fact that events reported by @c select() are persistent if not
* acted upon immediately. It works by remembering the event handler
* which was just activated, suspending it for further I/O activities,
* releasing the internal lock (so that another thread can start waiting
* in the event loop) and then dispatching the event's handler outside the
* scope of the reactor lock. After the event handler has been dispatched
* the event handler is resumed for further I/O activity.
*
* This reactor implementation is best suited for situations when the
* callbacks to event handlers can take arbitrarily long and/or a number
* of threads are available to run the event loop. Note that I/O-processing
* callback code in event handlers (e.g. handle_input()) does not have to
* be modified or made thread-safe for this reactor. This is because
* before an I/O event is dispatched to an event handler, the handler is
* suspended; it is resumed by the reactor after the upcall completes.
* Therefore, multiple I/O events will not be made to one event handler
* multiple threads simultaneously. This suspend/resume protection does not
* apply to either timers scheduled with the reactor or to notifications
* requested via the reactor. When using timers and/or notifications you
* must provide proper protection for your class in the context of multiple
* threads.
*/
class ACE_Export ACE_TP_Reactor : public ACE_Select_Reactor
{
public:
/// Initialize ACE_TP_Reactor with the default size.
ACE_TP_Reactor (ACE_Sig_Handler * = 0,
ACE_Timer_Queue * = 0,
bool mask_signals = true,
int s_queue = ACE_Select_Reactor_Token::FIFO);
/**
* Initialize the ACE_TP_Reactor to manage
* @a max_number_of_handles. If @a restart is non-0 then the
* ACE_Reactor's @c handle_events() method will be restarted
* automatically when @c EINTR occurs. If @a sh or
* @a tq are non-0 they are used as the signal handler and
* timer queue, respectively.
*/
ACE_TP_Reactor (size_t max_number_of_handles,
bool restart = false,
ACE_Sig_Handler *sh = 0,
ACE_Timer_Queue *tq = 0,
bool mask_signals = true,
int s_queue = ACE_Select_Reactor_Token::FIFO);
/**
* This event loop driver that blocks for @a max_wait_time before
* returning. It will return earlier if timer events, I/O events,
* or signal events occur. Note that @a max_wait_time can be 0, in
* which case this method blocks indefinitely until events occur.
*
* @a max_wait_time is decremented to reflect how much time this call
* took. For instance, if a time value of 3 seconds is passed to
* handle_events and an event occurs after 2 seconds,
* @a max_wait_time will equal 1 second. This can be used if an
* application wishes to handle events for some fixed amount of
* time.
*
* @return The total number of events that were dispatched; 0 if the
* @a max_wait_time elapsed without dispatching any handlers, or -1
* if an error occurs (check @c errno for more information).
*/
virtual int handle_events (ACE_Time_Value *max_wait_time = 0);
virtual int handle_events (ACE_Time_Value &max_wait_time);
/// Does the reactor allow the application to resume the handle on
/// its own ie. can it pass on the control of handle resumption to
/// the application. The TP reactor has can allow applications to
/// resume handles. So return a positive value.
virtual int resumable_handler (void);
/// Called from handle events
static void no_op_sleep_hook (void *);
/// The ACE_TP_Reactor implementation does not have a single owner thread.
/// Attempts to set the owner explicitly are ignored. The reported owner
/// thread is the current Leader in the pattern.
virtual int owner (ACE_thread_t n_id, ACE_thread_t *o_id = 0);
/// Return the thread ID of the current Leader.
virtual int owner (ACE_thread_t *t_id);
/// Declare the dynamic allocation hooks.
ACE_ALLOC_HOOK_DECLARE;
protected:
// = Internal methods that do the actual work.
/// Template method from the base class.
virtual void clear_dispatch_mask (ACE_HANDLE handle,
ACE_Reactor_Mask mask);
/// Dispatch just 1 signal, timer, notification handlers
int dispatch_i (ACE_Time_Value *max_wait_time,
ACE_TP_Token_Guard &guard);
/// Get the event that needs dispatching. It could be either a
/// signal, timer, notification handlers or return possibly 1 I/O
/// handler for dispatching. In the most common use case, this would
/// return 1 I/O handler for dispatching
int get_event_for_dispatching (ACE_Time_Value *max_wait_time);
#if 0
// @Ciju
// signal handling isn't in a production state yet.
// Commenting it out for now.
/// Method to handle signals
/// @note It is just busted at this point in time.
int handle_signals (int &event_count,
ACE_TP_Token_Guard &g);
#endif // #if 0
/// Handle timer events
int handle_timer_events (int &event_count,
ACE_TP_Token_Guard &g);
/// Handle notify events
int handle_notify_events (int &event_count,
ACE_TP_Token_Guard &g);
/// handle socket events
int handle_socket_events (int &event_count,
ACE_TP_Token_Guard &g);
/// This method shouldn't get called.
virtual void notify_handle (ACE_HANDLE handle,
ACE_Reactor_Mask mask,
ACE_Handle_Set &,
ACE_Event_Handler *eh,
ACE_EH_PTMF callback);
private:
/// Get the handle of the notify pipe from the ready set if there is
/// an event in the notify pipe.
ACE_HANDLE get_notify_handle (void);
/// Get socket event dispatch information.
int get_socket_event_info (ACE_EH_Dispatch_Info &info);
/// Notify the appropriate <callback> in the context of the <eh>
/// associated with <handle> that a particular event has occurred.
int dispatch_socket_event (ACE_EH_Dispatch_Info &dispatch_info);
/// Clear the @a handle from the read_set
void clear_handle_read_set (ACE_HANDLE handle);
int post_process_socket_event (ACE_EH_Dispatch_Info &dispatch_info,int status);
private:
/// Deny access since member-wise won't work...
ACE_TP_Reactor (const ACE_TP_Reactor &);
ACE_TP_Reactor &operator = (const ACE_TP_Reactor &);
};
ACE_END_VERSIONED_NAMESPACE_DECL
#if defined (__ACE_INLINE__)
#include "ace/TP_Reactor.inl"
#endif /* __ACE_INLINE__ */
#include /**/ "ace/post.h"
#endif /* ACE_TP_REACTOR_H */
|