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 unarrivedOf(long s)316 private static int unarrivedOf(long s) { 317 int counts = (int)s; 318 return (counts == EMPTY) ? 0 : (counts & UNARRIVED_MASK); 319 } 320 partiesOf(long s)321 private static int partiesOf(long s) { 322 return (int)s >>> PARTIES_SHIFT; 323 } 324 phaseOf(long s)325 private static int phaseOf(long s) { 326 return (int)(s >>> PHASE_SHIFT); 327 } 328 arrivedOf(long s)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 */ badArrive(long s)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 */ badRegister(long s)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 */ doArrive(int adjust)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 */ doRegister(int registrations)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 */ reconcileState()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 */ Phaser()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 */ Phaser(int parties)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 */ Phaser(Phaser parent)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 */ Phaser(Phaser parent, int parties)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 */ register()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 */ bulkRegister(int parties)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 */ arrive()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 */ arriveAndDeregister()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 */ arriveAndAwaitAdvance()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 */ awaitAdvance(int phase)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 */ awaitAdvanceInterruptibly(int phase)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 */ awaitAdvanceInterruptibly(int phase, long timeout, TimeUnit unit)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 */ forceTermination()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 */ getPhase()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 */ getRegisteredParties()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 */ getArrivedParties()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 */ getUnarrivedParties()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 */ getParent()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 */ getRoot()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 */ isTerminated()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 */ onAdvance(int phase, int registeredParties)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 */ toString()948 public String toString() { 949 return stateToString(reconcileState()); 950 } 951 952 /** 953 * Implementation of toString and string-based error messages. 954 */ stateToString(long s)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 */ releaseWaiters(int phase)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 */ abortWait(int phase)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 */ internalAwaitAdvance(int phase, QNode node)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 QNode(Phaser phaser, int phase, boolean interruptible, boolean timed, long nanos)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 isReleasable()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 block()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 } 1154