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.22 by jsr166, Sun Jul 26 17:33:37 2009 UTC vs.
Revision 1.40 by dl, Mon Aug 24 12:49:39 2009 UTC

# Line 12 | Line 12 | import java.util.concurrent.atomic.Atomi
12   import java.util.concurrent.locks.LockSupport;
13  
14   /**
15 < * A reusable synchronization barrier, similar in functionality to a
15 > * A reusable synchronization barrier, similar in functionality to
16   * {@link java.util.concurrent.CyclicBarrier CyclicBarrier} and
17   * {@link java.util.concurrent.CountDownLatch CountDownLatch}
18   * but supporting more flexible usage.
19   *
20 < * <ul>
21 < *
22 < * <li> The number of parties synchronizing on a phaser may vary over
23 < * time.  A task may register to be a party at any time, and may
24 < * deregister upon arriving at the barrier.  As is the case with most
25 < * basic synchronization constructs, registration and deregistration
26 < * affect only internal counts; they do not establish any further
27 < * internal bookkeeping, so tasks cannot query whether they are
28 < * registered. (However, you can introduce such bookkeeping by
29 < * subclassing this class.)
30 < *
31 < * <li> Each generation has an associated phase value, starting at
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:
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
24 > * constructors establishing initial numbers of parties), and
25 > * optionally deregistered upon any arrival (using {@link
26 > * #arriveAndDeregister}).  As is the case with most basic
27 > * synchronization constructs, registration and deregistration affect
28 > * only internal counts; they do not establish any further internal
29 > * bookkeeping, so tasks cannot query whether they are registered.
30 > * (However, you can introduce such bookkeeping by subclassing this
31 > * class.)
32 > *
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
37 > * generation of a {@code Phaser} has an associated phase number. The
38 > * phase number starts at zero, and advances when all parties arrive
39 > * at the barrier, wrapping around to zero after reaching {@code
40 > * Integer.MAX_VALUE}. The use of phase numbers enables independent
41 > * control of actions upon arrival at a barrier and upon awaiting
42 > * others, via two kinds of methods that may be invoked by any
43 > * registered party:
44   *
45   * <ul>
46   *
47 < *   <li> Arriving at a barrier. Methods {@code arrive} and
48 < *       {@code arriveAndDeregister} do not block, but return
49 < *       the phase value current upon entry to the method.
50 < *
51 < *   <li> Awaiting others. Method {@code awaitAdvance} requires an
52 < *       argument indicating the entry phase, and returns when the
53 < *       barrier advances to a new phase.
54 < * </ul>
47 > *   <li> <b>Arrival.</b> Methods {@link #arrive} and
48 > *       {@link #arriveAndDeregister} record arrival at a
49 > *       barrier. These methods do not block, but return an associated
50 > *       <em>arrival phase number</em>; that is, the phase number of
51 > *       the barrier to which the arrival applied. When the final
52 > *       party for a given phase arrives, an optional barrier action
53 > *       is performed and the phase advances.  Barrier actions,
54 > *       performed by the party triggering a phase advance, are
55 > *       arranged by overriding method {@link #onAdvance(int, int)},
56 > *       which also controls termination. Overriding this method is
57 > *       similar to, but more flexible than, providing a barrier
58 > *       action to a {@code CyclicBarrier}.
59 > *
60 > *   <li> <b>Waiting.</b> Method {@link #awaitAdvance} requires an
61 > *       argument indicating an arrival phase number, and returns when
62 > *       the barrier advances to (or is already at) a different phase.
63 > *       Unlike similar constructions using {@code CyclicBarrier},
64 > *       method {@code awaitAdvance} continues to wait even if the
65 > *       waiting thread is interrupted. Interruptible and timeout
66 > *       versions are also available, but exceptions encountered while
67 > *       tasks wait interruptibly or with timeout do not change the
68 > *       state of the barrier. If necessary, you can perform any
69 > *       associated recovery within handlers of those exceptions,
70 > *       often after invoking {@code forceTermination}.  Phasers may
71 > *       also be used by tasks executing in a {@link ForkJoinPool},
72 > *       which will ensure sufficient parallelism to execute tasks
73 > *       when others are blocked waiting for a phase to advance.
74   *
75 + * </ul>
76   *
77 < * <li> Barrier actions, performed by the task triggering a phase
78 < * advance while others may be waiting, are arranged by overriding
79 < * method {@code onAdvance}, that also controls termination.
80 < * Overriding this method may be used to similar but more flexible
81 < * effect as providing a barrier action to a CyclicBarrier.
82 < *
83 < * <li> Phasers may enter a <em>termination</em> state in which all
84 < * actions immediately return without updating phaser state or waiting
85 < * for advance, and indicating (via a negative phase value) that
86 < * execution is complete.  Termination is triggered by executing the
87 < * overridable {@code onAdvance} method that is invoked each time the
63 < * barrier is about to be tripped. When a Phaser is controlling an
64 < * action with a fixed number of iterations, it is often convenient to
65 < * override this method to cause termination when the current phase
66 < * number reaches a threshold. Method {@code forceTermination} is also
67 < * available to abruptly release waiting threads and allow them to
68 < * terminate.
77 > * <p> <b>Termination.</b> A {@code Phaser} may enter a
78 > * <em>termination</em> state in which all synchronization methods
79 > * immediately return without updating phaser state or waiting for
80 > * advance, and indicating (via a negative phase value) that execution
81 > * is complete.  Termination is triggered when an invocation of {@code
82 > * onAdvance} returns {@code true}.  As illustrated below, when
83 > * phasers control actions with a fixed number of iterations, it is
84 > * often convenient to override this method to cause termination when
85 > * the current phase number reaches a threshold. Method {@link
86 > * #forceTermination} is also available to abruptly release waiting
87 > * threads and allow them to terminate.
88   *
89 < * <li> Phasers may be tiered to reduce contention. Phasers with large
89 > * <p> <b>Tiering.</b> Phasers may be <em>tiered</em> (i.e., arranged
90 > * in tree structures) to reduce contention. Phasers with large
91   * numbers of parties that would otherwise experience heavy
92 < * synchronization contention costs may instead be arranged in trees.
93 < * This will typically greatly increase throughput even though it
94 < * incurs somewhat greater per-operation overhead.
95 < *
96 < * <li> By default, {@code awaitAdvance} continues to wait even if
97 < * the waiting thread is interrupted. And unlike the case in
98 < * CyclicBarriers, exceptions encountered while tasks wait
99 < * interruptibly or with timeout do not change the state of the
100 < * barrier. If necessary, you can perform any associated recovery
101 < * within handlers of those exceptions, often after invoking
102 < * {@code forceTermination}.
103 < *
104 < * <li>Phasers ensure lack of starvation when used by ForkJoinTasks.
105 < *
106 < * </ul>
92 > * synchronization contention costs may instead be set up so that
93 > * groups of sub-phasers share a common parent.  This may greatly
94 > * increase throughput even though it incurs greater per-operation
95 > * overhead.
96 > *
97 > * <p><b>Monitoring.</b> While synchronization methods may be invoked
98 > * only by registered parties, the current state of a phaser may be
99 > * monitored by any caller.  At any given moment there are {@link
100 > * #getRegisteredParties} parties in total, of which {@link
101 > * #getArrivedParties} have arrived at the current phase ({@link
102 > * #getPhase}).  When the remaining ({@link #getUnarrivedParties})
103 > * parties arrive, the phase advances; thus, this value is always
104 > * greater than zero if there are any registered parties.  The values
105 > * returned by these methods may reflect transient states and so are
106 > * not in general useful for synchronization control.  Method {@link
107 > * #toString} returns snapshots of these state queries in a form
108 > * convenient for informal monitoring.
109   *
110   * <p><b>Sample usages:</b>
111   *
112 < * <p>A Phaser may be used instead of a {@code CountDownLatch} to control
113 < * a one-shot action serving a variable number of parties. The typical
114 < * idiom is for the method setting this up to first register, then
115 < * start the actions, then deregister, as in:
112 > * <p>A {@code Phaser} may be used instead of a {@code CountDownLatch}
113 > * to control a one-shot action serving a variable number of
114 > * parties. The typical idiom is for the method setting this up to
115 > * first register, then start the actions, then deregister, as in:
116   *
117   *  <pre> {@code
118 < * void runTasks(List<Runnable> list) {
118 > * void runTasks(List<Runnable> tasks) {
119   *   final Phaser phaser = new Phaser(1); // "1" to register self
120 < *   for (Runnable r : list) {
120 > *   // create and start threads
121 > *   for (Runnable task : tasks) {
122   *     phaser.register();
123   *     new Thread() {
124   *       public void run() {
125   *         phaser.arriveAndAwaitAdvance(); // await all creation
126 < *         r.run();
104 < *         phaser.arriveAndDeregister();   // signal completion
126 > *         task.run();
127   *       }
128   *     }.start();
129   *   }
130   *
131 < *   doSomethingOnBehalfOfWorkers();
132 < *   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
131 > *   // allow threads to start and deregister self
132 > *   phaser.arriveAndDeregister();
133   * }}</pre>
134   *
135   * <p>One way to cause a set of threads to repeatedly perform actions
136   * for a given number of iterations is to override {@code onAdvance}:
137   *
138   *  <pre> {@code
139 < * void startTasks(List<Runnable> list, final int iterations) {
139 > * void startTasks(List<Runnable> tasks, final int iterations) {
140   *   final Phaser phaser = new Phaser() {
141 < *     public boolean onAdvance(int phase, int registeredParties) {
141 > *     protected boolean onAdvance(int phase, int registeredParties) {
142   *       return phase >= iterations || registeredParties == 0;
143   *     }
144   *   };
145   *   phaser.register();
146 < *   for (Runnable r : list) {
146 > *   for (Runnable task : tasks) {
147   *     phaser.register();
148   *     new Thread() {
149   *       public void run() {
150   *         do {
151 < *           r.run();
151 > *           task.run();
152   *           phaser.arriveAndAwaitAdvance();
153   *         } while(!phaser.isTerminated();
154   *       }
# Line 139 | Line 157 | import java.util.concurrent.locks.LockSu
157   *   phaser.arriveAndDeregister(); // deregister self, don't wait
158   * }}</pre>
159   *
160 < * <p> To create a set of tasks using a tree of Phasers,
160 > * If the main task must later await termination, it
161 > * may re-register and then execute a similar loop:
162 > * <pre> {@code
163 > *   // ...
164 > *   phaser.register();
165 > *   while (!phaser.isTerminated())
166 > *     phaser.arriveAndAwaitAdvance();
167 > * }</pre>
168 > *
169 > * Related constructions may be used to await particular phase numbers
170 > * in contexts where you are sure that the phase will never wrap around
171 > * {@code Integer.MAX_VALUE}. For example:
172 > *
173 > * <pre> {@code
174 > *   void awaitPhase(Phaser phaser, int phase) {
175 > *     int p = phaser.register(); // assumes caller not already registered
176 > *     while (p < phase) {
177 > *       if (phaser.isTerminated())
178 > *         // ... deal with unexpected termination
179 > *       else
180 > *         p = phaser.arriveAndAwaitAdvance();
181 > *     }
182 > *     phaser.arriveAndDeregister();
183 > *   }
184 > * }</pre>
185 > *
186 > *
187 > * <p>To create a set of tasks using a tree of phasers,
188   * you could use code of the following form, assuming a
189 < * Task class with a constructor accepting a Phaser that
189 > * Task class with a constructor accepting a phaser that
190   * it registers for upon construction:
191   *  <pre> {@code
192   * void build(Task[] actions, int lo, int hi, Phaser b) {
# Line 171 | Line 216 | import java.util.concurrent.locks.LockSu
216   *
217   * <p><b>Implementation notes</b>: This implementation restricts the
218   * maximum number of parties to 65535. Attempts to register additional
219 < * parties result in IllegalStateExceptions. However, you can and
219 > * parties result in {@code IllegalStateException}. However, you can and
220   * should create tiered phasers to accommodate arbitrarily large sets
221   * of participants.
222   *
# Line 249 | Line 294 | public class Phaser {
294      private final Phaser parent;
295  
296      /**
297 <     * The root of Phaser tree. Equals this if not in a tree.  Used to
297 >     * The root of phaser tree. Equals this if not in a tree.  Used to
298       * support faster state push-down.
299       */
300      private final Phaser root;
# Line 300 | Line 345 | public class Phaser {
345      }
346  
347      /**
348 <     * Creates a new Phaser without any initially registered parties,
348 >     * Creates a new phaser without any initially registered parties,
349       * initial phase number 0, and no parent. Any thread using this
350 <     * Phaser will need to first register for it.
350 >     * phaser will need to first register for it.
351       */
352      public Phaser() {
353          this(null);
354      }
355  
356      /**
357 <     * Creates a new Phaser with the given numbers of registered
357 >     * Creates a new phaser with the given numbers of registered
358       * unarrived parties, initial phase number 0, and no parent.
359       *
360       * @param parties the number of parties required to trip barrier
# Line 321 | Line 366 | public class Phaser {
366      }
367  
368      /**
369 <     * Creates a new Phaser with the given parent, without any
369 >     * Creates a new phaser with the given parent, without any
370       * initially registered parties. If parent is non-null this phaser
371       * is registered with the parent and its initial phase number is
372       * the same as that of parent phaser.
# Line 341 | Line 386 | public class Phaser {
386      }
387  
388      /**
389 <     * Creates a new Phaser with the given parent and numbers of
389 >     * Creates a new phaser with the given parent and numbers of
390       * registered unarrived parties. If parent is non-null, this phaser
391       * is registered with the parent and its initial phase number is
392       * the same as that of parent phaser.
# Line 368 | Line 413 | public class Phaser {
413      /**
414       * Adds a new unarrived party to this phaser.
415       *
416 <     * @return the current barrier phase number upon registration
416 >     * @return the arrival phase number to which this registration applied
417       * @throws IllegalStateException if attempting to register more
418       * than the maximum supported number of parties
419       */
# Line 380 | Line 425 | public class Phaser {
425       * Adds the given number of new unarrived parties to this phaser.
426       *
427       * @param parties the number of parties required to trip barrier
428 <     * @return the current barrier phase number upon registration
428 >     * @return the arrival phase number to which this registration applied
429       * @throws IllegalStateException if attempting to register more
430       * than the maximum supported number of parties
431       */
# Line 415 | Line 460 | public class Phaser {
460  
461      /**
462       * Arrives at the barrier, but does not wait for others.  (You can
463 <     * in turn wait for others via {@link #awaitAdvance}).
463 >     * in turn wait for others via {@link #awaitAdvance}).  It is an
464 >     * unenforced usage error for an unregistered party to invoke this
465 >     * method.
466       *
467 <     * @return the barrier phase number upon entry to this method, or a
421 <     * negative value if terminated
467 >     * @return the arrival phase number, or a negative value if terminated
468       * @throws IllegalStateException if not terminated and the number
469       * of unarrived parties would become negative
470       */
# Line 463 | Line 509 | public class Phaser {
509      }
510  
511      /**
512 <     * Arrives at the barrier, and deregisters from it, without
513 <     * waiting for others. Deregistration reduces number of parties
512 >     * Arrives at the barrier and deregisters from it without waiting
513 >     * for others. Deregistration reduces the number of parties
514       * required to trip the barrier in future phases.  If this phaser
515       * has a parent, and deregistration causes this phaser to have
516 <     * zero parties, this phaser is also deregistered from its parent.
516 >     * zero parties, this phaser also arrives at and is deregistered
517 >     * from its parent.  It is an unenforced usage error for an
518 >     * unregistered party to invoke this method.
519       *
520 <     * @return the current barrier phase number upon entry to
473 <     * this method, or a negative value if terminated
520 >     * @return the arrival phase number, or a negative value if terminated
521       * @throws IllegalStateException if not terminated and the number
522       * of registered or unarrived parties would become negative
523       */
# Line 520 | Line 567 | public class Phaser {
567  
568      /**
569       * Arrives at the barrier and awaits others. Equivalent in effect
570 <     * to {@code awaitAdvance(arrive())}.  If you instead need to
571 <     * await with interruption of timeout, and/or deregister upon
572 <     * arrival, you can arrange them using analogous constructions.
570 >     * to {@code awaitAdvance(arrive())}.  If you need to await with
571 >     * interruption or timeout, you can arrange this with an analogous
572 >     * construction using one of the other forms of the awaitAdvance
573 >     * method.  If instead you need to deregister upon arrival use
574 >     * {@code arriveAndDeregister}. It is an unenforced usage error
575 >     * for an unregistered party to invoke this method.
576       *
577 <     * @return the phase on entry to this method
577 >     * @return the arrival phase number, or a negative number if terminated
578       * @throws IllegalStateException if not terminated and the number
579       * of unarrived parties would become negative
580       */
# Line 533 | Line 583 | public class Phaser {
583      }
584  
585      /**
586 <     * Awaits the phase of the barrier to advance from the given
587 <     * value, or returns immediately if argument is negative or this
588 <     * barrier is terminated.
589 <     *
590 <     * @param phase the phase on entry to this method
591 <     * @return the phase on exit from this method
586 >     * Awaits the phase of the barrier to advance from the given phase
587 >     * value, returning immediately if the current phase of the
588 >     * barrier is not equal to the given phase value or this barrier
589 >     * is terminated.  It is an unenforced usage error for an
590 >     * unregistered party to invoke this method.
591 >     *
592 >     * @param phase an arrival phase number, or negative value if
593 >     * terminated; this argument is normally the value returned by a
594 >     * previous call to {@code arrive} or its variants
595 >     * @return the next arrival phase number, or a negative value
596 >     * if terminated or argument is negative
597       */
598      public int awaitAdvance(int phase) {
599          if (phase < 0)
# Line 554 | Line 609 | public class Phaser {
609      }
610  
611      /**
612 <     * Awaits the phase of the barrier to advance from the given
613 <     * value, or returns immediately if argument is negative or this
614 <     * barrier is terminated, or throws InterruptedException if
615 <     * interrupted while waiting.
616 <     *
617 <     * @param phase the phase on entry to this method
618 <     * @return the phase on exit from this method
612 >     * Awaits the phase of the barrier to advance from the given phase
613 >     * value, throwing {@code InterruptedException} if interrupted
614 >     * while waiting, or returning immediately if the current phase of
615 >     * the barrier is not equal to the given phase value or this
616 >     * barrier is terminated. It is an unenforced usage error for an
617 >     * unregistered party to invoke this method.
618 >     *
619 >     * @param phase an arrival phase number, or negative value if
620 >     * terminated; this argument is normally the value returned by a
621 >     * previous call to {@code arrive} or its variants
622 >     * @return the next arrival phase number, or a negative value
623 >     * if terminated or argument is negative
624       * @throws InterruptedException if thread interrupted while waiting
625       */
626      public int awaitAdvanceInterruptibly(int phase)
# Line 577 | Line 637 | public class Phaser {
637      }
638  
639      /**
640 <     * Awaits the phase of the barrier to advance from the given value
641 <     * or the given timeout elapses, or returns immediately if
642 <     * argument is negative or this barrier is terminated.
643 <     *
644 <     * @param phase the phase on entry to this method
645 <     * @return the phase on exit from this method
640 >     * Awaits the phase of the barrier to advance from the given phase
641 >     * value or the given timeout to elapse, throwing {@code
642 >     * InterruptedException} if interrupted while waiting, or
643 >     * returning immediately if the current phase of the barrier is
644 >     * not equal to the given phase value or this barrier is
645 >     * terminated.  It is an unenforced usage error for an
646 >     * unregistered party to invoke this method.
647 >     *
648 >     * @param phase an arrival phase number, or negative value if
649 >     * terminated; this argument is normally the value returned by a
650 >     * previous call to {@code arrive} or its variants
651 >     * @param timeout how long to wait before giving up, in units of
652 >     *        {@code unit}
653 >     * @param unit a {@code TimeUnit} determining how to interpret the
654 >     *        {@code timeout} parameter
655 >     * @return the next arrival phase number, or a negative value
656 >     * if terminated or argument is negative
657       * @throws InterruptedException if thread interrupted while waiting
658       * @throws TimeoutException if timed out while waiting
659       */
# Line 636 | Line 707 | public class Phaser {
707      }
708  
709      /**
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    /**
710       * Returns the number of parties registered at this barrier.
711       *
712       * @return the number of parties
# Line 655 | Line 716 | public class Phaser {
716      }
717  
718      /**
719 <     * Returns the number of parties that have arrived at the current
720 <     * phase of this barrier.
719 >     * Returns the number of registered parties that have arrived at
720 >     * the current phase of this barrier.
721       *
722       * @return the number of arrived parties
723       */
# Line 675 | Line 736 | public class Phaser {
736      }
737  
738      /**
739 <     * Returns the parent of this phaser, or null if none.
739 >     * Returns the parent of this phaser, or {@code null} if none.
740       *
741 <     * @return the parent of this phaser, or null if none
741 >     * @return the parent of this phaser, or {@code null} if none
742       */
743      public Phaser getParent() {
744          return parent;
# Line 706 | Line 767 | public class Phaser {
767       * Overridable method to perform an action upon phase advance, and
768       * to control termination. This method is invoked whenever the
769       * barrier is tripped (and thus all other waiting parties are
770 <     * dormant). If it returns true, then, rather than advance the
771 <     * phase number, this barrier will be set to a final termination
772 <     * state, and subsequent calls to {@code isTerminated} will
773 <     * return true.
770 >     * dormant). If it returns {@code true}, then, rather than advance
771 >     * the phase number, this barrier will be set to a final
772 >     * termination state, and subsequent calls to {@link #isTerminated}
773 >     * will return true.
774       *
775 <     * <p> The default version returns true when the number of
775 >     * <p>The default version returns {@code true} when the number of
776       * registered parties is zero. Normally, overrides that arrange
777       * termination for other reasons should also preserve this
778       * property.
779       *
780 <     * <p> You may override this method to perform an action with side
780 >     * <p>You may override this method to perform an action with side
781       * effects visible to participating tasks, but it is in general
782       * only sensible to do so in designs where all parties register
783 <     * before any arrive, and all {@code awaitAdvance} at each phase.
784 <     * Otherwise, you cannot ensure lack of interference. In
785 <     * 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.
783 >     * before any arrive, and all {@link #awaitAdvance} at each phase.
784 >     * Otherwise, you cannot ensure lack of interference from other
785 >     * parties during the invocation of this method.
786       *
787       * @param phase the phase number on entering the barrier
788       * @param registeredParties the current number of registered parties

Diff Legend

Removed lines
+ Added lines
< Changed lines
> Changed lines