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.15 by jsr166, Thu Jul 23 19:25:45 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 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 active; 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.
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 404 | Line 410 | public class ForkJoinWorkerThread extend
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		/**
#	Line 414 | Line 420 | public class ForkJoinWorkerThread extend
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	+	*
438		* @param t the task. Caller must ensure non-null.
439		*/
440		final void pushTask(ForkJoinTask<?> t) {
#	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 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 non-null
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 680 | Line 693 | public class ForkJoinWorkerThread extend
693		* Returns an estimate of the number of tasks in the queue.
694		*/
695		final int getQueueSize() {
696	<	int n = sp - base;
697	<	return n < 0? 0 : n; // suppress momentarily negative values
696	>	// suppress momentarily negative values
697	>	return Math.max(0, sp - base);
698		}
699
700		/**
#	Line 694 | Line 707 | public class ForkJoinWorkerThread extend
707		}
708
709		/**
710	<	* Scan, returning early if joinMe done
710	>	* Scans, returning early if joinMe done
711		*/
712		final ForkJoinTask<?> scanWhileJoining(ForkJoinTask<?> joinMe) {
713		ForkJoinTask<?> t = pollTask();
#	Line 706 | Line 719 | public class ForkJoinWorkerThread extend
719		}
720
721		/**
722	<	* Runs tasks until pool isQuiescent
722	>	* Runs tasks until pool isQuiescent.
723		*/
724		final void helpQuiescePool() {
725		for (;;) {
#	Line 745 | Line 758 | public class ForkJoinWorkerThread extend
758
759		private static long fieldOffset(String fieldName)
760		throws NoSuchFieldException {
761	<	return _unsafe.objectFieldOffset
761	>	return UNSAFE.objectFieldOffset
762		(ForkJoinWorkerThread.class.getDeclaredField(fieldName));
763		}
764
765	<	static final Unsafe _unsafe;
765	>	static final Unsafe UNSAFE;
766		static final long baseOffset;
767		static final long spOffset;
768		static final long runStateOffset;
#	Line 757 | Line 770 | public class ForkJoinWorkerThread extend
770		static final int qShift;
771		static {
772		try {
773	<	_unsafe = getUnsafe();
773	>	UNSAFE = getUnsafe();
774		baseOffset = fieldOffset("base");
775		spOffset = fieldOffset("sp");
776		runStateOffset = fieldOffset("runState");
777	<	qBase = _unsafe.arrayBaseOffset(ForkJoinTask[].class);
778	<	int s = _unsafe.arrayIndexScale(ForkJoinTask[].class);
777	>	qBase = UNSAFE.arrayBaseOffset(ForkJoinTask[].class);
778	>	int s = UNSAFE.arrayIndexScale(ForkJoinTask[].class);
779		if ((s & (s-1)) != 0)
780		throw new Error("data type scale not a power of two");
781		qShift = 31 - Integer.numberOfLeadingZeros(s);

Diff Legend

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