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.10 by jsr166, Mon Jul 20 23:07:43 2009 UTC vs.
Revision 1.23 by dl, Fri Jul 31 16:27:08 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 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 137 | Line 143 | public class ForkJoinWorkerThread extend
143		private static final int MAXIMUM_QUEUE_CAPACITY = 1 << 28;
144
145		/**
146	<	* The pool this thread works in. Accessed directly by ForkJoinTask
146	>	* The pool this thread works in. Accessed directly by ForkJoinTask.
147		*/
148		final ForkJoinPool pool;
149
#	Line 165 | Line 171 | public class ForkJoinWorkerThread extend
171		* Activity status. When true, this worker is considered active.
172		* Must be false upon construction. It must be true when executing
173		* tasks, and BEFORE stealing a task. It must be false before
174	<	* calling pool.sync
174	>	* calling pool.sync.
175		*/
176		private boolean active;
177
#	Line 188 | Line 194 | public class ForkJoinWorkerThread extend
194
195		/**
196		* Index of this worker in pool array. Set once by pool before
197	<	* running, and accessed directly by pool during cleanup etc
197	>	* running, and accessed directly by pool during cleanup etc.
198		*/
199		int poolIndex;
200
#	Line 205 | Line 211 | public class ForkJoinWorkerThread extend
211
212		/**
213		* Creates a ForkJoinWorkerThread operating in the given pool.
214	+	*
215		* @param pool the pool this thread works in
216		* @throws NullPointerException if pool is null
217		*/
#	Line 218 | Line 225 | public class ForkJoinWorkerThread extend
225		// Public access methods
226
227		/**
228	<	* Returns the pool hosting this thread
228	>	* Returns the pool hosting this thread.
229	>	*
230		* @return the pool
231		*/
232		public ForkJoinPool getPool() {
#	Line 231 | Line 239 | public class ForkJoinWorkerThread extend
239		* threads (minus one) that have ever been created in the pool.
240		* This method may be useful for applications that track status or
241		* collect results per-worker rather than per-task.
242	<	* @return the index number.
242	>	*
243	>	* @return the index number
244		*/
245		public int getPoolIndex() {
246		return poolIndex;
#	Line 240 | Line 249 | public class ForkJoinWorkerThread extend
249		/**
250		* Establishes local first-in-first-out scheduling mode for forked
251		* tasks that are never joined.
252	+	*
253		* @param async if true, use locally FIFO scheduling
254		*/
255		void setAsyncMode(boolean async) {
#	Line 261 | Line 271 | public class ForkJoinWorkerThread extend
271		final boolean shutdownNow()   { return transitionRunStateTo(TERMINATING); }
272
273		/**
274	<	* Transition to at least the given state. Return true if not
275	<	* already at least 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 (;;) {
280		int s = runState;
281		if (s >= state)
282		return false;
283	<	if (_unsafe.compareAndSwapInt(this, runStateOffset, s, state))
283	>	if (UNSAFE.compareAndSwapInt(this, runStateOffset, s, state))
284		return true;
285		}
286		}
287
288		/**
289	<	* Try to set status to active; fail on contention
289	>	* Tries to set status to active; fails on contention.
290		*/
291		private boolean tryActivate() {
292		if (!active) {
#	Line 287 | Line 298 | public class ForkJoinWorkerThread extend
298		}
299
300		/**
301	<	* Try to set status to active; fail on contention
301	>	* Tries to set status to inactive; fails on contention.
302		*/
303		private boolean tryInactivate() {
304		if (active) {
#	Line 299 | Line 310 | public class ForkJoinWorkerThread extend
310		}
311
312		/**
313	<	* Computes next value for random victim probe. Scans don't
313	>	* Computes next value for random victim probe.  Scans don't
314		* require a very high quality generator, but also not a crummy
315	<	* one. Marsaglia xor-shift is cheap and works well.
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;
310	<	return r;
318	>	r ^= (r << 13);
319	>	r ^= (r >>> 17);
320	>	return r ^ (r << 5);
321		}
322
323		// Lifecycle methods
#	Line 331 | Line 341 | public class ForkJoinWorkerThread extend
341		}
342
343		/**
344	<	* Execute tasks until shut down.
344	>	* Executes tasks until shut down.
345		*/
346		private void mainLoop() {
347		while (!isShutdown()) {
#	Line 363 | Line 373 | public class ForkJoinWorkerThread extend
373		}
374
375		/**
376	<	* Perform cleanup associated with termination of this worker
376	>	* Performs cleanup associated with termination of this worker
377		* thread.  If you override this method, you must invoke
378	<	* super.onTermination at the end of the overridden method.
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 377 | Line 387 | public class ForkJoinWorkerThread extend
387		ForkJoinTask<?> t = popTask();
388		if (t != null)
389		t.quietlyExec();
390	<	} catch(Throwable ex) {
390	>	} catch (Throwable ex) {
391		exception = ex;
392		}
393		}
394		// Cancel other tasks, transition status, notify pool, and
395		// propagate exception to uncaught exception handler
396		try {
397	<	do;while (!tryInactivate()); // ensure inactive
397	>	do {} while (!tryInactivate()); // ensure inactive
398		cancelTasks();
399		runState = TERMINATED;
400		pool.workerTerminated(this);
#	Line 404 | Line 414 | public class ForkJoinWorkerThread extend
414		* Caller must ensure q is non-null and index is in range.
415		*/
416		private static void setSlot(ForkJoinTask<?>[] q, int i,
417	<	ForkJoinTask<?> t){
418	<	_unsafe.putOrderedObject(q, (i << qShift) + qBase, t);
417	>	ForkJoinTask<?> t) {
418	>	UNSAFE.putOrderedObject(q, (i << qShift) + qBase, t);
419		}
420
421		/**
#	Line 414 | Line 424 | public class ForkJoinWorkerThread extend
424		*/
425		private static boolean casSlotNull(ForkJoinTask<?>[] q, int i,
426		ForkJoinTask<?> t) {
427	<	return _unsafe.compareAndSwapObject(q, (i << qShift) + qBase, t, null);
427	>	return UNSAFE.compareAndSwapObject(q, (i << qShift) + qBase, t, null);
428		}
429
430		/**
431		* Sets sp in store-order.
432		*/
433		private void storeSp(int s) {
434	<	_unsafe.putOrderedInt(this, spOffset, s);
434	>	UNSAFE.putOrderedInt(this, spOffset, s);
435		}
436
437		// Main queue methods
438
439		/**
440		* Pushes a task. Called only by current thread.
441	+	*
442		* @param t the task. Caller must ensure non-null.
443		*/
444		final void pushTask(ForkJoinTask<?> t) {
#	Line 445 | Line 456 | public class ForkJoinWorkerThread extend
456		/**
457		* Tries to take a task from the base of the queue, failing if
458		* either empty or contended.
459	<	* @return a task, or null if none or contended.
459	>	*
460	>	* @return a task, or null if none or contended
461		*/
462		final ForkJoinTask<?> deqTask() {
463		ForkJoinTask<?> t;
#	Line 463 | 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 487 | Line 521 | public class ForkJoinWorkerThread extend
521		* Specialized version of popTask to pop only if
522		* topmost element is the given task. Called only
523		* by current thread while active.
524	<	* @param t the task. Caller must ensure non-null
524	>	*
525	>	* @param t the task. Caller must ensure non-null.
526		*/
527		final boolean unpushTask(ForkJoinTask<?> t) {
528		ForkJoinTask<?>[] q = queue;
#	Line 501 | 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;
543		if (q == null)
544		return null;
545		int mask = q.length - 1;
546	<	int i = locallyFifo? base : (sp - 1);
546	>	int i = locallyFifo ? base : (sp - 1);
547		return q[i & mask];
548		}
549
#	Line 571 | Line 606 | public class ForkJoinWorkerThread extend
606		ForkJoinWorkerThread v = ws[mask & idx];
607		if (v == null || v.sp == v.base) {
608		if (probes <= mask)
609	<	idx = (probes++ < 0)? r : (idx + 1);
609	>	idx = (probes++ < 0) ? r : (idx + 1);
610		else
611		break;
612		}
#	Line 587 | Line 622 | public class ForkJoinWorkerThread extend
622		}
623
624		/**
625	<	* gets and removes a local or stolen a task
625	>	* Gets and removes a local or stolen task.
626	>	*
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;
634		}
635
636		/**
637	<	* gets a local task
637	>	* Gets a local task.
638	>	*
639		* @return a task, if available
640		*/
641		final ForkJoinTask<?> pollLocalTask() {
642	<	return locallyFifo? deqTask() : popTask();
642	>	return locallyFifo ? locallyDeqTask() : popTask();
643		}
644
645		/**
646		* Returns a pool submission, if one exists, activating first.
647	+	*
648		* @return a submission, if available
649		*/
650		private ForkJoinTask<?> pollSubmission() {
#	Line 632 | Line 670 | public class ForkJoinWorkerThread extend
670		}
671
672		/**
673	<	* Drains tasks to given collection c
673	>	* Drains tasks to given collection c.
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 646 | Line 685 | public class ForkJoinWorkerThread extend
685		}
686
687		/**
688	<	* Get and clear steal count for accumulation by pool.  Called
688	>	* Gets and clears steal count for accumulation by pool.  Called
689		* only when known to be idle (in pool.sync and termination).
690		*/
691		final int getAndClearStealCount() {
#	Line 656 | 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		*/
703		static boolean hasQueuedTasks(ForkJoinWorkerThread[] ws) {
#	Line 680 | Line 720 | public class ForkJoinWorkerThread extend
720		* Returns an estimate of the number of tasks in the queue.
721		*/
722		final int getQueueSize() {
723	<	int n = sp - base;
724	<	return n < 0? 0 : n; // suppress momentarily negative values
723	>	// suppress momentarily negative values
724	>	return Math.max(0, sp - base);
725		}
726
727		/**
#	Line 694 | Line 734 | public class ForkJoinWorkerThread extend
734		}
735
736		/**
737	<	* Scan, returning early if joinMe done
737	>	* Scans, returning early if joinMe done.
738		*/
739		final ForkJoinTask<?> scanWhileJoining(ForkJoinTask<?> joinMe) {
740		ForkJoinTask<?> t = pollTask();
#	Line 706 | Line 746 | public class ForkJoinWorkerThread extend
746		}
747
748		/**
749	<	* Runs tasks until pool isQuiescent
749	>	* Runs tasks until {@code pool.isQuiescent()}.
750		*/
751		final void helpQuiescePool() {
752		for (;;) {
#	Line 716 | Line 756 | public class ForkJoinWorkerThread extend
756		else if (tryInactivate() && pool.isQuiescent())
757		break;
758		}
759	<	do;while (!tryActivate()); // re-activate on exit
759	>	do {} while (!tryActivate()); // re-activate on exit
760		}
761
762	<	// Temporary Unsafe mechanics for preliminary release
763	<	private static Unsafe getUnsafe() throws Throwable {
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.getUnsafe();
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<Unsafe>() {
805	<	public Unsafe run() throws Exception {
806	<	return getUnsafePrivileged();
804	>	(new java.security
805	>	.PrivilegedExceptionAction<sun.misc.Unsafe>() {
806	>	public sun.misc.Unsafe run() throws Exception {
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 e.getCause();
813	>	throw new RuntimeException("Could not initialize intrinsics",
814	>	e.getCause());
815		}
816		}
817		}
738	–
739	–	private static Unsafe getUnsafePrivileged()
740	–	throws NoSuchFieldException, IllegalAccessException {
741	–	Field f = Unsafe.class.getDeclaredField("theUnsafe");
742	–	f.setAccessible(true);
743	–	return (Unsafe) f.get(null);
744	–	}
745	–
746	–	private static long fieldOffset(String fieldName)
747	–	throws NoSuchFieldException {
748	–	return _unsafe.objectFieldOffset
749	–	(ForkJoinWorkerThread.class.getDeclaredField(fieldName));
750	–	}
751	–
752	–	static final Unsafe _unsafe;
753	–	static final long baseOffset;
754	–	static final long spOffset;
755	–	static final long runStateOffset;
756	–	static final long qBase;
757	–	static final int qShift;
758	–	static {
759	–	try {
760	–	_unsafe = getUnsafe();
761	–	baseOffset = fieldOffset("base");
762	–	spOffset = fieldOffset("sp");
763	–	runStateOffset = fieldOffset("runState");
764	–	qBase = _unsafe.arrayBaseOffset(ForkJoinTask[].class);
765	–	int s = _unsafe.arrayIndexScale(ForkJoinTask[].class);
766	–	if ((s & (s-1)) != 0)
767	–	throw new Error("data type scale not a power of two");
768	–	qShift = 31 - Integer.numberOfLeadingZeros(s);
769	–	} catch (Throwable e) {
770	–	throw new RuntimeException("Could not initialize intrinsics", e);
771	–	}
772	–	}
818		}

Diff Legend

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