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.55 by dl, Sun Apr 18 13:59:57 2010 UTC vs.
Revision 1.57 by dl, Wed Jul 7 19:52:31 2010 UTC

#	Line 21 | Line 21 | import java.util.concurrent.CountDownLat
21		/**
22		* An {@link ExecutorService} for running {@link ForkJoinTask}s.
23		* A {@code ForkJoinPool} provides the entry point for submissions
24	<	* from non-{@code ForkJoinTask}s, as well as management and
24	>	* from non-{@code ForkJoinTask} clients, as well as management and
25		* monitoring operations.
26		*
27		* <p>A {@code ForkJoinPool} differs from other kinds of {@link
#	Line 30 | Line 30 | import java.util.concurrent.CountDownLat
30		* execute subtasks created by other active tasks (eventually blocking
31		* waiting for work if none exist). This enables efficient processing
32		* when most tasks spawn other subtasks (as do most {@code
33	<	* ForkJoinTask}s). A {@code ForkJoinPool} may also be used for mixed
34	<	* execution of some plain {@code Runnable}- or {@code Callable}-
35	<	* based activities along with {@code ForkJoinTask}s. When setting
36	<	* {@linkplain #setAsyncMode async mode}, a {@code ForkJoinPool} may
37	<	* also be appropriate for use with fine-grained tasks of any form
38	<	* that are never joined. Otherwise, other {@code ExecutorService}
39	<	* implementations are typically more appropriate choices.
33	>	* ForkJoinTask}s). When setting <em>asyncMode</em> to true in
34	>	* constructors, {@code ForkJoinPool}s may also be appropriate for use
35	>	* with event-style tasks that are never joined.
36		*
37		* <p>A {@code ForkJoinPool} is constructed with a given target
38		* parallelism level; by default, equal to the number of available
39	<	* processors. Unless configured otherwise via {@link
40	<	* #setMaintainsParallelism}, the pool attempts to maintain this
41	<	* number of active (or available) threads by dynamically adding,
42	<	* suspending, or resuming internal worker threads, even if some tasks
43	<	* are stalled waiting to join others. However, no such adjustments
44	<	* are performed in the face of blocked IO or other unmanaged
45	<	* synchronization. The nested {@link ManagedBlocker} interface
50	<	* enables extension of the kinds of synchronization accommodated.
51	<	* The target parallelism level may also be changed dynamically
52	<	* ({@link #setParallelism}). The total number of threads may be
53	<	* limited using method {@link #setMaximumPoolSize}, in which case it
54	<	* may become possible for the activities of a pool to stall due to
55	<	* the lack of available threads to process new tasks. When the pool
56	<	* is executing tasks, these and other configuration setting methods
57	<	* may only gradually affect actual pool sizes. It is normally best
58	<	* practice to invoke these methods only when the pool is known to be
59	<	* quiescent.
39	>	* processors. The pool attempts to maintain enough active (or
40	>	* available) threads by dynamically adding, suspending, or resuming
41	>	* internal worker threads, even if some tasks are stalled waiting to
42	>	* join others. However, no such adjustments are guaranteed in the
43	>	* face of blocked IO or other unmanaged synchronization. The nested
44	>	* {@link ManagedBlocker} interface enables extension of the kinds of
45	>	* synchronization accommodated.
46		*
47		* <p>In addition to execution and lifecycle control methods, this
48		* class provides status check methods (for example
#	Line 65 | Line 51 | import java.util.concurrent.CountDownLat
51		* {@link #toString} returns indications of pool state in a
52		* convenient form for informal monitoring.
53		*
54	+	* <p> As is the case with other ExecutorServices, there are three
55	+	* main task execution methods summarized in the follwoing
56	+	* table. These are designed to be used by clients not already engaged
57	+	* in fork/join computations in the current pool.  The main forms of
58	+	* these methods accept instances of {@code ForkJoinTask}, but
59	+	* overloaded forms also allow mixed execution of plain {@code
60	+	* Runnable}- or {@code Callable}- based activities as well.  However,
61	+	* tasks that are already executing in a pool should normally
62	+	* <em>NOT</em> use these pool execution methods, but instead use the
63	+	* within-computation forms listed in the table. To avoid inadvertant
64	+	* cyclic task dependencies and to improve performance, task
65	+	* submissions to the current pool by an ongoing fork/join
66	+	* computations may be implicitly translated to the corresponding
67	+	* ForkJoinTask forms.
68	+	*
69	+	* <table BORDER CELLPADDING=3 CELLSPACING=1>
70	+	*  <tr>
71	+	*    <td></td>
72	+	*    <td ALIGN=CENTER> <b>Call from non-fork/join clients</b></td>
73	+	*    <td ALIGN=CENTER> <b>Call from within fork/join computations</b></td>
74	+	*  </tr>
75	+	*  <tr>
76	+	*    <td> <b>Arange async execution</td>
77	+	*    <td> {@link #execute(ForkJoinTask)}</td>
78	+	*    <td> {@link ForkJoinTask#fork}</td>
79	+	*  </tr>
80	+	*  <tr>
81	+	*    <td> <b>Await and obtain result</td>
82	+	*    <td> {@link #invoke(ForkJoinTask)}</td>
83	+	*    <td> {@link ForkJoinTask#invoke}</td>
84	+	*  </tr>
85	+	*  <tr>
86	+	*    <td> <b>Arrange exec and obtain Future</td>
87	+	*    <td> {@link #submit(ForkJoinTask)}</td>
88	+	*    <td> {@link ForkJoinTask#fork} (ForkJoinTasks <em>are</em> Futures)</td>
89	+	*  </tr>
90	+	* </table>
91	+	*
92		* <p><b>Sample Usage.</b> Normally a single {@code ForkJoinPool} is
93		* used for all parallel task execution in a program or subsystem.
94		* Otherwise, use would not usually outweigh the construction and
#	Line 140 | Line 164 | public class ForkJoinPool extends Abstra
164		* (workerLock) but the array is otherwise concurrently readable,
165		* and accessed directly by workers. To simplify index-based
166		* operations, the array size is always a power of two, and all
167	<	* readers must tolerate null slots. Currently, all but the first
168	<	* worker thread creation is on-demand, triggered by task
169	<	* submissions, replacement of terminated workers, and/or
170	<	* compensation for blocked workers. However, all other support
171	<	* code is set up to work with other policies.
167	>	* readers must tolerate null slots. Currently, all worker thread
168	>	* creation is on-demand, triggered by task submissions,
169	>	* replacement of terminated workers, and/or compensation for
170	>	* blocked workers. However, all other support code is set up to
171	>	* work with other policies.
172		*
173		* 2. Bookkeeping for dynamically adding and removing workers. We
174	<	* maintain a given level of parallelism (or, if
175	<	* maintainsParallelism is false, at least avoid starvation). When
152	<	* some workers are known to be blocked (on joins or via
174	>	* aim to approximately maintain the given level of parallelism.
175	>	* When some workers are known to be blocked (on joins or via
176		* ManagedBlocker), we may create or resume others to take their
177		* place until they unblock (see below). Implementing this
178		* requires counts of the number of "running" threads (i.e., those
#	Line 157 | Line 180 | public class ForkJoinPool extends Abstra
180		* the total number.  These two values are packed into one field,
181		* "workerCounts" because we need accurate snapshots when deciding
182		* to create, resume or suspend.  To support these decisions,
183	<	* updates must be prospective (not retrospective).  For example,
184	<	* the running count is decremented before blocking by a thread
185	<	* about to block, but incremented by the thread about to unblock
186	<	* it. (In a few cases, these prospective updates may need to be
187	<	* rolled back, for example when deciding to create a new worker
188	<	* but the thread factory fails or returns null. In these cases,
189	<	* we are no worse off wrt other decisions than we would be
190	<	* otherwise.)  Updates to the workerCounts field sometimes
191	<	* transiently encounter a fair amount of contention when join
192	<	* dependencies are such that many threads block or unblock at
193	<	* about the same time. We alleviate this by sometimes bundling
194	<	* updates (for example blocking one thread on join and resuming a
172	<	* spare cancel each other out), and in most other cases
173	<	* performing an alternative action (like releasing waiters and
174	<	* finding spares; see below) as a more productive form of
175	<	* backoff.
183	>	* updates to spare counts must be prospective (not
184	>	* retrospective).  For example, the running count is decremented
185	>	* before blocking by a thread about to block as a spare, but
186	>	* incremented by the thread about to unblock it. Updates upon
187	>	* resumption ofr threads blocking in awaitJoin or awaitBlocker
188	>	* cannot usually be prospective, so the running count is in
189	>	* general an upper bound of the number of productively running
190	>	* threads Updates to the workerCounts field sometimes transiently
191	>	* encounter a fair amount of contention when join dependencies
192	>	* are such that many threads block or unblock at about the same
193	>	* time. We alleviate this by sometimes performing an alternative
194	>	* action on contention like releasing waiters or locating spares.
195		*
196		* 3. Maintaining global run state. The run state of the pool
197		* consists of a runLevel (SHUTDOWN, TERMINATING, etc) similar to
#	Line 221 | Line 240 | public class ForkJoinPool extends Abstra
240		* that only releases idle workers until it detects interference
241		* by other threads trying to release, and lets them take
242		* over. The net effect is a tree-like diffusion of signals, where
243	<	* released threads and possibly others) help with unparks.  To
243	>	* released threads (and possibly others) help with unparks.  To
244		* further reduce contention effects a bit, failed CASes to
245		* increment field eventCount are tolerated without retries.
246		* Conceptually they are merged into the same event, which is OK
#	Line 238 | Line 257 | public class ForkJoinPool extends Abstra
257		* "extra" spare threads from normal "core" threads: On each call
258		* to preStep (the only point at which we can do this) a worker
259		* checks to see if there are now too many running workers, and if
260	<	* so, suspends itself.  Methods preJoin and doBlock look for
261	<	* suspended threads to resume before considering creating a new
262	<	* replacement. We don't need a special data structure to maintain
263	<	* spares; simply scanning the workers array looking for
260	>	* so, suspends itself.  Methods awaitJoin and awaitBlocker look
261	>	* for suspended threads to resume before considering creating a
262	>	* new replacement. We don't need a special data structure to
263	>	* maintain spares; simply scanning the workers array looking for
264		* worker.isSuspended() is fine because the calling thread is
265		* otherwise not doing anything useful anyway; we are at least as
266		* happy if after locating a spare, the caller doesn't actually
#	Line 260 | Line 279 | public class ForkJoinPool extends Abstra
279		*
280		* 6. Deciding when to create new workers. The main dynamic
281		* control in this class is deciding when to create extra threads,
282	<	* in methods preJoin and doBlock. We always need to create one
283	<	* when the number of running threads becomes zero. But because
284	<	* blocked joins are typically dependent, we don't necessarily
285	<	* need or want one-to-one replacement. Using a one-to-one
286	<	* compensation rule often leads to enough useless overhead
287	<	* creating, suspending, resuming, and/or killing threads to
288	<	* signficantly degrade throughput.  We use a rule reflecting the
289	<	* idea that, the more spare threads you already have, the more
290	<	* evidence you need to create another one; where "evidence" is
291	<	* expressed as the current deficit -- target minus running
292	<	* threads. To reduce flickering and drift around target values,
293	<	* the relation is quadratic: adding a spare if (dc*dc)>=(sc*pc)
294	<	* (where dc is deficit, sc is number of spare threads and pc is
295	<	* target parallelism.)  This effectively reduces churn at the
296	<	* price of systematically undershooting target parallelism when
297	<	* many threads are blocked.  However, biasing toward undeshooting
298	<	* partially compensates for the above mechanics to suspend extra
280	<	* threads, that normally lead to overshoot because we can only
281	<	* suspend workers in-between top-level actions. It also better
282	<	* copes with the fact that some of the methods in this class tend
283	<	* to never become compiled (but are interpreted), so some
284	<	* components of the entire set of controls might execute many
285	<	* times faster than others. And similarly for cases where the
286	<	* apparent lack of work is just due to GC stalls and other
287	<	* transient system activity.
288	<	*
289	<	* 7. Maintaining other configuration parameters and monitoring
290	<	* statistics. Updates to fields controlling parallelism level,
291	<	* max size, etc can only meaningfully take effect for individual
292	<	* threads upon their next top-level actions; i.e., between
293	<	* stealing/running tasks/submission, which are separated by calls
294	<	* to preStep.  Memory ordering for these (assumed infrequent)
295	<	* reconfiguration calls is ensured by using reads and writes to
296	<	* volatile field workerCounts (that must be read in preStep anyway)
297	<	* as "fences" -- user-level reads are preceded by reads of
298	<	* workCounts, and writes are followed by no-op CAS to
299	<	* workerCounts. The values reported by other management and
300	<	* monitoring methods are either computed on demand, or are kept
301	<	* in fields that are only updated when threads are otherwise
302	<	* idle.
282	>	* in methods awaitJoin and awaitBlocker. We always need to create
283	>	* one when the number of running threads becomes zero. But
284	>	* because blocked joins are typically dependent, we don't
285	>	* necessarily need or want one-to-one replacement. Instead, we
286	>	* use a combination of heuristics that adds threads only when the
287	>	* pool appears to be approaching starvation.  These effectively
288	>	* reduce churn at the price of systematically undershooting
289	>	* target parallelism when many threads are blocked.  However,
290	>	* biasing toward undeshooting partially compensates for the above
291	>	* mechanics to suspend extra threads, that normally lead to
292	>	* overshoot because we can only suspend workers in-between
293	>	* top-level actions. It also better copes with the fact that some
294	>	* of the methods in this class tend to never become compiled (but
295	>	* are interpreted), so some components of the entire set of
296	>	* controls might execute many times faster than others. And
297	>	* similarly for cases where the apparent lack of work is just due
298	>	* to GC stalls and other transient system activity.
299		*
300		* Beware that there is a lot of representation-level coupling
301		* among classes ForkJoinPool, ForkJoinWorkerThread, and
#	Line 345 | Line 341 | public class ForkJoinPool extends Abstra
341		* Default ForkJoinWorkerThreadFactory implementation; creates a
342		* new ForkJoinWorkerThread.
343		*/
344	<	static class  DefaultForkJoinWorkerThreadFactory
344	>	static class DefaultForkJoinWorkerThreadFactory
345		implements ForkJoinWorkerThreadFactory {
346		public ForkJoinWorkerThread newThread(ForkJoinPool pool) {
347		return new ForkJoinWorkerThread(pool);
#	Line 413 | Line 409 | public class ForkJoinPool extends Abstra
409		/**
410		* Latch released upon termination.
411		*/
412	<	private final CountDownLatch terminationLatch;
412	>	private final Phaser termination;
413
414		/**
415		* Creation factory for worker threads.
#	Line 484 | Line 480 | public class ForkJoinPool extends Abstra
480		private static final int ONE_RUNNING        = 1;
481		private static final int ONE_TOTAL          = 1 << TOTAL_COUNT_SHIFT;
482
487	–	/*
488	–	* Fields parallelism. maxPoolSize, locallyFifo,
489	–	* maintainsParallelism, and ueh are non-volatile, but external
490	–	* reads/writes use workerCount fences to ensure visability.
491	–	*/
492	–
483		/**
484		* The target parallelism level.
485	+	* Accessed directly by ForkJoinWorkerThreads.
486		*/
487	<	private int parallelism;
497	<
498	<	/**
499	<	* The maximum allowed pool size.
500	<	*/
501	<	private int maxPoolSize;
487	>	final int parallelism;
488
489		/**
490		* True if use local fifo, not default lifo, for local polling
491	<	* Replicated by ForkJoinWorkerThreads
491	>	* Read by, and replicated by ForkJoinWorkerThreads
492		*/
493	<	private boolean locallyFifo;
493	>	final boolean locallyFifo;
494
495		/**
496	<	* Controls whether to add spares to maintain parallelism
496	>	* The uncaught exception handler used when any worker abruptly
497	>	* terminates.
498		*/
499	<	private boolean maintainsParallelism;
513	<
514	<	/**
515	<	* The uncaught exception handler used when any worker
516	<	* abruptly terminates
517	<	*/
518	<	private Thread.UncaughtExceptionHandler ueh;
499	>	private final Thread.UncaughtExceptionHandler ueh;
500
501		/**
502		* Pool number, just for assigning useful names to worker threads
#	Line 525 | Line 506 | public class ForkJoinPool extends Abstra
506		// utilities for updating fields
507
508		/**
509	<	* Adds delta to running count.  Used mainly by ForkJoinTask.
529	<	*
530	<	* @param delta the number to add
509	>	* Increments running count.  Also used by ForkJoinTask.
510		*/
511	<	final void updateRunningCount(int delta) {
512	<	int wc;
511	>	final void incrementRunningCount() {
512	>	int c;
513		do {} while (!UNSAFE.compareAndSwapInt(this, workerCountsOffset,
514	<	wc = workerCounts,
515	<	wc + delta));
537	<	}
538	<
539	<	/**
540	<	* Write fence for user modifications of pool parameters
541	<	* (parallelism. etc).  Note that it doesn't matter if CAS fails.
542	<	*/
543	<	private void workerCountWriteFence() {
544	<	int wc;
545	<	UNSAFE.compareAndSwapInt(this, workerCountsOffset,
546	<	wc = workerCounts, wc);
514	>	c = workerCounts,
515	>	c + ONE_RUNNING));
516		}
517	<
517	>
518		/**
519	<	* Read fence for external reads of pool parameters
551	<	* (parallelism. maxPoolSize, etc).
519	>	* Tries to decrement running count unless already zero
520		*/
521	<	private void workerCountReadFence() {
522	<	int ignore = workerCounts;
521	>	final boolean tryDecrementRunningCount() {
522	>	int wc = workerCounts;
523	>	if ((wc & RUNNING_COUNT_MASK) == 0)
524	>	return false;
525	>	return UNSAFE.compareAndSwapInt(this, workerCountsOffset,
526	>	wc, wc - ONE_RUNNING);
527		}
528
529		/**
#	Line 602 | Line 574 | public class ForkJoinPool extends Abstra
574		lock.lock();
575		try {
576		ForkJoinWorkerThread[] ws = workers;
577	<	int len = ws.length;
578	<	if (k < 0 || k >= len || ws[k] != null) {
579	<	for (k = 0; k < len && ws[k] != null; ++k)
577	>	int nws = ws.length;
578	>	if (k < 0 || k >= nws || ws[k] != null) {
579	>	for (k = 0; k < nws && ws[k] != null; ++k)
580		;
581	<	if (k == len)
582	<	ws = Arrays.copyOf(ws, len << 1);
581	>	if (k == nws)
582	>	ws = Arrays.copyOf(ws, nws << 1);
583		}
584		ws[k] = w;
585		workers = ws; // volatile array write ensures slot visibility
#	Line 653 | Line 625 | public class ForkJoinPool extends Abstra
625		return null;
626		}
627		}
628	<	w.start(recordWorker(w), locallyFifo, ueh);
628	>	w.start(recordWorker(w), ueh);
629		return w;
630		}
631
#	Line 661 | Line 633 | public class ForkJoinPool extends Abstra
633		* Adjusts counts upon failure to create worker
634		*/
635		private void onWorkerCreationFailure() {
636	<	int c;
637	<	do {} while (!UNSAFE.compareAndSwapInt(this, workerCountsOffset,
638	<	c = workerCounts,
639	<	c - (ONE_RUNNING|ONE_TOTAL)));
636	>	for (;;) {
637	>	int wc = workerCounts;
638	>	if ((wc >>> TOTAL_COUNT_SHIFT) > 0 &&
639	>	UNSAFE.compareAndSwapInt(this, workerCountsOffset,
640	>	wc, wc - (ONE_RUNNING|ONE_TOTAL)))
641	>	break;
642	>	}
643		tryTerminate(false); // in case of failure during shutdown
644		}
645
#	Line 674 | Line 649 | public class ForkJoinPool extends Abstra
649		*/
650		private void ensureEnoughTotalWorkers() {
651		int wc;
652	<	while (runState < TERMINATING &&
653	<	((wc = workerCounts) >>> TOTAL_COUNT_SHIFT) < parallelism) {
652	>	while (((wc = workerCounts) >>> TOTAL_COUNT_SHIFT) < parallelism &&
653	>	runState < TERMINATING) {
654		if ((UNSAFE.compareAndSwapInt(this, workerCountsOffset,
655		wc, wc + (ONE_RUNNING|ONE_TOTAL)) &&
656		addWorker() == null))
#	Line 698 | Line 673 | public class ForkJoinPool extends Abstra
673		}
674		forgetWorker(w);
675
676	<	// decrement total count, and if was running, running count
677	<	int unit = w.isTrimmed()? ONE_TOTAL : (ONE_RUNNING|ONE_TOTAL);
678	<	int wc;
679	<	do {} while (!UNSAFE.compareAndSwapInt(this, workerCountsOffset,
680	<	wc = workerCounts, wc - unit));
676	>	// Decrement total count, and if was running, running count
677	>	// Spin (waiting for other updates) if either would be negative
678	>	int nr = w.isTrimmed() ? 0 : ONE_RUNNING;
679	>	int unit = ONE_TOTAL + nr;
680	>	for (;;) {
681	>	int wc = workerCounts;
682	>	int rc = wc & RUNNING_COUNT_MASK;
683	>	if (rc - nr < 0 || (wc >>> TOTAL_COUNT_SHIFT) == 0)
684	>	Thread.yield(); // back off if waiting for other updates
685	>	else if (UNSAFE.compareAndSwapInt(this, workerCountsOffset,
686	>	wc, wc - unit))
687	>	break;
688	>	}
689
690		accumulateStealCount(w); // collect final count
691		if (!tryTerminate(false))
#	Line 712 | Line 695 | public class ForkJoinPool extends Abstra
695		// Waiting for and signalling events
696
697		/**
715	–	* Ensures eventCount on exit is different (mod 2^32) than on
716	–	* entry.  CAS failures are OK -- any change in count suffices.
717	–	*/
718	–	private void advanceEventCount() {
719	–	int c;
720	–	UNSAFE.compareAndSwapInt(this, eventCountOffset, c = eventCount, c+1);
721	–	}
722	–
723	–	/**
698		* Releases workers blocked on a count not equal to current count.
699		*/
700	<	final void releaseWaiters() {
700	>	private void releaseWaiters() {
701		long top;
702		int id;
703		while ((id = (int)((top = eventWaiters) & WAITER_INDEX_MASK)) > 0 &&
#	Line 738 | Line 712 | public class ForkJoinPool extends Abstra
712		}
713
714		/**
715	+	* Ensures eventCount on exit is different (mod 2^32) than on
716	+	* entry and wakes up all waiters
717	+	*/
718	+	private void signalEvent() {
719	+	int c;
720	+	do {} while (!UNSAFE.compareAndSwapInt(this, eventCountOffset,
721	+	c = eventCount, c+1));
722	+	releaseWaiters();
723	+	}
724	+
725	+	/**
726		* Advances eventCount and releases waiters until interference by
727		* other releasing threads is detected.
728		*/
729		final void signalWork() {
730	+	// EventCount CAS failures are OK -- any change in count suffices.
731		int ec;
732		UNSAFE.compareAndSwapInt(this, eventCountOffset, ec=eventCount, ec+1);
733		outer:for (;;) {
#	Line 820 | Line 806 | public class ForkJoinPool extends Abstra
806		boolean inactivate = !worked & active;
807		for (;;) {
808		if (inactivate) {
809	<	int c = runState;
809	>	int rs = runState;
810		if (UNSAFE.compareAndSwapInt(this, runStateOffset,
811	<	c, c - ONE_ACTIVE))
811	>	rs, rs - ONE_ACTIVE))
812		inactivate = active = w.active = false;
813		}
814		int wc = workerCounts;
#	Line 840 | Line 826 | public class ForkJoinPool extends Abstra
826		}
827
828		/**
829	<	* Adjusts counts and creates or resumes compensating threads for
830	<	* a worker about to block on task joinMe, returning early if
831	<	* joinMe becomes ready. First tries resuming an existing spare
832	<	* (which usually also avoids any count adjustment), but must then
833	<	* decrement running count to determine whether a new thread is
834	<	* needed. See above for fuller explanation.
835	<	*/
836	<	final void preJoin(ForkJoinTask<?> joinMe) {
837	<	boolean dec = false;       // true when running count decremented
838	<	for (;;) {
839	<	releaseWaiters();      // help other threads progress
840	<
841	<	if (joinMe.status < 0) // surround spare search with done checks
842	<	return;
843	<	ForkJoinWorkerThread spare = null;
844	<	for (ForkJoinWorkerThread w : workers) {
845	<	if (w != null && w.isSuspended()) {
846	<	spare = w;
829	>	* Tries to decrement running count, and if so, possibly creates
830	>	* or resumes compensating threads before blocking on task joinMe.
831	>	* This code is sprawled out with manual inlining to evade some
832	>	* JIT oddities.
833	>	*
834	>	* @param joinMe the task to join
835	>	* @return task status on exit
836	>	*/
837	>	final int tryAwaitJoin(ForkJoinTask<?> joinMe) {
838	>	int cw = workerCounts; // read now to spoil CAS if counts change as ...
839	>	releaseWaiters();      // ... a byproduct of releaseWaiters
840	>	int stat = joinMe.status;
841	>	if (stat >= 0 && // inline variant of tryDecrementRunningCount
842	>	(cw & RUNNING_COUNT_MASK) > 0 &&
843	>	UNSAFE.compareAndSwapInt(this, workerCountsOffset,
844	>	cw, cw - ONE_RUNNING)) {
845	>	int pc = parallelism;
846	>	int scans = 0;  // to require confirming passes to add threads
847	>	outer: while ((workerCounts & RUNNING_COUNT_MASK) < pc) {
848	>	if ((stat = joinMe.status) < 0)
849		break;
850	+	ForkJoinWorkerThread spare = null;
851	+	ForkJoinWorkerThread[] ws = workers;
852	+	int nws = ws.length;
853	+	for (int i = 0; i < nws; ++i) {
854	+	ForkJoinWorkerThread w = ws[i];
855	+	if (w != null && w.isSuspended()) {
856	+	spare = w;
857	+	break;
858	+	}
859	+	}
860	+	if ((stat = joinMe.status) < 0) // recheck to narrow race
861	+	break;
862	+	int wc = workerCounts;
863	+	int rc = wc & RUNNING_COUNT_MASK;
864	+	if (rc >= pc)
865	+	break;
866	+	if (spare != null) {
867	+	if (spare.tryUnsuspend()) {
868	+	int c; // inline incrementRunningCount
869	+	do {} while (!UNSAFE.compareAndSwapInt
870	+	(this, workerCountsOffset,
871	+	c = workerCounts, c + ONE_RUNNING));
872	+	LockSupport.unpark(spare);
873	+	break;
874	+	}
875	+	continue;
876		}
863	–	}
864	–	if (joinMe.status < 0)
865	–	return;
866	–
867	–	if (spare != null && spare.tryUnsuspend()) {
868	–	if (dec || joinMe.requestSignal() < 0) {
869	–	int c;
870	–	do {} while (!UNSAFE.compareAndSwapInt(this,
871	–	workerCountsOffset,
872	–	c = workerCounts,
873	–	c + ONE_RUNNING));
874	–	} // else no net count change
875	–	LockSupport.unpark(spare);
876	–	return;
877	–	}
878	–
879	–	int wc = workerCounts; // decrement running count
880	–	if (!dec && (wc & RUNNING_COUNT_MASK) != 0 &&
881	–	(dec = UNSAFE.compareAndSwapInt(this, workerCountsOffset,
882	–	wc, wc -= ONE_RUNNING)) &&
883	–	joinMe.requestSignal() < 0) { // cannot block
884	–	int c;                        // back out
885	–	do {} while (!UNSAFE.compareAndSwapInt(this,
886	–	workerCountsOffset,
887	–	c = workerCounts,
888	–	c + ONE_RUNNING));
889	–	return;
890	–	}
891	–
892	–	if (dec) {
877		int tc = wc >>> TOTAL_COUNT_SHIFT;
878	<	int pc = parallelism;
879	<	int dc = pc - (wc & RUNNING_COUNT_MASK); // deficit count
880	<	if ((dc < pc && (dc <= 0 || (dc * dc < (tc - pc) * pc) ||
881	<	!maintainsParallelism)) ||
882	<	tc >= maxPoolSize) // cannot add
883	<	return;
884	<	if (spare == null &&
878	>	int sc = tc - pc;
879	>	if (rc > 0) {
880	>	int p = pc;
881	>	int s = sc;
882	>	while (s-- >= 0) { // try keeping 3/4 live
883	>	if (rc > (p -= (p >>> 2) + 1))
884	>	break outer;
885	>	}
886	>	}
887	>	if (scans++ > sc && tc < MAX_THREADS &&
888		UNSAFE.compareAndSwapInt(this, workerCountsOffset, wc,
889		wc + (ONE_RUNNING|ONE_TOTAL))) {
890	<	addWorker();
891	<	return;
890	>	addWorker();
891	>	break;
892		}
893		}
894	+	if (stat >= 0)
895	+	stat = joinMe.internalAwaitDone();
896	+	int c; // inline incrementRunningCount
897	+	do {} while (!UNSAFE.compareAndSwapInt
898	+	(this, workerCountsOffset,
899	+	c = workerCounts, c + ONE_RUNNING));
900		}
901	+	return stat;
902		}
903
904		/**
905	<	* Same idea as preJoin but with too many differing details to
906	<	* integrate: There are no task-based signal counts, and only one
913	<	* way to do the actual blocking. So for simplicity it is directly
914	<	* incorporated into this method.
905	>	* Same idea as (and mostly pasted from) tryAwaitJoin, but
906	>	* self-contained
907		*/
908	<	final void doBlock(ManagedBlocker blocker, boolean maintainPar)
908	>	final void awaitBlocker(ManagedBlocker blocker)
909		throws InterruptedException {
918	–	maintainPar &= maintainsParallelism; // override
919	–	boolean dec = false;
920	–	boolean done = false;
910		for (;;) {
911	+	if (blocker.isReleasable())
912	+	return;
913	+	int cw = workerCounts;
914		releaseWaiters();
915	+	if ((cw & RUNNING_COUNT_MASK) > 0 &&
916	+	UNSAFE.compareAndSwapInt(this, workerCountsOffset,
917	+	cw, cw - ONE_RUNNING))
918	+	break;
919	+	}
920	+	boolean done = false;
921	+	int pc = parallelism;
922	+	int scans = 0;
923	+	outer: while ((workerCounts & RUNNING_COUNT_MASK) < pc) {
924		if (done = blocker.isReleasable())
925		break;
926		ForkJoinWorkerThread spare = null;
927	<	for (ForkJoinWorkerThread w : workers) {
927	>	ForkJoinWorkerThread[] ws = workers;
928	>	int nws = ws.length;
929	>	for (int i = 0; i < nws; ++i) {
930	>	ForkJoinWorkerThread w = ws[i];
931		if (w != null && w.isSuspended()) {
932		spare = w;
933		break;
#	Line 931 | Line 935 | public class ForkJoinPool extends Abstra
935		}
936		if (done = blocker.isReleasable())
937		break;
938	<	if (spare != null && spare.tryUnsuspend()) {
939	<	if (dec) {
938	>	int wc = workerCounts;
939	>	int rc = wc & RUNNING_COUNT_MASK;
940	>	if (rc >= pc)
941	>	break;
942	>	if (spare != null) {
943	>	if (spare.tryUnsuspend()) {
944		int c;
945	<	do {} while (!UNSAFE.compareAndSwapInt(this,
946	<	workerCountsOffset,
947	<	c = workerCounts,
948	<	c + ONE_RUNNING));
945	>	do {} while (!UNSAFE.compareAndSwapInt
946	>	(this, workerCountsOffset,
947	>	c = workerCounts, c + ONE_RUNNING));
948	>	LockSupport.unpark(spare);
949	>	break;
950		}
951	<	LockSupport.unpark(spare);
943	<	break;
951	>	continue;
952		}
953	<	int wc = workerCounts;
954	<	if (!dec && (wc & RUNNING_COUNT_MASK) != 0)
955	<	dec = UNSAFE.compareAndSwapInt(this, workerCountsOffset,
956	<	wc, wc -= ONE_RUNNING);
957	<	if (dec) {
958	<	int tc = wc >>> TOTAL_COUNT_SHIFT;
959	<	int pc = parallelism;
960	<	int dc = pc - (wc & RUNNING_COUNT_MASK);
953	<	if ((dc < pc && (dc <= 0 || (dc * dc < (tc - pc) * pc) ||
954	<	!maintainPar)) ||
955	<	tc >= maxPoolSize)
956	<	break;
957	<	if (spare == null &&
958	<	UNSAFE.compareAndSwapInt(this, workerCountsOffset, wc,
959	<	wc + (ONE_RUNNING|ONE_TOTAL))){
960	<	addWorker();
961	<	break;
953	>	int tc = wc >>> TOTAL_COUNT_SHIFT;
954	>	int sc = tc - pc;
955	>	if (rc > 0) {
956	>	int p = pc;
957	>	int s = sc;
958	>	while (s-- >= 0) {
959	>	if (rc > (p -= (p >>> 2) + 1))
960	>	break outer;
961		}
962		}
963	+	if (scans++ > sc && tc < MAX_THREADS &&
964	+	UNSAFE.compareAndSwapInt(this, workerCountsOffset, wc,
965	+	wc + (ONE_RUNNING|ONE_TOTAL))) {
966	+	addWorker();
967	+	break;
968	+	}
969		}
965	–
970		try {
971		if (!done)
972	<	do {} while (!blocker.isReleasable() && !blocker.block());
972	>	do {} while (!blocker.isReleasable() &&
973	>	!blocker.block());
974		} finally {
975	<	if (dec) {
976	<	int c;
977	<	do {} while (!UNSAFE.compareAndSwapInt(this,
978	<	workerCountsOffset,
974	<	c = workerCounts,
975	<	c + ONE_RUNNING));
976	<	}
977	<	}
978	<	}
979	<
980	<	/**
981	<	* Unless there are not enough other running threads, adjusts
982	<	* counts for a a worker in performing helpJoin that cannot find
983	<	* any work, so that this worker can now block.
984	<	*
985	<	* @return true if worker may block
986	<	*/
987	<	final boolean preBlockHelpingJoin(ForkJoinTask<?> joinMe) {
988	<	while (joinMe.status >= 0) {
989	<	releaseWaiters(); // help other threads progress
990	<
991	<	// if a spare exists, resume it to maintain parallelism level
992	<	if ((workerCounts & RUNNING_COUNT_MASK) <= parallelism) {
993	<	ForkJoinWorkerThread spare = null;
994	<	for (ForkJoinWorkerThread w : workers) {
995	<	if (w != null && w.isSuspended()) {
996	<	spare = w;
997	<	break;
998	<	}
999	<	}
1000	<	if (joinMe.status < 0)
1001	<	break;
1002	<	if (spare != null) {
1003	<	if (spare.tryUnsuspend()) {
1004	<	boolean canBlock = true;
1005	<	if (joinMe.requestSignal() < 0) {
1006	<	canBlock = false; // already done
1007	<	int c;
1008	<	do {} while (!UNSAFE.compareAndSwapInt
1009	<	(this, workerCountsOffset,
1010	<	c = workerCounts, c + ONE_RUNNING));
1011	<	}
1012	<	LockSupport.unpark(spare);
1013	<	return canBlock;
1014	<	}
1015	<	continue; // recheck -- another spare may exist
1016	<	}
1017	<	}
1018	<
1019	<	int wc = workerCounts; // reread to shorten CAS window
1020	<	int rc = wc & RUNNING_COUNT_MASK;
1021	<	if (rc <= 2) // keep this and at most one other thread alive
1022	<	break;
1023	<
1024	<	if (UNSAFE.compareAndSwapInt(this, workerCountsOffset,
1025	<	wc, wc - ONE_RUNNING)) {
1026	<	if (joinMe.requestSignal() >= 0)
1027	<	return true;
1028	<	int c;                        // back out
1029	<	do {} while (!UNSAFE.compareAndSwapInt
1030	<	(this, workerCountsOffset,
1031	<	c = workerCounts, c + ONE_RUNNING));
1032	<	break;
1033	<	}
975	>	int c;
976	>	do {} while (!UNSAFE.compareAndSwapInt
977	>	(this, workerCountsOffset,
978	>	c = workerCounts, c + ONE_RUNNING));
979		}
980	<	return false;
1036	<	}
980	>	}
981
982		/**
983		* Possibly initiates and/or completes termination.
#	Line 1056 | Line 1000 | public class ForkJoinPool extends Abstra
1000		// Finish now if all threads terminated; else in some subsequent call
1001		if ((workerCounts >>> TOTAL_COUNT_SHIFT) == 0) {
1002		advanceRunLevel(TERMINATED);
1003	<	terminationLatch.countDown();
1003	>	termination.arrive();
1004		}
1005		return true;
1006		}
#	Line 1065 | Line 1009 | public class ForkJoinPool extends Abstra
1009		* Actions on transition to TERMINATING
1010		*/
1011		private void startTerminating() {
1012	<	// Clear out and cancel submissions, ignoring exceptions
1012	>	for (int i = 0; i < 2; ++i) { // twice to mop up newly created workers
1013	>	cancelSubmissions();
1014	>	shutdownWorkers();
1015	>	cancelWorkerTasks();
1016	>	signalEvent();
1017	>	interruptWorkers();
1018	>	}
1019	>	}
1020	>
1021	>	/**
1022	>	* Clear out and cancel submissions, ignoring exceptions
1023	>	*/
1024	>	private void cancelSubmissions() {
1025		ForkJoinTask<?> task;
1026		while ((task = submissionQueue.poll()) != null) {
1027		try {
#	Line 1073 | Line 1029 | public class ForkJoinPool extends Abstra
1029		} catch (Throwable ignore) {
1030		}
1031		}
1032	<	// Propagate run level
1033	<	for (ForkJoinWorkerThread w : workers) {
1032	>	}
1033	>
1034	>	/**
1035	>	* Sets all worker run states to at least shutdown,
1036	>	* also resuming suspended workers
1037	>	*/
1038	>	private void shutdownWorkers() {
1039	>	ForkJoinWorkerThread[] ws = workers;
1040	>	int nws = ws.length;
1041	>	for (int i = 0; i < nws; ++i) {
1042	>	ForkJoinWorkerThread w = ws[i];
1043		if (w != null)
1044	<	w.shutdown();    // also resumes suspended workers
1044	>	w.shutdown();
1045		}
1046	<	// Ensure no straggling local tasks
1047	<	for (ForkJoinWorkerThread w : workers) {
1046	>	}
1047	>
1048	>	/**
1049	>	* Clears out and cancels all locally queued tasks
1050	>	*/
1051	>	private void cancelWorkerTasks() {
1052	>	ForkJoinWorkerThread[] ws = workers;
1053	>	int nws = ws.length;
1054	>	for (int i = 0; i < nws; ++i) {
1055	>	ForkJoinWorkerThread w = ws[i];
1056		if (w != null)
1057		w.cancelTasks();
1058		}
1059	<	// Wake up idle workers
1060	<	advanceEventCount();
1061	<	releaseWaiters();
1062	<	// Unstick pending joins
1063	<	for (ForkJoinWorkerThread w : workers) {
1059	>	}
1060	>
1061	>	/**
1062	>	* Unsticks all workers blocked on joins etc
1063	>	*/
1064	>	private void interruptWorkers() {
1065	>	ForkJoinWorkerThread[] ws = workers;
1066	>	int nws = ws.length;
1067	>	for (int i = 0; i < nws; ++i) {
1068	>	ForkJoinWorkerThread w = ws[i];
1069		if (w != null && !w.isTerminated()) {
1070		try {
1071		w.interrupt();
#	Line 1125 | Line 1103 | public class ForkJoinPool extends Abstra
1103		* active thread.
1104		*/
1105		final int idlePerActive() {
1128	–	int ac = runState;    // no mask -- artifically boosts during shutdown
1106		int pc = parallelism; // use targeted parallelism, not rc
1107	+	int ac = runState;    // no mask -- artifically boosts during shutdown
1108		// Use exact results for small values, saturate past 4
1109		return pc <= ac? 0 : pc >>> 1 <= ac? 1 : pc >>> 2 <= ac? 3 : pc >>> 3;
1110		}
#	Line 1137 | Line 1115 | public class ForkJoinPool extends Abstra
1115
1116		/**
1117		* Creates a {@code ForkJoinPool} with parallelism equal to {@link
1118	<	* java.lang.Runtime#availableProcessors}, and using the {@linkplain
1119	<	* #defaultForkJoinWorkerThreadFactory default thread factory}.
1118	>	* java.lang.Runtime#availableProcessors}, using the {@linkplain
1119	>	* #defaultForkJoinWorkerThreadFactory default thread factory},
1120	>	* no UncaughtExceptionHandler, and non-async LIFO processing mode.
1121		*
1122		* @throws SecurityException if a security manager exists and
1123		*         the caller is not permitted to modify threads
#	Line 1147 | Line 1126 | public class ForkJoinPool extends Abstra
1126		*/
1127		public ForkJoinPool() {
1128		this(Runtime.getRuntime().availableProcessors(),
1129	<	defaultForkJoinWorkerThreadFactory);
1129	>	defaultForkJoinWorkerThreadFactory, null, false);
1130		}
1131
1132		/**
1133		* Creates a {@code ForkJoinPool} with the indicated parallelism
1134	<	* level and using the {@linkplain
1135	<	* #defaultForkJoinWorkerThreadFactory default thread factory}.
1134	>	* level, the {@linkplain
1135	>	* #defaultForkJoinWorkerThreadFactory default thread factory},
1136	>	* no UncaughtExceptionHandler, and non-async LIFO processing mode.
1137		*
1138		* @param parallelism the parallelism level
1139		* @throws IllegalArgumentException if parallelism less than or
#	Line 1164 | Line 1144 | public class ForkJoinPool extends Abstra
1144		*         java.lang.RuntimePermission}{@code ("modifyThread")}
1145		*/
1146		public ForkJoinPool(int parallelism) {
1147	<	this(parallelism, defaultForkJoinWorkerThreadFactory);
1168	<	}
1169	<
1170	<	/**
1171	<	* Creates a {@code ForkJoinPool} with parallelism equal to {@link
1172	<	* java.lang.Runtime#availableProcessors}, and using the given
1173	<	* thread factory.
1174	<	*
1175	<	* @param factory the factory for creating new threads
1176	<	* @throws NullPointerException if the factory is null
1177	<	* @throws SecurityException if a security manager exists and
1178	<	*         the caller is not permitted to modify threads
1179	<	*         because it does not hold {@link
1180	<	*         java.lang.RuntimePermission}{@code ("modifyThread")}
1181	<	*/
1182	<	public ForkJoinPool(ForkJoinWorkerThreadFactory factory) {
1183	<	this(Runtime.getRuntime().availableProcessors(), factory);
1147	>	this(parallelism, defaultForkJoinWorkerThreadFactory, null, false);
1148		}
1149
1150		/**
1151	<	* Creates a {@code ForkJoinPool} with the given parallelism and
1188	<	* thread factory.
1151	>	* Creates a {@code ForkJoinPool} with the given parameters.
1152		*
1153	<	* @param parallelism the parallelism level
1154	<	* @param factory the factory for creating new threads
1153	>	* @param parallelism the parallelism level. For default value,
1154	>	* use {@link java.lang.Runtime#availableProcessors}.
1155	>	* @param factory the factory for creating new threads. For default value,
1156	>	* use {@link #defaultForkJoinWorkerThreadFactory}.
1157	>	* @param handler the handler for internal worker threads that
1158	>	* terminate due to unrecoverable errors encountered while executing
1159	>	* tasks. For default value, use <code>null</code>.
1160	>	* @param asyncMode if true,
1161	>	* establishes local first-in-first-out scheduling mode for forked
1162	>	* tasks that are never joined. This mode may be more appropriate
1163	>	* than default locally stack-based mode in applications in which
1164	>	* worker threads only process event-style asynchronous tasks.
1165	>	* For default value, use <code>false</code>.
1166		* @throws IllegalArgumentException if parallelism less than or
1167		*         equal to zero, or greater than implementation limit
1168		* @throws NullPointerException if the factory is null
#	Line 1197 | Line 1171 | public class ForkJoinPool extends Abstra
1171		*         because it does not hold {@link
1172		*         java.lang.RuntimePermission}{@code ("modifyThread")}
1173		*/
1174	<	public ForkJoinPool(int parallelism, ForkJoinWorkerThreadFactory factory) {
1174	>	public ForkJoinPool(int parallelism,
1175	>	ForkJoinWorkerThreadFactory factory,
1176	>	Thread.UncaughtExceptionHandler handler,
1177	>	boolean asyncMode) {
1178		checkPermission();
1179		if (factory == null)
1180		throw new NullPointerException();
1181		if (parallelism <= 0 || parallelism > MAX_THREADS)
1182		throw new IllegalArgumentException();
1206	–	this.poolNumber = poolNumberGenerator.incrementAndGet();
1207	–	int arraySize = initialArraySizeFor(parallelism);
1183		this.parallelism = parallelism;
1184		this.factory = factory;
1185	<	this.maxPoolSize = MAX_THREADS;
1186	<	this.maintainsParallelism = true;
1185	>	this.ueh = handler;
1186	>	this.locallyFifo = asyncMode;
1187	>	int arraySize = initialArraySizeFor(parallelism);
1188		this.workers = new ForkJoinWorkerThread[arraySize];
1189		this.submissionQueue = new LinkedTransferQueue<ForkJoinTask<?>>();
1190		this.workerLock = new ReentrantLock();
1191	<	this.terminationLatch = new CountDownLatch(1);
1192	<	// Start first worker; remaining workers added upon first submission
1217	<	workerCounts = ONE_RUNNING | ONE_TOTAL;
1218	<	addWorker();
1191	>	this.termination = new Phaser(1);
1192	>	this.poolNumber = poolNumberGenerator.incrementAndGet();
1193		}
1194
1195		/**
#	Line 1242 | Line 1216 | public class ForkJoinPool extends Abstra
1216		throw new NullPointerException();
1217		if (runState >= SHUTDOWN)
1218		throw new RejectedExecutionException();
1219	<	submissionQueue.offer(task);
1220	<	advanceEventCount();
1221	<	releaseWaiters();
1222	<	if ((workerCounts >>> TOTAL_COUNT_SHIFT) < parallelism)
1219	>	// Convert submissions to current pool into forks
1220	>	Thread t = Thread.currentThread();
1221	>	ForkJoinWorkerThread w;
1222	>	if ((t instanceof ForkJoinWorkerThread) &&
1223	>	(w = (ForkJoinWorkerThread) t).pool == this)
1224	>	w.pushTask(task);
1225	>	else {
1226	>	submissionQueue.offer(task);
1227	>	signalEvent();
1228		ensureEnoughTotalWorkers();
1229	+	}
1230		}
1231
1232		/**
1233		* Performs the given task, returning its result upon completion.
1234	+	* If the caller is already engaged in a fork/join computation in
1235	+	* the current pool, this method is equivalent in effect to
1236	+	* {@link ForkJoinTask#invoke}.
1237		*
1238		* @param task the task
1239		* @return the task's result
#	Line 1265 | Line 1248 | public class ForkJoinPool extends Abstra
1248
1249		/**
1250		* Arranges for (asynchronous) execution of the given task.
1251	+	* If the caller is already engaged in a fork/join computation in
1252	+	* the current pool, this method is equivalent in effect to
1253	+	* {@link ForkJoinTask#fork}.
1254		*
1255		* @param task the task
1256		* @throws NullPointerException if the task is null
#	Line 1292 | Line 1278 | public class ForkJoinPool extends Abstra
1278		}
1279
1280		/**
1281	+	* Submits a ForkJoinTask for execution.
1282	+	* If the caller is already engaged in a fork/join computation in
1283	+	* the current pool, this method is equivalent in effect to
1284	+	* {@link ForkJoinTask#fork}.
1285	+	*
1286	+	* @param task the task to submit
1287	+	* @return the task
1288	+	* @throws NullPointerException if the task is null
1289	+	* @throws RejectedExecutionException if the task cannot be
1290	+	*         scheduled for execution
1291	+	*/
1292	+	public <T> ForkJoinTask<T> submit(ForkJoinTask<T> task) {
1293	+	doSubmit(task);
1294	+	return task;
1295	+	}
1296	+
1297	+	/**
1298		* @throws NullPointerException if the task is null
1299		* @throws RejectedExecutionException if the task cannot be
1300		*         scheduled for execution
#	Line 1329 | Line 1332 | public class ForkJoinPool extends Abstra
1332		}
1333
1334		/**
1332	–	* Submits a ForkJoinTask for execution.
1333	–	*
1334	–	* @param task the task to submit
1335	–	* @return the task
1336	–	* @throws NullPointerException if the task is null
1337	–	* @throws RejectedExecutionException if the task cannot be
1338	–	*         scheduled for execution
1339	–	*/
1340	–	public <T> ForkJoinTask<T> submit(ForkJoinTask<T> task) {
1341	–	doSubmit(task);
1342	–	return task;
1343	–	}
1344	–
1345	–	/**
1335		* @throws NullPointerException       {@inheritDoc}
1336		* @throws RejectedExecutionException {@inheritDoc}
1337		*/
#	Line 1384 | Line 1373 | public class ForkJoinPool extends Abstra
1373		* @return the handler, or {@code null} if none
1374		*/
1375		public Thread.UncaughtExceptionHandler getUncaughtExceptionHandler() {
1387	–	workerCountReadFence();
1376		return ueh;
1377		}
1378
1379		/**
1392	–	* Sets the handler for internal worker threads that terminate due
1393	–	* to unrecoverable errors encountered while executing tasks.
1394	–	* Unless set, the current default or ThreadGroup handler is used
1395	–	* as handler.
1396	–	*
1397	–	* @param h the new handler
1398	–	* @return the old handler, or {@code null} if none
1399	–	* @throws SecurityException if a security manager exists and
1400	–	*         the caller is not permitted to modify threads
1401	–	*         because it does not hold {@link
1402	–	*         java.lang.RuntimePermission}{@code ("modifyThread")}
1403	–	*/
1404	–	public Thread.UncaughtExceptionHandler
1405	–	setUncaughtExceptionHandler(Thread.UncaughtExceptionHandler h) {
1406	–	checkPermission();
1407	–	workerCountReadFence();
1408	–	Thread.UncaughtExceptionHandler old = ueh;
1409	–	if (h != old) {
1410	–	ueh = h;
1411	–	workerCountWriteFence();
1412	–	for (ForkJoinWorkerThread w : workers) {
1413	–	if (w != null)
1414	–	w.setUncaughtExceptionHandler(h);
1415	–	}
1416	–	}
1417	–	return old;
1418	–	}
1419	–
1420	–	/**
1421	–	* Sets the target parallelism level of this pool.
1422	–	*
1423	–	* @param parallelism the target parallelism
1424	–	* @throws IllegalArgumentException if parallelism less than or
1425	–	* equal to zero or greater than maximum size bounds
1426	–	* @throws SecurityException if a security manager exists and
1427	–	*         the caller is not permitted to modify threads
1428	–	*         because it does not hold {@link
1429	–	*         java.lang.RuntimePermission}{@code ("modifyThread")}
1430	–	*/
1431	–	public void setParallelism(int parallelism) {
1432	–	checkPermission();
1433	–	if (parallelism <= 0 || parallelism > maxPoolSize)
1434	–	throw new IllegalArgumentException();
1435	–	workerCountReadFence();
1436	–	int pc = this.parallelism;
1437	–	if (pc != parallelism) {
1438	–	this.parallelism = parallelism;
1439	–	workerCountWriteFence();
1440	–	// Release spares. If too many, some will die after re-suspend
1441	–	for (ForkJoinWorkerThread w : workers) {
1442	–	if (w != null && w.tryUnsuspend()) {
1443	–	updateRunningCount(1);
1444	–	LockSupport.unpark(w);
1445	–	}
1446	–	}
1447	–	ensureEnoughTotalWorkers();
1448	–	advanceEventCount();
1449	–	releaseWaiters(); // force config recheck by existing workers
1450	–	}
1451	–	}
1452	–
1453	–	/**
1380		* Returns the targeted parallelism level of this pool.
1381		*
1382		* @return the targeted parallelism level of this pool
1383		*/
1384		public int getParallelism() {
1459	–	//        workerCountReadFence(); // inlined below
1460	–	int ignore = workerCounts;
1385		return parallelism;
1386		}
1387
#	Line 1474 | Line 1398 | public class ForkJoinPool extends Abstra
1398		}
1399
1400		/**
1477	–	* Returns the maximum number of threads allowed to exist in the
1478	–	* pool. Unless set using {@link #setMaximumPoolSize}, the
1479	–	* maximum is an implementation-defined value designed only to
1480	–	* prevent runaway growth.
1481	–	*
1482	–	* @return the maximum
1483	–	*/
1484	–	public int getMaximumPoolSize() {
1485	–	workerCountReadFence();
1486	–	return maxPoolSize;
1487	–	}
1488	–
1489	–	/**
1490	–	* Sets the maximum number of threads allowed to exist in the
1491	–	* pool. The given value should normally be greater than or equal
1492	–	* to the {@link #getParallelism parallelism} level. Setting this
1493	–	* value has no effect on current pool size. It controls
1494	–	* construction of new threads. The use of this method may cause
1495	–	* tasks that intrinsically require extra threads for dependent
1496	–	* computations to indefinitely stall. If you are instead trying
1497	–	* to minimize internal thread creation, consider setting {@link
1498	–	* #setMaintainsParallelism} as false.
1499	–	*
1500	–	* @throws IllegalArgumentException if negative or greater than
1501	–	* internal implementation limit
1502	–	*/
1503	–	public void setMaximumPoolSize(int newMax) {
1504	–	if (newMax < 0 || newMax > MAX_THREADS)
1505	–	throw new IllegalArgumentException();
1506	–	maxPoolSize = newMax;
1507	–	workerCountWriteFence();
1508	–	}
1509	–
1510	–	/**
1511	–	* Returns {@code true} if this pool dynamically maintains its
1512	–	* target parallelism level. If false, new threads are added only
1513	–	* to avoid possible starvation.  This setting is by default true.
1514	–	*
1515	–	* @return {@code true} if maintains parallelism
1516	–	*/
1517	–	public boolean getMaintainsParallelism() {
1518	–	workerCountReadFence();
1519	–	return maintainsParallelism;
1520	–	}
1521	–
1522	–	/**
1523	–	* Sets whether this pool dynamically maintains its target
1524	–	* parallelism level. If false, new threads are added only to
1525	–	* avoid possible starvation.
1526	–	*
1527	–	* @param enable {@code true} to maintain parallelism
1528	–	*/
1529	–	public void setMaintainsParallelism(boolean enable) {
1530	–	maintainsParallelism = enable;
1531	–	workerCountWriteFence();
1532	–	}
1533	–
1534	–	/**
1535	–	* Establishes local first-in-first-out scheduling mode for forked
1536	–	* tasks that are never joined. This mode may be more appropriate
1537	–	* than default locally stack-based mode in applications in which
1538	–	* worker threads only process asynchronous tasks.  This method is
1539	–	* designed to be invoked only when the pool is quiescent, and
1540	–	* typically only before any tasks are submitted. The effects of
1541	–	* invocations at other times may be unpredictable.
1542	–	*
1543	–	* @param async if {@code true}, use locally FIFO scheduling
1544	–	* @return the previous mode
1545	–	* @see #getAsyncMode
1546	–	*/
1547	–	public boolean setAsyncMode(boolean async) {
1548	–	workerCountReadFence();
1549	–	boolean oldMode = locallyFifo;
1550	–	if (oldMode != async) {
1551	–	locallyFifo = async;
1552	–	workerCountWriteFence();
1553	–	for (ForkJoinWorkerThread w : workers) {
1554	–	if (w != null)
1555	–	w.setAsyncMode(async);
1556	–	}
1557	–	}
1558	–	return oldMode;
1559	–	}
1560	–
1561	–	/**
1401		* Returns {@code true} if this pool uses local first-in-first-out
1402		* scheduling mode for forked tasks that are never joined.
1403		*
1404		* @return {@code true} if this pool uses async mode
1566	–	* @see #setAsyncMode
1405		*/
1406		public boolean getAsyncMode() {
1569	–	workerCountReadFence();
1407		return locallyFifo;
1408		}
1409
#	Line 1635 | Line 1472 | public class ForkJoinPool extends Abstra
1472		*/
1473		public long getQueuedTaskCount() {
1474		long count = 0;
1475	<	for (ForkJoinWorkerThread w : workers) {
1475	>	ForkJoinWorkerThread[] ws = workers;
1476	>	int nws = ws.length;
1477	>	for (int i = 0; i < nws; ++i) {
1478	>	ForkJoinWorkerThread w = ws[i];
1479		if (w != null)
1480		count += w.getQueueSize();
1481		}
#	Line 1693 | Line 1533 | public class ForkJoinPool extends Abstra
1533		*/
1534		protected int drainTasksTo(Collection<? super ForkJoinTask<?>> c) {
1535		int n = submissionQueue.drainTo(c);
1536	<	for (ForkJoinWorkerThread w : workers) {
1536	>	ForkJoinWorkerThread[] ws = workers;
1537	>	int nws = ws.length;
1538	>	for (int i = 0; i < nws; ++i) {
1539	>	ForkJoinWorkerThread w = ws[i];
1540		if (w != null)
1541		n += w.drainTasksTo(c);
1542		}
#	Line 1701 | Line 1544 | public class ForkJoinPool extends Abstra
1544		}
1545
1546		/**
1547	+	* Returns count of total parks by existing workers.
1548	+	* Used during development only since not meaningful to users.
1549	+	*/
1550	+	private int collectParkCount() {
1551	+	int count = 0;
1552	+	ForkJoinWorkerThread[] ws = workers;
1553	+	int nws = ws.length;
1554	+	for (int i = 0; i < nws; ++i) {
1555	+	ForkJoinWorkerThread w = ws[i];
1556	+	if (w != null)
1557	+	count += w.parkCount;
1558	+	}
1559	+	return count;
1560	+	}
1561	+
1562	+	/**
1563		* Returns a string identifying this pool, as well as its state,
1564		* including indications of run state, parallelism level, and
1565		* worker and task counts.
#	Line 1717 | Line 1576 | public class ForkJoinPool extends Abstra
1576		int pc = parallelism;
1577		int rs = runState;
1578		int ac = rs & ACTIVE_COUNT_MASK;
1579	+	//        int pk = collectParkCount();
1580		return super.toString() +
1581		"[" + runLevelToString(rs) +
1582		", parallelism = " + pc +
#	Line 1726 | Line 1586 | public class ForkJoinPool extends Abstra
1586		", steals = " + st +
1587		", tasks = " + qt +
1588		", submissions = " + qs +
1589	+	//            ", parks = " + pk +
1590		"]";
1591		}
1592
#	Line 1821 | Line 1682 | public class ForkJoinPool extends Abstra
1682		*/
1683		public boolean awaitTermination(long timeout, TimeUnit unit)
1684		throws InterruptedException {
1685	<	return terminationLatch.await(timeout, unit);
1685	>	try {
1686	>	return termination.awaitAdvanceInterruptibly(0, timeout, unit) > 0;
1687	>	} catch(TimeoutException ex) {
1688	>	return false;
1689	>	}
1690		}
1691
1692		/**
#	Line 1873 | Line 1738 | public class ForkJoinPool extends Abstra
1738		* Blocks in accord with the given blocker.  If the current thread
1739		* is a {@link ForkJoinWorkerThread}, this method possibly
1740		* arranges for a spare thread to be activated if necessary to
1741	<	* ensure parallelism while the current thread is blocked.
1877	<	*
1878	<	* <p>If {@code maintainParallelism} is {@code true} and the pool
1879	<	* supports it ({@link #getMaintainsParallelism}), this method
1880	<	* attempts to maintain the pool's nominal parallelism. Otherwise
1881	<	* it activates a thread only if necessary to avoid complete
1882	<	* starvation. This option may be preferable when blockages use
1883	<	* timeouts, or are almost always brief.
1741	>	* ensure sufficient parallelism while the current thread is blocked.
1742		*
1743		* <p>If the caller is not a {@link ForkJoinTask}, this method is
1744		* behaviorally equivalent to
#	Line 1894 | Line 1752 | public class ForkJoinPool extends Abstra
1752		* first be expanded to ensure parallelism, and later adjusted.
1753		*
1754		* @param blocker the blocker
1897	–	* @param maintainParallelism if {@code true} and supported by
1898	–	* this pool, attempt to maintain the pool's nominal parallelism;
1899	–	* otherwise activate a thread only if necessary to avoid
1900	–	* complete starvation.
1755		* @throws InterruptedException if blocker.block did so
1756		*/
1757	<	public static void managedBlock(ManagedBlocker blocker,
1904	<	boolean maintainParallelism)
1757	>	public static void managedBlock(ManagedBlocker blocker)
1758		throws InterruptedException {
1759		Thread t = Thread.currentThread();
1760		if (t instanceof ForkJoinWorkerThread)
1761	<	((ForkJoinWorkerThread) t).pool.
1762	<	doBlock(blocker, maintainParallelism);
1763	<	else
1764	<	awaitBlocker(blocker);
1912	<	}
1913	<
1914	<	/**
1915	<	* Performs Non-FJ blocking
1916	<	*/
1917	<	private static void awaitBlocker(ManagedBlocker blocker)
1918	<	throws InterruptedException {
1919	<	do {} while (!blocker.isReleasable() && !blocker.block());
1761	>	((ForkJoinWorkerThread) t).pool.awaitBlocker(blocker);
1762	>	else {
1763	>	do {} while (!blocker.isReleasable() && !blocker.block());
1764	>	}
1765		}
1766
1767		// AbstractExecutorService overrides.  These rely on undocumented
#	Line 1945 | Line 1790 | public class ForkJoinPool extends Abstra
1790		private static final long stealCountOffset =
1791		objectFieldOffset("stealCount",ForkJoinPool.class);
1792
1948	–
1793		private static long objectFieldOffset(String field, Class<?> klazz) {
1794		try {
1795		return UNSAFE.objectFieldOffset(klazz.getDeclaredField(field));

Diff Legend

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