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.74 by jsr166, Wed Jun 1 21:04:30 2011 UTC vs.
Revision 1.76 by jsr166, Sat Oct 15 21:46:25 2011 UTC

# Line 276 | Line 276 | public class Phaser {
276      // some special values
277      private static final int  ONE_ARRIVAL     = 1;
278      private static final int  ONE_PARTY       = 1 << PARTIES_SHIFT;
279 +    private static final int  ONE_DEREGISTER  = ONE_ARRIVAL|ONE_PARTY;
280      private static final int  EMPTY           = 1;
281  
282      // The following unpacking methods are usually manually inlined
283  
284      private static int unarrivedOf(long s) {
285          int counts = (int)s;
286 <        return (counts == EMPTY) ? 0 : counts & UNARRIVED_MASK;
286 >        return (counts == EMPTY) ? 0 : (counts & UNARRIVED_MASK);
287      }
288  
289      private static int partiesOf(long s) {
# Line 343 | Line 344 | public class Phaser {
344       * Manually tuned to speed up and minimize race windows for the
345       * common case of just decrementing unarrived field.
346       *
347 <     * @param deregister false for arrive, true for arriveAndDeregister
347 >     * @param adjust value to subtract from state;
348 >     *               ONE_ARRIVAL for arrive,
349 >     *               ONE_DEREGISTER for arriveAndDeregister
350       */
351 <    private int doArrive(boolean deregister) {
349 <        int adj = deregister ? ONE_ARRIVAL|ONE_PARTY : ONE_ARRIVAL;
351 >    private int doArrive(int adjust) {
352          final Phaser root = this.root;
353          for (;;) {
354              long s = (root == this) ? state : reconcileState();
355              int phase = (int)(s >>> PHASE_SHIFT);
354 –            int counts = (int)s;
355 –            int unarrived = (counts & UNARRIVED_MASK) - 1;
356              if (phase < 0)
357                  return phase;
358 <            else if (counts == EMPTY || unarrived < 0) {
359 <                if (root == this || reconcileState() == s)
360 <                    throw new IllegalStateException(badArrive(s));
361 <            }
362 <            else if (UNSAFE.compareAndSwapLong(this, stateOffset, s, s-=adj)) {
363 <                if (unarrived == 0) {
358 >            int counts = (int)s;
359 >            int unarrived = (counts == EMPTY) ? 0 : (counts & UNARRIVED_MASK);
360 >            if (unarrived <= 0)
361 >                throw new IllegalStateException(badArrive(s));
362 >            if (UNSAFE.compareAndSwapLong(this, stateOffset, s, s-=adjust)) {
363 >                if (unarrived == 1) {
364                      long n = s & PARTIES_MASK;  // base of next state
365                      int nextUnarrived = (int)n >>> PARTIES_SHIFT;
366 <                    if (root != this)
367 <                        return parent.doArrive(nextUnarrived == 0);
368 <                    if (onAdvance(phase, nextUnarrived))
369 <                        n |= TERMINATION_BIT;
370 <                    else if (nextUnarrived == 0)
371 <                        n |= EMPTY;
366 >                    if (root == this) {
367 >                        if (onAdvance(phase, nextUnarrived))
368 >                            n |= TERMINATION_BIT;
369 >                        else if (nextUnarrived == 0)
370 >                            n |= EMPTY;
371 >                        else
372 >                            n |= nextUnarrived;
373 >                        int nextPhase = (phase + 1) & MAX_PHASE;
374 >                        n |= (long)nextPhase << PHASE_SHIFT;
375 >                        UNSAFE.compareAndSwapLong(this, stateOffset, s, n);
376 >                        releaseWaiters(phase);
377 >                    }
378 >                    else if (nextUnarrived == 0) { // propagate deregistration
379 >                        phase = parent.doArrive(ONE_DEREGISTER);
380 >                        UNSAFE.compareAndSwapLong(this, stateOffset,
381 >                                                  s, s | EMPTY);
382 >                    }
383                      else
384 <                        n |= nextUnarrived;
374 <                    n |= (long)((phase + 1) & MAX_PHASE) << PHASE_SHIFT;
375 <                    UNSAFE.compareAndSwapLong(this, stateOffset, s, n);
376 <                    releaseWaiters(phase);
384 >                        phase = parent.doArrive(ONE_ARRIVAL);
385                  }
386                  return phase;
387              }
# Line 388 | Line 396 | public class Phaser {
396       */
397      private int doRegister(int registrations) {
398          // adjustment to state
399 <        long adj = ((long)registrations << PARTIES_SHIFT) | registrations;
399 >        long adjust = ((long)registrations << PARTIES_SHIFT) | registrations;
400          final Phaser parent = this.parent;
401          int phase;
402          for (;;) {
403 <            long s = state;
403 >            long s = (parent == null) ? state : reconcileState();
404              int counts = (int)s;
405              int parties = counts >>> PARTIES_SHIFT;
406              int unarrived = counts & UNARRIVED_MASK;
407              if (registrations > MAX_PARTIES - parties)
408                  throw new IllegalStateException(badRegister(s));
409 <            else if ((phase = (int)(s >>> PHASE_SHIFT)) < 0)
409 >            phase = (int)(s >>> PHASE_SHIFT);
410 >            if (phase < 0)
411                  break;
412 <            else if (counts != EMPTY) {             // not 1st registration
412 >            if (counts != EMPTY) {                  // not 1st registration
413                  if (parent == null || reconcileState() == s) {
414                      if (unarrived == 0)             // wait out advance
415                          root.internalAwaitAdvance(phase, null);
416                      else if (UNSAFE.compareAndSwapLong(this, stateOffset,
417 <                                                       s, s + adj))
417 >                                                       s, s + adjust))
418                          break;
419                  }
420              }
421              else if (parent == null) {              // 1st root registration
422 <                long next = ((long)phase << PHASE_SHIFT) | adj;
422 >                long next = ((long)phase << PHASE_SHIFT) | adjust;
423                  if (UNSAFE.compareAndSwapLong(this, stateOffset, s, next))
424                      break;
425              }
426              else {
427                  synchronized (this) {               // 1st sub registration
428                      if (state == s) {               // recheck under lock
429 <                        parent.doRegister(1);
430 <                        do {                        // force current phase
429 >                        phase = parent.doRegister(1);
430 >                        if (phase < 0)
431 >                            break;
432 >                        // finish registration whenever parent registration
433 >                        // succeeded, even when racing with termination,
434 >                        // since these are part of the same "transaction".
435 >                        while (!UNSAFE.compareAndSwapLong
436 >                               (this, stateOffset, s,
437 >                                ((long)phase << PHASE_SHIFT) | adjust)) {
438 >                            s = state;
439                              phase = (int)(root.state >>> PHASE_SHIFT);
440 <                            // assert phase < 0 || (int)state == EMPTY;
441 <                        } while (!UNSAFE.compareAndSwapLong
425 <                                 (this, stateOffset, state,
426 <                                  ((long)phase << PHASE_SHIFT) | adj));
440 >                            // assert (int)s == EMPTY;
441 >                        }
442                          break;
443                      }
444                  }
# Line 458 | Line 473 | public class Phaser {
473                      s = (((long)phase << PHASE_SHIFT) |
474                           (s & PARTIES_MASK) |
475                           ((p = (int)s >>> PARTIES_SHIFT) == 0 ? EMPTY :
476 <                          (u = (int)s & UNARRIVED_MASK) == 0 ? p : u))))
476 >                          ((u = (int)s & UNARRIVED_MASK) == 0 && phase >= 0) ?
477 >                          p : u))))
478                  s = state;
479          }
480          return s;
# Line 590 | Line 606 | public class Phaser {
606       * of unarrived parties would become negative
607       */
608      public int arrive() {
609 <        return doArrive(false);
609 >        return doArrive(ONE_ARRIVAL);
610      }
611  
612      /**
# Line 610 | Line 626 | public class Phaser {
626       * of registered or unarrived parties would become negative
627       */
628      public int arriveAndDeregister() {
629 <        return doArrive(true);
629 >        return doArrive(ONE_DEREGISTER);
630      }
631  
632      /**
# Line 637 | Line 653 | public class Phaser {
653          for (;;) {
654              long s = (root == this) ? state : reconcileState();
655              int phase = (int)(s >>> PHASE_SHIFT);
640 –            int counts = (int)s;
641 –            int unarrived = (counts & UNARRIVED_MASK) - 1;
656              if (phase < 0)
657                  return phase;
658 <            else if (counts == EMPTY || unarrived < 0) {
659 <                if (reconcileState() == s)
660 <                    throw new IllegalStateException(badArrive(s));
661 <            }
662 <            else if (UNSAFE.compareAndSwapLong(this, stateOffset, s,
663 <                                               s -= ONE_ARRIVAL)) {
664 <                if (unarrived != 0)
658 >            int counts = (int)s;
659 >            int unarrived = (counts == EMPTY) ? 0 : (counts & UNARRIVED_MASK);
660 >            if (unarrived <= 0)
661 >                throw new IllegalStateException(badArrive(s));
662 >            if (UNSAFE.compareAndSwapLong(this, stateOffset, s,
663 >                                          s -= ONE_ARRIVAL)) {
664 >                if (unarrived > 1)
665                      return root.internalAwaitAdvance(phase, null);
666                  if (root != this)
667                      return parent.arriveAndAwaitAdvance();
# Line 780 | Line 794 | public class Phaser {
794              if (UNSAFE.compareAndSwapLong(root, stateOffset,
795                                            s, s | TERMINATION_BIT)) {
796                  // signal all threads
797 <                releaseWaiters(0);
798 <                releaseWaiters(1);
797 >                releaseWaiters(0); // Waiters on evenQ
798 >                releaseWaiters(1); // Waiters on oddQ
799                  return;
800              }
801          }
# Line 987 | Line 1001 | public class Phaser {
1001  
1002      /**
1003       * Possibly blocks and waits for phase to advance unless aborted.
1004 <     * Call only from root node.
1004 >     * Call only on root phaser.
1005       *
1006       * @param phase current phase
1007       * @param node if non-null, the wait node to track interrupt and timeout;
# Line 995 | Line 1009 | public class Phaser {
1009       * @return current phase
1010       */
1011      private int internalAwaitAdvance(int phase, QNode node) {
1012 +        // assert root == this;
1013          releaseWaiters(phase-1);          // ensure old queue clean
1014          boolean queued = false;           // true when node is enqueued
1015          int lastUnarrived = 0;            // to increase spins upon change

Diff Legend

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