Erik de Castro Lopo pushed to branch wip/fix-9809 at Glasgow Haskell Compiler / GHC

Commits:

5 changed files:

Changes:

  • changelog.d/T9809-timer-manager-contention
    1
    +section: ghc-internal
    
    2
    +synopsis: Fix excessive GC, memory use and timer latency when many threads
    
    3
    +  register timeouts at once with ``threadDelay``, ``registerDelay`` or
    
    4
    +  ``System.Timeout.timeout``.
    
    5
    +issues: #9809
    
    6
    +mrs: !16863
    
    7
    +description: {
    
    8
    +    Threads registering timeouts now publish only fully evaluated updates to
    
    9
    +    the timer manager's timeout queue, using a compare-and-swap loop, instead
    
    10
    +    of building chains of unevaluated updates that other threads then had to
    
    11
    +    evaluate or block on.
    
    12
    +}

  • libraries/base/tests/.gitignore
    ... ... @@ -275,4 +275,5 @@
    275 275
     /weak001
    
    276 276
     /T9395
    
    277 277
     /T9532
    
    278
    +/T9809
    
    278 279
     /T11555

  • libraries/base/tests/T9809.hs
    1
    +-- #9809: when many threads registered timeouts at the same time, the
    
    2
    +-- TimerManager published unevaluated timeout queues with atomicModifyIORef'.
    
    3
    +-- The threads then built long chains of thunks and blocked on each other's
    
    4
    +-- black holes, and the heap grew to hundreds of megabytes. all.T runs this
    
    5
    +-- test with a heap limit that the old code exceeded.
    
    6
    +
    
    7
    +import Control.Concurrent
    
    8
    +import Control.Monad
    
    9
    +import Data.IORef
    
    10
    +import GHC.Conc (TVar, atomically, newTVarIO, readTVar, registerDelay, retry, writeTVar)
    
    11
    +
    
    12
    +main :: IO ()
    
    13
    +main = do
    
    14
    +    let n = 100000
    
    15
    +    ref <- newIORef []
    
    16
    +    registered <- newTVarIO (0 :: Int)
    
    17
    +    replicateM_ n . forkIO $ do
    
    18
    +        tv <- registerDelay 10
    
    19
    +        atomicModifyIORef' ref (\tvs -> (tv : tvs, ()))
    
    20
    +        atomically $ readTVar registered >>= writeTVar registered . (+ 1)
    
    21
    +    atomically $ do
    
    22
    +        r <- readTVar registered
    
    23
    +        when (r /= n) retry
    
    24
    +    readIORef ref >>= mapM_ waitFor
    
    25
    +
    
    26
    +-- | Block until a delay has fired, so that the test also checks that the
    
    27
    +-- TimerManager fires every timeout.
    
    28
    +waitFor :: TVar Bool -> IO ()
    
    29
    +waitFor tv = atomically $ readTVar tv >>= \fired -> unless fired retry

  • libraries/base/tests/all.T
    ... ... @@ -193,6 +193,12 @@ test('T7653', [when(opsys('mingw32'), skip),
    193 193
                    omit_ways(prof_ways + ghci_ways)], compile_and_run, [''])
    
    194 194
     test('T7787', normal, compile_and_run, [''])
    
    195 195
     
    
    196
    +# Before the fix for #9809 this needed more than 256MB of heap.
    
    197
    +test('T9809', [req_ghc_smp,
    
    198
    +               req_target_smp,
    
    199
    +               only_ways(['threaded1', 'threaded2', 'nonmoving_thr'])],
    
    200
    +               compile_and_run, ['-threaded -with-rtsopts="-N2 -M128m"'])
    
    201
    +
    
    196 202
     test('topHandler01', [when(opsys('mingw32'), skip), js_broken(22261), req_process], compile_and_run, [''])
    
    197 203
     test('topHandler02',
    
    198 204
          [when(opsys('mingw32'), skip),
    

  • libraries/ghc-internal/src/GHC/Internal/Event/TimerManager.hs
    ... ... @@ -13,8 +13,10 @@ import GHC.Internal.Types ()
    13 13
     {-# LANGUAGE CPP #-}
    
    14 14
     {-# LANGUAGE ExistentialQuantification #-}
    
    15 15
     {-# LANGUAGE FlexibleInstances #-}
    
    16
    +{-# LANGUAGE MagicHash #-}
    
    16 17
     {-# LANGUAGE NoImplicitPrelude #-}
    
    17 18
     {-# LANGUAGE Trustworthy #-}
    
    19
    +{-# LANGUAGE UnboxedTuples #-}
    
    18 20
     
    
    19 21
     -- TODO: use the new Windows IO manager
    
    20 22
     module GHC.Internal.Event.TimerManager
    
    ... ... @@ -53,6 +55,9 @@ import GHC.Internal.Control.Exception (finally)
    53 55
     import GHC.Internal.Data.Foldable (sequence_)
    
    54 56
     import GHC.Internal.Data.IORef (IORef, atomicModifyIORef', mkWeakIORef, newIORef, readIORef,
    
    55 57
                        writeIORef)
    
    58
    +import GHC.Internal.IORef (IORef(..))
    
    59
    +import GHC.Internal.Prim (casMutVar#, readMutVar#)
    
    60
    +import GHC.Internal.STRef (STRef(..))
    
    56 61
     import GHC.Internal.Err (errorWithoutStackTrace)
    
    57 62
     import GHC.Internal.Clock (getMonotonicTimeNSec)
    
    58 63
     import GHC.Internal.Conc.Signal (runHandlers)
    
    ... ... @@ -260,7 +265,7 @@ updateTimeout mgr (TK key) us = do
    260 265
     
    
    261 266
     editTimeouts :: TimerManager -> TimeoutEdit -> IO ()
    
    262 267
     editTimeouts mgr g = do
    
    263
    -  wake <- atomicModifyIORef' (emTimeouts mgr) f
    
    268
    +  wake <- modifyTimeouts (emTimeouts mgr) f
    
    264 269
       when wake (wakeManager mgr)
    
    265 270
       where
    
    266 271
         f q = (q', wake)
    
    ... ... @@ -276,4 +281,75 @@ editTimeouts mgr g = do
    276 281
                           t0 /= t1
    
    277 282
                         _ -> True
    
    278 283
     
    
    284
    +-- | Atomically replace the timeout queue with the first component of
    
    285
    +-- @f old@ and return the second component.
    
    286
    +--
    
    287
    +-- Unlike 'atomicModifyIORef'', the new queue is evaluated /before/ it is
    
    288
    +-- published, using a compare-and-swap loop. 'atomicModifyIORef'' installs
    
    289
    +-- the new contents as a thunk and only evaluates it afterwards, so when many
    
    290
    +-- threads register timeouts at once (#9809) each one builds its insert on
    
    291
    +-- top of the previous thread's unevaluated insert, giving long thunk chains,
    
    292
    +-- threads blocking on each other's half evaluated updates and lots of extra
    
    293
    +-- GC work. Since 'Q.IntPSQ' is spine strict, WHNF means fully evaluated, and
    
    294
    +-- other threads only ever see a fully evaluated queue.
    
    295
    +--
    
    296
    +-- The cost is that a thread whose compare-and-swap loses a race has to redo
    
    297
    +-- its O(log n) update. The update runs in constant stack (see #7653).
    
    298
    +--
    
    299
    +-- Only the short edits made by registering threads use this. The timer
    
    300
    +-- manager's own update in 'mkTimeout' (removing every expired timeout) can
    
    301
    +-- take much longer, and under a stream of registrations from other
    
    302
    +-- capabilities its compare-and-swap would almost never succeed, starving it
    
    303
    +-- so that expired timeouts don't fire. It therefore keeps using
    
    304
    +-- 'atomicModifyIORef'', which never retries: it publishes a single thunk and
    
    305
    +-- evaluates it straight away. A registering thread that reads that thunk
    
    306
    +-- evaluates it (or waits for the manager to) before publishing its own fully
    
    307
    +-- evaluated queue, so thunks still cannot pile up.
    
    308
    +--
    
    309
    +-- 'casMutVar#' compares pointers, so the expected value must be exactly the
    
    310
    +-- pointer read from the 'MutVar#'. See Note [Pointer equality in
    
    311
    +-- modifyTimeouts] for why this function is OPAQUE.
    
    312
    +{-# OPAQUE modifyTimeouts #-}
    
    313
    +modifyTimeouts :: IORef TimeoutQueue -> (TimeoutQueue -> (TimeoutQueue, b)) -> IO b
    
    314
    +modifyTimeouts (IORef (STRef ref)) f = IO modLoop
    
    315
    +  where
    
    316
    +    modLoop s0 = case readMutVar# ref s0 of
    
    317
    +      (# s1, oldVar #) -> case f oldVar of
    
    318
    +        (newVar, r) -> newVar `seq` case casMutVar# ref oldVar newVar s1 of
    
    319
    +          (# s2, 0#, _ #) -> (# s2, r #)
    
    320
    +          (# s2, _, _ #) -> modLoop s2
    
    321
    +
    
    322
    +{- Note [Pointer equality in modifyTimeouts]
    
    323
    +~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
    
    324
    +'casMutVar#' succeeds only if the expected value is the very pointer stored
    
    325
    +in the 'MutVar#', so all the difficulties of 'reallyUnsafePtrEquality#'
    
    326
    +apply to it. In 'modifyTimeouts' the expected value is @oldVar@, exactly as
    
    327
    +'readMutVar#' returned it, and that must survive optimisation.
    
    328
    +
    
    329
    +The danger is the optimiser seeing a @case@ on @oldVar@ in the same function
    
    330
    +as the 'casMutVar#'. That would happen if 'modifyTimeouts' were inlined or
    
    331
    +specialised at a call site, bringing in the body of @f@ (in 'editTimeouts',
    
    332
    +@f@ inspects the queue to decide whether to wake the manager). The binder
    
    333
    +swap (Note [Binder swap] in GHC.Core.Opt.OccurAnal) may then replace @oldVar@
    
    334
    +by the case binder, which is the /evaluated/ queue. That is a different
    
    335
    +pointer whenever the 'MutVar#' holds a thunk or an indirection, for example
    
    336
    +the selector thunk that 'atomicModifyIORef'' in 'mkTimeout' leaves behind.
    
    337
    +Case-of-known-constructor can likewise replace it by a freshly built
    
    338
    +constructor. Either way the compare-and-swap would keep failing and the
    
    339
    +registering thread would spin.
    
    340
    +
    
    341
    +The OPAQUE pragma keeps @f@ an unknown function inside 'modifyTimeouts', so
    
    342
    +there is never a @case@ on @oldVar@ to rewrite. OPAQUE rather than NOINLINE:
    
    343
    +both stop inlining and specialisation, but NOINLINE still allows the
    
    344
    +worker/wrapper transformation. Today that is harmless (the worker only
    
    345
    +unboxes the 'IORef' and still takes @f@ as an argument), but OPAQUE states
    
    346
    +the requirement directly: the body must be compiled as written and every call
    
    347
    +must go to it. It also costs nothing, since 'editTimeouts' passes a closure
    
    348
    +anyway.
    
    349
    +
    
    350
    +An alternative is to hide the expected pointer in a box built by a NOINLINE
    
    351
    +function before @f@ runs (like the @Ticket@ type of the atomic-primops
    
    352
    +package), but that allocates on every attempt.
    
    353
    +-}
    
    354
    +
    
    279 355
     #endif