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.75 by dl, Wed Sep 21 12:30:39 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 <                long n = s & PARTIES_MASK;  // base of next state
364 <                int nextUnarrived = (int)n >>> PARTIES_SHIFT;
365 <                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                          if (onAdvance(phase, nextUnarrived))
368                              n |= TERMINATION_BIT;
# Line 370 | Line 370 | public class Phaser {
370                              n |= EMPTY;
371                          else
372                              n |= nextUnarrived;
373 <                        n |= (long)((phase + 1) & MAX_PHASE) << PHASE_SHIFT;
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(true);
379 >                        phase = parent.doArrive(ONE_DEREGISTER);
380                          UNSAFE.compareAndSwapLong(this, stateOffset,
381                                                    s, s | EMPTY);
382                      }
383                      else
384 <                        phase = parent.doArrive(false);
383 <                    releaseWaiters(phase);
384 >                        phase = parent.doArrive(ONE_ARRIVAL);
385                  }
386                  return phase;
387              }
# Line 395 | 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 (;;) {
# Line 405 | Line 406 | public class Phaser {
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
432 <                                 (this, stateOffset, state,
433 <                                  ((long)phase << PHASE_SHIFT) | adj));
440 >                            // assert (int)s == EMPTY;
441 >                        }
442                          break;
443                      }
444                  }
# Line 598 | 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 618 | 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 645 | Line 653 | public class Phaser {
653          for (;;) {
654              long s = (root == this) ? state : reconcileState();
655              int phase = (int)(s >>> PHASE_SHIFT);
648 –            int counts = (int)s;
649 –            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 788 | 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 995 | 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 1003 | 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