ViewVC Help
View File | Revision Log | Show Annotations | Download File | Root Listing
root/jsr166/jsr166/src/main/java/util/concurrent/Phaser.java
(Generate patch)

Comparing jsr166/src/main/java/util/concurrent/Phaser.java (file contents):
Revision 1.9 by jsr166, Thu Aug 20 16:38:43 2009 UTC vs.
Revision 1.10 by jsr166, Mon Aug 31 23:39:57 2009 UTC

# Line 10 | Line 10 | import java.util.concurrent.atomic.Atomi
10   import java.util.concurrent.locks.LockSupport;
11  
12   /**
13 < * A reusable synchronization barrier, similar in functionality to a
13 > * A reusable synchronization barrier, similar in functionality to
14   * {@link java.util.concurrent.CyclicBarrier CyclicBarrier} and
15   * {@link java.util.concurrent.CountDownLatch CountDownLatch}
16   * but supporting more flexible usage.
17   *
18 < * <ul>
19 < *
20 < * <li> The number of parties synchronizing on a phaser may vary over
21 < * time.  A task may register to be a party at any time, and may
22 < * deregister upon arriving at the barrier.  As is the case with most
23 < * basic synchronization constructs, registration and deregistration
24 < * affect only internal counts; they do not establish any further
25 < * internal bookkeeping, so tasks cannot query whether they are
26 < * registered. (However, you can introduce such bookkeeping by
27 < * subclassing this class.)
28 < *
29 < * <li> Each generation has an associated phase value, starting at
30 < * zero, and advancing when all parties reach the barrier (wrapping
31 < * around to zero after reaching {@code Integer.MAX_VALUE}).
32 < *
33 < * <li> Like a {@code CyclicBarrier}, a phaser may be repeatedly
34 < * awaited.  Method {@link #arriveAndAwaitAdvance} has effect
35 < * analogous to {@link java.util.concurrent.CyclicBarrier#await
36 < * CyclicBarrier.await}.  However, phasers separate two aspects of
37 < * coordination, which may also be invoked independently:
18 > * <p> <b>Registration.</b> Unlike the case for other barriers, the
19 > * number of parties <em>registered</em> to synchronize on a phaser
20 > * may vary over time.  Tasks may be registered at any time (using
21 > * methods {@link #register}, {@link #bulkRegister}, or forms of
22 > * constructors establishing initial numbers of parties), and
23 > * optionally deregistered upon any arrival (using {@link
24 > * #arriveAndDeregister}).  As is the case with most basic
25 > * synchronization constructs, registration and deregistration affect
26 > * only internal counts; they do not establish any further internal
27 > * bookkeeping, so tasks cannot query whether they are registered.
28 > * (However, you can introduce such bookkeeping by subclassing this
29 > * class.)
30 > *
31 > * <p> <b>Synchronization.</b> Like a {@code CyclicBarrier}, a {@code
32 > * Phaser} may be repeatedly awaited.  Method {@link
33 > * #arriveAndAwaitAdvance} has effect analogous to {@link
34 > * java.util.concurrent.CyclicBarrier#await CyclicBarrier.await}. Each
35 > * generation of a {@code Phaser} has an associated phase number. The
36 > * phase number starts at zero, and advances when all parties arrive
37 > * at the barrier, wrapping around to zero after reaching {@code
38 > * Integer.MAX_VALUE}. The use of phase numbers enables independent
39 > * control of actions upon arrival at a barrier and upon awaiting
40 > * others, via two kinds of methods that may be invoked by any
41 > * registered party:
42   *
43   * <ul>
44   *
45 < *   <li> Arriving at a barrier. Methods {@link #arrive} and
46 < *       {@link #arriveAndDeregister} do not block, but return
47 < *       the phase value current upon entry to the method.
48 < *
49 < *   <li> Awaiting others. Method {@link #awaitAdvance} requires an
50 < *       argument indicating the entry phase, and returns when the
51 < *       barrier advances to a new phase.
52 < * </ul>
45 > *   <li> <b>Arrival.</b> Methods {@link #arrive} and
46 > *       {@link #arriveAndDeregister} record arrival at a
47 > *       barrier. These methods do not block, but return an associated
48 > *       <em>arrival phase number</em>; that is, the phase number of
49 > *       the barrier to which the arrival applied. When the final
50 > *       party for a given phase arrives, an optional barrier action
51 > *       is performed and the phase advances.  Barrier actions,
52 > *       performed by the party triggering a phase advance, are
53 > *       arranged by overriding method {@link #onAdvance(int, int)},
54 > *       which also controls termination. Overriding this method is
55 > *       similar to, but more flexible than, providing a barrier
56 > *       action to a {@code CyclicBarrier}.
57 > *
58 > *   <li> <b>Waiting.</b> Method {@link #awaitAdvance} requires an
59 > *       argument indicating an arrival phase number, and returns when
60 > *       the barrier advances to (or is already at) a different phase.
61 > *       Unlike similar constructions using {@code CyclicBarrier},
62 > *       method {@code awaitAdvance} continues to wait even if the
63 > *       waiting thread is interrupted. Interruptible and timeout
64 > *       versions are also available, but exceptions encountered while
65 > *       tasks wait interruptibly or with timeout do not change the
66 > *       state of the barrier. If necessary, you can perform any
67 > *       associated recovery within handlers of those exceptions,
68 > *       often after invoking {@code forceTermination}.  Phasers may
69 > *       also be used by tasks executing in a {@link ForkJoinPool},
70 > *       which will ensure sufficient parallelism to execute tasks
71 > *       when others are blocked waiting for a phase to advance.
72   *
73 + * </ul>
74   *
75 < * <li> Barrier actions, performed by the task triggering a phase
76 < * advance, are arranged by overriding method {@link #onAdvance(int,
77 < * int)}, which also controls termination. Overriding this method is
78 < * similar to, but more flexible than, providing a barrier action to a
79 < * {@code CyclicBarrier}.
80 < *
81 < * <li> Phasers may enter a <em>termination</em> state in which all
58 < * actions immediately return without updating phaser state or waiting
59 < * for advance, and indicating (via a negative phase value) that
60 < * execution is complete.  Termination is triggered when an invocation
61 < * of {@code onAdvance} returns {@code true}.  When a phaser is
62 < * controlling an action with a fixed number of iterations, it is
75 > * <p> <b>Termination.</b> A {@code Phaser} may enter a
76 > * <em>termination</em> state in which all synchronization methods
77 > * immediately return without updating phaser state or waiting for
78 > * advance, and indicating (via a negative phase value) that execution
79 > * is complete.  Termination is triggered when an invocation of {@code
80 > * onAdvance} returns {@code true}.  As illustrated below, when
81 > * phasers control actions with a fixed number of iterations, it is
82   * often convenient to override this method to cause termination when
83   * the current phase number reaches a threshold. Method {@link
84   * #forceTermination} is also available to abruptly release waiting
85   * threads and allow them to terminate.
86   *
87 < * <li> Phasers may be tiered to reduce contention. Phasers with large
87 > * <p> <b>Tiering.</b> Phasers may be <em>tiered</em> (i.e., arranged
88 > * in tree structures) to reduce contention. Phasers with large
89   * numbers of parties that would otherwise experience heavy
90 < * synchronization contention costs may instead be arranged in trees.
91 < * This will typically greatly increase throughput even though it
92 < * incurs somewhat greater per-operation overhead.
93 < *
94 < * <li> By default, {@code awaitAdvance} continues to wait even if
95 < * the waiting thread is interrupted. And unlike the case in
96 < * {@code CyclicBarrier}, exceptions encountered while tasks wait
97 < * interruptibly or with timeout do not change the state of the
98 < * barrier. If necessary, you can perform any associated recovery
99 < * within handlers of those exceptions, often after invoking
100 < * {@code forceTermination}.
101 < *
102 < * <li>Phasers may be used to coordinate tasks executing in a {@link
103 < * ForkJoinPool}, which will ensure sufficient parallelism to execute
104 < * tasks when others are blocked waiting for a phase to advance.
105 < *
86 < * </ul>
90 > * synchronization contention costs may instead be set up so that
91 > * groups of sub-phasers share a common parent.  This may greatly
92 > * increase throughput even though it incurs greater per-operation
93 > * overhead.
94 > *
95 > * <p><b>Monitoring.</b> While synchronization methods may be invoked
96 > * only by registered parties, the current state of a phaser may be
97 > * monitored by any caller.  At any given moment there are {@link
98 > * #getRegisteredParties} parties in total, of which {@link
99 > * #getArrivedParties} have arrived at the current phase ({@link
100 > * #getPhase}).  When the remaining ({@link #getUnarrivedParties})
101 > * parties arrive, the phase advances.  The values returned by these
102 > * methods may reflect transient states and so are not in general
103 > * useful for synchronization control.  Method {@link #toString}
104 > * returns snapshots of these state queries in a form convenient for
105 > * informal monitoring.
106   *
107   * <p><b>Sample usages:</b>
108   *
# Line 116 | Line 135 | import java.util.concurrent.locks.LockSu
135   *  <pre> {@code
136   * void startTasks(List<Runnable> tasks, final int iterations) {
137   *   final Phaser phaser = new Phaser() {
138 < *     public boolean onAdvance(int phase, int registeredParties) {
138 > *     protected boolean onAdvance(int phase, int registeredParties) {
139   *       return phase >= iterations || registeredParties == 0;
140   *     }
141   *   };
142   *   phaser.register();
143 < *   for (Runnable task : tasks) {
143 > *   for (final Runnable task : tasks) {
144   *     phaser.register();
145   *     new Thread() {
146   *       public void run() {
147   *         do {
148   *           task.run();
149   *           phaser.arriveAndAwaitAdvance();
150 < *         } while(!phaser.isTerminated();
150 > *         } while (!phaser.isTerminated());
151   *       }
152   *     }.start();
153   *   }
154   *   phaser.arriveAndDeregister(); // deregister self, don't wait
155   * }}</pre>
156   *
157 + * If the main task must later await termination, it
158 + * may re-register and then execute a similar loop:
159 + *  <pre> {@code
160 + *   // ...
161 + *   phaser.register();
162 + *   while (!phaser.isTerminated())
163 + *     phaser.arriveAndAwaitAdvance();}</pre>
164 + *
165 + * <p>Related constructions may be used to await particular phase numbers
166 + * in contexts where you are sure that the phase will never wrap around
167 + * {@code Integer.MAX_VALUE}. For example:
168 + *
169 + *  <pre> {@code
170 + * void awaitPhase(Phaser phaser, int phase) {
171 + *   int p = phaser.register(); // assumes caller not already registered
172 + *   while (p < phase) {
173 + *     if (phaser.isTerminated())
174 + *       // ... deal with unexpected termination
175 + *     else
176 + *       p = phaser.arriveAndAwaitAdvance();
177 + *   }
178 + *   phaser.arriveAndDeregister();
179 + * }}</pre>
180 + *
181 + *
182   * <p>To create a set of tasks using a tree of phasers,
183   * you could use code of the following form, assuming a
184   * Task class with a constructor accepting a phaser that
185   * it registers for upon construction:
186 + *
187   *  <pre> {@code
188 < * void build(Task[] actions, int lo, int hi, Phaser b) {
189 < *   int step = (hi - lo) / TASKS_PER_PHASER;
190 < *   if (step > 1) {
191 < *     int i = lo;
192 < *     while (i < hi) {
148 < *       int r = Math.min(i + step, hi);
149 < *       build(actions, i, r, new Phaser(b));
150 < *       i = r;
188 > * void build(Task[] actions, int lo, int hi, Phaser ph) {
189 > *   if (hi - lo > TASKS_PER_PHASER) {
190 > *     for (int i = lo; i < hi; i += TASKS_PER_PHASER) {
191 > *       int j = Math.min(i + TASKS_PER_PHASER, hi);
192 > *       build(actions, i, j, new Phaser(ph));
193   *     }
194   *   } else {
195   *     for (int i = lo; i < hi; ++i)
196 < *       actions[i] = new Task(b);
197 < *       // assumes new Task(b) performs b.register()
196 > *       actions[i] = new Task(ph);
197 > *       // assumes new Task(ph) performs ph.register()
198   *   }
199   * }
200   * // .. initially called, for n tasks via
# Line 201 | Line 243 | public class Phaser {
243       */
244      private volatile long state;
245  
204    private static final int ushortBits = 16;
246      private static final int ushortMask = 0xffff;
247      private static final int phaseMask  = 0x7fffffff;
248  
# Line 364 | Line 405 | public class Phaser {
405      /**
406       * Adds a new unarrived party to this phaser.
407       *
408 <     * @return the current barrier phase number upon registration
408 >     * @return the arrival phase number to which this registration applied
409       * @throws IllegalStateException if attempting to register more
410       * than the maximum supported number of parties
411       */
# Line 376 | Line 417 | public class Phaser {
417       * Adds the given number of new unarrived parties to this phaser.
418       *
419       * @param parties the number of parties required to trip barrier
420 <     * @return the current barrier phase number upon registration
420 >     * @return the arrival phase number to which this registration applied
421       * @throws IllegalStateException if attempting to register more
422       * than the maximum supported number of parties
423       */
# Line 411 | Line 452 | public class Phaser {
452  
453      /**
454       * Arrives at the barrier, but does not wait for others.  (You can
455 <     * in turn wait for others via {@link #awaitAdvance}).
455 >     * in turn wait for others via {@link #awaitAdvance}).  It is an
456 >     * unenforced usage error for an unregistered party to invoke this
457 >     * method.
458       *
459 <     * @return the barrier phase number upon entry to this method, or a
417 <     * negative value if terminated
459 >     * @return the arrival phase number, or a negative value if terminated
460       * @throws IllegalStateException if not terminated and the number
461       * of unarrived parties would become negative
462       */
# Line 464 | Line 506 | public class Phaser {
506       * required to trip the barrier in future phases.  If this phaser
507       * has a parent, and deregistration causes this phaser to have
508       * zero parties, this phaser also arrives at and is deregistered
509 <     * from its parent.
509 >     * from its parent.  It is an unenforced usage error for an
510 >     * unregistered party to invoke this method.
511       *
512 <     * @return the current barrier phase number upon entry to
470 <     * this method, or a negative value if terminated
512 >     * @return the arrival phase number, or a negative value if terminated
513       * @throws IllegalStateException if not terminated and the number
514       * of registered or unarrived parties would become negative
515       */
# Line 521 | Line 563 | public class Phaser {
563       * interruption or timeout, you can arrange this with an analogous
564       * construction using one of the other forms of the awaitAdvance
565       * method.  If instead you need to deregister upon arrival use
566 <     * {@code arriveAndDeregister}.
566 >     * {@code arriveAndDeregister}. It is an unenforced usage error
567 >     * for an unregistered party to invoke this method.
568       *
569 <     * @return the phase on entry to this method
569 >     * @return the arrival phase number, or a negative number if terminated
570       * @throws IllegalStateException if not terminated and the number
571       * of unarrived parties would become negative
572       */
# Line 535 | Line 578 | public class Phaser {
578       * Awaits the phase of the barrier to advance from the given phase
579       * value, returning immediately if the current phase of the
580       * barrier is not equal to the given phase value or this barrier
581 <     * is terminated.
581 >     * is terminated.  It is an unenforced usage error for an
582 >     * unregistered party to invoke this method.
583       *
584 <     * @param phase the phase on entry to this method
585 <     * @return the current barrier phase number upon exit of
586 <     * this method, or a negative value if terminated or
587 <     * argument is negative
584 >     * @param phase an arrival phase number, or negative value if
585 >     * terminated; this argument is normally the value returned by a
586 >     * previous call to {@code arrive} or its variants
587 >     * @return the next arrival phase number, or a negative value
588 >     * if terminated or argument is negative
589       */
590      public int awaitAdvance(int phase) {
591          if (phase < 0)
# Line 557 | Line 602 | public class Phaser {
602  
603      /**
604       * Awaits the phase of the barrier to advance from the given phase
605 <     * value, throwing {@code InterruptedException} if interrupted while
606 <     * waiting, or returning immediately if the current phase of the
607 <     * barrier is not equal to the given phase value or this barrier
608 <     * is terminated.
609 <     *
610 <     * @param phase the phase on entry to this method
611 <     * @return the current barrier phase number upon exit of
612 <     * this method, or a negative value if terminated or
613 <     * argument is negative
605 >     * value, throwing {@code InterruptedException} if interrupted
606 >     * while waiting, or returning immediately if the current phase of
607 >     * the barrier is not equal to the given phase value or this
608 >     * barrier is terminated. It is an unenforced usage error for an
609 >     * unregistered party to invoke this method.
610 >     *
611 >     * @param phase an arrival phase number, or negative value if
612 >     * terminated; this argument is normally the value returned by a
613 >     * previous call to {@code arrive} or its variants
614 >     * @return the next arrival phase number, or a negative value
615 >     * if terminated or argument is negative
616       * @throws InterruptedException if thread interrupted while waiting
617       */
618      public int awaitAdvanceInterruptibly(int phase)
# Line 583 | Line 630 | public class Phaser {
630  
631      /**
632       * Awaits the phase of the barrier to advance from the given phase
633 <     * value or the given timeout to elapse, throwing
634 <     * {@code InterruptedException} if interrupted while waiting, or
635 <     * returning immediately if the current phase of the barrier is not
636 <     * equal to the given phase value or this barrier is terminated.
637 <     *
638 <     * @param phase the phase on entry to this method
633 >     * value or the given timeout to elapse, throwing {@code
634 >     * InterruptedException} if interrupted while waiting, or
635 >     * returning immediately if the current phase of the barrier is
636 >     * not equal to the given phase value or this barrier is
637 >     * terminated.  It is an unenforced usage error for an
638 >     * unregistered party to invoke this method.
639 >     *
640 >     * @param phase an arrival phase number, or negative value if
641 >     * terminated; this argument is normally the value returned by a
642 >     * previous call to {@code arrive} or its variants
643       * @param timeout how long to wait before giving up, in units of
644       *        {@code unit}
645       * @param unit a {@code TimeUnit} determining how to interpret the
646       *        {@code timeout} parameter
647 <     * @return the current barrier phase number upon exit of
648 <     * this method, or a negative value if terminated or
598 <     * argument is negative
647 >     * @return the next arrival phase number, or a negative value
648 >     * if terminated or argument is negative
649       * @throws InterruptedException if thread interrupted while waiting
650       * @throws TimeoutException if timed out while waiting
651       */
# Line 658 | Line 708 | public class Phaser {
708      }
709  
710      /**
711 <     * Returns the number of parties that have arrived at the current
712 <     * phase of this barrier.
711 >     * Returns the number of registered parties that have arrived at
712 >     * the current phase of this barrier.
713       *
714       * @return the number of arrived parties
715       */
# Line 706 | Line 756 | public class Phaser {
756      }
757  
758      /**
759 <     * Overridable method to perform an action upon phase advance, and
760 <     * to control termination. This method is invoked whenever the
761 <     * barrier is tripped (and thus all other waiting parties are
762 <     * dormant). If it returns {@code true}, then, rather than advance
763 <     * the phase number, this barrier will be set to a final
764 <     * termination state, and subsequent calls to {@link #isTerminated}
765 <     * will return true.
759 >     * Overridable method to perform an action upon impending phase
760 >     * advance, and to control termination. This method is invoked
761 >     * upon arrival of the party tripping the barrier (when all other
762 >     * waiting parties are dormant).  If this method returns {@code
763 >     * true}, then, rather than advance the phase number, this barrier
764 >     * will be set to a final termination state, and subsequent calls
765 >     * to {@link #isTerminated} will return true. Any (unchecked)
766 >     * Exception or Error thrown by an invocation of this method is
767 >     * propagated to the party attempting to trip the barrier, in
768 >     * which case no advance occurs.
769 >     *
770 >     * <p>The arguments to this method provide the state of the phaser
771 >     * prevailing for the current transition. (When called from within
772 >     * an implementation of {@code onAdvance} the values returned by
773 >     * methods such as {@code getPhase} may or may not reliably
774 >     * indicate the state to which this transition applies.)
775       *
776       * <p>The default version returns {@code true} when the number of
777       * registered parties is zero. Normally, overrides that arrange
# Line 720 | Line 779 | public class Phaser {
779       * property.
780       *
781       * <p>You may override this method to perform an action with side
782 <     * effects visible to participating tasks, but it is in general
783 <     * only sensible to do so in designs where all parties register
784 <     * before any arrive, and all {@link #awaitAdvance} at each phase.
782 >     * effects visible to participating tasks, but it is only sensible
783 >     * to do so in designs where all parties register before any
784 >     * arrive, and all {@link #awaitAdvance} at each phase.
785       * Otherwise, you cannot ensure lack of interference from other
786 <     * parties during the invocation of this method.
786 >     * parties during the invocation of this method. Additionally,
787 >     * method {@code onAdvance} may be invoked more than once per
788 >     * transition if registrations are intermixed with arrivals.
789       *
790       * @param phase the phase number on entering the barrier
791       * @param registeredParties the current number of registered parties

Diff Legend

Removed lines
+ Added lines
< Changed lines
> Changed lines