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.6 by jsr166, Thu Mar 19 05:10:42 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
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	<	* cleanup methods surrounding the main task processing loop.  If you
19	<	* do create such a subclass, you will also need to supply a custom
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.
21		*
22	+	* @since 1.7
23	+	* @author Doug Lea
24		*/
25		public class ForkJoinWorkerThread extends Thread {
26		/*
#	Line 44 | Line 44 | public class ForkJoinWorkerThread extend
44		* of tasks. To accomplish this, we shift the CAS arbitrating pop
45		* vs deq (steal) from being on the indices ("base" and "sp") to
46		* the slots themselves (mainly via method "casSlotNull()"). So,
47	<	* both a successful pop and deq mainly entail CAS'ing a nonnull
47	>	* both a successful pop and deq mainly entail CAS'ing a non-null
48		* slot to null.  Because we rely on CASes of references, we do
49		* not need tag bits on base or sp.  They are simple ints as used
50		* in any circular array-based queue (see for example ArrayDeque).
#	Line 56 | Line 56 | public class ForkJoinWorkerThread extend
56		* considered individually, is not wait-free. One thief cannot
57		* successfully continue until another in-progress one (or, if
58		* previously empty, a push) completes.  However, in the
59	<	* aggregate, we ensure at least probablistic non-blockingness. If
59	>	* aggregate, we ensure at least probabilistic non-blockingness. If
60		* an attempted steal fails, a thief always chooses a different
61		* random victim target to try next. So, in order for one thief to
62		* progress, it suffices for any in-progress deq or new push on
#	Line 75 | Line 75 | public class ForkJoinWorkerThread extend
75		* push) require store order and CASes (in pop and deq) require
76		* (volatile) CAS semantics. Since these combinations aren't
77		* supported using ordinary volatiles, the only way to accomplish
78	<	* these effciently is to use direct Unsafe calls. (Using external
78	>	* these efficiently is to use direct Unsafe calls. (Using external
79		* AtomicIntegers and AtomicReferenceArrays for the indices and
80		* array is significantly slower because of memory locality and
81		* indirection effects.) Further, performance on most platforms is
#	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 199 | Line 199 | public class ForkJoinWorkerThread extend
199		long lastEventCount;
200
201		/**
202	+	* True if use local fifo, not default lifo, for local polling
203	+	*/
204	+	private boolean locallyFifo;
205	+
206	+	/**
207		* Creates a ForkJoinWorkerThread operating in the given pool.
208	+	*
209		* @param pool the pool this thread works in
210		* @throws NullPointerException if pool is null
211		*/
#	Line 213 | Line 219 | public class ForkJoinWorkerThread extend
219		// Public access methods
220
221		/**
222	<	* Returns the pool hosting this thread
222	>	* Returns the pool hosting this thread.
223	>	*
224		* @return the pool
225		*/
226		public ForkJoinPool getPool() {
#	Line 226 | Line 233 | public class ForkJoinWorkerThread extend
233		* threads (minus one) that have ever been created in the pool.
234		* This method may be useful for applications that track status or
235		* collect results per-worker rather than per-task.
236	<	* @return the index number.
236	>	*
237	>	* @return the index number
238		*/
239		public int getPoolIndex() {
240		return poolIndex;
241		}
242
243	+	/**
244	+	* Establishes local first-in-first-out scheduling mode for forked
245	+	* tasks that are never joined.
246	+	*
247	+	* @param async if true, use locally FIFO scheduling
248	+	*/
249	+	void setAsyncMode(boolean async) {
250	+	locallyFifo = async;
251	+	}
252
253		// Runstate management
254
#	Line 248 | Line 265 | public class ForkJoinWorkerThread extend
265		final boolean shutdownNow()   { return transitionRunStateTo(TERMINATING); }
266
267		/**
268	<	* Transition to at least the given state. Return true if not
269	<	* already at least given state.
268	>	* Transitions to at least the given state.  Returns true if not
269	>	* already at least at given state.
270		*/
271		private boolean transitionRunStateTo(int state) {
272		for (;;) {
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		}
280
281		/**
282	<	* Try to set status to active; fail on contention
282	>	* Tries to set status to active; fails on contention.
283		*/
284		private boolean tryActivate() {
285		if (!active) {
#	Line 274 | Line 291 | public class ForkJoinWorkerThread extend
291		}
292
293		/**
294	<	* Try to set status to active; fail on contention
294	>	* Tries to set status to inactive; fails on contention.
295		*/
296		private boolean tryInactivate() {
297		if (active) {
#	Line 286 | Line 303 | public class ForkJoinWorkerThread extend
303		}
304
305		/**
306	<	* Computes next value for random victim probe. Scans don't
306	>	* Computes next value for random victim probe.  Scans don't
307		* require a very high quality generator, but also not a crummy
308	<	* one. Marsaglia xor-shift is cheap and works well.
308	>	* one.  Marsaglia xor-shift is cheap and works well.
309		*/
310		private static int xorShift(int r) {
311		r ^= r << 1;
#	Line 318 | Line 335 | public class ForkJoinWorkerThread extend
335		}
336
337		/**
338	<	* Execute tasks until shut down.
338	>	* Executes tasks until shut down.
339		*/
340		private void mainLoop() {
341		while (!isShutdown()) {
#	Line 350 | Line 367 | public class ForkJoinWorkerThread extend
367		}
368
369		/**
370	<	* Perform cleanup associated with termination of this worker
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.
375	>	* to an unrecoverable error, or null if completed normally
376		*/
377		protected void onTermination(Throwable exception) {
378		// Execute remaining local tasks unless aborting or terminating
#	Line 364 | 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 387 | Line 404 | public class ForkJoinWorkerThread extend
404		// Intrinsics-based support for queue operations.
405
406		/**
407	<	* Add in store-order the given task at given slot of q to
408	<	* null. Caller must ensure q is nonnull and index is in range.
407	>	* Adds in store-order the given task at given slot of q to null.
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		/**
416	<	* CAS given slot of q to null. Caller must ensure q is nonnull
416	>	* CAS given slot of q to null. Caller must ensure q is non-null
417		* and index is in range.
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
432
433		/**
434		* Pushes a task. Called only by current thread.
435	<	* @param t the task. Caller must ensure nonnull
435	>	*
436	>	* @param t the task. Caller must ensure non-null.
437		*/
438		final void pushTask(ForkJoinTask<?> t) {
439		ForkJoinTask<?>[] q = queue;
#	Line 432 | Line 450 | public class ForkJoinWorkerThread extend
450		/**
451		* Tries to take a task from the base of the queue, failing if
452		* either empty or contended.
453	<	* @return a task, or null if none or contended.
453	>	*
454	>	* @return a task, or null if none or contended
455		*/
456	<	private ForkJoinTask<?> deqTask() {
456	>	final ForkJoinTask<?> deqTask() {
457		ForkJoinTask<?> t;
458		ForkJoinTask<?>[] q;
459		int i;
#	Line 451 | Line 470 | public class ForkJoinWorkerThread extend
470
471		/**
472		* Returns a popped task, or null if empty. Ensures active status
473	<	* if nonnull. Called only by current thread.
473	>	* if non-null. Called only by current thread.
474		*/
475		final ForkJoinTask<?> popTask() {
476		int s = sp;
#	Line 474 | Line 493 | public class ForkJoinWorkerThread extend
493		* Specialized version of popTask to pop only if
494		* topmost element is the given task. Called only
495		* by current thread while active.
496	<	* @param t the task. Caller must ensure nonnull
496	>	*
497	>	* @param t the task. Caller must ensure non-null.
498		*/
499		final boolean unpushTask(ForkJoinTask<?> t) {
500		ForkJoinTask<?>[] q = queue;
#	Line 488 | Line 508 | public class ForkJoinWorkerThread extend
508		}
509
510		/**
511	<	* Returns next task to pop.
511	>	* Returns next task.
512		*/
513		final ForkJoinTask<?> peekTask() {
514		ForkJoinTask<?>[] q = queue;
515	<	return q == null? null : q[(sp - 1) & (q.length - 1)];
515	>	if (q == null)
516	>	return null;
517	>	int mask = q.length - 1;
518	>	int i = locallyFifo ? base : (sp - 1);
519	>	return q[i & mask];
520		}
521
522		/**
#	Line 554 | 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 570 | Line 594 | public class ForkJoinWorkerThread extend
594		}
595
596		/**
597	<	* Pops or steals a task
597	>	* Gets and removes a local or stolen task.
598	>	*
599		* @return a task, if available
600		*/
601		final ForkJoinTask<?> pollTask() {
602	<	ForkJoinTask<?> t = popTask();
602	>	ForkJoinTask<?> t = locallyFifo ? deqTask() : popTask();
603		if (t == null && (t = scan()) != null)
604		++stealCount;
605		return t;
606		}
607
608		/**
609	+	* Gets a local task.
610	+	*
611	+	* @return a task, if available
612	+	*/
613	+	final ForkJoinTask<?> pollLocalTask() {
614	+	return locallyFifo ? deqTask() : popTask();
615	+	}
616	+
617	+	/**
618		* Returns a pool submission, if one exists, activating first.
619	+	*
620		* @return a submission, if available
621		*/
622		private ForkJoinTask<?> pollSubmission() {
#	Line 607 | Line 642 | public class ForkJoinWorkerThread extend
642		}
643
644		/**
645	<	* Get and clear steal count for accumulation by pool.  Called
645	>	* Drains tasks to given collection c.
646	>	*
647	>	* @return the number of tasks drained
648	>	*/
649	>	final int drainTasksTo(Collection<ForkJoinTask<?>> c) {
650	>	int n = 0;
651	>	ForkJoinTask<?> t;
652	>	while (base != sp && (t = deqTask()) != null) {
653	>	c.add(t);
654	>	++n;
655	>	}
656	>	return n;
657	>	}
658	>
659	>	/**
660	>	* Gets and clears steal count for accumulation by pool.  Called
661		* only when known to be idle (in pool.sync and termination).
662		*/
663		final int getAndClearStealCount() {
#	Line 619 | 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 641 | 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 655 | Line 706 | public class ForkJoinWorkerThread extend
706		}
707
708		/**
709	<	* Scan, returning early if joinMe done
709	>	* Scans, returning early if joinMe done.
710		*/
711		final ForkJoinTask<?> scanWhileJoining(ForkJoinTask<?> joinMe) {
712		ForkJoinTask<?> t = pollTask();
#	Line 667 | 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 677 | 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");
725	<	qBase = _unsafe.arrayBaseOffset(ForkJoinTask[].class);
726	<	int s = _unsafe.arrayIndexScale(ForkJoinTask[].class);
727	<	if ((s & (s-1)) != 0)
728	<	throw new Error("data type scale not a power of two");
729	<	qShift = 31 - Integer.numberOfLeadingZeros(s);
730	<	} catch (Throwable e) {
731	<	throw new RuntimeException("Could not initialize intrinsics", e);
732	<	}
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