--- jsr166/src/jsr166y/Phaser.java 2010/11/13 01:27:13 1.52 +++ jsr166/src/jsr166y/Phaser.java 2010/11/19 16:03:24 1.57 @@ -240,12 +240,12 @@ public class Phaser { */ private volatile long state; - private static final int MAX_COUNT = 0xffff; + private static final int MAX_PARTIES = 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 int UNARRIVED_MASK = 0xffff; + private static final long PARTIES_MASK = 0xffff0000L; // for masking long 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; @@ -253,11 +253,11 @@ public class Phaser { // The following unpacking methods are usually manually inlined private static int unarrivedOf(long s) { - return (int) (s & UNARRIVED_MASK); + return (int)s & UNARRIVED_MASK; } private static int partiesOf(long s) { - return ((int) (s & PARTIES_MASK)) >>> PARTIES_SHIFT; + return (int)s >>> PARTIES_SHIFT; } private static int phaseOf(long s) { @@ -302,50 +302,52 @@ public class Phaser { * 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(); - } + for (;;) { + long s = state; + int phase = (int)(s >>> PHASE_SHIFT); + if (phase < 0) + return phase; + int unarrived = (int)s & UNARRIVED_MASK; + if (unarrived == 0) + checkBadArrive(s); + else if (UNSAFE.compareAndSwapLong(this, stateOffset, s, s-=adj)) { + if (unarrived == 1) { + 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; + final Phaser parent = this.parent; + if (parent == null) { + if (onAdvance(phase, u)) + next |= TERMINATION_PHASE; // obliterate phase + UNSAFE.compareAndSwapLong(this, stateOffset, s, next); + releaseWaiters(phase); + } + else { + parent.doArrive((u == 0) ? + ONE_ARRIVAL|ONE_PARTY : ONE_ARRIVAL); + if ((int)(parent.state >>> PHASE_SHIFT) != nextPhase || + ((int)(state >>> PHASE_SHIFT) != nextPhase && + !UNSAFE.compareAndSwapLong(this, stateOffset, + s, next))) + reconcileState(); } - break; } + return phase; } - 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. + * Rechecks state and throws bounds exceptions on arrival -- called + * only if unarrived is apparently zero. */ - private String badArrive() { - return ("Attempted arrival of unregistered party for " + - this.toString()); + private void checkBadArrive(long s) { + if (reconcileState() == s) + throw new IllegalStateException + ("Attempted arrival of unregistered party for " + + stateToString(s)); } /** @@ -354,60 +356,57 @@ public class Phaser { * @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) + // assert registrations > 0; + // adjustment to state + long adj = ((long)registrations << PARTIES_SHIFT) | registrations; + final Phaser parent = this.parent; + for (;;) { + long s = (parent == null) ? state : reconcileState(); + int phase = (int)(s >>> PHASE_SHIFT); + if (phase < 0) + return phase; + int parties = (int)s >>> PARTIES_SHIFT; + if (parties != 0 && ((int)s & UNARRIVED_MASK) == 0) internalAwaitAdvance(phase, null); // wait for onAdvance - else if (parties + registrations > MAX_COUNT) - throw new IllegalStateException(badRegister()); + else if (registrations > MAX_PARTIES - parties) + throw new IllegalStateException(badRegister(s)); else if (UNSAFE.compareAndSwapLong(this, stateOffset, s, s + adj)) - break; + return phase; } - return phase; } /** - * Returns message string for bounds exceptions on registration + * Returns message string for out of bounds exceptions on registration. */ - private String badRegister() { - return ("Attempt to register more than " + MAX_COUNT + " parties for "+ - this.toString()); + private String badRegister(long s) { + return "Attempt to register more than " + + MAX_PARTIES + " parties for " + stateToString(s); } /** - * Recursively resolves lagged phase propagation from root if - * necessary. + * Recursively resolves lagged phase propagation from root if necessary. */ private long reconcileState() { 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; + long s = state; + if (par != null) { + Phaser rt = root; + int phase, rPhase; + while ((phase = (int)(s >>> PHASE_SHIFT)) >= 0 && + (rPhase = (int)(rt.state >>> PHASE_SHIFT)) != phase) { + if ((int)(par.state >>> PHASE_SHIFT) != rPhase) + par.reconcileState(); + else if (rPhase < 0 || ((int)s & UNARRIVED_MASK) == 0) { + long u = s & PARTIES_MASK; // reset unarrived to parties + long next = ((((long) rPhase) << PHASE_SHIFT) | u | + (u >>> PARTIES_SHIFT)); + if (state == s && + UNSAFE.compareAndSwapLong(this, stateOffset, + s, s = next)) + break; } + s = state; } - if (state == s) - releaseWaiters(phase); // help release others } return s; } @@ -457,7 +456,7 @@ public class Phaser { * or greater than the maximum number of parties supported */ public Phaser(Phaser parent, int parties) { - if (parties < 0 || parties > MAX_COUNT) + if (parties >>> PARTIES_SHIFT != 0) throw new IllegalArgumentException("Illegal number of parties"); int phase; this.parent = parent; @@ -475,7 +474,7 @@ public class Phaser { phase = 0; } long p = (long)parties; - this.state = (((long) phase) << PHASE_SHIFT) | p | (p << PARTIES_SHIFT); + this.state = (((long)phase) << PHASE_SHIFT) | p | (p << PARTIES_SHIFT); } /** @@ -505,8 +504,6 @@ public class Phaser { 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); @@ -575,10 +572,9 @@ public class Phaser { public int awaitAdvance(int phase) { if (phase < 0) return phase; - int p = (int)((parent==null? state : reconcileState()) >>> PHASE_SHIFT); - if (p != phase) - return p; - return internalAwaitAdvance(phase, null); + long s = (parent == null) ? state : reconcileState(); + int p = (int)(s >>> PHASE_SHIFT); + return (p != phase) ? p : internalAwaitAdvance(phase, null); } /** @@ -599,15 +595,15 @@ public class Phaser { throws InterruptedException { if (phase < 0) return phase; - int p = (int)((parent==null? state : reconcileState()) >>> PHASE_SHIFT); - if (p != phase) - return p; - QNode node = new QNode(this, phase, true, false, 0L); - p = internalAwaitAdvance(phase, node); - if (node.wasInterrupted) - throw new InterruptedException(); - else - return p; + long s = (parent == null) ? state : reconcileState(); + int p = (int)(s >>> PHASE_SHIFT); + if (p == phase) { + QNode node = new QNode(this, phase, true, false, 0L); + p = internalAwaitAdvance(phase, node); + if (node.wasInterrupted) + throw new InterruptedException(); + } + return p; } /** @@ -633,37 +629,43 @@ public class Phaser { public int awaitAdvanceInterruptibly(int phase, long timeout, TimeUnit unit) throws InterruptedException, TimeoutException { - long nanos = unit.toNanos(timeout); if (phase < 0) return phase; - int p = (int)((parent==null? state : reconcileState()) >>> PHASE_SHIFT); - if (p != phase) - return p; - 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; + long s = (parent == null) ? state : reconcileState(); + int p = (int)(s >>> PHASE_SHIFT); + if (p == phase) { + long nanos = 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(); + } + return p; } /** - * Forces this barrier to enter termination state. Counts of - * arrived and registered parties are unaffected. If this phaser - * has a parent, it too is terminated. This method may be useful - * for coordinating recovery after one or more tasks encounter - * unexpected exceptions. + * Forces this barrier to enter termination state. Counts of + * arrived and registered parties are unaffected. If this phaser + * is a member of a tiered set of phasers, then all of the phasers + * in the set are terminated. If this phaser is already + * terminated, this method has no effect. This method may be + * useful for coordinating recovery after one or more tasks + * encounter unexpected exceptions. */ public void forceTermination() { - Phaser r = root; // force at root then reconcile + // Only need to change root state + final Phaser root = this.root; long s; - while ((s = r.state) >= 0) - UNSAFE.compareAndSwapLong(r, stateOffset, s, s | TERMINATION_PHASE); - reconcileState(); - releaseWaiters(0); // signal all threads - releaseWaiters(1); + while ((s = root.state) >= 0) { + if (UNSAFE.compareAndSwapLong(root, stateOffset, + s, s | TERMINATION_PHASE)) { + releaseWaiters(0); // signal all threads + releaseWaiters(1); + return; + } + } } /** @@ -674,7 +676,7 @@ public class Phaser { * @return the phase number, or a negative value if terminated */ public final int getPhase() { - return (int)((parent == null? state : reconcileState()) >>> PHASE_SHIFT); + return (int)(root.state >>> PHASE_SHIFT); } /** @@ -683,7 +685,7 @@ public class Phaser { * @return the number of parties */ public int getRegisteredParties() { - return partiesOf(parent == null? state : reconcileState()); + return partiesOf(state); } /** @@ -693,7 +695,7 @@ public class Phaser { * @return the number of arrived parties */ public int getArrivedParties() { - return arrivedOf(parent == null? state : reconcileState()); + return arrivedOf(parent==null? state : reconcileState()); } /** @@ -703,7 +705,7 @@ public class Phaser { * @return the number of unarrived parties */ public int getUnarrivedParties() { - return unarrivedOf(parent == null? state : reconcileState()); + return unarrivedOf(parent==null? state : reconcileState()); } /** @@ -731,7 +733,7 @@ public class Phaser { * @return {@code true} if this barrier has been terminated */ public boolean isTerminated() { - return (parent == null? state : reconcileState()) < 0; + return root.state < 0L; } /** @@ -779,15 +781,23 @@ public class Phaser { * @return a string identifying this barrier, as well as its state */ public String toString() { - long s = reconcileState(); + return stateToString(reconcileState()); + } + + /** + * Implementation of toString and string-based error messages + */ + private String stateToString(long s) { return super.toString() + "[phase = " + phaseOf(s) + " parties = " + partiesOf(s) + " arrived = " + arrivedOf(s) + "]"; } + // Waiting mechanics + /** - * Removes and signals threads from queue for phase + * Removes and signals threads from queue for phase. */ private void releaseWaiters(int phase) { AtomicReference head = queueFor(phase); @@ -801,20 +811,6 @@ public class Phaser { } } - /** - * Tries to enqueue given node in the appropriate wait queue. - * - * @return true if successful - */ - 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(); @@ -826,46 +822,65 @@ public class Phaser { * 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. + * SPINS_PER_ARRIVAL more times before blocking. Plus, even on + * uniprocessors, there is at least one intervening Thread.yield + * before blocking. The value trades off good-citizenship vs big + * unnecessary slowdowns. */ - static final int SPINS_PER_ARRIVAL = NCPU < 2? 1 : 1 << 8; + 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 nonnull, the wait node to track interrupt and timeout; + * @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; - int spins = SPINS_PER_ARRIVAL; + boolean queued = false; // true when node is enqueued int lastUnarrived = -1; // to increase spins upon change + int spins = SPINS_PER_ARRIVAL; long s; int p; while ((p = (int)((s = current.state) >>> PHASE_SHIFT)) == phase) { - int unarrived = (int)(s & UNARRIVED_MASK); + Phaser par; + int unarrived = (int)s & UNARRIVED_MASK; if (unarrived != lastUnarrived) { + if (lastUnarrived == -1) // ensure old queue clean + releaseWaiters(phase-1); if ((lastUnarrived = unarrived) < NCPU) spins += SPINS_PER_ARRIVAL; } - else if (unarrived == 0 && par != null) { + else if (unarrived == 0 && (par = current.parent) != null) { current = par; // if all arrived, use parent par = par.parent; + lastUnarrived = -1; } - else if (spins > 0) - --spins; - else if (node == null) + else if (spins > 0) { + if (--spins == (SPINS_PER_ARRIVAL >>> 1)) + Thread.yield(); // yield midway through spin + } + else if (node == null) // must be noninterruptible node = new QNode(this, phase, false, false, 0L); - else if (node.isReleasable()) - break; - else if (!queued) - queued = tryEnqueue(phase, node); + else if (node.isReleasable()) { + if ((p = (int)(root.state >>> PHASE_SHIFT)) != phase) + break; + else + return phase; // aborted + } + else if (!queued) { // push onto queue + AtomicReference head = queueFor(phase); + QNode q = head.get(); + if (q == null || q.phase == phase) { + node.next = q; + if ((p = (int)(root.state >>> PHASE_SHIFT)) != phase) + break; // recheck to avoid stale enqueue + else + queued = head.compareAndSet(q, node); + } + } else { try { ForkJoinPool.managedBlock(node); @@ -874,16 +889,9 @@ public class Phaser { } } } - 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); + releaseWaiters(phase); + if (node != null) + node.onRelease(); return p; } @@ -956,6 +964,14 @@ public class Phaser { LockSupport.unpark(t); } } + + void onRelease() { // actions upon return from internalAwaitAdvance + if (!interruptible && wasInterrupted) + Thread.currentThread().interrupt(); + if (thread != null) + thread = null; + } + } // Unsafe mechanics