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.72 by dl, Mon May 16 11:41:14 2011 UTC vs.
Revision 1.76 by jsr166, Sat Oct 15 21:46:25 2011 UTC

# Line 237 | Line 237 | public class Phaser {
237       */
238  
239      /**
240 <     * Primary state representation, holding four fields:
240 >     * Primary state representation, holding four bit-fields:
241       *
242 <     * * unarrived -- the number of parties yet to hit barrier (bits  0-15)
243 <     * * parties -- the number of parties to wait              (bits 16-31)
244 <     * * phase -- the generation of the barrier                (bits 32-62)
245 <     * * terminated -- set if barrier is terminated            (bit  63 / sign)
242 >     * unarrived  -- the number of parties yet to hit barrier (bits  0-15)
243 >     * parties    -- the number of parties to wait            (bits 16-31)
244 >     * phase      -- the generation of the barrier            (bits 32-62)
245 >     * terminated -- set if barrier is terminated             (bit  63 / sign)
246       *
247       * Except that a phaser with no registered parties is
248 <     * distinguished with the otherwise illegal state of having zero
248 >     * distinguished by the otherwise illegal state of having zero
249       * parties and one unarrived parties (encoded as EMPTY below).
250       *
251       * To efficiently maintain atomicity, these values are packed into
# Line 266 | Line 266 | public class Phaser {
266      private volatile long state;
267  
268      private static final int  MAX_PARTIES     = 0xffff;
269 <    private static final int  MAX_PHASE       = 0x7fffffff;
269 >    private static final int  MAX_PHASE       = Integer.MAX_VALUE;
270      private static final int  PARTIES_SHIFT   = 16;
271      private static final int  PHASE_SHIFT     = 32;
272 –    private static final long PHASE_MASK      = -1L << PHASE_SHIFT;
272      private static final int  UNARRIVED_MASK  = 0xffff;      // to mask ints
273      private static final long PARTIES_MASK    = 0xffff0000L; // to mask longs
274      private static final long TERMINATION_BIT = 1L << 63;
# Line 277 | 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 291 | Line 291 | public class Phaser {
291      }
292  
293      private static int phaseOf(long s) {
294 <        return (int) (s >>> PHASE_SHIFT);
294 >        return (int)(s >>> PHASE_SHIFT);
295      }
296  
297      private static int arrivedOf(long s) {
# Line 344 | 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) {
350 <        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);
355 –            int counts = (int)s;
356 –            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;
365 >                    int nextUnarrived = (int)n >>> PARTIES_SHIFT;
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;
375 <                    n |= ((long)((phase + 1) & MAX_PHASE)) << PHASE_SHIFT;
376 <                    UNSAFE.compareAndSwapLong(this, stateOffset, s, n);
377 <                    releaseWaiters(phase);
384 >                        phase = parent.doArrive(ONE_ARRIVAL);
385                  }
386                  return phase;
387              }
# Line 389 | Line 396 | public class Phaser {
396       */
397      private int doRegister(int registrations) {
398          // adjustment to state
399 <        long adj = ((long)registrations << PARTIES_SHIFT) | registrations;
400 <        Phaser par = parent;
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
413 <                if (par == null || reconcileState() == s) {
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 (par == null) {                 // 1st root registration
422 <                long next = (((long) phase) << PHASE_SHIFT) | adj;
421 >            else if (parent == null) {              // 1st root registration
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 <                        par.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
426 <                                 (this, stateOffset, state,
427 <                                  (((long) phase) << PHASE_SHIFT) | adj));
440 >                            // assert (int)s == EMPTY;
441 >                        }
442                          break;
443                      }
444                  }
# Line 456 | Line 470 | public class Phaser {
470                     (int)(s >>> PHASE_SHIFT) &&
471                     !UNSAFE.compareAndSwapLong
472                     (this, stateOffset, s,
473 <                    s = ((((long) phase) << PHASE_SHIFT) | (s & PARTIES_MASK) |
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 525 | Line 541 | public class Phaser {
541              this.evenQ = new AtomicReference<QNode>();
542              this.oddQ = new AtomicReference<QNode>();
543          }
544 <        this.state = (parties == 0) ? (long) EMPTY :
545 <            ((((long) phase) << PHASE_SHIFT) |
546 <             (((long) parties) << PARTIES_SHIFT) |
547 <             ((long) parties));
544 >        this.state = (parties == 0) ? (long)EMPTY :
545 >            ((long)phase << PHASE_SHIFT) |
546 >            ((long)parties << PARTIES_SHIFT) |
547 >            ((long)parties);
548      }
549  
550      /**
# 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();
668                  long n = s & PARTIES_MASK;  // base of next state
669 <                int nextUnarrived = ((int)n) >>> PARTIES_SHIFT;
669 >                int nextUnarrived = (int)n >>> PARTIES_SHIFT;
670                  if (onAdvance(phase, nextUnarrived))
671                      n |= TERMINATION_BIT;
672                  else if (nextUnarrived == 0)
# 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
# Line 1108 | Line 1123 | public class Phaser {
1123  
1124      // Unsafe mechanics
1125  
1126 <    private static final sun.misc.Unsafe UNSAFE = getUnsafe();
1127 <    private static final long stateOffset =
1128 <        objectFieldOffset("state", Phaser.class);
1114 <
1115 <    private static long objectFieldOffset(String field, Class<?> klazz) {
1126 >    private static final sun.misc.Unsafe UNSAFE;
1127 >    private static final long stateOffset;
1128 >    static {
1129          try {
1130 <            return UNSAFE.objectFieldOffset(klazz.getDeclaredField(field));
1131 <        } catch (NoSuchFieldException e) {
1132 <            // Convert Exception to corresponding Error
1133 <            NoSuchFieldError error = new NoSuchFieldError(field);
1134 <            error.initCause(e);
1135 <            throw error;
1130 >            UNSAFE = getUnsafe();
1131 >            Class<?> k = Phaser.class;
1132 >            stateOffset = UNSAFE.objectFieldOffset
1133 >                (k.getDeclaredField("state"));
1134 >        } catch (Exception e) {
1135 >            throw new Error(e);
1136          }
1137      }
1138  

Diff Legend

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