--- jsr166/src/jsr166e/ForkJoinPool.java 2012/08/13 15:52:33 1.1
+++ jsr166/src/jsr166e/ForkJoinPool.java 2013/07/19 19:34:43 1.62
@@ -5,12 +5,13 @@
*/
package jsr166e;
+
+import java.lang.Thread.UncaughtExceptionHandler;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
import java.util.List;
-import java.util.Random;
import java.util.concurrent.AbstractExecutorService;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
@@ -18,10 +19,6 @@ import java.util.concurrent.Future;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.RunnableFuture;
import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicInteger;
-import java.util.concurrent.atomic.AtomicLong;
-import java.util.concurrent.locks.AbstractQueuedSynchronizer;
-import java.util.concurrent.locks.Condition;
/**
* An {@link ExecutorService} for running {@link ForkJoinTask}s.
@@ -41,14 +38,22 @@ import java.util.concurrent.locks.Condit
* 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. 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
+ *
A static {@link #commonPool()} is available and appropriate for
+ * most applications. The common pool is used by any ForkJoinTask that
+ * is not explicitly submitted to a specified pool. Using the common
+ * pool normally reduces resource usage (its threads are slowly
+ * reclaimed during periods of non-use, and reinstated upon subsequent
+ * use).
+ *
+ *
For applications that require separate or custom pools, a {@code
+ * ForkJoinPool} may be constructed with a given target parallelism
+ * level; by default, equal to the number of available 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
+ * I/O 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
@@ -58,7 +63,7 @@ import java.util.concurrent.locks.Condit
* {@link #toString} returns indications of pool state in a
* convenient form for informal monitoring.
*
- *
As is the case with other ExecutorServices, there are three
+ *
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 primarily by clients not already
* engaged in fork/join computations in the current pool. The main
@@ -71,44 +76,45 @@ import java.util.concurrent.locks.Condit
* there is little difference among choice of methods.
*
*
+ * Summary of task execution methods
*
* |
* Call from non-fork/join clients |
* Call from within fork/join computations |
*
*
- * | Arrange async execution |
+ * Arrange async execution |
* {@link #execute(ForkJoinTask)} |
* {@link ForkJoinTask#fork} |
*
*
- * | Await and obtain result |
+ * Await and obtain result |
* {@link #invoke(ForkJoinTask)} |
* {@link ForkJoinTask#invoke} |
*
*
- * | Arrange exec and obtain Future |
+ * 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
- * bookkeeping overhead of creating a large set of threads. For
- * example, a common pool could be used for the {@code SortTasks}
- * illustrated in {@link RecursiveAction}. Because {@code
- * ForkJoinPool} uses threads in {@linkplain java.lang.Thread#isDaemon
- * daemon} mode, there is typically no need to explicitly {@link
- * #shutdown} such a pool upon program exit.
- *
- *
{@code
- * static final ForkJoinPool mainPool = new ForkJoinPool();
- * ...
- * public void sort(long[] array) {
- * mainPool.invoke(new SortTask(array, 0, array.length));
- * }}
+ * The common pool is by default constructed with default
+ * parameters, but these may be controlled by setting three
+ * {@linkplain System#getProperty system properties}:
+ *
+ * - {@code java.util.concurrent.ForkJoinPool.common.parallelism}
+ * - the parallelism level, a non-negative integer
+ *
- {@code java.util.concurrent.ForkJoinPool.common.threadFactory}
+ * - the class name of a {@link ForkJoinWorkerThreadFactory}
+ *
- {@code java.util.concurrent.ForkJoinPool.common.exceptionHandler}
+ * - the class name of a {@link UncaughtExceptionHandler}
+ *
+ * The system class loader is used to load these classes.
+ * Upon any error in establishing these settings, default parameters
+ * are used. It is possible to disable or limit the use of threads in
+ * the common pool by setting the parallelism property to zero, and/or
+ * using a factory that may return {@code null}.
*
* Implementation notes: This implementation restricts the
* maximum number of running threads to 32767. Attempts to create
@@ -154,32 +160,35 @@ public class ForkJoinPool extends Abstra
* (http://research.sun.com/scalable/pubs/index.html) and
* "Idempotent work stealing" by Michael, Saraswat, and Vechev,
* PPoPP 2009 (http://portal.acm.org/citation.cfm?id=1504186).
- * The main differences ultimately stem from GC requirements that
- * we null out taken slots as soon as we can, to maintain as small
- * a footprint as possible even in programs generating huge
- * numbers of tasks. To accomplish this, we shift the CAS
- * arbitrating pop vs poll (steal) from being on the indices
- * ("base" and "top") to the slots themselves. So, both a
- * successful pop and poll mainly entail a CAS of a slot from
- * non-null to null. Because we rely on CASes of references, we
- * do not need tag bits on base or top. They are simple ints as
- * used in any circular array-based queue (see for example
- * ArrayDeque). Updates to the indices must still be ordered in a
- * way that guarantees that top == base means the queue is empty,
- * but otherwise may err on the side of possibly making the queue
- * appear nonempty when a push, pop, or poll have not fully
- * committed. Note that this means that the poll operation,
- * considered individually, is not wait-free. One thief cannot
- * successfully continue until another in-progress one (or, if
- * previously empty, a push) completes. However, in the
- * aggregate, we ensure at least probabilistic non-blockingness.
- * If an attempted steal fails, a thief always chooses a different
- * random victim target to try next. So, in order for one thief to
- * progress, it suffices for any in-progress poll or new push on
- * any empty queue to complete. (This is why we normally use
- * method pollAt and its variants that try once at the apparent
- * base index, else consider alternative actions, rather than
- * method poll.)
+ * See also "Correct and Efficient Work-Stealing for Weak Memory
+ * Models" by Le, Pop, Cohen, and Nardelli, PPoPP 2013
+ * (http://www.di.ens.fr/~zappa/readings/ppopp13.pdf) for an
+ * analysis of memory ordering (atomic, volatile etc) issues. The
+ * main differences ultimately stem from GC requirements that we
+ * null out taken slots as soon as we can, to maintain as small a
+ * footprint as possible even in programs generating huge numbers
+ * of tasks. To accomplish this, we shift the CAS arbitrating pop
+ * vs poll (steal) from being on the indices ("base" and "top") to
+ * the slots themselves. So, both a successful pop and poll
+ * mainly entail a CAS of a slot from non-null to null. Because
+ * we rely on CASes of references, we do not need tag bits on base
+ * or top. They are simple ints as used in any circular
+ * array-based queue (see for example ArrayDeque). Updates to the
+ * indices must still be ordered in a way that guarantees that top
+ * == base means the queue is empty, but otherwise may err on the
+ * side of possibly making the queue appear nonempty when a push,
+ * pop, or poll have not fully committed. Note that this means
+ * that the poll operation, considered individually, is not
+ * wait-free. One thief cannot successfully continue until another
+ * in-progress one (or, if previously empty, a push) completes.
+ * However, in the aggregate, we ensure at least probabilistic
+ * non-blockingness. If an attempted steal fails, a thief always
+ * chooses a different random victim target to try next. So, in
+ * order for one thief to progress, it suffices for any
+ * in-progress poll or new push on any empty queue to
+ * complete. (This is why we normally use method pollAt and its
+ * variants that try once at the apparent base index, else
+ * consider alternative actions, rather than method poll.)
*
* This approach also enables support of a user mode in which local
* task processing is in FIFO, not LIFO order, simply by using
@@ -196,21 +205,24 @@ public class ForkJoinPool extends Abstra
* WorkQueues are also used in a similar way for tasks submitted
* to the pool. We cannot mix these tasks in the same queues used
* for work-stealing (this would contaminate lifo/fifo
- * processing). Instead, we loosely associate submission queues
+ * processing). Instead, we randomly associate submission queues
* with submitting threads, using a form of hashing. The
- * ThreadLocal Submitter class contains a value initially used as
- * a hash code for choosing existing queues, but may be randomly
- * repositioned upon contention with other submitters. In
- * essence, submitters act like workers except that they never
- * take tasks, and they are multiplexed on to a finite number of
- * shared work queues. However, classes are set up so that future
- * extensions could allow submitters to optionally help perform
- * tasks as well. Insertion of tasks in shared mode requires a
- * lock (mainly to protect in the case of resizing) but we use
- * only a simple spinlock (using bits in field runState), because
- * submitters encountering a busy queue move on to try or create
- * other queues -- they block only when creating and registering
- * new queues.
+ * Submitter probe value serves as a hash code for
+ * choosing existing queues, and may be randomly repositioned upon
+ * contention with other submitters. In essence, submitters act
+ * like workers except that they are restricted to executing local
+ * tasks that they submitted (or in the case of CountedCompleters,
+ * others with the same root task). However, because most
+ * shared/external queue operations are more expensive than
+ * internal, and because, at steady state, external submitters
+ * will compete for CPU with workers, ForkJoinTask.join and
+ * related methods disable them from repeatedly helping to process
+ * tasks if all workers are active. Insertion of tasks in shared
+ * mode requires a lock (mainly to protect in the case of
+ * resizing) but we use only a simple spinlock (using bits in
+ * field qlock), because submitters encountering a busy queue move
+ * on to try or create other queues -- they block only when
+ * creating and registering new queues.
*
* Management
* ==========
@@ -232,23 +244,26 @@ public class ForkJoinPool extends Abstra
* and their negations (used for thresholding) to fit into 16bit
* fields.
*
- * Field "runState" contains 32 bits needed to register and
- * deregister WorkQueues, as well as to enable shutdown. It is
- * only modified under a lock (normally briefly held, but
- * occasionally protecting allocations and resizings) but even
- * when locked remains available to check consistency.
+ * Field "plock" is a form of sequence lock with a saturating
+ * shutdown bit (similarly for per-queue "qlocks"), mainly
+ * protecting updates to the workQueues array, as well as to
+ * enable shutdown. When used as a lock, it is normally only very
+ * briefly held, so is nearly always available after at most a
+ * brief spin, but we use a monitor-based backup strategy to
+ * block when needed.
*
* Recording WorkQueues. WorkQueues are recorded in the
- * "workQueues" array that is created upon pool construction and
- * expanded if necessary. Updates to the array while recording
- * new workers and unrecording terminated ones are protected from
- * each other by a lock but the array is otherwise concurrently
- * readable, and accessed directly. To simplify index-based
- * operations, the array size is always a power of two, and all
- * readers must tolerate null slots. Shared (submission) queues
- * are at even indices, worker queues at odd indices. Grouping
- * them together in this way simplifies and speeds up task
- * scanning.
+ * "workQueues" array that is created upon first use and expanded
+ * if necessary. Updates to the array while recording new workers
+ * and unrecording terminated ones are protected from each other
+ * by a lock but the array is otherwise concurrently readable, and
+ * accessed directly. To simplify index-based operations, the
+ * array size is always a power of two, and all readers must
+ * tolerate null slots. Worker queues are at odd indices. Shared
+ * (submission) queues are at even indices, up to a maximum of 64
+ * slots, to limit growth even if array needs to expand to add
+ * more workers. Grouping them together in this way simplifies and
+ * speeds up task scanning.
*
* All worker thread creation is on-demand, triggered by task
* submissions, replacement of terminated workers, and/or
@@ -293,40 +308,47 @@ public class ForkJoinPool extends Abstra
* has not yet entered the wait queue. We solve this by requiring
* a full sweep of all workers (via repeated calls to method
* scan()) both before and after a newly waiting worker is added
- * to the wait queue. During a rescan, the worker might release
- * some other queued worker rather than itself, which has the same
- * net effect. Because enqueued workers may actually be rescanning
- * rather than waiting, we set and clear the "parker" field of
- * WorkQueues to reduce unnecessary calls to unpark. (This
- * requires a secondary recheck to avoid missed signals.) Note
- * the unusual conventions about Thread.interrupts surrounding
- * parking and other blocking: Because interrupts are used solely
- * to alert threads to check termination, which is checked anyway
- * upon blocking, we clear status (using Thread.interrupted)
- * before any call to park, so that park does not immediately
- * return due to status being set via some other unrelated call to
- * interrupt in user code.
+ * to the wait queue. Because enqueued workers may actually be
+ * rescanning rather than waiting, we set and clear the "parker"
+ * field of WorkQueues to reduce unnecessary calls to unpark.
+ * (This requires a secondary recheck to avoid missed signals.)
+ * Note the unusual conventions about Thread.interrupts
+ * surrounding parking and other blocking: Because interrupts are
+ * used solely to alert threads to check termination, which is
+ * checked anyway upon blocking, we clear status (using
+ * Thread.interrupted) before any call to park, so that park does
+ * not immediately return due to status being set via some other
+ * unrelated call to interrupt in user code.
*
* Signalling. We create or wake up workers only when there
* appears to be at least one task they might be able to find and
* execute. When a submission is added or another worker adds a
- * task to a queue that previously had fewer than two tasks, they
- * signal waiting workers (or trigger creation of new ones if
- * fewer than the given parallelism level -- see signalWork).
- * These primary signals are buttressed by signals during rescans;
- * together these cover the signals needed in cases when more
- * tasks are pushed but untaken, and improve performance compared
- * to having one thread wake up all workers.
+ * task to a queue that has fewer than two tasks, they signal
+ * waiting workers (or trigger creation of new ones if fewer than
+ * the given parallelism level -- signalWork). These primary
+ * signals are buttressed by others whenever other threads remove
+ * a task from a queue and notice that there are other tasks there
+ * as well. So in general, pools will be over-signalled. On most
+ * platforms, signalling (unpark) overhead time is noticeably
+ * long, and the time between signalling a thread and it actually
+ * making progress can be very noticeably long, so it is worth
+ * offloading these delays from critical paths as much as
+ * possible. Additionally, workers spin-down gradually, by staying
+ * alive so long as they see the ctl state changing. Similar
+ * stability-sensing techniques are also used before blocking in
+ * awaitJoin and helpComplete.
*
* Trimming workers. To release resources after periods of lack of
* use, a worker starting to wait when the pool is quiescent will
- * time out and terminate if the pool has remained quiescent for
- * SHRINK_RATE nanosecs. This will slowly propagate, eventually
- * terminating all workers after long periods of non-use.
+ * time out and terminate if the pool has remained quiescent for a
+ * given period -- a short period if there are more threads than
+ * parallelism, longer as the number of threads decreases. This
+ * will slowly propagate, eventually terminating all workers after
+ * periods of non-use.
*
* Shutdown and Termination. A call to shutdownNow atomically sets
- * a runState bit and then (non-atomically) sets each worker's
- * runState status, cancels all unprocessed tasks, and wakes up
+ * a plock bit and then (non-atomically) sets each worker's
+ * qlock status, cancels all unprocessed tasks, and wakes up
* all waiting workers. Detecting whether termination should
* commence after a non-abrupt shutdown() call requires more work
* and bookkeeping. We need consensus about quiescence (i.e., that
@@ -354,13 +376,13 @@ public class ForkJoinPool extends Abstra
* method tryCompensate() may create or re-activate a spare
* thread to compensate for blocked joiners until they unblock.
*
- * A third form (implemented in tryRemoveAndExec and
- * tryPollForAndExec) amounts to helping a hypothetical
- * compensator: If we can readily tell that a possible action of a
- * compensator is to steal and execute the task being joined, the
- * joining thread can do so directly, without the need for a
- * compensation thread (although at the expense of larger run-time
- * stacks, but the tradeoff is typically worthwhile).
+ * A third form (implemented in tryRemoveAndExec) amounts to
+ * helping a hypothetical compensator: If we can readily tell that
+ * a possible action of a compensator is to steal and execute the
+ * task being joined, the joining thread can do so directly,
+ * without the need for a compensation thread (although at the
+ * expense of larger run-time stacks, but the tradeoff is
+ * typically worthwhile).
*
* The ManagedBlocker extension API can't use helping so relies
* only on compensation in method awaitBlocker.
@@ -382,12 +404,12 @@ public class ForkJoinPool extends Abstra
* steals, rather than use per-task bookkeeping. This sometimes
* requires a linear scan of workQueues array to locate stealers,
* but often doesn't because stealers leave hints (that may become
- * stale/wrong) of where to locate them. A stealHint is only a
- * hint because a worker might have had multiple steals and the
- * hint records only one of them (usually the most current).
- * Hinting isolates cost to when it is needed, rather than adding
- * to per-task overhead. (2) It is "shallow", ignoring nesting
- * and potentially cyclic mutual steals. (3) It is intentionally
+ * stale/wrong) of where to locate them. It is only a hint
+ * because a worker might have had multiple steals and the hint
+ * records only one of them (usually the most current). Hinting
+ * isolates cost to when it is needed, rather than adding to
+ * per-task overhead. (2) It is "shallow", ignoring nesting and
+ * potentially cyclic mutual steals. (3) It is intentionally
* racy: field currentJoin is updated only while actively joining,
* which means that we miss links in the chain during long-lived
* tasks, GC stalls etc (which is OK since blocking in such cases
@@ -395,6 +417,12 @@ public class ForkJoinPool extends Abstra
* to find work (see MAX_HELP) and fall back to suspending the
* worker and if necessary replacing it with another.
*
+ * Helping actions for CountedCompleters are much simpler: Method
+ * helpComplete can take and execute any task with the same root
+ * as the task being waited on. However, this still entails some
+ * traversal of completer chains, so is less efficient than using
+ * CountedCompleters without explicit joins.
+ *
* It is impossible to keep exactly the target parallelism number
* of threads running at any given time. Determining the
* existence of conservatively safe helping targets, the
@@ -416,30 +444,44 @@ public class ForkJoinPool extends Abstra
* intractable) game with an opponent that may choose the worst
* (for us) active thread to stall at any time. We take several
* precautions to bound losses (and thus bound gains), mainly in
- * methods tryCompensate and awaitJoin: (1) We only try
- * compensation after attempting enough helping steps (measured
- * via counting and timing) that we have already consumed the
- * estimated cost of creating and activating a new thread. (2) We
- * allow up to 50% of threads to be blocked before initially
- * adding any others, and unless completely saturated, check that
- * some work is available for a new worker before adding. Also, we
- * create up to only 50% more threads until entering a mode that
- * only adds a thread if all others are possibly blocked. All
- * together, this means that we might be half as fast to react,
- * and create half as many threads as possible in the ideal case,
- * but present vastly fewer anomalies in all other cases compared
- * to both more aggressive and more conservative alternatives.
- *
- * Style notes: There is a lot of representation-level coupling
- * among classes ForkJoinPool, ForkJoinWorkerThread, and
- * ForkJoinTask. The fields of WorkQueue maintain data structures
- * managed by ForkJoinPool, so are directly accessed. There is
- * little point trying to reduce this, since any associated future
- * changes in representations will need to be accompanied by
- * algorithmic changes anyway. Several methods intrinsically
- * sprawl because they must accumulate sets of consistent reads of
- * volatiles held in local variables. Methods signalWork() and
- * scan() are the main bottlenecks, so are especially heavily
+ * methods tryCompensate and awaitJoin.
+ *
+ * Common Pool
+ * ===========
+ *
+ * The static common pool always exists after static
+ * initialization. Since it (or any other created pool) need
+ * never be used, we minimize initial construction overhead and
+ * footprint to the setup of about a dozen fields, with no nested
+ * allocation. Most bootstrapping occurs within method
+ * fullExternalPush during the first submission to the pool.
+ *
+ * When external threads submit to the common pool, they can
+ * perform subtask processing (see externalHelpJoin and related
+ * methods). This caller-helps policy makes it sensible to set
+ * common pool parallelism level to one (or more) less than the
+ * total number of available cores, or even zero for pure
+ * caller-runs. We do not need to record whether external
+ * submissions are to the common pool -- if not, externalHelpJoin
+ * returns quickly (at the most helping to signal some common pool
+ * workers). These submitters would otherwise be blocked waiting
+ * for completion, so the extra effort (with liberally sprinkled
+ * task status checks) in inapplicable cases amounts to an odd
+ * form of limited spin-wait before blocking in ForkJoinTask.join.
+ *
+ * Style notes
+ * ===========
+ *
+ * There is a lot of representation-level coupling among classes
+ * ForkJoinPool, ForkJoinWorkerThread, and ForkJoinTask. The
+ * fields of WorkQueue maintain data structures managed by
+ * ForkJoinPool, so are directly accessed. There is little point
+ * trying to reduce this, since any associated future changes in
+ * representations will need to be accompanied by algorithmic
+ * changes anyway. Several methods intrinsically sprawl because
+ * they must accumulate sets of consistent reads of volatiles held
+ * in local variables. Methods signalWork() and scan() are the
+ * main bottlenecks, so are especially heavily
* micro-optimized/mangled. There are lots of inline assignments
* (of form "while ((local = field) != 0)") which are usually the
* simplest way to ensure the required read orderings (which are
@@ -447,7 +489,8 @@ public class ForkJoinPool extends Abstra
* declarations of these locals at the heads of methods or blocks.
* There are 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
+ * CAS'ed variable. There are also other coding oddities (including
+ * several unnecessary-looking hoisted null checks) that help
* some methods perform reasonably even when interpreted (not
* compiled).
*
@@ -487,6 +530,7 @@ public class ForkJoinPool extends Abstra
* Returns a new worker thread operating in the given pool.
*
* @param pool the pool this thread works in
+ * @return the new worker thread
* @throws NullPointerException if the pool is null
*/
public ForkJoinWorkerThread newThread(ForkJoinPool pool);
@@ -496,43 +540,21 @@ public class ForkJoinPool extends Abstra
* Default ForkJoinWorkerThreadFactory implementation; creates a
* new ForkJoinWorkerThread.
*/
- static class DefaultForkJoinWorkerThreadFactory
+ static final class DefaultForkJoinWorkerThreadFactory
implements ForkJoinWorkerThreadFactory {
- public ForkJoinWorkerThread newThread(ForkJoinPool pool) {
+ public final ForkJoinWorkerThread newThread(ForkJoinPool pool) {
return new ForkJoinWorkerThread(pool);
}
}
/**
- * A simple non-reentrant lock used for exclusion when managing
- * queues and workers. We use a custom lock so that we can readily
- * probe lock state in constructions that check among alternative
- * actions. The lock is normally only very briefly held, and
- * sometimes treated as a spinlock, but other usages block to
- * reduce overall contention in those cases where locked code
- * bodies perform allocation/resizing.
- */
- static final class Mutex extends AbstractQueuedSynchronizer {
- public final boolean tryAcquire(int ignore) {
- return compareAndSetState(0, 1);
- }
- public final boolean tryRelease(int ignore) {
- setState(0);
- return true;
- }
- public final void lock() { acquire(0); }
- public final void unlock() { release(0); }
- public final boolean isHeldExclusively() { return getState() == 1; }
- public final Condition newCondition() { return new ConditionObject(); }
- }
-
- /**
* Class for artificial tasks that are used to replace the target
* of local joins if they are removed from an interior queue slot
* in WorkQueue.tryRemoveAndExec. We don't need the proxy to
* actually do anything beyond having a unique identity.
*/
static final class EmptyTask extends ForkJoinTask {
+ private static final long serialVersionUID = -7721805057305804111L;
EmptyTask() { status = ForkJoinTask.NORMAL; } // force done
public final Void getRawResult() { return null; }
public final void setRawResult(Void x) {}
@@ -553,44 +575,39 @@ public class ForkJoinPool extends Abstra
*
* Field "top" is the index (mod array.length) of the next queue
* slot to push to or pop from. It is written only by owner thread
- * for push, or under lock for trySharedPush, and accessed by
- * other threads only after reading (volatile) base. Both top and
- * base are allowed to wrap around on overflow, but (top - base)
- * (or more commonly -(base - top) to force volatile read of base
- * before top) still estimates size.
+ * for push, or under lock for external/shared push, and accessed
+ * by other threads only after reading (volatile) base. Both top
+ * and base are allowed to wrap around on overflow, but (top -
+ * base) (or more commonly -(base - top) to force volatile read of
+ * base before top) still estimates size. The lock ("qlock") is
+ * forced to -1 on termination, causing all further lock attempts
+ * to fail. (Note: we don't need CAS for termination state because
+ * upon pool shutdown, all shared-queues will stop being used
+ * anyway.) Nearly all lock bodies are set up so that exceptions
+ * within lock bodies are "impossible" (modulo JVM errors that
+ * would cause failure anyway.)
*
* The array slots are read and written using the emulation of
* volatiles/atomics provided by Unsafe. Insertions must in
* general use putOrderedObject as a form of releasing store to
* ensure that all writes to the task object are ordered before
- * its publication in the queue. (Although we can avoid one case
- * of this when locked in trySharedPush.) All removals entail a
- * CAS to null. The array is always a power of two. To ensure
- * safety of Unsafe array operations, all accesses perform
- * explicit null checks and implicit bounds checks via
- * power-of-two masking.
+ * its publication in the queue. All removals entail a CAS to
+ * null. The array is always a power of two. To ensure safety of
+ * Unsafe array operations, all accesses perform explicit null
+ * checks and implicit bounds checks via power-of-two masking.
*
* In addition to basic queuing support, this class contains
* fields described elsewhere to control execution. It turns out
- * to work better memory-layout-wise to include them in this
- * class rather than a separate class.
+ * to work better memory-layout-wise to include them in this class
+ * rather than a separate class.
*
* Performance on most platforms is very sensitive to placement of
* instances of both WorkQueues and their arrays -- we absolutely
* do not want multiple WorkQueue instances or multiple queue
* arrays sharing cache lines. (It would be best for queue objects
* and their arrays to share, but there is nothing available to
- * help arrange that). Unfortunately, because they are recorded
- * in a common array, WorkQueue instances are often moved to be
- * adjacent by garbage collectors. To reduce impact, we use field
- * padding that works OK on common platforms; this effectively
- * trades off slightly slower average field access for the sake of
- * avoiding really bad worst-case access. (Until better JVM
- * support is in place, this padding is dependent on transient
- * properties of JVM field layout rules.) We also take care in
- * allocating, sizing and resizing the array. Non-shared queue
- * arrays are initialized (via method growArray) by workers before
- * use. Others are allocated on first use.
+ * help arrange that). The @Contended annotation alerts JVMs to
+ * try to keep instances apart.
*/
static final class WorkQueue {
/**
@@ -613,16 +630,16 @@ public class ForkJoinPool extends Abstra
*/
static final int MAXIMUM_QUEUE_CAPACITY = 1 << 26; // 64M
- volatile long totalSteals; // cumulative number of steals
- int seed; // for random scanning; initialize nonzero
+ // Heuristic padding to ameliorate unfortunate memory placements
+ volatile long pad00, pad01, pad02, pad03, pad04, pad05, pad06;
+
volatile int eventCount; // encoded inactivation count; < 0 if inactive
int nextWait; // encoded record of next event waiter
- int rescans; // remaining scans until block
- int nsteals; // top-level task executions since last idle
- final int mode; // lifo, fifo, or shared
- int poolIndex; // index of this queue in pool (or 0)
- int stealHint; // index of most recent known stealer
- volatile int runState; // 1: locked, -1: terminate; else 0
+ int nsteals; // number of steals
+ int hint; // steal index hint
+ short poolIndex; // index of this queue in pool
+ final short mode; // 0: lifo, > 0: fifo, < 0: shared
+ volatile int qlock; // 1: locked, -1: terminate; else 0
volatile int base; // index of next slot for poll
int top; // index of next slot for push
ForkJoinTask[] array; // the elements (initially unallocated)
@@ -631,14 +648,16 @@ public class ForkJoinPool extends Abstra
volatile Thread parker; // == owner during call to park; else null
volatile ForkJoinTask currentJoin; // task being joined in awaitJoin
ForkJoinTask currentSteal; // current non-local task being executed
- // Heuristic padding to ameliorate unfortunate memory placements
- Object p00, p01, p02, p03, p04, p05, p06, p07;
- Object p08, p09, p0a, p0b, p0c, p0d, p0e;
- WorkQueue(ForkJoinPool pool, ForkJoinWorkerThread owner, int mode) {
- this.mode = mode;
+ volatile Object pad10, pad11, pad12, pad13, pad14, pad15, pad16, pad17;
+ volatile Object pad18, pad19, pad1a, pad1b, pad1c, pad1d;
+
+ WorkQueue(ForkJoinPool pool, ForkJoinWorkerThread owner, int mode,
+ int seed) {
this.pool = pool;
this.owner = owner;
+ this.mode = (short)mode;
+ this.hint = seed; // store initial seed for runWorker
// Place indices in the center of array (that is not yet allocated)
base = top = INITIAL_QUEUE_CAPACITY >>> 1;
}
@@ -663,62 +682,61 @@ public class ForkJoinPool extends Abstra
(n == -1 &&
((a = array) == null ||
(m = a.length - 1) < 0 ||
- U.getObjectVolatile
- (a, ((m & (s - 1)) << ASHIFT) + ABASE) == null)));
+ U.getObject
+ (a, (long)((m & (s - 1)) << ASHIFT) + ABASE) == null)));
}
/**
- * Pushes a task. Call only by owner in unshared queues.
+ * Pushes a task. Call only by owner in unshared queues. (The
+ * shared-queue version is embedded in method externalPush.)
*
* @param task the task. Caller must ensure non-null.
- * @throw RejectedExecutionException if array cannot be resized
+ * @throws RejectedExecutionException if array cannot be resized
*/
final void push(ForkJoinTask task) {
ForkJoinTask[] a; ForkJoinPool p;
- int s = top, m, n;
+ int s = top, n;
if ((a = array) != null) { // ignore if queue removed
- U.putOrderedObject
- (a, (((m = a.length - 1) & s) << ASHIFT) + ABASE, task);
- if ((n = (top = s + 1) - base) <= 2) {
- if ((p = pool) != null)
- p.signalWork();
- }
+ int m = a.length - 1;
+ U.putOrderedObject(a, ((m & s) << ASHIFT) + ABASE, task);
+ if ((n = (top = s + 1) - base) <= 2)
+ (p = pool).signalWork(p.workQueues, this);
else if (n >= m)
- growArray(true);
+ growArray();
}
}
/**
- * Pushes a task if lock is free and array is either big
- * enough or can be resized to be big enough.
- *
- * @param task the task. Caller must ensure non-null.
- * @return true if submitted
+ * Initializes or doubles the capacity of array. Call either
+ * by owner or with lock held -- it is OK for base, but not
+ * top, to move while resizings are in progress.
*/
- final boolean trySharedPush(ForkJoinTask task) {
- boolean submitted = false;
- if (runState == 0 && U.compareAndSwapInt(this, RUNSTATE, 0, 1)) {
- ForkJoinTask[] a = array;
- int s = top;
- try {
- if ((a != null && a.length > s + 1 - base) ||
- (a = growArray(false)) != null) { // must presize
- int j = (((a.length - 1) & s) << ASHIFT) + ABASE;
- U.putObject(a, (long)j, task); // don't need "ordered"
- top = s + 1;
- submitted = true;
- }
- } finally {
- runState = 0; // unlock
- }
+ final ForkJoinTask[] growArray() {
+ ForkJoinTask[] oldA = array;
+ int size = oldA != null ? oldA.length << 1 : INITIAL_QUEUE_CAPACITY;
+ if (size > MAXIMUM_QUEUE_CAPACITY)
+ throw new RejectedExecutionException("Queue capacity exceeded");
+ int oldMask, t, b;
+ ForkJoinTask[] a = array = new ForkJoinTask[size];
+ if (oldA != null && (oldMask = oldA.length - 1) >= 0 &&
+ (t = top) - (b = base) > 0) {
+ int mask = size - 1;
+ do {
+ ForkJoinTask x;
+ int oldj = ((b & oldMask) << ASHIFT) + ABASE;
+ int j = ((b & mask) << ASHIFT) + ABASE;
+ x = (ForkJoinTask)U.getObjectVolatile(oldA, oldj);
+ if (x != null &&
+ U.compareAndSwapObject(oldA, oldj, x, null))
+ U.putObjectVolatile(a, j, x);
+ } while (++b != t);
}
- return submitted;
+ return a;
}
/**
* Takes next task, if one exists, in LIFO order. Call only
- * by owner in unshared queues. (We do not have a shared
- * version of this method because it is never needed.)
+ * by owner in unshared queues.
*/
final ForkJoinTask pop() {
ForkJoinTask[] a; ForkJoinTask t; int m;
@@ -746,9 +764,8 @@ public class ForkJoinPool extends Abstra
if ((a = array) != null) {
int j = (((a.length - 1) & b) << ASHIFT) + ABASE;
if ((t = (ForkJoinTask)U.getObjectVolatile(a, j)) != null &&
- base == b &&
- U.compareAndSwapObject(a, j, t, null)) {
- base = b + 1;
+ base == b && U.compareAndSwapObject(a, j, t, null)) {
+ U.putOrderedInt(this, QBASE, b + 1);
return t;
}
}
@@ -764,16 +781,15 @@ public class ForkJoinPool extends Abstra
int j = (((a.length - 1) & b) << ASHIFT) + ABASE;
t = (ForkJoinTask)U.getObjectVolatile(a, j);
if (t != null) {
- if (base == b &&
- U.compareAndSwapObject(a, j, t, null)) {
- base = b + 1;
+ if (U.compareAndSwapObject(a, j, t, null)) {
+ U.putOrderedInt(this, QBASE, b + 1);
return t;
}
}
else if (base == b) {
if (b + 1 == top)
break;
- Thread.yield(); // wait for lagging update
+ Thread.yield(); // wait for lagging update (very rare)
}
}
return null;
@@ -800,6 +816,7 @@ public class ForkJoinPool extends Abstra
/**
* Pops the given task only if it is at the current top.
+ * (A shared version is available only via FJP.tryExternalUnpush)
*/
final boolean tryUnpush(ForkJoinTask t) {
ForkJoinTask[] a; int s;
@@ -813,57 +830,6 @@ public class ForkJoinPool extends Abstra
}
/**
- * Polls the given task only if it is at the current base.
- */
- final boolean pollFor(ForkJoinTask task) {
- ForkJoinTask[] a; int b;
- if ((b = base) - top < 0 && (a = array) != null) {
- int j = (((a.length - 1) & b) << ASHIFT) + ABASE;
- if (U.getObjectVolatile(a, j) == task && base == b &&
- U.compareAndSwapObject(a, j, task, null)) {
- base = b + 1;
- return true;
- }
- }
- return false;
- }
-
- /**
- * Initializes or doubles the capacity of array. Call either
- * by owner or with lock held -- it is OK for base, but not
- * top, to move while resizings are in progress.
- *
- * @param rejectOnFailure if true, throw exception if capacity
- * exceeded (relayed ultimately to user); else return null.
- */
- final ForkJoinTask[] growArray(boolean rejectOnFailure) {
- ForkJoinTask[] oldA = array;
- int size = oldA != null ? oldA.length << 1 : INITIAL_QUEUE_CAPACITY;
- if (size <= MAXIMUM_QUEUE_CAPACITY) {
- int oldMask, t, b;
- ForkJoinTask[] a = array = new ForkJoinTask[size];
- if (oldA != null && (oldMask = oldA.length - 1) >= 0 &&
- (t = top) - (b = base) > 0) {
- int mask = size - 1;
- do {
- ForkJoinTask x;
- int oldj = ((b & oldMask) << ASHIFT) + ABASE;
- int j = ((b & mask) << ASHIFT) + ABASE;
- x = (ForkJoinTask)U.getObjectVolatile(oldA, oldj);
- if (x != null &&
- U.compareAndSwapObject(oldA, oldj, x, null))
- U.putObjectVolatile(a, j, x);
- } while (++b != t);
- }
- return a;
- }
- else if (!rejectOnFailure)
- return null;
- else
- throw new RejectedExecutionException("Queue capacity exceeded");
- }
-
- /**
* Removes and cancels all known tasks, ignoring any exceptions.
*/
final void cancelAll() {
@@ -873,65 +839,62 @@ public class ForkJoinPool extends Abstra
ForkJoinTask.cancelIgnoringExceptions(t);
}
- /**
- * Computes next value for random probes. Scans don't require
- * a very high quality generator, but also not a crummy one.
- * Marsaglia xor-shift is cheap and works well enough. Note:
- * This is manually inlined in its usages in ForkJoinPool to
- * avoid writes inside busy scan loops.
- */
- final int nextSeed() {
- int r = seed;
- r ^= r << 13;
- r ^= r >>> 17;
- return seed = r ^= r << 5;
- }
-
- // Execution methods
-
- /**
- * Pops and runs tasks until empty.
- */
- private void popAndExecAll() {
- // A bit faster than repeated pop calls
- ForkJoinTask[] a; int m, s; long j; ForkJoinTask t;
- while ((a = array) != null && (m = a.length - 1) >= 0 &&
- (s = top - 1) - base >= 0 &&
- (t = ((ForkJoinTask)
- U.getObject(a, j = ((m & s) << ASHIFT) + ABASE)))
- != null) {
- if (U.compareAndSwapObject(a, j, t, null)) {
- top = s;
- t.doExec();
- }
- }
- }
+ // Specialized execution methods
/**
* Polls and runs tasks until empty.
*/
- private void pollAndExecAll() {
+ final void pollAndExecAll() {
for (ForkJoinTask t; (t = poll()) != null;)
t.doExec();
}
/**
- * If present, removes from queue and executes the given task, or
- * any other cancelled task. Returns (true) immediately on any CAS
+ * Executes a top-level task and any local tasks remaining
+ * after execution.
+ */
+ final void runTask(ForkJoinTask task) {
+ if ((currentSteal = task) != null) {
+ task.doExec();
+ ForkJoinTask[] a = array;
+ int md = mode;
+ ++nsteals;
+ currentSteal = null;
+ if (md != 0)
+ pollAndExecAll();
+ else if (a != null) {
+ int s, m = a.length - 1;
+ while ((s = top - 1) - base >= 0) {
+ long i = ((m & s) << ASHIFT) + ABASE;
+ ForkJoinTask t = (ForkJoinTask)U.getObject(a, i);
+ if (t == null)
+ break;
+ if (U.compareAndSwapObject(a, i, t, null)) {
+ top = s;
+ t.doExec();
+ }
+ }
+ }
+ }
+ }
+
+ /**
+ * If present, removes from queue and executes the given task,
+ * or any other cancelled task. Returns (true) on any CAS
* or consistency check failure so caller can retry.
*
- * @return 0 if no progress can be made, else positive
- * (this unusual convention simplifies use with tryHelpStealer.)
+ * @return false if no progress can be made, else true
*/
- final int tryRemoveAndExec(ForkJoinTask task) {
- int stat = 1;
- boolean removed = false, empty = true;
+ final boolean tryRemoveAndExec(ForkJoinTask task) {
+ boolean stat;
ForkJoinTask[] a; int m, s, b, n;
- if ((a = array) != null && (m = a.length - 1) >= 0 &&
+ if (task != null && (a = array) != null && (m = a.length - 1) >= 0 &&
(n = (s = top) - (b = base)) > 0) {
+ boolean removed = false, empty = true;
+ stat = true;
for (ForkJoinTask t;;) { // traverse from s to b
- int j = ((--s & m) << ASHIFT) + ABASE;
- t = (ForkJoinTask)U.getObjectVolatile(a, j);
+ long j = ((--s & m) << ASHIFT) + ABASE;
+ t = (ForkJoinTask)U.getObject(a, j);
if (t == null) // inconsistent length
break;
else if (t == task) {
@@ -955,45 +918,99 @@ public class ForkJoinPool extends Abstra
}
if (--n == 0) {
if (!empty && base == b)
- stat = 0;
+ stat = false;
break;
}
}
+ if (removed)
+ task.doExec();
}
- if (removed)
- task.doExec();
+ else
+ stat = false;
return stat;
}
/**
- * Executes a top-level task and any local tasks remaining
- * after execution.
+ * Tries to poll for and execute the given task or any other
+ * task in its CountedCompleter computation.
*/
- final void runTask(ForkJoinTask t) {
- if (t != null) {
- currentSteal = t;
- t.doExec();
- if (top != base) { // process remaining local tasks
- if (mode == 0)
- popAndExecAll();
- else
- pollAndExecAll();
+ final boolean pollAndExecCC(CountedCompleter root) {
+ ForkJoinTask[] a; int b; Object o; CountedCompleter t, r;
+ if ((b = base) - top < 0 && (a = array) != null) {
+ long j = (((a.length - 1) & b) << ASHIFT) + ABASE;
+ if ((o = U.getObjectVolatile(a, j)) == null)
+ return true; // retry
+ if (o instanceof CountedCompleter) {
+ for (t = (CountedCompleter)o, r = t;;) {
+ if (r == root) {
+ if (base == b &&
+ U.compareAndSwapObject(a, j, t, null)) {
+ U.putOrderedInt(this, QBASE, b + 1);
+ t.doExec();
+ }
+ return true;
+ }
+ else if ((r = r.completer) == null)
+ break; // not part of root computation
+ }
}
- ++nsteals;
- currentSteal = null;
}
+ return false;
}
/**
- * Executes a non-top-level (stolen) task.
+ * Tries to pop and execute the given task or any other task
+ * in its CountedCompleter computation.
*/
- final void runSubtask(ForkJoinTask t) {
- if (t != null) {
- ForkJoinTask ps = currentSteal;
- currentSteal = t;
- t.doExec();
- currentSteal = ps;
+ final boolean externalPopAndExecCC(CountedCompleter root) {
+ ForkJoinTask[] a; int s; Object o; CountedCompleter t, r;
+ if (base - (s = top) < 0 && (a = array) != null) {
+ long j = (((a.length - 1) & (s - 1)) << ASHIFT) + ABASE;
+ if ((o = U.getObject(a, j)) instanceof CountedCompleter) {
+ for (t = (CountedCompleter)o, r = t;;) {
+ if (r == root) {
+ if (U.compareAndSwapInt(this, QLOCK, 0, 1)) {
+ if (top == s && array == a &&
+ U.compareAndSwapObject(a, j, t, null)) {
+ top = s - 1;
+ qlock = 0;
+ t.doExec();
+ }
+ else
+ qlock = 0;
+ }
+ return true;
+ }
+ else if ((r = r.completer) == null)
+ break;
+ }
+ }
}
+ return false;
+ }
+
+ /**
+ * Internal version
+ */
+ final boolean internalPopAndExecCC(CountedCompleter root) {
+ ForkJoinTask[] a; int s; Object o; CountedCompleter t, r;
+ if (base - (s = top) < 0 && (a = array) != null) {
+ long j = (((a.length - 1) & (s - 1)) << ASHIFT) + ABASE;
+ if ((o = U.getObject(a, j)) instanceof CountedCompleter) {
+ for (t = (CountedCompleter)o, r = t;;) {
+ if (r == root) {
+ if (U.compareAndSwapObject(a, j, t, null)) {
+ top = s - 1;
+ t.doExec();
+ }
+ return true;
+ }
+ else if ((r = r.completer) == null)
+ break;
+ }
+ }
+ }
+ return false;
}
/**
@@ -1008,77 +1025,44 @@ public class ForkJoinPool extends Abstra
s != Thread.State.TIMED_WAITING);
}
- /**
- * If this owned and is not already interrupted, try to
- * interrupt and/or unpark, ignoring exceptions.
- */
- final void interruptOwner() {
- Thread wt, p;
- if ((wt = owner) != null && !wt.isInterrupted()) {
- try {
- wt.interrupt();
- } catch (SecurityException ignore) {
- }
- }
- if ((p = parker) != null)
- U.unpark(p);
- }
-
// Unsafe mechanics
private static final sun.misc.Unsafe U;
- private static final long RUNSTATE;
+ private static final long QBASE;
+ private static final long QLOCK;
private static final int ABASE;
private static final int ASHIFT;
static {
- int s;
try {
U = getUnsafe();
Class k = WorkQueue.class;
Class ak = ForkJoinTask[].class;
- RUNSTATE = U.objectFieldOffset
- (k.getDeclaredField("runState"));
+ QBASE = U.objectFieldOffset
+ (k.getDeclaredField("base"));
+ QLOCK = U.objectFieldOffset
+ (k.getDeclaredField("qlock"));
ABASE = U.arrayBaseOffset(ak);
- s = U.arrayIndexScale(ak);
+ int scale = U.arrayIndexScale(ak);
+ if ((scale & (scale - 1)) != 0)
+ throw new Error("data type scale not a power of two");
+ ASHIFT = 31 - Integer.numberOfLeadingZeros(scale);
} catch (Exception e) {
throw new Error(e);
}
- if ((s & (s-1)) != 0)
- throw new Error("data type scale not a power of two");
- ASHIFT = 31 - Integer.numberOfLeadingZeros(s);
}
}
- /**
- * Per-thread records for threads that submit to pools. Currently
- * holds only pseudo-random seed / index that is used to choose
- * submission queues in method doSubmit. In the future, this may
- * also incorporate a means to implement different task rejection
- * and resubmission policies.
- *
- * Seeds for submitters and workers/workQueues work in basically
- * the same way but are initialized and updated using slightly
- * different mechanics. Both are initialized using the same
- * approach as in class ThreadLocal, where successive values are
- * unlikely to collide with previous values. This is done during
- * registration for workers, but requires a separate AtomicInteger
- * for submitters. Seeds are then randomly modified upon
- * collisions using xorshifts, which requires a non-zero seed.
- */
- static final class Submitter {
- int seed;
- Submitter() {
- int s = nextSubmitterSeed.getAndAdd(SEED_INCREMENT);
- seed = (s == 0) ? 1 : s; // ensure non-zero
- }
- }
-
- /** ThreadLocal class for Submitters */
- static final class ThreadSubmitter extends ThreadLocal {
- public Submitter initialValue() { return new Submitter(); }
- }
// static fields (initialized in static initializer below)
/**
+ * Per-thread submission bookkeeping. Shared across all pools
+ * to reduce ThreadLocal pollution and because random motion
+ * to avoid contention in one pool is likely to hold for others.
+ * Lazily initialized on first submission (but null-checked
+ * in other contexts to avoid unnecessary initialization).
+ */
+ static final ThreadLocal submitters;
+
+ /**
* Creates a new ForkJoinWorkerThread. This factory is used unless
* overridden in ForkJoinPool constructors.
*/
@@ -1086,54 +1070,65 @@ public class ForkJoinPool extends Abstra
defaultForkJoinWorkerThreadFactory;
/**
- * Generator for assigning sequence numbers as pool names.
+ * Permission required for callers of methods that may start or
+ * kill threads.
*/
- private static final AtomicInteger poolNumberGenerator;
+ private static final RuntimePermission modifyThreadPermission;
/**
- * Generator for initial hashes/seeds for submitters. Accessed by
- * Submitter class constructor.
+ * Common (static) pool. Non-null for public use unless a static
+ * construction exception, but internal usages null-check on use
+ * to paranoically avoid potential initialization circularities
+ * as well as to simplify generated code.
*/
- static final AtomicInteger nextSubmitterSeed;
+ static final ForkJoinPool common;
/**
- * Permission required for callers of methods that may start or
- * kill threads.
+ * Common pool parallelism. To allow simpler use and management
+ * when common pool threads are disabled, we allow the underlying
+ * common.parallelism field to be zero, but in that case still report
+ * parallelism as 1 to reflect resulting caller-runs mechanics.
*/
- private static final RuntimePermission modifyThreadPermission;
+ static final int commonParallelism;
/**
- * Per-thread submission bookeeping. Shared across all pools
- * to reduce ThreadLocal pollution and because random motion
- * to avoid contention in one pool is likely to hold for others.
+ * Sequence number for creating workerNamePrefix.
*/
- private static final ThreadSubmitter submitters;
+ private static int poolNumberSequence;
+
+ /**
+ * Returns the next sequence number. We don't expect this to
+ * ever contend, so use simple builtin sync.
+ */
+ private static final synchronized int nextPoolId() {
+ return ++poolNumberSequence;
+ }
// static constants
/**
- * The wakeup interval (in nanoseconds) for a worker waiting for a
- * task when the pool is quiescent to instead try to shrink the
- * number of workers. The exact value does not matter too
- * much. It must be short enough to release resources during
- * sustained periods of idleness, but not so short that threads
- * are continually re-created.
+ * Initial timeout value (in nanoseconds) for the thread
+ * triggering quiescence to park waiting for new work. On timeout,
+ * the thread will instead try to shrink the number of
+ * workers. The value should be large enough to avoid overly
+ * aggressive shrinkage during most transient stalls (long GCs
+ * etc).
+ */
+ private static final long IDLE_TIMEOUT = 2000L * 1000L * 1000L; // 2sec
+
+ /**
+ * Timeout value when there are more threads than parallelism level
*/
- private static final long SHRINK_RATE =
- 4L * 1000L * 1000L * 1000L; // 4 seconds
+ private static final long FAST_IDLE_TIMEOUT = 200L * 1000L * 1000L;
/**
- * The timeout value for attempted shrinkage, includes
- * some slop to cope with system timer imprecision.
+ * Tolerance for idle timeouts, to cope with timer undershoots
*/
- private static final long SHRINK_TIMEOUT = SHRINK_RATE - (SHRINK_RATE / 10);
+ private static final long TIMEOUT_SLOP = 2000000L;
/**
* The maximum stolen->joining link depth allowed in method
- * tryHelpStealer. Must be a power of two. This value also
- * controls the maximum number of times to try to help join a task
- * without any apparent progress or change in pool state before
- * giving up and blocking (see awaitJoin). Depths for legitimate
+ * tryHelpStealer. Must be a power of two. Depths for legitimate
* chains are unbounded, but we use a fixed constant to avoid
* (otherwise unchecked) cycles and to bound staleness of
* traversal parameters at the expense of sometimes blocking when
@@ -1142,22 +1137,12 @@ public class ForkJoinPool extends Abstra
private static final int MAX_HELP = 64;
/**
- * Secondary time-based bound (in nanosecs) for helping attempts
- * before trying compensated blocking in awaitJoin. Used in
- * conjunction with MAX_HELP to reduce variance due to different
- * polling rates associated with different helping options. The
- * value should roughly approximate the time required to create
- * and/or activate a worker thread.
- */
- private static final long COMPENSATION_DELAY = 1L << 18; // ~0.25 millisec
-
- /**
* Increment for seed generators. See class ThreadLocal for
* explanation.
*/
private static final int SEED_INCREMENT = 0x61c88647;
- /**
+ /*
* Bits and masks for control variables
*
* Field ctl is a long packed with:
@@ -1185,14 +1170,14 @@ public class ForkJoinPool extends Abstra
* scan for them to avoid queuing races. Note however that
* eventCount updates lag releases so usage requires care.
*
- * Field runState is an int packed with:
+ * Field plock is an int packed with:
* SHUTDOWN: true if shutdown is enabled (1 bit)
- * SEQ: a sequence number updated upon (de)registering workers (30 bits)
- * INIT: set true after workQueues array construction (1 bit)
+ * SEQ: a sequence lock, with PL_LOCK bit set if locked (30 bits)
+ * SIGNAL: set when threads may be waiting on the lock (1 bit)
*
* The sequence number enables simple consistency checks:
* Staleness of read-only operations on the workQueues array can
- * be checked by comparing runState before vs after the reads.
+ * be checked by comparing plock before vs after the reads.
*/
// bit positions/shifts for fields
@@ -1204,7 +1189,8 @@ public class ForkJoinPool extends Abstra
// bounds
private static final int SMASK = 0xffff; // short bits
private static final int MAX_CAP = 0x7fff; // max #workers - 1
- private static final int SQMASK = 0xfffe; // even short bits
+ private static final int EVENMASK = 0xfffe; // even short bits
+ private static final int SQMASK = 0x007e; // max 64 (even) slots
private static final int SHORT_SIGN = 1 << 15;
private static final int INT_SIGN = 1 << 31;
@@ -1229,91 +1215,142 @@ public class ForkJoinPool extends Abstra
private static final int E_MASK = 0x7fffffff; // no STOP_BIT
private static final int E_SEQ = 1 << EC_SHIFT;
- // runState bits
+ // plock bits
private static final int SHUTDOWN = 1 << 31;
+ private static final int PL_LOCK = 2;
+ private static final int PL_SIGNAL = 1;
+ private static final int PL_SPINS = 1 << 8;
// access mode for WorkQueue
static final int LIFO_QUEUE = 0;
static final int FIFO_QUEUE = 1;
static final int SHARED_QUEUE = -1;
- // Instance fields
-
- /*
- * Field layout order in this class tends to matter more than one
- * would like. Runtime layout order is only loosely related to
- * declaration order and may differ across JVMs, but the following
- * empirically works OK on current JVMs.
- */
+ // Heuristic padding to ameliorate unfortunate memory placements
+ volatile long pad00, pad01, pad02, pad03, pad04, pad05, pad06;
+ // Instance fields
+ volatile long stealCount; // collects worker counts
volatile long ctl; // main pool control
- final int parallelism; // parallelism level
- final int localMode; // per-worker scheduling mode
- final int submitMask; // submit queue index bound
- int nextSeed; // for initializing worker seeds
- volatile int runState; // shutdown status and seq
+ volatile int plock; // shutdown status and seqLock
+ volatile int indexSeed; // worker/submitter index seed
+ final short parallelism; // parallelism level
+ final short mode; // LIFO/FIFO
WorkQueue[] workQueues; // main registry
- final Mutex lock; // for registration
- final Condition termination; // for awaitTermination
- final ForkJoinWorkerThreadFactory factory; // factory for new workers
- final Thread.UncaughtExceptionHandler ueh; // per-worker UEH
- final AtomicLong stealCount; // collect counts when terminated
- final AtomicInteger nextWorkerNumber; // to create worker name string
+ final ForkJoinWorkerThreadFactory factory;
+ final UncaughtExceptionHandler ueh; // per-worker UEH
final String workerNamePrefix; // to create worker name string
- // Creating, registering, and deregistering workers
+ volatile Object pad10, pad11, pad12, pad13, pad14, pad15, pad16, pad17;
+ volatile Object pad18, pad19, pad1a, pad1b;
/**
- * Tries to create and start a worker
- */
- private void addWorker() {
- Throwable ex = null;
- ForkJoinWorkerThread wt = null;
- try {
- if ((wt = factory.newThread(this)) != null) {
- wt.start();
- return;
+ * Acquires the plock lock to protect worker array and related
+ * updates. This method is called only if an initial CAS on plock
+ * fails. This acts as a spinlock for normal cases, but falls back
+ * to builtin monitor to block when (rarely) needed. This would be
+ * a terrible idea for a highly contended lock, but works fine as
+ * a more conservative alternative to a pure spinlock.
+ */
+ private int acquirePlock() {
+ int spins = PL_SPINS, ps, nps;
+ for (;;) {
+ if (((ps = plock) & PL_LOCK) == 0 &&
+ U.compareAndSwapInt(this, PLOCK, ps, nps = ps + PL_LOCK))
+ return nps;
+ else if (spins >= 0) {
+ if (ThreadLocalRandom.current().nextInt() >= 0)
+ --spins;
+ }
+ else if (U.compareAndSwapInt(this, PLOCK, ps, ps | PL_SIGNAL)) {
+ synchronized (this) {
+ if ((plock & PL_SIGNAL) != 0) {
+ try {
+ wait();
+ } catch (InterruptedException ie) {
+ try {
+ Thread.currentThread().interrupt();
+ } catch (SecurityException ignore) {
+ }
+ }
+ }
+ else
+ notifyAll();
+ }
}
- } catch (Throwable e) {
- ex = e;
}
- deregisterWorker(wt, ex); // adjust counts etc on failure
}
/**
- * Callback from ForkJoinWorkerThread constructor to assign a
- * public name. This must be separate from registerWorker because
- * it is called during the "super" constructor call in
- * ForkJoinWorkerThread.
+ * Unlocks and signals any thread waiting for plock. Called only
+ * when CAS of seq value for unlock fails.
*/
- final String nextWorkerName() {
- return workerNamePrefix.concat
- (Integer.toString(nextWorkerNumber.addAndGet(1)));
+ private void releasePlock(int ps) {
+ plock = ps;
+ synchronized (this) { notifyAll(); }
}
/**
- * Callback from ForkJoinWorkerThread constructor to establish its
- * poolIndex and record its WorkQueue. To avoid scanning bias due
- * to packing entries in front of the workQueues array, we treat
- * the array as a simple power-of-two hash table using per-thread
- * seed as hash, expanding as needed.
- *
- * @param w the worker's queue
+ * Tries to create and start one worker if fewer than target
+ * parallelism level exist. Adjusts counts etc on failure.
*/
+ private void tryAddWorker() {
+ long c; int u, e;
+ while ((u = (int)((c = ctl) >>> 32)) < 0 &&
+ (u & SHORT_SIGN) != 0 && (e = (int)c) >= 0) {
+ long nc = ((long)(((u + UTC_UNIT) & UTC_MASK) |
+ ((u + UAC_UNIT) & UAC_MASK)) << 32) | (long)e;
+ if (U.compareAndSwapLong(this, CTL, c, nc)) {
+ ForkJoinWorkerThreadFactory fac;
+ Throwable ex = null;
+ ForkJoinWorkerThread wt = null;
+ try {
+ if ((fac = factory) != null &&
+ (wt = fac.newThread(this)) != null) {
+ wt.start();
+ break;
+ }
+ } catch (Throwable rex) {
+ ex = rex;
+ }
+ deregisterWorker(wt, ex);
+ break;
+ }
+ }
+ }
- final void registerWorker(WorkQueue w) {
- Mutex lock = this.lock;
- lock.lock();
+ // Registering and deregistering workers
+
+ /**
+ * Callback from ForkJoinWorkerThread to establish and record its
+ * WorkQueue. To avoid scanning bias due to packing entries in
+ * front of the workQueues array, we treat the array as a simple
+ * power-of-two hash table using per-thread seed as hash,
+ * expanding as needed.
+ *
+ * @param wt the worker thread
+ * @return the worker's queue
+ */
+ final WorkQueue registerWorker(ForkJoinWorkerThread wt) {
+ UncaughtExceptionHandler handler; WorkQueue[] ws; int s, ps;
+ wt.setDaemon(true);
+ if ((handler = ueh) != null)
+ wt.setUncaughtExceptionHandler(handler);
+ do {} while (!U.compareAndSwapInt(this, INDEXSEED, s = indexSeed,
+ s += SEED_INCREMENT) ||
+ s == 0); // skip 0
+ WorkQueue w = new WorkQueue(this, wt, mode, s);
+ if (((ps = plock) & PL_LOCK) != 0 ||
+ !U.compareAndSwapInt(this, PLOCK, ps, ps += PL_LOCK))
+ ps = acquirePlock();
+ int nps = (ps & SHUTDOWN) | ((ps + PL_LOCK) & ~SHUTDOWN);
try {
- WorkQueue[] ws = workQueues;
- if (w != null && ws != null) { // skip on shutdown/failure
- int rs, n = ws.length, m = n - 1;
- int s = nextSeed += SEED_INCREMENT; // rarely-colliding sequence
- w.seed = (s == 0) ? 1 : s; // ensure non-zero seed
- int r = (s << 1) | 1; // use odd-numbered indices
- if (ws[r &= m] != null) { // collision
- int probes = 0; // step by approx half size
- int step = (n <= 4) ? 2 : ((n >>> 1) & SQMASK) + 2;
+ if ((ws = workQueues) != null) { // skip if shutting down
+ int n = ws.length, m = n - 1;
+ int r = (s << 1) | 1; // use odd-numbered indices
+ if (ws[r &= m] != null) { // collision
+ int probes = 0; // step by approx half size
+ int step = (n <= 4) ? 2 : ((n >>> 1) & EVENMASK) + 2;
while (ws[r = (r + step) & m] != null) {
if (++probes >= n) {
workQueues = ws = Arrays.copyOf(ws, n <<= 1);
@@ -1322,104 +1359,230 @@ public class ForkJoinPool extends Abstra
}
}
}
- w.eventCount = w.poolIndex = r; // establish before recording
- ws[r] = w; // also update seq
- runState = ((rs = runState) & SHUTDOWN) | ((rs + 2) & ~SHUTDOWN);
+ w.poolIndex = (short)r;
+ w.eventCount = r; // volatile write orders
+ ws[r] = w;
}
} finally {
- lock.unlock();
+ if (!U.compareAndSwapInt(this, PLOCK, ps, nps))
+ releasePlock(nps);
}
+ wt.setName(workerNamePrefix.concat(Integer.toString(w.poolIndex >>> 1)));
+ return w;
}
/**
* Final callback from terminating worker, as well as upon failure
- * to construct or start a worker in addWorker. Removes record of
- * worker from array, and adjusts counts. If pool is shutting
- * down, tries to complete termination.
+ * to construct or start a worker. Removes record of worker from
+ * array, and adjusts counts. If pool is shutting down, tries to
+ * complete termination.
*
- * @param wt the worker thread or null if addWorker failed
+ * @param wt the worker thread, or null if construction failed
* @param ex the exception causing failure, or null if none
*/
final void deregisterWorker(ForkJoinWorkerThread wt, Throwable ex) {
- Mutex lock = this.lock;
WorkQueue w = null;
if (wt != null && (w = wt.workQueue) != null) {
- w.runState = -1; // ensure runState is set
- stealCount.getAndAdd(w.totalSteals + w.nsteals);
- int idx = w.poolIndex;
- lock.lock();
- try { // remove record from array
+ int ps; long sc;
+ w.qlock = -1; // ensure set
+ do {} while (!U.compareAndSwapLong(this, STEALCOUNT,
+ sc = stealCount,
+ sc + w.nsteals));
+ if (((ps = plock) & PL_LOCK) != 0 ||
+ !U.compareAndSwapInt(this, PLOCK, ps, ps += PL_LOCK))
+ ps = acquirePlock();
+ int nps = (ps & SHUTDOWN) | ((ps + PL_LOCK) & ~SHUTDOWN);
+ try {
+ int idx = w.poolIndex;
WorkQueue[] ws = workQueues;
if (ws != null && idx >= 0 && idx < ws.length && ws[idx] == w)
ws[idx] = null;
} finally {
- lock.unlock();
+ if (!U.compareAndSwapInt(this, PLOCK, ps, nps))
+ releasePlock(nps);
}
}
- long c; // adjust ctl counts
+ long c; // adjust ctl counts
do {} while (!U.compareAndSwapLong
(this, CTL, c = ctl, (((c - AC_UNIT) & AC_MASK) |
((c - TC_UNIT) & TC_MASK) |
(c & ~(AC_MASK|TC_MASK)))));
- if (!tryTerminate(false, false) && w != null) {
- w.cancelAll(); // cancel remaining tasks
- if (w.array != null) // suppress signal if never ran
- signalWork(); // wake up or create replacement
- if (ex == null) // help clean refs on way out
- ForkJoinTask.helpExpungeStaleExceptions();
+ if (!tryTerminate(false, false) && w != null && w.array != null) {
+ w.cancelAll(); // cancel remaining tasks
+ WorkQueue[] ws; WorkQueue v; Thread p; int u, i, e;
+ while ((u = (int)((c = ctl) >>> 32)) < 0 && (e = (int)c) >= 0) {
+ if (e > 0) { // activate or create replacement
+ if ((ws = workQueues) == null ||
+ (i = e & SMASK) >= ws.length ||
+ (v = ws[i]) == null)
+ break;
+ long nc = (((long)(v.nextWait & E_MASK)) |
+ ((long)(u + UAC_UNIT) << 32));
+ if (v.eventCount != (e | INT_SIGN))
+ break;
+ if (U.compareAndSwapLong(this, CTL, c, nc)) {
+ v.eventCount = (e + E_SEQ) & E_MASK;
+ if ((p = v.parker) != null)
+ U.unpark(p);
+ break;
+ }
+ }
+ else {
+ if ((short)u < 0)
+ tryAddWorker();
+ break;
+ }
+ }
}
-
- if (ex != null) // rethrow
- U.throwException(ex);
+ if (ex == null) // help clean refs on way out
+ ForkJoinTask.helpExpungeStaleExceptions();
+ else // rethrow
+ ForkJoinTask.rethrow(ex);
}
-
// Submissions
/**
+ * Per-thread records for threads that submit to pools. Currently
+ * holds only pseudo-random seed / index that is used to choose
+ * submission queues in method externalPush. In the future, this may
+ * also incorporate a means to implement different task rejection
+ * and resubmission policies.
+ *
+ * Seeds for submitters and workers/workQueues work in basically
+ * the same way but are initialized and updated using slightly
+ * different mechanics. Both are initialized using the same
+ * approach as in class ThreadLocal, where successive values are
+ * unlikely to collide with previous values. Seeds are then
+ * randomly modified upon collisions using xorshifts, which
+ * requires a non-zero seed.
+ */
+ static final class Submitter {
+ int seed;
+ Submitter(int s) { seed = s; }
+ }
+
+ /**
* Unless shutting down, adds the given task to a submission queue
* at submitter's current queue index (modulo submission
- * range). If no queue exists at the index, one is created. If
- * the queue is busy, another index is randomly chosen. The
- * submitMask bounds the effective number of queues to the
- * (nearest power of two for) parallelism level.
+ * range). Only the most common path is directly handled in this
+ * method. All others are relayed to fullExternalPush.
*
* @param task the task. Caller must ensure non-null.
*/
- private void doSubmit(ForkJoinTask task) {
- Submitter s = submitters.get();
- for (int r = s.seed, m = submitMask;;) {
- WorkQueue[] ws; WorkQueue q;
- int k = r & m & SQMASK; // use only even indices
- if (runState < 0 || (ws = workQueues) == null || ws.length <= k)
- throw new RejectedExecutionException(); // shutting down
- else if ((q = ws[k]) == null) { // create new queue
- WorkQueue nq = new WorkQueue(this, null, SHARED_QUEUE);
- Mutex lock = this.lock; // construct outside lock
- lock.lock();
- try { // recheck under lock
- int rs = runState; // to update seq
- if (ws == workQueues && ws[k] == null) {
- ws[k] = nq;
- runState = ((rs & SHUTDOWN) | ((rs + 2) & ~SHUTDOWN));
- }
- } finally {
- lock.unlock();
- }
- }
- else if (q.trySharedPush(task)) {
- signalWork();
+ final void externalPush(ForkJoinTask task) {
+ Submitter z = submitters.get();
+ WorkQueue q; int r, m, s, n, am; ForkJoinTask[] a;
+ int ps = plock;
+ WorkQueue[] ws = workQueues;
+ if (z != null && ps > 0 && ws != null && (m = (ws.length - 1)) >= 0 &&
+ (q = ws[m & (r = z.seed) & SQMASK]) != null && r != 0 &&
+ U.compareAndSwapInt(q, QLOCK, 0, 1)) { // lock
+ if ((a = q.array) != null &&
+ (am = a.length - 1) > (n = (s = q.top) - q.base)) {
+ int j = ((am & s) << ASHIFT) + ABASE;
+ U.putOrderedObject(a, j, task);
+ q.top = s + 1; // push on to deque
+ q.qlock = 0;
+ if (n <= 1)
+ signalWork(ws, q);
return;
}
- else if (m > 1) { // move to a different index
- r ^= r << 13; // same xorshift as WorkQueues
+ q.qlock = 0;
+ }
+ fullExternalPush(task);
+ }
+
+ /**
+ * Full version of externalPush. This method is called, among
+ * other times, upon the first submission of the first task to the
+ * pool, so must perform secondary initialization. It also
+ * detects first submission by an external thread by looking up
+ * its ThreadLocal, and creates a new shared queue if the one at
+ * index if empty or contended. The plock lock body must be
+ * exception-free (so no try/finally) so we optimistically
+ * allocate new queues outside the lock and throw them away if
+ * (very rarely) not needed.
+ *
+ * Secondary initialization occurs when plock is zero, to create
+ * workQueue array and set plock to a valid value. This lock body
+ * must also be exception-free. Because the plock seq value can
+ * eventually wrap around zero, this method harmlessly fails to
+ * reinitialize if workQueues exists, while still advancing plock.
+ */
+ private void fullExternalPush(ForkJoinTask task) {
+ int r = 0; // random index seed
+ for (Submitter z = submitters.get();;) {
+ WorkQueue[] ws; WorkQueue q; int ps, m, k;
+ if (z == null) {
+ if (U.compareAndSwapInt(this, INDEXSEED, r = indexSeed,
+ r += SEED_INCREMENT) && r != 0)
+ submitters.set(z = new Submitter(r));
+ }
+ else if (r == 0) { // move to a different index
+ r = z.seed;
+ r ^= r << 13; // same xorshift as WorkQueues
r ^= r >>> 17;
- s.seed = r ^= r << 5;
+ z.seed = r ^= (r << 5);
+ }
+ if ((ps = plock) < 0)
+ throw new RejectedExecutionException();
+ else if (ps == 0 || (ws = workQueues) == null ||
+ (m = ws.length - 1) < 0) { // initialize workQueues
+ int p = parallelism; // find power of two table size
+ int n = (p > 1) ? p - 1 : 1; // ensure at least 2 slots
+ n |= n >>> 1; n |= n >>> 2; n |= n >>> 4;
+ n |= n >>> 8; n |= n >>> 16; n = (n + 1) << 1;
+ WorkQueue[] nws = ((ws = workQueues) == null || ws.length == 0 ?
+ new WorkQueue[n] : null);
+ if (((ps = plock) & PL_LOCK) != 0 ||
+ !U.compareAndSwapInt(this, PLOCK, ps, ps += PL_LOCK))
+ ps = acquirePlock();
+ if (((ws = workQueues) == null || ws.length == 0) && nws != null)
+ workQueues = nws;
+ int nps = (ps & SHUTDOWN) | ((ps + PL_LOCK) & ~SHUTDOWN);
+ if (!U.compareAndSwapInt(this, PLOCK, ps, nps))
+ releasePlock(nps);
+ }
+ else if ((q = ws[k = r & m & SQMASK]) != null) {
+ if (q.qlock == 0 && U.compareAndSwapInt(q, QLOCK, 0, 1)) {
+ ForkJoinTask[] a = q.array;
+ int s = q.top;
+ boolean submitted = false;
+ try { // locked version of push
+ if ((a != null && a.length > s + 1 - q.base) ||
+ (a = q.growArray()) != null) { // must presize
+ int j = (((a.length - 1) & s) << ASHIFT) + ABASE;
+ U.putOrderedObject(a, j, task);
+ q.top = s + 1;
+ submitted = true;
+ }
+ } finally {
+ q.qlock = 0; // unlock
+ }
+ if (submitted) {
+ signalWork(ws, q);
+ return;
+ }
+ }
+ r = 0; // move on failure
+ }
+ else if (((ps = plock) & PL_LOCK) == 0) { // create new queue
+ q = new WorkQueue(this, null, SHARED_QUEUE, r);
+ q.poolIndex = (short)k;
+ if (((ps = plock) & PL_LOCK) != 0 ||
+ !U.compareAndSwapInt(this, PLOCK, ps, ps += PL_LOCK))
+ ps = acquirePlock();
+ if ((ws = workQueues) != null && k < ws.length && ws[k] == null)
+ ws[k] = q;
+ int nps = (ps & SHUTDOWN) | ((ps + PL_LOCK) & ~SHUTDOWN);
+ if (!U.compareAndSwapInt(this, PLOCK, ps, nps))
+ releasePlock(nps);
}
else
- Thread.yield(); // yield if no alternatives
+ r = 0;
}
}
@@ -1430,40 +1593,41 @@ public class ForkJoinPool extends Abstra
*/
final void incrementActiveCount() {
long c;
- do {} while (!U.compareAndSwapLong(this, CTL, c = ctl, c + AC_UNIT));
+ do {} while (!U.compareAndSwapLong
+ (this, CTL, c = ctl, ((c & ~AC_MASK) |
+ ((c & AC_MASK) + AC_UNIT))));
}
/**
- * Tries to activate or create a worker if too few are active.
+ * Tries to create or activate a worker if too few are active.
+ *
+ * @param ws the worker array to use to find signallees
+ * @param q if non-null, the queue holding tasks to be processed
*/
- final void signalWork() {
- long c; int u;
- while ((u = (int)((c = ctl) >>> 32)) < 0) { // too few active
- WorkQueue[] ws = workQueues; int e, i; WorkQueue w; Thread p;
- if ((e = (int)c) > 0) { // at least one waiting
- if (ws != null && (i = e & SMASK) < ws.length &&
- (w = ws[i]) != null && w.eventCount == (e | INT_SIGN)) {
- long nc = (((long)(w.nextWait & E_MASK)) |
- ((long)(u + UAC_UNIT) << 32));
- if (U.compareAndSwapLong(this, CTL, c, nc)) {
- w.eventCount = (e + E_SEQ) & E_MASK;
- if ((p = w.parker) != null)
- U.unpark(p); // activate and release
- break;
- }
- }
- else
- break;
+ final void signalWork(WorkQueue[] ws, WorkQueue q) {
+ for (;;) {
+ long c; int e, u, i; WorkQueue w; Thread p;
+ if ((u = (int)((c = ctl) >>> 32)) >= 0)
+ break;
+ if ((e = (int)c) <= 0) {
+ if ((short)u < 0)
+ tryAddWorker();
+ break;
}
- else if (e == 0 && (u & SHORT_SIGN) != 0) { // too few total
- long nc = (long)(((u + UTC_UNIT) & UTC_MASK) |
- ((u + UAC_UNIT) & UAC_MASK)) << 32;
- if (U.compareAndSwapLong(this, CTL, c, nc)) {
- addWorker();
- break;
- }
+ if (ws == null || ws.length <= (i = e & SMASK) ||
+ (w = ws[i]) == null)
+ break;
+ long nc = (((long)(w.nextWait & E_MASK)) |
+ ((long)(u + UAC_UNIT)) << 32);
+ int ne = (e + E_SEQ) & E_MASK;
+ if (w.eventCount == (e | INT_SIGN) &&
+ U.compareAndSwapLong(this, CTL, c, nc)) {
+ w.eventCount = ne;
+ if ((p = w.parker) != null)
+ U.unpark(p);
+ break;
}
- else
+ if (q != null && q.base >= q.top)
break;
}
}
@@ -1474,167 +1638,155 @@ public class ForkJoinPool extends Abstra
* Top-level runloop for workers, called by ForkJoinWorkerThread.run.
*/
final void runWorker(WorkQueue w) {
- w.growArray(false); // initialize queue array in this thread
- do { w.runTask(scan(w)); } while (w.runState >= 0);
+ w.growArray(); // allocate queue
+ for (int r = w.hint; scan(w, r) == 0; ) {
+ r ^= r << 13; r ^= r >>> 17; r ^= r << 5; // xorshift
+ }
}
/**
- * Scans for and, if found, returns one task, else possibly
+ * Scans for and, if found, runs one task, else possibly
* inactivates the worker. This method operates on single reads of
* volatile state and is designed to be re-invoked continuously,
* in part because it returns upon detecting inconsistencies,
* contention, or state changes that indicate possible success on
* re-invocation.
*
- * The scan searches for tasks across a random permutation of
- * queues (starting at a random index and stepping by a random
- * relative prime, checking each at least once). The scan
- * terminates upon either finding a non-empty queue, or completing
- * the sweep. If the worker is not inactivated, it takes and
- * returns a task from this queue. On failure to find a task, we
- * take one of the following actions, after which the caller will
- * retry calling this method unless terminated.
- *
- * * If pool is terminating, terminate the worker.
- *
- * * If not a complete sweep, try to release a waiting worker. If
- * the scan terminated because the worker is inactivated, then the
- * released worker will often be the calling worker, and it can
- * succeed obtaining a task on the next call. Or maybe it is
- * another worker, but with same net effect. Releasing in other
- * cases as well ensures that we have enough workers running.
- *
- * * If not already enqueued, try to inactivate and enqueue the
- * worker on wait queue. Or, if inactivating has caused the pool
- * to be quiescent, relay to idleAwaitWork to check for
- * termination and possibly shrink pool.
- *
- * * If already inactive, and the caller has run a task since the
- * last empty scan, return (to allow rescan) unless others are
- * also inactivated. Field WorkQueue.rescans counts down on each
- * scan to ensure eventual inactivation and blocking.
- *
- * * If already enqueued and none of the above apply, park
- * awaiting signal,
+ * The scan searches for tasks across queues starting at a random
+ * index, checking each at least twice. The scan terminates upon
+ * either finding a non-empty queue, or completing the sweep. If
+ * the worker is not inactivated, it takes and runs a task from
+ * this queue. Otherwise, if not activated, it tries to activate
+ * itself or some other worker by signalling. On failure to find a
+ * task, returns (for retry) if pool state may have changed during
+ * an empty scan, or tries to inactivate if active, else possibly
+ * blocks or terminates via method awaitWork.
*
* @param w the worker (via its WorkQueue)
- * @return a task or null of none found
+ * @param r a random seed
+ * @return worker qlock status if would have waited, else 0
*/
- private final ForkJoinTask scan(WorkQueue w) {
- WorkQueue[] ws; // first update random seed
- int r = w.seed; r ^= r << 13; r ^= r >>> 17; w.seed = r ^= r << 5;
- int rs = runState, m; // volatile read order matters
- if ((ws = workQueues) != null && (m = ws.length - 1) > 0) {
- int ec = w.eventCount; // ec is negative if inactive
- int step = (r >>> 16) | 1; // relative prime
- for (int j = (m + 1) << 2; ; r += step) {
- WorkQueue q; ForkJoinTask t; ForkJoinTask[] a; int b;
- if ((q = ws[r & m]) != null && (b = q.base) - q.top < 0 &&
- (a = q.array) != null) { // probably nonempty
- int i = (((a.length - 1) & b) << ASHIFT) + ABASE;
- t = (ForkJoinTask)U.getObjectVolatile(a, i);
- if (q.base == b && ec >= 0 && t != null &&
- U.compareAndSwapObject(a, i, t, null)) {
- if (q.top - (q.base = b + 1) > 1)
- signalWork(); // help pushes signal
- return t;
- }
- else if (ec < 0 || j <= m) {
- rs = 0; // mark scan as imcomplete
- break; // caller can retry after release
+ private final int scan(WorkQueue w, int r) {
+ WorkQueue[] ws; int m;
+ long c = ctl; // for consistency check
+ if ((ws = workQueues) != null && (m = ws.length - 1) >= 0 && w != null) {
+ for (int j = m + m + 1, ec = w.eventCount;;) {
+ WorkQueue q; int b, e; ForkJoinTask[] a; ForkJoinTask t;
+ if ((q = ws[(r - j) & m]) != null &&
+ (b = q.base) - q.top < 0 && (a = q.array) != null) {
+ long i = (((a.length - 1) & b) << ASHIFT) + ABASE;
+ if ((t = ((ForkJoinTask)
+ U.getObjectVolatile(a, i))) != null) {
+ if (ec < 0)
+ helpRelease(c, ws, w, q, b);
+ else if (q.base == b &&
+ U.compareAndSwapObject(a, i, t, null)) {
+ U.putOrderedInt(q, QBASE, b + 1);
+ if ((b + 1) - q.top < 0)
+ signalWork(ws, q);
+ w.runTask(t);
+ }
}
- }
- if (--j < 0)
break;
- }
-
- long c = ctl; int e = (int)c, a = (int)(c >> AC_SHIFT), nr, ns;
- if (e < 0) // decode ctl on empty scan
- w.runState = -1; // pool is terminating
- else if (rs == 0 || rs != runState) { // incomplete scan
- WorkQueue v; Thread p; // try to release a waiter
- if (e > 0 && a < 0 && w.eventCount == ec &&
- (v = ws[e & m]) != null && v.eventCount == (e | INT_SIGN)) {
- long nc = ((long)(v.nextWait & E_MASK) |
- ((c + AC_UNIT) & (AC_MASK|TC_MASK)));
- if (ctl == c && U.compareAndSwapLong(this, CTL, c, nc)) {
- v.eventCount = (e + E_SEQ) & E_MASK;
- if ((p = v.parker) != null)
- U.unpark(p);
+ }
+ else if (--j < 0) {
+ if ((ec | (e = (int)c)) < 0) // inactive or terminating
+ return awaitWork(w, c, ec);
+ else if (ctl == c) { // try to inactivate and enqueue
+ long nc = (long)ec | ((c - AC_UNIT) & (AC_MASK|TC_MASK));
+ w.nextWait = e;
+ w.eventCount = ec | INT_SIGN;
+ if (!U.compareAndSwapLong(this, CTL, c, nc))
+ w.eventCount = ec; // back out
}
+ break;
}
}
- else if (ec >= 0) { // try to enqueue/inactivate
- long nc = (long)ec | ((c - AC_UNIT) & (AC_MASK|TC_MASK));
- w.nextWait = e;
- w.eventCount = ec | INT_SIGN; // mark as inactive
- if (ctl != c || !U.compareAndSwapLong(this, CTL, c, nc))
- w.eventCount = ec; // unmark on CAS failure
- else {
- if ((ns = w.nsteals) != 0) {
- w.nsteals = 0; // set rescans if ran task
- w.rescans = (a > 0) ? 0 : a + parallelism;
- w.totalSteals += ns;
- }
- if (a == 1 - parallelism) // quiescent
- idleAwaitWork(w, nc, c);
- }
+ }
+ return 0;
+ }
+
+ /**
+ * A continuation of scan(), possibly blocking or terminating
+ * worker w. Returns without blocking if pool state has apparently
+ * changed since last invocation. Also, if inactivating w has
+ * caused the pool to become quiescent, checks for pool
+ * termination, and, so long as this is not the only worker, waits
+ * for event for up to a given duration. On timeout, if ctl has
+ * not changed, terminates the worker, which will in turn wake up
+ * another worker to possibly repeat this process.
+ *
+ * @param w the calling worker
+ * @param c the ctl value on entry to scan
+ * @param ec the worker's eventCount on entry to scan
+ */
+ private final int awaitWork(WorkQueue w, long c, int ec) {
+ int stat, ns; long parkTime, deadline;
+ if ((stat = w.qlock) >= 0 && w.eventCount == ec && ctl == c &&
+ !Thread.interrupted()) {
+ int e = (int)c;
+ int u = (int)(c >>> 32);
+ int d = (u >> UAC_SHIFT) + parallelism; // active count
+
+ if (e < 0 || (d <= 0 && tryTerminate(false, false)))
+ stat = w.qlock = -1; // pool is terminating
+ else if ((ns = w.nsteals) != 0) { // collect steals and retry
+ long sc;
+ w.nsteals = 0;
+ do {} while (!U.compareAndSwapLong(this, STEALCOUNT,
+ sc = stealCount, sc + ns));
}
- else if (w.eventCount < 0) { // already queued
- if ((nr = w.rescans) > 0) { // continue rescanning
- int ac = a + parallelism;
- if (((w.rescans = (ac < nr) ? ac : nr - 1) & 3) == 0)
- Thread.yield(); // yield before block
+ else {
+ long pc = ((d > 0 || ec != (e | INT_SIGN)) ? 0L :
+ ((long)(w.nextWait & E_MASK)) | // ctl to restore
+ ((long)(u + UAC_UNIT)) << 32);
+ if (pc != 0L) { // timed wait if last waiter
+ int dc = -(short)(c >>> TC_SHIFT);
+ parkTime = (dc < 0 ? FAST_IDLE_TIMEOUT:
+ (dc + 1) * IDLE_TIMEOUT);
+ deadline = System.nanoTime() + parkTime - TIMEOUT_SLOP;
}
- else {
- Thread.interrupted(); // clear status
+ else
+ parkTime = deadline = 0L;
+ if (w.eventCount == ec && ctl == c) {
Thread wt = Thread.currentThread();
U.putObject(wt, PARKBLOCKER, this);
w.parker = wt; // emulate LockSupport.park
- if (w.eventCount < 0) // recheck
- U.park(false, 0L);
+ if (w.eventCount == ec && ctl == c)
+ U.park(false, parkTime); // must recheck before park
w.parker = null;
U.putObject(wt, PARKBLOCKER, null);
+ if (parkTime != 0L && ctl == c &&
+ deadline - System.nanoTime() <= 0L &&
+ U.compareAndSwapLong(this, CTL, c, pc))
+ stat = w.qlock = -1; // shrink pool
}
}
}
- return null;
+ return stat;
}
/**
- * If inactivating worker w has caused the pool to become
- * quiescent, checks for pool termination, and, so long as this is
- * not the only worker, waits for event for up to SHRINK_RATE
- * nanosecs. On timeout, if ctl has not changed, terminates the
- * worker, which will in turn wake up another worker to possibly
- * repeat this process.
- *
- * @param w the calling worker
- * @param currentCtl the ctl value triggering possible quiescence
- * @param prevCtl the ctl value to restore if thread is terminated
- */
- private void idleAwaitWork(WorkQueue w, long currentCtl, long prevCtl) {
- if (w.eventCount < 0 && !tryTerminate(false, false) &&
- (int)prevCtl != 0 && !hasQueuedSubmissions() && ctl == currentCtl) {
- Thread wt = Thread.currentThread();
- Thread.yield(); // yield before block
- while (ctl == currentCtl) {
- long startTime = System.nanoTime();
- Thread.interrupted(); // timed variant of version in scan()
- U.putObject(wt, PARKBLOCKER, this);
- w.parker = wt;
- if (ctl == currentCtl)
- U.park(false, SHRINK_RATE);
- w.parker = null;
- U.putObject(wt, PARKBLOCKER, null);
- if (ctl != currentCtl)
- break;
- if (System.nanoTime() - startTime >= SHRINK_TIMEOUT &&
- U.compareAndSwapLong(this, CTL, currentCtl, prevCtl)) {
- w.eventCount = (w.eventCount + E_SEQ) | E_MASK;
- w.runState = -1; // shrink
- break;
- }
+ * Possibly releases (signals) a worker. Called only from scan()
+ * when a worker with apparently inactive status finds a non-empty
+ * queue. This requires revalidating all of the associated state
+ * from caller.
+ */
+ private final void helpRelease(long c, WorkQueue[] ws, WorkQueue w,
+ WorkQueue q, int b) {
+ WorkQueue v; int e, i; Thread p;
+ if (w != null && w.eventCount < 0 && (e = (int)c) > 0 &&
+ ws != null && ws.length > (i = e & SMASK) &&
+ (v = ws[i]) != null && ctl == c) {
+ long nc = (((long)(v.nextWait & E_MASK)) |
+ ((long)((int)(c >>> 32) + UAC_UNIT)) << 32);
+ int ne = (e + E_SEQ) & E_MASK;
+ if (q != null && q.base == b && w.eventCount < 0 &&
+ v.eventCount == (e | INT_SIGN) &&
+ U.compareAndSwapLong(this, CTL, c, nc)) {
+ v.eventCount = ne;
+ if ((p = v.parker) != null)
+ U.unpark(p);
}
}
}
@@ -1659,7 +1811,8 @@ public class ForkJoinPool extends Abstra
*/
private int tryHelpStealer(WorkQueue joiner, ForkJoinTask task) {
int stat = 0, steps = 0; // bound to avoid cycles
- if (joiner != null && task != null) { // hoist null checks
+ if (task != null && joiner != null &&
+ joiner.base - joiner.top >= 0) { // hoist checks
restart: for (;;) {
ForkJoinTask subtask = task; // current target
for (WorkQueue j = joiner, v;;) { // v is stealer of subtask
@@ -1670,7 +1823,7 @@ public class ForkJoinPool extends Abstra
}
if ((ws = workQueues) == null || (m = ws.length - 1) <= 0)
break restart; // shutting down
- if ((v = ws[h = (j.stealHint | 1) & m]) == null ||
+ if ((v = ws[h = (j.hint | 1) & m]) == null ||
v.currentSteal != subtask) {
for (int origin = h;;) { // find stealer
if (((h = (h + 2) & m) & 15) == 1 &&
@@ -1678,7 +1831,7 @@ public class ForkJoinPool extends Abstra
continue restart; // occasional staleness check
if ((v = ws[h]) != null &&
v.currentSteal == subtask) {
- j.stealHint = h; // save hint
+ j.hint = h; // save hint
break;
}
if (h == origin)
@@ -1686,7 +1839,7 @@ public class ForkJoinPool extends Abstra
}
}
for (;;) { // help stealer or descend to its stealer
- ForkJoinTask[] a; int b;
+ ForkJoinTask[] a; int b;
if (subtask.status < 0) // surround probes with
continue restart; // consistency checks
if ((b = v.base) - v.top < 0 && (a = v.array) != null) {
@@ -1697,13 +1850,23 @@ public class ForkJoinPool extends Abstra
v.currentSteal != subtask)
continue restart; // stale
stat = 1; // apparent progress
- if (t != null && v.base == b &&
- U.compareAndSwapObject(a, i, t, null)) {
- v.base = b + 1; // help stealer
- joiner.runSubtask(t);
+ if (v.base == b) {
+ if (t == null)
+ break restart;
+ if (U.compareAndSwapObject(a, i, t, null)) {
+ U.putOrderedInt(v, QBASE, b + 1);
+ ForkJoinTask ps = joiner.currentSteal;
+ int jt = joiner.top;
+ do {
+ joiner.currentSteal = t;
+ t.doExec(); // clear local tasks too
+ } while (task.status >= 0 &&
+ joiner.top != jt &&
+ (t = joiner.pop()) != null);
+ joiner.currentSteal = ps;
+ break restart;
+ }
}
- else if (v.base == b && ++steps == MAX_HELP)
- break restart; // v apparently stalled
}
else { // empty -- try to descend
ForkJoinTask next = v.currentJoin;
@@ -1726,88 +1889,88 @@ public class ForkJoinPool extends Abstra
}
/**
- * If task is at base of some steal queue, steals and executes it.
+ * Analog of tryHelpStealer for CountedCompleters. Tries to steal
+ * and run tasks within the target's computation.
*
- * @param joiner the joining worker
- * @param task the task
+ * @param task the task to join
*/
- private void tryPollForAndExec(WorkQueue joiner, ForkJoinTask task) {
- WorkQueue[] ws;
- if ((ws = workQueues) != null) {
- for (int j = 1; j < ws.length && task.status >= 0; j += 2) {
- WorkQueue q = ws[j];
- if (q != null && q.pollFor(task)) {
- joiner.runSubtask(task);
+ private int helpComplete(WorkQueue joiner, CountedCompleter task) {
+ WorkQueue[] ws; int m;
+ int s = 0;
+ if ((ws = workQueues) != null && (m = ws.length - 1) >= 0 &&
+ joiner != null && task != null) {
+ int j = joiner.poolIndex;
+ int scans = m + m + 1;
+ long c = 0L; // for stability check
+ for (int k = scans; ; j += 2) {
+ WorkQueue q;
+ if ((s = task.status) < 0)
+ break;
+ else if (joiner.internalPopAndExecCC(task))
+ k = scans;
+ else if ((s = task.status) < 0)
break;
+ else if ((q = ws[j & m]) != null && q.pollAndExecCC(task))
+ k = scans;
+ else if (--k < 0) {
+ if (c == (c = ctl))
+ break;
+ k = scans;
}
}
}
+ return s;
}
/**
* Tries to decrement active count (sometimes implicitly) and
* possibly release or create a compensating worker in preparation
* for blocking. Fails on contention or termination. Otherwise,
- * adds a new thread if no idle workers are available and either
- * pool would become completely starved or: (at least half
- * starved, and fewer than 50% spares exist, and there is at least
- * one task apparently available). Even though the availability
- * check requires a full scan, it is worthwhile in reducing false
- * alarms.
- *
- * @param task if non-null, a task being waited for
- * @param blocker if non-null, a blocker being waited for
- * @return true if the caller can block, else should recheck and retry
+ * adds a new thread if no idle workers are available and pool
+ * may become starved.
+ *
+ * @param c the assumed ctl value
*/
- final boolean tryCompensate(ForkJoinTask task, ManagedBlocker blocker) {
- int pc = parallelism, e;
- long c = ctl;
+ final boolean tryCompensate(long c) {
WorkQueue[] ws = workQueues;
- if ((e = (int)c) >= 0 && ws != null) {
- int u, a, ac, hc;
- int tc = (short)((u = (int)(c >>> 32)) >>> UTC_SHIFT) + pc;
- boolean replace = false;
- if ((a = u >> UAC_SHIFT) <= 0) {
- if ((ac = a + pc) <= 1)
- replace = true;
- else if ((e > 0 || (task != null &&
- ac <= (hc = pc >>> 1) && tc < pc + hc))) {
- WorkQueue w;
- for (int j = 0; j < ws.length; ++j) {
- if ((w = ws[j]) != null && !w.isEmpty()) {
- replace = true;
- break; // in compensation range and tasks available
- }
- }
- }
+ int pc = parallelism, e = (int)c, m, tc;
+ if (ws != null && (m = ws.length - 1) >= 0 && e >= 0 && ctl == c) {
+ WorkQueue w = ws[e & m];
+ if (e != 0 && w != null) {
+ Thread p;
+ long nc = ((long)(w.nextWait & E_MASK) |
+ (c & (AC_MASK|TC_MASK)));
+ int ne = (e + E_SEQ) & E_MASK;
+ if (w.eventCount == (e | INT_SIGN) &&
+ U.compareAndSwapLong(this, CTL, c, nc)) {
+ w.eventCount = ne;
+ if ((p = w.parker) != null)
+ U.unpark(p);
+ return true; // replace with idle worker
+ }
+ }
+ else if ((tc = (short)(c >>> TC_SHIFT)) >= 0 &&
+ (int)(c >> AC_SHIFT) + pc > 1) {
+ long nc = ((c - AC_UNIT) & AC_MASK) | (c & ~AC_MASK);
+ if (U.compareAndSwapLong(this, CTL, c, nc))
+ return true; // no compensation
}
- if ((task == null || task.status >= 0) && // recheck need to block
- (blocker == null || !blocker.isReleasable()) && ctl == c) {
- if (!replace) { // no compensation
- long nc = ((c - AC_UNIT) & AC_MASK) | (c & ~AC_MASK);
- if (U.compareAndSwapLong(this, CTL, c, nc))
- return true;
- }
- else if (e != 0) { // release an idle worker
- WorkQueue w; Thread p; int i;
- if ((i = e & SMASK) < ws.length && (w = ws[i]) != null) {
- long nc = ((long)(w.nextWait & E_MASK) |
- (c & (AC_MASK|TC_MASK)));
- if (w.eventCount == (e | INT_SIGN) &&
- U.compareAndSwapLong(this, CTL, c, nc)) {
- w.eventCount = (e + E_SEQ) & E_MASK;
- if ((p = w.parker) != null)
- U.unpark(p);
+ else if (tc + pc < MAX_CAP) {
+ long nc = ((c + TC_UNIT) & TC_MASK) | (c & ~TC_MASK);
+ if (U.compareAndSwapLong(this, CTL, c, nc)) {
+ ForkJoinWorkerThreadFactory fac;
+ Throwable ex = null;
+ ForkJoinWorkerThread wt = null;
+ try {
+ if ((fac = factory) != null &&
+ (wt = fac.newThread(this)) != null) {
+ wt.start();
return true;
}
+ } catch (Throwable rex) {
+ ex = rex;
}
- }
- else if (tc < MAX_CAP) { // create replacement
- long nc = ((c + TC_UNIT) & TC_MASK) | (c & ~TC_MASK);
- if (U.compareAndSwapLong(this, CTL, c, nc)) {
- addWorker();
- return true;
- }
+ deregisterWorker(wt, ex); // clean up and return false
}
}
}
@@ -1822,25 +1985,22 @@ public class ForkJoinPool extends Abstra
* @return task status on exit
*/
final int awaitJoin(WorkQueue joiner, ForkJoinTask task) {
- int s;
- if ((s = task.status) >= 0) {
+ int s = 0;
+ if (task != null && (s = task.status) >= 0 && joiner != null) {
ForkJoinTask prevJoin = joiner.currentJoin;
joiner.currentJoin = task;
- long startTime = 0L;
- for (int k = 0;;) {
- if ((s = (joiner.isEmpty() ? // try to help
- tryHelpStealer(joiner, task) :
- joiner.tryRemoveAndExec(task))) == 0 &&
+ do {} while (joiner.tryRemoveAndExec(task) && // process local tasks
+ (s = task.status) >= 0);
+ if (s >= 0 && (task instanceof CountedCompleter))
+ s = helpComplete(joiner, (CountedCompleter)task);
+ long cc = 0; // for stability checks
+ while (s >= 0 && (s = task.status) >= 0) {
+ if ((s = tryHelpStealer(joiner, task)) == 0 &&
(s = task.status) >= 0) {
- if (k == 0) {
- startTime = System.nanoTime();
- tryPollForAndExec(joiner, task); // check uncommon case
- }
- else if ((k & (MAX_HELP - 1)) == 0 &&
- System.nanoTime() - startTime >=
- COMPENSATION_DELAY &&
- tryCompensate(task, null)) {
- if (task.trySetSignal()) {
+ if (!tryCompensate(cc))
+ cc = ctl;
+ else {
+ if (task.trySetSignal() && (s = task.status) >= 0) {
synchronized (task) {
if (task.status >= 0) {
try { // see ForkJoinTask
@@ -1852,18 +2012,15 @@ public class ForkJoinPool extends Abstra
task.notifyAll();
}
}
- long c; // re-activate
+ long c; // reactivate
do {} while (!U.compareAndSwapLong
- (this, CTL, c = ctl, c + AC_UNIT));
+ (this, CTL, c = ctl,
+ ((c & ~AC_MASK) |
+ ((c & AC_MASK) + AC_UNIT))));
}
}
- if (s < 0 || (s = task.status) < 0) {
- joiner.currentJoin = prevJoin;
- break;
- }
- else if ((k++ & (MAX_HELP - 1)) == MAX_HELP >>> 1)
- Thread.yield(); // for politeness
}
+ joiner.currentJoin = prevJoin;
}
return s;
}
@@ -1875,46 +2032,45 @@ public class ForkJoinPool extends Abstra
*
* @param joiner the joining worker
* @param task the task
- * @return task status on exit
*/
- final int helpJoinOnce(WorkQueue joiner, ForkJoinTask task) {
+ final void helpJoinOnce(WorkQueue joiner, ForkJoinTask task) {
int s;
- while ((s = task.status) >= 0 &&
- (joiner.isEmpty() ?
- tryHelpStealer(joiner, task) :
- joiner.tryRemoveAndExec(task)) != 0)
- ;
- return s;
+ if (joiner != null && task != null && (s = task.status) >= 0) {
+ ForkJoinTask prevJoin = joiner.currentJoin;
+ joiner.currentJoin = task;
+ do {} while (joiner.tryRemoveAndExec(task) && // process local tasks
+ (s = task.status) >= 0);
+ if (s >= 0) {
+ if (task instanceof CountedCompleter)
+ helpComplete(joiner, (CountedCompleter)task);
+ do {} while (task.status >= 0 &&
+ tryHelpStealer(joiner, task) > 0);
+ }
+ joiner.currentJoin = prevJoin;
+ }
}
/**
* Returns a (probably) non-empty steal queue, if one is found
- * during a random, then cyclic scan, else null. This method must
- * be retried by caller if, by the time it tries to use the queue,
- * it is empty.
- */
- private WorkQueue findNonEmptyStealQueue(WorkQueue w) {
- // Similar to loop in scan(), but ignoring submissions
- int r = w.seed; r ^= r << 13; r ^= r >>> 17; w.seed = r ^= r << 5;
- int step = (r >>> 16) | 1;
- for (WorkQueue[] ws;;) {
- int rs = runState, m;
- if ((ws = workQueues) == null || (m = ws.length - 1) < 1)
- return null;
- for (int j = (m + 1) << 2; ; r += step) {
- WorkQueue q = ws[((r << 1) | 1) & m];
- if (q != null && !q.isEmpty())
- return q;
- else if (--j < 0) {
- if (runState == rs)
- return null;
- break;
+ * during a scan, else null. This method must be retried by
+ * caller if, by the time it tries to use the queue, it is empty.
+ */
+ private WorkQueue findNonEmptyStealQueue() {
+ int r = ThreadLocalRandom.current().nextInt();
+ for (;;) {
+ int ps = plock, m; WorkQueue[] ws; WorkQueue q;
+ if ((ws = workQueues) != null && (m = ws.length - 1) >= 0) {
+ for (int j = (m + 1) << 2; j >= 0; --j) {
+ if ((q = ws[(((r - j) << 1) | 1) & m]) != null &&
+ q.base - q.top < 0)
+ return q;
}
}
+ if (plock == ps)
+ return null;
}
}
-
/**
* Runs tasks until {@code isQuiescent()}. We piggyback on
* active count ctl maintenance, but rather than blocking
@@ -1922,37 +2078,36 @@ public class ForkJoinPool extends Abstra
* find tasks either.
*/
final void helpQuiescePool(WorkQueue w) {
+ ForkJoinTask ps = w.currentSteal;
for (boolean active = true;;) {
- ForkJoinTask localTask; // exhaust local queue
- while ((localTask = w.nextLocalTask()) != null)
- localTask.doExec();
- WorkQueue q = findNonEmptyStealQueue(w);
- if (q != null) {
- ForkJoinTask t; int b;
+ long c; WorkQueue q; ForkJoinTask t; int b;
+ while ((t = w.nextLocalTask()) != null)
+ t.doExec();
+ if ((q = findNonEmptyStealQueue()) != null) {
if (!active) { // re-establish active count
- long c;
active = true;
do {} while (!U.compareAndSwapLong
- (this, CTL, c = ctl, c + AC_UNIT));
+ (this, CTL, c = ctl,
+ ((c & ~AC_MASK) |
+ ((c & AC_MASK) + AC_UNIT))));
+ }
+ if ((b = q.base) - q.top < 0 && (t = q.pollAt(b)) != null) {
+ (w.currentSteal = t).doExec();
+ w.currentSteal = ps;
}
- if ((b = q.base) - q.top < 0 && (t = q.pollAt(b)) != null)
- w.runSubtask(t);
}
- else {
- long c;
- if (active) { // decrement active count without queuing
+ else if (active) { // decrement active count without queuing
+ long nc = ((c = ctl) & ~AC_MASK) | ((c & AC_MASK) - AC_UNIT);
+ if ((int)(nc >> AC_SHIFT) + parallelism == 0)
+ break; // bypass decrement-then-increment
+ if (U.compareAndSwapLong(this, CTL, c, nc))
active = false;
- do {} while (!U.compareAndSwapLong
- (this, CTL, c = ctl, c -= AC_UNIT));
- }
- else
- c = ctl; // re-increment on exit
- if ((int)(c >> AC_SHIFT) + parallelism == 0) {
- do {} while (!U.compareAndSwapLong
- (this, CTL, c = ctl, c + AC_UNIT));
- break;
- }
}
+ else if ((int)((c = ctl) >> AC_SHIFT) + parallelism <= 0 &&
+ U.compareAndSwapLong
+ (this, CTL, c, ((c & ~AC_MASK) |
+ ((c & AC_MASK) + AC_UNIT))))
+ break;
}
}
@@ -1966,7 +2121,7 @@ public class ForkJoinPool extends Abstra
WorkQueue q; int b;
if ((t = w.nextLocalTask()) != null)
return t;
- if ((q = findNonEmptyStealQueue(w)) == null)
+ if ((q = findNonEmptyStealQueue()) == null)
return null;
if ((b = q.base) - q.top < 0 && (t = q.pollAt(b)) != null)
return t;
@@ -1974,19 +2129,64 @@ public class ForkJoinPool extends Abstra
}
/**
- * Returns the approximate (non-atomic) number of idle threads per
- * active thread to offset steal queue size for method
- * ForkJoinTask.getSurplusQueuedTaskCount().
- */
- final int idlePerActive() {
- // Approximate at powers of two for small values, saturate past 4
- int p = parallelism;
- int a = p + (int)(ctl >> AC_SHIFT);
- return (a > (p >>>= 1) ? 0 :
- a > (p >>>= 1) ? 1 :
- a > (p >>>= 1) ? 2 :
- a > (p >>>= 1) ? 4 :
- 8);
+ * Returns a cheap heuristic guide for task partitioning when
+ * programmers, frameworks, tools, or languages have little or no
+ * idea about task granularity. In essence by offering this
+ * method, we ask users only about tradeoffs in overhead vs
+ * expected throughput and its variance, rather than how finely to
+ * partition tasks.
+ *
+ * In a steady state strict (tree-structured) computation, each
+ * thread makes available for stealing enough tasks for other
+ * threads to remain active. Inductively, if all threads play by
+ * the same rules, each thread should make available only a
+ * constant number of tasks.
+ *
+ * The minimum useful constant is just 1. But using a value of 1
+ * would require immediate replenishment upon each steal to
+ * maintain enough tasks, which is infeasible. Further,
+ * partitionings/granularities of offered tasks should minimize
+ * steal rates, which in general means that threads nearer the top
+ * of computation tree should generate more than those nearer the
+ * bottom. In perfect steady state, each thread is at
+ * approximately the same level of computation tree. However,
+ * producing extra tasks amortizes the uncertainty of progress and
+ * diffusion assumptions.
+ *
+ * So, users will want to use values larger (but not much larger)
+ * than 1 to both smooth over transient shortages and hedge
+ * against uneven progress; as traded off against the cost of
+ * extra task overhead. We leave the user to pick a threshold
+ * value to compare with the results of this call to guide
+ * decisions, but recommend values such as 3.
+ *
+ * When all threads are active, it is on average OK to estimate
+ * surplus strictly locally. In steady-state, if one thread is
+ * maintaining say 2 surplus tasks, then so are others. So we can
+ * just use estimated queue length. However, this strategy alone
+ * leads to serious mis-estimates in some non-steady-state
+ * conditions (ramp-up, ramp-down, other stalls). We can detect
+ * many of these by further considering the number of "idle"
+ * threads, that are known to have zero queued tasks, so
+ * compensate by a factor of (#idle/#active) threads.
+ *
+ * Note: The approximation of #busy workers as #active workers is
+ * not very good under current signalling scheme, and should be
+ * improved.
+ */
+ static int getSurplusQueuedTaskCount() {
+ Thread t; ForkJoinWorkerThread wt; ForkJoinPool pool; WorkQueue q;
+ if (((t = Thread.currentThread()) instanceof ForkJoinWorkerThread)) {
+ int p = (pool = (wt = (ForkJoinWorkerThread)t).pool).parallelism;
+ int n = (q = wt.workQueue).top - q.base;
+ int a = (int)(pool.ctl >> AC_SHIFT) + p;
+ return n - (a > (p >>>= 1) ? 0 :
+ a > (p >>>= 1) ? 1 :
+ a > (p >>>= 1) ? 2 :
+ a > (p >>>= 1) ? 4 :
+ 8);
+ }
+ return 0;
}
// Termination
@@ -2006,56 +2206,69 @@ public class ForkJoinPool extends Abstra
* @return true if now terminating or terminated
*/
private boolean tryTerminate(boolean now, boolean enable) {
- Mutex lock = this.lock;
+ int ps;
+ if (this == common) // cannot shut down
+ return false;
+ if ((ps = plock) >= 0) { // enable by setting plock
+ if (!enable)
+ return false;
+ if ((ps & PL_LOCK) != 0 ||
+ !U.compareAndSwapInt(this, PLOCK, ps, ps += PL_LOCK))
+ ps = acquirePlock();
+ int nps = ((ps + PL_LOCK) & ~SHUTDOWN) | SHUTDOWN;
+ if (!U.compareAndSwapInt(this, PLOCK, ps, nps))
+ releasePlock(nps);
+ }
for (long c;;) {
- if (((c = ctl) & STOP_BIT) != 0) { // already terminating
- if ((short)(c >>> TC_SHIFT) == -parallelism) {
- lock.lock(); // don't need try/finally
- termination.signalAll(); // signal when 0 workers
- lock.unlock();
+ if (((c = ctl) & STOP_BIT) != 0) { // already terminating
+ if ((short)(c >>> TC_SHIFT) + parallelism <= 0) {
+ synchronized (this) {
+ notifyAll(); // signal when 0 workers
+ }
}
return true;
}
- if (runState >= 0) { // not yet enabled
- if (!enable)
- return false;
- lock.lock();
- runState |= SHUTDOWN;
- lock.unlock();
- }
- if (!now) { // check if idle & no tasks
- if ((int)(c >> AC_SHIFT) != -parallelism ||
- hasQueuedSubmissions())
+ if (!now) { // check if idle & no tasks
+ WorkQueue[] ws; WorkQueue w;
+ if ((int)(c >> AC_SHIFT) + parallelism > 0)
return false;
- // Check for unqueued inactive workers. One pass suffices.
- WorkQueue[] ws = workQueues; WorkQueue w;
- if (ws != null) {
- for (int i = 1; i < ws.length; i += 2) {
- if ((w = ws[i]) != null && w.eventCount >= 0)
+ if ((ws = workQueues) != null) {
+ for (int i = 0; i < ws.length; ++i) {
+ if ((w = ws[i]) != null &&
+ (!w.isEmpty() ||
+ ((i & 1) != 0 && w.eventCount >= 0))) {
+ signalWork(ws, w);
return false;
+ }
}
}
}
if (U.compareAndSwapLong(this, CTL, c, c | STOP_BIT)) {
for (int pass = 0; pass < 3; ++pass) {
- WorkQueue[] ws = workQueues;
- if (ws != null) {
- WorkQueue w;
+ WorkQueue[] ws; WorkQueue w; Thread wt;
+ if ((ws = workQueues) != null) {
int n = ws.length;
for (int i = 0; i < n; ++i) {
if ((w = ws[i]) != null) {
- w.runState = -1;
+ w.qlock = -1;
if (pass > 0) {
w.cancelAll();
- if (pass > 1)
- w.interruptOwner();
+ if (pass > 1 && (wt = w.owner) != null) {
+ if (!wt.isInterrupted()) {
+ try {
+ wt.interrupt();
+ } catch (Throwable ignore) {
+ }
+ }
+ U.unpark(wt);
+ }
}
}
}
// Wake up workers parked on event queue
int i, e; long cc; Thread p;
while ((e = (int)(cc = ctl) & E_MASK) != 0 &&
- (i = e & SMASK) < n &&
+ (i = e & SMASK) < n && i >= 0 &&
(w = ws[i]) != null) {
long nc = ((long)(w.nextWait & E_MASK) |
((cc + AC_UNIT) & AC_MASK) |
@@ -2063,7 +2276,7 @@ public class ForkJoinPool extends Abstra
if (w.eventCount == (e | INT_SIGN) &&
U.compareAndSwapLong(this, CTL, cc, nc)) {
w.eventCount = (e + E_SEQ) & E_MASK;
- w.runState = -1;
+ w.qlock = -1;
if ((p = w.parker) != null)
U.unpark(p);
}
@@ -2074,6 +2287,77 @@ public class ForkJoinPool extends Abstra
}
}
+ // external operations on common pool
+
+ /**
+ * Returns common pool queue for a thread that has submitted at
+ * least one task.
+ */
+ static WorkQueue commonSubmitterQueue() {
+ Submitter z; ForkJoinPool p; WorkQueue[] ws; int m, r;
+ return ((z = submitters.get()) != null &&
+ (p = common) != null &&
+ (ws = p.workQueues) != null &&
+ (m = ws.length - 1) >= 0) ?
+ ws[m & z.seed & SQMASK] : null;
+ }
+
+ /**
+ * Tries to pop the given task from submitter's queue in common pool.
+ */
+ final boolean tryExternalUnpush(ForkJoinTask task) {
+ WorkQueue joiner; ForkJoinTask[] a; int m, s;
+ Submitter z = submitters.get();
+ WorkQueue[] ws = workQueues;
+ boolean popped = false;
+ if (z != null && ws != null && (m = ws.length - 1) >= 0 &&
+ (joiner = ws[z.seed & m & SQMASK]) != null &&
+ joiner.base != (s = joiner.top) &&
+ (a = joiner.array) != null) {
+ long j = (((a.length - 1) & (s - 1)) << ASHIFT) + ABASE;
+ if (U.getObject(a, j) == task &&
+ U.compareAndSwapInt(joiner, QLOCK, 0, 1)) {
+ if (joiner.top == s && joiner.array == a &&
+ U.compareAndSwapObject(a, j, task, null)) {
+ joiner.top = s - 1;
+ popped = true;
+ }
+ joiner.qlock = 0;
+ }
+ }
+ return popped;
+ }
+
+ final int externalHelpComplete(CountedCompleter task) {
+ WorkQueue joiner; int m, j;
+ Submitter z = submitters.get();
+ WorkQueue[] ws = workQueues;
+ int s = 0;
+ if (z != null && ws != null && (m = ws.length - 1) >= 0 &&
+ (joiner = ws[(j = z.seed) & m & SQMASK]) != null && task != null) {
+ int scans = m + m + 1;
+ long c = 0L; // for stability check
+ j |= 1; // poll odd queues
+ for (int k = scans; ; j += 2) {
+ WorkQueue q;
+ if ((s = task.status) < 0)
+ break;
+ else if (joiner.externalPopAndExecCC(task))
+ k = scans;
+ else if ((s = task.status) < 0)
+ break;
+ else if ((q = ws[j & m]) != null && q.pollAndExecCC(task))
+ k = scans;
+ else if (--k < 0) {
+ if (c == (c = ctl))
+ break;
+ k = scans;
+ }
+ }
+ }
+ return s;
+ }
+
// Exported methods
// Constructors
@@ -2090,7 +2374,7 @@ public class ForkJoinPool extends Abstra
* java.lang.RuntimePermission}{@code ("modifyThread")}
*/
public ForkJoinPool() {
- this(Runtime.getRuntime().availableProcessors(),
+ this(Math.min(MAX_CAP, Runtime.getRuntime().availableProcessors()),
defaultForkJoinWorkerThreadFactory, null, false);
}
@@ -2138,36 +2422,64 @@ public class ForkJoinPool extends Abstra
*/
public ForkJoinPool(int parallelism,
ForkJoinWorkerThreadFactory factory,
- Thread.UncaughtExceptionHandler handler,
+ UncaughtExceptionHandler handler,
boolean asyncMode) {
+ this(checkParallelism(parallelism),
+ checkFactory(factory),
+ handler,
+ (asyncMode ? FIFO_QUEUE : LIFO_QUEUE),
+ "ForkJoinPool-" + nextPoolId() + "-worker-");
checkPermission();
- if (factory == null)
- throw new NullPointerException();
+ }
+
+ private static int checkParallelism(int parallelism) {
if (parallelism <= 0 || parallelism > MAX_CAP)
throw new IllegalArgumentException();
- this.parallelism = parallelism;
+ return parallelism;
+ }
+
+ private static ForkJoinWorkerThreadFactory checkFactory
+ (ForkJoinWorkerThreadFactory factory) {
+ if (factory == null)
+ throw new NullPointerException();
+ return factory;
+ }
+
+ /**
+ * Creates a {@code ForkJoinPool} with the given parameters, without
+ * any security checks or parameter validation. Invoked directly by
+ * makeCommonPool.
+ */
+ private ForkJoinPool(int parallelism,
+ ForkJoinWorkerThreadFactory factory,
+ UncaughtExceptionHandler handler,
+ int mode,
+ String workerNamePrefix) {
+ this.workerNamePrefix = workerNamePrefix;
this.factory = factory;
this.ueh = handler;
- this.localMode = asyncMode ? FIFO_QUEUE : LIFO_QUEUE;
+ this.mode = (short)mode;
+ this.parallelism = (short)parallelism;
long np = (long)(-parallelism); // offset ctl counts
this.ctl = ((np << AC_SHIFT) & AC_MASK) | ((np << TC_SHIFT) & TC_MASK);
- // Use nearest power 2 for workQueues size. See Hackers Delight sec 3.2.
- int n = parallelism - 1;
- n |= n >>> 1; n |= n >>> 2; n |= n >>> 4; n |= n >>> 8; n |= n >>> 16;
- int size = (n + 1) << 1; // #slots = 2*#workers
- this.submitMask = size - 1; // room for max # of submit queues
- this.workQueues = new WorkQueue[size];
- this.termination = (this.lock = new Mutex()).newCondition();
- this.stealCount = new AtomicLong();
- this.nextWorkerNumber = new AtomicInteger();
- int pn = poolNumberGenerator.incrementAndGet();
- StringBuilder sb = new StringBuilder("ForkJoinPool-");
- sb.append(Integer.toString(pn));
- sb.append("-worker-");
- this.workerNamePrefix = sb.toString();
- lock.lock();
- this.runState = 1; // set init flag
- lock.unlock();
+ }
+
+ /**
+ * Returns the common pool instance. This pool is statically
+ * constructed; its run state is unaffected by attempts to {@link
+ * #shutdown} or {@link #shutdownNow}. However this pool and any
+ * ongoing processing are automatically terminated upon program
+ * {@link System#exit}. Any program that relies on asynchronous
+ * task processing to complete before program termination should
+ * invoke {@code commonPool().}{@link #awaitQuiescence awaitQuiescence},
+ * before exit.
+ *
+ * @return the common pool instance
+ * @since 1.8
+ */
+ public static ForkJoinPool commonPool() {
+ // assert common != null : "static init error";
+ return common;
}
// Execution methods
@@ -2191,7 +2503,7 @@ public class ForkJoinPool extends Abstra
public T invoke(ForkJoinTask task) {
if (task == null)
throw new NullPointerException();
- doSubmit(task);
+ externalPush(task);
return task.join();
}
@@ -2206,7 +2518,7 @@ public class ForkJoinPool extends Abstra
public void execute(ForkJoinTask task) {
if (task == null)
throw new NullPointerException();
- doSubmit(task);
+ externalPush(task);
}
// AbstractExecutorService methods
@@ -2223,8 +2535,8 @@ public class ForkJoinPool extends Abstra
if (task instanceof ForkJoinTask) // avoid re-wrap
job = (ForkJoinTask) task;
else
- job = new ForkJoinTask.AdaptedRunnableAction(task);
- doSubmit(job);
+ job = new ForkJoinTask.RunnableExecuteAction(task);
+ externalPush(job);
}
/**
@@ -2239,7 +2551,7 @@ public class ForkJoinPool extends Abstra
public ForkJoinTask submit(ForkJoinTask task) {
if (task == null)
throw new NullPointerException();
- doSubmit(task);
+ externalPush(task);
return task;
}
@@ -2250,7 +2562,7 @@ public class ForkJoinPool extends Abstra
*/
public ForkJoinTask submit(Callable task) {
ForkJoinTask job = new ForkJoinTask.AdaptedCallable(task);
- doSubmit(job);
+ externalPush(job);
return job;
}
@@ -2261,7 +2573,7 @@ public class ForkJoinPool extends Abstra
*/
public ForkJoinTask submit(Runnable task, T result) {
ForkJoinTask job = new ForkJoinTask.AdaptedRunnable(task, result);
- doSubmit(job);
+ externalPush(job);
return job;
}
@@ -2278,7 +2590,7 @@ public class ForkJoinPool extends Abstra
job = (ForkJoinTask) task;
else
job = new ForkJoinTask.AdaptedRunnableAction(task);
- doSubmit(job);
+ externalPush(job);
return job;
}
@@ -2290,27 +2602,23 @@ public class ForkJoinPool extends Abstra
// In previous versions of this class, this method constructed
// a task to run ForkJoinTask.invokeAll, but now external
// invocation of multiple tasks is at least as efficient.
- List> fs = new ArrayList>(tasks.size());
- // Workaround needed because method wasn't declared with
- // wildcards in return type but should have been.
- @SuppressWarnings({"unchecked", "rawtypes"})
- List> futures = (List>) (List) fs;
+ ArrayList> futures = new ArrayList>(tasks.size());
boolean done = false;
try {
for (Callable t : tasks) {
ForkJoinTask f = new ForkJoinTask.AdaptedCallable(t);
- doSubmit(f);
- fs.add(f);
+ futures.add(f);
+ externalPush(f);
}
- for (ForkJoinTask f : fs)
- f.quietlyJoin();
+ for (int i = 0, size = futures.size(); i < size; i++)
+ ((ForkJoinTask)futures.get(i)).quietlyJoin();
done = true;
return futures;
} finally {
if (!done)
- for (ForkJoinTask f : fs)
- f.cancel(false);
+ for (int i = 0, size = futures.size(); i < size; i++)
+ futures.get(i).cancel(false);
}
}
@@ -2329,7 +2637,7 @@ public class ForkJoinPool extends Abstra
*
* @return the handler, or {@code null} if none
*/
- public Thread.UncaughtExceptionHandler getUncaughtExceptionHandler() {
+ public UncaughtExceptionHandler getUncaughtExceptionHandler() {
return ueh;
}
@@ -2339,7 +2647,18 @@ public class ForkJoinPool extends Abstra
* @return the targeted parallelism level of this pool
*/
public int getParallelism() {
- return parallelism;
+ int par;
+ return ((par = parallelism) > 0) ? par : 1;
+ }
+
+ /**
+ * Returns the targeted parallelism level of the common pool.
+ *
+ * @return the targeted parallelism level of the common pool
+ * @since 1.8
+ */
+ public static int getCommonPoolParallelism() {
+ return commonParallelism;
}
/**
@@ -2361,7 +2680,7 @@ public class ForkJoinPool extends Abstra
* @return {@code true} if this pool uses async mode
*/
public boolean getAsyncMode() {
- return localMode != 0;
+ return mode == FIFO_QUEUE;
}
/**
@@ -2408,7 +2727,7 @@ public class ForkJoinPool extends Abstra
* @return {@code true} if all threads are currently idle
*/
public boolean isQuiescent() {
- return (int)(ctl >> AC_SHIFT) + parallelism == 0;
+ return parallelism + (int)(ctl >> AC_SHIFT) <= 0;
}
/**
@@ -2423,12 +2742,12 @@ public class ForkJoinPool extends Abstra
* @return the number of steals
*/
public long getStealCount() {
- long count = stealCount.get();
+ long count = stealCount;
WorkQueue[] ws; WorkQueue w;
if ((ws = workQueues) != null) {
for (int i = 1; i < ws.length; i += 2) {
if ((w = ws[i]) != null)
- count += w.totalSteals;
+ count += w.nsteals;
}
}
return count;
@@ -2553,7 +2872,7 @@ public class ForkJoinPool extends Abstra
public String toString() {
// Use a single pass through workQueues to collect counts
long qt = 0L, qs = 0L; int rc = 0;
- long st = stealCount.get();
+ long st = stealCount;
long c = ctl;
WorkQueue[] ws; WorkQueue w;
if ((ws = workQueues) != null) {
@@ -2564,7 +2883,7 @@ public class ForkJoinPool extends Abstra
qs += size;
else {
qt += size;
- st += w.totalSteals;
+ st += w.nsteals;
if (w.isApparentlyUnblocked())
++rc;
}
@@ -2580,7 +2899,7 @@ public class ForkJoinPool extends Abstra
if ((c & STOP_BIT) != 0)
level = (tc == 0) ? "Terminated" : "Terminating";
else
- level = runState < 0 ? "Shutting down" : "Running";
+ level = plock < 0 ? "Shutting down" : "Running";
return super.toString() +
"[" + level +
", parallelism = " + pc +
@@ -2594,11 +2913,13 @@ public class ForkJoinPool extends Abstra
}
/**
- * Initiates an orderly shutdown in which previously submitted
- * tasks are executed, but no new tasks will be accepted.
- * Invocation has no additional effect if already shut down.
- * Tasks that are in the process of being submitted concurrently
- * during the course of this method may or may not be rejected.
+ * Possibly initiates an orderly shutdown in which previously
+ * submitted tasks are executed, but no new tasks will be
+ * accepted. Invocation has no effect on execution state if this
+ * is the {@link #commonPool()}, and no additional effect if
+ * already shut down. Tasks that are in the process of being
+ * submitted concurrently during the course of this method may or
+ * may not be rejected.
*
* @throws SecurityException if a security manager exists and
* the caller is not permitted to modify threads
@@ -2611,14 +2932,16 @@ public class ForkJoinPool extends Abstra
}
/**
- * Attempts to cancel and/or stop all tasks, and reject all
- * subsequently submitted tasks. Tasks that are in the process of
- * being submitted or executed concurrently during the course of
- * this method may or may not be rejected. This method cancels
- * both existing and unexecuted tasks, in order to permit
- * termination in the presence of task dependencies. So the method
- * always returns an empty list (unlike the case for some other
- * Executors).
+ * Possibly attempts to cancel and/or stop all tasks, and reject
+ * all subsequently submitted tasks. Invocation has no effect on
+ * execution state if this is the {@link #commonPool()}, and no
+ * additional effect if already shut down. Otherwise, tasks that
+ * are in the process of being submitted or executed concurrently
+ * during the course of this method may or may not be
+ * rejected. This method cancels both existing and unexecuted
+ * tasks, in order to permit termination in the presence of task
+ * dependencies. So the method always returns an empty list
+ * (unlike the case for some other Executors).
*
* @return an empty list
* @throws SecurityException if a security manager exists and
@@ -2640,7 +2963,7 @@ public class ForkJoinPool extends Abstra
public boolean isTerminated() {
long c = ctl;
return ((c & STOP_BIT) != 0L &&
- (short)(c >>> TC_SHIFT) == -parallelism);
+ (short)(c >>> TC_SHIFT) + parallelism <= 0);
}
/**
@@ -2648,7 +2971,7 @@ public class ForkJoinPool extends Abstra
* commenced but not yet completed. This method may be useful for
* debugging. A return of {@code true} reported a sufficient
* period after shutdown may indicate that submitted tasks have
- * ignored or suppressed interruption, or are waiting for IO,
+ * ignored or suppressed interruption, or are waiting for I/O,
* causing this executor not to properly terminate. (See the
* advisory notes for class {@link ForkJoinTask} stating that
* tasks should not normally entail blocking operations. But if
@@ -2659,7 +2982,7 @@ public class ForkJoinPool extends Abstra
public boolean isTerminating() {
long c = ctl;
return ((c & STOP_BIT) != 0L &&
- (short)(c >>> TC_SHIFT) != -parallelism);
+ (short)(c >>> TC_SHIFT) + parallelism > 0);
}
/**
@@ -2668,13 +2991,16 @@ public class ForkJoinPool extends Abstra
* @return {@code true} if this pool has been shut down
*/
public boolean isShutdown() {
- return runState < 0;
+ return plock < 0;
}
/**
- * Blocks until all tasks have completed execution after a shutdown
- * request, or the timeout occurs, or the current thread is
- * interrupted, whichever happens first.
+ * Blocks until all tasks have completed execution after a
+ * shutdown request, or the timeout occurs, or the current thread
+ * is interrupted, whichever happens first. Because the {@link
+ * #commonPool()} never terminates until program shutdown, when
+ * applied to the common pool, this method is equivalent to {@link
+ * #awaitQuiescence(long, TimeUnit)} but always returns {@code false}.
*
* @param timeout the maximum time to wait
* @param unit the time unit of the timeout argument
@@ -2684,23 +3010,85 @@ public class ForkJoinPool extends Abstra
*/
public boolean awaitTermination(long timeout, TimeUnit unit)
throws InterruptedException {
+ if (Thread.interrupted())
+ throw new InterruptedException();
+ if (this == common) {
+ awaitQuiescence(timeout, unit);
+ return false;
+ }
long nanos = unit.toNanos(timeout);
- final Mutex lock = this.lock;
- lock.lock();
- try {
+ if (isTerminated())
+ return true;
+ if (nanos <= 0L)
+ return false;
+ long deadline = System.nanoTime() + nanos;
+ synchronized (this) {
for (;;) {
if (isTerminated())
return true;
- if (nanos <= 0)
+ if (nanos <= 0L)
return false;
- nanos = termination.awaitNanos(nanos);
+ long millis = TimeUnit.NANOSECONDS.toMillis(nanos);
+ wait(millis > 0L ? millis : 1L);
+ nanos = deadline - System.nanoTime();
}
- } finally {
- lock.unlock();
}
}
/**
+ * If called by a ForkJoinTask operating in this pool, equivalent
+ * in effect to {@link ForkJoinTask#helpQuiesce}. Otherwise,
+ * waits and/or attempts to assist performing tasks until this
+ * pool {@link #isQuiescent} or the indicated timeout elapses.
+ *
+ * @param timeout the maximum time to wait
+ * @param unit the time unit of the timeout argument
+ * @return {@code true} if quiescent; {@code false} if the
+ * timeout elapsed.
+ */
+ public boolean awaitQuiescence(long timeout, TimeUnit unit) {
+ long nanos = unit.toNanos(timeout);
+ ForkJoinWorkerThread wt;
+ Thread thread = Thread.currentThread();
+ if ((thread instanceof ForkJoinWorkerThread) &&
+ (wt = (ForkJoinWorkerThread)thread).pool == this) {
+ helpQuiescePool(wt.workQueue);
+ return true;
+ }
+ long startTime = System.nanoTime();
+ WorkQueue[] ws;
+ int r = 0, m;
+ boolean found = true;
+ while (!isQuiescent() && (ws = workQueues) != null &&
+ (m = ws.length - 1) >= 0) {
+ if (!found) {
+ if ((System.nanoTime() - startTime) > nanos)
+ return false;
+ Thread.yield(); // cannot block
+ }
+ found = false;
+ for (int j = (m + 1) << 2; j >= 0; --j) {
+ ForkJoinTask t; WorkQueue q; int b;
+ if ((q = ws[r++ & m]) != null && (b = q.base) - q.top < 0) {
+ found = true;
+ if ((t = q.pollAt(b)) != null)
+ t.doExec();
+ break;
+ }
+ }
+ }
+ return true;
+ }
+
+ /**
+ * Waits and/or attempts to assist performing tasks indefinitely
+ * until the {@link #commonPool()} {@link #isQuiescent}.
+ */
+ static void quiesceCommonPool() {
+ common.awaitQuiescence(Long.MAX_VALUE, TimeUnit.NANOSECONDS);
+ }
+
+ /**
* Interface for extending managed parallelism for tasks running
* in {@link ForkJoinPool}s.
*
@@ -2709,9 +3097,9 @@ public class ForkJoinPool extends Abstra
* not necessary. Method {@code block} blocks the current thread
* if necessary (perhaps internally invoking {@code isReleasable}
* before actually blocking). These actions are performed by any
- * thread invoking {@link ForkJoinPool#managedBlock}. The
- * unusual methods in this API accommodate synchronizers that may,
- * but don't usually, block for long periods. Similarly, they
+ * thread invoking {@link ForkJoinPool#managedBlock(ManagedBlocker)}.
+ * 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,
@@ -2769,6 +3157,7 @@ public class ForkJoinPool extends Abstra
/**
* Returns {@code true} if blocking is unnecessary.
+ * @return {@code true} if blocking is unnecessary
*/
boolean isReleasable();
}
@@ -2796,19 +3185,24 @@ public class ForkJoinPool extends Abstra
public static void managedBlock(ManagedBlocker blocker)
throws InterruptedException {
Thread t = Thread.currentThread();
- ForkJoinPool p = ((t instanceof ForkJoinWorkerThread) ?
- ((ForkJoinWorkerThread)t).pool : null);
- while (!blocker.isReleasable()) {
- if (p == null || p.tryCompensate(null, blocker)) {
- try {
- do {} while (!blocker.isReleasable() && !blocker.block());
- } finally {
- if (p != null)
+ if (t instanceof ForkJoinWorkerThread) {
+ ForkJoinPool p = ((ForkJoinWorkerThread)t).pool;
+ while (!blocker.isReleasable()) {
+ if (p.tryCompensate(p.ctl)) {
+ try {
+ do {} while (!blocker.isReleasable() &&
+ !blocker.block());
+ } finally {
p.incrementActiveCount();
+ }
+ break;
}
- break;
}
}
+ else {
+ do {} while (!blocker.isReleasable() &&
+ !blocker.block());
+ }
}
// AbstractExecutorService overrides. These rely on undocumented
@@ -2829,32 +3223,89 @@ public class ForkJoinPool extends Abstra
private static final long PARKBLOCKER;
private static final int ABASE;
private static final int ASHIFT;
+ private static final long STEALCOUNT;
+ private static final long PLOCK;
+ private static final long INDEXSEED;
+ private static final long QBASE;
+ private static final long QLOCK;
static {
- poolNumberGenerator = new AtomicInteger();
- nextSubmitterSeed = new AtomicInteger(0x55555555);
- modifyThreadPermission = new RuntimePermission("modifyThread");
- defaultForkJoinWorkerThreadFactory =
- new DefaultForkJoinWorkerThreadFactory();
- submitters = new ThreadSubmitter();
- int s;
+ // initialize field offsets for CAS etc
try {
U = getUnsafe();
Class k = ForkJoinPool.class;
- Class ak = ForkJoinTask[].class;
CTL = U.objectFieldOffset
(k.getDeclaredField("ctl"));
+ STEALCOUNT = U.objectFieldOffset
+ (k.getDeclaredField("stealCount"));
+ PLOCK = U.objectFieldOffset
+ (k.getDeclaredField("plock"));
+ INDEXSEED = U.objectFieldOffset
+ (k.getDeclaredField("indexSeed"));
Class tk = Thread.class;
PARKBLOCKER = U.objectFieldOffset
(tk.getDeclaredField("parkBlocker"));
+ Class wk = WorkQueue.class;
+ QBASE = U.objectFieldOffset
+ (wk.getDeclaredField("base"));
+ QLOCK = U.objectFieldOffset
+ (wk.getDeclaredField("qlock"));
+ Class ak = ForkJoinTask[].class;
ABASE = U.arrayBaseOffset(ak);
- s = U.arrayIndexScale(ak);
+ int scale = U.arrayIndexScale(ak);
+ if ((scale & (scale - 1)) != 0)
+ throw new Error("data type scale not a power of two");
+ ASHIFT = 31 - Integer.numberOfLeadingZeros(scale);
} catch (Exception e) {
throw new Error(e);
}
- if ((s & (s-1)) != 0)
- throw new Error("data type scale not a power of two");
- ASHIFT = 31 - Integer.numberOfLeadingZeros(s);
+
+ submitters = new ThreadLocal();
+ defaultForkJoinWorkerThreadFactory =
+ new DefaultForkJoinWorkerThreadFactory();
+ modifyThreadPermission = new RuntimePermission("modifyThread");
+
+ common = java.security.AccessController.doPrivileged
+ (new java.security.PrivilegedAction() {
+ public ForkJoinPool run() { return makeCommonPool(); }});
+ int par = common.parallelism; // report 1 even if threads disabled
+ commonParallelism = par > 0 ? par : 1;
+ }
+
+ /**
+ * Creates and returns the common pool, respecting user settings
+ * specified via system properties.
+ */
+ private static ForkJoinPool makeCommonPool() {
+ int parallelism = -1;
+ ForkJoinWorkerThreadFactory factory
+ = defaultForkJoinWorkerThreadFactory;
+ UncaughtExceptionHandler handler = null;
+ try { // ignore exceptions in accessing/parsing properties
+ String pp = System.getProperty
+ ("java.util.concurrent.ForkJoinPool.common.parallelism");
+ String fp = System.getProperty
+ ("java.util.concurrent.ForkJoinPool.common.threadFactory");
+ String hp = System.getProperty
+ ("java.util.concurrent.ForkJoinPool.common.exceptionHandler");
+ if (pp != null)
+ parallelism = Integer.parseInt(pp);
+ if (fp != null)
+ factory = ((ForkJoinWorkerThreadFactory)ClassLoader.
+ getSystemClassLoader().loadClass(fp).newInstance());
+ if (hp != null)
+ handler = ((UncaughtExceptionHandler)ClassLoader.
+ getSystemClassLoader().loadClass(hp).newInstance());
+ } catch (Exception ignore) {
+ }
+
+ if (parallelism < 0 && // default 1 less than #cores
+ (parallelism = Runtime.getRuntime().availableProcessors() - 1) < 0)
+ parallelism = 0;
+ if (parallelism > MAX_CAP)
+ parallelism = MAX_CAP;
+ return new ForkJoinPool(parallelism, factory, handler, LIFO_QUEUE,
+ "ForkJoinPool.commonPool-worker-");
}
/**
@@ -2867,22 +3318,23 @@ public class ForkJoinPool extends Abstra
private static sun.misc.Unsafe getUnsafe() {
try {
return sun.misc.Unsafe.getUnsafe();
- } catch (SecurityException se) {
- try {
- return java.security.AccessController.doPrivileged
- (new java.security
- .PrivilegedExceptionAction() {
- public sun.misc.Unsafe run() throws Exception {
- java.lang.reflect.Field f = sun.misc
- .Unsafe.class.getDeclaredField("theUnsafe");
- f.setAccessible(true);
- return (sun.misc.Unsafe) f.get(null);
- }});
- } catch (java.security.PrivilegedActionException e) {
- throw new RuntimeException("Could not initialize intrinsics",
- e.getCause());
- }
+ } catch (SecurityException tryReflectionInstead) {}
+ try {
+ return java.security.AccessController.doPrivileged
+ (new java.security.PrivilegedExceptionAction() {
+ public sun.misc.Unsafe run() throws Exception {
+ Class k = sun.misc.Unsafe.class;
+ for (java.lang.reflect.Field f : k.getDeclaredFields()) {
+ f.setAccessible(true);
+ Object x = f.get(null);
+ if (k.isInstance(x))
+ return k.cast(x);
+ }
+ throw new NoSuchFieldError("the Unsafe");
+ }});
+ } catch (java.security.PrivilegedActionException e) {
+ throw new RuntimeException("Could not initialize intrinsics",
+ e.getCause());
}
}
-
}