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.58 by jsr166, Sun Nov 21 08:18:19 2010 UTC vs.
Revision 1.59 by dl, Sun Nov 21 13:55:04 2010 UTC

#	Line 12 | Line 12 | import java.util.concurrent.locks.LockSu
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 398 | 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 600 | 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 856 | Line 870 | public class ForkJoinWorkerThread extend
870		*/
871		final void cancelTasks() {
872		ForkJoinTask<?> cj = currentJoin; // try to cancel ongoing tasks
873	<	if (cj != null) {
860	<	currentJoin = null;
873	>	if (cj != null && cj.status >= 0) {
874		cj.cancelIgnoringExceptions();
875		try {
876		this.interrupt(); // awaken wait
#	Line 865 | Line 878 | public class ForkJoinWorkerThread extend
878		}
879		}
880		ForkJoinTask<?> cs = currentSteal;
881	<	if (cs != null) {
869	<	currentSteal = null;
881	>	if (cs != null && cs.status >= 0)
882		cs.cancelIgnoringExceptions();
871	–	}
883		while (base != sp) {
884		ForkJoinTask<?> t = deqTask();
885		if (t != null)
#	Line 938 | Line 949 | public class ForkJoinWorkerThread extend
949		// currentJoin only written by this thread; only need ordered store
950		ForkJoinTask<?> prevJoin = currentJoin;
951		UNSAFE.putOrderedObject(this, currentJoinOffset, joinMe);
952	<	if (isTerminating())                // cancel if shutting down
942	<	joinMe.cancelIgnoringExceptions();
943	<	else {
944	<	if (sp != base)
945	<	localHelpJoinTask(joinMe);
946	<	if (joinMe.status >= 0)
947	<	pool.awaitJoin(joinMe, this, timed, nanos);
948	<	}
952	>	pool.awaitJoin(joinMe, this, timed, nanos);
953		UNSAFE.putOrderedObject(this, currentJoinOffset, prevJoin);
954		}
955
956		/**
953	–	* Run tasks in local queue until given task is done.
954	–	* Not currently used because it complicates semantics.
955	–	*
956	–	* @param joinMe the task to join
957	–	*/
958	–	private void localHelpJoinTask(ForkJoinTask<?> joinMe) {
959	–	int s;
960	–	ForkJoinTask<?>[] q;
961	–	while (joinMe.status >= 0 && (s = sp) != base && (q = queue) != null) {
962	–	int i = (q.length - 1) & --s;
963	–	long u = (i << qShift) + qBase; // raw offset
964	–	ForkJoinTask<?> t = q[i];
965	–	if (t == null)  // lost to a stealer
966	–	break;
967	–	if (UNSAFE.compareAndSwapObject(q, u, t, null)) {
968	–	/*
969	–	* This recheck (and similarly in helpJoinTask)
970	–	* handles cases where joinMe is independently
971	–	* cancelled or forced even though there is other work
972	–	* available. Back out of the pop by putting t back
973	–	* into slot before we commit by writing sp.
974	–	*/
975	–	if (joinMe.status < 0) {
976	–	UNSAFE.putObjectVolatile(q, u, t);
977	–	break;
978	–	}
979	–	sp = s;
980	–	// UNSAFE.putOrderedInt(this, spOffset, s);
981	–	t.quietlyExec();
982	–	}
983	–	}
984	–	}
985	–
986	–	/**
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
#	Line 998 | 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 ((ws = pool.workers) == null || (n = ws.length) <= 1)
978	<	return;                           // need at least 2 workers
979	<
980	<	ForkJoinTask<?> task = joinMe;        // base of chain
981	<	ForkJoinWorkerThread thread = this;   // thread with stolen task
982	<	for (int d = 0; d < MAX_HELP_DEPTH; ++d) { // chain length
983	<	// Try to find v, the stealer of task, by first using hint
984	<	ForkJoinWorkerThread v = ws[thread.stealHint & (n - 1)];
985	<	if (v == null || v.currentSteal != task) {
986	<	for (int j = 0; ; ++j) {      // search array
987	<	if (j < n) {
988	<	ForkJoinTask<?> vs;
989	<	if ((v = ws[j]) != null &&
990	<	(v != this || base == sp) &&
991	<	(vs = v.currentSteal) != null) {
992	<	if (joinMe.status < 0 || task.status < 0)
993	<	return;       // stale or done
994	<	if (vs == task) {
995	<	thread.stealHint = j;
996	<	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	>	else if ((s = sp) == base || (q = queue) == null)
991	>	break;                            // queue empty
992	>	else {
993	>	int i = (q.length - 1) & --s;
994	>	long u = (i << qShift) + qBase;   // raw offset
995	>	ForkJoinTask<?> t = q[i];
996	>	if (t == null)
997	>	break;                        // lost to a stealer
998	>	else if (t != joinMe && t.status >= 0)
999	>	return running;               // cannot safely help
1000	>	else if ((running ||
1001	>	(running = p.tryIncrementRunningCount())) &&
1002	>	UNSAFE.compareAndSwapObject(q, u, t, null)) {
1003	>	sp = s; // putOrderedInt may encourage more timely write
1004	>	// UNSAFE.putOrderedInt(this, spOffset, s);
1005	>	if (t.status >= 0)
1006	>	t.quietlyExec();
1007	>	}
1008	>	}
1009	>	}
1010	>
1011	>	int n;                                    // worker array size
1012	>	ForkJoinWorkerThread[] ws = p.workers;
1013	>	if (ws != null && (n = ws.length) > 1) {  // need at least 2 workers
1014	>	ForkJoinTask<?> task = joinMe;        // base of chain
1015	>	ForkJoinWorkerThread thread = this;   // thread with stolen task
1016	>
1017	>	outer:for (int d = 0; d < MAX_HELP_DEPTH; ++d) { // chain length
1018	>	// Try to find v, the stealer of task, by first using hint
1019	>	ForkJoinWorkerThread v = ws[thread.stealHint & (n - 1)];
1020	>	if (v == null || v.currentSteal != task) {
1021	>	for (int j = 0; ; ++j) {      // search array
1022	>	if (j < n) {
1023	>	ForkJoinTask<?> vs;
1024	>	if ((v = ws[j]) != null &&
1025	>	(vs = v.currentSteal) != null) {
1026	>	if (joinMe.status < 0)
1027	>	break outer;
1028	>	if (vs == task) {
1029	>	if (task.status < 0)
1030	>	break outer; // stale
1031	>	thread.stealHint = j;
1032	>	break;        // save hint for next time
1033	>	}
1034		}
1035		}
1036	+	else
1037	+	break outer;          // no stealer
1038		}
1030	–	else
1031	–	return;               // no stealer
1039		}
1040	<	}
1041	<	for (;;) { // Try to help v, using specialized form of deqTask
1042	<	if (joinMe.status < 0)
1043	<	return;
1044	<	int b = v.base;
1045	<	ForkJoinTask<?>[] q = v.queue;
1046	<	if (b == v.sp || q == null)
1047	<	break;
1048	<	int i = (q.length - 1) & b;
1049	<	long u = (i << qShift) + qBase;
1050	<	ForkJoinTask<?> t = q[i];
1051	<	int pid = poolIndex;
1052	<	ForkJoinTask<?> ps = currentSteal;
1053	<	if (task.status < 0)
1054	<	return;                   // stale or done
1055	<	if (t != null && v.base == b++ &&
1056	<	UNSAFE.compareAndSwapObject(q, u, t, null)) {
1057	<	if (joinMe.status < 0) {
1058	<	UNSAFE.putObjectVolatile(q, u, t);
1059	<	return;               // back out on cancel
1040	>
1041	>	// Try to help v, using specialized form of deqTask
1042	>	for (;;) {
1043	>	if (joinMe.status < 0)
1044	>	break outer;
1045	>	int b = v.base;
1046	>	ForkJoinTask<?>[] q = v.queue;
1047	>	if (b == v.sp || q == null)
1048	>	break;                    // empty
1049	>	int i = (q.length - 1) & b;
1050	>	long u = (i << qShift) + qBase;
1051	>	ForkJoinTask<?> t = q[i];
1052	>	if (task.status < 0)
1053	>	break outer;              // stale
1054	>	if (t != null &&
1055	>	(running ||
1056	>	(running = p.tryIncrementRunningCount())) &&
1057	>	v.base == b++ &&
1058	>	UNSAFE.compareAndSwapObject(q, u, t, null)) {
1059	>	if (t != joinMe && joinMe.status < 0) {
1060	>	UNSAFE.putObjectVolatile(q, u, t);
1061	>	break outer;          // joinMe cancelled; back out
1062	>	}
1063	>	v.base = b;
1064	>	if (t.status >= 0) {
1065	>	ForkJoinTask<?> ps = currentSteal;
1066	>	int pid = poolIndex;
1067	>	v.stealHint = pid;
1068	>	UNSAFE.putOrderedObject(this,
1069	>	currentStealOffset, t);
1070	>	t.quietlyExec();
1071	>	UNSAFE.putOrderedObject(this,
1072	>	currentStealOffset, ps);
1073	>	}
1074		}
1054	–	v.base = b;
1055	–	v.stealHint = pid;
1056	–	UNSAFE.putOrderedObject(this, currentStealOffset, t);
1057	–	t.quietlyExec();
1058	–	UNSAFE.putOrderedObject(this, currentStealOffset, ps);
1075		}
1076	+
1077	+	// Try to descend to find v's stealer
1078	+	ForkJoinTask<?> next = v.currentJoin;
1079	+	if (task.status < 0 || next == null || next == task)
1080	+	break;                       // stale, dead-end, or cyclic
1081	+	if ((runState & TERMINATING) != 0)
1082	+	joinMe.cancelIgnoringExceptions();
1083	+	if (joinMe.status < 0)
1084	+	break;
1085	+	task = next;
1086	+	thread = v;
1087		}
1061	–	// Try to descend to find v's stealer
1062	–	ForkJoinTask<?> next = v.currentJoin;
1063	–	if (task.status < 0 || next == null || next == task ||
1064	–	joinMe.status < 0)
1065	–	return;
1066	–	task = next;
1067	–	thread = v;
1088		}
1089	+	return running;
1090		}
1091
1092		/**

Diff Legend

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