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, 5 insertions, 6 deletions

diff --git a/rts/Task.c b/rts/Task.c
index 9a658e019c..253520f909 100644
--- a/rts/Task.c
+++ b/rts/Task.c
@@ -36,7 +36,6 @@ uint32_t peakWorkerCount;
 static int tasksInitialized = 0;
 
 static void   freeTask  (Task *task);
-static Task * allocTask (void);
 static Task * newTask   (rtsBool);
 
 #if defined(THREADED_RTS)
@@ -117,8 +116,7 @@ freeTaskManager (void)
     return tasksRunning;
 }
 
-static Task *
-allocTask (void)
+Task* getTask (void)
 {
     Task *task;
 
@@ -209,7 +207,7 @@ newTask (rtsBool worker)
 
     task->cap           = NULL;
     task->worker        = worker;
-    task->stopped       = rtsFalse;
+    task->stopped       = rtsTrue;
     task->running_finalizers = rtsFalse;
     task->n_spare_incalls = 0;
     task->spare_incalls = NULL;
@@ -304,7 +302,7 @@ newBoundTask (void)
         stg_exit(EXIT_FAILURE);
     }
 
-    task = allocTask();
+    task = getTask();
 
     task->stopped = rtsFalse;
 
@@ -452,6 +450,7 @@ startWorkerTask (Capability *cap)
 
   // A worker always gets a fresh Task structure.
   task = newTask(rtsTrue);
+  task->stopped = rtsFalse;
 
   // The lock here is to synchronise with taskStart(), to make sure
   // that we have finished setting up the Task structure before the
@@ -499,7 +498,7 @@ void rts_setInCallCapability (
     int preferred_capability,
     int affinity USED_IF_THREADS)
 {
-    Task *task = allocTask();
+    Task *task = getTask();
     task->preferred_capability = preferred_capability;
 
 #ifdef THREADED_RTS