CyclicBarrier and CountDownLatch but supporting
more flexible usage.
Registration. Unlike the case for other barriers, the
number of parties registered to synchronize on a phaser
may vary over time. Tasks may be registered at any time (using
methods register, bulkRegister, or forms of
constructors establishing initial numbers of parties), and
optionally deregistered upon any arrival (using 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.)
Synchronization. Like a CyclicBarrier, a Phaser may be repeatedly awaited. Method arriveAndAwaitAdvance has effect analogous to CyclicBarrier.. 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 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:
arrive and
arriveAndDeregister record arrival. These methods
do not block, but return an associated arrival phase
number; 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 onAdvance(int, int), which also controls
termination. Overriding this method is similar to, but more
flexible than, providing a barrier action to a CyclicBarrier.
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 CyclicBarrier,
method 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 forceTermination. Phasers may
also be used by tasks executing in a ForkJoinPool.
Progress is ensured if the pool's parallelism level can
accommodate the maximum number of simultaneously blocked
parties.
Termination. A phaser may enter a termination
state, that may be checked using method 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 onAdvance
returns true. The default implementation returns 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 forceTermination is
also available to abruptly release waiting threads and allow them
to terminate.
Tiering. Phasers may be tiered (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.
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 Phaser(Phaser, int)
constructor, register, or bulkRegister), the
child phaser is registered with its parent. Whenever the number of
registered parties becomes zero as the result of an invocation of
arriveAndDeregister, the child phaser is deregistered
from its parent.
Monitoring. 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 getRegisteredParties parties in total, of which getArrivedParties have arrived at the current phase (getPhase). When the remaining (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 toString
returns snapshots of these state queries in a form convenient for
informal monitoring.
Memory consistency effects: Actions prior to any form of arrive method happen-before a corresponding phase advance and onAdvance actions (if present), which in turn happen-before actions following the phase advance.
Sample usages:
A Phaser may be used instead of a 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:
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();
}
One way to cause a set of threads to repeatedly perform actions
for a given number of iterations is to override onAdvance:
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();
}
If the main task must later await termination, it
may re-register and then execute a similar loop:
// ...
phaser.register();
while (!phaser.isTerminated())
phaser.arriveAndAwaitAdvance();
Related constructions may be used to await particular phase numbers
in contexts where you are sure that the phase will never wrap around
Integer.MAX_VALUE. For example:
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();
}
To create a set of n tasks using a tree of phasers, you
could use code of the following form, assuming a Task class with a
constructor accepting a Phaser that it registers with upon
construction. After invocation of build(new Task[n], 0, n,
new Phaser()), these tasks could then be started, for example by
submitting to a pool:
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()
}
}
The best value of 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.Implementation Notes
This implementation restricts the
maximum number of parties to 65535. Attempts to register additional
parties result in IllegalStateException. However, you can and
should create tiered phasers to accommodate arbitrarily large sets
of participants.
| Modifier and Type | Class and Description |
|---|---|
| pack-priv static class | Phaser.
Wait nodes for Treiber stack representing wait queue. |
| Modifier and Type | Field and Description |
|---|---|
| private static final long | |
| private static final int | |
| private final AtomicReference | evenQ
Heads of Treiber stacks for waiting threads. |
| private static final int | |
| private static final int | |
| private static final int | NCPU
The number of CPUs, for spin control |
| private final AtomicReference | |
| private static final int | |
| private static final int | |
| private static final int | |
| private final Phaser | parent
The parent of this phaser, or null if none. |
| private static final long | |
| private static final int | |
| private static final int | |
| private final Phaser | root
The root of phaser tree. |
| pack-priv static final int | SPINS_PER_ARRIVAL
The number of times to spin before blocking while waiting for advance, per arrival while waiting. |
| private volatile long | state
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). |
| private static final VarHandle | |
| private static final long | |
| private static final int |
| Access | Constructor and Description |
|---|---|
| public | Phaser()
Creates a new phaser with no initially registered parties, no parent, and initial phase number 0. |
| public | Phaser(int
the number of parties required to advance to the
next phase parties)Creates a new phaser with the given number of registered unarrived parties, no parent, and initial phase number 0. |
| public | |
| public |
| Modifier and Type | Method and Description |
|---|---|
| private int | Returns: current phase on exitVariant of releaseWaiters that additionally tries to remove any nodes no longer waiting for advance due to timeout or interrupt. |
| public int | Returns: the arrival phase number, or a negative value if terminatedArrives at this phaser, without waiting for others to arrive. |
| public int | Returns: the arrival phase number, or the (negative) current phase if terminatedArrives at this phaser and awaits others. |
| public int | Returns: the arrival phase number, or a negative value if terminatedArrives at this phaser and deregisters from it without waiting for others to arrive. |
| private static int | |
| public int | Returns: the next arrival phase number, or the argument if it is negative, or the (negative) current phase if terminatedan arrival phase number, or negative value if
terminated; this argument is normally the value returned by a
previous call to phase)arrive or arriveAndDeregister.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. |
| public int | Returns: the next arrival phase number, or the argument if it is negative, or the (negative) current phase if terminatedan arrival phase number, or negative value if
terminated; this argument is normally the value returned by a
previous call to phase)arrive or arriveAndDeregister.Awaits the phase of this phaser to advance from the given phase
value, throwing |
| public int | Returns: the next arrival phase number, or the argument if it is negative, or the (negative) current phase if terminatedan arrival phase number, or negative value if
terminated; this argument is normally the value returned by a
previous call to phase, long arrive or arriveAndDeregister.how long to wait before giving up, in units of
timeout, TimeUnit unita unit)TimeUnit determining how to interpret the
timeout parameterAwaits the phase of this phaser to advance from the given phase
value or the given timeout to elapse, throwing |
| private String | |
| private String | |
| public int | Returns: 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.the number of additional parties required to
advance to the next phase parties)Adds the given number of new unarrived parties to this phaser. |
| private int | doArrive(int
value to subtract from state;
ONE_ARRIVAL for arrive,
ONE_DEREGISTER for arriveAndDeregister adjust)Main implementation for methods arrive and arriveAndDeregister. |
| private int | doRegister(int
number to add to both parties and
unarrived fields. Must be greater than zero. registrations)Implementation of register, bulkRegister. |
| public void | |
| public int | Returns: the number of arrived partiesReturns the number of registered parties that have arrived at the current phase of this phaser. |
| public Phaser | Returns: the parent of this phaser, ornull if noneReturns the parent of this phaser, or |
| public final int | Returns: the phase number, or a negative value if terminatedReturns the current phase number. |
| public int | Returns: the number of partiesReturns the number of parties registered at this phaser. |
| public Phaser | Returns: the root ancestor of this phaserReturns the root ancestor of this phaser, which is the same as this phaser if it has no parent. |
| public int | Returns: the number of unarrived partiesReturns the number of registered parties that have not yet arrived at the current phase of this phaser. |
| private int | Returns: current phasecurrent phase phase, Phaser.if non-null, the wait node to track interrupt and timeout;
if null, denotes noninterruptible wait nodePossibly blocks and waits for phase to advance unless aborted. |
| public boolean | Returns: true if this phaser has been terminatedReturns |
| protected boolean | Returns: true if this phaser should terminatethe current phase number on entry to this method,
before this phaser is advanced phase, int the current number of registered parties registeredParties)Overridable method to perform an action upon impending phase advance, and to control termination. |
| private static int | |
| private static int | |
| private long | |
| public int | Returns: 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.Adds a new unarrived party to this phaser. |
| private void | |
| private String | |
| public String | Returns: a string identifying this phaser, as well as its stateOverrides java. Returns a string identifying this phaser, as well as its state. |
| private static int |
| COUNTS_MASK | back to summary |
|---|---|
| private static final long COUNTS_MASK | |
| EMPTY | back to summary |
|---|---|
| private static final int EMPTY | |
| evenQ | back to summary |
|---|---|
| private final AtomicReference<Phaser. 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. | |
| MAX_PARTIES | back to summary |
|---|---|
| private static final int MAX_PARTIES | |
| MAX_PHASE | back to summary |
|---|---|
| private static final int MAX_PHASE | |
| NCPU | back to summary |
|---|---|
| private static final int NCPU The number of CPUs, for spin control | |
| oddQ | back to summary |
|---|---|
| private final AtomicReference<Phaser. | |
| ONE_ARRIVAL | back to summary |
|---|---|
| private static final int ONE_ARRIVAL | |
| ONE_DEREGISTER | back to summary |
|---|---|
| private static final int ONE_DEREGISTER | |
| ONE_PARTY | back to summary |
|---|---|
| private static final int ONE_PARTY | |
| parent | back to summary |
|---|---|
| private final Phaser parent The parent of this phaser, or null if none. | |
| PARTIES_MASK | back to summary |
|---|---|
| private static final long PARTIES_MASK | |
| PARTIES_SHIFT | back to summary |
|---|---|
| private static final int PARTIES_SHIFT | |
| PHASE_SHIFT | back to summary |
|---|---|
| private static final int PHASE_SHIFT | |
| root | back to summary |
|---|---|
| private final Phaser root The root of phaser tree. Equals this if not in a tree. | |
| SPINS_PER_ARRIVAL | back to summary |
|---|---|
| pack-priv static final int SPINS_PER_ARRIVAL 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. | |
| state | back to summary |
|---|---|
| private volatile long state 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. | |
| STATE | back to summary |
|---|---|
| private static final VarHandle STATE | |
| TERMINATION_BIT | back to summary |
|---|---|
| private static final long TERMINATION_BIT | |
| UNARRIVED_MASK | back to summary |
|---|---|
| private static final int UNARRIVED_MASK | |
| Phaser | back to summary |
|---|---|
| public Phaser() 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. | |
| Phaser | back to summary |
|---|---|
| public Phaser(int parties) Creates a new phaser with the given number of registered unarrived parties, no parent, and initial phase number 0.
| |
| Phaser | back to summary |
|---|---|
| public Phaser(Phaser parent) Equivalent to
| |
| Phaser | back to summary |
|---|---|
| public Phaser(Phaser parent, int parties) 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.
| |
| abortWait | back to summary |
|---|---|
| private int abortWait(int phase) 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.
| |
| arrive | back to summary |
|---|---|
| public int arrive() Arrives at this phaser, without waiting for others to arrive. It is a usage error for an unregistered party to invoke this
method. However, this error may result in an
| |
| arriveAndAwaitAdvance | back to summary |
|---|---|
| public int arriveAndAwaitAdvance() Arrives at this phaser and awaits others. Equivalent in effect
to It is a usage error for an unregistered party to invoke this
method. However, this error may result in an
| |
| arriveAndDeregister | back to summary |
|---|---|
| public int arriveAndDeregister() 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. It is a usage error for an unregistered party to invoke this
method. However, this error may result in an
| |
| arrivedOf | back to summary |
|---|---|
| private static int arrivedOf(long s) | |
| awaitAdvance | back to summary |
|---|---|
| public int awaitAdvance(int phase) 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.
| |
| awaitAdvanceInterruptibly | back to summary |
|---|---|
| public int awaitAdvanceInterruptibly(int phase) throws InterruptedException Awaits the phase of this phaser to advance from the given phase
value, throwing
| |
| awaitAdvanceInterruptibly | back to summary |
|---|---|
| public int awaitAdvanceInterruptibly(int phase, long timeout, TimeUnit unit) throws InterruptedException, TimeoutException Awaits the phase of this phaser to advance from the given phase
value or the given timeout to elapse, throwing
| |
| badArrive | back to summary |
|---|---|
| private String badArrive(long s) Returns message string for bounds exceptions on arrival. | |
| badRegister | back to summary |
|---|---|
| private String badRegister(long s) Returns message string for bounds exceptions on registration. | |
| bulkRegister | back to summary |
|---|---|
| public int bulkRegister(int parties) Adds the given number of new unarrived parties to this phaser.
If an ongoing invocation of
| |
| doArrive | back to summary |
|---|---|
| private int doArrive(int adjust) 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.
| |
| doRegister | back to summary |
|---|---|
| private int doRegister(int registrations) Implementation of register, bulkRegister.
| |
| forceTermination | back to summary |
|---|---|
| public void forceTermination() 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. | |
| getArrivedParties | back to summary |
|---|---|
| public int getArrivedParties() 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.
| |
| getParent | back to summary |
|---|---|
| public Phaser getParent() Returns the parent of this phaser, or
| |
| getPhase | back to summary |
|---|---|
| public final int getPhase() Returns the current phase number. The maximum phase number is
| |
| getRegisteredParties | back to summary |
|---|---|
| public int getRegisteredParties() Returns the number of parties registered at this phaser.
| |
| getRoot | back to summary |
|---|---|
| public Phaser getRoot() Returns the root ancestor of this phaser, which is the same as this phaser if it has no parent.
| |
| getUnarrivedParties | back to summary |
|---|---|
| public int getUnarrivedParties() 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.
| |
| internalAwaitAdvance | back to summary |
|---|---|
| private int internalAwaitAdvance(int phase, Phaser. Possibly blocks and waits for phase to advance unless aborted. Call only on root phaser. | |
| isTerminated | back to summary |
|---|---|
| public boolean isTerminated() Returns
| |
| onAdvance | back to summary |
|---|---|
| protected boolean onAdvance(int phase, int registeredParties) 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 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 If this phaser is a member of a tiered set of phasers, then
To support the most common use cases, the default
implementation of this method returns
| |
| partiesOf | back to summary |
|---|---|
| private static int partiesOf(long s) | |
| phaseOf | back to summary |
|---|---|
| private static int phaseOf(long s) | |
| reconcileState | back to summary |
|---|---|
| private long reconcileState() 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).
| |
| register | back to summary |
|---|---|
| public int register() Adds a new unarrived party to this phaser. If an ongoing
invocation of
| |
| releaseWaiters | back to summary |
|---|---|
| private void releaseWaiters(int phase) Removes and signals threads from queue for phase. | |
| stateToString | back to summary |
|---|---|
| private String stateToString(long s) Implementation of toString and string-based error messages. | |
| toString | back to summary |
|---|---|
| public String toString() Overrides java. Returns a string identifying this phaser, as well as its
state. The state, in brackets, includes the String
| |
| unarrivedOf | back to summary |
|---|---|
| private static int unarrivedOf(long s) | |
| Modifier and Type | Field and Description |
|---|---|
| pack-priv final long | |
| pack-priv final boolean | |
| pack-priv long | |
| pack-priv Phaser. | |
| pack-priv final int | |
| pack-priv final Phaser | |
| pack-priv volatile Thread | |
| pack-priv final boolean | |
| pack-priv boolean |
| Access | Constructor and Description |
|---|---|
| pack-priv |
| Modifier and Type | Method and Description |
|---|---|
| public boolean | block()
Implements java. Possibly blocks the current thread, for example waiting for a lock or condition. |
| public boolean | isReleasable()
Implements java. Returns |
| deadline | back to summary |
|---|---|
| pack-priv final long deadline | |
| interruptible | back to summary |
|---|---|
| pack-priv final boolean interruptible | |
| nanos | back to summary |
|---|---|
| pack-priv long nanos | |
| next | back to summary |
|---|---|
| pack-priv Phaser. | |
| phase | back to summary |
|---|---|
| pack-priv final int phase | |
| phaser | back to summary |
|---|---|
| pack-priv final Phaser phaser | |
| thread | back to summary |
|---|---|
| pack-priv volatile Thread thread | |
| timed | back to summary |
|---|---|
| pack-priv final boolean timed | |
| wasInterrupted | back to summary |
|---|---|
| pack-priv boolean wasInterrupted | |
| QNode | back to summary |
|---|---|
| pack-priv QNode(Phaser phaser, int phase, boolean interruptible, boolean timed, long nanos) | |
| block | back to summary |
|---|---|
| public boolean block() Implements java. Doc from java. Possibly blocks the current thread, for example waiting for a lock or condition. | |
| isReleasable | back to summary |
|---|---|
| public boolean isReleasable() Implements java. Doc from java. Returns | |