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
|
// $Id$
// ============================================================================
//
// = LIBRARY
// tests
//
// = FILENAME
// Reactors_Test.cpp
//
// = DESCRIPTION
// This is a test that illustrates how the Reactor's notify()
// method works under various settings of max_notify_iterations().
//
// = AUTHOR
// Douglas C. Schmidt
//
// ============================================================================
#include "ace/Synch.h"
#include "ace/Task.h"
#include "ace/Pipe.h"
#include "test_config.h"
#if defined (ACE_HAS_THREADS)
class Supplier_Task : public ACE_Task<ACE_MT_SYNCH>
{
public:
Supplier_Task (void);
// Constructor.
~Supplier_Task (void);
// Destructor.
virtual int open (void * = 0);
// Make this an Active Object.
virtual int close (u_long);
// Close down the supplier.
virtual int svc (void);
// Generates events and sends them to the <Reactor>'s <notify>
// method.
virtual int handle_exception (ACE_HANDLE);
// Releases the <waiter_> semaphore when called by the <Reactor>'s
// notify handler.
virtual int handle_output (ACE_HANDLE);
// Called every time through the main Reactor event loop to
// illustrate the difference between "limited" and "unlimited"
// notification.
private:
ACE_Thread_Semaphore waiter_;
// Used to hand-shake between the <Supplier_Task> and the
// <Reactor>'s notify mechanism.
ACE_Pipe pipe_;
// We use this pipe just so we can get a handle that is always
// "active."
};
Supplier_Task::Supplier_Task (void)
: waiter_ (0) // Make semaphore "locked" by default.
{
}
int
Supplier_Task::open (void *)
{
if (this->pipe_.open () == -1)
ACE_ERROR_RETURN ((LM_ERROR, "(%t) %p\n", "open failed"), -1);
else if (this->activate (THR_BOUND | THR_DETACHED))
// Make this an Active Object.
ACE_ERROR_RETURN ((LM_ERROR, "(%t) %p\n", "activate failed"), -1);
else if (ACE_Reactor::instance()->register_handler
(this->pipe_.write_handle (), this, ACE_Event_Handler::WRITE_MASK) == -1)
ACE_ERROR_RETURN ((LM_ERROR, "(%t) %p\n", "register_handler failed"), -1);
else
return 0;
}
int
Supplier_Task::close (u_long)
{
ACE_DEBUG ((LM_DEBUG, "(%t) Supplier_Task::close\n"));
if (ACE_Reactor::instance()->remove_handler
(this->pipe_.write_handle (), ACE_Event_Handler::WRITE_MASK) == -1)
ACE_ERROR ((LM_ERROR, "(%t) %p\n", "remove_handler failed"));
return 0;
}
Supplier_Task::~Supplier_Task (void)
{
ACE_DEBUG ((LM_DEBUG, "(%t) ~Supplier_Task\n"));
this->pipe_.close ();
}
int
Supplier_Task::svc (void)
{
size_t i;
ACE_DEBUG ((LM_DEBUG, "(%t) **** starting unlimited notifications test\n"));
// Allow an unlimited number of iterations per
// <ACE_Reactor::notify>.
ACE_Reactor::instance()->max_notify_iterations (-1);
for (i = 0; i < ACE_MAX_ITERATIONS; i++)
{
ACE_DEBUG ((LM_DEBUG, "(%t) notifying reactor\n"));
// Notify the Reactor.
if (ACE_Reactor::instance()->notify (this) == -1)
ACE_ERROR_RETURN ((LM_ERROR, "(%t) %p\n", "notify"), -1);
// Wait for our <handle_exception> method to release the
// semaphore.
else if (this->waiter_.acquire () == -1)
ACE_ERROR_RETURN ((LM_ERROR, "(%t) %p\n", "acquire"), -1);
}
ACE_DEBUG ((LM_DEBUG, "(%t) **** starting limited notifications test\n"));
// Only allow 1 iteration per <ACE_Reactor::notify>
ACE_Reactor::instance()->max_notify_iterations (1);
for (i = 0; i < ACE_MAX_ITERATIONS; i++)
{
ACE_DEBUG ((LM_DEBUG, "(%t) notifying reactor\n"));
// Notify the Reactor.
if (ACE_Reactor::instance()->notify (this) == -1)
ACE_ERROR_RETURN ((LM_ERROR, "(%t) %p\n", "notify"), -1);
// Wait for our <handle_exception> method to release the
// semaphore.
else if (this->waiter_.acquire () == -1)
ACE_ERROR_RETURN ((LM_ERROR, "(%t) %p\n", "acquire"), -1);
}
ACE_DEBUG ((LM_DEBUG, "(%t) **** exiting thread test\n"));
return 0;
}
int
Supplier_Task::handle_exception (ACE_HANDLE handle)
{
ACE_ASSERT (handle == ACE_INVALID_HANDLE);
ACE_DEBUG ((LM_DEBUG, "(%t) handle_exception\n"));
this->waiter_.release ();
return 0;
}
int
Supplier_Task::handle_output (ACE_HANDLE handle)
{
ACE_ASSERT (handle == this->pipe_.write_handle ());
ACE_DEBUG ((LM_DEBUG, "(%t) handle_output\n"));
return 0;
}
#endif /* ACE_HAS_THREADS */
int
main (int, char *[])
{
ACE_START_TEST ("Reactor_Notify_Test");
#if defined (ACE_HAS_THREADS)
ACE_ASSERT (ACE_LOG_MSG->op_status () != -1);
Supplier_Task task;
ACE_ASSERT (ACE_LOG_MSG->op_status () != -1);
if (task.open () == -1)
ACE_ERROR_RETURN ((LM_ERROR, "(%t) open failed\n"), -1);
else
{
ACE_Time_Value timeout (2);
int shutdown = 0;
for (int iteration = 1; !shutdown; iteration++)
{
// Use a timeout to inform the Reactor when to shutdown.
switch (ACE_Reactor::instance()->handle_events (timeout))
{
case -1:
ACE_ERROR_RETURN ((LM_ERROR, "(%t) %p\n", "reactor"), -1);
/* NOTREACHED */
case 0:
shutdown = 1;
break;
default:
// ACE_DEBUG ((LM_DEBUG, "(%t) done dispatching %d\n", iteration));
;
}
}
}
#else
ACE_ERROR ((LM_ERROR, "threads not supported on this platform\n"));
#endif /* ACE_HAS_THREADS */
ACE_END_TEST;
return 0;
}
|
