ViewVC Help

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

root/jsr166/jsr166/src/jsr166y/ForkJoinWorkerThread.java

Comparing jsr166/src/jsr166y/ForkJoinWorkerThread.java (file contents):
Revision 1.11 by jsr166, Tue Jul 21 00:15:14 2009 UTC vs.
Revision 1.19 by jsr166, Sun Jul 26 05:55:34 2009 UTC

#	Line 5 | Line 5
5		*/
6
7		package jsr166y;
8	<	import java.util.*;
8	>
9		import java.util.concurrent.*;
10	<	import java.util.concurrent.atomic.*;
11	<	import java.util.concurrent.locks.*;
12	<	import sun.misc.Unsafe;
13	<	import java.lang.reflect.*;
10	>
11	>	import java.util.Collection;
12
13		/**
14		* A thread managed by a {@link ForkJoinPool}.  This class is
#	Line 21 | Line 19 | import java.lang.reflect.*;
19		* create such a subclass, you will also need to supply a custom
20		* ForkJoinWorkerThreadFactory to use it in a ForkJoinPool.
21		*
22	+	* @since 1.7
23	+	* @author Doug Lea
24		*/
25		public class ForkJoinWorkerThread extends Thread {
26		/*
#	Line 137 | Line 137 | public class ForkJoinWorkerThread extend
137		private static final int MAXIMUM_QUEUE_CAPACITY = 1 << 28;
138
139		/**
140	<	* The pool this thread works in. Accessed directly by ForkJoinTask
140	>	* The pool this thread works in. Accessed directly by ForkJoinTask.
141		*/
142		final ForkJoinPool pool;
143
#	Line 165 | Line 165 | public class ForkJoinWorkerThread extend
165		* Activity status. When true, this worker is considered active.
166		* Must be false upon construction. It must be true when executing
167		* tasks, and BEFORE stealing a task. It must be false before
168	<	* calling pool.sync
168	>	* calling pool.sync.
169		*/
170		private boolean active;
171
#	Line 188 | Line 188 | public class ForkJoinWorkerThread extend
188
189		/**
190		* Index of this worker in pool array. Set once by pool before
191	<	* running, and accessed directly by pool during cleanup etc
191	>	* running, and accessed directly by pool during cleanup etc.
192		*/
193		int poolIndex;
194
#	Line 273 | Line 273 | public class ForkJoinWorkerThread extend
273		int s = runState;
274		if (s >= state)
275		return false;
276	<	if (_unsafe.compareAndSwapInt(this, runStateOffset, s, state))
276	>	if (UNSAFE.compareAndSwapInt(this, runStateOffset, s, state))
277		return true;
278		}
279		}
#	Line 291 | Line 291 | public class ForkJoinWorkerThread extend
291		}
292
293		/**
294	<	* Tries to set status to active; fails on contention.
294	>	* Tries to set status to inactive; fails on contention.
295		*/
296		private boolean tryInactivate() {
297		if (active) {
#	Line 369 | Line 369 | public class ForkJoinWorkerThread extend
369		/**
370		* Performs cleanup associated with termination of this worker
371		* thread.  If you override this method, you must invoke
372	<	* super.onTermination at the end of the overridden method.
372	>	* {@code super.onTermination} at the end of the overridden method.
373		*
374		* @param exception the exception causing this thread to abort due
375		* to an unrecoverable error, or null if completed normally
#	Line 381 | Line 381 | public class ForkJoinWorkerThread extend
381		ForkJoinTask<?> t = popTask();
382		if (t != null)
383		t.quietlyExec();
384	<	} catch(Throwable ex) {
384	>	} catch (Throwable ex) {
385		exception = ex;
386		}
387		}
388		// Cancel other tasks, transition status, notify pool, and
389		// propagate exception to uncaught exception handler
390		try {
391	<	do;while (!tryInactivate()); // ensure inactive
391	>	do {} while (!tryInactivate()); // ensure inactive
392		cancelTasks();
393		runState = TERMINATED;
394		pool.workerTerminated(this);
#	Line 408 | Line 408 | public class ForkJoinWorkerThread extend
408		* Caller must ensure q is non-null and index is in range.
409		*/
410		private static void setSlot(ForkJoinTask<?>[] q, int i,
411	<	ForkJoinTask<?> t){
412	<	_unsafe.putOrderedObject(q, (i << qShift) + qBase, t);
411	>	ForkJoinTask<?> t) {
412	>	UNSAFE.putOrderedObject(q, (i << qShift) + qBase, t);
413		}
414
415		/**
#	Line 418 | Line 418 | public class ForkJoinWorkerThread extend
418		*/
419		private static boolean casSlotNull(ForkJoinTask<?>[] q, int i,
420		ForkJoinTask<?> t) {
421	<	return _unsafe.compareAndSwapObject(q, (i << qShift) + qBase, t, null);
421	>	return UNSAFE.compareAndSwapObject(q, (i << qShift) + qBase, t, null);
422		}
423
424		/**
425		* Sets sp in store-order.
426		*/
427		private void storeSp(int s) {
428	<	_unsafe.putOrderedInt(this, spOffset, s);
428	>	UNSAFE.putOrderedInt(this, spOffset, s);
429		}
430
431		// Main queue methods
#	Line 515 | Line 515 | public class ForkJoinWorkerThread extend
515		if (q == null)
516		return null;
517		int mask = q.length - 1;
518	<	int i = locallyFifo? base : (sp - 1);
518	>	int i = locallyFifo ? base : (sp - 1);
519		return q[i & mask];
520		}
521
#	Line 578 | Line 578 | public class ForkJoinWorkerThread extend
578		ForkJoinWorkerThread v = ws[mask & idx];
579		if (v == null || v.sp == v.base) {
580		if (probes <= mask)
581	<	idx = (probes++ < 0)? r : (idx + 1);
581	>	idx = (probes++ < 0) ? r : (idx + 1);
582		else
583		break;
584		}
#	Line 599 | Line 599 | public class ForkJoinWorkerThread extend
599		* @return a task, if available
600		*/
601		final ForkJoinTask<?> pollTask() {
602	<	ForkJoinTask<?> t = locallyFifo? deqTask() : popTask();
602	>	ForkJoinTask<?> t = locallyFifo ? deqTask() : popTask();
603		if (t == null && (t = scan()) != null)
604		++stealCount;
605		return t;
#	Line 611 | Line 611 | public class ForkJoinWorkerThread extend
611		* @return a task, if available
612		*/
613		final ForkJoinTask<?> pollLocalTask() {
614	<	return locallyFifo? deqTask() : popTask();
614	>	return locallyFifo ? deqTask() : popTask();
615		}
616
617		/**
#	Line 669 | Line 669 | public class ForkJoinWorkerThread extend
669		/**
670		* Returns true if at least one worker in the given array appears
671		* to have at least one queued task.
672	+	*
673		* @param ws array of workers
674		*/
675		static boolean hasQueuedTasks(ForkJoinWorkerThread[] ws) {
#	Line 691 | Line 692 | public class ForkJoinWorkerThread extend
692		* Returns an estimate of the number of tasks in the queue.
693		*/
694		final int getQueueSize() {
695	<	int n = sp - base;
696	<	return n < 0? 0 : n; // suppress momentarily negative values
695	>	// suppress momentarily negative values
696	>	return Math.max(0, sp - base);
697		}
698
699		/**
#	Line 705 | Line 706 | public class ForkJoinWorkerThread extend
706		}
707
708		/**
709	<	* Scans, returning early if joinMe done
709	>	* Scans, returning early if joinMe done.
710		*/
711		final ForkJoinTask<?> scanWhileJoining(ForkJoinTask<?> joinMe) {
712		ForkJoinTask<?> t = pollTask();
#	Line 717 | Line 718 | public class ForkJoinWorkerThread extend
718		}
719
720		/**
721	<	* Runs tasks until pool isQuiescent.
721	>	* Runs tasks until {@code pool.isQuiescent()}.
722		*/
723		final void helpQuiescePool() {
724		for (;;) {
#	Line 727 | Line 728 | public class ForkJoinWorkerThread extend
728		else if (tryInactivate() && pool.isQuiescent())
729		break;
730		}
731	<	do;while (!tryActivate()); // re-activate on exit
731	>	do {} while (!tryActivate()); // re-activate on exit
732		}
733
734	<	// Temporary Unsafe mechanics for preliminary release
735	<	private static Unsafe getUnsafe() throws Throwable {
734	>	// Unsafe mechanics for jsr166y 3rd party package.
735	>	private static sun.misc.Unsafe getUnsafe() {
736		try {
737	<	return Unsafe.getUnsafe();
737	>	return sun.misc.Unsafe.getUnsafe();
738		} catch (SecurityException se) {
739		try {
740		return java.security.AccessController.doPrivileged
741	<	(new java.security.PrivilegedExceptionAction<Unsafe>() {
742	<	public Unsafe run() throws Exception {
743	<	return getUnsafePrivileged();
741	>	(new java.security.PrivilegedExceptionAction<sun.misc.Unsafe>() {
742	>	public sun.misc.Unsafe run() throws Exception {
743	>	return getUnsafeByReflection();
744		}});
745		} catch (java.security.PrivilegedActionException e) {
746	<	throw e.getCause();
746	>	throw new RuntimeException("Could not initialize intrinsics",
747	>	e.getCause());
748		}
749		}
750		}
751
752	<	private static Unsafe getUnsafePrivileged()
752	>	private static sun.misc.Unsafe getUnsafeByReflection()
753		throws NoSuchFieldException, IllegalAccessException {
754	<	Field f = Unsafe.class.getDeclaredField("theUnsafe");
754	>	java.lang.reflect.Field f =
755	>	sun.misc.Unsafe.class.getDeclaredField("theUnsafe");
756		f.setAccessible(true);
757	<	return (Unsafe) f.get(null);
757	>	return (sun.misc.Unsafe) f.get(null);
758		}
759
760	<	private static long fieldOffset(String fieldName)
761	<	throws NoSuchFieldException {
762	<	return _unsafe.objectFieldOffset
763	<	(ForkJoinWorkerThread.class.getDeclaredField(fieldName));
760	>	private static long fieldOffset(String fieldName, Class<?> klazz) {
761	>	try {
762	>	return UNSAFE.objectFieldOffset(klazz.getDeclaredField(fieldName));
763	>	} catch (NoSuchFieldException e) {
764	>	// Convert Exception to Error
765	>	NoSuchFieldError error = new NoSuchFieldError(fieldName);
766	>	error.initCause(e);
767	>	throw error;
768	>	}
769		}
770
771	<	static final Unsafe _unsafe;
772	<	static final long baseOffset;
773	<	static final long spOffset;
774	<	static final long runStateOffset;
775	<	static final long qBase;
776	<	static final int qShift;
771	>	private static final sun.misc.Unsafe UNSAFE = getUnsafe();
772	>	private static final long spOffset =
773	>	fieldOffset("sp", ForkJoinWorkerThread.class);
774	>	private static final long runStateOffset =
775	>	fieldOffset("runState", ForkJoinWorkerThread.class);
776	>	private static final long qBase;
777	>	private static final int qShift;
778	>
779		static {
780	<	try {
781	<	_unsafe = getUnsafe();
782	<	baseOffset = fieldOffset("base");
783	<	spOffset = fieldOffset("sp");
784	<	runStateOffset = fieldOffset("runState");
775	<	qBase = _unsafe.arrayBaseOffset(ForkJoinTask[].class);
776	<	int s = _unsafe.arrayIndexScale(ForkJoinTask[].class);
777	<	if ((s & (s-1)) != 0)
778	<	throw new Error("data type scale not a power of two");
779	<	qShift = 31 - Integer.numberOfLeadingZeros(s);
780	<	} catch (Throwable e) {
781	<	throw new RuntimeException("Could not initialize intrinsics", e);
782	<	}
780	>	qBase = UNSAFE.arrayBaseOffset(ForkJoinTask[].class);
781	>	int s = UNSAFE.arrayIndexScale(ForkJoinTask[].class);
782	>	if ((s & (s-1)) != 0)
783	>	throw new Error("data type scale not a power of two");
784	>	qShift = 31 - Integer.numberOfLeadingZeros(s);
785		}
786		}

Diff Legend

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