1155 lines
46 KiB
Java
1155 lines
46 KiB
Java
/*
|
|
* DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
|
|
*
|
|
* This code is free software; you can redistribute it and/or modify it
|
|
* under the terms of the GNU General Public License version 2 only, as
|
|
* published by the Free Software Foundation. Oracle designates this
|
|
* particular file as subject to the "Classpath" exception as provided
|
|
* by Oracle in the LICENSE file that accompanied this code.
|
|
*
|
|
* This code is distributed in the hope that it will be useful, but WITHOUT
|
|
* ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
|
|
* FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License
|
|
* version 2 for more details (a copy is included in the LICENSE file that
|
|
* accompanied this code).
|
|
*
|
|
* You should have received a copy of the GNU General Public License version
|
|
* 2 along with this work; if not, write to the Free Software Foundation,
|
|
* Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA.
|
|
*
|
|
* Please contact Oracle, 500 Oracle Parkway, Redwood Shores, CA 94065 USA
|
|
* or visit www.oracle.com if you need additional information or have any
|
|
* questions.
|
|
*/
|
|
|
|
/*
|
|
* This file is available under and governed by the GNU General Public
|
|
* License version 2 only, as published by the Free Software Foundation.
|
|
* However, the following notice accompanied the original version of this
|
|
* file:
|
|
*
|
|
* 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 parallelism level 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>Memory consistency effects: Actions prior to any form of arrive
|
|
* method <a href="package-summary.html#MemoryVisibility">
|
|
* <i>happen-before</i></a> a corresponding phase advance and
|
|
* onAdvance actions (if present), which in turn <i>happen-before</i>
|
|
* actions following the phase advance.
|
|
*
|
|
* <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;
|
|
}
|
|
// Android-removed: remove usage of Thread.onSpinWait. http://b/202837191
|
|
// 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;
|
|
}
|
|
}
|