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.56 by dl, Wed Nov 17 10:48:59 2010 UTC vs.
Revision 1.59 by dl, Sat Nov 27 16:46:53 2010 UTC

# Line 79 | Line 79 | import java.util.concurrent.locks.LockSu
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}.  As illustrated below, when
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
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
89   * threads and allow them to terminate.
90   *
91 < * <p> <b>Tiering.</b> Phasers may be <em>tiered</em> (i.e., arranged
92 < * in tree structures) to reduce contention. Phasers with large
93 < * numbers of parties that would otherwise experience heavy
91 > * <p> <b>Tiering.</b> Phasers may be <em>tiered</em> (i.e.,
92 > * constructed in tree structures) to reduce contention. Phasers with
93 > * large numbers of parties that would otherwise experience heavy
94   * synchronization contention costs may instead be set up so that
95   * groups of sub-phasers share a common parent.  This may greatly
96   * increase throughput even though it incurs greater per-operation
# Line 240 | Line 242 | public class Phaser {
242       */
243      private volatile long state;
244  
245 <    private static final int  MAX_PARTIES    = 0xffff;
246 <    private static final int  MAX_PHASE      = 0x7fffffff;
247 <    private static final int  PARTIES_SHIFT  = 16;
248 <    private static final int  PHASE_SHIFT    = 32;
249 <    private static final int  UNARRIVED_MASK = 0xffff;
250 <    private static final int  PARTIES_MASK   = 0xffff0000;
251 <    private static final long LPARTIES_MASK  = 0xffff0000L; // long version
252 <    private static final long ONE_ARRIVAL    = 1L;
253 <    private static final long ONE_PARTY      = 1L << PARTIES_SHIFT;
252 <    private static final long TERMINATION_PHASE  = -1L << PHASE_SHIFT;
245 >    private static final int  MAX_PARTIES     = 0xffff;
246 >    private static final int  MAX_PHASE       = 0x7fffffff;
247 >    private static final int  PARTIES_SHIFT   = 16;
248 >    private static final int  PHASE_SHIFT     = 32;
249 >    private static final int  UNARRIVED_MASK  = 0xffff;      // to mask ints
250 >    private static final long PARTIES_MASK    = 0xffff0000L; // to mask longs
251 >    private static final long ONE_ARRIVAL     = 1L;
252 >    private static final long ONE_PARTY       = 1L << PARTIES_SHIFT;
253 >    private static final long TERMINATION_BIT = 1L << 63;
254  
255      // The following unpacking methods are usually manually inlined
256  
# Line 294 | Line 295 | public class Phaser {
295      }
296  
297      /**
298 +     * Returns message string for bounds exceptions on arrival.
299 +     */
300 +    private String badArrive(long s) {
301 +        return "Attempted arrival of unregistered party for " +
302 +            stateToString(s);
303 +    }
304 +
305 +    /**
306 +     * Returns message string for bounds exceptions on registration.
307 +     */
308 +    private String badRegister(long s) {
309 +        return "Attempt to register more than " +
310 +            MAX_PARTIES + " parties for " + stateToString(s);
311 +    }
312 +
313 +    /**
314       * Main implementation for methods arrive and arriveAndDeregister.
315       * Manually tuned to speed up and minimize race windows for the
316       * common case of just decrementing unarrived field.
# Line 304 | Line 321 | public class Phaser {
321       */
322      private int doArrive(long adj) {
323          for (;;) {
324 <            long s;
325 <            int phase, unarrived;
326 <            if ((phase = (int)((s = state) >>> PHASE_SHIFT)) < 0)
324 >            long s = state;
325 >            int unarrived = (int)s & UNARRIVED_MASK;
326 >            int phase = (int)(s >>> PHASE_SHIFT);
327 >            if (phase < 0)
328                  return phase;
329 <            else if ((unarrived = (int)s & UNARRIVED_MASK) == 0)
330 <                checkBadArrive(s);
329 >            else if (unarrived == 0) {
330 >                if (reconcileState() == s)     // recheck
331 >                    throw new IllegalStateException(badArrive(s));
332 >            }
333              else if (UNSAFE.compareAndSwapLong(this, stateOffset, s, s-=adj)) {
334                  if (unarrived == 1) {
335 <                    Phaser par;
316 <                    long p = s & LPARTIES_MASK; // unshifted parties field
335 >                    long p = s & PARTIES_MASK; // unshifted parties field
336                      long lu = p >>> PARTIES_SHIFT;
337                      int u = (int)lu;
338                      int nextPhase = (phase + 1) & MAX_PHASE;
339                      long next = ((long)nextPhase << PHASE_SHIFT) | p | lu;
340 <                    if ((par = parent) == null) {
340 >                    final Phaser parent = this.parent;
341 >                    if (parent == null) {
342                          if (onAdvance(phase, u))
343 <                            next |= TERMINATION_PHASE; // obliterate phase
343 >                            next |= TERMINATION_BIT;
344                          UNSAFE.compareAndSwapLong(this, stateOffset, s, next);
345                          releaseWaiters(phase);
346                      }
347                      else {
348 <                        par.doArrive(u == 0?
349 <                                     ONE_ARRIVAL|ONE_PARTY : ONE_ARRIVAL);
350 <                        if ((int)(par.state >>> PHASE_SHIFT) != nextPhase ||
348 >                        parent.doArrive((u == 0) ?
349 >                                        ONE_ARRIVAL|ONE_PARTY : ONE_ARRIVAL);
350 >                        if ((int)(parent.state >>> PHASE_SHIFT) != nextPhase ||
351                              ((int)(state >>> PHASE_SHIFT) != nextPhase &&
352                               !UNSAFE.compareAndSwapLong(this, stateOffset,
353                                                          s, next)))
# Line 340 | Line 360 | public class Phaser {
360      }
361  
362      /**
343     * Rechecks state and throws bounds exceptions on arrival -- called
344     * only if unarrived is apparently zero.
345     */
346    private void checkBadArrive(long s) {
347        if (reconcileState() == s)
348            throw new IllegalStateException
349                ("Attempted arrival of unregistered party for " +
350                 stateToString(s));
351    }
352
353    /**
363       * Implementation of register, bulkRegister
364       *
365 <     * @param registrations number to add to both parties and unarrived fields
365 >     * @param registrations number to add to both parties and
366 >     * unarrived fields. Must be greater than zero.
367       */
368      private int doRegister(int registrations) {
369 <        long adj = (long)registrations; // adjustment to state
370 <        adj |= adj << PARTIES_SHIFT;
371 <        Phaser par = parent;
369 >        // adjustment to state
370 >        long adj = ((long)registrations << PARTIES_SHIFT) | registrations;
371 >        final Phaser parent = this.parent;
372          for (;;) {
373 <            int phase, parties;
374 <            long s = par == null? state : reconcileState();
375 <            if ((phase = (int)(s >>> PHASE_SHIFT)) < 0)
373 >            long s = (parent == null) ? state : reconcileState();
374 >            int parties = (int)s >>> PARTIES_SHIFT;
375 >            int phase = (int)(s >>> PHASE_SHIFT);
376 >            if (phase < 0)
377                  return phase;
378 <            if ((parties = (int)s >>> PARTIES_SHIFT) != 0 &&
368 <                ((int)s & UNARRIVED_MASK) == 0)
369 <                internalAwaitAdvance(phase, null); // wait for onAdvance
370 <            else if (parties + registrations > MAX_PARTIES)
378 >            else if (registrations > MAX_PARTIES - parties)
379                  throw new IllegalStateException(badRegister(s));
380 <            else if (UNSAFE.compareAndSwapLong(this, stateOffset, s, s + adj))
381 <                return phase;
380 >            else if ((parties == 0 && parent == null) || // first reg of root
381 >                     ((int)s & UNARRIVED_MASK) != 0) {   // not advancing
382 >                if (UNSAFE.compareAndSwapLong(this, stateOffset, s, s + adj))
383 >                    return phase;
384 >            }
385 >            else if (parties != 0)               // wait for onAdvance
386 >                internalAwaitAdvance(phase, null);
387 >            else {                               // 1st registration of child
388 >                synchronized(this) {             // register parent first
389 >                    if (reconcileState() == s) { // recheck under lock
390 >                        parent.doRegister(1);    // OK if throws IllegalState
391 >                        for (;;) {               // simpler form of outer loop
392 >                            s = reconcileState();
393 >                            phase = (int)(s >>> PHASE_SHIFT);
394 >                            if (phase < 0 ||
395 >                                UNSAFE.compareAndSwapLong(this, stateOffset,
396 >                                                          s, s + adj))
397 >                                return phase;
398 >                        }
399 >                    }
400 >                }
401 >            }
402          }
403      }
404  
405      /**
378     * Returns message string for out of bounds exceptions on registration.
379     */
380    private String badRegister(long s) {
381        return "Attempt to register more than " +
382            MAX_PARTIES + " parties for " + stateToString(s);
383    }
384
385    /**
406       * Recursively resolves lagged phase propagation from root if necessary.
407       */
408      private long reconcileState() {
409          Phaser par = parent;
410 <        if (par == null)
411 <            return state;
412 <        Phaser rt = root;
413 <        for (;;) {
414 <            long s, u;
415 <            int phase, rPhase, pPhase;
416 <            if ((phase = (int)((s = state)>>> PHASE_SHIFT)) < 0 ||
417 <                (rPhase = (int)(rt.state >>> PHASE_SHIFT)) == phase)
418 <                return s;
419 <            long pState = par.parent == null? par.state : par.reconcileState();
420 <            if (state == s) {
421 <                if ((rPhase < 0 || ((int)s & UNARRIVED_MASK) == 0) &&
422 <                    ((pPhase = (int)(pState >>> PHASE_SHIFT)) < 0 ||
423 <                     pPhase == ((phase + 1) & MAX_PHASE)))
424 <                    UNSAFE.compareAndSwapLong
405 <                        (this, stateOffset, s,
406 <                         (((long) pPhase) << PHASE_SHIFT) |
407 <                         (u = s & LPARTIES_MASK) |
408 <                         (u >>> PARTIES_SHIFT)); // reset unarrived to parties
409 <                else
410 <                    releaseWaiters(phase); // help release others
410 >        long s = state;
411 >        if (par != null) {
412 >            Phaser rt = root;
413 >            int phase, rPhase;
414 >            while ((phase = (int)(s >>> PHASE_SHIFT)) >= 0 &&
415 >                   (rPhase = (int)(rt.state >>> PHASE_SHIFT)) != phase) {
416 >                if ((int)(par.state >>> PHASE_SHIFT) != rPhase)
417 >                    par.reconcileState();
418 >                else if (rPhase < 0 || ((int)s & UNARRIVED_MASK) == 0) {
419 >                    long u = s & PARTIES_MASK; // reset unarrived to parties
420 >                    long next = ((((long) rPhase) << PHASE_SHIFT) | u |
421 >                                 (u >>> PARTIES_SHIFT));
422 >                    UNSAFE.compareAndSwapLong(this, stateOffset, s, next);
423 >                }
424 >                s = state;
425              }
426          }
427 +        return s;
428      }
429  
430      /**
# Line 434 | Line 449 | public class Phaser {
449      }
450  
451      /**
452 <     * Creates a new phaser with the given parent, without any
453 <     * initially registered parties. If parent is non-null this phaser
454 <     * is registered with the parent and its initial phase number is
455 <     * the same as that of parent phaser.
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 >     *
458 >     * Equivalent to {@link #Phaser(Phaser, int) Phaser(parent, 0)}.
459       *
460       * @param parent the parent phaser
461       */
# Line 447 | Line 465 | public class Phaser {
465  
466      /**
467       * Creates a new phaser with the given parent and number of
468 <     * registered unarrived parties. If parent is non-null, this phaser
469 <     * is registered with the parent and its initial phase number is
470 <     * the same as that of parent phaser.
468 >     * registered unarrived parties. If parent is non-null and
469 >     * the number of parties is non-zero, this phaser is registered
470 >     * with the parent.
471       *
472       * @param parent the parent phaser
473       * @param parties the number of parties required to trip barrier
# Line 466 | Line 484 | public class Phaser {
484              this.root = r;
485              this.evenQ = r.evenQ;
486              this.oddQ = r.oddQ;
487 <            phase = parent.register();
487 >            phase = (parties == 0) ? parent.getPhase() : parent.doRegister(1);
488          }
489          else {
490              this.root = this;
# Line 479 | Line 497 | public class Phaser {
497      }
498  
499      /**
500 <     * Adds a new unarrived party to this phaser.
501 <     * If an ongoing invocation of {@link #onAdvance} is in progress,
502 <     * this method may wait until its completion before registering.
500 >     * Adds a new unarrived party to this phaser.  If an ongoing
501 >     * invocation of {@link #onAdvance} is in progress, this method
502 >     * may wait until its completion before registering.  If this
503 >     * phaser has a parent, and this phaser previously had no
504 >     * registered parties, this phaser is also registered with its
505 >     * parent.
506       *
507       * @return the arrival phase number to which this registration applied
508       * @throws IllegalStateException if attempting to register more
# Line 495 | Line 516 | public class Phaser {
516       * Adds the given number of new unarrived parties to this phaser.
517       * If an ongoing invocation of {@link #onAdvance} is in progress,
518       * this method may wait until its completion before registering.
519 +     * If this phaser has a parent, and the given number of parities
520 +     * is greater than zero, and this phaser previously had no
521 +     * registered parties, this phaser is also registered with its
522 +     * parent.
523       *
524       * @param parties the number of additional parties required to trip barrier
525       * @return the arrival phase number to which this registration applied
# Line 505 | Line 530 | public class Phaser {
530      public int bulkRegister(int parties) {
531          if (parties < 0)
532              throw new IllegalArgumentException();
533 <        if (parties > MAX_PARTIES)
509 <            throw new IllegalStateException(badRegister(state));
510 <        if (parties == 0)
533 >        else if (parties == 0)
534              return getPhase();
535          return doRegister(parties);
536      }
537  
538      /**
539       * Arrives at the barrier, but does not wait for others.  (You can
540 <     * in turn wait for others via {@link #awaitAdvance}).  It is an
541 <     * unenforced usage error for an unregistered party to invoke this
542 <     * method.
540 >     * in turn wait for others via {@link #awaitAdvance}).  It is a
541 >     * usage error for an unregistered party to invoke this
542 >     * method. However, it is possible that this error will result in
543 >     * an {code IllegalStateException} only when some <em>other</em>
544 >     * party arrives.
545       *
546       * @return the arrival phase number, or a negative value if terminated
547       * @throws IllegalStateException if not terminated and the number
# Line 532 | Line 557 | public class Phaser {
557       * required to trip the barrier in future phases.  If this phaser
558       * has a parent, and deregistration causes this phaser to have
559       * zero parties, this phaser also arrives at and is deregistered
560 <     * from its parent.  It is an unenforced usage error for an
561 <     * unregistered party to invoke this method.
560 >     * from its parent.  It is a usage error for an unregistered party
561 >     * to invoke this method. However, it is possible that this error
562 >     * will result in an {code IllegalStateException} only when some
563 >     * <em>other</em> party arrives.
564       *
565       * @return the arrival phase number, or a negative value if terminated
566       * @throws IllegalStateException if not terminated and the number
# Line 549 | Line 576 | public class Phaser {
576       * interruption or timeout, you can arrange this with an analogous
577       * construction using one of the other forms of the {@code
578       * awaitAdvance} method.  If instead you need to deregister upon
579 <     * arrival, use {@link #arriveAndDeregister}. It is an unenforced
580 <     * usage error for an unregistered party to invoke this method.
579 >     * arrival, use {@link #arriveAndDeregister}.  It is a usage error
580 >     * for an unregistered party to invoke this method. However, it is
581 >     * possible that this error will result in an {code
582 >     * IllegalStateException} only when some <em>other</em> party
583 >     * arrives.
584       *
585       * @return the arrival phase number, or a negative number if terminated
586       * @throws IllegalStateException if not terminated and the number
# Line 573 | Line 603 | public class Phaser {
603       * if terminated or argument is negative
604       */
605      public int awaitAdvance(int phase) {
576        int p;
606          if (phase < 0)
607              return phase;
608 <        else if ((p = (int)((parent == null? state : reconcileState())
609 <                            >>> PHASE_SHIFT)) == phase)
610 <            return internalAwaitAdvance(phase, null);
582 <        else
583 <            return p;
608 >        long s = (parent == null) ? state : reconcileState();
609 >        int p = (int)(s >>> PHASE_SHIFT);
610 >        return (p != phase) ? p : internalAwaitAdvance(phase, null);
611      }
612  
613      /**
# Line 599 | Line 626 | public class Phaser {
626       */
627      public int awaitAdvanceInterruptibly(int phase)
628          throws InterruptedException {
602        int p;
629          if (phase < 0)
630              return phase;
631 <        if ((p = (int)((parent == null? state : reconcileState())
632 <                       >>> PHASE_SHIFT)) == phase) {
631 >        long s = (parent == null) ? state : reconcileState();
632 >        int p = (int)(s >>> PHASE_SHIFT);
633 >        if (p == phase) {
634              QNode node = new QNode(this, phase, true, false, 0L);
635              p = internalAwaitAdvance(phase, node);
636              if (node.wasInterrupted)
# Line 635 | Line 662 | public class Phaser {
662      public int awaitAdvanceInterruptibly(int phase,
663                                           long timeout, TimeUnit unit)
664          throws InterruptedException, TimeoutException {
638        long nanos = unit.toNanos(timeout);
639        int p;
665          if (phase < 0)
666              return phase;
667 <        if ((p = (int)((parent == null? state : reconcileState())
668 <                       >>> PHASE_SHIFT)) == phase) {
667 >        long s = (parent == null) ? state : reconcileState();
668 >        int p = (int)(s >>> PHASE_SHIFT);
669 >        if (p == phase) {
670 >            long nanos = unit.toNanos(timeout);
671              QNode node = new QNode(this, phase, true, true, nanos);
672              p = internalAwaitAdvance(phase, node);
673              if (node.wasInterrupted)
# Line 666 | Line 693 | public class Phaser {
693          long s;
694          while ((s = root.state) >= 0) {
695              if (UNSAFE.compareAndSwapLong(root, stateOffset,
696 <                                          s, s | TERMINATION_PHASE)) {
696 >                                          s, s | TERMINATION_BIT)) {
697                  releaseWaiters(0); // signal all threads
698                  releaseWaiters(1);
699                  return;
# Line 682 | Line 709 | public class Phaser {
709       * @return the phase number, or a negative value if terminated
710       */
711      public final int getPhase() {
712 <        return (int)((parent==null? state : reconcileState()) >>> PHASE_SHIFT);
712 >        return (int)(root.state >>> PHASE_SHIFT);
713      }
714  
715      /**
# Line 691 | Line 718 | public class Phaser {
718       * @return the number of parties
719       */
720      public int getRegisteredParties() {
721 <        return partiesOf(parent==null? state : reconcileState());
721 >        return partiesOf(state);
722      }
723  
724      /**
# Line 739 | Line 766 | public class Phaser {
766       * @return {@code true} if this barrier has been terminated
767       */
768      public boolean isTerminated() {
769 <        return (parent == null? state : reconcileState()) < 0;
769 >        return root.state < 0L;
770      }
771  
772      /**
# Line 764 | Line 791 | public class Phaser {
791       * {@code onAdvance} is invoked only for its root Phaser on each
792       * advance.
793       *
794 <     * <p>The default version returns {@code true} when the number of
795 <     * registered parties is zero. Normally, overrides that arrange
796 <     * termination for other reasons should also preserve this
797 <     * property.
794 >     * <p>To support the most common use cases, the default
795 >     * implementation of this method returns {@code true} when the
796 >     * number of registered parties has become zero as the result of a
797 >     * party invoking {@code arriveAndDeregister}.  You can disable
798 >     * this behavior, thus enabling continuation upon future
799 >     * registrations, by overriding this method to always return
800 >     * {@code false}:
801 >     *
802 >     * <pre> {@code
803 >     * Phaser phaser = new Phaser() {
804 >     *   protected boolean onAdvance(int phase, int parties) { return false; }
805 >     * }}</pre>
806       *
807       * @param phase the phase number on entering the barrier
808       * @param registeredParties the current number of registered parties
# Line 803 | Line 838 | public class Phaser {
838      // Waiting mechanics
839  
840      /**
841 <     * Removes and signals threads from queue for phase
841 >     * Removes and signals threads from queue for phase.
842       */
843      private void releaseWaiters(int phase) {
844          AtomicReference<QNode> head = queueFor(phase);
# Line 817 | Line 852 | public class Phaser {
852          }
853      }
854  
820    /**
821     * Tries to enqueue given node in the appropriate wait queue.
822     *
823     * @return true if successful
824     */
825    private boolean tryEnqueue(int phase, QNode node) {
826        releaseWaiters(phase-1); // ensure old queue clean
827        AtomicReference<QNode> head = queueFor(phase);
828        QNode q = head.get();
829        return ((q == null || q.phase == phase) &&
830                (int)(root.state >>> PHASE_SHIFT) == phase &&
831                head.compareAndSet(node.next = q, node));
832    }
833
855      /** The number of CPUs, for spin control */
856      private static final int NCPU = Runtime.getRuntime().availableProcessors();
857  
# Line 862 | Line 883 | public class Phaser {
883          boolean queued = false;      // true when node is enqueued
884          int lastUnarrived = -1;      // to increase spins upon change
885          int spins = SPINS_PER_ARRIVAL;
886 <        for (;;) {
887 <            int p, unarrived;
886 >        long s;
887 >        int p;
888 >        while ((p = (int)((s = current.state) >>> PHASE_SHIFT)) == phase) {
889              Phaser par;
890 <            long s = current.state;
891 <            if ((p = (int)(s >>> PHASE_SHIFT)) != phase) {
892 <                if (node != null)
893 <                    node.onRelease();
894 <                releaseWaiters(phase);
895 <                return p;
890 >            int unarrived = (int)s & UNARRIVED_MASK;
891 >            if (unarrived != lastUnarrived) {
892 >                if (lastUnarrived == -1) // ensure old queue clean
893 >                    releaseWaiters(phase-1);
894 >                if ((lastUnarrived = unarrived) < NCPU)
895 >                    spins += SPINS_PER_ARRIVAL;
896              }
897 <            else if ((unarrived = (int)s & UNARRIVED_MASK) == 0 &&
876 <                     (par = current.parent) != null) {
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              }
881            else if (unarrived != lastUnarrived) {
882                if ((lastUnarrived = unarrived) < NCPU)
883                    spins += SPINS_PER_ARRIVAL;
884            }
902              else if (spins > 0) {
903                  if (--spins == (SPINS_PER_ARRIVAL >>> 1))
904                      Thread.yield();  // yield midway through spin
# Line 889 | Line 906 | public class Phaser {
906              else if (node == null)   // must be noninterruptible
907                  node = new QNode(this, phase, false, false, 0L);
908              else if (node.isReleasable()) {
909 <                if ((int)(reconcileState() >>> PHASE_SHIFT) == phase)
909 >                if ((p = (int)(root.state >>> PHASE_SHIFT)) != phase)
910 >                    break;
911 >                else
912                      return phase;    // aborted
913              }
914 <            else if (!queued)
915 <                queued = tryEnqueue(phase, node);
914 >            else if (!queued) {      // push onto queue
915 >                AtomicReference<QNode> head = queueFor(phase);
916 >                QNode q = head.get();
917 >                if (q == null || q.phase == phase) {
918 >                    node.next = q;
919 >                    if ((p = (int)(root.state >>> PHASE_SHIFT)) != phase)
920 >                        break;       // recheck to avoid stale enqueue
921 >                    else
922 >                        queued = head.compareAndSet(q, node);
923 >                }
924 >            }
925              else {
926                  try {
927                      ForkJoinPool.managedBlock(node);
# Line 902 | Line 930 | public class Phaser {
930                  }
931              }
932          }
933 +        releaseWaiters(phase);
934 +        if (node != null)
935 +            node.onRelease();
936 +        return p;
937      }
938  
939      /**

Diff Legend

Removed lines
+ Added lines
< Changed lines
> Changed lines