Support for pausing other threadswip/pause-threads

7 files changed, 64 insertions, 1 deletions

diff --git a/compiler/prelude/primops.txt.pp b/compiler/prelude/primops.txt.pp
index 47a78e2c8d..887411c2a3 100644
--- a/compiler/prelude/primops.txt.pp
+++ b/compiler/prelude/primops.txt.pp
@@ -2892,6 +2892,12 @@ primop  IsCurrentThreadBoundOp "isCurrentThreadBound#" GenPrimOp
    out_of_line = True
    has_side_effects = True
 
+primop PauseThread "pauseThread#" GenPrimOp
+   ThreadId# -> State# RealWorld -> (# State# RealWorld, MVar# RealWorld () #)
+   with
+   out_of_line = True
+   has_side_effects = True
+
 primop  NoDuplicateOp "noDuplicate#" GenPrimOp
    State# s -> State# s
    with
diff --git a/includes/rts/storage/Closures.h b/includes/rts/storage/Closures.h
index 3e90306b65..4e66c0a492 100644
--- a/includes/rts/storage/Closures.h
+++ b/includes/rts/storage/Closures.h
@@ -171,6 +171,11 @@ typedef struct _StgUpdateFrame {
 
 typedef struct {
     StgHeader  header;
+    struct StgMVar_ *mvar;
+} StgPauseThread;
+
+typedef struct {
+    StgHeader  header;
     StgWord    exceptions_blocked;
     StgClosure *handler;
 } StgCatchFrame;
@@ -279,7 +284,7 @@ typedef struct StgMVarTSOQueue_ {
     struct StgTSO_          *tso;
 } StgMVarTSOQueue;
 
-typedef struct {
+typedef struct StgMVar_ {
     StgHeader                header;
     struct StgMVarTSOQueue_ *head;
     struct StgMVarTSOQueue_ *tail;
@@ -412,6 +417,13 @@ typedef struct MessageBlackHole_ {
     StgClosure *bh;
 } MessageBlackHole;
 
+typedef struct MessagePauseThread_ {
+    StgHeader   header;
+    Message    *link;
+    StgTSO     *tso;
+    StgMVar    *mvar;
+} MessagePauseThread;
+
 /* ----------------------------------------------------------------------------
    Compact Regions
    ------------------------------------------------------------------------- */
diff --git a/includes/stg/MiscClosures.h b/includes/stg/MiscClosures.h
index 4cec0b961c..618d436273 100644
--- a/includes/stg/MiscClosures.h
+++ b/includes/stg/MiscClosures.h
@@ -63,6 +63,7 @@ RTS_RET(stg_maskAsyncExceptionszh_ret);
 RTS_RET(stg_stack_underflow_frame);
 RTS_RET(stg_restore_cccs);
 RTS_RET(stg_restore_cccs_eval);
+RTS_RET(stg_pause_thread);
 
 // RTS_FUN(stg_interp_constr1_entry);
 // RTS_FUN(stg_interp_constr2_entry);
@@ -130,6 +131,7 @@ RTS_ENTRY(stg_STM_AWOKEN);
 RTS_ENTRY(stg_MSG_TRY_WAKEUP);
 RTS_ENTRY(stg_MSG_THROWTO);
 RTS_ENTRY(stg_MSG_BLACKHOLE);
+RTS_ENTRY(stg_MSG_PAUSE_THREAD);
 RTS_ENTRY(stg_MSG_NULL);
 RTS_ENTRY(stg_MVAR_TSO_QUEUE);
 RTS_ENTRY(stg_catch);
diff --git a/rts/Messages.c b/rts/Messages.c
index d878db5eda..fa3db268cd 100644
--- a/rts/Messages.c
+++ b/rts/Messages.c
@@ -77,6 +77,15 @@ loop:
                       (W_)tso->id);
         tryWakeupThread(cap, tso);
     }
+    else if (i == &stg_MSG_PAUSE_THREAD_info)
+    {
+        MessagePauseThread *t = (MessagePauseThread *)m;
+        StgStack *stack = t->tso->stackobj;
+        stack->sp += sizeofW(StgPauseThread);
+        StgPauseThread *frame = (StgPauseThread *) stack->sp;
+        frame->mvar = t->mvar;
+        frame->header.info = &stg_pause_thread_info;
+    }
     else if (i == &stg_MSG_THROWTO_info)
     {
         MessageThrowTo *t = (MessageThrowTo *)m;
diff --git a/rts/PrimOps.cmm b/rts/PrimOps.cmm
index d9a28d7396..5c4c4d01fc 100644
--- a/rts/PrimOps.cmm
+++ b/rts/PrimOps.cmm
@@ -1451,6 +1451,22 @@ stg_writeTVarzh (P_ tvar,     /* :: TVar a */
  *
  * -------------------------------------------------------------------------- */
 
+stg_pauseThread ( P_ thread_id ) /* :: TheadId -> State# RW -> (# State# RW, MVar# () #) */
+{
+    P_ msg;
+    ALLOC_PRIM_P (SIZEOF_MessagePauseThread, stg_pauseThread, thread_id);
+    msg = Hp - SIZEOF_MessagePauseThread + WDS(1);
+
+    (P_ mvar) = call stg_newMVarzh();
+
+    SET_HDR(msg, stg_MSG_PAUSE_THREAD_info, CCCS);
+    MessagePauseThread_mvar(msg) = mvar;
+    MessagePauseThread_tso(msg) = tso;
+    ccall sendMessage(cap, StgTSO_cap(tso), msg);
+    
+    return (mvar);
+}
+
 stg_isEmptyMVarzh ( P_ mvar /* :: MVar a */ )
 {
     if (StgMVar_value(mvar) == stg_END_TSO_QUEUE_closure) {
diff --git a/rts/StgMiscClosures.cmm b/rts/StgMiscClosures.cmm
index 03ea91fcb6..7ed617ce30 100644
--- a/rts/StgMiscClosures.cmm
+++ b/rts/StgMiscClosures.cmm
@@ -68,6 +68,16 @@ INFO_TABLE_RET (stg_restore_cccs_eval, RET_SMALL, W_ info_ptr, W_ cccs)
     jump stg_ap_0_fast(ret);
 }
 
+INFO_TABLE_RET (stg_pause_thread, RET_SMALL)
+    /* explicit stack */
+{
+    P_ mvar;
+    mvar = Sp(1);
+    Sp_adj(2);
+    R1 = mvar;
+    jump stg_takeMVarzh [R1];
+}
+
 /* ----------------------------------------------------------------------------
    Support for the bytecode interpreter.
    ------------------------------------------------------------------------- */
@@ -582,6 +592,9 @@ INFO_TABLE_CONSTR(stg_MSG_BLACKHOLE,3,0,0,PRIM,"MSG_BLACKHOLE","MSG_BLACKHOLE")
 INFO_TABLE_CONSTR(stg_MSG_NULL,1,0,0,PRIM,"MSG_NULL","MSG_NULL")
 { foreign "C" barf("MSG_NULL object (%p) entered!", R1) never returns; }
 
+INFO_TABLE_CONSTR(stg_MSG_PAUSE_THREAD,1,0,0,PRIM,"MSG_PAUSE_THREAD","MSG_PAUSE_THREAD")
+{ foreign "C" barf("MSG_PAUSE_THREAD object (%p) entered!", R1) never returns; }
+
 /* ----------------------------------------------------------------------------
    END_TSO_QUEUE
 
diff --git a/utils/deriveConstants/Main.hs b/utils/deriveConstants/Main.hs
index a812ac42c8..434dad0737 100644
--- a/utils/deriveConstants/Main.hs
+++ b/utils/deriveConstants/Main.hs
@@ -556,6 +556,11 @@ wanteds os = concat
           ,closureField C "MessageBlackHole" "tso"
           ,closureField C "MessageBlackHole" "bh"
 
+          ,closureSize  C "MessagePauseThread"
+          ,closureField C "MessagePauseThread" "link"
+          ,closureField C "MessagePauseThread" "tso"
+          ,closureField C "MessagePauseThread" "mvar"
+
           ,closureSize  C "StgCompactNFData"
           ,closureField C "StgCompactNFData" "totalW"
           ,closureField C "StgCompactNFData" "autoBlockW"