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.11 by jsr166, Thu Mar 19 04:49:44 2009 UTC vs.
Revision 1.23 by jsr166, Mon Jul 27 20:57:44 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 93 | Line 92 | import java.lang.reflect.*;
92   * idiom is for the method setting this up to first register, then
93   * start the actions, then deregister, as in:
94   *
95 < * <pre>
96 < *  void runTasks(List&lt;Runnable&gt; list) {
97 < *    final Phaser phaser = new Phaser(1); // "1" to register self
98 < *    for (Runnable r : list) {
99 < *      phaser.register();
100 < *      new Thread() {
101 < *        public void run() {
102 < *          phaser.arriveAndAwaitAdvance(); // await all creation
103 < *          r.run();
104 < *          phaser.arriveAndDeregister();   // signal completion
105 < *        }
106 < *      }.start();
95 > *  <pre> {@code
96 > * void runTasks(List<Runnable> list) {
97 > *   final Phaser phaser = new Phaser(1); // "1" to register self
98 > *   for (Runnable r : list) {
99 > *     phaser.register();
100 > *     new Thread() {
101 > *       public void run() {
102 > *         phaser.arriveAndAwaitAdvance(); // await all creation
103 > *         r.run();
104 > *         phaser.arriveAndDeregister();   // signal completion
105 > *       }
106 > *     }.start();
107   *   }
108   *
109   *   doSomethingOnBehalfOfWorkers();
# Line 113 | Line 112 | import java.lang.reflect.*;
112   *   p = phaser.awaitAdvance(p); // ... and await arrival
113   *   otherActions(); // do other things while tasks execute
114   *   phaser.awaitAdvance(p); // await final completion
115 < * }
117 < * </pre>
115 > * }}</pre>
116   *
117   * <p>One way to cause a set of threads to repeatedly perform actions
118   * for a given number of iterations is to override {@code onAdvance}:
119   *
120 < * <pre>
121 < *  void startTasks(List&lt;Runnable&gt; list, final int iterations) {
122 < *    final Phaser phaser = new Phaser() {
123 < *       public boolean onAdvance(int phase, int registeredParties) {
124 < *         return phase &gt;= iterations || registeredParties == 0;
120 > *  <pre> {@code
121 > * void startTasks(List<Runnable> list, final int iterations) {
122 > *   final Phaser phaser = new Phaser() {
123 > *     public boolean onAdvance(int phase, int registeredParties) {
124 > *       return phase >= iterations || registeredParties == 0;
125 > *     }
126 > *   };
127 > *   phaser.register();
128 > *   for (Runnable r : list) {
129 > *     phaser.register();
130 > *     new Thread() {
131 > *       public void run() {
132 > *         do {
133 > *           r.run();
134 > *           phaser.arriveAndAwaitAdvance();
135 > *         } while(!phaser.isTerminated();
136   *       }
137 < *    };
129 < *    phaser.register();
130 < *    for (Runnable r : list) {
131 < *      phaser.register();
132 < *      new Thread() {
133 < *        public void run() {
134 < *           do {
135 < *             r.run();
136 < *             phaser.arriveAndAwaitAdvance();
137 < *           } while(!phaser.isTerminated();
138 < *        }
139 < *      }.start();
137 > *     }.start();
138   *   }
139   *   phaser.arriveAndDeregister(); // deregister self, don't wait
140 < * }
143 < * </pre>
140 > * }}</pre>
141   *
142   * <p> To create a set of tasks using a tree of Phasers,
143   * you could use code of the following form, assuming a
144   * Task class with a constructor accepting a Phaser that
145   * it registers for upon construction:
146 < * <pre>
147 < *  void build(Task[] actions, int lo, int hi, Phaser b) {
148 < *    int step = (hi - lo) / TASKS_PER_PHASER;
149 < *    if (step &gt; 1) {
150 < *       int i = lo;
151 < *       while (i &lt; hi) {
152 < *         int r = Math.min(i + step, hi);
153 < *         build(actions, i, r, new Phaser(b));
154 < *         i = r;
155 < *       }
156 < *    }
157 < *    else {
158 < *      for (int i = lo; i &lt; hi; ++i)
159 < *        actions[i] = new Task(b);
160 < *        // assumes new Task(b) performs b.register()
161 < *    }
162 < *  }
163 < *  // .. initially called, for n tasks via
167 < *  build(new Task[n], 0, n, new Phaser());
168 < * </pre>
146 > *  <pre> {@code
147 > * void build(Task[] actions, int lo, int hi, Phaser b) {
148 > *   int step = (hi - lo) / TASKS_PER_PHASER;
149 > *   if (step > 1) {
150 > *     int i = lo;
151 > *     while (i < hi) {
152 > *       int r = Math.min(i + step, hi);
153 > *       build(actions, i, r, new Phaser(b));
154 > *       i = r;
155 > *     }
156 > *   } else {
157 > *     for (int i = lo; i < hi; ++i)
158 > *       actions[i] = new Task(b);
159 > *       // assumes new Task(b) performs b.register()
160 > *   }
161 > * }
162 > * // .. initially called, for n tasks via
163 > * build(new Task[n], 0, n, new Phaser());}</pre>
164   *
165   * The best value of {@code TASKS_PER_PHASER} depends mainly on
166   * expected barrier synchronization rates. A value as low as four may
# Line 179 | Line 174 | import java.lang.reflect.*;
174   * parties result in IllegalStateExceptions. However, you can and
175   * should create tiered phasers to accommodate arbitrarily large sets
176   * of participants.
177 + *
178 + * @since 1.7
179 + * @author Doug Lea
180   */
181   public class Phaser {
182      /*
# Line 212 | Line 210 | public class Phaser {
210      private static final int phaseMask  = 0x7fffffff;
211  
212      private static int unarrivedOf(long s) {
213 <        return (int)(s & ushortMask);
213 >        return (int) (s & ushortMask);
214      }
215  
216      private static int partiesOf(long s) {
217 <        return ((int)s) >>> 16;
217 >        return ((int) s) >>> 16;
218      }
219  
220      private static int phaseOf(long s) {
221 <        return (int)(s >>> 32);
221 >        return (int) (s >>> 32);
222      }
223  
224      private static int arrivedOf(long s) {
# Line 228 | Line 226 | public class Phaser {
226      }
227  
228      private static long stateFor(int phase, int parties, int unarrived) {
229 <        return ((((long)phase) << 32) | (((long)parties) << 16) |
230 <                (long)unarrived);
229 >        return ((((long) phase) << 32) | (((long) parties) << 16) |
230 >                (long) unarrived);
231      }
232  
233      private static long trippedStateFor(int phase, int parties) {
234 <        long lp = (long)parties;
235 <        return (((long)phase) << 32) | (lp << 16) | lp;
234 >        long lp = (long) parties;
235 >        return (((long) phase) << 32) | (lp << 16) | lp;
236      }
237  
238      /**
239 <     * Returns message string for bad bounds exceptions
239 >     * Returns message string for bad bounds exceptions.
240       */
241      private static String badBounds(int parties, int unarrived) {
242          return ("Attempt to set " + unarrived +
# Line 267 | Line 265 | public class Phaser {
265      private final AtomicReference<QNode> oddQ  = new AtomicReference<QNode>();
266  
267      private AtomicReference<QNode> queueFor(int phase) {
268 <        return (phase & 1) == 0? evenQ : oddQ;
268 >        return ((phase & 1) == 0) ? evenQ : oddQ;
269      }
270  
271      /**
# Line 275 | Line 273 | public class Phaser {
273       * root if necessary.
274       */
275      private long getReconciledState() {
276 <        return parent == null? state : reconcileState();
276 >        return (parent == null) ? state : reconcileState();
277      }
278  
279      /**
# Line 313 | Line 311 | public class Phaser {
311      /**
312       * Creates a new Phaser with the given numbers of registered
313       * unarrived parties, initial phase number 0, and no parent.
314 <     * @param parties the number of parties required to trip barrier.
314 >     *
315 >     * @param parties the number of parties required to trip barrier
316       * @throws IllegalArgumentException if parties less than zero
317 <     * or greater than the maximum number of parties supported.
317 >     * or greater than the maximum number of parties supported
318       */
319      public Phaser(int parties) {
320          this(null, parties);
# Line 326 | Line 325 | public class Phaser {
325       * initially registered parties. If parent is non-null this phaser
326       * is registered with the parent and its initial phase number is
327       * the same as that of parent phaser.
328 <     * @param parent the parent phaser.
328 >     *
329 >     * @param parent the parent phaser
330       */
331      public Phaser(Phaser parent) {
332          int phase = 0;
# Line 342 | Line 342 | public class Phaser {
342  
343      /**
344       * Creates a new Phaser with the given parent and numbers of
345 <     * registered unarrived parties. If parent is non-null this phaser
345 >     * registered unarrived parties. If parent is non-null, this phaser
346       * is registered with the parent and its initial phase number is
347       * the same as that of parent phaser.
348 <     * @param parent the parent phaser.
349 <     * @param parties the number of parties required to trip barrier.
348 >     *
349 >     * @param parent the parent phaser
350 >     * @param parties the number of parties required to trip barrier
351       * @throws IllegalArgumentException if parties less than zero
352 <     * or greater than the maximum number of parties supported.
352 >     * or greater than the maximum number of parties supported
353       */
354      public Phaser(Phaser parent, int parties) {
355          if (parties < 0 || parties > ushortMask)
# Line 366 | Line 367 | public class Phaser {
367  
368      /**
369       * Adds a new unarrived party to this phaser.
370 +     *
371       * @return the current barrier phase number upon registration
372       * @throws IllegalStateException if attempting to register more
373 <     * than the maximum supported number of parties.
373 >     * than the maximum supported number of parties
374       */
375      public int register() {
376          return doRegister(1);
# Line 376 | Line 378 | public class Phaser {
378  
379      /**
380       * Adds the given number of new unarrived parties to this phaser.
381 <     * @param parties the number of parties required to trip barrier.
381 >     *
382 >     * @param parties the number of parties required to trip barrier
383       * @return the current barrier phase number upon registration
384       * @throws IllegalStateException if attempting to register more
385 <     * than the maximum supported number of parties.
385 >     * than the maximum supported number of parties
386       */
387      public int bulkRegister(int parties) {
388          if (parties < 0)
# Line 399 | Line 402 | public class Phaser {
402              phase = phaseOf(s);
403              int unarrived = unarrivedOf(s) + registrations;
404              int parties = partiesOf(s) + registrations;
405 <            if (phase < 0)
405 >            if (phase < 0)
406                  break;
407              if (parties > ushortMask || unarrived > ushortMask)
408                  throw new IllegalStateException(badBounds(parties, unarrived));
# Line 415 | Line 418 | public class Phaser {
418       * in turn wait for others via {@link #awaitAdvance}).
419       *
420       * @return the barrier phase number upon entry to this method, or a
421 <     * negative value if terminated;
421 >     * negative value if terminated
422       * @throws IllegalStateException if not terminated and the number
423 <     * of unarrived parties would become negative.
423 >     * of unarrived parties would become negative
424       */
425      public int arrive() {
426          int phase;
# Line 437 | Line 440 | public class Phaser {
440                  if (par == null) {      // directly trip
441                      if (casState
442                          (s,
443 <                         trippedStateFor(onAdvance(phase, parties)? -1 :
443 >                         trippedStateFor(onAdvance(phase, parties) ? -1 :
444                                           ((phase + 1) & phaseMask), parties))) {
445                          releaseWaiters(phase);
446                          break;
# Line 467 | Line 470 | public class Phaser {
470       * zero parties, this phaser is also deregistered from its parent.
471       *
472       * @return the current barrier phase number upon entry to
473 <     * this method, or a negative value if terminated;
473 >     * this method, or a negative value if terminated
474       * @throws IllegalStateException if not terminated and the number
475 <     * of registered or unarrived parties would become negative.
475 >     * of registered or unarrived parties would become negative
476       */
477      public int arriveAndDeregister() {
478          // similar code to arrive, but too different to merge
# Line 498 | Line 501 | public class Phaser {
501                  if (unarrived == 0) {
502                      if (casState
503                          (s,
504 <                         trippedStateFor(onAdvance(phase, parties)? -1 :
504 >                         trippedStateFor(onAdvance(phase, parties) ? -1 :
505                                           ((phase + 1) & phaseMask), parties))) {
506                          releaseWaiters(phase);
507                          break;
# Line 520 | Line 523 | public class Phaser {
523       * to {@code awaitAdvance(arrive())}.  If you instead need to
524       * await with interruption of timeout, and/or deregister upon
525       * arrival, you can arrange them using analogous constructions.
526 +     *
527       * @return the phase on entry to this method
528       * @throws IllegalStateException if not terminated and the number
529 <     * of unarrived parties would become negative.
529 >     * of unarrived parties would become negative
530       */
531      public int arriveAndAwaitAdvance() {
532          return awaitAdvance(arrive());
# Line 532 | Line 536 | public class Phaser {
536       * Awaits the phase of the barrier to advance from the given
537       * value, or returns immediately if argument is negative or this
538       * barrier is terminated.
539 +     *
540       * @param phase the phase on entry to this method
541       * @return the phase on exit from this method
542       */
# Line 553 | Line 558 | public class Phaser {
558       * value, or returns immediately if argument is negative or this
559       * barrier is terminated, or throws InterruptedException if
560       * interrupted while waiting.
561 +     *
562       * @param phase the phase on entry to this method
563       * @return the phase on exit from this method
564       * @throws InterruptedException if thread interrupted while waiting
565       */
566 <    public int awaitAdvanceInterruptibly(int phase)
566 >    public int awaitAdvanceInterruptibly(int phase)
567          throws InterruptedException {
568          if (phase < 0)
569              return phase;
# Line 574 | Line 580 | public class Phaser {
580       * Awaits the phase of the barrier to advance from the given value
581       * or the given timeout elapses, or returns immediately if
582       * argument is negative or this barrier is terminated.
583 +     *
584       * @param phase the phase on entry to this method
585       * @return the phase on exit from this method
586       * @throws InterruptedException if thread interrupted while waiting
587       * @throws TimeoutException if timed out while waiting
588       */
589 <    public int awaitAdvanceInterruptibly(int phase, long timeout, TimeUnit unit)
589 >    public int awaitAdvanceInterruptibly(int phase,
590 >                                         long timeout, TimeUnit unit)
591          throws InterruptedException, TimeoutException {
592          if (phase < 0)
593              return phase;
# Line 620 | Line 628 | public class Phaser {
628       * Returns the current phase number. The maximum phase number is
629       * {@code Integer.MAX_VALUE}, after which it restarts at
630       * zero. Upon termination, the phase number is negative.
631 +     *
632       * @return the phase number, or a negative value if terminated
633       */
634      public final int getPhase() {
# Line 628 | Line 637 | public class Phaser {
637  
638      /**
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       */
# Line 637 | Line 647 | public class Phaser {
647  
648      /**
649       * Returns the number of parties registered at this barrier.
650 +     *
651       * @return the number of parties
652       */
653      public int getRegisteredParties() {
# Line 646 | Line 657 | public class Phaser {
657      /**
658       * Returns the number of parties that have arrived at the current
659       * phase of this barrier.
660 +     *
661       * @return the number of arrived parties
662       */
663      public int getArrivedParties() {
# Line 655 | Line 667 | public class Phaser {
667      /**
668       * Returns the number of registered parties that have not yet
669       * arrived at the current phase of this barrier.
670 +     *
671       * @return the number of unarrived parties
672       */
673      public int getUnarrivedParties() {
# Line 662 | Line 675 | public class Phaser {
675      }
676  
677      /**
678 <     * Returns the parent of this phaser, or null if none.
679 <     * @return the parent of this phaser, or null if none
678 >     * Returns the parent of this phaser, or {@code null} if none.
679 >     *
680 >     * @return the parent of this phaser, or {@code null} if none
681       */
682      public Phaser getParent() {
683          return parent;
# Line 672 | Line 686 | public class Phaser {
686      /**
687       * Returns the root ancestor of this phaser, which is the same as
688       * this phaser if it has no parent.
689 +     *
690       * @return the root ancestor of this phaser
691       */
692      public Phaser getRoot() {
# Line 680 | Line 695 | public class Phaser {
695  
696      /**
697       * Returns {@code true} if this barrier has been terminated.
698 +     *
699       * @return {@code true} if this barrier has been terminated
700       */
701      public boolean isTerminated() {
# Line 690 | Line 706 | public class Phaser {
706       * Overridable method to perform an action upon phase advance, and
707       * to control termination. This method is invoked whenever the
708       * barrier is tripped (and thus all other waiting parties are
709 <     * dormant). If it returns true, then, rather than advance the
710 <     * phase number, this barrier will be set to a final termination
711 <     * state, and subsequent calls to {@code isTerminated} will
712 <     * return true.
709 >     * dormant). If it returns {@code true}, then, rather than advance
710 >     * the phase number, this barrier will be set to a final
711 >     * termination state, and subsequent calls to {@link #isTerminated}
712 >     * will return true.
713       *
714 <     * <p> The default version returns true when the number of
714 >     * <p> The default version returns {@code true} when the number of
715       * registered parties is zero. Normally, overrides that arrange
716       * termination for other reasons should also preserve this
717       * property.
# Line 795 | Line 811 | public class Phaser {
811                  try {
812                      ForkJoinPool.managedBlock(this, false);
813                  } catch (InterruptedException ie) {
814 <                }
814 >                }
815              }
816              return wasInterrupted;
817          }
# Line 803 | Line 819 | public class Phaser {
819      }
820  
821      /**
822 <     * Removes and signals waiting threads from wait queue
822 >     * Removes and signals waiting threads from wait queue.
823       */
824      private void releaseWaiters(int phase) {
825          AtomicReference<QNode> head = queueFor(phase);
# Line 815 | Line 831 | public class Phaser {
831      }
832  
833      /**
834 <     * Tries to enqueue given node in the appropriate wait queue
834 >     * Tries to enqueue given node in the appropriate wait queue.
835 >     *
836       * @return true if successful
837       */
838      private boolean tryEnqueue(QNode node) {
# Line 825 | Line 842 | public class Phaser {
842  
843      /**
844       * Enqueues node and waits unless aborted or signalled.
845 +     *
846       * @return current phase
847       */
848      private int untimedWait(int phase) {
# Line 912 | Line 930 | public class Phaser {
930          return p;
931      }
932  
933 <    // Temporary Unsafe mechanics for preliminary release
916 <    private static Unsafe getUnsafe() throws Throwable {
917 <        try {
918 <            return Unsafe.getUnsafe();
919 <        } catch (SecurityException se) {
920 <            try {
921 <                return java.security.AccessController.doPrivileged
922 <                    (new java.security.PrivilegedExceptionAction<Unsafe>() {
923 <                        public Unsafe run() throws Exception {
924 <                            return getUnsafePrivileged();
925 <                        }});
926 <            } catch (java.security.PrivilegedActionException e) {
927 <                throw e.getCause();
928 <            }
929 <        }
930 <    }
933 >    // Unsafe mechanics
934  
935 <    private static Unsafe getUnsafePrivileged()
936 <            throws NoSuchFieldException, IllegalAccessException {
937 <        Field f = Unsafe.class.getDeclaredField("theUnsafe");
935 <        f.setAccessible(true);
936 <        return (Unsafe)f.get(null);
937 <    }
935 >    private static final sun.misc.Unsafe UNSAFE = getUnsafe();
936 >    private static final long stateOffset =
937 >        objectFieldOffset("state", Phaser.class);
938  
939 <    private static long fieldOffset(String fieldName)
940 <            throws NoSuchFieldException {
941 <        return _unsafe.objectFieldOffset
942 <            (Phaser.class.getDeclaredField(fieldName));
939 >    private final boolean casState(long cmp, long val) {
940 >        return UNSAFE.compareAndSwapLong(this, stateOffset, cmp, val);
941      }
942  
943 <    static final Unsafe _unsafe;
946 <    static final long stateOffset;
947 <
948 <    static {
943 >    private static long objectFieldOffset(String field, Class<?> klazz) {
944          try {
945 <            _unsafe = getUnsafe();
946 <            stateOffset = fieldOffset("state");
947 <        } catch (Exception e) {
948 <            throw new RuntimeException("Could not initialize intrinsics", e);
945 >            return UNSAFE.objectFieldOffset(klazz.getDeclaredField(field));
946 >        } catch (NoSuchFieldException e) {
947 >            // Convert Exception to corresponding Error
948 >            NoSuchFieldError error = new NoSuchFieldError(field);
949 >            error.initCause(e);
950 >            throw error;
951          }
952      }
953  
954 <    final boolean casState(long cmp, long val) {
955 <        return _unsafe.compareAndSwapLong(this, stateOffset, cmp, val);
954 >    /**
955 >     * Returns a sun.misc.Unsafe.  Suitable for use in a 3rd party package.
956 >     * Replace with a simple call to Unsafe.getUnsafe when integrating
957 >     * into a jdk.
958 >     *
959 >     * @return a sun.misc.Unsafe
960 >     */
961 >    private static sun.misc.Unsafe getUnsafe() {
962 >        try {
963 >            return sun.misc.Unsafe.getUnsafe();
964 >        } catch (SecurityException se) {
965 >            try {
966 >                return java.security.AccessController.doPrivileged
967 >                    (new java.security
968 >                     .PrivilegedExceptionAction<sun.misc.Unsafe>() {
969 >                        public sun.misc.Unsafe run() throws Exception {
970 >                            java.lang.reflect.Field f = sun.misc
971 >                                .Unsafe.class.getDeclaredField("theUnsafe");
972 >                            f.setAccessible(true);
973 >                            return (sun.misc.Unsafe) f.get(null);
974 >                        }});
975 >            } catch (java.security.PrivilegedActionException e) {
976 >                throw new RuntimeException("Could not initialize intrinsics",
977 >                                           e.getCause());
978 >            }
979 >        }
980      }
981   }

Diff Legend

– Removed lines
+ Added lines
< Changed lines
> Changed lines