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.85 by jsr166, Tue Jan 31 01:51:13 2012 UTC vs.
Revision 1.86 by dl, Mon Feb 20 18:20:06 2012 UTC

#	Line 197 | Line 197 | public abstract class ForkJoinTask<V> im
197		* methods in a way that flows well in javadocs.
198		*/
199
200	–	/**
201	–	* The number of times to try to help join a task without any
202	–	* apparent progress before giving up and blocking. The value is
203	–	* arbitrary but should be large enough to cope with transient
204	–	* stalls (due to GC etc) that can cause helping methods not to be
205	–	* able to proceed because other workers have not progressed to
206	–	* the point where subtasks can be found or taken.
207	–	*/
208	–	private static final int HELP_RETRIES = 32;
209	–
200		/*
201		* The status field holds run control status bits packed into a
202		* single int to minimize footprint and to ensure atomicity (via
203		* CAS).  Status is initially zero, and takes on nonnegative
204	<	* values until completed, upon which status holds value
205	<	* NORMAL, CANCELLED, or EXCEPTIONAL. Tasks undergoing blocking
206	<	* waits by other threads have the SIGNAL bit set.  Completion of
207	<	* a stolen task with SIGNAL set awakens any waiters via
208	<	* notifyAll. Even though suboptimal for some purposes, we use
209	<	* basic builtin wait/notify to take advantage of "monitor
210	<	* inflation" in JVMs that we would otherwise need to emulate to
211	<	* avoid adding further per-task bookkeeping overhead.  We want
212	<	* these monitors to be "fat", i.e., not use biasing or thin-lock
213	<	* techniques, so use some odd coding idioms that tend to avoid
214	<	* them.
204	>	* values until completed, upon which status (anded with
205	>	* DONE_MASK) holds value NORMAL, CANCELLED, or EXCEPTIONAL. Tasks
206	>	* undergoing blocking waits by other threads have the SIGNAL bit
207	>	* set.  Completion of a stolen task with SIGNAL set awakens any
208	>	* waiters via notifyAll. Even though suboptimal for some
209	>	* purposes, we use basic builtin wait/notify to take advantage of
210	>	* "monitor inflation" in JVMs that we would otherwise need to
211	>	* emulate to avoid adding further per-task bookkeeping overhead.
212	>	* We want these monitors to be "fat", i.e., not use biasing or
213	>	* thin-lock techniques, so use some odd coding idioms that tend
214	>	* to avoid them, mainly by arranging that every synchronized
215	>	* block performs a wait, notifyAll or both.
216		*/
217
218		/** The run status of this task */
219		volatile int status; // accessed directly by pool and workers
220	<	static final int NORMAL      = 0xfffffffc;  // negative with low 2 bits 0
221	<	static final int CANCELLED   = 0xfffffff8;  // must be < NORMAL
222	<	static final int EXCEPTIONAL = 0xfffffff4;  // must be < CANCELLED
220	>	static final int DONE_MASK   = 0xf0000000;  // mask out non-completion bits
221	>	static final int NORMAL      = 0xf0000000;  // must be negative
222	>	static final int CANCELLED   = 0xc0000000;  // must be < NORMAL
223	>	static final int EXCEPTIONAL = 0x80000000;  // must be < CANCELLED
224		static final int SIGNAL      = 0x00000001;
225		static final int MARKED      = 0x00000002;
226
227		/**
228		* Marks completion and wakes up threads waiting to join this
229	<	* task, also clearing signal request bits. A specialization for
230	<	* NORMAL completion is in method doExec.
229	>	* task. A specialization for NORMAL completion is in method
230	>	* doExec.
231		*
232		* @param completion one of NORMAL, CANCELLED, EXCEPTIONAL
233		* @return completion status on exit
#	Line 244 | Line 236 | public abstract class ForkJoinTask<V> im
236		for (int s;;) {
237		if ((s = status) < 0)
238		return s;
239	<	if (U.compareAndSwapInt(this, STATUS, s, (s & ~SIGNAL)|completion)) {
239	>	if (U.compareAndSwapInt(this, STATUS, s, s | completion)) {
240		if ((s & SIGNAL) != 0)
241		synchronized (this) { notifyAll(); }
242		return completion;
#	Line 268 | Line 260 | public abstract class ForkJoinTask<V> im
260		return setExceptionalCompletion(rex);
261		}
262		while ((s = status) >= 0 && completed) {
263	<	if (U.compareAndSwapInt(this, STATUS, s, (s & ~SIGNAL)|NORMAL)) {
263	>	if (U.compareAndSwapInt(this, STATUS, s, s | NORMAL)) {
264		if ((s & SIGNAL) != 0)
265		synchronized (this) { notifyAll(); }
266		return NORMAL;
#	Line 279 | Line 271 | public abstract class ForkJoinTask<V> im
271		}
272
273		/**
274	+	* Tries to set SIGNAL status. Used by ForkJoinPool. Other
275	+	* variants are directly incorporated into externalAwaitDone etc.
276	+	*
277	+	* @return true if successful
278	+	*/
279	+	final boolean trySetSignal() {
280	+	int s;
281	+	return U.compareAndSwapInt(this, STATUS, s = status, s | SIGNAL);
282	+	}
283	+
284	+	/**
285		* Blocks a non-worker-thread until completion.
286		* @return status upon completion
287		*/
288		private int externalAwaitDone() {
289	+	boolean interrupted = false;
290		int s;
291	<	if ((s = status) >= 0) {
292	<	boolean interrupted = false;
293	<	synchronized (this) {
294	<	while ((s = status) >= 0) {
291	<	if (U.compareAndSwapInt(this, STATUS, s, s | SIGNAL)) {
291	>	while ((s = status) >= 0) {
292	>	if (U.compareAndSwapInt(this, STATUS, s, s | SIGNAL)) {
293	>	synchronized (this) {
294	>	if (status >= 0) {
295		try {
296		wait();
297		} catch (InterruptedException ie) {
298		interrupted = true;
299		}
300		}
301	+	else
302	+	notifyAll();
303		}
304		}
300	–	if (interrupted)
301	–	Thread.currentThread().interrupt();
305		}
306	+	if (interrupted)
307	+	Thread.currentThread().interrupt();
308		return s;
309		}
310
311		/**
312	<	* Blocks a non-worker-thread until completion or interruption or timeout.
312	>	* Blocks a non-worker-thread until completion or interruption.
313		*/
314	<	private int externalInterruptibleAwaitDone(long millis)
310	<	throws InterruptedException {
314	>	private int externalInterruptibleAwaitDone() throws InterruptedException {
315		int s;
316		if (Thread.interrupted())
317		throw new InterruptedException();
318	<	if ((s = status) >= 0) {
319	<	synchronized (this) {
320	<	while ((s = status) >= 0) {
321	<	if (U.compareAndSwapInt(this, STATUS, s, s | SIGNAL)) {
322	<	wait(millis);
323	<	if (millis > 0L)
324	<	break;
321	<	}
318	>	while ((s = status) >= 0) {
319	>	if (U.compareAndSwapInt(this, STATUS, s, s | SIGNAL)) {
320	>	synchronized (this) {
321	>	if (status >= 0)
322	>	wait();
323	>	else
324	>	notifyAll();
325		}
326		}
327		}
#	Line 329 | Line 332 | public abstract class ForkJoinTask<V> im
332		/**
333		* Implementation for join, get, quietlyJoin. Directly handles
334		* only cases of already-completed, external wait, and
335	<	* unfork+exec.  Others are relayed to awaitJoin.
335	>	* unfork+exec.  Others are relayed to ForkJoinPool.awaitJoin.
336		*
337		* @return status upon completion
338		*/
339		private int doJoin() {
340		int s; Thread t; ForkJoinWorkerThread wt; ForkJoinPool.WorkQueue w;
341		if ((s = status) >= 0) {
342	<	if (!((t = Thread.currentThread()) instanceof ForkJoinWorkerThread))
343	<	s = externalAwaitDone();
344	<	else if (!(w = (wt = (ForkJoinWorkerThread)t).workQueue).
345	<	tryUnpush(this) || (s = doExec()) >= 0)
343	<	s = awaitJoin(w, wt.pool);
344	<	}
345	<	return s;
346	<	}
347	<
348	<	/**
349	<	* Helps and/or blocks until joined.
350	<	*
351	<	* @param w the joiner
352	<	* @param p the pool
353	<	* @return status upon completion
354	<	*/
355	<	private int awaitJoin(ForkJoinPool.WorkQueue w, ForkJoinPool p) {
356	<	int s;
357	<	ForkJoinTask<?> prevJoin = w.currentJoin;
358	<	w.currentJoin = this;
359	<	for (int k = HELP_RETRIES; (s = status) >= 0;) {
360	<	if ((w.queueSize() > 0) ?
361	<	w.tryRemoveAndExec(this) :        // self-help
362	<	p.tryHelpStealer(w, this))        // help process tasks
363	<	k = HELP_RETRIES;                 // reset if made progress
364	<	else if ((s = status) < 0)            // recheck
365	<	break;
366	<	else if (--k > 0) {
367	<	if ((k & 3) == 1)
368	<	Thread.yield();               // occasionally yield
369	<	}
370	<	else if (k == 0)
371	<	p.tryPollForAndExec(w, this);     // uncommon self-help case
372	<	else if (p.tryCompensate()) {         // true if can block
373	<	try {
374	<	int ss = status;
375	<	if (ss >= 0 &&                // assert need signal
376	<	U.compareAndSwapInt(this, STATUS, ss, ss | SIGNAL)) {
377	<	synchronized (this) {
378	<	if (status >= 0)      // block
379	<	wait();
380	<	}
381	<	}
382	<	} catch (InterruptedException ignore) {
383	<	} finally {
384	<	p.incrementActiveCount();     // re-activate
385	<	}
342	>	if (((t = Thread.currentThread()) instanceof ForkJoinWorkerThread)) {
343	>	if (!(w = (wt = (ForkJoinWorkerThread)t).workQueue).
344	>	tryUnpush(this) || (s = doExec()) >= 0)
345	>	s = wt.pool.awaitJoin(w, this);
346		}
347	+	else
348	+	s = externalAwaitDone();
349		}
388	–	w.currentJoin = prevJoin;
350		return s;
351		}
352
#	Line 395 | Line 356 | public abstract class ForkJoinTask<V> im
356		* @return status upon completion
357		*/
358		private int doInvoke() {
359	<	int s; Thread t;
359	>	int s; Thread t; ForkJoinWorkerThread wt;
360		if ((s = doExec()) >= 0) {
361	<	if (!((t = Thread.currentThread()) instanceof ForkJoinWorkerThread))
361	>	if ((t = Thread.currentThread()) instanceof ForkJoinWorkerThread)
362	>	s = (wt = (ForkJoinWorkerThread)t).pool.awaitJoin(wt.workQueue,
363	>	this);
364	>	else
365		s = externalAwaitDone();
402	–	else {
403	–	ForkJoinWorkerThread wt = (ForkJoinWorkerThread)t;
404	–	s = awaitJoin(wt.workQueue, wt.pool);
405	–	}
366		}
367		return s;
368		}
#	Line 539 | Line 499 | public abstract class ForkJoinTask<V> im
499		* @return the exception, or null if none
500		*/
501		private Throwable getThrowableException() {
502	<	if (status != EXCEPTIONAL)
502	>	if ((status & DONE_MASK) != EXCEPTIONAL)
503		return null;
504		int h = System.identityHashCode(this);
505		ExceptionNode e;
#	Line 624 | Line 584 | public abstract class ForkJoinTask<V> im
584		}
585
586		/**
587	<	* Report the result of invoke or join; called only upon
628	<	* non-normal return of internal versions.
587	>	* Throws exception, if any, associated with the given status.
588		*/
589	<	private V reportResult() {
590	<	int s; Throwable ex;
591	<	if ((s = status) == CANCELLED)
592	<	throw new CancellationException();
593	<	if (s == EXCEPTIONAL && (ex = getThrowableException()) != null)
589	>	private void reportException(int s) {
590	>	Throwable ex = ((s == CANCELLED) ?  new CancellationException() :
591	>	(s == EXCEPTIONAL) ? getThrowableException() :
592	>	null);
593	>	if (ex != null)
594		U.throwException(ex);
636	–	return getRawResult();
595		}
596
597		// public methods
#	Line 657 | Line 615 | public abstract class ForkJoinTask<V> im
615		* @return {@code this}, to simplify usage
616		*/
617		public final ForkJoinTask<V> fork() {
618	<	ForkJoinWorkerThread wt;
661	<	(wt = (ForkJoinWorkerThread)Thread.currentThread()).
662	<	workQueue.push(this, wt.pool);
618	>	((ForkJoinWorkerThread)Thread.currentThread()).workQueue.push(this);
619		return this;
620		}
621
#	Line 675 | Line 631 | public abstract class ForkJoinTask<V> im
631		* @return the computed result
632		*/
633		public final V join() {
634	<	if (doJoin() != NORMAL)
635	<	return reportResult();
636	<	else
637	<	return getRawResult();
634	>	int s;
635	>	if ((s = doJoin() & DONE_MASK) != NORMAL)
636	>	reportException(s);
637	>	return getRawResult();
638		}
639
640		/**
#	Line 690 | Line 646 | public abstract class ForkJoinTask<V> im
646		* @return the computed result
647		*/
648		public final V invoke() {
649	<	if (doInvoke() != NORMAL)
650	<	return reportResult();
651	<	else
652	<	return getRawResult();
649	>	int s;
650	>	if ((s = doInvoke() & DONE_MASK) != NORMAL)
651	>	reportException(s);
652	>	return getRawResult();
653		}
654
655		/**
#	Line 720 | Line 676 | public abstract class ForkJoinTask<V> im
676		* @throws NullPointerException if any task is null
677		*/
678		public static void invokeAll(ForkJoinTask<?> t1, ForkJoinTask<?> t2) {
679	+	int s1, s2;
680		t2.fork();
681	<	t1.invoke();
682	<	t2.join();
681	>	if ((s1 = t1.doInvoke() & DONE_MASK) != NORMAL)
682	>	t1.reportException(s1);
683	>	if ((s2 = t2.doJoin() & DONE_MASK) != NORMAL)
684	>	t2.reportException(s2);
685		}
686
687		/**
#	Line 859 | Line 818 | public abstract class ForkJoinTask<V> im
818		* @return {@code true} if this task is now cancelled
819		*/
820		public boolean cancel(boolean mayInterruptIfRunning) {
821	<	return setCompletion(CANCELLED) == CANCELLED;
821	>	return (setCompletion(CANCELLED) & DONE_MASK) == CANCELLED;
822		}
823
824		public final boolean isDone() {
#	Line 867 | Line 826 | public abstract class ForkJoinTask<V> im
826		}
827
828		public final boolean isCancelled() {
829	<	return status == CANCELLED;
829	>	return (status & DONE_MASK) == CANCELLED;
830		}
831
832		/**
#	Line 887 | Line 846 | public abstract class ForkJoinTask<V> im
846		* exception and was not cancelled
847		*/
848		public final boolean isCompletedNormally() {
849	<	return status == NORMAL;
849	>	return (status & DONE_MASK) == NORMAL;
850		}
851
852		/**
#	Line 898 | Line 857 | public abstract class ForkJoinTask<V> im
857		* @return the exception, or {@code null} if none
858		*/
859		public final Throwable getException() {
860	<	int s = status;
860	>	int s = status & DONE_MASK;
861		return ((s >= NORMAL)    ? null :
862		(s == CANCELLED) ? new CancellationException() :
863		getThrowableException());
#	Line 960 | Line 919 | public abstract class ForkJoinTask<V> im
919		*/
920		public final V get() throws InterruptedException, ExecutionException {
921		int s = (Thread.currentThread() instanceof ForkJoinWorkerThread) ?
922	<	doJoin() : externalInterruptibleAwaitDone(0L);
922	>	doJoin() : externalInterruptibleAwaitDone();
923		Throwable ex;
924	<	if (s == CANCELLED)
924	>	if ((s &= DONE_MASK) == CANCELLED)
925		throw new CancellationException();
926		if (s == EXCEPTIONAL && (ex = getThrowableException()) != null)
927		throw new ExecutionException(ex);
#	Line 985 | Line 944 | public abstract class ForkJoinTask<V> im
944		*/
945		public final V get(long timeout, TimeUnit unit)
946		throws InterruptedException, ExecutionException, TimeoutException {
947	<	// Messy in part because we measure in nanos, but wait in millis
948	<	int s; long millis, nanos;
949	<	Thread t = Thread.currentThread();
950	<	if (!(t instanceof ForkJoinWorkerThread)) {
951	<	if ((millis = unit.toMillis(timeout)) > 0L)
952	<	s = externalInterruptibleAwaitDone(millis);
953	<	else
954	<	s = status;
955	<	}
956	<	else if ((s = status) >= 0 && (nanos = unit.toNanos(timeout)) > 0L) {
957	<	long deadline = System.nanoTime() + nanos;
958	<	ForkJoinWorkerThread wt = (ForkJoinWorkerThread)t;
959	<	ForkJoinPool.WorkQueue w = wt.workQueue;
960	<	ForkJoinPool p = wt.pool;
961	<	if (w.tryUnpush(this))
962	<	doExec();
963	<	boolean blocking = false;
947	>	if (Thread.interrupted())
948	>	throw new InterruptedException();
949	>	// Messy in part because we measure in nanosecs, but wait in millisecs
950	>	int s; long ns, ms;
951	>	if ((s = status) >= 0 && (ns = unit.toNanos(timeout)) > 0L) {
952	>	long deadline = System.nanoTime() + ns;
953	>	ForkJoinPool p = null;
954	>	ForkJoinPool.WorkQueue w = null;
955	>	Thread t = Thread.currentThread();
956	>	if (t instanceof ForkJoinWorkerThread) {
957	>	ForkJoinWorkerThread wt = (ForkJoinWorkerThread)t;
958	>	p = wt.pool;
959	>	w = wt.workQueue;
960	>	s = p.helpJoinOnce(w, this); // no retries on failure
961	>	}
962	>	boolean canBlock = false;
963	>	boolean interrupted = false;
964		try {
965		while ((s = status) >= 0) {
966	<	if (w.runState < 0)
966	>	if (w != null && w.runState < 0)
967		cancelIgnoringExceptions(this);
968	<	else if (!blocking)
969	<	blocking = p.tryCompensate();
968	>	else if (!canBlock) {
969	>	if (p == null || p.tryCompensate(this, null))
970	>	canBlock = true;
971	>	}
972		else {
973	<	millis = TimeUnit.NANOSECONDS.toMillis(nanos);
1013	<	if (millis > 0L &&
973	>	if ((ms = TimeUnit.NANOSECONDS.toMillis(ns)) > 0L &&
974		U.compareAndSwapInt(this, STATUS, s, s | SIGNAL)) {
975	<	try {
976	<	synchronized (this) {
977	<	if (status >= 0)
978	<	wait(millis);
975	>	synchronized (this) {
976	>	if (status >= 0) {
977	>	try {
978	>	wait(ms);
979	>	} catch (InterruptedException ie) {
980	>	if (p == null)
981	>	interrupted = true;
982	>	}
983		}
984	<	} catch (InterruptedException ie) {
984	>	else
985	>	notifyAll();
986		}
987		}
988	<	if ((s = status) < 0 ||
989	<	(nanos = deadline - System.nanoTime()) <= 0L)
988	>	if ((s = status) < 0 || interrupted ||
989	>	(ns = deadline - System.nanoTime()) <= 0L)
990		break;
991		}
992		}
993		} finally {
994	<	if (blocking)
994	>	if (p != null && canBlock)
995		p.incrementActiveCount();
996		}
997	+	if (interrupted)
998	+	throw new InterruptedException();
999		}
1000	<	if (s != NORMAL) {
1000	>	if ((s &= DONE_MASK) != NORMAL) {
1001		Throwable ex;
1002		if (s == CANCELLED)
1003		throw new CancellationException();
#	Line 1097 | Line 1064 | public abstract class ForkJoinTask<V> im
1064		* setRawResult(null)}.
1065		*/
1066		public void reinitialize() {
1067	<	if (status == EXCEPTIONAL)
1067	>	if ((status & DONE_MASK) == EXCEPTIONAL)
1068		clearExceptionalCompletion();
1069		else
1070		status = 0;
#	Line 1385 | Line 1352 | public abstract class ForkJoinTask<V> im
1352		static final class AdaptedRunnable<T> extends ForkJoinTask<T>
1353		implements RunnableFuture<T> {
1354		final Runnable runnable;
1388	–	final T resultOnCompletion;
1355		T result;
1356		AdaptedRunnable(Runnable runnable, T result) {
1357		if (runnable == null) throw new NullPointerException();
1358		this.runnable = runnable;
1359	<	this.resultOnCompletion = result;
1359	>	this.result = result; // OK to set this even before completion
1360		}
1361	<	public T getRawResult() { return result; }
1362	<	public void setRawResult(T v) { result = v; }
1363	<	public boolean exec() {
1364	<	runnable.run();
1365	<	result = resultOnCompletion;
1366	<	return true;
1361	>	public final T getRawResult() { return result; }
1362	>	public final void setRawResult(T v) { result = v; }
1363	>	public final boolean exec() { runnable.run(); return true; }
1364	>	public final void run() { invoke(); }
1365	>	private static final long serialVersionUID = 5232453952276885070L;
1366	>	}
1367	>
1368	>	/**
1369	>	* Adaptor for Runnables without results
1370	>	*/
1371	>	static final class AdaptedRunnableAction extends ForkJoinTask<Void>
1372	>	implements RunnableFuture<Void> {
1373	>	final Runnable runnable;
1374	>	AdaptedRunnableAction(Runnable runnable) {
1375	>	if (runnable == null) throw new NullPointerException();
1376	>	this.runnable = runnable;
1377		}
1378	<	public void run() { invoke(); }
1378	>	public final Void getRawResult() { return null; }
1379	>	public final void setRawResult(Void v) { }
1380	>	public final boolean exec() { runnable.run(); return true; }
1381	>	public final void run() { invoke(); }
1382		private static final long serialVersionUID = 5232453952276885070L;
1383		}
1384
#	Line 1414 | Line 1393 | public abstract class ForkJoinTask<V> im
1393		if (callable == null) throw new NullPointerException();
1394		this.callable = callable;
1395		}
1396	<	public T getRawResult() { return result; }
1397	<	public void setRawResult(T v) { result = v; }
1398	<	public boolean exec() {
1396	>	public final T getRawResult() { return result; }
1397	>	public final void setRawResult(T v) { result = v; }
1398	>	public final boolean exec() {
1399		try {
1400		result = callable.call();
1401		return true;
#	Line 1428 | Line 1407 | public abstract class ForkJoinTask<V> im
1407		throw new RuntimeException(ex);
1408		}
1409		}
1410	<	public void run() { invoke(); }
1410	>	public final void run() { invoke(); }
1411		private static final long serialVersionUID = 2838392045355241008L;
1412		}
1413
#	Line 1441 | Line 1420 | public abstract class ForkJoinTask<V> im
1420		* @return the task
1421		*/
1422		public static ForkJoinTask<?> adapt(Runnable runnable) {
1423	<	return new AdaptedRunnable<Void>(runnable, null);
1423	>	return new AdaptedRunnableAction(runnable);
1424		}
1425
1426		/**

Diff Legend

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