--- jsr166/src/jsr166y/Phaser.java 2009/07/20 22:40:09 1.13 +++ jsr166/src/jsr166y/Phaser.java 2009/08/02 17:02:06 1.25 @@ -7,10 +7,9 @@ package jsr166y; import java.util.concurrent.*; -import java.util.concurrent.atomic.*; + +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 @@ -33,18 +32,19 @@ import java.lang.reflect.*; * zero, and advancing when all parties reach the barrier (wrapping * around to zero after reaching {@code Integer.MAX_VALUE}). * - *
Sample usages: * - *
A 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: + *
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* - *list) { @@ -140,9 +140,9 @@ import java.lang.reflect.*; * phaser.arriveAndDeregister(); // deregister self, don't wait * }}
To create a set of tasks using a tree of Phasers, + *
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 + * Task class with a constructor accepting a phaser that * it registers for upon construction: *
{@code * void build(Task[] actions, int lo, int hi, Phaser b) { @@ -175,6 +175,9 @@ import java.lang.reflect.*; * parties result in IllegalStateExceptions. However, you can and * should create tiered phasers to accommodate arbitrarily large sets * of participants. + * + * @since 1.7 + * @author Doug Lea */ public class Phaser { /* @@ -208,15 +211,15 @@ public class Phaser { private static final int phaseMask = 0x7fffffff; private static int unarrivedOf(long s) { - return (int)(s & ushortMask); + return (int) (s & ushortMask); } private static int partiesOf(long s) { - return ((int)s) >>> 16; + return ((int) s) >>> 16; } private static int phaseOf(long s) { - return (int)(s >>> 32); + return (int) (s >>> 32); } private static int arrivedOf(long s) { @@ -224,17 +227,17 @@ public class Phaser { } private static long stateFor(int phase, int parties, int unarrived) { - return ((((long)phase) << 32) | (((long)parties) << 16) | - (long)unarrived); + return ((((long) phase) << 32) | (((long) parties) << 16) | + (long) unarrived); } private static long trippedStateFor(int phase, int parties) { - long lp = (long)parties; - return (((long)phase) << 32) | (lp << 16) | lp; + long lp = (long) parties; + return (((long) phase) << 32) | (lp << 16) | lp; } /** - * Returns message string for bad bounds exceptions + * Returns message string for bad bounds exceptions. */ private static String badBounds(int parties, int unarrived) { return ("Attempt to set " + unarrived + @@ -247,7 +250,7 @@ public class Phaser { 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; @@ -263,7 +266,7 @@ public class Phaser { private final AtomicReferenceoddQ = new AtomicReference (); private AtomicReference queueFor(int phase) { - return (phase & 1) == 0? evenQ : oddQ; + return ((phase & 1) == 0) ? evenQ : oddQ; } /** @@ -271,7 +274,7 @@ public class Phaser { * root if necessary. */ private long getReconciledState() { - return parent == null? state : reconcileState(); + return (parent == null) ? state : reconcileState(); } /** @@ -298,31 +301,33 @@ public class Phaser { } /** - * Creates a new Phaser without any initially registered parties, + * 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. + * phaser will need to first register for it. */ public Phaser() { this(null); } /** - * Creates a new Phaser with the given numbers of registered + * Creates a new phaser with the given numbers 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; @@ -337,14 +342,15 @@ public class Phaser { } /** - * 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 numbers 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) @@ -362,9 +368,10 @@ public class Phaser { /** * Adds a new unarrived party to this phaser. + * * @return the current barrier phase number upon registration * @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); @@ -372,10 +379,11 @@ public class Phaser { /** * Adds the given number of new unarrived parties to this phaser. - * @param parties the number of parties required to trip barrier. + * + * @param parties the number of parties required to trip barrier * @return the current barrier phase number upon registration * @throws IllegalStateException if attempting to register more - * than the maximum supported number of parties. + * than the maximum supported number of parties */ public int bulkRegister(int parties) { if (parties < 0) @@ -411,9 +419,9 @@ public class Phaser { * in turn wait for others via {@link #awaitAdvance}). * * @return the barrier phase number upon entry to this method, or a - * negative value if terminated; + * 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; @@ -433,7 +441,7 @@ public class Phaser { if (par == null) { // directly trip if (casState (s, - trippedStateFor(onAdvance(phase, parties)? -1 : + trippedStateFor(onAdvance(phase, parties) ? -1 : ((phase + 1) & phaseMask), parties))) { releaseWaiters(phase); break; @@ -463,9 +471,9 @@ public class Phaser { * zero parties, this phaser is also deregistered from its parent. * * @return the current barrier phase number upon entry to - * this method, or a negative value if terminated; + * this method, 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 @@ -494,7 +502,7 @@ public class Phaser { if (unarrived == 0) { if (casState (s, - trippedStateFor(onAdvance(phase, parties)? -1 : + trippedStateFor(onAdvance(phase, parties) ? -1 : ((phase + 1) & phaseMask), parties))) { releaseWaiters(phase); break; @@ -516,9 +524,10 @@ public class Phaser { * to {@code 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 * @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()); @@ -528,6 +537,7 @@ public class Phaser { * 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 */ @@ -549,6 +559,7 @@ public class Phaser { * value, or returns immediately if argument 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 * @throws InterruptedException if thread interrupted while waiting @@ -570,12 +581,14 @@ public class Phaser { * 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 * @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 { if (phase < 0) return phase; @@ -616,6 +629,7 @@ public class Phaser { * Returns the current phase number. The maximum phase number is * {@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() { @@ -624,6 +638,7 @@ public class Phaser { /** * Returns {@code true} if the current phase number equals the given phase. + * * @param phase the phase * @return {@code true} if the current phase number equals the given phase */ @@ -633,6 +648,7 @@ public class Phaser { /** * Returns the number of parties registered at this barrier. + * * @return the number of parties */ public int getRegisteredParties() { @@ -642,6 +658,7 @@ public class Phaser { /** * Returns the number of parties that have arrived at the current * phase of this barrier. + * * @return the number of arrived parties */ public int getArrivedParties() { @@ -651,6 +668,7 @@ public class Phaser { /** * 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() { @@ -658,8 +676,9 @@ public class Phaser { } /** - * 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; @@ -668,6 +687,7 @@ 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 */ public Phaser getRoot() { @@ -676,6 +696,7 @@ public class Phaser { /** * Returns {@code true} if this barrier has been terminated. + * * @return {@code true} if this barrier has been terminated */ public boolean isTerminated() { @@ -686,20 +707,20 @@ public class Phaser { * 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 {@code isTerminated} will - * return true. + * dormant). If it 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. * - * The default version returns true when the number of + *
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 + *
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 {@code awaitAdvance} at each phase. + * before any arrive, and all {@link #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 @@ -799,7 +820,7 @@ public class Phaser { } /** - * Removes and signals waiting threads from wait queue + * Removes and signals waiting threads from wait queue. */ private void releaseWaiters(int phase) { AtomicReference
head = queueFor(phase); @@ -811,7 +832,8 @@ public class Phaser { } /** - * Tries to enqueue given node in the appropriate wait queue + * Tries to enqueue given node in the appropriate wait queue. + * * @return true if successful */ private boolean tryEnqueue(QNode node) { @@ -821,6 +843,7 @@ public class Phaser { /** * Enqueues node and waits unless aborted or signalled. + * * @return current phase */ private int untimedWait(int phase) { @@ -908,49 +931,52 @@ public class Phaser { return p; } - // Temporary Unsafe mechanics for preliminary release - private static Unsafe getUnsafe() throws Throwable { - try { - return Unsafe.getUnsafe(); - } catch (SecurityException se) { - try { - return java.security.AccessController.doPrivileged - (new java.security.PrivilegedExceptionAction () { - public Unsafe run() throws Exception { - return getUnsafePrivileged(); - }}); - } catch (java.security.PrivilegedActionException e) { - throw e.getCause(); - } - } - } + // Unsafe mechanics - private static Unsafe getUnsafePrivileged() - throws NoSuchFieldException, IllegalAccessException { - Field f = Unsafe.class.getDeclaredField("theUnsafe"); - f.setAccessible(true); - return (Unsafe) f.get(null); - } + private static final sun.misc.Unsafe UNSAFE = getUnsafe(); + private static final long stateOffset = + objectFieldOffset("state", Phaser.class); - private static long fieldOffset(String fieldName) - throws NoSuchFieldException { - return _unsafe.objectFieldOffset - (Phaser.class.getDeclaredField(fieldName)); + private final boolean casState(long cmp, long val) { + return UNSAFE.compareAndSwapLong(this, stateOffset, cmp, val); } - static final Unsafe _unsafe; - static final long stateOffset; - - static { + private static long objectFieldOffset(String field, Class> klazz) { try { - _unsafe = getUnsafe(); - stateOffset = fieldOffset("state"); - } catch (Throwable 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()); + } + } } }