X-Git-Url: http://git.megacz.com/?a=blobdiff_plain;f=rts%2FCapability.c;h=510656f473b6c4911f54e32b12cc50cca2e87b1f;hb=e576ba5d31fbae54c43e88316fb0dbdba9cbd4ff;hp=0415092a0308d5556e12a16a5125f4a6a8c321ba;hpb=5a2769f0273dd389977e8283375e7920d183bdd4;p=ghc-hetmet.git

diff --git a/rts/Capability.c b/rts/Capability.c
index 0415092..510656f 100644
--- a/rts/Capability.c
+++ b/rts/Capability.c
@@ -153,7 +153,8 @@ initCapability( Capability *cap, nat i )
 	cap->mut_lists[g] = NULL;
     }
 
-    cap->free_tvar_wait_queues = END_STM_WAIT_QUEUE;
+    cap->free_tvar_watch_queues = END_STM_WATCH_QUEUE;
+    cap->free_invariant_check_queues = END_INVARIANT_CHECK_QUEUE;
     cap->free_trec_chunks = END_STM_CHUNK_LIST;
     cap->free_trec_headers = NO_TREC;
     cap->transaction_tokens = 0;
@@ -518,8 +519,10 @@ wakeupThreadOnCapability (Capability *cap, StgTSO *tso)
 {
     ASSERT(tso->cap == cap);
     ASSERT(tso->bound ? tso->bound->cap == cap : 1);
+    ASSERT_LOCK_HELD(&cap->lock);
+
+    tso->cap = cap;
 
-    ACQUIRE_LOCK(&cap->lock);
     if (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.
@@ -535,6 +538,33 @@ wakeupThreadOnCapability (Capability *cap, StgTSO *tso)
 	// 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);
 }
 
@@ -617,7 +647,13 @@ shutdownCapability (Capability *cap, Task *task)
 
     task->cap = cap;
 
-    for (i = 0; i < 50; i++) {
+    // Loop indefinitely until all the workers have exited and there
+    // are no Haskell threads left.  We used to bail out after 50
+    // iterations of this loop, but that occasionally left a worker
+    // running which caused problems later (the closeMutex() below
+    // isn't safe, for one thing).
+
+    for (i = 0; /* i < 50 */; i++) {
 	debugTrace(DEBUG_sched, 
 		   "shutting down capability %d, attempt %d", cap->no, i);
 	ACQUIRE_LOCK(&cap->lock);
@@ -637,11 +673,17 @@ shutdownCapability (Capability *cap, Task *task)
 	    continue;
 	}
 	debugTrace(DEBUG_sched, "capability %d is stopped.", cap->no);
+    freeCapability(cap);
 	RELEASE_LOCK(&cap->lock);
 	break;
     }
     // we now have the Capability, its run queue and spare workers
     // list are both empty.
+
+    // ToDo: we can't drop this mutex, because there might still be
+    // threads performing foreign calls that will eventually try to 
+    // return via resumeThread() and attempt to grab cap->lock.
+    // closeMutex(&cap->lock);
 }
 
 /* ----------------------------------------------------------------------------
@@ -669,4 +711,11 @@ tryGrabCapability (Capability *cap, Task *task)
 
 #endif /* THREADED_RTS */
 
+void
+freeCapability (Capability *cap) {
+    stgFree(cap->mut_lists);
+#if defined(THREADED_RTS) || defined(PARALLEL_HASKELL)
+    freeSparkPool(&cap->r.rSparks);
+#endif
+}