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.51 by jsr166, Mon Sep 20 20:42:37 2010 UTC vs.
Revision 1.61 by jsr166, Tue Nov 23 06:21:54 2010 UTC

#	Line 6 | Line 6
6
7		package jsr166y;
8
9	–	import java.util.concurrent.*;
10	–
9		import java.util.Random;
10		import java.util.Collection;
11		import java.util.concurrent.locks.LockSupport;
12	+	import java.util.concurrent.RejectedExecutionException;
13
14		/**
15	<	* A thread managed by a {@link ForkJoinPool}.  This class is
16	<	* subclassable solely for the sake of adding functionality -- there
17	<	* are no overridable methods dealing with scheduling or execution.
18	<	* However, you can override initialization and termination methods
19	<	* surrounding the main task processing loop.  If you do create such a
20	<	* subclass, you will also need to supply a custom {@link
21	<	* ForkJoinPool.ForkJoinWorkerThreadFactory} to use it in a {@code
22	<	* ForkJoinPool}.
15	>	* A thread managed by a {@link ForkJoinPool}, which executes
16	>	* {@link ForkJoinTask}s.
17	>	* This class is subclassable solely for the sake of adding
18	>	* functionality -- there are no overridable methods dealing with
19	>	* scheduling or execution.  However, you can override initialization
20	>	* and termination methods surrounding the main task processing loop.
21	>	* If you do create such a subclass, you will also need to supply a
22	>	* custom {@link ForkJoinPool.ForkJoinWorkerThreadFactory} to use it
23	>	* in a {@code ForkJoinPool}.
24		*
25		* @since 1.7
26		* @author Doug Lea
#	Line 349 | Line 349 | public class ForkJoinWorkerThread extend
349		/**
350		* Initializes internal state after construction but before
351		* processing any tasks. If you override this method, you must
352	<	* invoke @code{super.onStart()} at the beginning of the method.
352	>	* invoke {@code super.onStart()} at the beginning of the method.
353		* Initialization requires care: Most fields must have legal
354		* default values, to ensure that attempted accesses from other
355		* threads work correctly even before this thread starts
#	Line 399 | Line 399 | public class ForkJoinWorkerThread extend
399		/**
400		* This method is required to be public, but should never be
401		* called explicitly. It performs the main run loop to execute
402	<	* ForkJoinTasks.
402	>	* {@link ForkJoinTask}s.
403		*/
404		public void run() {
405		Throwable exception = null;
#	Line 601 | Line 601 | public class ForkJoinWorkerThread extend
601		if (t == null)   // lost to stealer
602		break;
603		if (UNSAFE.compareAndSwapObject(q, u, t, null)) {
604	+	/*
605	+	* Note: here and in related methods, as a
606	+	* performance (not correctness) issue, we'd like
607	+	* to encourage compiler not to arbitrarily
608	+	* postpone setting sp after successful CAS.
609	+	* Currently there is no intrinsic for arranging
610	+	* this, but using Unsafe putOrderedInt may be a
611	+	* preferable strategy on some compilers even
612	+	* though its main effect is a pre-, not post-
613	+	* fence. To simplify possible changes, the option
614	+	* is left in comments next to the associated
615	+	* assignments.
616	+	*/
617		sp = s; // putOrderedInt may encourage more timely write
618		// UNSAFE.putOrderedInt(this, spOffset, s);
619		return t;
#	Line 750 | Line 763 | public class ForkJoinWorkerThread extend
763		// Run State management
764
765		// status check methods used mainly by ForkJoinPool
766	<	final boolean isRunning()     { return runState == 0; }
767	<	final boolean isTerminated()  { return (runState & TERMINATED) != 0; }
768	<	final boolean isSuspended()   { return (runState & SUSPENDED) != 0; }
769	<	final boolean isTrimmed()     { return (runState & TRIMMED) != 0; }
766	>	final boolean isRunning()    { return runState == 0; }
767	>	final boolean isTerminated() { return (runState & TERMINATED) != 0; }
768	>	final boolean isSuspended()  { return (runState & SUSPENDED) != 0; }
769	>	final boolean isTrimmed()    { return (runState & TRIMMED) != 0; }
770
771		final boolean isTerminating() {
772		if ((runState & TERMINATING) != 0)
#	Line 857 | Line 870 | public class ForkJoinWorkerThread extend
870		*/
871		final void cancelTasks() {
872		ForkJoinTask<?> cj = currentJoin; // try to cancel ongoing tasks
873	<	if (cj != null) {
861	<	currentJoin = null;
873	>	if (cj != null && cj.status >= 0) {
874		cj.cancelIgnoringExceptions();
875		try {
876		this.interrupt(); // awaken wait
#	Line 866 | Line 878 | public class ForkJoinWorkerThread extend
878		}
879		}
880		ForkJoinTask<?> cs = currentSteal;
881	<	if (cs != null) {
870	<	currentSteal = null;
881	>	if (cs != null && cs.status >= 0)
882		cs.cancelIgnoringExceptions();
872	–	}
883		while (base != sp) {
884		ForkJoinTask<?> t = deqTask();
885		if (t != null)
#	Line 932 | Line 942 | public class ForkJoinWorkerThread extend
942		* Possibly runs some tasks and/or blocks, until task is done.
943		*
944		* @param joinMe the task to join
945	+	* @param timed true if use timed wait
946	+	* @param nanos wait time if timed
947		*/
948	<	final void joinTask(ForkJoinTask<?> joinMe) {
948	>	final void joinTask(ForkJoinTask<?> joinMe, boolean timed, long nanos) {
949		// currentJoin only written by this thread; only need ordered store
950		ForkJoinTask<?> prevJoin = currentJoin;
951		UNSAFE.putOrderedObject(this, currentJoinOffset, joinMe);
952	<	if (sp != base)
941	<	localHelpJoinTask(joinMe);
942	<	if (joinMe.status >= 0)
943	<	pool.awaitJoin(joinMe, this);
952	>	pool.awaitJoin(joinMe, this, timed, nanos);
953		UNSAFE.putOrderedObject(this, currentJoinOffset, prevJoin);
954		}
955
956		/**
957	<	* Run tasks in local queue until given task is done.
958	<	*
959	<	* @param joinMe the task to join
960	<	*/
961	<	private void localHelpJoinTask(ForkJoinTask<?> joinMe) {
953	<	int s;
954	<	ForkJoinTask<?>[] q;
955	<	while (joinMe.status >= 0 && (s = sp) != base && (q = queue) != null) {
956	<	int i = (q.length - 1) & --s;
957	<	long u = (i << qShift) + qBase; // raw offset
958	<	ForkJoinTask<?> t = q[i];
959	<	if (t == null)  // lost to a stealer
960	<	break;
961	<	if (UNSAFE.compareAndSwapObject(q, u, t, null)) {
962	<	/*
963	<	* This recheck (and similarly in helpJoinTask)
964	<	* handles cases where joinMe is independently
965	<	* cancelled or forced even though there is other work
966	<	* available. Back out of the pop by putting t back
967	<	* into slot before we commit by writing sp.
968	<	*/
969	<	if (joinMe.status < 0) {
970	<	UNSAFE.putObjectVolatile(q, u, t);
971	<	break;
972	<	}
973	<	sp = s;
974	<	// UNSAFE.putOrderedInt(this, spOffset, s);
975	<	t.quietlyExec();
976	<	}
977	<	}
978	<	}
979	<
980	<	/**
981	<	* Unless terminating, tries to locate and help perform tasks for
982	<	* a stealer of the given task, or in turn one of its stealers.
983	<	* Traces currentSteal->currentJoin links looking for a thread
984	<	* working on a descendant of the given task and with a non-empty
985	<	* queue to steal back and execute tasks from.
957	>	* Tries to locate and help perform tasks for a stealer of the
958	>	* given task, or in turn one of its stealers.  Traces
959	>	* currentSteal->currentJoin links looking for a thread working on
960	>	* a descendant of the given task and with a non-empty queue to
961	>	* steal back and execute tasks from.
962		*
963		* The implementation is very branchy to cope with potential
964		* inconsistencies or loops encountering chains that are stale,
#	Line 992 | Line 968 | public class ForkJoinWorkerThread extend
968		* don't work out.
969		*
970		* @param joinMe the task to join
971	<	*/
972	<	final void helpJoinTask(ForkJoinTask<?> joinMe) {
973	<	ForkJoinWorkerThread[] ws;
974	<	int n;
975	<	if (joinMe.status < 0)                // already done
976	<	return;
977	<	if ((runState & TERMINATING) != 0) {  // cancel if shutting down
978	<	joinMe.cancelIgnoringExceptions();
979	<	return;
980	<	}
981	<	if ((ws = pool.workers) == null || (n = ws.length) <= 1)
982	<	return;                           // need at least 2 workers
983	<
984	<	ForkJoinTask<?> task = joinMe;        // base of chain
985	<	ForkJoinWorkerThread thread = this;   // thread with stolen task
986	<	for (int d = 0; d < MAX_HELP_DEPTH; ++d) { // chain length
987	<	// Try to find v, the stealer of task, by first using hint
988	<	ForkJoinWorkerThread v = ws[thread.stealHint & (n - 1)];
989	<	if (v == null || v.currentSteal != task) {
990	<	for (int j = 0; ; ++j) {      // search array
991	<	if (j < n) {
992	<	ForkJoinTask<?> vs;
993	<	if ((v = ws[j]) != null &&
994	<	(vs = v.currentSteal) != null) {
995	<	if (joinMe.status < 0 || task.status < 0)
996	<	return;       // stale or done
997	<	if (vs == task) {
998	<	thread.stealHint = j;
999	<	break;        // save hint for next time
971	>	* @param running if false, then must update pool count upon
972	>	*  running a task
973	>	* @return value of running on exit
974	>	*/
975	>	final boolean helpJoinTask(ForkJoinTask<?> joinMe, boolean running) {
976	>	/*
977	>	* Initial checks to (1) abort if terminating; (2) clean out
978	>	* old cancelled tasks from local queue; (3) if joinMe is next
979	>	* task, run it; (4) omit scan if local queue nonempty (since
980	>	* it may contain non-descendents of joinMe).
981	>	*/
982	>	ForkJoinPool p = pool;
983	>	for (;;) {
984	>	ForkJoinTask<?>[] q;
985	>	int s;
986	>	if (joinMe.status < 0)
987	>	return running;
988	>	else if ((runState & TERMINATING) != 0) {
989	>	joinMe.cancelIgnoringExceptions();
990	>	return running;
991	>	}
992	>	else if ((s = sp) == base || (q = queue) == null)
993	>	break;                            // queue empty
994	>	else {
995	>	int i = (q.length - 1) & --s;
996	>	long u = (i << qShift) + qBase;   // raw offset
997	>	ForkJoinTask<?> t = q[i];
998	>	if (t == null)
999	>	break;                        // lost to a stealer
1000	>	else if (t != joinMe && t.status >= 0)
1001	>	return running;               // cannot safely help
1002	>	else if ((running ||
1003	>	(running = p.tryIncrementRunningCount())) &&
1004	>	UNSAFE.compareAndSwapObject(q, u, t, null)) {
1005	>	sp = s; // putOrderedInt may encourage more timely write
1006	>	// UNSAFE.putOrderedInt(this, spOffset, s);
1007	>	t.quietlyExec();
1008	>	}
1009	>	}
1010	>	}
1011	>
1012	>	int n;                                    // worker array size
1013	>	ForkJoinWorkerThread[] ws = p.workers;
1014	>	if (ws != null && (n = ws.length) > 1) {  // need at least 2 workers
1015	>	ForkJoinTask<?> task = joinMe;        // base of chain
1016	>	ForkJoinWorkerThread thread = this;   // thread with stolen task
1017	>
1018	>	outer:for (int d = 0; d < MAX_HELP_DEPTH; ++d) { // chain length
1019	>	// Try to find v, the stealer of task, by first using hint
1020	>	ForkJoinWorkerThread v = ws[thread.stealHint & (n - 1)];
1021	>	if (v == null || v.currentSteal != task) {
1022	>	for (int j = 0; ; ++j) {      // search array
1023	>	if (j < n) {
1024	>	ForkJoinTask<?> vs;
1025	>	if ((v = ws[j]) != null &&
1026	>	(vs = v.currentSteal) != null) {
1027	>	if (joinMe.status < 0)
1028	>	break outer;
1029	>	if (vs == task) {
1030	>	if (task.status < 0)
1031	>	break outer; // stale
1032	>	thread.stealHint = j;
1033	>	break;        // save hint for next time
1034	>	}
1035		}
1036		}
1037	+	else
1038	+	break outer;          // no stealer
1039		}
1027	–	else
1028	–	return;               // no stealer
1040		}
1041	<	}
1042	<	for (;;) { // Try to help v, using specialized form of deqTask
1043	<	if (joinMe.status < 0)
1044	<	return;
1045	<	int b = v.base;
1046	<	ForkJoinTask<?>[] q = v.queue;
1047	<	if (b == v.sp || q == null)
1048	<	break;
1049	<	int i = (q.length - 1) & b;
1050	<	long u = (i << qShift) + qBase;
1051	<	ForkJoinTask<?> t = q[i];
1052	<	int pid = poolIndex;
1053	<	ForkJoinTask<?> ps = currentSteal;
1054	<	if (task.status < 0)
1055	<	return;                   // stale or done
1056	<	if (t != null && v.base == b++ &&
1057	<	UNSAFE.compareAndSwapObject(q, u, t, null)) {
1058	<	if (joinMe.status < 0) {
1059	<	UNSAFE.putObjectVolatile(q, u, t);
1060	<	return;               // back out on cancel
1041	>
1042	>	// Try to help v, using specialized form of deqTask
1043	>	for (;;) {
1044	>	if (joinMe.status < 0)
1045	>	break outer;
1046	>	int b = v.base;
1047	>	ForkJoinTask<?>[] q = v.queue;
1048	>	if (b == v.sp || q == null)
1049	>	break;                    // empty
1050	>	int i = (q.length - 1) & b;
1051	>	long u = (i << qShift) + qBase;
1052	>	ForkJoinTask<?> t = q[i];
1053	>	if (task.status < 0)
1054	>	break outer;              // stale
1055	>	if (t != null &&
1056	>	(running ||
1057	>	(running = p.tryIncrementRunningCount())) &&
1058	>	v.base == b++ &&
1059	>	UNSAFE.compareAndSwapObject(q, u, t, null)) {
1060	>	if (t != joinMe && joinMe.status < 0) {
1061	>	UNSAFE.putObjectVolatile(q, u, t);
1062	>	break outer;          // joinMe cancelled; back out
1063	>	}
1064	>	v.base = b;
1065	>	if (t.status >= 0) {
1066	>	ForkJoinTask<?> ps = currentSteal;
1067	>	int pid = poolIndex;
1068	>	v.stealHint = pid;
1069	>	UNSAFE.putOrderedObject(this,
1070	>	currentStealOffset, t);
1071	>	t.quietlyExec();
1072	>	UNSAFE.putOrderedObject(this,
1073	>	currentStealOffset, ps);
1074	>	}
1075	>	}
1076	>	else if ((runState & TERMINATING) != 0) {
1077	>	joinMe.cancelIgnoringExceptions();
1078	>	break outer;
1079		}
1051	–	v.base = b;
1052	–	v.stealHint = pid;
1053	–	UNSAFE.putOrderedObject(this, currentStealOffset, t);
1054	–	t.quietlyExec();
1055	–	UNSAFE.putOrderedObject(this, currentStealOffset, ps);
1080		}
1081	+
1082	+	// Try to descend to find v's stealer
1083	+	ForkJoinTask<?> next = v.currentJoin;
1084	+	if (task.status < 0 || next == null || next == task ||
1085	+	joinMe.status < 0)
1086	+	break;                 // done, stale, dead-end, or cyclic
1087	+	task = next;
1088	+	thread = v;
1089		}
1058	–	// Try to descend to find v's stealer
1059	–	ForkJoinTask<?> next = v.currentJoin;
1060	–	if (task.status < 0 || next == null || next == task ||
1061	–	joinMe.status < 0)
1062	–	return;
1063	–	task = next;
1064	–	thread = v;
1090		}
1091	+	return running;
1092		}
1093
1094		/**

Diff Legend

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