A {@code ForkJoinPool} differs from other kinds of {@link @@ -30,33 +30,19 @@ import java.util.concurrent.CountDownLat * execute subtasks created by other active tasks (eventually blocking * waiting for work if none exist). This enables efficient processing * when most tasks spawn other subtasks (as do most {@code - * ForkJoinTask}s). A {@code ForkJoinPool} may also be used for mixed - * execution of some plain {@code Runnable}- or {@code Callable}- - * based activities along with {@code ForkJoinTask}s. When setting - * {@linkplain #setAsyncMode async mode}, a {@code ForkJoinPool} may - * also be appropriate for use with fine-grained tasks of any form - * that are never joined. Otherwise, other {@code ExecutorService} - * implementations are typically more appropriate choices. + * ForkJoinTask}s). When setting asyncMode to true in + * constructors, {@code ForkJoinPool}s may also be appropriate for use + * with event-style tasks that are never joined. * *
A {@code ForkJoinPool} is constructed with a given target * parallelism level; by default, equal to the number of available - * processors. Unless configured otherwise via {@link - * #setMaintainsParallelism}, the pool attempts to maintain this - * number of active (or available) threads by dynamically adding, - * suspending, or resuming internal worker threads, even if some tasks - * are stalled waiting to join others. However, no such adjustments - * are performed in the face of blocked IO or other unmanaged - * synchronization. The nested {@link ManagedBlocker} interface - * enables extension of the kinds of synchronization accommodated. - * The target parallelism level may also be changed dynamically - * ({@link #setParallelism}). The total number of threads may be - * limited using method {@link #setMaximumPoolSize}, in which case it - * may become possible for the activities of a pool to stall due to - * the lack of available threads to process new tasks. When the pool - * is executing tasks, these and other configuration setting methods - * may only gradually affect actual pool sizes. It is normally best - * practice to invoke these methods only when the pool is known to be - * quiescent. + * processors. The pool attempts to maintain enough active (or + * available) threads by dynamically adding, suspending, or resuming + * internal worker threads, even if some tasks are stalled waiting to + * join others. However, no such adjustments are guaranteed in the + * face of blocked IO or other unmanaged synchronization. The nested + * {@link ManagedBlocker} interface enables extension of the kinds of + * synchronization accommodated. * *
In addition to execution and lifecycle control methods, this * class provides status check methods (for example @@ -65,6 +51,40 @@ import java.util.concurrent.CountDownLat * {@link #toString} returns indications of pool state in a * convenient form for informal monitoring. * + *
As is the case with other ExecutorServices, there are three + * main task execution methods summarized in the following + * table. These are designed to be used by clients not already engaged + * in fork/join computations in the current pool. The main forms of + * these methods accept instances of {@code ForkJoinTask}, but + * overloaded forms also allow mixed execution of plain {@code + * Runnable}- or {@code Callable}- based activities as well. However, + * tasks that are already executing in a pool should normally + * NOT use these pool execution methods, but instead use the + * within-computation forms listed in the table. + * + *
| + * | Call from non-fork/join clients | + *Call from within fork/join computations | + *
| Arange async execution | + *{@link #execute(ForkJoinTask)} | + *{@link ForkJoinTask#fork} | + *
| Await and obtain result | + *{@link #invoke(ForkJoinTask)} | + *{@link ForkJoinTask#invoke} | + *
| Arrange exec and obtain Future | + *{@link #submit(ForkJoinTask)} | + *{@link ForkJoinTask#fork} (ForkJoinTasks are Futures) | + *
Sample Usage. Normally a single {@code ForkJoinPool} is * used for all parallel task execution in a program or subsystem. * Otherwise, use would not usually outweigh the construction and @@ -89,7 +109,8 @@ import java.util.concurrent.CountDownLat * {@code IllegalArgumentException}. * *
This implementation rejects submitted tasks (that is, by throwing
- * {@link RejectedExecutionException}) only when the pool is shut down.
+ * {@link RejectedExecutionException}) only when the pool is shut down
+ * or internal resources have been exhuasted.
*
* @since 1.7
* @author Doug Lea
@@ -116,21 +137,59 @@ public class ForkJoinPool extends Abstra
* of tasks profit from cache affinities, but others are harmed by
* cache pollution effects.)
*
+ * Beyond work-stealing support and essential bookkeeping, the
+ * main responsibility of this framework is to take actions when
+ * one worker is waiting to join a task stolen (or always held by)
+ * another. Becauae we are multiplexing many tasks on to a pool
+ * of workers, we can't just let them block (as in Thread.join).
+ * We also cannot just reassign the joiner's run-time stack with
+ * another and replace it later, which would be a form of
+ * "continuation", that even if possible is not necessarily a good
+ * idea. Given that the creation costs of most threads on most
+ * systems mainly surrounds setting up runtime stacks, thread
+ * creation and switching is usually not much more expensive than
+ * stack creation and switching, and is more flexible). Instead we
+ * combine two tactics:
+ *
+ * Helping: Arranging for the joiner to execute some task that it
+ * would be running if the steal had not occurred. Method
+ * ForkJoinWorkerThread.helpJoinTask tracks joining->stealing
+ * links to try to find such a task.
+ *
+ * Compensating: Unless there are already enough live threads,
+ * method helpMaintainParallelism() may create or or
+ * re-activate a spare thread to compensate for blocked
+ * joiners until they unblock.
+ *
+ * Because the determining existence of conservatively safe
+ * helping targets, the availability of already-created spares,
+ * and the apparent need to create new spares are all racy and
+ * require heuristic guidance, we rely on multiple retries of
+ * each. Further, because it is impossible to keep exactly the
+ * target (parallelism) number of threads running at any given
+ * time, we allow compensation during joins to fail, and enlist
+ * all other threads to help out whenever they are not otherwise
+ * occupied (i.e., mainly in method preStep).
+ *
+ * The ManagedBlocker extension API can't use helping so relies
+ * only on compensation in method awaitBlocker.
+ *
* The main throughput advantages of work-stealing stem from
* decentralized control -- workers mostly steal tasks from each
* other. We do not want to negate this by creating bottlenecks
- * implementing the management responsibilities of this class. So
- * we use a collection of techniques that avoid, reduce, or cope
- * well with contention. These entail several instances of
- * bit-packing into CASable fields to maintain only the minimally
- * required atomicity. To enable such packing, we restrict maximum
- * parallelism to (1<<15)-1 (enabling twice this to fit into a 16
- * bit field), which is far in excess of normal operating range.
- * Even though updates to some of these bookkeeping fields do
- * sometimes contend with each other, they don't normally
- * cache-contend with updates to others enough to warrant memory
- * padding or isolation. So they are all held as fields of
- * ForkJoinPool objects. The main capabilities are as follows:
+ * implementing other management responsibilities. So we use a
+ * collection of techniques that avoid, reduce, or cope well with
+ * contention. These entail several instances of bit-packing into
+ * CASable fields to maintain only the minimally required
+ * atomicity. To enable such packing, we restrict maximum
+ * parallelism to (1<<15)-1 (enabling twice this (to accommodate
+ * unbalanced increments and decrements) to fit into a 16 bit
+ * field, which is far in excess of normal operating range. Even
+ * though updates to some of these bookkeeping fields do sometimes
+ * contend with each other, they don't normally cache-contend with
+ * updates to others enough to warrant memory padding or
+ * isolation. So they are all held as fields of ForkJoinPool
+ * objects. The main capabilities are as follows:
*
* 1. Creating and removing workers. Workers are recorded in the
* "workers" array. This is an array as opposed to some other data
@@ -146,31 +205,32 @@ public class ForkJoinPool extends Abstra
* blocked workers. However, all other support code is set up to
* work with other policies.
*
+ * To ensure that we do not hold on to worker references that
+ * would prevent GC, ALL accesses to workers are via indices into
+ * the workers array (which is one source of some of the unusual
+ * code constructions here). In essence, the workers array serves
+ * as a WeakReference mechanism. Thus for example the event queue
+ * stores worker indices, not worker references. Access to the
+ * workers in associated methods (for example releaseEventWaiters)
+ * must both index-check and null-check the IDs. All such accesses
+ * ignore bad IDs by returning out early from what they are doing,
+ * since this can only be associated with shutdown, in which case
+ * it is OK to give up. On termination, we just clobber these
+ * data structures without trying to use them.
+ *
* 2. Bookkeeping for dynamically adding and removing workers. We
- * maintain a given level of parallelism (or, if
- * maintainsParallelism is false, at least avoid starvation). When
- * some workers are known to be blocked (on joins or via
+ * aim to approximately maintain the given level of parallelism.
+ * When some workers are known to be blocked (on joins or via
* ManagedBlocker), we may create or resume others to take their
* place until they unblock (see below). Implementing this
* requires counts of the number of "running" threads (i.e., those
* that are neither blocked nor artifically suspended) as well as
* the total number. These two values are packed into one field,
* "workerCounts" because we need accurate snapshots when deciding
- * to create, resume or suspend. To support these decisions,
- * updates to spare counts must be prospective (not
- * retrospective). For example, the running count is decremented
- * before blocking by a thread about to block as a spare, but
- * incremented by the thread about to unblock it. Updates upon
- * resumption ofr threads blocking in awaitJoin or awaitBlocker
- * cannot usually be prospective, so the running count is in
- * general an upper bound of the number of productively running
- * threads Updates to the workerCounts field sometimes transiently
- * encounter a fair amount of contention when join dependencies
- * are such that many threads block or unblock at about the same
- * time. We alleviate this by sometimes bundling updates (for
- * example blocking one thread on join and resuming a spare cancel
- * each other out), and in most other cases performing an
- * alternative action like releasing waiters or locating spares.
+ * to create, resume or suspend. Note however that the
+ * correspondance of these counts to reality is not guaranteed. In
+ * particular updates for unblocked threads may lag until they
+ * actually wake up.
*
* 3. Maintaining global run state. The run state of the pool
* consists of a runLevel (SHUTDOWN, TERMINATING, etc) similar to
@@ -199,28 +259,28 @@ public class ForkJoinPool extends Abstra
* workers that previously could not find a task to now find one:
* Submission of a new task to the pool, or another worker pushing
* a task onto a previously empty queue. (We also use this
- * mechanism for termination and reconfiguration actions that
- * require wakeups of idle workers). Each worker maintains its
- * last known event count, and blocks when a scan for work did not
- * find a task AND its lastEventCount matches the current
- * eventCount. Waiting idle workers are recorded in a variant of
- * Treiber stack headed by field eventWaiters which, when nonzero,
- * encodes the thread index and count awaited for by the worker
- * thread most recently calling eventSync. This thread in turn has
- * a record (field nextEventWaiter) for the next waiting worker.
- * In addition to allowing simpler decisions about need for
- * wakeup, the event count bits in eventWaiters serve the role of
- * tags to avoid ABA errors in Treiber stacks. To reduce delays
- * in task diffusion, workers not otherwise occupied may invoke
- * method releaseWaiters, that removes and signals (unparks)
- * workers not waiting on current count. To minimize task
- * production stalls associate with signalling, any worker pushing
- * a task on an empty queue invokes the weaker method signalWork,
- * that only releases idle workers until it detects interference
- * by other threads trying to release, and lets them take
- * over. The net effect is a tree-like diffusion of signals, where
- * released threads (and possibly others) help with unparks. To
- * further reduce contention effects a bit, failed CASes to
+ * mechanism for termination actions that require wakeups of idle
+ * workers). Each worker maintains its last known event count,
+ * and blocks when a scan for work did not find a task AND its
+ * lastEventCount matches the current eventCount. Waiting idle
+ * workers are recorded in a variant of Treiber stack headed by
+ * field eventWaiters which, when nonzero, encodes the thread
+ * index and count awaited for by the worker thread most recently
+ * calling eventSync. This thread in turn has a record (field
+ * nextEventWaiter) for the next waiting worker. In addition to
+ * allowing simpler decisions about need for wakeup, the event
+ * count bits in eventWaiters serve the role of tags to avoid ABA
+ * errors in Treiber stacks. To reduce delays in task diffusion,
+ * workers not otherwise occupied may invoke method
+ * releaseEventWaiters, that removes and signals (unparks) workers
+ * not waiting on current count. To reduce stalls, To minimize
+ * task production stalls associate with signalling, any worker
+ * pushing a task on an empty queue invokes the weaker method
+ * signalWork, that only releases idle workers until it detects
+ * interference by other threads trying to release, and lets them
+ * take over. The net effect is a tree-like diffusion of signals,
+ * where released threads (and possibly others) help with unparks.
+ * To further reduce contention effects a bit, failed CASes to
* increment field eventCount are tolerated without retries.
* Conceptually they are merged into the same event, which is OK
* when their only purpose is to enable workers to scan for work.
@@ -228,80 +288,52 @@ public class ForkJoinPool extends Abstra
* 5. Managing suspension of extra workers. When a worker is about
* to block waiting for a join (or via ManagedBlockers), we may
* create a new thread to maintain parallelism level, or at least
- * avoid starvation (see below). Usually, extra threads are needed
- * for only very short periods, yet join dependencies are such
- * that we sometimes need them in bursts. Rather than create new
- * threads each time this happens, we suspend no-longer-needed
- * extra ones as "spares". For most purposes, we don't distinguish
- * "extra" spare threads from normal "core" threads: On each call
- * to preStep (the only point at which we can do this) a worker
+ * avoid starvation. Usually, extra threads are needed for only
+ * very short periods, yet join dependencies are such that we
+ * sometimes need them in bursts. Rather than create new threads
+ * each time this happens, we suspend no-longer-needed extra ones
+ * as "spares". For most purposes, we don't distinguish "extra"
+ * spare threads from normal "core" threads: On each call to
+ * preStep (the only point at which we can do this) a worker
* checks to see if there are now too many running workers, and if
- * so, suspends itself. Methods awaitJoin and awaitBlocker look
- * for suspended threads to resume before considering creating a
- * new replacement. We don't need a special data structure to
- * maintain spares; simply scanning the workers array looking for
- * worker.isSuspended() is fine because the calling thread is
- * otherwise not doing anything useful anyway; we are at least as
- * happy if after locating a spare, the caller doesn't actually
- * block because the join is ready before we try to adjust and
- * compensate. Note that this is intrinsically racy. One thread
+ * so, suspends itself. Method helpMaintainParallelism looks for
+ * suspended threads to resume before considering creating a new
+ * replacement. The spares themselves are encoded on another
+ * variant of a Treiber Stack, headed at field "spareWaiters".
+ * Note that the use of spares is intrinsically racy. One thread
* may become a spare at about the same time as another is
* needlessly being created. We counteract this and related slop
* in part by requiring resumed spares to immediately recheck (in
- * preStep) to see whether they they should re-suspend. The only
- * effective difference between "extra" and "core" threads is that
- * we allow the "extra" ones to time out and die if they are not
- * resumed within a keep-alive interval of a few seconds. This is
- * implemented mainly within ForkJoinWorkerThread, but requires
- * some coordination (isTrimmed() -- meaning killed while
- * suspended) to correctly maintain pool counts.
+ * preStep) to see whether they they should re-suspend. To avoid
+ * long-term build-up of spares, the oldest spare (see
+ * ForkJoinWorkerThread.suspendAsSpare) occasionally wakes up if
+ * not signalled and calls tryTrimSpare, which uses two different
+ * thresholds: Always killing if the number of spares is greater
+ * that 25% of total, and killing others only at a slower rate
+ * (UNUSED_SPARE_TRIM_RATE_NANOS).
*
* 6. Deciding when to create new workers. The main dynamic
- * control in this class is deciding when to create extra threads,
- * in methods awaitJoin and awaitBlocker. We always
- * need to create one when the number of running threads becomes
- * zero. But because blocked joins are typically dependent, we
- * don't necessarily need or want one-to-one replacement. Using a
- * one-to-one compensation rule often leads to enough useless
- * overhead creating, suspending, resuming, and/or killing threads
- * to signficantly degrade throughput. We use a rule reflecting
- * the idea that, the more spare threads you already have, the
- * more evidence you need to create another one. The "evidence"
- * here takes two forms: (1) Using a creation threshold expressed
- * in terms of the current deficit -- target minus running
- * threads. To reduce flickering and drift around target values,
- * the relation is quadratic: adding a spare if (dc*dc)>=(sc*pc)
- * (where dc is deficit, sc is number of spare threads and pc is
- * target parallelism.) (2) Using a form of adaptive
- * spionning. requiring a number of threshold checks proportional
- * to the number of spare threads. This effectively reduces churn
- * at the price of systematically undershooting target parallelism
- * when many threads are blocked. However, biasing toward
- * undeshooting partially compensates for the above mechanics to
- * suspend extra threads, that normally lead to overshoot because
- * we can only suspend workers in-between top-level actions. It
- * also better copes with the fact that some of the methods in
- * this class tend to never become compiled (but are interpreted),
- * so some components of the entire set of controls might execute
- * many times faster than others. And similarly for cases where
- * the apparent lack of work is just due to GC stalls and other
+ * control in this class is deciding when to create extra threads
+ * in method helpMaintainParallelism. We would like to keep
+ * exactly #parallelism threads running, which is an impossble
+ * task. We always need to create one when the number of running
+ * threads would become zero and all workers are busy. Beyond
+ * this, we must rely on heuristics that work well in the the
+ * presence of transients phenomena such as GC stalls, dynamic
+ * compilation, and wake-up lags. These transients are extremely
+ * common -- we are normally trying to fully saturate the CPUs on
+ * a machine, so almost any activity other than running tasks
+ * impedes accuracy. Our main defense is to allow some slack in
+ * creation thresholds, using rules that reflect the fact that the
+ * more threads we have running, the more likely that we are
+ * underestimating the number running threads. The rules also
+ * better cope with the fact that some of the methods in this
+ * class tend to never become compiled (but are interpreted), so
+ * some components of the entire set of controls might execute 100
+ * times faster than others. And similarly for cases where the
+ * apparent lack of work is just due to GC stalls and other
* transient system activity.
*
- * 7. Maintaining other configuration parameters and monitoring
- * statistics. Updates to fields controlling parallelism level,
- * max size, etc can only meaningfully take effect for individual
- * threads upon their next top-level actions; i.e., between
- * stealing/running tasks/submission, which are separated by calls
- * to preStep. Memory ordering for these (assumed infrequent)
- * reconfiguration calls is ensured by using reads and writes to
- * volatile field workerCounts (that must be read in preStep anyway)
- * as "fences" -- user-level reads are preceded by reads of
- * workCounts, and writes are followed by no-op CAS to
- * workerCounts. The values reported by other management and
- * monitoring methods are either computed on demand, or are kept
- * in fields that are only updated when threads are otherwise
- * idle.
- *
* Beware that there is a lot of representation-level coupling
* among classes ForkJoinPool, ForkJoinWorkerThread, and
* ForkJoinTask. For example, direct access to "workers" array by
@@ -313,11 +345,13 @@ public class ForkJoinPool extends Abstra
*
* Style notes: There are lots of inline assignments (of form
* "while ((local = field) != 0)") which are usually the simplest
- * way to ensure read orderings. Also several occurrences of the
- * unusual "do {} while(!cas...)" which is the simplest way to
- * force an update of a CAS'ed variable. There are also a few
- * other coding oddities that help some methods perform reasonably
- * even when interpreted (not compiled).
+ * way to ensure the required read orderings (which are sometimes
+ * critical). Also several occurrences of the unusual "do {}
+ * while(!cas...)" which is the simplest way to force an update of
+ * a CAS'ed variable. There are also other coding oddities that
+ * help some methods perform reasonably even when interpreted (not
+ * compiled), at the expense of some messy constructions that
+ * reduce byte code counts.
*
* The order of declarations in this file is: (1) statics (2)
* fields (along with constants used when unpacking some of them)
@@ -346,7 +380,7 @@ public class ForkJoinPool extends Abstra
* Default ForkJoinWorkerThreadFactory implementation; creates a
* new ForkJoinWorkerThread.
*/
- static class DefaultForkJoinWorkerThreadFactory
+ static class DefaultForkJoinWorkerThreadFactory
implements ForkJoinWorkerThreadFactory {
public ForkJoinWorkerThread newThread(ForkJoinPool pool) {
return new ForkJoinWorkerThread(pool);
@@ -385,10 +419,11 @@ public class ForkJoinPool extends Abstra
new AtomicInteger();
/**
- * Absolute bound for parallelism level. Twice this number must
- * fit into a 16bit field to enable word-packing for some counts.
+ * Absolute bound for parallelism level. Twice this number plus
+ * one (i.e., 0xfff) must fit into a 16bit field to enable
+ * word-packing for some counts and indices.
*/
- private static final int MAX_THREADS = 0x7fff;
+ private static final int MAX_WORKERS = 0x7fff;
/**
* Array holding all worker threads in the pool. Array size must
@@ -414,7 +449,7 @@ public class ForkJoinPool extends Abstra
/**
* Latch released upon termination.
*/
- private final CountDownLatch terminationLatch;
+ private final Phaser termination;
/**
* Creation factory for worker threads.
@@ -428,25 +463,47 @@ public class ForkJoinPool extends Abstra
private volatile long stealCount;
/**
+ * The last nanoTime that a spare thread was trimmed
+ */
+ private volatile long trimTime;
+
+ /**
+ * The rate at which to trim unused spares
+ */
+ static final long UNUSED_SPARE_TRIM_RATE_NANOS =
+ 1000L * 1000L * 1000L; // 1 sec
+
+ /**
* Encoded record of top of treiber stack of threads waiting for
* events. The top 32 bits contain the count being waited for. The
- * bottom word contains one plus the pool index of waiting worker
- * thread.
+ * bottom 16 bits contains one plus the pool index of waiting
+ * worker thread. (Bits 16-31 are unused.)
*/
private volatile long eventWaiters;
private static final int EVENT_COUNT_SHIFT = 32;
- private static final long WAITER_INDEX_MASK = (1L << EVENT_COUNT_SHIFT)-1L;
+ private static final long WAITER_ID_MASK = (1L << 16) - 1L;
/**
* A counter for events that may wake up worker threads:
* - Submission of a new task to the pool
* - A worker pushing a task on an empty queue
- * - termination and reconfiguration
+ * - termination
*/
private volatile int eventCount;
/**
+ * Encoded record of top of treiber stack of spare threads waiting
+ * for resumption. The top 16 bits contain an arbitrary count to
+ * avoid ABA effects. The bottom 16bits contains one plus the pool
+ * index of waiting worker thread.
+ */
+ private volatile int spareWaiters;
+
+ private static final int SPARE_COUNT_SHIFT = 16;
+ private static final int SPARE_ID_MASK = (1 << 16) - 1;
+
+ /**
* Lifecycle control. The low word contains the number of workers
* that are (probably) executing tasks. This value is atomically
* incremented before a worker gets a task to run, and decremented
@@ -475,8 +532,7 @@ public class ForkJoinPool extends Abstra
* making decisions about creating and suspending spare
* threads. Updated only by CAS. Note that adding a new worker
* requires incrementing both counts, since workers start off in
- * running state. This field is also used for memory-fencing
- * configuration parameters.
+ * running state.
*/
private volatile int workerCounts;
@@ -485,58 +541,45 @@ public class ForkJoinPool extends Abstra
private static final int ONE_RUNNING = 1;
private static final int ONE_TOTAL = 1 << TOTAL_COUNT_SHIFT;
- /*
- * Fields parallelism. maxPoolSize, and maintainsParallelism are
- * non-volatile, but external reads/writes use workerCount fences
- * to ensure visability.
- */
-
/**
* The target parallelism level.
+ * Accessed directly by ForkJoinWorkerThreads.
*/
- private int parallelism;
-
- /**
- * The maximum allowed pool size.
- */
- private int maxPoolSize;
+ final int parallelism;
/**
* True if use local fifo, not default lifo, for local polling
- * Replicated by ForkJoinWorkerThreads
+ * Read by, and replicated by ForkJoinWorkerThreads
*/
- private volatile boolean locallyFifo;
+ final boolean locallyFifo;
/**
- * Controls whether to add spares to maintain parallelism
+ * The uncaught exception handler used when any worker abruptly
+ * terminates.
*/
- private boolean maintainsParallelism;
-
- /**
- * The uncaught exception handler used when any worker
- * abruptly terminates
- */
- private volatile Thread.UncaughtExceptionHandler ueh;
+ private final Thread.UncaughtExceptionHandler ueh;
/**
* Pool number, just for assigning useful names to worker threads
*/
private final int poolNumber;
- // utilities for updating fields
+
+ // Utilities for CASing fields. Note that several of these
+ // are manually inlined by callers
/**
- * Adds delta to running count. Used mainly by ForkJoinTask.
+ * Increments running count part of workerCounts
*/
- final void updateRunningCount(int delta) {
- int wc;
+ final void incrementRunningCount() {
+ int c;
do {} while (!UNSAFE.compareAndSwapInt(this, workerCountsOffset,
- wc = workerCounts,
- wc + delta));
+ c = workerCounts,
+ c + ONE_RUNNING));
}
/**
- * Decrements running count unless already zero
+ * Tries to decrement running count unless already zero
*/
final boolean tryDecrementRunningCount() {
int wc = workerCounts;
@@ -547,21 +590,33 @@ public class ForkJoinPool extends Abstra
}
/**
- * Write fence for user modifications of pool parameters
- * (parallelism. etc). Note that it doesn't matter if CAS fails.
+ * Forces decrement of encoded workerCounts, awaiting nonzero if
+ * (rarely) necessary when other count updates lag.
+ *
+ * @param dr -- either zero or ONE_RUNNING
+ * @param dt == either zero or ONE_TOTAL
*/
- private void workerCountWriteFence() {
- int wc;
- UNSAFE.compareAndSwapInt(this, workerCountsOffset,
- wc = workerCounts, wc);
+ private void decrementWorkerCounts(int dr, int dt) {
+ for (;;) {
+ int wc = workerCounts;
+ if (wc == 0 && (runState & TERMINATED) != 0)
+ return; // lagging termination on a backout
+ if ((wc & RUNNING_COUNT_MASK) - dr < 0 ||
+ (wc >>> TOTAL_COUNT_SHIFT) - dt < 0)
+ Thread.yield();
+ if (UNSAFE.compareAndSwapInt(this, workerCountsOffset,
+ wc, wc - (dr + dt)))
+ return;
+ }
}
/**
- * Read fence for external reads of pool parameters
- * (parallelism. maxPoolSize, etc).
+ * Increments event count
*/
- private void workerCountReadFence() {
- int ignore = workerCounts;
+ private void advanceEventCount() {
+ int c;
+ do {} while(!UNSAFE.compareAndSwapInt(this, eventCountOffset,
+ c = eventCount, c+1));
}
/**
@@ -612,12 +667,12 @@ public class ForkJoinPool extends Abstra
lock.lock();
try {
ForkJoinWorkerThread[] ws = workers;
- int nws = ws.length;
- if (k < 0 || k >= nws || ws[k] != null) {
- for (k = 0; k < nws && ws[k] != null; ++k)
+ int n = ws.length;
+ if (k < 0 || k >= n || ws[k] != null) {
+ for (k = 0; k < n && ws[k] != null; ++k)
;
- if (k == nws)
- ws = Arrays.copyOf(ws, nws << 1);
+ if (k == n)
+ ws = Arrays.copyOf(ws, n << 1);
}
ws[k] = w;
workers = ws; // volatile array write ensures slot visibility
@@ -650,423 +705,342 @@ public class ForkJoinPool extends Abstra
* Tries to create and add new worker. Assumes that worker counts
* are already updated to accommodate the worker, so adjusts on
* failure.
- *
- * @return new worker or null if creation failed
*/
- private ForkJoinWorkerThread addWorker() {
+ private void addWorker() {
ForkJoinWorkerThread w = null;
try {
w = factory.newThread(this);
} finally { // Adjust on either null or exceptional factory return
if (w == null) {
- onWorkerCreationFailure();
- return null;
+ decrementWorkerCounts(ONE_RUNNING, ONE_TOTAL);
+ tryTerminate(false); // in case of failure during shutdown
}
}
- w.start(recordWorker(w), locallyFifo, ueh);
- return w;
- }
-
- /**
- * Adjusts counts upon failure to create worker
- */
- private void onWorkerCreationFailure() {
- for (;;) {
- int wc = workerCounts;
- if ((wc >>> TOTAL_COUNT_SHIFT) > 0 &&
- UNSAFE.compareAndSwapInt(this, workerCountsOffset,
- wc, wc - (ONE_RUNNING|ONE_TOTAL)))
- break;
- }
- tryTerminate(false); // in case of failure during shutdown
- }
-
- /**
- * Create enough total workers to establish target parallelism,
- * giving up if terminating or addWorker fails
- */
- private void ensureEnoughTotalWorkers() {
- int wc;
- while (((wc = workerCounts) >>> TOTAL_COUNT_SHIFT) < parallelism &&
- runState < TERMINATING) {
- if ((UNSAFE.compareAndSwapInt(this, workerCountsOffset,
- wc, wc + (ONE_RUNNING|ONE_TOTAL)) &&
- addWorker() == null))
- break;
- }
+ if (w != null)
+ w.start(recordWorker(w), ueh);
}
/**
* Final callback from terminating worker. Removes record of
* worker from array, and adjusts counts. If pool is shutting
- * down, tries to complete terminatation, else possibly replaces
- * the worker.
+ * down, tries to complete terminatation.
*
* @param w the worker
*/
final void workerTerminated(ForkJoinWorkerThread w) {
- if (w.active) { // force inactive
- w.active = false;
- do {} while (!tryDecrementActiveCount());
- }
forgetWorker(w);
-
- // Decrement total count, and if was running, running count
- // Spin (waiting for other updates) if either would be negative
- int nr = w.isTrimmed() ? 0 : ONE_RUNNING;
- int unit = ONE_TOTAL + nr;
- for (;;) {
- int wc = workerCounts;
- int rc = wc & RUNNING_COUNT_MASK;
- if (rc - nr < 0 || (wc >>> TOTAL_COUNT_SHIFT) == 0)
- Thread.yield(); // back off if waiting for other updates
- else if (UNSAFE.compareAndSwapInt(this, workerCountsOffset,
- wc, wc - unit))
- break;
- }
-
- accumulateStealCount(w); // collect final count
- if (!tryTerminate(false))
- ensureEnoughTotalWorkers();
+ decrementWorkerCounts(w.isTrimmed()? 0 : ONE_RUNNING, ONE_TOTAL);
+ while (w.stealCount != 0) // collect final count
+ tryAccumulateStealCount(w);
+ tryTerminate(false);
}
// Waiting for and signalling events
/**
- * Ensures eventCount on exit is different (mod 2^32) than on
- * entry. CAS failures are OK -- any change in count suffices.
- */
- private void advanceEventCount() {
- int c;
- UNSAFE.compareAndSwapInt(this, eventCountOffset, c = eventCount, c+1);
- }
-
- /**
* Releases workers blocked on a count not equal to current count.
+ * Normally called after precheck that eventWaiters isn't zero to
+ * avoid wasted array checks.
+ *
+ * @param signalling true if caller is a signalling worker so can
+ * exit upon (conservatively) detected contention by other threads
+ * who will continue to release
*/
- final void releaseWaiters() {
- long top;
- int id;
- while ((id = (int)((top = eventWaiters) & WAITER_INDEX_MASK)) > 0 &&
- (int)(top >>> EVENT_COUNT_SHIFT) != eventCount) {
- ForkJoinWorkerThread[] ws = workers;
- ForkJoinWorkerThread w;
- if (ws.length >= id && (w = ws[id - 1]) != null &&
- UNSAFE.compareAndSwapLong(this, eventWaitersOffset,
- top, w.nextWaiter))
+ private void releaseEventWaiters(boolean signalling) {
+ ForkJoinWorkerThread[] ws = workers;
+ int n = ws.length;
+ long h; // head of stack
+ ForkJoinWorkerThread w; int id, ec;
+ while ((id = ((int)((h = eventWaiters) & WAITER_ID_MASK)) - 1) >= 0 &&
+ (int)(h >>> EVENT_COUNT_SHIFT) != (ec = eventCount) &&
+ id < n && (w = ws[id]) != null) {
+ if (UNSAFE.compareAndSwapLong(this, eventWaitersOffset,
+ h, h = w.nextWaiter))
LockSupport.unpark(w);
+ if (signalling && (eventCount != ec || eventWaiters != h))
+ break;
}
}
/**
- * Advances eventCount and releases waiters until interference by
- * other releasing threads is detected.
+ * Tries to advance eventCount and releases waiters. Called only
+ * from workers.
*/
final void signalWork() {
- int ec;
- UNSAFE.compareAndSwapInt(this, eventCountOffset, ec=eventCount, ec+1);
- outer:for (;;) {
- long top = eventWaiters;
- ec = eventCount;
- for (;;) {
- ForkJoinWorkerThread[] ws; ForkJoinWorkerThread w;
- int id = (int)(top & WAITER_INDEX_MASK);
- if (id <= 0 || (int)(top >>> EVENT_COUNT_SHIFT) == ec)
- return;
- if ((ws = workers).length < id || (w = ws[id - 1]) == null ||
- !UNSAFE.compareAndSwapLong(this, eventWaitersOffset,
- top, top = w.nextWaiter))
- continue outer; // possibly stale; reread
- LockSupport.unpark(w);
- if (top != eventWaiters) // let someone else take over
- return;
- }
- }
+ int c; // try to increment event count -- CAS failure OK
+ UNSAFE.compareAndSwapInt(this, eventCountOffset, c = eventCount, c+1);
+ if (eventWaiters != 0L)
+ releaseEventWaiters(true);
}
/**
- * If worker is inactive, blocks until terminating or event count
- * advances from last value held by worker; in any case helps
- * release others.
+ * Blocks worker until terminating or event count
+ * advances from last value held by worker
*
* @param w the calling worker thread
*/
private void eventSync(ForkJoinWorkerThread w) {
- if (!w.active) {
- int prev = w.lastEventCount;
- long nextTop = (((long)prev << EVENT_COUNT_SHIFT) |
- ((long)(w.poolIndex + 1)));
- long top;
- while ((runState < SHUTDOWN || !tryTerminate(false)) &&
- (((int)(top = eventWaiters) & WAITER_INDEX_MASK) == 0 ||
- (int)(top >>> EVENT_COUNT_SHIFT) == prev) &&
- eventCount == prev) {
- if (UNSAFE.compareAndSwapLong(this, eventWaitersOffset,
- w.nextWaiter = top, nextTop)) {
- accumulateStealCount(w); // transfer steals while idle
- Thread.interrupted(); // clear/ignore interrupt
- while (eventCount == prev)
- w.doPark();
- break;
+ int wec = w.lastEventCount;
+ long nh = (((long)wec) << EVENT_COUNT_SHIFT) | ((long)(w.poolIndex+1));
+ long h;
+ while ((runState < SHUTDOWN || !tryTerminate(false)) &&
+ ((h = eventWaiters) == 0L ||
+ (int)(h >>> EVENT_COUNT_SHIFT) == wec) &&
+ eventCount == wec) {
+ if (UNSAFE.compareAndSwapLong(this, eventWaitersOffset,
+ w.nextWaiter = h, nh)) {
+ while (runState < TERMINATING && eventCount == wec) {
+ if (!tryAccumulateStealCount(w)) // transfer while idle
+ continue;
+ Thread.interrupted(); // clear/ignore interrupt
+ if (eventCount != wec)
+ break;
+ LockSupport.park(w);
}
+ break;
}
- w.lastEventCount = eventCount;
}
- releaseWaiters();
+ w.lastEventCount = eventCount;
}
+ // Maintaining spares
+
/**
- * Callback from workers invoked upon each top-level action (i.e.,
- * stealing a task or taking a submission and running
- * it). Performs one or both of the following:
- *
- * * If the worker cannot find work, updates its active status to
- * inactive and updates activeCount unless there is contention, in
- * which case it may try again (either in this or a subsequent
- * call). Additionally, awaits the next task event and/or helps
- * wake up other releasable waiters.
- *
- * * If there are too many running threads, suspends this worker
- * (first forcing inactivation if necessary). If it is not
- * resumed before a keepAlive elapses, the worker may be "trimmed"
- * -- killed while suspended within suspendAsSpare. Otherwise,
- * upon resume it rechecks to make sure that it is still needed.
- *
- * @param w the worker
- * @param worked false if the worker scanned for work but didn't
- * find any (in which case it may block waiting for work).
+ * Pushes worker onto the spare stack
*/
- final void preStep(ForkJoinWorkerThread w, boolean worked) {
- boolean active = w.active;
- boolean inactivate = !worked & active;
- for (;;) {
- if (inactivate) {
- int c = runState;
- if (UNSAFE.compareAndSwapInt(this, runStateOffset,
- c, c - ONE_ACTIVE))
- inactivate = active = w.active = false;
- }
- int wc = workerCounts;
- if ((wc & RUNNING_COUNT_MASK) <= parallelism) {
- if (!worked)
- eventSync(w);
- return;
+ final void pushSpare(ForkJoinWorkerThread w) {
+ int ns = (++w.spareCount << SPARE_COUNT_SHIFT) | (w.poolIndex+1);
+ do {} while (!UNSAFE.compareAndSwapInt(this, spareWaitersOffset,
+ w.nextSpare = spareWaiters,ns));
+ }
+
+ /**
+ * Tries (once) to resume a spare if running count is less than
+ * target parallelism. Fails on contention or stale workers.
+ */
+ private void tryResumeSpare() {
+ int sw, id;
+ ForkJoinWorkerThread w;
+ ForkJoinWorkerThread[] ws;
+ if ((id = ((sw = spareWaiters) & SPARE_ID_MASK) - 1) >= 0 &&
+ id < (ws = workers).length && (w = ws[id]) != null &&
+ (workerCounts & RUNNING_COUNT_MASK) < parallelism &&
+ eventWaiters == 0L &&
+ spareWaiters == sw &&
+ UNSAFE.compareAndSwapInt(this, spareWaitersOffset,
+ sw, w.nextSpare) &&
+ w.tryUnsuspend()) {
+ int c; // try increment; if contended, finish after unpark
+ boolean inc = UNSAFE.compareAndSwapInt(this, workerCountsOffset,
+ c = workerCounts,
+ c + ONE_RUNNING);
+ LockSupport.unpark(w);
+ if (!inc) {
+ do {} while(!UNSAFE.compareAndSwapInt(this, workerCountsOffset,
+ c = workerCounts,
+ c + ONE_RUNNING));
}
- if (!(inactivate |= active) && // must inactivate to suspend
- UNSAFE.compareAndSwapInt(this, workerCountsOffset,
- wc, wc - ONE_RUNNING) &&
- !w.suspendAsSpare()) // false if trimmed
- return;
}
}
/**
- * Adjusts counts and creates or resumes compensating threads for
- * a worker blocking on task joinMe. First tries resuming an
- * existing spare (which usually also avoids any count
- * adjustment), but must then decrement running count to determine
- * whether a new thread is needed. See above for fuller
- * explanation. This code is sprawled out non-modularly mainly
- * because adaptive spinning works best if the entire method is
- * either interpreted or compiled vs having only some pieces of it
- * compiled.
- *
- * @param joinMe the task to join
- * @return task status on exit (to simplify usage by callers)
- */
- final int awaitJoin(ForkJoinTask joinMe) {
+ * Callback from oldest spare occasionally waking up. Tries
+ * (once) to shutdown a spare if more than 25% spare overage, or
+ * if UNUSED_SPARE_TRIM_RATE_NANOS have elapsed and there are at
+ * least #parallelism running threads. Note that we don't need CAS
+ * or locks here because the method is called only from the oldest
+ * suspended spare occasionally waking (and even misfires are OK).
+ *
+ * @param now the wake up nanoTime of caller
+ */
+ final void tryTrimSpare(long now) {
+ long lastTrim = trimTime;
+ trimTime = now;
+ helpMaintainParallelism(); // first, help wake up any needed spares
+ int sw, id;
+ ForkJoinWorkerThread w;
+ ForkJoinWorkerThread[] ws;
int pc = parallelism;
- boolean adj = false; // true when running count adjusted
- int scans = 0;
-
- while (joinMe.status >= 0) {
- ForkJoinWorkerThread spare = null;
- if ((workerCounts & RUNNING_COUNT_MASK) < pc) {
- ForkJoinWorkerThread[] ws = workers;
- int nws = ws.length;
- for (int i = 0; i < nws; ++i) {
- ForkJoinWorkerThread w = ws[i];
- if (w != null && w.isSuspended()) {
- spare = w;
- break;
- }
- }
- if (joinMe.status < 0)
- break;
- }
- int wc = workerCounts;
- int rc = wc & RUNNING_COUNT_MASK;
- int dc = pc - rc;
- if (dc > 0 && spare != null && spare.tryUnsuspend()) {
- if (adj) {
- int c;
- do {} while (!UNSAFE.compareAndSwapInt
- (this, workerCountsOffset,
- c = workerCounts, c + ONE_RUNNING));
- }
- adj = true;
- LockSupport.unpark(spare);
- }
- else if (adj) {
- if (dc <= 0)
- break;
- int tc = wc >>> TOTAL_COUNT_SHIFT;
- if (scans > tc) {
- int ts = (tc - pc) * pc;
- if (rc != 0 && (dc * dc < ts || !maintainsParallelism))
- break;
- if (scans > ts && tc < maxPoolSize &&
- UNSAFE.compareAndSwapInt(this, workerCountsOffset, wc,
- wc+(ONE_RUNNING|ONE_TOTAL))){
- addWorker();
- break;
- }
- }
- }
- else if (rc != 0)
- adj = UNSAFE.compareAndSwapInt (this, workerCountsOffset,
- wc, wc - ONE_RUNNING);
- if ((scans++ & 1) == 0)
- releaseWaiters(); // help others progress
- else
- Thread.yield(); // avoid starving productive threads
- }
+ int wc = workerCounts;
+ if ((wc & RUNNING_COUNT_MASK) >= pc &&
+ (((wc >>> TOTAL_COUNT_SHIFT) - pc) > (pc >>> 2) + 1 ||// approx 25%
+ now - lastTrim >= UNUSED_SPARE_TRIM_RATE_NANOS) &&
+ (id = ((sw = spareWaiters) & SPARE_ID_MASK) - 1) >= 0 &&
+ id < (ws = workers).length && (w = ws[id]) != null &&
+ UNSAFE.compareAndSwapInt(this, spareWaitersOffset,
+ sw, w.nextSpare))
+ w.shutdown(false);
+ }
- if (adj) {
- joinMe.internalAwaitDone();
- int c;
- do {} while (!UNSAFE.compareAndSwapInt
- (this, workerCountsOffset,
- c = workerCounts, c + ONE_RUNNING));
+ /**
+ * Does at most one of:
+ *
+ * 1. Help wake up existing workers waiting for work via
+ * releaseEventWaiters. (If any exist, then it probably doesn't
+ * matter right now if under target parallelism level.)
+ *
+ * 2. If below parallelism level and a spare exists, try (once)
+ * to resume it via tryResumeSpare.
+ *
+ * 3. If neither of the above, tries (once) to add a new
+ * worker if either there are not enough total, or if all
+ * existing workers are busy, there are either no running
+ * workers or the deficit is at least twice the surplus.
+ */
+ private void helpMaintainParallelism() {
+ // uglified to work better when not compiled
+ int pc, wc, rc, tc, rs; long h;
+ if ((h = eventWaiters) != 0L) {
+ if ((int)(h >>> EVENT_COUNT_SHIFT) != eventCount)
+ releaseEventWaiters(false); // avoid useless call
+ }
+ else if ((pc = parallelism) >
+ (rc = ((wc = workerCounts) & RUNNING_COUNT_MASK))) {
+ if (spareWaiters != 0)
+ tryResumeSpare();
+ else if ((rs = runState) < TERMINATING &&
+ ((tc = wc >>> TOTAL_COUNT_SHIFT) < pc ||
+ (tc == (rs & ACTIVE_COUNT_MASK) && // all busy
+ (rc == 0 || // must add
+ rc < pc - ((tc - pc) << 1)) && // within slack
+ tc < MAX_WORKERS && runState == rs)) && // recheck busy
+ workerCounts == wc &&
+ UNSAFE.compareAndSwapInt(this, workerCountsOffset, wc,
+ wc + (ONE_RUNNING|ONE_TOTAL)))
+ addWorker();
}
- return joinMe.status;
}
/**
- * Same idea as awaitJoin
+ * Callback from workers invoked upon each top-level action (i.e.,
+ * stealing a task or taking a submission and running
+ * it). Performs one or more of the following:
+ *
+ * 1. If the worker cannot find work (misses > 0), updates its
+ * active status to inactive and updates activeCount unless
+ * this is the first miss and there is contention, in which
+ * case it may try again (either in this or a subsequent
+ * call).
+ *
+ * 2. If there are at least 2 misses, awaits the next task event
+ * via eventSync
+ *
+ * 3. If there are too many running threads, suspends this worker
+ * (first forcing inactivation if necessary). If it is not
+ * needed, it may be killed while suspended via
+ * tryTrimSpare. Otherwise, upon resume it rechecks to make
+ * sure that it is still needed.
+ *
+ * 4. Helps release and/or reactivate other workers via
+ * helpMaintainParallelism
+ *
+ * @param w the worker
+ * @param misses the number of scans by caller failing to find work
+ * (saturating at 2 just to avoid wraparound)
*/
- final void awaitBlocker(ManagedBlocker blocker, boolean maintainPar)
- throws InterruptedException {
- maintainPar &= maintainsParallelism;
+ final void preStep(ForkJoinWorkerThread w, int misses) {
+ boolean active = w.active;
int pc = parallelism;
- boolean adj = false; // true when running count adjusted
- int scans = 0;
- boolean done;
-
for (;;) {
- if (done = blocker.isReleasable())
- break;
- ForkJoinWorkerThread spare = null;
- if ((workerCounts & RUNNING_COUNT_MASK) < pc) {
- ForkJoinWorkerThread[] ws = workers;
- int nws = ws.length;
- for (int i = 0; i < nws; ++i) {
- ForkJoinWorkerThread w = ws[i];
- if (w != null && w.isSuspended()) {
- spare = w;
- break;
- }
- }
- if (done = blocker.isReleasable())
- break;
- }
int wc = workerCounts;
int rc = wc & RUNNING_COUNT_MASK;
- int dc = pc - rc;
- if (dc > 0 && spare != null && spare.tryUnsuspend()) {
- if (adj) {
- int c;
- do {} while (!UNSAFE.compareAndSwapInt
- (this, workerCountsOffset,
- c = workerCounts, c + ONE_RUNNING));
- }
- adj = true;
- LockSupport.unpark(spare);
+ if (active && (misses > 0 || rc > pc)) {
+ int rs; // try inactivate
+ if (UNSAFE.compareAndSwapInt(this, runStateOffset,
+ rs = runState, rs - ONE_ACTIVE))
+ active = w.active = false;
+ else if (misses > 1 || rc > pc ||
+ (rs & ACTIVE_COUNT_MASK) >= pc)
+ continue; // force inactivate
}
- else if (adj) {
- if (dc <= 0)
+ if (misses > 1) {
+ misses = 0; // don't re-sync
+ eventSync(w); // continue loop to recheck rc
+ }
+ else if (rc > pc) {
+ if (workerCounts == wc && // try to suspend as spare
+ UNSAFE.compareAndSwapInt(this, workerCountsOffset,
+ wc, wc - ONE_RUNNING) &&
+ !w.suspendAsSpare()) // false if killed
break;
- int tc = wc >>> TOTAL_COUNT_SHIFT;
- if (scans > tc) {
- int ts = (tc - pc) * pc;
- if (rc != 0 && (dc * dc < ts || !maintainPar))
- break;
- if (scans > ts && tc < maxPoolSize &&
- UNSAFE.compareAndSwapInt(this, workerCountsOffset, wc,
- wc+(ONE_RUNNING|ONE_TOTAL))){
- addWorker();
- break;
- }
- }
}
- else if (rc != 0)
- adj = UNSAFE.compareAndSwapInt (this, workerCountsOffset,
- wc, wc - ONE_RUNNING);
- if ((++scans & 1) == 0)
- releaseWaiters(); // help others progress
- else
- Thread.yield(); // avoid starving productive threads
+ else {
+ if (rc < pc || eventWaiters != 0L)
+ helpMaintainParallelism();
+ break;
+ }
}
+ }
- try {
- if (!done)
- do {} while (!blocker.isReleasable() && !blocker.block());
- } finally {
- if (adj) {
+ /**
+ * Helps and/or blocks awaiting join of the given task.
+ * Alternates between helpJoinTask() and helpMaintainParallelism()
+ * as many times as there is a deficit in running count (or longer
+ * if running count would become zero), then blocks if task still
+ * not done.
+ *
+ * @param joinMe the task to join
+ */
+ final void awaitJoin(ForkJoinTask joinMe, ForkJoinWorkerThread worker) {
+ int threshold = parallelism; // descend blocking thresholds
+ while (joinMe.status >= 0) {
+ boolean block; int wc;
+ worker.helpJoinTask(joinMe);
+ if (joinMe.status < 0)
+ break;
+ if (((wc = workerCounts) & RUNNING_COUNT_MASK) <= threshold) {
+ if (threshold > 0)
+ --threshold;
+ else
+ advanceEventCount(); // force release
+ block = false;
+ }
+ else
+ block = UNSAFE.compareAndSwapInt(this, workerCountsOffset,
+ wc, wc - ONE_RUNNING);
+ helpMaintainParallelism();
+ if (block) {
int c;
+ joinMe.internalAwaitDone();
do {} while (!UNSAFE.compareAndSwapInt
(this, workerCountsOffset,
c = workerCounts, c + ONE_RUNNING));
+ break;
}
}
}
/**
- * Unless there are not enough other running threads, adjusts
- * counts and blocks a worker performing helpJoin that cannot find
- * any work.
- *
- * @return true if joinMe now done
+ * Same idea as awaitJoin, but no helping
*/
- final boolean tryAwaitBusyJoin(ForkJoinTask joinMe) {
- int pc = parallelism;
- outer:for (;;) {
- releaseWaiters();
- if ((workerCounts & RUNNING_COUNT_MASK) < pc) {
- ForkJoinWorkerThread[] ws = workers;
- int nws = ws.length;
- for (int i = 0; i < nws; ++i) {
- ForkJoinWorkerThread w = ws[i];
- if (w != null && w.isSuspended()) {
- if (joinMe.status < 0)
- return true;
- if ((workerCounts & RUNNING_COUNT_MASK) > pc)
- break;
- if (w.tryUnsuspend()) {
- LockSupport.unpark(w);
- break outer;
- }
- continue outer;
- }
- }
+ final void awaitBlocker(ManagedBlocker blocker)
+ throws InterruptedException {
+ int threshold = parallelism;
+ while (!blocker.isReleasable()) {
+ boolean block; int wc;
+ if (((wc = workerCounts) & RUNNING_COUNT_MASK) <= threshold) {
+ if (threshold > 0)
+ --threshold;
+ else
+ advanceEventCount();
+ block = false;
}
- if (joinMe.status < 0)
- return true;
- int wc = workerCounts;
- if ((wc & RUNNING_COUNT_MASK) <= 2 ||
- (wc >>> TOTAL_COUNT_SHIFT) < pc)
- return false; // keep this thread alive
- if (UNSAFE.compareAndSwapInt(this, workerCountsOffset,
- wc, wc - ONE_RUNNING))
+ else
+ block = UNSAFE.compareAndSwapInt(this, workerCountsOffset,
+ wc, wc - ONE_RUNNING);
+ helpMaintainParallelism();
+ if (block) {
+ try {
+ do {} while (!blocker.isReleasable() && !blocker.block());
+ } finally {
+ int c;
+ do {} while (!UNSAFE.compareAndSwapInt
+ (this, workerCountsOffset,
+ c = workerCounts, c + ONE_RUNNING));
+ }
break;
+ }
}
-
- joinMe.internalAwaitDone();
- int c;
- do {} while (!UNSAFE.compareAndSwapInt
- (this, workerCountsOffset,
- c = workerCounts, c + ONE_RUNNING));
- return true;
}
/**
@@ -1090,22 +1064,44 @@ public class ForkJoinPool extends Abstra
// Finish now if all threads terminated; else in some subsequent call
if ((workerCounts >>> TOTAL_COUNT_SHIFT) == 0) {
advanceRunLevel(TERMINATED);
- terminationLatch.countDown();
+ termination.arrive();
}
return true;
}
/**
* Actions on transition to TERMINATING
+ *
+ * Runs up to four passes through workers: (0) shutting down each
+ * quietly (without waking up if parked) to quickly spread
+ * notifications without unnecessary bouncing around event queues
+ * etc (1) wake up and help cancel tasks (2) interrupt (3) mop up
+ * races with interrupted workers
*/
private void startTerminating() {
- for (int i = 0; i < 2; ++i) { // twice to mop up newly created workers
- cancelSubmissions();
- shutdownWorkers();
- cancelWorkerTasks();
+ cancelSubmissions();
+ for (int passes = 0; passes < 4 && workerCounts != 0; ++passes) {
advanceEventCount();
- releaseWaiters();
- interruptWorkers();
+ eventWaiters = 0L; // clobber lists
+ spareWaiters = 0;
+ ForkJoinWorkerThread[] ws = workers;
+ int n = ws.length;
+ for (int i = 0; i < n; ++i) {
+ ForkJoinWorkerThread w = ws[i];
+ if (w != null) {
+ w.shutdown(true);
+ if (passes > 0 && !w.isTerminated()) {
+ w.cancelTasks();
+ LockSupport.unpark(w);
+ if (passes > 1) {
+ try {
+ w.interrupt();
+ } catch (SecurityException ignore) {
+ }
+ }
+ }
+ }
+ }
}
}
@@ -1122,50 +1118,6 @@ public class ForkJoinPool extends Abstra
}
}
- /**
- * Sets all worker run states to at least shutdown,
- * also resuming suspended workers
- */
- private void shutdownWorkers() {
- ForkJoinWorkerThread[] ws = workers;
- int nws = ws.length;
- for (int i = 0; i < nws; ++i) {
- ForkJoinWorkerThread w = ws[i];
- if (w != null)
- w.shutdown();
- }
- }
-
- /**
- * Clears out and cancels all locally queued tasks
- */
- private void cancelWorkerTasks() {
- ForkJoinWorkerThread[] ws = workers;
- int nws = ws.length;
- for (int i = 0; i < nws; ++i) {
- ForkJoinWorkerThread w = ws[i];
- if (w != null)
- w.cancelTasks();
- }
- }
-
- /**
- * Unsticks all workers blocked on joins etc
- */
- private void interruptWorkers() {
- ForkJoinWorkerThread[] ws = workers;
- int nws = ws.length;
- for (int i = 0; i < nws; ++i) {
- ForkJoinWorkerThread w = ws[i];
- if (w != null && !w.isTerminated()) {
- try {
- w.interrupt();
- } catch (SecurityException ignore) {
- }
- }
- }
- }
-
// misc support for ForkJoinWorkerThread
/**
@@ -1176,17 +1128,21 @@ public class ForkJoinPool extends Abstra
}
/**
- * Accumulates steal count from a worker, clearing
- * the worker's value
+ * Tries to accumulates steal count from a worker, clearing
+ * the worker's value.
+ *
+ * @return true if worker steal count now zero
*/
- final void accumulateStealCount(ForkJoinWorkerThread w) {
+ final boolean tryAccumulateStealCount(ForkJoinWorkerThread w) {
int sc = w.stealCount;
- if (sc != 0) {
- long c;
- w.stealCount = 0;
- do {} while (!UNSAFE.compareAndSwapLong(this, stealCountOffset,
- c = stealCount, c + sc));
+ long c = stealCount;
+ // CAS even if zero, for fence effects
+ if (UNSAFE.compareAndSwapLong(this, stealCountOffset, c, c + sc)) {
+ if (sc != 0)
+ w.stealCount = 0;
+ return true;
}
+ return sc == 0;
}
/**
@@ -1194,8 +1150,8 @@ public class ForkJoinPool extends Abstra
* active thread.
*/
final int idlePerActive() {
+ int pc = parallelism; // use parallelism, not rc
int ac = runState; // no mask -- artifically boosts during shutdown
- int pc = parallelism; // use targeted parallelism, not rc
// Use exact results for small values, saturate past 4
return pc <= ac? 0 : pc >>> 1 <= ac? 1 : pc >>> 2 <= ac? 3 : pc >>> 3;
}
@@ -1206,8 +1162,9 @@ public class ForkJoinPool extends Abstra
/**
* Creates a {@code ForkJoinPool} with parallelism equal to {@link
- * java.lang.Runtime#availableProcessors}, and using the {@linkplain
- * #defaultForkJoinWorkerThreadFactory default thread factory}.
+ * java.lang.Runtime#availableProcessors}, using the {@linkplain
+ * #defaultForkJoinWorkerThreadFactory default thread factory},
+ * no UncaughtExceptionHandler, and non-async LIFO processing mode.
*
* @throws SecurityException if a security manager exists and
* the caller is not permitted to modify threads
@@ -1216,13 +1173,14 @@ public class ForkJoinPool extends Abstra
*/
public ForkJoinPool() {
this(Runtime.getRuntime().availableProcessors(),
- defaultForkJoinWorkerThreadFactory);
+ defaultForkJoinWorkerThreadFactory, null, false);
}
/**
* Creates a {@code ForkJoinPool} with the indicated parallelism
- * level and using the {@linkplain
- * #defaultForkJoinWorkerThreadFactory default thread factory}.
+ * level, the {@linkplain
+ * #defaultForkJoinWorkerThreadFactory default thread factory},
+ * no UncaughtExceptionHandler, and non-async LIFO processing mode.
*
* @param parallelism the parallelism level
* @throws IllegalArgumentException if parallelism less than or
@@ -1233,31 +1191,25 @@ public class ForkJoinPool extends Abstra
* java.lang.RuntimePermission}{@code ("modifyThread")}
*/
public ForkJoinPool(int parallelism) {
- this(parallelism, defaultForkJoinWorkerThreadFactory);
- }
-
- /**
- * Creates a {@code ForkJoinPool} with parallelism equal to {@link
- * java.lang.Runtime#availableProcessors}, and using the given
- * thread factory.
- *
- * @param factory the factory for creating new threads
- * @throws NullPointerException if the factory is null
- * @throws SecurityException if a security manager exists and
- * the caller is not permitted to modify threads
- * because it does not hold {@link
- * java.lang.RuntimePermission}{@code ("modifyThread")}
- */
- public ForkJoinPool(ForkJoinWorkerThreadFactory factory) {
- this(Runtime.getRuntime().availableProcessors(), factory);
+ this(parallelism, defaultForkJoinWorkerThreadFactory, null, false);
}
/**
- * Creates a {@code ForkJoinPool} with the given parallelism and
- * thread factory.
+ * Creates a {@code ForkJoinPool} with the given parameters.
*
- * @param parallelism the parallelism level
- * @param factory the factory for creating new threads
+ * @param parallelism the parallelism level. For default value,
+ * use {@link java.lang.Runtime#availableProcessors}.
+ * @param factory the factory for creating new threads. For default value,
+ * use {@link #defaultForkJoinWorkerThreadFactory}.
+ * @param handler the handler for internal worker threads that
+ * terminate due to unrecoverable errors encountered while executing
+ * tasks. For default value, use A {@code ManagedBlocker} provides two methods.
- * Method {@code isReleasable} must return {@code true} if
- * blocking is not necessary. Method {@code block} blocks the
- * current thread if necessary (perhaps internally invoking
- * {@code isReleasable} before actually blocking).
+ * A {@code ManagedBlocker} provides two methods. Method
+ * {@code isReleasable} must return {@code true} if blocking is
+ * not necessary. Method {@code block} blocks the current thread
+ * if necessary (perhaps internally invoking {@code isReleasable}
+ * before actually blocking). The unusual methods in this API
+ * accommodate synchronizers that may, but don't usually, block
+ * for long periods. Similarly, they allow more efficient internal
+ * handling of cases in which additional workers may be, but
+ * usually are not, needed to ensure sufficient parallelism.
+ * Toward this end, implementations of method {@code isReleasable}
+ * must be amenable to repeated invocation.
*
* For example, here is a ManagedBlocker based on a
* ReentrantLock:
@@ -1931,6 +1743,26 @@ public class ForkJoinPool extends Abstra
* return hasLock || (hasLock = lock.tryLock());
* }
* }}
+ *
+ * Here is a class that possibly blocks waiting for an
+ * item on a given queue:
+ * If {@code maintainParallelism} is {@code true} and the pool
- * supports it ({@link #getMaintainsParallelism}), this method
- * attempts to maintain the pool's nominal parallelism. Otherwise
- * it activates a thread only if necessary to avoid complete
- * starvation. This option may be preferable when blockages use
- * timeouts, or are almost always brief.
+ * ensure sufficient parallelism while the current thread is blocked.
*
* If the caller is not a {@link ForkJoinTask}, this method is
* behaviorally equivalent to
@@ -1975,29 +1800,18 @@ public class ForkJoinPool extends Abstra
* first be expanded to ensure parallelism, and later adjusted.
*
* @param blocker the blocker
- * @param maintainParallelism if {@code true} and supported by
- * this pool, attempt to maintain the pool's nominal parallelism;
- * otherwise activate a thread only if necessary to avoid
- * complete starvation.
* @throws InterruptedException if blocker.block did so
*/
- public static void managedBlock(ManagedBlocker blocker,
- boolean maintainParallelism)
+ public static void managedBlock(ManagedBlocker blocker)
throws InterruptedException {
Thread t = Thread.currentThread();
- if (t instanceof ForkJoinWorkerThread)
- ((ForkJoinWorkerThread) t).pool.
- awaitBlocker(blocker, maintainParallelism);
- else
- awaitBlocker(blocker);
- }
-
- /**
- * Performs Non-FJ blocking
- */
- private static void awaitBlocker(ManagedBlocker blocker)
- throws InterruptedException {
- do {} while (!blocker.isReleasable() && !blocker.block());
+ if (t instanceof ForkJoinWorkerThread) {
+ ForkJoinWorkerThread w = (ForkJoinWorkerThread) t;
+ w.pool.awaitBlocker(blocker);
+ }
+ else {
+ do {} while (!blocker.isReleasable() && !blocker.block());
+ }
}
// AbstractExecutorService overrides. These rely on undocumented
@@ -2025,7 +1839,8 @@ public class ForkJoinPool extends Abstra
objectFieldOffset("eventWaiters",ForkJoinPool.class);
private static final long stealCountOffset =
objectFieldOffset("stealCount",ForkJoinPool.class);
-
+ private static final long spareWaitersOffset =
+ objectFieldOffset("spareWaiters",ForkJoinPool.class);
private static long objectFieldOffset(String field, Class klazz) {
try {
null.
+ * @param asyncMode if true,
+ * establishes local first-in-first-out scheduling mode for forked
+ * tasks that are never joined. This mode may be more appropriate
+ * than default locally stack-based mode in applications in which
+ * worker threads only process event-style asynchronous tasks.
+ * For default value, use false.
* @throws IllegalArgumentException if parallelism less than or
* equal to zero, or greater than implementation limit
* @throws NullPointerException if the factory is null
@@ -1266,22 +1218,26 @@ public class ForkJoinPool extends Abstra
* because it does not hold {@link
* java.lang.RuntimePermission}{@code ("modifyThread")}
*/
- public ForkJoinPool(int parallelism, ForkJoinWorkerThreadFactory factory) {
+ public ForkJoinPool(int parallelism,
+ ForkJoinWorkerThreadFactory factory,
+ Thread.UncaughtExceptionHandler handler,
+ boolean asyncMode) {
checkPermission();
if (factory == null)
throw new NullPointerException();
- if (parallelism <= 0 || parallelism > MAX_THREADS)
+ if (parallelism <= 0 || parallelism > MAX_WORKERS)
throw new IllegalArgumentException();
- this.poolNumber = poolNumberGenerator.incrementAndGet();
- int arraySize = initialArraySizeFor(parallelism);
this.parallelism = parallelism;
this.factory = factory;
- this.maxPoolSize = MAX_THREADS;
- this.maintainsParallelism = true;
+ this.ueh = handler;
+ this.locallyFifo = asyncMode;
+ int arraySize = initialArraySizeFor(parallelism);
this.workers = new ForkJoinWorkerThread[arraySize];
this.submissionQueue = new LinkedTransferQueue {@code
+ * class QueueTaker
*/
public static interface ManagedBlocker {
/**
@@ -1954,14 +1786,7 @@ public class ForkJoinPool extends Abstra
* Blocks in accord with the given blocker. If the current thread
* is a {@link ForkJoinWorkerThread}, this method possibly
* arranges for a spare thread to be activated if necessary to
- * ensure parallelism while the current thread is blocked.
- *
- *