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