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