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.66 by dl, Sun Aug 29 23:34:46 2010 UTC vs.
Revision 1.84 by dl, Sat Nov 13 13:11:51 2010 UTC

#	Line 6 | Line 6
6
7		package jsr166y;
8
9	–	import java.util.concurrent.*;
10	–
9		import java.util.ArrayList;
10		import java.util.Arrays;
11		import java.util.Collection;
12		import java.util.Collections;
13		import java.util.List;
14	+	import java.util.concurrent.AbstractExecutorService;
15	+	import java.util.concurrent.Callable;
16	+	import java.util.concurrent.ExecutorService;
17	+	import java.util.concurrent.Future;
18	+	import java.util.concurrent.RejectedExecutionException;
19	+	import java.util.concurrent.RunnableFuture;
20	+	import java.util.concurrent.TimeUnit;
21	+	import java.util.concurrent.TimeoutException;
22	+	import java.util.concurrent.atomic.AtomicInteger;
23		import java.util.concurrent.locks.LockSupport;
24		import java.util.concurrent.locks.ReentrantLock;
18	–	import java.util.concurrent.atomic.AtomicInteger;
19	–	import java.util.concurrent.CountDownLatch;
25
26		/**
27		* An {@link ExecutorService} for running {@link ForkJoinTask}s.
#	Line 69 | Line 74 | import java.util.concurrent.CountDownLat
74		*    <td ALIGN=CENTER> <b>Call from within fork/join computations</b></td>
75		*  </tr>
76		*  <tr>
77	<	*    <td> <b>Arange async execution</td>
77	>	*    <td> <b>Arrange async execution</td>
78		*    <td> {@link #execute(ForkJoinTask)}</td>
79		*    <td> {@link ForkJoinTask#fork}</td>
80		*  </tr>
#	Line 140 | Line 145 | public class ForkJoinPool extends Abstra
145		* Beyond work-stealing support and essential bookkeeping, the
146		* main responsibility of this framework is to take actions when
147		* one worker is waiting to join a task stolen (or always held by)
148	<	* another.  Becauae we are multiplexing many tasks on to a pool
148	>	* another.  Because we are multiplexing many tasks on to a pool
149		* of workers, we can't just let them block (as in Thread.join).
150		* We also cannot just reassign the joiner's run-time stack with
151		* another and replace it later, which would be a form of
#	Line 157 | Line 162 | public class ForkJoinPool extends Abstra
162		*      links to try to find such a task.
163		*
164		*   Compensating: Unless there are already enough live threads,
165	<	*      method helpMaintainParallelism() may create or or
165	>	*      method helpMaintainParallelism() may create or
166		*      re-activate a spare thread to compensate for blocked
167		*      joiners until they unblock.
168		*
#	Line 226 | Line 231 | public class ForkJoinPool extends Abstra
231		* ManagedBlocker), we may create or resume others to take their
232		* place until they unblock (see below). Implementing this
233		* requires counts of the number of "running" threads (i.e., those
234	<	* that are neither blocked nor artifically suspended) as well as
234	>	* that are neither blocked nor artificially suspended) as well as
235		* the total number.  These two values are packed into one field,
236		* "workerCounts" because we need accurate snapshots when deciding
237		* to create, resume or suspend.  Note however that the
238	<	* correspondance of these counts to reality is not guaranteed. In
238	>	* correspondence of these counts to reality is not guaranteed. In
239		* particular updates for unblocked threads may lag until they
240		* actually wake up.
241		*
#	Line 301 | Line 306 | public class ForkJoinPool extends Abstra
306		* about the same time as another is needlessly being created. We
307		* counteract this and related slop in part by requiring resumed
308		* spares to immediately recheck (in preStep) to see whether they
309	<	* they should re-suspend.
309	>	* should re-suspend.
310		*
311		* 6. Killing off unneeded workers. A timeout mechanism is used to
312		* shed unused workers: The oldest (first) event queue waiter uses
#	Line 315 | Line 320 | public class ForkJoinPool extends Abstra
320		* 7. Deciding when to create new workers. The main dynamic
321		* control in this class is deciding when to create extra threads
322		* in method helpMaintainParallelism. We would like to keep
323	<	* exactly #parallelism threads running, which is an impossble
323	>	* exactly #parallelism threads running, which is an impossible
324		* task. We always need to create one when the number of running
325		* threads would become zero and all workers are busy. Beyond
326	<	* this, we must rely on heuristics that work well in the the
327	<	* presence of transients phenomena such as GC stalls, dynamic
326	>	* this, we must rely on heuristics that work well in the
327	>	* presence of transient phenomena such as GC stalls, dynamic
328		* compilation, and wake-up lags. These transients are extremely
329		* common -- we are normally trying to fully saturate the CPUs on
330		* a machine, so almost any activity other than running tasks
#	Line 346 | Line 351 | public class ForkJoinPool extends Abstra
351		* "while ((local = field) != 0)") which are usually the simplest
352		* way to ensure the required read orderings (which are sometimes
353		* critical). Also several occurrences of the unusual "do {}
354	<	* while(!cas...)" which is the simplest way to force an update of
354	>	* while (!cas...)" which is the simplest way to force an update of
355		* a CAS'ed variable. There are also other coding oddities that
356		* help some methods perform reasonably even when interpreted (not
357		* compiled), at the expense of some messy constructions that
#	Line 420 | Line 425 | public class ForkJoinPool extends Abstra
425		/**
426		* The time to block in a join (see awaitJoin) before checking if
427		* a new worker should be (re)started to maintain parallelism
428	<	* level. The value should be short enough to maintain gloabal
428	>	* level. The value should be short enough to maintain global
429		* responsiveness and progress but long enough to avoid
430		* counterproductive firings during GC stalls or unrelated system
431		* activity, and to not bog down systems with continual re-firings
#	Line 430 | Line 435 | public class ForkJoinPool extends Abstra
435
436		/**
437		* The wakeup interval (in nanoseconds) for the oldest worker
438	<	* worker waiting for an event invokes tryShutdownUnusedWorker to shrink
439	<	* the number of workers.  The exact value does not matter too
440	<	* much, but should be long enough to slowly release resources
441	<	* during long periods without use without disrupting normal use.
438	>	* waiting for an event to invoke tryShutdownUnusedWorker to
439	>	* shrink the number of workers.  The exact value does not matter
440	>	* too much. It must be short enough to release resources during
441	>	* sustained periods of idleness, but not so short that threads
442	>	* are continually re-created.
443		*/
444		private static final long SHRINK_RATE_NANOS =
445		30L * 1000L * 1000L * 1000L; // 2 per minute
#	Line 483 | Line 489 | public class ForkJoinPool extends Abstra
489		private volatile long stealCount;
490
491		/**
492	<	* Encoded record of top of treiber stack of threads waiting for
492	>	* Encoded record of top of Treiber stack of threads waiting for
493		* events. The top 32 bits contain the count being waited for. The
494		* bottom 16 bits contains one plus the pool index of waiting
495		* worker thread. (Bits 16-31 are unused.)
#	Line 502 | Line 508 | public class ForkJoinPool extends Abstra
508		private volatile int eventCount;
509
510		/**
511	<	* Encoded record of top of treiber stack of spare threads waiting
511	>	* Encoded record of top of Treiber stack of spare threads waiting
512		* for resumption. The top 16 bits contain an arbitrary count to
513		* avoid ABA effects. The bottom 16bits contains one plus the pool
514		* index of waiting worker thread.
#	Line 516 | Line 522 | public class ForkJoinPool extends Abstra
522		* Lifecycle control. The low word contains the number of workers
523		* that are (probably) executing tasks. This value is atomically
524		* incremented before a worker gets a task to run, and decremented
525	<	* when worker has no tasks and cannot find any.  Bits 16-18
525	>	* when a worker has no tasks and cannot find any.  Bits 16-18
526		* contain runLevel value. When all are zero, the pool is
527		* running. Level transitions are monotonic (running -> shutdown
528		* -> terminating -> terminated) so each transition adds a bit.
#	Line 605 | Line 611 | public class ForkJoinPool extends Abstra
611		* (rarely) necessary when other count updates lag.
612		*
613		* @param dr -- either zero or ONE_RUNNING
614	<	* @param dt == either zero or ONE_TOTAL
614	>	* @param dt -- either zero or ONE_TOTAL
615		*/
616		private void decrementWorkerCounts(int dr, int dt) {
617		for (;;) {
#	Line 674 | Line 680 | public class ForkJoinPool extends Abstra
680		}
681
682		/**
683	<	* Nulls out record of worker in workers array
683	>	* Nulls out record of worker in workers array.
684		*/
685		private void forgetWorker(ForkJoinWorkerThread w) {
686		int idx = w.poolIndex;
687	<	// Locking helps method recordWorker avoid unecessary expansion
687	>	// Locking helps method recordWorker avoid unnecessary expansion
688		final ReentrantLock lock = this.workerLock;
689		lock.lock();
690		try {
#	Line 693 | Line 699 | public class ForkJoinPool extends Abstra
699		/**
700		* Final callback from terminating worker.  Removes record of
701		* worker from array, and adjusts counts. If pool is shutting
702	<	* down, tries to complete terminatation.
702	>	* down, tries to complete termination.
703		*
704		* @param w the worker
705		*/
706		final void workerTerminated(ForkJoinWorkerThread w) {
707		forgetWorker(w);
708	<	decrementWorkerCounts(w.isTrimmed()? 0 : ONE_RUNNING, ONE_TOTAL);
708	>	decrementWorkerCounts(w.isTrimmed() ? 0 : ONE_RUNNING, ONE_TOTAL);
709		while (w.stealCount != 0) // collect final count
710		tryAccumulateStealCount(w);
711		tryTerminate(false);
#	Line 785 | Line 791 | public class ForkJoinPool extends Abstra
791		if (tryAccumulateStealCount(w)) { // transfer while idle
792		boolean untimed = (w.nextWaiter != 0L ||
793		(workerCounts & RUNNING_COUNT_MASK) <= 1);
794	<	long startTime = untimed? 0 : System.nanoTime();
794	>	long startTime = untimed ? 0 : System.nanoTime();
795		Thread.interrupted();         // clear/ignore interrupt
796	<	if (eventCount != ec || w.runState != 0 ||
797	<	runState >= TERMINATING)  // recheck after clear
792	<	break;
796	>	if (eventCount != ec || w.isTerminating())
797	>	break;                    // recheck after clear
798		if (untimed)
799		LockSupport.park(w);
800		else {
801		LockSupport.parkNanos(w, SHRINK_RATE_NANOS);
802	<	if (eventCount != ec || w.runState != 0 ||
798	<	runState >= TERMINATING)
802	>	if (eventCount != ec || w.isTerminating())
803		break;
804		if (System.nanoTime() - startTime >= SHRINK_RATE_NANOS)
805		tryShutdownUnusedWorker(ec);
#	Line 807 | Line 811 | public class ForkJoinPool extends Abstra
811		// Maintaining parallelism
812
813		/**
814	<	* Pushes worker onto the spare stack
814	>	* Pushes worker onto the spare stack.
815		*/
816		final void pushSpare(ForkJoinWorkerThread w) {
817		int ns = (++w.spareCount << SPARE_COUNT_SHIFT) | (w.poolIndex + 1);
#	Line 832 | Line 836 | public class ForkJoinPool extends Abstra
836		UNSAFE.compareAndSwapInt(this, spareWaitersOffset,
837		sw, w.nextSpare)) {
838		int c; // increment running count before resume
839	<	do {} while(!UNSAFE.compareAndSwapInt
840	<	(this, workerCountsOffset,
841	<	c = workerCounts, c + ONE_RUNNING));
839	>	do {} while (!UNSAFE.compareAndSwapInt
840	>	(this, workerCountsOffset,
841	>	c = workerCounts, c + ONE_RUNNING));
842		if (w.tryUnsuspend())
843		LockSupport.unpark(w);
844		else   // back out if w was shutdown
#	Line 846 | Line 850 | public class ForkJoinPool extends Abstra
850		* Tries to increase the number of running workers if below target
851		* parallelism: If a spare exists tries to resume it via
852		* tryResumeSpare.  Otherwise, if not enough total workers or all
853	<	* existing workers are busy, adds a new worker. In all casses also
853	>	* existing workers are busy, adds a new worker. In all cases also
854		* helps wake up releasable workers waiting for work.
855		*/
856		private void helpMaintainParallelism() {
#	Line 863 | Line 867 | public class ForkJoinPool extends Abstra
867		UNSAFE.compareAndSwapInt(this, workerCountsOffset, wc,
868		wc + (ONE_RUNNING|ONE_TOTAL))) {
869		ForkJoinWorkerThread w = null;
870	+	Throwable fail = null;
871		try {
872		w = factory.newThread(this);
873	<	} finally { // adjust on null or exceptional factory return
874	<	if (w == null) {
870	<	decrementWorkerCounts(ONE_RUNNING, ONE_TOTAL);
871	<	tryTerminate(false); // handle failure during shutdown
872	<	}
873	>	} catch (Throwable ex) {
874	>	fail = ex;
875		}
876	<	if (w == null)
876	>	if (w == null) { // null or exceptional factory return
877	>	decrementWorkerCounts(ONE_RUNNING, ONE_TOTAL);
878	>	tryTerminate(false); // handle failure during shutdown
879	>	// If originating from an external caller,
880	>	// propagate exception, else ignore
881	>	if (fail != null && runState < TERMINATING &&
882	>	!(Thread.currentThread() instanceof
883	>	ForkJoinWorkerThread))
884	>	UNSAFE.throwException(fail);
885		break;
886	+	}
887		w.start(recordWorker(w), ueh);
888		if ((workerCounts >>> TOTAL_COUNT_SHIFT) >= pc) {
889		int c; // advance event count
#	Line 961 | Line 972 | public class ForkJoinPool extends Abstra
972		boolean active = w.active;
973		boolean inactivate = false;
974		int pc = parallelism;
975	<	int rs;
976	<	while (w.runState == 0 && (rs = runState) < TERMINATING) {
975	>	while (w.runState == 0) {
976	>	int rs = runState;
977	>	if (rs >= TERMINATING) { // propagate shutdown
978	>	w.shutdown();
979	>	break;
980	>	}
981		if ((inactivate || (active && (rs & ACTIVE_COUNT_MASK) >= pc)) &&
982		UNSAFE.compareAndSwapInt(this, runStateOffset, rs, rs - 1))
983		inactivate = active = w.active = false;
#	Line 985 | Line 1000 | public class ForkJoinPool extends Abstra
1000		w.lastEventCount = ec;     // no need to wait
1001		break;
1002		}
1003	<	else if (!(inactivate |= active))
1003	>	else if (!(inactivate |= active))
1004		eventSync(w, wec);         // must inactivate before sync
1005		}
1006		else
#	Line 999 | Line 1014 | public class ForkJoinPool extends Abstra
1014		*
1015		* @param joinMe the task to join
1016		* @param worker the current worker thread
1017	+	* @param timed true if wait should time out
1018	+	* @param nanos timeout value if timed
1019		*/
1020	<	final void awaitJoin(ForkJoinTask<?> joinMe, ForkJoinWorkerThread worker) {
1020	>	final void awaitJoin(ForkJoinTask<?> joinMe, ForkJoinWorkerThread worker,
1021	>	boolean timed, long nanos) {
1022	>	long startTime = timed? System.nanoTime() : 0L;
1023		int retries = 2 + (parallelism >> 2); // #helpJoins before blocking
1024		while (joinMe.status >= 0) {
1025		int wc;
1026	+	long nt = 0L;
1027	+	if (runState >= TERMINATING) {
1028	+	joinMe.cancelIgnoringExceptions();
1029	+	break;
1030	+	}
1031		worker.helpJoinTask(joinMe);
1032		if (joinMe.status < 0)
1033		break;
1034		else if (retries > 0)
1035		--retries;
1036	+	else if (timed &&
1037	+	(nt = nanos - (System.nanoTime() - startTime)) <= 0L)
1038	+	break;
1039		else if (((wc = workerCounts) & RUNNING_COUNT_MASK) != 0 &&
1040		UNSAFE.compareAndSwapInt(this, workerCountsOffset,
1041		wc, wc - ONE_RUNNING)) {
#	Line 1017 | Line 1044 | public class ForkJoinPool extends Abstra
1044		(h = eventWaiters) != 0L && // help release others
1045		(int)(h >>> EVENT_COUNT_SHIFT) != eventCount)
1046		releaseEventWaiters();
1047	<	if (stat >= 0 &&
1048	<	((workerCounts & RUNNING_COUNT_MASK) == 0 ||
1049	<	(stat =
1050	<	joinMe.internalAwaitDone(JOIN_TIMEOUT_MILLIS)) >= 0))
1051	<	helpMaintainParallelism(); // timeout or no running workers
1047	>	if (stat >= 0) {
1048	>	if ((workerCounts & RUNNING_COUNT_MASK) != 0) {
1049	>	long ms; int ns;
1050	>	if (!timed) {
1051	>	ms = JOIN_TIMEOUT_MILLIS;
1052	>	ns = 0;
1053	>	}
1054	>	else { // at most JOIN_TIMEOUT_MILLIS per wait
1055	>	ms = nt / 1000000;
1056	>	if (ms > JOIN_TIMEOUT_MILLIS) {
1057	>	ms = JOIN_TIMEOUT_MILLIS;
1058	>	ns = 0;
1059	>	}
1060	>	else
1061	>	ns = (int) (nt % 1000000);
1062	>	}
1063	>	stat = joinMe.internalAwaitDone(ms, ns);
1064	>	}
1065	>	if (stat >= 0) // timeout or no running workers
1066	>	helpMaintainParallelism();
1067	>	}
1068		do {} while (!UNSAFE.compareAndSwapInt
1069		(this, workerCountsOffset,
1070		c = workerCounts, c + ONE_RUNNING));
#	Line 1085 | Line 1128 | public class ForkJoinPool extends Abstra
1128		// Finish now if all threads terminated; else in some subsequent call
1129		if ((workerCounts >>> TOTAL_COUNT_SHIFT) == 0) {
1130		advanceRunLevel(TERMINATED);
1131	<	termination.arrive();
1131	>	termination.forceTermination();
1132		}
1133		return true;
1134		}
1135
1136	+
1137		/**
1138		* Actions on transition to TERMINATING
1139		*
#	Line 1107 | Line 1151 | public class ForkJoinPool extends Abstra
1151		c = eventCount, c+1);
1152		eventWaiters = 0L; // clobber lists
1153		spareWaiters = 0;
1154	<	ForkJoinWorkerThread[] ws = workers;
1111	<	int n = ws.length;
1112	<	for (int i = 0; i < n; ++i) {
1113	<	ForkJoinWorkerThread w = ws[i];
1154	>	for (ForkJoinWorkerThread w : workers) {
1155		if (w != null) {
1156		w.shutdown();
1157		if (passes > 0 && !w.isTerminated()) {
1158		w.cancelTasks();
1159		LockSupport.unpark(w);
1160	<	if (passes > 1) {
1160	>	if (passes > 1 && !w.isInterrupted()) {
1161		try {
1162		w.interrupt();
1163		} catch (SecurityException ignore) {
#	Line 1129 | Line 1170 | public class ForkJoinPool extends Abstra
1170		}
1171
1172		/**
1173	<	* Clear out and cancel submissions, ignoring exceptions
1173	>	* Clears out and cancels submissions, ignoring exceptions.
1174		*/
1175		private void cancelSubmissions() {
1176		ForkJoinTask<?> task;
#	Line 1144 | Line 1185 | public class ForkJoinPool extends Abstra
1185		// misc support for ForkJoinWorkerThread
1186
1187		/**
1188	<	* Returns pool number
1188	>	* Returns pool number.
1189		*/
1190		final int getPoolNumber() {
1191		return poolNumber;
1192		}
1193
1194		/**
1195	<	* Tries to accumulates steal count from a worker, clearing
1196	<	* the worker's value.
1195	>	* Tries to accumulate steal count from a worker, clearing
1196	>	* the worker's value if successful.
1197		*
1198		* @return true if worker steal count now zero
1199		*/
#	Line 1174 | Line 1215 | public class ForkJoinPool extends Abstra
1215		*/
1216		final int idlePerActive() {
1217		int pc = parallelism; // use parallelism, not rc
1218	<	int ac = runState;    // no mask -- artifically boosts during shutdown
1218	>	int ac = runState;    // no mask -- artificially boosts during shutdown
1219		// Use exact results for small values, saturate past 4
1220	<	return pc <= ac? 0 : pc >>> 1 <= ac? 1 : pc >>> 2 <= ac? 3 : pc >>> 3;
1220	>	return ((pc <= ac) ? 0 :
1221	>	(pc >>> 1 <= ac) ? 1 :
1222	>	(pc >>> 2 <= ac) ? 3 :
1223	>	pc >>> 3);
1224		}
1225
1226		// Public and protected methods
#	Line 1226 | Line 1270 | public class ForkJoinPool extends Abstra
1270		* use {@link #defaultForkJoinWorkerThreadFactory}.
1271		* @param handler the handler for internal worker threads that
1272		* terminate due to unrecoverable errors encountered while executing
1273	<	* tasks. For default value, use <code>null</code>.
1273	>	* tasks. For default value, use {@code null}.
1274		* @param asyncMode if true,
1275		* establishes local first-in-first-out scheduling mode for forked
1276		* tasks that are never joined. This mode may be more appropriate
1277		* than default locally stack-based mode in applications in which
1278		* worker threads only process event-style asynchronous tasks.
1279	<	* For default value, use <code>false</code>.
1279	>	* For default value, use {@code false}.
1280		* @throws IllegalArgumentException if parallelism less than or
1281		*         equal to zero, or greater than implementation limit
1282		* @throws NullPointerException if the factory is null
#	Line 1280 | Line 1324 | public class ForkJoinPool extends Abstra
1324		// Execution methods
1325
1326		/**
1327	<	* Common code for execute, invoke and submit
1327	>	* Submits task and creates, starts, or resumes some workers if necessary
1328		*/
1329		private <T> void doSubmit(ForkJoinTask<T> task) {
1286	–	if (task == null)
1287	–	throw new NullPointerException();
1288	–	if (runState >= SHUTDOWN)
1289	–	throw new RejectedExecutionException();
1330		submissionQueue.offer(task);
1331		int c; // try to increment event count -- CAS failure OK
1332		UNSAFE.compareAndSwapInt(this, eventCountOffset, c = eventCount, c+1);
1333	<	helpMaintainParallelism(); // create, start, or resume some workers
1333	>	helpMaintainParallelism();
1334		}
1335
1336		/**
#	Line 1303 | Line 1343 | public class ForkJoinPool extends Abstra
1343		*         scheduled for execution
1344		*/
1345		public <T> T invoke(ForkJoinTask<T> task) {
1346	<	doSubmit(task);
1347	<	return task.join();
1346	>	if (task == null)
1347	>	throw new NullPointerException();
1348	>	if (runState >= SHUTDOWN)
1349	>	throw new RejectedExecutionException();
1350	>	Thread t = Thread.currentThread();
1351	>	if ((t instanceof ForkJoinWorkerThread) &&
1352	>	((ForkJoinWorkerThread)t).pool == this)
1353	>	return task.invoke();  // bypass submit if in same pool
1354	>	else {
1355	>	doSubmit(task);
1356	>	return task.join();
1357	>	}
1358	>	}
1359	>
1360	>	/**
1361	>	* Unless terminating, forks task if within an ongoing FJ
1362	>	* computation in the current pool, else submits as external task.
1363	>	*/
1364	>	private <T> void forkOrSubmit(ForkJoinTask<T> task) {
1365	>	if (runState >= SHUTDOWN)
1366	>	throw new RejectedExecutionException();
1367	>	Thread t = Thread.currentThread();
1368	>	if ((t instanceof ForkJoinWorkerThread) &&
1369	>	((ForkJoinWorkerThread)t).pool == this)
1370	>	task.fork();
1371	>	else
1372	>	doSubmit(task);
1373		}
1374
1375		/**
#	Line 1316 | Line 1381 | public class ForkJoinPool extends Abstra
1381		*         scheduled for execution
1382		*/
1383		public void execute(ForkJoinTask<?> task) {
1384	<	doSubmit(task);
1384	>	if (task == null)
1385	>	throw new NullPointerException();
1386	>	forkOrSubmit(task);
1387		}
1388
1389		// AbstractExecutorService methods
#	Line 1327 | Line 1394 | public class ForkJoinPool extends Abstra
1394		*         scheduled for execution
1395		*/
1396		public void execute(Runnable task) {
1397	+	if (task == null)
1398	+	throw new NullPointerException();
1399		ForkJoinTask<?> job;
1400		if (task instanceof ForkJoinTask<?>) // avoid re-wrap
1401		job = (ForkJoinTask<?>) task;
1402		else
1403		job = ForkJoinTask.adapt(task, null);
1404	<	doSubmit(job);
1404	>	forkOrSubmit(job);
1405		}
1406
1407		/**
#	Line 1345 | Line 1414 | public class ForkJoinPool extends Abstra
1414		*         scheduled for execution
1415		*/
1416		public <T> ForkJoinTask<T> submit(ForkJoinTask<T> task) {
1417	<	doSubmit(task);
1417	>	if (task == null)
1418	>	throw new NullPointerException();
1419	>	forkOrSubmit(task);
1420		return task;
1421		}
1422
#	Line 1355 | Line 1426 | public class ForkJoinPool extends Abstra
1426		*         scheduled for execution
1427		*/
1428		public <T> ForkJoinTask<T> submit(Callable<T> task) {
1429	+	if (task == null)
1430	+	throw new NullPointerException();
1431		ForkJoinTask<T> job = ForkJoinTask.adapt(task);
1432	<	doSubmit(job);
1432	>	forkOrSubmit(job);
1433		return job;
1434		}
1435
#	Line 1366 | Line 1439 | public class ForkJoinPool extends Abstra
1439		*         scheduled for execution
1440		*/
1441		public <T> ForkJoinTask<T> submit(Runnable task, T result) {
1442	+	if (task == null)
1443	+	throw new NullPointerException();
1444		ForkJoinTask<T> job = ForkJoinTask.adapt(task, result);
1445	<	doSubmit(job);
1445	>	forkOrSubmit(job);
1446		return job;
1447		}
1448
#	Line 1377 | Line 1452 | public class ForkJoinPool extends Abstra
1452		*         scheduled for execution
1453		*/
1454		public ForkJoinTask<?> submit(Runnable task) {
1455	+	if (task == null)
1456	+	throw new NullPointerException();
1457		ForkJoinTask<?> job;
1458		if (task instanceof ForkJoinTask<?>) // avoid re-wrap
1459		job = (ForkJoinTask<?>) task;
1460		else
1461		job = ForkJoinTask.adapt(task, null);
1462	<	doSubmit(job);
1462	>	forkOrSubmit(job);
1463		return job;
1464		}
1465
#	Line 1442 | Line 1519 | public class ForkJoinPool extends Abstra
1519
1520		/**
1521		* Returns the number of worker threads that have started but not
1522	<	* yet terminated.  This result returned by this method may differ
1522	>	* yet terminated.  The result returned by this method may differ
1523		* from {@link #getParallelism} when threads are created to
1524		* maintain parallelism when others are cooperatively blocked.
1525		*
#	Line 1527 | Line 1604 | public class ForkJoinPool extends Abstra
1604		*/
1605		public long getQueuedTaskCount() {
1606		long count = 0;
1607	<	ForkJoinWorkerThread[] ws = workers;
1531	<	int n = ws.length;
1532	<	for (int i = 0; i < n; ++i) {
1533	<	ForkJoinWorkerThread w = ws[i];
1607	>	for (ForkJoinWorkerThread w : workers)
1608		if (w != null)
1609		count += w.getQueueSize();
1536	–	}
1610		return count;
1611		}
1612
#	Line 1588 | Line 1661 | public class ForkJoinPool extends Abstra
1661		*/
1662		protected int drainTasksTo(Collection<? super ForkJoinTask<?>> c) {
1663		int count = submissionQueue.drainTo(c);
1664	<	ForkJoinWorkerThread[] ws = workers;
1592	<	int n = ws.length;
1593	<	for (int i = 0; i < n; ++i) {
1594	<	ForkJoinWorkerThread w = ws[i];
1664	>	for (ForkJoinWorkerThread w : workers)
1665		if (w != null)
1666		count += w.drainTasksTo(c);
1597	–	}
1667		return count;
1668		}
1669
#	Line 1698 | Line 1767 | public class ForkJoinPool extends Abstra
1767		}
1768
1769		/**
1770	+	* Returns true if terminating or terminated. Used by ForkJoinWorkerThread.
1771	+	*/
1772	+	final boolean isAtLeastTerminating() {
1773	+	return runState >= TERMINATING;
1774	+	}
1775	+
1776	+	/**
1777		* Returns {@code true} if this pool has been shut down.
1778		*
1779		* @return {@code true} if this pool has been shut down
#	Line 1720 | Line 1796 | public class ForkJoinPool extends Abstra
1796		public boolean awaitTermination(long timeout, TimeUnit unit)
1797		throws InterruptedException {
1798		try {
1799	<	return termination.awaitAdvanceInterruptibly(0, timeout, unit) > 0;
1800	<	} catch(TimeoutException ex) {
1799	>	termination.awaitAdvanceInterruptibly(0, timeout, unit);
1800	>	} catch (TimeoutException ex) {
1801		return false;
1802		}
1803	+	return true;
1804		}
1805
1806		/**
#	Line 1851 | Line 1928 | public class ForkJoinPool extends Abstra
1928		private static final long eventCountOffset =
1929		objectFieldOffset("eventCount", ForkJoinPool.class);
1930		private static final long eventWaitersOffset =
1931	<	objectFieldOffset("eventWaiters",ForkJoinPool.class);
1931	>	objectFieldOffset("eventWaiters", ForkJoinPool.class);
1932		private static final long stealCountOffset =
1933	<	objectFieldOffset("stealCount",ForkJoinPool.class);
1933	>	objectFieldOffset("stealCount", ForkJoinPool.class);
1934		private static final long spareWaitersOffset =
1935	<	objectFieldOffset("spareWaiters",ForkJoinPool.class);
1935	>	objectFieldOffset("spareWaiters", ForkJoinPool.class);
1936
1937		private static long objectFieldOffset(String field, Class<?> klazz) {
1938		try {

Diff Legend

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