Add hs_try_putmvar()

Summary:
This is a fast, non-blocking, asynchronous, interface to tryPutMVar that
can be called from C/C++.

It's useful for callback-based C/C++ APIs: the idea is that the callback
invokes hs_try_putmvar(), and the Haskell code waits for the callback to
run by blocking in takeMVar.

The callback doesn't block - this is often a requirement of
callback-based APIs.  The callback wakes up the Haskell thread with
minimal overhead and no unnecessary context-switches.

There are a couple of benchmarks in
testsuite/tests/concurrent/should_run.  Some example results comparing
hs_try_putmvar() with using a standard foreign export:

    ./hs_try_putmvar003 1 64 16 100 +RTS -s -N4     0.49s
    ./hs_try_putmvar003 2 64 16 100 +RTS -s -N4     2.30s

hs_try_putmvar() is 4x faster for this workload (see the source for
hs_try_putmvar003.hs for details of the workload).

An alternative solution is to use the IO Manager for this.  We've tried
it, but there are problems with that approach:
* Need to create a new file descriptor for each callback
* The IO Manger thread(s) become a bottleneck
* More potential for things to go wrong, e.g. throwing an exception in
  an IO Manager callback kills the IO Manager thread.

Test Plan: validate; new unit tests

Reviewers: niteria, erikd, ezyang, bgamari, austin, hvr

Subscribers: thomie

Differential Revision: https://phabricator.haskell.org/D2501

1 files changed, 12 insertions, 1 deletions

diff --git a/rts/Capability.h b/rts/Capability.h
index 8e0288b15e..bbf026279f 100644
--- a/rts/Capability.h
+++ b/rts/Capability.h
@@ -123,6 +123,7 @@ struct Capability_ {
     //    returning_tasks_{hd,tl}
     //    wakeup_queue
     //    inbox
+    //    putMVars
     Mutex lock;
 
     // Tasks waiting to return from a foreign call, or waiting to make
@@ -138,6 +139,10 @@ struct Capability_ {
     // Locks required: cap->lock
     Message *inbox;
 
+    // putMVars are really messages, but they're allocated with malloc() so they
+    // can't go on the inbox queue: the GC would get confused.
+    struct PutMVar_ *putMVars;
+
     SparkPool *sparks;
 
     // Stats on spark creation/conversion
@@ -378,6 +383,11 @@ extern uint32_t numa_map[MAX_NUMA_NODES];
    Messages
    -------------------------------------------------------------------------- */
 
+typedef struct PutMVar_ {
+    StgStablePtr mvar;
+    struct PutMVar_ *link;
+} PutMVar;
+
 #ifdef THREADED_RTS
 
 INLINE_HEADER rtsBool emptyInbox(Capability *cap);
@@ -459,7 +469,8 @@ contextSwitchCapability (Capability *cap)
 
 INLINE_HEADER rtsBool emptyInbox(Capability *cap)
 {
-    return (cap->inbox == (Message*)END_TSO_QUEUE);
+    return (cap->inbox == (Message*)END_TSO_QUEUE &&
+            cap->putMVars == NULL);
 }
 
 #endif