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.9 by jsr166, Mon Jul 20 22:26:03 2009 UTC vs.
Revision 1.16 by jsr166, Thu Jul 23 23:07:57 2009 UTC

#	Line 21 | Line 21 | import java.lang.reflect.*;
21		* create such a subclass, you will also need to supply a custom
22		* ForkJoinWorkerThreadFactory to use it in a ForkJoinPool.
23		*
24	+	* @since 1.7
25	+	* @author Doug Lea
26		*/
27		public class ForkJoinWorkerThread extends Thread {
28		/*
#	Line 44 | Line 46 | public class ForkJoinWorkerThread extend
46		* of tasks. To accomplish this, we shift the CAS arbitrating pop
47		* vs deq (steal) from being on the indices ("base" and "sp") to
48		* the slots themselves (mainly via method "casSlotNull()"). So,
49	<	* both a successful pop and deq mainly entail CAS'ing a nonnull
49	>	* both a successful pop and deq mainly entail CAS'ing a non-null
50		* slot to null.  Because we rely on CASes of references, we do
51		* not need tag bits on base or sp.  They are simple ints as used
52		* in any circular array-based queue (see for example ArrayDeque).
#	Line 137 | Line 139 | public class ForkJoinWorkerThread extend
139		private static final int MAXIMUM_QUEUE_CAPACITY = 1 << 28;
140
141		/**
142	<	* The pool this thread works in. Accessed directly by ForkJoinTask
142	>	* The pool this thread works in. Accessed directly by ForkJoinTask.
143		*/
144		final ForkJoinPool pool;
145
#	Line 165 | Line 167 | public class ForkJoinWorkerThread extend
167		* Activity status. When true, this worker is considered active.
168		* Must be false upon construction. It must be true when executing
169		* tasks, and BEFORE stealing a task. It must be false before
170	<	* calling pool.sync
170	>	* calling pool.sync.
171		*/
172		private boolean active;
173
#	Line 188 | Line 190 | public class ForkJoinWorkerThread extend
190
191		/**
192		* Index of this worker in pool array. Set once by pool before
193	<	* running, and accessed directly by pool during cleanup etc
193	>	* running, and accessed directly by pool during cleanup etc.
194		*/
195		int poolIndex;
196
#	Line 205 | Line 207 | public class ForkJoinWorkerThread extend
207
208		/**
209		* Creates a ForkJoinWorkerThread operating in the given pool.
210	+	*
211		* @param pool the pool this thread works in
212		* @throws NullPointerException if pool is null
213		*/
#	Line 218 | Line 221 | public class ForkJoinWorkerThread extend
221		// Public access methods
222
223		/**
224	<	* Returns the pool hosting this thread
224	>	* Returns the pool hosting this thread.
225	>	*
226		* @return the pool
227		*/
228		public ForkJoinPool getPool() {
#	Line 231 | Line 235 | public class ForkJoinWorkerThread extend
235		* threads (minus one) that have ever been created in the pool.
236		* This method may be useful for applications that track status or
237		* collect results per-worker rather than per-task.
238	<	* @return the index number.
238	>	*
239	>	* @return the index number
240		*/
241		public int getPoolIndex() {
242		return poolIndex;
#	Line 240 | Line 245 | public class ForkJoinWorkerThread extend
245		/**
246		* Establishes local first-in-first-out scheduling mode for forked
247		* tasks that are never joined.
248	+	*
249		* @param async if true, use locally FIFO scheduling
250		*/
251		void setAsyncMode(boolean async) {
#	Line 261 | Line 267 | public class ForkJoinWorkerThread extend
267		final boolean shutdownNow()   { return transitionRunStateTo(TERMINATING); }
268
269		/**
270	<	* Transition to at least the given state. Return true if not
271	<	* already at least given state.
270	>	* Transitions to at least the given state.  Returns true if not
271	>	* already at least at given state.
272		*/
273		private boolean transitionRunStateTo(int state) {
274		for (;;) {
275		int s = runState;
276		if (s >= state)
277		return false;
278	<	if (_unsafe.compareAndSwapInt(this, runStateOffset, s, state))
278	>	if (UNSAFE.compareAndSwapInt(this, runStateOffset, s, state))
279		return true;
280		}
281		}
282
283		/**
284	<	* Try to set status to active; fail on contention
284	>	* Tries to set status to active; fails on contention.
285		*/
286		private boolean tryActivate() {
287		if (!active) {
#	Line 287 | Line 293 | public class ForkJoinWorkerThread extend
293		}
294
295		/**
296	<	* Try to set status to active; fail on contention
296	>	* Tries to set status to inactive; fails on contention.
297		*/
298		private boolean tryInactivate() {
299		if (active) {
#	Line 299 | Line 305 | public class ForkJoinWorkerThread extend
305		}
306
307		/**
308	<	* Computes next value for random victim probe. Scans don't
308	>	* Computes next value for random victim probe.  Scans don't
309		* require a very high quality generator, but also not a crummy
310	<	* one. Marsaglia xor-shift is cheap and works well.
310	>	* one.  Marsaglia xor-shift is cheap and works well.
311		*/
312		private static int xorShift(int r) {
313		r ^= r << 1;
#	Line 331 | Line 337 | public class ForkJoinWorkerThread extend
337		}
338
339		/**
340	<	* Execute tasks until shut down.
340	>	* Executes tasks until shut down.
341		*/
342		private void mainLoop() {
343		while (!isShutdown()) {
#	Line 363 | Line 369 | public class ForkJoinWorkerThread extend
369		}
370
371		/**
372	<	* Perform cleanup associated with termination of this worker
372	>	* Performs cleanup associated with termination of this worker
373		* thread.  If you override this method, you must invoke
374	<	* super.onTermination at the end of the overridden method.
374	>	* {@code super.onTermination} at the end of the overridden method.
375		*
376		* @param exception the exception causing this thread to abort due
377	<	* to an unrecoverable error, or null if completed normally.
377	>	* to an unrecoverable error, or null if completed normally
378		*/
379		protected void onTermination(Throwable exception) {
380		// Execute remaining local tasks unless aborting or terminating
#	Line 377 | Line 383 | public class ForkJoinWorkerThread extend
383		ForkJoinTask<?> t = popTask();
384		if (t != null)
385		t.quietlyExec();
386	<	} catch(Throwable ex) {
386	>	} catch (Throwable ex) {
387		exception = ex;
388		}
389		}
390		// Cancel other tasks, transition status, notify pool, and
391		// propagate exception to uncaught exception handler
392		try {
393	<	do;while (!tryInactivate()); // ensure inactive
393	>	do {} while (!tryInactivate()); // ensure inactive
394		cancelTasks();
395		runState = TERMINATED;
396		pool.workerTerminated(this);
#	Line 400 | Line 406 | public class ForkJoinWorkerThread extend
406		// Intrinsics-based support for queue operations.
407
408		/**
409	<	* Add in store-order the given task at given slot of q to
410	<	* null. Caller must ensure q is nonnull and index is in range.
409	>	* Adds in store-order the given task at given slot of q to null.
410	>	* Caller must ensure q is non-null and index is in range.
411		*/
412		private static void setSlot(ForkJoinTask<?>[] q, int i,
413	<	ForkJoinTask<?> t){
414	<	_unsafe.putOrderedObject(q, (i << qShift) + qBase, t);
413	>	ForkJoinTask<?> t) {
414	>	UNSAFE.putOrderedObject(q, (i << qShift) + qBase, t);
415		}
416
417		/**
418	<	* CAS given slot of q to null. Caller must ensure q is nonnull
418	>	* CAS given slot of q to null. Caller must ensure q is non-null
419		* and index is in range.
420		*/
421		private static boolean casSlotNull(ForkJoinTask<?>[] q, int i,
422		ForkJoinTask<?> t) {
423	<	return _unsafe.compareAndSwapObject(q, (i << qShift) + qBase, t, null);
423	>	return UNSAFE.compareAndSwapObject(q, (i << qShift) + qBase, t, null);
424		}
425
426		/**
427		* Sets sp in store-order.
428		*/
429		private void storeSp(int s) {
430	<	_unsafe.putOrderedInt(this, spOffset, s);
430	>	UNSAFE.putOrderedInt(this, spOffset, s);
431		}
432
433		// Main queue methods
434
435		/**
436		* Pushes a task. Called only by current thread.
437	<	* @param t the task. Caller must ensure nonnull
437	>	*
438	>	* @param t the task. Caller must ensure non-null.
439		*/
440		final void pushTask(ForkJoinTask<?> t) {
441		ForkJoinTask<?>[] q = queue;
#	Line 445 | Line 452 | public class ForkJoinWorkerThread extend
452		/**
453		* Tries to take a task from the base of the queue, failing if
454		* either empty or contended.
455	<	* @return a task, or null if none or contended.
455	>	*
456	>	* @return a task, or null if none or contended
457		*/
458		final ForkJoinTask<?> deqTask() {
459		ForkJoinTask<?> t;
#	Line 464 | Line 472 | public class ForkJoinWorkerThread extend
472
473		/**
474		* Returns a popped task, or null if empty. Ensures active status
475	<	* if nonnull. Called only by current thread.
475	>	* if non-null. Called only by current thread.
476		*/
477		final ForkJoinTask<?> popTask() {
478		int s = sp;
#	Line 487 | Line 495 | public class ForkJoinWorkerThread extend
495		* Specialized version of popTask to pop only if
496		* topmost element is the given task. Called only
497		* by current thread while active.
498	<	* @param t the task. Caller must ensure nonnull
498	>	*
499	>	* @param t the task. Caller must ensure non-null.
500		*/
501		final boolean unpushTask(ForkJoinTask<?> t) {
502		ForkJoinTask<?>[] q = queue;
#	Line 508 | Line 517 | public class ForkJoinWorkerThread extend
517		if (q == null)
518		return null;
519		int mask = q.length - 1;
520	<	int i = locallyFifo? base : (sp - 1);
520	>	int i = locallyFifo ? base : (sp - 1);
521		return q[i & mask];
522		}
523
#	Line 571 | Line 580 | public class ForkJoinWorkerThread extend
580		ForkJoinWorkerThread v = ws[mask & idx];
581		if (v == null || v.sp == v.base) {
582		if (probes <= mask)
583	<	idx = (probes++ < 0)? r : (idx + 1);
583	>	idx = (probes++ < 0) ? r : (idx + 1);
584		else
585		break;
586		}
#	Line 587 | Line 596 | public class ForkJoinWorkerThread extend
596		}
597
598		/**
599	<	* gets and removes a local or stolen a task
599	>	* Gets and removes a local or stolen task.
600	>	*
601		* @return a task, if available
602		*/
603		final ForkJoinTask<?> pollTask() {
604	<	ForkJoinTask<?> t = locallyFifo? deqTask() : popTask();
604	>	ForkJoinTask<?> t = locallyFifo ? deqTask() : popTask();
605		if (t == null && (t = scan()) != null)
606		++stealCount;
607		return t;
608		}
609
610		/**
611	<	* gets a local task
611	>	* Gets a local task.
612	>	*
613		* @return a task, if available
614		*/
615		final ForkJoinTask<?> pollLocalTask() {
616	<	return locallyFifo? deqTask() : popTask();
616	>	return locallyFifo ? deqTask() : popTask();
617		}
618
619		/**
620		* Returns a pool submission, if one exists, activating first.
621	+	*
622		* @return a submission, if available
623		*/
624		private ForkJoinTask<?> pollSubmission() {
#	Line 632 | Line 644 | public class ForkJoinWorkerThread extend
644		}
645
646		/**
647	<	* Drains tasks to given collection c
647	>	* Drains tasks to given collection c.
648	>	*
649		* @return the number of tasks drained
650		*/
651		final int drainTasksTo(Collection<ForkJoinTask<?>> c) {
#	Line 646 | Line 659 | public class ForkJoinWorkerThread extend
659		}
660
661		/**
662	<	* Get and clear steal count for accumulation by pool.  Called
662	>	* Gets and clears steal count for accumulation by pool.  Called
663		* only when known to be idle (in pool.sync and termination).
664		*/
665		final int getAndClearStealCount() {
#	Line 658 | Line 671 | public class ForkJoinWorkerThread extend
671		/**
672		* Returns true if at least one worker in the given array appears
673		* to have at least one queued task.
674	+	*
675		* @param ws array of workers
676		*/
677		static boolean hasQueuedTasks(ForkJoinWorkerThread[] ws) {
#	Line 680 | Line 694 | public class ForkJoinWorkerThread extend
694		* Returns an estimate of the number of tasks in the queue.
695		*/
696		final int getQueueSize() {
697	<	int n = sp - base;
698	<	return n < 0? 0 : n; // suppress momentarily negative values
697	>	// suppress momentarily negative values
698	>	return Math.max(0, sp - base);
699		}
700
701		/**
#	Line 694 | Line 708 | public class ForkJoinWorkerThread extend
708		}
709
710		/**
711	<	* Scan, returning early if joinMe done
711	>	* Scans, returning early if joinMe done.
712		*/
713		final ForkJoinTask<?> scanWhileJoining(ForkJoinTask<?> joinMe) {
714		ForkJoinTask<?> t = pollTask();
#	Line 706 | Line 720 | public class ForkJoinWorkerThread extend
720		}
721
722		/**
723	<	* Runs tasks until pool isQuiescent
723	>	* Runs tasks until {@code pool.isQuiescent()}.
724		*/
725		final void helpQuiescePool() {
726		for (;;) {
#	Line 716 | Line 730 | public class ForkJoinWorkerThread extend
730		else if (tryInactivate() && pool.isQuiescent())
731		break;
732		}
733	<	do;while (!tryActivate()); // re-activate on exit
733	>	do {} while (!tryActivate()); // re-activate on exit
734		}
735
736		// Temporary Unsafe mechanics for preliminary release
#	Line 745 | Line 759 | public class ForkJoinWorkerThread extend
759
760		private static long fieldOffset(String fieldName)
761		throws NoSuchFieldException {
762	<	return _unsafe.objectFieldOffset
762	>	return UNSAFE.objectFieldOffset
763		(ForkJoinWorkerThread.class.getDeclaredField(fieldName));
764		}
765
766	<	static final Unsafe _unsafe;
766	>	static final Unsafe UNSAFE;
767		static final long baseOffset;
768		static final long spOffset;
769		static final long runStateOffset;
#	Line 757 | Line 771 | public class ForkJoinWorkerThread extend
771		static final int qShift;
772		static {
773		try {
774	<	_unsafe = getUnsafe();
774	>	UNSAFE = getUnsafe();
775		baseOffset = fieldOffset("base");
776		spOffset = fieldOffset("sp");
777		runStateOffset = fieldOffset("runState");
778	<	qBase = _unsafe.arrayBaseOffset(ForkJoinTask[].class);
779	<	int s = _unsafe.arrayIndexScale(ForkJoinTask[].class);
778	>	qBase = UNSAFE.arrayBaseOffset(ForkJoinTask[].class);
779	>	int s = UNSAFE.arrayIndexScale(ForkJoinTask[].class);
780		if ((s & (s-1)) != 0)
781		throw new Error("data type scale not a power of two");
782		qShift = 31 - Integer.numberOfLeadingZeros(s);

Diff Legend

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