--- jsr166/src/jsr166y/Phaser.java 2008/09/06 13:19:17 1.4 +++ jsr166/src/jsr166y/Phaser.java 2010/11/13 05:59:25 1.53 @@ -5,173 +5,216 @@ */ package jsr166y; -import java.util.concurrent.*; -import java.util.concurrent.atomic.*; + +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; +import java.util.concurrent.atomic.AtomicReference; import java.util.concurrent.locks.LockSupport; -import sun.misc.Unsafe; -import java.lang.reflect.*; /** - * A reusable synchronization barrier, similar in functionality to a - * {@link java.util.concurrent.CyclicBarrier} and {@link - * java.util.concurrent.CountDownLatch} but supporting more flexible - * usage. + * A reusable synchronization barrier, similar in functionality to + * {@link java.util.concurrent.CyclicBarrier CyclicBarrier} and + * {@link java.util.concurrent.CountDownLatch CountDownLatch} + * but supporting more flexible usage. + * + *

Registration. Unlike the case for other barriers, the + * number of parties registered to synchronize on a phaser + * may vary over time. Tasks may be registered at any time (using + * methods {@link #register}, {@link #bulkRegister}, or forms of + * constructors establishing initial numbers of parties), and + * optionally deregistered upon any arrival (using {@link + * #arriveAndDeregister}). As is the case with most basic + * synchronization constructs, registration and deregistration affect + * only internal counts; they do not establish any further internal + * bookkeeping, so tasks cannot query whether they are registered. + * (However, you can introduce such bookkeeping by subclassing this + * class.) + * + *

Synchronization. Like a {@code CyclicBarrier}, a {@code + * Phaser} may be repeatedly awaited. Method {@link + * #arriveAndAwaitAdvance} has effect analogous to {@link + * java.util.concurrent.CyclicBarrier#await CyclicBarrier.await}. Each + * generation of a {@code Phaser} has an associated phase number. The + * phase number starts at zero, and advances when all parties arrive + * at the barrier, wrapping around to zero after reaching {@code + * Integer.MAX_VALUE}. The use of phase numbers enables independent + * control of actions upon arrival at a barrier and upon awaiting + * others, via two kinds of methods that may be invoked by any + * registered party: * *

+ * synchronization contention costs may instead be set up so that + * groups of sub-phasers share a common parent. This may greatly + * increase throughput even though it incurs greater per-operation + * overhead. + * + *

Monitoring. While synchronization methods may be invoked + * only by registered parties, the current state of a phaser may be + * monitored by any caller. At any given moment there are {@link + * #getRegisteredParties} parties in total, of which {@link + * #getArrivedParties} have arrived at the current phase ({@link + * #getPhase}). When the remaining ({@link #getUnarrivedParties}) + * parties arrive, the phase advances. The values returned by these + * methods may reflect transient states and so are not in general + * useful for synchronization control. Method {@link #toString} + * returns snapshots of these state queries in a form convenient for + * informal monitoring. * *

Sample usages: * - *

A Phaser may be used instead of a CountdownLatch to control - * a one-shot action serving a variable number of parties. The typical - * idiom is for the method setting this up to first register, then - * start the actions, then deregister, as in: - * - * 

- *  void runTasks(List<Runnable> list) {
- *    final Phaser phaser = new Phaser(1); // "1" to register self
- *    for (Runnable r : list) {
- *      phaser.register();
- *      new Thread() {
- *        public void run() {
- *          phaser.arriveAndAwaitAdvance(); // await all creation
- *          r.run();
- *          phaser.arriveAndDeregister();   // signal completion
- *        }
- *      }.start();
+ * 
A {@code Phaser} may be used instead of a {@code CountDownLatch}
+ * to control a one-shot action serving a variable number of parties.
+ * The typical idiom is for the method setting this up to first
+ * register, then start the actions, then deregister, as in:
+ *
+ *  
 {@code
+ * void runTasks(List tasks) {
+ *   final Phaser phaser = new Phaser(1); // "1" to register self
+ *   // create and start threads
+ *   for (Runnable task : tasks) {
+ *     phaser.register();
+ *     new Thread() {
+ *       public void run() {
+ *         phaser.arriveAndAwaitAdvance(); // await all creation
+ *         task.run();
+ *       }
+ *     }.start();
  *   }
- *   phaser.arrive(); // allow threads to start
- *   int p = phaser.arriveAndDeregister(); // deregister self
- *   otherActions(); // do other things while tasks execute
- *   phaser.awaitAdvance(p); // wait for all tasks to arrive
- * }
- * 

+ *
+ *   // allow threads to start and deregister self
+ *   phaser.arriveAndDeregister();
+ * }}
* *

One way to cause a set of threads to repeatedly perform actions - * for a given number of iterations is to override onAdvance: + * for a given number of iterations is to override {@code onAdvance}: * - * 

- *  void startTasks(List<Runnable> list, final int iterations) {
- *    final Phaser phaser = new Phaser() {
- *       public boolean onAdvance(int phase, int registeredParties) {
- *         return phase >= iterations || registeredParties == 0;
+ *  
 {@code
+ * void startTasks(List tasks, final int iterations) {
+ *   final Phaser phaser = new Phaser() {
+ *     protected boolean onAdvance(int phase, int registeredParties) {
+ *       return phase >= iterations || registeredParties == 0;
+ *     }
+ *   };
+ *   phaser.register();
+ *   for (final Runnable task : tasks) {
+ *     phaser.register();
+ *     new Thread() {
+ *       public void run() {
+ *         do {
+ *           task.run();
+ *           phaser.arriveAndAwaitAdvance();
+ *         } while (!phaser.isTerminated());
  *       }
- *    };
- *    phaser.register();
- *    for (Runnable r : list) {
- *      phaser.register();
- *      new Thread() {
- *        public void run() {
- *           do {
- *             r.run();
- *             phaser.arriveAndAwaitAdvance();
- *           } while(!phaser.isTerminated();
- *        }
- *      }.start();
+ *     }.start();
  *   }
  *   phaser.arriveAndDeregister(); // deregister self, don't wait
- * }
- * 

+ * }}
* - *

To create a set of tasks using a tree of Phasers, + * If the main task must later await termination, it + * may re-register and then execute a similar loop: + * 

 {@code
+ *   // ...
+ *   phaser.register();
+ *   while (!phaser.isTerminated())
+ *     phaser.arriveAndAwaitAdvance();}
+ * + *

Related constructions may be used to await particular phase numbers + * in contexts where you are sure that the phase will never wrap around + * {@code Integer.MAX_VALUE}. For example: + * + * 

 {@code
+ * void awaitPhase(Phaser phaser, int phase) {
+ *   int p = phaser.register(); // assumes caller not already registered
+ *   while (p < phase) {
+ *     if (phaser.isTerminated())
+ *       // ... deal with unexpected termination
+ *     else
+ *       p = phaser.arriveAndAwaitAdvance();
+ *   }
+ *   phaser.arriveAndDeregister();
+ * }}
+ * + * + *

To create a set of tasks using a tree of phasers, * you could use code of the following form, assuming a - * Task class with a constructor accepting a Phaser that - * it registers for upon construction: - * 

- *  void build(Task[] actions, int lo, int hi, Phaser b) {
- *    int step = (hi - lo) / TASKS_PER_PHASER;
- *    if (step > 1) {
- *       int i = lo;
- *       while (i < hi) {
- *         int r = Math.min(i + step, hi);
- *         build(actions, i, r, new Phaser(b));
- *         i = r;
- *       }
- *    }
- *    else {
- *      for (int i = lo; i < hi; ++i)
- *        actions[i] = new Task(b);
- *        // assumes new Task(b) performs b.register()
- *    }
- *  }
- *  // .. initially called, for n tasks via
- *  build(new Task[n], 0, n, new Phaser());
- *
+ * Task class with a constructor accepting a phaser that + * it registers with upon construction: * - * The best value of TASKS_PER_PHASER depends mainly on + * 
 {@code
+ * void build(Task[] actions, int lo, int hi, Phaser ph) {
+ *   if (hi - lo > TASKS_PER_PHASER) {
+ *     for (int i = lo; i < hi; i += TASKS_PER_PHASER) {
+ *       int j = Math.min(i + TASKS_PER_PHASER, hi);
+ *       build(actions, i, j, new Phaser(ph));
+ *     }
+ *   } else {
+ *     for (int i = lo; i < hi; ++i)
+ *       actions[i] = new Task(ph);
+ *       // assumes new Task(ph) performs ph.register()
+ *   }
+ * }
+ * // .. initially called, for n tasks via
+ * build(new Task[n], 0, n, new Phaser());}
+ * + * The best value of {@code TASKS_PER_PHASER} depends mainly on * expected barrier synchronization rates. A value as low as four may * be appropriate for extremely small per-barrier task bodies (thus * high rates), or up to hundreds for extremely large ones. * - * - * *

Implementation notes: This implementation restricts the * maximum number of parties to 65535. Attempts to register additional - * parties result in IllegalStateExceptions. However, you can and + * parties result in {@code IllegalStateException}. However, you can and * should create tiered phasers to accommodate arbitrarily large sets * of participants. + * + * @since 1.7 + * @author Doug Lea */ public class Phaser { /* @@ -184,179 +227,265 @@ public class Phaser { * Barrier state representation. Conceptually, a barrier contains * four values: * - * * parties -- the number of parties to wait (16 bits) - * * unarrived -- the number of parties yet to hit barrier (16 bits) - * * phase -- the generation of the barrier (31 bits) - * * terminated -- set if barrier is terminated (1 bit) + * * unarrived -- the number of parties yet to hit barrier (bits 0-15) + * * parties -- the number of parties to wait (bits 16-31) + * * phase -- the generation of the barrier (bits 32-62) + * * terminated -- set if barrier is terminated (bit 63 / sign) * * However, to efficiently maintain atomicity, these values are * packed into a single (atomic) long. Termination uses the sign * bit of 32 bit representation of phase, so phase is set to -1 on - * termination. Good performace relies on keeping state decoding + * termination. Good performance relies on keeping state decoding * and encoding simple, and keeping race windows short. - * - * Note: there are some cheats in arrive() that rely on unarrived - * being lowest 16 bits. */ private volatile long state; - private static final int ushortBits = 16; - private static final int ushortMask = (1 << ushortBits) - 1; - private static final int phaseMask = 0x7fffffff; + private static final int MAX_COUNT = 0xffff; + private static final int MAX_PHASE = 0x7fffffff; + private static final int PARTIES_SHIFT = 16; + private static final int PHASE_SHIFT = 32; + private static final long UNARRIVED_MASK = 0xffffL; + private static final long PARTIES_MASK = 0xffff0000L; + private static final long ONE_ARRIVAL = 1L; + private static final long ONE_PARTY = 1L << PARTIES_SHIFT; + private static final long TERMINATION_PHASE = -1L << PHASE_SHIFT; + + // The following unpacking methods are usually manually inlined private static int unarrivedOf(long s) { - return (int)(s & ushortMask); + return (int) (s & UNARRIVED_MASK); } private static int partiesOf(long s) { - return (int)(s & (ushortMask << 16)) >>> 16; + return ((int) (s & PARTIES_MASK)) >>> PARTIES_SHIFT; } private static int phaseOf(long s) { - return (int)(s >>> 32); + return (int) (s >>> PHASE_SHIFT); } private static int arrivedOf(long s) { return partiesOf(s) - unarrivedOf(s); } - private static long stateFor(int phase, int parties, int unarrived) { - return (((long)phase) << 32) | ((parties << 16) | unarrived); - } - - private static long trippedStateFor(int phase, int parties) { - return (((long)phase) << 32) | ((parties << 16) | parties); - } - - private static IllegalStateException badBounds(int parties, int unarrived) { - return new IllegalStateException - ("Attempt to set " + unarrived + - " unarrived of " + parties + " parties"); - } - /** * The parent of this phaser, or null if none */ private final Phaser parent; /** - * The root of Phaser tree. Equals this if not in a tree. Used to + * The root of phaser tree. Equals this if not in a tree. Used to * support faster state push-down. */ private final Phaser root; - // Wait queues - /** - * Heads of Treiber stacks waiting for nonFJ threads. To eliminate - * contention while releasing some threads while adding others, we + * Heads of Treiber stacks for waiting threads. To eliminate + * contention when releasing some threads while adding others, we * use two of them, alternating across even and odd phases. + * Subphasers share queues with root to speed up releases. */ - private final AtomicReference evenQ = new AtomicReference(); - private final AtomicReference oddQ = new AtomicReference(); + private final AtomicReference evenQ; + private final AtomicReference oddQ; private AtomicReference queueFor(int phase) { - return (phase & 1) == 0? evenQ : oddQ; + return ((phase & 1) == 0) ? evenQ : oddQ; + } + + /** + * Main implementation for methods arrive and arriveAndDeregister. + * Manually tuned to speed up and minimize race windows for the + * common case of just decrementing unarrived field. + * + * @param adj - adjustment to apply to state -- either + * ONE_ARRIVAL (for arrive) or + * ONE_ARRIVAL|ONE_PARTY (for arriveAndDeregister) + */ + private int doArrive(long adj) { + long s; + int phase, unarrived; + while ((phase = (int)((s = state) >>> PHASE_SHIFT)) >= 0) { + if ((unarrived = (int)(s & UNARRIVED_MASK)) != 0) { + if (UNSAFE.compareAndSwapLong(this, stateOffset, s, s -= adj)) { + if (unarrived == 1) { + Phaser par; + long p = s & PARTIES_MASK; // unshifted parties field + long lu = p >>> PARTIES_SHIFT; + int u = (int)lu; + int nextPhase = (phase + 1) & MAX_PHASE; + long next = ((long)nextPhase << PHASE_SHIFT) | p | lu; + if ((par = parent) == null) { + UNSAFE.compareAndSwapLong + (this, stateOffset, s, onAdvance(phase, u)? + next | TERMINATION_PHASE : next); + releaseWaiters(phase); + } + else { + par.doArrive(u == 0? + ONE_ARRIVAL|ONE_PARTY : ONE_ARRIVAL); + if ((int)(par.state >>> PHASE_SHIFT) != nextPhase || + ((int)(state >>> PHASE_SHIFT) != nextPhase && + !UNSAFE.compareAndSwapLong(this, stateOffset, + s, next))) + reconcileState(); + } + } + break; + } + } + else if (state == s && reconcileState() == s) // recheck + throw new IllegalStateException(badArrive()); + } + return phase; + } + + /** + * Returns message string for bounds exceptions on arrival. + * Declared out of-line from doArrive to reduce string op bulk. + */ + private String badArrive() { + return ("Attempted arrival of unregistered party for " + + this.toString()); + } + + /** + * Implementation of register, bulkRegister + * + * @param registrations number to add to both parties and unarrived fields + */ + private int doRegister(int registrations) { + long adj = (long)registrations; // adjustment to state + adj |= adj << PARTIES_SHIFT; + Phaser par = parent; + long s; + int phase; + while ((phase = (int)((s = (par == null? state : reconcileState())) + >>> PHASE_SHIFT)) >= 0) { + int parties = ((int)(s & PARTIES_MASK)) >>> PARTIES_SHIFT; + if (parties != 0 && (s & UNARRIVED_MASK) == 0) + internalAwaitAdvance(phase, null); // wait for onAdvance + else if (parties + registrations > MAX_COUNT) + throw new IllegalStateException(badRegister()); + else if (UNSAFE.compareAndSwapLong(this, stateOffset, s, s + adj)) + break; + } + return phase; } /** - * Returns current state, first resolving lagged propagation from - * root if necessary. + * Returns message string for bounds exceptions on registration */ - private long getReconciledState() { - return parent == null? state : reconcileState(); + private String badRegister() { + return ("Attempt to register more than " + MAX_COUNT + " parties for "+ + this.toString()); } /** - * Recursively resolves state. + * Recursively resolves lagged phase propagation from root if + * necessary. */ private long reconcileState() { - Phaser p = parent; - long s = state; - if (p != null) { - while (unarrivedOf(s) == 0 && phaseOf(s) != phaseOf(root.state)) { - long parentState = p.getReconciledState(); - int parentPhase = phaseOf(parentState); - int phase = phaseOf(s = state); - if (phase != parentPhase) { - long next = trippedStateFor(parentPhase, partiesOf(s)); - if (casState(s, next)) { - releaseWaiters(phase); - s = next; - } + Phaser par = parent; + if (par == null) + return state; + Phaser rt = root; + long s; + int phase, rPhase; + while ((phase = (int)((s = state) >>> PHASE_SHIFT)) >= 0 && + (rPhase = (int)(rt.state >>> PHASE_SHIFT)) != phase) { + if (rPhase < 0 || (s & UNARRIVED_MASK) == 0) { + long ps = par.parent == null? par.state : par.reconcileState(); + int pPhase = (int)(ps >>> PHASE_SHIFT); + if (pPhase < 0 || pPhase == ((phase + 1) & MAX_PHASE)) { + if (state != s) + continue; + long p = s & PARTIES_MASK; + long next = ((((long) pPhase) << PHASE_SHIFT) | + (p >>> PARTIES_SHIFT) | p); + if (UNSAFE.compareAndSwapLong(this, stateOffset, s, next)) + return next; } } + if (state == s) + releaseWaiters(phase); // help release others } return s; } /** - * Creates a new Phaser without any initially registered parties, - * initial phase number 0, and no parent. + * Creates a new phaser without any initially registered parties, + * initial phase number 0, and no parent. Any thread using this + * phaser will need to first register for it. */ public Phaser() { - this(null); + this(null, 0); } /** - * Creates a new Phaser with the given numbers of registered + * Creates a new phaser with the given number of registered * unarrived parties, initial phase number 0, and no parent. - * @param parties the number of parties required to trip barrier. + * + * @param parties the number of parties required to trip barrier * @throws IllegalArgumentException if parties less than zero - * or greater than the maximum number of parties supported. + * or greater than the maximum number of parties supported */ public Phaser(int parties) { this(null, parties); } /** - * Creates a new Phaser with the given parent, without any + * Creates a new phaser with the given parent, without any * initially registered parties. If parent is non-null this phaser * is registered with the parent and its initial phase number is * the same as that of parent phaser. - * @param parent the parent phaser. + * + * @param parent the parent phaser */ public Phaser(Phaser parent) { - int phase = 0; - this.parent = parent; - if (parent != null) { - this.root = parent.root; - phase = parent.register(); - } - else - this.root = this; - this.state = trippedStateFor(phase, 0); + this(parent, 0); } /** - * Creates a new Phaser with the given parent and numbers of - * registered unarrived parties. If parent is non-null this phaser + * Creates a new phaser with the given parent and number of + * registered unarrived parties. If parent is non-null, this phaser * is registered with the parent and its initial phase number is * the same as that of parent phaser. - * @param parent the parent phaser. - * @param parties the number of parties required to trip barrier. + * + * @param parent the parent phaser + * @param parties the number of parties required to trip barrier * @throws IllegalArgumentException if parties less than zero - * or greater than the maximum number of parties supported. + * or greater than the maximum number of parties supported */ public Phaser(Phaser parent, int parties) { - if (parties < 0 || parties > ushortMask) + if (parties < 0 || parties > MAX_COUNT) throw new IllegalArgumentException("Illegal number of parties"); - int phase = 0; + int phase; this.parent = parent; if (parent != null) { - this.root = parent.root; + Phaser r = parent.root; + this.root = r; + this.evenQ = r.evenQ; + this.oddQ = r.oddQ; phase = parent.register(); } - else + else { this.root = this; - this.state = trippedStateFor(phase, parties); + this.evenQ = new AtomicReference(); + this.oddQ = new AtomicReference(); + phase = 0; + } + long p = (long)parties; + this.state = (((long) phase) << PHASE_SHIFT) | p | (p << PARTIES_SHIFT); } /** * Adds a new unarrived party to this phaser. - * @return the current barrier phase number upon registration + * If an ongoing invocation of {@link #onAdvance} is in progress, + * this method may wait until its completion before registering. + * + * @return the arrival phase number to which this registration applied * @throws IllegalStateException if attempting to register more - * than the maximum supported number of parties. + * than the maximum supported number of parties */ public int register() { return doRegister(1); @@ -364,219 +493,160 @@ public class Phaser { /** * Adds the given number of new unarrived parties to this phaser. - * @param parties the number of parties required to trip barrier. - * @return the current barrier phase number upon registration + * If an ongoing invocation of {@link #onAdvance} is in progress, + * this method may wait until its completion before registering. + * + * @param parties the number of additional parties required to trip barrier + * @return the arrival phase number to which this registration applied * @throws IllegalStateException if attempting to register more - * than the maximum supported number of parties. + * than the maximum supported number of parties + * @throws IllegalArgumentException if {@code parties < 0} */ public int bulkRegister(int parties) { if (parties < 0) throw new IllegalArgumentException(); + if (parties > MAX_COUNT) + throw new IllegalStateException(badRegister()); if (parties == 0) return getPhase(); return doRegister(parties); } /** - * Shared code for register, bulkRegister - */ - private int doRegister(int registrations) { - int phase; - for (;;) { - long s = getReconciledState(); - phase = phaseOf(s); - int unarrived = unarrivedOf(s) + registrations; - int parties = partiesOf(s) + registrations; - if (phase < 0) - break; - if (parties > ushortMask || unarrived > ushortMask) - throw badBounds(parties, unarrived); - if (phase == phaseOf(root.state) && - casState(s, stateFor(phase, parties, unarrived))) - break; - } - return phase; - } - - /** * Arrives at the barrier, but does not wait for others. (You can - * in turn wait for others via {@link #awaitAdvance}). + * in turn wait for others via {@link #awaitAdvance}). It is an + * unenforced usage error for an unregistered party to invoke this + * method. * - * @return the barrier phase number upon entry to this method, or a - * negative value if terminated; + * @return the arrival phase number, or a negative value if terminated * @throws IllegalStateException if not terminated and the number - * of unarrived parties would become negative. + * of unarrived parties would become negative */ public int arrive() { - int phase; - for (;;) { - long s = state; - phase = phaseOf(s); - int parties = partiesOf(s); - int unarrived = unarrivedOf(s) - 1; - if (unarrived > 0) { // Not the last arrival - if (casState(s, s - 1)) // s-1 adds one arrival - break; - } - else if (unarrived == 0) { // the last arrival - Phaser par = parent; - if (par == null) { // directly trip - if (casState - (s, - trippedStateFor(onAdvance(phase, parties)? -1 : - ((phase + 1) & phaseMask), parties))) { - releaseWaiters(phase); - break; - } - } - else { // cascade to parent - if (casState(s, s - 1)) { // zeroes unarrived - par.arrive(); - reconcileState(); - break; - } - } - } - else if (phase < 0) // Don't throw exception if terminated - break; - else if (phase != phaseOf(root.state)) // or if unreconciled - reconcileState(); - else - throw badBounds(parties, unarrived); - } - return phase; + return doArrive(ONE_ARRIVAL); } /** - * Arrives at the barrier, and deregisters from it, without - * waiting for others. Deregistration reduces number of parties + * Arrives at the barrier and deregisters from it without waiting + * for others. Deregistration reduces the number of parties * required to trip the barrier in future phases. If this phaser * has a parent, and deregistration causes this phaser to have - * zero parties, this phaser is also deregistered from its parent. + * zero parties, this phaser also arrives at and is deregistered + * from its parent. It is an unenforced usage error for an + * unregistered party to invoke this method. * - * @return the current barrier phase number upon entry to - * this method, or a negative value if terminated; + * @return the arrival phase number, or a negative value if terminated * @throws IllegalStateException if not terminated and the number - * of registered or unarrived parties would become negative. + * of registered or unarrived parties would become negative */ public int arriveAndDeregister() { - // similar code to arrive, but too different to merge - Phaser par = parent; - int phase; - for (;;) { - long s = state; - phase = phaseOf(s); - int parties = partiesOf(s) - 1; - int unarrived = unarrivedOf(s) - 1; - if (parties >= 0) { - if (unarrived > 0 || (unarrived == 0 && par != null)) { - if (casState - (s, - stateFor(phase, parties, unarrived))) { - if (unarrived == 0) { - par.arriveAndDeregister(); - reconcileState(); - } - break; - } - continue; - } - if (unarrived == 0) { - if (casState - (s, - trippedStateFor(onAdvance(phase, parties)? -1 : - ((phase + 1) & phaseMask), parties))) { - releaseWaiters(phase); - break; - } - continue; - } - if (phase < 0) - break; - if (par != null && phase != phaseOf(root.state)) { - reconcileState(); - continue; - } - } - throw badBounds(parties, unarrived); - } - return phase; + return doArrive(ONE_ARRIVAL|ONE_PARTY); } /** * Arrives at the barrier and awaits others. Equivalent in effect - * to awaitAdvance(arrive()). If you instead need to - * await with interruption of timeout, and/or deregister upon - * arrival, you can arrange them using analogous constructions. - * @return the phase on entry to this method + * to {@code awaitAdvance(arrive())}. If you need to await with + * interruption or timeout, you can arrange this with an analogous + * construction using one of the other forms of the {@code + * awaitAdvance} method. If instead you need to deregister upon + * arrival, use {@link #arriveAndDeregister}. It is an unenforced + * usage error for an unregistered party to invoke this method. + * + * @return the arrival phase number, or a negative number if terminated * @throws IllegalStateException if not terminated and the number - * of unarrived parties would become negative. + * of unarrived parties would become negative */ public int arriveAndAwaitAdvance() { return awaitAdvance(arrive()); } /** - * Awaits the phase of the barrier to advance from the given - * value, or returns immediately if argument is negative or this - * barrier is terminated. - * @param phase the phase on entry to this method - * @return the phase on exit from this method + * Awaits the phase of the barrier to advance from the given phase + * value, returning immediately if the current phase of the + * barrier is not equal to the given phase value or this barrier + * is terminated. + * + * @param phase an arrival phase number, or negative value if + * terminated; this argument is normally the value returned by a + * previous call to {@code arrive} or its variants + * @return the next arrival phase number, or a negative value + * if terminated or argument is negative */ public int awaitAdvance(int phase) { if (phase < 0) return phase; - long s = getReconciledState(); - int p = phaseOf(s); + int p = (int)((parent==null? state : reconcileState()) >>> PHASE_SHIFT); if (p != phase) return p; - if (unarrivedOf(s) == 0) - parent.awaitAdvance(phase); - // Fall here even if parent waited, to reconcile and help release - return untimedWait(phase); + return internalAwaitAdvance(phase, null); } /** - * Awaits the phase of the barrier to advance from the given - * value, or returns immediately if argumet is negative or this - * barrier is terminated, or throws InterruptedException if - * interrupted while waiting. - * @param phase the phase on entry to this method - * @return the phase on exit from this method + * Awaits the phase of the barrier to advance from the given phase + * value, throwing {@code InterruptedException} if interrupted + * while waiting, or returning immediately if the current phase of + * the barrier is not equal to the given phase value or this + * barrier is terminated. + * + * @param phase an arrival phase number, or negative value if + * terminated; this argument is normally the value returned by a + * previous call to {@code arrive} or its variants + * @return the next arrival phase number, or a negative value + * if terminated or argument is negative * @throws InterruptedException if thread interrupted while waiting */ - public int awaitAdvanceInterruptibly(int phase) throws InterruptedException { + public int awaitAdvanceInterruptibly(int phase) + throws InterruptedException { if (phase < 0) return phase; - long s = getReconciledState(); - int p = phaseOf(s); + int p = (int)((parent==null? state : reconcileState()) >>> PHASE_SHIFT); if (p != phase) return p; - if (unarrivedOf(s) != 0) - parent.awaitAdvanceInterruptibly(phase); - return interruptibleWait(phase); + QNode node = new QNode(this, phase, true, false, 0L); + p = internalAwaitAdvance(phase, node); + if (node.wasInterrupted) + throw new InterruptedException(); + else + return p; } /** - * Awaits the phase of the barrier to advance from the given value - * or the given timeout elapses, or returns immediately if - * argument is negative or this barrier is terminated. - * @param phase the phase on entry to this method - * @return the phase on exit from this method + * Awaits the phase of the barrier to advance from the given phase + * value or the given timeout to elapse, throwing {@code + * InterruptedException} if interrupted while waiting, or + * returning immediately if the current phase of the barrier is + * not equal to the given phase value or this barrier is + * terminated. + * + * @param phase an arrival phase number, or negative value if + * terminated; this argument is normally the value returned by a + * previous call to {@code arrive} or its variants + * @param timeout how long to wait before giving up, in units of + * {@code unit} + * @param unit a {@code TimeUnit} determining how to interpret the + * {@code timeout} parameter + * @return the next arrival phase number, or a negative value + * if terminated or argument is negative * @throws InterruptedException if thread interrupted while waiting * @throws TimeoutException if timed out while waiting */ - public int awaitAdvanceInterruptibly(int phase, long timeout, TimeUnit unit) + public int awaitAdvanceInterruptibly(int phase, + long timeout, TimeUnit unit) throws InterruptedException, TimeoutException { + long nanos = unit.toNanos(timeout); if (phase < 0) return phase; - long s = getReconciledState(); - int p = phaseOf(s); + int p = (int)((parent==null? state : reconcileState()) >>> PHASE_SHIFT); if (p != phase) return p; - if (unarrivedOf(s) == 0) - parent.awaitAdvanceInterruptibly(phase, timeout, unit); - return timedWait(phase, unit.toNanos(timeout)); + QNode node = new QNode(this, phase, true, true, nanos); + p = internalAwaitAdvance(phase, node); + if (node.wasInterrupted) + throw new InterruptedException(); + else if (p == phase) + throw new TimeoutException(); + else + return p; } /** @@ -587,70 +657,59 @@ public class Phaser { * unexpected exceptions. */ public void forceTermination() { - for (;;) { - long s = getReconciledState(); - int phase = phaseOf(s); - int parties = partiesOf(s); - int unarrived = unarrivedOf(s); - if (phase < 0 || - casState(s, stateFor(-1, parties, unarrived))) { - releaseWaiters(0); - releaseWaiters(1); - if (parent != null) - parent.forceTermination(); - return; - } - } + Phaser r = root; // force at root then reconcile + long s; + while ((s = r.state) >= 0) + UNSAFE.compareAndSwapLong(r, stateOffset, s, s | TERMINATION_PHASE); + reconcileState(); + releaseWaiters(0); // signal all threads + releaseWaiters(1); } /** * Returns the current phase number. The maximum phase number is - * Integer.MAX_VALUE, after which it restarts at + * {@code Integer.MAX_VALUE}, after which it restarts at * zero. Upon termination, the phase number is negative. + * * @return the phase number, or a negative value if terminated */ public final int getPhase() { - return phaseOf(getReconciledState()); - } - - /** - * Returns true if the current phase number equals the given phase. - * @param phase the phase - * @return true if the current phase number equals the given phase. - */ - public final boolean hasPhase(int phase) { - return phaseOf(getReconciledState()) == phase; + return (int)((parent == null? state : reconcileState()) >>> PHASE_SHIFT); } /** * Returns the number of parties registered at this barrier. + * * @return the number of parties */ public int getRegisteredParties() { - return partiesOf(state); + return partiesOf(parent == null? state : reconcileState()); } /** - * Returns the number of parties that have arrived at the current - * phase of this barrier. + * Returns the number of registered parties that have arrived at + * the current phase of this barrier. + * * @return the number of arrived parties */ public int getArrivedParties() { - return arrivedOf(state); + return arrivedOf(parent == null? state : reconcileState()); } /** * Returns the number of registered parties that have not yet * arrived at the current phase of this barrier. + * * @return the number of unarrived parties */ public int getUnarrivedParties() { - return unarrivedOf(state); + return unarrivedOf(parent == null? state : reconcileState()); } /** - * Returns the parent of this phaser, or null if none. - * @return the parent of this phaser, or null if none. + * Returns the parent of this phaser, or {@code null} if none. + * + * @return the parent of this phaser, or {@code null} if none */ public Phaser getParent() { return parent; @@ -659,49 +718,52 @@ public class Phaser { /** * Returns the root ancestor of this phaser, which is the same as * this phaser if it has no parent. - * @return the root ancestor of this phaser. + * + * @return the root ancestor of this phaser */ public Phaser getRoot() { return root; } /** - * Returns true if this barrier has been terminated. - * @return true if this barrier has been terminated + * Returns {@code true} if this barrier has been terminated. + * + * @return {@code true} if this barrier has been terminated */ public boolean isTerminated() { - return getPhase() < 0; + return (parent == null? state : reconcileState()) < 0; } /** - * Overridable method to perform an action upon phase advance, and - * to control termination. This method is invoked whenever the - * barrier is tripped (and thus all other waiting parties are - * dormant). If it returns true, then, rather than advance the - * phase number, this barrier will be set to a final termination - * state, and subsequent calls to isTerminated will - * return true. + * Overridable method to perform an action upon impending phase + * advance, and to control termination. This method is invoked + * upon arrival of the party tripping the barrier (when all other + * waiting parties are dormant). If this method returns {@code + * true}, then, rather than advance the phase number, this barrier + * will be set to a final termination state, and subsequent calls + * to {@link #isTerminated} will return true. Any (unchecked) + * Exception or Error thrown by an invocation of this method is + * propagated to the party attempting to trip the barrier, in + * which case no advance occurs. + * + *

The arguments to this method provide the state of the phaser + * prevailing for the current transition. The effects of invoking + * arrival, registration, and waiting methods on this Phaser from + * within {@code onAdvance} are unspecified and should not be + * relied on. * - *

The default version returns true when the number of + *

If this Phaser is a member of a tiered set of Phasers, then + * {@code onAdvance} is invoked only for its root Phaser on each + * advance. + * + *

The default version returns {@code true} when the number of * registered parties is zero. Normally, overrides that arrange * termination for other reasons should also preserve this * property. * - *

You may override this method to perform an action with side - * effects visible to participating tasks, but it is in general - * only sensible to do so in designs where all parties register - * before any arrive, and all awaitAdvance at each phase. - * Otherwise, you cannot ensure lack of interference. In - * particular, this method may be invoked more than once per - * transition if other parties successfully register while the - * invocation of this method is in progress, thus postponing the - * transition until those parties also arrive, re-triggering this - * method. - * * @param phase the phase number on entering the barrier - * @param registeredParties the current number of registered - * parties. - * @return true if this barrier should terminate + * @param registeredParties the current number of registered parties + * @return {@code true} if this barrier should terminate */ protected boolean onAdvance(int phase, int registeredParties) { return registeredParties <= 0; @@ -710,204 +772,234 @@ public class Phaser { /** * Returns a string identifying this phaser, as well as its * state. The state, in brackets, includes the String {@code - * "phase ="} followed by the phase number, {@code "parties ="} + * "phase = "} followed by the phase number, {@code "parties = "} * followed by the number of registered parties, and {@code - * "arrived ="} followed by the number of arrived parties + * "arrived = "} followed by the number of arrived parties. * * @return a string identifying this barrier, as well as its state */ public String toString() { - long s = getReconciledState(); - return super.toString() + "[phase = " + phaseOf(s) + " parties = " + partiesOf(s) + " arrived = " + arrivedOf(s) + "]"; + long s = reconcileState(); + return super.toString() + + "[phase = " + phaseOf(s) + + " parties = " + partiesOf(s) + + " arrived = " + arrivedOf(s) + "]"; } - // methods for waiting - - /** The number of CPUs, for spin control */ - static final int NCPUS = Runtime.getRuntime().availableProcessors(); - /** - * The number of times to spin before blocking in timed waits. - * The value is empirically derived. - */ - static final int maxTimedSpins = (NCPUS < 2)? 0 : 32; - - /** - * The number of times to spin before blocking in untimed waits. - * This is greater than timed value because untimed waits spin - * faster since they don't need to check times on each spin. - */ - static final int maxUntimedSpins = maxTimedSpins * 32; - - /** - * The number of nanoseconds for which it is faster to spin - * rather than to use timed park. A rough estimate suffices. - */ - static final long spinForTimeoutThreshold = 1000L; - - /** - * Wait nodes for Treiber stack representing wait queue for non-FJ - * tasks. - */ - static final class QNode { - QNode next; - volatile Thread thread; // nulled to cancel wait - QNode() { - thread = Thread.currentThread(); - } - void signal() { - Thread t = thread; - if (t != null) { - thread = null; - LockSupport.unpark(t); - } - } - } - - /** - * Removes and signals waiting threads from wait queue + * Removes and signals threads from queue for phase */ private void releaseWaiters(int phase) { AtomicReference head = queueFor(phase); QNode q; - while ((q = head.get()) != null) { + int p; + while ((q = head.get()) != null && + ((p = q.phase) == phase || + (int)(root.state >>> PHASE_SHIFT) != p)) { if (head.compareAndSet(q, q.next)) q.signal(); } } /** - * Enqueues node and waits unless aborted or signalled. + * Tries to enqueue given node in the appropriate wait queue. + * + * @return true if successful */ - private int untimedWait(int phase) { - int spins = maxUntimedSpins; - QNode node = null; - boolean interrupted = false; - boolean queued = false; - int p; - while ((p = getPhase()) == phase) { - interrupted = Thread.interrupted(); - if (node != null) { - if (!queued) { - AtomicReference head = queueFor(phase); - queued = head.compareAndSet(node.next = head.get(), node); - } - else if (node.thread != null) - LockSupport.park(this); - } - else if (spins <= 0) - node = new QNode(); - else - --spins; - } - if (node != null) - node.thread = null; - if (interrupted) - Thread.currentThread().interrupt(); - releaseWaiters(phase); - return p; + private boolean tryEnqueue(int phase, QNode node) { + releaseWaiters(phase-1); // ensure old queue clean + AtomicReference head = queueFor(phase); + QNode q = head.get(); + return ((q == null || q.phase == phase) && + (int)(root.state >>> PHASE_SHIFT) == phase && + head.compareAndSet(node.next = q, node)); } + /** The number of CPUs, for spin control */ + private static final int NCPU = Runtime.getRuntime().availableProcessors(); + /** - * Messier interruptible version + * The number of times to spin before blocking while waiting for + * advance, per arrival while waiting. On multiprocessors, fully + * blocking and waking up a large number of threads all at once is + * usually a very slow process, so we use rechargeable spins to + * avoid it when threads regularly arrive: When a thread in + * internalAwaitAdvance notices another arrival before blocking, + * and there appear to be enough CPUs available, it spins + * SPINS_PER_ARRIVAL more times before continuing to try to + * block. The value trades off good-citizenship vs big unnecessary + * slowdowns. */ - private int interruptibleWait(int phase) throws InterruptedException { - int spins = maxUntimedSpins; - QNode node = null; + static final int SPINS_PER_ARRIVAL = NCPU < 2? 1 : 1 << 8; + + /** + * Possibly blocks and waits for phase to advance unless aborted. + * + * @param phase current phase + * @param node if non-null, the wait node to track interrupt and timeout; + * if null, denotes noninterruptible wait + * @return current phase + */ + private int internalAwaitAdvance(int phase, QNode node) { + Phaser current = this; // to eventually wait at root if tiered + Phaser par = parent; boolean queued = false; - boolean interrupted = false; + int spins = SPINS_PER_ARRIVAL; + int lastUnarrived = -1; // to increase spins upon change + long s; int p; - while ((p = getPhase()) == phase) { - if (interrupted = Thread.interrupted()) + while ((p = (int)((s = current.state) >>> PHASE_SHIFT)) == phase) { + int unarrived = (int)(s & UNARRIVED_MASK); + if (unarrived != lastUnarrived) { + if ((lastUnarrived = unarrived) < NCPU) + spins += SPINS_PER_ARRIVAL; + } + else if (unarrived == 0 && par != null) { + current = par; // if all arrived, use parent + par = par.parent; + } + else if (spins > 0) + --spins; + else if (node == null) + node = new QNode(this, phase, false, false, 0L); + else if (node.isReleasable()) break; - if (node != null) { - if (!queued) { - AtomicReference head = queueFor(phase); - queued = head.compareAndSet(node.next = head.get(), node); + else if (!queued) + queued = tryEnqueue(phase, node); + else { + try { + ForkJoinPool.managedBlock(node); + } catch (InterruptedException ie) { + node.wasInterrupted = true; } - else if (node.thread != null) - LockSupport.park(this); } - else if (spins <= 0) - node = new QNode(); - else - --spins; } - if (node != null) - node.thread = null; - if (interrupted) - throw new InterruptedException(); - releaseWaiters(phase); + if (node != null) { + if (node.thread != null) + node.thread = null; + if (!node.interruptible && node.wasInterrupted) + Thread.currentThread().interrupt(); + } + if (p == phase) + p = (int)(reconcileState() >>> PHASE_SHIFT); + if (p != phase) + releaseWaiters(phase); return p; } /** - * Even messier timeout version. + * Wait nodes for Treiber stack representing wait queue */ - private int timedWait(int phase, long nanos) - throws InterruptedException, TimeoutException { - int p; - if ((p = getPhase()) == phase) { - long lastTime = System.nanoTime(); - int spins = maxTimedSpins; - QNode node = null; - boolean queued = false; - boolean interrupted = false; - while ((p = getPhase()) == phase) { - if (interrupted = Thread.interrupted()) - break; - long now = System.nanoTime(); - if ((nanos -= now - lastTime) <= 0) - break; - lastTime = now; - if (node != null) { - if (!queued) { - AtomicReference head = queueFor(phase); - queued = head.compareAndSet(node.next = head.get(), node); - } - else if (node.thread != null && - nanos > spinForTimeoutThreshold) { - LockSupport.parkNanos(this, nanos); + static final class QNode implements ForkJoinPool.ManagedBlocker { + final Phaser phaser; + final int phase; + final boolean interruptible; + final boolean timed; + boolean wasInterrupted; + long nanos; + long lastTime; + volatile Thread thread; // nulled to cancel wait + QNode next; + + QNode(Phaser phaser, int phase, boolean interruptible, + boolean timed, long nanos) { + this.phaser = phaser; + this.phase = phase; + this.interruptible = interruptible; + this.nanos = nanos; + this.timed = timed; + this.lastTime = timed? System.nanoTime() : 0L; + thread = Thread.currentThread(); + } + + public boolean isReleasable() { + Thread t = thread; + if (t != null) { + if (phaser.getPhase() != phase) + t = null; + else { + if (Thread.interrupted()) + wasInterrupted = true; + if (interruptible && wasInterrupted) + t = null; + else if (timed) { + if (nanos > 0) { + long now = System.nanoTime(); + nanos -= now - lastTime; + lastTime = now; + } + if (nanos <= 0) + t = null; } } - else if (spins <= 0) - node = new QNode(); - else - --spins; + if (t != null) + return false; + thread = null; + } + return true; + } + + public boolean block() { + if (isReleasable()) + return true; + else if (!timed) + LockSupport.park(this); + else if (nanos > 0) + LockSupport.parkNanos(this, nanos); + return isReleasable(); + } + + void signal() { + Thread t = thread; + if (t != null) { + thread = null; + LockSupport.unpark(t); } - if (node != null) - node.thread = null; - if (interrupted) - throw new InterruptedException(); - if (p == phase && (p = getPhase()) == phase) - throw new TimeoutException(); } - releaseWaiters(phase); - return p; } - // Temporary Unsafe mechanics for preliminary release + // Unsafe mechanics - static final Unsafe _unsafe; - static final long stateOffset; + private static final sun.misc.Unsafe UNSAFE = getUnsafe(); + private static final long stateOffset = + objectFieldOffset("state", Phaser.class); - static { + private static long objectFieldOffset(String field, Class klazz) { try { - if (Phaser.class.getClassLoader() != null) { - Field f = Unsafe.class.getDeclaredField("theUnsafe"); - f.setAccessible(true); - _unsafe = (Unsafe)f.get(null); - } - else - _unsafe = Unsafe.getUnsafe(); - stateOffset = _unsafe.objectFieldOffset - (Phaser.class.getDeclaredField("state")); - } catch (Exception e) { - throw new RuntimeException("Could not initialize intrinsics", e); + return UNSAFE.objectFieldOffset(klazz.getDeclaredField(field)); + } catch (NoSuchFieldException e) { + // Convert Exception to corresponding Error + NoSuchFieldError error = new NoSuchFieldError(field); + error.initCause(e); + throw error; } } - final boolean casState(long cmp, long val) { - return _unsafe.compareAndSwapLong(this, stateOffset, cmp, val); + /** + * Returns a sun.misc.Unsafe. Suitable for use in a 3rd party package. + * Replace with a simple call to Unsafe.getUnsafe when integrating + * into a jdk. + * + * @return a sun.misc.Unsafe + */ + private static sun.misc.Unsafe getUnsafe() { + try { + return sun.misc.Unsafe.getUnsafe(); + } catch (SecurityException se) { + try { + return java.security.AccessController.doPrivileged + (new java.security + .PrivilegedExceptionAction() { + public sun.misc.Unsafe run() throws Exception { + java.lang.reflect.Field f = sun.misc + .Unsafe.class.getDeclaredField("theUnsafe"); + f.setAccessible(true); + return (sun.misc.Unsafe) f.get(null); + }}); + } catch (java.security.PrivilegedActionException e) { + throw new RuntimeException("Could not initialize intrinsics", + e.getCause()); + } + } } }