ViewVC Help
View File | Revision Log | Show Annotations | Download File | Root Listing
root/jsr166/jsr166/src/jsr166y/Phaser.java
(Generate patch)

Comparing jsr166/src/jsr166y/Phaser.java (file contents):
Revision 1.6 by dl, Tue Oct 28 23:03:24 2008 UTC vs.
Revision 1.75 by dl, Wed Sep 21 12:30:39 2011 UTC

# Line 1 | Line 1
1   /*
2   * Written by Doug Lea with assistance from members of JCP JSR-166
3   * Expert Group and released to the public domain, as explained at
4 < * http://creativecommons.org/licenses/publicdomain
4 > * http://creativecommons.org/publicdomain/zero/1.0/
5   */
6  
7   package jsr166y;
8 < import java.util.concurrent.*;
9 < import java.util.concurrent.atomic.*;
8 >
9 > import java.util.concurrent.TimeUnit;
10 > import java.util.concurrent.TimeoutException;
11 > import java.util.concurrent.atomic.AtomicReference;
12   import java.util.concurrent.locks.LockSupport;
11 import sun.misc.Unsafe;
12 import java.lang.reflect.*;
13  
14   /**
15 < * A reusable synchronization barrier, similar in functionality to a
16 < * {@link java.util.concurrent.CyclicBarrier} and {@link
17 < * java.util.concurrent.CountDownLatch} but supporting more flexible
18 < * usage.
15 > * A reusable synchronization barrier, similar in functionality to
16 > * {@link java.util.concurrent.CyclicBarrier CyclicBarrier} and
17 > * {@link java.util.concurrent.CountDownLatch CountDownLatch}
18 > * but supporting more flexible usage.
19 > *
20 > * <p> <b>Registration.</b> Unlike the case for other barriers, the
21 > * number of parties <em>registered</em> to synchronize on a phaser
22 > * may vary over time.  Tasks may be registered at any time (using
23 > * methods {@link #register}, {@link #bulkRegister}, or forms of
24 > * constructors establishing initial numbers of parties), and
25 > * optionally deregistered upon any arrival (using {@link
26 > * #arriveAndDeregister}).  As is the case with most basic
27 > * synchronization constructs, registration and deregistration affect
28 > * only internal counts; they do not establish any further internal
29 > * bookkeeping, so tasks cannot query whether they are registered.
30 > * (However, you can introduce such bookkeeping by subclassing this
31 > * class.)
32 > *
33 > * <p> <b>Synchronization.</b> Like a {@code CyclicBarrier}, a {@code
34 > * Phaser} may be repeatedly awaited.  Method {@link
35 > * #arriveAndAwaitAdvance} has effect analogous to {@link
36 > * java.util.concurrent.CyclicBarrier#await CyclicBarrier.await}. Each
37 > * generation of a phaser has an associated phase number. The phase
38 > * number starts at zero, and advances when all parties arrive at the
39 > * phaser, wrapping around to zero after reaching {@code
40 > * Integer.MAX_VALUE}. The use of phase numbers enables independent
41 > * control of actions upon arrival at a phaser and upon awaiting
42 > * others, via two kinds of methods that may be invoked by any
43 > * registered party:
44   *
45   * <ul>
46   *
47 < * <li> The number of parties synchronizing on a phaser may vary over
48 < * time.  A task may register to be a party at any time, and may
49 < * deregister upon arriving at the barrier.  As is the case with most
50 < * basic synchronization constructs, registration and deregistration
51 < * affect only internal counts; they do not establish any further
52 < * internal bookkeeping, so tasks cannot query whether they are
53 < * registered. (However, you can introduce such bookkeeping in by
54 < * subclassing this class.)
55 < *
56 < * <li> Each generation has an associated phase value, starting at
57 < * zero, and advancing when all parties reach the barrier (wrapping
58 < * around to zero after reaching <tt>Integer.MAX_VALUE</tt>).
59 < *
60 < * <li> Like a CyclicBarrier, a Phaser may be repeatedly awaited.
61 < * Method <tt>arriveAndAwaitAdvance</tt> has effect analogous to
62 < * <tt>CyclicBarrier.await</tt>.  However, Phasers separate two
63 < * aspects of coordination, that may also be invoked independently:
64 < *
65 < * <ul>
47 > *   <li> <b>Arrival.</b> Methods {@link #arrive} and
48 > *       {@link #arriveAndDeregister} record arrival.  These methods
49 > *       do not block, but return an associated <em>arrival phase
50 > *       number</em>; that is, the phase number of the phaser to which
51 > *       the arrival applied. When the final party for a given phase
52 > *       arrives, an optional action is performed and the phase
53 > *       advances.  These actions are performed by the party
54 > *       triggering a phase advance, and are arranged by overriding
55 > *       method {@link #onAdvance(int, int)}, which also controls
56 > *       termination. Overriding this method is similar to, but more
57 > *       flexible than, providing a barrier action to a {@code
58 > *       CyclicBarrier}.
59 > *
60 > *   <li> <b>Waiting.</b> Method {@link #awaitAdvance} requires an
61 > *       argument indicating an arrival phase number, and returns when
62 > *       the phaser advances to (or is already at) a different phase.
63 > *       Unlike similar constructions using {@code CyclicBarrier},
64 > *       method {@code awaitAdvance} continues to wait even if the
65 > *       waiting thread is interrupted. Interruptible and timeout
66 > *       versions are also available, but exceptions encountered while
67 > *       tasks wait interruptibly or with timeout do not change the
68 > *       state of the phaser. If necessary, you can perform any
69 > *       associated recovery within handlers of those exceptions,
70 > *       often after invoking {@code forceTermination}.  Phasers may
71 > *       also be used by tasks executing in a {@link ForkJoinPool},
72 > *       which will ensure sufficient parallelism to execute tasks
73 > *       when others are blocked waiting for a phase to advance.
74   *
42 *   <li> Arriving at a barrier. Methods <tt>arrive</tt> and
43 *       <tt>arriveAndDeregister</tt> do not block, but return
44 *       the phase value current upon entry to the method.
45 *
46 *   <li> Awaiting others. Method <tt>awaitAdvance</tt> requires an
47 *       argument indicating the entry phase, and returns when the
48 *       barrier advances to a new phase.
75   * </ul>
76   *
77 < *
78 < * <li> Barrier actions, performed by the task triggering a phase
79 < * advance while others may be waiting, are arranged by overriding
80 < * method <tt>onAdvance</tt>, that also controls termination.
81 < * Overriding this method may be used to similar but more flexible
82 < * effect as providing a barrier action to a CyclicBarrier.
83 < *
84 < * <li> Phasers may enter a <em>termination</em> state in which all
85 < * await actions immediately return, indicating (via a negative phase
86 < * value) that execution is complete.  Termination is triggered by
87 < * executing the overridable <tt>onAdvance</tt> method that is invoked
88 < * each time the barrier is about to be tripped. When a Phaser is
89 < * controlling an action with a fixed number of iterations, it is
90 < * often convenient to override this method to cause termination when
91 < * the current phase number reaches a threshold. Method
92 < * <tt>forceTermination</tt> is also available to abruptly release
93 < * waiting threads and allow them to terminate.
94 < *
95 < * <li> Phasers may be tiered to reduce contention. Phasers with large
96 < * numbers of parties that would otherwise experience heavy
97 < * synchronization contention costs may instead be arranged in trees.
98 < * This will typically greatly increase throughput even though it
99 < * incurs somewhat greater per-operation overhead.
100 < *
101 < * <li> By default, <tt>awaitAdvance</tt> continues to wait even if
102 < * the waiting thread is interrupted. And unlike the case in
103 < * CyclicBarriers, exceptions encountered while tasks wait
104 < * interruptibly or with timeout do not change the state of the
105 < * barrier. If necessary, you can perform any associated recovery
106 < * within handlers of those exceptions, often after invoking
107 < * <tt>forceTermination</tt>.
108 < *
109 < * </ul>
77 > * <p> <b>Termination.</b> A phaser may enter a <em>termination</em>
78 > * state, that may be checked using method {@link #isTerminated}. Upon
79 > * termination, all synchronization methods immediately return without
80 > * waiting for advance, as indicated by a negative return value.
81 > * Similarly, attempts to register upon termination have no effect.
82 > * Termination is triggered when an invocation of {@code onAdvance}
83 > * returns {@code true}. The default implementation returns {@code
84 > * true} if a deregistration has caused the number of registered
85 > * parties to become zero.  As illustrated below, when phasers control
86 > * actions with a fixed number of iterations, it is often convenient
87 > * to override this method to cause termination when the current phase
88 > * number reaches a threshold. Method {@link #forceTermination} is
89 > * also available to abruptly release waiting threads and allow them
90 > * to terminate.
91 > *
92 > * <p> <b>Tiering.</b> Phasers may be <em>tiered</em> (i.e.,
93 > * constructed in tree structures) to reduce contention. Phasers with
94 > * large numbers of parties that would otherwise experience heavy
95 > * synchronization contention costs may instead be set up so that
96 > * groups of sub-phasers share a common parent.  This may greatly
97 > * increase throughput even though it incurs greater per-operation
98 > * overhead.
99 > *
100 > * <p>In a tree of tiered phasers, registration and deregistration of
101 > * child phasers with their parent are managed automatically.
102 > * Whenever the number of registered parties of a child phaser becomes
103 > * non-zero (as established in the {@link #Phaser(Phaser,int)}
104 > * constructor, {@link #register}, or {@link #bulkRegister}), the
105 > * child phaser is registered with its parent.  Whenever the number of
106 > * registered parties becomes zero as the result of an invocation of
107 > * {@link #arriveAndDeregister}, the child phaser is deregistered
108 > * from its parent.
109 > *
110 > * <p><b>Monitoring.</b> While synchronization methods may be invoked
111 > * only by registered parties, the current state of a phaser may be
112 > * monitored by any caller.  At any given moment there are {@link
113 > * #getRegisteredParties} parties in total, of which {@link
114 > * #getArrivedParties} have arrived at the current phase ({@link
115 > * #getPhase}).  When the remaining ({@link #getUnarrivedParties})
116 > * parties arrive, the phase advances.  The values returned by these
117 > * methods may reflect transient states and so are not in general
118 > * useful for synchronization control.  Method {@link #toString}
119 > * returns snapshots of these state queries in a form convenient for
120 > * informal monitoring.
121   *
122   * <p><b>Sample usages:</b>
123   *
124 < * <p>A Phaser may be used instead of a <tt>CountdownLatch</tt> to control
125 < * a one-shot action serving a variable number of parties. The typical
126 < * idiom is for the method setting this up to first register, then
127 < * start the actions, then deregister, as in:
128 < *
129 < * <pre>
130 < *  void runTasks(List&lt;Runnable&gt; list) {
131 < *    final Phaser phaser = new Phaser(1); // "1" to register self
132 < *    for (Runnable r : list) {
133 < *      phaser.register();
134 < *      new Thread() {
135 < *        public void run() {
136 < *          phaser.arriveAndAwaitAdvance(); // await all creation
137 < *          r.run();
138 < *          phaser.arriveAndDeregister();   // signal completion
139 < *        }
140 < *      }.start();
124 > * <p>A {@code Phaser} may be used instead of a {@code CountDownLatch}
125 > * to control a one-shot action serving a variable number of parties.
126 > * The typical idiom is for the method setting this up to first
127 > * register, then start the actions, then deregister, as in:
128 > *
129 > *  <pre> {@code
130 > * void runTasks(List<Runnable> tasks) {
131 > *   final Phaser phaser = new Phaser(1); // "1" to register self
132 > *   // create and start threads
133 > *   for (final Runnable task : tasks) {
134 > *     phaser.register();
135 > *     new Thread() {
136 > *       public void run() {
137 > *         phaser.arriveAndAwaitAdvance(); // await all creation
138 > *         task.run();
139 > *       }
140 > *     }.start();
141   *   }
142   *
143 < *   doSomethingOnBehalfOfWorkers();
144 < *   phaser.arrive(); // allow threads to start
145 < *   int p = phaser.arriveAndDeregister(); // deregister self  ...
109 < *   p = phaser.awaitAdvance(p); // ... and await arrival
110 < *   otherActions(); // do other things while tasks execute
111 < *   phaser.awaitAdvance(p); // awit final completion
112 < * }
113 < * </pre>
143 > *   // allow threads to start and deregister self
144 > *   phaser.arriveAndDeregister();
145 > * }}</pre>
146   *
147   * <p>One way to cause a set of threads to repeatedly perform actions
148 < * for a given number of iterations is to override <tt>onAdvance</tt>:
148 > * for a given number of iterations is to override {@code onAdvance}:
149   *
150 < * <pre>
151 < *  void startTasks(List&lt;Runnable&gt; list, final int iterations) {
152 < *    final Phaser phaser = new Phaser() {
153 < *       public boolean onAdvance(int phase, int registeredParties) {
154 < *         return phase &gt;= iterations || registeredParties == 0;
150 > *  <pre> {@code
151 > * void startTasks(List<Runnable> tasks, final int iterations) {
152 > *   final Phaser phaser = new Phaser() {
153 > *     protected boolean onAdvance(int phase, int registeredParties) {
154 > *       return phase >= iterations || registeredParties == 0;
155 > *     }
156 > *   };
157 > *   phaser.register();
158 > *   for (final Runnable task : tasks) {
159 > *     phaser.register();
160 > *     new Thread() {
161 > *       public void run() {
162 > *         do {
163 > *           task.run();
164 > *           phaser.arriveAndAwaitAdvance();
165 > *         } while (!phaser.isTerminated());
166   *       }
167 < *    };
125 < *    phaser.register();
126 < *    for (Runnable r : list) {
127 < *      phaser.register();
128 < *      new Thread() {
129 < *        public void run() {
130 < *           do {
131 < *             r.run();
132 < *             phaser.arriveAndAwaitAdvance();
133 < *           } while(!phaser.isTerminated();
134 < *        }
135 < *      }.start();
167 > *     }.start();
168   *   }
169   *   phaser.arriveAndDeregister(); // deregister self, don't wait
170 < * }
171 < * </pre>
170 > * }}</pre>
171 > *
172 > * If the main task must later await termination, it
173 > * may re-register and then execute a similar loop:
174 > *  <pre> {@code
175 > *   // ...
176 > *   phaser.register();
177 > *   while (!phaser.isTerminated())
178 > *     phaser.arriveAndAwaitAdvance();}</pre>
179 > *
180 > * <p>Related constructions may be used to await particular phase numbers
181 > * in contexts where you are sure that the phase will never wrap around
182 > * {@code Integer.MAX_VALUE}. For example:
183 > *
184 > *  <pre> {@code
185 > * void awaitPhase(Phaser phaser, int phase) {
186 > *   int p = phaser.register(); // assumes caller not already registered
187 > *   while (p < phase) {
188 > *     if (phaser.isTerminated())
189 > *       // ... deal with unexpected termination
190 > *     else
191 > *       p = phaser.arriveAndAwaitAdvance();
192 > *   }
193 > *   phaser.arriveAndDeregister();
194 > * }}</pre>
195   *
141 * <p> To create a set of tasks using a tree of Phasers,
142 * you could use code of the following form, assuming a
143 * Task class with a constructor accepting a Phaser that
144 * it registers for upon construction:
145 * <pre>
146 *  void build(Task[] actions, int lo, int hi, Phaser b) {
147 *    int step = (hi - lo) / TASKS_PER_PHASER;
148 *    if (step &gt; 1) {
149 *       int i = lo;
150 *       while (i &lt; hi) {
151 *         int r = Math.min(i + step, hi);
152 *         build(actions, i, r, new Phaser(b));
153 *         i = r;
154 *       }
155 *    }
156 *    else {
157 *      for (int i = lo; i &lt; hi; ++i)
158 *        actions[i] = new Task(b);
159 *        // assumes new Task(b) performs b.register()
160 *    }
161 *  }
162 *  // .. initially called, for n tasks via
163 *  build(new Task[n], 0, n, new Phaser());
164 * </pre>
165 *
166 * The best value of <tt>TASKS_PER_PHASER</tt> depends mainly on
167 * expected barrier synchronization rates. A value as low as four may
168 * be appropriate for extremely small per-barrier task bodies (thus
169 * high rates), or up to hundreds for extremely large ones.
196   *
197 < * </pre>
197 > * <p>To create a set of {@code n} tasks using a tree of phasers, you
198 > * could use code of the following form, assuming a Task class with a
199 > * constructor accepting a {@code Phaser} that it registers with upon
200 > * construction. After invocation of {@code build(new Task[n], 0, n,
201 > * new Phaser())}, these tasks could then be started, for example by
202 > * submitting to a pool:
203 > *
204 > *  <pre> {@code
205 > * void build(Task[] tasks, int lo, int hi, Phaser ph) {
206 > *   if (hi - lo > TASKS_PER_PHASER) {
207 > *     for (int i = lo; i < hi; i += TASKS_PER_PHASER) {
208 > *       int j = Math.min(i + TASKS_PER_PHASER, hi);
209 > *       build(tasks, i, j, new Phaser(ph));
210 > *     }
211 > *   } else {
212 > *     for (int i = lo; i < hi; ++i)
213 > *       tasks[i] = new Task(ph);
214 > *       // assumes new Task(ph) performs ph.register()
215 > *   }
216 > * }}</pre>
217 > *
218 > * The best value of {@code TASKS_PER_PHASER} depends mainly on
219 > * expected synchronization rates. A value as low as four may
220 > * be appropriate for extremely small per-phase task bodies (thus
221 > * high rates), or up to hundreds for extremely large ones.
222   *
223   * <p><b>Implementation notes</b>: This implementation restricts the
224   * maximum number of parties to 65535. Attempts to register additional
225 < * parties result in IllegalStateExceptions. However, you can and
225 > * parties result in {@code IllegalStateException}. However, you can and
226   * should create tiered phasers to accommodate arbitrarily large sets
227   * of participants.
228 + *
229 + * @since 1.7
230 + * @author Doug Lea
231   */
232   public class Phaser {
233      /*
# Line 184 | Line 237 | public class Phaser {
237       */
238  
239      /**
240 <     * Barrier state representation. Conceptually, a barrier contains
241 <     * four values:
240 >     * Primary state representation, holding four bit-fields:
241 >     *
242 >     * unarrived  -- the number of parties yet to hit barrier (bits  0-15)
243 >     * parties    -- the number of parties to wait            (bits 16-31)
244 >     * phase      -- the generation of the barrier            (bits 32-62)
245 >     * terminated -- set if barrier is terminated             (bit  63 / sign)
246 >     *
247 >     * Except that a phaser with no registered parties is
248 >     * distinguished by the otherwise illegal state of having zero
249 >     * parties and one unarrived parties (encoded as EMPTY below).
250 >     *
251 >     * To efficiently maintain atomicity, these values are packed into
252 >     * a single (atomic) long. Good performance relies on keeping
253 >     * state decoding and encoding simple, and keeping race windows
254 >     * short.
255       *
256 <     * * parties -- the number of parties to wait (16 bits)
257 <     * * unarrived -- the number of parties yet to hit barrier (16 bits)
258 <     * * phase -- the generation of the barrier (31 bits)
259 <     * * terminated -- set if barrier is terminated (1 bit)
260 <     *
195 <     * However, to efficiently maintain atomicity, these values are
196 <     * packed into a single (atomic) long. Termination uses the sign
197 <     * bit of 32 bit representation of phase, so phase is set to -1 on
198 <     * termination. Good performace relies on keeping state decoding
199 <     * and encoding simple, and keeping race windows short.
256 >     * All state updates are performed via CAS except initial
257 >     * registration of a sub-phaser (i.e., one with a non-null
258 >     * parent).  In this (relatively rare) case, we use built-in
259 >     * synchronization to lock while first registering with its
260 >     * parent.
261       *
262 <     * Note: there are some cheats in arrive() that rely on unarrived
263 <     * being lowest 16 bits.
262 >     * The phase of a subphaser is allowed to lag that of its
263 >     * ancestors until it is actually accessed -- see method
264 >     * reconcileState.
265       */
266      private volatile long state;
267  
268 <    private static final int ushortBits = 16;
269 <    private static final int ushortMask =  (1 << ushortBits) - 1;
270 <    private static final int phaseMask = 0x7fffffff;
268 >    private static final int  MAX_PARTIES     = 0xffff;
269 >    private static final int  MAX_PHASE       = Integer.MAX_VALUE;
270 >    private static final int  PARTIES_SHIFT   = 16;
271 >    private static final int  PHASE_SHIFT     = 32;
272 >    private static final int  UNARRIVED_MASK  = 0xffff;      // to mask ints
273 >    private static final long PARTIES_MASK    = 0xffff0000L; // to mask longs
274 >    private static final long TERMINATION_BIT = 1L << 63;
275 >
276 >    // some special values
277 >    private static final int  ONE_ARRIVAL     = 1;
278 >    private static final int  ONE_PARTY       = 1 << PARTIES_SHIFT;
279 >    private static final int  EMPTY           = 1;
280 >
281 >    // The following unpacking methods are usually manually inlined
282  
283      private static int unarrivedOf(long s) {
284 <        return (int)(s & ushortMask);
284 >        int counts = (int)s;
285 >        return (counts == EMPTY) ? 0 : counts & UNARRIVED_MASK;
286      }
287  
288      private static int partiesOf(long s) {
289 <        return (int)(s & (ushortMask << 16)) >>> 16;
289 >        return (int)s >>> PARTIES_SHIFT;
290      }
291  
292      private static int phaseOf(long s) {
293 <        return (int)(s >>> 32);
293 >        return (int)(s >>> PHASE_SHIFT);
294      }
295  
296      private static int arrivedOf(long s) {
297 <        return partiesOf(s) - unarrivedOf(s);
298 <    }
299 <
226 <    private static long stateFor(int phase, int parties, int unarrived) {
227 <        return (((long)phase) << 32) | ((parties << 16) | unarrived);
228 <    }
229 <
230 <    private static long trippedStateFor(int phase, int parties) {
231 <        return (((long)phase) << 32) | ((parties << 16) | parties);
232 <    }
233 <
234 <    private static IllegalStateException badBounds(int parties, int unarrived) {
235 <        return new IllegalStateException
236 <            ("Attempt to set " + unarrived +
237 <             " unarrived of " + parties + " parties");
297 >        int counts = (int)s;
298 >        return (counts == EMPTY) ? 0 :
299 >            (counts >>> PARTIES_SHIFT) - (counts & UNARRIVED_MASK);
300      }
301  
302      /**
# Line 243 | Line 305 | public class Phaser {
305      private final Phaser parent;
306  
307      /**
308 <     * The root of Phaser tree. Equals this if not in a tree.  Used to
247 <     * support faster state push-down.
308 >     * The root of phaser tree. Equals this if not in a tree.
309       */
310      private final Phaser root;
311  
251    // Wait queues
252
312      /**
313 <     * Heads of Treiber stacks waiting for nonFJ threads. To eliminate
314 <     * contention while releasing some threads while adding others, we
313 >     * Heads of Treiber stacks for waiting threads. To eliminate
314 >     * contention when releasing some threads while adding others, we
315       * use two of them, alternating across even and odd phases.
316 +     * Subphasers share queues with root to speed up releases.
317       */
318 <    private final AtomicReference<QNode> evenQ = new AtomicReference<QNode>();
319 <    private final AtomicReference<QNode> oddQ  = new AtomicReference<QNode>();
318 >    private final AtomicReference<QNode> evenQ;
319 >    private final AtomicReference<QNode> oddQ;
320  
321      private AtomicReference<QNode> queueFor(int phase) {
322 <        return (phase & 1) == 0? evenQ : oddQ;
322 >        return ((phase & 1) == 0) ? evenQ : oddQ;
323      }
324  
325      /**
326 <     * Returns current state, first resolving lagged propagation from
267 <     * root if necessary.
326 >     * Returns message string for bounds exceptions on arrival.
327       */
328 <    private long getReconciledState() {
329 <        return parent == null? state : reconcileState();
328 >    private String badArrive(long s) {
329 >        return "Attempted arrival of unregistered party for " +
330 >            stateToString(s);
331      }
332  
333      /**
334 <     * Recursively resolves state.
334 >     * Returns message string for bounds exceptions on registration.
335       */
336 <    private long reconcileState() {
337 <        Phaser p = parent;
338 <        long s = state;
339 <        if (p != null) {
340 <            while (unarrivedOf(s) == 0 && phaseOf(s) != phaseOf(root.state)) {
341 <                long parentState = p.getReconciledState();
342 <                int parentPhase = phaseOf(parentState);
343 <                int phase = phaseOf(s = state);
344 <                if (phase != parentPhase) {
345 <                    long next = trippedStateFor(parentPhase, partiesOf(s));
346 <                    if (casState(s, next)) {
347 <                        releaseWaiters(phase);
348 <                        s = next;
336 >    private String badRegister(long s) {
337 >        return "Attempt to register more than " +
338 >            MAX_PARTIES + " parties for " + stateToString(s);
339 >    }
340 >
341 >    /**
342 >     * Main implementation for methods arrive and arriveAndDeregister.
343 >     * Manually tuned to speed up and minimize race windows for the
344 >     * common case of just decrementing unarrived field.
345 >     *
346 >     * @param deregister false for arrive, true for arriveAndDeregister
347 >     */
348 >    private int doArrive(boolean deregister) {
349 >        int adj = deregister ? ONE_ARRIVAL|ONE_PARTY : ONE_ARRIVAL;
350 >        final Phaser root = this.root;
351 >        for (;;) {
352 >            long s = (root == this) ? state : reconcileState();
353 >            int phase = (int)(s >>> PHASE_SHIFT);
354 >            int counts = (int)s;
355 >            int unarrived = (counts & UNARRIVED_MASK) - 1;
356 >            if (phase < 0)
357 >                return phase;
358 >            else if (counts == EMPTY || unarrived < 0) {
359 >                if (root == this || reconcileState() == s)
360 >                    throw new IllegalStateException(badArrive(s));
361 >            }
362 >            else if (UNSAFE.compareAndSwapLong(this, stateOffset, s, s-=adj)) {
363 >                long n = s & PARTIES_MASK;  // base of next state
364 >                int nextUnarrived = (int)n >>> PARTIES_SHIFT;
365 >                if (unarrived == 0) {
366 >                    if (root == this) {
367 >                        if (onAdvance(phase, nextUnarrived))
368 >                            n |= TERMINATION_BIT;
369 >                        else if (nextUnarrived == 0)
370 >                            n |= EMPTY;
371 >                        else
372 >                            n |= nextUnarrived;
373 >                        n |= (long)((phase + 1) & MAX_PHASE) << PHASE_SHIFT;
374 >                        UNSAFE.compareAndSwapLong(this, stateOffset, s, n);
375                      }
376 +                    else if (nextUnarrived == 0) { // propagate deregistration
377 +                        phase = parent.doArrive(true);
378 +                        UNSAFE.compareAndSwapLong(this, stateOffset,
379 +                                                  s, s | EMPTY);
380 +                    }
381 +                    else
382 +                        phase = parent.doArrive(false);
383 +                    releaseWaiters(phase);
384                  }
385 +                return phase;
386              }
387          }
388 +    }
389 +
390 +    /**
391 +     * Implementation of register, bulkRegister
392 +     *
393 +     * @param registrations number to add to both parties and
394 +     * unarrived fields. Must be greater than zero.
395 +     */
396 +    private int doRegister(int registrations) {
397 +        // adjustment to state
398 +        long adj = ((long)registrations << PARTIES_SHIFT) | registrations;
399 +        final Phaser parent = this.parent;
400 +        int phase;
401 +        for (;;) {
402 +            long s = (parent == null) ? state : reconcileState();
403 +            int counts = (int)s;
404 +            int parties = counts >>> PARTIES_SHIFT;
405 +            int unarrived = counts & UNARRIVED_MASK;
406 +            if (registrations > MAX_PARTIES - parties)
407 +                throw new IllegalStateException(badRegister(s));
408 +            else if ((phase = (int)(s >>> PHASE_SHIFT)) < 0)
409 +                break;
410 +            else if (counts != EMPTY) {             // not 1st registration
411 +                if (parent == null || reconcileState() == s) {
412 +                    if (unarrived == 0)             // wait out advance
413 +                        root.internalAwaitAdvance(phase, null);
414 +                    else if (UNSAFE.compareAndSwapLong(this, stateOffset,
415 +                                                       s, s + adj))
416 +                        break;
417 +                }
418 +            }
419 +            else if (parent == null) {              // 1st root registration
420 +                long next = ((long)phase << PHASE_SHIFT) | adj;
421 +                if (UNSAFE.compareAndSwapLong(this, stateOffset, s, next))
422 +                    break;
423 +            }
424 +            else {
425 +                synchronized (this) {               // 1st sub registration
426 +                    if (state == s) {               // recheck under lock
427 +                        parent.doRegister(1);
428 +                        do {                        // force current phase
429 +                            phase = (int)(root.state >>> PHASE_SHIFT);
430 +                            // assert phase < 0 || (int)state == EMPTY;
431 +                        } while (!UNSAFE.compareAndSwapLong
432 +                                 (this, stateOffset, state,
433 +                                  ((long)phase << PHASE_SHIFT) | adj));
434 +                        break;
435 +                    }
436 +                }
437 +            }
438 +        }
439 +        return phase;
440 +    }
441 +
442 +    /**
443 +     * Resolves lagged phase propagation from root if necessary.
444 +     * Reconciliation normally occurs when root has advanced but
445 +     * subphasers have not yet done so, in which case they must finish
446 +     * their own advance by setting unarrived to parties (or if
447 +     * parties is zero, resetting to unregistered EMPTY state).
448 +     * However, this method may also be called when "floating"
449 +     * subphasers with possibly some unarrived parties are merely
450 +     * catching up to current phase, in which case counts are
451 +     * unaffected.
452 +     *
453 +     * @return reconciled state
454 +     */
455 +    private long reconcileState() {
456 +        final Phaser root = this.root;
457 +        long s = state;
458 +        if (root != this) {
459 +            int phase, u, p;
460 +            // CAS root phase with current parties; possibly trip unarrived
461 +            while ((phase = (int)(root.state >>> PHASE_SHIFT)) !=
462 +                   (int)(s >>> PHASE_SHIFT) &&
463 +                   !UNSAFE.compareAndSwapLong
464 +                   (this, stateOffset, s,
465 +                    s = (((long)phase << PHASE_SHIFT) |
466 +                         (s & PARTIES_MASK) |
467 +                         ((p = (int)s >>> PARTIES_SHIFT) == 0 ? EMPTY :
468 +                          ((u = (int)s & UNARRIVED_MASK) == 0 && phase >= 0) ?
469 +                          p : u))))
470 +                s = state;
471 +        }
472          return s;
473      }
474  
475      /**
476 <     * Creates a new Phaser without any initially registered parties,
477 <     * initial phase number 0, and no parent.
476 >     * Creates a new phaser with no initially registered parties, no
477 >     * parent, and initial phase number 0. Any thread using this
478 >     * phaser will need to first register for it.
479       */
480      public Phaser() {
481 <        this(null);
481 >        this(null, 0);
482      }
483  
484      /**
485 <     * Creates a new Phaser with the given numbers of registered
486 <     * unarrived parties, initial phase number 0, and no parent.
487 <     * @param parties the number of parties required to trip barrier.
485 >     * Creates a new phaser with the given number of registered
486 >     * unarrived parties, no parent, and initial phase number 0.
487 >     *
488 >     * @param parties the number of parties required to advance to the
489 >     * next phase
490       * @throws IllegalArgumentException if parties less than zero
491 <     * or greater than the maximum number of parties supported.
491 >     * or greater than the maximum number of parties supported
492       */
493      public Phaser(int parties) {
494          this(null, parties);
495      }
496  
497      /**
498 <     * Creates a new Phaser with the given parent, without any
499 <     * initially registered parties. If parent is non-null this phaser
500 <     * is registered with the parent and its initial phase number is
319 <     * the same as that of parent phaser.
320 <     * @param parent the parent phaser.
498 >     * Equivalent to {@link #Phaser(Phaser, int) Phaser(parent, 0)}.
499 >     *
500 >     * @param parent the parent phaser
501       */
502      public Phaser(Phaser parent) {
503 <        int phase = 0;
324 <        this.parent = parent;
325 <        if (parent != null) {
326 <            this.root = parent.root;
327 <            phase = parent.register();
328 <        }
329 <        else
330 <            this.root = this;
331 <        this.state = trippedStateFor(phase, 0);
503 >        this(parent, 0);
504      }
505  
506      /**
507 <     * Creates a new Phaser with the given parent and numbers of
508 <     * registered unarrived parties. If parent is non-null this phaser
509 <     * is registered with the parent and its initial phase number is
510 <     * the same as that of parent phaser.
511 <     * @param parent the parent phaser.
512 <     * @param parties the number of parties required to trip barrier.
507 >     * Creates a new phaser with the given parent and number of
508 >     * registered unarrived parties.  When the given parent is non-null
509 >     * and the given number of parties is greater than zero, this
510 >     * child phaser is registered with its parent.
511 >     *
512 >     * @param parent the parent phaser
513 >     * @param parties the number of parties required to advance to the
514 >     * next phase
515       * @throws IllegalArgumentException if parties less than zero
516 <     * or greater than the maximum number of parties supported.
516 >     * or greater than the maximum number of parties supported
517       */
518      public Phaser(Phaser parent, int parties) {
519 <        if (parties < 0 || parties > ushortMask)
519 >        if (parties >>> PARTIES_SHIFT != 0)
520              throw new IllegalArgumentException("Illegal number of parties");
521          int phase = 0;
522          this.parent = parent;
523          if (parent != null) {
524 <            this.root = parent.root;
525 <            phase = parent.register();
524 >            final Phaser root = parent.root;
525 >            this.root = root;
526 >            this.evenQ = root.evenQ;
527 >            this.oddQ = root.oddQ;
528 >            if (parties != 0)
529 >                phase = parent.doRegister(1);
530          }
531 <        else
531 >        else {
532              this.root = this;
533 <        this.state = trippedStateFor(phase, parties);
533 >            this.evenQ = new AtomicReference<QNode>();
534 >            this.oddQ = new AtomicReference<QNode>();
535 >        }
536 >        this.state = (parties == 0) ? (long)EMPTY :
537 >            ((long)phase << PHASE_SHIFT) |
538 >            ((long)parties << PARTIES_SHIFT) |
539 >            ((long)parties);
540      }
541  
542      /**
543 <     * Adds a new unarrived party to this phaser.
544 <     * @return the current barrier phase number upon registration
543 >     * Adds a new unarrived party to this phaser.  If an ongoing
544 >     * invocation of {@link #onAdvance} is in progress, this method
545 >     * may await its completion before returning.  If this phaser has
546 >     * a parent, and this phaser previously had no registered parties,
547 >     * this child phaser is also registered with its parent. If
548 >     * this phaser is terminated, the attempt to register has
549 >     * no effect, and a negative value is returned.
550 >     *
551 >     * @return the arrival phase number to which this registration
552 >     * applied.  If this value is negative, then this phaser has
553 >     * terminated, in which case registration has no effect.
554       * @throws IllegalStateException if attempting to register more
555 <     * than the maximum supported number of parties.
555 >     * than the maximum supported number of parties
556       */
557      public int register() {
558          return doRegister(1);
# Line 367 | Line 560 | public class Phaser {
560  
561      /**
562       * Adds the given number of new unarrived parties to this phaser.
563 <     * @param parties the number of parties required to trip barrier.
564 <     * @return the current barrier phase number upon registration
563 >     * If an ongoing invocation of {@link #onAdvance} is in progress,
564 >     * this method may await its completion before returning.  If this
565 >     * phaser has a parent, and the given number of parties is greater
566 >     * than zero, and this phaser previously had no registered
567 >     * parties, this child phaser is also registered with its parent.
568 >     * If this phaser is terminated, the attempt to register has no
569 >     * effect, and a negative value is returned.
570 >     *
571 >     * @param parties the number of additional parties required to
572 >     * advance to the next phase
573 >     * @return the arrival phase number to which this registration
574 >     * applied.  If this value is negative, then this phaser has
575 >     * terminated, in which case registration has no effect.
576       * @throws IllegalStateException if attempting to register more
577 <     * than the maximum supported number of parties.
577 >     * than the maximum supported number of parties
578 >     * @throws IllegalArgumentException if {@code parties < 0}
579       */
580      public int bulkRegister(int parties) {
581          if (parties < 0)
# Line 381 | Line 586 | public class Phaser {
586      }
587  
588      /**
589 <     * Shared code for register, bulkRegister
590 <     */
591 <    private int doRegister(int registrations) {
592 <        int phase;
593 <        for (;;) {
594 <            long s = getReconciledState();
390 <            phase = phaseOf(s);
391 <            int unarrived = unarrivedOf(s) + registrations;
392 <            int parties = partiesOf(s) + registrations;
393 <            if (phase < 0)
394 <                break;
395 <            if (parties > ushortMask || unarrived > ushortMask)
396 <                throw badBounds(parties, unarrived);
397 <            if (phase == phaseOf(root.state) &&
398 <                casState(s, stateFor(phase, parties, unarrived)))
399 <                break;
400 <        }
401 <        return phase;
402 <    }
403 <
404 <    /**
405 <     * Arrives at the barrier, but does not wait for others.  (You can
406 <     * in turn wait for others via {@link #awaitAdvance}).
589 >     * Arrives at this phaser, without waiting for others to arrive.
590 >     *
591 >     * <p>It is a usage error for an unregistered party to invoke this
592 >     * method.  However, this error may result in an {@code
593 >     * IllegalStateException} only upon some subsequent operation on
594 >     * this phaser, if ever.
595       *
596 <     * @return the barrier phase number upon entry to this method, or a
409 <     * negative value if terminated;
596 >     * @return the arrival phase number, or a negative value if terminated
597       * @throws IllegalStateException if not terminated and the number
598 <     * of unarrived parties would become negative.
598 >     * of unarrived parties would become negative
599       */
600      public int arrive() {
601 <        int phase;
415 <        for (;;) {
416 <            long s = state;
417 <            phase = phaseOf(s);
418 <            int parties = partiesOf(s);
419 <            int unarrived = unarrivedOf(s) - 1;
420 <            if (unarrived > 0) {        // Not the last arrival
421 <                if (casState(s, s - 1)) // s-1 adds one arrival
422 <                    break;
423 <            }
424 <            else if (unarrived == 0) {  // the last arrival
425 <                Phaser par = parent;
426 <                if (par == null) {      // directly trip
427 <                    if (casState
428 <                        (s,
429 <                         trippedStateFor(onAdvance(phase, parties)? -1 :
430 <                                         ((phase + 1) & phaseMask), parties))) {
431 <                        releaseWaiters(phase);
432 <                        break;
433 <                    }
434 <                }
435 <                else {                  // cascade to parent
436 <                    if (casState(s, s - 1)) { // zeroes unarrived
437 <                        par.arrive();
438 <                        reconcileState();
439 <                        break;
440 <                    }
441 <                }
442 <            }
443 <            else if (phase < 0) // Don't throw exception if terminated
444 <                break;
445 <            else if (phase != phaseOf(root.state)) // or if unreconciled
446 <                reconcileState();
447 <            else
448 <                throw badBounds(parties, unarrived);
449 <        }
450 <        return phase;
601 >        return doArrive(false);
602      }
603  
604      /**
605 <     * Arrives at the barrier, and deregisters from it, without
606 <     * waiting for others. Deregistration reduces number of parties
607 <     * required to trip the barrier in future phases.  If this phaser
605 >     * Arrives at this phaser and deregisters from it without waiting
606 >     * for others to arrive. Deregistration reduces the number of
607 >     * parties required to advance in future phases.  If this phaser
608       * has a parent, and deregistration causes this phaser to have
609       * zero parties, this phaser is also deregistered from its parent.
610       *
611 <     * @return the current barrier phase number upon entry to
612 <     * this method, or a negative value if terminated;
611 >     * <p>It is a usage error for an unregistered party to invoke this
612 >     * method.  However, this error may result in an {@code
613 >     * IllegalStateException} only upon some subsequent operation on
614 >     * this phaser, if ever.
615 >     *
616 >     * @return the arrival phase number, or a negative value if terminated
617       * @throws IllegalStateException if not terminated and the number
618 <     * of registered or unarrived parties would become negative.
618 >     * of registered or unarrived parties would become negative
619       */
620      public int arriveAndDeregister() {
621 <        // similar code to arrive, but too different to merge
467 <        Phaser par = parent;
468 <        int phase;
469 <        for (;;) {
470 <            long s = state;
471 <            phase = phaseOf(s);
472 <            int parties = partiesOf(s) - 1;
473 <            int unarrived = unarrivedOf(s) - 1;
474 <            if (parties >= 0) {
475 <                if (unarrived > 0 || (unarrived == 0 && par != null)) {
476 <                    if (casState
477 <                        (s,
478 <                         stateFor(phase, parties, unarrived))) {
479 <                        if (unarrived == 0) {
480 <                            par.arriveAndDeregister();
481 <                            reconcileState();
482 <                        }
483 <                        break;
484 <                    }
485 <                    continue;
486 <                }
487 <                if (unarrived == 0) {
488 <                    if (casState
489 <                        (s,
490 <                         trippedStateFor(onAdvance(phase, parties)? -1 :
491 <                                         ((phase + 1) & phaseMask), parties))) {
492 <                        releaseWaiters(phase);
493 <                        break;
494 <                    }
495 <                    continue;
496 <                }
497 <                if (phase < 0)
498 <                    break;
499 <                if (par != null && phase != phaseOf(root.state)) {
500 <                    reconcileState();
501 <                    continue;
502 <                }
503 <            }
504 <            throw badBounds(parties, unarrived);
505 <        }
506 <        return phase;
621 >        return doArrive(true);
622      }
623  
624      /**
625 <     * Arrives at the barrier and awaits others. Equivalent in effect
626 <     * to <tt>awaitAdvance(arrive())</tt>.  If you instead need to
627 <     * await with interruption of timeout, and/or deregister upon
628 <     * arrival, you can arrange them using analogous constructions.
629 <     * @return the phase on entry to this method
625 >     * Arrives at this phaser and awaits others. Equivalent in effect
626 >     * to {@code awaitAdvance(arrive())}.  If you need to await with
627 >     * interruption or timeout, you can arrange this with an analogous
628 >     * construction using one of the other forms of the {@code
629 >     * awaitAdvance} method.  If instead you need to deregister upon
630 >     * arrival, use {@code awaitAdvance(arriveAndDeregister())}.
631 >     *
632 >     * <p>It is a usage error for an unregistered party to invoke this
633 >     * method.  However, this error may result in an {@code
634 >     * IllegalStateException} only upon some subsequent operation on
635 >     * this phaser, if ever.
636 >     *
637 >     * @return the arrival phase number, or the (negative)
638 >     * {@linkplain #getPhase() current phase} if terminated
639       * @throws IllegalStateException if not terminated and the number
640 <     * of unarrived parties would become negative.
640 >     * of unarrived parties would become negative
641       */
642      public int arriveAndAwaitAdvance() {
643 <        return awaitAdvance(arrive());
643 >        // Specialization of doArrive+awaitAdvance eliminating some reads/paths
644 >        final Phaser root = this.root;
645 >        for (;;) {
646 >            long s = (root == this) ? state : reconcileState();
647 >            int phase = (int)(s >>> PHASE_SHIFT);
648 >            int counts = (int)s;
649 >            int unarrived = (counts & UNARRIVED_MASK) - 1;
650 >            if (phase < 0)
651 >                return phase;
652 >            else if (counts == EMPTY || unarrived < 0) {
653 >                if (reconcileState() == s)
654 >                    throw new IllegalStateException(badArrive(s));
655 >            }
656 >            else if (UNSAFE.compareAndSwapLong(this, stateOffset, s,
657 >                                               s -= ONE_ARRIVAL)) {
658 >                if (unarrived != 0)
659 >                    return root.internalAwaitAdvance(phase, null);
660 >                if (root != this)
661 >                    return parent.arriveAndAwaitAdvance();
662 >                long n = s & PARTIES_MASK;  // base of next state
663 >                int nextUnarrived = (int)n >>> PARTIES_SHIFT;
664 >                if (onAdvance(phase, nextUnarrived))
665 >                    n |= TERMINATION_BIT;
666 >                else if (nextUnarrived == 0)
667 >                    n |= EMPTY;
668 >                else
669 >                    n |= nextUnarrived;
670 >                int nextPhase = (phase + 1) & MAX_PHASE;
671 >                n |= (long)nextPhase << PHASE_SHIFT;
672 >                if (!UNSAFE.compareAndSwapLong(this, stateOffset, s, n))
673 >                    return (int)(state >>> PHASE_SHIFT); // terminated
674 >                releaseWaiters(phase);
675 >                return nextPhase;
676 >            }
677 >        }
678      }
679  
680      /**
681 <     * Awaits the phase of the barrier to advance from the given
682 <     * value, or returns immediately if argument is negative or this
683 <     * barrier is terminated.
684 <     * @param phase the phase on entry to this method
685 <     * @return the phase on exit from this method
681 >     * Awaits the phase of this phaser to advance from the given phase
682 >     * value, returning immediately if the current phase is not equal
683 >     * to the given phase value or this phaser is terminated.
684 >     *
685 >     * @param phase an arrival phase number, or negative value if
686 >     * terminated; this argument is normally the value returned by a
687 >     * previous call to {@code arrive} or {@code arriveAndDeregister}.
688 >     * @return the next arrival phase number, or the argument if it is
689 >     * negative, or the (negative) {@linkplain #getPhase() current phase}
690 >     * if terminated
691       */
692      public int awaitAdvance(int phase) {
693 +        final Phaser root = this.root;
694 +        long s = (root == this) ? state : reconcileState();
695 +        int p = (int)(s >>> PHASE_SHIFT);
696          if (phase < 0)
697              return phase;
698 <        long s = getReconciledState();
699 <        int p = phaseOf(s);
700 <        if (p != phase)
535 <            return p;
536 <        if (unarrivedOf(s) == 0)
537 <            parent.awaitAdvance(phase);
538 <        // Fall here even if parent waited, to reconcile and help release
539 <        return untimedWait(phase);
698 >        if (p == phase)
699 >            return root.internalAwaitAdvance(phase, null);
700 >        return p;
701      }
702  
703      /**
704 <     * Awaits the phase of the barrier to advance from the given
705 <     * value, or returns immediately if argumet is negative or this
706 <     * barrier is terminated, or throws InterruptedException if
707 <     * interrupted while waiting.
708 <     * @param phase the phase on entry to this method
709 <     * @return the phase on exit from this method
704 >     * Awaits the phase of this phaser to advance from the given phase
705 >     * value, throwing {@code InterruptedException} if interrupted
706 >     * while waiting, or returning immediately if the current phase is
707 >     * not equal to the given phase value or this phaser is
708 >     * terminated.
709 >     *
710 >     * @param phase an arrival phase number, or negative value if
711 >     * terminated; this argument is normally the value returned by a
712 >     * previous call to {@code arrive} or {@code arriveAndDeregister}.
713 >     * @return the next arrival phase number, or the argument if it is
714 >     * negative, or the (negative) {@linkplain #getPhase() current phase}
715 >     * if terminated
716       * @throws InterruptedException if thread interrupted while waiting
717       */
718 <    public int awaitAdvanceInterruptibly(int phase) throws InterruptedException {
718 >    public int awaitAdvanceInterruptibly(int phase)
719 >        throws InterruptedException {
720 >        final Phaser root = this.root;
721 >        long s = (root == this) ? state : reconcileState();
722 >        int p = (int)(s >>> PHASE_SHIFT);
723          if (phase < 0)
724              return phase;
725 <        long s = getReconciledState();
726 <        int p = phaseOf(s);
727 <        if (p != phase)
728 <            return p;
729 <        if (unarrivedOf(s) != 0)
730 <            parent.awaitAdvanceInterruptibly(phase);
731 <        return interruptibleWait(phase);
725 >        if (p == phase) {
726 >            QNode node = new QNode(this, phase, true, false, 0L);
727 >            p = root.internalAwaitAdvance(phase, node);
728 >            if (node.wasInterrupted)
729 >                throw new InterruptedException();
730 >        }
731 >        return p;
732      }
733  
734      /**
735 <     * Awaits the phase of the barrier to advance from the given value
736 <     * or the given timeout elapses, or returns immediately if
737 <     * argument is negative or this barrier is terminated.
738 <     * @param phase the phase on entry to this method
739 <     * @return the phase on exit from this method
735 >     * Awaits the phase of this phaser to advance from the given phase
736 >     * value or the given timeout to elapse, throwing {@code
737 >     * InterruptedException} if interrupted while waiting, or
738 >     * returning immediately if the current phase is not equal to the
739 >     * given phase value or this phaser is terminated.
740 >     *
741 >     * @param phase an arrival phase number, or negative value if
742 >     * terminated; this argument is normally the value returned by a
743 >     * previous call to {@code arrive} or {@code arriveAndDeregister}.
744 >     * @param timeout how long to wait before giving up, in units of
745 >     *        {@code unit}
746 >     * @param unit a {@code TimeUnit} determining how to interpret the
747 >     *        {@code timeout} parameter
748 >     * @return the next arrival phase number, or the argument if it is
749 >     * negative, or the (negative) {@linkplain #getPhase() current phase}
750 >     * if terminated
751       * @throws InterruptedException if thread interrupted while waiting
752       * @throws TimeoutException if timed out while waiting
753       */
754 <    public int awaitAdvanceInterruptibly(int phase, long timeout, TimeUnit unit)
754 >    public int awaitAdvanceInterruptibly(int phase,
755 >                                         long timeout, TimeUnit unit)
756          throws InterruptedException, TimeoutException {
757 +        long nanos = unit.toNanos(timeout);
758 +        final Phaser root = this.root;
759 +        long s = (root == this) ? state : reconcileState();
760 +        int p = (int)(s >>> PHASE_SHIFT);
761          if (phase < 0)
762              return phase;
763 <        long s = getReconciledState();
764 <        int p = phaseOf(s);
765 <        if (p != phase)
766 <            return p;
767 <        if (unarrivedOf(s) == 0)
768 <            parent.awaitAdvanceInterruptibly(phase, timeout, unit);
769 <        return timedWait(phase, unit.toNanos(timeout));
763 >        if (p == phase) {
764 >            QNode node = new QNode(this, phase, true, true, nanos);
765 >            p = root.internalAwaitAdvance(phase, node);
766 >            if (node.wasInterrupted)
767 >                throw new InterruptedException();
768 >            else if (p == phase)
769 >                throw new TimeoutException();
770 >        }
771 >        return p;
772      }
773  
774      /**
775 <     * Forces this barrier to enter termination state. Counts of
776 <     * arrived and registered parties are unaffected. If this phaser
777 <     * has a parent, it too is terminated. This method may be useful
778 <     * for coordinating recovery after one or more tasks encounter
775 >     * Forces this phaser to enter termination state.  Counts of
776 >     * registered parties are unaffected.  If this phaser is a member
777 >     * of a tiered set of phasers, then all of the phasers in the set
778 >     * are terminated.  If this phaser is already terminated, this
779 >     * method has no effect.  This method may be useful for
780 >     * coordinating recovery after one or more tasks encounter
781       * unexpected exceptions.
782       */
783      public void forceTermination() {
784 <        for (;;) {
785 <            long s = getReconciledState();
786 <            int phase = phaseOf(s);
787 <            int parties = partiesOf(s);
788 <            int unarrived = unarrivedOf(s);
789 <            if (phase < 0 ||
790 <                casState(s, stateFor(-1, parties, unarrived))) {
784 >        // Only need to change root state
785 >        final Phaser root = this.root;
786 >        long s;
787 >        while ((s = root.state) >= 0) {
788 >            if (UNSAFE.compareAndSwapLong(root, stateOffset,
789 >                                          s, s | TERMINATION_BIT)) {
790 >                // signal all threads
791                  releaseWaiters(0);
792                  releaseWaiters(1);
602                if (parent != null)
603                    parent.forceTermination();
793                  return;
794              }
795          }
# Line 608 | Line 797 | public class Phaser {
797  
798      /**
799       * Returns the current phase number. The maximum phase number is
800 <     * <tt>Integer.MAX_VALUE</tt>, after which it restarts at
801 <     * zero. Upon termination, the phase number is negative.
800 >     * {@code Integer.MAX_VALUE}, after which it restarts at
801 >     * zero. Upon termination, the phase number is negative,
802 >     * in which case the prevailing phase prior to termination
803 >     * may be obtained via {@code getPhase() + Integer.MIN_VALUE}.
804 >     *
805       * @return the phase number, or a negative value if terminated
806       */
807      public final int getPhase() {
808 <        return phaseOf(getReconciledState());
808 >        return (int)(root.state >>> PHASE_SHIFT);
809      }
810  
811      /**
812 <     * Returns true if the current phase number equals the given phase.
813 <     * @param phase the phase
622 <     * @return true if the current phase number equals the given phase.
623 <     */
624 <    public final boolean hasPhase(int phase) {
625 <        return phaseOf(getReconciledState()) == phase;
626 <    }
627 <
628 <    /**
629 <     * Returns the number of parties registered at this barrier.
812 >     * Returns the number of parties registered at this phaser.
813 >     *
814       * @return the number of parties
815       */
816      public int getRegisteredParties() {
# Line 634 | Line 818 | public class Phaser {
818      }
819  
820      /**
821 <     * Returns the number of parties that have arrived at the current
822 <     * phase of this barrier.
821 >     * Returns the number of registered parties that have arrived at
822 >     * the current phase of this phaser. If this phaser has terminated,
823 >     * the returned value is meaningless and arbitrary.
824 >     *
825       * @return the number of arrived parties
826       */
827      public int getArrivedParties() {
828 <        return arrivedOf(state);
828 >        return arrivedOf(reconcileState());
829      }
830  
831      /**
832       * Returns the number of registered parties that have not yet
833 <     * arrived at the current phase of this barrier.
833 >     * arrived at the current phase of this phaser. If this phaser has
834 >     * terminated, the returned value is meaningless and arbitrary.
835 >     *
836       * @return the number of unarrived parties
837       */
838      public int getUnarrivedParties() {
839 <        return unarrivedOf(state);
839 >        return unarrivedOf(reconcileState());
840      }
841  
842      /**
843 <     * Returns the parent of this phaser, or null if none.
844 <     * @return the parent of this phaser, or null if none.
843 >     * Returns the parent of this phaser, or {@code null} if none.
844 >     *
845 >     * @return the parent of this phaser, or {@code null} if none
846       */
847      public Phaser getParent() {
848          return parent;
# Line 662 | Line 851 | public class Phaser {
851      /**
852       * Returns the root ancestor of this phaser, which is the same as
853       * this phaser if it has no parent.
854 <     * @return the root ancestor of this phaser.
854 >     *
855 >     * @return the root ancestor of this phaser
856       */
857      public Phaser getRoot() {
858          return root;
859      }
860  
861      /**
862 <     * Returns true if this barrier has been terminated.
863 <     * @return true if this barrier has been terminated
862 >     * Returns {@code true} if this phaser has been terminated.
863 >     *
864 >     * @return {@code true} if this phaser has been terminated
865       */
866      public boolean isTerminated() {
867 <        return getPhase() < 0;
867 >        return root.state < 0L;
868      }
869  
870      /**
871 <     * Overridable method to perform an action upon phase advance, and
872 <     * to control termination. This method is invoked whenever the
873 <     * barrier is tripped (and thus all other waiting parties are
874 <     * dormant). If it returns true, then, rather than advance the
875 <     * phase number, this barrier will be set to a final termination
876 <     * state, and subsequent calls to <tt>isTerminated</tt> will
877 <     * return true.
878 <     *
879 <     * <p> The default version returns true when the number of
880 <     * registered parties is zero. Normally, overrides that arrange
881 <     * termination for other reasons should also preserve this
882 <     * property.
883 <     *
884 <     * <p> You may override this method to perform an action with side
885 <     * effects visible to participating tasks, but it is in general
886 <     * only sensible to do so in designs where all parties register
887 <     * before any arrive, and all <tt>awaitAdvance</tt> at each phase.
888 <     * Otherwise, you cannot ensure lack of interference. In
889 <     * particular, this method may be invoked more than once per
890 <     * transition if other parties successfully register while the
891 <     * invocation of this method is in progress, thus postponing the
892 <     * transition until those parties also arrive, re-triggering this
893 <     * method.
894 <     *
895 <     * @param phase the phase number on entering the barrier
896 <     * @param registeredParties the current number of registered
897 <     * parties.
898 <     * @return true if this barrier should terminate
871 >     * Overridable method to perform an action upon impending phase
872 >     * advance, and to control termination. This method is invoked
873 >     * upon arrival of the party advancing this phaser (when all other
874 >     * waiting parties are dormant).  If this method returns {@code
875 >     * true}, this phaser will be set to a final termination state
876 >     * upon advance, and subsequent calls to {@link #isTerminated}
877 >     * will return true. Any (unchecked) Exception or Error thrown by
878 >     * an invocation of this method is propagated to the party
879 >     * attempting to advance this phaser, in which case no advance
880 >     * occurs.
881 >     *
882 >     * <p>The arguments to this method provide the state of the phaser
883 >     * prevailing for the current transition.  The effects of invoking
884 >     * arrival, registration, and waiting methods on this phaser from
885 >     * within {@code onAdvance} are unspecified and should not be
886 >     * relied on.
887 >     *
888 >     * <p>If this phaser is a member of a tiered set of phasers, then
889 >     * {@code onAdvance} is invoked only for its root phaser on each
890 >     * advance.
891 >     *
892 >     * <p>To support the most common use cases, the default
893 >     * implementation of this method returns {@code true} when the
894 >     * number of registered parties has become zero as the result of a
895 >     * party invoking {@code arriveAndDeregister}.  You can disable
896 >     * this behavior, thus enabling continuation upon future
897 >     * registrations, by overriding this method to always return
898 >     * {@code false}:
899 >     *
900 >     * <pre> {@code
901 >     * Phaser phaser = new Phaser() {
902 >     *   protected boolean onAdvance(int phase, int parties) { return false; }
903 >     * }}</pre>
904 >     *
905 >     * @param phase the current phase number on entry to this method,
906 >     * before this phaser is advanced
907 >     * @param registeredParties the current number of registered parties
908 >     * @return {@code true} if this phaser should terminate
909       */
910      protected boolean onAdvance(int phase, int registeredParties) {
911 <        return registeredParties <= 0;
911 >        return registeredParties == 0;
912      }
913  
914      /**
915       * Returns a string identifying this phaser, as well as its
916       * state.  The state, in brackets, includes the String {@code
917 <     * "phase ="} followed by the phase number, {@code "parties ="}
917 >     * "phase = "} followed by the phase number, {@code "parties = "}
918       * followed by the number of registered parties, and {@code
919 <     * "arrived ="} followed by the number of arrived parties
919 >     * "arrived = "} followed by the number of arrived parties.
920       *
921 <     * @return a string identifying this barrier, as well as its state
921 >     * @return a string identifying this phaser, as well as its state
922       */
923      public String toString() {
924 <        long s = getReconciledState();
724 <        return super.toString() + "[phase = " + phaseOf(s) + " parties = " + partiesOf(s) + " arrived = " + arrivedOf(s) + "]";
924 >        return stateToString(reconcileState());
925      }
926  
727    // methods for waiting
728
729    /** The number of CPUs, for spin control */
730    static final int NCPUS = Runtime.getRuntime().availableProcessors();
731
732    /**
733     * The number of times to spin before blocking in timed waits.
734     * The value is empirically derived.
735     */
736    static final int maxTimedSpins = (NCPUS < 2)? 0 : 32;
737
927      /**
928 <     * The number of times to spin before blocking in untimed waits.
740 <     * This is greater than timed value because untimed waits spin
741 <     * faster since they don't need to check times on each spin.
928 >     * Implementation of toString and string-based error messages
929       */
930 <    static final int maxUntimedSpins = maxTimedSpins * 32;
930 >    private String stateToString(long s) {
931 >        return super.toString() +
932 >            "[phase = " + phaseOf(s) +
933 >            " parties = " + partiesOf(s) +
934 >            " arrived = " + arrivedOf(s) + "]";
935 >    }
936  
937 <    /**
746 <     * The number of nanoseconds for which it is faster to spin
747 <     * rather than to use timed park. A rough estimate suffices.
748 <     */
749 <    static final long spinForTimeoutThreshold = 1000L;
937 >    // Waiting mechanics
938  
939      /**
940 <     * Wait nodes for Treiber stack representing wait queue for non-FJ
753 <     * tasks.
940 >     * Removes and signals threads from queue for phase.
941       */
942 <    static final class QNode {
943 <        QNode next;
944 <        volatile Thread thread; // nulled to cancel wait
945 <        QNode() {
946 <            thread = Thread.currentThread();
947 <        }
948 <        void signal() {
949 <            Thread t = thread;
950 <            if (t != null) {
764 <                thread = null;
942 >    private void releaseWaiters(int phase) {
943 >        QNode q;   // first element of queue
944 >        Thread t;  // its thread
945 >        AtomicReference<QNode> head = (phase & 1) == 0 ? evenQ : oddQ;
946 >        while ((q = head.get()) != null &&
947 >               q.phase != (int)(root.state >>> PHASE_SHIFT)) {
948 >            if (head.compareAndSet(q, q.next) &&
949 >                (t = q.thread) != null) {
950 >                q.thread = null;
951                  LockSupport.unpark(t);
952              }
953          }
954      }
955  
956      /**
957 <     * Removes and signals waiting threads from wait queue
957 >     * Variant of releaseWaiters that additionally tries to remove any
958 >     * nodes no longer waiting for advance due to timeout or
959 >     * interrupt. Currently, nodes are removed only if they are at
960 >     * head of queue, which suffices to reduce memory footprint in
961 >     * most usages.
962 >     *
963 >     * @return current phase on exit
964       */
965 <    private void releaseWaiters(int phase) {
966 <        AtomicReference<QNode> head = queueFor(phase);
967 <        QNode q;
968 <        while ((q = head.get()) != null) {
969 <            if (head.compareAndSet(q, q.next))
970 <                q.signal();
965 >    private int abortWait(int phase) {
966 >        AtomicReference<QNode> head = (phase & 1) == 0 ? evenQ : oddQ;
967 >        for (;;) {
968 >            Thread t;
969 >            QNode q = head.get();
970 >            int p = (int)(root.state >>> PHASE_SHIFT);
971 >            if (q == null || ((t = q.thread) != null && q.phase == p))
972 >                return p;
973 >            if (head.compareAndSet(q, q.next) && t != null) {
974 >                q.thread = null;
975 >                LockSupport.unpark(t);
976 >            }
977          }
978      }
979  
980 +    /** The number of CPUs, for spin control */
981 +    private static final int NCPU = Runtime.getRuntime().availableProcessors();
982 +
983      /**
984 <     * Enqueues node and waits unless aborted or signalled.
984 >     * The number of times to spin before blocking while waiting for
985 >     * advance, per arrival while waiting. On multiprocessors, fully
986 >     * blocking and waking up a large number of threads all at once is
987 >     * usually a very slow process, so we use rechargeable spins to
988 >     * avoid it when threads regularly arrive: When a thread in
989 >     * internalAwaitAdvance notices another arrival before blocking,
990 >     * and there appear to be enough CPUs available, it spins
991 >     * SPINS_PER_ARRIVAL more times before blocking. The value trades
992 >     * off good-citizenship vs big unnecessary slowdowns.
993       */
994 <    private int untimedWait(int phase) {
786 <        int spins = maxUntimedSpins;
787 <        QNode node = null;
788 <        boolean interrupted = false;
789 <        boolean queued = false;
790 <        int p;
791 <        while ((p = getPhase()) == phase) {
792 <            interrupted = Thread.interrupted();
793 <            if (node != null) {
794 <                if (!queued) {
795 <                    AtomicReference<QNode> head = queueFor(phase);
796 <                    queued = head.compareAndSet(node.next = head.get(), node);
797 <                }
798 <                else if (node.thread != null)
799 <                    LockSupport.park(this);
800 <            }
801 <            else if (spins <= 0)
802 <                node = new QNode();
803 <            else
804 <                --spins;
805 <        }
806 <        if (node != null)
807 <            node.thread = null;
808 <        if (interrupted)
809 <            Thread.currentThread().interrupt();
810 <        releaseWaiters(phase);
811 <        return p;
812 <    }
994 >    static final int SPINS_PER_ARRIVAL = (NCPU < 2) ? 1 : 1 << 8;
995  
996      /**
997 <     * Messier interruptible version
998 <     */
999 <    private int interruptibleWait(int phase) throws InterruptedException {
1000 <        int spins = maxUntimedSpins;
1001 <        QNode node = null;
1002 <        boolean queued = false;
1003 <        boolean interrupted = false;
997 >     * Possibly blocks and waits for phase to advance unless aborted.
998 >     * Call only from root node.
999 >     *
1000 >     * @param phase current phase
1001 >     * @param node if non-null, the wait node to track interrupt and timeout;
1002 >     * if null, denotes noninterruptible wait
1003 >     * @return current phase
1004 >     */
1005 >    private int internalAwaitAdvance(int phase, QNode node) {
1006 >        releaseWaiters(phase-1);          // ensure old queue clean
1007 >        boolean queued = false;           // true when node is enqueued
1008 >        int lastUnarrived = 0;            // to increase spins upon change
1009 >        int spins = SPINS_PER_ARRIVAL;
1010 >        long s;
1011          int p;
1012 <        while ((p = getPhase()) == phase) {
1013 <            if (interrupted = Thread.interrupted())
1012 >        while ((p = (int)((s = state) >>> PHASE_SHIFT)) == phase) {
1013 >            if (node == null) {           // spinning in noninterruptible mode
1014 >                int unarrived = (int)s & UNARRIVED_MASK;
1015 >                if (unarrived != lastUnarrived &&
1016 >                    (lastUnarrived = unarrived) < NCPU)
1017 >                    spins += SPINS_PER_ARRIVAL;
1018 >                boolean interrupted = Thread.interrupted();
1019 >                if (interrupted || --spins < 0) { // need node to record intr
1020 >                    node = new QNode(this, phase, false, false, 0L);
1021 >                    node.wasInterrupted = interrupted;
1022 >                }
1023 >            }
1024 >            else if (node.isReleasable()) // done or aborted
1025                  break;
1026 <            if (node != null) {
1027 <                if (!queued) {
1028 <                    AtomicReference<QNode> head = queueFor(phase);
1029 <                    queued = head.compareAndSet(node.next = head.get(), node);
1026 >            else if (!queued) {           // push onto queue
1027 >                AtomicReference<QNode> head = (phase & 1) == 0 ? evenQ : oddQ;
1028 >                QNode q = node.next = head.get();
1029 >                if ((q == null || q.phase == phase) &&
1030 >                    (int)(state >>> PHASE_SHIFT) == phase) // avoid stale enq
1031 >                    queued = head.compareAndSet(q, node);
1032 >            }
1033 >            else {
1034 >                try {
1035 >                    ForkJoinPool.managedBlock(node);
1036 >                } catch (InterruptedException ie) {
1037 >                    node.wasInterrupted = true;
1038                  }
831                else if (node.thread != null)
832                    LockSupport.park(this);
1039              }
1040 <            else if (spins <= 0)
1041 <                node = new QNode();
1042 <            else
1043 <                --spins;
1044 <        }
1045 <        if (node != null)
1046 <            node.thread = null;
1047 <        if (interrupted)
1048 <            throw new InterruptedException();
1040 >        }
1041 >
1042 >        if (node != null) {
1043 >            if (node.thread != null)
1044 >                node.thread = null;       // avoid need for unpark()
1045 >            if (node.wasInterrupted && !node.interruptible)
1046 >                Thread.currentThread().interrupt();
1047 >            if (p == phase && (p = (int)(state >>> PHASE_SHIFT)) == phase)
1048 >                return abortWait(phase); // possibly clean up on abort
1049 >        }
1050          releaseWaiters(phase);
1051          return p;
1052      }
1053  
1054      /**
1055 <     * Even messier timeout version.
1055 >     * Wait nodes for Treiber stack representing wait queue
1056       */
1057 <    private int timedWait(int phase, long nanos)
1058 <        throws InterruptedException, TimeoutException {
1059 <        int p;
1060 <        if ((p = getPhase()) == phase) {
1061 <            long lastTime = System.nanoTime();
1062 <            int spins = maxTimedSpins;
1063 <            QNode node = null;
1064 <            boolean queued = false;
1065 <            boolean interrupted = false;
1066 <            while ((p = getPhase()) == phase) {
1067 <                if (interrupted = Thread.interrupted())
1068 <                    break;
1069 <                long now = System.nanoTime();
1070 <                if ((nanos -= now - lastTime) <= 0)
1071 <                    break;
1072 <                lastTime = now;
1073 <                if (node != null) {
1074 <                    if (!queued) {
1075 <                        AtomicReference<QNode> head = queueFor(phase);
1076 <                        queued = head.compareAndSet(node.next = head.get(), node);
1077 <                    }
1078 <                    else if (node.thread != null &&
1079 <                             nanos > spinForTimeoutThreshold) {
1080 <                        LockSupport.parkNanos(this, nanos);
1081 <                    }
1057 >    static final class QNode implements ForkJoinPool.ManagedBlocker {
1058 >        final Phaser phaser;
1059 >        final int phase;
1060 >        final boolean interruptible;
1061 >        final boolean timed;
1062 >        boolean wasInterrupted;
1063 >        long nanos;
1064 >        long lastTime;
1065 >        volatile Thread thread; // nulled to cancel wait
1066 >        QNode next;
1067 >
1068 >        QNode(Phaser phaser, int phase, boolean interruptible,
1069 >              boolean timed, long nanos) {
1070 >            this.phaser = phaser;
1071 >            this.phase = phase;
1072 >            this.interruptible = interruptible;
1073 >            this.nanos = nanos;
1074 >            this.timed = timed;
1075 >            this.lastTime = timed ? System.nanoTime() : 0L;
1076 >            thread = Thread.currentThread();
1077 >        }
1078 >
1079 >        public boolean isReleasable() {
1080 >            if (thread == null)
1081 >                return true;
1082 >            if (phaser.getPhase() != phase) {
1083 >                thread = null;
1084 >                return true;
1085 >            }
1086 >            if (Thread.interrupted())
1087 >                wasInterrupted = true;
1088 >            if (wasInterrupted && interruptible) {
1089 >                thread = null;
1090 >                return true;
1091 >            }
1092 >            if (timed) {
1093 >                if (nanos > 0L) {
1094 >                    long now = System.nanoTime();
1095 >                    nanos -= now - lastTime;
1096 >                    lastTime = now;
1097 >                }
1098 >                if (nanos <= 0L) {
1099 >                    thread = null;
1100 >                    return true;
1101                  }
876                else if (spins <= 0)
877                    node = new QNode();
878                else
879                    --spins;
1102              }
1103 <            if (node != null)
882 <                node.thread = null;
883 <            if (interrupted)
884 <                throw new InterruptedException();
885 <            if (p == phase && (p = getPhase()) == phase)
886 <                throw new TimeoutException();
1103 >            return false;
1104          }
888        releaseWaiters(phase);
889        return p;
890    }
1105  
1106 <    // Temporary Unsafe mechanics for preliminary release
1106 >        public boolean block() {
1107 >            if (isReleasable())
1108 >                return true;
1109 >            else if (!timed)
1110 >                LockSupport.park(this);
1111 >            else if (nanos > 0)
1112 >                LockSupport.parkNanos(this, nanos);
1113 >            return isReleasable();
1114 >        }
1115 >    }
1116  
1117 <    static final Unsafe _unsafe;
895 <    static final long stateOffset;
1117 >    // Unsafe mechanics
1118  
1119 +    private static final sun.misc.Unsafe UNSAFE;
1120 +    private static final long stateOffset;
1121      static {
1122          try {
1123 <            if (Phaser.class.getClassLoader() != null) {
1124 <                Field f = Unsafe.class.getDeclaredField("theUnsafe");
1125 <                f.setAccessible(true);
1126 <                _unsafe = (Unsafe)f.get(null);
903 <            }
904 <            else
905 <                _unsafe = Unsafe.getUnsafe();
906 <            stateOffset = _unsafe.objectFieldOffset
907 <                (Phaser.class.getDeclaredField("state"));
1123 >            UNSAFE = getUnsafe();
1124 >            Class<?> k = Phaser.class;
1125 >            stateOffset = UNSAFE.objectFieldOffset
1126 >                (k.getDeclaredField("state"));
1127          } catch (Exception e) {
1128 <            throw new RuntimeException("Could not initialize intrinsics", e);
1128 >            throw new Error(e);
1129          }
1130      }
1131  
1132 <    final boolean casState(long cmp, long val) {
1133 <        return _unsafe.compareAndSwapLong(this, stateOffset, cmp, val);
1132 >    /**
1133 >     * Returns a sun.misc.Unsafe.  Suitable for use in a 3rd party package.
1134 >     * Replace with a simple call to Unsafe.getUnsafe when integrating
1135 >     * into a jdk.
1136 >     *
1137 >     * @return a sun.misc.Unsafe
1138 >     */
1139 >    private static sun.misc.Unsafe getUnsafe() {
1140 >        try {
1141 >            return sun.misc.Unsafe.getUnsafe();
1142 >        } catch (SecurityException se) {
1143 >            try {
1144 >                return java.security.AccessController.doPrivileged
1145 >                    (new java.security
1146 >                     .PrivilegedExceptionAction<sun.misc.Unsafe>() {
1147 >                        public sun.misc.Unsafe run() throws Exception {
1148 >                            java.lang.reflect.Field f = sun.misc
1149 >                                .Unsafe.class.getDeclaredField("theUnsafe");
1150 >                            f.setAccessible(true);
1151 >                            return (sun.misc.Unsafe) f.get(null);
1152 >                        }});
1153 >            } catch (java.security.PrivilegedActionException e) {
1154 >                throw new RuntimeException("Could not initialize intrinsics",
1155 >                                           e.getCause());
1156 >            }
1157 >        }
1158      }
1159   }

Diff Legend

Removed lines
+ Added lines
< Changed lines
> Changed lines