KickJava   Java API By Example, From Geeks To Geeks.

Java > Open Source Codes > EDU > oswego > cs > dl > util > concurrent > CyclicBarrier


1 /*
2   File: CyclicBarrier.java
3
4   Originally written by Doug Lea and released into the public domain.
5   This may be used for any purposes whatsoever without acknowledgment.
6   Thanks for the assistance and support of Sun Microsystems Labs,
7   and everyone contributing, testing, and using this code.
8
9   History:
10   Date Who What
11   11Jul1998 dl Create public version
12   28Aug1998 dl minor code simplification
13 */

14
15 package EDU.oswego.cs.dl.util.concurrent;
16
17 /**
18  * A cyclic barrier is a reasonable choice for a barrier in contexts
19  * involving a fixed sized group of threads that
20  * must occasionally wait for each other.
21  * (A Rendezvous better handles applications in which
22  * any number of threads meet, n-at-a-time.)
23  * <p>
24  * CyclicBarriers use an all-or-none breakage model
25  * for failed synchronization attempts: If threads
26  * leave a barrier point prematurely because of timeout
27  * or interruption, others will also leave abnormally
28  * (via BrokenBarrierException), until
29  * the barrier is <code>restart</code>ed. This is usually
30  * the simplest and best strategy for sharing knowledge
31  * about failures among cooperating threads in the most
32  * common usages contexts of Barriers.
33  * This implementation has the property that interruptions
34  * among newly arriving threads can cause as-yet-unresumed
35  * threads from a previous barrier cycle to return out
36  * as broken. This transmits breakage
37  * as early as possible, but with the possible byproduct that
38  * only some threads returning out of a barrier will realize
39  * that it is newly broken. (Others will not realize this until a
40  * future cycle.) (The Rendezvous class has a more uniform, but
41  * sometimes less desirable policy.)
42  * <p>
43  * Barriers support an optional Runnable command
44  * that is run once per barrier point.
45  * <p>
46  * <b>Sample usage</b> Here is a code sketch of
47  * a barrier in a parallel decomposition design.
48  * <pre>
49  * class Solver {
50  * final int N;
51  * final float[][] data;
52  * final CyclicBarrier barrier;
53  *
54  * class Worker implements Runnable {
55  * int myRow;
56  * Worker(int row) { myRow = row; }
57  * public void run() {
58  * while (!done()) {
59  * processRow(myRow);
60  *
61  * try {
62  * barrier.barrier();
63  * }
64  * catch (InterruptedException ex) { return; }
65  * catch (BrokenBarrierException ex) { return; }
66  * }
67  * }
68  * }
69  *
70  * public Solver(float[][] matrix) {
71  * data = matrix;
72  * N = matrix.length;
73  * barrier = new CyclicBarrier(N);
74  * barrier.setBarrierCommand(new Runnable() {
75  * public void run() { mergeRows(...); }
76  * });
77  * for (int i = 0; i < N; ++i) {
78  * new Thread(new Worker(i)).start();
79  * waitUntilDone();
80  * }
81  * }
82  * </pre>
83  * <p>[<a HREF="http://gee.cs.oswego.edu/dl/classes/EDU/oswego/cs/dl/util/concurrent/intro.html"> Introduction to this package. </a>]
84
85  **/

