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.28 by jsr166, Wed Aug 12 02:24:35 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 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, or returns immediately if the current phase of the barrier
539 >     * is not equal to the given phase value or this barrier is
540 >     * terminated.
541       *
542       * @param phase the phase on entry to this method
543       * @return the phase on exit from this method
# Line 587 | Line 588 | public class Phaser {
588       * @throws InterruptedException if thread interrupted while waiting
589       * @throws TimeoutException if timed out while waiting
590       */
591 <    public int awaitAdvanceInterruptibly(int phase, long timeout, TimeUnit unit)
591 >    public int awaitAdvanceInterruptibly(int phase,
592 >                                         long timeout, TimeUnit unit)
593          throws InterruptedException, TimeoutException {
594          if (phase < 0)
595              return phase;
# Line 636 | Line 638 | public class Phaser {
638      }
639  
640      /**
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    /**
641       * Returns the number of parties registered at this barrier.
642       *
643       * @return the number of parties
# Line 675 | Line 667 | public class Phaser {
667      }
668  
669      /**
670 <     * Returns the parent of this phaser, or null if none.
670 >     * Returns the parent of this phaser, or {@code null} if none.
671       *
672 <     * @return the parent of this phaser, or null if none
672 >     * @return the parent of this phaser, or {@code null} if none
673       */
674      public Phaser getParent() {
675          return parent;
# Line 706 | Line 698 | public class Phaser {
698       * Overridable method to perform an action upon phase advance, and
699       * to control termination. This method is invoked whenever the
700       * barrier is tripped (and thus all other waiting parties are
701 <     * dormant). If it returns true, then, rather than advance the
702 <     * phase number, this barrier will be set to a final termination
703 <     * state, and subsequent calls to {@code isTerminated} will
704 <     * return true.
701 >     * dormant). If it returns {@code true}, then, rather than advance
702 >     * the phase number, this barrier will be set to a final
703 >     * termination state, and subsequent calls to {@link #isTerminated}
704 >     * will return true.
705       *
706 <     * <p> The default version returns true when the number of
706 >     * <p>The default version returns {@code true} when the number of
707       * registered parties is zero. Normally, overrides that arrange
708       * termination for other reasons should also preserve this
709       * property.
710       *
711 <     * <p> You may override this method to perform an action with side
711 >     * <p>You may override this method to perform an action with side
712       * effects visible to participating tasks, but it is in general
713       * only sensible to do so in designs where all parties register
714 <     * before any arrive, and all {@code awaitAdvance} at each phase.
715 <     * Otherwise, you cannot ensure lack of interference. In
716 <     * 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.
714 >     * before any arrive, and all {@link #awaitAdvance} at each phase.
715 >     * Otherwise, you cannot ensure lack of interference from other
716 >     * parties during the the invocation of this method.
717       *
718       * @param phase the phase number on entering the barrier
719       * @param registeredParties the current number of registered parties
# Line 930 | Line 918 | public class Phaser {
918          return p;
919      }
920  
921 <    // 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 <    }
921 >    // Unsafe mechanics
922  
923 <    private static Unsafe getUnsafePrivileged()
924 <            throws NoSuchFieldException, IllegalAccessException {
925 <        Field f = Unsafe.class.getDeclaredField("theUnsafe");
953 <        f.setAccessible(true);
954 <        return (Unsafe) f.get(null);
955 <    }
923 >    private static final sun.misc.Unsafe UNSAFE = getUnsafe();
924 >    private static final long stateOffset =
925 >        objectFieldOffset("state", Phaser.class);
926  
927 <    private static long fieldOffset(String fieldName)
928 <            throws NoSuchFieldException {
959 <        return UNSAFE.objectFieldOffset
960 <            (Phaser.class.getDeclaredField(fieldName));
927 >    private final boolean casState(long cmp, long val) {
928 >        return UNSAFE.compareAndSwapLong(this, stateOffset, cmp, val);
929      }
930  
931 <    static final Unsafe UNSAFE;
964 <    static final long stateOffset;
965 <
966 <    static {
931 >    private static long objectFieldOffset(String field, Class<?> klazz) {
932          try {
933 <            UNSAFE = getUnsafe();
934 <            stateOffset = fieldOffset("state");
935 <        } catch (Throwable e) {
936 <            throw new RuntimeException("Could not initialize intrinsics", e);
933 >            return UNSAFE.objectFieldOffset(klazz.getDeclaredField(field));
934 >        } catch (NoSuchFieldException e) {
935 >            // Convert Exception to corresponding Error
936 >            NoSuchFieldError error = new NoSuchFieldError(field);
937 >            error.initCause(e);
938 >            throw error;
939          }
940      }
941  
942 <    final boolean casState(long cmp, long val) {
943 <        return UNSAFE.compareAndSwapLong(this, stateOffset, cmp, val);
942 >    /**
943 >     * Returns a sun.misc.Unsafe.  Suitable for use in a 3rd party package.
944 >     * Replace with a simple call to Unsafe.getUnsafe when integrating
945 >     * into a jdk.
946 >     *
947 >     * @return a sun.misc.Unsafe
948 >     */
949 >    private static sun.misc.Unsafe getUnsafe() {
950 >        try {
951 >            return sun.misc.Unsafe.getUnsafe();
952 >        } catch (SecurityException se) {
953 >            try {
954 >                return java.security.AccessController.doPrivileged
955 >                    (new java.security
956 >                     .PrivilegedExceptionAction<sun.misc.Unsafe>() {
957 >                        public sun.misc.Unsafe run() throws Exception {
958 >                            java.lang.reflect.Field f = sun.misc
959 >                                .Unsafe.class.getDeclaredField("theUnsafe");
960 >                            f.setAccessible(true);
961 >                            return (sun.misc.Unsafe) f.get(null);
962 >                        }});
963 >            } catch (java.security.PrivilegedActionException e) {
964 >                throw new RuntimeException("Could not initialize intrinsics",
965 >                                           e.getCause());
966 >            }
967 >        }
968      }
969   }

Diff Legend

Removed lines
+ Added lines
< Changed lines
> Changed lines