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

Diff Legend

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