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.71 by jsr166, Tue Mar 15 19:47:02 2011 UTC vs.
Revision 1.81 by jsr166, Fri Jul 15 18:49:12 2016 UTC

# Line 17 | Line 17 | import java.util.concurrent.locks.LockSu
17   * {@link java.util.concurrent.CountDownLatch CountDownLatch}
18   * but supporting more flexible usage.
19   *
20 < * <p> <b>Registration.</b> Unlike the case for other barriers, the
20 > * <p><b>Registration.</b> Unlike the case for other barriers, the
21   * number of parties <em>registered</em> to synchronize on a phaser
22   * may vary over time.  Tasks may be registered at any time (using
23   * methods {@link #register}, {@link #bulkRegister}, or forms of
# Line 30 | Line 30 | import java.util.concurrent.locks.LockSu
30   * (However, you can introduce such bookkeeping by subclassing this
31   * class.)
32   *
33 < * <p> <b>Synchronization.</b> Like a {@code CyclicBarrier}, a {@code
33 > * <p><b>Synchronization.</b> Like a {@code CyclicBarrier}, a {@code
34   * Phaser} may be repeatedly awaited.  Method {@link
35   * #arriveAndAwaitAdvance} has effect analogous to {@link
36   * java.util.concurrent.CyclicBarrier#await CyclicBarrier.await}. Each
# Line 44 | Line 44 | import java.util.concurrent.locks.LockSu
44   *
45   * <ul>
46   *
47 < *   <li> <b>Arrival.</b> Methods {@link #arrive} and
47 > *   <li><b>Arrival.</b> Methods {@link #arrive} and
48   *       {@link #arriveAndDeregister} record arrival.  These methods
49   *       do not block, but return an associated <em>arrival phase
50   *       number</em>; that is, the phase number of the phaser to which
# Line 57 | Line 57 | import java.util.concurrent.locks.LockSu
57   *       flexible than, providing a barrier action to a {@code
58   *       CyclicBarrier}.
59   *
60 < *   <li> <b>Waiting.</b> Method {@link #awaitAdvance} requires an
60 > *   <li><b>Waiting.</b> Method {@link #awaitAdvance} requires an
61   *       argument indicating an arrival phase number, and returns when
62   *       the phaser advances to (or is already at) a different phase.
63   *       Unlike similar constructions using {@code CyclicBarrier},
# Line 74 | Line 74 | import java.util.concurrent.locks.LockSu
74   *
75   * </ul>
76   *
77 < * <p> <b>Termination.</b> A phaser may enter a <em>termination</em>
77 > * <p><b>Termination.</b> A phaser may enter a <em>termination</em>
78   * state, that may be checked using method {@link #isTerminated}. Upon
79   * termination, all synchronization methods immediately return without
80   * waiting for advance, as indicated by a negative return value.
# Line 89 | Line 89 | import java.util.concurrent.locks.LockSu
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.,
92 > * <p><b>Tiering.</b> Phasers may be <em>tiered</em> (i.e.,
93   * constructed in tree structures) to reduce contention. Phasers with
94   * large numbers of parties that would otherwise experience heavy
95   * synchronization contention costs may instead be set up so that
# Line 130 | Line 130 | import java.util.concurrent.locks.LockSu
130   * void runTasks(List<Runnable> tasks) {
131   *   final Phaser phaser = new Phaser(1); // "1" to register self
132   *   // create and start threads
133 < *   for (Runnable task : tasks) {
133 > *   for (final Runnable task : tasks) {
134   *     phaser.register();
135   *     new Thread() {
136   *       public void run() {
# Line 193 | Line 193 | import java.util.concurrent.locks.LockSu
193   *   phaser.arriveAndDeregister();
194   * }}</pre>
195   *
196 *
196   * <p>To create a set of {@code n} tasks using a tree of phasers, you
197   * could use code of the following form, assuming a Task class with a
198   * constructor accepting a {@code Phaser} that it registers with upon
# Line 237 | Line 236 | public class Phaser {
236       */
237  
238      /**
239 <     * Primary state representation, holding four fields:
239 >     * Primary state representation, holding four bit-fields:
240       *
241 <     * * unarrived -- the number of parties yet to hit barrier (bits  0-15)
242 <     * * parties -- the number of parties to wait              (bits 16-31)
243 <     * * phase -- the generation of the barrier                (bits 32-62)
244 <     * * terminated -- set if barrier is terminated            (bit  63 / sign)
241 >     * unarrived  -- the number of parties yet to hit barrier (bits  0-15)
242 >     * parties    -- the number of parties to wait            (bits 16-31)
243 >     * phase      -- the generation of the barrier            (bits 32-62)
244 >     * terminated -- set if barrier is terminated             (bit  63 / sign)
245       *
246       * Except that a phaser with no registered parties is
247 <     * distinguished with the otherwise illegal state of having zero
247 >     * distinguished by the otherwise illegal state of having zero
248       * parties and one unarrived parties (encoded as EMPTY below).
249       *
250       * To efficiently maintain atomicity, these values are packed into
# Line 266 | Line 265 | public class Phaser {
265      private volatile long state;
266  
267      private static final int  MAX_PARTIES     = 0xffff;
268 <    private static final int  MAX_PHASE       = 0x7fffffff;
268 >    private static final int  MAX_PHASE       = Integer.MAX_VALUE;
269      private static final int  PARTIES_SHIFT   = 16;
270      private static final int  PHASE_SHIFT     = 32;
272    private static final long PHASE_MASK      = -1L << PHASE_SHIFT;
271      private static final int  UNARRIVED_MASK  = 0xffff;      // to mask ints
272      private static final long PARTIES_MASK    = 0xffff0000L; // to mask longs
273 +    private static final long COUNTS_MASK     = 0xffffffffL;
274      private static final long TERMINATION_BIT = 1L << 63;
275  
276      // some special values
277      private static final int  ONE_ARRIVAL     = 1;
278      private static final int  ONE_PARTY       = 1 << PARTIES_SHIFT;
279 +    private static final int  ONE_DEREGISTER  = ONE_ARRIVAL|ONE_PARTY;
280      private static final int  EMPTY           = 1;
281  
282      // The following unpacking methods are usually manually inlined
283  
284      private static int unarrivedOf(long s) {
285          int counts = (int)s;
286 <        return (counts == EMPTY) ? 0 : counts & UNARRIVED_MASK;
286 >        return (counts == EMPTY) ? 0 : (counts & UNARRIVED_MASK);
287      }
288  
289      private static int partiesOf(long s) {
# Line 291 | Line 291 | public class Phaser {
291      }
292  
293      private static int phaseOf(long s) {
294 <        return (int) (s >>> PHASE_SHIFT);
294 >        return (int)(s >>> PHASE_SHIFT);
295      }
296  
297      private static int arrivedOf(long s) {
# Line 344 | Line 344 | public class Phaser {
344       * Manually tuned to speed up and minimize race windows for the
345       * common case of just decrementing unarrived field.
346       *
347 <     * @param deregister false for arrive, true for arriveAndDeregister
347 >     * @param adjust value to subtract from state;
348 >     *               ONE_ARRIVAL for arrive,
349 >     *               ONE_DEREGISTER for arriveAndDeregister
350       */
351 <    private int doArrive(boolean deregister) {
350 <        int adj = deregister ? ONE_ARRIVAL|ONE_PARTY : ONE_ARRIVAL;
351 >    private int doArrive(int adjust) {
352          final Phaser root = this.root;
353          for (;;) {
354              long s = (root == this) ? state : reconcileState();
355              int phase = (int)(s >>> PHASE_SHIFT);
355            int counts = (int)s;
356            int unarrived = (counts & UNARRIVED_MASK) - 1;
356              if (phase < 0)
357                  return phase;
358 <            else if (counts == EMPTY || unarrived < 0) {
359 <                if (root == this || reconcileState() == s)
360 <                    throw new IllegalStateException(badArrive(s));
361 <            }
362 <            else if (UNSAFE.compareAndSwapLong(this, stateOffset, s, s-=adj)) {
363 <                if (unarrived == 0) {
358 >            int counts = (int)s;
359 >            int unarrived = (counts == EMPTY) ? 0 : (counts & UNARRIVED_MASK);
360 >            if (unarrived <= 0)
361 >                throw new IllegalStateException(badArrive(s));
362 >            if (UNSAFE.compareAndSwapLong(this, stateOffset, s, s-=adjust)) {
363 >                if (unarrived == 1) {
364                      long n = s & PARTIES_MASK;  // base of next state
365 <                    int nextUnarrived = ((int)n) >>> PARTIES_SHIFT;
366 <                    if (root != this)
367 <                        return parent.doArrive(nextUnarrived == 0);
368 <                    if (onAdvance(phase, nextUnarrived))
369 <                        n |= TERMINATION_BIT;
370 <                    else if (nextUnarrived == 0)
371 <                        n |= EMPTY;
365 >                    int nextUnarrived = (int)n >>> PARTIES_SHIFT;
366 >                    if (root == this) {
367 >                        if (onAdvance(phase, nextUnarrived))
368 >                            n |= TERMINATION_BIT;
369 >                        else if (nextUnarrived == 0)
370 >                            n |= EMPTY;
371 >                        else
372 >                            n |= nextUnarrived;
373 >                        int nextPhase = (phase + 1) & MAX_PHASE;
374 >                        n |= (long)nextPhase << PHASE_SHIFT;
375 >                        UNSAFE.compareAndSwapLong(this, stateOffset, s, n);
376 >                        releaseWaiters(phase);
377 >                    }
378 >                    else if (nextUnarrived == 0) { // propagate deregistration
379 >                        phase = parent.doArrive(ONE_DEREGISTER);
380 >                        UNSAFE.compareAndSwapLong(this, stateOffset,
381 >                                                  s, s | EMPTY);
382 >                    }
383                      else
384 <                        n |= nextUnarrived;
375 <                    n |= ((long)((phase + 1) & MAX_PHASE)) << PHASE_SHIFT;
376 <                    UNSAFE.compareAndSwapLong(this, stateOffset, s, n);
377 <                    releaseWaiters(phase);
384 >                        phase = parent.doArrive(ONE_ARRIVAL);
385                  }
386                  return phase;
387              }
# Line 389 | Line 396 | public class Phaser {
396       */
397      private int doRegister(int registrations) {
398          // adjustment to state
399 <        long adj = ((long)registrations << PARTIES_SHIFT) | registrations;
400 <        Phaser par = parent;
399 >        long adjust = ((long)registrations << PARTIES_SHIFT) | registrations;
400 >        final Phaser parent = this.parent;
401          int phase;
402          for (;;) {
403 <            long s = state;
403 >            long s = (parent == null) ? state : reconcileState();
404              int counts = (int)s;
405              int parties = counts >>> PARTIES_SHIFT;
406              int unarrived = counts & UNARRIVED_MASK;
407              if (registrations > MAX_PARTIES - parties)
408                  throw new IllegalStateException(badRegister(s));
409 <            else if ((phase = (int)(s >>> PHASE_SHIFT)) < 0)
409 >            phase = (int)(s >>> PHASE_SHIFT);
410 >            if (phase < 0)
411                  break;
412 <            else if (counts != EMPTY) {             // not 1st registration
413 <                if (par == null || reconcileState() == s) {
412 >            if (counts != EMPTY) {                  // not 1st registration
413 >                if (parent == null || reconcileState() == s) {
414                      if (unarrived == 0)             // wait out advance
415                          root.internalAwaitAdvance(phase, null);
416                      else if (UNSAFE.compareAndSwapLong(this, stateOffset,
417 <                                                       s, s + adj))
417 >                                                       s, s + adjust))
418                          break;
419                  }
420              }
421 <            else if (par == null) {                 // 1st root registration
422 <                long next = (((long) phase) << PHASE_SHIFT) | adj;
421 >            else if (parent == null) {              // 1st root registration
422 >                long next = ((long)phase << PHASE_SHIFT) | adjust;
423                  if (UNSAFE.compareAndSwapLong(this, stateOffset, s, next))
424                      break;
425              }
426              else {
427                  synchronized (this) {               // 1st sub registration
428                      if (state == s) {               // recheck under lock
429 <                        par.doRegister(1);
430 <                        do {                        // force current phase
429 >                        phase = parent.doRegister(1);
430 >                        if (phase < 0)
431 >                            break;
432 >                        // finish registration whenever parent registration
433 >                        // succeeded, even when racing with termination,
434 >                        // since these are part of the same "transaction".
435 >                        while (!UNSAFE.compareAndSwapLong
436 >                               (this, stateOffset, s,
437 >                                ((long)phase << PHASE_SHIFT) | adjust)) {
438 >                            s = state;
439                              phase = (int)(root.state >>> PHASE_SHIFT);
440 <                            // assert phase < 0 || (int)state == EMPTY;
441 <                        } while (!UNSAFE.compareAndSwapLong
426 <                                 (this, stateOffset, state,
427 <                                  (((long) phase) << PHASE_SHIFT) | adj));
440 >                            // assert (int)s == EMPTY;
441 >                        }
442                          break;
443                      }
444                  }
# Line 439 | Line 453 | public class Phaser {
453       * subphasers have not yet done so, in which case they must finish
454       * their own advance by setting unarrived to parties (or if
455       * parties is zero, resetting to unregistered EMPTY state).
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.
456       *
457       * @return reconciled state
458       */
# Line 450 | Line 460 | public class Phaser {
460          final Phaser root = this.root;
461          long s = state;
462          if (root != this) {
463 <            int phase, u, p;
464 <            // CAS root phase with current parties; possibly trip unarrived
463 >            int phase, p;
464 >            // CAS to root phase with current parties, tripping unarrived
465              while ((phase = (int)(root.state >>> PHASE_SHIFT)) !=
466                     (int)(s >>> PHASE_SHIFT) &&
467                     !UNSAFE.compareAndSwapLong
468                     (this, stateOffset, s,
469 <                    s = ((((long) phase) << PHASE_SHIFT) | (s & PARTIES_MASK) |
470 <                         ((p = (int)s >>> PARTIES_SHIFT) == 0 ? EMPTY :
471 <                          (u = (int)s & UNARRIVED_MASK) == 0 ? p : u))))
469 >                    s = (((long)phase << PHASE_SHIFT) |
470 >                         ((phase < 0) ? (s & COUNTS_MASK) :
471 >                          (((p = (int)s >>> PARTIES_SHIFT) == 0) ? EMPTY :
472 >                           ((s & PARTIES_MASK) | p))))))
473                  s = state;
474          }
475          return s;
# Line 525 | Line 536 | public class Phaser {
536              this.evenQ = new AtomicReference<QNode>();
537              this.oddQ = new AtomicReference<QNode>();
538          }
539 <        this.state = (parties == 0) ? (long) EMPTY :
540 <            ((((long) phase) << PHASE_SHIFT) |
541 <             (((long) parties) << PARTIES_SHIFT) |
542 <             ((long) parties));
539 >        this.state = (parties == 0) ? (long)EMPTY :
540 >            ((long)phase << PHASE_SHIFT) |
541 >            ((long)parties << PARTIES_SHIFT) |
542 >            ((long)parties);
543      }
544  
545      /**
# Line 590 | Line 601 | public class Phaser {
601       * of unarrived parties would become negative
602       */
603      public int arrive() {
604 <        return doArrive(false);
604 >        return doArrive(ONE_ARRIVAL);
605      }
606  
607      /**
# Line 610 | Line 621 | public class Phaser {
621       * of registered or unarrived parties would become negative
622       */
623      public int arriveAndDeregister() {
624 <        return doArrive(true);
624 >        return doArrive(ONE_DEREGISTER);
625      }
626  
627      /**
# Line 637 | Line 648 | public class Phaser {
648          for (;;) {
649              long s = (root == this) ? state : reconcileState();
650              int phase = (int)(s >>> PHASE_SHIFT);
640            int counts = (int)s;
641            int unarrived = (counts & UNARRIVED_MASK) - 1;
651              if (phase < 0)
652                  return phase;
653 <            else if (counts == EMPTY || unarrived < 0) {
654 <                if (reconcileState() == s)
655 <                    throw new IllegalStateException(badArrive(s));
656 <            }
657 <            else if (UNSAFE.compareAndSwapLong(this, stateOffset, s,
658 <                                               s -= ONE_ARRIVAL)) {
659 <                if (unarrived != 0)
653 >            int counts = (int)s;
654 >            int unarrived = (counts == EMPTY) ? 0 : (counts & UNARRIVED_MASK);
655 >            if (unarrived <= 0)
656 >                throw new IllegalStateException(badArrive(s));
657 >            if (UNSAFE.compareAndSwapLong(this, stateOffset, s,
658 >                                          s -= ONE_ARRIVAL)) {
659 >                if (unarrived > 1)
660                      return root.internalAwaitAdvance(phase, null);
661                  if (root != this)
662                      return parent.arriveAndAwaitAdvance();
663                  long n = s & PARTIES_MASK;  // base of next state
664 <                int nextUnarrived = ((int)n) >>> PARTIES_SHIFT;
664 >                int nextUnarrived = (int)n >>> PARTIES_SHIFT;
665                  if (onAdvance(phase, nextUnarrived))
666                      n |= TERMINATION_BIT;
667                  else if (nextUnarrived == 0)
# Line 780 | Line 789 | public class Phaser {
789              if (UNSAFE.compareAndSwapLong(root, stateOffset,
790                                            s, s | TERMINATION_BIT)) {
791                  // signal all threads
792 <                releaseWaiters(0);
793 <                releaseWaiters(1);
792 >                releaseWaiters(0); // Waiters on evenQ
793 >                releaseWaiters(1); // Waiters on oddQ
794                  return;
795              }
796          }
# Line 987 | Line 996 | public class Phaser {
996  
997      /**
998       * Possibly blocks and waits for phase to advance unless aborted.
999 <     * Call only from root node.
999 >     * Call only on root phaser.
1000       *
1001       * @param phase current phase
1002       * @param node if non-null, the wait node to track interrupt and timeout;
# Line 995 | Line 1004 | public class Phaser {
1004       * @return current phase
1005       */
1006      private int internalAwaitAdvance(int phase, QNode node) {
1007 +        // assert root == this;
1008          releaseWaiters(phase-1);          // ensure old queue clean
1009          boolean queued = false;           // true when node is enqueued
1010          int lastUnarrived = 0;            // to increase spins upon change
# Line 1108 | Line 1118 | public class Phaser {
1118  
1119      // Unsafe mechanics
1120  
1121 <    private static final sun.misc.Unsafe UNSAFE = getUnsafe();
1122 <    private static final long stateOffset =
1123 <        objectFieldOffset("state", Phaser.class);
1114 <
1115 <    private static long objectFieldOffset(String field, Class<?> klazz) {
1121 >    private static final sun.misc.Unsafe UNSAFE;
1122 >    private static final long stateOffset;
1123 >    static {
1124          try {
1125 <            return UNSAFE.objectFieldOffset(klazz.getDeclaredField(field));
1126 <        } catch (NoSuchFieldException e) {
1127 <            // Convert Exception to corresponding Error
1128 <            NoSuchFieldError error = new NoSuchFieldError(field);
1129 <            error.initCause(e);
1130 <            throw error;
1125 >            UNSAFE = getUnsafe();
1126 >            Class<?> k = Phaser.class;
1127 >            stateOffset = UNSAFE.objectFieldOffset
1128 >                (k.getDeclaredField("state"));
1129 >        } catch (Exception e) {
1130 >            throw new Error(e);
1131          }
1132      }
1133  
# Line 1133 | Line 1141 | public class Phaser {
1141      private static sun.misc.Unsafe getUnsafe() {
1142          try {
1143              return sun.misc.Unsafe.getUnsafe();
1144 <        } catch (SecurityException se) {
1145 <            try {
1146 <                return java.security.AccessController.doPrivileged
1147 <                    (new java.security
1148 <                     .PrivilegedExceptionAction<sun.misc.Unsafe>() {
1149 <                        public sun.misc.Unsafe run() throws Exception {
1150 <                            java.lang.reflect.Field f = sun.misc
1151 <                                .Unsafe.class.getDeclaredField("theUnsafe");
1152 <                            f.setAccessible(true);
1153 <                            return (sun.misc.Unsafe) f.get(null);
1154 <                        }});
1155 <            } catch (java.security.PrivilegedActionException e) {
1156 <                throw new RuntimeException("Could not initialize intrinsics",
1157 <                                           e.getCause());
1158 <            }
1144 >        } catch (SecurityException tryReflectionInstead) {}
1145 >        try {
1146 >            return java.security.AccessController.doPrivileged
1147 >            (new java.security.PrivilegedExceptionAction<sun.misc.Unsafe>() {
1148 >                public sun.misc.Unsafe run() throws Exception {
1149 >                    Class<sun.misc.Unsafe> k = sun.misc.Unsafe.class;
1150 >                    for (java.lang.reflect.Field f : k.getDeclaredFields()) {
1151 >                        f.setAccessible(true);
1152 >                        Object x = f.get(null);
1153 >                        if (k.isInstance(x))
1154 >                            return k.cast(x);
1155 >                    }
1156 >                    throw new NoSuchFieldError("the Unsafe");
1157 >                }});
1158 >        } catch (java.security.PrivilegedActionException e) {
1159 >            throw new RuntimeException("Could not initialize intrinsics",
1160 >                                       e.getCause());
1161          }
1162      }
1163   }

Diff Legend

Removed lines
+ Added lines
< Changed lines
> Changed lines