--- jsr166/src/jsr166y/ForkJoinPool.java 2009/07/24 18:57:56 1.19 +++ jsr166/src/jsr166y/ForkJoinPool.java 2010/07/07 19:52:31 1.57 @@ -5,57 +5,115 @@ */ package jsr166y; -import java.util.*; + import java.util.concurrent.*; -import java.util.concurrent.locks.*; -import java.util.concurrent.atomic.*; -import sun.misc.Unsafe; -import java.lang.reflect.*; + +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collection; +import java.util.Collections; +import java.util.List; +import java.util.concurrent.locks.LockSupport; +import java.util.concurrent.locks.ReentrantLock; +import java.util.concurrent.atomic.AtomicInteger; +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. + * An {@link ExecutorService} for running {@link ForkJoinTask}s. + * 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 - * {@code setAsyncMode}, a ForkJoinPools 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 {@code ManagedBlocker} interface enables extension of - * the kinds of synchronization accommodated. The target parallelism - * level may also be changed dynamically ({@code setParallelism}) - * and thread construction can be limited using methods - * {@code setMaximumPoolSize} and/or - * {@code 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 - * {@code getStealCount}) that are intended to aid in developing, + * {@link #getStealCount}) that are intended to aid in developing, * tuning, and monitoring fork/join applications. Also, method - * {@code toString} returns indications of pool state in a + * {@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. To avoid inadvertant + * cyclic task dependencies and to improve performance, task + * submissions to the current pool by an ongoing fork/join + * computations may be implicitly translated to the corresponding + * ForkJoinTask forms. + * + * + * + * + * + * + * + * + * + * + * + * + * + * + * + * + * + * + * + * + * + * + *
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. * * @since 1.7 * @author Doug Lea @@ -63,28 +121,218 @@ import java.lang.reflect.*; 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.) + * + * The main throughput advantages of work-stealing stem from + * decentralized control -- workers mostly steal tasks from each + * other. We do not want to negate this by creating bottlenecks + * implementing the management responsibilities of this class. So + * we use a collection of techniques that avoid, reduce, or cope + * well with contention. These entail several instances of + * bit-packing into CASable fields to maintain only the minimally + * required atomicity. To enable such packing, we restrict maximum + * parallelism to (1<<15)-1 (enabling twice this to fit into a 16 + * bit field), which is far in excess of normal operating range. + * Even though updates to some of these bookkeeping fields do + * sometimes contend with each other, they don't normally + * cache-contend with updates to others enough to warrant memory + * padding or isolation. So they are all held as fields of + * ForkJoinPool objects. The main capabilities are as follows: + * + * 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. To support these decisions, + * updates to spare counts must be prospective (not + * retrospective). For example, the running count is decremented + * before blocking by a thread about to block as a spare, but + * incremented by the thread about to unblock it. Updates upon + * resumption ofr threads blocking in awaitJoin or awaitBlocker + * cannot usually be prospective, so the running count is in + * general an upper bound of the number of productively running + * threads Updates to the workerCounts field sometimes transiently + * encounter a fair amount of contention when join dependencies + * are such that many threads block or unblock at about the same + * time. We alleviate this by sometimes performing an alternative + * action on contention like releasing waiters or locating spares. + * + * 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 (see below). Usually, extra threads are needed + * for only very short periods, yet join dependencies are such + * that we sometimes need them in bursts. Rather than create new + * threads each time this happens, we suspend no-longer-needed + * extra ones as "spares". For most purposes, we don't distinguish + * "extra" spare threads from normal "core" threads: On each call + * to preStep (the only point at which we can do this) a worker + * checks to see if there are now too many running workers, and if + * so, suspends itself. Methods awaitJoin and awaitBlocker look + * for suspended threads to resume before considering creating a + * new replacement. We don't need a special data structure to + * maintain spares; simply scanning the workers array looking for + * worker.isSuspended() is fine because the calling thread is + * otherwise not doing anything useful anyway; we are at least as + * happy if after locating a spare, the caller doesn't actually + * block because the join is ready before we try to adjust and + * compensate. Note that this is intrinsically racy. One thread + * may become a spare at about the same time as another is + * needlessly being created. We counteract this and related slop + * in part by requiring resumed spares to immediately recheck (in + * preStep) to see whether they they should re-suspend. The only + * effective difference between "extra" and "core" threads is that + * we allow the "extra" ones to time out and die if they are not + * resumed within a keep-alive interval of a few seconds. This is + * implemented mainly within ForkJoinWorkerThread, but requires + * some coordination (isTrimmed() -- meaning killed while + * suspended) to correctly maintain pool counts. + * + * 6. Deciding when to create new workers. The main dynamic + * control in this class is deciding when to create extra threads, + * in methods awaitJoin and awaitBlocker. We always need to create + * one when the number of running threads becomes zero. But + * because blocked joins are typically dependent, we don't + * necessarily need or want one-to-one replacement. Instead, we + * use a combination of heuristics that adds threads only when the + * pool appears to be approaching starvation. These effectively + * reduce churn at the price of systematically undershooting + * target parallelism when many threads are blocked. However, + * biasing toward undeshooting partially compensates for the above + * mechanics to suspend extra threads, that normally lead to + * overshoot because we can only suspend workers in-between + * top-level actions. It also better copes with the fact that some + * of the methods in this class tend to never become compiled (but + * are interpreted), so some components of the entire set of + * controls might execute many times faster than others. And + * 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 a few + * other coding oddities that help some methods perform reasonably + * even when interpreted (not compiled). + * + * 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); } @@ -93,14 +341,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); } } @@ -136,403 +380,830 @@ 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; - /** - * Head of Treiber stack for barrier sync. See below for explanation. - */ - private volatile WaitQueueNode syncStack; + private static final int EVENT_COUNT_SHIFT = 32; + private static final long WAITER_INDEX_MASK = (1L << EVENT_COUNT_SHIFT)-1L; /** - * The count for event barrier + * 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 long eventCount; + private volatile int eventCount; /** - * Pool number, just for assigning useful names to worker threads - */ - private final int poolNumber; + * 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 /** - * The maximum allowed pool size + * 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 volatile int maxPoolSize; + 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 desired parallelism level, updated only under workerLock. + * The target parallelism level. + * Accessed directly by ForkJoinWorkerThreads. */ - private volatile int parallelism; + final int parallelism; /** * True if use local fifo, not default lifo, for local polling + * Read by, and replicated by ForkJoinWorkerThreads */ - private volatile boolean locallyFifo; + final boolean locallyFifo; /** - * 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. + * The uncaught exception handler used when any worker abruptly + * terminates. */ - private volatile int workerCounts; - - 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; } + private final Thread.UncaughtExceptionHandler ueh; /** - * 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 + * Pool number, just for assigning useful names to worker threads */ - final void updateRunningCount(int delta) { - int s; - do {} while (!casWorkerCounts(s = workerCounts, s + delta)); - } + private final int poolNumber; + + // utilities for updating fields /** - * 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 + * Increments running count. Also used by ForkJoinTask. */ - 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 void incrementRunningCount() { + int c; + do {} while (!UNSAFE.compareAndSwapInt(this, workerCountsOffset, + c = workerCounts, + c + 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 decrement running count unless already zero */ - 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 tryDecrementRunningCount() { + int wc = workerCounts; + if ((wc & RUNNING_COUNT_MASK) == 0) + return false; + return UNSAFE.compareAndSwapInt(this, workerCountsOffset, + wc, 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 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 && + 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. - * - * @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")} + * Create enough total workers to establish target parallelism, + * giving up if terminating or addWorker fails */ - public ForkJoinPool(ForkJoinWorkerThreadFactory factory) { - this(Runtime.getRuntime().availableProcessors(), factory); + private void ensureEnoughTotalWorkers() { + int wc; + while (((wc = workerCounts) >>> TOTAL_COUNT_SHIFT) < parallelism && + runState < TERMINATING) { + if ((UNSAFE.compareAndSwapInt(this, workerCountsOffset, + wc, wc + (ONE_RUNNING|ONE_TOTAL)) && + addWorker() == null)) + break; + } } /** - * 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)) + ensureEnoughTotalWorkers(); } + // 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. */ - 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; + int id; + while ((id = (int)((top = eventWaiters) & WAITER_INDEX_MASK)) > 0 && + (int)(top >>> EVENT_COUNT_SHIFT) != eventCount) { + ForkJoinWorkerThread[] ws = workers; + ForkJoinWorkerThread w; + if (ws.length >= id && (w = ws[id - 1]) != null && + UNSAFE.compareAndSwapLong(this, eventWaitersOffset, + top, w.nextWaiter)) + LockSupport.unpark(w); } - 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. - * - * @return the array + * Advances eventCount and releases waiters until interference by + * other releasing threads is detected. */ - 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; + final void signalWork() { + // EventCount CAS failures are OK -- any change in count suffices. + int ec; + UNSAFE.compareAndSwapInt(this, eventCountOffset, ec=eventCount, ec+1); + outer:for (;;) { + long top = eventWaiters; + ec = eventCount; + for (;;) { + ForkJoinWorkerThread[] ws; ForkJoinWorkerThread w; + int id = (int)(top & WAITER_INDEX_MASK); + if (id <= 0 || (int)(top >>> EVENT_COUNT_SHIFT) == ec) + return; + if ((ws = workers).length < id || (w = ws[id - 1]) == null || + !UNSAFE.compareAndSwapLong(this, eventWaitersOffset, + top, top = w.nextWaiter)) + continue outer; // possibly stale; reread + LockSupport.unpark(w); + if (top != eventWaiters) // let someone else take over + return; + } + } } /** - * Tries to shrink workers into smaller array after one or more terminate. + * If worker is inactive, blocks until terminating or event count + * advances from last value held by worker; in any case helps + * release others. + * + * @param w the calling worker thread */ - 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); + private void eventSync(ForkJoinWorkerThread w) { + if (!w.active) { + int prev = w.lastEventCount; + long nextTop = (((long)prev << EVENT_COUNT_SHIFT) | + ((long)(w.poolIndex + 1))); + long top; + while ((runState < SHUTDOWN || !tryTerminate(false)) && + (((int)(top = eventWaiters) & WAITER_INDEX_MASK) == 0 || + (int)(top >>> EVENT_COUNT_SHIFT) == prev) && + eventCount == prev) { + if (UNSAFE.compareAndSwapLong(this, eventWaitersOffset, + w.nextWaiter = top, nextTop)) { + accumulateStealCount(w); // transfer steals while idle + Thread.interrupted(); // clear/ignore interrupt + while (eventCount == prev) + w.doPark(); + break; + } + } + w.lastEventCount = eventCount; } + releaseWaiters(); } /** - * Initializes workers if necessary. + * Callback from workers invoked upon each top-level action (i.e., + * stealing a task or taking a submission and running + * it). Performs one or both of the following: + * + * * If the worker cannot find work, updates its active status to + * inactive and updates activeCount unless there is contention, in + * which case it may try again (either in this or a subsequent + * call). Additionally, awaits the next task event and/or helps + * wake up other releasable waiters. + * + * * If there are too many running threads, suspends this worker + * (first forcing inactivation if necessary). If it is not + * resumed before a keepAlive elapses, the worker may be "trimmed" + * -- killed while suspended within suspendAsSpare. Otherwise, + * upon resume it rechecks to make sure that it is still needed. + * + * @param w the worker + * @param worked false if the worker scanned for work but didn't + * find any (in which case it may block waiting for work). */ - 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); - if (w != null) { - ws[i] = w; - w.start(); - updateWorkerCount(1); - } + final void preStep(ForkJoinWorkerThread w, boolean worked) { + boolean active = w.active; + boolean inactivate = !worked & active; + for (;;) { + if (inactivate) { + int rs = runState; + if (UNSAFE.compareAndSwapInt(this, runStateOffset, + rs, rs - ONE_ACTIVE)) + inactivate = active = w.active = false; + } + int wc = workerCounts; + if ((wc & RUNNING_COUNT_MASK) <= parallelism) { + if (!worked) + eventSync(w); + return; + } + if (!(inactivate |= active) && // must inactivate to suspend + UNSAFE.compareAndSwapInt(this, workerCountsOffset, + wc, wc - ONE_RUNNING) && + !w.suspendAsSpare()) // false if trimmed + return; + } + } + + /** + * Tries to decrement running count, and if so, possibly creates + * or resumes compensating threads before blocking on task joinMe. + * This code is sprawled out with manual inlining to evade some + * JIT oddities. + * + * @param joinMe the task to join + * @return task status on exit + */ + final int tryAwaitJoin(ForkJoinTask joinMe) { + int cw = workerCounts; // read now to spoil CAS if counts change as ... + releaseWaiters(); // ... a byproduct of releaseWaiters + int stat = joinMe.status; + if (stat >= 0 && // inline variant of tryDecrementRunningCount + (cw & RUNNING_COUNT_MASK) > 0 && + UNSAFE.compareAndSwapInt(this, workerCountsOffset, + cw, cw - ONE_RUNNING)) { + int pc = parallelism; + int scans = 0; // to require confirming passes to add threads + outer: while ((workerCounts & RUNNING_COUNT_MASK) < pc) { + if ((stat = joinMe.status) < 0) + break; + ForkJoinWorkerThread spare = null; + ForkJoinWorkerThread[] ws = workers; + int nws = ws.length; + for (int i = 0; i < nws; ++i) { + ForkJoinWorkerThread w = ws[i]; + if (w != null && w.isSuspended()) { + spare = w; + break; } } - } finally { - lock.unlock(); + if ((stat = joinMe.status) < 0) // recheck to narrow race + break; + int wc = workerCounts; + int rc = wc & RUNNING_COUNT_MASK; + if (rc >= pc) + break; + if (spare != null) { + if (spare.tryUnsuspend()) { + int c; // inline incrementRunningCount + do {} while (!UNSAFE.compareAndSwapInt + (this, workerCountsOffset, + c = workerCounts, c + ONE_RUNNING)); + LockSupport.unpark(spare); + break; + } + continue; + } + int tc = wc >>> TOTAL_COUNT_SHIFT; + int sc = tc - pc; + if (rc > 0) { + int p = pc; + int s = sc; + while (s-- >= 0) { // try keeping 3/4 live + if (rc > (p -= (p >>> 2) + 1)) + break outer; + } + } + if (scans++ > sc && tc < MAX_THREADS && + UNSAFE.compareAndSwapInt(this, workerCountsOffset, wc, + wc + (ONE_RUNNING|ONE_TOTAL))) { + addWorker(); + break; + } } + if (stat >= 0) + stat = joinMe.internalAwaitDone(); + int c; // inline incrementRunningCount + do {} while (!UNSAFE.compareAndSwapInt + (this, workerCountsOffset, + c = workerCounts, c + ONE_RUNNING)); } + return stat; } /** - * Worker creation and startup for threads added via setParallelism. + * Same idea as (and mostly pasted from) tryAwaitJoin, but + * self-contained */ - 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) + final void awaitBlocker(ManagedBlocker blocker) + throws InterruptedException { + for (;;) { + if (blocker.isReleasable()) + return; + int cw = workerCounts; + releaseWaiters(); + if ((cw & RUNNING_COUNT_MASK) > 0 && + UNSAFE.compareAndSwapInt(this, workerCountsOffset, + cw, cw - ONE_RUNNING)) break; - if (casWorkerCounts (s, workerCountsFor(tc+1, rc+1))) { - ForkJoinWorkerThread w = createWorker(k); - if (w != null) { - ws[k++] = w; - w.start(); + } + boolean done = false; + int pc = parallelism; + int scans = 0; + outer: while ((workerCounts & RUNNING_COUNT_MASK) < pc) { + if (done = blocker.isReleasable()) + break; + ForkJoinWorkerThread spare = null; + ForkJoinWorkerThread[] ws = workers; + int nws = ws.length; + for (int i = 0; i < nws; ++i) { + ForkJoinWorkerThread w = ws[i]; + if (w != null && w.isSuspended()) { + spare = w; + break; } - else { - updateWorkerCount(-1); // back out on failed creation + } + if (done = blocker.isReleasable()) + break; + int wc = workerCounts; + int rc = wc & RUNNING_COUNT_MASK; + if (rc >= pc) + break; + if (spare != null) { + if (spare.tryUnsuspend()) { + int c; + do {} while (!UNSAFE.compareAndSwapInt + (this, workerCountsOffset, + c = workerCounts, c + ONE_RUNNING)); + LockSupport.unpark(spare); break; } + continue; + } + int tc = wc >>> TOTAL_COUNT_SHIFT; + int sc = tc - pc; + if (rc > 0) { + int p = pc; + int s = sc; + while (s-- >= 0) { + if (rc > (p -= (p >>> 2) + 1)) + break outer; + } + } + if (scans++ > sc && tc < MAX_THREADS && + UNSAFE.compareAndSwapInt(this, workerCountsOffset, wc, + wc + (ONE_RUNNING|ONE_TOTAL))) { + addWorker(); + break; } } + try { + if (!done) + do {} while (!blocker.isReleasable() && + !blocker.block()); + } finally { + int c; + do {} while (!UNSAFE.compareAndSwapInt + (this, workerCountsOffset, + c = workerCounts, c + ONE_RUNNING)); + } + } + + /** + * Possibly initiates and/or completes termination. + * + * @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 targeted 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 @@ -541,21 +1212,34 @@ public class ForkJoinPool extends Abstra * Common code for execute, invoke and submit */ private void doSubmit(ForkJoinTask task) { - if (isShutdown()) + if (task == null) + throw new NullPointerException(); + if (runState >= SHUTDOWN) throw new RejectedExecutionException(); - if (workers == null) - ensureWorkerInitialization(); - submissionQueue.offer(task); - signalIdleWorkers(); + // Convert submissions to current pool into forks + Thread t = Thread.currentThread(); + ForkJoinWorkerThread w; + if ((t instanceof ForkJoinWorkerThread) && + (w = (ForkJoinWorkerThread) t).pool == this) + w.pushTask(task); + else { + submissionQueue.offer(task); + signalEvent(); + ensureEnoughTotalWorkers(); + } } /** * 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); @@ -564,100 +1248,103 @@ 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) { - doSubmit(new AdaptedRunnable(task, null)); + ForkJoinTask job; + if (task instanceof ForkJoinTask) // avoid re-wrap + job = (ForkJoinTask) task; + else + job = ForkJoinTask.adapt(task, null); + 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 = new AdaptedCallable(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 = new AdaptedRunnable(task, 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 = new AdaptedRunnable(task, null); + ForkJoinTask job; + if (task instanceof ForkJoinTask) // avoid re-wrap + job = (ForkJoinTask) task; + else + job = ForkJoinTask.adapt(task, null); doSubmit(job); return job; } /** - * Adaptor for Runnables. This implements RunnableFuture - * to be compliant with AbstractExecutorService constraints. + * @throws NullPointerException {@inheritDoc} + * @throws RejectedExecutionException {@inheritDoc} */ - static final class AdaptedRunnable extends ForkJoinTask - implements RunnableFuture { - final Runnable runnable; - final T resultOnCompletion; - T result; - AdaptedRunnable(Runnable runnable, T result) { - if (runnable == null) throw new NullPointerException(); - this.runnable = runnable; - this.resultOnCompletion = result; - } - public T getRawResult() { return result; } - public void setRawResult(T v) { result = v; } - public boolean exec() { - runnable.run(); - result = resultOnCompletion; - return true; - } - public void run() { invoke(); } - private static final long serialVersionUID = 5232453952276885070L; - } - - /** - * Adaptor for Callables - */ - static final class AdaptedCallable extends ForkJoinTask - implements RunnableFuture { - final Callable callable; - T result; - AdaptedCallable(Callable callable) { - if (callable == null) throw new NullPointerException(); - this.callable = callable; - } - public T getRawResult() { return result; } - public void setRawResult(T v) { result = v; } - public boolean exec() { - try { - result = callable.call(); - return true; - } catch (Error err) { - throw err; - } catch (RuntimeException rex) { - throw rex; - } catch (Exception ex) { - throw new RuntimeException(ex); - } - } - public void run() { invoke(); } - private static final long serialVersionUID = 2838392045355241008L; - } - public List> invokeAll(Collection> tasks) { - ArrayList> ts = + ArrayList> forkJoinTasks = new ArrayList>(tasks.size()); - for (Callable c : tasks) - ts.add(new AdaptedCallable(c)); - invoke(new InvokeAll(ts)); - return (List>) (List) ts; + for (Callable task : tasks) + forkJoinTasks.add(ForkJoinTask.adapt(task)); + invoke(new InvokeAll(forkJoinTasks)); + + @SuppressWarnings({"unchecked", "rawtypes"}) + List> futures = (List>) (List) forkJoinTasks; + return futures; } static final class InvokeAll extends RecursiveAction { @@ -670,8 +1357,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. * @@ -685,93 +1370,16 @@ public class ForkJoinPool extends Abstra * Returns the handler for internal worker threads that terminate * due to unrecoverable errors encountered while executing tasks. * - * @return the handler, or null if none + * @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; + return ueh; } /** - * 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. + * Returns the targeted parallelism level of this pool. * - * @param h the new handler - * @return the old handler, or 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(); - } - - /** - * Returns the targeted number of worker threads in 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; @@ -780,95 +1388,20 @@ public class ForkJoinPool extends Abstra /** * Returns the number of worker threads that have started but not * yet terminated. This result returned by this method may differ - * from {@code getParallelism} when threads are created to + * from {@link #getParallelism} when threads are created to * maintain parallelism when others are cooperatively blocked. * * @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; + return workerCounts >>> TOTAL_COUNT_SHIFT; } /** - * 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 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 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 true to maintains 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 pool is quiescent, and - * typically only before any tasks are submitted. The effects of - * invocations at other times may be unpredictable. - * - * @param async if true, use locally FIFO scheduling - * @return the previous mode - */ - 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; - } - - /** - * Returns true if this pool uses local first-in-first-out + * Returns {@code true} if this pool uses local first-in-first-out * scheduling mode for forked tasks that are never joined. * - * @return true if this pool uses async mode + * @return {@code true} if this pool uses async mode */ public boolean getAsyncMode() { return locallyFifo; @@ -877,12 +1410,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; } /** @@ -893,34 +1427,22 @@ 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; } /** - * Returns true if all worker threads are currently idle. An idle - * worker is one that cannot obtain a task to execute because none - * are available to steal from other threads, and there are no - * pending submissions to the pool. This method is conservative; - * it might not return true immediately upon idleness of all - * threads, but will eventually become true if threads remain - * inactive. + * Returns {@code true} if all worker threads are currently idle. + * An idle worker is one that cannot obtain a task to execute + * because none are available to steal from other threads, and + * there are no pending submissions to the pool. This method is + * conservative; it might not return {@code true} immediately upon + * idleness of all threads, but will eventually become true if + * threads remain inactive. * - * @return true if all threads are currently idle + * @return {@code true} if all threads are currently idle */ public boolean isQuiescent() { - return activeCountOf(runControl) == 0; + return (runState & ACTIVE_COUNT_MASK) == 0; } /** @@ -935,17 +1457,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; } /** @@ -961,19 +1473,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 @@ -983,8 +1494,8 @@ public class ForkJoinPool extends Abstra } /** - * Returns true if there are any tasks submitted to this pool - * that have not yet begun executing. + * Returns {@code true} if there are any tasks submitted to this + * pool that have not yet begun executing. * * @return {@code true} if there are any queued submissions */ @@ -997,7 +1508,7 @@ public class ForkJoinPool extends Abstra * available. This method may be useful in extensions to this * class that re-assign work in systems with multiple pools. * - * @return the next submission, or null if none + * @return the next submission, or {@code null} if none */ protected ForkJoinTask pollSubmission() { return submissionQueue.poll(); @@ -1007,8 +1518,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 @@ -1020,20 +1531,35 @@ public class ForkJoinPool extends Abstra * @param c the collection to transfer elements into * @return the number of elements transferred */ - protected int drainTasksTo(Collection> c) { + 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. @@ -1041,36 +1567,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. @@ -1085,20 +1611,19 @@ public class ForkJoinPool extends Abstra */ public void shutdown() { checkPermission(); - transitionRunStateTo(SHUTDOWN); - if (canTerminateOnShutdown(runControl)) - 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 {@code 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 @@ -1108,7 +1633,7 @@ public class ForkJoinPool extends Abstra */ public List shutdownNow() { checkPermission(); - terminate(); + tryTerminate(true); return Collections.emptyList(); } @@ -1118,17 +1643,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; } /** @@ -1137,7 +1666,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; } /** @@ -1153,596 +1682,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 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 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 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 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. - * Method {@code isReleasable} must return true if blocking is not - * necessary. Method {@code block} blocks the current thread if - * necessary (perhaps internally invoking {@code isReleasable} - * before actually blocking.). + * 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). * *

For example, here is a ManagedBlocker based on a * ReentrantLock: @@ -1766,145 +1721,111 @@ public class ForkJoinPool extends Abstra * Possibly blocks the current thread, for example waiting for * a lock or condition. * - * @return true if no additional blocking is necessary (i.e., - * if isReleasable would return true) + * @return {@code true} if no additional blocking is necessary + * (i.e., if isReleasable would return true) * @throws InterruptedException if interrupted while waiting * (the method is not required to do so, but is allowed to) */ boolean block() throws InterruptedException; /** - * Returns true if blocking is unnecessary. + * Returns {@code true} if blocking is unnecessary. */ boolean isReleasable(); } /** * 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 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 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 + // AbstractExecutorService overrides. These rely on undocumented + // fact that ForkJoinTask.adapt returns ForkJoinTasks that also + // implement RunnableFuture. protected RunnableFuture newTaskFor(Runnable runnable, T value) { - return new AdaptedRunnable(runnable, value); + return (RunnableFuture) ForkJoinTask.adapt(runnable, value); } protected RunnableFuture newTaskFor(Callable callable) { - return new AdaptedCallable(callable); + return (RunnableFuture) ForkJoinTask.adapt(callable); } + // Unsafe mechanics + + private static final sun.misc.Unsafe UNSAFE = getUnsafe(); + private static final long workerCountsOffset = + objectFieldOffset("workerCounts", ForkJoinPool.class); + 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 { + return UNSAFE.objectFieldOffset(klazz.getDeclaredField(field)); + } catch (NoSuchFieldException e) { + // Convert Exception to corresponding Error + NoSuchFieldError error = new NoSuchFieldError(field); + error.initCause(e); + throw error; + } + } - // Temporary Unsafe mechanics for preliminary release - private static Unsafe getUnsafe() throws Throwable { + /** + * Returns a sun.misc.Unsafe. Suitable for use in a 3rd party package. + * Replace with a simple call to Unsafe.getUnsafe when integrating + * into a jdk. + * + * @return a sun.misc.Unsafe + */ + private static sun.misc.Unsafe getUnsafe() { try { - return Unsafe.getUnsafe(); + return sun.misc.Unsafe.getUnsafe(); } catch (SecurityException se) { try { return java.security.AccessController.doPrivileged - (new java.security.PrivilegedExceptionAction() { - public Unsafe run() throws Exception { - return getUnsafePrivileged(); + (new java.security + .PrivilegedExceptionAction() { + public sun.misc.Unsafe run() throws Exception { + java.lang.reflect.Field f = sun.misc + .Unsafe.class.getDeclaredField("theUnsafe"); + f.setAccessible(true); + return (sun.misc.Unsafe) f.get(null); }}); } catch (java.security.PrivilegedActionException e) { - throw e.getCause(); + throw new RuntimeException("Could not initialize intrinsics", + e.getCause()); } } } - - private static Unsafe getUnsafePrivileged() - throws NoSuchFieldException, IllegalAccessException { - Field f = Unsafe.class.getDeclaredField("theUnsafe"); - f.setAccessible(true); - return (Unsafe) f.get(null); - } - - private static long fieldOffset(String fieldName) - throws NoSuchFieldException { - return UNSAFE.objectFieldOffset - (ForkJoinPool.class.getDeclaredField(fieldName)); - } - - static final Unsafe UNSAFE; - static final long eventCountOffset; - static final long workerCountsOffset; - static final long runControlOffset; - static final long syncStackOffset; - static final long spareStackOffset; - - static { - try { - UNSAFE = getUnsafe(); - eventCountOffset = fieldOffset("eventCount"); - workerCountsOffset = fieldOffset("workerCounts"); - runControlOffset = fieldOffset("runControl"); - syncStackOffset = fieldOffset("syncStack"); - spareStackOffset = fieldOffset("spareStack"); - } catch (Throwable e) { - throw new RuntimeException("Could not initialize intrinsics", e); - } - } - - 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); - } }