thread-utils-context 0.4.1.0 → 0.4.1.1

raw patch · 5 files changed

+131/−16 lines, 5 filesPVP ok

version bump matches the API change (PVP)

API changes (from Hackage documentation)

Files

• ChangeLog.md +5/−0
• cbits/simd_search.c +14/−1
• src/Control/Concurrent/Thread/Storage.hs +41/−14
• test/Spec.hs +70/−0
• thread-utils-context.cabal +1/−1

@@ -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

@@ -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); }  /* -------------------------------------------------------------------

@@ -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

@@ -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 ()

@@ -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 <https://github.com/iand675/thread-utils-context#readme> category:       Concurrency