ForkJoinPool.java revision 2754
1771N/A * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER. 1771N/A * This code is free software; you can redistribute it and/or modify it 1771N/A * under the terms of the GNU General Public License version 2 only, as 2362N/A * published by the Free Software Foundation. Oracle designates this 1771N/A * particular file as subject to the "Classpath" exception as provided 2362N/A * by Oracle in the LICENSE file that accompanied this code. 1771N/A * This code is distributed in the hope that it will be useful, but WITHOUT 1771N/A * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or 1771N/A * FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License 1771N/A * version 2 for more details (a copy is included in the LICENSE file that 1771N/A * You should have received a copy of the GNU General Public License version 1771N/A * 2 along with this work; if not, write to the Free Software Foundation, 1771N/A * Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA. 2362N/A * Please contact Oracle, 500 Oracle Parkway, Redwood Shores, CA 94065 USA 2362N/A * or visit www.oracle.com if you need additional information or have any 1771N/A * This file is available under and governed by the GNU General Public 1771N/A * License version 2 only, as published by the Free Software Foundation. 1771N/A * However, the following notice accompanied the original version of this 1771N/A * Written by Doug Lea with assistance from members of JCP JSR-166 1771N/A * Expert Group and released to the public domain, as explained at 1771N/A * An {@link ExecutorService} for running {@link ForkJoinTask}s. 1771N/A * A {@code ForkJoinPool} provides the entry point for submissions 2754N/A * from non-{@code ForkJoinTask} clients, as well as management and 1771N/A * <p>A {@code ForkJoinPool} differs from other kinds of {@link 1771N/A * ExecutorService} mainly by virtue of employing 1771N/A * <em>work-stealing</em>: all threads in the pool attempt to find and 1771N/A * execute subtasks created by other active tasks (eventually blocking 1771N/A * waiting for work if none exist). This enables efficient processing 1771N/A * when most tasks spawn other subtasks (as do most {@code 2754N/A * ForkJoinTask}s). When setting <em>asyncMode</em> to true in 2754N/A * constructors, {@code ForkJoinPool}s may also be appropriate for use 2754N/A * with event-style tasks that are never joined. 1771N/A * <p>A {@code ForkJoinPool} is constructed with a given target 1771N/A * parallelism level; by default, equal to the number of available 2754N/A * processors. The pool attempts to maintain enough active (or 2754N/A * available) threads by dynamically adding, suspending, or resuming 2754N/A * internal worker threads, even if some tasks are stalled waiting to 2754N/A * join others. However, no such adjustments are guaranteed in the 2754N/A * face of blocked IO or other unmanaged synchronization. The nested 2754N/A * {@link ManagedBlocker} interface enables extension of the kinds of 2754N/A * synchronization accommodated. 1771N/A * <p>In addition to execution and lifecycle control methods, this 1771N/A * class provides status check methods (for example 1771N/A * {@link #getStealCount}) that are intended to aid in developing, 1771N/A * {@link #toString} returns indications of pool state in a 1771N/A * convenient form for informal monitoring. 2754N/A * <p> As is the case with other ExecutorServices, there are three 2754N/A * main task execution methods summarized in the following 2754N/A * table. These are designed to be used by clients not already engaged 2754N/A * these methods accept instances of {@code ForkJoinTask}, but 2754N/A * overloaded forms also allow mixed execution of plain {@code 2754N/A * Runnable}- or {@code Callable}- based activities as well. However, 2754N/A * tasks that are already executing in a pool should normally 2754N/A * <em>NOT</em> use these pool execution methods, but instead use the 2754N/A * within-computation forms listed in the table. 2754N/A * <table BORDER CELLPADDING=3 CELLSPACING=1> 2754N/A * <td ALIGN=CENTER> <b>Call from within fork/join computations</b></td> 2754N/A * <td> <b>Arrange async execution</td> 2754N/A * <td> {@link #execute(ForkJoinTask)}</td> 2754N/A * <td> {@link ForkJoinTask#fork}</td> 2754N/A * <td> <b>Await and obtain result</td> 2754N/A * <td> {@link #invoke(ForkJoinTask)}</td> 2754N/A * <td> {@link ForkJoinTask#invoke}</td> 2754N/A * <td> <b>Arrange exec and obtain Future</td> 2754N/A * <td> {@link #submit(ForkJoinTask)}</td> 2754N/A * <td> {@link ForkJoinTask#fork} (ForkJoinTasks <em>are</em> Futures)</td> 1771N/A * <p><b>Sample Usage.</b> Normally a single {@code ForkJoinPool} is 1771N/A * used for all parallel task execution in a program or subsystem. 1771N/A * Otherwise, use would not usually outweigh the construction and 1771N/A * bookkeeping overhead of creating a large set of threads. For 1771N/A * example, a common pool could be used for the {@code SortTasks} 1771N/A * illustrated in {@link RecursiveAction}. Because {@code 1771N/A * ForkJoinPool} uses threads in {@linkplain java.lang.Thread#isDaemon 1771N/A * daemon} mode, there is typically no need to explicitly {@link 1771N/A * #shutdown} such a pool upon program exit. 1771N/A * static final ForkJoinPool mainPool = new ForkJoinPool(); 1771N/A * public void sort(long[] array) { 1771N/A * mainPool.invoke(new SortTask(array, 0, array.length)); 1771N/A * <p><b>Implementation notes</b>: This implementation restricts the 1771N/A * maximum number of running threads to 32767. Attempts to create 1771N/A * pools with greater than the maximum number result in 1771N/A * {@code IllegalArgumentException}. 1771N/A * <p>This implementation rejects submitted tasks (that is, by throwing 2754N/A * {@link RejectedExecutionException}) only when the pool is shut down 2754N/A * or internal resources have been exhausted. 2754N/A * This class provides the central bookkeeping and control for a 2754N/A * set of worker threads: Submissions from non-FJ threads enter 2754N/A * into a submission queue. Workers take these tasks and typically 2754N/A * split them into subtasks that may be stolen by other workers. 2754N/A * The main work-stealing mechanics implemented in class 2754N/A * ForkJoinWorkerThread give first priority to processing tasks 2754N/A * from their own queues (LIFO or FIFO, depending on mode), then 2754N/A * to randomized FIFO steals of tasks in other worker queues, and 2754N/A * lastly to new submissions. These mechanics do not consider 2754N/A * affinities, loads, cache localities, etc, so rarely provide the 2754N/A * best possible performance on a given machine, but portably 2754N/A * provide good throughput by averaging over these factors. 2754N/A * (Further, even if we did try to use such information, we do not 2754N/A * usually have a basis for exploiting it. For example, some sets 2754N/A * of tasks profit from cache affinities, but others are harmed by 2754N/A * cache pollution effects.) 2754N/A * Beyond work-stealing support and essential bookkeeping, the 2754N/A * main responsibility of this framework is to take actions when 2754N/A * one worker is waiting to join a task stolen (or always held by) 2754N/A * another. Because we are multiplexing many tasks on to a pool 2754N/A * of workers, we can't just let them block (as in Thread.join). 2754N/A * We also cannot just reassign the joiner's run-time stack with 2754N/A * another and replace it later, which would be a form of 2754N/A * "continuation", that even if possible is not necessarily a good 2754N/A * idea. Given that the creation costs of most threads on most 2754N/A * systems mainly surrounds setting up runtime stacks, thread 2754N/A * creation and switching is usually not much more expensive than 2754N/A * stack creation and switching, and is more flexible). Instead we 2754N/A * Helping: Arranging for the joiner to execute some task that it 2754N/A * would be running if the steal had not occurred. Method 2754N/A * ForkJoinWorkerThread.helpJoinTask tracks joining->stealing 2754N/A * links to try to find such a task. 2754N/A * Compensating: Unless there are already enough live threads, 2754N/A * method helpMaintainParallelism() may create or 2754N/A * re-activate a spare thread to compensate for blocked 2754N/A * joiners until they unblock. 2754N/A * It is impossible to keep exactly the target (parallelism) 2754N/A * number of threads running at any given time. Determining 2754N/A * existence of conservatively safe helping targets, the 2754N/A * availability of already-created spares, and the apparent need 2754N/A * to create new spares are all racy and require heuristic 2754N/A * guidance, so we rely on multiple retries of each. Compensation 2754N/A * occurs in slow-motion. It is triggered only upon timeouts of 2754N/A * Object.wait used for joins. This reduces poor decisions that 2754N/A * would otherwise be made when threads are waiting for others 2754N/A * that are stalled because of unrelated activities such as 2754N/A * The ManagedBlocker extension API can't use helping so relies 2754N/A * only on compensation in method awaitBlocker. 2754N/A * The main throughput advantages of work-stealing stem from 2754N/A * decentralized control -- workers mostly steal tasks from each 2754N/A * other. We do not want to negate this by creating bottlenecks 2754N/A * implementing other management responsibilities. So we use a 2754N/A * collection of techniques that avoid, reduce, or cope well with 2754N/A * contention. These entail several instances of bit-packing into 2754N/A * CASable fields to maintain only the minimally required 2754N/A * atomicity. To enable such packing, we restrict maximum 2754N/A * parallelism to (1<<15)-1 (enabling twice this (to accommodate 2754N/A * unbalanced increments and decrements) to fit into a 16 bit 2754N/A * field, which is far in excess of normal operating range. Even 2754N/A * though updates to some of these bookkeeping fields do sometimes 2754N/A * contend with each other, they don't normally cache-contend with 2754N/A * updates to others enough to warrant memory padding or 2754N/A * isolation. So they are all held as fields of ForkJoinPool 2754N/A * objects. The main capabilities are as follows: 2754N/A * 1. Creating and removing workers. Workers are recorded in the 2754N/A * "workers" array. This is an array as opposed to some other data 2754N/A * structure to support index-based random steals by workers. 2754N/A * Updates to the array recording new workers and unrecording 2754N/A * terminated ones are protected from each other by a lock 2754N/A * (workerLock) but the array is otherwise concurrently readable, 2754N/A * and accessed directly by workers. To simplify index-based 2754N/A * operations, the array size is always a power of two, and all 2754N/A * readers must tolerate null slots. Currently, all worker thread 2754N/A * creation is on-demand, triggered by task submissions, 2754N/A * replacement of terminated workers, and/or compensation for 2754N/A * blocked workers. However, all other support code is set up to 2754N/A * work with other policies. 2754N/A * To ensure that we do not hold on to worker references that 2754N/A * would prevent GC, ALL accesses to workers are via indices into 2754N/A * the workers array (which is one source of some of the unusual 2754N/A * code constructions here). In essence, the workers array serves 2754N/A * as a WeakReference mechanism. Thus for example the event queue 2754N/A * stores worker indices, not worker references. Access to the 2754N/A * workers in associated methods (for example releaseEventWaiters) 2754N/A * must both index-check and null-check the IDs. All such accesses 2754N/A * ignore bad IDs by returning out early from what they are doing, 2754N/A * since this can only be associated with shutdown, in which case 2754N/A * it is OK to give up. On termination, we just clobber these 2754N/A * data structures without trying to use them. 2754N/A * 2. Bookkeeping for dynamically adding and removing workers. We 2754N/A * aim to approximately maintain the given level of parallelism. 2754N/A * When some workers are known to be blocked (on joins or via 2754N/A * ManagedBlocker), we may create or resume others to take their 2754N/A * place until they unblock (see below). Implementing this 2754N/A * requires counts of the number of "running" threads (i.e., those 2754N/A * that are neither blocked nor artificially suspended) as well as 2754N/A * the total number. These two values are packed into one field, 2754N/A * "workerCounts" because we need accurate snapshots when deciding 2754N/A * to create, resume or suspend. Note however that the 2754N/A * correspondence of these counts to reality is not guaranteed. In 2754N/A * particular updates for unblocked threads may lag until they 2754N/A * 3. Maintaining global run state. The run state of the pool 2754N/A * consists of a runLevel (SHUTDOWN, TERMINATING, etc) similar to 2754N/A * those in other Executor implementations, as well as a count of 2754N/A * "active" workers -- those that are, or soon will be, or 2754N/A * recently were executing tasks. The runLevel and active count 2754N/A * are packed together in order to correctly trigger shutdown and 2754N/A * termination. Without care, active counts can be subject to very 2754N/A * high contention. We substantially reduce this contention by 2754N/A * relaxing update rules. A worker must claim active status 2754N/A * prospectively, by activating if it sees that a submitted or 2754N/A * stealable task exists (it may find after activating that the 2754N/A * task no longer exists). It stays active while processing this 2754N/A * task (if it exists) and any other local subtasks it produces, 2754N/A * until it cannot find any other tasks. It then tries 2754N/A * inactivating (see method preStep), but upon update contention 2754N/A * instead scans for more tasks, later retrying inactivation if it 2754N/A * 4. Managing idle workers waiting for tasks. We cannot let 2754N/A * workers spin indefinitely scanning for tasks when none are 2754N/A * available. On the other hand, we must quickly prod them into 2754N/A * action when new tasks are submitted or generated. We 2754N/A * Field eventCount is incremented upon events that may enable 2754N/A * workers that previously could not find a task to now find one: 2754N/A * Submission of a new task to the pool, or another worker pushing 2754N/A * a task onto a previously empty queue. (We also use this 2754N/A * mechanism for configuration and termination actions that 2754N/A * require wakeups of idle workers). Each worker maintains its 2754N/A * last known event count, and blocks when a scan for work did not 2754N/A * find a task AND its lastEventCount matches the current 2754N/A * eventCount. Waiting idle workers are recorded in a variant of 2754N/A * Treiber stack headed by field eventWaiters which, when nonzero, 2754N/A * encodes the thread index and count awaited for by the worker 2754N/A * thread most recently calling eventSync. This thread in turn has 2754N/A * a record (field nextEventWaiter) for the next waiting worker. 2754N/A * In addition to allowing simpler decisions about need for 2754N/A * wakeup, the event count bits in eventWaiters serve the role of 2754N/A * tags to avoid ABA errors in Treiber stacks. Upon any wakeup, 2754N/A * released threads also try to release at most two others. The 2754N/A * net effect is a tree-like diffusion of signals, where released 2754N/A * threads (and possibly others) help with unparks. To further 2754N/A * reduce contention effects a bit, failed CASes to increment 2754N/A * field eventCount are tolerated without retries in signalWork. 2754N/A * Conceptually they are merged into the same event, which is OK 2754N/A * when their only purpose is to enable workers to scan for work. 2754N/A * 5. Managing suspension of extra workers. When a worker notices 2754N/A * (usually upon timeout of a wait()) that there are too few 2754N/A * running threads, we may create a new thread to maintain 2754N/A * parallelism level, or at least avoid starvation. Usually, extra 2754N/A * threads are needed for only very short periods, yet join 2754N/A * dependencies are such that we sometimes need them in 2754N/A * bursts. Rather than create new threads each time this happens, 2754N/A * we suspend no-longer-needed extra ones as "spares". For most 2754N/A * purposes, we don't distinguish "extra" spare threads from 2754N/A * normal "core" threads: On each call to preStep (the only point 2754N/A * at which we can do this) a worker checks to see if there are 2754N/A * now too many running workers, and if so, suspends itself. 2754N/A * Method helpMaintainParallelism looks for suspended threads to 2754N/A * resume before considering creating a new replacement. The 2754N/A * spares themselves are encoded on another variant of a Treiber 2754N/A * Stack, headed at field "spareWaiters". Note that the use of 2754N/A * spares is intrinsically racy. One thread may become a spare at 2754N/A * about the same time as another is needlessly being created. We 2754N/A * counteract this and related slop in part by requiring resumed 2754N/A * spares to immediately recheck (in preStep) to see whether they 2754N/A * 6. Killing off unneeded workers. A timeout mechanism is used to 2754N/A * shed unused workers: The oldest (first) event queue waiter uses 2754N/A * a timed rather than hard wait. When this wait times out without 2754N/A * a normal wakeup, it tries to shutdown any one (for convenience 2754N/A * the newest) other spare or event waiter via 2754N/A * tryShutdownUnusedWorker. This eventually reduces the number of 2754N/A * worker threads to a minimum of one after a long enough period 2754N/A * 7. Deciding when to create new workers. The main dynamic 2754N/A * control in this class is deciding when to create extra threads 2754N/A * in method helpMaintainParallelism. We would like to keep 2754N/A * exactly #parallelism threads running, which is an impossible 2754N/A * task. We always need to create one when the number of running 2754N/A * threads would become zero and all workers are busy. Beyond 2754N/A * this, we must rely on heuristics that work well in the 2754N/A * presence of transient phenomena such as GC stalls, dynamic 2754N/A * compilation, and wake-up lags. These transients are extremely 2754N/A * common -- we are normally trying to fully saturate the CPUs on 2754N/A * a machine, so almost any activity other than running tasks 2754N/A * impedes accuracy. Our main defense is to allow parallelism to 2754N/A * lapse for a while during joins, and use a timeout to see if, 2754N/A * after the resulting settling, there is still a need for 2754N/A * additional workers. This also better copes with the fact that 2754N/A * some of the methods in this class tend to never become compiled 2754N/A * (but are interpreted), so some components of the entire set of 2754N/A * controls might execute 100 times faster than others. And 2754N/A * similarly for cases where the apparent lack of work is just due 2754N/A * to GC stalls and other transient system activity. 2754N/A * Beware that there is a lot of representation-level coupling 2754N/A * among classes ForkJoinPool, ForkJoinWorkerThread, and 2754N/A * ForkJoinTask. For example, direct access to "workers" array by 2754N/A * workers, and direct access to ForkJoinTask.status by both 2754N/A * ForkJoinPool and ForkJoinWorkerThread. There is little point 2754N/A * trying to reduce this, since any associated future changes in 2754N/A * representations will need to be accompanied by algorithmic 2754N/A * Style notes: There are lots of inline assignments (of form 2754N/A * "while ((local = field) != 0)") which are usually the simplest 2754N/A * way to ensure the required read orderings (which are sometimes 2754N/A * critical). Also several occurrences of the unusual "do {} 2754N/A * while (!cas...)" which is the simplest way to force an update of 2754N/A * a CAS'ed variable. There are also other coding oddities that 2754N/A * help some methods perform reasonably even when interpreted (not 2754N/A * compiled), at the expense of some messy constructions that 2754N/A * The order of declarations in this file is: (1) statics (2) 2754N/A * fields (along with constants used when unpacking some of them) 2754N/A * (3) internal control methods (4) callbacks and other support 2754N/A * for ForkJoinTask and ForkJoinWorkerThread classes, (5) exported 2754N/A * methods (plus a few little helpers). 1771N/A * Factory for creating new {@link ForkJoinWorkerThread}s. 1771N/A * A {@code ForkJoinWorkerThreadFactory} must be defined and used 1771N/A * for {@code ForkJoinWorkerThread} subclasses that extend base 1771N/A * functionality or initialize threads with different contexts. 1771N/A * Returns a new worker thread operating in the given pool. 1771N/A * @param pool the pool this thread works in 1771N/A * @throws NullPointerException if the pool is null 1771N/A * Default ForkJoinWorkerThreadFactory implementation; creates a 1771N/A * new ForkJoinWorkerThread. 1771N/A * Creates a new ForkJoinWorkerThread. This factory is used unless 1771N/A * overridden in ForkJoinPool constructors. 1771N/A * Permission required for callers of methods that may start or 1771N/A * If there is a security manager, makes sure caller has 1771N/A * permission to modify threads. 1771N/A * Generator for assigning sequence numbers as pool names. 2754N/A * The time to block in a join (see awaitJoin) before checking if 2754N/A * a new worker should be (re)started to maintain parallelism 2754N/A * level. The value should be short enough to maintain global 2754N/A * responsiveness and progress but long enough to avoid 2754N/A * counterproductive firings during GC stalls or unrelated system 2754N/A * activity, and to not bog down systems with continual re-firings 2754N/A * on GCs or legitimately long waits. 2754N/A * The wakeup interval (in nanoseconds) for the oldest worker 2754N/A * waiting for an event to invoke tryShutdownUnusedWorker to 2754N/A * shrink the number of workers. The exact value does not matter 2754N/A * too much. It must be short enough to release resources during 2754N/A * sustained periods of idleness, but not so short that threads 2754N/A * are continually re-created. 2754N/A 30L *
1000L *
1000L *
1000L;
// 2 per minute 2754N/A * Absolute bound for parallelism level. Twice this number plus 2754N/A * one (i.e., 0xfff) must fit into a 16bit field to enable 2754N/A * word-packing for some counts and indices. 2754N/A * Array holding all worker threads in the pool. Array size must 2754N/A * be a power of two. Updates and replacements are protected by 2754N/A * workerLock, but the array is always kept in a consistent enough 2754N/A * state to be randomly accessed without locking by workers 2754N/A * performing work-stealing, as well as other traversal-based 2754N/A * methods in this class. All readers must tolerate that some 1771N/A * Queue for external submissions. 2754N/A * Lock protecting updates to workers array. 2754N/A * Latch released upon termination. 2754N/A * Creation factory for worker threads. 2754N/A * Sum of per-thread steal counts, updated only when threads are 2754N/A * Encoded record of top of Treiber stack of threads waiting for 2754N/A * events. The top 32 bits contain the count being waited for. The 2754N/A * bottom 16 bits contains one plus the pool index of waiting 2754N/A * worker thread. (Bits 16-31 are unused.) 2754N/A * A counter for events that may wake up worker threads: 2754N/A * - Submission of a new task to the pool 2754N/A * - A worker pushing a task on an empty queue 2754N/A * Encoded record of top of Treiber stack of spare threads waiting 2754N/A * for resumption. The top 16 bits contain an arbitrary count to 2754N/A * avoid ABA effects. The bottom 16bits contains one plus the pool 2754N/A * index of waiting worker thread. 2754N/A * Lifecycle control. The low word contains the number of workers 2754N/A * that are (probably) executing tasks. This value is atomically 2754N/A * incremented before a worker gets a task to run, and decremented 2754N/A * when a worker has no tasks and cannot find any. Bits 16-18 2754N/A * contain runLevel value. When all are zero, the pool is 2754N/A * running. Level transitions are monotonic (running -> shutdown 2754N/A * -> terminating -> terminated) so each transition adds a bit. 2754N/A * These are bundled together to ensure consistent read for 2754N/A * termination checks (i.e., that runLevel is at least SHUTDOWN 2754N/A * and active threads is zero). 2754N/A * Notes: Most direct CASes are dependent on these bitfield 2754N/A * positions. Also, this field is non-private to enable direct 2754N/A * performance-sensitive CASes in ForkJoinWorkerThread. 2754N/A // Note: The order among run level values matters. 2754N/A * Holds number of total (i.e., created and not yet terminated) 2754N/A * and running (i.e., not blocked on joins or other managed sync) 2754N/A * threads, packed together to ensure consistent snapshot when 2754N/A * making decisions about creating and suspending spare 2754N/A * threads. Updated only by CAS. Note that adding a new worker 2754N/A * requires incrementing both counts, since workers start off in 2754N/A * The target parallelism level. 2754N/A * Accessed directly by ForkJoinWorkerThreads. 2754N/A * True if use local fifo, not default lifo, for local polling 2754N/A * Read by, and replicated by ForkJoinWorkerThreads 2754N/A * The uncaught exception handler used when any worker abruptly 1771N/A * Pool number, just for assigning useful names to worker threads 2754N/A // Utilities for CASing fields. Note that most of these 2754N/A // are usually manually inlined by callers 2754N/A * Increments running count part of workerCounts 2754N/A * Tries to decrement running count unless already zero 2754N/A * Forces decrement of encoded workerCounts, awaiting nonzero if 2754N/A * (rarely) necessary when other count updates lag. 2754N/A * @param dr -- either zero or ONE_RUNNING 2754N/A * @param dt -- either zero or ONE_TOTAL 2754N/A return;
// lagging termination on a backout 1771N/A * Tries decrementing active count; fails on contention. 2754N/A * Called when workers cannot find tasks to run. 2754N/A * Advances to at least the given level. Returns true if not 2754N/A * already in at least the given level. 2754N/A // workers array maintenance 2754N/A * Records and returns a workers array index for new worker. 2754N/A // Try using slot totalCount-1. If not available, scan and/or resize 2754N/A * Nulls out record of worker in workers array. 2754N/A // Locking helps method recordWorker avoid unnecessary expansion 2754N/A * Final callback from terminating worker. Removes record of 2754N/A * worker from array, and adjusts counts. If pool is shutting 2754N/A * down, tries to complete termination. 2754N/A // Waiting for and signalling events 2754N/A * Releases workers blocked on a count not equal to current count. 2754N/A * Normally called after precheck that eventWaiters isn't zero to 2754N/A * avoid wasted array checks. Gives up upon a change in count or 2754N/A * upon releasing two workers, letting others take over. 2754N/A * Tries to advance eventCount and releases waiters. Called only 2754N/A int c;
// try to increment event count -- CAS failure OK 2754N/A * Adds the given worker to event queue and blocks until 2754N/A * terminating or event count advances from the given value 2754N/A * @param w the calling worker thread 2754N/A * Blocks the given worker (that has already been entered as an 2754N/A * event waiter) until terminating or event count advances from 2754N/A * the given value. The oldest (first) waiter uses a timed wait to 2754N/A * occasionally one-by-one shrink the number of workers (to a 2754N/A * minimum of one) if the pool has not been used for extended 2754N/A * @param w the calling worker thread 2754N/A * Pushes worker onto the spare stack. 2754N/A * Tries (once) to resume a spare if the number of running 2754N/A * threads is less than target. 2754N/A int c;
// increment running count before resume 2754N/A else // back out if w was shutdown 2754N/A * Tries to increase the number of running workers if below target 2754N/A * parallelism: If a spare exists tries to resume it via 2754N/A * tryResumeSpare. Otherwise, if not enough total workers or all 2754N/A * existing workers are busy, adds a new worker. In all cases also 2754N/A * helps wake up releasable workers waiting for work. 2754N/A }
finally {
// adjust on null or exceptional factory return 2754N/A int c;
// advance event count 2754N/A break;
// add at most one unless total below target 2754N/A * Callback from the oldest waiter in awaitEvent waking up after a 2754N/A * period of non-use. If all workers are idle, tries (once) to 2754N/A * shutdown an event waiter or a spare, if one exists. Note that 2754N/A * we don't need CAS or locks here because the method is called 2754N/A * only from one thread occasionally waking (and even misfires are 2754N/A * OK). Note that until the shutdown worker fully terminates, 2754N/A * workerCounts will overestimate total count, which is tolerable. 2754N/A * @param ec the event count waited on by caller (to abort 2754N/A * attempt if count has since changed). 2754N/A * Callback from workers invoked upon each top-level action (i.e., 2754N/A * stealing a task or taking a submission and running it). 2754N/A * Performs one or more of the following: 2754N/A * 1. If the worker is active and either did not run a task 2754N/A * or there are too many workers, try to set its active status 2754N/A * to inactive and update activeCount. On contention, we may 2754N/A * try again in this or a subsequent call. 2754N/A * 2. If not enough total workers, help create some. 2754N/A * 3. If there are too many running workers, suspend this worker 2754N/A * (first forcing inactive if necessary). If it is not needed, 2754N/A * it may be shutdown while suspended (via 2754N/A * tryShutdownUnusedWorker). Otherwise, upon resume it 2754N/A * rechecks running thread count and need for event sync. 2754N/A * 4. If worker did not run a task, await the next task event via 2754N/A * eventSync if necessary (first forcing inactivation), upon 2754N/A * which the worker may be shutdown via 2754N/A * tryShutdownUnusedWorker. Otherwise, help release any 2754N/A * existing event waiters that are now releasable, 2754N/A * @param ran true if worker ran a task since last call to this method 2754N/A * Helps and/or blocks awaiting join of the given task. 2754N/A * See above for explanation. 2754N/A * @param joinMe the task to join 2754N/A * @param worker the current worker thread 2754N/A * Same idea as awaitJoin, but no helping, retries, or timeouts. 2754N/A * @param now if true, unconditionally terminate, else only 2754N/A * if shutdown and empty queue and no active workers 2754N/A * @return true if now terminating or terminated 2754N/A // Finish now if all threads terminated; else in some subsequent call 2754N/A * Actions on transition to TERMINATING 2754N/A * Runs up to four passes through workers: (0) shutting down each 2754N/A * (without waking up if parked) to quickly spread notifications 2754N/A * without unnecessary bouncing around event queues etc (1) wake 2754N/A * up and help cancel tasks (2) interrupt (3) mop up races with 2754N/A int c;
// advance event count 2754N/A * Clears out and cancels submissions, ignoring exceptions. 2754N/A // misc support for ForkJoinWorkerThread 2754N/A * Tries to accumulate steal count from a worker, clearing 2754N/A * the worker's value if successful. 2754N/A * @return true if worker steal count now zero 2754N/A // CAS even if zero, for fence effects 2754N/A * Returns the approximate (non-atomic) number of idle threads per 2754N/A // Use exact results for small values, saturate past 4 2754N/A // Public and protected methods 1771N/A * Creates a {@code ForkJoinPool} with parallelism equal to {@link 2754N/A * java.lang.Runtime#availableProcessors}, using the {@linkplain 2754N/A * #defaultForkJoinWorkerThreadFactory default thread factory}, 2754N/A * no UncaughtExceptionHandler, and non-async LIFO processing mode. 1771N/A * @throws SecurityException if a security manager exists and 1771N/A * the caller is not permitted to modify threads 1771N/A * because it does not hold {@link 1771N/A * java.lang.RuntimePermission}{@code ("modifyThread")} 1771N/A * Creates a {@code ForkJoinPool} with the indicated parallelism 2754N/A * #defaultForkJoinWorkerThreadFactory default thread factory}, 2754N/A * no UncaughtExceptionHandler, and non-async LIFO processing mode. 1771N/A * @param parallelism the parallelism level 1771N/A * @throws IllegalArgumentException if parallelism less than or 1771N/A * equal to zero, or greater than implementation limit 1771N/A * @throws SecurityException if a security manager exists and 1771N/A * the caller is not permitted to modify threads 1771N/A * because it does not hold {@link 1771N/A * java.lang.RuntimePermission}{@code ("modifyThread")} 2754N/A * Creates a {@code ForkJoinPool} with the given parameters. 2754N/A * @param parallelism the parallelism level. For default value, 2754N/A * use {@link java.lang.Runtime#availableProcessors}. 2754N/A * @param factory the factory for creating new threads. For default value, 2754N/A * use {@link #defaultForkJoinWorkerThreadFactory}. 2754N/A * @param handler the handler for internal worker threads that 2754N/A * terminate due to unrecoverable errors encountered while executing 2754N/A * tasks. For default value, use {@code null}. 2754N/A * @param asyncMode if true, 2754N/A * establishes local first-in-first-out scheduling mode for forked 2754N/A * tasks that are never joined. This mode may be more appropriate 2754N/A * than default locally stack-based mode in applications in which 2754N/A * worker threads only process event-style asynchronous tasks. 2754N/A * For default value, use {@code false}. 1771N/A * @throws IllegalArgumentException if parallelism less than or 1771N/A * equal to zero, or greater than implementation limit 1771N/A * @throws NullPointerException if the factory is null 1771N/A * @throws SecurityException if a security manager exists and 1771N/A * the caller is not permitted to modify threads 1771N/A * because it does not hold {@link 1771N/A * java.lang.RuntimePermission}{@code ("modifyThread")} 2754N/A * Returns initial power of two size for workers array. 2754N/A * @param pc the initial parallelism level 2754N/A // If possible, initially allocate enough space for one spare 2754N/A // See Hackers Delight, sec 3.2. We know MAX_WORKERS < (1 >>> 16) 1771N/A * Common code for execute, invoke and submit 2754N/A int c;
// try to increment event count -- CAS failure OK 1771N/A * Performs the given task, returning its result upon completion. 1771N/A * @return the task's result 1771N/A * @throws NullPointerException if the task is null 1771N/A * @throws RejectedExecutionException if the task cannot be 1771N/A * Arranges for (asynchronous) execution of the given task. 1771N/A * @throws NullPointerException if the task is null 1771N/A * @throws RejectedExecutionException if the task cannot be 1771N/A // AbstractExecutorService methods 1771N/A * @throws NullPointerException if the task is null 1771N/A * @throws RejectedExecutionException if the task cannot be 2754N/A * Submits a ForkJoinTask for execution. 2754N/A * @param task the task to submit 2754N/A * @throws NullPointerException if the task is null 2754N/A * @throws RejectedExecutionException if the task cannot be 1771N/A * @throws NullPointerException if the task is null 1771N/A * @throws RejectedExecutionException if the task cannot be 1771N/A * @throws NullPointerException if the task is null 1771N/A * @throws RejectedExecutionException if the task cannot be 1771N/A * @throws NullPointerException if the task is null 1771N/A * @throws RejectedExecutionException if the task cannot be 1771N/A * @throws NullPointerException {@inheritDoc} 1771N/A * @throws RejectedExecutionException {@inheritDoc} 1771N/A * Returns the factory used for constructing new workers. 1771N/A * @return the factory used for constructing new workers 1771N/A * Returns the handler for internal worker threads that terminate 1771N/A * due to unrecoverable errors encountered while executing tasks. 1771N/A * @return the handler, or {@code null} if none 1771N/A * Returns the targeted parallelism level of this pool. 1771N/A * @return the targeted parallelism level of this pool 1771N/A * Returns the number of worker threads that have started but not 2754N/A * yet terminated. The result returned by this method may differ 1771N/A * from {@link #getParallelism} when threads are created to 1771N/A * maintain parallelism when others are cooperatively blocked. 1771N/A * @return the number of worker threads 1771N/A * Returns {@code true} if this pool uses local first-in-first-out 1771N/A * scheduling mode for forked tasks that are never joined. 1771N/A * @return {@code true} if this pool uses async mode 1771N/A * Returns an estimate of the number of worker threads that are 1771N/A * not blocked waiting to join tasks or for other managed 2754N/A * synchronization. This method may overestimate the 2754N/A * number of running threads. 1771N/A * @return the number of worker threads 1771N/A * Returns an estimate of the number of threads that are currently 1771N/A * stealing or executing tasks. This method may overestimate the 1771N/A * number of active threads. 1771N/A * @return the number of active threads 1771N/A * Returns {@code true} if all worker threads are currently idle. 1771N/A * An idle worker is one that cannot obtain a task to execute 1771N/A * because none are available to steal from other threads, and 1771N/A * there are no pending submissions to the pool. This method is 1771N/A * conservative; it might not return {@code true} immediately upon 1771N/A * idleness of all threads, but will eventually become true if 1771N/A * @return {@code true} if all threads are currently idle 1771N/A * Returns an estimate of the total number of tasks stolen from 1771N/A * one thread's work queue by another. The reported value 1771N/A * underestimates the actual total number of steals when the pool 1771N/A * is not quiescent. This value may be useful for monitoring and 1771N/A * high enough to keep threads busy, but low enough to avoid 1771N/A * overhead and contention across threads. 1771N/A * @return the number of steals 1771N/A * Returns an estimate of the total number of tasks currently held 1771N/A * in queues by worker threads (but not including tasks submitted 1771N/A * to the pool that have not begun executing). This value is only 1771N/A * an approximation, obtained by iterating across all threads in 1771N/A * the pool. This method may be useful for tuning task 1771N/A * @return the number of queued tasks 1771N/A * Returns an estimate of the number of tasks submitted to this 1771N/A * pool that have not yet begun executing. This method takes time 1771N/A * proportional to the number of submissions. 1771N/A * @return the number of queued submissions 1771N/A * Returns {@code true} if there are any tasks submitted to this 1771N/A * pool that have not yet begun executing. 1771N/A * @return {@code true} if there are any queued submissions 1771N/A * Removes and returns the next unexecuted submission if one is 1771N/A * available. This method may be useful in extensions to this 1771N/A * class that re-assign work in systems with multiple pools. 1771N/A * @return the next submission, or {@code null} if none 1771N/A * Removes all available unexecuted submitted and forked tasks 1771N/A * from scheduling queues and adds them to the given collection, 1771N/A * without altering their execution status. These may include 1771N/A * artificially generated or wrapped tasks. This method is 1771N/A * designed to be invoked only when the pool is known to be 1771N/A * quiescent. Invocations at other times may not remove all 1771N/A * tasks. A failure encountered while attempting to add elements 1771N/A * to collection {@code c} may result in elements being in 1771N/A * neither, either or both collections when the associated 1771N/A * exception is thrown. The behavior of this operation is 1771N/A * undefined if the specified collection is modified while the 1771N/A * operation is in progress. 1771N/A * @param c the collection to transfer elements into 1771N/A * @return the number of elements transferred 1771N/A * Returns a string identifying this pool, as well as its state, 1771N/A * including indications of run state, parallelism level, and 1771N/A * @return a string identifying this pool, as well as its state 1771N/A * Initiates an orderly shutdown in which previously submitted 1771N/A * tasks are executed, but no new tasks will be accepted. 1771N/A * Invocation has no additional effect if already shut down. 1771N/A * Tasks that are in the process of being submitted concurrently 1771N/A * during the course of this method may or may not be rejected. 1771N/A * @throws SecurityException if a security manager exists and 1771N/A * the caller is not permitted to modify threads 1771N/A * because it does not hold {@link 1771N/A * java.lang.RuntimePermission}{@code ("modifyThread")} 1771N/A * Attempts to cancel and/or stop all tasks, and reject all 1771N/A * subsequently submitted tasks. Tasks that are in the process of 1771N/A * being submitted or executed concurrently during the course of 1771N/A * this method may or may not be rejected. This method cancels 1771N/A * both existing and unexecuted tasks, in order to permit 1771N/A * termination in the presence of task dependencies. So the method 1771N/A * always returns an empty list (unlike the case for some other 1771N/A * @throws SecurityException if a security manager exists and 1771N/A * the caller is not permitted to modify threads 1771N/A * because it does not hold {@link 1771N/A * java.lang.RuntimePermission}{@code ("modifyThread")} 1771N/A * Returns {@code true} if all tasks have completed following shut down. 1771N/A * @return {@code true} if all tasks have completed following shut down 1771N/A * Returns {@code true} if the process of termination has 1771N/A * commenced but not yet completed. This method may be useful for 1771N/A * debugging. A return of {@code true} reported a sufficient 1771N/A * period after shutdown may indicate that submitted tasks have 1771N/A * ignored or suppressed interruption, causing this executor not 1771N/A * @return {@code true} if terminating but not yet terminated 1771N/A * Returns {@code true} if this pool has been shut down. 1771N/A * @return {@code true} if this pool has been shut down 1771N/A * Blocks until all tasks have completed execution after a shutdown 1771N/A * request, or the timeout occurs, or the current thread is 1771N/A * interrupted, whichever happens first. 1771N/A * @param timeout the maximum time to wait 1771N/A * @param unit the time unit of the timeout argument 1771N/A * @return {@code true} if this executor terminated and 1771N/A * {@code false} if the timeout elapsed before termination 1771N/A * @throws InterruptedException if interrupted while waiting 1771N/A * Interface for extending managed parallelism for tasks running 1771N/A * in {@link ForkJoinPool}s. 2754N/A * <p>A {@code ManagedBlocker} provides two methods. Method 2754N/A * {@code isReleasable} must return {@code true} if blocking is 2754N/A * not necessary. Method {@code block} blocks the current thread 2754N/A * if necessary (perhaps internally invoking {@code isReleasable} 2754N/A * before actually blocking). The unusual methods in this API 2754N/A * accommodate synchronizers that may, but don't usually, block 2754N/A * for long periods. Similarly, they allow more efficient internal 2754N/A * handling of cases in which additional workers may be, but 2754N/A * usually are not, needed to ensure sufficient parallelism. 2754N/A * Toward this end, implementations of method {@code isReleasable} 2754N/A * must be amenable to repeated invocation. 1771N/A * <p>For example, here is a ManagedBlocker based on a 1771N/A * class ManagedLocker implements ManagedBlocker { 1771N/A * final ReentrantLock lock; 1771N/A * ManagedLocker(ReentrantLock lock) { this.lock = lock; } 1771N/A * public boolean isReleasable() { 1771N/A * return hasLock || (hasLock = lock.tryLock()); 2754N/A * <p>Here is a class that possibly blocks waiting for an 2754N/A * class QueueTaker<E> implements ManagedBlocker { 2754N/A * final BlockingQueue<E> queue; 2754N/A * QueueTaker(BlockingQueue<E> q) { this.queue = q; } 2754N/A * public boolean block() throws InterruptedException { 2754N/A * public boolean isReleasable() { 2754N/A * return item != null || (item = queue.poll()) != null; 2754N/A * public E getItem() { // call after pool.managedBlock completes 1771N/A * Possibly blocks the current thread, for example waiting for 1771N/A * @return {@code true} if no additional blocking is necessary 1771N/A * (i.e., if isReleasable would return true) 1771N/A * @throws InterruptedException if interrupted while waiting 1771N/A * (the method is not required to do so, but is allowed to) 1771N/A * Returns {@code true} if blocking is unnecessary. 1771N/A * Blocks in accord with the given blocker. If the current thread 1771N/A * is a {@link ForkJoinWorkerThread}, this method possibly 1771N/A * arranges for a spare thread to be activated if necessary to 2754N/A * ensure sufficient parallelism while the current thread is blocked. 1771N/A * <p>If the caller is not a {@link ForkJoinTask}, this method is 1771N/A * behaviorally equivalent to 1771N/A * while (!blocker.isReleasable()) 1771N/A * If the caller is a {@code ForkJoinTask}, then the pool may 1771N/A * first be expanded to ensure parallelism, and later adjusted. 1771N/A * @param blocker the blocker 1771N/A * @throws InterruptedException if blocker.block did so 1771N/A // AbstractExecutorService overrides. These rely on undocumented 1771N/A // fact that ForkJoinTask.adapt returns ForkJoinTasks that also 1771N/A // implement RunnableFuture. 1771N/A // Convert Exception to corresponding Error