--- jsr166/src/jsr166y/ForkJoinPool.java 2009/07/31 20:17:52 1.34 +++ jsr166/src/jsr166y/ForkJoinPool.java 2010/07/23 13:07:43 1.58 @@ -13,44 +13,36 @@ import java.util.Arrays; import java.util.Collection; import java.util.Collections; import java.util.List; -import java.util.concurrent.locks.Condition; import java.util.concurrent.locks.LockSupport; import java.util.concurrent.locks.ReentrantLock; import java.util.concurrent.atomic.AtomicInteger; -import java.util.concurrent.atomic.AtomicLong; +import java.util.concurrent.CountDownLatch; /** * An {@link ExecutorService} for running {@link ForkJoinTask}s. - * A ForkJoinPool provides the entry point for submissions from - * non-ForkJoinTasks, as well as management and monitoring operations. - * Normally a single ForkJoinPool is used for a large number of - * submitted tasks. Otherwise, use would not usually outweigh the - * construction and bookkeeping overhead of creating a large set of - * threads. + * A {@code ForkJoinPool} provides the entry point for submissions + * from non-{@code ForkJoinTask} clients, as well as management and + * monitoring operations. * - *

ForkJoinPools differ from other kinds of Executors mainly in - * that they provide work-stealing: all threads in the pool - * attempt to find and execute subtasks created by other active tasks - * (eventually blocking if none exist). This makes them efficient when - * most tasks spawn other subtasks (as do most ForkJoinTasks), as well - * as the mixed execution of some plain Runnable- or Callable- based - * activities along with ForkJoinTasks. When setting {@linkplain - * #setAsyncMode async mode}, a ForkJoinPool may also be appropriate - * for use with fine-grained tasks that are never joined. Otherwise, - * other ExecutorService implementations are typically more - * appropriate choices. + *

A {@code ForkJoinPool} differs from other kinds of {@link + * ExecutorService} mainly by virtue of employing + * work-stealing: all threads in the pool attempt to find and + * execute subtasks created by other active tasks (eventually blocking + * waiting for work if none exist). This enables efficient processing + * when most tasks spawn other subtasks (as do most {@code + * ForkJoinTask}s). When setting asyncMode to true in + * constructors, {@code ForkJoinPool}s may also be appropriate for use + * with event-style tasks that are never joined. * - *

A ForkJoinPool may be constructed with a given parallelism level - * (target pool size), which it attempts to maintain by dynamically - * adding, suspending, or resuming threads, even if some tasks are - * waiting to join others. However, no such adjustments are performed - * in the face of blocked IO or other unmanaged synchronization. The - * nested {@link ManagedBlocker} interface enables extension of - * the kinds of synchronization accommodated. The target parallelism - * level may also be changed dynamically ({@link #setParallelism}) - * and thread construction can be limited using methods - * {@link #setMaximumPoolSize} and/or - * {@link #setMaintainsParallelism}. + *

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 + * synchronization accommodated. * *

In addition to execution and lifecycle control methods, this * class provides status check methods (for example @@ -59,10 +51,66 @@ import java.util.concurrent.atomic.Atomi * {@link #toString} returns indications of pool state in a * convenient form for informal monitoring. * + *

As is the case with other ExecutorServices, there are three + * main task execution methods summarized in the follwoing + * table. These are designed to be used by clients not already engaged + * in fork/join computations in the current pool. The main forms of + * these methods accept instances of {@code ForkJoinTask}, but + * overloaded forms also allow mixed execution of plain {@code + * Runnable}- or {@code Callable}- based activities as well. However, + * tasks that are already executing in a pool should normally + * NOT use these pool execution methods, but instead use the + * within-computation forms listed in the table. + * + * + * + * + * + * + * + * + * + * + * + * + * + * + * + * + * + * + * + * + * + * + *
Call from non-fork/join clients Call from within fork/join computations
Arange async execution {@link #execute(ForkJoinTask)} {@link ForkJoinTask#fork}
Await and obtain result {@link #invoke(ForkJoinTask)} {@link ForkJoinTask#invoke}
Arrange exec and obtain Future {@link #submit(ForkJoinTask)} {@link ForkJoinTask#fork} (ForkJoinTasks are Futures)
+ * + *

Sample Usage. Normally a single {@code ForkJoinPool} is + * used for all parallel task execution in a program or subsystem. + * Otherwise, use would not usually outweigh the construction and + * 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. + * + * 

+ * static final ForkJoinPool mainPool = new ForkJoinPool();
+ * ...
+ * public void sort(long[] array) {
+ *   mainPool.invoke(new SortTask(array, 0, array.length));
+ * }
+ *
+ * *

Implementation notes: This implementation restricts the * maximum number of running threads to 32767. Attempts to create - * pools with greater than the maximum result in - * IllegalArgumentExceptions. + * pools with greater than the maximum number result in + * {@code IllegalArgumentException}. + * + *

This implementation rejects submitted tasks (that is, by throwing + * {@link RejectedExecutionException}) only when the pool is shut down + * or internal resources have been exhuasted. * * @since 1.7 * @author Doug Lea @@ -70,28 +118,248 @@ import java.util.concurrent.atomic.Atomi public class ForkJoinPool extends AbstractExecutorService { /* - * See the extended comments interspersed below for design, - * rationale, and walkthroughs. - */ - - /** Mask for packing and unpacking shorts */ - private static final int shortMask = 0xffff; - - /** Max pool size -- must be a power of two minus 1 */ - private static final int MAX_THREADS = 0x7FFF; - - /** - * Factory for creating new ForkJoinWorkerThreads. A - * ForkJoinWorkerThreadFactory must be defined and used for - * ForkJoinWorkerThread subclasses that extend base functionality - * or initialize threads with different contexts. + * Implementation Overview + * + * This class provides the central bookkeeping and control for a + * set of worker threads: Submissions from non-FJ threads enter + * into a submission queue. Workers take these tasks and typically + * split them into subtasks that may be stolen by other workers. + * The main work-stealing mechanics implemented in class + * ForkJoinWorkerThread give first priority to processing tasks + * from their own queues (LIFO or FIFO, depending on mode), then + * to randomized FIFO steals of tasks in other worker queues, and + * lastly to new submissions. These mechanics do not consider + * affinities, loads, cache localities, etc, so rarely provide the + * best possible performance on a given machine, but portably + * provide good throughput by averaging over these factors. + * (Further, even if we did try to use such information, we do not + * usually have a basis for exploiting it. For example, some sets + * of tasks profit from cache affinities, but others are harmed by + * cache pollution effects.) + * + * Beyond work-stealing support and essential bookkeeping, the + * main responsibility of this framework is to arrange tactics for + * when one worker is waiting to join a task stolen (or always + * held by) another. Becauae we are multiplexing many tasks on to + * a pool of workers, we can't just let them block (as in + * Thread.join). We also cannot just reassign the joiner's + * run-time stack with another and replace it later, which would + * be a form of "continuation", that even if possible is not + * necessarily a good idea. Given that the creation costs of most + * threads on most systems mainly surrounds setting up runtime + * stacks, thread creation and switching is usually not much more + * expensive than stack creation and switching, and is more + * flexible). Instead we combine two tactics: + * + * 1. Arranging for the joiner to execute some task that it + * would be running if the steal had not occurred. Method + * ForkJoinWorkerThread.helpJoinTask tracks joining->stealing + * links to try to find such a task. + * + * 2. Unless there are already enough live threads, creating or + * or re-activating a spare thread to compensate for the + * (blocked) joiner until it unblocks. Spares then suspend + * at their next opportunity or eventually die if unused for + * too long. See below and the internal documentation + * for tryAwaitJoin for more details about compensation + * rules. + * + * Because the determining existence of conservatively safe + * helping targets, the availability of already-created spares, + * and the apparent need to create new spares are all racy and + * require heuristic guidance, joins (in + * ForkJoinWorkerThread.joinTask) interleave these options until + * successful. Creating a new spare always succeeds, but also + * increases application footprint, so we try to avoid it, within + * reason. + * + * The ManagedBlocker extension API can't use option (1) so uses a + * special version of (2) in method awaitBlocker. + * + * The main throughput advantages of work-stealing stem from + * decentralized control -- workers mostly steal tasks from each + * other. We do not want to negate this by creating bottlenecks + * implementing other management responsibilities. So we use a + * collection of techniques that avoid, reduce, or cope well with + * contention. These entail several instances of bit-packing into + * CASable fields to maintain only the minimally required + * atomicity. To enable such packing, we restrict maximum + * parallelism to (1<<15)-1 (enabling twice this (to accommodate + * unbalanced increments and decrements) to fit into a 16 bit + * field, which is far in excess of normal operating range. Even + * though updates to some of these bookkeeping fields do sometimes + * contend with each other, they don't normally cache-contend with + * updates to others enough to warrant memory padding or + * isolation. So they are all held as fields of ForkJoinPool + * objects. The main capabilities are as follows: + * + * 1. Creating and removing workers. Workers are recorded in the + * "workers" array. This is an array as opposed to some other data + * structure to support index-based random steals by workers. + * Updates to the array recording new workers and unrecording + * terminated ones are protected from each other by a lock + * (workerLock) but the array is otherwise concurrently readable, + * and accessed directly by workers. To simplify index-based + * operations, the array size is always a power of two, and all + * readers must tolerate null slots. Currently, all worker thread + * creation is on-demand, triggered by task submissions, + * replacement of terminated workers, and/or compensation for + * blocked workers. However, all other support code is set up to + * work with other policies. + * + * 2. Bookkeeping for dynamically adding and removing workers. We + * aim to approximately maintain the given level of parallelism. + * When some workers are known to be blocked (on joins or via + * ManagedBlocker), we may create or resume others to take their + * place until they unblock (see below). Implementing this + * requires counts of the number of "running" threads (i.e., those + * that are neither blocked nor artifically suspended) as well as + * the total number. These two values are packed into one field, + * "workerCounts" because we need accurate snapshots when deciding + * to create, resume or suspend. Note however that the + * correspondance of these counts to reality is not guaranteed. In + * particular updates for unblocked threads may lag until they + * actually wake up. + * + * 3. Maintaining global run state. The run state of the pool + * consists of a runLevel (SHUTDOWN, TERMINATING, etc) similar to + * those in other Executor implementations, as well as a count of + * "active" workers -- those that are, or soon will be, or + * recently were executing tasks. The runLevel and active count + * are packed together in order to correctly trigger shutdown and + * termination. Without care, active counts can be subject to very + * high contention. We substantially reduce this contention by + * relaxing update rules. A worker must claim active status + * prospectively, by activating if it sees that a submitted or + * stealable task exists (it may find after activating that the + * task no longer exists). It stays active while processing this + * task (if it exists) and any other local subtasks it produces, + * until it cannot find any other tasks. It then tries + * inactivating (see method preStep), but upon update contention + * instead scans for more tasks, later retrying inactivation if it + * doesn't find any. + * + * 4. Managing idle workers waiting for tasks. We cannot let + * workers spin indefinitely scanning for tasks when none are + * available. On the other hand, we must quickly prod them into + * action when new tasks are submitted or generated. We + * park/unpark these idle workers using an event-count scheme. + * Field eventCount is incremented upon events that may enable + * workers that previously could not find a task to now find one: + * Submission of a new task to the pool, or another worker pushing + * a task onto a previously empty queue. (We also use this + * mechanism for termination and reconfiguration actions that + * require wakeups of idle workers). Each worker maintains its + * last known event count, and blocks when a scan for work did not + * find a task AND its lastEventCount matches the current + * eventCount. Waiting idle workers are recorded in a variant of + * Treiber stack headed by field eventWaiters which, when nonzero, + * encodes the thread index and count awaited for by the worker + * thread most recently calling eventSync. This thread in turn has + * a record (field nextEventWaiter) for the next waiting worker. + * In addition to allowing simpler decisions about need for + * wakeup, the event count bits in eventWaiters serve the role of + * tags to avoid ABA errors in Treiber stacks. To reduce delays + * in task diffusion, workers not otherwise occupied may invoke + * method releaseWaiters, that removes and signals (unparks) + * workers not waiting on current count. To minimize task + * production stalls associate with signalling, any worker pushing + * a task on an empty queue invokes the weaker method signalWork, + * that only releases idle workers until it detects interference + * by other threads trying to release, and lets them take + * over. The net effect is a tree-like diffusion of signals, where + * released threads (and possibly others) help with unparks. To + * further reduce contention effects a bit, failed CASes to + * increment field eventCount are tolerated without retries. + * Conceptually they are merged into the same event, which is OK + * when their only purpose is to enable workers to scan for work. + * + * 5. Managing suspension of extra workers. When a worker is about + * to block waiting for a join (or via ManagedBlockers), we may + * create a new thread to maintain parallelism level, or at least + * avoid starvation. Usually, extra threads are needed for only + * very short periods, yet join dependencies are such that we + * sometimes need them in bursts. Rather than create new threads + * each time this happens, we suspend no-longer-needed extra ones + * as "spares". For most purposes, we don't distinguish "extra" + * spare threads from normal "core" threads: On each call to + * preStep (the only point at which we can do this) a worker + * checks to see if there are now too many running workers, and if + * so, suspends itself. Methods tryAwaitJoin and awaitBlocker + * look for suspended threads to resume before considering + * creating a new replacement. We don't need a special data + * structure to maintain spares; simply scanning the workers array + * looking for worker.isSuspended() is fine because the calling + * thread is otherwise not doing anything useful anyway; we are at + * least as happy if after locating a spare, the caller doesn't + * actually block because the join is ready before we try to + * adjust and compensate. Note that this is intrinsically racy. + * One thread may become a spare at about the same time as another + * is needlessly being created. We counteract this and related + * slop in part by requiring resumed spares to immediately recheck + * (in preStep) to see whether they they should re-suspend. The + * only effective difference between "extra" and "core" threads is + * that we allow the "extra" ones to time out and die if they are + * not resumed within a keep-alive interval of a few seconds. This + * is implemented mainly within ForkJoinWorkerThread, but requires + * some coordination (isTrimmed() -- meaning killed while + * suspended) to correctly maintain pool counts. + * + * 6. Deciding when to create new workers. The main dynamic + * control in this class is deciding when to create extra threads, + * in methods awaitJoin and awaitBlocker. We always need to create + * one when the number of running threads would become zero and + * all workers are busy. However, this is not easy to detect + * reliably in the presence of transients so we use retries and + * allow slack (in tryAwaitJoin) to reduce false alarms. These + * effectively reduce churn at the price of systematically + * undershooting target parallelism when many threads are blocked. + * However, biasing toward undeshooting partially compensates for + * the above mechanics to suspend extra threads, that normally + * lead to overshoot because we can only suspend workers + * in-between top-level actions. It also better copes with the + * fact that some of the methods in this class tend to never + * become compiled (but are interpreted), so some components of + * the entire set of controls might execute many times faster than + * others. And similarly for cases where the apparent lack of work + * is just due to GC stalls and other transient system activity. + * + * Beware that there is a lot of representation-level coupling + * among classes ForkJoinPool, ForkJoinWorkerThread, and + * ForkJoinTask. For example, direct access to "workers" array by + * workers, and direct access to ForkJoinTask.status by both + * ForkJoinPool and ForkJoinWorkerThread. There is little point + * trying to reduce this, since any associated future changes in + * representations will need to be accompanied by algorithmic + * changes anyway. + * + * Style notes: There are lots of inline assignments (of form + * "while ((local = field) != 0)") which are usually the simplest + * way to ensure read orderings. Also several occurrences of the + * unusual "do {} while(!cas...)" which is the simplest way to + * force an update of a CAS'ed variable. There are also other + * coding oddities that help some methods perform reasonably even + * when interpreted (not compiled), at the expense of messiness. + * + * The order of declarations in this file is: (1) statics (2) + * fields (along with constants used when unpacking some of them) + * (3) internal control methods (4) callbacks and other support + * for ForkJoinTask and ForkJoinWorkerThread classes, (5) exported + * methods (plus a few little helpers). + */ + + /** + * Factory for creating new {@link ForkJoinWorkerThread}s. + * A {@code ForkJoinWorkerThreadFactory} must be defined and used + * for {@code ForkJoinWorkerThread} subclasses that extend base + * functionality or initialize threads with different contexts. */ public static interface ForkJoinWorkerThreadFactory { /** * Returns a new worker thread operating in the given pool. * * @param pool the pool this thread works in - * @throws NullPointerException if pool is null + * @throws NullPointerException if the pool is null */ public ForkJoinWorkerThread newThread(ForkJoinPool pool); } @@ -100,14 +368,10 @@ public class ForkJoinPool extends Abstra * Default ForkJoinWorkerThreadFactory implementation; creates a * new ForkJoinWorkerThread. */ - static class DefaultForkJoinWorkerThreadFactory + static class DefaultForkJoinWorkerThreadFactory implements ForkJoinWorkerThreadFactory { public ForkJoinWorkerThread newThread(ForkJoinPool pool) { - try { - return new ForkJoinWorkerThread(pool); - } catch (OutOfMemoryError oom) { - return null; - } + return new ForkJoinWorkerThread(pool); } } @@ -143,405 +407,926 @@ public class ForkJoinPool extends Abstra new AtomicInteger(); /** - * Array holding all worker threads in the pool. Initialized upon - * first use. Array size must be a power of two. Updates and - * replacements are protected by workerLock, but it is always kept - * in a consistent enough state to be randomly accessed without - * locking by workers performing work-stealing. + * Absolute bound for parallelism level. Twice this number must + * fit into a 16bit field to enable word-packing for some counts. */ - volatile ForkJoinWorkerThread[] workers; + private static final int MAX_THREADS = 0x7fff; /** - * Lock protecting access to workers. + * Array holding all worker threads in the pool. Array size must + * be a power of two. Updates and replacements are protected by + * workerLock, but the array is always kept in a consistent enough + * state to be randomly accessed without locking by workers + * performing work-stealing, as well as other traversal-based + * methods in this class. All readers must tolerate that some + * array slots may be null. */ - private final ReentrantLock workerLock; + volatile ForkJoinWorkerThread[] workers; /** - * Condition for awaitTermination. + * Queue for external submissions. */ - private final Condition termination; + private final LinkedTransferQueue> submissionQueue; /** - * The uncaught exception handler used when any worker - * abruptly terminates + * Lock protecting updates to workers array. */ - private Thread.UncaughtExceptionHandler ueh; + private final ReentrantLock workerLock; /** - * Creation factory for worker threads. + * Latch released upon termination. */ - private final ForkJoinWorkerThreadFactory factory; + private final Phaser termination; /** - * Head of stack of threads that were created to maintain - * parallelism when other threads blocked, but have since - * suspended when the parallelism level rose. + * Creation factory for worker threads. */ - private volatile WaitQueueNode spareStack; + private final ForkJoinWorkerThreadFactory factory; /** * Sum of per-thread steal counts, updated only when threads are * idle or terminating. */ - private final AtomicLong stealCount; + private volatile long stealCount; /** - * Queue for external submissions. + * Encoded record of top of treiber stack of threads waiting for + * events. The top 32 bits contain the count being waited for. The + * bottom word contains one plus the pool index of waiting worker + * thread. */ - private final LinkedTransferQueue> submissionQueue; + private volatile long eventWaiters; + + private static final int EVENT_COUNT_SHIFT = 32; + private static final long WAITER_ID_MASK = (1L << EVENT_COUNT_SHIFT)-1L; /** - * Head of Treiber stack for barrier sync. See below for explanation. + * A counter for events that may wake up worker threads: + * - Submission of a new task to the pool + * - A worker pushing a task on an empty queue + * - termination and reconfiguration */ - private volatile WaitQueueNode syncStack; + private volatile int eventCount; /** - * The count for event barrier - */ - private volatile long eventCount; + * Lifecycle control. The low word contains the number of workers + * that are (probably) executing tasks. This value is atomically + * incremented before a worker gets a task to run, and decremented + * when worker has no tasks and cannot find any. Bits 16-18 + * contain runLevel value. When all are zero, the pool is + * running. Level transitions are monotonic (running -> shutdown + * -> terminating -> terminated) so each transition adds a bit. + * These are bundled together to ensure consistent read for + * termination checks (i.e., that runLevel is at least SHUTDOWN + * and active threads is zero). + */ + private volatile int runState; + + // Note: The order among run level values matters. + private static final int RUNLEVEL_SHIFT = 16; + private static final int SHUTDOWN = 1 << RUNLEVEL_SHIFT; + private static final int TERMINATING = 1 << (RUNLEVEL_SHIFT + 1); + private static final int TERMINATED = 1 << (RUNLEVEL_SHIFT + 2); + private static final int ACTIVE_COUNT_MASK = (1 << RUNLEVEL_SHIFT) - 1; + private static final int ONE_ACTIVE = 1; // active update delta /** - * Pool number, just for assigning useful names to worker threads + * Holds number of total (i.e., created and not yet terminated) + * and running (i.e., not blocked on joins or other managed sync) + * threads, packed together to ensure consistent snapshot when + * making decisions about creating and suspending spare + * threads. Updated only by CAS. Note that adding a new worker + * requires incrementing both counts, since workers start off in + * running state. This field is also used for memory-fencing + * configuration parameters. */ - private final int poolNumber; + private volatile int workerCounts; + + private static final int TOTAL_COUNT_SHIFT = 16; + private static final int RUNNING_COUNT_MASK = (1 << TOTAL_COUNT_SHIFT) - 1; + private static final int ONE_RUNNING = 1; + private static final int ONE_TOTAL = 1 << TOTAL_COUNT_SHIFT; /** - * The maximum allowed pool size + * The target parallelism level. + * Accessed directly by ForkJoinWorkerThreads. */ - private volatile int maxPoolSize; + final int parallelism; /** - * The desired parallelism level, updated only under workerLock. + * True if use local fifo, not default lifo, for local polling + * Read by, and replicated by ForkJoinWorkerThreads */ - private volatile int parallelism; + final boolean locallyFifo; /** - * True if use local fifo, not default lifo, for local polling + * The uncaught exception handler used when any worker abruptly + * terminates. */ - private volatile boolean locallyFifo; + private final Thread.UncaughtExceptionHandler ueh; /** - * Holds number of total (i.e., created and not yet terminated) - * and running (i.e., not blocked on joins or other managed sync) - * threads, packed into one int to ensure consistent snapshot when - * making decisions about creating and suspending spare - * threads. Updated only by CAS. Note: CASes in - * updateRunningCount and preJoin assume that running active count - * is in low word, so need to be modified if this changes. + * Pool number, just for assigning useful names to worker threads */ - private volatile int workerCounts; + private final int poolNumber; - private static int totalCountOf(int s) { return s >>> 16; } - private static int runningCountOf(int s) { return s & shortMask; } - private static int workerCountsFor(int t, int r) { return (t << 16) + r; } + // Utilities for CASing fields. Note that several of these + // are manually inlined by callers /** - * Adds delta (which may be negative) to running count. This must - * be called before (with negative arg) and after (with positive) - * any managed synchronization (i.e., mainly, joins). - * - * @param delta the number to add + * Increments running count. Also used by ForkJoinTask. */ - final void updateRunningCount(int delta) { - int s; - do {} while (!casWorkerCounts(s = workerCounts, s + delta)); + final void incrementRunningCount() { + int c; + do {} while (!UNSAFE.compareAndSwapInt(this, workerCountsOffset, + c = workerCounts, + c + ONE_RUNNING)); } /** - * Adds delta (which may be negative) to both total and running - * count. This must be called upon creation and termination of - * worker threads. - * - * @param delta the number to add + * Tries to decrement running count unless already zero */ - private void updateWorkerCount(int delta) { - int d = delta + (delta << 16); // add to both lo and hi parts - int s; - do {} while (!casWorkerCounts(s = workerCounts, s + d)); + final boolean tryDecrementRunningCount() { + int wc = workerCounts; + if ((wc & RUNNING_COUNT_MASK) == 0) + return false; + return UNSAFE.compareAndSwapInt(this, workerCountsOffset, + wc, wc - ONE_RUNNING); } /** - * Lifecycle control. High word contains runState, low word - * contains the number of workers that are (probably) executing - * tasks. This value is atomically incremented before a worker - * gets a task to run, and decremented when worker has no tasks - * and cannot find any. These two fields are bundled together to - * support correct termination triggering. Note: activeCount - * CAS'es cheat by assuming active count is in low word, so need - * to be modified if this changes + * Tries to increment running count */ - private volatile int runControl; - - // RunState values. Order among values matters - private static final int RUNNING = 0; - private static final int SHUTDOWN = 1; - private static final int TERMINATING = 2; - private static final int TERMINATED = 3; - - private static int runStateOf(int c) { return c >>> 16; } - private static int activeCountOf(int c) { return c & shortMask; } - private static int runControlFor(int r, int a) { return (r << 16) + a; } + final boolean tryIncrementRunningCount() { + int wc; + return UNSAFE.compareAndSwapInt(this, workerCountsOffset, + wc = workerCounts, wc + ONE_RUNNING); + } /** * Tries incrementing active count; fails on contention. - * Called by workers before/during executing tasks. + * Called by workers before executing tasks. * * @return true on success */ final boolean tryIncrementActiveCount() { - int c = runControl; - return casRunControl(c, c+1); + int c; + return UNSAFE.compareAndSwapInt(this, runStateOffset, + c = runState, c + ONE_ACTIVE); } /** * Tries decrementing active count; fails on contention. - * Possibly triggers termination on success. - * Called by workers when they can't find tasks. - * - * @return true on success + * Called when workers cannot find tasks to run. */ final boolean tryDecrementActiveCount() { - int c = runControl; - int nextc = c - 1; - if (!casRunControl(c, nextc)) - return false; - if (canTerminateOnShutdown(nextc)) - terminateOnShutdown(); - return true; + int c; + return UNSAFE.compareAndSwapInt(this, runStateOffset, + c = runState, c - ONE_ACTIVE); } /** - * Returns {@code true} if argument represents zero active count - * and nonzero runstate, which is the triggering condition for - * terminating on shutdown. + * Advances to at least the given level. Returns true if not + * already in at least the given level. */ - private static boolean canTerminateOnShutdown(int c) { - // i.e. least bit is nonzero runState bit - return ((c & -c) >>> 16) != 0; + private boolean advanceRunLevel(int level) { + for (;;) { + int s = runState; + if ((s & level) != 0) + return false; + if (UNSAFE.compareAndSwapInt(this, runStateOffset, s, s | level)) + return true; + } } + // workers array maintenance + /** - * Transition run state to at least the given state. Return true - * if not already at least given state. + * Records and returns a workers array index for new worker. */ - private boolean transitionRunStateTo(int state) { - for (;;) { - int c = runControl; - if (runStateOf(c) >= state) - return false; - if (casRunControl(c, runControlFor(state, activeCountOf(c)))) - return true; + private int recordWorker(ForkJoinWorkerThread w) { + // Try using slot totalCount-1. If not available, scan and/or resize + int k = (workerCounts >>> TOTAL_COUNT_SHIFT) - 1; + final ReentrantLock lock = this.workerLock; + lock.lock(); + try { + ForkJoinWorkerThread[] ws = workers; + int nws = ws.length; + if (k < 0 || k >= nws || ws[k] != null) { + for (k = 0; k < nws && ws[k] != null; ++k) + ; + if (k == nws) + ws = Arrays.copyOf(ws, nws << 1); + } + ws[k] = w; + workers = ws; // volatile array write ensures slot visibility + } finally { + lock.unlock(); } + return k; } /** - * Controls whether to add spares to maintain parallelism + * Nulls out record of worker in workers array */ - private volatile boolean maintainsParallelism; + private void forgetWorker(ForkJoinWorkerThread w) { + int idx = w.poolIndex; + // Locking helps method recordWorker avoid unecessary expansion + final ReentrantLock lock = this.workerLock; + lock.lock(); + try { + ForkJoinWorkerThread[] ws = workers; + if (idx >= 0 && idx < ws.length && ws[idx] == w) // verify + ws[idx] = null; + } finally { + lock.unlock(); + } + } - // Constructors + // adding and removing workers /** - * Creates a ForkJoinPool with a pool size equal to the number of - * processors available on the system, using the default - * ForkJoinWorkerThreadFactory. + * Tries to create and add new worker. Assumes that worker counts + * are already updated to accommodate the worker, so adjusts on + * failure. * - * @throws SecurityException if a security manager exists and - * the caller is not permitted to modify threads - * because it does not hold {@link - * java.lang.RuntimePermission}{@code ("modifyThread")} + * @return new worker or null if creation failed */ - public ForkJoinPool() { - this(Runtime.getRuntime().availableProcessors(), - defaultForkJoinWorkerThreadFactory); + private ForkJoinWorkerThread addWorker() { + ForkJoinWorkerThread w = null; + try { + w = factory.newThread(this); + } finally { // Adjust on either null or exceptional factory return + if (w == null) { + onWorkerCreationFailure(); + return null; + } + } + w.start(recordWorker(w), ueh); + return w; } /** - * Creates a ForkJoinPool with the indicated parallelism level - * threads and using the default ForkJoinWorkerThreadFactory. - * - * @param parallelism the number of worker threads - * @throws IllegalArgumentException if parallelism less than or - * equal to zero - * @throws SecurityException if a security manager exists and - * the caller is not permitted to modify threads - * because it does not hold {@link - * java.lang.RuntimePermission}{@code ("modifyThread")} + * Adjusts counts upon failure to create worker */ - public ForkJoinPool(int parallelism) { - this(parallelism, defaultForkJoinWorkerThreadFactory); + private void onWorkerCreationFailure() { + for (;;) { + int wc = workerCounts; + if ((wc >>> TOTAL_COUNT_SHIFT) == 0) + Thread.yield(); // wait for other counts to settle + else if (UNSAFE.compareAndSwapInt(this, workerCountsOffset, wc, + wc - (ONE_RUNNING|ONE_TOTAL))) + break; + } + tryTerminate(false); // in case of failure during shutdown } /** - * Creates a ForkJoinPool with parallelism equal to the number of - * processors available on the system and using the given - * ForkJoinWorkerThreadFactory. + * Creates and/or resumes enough workers to establish target + * parallelism, giving up if terminating or addWorker fails * - * @param factory the factory for creating new threads - * @throws NullPointerException if factory is null - * @throws SecurityException if a security manager exists and - * the caller is not permitted to modify threads - * because it does not hold {@link - * java.lang.RuntimePermission}{@code ("modifyThread")} + * TODO: recast this to support lazier creation and automated + * parallelism maintenance */ - public ForkJoinPool(ForkJoinWorkerThreadFactory factory) { - this(Runtime.getRuntime().availableProcessors(), factory); + private void ensureEnoughWorkers() { + for (;;) { + int pc = parallelism; + int wc = workerCounts; + int rc = wc & RUNNING_COUNT_MASK; + int tc = wc >>> TOTAL_COUNT_SHIFT; + if (tc < pc) { + if (runState == TERMINATING || + (UNSAFE.compareAndSwapInt + (this, workerCountsOffset, + wc, wc + (ONE_RUNNING|ONE_TOTAL)) && + addWorker() == null)) + break; + } + else if (tc > pc && rc < pc && + tc > (runState & ACTIVE_COUNT_MASK)) { + ForkJoinWorkerThread spare = null; + ForkJoinWorkerThread[] ws = workers; + int nws = ws.length; + for (int i = 0; i < nws; ++i) { + ForkJoinWorkerThread w = ws[i]; + if (w != null && w.isSuspended()) { + if ((workerCounts & RUNNING_COUNT_MASK) > pc || + runState == TERMINATING) + return; + if (w.tryResumeSpare()) + incrementRunningCount(); + break; + } + } + } + else + break; + } } /** - * Creates a ForkJoinPool with the given parallelism and factory. + * Final callback from terminating worker. Removes record of + * worker from array, and adjusts counts. If pool is shutting + * down, tries to complete terminatation, else possibly replaces + * the worker. * - * @param parallelism the targeted number of worker threads - * @param factory the factory for creating new threads - * @throws IllegalArgumentException if parallelism less than or - * equal to zero, or greater than implementation limit - * @throws NullPointerException if factory is null - * @throws SecurityException if a security manager exists and - * the caller is not permitted to modify threads - * because it does not hold {@link - * java.lang.RuntimePermission}{@code ("modifyThread")} + * @param w the worker */ - public ForkJoinPool(int parallelism, ForkJoinWorkerThreadFactory factory) { - if (parallelism <= 0 || parallelism > MAX_THREADS) - throw new IllegalArgumentException(); - if (factory == null) - throw new NullPointerException(); - checkPermission(); - this.factory = factory; - this.parallelism = parallelism; - this.maxPoolSize = MAX_THREADS; - this.maintainsParallelism = true; - this.poolNumber = poolNumberGenerator.incrementAndGet(); - this.workerLock = new ReentrantLock(); - this.termination = workerLock.newCondition(); - this.stealCount = new AtomicLong(); - this.submissionQueue = new LinkedTransferQueue>(); - // worker array and workers are lazily constructed + final void workerTerminated(ForkJoinWorkerThread w) { + if (w.active) { // force inactive + w.active = false; + do {} while (!tryDecrementActiveCount()); + } + forgetWorker(w); + + // Decrement total count, and if was running, running count + // Spin (waiting for other updates) if either would be negative + int nr = w.isTrimmed() ? 0 : ONE_RUNNING; + int unit = ONE_TOTAL + nr; + for (;;) { + int wc = workerCounts; + int rc = wc & RUNNING_COUNT_MASK; + if (rc - nr < 0 || (wc >>> TOTAL_COUNT_SHIFT) == 0) + Thread.yield(); // back off if waiting for other updates + else if (UNSAFE.compareAndSwapInt(this, workerCountsOffset, + wc, wc - unit)) + break; + } + + accumulateStealCount(w); // collect final count + if (!tryTerminate(false)) + ensureEnoughWorkers(); } + // Waiting for and signalling events + /** - * Creates a new worker thread using factory. - * - * @param index the index to assign worker - * @return new worker, or null of factory failed + * Releases workers blocked on a count not equal to current count. + * @return true if any released */ - private ForkJoinWorkerThread createWorker(int index) { - Thread.UncaughtExceptionHandler h = ueh; - ForkJoinWorkerThread w = factory.newThread(this); - if (w != null) { - w.poolIndex = index; - w.setDaemon(true); - w.setAsyncMode(locallyFifo); - w.setName("ForkJoinPool-" + poolNumber + "-worker-" + index); - if (h != null) - w.setUncaughtExceptionHandler(h); + private void releaseWaiters() { + long top; + while ((top = eventWaiters) != 0L) { + ForkJoinWorkerThread[] ws = workers; + int n = ws.length; + for (;;) { + int i = ((int)(top & WAITER_ID_MASK)) - 1; + if (i < 0 || (int)(top >>> EVENT_COUNT_SHIFT) == eventCount) + return; + ForkJoinWorkerThread w; + if (i < n && (w = ws[i]) != null && + UNSAFE.compareAndSwapLong(this, eventWaitersOffset, + top, w.nextWaiter)) { + LockSupport.unpark(w); + top = eventWaiters; + } + else + break; // possibly stale; reread + } } - return w; } /** - * Returns a good size for worker array given pool size. - * Currently requires size to be a power of two. + * Ensures eventCount on exit is different (mod 2^32) than on + * entry and wakes up all waiters */ - private static int arraySizeFor(int poolSize) { - return (poolSize <= 1) ? 1 : - (1 << (32 - Integer.numberOfLeadingZeros(poolSize-1))); + private void signalEvent() { + int c; + do {} while (!UNSAFE.compareAndSwapInt(this, eventCountOffset, + c = eventCount, c+1)); + releaseWaiters(); } /** - * Creates or resizes array if necessary to hold newLength. - * Call only under exclusion. + * Advances eventCount and releases waiters until interference by + * other releasing threads is detected. + */ + final void signalWork() { + int c; + UNSAFE.compareAndSwapInt(this, eventCountOffset, c=eventCount, c+1); + long top; + while ((top = eventWaiters) != 0L) { + int ec = eventCount; + ForkJoinWorkerThread[] ws = workers; + int n = ws.length; + for (;;) { + int i = ((int)(top & WAITER_ID_MASK)) - 1; + if (i < 0 || (int)(top >>> EVENT_COUNT_SHIFT) == ec) + return; + ForkJoinWorkerThread w; + if (i < n && (w = ws[i]) != null && + UNSAFE.compareAndSwapLong(this, eventWaitersOffset, + top, top = w.nextWaiter)) { + LockSupport.unpark(w); + if (top != eventWaiters) // let someone else take over + return; + } + else + break; // possibly stale; reread + } + } + } + + /** + * If worker is inactive, blocks until terminating or event count + * advances from last value held by worker; in any case helps + * release others. * - * @return the array + * @param w the calling worker thread + * @param retries the number of scans by caller failing to find work + * @return false if now too many threads running */ - private ForkJoinWorkerThread[] ensureWorkerArrayCapacity(int newLength) { - ForkJoinWorkerThread[] ws = workers; - if (ws == null) - return workers = new ForkJoinWorkerThread[arraySizeFor(newLength)]; - else if (newLength > ws.length) - return workers = Arrays.copyOf(ws, arraySizeFor(newLength)); - else - return ws; + private boolean eventSync(ForkJoinWorkerThread w, int retries) { + int wec = w.lastEventCount; + if (retries > 1) { // can only block after 2nd miss + long nextTop = (((long)wec << EVENT_COUNT_SHIFT) | + ((long)(w.poolIndex + 1))); + long top; + while ((runState < SHUTDOWN || !tryTerminate(false)) && + (((int)(top = eventWaiters) & WAITER_ID_MASK) == 0 || + (int)(top >>> EVENT_COUNT_SHIFT) == wec) && + eventCount == wec) { + if (UNSAFE.compareAndSwapLong(this, eventWaitersOffset, + w.nextWaiter = top, nextTop)) { + accumulateStealCount(w); // transfer steals while idle + Thread.interrupted(); // clear/ignore interrupt + while (eventCount == wec) + w.doPark(); + break; + } + } + wec = eventCount; + } + releaseWaiters(); + int wc = workerCounts; + if ((wc & RUNNING_COUNT_MASK) <= parallelism) { + w.lastEventCount = wec; + return true; + } + if (wec != w.lastEventCount) // back up if may re-wait + w.lastEventCount = wec - (wc >>> TOTAL_COUNT_SHIFT); + return false; } /** - * Tries to shrink workers into smaller array after one or more terminate. + * Callback from workers invoked upon each top-level action (i.e., + * stealing a task or taking a submission and running + * it). Performs one or both of the following: + * + * * If the worker cannot find work, updates its active status to + * inactive and updates activeCount unless there is contention, in + * which case it may try again (either in this or a subsequent + * call). Additionally, awaits the next task event and/or helps + * wake up other releasable waiters. + * + * * If there are too many running threads, suspends this worker + * (first forcing inactivation if necessary). If it is not + * resumed before a keepAlive elapses, the worker may be "trimmed" + * -- killed while suspended within suspendAsSpare. Otherwise, + * upon resume it rechecks to make sure that it is still needed. + * + * @param w the worker + * @param retries the number of scans by caller failing to find work + * find any (in which case it may block waiting for work). */ - private void tryShrinkWorkerArray() { - ForkJoinWorkerThread[] ws = workers; - if (ws != null) { - int len = ws.length; - int last = len - 1; - while (last >= 0 && ws[last] == null) - --last; - int newLength = arraySizeFor(last+1); - if (newLength < len) - workers = Arrays.copyOf(ws, newLength); + final void preStep(ForkJoinWorkerThread w, int retries) { + boolean active = w.active; + boolean inactivate = active && retries != 0; + for (;;) { + int rs, wc; + if (inactivate && + UNSAFE.compareAndSwapInt(this, runStateOffset, + rs = runState, rs - ONE_ACTIVE)) + inactivate = active = w.active = false; + if (((wc = workerCounts) & RUNNING_COUNT_MASK) <= parallelism) { + if (active || eventSync(w, retries)) + break; + } + else if (!(inactivate |= active) && // must inactivate to suspend + UNSAFE.compareAndSwapInt(this, workerCountsOffset, + wc, wc - ONE_RUNNING) && + !w.suspendAsSpare()) // false if trimmed + break; } } /** - * Initializes workers if necessary. + * Awaits join of the given task if enough threads, or can resume + * or create a spare. Fails (in which case the given task might + * not be done) upon contention or lack of decision about + * blocking. Returns void because caller must check + * task status on return anyway. + * + * We allow blocking if: + * + * 1. There would still be at least as many running threads as + * parallelism level if this thread blocks. + * + * 2. A spare is resumed to replace this worker. We tolerate + * slop in the decision to replace if a spare is found without + * first decrementing run count. This may release too many, + * but if so, the superfluous ones will re-suspend via + * preStep(). + * + * 3. After #spares repeated checks, there are no fewer than #spare + * threads not running. We allow this slack to avoid hysteresis + * and as a hedge against lag/uncertainty of running count + * estimates when signalling or unblocking stalls. + * + * 4. All existing workers are busy (as rechecked via repeated + * retries by caller) and a new spare is created. + * + * If none of the above hold, we try to escape out by + * re-incrementing count and returning to caller, which can retry + * later. + * + * @param joinMe the task to join + * @param retries if negative, then serve only as a precheck + * that the thread can be replaced by a spare. Otherwise, + * the number of repeated calls to this method returning busy + * @return true if the call must be retried because there + * none of the blocking checks hold */ - final void ensureWorkerInitialization() { - ForkJoinWorkerThread[] ws = workers; - if (ws == null) { - final ReentrantLock lock = this.workerLock; - lock.lock(); - try { - ws = workers; - if (ws == null) { - int ps = parallelism; - ws = ensureWorkerArrayCapacity(ps); - for (int i = 0; i < ps; ++i) { - ForkJoinWorkerThread w = createWorker(i); + final boolean tryAwaitJoin(ForkJoinTask joinMe, int retries) { + if (joinMe.status < 0) // precheck to prime loop + return false; + int pc = parallelism; + boolean running = true; // false when running count decremented + outer:for (;;) { + int wc = workerCounts; + int rc = wc & RUNNING_COUNT_MASK; + int tc = wc >>> TOTAL_COUNT_SHIFT; + if (running) { // replace with spare or decrement count + if (rc <= pc && tc > pc && + (retries > 0 || tc > (runState & ACTIVE_COUNT_MASK))) { + ForkJoinWorkerThread[] ws = workers; + int nws = ws.length; + for (int i = 0; i < nws; ++i) { // search for spare + ForkJoinWorkerThread w = ws[i]; if (w != null) { - ws[i] = w; - w.start(); - updateWorkerCount(1); + if (joinMe.status < 0) + return false; + if (w.isSuspended()) { + if ((workerCounts & RUNNING_COUNT_MASK)>=pc && + w.tryResumeSpare()) { + running = false; + break outer; + } + continue outer; // rescan + } } } } - } finally { - lock.unlock(); + if (retries < 0 || // < 0 means replacement check only + rc == 0 || joinMe.status < 0 || workerCounts != wc || + !UNSAFE.compareAndSwapInt(this, workerCountsOffset, + wc, wc - ONE_RUNNING)) + return false; // done or inconsistent or contended + running = false; + if (rc > pc) + break; + } + else { // allow blocking if enough threads + if (rc >= pc || joinMe.status < 0) + break; + int sc = tc - pc + 1; // = spare threads, plus the one to add + if (retries > sc) { + if (rc > 0 && rc >= pc - sc) // allow slack + break; + if (tc < MAX_THREADS && + tc == (runState & ACTIVE_COUNT_MASK) && + workerCounts == wc && + UNSAFE.compareAndSwapInt(this, workerCountsOffset, wc, + wc+(ONE_RUNNING|ONE_TOTAL))) { + addWorker(); + break; + } + } + if (workerCounts == wc && // back out to allow rescan + UNSAFE.compareAndSwapInt (this, workerCountsOffset, + wc, wc + ONE_RUNNING)) { + releaseWaiters(); // help others progress + return true; // let caller retry + } } } + // arrive here if can block + joinMe.internalAwaitDone(); + int c; // to inline incrementRunningCount + do {} while (!UNSAFE.compareAndSwapInt + (this, workerCountsOffset, + c = workerCounts, c + ONE_RUNNING)); + return false; } /** - * Worker creation and startup for threads added via setParallelism. + * Same idea as (and shares many code snippets with) tryAwaitJoin, + * but self-contained because there are no caller retries. + * TODO: Rework to use simpler API. */ - private void createAndStartAddedWorkers() { - resumeAllSpares(); // Allow spares to convert to nonspare - int ps = parallelism; - ForkJoinWorkerThread[] ws = ensureWorkerArrayCapacity(ps); - int len = ws.length; - // Sweep through slots, to keep lowest indices most populated - int k = 0; - while (k < len) { - if (ws[k] != null) { - ++k; - continue; - } - int s = workerCounts; - int tc = totalCountOf(s); - int rc = runningCountOf(s); - if (rc >= ps || tc >= ps) - break; - if (casWorkerCounts (s, workerCountsFor(tc+1, rc+1))) { - ForkJoinWorkerThread w = createWorker(k); - if (w != null) { - ws[k++] = w; - w.start(); + final void awaitBlocker(ManagedBlocker blocker) + throws InterruptedException { + boolean done; + if (done = blocker.isReleasable()) + return; + int pc = parallelism; + int retries = 0; + boolean running = true; // false when running count decremented + outer:for (;;) { + int wc = workerCounts; + int rc = wc & RUNNING_COUNT_MASK; + int tc = wc >>> TOTAL_COUNT_SHIFT; + if (running) { + if (rc <= pc && tc > pc && + (retries > 0 || tc > (runState & ACTIVE_COUNT_MASK))) { + ForkJoinWorkerThread[] ws = workers; + int nws = ws.length; + for (int i = 0; i < nws; ++i) { + ForkJoinWorkerThread w = ws[i]; + if (w != null) { + if (done = blocker.isReleasable()) + return; + if (w.isSuspended()) { + if ((workerCounts & RUNNING_COUNT_MASK)>=pc && + w.tryResumeSpare()) { + running = false; + break outer; + } + continue outer; // rescan + } + } + } } - else { - updateWorkerCount(-1); // back out on failed creation + if (done = blocker.isReleasable()) + return; + if (rc == 0 || workerCounts != wc || + !UNSAFE.compareAndSwapInt(this, workerCountsOffset, + wc, wc - ONE_RUNNING)) + continue; + running = false; + if (rc > pc) break; + } + else { + if (rc >= pc || (done = blocker.isReleasable())) + break; + int sc = tc - pc + 1; + if (retries++ > sc) { + if (rc > 0 && rc >= pc - sc) + break; + if (tc < MAX_THREADS && + tc == (runState & ACTIVE_COUNT_MASK) && + workerCounts == wc && + UNSAFE.compareAndSwapInt(this, workerCountsOffset, wc, + wc+(ONE_RUNNING|ONE_TOTAL))) { + addWorker(); + break; + } } + Thread.yield(); + } + } + + try { + if (!done) + do {} while (!blocker.isReleasable() && !blocker.block()); + } finally { + if (!running) { + int c; + do {} while (!UNSAFE.compareAndSwapInt + (this, workerCountsOffset, + c = workerCounts, c + ONE_RUNNING)); + } + } + } + + /** + * Possibly initiates and/or completes termination. + * + * @param now if true, unconditionally terminate, else only + * if shutdown and empty queue and no active workers + * @return true if now terminating or terminated + */ + private boolean tryTerminate(boolean now) { + if (now) + advanceRunLevel(SHUTDOWN); // ensure at least SHUTDOWN + else if (runState < SHUTDOWN || + !submissionQueue.isEmpty() || + (runState & ACTIVE_COUNT_MASK) != 0) + return false; + + if (advanceRunLevel(TERMINATING)) + startTerminating(); + + // Finish now if all threads terminated; else in some subsequent call + if ((workerCounts >>> TOTAL_COUNT_SHIFT) == 0) { + advanceRunLevel(TERMINATED); + termination.arrive(); + } + return true; + } + + /** + * Actions on transition to TERMINATING + */ + private void startTerminating() { + for (int i = 0; i < 2; ++i) { // twice to mop up newly created workers + cancelSubmissions(); + shutdownWorkers(); + cancelWorkerTasks(); + signalEvent(); + interruptWorkers(); + } + } + + /** + * Clear out and cancel submissions, ignoring exceptions + */ + private void cancelSubmissions() { + ForkJoinTask task; + while ((task = submissionQueue.poll()) != null) { + try { + task.cancel(false); + } catch (Throwable ignore) { } } } + /** + * Sets all worker run states to at least shutdown, + * also resuming suspended workers + */ + private void shutdownWorkers() { + ForkJoinWorkerThread[] ws = workers; + int nws = ws.length; + for (int i = 0; i < nws; ++i) { + ForkJoinWorkerThread w = ws[i]; + if (w != null) + w.shutdown(); + } + } + + /** + * Clears out and cancels all locally queued tasks + */ + private void cancelWorkerTasks() { + ForkJoinWorkerThread[] ws = workers; + int nws = ws.length; + for (int i = 0; i < nws; ++i) { + ForkJoinWorkerThread w = ws[i]; + if (w != null) + w.cancelTasks(); + } + } + + /** + * Unsticks all workers blocked on joins etc + */ + private void interruptWorkers() { + ForkJoinWorkerThread[] ws = workers; + int nws = ws.length; + for (int i = 0; i < nws; ++i) { + ForkJoinWorkerThread w = ws[i]; + if (w != null && !w.isTerminated()) { + try { + w.interrupt(); + } catch (SecurityException ignore) { + } + } + } + } + + // misc support for ForkJoinWorkerThread + + /** + * Returns pool number + */ + final int getPoolNumber() { + return poolNumber; + } + + /** + * Accumulates steal count from a worker, clearing + * the worker's value + */ + final void accumulateStealCount(ForkJoinWorkerThread w) { + int sc = w.stealCount; + if (sc != 0) { + long c; + w.stealCount = 0; + do {} while (!UNSAFE.compareAndSwapLong(this, stealCountOffset, + c = stealCount, c + sc)); + } + } + + /** + * Returns the approximate (non-atomic) number of idle threads per + * active thread. + */ + final int idlePerActive() { + int pc = parallelism; // use parallelism, not rc + int ac = runState; // no mask -- artifically boosts during shutdown + // Use exact results for small values, saturate past 4 + return pc <= ac? 0 : pc >>> 1 <= ac? 1 : pc >>> 2 <= ac? 3 : pc >>> 3; + } + + // Public and protected methods + + // Constructors + + /** + * Creates a {@code ForkJoinPool} with parallelism equal to {@link + * java.lang.Runtime#availableProcessors}, using the {@linkplain + * #defaultForkJoinWorkerThreadFactory default thread factory}, + * no UncaughtExceptionHandler, and non-async LIFO processing mode. + * + * @throws SecurityException if a security manager exists and + * the caller is not permitted to modify threads + * because it does not hold {@link + * java.lang.RuntimePermission}{@code ("modifyThread")} + */ + public ForkJoinPool() { + this(Runtime.getRuntime().availableProcessors(), + defaultForkJoinWorkerThreadFactory, null, false); + } + + /** + * Creates a {@code ForkJoinPool} with the indicated parallelism + * level, the {@linkplain + * #defaultForkJoinWorkerThreadFactory default thread factory}, + * no UncaughtExceptionHandler, and non-async LIFO processing mode. + * + * @param parallelism the parallelism level + * @throws IllegalArgumentException if parallelism less than or + * equal to zero, or greater than implementation limit + * @throws SecurityException if a security manager exists and + * the caller is not permitted to modify threads + * because it does not hold {@link + * java.lang.RuntimePermission}{@code ("modifyThread")} + */ + public ForkJoinPool(int parallelism) { + this(parallelism, defaultForkJoinWorkerThreadFactory, null, false); + } + + /** + * Creates a {@code ForkJoinPool} with the given parameters. + * + * @param parallelism the parallelism level. For default value, + * use {@link java.lang.Runtime#availableProcessors}. + * @param factory the factory for creating new threads. For default value, + * use {@link #defaultForkJoinWorkerThreadFactory}. + * @param handler the handler for internal worker threads that + * terminate due to unrecoverable errors encountered while executing + * tasks. For default value, use null. + * @param asyncMode if true, + * establishes local first-in-first-out scheduling mode for forked + * tasks that are never joined. This mode may be more appropriate + * than default locally stack-based mode in applications in which + * worker threads only process event-style asynchronous tasks. + * For default value, use false. + * @throws IllegalArgumentException if parallelism less than or + * equal to zero, or greater than implementation limit + * @throws NullPointerException if the factory is null + * @throws SecurityException if a security manager exists and + * the caller is not permitted to modify threads + * because it does not hold {@link + * java.lang.RuntimePermission}{@code ("modifyThread")} + */ + public ForkJoinPool(int parallelism, + ForkJoinWorkerThreadFactory factory, + Thread.UncaughtExceptionHandler handler, + boolean asyncMode) { + checkPermission(); + if (factory == null) + throw new NullPointerException(); + if (parallelism <= 0 || parallelism > MAX_THREADS) + throw new IllegalArgumentException(); + this.parallelism = parallelism; + this.factory = factory; + this.ueh = handler; + this.locallyFifo = asyncMode; + int arraySize = initialArraySizeFor(parallelism); + this.workers = new ForkJoinWorkerThread[arraySize]; + this.submissionQueue = new LinkedTransferQueue>(); + this.workerLock = new ReentrantLock(); + this.termination = new Phaser(1); + this.poolNumber = poolNumberGenerator.incrementAndGet(); + } + + /** + * Returns initial power of two size for workers array. + * @param pc the initial parallelism level + */ + private static int initialArraySizeFor(int pc) { + // See Hackers Delight, sec 3.2. We know MAX_THREADS < (1 >>> 16) + int size = pc < MAX_THREADS ? pc + 1 : MAX_THREADS; + size |= size >>> 1; + size |= size >>> 2; + size |= size >>> 4; + size |= size >>> 8; + return size + 1; + } + // Execution methods /** @@ -550,21 +1335,24 @@ public class ForkJoinPool extends Abstra private void doSubmit(ForkJoinTask task) { if (task == null) throw new NullPointerException(); - if (isShutdown()) + if (runState >= SHUTDOWN) throw new RejectedExecutionException(); - if (workers == null) - ensureWorkerInitialization(); submissionQueue.offer(task); - signalIdleWorkers(); + signalEvent(); + ensureEnoughWorkers(); } /** * Performs the given task, returning its result upon completion. + * If the caller is already engaged in a fork/join computation in + * the current pool, this method is equivalent in effect to + * {@link ForkJoinTask#invoke}. * * @param task the task * @return the task's result - * @throws NullPointerException if task is null - * @throws RejectedExecutionException if pool is shut down + * @throws NullPointerException if the task is null + * @throws RejectedExecutionException if the task cannot be + * scheduled for execution */ public T invoke(ForkJoinTask task) { doSubmit(task); @@ -573,17 +1361,26 @@ public class ForkJoinPool extends Abstra /** * Arranges for (asynchronous) execution of the given task. + * If the caller is already engaged in a fork/join computation in + * the current pool, this method is equivalent in effect to + * {@link ForkJoinTask#fork}. * * @param task the task - * @throws NullPointerException if task is null - * @throws RejectedExecutionException if pool is shut down + * @throws NullPointerException if the task is null + * @throws RejectedExecutionException if the task cannot be + * scheduled for execution */ - public void execute(ForkJoinTask task) { + public void execute(ForkJoinTask task) { doSubmit(task); } // AbstractExecutorService methods + /** + * @throws NullPointerException if the task is null + * @throws RejectedExecutionException if the task cannot be + * scheduled for execution + */ public void execute(Runnable task) { ForkJoinTask job; if (task instanceof ForkJoinTask) // avoid re-wrap @@ -593,18 +1390,50 @@ public class ForkJoinPool extends Abstra doSubmit(job); } + /** + * Submits a ForkJoinTask for execution. + * If the caller is already engaged in a fork/join computation in + * the current pool, this method is equivalent in effect to + * {@link ForkJoinTask#fork}. + * + * @param task the task to submit + * @return the task + * @throws NullPointerException if the task is null + * @throws RejectedExecutionException if the task cannot be + * scheduled for execution + */ + public ForkJoinTask submit(ForkJoinTask task) { + doSubmit(task); + return task; + } + + /** + * @throws NullPointerException if the task is null + * @throws RejectedExecutionException if the task cannot be + * scheduled for execution + */ public ForkJoinTask submit(Callable task) { ForkJoinTask job = ForkJoinTask.adapt(task); doSubmit(job); return job; } + /** + * @throws NullPointerException if the task is null + * @throws RejectedExecutionException if the task cannot be + * scheduled for execution + */ public ForkJoinTask submit(Runnable task, T result) { ForkJoinTask job = ForkJoinTask.adapt(task, result); doSubmit(job); return job; } + /** + * @throws NullPointerException if the task is null + * @throws RejectedExecutionException if the task cannot be + * scheduled for execution + */ public ForkJoinTask submit(Runnable task) { ForkJoinTask job; if (task instanceof ForkJoinTask) // avoid re-wrap @@ -616,20 +1445,9 @@ public class ForkJoinPool extends Abstra } /** - * Submits a ForkJoinTask for execution. - * - * @param task the task to submit - * @return the task - * @throws RejectedExecutionException if the task cannot be - * scheduled for execution - * @throws NullPointerException if the task is null + * @throws NullPointerException {@inheritDoc} + * @throws RejectedExecutionException {@inheritDoc} */ - public ForkJoinTask submit(ForkJoinTask task) { - doSubmit(task); - return task; - } - - public List> invokeAll(Collection> tasks) { ArrayList> forkJoinTasks = new ArrayList>(tasks.size()); @@ -638,7 +1456,7 @@ public class ForkJoinPool extends Abstra invoke(new InvokeAll(forkJoinTasks)); @SuppressWarnings({"unchecked", "rawtypes"}) - List> futures = (List>) (List) forkJoinTasks; + List> futures = (List>) (List) forkJoinTasks; return futures; } @@ -652,8 +1470,6 @@ public class ForkJoinPool extends Abstra private static final long serialVersionUID = -7914297376763021607L; } - // Configuration and status settings and queries - /** * Returns the factory used for constructing new workers. * @@ -670,90 +1486,13 @@ public class ForkJoinPool extends Abstra * @return the handler, or {@code null} if none */ public Thread.UncaughtExceptionHandler getUncaughtExceptionHandler() { - Thread.UncaughtExceptionHandler h; - final ReentrantLock lock = this.workerLock; - lock.lock(); - try { - h = ueh; - } finally { - lock.unlock(); - } - return h; - } - - /** - * Sets the handler for internal worker threads that terminate due - * to unrecoverable errors encountered while executing tasks. - * Unless set, the current default or ThreadGroup handler is used - * as handler. - * - * @param h the new handler - * @return the old handler, or {@code null} if none - * @throws SecurityException if a security manager exists and - * the caller is not permitted to modify threads - * because it does not hold {@link - * java.lang.RuntimePermission}{@code ("modifyThread")} - */ - public Thread.UncaughtExceptionHandler - setUncaughtExceptionHandler(Thread.UncaughtExceptionHandler h) { - checkPermission(); - Thread.UncaughtExceptionHandler old = null; - final ReentrantLock lock = this.workerLock; - lock.lock(); - try { - old = ueh; - ueh = h; - ForkJoinWorkerThread[] ws = workers; - if (ws != null) { - for (int i = 0; i < ws.length; ++i) { - ForkJoinWorkerThread w = ws[i]; - if (w != null) - w.setUncaughtExceptionHandler(h); - } - } - } finally { - lock.unlock(); - } - return old; - } - - - /** - * Sets the target parallelism level of this pool. - * - * @param parallelism the target parallelism - * @throws IllegalArgumentException if parallelism less than or - * equal to zero or greater than maximum size bounds - * @throws SecurityException if a security manager exists and - * the caller is not permitted to modify threads - * because it does not hold {@link - * java.lang.RuntimePermission}{@code ("modifyThread")} - */ - public void setParallelism(int parallelism) { - checkPermission(); - if (parallelism <= 0 || parallelism > maxPoolSize) - throw new IllegalArgumentException(); - final ReentrantLock lock = this.workerLock; - lock.lock(); - try { - if (!isTerminating()) { - int p = this.parallelism; - this.parallelism = parallelism; - if (parallelism > p) - createAndStartAddedWorkers(); - else - trimSpares(); - } - } finally { - lock.unlock(); - } - signalIdleWorkers(); + return ueh; } /** - * Returns the targeted number of worker threads in this pool. + * Returns the targeted parallelism level of this pool. * - * @return the targeted number of worker threads in this pool + * @return the targeted parallelism level of this pool */ public int getParallelism() { return parallelism; @@ -768,82 +1507,7 @@ public class ForkJoinPool extends Abstra * @return the number of worker threads */ public int getPoolSize() { - return totalCountOf(workerCounts); - } - - /** - * Returns the maximum number of threads allowed to exist in the - * pool, even if there are insufficient unblocked running threads. - * - * @return the maximum - */ - public int getMaximumPoolSize() { - return maxPoolSize; - } - - /** - * Sets the maximum number of threads allowed to exist in the - * pool, even if there are insufficient unblocked running threads. - * Setting this value has no effect on current pool size. It - * controls construction of new threads. - * - * @throws IllegalArgumentException if negative or greater then - * internal implementation limit - */ - public void setMaximumPoolSize(int newMax) { - if (newMax < 0 || newMax > MAX_THREADS) - throw new IllegalArgumentException(); - maxPoolSize = newMax; - } - - - /** - * Returns {@code true} if this pool dynamically maintains its - * target parallelism level. If false, new threads are added only - * to avoid possible starvation. This setting is by default true. - * - * @return {@code true} if maintains parallelism - */ - public boolean getMaintainsParallelism() { - return maintainsParallelism; - } - - /** - * Sets whether this pool dynamically maintains its target - * parallelism level. If false, new threads are added only to - * avoid possible starvation. - * - * @param enable {@code true} to maintain parallelism - */ - public void setMaintainsParallelism(boolean enable) { - maintainsParallelism = enable; - } - - /** - * Establishes local first-in-first-out scheduling mode for forked - * tasks that are never joined. This mode may be more appropriate - * than default locally stack-based mode in applications in which - * worker threads only process asynchronous tasks. This method is - * designed to be invoked only when the pool is quiescent, and - * typically only before any tasks are submitted. The effects of - * invocations at other times may be unpredictable. - * - * @param async if {@code true}, use locally FIFO scheduling - * @return the previous mode - * @see #getAsyncMode - */ - public boolean setAsyncMode(boolean async) { - boolean oldMode = locallyFifo; - locallyFifo = async; - ForkJoinWorkerThread[] ws = workers; - if (ws != null) { - for (int i = 0; i < ws.length; ++i) { - ForkJoinWorkerThread t = ws[i]; - if (t != null) - t.setAsyncMode(async); - } - } - return oldMode; + return workerCounts >>> TOTAL_COUNT_SHIFT; } /** @@ -851,7 +1515,6 @@ public class ForkJoinPool extends Abstra * scheduling mode for forked tasks that are never joined. * * @return {@code true} if this pool uses async mode - * @see #setAsyncMode */ public boolean getAsyncMode() { return locallyFifo; @@ -860,12 +1523,13 @@ public class ForkJoinPool extends Abstra /** * Returns an estimate of the number of worker threads that are * not blocked waiting to join tasks or for other managed - * synchronization. + * synchronization. This method may overestimate the + * number of running threads. * * @return the number of worker threads */ public int getRunningThreadCount() { - return runningCountOf(workerCounts); + return workerCounts & RUNNING_COUNT_MASK; } /** @@ -876,19 +1540,7 @@ public class ForkJoinPool extends Abstra * @return the number of active threads */ public int getActiveThreadCount() { - return activeCountOf(runControl); - } - - /** - * Returns an estimate of the number of threads that are currently - * idle waiting for tasks. This method may underestimate the - * number of idle threads. - * - * @return the number of idle threads - */ - final int getIdleThreadCount() { - int c = runningCountOf(workerCounts) - activeCountOf(runControl); - return (c <= 0) ? 0 : c; + return runState & ACTIVE_COUNT_MASK; } /** @@ -903,7 +1555,7 @@ public class ForkJoinPool extends Abstra * @return {@code true} if all threads are currently idle */ public boolean isQuiescent() { - return activeCountOf(runControl) == 0; + return (runState & ACTIVE_COUNT_MASK) == 0; } /** @@ -918,17 +1570,7 @@ public class ForkJoinPool extends Abstra * @return the number of steals */ public long getStealCount() { - return stealCount.get(); - } - - /** - * Accumulates steal count from a worker. - * Call only when worker known to be idle. - */ - private void updateStealCount(ForkJoinWorkerThread w) { - int sc = w.getAndClearStealCount(); - if (sc != 0) - stealCount.addAndGet(sc); + return stealCount; } /** @@ -944,19 +1586,18 @@ public class ForkJoinPool extends Abstra public long getQueuedTaskCount() { long count = 0; ForkJoinWorkerThread[] ws = workers; - if (ws != null) { - for (int i = 0; i < ws.length; ++i) { - ForkJoinWorkerThread t = ws[i]; - if (t != null) - count += t.getQueueSize(); - } + int nws = ws.length; + for (int i = 0; i < nws; ++i) { + ForkJoinWorkerThread w = ws[i]; + if (w != null) + count += w.getQueueSize(); } return count; } /** - * Returns an estimate of the number tasks submitted to this pool - * that have not yet begun executing. This method takes time + * Returns an estimate of the number of tasks submitted to this + * pool that have not yet begun executing. This method takes time * proportional to the number of submissions. * * @return the number of queued submissions @@ -990,8 +1631,8 @@ public class ForkJoinPool extends Abstra * Removes all available unexecuted submitted and forked tasks * from scheduling queues and adds them to the given collection, * without altering their execution status. These may include - * artificially generated or wrapped tasks. This method is designed - * to be invoked only when the pool is known to be + * artificially generated or wrapped tasks. This method is + * designed to be invoked only when the pool is known to be * quiescent. Invocations at other times may not remove all * tasks. A failure encountered while attempting to add elements * to collection {@code c} may result in elements being in @@ -1006,17 +1647,32 @@ public class ForkJoinPool extends Abstra protected int drainTasksTo(Collection> c) { int n = submissionQueue.drainTo(c); ForkJoinWorkerThread[] ws = workers; - if (ws != null) { - for (int i = 0; i < ws.length; ++i) { - ForkJoinWorkerThread w = ws[i]; - if (w != null) - n += w.drainTasksTo(c); - } + int nws = ws.length; + for (int i = 0; i < nws; ++i) { + ForkJoinWorkerThread w = ws[i]; + if (w != null) + n += w.drainTasksTo(c); } return n; } /** + * Returns count of total parks by existing workers. + * Used during development only since not meaningful to users. + */ + private int collectParkCount() { + int count = 0; + ForkJoinWorkerThread[] ws = workers; + int nws = ws.length; + for (int i = 0; i < nws; ++i) { + ForkJoinWorkerThread w = ws[i]; + if (w != null) + count += w.parkCount; + } + return count; + } + + /** * Returns a string identifying this pool, as well as its state, * including indications of run state, parallelism level, and * worker and task counts. @@ -1024,36 +1680,36 @@ public class ForkJoinPool extends Abstra * @return a string identifying this pool, as well as its state */ public String toString() { - int ps = parallelism; - int wc = workerCounts; - int rc = runControl; long st = getStealCount(); long qt = getQueuedTaskCount(); long qs = getQueuedSubmissionCount(); + int wc = workerCounts; + int tc = wc >>> TOTAL_COUNT_SHIFT; + int rc = wc & RUNNING_COUNT_MASK; + int pc = parallelism; + int rs = runState; + int ac = rs & ACTIVE_COUNT_MASK; + // int pk = collectParkCount(); return super.toString() + - "[" + runStateToString(runStateOf(rc)) + - ", parallelism = " + ps + - ", size = " + totalCountOf(wc) + - ", active = " + activeCountOf(rc) + - ", running = " + runningCountOf(wc) + + "[" + runLevelToString(rs) + + ", parallelism = " + pc + + ", size = " + tc + + ", active = " + ac + + ", running = " + rc + ", steals = " + st + ", tasks = " + qt + ", submissions = " + qs + + // ", parks = " + pk + "]"; } - private static String runStateToString(int rs) { - switch(rs) { - case RUNNING: return "Running"; - case SHUTDOWN: return "Shutting down"; - case TERMINATING: return "Terminating"; - case TERMINATED: return "Terminated"; - default: throw new Error("Unknown run state"); - } + private static String runLevelToString(int s) { + return ((s & TERMINATED) != 0 ? "Terminated" : + ((s & TERMINATING) != 0 ? "Terminating" : + ((s & SHUTDOWN) != 0 ? "Shutting down" : + "Running"))); } - // lifecycle control - /** * Initiates an orderly shutdown in which previously submitted * tasks are executed, but no new tasks will be accepted. @@ -1068,34 +1724,19 @@ public class ForkJoinPool extends Abstra */ public void shutdown() { checkPermission(); - transitionRunStateTo(SHUTDOWN); - if (canTerminateOnShutdown(runControl)) { - if (workers == null) { // shutting down before workers created - final ReentrantLock lock = this.workerLock; - lock.lock(); - try { - if (workers == null) { - terminate(); - transitionRunStateTo(TERMINATED); - termination.signalAll(); - } - } finally { - lock.unlock(); - } - } - terminateOnShutdown(); - } + advanceRunLevel(SHUTDOWN); + tryTerminate(false); } /** - * Attempts to stop all actively executing tasks, and cancels all - * waiting 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. Unlike some other executors, - * this method cancels rather than collects non-executed tasks - * upon termination, so always returns an empty list. However, you - * can use method {@link #drainTasksTo} before invoking this - * method to transfer unexecuted tasks to another collection. + * 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). * * @return an empty list * @throws SecurityException if a security manager exists and @@ -1105,7 +1746,7 @@ public class ForkJoinPool extends Abstra */ public List shutdownNow() { checkPermission(); - terminate(); + tryTerminate(true); return Collections.emptyList(); } @@ -1115,17 +1756,21 @@ public class ForkJoinPool extends Abstra * @return {@code true} if all tasks have completed following shut down */ public boolean isTerminated() { - return runStateOf(runControl) == TERMINATED; + return runState >= TERMINATED; } /** * Returns {@code true} if the process of termination has - * commenced but possibly not yet completed. + * 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, causing this executor not + * to properly terminate. * - * @return {@code true} if terminating + * @return {@code true} if terminating but not yet terminated */ public boolean isTerminating() { - return runStateOf(runControl) >= TERMINATING; + return (runState & (TERMINATING|TERMINATED)) == TERMINATING; } /** @@ -1134,7 +1779,7 @@ public class ForkJoinPool extends Abstra * @return {@code true} if this pool has been shut down */ public boolean isShutdown() { - return runStateOf(runControl) >= SHUTDOWN; + return runState >= SHUTDOWN; } /** @@ -1150,596 +1795,22 @@ public class ForkJoinPool extends Abstra */ public boolean awaitTermination(long timeout, TimeUnit unit) throws InterruptedException { - long nanos = unit.toNanos(timeout); - final ReentrantLock lock = this.workerLock; - lock.lock(); - try { - for (;;) { - if (isTerminated()) - return true; - if (nanos <= 0) - return false; - nanos = termination.awaitNanos(nanos); - } - } finally { - lock.unlock(); - } - } - - // Shutdown and termination support - - /** - * Callback from terminating worker. Nulls out the corresponding - * workers slot, and if terminating, tries to terminate; else - * tries to shrink workers array. - * - * @param w the worker - */ - final void workerTerminated(ForkJoinWorkerThread w) { - updateStealCount(w); - updateWorkerCount(-1); - final ReentrantLock lock = this.workerLock; - lock.lock(); try { - ForkJoinWorkerThread[] ws = workers; - if (ws != null) { - int idx = w.poolIndex; - if (idx >= 0 && idx < ws.length && ws[idx] == w) - ws[idx] = null; - if (totalCountOf(workerCounts) == 0) { - terminate(); // no-op if already terminating - transitionRunStateTo(TERMINATED); - termination.signalAll(); - } - else if (!isTerminating()) { - tryShrinkWorkerArray(); - tryResumeSpare(true); // allow replacement - } - } - } finally { - lock.unlock(); - } - signalIdleWorkers(); - } - - /** - * Initiates termination. - */ - private void terminate() { - if (transitionRunStateTo(TERMINATING)) { - stopAllWorkers(); - resumeAllSpares(); - signalIdleWorkers(); - cancelQueuedSubmissions(); - cancelQueuedWorkerTasks(); - interruptUnterminatedWorkers(); - signalIdleWorkers(); // resignal after interrupt - } - } - - /** - * Possibly terminates when on shutdown state. - */ - private void terminateOnShutdown() { - if (!hasQueuedSubmissions() && canTerminateOnShutdown(runControl)) - terminate(); - } - - /** - * Clears out and cancels submissions. - */ - private void cancelQueuedSubmissions() { - ForkJoinTask task; - while ((task = pollSubmission()) != null) - task.cancel(false); - } - - /** - * Cleans out worker queues. - */ - private void cancelQueuedWorkerTasks() { - final ReentrantLock lock = this.workerLock; - lock.lock(); - try { - ForkJoinWorkerThread[] ws = workers; - if (ws != null) { - for (int i = 0; i < ws.length; ++i) { - ForkJoinWorkerThread t = ws[i]; - if (t != null) - t.cancelTasks(); - } - } - } finally { - lock.unlock(); - } - } - - /** - * Sets each worker's status to terminating. Requires lock to avoid - * conflicts with add/remove. - */ - private void stopAllWorkers() { - final ReentrantLock lock = this.workerLock; - lock.lock(); - try { - ForkJoinWorkerThread[] ws = workers; - if (ws != null) { - for (int i = 0; i < ws.length; ++i) { - ForkJoinWorkerThread t = ws[i]; - if (t != null) - t.shutdownNow(); - } - } - } finally { - lock.unlock(); - } - } - - /** - * Interrupts all unterminated workers. This is not required for - * sake of internal control, but may help unstick user code during - * shutdown. - */ - private void interruptUnterminatedWorkers() { - final ReentrantLock lock = this.workerLock; - lock.lock(); - try { - ForkJoinWorkerThread[] ws = workers; - if (ws != null) { - for (int i = 0; i < ws.length; ++i) { - ForkJoinWorkerThread t = ws[i]; - if (t != null && !t.isTerminated()) { - try { - t.interrupt(); - } catch (SecurityException ignore) { - } - } - } - } - } finally { - lock.unlock(); - } - } - - - /* - * Nodes for event barrier to manage idle threads. Queue nodes - * are basic Treiber stack nodes, also used for spare stack. - * - * The event barrier has an event count and a wait queue (actually - * a Treiber stack). Workers are enabled to look for work when - * the eventCount is incremented. If they fail to find work, they - * may wait for next count. Upon release, threads help others wake - * up. - * - * Synchronization events occur only in enough contexts to - * maintain overall liveness: - * - * - Submission of a new task to the pool - * - Resizes or other changes to the workers array - * - pool termination - * - A worker pushing a task on an empty queue - * - * The case of pushing a task occurs often enough, and is heavy - * enough compared to simple stack pushes, to require special - * handling: Method signalWork returns without advancing count if - * the queue appears to be empty. This would ordinarily result in - * races causing some queued waiters not to be woken up. To avoid - * this, the first worker enqueued in method sync (see - * syncIsReleasable) rescans for tasks after being enqueued, and - * helps signal if any are found. This works well because the - * worker has nothing better to do, and so might as well help - * alleviate the overhead and contention on the threads actually - * doing work. Also, since event counts increments on task - * availability exist to maintain liveness (rather than to force - * refreshes etc), it is OK for callers to exit early if - * contending with another signaller. - */ - static final class WaitQueueNode { - WaitQueueNode next; // only written before enqueued - volatile ForkJoinWorkerThread thread; // nulled to cancel wait - final long count; // unused for spare stack - - WaitQueueNode(long c, ForkJoinWorkerThread w) { - count = c; - thread = w; - } - - /** - * Wakes up waiter, returning false if known to already - */ - boolean signal() { - ForkJoinWorkerThread t = thread; - if (t == null) - return false; - thread = null; - LockSupport.unpark(t); - return true; - } - - /** - * Awaits release on sync. - */ - void awaitSyncRelease(ForkJoinPool p) { - while (thread != null && !p.syncIsReleasable(this)) - LockSupport.park(this); - } - - /** - * Awaits resumption as spare. - */ - void awaitSpareRelease() { - while (thread != null) { - if (!Thread.interrupted()) - LockSupport.park(this); - } - } - } - - /** - * Ensures that no thread is waiting for count to advance from the - * current value of eventCount read on entry to this method, by - * releasing waiting threads if necessary. - * - * @return the count - */ - final long ensureSync() { - long c = eventCount; - WaitQueueNode q; - while ((q = syncStack) != null && q.count < c) { - if (casBarrierStack(q, null)) { - do { - q.signal(); - } while ((q = q.next) != null); - break; - } - } - return c; - } - - /** - * Increments event count and releases waiting threads. - */ - private void signalIdleWorkers() { - long c; - do {} while (!casEventCount(c = eventCount, c+1)); - ensureSync(); - } - - /** - * Signals threads waiting to poll a task. Because method sync - * rechecks availability, it is OK to only proceed if queue - * appears to be non-empty, and OK to skip under contention to - * increment count (since some other thread succeeded). - */ - final void signalWork() { - long c; - WaitQueueNode q; - if (syncStack != null && - casEventCount(c = eventCount, c+1) && - (((q = syncStack) != null && q.count <= c) && - (!casBarrierStack(q, q.next) || !q.signal()))) - ensureSync(); - } - - /** - * Waits until event count advances from last value held by - * caller, or if excess threads, caller is resumed as spare, or - * caller or pool is terminating. Updates caller's event on exit. - * - * @param w the calling worker thread - */ - final void sync(ForkJoinWorkerThread w) { - updateStealCount(w); // Transfer w's count while it is idle - - while (!w.isShutdown() && !isTerminating() && !suspendIfSpare(w)) { - long prev = w.lastEventCount; - WaitQueueNode node = null; - WaitQueueNode h; - while (eventCount == prev && - ((h = syncStack) == null || h.count == prev)) { - if (node == null) - node = new WaitQueueNode(prev, w); - if (casBarrierStack(node.next = h, node)) { - node.awaitSyncRelease(this); - break; - } - } - long ec = ensureSync(); - if (ec != prev) { - w.lastEventCount = ec; - break; - } - } - } - - /** - * Returns {@code true} if worker waiting on sync can proceed: - * - on signal (thread == null) - * - on event count advance (winning race to notify vs signaller) - * - on interrupt - * - if the first queued node, we find work available - * If node was not signalled and event count not advanced on exit, - * then we also help advance event count. - * - * @return {@code true} if node can be released - */ - final boolean syncIsReleasable(WaitQueueNode node) { - long prev = node.count; - if (!Thread.interrupted() && node.thread != null && - (node.next != null || - !ForkJoinWorkerThread.hasQueuedTasks(workers)) && - eventCount == prev) + return termination.awaitAdvanceInterruptibly(0, timeout, unit) > 0; + } catch(TimeoutException ex) { return false; - if (node.thread != null) { - node.thread = null; - long ec = eventCount; - if (prev <= ec) // help signal - casEventCount(ec, ec+1); - } - return true; - } - - /** - * Returns {@code true} if a new sync event occurred since last - * call to sync or this method, if so, updating caller's count. - */ - final boolean hasNewSyncEvent(ForkJoinWorkerThread w) { - long lc = w.lastEventCount; - long ec = ensureSync(); - if (ec == lc) - return false; - w.lastEventCount = ec; - return true; - } - - // Parallelism maintenance - - /** - * Decrements running count; if too low, adds spare. - * - * Conceptually, all we need to do here is add or resume a - * spare thread when one is about to block (and remove or - * suspend it later when unblocked -- see suspendIfSpare). - * However, implementing this idea requires coping with - * several problems: we have imperfect information about the - * states of threads. Some count updates can and usually do - * lag run state changes, despite arrangements to keep them - * accurate (for example, when possible, updating counts - * before signalling or resuming), especially when running on - * dynamic JVMs that don't optimize the infrequent paths that - * update counts. Generating too many threads can make these - * problems become worse, because excess threads are more - * likely to be context-switched with others, slowing them all - * down, especially if there is no work available, so all are - * busy scanning or idling. Also, excess spare threads can - * only be suspended or removed when they are idle, not - * immediately when they aren't needed. So adding threads will - * raise parallelism level for longer than necessary. Also, - * FJ applications often encounter highly transient peaks when - * many threads are blocked joining, but for less time than it - * takes to create or resume spares. - * - * @param joinMe if non-null, return early if done - * @param maintainParallelism if true, try to stay within - * target counts, else create only to avoid starvation - * @return true if joinMe known to be done - */ - final boolean preJoin(ForkJoinTask joinMe, - boolean maintainParallelism) { - maintainParallelism &= maintainsParallelism; // overrride - boolean dec = false; // true when running count decremented - while (spareStack == null || !tryResumeSpare(dec)) { - int counts = workerCounts; - if (dec || (dec = casWorkerCounts(counts, --counts))) { - // CAS cheat - if (!needSpare(counts, maintainParallelism)) - break; - if (joinMe.status < 0) - return true; - if (tryAddSpare(counts)) - break; - } - } - return false; - } - - /** - * Same idea as preJoin - */ - final boolean preBlock(ManagedBlocker blocker, - boolean maintainParallelism) { - maintainParallelism &= maintainsParallelism; - boolean dec = false; - while (spareStack == null || !tryResumeSpare(dec)) { - int counts = workerCounts; - if (dec || (dec = casWorkerCounts(counts, --counts))) { - if (!needSpare(counts, maintainParallelism)) - break; - if (blocker.isReleasable()) - return true; - if (tryAddSpare(counts)) - break; - } - } - return false; - } - - /** - * Returns {@code true} if a spare thread appears to be needed. - * If maintaining parallelism, returns true when the deficit in - * running threads is more than the surplus of total threads, and - * there is apparently some work to do. This self-limiting rule - * means that the more threads that have already been added, the - * less parallelism we will tolerate before adding another. - * - * @param counts current worker counts - * @param maintainParallelism try to maintain parallelism - */ - private boolean needSpare(int counts, boolean maintainParallelism) { - int ps = parallelism; - int rc = runningCountOf(counts); - int tc = totalCountOf(counts); - int runningDeficit = ps - rc; - int totalSurplus = tc - ps; - return (tc < maxPoolSize && - (rc == 0 || totalSurplus < 0 || - (maintainParallelism && - runningDeficit > totalSurplus && - ForkJoinWorkerThread.hasQueuedTasks(workers)))); - } - - /** - * Adds a spare worker if lock available and no more than the - * expected numbers of threads exist. - * - * @return true if successful - */ - private boolean tryAddSpare(int expectedCounts) { - final ReentrantLock lock = this.workerLock; - int expectedRunning = runningCountOf(expectedCounts); - int expectedTotal = totalCountOf(expectedCounts); - boolean success = false; - boolean locked = false; - // confirm counts while locking; CAS after obtaining lock - try { - for (;;) { - int s = workerCounts; - int tc = totalCountOf(s); - int rc = runningCountOf(s); - if (rc > expectedRunning || tc > expectedTotal) - break; - if (!locked && !(locked = lock.tryLock())) - break; - if (casWorkerCounts(s, workerCountsFor(tc+1, rc+1))) { - createAndStartSpare(tc); - success = true; - break; - } - } - } finally { - if (locked) - lock.unlock(); - } - return success; - } - - /** - * Adds the kth spare worker. On entry, pool counts are already - * adjusted to reflect addition. - */ - private void createAndStartSpare(int k) { - ForkJoinWorkerThread w = null; - ForkJoinWorkerThread[] ws = ensureWorkerArrayCapacity(k + 1); - int len = ws.length; - // Probably, we can place at slot k. If not, find empty slot - if (k < len && ws[k] != null) { - for (k = 0; k < len && ws[k] != null; ++k) - ; - } - if (k < len && !isTerminating() && (w = createWorker(k)) != null) { - ws[k] = w; - w.start(); - } - else - updateWorkerCount(-1); // adjust on failure - signalIdleWorkers(); - } - - /** - * Suspends calling thread w if there are excess threads. Called - * only from sync. Spares are enqueued in a Treiber stack using - * the same WaitQueueNodes as barriers. They are resumed mainly - * in preJoin, but are also woken on pool events that require all - * threads to check run state. - * - * @param w the caller - */ - private boolean suspendIfSpare(ForkJoinWorkerThread w) { - WaitQueueNode node = null; - int s; - while (parallelism < runningCountOf(s = workerCounts)) { - if (node == null) - node = new WaitQueueNode(0, w); - if (casWorkerCounts(s, s-1)) { // representation-dependent - // push onto stack - do {} while (!casSpareStack(node.next = spareStack, node)); - // block until released by resumeSpare - node.awaitSpareRelease(); - return true; - } - } - return false; - } - - /** - * Tries to pop and resume a spare thread. - * - * @param updateCount if true, increment running count on success - * @return true if successful - */ - private boolean tryResumeSpare(boolean updateCount) { - WaitQueueNode q; - while ((q = spareStack) != null) { - if (casSpareStack(q, q.next)) { - if (updateCount) - updateRunningCount(1); - q.signal(); - return true; - } - } - return false; - } - - /** - * Pops and resumes all spare threads. Same idea as ensureSync. - * - * @return true if any spares released - */ - private boolean resumeAllSpares() { - WaitQueueNode q; - while ( (q = spareStack) != null) { - if (casSpareStack(q, null)) { - do { - updateRunningCount(1); - q.signal(); - } while ((q = q.next) != null); - return true; - } - } - return false; - } - - /** - * Pops and shuts down excessive spare threads. Call only while - * holding lock. This is not guaranteed to eliminate all excess - * threads, only those suspended as spares, which are the ones - * unlikely to be needed in the future. - */ - private void trimSpares() { - int surplus = totalCountOf(workerCounts) - parallelism; - WaitQueueNode q; - while (surplus > 0 && (q = spareStack) != null) { - if (casSpareStack(q, null)) { - do { - updateRunningCount(1); - ForkJoinWorkerThread w = q.thread; - if (w != null && surplus > 0 && - runningCountOf(workerCounts) > 0 && w.shutdown()) - --surplus; - q.signal(); - } while ((q = q.next) != null); - } } } /** * Interface for extending managed parallelism for tasks running - * in ForkJoinPools. A ManagedBlocker provides two methods. + * in {@link ForkJoinPool}s. + * + *

