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.12 by jsr166, Tue Jul 21 18:11:44 2009 UTC

#	Line 17 | Line 17 | import java.lang.reflect.*;
17		* subclassable solely for the sake of adding functionality -- there
18		* are no overridable methods dealing with scheduling or
19		* execution. However, you can override initialization and termination
20	<	* cleanup methods surrounding the main task processing loop.  If you
21	<	* do create such a subclass, you will also need to supply a custom
20	>	* methods surrounding the main task processing loop.  If you do
21	>	* create such a subclass, you will also need to supply a custom
22		* ForkJoinWorkerThreadFactory to use it in a ForkJoinPool.
23		*
24		*/
#	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 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 active; 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.
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 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);
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 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 655 | Line 705 | public class ForkJoinWorkerThread extend
705		}
706
707		/**
708	<	* Scan, returning early if joinMe done
708	>	* Scans, returning early if joinMe done
709		*/
710		final ForkJoinTask<?> scanWhileJoining(ForkJoinTask<?> joinMe) {
711		ForkJoinTask<?> t = pollTask();
#	Line 667 | Line 717 | public class ForkJoinWorkerThread extend
717		}
718
719		/**
720	<	* Runs tasks until pool isQuiescent
720	>	* Runs tasks until pool isQuiescent.
721		*/
722		final void helpQuiescePool() {
723		for (;;) {
#	Line 706 | Line 756 | public class ForkJoinWorkerThread extend
756
757		private static long fieldOffset(String fieldName)
758		throws NoSuchFieldException {
759	<	return _unsafe.objectFieldOffset
759	>	return UNSAFE.objectFieldOffset
760		(ForkJoinWorkerThread.class.getDeclaredField(fieldName));
761		}
762
763	<	static final Unsafe _unsafe;
763	>	static final Unsafe UNSAFE;
764		static final long baseOffset;
765		static final long spOffset;
766		static final long runStateOffset;
#	Line 718 | Line 768 | public class ForkJoinWorkerThread extend
768		static final int qShift;
769		static {
770		try {
771	<	_unsafe = getUnsafe();
771	>	UNSAFE = getUnsafe();
772		baseOffset = fieldOffset("base");
773		spOffset = fieldOffset("sp");
774		runStateOffset = fieldOffset("runState");
775	<	qBase = _unsafe.arrayBaseOffset(ForkJoinTask[].class);
776	<	int s = _unsafe.arrayIndexScale(ForkJoinTask[].class);
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);

Diff Legend

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