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.49 by dl, Fri Nov 5 23:01:47 2010 UTC vs.
Revision 1.81 by jsr166, Fri Jul 15 18:49:12 2016 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  
9 < import java.util.concurrent.*;
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;
13  
# Line 16 | Line 17 | import java.util.concurrent.locks.LockSu
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
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
# Line 29 | Line 30 | import java.util.concurrent.locks.LockSu
30   * (However, you can introduce such bookkeeping by subclassing this
31   * class.)
32   *
33 < * <p> <b>Synchronization.</b> Like a {@code CyclicBarrier}, a {@code
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 {@code Phaser} has an associated phase number. The
38 < * phase number starts at zero, and advances when all parties arrive
39 < * at the barrier, wrapping around to zero after reaching {@code
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 barrier and upon awaiting
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> <b>Arrival.</b> Methods {@link #arrive} and
48 < *       {@link #arriveAndDeregister} record arrival at a
49 < *       barrier. These methods do not block, but return an associated
50 < *       <em>arrival phase number</em>; that is, the phase number of
51 < *       the barrier to which the arrival applied. When the final
52 < *       party for a given phase arrives, an optional barrier action
53 < *       is performed and the phase advances.  Barrier actions,
54 < *       performed by the party triggering a phase advance, are
55 < *       arranged by overriding method {@link #onAdvance(int, int)},
56 < *       which also controls termination. Overriding this method is
57 < *       similar to, but more flexible than, providing a barrier
58 < *       action to a {@code CyclicBarrier}.
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
60 > *   <li><b>Waiting.</b> Method {@link #awaitAdvance} requires an
61   *       argument indicating an arrival phase number, and returns when
62 < *       the barrier advances to (or is already at) a different phase.
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 barrier. If necessary, you can perform any
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},
# Line 73 | Line 74 | import java.util.concurrent.locks.LockSu
74   *
75   * </ul>
76   *
77 < * <p> <b>Termination.</b> A {@code Phaser} may enter a
78 < * <em>termination</em> state in which all synchronization methods
79 < * immediately return without updating phaser state or waiting for
80 < * advance, and indicating (via a negative phase value) that execution
81 < * is complete.  Termination is triggered when an invocation of {@code
82 < * onAdvance} returns {@code true}.  As illustrated below, when
83 < * phasers control actions with a fixed number of iterations, it is
84 < * often convenient to override this method to cause termination when
85 < * the current phase number reaches a threshold. Method {@link
86 < * #forceTermination} is also available to abruptly release waiting
87 < * threads and allow them to terminate.
88 < *
89 < * <p> <b>Tiering.</b> Phasers may be <em>tiered</em> (i.e., arranged
90 < * in tree structures) to reduce contention. Phasers with large
91 < * numbers of parties that would otherwise experience heavy
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
# Line 116 | Line 130 | import java.util.concurrent.locks.LockSu
130   * void runTasks(List<Runnable> tasks) {
131   *   final Phaser phaser = new Phaser(1); // "1" to register self
132   *   // create and start threads
133 < *   for (Runnable task : tasks) {
133 > *   for (final Runnable task : tasks) {
134   *     phaser.register();
135   *     new Thread() {
136   *       public void run() {
# Line 179 | Line 193 | import java.util.concurrent.locks.LockSu
193   *   phaser.arriveAndDeregister();
194   * }}</pre>
195   *
196 < *
197 < * <p>To create a set of tasks using a tree of phasers,
198 < * you could use code of the following form, assuming a
199 < * Task class with a constructor accepting a phaser that
200 < * it registers with upon construction:
196 > * <p>To create a set of {@code n} tasks using a tree of phasers, you
197 > * could use code of the following form, assuming a Task class with a
198 > * constructor accepting a {@code Phaser} that it registers with upon
199 > * construction. After invocation of {@code build(new Task[n], 0, n,
200 > * new Phaser())}, these tasks could then be started, for example by
201 > * submitting to a pool:
202   *
203   *  <pre> {@code
204 < * void build(Task[] actions, int lo, int hi, Phaser ph) {
204 > * void build(Task[] tasks, int lo, int hi, Phaser ph) {
205   *   if (hi - lo > TASKS_PER_PHASER) {
206   *     for (int i = lo; i < hi; i += TASKS_PER_PHASER) {
207   *       int j = Math.min(i + TASKS_PER_PHASER, hi);
208 < *       build(actions, i, j, new Phaser(ph));
208 > *       build(tasks, i, j, new Phaser(ph));
209   *     }
210   *   } else {
211   *     for (int i = lo; i < hi; ++i)
212 < *       actions[i] = new Task(ph);
212 > *       tasks[i] = new Task(ph);
213   *       // assumes new Task(ph) performs ph.register()
214   *   }
215 < * }
201 < * // .. initially called, for n tasks via
202 < * build(new Task[n], 0, n, new Phaser());}</pre>
215 > * }}</pre>
216   *
217   * The best value of {@code TASKS_PER_PHASER} depends mainly on
218 < * expected barrier synchronization rates. A value as low as four may
219 < * be appropriate for extremely small per-barrier task bodies (thus
218 > * expected synchronization rates. A value as low as four may
219 > * be appropriate for extremely small per-phase task bodies (thus
220   * high rates), or up to hundreds for extremely large ones.
221   *
222   * <p><b>Implementation notes</b>: This implementation restricts the
# Line 223 | Line 236 | public class Phaser {
236       */
237  
238      /**
239 <     * Barrier state representation. Conceptually, a barrier contains
227 <     * four values:
239 >     * Primary state representation, holding four bit-fields:
240       *
241 <     * * parties -- the number of parties to wait (16 bits)
242 <     * * unarrived -- the number of parties yet to hit barrier (16 bits)
243 <     * * phase -- the generation of the barrier (31 bits)
244 <     * * terminated -- set if barrier is terminated (1 bit)
245 <     *
246 <     * However, to efficiently maintain atomicity, these values are
247 <     * packed into a single (atomic) long. Termination uses the sign
248 <     * bit of 32 bit representation of phase, so phase is set to -1 on
249 <     * termination. Good performance relies on keeping state decoding
250 <     * and encoding simple, and keeping race windows short.
251 <     *
252 <     * Note: there are some cheats in arrive() that rely on unarrived
253 <     * count being lowest 16 bits.
241 >     * unarrived  -- the number of parties yet to hit barrier (bits  0-15)
242 >     * parties    -- the number of parties to wait            (bits 16-31)
243 >     * phase      -- the generation of the barrier            (bits 32-62)
244 >     * terminated -- set if barrier is terminated             (bit  63 / sign)
245 >     *
246 >     * Except that a phaser with no registered parties is
247 >     * distinguished by the otherwise illegal state of having zero
248 >     * parties and one unarrived parties (encoded as EMPTY below).
249 >     *
250 >     * To efficiently maintain atomicity, these values are packed into
251 >     * a single (atomic) long. Good performance relies on keeping
252 >     * state decoding and encoding simple, and keeping race windows
253 >     * short.
254 >     *
255 >     * All state updates are performed via CAS except initial
256 >     * registration of a sub-phaser (i.e., one with a non-null
257 >     * parent).  In this (relatively rare) case, we use built-in
258 >     * synchronization to lock while first registering with its
259 >     * parent.
260 >     *
261 >     * The phase of a subphaser is allowed to lag that of its
262 >     * ancestors until it is actually accessed -- see method
263 >     * reconcileState.
264       */
265      private volatile long state;
266  
267 <    private static final int ushortMask = 0xffff;
268 <    private static final int phaseMask  = 0x7fffffff;
267 >    private static final int  MAX_PARTIES     = 0xffff;
268 >    private static final int  MAX_PHASE       = Integer.MAX_VALUE;
269 >    private static final int  PARTIES_SHIFT   = 16;
270 >    private static final int  PHASE_SHIFT     = 32;
271 >    private static final int  UNARRIVED_MASK  = 0xffff;      // to mask ints
272 >    private static final long PARTIES_MASK    = 0xffff0000L; // to mask longs
273 >    private static final long COUNTS_MASK     = 0xffffffffL;
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  ONE_DEREGISTER  = ONE_ARRIVAL|ONE_PARTY;
280 >    private static final int  EMPTY           = 1;
281 >
282 >    // The following unpacking methods are usually manually inlined
283  
284      private static int unarrivedOf(long s) {
285 <        return (int) (s & ushortMask);
285 >        int counts = (int)s;
286 >        return (counts == EMPTY) ? 0 : (counts & UNARRIVED_MASK);
287      }
288  
289      private static int partiesOf(long s) {
290 <        return ((int) s) >>> 16;
290 >        return (int)s >>> PARTIES_SHIFT;
291      }
292  
293      private static int phaseOf(long s) {
294 <        return (int) (s >>> 32);
294 >        return (int)(s >>> PHASE_SHIFT);
295      }
296  
297      private static int arrivedOf(long s) {
298 <        return partiesOf(s) - unarrivedOf(s);
299 <    }
300 <
264 <    private static long stateFor(int phase, int parties, int unarrived) {
265 <        return ((((long) phase) << 32) | (((long) parties) << 16) |
266 <                (long) unarrived);
267 <    }
268 <
269 <    private static long trippedStateFor(int phase, int parties) {
270 <        long lp = (long) parties;
271 <        return (((long) phase) << 32) | (lp << 16) | lp;
272 <    }
273 <
274 <    /**
275 <     * Returns message string for bad bounds exceptions.
276 <     */
277 <    private static String badBounds(int parties, int unarrived) {
278 <        return ("Attempt to set " + unarrived +
279 <                " unarrived of " + parties + " parties");
298 >        int counts = (int)s;
299 >        return (counts == EMPTY) ? 0 :
300 >            (counts >>> PARTIES_SHIFT) - (counts & UNARRIVED_MASK);
301      }
302  
303      /**
# Line 285 | Line 306 | public class Phaser {
306      private final Phaser parent;
307  
308      /**
309 <     * The root of phaser tree. Equals this if not in a tree.  Used to
289 <     * support faster state push-down.
309 >     * The root of phaser tree. Equals this if not in a tree.
310       */
311      private final Phaser root;
312  
293    // Wait queues
294
313      /**
314       * Heads of Treiber stacks for waiting threads. To eliminate
315       * contention when releasing some threads while adding others, we
316       * use two of them, alternating across even and odd phases.
317       * Subphasers share queues with root to speed up releases.
318       */
319 <    private final AtomicReference<QNode> evenQ = new AtomicReference<QNode>();
320 <    private final AtomicReference<QNode> oddQ  = new AtomicReference<QNode>();
319 >    private final AtomicReference<QNode> evenQ;
320 >    private final AtomicReference<QNode> oddQ;
321  
322      private AtomicReference<QNode> queueFor(int phase) {
323 <        Phaser r = root;
306 <        return ((phase & 1) == 0) ? r.evenQ : r.oddQ;
323 >        return ((phase & 1) == 0) ? evenQ : oddQ;
324      }
325  
326      /**
327 <     * Returns current state, first resolving lagged propagation from
311 <     * root if necessary.
327 >     * Returns message string for bounds exceptions on arrival.
328       */
329 <    private long getReconciledState() {
330 <        return (parent == null) ? state : reconcileState();
329 >    private String badArrive(long s) {
330 >        return "Attempted arrival of unregistered party for " +
331 >            stateToString(s);
332 >    }
333 >
334 >    /**
335 >     * Returns message string for bounds exceptions on registration.
336 >     */
337 >    private String badRegister(long s) {
338 >        return "Attempt to register more than " +
339 >            MAX_PARTIES + " parties for " + stateToString(s);
340 >    }
341 >
342 >    /**
343 >     * Main implementation for methods arrive and arriveAndDeregister.
344 >     * Manually tuned to speed up and minimize race windows for the
345 >     * common case of just decrementing unarrived field.
346 >     *
347 >     * @param adjust value to subtract from state;
348 >     *               ONE_ARRIVAL for arrive,
349 >     *               ONE_DEREGISTER for arriveAndDeregister
350 >     */
351 >    private int doArrive(int adjust) {
352 >        final Phaser root = this.root;
353 >        for (;;) {
354 >            long s = (root == this) ? state : reconcileState();
355 >            int phase = (int)(s >>> PHASE_SHIFT);
356 >            if (phase < 0)
357 >                return phase;
358 >            int counts = (int)s;
359 >            int unarrived = (counts == EMPTY) ? 0 : (counts & UNARRIVED_MASK);
360 >            if (unarrived <= 0)
361 >                throw new IllegalStateException(badArrive(s));
362 >            if (UNSAFE.compareAndSwapLong(this, stateOffset, s, s-=adjust)) {
363 >                if (unarrived == 1) {
364 >                    long n = s & PARTIES_MASK;  // base of next state
365 >                    int nextUnarrived = (int)n >>> PARTIES_SHIFT;
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 >                        int nextPhase = (phase + 1) & MAX_PHASE;
374 >                        n |= (long)nextPhase << PHASE_SHIFT;
375 >                        UNSAFE.compareAndSwapLong(this, stateOffset, s, n);
376 >                        releaseWaiters(phase);
377 >                    }
378 >                    else if (nextUnarrived == 0) { // propagate deregistration
379 >                        phase = parent.doArrive(ONE_DEREGISTER);
380 >                        UNSAFE.compareAndSwapLong(this, stateOffset,
381 >                                                  s, s | EMPTY);
382 >                    }
383 >                    else
384 >                        phase = parent.doArrive(ONE_ARRIVAL);
385 >                }
386 >                return phase;
387 >            }
388 >        }
389      }
390  
391      /**
392 <     * Recursively resolves state.
392 >     * Implementation of register, bulkRegister
393 >     *
394 >     * @param registrations number to add to both parties and
395 >     * unarrived fields. Must be greater than zero.
396       */
397 <    private long reconcileState() {
398 <        Phaser par = parent;
399 <        long s = state;
400 <        if (par != null) {
401 <            int phase, rootPhase;
402 <            while ((phase = phaseOf(s)) >= 0 &&
403 <                   (rootPhase = phaseOf(root.state)) != phase &&
404 <                   (rootPhase < 0 || unarrivedOf(s) == 0)) {
405 <                long parentState = par.getReconciledState();
406 <                int parentPhase = phaseOf(parentState);
407 <                int parties = partiesOf(s);
408 <                long next = trippedStateFor(parentPhase, parties);
409 <                if (phaseOf(root.state) == rootPhase &&
410 <                    parentPhase != phase &&
411 <                    state == s && casState(s, next)) {
412 <                    releaseWaiters(phase);
413 <                    if (parties == 0) // exit if the final deregistration
397 >    private int doRegister(int registrations) {
398 >        // adjustment to state
399 >        long adjust = ((long)registrations << PARTIES_SHIFT) | registrations;
400 >        final Phaser parent = this.parent;
401 >        int phase;
402 >        for (;;) {
403 >            long s = (parent == null) ? state : reconcileState();
404 >            int counts = (int)s;
405 >            int parties = counts >>> PARTIES_SHIFT;
406 >            int unarrived = counts & UNARRIVED_MASK;
407 >            if (registrations > MAX_PARTIES - parties)
408 >                throw new IllegalStateException(badRegister(s));
409 >            phase = (int)(s >>> PHASE_SHIFT);
410 >            if (phase < 0)
411 >                break;
412 >            if (counts != EMPTY) {                  // not 1st registration
413 >                if (parent == null || reconcileState() == s) {
414 >                    if (unarrived == 0)             // wait out advance
415 >                        root.internalAwaitAdvance(phase, null);
416 >                    else if (UNSAFE.compareAndSwapLong(this, stateOffset,
417 >                                                       s, s + adjust))
418                          break;
419                  }
420 <                s = state;
420 >            }
421 >            else if (parent == null) {              // 1st root registration
422 >                long next = ((long)phase << PHASE_SHIFT) | adjust;
423 >                if (UNSAFE.compareAndSwapLong(this, stateOffset, s, next))
424 >                    break;
425 >            }
426 >            else {
427 >                synchronized (this) {               // 1st sub registration
428 >                    if (state == s) {               // recheck under lock
429 >                        phase = parent.doRegister(1);
430 >                        if (phase < 0)
431 >                            break;
432 >                        // finish registration whenever parent registration
433 >                        // succeeded, even when racing with termination,
434 >                        // since these are part of the same "transaction".
435 >                        while (!UNSAFE.compareAndSwapLong
436 >                               (this, stateOffset, s,
437 >                                ((long)phase << PHASE_SHIFT) | adjust)) {
438 >                            s = state;
439 >                            phase = (int)(root.state >>> PHASE_SHIFT);
440 >                            // assert (int)s == EMPTY;
441 >                        }
442 >                        break;
443 >                    }
444 >                }
445              }
446          }
447 +        return phase;
448 +    }
449 +
450 +    /**
451 +     * Resolves lagged phase propagation from root if necessary.
452 +     * Reconciliation normally occurs when root has advanced but
453 +     * subphasers have not yet done so, in which case they must finish
454 +     * their own advance by setting unarrived to parties (or if
455 +     * parties is zero, resetting to unregistered EMPTY state).
456 +     *
457 +     * @return reconciled state
458 +     */
459 +    private long reconcileState() {
460 +        final Phaser root = this.root;
461 +        long s = state;
462 +        if (root != this) {
463 +            int phase, p;
464 +            // CAS to root phase with current parties, tripping unarrived
465 +            while ((phase = (int)(root.state >>> PHASE_SHIFT)) !=
466 +                   (int)(s >>> PHASE_SHIFT) &&
467 +                   !UNSAFE.compareAndSwapLong
468 +                   (this, stateOffset, s,
469 +                    s = (((long)phase << PHASE_SHIFT) |
470 +                         ((phase < 0) ? (s & COUNTS_MASK) :
471 +                          (((p = (int)s >>> PARTIES_SHIFT) == 0) ? EMPTY :
472 +                           ((s & PARTIES_MASK) | p))))))
473 +                s = state;
474 +        }
475          return s;
476      }
477  
478      /**
479 <     * Creates a new phaser without any initially registered parties,
480 <     * initial phase number 0, and no parent. Any thread using this
479 >     * Creates a new phaser with no initially registered parties, no
480 >     * parent, and initial phase number 0. Any thread using this
481       * phaser will need to first register for it.
482       */
483      public Phaser() {
484 <        this(null);
484 >        this(null, 0);
485      }
486  
487      /**
488       * Creates a new phaser with the given number of registered
489 <     * unarrived parties, initial phase number 0, and no parent.
489 >     * unarrived parties, no parent, and initial phase number 0.
490       *
491 <     * @param parties the number of parties required to trip barrier
491 >     * @param parties the number of parties required to advance to the
492 >     * next phase
493       * @throws IllegalArgumentException if parties less than zero
494       * or greater than the maximum number of parties supported
495       */
# Line 364 | Line 498 | public class Phaser {
498      }
499  
500      /**
501 <     * Creates a new phaser with the given parent, without any
368 <     * initially registered parties. If parent is non-null this phaser
369 <     * is registered with the parent and its initial phase number is
370 <     * the same as that of parent phaser.
501 >     * Equivalent to {@link #Phaser(Phaser, int) Phaser(parent, 0)}.
502       *
503       * @param parent the parent phaser
504       */
505      public Phaser(Phaser parent) {
506 <        int phase = 0;
376 <        this.parent = parent;
377 <        if (parent != null) {
378 <            this.root = parent.root;
379 <            phase = parent.register();
380 <        }
381 <        else
382 <            this.root = this;
383 <        this.state = trippedStateFor(phase, 0);
506 >        this(parent, 0);
507      }
508  
509      /**
510       * Creates a new phaser with the given parent and number of
511 <     * registered unarrived parties. If parent is non-null, this phaser
512 <     * is registered with the parent and its initial phase number is
513 <     * the same as that of parent phaser.
511 >     * registered unarrived parties.  When the given parent is non-null
512 >     * and the given number of parties is greater than zero, this
513 >     * child phaser is registered with its parent.
514       *
515       * @param parent the parent phaser
516 <     * @param parties the number of parties required to trip barrier
516 >     * @param parties the number of parties required to advance to the
517 >     * next phase
518       * @throws IllegalArgumentException if parties less than zero
519       * or greater than the maximum number of parties supported
520       */
521      public Phaser(Phaser parent, int parties) {
522 <        if (parties < 0 || parties > ushortMask)
522 >        if (parties >>> PARTIES_SHIFT != 0)
523              throw new IllegalArgumentException("Illegal number of parties");
524          int phase = 0;
525          this.parent = parent;
526          if (parent != null) {
527 <            this.root = parent.root;
528 <            phase = parent.register();
527 >            final Phaser root = parent.root;
528 >            this.root = root;
529 >            this.evenQ = root.evenQ;
530 >            this.oddQ = root.oddQ;
531 >            if (parties != 0)
532 >                phase = parent.doRegister(1);
533          }
534 <        else
534 >        else {
535              this.root = this;
536 <        this.state = trippedStateFor(phase, parties);
536 >            this.evenQ = new AtomicReference<QNode>();
537 >            this.oddQ = new AtomicReference<QNode>();
538 >        }
539 >        this.state = (parties == 0) ? (long)EMPTY :
540 >            ((long)phase << PHASE_SHIFT) |
541 >            ((long)parties << PARTIES_SHIFT) |
542 >            ((long)parties);
543      }
544  
545      /**
546 <     * Adds a new unarrived party to this phaser.
547 <     * If an ongoing invocation of {@link #onAdvance} is in progress,
548 <     * this method waits until its completion before registering.
549 <     *
550 <     * @return the arrival phase number to which this registration applied
546 >     * Adds a new unarrived party to this phaser.  If an ongoing
547 >     * invocation of {@link #onAdvance} is in progress, this method
548 >     * may await its completion before returning.  If this phaser has
549 >     * a parent, and this phaser previously had no registered parties,
550 >     * this child phaser is also registered with its parent. If
551 >     * this phaser is terminated, the attempt to register has
552 >     * no effect, and a negative value is returned.
553 >     *
554 >     * @return the arrival phase number to which this registration
555 >     * applied.  If this value is negative, then this phaser has
556 >     * terminated, in which case registration has no effect.
557       * @throws IllegalStateException if attempting to register more
558       * than the maximum supported number of parties
559       */
# Line 424 | Line 564 | public class Phaser {
564      /**
565       * Adds the given number of new unarrived parties to this phaser.
566       * If an ongoing invocation of {@link #onAdvance} is in progress,
567 <     * this method waits until its completion before registering.
568 <     *
569 <     * @param parties the number of additional parties required to trip barrier
570 <     * @return the arrival phase number to which this registration applied
567 >     * this method may await its completion before returning.  If this
568 >     * phaser has a parent, and the given number of parties is greater
569 >     * than zero, and this phaser previously had no registered
570 >     * parties, this child phaser is also registered with its parent.
571 >     * If this phaser is terminated, the attempt to register has no
572 >     * effect, and a negative value is returned.
573 >     *
574 >     * @param parties the number of additional parties required to
575 >     * advance to the next phase
576 >     * @return the arrival phase number to which this registration
577 >     * applied.  If this value is negative, then this phaser has
578 >     * terminated, in which case registration has no effect.
579       * @throws IllegalStateException if attempting to register more
580       * than the maximum supported number of parties
581       * @throws IllegalArgumentException if {@code parties < 0}
# Line 441 | Line 589 | public class Phaser {
589      }
590  
591      /**
592 <     * Shared code for register, bulkRegister
593 <     */
594 <    private int doRegister(int registrations) {
595 <        Phaser par = parent;
596 <        long s;
597 <        int phase;
450 <        while ((phase = phaseOf(s = par==null? state:reconcileState())) >= 0) {
451 <            int p = partiesOf(s);
452 <            int u = unarrivedOf(s);
453 <            int unarrived = u + registrations;
454 <            int parties = p + registrations;
455 <            if (par == null || phase == phaseOf(root.state)) {
456 <                if (parties > ushortMask || unarrived > ushortMask)
457 <                    throw new IllegalStateException(badBounds(parties,
458 <                                                              unarrived));
459 <                else if (p != 0 && u == 0)       // back off if advancing
460 <                    Thread.yield();              // not worth actually blocking
461 <                else if (casState(s, stateFor(phase, parties, unarrived)))
462 <                    break;
463 <            }
464 <        }
465 <        return phase;
466 <    }
467 <
468 <    /**
469 <     * Arrives at the barrier, but does not wait for others.  (You can
470 <     * in turn wait for others via {@link #awaitAdvance}).  It is an
471 <     * unenforced usage error for an unregistered party to invoke this
472 <     * method.
592 >     * Arrives at this phaser, without waiting for others to arrive.
593 >     *
594 >     * <p>It is a usage error for an unregistered party to invoke this
595 >     * method.  However, this error may result in an {@code
596 >     * IllegalStateException} only upon some subsequent operation on
597 >     * this phaser, if ever.
598       *
599       * @return the arrival phase number, or a negative value if terminated
600       * @throws IllegalStateException if not terminated and the number
601       * of unarrived parties would become negative
602       */
603      public int arrive() {
604 <        Phaser par = parent;
480 <        long s;
481 <        int phase;
482 <        while ((phase = phaseOf(s = par==null? state:reconcileState())) >= 0) {
483 <            int parties = partiesOf(s);
484 <            int unarrived = unarrivedOf(s) - 1;
485 <            if (parties == 0 || unarrived < 0)
486 <                throw new IllegalStateException(badBounds(parties,
487 <                                                          unarrived));
488 <            else if (unarrived > 0) {           // Not the last arrival
489 <                if (casState(s, s - 1))         // s-1 adds one arrival
490 <                    break;
491 <            }
492 <            else if (par == null) {             // directly trip
493 <                if (casState(s, trippedStateFor(onAdvance(phase, parties) ? -1 :
494 <                                                ((phase + 1) & phaseMask),
495 <                                                parties))) {
496 <                    releaseWaiters(phase);
497 <                    break;
498 <                }
499 <            }
500 <            else if (phaseOf(root.state) == phase && casState(s, s - 1)) {
501 <                par.arrive();                   // cascade to parent
502 <                reconcileState();
503 <                break;
504 <            }
505 <        }
506 <        return phase;
604 >        return doArrive(ONE_ARRIVAL);
605      }
606  
607      /**
608 <     * Arrives at the barrier and deregisters from it without waiting
609 <     * for others. Deregistration reduces the number of parties
610 <     * required to trip the barrier in future phases.  If this phaser
608 >     * Arrives at this phaser and deregisters from it without waiting
609 >     * for others to arrive. Deregistration reduces the number of
610 >     * parties required to advance in future phases.  If this phaser
611       * has a parent, and deregistration causes this phaser to have
612 <     * zero parties, this phaser also arrives at and is deregistered
613 <     * from its parent.  It is an unenforced usage error for an
614 <     * unregistered party to invoke this method.
612 >     * zero parties, this phaser is also deregistered from its parent.
613 >     *
614 >     * <p>It is a usage error for an unregistered party to invoke this
615 >     * method.  However, this error may result in an {@code
616 >     * IllegalStateException} only upon some subsequent operation on
617 >     * this phaser, if ever.
618       *
619       * @return the arrival phase number, or a negative value if terminated
620       * @throws IllegalStateException if not terminated and the number
621       * of registered or unarrived parties would become negative
622       */
623      public int arriveAndDeregister() {
624 <        // similar to arrive, but too different to merge
524 <        Phaser par = parent;
525 <        long s;
526 <        int phase;
527 <        while ((phase = phaseOf(s = par==null? state:reconcileState())) >= 0) {
528 <            int parties = partiesOf(s) - 1;
529 <            int unarrived = unarrivedOf(s) - 1;
530 <            if (parties < 0 || unarrived < 0)
531 <                throw new IllegalStateException(badBounds(parties,
532 <                                                          unarrived));
533 <            else if (unarrived > 0) {
534 <                if (casState(s, stateFor(phase, parties, unarrived)))
535 <                    break;
536 <            }
537 <            else if (par == null) {
538 <                if (casState(s, trippedStateFor(onAdvance(phase, parties)? -1:
539 <                                                (phase + 1) & phaseMask,
540 <                                                parties))) {
541 <                    releaseWaiters(phase);
542 <                    break;
543 <                }
544 <            }
545 <            else if (phaseOf(root.state) == phase &&
546 <                     casState(s, stateFor(phase, parties, 0))) {
547 <                if (parties == 0)
548 <                    par.arriveAndDeregister();
549 <                else
550 <                    par.arrive();
551 <                reconcileState();
552 <                break;
553 <            }
554 <        }
555 <        return phase;
624 >        return doArrive(ONE_DEREGISTER);
625      }
626  
627      /**
628 <     * Arrives at the barrier and awaits others. Equivalent in effect
628 >     * Arrives at this phaser and awaits others. Equivalent in effect
629       * to {@code awaitAdvance(arrive())}.  If you need to await with
630       * interruption or timeout, you can arrange this with an analogous
631       * construction using one of the other forms of the {@code
632       * awaitAdvance} method.  If instead you need to deregister upon
633 <     * arrival, use {@link #arriveAndDeregister}. It is an unenforced
565 <     * usage error for an unregistered party to invoke this method.
633 >     * arrival, use {@code awaitAdvance(arriveAndDeregister())}.
634       *
635 <     * @return the arrival phase number, or a negative number if terminated
635 >     * <p>It is a usage error for an unregistered party to invoke this
636 >     * method.  However, this error may result in an {@code
637 >     * IllegalStateException} only upon some subsequent operation on
638 >     * this phaser, if ever.
639 >     *
640 >     * @return the arrival phase number, or the (negative)
641 >     * {@linkplain #getPhase() current phase} if terminated
642       * @throws IllegalStateException if not terminated and the number
643       * of unarrived parties would become negative
644       */
645      public int arriveAndAwaitAdvance() {
646 <        return awaitAdvance(arrive());
646 >        // Specialization of doArrive+awaitAdvance eliminating some reads/paths
647 >        final Phaser root = this.root;
648 >        for (;;) {
649 >            long s = (root == this) ? state : reconcileState();
650 >            int phase = (int)(s >>> PHASE_SHIFT);
651 >            if (phase < 0)
652 >                return phase;
653 >            int counts = (int)s;
654 >            int unarrived = (counts == EMPTY) ? 0 : (counts & UNARRIVED_MASK);
655 >            if (unarrived <= 0)
656 >                throw new IllegalStateException(badArrive(s));
657 >            if (UNSAFE.compareAndSwapLong(this, stateOffset, s,
658 >                                          s -= ONE_ARRIVAL)) {
659 >                if (unarrived > 1)
660 >                    return root.internalAwaitAdvance(phase, null);
661 >                if (root != this)
662 >                    return parent.arriveAndAwaitAdvance();
663 >                long n = s & PARTIES_MASK;  // base of next state
664 >                int nextUnarrived = (int)n >>> PARTIES_SHIFT;
665 >                if (onAdvance(phase, nextUnarrived))
666 >                    n |= TERMINATION_BIT;
667 >                else if (nextUnarrived == 0)
668 >                    n |= EMPTY;
669 >                else
670 >                    n |= nextUnarrived;
671 >                int nextPhase = (phase + 1) & MAX_PHASE;
672 >                n |= (long)nextPhase << PHASE_SHIFT;
673 >                if (!UNSAFE.compareAndSwapLong(this, stateOffset, s, n))
674 >                    return (int)(state >>> PHASE_SHIFT); // terminated
675 >                releaseWaiters(phase);
676 >                return nextPhase;
677 >            }
678 >        }
679      }
680  
681      /**
682 <     * Awaits the phase of the barrier to advance from the given phase
683 <     * value, returning immediately if the current phase of the
684 <     * barrier is not equal to the given phase value or this barrier
579 <     * is terminated.  It is an unenforced usage error for an
580 <     * unregistered party to invoke this method.
682 >     * Awaits the phase of this phaser to advance from the given phase
683 >     * value, returning immediately if the current phase is not equal
684 >     * to the given phase value or this phaser is terminated.
685       *
686       * @param phase an arrival phase number, or negative value if
687       * terminated; this argument is normally the value returned by a
688 <     * previous call to {@code arrive} or its variants
689 <     * @return the next arrival phase number, or a negative value
690 <     * if terminated or argument is negative
688 >     * previous call to {@code arrive} or {@code arriveAndDeregister}.
689 >     * @return the next arrival phase number, or the argument if it is
690 >     * negative, or the (negative) {@linkplain #getPhase() current phase}
691 >     * if terminated
692       */
693      public int awaitAdvance(int phase) {
694 +        final Phaser root = this.root;
695 +        long s = (root == this) ? state : reconcileState();
696 +        int p = (int)(s >>> PHASE_SHIFT);
697          if (phase < 0)
698              return phase;
699 <        int p = getPhase();
700 <        if (p != phase)
701 <            return p;
594 <        return untimedWait(phase);
699 >        if (p == phase)
700 >            return root.internalAwaitAdvance(phase, null);
701 >        return p;
702      }
703  
704      /**
705 <     * Awaits the phase of the barrier to advance from the given phase
705 >     * Awaits the phase of this phaser to advance from the given phase
706       * value, throwing {@code InterruptedException} if interrupted
707 <     * while waiting, or returning immediately if the current phase of
708 <     * the barrier is not equal to the given phase value or this
709 <     * barrier is terminated. It is an unenforced usage error for an
603 <     * unregistered party to invoke this method.
707 >     * while waiting, or returning immediately if the current phase is
708 >     * not equal to the given phase value or this phaser is
709 >     * terminated.
710       *
711       * @param phase an arrival phase number, or negative value if
712       * terminated; this argument is normally the value returned by a
713 <     * previous call to {@code arrive} or its variants
714 <     * @return the next arrival phase number, or a negative value
715 <     * if terminated or argument is negative
713 >     * previous call to {@code arrive} or {@code arriveAndDeregister}.
714 >     * @return the next arrival phase number, or the argument if it is
715 >     * negative, or the (negative) {@linkplain #getPhase() current phase}
716 >     * if terminated
717       * @throws InterruptedException if thread interrupted while waiting
718       */
719      public int awaitAdvanceInterruptibly(int phase)
720          throws InterruptedException {
721 +        final Phaser root = this.root;
722 +        long s = (root == this) ? state : reconcileState();
723 +        int p = (int)(s >>> PHASE_SHIFT);
724          if (phase < 0)
725              return phase;
726 <        int p = getPhase();
727 <        if (p != phase)
728 <            return p;
729 <        return interruptibleWait(phase);
726 >        if (p == phase) {
727 >            QNode node = new QNode(this, phase, true, false, 0L);
728 >            p = root.internalAwaitAdvance(phase, node);
729 >            if (node.wasInterrupted)
730 >                throw new InterruptedException();
731 >        }
732 >        return p;
733      }
734  
735      /**
736 <     * Awaits the phase of the barrier to advance from the given phase
736 >     * Awaits the phase of this phaser to advance from the given phase
737       * value or the given timeout to elapse, throwing {@code
738       * InterruptedException} if interrupted while waiting, or
739 <     * returning immediately if the current phase of the barrier is
740 <     * not equal to the given phase value or this barrier is
628 <     * terminated.  It is an unenforced usage error for an
629 <     * unregistered party to invoke this method.
739 >     * returning immediately if the current phase is not equal to the
740 >     * given phase value or this phaser is terminated.
741       *
742       * @param phase an arrival phase number, or negative value if
743       * terminated; this argument is normally the value returned by a
744 <     * previous call to {@code arrive} or its variants
744 >     * previous call to {@code arrive} or {@code arriveAndDeregister}.
745       * @param timeout how long to wait before giving up, in units of
746       *        {@code unit}
747       * @param unit a {@code TimeUnit} determining how to interpret the
748       *        {@code timeout} parameter
749 <     * @return the next arrival phase number, or a negative value
750 <     * if terminated or argument is negative
749 >     * @return the next arrival phase number, or the argument if it is
750 >     * negative, or the (negative) {@linkplain #getPhase() current phase}
751 >     * if terminated
752       * @throws InterruptedException if thread interrupted while waiting
753       * @throws TimeoutException if timed out while waiting
754       */
# Line 644 | Line 756 | public class Phaser {
756                                           long timeout, TimeUnit unit)
757          throws InterruptedException, TimeoutException {
758          long nanos = unit.toNanos(timeout);
759 +        final Phaser root = this.root;
760 +        long s = (root == this) ? state : reconcileState();
761 +        int p = (int)(s >>> PHASE_SHIFT);
762          if (phase < 0)
763              return phase;
764 <        int p = getPhase();
765 <        if (p != phase)
766 <            return p;
767 <        return timedWait(phase, nanos);
764 >        if (p == phase) {
765 >            QNode node = new QNode(this, phase, true, true, nanos);
766 >            p = root.internalAwaitAdvance(phase, node);
767 >            if (node.wasInterrupted)
768 >                throw new InterruptedException();
769 >            else if (p == phase)
770 >                throw new TimeoutException();
771 >        }
772 >        return p;
773      }
774  
775      /**
776 <     * Forces this barrier to enter termination state. Counts of
777 <     * arrived and registered parties are unaffected. If this phaser
778 <     * has a parent, it too is terminated. This method may be useful
779 <     * for coordinating recovery after one or more tasks encounter
776 >     * Forces this phaser to enter termination state.  Counts of
777 >     * registered parties are unaffected.  If this phaser is a member
778 >     * of a tiered set of phasers, then all of the phasers in the set
779 >     * are terminated.  If this phaser is already terminated, this
780 >     * method has no effect.  This method may be useful for
781 >     * coordinating recovery after one or more tasks encounter
782       * unexpected exceptions.
783       */
784      public void forceTermination() {
785 <        Phaser r = root;    // force at root then reconcile
785 >        // Only need to change root state
786 >        final Phaser root = this.root;
787          long s;
788 <        while (phaseOf(s = r.state) >= 0)
789 <            r.casState(s, stateFor(-1, partiesOf(s), unarrivedOf(s)));
790 <        reconcileState();
791 <        releaseWaiters(0);  // ensure wakeups on both queues
792 <        releaseWaiters(1);
788 >        while ((s = root.state) >= 0) {
789 >            if (UNSAFE.compareAndSwapLong(root, stateOffset,
790 >                                          s, s | TERMINATION_BIT)) {
791 >                // signal all threads
792 >                releaseWaiters(0); // Waiters on evenQ
793 >                releaseWaiters(1); // Waiters on oddQ
794 >                return;
795 >            }
796 >        }
797      }
798  
799      /**
800       * Returns the current phase number. The maximum phase number is
801       * {@code Integer.MAX_VALUE}, after which it restarts at
802 <     * zero. Upon termination, the phase number is negative.
802 >     * zero. Upon termination, the phase number is negative,
803 >     * in which case the prevailing phase prior to termination
804 >     * may be obtained via {@code getPhase() + Integer.MIN_VALUE}.
805       *
806       * @return the phase number, or a negative value if terminated
807       */
808      public final int getPhase() {
809 <        return phaseOf(getReconciledState());
809 >        return (int)(root.state >>> PHASE_SHIFT);
810      }
811  
812      /**
813 <     * Returns the number of parties registered at this barrier.
813 >     * Returns the number of parties registered at this phaser.
814       *
815       * @return the number of parties
816       */
817      public int getRegisteredParties() {
818 <        return partiesOf(getReconciledState());
818 >        return partiesOf(state);
819      }
820  
821      /**
822       * Returns the number of registered parties that have arrived at
823 <     * the current phase of this barrier.
823 >     * the current phase of this phaser. If this phaser has terminated,
824 >     * the returned value is meaningless and arbitrary.
825       *
826       * @return the number of arrived parties
827       */
828      public int getArrivedParties() {
829 <        return arrivedOf(getReconciledState());
829 >        return arrivedOf(reconcileState());
830      }
831  
832      /**
833       * Returns the number of registered parties that have not yet
834 <     * arrived at the current phase of this barrier.
834 >     * arrived at the current phase of this phaser. If this phaser has
835 >     * terminated, the returned value is meaningless and arbitrary.
836       *
837       * @return the number of unarrived parties
838       */
839      public int getUnarrivedParties() {
840 <        return unarrivedOf(getReconciledState());
840 >        return unarrivedOf(reconcileState());
841      }
842  
843      /**
# Line 729 | Line 860 | public class Phaser {
860      }
861  
862      /**
863 <     * Returns {@code true} if this barrier has been terminated.
863 >     * Returns {@code true} if this phaser has been terminated.
864       *
865 <     * @return {@code true} if this barrier has been terminated
865 >     * @return {@code true} if this phaser has been terminated
866       */
867      public boolean isTerminated() {
868 <        return getPhase() < 0;
868 >        return root.state < 0L;
869      }
870  
871      /**
872       * Overridable method to perform an action upon impending phase
873       * advance, and to control termination. This method is invoked
874 <     * upon arrival of the party tripping the barrier (when all other
874 >     * upon arrival of the party advancing this phaser (when all other
875       * waiting parties are dormant).  If this method returns {@code
876 <     * true}, then, rather than advance the phase number, this barrier
877 <     * will be set to a final termination state, and subsequent calls
878 <     * to {@link #isTerminated} will return true. Any (unchecked)
879 <     * Exception or Error thrown by an invocation of this method is
880 <     * propagated to the party attempting to trip the barrier, in
881 <     * which case no advance occurs.
876 >     * true}, this phaser will be set to a final termination state
877 >     * upon advance, and subsequent calls to {@link #isTerminated}
878 >     * will return true. Any (unchecked) Exception or Error thrown by
879 >     * an invocation of this method is propagated to the party
880 >     * attempting to advance this phaser, in which case no advance
881 >     * occurs.
882       *
883       * <p>The arguments to this method provide the state of the phaser
884 <     * prevailing for the current transition. (When called from within
885 <     * an implementation of {@code onAdvance} the values returned by
886 <     * methods such as {@code getPhase} may or may not reliably
887 <     * indicate the state to which this transition applies.)
888 <     *
889 <     * <p>The default version returns {@code true} when the number of
890 <     * registered parties is zero. Normally, overrides that arrange
891 <     * termination for other reasons should also preserve this
892 <     * property.
884 >     * prevailing for the current transition.  The effects of invoking
885 >     * arrival, registration, and waiting methods on this phaser from
886 >     * within {@code onAdvance} are unspecified and should not be
887 >     * relied on.
888 >     *
889 >     * <p>If this phaser is a member of a tiered set of phasers, then
890 >     * {@code onAdvance} is invoked only for its root phaser on each
891 >     * advance.
892 >     *
893 >     * <p>To support the most common use cases, the default
894 >     * implementation of this method returns {@code true} when the
895 >     * number of registered parties has become zero as the result of a
896 >     * party invoking {@code arriveAndDeregister}.  You can disable
897 >     * this behavior, thus enabling continuation upon future
898 >     * registrations, by overriding this method to always return
899 >     * {@code false}:
900 >     *
901 >     * <pre> {@code
902 >     * Phaser phaser = new Phaser() {
903 >     *   protected boolean onAdvance(int phase, int parties) { return false; }
904 >     * }}</pre>
905       *
906 <     * @param phase the phase number on entering the barrier
906 >     * @param phase the current phase number on entry to this method,
907 >     * before this phaser is advanced
908       * @param registeredParties the current number of registered parties
909 <     * @return {@code true} if this barrier should terminate
909 >     * @return {@code true} if this phaser should terminate
910       */
911      protected boolean onAdvance(int phase, int registeredParties) {
912 <        return registeredParties <= 0;
912 >        return registeredParties == 0;
913      }
914  
915      /**
# Line 775 | Line 919 | public class Phaser {
919       * followed by the number of registered parties, and {@code
920       * "arrived = "} followed by the number of arrived parties.
921       *
922 <     * @return a string identifying this barrier, as well as its state
922 >     * @return a string identifying this phaser, as well as its state
923       */
924      public String toString() {
925 <        long s = getReconciledState();
925 >        return stateToString(reconcileState());
926 >    }
927 >
928 >    /**
929 >     * Implementation of toString and string-based error messages
930 >     */
931 >    private String stateToString(long s) {
932          return super.toString() +
933              "[phase = " + phaseOf(s) +
934              " parties = " + partiesOf(s) +
935              " arrived = " + arrivedOf(s) + "]";
936      }
937  
938 <    // methods for waiting
938 >    // Waiting mechanics
939  
940      /**
941 <     * Wait nodes for Treiber stack representing wait queue
941 >     * Removes and signals threads from queue for phase.
942       */
943 <    static final class QNode implements ForkJoinPool.ManagedBlocker {
944 <        final Phaser phaser;
945 <        final int phase;
946 <        final long startTime;
947 <        final long nanos;
948 <        final boolean timed;
949 <        final boolean interruptible;
950 <        volatile boolean wasInterrupted = false;
951 <        volatile Thread thread; // nulled to cancel wait
802 <        QNode next;
803 <
804 <        QNode(Phaser phaser, int phase, boolean interruptible,
805 <              boolean timed, long startTime, long nanos) {
806 <            this.phaser = phaser;
807 <            this.phase = phase;
808 <            this.timed = timed;
809 <            this.interruptible = interruptible;
810 <            this.startTime = startTime;
811 <            this.nanos = nanos;
812 <            thread = Thread.currentThread();
813 <        }
814 <
815 <        public boolean isReleasable() {
816 <            return (thread == null ||
817 <                    phaser.getPhase() != phase ||
818 <                    (interruptible && wasInterrupted) ||
819 <                    (timed && (nanos - (System.nanoTime() - startTime)) <= 0));
820 <        }
821 <
822 <        public boolean block() {
823 <            if (Thread.interrupted()) {
824 <                wasInterrupted = true;
825 <                if (interruptible)
826 <                    return true;
827 <            }
828 <            if (!timed)
829 <                LockSupport.park(this);
830 <            else {
831 <                long waitTime = nanos - (System.nanoTime() - startTime);
832 <                if (waitTime <= 0)
833 <                    return true;
834 <                LockSupport.parkNanos(this, waitTime);
835 <            }
836 <            return isReleasable();
837 <        }
838 <
839 <        void signal() {
840 <            Thread t = thread;
841 <            if (t != null) {
842 <                thread = null;
943 >    private void releaseWaiters(int phase) {
944 >        QNode q;   // first element of queue
945 >        Thread t;  // its thread
946 >        AtomicReference<QNode> head = (phase & 1) == 0 ? evenQ : oddQ;
947 >        while ((q = head.get()) != null &&
948 >               q.phase != (int)(root.state >>> PHASE_SHIFT)) {
949 >            if (head.compareAndSet(q, q.next) &&
950 >                (t = q.thread) != null) {
951 >                q.thread = null;
952                  LockSupport.unpark(t);
953              }
954          }
846
847        boolean doWait() {
848            if (thread != null) {
849                try {
850                    ForkJoinPool.managedBlock(this);
851                } catch (InterruptedException ie) {
852                    wasInterrupted = true; // can't currently happen
853                }
854            }
855            return wasInterrupted;
856        }
955      }
956  
957      /**
958 <     * Removes and signals waiting threads from wait queue.
959 <     */
960 <    private void releaseWaiters(int phase) {
961 <        AtomicReference<QNode> head = queueFor(phase);
962 <        QNode q;
963 <        while ((q = head.get()) != null) {
964 <            if (head.compareAndSet(q, q.next))
965 <                q.signal();
958 >     * Variant of releaseWaiters that additionally tries to remove any
959 >     * nodes no longer waiting for advance due to timeout or
960 >     * interrupt. Currently, nodes are removed only if they are at
961 >     * head of queue, which suffices to reduce memory footprint in
962 >     * most usages.
963 >     *
964 >     * @return current phase on exit
965 >     */
966 >    private int abortWait(int phase) {
967 >        AtomicReference<QNode> head = (phase & 1) == 0 ? evenQ : oddQ;
968 >        for (;;) {
969 >            Thread t;
970 >            QNode q = head.get();
971 >            int p = (int)(root.state >>> PHASE_SHIFT);
972 >            if (q == null || ((t = q.thread) != null && q.phase == p))
973 >                return p;
974 >            if (head.compareAndSet(q, q.next) && t != null) {
975 >                q.thread = null;
976 >                LockSupport.unpark(t);
977 >            }
978          }
979      }
980  
981 +    /** The number of CPUs, for spin control */
982 +    private static final int NCPU = Runtime.getRuntime().availableProcessors();
983 +
984      /**
985 <     * Tries to enqueue given node in the appropriate wait queue.
986 <     *
987 <     * @return true if successful
985 >     * The number of times to spin before blocking while waiting for
986 >     * advance, per arrival while waiting. On multiprocessors, fully
987 >     * blocking and waking up a large number of threads all at once is
988 >     * usually a very slow process, so we use rechargeable spins to
989 >     * avoid it when threads regularly arrive: When a thread in
990 >     * internalAwaitAdvance notices another arrival before blocking,
991 >     * and there appear to be enough CPUs available, it spins
992 >     * SPINS_PER_ARRIVAL more times before blocking. The value trades
993 >     * off good-citizenship vs big unnecessary slowdowns.
994       */
995 <    private boolean tryEnqueue(QNode node) {
877 <        AtomicReference<QNode> head = queueFor(node.phase);
878 <        return head.compareAndSet(node.next = head.get(), node);
879 <    }
995 >    static final int SPINS_PER_ARRIVAL = (NCPU < 2) ? 1 : 1 << 8;
996  
997      /**
998 <     * Enqueues node and waits unless aborted or signalled.
998 >     * Possibly blocks and waits for phase to advance unless aborted.
999 >     * Call only on root phaser.
1000       *
1001 +     * @param phase current phase
1002 +     * @param node if non-null, the wait node to track interrupt and timeout;
1003 +     * if null, denotes noninterruptible wait
1004       * @return current phase
1005       */
1006 <    private int untimedWait(int phase) {
1007 <        QNode node = null;
1008 <        boolean queued = false;
1009 <        boolean interrupted = false;
1006 >    private int internalAwaitAdvance(int phase, QNode node) {
1007 >        // assert root == this;
1008 >        releaseWaiters(phase-1);          // ensure old queue clean
1009 >        boolean queued = false;           // true when node is enqueued
1010 >        int lastUnarrived = 0;            // to increase spins upon change
1011 >        int spins = SPINS_PER_ARRIVAL;
1012 >        long s;
1013          int p;
1014 <        while ((p = getPhase()) == phase) {
1015 <            if (Thread.interrupted())
1016 <                interrupted = true;
1017 <            else if (node == null)
1018 <                node = new QNode(this, phase, false, false, 0, 0);
1019 <            else if (!queued)
1020 <                queued = tryEnqueue(node);
1021 <            else if (node.doWait())
1022 <                interrupted = true;
1014 >        while ((p = (int)((s = state) >>> PHASE_SHIFT)) == phase) {
1015 >            if (node == null) {           // spinning in noninterruptible mode
1016 >                int unarrived = (int)s & UNARRIVED_MASK;
1017 >                if (unarrived != lastUnarrived &&
1018 >                    (lastUnarrived = unarrived) < NCPU)
1019 >                    spins += SPINS_PER_ARRIVAL;
1020 >                boolean interrupted = Thread.interrupted();
1021 >                if (interrupted || --spins < 0) { // need node to record intr
1022 >                    node = new QNode(this, phase, false, false, 0L);
1023 >                    node.wasInterrupted = interrupted;
1024 >                }
1025 >            }
1026 >            else if (node.isReleasable()) // done or aborted
1027 >                break;
1028 >            else if (!queued) {           // push onto queue
1029 >                AtomicReference<QNode> head = (phase & 1) == 0 ? evenQ : oddQ;
1030 >                QNode q = node.next = head.get();
1031 >                if ((q == null || q.phase == phase) &&
1032 >                    (int)(state >>> PHASE_SHIFT) == phase) // avoid stale enq
1033 >                    queued = head.compareAndSet(q, node);
1034 >            }
1035 >            else {
1036 >                try {
1037 >                    ForkJoinPool.managedBlock(node);
1038 >                } catch (InterruptedException ie) {
1039 >                    node.wasInterrupted = true;
1040 >                }
1041 >            }
1042 >        }
1043 >
1044 >        if (node != null) {
1045 >            if (node.thread != null)
1046 >                node.thread = null;       // avoid need for unpark()
1047 >            if (node.wasInterrupted && !node.interruptible)
1048 >                Thread.currentThread().interrupt();
1049 >            if (p == phase && (p = (int)(state >>> PHASE_SHIFT)) == phase)
1050 >                return abortWait(phase); // possibly clean up on abort
1051          }
901        if (node != null)
902            node.thread = null;
1052          releaseWaiters(phase);
904        if (interrupted)
905            Thread.currentThread().interrupt();
1053          return p;
1054      }
1055  
1056      /**
1057 <     * Interruptible version
911 <     * @return current phase
1057 >     * Wait nodes for Treiber stack representing wait queue
1058       */
1059 <    private int interruptibleWait(int phase) throws InterruptedException {
1060 <        QNode node = null;
1061 <        boolean queued = false;
1062 <        boolean interrupted = false;
1063 <        int p;
1064 <        while ((p = getPhase()) == phase && !interrupted) {
1065 <            if (Thread.interrupted())
1066 <                interrupted = true;
1067 <            else if (node == null)
1068 <                node = new QNode(this, phase, true, false, 0, 0);
923 <            else if (!queued)
924 <                queued = tryEnqueue(node);
925 <            else if (node.doWait())
926 <                interrupted = true;
927 <        }
928 <        if (node != null)
929 <            node.thread = null;
930 <        if (p != phase || (p = getPhase()) != phase)
931 <            releaseWaiters(phase);
932 <        if (interrupted)
933 <            throw new InterruptedException();
934 <        return p;
935 <    }
1059 >    static final class QNode implements ForkJoinPool.ManagedBlocker {
1060 >        final Phaser phaser;
1061 >        final int phase;
1062 >        final boolean interruptible;
1063 >        final boolean timed;
1064 >        boolean wasInterrupted;
1065 >        long nanos;
1066 >        long lastTime;
1067 >        volatile Thread thread; // nulled to cancel wait
1068 >        QNode next;
1069  
1070 <    /**
1071 <     * Timeout version.
1072 <     * @return current phase
1073 <     */
1074 <    private int timedWait(int phase, long nanos)
1075 <        throws InterruptedException, TimeoutException {
1076 <        long startTime = System.nanoTime();
1077 <        QNode node = null;
1078 <        boolean queued = false;
1079 <        boolean interrupted = false;
1080 <        int p;
1081 <        while ((p = getPhase()) == phase && !interrupted) {
1070 >        QNode(Phaser phaser, int phase, boolean interruptible,
1071 >              boolean timed, long nanos) {
1072 >            this.phaser = phaser;
1073 >            this.phase = phase;
1074 >            this.interruptible = interruptible;
1075 >            this.nanos = nanos;
1076 >            this.timed = timed;
1077 >            this.lastTime = timed ? System.nanoTime() : 0L;
1078 >            thread = Thread.currentThread();
1079 >        }
1080 >
1081 >        public boolean isReleasable() {
1082 >            if (thread == null)
1083 >                return true;
1084 >            if (phaser.getPhase() != phase) {
1085 >                thread = null;
1086 >                return true;
1087 >            }
1088              if (Thread.interrupted())
1089 <                interrupted = true;
1090 <            else if (nanos - (System.nanoTime() - startTime) <= 0)
1091 <                break;
1092 <            else if (node == null)
1093 <                node = new QNode(this, phase, true, true, startTime, nanos);
1094 <            else if (!queued)
1095 <                queued = tryEnqueue(node);
1096 <            else if (node.doWait())
1097 <                interrupted = true;
1098 <        }
1099 <        if (node != null)
1100 <            node.thread = null;
1101 <        if (p != phase || (p = getPhase()) != phase)
1102 <            releaseWaiters(phase);
1103 <        if (interrupted)
1104 <            throw new InterruptedException();
1105 <        if (p == phase)
1106 <            throw new TimeoutException();
1107 <        return p;
1089 >                wasInterrupted = true;
1090 >            if (wasInterrupted && interruptible) {
1091 >                thread = null;
1092 >                return true;
1093 >            }
1094 >            if (timed) {
1095 >                if (nanos > 0L) {
1096 >                    long now = System.nanoTime();
1097 >                    nanos -= now - lastTime;
1098 >                    lastTime = now;
1099 >                }
1100 >                if (nanos <= 0L) {
1101 >                    thread = null;
1102 >                    return true;
1103 >                }
1104 >            }
1105 >            return false;
1106 >        }
1107 >
1108 >        public boolean block() {
1109 >            if (isReleasable())
1110 >                return true;
1111 >            else if (!timed)
1112 >                LockSupport.park(this);
1113 >            else if (nanos > 0)
1114 >                LockSupport.parkNanos(this, nanos);
1115 >            return isReleasable();
1116 >        }
1117      }
1118  
1119      // Unsafe mechanics
1120  
1121 <    private static final sun.misc.Unsafe UNSAFE = getUnsafe();
1122 <    private static final long stateOffset =
1123 <        objectFieldOffset("state", Phaser.class);
976 <
977 <    private final boolean casState(long cmp, long val) {
978 <        return UNSAFE.compareAndSwapLong(this, stateOffset, cmp, val);
979 <    }
980 <
981 <    private static long objectFieldOffset(String field, Class<?> klazz) {
1121 >    private static final sun.misc.Unsafe UNSAFE;
1122 >    private static final long stateOffset;
1123 >    static {
1124          try {
1125 <            return UNSAFE.objectFieldOffset(klazz.getDeclaredField(field));
1126 <        } catch (NoSuchFieldException e) {
1127 <            // Convert Exception to corresponding Error
1128 <            NoSuchFieldError error = new NoSuchFieldError(field);
1129 <            error.initCause(e);
1130 <            throw error;
1125 >            UNSAFE = getUnsafe();
1126 >            Class<?> k = Phaser.class;
1127 >            stateOffset = UNSAFE.objectFieldOffset
1128 >                (k.getDeclaredField("state"));
1129 >        } catch (Exception e) {
1130 >            throw new Error(e);
1131          }
1132      }
1133  
# Line 999 | Line 1141 | public class Phaser {
1141      private static sun.misc.Unsafe getUnsafe() {
1142          try {
1143              return sun.misc.Unsafe.getUnsafe();
1144 <        } catch (SecurityException se) {
1145 <            try {
1146 <                return java.security.AccessController.doPrivileged
1147 <                    (new java.security
1148 <                     .PrivilegedExceptionAction<sun.misc.Unsafe>() {
1149 <                        public sun.misc.Unsafe run() throws Exception {
1150 <                            java.lang.reflect.Field f = sun.misc
1151 <                                .Unsafe.class.getDeclaredField("theUnsafe");
1152 <                            f.setAccessible(true);
1153 <                            return (sun.misc.Unsafe) f.get(null);
1154 <                        }});
1155 <            } catch (java.security.PrivilegedActionException e) {
1156 <                throw new RuntimeException("Could not initialize intrinsics",
1157 <                                           e.getCause());
1158 <            }
1144 >        } catch (SecurityException tryReflectionInstead) {}
1145 >        try {
1146 >            return java.security.AccessController.doPrivileged
1147 >            (new java.security.PrivilegedExceptionAction<sun.misc.Unsafe>() {
1148 >                public sun.misc.Unsafe run() throws Exception {
1149 >                    Class<sun.misc.Unsafe> k = sun.misc.Unsafe.class;
1150 >                    for (java.lang.reflect.Field f : k.getDeclaredFields()) {
1151 >                        f.setAccessible(true);
1152 >                        Object x = f.get(null);
1153 >                        if (k.isInstance(x))
1154 >                            return k.cast(x);
1155 >                    }
1156 >                    throw new NoSuchFieldError("the Unsafe");
1157 >                }});
1158 >        } catch (java.security.PrivilegedActionException e) {
1159 >            throw new RuntimeException("Could not initialize intrinsics",
1160 >                                       e.getCause());
1161          }
1162      }
1163   }

Diff Legend

Removed lines
+ Added lines
< Changed lines
> Changed lines