diff options
Diffstat (limited to 'src/forkjoin/scala/concurrent')
10 files changed, 0 insertions, 7328 deletions
diff --git a/src/forkjoin/scala/concurrent/forkjoin/ForkJoinPool.java b/src/forkjoin/scala/concurrent/forkjoin/ForkJoinPool.java deleted file mode 100644 index 6578504155..0000000000 --- a/src/forkjoin/scala/concurrent/forkjoin/ForkJoinPool.java +++ /dev/null @@ -1,3759 +0,0 @@ -/* - * 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/publicdomain/zero/1.0/ - */ - -package scala.concurrent.forkjoin; - -import java.util.ArrayList; -import java.util.Arrays; -import java.util.Collection; -import java.util.Collections; -import java.util.List; -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; - -/** - * @since 1.8 - * @author Doug Lea - */ -/*public*/ abstract class CountedCompleter<T> extends ForkJoinTask<T> { - private static final long serialVersionUID = 5232453752276485070L; - - /** This task's completer, or null if none */ - final CountedCompleter<?> completer; - /** The number of pending tasks until completion */ - volatile int pending; - - /** - * Creates a new CountedCompleter with the given completer - * and initial pending count. - * - * @param completer this task's completer, or {@code null} if none - * @param initialPendingCount the initial pending count - */ - protected CountedCompleter(CountedCompleter<?> completer, - int initialPendingCount) { - this.completer = completer; - this.pending = initialPendingCount; - } - - /** - * Creates a new CountedCompleter with the given completer - * and an initial pending count of zero. - * - * @param completer this task's completer, or {@code null} if none - */ - protected CountedCompleter(CountedCompleter<?> completer) { - this.completer = completer; - } - - /** - * Creates a new CountedCompleter with no completer - * and an initial pending count of zero. - */ - protected CountedCompleter() { - this.completer = null; - } - - /** - * The main computation performed by this task. - */ - public abstract void compute(); - - /** - * Performs an action when method {@link #tryComplete} is invoked - * and the pending count is zero, or when the unconditional - * method {@link #complete} is invoked. By default, this method - * does nothing. You can distinguish cases by checking the - * identity of the given caller argument. If not equal to {@code - * this}, then it is typically a subtask that may contain results - * (and/or links to other results) to combine. - * - * @param caller the task invoking this method (which may - * be this task itself) - */ - public void onCompletion(CountedCompleter<?> caller) { - } - - /** - * Performs an action when method {@link #completeExceptionally} - * is invoked or method {@link #compute} throws an exception, and - * this task has not otherwise already completed normally. On - * entry to this method, this task {@link - * ForkJoinTask#isCompletedAbnormally}. The return value of this - * method controls further propagation: If {@code true} and this - * task has a completer, then this completer is also completed - * exceptionally. The default implementation of this method does - * nothing except return {@code true}. - * - * @param ex the exception - * @param caller the task invoking this method (which may - * be this task itself) - * @return true if this exception should be propagated to this - * task's completer, if one exists - */ - public boolean onExceptionalCompletion(Throwable ex, CountedCompleter<?> caller) { - return true; - } - - /** - * Returns the completer established in this task's constructor, - * or {@code null} if none. - * - * @return the completer - */ - public final CountedCompleter<?> getCompleter() { - return completer; - } - - /** - * Returns the current pending count. - * - * @return the current pending count - */ - public final int getPendingCount() { - return pending; - } - - /** - * Sets the pending count to the given value. - * - * @param count the count - */ - public final void setPendingCount(int count) { - pending = count; - } - - /** - * Adds (atomically) the given value to the pending count. - * - * @param delta the value to add - */ - public final void addToPendingCount(int delta) { - int c; // note: can replace with intrinsic in jdk8 - do {} while (!U.compareAndSwapInt(this, PENDING, c = pending, c+delta)); - } - - /** - * Sets (atomically) the pending count to the given count only if - * it currently holds the given expected value. - * - * @param expected the expected value - * @param count the new value - * @return true if successful - */ - public final boolean compareAndSetPendingCount(int expected, int count) { - return U.compareAndSwapInt(this, PENDING, expected, count); - } - - /** - * If the pending count is nonzero, (atomically) decrements it. - * - * @return the initial (undecremented) pending count holding on entry - * to this method - */ - public final int decrementPendingCountUnlessZero() { - int c; - do {} while ((c = pending) != 0 && - !U.compareAndSwapInt(this, PENDING, c, c - 1)); - return c; - } - - /** - * Returns the root of the current computation; i.e., this - * task if it has no completer, else its completer's root. - * - * @return the root of the current computation - */ - public final CountedCompleter<?> getRoot() { - CountedCompleter<?> a = this, p; - while ((p = a.completer) != null) - a = p; - return a; - } - - /** - * If the pending count is nonzero, decrements the count; - * otherwise invokes {@link #onCompletion} and then similarly - * tries to complete this task's completer, if one exists, - * else marks this task as complete. - */ - public final void tryComplete() { - CountedCompleter<?> a = this, s = a; - for (int c;;) { - if ((c = a.pending) == 0) { - a.onCompletion(s); - if ((a = (s = a).completer) == null) { - s.quietlyComplete(); - return; - } - } - else if (U.compareAndSwapInt(a, PENDING, c, c - 1)) - return; - } - } - - /** - * Equivalent to {@link #tryComplete} but does not invoke {@link - * #onCompletion} along the completion path: If the pending count - * is nonzero, decrements the count; otherwise, similarly tries to - * complete this task's completer, if one exists, else marks this - * task as complete. This method may be useful in cases where - * {@code onCompletion} should not, or need not, be invoked for - * each completer in a computation. - */ - public final void propagateCompletion() { - CountedCompleter<?> a = this, s = a; - for (int c;;) { - if ((c = a.pending) == 0) { - if ((a = (s = a).completer) == null) { - s.quietlyComplete(); - return; - } - } - else if (U.compareAndSwapInt(a, PENDING, c, c - 1)) - return; - } - } - - /** - * Regardless of pending count, invokes {@link #onCompletion}, - * marks this task as complete and further triggers {@link - * #tryComplete} on this task's completer, if one exists. The - * given rawResult is used as an argument to {@link #setRawResult} - * before invoking {@link #onCompletion} or marking this task as - * complete; its value is meaningful only for classes overriding - * {@code setRawResult}. - * - * <p>This method may be useful when forcing completion as soon as - * any one (versus all) of several subtask results are obtained. - * However, in the common (and recommended) case in which {@code - * setRawResult} is not overridden, this effect can be obtained - * more simply using {@code quietlyCompleteRoot();}. - * - * @param rawResult the raw result - */ - public void complete(T rawResult) { - CountedCompleter<?> p; - setRawResult(rawResult); - onCompletion(this); - quietlyComplete(); - if ((p = completer) != null) - p.tryComplete(); - } - - - /** - * If this task's pending count is zero, returns this task; - * otherwise decrements its pending count and returns {@code - * null}. This method is designed to be used with {@link - * #nextComplete} in completion traversal loops. - * - * @return this task, if pending count was zero, else {@code null} - */ - public final CountedCompleter<?> firstComplete() { - for (int c;;) { - if ((c = pending) == 0) - return this; - else if (U.compareAndSwapInt(this, PENDING, c, c - 1)) - return null; - } - } - - /** - * If this task does not have a completer, invokes {@link - * ForkJoinTask#quietlyComplete} and returns {@code null}. Or, if - * this task's pending count is non-zero, decrements its pending - * count and returns {@code null}. Otherwise, returns the - * completer. This method can be used as part of a completion - * traversal loop for homogeneous task hierarchies: - * - * <pre> {@code - * for (CountedCompleter<?> c = firstComplete(); - * c != null; - * c = c.nextComplete()) { - * // ... process c ... - * }}</pre> - * - * @return the completer, or {@code null} if none - */ - public final CountedCompleter<?> nextComplete() { - CountedCompleter<?> p; - if ((p = completer) != null) - return p.firstComplete(); - else { - quietlyComplete(); - return null; - } - } - - /** - * Equivalent to {@code getRoot().quietlyComplete()}. - */ - public final void quietlyCompleteRoot() { - for (CountedCompleter<?> a = this, p;;) { - if ((p = a.completer) == null) { - a.quietlyComplete(); - return; - } - a = p; - } - } - - /** - * Supports ForkJoinTask exception propagation. - */ - void internalPropagateException(Throwable ex) { - CountedCompleter<?> a = this, s = a; - while (a.onExceptionalCompletion(ex, s) && - (a = (s = a).completer) != null && a.status >= 0) - a.recordExceptionalCompletion(ex); - } - - /** - * Implements execution conventions for CountedCompleters. - */ - protected final boolean exec() { - compute(); - return false; - } - - /** - * Returns the result of the computation. By default - * returns {@code null}, which is appropriate for {@code Void} - * actions, but in other cases should be overridden, almost - * always to return a field or function of a field that - * holds the result upon completion. - * - * @return the result of the computation - */ - public T getRawResult() { return null; } - - /** - * A method that result-bearing CountedCompleters may optionally - * use to help maintain result data. By default, does nothing. - * Overrides are not recommended. However, if this method is - * overridden to update existing objects or fields, then it must - * in general be defined to be thread-safe. - */ - protected void setRawResult(T t) { } - - // Unsafe mechanics - private static final sun.misc.Unsafe U; - private static final long PENDING; - static { - try { - U = getUnsafe(); - PENDING = U.objectFieldOffset - (CountedCompleter.class.getDeclaredField("pending")); - } catch (Exception e) { - throw new Error(e); - } - } - - /** - * Returns a sun.misc.Unsafe. Suitable for use in a 3rd party package. - * Replace with a simple call to Unsafe.getUnsafe when integrating - * into a jdk. - * - * @return a sun.misc.Unsafe - */ - private static sun.misc.Unsafe getUnsafe() { - return scala.concurrent.util.Unsafe.instance; - } -} - -/** - * 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. - * - * <p>A {@code ForkJoinPool} differs from other kinds of {@link - * ExecutorService} mainly by virtue of employing - * <em>work-stealing</em>: all threads in the pool attempt to find and - * execute tasks submitted to the pool and/or 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), as well as when many small - * tasks are submitted to the pool from external clients. Especially - * when setting <em>asyncMode</em> to true in constructors, {@code - * ForkJoinPool}s may also be appropriate for use with event-style - * tasks that are never joined. - * - * <p>A static {@link #commonPool()} is available and appropriate for - * most applications. The common pool is used by any ForkJoinTask that - * is not explicitly submitted to a specified pool. Using the common - * pool normally reduces resource usage (its threads are slowly - * reclaimed during periods of non-use, and reinstated upon subsequent - * use). - * - * <p>For applications that require separate or custom pools, a {@code - * ForkJoinPool} may be constructed with a given target parallelism - * level; by default, equal to the number of available processors. The - * pool attempts to maintain enough active (or available) threads by - * dynamically adding, suspending, or resuming internal worker - * threads, even if some tasks are stalled waiting to join - * others. However, no such adjustments are guaranteed in the face of - * blocked I/O or other unmanaged synchronization. The nested {@link - * ManagedBlocker} interface enables extension of the kinds of - * synchronization accommodated. - * - * <p>In addition to execution and lifecycle control methods, this - * class provides status check methods (for example - * {@link #getStealCount}) that are intended to aid in developing, - * tuning, and monitoring fork/join applications. Also, method - * {@link #toString} returns indications of pool state in a - * convenient form for informal monitoring. - * - * <p>As is the case with other ExecutorServices, there are three - * main task execution methods summarized in the following table. - * These are designed to be used primarily by clients not already - * engaged in fork/join computations in the current pool. The main - * 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 instead - * use the within-computation forms listed in the table unless using - * async event-style tasks that are not usually joined, in which case - * there is little difference among choice of methods. - * - * <table BORDER CELLPADDING=3 CELLSPACING=1> - * <tr> - * <td></td> - * <td ALIGN=CENTER> <b>Call from non-fork/join clients</b></td> - * <td ALIGN=CENTER> <b>Call from within fork/join computations</b></td> - * </tr> - * <tr> - * <td> <b>Arrange async execution</td> - * <td> {@link #execute(ForkJoinTask)}</td> - * <td> {@link ForkJoinTask#fork}</td> - * </tr> - * <tr> - * <td> <b>Await and obtain result</td> - * <td> {@link #invoke(ForkJoinTask)}</td> - * <td> {@link ForkJoinTask#invoke}</td> - * </tr> - * <tr> - * <td> <b>Arrange exec and obtain Future</td> - * <td> {@link #submit(ForkJoinTask)}</td> - * <td> {@link ForkJoinTask#fork} (ForkJoinTasks <em>are</em> Futures)</td> - * </tr> - * </table> - * - * <p>The common pool is by default constructed with default - * parameters, but these may be controlled by setting three {@link - * System#getProperty system properties} with prefix {@code - * java.util.concurrent.ForkJoinPool.common}: {@code parallelism} -- - * an integer greater than zero, {@code threadFactory} -- the class - * name of a {@link ForkJoinWorkerThreadFactory}, and {@code - * exceptionHandler} -- the class name of a {@link - * java.lang.Thread.UncaughtExceptionHandler - * Thread.UncaughtExceptionHandler}. Upon any error in establishing - * these settings, default parameters are used. - * - * <p><b>Implementation notes</b>: This implementation restricts the - * maximum number of running threads to 32767. Attempts to create - * pools with greater than the maximum number result in - * {@code IllegalArgumentException}. - * - * <p>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 { - - /* - * Implementation Overview - * - * This class and its nested classes provide the main - * functionality and control for a set of worker threads: - * Submissions from non-FJ threads enter into submission queues. - * 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 queues. - * - * WorkQueues - * ========== - * - * Most operations occur within work-stealing queues (in nested - * class WorkQueue). These are special forms of Deques that - * support only three of the four possible end-operations -- push, - * pop, and poll (aka steal), under the further constraints that - * push and pop are called only from the owning thread (or, as - * extended here, under a lock), while poll 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 poll (steal) from being on the indices - * ("base" and "top") to the slots themselves. So, both a - * successful pop and poll mainly entail a CAS of a slot from - * non-null to null. Because we rely on CASes of references, we - * do not need tag bits on base or top. They are simple ints as - * used in any circular array-based queue (see for example - * ArrayDeque). Updates to the indices must still be ordered in a - * way that guarantees that top == base means the queue is empty, - * but otherwise may err on the side of possibly making the queue - * appear nonempty when a push, pop, or poll have not fully - * committed. Note that this means that the poll operation, - * considered individually, is not wait-free. One thief cannot - * successfully continue until another in-progress one (or, if - * previously empty, a push) completes. However, in the - * aggregate, we ensure at least probabilistic non-blockingness. - * If an attempted steal fails, a thief always chooses a different - * random victim target to try next. So, in order for one thief to - * progress, it suffices for any in-progress poll or new push on - * any empty queue to complete. (This is why we normally use - * method pollAt and its variants that try once at the apparent - * base index, else consider alternative actions, rather than - * method poll.) - * - * This approach also enables support of a user mode in which local - * task processing is in FIFO, not LIFO order, simply by using - * poll rather than pop. This can be useful 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.) - * - * WorkQueues are also used in a similar way for tasks submitted - * to the pool. We cannot mix these tasks in the same queues used - * for work-stealing (this would contaminate lifo/fifo - * processing). Instead, we randomly associate submission queues - * with submitting threads, using a form of hashing. The - * ThreadLocal Submitter class contains a value initially used as - * a hash code for choosing existing queues, but may be randomly - * repositioned upon contention with other submitters. In - * essence, submitters act like workers except that they are - * restricted to executing local tasks that they submitted (or in - * the case of CountedCompleters, others with the same root task). - * However, because most shared/external queue operations are more - * expensive than internal, and because, at steady state, external - * submitters will compete for CPU with workers, ForkJoinTask.join - * and related methods disable them from repeatedly helping to - * process tasks if all workers are active. Insertion of tasks in - * shared mode requires a lock (mainly to protect in the case of - * resizing) but we use only a simple spinlock (using bits in - * field qlock), because submitters encountering a busy queue move - * on to try or create other queues -- they block only when - * creating and registering new queues. - * - * Management - * ========== - * - * 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 two volatile variables - * that are by far most often read (not written) as status and - * consistency checks. - * - * Field "ctl" contains 64 bits holding all the information needed - * to atomically decide to add, inactivate, enqueue (on an event - * queue), dequeue, and/or re-activate workers. To enable this - * 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. - * - * Field "plock" is a form of sequence lock with a saturating - * shutdown bit (similarly for per-queue "qlocks"), mainly - * protecting updates to the workQueues array, as well as to - * enable shutdown. When used as a lock, it is normally only very - * briefly held, so is nearly always available after at most a - * brief spin, but we use a monitor-based backup strategy to - * block when needed. - * - * Recording WorkQueues. WorkQueues are recorded in the - * "workQueues" array that is created upon first use and expanded - * if necessary. Updates to the array while recording new workers - * and unrecording terminated ones are protected from each other - * by a lock but the array is otherwise concurrently readable, and - * accessed directly. To simplify index-based operations, the - * array size is always a power of two, and all readers must - * tolerate null slots. Worker queues are at odd indices. Shared - * (submission) queues are at even indices, up to a maximum of 64 - * slots, to limit growth even if array needs to expand to add - * more workers. Grouping them together in this way simplifies and - * speeds up task scanning. - * - * All worker thread creation is on-demand, triggered by task - * submissions, replacement of terminated workers, and/or - * 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 workQueues are via indices into the workQueues - * array (which is one source of some of the messy code - * constructions here). In essence, the workQueues array serves as - * a weak reference mechanism. Thus for example the wait queue - * field of ctl stores indices, not references. Access to the - * workQueues 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 workQueues array - * also check that it 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. It also helps - * reduce JIT issuance of uncommon-trap code, which tends to - * unnecessarily complicate control flow in some methods. - * - * Event 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. In many usages, ramp-up time to activate workers is - * the main limiting factor in overall performance (this is - * compounded at program start-up by JIT compilation and - * allocation). So we try to streamline this as much as possible. - * 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 (that reflects the number of times a worker has - * been inactivated) to avoid ABA effects (we need only as many - * version numbers as worker threads). Successors are held in - * field WorkQueue.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 (via repeated calls to method - * scan()) both before and after a newly waiting worker is added - * to the wait queue. During a rescan, the worker might release - * some other queued worker rather than itself, which has the same - * net effect. Because enqueued workers may actually be rescanning - * rather than waiting, we set and clear the "parker" field of - * WorkQueues to reduce unnecessary calls to unpark. (This - * requires a secondary recheck to avoid missed signals.) Note - * the unusual conventions about Thread.interrupts surrounding - * parking and other blocking: Because interrupts are used solely - * to alert threads to check termination, which is checked anyway - * upon blocking, we clear status (using Thread.interrupted) - * before any call to park, so that park does not immediately - * return due to status being set via some other unrelated call to - * interrupt in user code. - * - * 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. However, many other threads may notice the same task - * and each signal to wake up a thread that might take it. So in - * general, pools will be over-signalled. When a submission is - * added or another worker adds a task to a queue that has fewer - * than two tasks, they signal waiting workers (or trigger - * creation of new ones if fewer than the given parallelism level - * -- signalWork), and may leave a hint to the unparked worker to - * help signal others upon wakeup). These primary signals are - * buttressed by others (see method helpSignal) whenever other - * threads scan for work or do not have a task to process. On - * most platforms, signalling (unpark) overhead time is noticeably - * long, and the time between signalling a thread and it actually - * making progress can be very noticeably long, so it is worth - * offloading these delays from critical paths as much as - * possible. - * - * 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 a - * given period -- a short period if there are more threads than - * parallelism, longer as the number of threads decreases. This - * will slowly propagate, eventually terminating all workers after - * periods of non-use. - * - * Shutdown and Termination. A call to shutdownNow atomically sets - * a plock bit and then (non-atomically) sets each worker's - * qlock status, cancels all unprocessed tasks, and wakes up - * all waiting workers. Detecting whether termination should - * commence after a non-abrupt shutdown() call requires more work - * and bookkeeping. We need consensus about quiescence (i.e., that - * there is no more work). The active count provides a primary - * indication but non-abrupt shutdown still requires a rechecking - * scan for any workers that are inactive but not queued. - * - * Joining Tasks - * ============= - * - * Any of several actions may be taken 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. - * - * Compensating: Unless there are already enough live threads, - * method tryCompensate() may create or re-activate a spare - * thread to compensate for blocked joiners until they unblock. - * - * A third form (implemented in tryRemoveAndExec) amounts to - * helping a hypothetical compensator: If we can readily tell that - * a possible action of a compensator is to steal and execute the - * task being joined, the joining thread can do so directly, - * without the need for a compensation thread (although at the - * expense of larger run-time stacks, but the tradeoff is - * typically worthwhile). - * - * The ManagedBlocker extension API can't use helping so relies - * only on compensation in method awaitBlocker. - * - * The algorithm in tryHelpStealer entails 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 tryHelpStealer 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 sometimes - * requires a linear scan of workQueues array to locate stealers, - * but often doesn't because stealers leave hints (that may become - * stale/wrong) of where to locate them. It is only a hint - * because a worker might have had multiple steals and the hint - * records only one of them (usually the most current). Hinting - * isolates cost to when it is needed, rather than adding to - * per-task overhead. (2) It is "shallow", ignoring nesting and - * potentially cyclic mutual steals. (3) It is intentionally - * racy: field currentJoin is updated only while actively joining, - * which means that we miss links in the chain during long-lived - * tasks, GC stalls etc (which is OK since blocking in such cases - * 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. - * - * Helping actions for CountedCompleters are much simpler: Method - * helpComplete can take and execute any task with the same root - * as the task being waited on. However, this still entails some - * traversal of completer chains, so is less efficient than using - * CountedCompleters without explicit joins. - * - * It is impossible to keep exactly the target parallelism number - * of threads running at any given time. Determining the - * existence of conservatively safe helping targets, the - * availability of already-created spares, and the apparent need - * to create new spares are all racy, so we rely on multiple - * retries of each. Compensation in the apparent absence of - * helping opportunities is challenging to control on JVMs, where - * GC and other activities can stall progress of tasks that in - * turn stall out many other dependent tasks, without us being - * able to determine whether they will ever require compensation. - * Even though work-stealing otherwise encounters little - * degradation in the presence of more threads than cores, - * aggressively adding new threads in such cases entails risk of - * unwanted positive feedback control loops in which more threads - * cause more dependent stalls (as well as delayed progress of - * unblocked threads to the point that we know they are available) - * leading to more situations requiring more threads, and so - * on. This aspect of control can be seen as an (analytically - * intractable) game with an opponent that may choose the worst - * (for us) active thread to stall at any time. We take several - * precautions to bound losses (and thus bound gains), mainly in - * methods tryCompensate and awaitJoin. - * - * Common Pool - * =========== - * - * The static common Pool always exists after static - * initialization. Since it (or any other created pool) need - * never be used, we minimize initial construction overhead and - * footprint to the setup of about a dozen fields, with no nested - * allocation. Most bootstrapping occurs within method - * fullExternalPush during the first submission to the pool. - * - * When external threads submit to the common pool, they can - * perform some subtask processing (see externalHelpJoin and - * related methods). We do not need to record whether these - * submissions are to the common pool -- if not, externalHelpJoin - * returns quickly (at the most helping to signal some common pool - * workers). These submitters would otherwise be blocked waiting - * for completion, so the extra effort (with liberally sprinkled - * task status checks) in inapplicable cases amounts to an odd - * form of limited spin-wait before blocking in ForkJoinTask.join. - * - * Style notes - * =========== - * - * There is a lot of representation-level coupling among classes - * ForkJoinPool, ForkJoinWorkerThread, and ForkJoinTask. The - * fields of WorkQueue maintain data structures managed by - * ForkJoinPool, so are directly accessed. There is little point - * trying to reduce this, since any associated future changes in - * representations will need to be accompanied by algorithmic - * changes anyway. Several methods intrinsically sprawl because - * they must accumulate sets of consistent reads of volatiles held - * in local variables. Methods signalWork() and scan() are the - * main bottlenecks, so are especially heavily - * micro-optimized/mangled. There are lots of inline assignments - * (of form "while ((local = field) != 0)") which are usually the - * simplest way to ensure the required read orderings (which are - * 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 (including - * several unnecessary-looking hoisted null checks) that help - * some methods perform reasonably even when interpreted (not - * compiled). - * - * The order of declarations in this file is: - * (1) Static utility functions - * (2) Nested (static) classes - * (3) Static fields - * (4) Fields, along with constants used when unpacking some of them - * (5) Internal control methods - * (6) Callbacks and other support for ForkJoinTask methods - * (7) Exported methods - * (8) Static block initializing statics in minimally dependent order - */ - - // Static utilities - - /** - * If there is a security manager, makes sure caller has - * permission to modify threads. - */ - private static void checkPermission() { - SecurityManager security = System.getSecurityManager(); - if (security != null) - security.checkPermission(modifyThreadPermission); - } - - // Nested classes - - /** - * 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 the pool is null - */ - public ForkJoinWorkerThread newThread(ForkJoinPool pool); - } - - /** - * Default ForkJoinWorkerThreadFactory implementation; creates a - * new ForkJoinWorkerThread. - */ - static final class DefaultForkJoinWorkerThreadFactory - implements ForkJoinWorkerThreadFactory { - public final ForkJoinWorkerThread newThread(ForkJoinPool pool) { - return new ForkJoinWorkerThread(pool); - } - } - - /** - * Per-thread records for threads that submit to pools. Currently - * holds only pseudo-random seed / index that is used to choose - * submission queues in method externalPush. In the future, this may - * also incorporate a means to implement different task rejection - * and resubmission policies. - * - * Seeds for submitters and workers/workQueues work in basically - * the same way but are initialized and updated using slightly - * different mechanics. Both are initialized using the same - * approach as in class ThreadLocal, where successive values are - * unlikely to collide with previous values. Seeds are then - * randomly modified upon collisions using xorshifts, which - * requires a non-zero seed. - */ - static final class Submitter { - int seed; - Submitter(int s) { seed = s; } - } - - /** - * Class for artificial tasks that are used to replace the target - * of local joins if they are removed from an interior queue slot - * in WorkQueue.tryRemoveAndExec. We don't need the proxy to - * actually do anything beyond having a unique identity. - */ - static final class EmptyTask extends ForkJoinTask<Void> { - private static final long serialVersionUID = -7721805057305804111L; - EmptyTask() { status = ForkJoinTask.NORMAL; } // force done - public final Void getRawResult() { return null; } - public final void setRawResult(Void x) {} - public final boolean exec() { return true; } - } - - /** - * Queues supporting work-stealing as well as external task - * submission. See above for main rationale and algorithms. - * Implementation relies heavily on "Unsafe" intrinsics - * and selective use of "volatile": - * - * Field "base" is the index (mod array.length) of the least valid - * queue slot, which is always the next position to steal (poll) - * from if nonempty. Reads and writes require volatile orderings - * but not CAS, because updates are only performed after slot - * CASes. - * - * Field "top" is the index (mod array.length) of the next queue - * slot to push to or pop from. It is written only by owner thread - * for push, or under lock for external/shared push, and accessed - * by other threads only after reading (volatile) base. Both top - * and base are allowed to wrap around on overflow, but (top - - * base) (or more commonly -(base - top) to force volatile read of - * base before top) still estimates size. The lock ("qlock") is - * forced to -1 on termination, causing all further lock attempts - * to fail. (Note: we don't need CAS for termination state because - * upon pool shutdown, all shared-queues will stop being used - * anyway.) Nearly all lock bodies are set up so that exceptions - * within lock bodies are "impossible" (modulo JVM errors that - * would cause failure anyway.) - * - * The array slots are read and written using the emulation of - * volatiles/atomics provided by Unsafe. Insertions must in - * general use putOrderedObject as a form of releasing store to - * ensure that all writes to the task object are ordered before - * its publication in the queue. All removals entail a CAS to - * null. The array is always a power of two. To ensure safety of - * Unsafe array operations, all accesses perform explicit null - * checks and implicit bounds checks via power-of-two masking. - * - * In addition to basic queuing support, this class contains - * fields described elsewhere to control execution. It turns out - * to work better memory-layout-wise to include them in this class - * rather than a separate class. - * - * Performance on most platforms is very sensitive to placement of - * instances of both WorkQueues and their arrays -- we absolutely - * do not want multiple WorkQueue instances or multiple queue - * arrays sharing cache lines. (It would be best for queue objects - * and their arrays to share, but there is nothing available to - * help arrange that). Unfortunately, because they are recorded - * in a common array, WorkQueue instances are often moved to be - * adjacent by garbage collectors. To reduce impact, we use field - * padding that works OK on common platforms; this effectively - * trades off slightly slower average field access for the sake of - * avoiding really bad worst-case access. (Until better JVM - * support is in place, this padding is dependent on transient - * properties of JVM field layout rules.) We also take care in - * allocating, sizing and resizing the array. Non-shared queue - * arrays are initialized by workers before use. Others are - * allocated on first use. - */ - static final class WorkQueue { - /** - * Capacity of work-stealing queue array upon initialization. - * Must be a power of two; at least 4, but should be larger to - * reduce or eliminate cacheline sharing among queues. - * Currently, it is much larger, as a partial workaround for - * the fact that JVMs often place arrays in locations that - * share GC bookkeeping (especially cardmarks) such that - * per-write accesses encounter serious memory contention. - */ - static final int INITIAL_QUEUE_CAPACITY = 1 << 13; - - /** - * Maximum size for queue arrays. Must be a power of two less - * than or equal to 1 << (31 - width of array entry) to ensure - * lack of wraparound of index calculations, but defined to a - * value a bit less than this to help users trap runaway - * programs before saturating systems. - */ - static final int MAXIMUM_QUEUE_CAPACITY = 1 << 26; // 64M - - // Heuristic padding to ameliorate unfortunate memory placements - volatile long pad00, pad01, pad02, pad03, pad04, pad05, pad06; - - int seed; // for random scanning; initialize nonzero - volatile int eventCount; // encoded inactivation count; < 0 if inactive - int nextWait; // encoded record of next event waiter - int hint; // steal or signal hint (index) - int poolIndex; // index of this queue in pool (or 0) - final int mode; // 0: lifo, > 0: fifo, < 0: shared - int nsteals; // number of steals - volatile int qlock; // 1: locked, -1: terminate; else 0 - volatile int base; // index of next slot for poll - int top; // index of next slot for push - ForkJoinTask<?>[] array; // the elements (initially unallocated) - final ForkJoinPool pool; // the containing pool (may be null) - final ForkJoinWorkerThread owner; // owning thread or null if shared - volatile Thread parker; // == owner during call to park; else null - volatile ForkJoinTask<?> currentJoin; // task being joined in awaitJoin - ForkJoinTask<?> currentSteal; // current non-local task being executed - - volatile Object pad10, pad11, pad12, pad13, pad14, pad15, pad16, pad17; - volatile Object pad18, pad19, pad1a, pad1b, pad1c, pad1d; - - WorkQueue(ForkJoinPool pool, ForkJoinWorkerThread owner, int mode, - int seed) { - this.pool = pool; - this.owner = owner; - this.mode = mode; - this.seed = seed; - // Place indices in the center of array (that is not yet allocated) - base = top = INITIAL_QUEUE_CAPACITY >>> 1; - } - - /** - * Returns the approximate number of tasks in the queue. - */ - final int queueSize() { - int n = base - top; // non-owner callers must read base first - return (n >= 0) ? 0 : -n; // ignore transient negative - } - - /** - * Provides a more accurate estimate of whether this queue has - * any tasks than does queueSize, by checking whether a - * near-empty queue has at least one unclaimed task. - */ - final boolean isEmpty() { - ForkJoinTask<?>[] a; int m, s; - int n = base - (s = top); - return (n >= 0 || - (n == -1 && - ((a = array) == null || - (m = a.length - 1) < 0 || - U.getObject - (a, (long)((m & (s - 1)) << ASHIFT) + ABASE) == null))); - } - - /** - * Pushes a task. Call only by owner in unshared queues. (The - * shared-queue version is embedded in method externalPush.) - * - * @param task the task. Caller must ensure non-null. - * @throws RejectedExecutionException if array cannot be resized - */ - final void push(ForkJoinTask<?> task) { - ForkJoinTask<?>[] a; ForkJoinPool p; - int s = top, m, n; - if ((a = array) != null) { // ignore if queue removed - int j = (((m = a.length - 1) & s) << ASHIFT) + ABASE; - U.putOrderedObject(a, j, task); - if ((n = (top = s + 1) - base) <= 2) { - if ((p = pool) != null) - p.signalWork(this); - } - else if (n >= m) - growArray(); - } - } - - /** - * Initializes or doubles the capacity of array. Call either - * by owner or with lock held -- it is OK for base, but not - * top, to move while resizings are in progress. - */ - final ForkJoinTask<?>[] growArray() { - ForkJoinTask<?>[] oldA = array; - int size = oldA != null ? oldA.length << 1 : INITIAL_QUEUE_CAPACITY; - if (size > MAXIMUM_QUEUE_CAPACITY) - throw new RejectedExecutionException("Queue capacity exceeded"); - int oldMask, t, b; - ForkJoinTask<?>[] a = array = new ForkJoinTask<?>[size]; - if (oldA != null && (oldMask = oldA.length - 1) >= 0 && - (t = top) - (b = base) > 0) { - int mask = size - 1; - do { - ForkJoinTask<?> x; - int oldj = ((b & oldMask) << ASHIFT) + ABASE; - int j = ((b & mask) << ASHIFT) + ABASE; - x = (ForkJoinTask<?>)U.getObjectVolatile(oldA, oldj); - if (x != null && - U.compareAndSwapObject(oldA, oldj, x, null)) - U.putObjectVolatile(a, j, x); - } while (++b != t); - } - return a; - } - - /** - * Takes next task, if one exists, in LIFO order. Call only - * by owner in unshared queues. - */ - final ForkJoinTask<?> pop() { - ForkJoinTask<?>[] a; ForkJoinTask<?> t; int m; - if ((a = array) != null && (m = a.length - 1) >= 0) { - for (int s; (s = top - 1) - base >= 0;) { - long j = ((m & s) << ASHIFT) + ABASE; - if ((t = (ForkJoinTask<?>)U.getObject(a, j)) == null) - break; - if (U.compareAndSwapObject(a, j, t, null)) { - top = s; - return t; - } - } - } - return null; - } - - /** - * Takes a task in FIFO order if b is base of queue and a task - * can be claimed without contention. Specialized versions - * appear in ForkJoinPool methods scan and tryHelpStealer. - */ - final ForkJoinTask<?> pollAt(int b) { - ForkJoinTask<?> t; ForkJoinTask<?>[] a; - if ((a = array) != null) { - int j = (((a.length - 1) & b) << ASHIFT) + ABASE; - if ((t = (ForkJoinTask<?>)U.getObjectVolatile(a, j)) != null && - base == b && - U.compareAndSwapObject(a, j, t, null)) { - base = b + 1; - return t; - } - } - return null; - } - - /** - * Takes next task, if one exists, in FIFO order. - */ - final ForkJoinTask<?> poll() { - ForkJoinTask<?>[] a; int b; ForkJoinTask<?> t; - while ((b = base) - top < 0 && (a = array) != null) { - int j = (((a.length - 1) & b) << ASHIFT) + ABASE; - t = (ForkJoinTask<?>)U.getObjectVolatile(a, j); - if (t != null) { - if (base == b && - U.compareAndSwapObject(a, j, t, null)) { - base = b + 1; - return t; - } - } - else if (base == b) { - if (b + 1 == top) - break; - Thread.yield(); // wait for lagging update (very rare) - } - } - return null; - } - - /** - * Takes next task, if one exists, in order specified by mode. - */ - final ForkJoinTask<?> nextLocalTask() { - return mode == 0 ? pop() : poll(); - } - - /** - * Returns next task, if one exists, in order specified by mode. - */ - final ForkJoinTask<?> peek() { - ForkJoinTask<?>[] a = array; int m; - if (a == null || (m = a.length - 1) < 0) - return null; - int i = mode == 0 ? top - 1 : base; - int j = ((i & m) << ASHIFT) + ABASE; - return (ForkJoinTask<?>)U.getObjectVolatile(a, j); - } - - /** - * Pops the given task only if it is at the current top. - * (A shared version is available only via FJP.tryExternalUnpush) - */ - final boolean tryUnpush(ForkJoinTask<?> t) { - ForkJoinTask<?>[] a; int s; - if ((a = array) != null && (s = top) != base && - U.compareAndSwapObject - (a, (((a.length - 1) & --s) << ASHIFT) + ABASE, t, null)) { - top = s; - return true; - } - return false; - } - - /** - * Removes and cancels all known tasks, ignoring any exceptions. - */ - final void cancelAll() { - ForkJoinTask.cancelIgnoringExceptions(currentJoin); - ForkJoinTask.cancelIgnoringExceptions(currentSteal); - for (ForkJoinTask<?> t; (t = poll()) != null; ) - ForkJoinTask.cancelIgnoringExceptions(t); - } - - /** - * Computes next value for random probes. Scans don't require - * a very high quality generator, but also not a crummy one. - * Marsaglia xor-shift is cheap and works well enough. Note: - * This is manually inlined in its usages in ForkJoinPool to - * avoid writes inside busy scan loops. - */ - final int nextSeed() { - int r = seed; - r ^= r << 13; - r ^= r >>> 17; - return seed = r ^= r << 5; - } - - // Specialized execution methods - - /** - * Pops and runs tasks until empty. - */ - private void popAndExecAll() { - // A bit faster than repeated pop calls - ForkJoinTask<?>[] a; int m, s; long j; ForkJoinTask<?> t; - while ((a = array) != null && (m = a.length - 1) >= 0 && - (s = top - 1) - base >= 0 && - (t = ((ForkJoinTask<?>) - U.getObject(a, j = ((m & s) << ASHIFT) + ABASE))) - != null) { - if (U.compareAndSwapObject(a, j, t, null)) { - top = s; - t.doExec(); - } - } - } - - /** - * Polls and runs tasks until empty. - */ - private void pollAndExecAll() { - for (ForkJoinTask<?> t; (t = poll()) != null;) - t.doExec(); - } - - /** - * If present, removes from queue and executes the given task, - * or any other cancelled task. Returns (true) on any CAS - * or consistency check failure so caller can retry. - * - * @return false if no progress can be made, else true - */ - final boolean tryRemoveAndExec(ForkJoinTask<?> task) { - boolean stat = true, removed = false, empty = true; - ForkJoinTask<?>[] a; int m, s, b, n; - if ((a = array) != null && (m = a.length - 1) >= 0 && - (n = (s = top) - (b = base)) > 0) { - for (ForkJoinTask<?> t;;) { // traverse from s to b - int j = ((--s & m) << ASHIFT) + ABASE; - t = (ForkJoinTask<?>)U.getObjectVolatile(a, j); - if (t == null) // inconsistent length - break; - else if (t == task) { - if (s + 1 == top) { // pop - if (!U.compareAndSwapObject(a, j, task, null)) - break; - top = s; - removed = true; - } - else if (base == b) // replace with proxy - removed = U.compareAndSwapObject(a, j, task, - new EmptyTask()); - break; - } - else if (t.status >= 0) - empty = false; - else if (s + 1 == top) { // pop and throw away - if (U.compareAndSwapObject(a, j, t, null)) - top = s; - break; - } - if (--n == 0) { - if (!empty && base == b) - stat = false; - break; - } - } - } - if (removed) - task.doExec(); - return stat; - } - - /** - * Polls for and executes the given task or any other task in - * its CountedCompleter computation. - */ - final boolean pollAndExecCC(ForkJoinTask<?> root) { - ForkJoinTask<?>[] a; int b; Object o; - outer: while ((b = base) - top < 0 && (a = array) != null) { - long j = (((a.length - 1) & b) << ASHIFT) + ABASE; - if ((o = U.getObject(a, j)) == null || - !(o instanceof CountedCompleter)) - break; - for (CountedCompleter<?> t = (CountedCompleter<?>)o, r = t;;) { - if (r == root) { - if (base == b && - U.compareAndSwapObject(a, j, t, null)) { - base = b + 1; - t.doExec(); - return true; - } - else - break; // restart - } - if ((r = r.completer) == null) - break outer; // not part of root computation - } - } - return false; - } - - /** - * Executes a top-level task and any local tasks remaining - * after execution. - */ - final void runTask(ForkJoinTask<?> t) { - if (t != null) { - (currentSteal = t).doExec(); - currentSteal = null; - ++nsteals; - if (base - top < 0) { // process remaining local tasks - if (mode == 0) - popAndExecAll(); - else - pollAndExecAll(); - } - } - } - - /** - * Executes a non-top-level (stolen) task. - */ - final void runSubtask(ForkJoinTask<?> t) { - if (t != null) { - ForkJoinTask<?> ps = currentSteal; - (currentSteal = t).doExec(); - currentSteal = ps; - } - } - - /** - * Returns true if owned and not known to be blocked. - */ - final boolean isApparentlyUnblocked() { - Thread wt; Thread.State s; - return (eventCount >= 0 && - (wt = owner) != null && - (s = wt.getState()) != Thread.State.BLOCKED && - s != Thread.State.WAITING && - s != Thread.State.TIMED_WAITING); - } - - // Unsafe mechanics - private static final sun.misc.Unsafe U; - private static final long QLOCK; - private static final int ABASE; - private static final int ASHIFT; - static { - try { - U = getUnsafe(); - Class<?> k = WorkQueue.class; - Class<?> ak = ForkJoinTask[].class; - QLOCK = U.objectFieldOffset - (k.getDeclaredField("qlock")); - ABASE = U.arrayBaseOffset(ak); - int scale = U.arrayIndexScale(ak); - if ((scale & (scale - 1)) != 0) - throw new Error("data type scale not a power of two"); - ASHIFT = 31 - Integer.numberOfLeadingZeros(scale); - } catch (Exception e) { - throw new Error(e); - } - } - } - - // static fields (initialized in static initializer below) - - /** - * Creates a new ForkJoinWorkerThread. This factory is used unless - * overridden in ForkJoinPool constructors. - */ - public static final ForkJoinWorkerThreadFactory - defaultForkJoinWorkerThreadFactory; - - /** - * Per-thread submission bookkeeping. Shared across all pools - * to reduce ThreadLocal pollution and because random motion - * to avoid contention in one pool is likely to hold for others. - * Lazily initialized on first submission (but null-checked - * in other contexts to avoid unnecessary initialization). - */ - static final ThreadLocal<Submitter> submitters; - - /** - * Permission required for callers of methods that may start or - * kill threads. - */ - private static final RuntimePermission modifyThreadPermission; - - /** - * Common (static) pool. Non-null for public use unless a static - * construction exception, but internal usages null-check on use - * to paranoically avoid potential initialization circularities - * as well as to simplify generated code. - */ - static final ForkJoinPool common; - - /** - * Common pool parallelism. Must equal common.parallelism. - */ - static final int commonParallelism; - - /** - * Sequence number for creating workerNamePrefix. - */ - private static int poolNumberSequence; - - /** - * Returns the next sequence number. We don't expect this to - * ever contend, so use simple builtin sync. - */ - private static final synchronized int nextPoolId() { - return ++poolNumberSequence; - } - - // static constants - - /** - * Initial timeout value (in nanoseconds) for the thread - * triggering quiescence to park waiting for new work. On timeout, - * the thread will instead try to shrink the number of - * workers. The value should be large enough to avoid overly - * aggressive shrinkage during most transient stalls (long GCs - * etc). - */ - private static final long IDLE_TIMEOUT = 2000L * 1000L * 1000L; // 2sec - - /** - * Timeout value when there are more threads than parallelism level - */ - private static final long FAST_IDLE_TIMEOUT = 200L * 1000L * 1000L; - - /** - * Tolerance for idle timeouts, to cope with timer undershoots - */ - private static final long TIMEOUT_SLOP = 2000000L; - - /** - * The maximum stolen->joining link depth allowed in method - * tryHelpStealer. Must be a power of two. Depths for legitimate - * chains are unbounded, but we use a fixed constant to avoid - * (otherwise unchecked) cycles and to bound staleness of - * traversal parameters at the expense of sometimes blocking when - * we could be helping. - */ - private static final int MAX_HELP = 64; - - /** - * Increment for seed generators. See class ThreadLocal for - * explanation. - */ - private static final int SEED_INCREMENT = 0x61c88647; - - /* - * Bits and masks for control variables - * - * Field ctl is a long packed with: - * AC: Number of active running workers minus target parallelism (16 bits) - * TC: Number of total workers minus target parallelism (16 bits) - * 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 waiters (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, 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. - * - * When a thread is queued (inactivated), its eventCount field is - * set negative, which is the only way to tell if a worker is - * prevented from executing tasks, even though it must continue to - * scan for them to avoid queuing races. Note however that - * eventCount updates lag releases so usage requires care. - * - * Field plock is an int packed with: - * SHUTDOWN: true if shutdown is enabled (1 bit) - * SEQ: a sequence lock, with PL_LOCK bit set if locked (30 bits) - * SIGNAL: set when threads may be waiting on the lock (1 bit) - * - * The sequence number enables simple consistency checks: - * Staleness of read-only operations on the workQueues array can - * be checked by comparing plock before vs after the reads. - */ - - // 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; - - // bounds - private static final int SMASK = 0xffff; // short bits - private static final int MAX_CAP = 0x7fff; // max #workers - 1 - private static final int EVENMASK = 0xfffe; // even short bits - private static final int SQMASK = 0x007e; // max 64 (even) slots - private static final int SHORT_SIGN = 1 << 15; - private static final int INT_SIGN = 1 << 31; - - // 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; - - // 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 E_SEQ = 1 << EC_SHIFT; - - // plock bits - private static final int SHUTDOWN = 1 << 31; - private static final int PL_LOCK = 2; - private static final int PL_SIGNAL = 1; - private static final int PL_SPINS = 1 << 8; - - // access mode for WorkQueue - static final int LIFO_QUEUE = 0; - static final int FIFO_QUEUE = 1; - static final int SHARED_QUEUE = -1; - - // bounds for #steps in scan loop -- must be power 2 minus 1 - private static final int MIN_SCAN = 0x1ff; // cover estimation slop - private static final int MAX_SCAN = 0x1ffff; // 4 * max workers - - // Instance fields - - /* - * Field layout of this class tends to matter more than one would - * like. Runtime layout order is only loosely related to - * declaration order and may differ across JVMs, but the following - * empirically works OK on current JVMs. - */ - - // Heuristic padding to ameliorate unfortunate memory placements - volatile long pad00, pad01, pad02, pad03, pad04, pad05, pad06; - - volatile long stealCount; // collects worker counts - volatile long ctl; // main pool control - volatile int plock; // shutdown status and seqLock - volatile int indexSeed; // worker/submitter index seed - final int config; // mode and parallelism level - WorkQueue[] workQueues; // main registry - final ForkJoinWorkerThreadFactory factory; - final Thread.UncaughtExceptionHandler ueh; // per-worker UEH - final String workerNamePrefix; // to create worker name string - - volatile Object pad10, pad11, pad12, pad13, pad14, pad15, pad16, pad17; - volatile Object pad18, pad19, pad1a, pad1b; - - /** - * Acquires the plock lock to protect worker array and related - * updates. This method is called only if an initial CAS on plock - * fails. This acts as a spinlock for normal cases, but falls back - * to builtin monitor to block when (rarely) needed. This would be - * a terrible idea for a highly contended lock, but works fine as - * a more conservative alternative to a pure spinlock. - */ - private int acquirePlock() { - int spins = PL_SPINS, r = 0, ps, nps; - for (;;) { - if (((ps = plock) & PL_LOCK) == 0 && - U.compareAndSwapInt(this, PLOCK, ps, nps = ps + PL_LOCK)) - return nps; - else if (r == 0) { // randomize spins if possible - Thread t = Thread.currentThread(); WorkQueue w; Submitter z; - if ((t instanceof ForkJoinWorkerThread) && - (w = ((ForkJoinWorkerThread)t).workQueue) != null) - r = w.seed; - else if ((z = submitters.get()) != null) - r = z.seed; - else - r = 1; - } - else if (spins >= 0) { - r ^= r << 1; r ^= r >>> 3; r ^= r << 10; // xorshift - if (r >= 0) - --spins; - } - else if (U.compareAndSwapInt(this, PLOCK, ps, ps | PL_SIGNAL)) { - synchronized (this) { - if ((plock & PL_SIGNAL) != 0) { - try { - wait(); - } catch (InterruptedException ie) { - try { - Thread.currentThread().interrupt(); - } catch (SecurityException ignore) { - } - } - } - else - notifyAll(); - } - } - } - } - - /** - * Unlocks and signals any thread waiting for plock. Called only - * when CAS of seq value for unlock fails. - */ - private void releasePlock(int ps) { - plock = ps; - synchronized (this) { notifyAll(); } - } - - /** - * Tries to create and start one worker if fewer than target - * parallelism level exist. Adjusts counts etc on failure. - */ - private void tryAddWorker() { - long c; int u; - while ((u = (int)((c = ctl) >>> 32)) < 0 && - (u & SHORT_SIGN) != 0 && (int)c == 0) { - long nc = (long)(((u + UTC_UNIT) & UTC_MASK) | - ((u + UAC_UNIT) & UAC_MASK)) << 32; - if (U.compareAndSwapLong(this, CTL, c, nc)) { - ForkJoinWorkerThreadFactory fac; - Throwable ex = null; - ForkJoinWorkerThread wt = null; - try { - if ((fac = factory) != null && - (wt = fac.newThread(this)) != null) { - wt.start(); - break; - } - } catch (Throwable e) { - ex = e; - } - deregisterWorker(wt, ex); - break; - } - } - } - - // Registering and deregistering workers - - /** - * Callback from ForkJoinWorkerThread to establish and record its - * WorkQueue. To avoid scanning bias due to packing entries in - * front of the workQueues array, we treat the array as a simple - * power-of-two hash table using per-thread seed as hash, - * expanding as needed. - * - * @param wt the worker thread - * @return the worker's queue - */ - final WorkQueue registerWorker(ForkJoinWorkerThread wt) { - Thread.UncaughtExceptionHandler handler; WorkQueue[] ws; int s, ps; - wt.setDaemon(true); - if ((handler = ueh) != null) - wt.setUncaughtExceptionHandler(handler); - do {} while (!U.compareAndSwapInt(this, INDEXSEED, s = indexSeed, - s += SEED_INCREMENT) || - s == 0); // skip 0 - WorkQueue w = new WorkQueue(this, wt, config >>> 16, s); - if (((ps = plock) & PL_LOCK) != 0 || - !U.compareAndSwapInt(this, PLOCK, ps, ps += PL_LOCK)) - ps = acquirePlock(); - int nps = (ps & SHUTDOWN) | ((ps + PL_LOCK) & ~SHUTDOWN); - try { - if ((ws = workQueues) != null) { // skip if shutting down - int n = ws.length, m = n - 1; - int r = (s << 1) | 1; // use odd-numbered indices - if (ws[r &= m] != null) { // collision - int probes = 0; // step by approx half size - int step = (n <= 4) ? 2 : ((n >>> 1) & EVENMASK) + 2; - while (ws[r = (r + step) & m] != null) { - if (++probes >= n) { - workQueues = ws = Arrays.copyOf(ws, n <<= 1); - m = n - 1; - probes = 0; - } - } - } - w.eventCount = w.poolIndex = r; // volatile write orders - ws[r] = w; - } - } finally { - if (!U.compareAndSwapInt(this, PLOCK, ps, nps)) - releasePlock(nps); - } - wt.setName(workerNamePrefix.concat(Integer.toString(w.poolIndex))); - return w; - } - - /** - * Final callback from terminating worker, as well as upon failure - * to construct or start a worker. Removes record of worker from - * array, and adjusts counts. If pool is shutting down, tries to - * complete termination. - * - * @param wt the worker thread or null if construction failed - * @param ex the exception causing failure, or null if none - */ - final void deregisterWorker(ForkJoinWorkerThread wt, Throwable ex) { - WorkQueue w = null; - if (wt != null && (w = wt.workQueue) != null) { - int ps; - w.qlock = -1; // ensure set - long ns = w.nsteals, sc; // collect steal count - do {} while (!U.compareAndSwapLong(this, STEALCOUNT, - sc = stealCount, sc + ns)); - if (((ps = plock) & PL_LOCK) != 0 || - !U.compareAndSwapInt(this, PLOCK, ps, ps += PL_LOCK)) - ps = acquirePlock(); - int nps = (ps & SHUTDOWN) | ((ps + PL_LOCK) & ~SHUTDOWN); - try { - int idx = w.poolIndex; - WorkQueue[] ws = workQueues; - if (ws != null && idx >= 0 && idx < ws.length && ws[idx] == w) - ws[idx] = null; - } finally { - if (!U.compareAndSwapInt(this, PLOCK, ps, nps)) - releasePlock(nps); - } - } - - long c; // adjust ctl counts - do {} while (!U.compareAndSwapLong - (this, CTL, c = ctl, (((c - AC_UNIT) & AC_MASK) | - ((c - TC_UNIT) & TC_MASK) | - (c & ~(AC_MASK|TC_MASK))))); - - if (!tryTerminate(false, false) && w != null && w.array != null) { - w.cancelAll(); // cancel remaining tasks - WorkQueue[] ws; WorkQueue v; Thread p; int u, i, e; - while ((u = (int)((c = ctl) >>> 32)) < 0 && (e = (int)c) >= 0) { - if (e > 0) { // activate or create replacement - if ((ws = workQueues) == null || - (i = e & SMASK) >= ws.length || - (v = ws[i]) == null) - break; - long nc = (((long)(v.nextWait & E_MASK)) | - ((long)(u + UAC_UNIT) << 32)); - if (v.eventCount != (e | INT_SIGN)) - break; - if (U.compareAndSwapLong(this, CTL, c, nc)) { - v.eventCount = (e + E_SEQ) & E_MASK; - if ((p = v.parker) != null) - U.unpark(p); - break; - } - } - else { - if ((short)u < 0) - tryAddWorker(); - break; - } - } - } - if (ex == null) // help clean refs on way out - ForkJoinTask.helpExpungeStaleExceptions(); - else // rethrow - ForkJoinTask.rethrow(ex); - } - - // Submissions - - /** - * Unless shutting down, adds the given task to a submission queue - * at submitter's current queue index (modulo submission - * range). Only the most common path is directly handled in this - * method. All others are relayed to fullExternalPush. - * - * @param task the task. Caller must ensure non-null. - */ - final void externalPush(ForkJoinTask<?> task) { - WorkQueue[] ws; WorkQueue q; Submitter z; int m; ForkJoinTask<?>[] a; - if ((z = submitters.get()) != null && plock > 0 && - (ws = workQueues) != null && (m = (ws.length - 1)) >= 0 && - (q = ws[m & z.seed & SQMASK]) != null && - U.compareAndSwapInt(q, QLOCK, 0, 1)) { // lock - int b = q.base, s = q.top, n, an; - if ((a = q.array) != null && (an = a.length) > (n = s + 1 - b)) { - int j = (((an - 1) & s) << ASHIFT) + ABASE; - U.putOrderedObject(a, j, task); - q.top = s + 1; // push on to deque - q.qlock = 0; - if (n <= 2) - signalWork(q); - return; - } - q.qlock = 0; - } - fullExternalPush(task); - } - - /** - * Full version of externalPush. This method is called, among - * other times, upon the first submission of the first task to the - * pool, so must perform secondary initialization. It also - * detects first submission by an external thread by looking up - * its ThreadLocal, and creates a new shared queue if the one at - * index if empty or contended. The plock lock body must be - * exception-free (so no try/finally) so we optimistically - * allocate new queues outside the lock and throw them away if - * (very rarely) not needed. - * - * Secondary initialization occurs when plock is zero, to create - * workQueue array and set plock to a valid value. This lock body - * must also be exception-free. Because the plock seq value can - * eventually wrap around zero, this method harmlessly fails to - * reinitialize if workQueues exists, while still advancing plock. - */ - private void fullExternalPush(ForkJoinTask<?> task) { - int r = 0; // random index seed - for (Submitter z = submitters.get();;) { - WorkQueue[] ws; WorkQueue q; int ps, m, k; - if (z == null) { - if (U.compareAndSwapInt(this, INDEXSEED, r = indexSeed, - r += SEED_INCREMENT) && r != 0) - submitters.set(z = new Submitter(r)); - } - else if (r == 0) { // move to a different index - r = z.seed; - r ^= r << 13; // same xorshift as WorkQueues - r ^= r >>> 17; - z.seed = r ^ (r << 5); - } - else if ((ps = plock) < 0) - throw new RejectedExecutionException(); - else if (ps == 0 || (ws = workQueues) == null || - (m = ws.length - 1) < 0) { // initialize workQueues - int p = config & SMASK; // find power of two table size - int n = (p > 1) ? p - 1 : 1; // ensure at least 2 slots - n |= n >>> 1; n |= n >>> 2; n |= n >>> 4; - n |= n >>> 8; n |= n >>> 16; n = (n + 1) << 1; - WorkQueue[] nws = ((ws = workQueues) == null || ws.length == 0 ? - new WorkQueue[n] : null); - if (((ps = plock) & PL_LOCK) != 0 || - !U.compareAndSwapInt(this, PLOCK, ps, ps += PL_LOCK)) - ps = acquirePlock(); - if (((ws = workQueues) == null || ws.length == 0) && nws != null) - workQueues = nws; - int nps = (ps & SHUTDOWN) | ((ps + PL_LOCK) & ~SHUTDOWN); - if (!U.compareAndSwapInt(this, PLOCK, ps, nps)) - releasePlock(nps); - } - else if ((q = ws[k = r & m & SQMASK]) != null) { - if (q.qlock == 0 && U.compareAndSwapInt(q, QLOCK, 0, 1)) { - ForkJoinTask<?>[] a = q.array; - int s = q.top; - boolean submitted = false; - try { // locked version of push - if ((a != null && a.length > s + 1 - q.base) || - (a = q.growArray()) != null) { // must presize - int j = (((a.length - 1) & s) << ASHIFT) + ABASE; - U.putOrderedObject(a, j, task); - q.top = s + 1; - submitted = true; - } - } finally { - q.qlock = 0; // unlock - } - if (submitted) { - signalWork(q); - return; - } - } - r = 0; // move on failure - } - else if (((ps = plock) & PL_LOCK) == 0) { // create new queue - q = new WorkQueue(this, null, SHARED_QUEUE, r); - if (((ps = plock) & PL_LOCK) != 0 || - !U.compareAndSwapInt(this, PLOCK, ps, ps += PL_LOCK)) - ps = acquirePlock(); - if ((ws = workQueues) != null && k < ws.length && ws[k] == null) - ws[k] = q; - int nps = (ps & SHUTDOWN) | ((ps + PL_LOCK) & ~SHUTDOWN); - if (!U.compareAndSwapInt(this, PLOCK, ps, nps)) - releasePlock(nps); - } - else - r = 0; // try elsewhere while lock held - } - } - - // Maintaining ctl counts - - /** - * Increments active count; mainly called upon return from blocking. - */ - final void incrementActiveCount() { - long c; - do {} while (!U.compareAndSwapLong(this, CTL, c = ctl, c + AC_UNIT)); - } - - /** - * Tries to create or activate a worker if too few are active. - * - * @param q the (non-null) queue holding tasks to be signalled - */ - final void signalWork(WorkQueue q) { - int hint = q.poolIndex; - long c; int e, u, i, n; WorkQueue[] ws; WorkQueue w; Thread p; - while ((u = (int)((c = ctl) >>> 32)) < 0) { - if ((e = (int)c) > 0) { - if ((ws = workQueues) != null && ws.length > (i = e & SMASK) && - (w = ws[i]) != null && w.eventCount == (e | INT_SIGN)) { - long nc = (((long)(w.nextWait & E_MASK)) | - ((long)(u + UAC_UNIT) << 32)); - if (U.compareAndSwapLong(this, CTL, c, nc)) { - w.hint = hint; - w.eventCount = (e + E_SEQ) & E_MASK; - if ((p = w.parker) != null) - U.unpark(p); - break; - } - if (q.top - q.base <= 0) - break; - } - else - break; - } - else { - if ((short)u < 0) - tryAddWorker(); - break; - } - } - } - - // Scanning for tasks - - /** - * Top-level runloop for workers, called by ForkJoinWorkerThread.run. - */ - final void runWorker(WorkQueue w) { - w.growArray(); // allocate queue - do { w.runTask(scan(w)); } while (w.qlock >= 0); - } - - /** - * Scans for and, if found, returns one task, else possibly - * inactivates the worker. This method operates on single reads of - * volatile state and is designed to be re-invoked continuously, - * in part because it returns upon detecting inconsistencies, - * contention, or state changes that indicate possible success on - * re-invocation. - * - * The scan searches for tasks across queues (starting at a random - * index, and relying on registerWorker to irregularly scatter - * them within array to avoid bias), checking each at least twice. - * The scan terminates upon either finding a non-empty queue, or - * completing the sweep. If the worker is not inactivated, it - * takes and returns a task from this queue. Otherwise, if not - * activated, it signals workers (that may include itself) and - * returns so caller can retry. Also returns for true if the - * worker array may have changed during an empty scan. On failure - * to find a task, we take one of the following actions, after - * which the caller will retry calling this method unless - * terminated. - * - * * If pool is terminating, terminate the worker. - * - * * If not already enqueued, try to inactivate and enqueue the - * worker on wait queue. Or, if inactivating has caused the pool - * to be quiescent, relay to idleAwaitWork to possibly shrink - * pool. - * - * * If already enqueued and none of the above apply, possibly - * park awaiting signal, else lingering to help scan and signal. - * - * * If a non-empty queue discovered or left as a hint, - * help wake up other workers before return. - * - * @param w the worker (via its WorkQueue) - * @return a task or null if none found - */ - private final ForkJoinTask<?> scan(WorkQueue w) { - WorkQueue[] ws; int m; - int ps = plock; // read plock before ws - if (w != null && (ws = workQueues) != null && (m = ws.length - 1) >= 0) { - int ec = w.eventCount; // ec is negative if inactive - int r = w.seed; r ^= r << 13; r ^= r >>> 17; w.seed = r ^= r << 5; - w.hint = -1; // update seed and clear hint - int j = ((m + m + 1) | MIN_SCAN) & MAX_SCAN; - do { - WorkQueue q; ForkJoinTask<?>[] a; int b; - if ((q = ws[(r + j) & m]) != null && (b = q.base) - q.top < 0 && - (a = q.array) != null) { // probably nonempty - int i = (((a.length - 1) & b) << ASHIFT) + ABASE; - ForkJoinTask<?> t = (ForkJoinTask<?>) - U.getObjectVolatile(a, i); - if (q.base == b && ec >= 0 && t != null && - U.compareAndSwapObject(a, i, t, null)) { - if ((q.base = b + 1) - q.top < 0) - signalWork(q); - return t; // taken - } - else if ((ec < 0 || j < m) && (int)(ctl >> AC_SHIFT) <= 0) { - w.hint = (r + j) & m; // help signal below - break; // cannot take - } - } - } while (--j >= 0); - - int h, e, ns; long c, sc; WorkQueue q; - if ((ns = w.nsteals) != 0) { - if (U.compareAndSwapLong(this, STEALCOUNT, - sc = stealCount, sc + ns)) - w.nsteals = 0; // collect steals and rescan - } - else if (plock != ps) // consistency check - ; // skip - else if ((e = (int)(c = ctl)) < 0) - w.qlock = -1; // pool is terminating - else { - if ((h = w.hint) < 0) { - if (ec >= 0) { // try to enqueue/inactivate - long nc = (((long)ec | - ((c - AC_UNIT) & (AC_MASK|TC_MASK)))); - w.nextWait = e; // link and mark inactive - w.eventCount = ec | INT_SIGN; - if (ctl != c || !U.compareAndSwapLong(this, CTL, c, nc)) - w.eventCount = ec; // unmark on CAS failure - else if ((int)(c >> AC_SHIFT) == 1 - (config & SMASK)) - idleAwaitWork(w, nc, c); - } - else if (w.eventCount < 0 && ctl == c) { - Thread wt = Thread.currentThread(); - Thread.interrupted(); // clear status - U.putObject(wt, PARKBLOCKER, this); - w.parker = wt; // emulate LockSupport.park - if (w.eventCount < 0) // recheck - U.park(false, 0L); // block - w.parker = null; - U.putObject(wt, PARKBLOCKER, null); - } - } - if ((h >= 0 || (h = w.hint) >= 0) && - (ws = workQueues) != null && h < ws.length && - (q = ws[h]) != null) { // signal others before retry - WorkQueue v; Thread p; int u, i, s; - for (int n = (config & SMASK) - 1;;) { - int idleCount = (w.eventCount < 0) ? 0 : -1; - if (((s = idleCount - q.base + q.top) <= n && - (n = s) <= 0) || - (u = (int)((c = ctl) >>> 32)) >= 0 || - (e = (int)c) <= 0 || m < (i = e & SMASK) || - (v = ws[i]) == null) - break; - long nc = (((long)(v.nextWait & E_MASK)) | - ((long)(u + UAC_UNIT) << 32)); - if (v.eventCount != (e | INT_SIGN) || - !U.compareAndSwapLong(this, CTL, c, nc)) - break; - v.hint = h; - v.eventCount = (e + E_SEQ) & E_MASK; - if ((p = v.parker) != null) - U.unpark(p); - if (--n <= 0) - break; - } - } - } - } - return null; - } - - /** - * If inactivating worker w has caused the pool to become - * quiescent, checks for pool termination, and, so long as this is - * not the only worker, waits for event for up to a given - * duration. On timeout, if ctl has not changed, terminates the - * worker, which will in turn wake up another worker to possibly - * repeat this process. - * - * @param w the calling worker - * @param currentCtl the ctl value triggering possible quiescence - * @param prevCtl the ctl value to restore if thread is terminated - */ - private void idleAwaitWork(WorkQueue w, long currentCtl, long prevCtl) { - if (w != null && w.eventCount < 0 && - !tryTerminate(false, false) && (int)prevCtl != 0 && - ctl == currentCtl) { - int dc = -(short)(currentCtl >>> TC_SHIFT); - long parkTime = dc < 0 ? FAST_IDLE_TIMEOUT: (dc + 1) * IDLE_TIMEOUT; - long deadline = System.nanoTime() + parkTime - TIMEOUT_SLOP; - Thread wt = Thread.currentThread(); - while (ctl == currentCtl) { - Thread.interrupted(); // timed variant of version in scan() - U.putObject(wt, PARKBLOCKER, this); - w.parker = wt; - if (ctl == currentCtl) - U.park(false, parkTime); - w.parker = null; - U.putObject(wt, PARKBLOCKER, null); - if (ctl != currentCtl) - break; - if (deadline - System.nanoTime() <= 0L && - U.compareAndSwapLong(this, CTL, currentCtl, prevCtl)) { - w.eventCount = (w.eventCount + E_SEQ) | E_MASK; - w.hint = -1; - w.qlock = -1; // shrink - break; - } - } - } - } - - /** - * Scans through queues looking for work while joining a task; if - * any present, signals. May return early if more signalling is - * detectably unneeded. - * - * @param task return early if done - * @param origin an index to start scan - */ - private void helpSignal(ForkJoinTask<?> task, int origin) { - WorkQueue[] ws; WorkQueue w; Thread p; long c; int m, u, e, i, s; - if (task != null && task.status >= 0 && - (u = (int)(ctl >>> 32)) < 0 && (u >> UAC_SHIFT) < 0 && - (ws = workQueues) != null && (m = ws.length - 1) >= 0) { - outer: for (int k = origin, j = m; j >= 0; --j) { - WorkQueue q = ws[k++ & m]; - for (int n = m;;) { // limit to at most m signals - if (task.status < 0) - break outer; - if (q == null || - ((s = -q.base + q.top) <= n && (n = s) <= 0)) - break; - if ((u = (int)((c = ctl) >>> 32)) >= 0 || - (e = (int)c) <= 0 || m < (i = e & SMASK) || - (w = ws[i]) == null) - break outer; - long nc = (((long)(w.nextWait & E_MASK)) | - ((long)(u + UAC_UNIT) << 32)); - if (w.eventCount != (e | INT_SIGN)) - break outer; - if (U.compareAndSwapLong(this, CTL, c, nc)) { - w.eventCount = (e + E_SEQ) & E_MASK; - if ((p = w.parker) != null) - U.unpark(p); - if (--n <= 0) - break; - } - } - } - } - } - - /** - * 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 first call to this method upon a - * waiting join will often entail scanning/search, (which is OK - * because the joiner has nothing better to do), but this method - * leaves hints in workers to speed up subsequent calls. The - * implementation is very branchy to cope with potential - * inconsistencies or loops encountering chains that are stale, - * unknown, or so long that they are likely cyclic. - * - * @param joiner the joining worker - * @param task the task to join - * @return 0 if no progress can be made, negative if task - * known complete, else positive - */ - private int tryHelpStealer(WorkQueue joiner, ForkJoinTask<?> task) { - int stat = 0, steps = 0; // bound to avoid cycles - if (joiner != null && task != null) { // hoist null checks - restart: for (;;) { - ForkJoinTask<?> subtask = task; // current target - for (WorkQueue j = joiner, v;;) { // v is stealer of subtask - WorkQueue[] ws; int m, s, h; - if ((s = task.status) < 0) { - stat = s; - break restart; - } - if ((ws = workQueues) == null || (m = ws.length - 1) <= 0) - break restart; // shutting down - if ((v = ws[h = (j.hint | 1) & m]) == null || - v.currentSteal != subtask) { - for (int origin = h;;) { // find stealer - if (((h = (h + 2) & m) & 15) == 1 && - (subtask.status < 0 || j.currentJoin != subtask)) - continue restart; // occasional staleness check - if ((v = ws[h]) != null && - v.currentSteal == subtask) { - j.hint = h; // save hint - break; - } - if (h == origin) - break restart; // cannot find stealer - } - } - for (;;) { // help stealer or descend to its stealer - ForkJoinTask[] a; int b; - if (subtask.status < 0) // surround probes with - continue restart; // consistency checks - if ((b = v.base) - v.top < 0 && (a = v.array) != null) { - int i = (((a.length - 1) & b) << ASHIFT) + ABASE; - ForkJoinTask<?> t = - (ForkJoinTask<?>)U.getObjectVolatile(a, i); - if (subtask.status < 0 || j.currentJoin != subtask || - v.currentSteal != subtask) - continue restart; // stale - stat = 1; // apparent progress - if (t != null && v.base == b && - U.compareAndSwapObject(a, i, t, null)) { - v.base = b + 1; // help stealer - joiner.runSubtask(t); - } - else if (v.base == b && ++steps == MAX_HELP) - break restart; // v apparently stalled - } - else { // empty -- try to descend - ForkJoinTask<?> next = v.currentJoin; - if (subtask.status < 0 || j.currentJoin != subtask || - v.currentSteal != subtask) - continue restart; // stale - else if (next == null || ++steps == MAX_HELP) - break restart; // dead-end or maybe cyclic - else { - subtask = next; - j = v; - break; - } - } - } - } - } - } - return stat; - } - - /** - * Analog of tryHelpStealer for CountedCompleters. Tries to steal - * and run tasks within the target's computation. - * - * @param task the task to join - * @param mode if shared, exit upon completing any task - * if all workers are active - */ - private int helpComplete(ForkJoinTask<?> task, int mode) { - WorkQueue[] ws; WorkQueue q; int m, n, s, u; - if (task != null && (ws = workQueues) != null && - (m = ws.length - 1) >= 0) { - for (int j = 1, origin = j;;) { - if ((s = task.status) < 0) - return s; - if ((q = ws[j & m]) != null && q.pollAndExecCC(task)) { - origin = j; - if (mode == SHARED_QUEUE && - ((u = (int)(ctl >>> 32)) >= 0 || (u >> UAC_SHIFT) >= 0)) - break; - } - else if ((j = (j + 2) & m) == origin) - break; - } - } - return 0; - } - - /** - * Tries to decrement active count (sometimes implicitly) and - * possibly release or create a compensating worker in preparation - * for blocking. Fails on contention or termination. Otherwise, - * adds a new thread if no idle workers are available and pool - * may become starved. - */ - final boolean tryCompensate() { - int pc = config & SMASK, e, i, tc; long c; - WorkQueue[] ws; WorkQueue w; Thread p; - if ((ws = workQueues) != null && (e = (int)(c = ctl)) >= 0) { - if (e != 0 && (i = e & SMASK) < ws.length && - (w = ws[i]) != null && w.eventCount == (e | INT_SIGN)) { - long nc = ((long)(w.nextWait & E_MASK) | - (c & (AC_MASK|TC_MASK))); - if (U.compareAndSwapLong(this, CTL, c, nc)) { - w.eventCount = (e + E_SEQ) & E_MASK; - if ((p = w.parker) != null) - U.unpark(p); - return true; // replace with idle worker - } - } - else if ((tc = (short)(c >>> TC_SHIFT)) >= 0 && - (int)(c >> AC_SHIFT) + pc > 1) { - long nc = ((c - AC_UNIT) & AC_MASK) | (c & ~AC_MASK); - if (U.compareAndSwapLong(this, CTL, c, nc)) - return true; // no compensation - } - else if (tc + pc < MAX_CAP) { - long nc = ((c + TC_UNIT) & TC_MASK) | (c & ~TC_MASK); - if (U.compareAndSwapLong(this, CTL, c, nc)) { - ForkJoinWorkerThreadFactory fac; - Throwable ex = null; - ForkJoinWorkerThread wt = null; - try { - if ((fac = factory) != null && - (wt = fac.newThread(this)) != null) { - wt.start(); - return true; - } - } catch (Throwable rex) { - ex = rex; - } - deregisterWorker(wt, ex); // clean up and return false - } - } - } - return false; - } - - /** - * Helps and/or blocks until the given task is done. - * - * @param joiner the joining worker - * @param task the task - * @return task status on exit - */ - final int awaitJoin(WorkQueue joiner, ForkJoinTask<?> task) { - int s = 0; - if (joiner != null && task != null && (s = task.status) >= 0) { - ForkJoinTask<?> prevJoin = joiner.currentJoin; - joiner.currentJoin = task; - do {} while ((s = task.status) >= 0 && !joiner.isEmpty() && - joiner.tryRemoveAndExec(task)); // process local tasks - if (s >= 0 && (s = task.status) >= 0) { - helpSignal(task, joiner.poolIndex); - if ((s = task.status) >= 0 && - (task instanceof CountedCompleter)) - s = helpComplete(task, LIFO_QUEUE); - } - while (s >= 0 && (s = task.status) >= 0) { - if ((!joiner.isEmpty() || // try helping - (s = tryHelpStealer(joiner, task)) == 0) && - (s = task.status) >= 0) { - helpSignal(task, joiner.poolIndex); - if ((s = task.status) >= 0 && tryCompensate()) { - if (task.trySetSignal() && (s = task.status) >= 0) { - synchronized (task) { - if (task.status >= 0) { - try { // see ForkJoinTask - task.wait(); // for explanation - } catch (InterruptedException ie) { - } - } - else - task.notifyAll(); - } - } - long c; // re-activate - do {} while (!U.compareAndSwapLong - (this, CTL, c = ctl, c + AC_UNIT)); - } - } - } - joiner.currentJoin = prevJoin; - } - return s; - } - - /** - * Stripped-down variant of awaitJoin used by timed joins. Tries - * to help join only while there is continuous progress. (Caller - * will then enter a timed wait.) - * - * @param joiner the joining worker - * @param task the task - */ - final void helpJoinOnce(WorkQueue joiner, ForkJoinTask<?> task) { - int s; - if (joiner != null && task != null && (s = task.status) >= 0) { - ForkJoinTask<?> prevJoin = joiner.currentJoin; - joiner.currentJoin = task; - do {} while ((s = task.status) >= 0 && !joiner.isEmpty() && - joiner.tryRemoveAndExec(task)); - if (s >= 0 && (s = task.status) >= 0) { - helpSignal(task, joiner.poolIndex); - if ((s = task.status) >= 0 && - (task instanceof CountedCompleter)) - s = helpComplete(task, LIFO_QUEUE); - } - if (s >= 0 && joiner.isEmpty()) { - do {} while (task.status >= 0 && - tryHelpStealer(joiner, task) > 0); - } - joiner.currentJoin = prevJoin; - } - } - - /** - * Returns a (probably) non-empty steal queue, if one is found - * during a scan, else null. This method must be retried by - * caller if, by the time it tries to use the queue, it is empty. - * @param r a (random) seed for scanning - */ - private WorkQueue findNonEmptyStealQueue(int r) { - for (;;) { - int ps = plock, m; WorkQueue[] ws; WorkQueue q; - if ((ws = workQueues) != null && (m = ws.length - 1) >= 0) { - for (int j = (m + 1) << 2; j >= 0; --j) { - if ((q = ws[(((r + j) << 1) | 1) & m]) != null && - q.base - q.top < 0) - return q; - } - } - if (plock == ps) - return null; - } - } - - /** - * Runs tasks until {@code isQuiescent()}. We piggyback on - * active count ctl maintenance, but rather than blocking - * when tasks cannot be found, we rescan until all others cannot - * find tasks either. - */ - final void helpQuiescePool(WorkQueue w) { - for (boolean active = true;;) { - long c; WorkQueue q; ForkJoinTask<?> t; int b; - while ((t = w.nextLocalTask()) != null) { - if (w.base - w.top < 0) - signalWork(w); - t.doExec(); - } - if ((q = findNonEmptyStealQueue(w.nextSeed())) != null) { - if (!active) { // re-establish active count - active = true; - do {} while (!U.compareAndSwapLong - (this, CTL, c = ctl, c + AC_UNIT)); - } - if ((b = q.base) - q.top < 0 && (t = q.pollAt(b)) != null) { - if (q.base - q.top < 0) - signalWork(q); - w.runSubtask(t); - } - } - else if (active) { // decrement active count without queuing - long nc = (c = ctl) - AC_UNIT; - if ((int)(nc >> AC_SHIFT) + (config & SMASK) == 0) - return; // bypass decrement-then-increment - if (U.compareAndSwapLong(this, CTL, c, nc)) - active = false; - } - else if ((int)((c = ctl) >> AC_SHIFT) + (config & SMASK) == 0 && - U.compareAndSwapLong(this, CTL, c, c + AC_UNIT)) - return; - } - } - - /** - * Gets and removes a local or stolen task for the given worker. - * - * @return a task, if available - */ - final ForkJoinTask<?> nextTaskFor(WorkQueue w) { - for (ForkJoinTask<?> t;;) { - WorkQueue q; int b; - if ((t = w.nextLocalTask()) != null) - return t; - if ((q = findNonEmptyStealQueue(w.nextSeed())) == null) - return null; - if ((b = q.base) - q.top < 0 && (t = q.pollAt(b)) != null) { - if (q.base - q.top < 0) - signalWork(q); - return t; - } - } - } - - /** - * Returns a cheap heuristic guide for task partitioning when - * programmers, frameworks, tools, or languages have little or no - * idea about task granularity. In essence by offering this - * method, we ask users only about tradeoffs in overhead vs - * expected throughput and its variance, rather than how finely to - * partition tasks. - * - * In a steady state strict (tree-structured) computation, each - * thread makes available for stealing enough tasks for other - * threads to remain active. Inductively, if all threads play by - * the same rules, each thread should make available only a - * constant number of tasks. - * - * The minimum useful constant is just 1. But using a value of 1 - * would require immediate replenishment upon each steal to - * maintain enough tasks, which is infeasible. Further, - * partitionings/granularities of offered tasks should minimize - * steal rates, which in general means that threads nearer the top - * of computation tree should generate more than those nearer the - * bottom. In perfect steady state, each thread is at - * approximately the same level of computation tree. However, - * producing extra tasks amortizes the uncertainty of progress and - * diffusion assumptions. - * - * So, users will want to use values larger (but not much larger) - * than 1 to both smooth over transient shortages and hedge - * against uneven progress; as traded off against the cost of - * extra task overhead. We leave the user to pick a threshold - * value to compare with the results of this call to guide - * decisions, but recommend values such as 3. - * - * When all threads are active, it is on average OK to estimate - * surplus strictly locally. In steady-state, if one thread is - * maintaining say 2 surplus tasks, then so are others. So we can - * just use estimated queue length. However, this strategy alone - * leads to serious mis-estimates in some non-steady-state - * conditions (ramp-up, ramp-down, other stalls). We can detect - * many of these by further considering the number of "idle" - * threads, that are known to have zero queued tasks, so - * compensate by a factor of (#idle/#active) threads. - * - * Note: The approximation of #busy workers as #active workers is - * not very good under current signalling scheme, and should be - * improved. - */ - static int getSurplusQueuedTaskCount() { - Thread t; ForkJoinWorkerThread wt; ForkJoinPool pool; WorkQueue q; - if (((t = Thread.currentThread()) instanceof ForkJoinWorkerThread)) { - int p = (pool = (wt = (ForkJoinWorkerThread)t).pool).config & SMASK; - int n = (q = wt.workQueue).top - q.base; - int a = (int)(pool.ctl >> AC_SHIFT) + p; - return n - (a > (p >>>= 1) ? 0 : - a > (p >>>= 1) ? 1 : - a > (p >>>= 1) ? 2 : - a > (p >>>= 1) ? 4 : - 8); - } - return 0; - } - - // Termination - - /** - * Possibly initiates and/or completes termination. The caller - * triggering termination runs three passes through workQueues: - * (0) Setting termination status, followed by wakeups of queued - * workers; (1) cancelling all 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. - * - * @param now if true, unconditionally terminate, else only - * if no work and no active workers - * @param enable if true, enable shutdown when next possible - * @return true if now terminating or terminated - */ - private boolean tryTerminate(boolean now, boolean enable) { - int ps; - if (this == common) // cannot shut down - return false; - if ((ps = plock) >= 0) { // enable by setting plock - if (!enable) - return false; - if ((ps & PL_LOCK) != 0 || - !U.compareAndSwapInt(this, PLOCK, ps, ps += PL_LOCK)) - ps = acquirePlock(); - int nps = ((ps + PL_LOCK) & ~SHUTDOWN) | SHUTDOWN; - if (!U.compareAndSwapInt(this, PLOCK, ps, nps)) - releasePlock(nps); - } - for (long c;;) { - if (((c = ctl) & STOP_BIT) != 0) { // already terminating - if ((short)(c >>> TC_SHIFT) == -(config & SMASK)) { - synchronized (this) { - notifyAll(); // signal when 0 workers - } - } - return true; - } - if (!now) { // check if idle & no tasks - WorkQueue[] ws; WorkQueue w; - if ((int)(c >> AC_SHIFT) != -(config & SMASK)) - return false; - if ((ws = workQueues) != null) { - for (int i = 0; i < ws.length; ++i) { - if ((w = ws[i]) != null) { - if (!w.isEmpty()) { // signal unprocessed tasks - signalWork(w); - return false; - } - if ((i & 1) != 0 && w.eventCount >= 0) - return false; // unqueued inactive worker - } - } - } - } - if (U.compareAndSwapLong(this, CTL, c, c | STOP_BIT)) { - for (int pass = 0; pass < 3; ++pass) { - WorkQueue[] ws; WorkQueue w; Thread wt; - if ((ws = workQueues) != null) { - int n = ws.length; - for (int i = 0; i < n; ++i) { - if ((w = ws[i]) != null) { - w.qlock = -1; - if (pass > 0) { - w.cancelAll(); - if (pass > 1 && (wt = w.owner) != null) { - if (!wt.isInterrupted()) { - try { - wt.interrupt(); - } catch (Throwable ignore) { - } - } - U.unpark(wt); - } - } - } - } - // Wake up workers parked on event queue - int i, e; long cc; Thread p; - while ((e = (int)(cc = ctl) & E_MASK) != 0 && - (i = e & SMASK) < n && i >= 0 && - (w = ws[i]) != null) { - long nc = ((long)(w.nextWait & E_MASK) | - ((cc + AC_UNIT) & AC_MASK) | - (cc & (TC_MASK|STOP_BIT))); - if (w.eventCount == (e | INT_SIGN) && - U.compareAndSwapLong(this, CTL, cc, nc)) { - w.eventCount = (e + E_SEQ) & E_MASK; - w.qlock = -1; - if ((p = w.parker) != null) - U.unpark(p); - } - } - } - } - } - } - } - - // external operations on common pool - - /** - * Returns common pool queue for a thread that has submitted at - * least one task. - */ - static WorkQueue commonSubmitterQueue() { - ForkJoinPool p; WorkQueue[] ws; int m; Submitter z; - return ((z = submitters.get()) != null && - (p = common) != null && - (ws = p.workQueues) != null && - (m = ws.length - 1) >= 0) ? - ws[m & z.seed & SQMASK] : null; - } - - /** - * Tries to pop the given task from submitter's queue in common pool. - */ - static boolean tryExternalUnpush(ForkJoinTask<?> t) { - ForkJoinPool p; WorkQueue[] ws; WorkQueue q; Submitter z; - ForkJoinTask<?>[] a; int m, s; - if (t != null && - (z = submitters.get()) != null && - (p = common) != null && - (ws = p.workQueues) != null && - (m = ws.length - 1) >= 0 && - (q = ws[m & z.seed & SQMASK]) != null && - (s = q.top) != q.base && - (a = q.array) != null) { - long j = (((a.length - 1) & (s - 1)) << ASHIFT) + ABASE; - if (U.getObject(a, j) == t && - U.compareAndSwapInt(q, QLOCK, 0, 1)) { - if (q.array == a && q.top == s && // recheck - U.compareAndSwapObject(a, j, t, null)) { - q.top = s - 1; - q.qlock = 0; - return true; - } - q.qlock = 0; - } - } - return false; - } - - /** - * Tries to pop and run local tasks within the same computation - * as the given root. On failure, tries to help complete from - * other queues via helpComplete. - */ - private void externalHelpComplete(WorkQueue q, ForkJoinTask<?> root) { - ForkJoinTask<?>[] a; int m; - if (q != null && (a = q.array) != null && (m = (a.length - 1)) >= 0 && - root != null && root.status >= 0) { - for (;;) { - int s, u; Object o; CountedCompleter<?> task = null; - if ((s = q.top) - q.base > 0) { - long j = ((m & (s - 1)) << ASHIFT) + ABASE; - if ((o = U.getObject(a, j)) != null && - (o instanceof CountedCompleter)) { - CountedCompleter<?> t = (CountedCompleter<?>)o, r = t; - do { - if (r == root) { - if (U.compareAndSwapInt(q, QLOCK, 0, 1)) { - if (q.array == a && q.top == s && - U.compareAndSwapObject(a, j, t, null)) { - q.top = s - 1; - task = t; - } - q.qlock = 0; - } - break; - } - } while ((r = r.completer) != null); - } - } - if (task != null) - task.doExec(); - if (root.status < 0 || - (u = (int)(ctl >>> 32)) >= 0 || (u >> UAC_SHIFT) >= 0) - break; - if (task == null) { - helpSignal(root, q.poolIndex); - if (root.status >= 0) - helpComplete(root, SHARED_QUEUE); - break; - } - } - } - } - - /** - * Tries to help execute or signal availability of the given task - * from submitter's queue in common pool. - */ - static void externalHelpJoin(ForkJoinTask<?> t) { - // Some hard-to-avoid overlap with tryExternalUnpush - ForkJoinPool p; WorkQueue[] ws; WorkQueue q, w; Submitter z; - ForkJoinTask<?>[] a; int m, s, n; - if (t != null && - (z = submitters.get()) != null && - (p = common) != null && - (ws = p.workQueues) != null && - (m = ws.length - 1) >= 0 && - (q = ws[m & z.seed & SQMASK]) != null && - (a = q.array) != null) { - int am = a.length - 1; - if ((s = q.top) != q.base) { - long j = ((am & (s - 1)) << ASHIFT) + ABASE; - if (U.getObject(a, j) == t && - U.compareAndSwapInt(q, QLOCK, 0, 1)) { - if (q.array == a && q.top == s && - U.compareAndSwapObject(a, j, t, null)) { - q.top = s - 1; - q.qlock = 0; - t.doExec(); - } - else - q.qlock = 0; - } - } - if (t.status >= 0) { - if (t instanceof CountedCompleter) - p.externalHelpComplete(q, t); - else - p.helpSignal(t, q.poolIndex); - } - } - } - - // 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(Math.min(MAX_CAP, Runtime.getRuntime().availableProcessors()), - defaultForkJoinWorkerThreadFactory, null, false); - } - - /** - * Creates a {@code ForkJoinPool} with the indicated parallelism - * level, the {@linkplain - * #defaultForkJoinWorkerThreadFactory default thread factory}, - * no UncaughtExceptionHandler, and non-async LIFO processing mode. - * - * @param parallelism the parallelism level - * @throws IllegalArgumentException if parallelism less than or - * equal to zero, or greater than implementation limit - * @throws SecurityException if a security manager exists and - * the caller is not permitted to modify threads - * because it does not hold {@link - * java.lang.RuntimePermission}{@code ("modifyThread")} - */ - public ForkJoinPool(int parallelism) { - this(parallelism, defaultForkJoinWorkerThreadFactory, null, false); - } - - /** - * Creates a {@code ForkJoinPool} with the given parameters. - * - * @param parallelism the parallelism level. For default value, - * use {@link java.lang.Runtime#availableProcessors}. - * @param factory the factory for creating new threads. For default value, - * use {@link #defaultForkJoinWorkerThreadFactory}. - * @param handler the handler for internal worker threads that - * terminate due to unrecoverable errors encountered while executing - * tasks. For default value, use {@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 (factory == null) - throw new NullPointerException(); - if (parallelism <= 0 || parallelism > MAX_CAP) - throw new IllegalArgumentException(); - this.factory = factory; - this.ueh = handler; - this.config = parallelism | (asyncMode ? (FIFO_QUEUE << 16) : 0); - long np = (long)(-parallelism); // offset ctl counts - this.ctl = ((np << AC_SHIFT) & AC_MASK) | ((np << TC_SHIFT) & TC_MASK); - int pn = nextPoolId(); - StringBuilder sb = new StringBuilder("ForkJoinPool-"); - sb.append(Integer.toString(pn)); - sb.append("-worker-"); - this.workerNamePrefix = sb.toString(); - } - - /** - * Constructor for common pool, suitable only for static initialization. - * Basically the same as above, but uses smallest possible initial footprint. - */ - ForkJoinPool(int parallelism, long ctl, - ForkJoinWorkerThreadFactory factory, - Thread.UncaughtExceptionHandler handler) { - this.config = parallelism; - this.ctl = ctl; - this.factory = factory; - this.ueh = handler; - this.workerNamePrefix = "ForkJoinPool.commonPool-worker-"; - } - - /** - * Returns the common pool instance. This pool is statically - * constructed; its run state is unaffected by attempts to {@link - * #shutdown} or {@link #shutdownNow}. However this pool and any - * ongoing processing are automatically terminated upon program - * {@link System#exit}. Any program that relies on asynchronous - * task processing to complete before program termination should - * invoke {@code commonPool().}{@link #awaitQuiescence}, before - * exit. - * - * @return the common pool instance - * @since 1.8 - */ - public static ForkJoinPool commonPool() { - // assert common != null : "static init error"; - return common; - } - - // Execution methods - - /** - * 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. - * - * @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 <T> T invoke(ForkJoinTask<T> task) { - if (task == null) - throw new NullPointerException(); - externalPush(task); - return task.join(); - } - - /** - * Arranges for (asynchronous) execution of the given task. - * - * @param task the task - * @throws NullPointerException if the task is null - * @throws RejectedExecutionException if the task cannot be - * scheduled for execution - */ - public void execute(ForkJoinTask<?> task) { - if (task == null) - throw new NullPointerException(); - externalPush(task); - } - - // AbstractExecutorService methods - - /** - * @throws NullPointerException if the task is null - * @throws RejectedExecutionException if the task cannot be - * scheduled for execution - */ - public void execute(Runnable task) { - if (task == null) - throw new NullPointerException(); - ForkJoinTask<?> job; - if (task instanceof ForkJoinTask<?>) // avoid re-wrap - job = (ForkJoinTask<?>) task; - else - job = new ForkJoinTask.AdaptedRunnableAction(task); - externalPush(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 <T> ForkJoinTask<T> submit(ForkJoinTask<T> task) { - if (task == null) - throw new NullPointerException(); - externalPush(task); - return task; - } - - /** - * @throws NullPointerException if the task is null - * @throws RejectedExecutionException if the task cannot be - * scheduled for execution - */ - public <T> ForkJoinTask<T> submit(Callable<T> task) { - ForkJoinTask<T> job = new ForkJoinTask.AdaptedCallable<T>(task); - externalPush(job); - return job; - } - - /** - * @throws NullPointerException if the task is null - * @throws RejectedExecutionException if the task cannot be - * scheduled for execution - */ - public <T> ForkJoinTask<T> submit(Runnable task, T result) { - ForkJoinTask<T> job = new ForkJoinTask.AdaptedRunnable<T>(task, result); - externalPush(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 = new ForkJoinTask.AdaptedRunnableAction(task); - externalPush(job); - return job; - } - - /** - * @throws NullPointerException {@inheritDoc} - * @throws RejectedExecutionException {@inheritDoc} - */ - public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks) { - // In previous versions of this class, this method constructed - // a task to run ForkJoinTask.invokeAll, but now external - // invocation of multiple tasks is at least as efficient. - ArrayList<Future<T>> futures = new ArrayList<Future<T>>(tasks.size()); - - boolean done = false; - try { - for (Callable<T> t : tasks) { - ForkJoinTask<T> f = new ForkJoinTask.AdaptedCallable<T>(t); - futures.add(f); - externalPush(f); - } - for (int i = 0, size = futures.size(); i < size; i++) - ((ForkJoinTask<?>)futures.get(i)).quietlyJoin(); - done = true; - return futures; - } finally { - if (!done) - for (int i = 0, size = futures.size(); i < size; i++) - futures.get(i).cancel(false); - } - } - - /** - * Returns the factory used for constructing new workers. - * - * @return the factory used for constructing new workers - */ - public ForkJoinWorkerThreadFactory getFactory() { - return factory; - } - - /** - * Returns the handler for internal worker threads that terminate - * due to unrecoverable errors encountered while executing tasks. - * - * @return the handler, or {@code null} if none - */ - public Thread.UncaughtExceptionHandler getUncaughtExceptionHandler() { - return ueh; - } - - /** - * Returns the targeted parallelism level of this pool. - * - * @return the targeted parallelism level of this pool - */ - public int getParallelism() { - return config & SMASK; - } - - /** - * Returns the targeted parallelism level of the common pool. - * - * @return the targeted parallelism level of the common pool - * @since 1.8 - */ - public static int getCommonPoolParallelism() { - return commonParallelism; - } - - /** - * 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 (config & SMASK) + (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 {@code true} if this pool uses async mode - */ - public boolean getAsyncMode() { - return (config >>> 16) == FIFO_QUEUE; - } - - /** - * Returns an estimate of the number of worker threads that are - * not blocked waiting to join tasks or for other managed - * synchronization. This method may overestimate the - * number of running threads. - * - * @return the number of worker threads - */ - public int getRunningThreadCount() { - int rc = 0; - WorkQueue[] ws; WorkQueue w; - if ((ws = workQueues) != null) { - for (int i = 1; i < ws.length; i += 2) { - if ((w = ws[i]) != null && w.isApparentlyUnblocked()) - ++rc; - } - } - return rc; - } - - /** - * 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 - */ - public int getActiveThreadCount() { - int r = (config & SMASK) + (int)(ctl >> AC_SHIFT); - return (r <= 0) ? 0 : r; // suppress momentarily negative values - } - - /** - * 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 (int)(ctl >> AC_SHIFT) + (config & SMASK) == 0; - } - - /** - * Returns an estimate of the total number of tasks stolen from - * 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 - * high enough to keep threads busy, but low enough to avoid - * overhead and contention across threads. - * - * @return the number of steals - */ - public long getStealCount() { - long count = stealCount; - WorkQueue[] ws; WorkQueue w; - if ((ws = workQueues) != null) { - for (int i = 1; i < ws.length; i += 2) { - if ((w = ws[i]) != null) - count += w.nsteals; - } - } - return count; - } - - /** - * Returns an estimate of the total number of tasks currently held - * in queues by worker threads (but not including tasks submitted - * to the pool that have not begun executing). This value is only - * 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 - */ - public long getQueuedTaskCount() { - long count = 0; - WorkQueue[] ws; WorkQueue w; - if ((ws = workQueues) != null) { - for (int i = 1; i < ws.length; i += 2) { - if ((w = ws[i]) != null) - count += w.queueSize(); - } - } - return count; - } - - /** - * 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() { - int count = 0; - WorkQueue[] ws; WorkQueue w; - if ((ws = workQueues) != null) { - for (int i = 0; i < ws.length; i += 2) { - if ((w = ws[i]) != null) - count += w.queueSize(); - } - } - return count; - } - - /** - * 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() { - WorkQueue[] ws; WorkQueue w; - if ((ws = workQueues) != null) { - for (int i = 0; i < ws.length; i += 2) { - if ((w = ws[i]) != null && !w.isEmpty()) - return true; - } - } - return false; - } - - /** - * 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 {@code null} if none - */ - protected ForkJoinTask<?> pollSubmission() { - WorkQueue[] ws; WorkQueue w; ForkJoinTask<?> t; - if ((ws = workQueues) != null) { - for (int i = 0; i < ws.length; i += 2) { - if ((w = ws[i]) != null && (t = w.poll()) != null) - 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 - * artificially generated or wrapped tasks. This method is - * designed to be invoked only when the pool is known to be - * quiescent. Invocations at other times may not remove all - * tasks. A failure encountered while attempting to add elements - * to collection {@code c} may result in elements being in - * 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<? super ForkJoinTask<?>> c) { - int count = 0; - WorkQueue[] ws; WorkQueue w; ForkJoinTask<?> t; - if ((ws = workQueues) != null) { - for (int i = 0; i < ws.length; ++i) { - if ((w = ws[i]) != null) { - while ((t = w.poll()) != null) { - c.add(t); - ++count; - } - } - } - } - return count; - } - - /** - * Returns a string identifying this pool, as well as its state, - * including indications of run state, parallelism level, and - * worker and task counts. - * - * @return a string identifying this pool, as well as its state - */ - public String toString() { - // Use a single pass through workQueues to collect counts - long qt = 0L, qs = 0L; int rc = 0; - long st = stealCount; - long c = ctl; - WorkQueue[] ws; WorkQueue w; - if ((ws = workQueues) != null) { - for (int i = 0; i < ws.length; ++i) { - if ((w = ws[i]) != null) { - int size = w.queueSize(); - if ((i & 1) == 0) - qs += size; - else { - qt += size; - st += w.nsteals; - if (w.isApparentlyUnblocked()) - ++rc; - } - } - } - } - int pc = (config & SMASK); - int tc = pc + (short)(c >>> TC_SHIFT); - int ac = pc + (int)(c >> AC_SHIFT); - if (ac < 0) // ignore transient negative - ac = 0; - String level; - if ((c & STOP_BIT) != 0) - level = (tc == 0) ? "Terminated" : "Terminating"; - else - level = plock < 0 ? "Shutting down" : "Running"; - return super.toString() + - "[" + level + - ", parallelism = " + pc + - ", size = " + tc + - ", active = " + ac + - ", running = " + rc + - ", steals = " + st + - ", tasks = " + qt + - ", submissions = " + qs + - "]"; - } - - /** - * Possibly initiates an orderly shutdown in which previously - * submitted tasks are executed, but no new tasks will be - * accepted. Invocation has no effect on execution state if this - * is the {@link #commonPool()}, and no additional effect if - * already shut down. Tasks that are in the process of being - * submitted concurrently during the course of this method may or - * may not be rejected. - * - * @throws SecurityException if a security manager exists and - * the caller is not permitted to modify threads - * because it does not hold {@link - * java.lang.RuntimePermission}{@code ("modifyThread")} - */ - public void shutdown() { - checkPermission(); - tryTerminate(false, true); - } - - /** - * Possibly attempts to cancel and/or stop all tasks, and reject - * all subsequently submitted tasks. Invocation has no effect on - * execution state if this is the {@link #commonPool()}, and no - * additional effect if already shut down. Otherwise, tasks that - * are in the process of being submitted or executed concurrently - * during the course of this method may or may not be - * rejected. This method cancels both existing and unexecuted - * tasks, in order to permit termination in the presence of task - * dependencies. So the method always returns an empty list - * (unlike the case for some other Executors). - * - * @return an empty list - * @throws SecurityException if a security manager exists and - * the caller is not permitted to modify threads - * because it does not hold {@link - * java.lang.RuntimePermission}{@code ("modifyThread")} - */ - public List<Runnable> shutdownNow() { - checkPermission(); - tryTerminate(true, true); - return Collections.emptyList(); - } - - /** - * Returns {@code true} if all tasks have completed following shut down. - * - * @return {@code true} if all tasks have completed following shut down - */ - public boolean isTerminated() { - long c = ctl; - return ((c & STOP_BIT) != 0L && - (short)(c >>> TC_SHIFT) == -(config & SMASK)); - } - - /** - * 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 I/O, - * causing this executor not to properly terminate. (See the - * advisory notes for class {@link ForkJoinTask} stating that - * tasks should not normally entail blocking operations. But if - * they do, they must abort them on interrupt.) - * - * @return {@code true} if terminating but not yet terminated - */ - public boolean isTerminating() { - long c = ctl; - return ((c & STOP_BIT) != 0L && - (short)(c >>> TC_SHIFT) != -(config & SMASK)); - } - - /** - * Returns {@code true} if this pool has been shut down. - * - * @return {@code true} if this pool has been shut down - */ - public boolean isShutdown() { - return plock < 0; - } - - /** - * Blocks until all tasks have completed execution after a - * shutdown request, or the timeout occurs, or the current thread - * is interrupted, whichever happens first. Because the {@link - * #commonPool()} never terminates until program shutdown, when - * applied to the common pool, this method is equivalent to {@link - * #awaitQuiescence} but always returns {@code false}. - * - * @param timeout the maximum time to wait - * @param unit the time unit of the timeout argument - * @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 { - if (Thread.interrupted()) - throw new InterruptedException(); - if (this == common) { - awaitQuiescence(timeout, unit); - return false; - } - long nanos = unit.toNanos(timeout); - if (isTerminated()) - return true; - long startTime = System.nanoTime(); - boolean terminated = false; - synchronized (this) { - for (long waitTime = nanos, millis = 0L;;) { - if (terminated = isTerminated() || - waitTime <= 0L || - (millis = unit.toMillis(waitTime)) <= 0L) - break; - wait(millis); - waitTime = nanos - (System.nanoTime() - startTime); - } - } - return terminated; - } - - /** - * If called by a ForkJoinTask operating in this pool, equivalent - * in effect to {@link ForkJoinTask#helpQuiesce}. Otherwise, - * waits and/or attempts to assist performing tasks until this - * pool {@link #isQuiescent} or the indicated timeout elapses. - * - * @param timeout the maximum time to wait - * @param unit the time unit of the timeout argument - * @return {@code true} if quiescent; {@code false} if the - * timeout elapsed. - */ - public boolean awaitQuiescence(long timeout, TimeUnit unit) { - long nanos = unit.toNanos(timeout); - ForkJoinWorkerThread wt; - Thread thread = Thread.currentThread(); - if ((thread instanceof ForkJoinWorkerThread) && - (wt = (ForkJoinWorkerThread)thread).pool == this) { - helpQuiescePool(wt.workQueue); - return true; - } - long startTime = System.nanoTime(); - WorkQueue[] ws; - int r = 0, m; - boolean found = true; - while (!isQuiescent() && (ws = workQueues) != null && - (m = ws.length - 1) >= 0) { - if (!found) { - if ((System.nanoTime() - startTime) > nanos) - return false; - Thread.yield(); // cannot block - } - found = false; - for (int j = (m + 1) << 2; j >= 0; --j) { - ForkJoinTask<?> t; WorkQueue q; int b; - if ((q = ws[r++ & m]) != null && (b = q.base) - q.top < 0) { - found = true; - if ((t = q.pollAt(b)) != null) { - if (q.base - q.top < 0) - signalWork(q); - t.doExec(); - } - break; - } - } - } - return true; - } - - /** - * Waits and/or attempts to assist performing tasks indefinitely - * until the {@link #commonPool()} {@link #isQuiescent}. - */ - static void quiesceCommonPool() { - common.awaitQuiescence(Long.MAX_VALUE, TimeUnit.NANOSECONDS); - } - - /** - * Interface for extending managed parallelism for tasks running - * in {@link ForkJoinPool}s. - * - * <p>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. - * - * <p>For example, here is a ManagedBlocker based on a - * ReentrantLock: - * <pre> {@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()); - * } - * }}</pre> - * - * <p>Here is a class that possibly blocks waiting for an - * item on a given queue: - * <pre> {@code - * class QueueTaker<E> implements ManagedBlocker { - * final BlockingQueue<E> queue; - * volatile E item = null; - * QueueTaker(BlockingQueue<E> 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; - * } - * }}</pre> - */ - public static interface ManagedBlocker { - /** - * Possibly blocks the current thread, for example waiting for - * a lock or condition. - * - * @return {@code true} if no additional blocking is necessary - * (i.e., if isReleasable would return true) - * @throws InterruptedException if interrupted while waiting - * (the method is not required to do so, but is allowed to) - */ - boolean block() throws InterruptedException; - - /** - * Returns {@code true} if blocking is unnecessary. - */ - boolean isReleasable(); - } - - /** - * Blocks in accord with the given blocker. If the current thread - * 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. - * - * <p>If the caller is not a {@link ForkJoinTask}, this method is - * behaviorally equivalent to - * <pre> {@code - * while (!blocker.isReleasable()) - * if (blocker.block()) - * return; - * }</pre> - * - * If the caller is a {@code ForkJoinTask}, then the pool may - * first be expanded to ensure parallelism, and later adjusted. - * - * @param blocker the blocker - * @throws InterruptedException if blocker.block did so - */ - public static void managedBlock(ManagedBlocker blocker) - throws InterruptedException { - Thread t = Thread.currentThread(); - if (t instanceof ForkJoinWorkerThread) { - ForkJoinPool p = ((ForkJoinWorkerThread)t).pool; - while (!blocker.isReleasable()) { // variant of helpSignal - WorkQueue[] ws; WorkQueue q; int m, u; - if ((ws = p.workQueues) != null && (m = ws.length - 1) >= 0) { - for (int i = 0; i <= m; ++i) { - if (blocker.isReleasable()) - return; - if ((q = ws[i]) != null && q.base - q.top < 0) { - p.signalWork(q); - if ((u = (int)(p.ctl >>> 32)) >= 0 || - (u >> UAC_SHIFT) >= 0) - break; - } - } - } - if (p.tryCompensate()) { - try { - do {} while (!blocker.isReleasable() && - !blocker.block()); - } finally { - p.incrementActiveCount(); - } - break; - } - } - } - else { - do {} while (!blocker.isReleasable() && - !blocker.block()); - } - } - - // AbstractExecutorService overrides. These rely on undocumented - // fact that ForkJoinTask.adapt returns ForkJoinTasks that also - // implement RunnableFuture. - - protected <T> RunnableFuture<T> newTaskFor(Runnable runnable, T value) { - return new ForkJoinTask.AdaptedRunnable<T>(runnable, value); - } - - protected <T> RunnableFuture<T> newTaskFor(Callable<T> callable) { - return new ForkJoinTask.AdaptedCallable<T>(callable); - } - - // Unsafe mechanics - private static final sun.misc.Unsafe U; - private static final long CTL; - private static final long PARKBLOCKER; - private static final int ABASE; - private static final int ASHIFT; - private static final long STEALCOUNT; - private static final long PLOCK; - private static final long INDEXSEED; - private static final long QLOCK; - - static { - // initialize field offsets for CAS etc - try { - U = getUnsafe(); - Class<?> k = ForkJoinPool.class; - CTL = U.objectFieldOffset - (k.getDeclaredField("ctl")); - STEALCOUNT = U.objectFieldOffset - (k.getDeclaredField("stealCount")); - PLOCK = U.objectFieldOffset - (k.getDeclaredField("plock")); - INDEXSEED = U.objectFieldOffset - (k.getDeclaredField("indexSeed")); - Class<?> tk = Thread.class; - PARKBLOCKER = U.objectFieldOffset - (tk.getDeclaredField("parkBlocker")); - Class<?> wk = WorkQueue.class; - QLOCK = U.objectFieldOffset - (wk.getDeclaredField("qlock")); - Class<?> ak = ForkJoinTask[].class; - ABASE = U.arrayBaseOffset(ak); - int scale = U.arrayIndexScale(ak); - if ((scale & (scale - 1)) != 0) - throw new Error("data type scale not a power of two"); - ASHIFT = 31 - Integer.numberOfLeadingZeros(scale); - } catch (Exception e) { - throw new Error(e); - } - - submitters = new ThreadLocal<Submitter>(); - ForkJoinWorkerThreadFactory fac = defaultForkJoinWorkerThreadFactory = - new DefaultForkJoinWorkerThreadFactory(); - modifyThreadPermission = new RuntimePermission("modifyThread"); - - /* - * Establish common pool parameters. For extra caution, - * computations to set up common pool state are here; the - * constructor just assigns these values to fields. - */ - - int par = 0; - Thread.UncaughtExceptionHandler handler = null; - try { // TBD: limit or report ignored exceptions? - String pp = System.getProperty - ("java.util.concurrent.ForkJoinPool.common.parallelism"); - String hp = System.getProperty - ("java.util.concurrent.ForkJoinPool.common.exceptionHandler"); - String fp = System.getProperty - ("java.util.concurrent.ForkJoinPool.common.threadFactory"); - if (fp != null) - fac = ((ForkJoinWorkerThreadFactory)ClassLoader. - getSystemClassLoader().loadClass(fp).newInstance()); - if (hp != null) - handler = ((Thread.UncaughtExceptionHandler)ClassLoader. - getSystemClassLoader().loadClass(hp).newInstance()); - if (pp != null) - par = Integer.parseInt(pp); - } catch (Exception ignore) { - } - - if (par <= 0) - par = Runtime.getRuntime().availableProcessors(); - if (par > MAX_CAP) - par = MAX_CAP; - commonParallelism = par; - long np = (long)(-par); // precompute initial ctl value - long ct = ((np << AC_SHIFT) & AC_MASK) | ((np << TC_SHIFT) & TC_MASK); - - common = new ForkJoinPool(par, ct, fac, handler); - } - - /** - * Returns a sun.misc.Unsafe. Suitable for use in a 3rd party package. - * Replace with a simple call to Unsafe.getUnsafe when integrating - * into a jdk. - * - * @return a sun.misc.Unsafe - */ - private static sun.misc.Unsafe getUnsafe() { - return scala.concurrent.util.Unsafe.instance; - } -} diff --git a/src/forkjoin/scala/concurrent/forkjoin/ForkJoinTask.java b/src/forkjoin/scala/concurrent/forkjoin/ForkJoinTask.java deleted file mode 100644 index fd1e132b07..0000000000 --- a/src/forkjoin/scala/concurrent/forkjoin/ForkJoinTask.java +++ /dev/null @@ -1,1488 +0,0 @@ -/* - * 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/publicdomain/zero/1.0/ - */ - -package scala.concurrent.forkjoin; - -import java.io.Serializable; -import java.util.Collection; -import java.util.List; -import java.util.RandomAccess; -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.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 {@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. - * - * <p>A "main" {@code ForkJoinTask} begins execution when it is - * explicitly submitted to a {@link ForkJoinPool}, or, if not already - * engaged in a ForkJoin computation, commenced in the {@link - * ForkJoinPool#commonPool()} via {@link #fork}, {@link #invoke}, or - * related methods. 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. - * - * <p>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 main 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 - * ideally 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. Subdividable tasks should also - * not perform blocking I/O, and should ideally access variables that - * are completely independent of those accessed by other running - * tasks. These guidelines are loosely 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. - * - * <p>It is possible to define and use ForkJoinTasks that may block, - * but doing do requires three further considerations: (1) Completion - * of few if any <em>other</em> tasks should be dependent on a task - * that blocks on external synchronization or I/O. Event-style async - * tasks that are never joined (for example, those subclassing {@link - * CountedCompleter}) often fall into this category. (2) To minimize - * resource impact, tasks should be small; ideally performing only the - * (possibly) blocking action. (3) Unless the {@link - * ForkJoinPool.ManagedBlocker} API is used, or the number of possibly - * blocked tasks is known to be less than the pool's {@link - * ForkJoinPool#getParallelism} level, the pool cannot guarantee that - * enough threads will be available to ensure progress or good - * performance. - * - * <p>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 {@code Future} - * conventions. Method {@link #invoke} is semantically - * equivalent to {@code fork(); join()} but always attempts to begin - * execution in the current thread. The "<em>quiet</em>" 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 {@code invokeAll} (available in multiple versions) - * performs the most common form of parallel invocation: forking a set - * of tasks and joining them all. - * - * <p>In the most typical usages, a fork-join pair act like a call - * (fork) and return (join) from a parallel recursive function. As is - * the case with other forms of recursive calls, returns (joins) - * should be performed innermost-first. For example, {@code a.fork(); - * b.fork(); b.join(); a.join();} is likely to be substantially more - * efficient than joining {@code a} before {@code b}. - * - * <p>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}. - * - * <p>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, typically {@link - * RecursiveAction} for most computations that do not return results, - * {@link RecursiveTask} for those that do, and {@link - * CountedCompleter} for those in which completed actions trigger - * other actions. Normally, a concrete ForkJoinTask subclass declares - * fields comprising its parameters, established in a constructor, and - * then defines a {@code compute} method that somehow uses the control - * methods supplied by this base class. - * - * <p>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. To support such usages a - * ForkJoinTask may be atomically <em>tagged</em> with a {@code short} - * value using {@link #setForkJoinTaskTag} or {@link - * #compareAndSetForkJoinTaskTag} and checked using {@link - * #getForkJoinTaskTag}. The ForkJoinTask implementation does not use - * these {@code protected} methods or tags for any purpose, but they - * may be of use in the construction of specialized subclasses. For - * example, parallel graph traversals can use the supplied methods to - * avoid revisiting nodes/tasks that have already been processed. - * (Method names for tagging are bulky in part to encourage definition - * of methods that reflect their usage patterns.) - * - * <p>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. - * - * <p>ForkJoinTasks should perform relatively small amounts of - * 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. - * - * <p>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 <em>asyncMode</em>. - * - * <p>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. - * - * @since 1.7 - * @author Doug Lea - */ -public abstract class ForkJoinTask<V> implements Future<V>, Serializable { - - /* - * 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. - */ - - /* - * 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 (anded with - * DONE_MASK) 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, mainly by arranging that every synchronized - * block performs a wait, notifyAll or both. - * - * These control bits occupy only (some of) the upper half (16 - * bits) of status field. The lower bits are used for user-defined - * tags. - */ - - /** The run status of this task */ - volatile int status; // accessed directly by pool and workers - static final int DONE_MASK = 0xf0000000; // mask out non-completion bits - static final int NORMAL = 0xf0000000; // must be negative - static final int CANCELLED = 0xc0000000; // must be < NORMAL - static final int EXCEPTIONAL = 0x80000000; // must be < CANCELLED - static final int SIGNAL = 0x00010000; // must be >= 1 << 16 - static final int SMASK = 0x0000ffff; // short bits for tags - - /** - * Marks completion and wakes up threads waiting to join this - * task. - * - * @param completion one of NORMAL, CANCELLED, EXCEPTIONAL - * @return completion status on exit - */ - private int setCompletion(int completion) { - for (int s;;) { - if ((s = status) < 0) - return s; - if (U.compareAndSwapInt(this, STATUS, s, s | completion)) { - if ((s >>> 16) != 0) - synchronized (this) { notifyAll(); } - return completion; - } - } - } - - /** - * Primary execution method for stolen tasks. Unless done, calls - * exec and records status if completed, but doesn't wait for - * completion otherwise. - * - * @return status on exit from this method - */ - final int doExec() { - int s; boolean completed; - if ((s = status) >= 0) { - try { - completed = exec(); - } catch (Throwable rex) { - return setExceptionalCompletion(rex); - } - if (completed) - s = setCompletion(NORMAL); - } - return s; - } - - /** - * Tries to set SIGNAL status unless already completed. Used by - * ForkJoinPool. Other variants are directly incorporated into - * externalAwaitDone etc. - * - * @return true if successful - */ - final boolean trySetSignal() { - int s = status; - return s >= 0 && U.compareAndSwapInt(this, STATUS, s, s | SIGNAL); - } - - /** - * Blocks a non-worker-thread until completion. - * @return status upon completion - */ - private int externalAwaitDone() { - int s; - ForkJoinPool.externalHelpJoin(this); - boolean interrupted = false; - while ((s = status) >= 0) { - if (U.compareAndSwapInt(this, STATUS, s, s | SIGNAL)) { - synchronized (this) { - if (status >= 0) { - try { - wait(); - } catch (InterruptedException ie) { - interrupted = true; - } - } - else - notifyAll(); - } - } - } - if (interrupted) - Thread.currentThread().interrupt(); - return s; - } - - /** - * Blocks a non-worker-thread until completion or interruption. - */ - private int externalInterruptibleAwaitDone() throws InterruptedException { - int s; - if (Thread.interrupted()) - throw new InterruptedException(); - ForkJoinPool.externalHelpJoin(this); - while ((s = status) >= 0) { - if (U.compareAndSwapInt(this, STATUS, s, s | SIGNAL)) { - synchronized (this) { - if (status >= 0) - wait(); - else - notifyAll(); - } - } - } - return s; - } - - - /** - * Implementation for join, get, quietlyJoin. Directly handles - * only cases of already-completed, external wait, and - * unfork+exec. Others are relayed to ForkJoinPool.awaitJoin. - * - * @return status upon completion - */ - private int doJoin() { - int s; Thread t; ForkJoinWorkerThread wt; ForkJoinPool.WorkQueue w; - return (s = status) < 0 ? s : - ((t = Thread.currentThread()) instanceof ForkJoinWorkerThread) ? - (w = (wt = (ForkJoinWorkerThread)t).workQueue). - tryUnpush(this) && (s = doExec()) < 0 ? s : - wt.pool.awaitJoin(w, this) : - externalAwaitDone(); - } - - /** - * Implementation for invoke, quietlyInvoke. - * - * @return status upon completion - */ - private int doInvoke() { - int s; Thread t; ForkJoinWorkerThread wt; - return (s = doExec()) < 0 ? s : - ((t = Thread.currentThread()) instanceof ForkJoinWorkerThread) ? - (wt = (ForkJoinWorkerThread)t).pool.awaitJoin(wt.workQueue, this) : - externalAwaitDone(); - } - - // Exception table support - - /** - * 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<Object> exceptionTableRefQueue; - - /** - * Fixed capacity for exceptionTable. - */ - private static final int EXCEPTION_MAP_CAPACITY = 32; - - /** - * 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. - */ - static final class ExceptionNode extends WeakReference<ForkJoinTask<?>> { - 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(); - } - } - - /** - * Records exception and sets status. - * - * @return status on exit - */ - final int recordExceptionalCompletion(Throwable ex) { - int s; - if ((s = status) >= 0) { - 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(); - } - s = setCompletion(EXCEPTIONAL); - } - return s; - } - - /** - * Records exception and possibly propagates. - * - * @return status on exit - */ - private int setExceptionalCompletion(Throwable ex) { - int s = recordExceptionalCompletion(ex); - if ((s & DONE_MASK) == EXCEPTIONAL) - internalPropagateException(ex); - return s; - } - - /** - * Hook for exception propagation support for tasks with completers. - */ - void internalPropagateException(Throwable ex) { - } - - /** - * 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. - */ - static final void cancelIgnoringExceptions(ForkJoinTask<?> t) { - if (t != null && t.status >= 0) { - try { - t.cancel(false); - } catch (Throwable ignore) { - } - } - } - - /** - * Removes exception node and clears status. - */ - 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(); - } - } - - /** - * 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 - */ - private Throwable getThrowableException() { - if ((status & DONE_MASK) != 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 (false && e.thrower != Thread.currentThread().getId()) { - Class<? extends Throwable> ec = ex.getClass(); - try { - 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) { - } - } - return ex; - } - - /** - * Poll stale refs and remove them. Call only while holding lock. - */ - 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; - } - } - } - } - - /** - * If lock is available, poll stale refs and remove them. - * Called from ForkJoinPool when pools become quiescent. - */ - static final void helpExpungeStaleExceptions() { - final ReentrantLock lock = exceptionTableLock; - if (lock.tryLock()) { - try { - expungeStaleExceptions(); - } finally { - lock.unlock(); - } - } - } - - /** - * A version of "sneaky throw" to relay exceptions - */ - static void rethrow(final Throwable ex) { - if (ex != null) { - if (ex instanceof Error) - throw (Error)ex; - if (ex instanceof RuntimeException) - throw (RuntimeException)ex; - ForkJoinTask.<RuntimeException>uncheckedThrow(ex); - } - } - - /** - * The sneaky part of sneaky throw, relying on generics - * limitations to evade compiler complaints about rethrowing - * unchecked exceptions - */ - @SuppressWarnings("unchecked") static <T extends Throwable> - void uncheckedThrow(Throwable t) throws T { - if (t != null) - throw (T)t; // rely on vacuous cast - } - - /** - * Throws exception, if any, associated with the given status. - */ - private void reportException(int s) { - if (s == CANCELLED) - throw new CancellationException(); - if (s == EXCEPTIONAL) - rethrow(getThrowableException()); - } - - // public methods - - /** - * Arranges to asynchronously execute this task in the pool the - * current task is running in, if applicable, or using the {@link - * ForkJoinPool#commonPool()} if not {@link #inForkJoinPool}. 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. 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}. - * - * @return {@code this}, to simplify usage - */ - public final ForkJoinTask<V> fork() { - Thread t; - if ((t = Thread.currentThread()) instanceof ForkJoinWorkerThread) - ((ForkJoinWorkerThread)t).workQueue.push(this); - else - ForkJoinPool.common.externalPush(this); - return this; - } - - /** - * 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 <em>not</em> cause the - * method to abruptly return by throwing {@code - * InterruptedException}. - * - * @return the computed result - */ - public final V join() { - int s; - if ((s = doJoin() & DONE_MASK) != NORMAL) - reportException(s); - return getRawResult(); - } - - /** - * Commences performing this task, awaits its completion if - * 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() { - int s; - if ((s = doInvoke() & DONE_MASK) != NORMAL) - reportException(s); - return getRawResult(); - } - - /** - * 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. - * - * @param t1 the first task - * @param t2 the second task - * @throws NullPointerException if any task is null - */ - public static void invokeAll(ForkJoinTask<?> t1, ForkJoinTask<?> t2) { - int s1, s2; - t2.fork(); - if ((s1 = t1.doInvoke() & DONE_MASK) != NORMAL) - t1.reportException(s1); - if ((s2 = t2.doJoin() & DONE_MASK) != NORMAL) - t2.reportException(s2); - } - - /** - * 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. - * - * @param tasks the tasks - * @throws NullPointerException if any task is null - */ - public static void invokeAll(ForkJoinTask<?>... tasks) { - Throwable ex = null; - int last = tasks.length - 1; - for (int i = last; i >= 0; --i) { - ForkJoinTask<?> t = tasks[i]; - if (t == null) { - if (ex == null) - ex = new NullPointerException(); - } - else if (i != 0) - t.fork(); - 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 if (t.doJoin() < NORMAL) - ex = t.getException(); - } - } - if (ex != null) - rethrow(ex); - } - - /** - * 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. - * - * @param tasks the collection of tasks - * @return the tasks argument, to simplify usage - * @throws NullPointerException if tasks or any element are null - */ - public static <T extends ForkJoinTask<?>> Collection<T> invokeAll(Collection<T> tasks) { - if (!(tasks instanceof RandomAccess) || !(tasks instanceof List<?>)) { - invokeAll(tasks.toArray(new ForkJoinTask<?>[tasks.size()])); - return tasks; - } - @SuppressWarnings("unchecked") - List<? extends ForkJoinTask<?>> ts = - (List<? extends ForkJoinTask<?>>) tasks; - Throwable ex = null; - int last = ts.size() - 1; - for (int i = last; i >= 0; --i) { - ForkJoinTask<?> t = ts.get(i); - if (t == null) { - if (ex == null) - ex = new NullPointerException(); - } - else if (i != 0) - t.fork(); - 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 if (t.doJoin() < NORMAL) - ex = t.getException(); - } - } - if (ex != null) - rethrow(ex); - return tasks; - } - - /** - * 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}. - * - * <p>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. - * - * <p>This method is designed to be invoked by <em>other</em> - * 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 boolean cancel(boolean mayInterruptIfRunning) { - return (setCompletion(CANCELLED) & DONE_MASK) == CANCELLED; - } - - public final boolean isDone() { - return status < 0; - } - - public final boolean isCancelled() { - return (status & DONE_MASK) == CANCELLED; - } - - /** - * Returns {@code true} if this task threw an exception or was cancelled. - * - * @return {@code true} if this task threw an exception or was cancelled - */ - public final boolean isCompletedAbnormally() { - return status < NORMAL; - } - - /** - * 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 isCompletedNormally() { - return (status & DONE_MASK) == NORMAL; - } - - /** - * Returns the exception thrown by the base computation, or a - * {@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 & DONE_MASK; - 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 - * {@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 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 - * {@code RuntimeException} or {@code Error}, the actual exception - * thrown will be a {@code RuntimeException} with cause {@code ex}. - */ - public void completeExceptionally(Throwable ex) { - setExceptionalCompletion((ex instanceof RuntimeException) || - (ex instanceof Error) ? ex : - new RuntimeException(ex)); - } - - /** - * Completes this task, and if not already aborted or cancelled, - * 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 - */ - public void complete(V value) { - try { - setRawResult(value); - } catch (Throwable rex) { - setExceptionalCompletion(rex); - return; - } - setCompletion(NORMAL); - } - - /** - * Completes this task normally without setting a value. The most - * recent value established by {@link #setRawResult} (or {@code - * null} by default) will be returned as the result of subsequent - * invocations of {@code join} and related operations. - * - * @since 1.8 - */ - public final void quietlyComplete() { - setCompletion(NORMAL); - } - - /** - * 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 get() throws InterruptedException, ExecutionException { - int s = (Thread.currentThread() instanceof ForkJoinWorkerThread) ? - doJoin() : externalInterruptibleAwaitDone(); - Throwable ex; - if ((s &= DONE_MASK) == CANCELLED) - throw new CancellationException(); - if (s == EXCEPTIONAL && (ex = getThrowableException()) != null) - throw new ExecutionException(ex); - return getRawResult(); - } - - /** - * 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 V get(long timeout, TimeUnit unit) - throws InterruptedException, ExecutionException, TimeoutException { - if (Thread.interrupted()) - throw new InterruptedException(); - // Messy in part because we measure in nanosecs, but wait in millisecs - int s; long ms; - long ns = unit.toNanos(timeout); - if ((s = status) >= 0 && ns > 0L) { - long deadline = System.nanoTime() + ns; - ForkJoinPool p = null; - ForkJoinPool.WorkQueue w = null; - Thread t = Thread.currentThread(); - if (t instanceof ForkJoinWorkerThread) { - ForkJoinWorkerThread wt = (ForkJoinWorkerThread)t; - p = wt.pool; - w = wt.workQueue; - p.helpJoinOnce(w, this); // no retries on failure - } - else - ForkJoinPool.externalHelpJoin(this); - boolean canBlock = false; - boolean interrupted = false; - try { - while ((s = status) >= 0) { - if (w != null && w.qlock < 0) - cancelIgnoringExceptions(this); - else if (!canBlock) { - if (p == null || p.tryCompensate()) - canBlock = true; - } - else { - if ((ms = TimeUnit.NANOSECONDS.toMillis(ns)) > 0L && - U.compareAndSwapInt(this, STATUS, s, s | SIGNAL)) { - synchronized (this) { - if (status >= 0) { - try { - wait(ms); - } catch (InterruptedException ie) { - if (p == null) - interrupted = true; - } - } - else - notifyAll(); - } - } - if ((s = status) < 0 || interrupted || - (ns = deadline - System.nanoTime()) <= 0L) - break; - } - } - } finally { - if (p != null && canBlock) - p.incrementActiveCount(); - } - if (interrupted) - throw new InterruptedException(); - } - if ((s &= DONE_MASK) != 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 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() { - doJoin(); - } - - /** - * Commences performing this task and awaits its completion if - * necessary, without returning its result or throwing its - * exception. - */ - public final void quietlyInvoke() { - doInvoke(); - } - - /** - * Possibly executes tasks until the pool hosting the current task - * {@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. - */ - public static void helpQuiesce() { - Thread t; - if ((t = Thread.currentThread()) instanceof ForkJoinWorkerThread) { - ForkJoinWorkerThread wt = (ForkJoinWorkerThread)t; - wt.pool.helpQuiescePool(wt.workQueue); - } - else - ForkJoinPool.quiesceCommonPool(); - } - - /** - * Resets the internal bookkeeping state of this task, allowing a - * 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. - * This method may be useful when executing - * pre-constructed trees of subtasks in loops. - * - * <p>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 & DONE_MASK) == EXCEPTIONAL) - clearExceptionalCompletion(); - else - status = 0; - } - - /** - * Returns the pool hosting the current task execution, or null - * 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; - } - - /** - * 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; - } - - /** - * Tries to unschedule this task for execution. This method will - * typically (but is not guaranteed to) succeed if this task is - * the most recently forked task 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. - * - * @return {@code true} if unforked - */ - public boolean tryUnfork() { - Thread t; - return (((t = Thread.currentThread()) instanceof ForkJoinWorkerThread) ? - ((ForkJoinWorkerThread)t).workQueue.tryUnpush(this) : - ForkJoinPool.tryExternalUnpush(this)); - } - - /** - * Returns an estimate of the number of tasks that have been - * forked by the current worker thread but not yet executed. This - * value may be useful for heuristic decisions about whether to - * fork other tasks. - * - * @return the number of tasks - */ - public static int getQueuedTaskCount() { - Thread t; ForkJoinPool.WorkQueue q; - if ((t = Thread.currentThread()) instanceof ForkJoinWorkerThread) - q = ((ForkJoinWorkerThread)t).workQueue; - else - q = ForkJoinPool.commonSubmitterQueue(); - return (q == null) ? 0 : q.queueSize(); - } - - /** - * 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, or zero if this thread is not - * operating in a ForkJoinPool. This value may be useful for - * heuristic decisions about whether to fork other tasks. In many - * usages of ForkJoinTasks, at steady state, each worker should - * aim to maintain a small constant surplus (for example, 3) of - * tasks, and to process computations locally if this threshold is - * exceeded. - * - * @return the surplus number of tasks, which may be negative - */ - public static int getSurplusQueuedTaskCount() { - return ForkJoinPool.getSurplusQueuedTaskCount(); - } - - // Extension methods - - /** - * 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 {@code null} if not completed - */ - public abstract V getRawResult(); - - /** - * Forces the given value to be returned as a result. This method - * is designed to support extensions, and should not in general be - * called otherwise. - * - * @param value the value - */ - protected abstract void setRawResult(V value); - - /** - * Immediately performs the base action of this task and returns - * true if, upon return from this method, this task is guaranteed - * to have completed normally. This method may return false - * otherwise, to indicate that this task is not necessarily - * complete (or is not known to be complete), for example in - * asynchronous actions that require explicit invocations of - * completion methods. This method may also throw an (unchecked) - * exception to indicate abnormal exit. This method is designed to - * support extensions, and should not in general be called - * otherwise. - * - * @return {@code true} if this task is known to have completed normally - */ - protected abstract boolean exec(); - - /** - * 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. 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 {@code null} if none are available - */ - protected static ForkJoinTask<?> peekNextLocalTask() { - Thread t; ForkJoinPool.WorkQueue q; - if ((t = Thread.currentThread()) instanceof ForkJoinWorkerThread) - q = ((ForkJoinWorkerThread)t).workQueue; - else - q = ForkJoinPool.commonSubmitterQueue(); - return (q == null) ? null : q.peek(); - } - - /** - * Unschedules and returns, without executing, the next task - * queued by the current thread but not yet executed, if the - * current thread is operating in a ForkJoinPool. This method is - * designed primarily to support extensions, and is unlikely to be - * useful otherwise. - * - * @return the next task, or {@code null} if none are available - */ - protected static ForkJoinTask<?> pollNextLocalTask() { - Thread t; - return ((t = Thread.currentThread()) instanceof ForkJoinWorkerThread) ? - ((ForkJoinWorkerThread)t).workQueue.nextLocalTask() : - null; - } - - /** - * If the current thread is operating in a ForkJoinPool, - * unschedules and returns, without executing, the next task - * 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 - * {@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. - * - * @return a task, or {@code null} if none are available - */ - protected static ForkJoinTask<?> pollTask() { - Thread t; ForkJoinWorkerThread wt; - return ((t = Thread.currentThread()) instanceof ForkJoinWorkerThread) ? - (wt = (ForkJoinWorkerThread)t).pool.nextTaskFor(wt.workQueue) : - null; - } - - // tag operations - - /** - * Returns the tag for this task. - * - * @return the tag for this task - * @since 1.8 - */ - public final short getForkJoinTaskTag() { - return (short)status; - } - - /** - * Atomically sets the tag value for this task. - * - * @param tag the tag value - * @return the previous value of the tag - * @since 1.8 - */ - public final short setForkJoinTaskTag(short tag) { - for (int s;;) { - if (U.compareAndSwapInt(this, STATUS, s = status, - (s & ~SMASK) | (tag & SMASK))) - return (short)s; - } - } - - /** - * Atomically conditionally sets the tag value for this task. - * Among other applications, tags can be used as visit markers - * in tasks operating on graphs, as in methods that check: {@code - * if (task.compareAndSetForkJoinTaskTag((short)0, (short)1))} - * before processing, otherwise exiting because the node has - * already been visited. - * - * @param e the expected tag value - * @param tag the new tag value - * @return true if successful; i.e., the current value was - * equal to e and is now tag. - * @since 1.8 - */ - public final boolean compareAndSetForkJoinTaskTag(short e, short tag) { - for (int s;;) { - if ((short)(s = status) != e) - return false; - if (U.compareAndSwapInt(this, STATUS, s, - (s & ~SMASK) | (tag & SMASK))) - return true; - } - } - - /** - * Adaptor for Runnables. This implements RunnableFuture - * to be compliant with AbstractExecutorService constraints - * when used in ForkJoinPool. - */ - static final class AdaptedRunnable<T> extends ForkJoinTask<T> - implements RunnableFuture<T> { - final Runnable runnable; - T result; - AdaptedRunnable(Runnable runnable, T result) { - if (runnable == null) throw new NullPointerException(); - this.runnable = runnable; - this.result = result; // OK to set this even before completion - } - public final T getRawResult() { return result; } - public final void setRawResult(T v) { result = v; } - public final boolean exec() { runnable.run(); return true; } - public final void run() { invoke(); } - private static final long serialVersionUID = 5232453952276885070L; - } - - /** - * Adaptor for Runnables without results - */ - static final class AdaptedRunnableAction extends ForkJoinTask<Void> - implements RunnableFuture<Void> { - final Runnable runnable; - AdaptedRunnableAction(Runnable runnable) { - if (runnable == null) throw new NullPointerException(); - this.runnable = runnable; - } - public final Void getRawResult() { return null; } - public final void setRawResult(Void v) { } - public final boolean exec() { runnable.run(); return true; } - public final void run() { invoke(); } - private static final long serialVersionUID = 5232453952276885070L; - } - - /** - * Adaptor for Callables - */ - static final class AdaptedCallable<T> extends ForkJoinTask<T> - implements RunnableFuture<T> { - final Callable<? extends T> callable; - T result; - AdaptedCallable(Callable<? extends T> callable) { - if (callable == null) throw new NullPointerException(); - this.callable = callable; - } - public final T getRawResult() { return result; } - public final void setRawResult(T v) { result = v; } - public final 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 final 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 AdaptedRunnableAction(runnable); - } - - /** - * 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 <T> ForkJoinTask<T> adapt(Runnable runnable, T result) { - return new AdaptedRunnable<T>(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 <T> ForkJoinTask<T> adapt(Callable<? extends T> callable) { - return new AdaptedCallable<T>(callable); - } - - // Serialization support - - private static final long serialVersionUID = -7721805057305804111L; - - /** - * Saves this task to a stream (that is, serializes it). - * - * @serialData the current run status and the exception thrown - * during execution, or {@code null} if none - */ - private void writeObject(java.io.ObjectOutputStream s) - throws java.io.IOException { - s.defaultWriteObject(); - s.writeObject(getException()); - } - - /** - * Reconstitutes this task from a stream (that is, deserializes it). - */ - private void readObject(java.io.ObjectInputStream s) - throws java.io.IOException, ClassNotFoundException { - s.defaultReadObject(); - Object ex = s.readObject(); - if (ex != null) - setExceptionalCompletion((Throwable)ex); - } - - // Unsafe mechanics - private static final sun.misc.Unsafe U; - private static final long STATUS; - - static { - exceptionTableLock = new ReentrantLock(); - exceptionTableRefQueue = new ReferenceQueue<Object>(); - exceptionTable = new ExceptionNode[EXCEPTION_MAP_CAPACITY]; - try { - U = getUnsafe(); - Class<?> k = ForkJoinTask.class; - STATUS = U.objectFieldOffset - (k.getDeclaredField("status")); - } catch (Exception e) { - throw new Error(e); - } - } - - /** - * Returns a sun.misc.Unsafe. Suitable for use in a 3rd party package. - * Replace with a simple call to Unsafe.getUnsafe when integrating - * into a jdk. - * - * @return a sun.misc.Unsafe - */ - private static sun.misc.Unsafe getUnsafe() { - return scala.concurrent.util.Unsafe.instance; - } -} diff --git a/src/forkjoin/scala/concurrent/forkjoin/ForkJoinWorkerThread.java b/src/forkjoin/scala/concurrent/forkjoin/ForkJoinWorkerThread.java deleted file mode 100644 index e62fc6eb71..0000000000 --- a/src/forkjoin/scala/concurrent/forkjoin/ForkJoinWorkerThread.java +++ /dev/null @@ -1,121 +0,0 @@ -/* - * 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/publicdomain/zero/1.0/ - */ - -package scala.concurrent.forkjoin; - -/** - * 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 { - /* - * ForkJoinWorkerThreads are managed by ForkJoinPools and perform - * ForkJoinTasks. For explanation, see the internal documentation - * of class ForkJoinPool. - * - * This class just maintains links to its pool and WorkQueue. The - * pool field is set immediately upon construction, but the - * workQueue field is not set until a call to registerWorker - * completes. This leads to a visibility race, that is tolerated - * by requiring that the workQueue field is only accessed by the - * owning thread. - */ - - final ForkJoinPool pool; // the pool this thread works in - final ForkJoinPool.WorkQueue workQueue; // work-stealing mechanics - - /** - * 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) { - // Use a placeholder until a useful name can be set in registerWorker - super("aForkJoinWorkerThread"); - this.pool = pool; - this.workQueue = pool.registerWorker(this); - } - - /** - * Returns the pool hosting this thread. - * - * @return the pool - */ - public ForkJoinPool getPool() { - return pool; - } - - /** - * Returns the index number of this thread in its pool. The - * returned value ranges from zero to the maximum number of - * 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 - */ - public int getPoolIndex() { - return workQueue.poolIndex; - } - - /** - * Initializes internal state after construction but before - * processing any tasks. If you override this method, you must - * 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() { - } - - /** - * Performs cleanup associated with termination of this worker - * thread. If you override this method, you must invoke - * {@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 {@code null} if completed normally - */ - protected void onTermination(Throwable exception) { - } - - /** - * This method is required to be public, but should never be - * called explicitly. It performs the main run loop to execute - * {@link ForkJoinTask}s. - */ - public void run() { - Throwable exception = null; - try { - onStart(); - pool.runWorker(workQueue); - } catch (Throwable ex) { - exception = ex; - } finally { - try { - onTermination(exception); - } catch (Throwable ex) { - if (exception == null) - exception = ex; - } finally { - pool.deregisterWorker(this, exception); - } - } - } -} diff --git a/src/forkjoin/scala/concurrent/forkjoin/LinkedTransferQueue.java b/src/forkjoin/scala/concurrent/forkjoin/LinkedTransferQueue.java deleted file mode 100644 index 07e81b395d..0000000000 --- a/src/forkjoin/scala/concurrent/forkjoin/LinkedTransferQueue.java +++ /dev/null @@ -1,1335 +0,0 @@ -/* - * 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/publicdomain/zero/1.0/ - */ - -package scala.concurrent.forkjoin; - -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 {@link TransferQueue} based on linked nodes. - * This queue orders elements FIFO (first-in-first-out) with respect - * to any given producer. The <em>head</em> of the queue is that - * element that has been on the queue the longest time for some - * producer. The <em>tail</em> of the queue is that element that has - * been on the queue the shortest time for some producer. - * - * <p>Beware that, unlike in most collections, the {@code size} method - * is <em>NOT</em> a constant-time operation. Because of the - * asynchronous nature of these queues, determining the current number - * 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 <em>not</em> guaranteed - * to be performed atomically. For example, an iterator operating - * concurrently with an {@code addAll} operation might view only some - * of the added elements. - * - * <p>This class and its iterator implement all of the - * <em>optional</em> methods of the {@link Collection} and {@link - * Iterator} interfaces. - * - * <p>Memory consistency effects: As with other concurrent - * collections, actions in a thread prior to placing an object into a - * {@code LinkedTransferQueue} - * <a href="package-summary.html#MemoryVisibility"><i>happen-before</i></a> - * actions subsequent to the access or removal of that element from - * the {@code LinkedTransferQueue} in another thread. - * - * <p>This class is a member of the - * <a href="{@docRoot}/../technotes/guides/collections/index.html"> - * Java Collections Framework</a>. - * - * @since 1.7 - * @author Doug Lea - * @param <E> the type of elements held in this collection - */ -public class LinkedTransferQueue<E> extends AbstractQueue<E> - implements TransferQueue<E>, java.io.Serializable { - private static final long serialVersionUID = -3223113410248163686L; - - /* - * *** 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. - * - * 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 it might 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. - */ - - /** True if on multiprocessor */ - private static final boolean MP = - Runtime.getRuntime().availableProcessors() > 1; - - /** - * 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. - */ - private static final int FRONT_SPINS = 1 << 7; - - /** - * 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. - */ - private static final int CHAINED_SPINS = FRONT_SPINS >>> 1; - - /** - * 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 int SWEEP_THRESHOLD = 32; - - /** - * 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 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; - } - - /** - * 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); - } - - /** - * 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); - } - - /** - * 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; - } - - /** - * 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 = 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; - - /** tail of the queue; null until first append */ - private transient volatile Node tail; - - /** The number of apparent failures to unsplice removed nodes */ - private transient volatile int sweepVotes; - - // 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 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> E cast(Object item) { - // assert item == null || item.getClass() != Node.class; - return (E) item; - } - - /** - * Implements all queuing methods. See above for explanation. - * - * @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 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 LinkedTransferQueue.<E>cast(item); - } - } - Node n = p.next; - p = (p != n) ? n : (h = head); // Use head if p offlist - } - - 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 - } - } - - /** - * 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 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 (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/yields/blocks until node s is matched or caller gives up. - * - * @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 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 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; // initialized after first item and cancel checks - ThreadLocalRandom randomYields = null; // bound if needed - - for (;;) { - Object item = s.item; - if (item != e) { // matched - // assert item != s; - s.forgetContents(); // avoid garbage - return LinkedTransferQueue.<E>cast(item); - } - if ((w.isInterrupted() || (timed && nanos <= 0)) && - s.casItem(e, s)) { // cancel - unsplice(pred, s); - return e; - } - - if (spins < 0) { // establish spins at/near front - if ((spins = spinsFor(pred, s.isData)) > 0) - randomYields = ThreadLocalRandom.current(); - } - else if (spins > 0) { // spin - --spins; - 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 { - LockSupport.park(this); - } - } - } - - /** - * Returns spin/yield value for a node with given predecessor and - * data mode. See above for explanation. - */ - 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 LinkedTransferQueue.<E>cast(item); - } - else if (item == null) - return null; - } - return null; - } - - /** - * Traverses and counts unmatched nodes of the given mode. - * Used by methods size and getWaitingConsumerCount. - */ - 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; - } - - final class Itr implements Iterator<E> { - 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 - - /** - * Moves to next node after prev, or first node if prev null. - */ - 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.<E>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); - } - 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 -------------- */ - - /** - * Unsplices (now or later) the given deleted/cancelled node with - * the given predecessor. - * - * @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 - */ - final void unsplice(Node pred, Node s) { - s.forgetContents(); // forget unneeded fields - /* - * 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. - */ - 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; - } - } - } - } - } - } - - /** - * 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 false; - } - - - /** - * Creates an initially empty {@code LinkedTransferQueue}. - */ - public LinkedTransferQueue() { - } - - /** - * Creates a {@code LinkedTransferQueue} - * initially containing the elements of the given collection, - * added in traversal order of the collection's iterator. - * - * @param c the collection of elements to initially contain - * @throws NullPointerException if the specified collection or any - * of its elements are null - */ - public LinkedTransferQueue(Collection<? extends E> c) { - this(); - addAll(c); - } - - /** - * 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); - } - - /** - * 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 java.util.concurrent.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) { - 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) { - xfer(e, true, ASYNC, 0); - return true; - } - - /** - * Transfers the element to a waiting consumer immediately, if possible. - * - * <p>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. - * - * <p>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 (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. - * - * <p>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 (xfer(e, true, TIMED, unit.toNanos(timeout)) == null) - return true; - if (!Thread.interrupted()) - return false; - throw new InterruptedException(); - } - - public E take() throws InterruptedException { - E e = xfer(null, false, SYNC, 0); - if (e != null) - return e; - Thread.interrupted(); - throw new InterruptedException(); - } - - public E poll(long timeout, TimeUnit unit) throws InterruptedException { - E e = xfer(null, false, TIMED, unit.toNanos(timeout)); - if (e != null || !Thread.interrupted()) - return e; - throw new InterruptedException(); - } - - public E poll() { - return xfer(null, false, NOW, 0); - } - - /** - * @throws NullPointerException {@inheritDoc} - * @throws IllegalArgumentException {@inheritDoc} - */ - public int drainTo(Collection<? super E> c) { - if (c == null) - throw new NullPointerException(); - if (c == this) - throw new IllegalArgumentException(); - int n = 0; - for (E e; (e = poll()) != null;) { - c.add(e); - ++n; - } - return n; - } - - /** - * @throws NullPointerException {@inheritDoc} - * @throws IllegalArgumentException {@inheritDoc} - */ - public int drainTo(Collection<? super E> c, int maxElements) { - if (c == null) - throw new NullPointerException(); - if (c == this) - throw new IllegalArgumentException(); - int n = 0; - for (E e; n < maxElements && (e = poll()) != null;) { - c.add(e); - ++n; - } - return n; - } - - /** - * 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). - * - * <p>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 - */ - public Iterator<E> iterator() { - return new Itr(); - } - - public E peek() { - return firstDataItem(); - } - - /** - * Returns {@code true} if this queue contains no elements. - * - * @return {@code true} if this queue contains no elements - */ - public boolean isEmpty() { - for (Node p = head; p != null; p = succ(p)) { - if (!p.isMatched()) - return !p.isData; - } - return true; - } - - public boolean hasWaitingConsumer() { - return firstOfMode(false) != null; - } - - /** - * Returns the number of elements in this queue. If this queue - * contains more than {@code Integer.MAX_VALUE} elements, returns - * {@code Integer.MAX_VALUE}. - * - * <p>Beware that, unlike in most collections, this method is - * <em>NOT</em> a constant-time operation. Because of the - * asynchronous nature of these queues, determining the current - * number of elements requires an O(n) traversal. - * - * @return the number of elements in this queue - */ - public int size() { - return countOfMode(true); - } - - public int getWaitingConsumerCount() { - return countOfMode(false); - } - - /** - * 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); - } - - /** - * 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; - } - 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 java.util.concurrent.BlockingQueue#remainingCapacity() - * BlockingQueue.remainingCapacity}) - */ - public int remainingCapacity() { - return Integer.MAX_VALUE; - } - - /** - * 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 - * @param s the stream - */ - private void writeObject(java.io.ObjectOutputStream s) - throws java.io.IOException { - s.defaultWriteObject(); - for (E e : this) - s.writeObject(e); - // Use trailing null as sentinel - s.writeObject(null); - } - - /** - * 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(); - for (;;) { - @SuppressWarnings("unchecked") - E item = (E) s.readObject(); - if (item == null) - break; - else - offer(item); - } - } - - // Unsafe mechanics - - private static final sun.misc.Unsafe UNSAFE; - private static final long headOffset; - private static final long tailOffset; - private static final long sweepVotesOffset; - static { - try { - 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); - } - } - - /** - * Returns a sun.misc.Unsafe. Suitable for use in a 3rd party package. - * Replace with a simple call to Unsafe.getUnsafe when integrating - * into a jdk. - * - * @return a sun.misc.Unsafe - */ - static sun.misc.Unsafe getUnsafe() { - return scala.concurrent.util.Unsafe.instance; - } - -} diff --git a/src/forkjoin/scala/concurrent/forkjoin/RecursiveAction.java b/src/forkjoin/scala/concurrent/forkjoin/RecursiveAction.java deleted file mode 100644 index 1e7cdd952d..0000000000 --- a/src/forkjoin/scala/concurrent/forkjoin/RecursiveAction.java +++ /dev/null @@ -1,164 +0,0 @@ -/* - * 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/publicdomain/zero/1.0/ - */ - -package scala.concurrent.forkjoin; - -/** - * 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 {@code join} - * always return {@code null} upon completion. - * - * <p><b>Sample Usages.</b> Here is a simple but complete ForkJoin - * sort that sorts a given {@code long[]} array: - * - * <pre> {@code - * static class SortTask extends RecursiveAction { - * final long[] array; final int lo, hi; - * SortTask(long[] array, int lo, int hi) { - * this.array = array; this.lo = lo; this.hi = hi; - * } - * SortTask(long[] array) { this(array, 0, array.length); } - * protected void compute() { - * if (hi - lo < THRESHOLD) - * sortSequentially(lo, hi); - * else { - * int mid = (lo + hi) >>> 1; - * invokeAll(new SortTask(array, lo, mid), - * new SortTask(array, mid, hi)); - * merge(lo, mid, hi); - * } - * } - * // implementation details follow: - * final static int THRESHOLD = 1000; - * void sortSequentially(int lo, int hi) { - * Arrays.sort(array, lo, hi); - * } - * void merge(int lo, int mid, int hi) { - * long[] buf = Arrays.copyOfRange(array, lo, mid); - * for (int i = 0, j = lo, k = mid; i < buf.length; j++) - * array[j] = (k == hi || buf[i] < array[k]) ? - * buf[i++] : array[k++]; - * } - * }}</pre> - * - * You could then sort {@code anArray} by creating {@code new - * SortTask(anArray)} and invoking it in a ForkJoinPool. As a more - * concrete simple example, the following task increments each element - * of an array: - * <pre> {@code - * class IncrementTask extends RecursiveAction { - * final long[] array; final int lo, 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) - * array[i]++; - * } - * else { - * int mid = (lo + hi) >>> 1; - * invokeAll(new IncrementTask(array, lo, mid), - * new IncrementTask(array, mid, hi)); - * } - * } - * }}</pre> - * - * <p>The following example illustrates some refinements and idioms - * that may lead to better performance: RecursiveActions need not be - * fully recursive, so long as they maintain the basic - * 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 {@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. - * - * <pre> {@code - * double sumOfSquares(ForkJoinPool pool, double[] array) { - * int n = array.length; - * 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; - * double result; - * Applyer next; // keeps track of right-hand-side tasks - * Applyer(double[] array, int lo, int hi, Applyer next) { - * this.array = array; this.lo = lo; this.hi = hi; - * this.next = next; - * } - * - * double atLeaf(int l, int h) { - * double sum = 0; - * for (int i = l; i < h; ++i) // perform leftmost base step - * sum += array[i] * array[i]; - * return sum; - * } - * - * protected void compute() { - * int l = lo; - * int h = hi; - * Applyer right = null; - * while (h - l > 1 && getSurplusQueuedTaskCount() <= 3) { - * int mid = (l + h) >>> 1; - * right = new Applyer(array, mid, h, right); - * right.fork(); - * h = mid; - * } - * double sum = atLeaf(l, h); - * while (right != null) { - * if (right.tryUnfork()) // directly calculate if not stolen - * sum += right.atLeaf(right.lo, right.hi); - * else { - * right.join(); - * sum += right.result; - * } - * right = right.next; - * } - * result = sum; - * } - * }}</pre> - * - * @since 1.7 - * @author Doug Lea - */ -public abstract class RecursiveAction extends ForkJoinTask<Void> { - private static final long serialVersionUID = 5232453952276485070L; - - /** - * The main computation performed by this task. - */ - protected abstract void compute(); - - /** - * Always returns {@code null}. - * - * @return {@code null} always - */ - public final Void getRawResult() { return null; } - - /** - * Requires null completion value. - */ - protected final void setRawResult(Void mustBeNull) { } - - /** - * Implements execution conventions for RecursiveActions. - */ - protected final boolean exec() { - compute(); - return true; - } - -} diff --git a/src/forkjoin/scala/concurrent/forkjoin/RecursiveTask.java b/src/forkjoin/scala/concurrent/forkjoin/RecursiveTask.java deleted file mode 100644 index d1e1547143..0000000000 --- a/src/forkjoin/scala/concurrent/forkjoin/RecursiveTask.java +++ /dev/null @@ -1,68 +0,0 @@ -/* - * 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/publicdomain/zero/1.0/ - */ - -package scala.concurrent.forkjoin; - -/** - * A recursive result-bearing {@link ForkJoinTask}. - * - * <p>For a classic example, here is a task computing Fibonacci numbers: - * - * <pre> {@code - * class Fibonacci extends RecursiveTask<Integer> { - * final int n; - * Fibonacci(int n) { this.n = n; } - * Integer compute() { - * if (n <= 1) - * return n; - * Fibonacci f1 = new Fibonacci(n - 1); - * f1.fork(); - * Fibonacci f2 = new Fibonacci(n - 2); - * return f2.compute() + f1.join(); - * } - * }}</pre> - * - * However, besides being a dumb way to compute Fibonacci functions - * (there is a simple fast linear algorithm that you'd use in - * practice), this is likely to perform poorly because the smallest - * subtasks are too small to be worthwhile splitting up. Instead, as - * is the case for nearly all fork/join applications, you'd pick some - * 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<V> extends ForkJoinTask<V> { - private static final long serialVersionUID = 5232453952276485270L; - - /** - * The result of the computation. - */ - V result; - - /** - * The main computation performed by this task. - */ - protected abstract V compute(); - - public final V getRawResult() { - return result; - } - - protected final void setRawResult(V value) { - result = value; - } - - /** - * Implements execution conventions for RecursiveTask. - */ - protected final boolean exec() { - result = compute(); - return true; - } - -} diff --git a/src/forkjoin/scala/concurrent/forkjoin/ThreadLocalRandom.java b/src/forkjoin/scala/concurrent/forkjoin/ThreadLocalRandom.java deleted file mode 100644 index a7ef492057..0000000000 --- a/src/forkjoin/scala/concurrent/forkjoin/ThreadLocalRandom.java +++ /dev/null @@ -1,197 +0,0 @@ -/* - * 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/publicdomain/zero/1.0/ - */ - -package scala.concurrent.forkjoin; - -import java.util.Random; - -/** - * 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. - * - * <p>Usages of this class should typically be of the form: - * {@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 - * accidentally share a {@code ThreadLocalRandom} across multiple threads. - * - * <p>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 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. - */ - private long rnd; - - /** - * 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; - - // Padding to help avoid memory contention among seed updates in - // different TLRs in the common case that they are located near - // each other. - private long pad0, pad1, pad2, pad3, pad4, pad5, pad6, pad7; - - /** - * The actual ThreadLocal - */ - private static final ThreadLocal<ThreadLocalRandom> localRandom = - new ThreadLocal<ThreadLocalRandom>() { - protected ThreadLocalRandom initialValue() { - return new ThreadLocalRandom(); - } - }; - - - /** - * Constructor called only by localRandom.initialValue. - */ - ThreadLocalRandom() { - super(); - initialized = true; - } - - /** - * Returns the current thread's {@code ThreadLocalRandom}. - * - * @return the current thread's {@code ThreadLocalRandom} - */ - public static ThreadLocalRandom current() { - return localRandom.get(); - } - - /** - * Throws {@code UnsupportedOperationException}. Setting seeds in - * this generator is not supported. - * - * @throws UnsupportedOperationException always - */ - public void setSeed(long seed) { - if (initialized) - throw new UnsupportedOperationException(); - rnd = (seed ^ multiplier) & mask; - } - - protected int next(int 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 - * to bound - * @return the next value - */ - public int nextInt(int least, int bound) { - if (least >= bound) - throw new IllegalArgumentException(); - return nextInt(bound - least) + least; - } - - /** - * Returns a pseudorandom, uniformly distributed value - * 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 - * @throws IllegalArgumentException if n is not positive - */ - public long nextLong(long n) { - if (n <= 0) - throw new IllegalArgumentException("n must be positive"); - // Divide n by two until small enough for nextInt. On each - // iteration (at most 31 of them but usually much less), - // randomly choose both whether to include high bit in result - // (offset) and whether to continue with the lower vs upper - // half (which makes a difference only if odd). - long offset = 0; - while (n >= Integer.MAX_VALUE) { - int bits = next(2); - long half = n >>> 1; - long nextn = ((bits & 2) == 0) ? half : n - half; - if ((bits & 1) == 0) - offset += n - nextn; - n = nextn; - } - 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 - * @throws IllegalArgumentException if least greater than or equal - * to bound - */ - public long nextLong(long least, long bound) { - if (least >= bound) - throw new IllegalArgumentException(); - return nextLong(bound - least) + least; - } - - /** - * Returns a pseudorandom, uniformly distributed {@code double} value - * 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 - * @throws IllegalArgumentException if n is not positive - */ - public double nextDouble(double n) { - if (n <= 0) - throw new IllegalArgumentException("n must be positive"); - return nextDouble() * 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 - * @throws IllegalArgumentException if least greater than or equal - * to bound - */ - public double nextDouble(double least, double bound) { - if (least >= bound) - throw new IllegalArgumentException(); - 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 deleted file mode 100644 index 7d149c7ae5..0000000000 --- a/src/forkjoin/scala/concurrent/forkjoin/TransferQueue.java +++ /dev/null @@ -1,133 +0,0 @@ -/* - * 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/publicdomain/zero/1.0/ - */ - -package scala.concurrent.forkjoin; -import java.util.concurrent.*; - -/** - * 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 {@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. - * - * <p>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. - * - * <p>This interface is a member of the - * <a href="{@docRoot}/../technotes/guides/collections/index.html"> - * Java Collections Framework</a>. - * - * @since 1.7 - * @author Doug Lea - * @param <E> the type of elements held in this collection - */ -public interface TransferQueue<E> extends BlockingQueue<E> { - /** - * Transfers the element to a waiting consumer immediately, if possible. - * - * <p>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 - * {@code false} - * @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 - * @throws IllegalArgumentException if some property of the specified - * element prevents it from being added to this queue - */ - boolean tryTransfer(E e); - - /** - * Transfers the element to a consumer, waiting if necessary to do so. - * - * <p>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 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 - * @throws IllegalArgumentException if some property of the specified - * element prevents it from being added to this queue - */ - void transfer(E e) throws InterruptedException; - - /** - * Transfers the element to a consumer if it is possible to do so - * before the timeout elapses. - * - * <p>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 - * {@code unit} - * @param unit a {@code TimeUnit} determining how to interpret the - * {@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 left enqueued - * @throws InterruptedException if interrupted while waiting, - * 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 - * @throws IllegalArgumentException if some property of the specified - * element prevents it from being added to this queue - */ - boolean tryTransfer(E e, long timeout, TimeUnit unit) - throws InterruptedException; - - /** - * Returns {@code true} if there is at least one consumer waiting - * 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 - */ - boolean hasWaitingConsumer(); - - /** - * Returns an estimate of the number of consumers waiting to - * 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 receive elements - */ - int getWaitingConsumerCount(); -} diff --git a/src/forkjoin/scala/concurrent/forkjoin/package-info.java b/src/forkjoin/scala/concurrent/forkjoin/package-info.java deleted file mode 100644 index 3561b9b44a..0000000000 --- a/src/forkjoin/scala/concurrent/forkjoin/package-info.java +++ /dev/null @@ -1,28 +0,0 @@ -/* - * 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/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. - * - * <p>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 <em>work-stealing</em> 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. - */ -package scala.concurrent.forkjoin; diff --git a/src/forkjoin/scala/concurrent/util/Unsafe.java b/src/forkjoin/scala/concurrent/util/Unsafe.java deleted file mode 100644 index ef893c94d9..0000000000 --- a/src/forkjoin/scala/concurrent/util/Unsafe.java +++ /dev/null @@ -1,35 +0,0 @@ -/* __ *\ -** ________ ___ / / ___ Scala API ** -** / __/ __// _ | / / / _ | (c) 2003-2013, LAMP/EPFL ** -** __\ \/ /__/ __ |/ /__/ __ | http://scala-lang.org/ ** -** /____/\___/_/ |_/____/_/ | | ** -** |/ ** -\* */ - -package scala.concurrent.util; - - - -import java.lang.reflect.Field; - - - -public final class Unsafe { - public final static sun.misc.Unsafe instance; - static { - try { - sun.misc.Unsafe found = null; - for(Field field : sun.misc.Unsafe.class.getDeclaredFields()) { - if (field.getType() == sun.misc.Unsafe.class) { - field.setAccessible(true); - found = (sun.misc.Unsafe) field.get(null); - break; - } - } - if (found == null) throw new IllegalStateException("Can't find instance of sun.misc.Unsafe"); - else instance = found; - } catch(Throwable t) { - throw new ExceptionInInitializerError(t); - } - } -} |