A {@code ManagedBlocker} provides two methods. * Method {@code isReleasable} must return {@code true} if * blocking is not necessary. Method {@code block} blocks the * current thread if necessary (perhaps internally invoking - * {@code isReleasable} before actually blocking.). + * {@code isReleasable} before actually blocking). * *

For example, here is a ManagedBlocker based on a * ReentrantLock: @@ -1778,56 +1849,34 @@ public class ForkJoinPool extends Abstra /** * Blocks in accord with the given blocker. If the current thread - * is a ForkJoinWorkerThread, this method possibly arranges for a - * spare thread to be activated if necessary to ensure parallelism - * while the current thread is blocked. If - * {@code maintainParallelism} is {@code true} and the pool supports - * it ({@link #getMaintainsParallelism}), this method attempts to - * maintain the pool's nominal parallelism. Otherwise it activates - * a thread only if necessary to avoid complete starvation. This - * option may be preferable when blockages use timeouts, or are - * almost always brief. + * is a {@link ForkJoinWorkerThread}, this method possibly + * arranges for a spare thread to be activated if necessary to + * ensure sufficient parallelism while the current thread is blocked. * - *

If the caller is not a ForkJoinTask, this method is behaviorally - * equivalent to + *

If the caller is not a {@link ForkJoinTask}, this method is + * behaviorally equivalent to * 

 {@code
      * while (!blocker.isReleasable())
      *   if (blocker.block())
      *     return;
      * }
- * If the caller is a ForkJoinTask, then the pool may first - * be expanded to ensure parallelism, and later adjusted. + * + * If the caller is a {@code ForkJoinTask}, then the pool may + * first be expanded to ensure parallelism, and later adjusted. * * @param blocker the blocker - * @param maintainParallelism if {@code true} and supported by - * this pool, attempt to maintain the pool's nominal parallelism; - * otherwise activate a thread only if necessary to avoid - * complete starvation. * @throws InterruptedException if blocker.block did so */ - public static void managedBlock(ManagedBlocker blocker, - boolean maintainParallelism) + public static void managedBlock(ManagedBlocker blocker) throws InterruptedException { Thread t = Thread.currentThread(); - ForkJoinPool pool = ((t instanceof ForkJoinWorkerThread) ? - ((ForkJoinWorkerThread) t).pool : null); - if (!blocker.isReleasable()) { - try { - if (pool == null || - !pool.preBlock(blocker, maintainParallelism)) - awaitBlocker(blocker); - } finally { - if (pool != null) - pool.updateRunningCount(1); - } + if (t instanceof ForkJoinWorkerThread) + ((ForkJoinWorkerThread) t).pool.awaitBlocker(blocker); + else { + do {} while (!blocker.isReleasable() && !blocker.block()); } } - private static void awaitBlocker(ManagedBlocker blocker) - throws InterruptedException { - do {} while (!blocker.isReleasable() && !blocker.block()); - } - // AbstractExecutorService overrides. These rely on undocumented // fact that ForkJoinTask.adapt returns ForkJoinTasks that also // implement RunnableFuture. @@ -1843,32 +1892,16 @@ public class ForkJoinPool extends Abstra // Unsafe mechanics private static final sun.misc.Unsafe UNSAFE = getUnsafe(); - private static final long eventCountOffset = - objectFieldOffset("eventCount", ForkJoinPool.class); private static final long workerCountsOffset = objectFieldOffset("workerCounts", ForkJoinPool.class); - private static final long runControlOffset = - objectFieldOffset("runControl", ForkJoinPool.class); - private static final long syncStackOffset = - objectFieldOffset("syncStack",ForkJoinPool.class); - private static final long spareStackOffset = - objectFieldOffset("spareStack", ForkJoinPool.class); - - private boolean casEventCount(long cmp, long val) { - return UNSAFE.compareAndSwapLong(this, eventCountOffset, cmp, val); - } - private boolean casWorkerCounts(int cmp, int val) { - return UNSAFE.compareAndSwapInt(this, workerCountsOffset, cmp, val); - } - private boolean casRunControl(int cmp, int val) { - return UNSAFE.compareAndSwapInt(this, runControlOffset, cmp, val); - } - private boolean casSpareStack(WaitQueueNode cmp, WaitQueueNode val) { - return UNSAFE.compareAndSwapObject(this, spareStackOffset, cmp, val); - } - private boolean casBarrierStack(WaitQueueNode cmp, WaitQueueNode val) { - return UNSAFE.compareAndSwapObject(this, syncStackOffset, cmp, val); - } + private static final long runStateOffset = + objectFieldOffset("runState", ForkJoinPool.class); + private static final long eventCountOffset = + objectFieldOffset("eventCount", ForkJoinPool.class); + private static final long eventWaitersOffset = + objectFieldOffset("eventWaiters",ForkJoinPool.class); + private static final long stealCountOffset = + objectFieldOffset("stealCount",ForkJoinPool.class); private static long objectFieldOffset(String field, Class klazz) { try {