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.69 by jsr166, Wed Sep 1 20:12:39 2010 UTC vs.
Revision 1.89 by dl, Wed Nov 24 10:50:38 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 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 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 711 | Line 728 | public class ForkJoinPool extends Abstra
728		* Releases workers blocked on a count not equal to current count.
729		* Normally called after precheck that eventWaiters isn't zero to
730		* avoid wasted array checks. Gives up upon a change in count or
731	<	* upon releasing two workers, letting others take over.
731	>	* upon releasing four workers, letting others take over.
732		*/
733		private void releaseEventWaiters() {
734		ForkJoinWorkerThread[] ws = workers;
735		int n = ws.length;
736		long h = eventWaiters;
737		int ec = eventCount;
738	<	boolean releasedOne = false;
738	>	int releases = 4;
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,
744		h,  w.nextWaiter)) {
745		LockSupport.unpark(w);
746	<	if (releasedOne) // exit on second release
746	>	if (--releases == 0)
747		break;
731	–	releasedOne = true;
748		}
749		if (eventCount != ec)
750		break;
#	Line 758 | Line 774 | public class ForkJoinPool extends Abstra
774		long nh = (((long)ec) << EVENT_COUNT_SHIFT) | ((long)(w.poolIndex+1));
775		long h;
776		while ((runState < SHUTDOWN || !tryTerminate(false)) &&
777	<	(((int)((h = eventWaiters) & WAITER_ID_MASK)) == 0 ||
777	>	(((int)(h = eventWaiters) & WAITER_ID_MASK) == 0 ||
778		(int)(h >>> EVENT_COUNT_SHIFT) == ec) &&
779		eventCount == ec) {
780		if (UNSAFE.compareAndSwapLong(this, eventWaitersOffset,
#	Line 785 | Line 801 | public class ForkJoinPool extends Abstra
801		if (tryAccumulateStealCount(w)) { // transfer while idle
802		boolean untimed = (w.nextWaiter != 0L ||
803		(workerCounts & RUNNING_COUNT_MASK) <= 1);
804	<	long startTime = untimed? 0 : System.nanoTime();
804	>	long startTime = untimed ? 0 : System.nanoTime();
805		Thread.interrupted();         // clear/ignore interrupt
806	<	if (eventCount != ec || w.runState != 0 ||
807	<	runState >= TERMINATING)  // recheck after clear
792	<	break;
806	>	if (w.isTerminating() || eventCount != ec)
807	>	break;                    // recheck after clear
808		if (untimed)
809		LockSupport.park(w);
810		else {
811		LockSupport.parkNanos(w, SHRINK_RATE_NANOS);
812	<	if (eventCount != ec || w.runState != 0 ||
798	<	runState >= TERMINATING)
812	>	if (eventCount != ec || w.isTerminating())
813		break;
814		if (System.nanoTime() - startTime >= SHRINK_RATE_NANOS)
815		tryShutdownUnusedWorker(ec);
#	Line 807 | Line 821 | public class ForkJoinPool extends Abstra
821		// Maintaining parallelism
822
823		/**
824	<	* Pushes worker onto the spare stack
824	>	* Pushes worker onto the spare stack.
825		*/
826		final void pushSpare(ForkJoinWorkerThread w) {
827		int ns = (++w.spareCount << SPARE_COUNT_SHIFT) | (w.poolIndex + 1);
#	Line 827 | Line 841 | public class ForkJoinPool extends Abstra
841		if ((sw = spareWaiters) != 0 &&
842		(id = (sw & SPARE_ID_MASK) - 1) >= 0 &&
843		id < n && (w = ws[id]) != null &&
844	<	(workerCounts & RUNNING_COUNT_MASK) < parallelism &&
844	>	(runState >= TERMINATING ||
845	>	(workerCounts & RUNNING_COUNT_MASK) < parallelism) &&
846		spareWaiters == sw &&
847		UNSAFE.compareAndSwapInt(this, spareWaitersOffset,
848		sw, w.nextSpare)) {
#	Line 863 | Line 878 | public class ForkJoinPool extends Abstra
878		UNSAFE.compareAndSwapInt(this, workerCountsOffset, wc,
879		wc + (ONE_RUNNING|ONE_TOTAL))) {
880		ForkJoinWorkerThread w = null;
881	+	Throwable fail = null;
882		try {
883		w = factory.newThread(this);
884	<	} finally { // adjust on null or exceptional factory return
885	<	if (w == null) {
870	<	decrementWorkerCounts(ONE_RUNNING, ONE_TOTAL);
871	<	tryTerminate(false); // handle failure during shutdown
872	<	}
884	>	} catch (Throwable ex) {
885	>	fail = ex;
886		}
887	<	if (w == null)
887	>	if (w == null) { // null or exceptional factory return
888	>	decrementWorkerCounts(ONE_RUNNING, ONE_TOTAL);
889	>	tryTerminate(false); // handle failure during shutdown
890	>	// If originating from an external caller,
891	>	// propagate exception, else ignore
892	>	if (fail != null && runState < TERMINATING &&
893	>	!(Thread.currentThread() instanceof
894	>	ForkJoinWorkerThread))
895	>	UNSAFE.throwException(fail);
896		break;
897	+	}
898		w.start(recordWorker(w), ueh);
899	<	if ((workerCounts >>> TOTAL_COUNT_SHIFT) >= pc) {
878	<	int c; // advance event count
879	<	UNSAFE.compareAndSwapInt(this, eventCountOffset,
880	<	c = eventCount, c+1);
899	>	if ((workerCounts >>> TOTAL_COUNT_SHIFT) >= pc)
900		break; // add at most one unless total below target
882	–	}
901		}
902		}
903		if (eventWaiters != 0L)
#	Line 915 | Line 933 | public class ForkJoinPool extends Abstra
933		}
934		else if ((h = eventWaiters) != 0L) {
935		long nh;
936	<	int id = ((int)(h & WAITER_ID_MASK)) - 1;
936	>	int id = (((int)h) & WAITER_ID_MASK) - 1;
937		if (id >= 0 && id < n && (w = ws[id]) != null &&
938		(nh = w.nextWaiter) != 0L && // keep at least one worker
939		UNSAFE.compareAndSwapLong(this, eventWaitersOffset, h, nh))
#	Line 961 | Line 979 | public class ForkJoinPool extends Abstra
979		boolean active = w.active;
980		boolean inactivate = false;
981		int pc = parallelism;
982	<	int rs;
983	<	while (w.runState == 0 && (rs = runState) < TERMINATING) {
982	>	while (w.runState == 0) {
983	>	int rs = runState;
984	>	if (rs >= TERMINATING) {           // propagate shutdown
985	>	w.shutdown();
986	>	break;
987	>	}
988		if ((inactivate || (active && (rs & ACTIVE_COUNT_MASK) >= pc)) &&
989	<	UNSAFE.compareAndSwapInt(this, runStateOffset, rs, rs - 1))
989	>	UNSAFE.compareAndSwapInt(this, runStateOffset, rs, --rs)) {
990		inactivate = active = w.active = false;
991	<	int wc = workerCounts;
991	>	if (rs == SHUTDOWN) {          // all inactive and shut down
992	>	tryTerminate(false);
993	>	continue;
994	>	}
995	>	}
996	>	int wc = workerCounts;             // try to suspend as spare
997		if ((wc & RUNNING_COUNT_MASK) > pc) {
998		if (!(inactivate |= active) && // must inactivate to suspend
999	<	workerCounts == wc &&      // try to suspend as spare
999	>	workerCounts == wc &&
1000		UNSAFE.compareAndSwapInt(this, workerCountsOffset,
1001		wc, wc - ONE_RUNNING))
1002		w.suspendAsSpare();
1003		}
1004		else if ((wc >>> TOTAL_COUNT_SHIFT) < pc)
1005		helpMaintainParallelism();     // not enough workers
1006	<	else if (!ran) {
1006	>	else if (ran)
1007	>	break;
1008	>	else {
1009		long h = eventWaiters;
1010		int ec = eventCount;
1011		if (h != 0L && (int)(h >>> EVENT_COUNT_SHIFT) != ec)
#	Line 988 | Line 1017 | public class ForkJoinPool extends Abstra
1017		else if (!(inactivate |= active))
1018		eventSync(w, wec);         // must inactivate before sync
1019		}
991	–	else
992	–	break;
1020		}
1021		}
1022
#	Line 999 | Line 1026 | public class ForkJoinPool extends Abstra
1026		*
1027		* @param joinMe the task to join
1028		* @param worker the current worker thread
1029	+	* @param timed true if wait should time out
1030	+	* @param nanos timeout value if timed
1031		*/
1032	<	final void awaitJoin(ForkJoinTask<?> joinMe, ForkJoinWorkerThread worker) {
1032	>	final void awaitJoin(ForkJoinTask<?> joinMe, ForkJoinWorkerThread worker,
1033	>	boolean timed, long nanos) {
1034	>	long startTime = timed? System.nanoTime() : 0L;
1035		int retries = 2 + (parallelism >> 2); // #helpJoins before blocking
1036	+	boolean running = true;               // false when count decremented
1037		while (joinMe.status >= 0) {
1038	<	int wc;
1039	<	worker.helpJoinTask(joinMe);
1038	>	if (runState >= TERMINATING) {
1039	>	joinMe.cancelIgnoringExceptions();
1040	>	break;
1041	>	}
1042	>	running = worker.helpJoinTask(joinMe, running);
1043		if (joinMe.status < 0)
1044		break;
1045	<	else if (retries > 0)
1045	>	if (retries > 0) {
1046		--retries;
1047	<	else if (((wc = workerCounts) & RUNNING_COUNT_MASK) != 0 &&
1048	<	UNSAFE.compareAndSwapInt(this, workerCountsOffset,
1049	<	wc, wc - ONE_RUNNING)) {
1050	<	int stat, c; long h;
1051	<	while ((stat = joinMe.status) >= 0 &&
1052	<	(h = eventWaiters) != 0L && // help release others
1053	<	(int)(h >>> EVENT_COUNT_SHIFT) != eventCount)
1047	>	continue;
1048	>	}
1049	>	int wc = workerCounts;
1050	>	if ((wc & RUNNING_COUNT_MASK) != 0) {
1051	>	if (running) {
1052	>	if (!UNSAFE.compareAndSwapInt(this, workerCountsOffset,
1053	>	wc, wc - ONE_RUNNING))
1054	>	continue;
1055	>	running = false;
1056	>	}
1057	>	long h = eventWaiters;
1058	>	if (h != 0L && (int)(h >>> EVENT_COUNT_SHIFT) != eventCount)
1059		releaseEventWaiters();
1060	<	if (stat >= 0 &&
1061	<	((workerCounts & RUNNING_COUNT_MASK) == 0 ||
1062	<	(stat =
1063	<	joinMe.internalAwaitDone(JOIN_TIMEOUT_MILLIS)) >= 0))
1064	<	helpMaintainParallelism(); // timeout or no running workers
1065	<	do {} while (!UNSAFE.compareAndSwapInt
1066	<	(this, workerCountsOffset,
1067	<	c = workerCounts, c + ONE_RUNNING));
1068	<	if (stat < 0)
1069	<	break;   // else restart
1060	>	if ((workerCounts & RUNNING_COUNT_MASK) != 0) {
1061	>	long ms; int ns;
1062	>	if (!timed) {
1063	>	ms = JOIN_TIMEOUT_MILLIS;
1064	>	ns = 0;
1065	>	}
1066	>	else { // at most JOIN_TIMEOUT_MILLIS per wait
1067	>	long nt = nanos - (System.nanoTime() - startTime);
1068	>	if (nt <= 0L)
1069	>	break;
1070	>	ms = nt / 1000000;
1071	>	if (ms > JOIN_TIMEOUT_MILLIS) {
1072	>	ms = JOIN_TIMEOUT_MILLIS;
1073	>	ns = 0;
1074	>	}
1075	>	else
1076	>	ns = (int) (nt % 1000000);
1077	>	}
1078	>	joinMe.internalAwaitDone(ms, ns);
1079	>	}
1080	>	if (joinMe.status < 0)
1081	>	break;
1082		}
1083	+	helpMaintainParallelism();
1084	+	}
1085	+	if (!running) {
1086	+	int c;
1087	+	do {} while (!UNSAFE.compareAndSwapInt
1088	+	(this, workerCountsOffset,
1089	+	c = workerCounts, c + ONE_RUNNING));
1090		}
1091		}
1092
#	Line 1038 | Line 1097 | public class ForkJoinPool extends Abstra
1097		throws InterruptedException {
1098		while (!blocker.isReleasable()) {
1099		int wc = workerCounts;
1100	<	if ((wc & RUNNING_COUNT_MASK) != 0 &&
1101	<	UNSAFE.compareAndSwapInt(this, workerCountsOffset,
1102	<	wc, wc - ONE_RUNNING)) {
1100	>	if ((wc & RUNNING_COUNT_MASK) == 0)
1101	>	helpMaintainParallelism();
1102	>	else if (UNSAFE.compareAndSwapInt(this, workerCountsOffset,
1103	>	wc, wc - ONE_RUNNING)) {
1104		try {
1105		while (!blocker.isReleasable()) {
1106		long h = eventWaiters;
#	Line 1085 | Line 1145 | public class ForkJoinPool extends Abstra
1145		// Finish now if all threads terminated; else in some subsequent call
1146		if ((workerCounts >>> TOTAL_COUNT_SHIFT) == 0) {
1147		advanceRunLevel(TERMINATED);
1148	<	termination.arrive();
1148	>	termination.forceTermination();
1149		}
1150		return true;
1151		}
#	Line 1107 | Line 1167 | public class ForkJoinPool extends Abstra
1167		c = eventCount, c+1);
1168		eventWaiters = 0L; // clobber lists
1169		spareWaiters = 0;
1170	<	ForkJoinWorkerThread[] ws = workers;
1111	<	int n = ws.length;
1112	<	for (int i = 0; i < n; ++i) {
1113	<	ForkJoinWorkerThread w = ws[i];
1170	>	for (ForkJoinWorkerThread w : workers) {
1171		if (w != null) {
1172		w.shutdown();
1173		if (passes > 0 && !w.isTerminated()) {
1174		w.cancelTasks();
1175		LockSupport.unpark(w);
1176	<	if (passes > 1) {
1176	>	if (passes > 1 && !w.isInterrupted()) {
1177		try {
1178		w.interrupt();
1179		} catch (SecurityException ignore) {
#	Line 1129 | Line 1186 | public class ForkJoinPool extends Abstra
1186		}
1187
1188		/**
1189	<	* Clear out and cancel submissions, ignoring exceptions
1189	>	* Clears out and cancels submissions, ignoring exceptions.
1190		*/
1191		private void cancelSubmissions() {
1192		ForkJoinTask<?> task;
#	Line 1144 | Line 1201 | public class ForkJoinPool extends Abstra
1201		// misc support for ForkJoinWorkerThread
1202
1203		/**
1204	<	* Returns pool number
1204	>	* Returns pool number.
1205		*/
1206		final int getPoolNumber() {
1207		return poolNumber;
1208		}
1209
1210		/**
1211	<	* Tries to accumulates steal count from a worker, clearing
1212	<	* the worker's value.
1211	>	* Tries to accumulate steal count from a worker, clearing
1212	>	* the worker's value if successful.
1213		*
1214		* @return true if worker steal count now zero
1215		*/
#	Line 1176 | Line 1233 | public class ForkJoinPool extends Abstra
1233		int pc = parallelism; // use parallelism, not rc
1234		int ac = runState;    // no mask -- artificially boosts during shutdown
1235		// Use exact results for small values, saturate past 4
1236	<	return pc <= ac? 0 : pc >>> 1 <= ac? 1 : pc >>> 2 <= ac? 3 : pc >>> 3;
1236	>	return ((pc <= ac) ? 0 :
1237	>	(pc >>> 1 <= ac) ? 1 :
1238	>	(pc >>> 2 <= ac) ? 3 :
1239	>	pc >>> 3);
1240		}
1241
1242		// Public and protected methods
#	Line 1226 | Line 1286 | public class ForkJoinPool extends Abstra
1286		* use {@link #defaultForkJoinWorkerThreadFactory}.
1287		* @param handler the handler for internal worker threads that
1288		* terminate due to unrecoverable errors encountered while executing
1289	<	* tasks. For default value, use <code>null</code>.
1289	>	* tasks. For default value, use {@code null}.
1290		* @param asyncMode if true,
1291		* establishes local first-in-first-out scheduling mode for forked
1292		* tasks that are never joined. This mode may be more appropriate
1293		* than default locally stack-based mode in applications in which
1294		* worker threads only process event-style asynchronous tasks.
1295	<	* For default value, use <code>false</code>.
1295	>	* For default value, use {@code false}.
1296		* @throws IllegalArgumentException if parallelism less than or
1297		*         equal to zero, or greater than implementation limit
1298		* @throws NullPointerException if the factory is null
#	Line 1280 | Line 1340 | public class ForkJoinPool extends Abstra
1340		// Execution methods
1341
1342		/**
1343	<	* Common code for execute, invoke and submit
1343	>	* Submits task and creates, starts, or resumes some workers if necessary
1344		*/
1345		private <T> void doSubmit(ForkJoinTask<T> task) {
1286	–	if (task == null)
1287	–	throw new NullPointerException();
1288	–	if (runState >= SHUTDOWN)
1289	–	throw new RejectedExecutionException();
1346		submissionQueue.offer(task);
1347		int c; // try to increment event count -- CAS failure OK
1348		UNSAFE.compareAndSwapInt(this, eventCountOffset, c = eventCount, c+1);
1349	<	helpMaintainParallelism(); // create, start, or resume some workers
1349	>	helpMaintainParallelism();
1350		}
1351
1352		/**
#	Line 1303 | Line 1359 | public class ForkJoinPool extends Abstra
1359		*         scheduled for execution
1360		*/
1361		public <T> T invoke(ForkJoinTask<T> task) {
1362	<	doSubmit(task);
1363	<	return task.join();
1362	>	if (task == null)
1363	>	throw new NullPointerException();
1364	>	if (runState >= SHUTDOWN)
1365	>	throw new RejectedExecutionException();
1366	>	Thread t = Thread.currentThread();
1367	>	if ((t instanceof ForkJoinWorkerThread) &&
1368	>	((ForkJoinWorkerThread)t).pool == this)
1369	>	return task.invoke();  // bypass submit if in same pool
1370	>	else {
1371	>	doSubmit(task);
1372	>	return task.join();
1373	>	}
1374	>	}
1375	>
1376	>	/**
1377	>	* Unless terminating, forks task if within an ongoing FJ
1378	>	* computation in the current pool, else submits as external task.
1379	>	*/
1380	>	private <T> void forkOrSubmit(ForkJoinTask<T> task) {
1381	>	if (runState >= SHUTDOWN)
1382	>	throw new RejectedExecutionException();
1383	>	Thread t = Thread.currentThread();
1384	>	if ((t instanceof ForkJoinWorkerThread) &&
1385	>	((ForkJoinWorkerThread)t).pool == this)
1386	>	task.fork();
1387	>	else
1388	>	doSubmit(task);
1389		}
1390
1391		/**
#	Line 1316 | Line 1397 | public class ForkJoinPool extends Abstra
1397		*         scheduled for execution
1398		*/
1399		public void execute(ForkJoinTask<?> task) {
1400	<	doSubmit(task);
1400	>	if (task == null)
1401	>	throw new NullPointerException();
1402	>	forkOrSubmit(task);
1403		}
1404
1405		// AbstractExecutorService methods
#	Line 1327 | Line 1410 | public class ForkJoinPool extends Abstra
1410		*         scheduled for execution
1411		*/
1412		public void execute(Runnable task) {
1413	+	if (task == null)
1414	+	throw new NullPointerException();
1415		ForkJoinTask<?> job;
1416		if (task instanceof ForkJoinTask<?>) // avoid re-wrap
1417		job = (ForkJoinTask<?>) task;
1418		else
1419		job = ForkJoinTask.adapt(task, null);
1420	<	doSubmit(job);
1420	>	forkOrSubmit(job);
1421		}
1422
1423		/**
#	Line 1345 | Line 1430 | public class ForkJoinPool extends Abstra
1430		*         scheduled for execution
1431		*/
1432		public <T> ForkJoinTask<T> submit(ForkJoinTask<T> task) {
1433	<	doSubmit(task);
1433	>	if (task == null)
1434	>	throw new NullPointerException();
1435	>	forkOrSubmit(task);
1436		return task;
1437		}
1438
#	Line 1355 | Line 1442 | public class ForkJoinPool extends Abstra
1442		*         scheduled for execution
1443		*/
1444		public <T> ForkJoinTask<T> submit(Callable<T> task) {
1445	+	if (task == null)
1446	+	throw new NullPointerException();
1447		ForkJoinTask<T> job = ForkJoinTask.adapt(task);
1448	<	doSubmit(job);
1448	>	forkOrSubmit(job);
1449		return job;
1450		}
1451
#	Line 1366 | Line 1455 | public class ForkJoinPool extends Abstra
1455		*         scheduled for execution
1456		*/
1457		public <T> ForkJoinTask<T> submit(Runnable task, T result) {
1458	+	if (task == null)
1459	+	throw new NullPointerException();
1460		ForkJoinTask<T> job = ForkJoinTask.adapt(task, result);
1461	<	doSubmit(job);
1461	>	forkOrSubmit(job);
1462		return job;
1463		}
1464
#	Line 1377 | Line 1468 | public class ForkJoinPool extends Abstra
1468		*         scheduled for execution
1469		*/
1470		public ForkJoinTask<?> submit(Runnable task) {
1471	+	if (task == null)
1472	+	throw new NullPointerException();
1473		ForkJoinTask<?> job;
1474		if (task instanceof ForkJoinTask<?>) // avoid re-wrap
1475		job = (ForkJoinTask<?>) task;
1476		else
1477		job = ForkJoinTask.adapt(task, null);
1478	<	doSubmit(job);
1478	>	forkOrSubmit(job);
1479		return job;
1480		}
1481
#	Line 1442 | Line 1535 | public class ForkJoinPool extends Abstra
1535
1536		/**
1537		* Returns the number of worker threads that have started but not
1538	<	* yet terminated.  This result returned by this method may differ
1538	>	* yet terminated.  The result returned by this method may differ
1539		* from {@link #getParallelism} when threads are created to
1540		* maintain parallelism when others are cooperatively blocked.
1541		*
#	Line 1527 | Line 1620 | public class ForkJoinPool extends Abstra
1620		*/
1621		public long getQueuedTaskCount() {
1622		long count = 0;
1623	<	ForkJoinWorkerThread[] ws = workers;
1531	<	int n = ws.length;
1532	<	for (int i = 0; i < n; ++i) {
1533	<	ForkJoinWorkerThread w = ws[i];
1623	>	for (ForkJoinWorkerThread w : workers)
1624		if (w != null)
1625		count += w.getQueueSize();
1536	–	}
1626		return count;
1627		}
1628
#	Line 1588 | Line 1677 | public class ForkJoinPool extends Abstra
1677		*/
1678		protected int drainTasksTo(Collection<? super ForkJoinTask<?>> c) {
1679		int count = submissionQueue.drainTo(c);
1680	<	ForkJoinWorkerThread[] ws = workers;
1592	<	int n = ws.length;
1593	<	for (int i = 0; i < n; ++i) {
1594	<	ForkJoinWorkerThread w = ws[i];
1680	>	for (ForkJoinWorkerThread w : workers)
1681		if (w != null)
1682		count += w.drainTasksTo(c);
1597	–	}
1683		return count;
1684		}
1685
#	Line 1688 | Line 1773 | public class ForkJoinPool extends Abstra
1773		* commenced but not yet completed.  This method may be useful for
1774		* debugging. A return of {@code true} reported a sufficient
1775		* period after shutdown may indicate that submitted tasks have
1776	<	* ignored or suppressed interruption, causing this executor not
1777	<	* to properly terminate.
1776	>	* ignored or suppressed interruption, or are waiting for IO,
1777	>	* causing this executor not to properly terminate. (See the
1778	>	* advisory notes for class {@link ForkJoinTask} stating that
1779	>	* tasks should not normally entail blocking operations.  But if
1780	>	* they do, they must abort them on interrupt.)
1781		*
1782		* @return {@code true} if terminating but not yet terminated
1783		*/
#	Line 1698 | Line 1786 | public class ForkJoinPool extends Abstra
1786		}
1787
1788		/**
1789	+	* Returns true if terminating or terminated. Used by ForkJoinWorkerThread.
1790	+	*/
1791	+	final boolean isAtLeastTerminating() {
1792	+	return runState >= TERMINATING;
1793	+	}
1794	+
1795	+	/**
1796		* Returns {@code true} if this pool has been shut down.
1797		*
1798		* @return {@code true} if this pool has been shut down
#	Line 1720 | Line 1815 | public class ForkJoinPool extends Abstra
1815		public boolean awaitTermination(long timeout, TimeUnit unit)
1816		throws InterruptedException {
1817		try {
1818	<	return termination.awaitAdvanceInterruptibly(0, timeout, unit) > 0;
1818	>	termination.awaitAdvanceInterruptibly(0, timeout, unit);
1819		} catch (TimeoutException ex) {
1820		return false;
1821		}
1822	+	return true;
1823		}
1824
1825		/**
#	Line 1851 | Line 1947 | public class ForkJoinPool extends Abstra
1947		private static final long eventCountOffset =
1948		objectFieldOffset("eventCount", ForkJoinPool.class);
1949		private static final long eventWaitersOffset =
1950	<	objectFieldOffset("eventWaiters",ForkJoinPool.class);
1950	>	objectFieldOffset("eventWaiters", ForkJoinPool.class);
1951		private static final long stealCountOffset =
1952	<	objectFieldOffset("stealCount",ForkJoinPool.class);
1952	>	objectFieldOffset("stealCount", ForkJoinPool.class);
1953		private static final long spareWaitersOffset =
1954	<	objectFieldOffset("spareWaiters",ForkJoinPool.class);
1954	>	objectFieldOffset("spareWaiters", ForkJoinPool.class);
1955
1956		private static long objectFieldOffset(String field, Class<?> klazz) {
1957		try {

Diff Legend

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