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.56 by dl, Thu May 27 16:46:48 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 147 | Line 171 | public class ForkJoinPool extends Abstra
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 167 | Line 190 | public class ForkJoinPool extends Abstra
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 bundling updates (for
194	<	* example blocking one thread on join and resuming a spare cancel
172	<	* each other out), and in most other cases performing an
173	<	* alternative action like releasing waiters or locating spares.
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 258 | 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 awaitJoin and awaitBlocker. We always
283	<	* need to create one when the number of running threads becomes
284	<	* zero. But because blocked joins are typically dependent, we
285	<	* don't necessarily need or want one-to-one replacement. Using a
286	<	* one-to-one compensation rule often leads to enough useless
287	<	* overhead creating, suspending, resuming, and/or killing threads
288	<	* to signficantly degrade throughput.  We use a rule reflecting
289	<	* the idea that, the more spare threads you already have, the
290	<	* more evidence you need to create another one. The "evidence"
291	<	* here takes two forms: (1) Using a creation threshold expressed
292	<	* in terms of the current deficit -- target minus running
293	<	* threads. To reduce flickering and drift around target values,
294	<	* the relation is quadratic: adding a spare if (dc*dc)>=(sc*pc)
295	<	* (where dc is deficit, sc is number of spare threads and pc is
296	<	* target parallelism.)  (2) Using a form of adaptive
297	<	* spionning. requiring a number of threshold checks proportional
298	<	* to the number of spare threads.  This effectively reduces churn
278	<	* at the price of systematically undershooting target parallelism
279	<	* when many threads are blocked.  However, biasing toward
280	<	* undeshooting partially compensates for the above mechanics to
281	<	* suspend extra threads, that normally lead to overshoot because
282	<	* we can only suspend workers in-between top-level actions. It
283	<	* also better copes with the fact that some of the methods in
284	<	* this class tend to never become compiled (but are interpreted),
285	<	* so some components of the entire set of controls might execute
286	<	* many times faster than others. And similarly for cases where
287	<	* the apparent lack of work is just due to GC stalls and other
288	<	* transient system activity.
289	<	*
290	<	* 7. Maintaining other configuration parameters and monitoring
291	<	* statistics. Updates to fields controlling parallelism level,
292	<	* max size, etc can only meaningfully take effect for individual
293	<	* threads upon their next top-level actions; i.e., between
294	<	* stealing/running tasks/submission, which are separated by calls
295	<	* to preStep.  Memory ordering for these (assumed infrequent)
296	<	* reconfiguration calls is ensured by using reads and writes to
297	<	* volatile field workerCounts (that must be read in preStep anyway)
298	<	* as "fences" -- user-level reads are preceded by reads of
299	<	* workCounts, and writes are followed by no-op CAS to
300	<	* workerCounts. The values reported by other management and
301	<	* monitoring methods are either computed on demand, or are kept
302	<	* in fields that are only updated when threads are otherwise
303	<	* 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 346 | 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 414 | 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 485 | 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
488	–	/*
489	–	* Fields parallelism. maxPoolSize, and maintainsParallelism are
490	–	* non-volatile, but external reads/writes use workerCount fences
491	–	* to ensure visability.
492	–	*/
493	–
483		/**
484		* The target parallelism level.
485	+	* Accessed directly by ForkJoinWorkerThreads.
486		*/
487	<	private int parallelism;
498	<
499	<	/**
500	<	* The maximum allowed pool size.
501	<	*/
502	<	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 volatile 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;
514	<
515	<	/**
516	<	* The uncaught exception handler used when any worker
517	<	* abruptly terminates
518	<	*/
519	<	private volatile Thread.UncaughtExceptionHandler ueh;
499	>	private final Thread.UncaughtExceptionHandler ueh;
500
501		/**
502		* Pool number, just for assigning useful names to worker threads
#	Line 526 | Line 506 | public class ForkJoinPool extends Abstra
506		// utilities for updating fields
507
508		/**
509	<	* Adds delta to running count.  Used mainly by ForkJoinTask.
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));
514	>	c = workerCounts,
515	>	c + ONE_RUNNING));
516		}
517	<
517	>
518		/**
519	<	* Decrements running count unless already zero
519	>	* Tries to decrement running count unless already zero
520		*/
521		final boolean tryDecrementRunningCount() {
522		int wc = workerCounts;
#	Line 547 | Line 527 | public class ForkJoinPool extends Abstra
527		}
528
529		/**
550	–	* Write fence for user modifications of pool parameters
551	–	* (parallelism. etc).  Note that it doesn't matter if CAS fails.
552	–	*/
553	–	private void workerCountWriteFence() {
554	–	int wc;
555	–	UNSAFE.compareAndSwapInt(this, workerCountsOffset,
556	–	wc = workerCounts, wc);
557	–	}
558	–
559	–	/**
560	–	* Read fence for external reads of pool parameters
561	–	* (parallelism. maxPoolSize, etc).
562	–	*/
563	–	private void workerCountReadFence() {
564	–	int ignore = workerCounts;
565	–	}
566	–
567	–	/**
530		* Tries incrementing active count; fails on contention.
531		* Called by workers before executing tasks.
532		*
#	Line 663 | 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 733 | Line 695 | public class ForkJoinPool extends Abstra
695		// Waiting for and signalling events
696
697		/**
736	–	* Ensures eventCount on exit is different (mod 2^32) than on
737	–	* entry.  CAS failures are OK -- any change in count suffices.
738	–	*/
739	–	private void advanceEventCount() {
740	–	int c;
741	–	UNSAFE.compareAndSwapInt(this, eventCountOffset, c = eventCount, c+1);
742	–	}
743	–
744	–	/**
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 759 | 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 841 | 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 861 | Line 826 | public class ForkJoinPool extends Abstra
826		}
827
828		/**
829	<	* Adjusts counts and creates or resumes compensating threads for
830	<	* a worker blocking on task joinMe.  First tries resuming an
831	<	* existing spare (which usually also avoids any count
832	<	* adjustment), but must then decrement running count to determine
868	<	* whether a new thread is needed. See above for fuller
869	<	* explanation. This code is sprawled out non-modularly mainly
870	<	* because adaptive spinning works best if the entire method is
871	<	* either interpreted or compiled vs having only some pieces of it
872	<	* compiled.
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 (to simplify usage by callers)
835	>	* @return task status on exit
836		*/
837	<	final int awaitJoin(ForkJoinTask<?> joinMe) {
838	<	int pc = parallelism;
839	<	boolean adj = false;        // true when running count adjusted
840	<	int scans = 0;
841	<
842	<	while (joinMe.status >= 0) {
843	<	ForkJoinWorkerThread spare = null;
844	<	if ((workerCounts & RUNNING_COUNT_MASK) < pc) {
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) {
#	Line 891 | Line 857 | public class ForkJoinPool extends Abstra
857		break;
858		}
859		}
860	<	if (joinMe.status < 0)
860	>	if ((stat = joinMe.status) < 0) // recheck to narrow race
861		break;
862	<	}
863	<	int wc = workerCounts;
864	<	int rc = wc & RUNNING_COUNT_MASK;
899	<	int dc = pc - rc;
900	<	if (dc > 0 && spare != null && spare.tryUnsuspend()) {
901	<	if (adj) {
902	<	int c;
903	<	do {} while (!UNSAFE.compareAndSwapInt
904	<	(this, workerCountsOffset,
905	<	c = workerCounts, c + ONE_RUNNING));
906	<	}
907	<	adj = true;
908	<	LockSupport.unpark(spare);
909	<	}
910	<	else if (adj) {
911	<	if (dc <= 0)
862	>	int wc = workerCounts;
863	>	int rc = wc & RUNNING_COUNT_MASK;
864	>	if (rc >= pc)
865		break;
866	<	int tc = wc >>> TOTAL_COUNT_SHIFT;
867	<	if (scans > tc) {
868	<	int ts = (tc - pc) * pc;
869	<	if (rc != 0 &&  (dc * dc < ts || !maintainsParallelism))
870	<	break;
871	<	if (scans > ts && tc < maxPoolSize &&
872	<	UNSAFE.compareAndSwapInt(this, workerCountsOffset, wc,
920	<	wc+(ONE_RUNNING|ONE_TOTAL))){
921	<	addWorker();
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	+	}
877	+	int tc = wc >>> TOTAL_COUNT_SHIFT;
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	+	break;
892		}
893		}
894	<	else if (rc != 0)
895	<	adj = UNSAFE.compareAndSwapInt (this, workerCountsOffset,
896	<	wc, wc - ONE_RUNNING);
929	<	if ((scans++ & 1) == 0)
930	<	releaseWaiters();   // help others progress
931	<	else
932	<	Thread.yield();     // avoid starving productive threads
933	<	}
934	<
935	<	if (adj) {
936	<	joinMe.internalAwaitDone();
937	<	int c;
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 joinMe.status;
901	>	return stat;
902		}
903
904		/**
905	<	* Same idea as awaitJoin
905	>	* Same idea as (and mostly pasted from) tryAwaitJoin, but
906	>	* self-contained
907		*/
908	<	final void awaitBlocker(ManagedBlocker blocker, boolean maintainPar)
908	>	final void awaitBlocker(ManagedBlocker blocker)
909		throws InterruptedException {
910	<	maintainPar &= maintainsParallelism;
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;
952	–	boolean adj = false;        // true when running count adjusted
922		int scans = 0;
923	<	boolean done;
955	<
956	<	for (;;) {
923	>	outer: while ((workerCounts & RUNNING_COUNT_MASK) < pc) {
924		if (done = blocker.isReleasable())
925		break;
926		ForkJoinWorkerThread spare = null;
927	<	if ((workerCounts & RUNNING_COUNT_MASK) < pc) {
928	<	ForkJoinWorkerThread[] ws = workers;
929	<	int nws = ws.length;
930	<	for (int i = 0; i < nws; ++i) {
931	<	ForkJoinWorkerThread w = ws[i];
932	<	if (w != null && w.isSuspended()) {
966	<	spare = w;
967	<	break;
968	<	}
969	<	}
970	<	if (done = blocker.isReleasable())
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;
934	+	}
935		}
936	+	if (done = blocker.isReleasable())
937	+	break;
938		int wc = workerCounts;
939		int rc = wc & RUNNING_COUNT_MASK;
940	<	int dc = pc - rc;
941	<	if (dc > 0 && spare != null && spare.tryUnsuspend()) {
942	<	if (adj) {
940	>	if (rc >= pc)
941	>	break;
942	>	if (spare != null) {
943	>	if (spare.tryUnsuspend()) {
944		int c;
945		do {} while (!UNSAFE.compareAndSwapInt
946		(this, workerCountsOffset,
947		c = workerCounts, c + ONE_RUNNING));
948	+	LockSupport.unpark(spare);
949	+	break;
950		}
951	<	adj = true;
984	<	LockSupport.unpark(spare);
951	>	continue;
952		}
953	<	else if (adj) {
954	<	if (dc <= 0)
955	<	break;
956	<	int tc = wc >>> TOTAL_COUNT_SHIFT;
957	<	if (scans > tc) {
958	<	int ts = (tc - pc) * pc;
959	<	if (rc != 0 &&  (dc * dc < ts || !maintainPar))
960	<	break;
994	<	if (scans > ts && tc < maxPoolSize &&
995	<	UNSAFE.compareAndSwapInt(this, workerCountsOffset, wc,
996	<	wc+(ONE_RUNNING|ONE_TOTAL))){
997	<	addWorker();
998	<	break;
999	<	}
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	<	else if (rc != 0)
964	<	adj = UNSAFE.compareAndSwapInt (this, workerCountsOffset,
965	<	wc, wc - ONE_RUNNING);
966	<	if ((++scans & 1) == 0)
967	<	releaseWaiters();   // help others progress
968	<	else
1008	<	Thread.yield();     // avoid starving productive threads
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		}
1010	–
970		try {
971		if (!done)
972	<	do {} while (!blocker.isReleasable() && !blocker.block());
972	>	do {} while (!blocker.isReleasable() &&
973	>	!blocker.block());
974		} finally {
975	<	if (adj) {
976	<	int c;
977	<	do {} while (!UNSAFE.compareAndSwapInt
978	<	(this, workerCountsOffset,
1019	<	c = workerCounts, c + ONE_RUNNING));
1020	<	}
1021	<	}
1022	<	}
1023	<
1024	<	/**
1025	<	* Unless there are not enough other running threads, adjusts
1026	<	* counts and blocks a worker performing helpJoin that cannot find
1027	<	* any work.
1028	<	*
1029	<	* @return true if joinMe now done
1030	<	*/
1031	<	final boolean tryAwaitBusyJoin(ForkJoinTask<?> joinMe) {
1032	<	int pc = parallelism;
1033	<	outer:for (;;) {
1034	<	releaseWaiters();
1035	<	if ((workerCounts & RUNNING_COUNT_MASK) < pc) {
1036	<	ForkJoinWorkerThread[] ws = workers;
1037	<	int nws = ws.length;
1038	<	for (int i = 0; i < nws; ++i) {
1039	<	ForkJoinWorkerThread w = ws[i];
1040	<	if (w != null && w.isSuspended()) {
1041	<	if (joinMe.status < 0)
1042	<	return true;
1043	<	if ((workerCounts & RUNNING_COUNT_MASK) > pc)
1044	<	break;
1045	<	if (w.tryUnsuspend()) {
1046	<	LockSupport.unpark(w);
1047	<	break outer;
1048	<	}
1049	<	continue outer;
1050	<	}
1051	<	}
1052	<	}
1053	<	if (joinMe.status < 0)
1054	<	return true;
1055	<	int wc = workerCounts;
1056	<	if ((wc & RUNNING_COUNT_MASK) <= 2 ||
1057	<	(wc >>> TOTAL_COUNT_SHIFT) < pc)
1058	<	return false;  // keep this thread alive
1059	<	if (UNSAFE.compareAndSwapInt(this, workerCountsOffset,
1060	<	wc, wc - ONE_RUNNING))
1061	<	break;
975	>	int c;
976	>	do {} while (!UNSAFE.compareAndSwapInt
977	>	(this, workerCountsOffset,
978	>	c = workerCounts, c + ONE_RUNNING));
979		}
980	<
1064	<	joinMe.internalAwaitDone();
1065	<	int c;
1066	<	do {} while (!UNSAFE.compareAndSwapInt
1067	<	(this, workerCountsOffset,
1068	<	c = workerCounts, c + ONE_RUNNING));
1069	<	return true;
1070	<	}
980	>	}
981
982		/**
983		* Possibly initiates and/or completes termination.
#	Line 1090 | 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 1103 | Line 1013 | public class ForkJoinPool extends Abstra
1013		cancelSubmissions();
1014		shutdownWorkers();
1015		cancelWorkerTasks();
1016	<	advanceEventCount();
1107	<	releaseWaiters();
1016	>	signalEvent();
1017		interruptWorkers();
1018		}
1019		}
#	Line 1194 | Line 1103 | public class ForkJoinPool extends Abstra
1103		* active thread.
1104		*/
1105		final int idlePerActive() {
1197	–	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 1206 | 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 1216 | 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 1233 | Line 1144 | public class ForkJoinPool extends Abstra
1144		*         java.lang.RuntimePermission}{@code ("modifyThread")}
1145		*/
1146		public ForkJoinPool(int parallelism) {
1147	<	this(parallelism, defaultForkJoinWorkerThreadFactory);
1237	<	}
1238	<
1239	<	/**
1240	<	* Creates a {@code ForkJoinPool} with parallelism equal to {@link
1241	<	* java.lang.Runtime#availableProcessors}, and using the given
1242	<	* thread factory.
1243	<	*
1244	<	* @param factory the factory for creating new threads
1245	<	* @throws NullPointerException if the factory is null
1246	<	* @throws SecurityException if a security manager exists and
1247	<	*         the caller is not permitted to modify threads
1248	<	*         because it does not hold {@link
1249	<	*         java.lang.RuntimePermission}{@code ("modifyThread")}
1250	<	*/
1251	<	public ForkJoinPool(ForkJoinWorkerThreadFactory factory) {
1252	<	this(Runtime.getRuntime().availableProcessors(), factory);
1147	>	this(parallelism, defaultForkJoinWorkerThreadFactory, null, false);
1148		}
1149
1150		/**
1151	<	* Creates a {@code ForkJoinPool} with the given parallelism and
1257	<	* 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 1266 | 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();
1275	–	this.poolNumber = poolNumberGenerator.incrementAndGet();
1276	–	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);
1191	>	this.termination = new Phaser(1);
1192	>	this.poolNumber = poolNumberGenerator.incrementAndGet();
1193		}
1194
1195		/**
#	Line 1308 | 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	<	ensureEnoughTotalWorkers();
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 1330 | 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 1357 | 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 1394 | Line 1332 | public class ForkJoinPool extends Abstra
1332		}
1333
1334		/**
1397	–	* Submits a ForkJoinTask for execution.
1398	–	*
1399	–	* @param task the task to submit
1400	–	* @return the task
1401	–	* @throws NullPointerException if the task is null
1402	–	* @throws RejectedExecutionException if the task cannot be
1403	–	*         scheduled for execution
1404	–	*/
1405	–	public <T> ForkJoinTask<T> submit(ForkJoinTask<T> task) {
1406	–	doSubmit(task);
1407	–	return task;
1408	–	}
1409	–
1410	–	/**
1335		* @throws NullPointerException       {@inheritDoc}
1336		* @throws RejectedExecutionException {@inheritDoc}
1337		*/
#	Line 1449 | Line 1373 | public class ForkJoinPool extends Abstra
1373		* @return the handler, or {@code null} if none
1374		*/
1375		public Thread.UncaughtExceptionHandler getUncaughtExceptionHandler() {
1452	–	workerCountReadFence();
1376		return ueh;
1377		}
1378
1379		/**
1457	–	* Sets the handler for internal worker threads that terminate due
1458	–	* to unrecoverable errors encountered while executing tasks.
1459	–	* Unless set, the current default or ThreadGroup handler is used
1460	–	* as handler.
1461	–	*
1462	–	* @param h the new handler
1463	–	* @return the old handler, or {@code null} if none
1464	–	* @throws SecurityException if a security manager exists and
1465	–	*         the caller is not permitted to modify threads
1466	–	*         because it does not hold {@link
1467	–	*         java.lang.RuntimePermission}{@code ("modifyThread")}
1468	–	*/
1469	–	public Thread.UncaughtExceptionHandler
1470	–	setUncaughtExceptionHandler(Thread.UncaughtExceptionHandler h) {
1471	–	checkPermission();
1472	–	Thread.UncaughtExceptionHandler old = ueh;
1473	–	if (h != old) {
1474	–	ueh = h;
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	–	w.setUncaughtExceptionHandler(h);
1481	–	}
1482	–	}
1483	–	return old;
1484	–	}
1485	–
1486	–	/**
1487	–	* Sets the target parallelism level of this pool.
1488	–	*
1489	–	* @param parallelism the target parallelism
1490	–	* @throws IllegalArgumentException if parallelism less than or
1491	–	* equal to zero or greater than maximum size bounds
1492	–	* @throws SecurityException if a security manager exists and
1493	–	*         the caller is not permitted to modify threads
1494	–	*         because it does not hold {@link
1495	–	*         java.lang.RuntimePermission}{@code ("modifyThread")}
1496	–	*/
1497	–	public void setParallelism(int parallelism) {
1498	–	checkPermission();
1499	–	if (parallelism <= 0 || parallelism > maxPoolSize)
1500	–	throw new IllegalArgumentException();
1501	–	workerCountReadFence();
1502	–	int pc = this.parallelism;
1503	–	if (pc != parallelism) {
1504	–	this.parallelism = parallelism;
1505	–	workerCountWriteFence();
1506	–	// Release spares. If too many, some will die after re-suspend
1507	–	ForkJoinWorkerThread[] ws = workers;
1508	–	int nws = ws.length;
1509	–	for (int i = 0; i < nws; ++i) {
1510	–	ForkJoinWorkerThread w = ws[i];
1511	–	if (w != null && w.tryUnsuspend()) {
1512	–	int c;
1513	–	do {} while (!UNSAFE.compareAndSwapInt
1514	–	(this, workerCountsOffset,
1515	–	c = workerCounts, c + ONE_RUNNING));
1516	–	LockSupport.unpark(w);
1517	–	}
1518	–	}
1519	–	ensureEnoughTotalWorkers();
1520	–	advanceEventCount();
1521	–	releaseWaiters(); // force config recheck by existing workers
1522	–	}
1523	–	}
1524	–
1525	–	/**
1380		* Returns the targeted parallelism level of this pool.
1381		*
1382		* @return the targeted parallelism level of this pool
1383		*/
1384		public int getParallelism() {
1531	–	//        workerCountReadFence(); // inlined below
1532	–	int ignore = workerCounts;
1385		return parallelism;
1386		}
1387
#	Line 1546 | Line 1398 | public class ForkJoinPool extends Abstra
1398		}
1399
1400		/**
1549	–	* Returns the maximum number of threads allowed to exist in the
1550	–	* pool. Unless set using {@link #setMaximumPoolSize}, the
1551	–	* maximum is an implementation-defined value designed only to
1552	–	* prevent runaway growth.
1553	–	*
1554	–	* @return the maximum
1555	–	*/
1556	–	public int getMaximumPoolSize() {
1557	–	workerCountReadFence();
1558	–	return maxPoolSize;
1559	–	}
1560	–
1561	–	/**
1562	–	* Sets the maximum number of threads allowed to exist in the
1563	–	* pool. The given value should normally be greater than or equal
1564	–	* to the {@link #getParallelism parallelism} level. Setting this
1565	–	* value has no effect on current pool size. It controls
1566	–	* construction of new threads. The use of this method may cause
1567	–	* tasks that intrinsically require extra threads for dependent
1568	–	* computations to indefinitely stall. If you are instead trying
1569	–	* to minimize internal thread creation, consider setting {@link
1570	–	* #setMaintainsParallelism} as false.
1571	–	*
1572	–	* @throws IllegalArgumentException if negative or greater than
1573	–	* internal implementation limit
1574	–	*/
1575	–	public void setMaximumPoolSize(int newMax) {
1576	–	if (newMax < 0 || newMax > MAX_THREADS)
1577	–	throw new IllegalArgumentException();
1578	–	maxPoolSize = newMax;
1579	–	workerCountWriteFence();
1580	–	}
1581	–
1582	–	/**
1583	–	* Returns {@code true} if this pool dynamically maintains its
1584	–	* target parallelism level. If false, new threads are added only
1585	–	* to avoid possible starvation.  This setting is by default true.
1586	–	*
1587	–	* @return {@code true} if maintains parallelism
1588	–	*/
1589	–	public boolean getMaintainsParallelism() {
1590	–	workerCountReadFence();
1591	–	return maintainsParallelism;
1592	–	}
1593	–
1594	–	/**
1595	–	* Sets whether this pool dynamically maintains its target
1596	–	* parallelism level. If false, new threads are added only to
1597	–	* avoid possible starvation.
1598	–	*
1599	–	* @param enable {@code true} to maintain parallelism
1600	–	*/
1601	–	public void setMaintainsParallelism(boolean enable) {
1602	–	maintainsParallelism = enable;
1603	–	workerCountWriteFence();
1604	–	}
1605	–
1606	–	/**
1607	–	* Establishes local first-in-first-out scheduling mode for forked
1608	–	* tasks that are never joined. This mode may be more appropriate
1609	–	* than default locally stack-based mode in applications in which
1610	–	* worker threads only process asynchronous tasks.  This method is
1611	–	* designed to be invoked only when the pool is quiescent, and
1612	–	* typically only before any tasks are submitted. The effects of
1613	–	* invocations at other times may be unpredictable.
1614	–	*
1615	–	* @param async if {@code true}, use locally FIFO scheduling
1616	–	* @return the previous mode
1617	–	* @see #getAsyncMode
1618	–	*/
1619	–	public boolean setAsyncMode(boolean async) {
1620	–	workerCountReadFence();
1621	–	boolean oldMode = locallyFifo;
1622	–	if (oldMode != async) {
1623	–	locallyFifo = async;
1624	–	workerCountWriteFence();
1625	–	ForkJoinWorkerThread[] ws = workers;
1626	–	int nws = ws.length;
1627	–	for (int i = 0; i < nws; ++i) {
1628	–	ForkJoinWorkerThread w = ws[i];
1629	–	if (w != null)
1630	–	w.setAsyncMode(async);
1631	–	}
1632	–	}
1633	–	return oldMode;
1634	–	}
1635	–
1636	–	/**
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
1641	–	* @see #setAsyncMode
1405		*/
1406		public boolean getAsyncMode() {
1644	–	workerCountReadFence();
1407		return locallyFifo;
1408		}
1409
#	Line 1782 | 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 1798 | 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 1807 | Line 1586 | public class ForkJoinPool extends Abstra
1586		", steals = " + st +
1587		", tasks = " + qt +
1588		", submissions = " + qs +
1589	+	//            ", parks = " + pk +
1590		"]";
1591		}
1592
#	Line 1902 | 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 1954 | 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.
1958	<	*
1959	<	* <p>If {@code maintainParallelism} is {@code true} and the pool
1960	<	* supports it ({@link #getMaintainsParallelism}), this method
1961	<	* attempts to maintain the pool's nominal parallelism. Otherwise
1962	<	* it activates a thread only if necessary to avoid complete
1963	<	* starvation. This option may be preferable when blockages use
1964	<	* 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 1975 | Line 1752 | public class ForkJoinPool extends Abstra
1752		* first be expanded to ensure parallelism, and later adjusted.
1753		*
1754		* @param blocker the blocker
1978	–	* @param maintainParallelism if {@code true} and supported by
1979	–	* this pool, attempt to maintain the pool's nominal parallelism;
1980	–	* otherwise activate a thread only if necessary to avoid
1981	–	* complete starvation.
1755		* @throws InterruptedException if blocker.block did so
1756		*/
1757	<	public static void managedBlock(ManagedBlocker blocker,
1985	<	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	<	awaitBlocker(blocker, maintainParallelism);
1763	<	else
1764	<	awaitBlocker(blocker);
1993	<	}
1994	<
1995	<	/**
1996	<	* Performs Non-FJ blocking
1997	<	*/
1998	<	private static void awaitBlocker(ManagedBlocker blocker)
1999	<	throws InterruptedException {
2000	<	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 2026 | Line 1790 | public class ForkJoinPool extends Abstra
1790		private static final long stealCountOffset =
1791		objectFieldOffset("stealCount",ForkJoinPool.class);
1792
2029	–
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