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.68 by dl, Sat Dec 4 15:25:08 2010 UTC vs.
Revision 1.71 by jsr166, Tue Mar 15 19:47:02 2011 UTC

# Line 1 | Line 1
1   /*
2   * Written by Doug Lea with assistance from members of JCP JSR-166
3   * Expert Group and released to the public domain, as explained at
4 < * http://creativecommons.org/licenses/publicdomain
4 > * http://creativecommons.org/publicdomain/zero/1.0/
5   */
6  
7   package jsr166y;
# Line 77 | Line 77 | import java.util.concurrent.locks.LockSu
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
81 < * value. Similarly, attempts to register upon termination have no
82 < * effect.  Termination is triggered when an invocation of {@code
83 < * onAdvance} returns {@code true}. The default implementation returns
84 < * {@code true} if a deregistration has caused the number of
85 < * registered parties to become zero.  As illustrated below, when
86 < * phasers control actions with a fixed number of iterations, it is
87 < * often convenient to override this method to cause termination when
88 < * the current phase number reaches a threshold. Method {@link
89 < * #forceTermination} is also available to abruptly release waiting
90 < * threads and allow them to terminate.
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
# Line 260 | Line 260 | public class Phaser {
260       * parent.
261       *
262       * The phase of a subphaser is allowed to lag that of its
263 <     * ancestors until it is actually accessed.  Method reconcileState
264 <     * is usually attempted only only when the number of unarrived
265 <     * parties appears to be zero, which indicates a potential lag in
266 <     * updating phase after the root advanced.
263 >     * ancestors until it is actually accessed -- see method
264 >     * reconcileState.
265       */
266      private volatile long state;
267  
# Line 271 | Line 269 | public class Phaser {
269      private static final int  MAX_PHASE       = 0x7fffffff;
270      private static final int  PARTIES_SHIFT   = 16;
271      private static final int  PHASE_SHIFT     = 32;
272 +    private static final long PHASE_MASK      = -1L << PHASE_SHIFT;
273      private static final int  UNARRIVED_MASK  = 0xffff;      // to mask ints
274      private static final long PARTIES_MASK    = 0xffff0000L; // to mask longs
275      private static final long TERMINATION_BIT = 1L << 63;
# Line 436 | Line 435 | public class Phaser {
435  
436      /**
437       * Resolves lagged phase propagation from root if necessary.
438 +     * Reconciliation normally occurs when root has advanced but
439 +     * subphasers have not yet done so, in which case they must finish
440 +     * their own advance by setting unarrived to parties (or if
441 +     * parties is zero, resetting to unregistered EMPTY state).
442 +     * However, this method may also be called when "floating"
443 +     * subphasers with possibly some unarrived parties are merely
444 +     * catching up to current phase, in which case counts are
445 +     * unaffected.
446 +     *
447 +     * @return reconciled state
448       */
449      private long reconcileState() {
450 <        Phaser rt = root;
450 >        final Phaser root = this.root;
451          long s = state;
452 <        if (rt != this) {
453 <            int phase;
454 <            while ((phase = (int)(rt.state >>> PHASE_SHIFT)) !=
455 <                   (int)(s >>> PHASE_SHIFT)) {
456 <                // assert phase < 0 || unarrivedOf(s) == 0
457 <                long t;                             // to reread s
458 <                long p = s & PARTIES_MASK;          // unshifted parties field
459 <                long n = (((long) phase) << PHASE_SHIFT) | p;
460 <                if (phase >= 0) {
461 <                    if (p == 0L)
462 <                        n |= EMPTY;                 // reset to empty
454 <                    else
455 <                        n |= p >>> PARTIES_SHIFT;   // set unarr to parties
456 <                }
457 <                if ((t = state) == s &&
458 <                    UNSAFE.compareAndSwapLong(this, stateOffset, s, s = n))
459 <                    break;
460 <                s = t;
461 <            }
452 >        if (root != this) {
453 >            int phase, u, p;
454 >            // CAS root phase with current parties; possibly trip unarrived
455 >            while ((phase = (int)(root.state >>> PHASE_SHIFT)) !=
456 >                   (int)(s >>> PHASE_SHIFT) &&
457 >                   !UNSAFE.compareAndSwapLong
458 >                   (this, stateOffset, s,
459 >                    s = ((((long) phase) << PHASE_SHIFT) | (s & PARTIES_MASK) |
460 >                         ((p = (int)s >>> PARTIES_SHIFT) == 0 ? EMPTY :
461 >                          (u = (int)s & UNARRIVED_MASK) == 0 ? p : u))))
462 >                s = state;
463          }
464          return s;
465      }
# Line 541 | Line 542 | public class Phaser {
542       *
543       * @return the arrival phase number to which this registration
544       * applied.  If this value is negative, then this phaser has
545 <     * terminated, in which casem registration has no effect.
545 >     * terminated, in which case registration has no effect.
546       * @throws IllegalStateException if attempting to register more
547       * than the maximum supported number of parties
548       */
# Line 563 | Line 564 | public class Phaser {
564       * advance to the next phase
565       * @return the arrival phase number to which this registration
566       * applied.  If this value is negative, then this phaser has
567 <     * terminated, in which casem registration has no effect.
567 >     * terminated, in which case registration has no effect.
568       * @throws IllegalStateException if attempting to register more
569       * than the maximum supported number of parties
570       * @throws IllegalArgumentException if {@code parties < 0}
# Line 682 | Line 683 | public class Phaser {
683       */
684      public int awaitAdvance(int phase) {
685          final Phaser root = this.root;
686 <        int p = (int)((root == this? state : reconcileState()) >>> PHASE_SHIFT);
686 >        long s = (root == this) ? state : reconcileState();
687 >        int p = (int)(s >>> PHASE_SHIFT);
688          if (phase < 0)
689              return phase;
690          if (p == phase)
# Line 708 | Line 710 | public class Phaser {
710      public int awaitAdvanceInterruptibly(int phase)
711          throws InterruptedException {
712          final Phaser root = this.root;
713 <        int p = (int)((root == this? state : reconcileState()) >>> PHASE_SHIFT);
713 >        long s = (root == this) ? state : reconcileState();
714 >        int p = (int)(s >>> PHASE_SHIFT);
715          if (phase < 0)
716              return phase;
717          if (p == phase) {
# Line 745 | Line 748 | public class Phaser {
748          throws InterruptedException, TimeoutException {
749          long nanos = unit.toNanos(timeout);
750          final Phaser root = this.root;
751 <        int p = (int)((root == this? state : reconcileState()) >>> PHASE_SHIFT);
751 >        long s = (root == this) ? state : reconcileState();
752 >        int p = (int)(s >>> PHASE_SHIFT);
753          if (phase < 0)
754              return phase;
755          if (p == phase) {
# Line 773 | Line 777 | public class Phaser {
777          final Phaser root = this.root;
778          long s;
779          while ((s = root.state) >= 0) {
780 <            long next = (s & ~((long)UNARRIVED_MASK)) | TERMINATION_BIT;
781 <            if (UNSAFE.compareAndSwapLong(root, stateOffset, s, next)) {
780 >            if (UNSAFE.compareAndSwapLong(root, stateOffset,
781 >                                          s, s | TERMINATION_BIT)) {
782                  // signal all threads
783                  releaseWaiters(0);
784                  releaseWaiters(1);
# Line 807 | Line 811 | public class Phaser {
811  
812      /**
813       * Returns the number of registered parties that have arrived at
814 <     * the current phase of this phaser.
814 >     * the current phase of this phaser. If this phaser has terminated,
815 >     * the returned value is meaningless and arbitrary.
816       *
817       * @return the number of arrived parties
818       */
# Line 817 | Line 822 | public class Phaser {
822  
823      /**
824       * Returns the number of registered parties that have not yet
825 <     * arrived at the current phase of this phaser.
825 >     * arrived at the current phase of this phaser. If this phaser has
826 >     * terminated, the returned value is meaningless and arbitrary.
827       *
828       * @return the number of unarrived parties
829       */

Diff Legend

– Removed lines
+ Added lines
< Changed lines
> Changed lines