GitLab

Marge Bot pushed to branch master at Glasgow Haskell Compiler / GHC

Commits:

5bd6a964

by Rodrigo Mesquita at 2026-04-30T04:54:08-04:00

New rts Message to {set,unset} TSO flags

This commit introduces stg_MSG_SET_TSO_FLAG_info and
stg_MSG_UNSET_TSO_FLAG_info, which allows setting flags of a TSO other
than yourself.

This is especially useful/necessary to set breakpoints and toggle
breakpoints of different threads, which is needed to safely implement
features like pausing, toggling step-out, toggling step-in per thread,
etc.

Fixes #27131

-------------------------
Metric Decrease:
    T3294
-------------------------

ce97fd3e

by Rodrigo Mesquita at 2026-04-30T04:54:08-04:00

test: Add test setting another TSO's flags

Introduces a test that runs on two capabilities. The main thread running
on Capability 0 sets the flags on a TSO running on Capability 1.
The TSO from Capability 1 itself checks whether its flags were set and
reports that back.

This validates that the RTS messages for setting TSO flags work, even if
it doesn't test a harsher scenario with race conditions to exercise why
the message passing is necessary for safely setting another TSO's flags.

Part of #27131

12 changed files:

+ changelog.d/T27131
rts/Interpreter.c
rts/Messages.c
rts/StgMiscClosures.cmm
rts/Threads.c
rts/Threads.h
rts/include/rts/storage/Closures.h
rts/include/stg/MiscClosures.h
+ testsuite/tests/rts/T27131.hs
+ testsuite/tests/rts/T27131.stdout
+ testsuite/tests/rts/T27131_c.c
testsuite/tests/rts/all.T

Changes:

changelog.d/T27131

 +section: rts
 +synopsis: Add rts Message to set/unset TSO flags
 +issues: #27131
 +mrs: !15831
 +description: This enables e.g. toggling breakpoints from different threads,
 +    which is necessary to safely implement features like pausing, per-thread
 +    step-in, and more in the haskell debugger.
++

rts/Interpreter.c

@@ -416,12 +416,22 @@ void rts_disableStopNextBreakpointAll(void)
  void rts_enableStopNextBreakpoint(StgTSO* tso)
+ {
 -    tso->flags |= TSO_STOP_NEXT_BREAKPOINT;
 +#if defined(THREADED_RTS)
 +  Capability* cap = rts_unsafeGetMyCapability();
 +  setThreadFlag(cap, tso, TSO_STOP_NEXT_BREAKPOINT);
 +#else
 +  tso->flags |= TSO_STOP_NEXT_BREAKPOINT;
 +#endif
+ }
  void rts_disableStopNextBreakpoint(StgTSO* tso)
