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.66 by dl, Sun Oct 24 19:37:26 2010 UTC vs.
Revision 1.70 by dl, Tue Nov 23 00:10:39 2010 UTC

# Line 110 | Line 110 | import java.util.concurrent.TimeoutExcep
110   * result in exceptions or errors, possibly including
111   * {@code ClassCastException}.
112   *
113 + * <p>Method {@link #join} and its variants are appropriate for use
114 + * only when completion dependencies are acyclic; that is, the
115 + * parallel computation can be described as a directed acyclic graph
116 + * (DAG). Otherwise, executions may encounter a form of deadlock as
117 + * tasks cyclically wait for each other.  However, this framework
118 + * supports other methods and techniques (for example the use of
119 + * {@link Phaser}, {@link #helpQuiesce}, and {@link #complete}) that
120 + * may be of use in constructing custom subclasses for problems that
121 + * are not statically structured as DAGs.
122 + *
123   * <p>Most base support methods are {@code final}, to prevent
124   * overriding of implementations that are intrinsically tied to the
125   * underlying lightweight task scheduling framework.  Developers
# Line 124 | Line 134 | import java.util.concurrent.TimeoutExcep
134   * computation. Large tasks should be split into smaller subtasks,
135   * usually via recursive decomposition. As a very rough rule of thumb,
136   * a task should perform more than 100 and less than 10000 basic
137 < * computational steps. If tasks are too big, then parallelism cannot
138 < * improve throughput. If too small, then memory and internal task
139 < * maintenance overhead may overwhelm processing.
137 > * computational steps, and should avoid indefinite looping. If tasks
138 > * are too big, then parallelism cannot improve throughput. If too
139 > * small, then memory and internal task maintenance overhead may
140 > * overwhelm processing.
141   *
142   * <p>This class provides {@code adapt} methods for {@link Runnable}
143   * and {@link Callable}, that may be of use when mixing execution of
# Line 223 | Line 234 | public abstract class ForkJoinTask<V> im
234      }
235  
236      /**
237 <     * Blocks a worker thread until completion. Called only by
238 <     * pool. Currently unused -- pool-based waits use timeout
228 <     * version below.
237 >     * Blocks a worker thread until completed or timed out.  Called
238 >     * only by pool.
239       */
240 <    final void internalAwaitDone() {
241 <        int s;         // the odd construction reduces lock bias effects
242 <        while ((s = status) >= 0) {
233 <            try {
240 >    final void internalAwaitDone(long millis, int nanos) {
241 >        if (status >= 0) {
242 >            try {     // the odd construction reduces lock bias effects
243                  synchronized (this) {
244 <                    if (UNSAFE.compareAndSwapInt(this, statusOffset, s,SIGNAL))
245 <                        wait();
244 >                    if (status > 0 ||
245 >                        UNSAFE.compareAndSwapInt(this, statusOffset,
246 >                                                 0, SIGNAL))
247 >                        wait(millis, nanos);
248                  }
249              } catch (InterruptedException ie) {
250                  cancelIfTerminating();
# Line 242 | Line 253 | public abstract class ForkJoinTask<V> im
253      }
254  
255      /**
256 <     * Blocks a worker thread until completed or timed out.  Called
246 <     * only by pool.
247 <     *
248 <     * @return status on exit
256 >     * Blocks a non-worker-thread until completion.
257       */
258 <    final int internalAwaitDone(long millis, int nanos) {
259 <        int s;
260 <        if ((s = status) >= 0) {
261 <            try {
262 <                synchronized (this) {
263 <                    if (UNSAFE.compareAndSwapInt(this, statusOffset, s,SIGNAL))
264 <                        wait(millis, nanos);
258 >    private void externalAwaitDone() {
259 >        if (status >= 0) {
260 >            boolean interrupted = false;
261 >            synchronized(this) {
262 >                int s;
263 >                while ((s = status) >= 0) {
264 >                    if (s == 0 &&
265 >                        !UNSAFE.compareAndSwapInt(this, statusOffset,
266 >                                                  0, SIGNAL))
267 >                        continue;
268 >                    try {
269 >                        wait();
270 >                    } catch (InterruptedException ie) {
271 >                        interrupted = true;
272 >                    }
273                  }
258            } catch (InterruptedException ie) {
259                cancelIfTerminating();
274              }
275 <            s = status;
275 >            if (interrupted)
276 >                Thread.currentThread().interrupt();
277          }
263        return s;
278      }
279  
280      /**
281 <     * Blocks a non-worker-thread until completion.
281 >     * Blocks a non-worker-thread until completion or interruption or timeout
282       */
283 <    private void externalAwaitDone() {
284 <        int s;
285 <        while ((s = status) >= 0) {
286 <            synchronized (this) {
287 <                if (UNSAFE.compareAndSwapInt(this, statusOffset, s, SIGNAL)) {
288 <                    boolean interrupted = false;
289 <                    while (status >= 0) {
290 <                        try {
291 <                            wait();
292 <                        } catch (InterruptedException ie) {
293 <                            interrupted = true;
294 <                        }
295 <                    }
296 <                    if (interrupted)
297 <                        Thread.currentThread().interrupt();
298 <                    break;
283 >    private void externalInterruptibleAwaitDone(boolean timed, long nanos)
284 >        throws InterruptedException {
285 >        if (Thread.interrupted())
286 >            throw new InterruptedException();
287 >        if (status >= 0) {
288 >            long startTime = timed ? System.nanoTime() : 0L;
289 >            synchronized(this) {
290 >                int s;
291 >                while ((s = status) >= 0) {
292 >                    long nt;
293 >                    if (s == 0 &&
294 >                        !UNSAFE.compareAndSwapInt(this, statusOffset,
295 >                                                  0, SIGNAL))
296 >                        continue;
297 >                    else if (!timed)
298 >                        wait();
299 >                    else if ((nt = nanos - (System.nanoTime()-startTime)) > 0L)
300 >                        wait(nt / 1000000, (int)(nt % 1000000));
301 >                    else
302 >                        break;
303                  }
304              }
305          }
# Line 316 | Line 334 | public abstract class ForkJoinTask<V> im
334       * #isDone} returning {@code true}.
335       *
336       * <p>This method may be invoked only from within {@code
337 <     * ForkJoinTask} computations (as may be determined using method
337 >     * ForkJoinPool} computations (as may be determined using method
338       * {@link #inForkJoinPool}).  Attempts to invoke in other contexts
339       * result in exceptions or errors, possibly including {@code
340       * ClassCastException}.
# Line 330 | Line 348 | public abstract class ForkJoinTask<V> im
348      }
349  
350      /**
351 <     * Returns the result of the computation when it {@link #isDone is done}.
352 <     * This method differs from {@link #get()} in that
351 >     * Returns the result of the computation when it {@link #isDone is
352 >     * done}.  This method differs from {@link #get()} in that
353       * abnormal completion results in {@code RuntimeException} or
354 <     * {@code Error}, not {@code ExecutionException}.
354 >     * {@code Error}, not {@code ExecutionException}, and that
355 >     * interrupts of the calling thread do <em>not</em> cause the
356 >     * method to abruptly return by throwing {@code
357 >     * InterruptedException}.
358       *
359       * @return the computed result
360       */
# Line 375 | Line 396 | public abstract class ForkJoinTask<V> im
396       * unprocessed.
397       *
398       * <p>This method may be invoked only from within {@code
399 <     * ForkJoinTask} computations (as may be determined using method
399 >     * ForkJoinPool} computations (as may be determined using method
400       * {@link #inForkJoinPool}).  Attempts to invoke in other contexts
401       * result in exceptions or errors, possibly including {@code
402       * ClassCastException}.
# Line 403 | Line 424 | public abstract class ForkJoinTask<V> im
424       * normally or exceptionally, or left unprocessed.
425       *
426       * <p>This method may be invoked only from within {@code
427 <     * ForkJoinTask} computations (as may be determined using method
427 >     * ForkJoinPool} computations (as may be determined using method
428       * {@link #inForkJoinPool}).  Attempts to invoke in other contexts
429       * result in exceptions or errors, possibly including {@code
430       * ClassCastException}.
# Line 458 | Line 479 | public abstract class ForkJoinTask<V> im
479       * unprocessed.
480       *
481       * <p>This method may be invoked only from within {@code
482 <     * ForkJoinTask} computations (as may be determined using method
482 >     * ForkJoinPool} computations (as may be determined using method
483       * {@link #inForkJoinPool}).  Attempts to invoke in other contexts
484       * result in exceptions or errors, possibly including {@code
485       * ClassCastException}.
# Line 510 | Line 531 | public abstract class ForkJoinTask<V> im
531  
532      /**
533       * Attempts to cancel execution of this task. This attempt will
534 <     * fail if the task has already completed, has already been
535 <     * cancelled, or could not be cancelled for some other reason. If
536 <     * successful, and this task has not started when cancel is
537 <     * called, execution of this task is suppressed, {@link
538 <     * #isCancelled} will report true, and {@link #join} will result
539 <     * in a {@code CancellationException} being thrown.
534 >     * fail if the task has already completed or could not be
535 >     * cancelled for some other reason. If successful, and this task
536 >     * has not started when {@code cancel} is called, execution of
537 >     * this task is suppressed. After this method returns
538 >     * successfully, unless there is an intervening call to {@link
539 >     * #reinitialize}, subsequent calls to {@link #isCancelled},
540 >     * {@link #isDone}, and {@code cancel} will return {@code true}
541 >     * and calls to {@link #join} and related methods will result in
542 >     * {@code CancellationException}.
543       *
544       * <p>This method may be overridden in subclasses, but if so, must
545 <     * still ensure that these minimal properties hold. In particular,
546 <     * the {@code cancel} method itself must not throw exceptions.
545 >     * still ensure that these properties hold. In particular, the
546 >     * {@code cancel} method itself must not throw exceptions.
547       *
548       * <p>This method is designed to be invoked by <em>other</em>
549       * tasks. To terminate the current task, you can just return or
550       * throw an unchecked exception from its computation method, or
551       * invoke {@link #completeExceptionally}.
552       *
553 <     * @param mayInterruptIfRunning this value is ignored in the
554 <     * default implementation because tasks are not
555 <     * cancelled via interruption
553 >     * @param mayInterruptIfRunning this value has no effect in the
554 >     * default implementation because interrupts are not used to
555 >     * control cancellation.
556       *
557       * @return {@code true} if this task is now cancelled
558       */
# Line 662 | Line 686 | public abstract class ForkJoinTask<V> im
686       * member of a ForkJoinPool and was interrupted while waiting
687       */
688      public final V get() throws InterruptedException, ExecutionException {
689 <        int s;
690 <        if (Thread.currentThread() instanceof ForkJoinWorkerThread) {
689 >        Thread t = Thread.currentThread();
690 >        if (t instanceof ForkJoinWorkerThread)
691              quietlyJoin();
692 <            s = status;
693 <        }
694 <        else {
695 <            while ((s = status) >= 0) {
672 <                synchronized (this) { // interruptible form of awaitDone
673 <                    if (UNSAFE.compareAndSwapInt(this, statusOffset,
674 <                                                 s, SIGNAL)) {
675 <                        while (status >= 0)
676 <                            wait();
677 <                    }
678 <                }
679 <            }
680 <        }
681 <        if (s < NORMAL) {
692 >        else
693 >            externalInterruptibleAwaitDone(false, 0L);
694 >        int s = status;
695 >        if (s != NORMAL) {
696              Throwable ex;
697              if (s == CANCELLED)
698                  throw new CancellationException();
# Line 705 | Line 719 | public abstract class ForkJoinTask<V> im
719      public final V get(long timeout, TimeUnit unit)
720          throws InterruptedException, ExecutionException, TimeoutException {
721          long nanos = unit.toNanos(timeout);
722 <        if (status >= 0) {
723 <            Thread t = Thread.currentThread();
724 <            if (t instanceof ForkJoinWorkerThread) {
725 <                ForkJoinWorkerThread w = (ForkJoinWorkerThread) t;
726 <                boolean completed = false; // timed variant of quietlyJoin
727 <                if (w.unpushTask(this)) {
728 <                    try {
715 <                        completed = exec();
716 <                    } catch (Throwable rex) {
717 <                        setExceptionalCompletion(rex);
718 <                    }
719 <                }
720 <                if (completed)
721 <                    setCompletion(NORMAL);
722 <                else if (status >= 0)
723 <                    w.joinTask(this, true, nanos);
724 <            }
725 <            else if (Thread.interrupted())
726 <                throw new InterruptedException();
727 <            else {
728 <                long startTime = System.nanoTime();
729 <                int s; long nt;
730 <                while ((s = status) >= 0 &&
731 <                       (nt = nanos - (System.nanoTime() - startTime)) > 0) {
732 <                    if (UNSAFE.compareAndSwapInt(this, statusOffset, s,
733 <                                                 SIGNAL)) {
734 <                        long ms = nt / 1000000;
735 <                        int ns = (int) (nt % 1000000);
736 <                        synchronized (this) {
737 <                            if (status >= 0)
738 <                                wait(ms, ns); // exit on IE throw
739 <                        }
740 <                    }
741 <                }
742 <            }
743 <        }
744 <        int es = status;
745 <        if (es != NORMAL) {
722 >        Thread t = Thread.currentThread();
723 >        if (t instanceof ForkJoinWorkerThread)
724 >            ((ForkJoinWorkerThread)t).joinTask(this, true, nanos);
725 >        else
726 >            externalInterruptibleAwaitDone(true, nanos);
727 >        int s = status;
728 >        if (s != NORMAL) {
729              Throwable ex;
730 <            if (es == CANCELLED)
730 >            if (s == CANCELLED)
731                  throw new CancellationException();
732 <            if (es == EXCEPTIONAL && (ex = exceptionMap.get(this)) != null)
732 >            if (s == EXCEPTIONAL && (ex = exceptionMap.get(this)) != null)
733                  throw new ExecutionException(ex);
734              throw new TimeoutException();
735          }
# Line 813 | Line 796 | public abstract class ForkJoinTask<V> im
796       * processed.
797       *
798       * <p>This method may be invoked only from within {@code
799 <     * ForkJoinTask} computations (as may be determined using method
799 >     * ForkJoinPool} computations (as may be determined using method
800       * {@link #inForkJoinPool}).  Attempts to invoke in other contexts
801       * result in exceptions or errors, possibly including {@code
802       * ClassCastException}.
# Line 832 | Line 815 | public abstract class ForkJoinTask<V> im
815       * under any other usage conditions are not guaranteed.
816       * This method may be useful when executing
817       * pre-constructed trees of subtasks in loops.
818 +     *
819 +     * <p>Upon completion of this method, {@code isDone()} reports
820 +     * {@code false}, and {@code getException()} reports {@code
821 +     * null}. However, the value returned by {@code getRawResult} is
822 +     * unaffected. To clear this value, you can invoke {@code
823 +     * setRawResult(null)}.
824       */
825      public void reinitialize() {
826          if (status == EXCEPTIONAL)
# Line 872 | Line 861 | public abstract class ForkJoinTask<V> im
861       * were not, stolen.
862       *
863       * <p>This method may be invoked only from within {@code
864 <     * ForkJoinTask} computations (as may be determined using method
864 >     * ForkJoinPool} computations (as may be determined using method
865       * {@link #inForkJoinPool}).  Attempts to invoke in other contexts
866       * result in exceptions or errors, possibly including {@code
867       * ClassCastException}.
# Line 891 | Line 880 | public abstract class ForkJoinTask<V> im
880       * fork other tasks.
881       *
882       * <p>This method may be invoked only from within {@code
883 <     * ForkJoinTask} computations (as may be determined using method
883 >     * ForkJoinPool} computations (as may be determined using method
884       * {@link #inForkJoinPool}).  Attempts to invoke in other contexts
885       * result in exceptions or errors, possibly including {@code
886       * ClassCastException}.
# Line 914 | Line 903 | public abstract class ForkJoinTask<V> im
903       * exceeded.
904       *
905       * <p>This method may be invoked only from within {@code
906 <     * ForkJoinTask} computations (as may be determined using method
906 >     * ForkJoinPool} computations (as may be determined using method
907       * {@link #inForkJoinPool}).  Attempts to invoke in other contexts
908       * result in exceptions or errors, possibly including {@code
909       * ClassCastException}.
# Line 972 | Line 961 | public abstract class ForkJoinTask<V> im
961       * otherwise.
962       *
963       * <p>This method may be invoked only from within {@code
964 <     * ForkJoinTask} computations (as may be determined using method
964 >     * ForkJoinPool} computations (as may be determined using method
965       * {@link #inForkJoinPool}).  Attempts to invoke in other contexts
966       * result in exceptions or errors, possibly including {@code
967       * ClassCastException}.
# Line 991 | Line 980 | public abstract class ForkJoinTask<V> im
980       * be useful otherwise.
981       *
982       * <p>This method may be invoked only from within {@code
983 <     * ForkJoinTask} computations (as may be determined using method
983 >     * ForkJoinPool} computations (as may be determined using method
984       * {@link #inForkJoinPool}).  Attempts to invoke in other contexts
985       * result in exceptions or errors, possibly including {@code
986       * ClassCastException}.
# Line 1014 | Line 1003 | public abstract class ForkJoinTask<V> im
1003       * otherwise.
1004       *
1005       * <p>This method may be invoked only from within {@code
1006 <     * ForkJoinTask} computations (as may be determined using method
1006 >     * ForkJoinPool} computations (as may be determined using method
1007       * {@link #inForkJoinPool}).  Attempts to invoke in other contexts
1008       * result in exceptions or errors, possibly including {@code
1009       * ClassCastException}.

Diff Legend

Removed lines
+ Added lines
< Changed lines
> Changed lines