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

Comparing jsr166/src/jsr166y/ForkJoinTask.java (file contents):
Revision 1.85 by jsr166, Tue Jan 31 01:51:13 2012 UTC vs.
Revision 1.87 by dl, Sun Mar 4 15:52:45 2012 UTC

# Line 137 | Line 137 | import java.lang.reflect.Constructor;
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>marked</em> using {@link
141 < * #markForkJoinTask} and checked for marking using {@link
142 < * #isMarkedForkJoinTask}. The ForkJoinTask implementation does not
143 < * use these {@code protected} methods or marks for any purpose, but
140 > * ForkJoinTask may be atomically <em>tagged</em> with a {@code
141 > * short} value using {@link #setForkJoinTaskTag} or {@link
142 > * #compareAndSetForkJoinTaskTag} and checked using {@link
143 > * #getForkJoinTaskTag}. The ForkJoinTask implementation does not
144 > * use these {@code protected} methods or tags for any purpose, but
145   * they may be of use in the construction of specialized subclasses.
146   * For example, parallel graph traversals can use the supplied methods
147   * to avoid revisiting nodes/tasks that have already been processed.
148 < * Also, completion based designs can use them to record that one
149 < * subtask has completed. (Method names for marking are bulky in part
150 < * to encourage definition of methods that reflect their usage
150 < * patterns.)
148 > * Also, completion based designs can use them to record that subtasks
149 > * have completed. (Method names for tagging are bulky in part to
150 > * encourage definition of methods that reflect their usage patterns.)
151   *
152   * <p>Most base support methods are {@code final}, to prevent
153   * overriding of implementations that are intrinsically tied to the
# Line 197 | Line 197 | public abstract class ForkJoinTask<V> im
197       * methods in a way that flows well in javadocs.
198       */
199  
200    /**
201     * The number of times to try to help join a task without any
202     * apparent progress before giving up and blocking. The value is
203     * arbitrary but should be large enough to cope with transient
204     * stalls (due to GC etc) that can cause helping methods not to be
205     * able to proceed because other workers have not progressed to
206     * the point where subtasks can be found or taken.
207     */
208    private static final int HELP_RETRIES = 32;
209
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 holds value
205 <     * NORMAL, CANCELLED, or EXCEPTIONAL. Tasks undergoing blocking
206 <     * waits by other threads have the SIGNAL bit set.  Completion of
207 <     * a stolen task with SIGNAL set awakens any waiters via
208 <     * notifyAll. Even though suboptimal for some purposes, we use
209 <     * basic builtin wait/notify to take advantage of "monitor
210 <     * inflation" in JVMs that we would otherwise need to emulate to
211 <     * avoid adding further per-task bookkeeping overhead.  We want
212 <     * these monitors to be "fat", i.e., not use biasing or thin-lock
213 <     * techniques, so use some odd coding idioms that tend to avoid
214 <     * them.
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 NORMAL      = 0xfffffffc;  // negative with low 2 bits 0
225 <    static final int CANCELLED   = 0xfffffff8;  // must be < NORMAL
226 <    static final int EXCEPTIONAL = 0xfffffff4;  // must be < CANCELLED
227 <    static final int SIGNAL      = 0x00000001;
228 <    static final int MARKED      = 0x00000002;
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, also clearing signal request bits. A specialization for
238 <     * NORMAL completion is in method doExec.
233 >     * task.
234       *
235       * @param completion one of NORMAL, CANCELLED, EXCEPTIONAL
236       * @return completion status on exit
# Line 244 | Line 239 | public abstract class ForkJoinTask<V> im
239          for (int s;;) {
240              if ((s = status) < 0)
241                  return s;
242 <            if (U.compareAndSwapInt(this, STATUS, s, (s & ~SIGNAL)|completion)) {
243 <                if ((s & SIGNAL) != 0)
242 >            if (U.compareAndSwapInt(this, STATUS, s, s | completion)) {
243 >                if ((s >>> 16) != 0)
244                      synchronized (this) { notifyAll(); }
245                  return completion;
246              }
# Line 267 | Line 262 | public abstract class ForkJoinTask<V> im
262              } catch (Throwable rex) {
263                  return setExceptionalCompletion(rex);
264              }
265 <            while ((s = status) >= 0 && completed) {
266 <                if (U.compareAndSwapInt(this, STATUS, s, (s & ~SIGNAL)|NORMAL)) {
272 <                    if ((s & SIGNAL) != 0)
273 <                        synchronized (this) { notifyAll(); }
274 <                    return NORMAL;
275 <                }
276 <            }
265 >            if (completed)
266 >                s = setCompletion(NORMAL);
267          }
268          return s;
269      }
270  
271      /**
272 +     * Tries to set SIGNAL status. Used by ForkJoinPool. Other
273 +     * variants are directly incorporated into externalAwaitDone etc.
274 +     *
275 +     * @return true if successful
276 +     */
277 +    final boolean trySetSignal() {
278 +        int s;
279 +        return U.compareAndSwapInt(this, STATUS, s = status, s | SIGNAL);
280 +    }
281 +
282 +    /**
283       * Blocks a non-worker-thread until completion.
284       * @return status upon completion
285       */
286      private int externalAwaitDone() {
287 +        boolean interrupted = false;
288          int s;
289 <        if ((s = status) >= 0) {
290 <            boolean interrupted = false;
291 <            synchronized (this) {
292 <                while ((s = status) >= 0) {
291 <                    if (U.compareAndSwapInt(this, STATUS, s, s | SIGNAL)) {
289 >        while ((s = status) >= 0) {
290 >            if (U.compareAndSwapInt(this, STATUS, s, s | SIGNAL)) {
291 >                synchronized (this) {
292 >                    if (status >= 0) {
293                          try {
294                              wait();
295                          } catch (InterruptedException ie) {
296                              interrupted = true;
297                          }
298                      }
299 +                    else
300 +                        notifyAll();
301                  }
302              }
300            if (interrupted)
301                Thread.currentThread().interrupt();
303          }
304 +        if (interrupted)
305 +            Thread.currentThread().interrupt();
306          return s;
307      }
308  
309      /**
310 <     * Blocks a non-worker-thread until completion or interruption or timeout.
310 >     * Blocks a non-worker-thread until completion or interruption.
311       */
312 <    private int externalInterruptibleAwaitDone(long millis)
310 <        throws InterruptedException {
312 >    private int externalInterruptibleAwaitDone() throws InterruptedException {
313          int s;
314          if (Thread.interrupted())
315              throw new InterruptedException();
316 <        if ((s = status) >= 0) {
317 <            synchronized (this) {
318 <                while ((s = status) >= 0) {
319 <                    if (U.compareAndSwapInt(this, STATUS, s, s | SIGNAL)) {
320 <                        wait(millis);
321 <                        if (millis > 0L)
322 <                            break;
321 <                    }
316 >        while ((s = status) >= 0) {
317 >            if (U.compareAndSwapInt(this, STATUS, s, s | SIGNAL)) {
318 >                synchronized (this) {
319 >                    if (status >= 0)
320 >                        wait();
321 >                    else
322 >                        notifyAll();
323                  }
324              }
325          }
326          return s;
327      }
328  
328
329      /**
330       * Implementation for join, get, quietlyJoin. Directly handles
331       * only cases of already-completed, external wait, and
332 <     * unfork+exec.  Others are relayed to awaitJoin.
332 >     * unfork+exec.  Others are relayed to ForkJoinPool.awaitJoin.
333       *
334       * @return status upon completion
335       */
336      private int doJoin() {
337          int s; Thread t; ForkJoinWorkerThread wt; ForkJoinPool.WorkQueue w;
338          if ((s = status) >= 0) {
339 <            if (!((t = Thread.currentThread()) instanceof ForkJoinWorkerThread))
340 <                s = externalAwaitDone();
341 <            else if (!(w = (wt = (ForkJoinWorkerThread)t).workQueue).
342 <                     tryUnpush(this) || (s = doExec()) >= 0)
343 <                s = awaitJoin(w, wt.pool);
344 <        }
345 <        return s;
346 <    }
347 <
348 <    /**
349 <     * Helps and/or blocks until joined.
350 <     *
351 <     * @param w the joiner
352 <     * @param p the pool
353 <     * @return status upon completion
354 <     */
355 <    private int awaitJoin(ForkJoinPool.WorkQueue w, ForkJoinPool p) {
356 <        int s;
357 <        ForkJoinTask<?> prevJoin = w.currentJoin;
358 <        w.currentJoin = this;
359 <        for (int k = HELP_RETRIES; (s = status) >= 0;) {
360 <            if ((w.queueSize() > 0) ?
361 <                w.tryRemoveAndExec(this) :        // self-help
362 <                p.tryHelpStealer(w, this))        // help process tasks
363 <                k = HELP_RETRIES;                 // reset if made progress
364 <            else if ((s = status) < 0)            // recheck
365 <                break;
366 <            else if (--k > 0) {
367 <                if ((k & 3) == 1)
368 <                    Thread.yield();               // occasionally yield
369 <            }
370 <            else if (k == 0)
371 <                p.tryPollForAndExec(w, this);     // uncommon self-help case
372 <            else if (p.tryCompensate()) {         // true if can block
373 <                try {
374 <                    int ss = status;
375 <                    if (ss >= 0 &&                // assert need signal
376 <                        U.compareAndSwapInt(this, STATUS, ss, ss | SIGNAL)) {
377 <                        synchronized (this) {
378 <                            if (status >= 0)      // block
379 <                                wait();
380 <                        }
381 <                    }
382 <                } catch (InterruptedException ignore) {
383 <                } finally {
384 <                    p.incrementActiveCount();     // re-activate
385 <                }
339 >            if (((t = Thread.currentThread()) instanceof ForkJoinWorkerThread)) {
340 >                if (!(w = (wt = (ForkJoinWorkerThread)t).workQueue).
341 >                    tryUnpush(this) || (s = doExec()) >= 0)
342 >                    s = wt.pool.awaitJoin(w, this);
343              }
344 +            else
345 +                s = externalAwaitDone();
346          }
388        w.currentJoin = prevJoin;
347          return s;
348      }
349  
# Line 395 | Line 353 | public abstract class ForkJoinTask<V> im
353       * @return status upon completion
354       */
355      private int doInvoke() {
356 <        int s; Thread t;
356 >        int s; Thread t; ForkJoinWorkerThread wt;
357          if ((s = doExec()) >= 0) {
358 <            if (!((t = Thread.currentThread()) instanceof ForkJoinWorkerThread))
358 >            if ((t = Thread.currentThread()) instanceof ForkJoinWorkerThread)
359 >                s = (wt = (ForkJoinWorkerThread)t).pool.awaitJoin(wt.workQueue,
360 >                                                                  this);
361 >            else
362                  s = externalAwaitDone();
402            else {
403                ForkJoinWorkerThread wt = (ForkJoinWorkerThread)t;
404                s = awaitJoin(wt.workQueue, wt.pool);
405            }
363          }
364          return s;
365      }
# Line 539 | Line 496 | public abstract class ForkJoinTask<V> im
496       * @return the exception, or null if none
497       */
498      private Throwable getThrowableException() {
499 <        if (status != EXCEPTIONAL)
499 >        if ((status & DONE_MASK) != EXCEPTIONAL)
500              return null;
501          int h = System.identityHashCode(this);
502          ExceptionNode e;
# Line 624 | Line 581 | public abstract class ForkJoinTask<V> im
581      }
582  
583      /**
584 <     * Report the result of invoke or join; called only upon
628 <     * non-normal return of internal versions.
584 >     * Throws exception, if any, associated with the given status.
585       */
586 <    private V reportResult() {
587 <        int s; Throwable ex;
588 <        if ((s = status) == CANCELLED)
589 <            throw new CancellationException();
590 <        if (s == EXCEPTIONAL && (ex = getThrowableException()) != null)
586 >    private void reportException(int s) {
587 >        Throwable ex = ((s == CANCELLED) ?  new CancellationException() :
588 >                        (s == EXCEPTIONAL) ? getThrowableException() :
589 >                        null);
590 >        if (ex != null)
591              U.throwException(ex);
636        return getRawResult();
592      }
593  
594      // public methods
# Line 657 | Line 612 | public abstract class ForkJoinTask<V> im
612       * @return {@code this}, to simplify usage
613       */
614      public final ForkJoinTask<V> fork() {
615 <        ForkJoinWorkerThread wt;
661 <        (wt = (ForkJoinWorkerThread)Thread.currentThread()).
662 <            workQueue.push(this, wt.pool);
615 >        ((ForkJoinWorkerThread)Thread.currentThread()).workQueue.push(this);
616          return this;
617      }
618  
# Line 675 | Line 628 | public abstract class ForkJoinTask<V> im
628       * @return the computed result
629       */
630      public final V join() {
631 <        if (doJoin() != NORMAL)
632 <            return reportResult();
633 <        else
634 <            return getRawResult();
631 >        int s;
632 >        if ((s = doJoin() & DONE_MASK) != NORMAL)
633 >            reportException(s);
634 >        return getRawResult();
635      }
636  
637      /**
# Line 690 | Line 643 | public abstract class ForkJoinTask<V> im
643       * @return the computed result
644       */
645      public final V invoke() {
646 <        if (doInvoke() != NORMAL)
647 <            return reportResult();
648 <        else
649 <            return getRawResult();
646 >        int s;
647 >        if ((s = doInvoke() & DONE_MASK) != NORMAL)
648 >            reportException(s);
649 >        return getRawResult();
650      }
651  
652      /**
# Line 720 | Line 673 | public abstract class ForkJoinTask<V> im
673       * @throws NullPointerException if any task is null
674       */
675      public static void invokeAll(ForkJoinTask<?> t1, ForkJoinTask<?> t2) {
676 +        int s1, s2;
677          t2.fork();
678 <        t1.invoke();
679 <        t2.join();
678 >        if ((s1 = t1.doInvoke() & DONE_MASK) != NORMAL)
679 >            t1.reportException(s1);
680 >        if ((s2 = t2.doJoin() & DONE_MASK) != NORMAL)
681 >            t2.reportException(s2);
682      }
683  
684      /**
# Line 859 | Line 815 | public abstract class ForkJoinTask<V> im
815       * @return {@code true} if this task is now cancelled
816       */
817      public boolean cancel(boolean mayInterruptIfRunning) {
818 <        return setCompletion(CANCELLED) == CANCELLED;
818 >        return (setCompletion(CANCELLED) & DONE_MASK) == CANCELLED;
819      }
820  
821      public final boolean isDone() {
# Line 867 | Line 823 | public abstract class ForkJoinTask<V> im
823      }
824  
825      public final boolean isCancelled() {
826 <        return status == CANCELLED;
826 >        return (status & DONE_MASK) == CANCELLED;
827      }
828  
829      /**
# Line 887 | Line 843 | public abstract class ForkJoinTask<V> im
843       * exception and was not cancelled
844       */
845      public final boolean isCompletedNormally() {
846 <        return status == NORMAL;
846 >        return (status & DONE_MASK) == NORMAL;
847      }
848  
849      /**
# Line 898 | Line 854 | public abstract class ForkJoinTask<V> im
854       * @return the exception, or {@code null} if none
855       */
856      public final Throwable getException() {
857 <        int s = status;
857 >        int s = status & DONE_MASK;
858          return ((s >= NORMAL)    ? null :
859                  (s == CANCELLED) ? new CancellationException() :
860                  getThrowableException());
# Line 948 | Line 904 | public abstract class ForkJoinTask<V> im
904      }
905  
906      /**
907 +     * Completes this task. The most recent value established by
908 +     * {@link #setRawResult} (or {@code null}) will be returned as the
909 +     * result of subsequent invocations of {@code join} and related
910 +     * operations. This method may be useful when processing sets of
911 +     * tasks when some do not otherwise complete normally. Its use in
912 +     * other situations is discouraged.
913 +     */
914 +    public final void quietlyComplete() {
915 +        setCompletion(NORMAL);
916 +    }
917 +
918 +    /**
919       * Waits if necessary for the computation to complete, and then
920       * retrieves its result.
921       *
# Line 960 | Line 928 | public abstract class ForkJoinTask<V> im
928       */
929      public final V get() throws InterruptedException, ExecutionException {
930          int s = (Thread.currentThread() instanceof ForkJoinWorkerThread) ?
931 <            doJoin() : externalInterruptibleAwaitDone(0L);
931 >            doJoin() : externalInterruptibleAwaitDone();
932          Throwable ex;
933 <        if (s == CANCELLED)
933 >        if ((s &= DONE_MASK) == CANCELLED)
934              throw new CancellationException();
935          if (s == EXCEPTIONAL && (ex = getThrowableException()) != null)
936              throw new ExecutionException(ex);
# Line 985 | Line 953 | public abstract class ForkJoinTask<V> im
953       */
954      public final V get(long timeout, TimeUnit unit)
955          throws InterruptedException, ExecutionException, TimeoutException {
956 <        // Messy in part because we measure in nanos, but wait in millis
957 <        int s; long millis, nanos;
958 <        Thread t = Thread.currentThread();
959 <        if (!(t instanceof ForkJoinWorkerThread)) {
960 <            if ((millis = unit.toMillis(timeout)) > 0L)
961 <                s = externalInterruptibleAwaitDone(millis);
962 <            else
963 <                s = status;
964 <        }
965 <        else if ((s = status) >= 0 && (nanos = unit.toNanos(timeout)) > 0L) {
966 <            long deadline = System.nanoTime() + nanos;
967 <            ForkJoinWorkerThread wt = (ForkJoinWorkerThread)t;
968 <            ForkJoinPool.WorkQueue w = wt.workQueue;
969 <            ForkJoinPool p = wt.pool;
970 <            if (w.tryUnpush(this))
971 <                doExec();
972 <            boolean blocking = false;
956 >        if (Thread.interrupted())
957 >            throw new InterruptedException();
958 >        // Messy in part because we measure in nanosecs, but wait in millisecs
959 >        int s; long ns, ms;
960 >        if ((s = status) >= 0 && (ns = unit.toNanos(timeout)) > 0L) {
961 >            long deadline = System.nanoTime() + ns;
962 >            ForkJoinPool p = null;
963 >            ForkJoinPool.WorkQueue w = null;
964 >            Thread t = Thread.currentThread();
965 >            if (t instanceof ForkJoinWorkerThread) {
966 >                ForkJoinWorkerThread wt = (ForkJoinWorkerThread)t;
967 >                p = wt.pool;
968 >                w = wt.workQueue;
969 >                s = p.helpJoinOnce(w, this); // no retries on failure
970 >            }
971 >            boolean canBlock = false;
972 >            boolean interrupted = false;
973              try {
974                  while ((s = status) >= 0) {
975 <                    if (w.runState < 0)
975 >                    if (w != null && w.runState < 0)
976                          cancelIgnoringExceptions(this);
977 <                    else if (!blocking)
978 <                        blocking = p.tryCompensate();
977 >                    else if (!canBlock) {
978 >                        if (p == null || p.tryCompensate(this, null))
979 >                            canBlock = true;
980 >                    }
981                      else {
982 <                        millis = TimeUnit.NANOSECONDS.toMillis(nanos);
1013 <                        if (millis > 0L &&
982 >                        if ((ms = TimeUnit.NANOSECONDS.toMillis(ns)) > 0L &&
983                              U.compareAndSwapInt(this, STATUS, s, s | SIGNAL)) {
984 <                            try {
985 <                                synchronized (this) {
986 <                                    if (status >= 0)
987 <                                        wait(millis);
984 >                            synchronized (this) {
985 >                                if (status >= 0) {
986 >                                    try {
987 >                                        wait(ms);
988 >                                    } catch (InterruptedException ie) {
989 >                                        if (p == null)
990 >                                            interrupted = true;
991 >                                    }
992                                  }
993 <                            } catch (InterruptedException ie) {
993 >                                else
994 >                                    notifyAll();
995                              }
996                          }
997 <                        if ((s = status) < 0 ||
998 <                            (nanos = deadline - System.nanoTime()) <= 0L)
997 >                        if ((s = status) < 0 || interrupted ||
998 >                            (ns = deadline - System.nanoTime()) <= 0L)
999                              break;
1000                      }
1001                  }
1002              } finally {
1003 <                if (blocking)
1003 >                if (p != null && canBlock)
1004                      p.incrementActiveCount();
1005              }
1006 +            if (interrupted)
1007 +                throw new InterruptedException();
1008          }
1009 <        if (s != NORMAL) {
1009 >        if ((s &= DONE_MASK) != NORMAL) {
1010              Throwable ex;
1011              if (s == CANCELLED)
1012                  throw new CancellationException();
# Line 1097 | Line 1073 | public abstract class ForkJoinTask<V> im
1073       * setRawResult(null)}.
1074       */
1075      public void reinitialize() {
1076 <        if (status == EXCEPTIONAL)
1076 >        if ((status & DONE_MASK) == EXCEPTIONAL)
1077              clearExceptionalCompletion();
1078          else
1079              status = 0;
# Line 1335 | Line 1311 | public abstract class ForkJoinTask<V> im
1311          return wt.pool.nextTaskFor(wt.workQueue);
1312      }
1313  
1314 <    // Mark-bit operations
1314 >    // tag operations
1315  
1316      /**
1317 <     * Returns true if this task is marked.
1317 >     * Returns the tag for this task.
1318       *
1319 <     * @return true if this task is marked
1319 >     * @return the tag for this task
1320       * @since 1.8
1321       */
1322 <    public final boolean isMarkedForkJoinTask() {
1323 <        return (status & MARKED) != 0;
1322 >    public final short getForkJoinTaskTag() {
1323 >        return (short)status;
1324      }
1325  
1326      /**
1327 <     * Atomically sets the mark on this task.
1327 >     * Atomically sets the tag value for this task.
1328       *
1329 <     * @return true if this task was previously unmarked
1329 >     * @param tag the tag value
1330 >     * @return the previous value of the tag
1331       * @since 1.8
1332       */
1333 <    public final boolean markForkJoinTask() {
1333 >    public final short setForkJoinTaskTag(short tag) {
1334          for (int s;;) {
1335 <            if (((s = status) & MARKED) != 0)
1336 <                return false;
1337 <            if (U.compareAndSwapInt(this, STATUS, s, s | MARKED))
1361 <                return true;
1335 >            if (U.compareAndSwapInt(this, STATUS, s = status,
1336 >                                    (s & ~SMASK) | (tag & SMASK)))
1337 >                return (short)s;
1338          }
1339      }
1340  
1341      /**
1342 <     * Atomically clears the mark on this task.
1342 >     * Atomically conditionally sets the tag value for this task.
1343 >     * Among other applications, tags can be used as visit markers
1344 >     * in tasks operating on graphs, as in mathods that check: {@code
1345 >     * if (task.compareAndSetForkJoinTaskTag((short)0, (short)1))}
1346 >     * before processing, otherwise exiting because the node has
1347 >     * already been visited.
1348       *
1349 <     * @return true if this task was previously marked
1349 >     * @param e the expected tag value
1350 >     * @param tag the new tag value
1351 >     * @return true if successful; i.e., the current value was
1352 >     * equal to e and is now tag.
1353       * @since 1.8
1354       */
1355 <    public final boolean unmarkForkJoinTask() {
1355 >    public final boolean compareAndSetForkJoinTaskTag(short e, short tag) {
1356          for (int s;;) {
1357 <            if (((s = status) & MARKED) == 0)
1357 >            if ((short)(s = status) != e)
1358                  return false;
1359 <            if (U.compareAndSwapInt(this, STATUS, s, s & ~MARKED))
1359 >            if (U.compareAndSwapInt(this, STATUS, s,
1360 >                                    (s & ~SMASK) | (tag & SMASK)))
1361                  return true;
1362          }
1363      }
# Line 1385 | Line 1370 | public abstract class ForkJoinTask<V> im
1370      static final class AdaptedRunnable<T> extends ForkJoinTask<T>
1371          implements RunnableFuture<T> {
1372          final Runnable runnable;
1388        final T resultOnCompletion;
1373          T result;
1374          AdaptedRunnable(Runnable runnable, T result) {
1375              if (runnable == null) throw new NullPointerException();
1376              this.runnable = runnable;
1377 <            this.resultOnCompletion = result;
1377 >            this.result = result; // OK to set this even before completion
1378          }
1379 <        public T getRawResult() { return result; }
1380 <        public void setRawResult(T v) { result = v; }
1381 <        public boolean exec() {
1382 <            runnable.run();
1383 <            result = resultOnCompletion;
1384 <            return true;
1379 >        public final T getRawResult() { return result; }
1380 >        public final void setRawResult(T v) { result = v; }
1381 >        public final boolean exec() { runnable.run(); return true; }
1382 >        public final void run() { invoke(); }
1383 >        private static final long serialVersionUID = 5232453952276885070L;
1384 >    }
1385 >
1386 >    /**
1387 >     * Adaptor for Runnables without results
1388 >     */
1389 >    static final class AdaptedRunnableAction extends ForkJoinTask<Void>
1390 >        implements RunnableFuture<Void> {
1391 >        final Runnable runnable;
1392 >        AdaptedRunnableAction(Runnable runnable) {
1393 >            if (runnable == null) throw new NullPointerException();
1394 >            this.runnable = runnable;
1395          }
1396 <        public void run() { invoke(); }
1396 >        public final Void getRawResult() { return null; }
1397 >        public final void setRawResult(Void v) { }
1398 >        public final boolean exec() { runnable.run(); return true; }
1399 >        public final void run() { invoke(); }
1400          private static final long serialVersionUID = 5232453952276885070L;
1401      }
1402  
# Line 1414 | Line 1411 | public abstract class ForkJoinTask<V> im
1411              if (callable == null) throw new NullPointerException();
1412              this.callable = callable;
1413          }
1414 <        public T getRawResult() { return result; }
1415 <        public void setRawResult(T v) { result = v; }
1416 <        public boolean exec() {
1414 >        public final T getRawResult() { return result; }
1415 >        public final void setRawResult(T v) { result = v; }
1416 >        public final boolean exec() {
1417              try {
1418                  result = callable.call();
1419                  return true;
# Line 1428 | Line 1425 | public abstract class ForkJoinTask<V> im
1425                  throw new RuntimeException(ex);
1426              }
1427          }
1428 <        public void run() { invoke(); }
1428 >        public final void run() { invoke(); }
1429          private static final long serialVersionUID = 2838392045355241008L;
1430      }
1431  
# Line 1441 | Line 1438 | public abstract class ForkJoinTask<V> im
1438       * @return the task
1439       */
1440      public static ForkJoinTask<?> adapt(Runnable runnable) {
1441 <        return new AdaptedRunnable<Void>(runnable, null);
1441 >        return new AdaptedRunnableAction(runnable);
1442      }
1443  
1444      /**

Diff Legend

Removed lines
+ Added lines
< Changed lines
> Changed lines