+ {
 -    tso->flags &= ~TSO_STOP_NEXT_BREAKPOINT;
 +#if defined(THREADED_RTS)
 +  Capability* cap = rts_unsafeGetMyCapability();
 +  unsetThreadFlag(cap, tso, TSO_STOP_NEXT_BREAKPOINT);
 +#else
 +  tso->flags &= ~TSO_STOP_NEXT_BREAKPOINT;
 +#endif
+ }
  /* ---------------------------------------------------------------------------
@@ -430,12 +440,22 @@ void rts_disableStopNextBreakpoint(StgTSO* tso)
  void rts_enableStopAfterReturn(StgTSO* tso)
+ {
 +#if defined(THREADED_RTS)
 +  Capability* cap = rts_unsafeGetMyCapability();
 +  setThreadFlag(cap, tso, TSO_STOP_AFTER_RETURN);
 +#else
    tso->flags |= TSO_STOP_AFTER_RETURN;
 +#endif
+ }
  void rts_disableStopAfterReturn(StgTSO* tso)
+ {
 +#if defined(THREADED_RTS)
 +  Capability* cap = rts_unsafeGetMyCapability();
 +  unsetThreadFlag(cap, tso, TSO_STOP_AFTER_RETURN);
 +#else
    tso->flags &= ~TSO_STOP_AFTER_RETURN;
 +#endif
+ }
  /*

rts/Messages.c

@@ -35,7 +35,9 @@ void sendMessage(Capability *from_cap, Capability *to_cap, Message *msg)
              i != &stg_MSG_TRY_WAKEUP_info &&
              i != &stg_IND_info && // can happen if a MSG_BLACKHOLE is revoked
              i != &stg_WHITEHOLE_info &&
 -            i != &stg_MSG_CLONE_STACK_info) {
 +            i != &stg_MSG_CLONE_STACK_info &&
 +            i != &stg_MSG_SET_TSO_FLAG_info &&
 +            i != &stg_MSG_UNSET_TSO_FLAG_info) {
              barf("sendMessage: %p", i);
+         }
+     }
@@ -137,6 +139,16 @@ loop:
          MessageCloneStack *cloneStackMessage = (MessageCloneStack*) m;
          handleCloneStackMessage(cap, cloneStackMessage);
+     }
 +    else if(i == &stg_MSG_SET_TSO_FLAG_info){
 +        MessageUpdTSOFlag *u = (MessageUpdTSOFlag*) m;
 +        u->tso->flags |= u->flag;
 +        return;
 +    }
 +    else if(i == &stg_MSG_UNSET_TSO_FLAG_info){
 +        MessageUpdTSOFlag *u = (MessageUpdTSOFlag*) m;
 +        u->tso->flags &= ~u->flag;
 +        return;
 +    }
      else
+     {
          barf("executeMessage: %p", i);

rts/StgMiscClosures.cmm

@@ -855,6 +855,12 @@ INFO_TABLE_CONSTR(stg_MSG_NULL,1,0,0,PRIM,"MSG_NULL","MSG_NULL")
  INFO_TABLE_CONSTR(stg_MSG_CLONE_STACK,3,0,0,PRIM,"MSG_CLONE_STACK","MSG_CLONE_STACK")
  { ccall pbarf("stg_MSG_CLONE_STACK object (%p) entered!", R1 "ptr") never returns; }
 +INFO_TABLE_CONSTR(stg_MSG_SET_TSO_FLAG,2,1,0,PRIM,"MSG_SET_TSO_FLAG","MSG_SET_TSO_FLAG")
 +{ foreign "C" barf("stg_MSG_SET_TSO_FLAG object (%p) entered!", R1) never returns; }
++
 +INFO_TABLE_CONSTR(stg_MSG_UNSET_TSO_FLAG,2,1,0,PRIM,"MSG_UNSET_TSO_FLAG","MSG_UNSET_TSO_FLAG")
 +{ foreign "C" barf("stg_MSG_UNSET_TSO_FLAG object (%p) entered!", R1) never returns; }
++
  /* ----------------------------------------------------------------------------
     END_TSO_QUEUE

rts/Threads.c

@@ -376,6 +376,38 @@ migrateThread (Capability *from, StgTSO *tso, Capability *to)
      tryWakeupThread(from, tso);
+ }
 +/* ----------------------------------------------------------------------------
 +   {set,unset}ThreadFlag
++
 +   sets or unsets a flag in a given TSO
 +   ------------------------------------------------------------------------- */
++
 +#if defined(THREADED_RTS)
 +static void
 +updThreadFlag(Capability *from, StgTSO *tso, StgWord32 flag, const StgInfoTable* info);
++
 +void setThreadFlag(Capability *from, StgTSO *tso, StgWord32 flag)
 +{
 +    updThreadFlag(from, tso, flag, &stg_MSG_SET_TSO_FLAG_info);
 +}
++
 +void unsetThreadFlag(Capability *from, StgTSO *tso, StgWord32 flag)
 +{
 +    updThreadFlag(from, tso, flag, &stg_MSG_UNSET_TSO_FLAG_info);
 +}
