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

Comparing jsr166/src/jsr166y/ForkJoinPool.java (file contents):
Revision 1.28 by jsr166, Mon Jul 27 20:57:44 2009 UTC vs.
Revision 1.33 by dl, Fri Jul 31 16:27:08 2009 UTC

# Line 20 | Line 20 | import java.util.concurrent.atomic.Atomi
20   import java.util.concurrent.atomic.AtomicLong;
21  
22   /**
23 < * An {@link ExecutorService} for running {@link ForkJoinTask}s.  A
24 < * ForkJoinPool provides the entry point for submissions from
23 > * An {@link ExecutorService} for running {@link ForkJoinTask}s.
24 > * A ForkJoinPool provides the entry point for submissions from
25   * non-ForkJoinTasks, as well as management and monitoring operations.
26   * Normally a single ForkJoinPool is used for a large number of
27   * submitted tasks. Otherwise, use would not usually outweigh the
# Line 34 | Line 34 | import java.util.concurrent.atomic.Atomi
34   * (eventually blocking if none exist). This makes them efficient when
35   * most tasks spawn other subtasks (as do most ForkJoinTasks), as well
36   * as the mixed execution of some plain Runnable- or Callable- based
37 < * activities along with ForkJoinTasks. When setting
38 < * {@code setAsyncMode}, a ForkJoinPools may also be appropriate for
39 < * use with fine-grained tasks that are never joined. Otherwise, other
40 < * ExecutorService implementations are typically more appropriate
41 < * choices.
37 > * activities along with ForkJoinTasks. When setting {@linkplain
38 > * #setAsyncMode async mode}, a ForkJoinPool may also be appropriate
39 > * for use with fine-grained tasks that are never joined. Otherwise,
40 > * other ExecutorService implementations are typically more
41 > * appropriate choices.
42   *
43   * <p>A ForkJoinPool may be constructed with a given parallelism level
44   * (target pool size), which it attempts to maintain by dynamically
45   * adding, suspending, or resuming threads, even if some tasks are
46   * waiting to join others. However, no such adjustments are performed
47   * in the face of blocked IO or other unmanaged synchronization. The
48 < * nested {@code ManagedBlocker} interface enables extension of
48 > * nested {@link ManagedBlocker} interface enables extension of
49   * the kinds of synchronization accommodated.  The target parallelism
50 < * level may also be changed dynamically ({@code setParallelism})
50 > * level may also be changed dynamically ({@link #setParallelism})
51   * and thread construction can be limited using methods
52 < * {@code setMaximumPoolSize} and/or
53 < * {@code setMaintainsParallelism}.
52 > * {@link #setMaximumPoolSize} and/or
53 > * {@link #setMaintainsParallelism}.
54   *
55   * <p>In addition to execution and lifecycle control methods, this
56   * class provides status check methods (for example
57 < * {@code getStealCount}) that are intended to aid in developing,
57 > * {@link #getStealCount}) that are intended to aid in developing,
58   * tuning, and monitoring fork/join applications. Also, method
59 < * {@code toString} returns indications of pool state in a
59 > * {@link #toString} returns indications of pool state in a
60   * convenient form for informal monitoring.
61   *
62   * <p><b>Implementation notes</b>: This implementation restricts the
# Line 589 | Line 589 | public class ForkJoinPool extends Abstra
589          if (task instanceof ForkJoinTask<?>) // avoid re-wrap
590              job = (ForkJoinTask<?>) task;
591          else
592 <            job = new AdaptedRunnable<Void>(task, null);
592 >            job = ForkJoinTask.adapt(task, null);
593          doSubmit(job);
594      }
595  
596      public <T> ForkJoinTask<T> submit(Callable<T> task) {
597 <        ForkJoinTask<T> job = new AdaptedCallable<T>(task);
597 >        ForkJoinTask<T> job = ForkJoinTask.adapt(task);
598          doSubmit(job);
599          return job;
600      }
601  
602      public <T> ForkJoinTask<T> submit(Runnable task, T result) {
603 <        ForkJoinTask<T> job = new AdaptedRunnable<T>(task, result);
603 >        ForkJoinTask<T> job = ForkJoinTask.adapt(task, result);
604          doSubmit(job);
605          return job;
606      }
# Line 610 | Line 610 | public class ForkJoinPool extends Abstra
610          if (task instanceof ForkJoinTask<?>) // avoid re-wrap
611              job = (ForkJoinTask<?>) task;
612          else
613 <            job = new AdaptedRunnable<Void>(task, null);
613 >            job = ForkJoinTask.adapt(task, null);
614          doSubmit(job);
615          return job;
616      }
# Line 629 | Line 629 | public class ForkJoinPool extends Abstra
629          return task;
630      }
631  
632 –    /**
633 –     * Adaptor for Runnables. This implements RunnableFuture
634 –     * to be compliant with AbstractExecutorService constraints.
635 –     */
636 –    static final class AdaptedRunnable<T> extends ForkJoinTask<T>
637 –        implements RunnableFuture<T> {
638 –        final Runnable runnable;
639 –        final T resultOnCompletion;
640 –        T result;
641 –        AdaptedRunnable(Runnable runnable, T result) {
642 –            if (runnable == null) throw new NullPointerException();
643 –            this.runnable = runnable;
644 –            this.resultOnCompletion = result;
645 –        }
646 –        public T getRawResult() { return result; }
647 –        public void setRawResult(T v) { result = v; }
648 –        public boolean exec() {
649 –            runnable.run();
650 –            result = resultOnCompletion;
651 –            return true;
652 –        }
653 –        public void run() { invoke(); }
654 –        private static final long serialVersionUID = 5232453952276885070L;
655 –    }
656 –
657 –    /**
658 –     * Adaptor for Callables
659 –     */
660 –    static final class AdaptedCallable<T> extends ForkJoinTask<T>
661 –        implements RunnableFuture<T> {
662 –        final Callable<T> callable;
663 –        T result;
664 –        AdaptedCallable(Callable<T> callable) {
665 –            if (callable == null) throw new NullPointerException();
666 –            this.callable = callable;
667 –        }
668 –        public T getRawResult() { return result; }
669 –        public void setRawResult(T v) { result = v; }
670 –        public boolean exec() {
671 –            try {
672 –                result = callable.call();
673 –                return true;
674 –            } catch (Error err) {
675 –                throw err;
676 –            } catch (RuntimeException rex) {
677 –                throw rex;
678 –            } catch (Exception ex) {
679 –                throw new RuntimeException(ex);
680 –            }
681 –        }
682 –        public void run() { invoke(); }
683 –        private static final long serialVersionUID = 2838392045355241008L;
684 –    }
632  
633      public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks) {
634          ArrayList<ForkJoinTask<T>> forkJoinTasks =
635              new ArrayList<ForkJoinTask<T>>(tasks.size());
636          for (Callable<T> task : tasks)
637 <            forkJoinTasks.add(new AdaptedCallable<T>(task));
637 >            forkJoinTasks.add(ForkJoinTask.adapt(task));
638          invoke(new InvokeAll<T>(forkJoinTasks));
639  
640          @SuppressWarnings({"unchecked", "rawtypes"})
# Line 815 | Line 762 | public class ForkJoinPool extends Abstra
762      /**
763       * Returns the number of worker threads that have started but not
764       * yet terminated.  This result returned by this method may differ
765 <     * from {@code getParallelism} when threads are created to
765 >     * from {@link #getParallelism} when threads are created to
766       * maintain parallelism when others are cooperatively blocked.
767       *
768       * @return the number of worker threads
# Line 877 | Line 824 | public class ForkJoinPool extends Abstra
824       * tasks that are never joined. This mode may be more appropriate
825       * than default locally stack-based mode in applications in which
826       * worker threads only process asynchronous tasks.  This method is
827 <     * designed to be invoked only when pool is quiescent, and
827 >     * designed to be invoked only when the pool is quiescent, and
828       * typically only before any tasks are submitted. The effects of
829       * invocations at other times may be unpredictable.
830       *
831 <     * @param async if true, use locally FIFO scheduling
831 >     * @param async if {@code true}, use locally FIFO scheduling
832       * @return the previous mode
833 +     * @see #getAsyncMode
834       */
835      public boolean setAsyncMode(boolean async) {
836          boolean oldMode = locallyFifo;
# Line 903 | Line 851 | public class ForkJoinPool extends Abstra
851       * scheduling mode for forked tasks that are never joined.
852       *
853       * @return {@code true} if this pool uses async mode
854 +     * @see #setAsyncMode
855       */
856      public boolean getAsyncMode() {
857          return locallyFifo;
# Line 1054 | Line 1003 | public class ForkJoinPool extends Abstra
1003       * @param c the collection to transfer elements into
1004       * @return the number of elements transferred
1005       */
1006 <    protected int drainTasksTo(Collection<ForkJoinTask<?>> c) {
1006 >    protected int drainTasksTo(Collection<? super ForkJoinTask<?>> c) {
1007          int n = submissionQueue.drainTo(c);
1008          ForkJoinWorkerThread[] ws = workers;
1009          if (ws != null) {
# Line 1120 | Line 1069 | public class ForkJoinPool extends Abstra
1069      public void shutdown() {
1070          checkPermission();
1071          transitionRunStateTo(SHUTDOWN);
1072 <        if (canTerminateOnShutdown(runControl))
1072 >        if (canTerminateOnShutdown(runControl)) {
1073 >            if (workers == null) { // shutting down before workers created
1074 >                final ReentrantLock lock = this.workerLock;
1075 >                lock.lock();
1076 >                try {
1077 >                    if (workers == null) {
1078 >                        terminate();
1079 >                        transitionRunStateTo(TERMINATED);
1080 >                        termination.signalAll();
1081 >                    }
1082 >                } finally {
1083 >                    lock.unlock();
1084 >                }
1085 >            }
1086              terminateOnShutdown();
1087 +        }
1088      }
1089  
1090      /**
# Line 1131 | Line 1094 | public class ForkJoinPool extends Abstra
1094       * method may or may not be rejected. Unlike some other executors,
1095       * this method cancels rather than collects non-executed tasks
1096       * upon termination, so always returns an empty list. However, you
1097 <     * can use method {@code drainTasksTo} before invoking this
1097 >     * can use method {@link #drainTasksTo} before invoking this
1098       * method to transfer unexecuted tasks to another collection.
1099       *
1100       * @return an empty list
# Line 1865 | Line 1828 | public class ForkJoinPool extends Abstra
1828          do {} while (!blocker.isReleasable() && !blocker.block());
1829      }
1830  
1831 <    // AbstractExecutorService overrides
1831 >    // AbstractExecutorService overrides.  These rely on undocumented
1832 >    // fact that ForkJoinTask.adapt returns ForkJoinTasks that also
1833 >    // implement RunnableFuture.
1834  
1835      protected <T> RunnableFuture<T> newTaskFor(Runnable runnable, T value) {
1836 <        return new AdaptedRunnable<T>(runnable, value);
1836 >        return (RunnableFuture<T>)ForkJoinTask.adapt(runnable, value);
1837      }
1838  
1839      protected <T> RunnableFuture<T> newTaskFor(Callable<T> callable) {
1840 <        return new AdaptedCallable<T>(callable);
1840 >        return (RunnableFuture<T>)ForkJoinTask.adapt(callable);
1841      }
1842  
1843      // Unsafe mechanics

Diff Legend

– Removed lines
+ Added lines
< Changed lines
> Changed lines