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
|
<HTML>
<HEAD>
<META HTTP-EQUIV="Content-Type" CONTENT="text/html; charset=iso-8859-1">
<META NAME="Author" CONTENT="James CE Johnson">
<TITLE>ACE Tutorial 017</TITLE>
</HEAD>
<BODY TEXT="#000000" BGCOLOR="#FFFFFF" LINK="#000FFF" VLINK="#FF0F0F">
<CENTER><B><FONT SIZE=+2>ACE Tutorial 017</FONT></B></CENTER>
<CENTER><B><FONT SIZE=+2>Using the ACE_Barrier synch object</FONT></B></CENTER>
<P>
<HR WIDTH="100%">
I could have included this in the first Test object of the tutorial
but that may have complicated things a bit. What we're doing here is
recognizing when the "owner" thread adds more threads to the pool.
When we notice that, we use the barrier to wait until everything
stabilizes and then we recalibrate and move on.
<P>
The source is <A HREF="barrier2.cpp">here</A>.
<HR>
<PRE>
<font color=red>// $Id$</font>
<font color=blue>#include</font> "<font color=green>Barrier_i.h</font>"
<font color=blue>#include</font> "<A HREF="../../../ace/Task.h">ace/Task.h</A>"
<font color=red>/* We'll use a simple Task<> derivative to test our new Barrier
object.
*/</font>
class Test : public ACE_Task<ACE_NULL_SYNCH>
{
public:
<font color=red>// Construct the object with a desired thread count</font>
Test(int _threads);
<font color=red>// Open/begin the test. As usual, we have to match the</font>
<font color=red>// ACE_Task signature.</font>
int open(void * _unused = 0);
<font color=red>// Change the threads_ value for the next invocation of open()</font>
void threads(int _threads);
<font color=red>// Get the current threads_ value.</font>
int threads(void);
<font color=red>// Perform the test</font>
int svc(void);
protected:
<font color=red>// How many threads the barrier will test.</font>
u_int threads_;
<font color=red>// The Barrier object we'll use in our tests below</font>
Barrier barrier_;
<font color=red>// This lets us pick one (eg -- the first) thread as the</font>
<font color=red>// "<font color=green>controller</font>" for our little test...</font>
ACE_Atomic_Op<ACE_Mutex,u_int> tcount_;
};
<font color=red>/* Construct the object & initialize the threads value for open() to
use.
*/</font>
<font color=#008888>Test::Test</font>(int _threads)
: threads_(_threads), tcount_(0)
{
}
<font color=red>/* As usual, our open() will create one or more threads where we'll do
the interesting work.
*/</font>
int <font color=#008888>Test::open</font>(void * _unused)
{
ACE_UNUSED_ARG(_unused);
<font color=red>// One thing about the barrier: You have to tell it how many</font>
<font color=red>// threads it will be synching. The threads() mutator on my</font>
<font color=red>// Barrier class lets you do that and hides the implementation </font>
<font color=red>// details at the same time.</font>
barrier_.threads(threads_);
<font color=red>// Activate the tasks as usual...</font>
return this->activate(THR_NEW_LWP, threads_, 1);
}
void <font color=#008888>Test::threads</font>(int _threads)
{
threads_ = _threads;
}
int <font color=#008888>Test::threads</font>(void)
{
return threads_;
}
<font color=red>/* svc() will execute in each thread & do a few things with the
Barrier we have.
*/</font>
int <font color=#008888>Test::svc</font>(void)
{
<font color=red>// Say hello to everyone first.</font>
ACE_DEBUG(( LM_INFO, "<font color=green>(%P|%t|%T) Created\n</font>" ));
<font color=red>// Increment and save the "<font color=green>tcount</font>" value. We'll use it in</font>
<font color=red>// just a moment...</font>
int me = ++tcount_;
<font color=red>// Wait for all initial threads to get to this point before we</font>
<font color=red>// go any further. This is standard barrier usage...</font>
barrier_.wait();
<font color=red>// Setup our random number generator.</font>
ACE_Time_Value now(<font color=#008888>ACE_OS::gettimeofday</font>());
ACE_RANDR_TYPE seed = now.usec();
<font color=#008888>ACE_OS::srand</font>(seed);
int delay;
<font color=red>// We'll arbitrarily choose the first activated thread to be</font>
<font color=red>// the controller. After it sleeps a few seconds, it will add </font>
<font color=red>// five threads.</font>
if( me == 1 )
{
<font color=red>// Sleep from 1 to 10 seconds so that some of the other</font>
<font color=red>// threads will be into their for() loop.</font>
delay = <font color=#008888>ACE_OS::rand_r</font>(seed)%10;
<font color=#008888>ACE_OS::sleep</font>(abs(delay)+1);
<font color=red>// Make ourselves the barrier owner so that we can change</font>
<font color=red>// the number of threads. This should be done with care...</font>
barrier_.owner( <font color=#008888>ACE_OS::thr_self</font>() );
<font color=red>// Add 5 threads to the barrier and then activate() to</font>
<font color=red>// make them real. Notice the third parameter to</font>
<font color=red>// activate(). Without this parameter, the threads won't</font>
<font color=red>// be created.</font>
if( barrier_.threads(threads_+5) == 0 )
{
this->activate(THR_NEW_LWP,5,1);
}
}
<font color=red>// This for() loop represents an "<font color=green>infinite</font>" work loop in an</font>
<font color=red>// application. The theory is that the threads are dividing up </font>
<font color=red>// some work but need to "<font color=green>recalibrate</font>" if more threads are</font>
<font color=red>// added. I'll just do five iterations so that the test</font>
<font color=red>// doesn't run forever.</font>
int i;
for( i = 0 ; i < 5 ; ++i )
{
<font color=red>// The sleep() represents time doing work.</font>
delay = <font color=#008888>ACE_OS::rand_r</font>(seed)%7;
<font color=#008888>ACE_OS::sleep</font>(abs(delay)+1);
ACE_DEBUG(( LM_INFO, "<font color=green>(%P|%t|%T)\tThread %.2d of %.2d iteration %.2d\n</font>", me, threads_, i ));
<font color=red>// If the local threads_ variable doesn't match the number </font>
<font color=red>// in the barrier, then the controller must have changed</font>
<font color=red>// the thread count. We'll wait() for everyone and then</font>
<font color=red>// recalibrate ourselves before continuing.</font>
if( this->threads_ != barrier_.threads() )
{
ACE_DEBUG(( LM_INFO, "<font color=green>(%P|%t|%T) Waiting for thread count to increase to %d from %d\n</font>",
barrier_.threads(), this->threads_ ));
<font color=red>// Wait for all our sibling threads...</font>
barrier_.wait();
<font color=red>// Set our local variable so that we don't come here again.</font>
this->threads_ = barrier_.threads();
<font color=red>// Recalibration can be anything you want. At this</font>
<font color=red>// point, we know that all of the threads are synch'd</font>
<font color=red>// and ready to go.</font>
}
}
<font color=red>// Re-synch all of the threads before they exit. This isn't</font>
<font color=red>// really necessary but I like to do it.</font>
barrier_.done();
return(0);
}
<font color=red>/* Our test application...
*/</font>
int main(int, char**)
{
<font color=red>// Create the test object with 5 threads</font>
Test test(5);
<font color=red>// and open it to test the barrier.</font>
test.open();
<font color=red>// Now wait for them all to exit.</font>
test.wait();
return(0);
}
</PRE>
<P><HR WIDTH="100%">
<CENTER>[<A HREF="../online-tutorials.html">Tutorial Index</A>] </CENTER>
|
