Marge Bot pushed to branch master at Glasgow Haskell Compiler / GHC Commits: bb324171 by Rodrigo Mesquita at 2026-08-26T15:10:59-04:00 rts: refactor to reduce THREADED_RTS in MSG_UPD_TSO_FLAGS - No behavior change in this commit (well, a small optimization here makes us do less work if the target TSO owned by the curr. capability) - Move all THREADED_RTS CPP needed into `updThreadFlag` - Merge MSG_SET_TSO_FLAGS and MSG_UNSET_TSO_FLAGS into MSG_UPD_TSO_FLAGS plus a `set` bool field in the MessageUpdTSOFlag struct Towards #27729 - - - - - ed99b7b7 by Rodrigo Mesquita at 2026-08-26T15:10:59-04:00 rts: Fix race condition in MSG_UPD_TSO_FLAGS execution The code for processing the MSG_UPD_TSO_FLAGS message was not taking into consideration that the TSO's owner might have moved in between that capability receiving the message (since it was its previous owner) and starting to process its inbox (a point at which it was no longer the owner) Added Note [TSO owner may change in between Msg being sent and received] to explain this race and the pattern used to fix this, where we just forward the message to the new owner. Fixes #27729 - - - - - 8 changed files: - rts/CloneStack.c - 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 Changes: ===================================== rts/CloneStack.c ===================================== @@ -88,6 +88,7 @@ void sendCloneStackMessage(StgTSO *tso, HsStablePtr mvar) { void handleCloneStackMessage(Capability *cap, MessageCloneStack *msg){ // We must check that the current owner of the thread we want to clone the stack for // is still this capability. + // See Note [TSO owner may change in between Msg being sent and received] Capability *owner = RELAXED_LOAD(&msg->tso->cap); if (owner != cap) { // The target TSO may have migrated after the message was queued on the old ===================================== rts/Interpreter.c ===================================== @@ -416,22 +416,14 @@ void rts_disableStopNextBreakpointAll(void) void rts_enableStopNextBreakpoint(StgTSO* tso) { -#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) { -#if defined(THREADED_RTS) Capability* cap = rts_unsafeGetMyCapability(); unsetThreadFlag(cap, tso, TSO_STOP_NEXT_BREAKPOINT); -#else - tso->flags &= ~TSO_STOP_NEXT_BREAKPOINT; -#endif } /* --------------------------------------------------------------------------- @@ -440,22 +432,14 @@ 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 ===================================== @@ -36,8 +36,7 @@ void sendMessage(Capability *from_cap, Capability *to_cap, Message *msg) 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_SET_TSO_FLAG_info && - i != &stg_MSG_UNSET_TSO_FLAG_info) { + i != &stg_MSG_UPD_TSO_FLAG_info) { barf("sendMessage: %p", i); } } @@ -67,6 +66,62 @@ void sendMessage(Capability *from_cap, Capability *to_cap, Message *msg) Handle a message ------------------------------------------------------------------------- */ +/* +Note [TSO owner may change in between Msg being sent and received] +~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ +When a message is sent from Capability (C1) to a target TSO (T2) (e.g. +MessageUpdTSOFlag, MessageCloneStack, ...), it is queued on the TSO's owner +Capability (C3) inbox (inboxes are owned by Capabilities, not TSOs). + +At a later point, the Capability (C3) will process its inbox. Upon receiving +the message meant for a specific TSO (T2), it must first always check that the +TSO's owner is *still* itself (C3). + +The target TSO (T2) may have migrated after the message was queued on its old +capability (C3). In that case we must forward the request to the new owner +(say, C4); otherwise the Capability C3 could be modifying a TSO it no longer +owns, racing with its actual owner mutating it, since it is no longer the owner. + +The message meant for a TSO should only be executed when the receiving +Capability is still the owner of that TSO. Otherwise, it must be forwarded to +the new owner. + +The general pattern is one where there's a top-level function which assumes it +can be called by capabilities other than the TSO's owner. The function checks +whether the current capability is the TSO owner. If yes, execute the action. If +not, then it sends a message to the current TSO's owner. On receiving the +message, the new capability will just call that top-level function, which will +ensure the message is forwarded again if the TSO owner changed. +It will look something like: + + runMyMsg(Capability *from, StgTSO *target, ...) { + +#if defined(THREADED_RTS) + Capability *owner = RELAXED_LOAD(&target->cap) + if (owner != from) { + MessageMyMsg* msg = ... + sendMessage(cap, owner, msg) + return + } +#endif + + actuallyDoTheWork(...) + } + + executeMessage(...) { + + if (i == &stg_MY_MSG_info) { + + MessageMyMsg* msg = (MessageMyMsg*) m + runMyMsg(cap, m->tso, ...) + + } + } + +See example `updThreadFlag` and `executeMessage`'s `stg_MSG_UPD_TSO_FLAG_info`, +or `tryWakeUpThread` and `stg_MSG_TRY_WAKEUP_info` for two live examples. +*/ + #if defined(THREADED_RTS) void @@ -141,15 +196,11 @@ loop: MessageCloneStack *cloneStackMessage = (MessageCloneStack*) m; handleCloneStackMessage(cap, cloneStackMessage); } - else if(i == &stg_MSG_SET_TSO_FLAG_info){ + else if(i == &stg_MSG_UPD_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; + + StgTSO *tso = RELAXED_LOAD(&u->tso); + updThreadFlag(cap, tso, u->flag, u->set); } else { ===================================== rts/StgMiscClosures.cmm ===================================== @@ -855,11 +855,8 @@ 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; } +INFO_TABLE_CONSTR(stg_MSG_UPD_TSO_FLAG,2,2,0,PRIM,"MSG_UPD_TSO_FLAG","MSG_UPD_TSO_FLAG") +{ foreign "C" barf("stg_MSG_UPD_TSO_FLAG object (%p) entered!", R1) never returns; } /* ---------------------------------------------------------------------------- END_TSO_QUEUE ===================================== rts/Threads.c ===================================== @@ -379,32 +379,46 @@ migrateThread (Capability *from, StgTSO *tso, Capability *to) 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); + updThreadFlag(from, tso, flag, true); } void unsetThreadFlag(Capability *from, StgTSO *tso, StgWord32 flag) { - updThreadFlag(from, tso, flag, &stg_MSG_UNSET_TSO_FLAG_info); + updThreadFlag(from, tso, flag, false); } -static void -updThreadFlag(Capability *from, StgTSO *tso, StgWord32 flag, const StgInfoTable* info) +void +updThreadFlag(Capability *from USED_IF_THREADS, StgTSO *tso, StgWord32 flag, StgBool set /* true=set, false=unset */) { - 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); -} +#if defined(THREADED_RTS) + // If we're the current owner of the thread we want to modify, do it. + // Otherwise, we must forward the message to the actual owner. + // When executing the upd message, we check again that we're still the TSO + // owner (which may have changed since the message was queued on this cap.) + // See Note [TSO owner may change in between Msg being sent and received] + Capability *tso_owner = RELAXED_LOAD(&tso->cap); + if (from != tso_owner) { + MessageUpdTSOFlag *msg; + msg = (MessageUpdTSOFlag *)allocate(from,sizeofW(MessageUpdTSOFlag)); + msg->tso = tso; + msg->flag = flag; + msg->set = set; + SET_HDR_RELEASE(msg, &stg_MSG_UPD_TSO_FLAG_info, CCS_SYSTEM); + sendMessage(from, tso_owner, (Message*)msg); + return; + } #endif + if (set) { + tso->flags |= flag; + } + else { + tso->flags &= ~flag; + } +} + /* ---------------------------------------------------------------------------- awakenBlockedQueue ===================================== rts/Threads.h ===================================== @@ -19,10 +19,9 @@ 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 +void updThreadFlag (Capability *from, StgTSO *tso, StgWord32 flag, StgBool set); // 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 ===================================== @@ -625,6 +625,7 @@ typedef struct MessageUpdTSOFlag_ { Message *link; StgTSO *tso; StgWord flag; + StgWord set; // bool: true=SET; false=UNSET } MessageUpdTSOFlag; /* ---------------------------------------------------------------------------- ===================================== rts/include/stg/MiscClosures.h ===================================== @@ -151,8 +151,7 @@ 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_UPD_TSO_FLAG); RTS_ENTRY(stg_MSG_NULL); RTS_ENTRY(stg_MVAR_TSO_QUEUE); RTS_ENTRY(stg_catch); View it on GitLab: https://gitlab.haskell.org/ghc/ghc/-/compare/fd22f71ee92595f4634206455d8eab1... -- View it on GitLab: https://gitlab.haskell.org/ghc/ghc/-/compare/fd22f71ee92595f4634206455d8eab1... You're receiving this email because of your account on gitlab.haskell.org. Manage all notifications: https://gitlab.haskell.org/-/profile/notifications | Help: https://gitlab.haskell.org/help