diff --git a/ChangeLog.md b/ChangeLog.md --- a/ChangeLog.md +++ b/ChangeLog.md @@ -1,5 +1,10 @@ # Changelog for thread-utils-context +## 0.4.1.1 + +- Fix `purgeDeadThreads` retaining finished threads and evicting live ones. +- Stop `purgeDeadThreads` from retaining stale TSO pointers after collection. + ## 0.4.1.0 - Fix space leak: repeated `attach`/`detach` on long-lived threads no longer diff --git a/cbits/simd_search.c b/cbits/simd_search.c --- a/cbits/simd_search.c +++ b/cbits/simd_search.c @@ -23,9 +23,19 @@ * `sorted` must be in ascending order. The comparison `base[half] < * needle` compiles to CMOV on both x86-64 and AArch64 at -O2, so no * branch mispredictions. + * + * The loop narrows to a lower-bound *candidate*, not to the answer: on + * exit the lower bound is either `base` or `base + 1`. Lemire's original + * returns the index `(*base < target) + (base - source)` precisely to + * account for that. Both slots must therefore be compared here -- + * testing only `*base` reports "absent" for almost every element that is + * actually present (e.g. 199 of 200 for a contiguous run), because the + * loop most often stops one short. * ------------------------------------------------------------------- */ static inline int contains_bsearch(HsInt needle, const HsInt *sorted, HsInt n) { + if (n <= 0) + return 0; const HsInt *base = sorted; HsInt len = n; while (len > 1) { @@ -33,7 +43,10 @@ base += (base[half] < needle) ? half : 0; len -= half; } - return (n > 0) && (*base == needle); + if (*base == needle) + return 1; + const HsInt *next = base + 1; + return (next < sorted + n) && (*next == needle); } /* ------------------------------------------------------------------- diff --git a/src/Control/Concurrent/Thread/Storage.hs b/src/Control/Concurrent/Thread/Storage.hs --- a/src/Control/Concurrent/Thread/Storage.hs +++ b/src/Control/Concurrent/Thread/Storage.hs @@ -148,15 +148,14 @@ import Data.IORef import Foreign.C.Types (CULLong (..)) import Foreign.Storable (sizeOf) -import GHC.Base (Addr#) import GHC.Conc (getNumCapabilities, yield) import GHC.Conc.Sync (ThreadId (..)) -import GHC.Exts (Int (..), Int#, isTrue#, unsafeCoerce#, (==#), (>=#)) +import GHC.Exts (Int (..), Int#, ThreadId#, isTrue#, unsafeCoerce#, (==#), (>=#)) import qualified GHC.Exts as Exts import GHC.IO (IO (..)) import System.IO.Unsafe (unsafePerformIO) #if MIN_VERSION_base(4,18,0) -import GHC.Conc (listThreads) +import GHC.Conc (ThreadStatus (..), listThreads, threadStatus) #endif import Prelude hiding (lookup) @@ -202,7 +201,21 @@ {-# INLINE getCurrentThreadId #-} -foreign import ccall unsafe "rts_getThreadId" c_getThreadId :: Addr# -> CULLong +-- | @rts_getThreadId@ takes the TSO pointer behind a 'ThreadId'. +-- +-- The argument MUST be declared as 'ThreadId#' rather than coerced to +-- 'Addr#'. A 'ThreadId#' is an ordinary movable heap pointer: GHC's +-- generational collector relocates TSOs when it promotes them. Declaring +-- it as 'ThreadId#' keeps it in a pointer slot, so the collector traces +-- and updates it, and (because the call is @unsafe@) no GC can run +-- between the argument being read and the callee dereferencing it. +-- +-- Coercing to 'Addr#' launders the pointer into a non-pointer slot that +-- the collector neither traces nor updates. If a GC lands while the +-- laundered word is live, the callee dereferences a stale TSO address and +-- the process segfaults. This is the same signature @base@ uses in +-- "GHC.Conc.Sync". +foreign import ccall unsafe "rts_getThreadId" c_getThreadId :: ThreadId# -> CULLong -- | Extract the numeric thread ID from an existing 'ThreadId'. @@ -211,12 +224,12 @@ -- 'ThreadId' and need its numeric form for 'lookupRaw' or 'updateRaw', use -- this. Otherwise prefer 'getCurrentThreadId'. getThreadId :: ThreadId -> Word -getThreadId (ThreadId tid#) = fromIntegral (c_getThreadId (unsafeCoerce# tid#)) +getThreadId (ThreadId tid#) = fromIntegral (c_getThreadId tid#) {-# INLINE getThreadId #-} getThreadIdInt :: ThreadId -> Int -getThreadIdInt (ThreadId tid#) = fromIntegral (c_getThreadId (unsafeCoerce# tid#)) +getThreadIdInt (ThreadId tid#) = fromIntegral (c_getThreadId tid#) {-# INLINE getThreadIdInt #-} @@ -1131,19 +1144,32 @@ s' -> (# s', () #) --- | Fill a 'MutIntArray' with numeric thread IDs from a @['ThreadId']@. +-- | Fill a 'MutIntArray' with the numeric IDs of the threads that can still +-- run, returning how many were written. +-- +-- Threads whose 'threadStatus' is 'ThreadFinished' or 'ThreadDied' are +-- skipped. 'listThreads' enumerates the RTS generation thread lists, and a +-- TSO is only unlinked from those by a GC that collects its generation -- +-- so a thread that has exited keeps being listed until then, and once it has +-- been promoted, until the next /major/ GC. +-- -- The array is left unsorted; the C-side 'c_purge_find_dead' sorts it -- in place via @qsort@ before scanning. buildLiveSet :: [ThreadId] -> IO (MutIntArray, Int) buildLiveSet tids = do let !n = length tids arr <- newMutIntArray (max 1 n) - let fill [] _ = pure () + let fill [] !i = pure i fill (t : ts) !i = do - writeMutInt arr i (getThreadIdInt t) - fill ts (i + 1) - fill tids 0 - pure (arr, n) + status <- threadStatus t + case status of + ThreadFinished -> fill ts i + ThreadDied -> fill ts i + _ -> do + writeMutInt arr i (getThreadIdInt t) + fill ts (i + 1) + nLive <- fill tids 0 + pure (arr, nLive) -- | Batch membership scan implemented in C with architecture-dispatched @@ -1174,8 +1200,9 @@ -- -- Normally, slots are cleaned up by GC finalizers attached to the owning -- 'ThreadId'. This function provides an eager alternative: it calls --- 'GHC.Conc.listThreads' to obtain the set of live threads and tombstones --- any slot whose key is not in that set. +-- 'GHC.Conc.listThreads', discards the entries that have already finished or +-- died (a 'listThreads' result keeps naming exited threads until a GC unlinks +-- their TSOs), and tombstones any slot whose key is not in what remains. -- -- Internally builds a flat array of live thread IDs and passes it to a -- C function that @qsort@s it, then batch-scans the key array using diff --git a/test/Spec.hs b/test/Spec.hs --- a/test/Spec.hs +++ b/test/Spec.hs @@ -1,4 +1,5 @@ {-# LANGUAGE BangPatterns #-} +{-# LANGUAGE CPP #-} {-# LANGUAGE NumericUnderscores #-} import System.Mem import Control.Concurrent @@ -226,6 +227,75 @@ -- Fixed code => bounded constant (one Weak# per thread). let growth = fromIntegral afterLive - fromIntegral beforeLive :: Int growth `shouldSatisfy` (< 1_000_000) + +#if MIN_VERSION_base(4,18,0) + describe "purgeDeadThreads" $ do + it "reclaims entries for exited threads without waiting for GC" $ do + let n = 500 + gate <- newEmptyMVar + doneRef <- newIORef (0 :: Int) + tsm <- newThreadStorageMap + replicateM_ n $ forkIO $ do + attach tsm () + readMVar gate + atomicModifyIORef' doneRef (\x -> (x + 1, ())) + + waitForCount tsm n + putMVar gate () + spinUntil $ (>= n) <$> readIORef doneRef + + -- The whole point of purgeDeadThreads is to reclaim eagerly rather + -- than waiting on GC finalizers, so spin without performGC here. + spinUntil $ do + purgeDeadThreads tsm + null <$> storedItems tsm + + leftovers <- storedItems tsm + leftovers `shouldBe` [] + + it "keeps entries for threads that are still alive" $ do + let n = 200 + gate <- newEmptyMVar + tsm <- newThreadStorageMap + replicateM_ n $ forkIO $ do + attach tsm () + readMVar gate + + waitForCount tsm n + purgeDeadThreads tsm + survivors <- storedItems tsm + putMVar gate () + length survivors `shouldBe` n + + -- Regression test for a laundered TSO pointer in getThreadIdInt. + -- + -- purgeDeadThreads calls listThreads and converts every ThreadId in the + -- process to its numeric id. That conversion used to coerce ThreadId# to + -- Addr# before handing it to rts_getThreadId. ThreadId# is a movable heap + -- pointer, so once laundered into a non-pointer slot the collector would + -- neither trace nor relocate it; a GC landing mid-traversal left the C + -- call dereferencing a stale TSO and the process segfaulted. + -- + -- This is a race, so it is probabilistic rather than deterministic: many + -- live threads (a long listThreads result) interleaved with forced GCs + -- is the shape that reproduces it. + it "tolerates GC relocating TSOs during the live-thread scan" $ do + let n = 2_000 + gate <- newEmptyMVar + tsm <- newThreadStorageMap + replicateM_ n $ forkIO $ do + attach tsm (0 :: Int) + readMVar gate + + waitForCount tsm n + replicateM_ 100 $ do + performGC + purgeDeadThreads tsm + + survivors <- storedItems tsm + putMVar gate () + length survivors `shouldBe` n +#endif waitForCount :: ThreadStorageMap a -> Int -> IO () diff --git a/thread-utils-context.cabal b/thread-utils-context.cabal --- a/thread-utils-context.cabal +++ b/thread-utils-context.cabal @@ -1,7 +1,7 @@ cabal-version: 3.0 name: thread-utils-context -version: 0.4.1.0 +version: 0.4.1.1 synopsis: Garbage-collected thread local storage description: Please see the README on GitHub at category: Concurrency