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.41 by jsr166, Mon Aug 3 01:11:58 2009 UTC vs.
Revision 1.42 by dl, Mon Aug 3 13:01:15 2009 UTC

#	Line 23 | Line 23 | import java.util.concurrent.atomic.Atomi
23		* An {@link ExecutorService} for running {@link ForkJoinTask}s.
24		* A {@code ForkJoinPool} provides the entry point for submissions
25		* from non-{@code ForkJoinTask}s, as well as management and
26	<	* monitoring operations.  Normally a single {@code ForkJoinPool} is
27	<	* used for a large number of submitted tasks. Otherwise, use would
28	<	* not usually outweigh the construction and bookkeeping overhead of
29	<	* creating a large set of threads.
26	>	* monitoring operations.
27		*
28	<	* <p>{@code ForkJoinPool}s differ from other kinds of {@link
29	<	* Executor}s mainly in that they provide <em>work-stealing</em>: all
30	<	* threads in the pool attempt to find and execute subtasks created by
31	<	* other active tasks (eventually blocking if none exist). This makes
32	<	* them efficient when most tasks spawn other subtasks (as do most
33	<	* {@code ForkJoinTask}s), as well as the mixed execution of some
34	<	* plain {@code Runnable}- or {@code Callable}- based activities along
35	<	* with {@code ForkJoinTask}s. When setting {@linkplain #setAsyncMode
36	<	* async mode}, a {@code ForkJoinPool} may also be appropriate for use
37	<	* with fine-grained tasks that are never joined. Otherwise, other
38	<	* {@code ExecutorService} implementations are typically more
39	<	* appropriate choices.
28	>	* <p>A {@code ForkJoinPool} differs from other kinds of {@link
29	>	* ExecutorService} mainly by virtue of employing
30	>	* <em>work-stealing</em>: all threads in the pool attempt to find and
31	>	* execute subtasks created by other active tasks (eventually blocking
32	>	* waiting for work if none exist). This enables efficient processing
33	>	* when most tasks spawn other subtasks (as do most {@code
34	>	* ForkJoinTask}s). A {@code ForkJoinPool} may also be used for mixed
35	>	* execution of some plain {@code Runnable}- or {@code Callable}-
36	>	* based activities along with {@code ForkJoinTask}s. When setting
37	>	* {@linkplain #setAsyncMode async mode}, a {@code ForkJoinPool} may
38	>	* also be appropriate for use with fine-grained tasks of any form
39	>	* that are never joined. Otherwise, other {@code ExecutorService}
40	>	* implementations are typically more appropriate choices.
41		*
42	<	* <p>A {@code ForkJoinPool} may be constructed with a given
43	<	* parallelism level (target pool size), which it attempts to maintain
44	<	* by dynamically adding, suspending, or resuming threads, even if
45	<	* some tasks are waiting to join others. However, no such adjustments
46	<	* are performed in the face of blocked IO or other unmanaged
42	>	* <p>A {@code ForkJoinPool} is constructed with a given target
43	>	* parallelism level; by default, equal to the number of available
44	>	* processors. Unless configured otherwise via {@link
45	>	* #setMaintainsParallelism}, the pool attempts to maintain this
46	>	* number of active (or available) threads by dynamically adding,
47	>	* suspending, or resuming internal worker threads, even if some tasks
48	>	* are waiting to join others. However, no such adjustments are
49	>	* performed in the face of blocked IO or other unmanaged
50		* synchronization. The nested {@link ManagedBlocker} interface
51		* enables extension of the kinds of synchronization accommodated.
52		* The target parallelism level may also be changed dynamically
53	<	* ({@link #setParallelism}) and thread construction can be limited
54	<	* using methods {@link #setMaximumPoolSize} and/or {@link
55	<	* #setMaintainsParallelism}.
53	>	* ({@link #setParallelism}). The total number of threads may be
54	>	* limited using method {@link #setMaximumPoolSize}, in which case it
55	>	* may become possible for the activities of a pool to stall due to
56	>	* the lack of available threads to process new tasks.
57		*
58		* <p>In addition to execution and lifecycle control methods, this
59		* class provides status check methods (for example
#	Line 60 | Line 62 | import java.util.concurrent.atomic.Atomi
62		* {@link #toString} returns indications of pool state in a
63		* convenient form for informal monitoring.
64		*
65	+	* <p><b>Sample Usage.</b> Normally a single {@code ForkJoinPool} is
66	+	* used for all parallel task execution in a program or subsystem.
67	+	* Otherwise, use would not usually outweigh the construction and
68	+	* bookkeeping overhead of creating a large set of threads. For
69	+	* example a common pool could be used for the {@code SortTasks}
70	+	* illustrated in {@link RecursiveAction}. Because {@code
71	+	* ForkJoinPool} uses threads in {@linkplain java.lang.Thread#isDaemon
72	+	* daemon} mode, there is typically no need to explictly {@link
73	+	* #shutdown} such a pool upon program exit.
74	+	*
75	+	* <pre>
76	+	* static final ForkJoinPool mainPool = new ForkJoinPool();
77	+	* ...
78	+	* public void sort(long[] array) {
79	+	*   mainPool.invoke(new SortTask(array, 0, array.length));
80	+	* }
81	+	* </pre>
82	+	*
83		* <p><b>Implementation notes</b>: This implementation restricts the
84		* maximum number of running threads to 32767. Attempts to create
85		* pools with greater than the maximum result in
#	Line 343 | Line 363 | public class ForkJoinPool extends Abstra
363		// Constructors
364
365		/**
366	<	* Creates a {@code ForkJoinPool} with a pool size equal to the
367	<	* number of processors available on the system, using the
368	<	* {@linkplain #defaultForkJoinWorkerThreadFactory default thread factory}.
366	>	* Creates a {@code ForkJoinPool} with parallelism equal to {@link
367	>	* java.lang.Runtime#availableProcessors}, and using the {@linkplain
368	>	* #defaultForkJoinWorkerThreadFactory default thread factory}.
369		*
370		* @throws SecurityException if a security manager exists and
371		*         the caller is not permitted to modify threads
#	Line 358 | Line 378 | public class ForkJoinPool extends Abstra
378		}
379
380		/**
381	<	* Creates a {@code ForkJoinPool} with the indicated parallelism level
382	<	* threads and using the
383	<	* {@linkplain #defaultForkJoinWorkerThreadFactory default thread factory}.
381	>	* Creates a {@code ForkJoinPool} with the indicated parallelism
382	>	* level and using the {@linkplain
383	>	* #defaultForkJoinWorkerThreadFactory default thread factory}.
384		*
385	<	* @param parallelism the number of worker threads
385	>	* @param parallelism the parallelism level
386		* @throws IllegalArgumentException if parallelism less than or
387		* equal to zero
388		* @throws SecurityException if a security manager exists and
#	Line 375 | Line 395 | public class ForkJoinPool extends Abstra
395		}
396
397		/**
398	<	* Creates a {@code ForkJoinPool} with parallelism equal to the
399	<	* number of processors available on the system and using the
400	<	* given thread factory.
398	>	* Creates a {@code ForkJoinPool} with parallelism equal to {@link
399	>	* java.lang.Runtime#availableProcessors}, and using the given
400	>	* thread factory.
401		*
402		* @param factory the factory for creating new threads
403		* @throws NullPointerException if factory is null
#	Line 394 | Line 414 | public class ForkJoinPool extends Abstra
414		* Creates a {@code ForkJoinPool} with the given parallelism and
415		* thread factory.
416		*
417	<	* @param parallelism the targeted number of worker threads
417	>	* @param parallelism the parallelism level
418		* @param factory the factory for creating new threads
419		* @throws IllegalArgumentException if parallelism less than or
420		* equal to zero, or greater than implementation limit
#	Line 447 | Line 467 | public class ForkJoinPool extends Abstra
467		* Currently requires size to be a power of two.
468		*/
469		private static int arraySizeFor(int poolSize) {
470	<	return (poolSize <= 1) ? 1 :
471	<	(1 << (32 - Integer.numberOfLeadingZeros(poolSize-1)));
470	>	if (poolSize <= 1)
471	>	return 1;
472	>	// See Hackers Delight, sec 3.2
473	>	int c = poolSize >= MAX_THREADS ? MAX_THREADS : (poolSize - 1);
474	>	c |= c >>>  1;
475	>	c |= c >>>  2;
476	>	c |= c >>>  4;
477	>	c |= c >>>  8;
478	>	c |= c >>> 16;
479	>	return c + 1;
480		}
481
482		/**
#	Line 739 | Line 767 | public class ForkJoinPool extends Abstra
767		final ReentrantLock lock = this.workerLock;
768		lock.lock();
769		try {
770	<	if (!isTerminating()) {
770	>	if (isProcessingTasks()) {
771		int p = this.parallelism;
772		this.parallelism = parallelism;
773		if (parallelism > p)
#	Line 754 | Line 782 | public class ForkJoinPool extends Abstra
782		}
783
784		/**
785	<	* Returns the targeted number of worker threads in this pool.
785	>	* Returns the targeted parallelism level of this pool.
786		*
787	<	* @return the targeted number of worker threads in this pool
787	>	* @return the targeted parallelism level of this pool
788		*/
789		public int getParallelism() {
790		return parallelism;
#	Line 776 | Line 804 | public class ForkJoinPool extends Abstra
804
805		/**
806		* Returns the maximum number of threads allowed to exist in the
807	<	* pool, even if there are insufficient unblocked running threads.
807	>	* pool.  Unless set using {@link #setMaximumPoolSize}, the
808	>	* maximum is an implementation-defined value designed only to
809	>	* prevent runaway growth.
810		*
811		* @return the maximum
812		*/
#	Line 786 | Line 816 | public class ForkJoinPool extends Abstra
816
817		/**
818		* Sets the maximum number of threads allowed to exist in the
819	<	* pool, even if there are insufficient unblocked running threads.
820	<	* Setting this value has no effect on current pool size. It
791	<	* controls construction of new threads.
819	>	* pool.  Setting this value has no effect on current pool
820	>	* size. It controls construction of new threads.
821		*
822		* @throws IllegalArgumentException if negative or greater than
823		* internal implementation limit
#	Line 1091 | Line 1120 | public class ForkJoinPool extends Abstra
1120		}
1121
1122		/**
1123	<	* Attempts to stop all actively executing tasks, and cancels all
1124	<	* waiting tasks.  Tasks that are in the process of being
1125	<	* submitted or executed concurrently during the course of this
1126	<	* method may or may not be rejected. Unlike some other executors,
1127	<	* this method cancels rather than collects non-executed tasks
1128	<	* upon termination, so always returns an empty list. However, you
1129	<	* can use method {@link #drainTasksTo} before invoking this
1130	<	* method to transfer unexecuted tasks to another collection.
1123	>	* Attempts to cancel and/or stop all tasks, and reject all
1124	>	* subsequently submitted tasks.  Tasks that are in the process of
1125	>	* being submitted or executed concurrently during the course of
1126	>	* this method may or may not be rejected. This method cancels
1127	>	* both existing and unexecuted tasks, in order to permit
1128	>	* termination in the presence of task dependencies. So the method
1129	>	* always returns an empty list (unlike the case for some other
1130	>	* Executors).
1131		*
1132		* @return an empty list
1133		* @throws SecurityException if a security manager exists and
#	Line 1123 | Line 1152 | public class ForkJoinPool extends Abstra
1152
1153		/**
1154		* Returns {@code true} if the process of termination has
1155	<	* commenced but possibly not yet completed.
1155	>	* commenced but not yet completed.  This method may be useful for
1156	>	* debugging. A return of {@code true} reported a sufficient
1157	>	* period after shutdown may indicate that submitted tasks have
1158	>	* ignored or suppressed interruption, causing this executor not
1159	>	* to properly terminate.
1160		*
1161	<	* @return {@code true} if terminating
1161	>	* @return {@code true} if terminating but not yet terminated
1162		*/
1163		public boolean isTerminating() {
1164	<	return runStateOf(runControl) >= TERMINATING;
1164	>	return runStateOf(runControl) == TERMINATING;
1165		}
1166
1167		/**
#	Line 1141 | Line 1174 | public class ForkJoinPool extends Abstra
1174		}
1175
1176		/**
1177	+	* Returns true if pool is not terminating or terminated.
1178	+	* Used internally to suppress execution when terminating.
1179	+	*/
1180	+	final boolean isProcessingTasks() {
1181	+	return runStateOf(runControl) < TERMINATING;
1182	+	}
1183	+
1184	+	/**
1185		* Blocks until all tasks have completed execution after a shutdown
1186		* request, or the timeout occurs, or the current thread is
1187		* interrupted, whichever happens first.
#	Line 1194 | Line 1235 | public class ForkJoinPool extends Abstra
1235		transitionRunStateTo(TERMINATED);
1236		termination.signalAll();
1237		}
1238	<	else if (!isTerminating()) {
1238	>	else if (isProcessingTasks()) {
1239		tryShrinkWorkerArray();
1240		tryResumeSpare(true); // allow replacement
1241		}
#	Line 1435 | Line 1476 | public class ForkJoinPool extends Abstra
1476		final void sync(ForkJoinWorkerThread w) {
1477		updateStealCount(w); // Transfer w's count while it is idle
1478
1479	<	while (!w.isShutdown() && !isTerminating() && !suspendIfSpare(w)) {
1479	>	while (!w.isShutdown() && isProcessingTasks() && !suspendIfSpare(w)) {
1480		long prev = w.lastEventCount;
1481		WaitQueueNode node = null;
1482		WaitQueueNode h;
#	Line 1640 | Line 1681 | public class ForkJoinPool extends Abstra
1681		for (k = 0; k < len && ws[k] != null; ++k)
1682		;
1683		}
1684	<	if (k < len && !isTerminating() && (w = createWorker(k)) != null) {
1684	>	if (k < len && isProcessingTasks() && (w = createWorker(k)) != null) {
1685		ws[k] = w;
1686		w.start();
1687		}

Diff Legend

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