Home | History | Annotate | Download | only in concurrent
      1 /*
      2  * Written by Doug Lea with assistance from members of JCP JSR-166
      3  * Expert Group and released to the public domain, as explained at
      4  * http://creativecommons.org/publicdomain/zero/1.0/
      5  */
      6 
      7 package java.util.concurrent;
      8 
      9 import java.io.Serializable;
     10 import java.util.Collection;
     11 import java.util.List;
     12 import java.util.RandomAccess;
     13 import java.lang.ref.WeakReference;
     14 import java.lang.ref.ReferenceQueue;
     15 import java.util.concurrent.Callable;
     16 import java.util.concurrent.CancellationException;
     17 import java.util.concurrent.ExecutionException;
     18 import java.util.concurrent.Future;
     19 import java.util.concurrent.RejectedExecutionException;
     20 import java.util.concurrent.RunnableFuture;
     21 import java.util.concurrent.TimeUnit;
     22 import java.util.concurrent.TimeoutException;
     23 import java.util.concurrent.locks.ReentrantLock;
     24 import java.lang.reflect.Constructor;
     25 
     26 /**
     27  * Abstract base class for tasks that run within a {@link ForkJoinPool}.
     28  * A {@code ForkJoinTask} is a thread-like entity that is much
     29  * lighter weight than a normal thread.  Huge numbers of tasks and
     30  * subtasks may be hosted by a small number of actual threads in a
     31  * ForkJoinPool, at the price of some usage limitations.
     32  *
     33  * <p>A "main" {@code ForkJoinTask} begins execution when it is
     34  * explicitly submitted to a {@link ForkJoinPool}, or, if not already
     35  * engaged in a ForkJoin computation, commenced in the {@link
     36  * ForkJoinPool#commonPool()} via {@link #fork}, {@link #invoke}, or
     37  * related methods.  Once started, it will usually in turn start other
     38  * subtasks.  As indicated by the name of this class, many programs
     39  * using {@code ForkJoinTask} employ only methods {@link #fork} and
     40  * {@link #join}, or derivatives such as {@link
     41  * #invokeAll(ForkJoinTask...) invokeAll}.  However, this class also
     42  * provides a number of other methods that can come into play in
     43  * advanced usages, as well as extension mechanics that allow support
     44  * of new forms of fork/join processing.
     45  *
     46  * <p>A {@code ForkJoinTask} is a lightweight form of {@link Future}.
     47  * The efficiency of {@code ForkJoinTask}s stems from a set of
     48  * restrictions (that are only partially statically enforceable)
     49  * reflecting their main use as computational tasks calculating pure
     50  * functions or operating on purely isolated objects.  The primary
     51  * coordination mechanisms are {@link #fork}, that arranges
     52  * asynchronous execution, and {@link #join}, that doesn't proceed
     53  * until the task's result has been computed.  Computations should
     54  * ideally avoid {@code synchronized} methods or blocks, and should
     55  * minimize other blocking synchronization apart from joining other
     56  * tasks or using synchronizers such as Phasers that are advertised to
     57  * cooperate with fork/join scheduling. Subdividable tasks should also
     58  * not perform blocking I/O, and should ideally access variables that
     59  * are completely independent of those accessed by other running
     60  * tasks. These guidelines are loosely enforced by not permitting
     61  * checked exceptions such as {@code IOExceptions} to be
     62  * thrown. However, computations may still encounter unchecked
     63  * exceptions, that are rethrown to callers attempting to join
     64  * them. These exceptions may additionally include {@link
     65  * RejectedExecutionException} stemming from internal resource
     66  * exhaustion, such as failure to allocate internal task
     67  * queues. Rethrown exceptions behave in the same way as regular
     68  * exceptions, but, when possible, contain stack traces (as displayed
     69  * for example using {@code ex.printStackTrace()}) of both the thread
     70  * that initiated the computation as well as the thread actually
     71  * encountering the exception; minimally only the latter.
     72  *
     73  * <p>It is possible to define and use ForkJoinTasks that may block,
     74  * but doing do requires three further considerations: (1) Completion
     75  * of few if any <em>other</em> tasks should be dependent on a task
     76  * that blocks on external synchronization or I/O. Event-style async
     77  * tasks that are never joined (for example, those subclassing {@link
     78  * CountedCompleter}) often fall into this category.  (2) To minimize
     79  * resource impact, tasks should be small; ideally performing only the
     80  * (possibly) blocking action. (3) Unless the {@link
     81  * ForkJoinPool.ManagedBlocker} API is used, or the number of possibly
     82  * blocked tasks is known to be less than the pool's {@link
     83  * ForkJoinPool#getParallelism} level, the pool cannot guarantee that
     84  * enough threads will be available to ensure progress or good
     85  * performance.
     86  *
     87  * <p>The primary method for awaiting completion and extracting
     88  * results of a task is {@link #join}, but there are several variants:
     89  * The {@link Future#get} methods support interruptible and/or timed
     90  * waits for completion and report results using {@code Future}
     91  * conventions. Method {@link #invoke} is semantically
     92  * equivalent to {@code fork(); join()} but always attempts to begin
     93  * execution in the current thread. The "<em>quiet</em>" forms of
     94  * these methods do not extract results or report exceptions. These
     95  * may be useful when a set of tasks are being executed, and you need
     96  * to delay processing of results or exceptions until all complete.
     97  * Method {@code invokeAll} (available in multiple versions)
     98  * performs the most common form of parallel invocation: forking a set
     99  * of tasks and joining them all.
    100  *
    101  * <p>In the most typical usages, a fork-join pair act like a call
    102  * (fork) and return (join) from a parallel recursive function. As is
    103  * the case with other forms of recursive calls, returns (joins)
    104  * should be performed innermost-first. For example, {@code a.fork();
    105  * b.fork(); b.join(); a.join();} is likely to be substantially more
    106  * efficient than joining {@code a} before {@code b}.
    107  *
    108  * <p>The execution status of tasks may be queried at several levels
    109  * of detail: {@link #isDone} is true if a task completed in any way
    110  * (including the case where a task was cancelled without executing);
    111  * {@link #isCompletedNormally} is true if a task completed without
    112  * cancellation or encountering an exception; {@link #isCancelled} is
    113  * true if the task was cancelled (in which case {@link #getException}
    114  * returns a {@link java.util.concurrent.CancellationException}); and
    115  * {@link #isCompletedAbnormally} is true if a task was either
    116  * cancelled or encountered an exception, in which case {@link
    117  * #getException} will return either the encountered exception or
    118  * {@link java.util.concurrent.CancellationException}.
    119  *
    120  * <p>The ForkJoinTask class is not usually directly subclassed.
    121  * Instead, you subclass one of the abstract classes that support a
    122  * particular style of fork/join processing, typically {@link
    123  * RecursiveAction} for most computations that do not return results,
    124  * {@link RecursiveTask} for those that do, and {@link
    125  * CountedCompleter} for those in which completed actions trigger
    126  * other actions.  Normally, a concrete ForkJoinTask subclass declares
    127  * fields comprising its parameters, established in a constructor, and
    128  * then defines a {@code compute} method that somehow uses the control
    129  * methods supplied by this base class.
    130  *
    131  * <p>Method {@link #join} and its variants are appropriate for use
    132  * only when completion dependencies are acyclic; that is, the
    133  * parallel computation can be described as a directed acyclic graph
    134  * (DAG). Otherwise, executions may encounter a form of deadlock as
    135  * tasks cyclically wait for each other.  However, this framework
    136  * supports other methods and techniques (for example the use of
    137  * {@link Phaser}, {@link #helpQuiesce}, and {@link #complete}) that
    138  * may be of use in constructing custom subclasses for problems that
    139  * are not statically structured as DAGs. To support such usages a
    140  * ForkJoinTask may be atomically <em>tagged</em> with a {@code short}
    141  * value using {@link #setForkJoinTaskTag} or {@link
    142  * #compareAndSetForkJoinTaskTag} and checked using {@link
    143  * #getForkJoinTaskTag}. The ForkJoinTask implementation does not use
    144  * these {@code protected} methods or tags for any purpose, but they
    145  * may be of use in the construction of specialized subclasses.  For
    146  * example, parallel graph traversals can use the supplied methods to
    147  * avoid revisiting nodes/tasks that have already been processed.
    148  * (Method names for tagging are bulky in part to encourage definition
    149  * of methods that reflect their usage patterns.)
    150  *
    151  * <p>Most base support methods are {@code final}, to prevent
    152  * overriding of implementations that are intrinsically tied to the
    153  * underlying lightweight task scheduling framework.  Developers
    154  * creating new basic styles of fork/join processing should minimally
    155  * implement {@code protected} methods {@link #exec}, {@link
    156  * #setRawResult}, and {@link #getRawResult}, while also introducing
    157  * an abstract computational method that can be implemented in its
    158  * subclasses, possibly relying on other {@code protected} methods
    159  * provided by this class.
    160  *
    161  * <p>ForkJoinTasks should perform relatively small amounts of
    162  * computation. Large tasks should be split into smaller subtasks,
    163  * usually via recursive decomposition. As a very rough rule of thumb,
    164  * a task should perform more than 100 and less than 10000 basic
    165  * computational steps, and should avoid indefinite looping. If tasks
    166  * are too big, then parallelism cannot improve throughput. If too
    167  * small, then memory and internal task maintenance overhead may
    168  * overwhelm processing.
    169  *
    170  * <p>This class provides {@code adapt} methods for {@link Runnable}
    171  * and {@link Callable}, that may be of use when mixing execution of
    172  * {@code ForkJoinTasks} with other kinds of tasks. When all tasks are
    173  * of this form, consider using a pool constructed in <em>asyncMode</em>.
    174  *
    175  * <p>ForkJoinTasks are {@code Serializable}, which enables them to be
    176  * used in extensions such as remote execution frameworks. It is
    177  * sensible to serialize tasks only before or after, but not during,
    178  * execution. Serialization is not relied on during execution itself.
    179  *
    180  * @since 1.7
    181  * @hide
    182  * @author Doug Lea
    183  */
    184 public abstract class ForkJoinTask<V> implements Future<V>, Serializable {
    185 
    186     /*
    187      * See the internal documentation of class ForkJoinPool for a
    188      * general implementation overview.  ForkJoinTasks are mainly
    189      * responsible for maintaining their "status" field amidst relays
    190      * to methods in ForkJoinWorkerThread and ForkJoinPool.
    191      *
    192      * The methods of this class are more-or-less layered into
    193      * (1) basic status maintenance
    194      * (2) execution and awaiting completion
    195      * (3) user-level methods that additionally report results.
    196      * This is sometimes hard to see because this file orders exported
    197      * methods in a way that flows well in javadocs.
    198      */
    199 
    200     /*
    201      * The status field holds run control status bits packed into a
    202      * single int to minimize footprint and to ensure atomicity (via
    203      * CAS).  Status is initially zero, and takes on nonnegative
    204      * values until completed, upon which status (anded with
    205      * DONE_MASK) holds value NORMAL, CANCELLED, or EXCEPTIONAL. Tasks
    206      * undergoing blocking waits by other threads have the SIGNAL bit
    207      * set.  Completion of a stolen task with SIGNAL set awakens any
    208      * waiters via notifyAll. Even though suboptimal for some
    209      * purposes, we use basic builtin wait/notify to take advantage of
    210      * "monitor inflation" in JVMs that we would otherwise need to
    211      * emulate to avoid adding further per-task bookkeeping overhead.
    212      * We want these monitors to be "fat", i.e., not use biasing or
    213      * thin-lock techniques, so use some odd coding idioms that tend
    214      * to avoid them, mainly by arranging that every synchronized
    215      * block performs a wait, notifyAll or both.
    216      *
    217      * These control bits occupy only (some of) the upper half (16
    218      * bits) of status field. The lower bits are used for user-defined
    219      * tags.
    220      */
    221 
    222     /** The run status of this task */
    223     volatile int status; // accessed directly by pool and workers
    224     static final int DONE_MASK   = 0xf0000000;  // mask out non-completion bits
    225     static final int NORMAL      = 0xf0000000;  // must be negative
    226     static final int CANCELLED   = 0xc0000000;  // must be < NORMAL
    227     static final int EXCEPTIONAL = 0x80000000;  // must be < CANCELLED
    228     static final int SIGNAL      = 0x00010000;  // must be >= 1 << 16
    229     static final int SMASK       = 0x0000ffff;  // short bits for tags
    230 
    231     /**
    232      * Marks completion and wakes up threads waiting to join this
    233      * task.
    234      *
    235      * @param completion one of NORMAL, CANCELLED, EXCEPTIONAL
    236      * @return completion status on exit
    237      */
    238     private int setCompletion(int completion) {
    239         for (int s;;) {
    240             if ((s = status) < 0)
    241                 return s;
    242             if (U.compareAndSwapInt(this, STATUS, s, s | completion)) {
    243                 if ((s >>> 16) != 0)
    244                     synchronized (this) { notifyAll(); }
    245                 return completion;
    246             }
    247         }
    248     }
    249 
    250     /**
    251      * Primary execution method for stolen tasks. Unless done, calls
    252      * exec and records status if completed, but doesn't wait for
    253      * completion otherwise.
    254      *
    255      * @return status on exit from this method
    256      */
    257     final int doExec() {
    258         int s; boolean completed;
    259         if ((s = status) >= 0) {
    260             try {
    261                 completed = exec();
    262             } catch (Throwable rex) {
    263                 return setExceptionalCompletion(rex);
    264             }
    265             if (completed)
    266                 s = setCompletion(NORMAL);
    267         }
    268         return s;
    269     }
    270 
    271     /**
    272      * Tries to set SIGNAL status unless already completed. Used by
    273      * ForkJoinPool. Other variants are directly incorporated into
    274      * externalAwaitDone etc.
    275      *
    276      * @return true if successful
    277      */
    278     final boolean trySetSignal() {
    279         int s = status;
    280         return s >= 0 && U.compareAndSwapInt(this, STATUS, s, s | SIGNAL);
    281     }
    282 
    283     /**
    284      * Blocks a non-worker-thread until completion.
    285      * @return status upon completion
    286      */
    287     private int externalAwaitDone() {
    288         int s;
    289         ForkJoinPool.externalHelpJoin(this);
    290         boolean interrupted = false;
    291         while ((s = status) >= 0) {
    292             if (U.compareAndSwapInt(this, STATUS, s, s | SIGNAL)) {
    293                 synchronized (this) {
    294                     if (status >= 0) {
    295                         try {
    296                             wait();
    297                         } catch (InterruptedException ie) {
    298                             interrupted = true;
    299                         }
    300                     }
    301                     else
    302                         notifyAll();
    303                 }
    304             }
    305         }
    306         if (interrupted)
    307             Thread.currentThread().interrupt();
    308         return s;
    309     }
    310 
    311     /**
    312      * Blocks a non-worker-thread until completion or interruption.
    313      */
    314     private int externalInterruptibleAwaitDone() throws InterruptedException {
    315         int s;
    316         if (Thread.interrupted())
    317             throw new InterruptedException();
    318         ForkJoinPool.externalHelpJoin(this);
    319         while ((s = status) >= 0) {
    320             if (U.compareAndSwapInt(this, STATUS, s, s | SIGNAL)) {
    321                 synchronized (this) {
    322                     if (status >= 0)
    323                         wait();
    324                     else
    325                         notifyAll();
    326                 }
    327             }
    328         }
    329         return s;
    330     }
    331 
    332 
    333     /**
    334      * Implementation for join, get, quietlyJoin. Directly handles
    335      * only cases of already-completed, external wait, and
    336      * unfork+exec.  Others are relayed to ForkJoinPool.awaitJoin.
    337      *
    338      * @return status upon completion
    339      */
    340     private int doJoin() {
    341         int s; Thread t; ForkJoinWorkerThread wt; ForkJoinPool.WorkQueue w;
    342         return (s = status) < 0 ? s :
    343             ((t = Thread.currentThread()) instanceof ForkJoinWorkerThread) ?
    344             (w = (wt = (ForkJoinWorkerThread)t).workQueue).
    345             tryUnpush(this) && (s = doExec()) < 0 ? s :
    346             wt.pool.awaitJoin(w, this) :
    347             externalAwaitDone();
    348     }
    349 
    350     /**
    351      * Implementation for invoke, quietlyInvoke.
    352      *
    353      * @return status upon completion
    354      */
    355     private int doInvoke() {
    356         int s; Thread t; ForkJoinWorkerThread wt;
    357         return (s = doExec()) < 0 ? s :
    358             ((t = Thread.currentThread()) instanceof ForkJoinWorkerThread) ?
    359             (wt = (ForkJoinWorkerThread)t).pool.awaitJoin(wt.workQueue, this) :
    360             externalAwaitDone();
    361     }
    362 
    363     // Exception table support
    364 
    365     /**
    366      * Table of exceptions thrown by tasks, to enable reporting by
    367      * callers. Because exceptions are rare, we don't directly keep
    368      * them with task objects, but instead use a weak ref table.  Note
    369      * that cancellation exceptions don't appear in the table, but are
    370      * instead recorded as status values.
    371      *
    372      * Note: These statics are initialized below in static block.
    373      */
    374     private static final ExceptionNode[] exceptionTable;
    375     private static final ReentrantLock exceptionTableLock;
    376     private static final ReferenceQueue<Object> exceptionTableRefQueue;
    377 
    378     /**
    379      * Fixed capacity for exceptionTable.
    380      */
    381     private static final int EXCEPTION_MAP_CAPACITY = 32;
    382 
    383     /**
    384      * Key-value nodes for exception table.  The chained hash table
    385      * uses identity comparisons, full locking, and weak references
    386      * for keys. The table has a fixed capacity because it only
    387      * maintains task exceptions long enough for joiners to access
    388      * them, so should never become very large for sustained
    389      * periods. However, since we do not know when the last joiner
    390      * completes, we must use weak references and expunge them. We do
    391      * so on each operation (hence full locking). Also, some thread in
    392      * any ForkJoinPool will call helpExpungeStaleExceptions when its
    393      * pool becomes isQuiescent.
    394      */
    395     static final class ExceptionNode extends WeakReference<ForkJoinTask<?>> {
    396         final Throwable ex;
    397         ExceptionNode next;
    398         final long thrower;  // use id not ref to avoid weak cycles
    399         ExceptionNode(ForkJoinTask<?> task, Throwable ex, ExceptionNode next) {
    400             super(task, exceptionTableRefQueue);
    401             this.ex = ex;
    402             this.next = next;
    403             this.thrower = Thread.currentThread().getId();
    404         }
    405     }
    406 
    407     /**
    408      * Records exception and sets status.
    409      *
    410      * @return status on exit
    411      */
    412     final int recordExceptionalCompletion(Throwable ex) {
    413         int s;
    414         if ((s = status) >= 0) {
    415             int h = System.identityHashCode(this);
    416             final ReentrantLock lock = exceptionTableLock;
    417             lock.lock();
    418             try {
    419                 expungeStaleExceptions();
    420                 ExceptionNode[] t = exceptionTable;
    421                 int i = h & (t.length - 1);
    422                 for (ExceptionNode e = t[i]; ; e = e.next) {
    423                     if (e == null) {
    424                         t[i] = new ExceptionNode(this, ex, t[i]);
    425                         break;
    426                     }
    427                     if (e.get() == this) // already present
    428                         break;
    429                 }
    430             } finally {
    431                 lock.unlock();
    432             }
    433             s = setCompletion(EXCEPTIONAL);
    434         }
    435         return s;
    436     }
    437 
    438     /**
    439      * Records exception and possibly propagates.
    440      *
    441      * @return status on exit
    442      */
    443     private int setExceptionalCompletion(Throwable ex) {
    444         int s = recordExceptionalCompletion(ex);
    445         if ((s & DONE_MASK) == EXCEPTIONAL)
    446             internalPropagateException(ex);
    447         return s;
    448     }
    449 
    450     /**
    451      * Hook for exception propagation support for tasks with completers.
    452      */
    453     void internalPropagateException(Throwable ex) {
    454     }
    455 
    456     /**
    457      * Cancels, ignoring any exceptions thrown by cancel. Used during
    458      * worker and pool shutdown. Cancel is spec'ed not to throw any
    459      * exceptions, but if it does anyway, we have no recourse during
    460      * shutdown, so guard against this case.
    461      */
    462     static final void cancelIgnoringExceptions(ForkJoinTask<?> t) {
    463         if (t != null && t.status >= 0) {
    464             try {
    465                 t.cancel(false);
    466             } catch (Throwable ignore) {
    467             }
    468         }
    469     }
    470 
    471     /**
    472      * Removes exception node and clears status.
    473      */
    474     private void clearExceptionalCompletion() {
    475         int h = System.identityHashCode(this);
    476         final ReentrantLock lock = exceptionTableLock;
    477         lock.lock();
    478         try {
    479             ExceptionNode[] t = exceptionTable;
    480             int i = h & (t.length - 1);
    481             ExceptionNode e = t[i];
    482             ExceptionNode pred = null;
    483             while (e != null) {
    484                 ExceptionNode next = e.next;
    485                 if (e.get() == this) {
    486                     if (pred == null)
    487                         t[i] = next;
    488                     else
    489                         pred.next = next;
    490                     break;
    491                 }
    492                 pred = e;
    493                 e = next;
    494             }
    495             expungeStaleExceptions();
    496             status = 0;
    497         } finally {
    498             lock.unlock();
    499         }
    500     }
    501 
    502     /**
    503      * Returns a rethrowable exception for the given task, if
    504      * available. To provide accurate stack traces, if the exception
    505      * was not thrown by the current thread, we try to create a new
    506      * exception of the same type as the one thrown, but with the
    507      * recorded exception as its cause. If there is no such
    508      * constructor, we instead try to use a no-arg constructor,
    509      * followed by initCause, to the same effect. If none of these
    510      * apply, or any fail due to other exceptions, we return the
    511      * recorded exception, which is still correct, although it may
    512      * contain a misleading stack trace.
    513      *
    514      * @return the exception, or null if none
    515      */
    516     private Throwable getThrowableException() {
    517         if ((status & DONE_MASK) != EXCEPTIONAL)
    518             return null;
    519         int h = System.identityHashCode(this);
    520         ExceptionNode e;
    521         final ReentrantLock lock = exceptionTableLock;
    522         lock.lock();
    523         try {
    524             expungeStaleExceptions();
    525             ExceptionNode[] t = exceptionTable;
    526             e = t[h & (t.length - 1)];
    527             while (e != null && e.get() != this)
    528                 e = e.next;
    529         } finally {
    530             lock.unlock();
    531         }
    532         Throwable ex;
    533         if (e == null || (ex = e.ex) == null)
    534             return null;
    535         if (false && e.thrower != Thread.currentThread().getId()) {
    536             Class<? extends Throwable> ec = ex.getClass();
    537             try {
    538                 Constructor<?> noArgCtor = null;
    539                 Constructor<?>[] cs = ec.getConstructors();// public ctors only
    540                 for (int i = 0; i < cs.length; ++i) {
    541                     Constructor<?> c = cs[i];
    542                     Class<?>[] ps = c.getParameterTypes();
    543                     if (ps.length == 0)
    544                         noArgCtor = c;
    545                     else if (ps.length == 1 && ps[0] == Throwable.class)
    546                         return (Throwable)(c.newInstance(ex));
    547                 }
    548                 if (noArgCtor != null) {
    549                     Throwable wx = (Throwable)(noArgCtor.newInstance());
    550                     wx.initCause(ex);
    551                     return wx;
    552                 }
    553             } catch (Exception ignore) {
    554             }
    555         }
    556         return ex;
    557     }
    558 
    559     /**
    560      * Poll stale refs and remove them. Call only while holding lock.
    561      */
    562     private static void expungeStaleExceptions() {
    563         for (Object x; (x = exceptionTableRefQueue.poll()) != null;) {
    564             if (x instanceof ExceptionNode) {
    565                 ForkJoinTask<?> key = ((ExceptionNode)x).get();
    566                 ExceptionNode[] t = exceptionTable;
    567                 int i = System.identityHashCode(key) & (t.length - 1);
    568                 ExceptionNode e = t[i];
    569                 ExceptionNode pred = null;
    570                 while (e != null) {
    571                     ExceptionNode next = e.next;
    572                     if (e == x) {
    573                         if (pred == null)
    574                             t[i] = next;
    575                         else
    576                             pred.next = next;
    577                         break;
    578                     }
    579                     pred = e;
    580                     e = next;
    581                 }
    582             }
    583         }
    584     }
    585 
    586     /**
    587      * If lock is available, poll stale refs and remove them.
    588      * Called from ForkJoinPool when pools become quiescent.
    589      */
    590     static final void helpExpungeStaleExceptions() {
    591         final ReentrantLock lock = exceptionTableLock;
    592         if (lock.tryLock()) {
    593             try {
    594                 expungeStaleExceptions();
    595             } finally {
    596                 lock.unlock();
    597             }
    598         }
    599     }
    600 
    601     /**
    602      * A version of "sneaky throw" to relay exceptions
    603      */
    604     static void rethrow(final Throwable ex) {
    605         if (ex != null) {
    606             if (ex instanceof Error)
    607                 throw (Error)ex;
    608             if (ex instanceof RuntimeException)
    609                 throw (RuntimeException)ex;
    610             throw uncheckedThrowable(ex, RuntimeException.class);
    611         }
    612     }
    613 
    614     /**
    615      * The sneaky part of sneaky throw, relying on generics
    616      * limitations to evade compiler complaints about rethrowing
    617      * unchecked exceptions
    618      */
    619     @SuppressWarnings("unchecked") static <T extends Throwable>
    620         T uncheckedThrowable(final Throwable t, final Class<T> c) {
    621         return (T)t; // rely on vacuous cast
    622     }
    623 
    624     /**
    625      * Throws exception, if any, associated with the given status.
    626      */
    627     private void reportException(int s) {
    628         if (s == CANCELLED)
    629             throw new CancellationException();
    630         if (s == EXCEPTIONAL)
    631             rethrow(getThrowableException());
    632     }
    633 
    634     // public methods
    635 
    636     /**
    637      * Arranges to asynchronously execute this task in the pool the
    638      * current task is running in, if applicable, or using the {@link
    639      * ForkJoinPool#commonPool()} if not {@link #inForkJoinPool}.  While
    640      * it is not necessarily enforced, it is a usage error to fork a
    641      * task more than once unless it has completed and been
    642      * reinitialized.  Subsequent modifications to the state of this
    643      * task or any data it operates on are not necessarily
    644      * consistently observable by any thread other than the one
    645      * executing it unless preceded by a call to {@link #join} or
    646      * related methods, or a call to {@link #isDone} returning {@code
    647      * true}.
    648      *
    649      * @return {@code this}, to simplify usage
    650      */
    651     public final ForkJoinTask<V> fork() {
    652         Thread t;
    653         if ((t = Thread.currentThread()) instanceof ForkJoinWorkerThread)
    654             ((ForkJoinWorkerThread)t).workQueue.push(this);
    655         else
    656             ForkJoinPool.commonPool.externalPush(this);
    657         return this;
    658     }
    659 
    660     /**
    661      * Returns the result of the computation when it {@link #isDone is
    662      * done}.  This method differs from {@link #get()} in that
    663      * abnormal completion results in {@code RuntimeException} or
    664      * {@code Error}, not {@code ExecutionException}, and that
    665      * interrupts of the calling thread do <em>not</em> cause the
    666      * method to abruptly return by throwing {@code
    667      * InterruptedException}.
    668      *
    669      * @return the computed result
    670      */
    671     public final V join() {
    672         int s;
    673         if ((s = doJoin() & DONE_MASK) != NORMAL)
    674             reportException(s);
    675         return getRawResult();
    676     }
    677 
    678     /**
    679      * Commences performing this task, awaits its completion if
    680      * necessary, and returns its result, or throws an (unchecked)
    681      * {@code RuntimeException} or {@code Error} if the underlying
    682      * computation did so.
    683      *
    684      * @return the computed result
    685      */
    686     public final V invoke() {
    687         int s;
    688         if ((s = doInvoke() & DONE_MASK) != NORMAL)
    689             reportException(s);
    690         return getRawResult();
    691     }
    692 
    693     /**
    694      * Forks the given tasks, returning when {@code isDone} holds for
    695      * each task or an (unchecked) exception is encountered, in which
    696      * case the exception is rethrown. If more than one task
    697      * encounters an exception, then this method throws any one of
    698      * these exceptions. If any task encounters an exception, the
    699      * other may be cancelled. However, the execution status of
    700      * individual tasks is not guaranteed upon exceptional return. The
    701      * status of each task may be obtained using {@link
    702      * #getException()} and related methods to check if they have been
    703      * cancelled, completed normally or exceptionally, or left
    704      * unprocessed.
    705      *
    706      * @param t1 the first task
    707      * @param t2 the second task
    708      * @throws NullPointerException if any task is null
    709      */
    710     public static void invokeAll(ForkJoinTask<?> t1, ForkJoinTask<?> t2) {
    711         int s1, s2;
    712         t2.fork();
    713         if ((s1 = t1.doInvoke() & DONE_MASK) != NORMAL)
    714             t1.reportException(s1);
    715         if ((s2 = t2.doJoin() & DONE_MASK) != NORMAL)
    716             t2.reportException(s2);
    717     }
    718 
    719     /**
    720      * Forks the given tasks, returning when {@code isDone} holds for
    721      * each task or an (unchecked) exception is encountered, in which
    722      * case the exception is rethrown. If more than one task
    723      * encounters an exception, then this method throws any one of
    724      * these exceptions. If any task encounters an exception, others
    725      * may be cancelled. However, the execution status of individual
    726      * tasks is not guaranteed upon exceptional return. The status of
    727      * each task may be obtained using {@link #getException()} and
    728      * related methods to check if they have been cancelled, completed
    729      * normally or exceptionally, or left unprocessed.
    730      *
    731      * @param tasks the tasks
    732      * @throws NullPointerException if any task is null
    733      */
    734     public static void invokeAll(ForkJoinTask<?>... tasks) {
    735         Throwable ex = null;
    736         int last = tasks.length - 1;
    737         for (int i = last; i >= 0; --i) {
    738             ForkJoinTask<?> t = tasks[i];
    739             if (t == null) {
    740                 if (ex == null)
    741                     ex = new NullPointerException();
    742             }
    743             else if (i != 0)
    744                 t.fork();
    745             else if (t.doInvoke() < NORMAL && ex == null)
    746                 ex = t.getException();
    747         }
    748         for (int i = 1; i <= last; ++i) {
    749             ForkJoinTask<?> t = tasks[i];
    750             if (t != null) {
    751                 if (ex != null)
    752                     t.cancel(false);
    753                 else if (t.doJoin() < NORMAL)
    754                     ex = t.getException();
    755             }
    756         }
    757         if (ex != null)
    758             rethrow(ex);
    759     }
    760 
    761     /**
    762      * Forks all tasks in the specified collection, returning when
    763      * {@code isDone} holds for each task or an (unchecked) exception
    764      * is encountered, in which case the exception is rethrown. If
    765      * more than one task encounters an exception, then this method
    766      * throws any one of these exceptions. If any task encounters an
    767      * exception, others may be cancelled. However, the execution
    768      * status of individual tasks is not guaranteed upon exceptional
    769      * return. The status of each task may be obtained using {@link
    770      * #getException()} and related methods to check if they have been
    771      * cancelled, completed normally or exceptionally, or left
    772      * unprocessed.
    773      *
    774      * @param tasks the collection of tasks
    775      * @return the tasks argument, to simplify usage
    776      * @throws NullPointerException if tasks or any element are null
    777 
    778      * @hide
    779      */
    780     public static <T extends ForkJoinTask<?>> Collection<T> invokeAll(Collection<T> tasks) {
    781         if (!(tasks instanceof RandomAccess) || !(tasks instanceof List<?>)) {
    782             invokeAll(tasks.toArray(new ForkJoinTask<?>[tasks.size()]));
    783             return tasks;
    784         }
    785         @SuppressWarnings("unchecked")
    786         List<? extends ForkJoinTask<?>> ts =
    787             (List<? extends ForkJoinTask<?>>) tasks;
    788         Throwable ex = null;
    789         int last = ts.size() - 1;
    790         for (int i = last; i >= 0; --i) {
    791             ForkJoinTask<?> t = ts.get(i);
    792             if (t == null) {
    793                 if (ex == null)
    794                     ex = new NullPointerException();
    795             }
    796             else if (i != 0)
    797                 t.fork();
    798             else if (t.doInvoke() < NORMAL && ex == null)
    799                 ex = t.getException();
    800         }
    801         for (int i = 1; i <= last; ++i) {
    802             ForkJoinTask<?> t = ts.get(i);
    803             if (t != null) {
    804                 if (ex != null)
    805                     t.cancel(false);
    806                 else if (t.doJoin() < NORMAL)
    807                     ex = t.getException();
    808             }
    809         }
    810         if (ex != null)
    811             rethrow(ex);
    812         return tasks;
    813     }
    814 
    815     /**
    816      * Attempts to cancel execution of this task. This attempt will
    817      * fail if the task has already completed or could not be
    818      * cancelled for some other reason. If successful, and this task
    819      * has not started when {@code cancel} is called, execution of
    820      * this task is suppressed. After this method returns
    821      * successfully, unless there is an intervening call to {@link
    822      * #reinitialize}, subsequent calls to {@link #isCancelled},
    823      * {@link #isDone}, and {@code cancel} will return {@code true}
    824      * and calls to {@link #join} and related methods will result in
    825      * {@code CancellationException}.
    826      *
    827      * <p>This method may be overridden in subclasses, but if so, must
    828      * still ensure that these properties hold. In particular, the
    829      * {@code cancel} method itself must not throw exceptions.
    830      *
    831      * <p>This method is designed to be invoked by <em>other</em>
    832      * tasks. To terminate the current task, you can just return or
    833      * throw an unchecked exception from its computation method, or
    834      * invoke {@link #completeExceptionally}.
    835      *
    836      * @param mayInterruptIfRunning this value has no effect in the
    837      * default implementation because interrupts are not used to
    838      * control cancellation.
    839      *
    840      * @return {@code true} if this task is now cancelled
    841      */
    842     public boolean cancel(boolean mayInterruptIfRunning) {
    843         return (setCompletion(CANCELLED) & DONE_MASK) == CANCELLED;
    844     }
    845 
    846     public final boolean isDone() {
    847         return status < 0;
    848     }
    849 
    850     public final boolean isCancelled() {
    851         return (status & DONE_MASK) == CANCELLED;
    852     }
    853 
    854     /**
    855      * Returns {@code true} if this task threw an exception or was cancelled.
    856      *
    857      * @return {@code true} if this task threw an exception or was cancelled
    858      */
    859     public final boolean isCompletedAbnormally() {
    860         return status < NORMAL;
    861     }
    862 
    863     /**
    864      * Returns {@code true} if this task completed without throwing an
    865      * exception and was not cancelled.
    866      *
    867      * @return {@code true} if this task completed without throwing an
    868      * exception and was not cancelled
    869      */
    870     public final boolean isCompletedNormally() {
    871         return (status & DONE_MASK) == NORMAL;
    872     }
    873 
    874     /**
    875      * Returns the exception thrown by the base computation, or a
    876      * {@code CancellationException} if cancelled, or {@code null} if
    877      * none or if the method has not yet completed.
    878      *
    879      * @return the exception, or {@code null} if none
    880      */
    881     public final Throwable getException() {
    882         int s = status & DONE_MASK;
    883         return ((s >= NORMAL)    ? null :
    884                 (s == CANCELLED) ? new CancellationException() :
    885                 getThrowableException());
    886     }
    887 
    888     /**
    889      * Completes this task abnormally, and if not already aborted or
    890      * cancelled, causes it to throw the given exception upon
    891      * {@code join} and related operations. This method may be used
    892      * to induce exceptions in asynchronous tasks, or to force
    893      * completion of tasks that would not otherwise complete.  Its use
    894      * in other situations is discouraged.  This method is
    895      * overridable, but overridden versions must invoke {@code super}
    896      * implementation to maintain guarantees.
    897      *
    898      * @param ex the exception to throw. If this exception is not a
    899      * {@code RuntimeException} or {@code Error}, the actual exception
    900      * thrown will be a {@code RuntimeException} with cause {@code ex}.
    901      */
    902     public void completeExceptionally(Throwable ex) {
    903         setExceptionalCompletion((ex instanceof RuntimeException) ||
    904                                  (ex instanceof Error) ? ex :
    905                                  new RuntimeException(ex));
    906     }
    907 
    908     /**
    909      * Completes this task, and if not already aborted or cancelled,
    910      * returning the given value as the result of subsequent
    911      * invocations of {@code join} and related operations. This method
    912      * may be used to provide results for asynchronous tasks, or to
    913      * provide alternative handling for tasks that would not otherwise
    914      * complete normally. Its use in other situations is
    915      * discouraged. This method is overridable, but overridden
    916      * versions must invoke {@code super} implementation to maintain
    917      * guarantees.
    918      *
    919      * @param value the result value for this task
    920      */
    921     public void complete(V value) {
    922         try {
    923             setRawResult(value);
    924         } catch (Throwable rex) {
    925             setExceptionalCompletion(rex);
    926             return;
    927         }
    928         setCompletion(NORMAL);
    929     }
    930 
    931     /**
    932      * Completes this task normally without setting a value. The most
    933      * recent value established by {@link #setRawResult} (or {@code
    934      * null} by default) will be returned as the result of subsequent
    935      * invocations of {@code join} and related operations.
    936      *
    937      * @since 1.8
    938      * @hide
    939      */
    940     public final void quietlyComplete() {
    941         setCompletion(NORMAL);
    942     }
    943 
    944     /**
    945      * Waits if necessary for the computation to complete, and then
    946      * retrieves its result.
    947      *
    948      * @return the computed result
    949      * @throws CancellationException if the computation was cancelled
    950      * @throws ExecutionException if the computation threw an
    951      * exception
    952      * @throws InterruptedException if the current thread is not a
    953      * member of a ForkJoinPool and was interrupted while waiting
    954      */
    955     public final V get() throws InterruptedException, ExecutionException {
    956         int s = (Thread.currentThread() instanceof ForkJoinWorkerThread) ?
    957             doJoin() : externalInterruptibleAwaitDone();
    958         Throwable ex;
    959         if ((s &= DONE_MASK) == CANCELLED)
    960             throw new CancellationException();
    961         if (s == EXCEPTIONAL && (ex = getThrowableException()) != null)
    962             throw new ExecutionException(ex);
    963         return getRawResult();
    964     }
    965 
    966     /**
    967      * Waits if necessary for at most the given time for the computation
    968      * to complete, and then retrieves its result, if available.
    969      *
    970      * @param timeout the maximum time to wait
    971      * @param unit the time unit of the timeout argument
    972      * @return the computed result
    973      * @throws CancellationException if the computation was cancelled
    974      * @throws ExecutionException if the computation threw an
    975      * exception
    976      * @throws InterruptedException if the current thread is not a
    977      * member of a ForkJoinPool and was interrupted while waiting
    978      * @throws TimeoutException if the wait timed out
    979      */
    980     public final V get(long timeout, TimeUnit unit)
    981         throws InterruptedException, ExecutionException, TimeoutException {
    982         if (Thread.interrupted())
    983             throw new InterruptedException();
    984         // Messy in part because we measure in nanosecs, but wait in millisecs
    985         int s; long ms;
    986         long ns = unit.toNanos(timeout);
    987         if ((s = status) >= 0 && ns > 0L) {
    988             long deadline = System.nanoTime() + ns;
    989             ForkJoinPool p = null;
    990             ForkJoinPool.WorkQueue w = null;
    991             Thread t = Thread.currentThread();
    992             if (t instanceof ForkJoinWorkerThread) {
    993                 ForkJoinWorkerThread wt = (ForkJoinWorkerThread)t;
    994                 p = wt.pool;
    995                 w = wt.workQueue;
    996                 p.helpJoinOnce(w, this); // no retries on failure
    997             }
    998             else
    999                 ForkJoinPool.externalHelpJoin(this);
   1000             boolean canBlock = false;
   1001             boolean interrupted = false;
   1002             try {
   1003                 while ((s = status) >= 0) {
   1004                     if (w != null && w.qlock < 0)
   1005                         cancelIgnoringExceptions(this);
   1006                     else if (!canBlock) {
   1007                         if (p == null || p.tryCompensate())
   1008                             canBlock = true;
   1009                     }
   1010                     else {
   1011                         if ((ms = TimeUnit.NANOSECONDS.toMillis(ns)) > 0L &&
   1012                             U.compareAndSwapInt(this, STATUS, s, s | SIGNAL)) {
   1013                             synchronized (this) {
   1014                                 if (status >= 0) {
   1015                                     try {
   1016                                         wait(ms);
   1017                                     } catch (InterruptedException ie) {
   1018                                         if (p == null)
   1019                                             interrupted = true;
   1020                                     }
   1021                                 }
   1022                                 else
   1023                                     notifyAll();
   1024                             }
   1025                         }
   1026                         if ((s = status) < 0 || interrupted ||
   1027                             (ns = deadline - System.nanoTime()) <= 0L)
   1028                             break;
   1029                     }
   1030                 }
   1031             } finally {
   1032                 if (p != null && canBlock)
   1033                     p.incrementActiveCount();
   1034             }
   1035             if (interrupted)
   1036                 throw new InterruptedException();
   1037         }
   1038         if ((s &= DONE_MASK) != NORMAL) {
   1039             Throwable ex;
   1040             if (s == CANCELLED)
   1041                 throw new CancellationException();
   1042             if (s != EXCEPTIONAL)
   1043                 throw new TimeoutException();
   1044             if ((ex = getThrowableException()) != null)
   1045                 throw new ExecutionException(ex);
   1046         }
   1047         return getRawResult();
   1048     }
   1049 
   1050     /**
   1051      * Joins this task, without returning its result or throwing its
   1052      * exception. This method may be useful when processing
   1053      * collections of tasks when some have been cancelled or otherwise
   1054      * known to have aborted.
   1055      */
   1056     public final void quietlyJoin() {
   1057         doJoin();
   1058     }
   1059 
   1060     /**
   1061      * Commences performing this task and awaits its completion if
   1062      * necessary, without returning its result or throwing its
   1063      * exception.
   1064      */
   1065     public final void quietlyInvoke() {
   1066         doInvoke();
   1067     }
   1068 
   1069     /**
   1070      * Possibly executes tasks until the pool hosting the current task
   1071      * {@link ForkJoinPool#isQuiescent is quiescent}. This method may
   1072      * be of use in designs in which many tasks are forked, but none
   1073      * are explicitly joined, instead executing them until all are
   1074      * processed.
   1075      */
   1076     public static void helpQuiesce() {
   1077         Thread t;
   1078         if ((t = Thread.currentThread()) instanceof ForkJoinWorkerThread) {
   1079             ForkJoinWorkerThread wt = (ForkJoinWorkerThread)t;
   1080             wt.pool.helpQuiescePool(wt.workQueue);
   1081         }
   1082         else
   1083             ForkJoinPool.externalHelpQuiescePool();
   1084     }
   1085 
   1086     /**
   1087      * Resets the internal bookkeeping state of this task, allowing a
   1088      * subsequent {@code fork}. This method allows repeated reuse of
   1089      * this task, but only if reuse occurs when this task has either
   1090      * never been forked, or has been forked, then completed and all
   1091      * outstanding joins of this task have also completed. Effects
   1092      * under any other usage conditions are not guaranteed.
   1093      * This method may be useful when executing
   1094      * pre-constructed trees of subtasks in loops.
   1095      *
   1096      * <p>Upon completion of this method, {@code isDone()} reports
   1097      * {@code false}, and {@code getException()} reports {@code
   1098      * null}. However, the value returned by {@code getRawResult} is
   1099      * unaffected. To clear this value, you can invoke {@code
   1100      * setRawResult(null)}.
   1101      */
   1102     public void reinitialize() {
   1103         if ((status & DONE_MASK) == EXCEPTIONAL)
   1104             clearExceptionalCompletion();
   1105         else
   1106             status = 0;
   1107     }
   1108 
   1109     /**
   1110      * Returns the pool hosting the current task execution, or null
   1111      * if this task is executing outside of any ForkJoinPool.
   1112      *
   1113      * @see #inForkJoinPool
   1114      * @return the pool, or {@code null} if none
   1115      */
   1116     public static ForkJoinPool getPool() {
   1117         Thread t = Thread.currentThread();
   1118         return (t instanceof ForkJoinWorkerThread) ?
   1119             ((ForkJoinWorkerThread) t).pool : null;
   1120     }
   1121 
   1122     /**
   1123      * Returns {@code true} if the current thread is a {@link
   1124      * ForkJoinWorkerThread} executing as a ForkJoinPool computation.
   1125      *
   1126      * @return {@code true} if the current thread is a {@link
   1127      * ForkJoinWorkerThread} executing as a ForkJoinPool computation,
   1128      * or {@code false} otherwise
   1129      */
   1130     public static boolean inForkJoinPool() {
   1131         return Thread.currentThread() instanceof ForkJoinWorkerThread;
   1132     }
   1133 
   1134     /**
   1135      * Tries to unschedule this task for execution. This method will
   1136      * typically (but is not guaranteed to) succeed if this task is
   1137      * the most recently forked task by the current thread, and has
   1138      * not commenced executing in another thread.  This method may be
   1139      * useful when arranging alternative local processing of tasks
   1140      * that could have been, but were not, stolen.
   1141      *
   1142      * @return {@code true} if unforked
   1143      */
   1144     public boolean tryUnfork() {
   1145         Thread t;
   1146         return (((t = Thread.currentThread()) instanceof ForkJoinWorkerThread) ?
   1147                 ((ForkJoinWorkerThread)t).workQueue.tryUnpush(this) :
   1148                 ForkJoinPool.tryExternalUnpush(this));
   1149     }
   1150 
   1151     /**
   1152      * Returns an estimate of the number of tasks that have been
   1153      * forked by the current worker thread but not yet executed. This
   1154      * value may be useful for heuristic decisions about whether to
   1155      * fork other tasks.
   1156      *
   1157      * @return the number of tasks
   1158      */
   1159     public static int getQueuedTaskCount() {
   1160         Thread t; ForkJoinPool.WorkQueue q;
   1161         if ((t = Thread.currentThread()) instanceof ForkJoinWorkerThread)
   1162             q = ((ForkJoinWorkerThread)t).workQueue;
   1163         else
   1164             q = ForkJoinPool.commonSubmitterQueue();
   1165         return (q == null) ? 0 : q.queueSize();
   1166     }
   1167 
   1168     /**
   1169      * Returns an estimate of how many more locally queued tasks are
   1170      * held by the current worker thread than there are other worker
   1171      * threads that might steal them, or zero if this thread is not
   1172      * operating in a ForkJoinPool. This value may be useful for
   1173      * heuristic decisions about whether to fork other tasks. In many
   1174      * usages of ForkJoinTasks, at steady state, each worker should
   1175      * aim to maintain a small constant surplus (for example, 3) of
   1176      * tasks, and to process computations locally if this threshold is
   1177      * exceeded.
   1178      *
   1179      * @return the surplus number of tasks, which may be negative
   1180      */
   1181     public static int getSurplusQueuedTaskCount() {
   1182         return ForkJoinPool.getSurplusQueuedTaskCount();
   1183     }
   1184 
   1185     // Extension methods
   1186 
   1187     /**
   1188      * Returns the result that would be returned by {@link #join}, even
   1189      * if this task completed abnormally, or {@code null} if this task
   1190      * is not known to have been completed.  This method is designed
   1191      * to aid debugging, as well as to support extensions. Its use in
   1192      * any other context is discouraged.
   1193      *
   1194      * @return the result, or {@code null} if not completed
   1195      */
   1196     public abstract V getRawResult();
   1197 
   1198     /**
   1199      * Forces the given value to be returned as a result.  This method
   1200      * is designed to support extensions, and should not in general be
   1201      * called otherwise.
   1202      *
   1203      * @param value the value
   1204      */
   1205     protected abstract void setRawResult(V value);
   1206 
   1207     /**
   1208      * Immediately performs the base action of this task and returns
   1209      * true if, upon return from this method, this task is guaranteed
   1210      * to have completed normally. This method may return false
   1211      * otherwise, to indicate that this task is not necessarily
   1212      * complete (or is not known to be complete), for example in
   1213      * asynchronous actions that require explicit invocations of
   1214      * completion methods. This method may also throw an (unchecked)
   1215      * exception to indicate abnormal exit. This method is designed to
   1216      * support extensions, and should not in general be called
   1217      * otherwise.
   1218      *
   1219      * @return {@code true} if this task is known to have completed normally
   1220      */
   1221     protected abstract boolean exec();
   1222 
   1223     /**
   1224      * Returns, but does not unschedule or execute, a task queued by
   1225      * the current thread but not yet executed, if one is immediately
   1226      * available. There is no guarantee that this task will actually
   1227      * be polled or executed next. Conversely, this method may return
   1228      * null even if a task exists but cannot be accessed without
   1229      * contention with other threads.  This method is designed
   1230      * primarily to support extensions, and is unlikely to be useful
   1231      * otherwise.
   1232      *
   1233      * @return the next task, or {@code null} if none are available
   1234      */
   1235     protected static ForkJoinTask<?> peekNextLocalTask() {
   1236         Thread t; ForkJoinPool.WorkQueue q;
   1237         if ((t = Thread.currentThread()) instanceof ForkJoinWorkerThread)
   1238             q = ((ForkJoinWorkerThread)t).workQueue;
   1239         else
   1240             q = ForkJoinPool.commonSubmitterQueue();
   1241         return (q == null) ? null : q.peek();
   1242     }
   1243 
   1244     /**
   1245      * Unschedules and returns, without executing, the next task
   1246      * queued by the current thread but not yet executed, if the
   1247      * current thread is operating in a ForkJoinPool.  This method is
   1248      * designed primarily to support extensions, and is unlikely to be
   1249      * useful otherwise.
   1250      *
   1251      * @return the next task, or {@code null} if none are available
   1252      */
   1253     protected static ForkJoinTask<?> pollNextLocalTask() {
   1254         Thread t;
   1255         return ((t = Thread.currentThread()) instanceof ForkJoinWorkerThread) ?
   1256             ((ForkJoinWorkerThread)t).workQueue.nextLocalTask() :
   1257             null;
   1258     }
   1259 
   1260     /**
   1261      * If the current thread is operating in a ForkJoinPool,
   1262      * unschedules and returns, without executing, the next task
   1263      * queued by the current thread but not yet executed, if one is
   1264      * available, or if not available, a task that was forked by some
   1265      * other thread, if available. Availability may be transient, so a
   1266      * {@code null} result does not necessarily imply quiescence of
   1267      * the pool this task is operating in.  This method is designed
   1268      * primarily to support extensions, and is unlikely to be useful
   1269      * otherwise.
   1270      *
   1271      * @return a task, or {@code null} if none are available
   1272      */
   1273     protected static ForkJoinTask<?> pollTask() {
   1274         Thread t; ForkJoinWorkerThread wt;
   1275         return ((t = Thread.currentThread()) instanceof ForkJoinWorkerThread) ?
   1276             (wt = (ForkJoinWorkerThread)t).pool.nextTaskFor(wt.workQueue) :
   1277             null;
   1278     }
   1279 
   1280     // tag operations
   1281 
   1282     /**
   1283      * Returns the tag for this task.
   1284      *
   1285      * @return the tag for this task
   1286      * @since 1.8
   1287      * @hide
   1288      */
   1289     public final short getForkJoinTaskTag() {
   1290         return (short)status;
   1291     }
   1292 
   1293     /**
   1294      * Atomically sets the tag value for this task.
   1295      *
   1296      * @param tag the tag value
   1297      * @return the previous value of the tag
   1298      * @since 1.8
   1299      * @hide
   1300      */
   1301     public final short setForkJoinTaskTag(short tag) {
   1302         for (int s;;) {
   1303             if (U.compareAndSwapInt(this, STATUS, s = status,
   1304                                     (s & ~SMASK) | (tag & SMASK)))
   1305                 return (short)s;
   1306         }
   1307     }
   1308 
   1309     /**
   1310      * Atomically conditionally sets the tag value for this task.
   1311      * Among other applications, tags can be used as visit markers
   1312      * in tasks operating on graphs, as in methods that check: {@code
   1313      * if (task.compareAndSetForkJoinTaskTag((short)0, (short)1))}
   1314      * before processing, otherwise exiting because the node has
   1315      * already been visited.
   1316      *
   1317      * @param e the expected tag value
   1318      * @param tag the new tag value
   1319      * @return true if successful; i.e., the current value was
   1320      * equal to e and is now tag.
   1321      * @since 1.8
   1322      * @hide
   1323      */
   1324     public final boolean compareAndSetForkJoinTaskTag(short e, short tag) {
   1325         for (int s;;) {
   1326             if ((short)(s = status) != e)
   1327                 return false;
   1328             if (U.compareAndSwapInt(this, STATUS, s,
   1329                                     (s & ~SMASK) | (tag & SMASK)))
   1330                 return true;
   1331         }
   1332     }
   1333 
   1334     /**
   1335      * Adaptor for Runnables. This implements RunnableFuture
   1336      * to be compliant with AbstractExecutorService constraints
   1337      * when used in ForkJoinPool.
   1338      */
   1339     static final class AdaptedRunnable<T> extends ForkJoinTask<T>
   1340         implements RunnableFuture<T> {
   1341         final Runnable runnable;
   1342         T result;
   1343         AdaptedRunnable(Runnable runnable, T result) {
   1344             if (runnable == null) throw new NullPointerException();
   1345             this.runnable = runnable;
   1346             this.result = result; // OK to set this even before completion
   1347         }
   1348         public final T getRawResult() { return result; }
   1349         public final void setRawResult(T v) { result = v; }
   1350         public final boolean exec() { runnable.run(); return true; }
   1351         public final void run() { invoke(); }
   1352         private static final long serialVersionUID = 5232453952276885070L;
   1353     }
   1354 
   1355     /**
   1356      * Adaptor for Runnables without results
   1357      */
   1358     static final class AdaptedRunnableAction extends ForkJoinTask<Void>
   1359         implements RunnableFuture<Void> {
   1360         final Runnable runnable;
   1361         AdaptedRunnableAction(Runnable runnable) {
   1362             if (runnable == null) throw new NullPointerException();
   1363             this.runnable = runnable;
   1364         }
   1365         public final Void getRawResult() { return null; }
   1366         public final void setRawResult(Void v) { }
   1367         public final boolean exec() { runnable.run(); return true; }
   1368         public final void run() { invoke(); }
   1369         private static final long serialVersionUID = 5232453952276885070L;
   1370     }
   1371 
   1372     /**
   1373      * Adaptor for Callables
   1374      */
   1375     static final class AdaptedCallable<T> extends ForkJoinTask<T>
   1376         implements RunnableFuture<T> {
   1377         final Callable<? extends T> callable;
   1378         T result;
   1379         AdaptedCallable(Callable<? extends T> callable) {
   1380             if (callable == null) throw new NullPointerException();
   1381             this.callable = callable;
   1382         }
   1383         public final T getRawResult() { return result; }
   1384         public final void setRawResult(T v) { result = v; }
   1385         public final boolean exec() {
   1386             try {
   1387                 result = callable.call();
   1388                 return true;
   1389             } catch (Error err) {
   1390                 throw err;
   1391             } catch (RuntimeException rex) {
   1392                 throw rex;
   1393             } catch (Exception ex) {
   1394                 throw new RuntimeException(ex);
   1395             }
   1396         }
   1397         public final void run() { invoke(); }
   1398         private static final long serialVersionUID = 2838392045355241008L;
   1399     }
   1400 
   1401     /**
   1402      * Returns a new {@code ForkJoinTask} that performs the {@code run}
   1403      * method of the given {@code Runnable} as its action, and returns
   1404      * a null result upon {@link #join}.
   1405      *
   1406      * @param runnable the runnable action
   1407      * @return the task
   1408      */
   1409     public static ForkJoinTask<?> adapt(Runnable runnable) {
   1410         return new AdaptedRunnableAction(runnable);
   1411     }
   1412 
   1413     /**
   1414      * Returns a new {@code ForkJoinTask} that performs the {@code run}
   1415      * method of the given {@code Runnable} as its action, and returns
   1416      * the given result upon {@link #join}.
   1417      *
   1418      * @param runnable the runnable action
   1419      * @param result the result upon completion
   1420      * @return the task
   1421      */
   1422     public static <T> ForkJoinTask<T> adapt(Runnable runnable, T result) {
   1423         return new AdaptedRunnable<T>(runnable, result);
   1424     }
   1425 
   1426     /**
   1427      * Returns a new {@code ForkJoinTask} that performs the {@code call}
   1428      * method of the given {@code Callable} as its action, and returns
   1429      * its result upon {@link #join}, translating any checked exceptions
   1430      * encountered into {@code RuntimeException}.
   1431      *
   1432      * @param callable the callable action
   1433      * @return the task
   1434      */
   1435     public static <T> ForkJoinTask<T> adapt(Callable<? extends T> callable) {
   1436         return new AdaptedCallable<T>(callable);
   1437     }
   1438 
   1439     // Serialization support
   1440 
   1441     private static final long serialVersionUID = -7721805057305804111L;
   1442 
   1443     /**
   1444      * Saves this task to a stream (that is, serializes it).
   1445      *
   1446      * @serialData the current run status and the exception thrown
   1447      * during execution, or {@code null} if none
   1448      */
   1449     private void writeObject(java.io.ObjectOutputStream s)
   1450         throws java.io.IOException {
   1451         s.defaultWriteObject();
   1452         s.writeObject(getException());
   1453     }
   1454 
   1455     /**
   1456      * Reconstitutes this task from a stream (that is, deserializes it).
   1457      */
   1458     private void readObject(java.io.ObjectInputStream s)
   1459         throws java.io.IOException, ClassNotFoundException {
   1460         s.defaultReadObject();
   1461         Object ex = s.readObject();
   1462         if (ex != null)
   1463             setExceptionalCompletion((Throwable)ex);
   1464     }
   1465 
   1466     // Unsafe mechanics
   1467     private static final sun.misc.Unsafe U;
   1468     private static final long STATUS;
   1469 
   1470     static {
   1471         exceptionTableLock = new ReentrantLock();
   1472         exceptionTableRefQueue = new ReferenceQueue<Object>();
   1473         exceptionTable = new ExceptionNode[EXCEPTION_MAP_CAPACITY];
   1474         try {
   1475             U = sun.misc.Unsafe.getUnsafe();
   1476             Class<?> k = ForkJoinTask.class;
   1477             STATUS = U.objectFieldOffset
   1478                 (k.getDeclaredField("status"));
   1479         } catch (Exception e) {
   1480             throw new Error(e);
   1481         }
   1482     }
   1483 
   1484 }
   1485