ViewVC Help

View File | Revision Log | Show Annotations | Download File | Root Listing

root/jsr166/jsr166/src/jsr166y/ForkJoinTask.java

Comparing jsr166/src/jsr166y/ForkJoinTask.java (file contents):
Revision 1.47 by dl, Sun Apr 18 12:51:18 2010 UTC vs.
Revision 1.48 by dl, Thu May 27 16:46:48 2010 UTC

#	Line 157 | Line 157 | public abstract class ForkJoinTask<V> im
157		* initially zero, and takes on nonnegative values until
158		* completed, upon which status holds COMPLETED. CANCELLED, or
159		* EXCEPTIONAL, which use the top 3 bits.  Tasks undergoing
160	<	* blocking waits by other threads have SIGNAL_MASK bits set --
161	<	* bit 15 for external (nonFJ) waits, and the rest a count of
162	<	* waiting FJ threads.  (This representation relies on
163	<	* ForkJoinPool max thread limits). Signal counts are not directly
164	<	* incremented by ForkJoinTask methods, but instead via a call to
165	<	* requestSignal within ForkJoinPool.preJoin, once their need is
166	<	* established.
167	<	*
168	<	* Completion of a stolen task with SIGNAL_MASK bits set awakens
169	<	* any waiters via notifyAll. Even though suboptimal for some
170	<	* purposes, we use basic builtin wait/notify to take advantage of
171	<	* "monitor inflation" in JVMs that we would otherwise need to
172	<	* emulate to avoid adding further per-task bookkeeping overhead.
173	<	* We want these monitors to be "fat", i.e., not use biasing or
174	<	* thin-lock techniques, so use some odd coding idioms that tend
175	<	* to avoid them.
160	>	* blocking waits by other threads have the SIGNAL bit set.
161		*
162	<	* Note that bits 16-28 are currently unused. Also value
162	>	* Completion of a stolen task with SIGNAL set awakens any waiters
163	>	* via notifyAll. Even though suboptimal for some purposes, we use
164	>	* basic builtin wait/notify to take advantage of "monitor
165	>	* inflation" in JVMs that we would otherwise need to emulate to
166	>	* avoid adding further per-task bookkeeping overhead.  We want
167	>	* these monitors to be "fat", i.e., not use biasing or thin-lock
168	>	* techniques, so use some odd coding idioms that tend to avoid
169	>	* them.
170	>	*
171	>	* Note that bits 1-28 are currently unused. Also value
172		* 0x80000000 is available as spare completion value.
173		*/
174		volatile int status; // accessed directly by pool and workers
#	Line 183 | Line 177 | public abstract class ForkJoinTask<V> im
177		private static final int NORMAL               = 0xe0000000; // == mask
178		private static final int CANCELLED            = 0xc0000000;
179		private static final int EXCEPTIONAL          = 0xa0000000;
180	<	private static final int SIGNAL_MASK          = 0x0000ffff;
187	<	private static final int INTERNAL_SIGNAL_MASK = 0x00007fff;
188	<	private static final int EXTERNAL_SIGNAL      = 0x00008000;
180	>	private static final int SIGNAL               = 0x00000001;
181
182		/**
183		* Table of exceptions thrown by tasks, to enable reporting by
#	Line 206 | Line 198 | public abstract class ForkJoinTask<V> im
198		* also clearing signal request bits.
199		*
200		* @param completion one of NORMAL, CANCELLED, EXCEPTIONAL
201	+	* @return status on exit
202		*/
203	<	private void setCompletion(int completion) {
203	>	private int setCompletion(int completion) {
204		int s;
205		while ((s = status) >= 0) {
206		if (UNSAFE.compareAndSwapInt(this, statusOffset, s, completion)) {
207	<	if ((s & SIGNAL_MASK) != 0) {
215	<	Thread t = Thread.currentThread();
216	<	if (t instanceof ForkJoinWorkerThread)
217	<	((ForkJoinWorkerThread) t).pool.updateRunningCount
218	<	(s & INTERNAL_SIGNAL_MASK);
207	>	if ((s & SIGNAL) != 0)
208		synchronized (this) { notifyAll(); }
209	<	}
221	<	return;
209	>	return completion;
210		}
211		}
212	+	return s;
213		}
214
215		/**
216		* Record exception and set exceptional completion
217	+	* @return status on exit
218		*/
219	<	private void setDoneExceptionally(Throwable rex) {
219	>	private int setExceptionalCompletion(Throwable rex) {
220		exceptionMap.put(this, rex);
221	<	setCompletion(EXCEPTIONAL);
232	<	}
233	<
234	<	/**
235	<	* Main internal execution method: Unless done, calls exec and
236	<	* records completion.
237	<	*
238	<	* @return true if ran and completed normally
239	<	*/
240	<	final boolean tryExec() {
241	<	try {
242	<	if (status < 0 || !exec())
243	<	return false;
244	<	} catch (Throwable rex) {
245	<	setDoneExceptionally(rex);
246	<	return false;
247	<	}
248	<	setCompletion(NORMAL); // must be outside try block
249	<	return true;
250	<	}
251	<
252	<	/**
253	<	* Increments internal signal count (thus requesting signal upon
254	<	* completion) unless already done.  Call only once per join.
255	<	* Used by ForkJoinPool.preJoin.
256	<	*
257	<	* @return status
258	<	*/
259	<	final int requestSignal() {
260	<	int s;
261	<	do {} while ((s = status) >= 0 &&
262	<	!UNSAFE.compareAndSwapInt(this, statusOffset, s, s + 1));
263	<	return s;
221	>	return setCompletion(EXCEPTIONAL);
222		}
223
224		/**
225	<	* Sets external signal request unless already done.
268	<	*
269	<	* @return status
225	>	* Blocks a worker thread until completion. Called only by pool.
226		*/
227	<	private int requestExternalSignal() {
227	>	final void internalAwaitDone() {
228		int s;
229	<	do {} while ((s = status) >= 0 &&
230	<	!UNSAFE.compareAndSwapInt(this, statusOffset,
231	<	s, s | EXTERNAL_SIGNAL));
232	<	return s;
233	<	}
278	<
279	<	/*
280	<	* Awaiting completion. The four versions, internal vs external X
281	<	* untimed vs timed, have the same overall structure but differ
282	<	* from each other enough to defy simple integration.
283	<	*/
284	<
285	<	/**
286	<	* Blocks a worker until this task is done, also maintaining pool
287	<	* and signal counts
288	<	*/
289	<	private void awaitDone(ForkJoinWorkerThread w) {
290	<	if (status >= 0) {
291	<	w.pool.preJoin(this);
292	<	while (status >= 0) {
293	<	try { // minimize lock scope
294	<	synchronized(this) {
295	<	if (status >= 0)
229	>	while ((s = status) >= 0) {
230	>	synchronized(this) {
231	>	if (UNSAFE.compareAndSwapInt(this, statusOffset, s, s|SIGNAL)){
232	>	do {
233	>	try {
234		wait();
235	<	else { // help release; also helps avoid lock-biasing
236	<	notifyAll();
299	<	break;
235	>	} catch (InterruptedException ie) {
236	>	cancelIfTerminating();
237		}
238	<	}
239	<	} catch (InterruptedException ie) {
303	<	cancelIfTerminating();
238	>	} while (status >= 0);
239	>	break;
240		}
241		}
242		}
243		}
244
245		/**
246	<	* Blocks a non-ForkJoin thread until this task is done.
246	>	* Blocks a non-worker-thread until completion.
247	>	* @return status on exit
248		*/
249	<	private void externalAwaitDone() {
250	<	if (requestExternalSignal() >= 0) {
251	<	boolean interrupted = false;
252	<	while (status >= 0) {
253	<	try {
254	<	synchronized(this) {
255	<	if (status >= 0)
249	>	private int externalAwaitDone() {
250	>	int s;
251	>	while ((s = status) >= 0) {
252	>	synchronized(this) {
253	>	if (UNSAFE.compareAndSwapInt(this, statusOffset, s, s|SIGNAL)){
254	>	boolean interrupted = false;
255	>	do {
256	>	try {
257		wait();
258	<	else {
259	<	notifyAll();
322	<	break;
258	>	} catch (InterruptedException ie) {
259	>	interrupted = true;
260		}
261	<	}
262	<	} catch (InterruptedException ie) {
263	<	interrupted = true;
261	>	} while ((s = status) >= 0);
262	>	if (interrupted)
263	>	Thread.currentThread().interrupt();
264	>	break;
265		}
266		}
329	–	if (interrupted)
330	–	Thread.currentThread().interrupt();
267		}
268	+	return s;
269		}
270
271		/**
272	<	* Blocks a worker until this task is done or timeout elapses
272	>	* Unless done, calls exec and records status if completed, but
273	>	* doesn't wait for completion otherwise.
274		*/
275	<	private void timedAwaitDone(ForkJoinWorkerThread w, long nanos) {
276	<	if (status >= 0) {
277	<	long startTime = System.nanoTime();
278	<	ForkJoinPool pool = w.pool;
279	<	pool.preJoin(this);
280	<	while (status >= 0) {
281	<	long nt = nanos - (System.nanoTime() - startTime);
344	<	if (nt > 0) {
345	<	long ms = nt / 1000000;
346	<	int ns = (int) (nt % 1000000);
347	<	try {
348	<	synchronized(this) { if (status >= 0) wait(ms, ns); }
349	<	} catch (InterruptedException ie) {
350	<	cancelIfTerminating();
351	<	}
352	<	}
353	<	else {
354	<	int s; // adjust running count on timeout
355	<	while ((s = status) >= 0 &&
356	<	(s & INTERNAL_SIGNAL_MASK) != 0) {
357	<	if (UNSAFE.compareAndSwapInt(this, statusOffset,
358	<	s, s - 1)) {
359	<	pool.updateRunningCount(1);
360	<	break;
361	<	}
362	<	}
363	<	break;
364	<	}
365	<	}
275	>	final void tryExec() {
276	>	try {
277	>	if (status < 0 || !exec())
278	>	return;
279	>	} catch (Throwable rex) {
280	>	setExceptionalCompletion(rex);
281	>	return;
282		}
283	+	setCompletion(NORMAL); // must be outside try block
284		}
285
286		/**
287	<	* Blocks a non-ForkJoin thread until this task is done or timeout elapses
288	<	*/
289	<	private void externalTimedAwaitDone(long nanos) {
290	<	if (requestExternalSignal() >= 0) {
291	<	long startTime = System.nanoTime();
292	<	boolean interrupted = false;
293	<	while (status >= 0) {
294	<	long nt = nanos - (System.nanoTime() - startTime);
295	<	if (nt <= 0)
296	<	break;
297	<	long ms = nt / 1000000;
298	<	int ns = (int) (nt % 1000000);
287	>	* If not done and this task is next in worker queue, runs it,
288	>	* else waits for it.
289	>	* @return status on exit
290	>	*/
291	>	private int waitingJoin() {
292	>	int s = status;
293	>	if (s < 0)
294	>	return s;
295	>	Thread t = Thread.currentThread();
296	>	if (t instanceof ForkJoinWorkerThread) {
297	>	ForkJoinWorkerThread w = (ForkJoinWorkerThread) t;
298	>	if (w.unpushTask(this)) {
299	>	boolean completed;
300		try {
301	<	synchronized(this) { if (status >= 0) wait(ms, ns); }
302	<	} catch (InterruptedException ie) {
303	<	interrupted = true;
301	>	completed = exec();
302	>	} catch (Throwable rex) {
303	>	return setExceptionalCompletion(rex);
304		}
305	+	if (completed)
306	+	return setCompletion(NORMAL);
307		}
308	<	if (interrupted)
389	<	Thread.currentThread().interrupt();
308	>	return w.pool.awaitJoin(this);
309		}
310	+	else
311	+	return externalAwaitDone();
312		}
313
393	–	// reporting results
394	–
314		/**
315	<	* Returns result or throws the exception associated with status.
316	<	* Uses Unsafe as a workaround for javac not allowing rethrow of
317	<	* unchecked exceptions.
315	>	* Unless done, calls exec and records status if completed, or
316	>	* waits for completion otherwise.
317	>	* @return status on exit
318		*/
319	<	private V reportResult() {
320	<	if ((status & COMPLETION_MASK) < NORMAL) {
321	<	Throwable ex = getException();
322	<	if (ex != null)
323	<	UNSAFE.throwException(ex);
319	>	private int waitingInvoke() {
320	>	int s = status;
321	>	if (s < 0)
322	>	return s;
323	>	boolean completed;
324	>	try {
325	>	completed = exec();
326	>	} catch (Throwable rex) {
327	>	return setExceptionalCompletion(rex);
328		}
329	<	return getRawResult();
329	>	if (completed)
330	>	return setCompletion(NORMAL);
331	>	return waitingJoin();
332		}
333
334		/**
335	<	* Returns result or throws exception using j.u.c.Future conventions.
336	<	* Only call when {@code isDone} known to be true or thread known
337	<	* to be interrupted.
335	>	* If this task is next in worker queue, runs it, else processes other
336	>	* tasks until complete.
337	>	* @return status on exit
338		*/
339	<	private V reportFutureResult()
340	<	throws InterruptedException, ExecutionException {
341	<	if (Thread.interrupted())
342	<	throw new InterruptedException();
343	<	int s = status & COMPLETION_MASK;
344	<	if (s < NORMAL) {
345	<	Throwable ex;
346	<	if (s == CANCELLED)
347	<	throw new CancellationException();
348	<	if (s == EXCEPTIONAL && (ex = exceptionMap.get(this)) != null)
349	<	throw new ExecutionException(ex);
339	>	private int busyJoin() {
340	>	int s = status;
341	>	if (s < 0)
342	>	return s;
343	>	ForkJoinWorkerThread w = (ForkJoinWorkerThread) Thread.currentThread();
344	>	if (w.unpushTask(this)) {
345	>	boolean completed;
346	>	try {
347	>	completed = exec();
348	>	} catch (Throwable rex) {
349	>	return setExceptionalCompletion(rex);
350	>	}
351	>	if (completed)
352	>	return setCompletion(NORMAL);
353		}
354	<	return getRawResult();
354	>	return w.execWhileJoining(this);
355		}
356
357		/**
358	<	* Returns result or throws exception using j.u.c.Future conventions
359	<	* with timeouts.
358	>	* Returns result or throws exception associated with given status.
359	>	* @param s the status
360		*/
361	<	private V reportTimedFutureResult()
434	<	throws InterruptedException, ExecutionException, TimeoutException {
435	<	if (Thread.interrupted())
436	<	throw new InterruptedException();
361	>	private V reportResult(int s) {
362		Throwable ex;
363	<	int s = status & COMPLETION_MASK;
364	<	if (s == NORMAL)
365	<	return getRawResult();
441	<	else if (s == CANCELLED)
442	<	throw new CancellationException();
443	<	else if (s == EXCEPTIONAL && (ex = exceptionMap.get(this)) != null)
444	<	throw new ExecutionException(ex);
445	<	else
446	<	throw new TimeoutException();
363	>	if (s < NORMAL && (ex = getException()) != null)
364	>	UNSAFE.throwException(ex);
365	>	return getRawResult();
366		}
367
368		// public methods
#	Line 481 | Line 400 | public abstract class ForkJoinTask<V> im
400		* @return the computed result
401		*/
402		public final V join() {
403	<	quietlyJoin();
485	<	return reportResult();
403	>	return reportResult(waitingJoin());
404		}
405
406		/**
#	Line 493 | Line 411 | public abstract class ForkJoinTask<V> im
411		* @return the computed result
412		*/
413		public final V invoke() {
414	<	if (!tryExec())
497	<	quietlyJoin();
498	<	return reportResult();
414	>	return reportResult(waitingInvoke());
415		}
416
417		/**
#	Line 553 | Line 469 | public abstract class ForkJoinTask<V> im
469		}
470		else if (i != 0)
471		t.fork();
472	<	else {
473	<	t.quietlyInvoke();
558	<	if (ex == null)
559	<	ex = t.getException();
560	<	}
472	>	else if (t.waitingInvoke() < NORMAL && ex == null)
473	>	ex = t.getException();
474		}
475		for (int i = 1; i <= last; ++i) {
476		ForkJoinTask<?> t = tasks[i];
477		if (t != null) {
478		if (ex != null)
479		t.cancel(false);
480	<	else {
481	<	t.quietlyJoin();
569	<	if (ex == null)
570	<	ex = t.getException();
571	<	}
480	>	else if (t.waitingJoin() < NORMAL && ex == null)
481	>	ex = t.getException();
482		}
483		}
484		if (ex != null)
#	Line 615 | Line 525 | public abstract class ForkJoinTask<V> im
525		}
526		else if (i != 0)
527		t.fork();
528	<	else {
529	<	t.quietlyInvoke();
620	<	if (ex == null)
621	<	ex = t.getException();
622	<	}
528	>	else if (t.waitingInvoke() < NORMAL && ex == null)
529	>	ex = t.getException();
530		}
531		for (int i = 1; i <= last; ++i) {
532		ForkJoinTask<?> t = ts.get(i);
533		if (t != null) {
534		if (ex != null)
535		t.cancel(false);
536	<	else {
537	<	t.quietlyJoin();
631	<	if (ex == null)
632	<	ex = t.getException();
633	<	}
536	>	else if (t.waitingJoin() < NORMAL && ex == null)
537	>	ex = t.getException();
538		}
539		}
540		if (ex != null)
#	Line 749 | Line 653 | public abstract class ForkJoinTask<V> im
653		* thrown will be a {@code RuntimeException} with cause {@code ex}.
654		*/
655		public void completeExceptionally(Throwable ex) {
656	<	setDoneExceptionally((ex instanceof RuntimeException) ||
657	<	(ex instanceof Error) ? ex :
658	<	new RuntimeException(ex));
656	>	setExceptionalCompletion((ex instanceof RuntimeException) ||
657	>	(ex instanceof Error) ? ex :
658	>	new RuntimeException(ex));
659		}
660
661		/**
#	Line 770 | Line 674 | public abstract class ForkJoinTask<V> im
674		try {
675		setRawResult(value);
676		} catch (Throwable rex) {
677	<	setDoneExceptionally(rex);
677	>	setExceptionalCompletion(rex);
678		return;
679		}
680		setCompletion(NORMAL);
681		}
682
683		public final V get() throws InterruptedException, ExecutionException {
684	<	quietlyJoin();
685	<	return reportFutureResult();
684	>	int s = waitingJoin() & COMPLETION_MASK;
685	>	if (Thread.interrupted())
686	>	throw new InterruptedException();
687	>	if (s < NORMAL) {
688	>	Throwable ex;
689	>	if (s == CANCELLED)
690	>	throw new CancellationException();
691	>	if (s == EXCEPTIONAL && (ex = exceptionMap.get(this)) != null)
692	>	throw new ExecutionException(ex);
693	>	}
694	>	return getRawResult();
695		}
696
697		public final V get(long timeout, TimeUnit unit)
698		throws InterruptedException, ExecutionException, TimeoutException {
786	–	long nanos = unit.toNanos(timeout);
699		Thread t = Thread.currentThread();
700	+	ForkJoinPool pool;
701		if (t instanceof ForkJoinWorkerThread) {
702		ForkJoinWorkerThread w = (ForkJoinWorkerThread) t;
703	<	if (!w.unpushTask(this) || !tryExec())
704	<	timedAwaitDone(w, nanos);
703	>	if (status >= 0 && w.unpushTask(this))
704	>	tryExec();
705	>	pool = w.pool;
706		}
707		else
708	<	externalTimedAwaitDone(nanos);
709	<	return reportTimedFutureResult();
708	>	pool = null;
709	>	/*
710	>	* Timed wait loop intermixes cases for fj (pool != null) and
711	>	* non FJ threads. For FJ, decrement pool count but don't try
712	>	* for replacement; increment count on completion. For non-FJ,
713	>	* deal with interrupts. This is messy, but a little less so
714	>	* than is splitting the FJ and nonFJ cases.
715	>	*/
716	>	boolean interrupted = false;
717	>	boolean dec = false; // true if pool count decremented
718	>	for (;;) {
719	>	if (Thread.interrupted() && pool == null) {
720	>	interrupted = true;
721	>	break;
722	>	}
723	>	int s = status;
724	>	if (s < 0)
725	>	break;
726	>	if (UNSAFE.compareAndSwapInt(this, statusOffset,
727	>	s, s | SIGNAL)) {
728	>	long startTime = System.nanoTime();
729	>	long nanos = unit.toNanos(timeout);
730	>	long nt; // wait time
731	>	while (status >= 0 &&
732	>	(nt = nanos - (System.nanoTime() - startTime)) > 0) {
733	>	if (pool != null && !dec)
734	>	dec = pool.tryDecrementRunningCount();
735	>	else {
736	>	long ms = nt / 1000000;
737	>	int ns = (int) (nt % 1000000);
738	>	try {
739	>	synchronized(this) {
740	>	if (status >= 0)
741	>	wait(ms, ns);
742	>	}
743	>	} catch (InterruptedException ie) {
744	>	if (pool != null)
745	>	cancelIfTerminating();
746	>	else {
747	>	interrupted = true;
748	>	break;
749	>	}
750	>	}
751	>	}
752	>	}
753	>	break;
754	>	}
755	>	}
756	>	if (pool != null && dec)
757	>	pool.updateRunningCount(1);
758	>	if (interrupted)
759	>	throw new InterruptedException();
760	>	int es = status & COMPLETION_MASK;
761	>	if (es != NORMAL) {
762	>	Throwable ex;
763	>	if (es == CANCELLED)
764	>	throw new CancellationException();
765	>	if (es == EXCEPTIONAL && (ex = exceptionMap.get(this)) != null)
766	>	throw new ExecutionException(ex);
767	>	throw new TimeoutException();
768	>	}
769	>	return getRawResult();
770		}
771
772		/**
#	Line 813 | Line 787 | public abstract class ForkJoinTask<V> im
787		* @return the computed result
788		*/
789		public final V helpJoin() {
790	<	quietlyHelpJoin();
817	<	return reportResult();
790	>	return reportResult(busyJoin());
791		}
792
793		/**
#	Line 830 | Line 803 | public abstract class ForkJoinTask<V> im
803		* ClassCastException}.
804		*/
805		public final void quietlyHelpJoin() {
806	<	ForkJoinWorkerThread w = (ForkJoinWorkerThread) Thread.currentThread();
834	<	if (!w.unpushTask(this) || !tryExec()) {
835	<	for (;;) {
836	<	ForkJoinTask<?> t;
837	<	if (status < 0)
838	<	return;
839	<	else if ((t = w.scanWhileJoining(this)) != null)
840	<	t.tryExec();
841	<	else if (status < 0)
842	<	return;
843	<	else if (w.pool.preBlockHelpingJoin(this)) {
844	<	while (status >= 0) { // variant of awaitDone
845	<	try {
846	<	synchronized(this) {
847	<	if (status >= 0)
848	<	wait();
849	<	else {
850	<	notifyAll();
851	<	break;
852	<	}
853	<	}
854	<	} catch (InterruptedException ie) {
855	<	cancelIfTerminating();
856	<	}
857	<	}
858	<	return;
859	<	}
860	<	}
861	<	}
806	>	busyJoin();
807		}
808
809		/**
#	Line 868 | Line 813 | public abstract class ForkJoinTask<V> im
813		* known to have aborted.
814		*/
815		public final void quietlyJoin() {
816	<	Thread t = Thread.currentThread();
872	<	if (t instanceof ForkJoinWorkerThread) {
873	<	ForkJoinWorkerThread w = (ForkJoinWorkerThread) t;
874	<	if (!w.unpushTask(this) || !tryExec())
875	<	awaitDone(w);
876	<	}
877	<	else
878	<	externalAwaitDone();
816	>	waitingJoin();
817		}
818
819		/**
#	Line 886 | Line 824 | public abstract class ForkJoinTask<V> im
824		* known to have aborted.
825		*/
826		public final void quietlyInvoke() {
827	<	if (!tryExec())
890	<	quietlyJoin();
827	>	waitingInvoke();
828		}
829
830		/**
#	Line 1229 | Line 1166 | public abstract class ForkJoinTask<V> im
1166		private void readObject(java.io.ObjectInputStream s)
1167		throws java.io.IOException, ClassNotFoundException {
1168		s.defaultReadObject();
1169	<	status &= ~INTERNAL_SIGNAL_MASK; // clear internal signal counts
1233	<	status |= EXTERNAL_SIGNAL; // conservatively set external signal
1169	>	status |= SIGNAL; // conservatively set external signal
1170		Object ex = s.readObject();
1171		if (ex != null)
1172	<	setDoneExceptionally((Throwable) ex);
1172	>	setExceptionalCompletion((Throwable) ex);
1173		}
1174
1175		// Unsafe mechanics

Diff Legend

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