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.48 by dl, Sun Oct 24 21:45:39 2010 UTC vs.
Revision 1.49 by dl, Fri Nov 5 23:01:47 2010 UTC

# Line 7 | Line 7
7   package jsr166y;
8  
9   import java.util.concurrent.*;
10 –
10   import java.util.concurrent.atomic.AtomicReference;
11   import java.util.concurrent.locks.LockSupport;
12  
# Line 295 | Line 294 | public class Phaser {
294  
295      /**
296       * Heads of Treiber stacks for waiting threads. To eliminate
297 <     * contention while releasing some threads while adding others, we
297 >     * contention when releasing some threads while adding others, we
298       * use two of them, alternating across even and odd phases.
299 +     * Subphasers share queues with root to speed up releases.
300       */
301      private final AtomicReference<QNode> evenQ = new AtomicReference<QNode>();
302      private final AtomicReference<QNode> oddQ  = new AtomicReference<QNode>();
303  
304      private AtomicReference<QNode> queueFor(int phase) {
305 <        return ((phase & 1) == 0) ? evenQ : oddQ;
305 >        Phaser r = root;
306 >        return ((phase & 1) == 0) ? r.evenQ : r.oddQ;
307      }
308  
309      /**
# Line 317 | Line 318 | public class Phaser {
318       * Recursively resolves state.
319       */
320      private long reconcileState() {
321 <        Phaser p = parent;
321 >        Phaser par = parent;
322          long s = state;
323 <        if (p != null) {
324 <            while (unarrivedOf(s) == 0 && phaseOf(s) != phaseOf(root.state)) {
325 <                long parentState = p.getReconciledState();
323 >        if (par != null) {
324 >            int phase, rootPhase;
325 >            while ((phase = phaseOf(s)) >= 0 &&
326 >                   (rootPhase = phaseOf(root.state)) != phase &&
327 >                   (rootPhase < 0 || unarrivedOf(s) == 0)) {
328 >                long parentState = par.getReconciledState();
329                  int parentPhase = phaseOf(parentState);
330 <                int phase = phaseOf(s = state);
331 <                if (phase != parentPhase) {
332 <                    long next = trippedStateFor(parentPhase, partiesOf(s));
333 <                    if (casState(s, next)) {
334 <                        releaseWaiters(phase);
335 <                        s = next;
336 <                    }
330 >                int parties = partiesOf(s);
331 >                long next = trippedStateFor(parentPhase, parties);
332 >                if (phaseOf(root.state) == rootPhase &&
333 >                    parentPhase != phase &&
334 >                    state == s && casState(s, next)) {
335 >                    releaseWaiters(phase);
336 >                    if (parties == 0) // exit if the final deregistration
337 >                        break;
338                  }
339 +                s = state;
340              }
341          }
342          return s;
# Line 404 | Line 410 | public class Phaser {
410  
411      /**
412       * Adds a new unarrived party to this phaser.
413 +     * If an ongoing invocation of {@link #onAdvance} is in progress,
414 +     * this method waits until its completion before registering.
415       *
416       * @return the arrival phase number to which this registration applied
417       * @throws IllegalStateException if attempting to register more
# Line 415 | Line 423 | public class Phaser {
423  
424      /**
425       * Adds the given number of new unarrived parties to this phaser.
426 +     * If an ongoing invocation of {@link #onAdvance} is in progress,
427 +     * this method waits until its completion before registering.
428       *
429       * @param parties the number of additional parties required to trip barrier
430       * @return the arrival phase number to which this registration applied
# Line 434 | Line 444 | public class Phaser {
444       * Shared code for register, bulkRegister
445       */
446      private int doRegister(int registrations) {
447 +        Phaser par = parent;
448 +        long s;
449          int phase;
450 <        for (;;) {
451 <            long s = getReconciledState();
452 <            phase = phaseOf(s);
453 <            int unarrived = unarrivedOf(s) + registrations;
454 <            int parties = partiesOf(s) + registrations;
455 <            if (phase < 0)
456 <                break;
457 <            if (parties > ushortMask || unarrived > ushortMask)
458 <                throw new IllegalStateException(badBounds(parties, unarrived));
459 <            if (phase == phaseOf(root.state) &&
460 <                casState(s, stateFor(phase, parties, unarrived)))
461 <                break;
450 >        while ((phase = phaseOf(s = par==null? state:reconcileState())) >= 0) {
451 >            int p = partiesOf(s);
452 >            int u = unarrivedOf(s);
453 >            int unarrived = u + registrations;
454 >            int parties = p + registrations;
455 >            if (par == null || phase == phaseOf(root.state)) {
456 >                if (parties > ushortMask || unarrived > ushortMask)
457 >                    throw new IllegalStateException(badBounds(parties,
458 >                                                              unarrived));
459 >                else if (p != 0 && u == 0)       // back off if advancing
460 >                    Thread.yield();              // not worth actually blocking
461 >                else if (casState(s, stateFor(phase, parties, unarrived)))
462 >                    break;
463 >            }
464          }
465          return phase;
466      }
# Line 462 | Line 476 | public class Phaser {
476       * of unarrived parties would become negative
477       */
478      public int arrive() {
479 +        Phaser par = parent;
480 +        long s;
481          int phase;
482 <        for (;;) {
467 <            long s = state;
468 <            phase = phaseOf(s);
469 <            if (phase < 0)
470 <                break;
482 >        while ((phase = phaseOf(s = par==null? state:reconcileState())) >= 0) {
483              int parties = partiesOf(s);
484              int unarrived = unarrivedOf(s) - 1;
485 <            if (unarrived > 0) {        // Not the last arrival
486 <                if (casState(s, s - 1)) // s-1 adds one arrival
485 >            if (parties == 0 || unarrived < 0)
486 >                throw new IllegalStateException(badBounds(parties,
487 >                                                          unarrived));
488 >            else if (unarrived > 0) {           // Not the last arrival
489 >                if (casState(s, s - 1))         // s-1 adds one arrival
490                      break;
491              }
492 <            else if (unarrived == 0) {  // the last arrival
493 <                Phaser par = parent;
494 <                if (par == null) {      // directly trip
495 <                    if (casState
496 <                        (s,
497 <                         trippedStateFor(onAdvance(phase, parties) ? -1 :
483 <                                         ((phase + 1) & phaseMask), parties))) {
484 <                        releaseWaiters(phase);
485 <                        break;
486 <                    }
487 <                }
488 <                else {                  // cascade to parent
489 <                    if (casState(s, s - 1)) { // zeroes unarrived
490 <                        par.arrive();
491 <                        reconcileState();
492 <                        break;
493 <                    }
492 >            else if (par == null) {             // directly trip
493 >                if (casState(s, trippedStateFor(onAdvance(phase, parties) ? -1 :
494 >                                                ((phase + 1) & phaseMask),
495 >                                                parties))) {
496 >                    releaseWaiters(phase);
497 >                    break;
498                  }
499              }
500 <            else if (phase != phaseOf(root.state)) // or if unreconciled
500 >            else if (phaseOf(root.state) == phase && casState(s, s - 1)) {
501 >                par.arrive();                   // cascade to parent
502                  reconcileState();
503 <            else
504 <                throw new IllegalStateException(badBounds(parties, unarrived));
503 >                break;
504 >            }
505          }
506          return phase;
507      }
# Line 515 | Line 520 | public class Phaser {
520       * of registered or unarrived parties would become negative
521       */
522      public int arriveAndDeregister() {
523 <        // similar code to arrive, but too different to merge
523 >        // similar to arrive, but too different to merge
524          Phaser par = parent;
525 +        long s;
526          int phase;
527 <        for (;;) {
522 <            long s = state;
523 <            phase = phaseOf(s);
524 <            if (phase < 0)
525 <                break;
527 >        while ((phase = phaseOf(s = par==null? state:reconcileState())) >= 0) {
528              int parties = partiesOf(s) - 1;
529              int unarrived = unarrivedOf(s) - 1;
530 <            if (parties >= 0) {
531 <                if (unarrived > 0 || (unarrived == 0 && par != null)) {
532 <                    if (casState
533 <                        (s,
534 <                         stateFor(phase, parties, unarrived))) {
535 <                        if (unarrived == 0) {
536 <                            par.arriveAndDeregister();
537 <                            reconcileState();
538 <                        }
539 <                        break;
540 <                    }
541 <                    continue;
542 <                }
541 <                if (unarrived == 0) {
542 <                    if (casState
543 <                        (s,
544 <                         trippedStateFor(onAdvance(phase, parties) ? -1 :
545 <                                         ((phase + 1) & phaseMask), parties))) {
546 <                        releaseWaiters(phase);
547 <                        break;
548 <                    }
549 <                    continue;
550 <                }
551 <                if (par != null && phase != phaseOf(root.state)) {
552 <                    reconcileState();
553 <                    continue;
530 >            if (parties < 0 || unarrived < 0)
531 >                throw new IllegalStateException(badBounds(parties,
532 >                                                          unarrived));
533 >            else if (unarrived > 0) {
534 >                if (casState(s, stateFor(phase, parties, unarrived)))
535 >                    break;
536 >            }
537 >            else if (par == null) {
538 >                if (casState(s, trippedStateFor(onAdvance(phase, parties)? -1:
539 >                                                (phase + 1) & phaseMask,
540 >                                                parties))) {
541 >                    releaseWaiters(phase);
542 >                    break;
543                  }
544              }
545 <            throw new IllegalStateException(badBounds(parties, unarrived));
545 >            else if (phaseOf(root.state) == phase &&
546 >                     casState(s, stateFor(phase, parties, 0))) {
547 >                if (parties == 0)
548 >                    par.arriveAndDeregister();
549 >                else
550 >                    par.arrive();
551 >                reconcileState();
552 >                break;
553 >            }
554          }
555          return phase;
556      }
# Line 591 | Line 588 | public class Phaser {
588      public int awaitAdvance(int phase) {
589          if (phase < 0)
590              return phase;
591 <        long s = getReconciledState();
595 <        int p = phaseOf(s);
591 >        int p = getPhase();
592          if (p != phase)
593              return p;
598 –        if (unarrivedOf(s) == 0 && parent != null)
599 –            parent.awaitAdvance(phase);
600 –        // Fall here even if parent waited, to reconcile and help release
594          return untimedWait(phase);
595      }
596  
# Line 620 | Line 613 | public class Phaser {
613          throws InterruptedException {
614          if (phase < 0)
615              return phase;
616 <        long s = getReconciledState();
624 <        int p = phaseOf(s);
616 >        int p = getPhase();
617          if (p != phase)
618              return p;
627 –        if (unarrivedOf(s) == 0 && parent != null)
628 –            parent.awaitAdvanceInterruptibly(phase);
619          return interruptibleWait(phase);
620      }
621  
# Line 653 | Line 643 | public class Phaser {
643      public int awaitAdvanceInterruptibly(int phase,
644                                           long timeout, TimeUnit unit)
645          throws InterruptedException, TimeoutException {
646 +        long nanos = unit.toNanos(timeout);
647          if (phase < 0)
648              return phase;
649 <        long s = getReconciledState();
659 <        int p = phaseOf(s);
649 >        int p = getPhase();
650          if (p != phase)
651              return p;
652 <        if (unarrivedOf(s) == 0 && parent != null)
663 <            parent.awaitAdvanceInterruptibly(phase, timeout, unit);
664 <        return timedWait(phase, unit.toNanos(timeout));
652 >        return timedWait(phase, nanos);
653      }
654  
655      /**
# Line 672 | Line 660 | public class Phaser {
660       * unexpected exceptions.
661       */
662      public void forceTermination() {
663 <        for (;;) {
664 <            long s = getReconciledState();
665 <            int phase = phaseOf(s);
666 <            int parties = partiesOf(s);
667 <            int unarrived = unarrivedOf(s);
668 <            if (phase < 0 ||
669 <                casState(s, stateFor(-1, parties, unarrived))) {
682 <                releaseWaiters(0);
683 <                releaseWaiters(1);
684 <                if (parent != null)
685 <                    parent.forceTermination();
686 <                return;
687 <            }
688 <        }
663 >        Phaser r = root;    // force at root then reconcile
664 >        long s;
665 >        while (phaseOf(s = r.state) >= 0)
666 >            r.casState(s, stateFor(-1, partiesOf(s), unarrivedOf(s)));
667 >        reconcileState();
668 >        releaseWaiters(0);  // ensure wakeups on both queues
669 >        releaseWaiters(1);
670      }
671  
672      /**
# Line 705 | Line 686 | public class Phaser {
686       * @return the number of parties
687       */
688      public int getRegisteredParties() {
689 <        return partiesOf(state);
689 >        return partiesOf(getReconciledState());
690      }
691  
692      /**
# Line 715 | Line 696 | public class Phaser {
696       * @return the number of arrived parties
697       */
698      public int getArrivedParties() {
699 <        return arrivedOf(state);
699 >        return arrivedOf(getReconciledState());
700      }
701  
702      /**
# Line 725 | Line 706 | public class Phaser {
706       * @return the number of unarrived parties
707       */
708      public int getUnarrivedParties() {
709 <        return unarrivedOf(state);
709 >        return unarrivedOf(getReconciledState());
710      }
711  
712      /**
# Line 779 | Line 760 | public class Phaser {
760       * termination for other reasons should also preserve this
761       * property.
762       *
782 –     * <p>You may override this method to perform an action with side
783 –     * effects visible to participating tasks, but it is only sensible
784 –     * to do so in designs where all parties register before any
785 –     * arrive, and all {@link #awaitAdvance} at each phase.
786 –     * Otherwise, you cannot ensure lack of interference from other
787 –     * parties during the invocation of this method. Additionally,
788 –     * method {@code onAdvance} may be invoked more than once per
789 –     * transition if registrations are intermixed with arrivals.
790 –     *
763       * @param phase the phase number on entering the barrier
764       * @param registeredParties the current number of registered parties
765       * @return {@code true} if this barrier should terminate

Diff Legend

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