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.86 by dl, Sun Nov 21 20:42:18 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 590 | Line 596 | public class ForkJoinPool extends Abstra
596		}
597
598		/**
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() {
#	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 (joinMe.status < 0)
1062	>	break;
1063	>	if ((workerCounts & RUNNING_COUNT_MASK) != 0) {
1064	>	long ms; int ns;
1065	>	if (!timed) {
1066	>	ms = JOIN_TIMEOUT_MILLIS;
1067	>	ns = 0;
1068	>	}
1069	>	else { // at most JOIN_TIMEOUT_MILLIS per wait
1070	>	long nt = nanos - (System.nanoTime() - startTime);
1071	>	if (nt <= 0L)
1072	>	break;
1073	>	ms = nt / 1000000;
1074	>	if (ms > JOIN_TIMEOUT_MILLIS) {
1075	>	ms = JOIN_TIMEOUT_MILLIS;
1076	>	ns = 0;
1077	>	}
1078	>	else
1079	>	ns = (int) (nt % 1000000);
1080	>	}
1081	>	if (joinMe.internalAwaitDone(ms, ns) < 0)
1082	>	break;
1083	>	}
1084		}
1085	+	helpMaintainParallelism();
1086	+	}
1087	+	if (!running) {
1088	+	int c;
1089	+	do {} while (!UNSAFE.compareAndSwapInt
1090	+	(this, workerCountsOffset,
1091	+	c = workerCounts, c + ONE_RUNNING));
1092		}
1093		}
1094
#	Line 1038 | Line 1099 | public class ForkJoinPool extends Abstra
1099		throws InterruptedException {
1100		while (!blocker.isReleasable()) {
1101		int wc = workerCounts;
1102	<	if ((wc & RUNNING_COUNT_MASK) != 0 &&
1103	<	UNSAFE.compareAndSwapInt(this, workerCountsOffset,
1104	<	wc, wc - ONE_RUNNING)) {
1102	>	if ((wc & RUNNING_COUNT_MASK) == 0)
1103	>	helpMaintainParallelism();
1104	>	else if (UNSAFE.compareAndSwapInt(this, workerCountsOffset,
1105	>	wc, wc - ONE_RUNNING)) {
1106		try {
1107		while (!blocker.isReleasable()) {
1108		long h = eventWaiters;
#	Line 1085 | Line 1147 | public class ForkJoinPool extends Abstra
1147		// Finish now if all threads terminated; else in some subsequent call
1148		if ((workerCounts >>> TOTAL_COUNT_SHIFT) == 0) {
1149		advanceRunLevel(TERMINATED);
1150	<	termination.arrive();
1150	>	termination.forceTermination();
1151		}
1152		return true;
1153		}
#	Line 1107 | Line 1169 | public class ForkJoinPool extends Abstra
1169		c = eventCount, c+1);
1170		eventWaiters = 0L; // clobber lists
1171		spareWaiters = 0;
1172	<	ForkJoinWorkerThread[] ws = workers;
1111	<	int n = ws.length;
1112	<	for (int i = 0; i < n; ++i) {
1113	<	ForkJoinWorkerThread w = ws[i];
1172	>	for (ForkJoinWorkerThread w : workers) {
1173		if (w != null) {
1174		w.shutdown();
1175		if (passes > 0 && !w.isTerminated()) {
1176		w.cancelTasks();
1177		LockSupport.unpark(w);
1178	<	if (passes > 1) {
1178	>	if (passes > 1 && !w.isInterrupted()) {
1179		try {
1180		w.interrupt();
1181		} catch (SecurityException ignore) {
#	Line 1129 | Line 1188 | public class ForkJoinPool extends Abstra
1188		}
1189
1190		/**
1191	<	* Clear out and cancel submissions, ignoring exceptions
1191	>	* Clears out and cancels submissions, ignoring exceptions.
1192		*/
1193		private void cancelSubmissions() {
1194		ForkJoinTask<?> task;
#	Line 1144 | Line 1203 | public class ForkJoinPool extends Abstra
1203		// misc support for ForkJoinWorkerThread
1204
1205		/**
1206	<	* Returns pool number
1206	>	* Returns pool number.
1207		*/
1208		final int getPoolNumber() {
1209		return poolNumber;
1210		}
1211
1212		/**
1213	<	* Tries to accumulates steal count from a worker, clearing
1214	<	* the worker's value.
1213	>	* Tries to accumulate steal count from a worker, clearing
1214	>	* the worker's value if successful.
1215		*
1216		* @return true if worker steal count now zero
1217		*/
#	Line 1176 | Line 1235 | public class ForkJoinPool extends Abstra
1235		int pc = parallelism; // use parallelism, not rc
1236		int ac = runState;    // no mask -- artificially boosts during shutdown
1237		// Use exact results for small values, saturate past 4
1238	<	return pc <= ac? 0 : pc >>> 1 <= ac? 1 : pc >>> 2 <= ac? 3 : pc >>> 3;
1238	>	return ((pc <= ac) ? 0 :
1239	>	(pc >>> 1 <= ac) ? 1 :
1240	>	(pc >>> 2 <= ac) ? 3 :
1241	>	pc >>> 3);
1242		}
1243
1244		// Public and protected methods
#	Line 1226 | Line 1288 | public class ForkJoinPool extends Abstra
1288		* use {@link #defaultForkJoinWorkerThreadFactory}.
1289		* @param handler the handler for internal worker threads that
1290		* terminate due to unrecoverable errors encountered while executing
1291	<	* tasks. For default value, use <code>null</code>.
1291	>	* tasks. For default value, use {@code null}.
1292		* @param asyncMode if true,
1293		* establishes local first-in-first-out scheduling mode for forked
1294		* tasks that are never joined. This mode may be more appropriate
1295		* than default locally stack-based mode in applications in which
1296		* worker threads only process event-style asynchronous tasks.
1297	<	* For default value, use <code>false</code>.
1297	>	* For default value, use {@code false}.
1298		* @throws IllegalArgumentException if parallelism less than or
1299		*         equal to zero, or greater than implementation limit
1300		* @throws NullPointerException if the factory is null
#	Line 1280 | Line 1342 | public class ForkJoinPool extends Abstra
1342		// Execution methods
1343
1344		/**
1345	<	* Common code for execute, invoke and submit
1345	>	* Submits task and creates, starts, or resumes some workers if necessary
1346		*/
1347		private <T> void doSubmit(ForkJoinTask<T> task) {
1286	–	if (task == null)
1287	–	throw new NullPointerException();
1288	–	if (runState >= SHUTDOWN)
1289	–	throw new RejectedExecutionException();
1348		submissionQueue.offer(task);
1349		int c; // try to increment event count -- CAS failure OK
1350		UNSAFE.compareAndSwapInt(this, eventCountOffset, c = eventCount, c+1);
1351	<	helpMaintainParallelism(); // create, start, or resume some workers
1351	>	helpMaintainParallelism();
1352		}
1353
1354		/**
#	Line 1303 | Line 1361 | public class ForkJoinPool extends Abstra
1361		*         scheduled for execution
1362		*/
1363		public <T> T invoke(ForkJoinTask<T> task) {
1364	<	doSubmit(task);
1365	<	return task.join();
1364	>	if (task == null)
1365	>	throw new NullPointerException();
1366	>	if (runState >= SHUTDOWN)
1367	>	throw new RejectedExecutionException();
1368	>	Thread t = Thread.currentThread();
1369	>	if ((t instanceof ForkJoinWorkerThread) &&
1370	>	((ForkJoinWorkerThread)t).pool == this)
1371	>	return task.invoke();  // bypass submit if in same pool
1372	>	else {
1373	>	doSubmit(task);
1374	>	return task.join();
1375	>	}
1376	>	}
1377	>
1378	>	/**
1379	>	* Unless terminating, forks task if within an ongoing FJ
1380	>	* computation in the current pool, else submits as external task.
1381	>	*/
1382	>	private <T> void forkOrSubmit(ForkJoinTask<T> task) {
1383	>	if (runState >= SHUTDOWN)
1384	>	throw new RejectedExecutionException();
1385	>	Thread t = Thread.currentThread();
1386	>	if ((t instanceof ForkJoinWorkerThread) &&
1387	>	((ForkJoinWorkerThread)t).pool == this)
1388	>	task.fork();
1389	>	else
1390	>	doSubmit(task);
1391		}
1392
1393		/**
#	Line 1316 | Line 1399 | public class ForkJoinPool extends Abstra
1399		*         scheduled for execution
1400		*/
1401		public void execute(ForkJoinTask<?> task) {
1402	<	doSubmit(task);
1402	>	if (task == null)
1403	>	throw new NullPointerException();
1404	>	forkOrSubmit(task);
1405		}
1406
1407		// AbstractExecutorService methods
#	Line 1327 | Line 1412 | public class ForkJoinPool extends Abstra
1412		*         scheduled for execution
1413		*/
1414		public void execute(Runnable task) {
1415	+	if (task == null)
1416	+	throw new NullPointerException();
1417		ForkJoinTask<?> job;
1418		if (task instanceof ForkJoinTask<?>) // avoid re-wrap
1419		job = (ForkJoinTask<?>) task;
1420		else
1421		job = ForkJoinTask.adapt(task, null);
1422	<	doSubmit(job);
1422	>	forkOrSubmit(job);
1423		}
1424
1425		/**
#	Line 1345 | Line 1432 | public class ForkJoinPool extends Abstra
1432		*         scheduled for execution
1433		*/
1434		public <T> ForkJoinTask<T> submit(ForkJoinTask<T> task) {
1435	<	doSubmit(task);
1435	>	if (task == null)
1436	>	throw new NullPointerException();
1437	>	forkOrSubmit(task);
1438		return task;
1439		}
1440
#	Line 1355 | Line 1444 | public class ForkJoinPool extends Abstra
1444		*         scheduled for execution
1445		*/
1446		public <T> ForkJoinTask<T> submit(Callable<T> task) {
1447	+	if (task == null)
1448	+	throw new NullPointerException();
1449		ForkJoinTask<T> job = ForkJoinTask.adapt(task);
1450	<	doSubmit(job);
1450	>	forkOrSubmit(job);
1451		return job;
1452		}
1453
#	Line 1366 | Line 1457 | public class ForkJoinPool extends Abstra
1457		*         scheduled for execution
1458		*/
1459		public <T> ForkJoinTask<T> submit(Runnable task, T result) {
1460	+	if (task == null)
1461	+	throw new NullPointerException();
1462		ForkJoinTask<T> job = ForkJoinTask.adapt(task, result);
1463	<	doSubmit(job);
1463	>	forkOrSubmit(job);
1464		return job;
1465		}
1466
#	Line 1377 | Line 1470 | public class ForkJoinPool extends Abstra
1470		*         scheduled for execution
1471		*/
1472		public ForkJoinTask<?> submit(Runnable task) {
1473	+	if (task == null)
1474	+	throw new NullPointerException();
1475		ForkJoinTask<?> job;
1476		if (task instanceof ForkJoinTask<?>) // avoid re-wrap
1477		job = (ForkJoinTask<?>) task;
1478		else
1479		job = ForkJoinTask.adapt(task, null);
1480	<	doSubmit(job);
1480	>	forkOrSubmit(job);
1481		return job;
1482		}
1483
#	Line 1442 | Line 1537 | public class ForkJoinPool extends Abstra
1537
1538		/**
1539		* Returns the number of worker threads that have started but not
1540	<	* yet terminated.  This result returned by this method may differ
1540	>	* yet terminated.  The result returned by this method may differ
1541		* from {@link #getParallelism} when threads are created to
1542		* maintain parallelism when others are cooperatively blocked.
1543		*
#	Line 1527 | Line 1622 | public class ForkJoinPool extends Abstra
1622		*/
1623		public long getQueuedTaskCount() {
1624		long count = 0;
1625	<	ForkJoinWorkerThread[] ws = workers;
1531	<	int n = ws.length;
1532	<	for (int i = 0; i < n; ++i) {
1533	<	ForkJoinWorkerThread w = ws[i];
1625	>	for (ForkJoinWorkerThread w : workers)
1626		if (w != null)
1627		count += w.getQueueSize();
1536	–	}
1628		return count;
1629		}
1630
#	Line 1588 | Line 1679 | public class ForkJoinPool extends Abstra
1679		*/
1680		protected int drainTasksTo(Collection<? super ForkJoinTask<?>> c) {
1681		int count = submissionQueue.drainTo(c);
1682	<	ForkJoinWorkerThread[] ws = workers;
1592	<	int n = ws.length;
1593	<	for (int i = 0; i < n; ++i) {
1594	<	ForkJoinWorkerThread w = ws[i];
1682	>	for (ForkJoinWorkerThread w : workers)
1683		if (w != null)
1684		count += w.drainTasksTo(c);
1597	–	}
1685		return count;
1686		}
1687
#	Line 1698 | Line 1785 | public class ForkJoinPool extends Abstra
1785		}
1786
1787		/**
1788	+	* Returns true if terminating or terminated. Used by ForkJoinWorkerThread.
1789	+	*/
1790	+	final boolean isAtLeastTerminating() {
1791	+	return runState >= TERMINATING;
1792	+	}
1793	+
1794	+	/**
1795		* Returns {@code true} if this pool has been shut down.
1796		*
1797		* @return {@code true} if this pool has been shut down
#	Line 1720 | Line 1814 | public class ForkJoinPool extends Abstra
1814		public boolean awaitTermination(long timeout, TimeUnit unit)
1815		throws InterruptedException {
1816		try {
1817	<	return termination.awaitAdvanceInterruptibly(0, timeout, unit) > 0;
1818	<	} catch(TimeoutException ex) {
1817	>	termination.awaitAdvanceInterruptibly(0, timeout, unit);
1818	>	} catch (TimeoutException ex) {
1819		return false;
1820		}
1821	+	return true;
1822		}
1823
1824		/**
#	Line 1851 | Line 1946 | public class ForkJoinPool extends Abstra
1946		private static final long eventCountOffset =
1947		objectFieldOffset("eventCount", ForkJoinPool.class);
1948		private static final long eventWaitersOffset =
1949	<	objectFieldOffset("eventWaiters",ForkJoinPool.class);
1949	>	objectFieldOffset("eventWaiters", ForkJoinPool.class);
1950		private static final long stealCountOffset =
1951	<	objectFieldOffset("stealCount",ForkJoinPool.class);
1951	>	objectFieldOffset("stealCount", ForkJoinPool.class);
1952		private static final long spareWaitersOffset =
1953	<	objectFieldOffset("spareWaiters",ForkJoinPool.class);
1953	>	objectFieldOffset("spareWaiters", ForkJoinPool.class);
1954
1955		private static long objectFieldOffset(String field, Class<?> klazz) {
1956		try {

Diff Legend

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