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.16 by jsr166, Wed Jul 22 01:36:51 2009 UTC vs.
Revision 1.32 by jsr166, Wed Aug 19 17:44:45 2009 UTC

# Line 7 | Line 7
7   package jsr166y;
8  
9   import java.util.concurrent.*;
10 < import java.util.concurrent.atomic.*;
10 >
11 > import java.util.concurrent.atomic.AtomicReference;
12   import java.util.concurrent.locks.LockSupport;
12 import sun.misc.Unsafe;
13 import java.lang.reflect.*;
13  
14   /**
15   * A reusable synchronization barrier, similar in functionality to a
# Line 33 | Line 32 | import java.lang.reflect.*;
32   * zero, and advancing when all parties reach the barrier (wrapping
33   * around to zero after reaching {@code Integer.MAX_VALUE}).
34   *
35 < * <li> Like a CyclicBarrier, a Phaser may be repeatedly awaited.
36 < * Method {@code arriveAndAwaitAdvance} has effect analogous to
37 < * {@code CyclicBarrier.await}.  However, Phasers separate two
38 < * aspects of coordination, that may also be invoked independently:
35 > * <li> Like a {@code CyclicBarrier}, a phaser may be repeatedly
36 > * awaited.  Method {@link #arriveAndAwaitAdvance} has effect
37 > * analogous to {@link java.util.concurrent.CyclicBarrier#await
38 > * CyclicBarrier.await}.  However, phasers separate two aspects of
39 > * coordination, which may also be invoked independently:
40   *
41   * <ul>
42   *
43 < *   <li> Arriving at a barrier. Methods {@code arrive} and
44 < *       {@code arriveAndDeregister} do not block, but return
43 > *   <li> Arriving at a barrier. Methods {@link #arrive} and
44 > *       {@link #arriveAndDeregister} do not block, but return
45   *       the phase value current upon entry to the method.
46   *
47 < *   <li> Awaiting others. Method {@code awaitAdvance} requires an
47 > *   <li> Awaiting others. Method {@link #awaitAdvance} requires an
48   *       argument indicating the entry phase, and returns when the
49   *       barrier advances to a new phase.
50   * </ul>
51   *
52   *
53   * <li> Barrier actions, performed by the task triggering a phase
54 < * advance while others may be waiting, are arranged by overriding
55 < * method {@code onAdvance}, that also controls termination.
56 < * Overriding this method may be used to similar but more flexible
57 < * effect as providing a barrier action to a CyclicBarrier.
54 > * advance, are arranged by overriding method {@link #onAdvance(int,
55 > * int)}, which also controls termination. Overriding this method is
56 > * similar to, but more flexible than, providing a barrier action to a
57 > * {@code CyclicBarrier}.
58   *
59   * <li> Phasers may enter a <em>termination</em> state in which all
60   * actions immediately return without updating phaser state or waiting
61   * for advance, and indicating (via a negative phase value) that
62 < * execution is complete.  Termination is triggered by executing the
63 < * overridable {@code onAdvance} method that is invoked each time the
64 < * barrier is about to be tripped. When a Phaser is controlling an
65 < * action with a fixed number of iterations, it is often convenient to
66 < * override this method to cause termination when the current phase
67 < * number reaches a threshold. Method {@code forceTermination} is also
68 < * available to abruptly release waiting threads and allow them to
69 < * terminate.
62 > * execution is complete.  Termination is triggered when an invocation
63 > * of {@code onAdvance} returns {@code true}.  When a phaser is
64 > * controlling an action with a fixed number of iterations, it is
65 > * often convenient to override this method to cause termination when
66 > * the current phase number reaches a threshold. Method {@link
67 > * #forceTermination} is also available to abruptly release waiting
68 > * threads and allow them to terminate.
69   *
70   * <li> Phasers may be tiered to reduce contention. Phasers with large
71   * numbers of parties that would otherwise experience heavy
# Line 76 | Line 75 | import java.lang.reflect.*;
75   *
76   * <li> By default, {@code awaitAdvance} continues to wait even if
77   * the waiting thread is interrupted. And unlike the case in
78 < * CyclicBarriers, exceptions encountered while tasks wait
78 > * {@code CyclicBarrier}, exceptions encountered while tasks wait
79   * interruptibly or with timeout do not change the state of the
80   * barrier. If necessary, you can perform any associated recovery
81   * within handlers of those exceptions, often after invoking
82   * {@code forceTermination}.
83   *
84 < * <li>Phasers ensure lack of starvation when used by ForkJoinTasks.
84 > * <li>Phasers may be used to coordinate tasks executing in a {@link
85 > * ForkJoinPool}, which will ensure sufficient parallelism to execute
86 > * tasks when others are blocked waiting for a phase to advance.
87   *
88   * </ul>
89   *
90   * <p><b>Sample usages:</b>
91   *
92 < * <p>A Phaser may be used instead of a {@code CountDownLatch} to control
93 < * a one-shot action serving a variable number of parties. The typical
94 < * idiom is for the method setting this up to first register, then
95 < * start the actions, then deregister, as in:
92 > * <p>A {@code Phaser} may be used instead of a {@code CountDownLatch}
93 > * to control a one-shot action serving a variable number of
94 > * parties. The typical idiom is for the method setting this up to
95 > * first register, then start the actions, then deregister, as in:
96   *
97   *  <pre> {@code
98   * void runTasks(List<Runnable> list) {
99   *   final Phaser phaser = new Phaser(1); // "1" to register self
100 + *   // create and start threads
101   *   for (Runnable r : list) {
102   *     phaser.register();
103   *     new Thread() {
104   *       public void run() {
105   *         phaser.arriveAndAwaitAdvance(); // await all creation
106   *         r.run();
105 *         phaser.arriveAndDeregister();   // signal completion
107   *       }
108   *     }.start();
109   *   }
110   *
111 < *   doSomethingOnBehalfOfWorkers();
112 < *   phaser.arrive(); // allow threads to start
112 < *   int p = phaser.arriveAndDeregister(); // deregister self  ...
113 < *   p = phaser.awaitAdvance(p); // ... and await arrival
114 < *   otherActions(); // do other things while tasks execute
115 < *   phaser.awaitAdvance(p); // await final completion
111 > *   // allow threads to start and deregister self
112 > *   phaser.arriveAndDeregister();
113   * }}</pre>
114   *
115   * <p>One way to cause a set of threads to repeatedly perform actions
# Line 140 | Line 137 | import java.lang.reflect.*;
137   *   phaser.arriveAndDeregister(); // deregister self, don't wait
138   * }}</pre>
139   *
140 < * <p> To create a set of tasks using a tree of Phasers,
140 > * <p>To create a set of tasks using a tree of phasers,
141   * you could use code of the following form, assuming a
142 < * Task class with a constructor accepting a Phaser that
142 > * Task class with a constructor accepting a phaser that
143   * it registers for upon construction:
144   *  <pre> {@code
145   * void build(Task[] actions, int lo, int hi, Phaser b) {
# Line 172 | Line 169 | import java.lang.reflect.*;
169   *
170   * <p><b>Implementation notes</b>: This implementation restricts the
171   * maximum number of parties to 65535. Attempts to register additional
172 < * parties result in IllegalStateExceptions. However, you can and
172 > * parties result in {@code IllegalStateException}. However, you can and
173   * should create tiered phasers to accommodate arbitrarily large sets
174   * of participants.
175   *
# Line 211 | Line 208 | public class Phaser {
208      private static final int phaseMask  = 0x7fffffff;
209  
210      private static int unarrivedOf(long s) {
211 <        return (int)(s & ushortMask);
211 >        return (int) (s & ushortMask);
212      }
213  
214      private static int partiesOf(long s) {
215 <        return ((int)s) >>> 16;
215 >        return ((int) s) >>> 16;
216      }
217  
218      private static int phaseOf(long s) {
219 <        return (int)(s >>> 32);
219 >        return (int) (s >>> 32);
220      }
221  
222      private static int arrivedOf(long s) {
# Line 227 | Line 224 | public class Phaser {
224      }
225  
226      private static long stateFor(int phase, int parties, int unarrived) {
227 <        return ((((long)phase) << 32) | (((long)parties) << 16) |
228 <                (long)unarrived);
227 >        return ((((long) phase) << 32) | (((long) parties) << 16) |
228 >                (long) unarrived);
229      }
230  
231      private static long trippedStateFor(int phase, int parties) {
232 <        long lp = (long)parties;
233 <        return (((long)phase) << 32) | (lp << 16) | lp;
232 >        long lp = (long) parties;
233 >        return (((long) phase) << 32) | (lp << 16) | lp;
234      }
235  
236      /**
# Line 250 | Line 247 | public class Phaser {
247      private final Phaser parent;
248  
249      /**
250 <     * The root of Phaser tree. Equals this if not in a tree.  Used to
250 >     * The root of phaser tree. Equals this if not in a tree.  Used to
251       * support faster state push-down.
252       */
253      private final Phaser root;
# Line 266 | Line 263 | public class Phaser {
263      private final AtomicReference<QNode> oddQ  = new AtomicReference<QNode>();
264  
265      private AtomicReference<QNode> queueFor(int phase) {
266 <        return (phase & 1) == 0? evenQ : oddQ;
266 >        return ((phase & 1) == 0) ? evenQ : oddQ;
267      }
268  
269      /**
# Line 274 | Line 271 | public class Phaser {
271       * root if necessary.
272       */
273      private long getReconciledState() {
274 <        return parent == null? state : reconcileState();
274 >        return (parent == null) ? state : reconcileState();
275      }
276  
277      /**
# Line 301 | Line 298 | public class Phaser {
298      }
299  
300      /**
301 <     * Creates a new Phaser without any initially registered parties,
301 >     * Creates a new phaser without any initially registered parties,
302       * initial phase number 0, and no parent. Any thread using this
303 <     * Phaser will need to first register for it.
303 >     * phaser will need to first register for it.
304       */
305      public Phaser() {
306          this(null);
307      }
308  
309      /**
310 <     * Creates a new Phaser with the given numbers of registered
310 >     * Creates a new phaser with the given numbers of registered
311       * unarrived parties, initial phase number 0, and no parent.
312       *
313       * @param parties the number of parties required to trip barrier
# Line 322 | Line 319 | public class Phaser {
319      }
320  
321      /**
322 <     * Creates a new Phaser with the given parent, without any
322 >     * Creates a new phaser with the given parent, without any
323       * initially registered parties. If parent is non-null this phaser
324       * is registered with the parent and its initial phase number is
325       * the same as that of parent phaser.
# Line 342 | Line 339 | public class Phaser {
339      }
340  
341      /**
342 <     * Creates a new Phaser with the given parent and numbers of
342 >     * Creates a new phaser with the given parent and numbers of
343       * registered unarrived parties. If parent is non-null, this phaser
344       * is registered with the parent and its initial phase number is
345       * the same as that of parent phaser.
# Line 441 | Line 438 | public class Phaser {
438                  if (par == null) {      // directly trip
439                      if (casState
440                          (s,
441 <                         trippedStateFor(onAdvance(phase, parties)? -1 :
441 >                         trippedStateFor(onAdvance(phase, parties) ? -1 :
442                                           ((phase + 1) & phaseMask), parties))) {
443                          releaseWaiters(phase);
444                          break;
# Line 464 | Line 461 | public class Phaser {
461      }
462  
463      /**
464 <     * Arrives at the barrier, and deregisters from it, without
465 <     * waiting for others. Deregistration reduces number of parties
464 >     * Arrives at the barrier and deregisters from it without waiting
465 >     * for others. Deregistration reduces the number of parties
466       * required to trip the barrier in future phases.  If this phaser
467       * has a parent, and deregistration causes this phaser to have
468 <     * zero parties, this phaser is also deregistered from its parent.
468 >     * zero parties, this phaser also arrives at and is deregistered
469 >     * from its parent.
470       *
471       * @return the current barrier phase number upon entry to
472       * this method, or a negative value if terminated
# Line 502 | Line 500 | public class Phaser {
500                  if (unarrived == 0) {
501                      if (casState
502                          (s,
503 <                         trippedStateFor(onAdvance(phase, parties)? -1 :
503 >                         trippedStateFor(onAdvance(phase, parties) ? -1 :
504                                           ((phase + 1) & phaseMask), parties))) {
505                          releaseWaiters(phase);
506                          break;
# Line 521 | Line 519 | public class Phaser {
519  
520      /**
521       * Arrives at the barrier and awaits others. Equivalent in effect
522 <     * to {@code awaitAdvance(arrive())}.  If you instead need to
523 <     * await with interruption of timeout, and/or deregister upon
524 <     * arrival, you can arrange them using analogous constructions.
522 >     * to {@code awaitAdvance(arrive())}.  If you need to await with
523 >     * interruption or timeout, you can arrange this with an analogous
524 >     * construction using one of the other forms of the awaitAdvance
525 >     * method.  If instead you need to deregister upon arrival use
526 >     * {@code arriveAndDeregister}.
527       *
528       * @return the phase on entry to this method
529       * @throws IllegalStateException if not terminated and the number
# Line 534 | Line 534 | public class Phaser {
534      }
535  
536      /**
537 <     * Awaits the phase of the barrier to advance from the given
538 <     * value, or returns immediately if argument is negative or this
539 <     * barrier is terminated.
537 >     * Awaits the phase of the barrier to advance from the given phase
538 >     * value, returning immediately if the current phase of the
539 >     * barrier is not equal to the given phase value or this barrier
540 >     * is terminated.
541       *
542       * @param phase the phase on entry to this method
543       * @return the phase on exit from this method
# Line 555 | Line 556 | public class Phaser {
556      }
557  
558      /**
559 <     * Awaits the phase of the barrier to advance from the given
560 <     * value, or returns immediately if argument is negative or this
561 <     * barrier is terminated, or throws InterruptedException if
562 <     * interrupted while waiting.
559 >     * Awaits the phase of the barrier to advance from the given phase
560 >     * value, throwing {@code InterruptedException} if interrupted while
561 >     * waiting, or returning immediately if the current phase of the
562 >     * barrier is not equal to the given phase value or this barrier
563 >     * is terminated.
564       *
565       * @param phase the phase on entry to this method
566       * @return the phase on exit from this method
# Line 578 | Line 580 | public class Phaser {
580      }
581  
582      /**
583 <     * Awaits the phase of the barrier to advance from the given value
584 <     * or the given timeout elapses, or returns immediately if
585 <     * argument is negative or this barrier is terminated.
583 >     * Awaits the phase of the barrier to advance from the given phase
584 >     * value or the given timeout to elapse, throwing
585 >     * {@code InterruptedException} if interrupted while waiting, or
586 >     * returning immediately if the current phase of the barrier is not
587 >     * equal to the given phase value or this barrier is terminated.
588       *
589       * @param phase the phase on entry to this method
590 +     * @param timeout how long to wait before giving up, in units of
591 +     *        {@code unit}
592 +     * @param unit a {@code TimeUnit} determining how to interpret the
593 +     *        {@code timeout} parameter
594       * @return the phase on exit from this method
595       * @throws InterruptedException if thread interrupted while waiting
596       * @throws TimeoutException if timed out while waiting
597       */
598 <    public int awaitAdvanceInterruptibly(int phase, long timeout, TimeUnit unit)
598 >    public int awaitAdvanceInterruptibly(int phase,
599 >                                         long timeout, TimeUnit unit)
600          throws InterruptedException, TimeoutException {
601          if (phase < 0)
602              return phase;
# Line 636 | Line 645 | public class Phaser {
645      }
646  
647      /**
639     * Returns {@code true} if the current phase number equals the given phase.
640     *
641     * @param phase the phase
642     * @return {@code true} if the current phase number equals the given phase
643     */
644    public final boolean hasPhase(int phase) {
645        return phaseOf(getReconciledState()) == phase;
646    }
647
648    /**
648       * Returns the number of parties registered at this barrier.
649       *
650       * @return the number of parties
# Line 675 | Line 674 | public class Phaser {
674      }
675  
676      /**
677 <     * Returns the parent of this phaser, or null if none.
677 >     * Returns the parent of this phaser, or {@code null} if none.
678       *
679 <     * @return the parent of this phaser, or null if none
679 >     * @return the parent of this phaser, or {@code null} if none
680       */
681      public Phaser getParent() {
682          return parent;
# Line 706 | Line 705 | public class Phaser {
705       * Overridable method to perform an action upon phase advance, and
706       * to control termination. This method is invoked whenever the
707       * barrier is tripped (and thus all other waiting parties are
708 <     * dormant). If it returns true, then, rather than advance the
709 <     * phase number, this barrier will be set to a final termination
710 <     * state, and subsequent calls to {@code isTerminated} will
711 <     * return true.
708 >     * dormant). If it returns {@code true}, then, rather than advance
709 >     * the phase number, this barrier will be set to a final
710 >     * termination state, and subsequent calls to {@link #isTerminated}
711 >     * will return true.
712       *
713 <     * <p> The default version returns true when the number of
713 >     * <p>The default version returns {@code true} when the number of
714       * registered parties is zero. Normally, overrides that arrange
715       * termination for other reasons should also preserve this
716       * property.
717       *
718 <     * <p> You may override this method to perform an action with side
718 >     * <p>You may override this method to perform an action with side
719       * effects visible to participating tasks, but it is in general
720       * only sensible to do so in designs where all parties register
721 <     * before any arrive, and all {@code awaitAdvance} at each phase.
722 <     * Otherwise, you cannot ensure lack of interference. In
723 <     * particular, this method may be invoked more than once per
725 <     * transition if other parties successfully register while the
726 <     * invocation of this method is in progress, thus postponing the
727 <     * transition until those parties also arrive, re-triggering this
728 <     * method.
721 >     * before any arrive, and all {@link #awaitAdvance} at each phase.
722 >     * Otherwise, you cannot ensure lack of interference from other
723 >     * parties during the invocation of this method.
724       *
725       * @param phase the phase number on entering the barrier
726       * @param registeredParties the current number of registered parties
# Line 930 | Line 925 | public class Phaser {
925          return p;
926      }
927  
928 <    // Temporary Unsafe mechanics for preliminary release
934 <    private static Unsafe getUnsafe() throws Throwable {
935 <        try {
936 <            return Unsafe.getUnsafe();
937 <        } catch (SecurityException se) {
938 <            try {
939 <                return java.security.AccessController.doPrivileged
940 <                    (new java.security.PrivilegedExceptionAction<Unsafe>() {
941 <                        public Unsafe run() throws Exception {
942 <                            return getUnsafePrivileged();
943 <                        }});
944 <            } catch (java.security.PrivilegedActionException e) {
945 <                throw e.getCause();
946 <            }
947 <        }
948 <    }
928 >    // Unsafe mechanics
929  
930 <    private static Unsafe getUnsafePrivileged()
931 <            throws NoSuchFieldException, IllegalAccessException {
932 <        Field f = Unsafe.class.getDeclaredField("theUnsafe");
953 <        f.setAccessible(true);
954 <        return (Unsafe) f.get(null);
955 <    }
930 >    private static final sun.misc.Unsafe UNSAFE = getUnsafe();
931 >    private static final long stateOffset =
932 >        objectFieldOffset("state", Phaser.class);
933  
934 <    private static long fieldOffset(String fieldName)
935 <            throws NoSuchFieldException {
959 <        return UNSAFE.objectFieldOffset
960 <            (Phaser.class.getDeclaredField(fieldName));
934 >    private final boolean casState(long cmp, long val) {
935 >        return UNSAFE.compareAndSwapLong(this, stateOffset, cmp, val);
936      }
937  
938 <    static final Unsafe UNSAFE;
964 <    static final long stateOffset;
965 <
966 <    static {
938 >    private static long objectFieldOffset(String field, Class<?> klazz) {
939          try {
940 <            UNSAFE = getUnsafe();
941 <            stateOffset = fieldOffset("state");
942 <        } catch (Throwable e) {
943 <            throw new RuntimeException("Could not initialize intrinsics", e);
940 >            return UNSAFE.objectFieldOffset(klazz.getDeclaredField(field));
941 >        } catch (NoSuchFieldException e) {
942 >            // Convert Exception to corresponding Error
943 >            NoSuchFieldError error = new NoSuchFieldError(field);
944 >            error.initCause(e);
945 >            throw error;
946          }
947      }
948  
949 <    final boolean casState(long cmp, long val) {
950 <        return UNSAFE.compareAndSwapLong(this, stateOffset, cmp, val);
949 >    /**
950 >     * Returns a sun.misc.Unsafe.  Suitable for use in a 3rd party package.
951 >     * Replace with a simple call to Unsafe.getUnsafe when integrating
952 >     * into a jdk.
953 >     *
954 >     * @return a sun.misc.Unsafe
955 >     */
956 >    private static sun.misc.Unsafe getUnsafe() {
957 >        try {
958 >            return sun.misc.Unsafe.getUnsafe();
959 >        } catch (SecurityException se) {
960 >            try {
961 >                return java.security.AccessController.doPrivileged
962 >                    (new java.security
963 >                     .PrivilegedExceptionAction<sun.misc.Unsafe>() {
964 >                        public sun.misc.Unsafe run() throws Exception {
965 >                            java.lang.reflect.Field f = sun.misc
966 >                                .Unsafe.class.getDeclaredField("theUnsafe");
967 >                            f.setAccessible(true);
968 >                            return (sun.misc.Unsafe) f.get(null);
969 >                        }});
970 >            } catch (java.security.PrivilegedActionException e) {
971 >                throw new RuntimeException("Could not initialize intrinsics",
972 >                                           e.getCause());
973 >            }
974 >        }
975      }
976   }

Diff Legend

Removed lines
+ Added lines
< Changed lines
> Changed lines