Fix race condition in wakeupThreadOnCapability() (#2574)

wakeupThreadOnCapbility() is used to signal another capability that
there is a thread waiting to be added to its run queue.  It adds the
thread to the (locked) wakeup queue on the remote capability.  In
order to do this, it has to modify the TSO's link field, which has a
write barrier.  The write barrier might put the TSO on the mutable
list, and the bug was that it was using the mutable list of the
*target* capability, which we do not have exclusive access to.  We
should be using the current Capabilty's mutable list in this case.

1 files changed, 23 insertions, 40 deletions

diff --git a/rts/Capability.c b/rts/Capability.c
index 0f03621003..1f1a1aea2a 100644
--- a/rts/Capability.c
+++ b/rts/Capability.c
@@ -540,57 +540,40 @@ yieldCapability (Capability** pCap, Task *task)
  * ------------------------------------------------------------------------- */
 
 void
-wakeupThreadOnCapability (Capability *cap, StgTSO *tso)
+wakeupThreadOnCapability (Capability *my_cap, 
+                          Capability *other_cap, 
+                          StgTSO *tso)
 {
-    ASSERT(tso->cap == cap);
-    ASSERT(tso->bound ? tso->bound->cap == cap : 1);
-    ASSERT_LOCK_HELD(&cap->lock);
+    ACQUIRE_LOCK(&other_cap->lock);
 
-    tso->cap = cap;
+    // ASSUMES: cap->lock is held (asserted in wakeupThreadOnCapability)
+    if (tso->bound) {
+	ASSERT(tso->bound->cap == tso->cap);
+    	tso->bound->cap = other_cap;
+    }
+    tso->cap = other_cap;
+
+    ASSERT(tso->bound ? tso->bound->cap == other_cap : 1);
 
-    if (cap->running_task == NULL) {
+    if (other_cap->running_task == NULL) {
 	// nobody is running this Capability, we can add our thread
 	// directly onto the run queue and start up a Task to run it.
-	appendToRunQueue(cap,tso);
 
-	// start it up
-	cap->running_task = myTask(); // precond for releaseCapability_()
-	trace(TRACE_sched, "resuming capability %d", cap->no);
-	releaseCapability_(cap);
+	other_cap->running_task = myTask(); 
+            // precond for releaseCapability_() and appendToRunQueue()
+
+	appendToRunQueue(other_cap,tso);
+
+	trace(TRACE_sched, "resuming capability %d", other_cap->no);
+	releaseCapability_(other_cap);
     } else {
-	appendToWakeupQueue(cap,tso);
+	appendToWakeupQueue(my_cap,other_cap,tso);
 	// someone is running on this Capability, so it cannot be
 	// freed without first checking the wakeup queue (see
 	// releaseCapability_).
     }
-}
 
-void
-wakeupThreadOnCapability_lock (Capability *cap, StgTSO *tso)
-{
-    ACQUIRE_LOCK(&cap->lock);
-    migrateThreadToCapability (cap, tso);
-    RELEASE_LOCK(&cap->lock);
-}
-
-void
-migrateThreadToCapability (Capability *cap, StgTSO *tso)
-{
-    // ASSUMES: cap->lock is held (asserted in wakeupThreadOnCapability)
-    if (tso->bound) {
-	ASSERT(tso->bound->cap == tso->cap);
-    	tso->bound->cap = cap;
-    }
-    tso->cap = cap;
-    wakeupThreadOnCapability(cap,tso);
-}
-
-void
-migrateThreadToCapability_lock (Capability *cap, StgTSO *tso)
-{
-    ACQUIRE_LOCK(&cap->lock);
-    migrateThreadToCapability (cap, tso);
-    RELEASE_LOCK(&cap->lock);
+    RELEASE_LOCK(&other_cap->lock);
 }
 
 /* ----------------------------------------------------------------------------
@@ -818,7 +801,7 @@ markSomeCapabilities (evac_fn evac, void *user, nat i0, nat delta)
 	}
 
 #if defined(THREADED_RTS)
-        markSparkQueue (evac, user, cap);
+        traverseSparkQueue (evac, user, cap);
 #endif
     }