New implementation of BLACKHOLEs

This replaces the global blackhole_queue with a clever scheme that
enables us to queue up blocked threads on the closure that they are
blocked on, while still avoiding atomic instructions in the common
case.

Advantages:

 - gets rid of a locked global data structure and some tricky GC code
   (replacing it with some per-thread data structures and different
   tricky GC code :)

 - wakeups are more prompt: parallel/concurrent performance should
   benefit.  I haven't seen anything dramatic in the parallel
   benchmarks so far, but a couple of threading benchmarks do improve
   a bit.

 - waking up a thread blocked on a blackhole is now O(1) (e.g. if
   it is the target of throwTo).

 - less sharing and better separation of Capabilities: communication
   is done with messages, the data structures are strictly owned by a
   Capability and cannot be modified except by sending messages.

 - this change will utlimately enable us to do more intelligent
   scheduling when threads block on each other.  This is what started
   off the whole thing, but it isn't done yet (#3838).

I'll be documenting all this on the wiki in due course.

1 files changed, 296 insertions, 0 deletions

diff --git a/rts/Messages.c b/rts/Messages.c
new file mode 100644
index 0000000000..2b40f76e93
--- /dev/null
+++ b/rts/Messages.c
@@ -0,0 +1,296 @@
+/* ---------------------------------------------------------------------------
+ *
+ * (c) The GHC Team, 2010
+ *
+ * Inter-Capability message passing
+ *
+ * --------------------------------------------------------------------------*/
+
+#include "Rts.h"
+#include "Messages.h"
+#include "Trace.h"
+#include "Capability.h"
+#include "Schedule.h"
+#include "Threads.h"
+#include "RaiseAsync.h"
+#include "sm/Storage.h"
+
+/* ----------------------------------------------------------------------------
+   Send a message to another Capability
+   ------------------------------------------------------------------------- */
+
+#ifdef THREADED_RTS
+
+void sendMessage(Capability *from_cap, Capability *to_cap, Message *msg)
+{
+    ACQUIRE_LOCK(&to_cap->lock);
+
+#ifdef DEBUG    
+    {
+        const StgInfoTable *i = msg->header.info;
+        if (i != &stg_MSG_WAKEUP_info &&
+            i != &stg_MSG_THROWTO_info &&
+            i != &stg_MSG_BLACKHOLE_info &&
+            i != &stg_MSG_TRY_WAKEUP_info &&
+            i != &stg_IND_info && // can happen if a MSG_BLACKHOLE is revoked
+            i != &stg_WHITEHOLE_info) {
+            barf("sendMessage: %p", i);
+        }
+    }
+#endif
+
+    msg->link = to_cap->inbox;
+    to_cap->inbox = msg;
+
+    recordClosureMutated(from_cap,(StgClosure*)msg);
+
+    if (to_cap->running_task == NULL) {
+	to_cap->running_task = myTask(); 
+            // precond for releaseCapability_()
+	releaseCapability_(to_cap,rtsFalse);
+    } else {
+        contextSwitchCapability(to_cap);
+    }
+
+    RELEASE_LOCK(&to_cap->lock);
+}
+
+#endif /* THREADED_RTS */
+
+/* ----------------------------------------------------------------------------
+   Handle a message
+   ------------------------------------------------------------------------- */
+
+#ifdef THREADED_RTS
+
+void
+executeMessage (Capability *cap, Message *m)
+{
+    const StgInfoTable *i;
+
+loop:
+    write_barrier(); // allow m->header to be modified by another thread
+    i = m->header.info;
+    if (i == &stg_MSG_WAKEUP_info)
+    {
+        // the plan is to eventually get rid of these and use
+        // TRY_WAKEUP instead.
+        MessageWakeup *w = (MessageWakeup *)m;
+        StgTSO *tso = w->tso;
+        debugTraceCap(DEBUG_sched, cap, "message: wakeup thread %ld", 
+                      (lnat)tso->id);
+        ASSERT(tso->cap == cap);
+        ASSERT(tso->why_blocked == BlockedOnMsgWakeup);
+        ASSERT(tso->block_info.closure == (StgClosure *)m);
+        tso->why_blocked = NotBlocked;
+        appendToRunQueue(cap, tso);
+    }
+    else if (i == &stg_MSG_TRY_WAKEUP_info)
+    {
+        StgTSO *tso = ((MessageWakeup *)m)->tso;
+        debugTraceCap(DEBUG_sched, cap, "message: try wakeup thread %ld", 
+                      (lnat)tso->id);
+        tryWakeupThread(cap, tso);
+    }
+    else if (i == &stg_MSG_THROWTO_info)
+    {
+        MessageThrowTo *t = (MessageThrowTo *)m;
+        nat r;
+        const StgInfoTable *i;
+
+        i = lockClosure((StgClosure*)m);
+        if (i != &stg_MSG_THROWTO_info) {
+            unlockClosure((StgClosure*)m, i);
+            goto loop;
+        }
+
+        debugTraceCap(DEBUG_sched, cap, "message: throwTo %ld -> %ld", 
+                      (lnat)t->source->id, (lnat)t->target->id);
+
+        ASSERT(t->source->why_blocked == BlockedOnMsgThrowTo);
+        ASSERT(t->source->block_info.closure == (StgClosure *)m);
+
+        r = throwToMsg(cap, t);
+
+        switch (r) {
+        case THROWTO_SUCCESS:
+            ASSERT(t->source->sp[0] == (StgWord)&stg_block_throwto_info);
+            t->source->sp += 3;
+            unblockOne(cap, t->source);
+            // this message is done
+            unlockClosure((StgClosure*)m, &stg_IND_info);
+            break;
+        case THROWTO_BLOCKED:
+            // unlock the message
+            unlockClosure((StgClosure*)m, &stg_MSG_THROWTO_info);
+            break;
+        }
+    }
+    else if (i == &stg_MSG_BLACKHOLE_info)
+    {
+        nat r;
+        MessageBlackHole *b = (MessageBlackHole*)m;
+
+        r = messageBlackHole(cap, b);
+        if (r == 0) {
+            tryWakeupThread(cap, b->tso);
+        }
+        return;
+    }
+    else if (i == &stg_IND_info)
+    {
+        // message was revoked
+        return;
+    }
+    else if (i == &stg_WHITEHOLE_info)
+    {
+        goto loop;
+    }
+    else
+    {
+        barf("executeMessage: %p", i);
+    }
+}
+
+#endif
+
+/* ----------------------------------------------------------------------------
+   Handle a MSG_BLACKHOLE message
+
+   This is called from two places: either we just entered a BLACKHOLE
+   (stg_BLACKHOLE_info), or we received a MSG_BLACKHOLE in our
+   cap->inbox.  
+
+   We need to establish whether the BLACKHOLE belongs to
+   this Capability, and 
+     - if so, arrange to block the current thread on it
+     - otherwise, forward the message to the right place
+
+   Returns:
+     - 0 if the blocked thread can be woken up by the caller
+     - 1 if the thread is still blocked, and we promise to send a MSG_TRY_WAKEUP
+       at some point in the future.
+
+   ------------------------------------------------------------------------- */
+
+nat messageBlackHole(Capability *cap, MessageBlackHole *msg)
+{
+    const StgInfoTable *info;
+    StgClosure *p;
+    StgBlockingQueue *bq;
+    StgClosure *bh = msg->bh;
+    StgTSO *owner;
+
+    debugTraceCap(DEBUG_sched, cap, "message: thread %d blocking on blackhole %p", 
+                  (lnat)msg->tso->id, msg->bh);
+
+    info = bh->header.info;
+
+    // If we got this message in our inbox, it might be that the
+    // BLACKHOLE has already been updated, and GC has shorted out the
+    // indirection, so the pointer no longer points to a BLACKHOLE at
+    // all.
+    if (info != &stg_BLACKHOLE_info && 
+        info != &stg_CAF_BLACKHOLE_info && 
+        info != &stg_WHITEHOLE_info) {
+        // if it is a WHITEHOLE, then a thread is in the process of
+        // trying to BLACKHOLE it.  But we know that it was once a
+        // BLACKHOLE, so there is at least a valid pointer in the
+        // payload, so we can carry on.
+        return 0;
+    }
+
+    // we know at this point that the closure 
+loop:
+    p = ((StgInd*)bh)->indirectee;
+    info = p->header.info;
+
+    if (info == &stg_IND_info)
+    {
+        // This could happen, if e.g. we got a BLOCKING_QUEUE that has
+        // just been replaced with an IND by another thread in
+        // updateThunk().  In which case, if we read the indirectee
+        // again we should get the value.
+        goto loop;
+    }
+
+    else if (info == &stg_TSO_info)
+    {
+        owner = deRefTSO((StgTSO *)p);
+
+#ifdef THREADED_RTS
+        if (owner->cap != cap) {
+            sendMessage(cap, owner->cap, (Message*)msg);
+            debugTraceCap(DEBUG_sched, cap, "forwarding message to cap %d", owner->cap->no);
+            return 1;
+        }
+#endif
+        // owner is the owner of the BLACKHOLE, and resides on this
+        // Capability.  msg->tso is the first thread to block on this
+        // BLACKHOLE, so we first create a BLOCKING_QUEUE object.
+
+        bq = (StgBlockingQueue*)allocate(cap, sizeofW(StgBlockingQueue));
+            
+        // initialise the BLOCKING_QUEUE object
+        SET_HDR(bq, &stg_BLOCKING_QUEUE_DIRTY_info, CCS_SYSTEM);
+        bq->bh = bh;
+        bq->queue = msg;
+        bq->owner = owner;
+        
+        msg->link = (MessageBlackHole*)END_TSO_QUEUE;
+        
+        // All BLOCKING_QUEUES are linked in a list on owner->bq, so
+        // that we can search through them in the event that there is
+        // a collision to update a BLACKHOLE and a BLOCKING_QUEUE
+        // becomes orphaned (see updateThunk()).
+        bq->link = owner->bq;
+        owner->bq = bq;
+        dirty_TSO(cap, owner); // we modified owner->bq
+        
+        // point to the BLOCKING_QUEUE from the BLACKHOLE
+        write_barrier(); // make the BQ visible
+        ((StgInd*)bh)->indirectee = (StgClosure *)bq;
+        recordClosureMutated(cap,bh); // bh was mutated
+
+        debugTraceCap(DEBUG_sched, cap, "thread %d blocked on thread %d", 
+                      (lnat)msg->tso->id, (lnat)owner->id);
+
+        return 1; // blocked
+    }
+    else if (info == &stg_BLOCKING_QUEUE_CLEAN_info || 
+             info == &stg_BLOCKING_QUEUE_DIRTY_info)
+    {
+        StgBlockingQueue *bq = (StgBlockingQueue *)p;
+
+        ASSERT(bq->bh == bh);
+
+        owner = deRefTSO(bq->owner);
+
+        ASSERT(owner != END_TSO_QUEUE);
+
+#ifdef THREADED_RTS
+        if (owner->cap != cap) {
+            sendMessage(cap, owner->cap, (Message*)msg);
+            debugTraceCap(DEBUG_sched, cap, "forwarding message to cap %d", owner->cap->no);
+            return 1;
+        }
+#endif
+
+        msg->link = bq->queue;
+        bq->queue = msg;
+        recordClosureMutated(cap,(StgClosure*)msg);
+
+        if (info == &stg_BLOCKING_QUEUE_CLEAN_info) {
+            bq->header.info = &stg_BLOCKING_QUEUE_DIRTY_info;
+            recordClosureMutated(cap,bq);
+        }
+
+        debugTraceCap(DEBUG_sched, cap, "thread %d blocked on thread %d", 
+                      (lnat)msg->tso->id, (lnat)owner->id);
+
+        return 1; // blocked
+    }
+    
+    return 0; // not blocked
+}
+