As is the case with other ExecutorServices, there are three - * main task execution methods summarized in the follwoing + * 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 @@ -60,11 +60,7 @@ import java.util.concurrent.CountDownLat * 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. To avoid inadvertant - * cyclic task dependencies and to improve performance, task - * submissions to the current pool by an ongoing fork/join - * computations may be implicitly translated to the corresponding - * ForkJoinTask forms. + * within-computation forms listed in the table. * *
| Call from within fork/join computations | *|||
| Arange async execution | + *Arrange async execution | *{@link #execute(ForkJoinTask)} | *{@link ForkJoinTask#fork} | *{@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 @@ -113,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 exhausted.
*
* @since 1.7
* @author Doug Lea
@@ -140,21 +137,61 @@ 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. Because 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
+ * re-activate a spare thread to compensate for blocked
+ * joiners until they unblock.
+ *
+ * It is impossible to keep exactly the target (parallelism)
+ * number of threads running at any given time. 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, so we rely on multiple retries of each. Compensation
+ * occurs in slow-motion. It is triggered only upon timeouts of
+ * Object.wait used for joins. This reduces poor decisions that
+ * would otherwise be made when threads are waiting for others
+ * that are stalled because of unrelated activities such as
+ * garbage collection.
+ *
+ * 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
@@ -170,28 +207,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
* 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
+ * that are neither blocked nor artificially 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 performing an alternative
- * action on contention like releasing waiters or locating spares.
+ * to create, resume or suspend. Note however that the
+ * correspondence 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
@@ -220,7 +261,7 @@ 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
+ * mechanism for configuration and 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
@@ -231,69 +272,64 @@ public class ForkJoinPool extends Abstra
* 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
- * increment field eventCount are tolerated without retries.
+ * tags to avoid ABA errors in Treiber stacks. Upon any wakeup,
+ * released threads also try to release at most two others. 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 in signalWork.
* Conceptually they are merged into the same event, which is OK
* when their only purpose is to enable workers to scan for work.
*
- * 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
- * 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
- * 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.
- *
- * 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. Instead, we
- * use a combination of heuristics that adds threads only when the
- * pool appears to be approaching starvation. These effectively
- * reduce 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
+ * 5. Managing suspension of extra workers. When a worker notices
+ * (usually upon timeout of a wait()) that there are too few
+ * running threads, we may create a new thread to maintain
+ * parallelism level, or at least 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.
+ * 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.
+ *
+ * 6. Killing off unneeded workers. A timeout mechanism is used to
+ * shed unused workers: The oldest (first) event queue waiter uses
+ * a timed rather than hard wait. When this wait times out without
+ * a normal wakeup, it tries to shutdown any one (for convenience
+ * the newest) other spare or event waiter via
+ * tryShutdownUnusedWorker. This eventually reduces the number of
+ * worker threads to a minimum of one after a long enough period
+ * without use.
+ *
+ * 7. Deciding when to create new workers. The main dynamic
+ * 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 impossible
+ * 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
+ * presence of transient 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 parallelism to
+ * lapse for a while during joins, and use a timeout to see if,
+ * after the resulting settling, there is still a need for
+ * additional workers. This 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 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.
*
@@ -308,11 +344,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)
@@ -380,10 +418,32 @@ 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.
+ * The time to block in a join (see awaitJoin) before checking if
+ * a new worker should be (re)started to maintain parallelism
+ * level. The value should be short enough to maintain global
+ * responsiveness and progress but long enough to avoid
+ * counterproductive firings during GC stalls or unrelated system
+ * activity, and to not bog down systems with continual re-firings
+ * on GCs or legitimately long waits.
+ */
+ private static final long JOIN_TIMEOUT_MILLIS = 250L; // 4 per second
+
+ /**
+ * The wakeup interval (in nanoseconds) for the oldest worker
+ * worker waiting for an event invokes tryShutdownUnusedWorker to shrink
+ * the number of workers. The exact value does not matter too
+ * much, but should be long enough to slowly release resources
+ * during long periods without use without disrupting normal use.
*/
- private static final int MAX_THREADS = 0x7fff;
+ private static final long SHRINK_RATE_NANOS =
+ 30L * 1000L * 1000L * 1000L; // 2 per minute
+
+ /**
+ * 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_WORKERS = 0x7fff;
/**
* Array holding all worker threads in the pool. Array size must
@@ -423,25 +483,36 @@ public class ForkJoinPool extends Abstra
private volatile long stealCount;
/**
- * Encoded record of top of treiber stack of threads waiting for
+ * 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
@@ -452,8 +523,12 @@ public class ForkJoinPool extends Abstra
* These are bundled together to ensure consistent read for
* termination checks (i.e., that runLevel is at least SHUTDOWN
* and active threads is zero).
+ *
+ * Notes: Most direct CASes are dependent on these bitfield
+ * positions. Also, this field is non-private to enable direct
+ * performance-sensitive CASes in ForkJoinWorkerThread.
*/
- private volatile int runState;
+ volatile int runState;
// Note: The order among run level values matters.
private static final int RUNLEVEL_SHIFT = 16;
@@ -461,7 +536,6 @@ public class ForkJoinPool extends Abstra
private static final int TERMINATING = 1 << (RUNLEVEL_SHIFT + 1);
private static final int TERMINATED = 1 << (RUNLEVEL_SHIFT + 2);
private static final int ACTIVE_COUNT_MASK = (1 << RUNLEVEL_SHIFT) - 1;
- private static final int ONE_ACTIVE = 1; // active update delta
/**
* Holds number of total (i.e., created and not yet terminated)
@@ -470,8 +544,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;
@@ -503,18 +576,19 @@ public class ForkJoinPool extends Abstra
*/
private final int poolNumber;
- // utilities for updating fields
+ // Utilities for CASing fields. Note that most of these
+ // are usually manually inlined by callers
/**
- * Increments running count. Also used by ForkJoinTask.
+ * Increments running count part of workerCounts
*/
final void incrementRunningCount() {
int c;
do {} while (!UNSAFE.compareAndSwapInt(this, workerCountsOffset,
- c = workerCounts,
+ c = workerCounts,
c + ONE_RUNNING));
}
-
+
/**
* Tries to decrement running count unless already zero
*/
@@ -527,15 +601,25 @@ public class ForkJoinPool extends Abstra
}
/**
- * Tries incrementing active count; fails on contention.
- * Called by workers before executing tasks.
+ * Forces decrement of encoded workerCounts, awaiting nonzero if
+ * (rarely) necessary when other count updates lag.
*
- * @return true on success
+ * @param dr -- either zero or ONE_RUNNING
+ * @param dt == either zero or ONE_TOTAL
*/
- final boolean tryIncrementActiveCount() {
- int c;
- return UNSAFE.compareAndSwapInt(this, runStateOffset,
- c = runState, c + ONE_ACTIVE);
+ private void decrementWorkerCounts(int dr, int dt) {
+ for (;;) {
+ int wc = workerCounts;
+ if ((wc & RUNNING_COUNT_MASK) - dr < 0 ||
+ (wc >>> TOTAL_COUNT_SHIFT) - dt < 0) {
+ if ((runState & TERMINATED) != 0)
+ return; // lagging termination on a backout
+ Thread.yield();
+ }
+ if (UNSAFE.compareAndSwapInt(this, workerCountsOffset,
+ wc, wc - (dr + dt)))
+ return;
+ }
}
/**
@@ -545,7 +629,7 @@ public class ForkJoinPool extends Abstra
final boolean tryDecrementActiveCount() {
int c;
return UNSAFE.compareAndSwapInt(this, runStateOffset,
- c = runState, c - ONE_ACTIVE);
+ c = runState, c - 1);
}
/**
@@ -574,12 +658,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
@@ -594,7 +678,7 @@ public class ForkJoinPool extends Abstra
*/
private void forgetWorker(ForkJoinWorkerThread w) {
int idx = w.poolIndex;
- // Locking helps method recordWorker avoid unecessary expansion
+ // Locking helps method recordWorker avoid unnecessary expansion
final ReentrantLock lock = this.workerLock;
lock.lock();
try {
@@ -606,378 +690,379 @@ public class ForkJoinPool extends Abstra
}
}
- // adding and removing workers
-
/**
- * Tries to create and add new worker. Assumes that worker counts
- * are already updated to accommodate the worker, so adjusts on
- * failure.
+ * Final callback from terminating worker. Removes record of
+ * worker from array, and adjusts counts. If pool is shutting
+ * down, tries to complete termination.
*
- * @return new worker or null if creation failed
+ * @param w the worker
*/
- private ForkJoinWorkerThread addWorker() {
- ForkJoinWorkerThread w = null;
- try {
- w = factory.newThread(this);
- } finally { // Adjust on either null or exceptional factory return
- if (w == null) {
- onWorkerCreationFailure();
- return null;
- }
- }
- w.start(recordWorker(w), ueh);
- return w;
+ final void workerTerminated(ForkJoinWorkerThread w) {
+ forgetWorker(w);
+ decrementWorkerCounts(w.isTrimmed()? 0 : ONE_RUNNING, ONE_TOTAL);
+ while (w.stealCount != 0) // collect final count
+ tryAccumulateStealCount(w);
+ tryTerminate(false);
}
+ // Waiting for and signalling events
+
/**
- * Adjusts counts upon failure to create worker
+ * 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. Gives up upon a change in count or
+ * upon releasing two workers, letting others take over.
*/
- private void onWorkerCreationFailure() {
- for (;;) {
- int wc = workerCounts;
- if ((wc >>> TOTAL_COUNT_SHIFT) > 0 &&
- UNSAFE.compareAndSwapInt(this, workerCountsOffset,
- wc, wc - (ONE_RUNNING|ONE_TOTAL)))
+ private void releaseEventWaiters() {
+ ForkJoinWorkerThread[] ws = workers;
+ int n = ws.length;
+ long h = eventWaiters;
+ int ec = eventCount;
+ boolean releasedOne = false;
+ ForkJoinWorkerThread w; int id;
+ while ((id = ((int)(h & WAITER_ID_MASK)) - 1) >= 0 &&
+ (int)(h >>> EVENT_COUNT_SHIFT) != ec &&
+ id < n && (w = ws[id]) != null) {
+ if (UNSAFE.compareAndSwapLong(this, eventWaitersOffset,
+ h, w.nextWaiter)) {
+ LockSupport.unpark(w);
+ if (releasedOne) // exit on second release
+ break;
+ releasedOne = true;
+ }
+ if (eventCount != ec)
break;
+ h = eventWaiters;
}
- tryTerminate(false); // in case of failure during shutdown
}
/**
- * Create enough total workers to establish target parallelism,
- * giving up if terminating or addWorker fails
+ * Tries to advance eventCount and releases waiters. Called only
+ * from workers.
*/
- 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;
- }
+ final void signalWork() {
+ int c; // try to increment event count -- CAS failure OK
+ UNSAFE.compareAndSwapInt(this, eventCountOffset, c = eventCount, c+1);
+ if (eventWaiters != 0L)
+ releaseEventWaiters();
}
/**
- * 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.
+ * Adds the given worker to event queue and blocks until
+ * terminating or event count advances from the given value
*
- * @param w the worker
+ * @param w the calling worker thread
+ * @param ec the count
*/
- 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))
+ private void eventSync(ForkJoinWorkerThread w, int ec) {
+ long nh = (((long)ec) << EVENT_COUNT_SHIFT) | ((long)(w.poolIndex+1));
+ long h;
+ while ((runState < SHUTDOWN || !tryTerminate(false)) &&
+ (((int)((h = eventWaiters) & WAITER_ID_MASK)) == 0 ||
+ (int)(h >>> EVENT_COUNT_SHIFT) == ec) &&
+ eventCount == ec) {
+ if (UNSAFE.compareAndSwapLong(this, eventWaitersOffset,
+ w.nextWaiter = h, nh)) {
+ awaitEvent(w, ec);
break;
+ }
}
-
- accumulateStealCount(w); // collect final count
- if (!tryTerminate(false))
- ensureEnoughTotalWorkers();
}
- // Waiting for and signalling events
-
/**
- * Releases workers blocked on a count not equal to current count.
+ * Blocks the given worker (that has already been entered as an
+ * event waiter) until terminating or event count advances from
+ * the given value. The oldest (first) waiter uses a timed wait to
+ * occasionally one-by-one shrink the number of workers (to a
+ * minimum of one) if the pool has not been used for extended
+ * periods.
+ *
+ * @param w the calling worker thread
+ * @param ec the count
*/
- private 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))
- LockSupport.unpark(w);
+ private void awaitEvent(ForkJoinWorkerThread w, int ec) {
+ while (eventCount == ec) {
+ if (tryAccumulateStealCount(w)) { // transfer while idle
+ boolean untimed = (w.nextWaiter != 0L ||
+ (workerCounts & RUNNING_COUNT_MASK) <= 1);
+ long startTime = untimed? 0 : System.nanoTime();
+ Thread.interrupted(); // clear/ignore interrupt
+ if (eventCount != ec || w.runState != 0 ||
+ runState >= TERMINATING) // recheck after clear
+ break;
+ if (untimed)
+ LockSupport.park(w);
+ else {
+ LockSupport.parkNanos(w, SHRINK_RATE_NANOS);
+ if (eventCount != ec || w.runState != 0 ||
+ runState >= TERMINATING)
+ break;
+ if (System.nanoTime() - startTime >= SHRINK_RATE_NANOS)
+ tryShutdownUnusedWorker(ec);
+ }
+ }
}
}
+ // Maintaining parallelism
+
/**
- * Ensures eventCount on exit is different (mod 2^32) than on
- * entry and wakes up all waiters
+ * Pushes worker onto the spare stack
*/
- private void signalEvent() {
- int c;
- do {} while (!UNSAFE.compareAndSwapInt(this, eventCountOffset,
- c = eventCount, c+1));
- releaseWaiters();
+ 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));
}
/**
- * Advances eventCount and releases waiters until interference by
- * other releasing threads is detected.
+ * Tries (once) to resume a spare if the number of running
+ * threads is less than target.
*/
- final void signalWork() {
- // EventCount CAS failures are OK -- any change in count suffices.
- 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
+ private void tryResumeSpare() {
+ int sw, id;
+ ForkJoinWorkerThread[] ws = workers;
+ int n = ws.length;
+ ForkJoinWorkerThread w;
+ if ((sw = spareWaiters) != 0 &&
+ (id = (sw & SPARE_ID_MASK) - 1) >= 0 &&
+ id < n && (w = ws[id]) != null &&
+ (workerCounts & RUNNING_COUNT_MASK) < parallelism &&
+ spareWaiters == sw &&
+ UNSAFE.compareAndSwapInt(this, spareWaitersOffset,
+ sw, w.nextSpare)) {
+ int c; // increment running count before resume
+ do {} while(!UNSAFE.compareAndSwapInt
+ (this, workerCountsOffset,
+ c = workerCounts, c + ONE_RUNNING));
+ if (w.tryUnsuspend())
LockSupport.unpark(w);
- if (top != eventWaiters) // let someone else take over
- return;
- }
+ else // back out if w was shutdown
+ decrementWorkerCounts(ONE_RUNNING, 0);
}
}
/**
- * If worker is inactive, blocks until terminating or event count
- * advances from last value held by worker; in any case helps
- * release others.
- *
- * @param w the calling worker thread
+ * Tries to increase the number of running workers if below target
+ * parallelism: If a spare exists tries to resume it via
+ * tryResumeSpare. Otherwise, if not enough total workers or all
+ * existing workers are busy, adds a new worker. In all cases also
+ * helps wake up releasable workers waiting for work.
*/
- 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();
+ private void helpMaintainParallelism() {
+ int pc = parallelism;
+ int wc, rs, tc;
+ while (((wc = workerCounts) & RUNNING_COUNT_MASK) < pc &&
+ (rs = runState) < TERMINATING) {
+ if (spareWaiters != 0)
+ tryResumeSpare();
+ else if ((tc = wc >>> TOTAL_COUNT_SHIFT) >= MAX_WORKERS ||
+ (tc >= pc && (rs & ACTIVE_COUNT_MASK) != tc))
+ break; // enough total
+ else if (runState == rs && workerCounts == wc &&
+ UNSAFE.compareAndSwapInt(this, workerCountsOffset, wc,
+ wc + (ONE_RUNNING|ONE_TOTAL))) {
+ ForkJoinWorkerThread w = null;
+ try {
+ w = factory.newThread(this);
+ } finally { // adjust on null or exceptional factory return
+ if (w == null) {
+ decrementWorkerCounts(ONE_RUNNING, ONE_TOTAL);
+ tryTerminate(false); // handle failure during shutdown
+ }
+ }
+ if (w == null)
break;
+ w.start(recordWorker(w), ueh);
+ if ((workerCounts >>> TOTAL_COUNT_SHIFT) >= pc) {
+ int c; // advance event count
+ UNSAFE.compareAndSwapInt(this, eventCountOffset,
+ c = eventCount, c+1);
+ break; // add at most one unless total below target
}
}
- w.lastEventCount = eventCount;
}
- releaseWaiters();
+ if (eventWaiters != 0L)
+ releaseEventWaiters();
+ }
+
+ /**
+ * Callback from the oldest waiter in awaitEvent waking up after a
+ * period of non-use. If all workers are idle, tries (once) to
+ * shutdown an event waiter or a spare, if one exists. Note that
+ * we don't need CAS or locks here because the method is called
+ * only from one thread occasionally waking (and even misfires are
+ * OK). Note that until the shutdown worker fully terminates,
+ * workerCounts will overestimate total count, which is tolerable.
+ *
+ * @param ec the event count waited on by caller (to abort
+ * attempt if count has since changed).
+ */
+ private void tryShutdownUnusedWorker(int ec) {
+ if (runState == 0 && eventCount == ec) { // only trigger if all idle
+ ForkJoinWorkerThread[] ws = workers;
+ int n = ws.length;
+ ForkJoinWorkerThread w = null;
+ boolean shutdown = false;
+ int sw;
+ long h;
+ if ((sw = spareWaiters) != 0) { // prefer killing spares
+ int id = (sw & SPARE_ID_MASK) - 1;
+ if (id >= 0 && id < n && (w = ws[id]) != null &&
+ UNSAFE.compareAndSwapInt(this, spareWaitersOffset,
+ sw, w.nextSpare))
+ shutdown = true;
+ }
+ else if ((h = eventWaiters) != 0L) {
+ long nh;
+ int id = ((int)(h & WAITER_ID_MASK)) - 1;
+ if (id >= 0 && id < n && (w = ws[id]) != null &&
+ (nh = w.nextWaiter) != 0L && // keep at least one worker
+ UNSAFE.compareAndSwapLong(this, eventWaitersOffset, h, nh))
+ shutdown = true;
+ }
+ if (w != null && shutdown) {
+ w.shutdown();
+ LockSupport.unpark(w);
+ }
+ }
+ releaseEventWaiters(); // in case of interference
}
/**
* 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:
+ * stealing a task or taking a submission and running it).
+ * Performs one or more 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.
+ * 1. If the worker is active and either did not run a task
+ * or there are too many workers, try to set its active status
+ * to inactive and update activeCount. On contention, we may
+ * try again in this or a subsequent call.
+ *
+ * 2. If not enough total workers, help create some.
+ *
+ * 3. If there are too many running workers, suspend this worker
+ * (first forcing inactive if necessary). If it is not needed,
+ * it may be shutdown while suspended (via
+ * tryShutdownUnusedWorker). Otherwise, upon resume it
+ * rechecks running thread count and need for event sync.
+ *
+ * 4. If worker did not run a task, await the next task event via
+ * eventSync if necessary (first forcing inactivation), upon
+ * which the worker may be shutdown via
+ * tryShutdownUnusedWorker. Otherwise, help release any
+ * existing event waiters that are now releasable,
*
* @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).
+ * @param ran true if worker ran a task since last call to this method
*/
- final void preStep(ForkJoinWorkerThread w, boolean worked) {
+ final void preStep(ForkJoinWorkerThread w, boolean ran) {
+ int wec = w.lastEventCount;
boolean active = w.active;
- boolean inactivate = !worked & active;
- for (;;) {
- if (inactivate) {
- int rs = runState;
- if (UNSAFE.compareAndSwapInt(this, runStateOffset,
- rs, rs - ONE_ACTIVE))
- inactivate = active = w.active = false;
- }
+ boolean inactivate = false;
+ int pc = parallelism;
+ int rs;
+ while (w.runState == 0 && (rs = runState) < TERMINATING) {
+ if ((inactivate || (active && (rs & ACTIVE_COUNT_MASK) >= pc)) &&
+ UNSAFE.compareAndSwapInt(this, runStateOffset, rs, rs - 1))
+ inactivate = active = w.active = false;
int wc = workerCounts;
- if ((wc & RUNNING_COUNT_MASK) <= parallelism) {
- if (!worked)
- eventSync(w);
- return;
+ if ((wc & RUNNING_COUNT_MASK) > pc) {
+ if (!(inactivate |= active) && // must inactivate to suspend
+ workerCounts == wc && // try to suspend as spare
+ UNSAFE.compareAndSwapInt(this, workerCountsOffset,
+ wc, wc - ONE_RUNNING))
+ w.suspendAsSpare();
}
- if (!(inactivate |= active) && // must inactivate to suspend
- UNSAFE.compareAndSwapInt(this, workerCountsOffset,
- wc, wc - ONE_RUNNING) &&
- !w.suspendAsSpare()) // false if trimmed
- return;
+ else if ((wc >>> TOTAL_COUNT_SHIFT) < pc)
+ helpMaintainParallelism(); // not enough workers
+ else if (!ran) {
+ long h = eventWaiters;
+ int ec = eventCount;
+ if (h != 0L && (int)(h >>> EVENT_COUNT_SHIFT) != ec)
+ releaseEventWaiters(); // release others before waiting
+ else if (ec != wec) {
+ w.lastEventCount = ec; // no need to wait
+ break;
+ }
+ else if (!(inactivate |= active))
+ eventSync(w, wec); // must inactivate before sync
+ }
+ else
+ break;
}
}
/**
- * Tries to decrement running count, and if so, possibly creates
- * or resumes compensating threads before blocking on task joinMe.
- * This code is sprawled out with manual inlining to evade some
- * JIT oddities.
+ * Helps and/or blocks awaiting join of the given task.
+ * See above for explanation.
*
* @param joinMe the task to join
- * @return task status on exit
+ * @param worker the current worker thread
*/
- final int tryAwaitJoin(ForkJoinTask joinMe) {
- int cw = workerCounts; // read now to spoil CAS if counts change as ...
- releaseWaiters(); // ... a byproduct of releaseWaiters
- int stat = joinMe.status;
- if (stat >= 0 && // inline variant of tryDecrementRunningCount
- (cw & RUNNING_COUNT_MASK) > 0 &&
- UNSAFE.compareAndSwapInt(this, workerCountsOffset,
- cw, cw - ONE_RUNNING)) {
- int pc = parallelism;
- int scans = 0; // to require confirming passes to add threads
- outer: while ((workerCounts & RUNNING_COUNT_MASK) < pc) {
- if ((stat = joinMe.status) < 0)
- break;
- ForkJoinWorkerThread spare = null;
- 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 ((stat = joinMe.status) < 0) // recheck to narrow race
- break;
- int wc = workerCounts;
- int rc = wc & RUNNING_COUNT_MASK;
- if (rc >= pc)
- break;
- if (spare != null) {
- if (spare.tryUnsuspend()) {
- int c; // inline incrementRunningCount
- do {} while (!UNSAFE.compareAndSwapInt
- (this, workerCountsOffset,
- c = workerCounts, c + ONE_RUNNING));
- LockSupport.unpark(spare);
- break;
- }
- continue;
- }
- int tc = wc >>> TOTAL_COUNT_SHIFT;
- int sc = tc - pc;
- if (rc > 0) {
- int p = pc;
- int s = sc;
- while (s-- >= 0) { // try keeping 3/4 live
- if (rc > (p -= (p >>> 2) + 1))
- break outer;
- }
- }
- if (scans++ > sc && tc < MAX_THREADS &&
- UNSAFE.compareAndSwapInt(this, workerCountsOffset, wc,
- wc + (ONE_RUNNING|ONE_TOTAL))) {
- addWorker();
- break;
- }
+ final void awaitJoin(ForkJoinTask joinMe, ForkJoinWorkerThread worker) {
+ int retries = 2 + (parallelism >> 2); // #helpJoins before blocking
+ while (joinMe.status >= 0) {
+ int wc;
+ worker.helpJoinTask(joinMe);
+ if (joinMe.status < 0)
+ break;
+ else if (retries > 0)
+ --retries;
+ else if (((wc = workerCounts) & RUNNING_COUNT_MASK) != 0 &&
+ UNSAFE.compareAndSwapInt(this, workerCountsOffset,
+ wc, wc - ONE_RUNNING)) {
+ int stat, c; long h;
+ while ((stat = joinMe.status) >= 0 &&
+ (h = eventWaiters) != 0L && // help release others
+ (int)(h >>> EVENT_COUNT_SHIFT) != eventCount)
+ releaseEventWaiters();
+ if (stat >= 0 &&
+ ((workerCounts & RUNNING_COUNT_MASK) == 0 ||
+ (stat =
+ joinMe.internalAwaitDone(JOIN_TIMEOUT_MILLIS)) >= 0))
+ helpMaintainParallelism(); // timeout or no running workers
+ do {} while (!UNSAFE.compareAndSwapInt
+ (this, workerCountsOffset,
+ c = workerCounts, c + ONE_RUNNING));
+ if (stat < 0)
+ break; // else restart
}
- if (stat >= 0)
- stat = joinMe.internalAwaitDone();
- int c; // inline incrementRunningCount
- do {} while (!UNSAFE.compareAndSwapInt
- (this, workerCountsOffset,
- c = workerCounts, c + ONE_RUNNING));
}
- return stat;
}
/**
- * Same idea as (and mostly pasted from) tryAwaitJoin, but
- * self-contained
+ * Same idea as awaitJoin, but no helping, retries, or timeouts.
*/
final void awaitBlocker(ManagedBlocker blocker)
throws InterruptedException {
- for (;;) {
- if (blocker.isReleasable())
- return;
- int cw = workerCounts;
- releaseWaiters();
- if ((cw & RUNNING_COUNT_MASK) > 0 &&
- UNSAFE.compareAndSwapInt(this, workerCountsOffset,
- cw, cw - ONE_RUNNING))
- break;
- }
- boolean done = false;
- int pc = parallelism;
- int scans = 0;
- outer: while ((workerCounts & RUNNING_COUNT_MASK) < pc) {
- if (done = blocker.isReleasable())
- break;
- ForkJoinWorkerThread spare = null;
- 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;
+ while (!blocker.isReleasable()) {
int wc = workerCounts;
- int rc = wc & RUNNING_COUNT_MASK;
- if (rc >= pc)
- break;
- if (spare != null) {
- if (spare.tryUnsuspend()) {
+ if ((wc & RUNNING_COUNT_MASK) != 0 &&
+ UNSAFE.compareAndSwapInt(this, workerCountsOffset,
+ wc, wc - ONE_RUNNING)) {
+ try {
+ while (!blocker.isReleasable()) {
+ long h = eventWaiters;
+ if (h != 0L &&
+ (int)(h >>> EVENT_COUNT_SHIFT) != eventCount)
+ releaseEventWaiters();
+ else if ((workerCounts & RUNNING_COUNT_MASK) == 0 &&
+ runState < TERMINATING)
+ helpMaintainParallelism();
+ else if (blocker.block())
+ break;
+ }
+ } finally {
int c;
do {} while (!UNSAFE.compareAndSwapInt
(this, workerCountsOffset,
c = workerCounts, c + ONE_RUNNING));
- LockSupport.unpark(spare);
- break;
- }
- continue;
- }
- int tc = wc >>> TOTAL_COUNT_SHIFT;
- int sc = tc - pc;
- if (rc > 0) {
- int p = pc;
- int s = sc;
- while (s-- >= 0) {
- if (rc > (p -= (p >>> 2) + 1))
- break outer;
}
- }
- if (scans++ > sc && tc < MAX_THREADS &&
- UNSAFE.compareAndSwapInt(this, workerCountsOffset, wc,
- wc + (ONE_RUNNING|ONE_TOTAL))) {
- addWorker();
break;
}
}
- try {
- if (!done)
- do {} while (!blocker.isReleasable() &&
- !blocker.block());
- } finally {
- int c;
- do {} while (!UNSAFE.compareAndSwapInt
- (this, workerCountsOffset,
- c = workerCounts, c + ONE_RUNNING));
- }
- }
+ }
/**
* Possibly initiates and/or completes termination.
@@ -1007,14 +1092,39 @@ public class ForkJoinPool extends Abstra
/**
* Actions on transition to TERMINATING
+ *
+ * Runs up to four passes through workers: (0) shutting down each
+ * (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();
- signalEvent();
- interruptWorkers();
+ cancelSubmissions();
+ for (int passes = 0; passes < 4 && workerCounts != 0; ++passes) {
+ int c; // advance event count
+ UNSAFE.compareAndSwapInt(this, eventCountOffset,
+ c = eventCount, c+1);
+ 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();
+ if (passes > 0 && !w.isTerminated()) {
+ w.cancelTasks();
+ LockSupport.unpark(w);
+ if (passes > 1) {
+ try {
+ w.interrupt();
+ } catch (SecurityException ignore) {
+ }
+ }
+ }
+ }
+ }
}
}
@@ -1031,50 +1141,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
/**
@@ -1085,17 +1151,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;
}
/**
@@ -1103,8 +1173,8 @@ public class ForkJoinPool extends Abstra
* active thread.
*/
final int idlePerActive() {
- int pc = parallelism; // use targeted parallelism, not rc
- int ac = runState; // no mask -- artifically boosts during shutdown
+ int pc = parallelism; // use parallelism, not rc
+ int ac = runState; // no mask -- artificially boosts during shutdown
// Use exact results for small values, saturate past 4
return pc <= ac? 0 : pc >>> 1 <= ac? 1 : pc >>> 2 <= ac? 3 : pc >>> 3;
}
@@ -1154,10 +1224,10 @@ public class ForkJoinPool extends Abstra
* 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
+ * @param handler the handler for internal worker threads that
+ * terminate due to unrecoverable errors encountered while executing
* tasks. For default value, use null.
- * @param asyncMode if true,
+ * @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
@@ -1171,14 +1241,14 @@ public class ForkJoinPool extends Abstra
* because it does not hold {@link
* java.lang.RuntimePermission}{@code ("modifyThread")}
*/
- public ForkJoinPool(int parallelism,
+ 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.parallelism = parallelism;
this.factory = factory;
@@ -1197,8 +1267,9 @@ public class ForkJoinPool extends Abstra
* @param pc the initial parallelism level
*/
private static int initialArraySizeFor(int pc) {
- // See Hackers Delight, sec 3.2. We know MAX_THREADS < (1 >>> 16)
- int size = pc < MAX_THREADS ? pc + 1 : MAX_THREADS;
+ // If possible, initially allocate enough space for one spare
+ int size = pc < MAX_WORKERS ? pc + 1 : MAX_WORKERS;
+ // See Hackers Delight, sec 3.2. We know MAX_WORKERS < (1 >>> 16)
size |= size >>> 1;
size |= size >>> 2;
size |= size >>> 4;
@@ -1216,24 +1287,14 @@ public class ForkJoinPool extends Abstra
throw new NullPointerException();
if (runState >= SHUTDOWN)
throw new RejectedExecutionException();
- // Convert submissions to current pool into forks
- Thread t = Thread.currentThread();
- ForkJoinWorkerThread w;
- if ((t instanceof ForkJoinWorkerThread) &&
- (w = (ForkJoinWorkerThread) t).pool == this)
- w.pushTask(task);
- else {
- submissionQueue.offer(task);
- signalEvent();
- ensureEnoughTotalWorkers();
- }
+ submissionQueue.offer(task);
+ int c; // try to increment event count -- CAS failure OK
+ UNSAFE.compareAndSwapInt(this, eventCountOffset, c = eventCount, c+1);
+ helpMaintainParallelism(); // create, start, or resume some workers
}
/**
* Performs the given task, returning its result upon completion.
- * If the caller is already engaged in a fork/join computation in
- * the current pool, this method is equivalent in effect to
- * {@link ForkJoinTask#invoke}.
*
* @param task the task
* @return the task's result
@@ -1248,9 +1309,6 @@ public class ForkJoinPool extends Abstra
/**
* Arranges for (asynchronous) execution of the given task.
- * If the caller is already engaged in a fork/join computation in
- * the current pool, this method is equivalent in effect to
- * {@link ForkJoinTask#fork}.
*
* @param task the task
* @throws NullPointerException if the task is null
@@ -1279,9 +1337,6 @@ public class ForkJoinPool extends Abstra
/**
* Submits a ForkJoinTask for execution.
- * If the caller is already engaged in a fork/join computation in
- * the current pool, this method is equivalent in effect to
- * {@link ForkJoinTask#fork}.
*
* @param task the task to submit
* @return the task
@@ -1473,8 +1528,8 @@ public class ForkJoinPool extends Abstra
public long getQueuedTaskCount() {
long count = 0;
ForkJoinWorkerThread[] ws = workers;
- int nws = ws.length;
- for (int i = 0; i < nws; ++i) {
+ int n = ws.length;
+ for (int i = 0; i < n; ++i) {
ForkJoinWorkerThread w = ws[i];
if (w != null)
count += w.getQueueSize();
@@ -1532,29 +1587,13 @@ public class ForkJoinPool extends Abstra
* @return the number of elements transferred
*/
protected int drainTasksTo(Collection> c) {
- int n = submissionQueue.drainTo(c);
- ForkJoinWorkerThread[] ws = workers;
- int nws = ws.length;
- for (int i = 0; i < nws; ++i) {
- ForkJoinWorkerThread w = ws[i];
- if (w != null)
- n += w.drainTasksTo(c);
- }
- return n;
- }
-
- /**
- * Returns count of total parks by existing workers.
- * Used during development only since not meaningful to users.
- */
- private int collectParkCount() {
- int count = 0;
+ int count = submissionQueue.drainTo(c);
ForkJoinWorkerThread[] ws = workers;
- int nws = ws.length;
- for (int i = 0; i < nws; ++i) {
+ int n = ws.length;
+ for (int i = 0; i < n; ++i) {
ForkJoinWorkerThread w = ws[i];
if (w != null)
- count += w.parkCount;
+ count += w.drainTasksTo(c);
}
return count;
}
@@ -1576,7 +1615,6 @@ public class ForkJoinPool extends Abstra
int pc = parallelism;
int rs = runState;
int ac = rs & ACTIVE_COUNT_MASK;
- // int pk = collectParkCount();
return super.toString() +
"[" + runLevelToString(rs) +
", parallelism = " + pc +
@@ -1586,7 +1624,6 @@ public class ForkJoinPool extends Abstra
", steals = " + st +
", tasks = " + qt +
", submissions = " + qs +
- // ", parks = " + pk +
"]";
}
@@ -1693,11 +1730,17 @@ public class ForkJoinPool extends Abstra
* Interface for extending managed parallelism for tasks running
* in {@link ForkJoinPool}s.
*
- *
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: @@ -1715,6 +1758,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: + *
{@code
+ * class QueueTaker implements ManagedBlocker {
+ * final BlockingQueue queue;
+ * volatile E item = null;
+ * QueueTaker(BlockingQueue q) { this.queue = q; }
+ * public boolean block() throws InterruptedException {
+ * if (item == null)
+ * item = queue.take();
+ * return true;
+ * }
+ * public boolean isReleasable() {
+ * return item != null || (item = queue.poll()) != null;
+ * }
+ * public E getItem() { // call after pool.managedBlock completes
+ * return item;
+ * }
+ * }}
*/
public static interface ManagedBlocker {
/**
@@ -1757,8 +1820,10 @@ public class ForkJoinPool extends Abstra
public static void managedBlock(ManagedBlocker blocker)
throws InterruptedException {
Thread t = Thread.currentThread();
- if (t instanceof ForkJoinWorkerThread)
- ((ForkJoinWorkerThread) t).pool.awaitBlocker(blocker);
+ if (t instanceof ForkJoinWorkerThread) {
+ ForkJoinWorkerThread w = (ForkJoinWorkerThread) t;
+ w.pool.awaitBlocker(blocker);
+ }
else {
do {} while (!blocker.isReleasable() && !blocker.block());
}
@@ -1789,6 +1854,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 {