sheaf pushed to branch wip/romes/27514 at Glasgow Haskell Compiler / GHC

Commits:

3 changed files:

Changes:

  • compiler/GHC/Driver/Concurrency.hs
    ... ... @@ -12,7 +12,8 @@ module GHC.Driver.Concurrency
    12 12 
         -- * Concurrent worker scheduling
    13 13 
       , ConcurrentWorkerEnv(..)
    14 14 
       , mapConcurrentWorkers
    15 
    -  , concurrentTraversal_DF
    15 
    +  , NodeExpander(..)
    16 
    +  , concurrentTraversal
    16 17 
       )
    17 18 
       where
    18 19
    ... ... @@ -20,12 +21,16 @@ import GHC.Prelude
    20 21
    21 22 
     import GHC.Driver.MakeSem
    22 23 
     import GHC.Driver.Pipeline.LogQueue
    23 
    -  ( LogQueueQueue, finishLogQueue, initLogQueue, logThread
    24 
    -  , newLogQueue, newLogQueueQueue, parLogAction )
    24 
    +  ( LogQueue, LogQueueQueue, finishLogQueue, initLogQueue, logThread
    25 
    +  , newLogQueue, newLogQueueQueue, parLogAction, printLogs )
    25 26 
     import GHC.Utils.Logger
    26 27 
       ( Logger, makeThreadSafe, pushLogHook )
    28 
    +import GHC.Utils.Misc
    29 
    +  ( HasDebugCallStack )
    30 
    +import GHC.Utils.Outputable
    31 
    +  ( Outputable(..), text, (<+>) )
    27 32 
     import GHC.Utils.Panic
    28 
    -  ( panic )
    33 
    +  ( massertPpr, pprPanic )
    29 34 
     import GHC.Utils.TmpFs
    30 35 
       ( TmpFs, forkTmpFsFrom, mergeTmpFsInto, withLocalTmpFS )
    31 36
    ... ... @@ -36,31 +41,30 @@ import System.Semaphore
    36 41 
     import Control.Concurrent
    37 42 
       ( ThreadId, forkIOWithUnmask, killThread, myThreadId )
    38 43 
     import Control.Concurrent.MVar
    39 
    -  ( MVar, newEmptyMVar, newMVar, putMVar, takeMVar )
    44 
    +  ( MVar, newMVar, putMVar, takeMVar )
    40 45 
     import GHC.Conc
    41 46 
       ( labelThread )
    42 47 
     #else
    43 48 
     import Control.Concurrent
    44 49 
       ( ThreadId, forkIOWithUnmask, killThread, myThreadId
    45 
    -  , newQSem, signalQSem, waitQSem, MVar, takeMVar, putMVar, newEmptyMVar )
    46 
    -import Control.Monad
    47 
    -  ( unless )
    50 
    +  , newQSem, signalQSem, waitQSem )
    48 51 
     import qualified Control.Monad.Catch as MC
    49 52 
     import GHC.Conc
    50 53 
       ( getNumCapabilities, getNumProcessors, labelThread, setNumCapabilities )
    51 54 
     #endif
    52 55 
     import Control.Concurrent.STM
    53 
    -  ( TVar, atomically, check, modifyTVar', newTVarIO, readTVar, writeTVar )
    56 
    +  ( TVar, atomically, check, modifyTVar', newTVarIO, readTVar, readTVarIO
    57 
    +  , writeTVar )
    54 58 
     import Control.Exception
    55 59 
       ( AsyncException(ThreadKilled), SomeAsyncException, SomeException
    56 
    -  , finally, fromException, mask, mask_, onException
    60 
    +  , catch, finally, fromException, mask, mask_, onException
    57 61 
       , throwIO, try, uninterruptibleMask_ )
    58 62 
     import Control.Monad
    59 
    -  ( replicateM )
    63 
    +  ( unless )
    60 64 
     import Data.Foldable
    61 
    -  ( for_ )
    65 
    +  ( for_, traverse_ )
    62 66 
     import Data.IORef
    63 
    -  ( IORef, atomicModifyIORef', modifyIORef', newIORef, readIORef, writeIORef )
    67 
    +  ( IORef, atomicModifyIORef', modifyIORef', newIORef, readIORef )
    64 68 
     import qualified Data.Map as Map
    65 69 
     import qualified Data.Sequence as Seq
    66 70 
     import qualified Data.Set as Set
    ... ... @@ -211,72 +215,163 @@ withWorkerLimit logger tmpfs report_semaphore_failure limit action
    211 215 
               action conc parent_work_env
    212 216
    213 217 
     --------------------------------------------------------------------------------
    214 
    --- * Scheduling concurrent workers
    218 
    +-- * Monotone data structures
    215 219 
     --------------------------------------------------------------------------------
    216 220
    217 
    --- | Internal scheduler abstraction with two capabilities:
    221 
    +-- | A map that only ever grows, and whose entries are written at most once.
    222 
    +newtype MonotoneMap k v = MonotoneMap ( IORef ( Map.Map k v ) )
    223 
    +
    224 
    +newMonotoneMap :: Map.Map k v -> IO ( MonotoneMap k v )
    225 
    +newMonotoneMap initial = MonotoneMap <$> newIORef initial
    226 
    +
    227 
    +-- | The outcome of inserting into a 'MonotoneMap' or a 'MonotoneSet'.
    228 
    +data InsertionResult
    229 
    +  -- | The key was absent before the insertion.
    230 
    +  = Inserted
    231 
    +  -- | The key was already present; the container is unchanged.
    232 
    +  | AlreadyPresent
    233 
    +
    234 
    +-- | Write an entry into a 'MonotoneMap' unless the key is already present.
    235 
    +insertMonotoneMap :: Ord k => MonotoneMap k v -> k -> v -> IO InsertionResult
    236 
    +insertMonotoneMap ( MonotoneMap ref ) k v =
    237 
    +  atomicModifyIORef' ref \ m ->
    238 
    +    case Map.insertLookupWithKey ( \ _ _ old -> old ) k v m of
    239 
    +      ( Nothing , m' ) -> ( m', Inserted )
    240 
    +      ( Just _  , _  ) -> ( m , AlreadyPresent )
    241 
    +
    242 
    +-- | Write a new entry into a 'MonotoneMap'.
    218 243 
     --
    219 
    ---  - spawn a new worker thread
    220 
    ---  - wait for a worker thread to complete
    221 
    -data Scheduler r = Scheduler
    222 
    -  { spawnWorker :: (ConcurrentWorkerEnv -> IO r) -> IO ()
    223 
    -    -- ^ Spawn one concurrent worker.
    224 
    -    --
    225 
    -    -- The worker does not hold a token of the concurrency semaphore: the
    226 
    -    -- worker action should use 'withConcurrency' around the work whose
    227 
    -    -- concurrency should be limited.
    228 
    -  , awaitWorker :: IO (Either SomeException r)
    229 
    -    -- ^ Wait for one worker to complete.
    230 
    -    --
    231 
    -    -- Will crash if there are no outstanding workers.
    232 
    -  }
    244 
    +-- Panics if the entry is already present.
    245 
    +insertMonotoneMap_new
    246 
    +  :: ( HasDebugCallStack, Ord k, Outputable k )
    247 
    +  => MonotoneMap k v -> k -> v -> IO ()
    248 
    +insertMonotoneMap_new  mm k v =
    249 
    +  insertMonotoneMap mm k v >>= \case
    250 
    +    Inserted       -> pure ()
    251 
    +    AlreadyPresent -> pprPanic "monotone map: duplicate key" $ ppr k
    252 
    +
    253 
    +-- | The contents of a monotone map.
    254 
    +freezeMonotoneMap :: MonotoneMap k v -> IO ( Map.Map k v )
    255 
    +freezeMonotoneMap ( MonotoneMap ref ) = readIORef ref
    256 
    +
    257 
    +-- | A set that only ever grows.
    258 
    +newtype MonotoneSet k = MonotoneSet ( IORef ( Set.Set k ) )
    259 
    +
    260 
    +newMonotoneSet :: Set.Set k -> IO ( MonotoneSet k )
    261 
    +newMonotoneSet initial = MonotoneSet <$> newIORef initial
    262 
    +
    263 
    +-- | Add an element, unless it is already present.
    264 
    +insertMonotoneSet :: Ord k => MonotoneSet k -> k -> IO InsertionResult
    265 
    +insertMonotoneSet ( MonotoneSet ref ) k =
    266 
    +  atomicModifyIORef' ref \ s ->
    267 
    +    if k `Set.member` s
    268 
    +    then ( s              , AlreadyPresent )
    269 
    +    else ( Set.insert k s , Inserted )
    270 
    +
    271 
    +--------------------------------------------------------------------------------
    272 
    +-- * Pools of concurrent workers
    273 
    +--------------------------------------------------------------------------------
    233 274
    234 
    --- | Internal implementation of a concurrent worker scheduler.
    275 
    +-- | The order in which logging should happen when using concurrent workers.
    276 
    +data LogOrder
    277 
    +  -- | Log as we go.
    278 
    +  --
    279 
    +  -- Only valid when workers are spawned in a deterministic order.
    280 
    +  = LogAsWeGo
    281 
    +  -- | Accumulate logs per worker. Once all work is done, sort the logs
    282 
    +  -- before proceeding.
    283 
    +  --
    284 
    +  -- Used when workers may be spawned in a non-deterministic order.
    285 
    +  | SortLogs
    286 
    +
    287 
    +-- | A pool of concurrent workers with a given worker key type, supporting two
    288 
    +-- operations:
    235 289 
     --
    236 
    --- Usage of this function requires the following:
    290 
    +--  - spawning a new worker,
    291 
    +--  - waiting on all workers to finish.
    237 292 
     --
    238 
    ---  - all spawn/await actions are performed by a single thread,
    239 
    ---  - we never wait for more workers than were spawned,
    240 
    ---  - no worker outlives 'run_schedule'.
    241 
    -run_schedule
    242 
    -  :: forall r a
    243 
    -  .  String
    244 
    -      -- ^ thread label for workers
    293 
    +-- See Note [Deterministic concurrent workers].
    294 
    +data WorkerPool worker_key =
    295 
    +  WorkerPool
    296 
    +    { spawnWorker :: worker_key -> ( ConcurrentWorkerEnv -> IO () ) -> IO ()
    297 
    +      -- ^ Spawn one worker with the given worker key.
    298 
    +      --
    299 
    +      -- May be called from inside another worker.
    300 
    +      --
    301 
    +      -- The worker does not hold a token of the concurrency semaphore: the
    302 
    +      -- worker action should use 'withConcurrency' around the work whose
    303 
    +      -- concurrency should be limited.
    304 
    +      --
    305 
    +      -- An exception escaping the action stops further workers from being
    306 
    +      -- spawned, and is rethrown by 'waitForWorkers'.
    307 
    +    , waitForWorkers :: IO ()
    308 
    +      -- ^ Wait until all workers are done, throwing an exception if any
    309 
    +      -- worker failed (which exception is thrown is not deterministic).
    310 
    +    }
    311 
    +
    312 
    +-- | Internal implementation of a pool of concurrent workers.
    313 
    +run_pool
    314 
    +  :: forall worker_key a
    315 
    +  .  ( HasDebugCallStack, Ord worker_key, Outputable worker_key )
    316 
    +  => String -- ^ thread label for workers
    317 
    +  -> LogOrder
    245 318 
       -> Concurrency
    246 319 
       -> ConcurrentWorkerEnv
    247 
    -  -> (Scheduler r -> IO a)
    248 
    -        -- ^ worker action
    320 
    +  -> ( WorkerPool worker_key -> IO a )
    249 321 
       -> IO a
    250 
    -run_schedule worker_label conc parent_work_env withScheduler =
    322 
    +run_pool worker_label log_order conc parent_work_env withPool =
    251 323 
       case conc of
    252 324
    253 325 
         Serial -> do
    254 
    -      results_var <- newIORef Seq.empty
    326 
    +      queued_var <- newIORef $ Seq.empty @( worker_key, ConcurrentWorkerEnv -> IO () )
    327 
    +      logs_var   <- newMonotoneMap $ Map.empty @worker_key @LogQueue
    328 
    +
    255 329 
           let
    256 
    -        spawnWorker :: (ConcurrentWorkerEnv -> IO r) -> IO ()
    257 
    -        spawnWorker action = do
    258 
    -          res <- try @SomeException $
    330 
    +        spawnWorker :: worker_key -> ( ConcurrentWorkerEnv -> IO () ) -> IO ()
    331 
    +        spawnWorker worker_key action =
    332 
    +          modifyIORef' queued_var ( Seq.|> ( worker_key, action ) )
    333 
    +
    334 
    +        run_worker :: worker_key -> ( ConcurrentWorkerEnv -> IO () ) -> IO ()
    335 
    +        run_worker worker_key action = case log_order of
    336 
    +          LogAsWeGo ->
    259 337 
                 workerEnv_withLocalTmpFS parent_work_env action
    260 
    -          case res of
    261 
    -            Left e
    262 
    -              | Just _ <- fromException @SomeAsyncException e
    263 
    -              -> throwIO e
    264 
    -            _ -> modifyIORef' results_var (Seq.|> res)
    265 
    -
    266 
    -        awaitWorker :: IO (Either SomeException r)
    267 
    -        awaitWorker =
    268 
    -          readIORef results_var >>= \case
    269 
    -            res Seq.:<| rest -> do
    270 
    -              writeIORef results_var rest
    271 
    -              pure res
    272 
    -            Seq.Empty ->
    273 
    -              panic "run_schedule: no outstanding job"
    274 
    -
    275 
    -      withScheduler $ Scheduler { spawnWorker, awaitWorker }
    338 
    +          SortLogs -> do
    339 
    +            -- Use a log queue for consistency with the concurrent case.
    340 
    +            lq <- newLogQueue
    341 
    +            insertMonotoneMap_new logs_var worker_key lq
    342 
    +            let
    343 
    +              worker_work_env :: ConcurrentWorkerEnv
    344 
    +              worker_work_env =
    345 
    +                parent_work_env
    346 
    +                  { cwe_logger = pushLogHook ( const ( parLogAction lq ) )
    347 
    +                                   ( cwe_logger parent_work_env ) }
    348 
    +            workerEnv_withLocalTmpFS worker_work_env action
    349 
    +              `finally` finishLogQueue lq
    350 
    +
    351 
    +        waitForWorkers :: IO ()
    352 
    +        waitForWorkers = do
    353 
    +          next <- atomicModifyIORef' queued_var \ queued ->
    354 
    +            case queued of
    355 
    +              work Seq.:<| rest -> ( rest  , Just work )
    356 
    +              Seq.Empty         -> ( queued, Nothing )
    357 
    +          case next of
    358 
    +            Nothing -> pure ()
    359 
    +            Just ( worker_key, action ) ->
    360 
    +              run_worker worker_key action *> waitForWorkers
    361 
    +
    362 
    +        print_logs :: IO ()
    363 
    +        print_logs = do
    364 
    +          logs <- freezeMonotoneMap logs_var
    365 
    +          for_ ( Map.elems logs ) $ printLogs ( cwe_logger parent_work_env )
    366 
    +
    367 
    +      withPool ( WorkerPool { spawnWorker, waitForWorkers } )
    368 
    +        `finally` print_logs
    276 369
    277 370 
         Concurrent ( ConcurrencyEnv { ce_next_log_queue_id, ce_log_queue_queue } ) -> do
    278 
    -      worker_tids_var <- newTVarIO $ Set.empty @ThreadId
    279 
    -      all_results_vars_var <- newIORef $ Seq.empty @(MVar (Either SomeException r))
    371 
    +      worker_tids_var  <- newTVarIO $ Set.empty @ThreadId
    372 
    +      failure_var      <- newTVarIO $ Nothing @SomeException
    373 
    +      logs_var         <- newMonotoneMap $ Map.empty @worker_key @LogQueue
    374 
    +      last_spawned_var <- newIORef $ Nothing @worker_key
    280 375
    281 376 
           let
    282 377 
             wait_for_workers :: IO ()
    ... ... @@ -291,92 +386,118 @@ run_schedule worker_label conc parent_work_env withScheduler =
    291 386 
                 for_ tids killThread
    292 387 
               wait_for_workers
    293 388
    294 
    -        awaitWorker :: IO (Either SomeException r)
    295 
    -        awaitWorker =
    296 
    -          readIORef all_results_vars_var >>= \case
    297 
    -            first_worker_res_var Seq.:<| rest -> do
    298 
    -              writeIORef all_results_vars_var rest
    299 
    -              -- block on the earliest-spawned outstanding worker
    300 
    -              takeMVar first_worker_res_var
    301 
    -            Seq.Empty ->
    302 
    -              panic "run_schedule: no outstanding job"
    303 
    -
    304 
    -        spawnWorker :: (ConcurrentWorkerEnv -> IO r) -> IO ()
    305 
    -        spawnWorker action = mask_ do
    306 
    -
    307 
    -          worker_res_var <- newEmptyMVar
    308 
    -
    309 
    -          -- TmpFs
    310 
    -          lcl_tmpfs <- forkTmpFsFrom (cwe_tmpfs parent_work_env)
    311 
    -
    312 
    -          -- LogQueue
    313 
    -          lq <- do
    314 
    -            job_id <- atomicModifyIORef' ce_next_log_queue_id \n -> (n + 1, n)
    315 
    -            lq <- newLogQueue job_id
    316 
    -            atomically $ initLogQueue ce_log_queue_queue lq
    317 
    -            pure lq
    389 
    +        -- Record a worker failure, preventing any further work from starting.
    390 
    +        record_failure :: SomeException -> IO ()
    391 
    +        record_failure e =
    392 
    +          atomically $ modifyTVar' failure_var \ failure ->
    393 
    +            case failure of
    394 
    +              Nothing -> Just e
    395 
    +              Just {} -> failure
    318 396
    319 
    -          let
    397 
    +        waitForWorkers :: IO ()
    398 
    +        waitForWorkers = do
    399 
    +          wait_for_workers
    400 
    +          traverse_ throwIO =<< readTVarIO failure_var
    401 
    +
    402 
    +        -- Create the log queue of a worker, ordering it according to the worker key.
    403 
    +        new_worker_log_queue :: worker_key -> IO LogQueue
    404 
    +        new_worker_log_queue worker_key = do
    405 
    +          lq <- newLogQueue
    406 
    +          case log_order of
    407 
    +            LogAsWeGo -> do
    408 
    +              last_spawned <-
    409 
    +                atomicModifyIORef' last_spawned_var \ last_spawned ->
    410 
    +                  ( Just worker_key, last_spawned )
    411 
    +              massertPpr ( all ( < worker_key ) last_spawned ) $
    412 
    +                text "run_pool: LogAsWeGo workers spawned out of order:"
    413 
    +                  <+> ppr last_spawned <+> text "then" <+> ppr worker_key
    414 
    +              job_id <- atomicModifyIORef' ce_next_log_queue_id \ n -> ( n + 1, n )
    415 
    +              atomically $ initLogQueue ce_log_queue_queue job_id lq
    416 
    +            SortLogs ->
    417 
    +              insertMonotoneMap_new logs_var worker_key lq
    418 
    +          pure lq
    419 
    +
    420 
    +        -- Hand the log queues over for printing, in worker key order.
    421 
    +        release_queued_logs :: IO ()
    422 
    +        release_queued_logs = do
    423 
    +          queued <- freezeMonotoneMap logs_var
    424 
    +          unless ( Map.null queued ) do
    425 
    +            first_id <-
    426 
    +              atomicModifyIORef' ce_next_log_queue_id \ n ->
    427 
    +                ( n + Map.size queued, n )
    428 
    +            atomically $
    429 
    +              for_ ( zip [ first_id .. ] ( Map.elems queued ) ) \ ( job_id, lq ) ->
    430 
    +                initLogQueue ce_log_queue_queue job_id lq
    431 
    +
    432 
    +        spawnWorker :: worker_key -> ( ConcurrentWorkerEnv -> IO () ) -> IO ()
    433 
    +        spawnWorker worker_key action = mask_ do
    434 
    +          failure <- readTVarIO failure_var
    435 
    +          case failure of
    436 
    +            -- A worker has failed: don't start any more work.
    437 
    +            Just {} -> pure ()
    438 
    +            Nothing -> do
    439 
    +
    440 
    +              -- TmpFs
    441 
    +              lcl_tmpfs <- forkTmpFsFrom ( cwe_tmpfs parent_work_env )
    442 
    +
    443 
    +              -- LogQueue
    444 
    +              lq <- new_worker_log_queue worker_key
    320 445
    321 
    -            worker_work_env :: ConcurrentWorkerEnv
    322 
    -            worker_work_env =
    323 
    -              parent_work_env
    324 
    -                { cwe_tmpfs  = lcl_tmpfs
    325 
    -                , cwe_logger = pushLogHook (const (parLogAction lq))
    326 
    -                                (cwe_logger parent_work_env)
    327 
    -                }
    328 
    -
    329 
    -            -- Run a worker action and record its result.
    330 
    -            run_worker_and_record :: IO r -> IO ()
    331 
    -            run_worker_and_record worker_action = do
    332 
    -              res <- try @SomeException worker_action
    333 
    -              case res of
    334 
    -                Left e
    335 
    -                  -- Worker is being cancelled: don't record anything.
    336 
    -                  | Just ThreadKilled <- fromException e
    337 
    -                  -> pure ()
    338 
    -                _ -> putMVar worker_res_var res
    339 
    -
    340 
    -            -- Record that a worker thread is done.
    341 
    -            mark_worker_done :: ThreadId -> IO ()
    342 
    -            mark_worker_done tid =
    343 
    -              uninterruptibleMask_ do
    344 
    -                -- Uninterruptible: the deletion below /must/ occur.
    345 
    -                -- An uninterruptible mask is OK as we only ever block for (GAP) below.
    346 
    -                mergeTmpFsInto lcl_tmpfs $ cwe_tmpfs parent_work_env
    347 
    -                finishLogQueue lq
    348 
    -                atomically do
    349 
    -                  tids <- readTVar worker_tids_var
    350 
    -                  check $ tid `Set.member` tids
    351 
    -                    -- Ensure we never end up with a dead ThreadId in 'worker_tids_var'
    352 
    -                    -- (if the worker thread finishes before the parent thread has
    353 
    -                    -- the time to add its ThreadId to 'worker_tids_var').
    354 
    -
    355 
    -                  writeTVar worker_tids_var $ Set.delete tid tids
    356 
    -
    357 
    -            run_worker :: (forall b. IO b -> IO b) -> IO ()
    358 
    -            run_worker unmask = do
    359 
    -              tid <- myThreadId
    360 
    -              labelThread tid worker_label
    361 446 
                   let
    362 
    -                worker_action :: IO r
    363 
    -                worker_action = unmask $ action worker_work_env
    364 447
    365 
    -              run_worker_and_record worker_action `finally`
    366 
    -                mark_worker_done tid
    367 
    -
    368 
    -          worker_tid <-
    369 
    -            forkIOWithUnmask run_worker
    370 
    -              `onException` finishLogQueue lq
    371 
    -          -- Very short (GAP) between forking the thread and recording its ThreadId.
    372 
    -          atomically $ modifyTVar' worker_tids_var $ Set.insert worker_tid
    373 
    -          modifyIORef' all_results_vars_var (Seq.|> worker_res_var)
    374 
    -
    375 
    -      mask \ restore -> do
    376 
    -        result <- restore (withScheduler $ Scheduler { spawnWorker, awaitWorker })
    377 
    -                    `onException` cancel_workers
    378 
    -        restore wait_for_workers `onException` cancel_workers
    379 
    -        pure result
    448 
    +                worker_work_env :: ConcurrentWorkerEnv
    449 
    +                worker_work_env =
    450 
    +                  parent_work_env
    451 
    +                    { cwe_tmpfs  = lcl_tmpfs
    452 
    +                    , cwe_logger = pushLogHook ( const ( parLogAction lq ) )
    453 
    +                                    ( cwe_logger parent_work_env )
    454 
    +                    }
    455 
    +
    456 
    +                -- Record that a worker thread is done.
    457 
    +                mark_worker_done :: ThreadId -> IO ()
    458 
    +                mark_worker_done tid =
    459 
    +                  uninterruptibleMask_ do
    460 
    +                    -- Uninterruptible: the deletion below /must/ occur.
    461 
    +                    -- An uninterruptible mask is OK as we only ever block for (GAP) below.
    462 
    +                    mergeTmpFsInto lcl_tmpfs $ cwe_tmpfs parent_work_env
    463 
    +                    finishLogQueue lq
    464 
    +                    atomically do
    465 
    +                      tids <- readTVar worker_tids_var
    466 
    +                      check $ tid `Set.member` tids
    467 
    +                        -- Ensure we never end up with a dead ThreadId in 'worker_tids_var'
    468 
    +                        -- (if the worker thread finishes before the parent thread has
    469 
    +                        -- the time to add its ThreadId to 'worker_tids_var').
    470 
    +
    471 
    +                      writeTVar worker_tids_var $ Set.delete tid tids
    472 
    +
    473 
    +                handle_worker_exception :: SomeException -> IO ()
    474 
    +                handle_worker_exception e
    475 
    +                  -- Worker is being cancelled: not a failure to report.
    476 
    +                  | Just ThreadKilled <- fromException e
    477 
    +                  = pure ()
    478 
    +                  | otherwise
    479 
    +                  = record_failure e
    480 
    +
    481 
    +                run_worker :: ( forall b. IO b -> IO b ) -> IO ()
    482 
    +                run_worker unmask = do
    483 
    +                  tid <- myThreadId
    484 
    +                  labelThread tid worker_label
    485 
    +                  ( unmask ( action worker_work_env )
    486 
    +                      `catch` handle_worker_exception )
    487 
    +                    `finally` mark_worker_done tid
    488 
    +
    489 
    +              worker_tid <-
    490 
    +                forkIOWithUnmask run_worker
    491 
    +                  `onException` finishLogQueue lq
    492 
    +              -- Very short (GAP) between forking the thread and recording its ThreadId.
    493 
    +              atomically $ modifyTVar' worker_tids_var $ Set.insert worker_tid
    494 
    +
    495 
    +      ( `finally` release_queued_logs ) $
    496 
    +        mask \ restore -> do
    497 
    +          result <- restore ( withPool $ WorkerPool { spawnWorker, waitForWorkers } )
    498 
    +                      `onException` cancel_workers
    499 
    +          restore wait_for_workers `onException` cancel_workers
    500 
    +          pure result
    380 501
    381 502 
     --------------------------------------------------------------------------------
    382 503 
     -- * Derived scheduling functionality
    ... ... @@ -387,7 +508,9 @@ run_schedule worker_label conc parent_work_env withScheduler =
    387 508 
     -- Workers run to completion (no early abort); the first exception
    388 509 
     -- (in input order) is rethrown at the end.
    389 510 
     mapConcurrentWorkers
    390 
    -  :: String -- ^ thread label for workers
    511 
    +  :: forall a b
    512 
    +  .  HasDebugCallStack
    513 
    +  => String -- ^ thread label for workers
    391 514 
       -> Concurrency
    392 515 
       -> ConcurrentWorkerEnv
    393 516 
       -> (ConcurrentWorkerEnv -> a -> IO b)
    ... ... @@ -398,67 +521,111 @@ mapConcurrentWorkers
    398 521 
       -> [a]
    399 522 
       -> IO [b]
    400 523 
     mapConcurrentWorkers worker_label conc work_env f xs =
    401 
    -  run_schedule worker_label conc work_env \ scheduler -> do
    402 
    -    for_ xs \ x -> spawnWorker scheduler \ worker_env -> f worker_env x
    403 
    -    results <- replicateM (length xs) (awaitWorker scheduler)
    404 
    -    either throwIO pure (sequence results)
    524 
    +  -- LogAsWeGo: workers are keyed by their position in the input list and
    525 
    +  -- spawned in that same order, so their output can be printed as it is produced.
    526 
    +  run_pool worker_label LogAsWeGo conc work_env \ pool -> do
    527 
    +    results <- newMonotoneMap $ Map.empty @Int @( Either SomeException b )
    528 
    +    for_ ( zip [ 0 .. ] xs ) \ ( i, x ) ->
    529 
    +      spawnWorker pool i \ worker_env -> do
    530 
    +        res <- try @SomeException $ f worker_env x
    531 
    +        case res of
    532 
    +          Left e
    533 
    +            -- Take care to avoid swallowing async exceptions.
    534 
    +            | Just _ <- fromException @SomeAsyncException e
    535 
    +            -> throwIO e
    536 
    +          _ -> insertMonotoneMap_new results i res
    537 
    +    waitForWorkers pool
    538 
    +    all_results <- freezeMonotoneMap results
    539 
    +    massertPpr ( Map.size all_results == length xs ) $
    540 
    +      text "mapConcurrentWorkers: missing results"
    541 
    +    either throwIO pure $ sequence $ Map.elems all_results
    542 
    +
    543 
    +-- | How to expand a node in a graph for 'concurrentTraversal'.
    544 
    +data NodeExpander k n v =
    545 
    +  NodeExpander
    546 
    +    { nodeKey :: n -> k
    547 
    +      -- ^ The identity of a node.
    548 
    +    , expandNode :: ConcurrentWorkerEnv -> n -> IO ( v, [n] )
    549 
    +      -- ^ Expand a node into its result and the children to visit next.
    550 
    +      --
    551 
    +      -- To guarantee determinism, the children must be a pure function of the
    552 
    +      -- input, and IO effects must not observably depend on the order in
    553 
    +      -- which nodes are expanded.
    554 
    +      --
    555 
    +      -- NB: workers do not hold semaphore tokens by default; use
    556 
    +      -- 'withConcurrency' to acquire one
    557 
    +    }
    405 558
    406 
    --- | Depth-first traversal with on-the-fly expansion of nodes.
    559 
    +-- | Deterministically traverse a graph whose nodes are discovered as they are
    560 
    +-- expanded.
    407 561 
     --
    408 
    --- Each expansion step is handled by a worker thread under the given
    409 
    --- concurrency control.
    410 
    ---
    411 
    --- Deterministic: expansions are consumed in the order the nodes were
    412 
    --- discovered, so the traversal is a function of the node graph alone.
    413 
    ---
    414 
    --- Fails fast: the first worker exception cancels the outstanding workers and
    415 
    --- is rethrown.
    416 
    -concurrentTraversal_DF
    417 
    -  :: forall k n r
    418 
    -  .  Ord k
    562 
    +-- Fails fast: once a worker throws an exception, no further work is started,
    563 
    +-- and the exception is rethrown once the outstanding workers finish.
    564 
    +concurrentTraversal
    565 
    +  :: forall k n v
    566 
    +  .  ( HasDebugCallStack, Ord k, Outputable k )
    419 567 
       => String -- ^ thread label for workers
    420 568 
       -> Concurrency
    421 569 
       -> ConcurrentWorkerEnv
    422 
    -  -> Map.Map k r
    570 
    +  -> NodeExpander k n v
    571 
    +  -> Map.Map k v
    423 572 
          -- ^ results known ahead of time (no expansion needed)
    424 573 
       -> [n]
    425 574 
          -- ^ root nodes
    426 
    -  -> (n -> k)
    427 
    -     -- ^ node key from node
    428 
    -  -> (ConcurrentWorkerEnv -> n -> IO (r, [n]))
    429 
    -     -- ^ worker action: expand a node into its result and the children to visit next
    430 
    -     --
    431 
    -     -- NB: workers do not hold semaphore tokens by default; use
    432 
    -     -- 'withConcurrency' to acquire one
    433 
    -  -> IO (Map.Map k r)
    434 
    -concurrentTraversal_DF worker_label conc work_env base_map roots key expand =
    435 
    -  run_schedule worker_label conc work_env \ scheduler -> do
    575 
    +  -> IO ( Map.Map k v )
    576 
    +concurrentTraversal
    577 
    +  worker_label conc work_env
    578 
    +  ( NodeExpander { nodeKey, expandNode } )
    579 
    +  base_map roots
    580 
    +  =
    581 
    +  -- SortLogs: nodes are discovered in an order that depends on the schedule,
    582 
    +  -- so the workers' logs must be ordered before being printed.
    583 
    +  run_pool worker_label SortLogs conc work_env \ pool -> do
    584 
    +
    585 
    +    -- The keys whose expansion has been started.
    586 
    +    claims  <- newMonotoneSet $ Map.keysSet base_map
    587 
    +    results <- newMonotoneMap base_map
    588 
    +
    436 589 
         let
    437 
    -      expand_node :: n -> ConcurrentWorkerEnv -> IO (k, (r, [n]))
    438 
    -      expand_node node worker_env = do
    439 
    -        res <- expand worker_env node
    440 
    -        pure (key node, res)
    441 
    -
    442 
    -      go
    443 
    -        :: Map.Map k r -- expanded nodes and their results
    444 
    -        -> Set.Set k   -- nodes currently being expanded
    445 
    -        -> [n]         -- discovered nodes, to expand next
    446 
    -        -> IO (Map.Map k r)
    447 
    -      go !visited !pending (node : worklist)
    448 
    -        | k `Set.member` pending || k `Map.member` visited
    449 
    -        = go visited pending worklist
    450 
    -        | otherwise
    451 
    -        = do spawnWorker scheduler (expand_node node)
    452 
    -             go visited (Set.insert k pending) worklist
    590 
    +      discover :: n -> IO ()
    591 
    +      discover node =
    592 
    +        -- Claim the work for this node to avoid any other worker duplicating it.
    593 
    +        insertMonotoneSet claims key >>= \case
    594 
    +          AlreadyPresent -> pure ()
    595 
    +          Inserted ->
    596 
    +            spawnWorker pool key \ worker_env -> do
    597 
    +              ( result, children ) <- expandNode worker_env node
    598 
    +              insertMonotoneMap_new results key result
    599 
    +              for_ children discover
    453 600 
             where
    454 
    -          k = key node
    455 
    -      go visited pending []
    456 
    -        | Set.null pending
    457 
    -        = pure visited
    458 
    -        | otherwise
    459 
    -        = awaitWorker scheduler >>= \case
    460 
    -            Left e -> throwIO e
    461 
    -            Right (k, (result, children)) ->
    462 
    -              go (Map.insert k result visited) (Set.delete k pending) children
    463 
    -
    464 
    -    go base_map Set.empty roots
    601 
    +          key = nodeKey node
    602 
    +
    603 
    +    for_ roots discover
    604 
    +    waitForWorkers pool
    605 
    +    freezeMonotoneMap results
    606 
    +
    607 
    +{- Note [Deterministic concurrent workers]
    608 
    +~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
    609 
    +To provide deterministic output when doing graph traversal (as in downsweep)
    610 
    +despite using concurrent workers, we ensure that nothing can observe the order
    611 
    +in which workers do their work:
    612 
    +
    613 
    +  1. Any chunk of work is performed at most once: every worker atomically claims
    614 
    +     ownership of the work it is going to do before it starts that work.
    615 
    +
    616 
    +  2. Workers report results by writing to a 'MonotoneMap', whose entries are
    617 
    +     written at most once. Other outputs (such as logging output) is accumulated
    618 
    +     in a deterministic order and reported at the end.
    619 
    +
    620 
    +This scheme allows us to retain maximum concurrency: it allows new edges to be
    621 
    +discovered by any worker and immediately processed.
    622 
    +
    623 
    +For this scheme to provide deterministic output, we require that:
    624 
    +
    625 
    +  * The expansion of a node is a pure function of the node.
    626 
    +  * The work itself should not observably depend on when it was run.
    627 
    +
    628 
    +Failure is not deterministic: which worker's exception is reported depends on
    629 
    +the schedule. When deterministic error messages are desired, the workers should
    630 
    +return an error value instead (as 'mapConcurrentWorkers' does).
    631 
    +-}

  • compiler/GHC/Driver/Downsweep.hs
    ... ... @@ -174,9 +174,8 @@ incrementally constructing a ModuleGraph using the GHC API; See #27054). So
    174 174 
     `downsweep` takes a `Maybe ModuleGraph` as one of its arguments.
    175 175
    176 176 
     Downsweep iteratively *expands* each so-called 'DownsweepNode' into a list of
    177 
    -its dependencies, and recursively traverses all reachable nodes in a parallel
    178 
    -non-det-depth-first order using 'parDfsBuild'. A 'DownsweepNode' is *expanded*
    179 
    -by 'dsNodeExpand':
    177 
    +its dependencies, and recursively traverses all reachable nodes concurrently
    178 
    +using 'parBuild'. A 'DownsweepNode' is *expanded* by 'dsNodeExpand':
    180 179
    181 180 
       dsNodeExpand :: DownsweepNode -> DownsweepM (NodeRes (ModuleGraphNode, [DownsweepNode]))
    182 181
    ... ... @@ -591,7 +590,7 @@ loopModuleNodeInfos :: M.Map NodeKey (NodeRes ModuleGraphNode) -> [ModuleNodeInf
    591 590 
     loopUnits           :: M.Map NodeKey (NodeRes ModuleGraphNode) -> UnitId -> [UnitId]            -> DownsweepM (M.Map NodeKey (NodeRes ModuleGraphNode))
    592 591 
     loopInstantiations  :: M.Map NodeKey (NodeRes ModuleGraphNode) -> [(UnitId, InstantiatedUnit)]  -> DownsweepM (M.Map NodeKey (NodeRes ModuleGraphNode))
    593 592 
     loopFromInteractive :: M.Map NodeKey (NodeRes ModuleGraphNode) -> Module -> [InteractiveImport] -> DownsweepM (M.Map NodeKey (NodeRes ModuleGraphNode))
    594 
    -loopDownsweepNodes  base_map nodes = parDfsBuild (Just base_map) nodes dsNodeInfoKey dsNodeExpand
    593 
    +loopDownsweepNodes  base_map nodes = parBuild (Just base_map) nodes dsNodeInfoKey dsNodeExpand
    595 594 
     loopModuleNodeInfos base_map       = loopDownsweepNodes base_map . map DSMod
    596 595 
     loopUnits           base_map homud = loopDownsweepNodes base_map . map (DSUnit homud)
    597 596 
     loopInstantiations  base_map       = loopDownsweepNodes base_map . map (uncurry DSInst)
    ... ... @@ -1732,10 +1731,10 @@ getPreprocessedImports hsc_env src_fn mb_phase maybe_buf = do
    1732 1731 
       return PreprocessedImports {..}
    1733 1732
    1734 1733 
     --------------------------------------------------------------------------------
    1735 
    --- * Generic traversal of iteratively-built graph: parDfsBuild
    1734 
    +-- * Generic traversal of iteratively-built graph: parBuild
    1736 1735 
     --------------------------------------------------------------------------------
    1737 1736
    1738 
    --- | The result of expanding a node in 'parDfsBuild'.
    1737 
    +-- | The result of expanding a node in 'parBuild'.
    1739 1738 
     data NodeRes v
    1740 1739 
       -- | Computed the node payload successfully
    1741 1740 
       = NSuccess v
    ... ... @@ -1748,20 +1747,18 @@ data NodeRes v
    1748 1747 
       -- abort.
    1749 1748 
       | NSkip
    1750 1749
    1751 
    --- | In a parallel non-det-depth-first order, and starting from the given roots, traverse a
    1752 
    --- graph by iteratively expanding a node into a payload and a list of children
    1753 
    --- nodes to visit next.
    1750 
    +-- | Starting from the given roots, traverse a graph by iteratively expanding a
    1751 
    +-- node into a payload and a list of children nodes to visit next, expanding
    1752 
    +-- nodes concurrently.
    1754 1753 
     --
    1755 
    --- A node is NEVER visited/expanded more than once, as long as the node key
    1756 
    --- @k@, computed from the node @n@, uniquely identifies that node.
    1757 
    ---
    1758 
    --- The first argument @base_map@ is the starting set of already visited nodes
    1759 
    --- (these nodes won't be expanded again!).
    1754 
    +-- A node is NEVER visited/expanded more than once: nodes are identified by
    1755 
    +-- their key @k@, and the first node discovered under a key is the one that is
    1756 
    +-- expanded.
    1760 1757 
     --
    1761 1758 
     -- The result is a mapping from the key of every node transitively reachable
    1762 1759 
     -- from the root nodes (inclusively) to the payload returned by expanding that
    1763 1760 
     -- node. The result includes the previously visited nodes given in @base_map@,
    1764 
    --- s.t. @parDfsBuild base_map [] _ _ == base_map@.
    1761 
    +-- s.t. @parBuild base_map [] _ _ _ == base_map@.
    1765 1762 
     --
    1766 1763 
     -- The @expand@ function returns a 'NodeRes'. See the 'NodeRes' documentation
    1767 1764 
     -- for more information about each result type.
    ... ... @@ -1771,7 +1768,7 @@ data NodeRes v
    1771 1768 
     -- See Note [Parallel Downsweep] for more information about how parallelism is
    1772 1769 
     -- achieved, and See Note [Downsweep Control Flow and Caching] for information
    1773 1770 
     -- about the various caches used.
    1774 
    -parDfsBuild :: forall k v n. Ord k
    1771 
    +parBuild :: forall k v n. (Ord k, Outputable k)
    1775 1772 
              => Maybe (Map.Map k (NodeRes v))
    1776 1773 
              -- ^ Base map, existing results. We won't re-expand any of the nodes
    1777 1774 
              -- already present in this map.
    ... ... @@ -1785,22 +1782,26 @@ parDfsBuild :: forall k v n. Ord k
    1785 1782 
              -> DownsweepM (Map.Map k (NodeRes v))
    1786 1783 
              -- ^ The result accumulates the payload of expanding the root nodes
    1787 1784 
              -- and all nodes transitively reachable from those roots.
    1788 
    -parDfsBuild base_map roots key expand = ReaderT $ \ds_env -> do
    1785 
    +parBuild base_map roots nodeKey expand = ReaderT $ \ds_env -> do
    1789 1786
    1790 1787 
       let
    1791 1788 
         conc :: Concurrency
    1792 1789 
         conc = ds_concurrency ds_env
    1793 1790
    1794 
    -    expand_node :: ConcurrentWorkerEnv -> n -> IO (NodeRes v, [n])
    1795 
    -    expand_node worker_env node = do
    1791 
    +    expandNode :: ConcurrentWorkerEnv -> n -> IO (NodeRes v, [n])
    1792 
    +    expandNode worker_env node = do
    1796 1793 
           result <- withConcurrency conc $
    1797 1794 
             runDownsweepM (setDownsweepWorkerEnv worker_env ds_env) (expand node)
    1798 1795 
           pure $ case result of
    1799 1796 
             NSkip                    -> (NSkip, [])
    1800 1797 
             NSuccess (val, new_work) -> (NSuccess val, new_work)
    1801 1798
    1802 
    -  concurrentTraversal_DF "downsweep_worker" conc (viewHscWorkerEnv $ ds_hsc_env ds_env)
    1803 
    -    (fromMaybe mempty base_map) roots key expand_node
    1799 
    +  concurrentTraversal "downsweep_worker"
    1800 
    +    conc
    1801 
    +    (viewHscWorkerEnv $ ds_hsc_env ds_env)
    1802 
    +    (NodeExpander { nodeKey, expandNode })
    1803 
    +    (fromMaybe mempty base_map)
    1804 
    +    roots
    1804 1805
    1805 1806 
     setDownsweepWorkerEnv :: ConcurrentWorkerEnv -> DownsweepEnv -> DownsweepEnv
    1806 1807 
     setDownsweepWorkerEnv work_env env =
    ... ... @@ -1809,7 +1810,7 @@ setDownsweepWorkerEnv work_env env =
    1809 1810 
     {-
    1810 1811 
     Note [Downsweep Control Flow and Caching]
    1811 1812 
     ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
    1812 
    -The control flow of downsweep is extracted into a single function `parDfsBuild`,
    1813 
    +The control flow of downsweep is extracted into a single function `parBuild`,
    1813 1814 
     which takes care of iteratively expanding and traversing all nodes of the
    1814 1815 
     in-construction module graph necessary to build a full `ModuleGraph` at the
    1815 1816 
     end.
    ... ... @@ -1818,7 +1819,7 @@ There are three levels of caching going on, all of which are necessary to make
    1818 1819 
     sure we don't do repeated work (notably, we NEVER summarise the same module
    1819 1820 
     twice).
    1820 1821
    1821 
    -1. `parDfsBuild` accumulates the final module graph and never revisits the
    1822 
    +1. `parBuild` accumulates the final module graph and never revisits the
    1822 1823 
        same node of the module graph. Cache is keyed by the final
    1823 1824 
        `ModuleGraph`s `NodeKey`s.
    1824 1825
    ... ... @@ -1900,9 +1901,12 @@ things, and that processing can often be costly (e.g. see `expandModuleSummary`)
    1900 1901 
     We leverage multiple threads in this traversal to expand more than one module
    1901 1902 
     at once, respecting -j<N> to mean we never expand more than N modules at once.
    1902 1903
    1903 
    -We use the concurrent scheduling abstraction from GHC.Driver.Concurrency
    1904 
    -('concurrentTraversal_DF'). Each time we discover a new node, a worker is
    1905 
    -spawned to expand it. After each expansion completes, the resulting children
    1906 
    -nodes are pushed onto the worklist. With -j1 no threads are involved: each
    1907 
    -expansion runs in sequence.
    1904 
    +We use the concurrent traversal abstraction from GHC.Driver.Concurrency
    1905 
    +('concurrentTraversal'). Each time we discover a new node, a worker is spawned
    1906 
    +to expand it; a worker spawns the workers for the children it discovers itself,
    1907 
    +so an expansion never waits on an unrelated one. With -j1 no threads are
    1908 
    +involved: each expansion runs in sequence.
    1909 
    +
    1910 
    +The traversal is deterministic despite the fact that the order in which workers
    1911 
    +are spawned is not: see Note [Deterministic concurrent workers] in GHC.Driver.Concurrency.
    1908 1912 
     -}

  • compiler/GHC/Driver/Pipeline/LogQueue.hs
    ... ... @@ -4,6 +4,7 @@ module GHC.Driver.Pipeline.LogQueue ( LogQueue(..)
    4 4 
                                       , finishLogQueue
    5 5 
                                       , writeLogQueue
    6 6 
                                       , parLogAction
    7 
    +                                  , printLogs
    7 8
    8 9 
                                       , LogQueueQueue(..)
    9 10 
                                       , initLogQueue
    ... ... @@ -25,19 +26,18 @@ import Control.Monad
    25 26
    26 27 
     -- LogQueue Abstraction
    27 28
    28 
    --- | Each module is given a unique 'LogQueue' to redirect compilation messages
    29 
    --- to. A 'Nothing' value contains the result of compilation, and denotes the
    30 
    --- end of the message queue.
    31 
    -data LogQueue = LogQueue { logQueueId :: !Int
    32 
    -                         , logQueueMessages :: !(IORef [Maybe (MessageClass, SrcSpan, SDoc, LogFlags)])
    29 
    +-- | A buffer of compilation messages produced by one worker.
    30 
    +--
    31 
    +-- A 'Nothing' value denotes the end of the message queue.
    32 
    +data LogQueue = LogQueue { logQueueMessages  :: !(IORef [Maybe (MessageClass, SrcSpan, SDoc, LogFlags)])
    33 33 
                              , logQueueSemaphore :: !(MVar ())
    34 34 
                              }
    35 35
    36 
    -newLogQueue :: Int -> IO LogQueue
    37 
    -newLogQueue n = do
    36 
    +newLogQueue :: IO LogQueue
    37 
    +newLogQueue = do
    38 38 
       mqueue <- newIORef []
    39 39 
       sem <- newMVar ()
    40 
    -  return (LogQueue n mqueue sem)
    40 
    +  return (LogQueue mqueue sem)
    41 41
    42 42 
     finishLogQueue :: LogQueue -> IO ()
    43 43 
     finishLogQueue lq = do
    ... ... @@ -50,7 +50,7 @@ writeLogQueue lq msg = do
    50 50
    51 51 
     -- | Internal helper for writing log messages
    52 52 
     writeLogQueueInternal :: LogQueue -> Maybe (MessageClass,SrcSpan,SDoc, LogFlags) -> IO ()
    53 
    -writeLogQueueInternal (LogQueue _n ref sem) msg = do
    53 
    +writeLogQueueInternal (LogQueue ref sem) msg = do
    54 54 
         atomicModifyIORef' ref $ \msgs -> (msg:msgs,())
    55 55 
         _ <- tryPutMVar sem ()
    56 56 
         return ()
    ... ... @@ -61,9 +61,11 @@ parLogAction :: LogQueue -> LogAction
    61 61 
     parLogAction log_queue log_flags !msgClass !srcSpan !msg =
    62 62 
         writeLogQueue log_queue (msgClass,srcSpan,msg, log_flags)
    63 63
    64 
    --- Print each message from the log_queue using the global logger
    64 
    +-- | Print each message from the log queue using the given logger.
    65 
    +--
    66 
    +-- Blocks until the queue has been finished with 'finishLogQueue'.
    65 67 
     printLogs :: Logger -> LogQueue -> IO ()
    66 
    -printLogs !logger (LogQueue _n ref sem) = read_msgs
    68 
    +printLogs !logger (LogQueue ref sem) = read_msgs
    67 69 
       where read_msgs = do
    68 70 
                 takeMVar sem
    69 71 
                 msgs <- atomicModifyIORef' ref $ \xs -> ([], reverse xs)
    ... ... @@ -84,11 +86,23 @@ data LogQueueQueue = LogQueueQueue Int (IM.IntMap LogQueue)
    84 86 
     newLogQueueQueue :: LogQueueQueue
    85 87 
     newLogQueueQueue = LogQueueQueue 1 IM.empty
    86 88
    87 
    -addToQueueQueue :: LogQueue -> LogQueueQueue -> LogQueueQueue
    88 
    -addToQueueQueue lq (LogQueueQueue n im) = LogQueueQueue n (IM.insert (logQueueId lq) lq im)
    89 
    -
    90 
    -initLogQueue :: TVar LogQueueQueue -> LogQueue -> STM ()
    91 
    -initLogQueue lqq lq = modifyTVar lqq (addToQueueQueue lq)
    89 
    +addToQueueQueue
    90 
    +  :: Int -- ^ 1-indexed position in which to add
    91 
    +  -> LogQueue
    92 
    +  -> LogQueueQueue
    93 
    +  -> LogQueueQueue
    94 
    +addToQueueQueue i lq (LogQueueQueue n im) = LogQueueQueue n (IM.insert i lq im)
    95 
    +
    96 
    +-- | Hand a log queue to the log thread, to be printed at the given position.
    97 
    +--
    98 
    +-- Positions must be contiguous: the log thread prints position @n@ only after
    99 
    +-- every position below @n@ is done.
    100 
    +initLogQueue
    101 
    +  :: TVar LogQueueQueue
    102 
    +  -> Int -- ^ position in the 'LogQueueQueue' in which to insert the 'LogQueue'
    103 
    +  -> LogQueue
    104 
    +  -> STM ()
    105 
    +initLogQueue lqq i lq = modifyTVar lqq (addToQueueQueue i lq)
    92 106
    93 107 
     -- | Return all items in the queue in ascending order
    94 108 
     allLogQueues :: LogQueueQueue -> [LogQueue]

    • —
    View it on GitLab.
    You're receiving this email because of your account on gitlab.haskell.org. Manage all notifications · Help Notification message regarding https://gitlab.haskell.org/ghc/ghc/-/commit/04ba945021e675d73807501cf132cec7226812a2 at 1787134336