ViewVC Help
View File | Revision Log | Show Annotations | Download File | Root Listing
root/jsr166/jsr166/src/jsr166e/ForkJoinTask.java
(Generate patch)

Comparing jsr166/src/jsr166e/ForkJoinTask.java (file contents):
Revision 1.1 by dl, Mon Aug 13 15:52:33 2012 UTC vs.
Revision 1.14 by jsr166, Sun Jul 21 06:32:28 2013 UTC

# Line 30 | Line 30 | import java.lang.reflect.Constructor;
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 submitted
34 < * to a {@link ForkJoinPool}.  Once started, it will usually in turn
35 < * start other subtasks.  As indicated by the name of this class,
36 < * many programs using {@code ForkJoinTask} employ only methods
37 < * {@link #fork} and {@link #join}, or derivatives such as {@link
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
44 < * support of new forms of fork/join processing.
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
# Line 52 | Line 55 | import java.lang.reflect.Constructor;
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 IO, and should ideally access variables that
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
# Line 70 | Line 73 | import java.lang.reflect.Constructor;
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 IO. Event-style async
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
# Line 123 | Line 126 | import java.lang.reflect.Constructor;
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. While these methods have
127 < * {@code public} access (to allow instances of different task
128 < * subclasses to call each other's methods), some of them may only be
129 < * called from within other ForkJoinTasks (as may be determined using
130 < * method {@link #inForkJoinPool}).  Attempts to invoke them in other
131 < * contexts result in exceptions or errors, possibly including {@code
132 < * ClassCastException}.
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
# Line 137 | Line 134 | import java.lang.reflect.Constructor;
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
137 > * {@link java.util.concurrent.Phaser 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
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
# Line 287 | Line 284 | public abstract class ForkJoinTask<V> im
284       * @return status upon completion
285       */
286      private int externalAwaitDone() {
290        boolean interrupted = false;
287          int s;
288 <        while ((s = status) >= 0) {
289 <            if (U.compareAndSwapInt(this, STATUS, s, s | SIGNAL)) {
290 <                synchronized (this) {
291 <                    if (status >= 0) {
292 <                        try {
293 <                            wait();
294 <                        } catch (InterruptedException ie) {
295 <                            interrupted = true;
288 >        ForkJoinPool cp = ForkJoinPool.common;
289 >        if ((s = status) >= 0) {
290 >            if (cp != null) {
291 >                if (this instanceof CountedCompleter)
292 >                    s = cp.externalHelpComplete((CountedCompleter<?>)this);
293 >                else if (cp.tryExternalUnpush(this))
294 >                    s = doExec();
295 >            }
296 >            if (s >= 0 && (s = status) >= 0) {
297 >                boolean interrupted = false;
298 >                do {
299 >                    if (U.compareAndSwapInt(this, STATUS, s, s | SIGNAL)) {
300 >                        synchronized (this) {
301 >                            if (status >= 0) {
302 >                                try {
303 >                                    wait();
304 >                                } catch (InterruptedException ie) {
305 >                                    interrupted = true;
306 >                                }
307 >                            }
308 >                            else
309 >                                notifyAll();
310                          }
311                      }
312 <                    else
313 <                        notifyAll();
314 <                }
312 >                } while ((s = status) >= 0);
313 >                if (interrupted)
314 >                    Thread.currentThread().interrupt();
315              }
316          }
307        if (interrupted)
308            Thread.currentThread().interrupt();
317          return s;
318      }
319  
# Line 314 | Line 322 | public abstract class ForkJoinTask<V> im
322       */
323      private int externalInterruptibleAwaitDone() throws InterruptedException {
324          int s;
325 +        ForkJoinPool cp = ForkJoinPool.common;
326          if (Thread.interrupted())
327              throw new InterruptedException();
328 +        if ((s = status) >= 0 && cp != null) {
329 +            if (this instanceof CountedCompleter)
330 +                cp.externalHelpComplete((CountedCompleter<?>)this);
331 +            else if (cp.tryExternalUnpush(this))
332 +                doExec();
333 +        }
334          while ((s = status) >= 0) {
335              if (U.compareAndSwapInt(this, STATUS, s, s | SIGNAL)) {
336                  synchronized (this) {
# Line 329 | Line 344 | public abstract class ForkJoinTask<V> im
344          return s;
345      }
346  
347 +
348      /**
349       * Implementation for join, get, quietlyJoin. Directly handles
350       * only cases of already-completed, external wait, and
# Line 338 | Line 354 | public abstract class ForkJoinTask<V> im
354       */
355      private int doJoin() {
356          int s; Thread t; ForkJoinWorkerThread wt; ForkJoinPool.WorkQueue w;
357 <        if ((s = status) >= 0) {
358 <            if (((t = Thread.currentThread()) instanceof ForkJoinWorkerThread)) {
359 <                if (!(w = (wt = (ForkJoinWorkerThread)t).workQueue).
360 <                    tryUnpush(this) || (s = doExec()) >= 0)
361 <                    s = wt.pool.awaitJoin(w, this);
362 <            }
347 <            else
348 <                s = externalAwaitDone();
349 <        }
350 <        return s;
357 >        return (s = status) < 0 ? s :
358 >            ((t = Thread.currentThread()) instanceof ForkJoinWorkerThread) ?
359 >            (w = (wt = (ForkJoinWorkerThread)t).workQueue).
360 >            tryUnpush(this) && (s = doExec()) < 0 ? s :
361 >            wt.pool.awaitJoin(w, this) :
362 >            externalAwaitDone();
363      }
364  
365      /**
# Line 357 | Line 369 | public abstract class ForkJoinTask<V> im
369       */
370      private int doInvoke() {
371          int s; Thread t; ForkJoinWorkerThread wt;
372 <        if ((s = doExec()) >= 0) {
373 <            if ((t = Thread.currentThread()) instanceof ForkJoinWorkerThread)
374 <                s = (wt = (ForkJoinWorkerThread)t).pool.awaitJoin(wt.workQueue,
375 <                                                                  this);
364 <            else
365 <                s = externalAwaitDone();
366 <        }
367 <        return s;
372 >        return (s = doExec()) < 0 ? s :
373 >            ((t = Thread.currentThread()) instanceof ForkJoinWorkerThread) ?
374 >            (wt = (ForkJoinWorkerThread)t).pool.awaitJoin(wt.workQueue, this) :
375 >            externalAwaitDone();
376      }
377  
378      // Exception table support
# Line 443 | Line 451 | public abstract class ForkJoinTask<V> im
451      }
452  
453      /**
454 <     * Records exception and possibly propagates
454 >     * Records exception and possibly propagates.
455       *
456       * @return status on exit
457       */
# Line 476 | Line 484 | public abstract class ForkJoinTask<V> im
484      }
485  
486      /**
487 <     * Removes exception node and clears status
487 >     * Removes exception node and clears status.
488       */
489      private void clearExceptionalCompletion() {
490          int h = System.identityHashCode(this);
# Line 606 | Line 614 | public abstract class ForkJoinTask<V> im
614      }
615  
616      /**
617 +     * A version of "sneaky throw" to relay exceptions
618 +     */
619 +    static void rethrow(Throwable ex) {
620 +        if (ex != null)
621 +            ForkJoinTask.<RuntimeException>uncheckedThrow(ex);
622 +    }
623 +
624 +    /**
625 +     * The sneaky part of sneaky throw, relying on generics
626 +     * limitations to evade compiler complaints about rethrowing
627 +     * unchecked exceptions
628 +     */
629 +    @SuppressWarnings("unchecked") static <T extends Throwable>
630 +        void uncheckedThrow(Throwable t) throws T {
631 +        throw (T)t; // rely on vacuous cast
632 +    }
633 +
634 +    /**
635       * Throws exception, if any, associated with the given status.
636       */
637      private void reportException(int s) {
638 <        Throwable ex = ((s == CANCELLED) ?  new CancellationException() :
639 <                        (s == EXCEPTIONAL) ? getThrowableException() :
640 <                        null);
641 <        if (ex != null)
616 <            U.throwException(ex);
638 >        if (s == CANCELLED)
639 >            throw new CancellationException();
640 >        if (s == EXCEPTIONAL)
641 >            rethrow(getThrowableException());
642      }
643  
644      // public methods
645  
646      /**
647 <     * Arranges to asynchronously execute this task.  While it is not
648 <     * necessarily enforced, it is a usage error to fork a task more
649 <     * than once unless it has completed and been reinitialized.
650 <     * Subsequent modifications to the state of this task or any data
651 <     * it operates on are not necessarily consistently observable by
652 <     * any thread other than the one executing it unless preceded by a
653 <     * call to {@link #join} or related methods, or a call to {@link
654 <     * #isDone} returning {@code true}.
655 <     *
656 <     * <p>This method may be invoked only from within {@code
657 <     * ForkJoinPool} computations (as may be determined using method
633 <     * {@link #inForkJoinPool}).  Attempts to invoke in other contexts
634 <     * result in exceptions or errors, possibly including {@code
635 <     * ClassCastException}.
647 >     * Arranges to asynchronously execute this task in the pool the
648 >     * current task is running in, if applicable, or using the {@link
649 >     * ForkJoinPool#commonPool()} if not {@link #inForkJoinPool}.  While
650 >     * it is not necessarily enforced, it is a usage error to fork a
651 >     * task more than once unless it has completed and been
652 >     * reinitialized.  Subsequent modifications to the state of this
653 >     * task or any data it operates on are not necessarily
654 >     * consistently observable by any thread other than the one
655 >     * executing it unless preceded by a call to {@link #join} or
656 >     * related methods, or a call to {@link #isDone} returning {@code
657 >     * true}.
658       *
659       * @return {@code this}, to simplify usage
660       */
661      public final ForkJoinTask<V> fork() {
662 <        ((ForkJoinWorkerThread)Thread.currentThread()).workQueue.push(this);
662 >        Thread t;
663 >        if ((t = Thread.currentThread()) instanceof ForkJoinWorkerThread)
664 >            ((ForkJoinWorkerThread)t).workQueue.push(this);
665 >        else
666 >            ForkJoinPool.common.externalPush(this);
667          return this;
668      }
669  
# Line 687 | Line 713 | public abstract class ForkJoinTask<V> im
713       * cancelled, completed normally or exceptionally, or left
714       * unprocessed.
715       *
690     * <p>This method may be invoked only from within {@code
691     * ForkJoinPool} computations (as may be determined using method
692     * {@link #inForkJoinPool}).  Attempts to invoke in other contexts
693     * result in exceptions or errors, possibly including {@code
694     * ClassCastException}.
695     *
716       * @param t1 the first task
717       * @param t2 the second task
718       * @throws NullPointerException if any task is null
# Line 718 | Line 738 | public abstract class ForkJoinTask<V> im
738       * related methods to check if they have been cancelled, completed
739       * normally or exceptionally, or left unprocessed.
740       *
721     * <p>This method may be invoked only from within {@code
722     * ForkJoinPool} computations (as may be determined using method
723     * {@link #inForkJoinPool}).  Attempts to invoke in other contexts
724     * result in exceptions or errors, possibly including {@code
725     * ClassCastException}.
726     *
741       * @param tasks the tasks
742       * @throws NullPointerException if any task is null
743       */
# Line 751 | Line 765 | public abstract class ForkJoinTask<V> im
765              }
766          }
767          if (ex != null)
768 <            U.throwException(ex);
768 >            rethrow(ex);
769      }
770  
771      /**
# Line 767 | Line 781 | public abstract class ForkJoinTask<V> im
781       * cancelled, completed normally or exceptionally, or left
782       * unprocessed.
783       *
770     * <p>This method may be invoked only from within {@code
771     * ForkJoinPool} computations (as may be determined using method
772     * {@link #inForkJoinPool}).  Attempts to invoke in other contexts
773     * result in exceptions or errors, possibly including {@code
774     * ClassCastException}.
775     *
784       * @param tasks the collection of tasks
785 +     * @param <T> the type of the values returned from the tasks
786       * @return the tasks argument, to simplify usage
787       * @throws NullPointerException if tasks or any element are null
788       */
# Line 808 | Line 817 | public abstract class ForkJoinTask<V> im
817              }
818          }
819          if (ex != null)
820 <            U.throwException(ex);
820 >            rethrow(ex);
821          return tasks;
822      }
823  
# Line 831 | Line 840 | public abstract class ForkJoinTask<V> im
840       * <p>This method is designed to be invoked by <em>other</em>
841       * tasks. To terminate the current task, you can just return or
842       * throw an unchecked exception from its computation method, or
843 <     * invoke {@link #completeExceptionally}.
843 >     * invoke {@link #completeExceptionally(Throwable)}.
844       *
845       * @param mayInterruptIfRunning this value has no effect in the
846       * default implementation because interrupts are not used to
# Line 981 | Line 990 | public abstract class ForkJoinTask<V> im
990          if (Thread.interrupted())
991              throw new InterruptedException();
992          // Messy in part because we measure in nanosecs, but wait in millisecs
993 <        int s; long ns, ms;
994 <        if ((s = status) >= 0 && (ns = unit.toNanos(timeout)) > 0L) {
993 >        int s; long ms;
994 >        long ns = unit.toNanos(timeout);
995 >        ForkJoinPool cp;
996 >        if ((s = status) >= 0 && ns > 0L) {
997              long deadline = System.nanoTime() + ns;
998              ForkJoinPool p = null;
999              ForkJoinPool.WorkQueue w = null;
# Line 991 | Line 1002 | public abstract class ForkJoinTask<V> im
1002                  ForkJoinWorkerThread wt = (ForkJoinWorkerThread)t;
1003                  p = wt.pool;
1004                  w = wt.workQueue;
1005 <                s = p.helpJoinOnce(w, this); // no retries on failure
1005 >                p.helpJoinOnce(w, this); // no retries on failure
1006 >            }
1007 >            else if ((cp = ForkJoinPool.common) != null) {
1008 >                if (this instanceof CountedCompleter)
1009 >                    cp.externalHelpComplete((CountedCompleter<?>)this);
1010 >                else if (cp.tryExternalUnpush(this))
1011 >                    doExec();
1012              }
1013              boolean canBlock = false;
1014              boolean interrupted = false;
1015              try {
1016                  while ((s = status) >= 0) {
1017 <                    if (w != null && w.runState < 0)
1017 >                    if (w != null && w.qlock < 0)
1018                          cancelIgnoringExceptions(this);
1019                      else if (!canBlock) {
1020 <                        if (p == null || p.tryCompensate(this, null))
1020 >                        if (p == null || p.tryCompensate(p.ctl))
1021                              canBlock = true;
1022                      }
1023                      else {
# Line 1068 | Line 1085 | public abstract class ForkJoinTask<V> im
1085       * be of use in designs in which many tasks are forked, but none
1086       * are explicitly joined, instead executing them until all are
1087       * processed.
1071     *
1072     * <p>This method may be invoked only from within {@code
1073     * ForkJoinPool} computations (as may be determined using method
1074     * {@link #inForkJoinPool}).  Attempts to invoke in other contexts
1075     * result in exceptions or errors, possibly including {@code
1076     * ClassCastException}.
1088       */
1089      public static void helpQuiesce() {
1090 <        ForkJoinWorkerThread wt =
1091 <            (ForkJoinWorkerThread)Thread.currentThread();
1092 <        wt.pool.helpQuiescePool(wt.workQueue);
1090 >        Thread t;
1091 >        if ((t = Thread.currentThread()) instanceof ForkJoinWorkerThread) {
1092 >            ForkJoinWorkerThread wt = (ForkJoinWorkerThread)t;
1093 >            wt.pool.helpQuiescePool(wt.workQueue);
1094 >        }
1095 >        else
1096 >            ForkJoinPool.quiesceCommonPool();
1097      }
1098  
1099      /**
# Line 1131 | Line 1146 | public abstract class ForkJoinTask<V> im
1146  
1147      /**
1148       * Tries to unschedule this task for execution. This method will
1149 <     * typically succeed if this task is the most recently forked task
1150 <     * by the current thread, and has not commenced executing in
1151 <     * another thread.  This method may be useful when arranging
1152 <     * alternative local processing of tasks that could have been, but
1153 <     * were not, stolen.
1139 <     *
1140 <     * <p>This method may be invoked only from within {@code
1141 <     * ForkJoinPool} computations (as may be determined using method
1142 <     * {@link #inForkJoinPool}).  Attempts to invoke in other contexts
1143 <     * result in exceptions or errors, possibly including {@code
1144 <     * ClassCastException}.
1149 >     * typically (but is not guaranteed to) succeed if this task is
1150 >     * the most recently forked task by the current thread, and has
1151 >     * not commenced executing in another thread.  This method may be
1152 >     * useful when arranging alternative local processing of tasks
1153 >     * that could have been, but were not, stolen.
1154       *
1155       * @return {@code true} if unforked
1156       */
1157      public boolean tryUnfork() {
1158 <        return ((ForkJoinWorkerThread)Thread.currentThread())
1159 <            .workQueue.tryUnpush(this);
1158 >        Thread t;
1159 >        return (((t = Thread.currentThread()) instanceof ForkJoinWorkerThread) ?
1160 >                ((ForkJoinWorkerThread)t).workQueue.tryUnpush(this) :
1161 >                ForkJoinPool.common.tryExternalUnpush(this));
1162      }
1163  
1164      /**
# Line 1156 | Line 1167 | public abstract class ForkJoinTask<V> im
1167       * value may be useful for heuristic decisions about whether to
1168       * fork other tasks.
1169       *
1159     * <p>This method may be invoked only from within {@code
1160     * ForkJoinPool} computations (as may be determined using method
1161     * {@link #inForkJoinPool}).  Attempts to invoke in other contexts
1162     * result in exceptions or errors, possibly including {@code
1163     * ClassCastException}.
1164     *
1170       * @return the number of tasks
1171       */
1172      public static int getQueuedTaskCount() {
1173 <        return ((ForkJoinWorkerThread) Thread.currentThread())
1174 <            .workQueue.queueSize();
1173 >        Thread t; ForkJoinPool.WorkQueue q;
1174 >        if ((t = Thread.currentThread()) instanceof ForkJoinWorkerThread)
1175 >            q = ((ForkJoinWorkerThread)t).workQueue;
1176 >        else
1177 >            q = ForkJoinPool.commonSubmitterQueue();
1178 >        return (q == null) ? 0 : q.queueSize();
1179      }
1180  
1181      /**
1182       * Returns an estimate of how many more locally queued tasks are
1183       * held by the current worker thread than there are other worker
1184 <     * threads that might steal them.  This value may be useful for
1184 >     * threads that might steal them, or zero if this thread is not
1185 >     * operating in a ForkJoinPool. This value may be useful for
1186       * heuristic decisions about whether to fork other tasks. In many
1187       * usages of ForkJoinTasks, at steady state, each worker should
1188       * aim to maintain a small constant surplus (for example, 3) of
1189       * tasks, and to process computations locally if this threshold is
1190       * exceeded.
1191       *
1182     * <p>This method may be invoked only from within {@code
1183     * ForkJoinPool} computations (as may be determined using method
1184     * {@link #inForkJoinPool}).  Attempts to invoke in other contexts
1185     * result in exceptions or errors, possibly including {@code
1186     * ClassCastException}.
1187     *
1192       * @return the surplus number of tasks, which may be negative
1193       */
1194      public static int getSurplusQueuedTaskCount() {
1195 <        /*
1192 <         * The aim of this method is to return a cheap heuristic guide
1193 <         * for task partitioning when programmers, frameworks, tools,
1194 <         * or languages have little or no idea about task granularity.
1195 <         * In essence by offering this method, we ask users only about
1196 <         * tradeoffs in overhead vs expected throughput and its
1197 <         * variance, rather than how finely to partition tasks.
1198 <         *
1199 <         * In a steady state strict (tree-structured) computation,
1200 <         * each thread makes available for stealing enough tasks for
1201 <         * other threads to remain active. Inductively, if all threads
1202 <         * play by the same rules, each thread should make available
1203 <         * only a constant number of tasks.
1204 <         *
1205 <         * The minimum useful constant is just 1. But using a value of
1206 <         * 1 would require immediate replenishment upon each steal to
1207 <         * maintain enough tasks, which is infeasible.  Further,
1208 <         * partitionings/granularities of offered tasks should
1209 <         * minimize steal rates, which in general means that threads
1210 <         * nearer the top of computation tree should generate more
1211 <         * than those nearer the bottom. In perfect steady state, each
1212 <         * thread is at approximately the same level of computation
1213 <         * tree. However, producing extra tasks amortizes the
1214 <         * uncertainty of progress and diffusion assumptions.
1215 <         *
1216 <         * So, users will want to use values larger, but not much
1217 <         * larger than 1 to both smooth over transient shortages and
1218 <         * hedge against uneven progress; as traded off against the
1219 <         * cost of extra task overhead. We leave the user to pick a
1220 <         * threshold value to compare with the results of this call to
1221 <         * guide decisions, but recommend values such as 3.
1222 <         *
1223 <         * When all threads are active, it is on average OK to
1224 <         * estimate surplus strictly locally. In steady-state, if one
1225 <         * thread is maintaining say 2 surplus tasks, then so are
1226 <         * others. So we can just use estimated queue length.
1227 <         * However, this strategy alone leads to serious mis-estimates
1228 <         * in some non-steady-state conditions (ramp-up, ramp-down,
1229 <         * other stalls). We can detect many of these by further
1230 <         * considering the number of "idle" threads, that are known to
1231 <         * have zero queued tasks, so compensate by a factor of
1232 <         * (#idle/#active) threads.
1233 <         */
1234 <        ForkJoinWorkerThread wt =
1235 <            (ForkJoinWorkerThread)Thread.currentThread();
1236 <        return wt.workQueue.queueSize() - wt.pool.idlePerActive();
1195 >        return ForkJoinPool.getSurplusQueuedTaskCount();
1196      }
1197  
1198      // Extension methods
# Line 1284 | Line 1243 | public abstract class ForkJoinTask<V> im
1243       * primarily to support extensions, and is unlikely to be useful
1244       * otherwise.
1245       *
1287     * <p>This method may be invoked only from within {@code
1288     * ForkJoinPool} computations (as may be determined using method
1289     * {@link #inForkJoinPool}).  Attempts to invoke in other contexts
1290     * result in exceptions or errors, possibly including {@code
1291     * ClassCastException}.
1292     *
1246       * @return the next task, or {@code null} if none are available
1247       */
1248      protected static ForkJoinTask<?> peekNextLocalTask() {
1249 <        return ((ForkJoinWorkerThread) Thread.currentThread()).workQueue.peek();
1249 >        Thread t; ForkJoinPool.WorkQueue q;
1250 >        if ((t = Thread.currentThread()) instanceof ForkJoinWorkerThread)
1251 >            q = ((ForkJoinWorkerThread)t).workQueue;
1252 >        else
1253 >            q = ForkJoinPool.commonSubmitterQueue();
1254 >        return (q == null) ? null : q.peek();
1255      }
1256  
1257      /**
1258       * Unschedules and returns, without executing, the next task
1259 <     * queued by the current thread but not yet executed.  This method
1260 <     * is designed primarily to support extensions, and is unlikely to
1261 <     * be useful otherwise.
1262 <     *
1305 <     * <p>This method may be invoked only from within {@code
1306 <     * ForkJoinPool} computations (as may be determined using method
1307 <     * {@link #inForkJoinPool}).  Attempts to invoke in other contexts
1308 <     * result in exceptions or errors, possibly including {@code
1309 <     * ClassCastException}.
1259 >     * queued by the current thread but not yet executed, if the
1260 >     * current thread is operating in a ForkJoinPool.  This method is
1261 >     * designed primarily to support extensions, and is unlikely to be
1262 >     * useful otherwise.
1263       *
1264       * @return the next task, or {@code null} if none are available
1265       */
1266      protected static ForkJoinTask<?> pollNextLocalTask() {
1267 <        return ((ForkJoinWorkerThread) Thread.currentThread())
1268 <            .workQueue.nextLocalTask();
1267 >        Thread t;
1268 >        return ((t = Thread.currentThread()) instanceof ForkJoinWorkerThread) ?
1269 >            ((ForkJoinWorkerThread)t).workQueue.nextLocalTask() :
1270 >            null;
1271      }
1272  
1273      /**
1274 <     * Unschedules and returns, without executing, the next task
1274 >     * If the current thread is operating in a ForkJoinPool,
1275 >     * unschedules and returns, without executing, the next task
1276       * queued by the current thread but not yet executed, if one is
1277       * available, or if not available, a task that was forked by some
1278       * other thread, if available. Availability may be transient, so a
1279 <     * {@code null} result does not necessarily imply quiescence
1280 <     * of the pool this task is operating in.  This method is designed
1279 >     * {@code null} result does not necessarily imply quiescence of
1280 >     * the pool this task is operating in.  This method is designed
1281       * primarily to support extensions, and is unlikely to be useful
1282       * otherwise.
1283       *
1328     * <p>This method may be invoked only from within {@code
1329     * ForkJoinPool} computations (as may be determined using method
1330     * {@link #inForkJoinPool}).  Attempts to invoke in other contexts
1331     * result in exceptions or errors, possibly including {@code
1332     * ClassCastException}.
1333     *
1284       * @return a task, or {@code null} if none are available
1285       */
1286      protected static ForkJoinTask<?> pollTask() {
1287 <        ForkJoinWorkerThread wt =
1288 <            (ForkJoinWorkerThread)Thread.currentThread();
1289 <        return wt.pool.nextTaskFor(wt.workQueue);
1287 >        Thread t; ForkJoinWorkerThread wt;
1288 >        return ((t = Thread.currentThread()) instanceof ForkJoinWorkerThread) ?
1289 >            (wt = (ForkJoinWorkerThread)t).pool.nextTaskFor(wt.workQueue) :
1290 >            null;
1291      }
1292  
1293      // tag operations
# Line 1376 | Line 1327 | public abstract class ForkJoinTask<V> im
1327       *
1328       * @param e the expected tag value
1329       * @param tag the new tag value
1330 <     * @return true if successful; i.e., the current value was
1330 >     * @return {@code true} if successful; i.e., the current value was
1331       * equal to e and is now tag.
1332       * @since 1.8
1333       */
# Line 1429 | Line 1380 | public abstract class ForkJoinTask<V> im
1380      }
1381  
1382      /**
1383 +     * Adaptor for Runnables in which failure forces worker exception
1384 +     */
1385 +    static final class RunnableExecuteAction extends ForkJoinTask<Void> {
1386 +        final Runnable runnable;
1387 +        RunnableExecuteAction(Runnable runnable) {
1388 +            if (runnable == null) throw new NullPointerException();
1389 +            this.runnable = runnable;
1390 +        }
1391 +        public final Void getRawResult() { return null; }
1392 +        public final void setRawResult(Void v) { }
1393 +        public final boolean exec() { runnable.run(); return true; }
1394 +        void internalPropagateException(Throwable ex) {
1395 +            rethrow(ex); // rethrow outside exec() catches.
1396 +        }
1397 +        private static final long serialVersionUID = 5232453952276885070L;
1398 +    }
1399 +
1400 +    /**
1401       * Adaptor for Callables
1402       */
1403      static final class AdaptedCallable<T> extends ForkJoinTask<T>
# Line 1476 | Line 1445 | public abstract class ForkJoinTask<V> im
1445       *
1446       * @param runnable the runnable action
1447       * @param result the result upon completion
1448 +     * @param <T> the type of the result
1449       * @return the task
1450       */
1451      public static <T> ForkJoinTask<T> adapt(Runnable runnable, T result) {
# Line 1489 | Line 1459 | public abstract class ForkJoinTask<V> im
1459       * encountered into {@code RuntimeException}.
1460       *
1461       * @param callable the callable action
1462 +     * @param <T> the type of the callable's result
1463       * @return the task
1464       */
1465      public static <T> ForkJoinTask<T> adapt(Callable<? extends T> callable) {
# Line 1525 | Line 1496 | public abstract class ForkJoinTask<V> im
1496      // Unsafe mechanics
1497      private static final sun.misc.Unsafe U;
1498      private static final long STATUS;
1499 +
1500      static {
1501          exceptionTableLock = new ReentrantLock();
1502          exceptionTableRefQueue = new ReferenceQueue<Object>();
1503          exceptionTable = new ExceptionNode[EXCEPTION_MAP_CAPACITY];
1504          try {
1505              U = getUnsafe();
1506 +            Class<?> k = ForkJoinTask.class;
1507              STATUS = U.objectFieldOffset
1508 <                (ForkJoinTask.class.getDeclaredField("status"));
1508 >                (k.getDeclaredField("status"));
1509          } catch (Exception e) {
1510              throw new Error(e);
1511          }
# Line 1548 | Line 1521 | public abstract class ForkJoinTask<V> im
1521      private static sun.misc.Unsafe getUnsafe() {
1522          try {
1523              return sun.misc.Unsafe.getUnsafe();
1524 <        } catch (SecurityException se) {
1525 <            try {
1526 <                return java.security.AccessController.doPrivileged
1527 <                    (new java.security
1528 <                     .PrivilegedExceptionAction<sun.misc.Unsafe>() {
1529 <                        public sun.misc.Unsafe run() throws Exception {
1530 <                            java.lang.reflect.Field f = sun.misc
1531 <                                .Unsafe.class.getDeclaredField("theUnsafe");
1532 <                            f.setAccessible(true);
1533 <                            return (sun.misc.Unsafe) f.get(null);
1534 <                        }});
1535 <            } catch (java.security.PrivilegedActionException e) {
1536 <                throw new RuntimeException("Could not initialize intrinsics",
1537 <                                           e.getCause());
1538 <            }
1524 >        } catch (SecurityException tryReflectionInstead) {}
1525 >        try {
1526 >            return java.security.AccessController.doPrivileged
1527 >            (new java.security.PrivilegedExceptionAction<sun.misc.Unsafe>() {
1528 >                public sun.misc.Unsafe run() throws Exception {
1529 >                    Class<sun.misc.Unsafe> k = sun.misc.Unsafe.class;
1530 >                    for (java.lang.reflect.Field f : k.getDeclaredFields()) {
1531 >                        f.setAccessible(true);
1532 >                        Object x = f.get(null);
1533 >                        if (k.isInstance(x))
1534 >                            return k.cast(x);
1535 >                    }
1536 >                    throw new NoSuchFieldError("the Unsafe");
1537 >                }});
1538 >        } catch (java.security.PrivilegedActionException e) {
1539 >            throw new RuntimeException("Could not initialize intrinsics",
1540 >                                       e.getCause());
1541          }
1542      }
1543   }

Diff Legend

Removed lines
+ Added lines
< Changed lines
> Changed lines