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.18 by jsr166, Sat Jul 25 00:34:00 2009 UTC vs.
Revision 1.26 by dl, Sun Aug 2 11:54:31 2009 UTC

#	Line 13 | Line 13 | import java.util.Collection;
13		/**
14		* A thread managed by a {@link ForkJoinPool}.  This class is
15		* subclassable solely for the sake of adding functionality -- there
16	<	* are no overridable methods dealing with scheduling or
17	<	* execution. However, you can override initialization and termination
18	<	* methods surrounding the main task processing loop.  If you do
19	<	* create such a subclass, you will also need to supply a custom
20	<	* ForkJoinWorkerThreadFactory to use it in a ForkJoinPool.
16	>	* are no overridable methods dealing with scheduling or execution.
17	>	* However, you can override initialization and termination methods
18	>	* surrounding the main task processing loop.  If you do create such a
19	>	* subclass, you will also need to supply a custom {@link
20	>	* ForkJoinPool.ForkJoinWorkerThreadFactory} to use it in a {@code
21	>	* ForkJoinPool}.
22		*
23		* @since 1.7
24		* @author Doug Lea
#	Line 65 | Line 66 | public class ForkJoinWorkerThread extend
66		* which gives threads a chance to activate if necessary before
67		* stealing (see below).
68		*
69	+	* This approach also enables support for "async mode" where local
70	+	* task processing is in FIFO, not LIFO order; simply by using a
71	+	* version of deq rather than pop when locallyFifo is true (as set
72	+	* by the ForkJoinPool).  This allows use in message-passing
73	+	* frameworks in which tasks are never joined.
74	+	*
75		* Efficient implementation of this approach currently relies on
76		* an uncomfortable amount of "Unsafe" mechanics. To maintain
77		* correct orderings, reads and writes of variable base require
#	Line 265 | Line 272 | public class ForkJoinWorkerThread extend
272		final boolean shutdownNow()   { return transitionRunStateTo(TERMINATING); }
273
274		/**
275	<	* Transitions to at least the given state.  Returns true if not
276	<	* already at least at given state.
275	>	* Transitions to at least the given state.
276	>	*
277	>	* @return {@code true} if not already at least at given state
278		*/
279		private boolean transitionRunStateTo(int state) {
280		for (;;) {
#	Line 308 | Line 316 | public class ForkJoinWorkerThread extend
316		* one.  Marsaglia xor-shift is cheap and works well.
317		*/
318		private static int xorShift(int r) {
319	<	r ^= r << 1;
320	<	r ^= r >>> 3;
321	<	r ^= r << 10;
314	<	return r;
319	>	r ^= (r << 13);
320	>	r ^= (r >>> 17);
321	>	return r ^ (r << 5);
322		}
323
324		// Lifecycle methods
#	Line 372 | Line 379 | public class ForkJoinWorkerThread extend
379		* {@code super.onTermination} at the end of the overridden method.
380		*
381		* @param exception the exception causing this thread to abort due
382	<	* to an unrecoverable error, or null if completed normally
382	>	* to an unrecoverable error, or {@code null} if completed normally
383		*/
384		protected void onTermination(Throwable exception) {
385		// Execute remaining local tasks unless aborting or terminating
#	Line 469 | Line 476 | public class ForkJoinWorkerThread extend
476		}
477
478		/**
479	+	* Tries to take a task from the base of own queue, activating if
480	+	* necessary, failing only if empty. Called only by current thread.
481	+	*
482	+	* @return a task, or null if none
483	+	*/
484	+	final ForkJoinTask<?> locallyDeqTask() {
485	+	int b;
486	+	while (sp != (b = base)) {
487	+	if (tryActivate()) {
488	+	ForkJoinTask<?>[] q = queue;
489	+	int i = (q.length - 1) & b;
490	+	ForkJoinTask<?> t = q[i];
491	+	if (t != null && casSlotNull(q, i, t)) {
492	+	base = b + 1;
493	+	return t;
494	+	}
495	+	}
496	+	}
497	+	return null;
498	+	}
499	+
500	+	/**
501		* Returns a popped task, or null if empty. Ensures active status
502		* if non-null. Called only by current thread.
503		*/
#	Line 508 | Line 537 | public class ForkJoinWorkerThread extend
537		}
538
539		/**
540	<	* Returns next task.
540	>	* Returns next task or null if empty or contended
541		*/
542		final ForkJoinTask<?> peekTask() {
543		ForkJoinTask<?>[] q = queue;
#	Line 599 | Line 628 | public class ForkJoinWorkerThread extend
628		* @return a task, if available
629		*/
630		final ForkJoinTask<?> pollTask() {
631	<	ForkJoinTask<?> t = locallyFifo ? deqTask() : popTask();
631	>	ForkJoinTask<?> t = locallyFifo ? locallyDeqTask() : popTask();
632		if (t == null && (t = scan()) != null)
633		++stealCount;
634		return t;
#	Line 611 | Line 640 | public class ForkJoinWorkerThread extend
640		* @return a task, if available
641		*/
642		final ForkJoinTask<?> pollLocalTask() {
643	<	return locallyFifo ? deqTask() : popTask();
643	>	return locallyFifo ? locallyDeqTask() : popTask();
644		}
645
646		/**
#	Line 646 | Line 675 | public class ForkJoinWorkerThread extend
675		*
676		* @return the number of tasks drained
677		*/
678	<	final int drainTasksTo(Collection<ForkJoinTask<?>> c) {
678	>	final int drainTasksTo(Collection<? super ForkJoinTask<?>> c) {
679		int n = 0;
680		ForkJoinTask<?> t;
681		while (base != sp && (t = deqTask()) != null) {
#	Line 667 | Line 696 | public class ForkJoinWorkerThread extend
696		}
697
698		/**
699	<	* Returns true if at least one worker in the given array appears
700	<	* to have at least one queued task.
699	>	* Returns {@code true} if at least one worker in the given array
700	>	* appears to have at least one queued task.
701		*
702		* @param ws array of workers
703		*/
#	Line 731 | Line 760 | public class ForkJoinWorkerThread extend
760		do {} while (!tryActivate()); // re-activate on exit
761		}
762
763	<	// Unsafe mechanics for jsr166y 3rd party package.
763	>	// Unsafe mechanics
764	>
765	>	private static final sun.misc.Unsafe UNSAFE = getUnsafe();
766	>	private static final long spOffset =
767	>	objectFieldOffset("sp", ForkJoinWorkerThread.class);
768	>	private static final long runStateOffset =
769	>	objectFieldOffset("runState", ForkJoinWorkerThread.class);
770	>	private static final long qBase;
771	>	private static final int qShift;
772	>
773	>	static {
774	>	qBase = UNSAFE.arrayBaseOffset(ForkJoinTask[].class);
775	>	int s = UNSAFE.arrayIndexScale(ForkJoinTask[].class);
776	>	if ((s & (s-1)) != 0)
777	>	throw new Error("data type scale not a power of two");
778	>	qShift = 31 - Integer.numberOfLeadingZeros(s);
779	>	}
780	>
781	>	private static long objectFieldOffset(String field, Class<?> klazz) {
782	>	try {
783	>	return UNSAFE.objectFieldOffset(klazz.getDeclaredField(field));
784	>	} catch (NoSuchFieldException e) {
785	>	// Convert Exception to corresponding Error
786	>	NoSuchFieldError error = new NoSuchFieldError(field);
787	>	error.initCause(e);
788	>	throw error;
789	>	}
790	>	}
791	>
792	>	/**
793	>	* Returns a sun.misc.Unsafe.  Suitable for use in a 3rd party package.
794	>	* Replace with a simple call to Unsafe.getUnsafe when integrating
795	>	* into a jdk.
796	>	*
797	>	* @return a sun.misc.Unsafe
798	>	*/
799		private static sun.misc.Unsafe getUnsafe() {
800		try {
801		return sun.misc.Unsafe.getUnsafe();
802		} catch (SecurityException se) {
803		try {
804		return java.security.AccessController.doPrivileged
805	<	(new java.security.PrivilegedExceptionAction<sun.misc.Unsafe>() {
805	>	(new java.security
806	>	.PrivilegedExceptionAction<sun.misc.Unsafe>() {
807		public sun.misc.Unsafe run() throws Exception {
808	<	return getUnsafeByReflection();
808	>	java.lang.reflect.Field f = sun.misc
809	>	.Unsafe.class.getDeclaredField("theUnsafe");
810	>	f.setAccessible(true);
811	>	return (sun.misc.Unsafe) f.get(null);
812		}});
813		} catch (java.security.PrivilegedActionException e) {
814		throw new RuntimeException("Could not initialize intrinsics",
#	Line 748 | Line 816 | public class ForkJoinWorkerThread extend
816		}
817		}
818		}
751	–
752	–	private static sun.misc.Unsafe getUnsafeByReflection()
753	–	throws NoSuchFieldException, IllegalAccessException {
754	–	java.lang.reflect.Field f =
755	–	sun.misc.Unsafe.class.getDeclaredField("theUnsafe");
756	–	f.setAccessible(true);
757	–	return (sun.misc.Unsafe) f.get(null);
758	–	}
759	–
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	–	private static final sun.misc.Unsafe UNSAFE = getUnsafe();
772	–	static final long baseOffset =
773	–	fieldOffset("base", ForkJoinWorkerThread.class);
774	–	static final long spOffset =
775	–	fieldOffset("sp", ForkJoinWorkerThread.class);
776	–	static final long runStateOffset =
777	–	fieldOffset("runState", ForkJoinWorkerThread.class);
778	–	static final long qBase;
779	–	static final int qShift;
780	–
781	–	static {
782	–	qBase = UNSAFE.arrayBaseOffset(ForkJoinTask[].class);
783	–	int s = UNSAFE.arrayIndexScale(ForkJoinTask[].class);
784	–	if ((s & (s-1)) != 0)
785	–	throw new Error("data type scale not a power of two");
786	–	qShift = 31 - Integer.numberOfLeadingZeros(s);
787	–	}
819		}

Diff Legend

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