From 0dda8885a980e3796ea85e32161383c6f8f5613e Mon Sep 17 00:00:00 2001 From: Philipp Haller Date: Wed, 3 Aug 2011 21:13:33 +0000 Subject: Update fork/join framework to JDK release 1.7.0 --- .../scala/concurrent/forkjoin/ForkJoinPool.java | 3040 +++++++++++--------- .../scala/concurrent/forkjoin/ForkJoinTask.java | 1582 +++++----- .../concurrent/forkjoin/ForkJoinWorkerThread.java | 1240 ++++---- .../concurrent/forkjoin/LinkedTransferQueue.java | 1564 ++++++---- .../scala/concurrent/forkjoin/RecursiveAction.java | 89 +- .../scala/concurrent/forkjoin/RecursiveTask.java | 31 +- .../concurrent/forkjoin/ThreadLocalRandom.java | 81 +- .../scala/concurrent/forkjoin/TransferQueue.java | 89 +- .../scala/concurrent/forkjoin/package-info.java | 281 +- 9 files changed, 4760 insertions(+), 3237 deletions(-) (limited to 'src/forkjoin') diff --git a/src/forkjoin/scala/concurrent/forkjoin/ForkJoinPool.java b/src/forkjoin/scala/concurrent/forkjoin/ForkJoinPool.java index 3fad92cbf1..401ce6c5c9 100644 --- a/src/forkjoin/scala/concurrent/forkjoin/ForkJoinPool.java +++ b/src/forkjoin/scala/concurrent/forkjoin/ForkJoinPool.java @@ -1,107 +1,370 @@ /* * Written by Doug Lea with assistance from members of JCP JSR-166 * Expert Group and released to the public domain, as explained at - * http://creativecommons.org/licenses/publicdomain + * http://creativecommons.org/publicdomain/zero/1.0/ */ package scala.concurrent.forkjoin; -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.Random; +import java.util.concurrent.AbstractExecutorService; +import java.util.concurrent.Callable; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Future; +import java.util.concurrent.RejectedExecutionException; +import java.util.concurrent.RunnableFuture; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.locks.LockSupport; +import java.util.concurrent.locks.ReentrantLock; +import java.util.concurrent.locks.Condition; /** - * 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 - * 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 ManagedBlocker interface enables extension of - * the kinds of synchronization accommodated. The target parallelism - * level may also be changed dynamically (setParallelism) - * and thread construction can be limited using methods - * setMaximumPoolSize and/or - * 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 - * 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 - * 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 following + * table. These are designed to be used by clients not already engaged + * in fork/join computations in the current pool. The main forms of + * these methods accept instances of {@code ForkJoinTask}, but + * overloaded forms also allow mixed execution of plain {@code + * Runnable}- or {@code Callable}- based activities as well. However, + * tasks that are already executing in a pool should normally + * NOT use these pool execution methods, but instead use the + * within-computation forms listed in the table. + * + * + * + * + * + * + * + * + * + * + * + * + * + * + * + * + * + * + * + * + * + * + *
Call from non-fork/join clients Call from within fork/join computations
Arrange async execution {@link #execute(ForkJoinTask)} {@link ForkJoinTask#fork}
Await and obtain result {@link #invoke(ForkJoinTask)} {@link ForkJoinTask#invoke}
Arrange exec and obtain Future {@link #submit(ForkJoinTask)} {@link ForkJoinTask#fork} (ForkJoinTasks are Futures)
+ * + *

Sample Usage. Normally a single {@code ForkJoinPool} is + * used for all parallel task execution in a program or subsystem. + * Otherwise, use would not usually outweigh the construction and + * bookkeeping overhead of creating a large set of threads. For + * example, a common pool could be used for the {@code SortTasks} + * illustrated in {@link RecursiveAction}. Because {@code + * ForkJoinPool} uses threads in {@linkplain java.lang.Thread#isDaemon + * daemon} mode, there is typically no need to explicitly {@link + * #shutdown} such a pool upon program exit. + * + * 

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

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

This implementation rejects submitted tasks (that is, by throwing + * {@link RejectedExecutionException}) only when the pool is shut down + * or internal resources have been exhausted. + * + * @since 1.7 + * @author Doug Lea */ public class ForkJoinPool /*extends AbstractExecutorService*/ { /* - * See the extended comments interspersed below for design, - * rationale, and walkthroughs. + * 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. + * Preference rules 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. + * + * The main throughput advantages of work-stealing stem from + * decentralized control -- workers mostly take tasks from + * themselves or each other. We cannot negate this in the + * implementation of other management responsibilities. The main + * tactic for avoiding bottlenecks is packing nearly all + * essentially atomic control state into a single 64bit volatile + * variable ("ctl"). This variable is read on the order of 10-100 + * times as often as it is modified (always via CAS). (There is + * some additional control state, for example variable "shutdown" + * for which we can cope with uncoordinated updates.) This + * streamlines synchronization and control at the expense of messy + * constructions needed to repack status bits upon updates. + * Updates tend not to contend with each other except during + * bursts while submitted tasks begin or end. In some cases when + * they do contend, threads can instead do something else + * (usually, scan for tasks) until contention subsides. + * + * To enable packing, we restrict maximum parallelism to (1<<15)-1 + * (which is far in excess of normal operating range) to allow + * ids, counts, and their negations (used for thresholding) to fit + * into 16bit fields. + * + * Recording Workers. Workers are recorded in the "workers" array + * that is created upon pool construction and expanded if (rarely) + * necessary. 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 seqLock + * (scanGuard) 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. To avoid flailing during + * start-up, the array is presized to hold twice #parallelism + * workers (which is unlikely to need further resizing during + * execution). But to avoid dealing with so many null slots, + * variable scanGuard includes a mask for the nearest power of two + * that contains all current workers. 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. To ensure that we do not hold on to worker + * references that would prevent GC, ALL accesses to workers are + * via indices into the workers array (which is one source of some + * of the messy code constructions here). In essence, the workers + * array serves as a weak reference mechanism. Thus for example + * the wait queue field of ctl stores worker indices, not worker + * references. Access to the workers in associated methods (for + * example signalWork) must both index-check and null-check the + * IDs. All such accesses ignore bad IDs by returning out early + * from what they are doing, since this can only be associated + * with termination, in which case it is OK to give up. + * + * All uses of the workers array, as well as queue arrays, check + * that the array is non-null (even if previously non-null). This + * allows nulling during termination, which is currently not + * necessary, but remains an option for resource-revocation-based + * shutdown schemes. + * + * Wait Queuing. Unlike HPC work-stealing frameworks, we cannot + * let workers spin indefinitely scanning for tasks when none can + * be found immediately, and we cannot start/resume workers unless + * there appear to be tasks available. On the other hand, we must + * quickly prod them into action when new tasks are submitted or + * generated. We park/unpark workers after placing in an event + * wait queue when they cannot find work. This "queue" is actually + * a simple Treiber stack, headed by the "id" field of ctl, plus a + * 15bit counter value to both wake up waiters (by advancing their + * count) and avoid ABA effects. Successors are held in worker + * field "nextWait". Queuing deals with several intrinsic races, + * mainly that a task-producing thread can miss seeing (and + * signalling) another thread that gave up looking for work but + * has not yet entered the wait queue. We solve this by requiring + * a full sweep of all workers both before (in scan()) and after + * (in tryAwaitWork()) a newly waiting worker is added to the wait + * queue. During a rescan, the worker might release some other + * queued worker rather than itself, which has the same net + * effect. Because enqueued workers may actually be rescanning + * rather than waiting, we set and clear the "parked" field of + * ForkJoinWorkerThread to reduce unnecessary calls to unpark. + * (Use of the parked field requires a secondary recheck to avoid + * missed signals.) + * + * Signalling. We create or wake up workers only when there + * appears to be at least one task they might be able to find and + * execute. When a submission is added or another worker adds a + * task to a queue that previously had two or fewer tasks, they + * signal waiting workers (or trigger creation of new ones if + * fewer than the given parallelism level -- see signalWork). + * These primary signals are buttressed by signals during rescans + * as well as those performed when a worker steals a task and + * notices that there are more tasks too; together these cover the + * signals needed in cases when more than two tasks are pushed + * but untaken. + * + * Trimming workers. To release resources after periods of lack of + * use, a worker starting to wait when the pool is quiescent will + * time out and terminate if the pool has remained quiescent for + * SHRINK_RATE nanosecs. This will slowly propagate, eventually + * terminating all workers after long periods of non-use. + * + * Submissions. External submissions are maintained in an + * array-based queue that is structured identically to + * ForkJoinWorkerThread queues except for the use of + * submissionLock in method addSubmission. Unlike the case for + * worker queues, multiple external threads can add new + * submissions, so adding requires a lock. + * + * Compensation. Beyond work-stealing support and lifecycle + * control, the main responsibility of this framework is to take + * actions when one worker is waiting to join a task stolen (or + * always held by) another. Because we are multiplexing many + * tasks on to a pool of workers, we can't just let them block (as + * in Thread.join). We also cannot just reassign the joiner's + * run-time stack with another and replace it later, which would + * be a form of "continuation", that even if possible is not + * necessarily a good idea since we sometimes need both an + * unblocked task and its continuation to progress. Instead we + * combine two tactics: + * + * Helping: Arranging for the joiner to execute some task that it + * would be running if the steal had not occurred. Method + * ForkJoinWorkerThread.joinTask tracks joining->stealing + * links to try to find such a task. + * + * Compensating: Unless there are already enough live threads, + * method tryPreBlock() may create or re-activate a spare + * thread to compensate for blocked joiners until they + * unblock. + * + * The ManagedBlocker extension API can't use helping so relies + * only on compensation in method awaitBlocker. + * + * It is impossible to keep exactly the target parallelism number + * of threads running at any given time. Determining the + * existence of conservatively safe helping targets, the + * availability of already-created spares, and the apparent need + * to create new spares are all racy and require heuristic + * guidance, so we rely on multiple retries of each. Currently, + * in keeping with on-demand signalling policy, we compensate only + * if blocking would leave less than one active (non-waiting, + * non-blocked) worker. Additionally, to avoid some false alarms + * due to GC, lagging counters, system activity, etc, compensated + * blocking for joins is only attempted after rechecks stabilize + * (retries are interspersed with Thread.yield, for good + * citizenship). The variable blockedCount, incremented before + * blocking and decremented after, is sometimes needed to + * distinguish cases of waiting for work vs blocking on joins or + * other managed sync. Both cases are equivalent for most pool + * control, so we can update non-atomically. (Additionally, + * contention on blockedCount alleviates some contention on ctl). + * + * Shutdown and Termination. A call to shutdownNow atomically sets + * the ctl stop bit and then (non-atomically) sets each workers + * "terminate" status, cancels all unprocessed tasks, and wakes up + * all waiting workers. Detecting whether termination should + * commence after a non-abrupt shutdown() call requires more work + * and bookkeeping. We need consensus about quiesence (i.e., that + * there is no more work) which is reflected in active counts so + * long as there are no current blockers, as well as possible + * re-evaluations during independent changes in blocking or + * quiescing workers. + * + * Style notes: There is a lot of representation-level coupling + * among classes ForkJoinPool, ForkJoinWorkerThread, and + * ForkJoinTask. Most fields of ForkJoinWorkerThread maintain + * data structures managed by ForkJoinPool, so are directly + * accessed. Conversely we allow 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. All together, these low-level implementation + * choices produce as much as a factor of 4 performance + * improvement compared to naive implementations, and enable the + * processing of billions of tasks per second, at the expense of + * some ugliness. + * + * Methods signalWork() and scan() are the main bottlenecks so are + * especially heavily micro-optimized/mangled. There are lots of + * inline assignments (of form "while ((local = field) != 0)") + * which are usually the simplest way to ensure the required read + * orderings (which are sometimes critical). This leads to a + * "C"-like style of listing declarations of these locals at the + * heads of methods or blocks. There are several occurrences of + * the unusual "do {} while (!cas...)" which is the simplest way + * to force an update of a CAS'ed variable. There are also other + * coding oddities that help some methods perform reasonably even + * when interpreted (not compiled). + * + * The order of declarations in this file is: (1) declarations of + * statics (2) fields (along with constants used when unpacking + * some of them), listed in an order that tends to reduce + * contention among them a bit under most JVMs. (3) internal + * control methods (4) callbacks and other support for + * ForkJoinTask and ForkJoinWorkerThread classes, (5) exported + * methods (plus a few little helpers). (6) static block + * initializing all statics in a minimally dependent order. */ - /** 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; - - // placeholder for java.util.concurrent.RunnableFuture - interface RunnableFuture extends Runnable { + public static ForkJoinWorkerThread[] copyOfWorkers(ForkJoinWorkerThread[] original, int newLength) { + ForkJoinWorkerThread[] copy = new ForkJoinWorkerThread[newLength]; + System.arraycopy(original, 0, copy, 0, Math.min(newLength, original.length)); + return copy; } /** - * 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. + * 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); } /** - * Default ForkJoinWorkerThreadFactory implementation, creates a + * 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); } } @@ -110,15 +373,13 @@ public class ForkJoinPool /*extends AbstractExecutorService*/ { * overridden in ForkJoinPool constructors. */ public static final ForkJoinWorkerThreadFactory - defaultForkJoinWorkerThreadFactory = - new DefaultForkJoinWorkerThreadFactory(); + defaultForkJoinWorkerThreadFactory; /** * Permission required for callers of methods that may start or * kill threads. */ - private static final RuntimePermission modifyThreadPermission = - new RuntimePermission("modifyThread"); + private static final RuntimePermission modifyThreadPermission; /** * If there is a security manager, makes sure caller has @@ -133,726 +394,1289 @@ public class ForkJoinPool /*extends AbstractExecutorService*/ { /** * Generator for assigning sequence numbers as pool names. */ - private static final AtomicInteger poolNumberGenerator = - new AtomicInteger(); + private static final AtomicInteger poolNumberGenerator; /** - * 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. + * Generator for initial random seeds for worker victim + * selection. This is used only to create initial seeds. Random + * steals use a cheaper xorshift generator per steal attempt. We + * don't expect much contention on seedGenerator, so just use a + * plain Random. */ - public volatile ForkJoinWorkerThread[] workers; + static final Random workerSeedGenerator; /** - * Lock protecting access to workers. + * Array holding all worker threads in the pool. Initialized upon + * construction. Array size must be a power of two. Updates and + * replacements are protected by scanGuard, 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, so long + * as reads memory-acquire by first reading ctl. All readers must + * tolerate that some array slots may be null. */ - private final ReentrantLock workerLock; + ForkJoinWorkerThread[] workers; /** - * Condition for awaitTermination. + * Initial size for submission queue array. Must be a power of + * two. In many applications, these always stay small so we use a + * small initial cap. */ - private final Condition termination; + private static final int INITIAL_QUEUE_CAPACITY = 8; /** - * The uncaught exception handler used when any worker - * abrupty terminates + * Maximum size for submission queue array. Must be a power of two + * less than or equal to 1 << (31 - width of array entry) to + * ensure lack of index wraparound, but is capped at a lower + * value to help users trap runaway computations. */ - private Thread.UncaughtExceptionHandler ueh; + private static final int MAXIMUM_QUEUE_CAPACITY = 1 << 24; // 16M /** - * Creation factory for worker threads. + * Array serving as submission queue. Initialized upon construction. */ - private final ForkJoinWorkerThreadFactory factory; + private ForkJoinTask[] submissionQueue; /** - * Head of stack of threads that were created to maintain - * parallelism when other threads blocked, but have since - * suspended when the parallelism level rose. + * Lock protecting submissions array for addSubmission */ - private volatile WaitQueueNode spareStack; + private final ReentrantLock submissionLock; /** - * Sum of per-thread steal counts, updated only when threads are - * idle or terminating. + * Condition for awaitTermination, using submissionLock for + * convenience. */ - private final AtomicLong stealCount; + private final Condition termination; /** - * Queue for external submissions. + * Creation factory for worker threads. */ - private final LinkedTransferQueue> submissionQueue; + private final ForkJoinWorkerThreadFactory factory; /** - * Head of Treiber stack for barrier sync. See below for explanation + * The uncaught exception handler used when any worker abruptly + * terminates. */ - private volatile WaitQueueNode syncStack; + final Thread.UncaughtExceptionHandler ueh; /** - * The count for event barrier + * Prefix for assigning names to worker threads */ - private volatile long eventCount; + private final String workerNamePrefix; /** - * Pool number, just for assigning useful names to worker threads + * Sum of per-thread steal counts, updated only when threads are + * idle or terminating. */ - private final int poolNumber; + private volatile long stealCount; /** - * The maximum allowed pool size + * Main pool control -- a long packed with: + * AC: Number of active running workers minus target parallelism (16 bits) + * TC: Number of total workers minus target parallelism (16bits) + * ST: true if pool is terminating (1 bit) + * EC: the wait count of top waiting thread (15 bits) + * ID: ~poolIndex of top of Treiber stack of waiting threads (16 bits) + * + * When convenient, we can extract the upper 32 bits of counts and + * the lower 32 bits of queue state, u = (int)(ctl >>> 32) and e = + * (int)ctl. The ec field is never accessed alone, but always + * together with id and st. The offsets of counts by the target + * parallelism and the positionings of fields makes it possible to + * perform the most common checks via sign tests of fields: When + * ac is negative, there are not enough active workers, when tc is + * negative, there are not enough total workers, when id is + * negative, there is at least one waiting worker, and when e is + * negative, the pool is terminating. To deal with these possibly + * negative fields, we use casts in and out of "short" and/or + * signed shifts to maintain signedness. */ - private volatile int maxPoolSize; + volatile long ctl; - /** - * The desired parallelism level, updated only under workerLock. - */ - private volatile int parallelism; + // bit positions/shifts for fields + private static final int AC_SHIFT = 48; + private static final int TC_SHIFT = 32; + private static final int ST_SHIFT = 31; + private static final int EC_SHIFT = 16; - /** - * True if use local fifo, not default lifo, for local polling - */ - private volatile boolean locallyFifo; + // bounds + private static final int MAX_ID = 0x7fff; // max poolIndex + private static final int SMASK = 0xffff; // mask short bits + private static final int SHORT_SIGN = 1 << 15; + private static final int INT_SIGN = 1 << 31; - /** - * 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 running active count is in low - * word, so need to be modified if this changes - */ - private volatile int workerCounts; + // masks + private static final long STOP_BIT = 0x0001L << ST_SHIFT; + private static final long AC_MASK = ((long)SMASK) << AC_SHIFT; + private static final long TC_MASK = ((long)SMASK) << TC_SHIFT; - 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; } + // units for incrementing and decrementing + private static final long TC_UNIT = 1L << TC_SHIFT; + private static final long AC_UNIT = 1L << AC_SHIFT; + + // masks and units for dealing with u = (int)(ctl >>> 32) + private static final int UAC_SHIFT = AC_SHIFT - 32; + private static final int UTC_SHIFT = TC_SHIFT - 32; + private static final int UAC_MASK = SMASK << UAC_SHIFT; + private static final int UTC_MASK = SMASK << UTC_SHIFT; + private static final int UAC_UNIT = 1 << UAC_SHIFT; + private static final int UTC_UNIT = 1 << UTC_SHIFT; + + // masks and units for dealing with e = (int)ctl + private static final int E_MASK = 0x7fffffff; // no STOP_BIT + private static final int EC_UNIT = 1 << EC_SHIFT; /** - * Add 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 + * The target parallelism level. */ - final void updateRunningCount(int delta) { - int s; - do;while (!casWorkerCounts(s = workerCounts, s + delta)); - } + final int parallelism; /** - * Add 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 + * Index (mod submission queue length) of next element to take + * from submission queue. Usage is identical to that for + * per-worker queues -- see ForkJoinWorkerThread internal + * documentation. */ - 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)); - } + volatile int queueBase; /** - * 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 + * Index (mod submission queue length) of next element to add + * in submission queue. Usage is identical to that for + * per-worker queues -- see ForkJoinWorkerThread internal + * documentation. */ - 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; } + int queueTop; /** - * Try incrementing active count; fail on contention. Called by - * workers before/during executing tasks. - * @return true on success; + * True when shutdown() has been called. */ - final boolean tryIncrementActiveCount() { - int c = runControl; - return casRunControl(c, c+1); - } + volatile boolean shutdown; /** - * Try decrementing active count; fail on contention. - * Possibly trigger termination on success - * Called by workers when they can't find tasks. - * @return true on success + * True if use local fifo, not default lifo, for local polling + * Read by, and replicated by ForkJoinWorkerThreads */ - final boolean tryDecrementActiveCount() { - int c = runControl; - int nextc = c - 1; - if (!casRunControl(c, nextc)) - return false; - if (canTerminateOnShutdown(nextc)) - terminateOnShutdown(); - return true; - } + final boolean locallyFifo; /** - * Return true if argument represents zero active count and - * nonzero runstate, which is the triggering condition for - * terminating on shutdown. + * The number of threads in ForkJoinWorkerThreads.helpQuiescePool. + * When non-zero, suppresses automatic shutdown when active + * counts become zero. */ - private static boolean canTerminateOnShutdown(int c) { - return ((c & -c) >>> 16) != 0; // i.e. least bit is nonzero runState bit - } + volatile int quiescerCount; /** - * Transition run state to at least the given state. Return true - * if not already at least given state. + * The number of threads blocked in join. */ - private boolean transitionRunStateTo(int state) { - for (;;) { - int c = runControl; - if (runStateOf(c) >= state) - return false; - if (casRunControl(c, runControlFor(state, activeCountOf(c)))) - return true; - } - } + volatile int blockedCount; /** - * Controls whether to add spares to maintain parallelism + * Counter for worker Thread names (unrelated to their poolIndex) */ - private volatile boolean maintainsParallelism; - - // Constructors + private volatile int nextWorkerNumber; /** - * Creates a ForkJoinPool with a pool size equal to the number of - * processors available on the system and using the default - * ForkJoinWorkerThreadFactory, - * @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}("modifyThread"), + * The index for the next created worker. Accessed under scanGuard. */ - public ForkJoinPool() { - this(Runtime.getRuntime().availableProcessors(), - defaultForkJoinWorkerThreadFactory); - } + private int nextWorkerIndex; /** - * Creates a ForkJoinPool with the indicated parellelism 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}("modifyThread"), + * SeqLock and index masking for updates to workers array. Locked + * when SG_UNIT is set. Unlocking clears bit by adding + * SG_UNIT. Staleness of read-only operations can be checked by + * comparing scanGuard to value before the reads. The low 16 bits + * (i.e, anding with SMASK) hold (the smallest power of two + * covering all worker indices, minus one, and is used to avoid + * dealing with large numbers of null slots when the workers array + * is overallocated. */ - public ForkJoinPool(int parallelism) { - this(parallelism, defaultForkJoinWorkerThreadFactory); - } + volatile int scanGuard; + + private static final int SG_UNIT = 1 << 16; /** - * 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}("modifyThread"), + * The wakeup interval (in nanoseconds) for a worker waiting for a + * task when the pool is quiescent to instead try to shrink the + * number of workers. The exact value does not matter too + * much. It must be short enough to release resources during + * sustained periods of idleness, but not so short that threads + * are continually re-created. */ - public ForkJoinPool(ForkJoinWorkerThreadFactory factory) { - this(Runtime.getRuntime().availableProcessors(), factory); - } + private static final long SHRINK_RATE = + 4L * 1000L * 1000L * 1000L; // 4 seconds /** - * Creates a ForkJoinPool with the given parallelism and factory. + * Top-level loop for worker threads: On each step: if the + * previous step swept through all queues and found no tasks, or + * there are excess threads, then possibly blocks. Otherwise, + * scans for and, if found, executes a task. Returns when pool + * and/or worker terminate. * - * @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}("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 - } - - /** - * Create new worker using factory. - * @param index the index to assign worker - * @return new worker, or null of factory failed - */ - 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); + final void work(ForkJoinWorkerThread w) { + boolean swept = false; // true on empty scans + long c; + while (!w.terminate && (int)(c = ctl) >= 0) { + int a; // active count + if (!swept && (a = (int)(c >> AC_SHIFT)) <= 0) + swept = scan(w, a); + else if (tryAwaitWork(w, c)) + swept = false; } - return w; } + // Signalling + /** - * Return a good size for worker array given pool size. - * Currently requires size to be a power of two. + * Wakes up or creates a worker. */ - private static int arraySizeFor(int ps) { - return ps <= 1? 1 : (1 << (32 - Integer.numberOfLeadingZeros(ps-1))); - } - - public static ForkJoinWorkerThread[] copyOfWorkers(ForkJoinWorkerThread[] original, int newLength) { - ForkJoinWorkerThread[] copy = new ForkJoinWorkerThread[newLength]; - System.arraycopy(original, 0, copy, 0, Math.min(newLength, original.length)); - return copy; + final void signalWork() { + /* + * The while condition is true if: (there is are too few total + * workers OR there is at least one waiter) AND (there are too + * few active workers OR the pool is terminating). The value + * of e distinguishes the remaining cases: zero (no waiters) + * for create, negative if terminating (in which case do + * nothing), else release a waiter. The secondary checks for + * release (non-null array etc) can fail if the pool begins + * terminating after the test, and don't impose any added cost + * because JVMs must perform null and bounds checks anyway. + */ + long c; int e, u; + while ((((e = (int)(c = ctl)) | (u = (int)(c >>> 32))) & + (INT_SIGN|SHORT_SIGN)) == (INT_SIGN|SHORT_SIGN) && e >= 0) { + if (e > 0) { // release a waiting worker + int i; ForkJoinWorkerThread w; ForkJoinWorkerThread[] ws; + if ((ws = workers) == null || + (i = ~e & SMASK) >= ws.length || + (w = ws[i]) == null) + break; + long nc = (((long)(w.nextWait & E_MASK)) | + ((long)(u + UAC_UNIT) << 32)); + if (w.eventCount == e && + UNSAFE.compareAndSwapLong(this, ctlOffset, c, nc)) { + w.eventCount = (e + EC_UNIT) & E_MASK; + if (w.parked) + UNSAFE.unpark(w); + break; + } + } + else if (UNSAFE.compareAndSwapLong + (this, ctlOffset, c, + (long)(((u + UTC_UNIT) & UTC_MASK) | + ((u + UAC_UNIT) & UAC_MASK)) << 32)) { + addWorker(); + break; + } + } } /** - * Create or resize array if necessary to hold newLength. - * Call only under exlusion or lock - * @return the array - */ - private ForkJoinWorkerThread[] ensureWorkerArrayCapacity(int newLength) { - ForkJoinWorkerThread[] ws = workers; - if (ws == null) - return workers = new ForkJoinWorkerThread[arraySizeFor(newLength)]; - else if (newLength > ws.length) - return workers = copyOfWorkers(ws, arraySizeFor(newLength)); - else - return ws; + * Variant of signalWork to help release waiters on rescans. + * Tries once to release a waiter if active count < 0. + * + * @return false if failed due to contention, else true + */ + private boolean tryReleaseWaiter() { + long c; int e, i; ForkJoinWorkerThread w; ForkJoinWorkerThread[] ws; + if ((e = (int)(c = ctl)) > 0 && + (int)(c >> AC_SHIFT) < 0 && + (ws = workers) != null && + (i = ~e & SMASK) < ws.length && + (w = ws[i]) != null) { + long nc = ((long)(w.nextWait & E_MASK) | + ((c + AC_UNIT) & (AC_MASK|TC_MASK))); + if (w.eventCount != e || + !UNSAFE.compareAndSwapLong(this, ctlOffset, c, nc)) + return false; + w.eventCount = (e + EC_UNIT) & E_MASK; + if (w.parked) + UNSAFE.unpark(w); + } + return true; } + // Scanning for tasks + /** - * Try to shrink workers into smaller array after one or more terminate + * Scans for and, if found, executes one task. Scans start at a + * random index of workers array, and randomly select the first + * (2*#workers)-1 probes, and then, if all empty, resort to 2 + * circular sweeps, which is necessary to check quiescence. and + * taking a submission only if no stealable tasks were found. The + * steal code inside the loop is a specialized form of + * ForkJoinWorkerThread.deqTask, followed bookkeeping to support + * helpJoinTask and signal propagation. The code for submission + * queues is almost identical. On each steal, the worker completes + * not only the task, but also all local tasks that this task may + * have generated. On detecting staleness or contention when + * trying to take a task, this method returns without finishing + * sweep, which allows global state rechecks before retry. + * + * @param w the worker + * @param a the number of active workers + * @return true if swept all queues without finding a task */ - private void tryShrinkWorkerArray() { + private boolean scan(ForkJoinWorkerThread w, int a) { + int g = scanGuard; // mask 0 avoids useless scans if only one active + int m = (parallelism == 1 - a && blockedCount == 0) ? 0 : g & SMASK; 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 = copyOfWorkers(ws, newLength); + if (ws == null || ws.length <= m) // staleness check + return false; + for (int r = w.seed, k = r, j = -(m + m); j <= m + m; ++j) { + ForkJoinTask t; ForkJoinTask[] q; int b, i; + ForkJoinWorkerThread v = ws[k & m]; + if (v != null && (b = v.queueBase) != v.queueTop && + (q = v.queue) != null && (i = (q.length - 1) & b) >= 0) { + long u = (i << ASHIFT) + ABASE; + if ((t = q[i]) != null && v.queueBase == b && + UNSAFE.compareAndSwapObject(q, u, t, null)) { + int d = (v.queueBase = b + 1) - v.queueTop; + v.stealHint = w.poolIndex; + if (d != 0) + signalWork(); // propagate if nonempty + w.execTask(t); + } + r ^= r << 13; r ^= r >>> 17; w.seed = r ^ (r << 5); + return false; // store next seed + } + else if (j < 0) { // xorshift + r ^= r << 13; r ^= r >>> 17; k = r ^= r << 5; + } + else + ++k; + } + if (scanGuard != g) // staleness check + return false; + else { // try to take submission + ForkJoinTask t; ForkJoinTask[] q; int b, i; + if ((b = queueBase) != queueTop && + (q = submissionQueue) != null && + (i = (q.length - 1) & b) >= 0) { + long u = (i << ASHIFT) + ABASE; + if ((t = q[i]) != null && queueBase == b && + UNSAFE.compareAndSwapObject(q, u, t, null)) { + queueBase = b + 1; + w.execTask(t); + } + return false; + } + return true; // all queues empty } } /** - * Initialize workers if necessary - */ - 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); + * Tries to enqueue worker w in wait queue and await change in + * worker's eventCount. If the pool is quiescent and there is + * more than one worker, possibly terminates worker upon exit. + * Otherwise, before blocking, rescans queues to avoid missed + * signals. Upon finding work, releases at least one worker + * (which may be the current worker). Rescans restart upon + * detected staleness or failure to release due to + * contention. Note the unusual conventions about Thread.interrupt + * here and elsewhere: Because interrupts are used solely to alert + * threads to check termination, which is checked here anyway, we + * clear status (using Thread.interrupted) before any call to + * park, so that park does not immediately return due to status + * being set via some other unrelated call to interrupt in user + * code. + * + * @param w the calling worker + * @param c the ctl value on entry + * @return true if waited or another thread was released upon enq + */ + private boolean tryAwaitWork(ForkJoinWorkerThread w, long c) { + int v = w.eventCount; + w.nextWait = (int)c; // w's successor record + long nc = (long)(v & E_MASK) | ((c - AC_UNIT) & (AC_MASK|TC_MASK)); + if (ctl != c || !UNSAFE.compareAndSwapLong(this, ctlOffset, c, nc)) { + long d = ctl; // return true if lost to a deq, to force scan + return (int)d != (int)c && ((d - c) & AC_MASK) >= 0L; + } + for (int sc = w.stealCount; sc != 0;) { // accumulate stealCount + long s = stealCount; + if (UNSAFE.compareAndSwapLong(this, stealCountOffset, s, s + sc)) + sc = w.stealCount = 0; + else if (w.eventCount != v) + return true; // update next time + } + if ((!shutdown || !tryTerminate(false)) && + (int)c != 0 && parallelism + (int)(nc >> AC_SHIFT) == 0 && + blockedCount == 0 && quiescerCount == 0) + idleAwaitWork(w, nc, c, v); // quiescent + for (boolean rescanned = false;;) { + if (w.eventCount != v) + return true; + if (!rescanned) { + int g = scanGuard, m = g & SMASK; + ForkJoinWorkerThread[] ws = workers; + if (ws != null && m < ws.length) { + rescanned = true; + for (int i = 0; i <= m; ++i) { + ForkJoinWorkerThread u = ws[i]; + if (u != null) { + if (u.queueBase != u.queueTop && + !tryReleaseWaiter()) + rescanned = false; // contended + if (w.eventCount != v) + return true; } } } - } finally { - lock.unlock(); + if (scanGuard != g || // stale + (queueBase != queueTop && !tryReleaseWaiter())) + rescanned = false; + if (!rescanned) + Thread.yield(); // reduce contention + else + Thread.interrupted(); // clear before park + } + else { + w.parked = true; // must recheck + if (w.eventCount != v) { + w.parked = false; + return true; + } + LockSupport.park(this); + rescanned = w.parked = false; } } } /** - * Worker creation and startup for threads added via setParallelism. - */ - private void createAndStartAddedWorkers() { - resumeAllSpares(); // Allow spares to convert to nonspare - int ps = parallelism; - ForkJoinWorkerThread[] ws = ensureWorkerArrayCapacity(ps); - int len = ws.length; - // Sweep through slots, to keep lowest indices most populated - int k = 0; - while (k < len) { - if (ws[k] != null) { - ++k; - continue; - } - int s = workerCounts; - int tc = totalCountOf(s); - int rc = runningCountOf(s); - if (rc >= ps || tc >= ps) - break; - if (casWorkerCounts (s, workerCountsFor(tc+1, rc+1))) { - ForkJoinWorkerThread w = createWorker(k); - if (w != null) { - ws[k++] = w; - w.start(); - } - else { - updateWorkerCount(-1); // back out on failed creation + * If inactivating worker w has caused pool to become + * quiescent, check for pool termination, and wait for event + * for up to SHRINK_RATE nanosecs (rescans are unnecessary in + * this case because quiescence reflects consensus about lack + * of work). On timeout, if ctl has not changed, terminate the + * worker. Upon its termination (see deregisterWorker), it may + * wake up another worker to possibly repeat this process. + * + * @param w the calling worker + * @param currentCtl the ctl value after enqueuing w + * @param prevCtl the ctl value if w terminated + * @param v the eventCount w awaits change + */ + private void idleAwaitWork(ForkJoinWorkerThread w, long currentCtl, + long prevCtl, int v) { + if (w.eventCount == v) { + if (shutdown) + tryTerminate(false); + ForkJoinTask.helpExpungeStaleExceptions(); // help clean weak refs + while (ctl == currentCtl) { + long startTime = System.nanoTime(); + w.parked = true; + if (w.eventCount == v) // must recheck + LockSupport.parkNanos(this, SHRINK_RATE); + w.parked = false; + if (w.eventCount != v) + break; + else if (System.nanoTime() - startTime < + SHRINK_RATE - (SHRINK_RATE / 10)) // timing slop + Thread.interrupted(); // spurious wakeup + else if (UNSAFE.compareAndSwapLong(this, ctlOffset, + currentCtl, prevCtl)) { + w.terminate = true; // restore previous + w.eventCount = ((int)currentCtl + EC_UNIT) & E_MASK; break; } } } } - // Execution methods + // Submissions /** - * Common code for execute, invoke and submit + * Enqueues the given task in the submissionQueue. Same idea as + * ForkJoinWorkerThread.pushTask except for use of submissionLock. + * + * @param t the task */ - private void doSubmit(ForkJoinTask task) { - if (isShutdown()) - throw new RejectedExecutionException(); - if (workers == null) - ensureWorkerInitialization(); - submissionQueue.offer(task); - signalIdleWorkers(); + private void addSubmission(ForkJoinTask t) { + final ReentrantLock lock = this.submissionLock; + lock.lock(); + try { + ForkJoinTask[] q; int s, m; + if ((q = submissionQueue) != null) { // ignore if queue removed + long u = (((s = queueTop) & (m = q.length-1)) << ASHIFT)+ABASE; + UNSAFE.putOrderedObject(q, u, t); + queueTop = s + 1; + if (s - queueBase == m) + growSubmissionQueue(); + } + } finally { + lock.unlock(); + } + signalWork(); + } + + // (pollSubmission is defined below with exported methods) + + /** + * Creates or doubles submissionQueue array. + * Basically identical to ForkJoinWorkerThread version. + */ + private void growSubmissionQueue() { + ForkJoinTask[] oldQ = submissionQueue; + int size = oldQ != null ? oldQ.length << 1 : INITIAL_QUEUE_CAPACITY; + if (size > MAXIMUM_QUEUE_CAPACITY) + throw new RejectedExecutionException("Queue capacity exceeded"); + if (size < INITIAL_QUEUE_CAPACITY) + size = INITIAL_QUEUE_CAPACITY; + ForkJoinTask[] q = submissionQueue = new ForkJoinTask[size]; + int mask = size - 1; + int top = queueTop; + int oldMask; + if (oldQ != null && (oldMask = oldQ.length - 1) >= 0) { + for (int b = queueBase; b != top; ++b) { + long u = ((b & oldMask) << ASHIFT) + ABASE; + Object x = UNSAFE.getObjectVolatile(oldQ, u); + if (x != null && UNSAFE.compareAndSwapObject(oldQ, u, x, null)) + UNSAFE.putObjectVolatile + (q, ((b & mask) << ASHIFT) + ABASE, x); + } + } } + // Blocking support + /** - * Performs the given task; returning its result upon completion - * @param task the task - * @return the task's result - * @throws NullPointerException if task is null - * @throws RejectedExecutionException if pool is shut down - */ - public T invoke(ForkJoinTask task) { - doSubmit(task); - return task.join(); + * Tries to increment blockedCount, decrement active count + * (sometimes implicitly) and possibly release or create a + * compensating worker in preparation for blocking. Fails + * on contention or termination. + * + * @return true if the caller can block, else should recheck and retry + */ + private boolean tryPreBlock() { + int b = blockedCount; + if (UNSAFE.compareAndSwapInt(this, blockedCountOffset, b, b + 1)) { + int pc = parallelism; + do { + ForkJoinWorkerThread[] ws; ForkJoinWorkerThread w; + int e, ac, tc, rc, i; + long c = ctl; + int u = (int)(c >>> 32); + if ((e = (int)c) < 0) { + // skip -- terminating + } + else if ((ac = (u >> UAC_SHIFT)) <= 0 && e != 0 && + (ws = workers) != null && + (i = ~e & SMASK) < ws.length && + (w = ws[i]) != null) { + long nc = ((long)(w.nextWait & E_MASK) | + (c & (AC_MASK|TC_MASK))); + if (w.eventCount == e && + UNSAFE.compareAndSwapLong(this, ctlOffset, c, nc)) { + w.eventCount = (e + EC_UNIT) & E_MASK; + if (w.parked) + UNSAFE.unpark(w); + return true; // release an idle worker + } + } + else if ((tc = (short)(u >>> UTC_SHIFT)) >= 0 && ac + pc > 1) { + long nc = ((c - AC_UNIT) & AC_MASK) | (c & ~AC_MASK); + if (UNSAFE.compareAndSwapLong(this, ctlOffset, c, nc)) + return true; // no compensation needed + } + else if (tc + pc < MAX_ID) { + long nc = ((c + TC_UNIT) & TC_MASK) | (c & ~TC_MASK); + if (UNSAFE.compareAndSwapLong(this, ctlOffset, c, nc)) { + addWorker(); + return true; // create a replacement + } + } + // try to back out on any failure and let caller retry + } while (!UNSAFE.compareAndSwapInt(this, blockedCountOffset, + b = blockedCount, b - 1)); + } + return false; } /** - * Arranges for (asynchronous) execution of the given task. - * @param task the task - * @throws NullPointerException if task is null - * @throws RejectedExecutionException if pool is shut down + * Decrements blockedCount and increments active count */ - public void execute(ForkJoinTask task) { - doSubmit(task); + private void postBlock() { + long c; + do {} while (!UNSAFE.compareAndSwapLong(this, ctlOffset, // no mask + c = ctl, c + AC_UNIT)); + int b; + do {} while (!UNSAFE.compareAndSwapInt(this, blockedCountOffset, + b = blockedCount, b - 1)); } - // AbstractExecutorService methods - - public void execute(Runnable task) { - doSubmit(new AdaptedRunnable(task, null)); + /** + * Possibly blocks waiting for the given task to complete, or + * cancels the task if terminating. Fails to wait if contended. + * + * @param joinMe the task + */ + final void tryAwaitJoin(ForkJoinTask joinMe) { + int s; + Thread.interrupted(); // clear interrupts before checking termination + if (joinMe.status >= 0) { + if (tryPreBlock()) { + joinMe.tryAwaitDone(0L); + postBlock(); + } + else if ((ctl & STOP_BIT) != 0L) + joinMe.cancelIgnoringExceptions(); + } } - public ForkJoinTask submit(Callable task) { - ForkJoinTask job = new AdaptedCallable(task); - doSubmit(job); - return job; + /** + * Possibly blocks the given worker waiting for joinMe to + * complete or timeout + * + * @param joinMe the task + * @param millis the wait time for underlying Object.wait + */ + final void timedAwaitJoin(ForkJoinTask joinMe, long nanos) { + while (joinMe.status >= 0) { + Thread.interrupted(); + if ((ctl & STOP_BIT) != 0L) { + joinMe.cancelIgnoringExceptions(); + break; + } + if (tryPreBlock()) { + long last = System.nanoTime(); + while (joinMe.status >= 0) { + long millis = TimeUnit.NANOSECONDS.toMillis(nanos); + if (millis <= 0) + break; + joinMe.tryAwaitDone(millis); + if (joinMe.status < 0) + break; + if ((ctl & STOP_BIT) != 0L) { + joinMe.cancelIgnoringExceptions(); + break; + } + long now = System.nanoTime(); + nanos -= now - last; + last = now; + } + postBlock(); + break; + } + } } - public ForkJoinTask submit(Runnable task, T result) { - ForkJoinTask job = new AdaptedRunnable(task, result); - doSubmit(job); - return job; + /** + * If necessary, compensates for blocker, and blocks + */ + private void awaitBlocker(ManagedBlocker blocker) + throws InterruptedException { + while (!blocker.isReleasable()) { + if (tryPreBlock()) { + try { + do {} while (!blocker.isReleasable() && !blocker.block()); + } finally { + postBlock(); + } + break; + } + } } - public ForkJoinTask submit(Runnable task) { - ForkJoinTask job = new AdaptedRunnable(task, null); - doSubmit(job); - return job; - } + // Creating, registering and deregistring workers /** - * Adaptor for Runnables. This implements RunnableFuture - * to be compliant with AbstractExecutorService constraints + * Tries to create and start a worker; minimally rolls back counts + * on failure. */ - 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; + private void addWorker() { + Throwable ex = null; + ForkJoinWorkerThread t = null; + try { + t = factory.newThread(this); + } catch (Throwable e) { + ex = e; } - public T getRawResult() { return result; } - public void setRawResult(T v) { result = v; } - public boolean exec() { - runnable.run(); - result = resultOnCompletion; - return true; + if (t == null) { // null or exceptional factory return + long c; // adjust counts + do {} while (!UNSAFE.compareAndSwapLong + (this, ctlOffset, c = ctl, + (((c - AC_UNIT) & AC_MASK) | + ((c - TC_UNIT) & TC_MASK) | + (c & ~(AC_MASK|TC_MASK))))); + // Propagate exception if originating from an external caller + if (!tryTerminate(false) && ex != null && + !(Thread.currentThread() instanceof ForkJoinWorkerThread)) + UNSAFE.throwException(ex); } - public void run() { invoke(); } + else + t.start(); } /** - * Adaptor for Callables + * Callback from ForkJoinWorkerThread constructor to assign a + * public name */ - 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); - } + final String nextWorkerName() { + for (int n;;) { + if (UNSAFE.compareAndSwapInt(this, nextWorkerNumberOffset, + n = nextWorkerNumber, ++n)) + return workerNamePrefix + n; } - public void run() { invoke(); } - } - - public List> invokeAll(Collection> tasks) { - ArrayList> ts = - new ArrayList>(tasks.size()); - for (Callable c : tasks) - ts.add(new AdaptedCallable(c)); - invoke(new InvokeAll(ts)); - return (List>)(List)ts; } - static final class InvokeAll extends RecursiveAction { - final ArrayList> tasks; - InvokeAll(ArrayList> tasks) { this.tasks = tasks; } - public void compute() { - try { invokeAll(tasks); } catch(Exception ignore) {} + /** + * Callback from ForkJoinWorkerThread constructor to + * determine its poolIndex and record in workers array. + * + * @param w the worker + * @return the worker's pool index + */ + final int registerWorker(ForkJoinWorkerThread w) { + /* + * In the typical case, a new worker acquires the lock, uses + * next available index and returns quickly. Since we should + * not block callers (ultimately from signalWork or + * tryPreBlock) waiting for the lock needed to do this, we + * instead help release other workers while waiting for the + * lock. + */ + for (int g;;) { + ForkJoinWorkerThread[] ws; + if (((g = scanGuard) & SG_UNIT) == 0 && + UNSAFE.compareAndSwapInt(this, scanGuardOffset, + g, g | SG_UNIT)) { + int k = nextWorkerIndex; + try { + if ((ws = workers) != null) { // ignore on shutdown + int n = ws.length; + if (k < 0 || k >= n || ws[k] != null) { + for (k = 0; k < n && ws[k] != null; ++k) + ; + if (k == n) + ws = workers = Arrays.copyOf(ws, n << 1); + } + ws[k] = w; + nextWorkerIndex = k + 1; + int m = g & SMASK; + g = (k > m) ? ((m << 1) + 1) & SMASK : g + (SG_UNIT<<1); + } + } finally { + scanGuard = g; + } + return k; + } + else if ((ws = workers) != null) { // help release others + for (ForkJoinWorkerThread u : ws) { + if (u != null && u.queueBase != u.queueTop) { + if (tryReleaseWaiter()) + break; + } + } + } } } - // Configuration and status settings and queries - /** - * Returns the factory used for constructing new workers + * Final callback from terminating worker. Removes record of + * worker from array, and adjusts counts. If pool is shutting + * down, tries to complete termination. * - * @return the factory used for constructing new workers + * @param w the worker */ - public ForkJoinWorkerThreadFactory getFactory() { - return factory; + final void deregisterWorker(ForkJoinWorkerThread w, Throwable ex) { + int idx = w.poolIndex; + int sc = w.stealCount; + int steps = 0; + // Remove from array, adjust worker counts and collect steal count. + // We can intermix failed removes or adjusts with steal updates + do { + long s, c; + int g; + if (steps == 0 && ((g = scanGuard) & SG_UNIT) == 0 && + UNSAFE.compareAndSwapInt(this, scanGuardOffset, + g, g |= SG_UNIT)) { + ForkJoinWorkerThread[] ws = workers; + if (ws != null && idx >= 0 && + idx < ws.length && ws[idx] == w) + ws[idx] = null; // verify + nextWorkerIndex = idx; + scanGuard = g + SG_UNIT; + steps = 1; + } + if (steps == 1 && + UNSAFE.compareAndSwapLong(this, ctlOffset, c = ctl, + (((c - AC_UNIT) & AC_MASK) | + ((c - TC_UNIT) & TC_MASK) | + (c & ~(AC_MASK|TC_MASK))))) + steps = 2; + if (sc != 0 && + UNSAFE.compareAndSwapLong(this, stealCountOffset, + s = stealCount, s + sc)) + sc = 0; + } while (steps != 2 || sc != 0); + if (!tryTerminate(false)) { + if (ex != null) // possibly replace if died abnormally + signalWork(); + else + tryReleaseWaiter(); + } } + // Shutdown and termination + /** - * Returns the handler for internal worker threads that terminate - * due to unrecoverable errors encountered while executing tasks. - * @return the handler, or null if none + * 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 */ - public Thread.UncaughtExceptionHandler getUncaughtExceptionHandler() { - Thread.UncaughtExceptionHandler h; - final ReentrantLock lock = this.workerLock; - lock.lock(); - try { - h = ueh; - } finally { - lock.unlock(); + private boolean tryTerminate(boolean now) { + long c; + while (((c = ctl) & STOP_BIT) == 0) { + if (!now) { + if ((int)(c >> AC_SHIFT) != -parallelism) + return false; + if (!shutdown || blockedCount != 0 || quiescerCount != 0 || + queueBase != queueTop) { + if (ctl == c) // staleness check + return false; + continue; + } + } + if (UNSAFE.compareAndSwapLong(this, ctlOffset, c, c | STOP_BIT)) + startTerminating(); } - return h; + if ((short)(c >>> TC_SHIFT) == -parallelism) { // signal when 0 workers + final ReentrantLock lock = this.submissionLock; + lock.lock(); + try { + termination.signalAll(); + } finally { + lock.unlock(); + } + } + return true; } /** - * Sets the handler for internal worker threads that terminate due - * to unrecoverable errors encountered while executing tasks. - * Unless set, the current default or ThreadGroup handler is used - * as handler. - * - * @param h the new handler - * @return the old handler, or 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}("modifyThread"), + * Runs up to three passes through workers: (0) Setting + * termination status for each worker, followed by wakeups up to + * queued workers; (1) helping cancel tasks; (2) interrupting + * lagging threads (likely in external tasks, but possibly also + * blocked in joins). Each pass repeats previous steps because of + * potential lagging thread creation. */ - public Thread.UncaughtExceptionHandler - setUncaughtExceptionHandler(Thread.UncaughtExceptionHandler h) { - checkPermission(); - Thread.UncaughtExceptionHandler old = null; - final ReentrantLock lock = this.workerLock; - lock.lock(); - try { - old = ueh; - ueh = h; + private void startTerminating() { + cancelSubmissions(); + for (int pass = 0; pass < 3; ++pass) { ForkJoinWorkerThread[] ws = workers; if (ws != null) { - for (int i = 0; i < ws.length; ++i) { - ForkJoinWorkerThread w = ws[i]; - if (w != null) - w.setUncaughtExceptionHandler(h); + for (ForkJoinWorkerThread w : ws) { + if (w != null) { + w.terminate = true; + if (pass > 0) { + w.cancelTasks(); + if (pass > 1 && !w.isInterrupted()) { + try { + w.interrupt(); + } catch (SecurityException ignore) { + } + } + } + } + } + terminateWaiters(); + } + } + } + + /** + * Polls and cancels all submissions. Called only during termination. + */ + private void cancelSubmissions() { + while (queueBase != queueTop) { + ForkJoinTask task = pollSubmission(); + if (task != null) { + try { + task.cancel(false); + } catch (Throwable ignore) { + } + } + } + } + + /** + * Tries to set the termination status of waiting workers, and + * then wakes them up (after which they will terminate). + */ + private void terminateWaiters() { + ForkJoinWorkerThread[] ws = workers; + if (ws != null) { + ForkJoinWorkerThread w; long c; int i, e; + int n = ws.length; + while ((i = ~(e = (int)(c = ctl)) & SMASK) < n && + (w = ws[i]) != null && w.eventCount == (e & E_MASK)) { + if (UNSAFE.compareAndSwapLong(this, ctlOffset, c, + (long)(w.nextWait & E_MASK) | + ((c + AC_UNIT) & AC_MASK) | + (c & (TC_MASK|STOP_BIT)))) { + w.terminate = true; + w.eventCount = e + EC_UNIT; + if (w.parked) + UNSAFE.unpark(w); } } - } finally { - lock.unlock(); } - return old; } + // misc ForkJoinWorkerThread support /** - * Sets the target paralleism level of this pool. - * @param parallelism the target parallelism + * Increment or decrement quiescerCount. Needed only to prevent + * triggering shutdown if a worker is transiently inactive while + * checking quiescence. + * + * @param delta 1 for increment, -1 for decrement + */ + final void addQuiescerCount(int delta) { + int c; + do {} while (!UNSAFE.compareAndSwapInt(this, quiescerCountOffset, + c = quiescerCount, c + delta)); + } + + /** + * Directly increment or decrement active count without + * queuing. This method is used to transiently assert inactivation + * while checking quiescence. + * + * @param delta 1 for increment, -1 for decrement + */ + final void addActiveCount(int delta) { + long d = delta < 0 ? -AC_UNIT : AC_UNIT; + long c; + do {} while (!UNSAFE.compareAndSwapLong(this, ctlOffset, c = ctl, + ((c + d) & AC_MASK) | + (c & ~AC_MASK))); + } + + /** + * Returns the approximate (non-atomic) number of idle threads per + * active thread. + */ + final int idlePerActive() { + // Approximate at powers of two for small values, saturate past 4 + int p = parallelism; + int a = p + (int)(ctl >> AC_SHIFT); + return (a > (p >>>= 1) ? 0 : + a > (p >>>= 1) ? 1 : + a > (p >>>= 1) ? 2 : + a > (p >>>= 1) ? 4 : + 8); + } + + // Exported 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 maximum size bounds. + * 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}("modifyThread"), + * java.lang.RuntimePermission}{@code ("modifyThread")} */ - public void setParallelism(int parallelism) { + 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 {@code 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 {@code 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 (parallelism <= 0 || parallelism > maxPoolSize) + if (factory == null) + throw new NullPointerException(); + if (parallelism <= 0 || parallelism > MAX_ID) 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(); + this.parallelism = parallelism; + this.factory = factory; + this.ueh = handler; + this.locallyFifo = asyncMode; + long np = (long)(-parallelism); // offset ctl counts + this.ctl = ((np << AC_SHIFT) & AC_MASK) | ((np << TC_SHIFT) & TC_MASK); + this.submissionQueue = new ForkJoinTask[INITIAL_QUEUE_CAPACITY]; + // initialize workers array with room for 2*parallelism if possible + int n = parallelism << 1; + if (n >= MAX_ID) + n = MAX_ID; + else { // See Hackers Delight, sec 3.2, where n < (1 << 16) + n |= n >>> 1; n |= n >>> 2; n |= n >>> 4; n |= n >>> 8; } - signalIdleWorkers(); + workers = new ForkJoinWorkerThread[n + 1]; + this.submissionLock = new ReentrantLock(); + this.termination = submissionLock.newCondition(); + StringBuilder sb = new StringBuilder("ForkJoinPool-"); + sb.append(poolNumberGenerator.incrementAndGet()); + sb.append("-worker-"); + this.workerNamePrefix = sb.toString(); } + // Execution methods + /** - * Returns the targeted number of worker threads in this pool. + * Performs the given task, returning its result upon completion. + * If the computation encounters an unchecked Exception or Error, + * it is rethrown as the outcome of this invocation. Rethrown + * exceptions behave in the same way as regular exceptions, but, + * when possible, contain stack traces (as displayed for example + * using {@code ex.printStackTrace()}) of both the current thread + * as well as the thread actually encountering the exception; + * minimally only the latter. * - * @return the targeted number of worker threads in this pool + * @param task the task + * @return the task's result + * @throws NullPointerException if the task is null + * @throws RejectedExecutionException if the task cannot be + * scheduled for execution */ - public int getParallelism() { - return parallelism; + public T invoke(ForkJoinTask task) { + Thread t = Thread.currentThread(); + if (task == null) + throw new NullPointerException(); + if (shutdown) + throw new RejectedExecutionException(); + if ((t instanceof ForkJoinWorkerThread) && + ((ForkJoinWorkerThread)t).pool == this) + return task.invoke(); // bypass submit if in same pool + else { + addSubmission(task); + return task.join(); + } } /** - * Returns the number of worker threads that have started but not - * yet terminated. This result returned by this method may differ - * from getParallelism when threads are created to - * maintain parallelism when others are cooperatively blocked. + * Unless terminating, forks task if within an ongoing FJ + * computation in the current pool, else submits as external task. + */ + private void forkOrSubmit(ForkJoinTask task) { + ForkJoinWorkerThread w; + Thread t = Thread.currentThread(); + if (shutdown) + throw new RejectedExecutionException(); + if ((t instanceof ForkJoinWorkerThread) && + (w = (ForkJoinWorkerThread)t).pool == this) + w.pushTask(task); + else + addSubmission(task); + } + + /** + * Arranges for (asynchronous) execution of the given task. * - * @return the number of worker threads + * @param task the task + * @throws NullPointerException if the task is null + * @throws RejectedExecutionException if the task cannot be + * scheduled for execution */ - public int getPoolSize() { - return totalCountOf(workerCounts); + public void execute(ForkJoinTask task) { + if (task == null) + throw new NullPointerException(); + forkOrSubmit(task); } + // AbstractExecutorService methods + /** - * Returns the maximum number of threads allowed to exist in the - * pool, even if there are insufficient unblocked running threads. - * @return the maximum + * @throws NullPointerException if the task is null + * @throws RejectedExecutionException if the task cannot be + * scheduled for execution */ - public int getMaximumPoolSize() { - return maxPoolSize; + public void execute(Runnable task) { + if (task == null) + throw new NullPointerException(); + ForkJoinTask job; + if (task instanceof ForkJoinTask) // avoid re-wrap + job = (ForkJoinTask) task; + else + job = ForkJoinTask.adapt(task, null); + forkOrSubmit(job); + } + + /** + * Submits a ForkJoinTask for execution. + * + * @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) { + if (task == null) + throw new NullPointerException(); + forkOrSubmit(task); + return task; } /** - * 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. + * @throws NullPointerException if the task is null + * @throws RejectedExecutionException if the task cannot be + * scheduled for execution */ - public void setMaximumPoolSize(int newMax) { - if (newMax < 0 || newMax > MAX_THREADS) - throw new IllegalArgumentException(); - maxPoolSize = newMax; + public ForkJoinTask submit(Callable task) { + if (task == null) + throw new NullPointerException(); + ForkJoinTask job = ForkJoinTask.adapt(task); + forkOrSubmit(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) { + if (task == null) + throw new NullPointerException(); + ForkJoinTask job = ForkJoinTask.adapt(task, result); + forkOrSubmit(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) { + if (task == null) + throw new NullPointerException(); + ForkJoinTask job; + if (task instanceof ForkJoinTask) // avoid re-wrap + job = (ForkJoinTask) task; + else + job = ForkJoinTask.adapt(task, null); + forkOrSubmit(job); + return job; + } /** - * 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 + * @throws NullPointerException {@inheritDoc} + * @throws RejectedExecutionException {@inheritDoc} */ - public boolean getMaintainsParallelism() { - return maintainsParallelism; + public List> invokeAll(Collection> tasks) { + ArrayList> forkJoinTasks = + new ArrayList>(tasks.size()); + 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 { + final ArrayList> tasks; + InvokeAll(ArrayList> tasks) { this.tasks = tasks; } + public void compute() { + try { invokeAll(tasks); } + catch (Exception ignore) {} + } + private static final long serialVersionUID = -7914297376763021607L; } /** - * 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 + * Returns the factory used for constructing new workers. + * + * @return the factory used for constructing new workers */ - public void setMaintainsParallelism(boolean enable) { - maintainsParallelism = enable; + public ForkJoinWorkerThreadFactory getFactory() { + return factory; } /** - * 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 ather times may be unpredictable. + * Returns the handler for internal worker threads that terminate + * due to unrecoverable errors encountered while executing tasks. * - * @param async if true, use locally FIFO scheduling - * @return the previous mode. + * @return the handler, or {@code null} if none */ - 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; + public Thread.UncaughtExceptionHandler getUncaughtExceptionHandler() { + return ueh; } /** - * Returns true if this pool uses local first-in-first-out + * Returns the targeted parallelism level of this pool. + * + * @return the targeted parallelism level of this pool + */ + public int getParallelism() { + return parallelism; + } + + /** + * Returns the number of worker threads that have started but not + * yet terminated. The result returned by this method may differ + * 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 parallelism + (short)(ctl >>> TC_SHIFT); + } + + /** + * 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; @@ -861,47 +1685,41 @@ public class ForkJoinPool /*extends AbstractExecutorService*/ { /** * 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); + int r = parallelism + (int)(ctl >> AC_SHIFT); + return (r <= 0) ? 0 : r; // suppress momentarily negative values } /** * Returns an estimate of the number of threads that are currently * stealing or executing tasks. This method may overestimate the * number of active threads. - * @return the number of active threads. + * + * @return the number of active threads */ public int getActiveThreadCount() { - return activeCountOf(runControl); + int r = parallelism + (int)(ctl >> AC_SHIFT) + blockedCount; + return (r <= 0) ? 0 : r; // suppress momentarily negative values } /** - * 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; - } - - /** - * 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. - * @return true if all threads are currently idle + * 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 {@code true} if all threads are currently idle */ public boolean isQuiescent() { - return activeCountOf(runControl) == 0; + return parallelism + (int)(ctl >> AC_SHIFT) + blockedCount == 0; } /** @@ -909,23 +1727,14 @@ public class ForkJoinPool /*extends AbstractExecutorService*/ { * one thread's work queue by another. The reported value * underestimates the actual total number of steals when the pool * is not quiescent. This value may be useful for monitoring and - * tuning fork/join programs: In general, steal counts should be + * tuning fork/join programs: in general, steal counts should be * high enough to keep threads busy, but low enough to avoid * overhead and contention across threads. - * @return the number of steals. + * + * @return the number of steals */ public long getStealCount() { - return stealCount.get(); - } - - /** - * Accumulate 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; } /** @@ -935,77 +1744,99 @@ public class ForkJoinPool /*extends AbstractExecutorService*/ { * an approximation, obtained by iterating across all threads in * the pool. This method may be useful for tuning task * granularities. - * @return the number of queued tasks. + * + * @return the number of queued tasks */ 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(); - } + ForkJoinWorkerThread[] ws; + if ((short)(ctl >>> TC_SHIFT) > -parallelism && + (ws = workers) != null) { + for (ForkJoinWorkerThread w : ws) + if (w != null) + count -= w.queueBase - w.queueTop; // must read base first } return count; } /** - * Returns an estimate of the number 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. + * Returns an estimate of the number of tasks submitted to this + * pool that have not yet begun executing. This method may take + * time proportional to the number of submissions. + * + * @return the number of queued submissions */ public int getQueuedSubmissionCount() { - return submissionQueue.size(); + return -queueBase + queueTop; } /** - * Returns true if there are any tasks submitted to this pool - * that have not yet begun executing. - * @return true if there are any queued submissions. + * 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 */ public boolean hasQueuedSubmissions() { - return !submissionQueue.isEmpty(); + return queueBase != queueTop; } /** * Removes and returns the next unexecuted submission if one is * 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(); + ForkJoinTask t; ForkJoinTask[] q; int b, i; + while ((b = queueBase) != queueTop && + (q = submissionQueue) != null && + (i = (q.length - 1) & b) >= 0) { + long u = (i << ASHIFT) + ABASE; + if ((t = q[i]) != null && + queueBase == b && + UNSAFE.compareAndSwapObject(q, u, t, null)) { + queueBase = b + 1; + return t; + } + } + return null; } /** * 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 - * artifically generated or wrapped tasks. This method id 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 c may result in elements being in + * to collection {@code c} may result in elements being in * neither, either or both collections when the associated * exception is thrown. The behavior of this operation is * undefined if the specified collection is modified while the * operation is in progress. + * * @param c the collection to transfer elements into * @return the number of elements transferred */ - 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); + protected int drainTasksTo(Collection> c) { + int count = 0; + while (queueBase != queueTop) { + ForkJoinTask t = pollSubmission(); + if (t != null) { + c.add(t); + ++count; } } - return n; + ForkJoinWorkerThread[] ws; + if ((short)(ctl >>> TC_SHIFT) > -parallelism && + (ws = workers) != null) { + for (ForkJoinWorkerThread w : ws) + if (w != null) + count += w.drainTasksTo(c); + } + return count; } /** @@ -1016,101 +1847,118 @@ public class ForkJoinPool /*extends AbstractExecutorService*/ { * @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 pc = parallelism; + long c = ctl; + int tc = pc + (short)(c >>> TC_SHIFT); + int rc = pc + (int)(c >> AC_SHIFT); + if (rc < 0) // ignore transient negative + rc = 0; + int ac = rc + blockedCount; + String level; + if ((c & STOP_BIT) != 0) + level = (tc == 0) ? "Terminated" : "Terminating"; + else + level = shutdown ? "Shutting down" : "Running"; return super.toString() + - "[" + runStateToString(runStateOf(rc)) + - ", parallelism = " + ps + - ", size = " + totalCountOf(wc) + - ", active = " + activeCountOf(rc) + - ", running = " + runningCountOf(wc) + + "[" + level + + ", parallelism = " + pc + + ", size = " + tc + + ", active = " + ac + + ", running = " + rc + ", steals = " + st + ", tasks = " + qt + ", submissions = " + qs + "]"; } - 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"); - } - } - - // lifecycle control - /** * Initiates an orderly shutdown in which previously submitted * tasks are executed, but no new tasks will be accepted. * Invocation has no additional effect if already shut down. * Tasks that are in the process of being submitted concurrently * during the course of this method may or may not be rejected. + * * @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}("modifyThread"), + * java.lang.RuntimePermission}{@code ("modifyThread")} */ public void shutdown() { checkPermission(); - transitionRunStateTo(SHUTDOWN); - if (canTerminateOnShutdown(runControl)) - terminateOnShutdown(); + shutdown = true; + 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 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 * the caller is not permitted to modify threads * because it does not hold {@link - * java.lang.RuntimePermission}("modifyThread"), + * java.lang.RuntimePermission}{@code ("modifyThread")} */ public List shutdownNow() { checkPermission(); - terminate(); + shutdown = true; + tryTerminate(true); return Collections.emptyList(); } /** - * Returns true if all tasks have completed following shut down. + * Returns {@code true} if all tasks have completed following shut down. * - * @return true if all tasks have completed following shut down + * @return {@code true} if all tasks have completed following shut down */ public boolean isTerminated() { - return runStateOf(runControl) == TERMINATED; + long c = ctl; + return ((c & STOP_BIT) != 0L && + (short)(c >>> TC_SHIFT) == -parallelism); } /** - * Returns true if the process of termination has - * commenced but possibly not yet completed. + * Returns {@code true} if the process of termination has + * commenced but not yet completed. This method may be useful for + * debugging. A return of {@code true} reported a sufficient + * period after shutdown may indicate that submitted tasks have + * ignored or suppressed interruption, or are waiting for IO, + * causing this executor not to properly terminate. (See the + * advisory notes for class {@link ForkJoinTask} stating that + * tasks should not normally entail blocking operations. But if + * they do, they must abort them on interrupt.) * - * @return true if terminating + * @return {@code true} if terminating but not yet terminated */ public boolean isTerminating() { - return runStateOf(runControl) >= TERMINATING; + long c = ctl; + return ((c & STOP_BIT) != 0L && + (short)(c >>> TC_SHIFT) != -parallelism); } /** - * Returns true if this pool has been shut down. + * Returns true if terminating or terminated. Used by ForkJoinWorkerThread. + */ + final boolean isAtLeastTerminating() { + return (ctl & STOP_BIT) != 0L; + } + + /** + * Returns {@code true} if this pool has been shut down. * - * @return true if this pool has been shut down + * @return {@code true} if this pool has been shut down */ public boolean isShutdown() { - return runStateOf(runControl) >= SHUTDOWN; + return shutdown; } /** @@ -1120,14 +1968,14 @@ public class ForkJoinPool /*extends AbstractExecutorService*/ { * * @param timeout the maximum time to wait * @param unit the time unit of the timeout argument - * @return true if this executor terminated and - * false if the timeout elapsed before termination + * @return {@code true} if this executor terminated and + * {@code false} if the timeout elapsed before termination * @throws InterruptedException if interrupted while waiting */ public boolean awaitTermination(long timeout, TimeUnit unit) throws InterruptedException { long nanos = unit.toNanos(timeout); - final ReentrantLock lock = this.workerLock; + final ReentrantLock lock = this.submissionLock; lock.lock(); try { for (;;) { @@ -1142,729 +1990,165 @@ public class ForkJoinPool /*extends AbstractExecutorService*/ { } } - // Shutdown and termination support - - /** - * Callback from terminating worker. Null out the corresponding - * workers slot, and if terminating, try to terminate, else try 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(); - } - - /** - * Initiate termination. - */ - private void terminate() { - if (transitionRunStateTo(TERMINATING)) { - stopAllWorkers(); - resumeAllSpares(); - signalIdleWorkers(); - cancelQueuedSubmissions(); - cancelQueuedWorkerTasks(); - interruptUnterminatedWorkers(); - signalIdleWorkers(); // resignal after interrupt - } - } - - /** - * Possibly terminate when on shutdown state - */ - private void terminateOnShutdown() { - if (!hasQueuedSubmissions() && canTerminateOnShutdown(runControl)) - terminate(); - } - - /** - * Clear out and cancel submissions - */ - private void cancelQueuedSubmissions() { - ForkJoinTask task; - while ((task = pollSubmission()) != null) - task.cancel(false); - } - - /** - * Clean 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(); - } - } - - /** - * Set 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(); - } - } - - /** - * Interrupt 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; - } - - /** - * Wake 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; - } - - /** - * Await release on sync - */ - void awaitSyncRelease(ForkJoinPool p) { - while (thread != null && !p.syncIsReleasable(this)) - LockSupport.park(this); - } - - /** - * Await 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(); - } - - /** - * Signal 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 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 - - /** - * Decrement running count; if too low, add 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 enounter 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)))); - } - - /** - * Add 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; - } - - /** - * Add the kth spare worker. On entry, pool coounts 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(); - } - - /** - * Suspend 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; - } - - /** - * Try 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; - } - - /** - * Pop and resume 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; - } - - /** - * Pop and shutdown 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 isReleasable must return true if blocking is not - * necessary. Method block blocks the current thread - * if necessary (perhaps internally invoking 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). These actions are performed by any + * thread invoking {@link ForkJoinPool#managedBlock}. The + * unusual methods in this API accommodate synchronizers that may, + * but don't usually, block for long periods. Similarly, they + * allow more efficient internal handling of cases in which + * additional workers may be, but usually are not, needed to + * ensure sufficient parallelism. Toward this end, + * implementations of method {@code isReleasable} must be amenable + * to repeated invocation. + * *

For example, here is a ManagedBlocker based on a * ReentrantLock: - * 

-     *   class ManagedLocker implements ManagedBlocker {
-     *     final ReentrantLock lock;
-     *     boolean hasLock = false;
-     *     ManagedLocker(ReentrantLock lock) { this.lock = lock; }
-     *     public boolean block() {
-     *        if (!hasLock)
-     *           lock.lock();
-     *        return true;
-     *     }
-     *     public boolean isReleasable() {
-     *        return hasLock || (hasLock = lock.tryLock());
-     *     }
+     *  
 {@code
+     * class ManagedLocker implements ManagedBlocker {
+     *   final ReentrantLock lock;
+     *   boolean hasLock = false;
+     *   ManagedLocker(ReentrantLock lock) { this.lock = lock; }
+     *   public boolean block() {
+     *     if (!hasLock)
+     *       lock.lock();
+     *     return true;
      *   }
-     * 

+     *   public boolean isReleasable() {
+     *     return hasLock || (hasLock = lock.tryLock());
+     *   }
+     * }}
+ * + *

Here is a class that possibly blocks waiting for an + * item on a given queue: + * 

 {@code
+     * class QueueTaker implements ManagedBlocker {
+     *   final BlockingQueue queue;
+     *   volatile E item = null;
+     *   QueueTaker(BlockingQueue q) { this.queue = q; }
+     *   public boolean block() throws InterruptedException {
+     *     if (item == null)
+     *       item = queue.take();
+     *     return true;
+     *   }
+     *   public boolean isReleasable() {
+     *     return item != null || (item = queue.poll()) != null;
+     *   }
+     *   public E getItem() { // call after pool.managedBlock completes
+     *     return item;
+     *   }
+     * }}
*/ public static interface ManagedBlocker { /** * 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 allowe to). + * (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 - * maintainParallelism is true and the pool supports - * it ({@link #getMaintainsParallelism}), this method attempts to - * maintain the pool's nominal parallelism. Otherwise if activates - * a thread only if necessary to avoid complete starvation. This - * option may be preferable when blockages use timeouts, or are - * almost always brief. - * - *

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

-     *   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. + * 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 {@link ForkJoinTask}, this method is + * behaviorally equivalent to + * 

 {@code
+     * while (!blocker.isReleasable())
+     *   if (blocker.block())
+     *     return;
+     * }
+ * + * 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) + * @throws InterruptedException if blocker.block did so + */ + 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 w = (ForkJoinWorkerThread) t; + w.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); - } - - - // Temporary Unsafe mechanics for preliminary release - private static Unsafe getUnsafe() throws Throwable { - try { - return Unsafe.getUnsafe(); - } catch (SecurityException se) { - try { - return java.security.AccessController.doPrivileged - (new java.security.PrivilegedExceptionAction() { - public Unsafe run() throws Exception { - return getUnsafePrivileged(); - }}); - } catch (java.security.PrivilegedActionException e) { - throw e.getCause(); - } - } + return (RunnableFuture) ForkJoinTask.adapt(callable); } - 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; + // Unsafe mechanics + private static final sun.misc.Unsafe UNSAFE; + private static final long ctlOffset; + private static final long stealCountOffset; + private static final long blockedCountOffset; + private static final long quiescerCountOffset; + private static final long scanGuardOffset; + private static final long nextWorkerNumberOffset; + private static final long ABASE; + private static final int ASHIFT; static { + poolNumberGenerator = new AtomicInteger(); + workerSeedGenerator = new Random(); + modifyThreadPermission = new RuntimePermission("modifyThread"); + defaultForkJoinWorkerThreadFactory = + new DefaultForkJoinWorkerThreadFactory(); + int s; 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); + UNSAFE = sun.misc.Unsafe.getUnsafe(); + Class k = ForkJoinPool.class; + ctlOffset = UNSAFE.objectFieldOffset + (k.getDeclaredField("ctl")); + stealCountOffset = UNSAFE.objectFieldOffset + (k.getDeclaredField("stealCount")); + blockedCountOffset = UNSAFE.objectFieldOffset + (k.getDeclaredField("blockedCount")); + quiescerCountOffset = UNSAFE.objectFieldOffset + (k.getDeclaredField("quiescerCount")); + scanGuardOffset = UNSAFE.objectFieldOffset + (k.getDeclaredField("scanGuard")); + nextWorkerNumberOffset = UNSAFE.objectFieldOffset + (k.getDeclaredField("nextWorkerNumber")); + Class a = ForkJoinTask[].class; + ABASE = UNSAFE.arrayBaseOffset(a); + s = UNSAFE.arrayIndexScale(a); + } catch (Exception e) { + throw new Error(e); } + if ((s & (s-1)) != 0) + throw new Error("data type scale not a power of two"); + ASHIFT = 31 - Integer.numberOfLeadingZeros(s); } - 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); - } } diff --git a/src/forkjoin/scala/concurrent/forkjoin/ForkJoinTask.java b/src/forkjoin/scala/concurrent/forkjoin/ForkJoinTask.java index dc1a6bcccc..1233195b71 100644 --- a/src/forkjoin/scala/concurrent/forkjoin/ForkJoinTask.java +++ b/src/forkjoin/scala/concurrent/forkjoin/ForkJoinTask.java @@ -1,470 +1,572 @@ /* * Written by Doug Lea with assistance from members of JCP JSR-166 * Expert Group and released to the public domain, as explained at - * http://creativecommons.org/licenses/publicdomain + * http://creativecommons.org/publicdomain/zero/1.0/ */ package scala.concurrent.forkjoin; + import java.io.Serializable; -import java.util.*; -import java.util.concurrent.*; -import java.util.concurrent.atomic.*; -import sun.misc.Unsafe; -import java.lang.reflect.*; +import java.util.Collection; +import java.util.Collections; +import java.util.List; +import java.util.RandomAccess; +import java.util.Map; +import java.lang.ref.WeakReference; +import java.lang.ref.ReferenceQueue; +import java.util.concurrent.Callable; +import java.util.concurrent.CancellationException; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.Executor; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Future; +import java.util.concurrent.RejectedExecutionException; +import java.util.concurrent.RunnableFuture; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; +import java.util.concurrent.locks.ReentrantLock; +import java.lang.reflect.Constructor; /** - * Abstract base class for tasks that run within a {@link - * ForkJoinPool}. A ForkJoinTask is a thread-like entity that is much + * Abstract base class for tasks that run within a {@link ForkJoinPool}. + * A {@code ForkJoinTask} is a thread-like entity that is much * lighter weight than a normal thread. Huge numbers of tasks and * subtasks may be hosted by a small number of actual threads in a * ForkJoinPool, at the price of some usage limitations. * - *

A "main" ForkJoinTask begins execution when submitted to a - * {@link ForkJoinPool}. Once started, it will usually in turn start - * other subtasks. As indicated by the name of this class, many - * programs using ForkJoinTasks employ only methods fork - * and join, or derivatives such as - * invokeAll. However, this class also provides a number - * of other methods that can come into play in advanced usages, as - * well as extension mechanics that allow support of new forms of - * fork/join processing. + *

A "main" {@code ForkJoinTask} begins execution when submitted + * to a {@link ForkJoinPool}. Once started, it will usually in turn + * start other subtasks. As indicated by the name of this class, + * many programs using {@code ForkJoinTask} employ only methods + * {@link #fork} and {@link #join}, or derivatives such as {@link + * #invokeAll(ForkJoinTask...) invokeAll}. However, this class also + * provides a number of other methods that can come into play in + * advanced usages, as well as extension mechanics that allow + * support of new forms of fork/join processing. * - *

A ForkJoinTask is a lightweight form of {@link Future}. The - * efficiency of ForkJoinTasks stems from a set of restrictions (that - * are only partially statically enforceable) reflecting their - * intended use as computational tasks calculating pure functions or - * operating on purely isolated objects. The primary coordination - * mechanisms are {@link #fork}, that arranges asynchronous execution, - * and {@link #join}, that doesn't proceed until the task's result has - * been computed. Computations should avoid synchronized - * methods or blocks, and should minimize other blocking - * synchronization apart from joining other tasks or using - * synchronizers such as Phasers that are advertised to cooperate with - * fork/join scheduling. Tasks should also not perform blocking IO, - * and should ideally access variables that are completely independent - * of those accessed by other running tasks. Minor breaches of these - * restrictions, for example using shared output streams, may be - * tolerable in practice, but frequent use may result in poor - * performance, and the potential to indefinitely stall if the number - * of threads not waiting for IO or other external synchronization - * becomes exhausted. This usage restriction is in part enforced by - * not permitting checked exceptions such as IOExceptions - * to be thrown. However, computations may still encounter unchecked - * exceptions, that are rethrown to callers attempting join - * them. These exceptions may additionally include - * RejectedExecutionExceptions stemming from internal resource - * exhaustion such as failure to allocate internal task queues. + *

A {@code ForkJoinTask} is a lightweight form of {@link Future}. + * The efficiency of {@code ForkJoinTask}s stems from a set of + * restrictions (that are only partially statically enforceable) + * reflecting their intended use as computational tasks calculating + * pure functions or operating on purely isolated objects. The + * primary coordination mechanisms are {@link #fork}, that arranges + * asynchronous execution, and {@link #join}, that doesn't proceed + * until the task's result has been computed. Computations should + * avoid {@code synchronized} methods or blocks, and should minimize + * other blocking synchronization apart from joining other tasks or + * using synchronizers such as Phasers that are advertised to + * cooperate with fork/join scheduling. Tasks should also not perform + * blocking IO, and should ideally access variables that are + * completely independent of those accessed by other running + * tasks. Minor breaches of these restrictions, for example using + * shared output streams, may be tolerable in practice, but frequent + * use may result in poor performance, and the potential to + * indefinitely stall if the number of threads not waiting for IO or + * other external synchronization becomes exhausted. This usage + * restriction is in part enforced by not permitting checked + * exceptions such as {@code IOExceptions} to be thrown. However, + * computations may still encounter unchecked exceptions, that are + * rethrown to callers attempting to join them. These exceptions may + * additionally include {@link RejectedExecutionException} stemming + * from internal resource exhaustion, such as failure to allocate + * internal task queues. Rethrown exceptions behave in the same way as + * regular exceptions, but, when possible, contain stack traces (as + * displayed for example using {@code ex.printStackTrace()}) of both + * the thread that initiated the computation as well as the thread + * actually encountering the exception; minimally only the latter. * *

The primary method for awaiting completion and extracting * results of a task is {@link #join}, but there are several variants: * The {@link Future#get} methods support interruptible and/or timed - * waits for completion and report results using Future - * conventions. Method {@link #helpJoin} enables callers to actively - * execute other tasks while awaiting joins, which is sometimes more - * efficient but only applies when all subtasks are known to be - * strictly tree-structured. Method {@link #invoke} is semantically - * equivalent to fork(); join() but always attempts to - * begin execution in the current thread. The "quiet" forms - * of these methods do not extract results or report exceptions. These + * waits for completion and report results using {@code Future} + * conventions. Method {@link #invoke} is semantically + * equivalent to {@code fork(); join()} but always attempts to begin + * execution in the current thread. The "quiet" forms of + * these methods do not extract results or report exceptions. These * may be useful when a set of tasks are being executed, and you need * to delay processing of results or exceptions until all complete. - * Method invokeAll (available in multiple versions) + * Method {@code invokeAll} (available in multiple versions) * performs the most common form of parallel invocation: forking a set * of tasks and joining them all. * - *

The ForkJoinTask class is not usually directly subclassed. + *

The execution status of tasks may be queried at several levels + * of detail: {@link #isDone} is true if a task completed in any way + * (including the case where a task was cancelled without executing); + * {@link #isCompletedNormally} is true if a task completed without + * cancellation or encountering an exception; {@link #isCancelled} is + * true if the task was cancelled (in which case {@link #getException} + * returns a {@link java.util.concurrent.CancellationException}); and + * {@link #isCompletedAbnormally} is true if a task was either + * cancelled or encountered an exception, in which case {@link + * #getException} will return either the encountered exception or + * {@link java.util.concurrent.CancellationException}. + * + *

The ForkJoinTask class is not usually directly subclassed. * Instead, you subclass one of the abstract classes that support a - * particular style of fork/join processing. Normally, a concrete + * particular style of fork/join processing, typically {@link + * RecursiveAction} for computations that do not return results, or + * {@link RecursiveTask} for those that do. Normally, a concrete * ForkJoinTask subclass declares fields comprising its parameters, - * established in a constructor, and then defines a compute + * established in a constructor, and then defines a {@code compute} * method that somehow uses the control methods supplied by this base - * class. While these methods have public access (to allow - * instances of different task subclasses to call each others + * class. While these methods have {@code public} access (to allow + * instances of different task subclasses to call each other's * methods), some of them may only be called from within other - * ForkJoinTasks. Attempts to invoke them in other contexts result in - * exceptions or errors possibly including ClassCastException. + * ForkJoinTasks (as may be determined using method {@link + * #inForkJoinPool}). Attempts to invoke them in other contexts + * result in exceptions or errors, possibly including + * {@code ClassCastException}. + * + *

Method {@link #join} and its variants are appropriate for use + * only when completion dependencies are acyclic; that is, the + * parallel computation can be described as a directed acyclic graph + * (DAG). Otherwise, executions may encounter a form of deadlock as + * tasks cyclically wait for each other. However, this framework + * supports other methods and techniques (for example the use of + * {@link Phaser}, {@link #helpQuiesce}, and {@link #complete}) that + * may be of use in constructing custom subclasses for problems that + * are not statically structured as DAGs. * - *

Most base support methods are final because their - * implementations are intrinsically tied to the underlying - * lightweight task scheduling framework, and so cannot be overridden. - * Developers creating new basic styles of fork/join processing should - * minimally implement protected methods - * exec, setRawResult, and - * getRawResult, while also introducing an abstract - * computational method that can be implemented in its subclasses, - * possibly relying on other protected methods provided - * by this class. + *

Most base support methods are {@code final}, to prevent + * overriding of implementations that are intrinsically tied to the + * underlying lightweight task scheduling framework. Developers + * creating new basic styles of fork/join processing should minimally + * implement {@code protected} methods {@link #exec}, {@link + * #setRawResult}, and {@link #getRawResult}, while also introducing + * an abstract computational method that can be implemented in its + * subclasses, possibly relying on other {@code protected} methods + * provided by this class. * *

ForkJoinTasks should perform relatively small amounts of - * computations, othewise splitting into smaller tasks. As a very - * rough rule of thumb, a task should perform more than 100 and less - * than 10000 basic computational steps. If tasks are too big, then - * parellelism cannot improve throughput. If too small, then memory - * and internal task maintenance overhead may overwhelm processing. + * computation. Large tasks should be split into smaller subtasks, + * usually via recursive decomposition. As a very rough rule of thumb, + * a task should perform more than 100 and less than 10000 basic + * computational steps, and should avoid indefinite looping. If tasks + * are too big, then parallelism cannot improve throughput. If too + * small, then memory and internal task maintenance overhead may + * overwhelm processing. + * + *

This class provides {@code adapt} methods for {@link Runnable} + * and {@link Callable}, that may be of use when mixing execution of + * {@code ForkJoinTasks} with other kinds of tasks. When all tasks are + * of this form, consider using a pool constructed in asyncMode. + * + *

ForkJoinTasks are {@code Serializable}, which enables them to be + * used in extensions such as remote execution frameworks. It is + * sensible to serialize tasks only before or after, but not during, + * execution. Serialization is not relied on during execution itself. * - *

ForkJoinTasks are Serializable, which enables them - * to be used in extensions such as remote execution frameworks. It is - * in general sensible to serialize tasks only before or after, but - * not during execution. Serialization is not relied on during - * execution itself. + * @since 1.7 + * @author Doug Lea */ public abstract class ForkJoinTask implements Future, Serializable { - /** - * Run control status bits packed into a single int to minimize - * footprint and to ensure atomicity (via CAS). Status is - * initially zero, and takes on nonnegative values until - * completed, upon which status holds COMPLETED. CANCELLED, or - * EXCEPTIONAL, which use the top 3 bits. Tasks undergoing - * blocking waits by other threads have SIGNAL_MASK bits set -- - * bit 15 for external (nonFJ) waits, and the rest a count of - * waiting FJ threads. (This representation relies on - * ForkJoinPool max thread limits). Completion of a stolen task - * with SIGNAL_MASK bits set awakens waiter via notifyAll. Even - * though suboptimal for some purposes, we use basic builtin - * wait/notify to take advantage of "monitor inflation" in JVMs - * that we would otherwise need to emulate to avoid adding further - * per-task bookkeeping overhead. Note that bits 16-28 are - * currently unused. Also value 0x80000000 is available as spare - * completion value. + /* + * See the internal documentation of class ForkJoinPool for a + * general implementation overview. ForkJoinTasks are mainly + * responsible for maintaining their "status" field amidst relays + * to methods in ForkJoinWorkerThread and ForkJoinPool. The + * methods of this class are more-or-less layered into (1) basic + * status maintenance (2) execution and awaiting completion (3) + * user-level methods that additionally report results. This is + * sometimes hard to see because this file orders exported methods + * in a way that flows well in javadocs. */ - volatile int status; // accessed directy by pool and workers - - static final int COMPLETION_MASK = 0xe0000000; - static final int NORMAL = 0xe0000000; // == mask - static final int CANCELLED = 0xc0000000; - static final int EXCEPTIONAL = 0xa0000000; - static final int SIGNAL_MASK = 0x0000ffff; - static final int INTERNAL_SIGNAL_MASK = 0x00007fff; - static final int EXTERNAL_SIGNAL = 0x00008000; // top bit of low word - /** - * Table of exceptions thrown by tasks, to enable reporting by - * callers. Because exceptions are rare, we don't directly keep - * them with task objects, but instead us a weak ref table. Note - * that cancellation exceptions don't appear in the table, but are - * instead recorded as status values. - * Todo: Use ConcurrentReferenceHashMap + /* + * The status field holds run control status bits packed into a + * single int to minimize footprint and to ensure atomicity (via + * CAS). Status is initially zero, and takes on nonnegative + * values until completed, upon which status holds value + * NORMAL, CANCELLED, or EXCEPTIONAL. Tasks undergoing blocking + * waits by other threads have the SIGNAL bit set. Completion of + * a stolen task with SIGNAL set awakens any waiters via + * notifyAll. Even though suboptimal for some purposes, we use + * basic builtin wait/notify to take advantage of "monitor + * inflation" in JVMs that we would otherwise need to emulate to + * avoid adding further per-task bookkeeping overhead. We want + * these monitors to be "fat", i.e., not use biasing or thin-lock + * techniques, so use some odd coding idioms that tend to avoid + * them. */ - static final Map, Throwable> exceptionMap = - Collections.synchronizedMap - (new WeakHashMap, Throwable>()); - // within-package utilities + /** The run status of this task */ + volatile int status; // accessed directly by pool and workers + private static final int NORMAL = -1; + private static final int CANCELLED = -2; + private static final int EXCEPTIONAL = -3; + private static final int SIGNAL = 1; /** - * Get current worker thread, or null if not a worker thread - */ - static ForkJoinWorkerThread getWorker() { - Thread t = Thread.currentThread(); - return ((t instanceof ForkJoinWorkerThread)? - (ForkJoinWorkerThread)t : null); - } - - final boolean casStatus(int cmp, int val) { - return _unsafe.compareAndSwapInt(this, statusOffset, cmp, val); - } - - /** - * Workaround for not being able to rethrow unchecked exceptions. - */ - static void rethrowException(Throwable ex) { - if (ex != null) - _unsafe.throwException(ex); - } - - // Setting completion status - - /** - * Mark completion and wake up threads waiting to join this task. + * Marks completion and wakes up threads waiting to join this task, + * also clearing signal request bits. + * * @param completion one of NORMAL, CANCELLED, EXCEPTIONAL + * @return completion status on exit */ - final void setCompletion(int completion) { - ForkJoinPool pool = getPool(); - if (pool != null) { - int s; // Clear signal bits while setting completion status - do;while ((s = status) >= 0 && !casStatus(s, completion)); - - if ((s & SIGNAL_MASK) != 0) { - if ((s &= INTERNAL_SIGNAL_MASK) != 0) - pool.updateRunningCount(s); - synchronized(this) { notifyAll(); } + private int setCompletion(int completion) { + for (int s;;) { + if ((s = status) < 0) + return s; + if (UNSAFE.compareAndSwapInt(this, statusOffset, s, completion)) { + if (s != 0) + synchronized (this) { notifyAll(); } + return completion; } } - else - externallySetCompletion(completion); } /** - * Version of setCompletion for non-FJ threads. Leaves signal - * bits for unblocked threads to adjust, and always notifies. + * Tries to block a worker thread until completed or timed out. + * Uses Object.wait time argument conventions. + * May fail on contention or interrupt. + * + * @param millis if > 0, wait time. */ - private void externallySetCompletion(int completion) { + final void tryAwaitDone(long millis) { int s; - do;while ((s = status) >= 0 && - !casStatus(s, (s & SIGNAL_MASK) | completion)); - synchronized(this) { notifyAll(); } - } - - /** - * Sets status to indicate normal completion - */ - final void setNormalCompletion() { - // Try typical fast case -- single CAS, no signal, not already done. - // Manually expand casStatus to improve chances of inlining it - if (!_unsafe.compareAndSwapInt(this, statusOffset, 0, NORMAL)) - setCompletion(NORMAL); - } - - // internal waiting and notification - - /** - * Performs the actual monitor wait for awaitDone - */ - private void doAwaitDone() { - // Minimize lock bias and in/de-flation effects by maximizing - // chances of waiting inside sync try { - while (status >= 0) - synchronized(this) { if (status >= 0) wait(); } - } catch (InterruptedException ie) { - onInterruptedWait(); - } - } - - /** - * Performs the actual monitor wait for awaitDone - */ - private void doAwaitDone(long startTime, long nanos) { - synchronized(this) { - try { - while (status >= 0) { - long nt = nanos - System.nanoTime() - startTime; - if (nt <= 0) - break; - wait(nt / 1000000, (int)(nt % 1000000)); + if (((s = status) > 0 || + (s == 0 && + UNSAFE.compareAndSwapInt(this, statusOffset, 0, SIGNAL))) && + status > 0) { + synchronized (this) { + if (status > 0) + wait(millis); } - } catch (InterruptedException ie) { - onInterruptedWait(); } + } catch (InterruptedException ie) { + // caller must check termination } } - // Awaiting completion - /** - * Sets status to indicate there is joiner, then waits for join, - * surrounded with pool notifications. - * @return status upon exit + * Blocks a non-worker-thread until completion. + * @return status upon completion */ - private int awaitDone(ForkJoinWorkerThread w, boolean maintainParallelism) { - ForkJoinPool pool = w == null? null : w.pool; + private int externalAwaitDone() { int s; - while ((s = status) >= 0) { - if (casStatus(s, pool == null? s|EXTERNAL_SIGNAL : s+1)) { - if (pool == null || !pool.preJoin(this, maintainParallelism)) - doAwaitDone(); - if (((s = status) & INTERNAL_SIGNAL_MASK) != 0) - adjustPoolCountsOnUnblock(pool); - break; + if ((s = status) >= 0) { + boolean interrupted = false; + synchronized (this) { + while ((s = status) >= 0) { + if (s == 0) + UNSAFE.compareAndSwapInt(this, statusOffset, + 0, SIGNAL); + else { + try { + wait(); + } catch (InterruptedException ie) { + interrupted = true; + } + } + } } + if (interrupted) + Thread.currentThread().interrupt(); } return s; } /** - * Timed version of awaitDone - * @return status upon exit + * Blocks a non-worker-thread until completion or interruption or timeout. */ - private int awaitDone(ForkJoinWorkerThread w, long nanos) { - ForkJoinPool pool = w == null? null : w.pool; + private int externalInterruptibleAwaitDone(long millis) + throws InterruptedException { int s; - while ((s = status) >= 0) { - if (casStatus(s, pool == null? s|EXTERNAL_SIGNAL : s+1)) { - long startTime = System.nanoTime(); - if (pool == null || !pool.preJoin(this, false)) - doAwaitDone(startTime, nanos); - if ((s = status) >= 0) { - adjustPoolCountsOnCancelledWait(pool); - s = status; + if (Thread.interrupted()) + throw new InterruptedException(); + if ((s = status) >= 0) { + synchronized (this) { + while ((s = status) >= 0) { + if (s == 0) + UNSAFE.compareAndSwapInt(this, statusOffset, + 0, SIGNAL); + else { + wait(millis); + if (millis > 0L) + break; + } } - if (s < 0 && (s & INTERNAL_SIGNAL_MASK) != 0) - adjustPoolCountsOnUnblock(pool); - break; } } return s; } /** - * Notify pool that thread is unblocked. Called by signalled - * threads when woken by non-FJ threads (which is atypical). + * Primary execution method for stolen tasks. Unless done, calls + * exec and records status if completed, but doesn't wait for + * completion otherwise. */ - private void adjustPoolCountsOnUnblock(ForkJoinPool pool) { - int s; - do;while ((s = status) < 0 && !casStatus(s, s & COMPLETION_MASK)); - if (pool != null && (s &= INTERNAL_SIGNAL_MASK) != 0) - pool.updateRunningCount(s); + final void doExec() { + if (status >= 0) { + boolean completed; + try { + completed = exec(); + } catch (Throwable rex) { + setExceptionalCompletion(rex); + return; + } + if (completed) + setCompletion(NORMAL); // must be outside try block + } } /** - * Notify pool to adjust counts on cancelled or timed out wait + * Primary mechanics for join, get, quietlyJoin. + * @return status upon completion */ - private void adjustPoolCountsOnCancelledWait(ForkJoinPool pool) { - if (pool != null) { - int s; - while ((s = status) >= 0 && (s & INTERNAL_SIGNAL_MASK) != 0) { - if (casStatus(s, s - 1)) { - pool.updateRunningCount(1); - break; + private int doJoin() { + Thread t; ForkJoinWorkerThread w; int s; boolean completed; + if ((t = Thread.currentThread()) instanceof ForkJoinWorkerThread) { + if ((s = status) < 0) + return s; + if ((w = (ForkJoinWorkerThread)t).unpushTask(this)) { + try { + completed = exec(); + } catch (Throwable rex) { + return setExceptionalCompletion(rex); } + if (completed) + return setCompletion(NORMAL); } + return w.joinTask(this); } + else + return externalAwaitDone(); } /** - * Handle interruptions during waits. + * Primary mechanics for invoke, quietlyInvoke. + * @return status upon completion */ - private void onInterruptedWait() { - ForkJoinWorkerThread w = getWorker(); - if (w == null) - Thread.currentThread().interrupt(); // re-interrupt - else if (w.isTerminating()) - cancelIgnoringExceptions(); - // else if FJworker, ignore interrupt + private int doInvoke() { + int s; boolean completed; + if ((s = status) < 0) + return s; + try { + completed = exec(); + } catch (Throwable rex) { + return setExceptionalCompletion(rex); + } + if (completed) + return setCompletion(NORMAL); + else + return doJoin(); } - // Recording and reporting exceptions + // Exception table support - private void setDoneExceptionally(Throwable rex) { - exceptionMap.put(this, rex); - setCompletion(EXCEPTIONAL); - } + /** + * Table of exceptions thrown by tasks, to enable reporting by + * callers. Because exceptions are rare, we don't directly keep + * them with task objects, but instead use a weak ref table. Note + * that cancellation exceptions don't appear in the table, but are + * instead recorded as status values. + * + * Note: These statics are initialized below in static block. + */ + private static final ExceptionNode[] exceptionTable; + private static final ReentrantLock exceptionTableLock; + private static final ReferenceQueue exceptionTableRefQueue; /** - * Throws the exception associated with status s; - * @throws the exception + * Fixed capacity for exceptionTable. */ - private void reportException(int s) { - if ((s &= COMPLETION_MASK) < NORMAL) { - if (s == CANCELLED) - throw new CancellationException(); - else - rethrowException(exceptionMap.get(this)); - } - } + private static final int EXCEPTION_MAP_CAPACITY = 32; /** - * Returns result or throws exception using j.u.c.Future conventions - * Only call when isDone known to be true. + * Key-value nodes for exception table. The chained hash table + * uses identity comparisons, full locking, and weak references + * for keys. The table has a fixed capacity because it only + * maintains task exceptions long enough for joiners to access + * them, so should never become very large for sustained + * periods. However, since we do not know when the last joiner + * completes, we must use weak references and expunge them. We do + * so on each operation (hence full locking). Also, some thread in + * any ForkJoinPool will call helpExpungeStaleExceptions when its + * pool becomes isQuiescent. */ - private V reportFutureResult() - throws ExecutionException, InterruptedException { - int s = status & COMPLETION_MASK; - if (s < NORMAL) { - Throwable ex; - if (s == CANCELLED) - throw new CancellationException(); - if (s == EXCEPTIONAL && (ex = exceptionMap.get(this)) != null) - throw new ExecutionException(ex); - if (Thread.interrupted()) - throw new InterruptedException(); + static final class ExceptionNode extends WeakReference>{ + final Throwable ex; + ExceptionNode next; + final long thrower; // use id not ref to avoid weak cycles + ExceptionNode(ForkJoinTask task, Throwable ex, ExceptionNode next) { + super(task, exceptionTableRefQueue); + this.ex = ex; + this.next = next; + this.thrower = Thread.currentThread().getId(); } - return getRawResult(); } /** - * Returns result or throws exception using j.u.c.Future conventions - * with timeouts + * Records exception and sets exceptional completion. + * + * @return status on exit */ - private V reportTimedFutureResult() - throws InterruptedException, ExecutionException, TimeoutException { - Throwable ex; - int s = status & COMPLETION_MASK; - if (s == NORMAL) - return getRawResult(); - if (s == CANCELLED) - throw new CancellationException(); - if (s == EXCEPTIONAL && (ex = exceptionMap.get(this)) != null) - throw new ExecutionException(ex); - if (Thread.interrupted()) - throw new InterruptedException(); - throw new TimeoutException(); + private int setExceptionalCompletion(Throwable ex) { + int h = System.identityHashCode(this); + final ReentrantLock lock = exceptionTableLock; + lock.lock(); + try { + expungeStaleExceptions(); + ExceptionNode[] t = exceptionTable; + int i = h & (t.length - 1); + for (ExceptionNode e = t[i]; ; e = e.next) { + if (e == null) { + t[i] = new ExceptionNode(this, ex, t[i]); + break; + } + if (e.get() == this) // already present + break; + } + } finally { + lock.unlock(); + } + return setCompletion(EXCEPTIONAL); } - // internal execution methods - /** - * Calls exec, recording completion, and rethrowing exception if - * encountered. Caller should normally check status before calling - * @return true if completed normally + * Removes exception node and clears status */ - private boolean tryExec() { - try { // try block must contain only call to exec - if (!exec()) - return false; - } catch (Throwable rex) { - setDoneExceptionally(rex); - rethrowException(rex); - return false; // not reached + private void clearExceptionalCompletion() { + int h = System.identityHashCode(this); + final ReentrantLock lock = exceptionTableLock; + lock.lock(); + try { + ExceptionNode[] t = exceptionTable; + int i = h & (t.length - 1); + ExceptionNode e = t[i]; + ExceptionNode pred = null; + while (e != null) { + ExceptionNode next = e.next; + if (e.get() == this) { + if (pred == null) + t[i] = next; + else + pred.next = next; + break; + } + pred = e; + e = next; + } + expungeStaleExceptions(); + status = 0; + } finally { + lock.unlock(); } - setNormalCompletion(); - return true; } /** - * Main execution method used by worker threads. Invokes - * base computation unless already complete + * Returns a rethrowable exception for the given task, if + * available. To provide accurate stack traces, if the exception + * was not thrown by the current thread, we try to create a new + * exception of the same type as the one thrown, but with the + * recorded exception as its cause. If there is no such + * constructor, we instead try to use a no-arg constructor, + * followed by initCause, to the same effect. If none of these + * apply, or any fail due to other exceptions, we return the + * recorded exception, which is still correct, although it may + * contain a misleading stack trace. + * + * @return the exception, or null if none */ - final void quietlyExec() { - if (status >= 0) { + private Throwable getThrowableException() { + if (status != EXCEPTIONAL) + return null; + int h = System.identityHashCode(this); + ExceptionNode e; + final ReentrantLock lock = exceptionTableLock; + lock.lock(); + try { + expungeStaleExceptions(); + ExceptionNode[] t = exceptionTable; + e = t[h & (t.length - 1)]; + while (e != null && e.get() != this) + e = e.next; + } finally { + lock.unlock(); + } + Throwable ex; + if (e == null || (ex = e.ex) == null) + return null; + if (e.thrower != Thread.currentThread().getId()) { + Class ec = ex.getClass(); try { - if (!exec()) - return; - } catch(Throwable rex) { - setDoneExceptionally(rex); - return; + Constructor noArgCtor = null; + Constructor[] cs = ec.getConstructors();// public ctors only + for (int i = 0; i < cs.length; ++i) { + Constructor c = cs[i]; + Class[] ps = c.getParameterTypes(); + if (ps.length == 0) + noArgCtor = c; + else if (ps.length == 1 && ps[0] == Throwable.class) + return (Throwable)(c.newInstance(ex)); + } + if (noArgCtor != null) { + Throwable wx = (Throwable)(noArgCtor.newInstance()); + wx.initCause(ex); + return wx; + } + } catch (Exception ignore) { } - setNormalCompletion(); } + return ex; } /** - * Calls exec, recording but not rethrowing exception - * Caller should normally check status before calling - * @return true if completed normally + * Poll stale refs and remove them. Call only while holding lock. */ - private boolean tryQuietlyInvoke() { - try { - if (!exec()) - return false; - } catch (Throwable rex) { - setDoneExceptionally(rex); - return false; + private static void expungeStaleExceptions() { + for (Object x; (x = exceptionTableRefQueue.poll()) != null;) { + if (x instanceof ExceptionNode) { + ForkJoinTask key = ((ExceptionNode)x).get(); + ExceptionNode[] t = exceptionTable; + int i = System.identityHashCode(key) & (t.length - 1); + ExceptionNode e = t[i]; + ExceptionNode pred = null; + while (e != null) { + ExceptionNode next = e.next; + if (e == x) { + if (pred == null) + t[i] = next; + else + pred.next = next; + break; + } + pred = e; + e = next; + } + } } - setNormalCompletion(); - return true; } /** - * Cancel, ignoring any exceptions it throws + * If lock is available, poll stale refs and remove them. + * Called from ForkJoinPool when pools become quiescent. */ - final void cancelIgnoringExceptions() { - try { - cancel(false); - } catch(Throwable ignore) { + static final void helpExpungeStaleExceptions() { + final ReentrantLock lock = exceptionTableLock; + if (lock.tryLock()) { + try { + expungeStaleExceptions(); + } finally { + lock.unlock(); + } } } /** - * Main implementation of helpJoin + * Report the result of invoke or join; called only upon + * non-normal return of internal versions. */ - private int busyJoin(ForkJoinWorkerThread w) { - int s; - ForkJoinTask t; - while ((s = status) >= 0 && (t = w.scanWhileJoining(this)) != null) - t.quietlyExec(); - return (s >= 0)? awaitDone(w, false) : s; // block if no work + private V reportResult() { + int s; Throwable ex; + if ((s = status) == CANCELLED) + throw new CancellationException(); + if (s == EXCEPTIONAL && (ex = getThrowableException()) != null) + UNSAFE.throwException(ex); + return getRawResult(); } // public methods @@ -472,70 +574,109 @@ public abstract class ForkJoinTask implements Future, Serializable { /** * Arranges to asynchronously execute this task. While it is not * necessarily enforced, it is a usage error to fork a task more - * than once unless it has completed and been reinitialized. This - * method may be invoked only from within ForkJoinTask - * computations. Attempts to invoke in other contexts result in - * exceptions or errors possibly including ClassCastException. + * than once unless it has completed and been reinitialized. + * Subsequent modifications to the state of this task or any data + * it operates on are not necessarily consistently observable by + * any thread other than the one executing it unless preceded by a + * call to {@link #join} or related methods, or a call to {@link + * #isDone} returning {@code true}. + * + *

This method may be invoked only from within {@code + * ForkJoinPool} computations (as may be determined using method + * {@link #inForkJoinPool}). Attempts to invoke in other contexts + * result in exceptions or errors, possibly including {@code + * ClassCastException}. + * + * @return {@code this}, to simplify usage */ - public final void fork() { - ((ForkJoinWorkerThread)(Thread.currentThread())).pushTask(this); + public final ForkJoinTask fork() { + ((ForkJoinWorkerThread) Thread.currentThread()) + .pushTask(this); + return this; } /** - * Returns the result of the computation when it is ready. - * This method differs from get in that abnormal - * completion results in RuntimeExceptions or Errors, not - * ExecutionExceptions. + * Returns the result of the computation when it {@link #isDone is + * done}. This method differs from {@link #get()} in that + * abnormal completion results in {@code RuntimeException} or + * {@code Error}, not {@code ExecutionException}, and that + * interrupts of the calling thread do not cause the + * method to abruptly return by throwing {@code + * InterruptedException}. * * @return the computed result */ public final V join() { - ForkJoinWorkerThread w = getWorker(); - if (w == null || status < 0 || !w.unpushTask(this) || !tryExec()) - reportException(awaitDone(w, true)); - return getRawResult(); + if (doJoin() != NORMAL) + return reportResult(); + else + return getRawResult(); } /** * Commences performing this task, awaits its completion if - * necessary, and return its result. - * @throws Throwable (a RuntimeException, Error, or unchecked - * exception) if the underlying computation did so. + * necessary, and returns its result, or throws an (unchecked) + * {@code RuntimeException} or {@code Error} if the underlying + * computation did so. + * * @return the computed result */ public final V invoke() { - if (status >= 0 && tryExec()) - return getRawResult(); + if (doInvoke() != NORMAL) + return reportResult(); else - return join(); + return getRawResult(); } /** - * Forks both tasks, returning when isDone holds for - * both of them or an exception is encountered. This method may be - * invoked only from within ForkJoinTask computations. Attempts to - * invoke in other contexts result in exceptions or errors - * possibly including ClassCastException. - * @param t1 one task - * @param t2 the other task - * @throws NullPointerException if t1 or t2 are null - * @throws RuntimeException or Error if either task did so. + * Forks the given tasks, returning when {@code isDone} holds for + * each task or an (unchecked) exception is encountered, in which + * case the exception is rethrown. If more than one task + * encounters an exception, then this method throws any one of + * these exceptions. If any task encounters an exception, the + * other may be cancelled. However, the execution status of + * individual tasks is not guaranteed upon exceptional return. The + * status of each task may be obtained using {@link + * #getException()} and related methods to check if they have been + * cancelled, completed normally or exceptionally, or left + * unprocessed. + * + *

This method may be invoked only from within {@code + * ForkJoinPool} computations (as may be determined using method + * {@link #inForkJoinPool}). Attempts to invoke in other contexts + * result in exceptions or errors, possibly including {@code + * ClassCastException}. + * + * @param t1 the first task + * @param t2 the second task + * @throws NullPointerException if any task is null */ - public static void invokeAll(ForkJoinTaskt1, ForkJoinTask t2) { + public static void invokeAll(ForkJoinTask t1, ForkJoinTask t2) { t2.fork(); t1.invoke(); t2.join(); } /** - * Forks the given tasks, returning when isDone holds - * for all of them. If any task encounters an exception, others - * may be cancelled. This method may be invoked only from within - * ForkJoinTask computations. Attempts to invoke in other contexts - * result in exceptions or errors possibly including ClassCastException. - * @param tasks the array of tasks - * @throws NullPointerException if tasks or any element are null. - * @throws RuntimeException or Error if any task did so. + * Forks the given tasks, returning when {@code isDone} holds for + * each task or an (unchecked) exception is encountered, in which + * case the exception is rethrown. If more than one task + * encounters an exception, then this method throws any one of + * these exceptions. If any task encounters an exception, others + * may be cancelled. However, the execution status of individual + * tasks is not guaranteed upon exceptional return. The status of + * each task may be obtained using {@link #getException()} and + * related methods to check if they have been cancelled, completed + * normally or exceptionally, or left unprocessed. + * + *

This method may be invoked only from within {@code + * ForkJoinPool} computations (as may be determined using method + * {@link #inForkJoinPool}). Attempts to invoke in other contexts + * result in exceptions or errors, possibly including {@code + * ClassCastException}. + * + * @param tasks the tasks + * @throws NullPointerException if any task is null */ public static void invokeAll(ForkJoinTask... tasks) { Throwable ex = null; @@ -548,46 +689,53 @@ public abstract class ForkJoinTask implements Future, Serializable { } else if (i != 0) t.fork(); - else { - t.quietlyInvoke(); - if (ex == null) - ex = t.getException(); - } + else if (t.doInvoke() < NORMAL && ex == null) + ex = t.getException(); } for (int i = 1; i <= last; ++i) { ForkJoinTask t = tasks[i]; if (t != null) { if (ex != null) t.cancel(false); - else { - t.quietlyJoin(); - if (ex == null) - ex = t.getException(); - } + else if (t.doJoin() < NORMAL && ex == null) + ex = t.getException(); } } if (ex != null) - rethrowException(ex); + UNSAFE.throwException(ex); } /** - * Forks all tasks in the collection, returning when - * isDone holds for all of them. If any task - * encounters an exception, others may be cancelled. This method - * may be invoked only from within ForkJoinTask - * computations. Attempts to invoke in other contexts resul!t in - * exceptions or errors possibly including ClassCastException. + * Forks all tasks in the specified collection, returning when + * {@code isDone} holds for each task or an (unchecked) exception + * is encountered, in which case the exception is rethrown. If + * more than one task encounters an exception, then this method + * throws any one of these exceptions. If any task encounters an + * exception, others may be cancelled. However, the execution + * status of individual tasks is not guaranteed upon exceptional + * return. The status of each task may be obtained using {@link + * #getException()} and related methods to check if they have been + * cancelled, completed normally or exceptionally, or left + * unprocessed. + * + *

This method may be invoked only from within {@code + * ForkJoinPool} computations (as may be determined using method + * {@link #inForkJoinPool}). Attempts to invoke in other contexts + * result in exceptions or errors, possibly including {@code + * ClassCastException}. + * * @param tasks the collection of tasks - * @throws NullPointerException if tasks or any element are null. - * @throws RuntimeException or Error if any task did so. + * @return the tasks argument, to simplify usage + * @throws NullPointerException if tasks or any element are null */ - public static void invokeAll(Collection> tasks) { - if (!(tasks instanceof List)) { - invokeAll(tasks.toArray(new ForkJoinTask[tasks.size()])); - return; + public static > Collection invokeAll(Collection tasks) { + if (!(tasks instanceof RandomAccess) || !(tasks instanceof List)) { + invokeAll(tasks.toArray(new ForkJoinTask[tasks.size()])); + return tasks; } + @SuppressWarnings("unchecked") List> ts = - (List>)tasks; + (List>) tasks; Throwable ex = null; int last = ts.size() - 1; for (int i = last; i >= 0; --i) { @@ -598,253 +746,310 @@ public abstract class ForkJoinTask implements Future, Serializable { } else if (i != 0) t.fork(); - else { - t.quietlyInvoke(); - if (ex == null) - ex = t.getException(); - } + else if (t.doInvoke() < NORMAL && ex == null) + ex = t.getException(); } for (int i = 1; i <= last; ++i) { ForkJoinTask t = ts.get(i); if (t != null) { if (ex != null) t.cancel(false); - else { - t.quietlyJoin(); - if (ex == null) - ex = t.getException(); - } + else if (t.doJoin() < NORMAL && ex == null) + ex = t.getException(); } } if (ex != null) - rethrowException(ex); + UNSAFE.throwException(ex); + return tasks; } /** - * Returns true if the computation performed by this task has - * completed (or has been cancelled). - * @return true if this computation has completed + * Attempts to cancel execution of this task. This attempt will + * fail if the task has already completed or could not be + * cancelled for some other reason. If successful, and this task + * has not started when {@code cancel} is called, execution of + * this task is suppressed. After this method returns + * successfully, unless there is an intervening call to {@link + * #reinitialize}, subsequent calls to {@link #isCancelled}, + * {@link #isDone}, and {@code cancel} will return {@code true} + * and calls to {@link #join} and related methods will result in + * {@code CancellationException}. + * + *

This method may be overridden in subclasses, but if so, must + * still ensure that these properties hold. In particular, the + * {@code cancel} method itself must not throw exceptions. + * + *

This method is designed to be invoked by other + * tasks. To terminate the current task, you can just return or + * throw an unchecked exception from its computation method, or + * invoke {@link #completeExceptionally}. + * + * @param mayInterruptIfRunning this value has no effect in the + * default implementation because interrupts are not used to + * control cancellation. + * + * @return {@code true} if this task is now cancelled */ - public final boolean isDone() { - return status < 0; + public boolean cancel(boolean mayInterruptIfRunning) { + return setCompletion(CANCELLED) == CANCELLED; } /** - * Returns true if this task was cancelled. - * @return true if this task was cancelled + * Cancels, ignoring any exceptions thrown by cancel. Used during + * worker and pool shutdown. Cancel is spec'ed not to throw any + * exceptions, but if it does anyway, we have no recourse during + * shutdown, so guard against this case. */ + final void cancelIgnoringExceptions() { + try { + cancel(false); + } catch (Throwable ignore) { + } + } + + public final boolean isDone() { + return status < 0; + } + public final boolean isCancelled() { - return (status & COMPLETION_MASK) == CANCELLED; + return status == CANCELLED; } /** - * Asserts that the results of this task's computation will not be - * used. If a cancellation occurs before atempting to execute this - * task, then execution will be suppressed, isCancelled - * will report true, and join will result in a - * CancellationException being thrown. Otherwise, when - * cancellation races with completion, there are no guarantees - * about whether isCancelled will report true, whether - * join will return normally or via an exception, or - * whether these behaviors will remain consistent upon repeated - * invocation. - * - *

This method may be overridden in subclasses, but if so, must - * still ensure that these minimal properties hold. In particular, - * the cancel method itself must not throw exceptions. - * - *

This method is designed to be invoked by other - * tasks. To terminate the current task, you can just return or - * throw an unchecked exception from its computation method, or - * invoke completeExceptionally. - * - * @param mayInterruptIfRunning this value is ignored in the - * default implementation because tasks are not in general - * cancelled via interruption. + * Returns {@code true} if this task threw an exception or was cancelled. * - * @return true if this task is now cancelled + * @return {@code true} if this task threw an exception or was cancelled */ - public boolean cancel(boolean mayInterruptIfRunning) { - setCompletion(CANCELLED); - return (status & COMPLETION_MASK) == CANCELLED; + public final boolean isCompletedAbnormally() { + return status < NORMAL; } /** - * Returns true if this task threw an exception or was cancelled - * @return true if this task threw an exception or was cancelled + * Returns {@code true} if this task completed without throwing an + * exception and was not cancelled. + * + * @return {@code true} if this task completed without throwing an + * exception and was not cancelled */ - public final boolean isCompletedAbnormally() { - return (status & COMPLETION_MASK) < NORMAL; + public final boolean isCompletedNormally() { + return status == NORMAL; } /** * Returns the exception thrown by the base computation, or a - * CancellationException if cancelled, or null if none or if the - * method has not yet completed. - * @return the exception, or null if none + * {@code CancellationException} if cancelled, or {@code null} if + * none or if the method has not yet completed. + * + * @return the exception, or {@code null} if none */ public final Throwable getException() { - int s = status & COMPLETION_MASK; - if (s >= NORMAL) - return null; - if (s == CANCELLED) - return new CancellationException(); - return exceptionMap.get(this); + int s = status; + return ((s >= NORMAL) ? null : + (s == CANCELLED) ? new CancellationException() : + getThrowableException()); } /** * Completes this task abnormally, and if not already aborted or * cancelled, causes it to throw the given exception upon - * join and related operations. This method may be used + * {@code join} and related operations. This method may be used * to induce exceptions in asynchronous tasks, or to force * completion of tasks that would not otherwise complete. Its use - * in other situations is likely to be wrong. This method is - * overridable, but overridden versions must invoke super + * in other situations is discouraged. This method is + * overridable, but overridden versions must invoke {@code super} * implementation to maintain guarantees. * - * @param ex the exception to throw. If this exception is - * not a RuntimeException or Error, the actual exception thrown - * will be a RuntimeException with cause ex. + * @param ex the exception to throw. If this exception is not a + * {@code RuntimeException} or {@code Error}, the actual exception + * thrown will be a {@code RuntimeException} with cause {@code ex}. */ public void completeExceptionally(Throwable ex) { - setDoneExceptionally((ex instanceof RuntimeException) || - (ex instanceof Error)? ex : - new RuntimeException(ex)); + setExceptionalCompletion((ex instanceof RuntimeException) || + (ex instanceof Error) ? ex : + new RuntimeException(ex)); } /** * Completes this task, and if not already aborted or cancelled, - * returning a null result upon join and related - * operations. This method may be used to provide results for - * asynchronous tasks, or to provide alternative handling for - * tasks that would not otherwise complete normally. Its use in - * other situations is likely to be wrong. This method is - * overridable, but overridden versions must invoke super - * implementation to maintain guarantees. + * returning the given value as the result of subsequent + * invocations of {@code join} and related operations. This method + * may be used to provide results for asynchronous tasks, or to + * provide alternative handling for tasks that would not otherwise + * complete normally. Its use in other situations is + * discouraged. This method is overridable, but overridden + * versions must invoke {@code super} implementation to maintain + * guarantees. * - * @param value the result value for this task. + * @param value the result value for this task */ public void complete(V value) { try { setRawResult(value); - } catch(Throwable rex) { - setDoneExceptionally(rex); + } catch (Throwable rex) { + setExceptionalCompletion(rex); return; } - setNormalCompletion(); - } - - public final V get() throws InterruptedException, ExecutionException { - ForkJoinWorkerThread w = getWorker(); - if (w == null || status < 0 || !w.unpushTask(this) || !tryQuietlyInvoke()) - awaitDone(w, true); - return reportFutureResult(); - } - - public final V get(long timeout, TimeUnit unit) - throws InterruptedException, ExecutionException, TimeoutException { - ForkJoinWorkerThread w = getWorker(); - if (w == null || status < 0 || !w.unpushTask(this) || !tryQuietlyInvoke()) - awaitDone(w, unit.toNanos(timeout)); - return reportTimedFutureResult(); + setCompletion(NORMAL); } /** - * Possibly executes other tasks until this task is ready, then - * returns the result of the computation. This method may be more - * efficient than join, but is only applicable when - * there are no potemtial dependencies between continuation of the - * current task and that of any other task that might be executed - * while helping. (This usually holds for pure divide-and-conquer - * tasks). This method may be invoked only from within - * ForkJoinTask computations. Attempts to invoke in other contexts - * resul!t in exceptions or errors possibly including ClassCastException. + * Waits if necessary for the computation to complete, and then + * retrieves its result. + * * @return the computed result + * @throws CancellationException if the computation was cancelled + * @throws ExecutionException if the computation threw an + * exception + * @throws InterruptedException if the current thread is not a + * member of a ForkJoinPool and was interrupted while waiting */ - public final V helpJoin() { - ForkJoinWorkerThread w = (ForkJoinWorkerThread)(Thread.currentThread()); - if (status < 0 || !w.unpushTask(this) || !tryExec()) - reportException(busyJoin(w)); + public final V get() throws InterruptedException, ExecutionException { + int s = (Thread.currentThread() instanceof ForkJoinWorkerThread) ? + doJoin() : externalInterruptibleAwaitDone(0L); + Throwable ex; + if (s == CANCELLED) + throw new CancellationException(); + if (s == EXCEPTIONAL && (ex = getThrowableException()) != null) + throw new ExecutionException(ex); return getRawResult(); } /** - * Possibly executes other tasks until this task is ready. This - * method may be invoked only from within ForkJoinTask - * computations. Attempts to invoke in other contexts resul!t in - * exceptions or errors possibly including ClassCastException. + * Waits if necessary for at most the given time for the computation + * to complete, and then retrieves its result, if available. + * + * @param timeout the maximum time to wait + * @param unit the time unit of the timeout argument + * @return the computed result + * @throws CancellationException if the computation was cancelled + * @throws ExecutionException if the computation threw an + * exception + * @throws InterruptedException if the current thread is not a + * member of a ForkJoinPool and was interrupted while waiting + * @throws TimeoutException if the wait timed out */ - public final void quietlyHelpJoin() { - if (status >= 0) { - ForkJoinWorkerThread w = - (ForkJoinWorkerThread)(Thread.currentThread()); - if (!w.unpushTask(this) || !tryQuietlyInvoke()) - busyJoin(w); + public final V get(long timeout, TimeUnit unit) + throws InterruptedException, ExecutionException, TimeoutException { + Thread t = Thread.currentThread(); + if (t instanceof ForkJoinWorkerThread) { + ForkJoinWorkerThread w = (ForkJoinWorkerThread) t; + long nanos = unit.toNanos(timeout); + if (status >= 0) { + boolean completed = false; + if (w.unpushTask(this)) { + try { + completed = exec(); + } catch (Throwable rex) { + setExceptionalCompletion(rex); + } + } + if (completed) + setCompletion(NORMAL); + else if (status >= 0 && nanos > 0) + w.pool.timedAwaitJoin(this, nanos); + } + } + else { + long millis = unit.toMillis(timeout); + if (millis > 0) + externalInterruptibleAwaitDone(millis); + } + int s = status; + if (s != NORMAL) { + Throwable ex; + if (s == CANCELLED) + throw new CancellationException(); + if (s != EXCEPTIONAL) + throw new TimeoutException(); + if ((ex = getThrowableException()) != null) + throw new ExecutionException(ex); } + return getRawResult(); } /** - * Joins this task, without returning its result or throwing an + * Joins this task, without returning its result or throwing its * exception. This method may be useful when processing * collections of tasks when some have been cancelled or otherwise * known to have aborted. */ public final void quietlyJoin() { - if (status >= 0) { - ForkJoinWorkerThread w = getWorker(); - if (w == null || !w.unpushTask(this) || !tryQuietlyInvoke()) - awaitDone(w, true); - } + doJoin(); } /** * Commences performing this task and awaits its completion if - * necessary, without returning its result or throwing an - * exception. This method may be useful when processing - * collections of tasks when some have been cancelled or otherwise - * known to have aborted. + * necessary, without returning its result or throwing its + * exception. */ public final void quietlyInvoke() { - if (status >= 0 && !tryQuietlyInvoke()) - quietlyJoin(); + doInvoke(); } /** * Possibly executes tasks until the pool hosting the current task - * {@link ForkJoinPool#isQuiescent}. This method may be of use in - * designs in which many tasks are forked, but none are explicitly - * joined, instead executing them until all are processed. + * {@link ForkJoinPool#isQuiescent is quiescent}. This method may + * be of use in designs in which many tasks are forked, but none + * are explicitly joined, instead executing them until all are + * processed. + * + *

This method may be invoked only from within {@code + * ForkJoinPool} computations (as may be determined using method + * {@link #inForkJoinPool}). Attempts to invoke in other contexts + * result in exceptions or errors, possibly including {@code + * ClassCastException}. */ public static void helpQuiesce() { - ((ForkJoinWorkerThread)(Thread.currentThread())). - helpQuiescePool(); + ((ForkJoinWorkerThread) Thread.currentThread()) + .helpQuiescePool(); } /** * Resets the internal bookkeeping state of this task, allowing a - * subsequent fork. This method allows repeated reuse of + * subsequent {@code fork}. This method allows repeated reuse of * this task, but only if reuse occurs when this task has either * never been forked, or has been forked, then completed and all * outstanding joins of this task have also completed. Effects - * under any other usage conditions are not guaranteed, and are - * almost surely wrong. This method may be useful when executing + * under any other usage conditions are not guaranteed. + * This method may be useful when executing * pre-constructed trees of subtasks in loops. + * + *

Upon completion of this method, {@code isDone()} reports + * {@code false}, and {@code getException()} reports {@code + * null}. However, the value returned by {@code getRawResult} is + * unaffected. To clear this value, you can invoke {@code + * setRawResult(null)}. */ public void reinitialize() { - if ((status & COMPLETION_MASK) == EXCEPTIONAL) - exceptionMap.remove(this); - status = 0; + if (status == EXCEPTIONAL) + clearExceptionalCompletion(); + else + status = 0; } /** * Returns the pool hosting the current task execution, or null - * if this task is executing outside of any pool. - * @return the pool, or null if none. + * if this task is executing outside of any ForkJoinPool. + * + * @see #inForkJoinPool + * @return the pool, or {@code null} if none */ public static ForkJoinPool getPool() { Thread t = Thread.currentThread(); - return ((t instanceof ForkJoinWorkerThread)? - ((ForkJoinWorkerThread)t).pool : null); + return (t instanceof ForkJoinWorkerThread) ? + ((ForkJoinWorkerThread) t).pool : null; + } + + /** + * Returns {@code true} if the current thread is a {@link + * ForkJoinWorkerThread} executing as a ForkJoinPool computation. + * + * @return {@code true} if the current thread is a {@link + * ForkJoinWorkerThread} executing as a ForkJoinPool computation, + * or {@code false} otherwise + */ + public static boolean inForkJoinPool() { + return Thread.currentThread() instanceof ForkJoinWorkerThread; } /** @@ -853,13 +1058,19 @@ public abstract class ForkJoinTask implements Future, Serializable { * by the current thread, and has not commenced executing in * another thread. This method may be useful when arranging * alternative local processing of tasks that could have been, but - * were not, stolen. This method may be invoked only from within - * ForkJoinTask computations. Attempts to invoke in other contexts - * result in exceptions or errors possibly including ClassCastException. - * @return true if unforked + * were not, stolen. + * + *

This method may be invoked only from within {@code + * ForkJoinPool} computations (as may be determined using method + * {@link #inForkJoinPool}). Attempts to invoke in other contexts + * result in exceptions or errors, possibly including {@code + * ClassCastException}. + * + * @return {@code true} if unforked */ public boolean tryUnfork() { - return ((ForkJoinWorkerThread)(Thread.currentThread())).unpushTask(this); + return ((ForkJoinWorkerThread) Thread.currentThread()) + .unpushTask(this); } /** @@ -867,15 +1078,22 @@ public abstract class ForkJoinTask implements Future, Serializable { * forked by the current worker thread but not yet executed. This * value may be useful for heuristic decisions about whether to * fork other tasks. + * + *

This method may be invoked only from within {@code + * ForkJoinPool} computations (as may be determined using method + * {@link #inForkJoinPool}). Attempts to invoke in other contexts + * result in exceptions or errors, possibly including {@code + * ClassCastException}. + * * @return the number of tasks */ public static int getQueuedTaskCount() { - return ((ForkJoinWorkerThread)(Thread.currentThread())). - getQueueSize(); + return ((ForkJoinWorkerThread) Thread.currentThread()) + .getQueueSize(); } /** - * Returns a estimate of how many more locally queued tasks are + * Returns an estimate of how many more locally queued tasks are * held by the current worker thread than there are other worker * threads that might steal them. This value may be useful for * heuristic decisions about whether to fork other tasks. In many @@ -883,23 +1101,30 @@ public abstract class ForkJoinTask implements Future, Serializable { * aim to maintain a small constant surplus (for example, 3) of * tasks, and to process computations locally if this threshold is * exceeded. + * + *

This method may be invoked only from within {@code + * ForkJoinPool} computations (as may be determined using method + * {@link #inForkJoinPool}). Attempts to invoke in other contexts + * result in exceptions or errors, possibly including {@code + * ClassCastException}. + * * @return the surplus number of tasks, which may be negative */ public static int getSurplusQueuedTaskCount() { - return ((ForkJoinWorkerThread)(Thread.currentThread())) + return ((ForkJoinWorkerThread) Thread.currentThread()) .getEstimatedSurplusTaskCount(); } // Extension methods /** - * Returns the result that would be returned by join, - * even if this task completed abnormally, or null if this task is - * not known to have been completed. This method is designed to - * aid debugging, as well as to support extensions. Its use in any - * other context is discouraged. + * Returns the result that would be returned by {@link #join}, even + * if this task completed abnormally, or {@code null} if this task + * is not known to have been completed. This method is designed + * to aid debugging, as well as to support extensions. Its use in + * any other context is discouraged. * - * @return the result, or null if not completed. + * @return the result, or {@code null} if not completed */ public abstract V getRawResult(); @@ -918,42 +1143,53 @@ public abstract class ForkJoinTask implements Future, Serializable { * called otherwise. The return value controls whether this task * is considered to be done normally. It may return false in * asynchronous actions that require explicit invocations of - * complete to become joinable. It may throw exceptions - * to indicate abnormal exit. - * @return true if completed normally - * @throws Error or RuntimeException if encountered during computation + * {@link #complete} to become joinable. It may also throw an + * (unchecked) exception to indicate abnormal exit. + * + * @return {@code true} if completed normally */ protected abstract boolean exec(); /** - * Returns, but does not unschedule or execute, the task queued by - * the current thread but not yet executed, if one is + * Returns, but does not unschedule or execute, a task queued by + * the current thread but not yet executed, if one is immediately * available. There is no guarantee that this task will actually - * be polled or executed next. This method is designed primarily - * to support extensions, and is unlikely to be useful otherwise. - * This method may be invoked only from within ForkJoinTask - * computations. Attempts to invoke in other contexts result in - * exceptions or errors possibly including ClassCastException. + * be polled or executed next. Conversely, this method may return + * null even if a task exists but cannot be accessed without + * contention with other threads. This method is designed + * primarily to support extensions, and is unlikely to be useful + * otherwise. * - * @return the next task, or null if none are available + *

This method may be invoked only from within {@code + * ForkJoinPool} computations (as may be determined using method + * {@link #inForkJoinPool}). Attempts to invoke in other contexts + * result in exceptions or errors, possibly including {@code + * ClassCastException}. + * + * @return the next task, or {@code null} if none are available */ protected static ForkJoinTask peekNextLocalTask() { - return ((ForkJoinWorkerThread)(Thread.currentThread())).peekTask(); + return ((ForkJoinWorkerThread) Thread.currentThread()) + .peekTask(); } /** * Unschedules and returns, without executing, the next task * queued by the current thread but not yet executed. This method * is designed primarily to support extensions, and is unlikely to - * be useful otherwise. This method may be invoked only from - * within ForkJoinTask computations. Attempts to invoke in other - * contexts result in exceptions or errors possibly including - * ClassCastException. + * be useful otherwise. + * + *

This method may be invoked only from within {@code + * ForkJoinPool} computations (as may be determined using method + * {@link #inForkJoinPool}). Attempts to invoke in other contexts + * result in exceptions or errors, possibly including {@code + * ClassCastException}. * - * @return the next task, or null if none are available + * @return the next task, or {@code null} if none are available */ protected static ForkJoinTask pollNextLocalTask() { - return ((ForkJoinWorkerThread)(Thread.currentThread())).pollLocalTask(); + return ((ForkJoinWorkerThread) Thread.currentThread()) + .pollLocalTask(); } /** @@ -961,19 +1197,115 @@ public abstract class ForkJoinTask implements Future, Serializable { * queued by the current thread but not yet executed, if one is * available, or if not available, a task that was forked by some * other thread, if available. Availability may be transient, so a - * null result does not necessarily imply quiecence + * {@code null} result does not necessarily imply quiescence * of the pool this task is operating in. This method is designed * primarily to support extensions, and is unlikely to be useful - * otherwise. This method may be invoked only from within - * ForkJoinTask computations. Attempts to invoke in other contexts - * result in exceptions or errors possibly including - * ClassCastException. + * otherwise. * - * @return a task, or null if none are available + *

This method may be invoked only from within {@code + * ForkJoinPool} computations (as may be determined using method + * {@link #inForkJoinPool}). Attempts to invoke in other contexts + * result in exceptions or errors, possibly including {@code + * ClassCastException}. + * + * @return a task, or {@code null} if none are available */ protected static ForkJoinTask pollTask() { - return ((ForkJoinWorkerThread)(Thread.currentThread())). - pollTask(); + return ((ForkJoinWorkerThread) Thread.currentThread()) + .pollTask(); + } + + /** + * Adaptor for Runnables. This implements RunnableFuture + * to be compliant with AbstractExecutorService constraints + * when used in ForkJoinPool. + */ + 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; + } + + /** + * Returns a new {@code ForkJoinTask} that performs the {@code run} + * method of the given {@code Runnable} as its action, and returns + * a null result upon {@link #join}. + * + * @param runnable the runnable action + * @return the task + */ + public static ForkJoinTask adapt(Runnable runnable) { + return new AdaptedRunnable(runnable, null); + } + + /** + * Returns a new {@code ForkJoinTask} that performs the {@code run} + * method of the given {@code Runnable} as its action, and returns + * the given result upon {@link #join}. + * + * @param runnable the runnable action + * @param result the result upon completion + * @return the task + */ + public static ForkJoinTask adapt(Runnable runnable, T result) { + return new AdaptedRunnable(runnable, result); + } + + /** + * Returns a new {@code ForkJoinTask} that performs the {@code call} + * method of the given {@code Callable} as its action, and returns + * its result upon {@link #join}, translating any checked exceptions + * encountered into {@code RuntimeException}. + * + * @param callable the callable action + * @return the task + */ + public static ForkJoinTask adapt(Callable callable) { + return new AdaptedCallable(callable); } // Serialization support @@ -981,10 +1313,10 @@ public abstract class ForkJoinTask implements Future, Serializable { private static final long serialVersionUID = -7721805057305804111L; /** - * Save the state to a stream. + * Saves the state to a stream (that is, serializes it). * * @serialData the current run status and the exception thrown - * during execution, or null if none. + * during execution, or {@code null} if none * @param s the stream */ private void writeObject(java.io.ObjectOutputStream s) @@ -994,70 +1326,32 @@ public abstract class ForkJoinTask implements Future, Serializable { } /** - * Reconstitute the instance from a stream. + * Reconstitutes the instance from a stream (that is, deserializes it). + * * @param s the stream */ private void readObject(java.io.ObjectInputStream s) throws java.io.IOException, ClassNotFoundException { s.defaultReadObject(); - status &= ~INTERNAL_SIGNAL_MASK; // clear internal signal counts - status |= EXTERNAL_SIGNAL; // conservatively set external signal Object ex = s.readObject(); if (ex != null) - setDoneExceptionally((Throwable)ex); + setExceptionalCompletion((Throwable)ex); } - // Temporary Unsafe mechanics for preliminary release - private static Unsafe getUnsafe() throws Throwable { - try { - return Unsafe.getUnsafe(); - } catch (SecurityException se) { - try { - return java.security.AccessController.doPrivileged - (new java.security.PrivilegedExceptionAction() { - public Unsafe run() throws Exception { - return getUnsafePrivileged(); - }}); - } catch (java.security.PrivilegedActionException e) { - throw 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, Unsafe unsafe) - throws NoSuchFieldException { - // do not use _unsafe to avoid NPE - return unsafe.objectFieldOffset - (ForkJoinTask.class.getDeclaredField(fieldName)); - } - - static final Unsafe _unsafe; - static final long statusOffset; - + // Unsafe mechanics + private static final sun.misc.Unsafe UNSAFE; + private static final long statusOffset; static { - Unsafe tmpUnsafe = null; - long tmpStatusOffset = 0; + exceptionTableLock = new ReentrantLock(); + exceptionTableRefQueue = new ReferenceQueue(); + exceptionTable = new ExceptionNode[EXCEPTION_MAP_CAPACITY]; try { - tmpUnsafe = getUnsafe(); - tmpStatusOffset = fieldOffset("status", tmpUnsafe); - } catch (Throwable e) { - // Ignore the failure to load sun.misc.Unsafe on Android so - // that platform can use the actor library without the - // fork/join scheduler. - String vmVendor = System.getProperty("java.vm.vendor"); - if (!vmVendor.contains("Android")) { - throw new RuntimeException("Could not initialize intrinsics", e); - } + UNSAFE = sun.misc.Unsafe.getUnsafe(); + statusOffset = UNSAFE.objectFieldOffset + (ForkJoinTask.class.getDeclaredField("status")); + } catch (Exception e) { + throw new Error(e); } - _unsafe = tmpUnsafe; - statusOffset = tmpStatusOffset; } } diff --git a/src/forkjoin/scala/concurrent/forkjoin/ForkJoinWorkerThread.java b/src/forkjoin/scala/concurrent/forkjoin/ForkJoinWorkerThread.java index b4d889750c..79879b19c7 100644 --- a/src/forkjoin/scala/concurrent/forkjoin/ForkJoinWorkerThread.java +++ b/src/forkjoin/scala/concurrent/forkjoin/ForkJoinWorkerThread.java @@ -1,224 +1,287 @@ /* * Written by Doug Lea with assistance from members of JCP JSR-166 * Expert Group and released to the public domain, as explained at - * http://creativecommons.org/licenses/publicdomain + * http://creativecommons.org/publicdomain/zero/1.0/ */ package scala.concurrent.forkjoin; -import java.util.*; -import java.util.concurrent.*; -import java.util.concurrent.atomic.*; -import java.util.concurrent.locks.*; -import sun.misc.Unsafe; -import java.lang.reflect.*; + +import java.util.Collection; +import java.util.concurrent.RejectedExecutionException; /** - * A thread managed by a {@link ForkJoinPool}. This class is - * subclassable solely for the sake of adding functionality -- there - * are no overridable methods dealing with scheduling or - * execution. However, you can override initialization and termination - * methods surrounding the main task processing loop. If you do - * create such a subclass, you will also need to supply a custom - * ForkJoinWorkerThreadFactory to use it in a ForkJoinPool. + * A thread managed by a {@link ForkJoinPool}, which executes + * {@link ForkJoinTask}s. + * This class is subclassable solely for the sake of adding + * functionality -- there are no overridable methods dealing with + * scheduling or execution. However, you can override initialization + * and termination methods surrounding the main task processing loop. + * If you do create such a subclass, you will also need to supply a + * custom {@link ForkJoinPool.ForkJoinWorkerThreadFactory} to use it + * in a {@code ForkJoinPool}. * + * @since 1.7 + * @author Doug Lea */ public class ForkJoinWorkerThread extends Thread { /* - * Algorithm overview: + * Overview: + * + * ForkJoinWorkerThreads are managed by ForkJoinPools and perform + * ForkJoinTasks. This class includes bookkeeping in support of + * worker activation, suspension, and lifecycle control described + * in more detail in the internal documentation of class + * ForkJoinPool. And as described further below, this class also + * includes special-cased support for some ForkJoinTask + * methods. But the main mechanics involve work-stealing: * - * 1. Work-Stealing: Work-stealing queues are special forms of - * Deques that support only three of the four possible - * end-operations -- push, pop, and deq (aka steal), and only do - * so under the constraints that push and pop are called only from - * the owning thread, while deq may be called from other threads. - * (If you are unfamiliar with them, you probably want to read - * Herlihy and Shavit's book "The Art of Multiprocessor - * programming", chapter 16 describing these in more detail before - * proceeding.) The main work-stealing queue design is roughly - * similar to "Dynamic Circular Work-Stealing Deque" by David - * Chase and Yossi Lev, SPAA 2005 - * (http://research.sun.com/scalable/pubs/index.html). The main - * difference ultimately stems from gc requirements that we null - * out taken slots as soon as we can, to maintain as small a - * footprint as possible even in programs generating huge numbers - * of tasks. To accomplish this, we shift the CAS arbitrating pop - * vs deq (steal) from being on the indices ("base" and "sp") to - * the slots themselves (mainly via method "casSlotNull()"). So, - * both a successful pop and deq mainly entail CAS'ing a nonnull - * slot to null. Because we rely on CASes of references, we do - * not need tag bits on base or sp. They are simple ints as used - * in any circular array-based queue (see for example ArrayDeque). + * Work-stealing queues are special forms of Deques that support + * only three of the four possible end-operations -- push, pop, + * and deq (aka steal), under the further constraints that push + * and pop are called only from the owning thread, while deq may + * be called from other threads. (If you are unfamiliar with + * them, you probably want to read Herlihy and Shavit's book "The + * Art of Multiprocessor programming", chapter 16 describing these + * in more detail before proceeding.) The main work-stealing + * queue design is roughly similar to those in the papers "Dynamic + * Circular Work-Stealing Deque" by Chase and Lev, SPAA 2005 + * (http://research.sun.com/scalable/pubs/index.html) and + * "Idempotent work stealing" by Michael, Saraswat, and Vechev, + * PPoPP 2009 (http://portal.acm.org/citation.cfm?id=1504186). + * The main differences ultimately stem from gc requirements that + * we null out taken slots as soon as we can, to maintain as small + * a footprint as possible even in programs generating huge + * numbers of tasks. To accomplish this, we shift the CAS + * arbitrating pop vs deq (steal) from being on the indices + * ("queueBase" and "queueTop") to the slots themselves (mainly + * via method "casSlotNull()"). So, both a successful pop and deq + * mainly entail a CAS of a slot from non-null to null. Because + * we rely on CASes of references, we do not need tag bits on + * queueBase or queueTop. They are simple ints as used in any + * circular array-based queue (see for example ArrayDeque). * Updates to the indices must still be ordered in a way that - * guarantees that (sp - base) > 0 means the queue is empty, but - * otherwise may err on the side of possibly making the queue + * guarantees that queueTop == queueBase means the queue is empty, + * but otherwise may err on the side of possibly making the queue * appear nonempty when a push, pop, or deq have not fully * committed. Note that this means that the deq operation, * considered individually, is not wait-free. One thief cannot * successfully continue until another in-progress one (or, if * previously empty, a push) completes. However, in the - * aggregate, we ensure at least probablistic non-blockingness. If - * an attempted steal fails, a thief always chooses a different + * aggregate, we ensure at least probabilistic non-blockingness. + * If an attempted steal fails, a thief always chooses a different * random victim target to try next. So, in order for one thief to * progress, it suffices for any in-progress deq or new push on - * any empty queue to complete. One reason this works well here is - * that apparently-nonempty often means soon-to-be-stealable, - * which gives threads a chance to activate if necessary before - * stealing (see below). + * any empty queue to complete. + * + * This approach also enables support for "async mode" where local + * task processing is in FIFO, not LIFO order; simply by using a + * version of deq rather than pop when locallyFifo is true (as set + * by the ForkJoinPool). This allows use in message-passing + * frameworks in which tasks are never joined. However neither + * mode considers 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.) + * + * When a worker would otherwise be blocked waiting to join a + * task, it first tries a form of linear helping: Each worker + * records (in field currentSteal) the most recent task it stole + * from some other worker. Plus, it records (in field currentJoin) + * the task it is currently actively joining. Method joinTask uses + * these markers to try to find a worker to help (i.e., steal back + * a task from and execute it) that could hasten completion of the + * actively joined task. In essence, the joiner executes a task + * that would be on its own local deque had the to-be-joined task + * not been stolen. This may be seen as a conservative variant of + * the approach in Wagner & Calder "Leapfrogging: a portable + * technique for implementing efficient futures" SIGPLAN Notices, + * 1993 (http://portal.acm.org/citation.cfm?id=155354). It differs + * in that: (1) We only maintain dependency links across workers + * upon steals, rather than use per-task bookkeeping. This may + * require a linear scan of workers array to locate stealers, but + * usually doesn't because stealers leave hints (that may become + * stale/wrong) of where to locate them. This isolates cost to + * when it is needed, rather than adding to per-task overhead. + * (2) It is "shallow", ignoring nesting and potentially cyclic + * mutual steals. (3) It is intentionally racy: field currentJoin + * is updated only while actively joining, which means that we + * miss links in the chain during long-lived tasks, GC stalls etc + * (which is OK since blocking in such cases is usually a good + * idea). (4) We bound the number of attempts to find work (see + * MAX_HELP) and fall back to suspending the worker and if + * necessary replacing it with another. * - * Efficient implementation of this approach currently relies on - * an uncomfortable amount of "Unsafe" mechanics. To maintain - * correct orderings, reads and writes of variable base require - * volatile ordering. Variable sp does not require volatile write - * but needs cheaper store-ordering on writes. Because they are - * protected by volatile base reads, reads of the queue array and - * its slots do not need volatile load semantics, but writes (in - * push) require store order and CASes (in pop and deq) require - * (volatile) CAS semantics. Since these combinations aren't - * supported using ordinary volatiles, the only way to accomplish - * these effciently is to use direct Unsafe calls. (Using external + * Efficient implementation of these algorithms currently relies + * on an uncomfortable amount of "Unsafe" mechanics. To maintain + * correct orderings, reads and writes of variable queueBase + * require volatile ordering. Variable queueTop need not be + * volatile because non-local reads always follow those of + * queueBase. Similarly, because they are protected by volatile + * queueBase reads, reads of the queue array and its slots by + * other threads do not need volatile load semantics, but writes + * (in push) require store order and CASes (in pop and deq) + * require (volatile) CAS semantics. (Michael, Saraswat, and + * Vechev's algorithm has similar properties, but without support + * for nulling slots.) Since these combinations aren't supported + * using ordinary volatiles, the only way to accomplish these + * efficiently is to use direct Unsafe calls. (Using external * AtomicIntegers and AtomicReferenceArrays for the indices and * array is significantly slower because of memory locality and - * indirection effects.) Further, performance on most platforms is - * very sensitive to placement and sizing of the (resizable) queue - * array. Even though these queues don't usually become all that - * big, the initial size must be large enough to counteract cache + * indirection effects.) + * + * Further, performance on most platforms is very sensitive to + * placement and sizing of the (resizable) queue array. Even + * though these queues don't usually become all that big, the + * initial size must be large enough to counteract cache * contention effects across multiple queues (especially in the * presence of GC cardmarking). Also, to improve thread-locality, - * queues are currently initialized immediately after the thread - * gets the initial signal to start processing tasks. However, - * all queue-related methods except pushTask are written in a way - * that allows them to instead be lazily allocated and/or disposed - * of when empty. All together, these low-level implementation - * choices produce as much as a factor of 4 performance - * improvement compared to naive implementations, and enable the - * processing of billions of tasks per second, sometimes at the - * expense of ugliness. - * - * 2. Run control: The primary run control is based on a global - * counter (activeCount) held by the pool. It uses an algorithm - * similar to that in Herlihy and Shavit section 17.6 to cause - * threads to eventually block when all threads declare they are - * inactive. (See variable "scans".) For this to work, threads - * must be declared active when executing tasks, and before - * stealing a task. They must be inactive before blocking on the - * Pool Barrier (awaiting a new submission or other Pool - * event). In between, there is some free play which we take - * advantage of to avoid contention and rapid flickering of the - * global activeCount: If inactive, we activate only if a victim - * queue appears to be nonempty (see above). Similarly, a thread - * tries to inactivate only after a full scan of other threads. - * The net effect is that contention on activeCount is rarely a - * measurable performance issue. (There are also a few other cases - * where we scan for work rather than retry/block upon - * contention.) - * - * 3. Selection control. We maintain policy of always choosing to - * run local tasks rather than stealing, and always trying to - * steal tasks before trying to run a new submission. All steals - * are currently performed in randomly-chosen deq-order. It may be - * worthwhile to bias these with locality / anti-locality - * information, but doing this well probably requires more - * lower-level information from JVMs than currently provided. + * queues are initialized after starting. */ + /** + * Mask for pool indices encoded as shorts + */ + private static final int SMASK = 0xffff; + /** * Capacity of work-stealing queue array upon initialization. - * Must be a power of two. Initial size must be at least 2, but is + * Must be a power of two. Initial size must be at least 4, but is * padded to minimize cache effects. */ private static final int INITIAL_QUEUE_CAPACITY = 1 << 13; /** - * Maximum work-stealing queue array size. Must be less than or - * equal to 1 << 28 to ensure lack of index wraparound. (This - * is less than usual bounds, because we need leftshift by 3 - * to be in int range). + * Maximum size for queue array. Must be a power of two + * less than or equal to 1 << (31 - width of array entry) to + * ensure lack of index wraparound, but is capped at a lower + * value to help users trap runaway computations. */ - private static final int MAXIMUM_QUEUE_CAPACITY = 1 << 28; + private static final int MAXIMUM_QUEUE_CAPACITY = 1 << 24; // 16M /** - * The pool this thread works in. Accessed directly by ForkJoinTask + * The work-stealing queue array. Size must be a power of two. + * Initialized when started (as oposed to when constructed), to + * improve memory locality. */ - final ForkJoinPool pool; + ForkJoinTask[] queue; /** - * The work-stealing queue array. Size must be a power of two. - * Initialized when thread starts, to improve memory locality. + * The pool this thread works in. Accessed directly by ForkJoinTask. */ - private ForkJoinTask[] queue; + final ForkJoinPool pool; /** * Index (mod queue.length) of next queue slot to push to or pop - * from. It is written only by owner thread, via ordered store. - * Both sp and base are allowed to wrap around on overflow, but - * (sp - base) still estimates size. + * from. It is written only by owner thread, and accessed by other + * threads only after reading (volatile) queueBase. Both queueTop + * and queueBase are allowed to wrap around on overflow, but + * (queueTop - queueBase) still estimates size. */ - private volatile int sp; + int queueTop; /** * Index (mod queue.length) of least valid queue slot, which is * always the next position to steal from if nonempty. */ - private volatile int base; + volatile int queueBase; + + /** + * The index of most recent stealer, used as a hint to avoid + * traversal in method helpJoinTask. This is only a hint because a + * worker might have had multiple steals and this only holds one + * of them (usually the most current). Declared non-volatile, + * relying on other prevailing sync to keep reasonably current. + */ + int stealHint; + + /** + * Index of this worker in pool array. Set once by pool before + * running, and accessed directly by pool to locate this worker in + * its workers array. + */ + final int poolIndex; /** - * Activity status. When true, this worker is considered active. - * Must be false upon construction. It must be true when executing - * tasks, and BEFORE stealing a task. It must be false before - * calling pool.sync + * Encoded record for pool task waits. Usages are always + * surrounded by volatile reads/writes */ - private boolean active; + int nextWait; /** - * Run state of this worker. Supports simple versions of the usual - * shutdown/shutdownNow control. + * Complement of poolIndex, offset by count of entries of task + * waits. Accessed by ForkJoinPool to manage event waiters. */ - private volatile int runState; + volatile int eventCount; /** * Seed for random number generator for choosing steal victims. - * Uses Marsaglia xorshift. Must be nonzero upon initialization. + * Uses Marsaglia xorshift. Must be initialized as nonzero. */ - private int seed; + int seed; /** - * Number of steals, transferred to pool when idle + * Number of steals. Directly accessed (and reset) by pool when + * idle. */ - private int stealCount; + int stealCount; /** - * Index of this worker in pool array. Set once by pool before - * running, and accessed directly by pool during cleanup etc + * True if this worker should or did terminate */ - int poolIndex; + volatile boolean terminate; /** - * The last barrier event waited for. Accessed in pool callback - * methods, but only by current thread. + * Set to true before LockSupport.park; false on return */ - long lastEventCount; + volatile boolean parked; /** - * True if use local fifo, not default lifo, for local polling + * True if use local fifo, not default lifo, for local polling. + * Shadows value from ForkJoinPool. */ - private boolean locallyFifo; + final boolean locallyFifo; + + /** + * The task most recently stolen from another worker (or + * submission queue). All uses are surrounded by enough volatile + * reads/writes to maintain as non-volatile. + */ + ForkJoinTask currentSteal; + + /** + * The task currently being joined, set only when actively trying + * to help other stealers in helpJoinTask. All uses are surrounded + * by enough volatile reads/writes to maintain as non-volatile. + */ + ForkJoinTask currentJoin; /** * Creates a ForkJoinWorkerThread operating in the given pool. + * * @param pool the pool this thread works in * @throws NullPointerException if pool is null */ protected ForkJoinWorkerThread(ForkJoinPool pool) { - if (pool == null) throw new NullPointerException(); + super(pool.nextWorkerName()); this.pool = pool; - // Note: poolIndex is set by pool during construction - // Remaining initialization is deferred to onStart + int k = pool.registerWorker(this); + poolIndex = k; + eventCount = ~k & SMASK; // clear wait count + locallyFifo = pool.locallyFifo; + Thread.UncaughtExceptionHandler ueh = pool.ueh; + if (ueh != null) + setUncaughtExceptionHandler(ueh); + setDaemon(true); } - // Public access methods + // Public methods /** - * Returns the pool hosting this thread + * Returns the pool hosting this thread. + * * @return the pool */ public ForkJoinPool getPool() { @@ -231,543 +294,676 @@ public class ForkJoinWorkerThread extends Thread { * threads (minus one) that have ever been created in the pool. * This method may be useful for applications that track status or * collect results per-worker rather than per-task. - * @return the index number. + * + * @return the index number */ public int getPoolIndex() { return poolIndex; } - /** - * Establishes local first-in-first-out scheduling mode for forked - * tasks that are never joined. - * @param async if true, use locally FIFO scheduling - */ - void setAsyncMode(boolean async) { - locallyFifo = async; - } - - // Runstate management - - // Runstate values. Order 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; - - final boolean isShutdown() { return runState >= SHUTDOWN; } - final boolean isTerminating() { return runState >= TERMINATING; } - final boolean isTerminated() { return runState == TERMINATED; } - final boolean shutdown() { return transitionRunStateTo(SHUTDOWN); } - final boolean shutdownNow() { return transitionRunStateTo(TERMINATING); } - - /** - * Transition to at least the given state. Return true if not - * already at least given state. - */ - private boolean transitionRunStateTo(int state) { - for (;;) { - int s = runState; - if (s >= state) - return false; - if (_unsafe.compareAndSwapInt(this, runStateOffset, s, state)) - return true; - } - } - - /** - * Try to set status to active; fail on contention - */ - private boolean tryActivate() { - if (!active) { - if (!pool.tryIncrementActiveCount()) - return false; - active = true; - } - return true; - } + // Randomization /** - * Try to set status to active; fail on contention + * Computes next value for random victim probes and backoffs. + * Scans don't require a very high quality generator, but also not + * a crummy one. Marsaglia xor-shift is cheap and works well + * enough. Note: This is manually inlined in FJP.scan() to avoid + * writes inside busy loops. */ - private boolean tryInactivate() { - if (active) { - if (!pool.tryDecrementActiveCount()) - return false; - active = false; - } - return true; + private int nextSeed() { + int r = seed; + r ^= r << 13; + r ^= r >>> 17; + r ^= r << 5; + return seed = r; } - /** - * Computes next value for random victim probe. Scans don't - * require a very high quality generator, but also not a crummy - * one. Marsaglia xor-shift is cheap and works well. - */ - private static int xorShift(int r) { - r ^= r << 1; - r ^= r >>> 3; - r ^= r << 10; - return r; - } - - // Lifecycle methods - - /** - * This method is required to be public, but should never be - * called explicitly. It performs the main run loop to execute - * ForkJoinTasks. - */ - public void run() { - Throwable exception = null; - try { - onStart(); - pool.sync(this); // await first pool event - mainLoop(); - } catch (Throwable ex) { - exception = ex; - } finally { - onTermination(exception); - } - } - - /** - * Execute tasks until shut down. - */ - private void mainLoop() { - while (!isShutdown()) { - ForkJoinTask t = pollTask(); - if (t != null || (t = pollSubmission()) != null) - t.quietlyExec(); - else if (tryInactivate()) - pool.sync(this); - } - } + // Run State management /** * Initializes internal state after construction but before * processing any tasks. If you override this method, you must - * invoke super.onStart() at the beginning of the method. + * invoke {@code super.onStart()} at the beginning of the method. * Initialization requires care: Most fields must have legal * default values, to ensure that attempted accesses from other * threads work correctly even before this thread starts * processing tasks. */ protected void onStart() { - // Allocate while starting to improve chances of thread-local - // isolation queue = new ForkJoinTask[INITIAL_QUEUE_CAPACITY]; - // Initial value of seed need not be especially random but - // should differ across workers and must be nonzero - int p = poolIndex + 1; - seed = p + (p << 8) + (p << 16) + (p << 24); // spread bits + int r = pool.workerSeedGenerator.nextInt(); + seed = (r == 0) ? 1 : r; // must be nonzero } /** - * Perform cleanup associated with termination of this worker + * Performs cleanup associated with termination of this worker * thread. If you override this method, you must invoke - * super.onTermination at the end of the overridden method. + * {@code super.onTermination} at the end of the overridden method. * * @param exception the exception causing this thread to abort due - * to an unrecoverable error, or null if completed normally. + * to an unrecoverable error, or {@code null} if completed normally */ protected void onTermination(Throwable exception) { - // Execute remaining local tasks unless aborting or terminating - while (exception == null && !pool.isTerminating() && base != sp) { - try { - ForkJoinTask t = popTask(); - if (t != null) - t.quietlyExec(); - } catch(Throwable ex) { - exception = ex; - } - } - // Cancel other tasks, transition status, notify pool, and - // propagate exception to uncaught exception handler try { - do;while (!tryInactivate()); // ensure inactive + terminate = true; cancelTasks(); - runState = TERMINATED; - pool.workerTerminated(this); + pool.deregisterWorker(this, exception); } catch (Throwable ex) { // Shouldn't ever happen if (exception == null) // but if so, at least rethrown exception = ex; } finally { if (exception != null) - ForkJoinTask.rethrowException(exception); + UNSAFE.throwException(exception); } } - // Intrinsics-based support for queue operations. - /** - * Add in store-order the given task at given slot of q to - * null. Caller must ensure q is nonnull and index is in range. + * This method is required to be public, but should never be + * called explicitly. It performs the main run loop to execute + * {@link ForkJoinTask}s. */ - private static void setSlot(ForkJoinTask[] q, int i, - ForkJoinTask t){ - _unsafe.putOrderedObject(q, (i << qShift) + qBase, t); + public void run() { + Throwable exception = null; + try { + onStart(); + pool.work(this); + } catch (Throwable ex) { + exception = ex; + } finally { + onTermination(exception); + } } + /* + * Intrinsics-based atomic writes for queue slots. These are + * basically the same as methods in AtomicReferenceArray, but + * specialized for (1) ForkJoinTask elements (2) requirement that + * nullness and bounds checks have already been performed by + * callers and (3) effective offsets are known not to overflow + * from int to long (because of MAXIMUM_QUEUE_CAPACITY). We don't + * need corresponding version for reads: plain array reads are OK + * because they are protected by other volatile reads and are + * confirmed by CASes. + * + * Most uses don't actually call these methods, but instead + * contain inlined forms that enable more predictable + * optimization. We don't define the version of write used in + * pushTask at all, but instead inline there a store-fenced array + * slot write. + * + * Also in most methods, as a performance (not correctness) issue, + * we'd like to encourage compilers not to arbitrarily postpone + * setting queueTop after writing slot. Currently there is no + * intrinsic for arranging this, but using Unsafe putOrderedInt + * may be a preferable strategy on some compilers even though its + * main effect is a pre-, not post- fence. To simplify possible + * changes, the option is left in comments next to the associated + * assignments. + */ + /** - * CAS given slot of q to null. Caller must ensure q is nonnull - * and index is in range. + * CASes slot i of array q from t to null. Caller must ensure q is + * non-null and index is in range. */ - private static boolean casSlotNull(ForkJoinTask[] q, int i, - ForkJoinTask t) { - return _unsafe.compareAndSwapObject(q, (i << qShift) + qBase, t, null); + private static final boolean casSlotNull(ForkJoinTask[] q, int i, + ForkJoinTask t) { + return UNSAFE.compareAndSwapObject(q, (i << ASHIFT) + ABASE, t, null); } /** - * Sets sp in store-order. + * Performs a volatile write of the given task at given slot of + * array q. Caller must ensure q is non-null and index is in + * range. This method is used only during resets and backouts. */ - private void storeSp(int s) { - _unsafe.putOrderedInt(this, spOffset, s); + private static final void writeSlot(ForkJoinTask[] q, int i, + ForkJoinTask t) { + UNSAFE.putObjectVolatile(q, (i << ASHIFT) + ABASE, t); } - // Main queue methods + // queue methods /** - * Pushes a task. Called only by current thread. - * @param t the task. Caller must ensure nonnull + * Pushes a task. Call only from this thread. + * + * @param t the task. Caller must ensure non-null. */ final void pushTask(ForkJoinTask t) { - ForkJoinTask[] q = queue; - int mask = q.length - 1; - int s = sp; - setSlot(q, s & mask, t); - storeSp(++s); - if ((s -= base) == 1) - pool.signalWork(); - else if (s >= mask) - growQueue(); + ForkJoinTask[] q; int s, m; + if ((q = queue) != null) { // ignore if queue removed + long u = (((s = queueTop) & (m = q.length - 1)) << ASHIFT) + ABASE; + UNSAFE.putOrderedObject(q, u, t); + queueTop = s + 1; // or use putOrderedInt + if ((s -= queueBase) <= 2) + pool.signalWork(); + else if (s == m) + growQueue(); + } + } + + /** + * Creates or doubles queue array. Transfers elements by + * emulating steals (deqs) from old array and placing, oldest + * first, into new array. + */ + private void growQueue() { + ForkJoinTask[] oldQ = queue; + int size = oldQ != null ? oldQ.length << 1 : INITIAL_QUEUE_CAPACITY; + if (size > MAXIMUM_QUEUE_CAPACITY) + throw new RejectedExecutionException("Queue capacity exceeded"); + if (size < INITIAL_QUEUE_CAPACITY) + size = INITIAL_QUEUE_CAPACITY; + ForkJoinTask[] q = queue = new ForkJoinTask[size]; + int mask = size - 1; + int top = queueTop; + int oldMask; + if (oldQ != null && (oldMask = oldQ.length - 1) >= 0) { + for (int b = queueBase; b != top; ++b) { + long u = ((b & oldMask) << ASHIFT) + ABASE; + Object x = UNSAFE.getObjectVolatile(oldQ, u); + if (x != null && UNSAFE.compareAndSwapObject(oldQ, u, x, null)) + UNSAFE.putObjectVolatile + (q, ((b & mask) << ASHIFT) + ABASE, x); + } + } } /** * Tries to take a task from the base of the queue, failing if - * either empty or contended. - * @return a task, or null if none or contended. + * empty or contended. Note: Specializations of this code appear + * in locallyDeqTask and elsewhere. + * + * @return a task, or null if none or contended */ final ForkJoinTask deqTask() { - ForkJoinTask t; - ForkJoinTask[] q; - int i; - int b; - if (sp != (b = base) && + ForkJoinTask t; ForkJoinTask[] q; int b, i; + if (queueTop != (b = queueBase) && (q = queue) != null && // must read q after b - (t = q[i = (q.length - 1) & b]) != null && - casSlotNull(q, i, t)) { - base = b + 1; + (i = (q.length - 1) & b) >= 0 && + (t = q[i]) != null && queueBase == b && + UNSAFE.compareAndSwapObject(q, (i << ASHIFT) + ABASE, t, null)) { + queueBase = b + 1; return t; } return null; } /** - * Returns a popped task, or null if empty. Ensures active status - * if nonnull. Called only by current thread. + * Tries to take a task from the base of own queue. Called only + * by this thread. + * + * @return a task, or null if none + */ + final ForkJoinTask locallyDeqTask() { + ForkJoinTask t; int m, b, i; + ForkJoinTask[] q = queue; + if (q != null && (m = q.length - 1) >= 0) { + while (queueTop != (b = queueBase)) { + if ((t = q[i = m & b]) != null && + queueBase == b && + UNSAFE.compareAndSwapObject(q, (i << ASHIFT) + ABASE, + t, null)) { + queueBase = b + 1; + return t; + } + } + } + return null; + } + + /** + * Returns a popped task, or null if empty. + * Called only by this thread. */ - final ForkJoinTask popTask() { - int s = sp; - while (s != base) { - if (tryActivate()) { - ForkJoinTask[] q = queue; - int mask = q.length - 1; - int i = (s - 1) & mask; + private ForkJoinTask popTask() { + int m; + ForkJoinTask[] q = queue; + if (q != null && (m = q.length - 1) >= 0) { + for (int s; (s = queueTop) != queueBase;) { + int i = m & --s; + long u = (i << ASHIFT) + ABASE; // raw offset ForkJoinTask t = q[i]; - if (t == null || !casSlotNull(q, i, t)) + if (t == null) // lost to stealer break; - storeSp(s - 1); - return t; + if (UNSAFE.compareAndSwapObject(q, u, t, null)) { + queueTop = s; // or putOrderedInt + return t; + } } } return null; } /** - * Specialized version of popTask to pop only if - * topmost element is the given task. Called only - * by current thread while active. - * @param t the task. Caller must ensure nonnull + * Specialized version of popTask to pop only if topmost element + * is the given task. Called only by this thread. + * + * @param t the task. Caller must ensure non-null. */ final boolean unpushTask(ForkJoinTask t) { - ForkJoinTask[] q = queue; - int mask = q.length - 1; - int s = sp - 1; - if (casSlotNull(q, s & mask, t)) { - storeSp(s); + ForkJoinTask[] q; + int s; + if ((q = queue) != null && (s = queueTop) != queueBase && + UNSAFE.compareAndSwapObject + (q, (((q.length - 1) & --s) << ASHIFT) + ABASE, t, null)) { + queueTop = s; // or putOrderedInt return true; } return false; } /** - * Returns next task. + * Returns next task, or null if empty or contended. */ final ForkJoinTask peekTask() { + int m; ForkJoinTask[] q = queue; - if (q == null) + if (q == null || (m = q.length - 1) < 0) return null; - int mask = q.length - 1; - int i = locallyFifo? base : (sp - 1); - return q[i & mask]; + int i = locallyFifo ? queueBase : (queueTop - 1); + return q[i & m]; } + // Support methods for ForkJoinPool + /** - * Doubles queue array size. Transfers elements by emulating - * steals (deqs) from old array and placing, oldest first, into - * new array. + * Runs the given task, plus any local tasks until queue is empty */ - private void growQueue() { - ForkJoinTask[] oldQ = queue; - int oldSize = oldQ.length; - int newSize = oldSize << 1; - if (newSize > MAXIMUM_QUEUE_CAPACITY) - throw new RejectedExecutionException("Queue capacity exceeded"); - ForkJoinTask[] newQ = queue = new ForkJoinTask[newSize]; - - int b = base; - int bf = b + oldSize; - int oldMask = oldSize - 1; - int newMask = newSize - 1; - do { - int oldIndex = b & oldMask; - ForkJoinTask t = oldQ[oldIndex]; - if (t != null && !casSlotNull(oldQ, oldIndex, t)) - t = null; - setSlot(newQ, b & newMask, t); - } while (++b != bf); - pool.signalWork(); + final void execTask(ForkJoinTask t) { + currentSteal = t; + for (;;) { + if (t != null) + t.doExec(); + if (queueTop == queueBase) + break; + t = locallyFifo ? locallyDeqTask() : popTask(); + } + ++stealCount; + currentSteal = null; } /** - * Tries to steal a task from another worker. Starts at a random - * index of workers array, and probes workers until finding one - * with non-empty queue or finding that all are empty. It - * randomly selects the first n probes. If these are empty, it - * resorts to a full circular traversal, which is necessary to - * accurately set active status by caller. Also restarts if pool - * events occurred since last scan, which forces refresh of - * workers array, in case barrier was associated with resize. - * - * This method must be both fast and quiet -- usually avoiding - * memory accesses that could disrupt cache sharing etc other than - * those needed to check for and take tasks. This accounts for, - * among other things, updating random seed in place without - * storing it until exit. + * Removes and cancels all tasks in queue. Can be called from any + * thread. + */ + final void cancelTasks() { + ForkJoinTask cj = currentJoin; // try to cancel ongoing tasks + if (cj != null && cj.status >= 0) + cj.cancelIgnoringExceptions(); + ForkJoinTask cs = currentSteal; + if (cs != null && cs.status >= 0) + cs.cancelIgnoringExceptions(); + while (queueBase != queueTop) { + ForkJoinTask t = deqTask(); + if (t != null) + t.cancelIgnoringExceptions(); + } + } + + /** + * Drains tasks to given collection c. * - * @return a task, or null if none found - */ - private ForkJoinTask scan() { - ForkJoinTask t = null; - int r = seed; // extract once to keep scan quiet - ForkJoinWorkerThread[] ws; // refreshed on outer loop - int mask; // must be power 2 minus 1 and > 0 - outer:do { - if ((ws = pool.workers) != null && (mask = ws.length - 1) > 0) { - int idx = r; - int probes = ~mask; // use random index while negative - for (;;) { - r = xorShift(r); // update random seed - ForkJoinWorkerThread v = ws[mask & idx]; - if (v == null || v.sp == v.base) { - if (probes <= mask) - idx = (probes++ < 0)? r : (idx + 1); - else - break; - } - else if (!tryActivate() || (t = v.deqTask()) == null) - continue outer; // restart on contention - else - break outer; - } + * @return the number of tasks drained + */ + final int drainTasksTo(Collection> c) { + int n = 0; + while (queueBase != queueTop) { + ForkJoinTask t = deqTask(); + if (t != null) { + c.add(t); + ++n; } - } while (pool.hasNewSyncEvent(this)); // retry on pool events - seed = r; - return t; + } + return n; } + // Support methods for ForkJoinTask + /** - * gets and removes a local or stolen a task - * @return a task, if available + * Returns an estimate of the number of tasks in the queue. */ - final ForkJoinTask pollTask() { - ForkJoinTask t = locallyFifo? deqTask() : popTask(); - if (t == null && (t = scan()) != null) - ++stealCount; - return t; + final int getQueueSize() { + return queueTop - queueBase; } /** - * gets a local task + * Gets and removes a local task. + * * @return a task, if available */ final ForkJoinTask pollLocalTask() { - return locallyFifo? deqTask() : popTask(); + return locallyFifo ? locallyDeqTask() : popTask(); } /** - * Returns a pool submission, if one exists, activating first. - * @return a submission, if available + * Gets and removes a local or stolen task. + * + * @return a task, if available */ - private ForkJoinTask pollSubmission() { - ForkJoinPool p = pool; - while (p.hasQueuedSubmissions()) { - ForkJoinTask t; - if (tryActivate() && (t = p.pollSubmission()) != null) - return t; + final ForkJoinTask pollTask() { + ForkJoinWorkerThread[] ws; + ForkJoinTask t = pollLocalTask(); + if (t != null || (ws = pool.workers) == null) + return t; + int n = ws.length; // cheap version of FJP.scan + int steps = n << 1; + int r = nextSeed(); + int i = 0; + while (i < steps) { + ForkJoinWorkerThread w = ws[(i++ + r) & (n - 1)]; + if (w != null && w.queueBase != w.queueTop && w.queue != null) { + if ((t = w.deqTask()) != null) + return t; + i = 0; + } } return null; } - // Methods accessed only by Pool - /** - * Removes and cancels all tasks in queue. Can be called from any - * thread. + * The maximum stolen->joining link depth allowed in helpJoinTask, + * as well as the maximum number of retries (allowing on average + * one staleness retry per level) per attempt to instead try + * compensation. Depths for legitimate chains are unbounded, but + * we use a fixed constant to avoid (otherwise unchecked) cycles + * and bound staleness of traversal parameters at the expense of + * sometimes blocking when we could be helping. */ - final void cancelTasks() { - ForkJoinTask t; - while (base != sp && (t = deqTask()) != null) - t.cancelIgnoringExceptions(); - } + private static final int MAX_HELP = 16; /** - * Drains tasks to given collection c - * @return the number of tasks drained - */ - final int drainTasksTo(Collection> c) { - int n = 0; - ForkJoinTask t; - while (base != sp && (t = deqTask()) != null) { - c.add(t); - ++n; + * Possibly runs some tasks and/or blocks, until joinMe is done. + * + * @param joinMe the task to join + * @return completion status on exit + */ + final int joinTask(ForkJoinTask joinMe) { + ForkJoinTask prevJoin = currentJoin; + currentJoin = joinMe; + for (int s, retries = MAX_HELP;;) { + if ((s = joinMe.status) < 0) { + currentJoin = prevJoin; + return s; + } + if (retries > 0) { + if (queueTop != queueBase) { + if (!localHelpJoinTask(joinMe)) + retries = 0; // cannot help + } + else if (retries == MAX_HELP >>> 1) { + --retries; // check uncommon case + if (tryDeqAndExec(joinMe) >= 0) + Thread.yield(); // for politeness + } + else + retries = helpJoinTask(joinMe) ? MAX_HELP : retries - 1; + } + else { + retries = MAX_HELP; // restart if not done + pool.tryAwaitJoin(joinMe); + } } - return n; } /** - * Get and clear steal count for accumulation by pool. Called - * only when known to be idle (in pool.sync and termination). - */ - final int getAndClearStealCount() { - int sc = stealCount; - stealCount = 0; - return sc; + * If present, pops and executes the given task, or any other + * cancelled task + * + * @return false if any other non-cancelled task exists in local queue + */ + private boolean localHelpJoinTask(ForkJoinTask joinMe) { + int s, i; ForkJoinTask[] q; ForkJoinTask t; + if ((s = queueTop) != queueBase && (q = queue) != null && + (i = (q.length - 1) & --s) >= 0 && + (t = q[i]) != null) { + if (t != joinMe && t.status >= 0) + return false; + if (UNSAFE.compareAndSwapObject + (q, (i << ASHIFT) + ABASE, t, null)) { + queueTop = s; // or putOrderedInt + t.doExec(); + } + } + return true; } /** - * Returns true if at least one worker in the given array appears - * to have at least one queued task. - * @param ws array of workers - */ - static boolean hasQueuedTasks(ForkJoinWorkerThread[] ws) { - if (ws != null) { - int len = ws.length; - for (int j = 0; j < 2; ++j) { // need two passes for clean sweep - for (int i = 0; i < len; ++i) { - ForkJoinWorkerThread w = ws[i]; - if (w != null && w.sp != w.base) - return true; + * Tries to locate and execute tasks for a stealer of the given + * task, or in turn one of its stealers, Traces + * currentSteal->currentJoin links looking for a thread working on + * a descendant of the given task and with a non-empty queue to + * steal back and execute tasks from. The implementation is very + * branchy to cope with potential inconsistencies or loops + * encountering chains that are stale, unknown, or of length + * greater than MAX_HELP links. All of these cases are dealt with + * by just retrying by caller. + * + * @param joinMe the task to join + * @param canSteal true if local queue is empty + * @return true if ran a task + */ + private boolean helpJoinTask(ForkJoinTask joinMe) { + boolean helped = false; + int m = pool.scanGuard & SMASK; + ForkJoinWorkerThread[] ws = pool.workers; + if (ws != null && ws.length > m && joinMe.status >= 0) { + int levels = MAX_HELP; // remaining chain length + ForkJoinTask task = joinMe; // base of chain + outer:for (ForkJoinWorkerThread thread = this;;) { + // Try to find v, the stealer of task, by first using hint + ForkJoinWorkerThread v = ws[thread.stealHint & m]; + if (v == null || v.currentSteal != task) { + for (int j = 0; ;) { // search array + if ((v = ws[j]) != null && v.currentSteal == task) { + thread.stealHint = j; + break; // save hint for next time + } + if (++j > m) + break outer; // can't find stealer + } + } + // Try to help v, using specialized form of deqTask + for (;;) { + ForkJoinTask[] q; int b, i; + if (joinMe.status < 0) + break outer; + if ((b = v.queueBase) == v.queueTop || + (q = v.queue) == null || + (i = (q.length-1) & b) < 0) + break; // empty + long u = (i << ASHIFT) + ABASE; + ForkJoinTask t = q[i]; + if (task.status < 0) + break outer; // stale + if (t != null && v.queueBase == b && + UNSAFE.compareAndSwapObject(q, u, t, null)) { + v.queueBase = b + 1; + v.stealHint = poolIndex; + ForkJoinTask ps = currentSteal; + currentSteal = t; + t.doExec(); + currentSteal = ps; + helped = true; + } + } + // Try to descend to find v's stealer + ForkJoinTask next = v.currentJoin; + if (--levels > 0 && task.status >= 0 && + next != null && next != task) { + task = next; + thread = v; } + else + break; // max levels, stale, dead-end, or cyclic } } - return false; + return helped; } - // Support methods for ForkJoinTask - /** - * Returns an estimate of the number of tasks in the queue. - */ - final int getQueueSize() { - int n = sp - base; - return n < 0? 0 : n; // suppress momentarily negative values + * Performs an uncommon case for joinTask: If task t is at base of + * some workers queue, steals and executes it. + * + * @param t the task + * @return t's status + */ + private int tryDeqAndExec(ForkJoinTask t) { + int m = pool.scanGuard & SMASK; + ForkJoinWorkerThread[] ws = pool.workers; + if (ws != null && ws.length > m && t.status >= 0) { + for (int j = 0; j <= m; ++j) { + ForkJoinTask[] q; int b, i; + ForkJoinWorkerThread v = ws[j]; + if (v != null && + (b = v.queueBase) != v.queueTop && + (q = v.queue) != null && + (i = (q.length - 1) & b) >= 0 && + q[i] == t) { + long u = (i << ASHIFT) + ABASE; + if (v.queueBase == b && + UNSAFE.compareAndSwapObject(q, u, t, null)) { + v.queueBase = b + 1; + v.stealHint = poolIndex; + ForkJoinTask ps = currentSteal; + currentSteal = t; + t.doExec(); + currentSteal = ps; + } + break; + } + } + } + return t.status; } /** - * Returns an estimate of the number of tasks, offset by a - * function of number of idle workers. + * Implements ForkJoinTask.getSurplusQueuedTaskCount(). Returns + * an estimate of the number of tasks, offset by a function of + * number of idle workers. + * + * This method provides a cheap heuristic guide for task + * partitioning when programmers, frameworks, tools, or languages + * have little or no idea about task granularity. In essence by + * offering this method, we ask users only about tradeoffs in + * overhead vs expected throughput and its variance, rather than + * how finely to partition tasks. + * + * In a steady state strict (tree-structured) computation, each + * thread makes available for stealing enough tasks for other + * threads to remain active. Inductively, if all threads play by + * the same rules, each thread should make available only a + * constant number of tasks. + * + * The minimum useful constant is just 1. But using a value of 1 + * would require immediate replenishment upon each steal to + * maintain enough tasks, which is infeasible. Further, + * partitionings/granularities of offered tasks should minimize + * steal rates, which in general means that threads nearer the top + * of computation tree should generate more than those nearer the + * bottom. In perfect steady state, each thread is at + * approximately the same level of computation tree. However, + * producing extra tasks amortizes the uncertainty of progress and + * diffusion assumptions. + * + * So, users will want to use values larger, but not much larger + * than 1 to both smooth over transient shortages and hedge + * against uneven progress; as traded off against the cost of + * extra task overhead. We leave the user to pick a threshold + * value to compare with the results of this call to guide + * decisions, but recommend values such as 3. + * + * When all threads are active, it is on average OK to estimate + * surplus strictly locally. In steady-state, if one thread is + * maintaining say 2 surplus tasks, then so are others. So we can + * just use estimated queue length (although note that (queueTop - + * queueBase) can be an overestimate because of stealers lagging + * increments of queueBase). However, this strategy alone leads + * to serious mis-estimates in some non-steady-state conditions + * (ramp-up, ramp-down, other stalls). We can detect many of these + * by further considering the number of "idle" threads, that are + * known to have zero queued tasks, so compensate by a factor of + * (#idle/#active) threads. */ final int getEstimatedSurplusTaskCount() { - // The halving approximates weighting idle vs non-idle workers - return (sp - base) - (pool.getIdleThreadCount() >>> 1); + return queueTop - queueBase - pool.idlePerActive(); } /** - * Scan, returning early if joinMe done - */ - final ForkJoinTask scanWhileJoining(ForkJoinTask joinMe) { - ForkJoinTask t = pollTask(); - if (t != null && joinMe.status < 0 && sp == base) { - pushTask(t); // unsteal if done and this task would be stealable - t = null; - } - return t; - } - - /** - * Runs tasks until pool isQuiescent + * Runs tasks until {@code pool.isQuiescent()}. We piggyback on + * pool's active count ctl maintenance, but rather than blocking + * when tasks cannot be found, we rescan until all others cannot + * find tasks either. The bracketing by pool quiescerCounts + * updates suppresses pool auto-shutdown mechanics that could + * otherwise prematurely terminate the pool because all threads + * appear to be inactive. */ final void helpQuiescePool() { + boolean active = true; + ForkJoinTask ps = currentSteal; // to restore below + ForkJoinPool p = pool; + p.addQuiescerCount(1); for (;;) { - ForkJoinTask t = pollTask(); - if (t != null) - t.quietlyExec(); - else if (tryInactivate() && pool.isQuiescent()) - break; - } - do;while (!tryActivate()); // re-activate on exit - } - - // Temporary Unsafe mechanics for preliminary release - private static Unsafe getUnsafe() throws Throwable { - try { - return Unsafe.getUnsafe(); - } catch (SecurityException se) { - try { - return java.security.AccessController.doPrivileged - (new java.security.PrivilegedExceptionAction() { - public Unsafe run() throws Exception { - return getUnsafePrivileged(); - }}); - } catch (java.security.PrivilegedActionException e) { - throw e.getCause(); + ForkJoinWorkerThread[] ws = p.workers; + ForkJoinWorkerThread v = null; + int n; + if (queueTop != queueBase) + v = this; + else if (ws != null && (n = ws.length) > 1) { + ForkJoinWorkerThread w; + int r = nextSeed(); // cheap version of FJP.scan + int steps = n << 1; + for (int i = 0; i < steps; ++i) { + if ((w = ws[(i + r) & (n - 1)]) != null && + w.queueBase != w.queueTop) { + v = w; + break; + } + } + } + if (v != null) { + ForkJoinTask t; + if (!active) { + active = true; + p.addActiveCount(1); + } + if ((t = (v != this) ? v.deqTask() : + locallyFifo ? locallyDeqTask() : popTask()) != null) { + currentSteal = t; + t.doExec(); + currentSteal = ps; + } + } + else { + if (active) { + active = false; + p.addActiveCount(-1); + } + if (p.isQuiescent()) { + p.addActiveCount(1); + p.addQuiescerCount(-1); + break; + } } } } - private static Unsafe getUnsafePrivileged() - throws NoSuchFieldException, IllegalAccessException { - Field f = Unsafe.class.getDeclaredField("theUnsafe"); - f.setAccessible(true); - return (Unsafe) f.get(null); - } + // Unsafe mechanics + private static final sun.misc.Unsafe UNSAFE; + private static final long ABASE; + private static final int ASHIFT; - private static long fieldOffset(String fieldName) - throws NoSuchFieldException { - return _unsafe.objectFieldOffset - (ForkJoinWorkerThread.class.getDeclaredField(fieldName)); - } - - static final Unsafe _unsafe; - static final long baseOffset; - static final long spOffset; - static final long runStateOffset; - static final long qBase; - static final int qShift; static { + int s; try { - _unsafe = getUnsafe(); - baseOffset = fieldOffset("base"); - spOffset = fieldOffset("sp"); - runStateOffset = fieldOffset("runState"); - qBase = _unsafe.arrayBaseOffset(ForkJoinTask[].class); - int s = _unsafe.arrayIndexScale(ForkJoinTask[].class); - if ((s & (s-1)) != 0) - throw new Error("data type scale not a power of two"); - qShift = 31 - Integer.numberOfLeadingZeros(s); - } catch (Throwable e) { - throw new RuntimeException("Could not initialize intrinsics", e); + UNSAFE = sun.misc.Unsafe.getUnsafe(); + Class a = ForkJoinTask[].class; + ABASE = UNSAFE.arrayBaseOffset(a); + s = UNSAFE.arrayIndexScale(a); + } catch (Exception e) { + throw new Error(e); } + if ((s & (s-1)) != 0) + throw new Error("data type scale not a power of two"); + ASHIFT = 31 - Integer.numberOfLeadingZeros(s); } + } diff --git a/src/forkjoin/scala/concurrent/forkjoin/LinkedTransferQueue.java b/src/forkjoin/scala/concurrent/forkjoin/LinkedTransferQueue.java index 3b46c176ff..2d9ca99737 100644 --- a/src/forkjoin/scala/concurrent/forkjoin/LinkedTransferQueue.java +++ b/src/forkjoin/scala/concurrent/forkjoin/LinkedTransferQueue.java @@ -1,30 +1,38 @@ /* * Written by Doug Lea with assistance from members of JCP JSR-166 * Expert Group and released to the public domain, as explained at - * http://creativecommons.org/licenses/publicdomain + * http://creativecommons.org/publicdomain/zero/1.0/ */ package scala.concurrent.forkjoin; -import java.util.concurrent.*; -import java.util.concurrent.locks.*; -import java.util.concurrent.atomic.*; -import java.util.*; -import java.io.*; -import sun.misc.Unsafe; -import java.lang.reflect.*; + +import java.util.AbstractQueue; +import java.util.Collection; +import java.util.Iterator; +import java.util.NoSuchElementException; +import java.util.Queue; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.locks.LockSupport; /** - * An unbounded {@linkplain TransferQueue} based on linked nodes. + * An unbounded {@link TransferQueue} based on linked nodes. * This queue orders elements FIFO (first-in-first-out) with respect * to any given producer. The head of the queue is that * element that has been on the queue the longest time for some * producer. The tail of the queue is that element that has * been on the queue the shortest time for some producer. * - *

Beware that, unlike in most collections, the {@code size} - * method is NOT a constant-time operation. Because of the + *

Beware that, unlike in most collections, the {@code size} method + * is NOT a constant-time operation. Because of the * asynchronous nature of these queues, determining the current number - * of elements requires a traversal of the elements. + * of elements requires a traversal of the elements, and so may report + * inaccurate results if this collection is modified during traversal. + * Additionally, the bulk operations {@code addAll}, + * {@code removeAll}, {@code retainAll}, {@code containsAll}, + * {@code equals}, and {@code toArray} are not guaranteed + * to be performed atomically. For example, an iterator operating + * concurrently with an {@code addAll} operation might view only some + * of the added elements. * *

This class and its iterator implement all of the * optional methods of the {@link Collection} and {@link @@ -44,381 +52,938 @@ import java.lang.reflect.*; * @since 1.7 * @author Doug Lea * @param the type of elements held in this collection - * */ public class LinkedTransferQueue extends AbstractQueue implements TransferQueue, java.io.Serializable { private static final long serialVersionUID = -3223113410248163686L; /* - * This class extends the approach used in FIFO-mode - * SynchronousQueues. See the internal documentation, as well as - * the PPoPP 2006 paper "Scalable Synchronous Queues" by Scherer, - * Lea & Scott - * (http://www.cs.rice.edu/~wns1/papers/2006-PPoPP-SQ.pdf) + * *** Overview of Dual Queues with Slack *** + * + * Dual Queues, introduced by Scherer and Scott + * (http://www.cs.rice.edu/~wns1/papers/2004-DISC-DDS.pdf) are + * (linked) queues in which nodes may represent either data or + * requests. When a thread tries to enqueue a data node, but + * encounters a request node, it instead "matches" and removes it; + * and vice versa for enqueuing requests. Blocking Dual Queues + * arrange that threads enqueuing unmatched requests block until + * other threads provide the match. Dual Synchronous Queues (see + * Scherer, Lea, & Scott + * http://www.cs.rochester.edu/u/scott/papers/2009_Scherer_CACM_SSQ.pdf) + * additionally arrange that threads enqueuing unmatched data also + * block. Dual Transfer Queues support all of these modes, as + * dictated by callers. + * + * A FIFO dual queue may be implemented using a variation of the + * Michael & Scott (M&S) lock-free queue algorithm + * (http://www.cs.rochester.edu/u/scott/papers/1996_PODC_queues.pdf). + * It maintains two pointer fields, "head", pointing to a + * (matched) node that in turn points to the first actual + * (unmatched) queue node (or null if empty); and "tail" that + * points to the last node on the queue (or again null if + * empty). For example, here is a possible queue with four data + * elements: + * + * head tail + * | | + * v v + * M -> U -> U -> U -> U + * + * The M&S queue algorithm is known to be prone to scalability and + * overhead limitations when maintaining (via CAS) these head and + * tail pointers. This has led to the development of + * contention-reducing variants such as elimination arrays (see + * Moir et al http://portal.acm.org/citation.cfm?id=1074013) and + * optimistic back pointers (see Ladan-Mozes & Shavit + * http://people.csail.mit.edu/edya/publications/OptimisticFIFOQueue-journal.pdf). + * However, the nature of dual queues enables a simpler tactic for + * improving M&S-style implementations when dual-ness is needed. + * + * In a dual queue, each node must atomically maintain its match + * status. While there are other possible variants, we implement + * this here as: for a data-mode node, matching entails CASing an + * "item" field from a non-null data value to null upon match, and + * vice-versa for request nodes, CASing from null to a data + * value. (Note that the linearization properties of this style of + * queue are easy to verify -- elements are made available by + * linking, and unavailable by matching.) Compared to plain M&S + * queues, this property of dual queues requires one additional + * successful atomic operation per enq/deq pair. But it also + * enables lower cost variants of queue maintenance mechanics. (A + * variation of this idea applies even for non-dual queues that + * support deletion of interior elements, such as + * j.u.c.ConcurrentLinkedQueue.) + * + * Once a node is matched, its match status can never again + * change. We may thus arrange that the linked list of them + * contain a prefix of zero or more matched nodes, followed by a + * suffix of zero or more unmatched nodes. (Note that we allow + * both the prefix and suffix to be zero length, which in turn + * means that we do not use a dummy header.) If we were not + * concerned with either time or space efficiency, we could + * correctly perform enqueue and dequeue operations by traversing + * from a pointer to the initial node; CASing the item of the + * first unmatched node on match and CASing the next field of the + * trailing node on appends. (Plus some special-casing when + * initially empty). While this would be a terrible idea in + * itself, it does have the benefit of not requiring ANY atomic + * updates on head/tail fields. + * + * We introduce here an approach that lies between the extremes of + * never versus always updating queue (head and tail) pointers. + * This offers a tradeoff between sometimes requiring extra + * traversal steps to locate the first and/or last unmatched + * nodes, versus the reduced overhead and contention of fewer + * updates to queue pointers. For example, a possible snapshot of + * a queue is: + * + * head tail + * | | + * v v + * M -> M -> U -> U -> U -> U + * + * The best value for this "slack" (the targeted maximum distance + * between the value of "head" and the first unmatched node, and + * similarly for "tail") is an empirical matter. We have found + * that using very small constants in the range of 1-3 work best + * over a range of platforms. Larger values introduce increasing + * costs of cache misses and risks of long traversal chains, while + * smaller values increase CAS contention and overhead. + * + * Dual queues with slack differ from plain M&S dual queues by + * virtue of only sometimes updating head or tail pointers when + * matching, appending, or even traversing nodes; in order to + * maintain a targeted slack. The idea of "sometimes" may be + * operationalized in several ways. The simplest is to use a + * per-operation counter incremented on each traversal step, and + * to try (via CAS) to update the associated queue pointer + * whenever the count exceeds a threshold. Another, that requires + * more overhead, is to use random number generators to update + * with a given probability per traversal step. + * + * In any strategy along these lines, because CASes updating + * fields may fail, the actual slack may exceed targeted + * slack. However, they may be retried at any time to maintain + * targets. Even when using very small slack values, this + * approach works well for dual queues because it allows all + * operations up to the point of matching or appending an item + * (hence potentially allowing progress by another thread) to be + * read-only, thus not introducing any further contention. As + * described below, we implement this by performing slack + * maintenance retries only after these points. + * + * As an accompaniment to such techniques, traversal overhead can + * be further reduced without increasing contention of head + * pointer updates: Threads may sometimes shortcut the "next" link + * path from the current "head" node to be closer to the currently + * known first unmatched node, and similarly for tail. Again, this + * may be triggered with using thresholds or randomization. + * + * These ideas must be further extended to avoid unbounded amounts + * of costly-to-reclaim garbage caused by the sequential "next" + * links of nodes starting at old forgotten head nodes: As first + * described in detail by Boehm + * (http://portal.acm.org/citation.cfm?doid=503272.503282) if a GC + * delays noticing that any arbitrarily old node has become + * garbage, all newer dead nodes will also be unreclaimed. + * (Similar issues arise in non-GC environments.) To cope with + * this in our implementation, upon CASing to advance the head + * pointer, we set the "next" link of the previous head to point + * only to itself; thus limiting the length of connected dead lists. + * (We also take similar care to wipe out possibly garbage + * retaining values held in other Node fields.) However, doing so + * adds some further complexity to traversal: If any "next" + * pointer links to itself, it indicates that the current thread + * has lagged behind a head-update, and so the traversal must + * continue from the "head". Traversals trying to find the + * current tail starting from "tail" may also encounter + * self-links, in which case they also continue at "head". + * + * It is tempting in slack-based scheme to not even use CAS for + * updates (similarly to Ladan-Mozes & Shavit). However, this + * cannot be done for head updates under the above link-forgetting + * mechanics because an update may leave head at a detached node. + * And while direct writes are possible for tail updates, they + * increase the risk of long retraversals, and hence long garbage + * chains, which can be much more costly than is worthwhile + * considering that the cost difference of performing a CAS vs + * write is smaller when they are not triggered on each operation + * (especially considering that writes and CASes equally require + * additional GC bookkeeping ("write barriers") that are sometimes + * more costly than the writes themselves because of contention). + * + * *** Overview of implementation *** + * + * We use a threshold-based approach to updates, with a slack + * threshold of two -- that is, we update head/tail when the + * current pointer appears to be two or more steps away from the + * first/last node. The slack value is hard-wired: a path greater + * than one is naturally implemented by checking equality of + * traversal pointers except when the list has only one element, + * in which case we keep slack threshold at one. Avoiding tracking + * explicit counts across method calls slightly simplifies an + * already-messy implementation. Using randomization would + * probably work better if there were a low-quality dirt-cheap + * per-thread one available, but even ThreadLocalRandom is too + * heavy for these purposes. + * + * With such a small slack threshold value, it is not worthwhile + * to augment this with path short-circuiting (i.e., unsplicing + * interior nodes) except in the case of cancellation/removal (see + * below). + * + * We allow both the head and tail fields to be null before any + * nodes are enqueued; initializing upon first append. This + * simplifies some other logic, as well as providing more + * efficient explicit control paths instead of letting JVMs insert + * implicit NullPointerExceptions when they are null. While not + * currently fully implemented, we also leave open the possibility + * of re-nulling these fields when empty (which is complicated to + * arrange, for little benefit.) + * + * All enqueue/dequeue operations are handled by the single method + * "xfer" with parameters indicating whether to act as some form + * of offer, put, poll, take, or transfer (each possibly with + * timeout). The relative complexity of using one monolithic + * method outweighs the code bulk and maintenance problems of + * using separate methods for each case. * - * The main extension is to provide different Wait modes for the - * main "xfer" method that puts or takes items. These don't - * impact the basic dual-queue logic, but instead control whether - * or how threads block upon insertion of request or data nodes - * into the dual queue. It also uses slightly different - * conventions for tracking whether nodes are off-list or - * cancelled. + * Operation consists of up to three phases. The first is + * implemented within method xfer, the second in tryAppend, and + * the third in method awaitMatch. + * + * 1. Try to match an existing node + * + * Starting at head, skip already-matched nodes until finding + * an unmatched node of opposite mode, if one exists, in which + * case matching it and returning, also if necessary updating + * head to one past the matched node (or the node itself if the + * list has no other unmatched nodes). If the CAS misses, then + * a loop retries advancing head by two steps until either + * success or the slack is at most two. By requiring that each + * attempt advances head by two (if applicable), we ensure that + * the slack does not grow without bound. Traversals also check + * if the initial head is now off-list, in which case they + * start at the new head. + * + * If no candidates are found and the call was untimed + * poll/offer, (argument "how" is NOW) return. + * + * 2. Try to append a new node (method tryAppend) + * + * Starting at current tail pointer, find the actual last node + * and try to append a new node (or if head was null, establish + * the first node). Nodes can be appended only if their + * predecessors are either already matched or are of the same + * mode. If we detect otherwise, then a new node with opposite + * mode must have been appended during traversal, so we must + * restart at phase 1. The traversal and update steps are + * otherwise similar to phase 1: Retrying upon CAS misses and + * checking for staleness. In particular, if a self-link is + * encountered, then we can safely jump to a node on the list + * by continuing the traversal at current head. + * + * On successful append, if the call was ASYNC, return. + * + * 3. Await match or cancellation (method awaitMatch) + * + * Wait for another thread to match node; instead cancelling if + * the current thread was interrupted or the wait timed out. On + * multiprocessors, we use front-of-queue spinning: If a node + * appears to be the first unmatched node in the queue, it + * spins a bit before blocking. In either case, before blocking + * it tries to unsplice any nodes between the current "head" + * and the first unmatched node. + * + * Front-of-queue spinning vastly improves performance of + * heavily contended queues. And so long as it is relatively + * brief and "quiet", spinning does not much impact performance + * of less-contended queues. During spins threads check their + * interrupt status and generate a thread-local random number + * to decide to occasionally perform a Thread.yield. While + * yield has underdefined specs, we assume that might it help, + * and will not hurt in limiting impact of spinning on busy + * systems. We also use smaller (1/2) spins for nodes that are + * not known to be front but whose predecessors have not + * blocked -- these "chained" spins avoid artifacts of + * front-of-queue rules which otherwise lead to alternating + * nodes spinning vs blocking. Further, front threads that + * represent phase changes (from data to request node or vice + * versa) compared to their predecessors receive additional + * chained spins, reflecting longer paths typically required to + * unblock threads during phase changes. + * + * + * ** Unlinking removed interior nodes ** + * + * In addition to minimizing garbage retention via self-linking + * described above, we also unlink removed interior nodes. These + * may arise due to timed out or interrupted waits, or calls to + * remove(x) or Iterator.remove. Normally, given a node that was + * at one time known to be the predecessor of some node s that is + * to be removed, we can unsplice s by CASing the next field of + * its predecessor if it still points to s (otherwise s must + * already have been removed or is now offlist). But there are two + * situations in which we cannot guarantee to make node s + * unreachable in this way: (1) If s is the trailing node of list + * (i.e., with null next), then it is pinned as the target node + * for appends, so can only be removed later after other nodes are + * appended. (2) We cannot necessarily unlink s given a + * predecessor node that is matched (including the case of being + * cancelled): the predecessor may already be unspliced, in which + * case some previous reachable node may still point to s. + * (For further explanation see Herlihy & Shavit "The Art of + * Multiprocessor Programming" chapter 9). Although, in both + * cases, we can rule out the need for further action if either s + * or its predecessor are (or can be made to be) at, or fall off + * from, the head of list. + * + * Without taking these into account, it would be possible for an + * unbounded number of supposedly removed nodes to remain + * reachable. Situations leading to such buildup are uncommon but + * can occur in practice; for example when a series of short timed + * calls to poll repeatedly time out but never otherwise fall off + * the list because of an untimed call to take at the front of the + * queue. + * + * When these cases arise, rather than always retraversing the + * entire list to find an actual predecessor to unlink (which + * won't help for case (1) anyway), we record a conservative + * estimate of possible unsplice failures (in "sweepVotes"). + * We trigger a full sweep when the estimate exceeds a threshold + * ("SWEEP_THRESHOLD") indicating the maximum number of estimated + * removal failures to tolerate before sweeping through, unlinking + * cancelled nodes that were not unlinked upon initial removal. + * We perform sweeps by the thread hitting threshold (rather than + * background threads or by spreading work to other threads) + * because in the main contexts in which removal occurs, the + * caller is already timed-out, cancelled, or performing a + * potentially O(n) operation (e.g. remove(x)), none of which are + * time-critical enough to warrant the overhead that alternatives + * would impose on other threads. + * + * Because the sweepVotes estimate is conservative, and because + * nodes become unlinked "naturally" as they fall off the head of + * the queue, and because we allow votes to accumulate even while + * sweeps are in progress, there are typically significantly fewer + * such nodes than estimated. Choice of a threshold value + * balances the likelihood of wasted effort and contention, versus + * providing a worst-case bound on retention of interior nodes in + * quiescent queues. The value defined below was chosen + * empirically to balance these under various timeout scenarios. + * + * Note that we cannot self-link unlinked interior nodes during + * sweeps. However, the associated garbage chains terminate when + * some successor ultimately falls off the head of the list and is + * self-linked. */ - // Wait modes for xfer method - static final int NOWAIT = 0; - static final int TIMEOUT = 1; - static final int WAIT = 2; - - /** The number of CPUs, for spin control */ - static final int NCPUS = Runtime.getRuntime().availableProcessors(); + /** True if on multiprocessor */ + private static final boolean MP = + Runtime.getRuntime().availableProcessors() > 1; /** - * The number of times to spin before blocking in timed waits. - * The value is empirically derived -- it works well across a - * variety of processors and OSes. Empirically, the best value - * seems not to vary with number of CPUs (beyond 2) so is just - * a constant. + * The number of times to spin (with randomly interspersed calls + * to Thread.yield) on multiprocessor before blocking when a node + * is apparently the first waiter in the queue. See above for + * explanation. Must be a power of two. The value is empirically + * derived -- it works pretty well across a variety of processors, + * numbers of CPUs, and OSes. */ - static final int maxTimedSpins = (NCPUS < 2)? 0 : 32; + private static final int FRONT_SPINS = 1 << 7; /** - * The number of times to spin before blocking in untimed waits. - * This is greater than timed value because untimed waits spin - * faster since they don't need to check times on each spin. + * The number of times to spin before blocking when a node is + * preceded by another node that is apparently spinning. Also + * serves as an increment to FRONT_SPINS on phase changes, and as + * base average frequency for yielding during spins. Must be a + * power of two. */ - static final int maxUntimedSpins = maxTimedSpins * 16; + private static final int CHAINED_SPINS = FRONT_SPINS >>> 1; /** - * The number of nanoseconds for which it is faster to spin - * rather than to use timed park. A rough estimate suffices. + * The maximum number of estimated removal failures (sweepVotes) + * to tolerate before sweeping through the queue unlinking + * cancelled nodes that were not unlinked upon initial + * removal. See above for explanation. The value must be at least + * two to avoid useless sweeps when removing trailing nodes. */ - static final long spinForTimeoutThreshold = 1000L; + static final int SWEEP_THRESHOLD = 32; /** - * Node class for LinkedTransferQueue. Opportunistically - * subclasses from AtomicReference to represent item. Uses Object, - * not E, to allow setting item to "this" after use, to avoid - * garbage retention. Similarly, setting the next field to this is - * used as sentinel that node is off list. + * Queue nodes. Uses Object, not E, for items to allow forgetting + * them after use. Relies heavily on Unsafe mechanics to minimize + * unnecessary ordering constraints: Writes that are intrinsically + * ordered wrt other accesses or CASes use simple relaxed forms. */ - static final class QNode extends AtomicReference { - volatile QNode next; - volatile Thread waiter; // to control park/unpark - final boolean isData; - QNode(Object item, boolean isData) { - super(item); + static final class Node { + final boolean isData; // false if this is a request node + volatile Object item; // initially non-null if isData; CASed to match + volatile Node next; + volatile Thread waiter; // null until waiting + + // CAS methods for fields + final boolean casNext(Node cmp, Node val) { + return UNSAFE.compareAndSwapObject(this, nextOffset, cmp, val); + } + + final boolean casItem(Object cmp, Object val) { + // assert cmp == null || cmp.getClass() != Node.class; + return UNSAFE.compareAndSwapObject(this, itemOffset, cmp, val); + } + + /** + * Constructs a new node. Uses relaxed write because item can + * only be seen after publication via casNext. + */ + Node(Object item, boolean isData) { + UNSAFE.putObject(this, itemOffset, item); // relaxed write this.isData = isData; } - static final AtomicReferenceFieldUpdater - nextUpdater = AtomicReferenceFieldUpdater.newUpdater - (QNode.class, QNode.class, "next"); + /** + * Links node to itself to avoid garbage retention. Called + * only after CASing head field, so uses relaxed write. + */ + final void forgetNext() { + UNSAFE.putObject(this, nextOffset, this); + } + + /** + * Sets item to self and waiter to null, to avoid garbage + * retention after matching or cancelling. Uses relaxed writes + * because order is already constrained in the only calling + * contexts: item is forgotten only after volatile/atomic + * mechanics that extract items. Similarly, clearing waiter + * follows either CAS or return from park (if ever parked; + * else we don't care). + */ + final void forgetContents() { + UNSAFE.putObject(this, itemOffset, this); + UNSAFE.putObject(this, waiterOffset, null); + } - final boolean casNext(QNode cmp, QNode val) { - return nextUpdater.compareAndSet(this, cmp, val); + /** + * Returns true if this node has been matched, including the + * case of artificial matches due to cancellation. + */ + final boolean isMatched() { + Object x = item; + return (x == this) || ((x == null) == isData); } - final void clearNext() { - nextUpdater.lazySet(this, this); + /** + * Returns true if this is an unmatched request node. + */ + final boolean isUnmatchedRequest() { + return !isData && item == null; } - } + /** + * Returns true if a node with the given mode cannot be + * appended to this node because this node is unmatched and + * has opposite data mode. + */ + final boolean cannotPrecede(boolean haveData) { + boolean d = isData; + Object x; + return d != haveData && (x = item) != this && (x != null) == d; + } - /** - * Padded version of AtomicReference used for head, tail and - * cleanMe, to alleviate contention across threads CASing one vs - * the other. - */ - static final class PaddedAtomicReference extends AtomicReference { - // enough padding for 64bytes with 4byte refs - Object p0, p1, p2, p3, p4, p5, p6, p7, p8, p9, pa, pb, pc, pd, pe; - PaddedAtomicReference(T r) { super(r); } + /** + * Tries to artificially match a data node -- used by remove. + */ + final boolean tryMatchData() { + // assert isData; + Object x = item; + if (x != null && x != this && casItem(x, null)) { + LockSupport.unpark(waiter); + return true; + } + return false; + } + + private static final long serialVersionUID = -3375979862319811754L; + + // Unsafe mechanics + private static final sun.misc.Unsafe UNSAFE; + private static final long itemOffset; + private static final long nextOffset; + private static final long waiterOffset; + static { + try { + UNSAFE = sun.misc.Unsafe.getUnsafe(); + Class k = Node.class; + itemOffset = UNSAFE.objectFieldOffset + (k.getDeclaredField("item")); + nextOffset = UNSAFE.objectFieldOffset + (k.getDeclaredField("next")); + waiterOffset = UNSAFE.objectFieldOffset + (k.getDeclaredField("waiter")); + } catch (Exception e) { + throw new Error(e); + } + } } + /** head of the queue; null until first enqueue */ + transient volatile Node head; - /** head of the queue */ - private transient final PaddedAtomicReference head; - /** tail of the queue */ - private transient final PaddedAtomicReference tail; + /** tail of the queue; null until first append */ + private transient volatile Node tail; - /** - * Reference to a cancelled node that might not yet have been - * unlinked from queue because it was the last inserted node - * when it cancelled. - */ - private transient final PaddedAtomicReference cleanMe; + /** The number of apparent failures to unsplice removed nodes */ + private transient volatile int sweepVotes; - /** - * Tries to cas nh as new head; if successful, unlink - * old head's next node to avoid garbage retention. + // CAS methods for fields + private boolean casTail(Node cmp, Node val) { + return UNSAFE.compareAndSwapObject(this, tailOffset, cmp, val); + } + + private boolean casHead(Node cmp, Node val) { + return UNSAFE.compareAndSwapObject(this, headOffset, cmp, val); + } + + private boolean casSweepVotes(int cmp, int val) { + return UNSAFE.compareAndSwapInt(this, sweepVotesOffset, cmp, val); + } + + /* + * Possible values for "how" argument in xfer method. */ - private boolean advanceHead(QNode h, QNode nh) { - if (h == head.get() && head.compareAndSet(h, nh)) { - h.clearNext(); // forget old next - return true; - } - return false; + private static final int NOW = 0; // for untimed poll, tryTransfer + private static final int ASYNC = 1; // for offer, put, add + private static final int SYNC = 2; // for transfer, take + private static final int TIMED = 3; // for timed poll, tryTransfer + + @SuppressWarnings("unchecked") + static E cast(Object item) { + // assert item == null || item.getClass() != Node.class; + return (E) item; } /** - * Puts or takes an item. Used for most queue operations (except - * poll() and tryTransfer()). See the similar code in - * SynchronousQueue for detailed explanation. + * Implements all queuing methods. See above for explanation. * - * @param e the item or if null, signifies that this is a take - * @param mode the wait mode: NOWAIT, TIMEOUT, WAIT - * @param nanos timeout in nanosecs, used only if mode is TIMEOUT - * @return an item, or null on failure + * @param e the item or null for take + * @param haveData true if this is a put, else a take + * @param how NOW, ASYNC, SYNC, or TIMED + * @param nanos timeout in nanosecs, used only if mode is TIMED + * @return an item if matched, else e + * @throws NullPointerException if haveData mode but e is null */ - private Object xfer(Object e, int mode, long nanos) { - boolean isData = (e != null); - QNode s = null; - final PaddedAtomicReference head = this.head; - final PaddedAtomicReference tail = this.tail; - - for (;;) { - QNode t = tail.get(); - QNode h = head.get(); - - if (t != null && (t == h || t.isData == isData)) { - if (s == null) - s = new QNode(e, isData); - QNode last = t.next; - if (last != null) { - if (t == tail.get()) - tail.compareAndSet(t, last); - } - else if (t.casNext(null, s)) { - tail.compareAndSet(t, s); - return awaitFulfill(t, s, e, mode, nanos); + private E xfer(E e, boolean haveData, int how, long nanos) { + if (haveData && (e == null)) + throw new NullPointerException(); + Node s = null; // the node to append, if needed + + retry: + for (;;) { // restart on append race + + for (Node h = head, p = h; p != null;) { // find & match first node + boolean isData = p.isData; + Object item = p.item; + if (item != p && (item != null) == isData) { // unmatched + if (isData == haveData) // can't match + break; + if (p.casItem(item, e)) { // match + for (Node q = p; q != h;) { + Node n = q.next; // update by 2 unless singleton + if (head == h && casHead(h, n == null ? q : n)) { + h.forgetNext(); + break; + } // advance and retry + if ((h = head) == null || + (q = h.next) == null || !q.isMatched()) + break; // unless slack < 2 + } + LockSupport.unpark(p.waiter); + return this.cast(item); + } } + Node n = p.next; + p = (p != n) ? n : (h = head); // Use head if p offlist } - else if (h != null) { - QNode first = h.next; - if (t == tail.get() && first != null && - advanceHead(h, first)) { - Object x = first.get(); - if (x != first && first.compareAndSet(x, e)) { - LockSupport.unpark(first.waiter); - return isData? e : x; - } - } + if (how != NOW) { // No matches available + if (s == null) + s = new Node(e, haveData); + Node pred = tryAppend(s, haveData); + if (pred == null) + continue retry; // lost race vs opposite mode + if (how != ASYNC) + return awaitMatch(s, pred, e, (how == TIMED), nanos); } + return e; // not waiting } } - /** - * Version of xfer for poll() and tryTransfer, which - * simplifies control paths both here and in xfer. + * Tries to append node s as tail. + * + * @param s the node to append + * @param haveData true if appending in data mode + * @return null on failure due to losing race with append in + * different mode, else s's predecessor, or s itself if no + * predecessor */ - private Object fulfill(Object e) { - boolean isData = (e != null); - final PaddedAtomicReference head = this.head; - final PaddedAtomicReference tail = this.tail; - - for (;;) { - QNode t = tail.get(); - QNode h = head.get(); - - if (t != null && (t == h || t.isData == isData)) { - QNode last = t.next; - if (t == tail.get()) { - if (last != null) - tail.compareAndSet(t, last); - else - return null; - } + private Node tryAppend(Node s, boolean haveData) { + for (Node t = tail, p = t;;) { // move p to last node and append + Node n, u; // temps for reads of next & tail + if (p == null && (p = head) == null) { + if (casHead(null, s)) + return s; // initialize } - else if (h != null) { - QNode first = h.next; - if (t == tail.get() && - first != null && - advanceHead(h, first)) { - Object x = first.get(); - if (x != first && first.compareAndSet(x, e)) { - LockSupport.unpark(first.waiter); - return isData? e : x; - } + else if (p.cannotPrecede(haveData)) + return null; // lost race vs opposite mode + else if ((n = p.next) != null) // not last; keep traversing + p = p != t && t != (u = tail) ? (t = u) : // stale tail + (p != n) ? n : null; // restart if off list + else if (!p.casNext(null, s)) + p = p.next; // re-read on CAS failure + else { + if (p != t) { // update if slack now >= 2 + while ((tail != t || !casTail(t, s)) && + (t = tail) != null && + (s = t.next) != null && // advance and retry + (s = s.next) != null && s != t); } + return p; } } } /** - * Spins/blocks until node s is fulfilled or caller gives up, - * depending on wait mode. + * Spins/yields/blocks until node s is matched or caller gives up. * - * @param pred the predecessor of waiting node * @param s the waiting node + * @param pred the predecessor of s, or s itself if it has no + * predecessor, or null if unknown (the null case does not occur + * in any current calls but may in possible future extensions) * @param e the comparison value for checking match - * @param mode mode - * @param nanos timeout value - * @return matched item, or s if cancelled + * @param timed if true, wait only until timeout elapses + * @param nanos timeout in nanosecs, used only if timed is true + * @return matched item, or e if unmatched on interrupt or timeout */ - private Object awaitFulfill(QNode pred, QNode s, Object e, - int mode, long nanos) { - if (mode == NOWAIT) - return null; - - long lastTime = (mode == TIMEOUT)? System.nanoTime() : 0; + private E awaitMatch(Node s, Node pred, E e, boolean timed, long nanos) { + long lastTime = timed ? System.nanoTime() : 0L; Thread w = Thread.currentThread(); - int spins = -1; // set to desired spin count below + int spins = -1; // initialized after first item and cancel checks + ThreadLocalRandom randomYields = null; // bound if needed + for (;;) { - if (w.isInterrupted()) - s.compareAndSet(e, s); - Object x = s.get(); - if (x != e) { // Node was matched or cancelled - advanceHead(pred, s); // unlink if head - if (x == s) { // was cancelled - clean(pred, s); - return null; - } - else if (x != null) { - s.set(s); // avoid garbage retention - return x; - } - else - return e; + Object item = s.item; + if (item != e) { // matched + // assert item != s; + s.forgetContents(); // avoid garbage + return this.cast(item); } - if (mode == TIMEOUT) { - long now = System.nanoTime(); - nanos -= now - lastTime; - lastTime = now; - if (nanos <= 0) { - s.compareAndSet(e, s); // try to cancel - continue; - } + if ((w.isInterrupted() || (timed && nanos <= 0)) && + s.casItem(e, s)) { // cancel + unsplice(pred, s); + return e; } - if (spins < 0) { - QNode h = head.get(); // only spin if at head - spins = ((h != null && h.next == s) ? - (mode == TIMEOUT? - maxTimedSpins : maxUntimedSpins) : 0); + + if (spins < 0) { // establish spins at/near front + if ((spins = spinsFor(pred, s.isData)) > 0) + randomYields = ThreadLocalRandom.current(); } - if (spins > 0) + else if (spins > 0) { // spin --spins; - else if (s.waiter == null) - s.waiter = w; - else if (mode != TIMEOUT) { - LockSupport.park(this); - s.waiter = null; - spins = -1; + if (randomYields.nextInt(CHAINED_SPINS) == 0) + Thread.yield(); // occasionally yield + } + else if (s.waiter == null) { + s.waiter = w; // request unpark then recheck + } + else if (timed) { + long now = System.nanoTime(); + if ((nanos -= now - lastTime) > 0) + LockSupport.parkNanos(this, nanos); + lastTime = now; } - else if (nanos > spinForTimeoutThreshold) { - LockSupport.parkNanos(this, nanos); - s.waiter = null; - spins = -1; + else { + LockSupport.park(this); } } } /** - * Returns validated tail for use in cleaning methods. + * Returns spin/yield value for a node with given predecessor and + * data mode. See above for explanation. */ - private QNode getValidatedTail() { - for (;;) { - QNode h = head.get(); - QNode first = h.next; - if (first != null && first.next == first) { // help advance - advanceHead(h, first); - continue; - } - QNode t = tail.get(); - QNode last = t.next; - if (t == tail.get()) { - if (last != null) - tail.compareAndSet(t, last); // help advance - else - return t; + private static int spinsFor(Node pred, boolean haveData) { + if (MP && pred != null) { + if (pred.isData != haveData) // phase change + return FRONT_SPINS + CHAINED_SPINS; + if (pred.isMatched()) // probably at front + return FRONT_SPINS; + if (pred.waiter == null) // pred apparently spinning + return CHAINED_SPINS; + } + return 0; + } + + /* -------------- Traversal methods -------------- */ + + /** + * Returns the successor of p, or the head node if p.next has been + * linked to self, which will only be true if traversing with a + * stale pointer that is now off the list. + */ + final Node succ(Node p) { + Node next = p.next; + return (p == next) ? head : next; + } + + /** + * Returns the first unmatched node of the given mode, or null if + * none. Used by methods isEmpty, hasWaitingConsumer. + */ + private Node firstOfMode(boolean isData) { + for (Node p = head; p != null; p = succ(p)) { + if (!p.isMatched()) + return (p.isData == isData) ? p : null; + } + return null; + } + + /** + * Returns the item in the first unmatched node with isData; or + * null if none. Used by peek. + */ + private E firstDataItem() { + for (Node p = head; p != null; p = succ(p)) { + Object item = p.item; + if (p.isData) { + if (item != null && item != p) + return this.cast(item); } + else if (item == null) + return null; } + return null; } /** - * Gets rid of cancelled node s with original predecessor pred. - * - * @param pred predecessor of cancelled node - * @param s the cancelled node + * Traverses and counts unmatched nodes of the given mode. + * Used by methods size and getWaitingConsumerCount. */ - private void clean(QNode pred, QNode s) { - Thread w = s.waiter; - if (w != null) { // Wake up thread - s.waiter = null; - if (w != Thread.currentThread()) - LockSupport.unpark(w); + private int countOfMode(boolean data) { + int count = 0; + for (Node p = head; p != null; ) { + if (!p.isMatched()) { + if (p.isData != data) + return 0; + if (++count == Integer.MAX_VALUE) // saturated + break; + } + Node n = p.next; + if (n != p) + p = n; + else { + count = 0; + p = head; + } } + return count; + } - if (pred == null) - return; + final class Itr implements Iterator { + private Node nextNode; // next node to return item for + private E nextItem; // the corresponding item + private Node lastRet; // last returned node, to support remove + private Node lastPred; // predecessor to unlink lastRet - /* - * At any given time, exactly one node on list cannot be - * deleted -- the last inserted node. To accommodate this, if - * we cannot delete s, we save its predecessor as "cleanMe", - * processing the previously saved version first. At least one - * of node s or the node previously saved can always be - * processed, so this always terminates. + /** + * Moves to next node after prev, or first node if prev null. */ - while (pred.next == s) { - QNode oldpred = reclean(); // First, help get rid of cleanMe - QNode t = getValidatedTail(); - if (s != t) { // If not tail, try to unsplice - QNode sn = s.next; // s.next == s means s already off list - if (sn == s || pred.casNext(s, sn)) + private void advance(Node prev) { + /* + * To track and avoid buildup of deleted nodes in the face + * of calls to both Queue.remove and Itr.remove, we must + * include variants of unsplice and sweep upon each + * advance: Upon Itr.remove, we may need to catch up links + * from lastPred, and upon other removes, we might need to + * skip ahead from stale nodes and unsplice deleted ones + * found while advancing. + */ + + Node r, b; // reset lastPred upon possible deletion of lastRet + if ((r = lastRet) != null && !r.isMatched()) + lastPred = r; // next lastPred is old lastRet + else if ((b = lastPred) == null || b.isMatched()) + lastPred = null; // at start of list + else { + Node s, n; // help with removal of lastPred.next + while ((s = b.next) != null && + s != b && s.isMatched() && + (n = s.next) != null && n != s) + b.casNext(s, n); + } + + this.lastRet = prev; + + for (Node p = prev, s, n;;) { + s = (p == null) ? head : p.next; + if (s == null) + break; + else if (s == p) { + p = null; + continue; + } + Object item = s.item; + if (s.isData) { + if (item != null && item != s) { + nextItem = LinkedTransferQueue.cast(item); + nextNode = s; + return; + } + } + else if (item == null) + break; + // assert s.isMatched(); + if (p == null) + p = s; + else if ((n = s.next) == null) break; + else if (s == n) + p = null; + else + p.casNext(s, n); } - else if (oldpred == pred || // Already saved - (oldpred == null && cleanMe.compareAndSet(null, pred))) - break; // Postpone cleaning + nextNode = null; + nextItem = null; + } + + Itr() { + advance(null); + } + + public final boolean hasNext() { + return nextNode != null; + } + + public final E next() { + Node p = nextNode; + if (p == null) throw new NoSuchElementException(); + E e = nextItem; + advance(p); + return e; + } + + public final void remove() { + final Node lastRet = this.lastRet; + if (lastRet == null) + throw new IllegalStateException(); + this.lastRet = null; + if (lastRet.tryMatchData()) + unsplice(lastPred, lastRet); } } + /* -------------- Removal methods -------------- */ + /** - * Tries to unsplice the cancelled node held in cleanMe that was - * previously uncleanable because it was at tail. + * Unsplices (now or later) the given deleted/cancelled node with + * the given predecessor. * - * @return current cleanMe node (or null) + * @param pred a node that was at one time known to be the + * predecessor of s, or null or s itself if s is/was at head + * @param s the node to be unspliced */ - private QNode reclean() { + final void unsplice(Node pred, Node s) { + s.forgetContents(); // forget unneeded fields /* - * cleanMe is, or at one time was, predecessor of cancelled - * node s that was the tail so could not be unspliced. If s - * is no longer the tail, try to unsplice if necessary and - * make cleanMe slot available. This differs from similar - * code in clean() because we must check that pred still - * points to a cancelled node that must be unspliced -- if - * not, we can (must) clear cleanMe without unsplicing. - * This can loop only due to contention on casNext or - * clearing cleanMe. + * See above for rationale. Briefly: if pred still points to + * s, try to unlink s. If s cannot be unlinked, because it is + * trailing node or pred might be unlinked, and neither pred + * nor s are head or offlist, add to sweepVotes, and if enough + * votes have accumulated, sweep. */ - QNode pred; - while ((pred = cleanMe.get()) != null) { - QNode t = getValidatedTail(); - QNode s = pred.next; - if (s != t) { - QNode sn; - if (s == null || s == pred || s.get() != s || - (sn = s.next) == s || pred.casNext(s, sn)) - cleanMe.compareAndSet(pred, null); + if (pred != null && pred != s && pred.next == s) { + Node n = s.next; + if (n == null || + (n != s && pred.casNext(s, n) && pred.isMatched())) { + for (;;) { // check if at, or could be, head + Node h = head; + if (h == pred || h == s || h == null) + return; // at head or list empty + if (!h.isMatched()) + break; + Node hn = h.next; + if (hn == null) + return; // now empty + if (hn != h && casHead(h, hn)) + h.forgetNext(); // advance head + } + if (pred.next != pred && s.next != s) { // recheck if offlist + for (;;) { // sweep now if enough votes + int v = sweepVotes; + if (v < SWEEP_THRESHOLD) { + if (casSweepVotes(v, v + 1)) + break; + } + else if (casSweepVotes(v, 0)) { + sweep(); + break; + } + } + } } - else // s is still tail; cannot clean + } + } + + /** + * Unlinks matched (typically cancelled) nodes encountered in a + * traversal from head. + */ + private void sweep() { + for (Node p = head, s, n; p != null && (s = p.next) != null; ) { + if (!s.isMatched()) + // Unmatched nodes are never self-linked + p = s; + else if ((n = s.next) == null) // trailing node is pinned break; + else if (s == n) // stale + // No need to also check for p == s, since that implies s == n + p = head; + else + p.casNext(s, n); + } + } + + /** + * Main implementation of remove(Object) + */ + private boolean findAndRemove(Object e) { + if (e != null) { + for (Node pred = null, p = head; p != null; ) { + Object item = p.item; + if (p.isData) { + if (item != null && item != p && e.equals(item) && + p.tryMatchData()) { + unsplice(pred, p); + return true; + } + } + else if (item == null) + break; + pred = p; + if ((p = p.next) == pred) { // stale + pred = null; + p = head; + } + } } - return pred; + return false; } + /** * Creates an initially empty {@code LinkedTransferQueue}. */ public LinkedTransferQueue() { - QNode dummy = new QNode(null, false); - head = new PaddedAtomicReference(dummy); - tail = new PaddedAtomicReference(dummy); - cleanMe = new PaddedAtomicReference(null); } /** @@ -435,74 +1000,133 @@ public class LinkedTransferQueue extends AbstractQueue addAll(c); } - public void put(E e) throws InterruptedException { - if (e == null) throw new NullPointerException(); - if (Thread.interrupted()) throw new InterruptedException(); - xfer(e, NOWAIT, 0); + /** + * Inserts the specified element at the tail of this queue. + * As the queue is unbounded, this method will never block. + * + * @throws NullPointerException if the specified element is null + */ + public void put(E e) { + xfer(e, true, ASYNC, 0); } - public boolean offer(E e, long timeout, TimeUnit unit) - throws InterruptedException { - if (e == null) throw new NullPointerException(); - if (Thread.interrupted()) throw new InterruptedException(); - xfer(e, NOWAIT, 0); + /** + * Inserts the specified element at the tail of this queue. + * As the queue is unbounded, this method will never block or + * return {@code false}. + * + * @return {@code true} (as specified by + * {@link BlockingQueue#offer(Object,long,TimeUnit) BlockingQueue.offer}) + * @throws NullPointerException if the specified element is null + */ + public boolean offer(E e, long timeout, TimeUnit unit) { + xfer(e, true, ASYNC, 0); return true; } + /** + * Inserts the specified element at the tail of this queue. + * As the queue is unbounded, this method will never return {@code false}. + * + * @return {@code true} (as specified by {@link Queue#offer}) + * @throws NullPointerException if the specified element is null + */ public boolean offer(E e) { - if (e == null) throw new NullPointerException(); - xfer(e, NOWAIT, 0); + xfer(e, true, ASYNC, 0); return true; } + /** + * Inserts the specified element at the tail of this queue. + * As the queue is unbounded, this method will never throw + * {@link IllegalStateException} or return {@code false}. + * + * @return {@code true} (as specified by {@link Collection#add}) + * @throws NullPointerException if the specified element is null + */ public boolean add(E e) { - if (e == null) throw new NullPointerException(); - xfer(e, NOWAIT, 0); + xfer(e, true, ASYNC, 0); return true; } + /** + * Transfers the element to a waiting consumer immediately, if possible. + * + *

More precisely, transfers the specified element immediately + * if there exists a consumer already waiting to receive it (in + * {@link #take} or timed {@link #poll(long,TimeUnit) poll}), + * otherwise returning {@code false} without enqueuing the element. + * + * @throws NullPointerException if the specified element is null + */ + public boolean tryTransfer(E e) { + return xfer(e, true, NOW, 0) == null; + } + + /** + * Transfers the element to a consumer, waiting if necessary to do so. + * + *

More precisely, transfers the specified element immediately + * if there exists a consumer already waiting to receive it (in + * {@link #take} or timed {@link #poll(long,TimeUnit) poll}), + * else inserts the specified element at the tail of this queue + * and waits until the element is received by a consumer. + * + * @throws NullPointerException if the specified element is null + */ public void transfer(E e) throws InterruptedException { - if (e == null) throw new NullPointerException(); - if (xfer(e, WAIT, 0) == null) { - Thread.interrupted(); + if (xfer(e, true, SYNC, 0) != null) { + Thread.interrupted(); // failure possible only due to interrupt throw new InterruptedException(); } } + /** + * Transfers the element to a consumer if it is possible to do so + * before the timeout elapses. + * + *

More precisely, transfers the specified element immediately + * if there exists a consumer already waiting to receive it (in + * {@link #take} or timed {@link #poll(long,TimeUnit) poll}), + * else inserts the specified element at the tail of this queue + * and waits until the element is received by a consumer, + * returning {@code false} if the specified wait time elapses + * before the element can be transferred. + * + * @throws NullPointerException if the specified element is null + */ public boolean tryTransfer(E e, long timeout, TimeUnit unit) throws InterruptedException { - if (e == null) throw new NullPointerException(); - if (xfer(e, TIMEOUT, unit.toNanos(timeout)) != null) + if (xfer(e, true, TIMED, unit.toNanos(timeout)) == null) return true; if (!Thread.interrupted()) return false; throw new InterruptedException(); } - public boolean tryTransfer(E e) { - if (e == null) throw new NullPointerException(); - return fulfill(e) != null; - } - public E take() throws InterruptedException { - Object e = xfer(null, WAIT, 0); + E e = xfer(null, false, SYNC, 0); if (e != null) - return (E)e; + return e; Thread.interrupted(); throw new InterruptedException(); } public E poll(long timeout, TimeUnit unit) throws InterruptedException { - Object e = xfer(null, TIMEOUT, unit.toNanos(timeout)); + E e = xfer(null, false, TIMED, unit.toNanos(timeout)); if (e != null || !Thread.interrupted()) - return (E)e; + return e; throw new InterruptedException(); } public E poll() { - return (E)fulfill(null); + return xfer(null, false, NOW, 0); } + /** + * @throws NullPointerException {@inheritDoc} + * @throws IllegalArgumentException {@inheritDoc} + */ public int drainTo(Collection c) { if (c == null) throw new NullPointerException(); @@ -517,6 +1141,10 @@ public class LinkedTransferQueue extends AbstractQueue return n; } + /** + * @throws NullPointerException {@inheritDoc} + * @throws IllegalArgumentException {@inheritDoc} + */ public int drainTo(Collection c, int maxElements) { if (c == null) throw new NullPointerException(); @@ -531,156 +1159,42 @@ public class LinkedTransferQueue extends AbstractQueue return n; } - // Traversal-based methods - /** - * Returns head after performing any outstanding helping steps. + * Returns an iterator over the elements in this queue in proper sequence. + * The elements will be returned in order from first (head) to last (tail). + * + *

The returned iterator is a "weakly consistent" iterator that + * will never throw {@link java.util.ConcurrentModificationException + * ConcurrentModificationException}, and guarantees to traverse + * elements as they existed upon construction of the iterator, and + * may (but is not guaranteed to) reflect any modifications + * subsequent to construction. + * + * @return an iterator over the elements in this queue in proper sequence */ - private QNode traversalHead() { - for (;;) { - QNode t = tail.get(); - QNode h = head.get(); - if (h != null && t != null) { - QNode last = t.next; - QNode first = h.next; - if (t == tail.get()) { - if (last != null) - tail.compareAndSet(t, last); - else if (first != null) { - Object x = first.get(); - if (x == first) - advanceHead(h, first); - else - return h; - } - else - return h; - } - } - reclean(); - } - } - - public Iterator iterator() { return new Itr(); } - /** - * Iterators. Basic strategy is to traverse list, treating - * non-data (i.e., request) nodes as terminating list. - * Once a valid data node is found, the item is cached - * so that the next call to next() will return it even - * if subsequently removed. - */ - class Itr implements Iterator { - QNode next; // node to return next - QNode pnext; // predecessor of next - QNode snext; // successor of next - QNode curr; // last returned node, for remove() - QNode pcurr; // predecessor of curr, for remove() - E nextItem; // Cache of next item, once commited to in next - - Itr() { - findNext(); - } - - /** - * Ensures next points to next valid node, or null if none. - */ - void findNext() { - for (;;) { - QNode pred = pnext; - QNode q = next; - if (pred == null || pred == q) { - pred = traversalHead(); - q = pred.next; - } - if (q == null || !q.isData) { - next = null; - return; - } - Object x = q.get(); - QNode s = q.next; - if (x != null && q != x && q != s) { - nextItem = (E)x; - snext = s; - pnext = pred; - next = q; - return; - } - pnext = q; - next = s; - } - } - - public boolean hasNext() { - return next != null; - } - - public E next() { - if (next == null) throw new NoSuchElementException(); - pcurr = pnext; - curr = next; - pnext = next; - next = snext; - E x = nextItem; - findNext(); - return x; - } - - public void remove() { - QNode p = curr; - if (p == null) - throw new IllegalStateException(); - Object x = p.get(); - if (x != null && x != p && p.compareAndSet(x, p)) - clean(pcurr, p); - } - } - public E peek() { - for (;;) { - QNode h = traversalHead(); - QNode p = h.next; - if (p == null) - return null; - Object x = p.get(); - if (p != x) { - if (!p.isData) - return null; - if (x != null) - return (E)x; - } - } + return firstDataItem(); } + /** + * Returns {@code true} if this queue contains no elements. + * + * @return {@code true} if this queue contains no elements + */ public boolean isEmpty() { - for (;;) { - QNode h = traversalHead(); - QNode p = h.next; - if (p == null) - return true; - Object x = p.get(); - if (p != x) { - if (!p.isData) - return true; - if (x != null) - return false; - } + for (Node p = head; p != null; p = succ(p)) { + if (!p.isMatched()) + return !p.isData; } + return true; } public boolean hasWaitingConsumer() { - for (;;) { - QNode h = traversalHead(); - QNode p = h.next; - if (p == null) - return false; - Object x = p.get(); - if (p != x) - return !p.isData; - } + return firstOfMode(false) != null; } /** @@ -696,58 +1210,63 @@ public class LinkedTransferQueue extends AbstractQueue * @return the number of elements in this queue */ public int size() { - int count = 0; - QNode h = traversalHead(); - for (QNode p = h.next; p != null && p.isData; p = p.next) { - Object x = p.get(); - if (x != null && x != p) { - if (++count == Integer.MAX_VALUE) // saturated - break; - } - } - return count; + return countOfMode(true); } public int getWaitingConsumerCount() { - int count = 0; - QNode h = traversalHead(); - for (QNode p = h.next; p != null && !p.isData; p = p.next) { - if (p.get() == null) { - if (++count == Integer.MAX_VALUE) - break; - } - } - return count; + return countOfMode(false); } - public int remainingCapacity() { - return Integer.MAX_VALUE; + /** + * Removes a single instance of the specified element from this queue, + * if it is present. More formally, removes an element {@code e} such + * that {@code o.equals(e)}, if this queue contains one or more such + * elements. + * Returns {@code true} if this queue contained the specified element + * (or equivalently, if this queue changed as a result of the call). + * + * @param o element to be removed from this queue, if present + * @return {@code true} if this queue changed as a result of the call + */ + public boolean remove(Object o) { + return findAndRemove(o); } - public boolean remove(Object o) { - if (o == null) - return false; - for (;;) { - QNode pred = traversalHead(); - for (;;) { - QNode q = pred.next; - if (q == null || !q.isData) - return false; - if (q == pred) // restart - break; - Object x = q.get(); - if (x != null && x != q && o.equals(x) && - q.compareAndSet(x, q)) { - clean(pred, q); + /** + * Returns {@code true} if this queue contains the specified element. + * More formally, returns {@code true} if and only if this queue contains + * at least one element {@code e} such that {@code o.equals(e)}. + * + * @param o object to be checked for containment in this queue + * @return {@code true} if this queue contains the specified element + */ + public boolean contains(Object o) { + if (o == null) return false; + for (Node p = head; p != null; p = succ(p)) { + Object item = p.item; + if (p.isData) { + if (item != null && item != p && o.equals(item)) return true; - } - pred = q; } + else if (item == null) + break; } + return false; + } + + /** + * Always returns {@code Integer.MAX_VALUE} because a + * {@code LinkedTransferQueue} is not capacity constrained. + * + * @return {@code Integer.MAX_VALUE} (as specified by + * {@link BlockingQueue#remainingCapacity()}) + */ + public int remainingCapacity() { + return Integer.MAX_VALUE; } /** - * Save the state to a stream (that is, serialize it). + * Saves the state to a stream (that is, serializes it). * * @serialData All of the elements (each an {@code E}) in * the proper order, followed by a null @@ -763,16 +1282,16 @@ public class LinkedTransferQueue extends AbstractQueue } /** - * Reconstitute the Queue instance from a stream (that is, - * deserialize it). + * Reconstitutes the Queue instance from a stream (that is, + * deserializes it). + * * @param s the stream */ private void readObject(java.io.ObjectInputStream s) throws java.io.IOException, ClassNotFoundException { s.defaultReadObject(); - resetHeadAndTail(); for (;;) { - E item = (E)s.readObject(); + @SuppressWarnings("unchecked") E item = (E) s.readObject(); if (item == null) break; else @@ -780,61 +1299,24 @@ public class LinkedTransferQueue extends AbstractQueue } } + // Unsafe mechanics - // Support for resetting head/tail while deserializing - private void resetHeadAndTail() { - QNode dummy = new QNode(null, false); - _unsafe.putObjectVolatile(this, headOffset, - new PaddedAtomicReference(dummy)); - _unsafe.putObjectVolatile(this, tailOffset, - new PaddedAtomicReference(dummy)); - _unsafe.putObjectVolatile(this, cleanMeOffset, - new PaddedAtomicReference(null)); - } - - // Temporary Unsafe mechanics for preliminary release - private static Unsafe getUnsafe() throws Throwable { - try { - return Unsafe.getUnsafe(); - } catch (SecurityException se) { - try { - return java.security.AccessController.doPrivileged - (new java.security.PrivilegedExceptionAction() { - public Unsafe run() throws Exception { - return getUnsafePrivileged(); - }}); - } catch (java.security.PrivilegedActionException e) { - throw 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 - (LinkedTransferQueue.class.getDeclaredField(fieldName)); - } - - private static final Unsafe _unsafe; + private static final sun.misc.Unsafe UNSAFE; private static final long headOffset; private static final long tailOffset; - private static final long cleanMeOffset; + private static final long sweepVotesOffset; static { try { - _unsafe = getUnsafe(); - headOffset = fieldOffset("head"); - tailOffset = fieldOffset("tail"); - cleanMeOffset = fieldOffset("cleanMe"); - } catch (Throwable e) { - throw new RuntimeException("Could not initialize intrinsics", e); + UNSAFE = sun.misc.Unsafe.getUnsafe(); + Class k = LinkedTransferQueue.class; + headOffset = UNSAFE.objectFieldOffset + (k.getDeclaredField("head")); + tailOffset = UNSAFE.objectFieldOffset + (k.getDeclaredField("tail")); + sweepVotesOffset = UNSAFE.objectFieldOffset + (k.getDeclaredField("sweepVotes")); + } catch (Exception e) { + throw new Error(e); } } - } diff --git a/src/forkjoin/scala/concurrent/forkjoin/RecursiveAction.java b/src/forkjoin/scala/concurrent/forkjoin/RecursiveAction.java index 2d36f7eb33..bb24d9e575 100644 --- a/src/forkjoin/scala/concurrent/forkjoin/RecursiveAction.java +++ b/src/forkjoin/scala/concurrent/forkjoin/RecursiveAction.java @@ -1,64 +1,61 @@ /* * Written by Doug Lea with assistance from members of JCP JSR-166 * Expert Group and released to the public domain, as explained at - * http://creativecommons.org/licenses/publicdomain + * http://creativecommons.org/publicdomain/zero/1.0/ */ package scala.concurrent.forkjoin; /** - * Recursive resultless ForkJoinTasks. This class establishes - * conventions to parameterize resultless actions as Void - * ForkJoinTasks. Because null is the only valid value of - * Void, methods such as join always return null - * upon completion. + * A recursive resultless {@link ForkJoinTask}. This class + * establishes conventions to parameterize resultless actions as + * {@code Void} {@code ForkJoinTask}s. Because {@code null} is the + * only valid value of type {@code Void}, methods such as join always + * return {@code null} upon completion. * *

Sample Usages. Here is a sketch of a ForkJoin sort that - * sorts a given long[] array: + * sorts a given {@code long[]} array: * - * 

+ *  
 {@code
  * class SortTask extends RecursiveAction {
  *   final long[] array; final int lo; final int hi;
  *   SortTask(long[] array, int lo, int hi) {
  *     this.array = array; this.lo = lo; this.hi = hi;
  *   }
  *   protected void compute() {
- *     if (hi - lo < THRESHOLD)
+ *     if (hi - lo < THRESHOLD)
  *       sequentiallySort(array, lo, hi);
  *     else {
- *       int mid = (lo + hi) >>> 1;
+ *       int mid = (lo + hi) >>> 1;
  *       invokeAll(new SortTask(array, lo, mid),
  *                 new SortTask(array, mid, hi));
  *       merge(array, lo, hi);
  *     }
  *   }
- * }
- * 

+ * }}
* - * You could then sort anArray by creating new SortTask(anArray, 0, - * anArray.length-1) and invoking it in a ForkJoinPool. - * As a more concrete simple example, the following task increments - * each element of an array: - * 
+ * You could then sort {@code anArray} by creating {@code new
+ * SortTask(anArray, 0, anArray.length-1) } and invoking it in a
+ * ForkJoinPool.  As a more concrete simple example, the following
+ * task increments each element of an array:
+ *  
 {@code
  * class IncrementTask extends RecursiveAction {
  *   final long[] array; final int lo; final int hi;
  *   IncrementTask(long[] array, int lo, int hi) {
  *     this.array = array; this.lo = lo; this.hi = hi;
  *   }
  *   protected void compute() {
- *     if (hi - lo < THRESHOLD) {
- *       for (int i = lo; i < hi; ++i)
+ *     if (hi - lo < THRESHOLD) {
+ *       for (int i = lo; i < hi; ++i)
  *         array[i]++;
  *     }
  *     else {
- *       int mid = (lo + hi) >>> 1;
+ *       int mid = (lo + hi) >>> 1;
  *       invokeAll(new IncrementTask(array, lo, mid),
  *                 new IncrementTask(array, mid, hi));
  *     }
  *   }
- * }
- * 

- *
+ * }}
* *

The following example illustrates some refinements and idioms * that may lead to better performance: RecursiveActions need not be @@ -66,33 +63,33 @@ package scala.concurrent.forkjoin; * divide-and-conquer approach. Here is a class that sums the squares * of each element of a double array, by subdividing out only the * right-hand-sides of repeated divisions by two, and keeping track of - * them with a chain of next references. It uses a dynamic - * threshold based on method surplus, but counterbalances - * potential excess partitioning by directly performing leaf actions - * on unstolen tasks rather than further subdividing. + * them with a chain of {@code next} references. It uses a dynamic + * threshold based on method {@code getSurplusQueuedTaskCount}, but + * counterbalances potential excess partitioning by directly + * performing leaf actions on unstolen tasks rather than further + * subdividing. * - * 

+ *  
 {@code
  * double sumOfSquares(ForkJoinPool pool, double[] array) {
  *   int n = array.length;
- *   int seqSize = 1 + n / (8 * pool.getParallelism());
- *   Applyer a = new Applyer(array, 0, n, seqSize, null);
+ *   Applyer a = new Applyer(array, 0, n, null);
  *   pool.invoke(a);
  *   return a.result;
  * }
  *
  * class Applyer extends RecursiveAction {
  *   final double[] array;
- *   final int lo, hi, seqSize;
+ *   final int lo, hi;
  *   double result;
  *   Applyer next; // keeps track of right-hand-side tasks
- *   Applyer(double[] array, int lo, int hi, int seqSize, Applyer next) {
+ *   Applyer(double[] array, int lo, int hi, Applyer next) {
  *     this.array = array; this.lo = lo; this.hi = hi;
- *     this.seqSize = seqSize; this.next = next;
+ *     this.next = next;
  *   }
  *
- *   double atLeaf(int l, int r) {
+ *   double atLeaf(int l, int h) {
  *     double sum = 0;
- *     for (int i = l; i < h; ++i) // perform leftmost base step
+ *     for (int i = l; i < h; ++i) // perform leftmost base step
  *       sum += array[i] * array[i];
  *     return sum;
  *   }
@@ -101,10 +98,9 @@ package scala.concurrent.forkjoin;
  *     int l = lo;
  *     int h = hi;
  *     Applyer right = null;
- *     while (h - l > 1 &&
- *        ForkJoinWorkerThread.getEstimatedSurplusTaskCount() <= 3) {
- *        int mid = (l + h) >>> 1;
- *        right = new Applyer(array, mid, h, seqSize, right);
+ *     while (h - l > 1 && getSurplusQueuedTaskCount() <= 3) {
+ *        int mid = (l + h) >>> 1;
+ *        right = new Applyer(array, mid, h, right);
  *        right.fork();
  *        h = mid;
  *     }
@@ -113,17 +109,20 @@ package scala.concurrent.forkjoin;
  *        if (right.tryUnfork()) // directly calculate if not stolen
  *          sum += right.atLeaf(right.lo, right.hi);
  *       else {
- *          right.helpJoin();
+ *          right.join();
  *          sum += right.result;
  *        }
  *        right = right.next;
  *      }
  *     result = sum;
  *   }
- * }
- * 

+ * }}
+ * + * @since 1.7 + * @author Doug Lea */ public abstract class RecursiveAction extends ForkJoinTask { + private static final long serialVersionUID = 5232453952276485070L; /** * The main computation performed by this task. @@ -131,7 +130,9 @@ public abstract class RecursiveAction extends ForkJoinTask { protected abstract void compute(); /** - * Always returns null + * Always returns {@code null}. + * + * @return {@code null} always */ public final Void getRawResult() { return null; } @@ -141,7 +142,7 @@ public abstract class RecursiveAction extends ForkJoinTask { protected final void setRawResult(Void mustBeNull) { } /** - * Implements execution conventions for RecursiveActions + * Implements execution conventions for RecursiveActions. */ protected final boolean exec() { compute(); diff --git a/src/forkjoin/scala/concurrent/forkjoin/RecursiveTask.java b/src/forkjoin/scala/concurrent/forkjoin/RecursiveTask.java index a526f75597..d1e1547143 100644 --- a/src/forkjoin/scala/concurrent/forkjoin/RecursiveTask.java +++ b/src/forkjoin/scala/concurrent/forkjoin/RecursiveTask.java @@ -1,29 +1,29 @@ /* * Written by Doug Lea with assistance from members of JCP JSR-166 * Expert Group and released to the public domain, as explained at - * http://creativecommons.org/licenses/publicdomain + * http://creativecommons.org/publicdomain/zero/1.0/ */ package scala.concurrent.forkjoin; /** - * Recursive result-bearing ForkJoinTasks. - *

For a classic example, here is a task computing Fibonacci numbers: + * A recursive result-bearing {@link ForkJoinTask}. * - * 

- * class Fibonacci extends RecursiveTask<Integer> {
+ * 
For a classic example, here is a task computing Fibonacci numbers:
+ *
+ *  
 {@code
+ * class Fibonacci extends RecursiveTask {
  *   final int n;
- *   Fibonnaci(int n) { this.n = n; }
+ *   Fibonacci(int n) { this.n = n; }
  *   Integer compute() {
- *     if (n <= 1)
+ *     if (n <= 1)
  *        return n;
  *     Fibonacci f1 = new Fibonacci(n - 1);
  *     f1.fork();
  *     Fibonacci f2 = new Fibonacci(n - 2);
  *     return f2.compute() + f1.join();
  *   }
- * }
- * 

+ * }}
* * However, besides being a dumb way to compute Fibonacci functions * (there is a simple fast linear algorithm that you'd use in @@ -33,17 +33,14 @@ package scala.concurrent.forkjoin; * minimum granularity size (for example 10 here) for which you always * sequentially solve rather than subdividing. * + * @since 1.7 + * @author Doug Lea */ public abstract class RecursiveTask extends ForkJoinTask { + private static final long serialVersionUID = 5232453952276485270L; /** - * Empty constructor for use by subclasses. - */ - protected RecursiveTask() { - } - - /** - * The result returned by compute method. + * The result of the computation. */ V result; @@ -61,7 +58,7 @@ public abstract class RecursiveTask extends ForkJoinTask { } /** - * Implements execution conventions for RecursiveTask + * Implements execution conventions for RecursiveTask. */ protected final boolean exec() { result = compute(); diff --git a/src/forkjoin/scala/concurrent/forkjoin/ThreadLocalRandom.java b/src/forkjoin/scala/concurrent/forkjoin/ThreadLocalRandom.java index 34e2e37f37..19237c9092 100644 --- a/src/forkjoin/scala/concurrent/forkjoin/ThreadLocalRandom.java +++ b/src/forkjoin/scala/concurrent/forkjoin/ThreadLocalRandom.java @@ -1,49 +1,53 @@ /* * Written by Doug Lea with assistance from members of JCP JSR-166 * Expert Group and released to the public domain, as explained at - * http://creativecommons.org/licenses/publicdomain + * http://creativecommons.org/publicdomain/zero/1.0/ */ package scala.concurrent.forkjoin; -import java.util.*; + +import java.util.Random; /** - * A random number generator with the same properties as class {@link - * Random} but isolated to the current Thread. Like the global - * generator used by the {@link java.lang.Math} class, a - * ThreadLocalRandom is initialized with an internally generated seed - * that may not otherwise be modified. When applicable, use of - * ThreadLocalRandom rather than shared Random objects in concurrent - * programs will typically encounter much less overhead and - * contention. ThreadLocalRandoms are particularly appropriate when - * multiple tasks (for example, each a {@link ForkJoinTask}), use - * random numbers in parallel in thread pools. + * A random number generator isolated to the current thread. Like the + * global {@link java.util.Random} generator used by the {@link + * java.lang.Math} class, a {@code ThreadLocalRandom} is initialized + * with an internally generated seed that may not otherwise be + * modified. When applicable, use of {@code ThreadLocalRandom} rather + * than shared {@code Random} objects in concurrent programs will + * typically encounter much less overhead and contention. Use of + * {@code ThreadLocalRandom} is particularly appropriate when multiple + * tasks (for example, each a {@link ForkJoinTask}) use random numbers + * in parallel in thread pools. * *

Usages of this class should typically be of the form: - * ThreadLocalRandom.current().nextX(...) (where - * X is Int, Long, etc). + * {@code ThreadLocalRandom.current().nextX(...)} (where + * {@code X} is {@code Int}, {@code Long}, etc). * When all usages are of this form, it is never possible to - * accidently share ThreadLocalRandoms across multiple threads. + * accidently share a {@code ThreadLocalRandom} across multiple threads. * *

This class also provides additional commonly used bounded random * generation methods. + * + * @since 1.7 + * @author Doug Lea */ public class ThreadLocalRandom extends Random { // same constants as Random, but must be redeclared because private - private final static long multiplier = 0x5DEECE66DL; - private final static long addend = 0xBL; - private final static long mask = (1L << 48) - 1; + private static final long multiplier = 0x5DEECE66DL; + private static final long addend = 0xBL; + private static final long mask = (1L << 48) - 1; /** - * The random seed. We can't use super.seed + * The random seed. We can't use super.seed. */ private long rnd; /** - * Initialization flag to permit the first and only allowed call - * to setSeed (inside Random constructor) to succeed. We can't - * allow others since it would cause setting seed in one part of a - * program to unintentionally impact other usages by the thread. + * Initialization flag to permit calls to setSeed to succeed only + * while executing the Random constructor. We can't allow others + * since it would cause setting seed in one part of a program to + * unintentionally impact other usages by the thread. */ boolean initialized; @@ -65,40 +69,42 @@ public class ThreadLocalRandom extends Random { /** * Constructor called only by localRandom.initialValue. - * We rely on the fact that the superclass no-arg constructor - * invokes setSeed exactly once to initialize. */ ThreadLocalRandom() { super(); + initialized = true; } /** - * Returns the current Thread's ThreadLocalRandom - * @return the current Thread's ThreadLocalRandom + * Returns the current thread's {@code ThreadLocalRandom}. + * + * @return the current thread's {@code ThreadLocalRandom} */ public static ThreadLocalRandom current() { return localRandom.get(); } /** - * Throws UnsupportedOperationException. Setting seeds in this - * generator is unsupported. + * Throws {@code UnsupportedOperationException}. Setting seeds in + * this generator is not supported. + * * @throws UnsupportedOperationException always */ public void setSeed(long seed) { if (initialized) throw new UnsupportedOperationException(); - initialized = true; rnd = (seed ^ multiplier) & mask; } protected int next(int bits) { - return (int)((rnd = (rnd * multiplier + addend) & mask) >>> (48-bits)); + rnd = (rnd * multiplier + addend) & mask; + return (int) (rnd >>> (48-bits)); } /** * Returns a pseudorandom, uniformly distributed value between the * given least value (inclusive) and bound (exclusive). + * * @param least the least value returned * @param bound the upper bound (exclusive) * @throws IllegalArgumentException if least greater than or equal @@ -113,7 +119,8 @@ public class ThreadLocalRandom extends Random { /** * Returns a pseudorandom, uniformly distributed value - * between 0 (inclusive) and the specified value (exclusive) + * between 0 (inclusive) and the specified value (exclusive). + * * @param n the bound on the random number to be returned. Must be * positive. * @return the next value @@ -131,17 +138,18 @@ public class ThreadLocalRandom extends Random { while (n >= Integer.MAX_VALUE) { int bits = next(2); long half = n >>> 1; - long nextn = ((bits & 2) == 0)? half : n - half; + long nextn = ((bits & 2) == 0) ? half : n - half; if ((bits & 1) == 0) offset += n - nextn; n = nextn; } - return offset + nextInt((int)n); + return offset + nextInt((int) n); } /** * Returns a pseudorandom, uniformly distributed value between the * given least value (inclusive) and bound (exclusive). + * * @param least the least value returned * @param bound the upper bound (exclusive) * @return the next value @@ -156,7 +164,8 @@ public class ThreadLocalRandom extends Random { /** * Returns a pseudorandom, uniformly distributed {@code double} value - * between 0 (inclusive) and the specified value (exclusive) + * between 0 (inclusive) and the specified value (exclusive). + * * @param n the bound on the random number to be returned. Must be * positive. * @return the next value @@ -171,6 +180,7 @@ public class ThreadLocalRandom extends Random { /** * Returns a pseudorandom, uniformly distributed value between the * given least value (inclusive) and bound (exclusive). + * * @param least the least value returned * @param bound the upper bound (exclusive) * @return the next value @@ -183,4 +193,5 @@ public class ThreadLocalRandom extends Random { return nextDouble() * (bound - least) + least; } + private static final long serialVersionUID = -5851777807851030925L; } diff --git a/src/forkjoin/scala/concurrent/forkjoin/TransferQueue.java b/src/forkjoin/scala/concurrent/forkjoin/TransferQueue.java index 9c7b2289c4..422846be78 100644 --- a/src/forkjoin/scala/concurrent/forkjoin/TransferQueue.java +++ b/src/forkjoin/scala/concurrent/forkjoin/TransferQueue.java @@ -1,31 +1,35 @@ /* * Written by Doug Lea with assistance from members of JCP JSR-166 * Expert Group and released to the public domain, as explained at - * http://creativecommons.org/licenses/publicdomain + * http://creativecommons.org/publicdomain/zero/1.0/ */ package scala.concurrent.forkjoin; -import java.util.concurrent.*; + +import java.util.concurrent.BlockingQueue; +import java.util.concurrent.TimeUnit; /** * A {@link BlockingQueue} in which producers may wait for consumers * to receive elements. A {@code TransferQueue} may be useful for * example in message passing applications in which producers - * sometimes (using method {@code transfer}) await receipt of - * elements by consumers invoking {@code take} or {@code poll}, - * while at other times enqueue elements (via method {@code put}) - * without waiting for receipt. Non-blocking and time-out versions of - * {@code tryTransfer} are also available. A TransferQueue may also - * be queried via {@code hasWaitingConsumer} whether there are any - * threads waiting for items, which is a converse analogy to a - * {@code peek} operation. + * sometimes (using method {@link #transfer}) await receipt of + * elements by consumers invoking {@code take} or {@code poll}, while + * at other times enqueue elements (via method {@code put}) without + * waiting for receipt. + * {@linkplain #tryTransfer(Object) Non-blocking} and + * {@linkplain #tryTransfer(Object,long,TimeUnit) time-out} versions of + * {@code tryTransfer} are also available. + * A {@code TransferQueue} may also be queried, via {@link + * #hasWaitingConsumer}, whether there are any threads waiting for + * items, which is a converse analogy to a {@code peek} operation. * - *

Like any {@code BlockingQueue}, a {@code TransferQueue} may be - * capacity bounded. If so, an attempted {@code transfer} operation - * may initially block waiting for available space, and/or - * subsequently block waiting for reception by a consumer. Note that - * in a queue with zero capacity, such as {@link SynchronousQueue}, - * {@code put} and {@code transfer} are effectively synonymous. + *

Like other blocking queues, a {@code TransferQueue} may be + * capacity bounded. If so, an attempted transfer operation may + * initially block waiting for available space, and/or subsequently + * block waiting for reception by a consumer. Note that in a queue + * with zero capacity, such as {@link SynchronousQueue}, {@code put} + * and {@code transfer} are effectively synonymous. * *

This interface is a member of the * @@ -37,9 +41,12 @@ import java.util.concurrent.*; */ public interface TransferQueue extends BlockingQueue { /** - * Transfers the specified element if there exists a consumer - * already waiting to receive it, otherwise returning {@code false} - * without enqueuing the element. + * Transfers the element to a waiting consumer immediately, if possible. + * + *

More precisely, transfers the specified element immediately + * if there exists a consumer already waiting to receive it (in + * {@link #take} or timed {@link #poll(long,TimeUnit) poll}), + * otherwise returning {@code false} without enqueuing the element. * * @param e the element to transfer * @return {@code true} if the element was transferred, else @@ -53,13 +60,16 @@ public interface TransferQueue extends BlockingQueue { boolean tryTransfer(E e); /** - * Inserts the specified element into this queue, waiting if - * necessary for space to become available and the element to be - * dequeued by a consumer invoking {@code take} or {@code poll}. + * Transfers the element to a consumer, waiting if necessary to do so. + * + *

More precisely, transfers the specified element immediately + * if there exists a consumer already waiting to receive it (in + * {@link #take} or timed {@link #poll(long,TimeUnit) poll}), + * else waits until the element is received by a consumer. * * @param e the element to transfer * @throws InterruptedException if interrupted while waiting, - * in which case the element is not enqueued. + * in which case the element is not left enqueued * @throws ClassCastException if the class of the specified element * prevents it from being added to this queue * @throws NullPointerException if the specified element is null @@ -69,10 +79,15 @@ public interface TransferQueue extends BlockingQueue { void transfer(E e) throws InterruptedException; /** - * Inserts the specified element into this queue, waiting up to - * the specified wait time if necessary for space to become - * available and the element to be dequeued by a consumer invoking - * {@code take} or {@code poll}. + * Transfers the element to a consumer if it is possible to do so + * before the timeout elapses. + * + *

More precisely, transfers the specified element immediately + * if there exists a consumer already waiting to receive it (in + * {@link #take} or timed {@link #poll(long,TimeUnit) poll}), + * else waits until the element is received by a consumer, + * returning {@code false} if the specified wait time elapses + * before the element can be transferred. * * @param e the element to transfer * @param timeout how long to wait before giving up, in units of @@ -81,9 +96,9 @@ public interface TransferQueue extends BlockingQueue { * {@code timeout} parameter * @return {@code true} if successful, or {@code false} if * the specified waiting time elapses before completion, - * in which case the element is not enqueued. + * in which case the element is not left enqueued * @throws InterruptedException if interrupted while waiting, - * in which case the element is not enqueued. + * in which case the element is not left enqueued * @throws ClassCastException if the class of the specified element * prevents it from being added to this queue * @throws NullPointerException if the specified element is null @@ -95,7 +110,8 @@ public interface TransferQueue extends BlockingQueue { /** * Returns {@code true} if there is at least one consumer waiting - * to dequeue an element via {@code take} or {@code poll}. + * to receive an element via {@link #take} or + * timed {@link #poll(long,TimeUnit) poll}. * The return value represents a momentary state of affairs. * * @return {@code true} if there is at least one waiting consumer @@ -104,15 +120,16 @@ public interface TransferQueue extends BlockingQueue { /** * Returns an estimate of the number of consumers waiting to - * dequeue elements via {@code take} or {@code poll}. The return - * value is an approximation of a momentary state of affairs, that - * may be inaccurate if consumers have completed or given up - * waiting. The value may be useful for monitoring and heuristics, - * but not for synchronization control. Implementations of this + * receive elements via {@link #take} or timed + * {@link #poll(long,TimeUnit) poll}. The return value is an + * approximation of a momentary state of affairs, that may be + * inaccurate if consumers have completed or given up waiting. + * The value may be useful for monitoring and heuristics, but + * not for synchronization control. Implementations of this * method are likely to be noticeably slower than those for * {@link #hasWaitingConsumer}. * - * @return the number of consumers waiting to dequeue elements + * @return the number of consumers waiting to receive elements */ int getWaitingConsumerCount(); } diff --git a/src/forkjoin/scala/concurrent/forkjoin/package-info.java b/src/forkjoin/scala/concurrent/forkjoin/package-info.java index b8fa0fad02..33df96f186 100644 --- a/src/forkjoin/scala/concurrent/forkjoin/package-info.java +++ b/src/forkjoin/scala/concurrent/forkjoin/package-info.java @@ -1,29 +1,270 @@ /* * Written by Doug Lea with assistance from members of JCP JSR-166 * Expert Group and released to the public domain, as explained at - * http://creativecommons.org/licenses/publicdomain + * http://creativecommons.org/publicdomain/zero/1.0/ */ - /** - * Preview versions of classes targeted for Java 7. Includes a - * fine-grained parallel computation framework: ForkJoinTasks and - * their related support classes provide a very efficient basis for - * obtaining platform-independent parallel speed-ups of - * computation-intensive operations. They are not a full substitute - * for the kinds of arbitrary processing supported by Executors or - * Threads. However, when applicable, they typically provide - * significantly greater performance on multiprocessor platforms. - * - *

Candidates for fork/join processing mainly include those that - * can be expressed using parallel divide-and-conquer techniques: To - * solve a problem, break it in two (or more) parts, and then solve - * those parts in parallel, continuing on in this way until the - * problem is too small to be broken up, so is solved directly. The - * underlying work-stealing framework makes subtasks - * available to other threads (normally one per CPU), that help - * complete the tasks. In general, the most efficient ForkJoinTasks - * are those that directly implement this algorithmic design pattern. + * Utility classes commonly useful in concurrent programming. This + * package includes a few small standardized extensible frameworks, as + * well as some classes that provide useful functionality and are + * otherwise tedious or difficult to implement. Here are brief + * descriptions of the main components. See also the + * {@link java.util.concurrent.locks} and + * {@link java.util.concurrent.atomic} packages. + * + *

Executors

+ * + * Interfaces. + * + * {@link java.util.concurrent.Executor} is a simple standardized + * interface for defining custom thread-like subsystems, including + * thread pools, asynchronous IO, and lightweight task frameworks. + * Depending on which concrete Executor class is being used, tasks may + * execute in a newly created thread, an existing task-execution thread, + * or the thread calling {@link java.util.concurrent.Executor#execute + * execute}, and may execute sequentially or concurrently. + * + * {@link java.util.concurrent.ExecutorService} provides a more + * complete asynchronous task execution framework. An + * ExecutorService manages queuing and scheduling of tasks, + * and allows controlled shutdown. + * + * The {@link java.util.concurrent.ScheduledExecutorService} + * subinterface and associated interfaces add support for + * delayed and periodic task execution. ExecutorServices + * provide methods arranging asynchronous execution of any + * function expressed as {@link java.util.concurrent.Callable}, + * the result-bearing analog of {@link java.lang.Runnable}. + * + * A {@link java.util.concurrent.Future} returns the results of + * a function, allows determination of whether execution has + * completed, and provides a means to cancel execution. + * + * A {@link java.util.concurrent.RunnableFuture} is a {@code Future} + * that possesses a {@code run} method that upon execution, + * sets its results. + * + *

+ * + * Implementations. + * + * Classes {@link java.util.concurrent.ThreadPoolExecutor} and + * {@link java.util.concurrent.ScheduledThreadPoolExecutor} + * provide tunable, flexible thread pools. + * + * The {@link java.util.concurrent.Executors} class provides + * factory methods for the most common kinds and configurations + * of Executors, as well as a few utility methods for using + * them. Other utilities based on {@code Executors} include the + * concrete class {@link java.util.concurrent.FutureTask} + * providing a common extensible implementation of Futures, and + * {@link java.util.concurrent.ExecutorCompletionService}, that + * assists in coordinating the processing of groups of + * asynchronous tasks. + * + *

Class {@link java.util.concurrent.ForkJoinPool} provides an + * Executor primarily designed for processing instances of {@link + * java.util.concurrent.ForkJoinTask} and its subclasses. These + * classes employ a work-stealing scheduler that attains high + * throughput for tasks conforming to restrictions that often hold in + * computation-intensive parallel processing. + * + *

Queues

+ * + * The {@link java.util.concurrent.ConcurrentLinkedQueue} class + * supplies an efficient scalable thread-safe non-blocking FIFO + * queue. + * + *

Five implementations in {@code java.util.concurrent} support + * the extended {@link java.util.concurrent.BlockingQueue} + * interface, that defines blocking versions of put and take: + * {@link java.util.concurrent.LinkedBlockingQueue}, + * {@link java.util.concurrent.ArrayBlockingQueue}, + * {@link java.util.concurrent.SynchronousQueue}, + * {@link java.util.concurrent.PriorityBlockingQueue}, and + * {@link java.util.concurrent.DelayQueue}. + * The different classes cover the most common usage contexts + * for producer-consumer, messaging, parallel tasking, and + * related concurrent designs. + * + *

Extended interface {@link java.util.concurrent.TransferQueue}, + * and implementation {@link java.util.concurrent.LinkedTransferQueue} + * introduce a synchronous {@code transfer} method (along with related + * features) in which a producer may optionally block awaiting its + * consumer. + * + *

The {@link java.util.concurrent.BlockingDeque} interface + * extends {@code BlockingQueue} to support both FIFO and LIFO + * (stack-based) operations. + * Class {@link java.util.concurrent.LinkedBlockingDeque} + * provides an implementation. + * + *

Timing

+ * + * The {@link java.util.concurrent.TimeUnit} class provides + * multiple granularities (including nanoseconds) for + * specifying and controlling time-out based operations. Most + * classes in the package contain operations based on time-outs + * in addition to indefinite waits. In all cases that + * time-outs are used, the time-out specifies the minimum time + * that the method should wait before indicating that it + * timed-out. Implementations make a "best effort" + * to detect time-outs as soon as possible after they occur. + * However, an indefinite amount of time may elapse between a + * time-out being detected and a thread actually executing + * again after that time-out. All methods that accept timeout + * parameters treat values less than or equal to zero to mean + * not to wait at all. To wait "forever", you can use a value + * of {@code Long.MAX_VALUE}. + * + *

Synchronizers

+ * + * Five classes aid common special-purpose synchronization idioms. + *
    + * + *
  • {@link java.util.concurrent.Semaphore} is a classic concurrency tool. + * + *
  • {@link java.util.concurrent.CountDownLatch} is a very simple yet + * very common utility for blocking until a given number of signals, + * events, or conditions hold. + * + *
  • A {@link java.util.concurrent.CyclicBarrier} is a resettable + * multiway synchronization point useful in some styles of parallel + * programming. + * + *
  • A {@link java.util.concurrent.Phaser} provides + * a more flexible form of barrier that may be used to control phased + * computation among multiple threads. + * + *
  • An {@link java.util.concurrent.Exchanger} allows two threads to + * exchange objects at a rendezvous point, and is useful in several + * pipeline designs. + * + *
+ * + *

Concurrent Collections

+ * + * Besides Queues, this package supplies Collection implementations + * designed for use in multithreaded contexts: + * {@link java.util.concurrent.ConcurrentHashMap}, + * {@link java.util.concurrent.ConcurrentSkipListMap}, + * {@link java.util.concurrent.ConcurrentSkipListSet}, + * {@link java.util.concurrent.CopyOnWriteArrayList}, and + * {@link java.util.concurrent.CopyOnWriteArraySet}. + * When many threads are expected to access a given collection, a + * {@code ConcurrentHashMap} is normally preferable to a synchronized + * {@code HashMap}, and a {@code ConcurrentSkipListMap} is normally + * preferable to a synchronized {@code TreeMap}. + * A {@code CopyOnWriteArrayList} is preferable to a synchronized + * {@code ArrayList} when the expected number of reads and traversals + * greatly outnumber the number of updates to a list. + * + *

The "Concurrent" prefix used with some classes in this package + * is a shorthand indicating several differences from similar + * "synchronized" classes. For example {@code java.util.Hashtable} and + * {@code Collections.synchronizedMap(new HashMap())} are + * synchronized. But {@link + * java.util.concurrent.ConcurrentHashMap} is "concurrent". A + * concurrent collection is thread-safe, but not governed by a + * single exclusion lock. In the particular case of + * ConcurrentHashMap, it safely permits any number of + * concurrent reads as well as a tunable number of concurrent + * writes. "Synchronized" classes can be useful when you need + * to prevent all access to a collection via a single lock, at + * the expense of poorer scalability. In other cases in which + * multiple threads are expected to access a common collection, + * "concurrent" versions are normally preferable. And + * unsynchronized collections are preferable when either + * collections are unshared, or are accessible only when + * holding other locks. + * + *

Most concurrent Collection implementations (including most + * Queues) also differ from the usual java.util conventions in that + * their Iterators provide weakly consistent rather than + * fast-fail traversal. A weakly consistent iterator is thread-safe, + * but does not necessarily freeze the collection while iterating, so + * it may (or may not) reflect any updates since the iterator was + * created. + * + *

Memory Consistency Properties

+ * + * + * Chapter 17 of the Java Language Specification defines the + * happens-before relation on memory operations such as reads and + * writes of shared variables. The results of a write by one thread are + * guaranteed to be visible to a read by another thread only if the write + * operation happens-before the read operation. The + * {@code synchronized} and {@code volatile} constructs, as well as the + * {@code Thread.start()} and {@code Thread.join()} methods, can form + * happens-before relationships. In particular: + * + *
    + *
  • Each action in a thread happens-before every action in that + * thread that comes later in the program's order. + * + *
  • An unlock ({@code synchronized} block or method exit) of a + * monitor happens-before every subsequent lock ({@code synchronized} + * block or method entry) of that same monitor. And because + * the happens-before relation is transitive, all actions + * of a thread prior to unlocking happen-before all actions + * subsequent to any thread locking that monitor. + * + *
  • A write to a {@code volatile} field happens-before every + * subsequent read of that same field. Writes and reads of + * {@code volatile} fields have similar memory consistency effects + * as entering and exiting monitors, but do not entail + * mutual exclusion locking. + * + *
  • A call to {@code start} on a thread happens-before any + * action in the started thread. + * + *
  • All actions in a thread happen-before any other thread + * successfully returns from a {@code join} on that thread. + * + *
+ * + * + * The methods of all classes in {@code java.util.concurrent} and its + * subpackages extend these guarantees to higher-level + * synchronization. In particular: + * + *
    + * + *
  • Actions in a thread prior to placing an object into any concurrent + * collection happen-before actions subsequent to the access or + * removal of that element from the collection in another thread. + * + *
  • Actions in a thread prior to the submission of a {@code Runnable} + * to an {@code Executor} happen-before its execution begins. + * Similarly for {@code Callables} submitted to an {@code ExecutorService}. + * + *
  • Actions taken by the asynchronous computation represented by a + * {@code Future} happen-before actions subsequent to the + * retrieval of the result via {@code Future.get()} in another thread. + * + *
  • Actions prior to "releasing" synchronizer methods such as + * {@code Lock.unlock}, {@code Semaphore.release}, and + * {@code CountDownLatch.countDown} happen-before actions + * subsequent to a successful "acquiring" method such as + * {@code Lock.lock}, {@code Semaphore.acquire}, + * {@code Condition.await}, and {@code CountDownLatch.await} on the + * same synchronizer object in another thread. + * + *
  • For each pair of threads that successfully exchange objects via + * an {@code Exchanger}, actions prior to the {@code exchange()} + * in each thread happen-before those subsequent to the + * corresponding {@code exchange()} in another thread. + * + *
  • Actions prior to calling {@code CyclicBarrier.await} and + * {@code Phaser.awaitAdvance} (as well as its variants) + * happen-before actions performed by the barrier action, and + * actions performed by the barrier action happen-before actions + * subsequent to a successful return from the corresponding {@code await} + * in other threads. + * + *
* + * @since 1.5 */ package scala.concurrent.forkjoin; -- cgit v1.2.3