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.59 by dl, Sat Nov 27 16:46:53 2010 UTC vs.
Revision 1.60 by dl, Sun Nov 28 15:49:49 2010 UTC

# Line 18 | Line 18 | import java.util.concurrent.locks.LockSu
18   * but supporting more flexible usage.
19   *
20   * <p> <b>Registration.</b> Unlike the case for other barriers, the
21 < * number of parties <em>registered</em> to synchronize on a phaser
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
# Line 76 | Line 76 | import java.util.concurrent.locks.LockSu
76   *
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
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}. The default implementation returns
83   * {@code true} if a deregistration has caused the number of
84   * registered parties to become zero.  As illustrated below, when
85 < * phasers control actions with a fixed number of iterations, it is
85 > * Phasers control actions with a fixed number of iterations, it is
86   * often convenient to override this method to cause termination when
87   * the current phase number reaches a threshold. Method {@link
88   * #forceTermination} is also available to abruptly release waiting
# Line 97 | Line 97 | import java.util.concurrent.locks.LockSu
97   * overhead.
98   *
99   * <p><b>Monitoring.</b> While synchronization methods may be invoked
100 < * only by registered parties, the current state of a phaser may be
100 > * only by registered parties, the current state of a Phaser may be
101   * monitored by any caller.  At any given moment there are {@link
102   * #getRegisteredParties} parties in total, of which {@link
103   * #getArrivedParties} have arrived at the current phase ({@link
# Line 183 | Line 183 | import java.util.concurrent.locks.LockSu
183   * }}</pre>
184   *
185   *
186 < * <p>To create a set of tasks using a tree of phasers,
186 > * <p>To create a set of tasks using a tree of Phasers,
187   * you could use code of the following form, assuming a
188 < * Task class with a constructor accepting a phaser that
188 > * Task class with a constructor accepting a Phaser that
189   * it registers with upon construction:
190   *
191   *  <pre> {@code
# Line 212 | Line 212 | import java.util.concurrent.locks.LockSu
212   * <p><b>Implementation notes</b>: This implementation restricts the
213   * maximum number of parties to 65535. Attempts to register additional
214   * parties result in {@code IllegalStateException}. However, you can and
215 < * should create tiered phasers to accommodate arbitrarily large sets
215 > * should create tiered Phasers to accommodate arbitrarily large sets
216   * of participants.
217   *
218   * @since 1.7
# Line 383 | Line 383 | public class Phaser {
383                      return phase;
384              }
385              else if (parties != 0)               // wait for onAdvance
386 <                internalAwaitAdvance(phase, null);
386 >                root.internalAwaitAdvance(phase, null);
387              else {                               // 1st registration of child
388                  synchronized(this) {             // register parent first
389                      if (reconcileState() == s) { // recheck under lock
# Line 428 | Line 428 | public class Phaser {
428      }
429  
430      /**
431 <     * Creates a new phaser without any initially registered parties,
431 >     * Creates a new Phaser without any initially registered parties,
432       * initial phase number 0, and no parent. Any thread using this
433 <     * phaser will need to first register for it.
433 >     * Phaser will need to first register for it.
434       */
435      public Phaser() {
436          this(null, 0);
437      }
438  
439      /**
440 <     * Creates a new phaser with the given number of registered
440 >     * Creates a new Phaser with the given number of registered
441       * unarrived parties, initial phase number 0, and no parent.
442       *
443       * @param parties the number of parties required to trip barrier
# Line 449 | Line 449 | public class Phaser {
449      }
450  
451      /**
452     * Creates a new phaser with the given parent, and without any
453     * initially registered parties.  Any thread using this phaser
454     * will need to first register for it, at which point, if the
455     * given parent is non-null, this phaser will also be registered
456     * with the parent.
457     *
452       * Equivalent to {@link #Phaser(Phaser, int) Phaser(parent, 0)}.
453       *
454 <     * @param parent the parent phaser
454 >     * @param parent the parent Phaser
455       */
456      public Phaser(Phaser parent) {
457          this(parent, 0);
458      }
459  
460      /**
461 <     * Creates a new phaser with the given parent and number of
462 <     * registered unarrived parties. If parent is non-null and
463 <     * the number of parties is non-zero, this phaser is registered
464 <     * with the parent.
461 >     * Creates a new Phaser with the given parent and number of
462 >     * registered unarrived parties. Registration and deregistration
463 >     * of this child Phaser with its parent are managed automatically.
464 >     * If the given parent is non-null, whenever this child Phaser has
465 >     * any registered parties (as established in this constructor,
466 >     * {@link #register}, or {@link #bulkRegister}), this child Phaser
467 >     * is registered with its parent. Whenever the number of
468 >     * registered parties becomes zero as the result of an invocation
469 >     * of {@link #arriveAndDeregister}, this child Phaser is
470 >     * deregistered from its parent.
471       *
472 <     * @param parent the parent phaser
472 >     * @param parent the parent Phaser
473       * @param parties the number of parties required to trip barrier
474       * @throws IllegalArgumentException if parties less than zero
475       * or greater than the maximum number of parties supported
# Line 477 | Line 477 | public class Phaser {
477      public Phaser(Phaser parent, int parties) {
478          if (parties >>> PARTIES_SHIFT != 0)
479              throw new IllegalArgumentException("Illegal number of parties");
480 <        int phase;
480 >        long s = ((long) parties) | (((long) parties) << PARTIES_SHIFT);
481          this.parent = parent;
482          if (parent != null) {
483              Phaser r = parent.root;
484              this.root = r;
485              this.evenQ = r.evenQ;
486              this.oddQ = r.oddQ;
487 <            phase = (parties == 0) ? parent.getPhase() : parent.doRegister(1);
487 >            if (parties != 0)
488 >                s |= ((long)(parent.doRegister(1))) << PHASE_SHIFT;
489          }
490          else {
491              this.root = this;
492              this.evenQ = new AtomicReference<QNode>();
493              this.oddQ = new AtomicReference<QNode>();
493            phase = 0;
494          }
495 <        long p = (long)parties;
496 <        this.state = (((long)phase) << PHASE_SHIFT) | p | (p << PARTIES_SHIFT);
495 >        this.state = s;
496      }
497  
498      /**
499 <     * Adds a new unarrived party to this phaser.  If an ongoing
499 >     * Adds a new unarrived party to this Phaser.  If an ongoing
500       * invocation of {@link #onAdvance} is in progress, this method
501 <     * may wait until its completion before registering.  If this
502 <     * phaser has a parent, and this phaser previously had no
503 <     * registered parties, this phaser is also registered with its
505 <     * parent.
501 >     * may await its completion before returning.  If this Phaser has
502 >     * a parent, and this Phaser previously had no registered parties,
503 >     * this Phaser is also registered with its parent.
504       *
505       * @return the arrival phase number to which this registration applied
506       * @throws IllegalStateException if attempting to register more
# Line 513 | Line 511 | public class Phaser {
511      }
512  
513      /**
514 <     * Adds the given number of new unarrived parties to this phaser.
514 >     * Adds the given number of new unarrived parties to this Phaser.
515       * If an ongoing invocation of {@link #onAdvance} is in progress,
516 <     * this method may wait until its completion before registering.
517 <     * If this phaser has a parent, and the given number of parities
518 <     * is greater than zero, and this phaser previously had no
519 <     * registered parties, this phaser is also registered with its
522 <     * parent.
516 >     * this method may await its completion before returning.  If this
517 >     * Phaser has a parent, and the given number of parities is
518 >     * greater than zero, and this Phaser previously had no registered
519 >     * parties, this Phaser is also registered with its parent.
520       *
521       * @param parties the number of additional parties required to trip barrier
522       * @return the arrival phase number to which this registration applied
# Line 530 | Line 527 | public class Phaser {
527      public int bulkRegister(int parties) {
528          if (parties < 0)
529              throw new IllegalArgumentException();
530 <        else if (parties == 0)
530 >        if (parties == 0)
531              return getPhase();
532          return doRegister(parties);
533      }
534  
535      /**
536 <     * Arrives at the barrier, but does not wait for others.  (You can
537 <     * in turn wait for others via {@link #awaitAdvance}).  It is a
538 <     * usage error for an unregistered party to invoke this
539 <     * method. However, it is possible that this error will result in
540 <     * an {code IllegalStateException} only when some <em>other</em>
541 <     * party arrives.
536 >     * Arrives at the barrier, without waiting for others to arrive.
537 >     *
538 >     * <p>It is a usage error for an unregistered party to invoke this
539 >     * method.  However, this error may result in an {@code
540 >     * IllegalStateException} only upon some subsequent operation on
541 >     * this Phaser, if ever.
542       *
543       * @return the arrival phase number, or a negative value if terminated
544       * @throws IllegalStateException if not terminated and the number
# Line 553 | Line 550 | public class Phaser {
550  
551      /**
552       * Arrives at the barrier and deregisters from it without waiting
553 <     * for others. Deregistration reduces the number of parties
554 <     * required to trip the barrier in future phases.  If this phaser
555 <     * has a parent, and deregistration causes this phaser to have
556 <     * zero parties, this phaser also arrives at and is deregistered
557 <     * from its parent.  It is a usage error for an unregistered party
558 <     * to invoke this method. However, it is possible that this error
559 <     * will result in an {code IllegalStateException} only when some
560 <     * <em>other</em> party arrives.
553 >     * for others to arrive. Deregistration reduces the number of
554 >     * parties required to trip the barrier in future phases.  If this
555 >     * Phaser has a parent, and deregistration causes this Phaser to
556 >     * have zero parties, this Phaser is also deregistered from its
557 >     * parent.
558 >     *
559 >     * <p>It is a usage error for an unregistered party to invoke this
560 >     * method.  However, this error may result in an {@code
561 >     * IllegalStateException} only upon some subsequent operation on
562 >     * this Phaser, if ever.
563       *
564       * @return the arrival phase number, or a negative value if terminated
565       * @throws IllegalStateException if not terminated and the number
# Line 576 | Line 575 | public class Phaser {
575       * interruption or timeout, you can arrange this with an analogous
576       * construction using one of the other forms of the {@code
577       * awaitAdvance} method.  If instead you need to deregister upon
578 <     * arrival, use {@link #arriveAndDeregister}.  It is a usage error
579 <     * for an unregistered party to invoke this method. However, it is
580 <     * possible that this error will result in an {code
581 <     * IllegalStateException} only when some <em>other</em> party
582 <     * arrives.
578 >     * arrival, use {@code awaitAdvance(arriveAndDeregister())}.
579 >     *
580 >     * <p>It is a usage error for an unregistered party to invoke this
581 >     * method.  However, this error may result in an {@code
582 >     * IllegalStateException} only upon some subsequent operation on
583 >     * this Phaser, if ever.
584       *
585       * @return the arrival phase number, or a negative number if terminated
586       * @throws IllegalStateException if not terminated and the number
# Line 603 | Line 603 | public class Phaser {
603       * if terminated or argument is negative
604       */
605      public int awaitAdvance(int phase) {
606 +        Phaser r;
607 +        int p = (int)(state >>> PHASE_SHIFT);
608          if (phase < 0)
609              return phase;
610 <        long s = (parent == null) ? state : reconcileState();
611 <        int p = (int)(s >>> PHASE_SHIFT);
612 <        return (p != phase) ? p : internalAwaitAdvance(phase, null);
610 >        if (p == phase &&
611 >            (p = (int)((r = root).state >>> PHASE_SHIFT)) == phase)
612 >            return r.internalAwaitAdvance(phase, null);
613 >        return p;
614      }
615  
616      /**
# Line 626 | Line 629 | public class Phaser {
629       */
630      public int awaitAdvanceInterruptibly(int phase)
631          throws InterruptedException {
632 +        Phaser r;
633 +        int p = (int)(state >>> PHASE_SHIFT);
634          if (phase < 0)
635              return phase;
636 <        long s = (parent == null) ? state : reconcileState();
637 <        int p = (int)(s >>> PHASE_SHIFT);
633 <        if (p == phase) {
636 >        if (p == phase &&
637 >            (p = (int)((r = root).state >>> PHASE_SHIFT)) == phase) {
638              QNode node = new QNode(this, phase, true, false, 0L);
639 <            p = internalAwaitAdvance(phase, node);
639 >            p = r.internalAwaitAdvance(phase, node);
640              if (node.wasInterrupted)
641                  throw new InterruptedException();
642          }
# Line 662 | Line 666 | public class Phaser {
666      public int awaitAdvanceInterruptibly(int phase,
667                                           long timeout, TimeUnit unit)
668          throws InterruptedException, TimeoutException {
669 +        long nanos = unit.toNanos(timeout);
670 +        Phaser r;
671 +        int p = (int)(state >>> PHASE_SHIFT);
672          if (phase < 0)
673              return phase;
674 <        long s = (parent == null) ? state : reconcileState();
675 <        int p = (int)(s >>> PHASE_SHIFT);
669 <        if (p == phase) {
670 <            long nanos = unit.toNanos(timeout);
674 >        if (p == phase &&
675 >            (p = (int)((r = root).state >>> PHASE_SHIFT)) == phase) {
676              QNode node = new QNode(this, phase, true, true, nanos);
677 <            p = internalAwaitAdvance(phase, node);
677 >            p = r.internalAwaitAdvance(phase, node);
678              if (node.wasInterrupted)
679                  throw new InterruptedException();
680              else if (p == phase)
# Line 680 | Line 685 | public class Phaser {
685  
686      /**
687       * Forces this barrier to enter termination state.  Counts of
688 <     * arrived and registered parties are unaffected.  If this phaser
689 <     * is a member of a tiered set of phasers, then all of the phasers
690 <     * in the set are terminated.  If this phaser is already
688 >     * arrived and registered parties are unaffected.  If this Phaser
689 >     * is a member of a tiered set of Phasers, then all of the Phasers
690 >     * in the set are terminated.  If this Phaser is already
691       * terminated, this method has no effect.  This method may be
692       * useful for coordinating recovery after one or more tasks
693       * encounter unexpected exceptions.
# Line 728 | Line 733 | public class Phaser {
733       * @return the number of arrived parties
734       */
735      public int getArrivedParties() {
736 <        return arrivedOf(parent==null? state : reconcileState());
736 >        long s = state;
737 >        int u = unarrivedOf(s); // only reconcile if possibly needed
738 >        return (u != 0 || parent == null) ?
739 >            partiesOf(s) - u :
740 >            arrivedOf(reconcileState());
741      }
742  
743      /**
# Line 738 | Line 747 | public class Phaser {
747       * @return the number of unarrived parties
748       */
749      public int getUnarrivedParties() {
750 <        return unarrivedOf(parent==null? state : reconcileState());
750 >        int u = unarrivedOf(state);
751 >        return (u != 0 || parent == null) ? u : unarrivedOf(reconcileState());
752      }
753  
754      /**
755 <     * Returns the parent of this phaser, or {@code null} if none.
755 >     * Returns the parent of this Phaser, or {@code null} if none.
756       *
757 <     * @return the parent of this phaser, or {@code null} if none
757 >     * @return the parent of this Phaser, or {@code null} if none
758       */
759      public Phaser getParent() {
760          return parent;
761      }
762  
763      /**
764 <     * Returns the root ancestor of this phaser, which is the same as
765 <     * this phaser if it has no parent.
764 >     * Returns the root ancestor of this Phaser, which is the same as
765 >     * this Phaser if it has no parent.
766       *
767 <     * @return the root ancestor of this phaser
767 >     * @return the root ancestor of this Phaser
768       */
769      public Phaser getRoot() {
770          return root;
# Line 781 | Line 791 | public class Phaser {
791       * propagated to the party attempting to trip the barrier, in
792       * which case no advance occurs.
793       *
794 <     * <p>The arguments to this method provide the state of the phaser
794 >     * <p>The arguments to this method provide the state of the Phaser
795       * prevailing for the current transition.  The effects of invoking
796       * arrival, registration, and waiting methods on this Phaser from
797       * within {@code onAdvance} are unspecified and should not be
# Line 813 | Line 823 | public class Phaser {
823      }
824  
825      /**
826 <     * Returns a string identifying this phaser, as well as its
826 >     * Returns a string identifying this Phaser, as well as its
827       * state.  The state, in brackets, includes the String {@code
828       * "phase = "} followed by the phase number, {@code "parties = "}
829       * followed by the number of registered parties, and {@code
# Line 872 | Line 882 | public class Phaser {
882  
883      /**
884       * Possibly blocks and waits for phase to advance unless aborted.
885 +     * Call only from root node.
886       *
887       * @param phase current phase
888       * @param node if non-null, the wait node to track interrupt and timeout;
# Line 879 | Line 890 | public class Phaser {
890       * @return current phase
891       */
892      private int internalAwaitAdvance(int phase, QNode node) {
882        Phaser current = this;       // to eventually wait at root if tiered
893          boolean queued = false;      // true when node is enqueued
894          int lastUnarrived = -1;      // to increase spins upon change
895          int spins = SPINS_PER_ARRIVAL;
896          long s;
897          int p;
898 <        while ((p = (int)((s = current.state) >>> PHASE_SHIFT)) == phase) {
889 <            Phaser par;
898 >        while ((p = (int)((s = state) >>> PHASE_SHIFT)) == phase) {
899              int unarrived = (int)s & UNARRIVED_MASK;
900              if (unarrived != lastUnarrived) {
901                  if (lastUnarrived == -1) // ensure old queue clean
# Line 894 | Line 903 | public class Phaser {
903                  if ((lastUnarrived = unarrived) < NCPU)
904                      spins += SPINS_PER_ARRIVAL;
905              }
897            else if (unarrived == 0 && (par = current.parent) != null) {
898                current = par;       // if all arrived, use parent
899                par = par.parent;
900                lastUnarrived = -1;
901            }
906              else if (spins > 0) {
907                  if (--spins == (SPINS_PER_ARRIVAL >>> 1))
908                      Thread.yield();  // yield midway through spin
# Line 906 | Line 910 | public class Phaser {
910              else if (node == null)   // must be noninterruptible
911                  node = new QNode(this, phase, false, false, 0L);
912              else if (node.isReleasable()) {
913 <                if ((p = (int)(root.state >>> PHASE_SHIFT)) != phase)
914 <                    break;
911 <                else
912 <                    return phase;    // aborted
913 >                p = (int)(state >>> PHASE_SHIFT);
914 >                break;               // aborted
915              }
916              else if (!queued) {      // push onto queue
917                  AtomicReference<QNode> head = queueFor(phase);
918                  QNode q = head.get();
919                  if (q == null || q.phase == phase) {
920                      node.next = q;
921 <                    if ((p = (int)(root.state >>> PHASE_SHIFT)) != phase)
921 >                    if ((p = (int)(state >>> PHASE_SHIFT)) != phase)
922                          break;       // recheck to avoid stale enqueue
923                      else
924                          queued = head.compareAndSet(q, node);
# Line 930 | Line 932 | public class Phaser {
932                  }
933              }
934          }
935 <        releaseWaiters(phase);
936 <        if (node != null)
937 <            node.onRelease();
935 >
936 >        if (node != null) {
937 >            if (node.thread != null)
938 >                node.thread = null; // disable unpark() in node.signal
939 >            if (!node.interruptible && node.wasInterrupted)
940 >                Thread.currentThread().interrupt();
941 >        }
942 >        if (p != phase)
943 >            releaseWaiters(phase);
944          return p;
945      }
946  
# Line 1005 | Line 1013 | public class Phaser {
1013                  LockSupport.unpark(t);
1014              }
1015          }
1008
1009        void onRelease() { // actions upon return from internalAwaitAdvance
1010            if (!interruptible && wasInterrupted)
1011                Thread.currentThread().interrupt();
1012            if (thread != null)
1013                thread = null;
1014        }
1015
1016      }
1017  
1018      // Unsafe mechanics

Diff Legend

Removed lines
+ Added lines
< Changed lines
> Changed lines