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.67 by jsr166, Wed Sep 1 03:32:03 2010 UTC vs.
Revision 1.88 by dl, Tue Nov 23 01:06:00 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 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 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 318 | Line 323 | public class ForkJoinPool extends Abstra
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 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 490 | Line 496 | public class ForkJoinPool extends Abstra
496		*/
497		private volatile long eventWaiters;
498
499	<	private static final int  EVENT_COUNT_SHIFT = 32;
500	<	private static final long WAITER_ID_MASK    = (1L << 16) - 1L;
499	>	private static final int EVENT_COUNT_SHIFT = 32;
500	>	private static final int WAITER_ID_MASK    = (1 << 16) - 1;
501
502		/**
503		* A counter for events that may wake up worker threads:
#	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 580 | Line 586 | public class ForkJoinPool extends Abstra
586		// are usually manually inlined by callers
587
588		/**
589	<	* Increments running count part of workerCounts
589	>	* Increments running count part of workerCounts.
590		*/
591		final void incrementRunningCount() {
592		int c;
#	Line 590 | Line 596 | public class ForkJoinPool extends Abstra
596		}
597
598		/**
599	<	* Tries to decrement running count unless already zero
599	>	* Tries to increment running count part of workerCounts.
600	>	*/
601	>	final boolean tryIncrementRunningCount() {
602	>	int c;
603	>	return UNSAFE.compareAndSwapInt(this, workerCountsOffset,
604	>	c = workerCounts,
605	>	c + ONE_RUNNING);
606	>	}
607	>
608	>	/**
609	>	* Tries to decrement running count unless already zero.
610		*/
611		final boolean tryDecrementRunningCount() {
612		int wc = workerCounts;
#	Line 605 | Line 621 | public class ForkJoinPool extends Abstra
621		* (rarely) necessary when other count updates lag.
622		*
623		* @param dr -- either zero or ONE_RUNNING
624	<	* @param dt == either zero or ONE_TOTAL
624	>	* @param dt -- either zero or ONE_TOTAL
625		*/
626		private void decrementWorkerCounts(int dr, int dt) {
627		for (;;) {
#	Line 663 | Line 679 | public class ForkJoinPool extends Abstra
679		for (k = 0; k < n && ws[k] != null; ++k)
680		;
681		if (k == n)
682	<	ws = Arrays.copyOf(ws, n << 1);
682	>	ws = workers = Arrays.copyOf(ws, n << 1);
683		}
684		ws[k] = w;
685	<	workers = ws; // volatile array write ensures slot visibility
685	>	int c = eventCount; // advance event count to ensure visibility
686	>	UNSAFE.compareAndSwapInt(this, eventCountOffset, c, c+1);
687		} finally {
688		lock.unlock();
689		}
#	Line 674 | Line 691 | public class ForkJoinPool extends Abstra
691		}
692
693		/**
694	<	* Nulls out record of worker in workers array
694	>	* Nulls out record of worker in workers array.
695		*/
696		private void forgetWorker(ForkJoinWorkerThread w) {
697		int idx = w.poolIndex;
#	Line 699 | Line 716 | public class ForkJoinPool extends Abstra
716		*/
717		final void workerTerminated(ForkJoinWorkerThread w) {
718		forgetWorker(w);
719	<	decrementWorkerCounts(w.isTrimmed()? 0 : ONE_RUNNING, ONE_TOTAL);
719	>	decrementWorkerCounts(w.isTrimmed() ? 0 : ONE_RUNNING, ONE_TOTAL);
720		while (w.stealCount != 0) // collect final count
721		tryAccumulateStealCount(w);
722		tryTerminate(false);
#	Line 720 | Line 737 | public class ForkJoinPool extends Abstra
737		int ec = eventCount;
738		boolean releasedOne = false;
739		ForkJoinWorkerThread w; int id;
740	<	while ((id = ((int)(h & WAITER_ID_MASK)) - 1) >= 0 &&
740	>	while ((id = (((int)h) & WAITER_ID_MASK) - 1) >= 0 &&
741		(int)(h >>> EVENT_COUNT_SHIFT) != ec &&
742		id < n && (w = ws[id]) != null) {
743		if (UNSAFE.compareAndSwapLong(this, eventWaitersOffset,
#	Line 758 | Line 775 | public class ForkJoinPool extends Abstra
775		long nh = (((long)ec) << EVENT_COUNT_SHIFT) | ((long)(w.poolIndex+1));
776		long h;
777		while ((runState < SHUTDOWN || !tryTerminate(false)) &&
778	<	(((int)((h = eventWaiters) & WAITER_ID_MASK)) == 0 ||
778	>	(((int)(h = eventWaiters) & WAITER_ID_MASK) == 0 ||
779		(int)(h >>> EVENT_COUNT_SHIFT) == ec) &&
780		eventCount == ec) {
781		if (UNSAFE.compareAndSwapLong(this, eventWaitersOffset,
#	Line 785 | Line 802 | public class ForkJoinPool extends Abstra
802		if (tryAccumulateStealCount(w)) { // transfer while idle
803		boolean untimed = (w.nextWaiter != 0L ||
804		(workerCounts & RUNNING_COUNT_MASK) <= 1);
805	<	long startTime = untimed? 0 : System.nanoTime();
805	>	long startTime = untimed ? 0 : System.nanoTime();
806		Thread.interrupted();         // clear/ignore interrupt
807	<	if (eventCount != ec || w.runState != 0 ||
808	<	runState >= TERMINATING)  // recheck after clear
792	<	break;
807	>	if (w.isTerminating() || eventCount != ec)
808	>	break;                    // recheck after clear
809		if (untimed)
810		LockSupport.park(w);
811		else {
812		LockSupport.parkNanos(w, SHRINK_RATE_NANOS);
813	<	if (eventCount != ec || w.runState != 0 ||
798	<	runState >= TERMINATING)
813	>	if (eventCount != ec || w.isTerminating())
814		break;
815		if (System.nanoTime() - startTime >= SHRINK_RATE_NANOS)
816		tryShutdownUnusedWorker(ec);
#	Line 807 | Line 822 | public class ForkJoinPool extends Abstra
822		// Maintaining parallelism
823
824		/**
825	<	* Pushes worker onto the spare stack
825	>	* Pushes worker onto the spare stack.
826		*/
827		final void pushSpare(ForkJoinWorkerThread w) {
828		int ns = (++w.spareCount << SPARE_COUNT_SHIFT) | (w.poolIndex + 1);
#	Line 827 | Line 842 | public class ForkJoinPool extends Abstra
842		if ((sw = spareWaiters) != 0 &&
843		(id = (sw & SPARE_ID_MASK) - 1) >= 0 &&
844		id < n && (w = ws[id]) != null &&
845	<	(workerCounts & RUNNING_COUNT_MASK) < parallelism &&
845	>	(runState >= TERMINATING ||
846	>	(workerCounts & RUNNING_COUNT_MASK) < parallelism) &&
847		spareWaiters == sw &&
848		UNSAFE.compareAndSwapInt(this, spareWaitersOffset,
849		sw, w.nextSpare)) {
850		int c; // increment running count before resume
851	<	do {} while(!UNSAFE.compareAndSwapInt
852	<	(this, workerCountsOffset,
853	<	c = workerCounts, c + ONE_RUNNING));
851	>	do {} while (!UNSAFE.compareAndSwapInt
852	>	(this, workerCountsOffset,
853	>	c = workerCounts, c + ONE_RUNNING));
854		if (w.tryUnsuspend())
855		LockSupport.unpark(w);
856		else   // back out if w was shutdown
#	Line 863 | Line 879 | public class ForkJoinPool extends Abstra
879		UNSAFE.compareAndSwapInt(this, workerCountsOffset, wc,
880		wc + (ONE_RUNNING|ONE_TOTAL))) {
881		ForkJoinWorkerThread w = null;
882	+	Throwable fail = null;
883		try {
884		w = factory.newThread(this);
885	<	} finally { // adjust on null or exceptional factory return
886	<	if (w == null) {
870	<	decrementWorkerCounts(ONE_RUNNING, ONE_TOTAL);
871	<	tryTerminate(false); // handle failure during shutdown
872	<	}
885	>	} catch (Throwable ex) {
886	>	fail = ex;
887		}
888	<	if (w == null)
888	>	if (w == null) { // null or exceptional factory return
889	>	decrementWorkerCounts(ONE_RUNNING, ONE_TOTAL);
890	>	tryTerminate(false); // handle failure during shutdown
891	>	// If originating from an external caller,
892	>	// propagate exception, else ignore
893	>	if (fail != null && runState < TERMINATING &&
894	>	!(Thread.currentThread() instanceof
895	>	ForkJoinWorkerThread))
896	>	UNSAFE.throwException(fail);
897		break;
898	+	}
899		w.start(recordWorker(w), ueh);
900	<	if ((workerCounts >>> TOTAL_COUNT_SHIFT) >= pc) {
878	<	int c; // advance event count
879	<	UNSAFE.compareAndSwapInt(this, eventCountOffset,
880	<	c = eventCount, c+1);
900	>	if ((workerCounts >>> TOTAL_COUNT_SHIFT) >= pc)
901		break; // add at most one unless total below target
882	–	}
902		}
903		}
904		if (eventWaiters != 0L)
#	Line 915 | Line 934 | public class ForkJoinPool extends Abstra
934		}
935		else if ((h = eventWaiters) != 0L) {
936		long nh;
937	<	int id = ((int)(h & WAITER_ID_MASK)) - 1;
937	>	int id = (((int)h) & WAITER_ID_MASK) - 1;
938		if (id >= 0 && id < n && (w = ws[id]) != null &&
939		(nh = w.nextWaiter) != 0L && // keep at least one worker
940		UNSAFE.compareAndSwapLong(this, eventWaitersOffset, h, nh))
#	Line 961 | Line 980 | public class ForkJoinPool extends Abstra
980		boolean active = w.active;
981		boolean inactivate = false;
982		int pc = parallelism;
983	<	int rs;
984	<	while (w.runState == 0 && (rs = runState) < TERMINATING) {
983	>	while (w.runState == 0) {
984	>	int rs = runState;
985	>	if (rs >= TERMINATING) {           // propagate shutdown
986	>	w.shutdown();
987	>	break;
988	>	}
989		if ((inactivate || (active && (rs & ACTIVE_COUNT_MASK) >= pc)) &&
990	<	UNSAFE.compareAndSwapInt(this, runStateOffset, rs, rs - 1))
990	>	UNSAFE.compareAndSwapInt(this, runStateOffset, rs, --rs)) {
991		inactivate = active = w.active = false;
992	<	int wc = workerCounts;
992	>	if (rs == SHUTDOWN) {          // all inactive and shut down
993	>	tryTerminate(false);
994	>	continue;
995	>	}
996	>	}
997	>	int wc = workerCounts;             // try to suspend as spare
998		if ((wc & RUNNING_COUNT_MASK) > pc) {
999		if (!(inactivate |= active) && // must inactivate to suspend
1000	<	workerCounts == wc &&      // try to suspend as spare
1000	>	workerCounts == wc &&
1001		UNSAFE.compareAndSwapInt(this, workerCountsOffset,
1002		wc, wc - ONE_RUNNING))
1003		w.suspendAsSpare();
1004		}
1005		else if ((wc >>> TOTAL_COUNT_SHIFT) < pc)
1006		helpMaintainParallelism();     // not enough workers
1007	<	else if (!ran) {
1007	>	else if (ran)
1008	>	break;
1009	>	else {
1010		long h = eventWaiters;
1011		int ec = eventCount;
1012		if (h != 0L && (int)(h >>> EVENT_COUNT_SHIFT) != ec)
#	Line 985 | Line 1015 | public class ForkJoinPool extends Abstra
1015		w.lastEventCount = ec;     // no need to wait
1016		break;
1017		}
1018	<	else if (!(inactivate |= active))
1018	>	else if (!(inactivate |= active))
1019		eventSync(w, wec);         // must inactivate before sync
1020		}
991	–	else
992	–	break;
1021		}
1022		}
1023
#	Line 999 | Line 1027 | public class ForkJoinPool extends Abstra
1027		*
1028		* @param joinMe the task to join
1029		* @param worker the current worker thread
1030	+	* @param timed true if wait should time out
1031	+	* @param nanos timeout value if timed
1032		*/
1033	<	final void awaitJoin(ForkJoinTask<?> joinMe, ForkJoinWorkerThread worker) {
1033	>	final void awaitJoin(ForkJoinTask<?> joinMe, ForkJoinWorkerThread worker,
1034	>	boolean timed, long nanos) {
1035	>	long startTime = timed? System.nanoTime() : 0L;
1036		int retries = 2 + (parallelism >> 2); // #helpJoins before blocking
1037	+	boolean running = true;               // false when count decremented
1038		while (joinMe.status >= 0) {
1039	<	int wc;
1040	<	worker.helpJoinTask(joinMe);
1039	>	if (runState >= TERMINATING) {
1040	>	joinMe.cancelIgnoringExceptions();
1041	>	break;
1042	>	}
1043	>	running = worker.helpJoinTask(joinMe, running);
1044		if (joinMe.status < 0)
1045		break;
1046	<	else if (retries > 0)
1046	>	if (retries > 0) {
1047		--retries;
1048	<	else if (((wc = workerCounts) & RUNNING_COUNT_MASK) != 0 &&
1049	<	UNSAFE.compareAndSwapInt(this, workerCountsOffset,
1050	<	wc, wc - ONE_RUNNING)) {
1051	<	int stat, c; long h;
1052	<	while ((stat = joinMe.status) >= 0 &&
1053	<	(h = eventWaiters) != 0L && // help release others
1054	<	(int)(h >>> EVENT_COUNT_SHIFT) != eventCount)
1048	>	continue;
1049	>	}
1050	>	int wc = workerCounts;
1051	>	if ((wc & RUNNING_COUNT_MASK) != 0) {
1052	>	if (running) {
1053	>	if (!UNSAFE.compareAndSwapInt(this, workerCountsOffset,
1054	>	wc, wc - ONE_RUNNING))
1055	>	continue;
1056	>	running = false;
1057	>	}
1058	>	long h = eventWaiters;
1059	>	if (h != 0L && (int)(h >>> EVENT_COUNT_SHIFT) != eventCount)
1060		releaseEventWaiters();
1061	<	if (stat >= 0 &&
1062	<	((workerCounts & RUNNING_COUNT_MASK) == 0 ||
1063	<	(stat =
1064	<	joinMe.internalAwaitDone(JOIN_TIMEOUT_MILLIS)) >= 0))
1065	<	helpMaintainParallelism(); // timeout or no running workers
1066	<	do {} while (!UNSAFE.compareAndSwapInt
1067	<	(this, workerCountsOffset,
1068	<	c = workerCounts, c + ONE_RUNNING));
1069	<	if (stat < 0)
1070	<	break;   // else restart
1061	>	if ((workerCounts & RUNNING_COUNT_MASK) != 0) {
1062	>	long ms; int ns;
1063	>	if (!timed) {
1064	>	ms = JOIN_TIMEOUT_MILLIS;
1065	>	ns = 0;
1066	>	}
1067	>	else { // at most JOIN_TIMEOUT_MILLIS per wait
1068	>	long nt = nanos - (System.nanoTime() - startTime);
1069	>	if (nt <= 0L)
1070	>	break;
1071	>	ms = nt / 1000000;
1072	>	if (ms > JOIN_TIMEOUT_MILLIS) {
1073	>	ms = JOIN_TIMEOUT_MILLIS;
1074	>	ns = 0;
1075	>	}
1076	>	else
1077	>	ns = (int) (nt % 1000000);
1078	>	}
1079	>	joinMe.internalAwaitDone(ms, ns);
1080	>	}
1081	>	if (joinMe.status < 0)
1082	>	break;
1083		}
1084	+	helpMaintainParallelism();
1085	+	}
1086	+	if (!running) {
1087	+	int c;
1088	+	do {} while (!UNSAFE.compareAndSwapInt
1089	+	(this, workerCountsOffset,
1090	+	c = workerCounts, c + ONE_RUNNING));
1091		}
1092		}
1093
#	Line 1038 | Line 1098 | public class ForkJoinPool extends Abstra
1098		throws InterruptedException {
1099		while (!blocker.isReleasable()) {
1100		int wc = workerCounts;
1101	<	if ((wc & RUNNING_COUNT_MASK) != 0 &&
1102	<	UNSAFE.compareAndSwapInt(this, workerCountsOffset,
1103	<	wc, wc - ONE_RUNNING)) {
1101	>	if ((wc & RUNNING_COUNT_MASK) == 0)
1102	>	helpMaintainParallelism();
1103	>	else if (UNSAFE.compareAndSwapInt(this, workerCountsOffset,
1104	>	wc, wc - ONE_RUNNING)) {
1105		try {
1106		while (!blocker.isReleasable()) {
1107		long h = eventWaiters;
#	Line 1085 | Line 1146 | public class ForkJoinPool extends Abstra
1146		// Finish now if all threads terminated; else in some subsequent call
1147		if ((workerCounts >>> TOTAL_COUNT_SHIFT) == 0) {
1148		advanceRunLevel(TERMINATED);
1149	<	termination.arrive();
1149	>	termination.forceTermination();
1150		}
1151		return true;
1152		}
#	Line 1107 | Line 1168 | public class ForkJoinPool extends Abstra
1168		c = eventCount, c+1);
1169		eventWaiters = 0L; // clobber lists
1170		spareWaiters = 0;
1171	<	ForkJoinWorkerThread[] ws = workers;
1111	<	int n = ws.length;
1112	<	for (int i = 0; i < n; ++i) {
1113	<	ForkJoinWorkerThread w = ws[i];
1171	>	for (ForkJoinWorkerThread w : workers) {
1172		if (w != null) {
1173		w.shutdown();
1174		if (passes > 0 && !w.isTerminated()) {
1175		w.cancelTasks();
1176		LockSupport.unpark(w);
1177	<	if (passes > 1) {
1177	>	if (passes > 1 && !w.isInterrupted()) {
1178		try {
1179		w.interrupt();
1180		} catch (SecurityException ignore) {
#	Line 1129 | Line 1187 | public class ForkJoinPool extends Abstra
1187		}
1188
1189		/**
1190	<	* Clear out and cancel submissions, ignoring exceptions
1190	>	* Clears out and cancels submissions, ignoring exceptions.
1191		*/
1192		private void cancelSubmissions() {
1193		ForkJoinTask<?> task;
#	Line 1144 | Line 1202 | public class ForkJoinPool extends Abstra
1202		// misc support for ForkJoinWorkerThread
1203
1204		/**
1205	<	* Returns pool number
1205	>	* Returns pool number.
1206		*/
1207		final int getPoolNumber() {
1208		return poolNumber;
1209		}
1210
1211		/**
1212	<	* Tries to accumulates steal count from a worker, clearing
1213	<	* the worker's value.
1212	>	* Tries to accumulate steal count from a worker, clearing
1213	>	* the worker's value if successful.
1214		*
1215		* @return true if worker steal count now zero
1216		*/
#	Line 1176 | Line 1234 | public class ForkJoinPool extends Abstra
1234		int pc = parallelism; // use parallelism, not rc
1235		int ac = runState;    // no mask -- artificially boosts during shutdown
1236		// Use exact results for small values, saturate past 4
1237	<	return pc <= ac? 0 : pc >>> 1 <= ac? 1 : pc >>> 2 <= ac? 3 : pc >>> 3;
1237	>	return ((pc <= ac) ? 0 :
1238	>	(pc >>> 1 <= ac) ? 1 :
1239	>	(pc >>> 2 <= ac) ? 3 :
1240	>	pc >>> 3);
1241		}
1242
1243		// Public and protected methods
#	Line 1226 | Line 1287 | public class ForkJoinPool extends Abstra
1287		* use {@link #defaultForkJoinWorkerThreadFactory}.
1288		* @param handler the handler for internal worker threads that
1289		* terminate due to unrecoverable errors encountered while executing
1290	<	* tasks. For default value, use <code>null</code>.
1290	>	* tasks. For default value, use {@code null}.
1291		* @param asyncMode if true,
1292		* establishes local first-in-first-out scheduling mode for forked
1293		* tasks that are never joined. This mode may be more appropriate
1294		* than default locally stack-based mode in applications in which
1295		* worker threads only process event-style asynchronous tasks.
1296	<	* For default value, use <code>false</code>.
1296	>	* For default value, use {@code false}.
1297		* @throws IllegalArgumentException if parallelism less than or
1298		*         equal to zero, or greater than implementation limit
1299		* @throws NullPointerException if the factory is null
#	Line 1280 | Line 1341 | public class ForkJoinPool extends Abstra
1341		// Execution methods
1342
1343		/**
1344	<	* Common code for execute, invoke and submit
1344	>	* Submits task and creates, starts, or resumes some workers if necessary
1345		*/
1346		private <T> void doSubmit(ForkJoinTask<T> task) {
1286	–	if (task == null)
1287	–	throw new NullPointerException();
1288	–	if (runState >= SHUTDOWN)
1289	–	throw new RejectedExecutionException();
1347		submissionQueue.offer(task);
1348		int c; // try to increment event count -- CAS failure OK
1349		UNSAFE.compareAndSwapInt(this, eventCountOffset, c = eventCount, c+1);
1350	<	helpMaintainParallelism(); // create, start, or resume some workers
1350	>	helpMaintainParallelism();
1351		}
1352
1353		/**
#	Line 1303 | Line 1360 | public class ForkJoinPool extends Abstra
1360		*         scheduled for execution
1361		*/
1362		public <T> T invoke(ForkJoinTask<T> task) {
1363	<	doSubmit(task);
1364	<	return task.join();
1363	>	if (task == null)
1364	>	throw new NullPointerException();
1365	>	if (runState >= SHUTDOWN)
1366	>	throw new RejectedExecutionException();
1367	>	Thread t = Thread.currentThread();
1368	>	if ((t instanceof ForkJoinWorkerThread) &&
1369	>	((ForkJoinWorkerThread)t).pool == this)
1370	>	return task.invoke();  // bypass submit if in same pool
1371	>	else {
1372	>	doSubmit(task);
1373	>	return task.join();
1374	>	}
1375	>	}
1376	>
1377	>	/**
1378	>	* Unless terminating, forks task if within an ongoing FJ
1379	>	* computation in the current pool, else submits as external task.
1380	>	*/
1381	>	private <T> void forkOrSubmit(ForkJoinTask<T> task) {
1382	>	if (runState >= SHUTDOWN)
1383	>	throw new RejectedExecutionException();
1384	>	Thread t = Thread.currentThread();
1385	>	if ((t instanceof ForkJoinWorkerThread) &&
1386	>	((ForkJoinWorkerThread)t).pool == this)
1387	>	task.fork();
1388	>	else
1389	>	doSubmit(task);
1390		}
1391
1392		/**
#	Line 1316 | Line 1398 | public class ForkJoinPool extends Abstra
1398		*         scheduled for execution
1399		*/
1400		public void execute(ForkJoinTask<?> task) {
1401	<	doSubmit(task);
1401	>	if (task == null)
1402	>	throw new NullPointerException();
1403	>	forkOrSubmit(task);
1404		}
1405
1406		// AbstractExecutorService methods
#	Line 1327 | Line 1411 | public class ForkJoinPool extends Abstra
1411		*         scheduled for execution
1412		*/
1413		public void execute(Runnable task) {
1414	+	if (task == null)
1415	+	throw new NullPointerException();
1416		ForkJoinTask<?> job;
1417		if (task instanceof ForkJoinTask<?>) // avoid re-wrap
1418		job = (ForkJoinTask<?>) task;
1419		else
1420		job = ForkJoinTask.adapt(task, null);
1421	<	doSubmit(job);
1421	>	forkOrSubmit(job);
1422		}
1423
1424		/**
#	Line 1345 | Line 1431 | public class ForkJoinPool extends Abstra
1431		*         scheduled for execution
1432		*/
1433		public <T> ForkJoinTask<T> submit(ForkJoinTask<T> task) {
1434	<	doSubmit(task);
1434	>	if (task == null)
1435	>	throw new NullPointerException();
1436	>	forkOrSubmit(task);
1437		return task;
1438		}
1439
#	Line 1355 | Line 1443 | public class ForkJoinPool extends Abstra
1443		*         scheduled for execution
1444		*/
1445		public <T> ForkJoinTask<T> submit(Callable<T> task) {
1446	+	if (task == null)
1447	+	throw new NullPointerException();
1448		ForkJoinTask<T> job = ForkJoinTask.adapt(task);
1449	<	doSubmit(job);
1449	>	forkOrSubmit(job);
1450		return job;
1451		}
1452
#	Line 1366 | Line 1456 | public class ForkJoinPool extends Abstra
1456		*         scheduled for execution
1457		*/
1458		public <T> ForkJoinTask<T> submit(Runnable task, T result) {
1459	+	if (task == null)
1460	+	throw new NullPointerException();
1461		ForkJoinTask<T> job = ForkJoinTask.adapt(task, result);
1462	<	doSubmit(job);
1462	>	forkOrSubmit(job);
1463		return job;
1464		}
1465
#	Line 1377 | Line 1469 | public class ForkJoinPool extends Abstra
1469		*         scheduled for execution
1470		*/
1471		public ForkJoinTask<?> submit(Runnable task) {
1472	+	if (task == null)
1473	+	throw new NullPointerException();
1474		ForkJoinTask<?> job;
1475		if (task instanceof ForkJoinTask<?>) // avoid re-wrap
1476		job = (ForkJoinTask<?>) task;
1477		else
1478		job = ForkJoinTask.adapt(task, null);
1479	<	doSubmit(job);
1479	>	forkOrSubmit(job);
1480		return job;
1481		}
1482
#	Line 1442 | Line 1536 | public class ForkJoinPool extends Abstra
1536
1537		/**
1538		* Returns the number of worker threads that have started but not
1539	<	* yet terminated.  This result returned by this method may differ
1539	>	* yet terminated.  The result returned by this method may differ
1540		* from {@link #getParallelism} when threads are created to
1541		* maintain parallelism when others are cooperatively blocked.
1542		*
#	Line 1527 | Line 1621 | public class ForkJoinPool extends Abstra
1621		*/
1622		public long getQueuedTaskCount() {
1623		long count = 0;
1624	<	ForkJoinWorkerThread[] ws = workers;
1531	<	int n = ws.length;
1532	<	for (int i = 0; i < n; ++i) {
1533	<	ForkJoinWorkerThread w = ws[i];
1624	>	for (ForkJoinWorkerThread w : workers)
1625		if (w != null)
1626		count += w.getQueueSize();
1536	–	}
1627		return count;
1628		}
1629
#	Line 1588 | Line 1678 | public class ForkJoinPool extends Abstra
1678		*/
1679		protected int drainTasksTo(Collection<? super ForkJoinTask<?>> c) {
1680		int count = submissionQueue.drainTo(c);
1681	<	ForkJoinWorkerThread[] ws = workers;
1592	<	int n = ws.length;
1593	<	for (int i = 0; i < n; ++i) {
1594	<	ForkJoinWorkerThread w = ws[i];
1681	>	for (ForkJoinWorkerThread w : workers)
1682		if (w != null)
1683		count += w.drainTasksTo(c);
1597	–	}
1684		return count;
1685		}
1686
#	Line 1688 | Line 1774 | public class ForkJoinPool extends Abstra
1774		* commenced but not yet completed.  This method may be useful for
1775		* debugging. A return of {@code true} reported a sufficient
1776		* period after shutdown may indicate that submitted tasks have
1777	<	* ignored or suppressed interruption, causing this executor not
1778	<	* to properly terminate.
1777	>	* ignored or suppressed interruption, or are waiting for IO,
1778	>	* causing this executor not to properly terminate. (See the
1779	>	* advisory notes for class {@link ForkJoinTask} stating that
1780	>	* tasks should not normally entail blocking operations.  But if
1781	>	* they do, they must abort them on interrupt.)
1782		*
1783		* @return {@code true} if terminating but not yet terminated
1784		*/
#	Line 1698 | Line 1787 | public class ForkJoinPool extends Abstra
1787		}
1788
1789		/**
1790	+	* Returns true if terminating or terminated. Used by ForkJoinWorkerThread.
1791	+	*/
1792	+	final boolean isAtLeastTerminating() {
1793	+	return runState >= TERMINATING;
1794	+	}
1795	+
1796	+	/**
1797		* Returns {@code true} if this pool has been shut down.
1798		*
1799		* @return {@code true} if this pool has been shut down
#	Line 1720 | Line 1816 | public class ForkJoinPool extends Abstra
1816		public boolean awaitTermination(long timeout, TimeUnit unit)
1817		throws InterruptedException {
1818		try {
1819	<	return termination.awaitAdvanceInterruptibly(0, timeout, unit) > 0;
1820	<	} catch(TimeoutException ex) {
1819	>	termination.awaitAdvanceInterruptibly(0, timeout, unit);
1820	>	} catch (TimeoutException ex) {
1821		return false;
1822		}
1823	+	return true;
1824		}
1825
1826		/**
#	Line 1851 | Line 1948 | public class ForkJoinPool extends Abstra
1948		private static final long eventCountOffset =
1949		objectFieldOffset("eventCount", ForkJoinPool.class);
1950		private static final long eventWaitersOffset =
1951	<	objectFieldOffset("eventWaiters",ForkJoinPool.class);
1951	>	objectFieldOffset("eventWaiters", ForkJoinPool.class);
1952		private static final long stealCountOffset =
1953	<	objectFieldOffset("stealCount",ForkJoinPool.class);
1953	>	objectFieldOffset("stealCount", ForkJoinPool.class);
1954		private static final long spareWaitersOffset =
1955	<	objectFieldOffset("spareWaiters",ForkJoinPool.class);
1955	>	objectFieldOffset("spareWaiters", ForkJoinPool.class);
1956
1957		private static long objectFieldOffset(String field, Class<?> klazz) {
1958		try {

Diff Legend

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