++
 +static void
 +updThreadFlag(Capability *from, StgTSO *tso, StgWord32 flag, const StgInfoTable* info)
 +{
 +    MessageUpdTSOFlag *msg;
 +    msg = (MessageUpdTSOFlag *)allocate(from,sizeofW(MessageUpdTSOFlag));
 +    msg->tso  = tso;
 +    msg->flag = flag;
 +    SET_HDR_RELEASE(msg, info, CCS_SYSTEM);
 +    sendMessage(from, tso->cap, (Message*)msg);
 +}
 +#endif
++
  /* ----------------------------------------------------------------------------
     awakenBlockedQueue

rts/Threads.h

@@ -19,6 +19,11 @@ void checkBlockingQueues (Capability *cap, StgTSO *tso);
  void tryWakeupThread     (Capability *cap, StgTSO *tso);
  void migrateThread       (Capability *from, StgTSO *tso, Capability *to);
 +#if defined(THREADED_RTS)
 +void setThreadFlag       (Capability *from, StgTSO *tso, StgWord32 flag);
 +void unsetThreadFlag     (Capability *from, StgTSO *tso, StgWord32 flag);
 +#endif
++
  // Wakes up a thread on a Capability (probably a different Capability
  // from the one held by the current Task).
  //

rts/include/rts/storage/Closures.h

@@ -620,6 +620,12 @@ typedef struct MessageCloneStack_ {
      StgTSO    *tso;
  } MessageCloneStack;
 +typedef struct MessageUpdTSOFlag_ {
 +    StgHeader header;
 +    Message   *link;
 +    StgTSO    *tso;
 +    StgWord   flag;
 +} MessageUpdTSOFlag;
  /* ----------------------------------------------------------------------------
     Compact Regions

rts/include/stg/MiscClosures.h

@@ -152,6 +152,8 @@ RTS_ENTRY(stg_MSG_TRY_WAKEUP);
  RTS_ENTRY(stg_MSG_THROWTO);
  RTS_ENTRY(stg_MSG_BLACKHOLE);
  RTS_ENTRY(stg_MSG_CLONE_STACK);
 +RTS_ENTRY(stg_MSG_SET_TSO_FLAG);
 +RTS_ENTRY(stg_MSG_UNSET_TSO_FLAG);
  RTS_ENTRY(stg_MSG_NULL);
  RTS_ENTRY(stg_MVAR_TSO_QUEUE);
  RTS_ENTRY(stg_catch);

testsuite/tests/rts/T27131.hs

 +{-# LANGUAGE MagicHash #-}
 +{-# LANGUAGE UnliftedFFITypes #-}
++
 +module Main where
++
 +import Control.Concurrent
 +import Control.Monad
 +import Foreign.C.Types
 +import GHC.Conc.Sync (ThreadId(..), forkOn, myThreadId, setNumCapabilities)
 +import GHC.Exts (ThreadId#)
++
 +foreign import ccall unsafe "rts_enableStopNextBreakpoint"
 +  rts_enableStopNextBreakpoint :: ThreadId# -> IO ()
++
 +foreign import ccall unsafe "rts_disableStopNextBreakpoint"
 +  rts_disableStopNextBreakpoint :: ThreadId# -> IO ()
++
 +foreign import ccall unsafe "rts_enableStopAfterReturn"
 +  rts_enableStopAfterReturn :: ThreadId# -> IO ()
++
 +foreign import ccall unsafe "rts_disableStopAfterReturn"
 +  rts_disableStopAfterReturn :: ThreadId# -> IO ()
++
 +foreign import ccall unsafe "has_local_stop_next_breakpoint"
 +  c_hasLocalStopNextBreakpoint :: IO CInt
++
 +foreign import ccall unsafe "has_local_stop_after_return"
 +  c_hasLocalStopAfterReturn :: IO CInt
++
 +main :: IO ()
 +main = do
 +  setNumCapabilities 2
 +  checkFlag
 +    "TSO_STOP_NEXT_BREAKPOINT"
 +    rts_enableStopNextBreakpoint
 +    rts_disableStopNextBreakpoint
 +    c_hasLocalStopNextBreakpoint
 +  checkFlag
 +    "TSO_STOP_AFTER_RETURN"
 +    rts_enableStopAfterReturn
 +    rts_disableStopAfterReturn
 +    c_hasLocalStopAfterReturn
++
 +checkFlag
 +  :: String
 +  -> (ThreadId# -> IO ())
 +  -> (ThreadId# -> IO ())
 +  -> IO CInt
 +  -> IO ()
 +checkFlag label enable disable isMyThreadFlagSet = do
 +  -- Print the main thread's capability (should be 0)
 +  print =<< threadCapability =<< myThreadId
++
 +  -- Target thread will write its own flag value here
 +  targetCheckVar <- newEmptyMVar
++
 +  -- Run the new TSO runs on capability 1
 +  ThreadId tid# <- forkOn 1 $ do
 +    replicateM_ 2 $ do
 +      replyVar <- takeMVar targetCheckVar
 +      isSet <- (/= 0) <$> isMyThreadFlagSet
 +      putMVar replyVar isSet
++
 +  -- Enable the other TSO's flag
 +  enable tid#
 +  -- It will check whether it is set and reply here
 +  renderCheck label "set" =<< checkTarget targetCheckVar
++
 +  -- Ditto.
 +  disable tid#
 +  renderCheck label "unset" . not =<< checkTarget targetCheckVar
++
 +checkTarget :: MVar (MVar Bool) -> IO Bool
 +checkTarget targetCheckVar = do
 +  replyVar <- newEmptyMVar
 +  putMVar targetCheckVar replyVar
 +  takeMVar replyVar
++
 +renderCheck :: String -> String -> Bool -> IO ()
 +renderCheck label state ok = putStrLn $
 +  label ++ " " ++ state ++ ": " ++ if ok then "ok" else "failed"

testsuite/tests/rts/T27131.stdout

 +(0,False)
 +TSO_STOP_NEXT_BREAKPOINT set: ok
 +TSO_STOP_NEXT_BREAKPOINT unset: ok
 +(0,False)
 +TSO_STOP_AFTER_RETURN set: ok
 +TSO_STOP_AFTER_RETURN unset: ok

testsuite/tests/rts/T27131_c.c

 +#include "Rts.h"
++
 +int has_local_stop_next_breakpoint(void)
 +{
 +    CapabilityPublic *cap = (CapabilityPublic *) rts_unsafeGetMyCapability();
 +    StgTSO *tso = cap->r.rCurrentTSO;
 +    return (tso->flags & TSO_STOP_NEXT_BREAKPOINT) != 0;
 +}
++
 +int has_local_stop_after_return(void)
 +{
 +    CapabilityPublic *cap = (CapabilityPublic *) rts_unsafeGetMyCapability();
 +    StgTSO *tso = cap->r.rCurrentTSO;
 +    return (tso->flags & TSO_STOP_AFTER_RETURN) != 0;
 +}

testsuite/tests/rts/all.T

@@ -623,6 +623,13 @@ test('T20201b', [js_skip, exit_code(1)], compile_and_run, ['-with-rtsopts -A64z'
  test('T22012', [js_skip, extra_ways(['ghci'])], compile_and_run, ['T22012_c.c'])
 +test('T27131',
 +     [ only_ways(['threaded1', 'threaded2'])
 +     , req_ghc_with_threaded_rts
 +     , req_target_smp
 +     ],
 +     compile_and_run, ['T27131_c.c'])
++
  # Skip for JS platform as the JS RTS is always single threaded
  test('T22795a', [only_ways(['normal']), js_skip, req_ghc_with_threaded_rts], compile_and_run, ['-threaded'])
  test('T22795b', [only_ways(['normal']), js_skip], compile_and_run, ['-single-threaded'])