ViewVC Help

View File | Revision Log | Show Annotations | Download File | Root Listing

root/jsr166/jsr166/src/jsr166y/ForkJoinPool.java

Comparing jsr166/src/jsr166y/ForkJoinPool.java (file contents):
Revision 1.28 by jsr166, Mon Jul 27 20:57:44 2009 UTC vs.
Revision 1.38 by jsr166, Sun Aug 2 17:02:06 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 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 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 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 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 1772 | 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 1815 | 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 1865 | 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