1 /*
   2  * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
   3  *
   4  * This code is free software; you can redistribute it and/or modify it
   5  * under the terms of the GNU General Public License version 2 only, as
   6  * published by the Free Software Foundation.  Oracle designates this
   7  * particular file as subject to the "Classpath" exception as provided
   8  * by Oracle in the LICENSE file that accompanied this code.
   9  *
  10  * This code is distributed in the hope that it will be useful, but WITHOUT
  11  * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
  12  * FITNESS FOR A PARTICULAR PURPOSE.  See the GNU General Public License
  13  * version 2 for more details (a copy is included in the LICENSE file that
  14  * accompanied this code).
  15  *
  16  * You should have received a copy of the GNU General Public License version
  17  * 2 along with this work; if not, write to the Free Software Foundation,
  18  * Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA.
  19  *
  20  * Please contact Oracle, 500 Oracle Parkway, Redwood Shores, CA 94065 USA
  21  * or visit www.oracle.com if you need additional information or have any
  22  * questions.
  23  */
  24 
  25 /*
  26  * This file is available under and governed by the GNU General Public
  27  * License version 2 only, as published by the Free Software Foundation.
  28  * However, the following notice accompanied the original version of this
  29  * file:
  30  *
  31  * Written by Doug Lea with assistance from members of JCP JSR-166
  32  * Expert Group and released to the public domain, as explained at
  33  * http://creativecommons.org/licenses/publicdomain
  34  */
  35 
  36 package java.util.concurrent;
  37 
  38 import java.util.ArrayList;
  39 import java.util.Arrays;
  40 import java.util.Collection;
  41 import java.util.Collections;
  42 import java.util.List;
  43 import java.util.concurrent.AbstractExecutorService;
  44 import java.util.concurrent.Callable;
  45 import java.util.concurrent.ExecutorService;
  46 import java.util.concurrent.Future;
  47 import java.util.concurrent.RejectedExecutionException;
  48 import java.util.concurrent.RunnableFuture;
  49 import java.util.concurrent.TimeUnit;
  50 import java.util.concurrent.TimeoutException;
  51 import java.util.concurrent.atomic.AtomicInteger;
  52 import java.util.concurrent.locks.LockSupport;
  53 import java.util.concurrent.locks.ReentrantLock;
  54 
  55 /**
  56  * An {@link ExecutorService} for running {@link ForkJoinTask}s.
  57  * A {@code ForkJoinPool} provides the entry point for submissions
  58  * from non-{@code ForkJoinTask} clients, as well as management and
  59  * monitoring operations.
  60  *
  61  * <p>A {@code ForkJoinPool} differs from other kinds of {@link
  62  * ExecutorService} mainly by virtue of employing
  63  * <em>work-stealing</em>: all threads in the pool attempt to find and
  64  * execute subtasks created by other active tasks (eventually blocking
  65  * waiting for work if none exist). This enables efficient processing
  66  * when most tasks spawn other subtasks (as do most {@code
  67  * ForkJoinTask}s). When setting <em>asyncMode</em> to true in
  68  * constructors, {@code ForkJoinPool}s may also be appropriate for use
  69  * with event-style tasks that are never joined.
  70  *
  71  * <p>A {@code ForkJoinPool} is constructed with a given target
  72  * parallelism level; by default, equal to the number of available
  73  * processors. The pool attempts to maintain enough active (or
  74  * available) threads by dynamically adding, suspending, or resuming
  75  * internal worker threads, even if some tasks are stalled waiting to
  76  * join others. However, no such adjustments are guaranteed in the
  77  * face of blocked IO or other unmanaged synchronization. The nested
  78  * {@link ManagedBlocker} interface enables extension of the kinds of
  79  * synchronization accommodated.
  80  *
  81  * <p>In addition to execution and lifecycle control methods, this
  82  * class provides status check methods (for example
  83  * {@link #getStealCount}) that are intended to aid in developing,
  84  * tuning, and monitoring fork/join applications. Also, method
  85  * {@link #toString} returns indications of pool state in a
  86  * convenient form for informal monitoring.
  87  *
  88  * <p> As is the case with other ExecutorServices, there are three
  89  * main task execution methods summarized in the following
  90  * table. These are designed to be used by clients not already engaged
  91  * in fork/join computations in the current pool.  The main forms of
  92  * these methods accept instances of {@code ForkJoinTask}, but
  93  * overloaded forms also allow mixed execution of plain {@code
  94  * Runnable}- or {@code Callable}- based activities as well.  However,
  95  * tasks that are already executing in a pool should normally
  96  * <em>NOT</em> use these pool execution methods, but instead use the
  97  * within-computation forms listed in the table.
  98  *
  99  * <table BORDER CELLPADDING=3 CELLSPACING=1>
 100  *  <tr>
 101  *    <td></td>
 102  *    <td ALIGN=CENTER> <b>Call from non-fork/join clients</b></td>
 103  *    <td ALIGN=CENTER> <b>Call from within fork/join computations</b></td>
 104  *  </tr>
 105  *  <tr>
 106  *    <td> <b>Arrange async execution</td>
 107  *    <td> {@link #execute(ForkJoinTask)}</td>
 108  *    <td> {@link ForkJoinTask#fork}</td>
 109  *  </tr>
 110  *  <tr>
 111  *    <td> <b>Await and obtain result</td>
 112  *    <td> {@link #invoke(ForkJoinTask)}</td>
 113  *    <td> {@link ForkJoinTask#invoke}</td>
 114  *  </tr>
 115  *  <tr>
 116  *    <td> <b>Arrange exec and obtain Future</td>
 117  *    <td> {@link #submit(ForkJoinTask)}</td>
 118  *    <td> {@link ForkJoinTask#fork} (ForkJoinTasks <em>are</em> Futures)</td>
 119  *  </tr>
 120  * </table>
 121  *
 122  * <p><b>Sample Usage.</b> Normally a single {@code ForkJoinPool} is
 123  * used for all parallel task execution in a program or subsystem.
 124  * Otherwise, use would not usually outweigh the construction and
 125  * bookkeeping overhead of creating a large set of threads. For
 126  * example, a common pool could be used for the {@code SortTasks}
 127  * illustrated in {@link RecursiveAction}. Because {@code
 128  * ForkJoinPool} uses threads in {@linkplain java.lang.Thread#isDaemon
 129  * daemon} mode, there is typically no need to explicitly {@link
 130  * #shutdown} such a pool upon program exit.
 131  *
 132  * <pre>
 133  * static final ForkJoinPool mainPool = new ForkJoinPool();
 134  * ...
 135  * public void sort(long[] array) {
 136  *   mainPool.invoke(new SortTask(array, 0, array.length));
 137  * }
 138  * </pre>
 139  *
 140  * <p><b>Implementation notes</b>: This implementation restricts the
 141  * maximum number of running threads to 32767. Attempts to create
 142  * pools with greater than the maximum number result in
 143  * {@code IllegalArgumentException}.
 144  *
 145  * <p>This implementation rejects submitted tasks (that is, by throwing
 146  * {@link RejectedExecutionException}) only when the pool is shut down
 147  * or internal resources have been exhausted.
 148  *
 149  * @since 1.7
 150  * @author Doug Lea
 151  */
 152 public class ForkJoinPool extends AbstractExecutorService {
 153 
 154     /*
 155      * Implementation Overview
 156      *
 157      * This class provides the central bookkeeping and control for a
 158      * set of worker threads: Submissions from non-FJ threads enter
 159      * into a submission queue. Workers take these tasks and typically
 160      * split them into subtasks that may be stolen by other workers.
 161      * The main work-stealing mechanics implemented in class
 162      * ForkJoinWorkerThread give first priority to processing tasks
 163      * from their own queues (LIFO or FIFO, depending on mode), then
 164      * to randomized FIFO steals of tasks in other worker queues, and
 165      * lastly to new submissions. These mechanics do not consider
 166      * affinities, loads, cache localities, etc, so rarely provide the
 167      * best possible performance on a given machine, but portably
 168      * provide good throughput by averaging over these factors.
 169      * (Further, even if we did try to use such information, we do not
 170      * usually have a basis for exploiting it. For example, some sets
 171      * of tasks profit from cache affinities, but others are harmed by
 172      * cache pollution effects.)
 173      *
 174      * Beyond work-stealing support and essential bookkeeping, the
 175      * main responsibility of this framework is to take actions when
 176      * one worker is waiting to join a task stolen (or always held by)
 177      * another.  Because we are multiplexing many tasks on to a pool
 178      * of workers, we can't just let them block (as in Thread.join).
 179      * We also cannot just reassign the joiner's run-time stack with
 180      * another and replace it later, which would be a form of
 181      * "continuation", that even if possible is not necessarily a good
 182      * idea. Given that the creation costs of most threads on most
 183      * systems mainly surrounds setting up runtime stacks, thread
 184      * creation and switching is usually not much more expensive than
 185      * stack creation and switching, and is more flexible). Instead we
 186      * combine two tactics:
 187      *
 188      *   Helping: Arranging for the joiner to execute some task that it
 189      *      would be running if the steal had not occurred.  Method
 190      *      ForkJoinWorkerThread.helpJoinTask tracks joining->stealing
 191      *      links to try to find such a task.
 192      *
 193      *   Compensating: Unless there are already enough live threads,
 194      *      method helpMaintainParallelism() may create or
 195      *      re-activate a spare thread to compensate for blocked
 196      *      joiners until they unblock.
 197      *
 198      * It is impossible to keep exactly the target (parallelism)
 199      * number of threads running at any given time.  Determining
 200      * existence of conservatively safe helping targets, the
 201      * availability of already-created spares, and the apparent need
 202      * to create new spares are all racy and require heuristic
 203      * guidance, so we rely on multiple retries of each.  Compensation
 204      * occurs in slow-motion. It is triggered only upon timeouts of
 205      * Object.wait used for joins. This reduces poor decisions that
 206      * would otherwise be made when threads are waiting for others
 207      * that are stalled because of unrelated activities such as
 208      * garbage collection.
 209      *
 210      * The ManagedBlocker extension API can't use helping so relies
 211      * only on compensation in method awaitBlocker.
 212      *
 213      * The main throughput advantages of work-stealing stem from
 214      * decentralized control -- workers mostly steal tasks from each
 215      * other. We do not want to negate this by creating bottlenecks
 216      * implementing other management responsibilities. So we use a
 217      * collection of techniques that avoid, reduce, or cope well with
 218      * contention. These entail several instances of bit-packing into
 219      * CASable fields to maintain only the minimally required
 220      * atomicity. To enable such packing, we restrict maximum
 221      * parallelism to (1<<15)-1 (enabling twice this (to accommodate
 222      * unbalanced increments and decrements) to fit into a 16 bit
 223      * field, which is far in excess of normal operating range.  Even
 224      * though updates to some of these bookkeeping fields do sometimes
 225      * contend with each other, they don't normally cache-contend with
 226      * updates to others enough to warrant memory padding or
 227      * isolation. So they are all held as fields of ForkJoinPool
 228      * objects.  The main capabilities are as follows:
 229      *
 230      * 1. Creating and removing workers. Workers are recorded in the
 231      * "workers" array. This is an array as opposed to some other data
 232      * structure to support index-based random steals by workers.
 233      * Updates to the array recording new workers and unrecording
 234      * terminated ones are protected from each other by a lock
 235      * (workerLock) but the array is otherwise concurrently readable,
 236      * and accessed directly by workers. To simplify index-based
 237      * operations, the array size is always a power of two, and all
 238      * readers must tolerate null slots. Currently, all worker thread
 239      * creation is on-demand, triggered by task submissions,
 240      * replacement of terminated workers, and/or compensation for
 241      * blocked workers. However, all other support code is set up to
 242      * work with other policies.
 243      *
 244      * To ensure that we do not hold on to worker references that
 245      * would prevent GC, ALL accesses to workers are via indices into
 246      * the workers array (which is one source of some of the unusual
 247      * code constructions here). In essence, the workers array serves
 248      * as a WeakReference mechanism. Thus for example the event queue
 249      * stores worker indices, not worker references. Access to the
 250      * workers in associated methods (for example releaseEventWaiters)
 251      * must both index-check and null-check the IDs. All such accesses
 252      * ignore bad IDs by returning out early from what they are doing,
 253      * since this can only be associated with shutdown, in which case
 254      * it is OK to give up. On termination, we just clobber these
 255      * data structures without trying to use them.
 256      *
 257      * 2. Bookkeeping for dynamically adding and removing workers. We
 258      * aim to approximately maintain the given level of parallelism.
 259      * When some workers are known to be blocked (on joins or via
 260      * ManagedBlocker), we may create or resume others to take their
 261      * place until they unblock (see below). Implementing this
 262      * requires counts of the number of "running" threads (i.e., those
 263      * that are neither blocked nor artificially suspended) as well as
 264      * the total number.  These two values are packed into one field,
 265      * "workerCounts" because we need accurate snapshots when deciding
 266      * to create, resume or suspend.  Note however that the
 267      * correspondence of these counts to reality is not guaranteed. In
 268      * particular updates for unblocked threads may lag until they
 269      * actually wake up.
 270      *
 271      * 3. Maintaining global run state. The run state of the pool
 272      * consists of a runLevel (SHUTDOWN, TERMINATING, etc) similar to
 273      * those in other Executor implementations, as well as a count of
 274      * "active" workers -- those that are, or soon will be, or
 275      * recently were executing tasks. The runLevel and active count
 276      * are packed together in order to correctly trigger shutdown and
 277      * termination. Without care, active counts can be subject to very
 278      * high contention.  We substantially reduce this contention by
 279      * relaxing update rules.  A worker must claim active status
 280      * prospectively, by activating if it sees that a submitted or
 281      * stealable task exists (it may find after activating that the
 282      * task no longer exists). It stays active while processing this
 283      * task (if it exists) and any other local subtasks it produces,
 284      * until it cannot find any other tasks. It then tries
 285      * inactivating (see method preStep), but upon update contention
 286      * instead scans for more tasks, later retrying inactivation if it
 287      * doesn't find any.
 288      *
 289      * 4. Managing idle workers waiting for tasks. We cannot let
 290      * workers spin indefinitely scanning for tasks when none are
 291      * available. On the other hand, we must quickly prod them into
 292      * action when new tasks are submitted or generated.  We
 293      * park/unpark these idle workers using an event-count scheme.
 294      * Field eventCount is incremented upon events that may enable
 295      * workers that previously could not find a task to now find one:
 296      * Submission of a new task to the pool, or another worker pushing
 297      * a task onto a previously empty queue.  (We also use this
 298      * mechanism for configuration and termination actions that
 299      * require wakeups of idle workers).  Each worker maintains its
 300      * last known event count, and blocks when a scan for work did not
 301      * find a task AND its lastEventCount matches the current
 302      * eventCount. Waiting idle workers are recorded in a variant of
 303      * Treiber stack headed by field eventWaiters which, when nonzero,
 304      * encodes the thread index and count awaited for by the worker
 305      * thread most recently calling eventSync. This thread in turn has
 306      * a record (field nextEventWaiter) for the next waiting worker.
 307      * In addition to allowing simpler decisions about need for
 308      * wakeup, the event count bits in eventWaiters serve the role of
 309      * tags to avoid ABA errors in Treiber stacks. Upon any wakeup,
 310      * released threads also try to release at most two others.  The
 311      * net effect is a tree-like diffusion of signals, where released
 312      * threads (and possibly others) help with unparks.  To further
 313      * reduce contention effects a bit, failed CASes to increment
 314      * field eventCount are tolerated without retries in signalWork.
 315      * Conceptually they are merged into the same event, which is OK
 316      * when their only purpose is to enable workers to scan for work.
 317      *
 318      * 5. Managing suspension of extra workers. When a worker notices
 319      * (usually upon timeout of a wait()) that there are too few
 320      * running threads, we may create a new thread to maintain
 321      * parallelism level, or at least avoid starvation. Usually, extra
 322      * threads are needed for only very short periods, yet join
 323      * dependencies are such that we sometimes need them in
 324      * bursts. Rather than create new threads each time this happens,
 325      * we suspend no-longer-needed extra ones as "spares". For most
 326      * purposes, we don't distinguish "extra" spare threads from
 327      * normal "core" threads: On each call to preStep (the only point
 328      * at which we can do this) a worker checks to see if there are
 329      * now too many running workers, and if so, suspends itself.
 330      * Method helpMaintainParallelism looks for suspended threads to
 331      * resume before considering creating a new replacement. The
 332      * spares themselves are encoded on another variant of a Treiber
 333      * Stack, headed at field "spareWaiters".  Note that the use of
 334      * spares is intrinsically racy.  One thread may become a spare at
 335      * about the same time as another is needlessly being created. We
 336      * counteract this and related slop in part by requiring resumed
 337      * spares to immediately recheck (in preStep) to see whether they
 338      * should re-suspend.
 339      *
 340      * 6. Killing off unneeded workers. A timeout mechanism is used to
 341      * shed unused workers: The oldest (first) event queue waiter uses
 342      * a timed rather than hard wait. When this wait times out without
 343      * a normal wakeup, it tries to shutdown any one (for convenience
 344      * the newest) other spare or event waiter via
 345      * tryShutdownUnusedWorker. This eventually reduces the number of
 346      * worker threads to a minimum of one after a long enough period
 347      * without use.
 348      *
 349      * 7. Deciding when to create new workers. The main dynamic
 350      * control in this class is deciding when to create extra threads
 351      * in method helpMaintainParallelism. We would like to keep
 352      * exactly #parallelism threads running, which is an impossible
 353      * task. We always need to create one when the number of running
 354      * threads would become zero and all workers are busy. Beyond
 355      * this, we must rely on heuristics that work well in the
 356      * presence of transient phenomena such as GC stalls, dynamic
 357      * compilation, and wake-up lags. These transients are extremely
 358      * common -- we are normally trying to fully saturate the CPUs on
 359      * a machine, so almost any activity other than running tasks
 360      * impedes accuracy. Our main defense is to allow parallelism to
 361      * lapse for a while during joins, and use a timeout to see if,
 362      * after the resulting settling, there is still a need for
 363      * additional workers.  This also better copes with the fact that
 364      * some of the methods in this class tend to never become compiled
 365      * (but are interpreted), so some components of the entire set of
 366      * controls might execute 100 times faster than others. And
 367      * similarly for cases where the apparent lack of work is just due
 368      * to GC stalls and other transient system activity.
 369      *
 370      * Beware that there is a lot of representation-level coupling
 371      * among classes ForkJoinPool, ForkJoinWorkerThread, and
 372      * ForkJoinTask.  For example, direct access to "workers" array by
 373      * workers, and direct access to ForkJoinTask.status by both
 374      * ForkJoinPool and ForkJoinWorkerThread.  There is little point
 375      * trying to reduce this, since any associated future changes in
 376      * representations will need to be accompanied by algorithmic
 377      * changes anyway.
 378      *
 379      * Style notes: There are lots of inline assignments (of form
 380      * "while ((local = field) != 0)") which are usually the simplest
 381      * way to ensure the required read orderings (which are sometimes
 382      * critical). Also several occurrences of the unusual "do {}
 383      * while (!cas...)" which is the simplest way to force an update of
 384      * a CAS'ed variable. There are also other coding oddities that
 385      * help some methods perform reasonably even when interpreted (not
 386      * compiled), at the expense of some messy constructions that
 387      * reduce byte code counts.
 388      *
 389      * The order of declarations in this file is: (1) statics (2)
 390      * fields (along with constants used when unpacking some of them)
 391      * (3) internal control methods (4) callbacks and other support
 392      * for ForkJoinTask and ForkJoinWorkerThread classes, (5) exported
 393      * methods (plus a few little helpers).
 394      */
 395 
 396     /**
 397      * Factory for creating new {@link ForkJoinWorkerThread}s.
 398      * A {@code ForkJoinWorkerThreadFactory} must be defined and used
 399      * for {@code ForkJoinWorkerThread} subclasses that extend base
 400      * functionality or initialize threads with different contexts.
 401      */
 402     public static interface ForkJoinWorkerThreadFactory {
 403         /**
 404          * Returns a new worker thread operating in the given pool.
 405          *
 406          * @param pool the pool this thread works in
 407          * @throws NullPointerException if the pool is null
 408          */
 409         public ForkJoinWorkerThread newThread(ForkJoinPool pool);
 410     }
 411 
 412     /**
 413      * Default ForkJoinWorkerThreadFactory implementation; creates a
 414      * new ForkJoinWorkerThread.
 415      */
 416     static class DefaultForkJoinWorkerThreadFactory
 417         implements ForkJoinWorkerThreadFactory {
 418         public ForkJoinWorkerThread newThread(ForkJoinPool pool) {
 419             return new ForkJoinWorkerThread(pool);
 420         }
 421     }
 422 
 423     /**
 424      * Creates a new ForkJoinWorkerThread. This factory is used unless
 425      * overridden in ForkJoinPool constructors.
 426      */
 427     public static final ForkJoinWorkerThreadFactory
 428         defaultForkJoinWorkerThreadFactory =
 429         new DefaultForkJoinWorkerThreadFactory();
 430 
 431     /**
 432      * Permission required for callers of methods that may start or
 433      * kill threads.
 434      */
 435     private static final RuntimePermission modifyThreadPermission =
 436         new RuntimePermission("modifyThread");
 437 
 438     /**
 439      * If there is a security manager, makes sure caller has
 440      * permission to modify threads.
 441      */
 442     private static void checkPermission() {
 443         SecurityManager security = System.getSecurityManager();
 444         if (security != null)
 445             security.checkPermission(modifyThreadPermission);
 446     }
 447 
 448     /**
 449      * Generator for assigning sequence numbers as pool names.
 450      */
 451     private static final AtomicInteger poolNumberGenerator =
 452         new AtomicInteger();
 453 
 454     /**
 455      * The time to block in a join (see awaitJoin) before checking if
 456      * a new worker should be (re)started to maintain parallelism
 457      * level. The value should be short enough to maintain global
 458      * responsiveness and progress but long enough to avoid
 459      * counterproductive firings during GC stalls or unrelated system
 460      * activity, and to not bog down systems with continual re-firings
 461      * on GCs or legitimately long waits.
 462      */
 463     private static final long JOIN_TIMEOUT_MILLIS = 250L; // 4 per second
 464 
 465     /**
 466      * The wakeup interval (in nanoseconds) for the oldest worker
 467      * waiting for an event to invoke tryShutdownUnusedWorker to
 468      * shrink the number of workers.  The exact value does not matter
 469      * too much. It must be short enough to release resources during
 470      * sustained periods of idleness, but not so short that threads
 471      * are continually re-created.
 472      */
 473     private static final long SHRINK_RATE_NANOS =
 474         30L * 1000L * 1000L * 1000L; // 2 per minute
 475 
 476     /**
 477      * Absolute bound for parallelism level. Twice this number plus
 478      * one (i.e., 0xfff) must fit into a 16bit field to enable
 479      * word-packing for some counts and indices.
 480      */
 481     private static final int MAX_WORKERS   = 0x7fff;
 482 
 483     /**
 484      * Array holding all worker threads in the pool.  Array size must
 485      * be a power of two.  Updates and replacements are protected by
 486      * workerLock, but the array is always kept in a consistent enough
 487      * state to be randomly accessed without locking by workers
 488      * performing work-stealing, as well as other traversal-based
 489      * methods in this class. All readers must tolerate that some
 490      * array slots may be null.
 491      */
 492     volatile ForkJoinWorkerThread[] workers;
 493 
 494     /**
 495      * Queue for external submissions.
 496      */
 497     private final LinkedTransferQueue<ForkJoinTask<?>> submissionQueue;
 498 
 499     /**
 500      * Lock protecting updates to workers array.
 501      */
 502     private final ReentrantLock workerLock;
 503 
 504     /**
 505      * Latch released upon termination.
 506      */
 507     private final Phaser termination;
 508 
 509     /**
 510      * Creation factory for worker threads.
 511      */
 512     private final ForkJoinWorkerThreadFactory factory;
 513 
 514     /**
 515      * Sum of per-thread steal counts, updated only when threads are
 516      * idle or terminating.
 517      */
 518     private volatile long stealCount;
 519 
 520     /**
 521      * Encoded record of top of Treiber stack of threads waiting for
 522      * events. The top 32 bits contain the count being waited for. The
 523      * bottom 16 bits contains one plus the pool index of waiting
 524      * worker thread. (Bits 16-31 are unused.)
 525      */
 526     private volatile long eventWaiters;
 527 
 528     private static final int EVENT_COUNT_SHIFT = 32;
 529     private static final int WAITER_ID_MASK    = (1 << 16) - 1;
 530 
 531     /**
 532      * A counter for events that may wake up worker threads:
 533      *   - Submission of a new task to the pool
 534      *   - A worker pushing a task on an empty queue
 535      *   - termination
 536      */
 537     private volatile int eventCount;
 538 
 539     /**
 540      * Encoded record of top of Treiber stack of spare threads waiting
 541      * for resumption. The top 16 bits contain an arbitrary count to
 542      * avoid ABA effects. The bottom 16bits contains one plus the pool
 543      * index of waiting worker thread.
 544      */
 545     private volatile int spareWaiters;
 546 
 547     private static final int SPARE_COUNT_SHIFT = 16;
 548     private static final int SPARE_ID_MASK     = (1 << 16) - 1;
 549 
 550     /**
 551      * Lifecycle control. The low word contains the number of workers
 552      * that are (probably) executing tasks. This value is atomically
 553      * incremented before a worker gets a task to run, and decremented
 554      * when a worker has no tasks and cannot find any.  Bits 16-18
 555      * contain runLevel value. When all are zero, the pool is
 556      * running. Level transitions are monotonic (running -> shutdown
 557      * -> terminating -> terminated) so each transition adds a bit.
 558      * These are bundled together to ensure consistent read for
 559      * termination checks (i.e., that runLevel is at least SHUTDOWN
 560      * and active threads is zero).
 561      *
 562      * Notes: Most direct CASes are dependent on these bitfield
 563      * positions.  Also, this field is non-private to enable direct
 564      * performance-sensitive CASes in ForkJoinWorkerThread.
 565      */
 566     volatile int runState;
 567 
 568     // Note: The order among run level values matters.
 569     private static final int RUNLEVEL_SHIFT     = 16;
 570     private static final int SHUTDOWN           = 1 << RUNLEVEL_SHIFT;
 571     private static final int TERMINATING        = 1 << (RUNLEVEL_SHIFT + 1);
 572     private static final int TERMINATED         = 1 << (RUNLEVEL_SHIFT + 2);
 573     private static final int ACTIVE_COUNT_MASK  = (1 << RUNLEVEL_SHIFT) - 1;
 574 
 575     /**
 576      * Holds number of total (i.e., created and not yet terminated)
 577      * and running (i.e., not blocked on joins or other managed sync)
 578      * threads, packed together to ensure consistent snapshot when
 579      * making decisions about creating and suspending spare
 580      * threads. Updated only by CAS. Note that adding a new worker
 581      * requires incrementing both counts, since workers start off in
 582      * running state.
 583      */
 584     private volatile int workerCounts;
 585 
 586     private static final int TOTAL_COUNT_SHIFT  = 16;
 587     private static final int RUNNING_COUNT_MASK = (1 << TOTAL_COUNT_SHIFT) - 1;
 588     private static final int ONE_RUNNING        = 1;
 589     private static final int ONE_TOTAL          = 1 << TOTAL_COUNT_SHIFT;
 590 
 591     /**
 592      * The target parallelism level.
 593      * Accessed directly by ForkJoinWorkerThreads.
 594      */
 595     final int parallelism;
 596 
 597     /**
 598      * True if use local fifo, not default lifo, for local polling
 599      * Read by, and replicated by ForkJoinWorkerThreads
 600      */
 601     final boolean locallyFifo;
 602 
 603     /**
 604      * The uncaught exception handler used when any worker abruptly
 605      * terminates.
 606      */
 607     private final Thread.UncaughtExceptionHandler ueh;
 608 
 609     /**
 610      * Pool number, just for assigning useful names to worker threads
 611      */
 612     private final int poolNumber;
 613 
 614     // Utilities for CASing fields. Note that most of these
 615     // are usually manually inlined by callers
 616 
 617     /**
 618      * Increments running count part of workerCounts.
 619      */
 620     final void incrementRunningCount() {
 621         int c;
 622         do {} while (!UNSAFE.compareAndSwapInt(this, workerCountsOffset,
 623                                                c = workerCounts,
 624                                                c + ONE_RUNNING));
 625     }
 626 
 627     /**
 628      * Tries to increment running count part of workerCounts.
 629      */
 630     final boolean tryIncrementRunningCount() {
 631         int c;
 632         return UNSAFE.compareAndSwapInt(this, workerCountsOffset,
 633                                         c = workerCounts,
 634                                         c + ONE_RUNNING);
 635     }
 636 
 637     /**
 638      * Tries to decrement running count unless already zero.
 639      */
 640     final boolean tryDecrementRunningCount() {
 641         int wc = workerCounts;
 642         if ((wc & RUNNING_COUNT_MASK) == 0)
 643             return false;
 644         return UNSAFE.compareAndSwapInt(this, workerCountsOffset,
 645                                         wc, wc - ONE_RUNNING);
 646     }
 647 
 648     /**
 649      * Forces decrement of encoded workerCounts, awaiting nonzero if
 650      * (rarely) necessary when other count updates lag.
 651      *
 652      * @param dr -- either zero or ONE_RUNNING
 653      * @param dt -- either zero or ONE_TOTAL
 654      */
 655     private void decrementWorkerCounts(int dr, int dt) {
 656         for (;;) {
 657             int wc = workerCounts;
 658             if ((wc & RUNNING_COUNT_MASK)  - dr < 0 ||
 659                 (wc >>> TOTAL_COUNT_SHIFT) - dt < 0) {
 660                 if ((runState & TERMINATED) != 0)
 661                     return; // lagging termination on a backout
 662                 Thread.yield();
 663             }
 664             if (UNSAFE.compareAndSwapInt(this, workerCountsOffset,
 665                                          wc, wc - (dr + dt)))
 666                 return;
 667         }
 668     }
 669 
 670     /**
 671      * Tries decrementing active count; fails on contention.
 672      * Called when workers cannot find tasks to run.
 673      */
 674     final boolean tryDecrementActiveCount() {
 675         int c;
 676         return UNSAFE.compareAndSwapInt(this, runStateOffset,
 677                                         c = runState, c - 1);
 678     }
 679 
 680     /**
 681      * Advances to at least the given level. Returns true if not
 682      * already in at least the given level.
 683      */
 684     private boolean advanceRunLevel(int level) {
 685         for (;;) {
 686             int s = runState;
 687             if ((s & level) != 0)
 688                 return false;
 689             if (UNSAFE.compareAndSwapInt(this, runStateOffset, s, s | level))
 690                 return true;
 691         }
 692     }
 693 
 694     // workers array maintenance
 695 
 696     /**
 697      * Records and returns a workers array index for new worker.
 698      */
 699     private int recordWorker(ForkJoinWorkerThread w) {
 700         // Try using slot totalCount-1. If not available, scan and/or resize
 701         int k = (workerCounts >>> TOTAL_COUNT_SHIFT) - 1;
 702         final ReentrantLock lock = this.workerLock;
 703         lock.lock();
 704         try {
 705             ForkJoinWorkerThread[] ws = workers;
 706             int n = ws.length;
 707             if (k < 0 || k >= n || ws[k] != null) {
 708                 for (k = 0; k < n && ws[k] != null; ++k)
 709                     ;
 710                 if (k == n)
 711                     ws = workers = Arrays.copyOf(ws, n << 1);
 712             }
 713             ws[k] = w;
 714             int c = eventCount; // advance event count to ensure visibility
 715             UNSAFE.compareAndSwapInt(this, eventCountOffset, c, c+1);
 716         } finally {
 717             lock.unlock();
 718         }
 719         return k;
 720     }
 721 
 722     /**
 723      * Nulls out record of worker in workers array.
 724      */
 725     private void forgetWorker(ForkJoinWorkerThread w) {
 726         int idx = w.poolIndex;
 727         // Locking helps method recordWorker avoid unnecessary expansion
 728         final ReentrantLock lock = this.workerLock;
 729         lock.lock();
 730         try {
 731             ForkJoinWorkerThread[] ws = workers;
 732             if (idx >= 0 && idx < ws.length && ws[idx] == w) // verify
 733                 ws[idx] = null;
 734         } finally {
 735             lock.unlock();
 736         }
 737     }
 738 
 739     /**
 740      * Final callback from terminating worker.  Removes record of
 741      * worker from array, and adjusts counts. If pool is shutting
 742      * down, tries to complete termination.
 743      *
 744      * @param w the worker
 745      */
 746     final void workerTerminated(ForkJoinWorkerThread w) {
 747         forgetWorker(w);
 748         decrementWorkerCounts(w.isTrimmed() ? 0 : ONE_RUNNING, ONE_TOTAL);
 749         while (w.stealCount != 0) // collect final count
 750             tryAccumulateStealCount(w);
 751         tryTerminate(false);
 752     }
 753 
 754     // Waiting for and signalling events
 755 
 756     /**
 757      * Releases workers blocked on a count not equal to current count.
 758      * Normally called after precheck that eventWaiters isn't zero to
 759      * avoid wasted array checks. Gives up upon a change in count or
 760      * upon releasing four workers, letting others take over.
 761      */
 762     private void releaseEventWaiters() {
 763         ForkJoinWorkerThread[] ws = workers;
 764         int n = ws.length;
 765         long h = eventWaiters;
 766         int ec = eventCount;
 767         int releases = 4;
 768         ForkJoinWorkerThread w; int id;
 769         while ((id = (((int)h) & WAITER_ID_MASK) - 1) >= 0 &&
 770                (int)(h >>> EVENT_COUNT_SHIFT) != ec &&
 771                id < n && (w = ws[id]) != null) {
 772             if (UNSAFE.compareAndSwapLong(this, eventWaitersOffset,
 773                                           h,  w.nextWaiter)) {
 774                 LockSupport.unpark(w);
 775                 if (--releases == 0)
 776                     break;

 777             }
 778             if (eventCount != ec)
 779                 break;
 780             h = eventWaiters;
 781         }
 782     }
 783 
 784     /**
 785      * Tries to advance eventCount and releases waiters. Called only
 786      * from workers.
 787      */
 788     final void signalWork() {
 789         int c; // try to increment event count -- CAS failure OK
 790         UNSAFE.compareAndSwapInt(this, eventCountOffset, c = eventCount, c+1);
 791         if (eventWaiters != 0L)
 792             releaseEventWaiters();
 793     }
 794 
 795     /**
 796      * Adds the given worker to event queue and blocks until
 797      * terminating or event count advances from the given value
 798      *
 799      * @param w the calling worker thread
 800      * @param ec the count
 801      */
 802     private void eventSync(ForkJoinWorkerThread w, int ec) {
 803         long nh = (((long)ec) << EVENT_COUNT_SHIFT) | ((long)(w.poolIndex+1));
 804         long h;
 805         while ((runState < SHUTDOWN || !tryTerminate(false)) &&
 806                (((int)(h = eventWaiters) & WAITER_ID_MASK) == 0 ||
 807                 (int)(h >>> EVENT_COUNT_SHIFT) == ec) &&
 808                eventCount == ec) {
 809             if (UNSAFE.compareAndSwapLong(this, eventWaitersOffset,
 810                                           w.nextWaiter = h, nh)) {
 811                 awaitEvent(w, ec);
 812                 break;
 813             }
 814         }
 815     }
 816 
 817     /**
 818      * Blocks the given worker (that has already been entered as an
 819      * event waiter) until terminating or event count advances from
 820      * the given value. The oldest (first) waiter uses a timed wait to
 821      * occasionally one-by-one shrink the number of workers (to a
 822      * minimum of one) if the pool has not been used for extended
 823      * periods.
 824      *
 825      * @param w the calling worker thread
 826      * @param ec the count
 827      */
 828     private void awaitEvent(ForkJoinWorkerThread w, int ec) {
 829         while (eventCount == ec) {
 830             if (tryAccumulateStealCount(w)) { // transfer while idle
 831                 boolean untimed = (w.nextWaiter != 0L ||
 832                                    (workerCounts & RUNNING_COUNT_MASK) <= 1);
 833                 long startTime = untimed ? 0 : System.nanoTime();
 834                 Thread.interrupted();         // clear/ignore interrupt
 835                 if (w.isTerminating() || eventCount != ec)
 836                     break;                    // recheck after clear
 837                 if (untimed)
 838                     LockSupport.park(w);
 839                 else {
 840                     LockSupport.parkNanos(w, SHRINK_RATE_NANOS);
 841                     if (eventCount != ec || w.isTerminating())
 842                         break;
 843                     if (System.nanoTime() - startTime >= SHRINK_RATE_NANOS)
 844                         tryShutdownUnusedWorker(ec);
 845                 }
 846             }
 847         }
 848     }
 849 
 850     // Maintaining parallelism
 851 
 852     /**
 853      * Pushes worker onto the spare stack.
 854      */
 855     final void pushSpare(ForkJoinWorkerThread w) {
 856         int ns = (++w.spareCount << SPARE_COUNT_SHIFT) | (w.poolIndex + 1);
 857         do {} while (!UNSAFE.compareAndSwapInt(this, spareWaitersOffset,
 858                                                w.nextSpare = spareWaiters,ns));
 859     }
 860 
 861     /**
 862      * Tries (once) to resume a spare if the number of running
 863      * threads is less than target.
 864      */
 865     private void tryResumeSpare() {
 866         int sw, id;
 867         ForkJoinWorkerThread[] ws = workers;
 868         int n = ws.length;
 869         ForkJoinWorkerThread w;
 870         if ((sw = spareWaiters) != 0 &&
 871             (id = (sw & SPARE_ID_MASK) - 1) >= 0 &&
 872             id < n && (w = ws[id]) != null &&
 873             (runState >= TERMINATING ||
 874              (workerCounts & RUNNING_COUNT_MASK) < parallelism) &&
 875             spareWaiters == sw &&
 876             UNSAFE.compareAndSwapInt(this, spareWaitersOffset,
 877                                      sw, w.nextSpare)) {
 878             int c; // increment running count before resume
 879             do {} while (!UNSAFE.compareAndSwapInt
 880                          (this, workerCountsOffset,
 881                           c = workerCounts, c + ONE_RUNNING));
 882             if (w.tryUnsuspend())
 883                 LockSupport.unpark(w);
 884             else   // back out if w was shutdown
 885                 decrementWorkerCounts(ONE_RUNNING, 0);
 886         }
 887     }
 888 
 889     /**
 890      * Tries to increase the number of running workers if below target
 891      * parallelism: If a spare exists tries to resume it via
 892      * tryResumeSpare.  Otherwise, if not enough total workers or all
 893      * existing workers are busy, adds a new worker. In all cases also
 894      * helps wake up releasable workers waiting for work.
 895      */
 896     private void helpMaintainParallelism() {
 897         int pc = parallelism;
 898         int wc, rs, tc;
 899         while (((wc = workerCounts) & RUNNING_COUNT_MASK) < pc &&
 900                (rs = runState) < TERMINATING) {
 901             if (spareWaiters != 0)
 902                 tryResumeSpare();
 903             else if ((tc = wc >>> TOTAL_COUNT_SHIFT) >= MAX_WORKERS ||
 904                      (tc >= pc && (rs & ACTIVE_COUNT_MASK) != tc))
 905                 break;   // enough total
 906             else if (runState == rs && workerCounts == wc &&
 907                      UNSAFE.compareAndSwapInt(this, workerCountsOffset, wc,
 908                                               wc + (ONE_RUNNING|ONE_TOTAL))) {
 909                 ForkJoinWorkerThread w = null;
 910                 Throwable fail = null;
 911                 try {
 912                     w = factory.newThread(this);
 913                 } catch (Throwable ex) {
 914                     fail = ex;
 915                 }
 916                 if (w == null) { // null or exceptional factory return
 917                     decrementWorkerCounts(ONE_RUNNING, ONE_TOTAL);
 918                     tryTerminate(false); // handle failure during shutdown
 919                     // If originating from an external caller,
 920                     // propagate exception, else ignore
 921                     if (fail != null && runState < TERMINATING &&
 922                         !(Thread.currentThread() instanceof
 923                           ForkJoinWorkerThread))
 924                         UNSAFE.throwException(fail);
 925                     break;
 926                 }
 927                 w.start(recordWorker(w), ueh);
 928                 if ((workerCounts >>> TOTAL_COUNT_SHIFT) >= pc)



 929                     break; // add at most one unless total below target
 930             }
 931         }

 932         if (eventWaiters != 0L)
 933             releaseEventWaiters();
 934     }
 935 
 936     /**
 937      * Callback from the oldest waiter in awaitEvent waking up after a
 938      * period of non-use. If all workers are idle, tries (once) to
 939      * shutdown an event waiter or a spare, if one exists. Note that
 940      * we don't need CAS or locks here because the method is called
 941      * only from one thread occasionally waking (and even misfires are
 942      * OK). Note that until the shutdown worker fully terminates,
 943      * workerCounts will overestimate total count, which is tolerable.
 944      *
 945      * @param ec the event count waited on by caller (to abort
 946      * attempt if count has since changed).
 947      */
 948     private void tryShutdownUnusedWorker(int ec) {
 949         if (runState == 0 && eventCount == ec) { // only trigger if all idle
 950             ForkJoinWorkerThread[] ws = workers;
 951             int n = ws.length;
 952             ForkJoinWorkerThread w = null;
 953             boolean shutdown = false;
 954             int sw;
 955             long h;
 956             if ((sw = spareWaiters) != 0) { // prefer killing spares
 957                 int id = (sw & SPARE_ID_MASK) - 1;
 958                 if (id >= 0 && id < n && (w = ws[id]) != null &&
 959                     UNSAFE.compareAndSwapInt(this, spareWaitersOffset,
 960                                              sw, w.nextSpare))
 961                     shutdown = true;
 962             }
 963             else if ((h = eventWaiters) != 0L) {
 964                 long nh;
 965                 int id = (((int)h) & WAITER_ID_MASK) - 1;
 966                 if (id >= 0 && id < n && (w = ws[id]) != null &&
 967                     (nh = w.nextWaiter) != 0L && // keep at least one worker
 968                     UNSAFE.compareAndSwapLong(this, eventWaitersOffset, h, nh))
 969                     shutdown = true;
 970             }
 971             if (w != null && shutdown) {
 972                 w.shutdown();
 973                 LockSupport.unpark(w);
 974             }
 975         }
 976         releaseEventWaiters(); // in case of interference
 977     }
 978 
 979     /**
 980      * Callback from workers invoked upon each top-level action (i.e.,
 981      * stealing a task or taking a submission and running it).
 982      * Performs one or more of the following:
 983      *
 984      * 1. If the worker is active and either did not run a task
 985      *    or there are too many workers, try to set its active status
 986      *    to inactive and update activeCount. On contention, we may
 987      *    try again in this or a subsequent call.
 988      *
 989      * 2. If not enough total workers, help create some.
 990      *
 991      * 3. If there are too many running workers, suspend this worker
 992      *    (first forcing inactive if necessary).  If it is not needed,
 993      *    it may be shutdown while suspended (via
 994      *    tryShutdownUnusedWorker).  Otherwise, upon resume it
 995      *    rechecks running thread count and need for event sync.
 996      *
 997      * 4. If worker did not run a task, await the next task event via
 998      *    eventSync if necessary (first forcing inactivation), upon
 999      *    which the worker may be shutdown via
1000      *    tryShutdownUnusedWorker.  Otherwise, help release any
1001      *    existing event waiters that are now releasable,
1002      *
1003      * @param w the worker
1004      * @param ran true if worker ran a task since last call to this method
1005      */
1006     final void preStep(ForkJoinWorkerThread w, boolean ran) {
1007         int wec = w.lastEventCount;
1008         boolean active = w.active;
1009         boolean inactivate = false;
1010         int pc = parallelism;
1011         while (w.runState == 0) {
1012             int rs = runState;
1013             if (rs >= TERMINATING) {           // propagate shutdown
1014                 w.shutdown();
1015                 break;
1016             }
1017             if ((inactivate || (active && (rs & ACTIVE_COUNT_MASK) >= pc)) &&
1018                 UNSAFE.compareAndSwapInt(this, runStateOffset, rs, --rs)) {
1019                 inactivate = active = w.active = false;
1020                 if (rs == SHUTDOWN) {          // all inactive and shut down
1021                     tryTerminate(false);
1022                     continue;
1023                 }
1024             }
1025             int wc = workerCounts;             // try to suspend as spare
1026             if ((wc & RUNNING_COUNT_MASK) > pc) {
1027                 if (!(inactivate |= active) && // must inactivate to suspend
1028                     workerCounts == wc &&
1029                     UNSAFE.compareAndSwapInt(this, workerCountsOffset,
1030                                              wc, wc - ONE_RUNNING))
1031                     w.suspendAsSpare();
1032             }
1033             else if ((wc >>> TOTAL_COUNT_SHIFT) < pc)
1034                 helpMaintainParallelism();     // not enough workers
1035             else if (ran)
1036                 break;
1037             else {
1038                 long h = eventWaiters;
1039                 int ec = eventCount;
1040                 if (h != 0L && (int)(h >>> EVENT_COUNT_SHIFT) != ec)
1041                     releaseEventWaiters();     // release others before waiting
1042                 else if (ec != wec) {
1043                     w.lastEventCount = ec;     // no need to wait
1044                     break;
1045                 }
1046                 else if (!(inactivate |= active))
1047                     eventSync(w, wec);         // must inactivate before sync
1048             }


1049         }
1050     }
1051 
1052     /**
1053      * Helps and/or blocks awaiting join of the given task.
1054      * See above for explanation.
1055      *
1056      * @param joinMe the task to join
1057      * @param worker the current worker thread
1058      * @param timed true if wait should time out
1059      * @param nanos timeout value if timed
1060      */
1061     final void awaitJoin(ForkJoinTask<?> joinMe, ForkJoinWorkerThread worker,
1062                          boolean timed, long nanos) {
1063         long startTime = timed ? System.nanoTime() : 0L;
1064         int retries = 2 + (parallelism >> 2); // #helpJoins before blocking
1065         boolean running = true;               // false when count decremented
1066         while (joinMe.status >= 0) {
1067             if (runState >= TERMINATING) {
1068                 joinMe.cancelIgnoringExceptions();
1069                 break;
1070             }
1071             running = worker.helpJoinTask(joinMe, running);
1072             if (joinMe.status < 0)
1073                 break;
1074             if (retries > 0) {
1075                 --retries;
1076                 continue;
1077             }
1078             int wc = workerCounts;
1079             if ((wc & RUNNING_COUNT_MASK) != 0) {
1080                 if (running) {
1081                     if (!UNSAFE.compareAndSwapInt(this, workerCountsOffset,
1082                                                   wc, wc - ONE_RUNNING))
1083                         continue;
1084                     running = false;
1085                 }
1086                 long h = eventWaiters;
1087                 if (h != 0L && (int)(h >>> EVENT_COUNT_SHIFT) != eventCount)
1088                     releaseEventWaiters();
1089                 if ((workerCounts & RUNNING_COUNT_MASK) != 0) {
1090                     long ms; int ns;
1091                     if (!timed) {
1092                         ms = JOIN_TIMEOUT_MILLIS;
1093                         ns = 0;
1094                     }
1095                     else { // at most JOIN_TIMEOUT_MILLIS per wait
1096                         long nt = nanos - (System.nanoTime() - startTime);
1097                         if (nt <= 0L)
1098                             break;
1099                         ms = nt / 1000000;
1100                         if (ms > JOIN_TIMEOUT_MILLIS) {
1101                             ms = JOIN_TIMEOUT_MILLIS;
1102                             ns = 0;
1103                         }
1104                         else
1105                             ns = (int) (nt % 1000000);
1106                     }
1107                     joinMe.internalAwaitDone(ms, ns);
1108                 }
1109                 if (joinMe.status < 0)
1110                     break;
1111             }
1112             helpMaintainParallelism();
1113         }
1114         if (!running) {
1115             int c;
1116             do {} while (!UNSAFE.compareAndSwapInt
1117                          (this, workerCountsOffset,
1118                           c = workerCounts, c + ONE_RUNNING));


1119         }
1120     }

1121 
1122     /**
1123      * Same idea as awaitJoin, but no helping, retries, or timeouts.
1124      */
1125     final void awaitBlocker(ManagedBlocker blocker)
1126         throws InterruptedException {
1127         while (!blocker.isReleasable()) {
1128             int wc = workerCounts;
1129             if ((wc & RUNNING_COUNT_MASK) == 0)
1130                 helpMaintainParallelism();
1131             else if (UNSAFE.compareAndSwapInt(this, workerCountsOffset,
1132                                               wc, wc - ONE_RUNNING)) {
1133                 try {
1134                     while (!blocker.isReleasable()) {
1135                         long h = eventWaiters;
1136                         if (h != 0L &&
1137                             (int)(h >>> EVENT_COUNT_SHIFT) != eventCount)
1138                             releaseEventWaiters();
1139                         else if ((workerCounts & RUNNING_COUNT_MASK) == 0 &&
1140                                  runState < TERMINATING)
1141                             helpMaintainParallelism();
1142                         else if (blocker.block())
1143                             break;
1144                     }
1145                 } finally {
1146                     int c;
1147                     do {} while (!UNSAFE.compareAndSwapInt
1148                                  (this, workerCountsOffset,
1149                                   c = workerCounts, c + ONE_RUNNING));
1150                 }
1151                 break;
1152             }
1153         }
1154     }
1155 
1156     /**
1157      * Possibly initiates and/or completes termination.
1158      *
1159      * @param now if true, unconditionally terminate, else only
1160      * if shutdown and empty queue and no active workers
1161      * @return true if now terminating or terminated
1162      */
1163     private boolean tryTerminate(boolean now) {
1164         if (now)
1165             advanceRunLevel(SHUTDOWN); // ensure at least SHUTDOWN
1166         else if (runState < SHUTDOWN ||
1167                  !submissionQueue.isEmpty() ||
1168                  (runState & ACTIVE_COUNT_MASK) != 0)
1169             return false;
1170 
1171         if (advanceRunLevel(TERMINATING))
1172             startTerminating();
1173 
1174         // Finish now if all threads terminated; else in some subsequent call
1175         if ((workerCounts >>> TOTAL_COUNT_SHIFT) == 0) {
1176             advanceRunLevel(TERMINATED);
1177             termination.forceTermination();
1178         }
1179         return true;
1180     }
1181 

1182     /**
1183      * Actions on transition to TERMINATING
1184      *
1185      * Runs up to four passes through workers: (0) shutting down each
1186      * (without waking up if parked) to quickly spread notifications
1187      * without unnecessary bouncing around event queues etc (1) wake
1188      * up and help cancel tasks (2) interrupt (3) mop up races with
1189      * interrupted workers
1190      */
1191     private void startTerminating() {
1192         cancelSubmissions();
1193         for (int passes = 0; passes < 4 && workerCounts != 0; ++passes) {
1194             int c; // advance event count
1195             UNSAFE.compareAndSwapInt(this, eventCountOffset,
1196                                      c = eventCount, c+1);
1197             eventWaiters = 0L; // clobber lists
1198             spareWaiters = 0;
1199             for (ForkJoinWorkerThread w : workers) {
1200                 if (w != null) {
1201                     w.shutdown();
1202                     if (passes > 0 && !w.isTerminated()) {
1203                         w.cancelTasks();
1204                         LockSupport.unpark(w);
1205                         if (passes > 1 && !w.isInterrupted()) {
1206                             try {
1207                                 w.interrupt();
1208                             } catch (SecurityException ignore) {
1209                             }
1210                         }
1211                     }
1212                 }
1213             }
1214         }
1215     }
1216 
1217     /**
1218      * Clears out and cancels submissions, ignoring exceptions.
1219      */
1220     private void cancelSubmissions() {
1221         ForkJoinTask<?> task;
1222         while ((task = submissionQueue.poll()) != null) {
1223             try {
1224                 task.cancel(false);
1225             } catch (Throwable ignore) {
1226             }
1227         }
1228     }
1229 
1230     // misc support for ForkJoinWorkerThread
1231 
1232     /**
1233      * Returns pool number.
1234      */
1235     final int getPoolNumber() {
1236         return poolNumber;
1237     }
1238 
1239     /**
1240      * Tries to accumulate steal count from a worker, clearing
1241      * the worker's value if successful.
1242      *
1243      * @return true if worker steal count now zero
1244      */
1245     final boolean tryAccumulateStealCount(ForkJoinWorkerThread w) {
1246         int sc = w.stealCount;
1247         long c = stealCount;
1248         // CAS even if zero, for fence effects
1249         if (UNSAFE.compareAndSwapLong(this, stealCountOffset, c, c + sc)) {
1250             if (sc != 0)
1251                 w.stealCount = 0;
1252             return true;
1253         }
1254         return sc == 0;
1255     }
1256 
1257     /**
1258      * Returns the approximate (non-atomic) number of idle threads per
1259      * active thread.
1260      */
1261     final int idlePerActive() {
1262         int pc = parallelism; // use parallelism, not rc
1263         int ac = runState;    // no mask -- artificially boosts during shutdown
1264         // Use exact results for small values, saturate past 4
1265         return ((pc <= ac) ? 0 :
1266                 (pc >>> 1 <= ac) ? 1 :
1267                 (pc >>> 2 <= ac) ? 3 :
1268                 pc >>> 3);
1269     }
1270 
1271     // Public and protected methods
1272 
1273     // Constructors
1274 
1275     /**
1276      * Creates a {@code ForkJoinPool} with parallelism equal to {@link
1277      * java.lang.Runtime#availableProcessors}, using the {@linkplain
1278      * #defaultForkJoinWorkerThreadFactory default thread factory},
1279      * no UncaughtExceptionHandler, and non-async LIFO processing mode.
1280      *
1281      * @throws SecurityException if a security manager exists and
1282      *         the caller is not permitted to modify threads
1283      *         because it does not hold {@link
1284      *         java.lang.RuntimePermission}{@code ("modifyThread")}
1285      */
1286     public ForkJoinPool() {
1287         this(Runtime.getRuntime().availableProcessors(),
1288              defaultForkJoinWorkerThreadFactory, null, false);
1289     }
1290 
1291     /**
1292      * Creates a {@code ForkJoinPool} with the indicated parallelism
1293      * level, the {@linkplain
1294      * #defaultForkJoinWorkerThreadFactory default thread factory},
1295      * no UncaughtExceptionHandler, and non-async LIFO processing mode.
1296      *
1297      * @param parallelism the parallelism level
1298      * @throws IllegalArgumentException if parallelism less than or
1299      *         equal to zero, or greater than implementation limit
1300      * @throws SecurityException if a security manager exists and
1301      *         the caller is not permitted to modify threads
1302      *         because it does not hold {@link
1303      *         java.lang.RuntimePermission}{@code ("modifyThread")}
1304      */
1305     public ForkJoinPool(int parallelism) {
1306         this(parallelism, defaultForkJoinWorkerThreadFactory, null, false);
1307     }
1308 
1309     /**
1310      * Creates a {@code ForkJoinPool} with the given parameters.
1311      *
1312      * @param parallelism the parallelism level. For default value,
1313      * use {@link java.lang.Runtime#availableProcessors}.
1314      * @param factory the factory for creating new threads. For default value,
1315      * use {@link #defaultForkJoinWorkerThreadFactory}.
1316      * @param handler the handler for internal worker threads that
1317      * terminate due to unrecoverable errors encountered while executing
1318      * tasks. For default value, use {@code null}.
1319      * @param asyncMode if true,
1320      * establishes local first-in-first-out scheduling mode for forked
1321      * tasks that are never joined. This mode may be more appropriate
1322      * than default locally stack-based mode in applications in which
1323      * worker threads only process event-style asynchronous tasks.
1324      * For default value, use {@code false}.
1325      * @throws IllegalArgumentException if parallelism less than or
1326      *         equal to zero, or greater than implementation limit
1327      * @throws NullPointerException if the factory is null
1328      * @throws SecurityException if a security manager exists and
1329      *         the caller is not permitted to modify threads
1330      *         because it does not hold {@link
1331      *         java.lang.RuntimePermission}{@code ("modifyThread")}
1332      */
1333     public ForkJoinPool(int parallelism,
1334                         ForkJoinWorkerThreadFactory factory,
1335                         Thread.UncaughtExceptionHandler handler,
1336                         boolean asyncMode) {
1337         checkPermission();
1338         if (factory == null)
1339             throw new NullPointerException();
1340         if (parallelism <= 0 || parallelism > MAX_WORKERS)
1341             throw new IllegalArgumentException();
1342         this.parallelism = parallelism;
1343         this.factory = factory;
1344         this.ueh = handler;
1345         this.locallyFifo = asyncMode;
1346         int arraySize = initialArraySizeFor(parallelism);
1347         this.workers = new ForkJoinWorkerThread[arraySize];
1348         this.submissionQueue = new LinkedTransferQueue<ForkJoinTask<?>>();
1349         this.workerLock = new ReentrantLock();
1350         this.termination = new Phaser(1);
1351         this.poolNumber = poolNumberGenerator.incrementAndGet();
1352     }
1353 
1354     /**
1355      * Returns initial power of two size for workers array.
1356      * @param pc the initial parallelism level
1357      */
1358     private static int initialArraySizeFor(int pc) {
1359         // If possible, initially allocate enough space for one spare
1360         int size = pc < MAX_WORKERS ? pc + 1 : MAX_WORKERS;
1361         // See Hackers Delight, sec 3.2. We know MAX_WORKERS < (1 >>> 16)
1362         size |= size >>> 1;
1363         size |= size >>> 2;
1364         size |= size >>> 4;
1365         size |= size >>> 8;
1366         return size + 1;
1367     }
1368 
1369     // Execution methods
1370 
1371     /**
1372      * Submits task and creates, starts, or resumes some workers if necessary
1373      */
1374     private <T> void doSubmit(ForkJoinTask<T> task) {




1375         submissionQueue.offer(task);
1376         int c; // try to increment event count -- CAS failure OK
1377         UNSAFE.compareAndSwapInt(this, eventCountOffset, c = eventCount, c+1);
1378         helpMaintainParallelism();
1379     }
1380 
1381     /**
1382      * Performs the given task, returning its result upon completion.
1383      *
1384      * @param task the task
1385      * @return the task's result
1386      * @throws NullPointerException if the task is null
1387      * @throws RejectedExecutionException if the task cannot be
1388      *         scheduled for execution
1389      */
1390     public <T> T invoke(ForkJoinTask<T> task) {
1391         if (task == null)
1392             throw new NullPointerException();
1393         if (runState >= SHUTDOWN)
1394             throw new RejectedExecutionException();
1395         Thread t = Thread.currentThread();
1396         if ((t instanceof ForkJoinWorkerThread) &&
1397             ((ForkJoinWorkerThread)t).pool == this)
1398             return task.invoke();  // bypass submit if in same pool
1399         else {
1400             doSubmit(task);
1401             return task.join();
1402         }
1403     }
1404 
1405     /**
1406      * Unless terminating, forks task if within an ongoing FJ
1407      * computation in the current pool, else submits as external task.
1408      */
1409     private <T> void forkOrSubmit(ForkJoinTask<T> task) {
1410         if (runState >= SHUTDOWN)
1411             throw new RejectedExecutionException();
1412         Thread t = Thread.currentThread();
1413         if ((t instanceof ForkJoinWorkerThread) &&
1414             ((ForkJoinWorkerThread)t).pool == this)
1415             task.fork();
1416         else
1417             doSubmit(task);
1418     }
1419 
1420     /**
1421      * Arranges for (asynchronous) execution of the given task.
1422      *
1423      * @param task the task
1424      * @throws NullPointerException if the task is null
1425      * @throws RejectedExecutionException if the task cannot be
1426      *         scheduled for execution
1427      */
1428     public void execute(ForkJoinTask<?> task) {
1429         if (task == null)
1430             throw new NullPointerException();
1431         forkOrSubmit(task);
1432     }
1433 
1434     // AbstractExecutorService methods
1435 
1436     /**
1437      * @throws NullPointerException if the task is null
1438      * @throws RejectedExecutionException if the task cannot be
1439      *         scheduled for execution
1440      */
1441     public void execute(Runnable task) {
1442         if (task == null)
1443             throw new NullPointerException();
1444         ForkJoinTask<?> job;
1445         if (task instanceof ForkJoinTask<?>) // avoid re-wrap
1446             job = (ForkJoinTask<?>) task;
1447         else
1448             job = ForkJoinTask.adapt(task, null);
1449         forkOrSubmit(job);
1450     }
1451 
1452     /**
1453      * Submits a ForkJoinTask for execution.
1454      *
1455      * @param task the task to submit
1456      * @return the task
1457      * @throws NullPointerException if the task is null
1458      * @throws RejectedExecutionException if the task cannot be
1459      *         scheduled for execution
1460      */
1461     public <T> ForkJoinTask<T> submit(ForkJoinTask<T> task) {
1462         if (task == null)
1463             throw new NullPointerException();
1464         forkOrSubmit(task);
1465         return task;
1466     }
1467 
1468     /**
1469      * @throws NullPointerException if the task is null
1470      * @throws RejectedExecutionException if the task cannot be
1471      *         scheduled for execution
1472      */
1473     public <T> ForkJoinTask<T> submit(Callable<T> task) {
1474         if (task == null)
1475             throw new NullPointerException();
1476         ForkJoinTask<T> job = ForkJoinTask.adapt(task);
1477         forkOrSubmit(job);
1478         return job;
1479     }
1480 
1481     /**
1482      * @throws NullPointerException if the task is null
1483      * @throws RejectedExecutionException if the task cannot be
1484      *         scheduled for execution
1485      */
1486     public <T> ForkJoinTask<T> submit(Runnable task, T result) {
1487         if (task == null)
1488             throw new NullPointerException();
1489         ForkJoinTask<T> job = ForkJoinTask.adapt(task, result);
1490         forkOrSubmit(job);
1491         return job;
1492     }
1493 
1494     /**
1495      * @throws NullPointerException if the task is null
1496      * @throws RejectedExecutionException if the task cannot be
1497      *         scheduled for execution
1498      */
1499     public ForkJoinTask<?> submit(Runnable task) {
1500         if (task == null)
1501             throw new NullPointerException();
1502         ForkJoinTask<?> job;
1503         if (task instanceof ForkJoinTask<?>) // avoid re-wrap
1504             job = (ForkJoinTask<?>) task;
1505         else
1506             job = ForkJoinTask.adapt(task, null);
1507         forkOrSubmit(job);
1508         return job;
1509     }
1510 
1511     /**
1512      * @throws NullPointerException       {@inheritDoc}
1513      * @throws RejectedExecutionException {@inheritDoc}
1514      */
1515     public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks) {
1516         ArrayList<ForkJoinTask<T>> forkJoinTasks =
1517             new ArrayList<ForkJoinTask<T>>(tasks.size());
1518         for (Callable<T> task : tasks)
1519             forkJoinTasks.add(ForkJoinTask.adapt(task));
1520         invoke(new InvokeAll<T>(forkJoinTasks));
1521 
1522         @SuppressWarnings({"unchecked", "rawtypes"})
1523             List<Future<T>> futures = (List<Future<T>>) (List) forkJoinTasks;
1524         return futures;
1525     }
1526 
1527     static final class InvokeAll<T> extends RecursiveAction {
1528         final ArrayList<ForkJoinTask<T>> tasks;
1529         InvokeAll(ArrayList<ForkJoinTask<T>> tasks) { this.tasks = tasks; }
1530         public void compute() {
1531             try { invokeAll(tasks); }
1532             catch (Exception ignore) {}
1533         }
1534         private static final long serialVersionUID = -7914297376763021607L;
1535     }
1536 
1537     /**
1538      * Returns the factory used for constructing new workers.
1539      *
1540      * @return the factory used for constructing new workers
1541      */
1542     public ForkJoinWorkerThreadFactory getFactory() {
1543         return factory;
1544     }
1545 
1546     /**
1547      * Returns the handler for internal worker threads that terminate
1548      * due to unrecoverable errors encountered while executing tasks.
1549      *
1550      * @return the handler, or {@code null} if none
1551      */
1552     public Thread.UncaughtExceptionHandler getUncaughtExceptionHandler() {
1553         return ueh;
1554     }
1555 
1556     /**
1557      * Returns the targeted parallelism level of this pool.
1558      *
1559      * @return the targeted parallelism level of this pool
1560      */
1561     public int getParallelism() {
1562         return parallelism;
1563     }
1564 
1565     /**
1566      * Returns the number of worker threads that have started but not
1567      * yet terminated.  The result returned by this method may differ
1568      * from {@link #getParallelism} when threads are created to
1569      * maintain parallelism when others are cooperatively blocked.
1570      *
1571      * @return the number of worker threads
1572      */
1573     public int getPoolSize() {
1574         return workerCounts >>> TOTAL_COUNT_SHIFT;
1575     }
1576 
1577     /**
1578      * Returns {@code true} if this pool uses local first-in-first-out
1579      * scheduling mode for forked tasks that are never joined.
1580      *
1581      * @return {@code true} if this pool uses async mode
1582      */
1583     public boolean getAsyncMode() {
1584         return locallyFifo;
1585     }
1586 
1587     /**
1588      * Returns an estimate of the number of worker threads that are
1589      * not blocked waiting to join tasks or for other managed
1590      * synchronization. This method may overestimate the
1591      * number of running threads.
1592      *
1593      * @return the number of worker threads
1594      */
1595     public int getRunningThreadCount() {
1596         return workerCounts & RUNNING_COUNT_MASK;
1597     }
1598 
1599     /**
1600      * Returns an estimate of the number of threads that are currently
1601      * stealing or executing tasks. This method may overestimate the
1602      * number of active threads.
1603      *
1604      * @return the number of active threads
1605      */
1606     public int getActiveThreadCount() {
1607         return runState & ACTIVE_COUNT_MASK;
1608     }
1609 
1610     /**
1611      * Returns {@code true} if all worker threads are currently idle.
1612      * An idle worker is one that cannot obtain a task to execute
1613      * because none are available to steal from other threads, and
1614      * there are no pending submissions to the pool. This method is
1615      * conservative; it might not return {@code true} immediately upon
1616      * idleness of all threads, but will eventually become true if
1617      * threads remain inactive.
1618      *
1619      * @return {@code true} if all threads are currently idle
1620      */
1621     public boolean isQuiescent() {
1622         return (runState & ACTIVE_COUNT_MASK) == 0;
1623     }
1624 
1625     /**
1626      * Returns an estimate of the total number of tasks stolen from
1627      * one thread's work queue by another. The reported value
1628      * underestimates the actual total number of steals when the pool
1629      * is not quiescent. This value may be useful for monitoring and
1630      * tuning fork/join programs: in general, steal counts should be
1631      * high enough to keep threads busy, but low enough to avoid
1632      * overhead and contention across threads.
1633      *
1634      * @return the number of steals
1635      */
1636     public long getStealCount() {
1637         return stealCount;
1638     }
1639 
1640     /**
1641      * Returns an estimate of the total number of tasks currently held
1642      * in queues by worker threads (but not including tasks submitted
1643      * to the pool that have not begun executing). This value is only
1644      * an approximation, obtained by iterating across all threads in
1645      * the pool. This method may be useful for tuning task
1646      * granularities.
1647      *
1648      * @return the number of queued tasks
1649      */
1650     public long getQueuedTaskCount() {
1651         long count = 0;
1652         for (ForkJoinWorkerThread w : workers)
1653             if (w != null)
1654                 count += w.getQueueSize();
1655         return count;
1656     }
1657 
1658     /**
1659      * Returns an estimate of the number of tasks submitted to this
1660      * pool that have not yet begun executing.  This method takes time
1661      * proportional to the number of submissions.
1662      *
1663      * @return the number of queued submissions
1664      */
1665     public int getQueuedSubmissionCount() {
1666         return submissionQueue.size();
1667     }
1668 
1669     /**
1670      * Returns {@code true} if there are any tasks submitted to this
1671      * pool that have not yet begun executing.
1672      *
1673      * @return {@code true} if there are any queued submissions
1674      */
1675     public boolean hasQueuedSubmissions() {
1676         return !submissionQueue.isEmpty();
1677     }
1678 
1679     /**
1680      * Removes and returns the next unexecuted submission if one is
1681      * available.  This method may be useful in extensions to this
1682      * class that re-assign work in systems with multiple pools.
1683      *
1684      * @return the next submission, or {@code null} if none
1685      */
1686     protected ForkJoinTask<?> pollSubmission() {
1687         return submissionQueue.poll();
1688     }
1689 
1690     /**
1691      * Removes all available unexecuted submitted and forked tasks
1692      * from scheduling queues and adds them to the given collection,
1693      * without altering their execution status. These may include
1694      * artificially generated or wrapped tasks. This method is
1695      * designed to be invoked only when the pool is known to be
1696      * quiescent. Invocations at other times may not remove all
1697      * tasks. A failure encountered while attempting to add elements
1698      * to collection {@code c} may result in elements being in
1699      * neither, either or both collections when the associated
1700      * exception is thrown.  The behavior of this operation is
1701      * undefined if the specified collection is modified while the
1702      * operation is in progress.
1703      *
1704      * @param c the collection to transfer elements into
1705      * @return the number of elements transferred
1706      */
1707     protected int drainTasksTo(Collection<? super ForkJoinTask<?>> c) {
1708         int count = submissionQueue.drainTo(c);
1709         for (ForkJoinWorkerThread w : workers)
1710             if (w != null)
1711                 count += w.drainTasksTo(c);
1712         return count;
1713     }
1714 
1715     /**
1716      * Returns a string identifying this pool, as well as its state,
1717      * including indications of run state, parallelism level, and
1718      * worker and task counts.
1719      *
1720      * @return a string identifying this pool, as well as its state
1721      */
1722     public String toString() {
1723         long st = getStealCount();
1724         long qt = getQueuedTaskCount();
1725         long qs = getQueuedSubmissionCount();
1726         int wc = workerCounts;
1727         int tc = wc >>> TOTAL_COUNT_SHIFT;
1728         int rc = wc & RUNNING_COUNT_MASK;
1729         int pc = parallelism;
1730         int rs = runState;
1731         int ac = rs & ACTIVE_COUNT_MASK;
1732         return super.toString() +
1733             "[" + runLevelToString(rs) +
1734             ", parallelism = " + pc +
1735             ", size = " + tc +
1736             ", active = " + ac +
1737             ", running = " + rc +
1738             ", steals = " + st +
1739             ", tasks = " + qt +
1740             ", submissions = " + qs +
1741             "]";
1742     }
1743 
1744     private static String runLevelToString(int s) {
1745         return ((s & TERMINATED) != 0 ? "Terminated" :
1746                 ((s & TERMINATING) != 0 ? "Terminating" :
1747                  ((s & SHUTDOWN) != 0 ? "Shutting down" :
1748                   "Running")));
1749     }
1750 
1751     /**
1752      * Initiates an orderly shutdown in which previously submitted
1753      * tasks are executed, but no new tasks will be accepted.
1754      * Invocation has no additional effect if already shut down.
1755      * Tasks that are in the process of being submitted concurrently
1756      * during the course of this method may or may not be rejected.
1757      *
1758      * @throws SecurityException if a security manager exists and
1759      *         the caller is not permitted to modify threads
1760      *         because it does not hold {@link
1761      *         java.lang.RuntimePermission}{@code ("modifyThread")}
1762      */
1763     public void shutdown() {
1764         checkPermission();
1765         advanceRunLevel(SHUTDOWN);
1766         tryTerminate(false);
1767     }
1768 
1769     /**
1770      * Attempts to cancel and/or stop all tasks, and reject all
1771      * subsequently submitted tasks.  Tasks that are in the process of
1772      * being submitted or executed concurrently during the course of
1773      * this method may or may not be rejected. This method cancels
1774      * both existing and unexecuted tasks, in order to permit
1775      * termination in the presence of task dependencies. So the method
1776      * always returns an empty list (unlike the case for some other
1777      * Executors).
1778      *
1779      * @return an empty list
1780      * @throws SecurityException if a security manager exists and
1781      *         the caller is not permitted to modify threads
1782      *         because it does not hold {@link
1783      *         java.lang.RuntimePermission}{@code ("modifyThread")}
1784      */
1785     public List<Runnable> shutdownNow() {
1786         checkPermission();
1787         tryTerminate(true);
1788         return Collections.emptyList();
1789     }
1790 
1791     /**
1792      * Returns {@code true} if all tasks have completed following shut down.
1793      *
1794      * @return {@code true} if all tasks have completed following shut down
1795      */
1796     public boolean isTerminated() {
1797         return runState >= TERMINATED;
1798     }
1799 
1800     /**
1801      * Returns {@code true} if the process of termination has
1802      * commenced but not yet completed.  This method may be useful for
1803      * debugging. A return of {@code true} reported a sufficient
1804      * period after shutdown may indicate that submitted tasks have
1805      * ignored or suppressed interruption, or are waiting for IO,
1806      * causing this executor not to properly terminate. (See the
1807      * advisory notes for class {@link ForkJoinTask} stating that
1808      * tasks should not normally entail blocking operations.  But if
1809      * they do, they must abort them on interrupt.)
1810      *
1811      * @return {@code true} if terminating but not yet terminated
1812      */
1813     public boolean isTerminating() {
1814         return (runState & (TERMINATING|TERMINATED)) == TERMINATING;
1815     }
1816 
1817     /**
1818      * Returns true if terminating or terminated. Used by ForkJoinWorkerThread.
1819      */
1820     final boolean isAtLeastTerminating() {
1821         return runState >= TERMINATING;
1822     }
1823 
1824     /**
1825      * Returns {@code true} if this pool has been shut down.
1826      *
1827      * @return {@code true} if this pool has been shut down
1828      */
1829     public boolean isShutdown() {
1830         return runState >= SHUTDOWN;
1831     }
1832 
1833     /**
1834      * Blocks until all tasks have completed execution after a shutdown
1835      * request, or the timeout occurs, or the current thread is
1836      * interrupted, whichever happens first.
1837      *
1838      * @param timeout the maximum time to wait
1839      * @param unit the time unit of the timeout argument
1840      * @return {@code true} if this executor terminated and
1841      *         {@code false} if the timeout elapsed before termination
1842      * @throws InterruptedException if interrupted while waiting
1843      */
1844     public boolean awaitTermination(long timeout, TimeUnit unit)
1845         throws InterruptedException {
1846         try {
1847             termination.awaitAdvanceInterruptibly(0, timeout, unit);
1848         } catch (TimeoutException ex) {
1849             return false;
1850         }
1851         return true;
1852     }
1853 
1854     /**
1855      * Interface for extending managed parallelism for tasks running
1856      * in {@link ForkJoinPool}s.
1857      *
1858      * <p>A {@code ManagedBlocker} provides two methods.  Method
1859      * {@code isReleasable} must return {@code true} if blocking is
1860      * not necessary. Method {@code block} blocks the current thread
1861      * if necessary (perhaps internally invoking {@code isReleasable}
1862      * before actually blocking). The unusual methods in this API
1863      * accommodate synchronizers that may, but don't usually, block
1864      * for long periods. Similarly, they allow more efficient internal
1865      * handling of cases in which additional workers may be, but
1866      * usually are not, needed to ensure sufficient parallelism.
1867      * Toward this end, implementations of method {@code isReleasable}
1868      * must be amenable to repeated invocation.
1869      *
1870      * <p>For example, here is a ManagedBlocker based on a
1871      * ReentrantLock:
1872      *  <pre> {@code
1873      * class ManagedLocker implements ManagedBlocker {
1874      *   final ReentrantLock lock;
1875      *   boolean hasLock = false;
1876      *   ManagedLocker(ReentrantLock lock) { this.lock = lock; }
1877      *   public boolean block() {
1878      *     if (!hasLock)
1879      *       lock.lock();
1880      *     return true;
1881      *   }
1882      *   public boolean isReleasable() {
1883      *     return hasLock || (hasLock = lock.tryLock());
1884      *   }
1885      * }}</pre>
1886      *
1887      * <p>Here is a class that possibly blocks waiting for an
1888      * item on a given queue:
1889      *  <pre> {@code
1890      * class QueueTaker<E> implements ManagedBlocker {
1891      *   final BlockingQueue<E> queue;
1892      *   volatile E item = null;
1893      *   QueueTaker(BlockingQueue<E> q) { this.queue = q; }
1894      *   public boolean block() throws InterruptedException {
1895      *     if (item == null)
1896      *       item = queue.take();
1897      *     return true;
1898      *   }
1899      *   public boolean isReleasable() {
1900      *     return item != null || (item = queue.poll()) != null;
1901      *   }
1902      *   public E getItem() { // call after pool.managedBlock completes
1903      *     return item;
1904      *   }
1905      * }}</pre>
1906      */
1907     public static interface ManagedBlocker {
1908         /**
1909          * Possibly blocks the current thread, for example waiting for
1910          * a lock or condition.
1911          *
1912          * @return {@code true} if no additional blocking is necessary
1913          * (i.e., if isReleasable would return true)
1914          * @throws InterruptedException if interrupted while waiting
1915          * (the method is not required to do so, but is allowed to)
1916          */
1917         boolean block() throws InterruptedException;
1918 
1919         /**
1920          * Returns {@code true} if blocking is unnecessary.
1921          */
1922         boolean isReleasable();
1923     }
1924 
1925     /**
1926      * Blocks in accord with the given blocker.  If the current thread
1927      * is a {@link ForkJoinWorkerThread}, this method possibly
1928      * arranges for a spare thread to be activated if necessary to
1929      * ensure sufficient parallelism while the current thread is blocked.
1930      *
1931      * <p>If the caller is not a {@link ForkJoinTask}, this method is
1932      * behaviorally equivalent to
1933      *  <pre> {@code
1934      * while (!blocker.isReleasable())
1935      *   if (blocker.block())
1936      *     return;
1937      * }</pre>
1938      *
1939      * If the caller is a {@code ForkJoinTask}, then the pool may
1940      * first be expanded to ensure parallelism, and later adjusted.
1941      *
1942      * @param blocker the blocker
1943      * @throws InterruptedException if blocker.block did so
1944      */
1945     public static void managedBlock(ManagedBlocker blocker)
1946         throws InterruptedException {
1947         Thread t = Thread.currentThread();
1948         if (t instanceof ForkJoinWorkerThread) {
1949             ForkJoinWorkerThread w = (ForkJoinWorkerThread) t;
1950             w.pool.awaitBlocker(blocker);
1951         }
1952         else {
1953             do {} while (!blocker.isReleasable() && !blocker.block());
1954         }
1955     }
1956 
1957     // AbstractExecutorService overrides.  These rely on undocumented
1958     // fact that ForkJoinTask.adapt returns ForkJoinTasks that also
1959     // implement RunnableFuture.
1960 
1961     protected <T> RunnableFuture<T> newTaskFor(Runnable runnable, T value) {
1962         return (RunnableFuture<T>) ForkJoinTask.adapt(runnable, value);
1963     }
1964 
1965     protected <T> RunnableFuture<T> newTaskFor(Callable<T> callable) {
1966         return (RunnableFuture<T>) ForkJoinTask.adapt(callable);
1967     }
1968 
1969     // Unsafe mechanics
1970 
1971     private static final sun.misc.Unsafe UNSAFE = sun.misc.Unsafe.getUnsafe();
1972     private static final long workerCountsOffset =
1973         objectFieldOffset("workerCounts", ForkJoinPool.class);
1974     private static final long runStateOffset =
1975         objectFieldOffset("runState", ForkJoinPool.class);
1976     private static final long eventCountOffset =
1977         objectFieldOffset("eventCount", ForkJoinPool.class);
1978     private static final long eventWaitersOffset =
1979         objectFieldOffset("eventWaiters", ForkJoinPool.class);
1980     private static final long stealCountOffset =
1981         objectFieldOffset("stealCount", ForkJoinPool.class);
1982     private static final long spareWaitersOffset =
1983         objectFieldOffset("spareWaiters", ForkJoinPool.class);
1984 
1985     private static long objectFieldOffset(String field, Class<?> klazz) {
1986         try {
1987             return UNSAFE.objectFieldOffset(klazz.getDeclaredField(field));
1988         } catch (NoSuchFieldException e) {
1989             // Convert Exception to corresponding Error
1990             NoSuchFieldError error = new NoSuchFieldError(field);
1991             error.initCause(e);
1992             throw error;
1993         }
1994     }
1995 }
--- EOF ---