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.21 by jsr166, Sun Jul 26 05:55:34 2009 UTC vs.
Revision 1.29 by jsr166, Wed Aug 12 04:02:45 2009 UTC

# Line 32 | Line 32 | import java.util.concurrent.locks.LockSu
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
68 < * 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 75 | Line 75 | import java.util.concurrent.locks.LockSu
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();
104 *         phaser.arriveAndDeregister();   // signal completion
107   *       }
108   *     }.start();
109   *   }
110   *
111 < *   doSomethingOnBehalfOfWorkers();
112 < *   phaser.arrive(); // allow threads to start
111 < *   int p = phaser.arriveAndDeregister(); // deregister self  ...
112 < *   p = phaser.awaitAdvance(p); // ... and await arrival
113 < *   otherActions(); // do other things while tasks execute
114 < *   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 139 | Line 137 | import java.util.concurrent.locks.LockSu
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 249 | 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 300 | 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 321 | 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 341 | 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 463 | 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 520 | 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 533 | 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 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 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 <    // Unsafe mechanics for jsr166y 3rd party package.
921 >    // Unsafe mechanics
922 >
923 >    private static final sun.misc.Unsafe UNSAFE = getUnsafe();
924 >    private static final long stateOffset =
925 >        objectFieldOffset("state", Phaser.class);
926 >
927 >    private final boolean casState(long cmp, long val) {
928 >        return UNSAFE.compareAndSwapLong(this, stateOffset, cmp, val);
929 >    }
930 >
931 >    private static long objectFieldOffset(String field, Class<?> klazz) {
932 >        try {
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 >    /**
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.PrivilegedExceptionAction<sun.misc.Unsafe>() {
955 >                    (new java.security
956 >                     .PrivilegedExceptionAction<sun.misc.Unsafe>() {
957                          public sun.misc.Unsafe run() throws Exception {
958 <                            return getUnsafeByReflection();
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",
# Line 947 | Line 966 | public class Phaser {
966              }
967          }
968      }
950
951    private static sun.misc.Unsafe getUnsafeByReflection()
952            throws NoSuchFieldException, IllegalAccessException {
953        java.lang.reflect.Field f =
954            sun.misc.Unsafe.class.getDeclaredField("theUnsafe");
955        f.setAccessible(true);
956        return (sun.misc.Unsafe) f.get(null);
957    }
958
959    private static long fieldOffset(String fieldName, Class<?> klazz) {
960        try {
961            return UNSAFE.objectFieldOffset(klazz.getDeclaredField(fieldName));
962        } catch (NoSuchFieldException e) {
963            // Convert Exception to Error
964            NoSuchFieldError error = new NoSuchFieldError(fieldName);
965            error.initCause(e);
966            throw error;
967        }
968    }
969
970    private static final sun.misc.Unsafe UNSAFE = getUnsafe();
971    private static final long stateOffset =
972        fieldOffset("state", Phaser.class);
973
974    private final boolean casState(long cmp, long val) {
975        return UNSAFE.compareAndSwapLong(this, stateOffset, cmp, val);
976    }
969   }

Diff Legend

Removed lines
+ Added lines
< Changed lines
> Changed lines