86 public class CyclicBarrier implements Barrier {
87
88   protected final int parties_;
89   protected boolean broken_ = false;
90   protected Runnable JavaDoc barrierCommand_ = null;
91   protected int count_; // number of parties still waiting
92
protected int resets_ = 0; // incremented on each release
93

94   /**
95    * Create a CyclicBarrier for the indicated number of parties,
96    * and no command to run at each barrier.
97    * @exception IllegalArgumentException if parties less than or equal to zero.
98    **/

99
100   public CyclicBarrier(int parties) { this(parties, null); }
101
102   /**
103    * Create a CyclicBarrier for the indicated number of parties.
104    * and the given command to run at each barrier point.
105    * @exception IllegalArgumentException if parties less than or equal to zero.
106    **/

107
108   public CyclicBarrier(int parties, Runnable JavaDoc command) {
109     if (parties <= 0) throw new IllegalArgumentException JavaDoc();
110     parties_ = parties;
111     count_ = parties;
112     barrierCommand_ = command;
113   }
114
115   /**
116    * Set the command to run at the point at which all threads reach the
117    * barrier. This command is run exactly once, by the thread
118    * that trips the barrier. The command is not run if the barrier is
119    * broken.
120    * @param command the command to run. If null, no command is run.
121    * @return the previous command
122    **/

123
124   public synchronized Runnable JavaDoc setBarrierCommand(Runnable JavaDoc command) {
125     Runnable JavaDoc old = barrierCommand_;
126     barrierCommand_ = command;
127     return old;
128   }
129
130   public synchronized boolean broken() { return broken_; }
131
132   /**
133    * Reset to initial state. Clears both the broken status
134    * and any record of waiting threads, and releases all
135    * currently waiting threads with indeterminate return status.
136    * This method is intended only for use in recovery actions
137    * in which it is somehow known
138    * that no thread could possibly be relying on the
139    * the synchronization properties of this barrier.
140    **/

141
142   public synchronized void restart() {
143     broken_ = false;
144     ++resets_;
145     count_ = parties_;
146     notifyAll();
147   }
148   
149  
150   public int parties() { return parties_; }
151
152   /**
153    * Enter barrier and wait for the other parties()-1 threads.
154    * @return the arrival index: the number of other parties
155    * that were still waiting
156    * upon entry. This is a unique value from zero to parties()-1.
157    * If it is zero, then the current
158    * thread was the last party to hit barrier point
159    * and so was responsible for releasing the others.
160    * @exception BrokenBarrierException if any other thread
161    * in any previous or current barrier
162    * since either creation or the last <code>restart</code>
163    * operation left the barrier
164    * prematurely due to interruption or time-out. (If so,
165    * the <code>broken</code> status is also set.)
166    * Threads that are noticied to have been
167    * interrupted <em>after</em> being released are not considered
168    * to have broken the barrier.
169    * In all cases, the interruption
170    * status of the current thread is preserved, so can be tested
171    * by checking <code>Thread.interrupted</code>.
172    * @exception InterruptedException if this thread was interrupted
173    * during the barrier, and was the one causing breakage.
174    * If so, <code>broken</code> status is also set.
175    **/

176
177   public int barrier() throws InterruptedException JavaDoc, BrokenBarrierException {
178     return doBarrier(false, 0);
179   }
180
181   /**
182    * Enter barrier and wait at most msecs for the other parties()-1 threads.
183    * @return if not timed out, the arrival index: the number of other parties
184    * that were still waiting
185    * upon entry. This is a unique value from zero to parties()-1.
186    * If it is zero, then the current
187    * thread was the last party to hit barrier point
188    * and so was responsible for releasing the others.
189    * @exception BrokenBarrierException
190    * if any other thread
191    * in any previous or current barrier
192    * since either creation or the last <code>restart</code>
193    * operation left the barrier
194    * prematurely due to interruption or time-out. (If so,
195    * the <code>broken</code> status is also set.)
196    * Threads that are noticed to have been
197    * interrupted <em>after</em> being released are not considered
198    * to have broken the barrier.
199    * In all cases, the interruption
200    * status of the current thread is preserved, so can be tested
201    * by checking <code>Thread.interrupted</code>.
202    * @exception InterruptedException if this thread was interrupted
203    * during the barrier. If so, <code>broken</code> status is also set.
204    * @exception TimeoutException if this thread timed out waiting for
205    * the barrier. If the timeout occured while already in the
206    * barrier, <code>broken</code> status is also set.
207    **/

208
209   public int attemptBarrier(long msecs)
210     throws InterruptedException JavaDoc, TimeoutException, BrokenBarrierException {
211     return doBarrier(true, msecs);
212   }
213
214   protected synchronized int doBarrier(boolean timed, long msecs)
215     throws InterruptedException JavaDoc, TimeoutException, BrokenBarrierException {
216     
217     int index = --count_;
218
219     if (broken_) {
220       throw new BrokenBarrierException(index);
221     }
222     else if (Thread.interrupted()) {
223       broken_ = true;
224       notifyAll();
225       throw new InterruptedException JavaDoc();
226     }
227     else if (index == 0) { // tripped
228
count_ = parties_;
229       ++resets_;
230       notifyAll();
231       try {
232         if (barrierCommand_ != null)
233           barrierCommand_.run();
234         return 0;
235       }
236       catch (RuntimeException JavaDoc ex) {
237         broken_ = true;
238         return 0;
239       }
240     }
241     else if (timed && msecs <= 0) {
242       broken_ = true;
243       notifyAll();
244       throw new TimeoutException(msecs);
245     }
246     else { // wait until next reset
247
int r = resets_;
248       long startTime = (timed)? System.currentTimeMillis() : 0;
249       long waitTime = msecs;
250       for (;;) {
251         try {
252           wait(waitTime);
253         }
254         catch (InterruptedException JavaDoc ex) {
255           // Only claim that broken if interrupted before reset
256
if (resets_ == r) {
257             broken_ = true;
258             notifyAll();
259             throw ex;
260           }
261           else {
262             Thread.currentThread().interrupt(); // propagate
263
}
264         }
265
266         if (broken_)
267           throw new BrokenBarrierException(index);
268
269         else if (r != resets_)
270           return index;
271
272         else if (timed) {
273           waitTime = msecs - (System.currentTimeMillis() - startTime);
274           if (waitTime <= 0) {
275             broken_ = true;
276             notifyAll();
277             throw new TimeoutException(msecs);
278           }
279         }
280       }
281     }
282   }
283
284 }
285
Popular Tags