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.31 by dl, Wed Jul 29 22:48:54 2009 UTC vs.
Revision 1.38 by jsr166, Sun Aug 2 17:02:06 2009 UTC

# Line 81 | Line 81 | public class ForkJoinPool extends Abstra
81      private static final int MAX_THREADS =  0x7FFF;
82  
83      /**
84 <     * Factory for creating new ForkJoinWorkerThreads.  A
85 <     * ForkJoinWorkerThreadFactory must be defined and used for
86 <     * ForkJoinWorkerThread subclasses that extend base functionality
87 <     * or initialize threads with different contexts.
84 >     * Factory for creating new {@link ForkJoinWorkerThread}s.
85 >     * A {@code ForkJoinWorkerThreadFactory} must be defined and used
86 >     * for {@code ForkJoinWorkerThread} subclasses that extend base
87 >     * functionality or initialize threads with different contexts.
88       */
89      public static interface ForkJoinWorkerThreadFactory {
90          /**
# Line 578 | Line 578 | public class ForkJoinPool extends Abstra
578       * @throws NullPointerException if task is null
579       * @throws RejectedExecutionException if pool is shut down
580       */
581 <    public <T> void execute(ForkJoinTask<T> task) {
581 >    public void execute(ForkJoinTask<?> task) {
582          doSubmit(task);
583      }
584  
# 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 840 | Line 787 | public class ForkJoinPool extends Abstra
787       * Setting this value has no effect on current pool size. It
788       * controls construction of new threads.
789       *
790 <     * @throws IllegalArgumentException if negative or greater then
790 >     * @throws IllegalArgumentException if negative or greater than
791       * internal implementation limit
792       */
793      public void setMaximumPoolSize(int newMax) {
# Line 1132 | Line 1079 | public class ForkJoinPool extends Abstra
1079                          transitionRunStateTo(TERMINATED);
1080                          termination.signalAll();
1081                      }
1135 –                    
1082                  } finally {
1083                      lock.unlock();
1084                  }
# Line 1789 | Line 1735 | public class ForkJoinPool extends Abstra
1735  
1736      /**
1737       * Interface for extending managed parallelism for tasks running
1738 <     * in ForkJoinPools. A ManagedBlocker provides two methods.
1738 >     * in {@link ForkJoinPool}s.
1739 >     *
1740 >     * <p>A {@code ManagedBlocker} provides two methods.
1741       * Method {@code isReleasable} must return {@code true} if
1742       * blocking is not necessary. Method {@code block} blocks the
1743       * current thread if necessary (perhaps internally invoking
# Line 1832 | Line 1780 | public class ForkJoinPool extends Abstra
1780  
1781      /**
1782       * Blocks in accord with the given blocker.  If the current thread
1783 <     * is a ForkJoinWorkerThread, this method possibly arranges for a
1784 <     * spare thread to be activated if necessary to ensure parallelism
1785 <     * while the current thread is blocked.  If
1786 <     * {@code maintainParallelism} is {@code true} and the pool supports
1787 <     * it ({@link #getMaintainsParallelism}), this method attempts to
1788 <     * maintain the pool's nominal parallelism. Otherwise it activates
1789 <     * a thread only if necessary to avoid complete starvation. This
1790 <     * option may be preferable when blockages use timeouts, or are
1791 <     * almost always brief.
1783 >     * is a {@link ForkJoinWorkerThread}, this method possibly
1784 >     * arranges for a spare thread to be activated if necessary to
1785 >     * ensure parallelism while the current thread is blocked.
1786 >     *
1787 >     * <p>If {@code maintainParallelism} is {@code true} and the pool
1788 >     * supports it ({@link #getMaintainsParallelism}), this method
1789 >     * attempts to maintain the pool's nominal parallelism. Otherwise
1790 >     * it activates a thread only if necessary to avoid complete
1791 >     * starvation. This option may be preferable when blockages use
1792 >     * timeouts, or are almost always brief.
1793       *
1794 <     * <p> If the caller is not a ForkJoinTask, this method is behaviorally
1795 <     * equivalent to
1794 >     * <p>If the caller is not a {@link ForkJoinTask}, this method is
1795 >     * behaviorally equivalent to
1796       *  <pre> {@code
1797       * while (!blocker.isReleasable())
1798       *   if (blocker.block())
1799       *     return;
1800       * }</pre>
1801 <     * If the caller is a ForkJoinTask, then the pool may first
1802 <     * be expanded to ensure parallelism, and later adjusted.
1801 >     *
1802 >     * If the caller is a {@code ForkJoinTask}, then the pool may
1803 >     * first be expanded to ensure parallelism, and later adjusted.
1804       *
1805       * @param blocker the blocker
1806       * @param maintainParallelism if {@code true} and supported by
# Line 1882 | Line 1832 | public class ForkJoinPool extends Abstra
1832          do {} while (!blocker.isReleasable() && !blocker.block());
1833      }
1834  
1835 <    // AbstractExecutorService overrides
1835 >    // AbstractExecutorService overrides.  These rely on undocumented
1836 >    // fact that ForkJoinTask.adapt returns ForkJoinTasks that also
1837 >    // implement RunnableFuture.
1838  
1839      protected <T> RunnableFuture<T> newTaskFor(Runnable runnable, T value) {
1840 <        return new AdaptedRunnable<T>(runnable, value);
1840 >        return (RunnableFuture<T>) ForkJoinTask.adapt(runnable, value);
1841      }
1842  
1843      protected <T> RunnableFuture<T> newTaskFor(Callable<T> callable) {
1844 <        return new AdaptedCallable<T>(callable);
1844 >        return (RunnableFuture<T>) ForkJoinTask.adapt(callable);
1845      }
1846  
1847      // Unsafe mechanics

Diff Legend

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