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.146 by dl, Mon Nov 19 18:12:42 2012 UTC vs.
Revision 1.157 by jsr166, Fri Dec 14 16:33:42 2012 UTC

#	Line 51 | Line 51 | import java.util.concurrent.TimeUnit;
51		* dynamically adding, suspending, or resuming internal worker
52		* threads, even if some tasks are stalled waiting to join
53		* others. However, no such adjustments are guaranteed in the face of
54	<	* blocked IO or other unmanaged synchronization. The nested {@link
54	>	* blocked I/O or other unmanaged synchronization. The nested {@link
55		* ManagedBlocker} interface enables extension of the kinds of
56		* synchronization accommodated.
57		*
#	Line 99 | Line 99 | import java.util.concurrent.TimeUnit;
99		*
100		* <p>The common pool is by default constructed with default
101		* parameters, but these may be controlled by setting three {@link
102	<	* System#getProperty properties} with prefix {@code
102	>	* System#getProperty system properties} with prefix {@code
103		* java.util.concurrent.ForkJoinPool.common}: {@code parallelism} --
104		* an integer greater than zero, {@code threadFactory} -- the class
105		* name of a {@link ForkJoinWorkerThreadFactory}, and {@code
106	–	<<<<<<< ForkJoinPool.java
107	–	* exceptionHandler} -- the class name of a {@code
108	–	=======
106		* exceptionHandler} -- the class name of a {@link
107		* java.lang.Thread.UncaughtExceptionHandler
111	–	>>>>>>> 1.111
108		* Thread.UncaughtExceptionHandler}. Upon any error in establishing
109		* these settings, default parameters are used.
110		*
#	Line 320 | Line 316 | public class ForkJoinPool extends Abstra
316		* execute. However, many other threads may notice the same task
317		* and each signal to wake up a thread that might take it. So in
318		* general, pools will be over-signalled.  When a submission is
319	<	* added or another worker adds a task to a queue that is
320	<	* apparently empty, they signal waiting workers (or trigger
321	<	* creation of new ones if fewer than the given parallelism
322	<	* level).  These primary signals are buttressed by signals
323	<	* whenever other threads scan for work or do not have a task to
324	<	* process (including the case of leaving a hint to unparked
325	<	* threads to help signal others upon wakeup).  On most platforms,
326	<	* signalling (unpark) overhead time is noticeably long, and the
327	<	* time between signalling a thread and it actually making
328	<	* progress can be very noticeably long, so it is worth offloading
329	<	* these delays from critical paths as much as possible.
319	>	* added or another worker adds a task to a queue that has fewer
320	>	* than two tasks, they signal waiting workers (or trigger
321	>	* creation of new ones if fewer than the given parallelism level
322	>	* -- signalWork), and may leave a hint to the unparked worker to
323	>	* help signal others upon wakeup).  These primary signals are
324	>	* buttressed by others (see method helpSignal) whenever other
325	>	* threads scan for work or do not have a task to process.  On
326	>	* most platforms, signalling (unpark) overhead time is noticeably
327	>	* long, and the time between signalling a thread and it actually
328	>	* making progress can be very noticeably long, so it is worth
329	>	* offloading these delays from critical paths as much as
330	>	* possible.
331		*
332		* Trimming workers. To release resources after periods of lack of
333		* use, a worker starting to wait when the pool is quiescent will
#	Line 538 | Line 535 | public class ForkJoinPool extends Abstra
535		}
536
537		/**
538	+	* Per-thread records for threads that submit to pools. Currently
539	+	* holds only pseudo-random seed / index that is used to choose
540	+	* submission queues in method externalPush. In the future, this may
541	+	* also incorporate a means to implement different task rejection
542	+	* and resubmission policies.
543	+	*
544	+	* Seeds for submitters and workers/workQueues work in basically
545	+	* the same way but are initialized and updated using slightly
546	+	* different mechanics. Both are initialized using the same
547	+	* approach as in class ThreadLocal, where successive values are
548	+	* unlikely to collide with previous values. Seeds are then
549	+	* randomly modified upon collisions using xorshifts, which
550	+	* requires a non-zero seed.
551	+	*/
552	+	static final class Submitter {
553	+	int seed;
554	+	Submitter(int s) { seed = s; }
555	+	}
556	+
557	+	/**
558		* Class for artificial tasks that are used to replace the target
559		* of local joins if they are removed from an interior queue slot
560		* in WorkQueue.tryRemoveAndExec. We don't need the proxy to
#	Line 603 | Line 620 | public class ForkJoinPool extends Abstra
620		* trades off slightly slower average field access for the sake of
621		* avoiding really bad worst-case access. (Until better JVM
622		* support is in place, this padding is dependent on transient
623	<	* properties of JVM field layout rules.)
623	>	* properties of JVM field layout rules.) We also take care in
624	>	* allocating, sizing and resizing the array. Non-shared queue
625	>	* arrays are initialized by workers before use. Others are
626	>	* allocated on first use.
627		*/
628		static final class WorkQueue {
629		/**
#	Line 626 | Line 646 | public class ForkJoinPool extends Abstra
646		*/
647		static final int MAXIMUM_QUEUE_CAPACITY = 1 << 26; // 64M
648
649	+	// Heuristic padding to ameliorate unfortunate memory placements
650	+	volatile long pad00, pad01, pad02, pad03, pad04, pad05, pad06;
651	+
652		int seed;                  // for random scanning; initialize nonzero
653		volatile int eventCount;   // encoded inactivation count; < 0 if inactive
654		int nextWait;              // encoded record of next event waiter
#	Line 643 | Line 666 | public class ForkJoinPool extends Abstra
666		volatile ForkJoinTask<?> currentJoin;  // task being joined in awaitJoin
667		ForkJoinTask<?> currentSteal; // current non-local task being executed
668
669	<	// Heuristic padding to ameliorate unfortunate memory placements
670	<	Object p00, p01, p02, p03, p04, p05, p06, p07;
648	<	Object p08, p09, p0a, p0b, p0c;
669	>	volatile Object pad10, pad11, pad12, pad13, pad14, pad15, pad16, pad17;
670	>	volatile Object pad18, pad19, pad1a, pad1b, pad1c, pad1d;
671
672		WorkQueue(ForkJoinPool pool, ForkJoinWorkerThread owner, int mode,
673		int seed) {
652	–	this.array = new ForkJoinTask<?>[WorkQueue.INITIAL_QUEUE_CAPACITY];
674		this.pool = pool;
675		this.owner = owner;
676		this.mode = mode;
677		this.seed = seed;
678	<	// Place indices in the center of array
678	>	// Place indices in the center of array (that is not yet allocated)
679		base = top = INITIAL_QUEUE_CAPACITY >>> 1;
680		}
681
682		/**
683	<	* Pushes a task. Call only by owner in unshared queues.
684	<	* Cases needing resizing or rejection are relayed to fullPush
685	<	* (that also handles shared queues).
683	>	* Returns the approximate number of tasks in the queue.
684	>	*/
685	>	final int queueSize() {
686	>	int n = base - top;       // non-owner callers must read base first
687	>	return (n >= 0) ? 0 : -n; // ignore transient negative
688	>	}
689	>
690	>	/**
691	>	* Provides a more accurate estimate of whether this queue has
692	>	* any tasks than does queueSize, by checking whether a
693	>	* near-empty queue has at least one unclaimed task.
694	>	*/
695	>	final boolean isEmpty() {
696	>	ForkJoinTask<?>[] a; int m, s;
697	>	int n = base - (s = top);
698	>	return (n >= 0 ||
699	>	(n == -1 &&
700	>	((a = array) == null ||
701	>	(m = a.length - 1) < 0 ||
702	>	U.getObject
703	>	(a, (long)((m & (s - 1)) << ASHIFT) + ABASE) == null)));
704	>	}
705	>
706	>	/**
707	>	* Pushes a task. Call only by owner in unshared queues.  (The
708	>	* shared-queue version is embedded in method externalPush.)
709		*
710		* @param task the task. Caller must ensure non-null.
711		* @throw RejectedExecutionException if array cannot be resized
#	Line 670 | Line 714 | public class ForkJoinPool extends Abstra
714		ForkJoinTask<?>[] a; ForkJoinPool p;
715		int s = top, m, n;
716		if ((a = array) != null) {    // ignore if queue removed
717	<	U.putOrderedObject
718	<	(a, (((m = a.length - 1) & s) << ASHIFT) + ABASE, task);
719	<	if ((n = (top = s + 1) - base) <= 1) {
717	>	int j = (((m = a.length - 1) & s) << ASHIFT) + ABASE;
718	>	U.putOrderedObject(a, j, task);
719	>	if ((n = (top = s + 1) - base) <= 2) {
720		if ((p = pool) != null)
721	<	p.signalWork(this, 0);
721	>	p.signalWork(this);
722		}
723		else if (n >= m)
724		growArray();
725		}
726		}
727
684	–	/**
685	–	* Pushes a task if lock is free and array is either big
686	–	* enough or can be resized to be big enough.
687	–	*
688	–	* @param task the task. Caller must ensure non-null.
689	–	* @return true if submitted
690	–	*/
691	–	final boolean trySharedPush(ForkJoinTask<?> task) {
692	–	boolean submitted = false;
693	–	if (qlock == 0 && U.compareAndSwapInt(this, QLOCK, 0, 1)) {
694	–	ForkJoinTask<?>[] a = array;  ForkJoinPool p;
695	–	int s = top;
696	–	try {
697	–	if ((a != null && a.length > s + 1 - base) ||
698	–	(a = growArray()) != null) {   // must presize
699	–	int j = (((a.length - 1) & s) << ASHIFT) + ABASE;
700	–	U.putOrderedObject(a, j, task);
701	–	top = s + 1;
702	–	submitted = true;
703	–	}
704	–	} finally {
705	–	qlock = 0;                         // unlock
706	–	}
707	–	if (submitted && (p = pool) != null)
708	–	p.signalWork(this, 0);
709	–	}
710	–	return submitted;
711	–	}
712	–
728		/**
729		* Initializes or doubles the capacity of array. Call either
730		* by owner or with lock held -- it is OK for base, but not
#	Line 859 | Line 874 | public class ForkJoinPool extends Abstra
874		return seed = r ^= r << 5;
875		}
876
862	–	/**
863	–	* Provides a more accurate estimate of size than (top - base)
864	–	* by ordering reads and checking whether a near-empty queue
865	–	* has at least one unclaimed task.
866	–	*/
867	–	final int queueSize() {
868	–	ForkJoinTask<?>[] a; int k, s, n;
869	–	return ((n = base - (s = top)) < 0 &&
870	–	(n != -1 ||
871	–	((a = array) != null && (k = a.length) > 0 &&
872	–	U.getObject
873	–	(a, (long)((((k - 1) & (s - 1)) << ASHIFT) + ABASE)) != null))) ?
874	–	-n : 0;
875	–	}
876	–
877		// Specialized execution methods
878
879		/**
#	Line 987 | Line 987 | public class ForkJoinPool extends Abstra
987		if (t != null) {
988		(currentSteal = t).doExec();
989		currentSteal = null;
990	<	++nsteals;
991	<	if (top != base) {       // process remaining local tasks
990	>	if (base - top < 0) {       // process remaining local tasks
991		if (mode == 0)
992		popAndExecAll();
993		else
994		pollAndExecAll();
995		}
996	+	++nsteals;
997	+	hint = -1;
998		}
999		}
1000
#	Line 1070 | Line 1071 | public class ForkJoinPool extends Abstra
1071		defaultForkJoinWorkerThreadFactory;
1072
1073		/**
1073	–	* Per-thread records for threads that submit to pools. Currently
1074	–	* holds only pseudo-random seed / index that is used to choose
1075	–	* submission queues in method externalPush. In the future, this may
1076	–	* also incorporate a means to implement different task rejection
1077	–	* and resubmission policies.
1078	–	*
1079	–	* Seeds for submitters and workers/workQueues work in basically
1080	–	* the same way but are initialized and updated using slightly
1081	–	* different mechanics. Both are initialized using the same
1082	–	* approach as in class ThreadLocal, where successive values are
1083	–	* unlikely to collide with previous values. Seeds are then
1084	–	* randomly modified upon collisions using xorshifts, which
1085	–	* requires a non-zero seed.
1086	–	*/
1087	–	static final class Submitter {
1088	–	int seed;
1089	–	Submitter(int s) { seed = s; }
1090	–	}
1091	–
1092	–	/**
1074		* Per-thread submission bookkeeping. Shared across all pools
1075		* to reduce ThreadLocal pollution and because random motion
1076		* to avoid contention in one pool is likely to hold for others.
#	Line 1099 | Line 1080 | public class ForkJoinPool extends Abstra
1080		static final ThreadLocal<Submitter> submitters;
1081
1082		/**
1083	+	* Permission required for callers of methods that may start or
1084	+	* kill threads.
1085	+	*/
1086	+	private static final RuntimePermission modifyThreadPermission;
1087	+
1088	+	/**
1089		* Common (static) pool. Non-null for public use unless a static
1090		* construction exception, but internal usages null-check on use
1091		* to paranoically avoid potential initialization circularities
#	Line 1107 | Line 1094 | public class ForkJoinPool extends Abstra
1094		static final ForkJoinPool commonPool;
1095
1096		/**
1110	–	* Permission required for callers of methods that may start or
1111	–	* kill threads.
1112	–	*/
1113	–	private static final RuntimePermission modifyThreadPermission;
1114	–
1115	–	/**
1097		* Common pool parallelism. Must equal commonPool.parallelism.
1098		*/
1099		static final int commonPoolParallelism;
#	Line 1148 | Line 1129 | public class ForkJoinPool extends Abstra
1129		private static final long FAST_IDLE_TIMEOUT =  200L * 1000L * 1000L;
1130
1131		/**
1132	+	* Tolerance for idle timeouts, to cope with timer undershoots
1133	+	*/
1134	+	private static final long TIMEOUT_SLOP = 2000000L; // 20ms
1135	+
1136	+	/**
1137		* The maximum stolen->joining link depth allowed in method
1138		* tryHelpStealer.  Must be a power of two.  Depths for legitimate
1139		* chains are unbounded, but we use a fixed constant to avoid
#	Line 1259 | Line 1245 | public class ForkJoinPool extends Abstra
1245		* declaration order and may differ across JVMs, but the following
1246		* empirically works OK on current JVMs.
1247		*/
1248	+
1249	+	// Heuristic padding to ameliorate unfortunate memory placements
1250	+	volatile long pad00, pad01, pad02, pad03, pad04, pad05, pad06;
1251	+
1252		volatile long stealCount;                  // collects worker counts
1253		volatile long ctl;                         // main pool control
1254		volatile int plock;                        // shutdown status and seqLock
#	Line 1269 | Line 1259 | public class ForkJoinPool extends Abstra
1259		final Thread.UncaughtExceptionHandler ueh; // per-worker UEH
1260		final String workerNamePrefix;             // to create worker name string
1261
1262	+	volatile Object pad10, pad11, pad12, pad13, pad14, pad15, pad16, pad17;
1263	+	volatile Object pad18, pad19, pad1a, pad1b;
1264	+
1265		/*
1266		* Acquires the plock lock to protect worker array and related
1267		* updates. This method is called only if an initial CAS on plock
#	Line 1329 | Line 1322 | public class ForkJoinPool extends Abstra
1322		}
1323
1324		/**
1332	–	* Tries to create and start a worker; adjusts counts etc on failure
1333	–	*/
1334	–	private void addWorker() {
1335	–	ForkJoinWorkerThread wt = null;
1336	–	try {
1337	–	(wt = factory.newThread(this)).start();
1338	–	} catch (Throwable ex) {
1339	–	deregisterWorker(wt, ex); // adjust on failure
1340	–	}
1341	–	}
1342	–
1343	–	/**
1325		* Performs secondary initialization, called when plock is zero.
1326		* Creates workQueue array and sets plock to a valid value.  The
1327		* lock body must be exception-free (so no try/finally) so we
#	Line 1349 | Line 1330 | public class ForkJoinPool extends Abstra
1330		* fullExternalPush.)  Because the plock seq value can eventually
1331		* wrap around zero, this method harmlessly fails to reinitialize
1332		* if workQueues exists, while still advancing plock.
1333	+	*
1334	+	* Additionally tries to create the first worker.
1335		*/
1336	<	private void initWorkQueuesArray() {
1337	<	WorkQueue[] ws; int ps;
1336	>	private void initWorkers() {
1337	>	WorkQueue[] ws, nws; int ps;
1338		int p = config & SMASK;        // find power of two table size
1339		int n = (p > 1) ? p - 1 : 1;   // ensure at least 2 slots
1340		n |= n >>> 1; n |= n >>> 2; n |= n >>> 4; n |= n >>> 8; n |= n >>> 16;
1341	<	WorkQueue[] nws = new WorkQueue[(n + 1) << 1];
1341	>	n = (n + 1) << 1;
1342	>	if ((ws = workQueues) == null || ws.length == 0)
1343	>	nws = new WorkQueue[n];
1344	>	else
1345	>	nws = null;
1346		if (((ps = plock) & PL_LOCK) != 0 ||
1347		!U.compareAndSwapInt(this, PLOCK, ps, ps += PL_LOCK))
1348		ps = acquirePlock();
1349	<	if ((ws = workQueues) == null || ws.length == 0)
1349	>	if (((ws = workQueues) == null || ws.length == 0) && nws != null)
1350		workQueues = nws;
1351		int nps = (ps & SHUTDOWN) | ((ps + PL_LOCK) & ~SHUTDOWN);
1352		if (!U.compareAndSwapInt(this, PLOCK, ps, nps))
1353		releasePlock(nps);
1354	+	tryAddWorker();
1355	+	}
1356	+
1357	+	/**
1358	+	* Tries to create and start one worker if fewer than target
1359	+	* parallelism level exist. Adjusts counts etc on failure.
1360	+	*/
1361	+	private void tryAddWorker() {
1362		long c; int u;
1363	<	if ((u = (int)((c = ctl) >>> 32)) < 0 && (int)c == 0) {
1363	>	while ((u = (int)((c = ctl) >>> 32)) < 0 &&
1364	>	(u & SHORT_SIGN) != 0 && (int)c == 0) {
1365		long nc = (long)(((u + UTC_UNIT) & UTC_MASK) |
1366		((u + UAC_UNIT) & UAC_MASK)) << 32;
1367	<	if (U.compareAndSwapLong(this, CTL, c, nc))
1368	<	addWorker();
1367	>	if (U.compareAndSwapLong(this, CTL, c, nc)) {
1368	>	ForkJoinWorkerThreadFactory fac;
1369	>	Throwable ex = null;
1370	>	ForkJoinWorkerThread wt = null;
1371	>	try {
1372	>	if ((fac = factory) != null &&
1373	>	(wt = fac.newThread(this)) != null) {
1374	>	wt.start();
1375	>	break;
1376	>	}
1377	>	} catch (Throwable e) {
1378	>	ex = e;
1379	>	}
1380	>	deregisterWorker(wt, ex);
1381	>	break;
1382	>	}
1383		}
1374	–
1384		}
1385
1386		//  Registering and deregistering workers
#	Line 1384 | Line 1393 | public class ForkJoinPool extends Abstra
1393		* expanding as needed.
1394		*
1395		* @param wt the worker thread
1396	+	* @return the worker's queue
1397		*/
1398	<	final void registerWorker(ForkJoinWorkerThread wt) {
1399	<	if (wt != null && wt.workQueue == null) {
1400	<	int s, ps;    // generate a rarely colliding candidate index seed
1401	<	do {} while (!U.compareAndSwapInt(this, INDEXSEED, s = indexSeed,
1402	<	s += SEED_INCREMENT) ||
1403	<	s == 0); // skip 0
1404	<	WorkQueue w = new WorkQueue(this, wt, config >>> 16, s);
1405	<	if (((ps = plock) & PL_LOCK) != 0 ||
1406	<	!U.compareAndSwapInt(this, PLOCK, ps, ps += PL_LOCK))
1407	<	ps = acquirePlock();
1408	<	int nps = (ps & SHUTDOWN) | ((ps + PL_LOCK) & ~SHUTDOWN);
1409	<	try {
1410	<	WorkQueue[] ws;
1411	<	if ((ws = workQueues) != null && wt.workQueue == null) {
1412	<	int n = ws.length, m = n - 1;
1413	<	int r = (s << 1) | 1;           // use odd-numbered indices
1414	<	if (ws[r &= m] != null) {       // collision
1415	<	int probes = 0;             // step by approx half size
1416	<	int step = (n <= 4) ? 2 : ((n >>> 1) & EVENMASK) + 2;
1417	<	while (ws[r = (r + step) & m] != null) {
1418	<	if (++probes >= n) {
1419	<	workQueues = ws = Arrays.copyOf(ws, n <<= 1);
1420	<	m = n - 1;
1421	<	probes = 0;
1422	<	}
1398	>	final WorkQueue registerWorker(ForkJoinWorkerThread wt) {
1399	>	Thread.UncaughtExceptionHandler handler; WorkQueue[] ws; int s, ps;
1400	>	wt.setDaemon(true);
1401	>	if ((handler = ueh) != null)
1402	>	wt.setUncaughtExceptionHandler(handler);
1403	>	do {} while (!U.compareAndSwapInt(this, INDEXSEED, s = indexSeed,
1404	>	s += SEED_INCREMENT) ||
1405	>	s == 0); // skip 0
1406	>	WorkQueue w = new WorkQueue(this, wt, config >>> 16, s);
1407	>	if (((ps = plock) & PL_LOCK) != 0 ||
1408	>	!U.compareAndSwapInt(this, PLOCK, ps, ps += PL_LOCK))
1409	>	ps = acquirePlock();
1410	>	int nps = (ps & SHUTDOWN) | ((ps + PL_LOCK) & ~SHUTDOWN);
1411	>	try {
1412	>	if ((ws = workQueues) != null) {    // skip if shutting down
1413	>	int n = ws.length, m = n - 1;
1414	>	int r = (s << 1) | 1;           // use odd-numbered indices
1415	>	if (ws[r &= m] != null) {       // collision
1416	>	int probes = 0;             // step by approx half size
1417	>	int step = (n <= 4) ? 2 : ((n >>> 1) & EVENMASK) + 2;
1418	>	while (ws[r = (r + step) & m] != null) {
1419	>	if (++probes >= n) {
1420	>	workQueues = ws = Arrays.copyOf(ws, n <<= 1);
1421	>	m = n - 1;
1422	>	probes = 0;
1423		}
1424		}
1415	–	w.eventCount = w.poolIndex = r; // volatile write orders
1416	–	wt.workQueue = ws[r] = w;
1425		}
1426	<	} finally {
1427	<	if (!U.compareAndSwapInt(this, PLOCK, ps, nps))
1420	<	releasePlock(nps);
1426	>	w.eventCount = w.poolIndex = r; // volatile write orders
1427	>	ws[r] = w;
1428		}
1429	+	} finally {
1430	+	if (!U.compareAndSwapInt(this, PLOCK, ps, nps))
1431	+	releasePlock(nps);
1432		}
1433	+	wt.setName(workerNamePrefix.concat(Integer.toString(w.poolIndex)));
1434	+	return w;
1435		}
1436
1437		/**
#	Line 1460 | Line 1472 | public class ForkJoinPool extends Abstra
1472		((c - TC_UNIT) & TC_MASK) |
1473		(c & ~(AC_MASK|TC_MASK)))));
1474
1475	<	if (!tryTerminate(false, false) && w != null) {
1475	>	if (!tryTerminate(false, false) && w != null && w.array != null) {
1476		w.cancelAll();                  // cancel remaining tasks
1477	<	if (w.array != null)            // suppress signal if never ran
1478	<	helpSignal(null, 0);        // wake up or create replacement
1479	<	if (ex == null)                 // help clean refs on way out
1480	<	ForkJoinTask.helpExpungeStaleExceptions();
1477	>	int e, u, i, n; WorkQueue[] ws; WorkQueue v; Thread p;
1478	>	while ((u = (int)((c = ctl) >>> 32)) < 0) {
1479	>	if ((e = (int)c) > 0) {     // activate or create replacement
1480	>	if ((ws = workQueues) != null &&
1481	>	ws.length > (i = e & SMASK) &&
1482	>	(v = ws[i]) != null && v.eventCount == (e | INT_SIGN)) {
1483	>	long nc = (((long)(v.nextWait & E_MASK)) |
1484	>	((long)(u + UAC_UNIT) << 32));
1485	>	if (U.compareAndSwapLong(this, CTL, c, nc)) {
1486	>	v.eventCount = (e + E_SEQ) & E_MASK;
1487	>	if ((p = v.parker) != null)
1488	>	U.unpark(p);
1489	>	break;
1490	>	}
1491	>	}
1492	>	else
1493	>	break;
1494	>	}
1495	>	else {
1496	>	if ((short)u < 0)
1497	>	tryAddWorker();
1498	>	break;
1499	>	}
1500	>	}
1501		}
1502	<
1503	<	if (ex != null)                     // rethrow
1502	>	if (ex == null)                     // help clean refs on way out
1503	>	ForkJoinTask.helpExpungeStaleExceptions();
1504	>	else                                // rethrow
1505		ForkJoinTask.rethrow(ex);
1506		}
1507
#	Line 1490 | Line 1523 | public class ForkJoinPool extends Abstra
1523		U.compareAndSwapInt(q, QLOCK, 0, 1)) { // lock
1524		int b = q.base, s = q.top, n, an;
1525		if ((a = q.array) != null && (an = a.length) > (n = s + 1 - b)) {
1526	<	U.putObject(a, (long)(((an - 1) & s) << ASHIFT) + ABASE, task);
1526	>	int j = (((an - 1) & s) << ASHIFT) + ABASE;
1527	>	U.putOrderedObject(a, j, task);
1528		q.top = s + 1;                     // push on to deque
1529		q.qlock = 0;
1530		if (n <= 2)
1531	<	signalWork(q, 0);
1531	>	signalWork(q);
1532		return;
1533		}
1534		q.qlock = 0;
#	Line 1506 | Line 1540 | public class ForkJoinPool extends Abstra
1540		* Full version of externalPush. This method is called, among
1541		* other times, upon the first submission of the first task to the
1542		* pool, so must perform secondary initialization (via
1543	<	* initWorkQueuesArray). It also detects first submission by an
1544	<	* external thread by looking up its ThreadLocal, and creates a
1545	<	* new shared queue if the one at index if empty or contended. The
1546	<	* lock body must be exception-free (so no try/finally) so we
1543	>	* initWorkers). It also detects first submission by an external
1544	>	* thread by looking up its ThreadLocal, and creates a new shared
1545	>	* queue if the one at index if empty or contended. The plock lock
1546	>	* body must be exception-free (so no try/finally) so we
1547		* optimistically allocate new queues outside the lock and throw
1548		* them away if (very rarely) not needed.
1549		*/
1550		private void fullExternalPush(ForkJoinTask<?> task) {
1551	<	int r = 0;
1551	>	int r = 0; // random index seed
1552		for (Submitter z = submitters.get();;) {
1553		WorkQueue[] ws; WorkQueue q; int ps, m, k;
1554		if (z == null) {
#	Line 1532 | Line 1566 | public class ForkJoinPool extends Abstra
1566		throw new RejectedExecutionException();
1567		else if (ps == 0 || (ws = workQueues) == null ||
1568		(m = ws.length - 1) < 0)
1569	<	initWorkQueuesArray();
1569	>	initWorkers();
1570		else if ((q = ws[k = r & m & SQMASK]) != null) {
1571	<	if (q.trySharedPush(task))
1572	<	return;
1573	<	else
1574	<	r = 0; // move on contention
1571	>	if (q.qlock == 0 && U.compareAndSwapInt(q, QLOCK, 0, 1)) {
1572	>	ForkJoinTask<?>[] a = q.array;
1573	>	int s = q.top;
1574	>	boolean submitted = false;
1575	>	try {                      // locked version of push
1576	>	if ((a != null && a.length > s + 1 - q.base) ||
1577	>	(a = q.growArray()) != null) {   // must presize
1578	>	int j = (((a.length - 1) & s) << ASHIFT) + ABASE;
1579	>	U.putOrderedObject(a, j, task);
1580	>	q.top = s + 1;
1581	>	submitted = true;
1582	>	}
1583	>	} finally {
1584	>	q.qlock = 0;  // unlock
1585	>	}
1586	>	if (submitted) {
1587	>	signalWork(q);
1588	>	return;
1589	>	}
1590	>	}
1591	>	r = 0; // move on failure
1592		}
1593		else if (((ps = plock) & PL_LOCK) == 0) { // create new queue
1594		q = new WorkQueue(this, null, SHARED_QUEUE, r);
#	Line 1566 | Line 1617 | public class ForkJoinPool extends Abstra
1617		}
1618
1619		/**
1620	<	* Tries to create (at most one) or activate (possibly several)
1621	<	* workers if too few are active. On contention failure, continues
1622	<	* until at least one worker is signalled or the given queue is
1572	<	* empty or all workers are active.
1573	<	*
1574	<	* @param q if non-null, the queue holding tasks to be signalled
1575	<	* @param signals the target number of signals (at least one --
1576	<	* if argument is zero also sets signallee hint if parked).
1620	>	* Tries to create or activate a worker if too few are active.
1621	>	*
1622	>	* @param q the (non-null) queue holding tasks to be signalled
1623		*/
1624	<	final void signalWork(WorkQueue q, int signals) {
1625	<	long c; int e, u, i, s; WorkQueue[] ws; WorkQueue w; Thread p;
1624	>	final void signalWork(WorkQueue q) {
1625	>	int hint = q.poolIndex;
1626	>	long c; int e, u, i, n; WorkQueue[] ws; WorkQueue w; Thread p;
1627		while ((u = (int)((c = ctl) >>> 32)) < 0) {
1628		if ((e = (int)c) > 0) {
1629		if ((ws = workQueues) != null && ws.length > (i = e & SMASK) &&
#	Line 1584 | Line 1631 | public class ForkJoinPool extends Abstra
1631		long nc = (((long)(w.nextWait & E_MASK)) |
1632		((long)(u + UAC_UNIT) << 32));
1633		if (U.compareAndSwapLong(this, CTL, c, nc)) {
1634	+	w.hint = hint;
1635		w.eventCount = (e + E_SEQ) & E_MASK;
1636	<	if ((p = w.parker) != null) {
1589	<	if (q != null && signals == 0)
1590	<	w.hint = q.poolIndex;
1636	>	if ((p = w.parker) != null)
1637		U.unpark(p);
1638	<	}
1593	<	if (--signals <= 0)
1594	<	break;
1638	>	break;
1639		}
1640	<	if (q != null && (s = q.queueSize()) <= signals &&
1597	<	(signals = s) <= 0)
1640	>	if (q.top - q.base <= 0)
1641		break;
1642		}
1643		else
1644		break;
1645		}
1646	<	else if (e == 0 && (u & SHORT_SIGN) != 0) {
1647	<	long nc = (long)(((u + UTC_UNIT) & UTC_MASK) |
1648	<	((u + UAC_UNIT) & UAC_MASK)) << 32;
1606	<	if (U.compareAndSwapLong(this, CTL, c, nc)) {
1607	<	addWorker();
1608	<	break;
1609	<	}
1610	<	}
1611	<	else
1646	>	else {
1647	>	if ((short)u < 0)
1648	>	tryAddWorker();
1649		break;
1650	+	}
1651		}
1652		}
1653
#	Line 1619 | Line 1657 | public class ForkJoinPool extends Abstra
1657		* Top-level runloop for workers, called by ForkJoinWorkerThread.run.
1658		*/
1659		final void runWorker(WorkQueue w) {
1660	<	if (w != null) // skip on initialization failure
1661	<	do { w.runTask(scan(w)); } while (w.qlock >= 0);
1660	>	w.growArray(); // allocate queue
1661	>	do { w.runTask(scan(w)); } while (w.qlock >= 0);
1662		}
1663
1664		/**
#	Line 1659 | Line 1697 | public class ForkJoinPool extends Abstra
1697		* @return a task or null if none found
1698		*/
1699		private final ForkJoinTask<?> scan(WorkQueue w) {
1700	<	WorkQueue[] ws; int m, hint;
1700	>	WorkQueue[] ws; int m;
1701		int ps = plock;                          // read plock before ws
1702		if (w != null && (ws = workQueues) != null && (m = ws.length - 1) >= 0) {
1703		int ec = w.eventCount;               // ec is negative if inactive
1704		int r = w.seed; r ^= r << 13; r ^= r >>> 17; w.seed = r ^= r << 5;
1705	<	for (int j = ((m + m + 1) | MIN_SCAN) & MAX_SCAN; ; --j) {
1705	>	int j = ((m + m + 1) | MIN_SCAN) & MAX_SCAN;
1706	>	do {
1707		WorkQueue q; ForkJoinTask<?>[] a; int b;
1708		if ((q = ws[(r + j) & m]) != null && (b = q.base) - q.top < 0 &&
1709		(a = q.array) != null) {     // probably nonempty
#	Line 1674 | Line 1713 | public class ForkJoinPool extends Abstra
1713		if (q.base == b && ec >= 0 && t != null &&
1714		U.compareAndSwapObject(a, i, t, null)) {
1715		if ((q.base = b + 1) - q.top < 0)
1716	<	signalWork(q, 0);
1716	>	signalWork(q);
1717		return t;                // taken
1718		}
1719	<	else if (ec < 0 || j < m) {  // cannot take or cannot rescan
1720	<	w.hint = q.poolIndex;    // use hint below
1721	<	break;                   // let caller retry after signal
1722	<	}
1719	>	else if ((ec < 0 || j < m) && (int)(ctl >> AC_SHIFT) <= 0) {
1720	>	w.hint = (r + j) & m;    // help signal below
1721	>	break;                   // cannot take
1722	>	}
1723	>	}
1724	>	} while (--j >= 0);
1725	>
1726	>	long c, sc; int e, ns, h;
1727	>	if ((h = w.hint) < 0) {
1728	>	if ((ns = w.nsteals) != 0) {
1729	>	if (U.compareAndSwapLong(this, STEALCOUNT,
1730	>	sc = stealCount, sc + ns))
1731	>	w.nsteals = 0;           // collect steals
1732	>	}
1733	>	else if (plock != ps)            // consistency check
1734	>	;                            // skip
1735	>	else if ((e = (int)(c = ctl)) < 0)
1736	>	w.qlock = -1;                // pool is terminating
1737	>	else if (ec >= 0) {              // try to enqueue/inactivate
1738	>	long nc = ((long)ec | ((c - AC_UNIT) & (AC_MASK|TC_MASK)));
1739	>	w.nextWait = e;              // link and mark inactive
1740	>	w.eventCount = ec | INT_SIGN;
1741	>	if (ctl != c || !U.compareAndSwapLong(this, CTL, c, nc))
1742	>	w.eventCount = ec;       // unmark on CAS failure
1743	>	else if ((int)(c >> AC_SHIFT) == 1 - (config & SMASK))
1744	>	idleAwaitWork(w, nc, c);
1745	>	}
1746	>	else if (w.eventCount < 0) {     // block
1747	>	Thread wt = Thread.currentThread();
1748	>	Thread.interrupted();        // clear status
1749	>	U.putObject(wt, PARKBLOCKER, this);
1750	>	w.parker = wt;               // emulate LockSupport.park
1751	>	if (w.eventCount < 0)        // recheck
1752	>	U.park(false, 0L);
1753	>	w.parker = null;
1754	>	U.putObject(wt, PARKBLOCKER, null);
1755		}
1685	–	else if (j < 0) { // end of scan; in loop to simplify code
1686	–	long c, sc; int e, ns;
1687	–	if ((ns = w.nsteals) != 0) {
1688	–	if (U.compareAndSwapLong(this, STEALCOUNT,
1689	–	sc = stealCount, sc + ns))
1690	–	w.nsteals = 0;       // collect steals
1691	–	}
1692	–	else if (plock != ps)        // ws may have changed
1693	–	break;
1694	–	else if ((e = (int)(c = ctl)) < 0)
1695	–	w.qlock = -1;            // pool is terminating
1696	–	else if (ec >= 0) {          // try to enqueue/inactivate
1697	–	long nc = ((long)ec |
1698	–	((c - AC_UNIT) & (AC_MASK|TC_MASK)));
1699	–	w.nextWait = e;          // link and mark inactive
1700	–	w.hint = -1;             // use hint if set while parked
1701	–	w.eventCount = ec | INT_SIGN;
1702	–	if (ctl != c ||
1703	–	!U.compareAndSwapLong(this, CTL, c, nc))
1704	–	w.eventCount = ec;  // unmark on CAS failure
1705	–	else if ((int)(c >> AC_SHIFT) == 1 - (config & SMASK))
1706	–	idleAwaitWork(w, nc, c);
1707	–	}
1708	–	else if (w.eventCount < 0) { // block
1709	–	Thread wt = Thread.currentThread();
1710	–	Thread.interrupted();    // clear status
1711	–	U.putObject(wt, PARKBLOCKER, this);
1712	–	w.parker = wt;           // emulate LockSupport.park
1713	–	if (w.eventCount < 0)    // recheck
1714	–	U.park(false, 0L);
1715	–	w.parker = null;
1716	–	U.putObject(wt, PARKBLOCKER, null);
1717	–	}
1718	–	break;
1719	–	}
1720	–	}
1721	–	if ((hint = w.hint) >= 0) {          // help signal
1722	–	WorkQueue[] vs; WorkQueue v; int k;
1723	–	w.hint = -1;                     // suppress resignal
1724	–	if ((vs = workQueues) != null && hint < vs.length &&
1725	–	(v = vs[hint]) != null && (k = v.base - v.top) < -1)
1726	–	signalWork(v, 1 - k);
1756		}
1757	+	if (h >= 0 || w.hint >= 0)           // signal others before retry
1758	+	helpSignalHint(w);
1759		}
1760		return null;
1761		}
#	Line 1746 | Line 1777 | public class ForkJoinPool extends Abstra
1777		!tryTerminate(false, false) && (int)prevCtl != 0) {
1778		int dc = -(short)(currentCtl >>> TC_SHIFT);
1779		long parkTime = dc < 0 ? FAST_IDLE_TIMEOUT: (dc + 1) * IDLE_TIMEOUT;
1780	<	long deadline = System.nanoTime() + parkTime - 100000L; // 1ms slop
1780	>	long deadline = System.nanoTime() + parkTime - TIMEOUT_SLOP;
1781		Thread wt = Thread.currentThread();
1782		while (ctl == currentCtl) {
1783		Thread.interrupted();  // timed variant of version in scan()
#	Line 1762 | Line 1793 | public class ForkJoinPool extends Abstra
1793		U.compareAndSwapLong(this, CTL, currentCtl, prevCtl)) {
1794		w.eventCount = (w.eventCount + E_SEQ) | E_MASK;
1795		w.qlock = -1;   // shrink
1765	–	w.hint = -1;    // suppress helping
1796		break;
1797		}
1798		}
#	Line 1770 | Line 1800 | public class ForkJoinPool extends Abstra
1800		}
1801
1802		/**
1803	<	* Scans through queues looking for work (optionally, while
1804	<	* joining a task); if any are present, signals. May return early
1805	<	* if more signalling is detectably unneeded.
1803	>	* Scans through queues looking for work while joining a task; if
1804	>	* any present, signals. May return early if more signalling is
1805	>	* detectably unneeded.
1806		*
1807	<	* @param task if non-null, return early if done
1807	>	* @param task return early if done
1808		* @param origin an index to start scan
1809		*/
1810	<	final int helpSignal(ForkJoinTask<?> task, int origin) {
1811	<	WorkQueue[] ws; WorkQueue q; int m, n, s, u;
1812	<	if ((ws = workQueues) != null && (m = ws.length - 1) >= 0) {
1813	<	for (int i = 0; i <= m; ++i) {
1814	<	if (task != null && (s = task.status) < 0)
1815	<	return s;
1816	<	if ((q = ws[(i + origin) & m]) != null &&
1817	<	(n = q.queueSize()) > 0) {
1818	<	signalWork(q, n);
1819	<	if ((u = (int)(ctl >>> 32)) >= 0 || (u >> UAC_SHIFT) >= 0)
1810	>	private void helpSignal(ForkJoinTask<?> task, int origin) {
1811	>	WorkQueue[] ws; WorkQueue w; Thread p; long c; int m, u, e, i, s;
1812	>	if (task != null && task.status >= 0 &&
1813	>	(u = (int)(ctl >>> 32)) < 0 && (u >> UAC_SHIFT) < 0 &&
1814	>	(ws = workQueues) != null && (m = ws.length - 1) >= 0) {
1815	>	outer: for (int k = origin, j = m; j >= 0; --j) {
1816	>	WorkQueue q = ws[k++ & m];
1817	>	for (int n = m;;) { // limit to at most m signals
1818	>	if (task.status < 0)
1819	>	break outer;
1820	>	if (q == null ||
1821	>	((s = -q.base + q.top) <= n && (n = s) <= 0))
1822	>	break;
1823	>	if ((u = (int)((c = ctl) >>> 32)) >= 0 ||
1824	>	(e = (int)c) <= 0 || m < (i = e & SMASK) ||
1825	>	(w = ws[i]) == null)
1826	>	break outer;
1827	>	long nc = (((long)(w.nextWait & E_MASK)) |
1828	>	((long)(u + UAC_UNIT) << 32));
1829	>	if (w.eventCount == (e | INT_SIGN) &&
1830	>	U.compareAndSwapLong(this, CTL, c, nc)) {
1831	>	w.eventCount = (e + E_SEQ) & E_MASK;
1832	>	if ((p = w.parker) != null)
1833	>	U.unpark(p);
1834	>	if (--n <= 0)
1835	>	break;
1836	>	}
1837	>	}
1838	>	}
1839	>	}
1840	>	}
1841	>
1842	>	/**
1843	>	* Signals other workers if tasks are present in hinted queue.
1844	>	*
1845	>	* @param caller the worker with the hint
1846	>	*/
1847	>	private void helpSignalHint(WorkQueue caller) {
1848	>	WorkQueue[] ws; WorkQueue q, w; Thread p; long c; int h, m, u, e, i, s;
1849	>	if (caller != null && (h = caller.hint) >= 0) {
1850	>	caller.hint = -1;
1851	>	if ((u = (int)(ctl >>> 32)) < 0 && (u >> UAC_SHIFT) < 0 &&
1852	>	(ws = workQueues) != null && (m = ws.length - 1) >= 0 &&
1853	>	(q = ws[h & m]) != null) {
1854	>	for (int n = 2;;) { // limit to at most 2 signals
1855	>	int idleCount = (caller.eventCount < 0) ? 0 : -1;
1856	>	if (((s = idleCount - q.base + q.top) <= n &&
1857	>	(n = s) <= 0) ||
1858	>	(u = (int)((c = ctl) >>> 32)) >= 0 ||
1859	>	(e = (int)c) <= 0 || m < (i = e & SMASK) ||
1860	>	(w = ws[i]) == null)
1861		break;
1862	+	long nc = (((long)(w.nextWait & E_MASK)) |
1863	+	((long)(u + UAC_UNIT) << 32));
1864	+	if (w.eventCount == (e | INT_SIGN) &&
1865	+	U.compareAndSwapLong(this, CTL, c, nc)) {
1866	+	w.hint = h;
1867	+	w.eventCount = (e + E_SEQ) & E_MASK;
1868	+	if ((p = w.parker) != null)
1869	+	U.unpark(p);
1870	+	if (--n <= 0)
1871	+	break;
1872	+	}
1873		}
1874		}
1875		}
1794	–	return 0;
1876		}
1877
1878		/**
#	Line 1940 | Line 2021 | public class ForkJoinPool extends Abstra
2021		else if (tc + pc < MAX_CAP) {
2022		long nc = ((c + TC_UNIT) & TC_MASK) | (c & ~TC_MASK);
2023		if (U.compareAndSwapLong(this, CTL, c, nc)) {
2024	<	addWorker();
2025	<	return true;
2024	>	ForkJoinWorkerThreadFactory fac;
2025	>	Throwable ex = null;
2026	>	ForkJoinWorkerThread wt = null;
2027	>	try {
2028	>	if ((fac = factory) != null &&
2029	>	(wt = fac.newThread(this)) != null) {
2030	>	wt.start();
2031	>	return true;
2032	>	}
2033	>	} catch (Throwable rex) {
2034	>	ex = rex;
2035	>	}
2036	>	deregisterWorker(wt, ex); // clean up and return false
2037		}
2038		}
2039		}
#	Line 1960 | Line 2052 | public class ForkJoinPool extends Abstra
2052		if (joiner != null && task != null && (s = task.status) >= 0) {
2053		ForkJoinTask<?> prevJoin = joiner.currentJoin;
2054		joiner.currentJoin = task;
2055	<	do {} while ((s = task.status) >= 0 &&
1964	<	joiner.queueSize() > 0 &&
2055	>	do {} while ((s = task.status) >= 0 && !joiner.isEmpty() &&
2056		joiner.tryRemoveAndExec(task)); // process local tasks
2057	<	if (s >= 0 && (s = task.status) >= 0 &&
2058	<	(s = helpSignal(task, joiner.poolIndex)) >= 0 &&
2059	<	(task instanceof CountedCompleter))
2060	<	s = helpComplete(task, LIFO_QUEUE);
2061	<	int k = 0; // to perform pre-block yield for politeness
2057	>	if (s >= 0 && (s = task.status) >= 0) {
2058	>	helpSignal(task, joiner.poolIndex);
2059	>	if ((s = task.status) >= 0 &&
2060	>	(task instanceof CountedCompleter))
2061	>	s = helpComplete(task, LIFO_QUEUE);
2062	>	}
2063		while (s >= 0 && (s = task.status) >= 0) {
2064	<	if ((joiner.queueSize() > 0 ||           // try helping
2064	>	if ((!joiner.isEmpty() ||           // try helping
2065		(s = tryHelpStealer(joiner, task)) == 0) &&
2066		(s = task.status) >= 0) {
2067	<	if (k < 3) {
2068	<	if (++k < 3)
1977	<	s = helpSignal(task, joiner.poolIndex);
1978	<	else
1979	<	Thread.yield();
1980	<	}
1981	<	else if (!tryCompensate())
1982	<	k = 0;
1983	<	else {
2067	>	helpSignal(task, joiner.poolIndex);
2068	>	if ((s = task.status) >= 0 && tryCompensate()) {
2069		if (task.trySetSignal() && (s = task.status) >= 0) {
2070		synchronized (task) {
2071		if (task.status >= 0) {
#	Line 2017 | Line 2102 | public class ForkJoinPool extends Abstra
2102		if (joiner != null && task != null && (s = task.status) >= 0) {
2103		ForkJoinTask<?> prevJoin = joiner.currentJoin;
2104		joiner.currentJoin = task;
2105	<	do {} while ((s = task.status) >= 0 &&
2021	<	joiner.queueSize() > 0 &&
2105	>	do {} while ((s = task.status) >= 0 && !joiner.isEmpty() &&
2106		joiner.tryRemoveAndExec(task));
2107	<	if (s >= 0 && (s = task.status) >= 0 &&
2108	<	(s = helpSignal(task, joiner.poolIndex)) >= 0 &&
2109	<	(task instanceof CountedCompleter))
2110	<	s = helpComplete(task, LIFO_QUEUE);
2111	<	if (s >= 0 && joiner.queueSize() == 0) {
2107	>	if (s >= 0 && (s = task.status) >= 0) {
2108	>	helpSignal(task, joiner.poolIndex);
2109	>	if ((s = task.status) >= 0 &&
2110	>	(task instanceof CountedCompleter))
2111	>	s = helpComplete(task, LIFO_QUEUE);
2112	>	}
2113	>	if (s >= 0 && joiner.isEmpty()) {
2114		do {} while (task.status >= 0 &&
2115		tryHelpStealer(joiner, task) > 0);
2116		}
#	Line 2046 | Line 2132 | public class ForkJoinPool extends Abstra
2132		return null;
2133		for (int j = (m + 1) << 2; ;) {
2134		WorkQueue q = ws[(((r + j) << 1) | 1) & m];
2135	<	if (q != null && (n = q.queueSize()) > 0) {
2136	<	if (n > 1)
2137	<	signalWork(q, 0);
2135	>	if (q != null && (n = q.base - q.top) < 0) {
2136	>	if (n < -1)
2137	>	signalWork(q);
2138		return q;
2139		}
2140		else if (--j < 0) {
#	Line 2290 | Line 2376 | public class ForkJoinPool extends Abstra
2376		*/
2377		static boolean tryExternalUnpush(ForkJoinTask<?> t) {
2378		ForkJoinPool p; WorkQueue[] ws; WorkQueue q; Submitter z;
2379	<	ForkJoinTask<?>[] a;  int m, s; long j;
2380	<	if ((z = submitters.get()) != null &&
2379	>	ForkJoinTask<?>[] a;  int m, s;
2380	>	if (t != null &&
2381	>	(z = submitters.get()) != null &&
2382		(p = commonPool) != null &&
2383		(ws = p.workQueues) != null &&
2384		(m = ws.length - 1) >= 0 &&
2385		(q = ws[m & z.seed & SQMASK]) != null &&
2386		(s = q.top) != q.base &&
2387	<	(a = q.array) != null &&
2388	<	U.getObjectVolatile
2389	<	(a, j = (((a.length - 1) & (s - 1)) << ASHIFT) + ABASE) == t &&
2390	<	U.compareAndSwapInt(q, QLOCK, 0, 1)) {
2391	<	if (q.array == a && q.top == s && // recheck
2392	<	U.compareAndSwapObject(a, j, t, null)) {
2393	<	q.top = s - 1;
2387	>	(a = q.array) != null) {
2388	>	long j = (((a.length - 1) & (s - 1)) << ASHIFT) + ABASE;
2389	>	if (U.getObject(a, j) == t &&
2390	>	U.compareAndSwapInt(q, QLOCK, 0, 1)) {
2391	>	if (q.array == a && q.top == s && // recheck
2392	>	U.compareAndSwapObject(a, j, t, null)) {
2393	>	q.top = s - 1;
2394	>	q.qlock = 0;
2395	>	return true;
2396	>	}
2397		q.qlock = 0;
2308	–	return true;
2398		}
2310	–	q.qlock = 0;
2399		}
2400		return false;
2401		}
#	Line 2349 | Line 2437 | public class ForkJoinPool extends Abstra
2437		(u = (int)(ctl >>> 32)) >= 0 || (u >> UAC_SHIFT) >= 0)
2438		break;
2439		if (task == null) {
2440	<	if (helpSignal(root, q.poolIndex) >= 0)
2440	>	helpSignal(root, q.poolIndex);
2441	>	if (root.status >= 0)
2442		helpComplete(root, SHARED_QUEUE);
2443		break;
2444		}
#	Line 2364 | Line 2453 | public class ForkJoinPool extends Abstra
2453		static void externalHelpJoin(ForkJoinTask<?> t) {
2454		// Some hard-to-avoid overlap with tryExternalUnpush
2455		ForkJoinPool p; WorkQueue[] ws; WorkQueue q, w; Submitter z;
2456	<	ForkJoinTask<?>[] a;  int m, s, n; long j;
2456	>	ForkJoinTask<?>[] a;  int m, s, n;
2457		if (t != null &&
2458		(z = submitters.get()) != null &&
2459		(p = commonPool) != null &&
2460		(ws = p.workQueues) != null &&
2461		(m = ws.length - 1) >= 0 &&
2462		(q = ws[m & z.seed & SQMASK]) != null &&
2463	<	(a = q.array) != null &&
2464	<	t.status >= 0) {
2465	<	if ((s = q.top) != q.base &&
2466	<	U.getObjectVolatile
2467	<	(a, j = (((a.length - 1) & (s - 1)) << ASHIFT) + ABASE) == t &&
2468	<	U.compareAndSwapInt(q, QLOCK, 0, 1)) {
2469	<	if (q.array == a && q.top == s &&
2470	<	U.compareAndSwapObject(a, j, t, null)) {
2471	<	q.top = s - 1;
2472	<	q.qlock = 0;
2473	<	t.doExec();
2463	>	(a = q.array) != null) {
2464	>	int am = a.length - 1;
2465	>	if ((s = q.top) != q.base) {
2466	>	long j = ((am & (s - 1)) << ASHIFT) + ABASE;
2467	>	if (U.getObject(a, j) == t &&
2468	>	U.compareAndSwapInt(q, QLOCK, 0, 1)) {
2469	>	if (q.array == a && q.top == s &&
2470	>	U.compareAndSwapObject(a, j, t, null)) {
2471	>	q.top = s - 1;
2472	>	q.qlock = 0;
2473	>	t.doExec();
2474	>	}
2475	>	else
2476	>	q.qlock = 0;
2477		}
2386	–	else
2387	–	q.qlock = 0;
2478		}
2479		if (t.status >= 0) {
2480		if (t instanceof CountedCompleter)
#	Line 2480 | Line 2570 | public class ForkJoinPool extends Abstra
2570		throw new IllegalArgumentException();
2571		this.factory = factory;
2572		this.ueh = handler;
2573	<	this.config = parallelism | (asyncMode? (FIFO_QUEUE << 16) : 0);
2573	>	this.config = parallelism | (asyncMode ? (FIFO_QUEUE << 16) : 0);
2574		long np = (long)(-parallelism); // offset ctl counts
2575		this.ctl = ((np << AC_SHIFT) & AC_MASK) | ((np << TC_SHIFT) & TC_MASK);
2576		int pn = nextPoolId();
#	Line 2838 | Line 2928 | public class ForkJoinPool extends Abstra
2928		WorkQueue[] ws; WorkQueue w;
2929		if ((ws = workQueues) != null) {
2930		for (int i = 0; i < ws.length; i += 2) {
2931	<	if ((w = ws[i]) != null && w.queueSize() != 0)
2931	>	if ((w = ws[i]) != null && !w.isEmpty())
2932		return true;
2933		}
2934		}
#	Line 3005 | Line 3095 | public class ForkJoinPool extends Abstra
3095		* commenced but not yet completed.  This method may be useful for
3096		* debugging. A return of {@code true} reported a sufficient
3097		* period after shutdown may indicate that submitted tasks have
3098	<	* ignored or suppressed interruption, or are waiting for IO,
3098	>	* ignored or suppressed interruption, or are waiting for I/O,
3099		* causing this executor not to properly terminate. (See the
3100		* advisory notes for class {@link ForkJoinTask} stating that
3101		* tasks should not normally entail blocking operations.  But if
#	Line 3160 | Line 3250 | public class ForkJoinPool extends Abstra
3250		if (t instanceof ForkJoinWorkerThread) {
3251		ForkJoinPool p = ((ForkJoinWorkerThread)t).pool;
3252		while (!blocker.isReleasable()) { // variant of helpSignal
3253	<	WorkQueue[] ws; WorkQueue q; int m, n, u;
3253	>	WorkQueue[] ws; WorkQueue q; int m, u;
3254		if ((ws = p.workQueues) != null && (m = ws.length - 1) >= 0) {
3255		for (int i = 0; i <= m; ++i) {
3256		if (blocker.isReleasable())
3257		return;
3258	<	if ((q = ws[i]) != null && (n = q.queueSize()) > 0) {
3259	<	p.signalWork(q, n);
3258	>	if ((q = ws[i]) != null && q.base - q.top < 0) {
3259	>	p.signalWork(q);
3260		if ((u = (int)(p.ctl >>> 32)) >= 0 ||
3261		(u >> UAC_SHIFT) >= 0)
3262		break;
#	Line 3245 | Line 3335 | public class ForkJoinPool extends Abstra
3335		submitters = new ThreadLocal<Submitter>();
3336		ForkJoinWorkerThreadFactory fac = defaultForkJoinWorkerThreadFactory =
3337		new DefaultForkJoinWorkerThreadFactory();
3338	+	modifyThreadPermission = new RuntimePermission("modifyThread");
3339	+
3340		/*
3341		* Establish common pool parameters.  For extra caution,
3342		* computations to set up common pool state are here; the
#	Line 3280 | Line 3372 | public class ForkJoinPool extends Abstra
3372		long ct = ((np << AC_SHIFT) & AC_MASK) | ((np << TC_SHIFT) & TC_MASK);
3373
3374		commonPool = new ForkJoinPool(par, ct, fac, handler);
3283	–	modifyThreadPermission = new RuntimePermission("modifyThread");
3375		}
3376
3377		/**

Diff Legend

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