Categories:
Audio (13)
Biotech (29)
Bytecode (36)
Database (77)
Framework (7)
Game (7)
General (507)
Graphics (53)
I/O (35)
IDE (2)
JAR Tools (101)
JavaBeans (21)
JDBC (121)
JDK (426)
JSP (20)
Logging (108)
Mail (58)
Messaging (8)
Network (84)
PDF (97)
Report (7)
Scripting (84)
Security (32)
Server (121)
Servlet (26)
SOAP (24)
Testing (54)
Web (15)
XML (309)
Collections:
Other Resources:
JDK 11 java.base.jmod - Base Module
JDK 11 java.base.jmod is the JMOD file for JDK 11 Base module.
JDK 11 Base module compiled class files are stored in \fyicenter\jdk-11.0.1\jmods\java.base.jmod.
JDK 11 Base module compiled class files are also linked and stored in the \fyicenter\jdk-11.0.1\lib\modules JImage file.
JDK 11 Base module source code files are stored in \fyicenter\jdk-11.0.1\lib\src.zip\java.base.
You can click and view the content of each source code file in the list below.
✍: FYIcenter
⏎ java/util/concurrent/Phaser.java
/* * ORACLE PROPRIETARY/CONFIDENTIAL. Use is subject to license terms. * * * * * * * * * * * * * * * * * * * * */ /* * * * * * * Written by Doug Lea with assistance from members of JCP JSR-166 * Expert Group and released to the public domain, as explained at * http://creativecommons.org/publicdomain/zero/1.0/ */ package java.util.concurrent; import java.lang.invoke.MethodHandles; import java.lang.invoke.VarHandle; import java.util.concurrent.atomic.AtomicReference; import java.util.concurrent.locks.LockSupport; /** * A reusable synchronization barrier, similar in functionality to * {@link CyclicBarrier} and {@link CountDownLatch} but supporting * more flexible usage. * * <p><b>Registration.</b> Unlike the case for other barriers, the * number of parties <em>registered</em> to synchronize on a phaser * may vary over time. Tasks may be registered at any time (using * methods {@link #register}, {@link #bulkRegister}, or forms of * constructors establishing initial numbers of parties), and * optionally deregistered upon any arrival (using {@link * #arriveAndDeregister}). As is the case with most basic * synchronization constructs, registration and deregistration affect * only internal counts; they do not establish any further internal * bookkeeping, so tasks cannot query whether they are registered. * (However, you can introduce such bookkeeping by subclassing this * class.) * * <p><b>Synchronization.</b> Like a {@code CyclicBarrier}, a {@code * Phaser} may be repeatedly awaited. Method {@link * #arriveAndAwaitAdvance} has effect analogous to {@link * java.util.concurrent.CyclicBarrier#await CyclicBarrier.await}. Each * generation of a phaser has an associated phase number. The phase * number starts at zero, and advances when all parties arrive at the * phaser, wrapping around to zero after reaching {@code * Integer.MAX_VALUE}. The use of phase numbers enables independent * control of actions upon arrival at a phaser and upon awaiting * others, via two kinds of methods that may be invoked by any * registered party: * * <ul> * * <li><b>Arrival.</b> Methods {@link #arrive} and * {@link #arriveAndDeregister} record arrival. These methods * do not block, but return an associated <em>arrival phase * number</em>; that is, the phase number of the phaser to which * the arrival applied. When the final party for a given phase * arrives, an optional action is performed and the phase * advances. These actions are performed by the party * triggering a phase advance, and are arranged by overriding * method {@link #onAdvance(int, int)}, which also controls * termination. Overriding this method is similar to, but more * flexible than, providing a barrier action to a {@code * CyclicBarrier}. * * <li><b>Waiting.</b> Method {@link #awaitAdvance} requires an * argument indicating an arrival phase number, and returns when * the phaser advances to (or is already at) a different phase. * Unlike similar constructions using {@code CyclicBarrier}, * method {@code awaitAdvance} continues to wait even if the * waiting thread is interrupted. Interruptible and timeout * versions are also available, but exceptions encountered while * tasks wait interruptibly or with timeout do not change the * state of the phaser. If necessary, you can perform any * associated recovery within handlers of those exceptions, * often after invoking {@code forceTermination}. Phasers may * also be used by tasks executing in a {@link ForkJoinPool}. * Progress is ensured if the pool's parallelismLevel can * accommodate the maximum number of simultaneously blocked * parties. * * </ul> * * <p><b>Termination.</b> A phaser may enter a <em>termination</em> * state, that may be checked using method {@link #isTerminated}. Upon * termination, all synchronization methods immediately return without * waiting for advance, as indicated by a negative return value. * Similarly, attempts to register upon termination have no effect. * Termination is triggered when an invocation of {@code onAdvance} * returns {@code true}. The default implementation returns {@code * true} if a deregistration has caused the number of registered * parties to become zero. As illustrated below, when phasers control * actions with a fixed number of iterations, it is often convenient * to override this method to cause termination when the current phase * number reaches a threshold. Method {@link #forceTermination} is * also available to abruptly release waiting threads and allow them * to terminate. * * <p><b>Tiering.</b> Phasers may be <em>tiered</em> (i.e., * constructed in tree structures) to reduce contention. Phasers with * large numbers of parties that would otherwise experience heavy * synchronization contention costs may instead be set up so that * groups of sub-phasers share a common parent. This may greatly * increase throughput even though it incurs greater per-operation * overhead. * * <p>In a tree of tiered phasers, registration and deregistration of * child phasers with their parent are managed automatically. * Whenever the number of registered parties of a child phaser becomes * non-zero (as established in the {@link #Phaser(Phaser,int)} * constructor, {@link #register}, or {@link #bulkRegister}), the * child phaser is registered with its parent. Whenever the number of * registered parties becomes zero as the result of an invocation of * {@link #arriveAndDeregister}, the child phaser is deregistered * from its parent. * * <p><b>Monitoring.</b> While synchronization methods may be invoked * only by registered parties, the current state of a phaser may be * monitored by any caller. At any given moment there are {@link * #getRegisteredParties} parties in total, of which {@link * #getArrivedParties} have arrived at the current phase ({@link * #getPhase}). When the remaining ({@link #getUnarrivedParties}) * parties arrive, the phase advances. The values returned by these * methods may reflect transient states and so are not in general * useful for synchronization control. Method {@link #toString} * returns snapshots of these state queries in a form convenient for * informal monitoring. * * <p><b>Sample usages:</b> * * <p>A {@code Phaser} may be used instead of a {@code CountDownLatch} * to control a one-shot action serving a variable number of parties. * The typical idiom is for the method setting this up to first * register, then start all the actions, then deregister, as in: * * <pre> {@code * void runTasks(List<Runnable> tasks) { * Phaser startingGate = new Phaser(1); // "1" to register self * // create and start threads * for (Runnable task : tasks) { * startingGate.register(); * new Thread(() -> { * startingGate.arriveAndAwaitAdvance(); * task.run(); * }).start(); * } * * // deregister self to allow threads to proceed * startingGate.arriveAndDeregister(); * }}</pre> * * <p>One way to cause a set of threads to repeatedly perform actions * for a given number of iterations is to override {@code onAdvance}: * * <pre> {@code * void startTasks(List<Runnable> tasks, int iterations) { * Phaser phaser = new Phaser() { * protected boolean onAdvance(int phase, int registeredParties) { * return phase >= iterations - 1 || registeredParties == 0; * } * }; * phaser.register(); * for (Runnable task : tasks) { * phaser.register(); * new Thread(() -> { * do { * task.run(); * phaser.arriveAndAwaitAdvance(); * } while (!phaser.isTerminated()); * }).start(); * } * // allow threads to proceed; don't wait for them * phaser.arriveAndDeregister(); * }}</pre> * * If the main task must later await termination, it * may re-register and then execute a similar loop: * <pre> {@code * // ... * phaser.register(); * while (!phaser.isTerminated()) * phaser.arriveAndAwaitAdvance();}</pre> * * <p>Related constructions may be used to await particular phase numbers * in contexts where you are sure that the phase will never wrap around * {@code Integer.MAX_VALUE}. For example: * * <pre> {@code * void awaitPhase(Phaser phaser, int phase) { * int p = phaser.register(); // assumes caller not already registered * while (p < phase) { * if (phaser.isTerminated()) * // ... deal with unexpected termination * else * p = phaser.arriveAndAwaitAdvance(); * } * phaser.arriveAndDeregister(); * }}</pre> * * <p>To create a set of {@code n} tasks using a tree of phasers, you * could use code of the following form, assuming a Task class with a * constructor accepting a {@code Phaser} that it registers with upon * construction. After invocation of {@code build(new Task[n], 0, n, * new Phaser())}, these tasks could then be started, for example by * submitting to a pool: * * <pre> {@code * void build(Task[] tasks, int lo, int hi, Phaser ph) { * if (hi - lo > TASKS_PER_PHASER) { * for (int i = lo; i < hi; i += TASKS_PER_PHASER) { * int j = Math.min(i + TASKS_PER_PHASER, hi); * build(tasks, i, j, new Phaser(ph)); * } * } else { * for (int i = lo; i < hi; ++i) * tasks[i] = new Task(ph); * // assumes new Task(ph) performs ph.register() * } * }}</pre> * * The best value of {@code TASKS_PER_PHASER} depends mainly on * expected synchronization rates. A value as low as four may * be appropriate for extremely small per-phase task bodies (thus * high rates), or up to hundreds for extremely large ones. * * <p><b>Implementation notes</b>: This implementation restricts the * maximum number of parties to 65535. Attempts to register additional * parties result in {@code IllegalStateException}. However, you can and * should create tiered phasers to accommodate arbitrarily large sets * of participants. * * @since 1.7 * @author Doug Lea */ public class Phaser { /* * This class implements an extension of X10 "clocks". Thanks to * Vijay Saraswat for the idea, and to Vivek Sarkar for * enhancements to extend functionality. */ /** * Primary state representation, holding four bit-fields: * * unarrived -- the number of parties yet to hit barrier (bits 0-15) * parties -- the number of parties to wait (bits 16-31) * phase -- the generation of the barrier (bits 32-62) * terminated -- set if barrier is terminated (bit 63 / sign) * * Except that a phaser with no registered parties is * distinguished by the otherwise illegal state of having zero * parties and one unarrived parties (encoded as EMPTY below). * * To efficiently maintain atomicity, these values are packed into * a single (atomic) long. Good performance relies on keeping * state decoding and encoding simple, and keeping race windows * short. * * All state updates are performed via CAS except initial * registration of a sub-phaser (i.e., one with a non-null * parent). In this (relatively rare) case, we use built-in * synchronization to lock while first registering with its * parent. * * The phase of a subphaser is allowed to lag that of its * ancestors until it is actually accessed -- see method * reconcileState. */ private volatile long state; private static final int MAX_PARTIES = 0xffff; private static final int MAX_PHASE = Integer.MAX_VALUE; private static final int PARTIES_SHIFT = 16; private static final int PHASE_SHIFT = 32; private static final int UNARRIVED_MASK = 0xffff; // to mask ints private static final long PARTIES_MASK = 0xffff0000L; // to mask longs private static final long COUNTS_MASK = 0xffffffffL; private static final long TERMINATION_BIT = 1L << 63; // some special values private static final int ONE_ARRIVAL = 1; private static final int ONE_PARTY = 1 << PARTIES_SHIFT; private static final int ONE_DEREGISTER = ONE_ARRIVAL|ONE_PARTY; private static final int EMPTY = 1; // The following unpacking methods are usually manually inlined private static int unarrivedOf(long s) { int counts = (int)s; return (counts == EMPTY) ? 0 : (counts & UNARRIVED_MASK); } private static int partiesOf(long s) { return (int)s >>> PARTIES_SHIFT; } private static int phaseOf(long s) { return (int)(s >>> PHASE_SHIFT); } private static int arrivedOf(long s) { int counts = (int)s; return (counts == EMPTY) ? 0 : (counts >>> PARTIES_SHIFT) - (counts & UNARRIVED_MASK); } /** * The parent of this phaser, or null if none. */ private final Phaser parent; /** * The root of phaser tree. Equals this if not in a tree. */ private final Phaser root; /** * Heads of Treiber stacks for waiting threads. To eliminate * contention when releasing some threads while adding others, we * use two of them, alternating across even and odd phases. * Subphasers share queues with root to speed up releases. */ private final AtomicReference<QNode> evenQ; private final AtomicReference<QNode> oddQ; /** * Returns message string for bounds exceptions on arrival. */ private String badArrive(long s) { return "Attempted arrival of unregistered party for " + stateToString(s); } /** * Returns message string for bounds exceptions on registration. */ private String badRegister(long s) { return "Attempt to register more than " + MAX_PARTIES + " parties for " + stateToString(s); } /** * Main implementation for methods arrive and arriveAndDeregister. * Manually tuned to speed up and minimize race windows for the * common case of just decrementing unarrived field. * * @param adjust value to subtract from state; * ONE_ARRIVAL for arrive, * ONE_DEREGISTER for arriveAndDeregister */ private int doArrive(int adjust) { final Phaser root = this.root; for (;;) { long s = (root == this) ? state : reconcileState(); int phase = (int)(s >>> PHASE_SHIFT); if (phase < 0) return phase; int counts = (int)s; int unarrived = (counts == EMPTY) ? 0 : (counts & UNARRIVED_MASK); if (unarrived <= 0) throw new IllegalStateException(badArrive(s)); if (STATE.compareAndSet(this, s, s-=adjust)) { if (unarrived == 1) { long n = s & PARTIES_MASK; // base of next state int nextUnarrived = (int)n >>> PARTIES_SHIFT; if (root == this) { if (onAdvance(phase, nextUnarrived)) n |= TERMINATION_BIT; else if (nextUnarrived == 0) n |= EMPTY; else n |= nextUnarrived; int nextPhase = (phase + 1) & MAX_PHASE; n |= (long)nextPhase << PHASE_SHIFT; STATE.compareAndSet(this, s, n); releaseWaiters(phase); } else if (nextUnarrived == 0) { // propagate deregistration phase = parent.doArrive(ONE_DEREGISTER); STATE.compareAndSet(this, s, s | EMPTY); } else phase = parent.doArrive(ONE_ARRIVAL); } return phase; } } } /** * Implementation of register, bulkRegister. * * @param registrations number to add to both parties and * unarrived fields. Must be greater than zero. */ private int doRegister(int registrations) { // adjustment to state long adjust = ((long)registrations << PARTIES_SHIFT) | registrations; final Phaser parent = this.parent; int phase; for (;;) { long s = (parent == null) ? state : reconcileState(); int counts = (int)s; int parties = counts >>> PARTIES_SHIFT; int unarrived = counts & UNARRIVED_MASK; if (registrations > MAX_PARTIES - parties) throw new IllegalStateException(badRegister(s)); phase = (int)(s >>> PHASE_SHIFT); if (phase < 0) break; if (counts != EMPTY) { // not 1st registration if (parent == null || reconcileState() == s) { if (unarrived == 0) // wait out advance root.internalAwaitAdvance(phase, null); else if (STATE.compareAndSet(this, s, s + adjust)) break; } } else if (parent == null) { // 1st root registration long next = ((long)phase << PHASE_SHIFT) | adjust; if (STATE.compareAndSet(this, s, next)) break; } else { synchronized (this) { // 1st sub registration if (state == s) { // recheck under lock phase = parent.doRegister(1); if (phase < 0) break; // finish registration whenever parent registration // succeeded, even when racing with termination, // since these are part of the same "transaction". while (!STATE.weakCompareAndSet (this, s, ((long)phase << PHASE_SHIFT) | adjust)) { s = state; phase = (int)(root.state >>> PHASE_SHIFT); // assert (int)s == EMPTY; } break; } } } } return phase; } /** * Resolves lagged phase propagation from root if necessary. * Reconciliation normally occurs when root has advanced but * subphasers have not yet done so, in which case they must finish * their own advance by setting unarrived to parties (or if * parties is zero, resetting to unregistered EMPTY state). * * @return reconciled state */ private long reconcileState() { final Phaser root = this.root; long s = state; if (root != this) { int phase, p; // CAS to root phase with current parties, tripping unarrived while ((phase = (int)(root.state >>> PHASE_SHIFT)) != (int)(s >>> PHASE_SHIFT) && !STATE.weakCompareAndSet (this, s, s = (((long)phase << PHASE_SHIFT) | ((phase < 0) ? (s & COUNTS_MASK) : (((p = (int)s >>> PARTIES_SHIFT) == 0) ? EMPTY : ((s & PARTIES_MASK) | p)))))) s = state; } return s; } /** * Creates a new phaser with no initially registered parties, no * parent, and initial phase number 0. Any thread using this * phaser will need to first register for it. */ public Phaser() { this(null, 0); } /** * Creates a new phaser with the given number of registered * unarrived parties, no parent, and initial phase number 0. * * @param parties the number of parties required to advance to the * next phase * @throws IllegalArgumentException if parties less than zero * or greater than the maximum number of parties supported */ public Phaser(int parties) { this(null, parties); } /** * Equivalent to {@link #Phaser(Phaser, int) Phaser(parent, 0)}. * * @param parent the parent phaser */ public Phaser(Phaser parent) { this(parent, 0); } /** * Creates a new phaser with the given parent and number of * registered unarrived parties. When the given parent is non-null * and the given number of parties is greater than zero, this * child phaser is registered with its parent. * * @param parent the parent phaser * @param parties the number of parties required to advance to the * next phase * @throws IllegalArgumentException if parties less than zero * or greater than the maximum number of parties supported */ public Phaser(Phaser parent, int parties) { if (parties >>> PARTIES_SHIFT != 0) throw new IllegalArgumentException("Illegal number of parties"); int phase = 0; this.parent = parent; if (parent != null) { final Phaser root = parent.root; this.root = root; this.evenQ = root.evenQ; this.oddQ = root.oddQ; if (parties != 0) phase = parent.doRegister(1); } else { this.root = this; this.evenQ = new AtomicReference<QNode>(); this.oddQ = new AtomicReference<QNode>(); } this.state = (parties == 0) ? (long)EMPTY : ((long)phase << PHASE_SHIFT) | ((long)parties << PARTIES_SHIFT) | ((long)parties); } /** * Adds a new unarrived party to this phaser. If an ongoing * invocation of {@link #onAdvance} is in progress, this method * may await its completion before returning. If this phaser has * a parent, and this phaser previously had no registered parties, * this child phaser is also registered with its parent. If * this phaser is terminated, the attempt to register has * no effect, and a negative value is returned. * * @return the arrival phase number to which this registration * applied. If this value is negative, then this phaser has * terminated, in which case registration has no effect. * @throws IllegalStateException if attempting to register more * than the maximum supported number of parties */ public int register() { return doRegister(1); } /** * Adds the given number of new unarrived parties to this phaser. * If an ongoing invocation of {@link #onAdvance} is in progress, * this method may await its completion before returning. If this * phaser has a parent, and the given number of parties is greater * than zero, and this phaser previously had no registered * parties, this child phaser is also registered with its parent. * If this phaser is terminated, the attempt to register has no * effect, and a negative value is returned. * * @param parties the number of additional parties required to * advance to the next phase * @return the arrival phase number to which this registration * applied. If this value is negative, then this phaser has * terminated, in which case registration has no effect. * @throws IllegalStateException if attempting to register more * than the maximum supported number of parties * @throws IllegalArgumentException if {@code parties < 0} */ public int bulkRegister(int parties) { if (parties < 0) throw new IllegalArgumentException(); if (parties == 0) return getPhase(); return doRegister(parties); } /** * Arrives at this phaser, without waiting for others to arrive. * * <p>It is a usage error for an unregistered party to invoke this * method. However, this error may result in an {@code * IllegalStateException} only upon some subsequent operation on * this phaser, if ever. * * @return the arrival phase number, or a negative value if terminated * @throws IllegalStateException if not terminated and the number * of unarrived parties would become negative */ public int arrive() { return doArrive(ONE_ARRIVAL); } /** * Arrives at this phaser and deregisters from it without waiting * for others to arrive. Deregistration reduces the number of * parties required to advance in future phases. If this phaser * has a parent, and deregistration causes this phaser to have * zero parties, this phaser is also deregistered from its parent. * * <p>It is a usage error for an unregistered party to invoke this * method. However, this error may result in an {@code * IllegalStateException} only upon some subsequent operation on * this phaser, if ever. * * @return the arrival phase number, or a negative value if terminated * @throws IllegalStateException if not terminated and the number * of registered or unarrived parties would become negative */ public int arriveAndDeregister() { return doArrive(ONE_DEREGISTER); } /** * Arrives at this phaser and awaits others. Equivalent in effect * to {@code awaitAdvance(arrive())}. If you need to await with * interruption or timeout, you can arrange this with an analogous * construction using one of the other forms of the {@code * awaitAdvance} method. If instead you need to deregister upon * arrival, use {@code awaitAdvance(arriveAndDeregister())}. * * <p>It is a usage error for an unregistered party to invoke this * method. However, this error may result in an {@code * IllegalStateException} only upon some subsequent operation on * this phaser, if ever. * * @return the arrival phase number, or the (negative) * {@linkplain #getPhase() current phase} if terminated * @throws IllegalStateException if not terminated and the number * of unarrived parties would become negative */ public int arriveAndAwaitAdvance() { // Specialization of doArrive+awaitAdvance eliminating some reads/paths final Phaser root = this.root; for (;;) { long s = (root == this) ? state : reconcileState(); int phase = (int)(s >>> PHASE_SHIFT); if (phase < 0) return phase; int counts = (int)s; int unarrived = (counts == EMPTY) ? 0 : (counts & UNARRIVED_MASK); if (unarrived <= 0) throw new IllegalStateException(badArrive(s)); if (STATE.compareAndSet(this, s, s -= ONE_ARRIVAL)) { if (unarrived > 1) return root.internalAwaitAdvance(phase, null); if (root != this) return parent.arriveAndAwaitAdvance(); long n = s & PARTIES_MASK; // base of next state int nextUnarrived = (int)n >>> PARTIES_SHIFT; if (onAdvance(phase, nextUnarrived)) n |= TERMINATION_BIT; else if (nextUnarrived == 0) n |= EMPTY; else n |= nextUnarrived; int nextPhase = (phase + 1) & MAX_PHASE; n |= (long)nextPhase << PHASE_SHIFT; if (!STATE.compareAndSet(this, s, n)) return (int)(state >>> PHASE_SHIFT); // terminated releaseWaiters(phase); return nextPhase; } } } /** * Awaits the phase of this phaser to advance from the given phase * value, returning immediately if the current phase is not equal * to the given phase value or this phaser is terminated. * * @param phase an arrival phase number, or negative value if * terminated; this argument is normally the value returned by a * previous call to {@code arrive} or {@code arriveAndDeregister}. * @return the next arrival phase number, or the argument if it is * negative, or the (negative) {@linkplain #getPhase() current phase} * if terminated */ public int awaitAdvance(int phase) { final Phaser root = this.root; long s = (root == this) ? state : reconcileState(); int p = (int)(s >>> PHASE_SHIFT); if (phase < 0) return phase; if (p == phase) return root.internalAwaitAdvance(phase, null); return p; } /** * Awaits the phase of this phaser to advance from the given phase * value, throwing {@code InterruptedException} if interrupted * while waiting, or returning immediately if the current phase is * not equal to the given phase value or this phaser is * terminated. * * @param phase an arrival phase number, or negative value if * terminated; this argument is normally the value returned by a * previous call to {@code arrive} or {@code arriveAndDeregister}. * @return the next arrival phase number, or the argument if it is * negative, or the (negative) {@linkplain #getPhase() current phase} * if terminated * @throws InterruptedException if thread interrupted while waiting */ public int awaitAdvanceInterruptibly(int phase) throws InterruptedException { final Phaser root = this.root; long s = (root == this) ? state : reconcileState(); int p = (int)(s >>> PHASE_SHIFT); if (phase < 0) return phase; if (p == phase) { QNode node = new QNode(this, phase, true, false, 0L); p = root.internalAwaitAdvance(phase, node); if (node.wasInterrupted) throw new InterruptedException(); } return p; } /** * Awaits the phase of this phaser to advance from the given phase * value or the given timeout to elapse, throwing {@code * InterruptedException} if interrupted while waiting, or * returning immediately if the current phase is not equal to the * given phase value or this phaser is terminated. * * @param phase an arrival phase number, or negative value if * terminated; this argument is normally the value returned by a * previous call to {@code arrive} or {@code arriveAndDeregister}. * @param timeout how long to wait before giving up, in units of * {@code unit} * @param unit a {@code TimeUnit} determining how to interpret the * {@code timeout} parameter * @return the next arrival phase number, or the argument if it is * negative, or the (negative) {@linkplain #getPhase() current phase} * if terminated * @throws InterruptedException if thread interrupted while waiting * @throws TimeoutException if timed out while waiting */ public int awaitAdvanceInterruptibly(int phase, long timeout, TimeUnit unit) throws InterruptedException, TimeoutException { long nanos = unit.toNanos(timeout); final Phaser root = this.root; long s = (root == this) ? state : reconcileState(); int p = (int)(s >>> PHASE_SHIFT); if (phase < 0) return phase; if (p == phase) { QNode node = new QNode(this, phase, true, true, nanos); p = root.internalAwaitAdvance(phase, node); if (node.wasInterrupted) throw new InterruptedException(); else if (p == phase) throw new TimeoutException(); } return p; } /** * Forces this phaser to enter termination state. Counts of * registered parties are unaffected. If this phaser is a member * of a tiered set of phasers, then all of the phasers in the set * are terminated. If this phaser is already terminated, this * method has no effect. This method may be useful for * coordinating recovery after one or more tasks encounter * unexpected exceptions. */ public void forceTermination() { // Only need to change root state final Phaser root = this.root; long s; while ((s = root.state) >= 0) { if (STATE.compareAndSet(root, s, s | TERMINATION_BIT)) { // signal all threads releaseWaiters(0); // Waiters on evenQ releaseWaiters(1); // Waiters on oddQ return; } } } /** * Returns the current phase number. The maximum phase number is * {@code Integer.MAX_VALUE}, after which it restarts at * zero. Upon termination, the phase number is negative, * in which case the prevailing phase prior to termination * may be obtained via {@code getPhase() + Integer.MIN_VALUE}. * * @return the phase number, or a negative value if terminated */ public final int getPhase() { return (int)(root.state >>> PHASE_SHIFT); } /** * Returns the number of parties registered at this phaser. * * @return the number of parties */ public int getRegisteredParties() { return partiesOf(state); } /** * Returns the number of registered parties that have arrived at * the current phase of this phaser. If this phaser has terminated, * the returned value is meaningless and arbitrary. * * @return the number of arrived parties */ public int getArrivedParties() { return arrivedOf(reconcileState()); } /** * Returns the number of registered parties that have not yet * arrived at the current phase of this phaser. If this phaser has * terminated, the returned value is meaningless and arbitrary. * * @return the number of unarrived parties */ public int getUnarrivedParties() { return unarrivedOf(reconcileState()); } /** * Returns the parent of this phaser, or {@code null} if none. * * @return the parent of this phaser, or {@code null} if none */ public Phaser getParent() { return parent; } /** * Returns the root ancestor of this phaser, which is the same as * this phaser if it has no parent. * * @return the root ancestor of this phaser */ public Phaser getRoot() { return root; } /** * Returns {@code true} if this phaser has been terminated. * * @return {@code true} if this phaser has been terminated */ public boolean isTerminated() { return root.state < 0L; } /** * Overridable method to perform an action upon impending phase * advance, and to control termination. This method is invoked * upon arrival of the party advancing this phaser (when all other * waiting parties are dormant). If this method returns {@code * true}, this phaser will be set to a final termination state * upon advance, and subsequent calls to {@link #isTerminated} * will return true. Any (unchecked) Exception or Error thrown by * an invocation of this method is propagated to the party * attempting to advance this phaser, in which case no advance * occurs. * * <p>The arguments to this method provide the state of the phaser * prevailing for the current transition. The effects of invoking * arrival, registration, and waiting methods on this phaser from * within {@code onAdvance} are unspecified and should not be * relied on. * * <p>If this phaser is a member of a tiered set of phasers, then * {@code onAdvance} is invoked only for its root phaser on each * advance. * * <p>To support the most common use cases, the default * implementation of this method returns {@code true} when the * number of registered parties has become zero as the result of a * party invoking {@code arriveAndDeregister}. You can disable * this behavior, thus enabling continuation upon future * registrations, by overriding this method to always return * {@code false}: * * <pre> {@code * Phaser phaser = new Phaser() { * protected boolean onAdvance(int phase, int parties) { return false; } * }}</pre> * * @param phase the current phase number on entry to this method, * before this phaser is advanced * @param registeredParties the current number of registered parties * @return {@code true} if this phaser should terminate */ protected boolean onAdvance(int phase, int registeredParties) { return registeredParties == 0; } /** * Returns a string identifying this phaser, as well as its * state. The state, in brackets, includes the String {@code * "phase = "} followed by the phase number, {@code "parties = "} * followed by the number of registered parties, and {@code * "arrived = "} followed by the number of arrived parties. * * @return a string identifying this phaser, as well as its state */ public String toString() { return stateToString(reconcileState()); } /** * Implementation of toString and string-based error messages. */ private String stateToString(long s) { return super.toString() + "[phase = " + phaseOf(s) + " parties = " + partiesOf(s) + " arrived = " + arrivedOf(s) + "]"; } // Waiting mechanics /** * Removes and signals threads from queue for phase. */ private void releaseWaiters(int phase) { QNode q; // first element of queue Thread t; // its thread AtomicReference<QNode> head = (phase & 1) == 0 ? evenQ : oddQ; while ((q = head.get()) != null && q.phase != (int)(root.state >>> PHASE_SHIFT)) { if (head.compareAndSet(q, q.next) && (t = q.thread) != null) { q.thread = null; LockSupport.unpark(t); } } } /** * Variant of releaseWaiters that additionally tries to remove any * nodes no longer waiting for advance due to timeout or * interrupt. Currently, nodes are removed only if they are at * head of queue, which suffices to reduce memory footprint in * most usages. * * @return current phase on exit */ private int abortWait(int phase) { AtomicReference<QNode> head = (phase & 1) == 0 ? evenQ : oddQ; for (;;) { Thread t; QNode q = head.get(); int p = (int)(root.state >>> PHASE_SHIFT); if (q == null || ((t = q.thread) != null && q.phase == p)) return p; if (head.compareAndSet(q, q.next) && t != null) { q.thread = null; LockSupport.unpark(t); } } } /** The number of CPUs, for spin control */ private static final int NCPU = Runtime.getRuntime().availableProcessors(); /** * The number of times to spin before blocking while waiting for * advance, per arrival while waiting. On multiprocessors, fully * blocking and waking up a large number of threads all at once is * usually a very slow process, so we use rechargeable spins to * avoid it when threads regularly arrive: When a thread in * internalAwaitAdvance notices another arrival before blocking, * and there appear to be enough CPUs available, it spins * SPINS_PER_ARRIVAL more times before blocking. The value trades * off good-citizenship vs big unnecessary slowdowns. */ static final int SPINS_PER_ARRIVAL = (NCPU < 2) ? 1 : 1 << 8; /** * Possibly blocks and waits for phase to advance unless aborted. * Call only on root phaser. * * @param phase current phase * @param node if non-null, the wait node to track interrupt and timeout; * if null, denotes noninterruptible wait * @return current phase */ private int internalAwaitAdvance(int phase, QNode node) { // assert root == this; releaseWaiters(phase-1); // ensure old queue clean boolean queued = false; // true when node is enqueued int lastUnarrived = 0; // to increase spins upon change int spins = SPINS_PER_ARRIVAL; long s; int p; while ((p = (int)((s = state) >>> PHASE_SHIFT)) == phase) { if (node == null) { // spinning in noninterruptible mode int unarrived = (int)s & UNARRIVED_MASK; if (unarrived != lastUnarrived && (lastUnarrived = unarrived) < NCPU) spins += SPINS_PER_ARRIVAL; boolean interrupted = Thread.interrupted(); if (interrupted || --spins < 0) { // need node to record intr node = new QNode(this, phase, false, false, 0L); node.wasInterrupted = interrupted; } else Thread.onSpinWait(); } else if (node.isReleasable()) // done or aborted break; else if (!queued) { // push onto queue AtomicReference<QNode> head = (phase & 1) == 0 ? evenQ : oddQ; QNode q = node.next = head.get(); if ((q == null || q.phase == phase) && (int)(state >>> PHASE_SHIFT) == phase) // avoid stale enq queued = head.compareAndSet(q, node); } else { try { ForkJoinPool.managedBlock(node); } catch (InterruptedException cantHappen) { node.wasInterrupted = true; } } } if (node != null) { if (node.thread != null) node.thread = null; // avoid need for unpark() if (node.wasInterrupted && !node.interruptible) Thread.currentThread().interrupt(); if (p == phase && (p = (int)(state >>> PHASE_SHIFT)) == phase) return abortWait(phase); // possibly clean up on abort } releaseWaiters(phase); return p; } /** * Wait nodes for Treiber stack representing wait queue. */ static final class QNode implements ForkJoinPool.ManagedBlocker { final Phaser phaser; final int phase; final boolean interruptible; final boolean timed; boolean wasInterrupted; long nanos; final long deadline; volatile Thread thread; // nulled to cancel wait QNode next; QNode(Phaser phaser, int phase, boolean interruptible, boolean timed, long nanos) { this.phaser = phaser; this.phase = phase; this.interruptible = interruptible; this.nanos = nanos; this.timed = timed; this.deadline = timed ? System.nanoTime() + nanos : 0L; thread = Thread.currentThread(); } public boolean isReleasable() { if (thread == null) return true; if (phaser.getPhase() != phase) { thread = null; return true; } if (Thread.interrupted()) wasInterrupted = true; if (wasInterrupted && interruptible) { thread = null; return true; } if (timed && (nanos <= 0L || (nanos = deadline - System.nanoTime()) <= 0L)) { thread = null; return true; } return false; } public boolean block() { while (!isReleasable()) { if (timed) LockSupport.parkNanos(this, nanos); else LockSupport.park(this); } return true; } } // VarHandle mechanics private static final VarHandle STATE; static { try { MethodHandles.Lookup l = MethodHandles.lookup(); STATE = l.findVarHandle(Phaser.class, "state", long.class); } catch (ReflectiveOperationException e) { throw new ExceptionInInitializerError(e); } // Reduce the risk of rare disastrous classloading in first call to // LockSupport.park: https://bugs.openjdk.java.net/browse/JDK-8074773 Class<?> ensureLoaded = LockSupport.class; } }
⏎ java/util/concurrent/Phaser.java
Or download all of them as a single archive file:
File name: java.base-11.0.1-src.zip File size: 8740354 bytes Release date: 2018-11-04 Download
2020-05-29, 205091👍, 0💬
Popular Posts:
kernel.jar is a component in iText Java library to provide low-level functionalities. iText Java lib...
commons-collections4-4.4 -sources.jaris the source JAR file for Apache Commons Collections 4.2, whic...
JDK 7 tools.jar is the JAR file for JDK 7 tools. It contains Java classes to support different JDK t...
Saxon-HE (home edition) is an open source product available under the Mozilla Public License. It pro...
How to display types defined in an XML Schema file with the xs\QueryXS.java provided in the Apache X...