1 /*
   2  * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
   3  *
   4  * This code is free software; you can redistribute it and/or modify it
   5  * under the terms of the GNU General Public License version 2 only, as
   6  * published by the Free Software Foundation.  Oracle designates this
   7  * particular file as subject to the "Classpath" exception as provided
   8  * by Oracle in the LICENSE file that accompanied this code.
   9  *
  10  * This code is distributed in the hope that it will be useful, but WITHOUT
  11  * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
  12  * FITNESS FOR A PARTICULAR PURPOSE.  See the GNU General Public License
  13  * version 2 for more details (a copy is included in the LICENSE file that
  14  * accompanied this code).
  15  *
  16  * You should have received a copy of the GNU General Public License version
  17  * 2 along with this work; if not, write to the Free Software Foundation,
  18  * Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA.
  19  *
  20  * Please contact Oracle, 500 Oracle Parkway, Redwood Shores, CA 94065 USA
  21  * or visit www.oracle.com if you need additional information or have any
  22  * questions.
  23  */
  24 
  25 /*
  26  * This file is available under and governed by the GNU General Public
  27  * License version 2 only, as published by the Free Software Foundation.
  28  * However, the following notice accompanied the original version of this
  29  * file:
  30  *
  31  * Written by Doug Lea with assistance from members of JCP JSR-166
  32  * Expert Group and released to the public domain, as explained at
  33  * http://creativecommons.org/publicdomain/zero/1.0/
  34  */
  35 
  36 package java.util.concurrent;
  37 
  38 import java.lang.invoke.MethodHandles;
  39 import java.lang.invoke.VarHandle;
  40 import java.util.concurrent.atomic.AtomicReference;
  41 import java.util.concurrent.locks.LockSupport;
  42 
  43 /**
  44  * A reusable synchronization barrier, similar in functionality to
  45  * {@link CyclicBarrier} and {@link CountDownLatch} but supporting
  46  * more flexible usage.
  47  *
  48  * <p><b>Registration.</b> Unlike the case for other barriers, the
  49  * number of parties <em>registered</em> to synchronize on a phaser
  50  * may vary over time.  Tasks may be registered at any time (using
  51  * methods {@link #register}, {@link #bulkRegister}, or forms of
  52  * constructors establishing initial numbers of parties), and
  53  * optionally deregistered upon any arrival (using {@link
  54  * #arriveAndDeregister}).  As is the case with most basic
  55  * synchronization constructs, registration and deregistration affect
  56  * only internal counts; they do not establish any further internal
  57  * bookkeeping, so tasks cannot query whether they are registered.
  58  * (However, you can introduce such bookkeeping by subclassing this
  59  * class.)
  60  *
  61  * <p><b>Synchronization.</b> Like a {@code CyclicBarrier}, a {@code
  62  * Phaser} may be repeatedly awaited.  Method {@link
  63  * #arriveAndAwaitAdvance} has effect analogous to {@link
  64  * java.util.concurrent.CyclicBarrier#await CyclicBarrier.await}. Each
  65  * generation of a phaser has an associated phase number. The phase
  66  * number starts at zero, and advances when all parties arrive at the
  67  * phaser, wrapping around to zero after reaching {@code
  68  * Integer.MAX_VALUE}. The use of phase numbers enables independent
  69  * control of actions upon arrival at a phaser and upon awaiting
  70  * others, via two kinds of methods that may be invoked by any
  71  * registered party:
  72  *
  73  * <ul>
  74  *
  75  *   <li><b>Arrival.</b> Methods {@link #arrive} and
  76  *       {@link #arriveAndDeregister} record arrival.  These methods
  77  *       do not block, but return an associated <em>arrival phase
  78  *       number</em>; that is, the phase number of the phaser to which
  79  *       the arrival applied. When the final party for a given phase
  80  *       arrives, an optional action is performed and the phase
  81  *       advances.  These actions are performed by the party
  82  *       triggering a phase advance, and are arranged by overriding
  83  *       method {@link #onAdvance(int, int)}, which also controls
  84  *       termination. Overriding this method is similar to, but more
  85  *       flexible than, providing a barrier action to a {@code
  86  *       CyclicBarrier}.
  87  *
  88  *   <li><b>Waiting.</b> Method {@link #awaitAdvance} requires an
  89  *       argument indicating an arrival phase number, and returns when
  90  *       the phaser advances to (or is already at) a different phase.
  91  *       Unlike similar constructions using {@code CyclicBarrier},
  92  *       method {@code awaitAdvance} continues to wait even if the
  93  *       waiting thread is interrupted. Interruptible and timeout
  94  *       versions are also available, but exceptions encountered while
  95  *       tasks wait interruptibly or with timeout do not change the
  96  *       state of the phaser. If necessary, you can perform any
  97  *       associated recovery within handlers of those exceptions,
  98  *       often after invoking {@code forceTermination}.  Phasers may
  99  *       also be used by tasks executing in a {@link ForkJoinPool}.
 100  *       Progress is ensured if the pool's parallelism level can
 101  *       accommodate the maximum number of simultaneously blocked
 102  *       parties.
 103  *
 104  * </ul>
 105  *
 106  * <p><b>Termination.</b> A phaser may enter a <em>termination</em>
 107  * state, that may be checked using method {@link #isTerminated}. Upon
 108  * termination, all synchronization methods immediately return without
 109  * waiting for advance, as indicated by a negative return value.
 110  * Similarly, attempts to register upon termination have no effect.
 111  * Termination is triggered when an invocation of {@code onAdvance}
 112  * returns {@code true}. The default implementation returns {@code
 113  * true} if a deregistration has caused the number of registered
 114  * parties to become zero.  As illustrated below, when phasers control
 115  * actions with a fixed number of iterations, it is often convenient
 116  * to override this method to cause termination when the current phase
 117  * number reaches a threshold. Method {@link #forceTermination} is
 118  * also available to abruptly release waiting threads and allow them
 119  * to terminate.
 120  *
 121  * <p><b>Tiering.</b> Phasers may be <em>tiered</em> (i.e.,
 122  * constructed in tree structures) to reduce contention. Phasers with
 123  * large numbers of parties that would otherwise experience heavy
 124  * synchronization contention costs may instead be set up so that
 125  * groups of sub-phasers share a common parent.  This may greatly
 126  * increase throughput even though it incurs greater per-operation
 127  * overhead.
 128  *
 129  * <p>In a tree of tiered phasers, registration and deregistration of
 130  * child phasers with their parent are managed automatically.
 131  * Whenever the number of registered parties of a child phaser becomes
 132  * non-zero (as established in the {@link #Phaser(Phaser,int)}
 133  * constructor, {@link #register}, or {@link #bulkRegister}), the
 134  * child phaser is registered with its parent.  Whenever the number of
 135  * registered parties becomes zero as the result of an invocation of
 136  * {@link #arriveAndDeregister}, the child phaser is deregistered
 137  * from its parent.
 138  *
 139  * <p><b>Monitoring.</b> While synchronization methods may be invoked
 140  * only by registered parties, the current state of a phaser may be
 141  * monitored by any caller.  At any given moment there are {@link
 142  * #getRegisteredParties} parties in total, of which {@link
 143  * #getArrivedParties} have arrived at the current phase ({@link
 144  * #getPhase}).  When the remaining ({@link #getUnarrivedParties})
 145  * parties arrive, the phase advances.  The values returned by these
 146  * methods may reflect transient states and so are not in general
 147  * useful for synchronization control.  Method {@link #toString}
 148  * returns snapshots of these state queries in a form convenient for
 149  * informal monitoring.
 150  *
 151  * <p>Memory consistency effects: Actions prior to any form of arrive
 152  * method <a href="package-summary.html#MemoryVisibility">
 153  * <i>happen-before</i></a> a corresponding phase advance and
 154  * onAdvance actions (if present), which in turn <i>happen-before</i>
 155  * actions following the phase advance.
 156  *
 157  * <p><b>Sample usages:</b>
 158  *
 159  * <p>A {@code Phaser} may be used instead of a {@code CountDownLatch}
 160  * to control a one-shot action serving a variable number of parties.
 161  * The typical idiom is for the method setting this up to first
 162  * register, then start all the actions, then deregister, as in:
 163  *
 164  * <pre> {@code
 165  * void runTasks(List<Runnable> tasks) {
 166  *   Phaser startingGate = new Phaser(1); // "1" to register self
 167  *   // create and start threads
 168  *   for (Runnable task : tasks) {
 169  *     startingGate.register();
 170  *     new Thread(() -> {
 171  *       startingGate.arriveAndAwaitAdvance();
 172  *       task.run();
 173  *     }).start();
 174  *   }
 175  *
 176  *   // deregister self to allow threads to proceed
 177  *   startingGate.arriveAndDeregister();
 178  * }}</pre>
 179  *
 180  * <p>One way to cause a set of threads to repeatedly perform actions
 181  * for a given number of iterations is to override {@code onAdvance}:
 182  *
 183  * <pre> {@code
 184  * void startTasks(List<Runnable> tasks, int iterations) {
 185  *   Phaser phaser = new Phaser() {
 186  *     protected boolean onAdvance(int phase, int registeredParties) {
 187  *       return phase >= iterations - 1 || registeredParties == 0;
 188  *     }
 189  *   };
 190  *   phaser.register();
 191  *   for (Runnable task : tasks) {
 192  *     phaser.register();
 193  *     new Thread(() -> {
 194  *       do {
 195  *         task.run();
 196  *         phaser.arriveAndAwaitAdvance();
 197  *       } while (!phaser.isTerminated());
 198  *     }).start();
 199  *   }
 200  *   // allow threads to proceed; don't wait for them
 201  *   phaser.arriveAndDeregister();
 202  * }}</pre>
 203  *
 204  * If the main task must later await termination, it
 205  * may re-register and then execute a similar loop:
 206  * <pre> {@code
 207  *   // ...
 208  *   phaser.register();
 209  *   while (!phaser.isTerminated())
 210  *     phaser.arriveAndAwaitAdvance();}</pre>
 211  *
 212  * <p>Related constructions may be used to await particular phase numbers
 213  * in contexts where you are sure that the phase will never wrap around
 214  * {@code Integer.MAX_VALUE}. For example:
 215  *
 216  * <pre> {@code
 217  * void awaitPhase(Phaser phaser, int phase) {
 218  *   int p = phaser.register(); // assumes caller not already registered
 219  *   while (p < phase) {
 220  *     if (phaser.isTerminated())
 221  *       // ... deal with unexpected termination
 222  *     else
 223  *       p = phaser.arriveAndAwaitAdvance();
 224  *   }
 225  *   phaser.arriveAndDeregister();
 226  * }}</pre>
 227  *
 228  * <p>To create a set of {@code n} tasks using a tree of phasers, you
 229  * could use code of the following form, assuming a Task class with a
 230  * constructor accepting a {@code Phaser} that it registers with upon
 231  * construction. After invocation of {@code build(new Task[n], 0, n,
 232  * new Phaser())}, these tasks could then be started, for example by
 233  * submitting to a pool:
 234  *
 235  * <pre> {@code
 236  * void build(Task[] tasks, int lo, int hi, Phaser ph) {
 237  *   if (hi - lo > TASKS_PER_PHASER) {
 238  *     for (int i = lo; i < hi; i += TASKS_PER_PHASER) {
 239  *       int j = Math.min(i + TASKS_PER_PHASER, hi);
 240  *       build(tasks, i, j, new Phaser(ph));
 241  *     }
 242  *   } else {
 243  *     for (int i = lo; i < hi; ++i)
 244  *       tasks[i] = new Task(ph);
 245  *       // assumes new Task(ph) performs ph.register()
 246  *   }
 247  * }}</pre>
 248  *
 249  * The best value of {@code TASKS_PER_PHASER} depends mainly on
 250  * expected synchronization rates. A value as low as four may
 251  * be appropriate for extremely small per-phase task bodies (thus
 252  * high rates), or up to hundreds for extremely large ones.
 253  *
 254  * <p><b>Implementation notes:</b> This implementation restricts the
 255  * maximum number of parties to 65535. Attempts to register additional
 256  * parties result in {@code IllegalStateException}. However, you can and
 257  * should create tiered phasers to accommodate arbitrarily large sets
 258  * of participants.
 259  *
 260  * @since 1.7
 261  * @author Doug Lea
 262  */
 263 public class Phaser {
 264     /*
 265      * This class implements an extension of X10 "clocks".  Thanks to
 266      * Vijay Saraswat for the idea, and to Vivek Sarkar for
 267      * enhancements to extend functionality.
 268      */
 269 
 270     /**
 271      * Primary state representation, holding four bit-fields:
 272      *
 273      * unarrived  -- the number of parties yet to hit barrier (bits  0-15)
 274      * parties    -- the number of parties to wait            (bits 16-31)
 275      * phase      -- the generation of the barrier            (bits 32-62)
 276      * terminated -- set if barrier is terminated             (bit  63 / sign)
 277      *
 278      * Except that a phaser with no registered parties is
 279      * distinguished by the otherwise illegal state of having zero
 280      * parties and one unarrived parties (encoded as EMPTY below).
 281      *
 282      * To efficiently maintain atomicity, these values are packed into
 283      * a single (atomic) long. Good performance relies on keeping
 284      * state decoding and encoding simple, and keeping race windows
 285      * short.
 286      *
 287      * All state updates are performed via CAS except initial
 288      * registration of a sub-phaser (i.e., one with a non-null
 289      * parent).  In this (relatively rare) case, we use built-in
 290      * synchronization to lock while first registering with its
 291      * parent.
 292      *
 293      * The phase of a subphaser is allowed to lag that of its
 294      * ancestors until it is actually accessed -- see method
 295      * reconcileState.
 296      */
 297     private volatile long state;
 298 
 299     private static final int  MAX_PARTIES     = 0xffff;
 300     private static final int  MAX_PHASE       = Integer.MAX_VALUE;
 301     private static final int  PARTIES_SHIFT   = 16;
 302     private static final int  PHASE_SHIFT     = 32;
 303     private static final int  UNARRIVED_MASK  = 0xffff;      // to mask ints
 304     private static final long PARTIES_MASK    = 0xffff0000L; // to mask longs
 305     private static final long COUNTS_MASK     = 0xffffffffL;
 306     private static final long TERMINATION_BIT = 1L << 63;
 307 
 308     // some special values
 309     private static final int  ONE_ARRIVAL     = 1;
 310     private static final int  ONE_PARTY       = 1 << PARTIES_SHIFT;
 311     private static final int  ONE_DEREGISTER  = ONE_ARRIVAL|ONE_PARTY;
 312     private static final int  EMPTY           = 1;
 313 
 314     // The following unpacking methods are usually manually inlined
 315 
 316     private static int unarrivedOf(long s) {
 317         int counts = (int)s;
 318         return (counts == EMPTY) ? 0 : (counts & UNARRIVED_MASK);
 319     }
 320 
 321     private static int partiesOf(long s) {
 322         return (int)s >>> PARTIES_SHIFT;
 323     }
 324 
 325     private static int phaseOf(long s) {
 326         return (int)(s >>> PHASE_SHIFT);
 327     }
 328 
 329     private static int arrivedOf(long s) {
 330         int counts = (int)s;
 331         return (counts == EMPTY) ? 0 :
 332             (counts >>> PARTIES_SHIFT) - (counts & UNARRIVED_MASK);
 333     }
 334 
 335     /**
 336      * The parent of this phaser, or null if none.
 337      */
 338     private final Phaser parent;
 339 
 340     /**
 341      * The root of phaser tree. Equals this if not in a tree.
 342      */
 343     private final Phaser root;
 344 
 345     /**
 346      * Heads of Treiber stacks for waiting threads. To eliminate
 347      * contention when releasing some threads while adding others, we
 348      * use two of them, alternating across even and odd phases.
 349      * Subphasers share queues with root to speed up releases.
 350      */
 351     private final AtomicReference<QNode> evenQ;
 352     private final AtomicReference<QNode> oddQ;
 353 
 354     /**
 355      * Returns message string for bounds exceptions on arrival.
 356      */
 357     private String badArrive(long s) {
 358         return "Attempted arrival of unregistered party for " +
 359             stateToString(s);
 360     }
 361 
 362     /**
 363      * Returns message string for bounds exceptions on registration.
 364      */
 365     private String badRegister(long s) {
 366         return "Attempt to register more than " +
 367             MAX_PARTIES + " parties for " + stateToString(s);
 368     }
 369 
 370     /**
 371      * Main implementation for methods arrive and arriveAndDeregister.
 372      * Manually tuned to speed up and minimize race windows for the
 373      * common case of just decrementing unarrived field.
 374      *
 375      * @param adjust value to subtract from state;
 376      *               ONE_ARRIVAL for arrive,
 377      *               ONE_DEREGISTER for arriveAndDeregister
 378      */
 379     private int doArrive(int adjust) {
 380         final Phaser root = this.root;
 381         for (;;) {
 382             long s = (root == this) ? state : reconcileState();
 383             int phase = (int)(s >>> PHASE_SHIFT);
 384             if (phase < 0)
 385                 return phase;
 386             int counts = (int)s;
 387             int unarrived = (counts == EMPTY) ? 0 : (counts & UNARRIVED_MASK);
 388             if (unarrived <= 0)
 389                 throw new IllegalStateException(badArrive(s));
 390             if (STATE.compareAndSet(this, s, s-=adjust)) {
 391                 if (unarrived == 1) {
 392                     long n = s & PARTIES_MASK;  // base of next state
 393                     int nextUnarrived = (int)n >>> PARTIES_SHIFT;
 394                     if (root == this) {
 395                         if (onAdvance(phase, nextUnarrived))
 396                             n |= TERMINATION_BIT;
 397                         else if (nextUnarrived == 0)
 398                             n |= EMPTY;
 399                         else
 400                             n |= nextUnarrived;
 401                         int nextPhase = (phase + 1) & MAX_PHASE;
 402                         n |= (long)nextPhase << PHASE_SHIFT;
 403                         STATE.compareAndSet(this, s, n);
 404                         releaseWaiters(phase);
 405                     }
 406                     else if (nextUnarrived == 0) { // propagate deregistration
 407                         phase = parent.doArrive(ONE_DEREGISTER);
 408                         STATE.compareAndSet(this, s, s | EMPTY);
 409                     }
 410                     else
 411                         phase = parent.doArrive(ONE_ARRIVAL);
 412                 }
 413                 return phase;
 414             }
 415         }
 416     }
 417 
 418     /**
 419      * Implementation of register, bulkRegister.
 420      *
 421      * @param registrations number to add to both parties and
 422      * unarrived fields. Must be greater than zero.
 423      */
 424     private int doRegister(int registrations) {
 425         // adjustment to state
 426         long adjust = ((long)registrations << PARTIES_SHIFT) | registrations;
 427         final Phaser parent = this.parent;
 428         int phase;
 429         for (;;) {
 430             long s = (parent == null) ? state : reconcileState();
 431             int counts = (int)s;
 432             int parties = counts >>> PARTIES_SHIFT;
 433             int unarrived = counts & UNARRIVED_MASK;
 434             if (registrations > MAX_PARTIES - parties)
 435                 throw new IllegalStateException(badRegister(s));
 436             phase = (int)(s >>> PHASE_SHIFT);
 437             if (phase < 0)
 438                 break;
 439             if (counts != EMPTY) {                  // not 1st registration
 440                 if (parent == null || reconcileState() == s) {
 441                     if (unarrived == 0)             // wait out advance
 442                         root.internalAwaitAdvance(phase, null);
 443                     else if (STATE.compareAndSet(this, s, s + adjust))
 444                         break;
 445                 }
 446             }
 447             else if (parent == null) {              // 1st root registration
 448                 long next = ((long)phase << PHASE_SHIFT) | adjust;
 449                 if (STATE.compareAndSet(this, s, next))
 450                     break;
 451             }
 452             else {
 453                 synchronized (this) {               // 1st sub registration
 454                     if (state == s) {               // recheck under lock
 455                         phase = parent.doRegister(1);
 456                         if (phase < 0)
 457                             break;
 458                         // finish registration whenever parent registration
 459                         // succeeded, even when racing with termination,
 460                         // since these are part of the same "transaction".
 461                         while (!STATE.weakCompareAndSet
 462                                (this, s,
 463                                 ((long)phase << PHASE_SHIFT) | adjust)) {
 464                             s = state;
 465                             phase = (int)(root.state >>> PHASE_SHIFT);
 466                             // assert (int)s == EMPTY;
 467                         }
 468                         break;
 469                     }
 470                 }
 471             }
 472         }
 473         return phase;
 474     }
 475 
 476     /**
 477      * Resolves lagged phase propagation from root if necessary.
 478      * Reconciliation normally occurs when root has advanced but
 479      * subphasers have not yet done so, in which case they must finish
 480      * their own advance by setting unarrived to parties (or if
 481      * parties is zero, resetting to unregistered EMPTY state).
 482      *
 483      * @return reconciled state
 484      */
 485     private long reconcileState() {
 486         final Phaser root = this.root;
 487         long s = state;
 488         if (root != this) {
 489             int phase, p;
 490             // CAS to root phase with current parties, tripping unarrived
 491             while ((phase = (int)(root.state >>> PHASE_SHIFT)) !=
 492                    (int)(s >>> PHASE_SHIFT) &&
 493                    !STATE.weakCompareAndSet
 494                    (this, s,
 495                     s = (((long)phase << PHASE_SHIFT) |
 496                          ((phase < 0) ? (s & COUNTS_MASK) :
 497                           (((p = (int)s >>> PARTIES_SHIFT) == 0) ? EMPTY :
 498                            ((s & PARTIES_MASK) | p))))))
 499                 s = state;
 500         }
 501         return s;
 502     }
 503 
 504     /**
 505      * Creates a new phaser with no initially registered parties, no
 506      * parent, and initial phase number 0. Any thread using this
 507      * phaser will need to first register for it.
 508      */
 509     public Phaser() {
 510         this(null, 0);
 511     }
 512 
 513     /**
 514      * Creates a new phaser with the given number of registered
 515      * unarrived parties, no parent, and initial phase number 0.
 516      *
 517      * @param parties the number of parties required to advance to the
 518      * next phase
 519      * @throws IllegalArgumentException if parties less than zero
 520      * or greater than the maximum number of parties supported
 521      */
 522     public Phaser(int parties) {
 523         this(null, parties);
 524     }
 525 
 526     /**
 527      * Equivalent to {@link #Phaser(Phaser, int) Phaser(parent, 0)}.
 528      *
 529      * @param parent the parent phaser
 530      */
 531     public Phaser(Phaser parent) {
 532         this(parent, 0);
 533     }
 534 
 535     /**
 536      * Creates a new phaser with the given parent and number of
 537      * registered unarrived parties.  When the given parent is non-null
 538      * and the given number of parties is greater than zero, this
 539      * child phaser is registered with its parent.
 540      *
 541      * @param parent the parent phaser
 542      * @param parties the number of parties required to advance to the
 543      * next phase
 544      * @throws IllegalArgumentException if parties less than zero
 545      * or greater than the maximum number of parties supported
 546      */
 547     public Phaser(Phaser parent, int parties) {
 548         if (parties >>> PARTIES_SHIFT != 0)
 549             throw new IllegalArgumentException("Illegal number of parties");
 550         int phase = 0;
 551         this.parent = parent;
 552         if (parent != null) {
 553             final Phaser root = parent.root;
 554             this.root = root;
 555             this.evenQ = root.evenQ;
 556             this.oddQ = root.oddQ;
 557             if (parties != 0)
 558                 phase = parent.doRegister(1);
 559         }
 560         else {
 561             this.root = this;
 562             this.evenQ = new AtomicReference<QNode>();
 563             this.oddQ = new AtomicReference<QNode>();
 564         }
 565         this.state = (parties == 0) ? (long)EMPTY :
 566             ((long)phase << PHASE_SHIFT) |
 567             ((long)parties << PARTIES_SHIFT) |
 568             ((long)parties);
 569     }
 570 
 571     /**
 572      * Adds a new unarrived party to this phaser.  If an ongoing
 573      * invocation of {@link #onAdvance} is in progress, this method
 574      * may await its completion before returning.  If this phaser has
 575      * a parent, and this phaser previously had no registered parties,
 576      * this child phaser is also registered with its parent. If
 577      * this phaser is terminated, the attempt to register has
 578      * no effect, and a negative value is returned.
 579      *
 580      * @return the arrival phase number to which this registration
 581      * applied.  If this value is negative, then this phaser has
 582      * terminated, in which case registration has no effect.
 583      * @throws IllegalStateException if attempting to register more
 584      * than the maximum supported number of parties
 585      */
 586     public int register() {
 587         return doRegister(1);
 588     }
 589 
 590     /**
 591      * Adds the given number of new unarrived parties to this phaser.
 592      * If an ongoing invocation of {@link #onAdvance} is in progress,
 593      * this method may await its completion before returning.  If this
 594      * phaser has a parent, and the given number of parties is greater
 595      * than zero, and this phaser previously had no registered
 596      * parties, this child phaser is also registered with its parent.
 597      * If this phaser is terminated, the attempt to register has no
 598      * effect, and a negative value is returned.
 599      *
 600      * @param parties the number of additional parties required to
 601      * advance to the next phase
 602      * @return the arrival phase number to which this registration
 603      * applied.  If this value is negative, then this phaser has
 604      * terminated, in which case registration has no effect.
 605      * @throws IllegalStateException if attempting to register more
 606      * than the maximum supported number of parties
 607      * @throws IllegalArgumentException if {@code parties < 0}
 608      */
 609     public int bulkRegister(int parties) {
 610         if (parties < 0)
 611             throw new IllegalArgumentException();
 612         if (parties == 0)
 613             return getPhase();
 614         return doRegister(parties);
 615     }
 616 
 617     /**
 618      * Arrives at this phaser, without waiting for others to arrive.
 619      *
 620      * <p>It is a usage error for an unregistered party to invoke this
 621      * method.  However, this error may result in an {@code
 622      * IllegalStateException} only upon some subsequent operation on
 623      * this phaser, if ever.
 624      *
 625      * @return the arrival phase number, or a negative value if terminated
 626      * @throws IllegalStateException if not terminated and the number
 627      * of unarrived parties would become negative
 628      */
 629     public int arrive() {
 630         return doArrive(ONE_ARRIVAL);
 631     }
 632 
 633     /**
 634      * Arrives at this phaser and deregisters from it without waiting
 635      * for others to arrive. Deregistration reduces the number of
 636      * parties required to advance in future phases.  If this phaser
 637      * has a parent, and deregistration causes this phaser to have
 638      * zero parties, this phaser is also deregistered from its parent.
 639      *
 640      * <p>It is a usage error for an unregistered party to invoke this
 641      * method.  However, this error may result in an {@code
 642      * IllegalStateException} only upon some subsequent operation on
 643      * this phaser, if ever.
 644      *
 645      * @return the arrival phase number, or a negative value if terminated
 646      * @throws IllegalStateException if not terminated and the number
 647      * of registered or unarrived parties would become negative
 648      */
 649     public int arriveAndDeregister() {
 650         return doArrive(ONE_DEREGISTER);
 651     }
 652 
 653     /**
 654      * Arrives at this phaser and awaits others. Equivalent in effect
 655      * to {@code awaitAdvance(arrive())}.  If you need to await with
 656      * interruption or timeout, you can arrange this with an analogous
 657      * construction using one of the other forms of the {@code
 658      * awaitAdvance} method.  If instead you need to deregister upon
 659      * arrival, use {@code awaitAdvance(arriveAndDeregister())}.
 660      *
 661      * <p>It is a usage error for an unregistered party to invoke this
 662      * method.  However, this error may result in an {@code
 663      * IllegalStateException} only upon some subsequent operation on
 664      * this phaser, if ever.
 665      *
 666      * @return the arrival phase number, or the (negative)
 667      * {@linkplain #getPhase() current phase} if terminated
 668      * @throws IllegalStateException if not terminated and the number
 669      * of unarrived parties would become negative
 670      */
 671     public int arriveAndAwaitAdvance() {
 672         // Specialization of doArrive+awaitAdvance eliminating some reads/paths
 673         final Phaser root = this.root;
 674         for (;;) {
 675             long s = (root == this) ? state : reconcileState();
 676             int phase = (int)(s >>> PHASE_SHIFT);
 677             if (phase < 0)
 678                 return phase;
 679             int counts = (int)s;
 680             int unarrived = (counts == EMPTY) ? 0 : (counts & UNARRIVED_MASK);
 681             if (unarrived <= 0)
 682                 throw new IllegalStateException(badArrive(s));
 683             if (STATE.compareAndSet(this, s, s -= ONE_ARRIVAL)) {
 684                 if (unarrived > 1)
 685                     return root.internalAwaitAdvance(phase, null);
 686                 if (root != this)
 687                     return parent.arriveAndAwaitAdvance();
 688                 long n = s & PARTIES_MASK;  // base of next state
 689                 int nextUnarrived = (int)n >>> PARTIES_SHIFT;
 690                 if (onAdvance(phase, nextUnarrived))
 691                     n |= TERMINATION_BIT;
 692                 else if (nextUnarrived == 0)
 693                     n |= EMPTY;
 694                 else
 695                     n |= nextUnarrived;
 696                 int nextPhase = (phase + 1) & MAX_PHASE;
 697                 n |= (long)nextPhase << PHASE_SHIFT;
 698                 if (!STATE.compareAndSet(this, s, n))
 699                     return (int)(state >>> PHASE_SHIFT); // terminated
 700                 releaseWaiters(phase);
 701                 return nextPhase;
 702             }
 703         }
 704     }
 705 
 706     /**
 707      * Awaits the phase of this phaser to advance from the given phase
 708      * value, returning immediately if the current phase is not equal
 709      * to the given phase value or this phaser is terminated.
 710      *
 711      * @param phase an arrival phase number, or negative value if
 712      * terminated; this argument is normally the value returned by a
 713      * previous call to {@code arrive} or {@code arriveAndDeregister}.
 714      * @return the next arrival phase number, or the argument if it is
 715      * negative, or the (negative) {@linkplain #getPhase() current phase}
 716      * if terminated
 717      */
 718     public int awaitAdvance(int phase) {
 719         final Phaser root = this.root;
 720         long s = (root == this) ? state : reconcileState();
 721         int p = (int)(s >>> PHASE_SHIFT);
 722         if (phase < 0)
 723             return phase;
 724         if (p == phase)
 725             return root.internalAwaitAdvance(phase, null);
 726         return p;
 727     }
 728 
 729     /**
 730      * Awaits the phase of this phaser to advance from the given phase
 731      * value, throwing {@code InterruptedException} if interrupted
 732      * while waiting, or returning immediately if the current phase is
 733      * not equal to the given phase value or this phaser is
 734      * terminated.
 735      *
 736      * @param phase an arrival phase number, or negative value if
 737      * terminated; this argument is normally the value returned by a
 738      * previous call to {@code arrive} or {@code arriveAndDeregister}.
 739      * @return the next arrival phase number, or the argument if it is
 740      * negative, or the (negative) {@linkplain #getPhase() current phase}
 741      * if terminated
 742      * @throws InterruptedException if thread interrupted while waiting
 743      */
 744     public int awaitAdvanceInterruptibly(int phase)
 745         throws InterruptedException {
 746         final Phaser root = this.root;
 747         long s = (root == this) ? state : reconcileState();
 748         int p = (int)(s >>> PHASE_SHIFT);
 749         if (phase < 0)
 750             return phase;
 751         if (p == phase) {
 752             QNode node = new QNode(this, phase, true, false, 0L);
 753             p = root.internalAwaitAdvance(phase, node);
 754             if (node.wasInterrupted)
 755                 throw new InterruptedException();
 756         }
 757         return p;
 758     }
 759 
 760     /**
 761      * Awaits the phase of this phaser to advance from the given phase
 762      * value or the given timeout to elapse, throwing {@code
 763      * InterruptedException} if interrupted while waiting, or
 764      * returning immediately if the current phase is not equal to the
 765      * given phase value or this phaser is terminated.
 766      *
 767      * @param phase an arrival phase number, or negative value if
 768      * terminated; this argument is normally the value returned by a
 769      * previous call to {@code arrive} or {@code arriveAndDeregister}.
 770      * @param timeout how long to wait before giving up, in units of
 771      *        {@code unit}
 772      * @param unit a {@code TimeUnit} determining how to interpret the
 773      *        {@code timeout} parameter
 774      * @return the next arrival phase number, or the argument if it is
 775      * negative, or the (negative) {@linkplain #getPhase() current phase}
 776      * if terminated
 777      * @throws InterruptedException if thread interrupted while waiting
 778      * @throws TimeoutException if timed out while waiting
 779      */
 780     public int awaitAdvanceInterruptibly(int phase,
 781                                          long timeout, TimeUnit unit)
 782         throws InterruptedException, TimeoutException {
 783         long nanos = unit.toNanos(timeout);
 784         final Phaser root = this.root;
 785         long s = (root == this) ? state : reconcileState();
 786         int p = (int)(s >>> PHASE_SHIFT);
 787         if (phase < 0)
 788             return phase;
 789         if (p == phase) {
 790             QNode node = new QNode(this, phase, true, true, nanos);
 791             p = root.internalAwaitAdvance(phase, node);
 792             if (node.wasInterrupted)
 793                 throw new InterruptedException();
 794             else if (p == phase)
 795                 throw new TimeoutException();
 796         }
 797         return p;
 798     }
 799 
 800     /**
 801      * Forces this phaser to enter termination state.  Counts of
 802      * registered parties are unaffected.  If this phaser is a member
 803      * of a tiered set of phasers, then all of the phasers in the set
 804      * are terminated.  If this phaser is already terminated, this
 805      * method has no effect.  This method may be useful for
 806      * coordinating recovery after one or more tasks encounter
 807      * unexpected exceptions.
 808      */
 809     public void forceTermination() {
 810         // Only need to change root state
 811         final Phaser root = this.root;
 812         long s;
 813         while ((s = root.state) >= 0) {
 814             if (STATE.compareAndSet(root, s, s | TERMINATION_BIT)) {
 815                 // signal all threads
 816                 releaseWaiters(0); // Waiters on evenQ
 817                 releaseWaiters(1); // Waiters on oddQ
 818                 return;
 819             }
 820         }
 821     }
 822 
 823     /**
 824      * Returns the current phase number. The maximum phase number is
 825      * {@code Integer.MAX_VALUE}, after which it restarts at
 826      * zero. Upon termination, the phase number is negative,
 827      * in which case the prevailing phase prior to termination
 828      * may be obtained via {@code getPhase() + Integer.MIN_VALUE}.
 829      *
 830      * @return the phase number, or a negative value if terminated
 831      */
 832     public final int getPhase() {
 833         return (int)(root.state >>> PHASE_SHIFT);
 834     }
 835 
 836     /**
 837      * Returns the number of parties registered at this phaser.
 838      *
 839      * @return the number of parties
 840      */
 841     public int getRegisteredParties() {
 842         return partiesOf(state);
 843     }
 844 
 845     /**
 846      * Returns the number of registered parties that have arrived at
 847      * the current phase of this phaser. If this phaser has terminated,
 848      * the returned value is meaningless and arbitrary.
 849      *
 850      * @return the number of arrived parties
 851      */
 852     public int getArrivedParties() {
 853         return arrivedOf(reconcileState());
 854     }
 855 
 856     /**
 857      * Returns the number of registered parties that have not yet
 858      * arrived at the current phase of this phaser. If this phaser has
 859      * terminated, the returned value is meaningless and arbitrary.
 860      *
 861      * @return the number of unarrived parties
 862      */
 863     public int getUnarrivedParties() {
 864         return unarrivedOf(reconcileState());
 865     }
 866 
 867     /**
 868      * Returns the parent of this phaser, or {@code null} if none.
 869      *
 870      * @return the parent of this phaser, or {@code null} if none
 871      */
 872     public Phaser getParent() {
 873         return parent;
 874     }
 875 
 876     /**
 877      * Returns the root ancestor of this phaser, which is the same as
 878      * this phaser if it has no parent.
 879      *
 880      * @return the root ancestor of this phaser
 881      */
 882     public Phaser getRoot() {
 883         return root;
 884     }
 885 
 886     /**
 887      * Returns {@code true} if this phaser has been terminated.
 888      *
 889      * @return {@code true} if this phaser has been terminated
 890      */
 891     public boolean isTerminated() {
 892         return root.state < 0L;
 893     }
 894 
 895     /**
 896      * Overridable method to perform an action upon impending phase
 897      * advance, and to control termination. This method is invoked
 898      * upon arrival of the party advancing this phaser (when all other
 899      * waiting parties are dormant).  If this method returns {@code
 900      * true}, this phaser will be set to a final termination state
 901      * upon advance, and subsequent calls to {@link #isTerminated}
 902      * will return true. Any (unchecked) Exception or Error thrown by
 903      * an invocation of this method is propagated to the party
 904      * attempting to advance this phaser, in which case no advance
 905      * occurs.
 906      *
 907      * <p>The arguments to this method provide the state of the phaser
 908      * prevailing for the current transition.  The effects of invoking
 909      * arrival, registration, and waiting methods on this phaser from
 910      * within {@code onAdvance} are unspecified and should not be
 911      * relied on.
 912      *
 913      * <p>If this phaser is a member of a tiered set of phasers, then
 914      * {@code onAdvance} is invoked only for its root phaser on each
 915      * advance.
 916      *
 917      * <p>To support the most common use cases, the default
 918      * implementation of this method returns {@code true} when the
 919      * number of registered parties has become zero as the result of a
 920      * party invoking {@code arriveAndDeregister}.  You can disable
 921      * this behavior, thus enabling continuation upon future
 922      * registrations, by overriding this method to always return
 923      * {@code false}:
 924      *
 925      * <pre> {@code
 926      * Phaser phaser = new Phaser() {
 927      *   protected boolean onAdvance(int phase, int parties) { return false; }
 928      * };}</pre>
 929      *
 930      * @param phase the current phase number on entry to this method,
 931      * before this phaser is advanced
 932      * @param registeredParties the current number of registered parties
 933      * @return {@code true} if this phaser should terminate
 934      */
 935     protected boolean onAdvance(int phase, int registeredParties) {
 936         return registeredParties == 0;
 937     }
 938 
 939     /**
 940      * Returns a string identifying this phaser, as well as its
 941      * state.  The state, in brackets, includes the String {@code
 942      * "phase = "} followed by the phase number, {@code "parties = "}
 943      * followed by the number of registered parties, and {@code
 944      * "arrived = "} followed by the number of arrived parties.
 945      *
 946      * @return a string identifying this phaser, as well as its state
 947      */
 948     public String toString() {
 949         return stateToString(reconcileState());
 950     }
 951 
 952     /**
 953      * Implementation of toString and string-based error messages.
 954      */
 955     private String stateToString(long s) {
 956         return super.toString() +
 957             "[phase = " + phaseOf(s) +
 958             " parties = " + partiesOf(s) +
 959             " arrived = " + arrivedOf(s) + "]";
 960     }
 961 
 962     // Waiting mechanics
 963 
 964     /**
 965      * Removes and signals threads from queue for phase.
 966      */
 967     private void releaseWaiters(int phase) {
 968         QNode q;   // first element of queue
 969         Thread t;  // its thread
 970         AtomicReference<QNode> head = (phase & 1) == 0 ? evenQ : oddQ;
 971         while ((q = head.get()) != null &&
 972                q.phase != (int)(root.state >>> PHASE_SHIFT)) {
 973             if (head.compareAndSet(q, q.next) &&
 974                 (t = q.thread) != null) {
 975                 q.thread = null;
 976                 LockSupport.unpark(t);
 977             }
 978         }
 979     }
 980 
 981     /**
 982      * Variant of releaseWaiters that additionally tries to remove any
 983      * nodes no longer waiting for advance due to timeout or
 984      * interrupt. Currently, nodes are removed only if they are at
 985      * head of queue, which suffices to reduce memory footprint in
 986      * most usages.
 987      *
 988      * @return current phase on exit
 989      */
 990     private int abortWait(int phase) {
 991         AtomicReference<QNode> head = (phase & 1) == 0 ? evenQ : oddQ;
 992         for (;;) {
 993             Thread t;
 994             QNode q = head.get();
 995             int p = (int)(root.state >>> PHASE_SHIFT);
 996             if (q == null || ((t = q.thread) != null && q.phase == p))
 997                 return p;
 998             if (head.compareAndSet(q, q.next) && t != null) {
 999                 q.thread = null;
1000                 LockSupport.unpark(t);
1001             }
1002         }
1003     }
1004 
1005     /** The number of CPUs, for spin control */
1006     private static final int NCPU = Runtime.getRuntime().availableProcessors();
1007 
1008     /**
1009      * The number of times to spin before blocking while waiting for
1010      * advance, per arrival while waiting. On multiprocessors, fully
1011      * blocking and waking up a large number of threads all at once is
1012      * usually a very slow process, so we use rechargeable spins to
1013      * avoid it when threads regularly arrive: When a thread in
1014      * internalAwaitAdvance notices another arrival before blocking,
1015      * and there appear to be enough CPUs available, it spins
1016      * SPINS_PER_ARRIVAL more times before blocking. The value trades
1017      * off good-citizenship vs big unnecessary slowdowns.
1018      */
1019     static final int SPINS_PER_ARRIVAL = (NCPU < 2) ? 1 : 1 << 8;
1020 
1021     /**
1022      * Possibly blocks and waits for phase to advance unless aborted.
1023      * Call only on root phaser.
1024      *
1025      * @param phase current phase
1026      * @param node if non-null, the wait node to track interrupt and timeout;
1027      * if null, denotes noninterruptible wait
1028      * @return current phase
1029      */
1030     private int internalAwaitAdvance(int phase, QNode node) {
1031         // assert root == this;
1032         releaseWaiters(phase-1);          // ensure old queue clean
1033         boolean queued = false;           // true when node is enqueued
1034         int lastUnarrived = 0;            // to increase spins upon change
1035         int spins = SPINS_PER_ARRIVAL;
1036         long s;
1037         int p;
1038         while ((p = (int)((s = state) >>> PHASE_SHIFT)) == phase) {
1039             if (node == null) {           // spinning in noninterruptible mode
1040                 int unarrived = (int)s & UNARRIVED_MASK;
1041                 if (unarrived != lastUnarrived &&
1042                     (lastUnarrived = unarrived) < NCPU)
1043                     spins += SPINS_PER_ARRIVAL;
1044                 boolean interrupted = Thread.interrupted();
1045                 if (interrupted || --spins < 0) { // need node to record intr
1046                     node = new QNode(this, phase, false, false, 0L);
1047                     node.wasInterrupted = interrupted;
1048                 }
1049                 else
1050                     Thread.onSpinWait();
1051             }
1052             else if (node.isReleasable()) // done or aborted
1053                 break;
1054             else if (!queued) {           // push onto queue
1055                 AtomicReference<QNode> head = (phase & 1) == 0 ? evenQ : oddQ;
1056                 QNode q = node.next = head.get();
1057                 if ((q == null || q.phase == phase) &&
1058                     (int)(state >>> PHASE_SHIFT) == phase) // avoid stale enq
1059                     queued = head.compareAndSet(q, node);
1060             }
1061             else {
1062                 try {
1063                     ForkJoinPool.managedBlock(node);
1064                 } catch (InterruptedException cantHappen) {
1065                     node.wasInterrupted = true;
1066                 }
1067             }
1068         }
1069 
1070         if (node != null) {
1071             if (node.thread != null)
1072                 node.thread = null;       // avoid need for unpark()
1073             if (node.wasInterrupted && !node.interruptible)
1074                 Thread.currentThread().interrupt();
1075             if (p == phase && (p = (int)(state >>> PHASE_SHIFT)) == phase)
1076                 return abortWait(phase); // possibly clean up on abort
1077         }
1078         releaseWaiters(phase);
1079         return p;
1080     }
1081 
1082     /**
1083      * Wait nodes for Treiber stack representing wait queue.
1084      */
1085     static final class QNode implements ForkJoinPool.ManagedBlocker {
1086         final Phaser phaser;
1087         final int phase;
1088         final boolean interruptible;
1089         final boolean timed;
1090         boolean wasInterrupted;
1091         long nanos;
1092         final long deadline;
1093         volatile Thread thread; // nulled to cancel wait
1094         QNode next;
1095 
1096         QNode(Phaser phaser, int phase, boolean interruptible,
1097               boolean timed, long nanos) {
1098             this.phaser = phaser;
1099             this.phase = phase;
1100             this.interruptible = interruptible;
1101             this.nanos = nanos;
1102             this.timed = timed;
1103             this.deadline = timed ? System.nanoTime() + nanos : 0L;
1104             thread = Thread.currentThread();
1105         }
1106 
1107         public boolean isReleasable() {
1108             if (thread == null)
1109                 return true;
1110             if (phaser.getPhase() != phase) {
1111                 thread = null;
1112                 return true;
1113             }
1114             if (Thread.interrupted())
1115                 wasInterrupted = true;
1116             if (wasInterrupted && interruptible) {
1117                 thread = null;
1118                 return true;
1119             }
1120             if (timed &&
1121                 (nanos <= 0L || (nanos = deadline - System.nanoTime()) <= 0L)) {
1122                 thread = null;
1123                 return true;
1124             }
1125             return false;
1126         }
1127 
1128         public boolean block() {
1129             while (!isReleasable()) {
1130                 if (timed)
1131                     LockSupport.parkNanos(this, nanos);
1132                 else
1133                     LockSupport.park(this);
1134             }
1135             return true;
1136         }
1137     }
1138 
1139     // VarHandle mechanics
1140     private static final VarHandle STATE;
1141     static {
1142         try {
1143             MethodHandles.Lookup l = MethodHandles.lookup();
1144             STATE = l.findVarHandle(Phaser.class, "state", long.class);
1145         } catch (ReflectiveOperationException e) {
1146             throw new ExceptionInInitializerError(e);
1147         }
1148 
1149         // Reduce the risk of rare disastrous classloading in first call to
1150         // LockSupport.park: https://bugs.openjdk.java.net/browse/JDK-8074773
1151         Class<?> ensureLoaded = LockSupport.class;
1152     }
1153 }