GitLab

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

Commits:

566d789d

by sheaf at 2026-08-14T23:26:51+02:00

Concurrent worker abstraction for downsweep/make

This commit introduces GHC.Driver.Concurrency which provides a high-level
interface over scheduling concurrent workers, used by downsweep and
--make.

Summary of changes:

  - The ad-hoc plumbing of AbstractSem is replaced by the dedicated
    'data Concurrency = Serial | Concurrent ConcurrencyEnv'.
    This avoids a footgun in which one could try to use an 'AbstractSem'
    as a lock in the serial case.

  - We no longer wastefully re-run an entire action upon a semaphore
    failure. The "fallback to -j1" logic is preserved, but it only
    applies to the initial attempt at opening a semaphore, not on late
    semaphore errors that occur partway through a lengthy computation.

  - Logger threads are now properly cleaned up on exception.

  - Drop the 'GhcMessage -> AnyGhcDiagnostic' and 'Maybe Messager'
    arguments to 'depanalE', 'depanalPartial' and 'downsweep', which
    were all dead in practice.

9 changed files:

+ compiler/GHC/Driver/Concurrency.hs
+ compiler/GHC/Driver/Config/Concurrency.hs
compiler/GHC/Driver/Downsweep.hs
compiler/GHC/Driver/Make.hs
compiler/GHC/Driver/MakeAction.hs
compiler/GHC/Driver/MakeSem.hs
compiler/GHC/Utils/TmpFs.hs
compiler/ghc.cabal.in
utils/haddock/haddock-api/src/Haddock/Interface.hs

Changes:

compiler/GHC/Driver/Concurrency.hs

 +{-# LANGUAGE CPP #-}
++
 +{-# LANGUAGE BlockArguments #-}
++
 +module GHC.Driver.Concurrency
 +  ( -- * Worker limit and concurrency
 +    WorkerLimit(..)
 +  , isWorkerLimitSequential
 +  , withWorkerLimit
 +  , Concurrency
 +  , withConcurrency
 +    -- * Concurrent worker scheduling
 +  , ConcurrentWorkerEnv(..)
 +  , mapConcurrentWorkers
 +  , concurrentTraversal_DF
 +  )
 +  where
++
 +import GHC.Prelude
++
 +import GHC.Driver.MakeSem
 +import GHC.Driver.Pipeline.LogQueue
 +  ( LogQueueQueue, finishLogQueue, initLogQueue, logThread
 +  , newLogQueue, newLogQueueQueue, parLogAction )
 +import GHC.Utils.Logger
 +  ( Logger, makeThreadSafe, pushLogHook )
 +import GHC.Utils.Panic
 +  ( panic )
 +import GHC.Utils.TmpFs
 +  ( TmpFs, forkTmpFsFrom, mergeTmpFsInto, withLocalTmpFS )
++
 +import System.Semaphore
 +  ( SemaphoreError, SemaphoreIdentifier )
++
 +#if defined(wasm32_HOST_ARCH) || defined(javascript_HOST_ARCH)
 +import Control.Concurrent
 +  ( ThreadId, forkIOWithUnmask, killThread, myThreadId )
 +import Control.Concurrent.MVar
 +  ( MVar, newEmptyMVar, newMVar, putMVar, takeMVar )
 +import GHC.Conc
 +  ( labelThread )
 +#else
 +import Control.Concurrent
 +  ( ThreadId, forkIOWithUnmask, killThread, myThreadId
 +  , newQSem, signalQSem, waitQSem, MVar, takeMVar, putMVar, newEmptyMVar )
 +import Control.Monad
 +  ( unless )
 +import qualified Control.Monad.Catch as MC
 +import GHC.Conc
 +  ( getNumCapabilities, getNumProcessors, labelThread, setNumCapabilities )
 +#endif
 +import Control.Concurrent.STM
 +  ( TVar, atomically, check, modifyTVar', newTVarIO, readTVar, writeTVar )
 +import Control.Exception
 +  ( AsyncException(ThreadKilled), SomeAsyncException, SomeException
 +  , finally, fromException, mask, mask_, onException
 +  , throwIO, try, uninterruptibleMask_ )
 +import Control.Monad
 +  ( replicateM )
 +import Data.Foldable
 +  ( for_ )
 +import Data.IORef
 +  ( IORef, atomicModifyIORef', modifyIORef', newIORef, readIORef, writeIORef )
 +import qualified Data.Map as Map
 +import qualified Data.Sequence as Seq
 +import qualified Data.Set as Set
++
 +--------------------------------------------------------------------------------
 +-- * Worker Limit
 +--------------------------------------------------------------------------------
++
 +-- | A limit on the number of concurrent worker threads.
 +data WorkerLimit
 +  -- | Fixed concurrent worker count limit @-jN@
 +  = NumProcessorsLimit Int
 +  -- | The concurrent worker count is limited by a @-jsem@ semaphore
 +  | JSemLimit
 +      SemaphoreIdentifier
 +        -- ^ Semaphore identifier (from the @semaphore-compat@ library)
 +  deriving Eq
++
 +isWorkerLimitSequential :: WorkerLimit -> Bool
 +isWorkerLimitSequential (NumProcessorsLimit x) = x <= 1
 +isWorkerLimitSequential (JSemLimit {})         = False
++
 +runWorkerLimit
 +  :: (SemaphoreError -> IO ())
 +     -- ^ report failure when opening the @-jsem@ semaphore
 +     -- (after which we fall back to running with a single job)
 +  -> WorkerLimit -> (AbstractSem -> IO a) -> IO a
 +#if defined(wasm32_HOST_ARCH) || defined(javascript_HOST_ARCH)
 +runWorkerLimit _report_semaphore_failure _ action = do
 +  lock <- newMVar ()
 +  action $ AbstractSem (takeMVar lock) (putMVar lock ())
 +#else
 +runWorkerLimit report_semaphore_failure worker_limit action = case worker_limit of
 +    NumProcessorsLimit n_jobs ->
 +      runNjobsAbstractSem n_jobs action
 +    JSemLimit sem_ident ->
 +      runJSemAbstractSem sem_ident action >>= \case
 +        Right a -> return a
 +        Left err -> do
 +          report_semaphore_failure err
 +          runNjobsAbstractSem 1 action
 +#endif
++
 +#if !(defined(wasm32_HOST_ARCH) || defined(javascript_HOST_ARCH))
 +runNjobsAbstractSem :: Int -> (AbstractSem -> IO a) -> IO a
 +runNjobsAbstractSem n_jobs action = do
 +  compile_sem <- newQSem n_jobs
 +  n_capabilities <- getNumCapabilities
 +  n_cpus <- getNumProcessors
 +  let
 +    asem = AbstractSem (waitQSem compile_sem) (signalQSem compile_sem)
 +    set_num_caps n = unless (n_capabilities /= 1) $ setNumCapabilities n
 +    updNumCapabilities =  do
 +      -- Setting number of capabilities more than
 +      -- CPU count usually leads to high userspace
 +      -- lock contention. #9221
 +      set_num_caps $ min n_jobs n_cpus
 +    resetNumCapabilities = set_num_caps n_capabilities
 +  MC.bracket_ updNumCapabilities resetNumCapabilities $ action asem
++
 +#endif
++
 +--------------------------------------------------------------------------------
 +-- * Workers
 +--------------------------------------------------------------------------------
++
 +data Concurrency
 +  = Serial
 +  | Concurrent !ConcurrencyEnv
++
 +-- | Run an action with the given concurrency control (serial or concurrent).
 +withConcurrency :: Concurrency -> IO a -> IO a
 +withConcurrency conc act =
 +  case conc of
 +    Serial -> act
 +    Concurrent ( ConcurrencyEnv { ce_semaphore } ) ->
 +      withAbstractSem ce_semaphore act
++
 +data ConcurrencyEnv =
 +  ConcurrencyEnv
 +    { ce_semaphore         :: !AbstractSem
 +    , ce_log_queue_queue   :: !( TVar LogQueueQueue )
 +    , ce_next_log_queue_id :: !( IORef Int )
 +    }
++
 +-- | The local environment of a worker thread that may be scheduled concurrently.
 +data ConcurrentWorkerEnv = ConcurrentWorkerEnv
 +  { cwe_logger :: !Logger
 +  , cwe_tmpfs  :: !TmpFs
 +  }
++
 +-- | Run an action with a local 'TmpFs', merging in the resulting temporary file
 +-- accumulator into the parent afterwards.
 +workerEnv_withLocalTmpFS :: ConcurrentWorkerEnv -> (ConcurrentWorkerEnv -> IO a) -> IO a
 +workerEnv_withLocalTmpFS env use =
 +  withLocalTmpFS (cwe_tmpfs env) \ lcl_tmpfs ->
 +    use env { cwe_tmpfs = lcl_tmpfs }
++
 +-- | Run an action either serially or concurrently based on the provided
 +-- 'WorkerLimit'.
 +withWorkerLimit
 +  :: Logger
 +  -> TmpFs
 +  -> (SemaphoreError -> IO ())
 +     -- ^ report a failure to open the @-jsem@ semaphore
 +     -- (after which we fall back to running with a single job)
 +  -> WorkerLimit
 +  -> (Concurrency -> ConcurrentWorkerEnv -> IO a) -- ^ action to run
 +  -> IO a
 +withWorkerLimit logger tmpfs report_semaphore_failure limit action
 +  | isWorkerLimitSequential limit
 +  = action Serial $
 +      ConcurrentWorkerEnv
 +        { cwe_logger = logger
 +        , cwe_tmpfs  = tmpfs
 +        }
 +  | otherwise
 +  = do
 +      safe_logger     <- makeThreadSafe logger
 +      lqq_var         <- newTVarIO newLogQueueQueue
 +      stopped_var     <- newTVarIO False
 +      wait_log_thread <- logThread safe_logger stopped_var lqq_var
 +      next_logq_var   <- newIORef 1
++
 +      let
 +        stop_logging :: IO ()
 +        stop_logging = do
 +          atomically $ writeTVar stopped_var True
 +          wait_log_thread
++
 +        parent_work_env :: ConcurrentWorkerEnv
 +        parent_work_env =
 +          ConcurrentWorkerEnv
 +            { cwe_logger = safe_logger
 +            , cwe_tmpfs  = tmpfs
 +            }
++
 +      ( `finally` stop_logging ) $
 +        runWorkerLimit report_semaphore_failure limit \ sem -> do
 +          let
 +            conc =
 +              Concurrent $
 +                ConcurrencyEnv
 +                  { ce_semaphore         = sem
 +                  , ce_log_queue_queue   = lqq_var
 +                  , ce_next_log_queue_id = next_logq_var
 +                  }
 +          action conc parent_work_env
++
 +--------------------------------------------------------------------------------
 +-- * Scheduling concurrent workers
 +--------------------------------------------------------------------------------
++
 +-- | Internal scheduler abstraction with two capabilities:
 +--
 +--  - spawn a new worker thread
 +--  - wait for a worker thread to complete
 +data Scheduler r = Scheduler
 +  { spawnWorker :: (ConcurrentWorkerEnv -> IO r) -> IO ()
 +    -- ^ Spawn one concurrent worker.
 +    --
 +    -- The worker does not hold a token of the concurrency semaphore: the
 +    -- worker action should use 'withConcurrency' around the work whose
 +    -- concurrency should be limited.
 +  , awaitWorker :: IO (Either SomeException r)
 +    -- ^ Wait for one worker to complete.
 +    --
 +    -- Will crash if there are no outstanding workers.
 +  }
++
 +-- | Internal implementation of a concurrent worker scheduler.
 +--
 +-- Usage of this function requires the following:
 +--
 +--  - all spawn/await actions are performed by a single thread,
 +--  - we never wait for more workers than were spawned,
 +--  - no worker outlives 'run_schedule'.
 +run_schedule
 +  :: forall r a
 +  .  String
 +      -- ^ thread label for workers
 +  -> Concurrency
 +  -> ConcurrentWorkerEnv
 +  -> (Scheduler r -> IO a)
 +        -- ^ worker action
 +  -> IO a
 +run_schedule worker_label conc parent_work_env withScheduler =
 +  case conc of
++
 +    Serial -> do
 +      results_var <- newIORef Seq.empty
 +      let
 +        spawnWorker :: (ConcurrentWorkerEnv -> IO r) -> IO ()
 +        spawnWorker action = do
 +          res <- try @SomeException $
 +            workerEnv_withLocalTmpFS parent_work_env action
 +          case res of
 +            Left e
 +              | Just _ <- fromException @SomeAsyncException e
 +              -> throwIO e
 +            _ -> modifyIORef' results_var (Seq.|> res)
++
 +        awaitWorker :: IO (Either SomeException r)
 +        awaitWorker =
 +          readIORef results_var >>= \case
 +            res Seq.:<| rest -> do
 +              writeIORef results_var rest
 +              pure res
 +            Seq.Empty ->
 +              panic "run_schedule: no outstanding job"
++
 +      withScheduler $ Scheduler { spawnWorker, awaitWorker }
++
 +    Concurrent ( ConcurrencyEnv { ce_next_log_queue_id, ce_log_queue_queue } ) -> do
 +      worker_tids_var <- newTVarIO $ Set.empty @ThreadId
 +      all_results_vars_var <- newIORef $ Seq.empty @(MVar (Either SomeException r))
++
 +      let
 +        wait_for_workers :: IO ()
 +        wait_for_workers =
 +          atomically $
 +            check . Set.null =<< readTVar worker_tids_var
++
 +        cancel_workers :: IO ()
 +        cancel_workers = do
 +          uninterruptibleMask_ do
 +            tids <- atomically $ readTVar worker_tids_var
 +            for_ tids killThread
 +          wait_for_workers
++
 +        awaitWorker :: IO (Either SomeException r)
 +        awaitWorker =
 +          readIORef all_results_vars_var >>= \case
 +            first_worker_res_var Seq.:<| rest -> do
 +              writeIORef all_results_vars_var rest
 +              -- block on the earliest-spawned outstanding worker
 +              takeMVar first_worker_res_var
 +            Seq.Empty ->
 +              panic "run_schedule: no outstanding job"
++
 +        spawnWorker :: (ConcurrentWorkerEnv -> IO r) -> IO ()
 +        spawnWorker action = mask_ do
++
 +          worker_res_var <- newEmptyMVar
++
 +          -- TmpFs
 +          lcl_tmpfs <- forkTmpFsFrom (cwe_tmpfs parent_work_env)
++
 +          -- LogQueue
 +          lq <- do
 +            job_id <- atomicModifyIORef' ce_next_log_queue_id \n -> (n + 1, n)
 +            lq <- newLogQueue job_id
 +            atomically $ initLogQueue ce_log_queue_queue lq
 +            pure lq
++
 +          let
++
 +            worker_work_env :: ConcurrentWorkerEnv
 +            worker_work_env =
 +              parent_work_env
 +                { cwe_tmpfs  = lcl_tmpfs
 +                , cwe_logger = pushLogHook (const (parLogAction lq))
 +                                (cwe_logger parent_work_env)
 +                }
++
 +            -- Run a worker action and record its result.
 +            run_worker_and_record :: IO r -> IO ()
 +            run_worker_and_record worker_action = do
 +              res <- try @SomeException worker_action
 +              case res of
 +                Left e
 +                  -- Worker is being cancelled: don't record anything.
 +                  | Just ThreadKilled <- fromException e
 +                  -> pure ()
 +                _ -> putMVar worker_res_var res
++
 +            -- Record that a worker thread is done.
 +            mark_worker_done :: ThreadId -> IO ()
 +            mark_worker_done tid =
 +              uninterruptibleMask_ do
 +                -- Uninterruptible: the deletion below /must/ occur.
 +                -- An uninterruptible mask is OK as we only ever block for (GAP) below.
 +                mergeTmpFsInto lcl_tmpfs $ cwe_tmpfs parent_work_env
 +                finishLogQueue lq
 +                atomically do
 +                  tids <- readTVar worker_tids_var
 +                  check $ tid `Set.member` tids
 +                    -- Ensure we never end up with a dead ThreadId in 'worker_tids_var'
 +                    -- (if the worker thread finishes before the parent thread has
 +                    -- the time to add its ThreadId to 'worker_tids_var').
++
 +                  writeTVar worker_tids_var $ Set.delete tid tids
++
 +            run_worker :: (forall b. IO b -> IO b) -> IO ()
 +            run_worker unmask = do
 +              tid <- myThreadId
 +              labelThread tid worker_label
 +              let
 +                worker_action :: IO r
 +                worker_action = unmask $ action worker_work_env
++
 +              run_worker_and_record worker_action `finally`
 +                mark_worker_done tid
++
 +          worker_tid <-
 +            forkIOWithUnmask run_worker
 +              `onException` finishLogQueue lq
 +          -- Very short (GAP) between forking the thread and recording its ThreadId.
 +          atomically $ modifyTVar' worker_tids_var $ Set.insert worker_tid
 +          modifyIORef' all_results_vars_var (Seq.|> worker_res_var)
++
 +      mask \ restore -> do
 +        result <- restore (withScheduler $ Scheduler { spawnWorker, awaitWorker })
 +                    `onException` cancel_workers
 +        restore wait_for_workers `onException` cancel_workers
 +        pure result
++
 +--------------------------------------------------------------------------------
 +-- * Derived scheduling functionality
 +--------------------------------------------------------------------------------
++
 +-- | Map a worker action over the input list with the given concurrency control.
 +--
 +-- Workers run to completion (no early abort); the first exception
 +-- (in input order) is rethrown at the end.
 +mapConcurrentWorkers
 +  :: String -- ^ thread label for workers
 +  -> Concurrency
 +  -> ConcurrentWorkerEnv
 +  -> (ConcurrentWorkerEnv -> a -> IO b)
 +      -- ^ individual worker action
 +      --
 +      -- NB: workers do not hold semaphore tokens by default; use
 +      -- 'withConcurrency' to acquire one
 +  -> [a]
 +  -> IO [b]
 +mapConcurrentWorkers worker_label conc work_env f xs =
 +  run_schedule worker_label conc work_env \ scheduler -> do
 +    for_ xs \ x -> spawnWorker scheduler \ worker_env -> f worker_env x
 +    results <- replicateM (length xs) (awaitWorker scheduler)
 +    either throwIO pure (sequence results)
++
 +-- | Depth-first traversal with on-the-fly expansion of nodes.
 +--
 +-- Each expansion step is handled by a worker thread under the given
 +-- concurrency control.
 +--
 +-- Deterministic: expansions are consumed in the order the nodes were
 +-- discovered, so the traversal is a function of the node graph alone.
 +--
 +-- Fails fast: the first worker exception cancels the outstanding workers and
 +-- is rethrown.
 +concurrentTraversal_DF
 +  :: forall k n r
 +  .  Ord k
 +  => String -- ^ thread label for workers
 +  -> Concurrency
 +  -> ConcurrentWorkerEnv
 +  -> Map.Map k r
 +     -- ^ results known ahead of time (no expansion needed)
 +  -> [n]
 +     -- ^ root nodes
 +  -> (n -> k)
 +     -- ^ node key from node
 +  -> (ConcurrentWorkerEnv -> n -> IO (r, [n]))
 +     -- ^ worker action: expand a node into its result and the children to visit next
 +     --
 +     -- NB: workers do not hold semaphore tokens by default; use
 +     -- 'withConcurrency' to acquire one
 +  -> IO (Map.Map k r)
 +concurrentTraversal_DF worker_label conc work_env base_map roots key expand =
 +  run_schedule worker_label conc work_env \ scheduler -> do
 +    let
 +      expand_node :: n -> ConcurrentWorkerEnv -> IO (k, (r, [n]))
 +      expand_node node worker_env = do
 +        res <- expand worker_env node
 +        pure (key node, res)
++
 +      go
 +        :: Map.Map k r -- expanded nodes and their results
 +        -> Set.Set k   -- nodes currently being expanded
 +        -> [n]         -- discovered nodes, to expand next
 +        -> IO (Map.Map k r)
 +      go !visited !pending (node : worklist)
 +        | k `Set.member` pending || k `Map.member` visited
 +        = go visited pending worklist
 +        | otherwise
 +        = do spawnWorker scheduler (expand_node node)
 +             go visited (Set.insert k pending) worklist
 +        where
 +          k = key node
 +      go visited pending []
 +        | Set.null pending
 +        = pure visited
 +        | otherwise
 +        = awaitWorker scheduler >>= \case
 +            Left e -> throwIO e
 +            Right (k, (result, children)) ->
 +              go (Map.insert k result visited) (Set.delete k pending) children
++
 +    go base_map Set.empty roots

compiler/GHC/Driver/Config/Concurrency.hs

 +-- | Subsystem configuration for 'GHC.Driver.Concurrency'.
 +module GHC.Driver.Config.Concurrency
 +  ( mkWorkerLimit
 +  , semaphoreOpenFailureHandler
 +  ) where
++
 +import GHC.Prelude
++
 +import GHC.Driver.Concurrency
 +import GHC.Driver.Config.Diagnostic ( initDiagOpts, initPrintConfig )
 +import GHC.Driver.DynFlags
 +import GHC.Driver.Errors ( printOrThrowDiagnostics )
 +import GHC.Driver.Errors.Types
++
 +import GHC.Types.Error ( singleMessage )
 +import GHC.Types.SrcLoc ( noSrcSpan )
 +import GHC.Utils.Error ( mkPlainMsgEnvelope )
 +import GHC.Utils.Logger ( Logger )
++
 +import GHC.Conc ( getNumProcessors )
 +import System.Semaphore ( SemaphoreError )
++
 +--------------------------------------------------------------------------------
++
 +-- | Compute the 'WorkerLimit' from the @-j@\/@-jsem@ flags.
 +mkWorkerLimit :: DynFlags -> IO WorkerLimit
 +mkWorkerLimit dflags =
 +  case parMakeCount dflags of
 +    Nothing -> pure $ num_procs 1
 +    Just (ParMakeSemaphore h) -> pure (JSemLimit h)
 +    Just ParMakeNumProcessors -> num_procs <$> getNumProcessors
 +    Just (ParMakeThisMany n) -> pure $ num_procs n
 +  where
 +    num_procs x = NumProcessorsLimit (max 1 x)
++
 +-- | Report that the semaphore specified using the @-jsem@ flag could not be opened.
 +semaphoreOpenFailureHandler :: Logger -> DynFlags -> SemaphoreError -> IO ()
 +semaphoreOpenFailureHandler logger dflags err = do
 +  let diag = DriverSemaphoreOpenFailure (checkBuildingCabalPackage dflags) err
 +      msg  = singleMessage $ mkPlainMsgEnvelope (initDiagOpts dflags) noSrcSpan diag
 +  printOrThrowDiagnostics logger (initPrintConfig dflags) (initDiagOpts dflags) (GhcDriverMessage <$> msg)

compiler/GHC/Driver/Downsweep.hs

@@ -40,9 +40,9 @@ import GHC.Driver.Monad
  import GHC.Driver.Env
  import GHC.Driver.Errors
  import GHC.Driver.Errors.Types
 -import GHC.Driver.Messager
 -import GHC.Driver.MakeSem
 +import GHC.Driver.Concurrency
  import GHC.Driver.MakeAction
 +import GHC.Driver.Config.Concurrency
  import GHC.Driver.Config.Diagnostic
  import GHC.Driver.Ppr
@@ -64,7 +64,6 @@ import GHC.Data.OsPath     ( OsPath, unsafeEncodeUtf )
  import GHC.Data.StringBuffer
  import GHC.Data.Graph.Directed.Reachability
 -import GHC.Utils.Exception ( throwIO, SomeAsyncException, AsyncException (..) )
  import GHC.Utils.Outputable
  import GHC.Utils.Panic
  import GHC.Utils.Misc
@@ -101,7 +100,6 @@ import qualified Data.Set as Set
  import Control.Concurrent.MVar
  import Control.Monad
  import Control.Monad.Trans.Except ( ExceptT(..), runExceptT, throwE )
 -import qualified Control.Monad.Catch as MC
  import Data.Maybe
  import Data.List (partition)
  import Data.Time
@@ -112,13 +110,8 @@ import System.FilePath
  import Control.Monad.Trans.Reader
  import qualified Data.Map.Strict as M
 -import Control.Monad.Trans.Class
  import System.IO.Unsafe (unsafeInterleaveIO)
  import qualified Data.List.NonEmpty as NE
 -import Control.Concurrent
 -import Control.Concurrent.STM.TQueue
 -import Control.Concurrent.STM
 -import Control.Applicative
  {-
  Note [The ModuleGraph]
@@ -245,8 +238,6 @@ See Note [The ModuleGraph] for an overview when we do downsweep.
  --
  -- See also Note [The ModuleGraph]
  downsweep :: HscEnv
 -          -> (GhcMessage -> AnyGhcDiagnostic)
 -          -> Maybe Messager
            -> [ModSummary]
            -- ^ Old summaries
            -> Maybe ModuleGraph
@@ -260,13 +251,14 @@ downsweep :: HscEnv
                  -- The non-error elements of the returned list all have distinct
                  -- (Modules, IsBoot) identifiers, unless the Bool is true in
                  -- which case there can be repeats
 -downsweep hsc_env diag_wrapper msg old_summaries maybe_base_graph excl_mods allow_dup_roots = do
 +downsweep hsc_env old_summaries maybe_base_graph excl_mods allow_dup_roots = do
    n_jobs     <- mkWorkerLimit (hsc_dflags hsc_env)
    summ_cache <- newMVar (mkModSummaryCache (zip old_summaries (repeat SummOld)))
    imps_cache <- newMVar Map.empty
 -  withMakeEnv n_jobs hsc_env diag_wrapper msg $ \make_env -> do
 -    (root_errs, root_summaries) <- rootSummariesParallel n_jobs make_env (hsc_targets hsc_env)
 -                                     (getRootSummary excl_mods summ_cache imps_cache)
 +  withWorkerLimitHsc hsc_env n_jobs $ \conc hsc_env' -> do
 +    (root_errs, root_summaries) <-
 +      rootSummariesParallel conc hsc_env' (hsc_targets hsc_env)
 +        (getRootSummary excl_mods summ_cache imps_cache)
      let closure_errs = checkHomeUnitsClosed unit_env
          unit_env = hsc_unit_env hsc_env
@@ -275,13 +267,12 @@ downsweep hsc_env diag_wrapper msg old_summaries maybe_base_graph excl_mods allo
      case all_errs of
        [] -> do
           let env = DownsweepEnv
 -               { ds_hsc_env         = hsc_env
 +               { ds_hsc_env         = hsc_env'
                 , ds_summaries_cache = summ_cache
                 , ds_imports_cache   = imps_cache
                 , ds_mode            = DownsweepUseCompile
                 , ds_excl_mods       = excl_mods
 -               , ds_n_jobs          = n_jobs
 -               , ds_make_env        = make_env
 +               , ds_concurrency     = conc
+                }
           (downsweep_errs, downsweep_nodes) <- runDownsweepM env $
              downsweepFromRootNodes maybe_base_graph allow_dup_roots
@@ -349,15 +340,14 @@ downsweepThunk hsc_env mod_summary = unsafeInterleaveIO $ do
    njobs <- mkWorkerLimit (hsc_dflags hsc_env)
    summs <- newMVar (mkModSummaryCache [(mod_summary,SummOld)])
    imps  <- newMVar mempty
 -  withMakeEnv njobs hsc_env mkUnknownDiagnostic Nothing $ \make_env -> do
 +  withWorkerLimitHsc hsc_env njobs $ \conc hsc_env' -> do
      let env = DownsweepEnv
 -          { ds_hsc_env         = hsc_env
 +          { ds_hsc_env         = hsc_env'
            , ds_summaries_cache = summs
            , ds_imports_cache   = imps
            , ds_mode            = DownsweepUseFixed
            , ds_excl_mods       = []
 -          , ds_n_jobs          = njobs
 -          , ds_make_env        = make_env
 +          , ds_concurrency     = conc
+           }
      ~(errs, mg) <- runDownsweepM env $
        downsweepFromRootNodes Nothing True
@@ -394,15 +384,14 @@ downsweepInteractiveImports hsc_env ic = unsafeInterleaveIO $ do
    n_jobs     <- mkWorkerLimit (hsc_dflags hsc_env)
    summ_cache <- newMVar mempty
    imps_cache <- newMVar mempty
 -  withMakeEnv n_jobs hsc_env mkUnknownDiagnostic Nothing $ \make_env -> do
 +  withWorkerLimitHsc hsc_env n_jobs $ \conc hsc_env' -> do
      let env = DownsweepEnv
 -          { ds_hsc_env         = hsc_env
 +          { ds_hsc_env         = hsc_env'
            , ds_mode            = DownsweepUseFixed{-or DownsweepUseCompile?-}
            , ds_summaries_cache = summ_cache
            , ds_imports_cache   = imps_cache
            , ds_excl_mods       = []
 -          , ds_n_jobs          = n_jobs
 -          , ds_make_env        = make_env
 +          , ds_concurrency     = conc
+           }
      graph <- runDownsweepM env do
        loopFromInteractive cached_nodes interactive_mn imps
@@ -439,15 +428,14 @@ downsweepInstalledModules hsc_env mods = do
      nodes <- mapM process installed_mods
      summs <- newMVar mempty
      imps  <- newMVar mempty
 -    withMakeEnv njobs hsc_env mkUnknownDiagnostic Nothing $ \make_env -> do
 +    withWorkerLimitHsc hsc_env njobs $ \conc hsc_env' -> do
        let env = DownsweepEnv
 -            { ds_hsc_env         = hsc_env
 +            { ds_hsc_env         = hsc_env'
              , ds_summaries_cache = summs
              , ds_imports_cache   = imps
              , ds_mode            = DownsweepUseFixed
              , ds_excl_mods       = []
 -            , ds_n_jobs          = njobs
 -            , ds_make_env        = make_env
 +            , ds_concurrency     = conc
+             }
        (errs, mg) <- runDownsweepM env $
          downsweepFromRootNodes Nothing True nodes external_uids
@@ -562,8 +550,8 @@ data DownsweepEnv = DownsweepEnv {
      , ds_summaries_cache :: ModSummaryCache
      , ds_imports_cache   :: ImportsCache
      , ds_excl_mods       :: [ModuleName]
 -    , ds_n_jobs          :: WorkerLimit
 -    , ds_make_env        :: MakeEnv
 +    , ds_concurrency     :: Concurrency
 +      -- ^ The concurrency to use for downsweep
+ }
  mkModSummaryCache :: [(ModSummary, SummProvenance)] -> ModSummaryCacheMap
@@ -928,15 +916,28 @@ getRootSummary excl_mods summ_cache imports_cache hsc_env target
        rootLoc = mkGeneralSrcSpan (fsLit "<command line>")
        dflags = homeUnitEnv_dflags (ue_findHomeUnitEnv uid (hsc_unit_env hsc_env))
 --- | Execute 'getRootSummary' for the 'Target's using the parallelism pipeline system.
 +-- | Execute 'getRootSummary' for the 'Target's in bundles, spawning one
 +-- worker per bundle. The number of bundles processed at once is limited by
 +-- the given 'Concurrency'.
  rootSummariesParallel
 -  :: WorkerLimit -> MakeEnv -> [Target]
 +  :: Concurrency -> HscEnv -> [Target]
    -> (HscEnv -> Target -> IO (Either DriverMessages ModSummary))
    -> IO ([DriverMessages], [ModSummary])
 -rootSummariesParallel n_jobs make_env targets get_summary = do
 -  partitionEithers <$> mapConcDS n_jobs bundle_size make_env get_summary targets
 -    where
 -      bundle_size = 20
 +rootSummariesParallel conc hsc_env targets get_summary = do
 +  results <-
 +    mapConcurrentWorkers "root_summary_worker" conc (viewHscWorkerEnv hsc_env)
 +      ( \ work_env bundle ->
 +          withConcurrency conc $
 +            mapM (get_summary (setHscWorkerEnv work_env hsc_env)) bundle )
 +      bundles
 +  pure $ partitionEithers (concat results)
 +  where
 +    bundle_size = 20
++
 +    bundles = mk_bundles targets
 +    mk_bundles = unfoldr \case
 +      [] -> Nothing
 +      ts -> Just (splitAt bundle_size ts)
  --------------------------------------------------------------------------------
  -- * Check/validate properties and error out
@@ -1762,7 +1763,7 @@ data NodeRes v
  -- node. The result includes the previously visited nodes given in @base_map@,
  -- s.t. @parDfsBuild base_map [] _ _ == base_map@.
  --
 --- The @expand@ function returns an 'NResult'. See the 'NResult' documentation
 +-- The @expand@ function returns a 'NodeRes'. See the 'NodeRes' documentation
  -- for more information about each result type.
  --
  -- Example usage: @n@ is instanced to @DownsweepNode@, @k@ is @NodeKey@, and @v@ is @ModuleNodeEdge@.
@@ -1785,90 +1786,25 @@ parDfsBuild :: forall k v n. Ord k
           -- ^ The result accumulates the payload of expanding the root nodes
           -- and all nodes transitively reachable from those roots.
  parDfsBuild base_map roots key expand = ReaderT $ \ds_env -> do
 -    exc_var     <- newTVarIO $ Nothing @MC.SomeException
 -    visited_var <- newTVarIO $ fromMaybe Map.empty base_map
 -    pending     <- newTVarIO $ Set.empty @k
 -    worklist    <- newTQueueIO @n
 -    threads     <- newTVarIO []
+-
 -    coord_tid   <- forkIO $
 -      coordinator ds_env exc_var visited_var worklist pending threads
 -        `MC.catch` \case
 -          (e::MC.SomeException)
 -            -- exit cleanly when killed
 -            | Just ThreadKilled <- fromException e -> return ()
 -            -- if the coordinator somehow else crashes,
 -            -- signal the exc_var for the main thread to throw it
 -            | otherwise -> atomically (modifyTVar' exc_var (<|> Just e))
+-
 -    atomically $ mapM_ (writeTQueue worklist) roots
+-
 -    mb_exc <- wait_done exc_var worklist pending
 -      `MC.finally` do
 -        killThread coord_tid
 -        mapM_ killThread =<< readTVarIO threads
+-
 -    case mb_exc of
 -      Just e  -> throwIO e
 -      Nothing -> readTVarIO visited_var
 -  where
 -    wait_done exc_var worklist pending =
 -      -- this txn retries until all work is done or an exception is signaled
 -      atomically $ do
 -        readTVar exc_var >>= \case
 -          Just e -> return (Just e)
 -          Nothing -> do
 -            empty_worklist <- isEmptyTQueue worklist
 -            empty_pending  <- Set.null <$> readTVar pending
 -            check (empty_worklist && empty_pending)
 -            return Nothing
+-
 -    coordinator ds_env exc_var visvar worklist pendvar threads = forever $ do
 -      mb_node_to_expand <- atomically $ do
 -        node <- readTQueue worklist
 -        let k = key node
+-
 -        visited <- readTVar visvar
 -        pending <- readTVar pendvar
+-
 -        if (k `Set.member` pending || k `Map.member` visited)
 -          then return Nothing
 -          else do
 -            -- must add to pending in the same transaction as worklist dequeue,
 -            -- otherwise the main thread may find both the worklist and pending
 -            -- lists empty and exit prematurely.
 -            modifyTVar' pendvar (Set.insert k)
 -            return (Just (k, node))
+-
 -      case mb_node_to_expand of
 -        Nothing        -> return ()
 -        Just (k, node) -> do
 -          tid <- MC.mask_ $ forkIOWithUnmask $ \unmask ->
 -            unmask (withLocalTmpFSMake (ds_make_env ds_env) $ \make_env ->
 -              worker ds_env{ds_make_env = make_env} visvar worklist pendvar k node)
 -                `MC.catch` \case
 -                  e | Just (_ :: SomeAsyncException) <- fromException e
 -                    -> throwIO e -- async exceptions like KillThread get thrown
 -                    | otherwise  -- exceptions in workers are written for main thread
 -                    -> atomically (modifyTVar' exc_var (<|> Just e))
+-
 -          atomically $ modifyTVar' threads (tid:)
+-
 -    worker ds_env@DownsweepEnv{..} visvar worklist pendvar k node =
 -      withAbstractSem (compile_sem ds_make_env) $ do
 -        r <- runDownsweepM ds_env $
 -             expand node -- do the main work!
+-
 -        atomically $ do
 -          case r of
 -            NSkip ->
 -              modifyTVar' visvar (Map.insert k NSkip)
 -            NSuccess (v,ns) -> do
 -              modifyTVar' visvar (Map.insert k (NSuccess v))
 -              mapM_ (writeTQueue worklist) ns
+-
 -          modifyTVar' pendvar (Set.delete k)
 +  let
 +    conc :: Concurrency
 +    conc = ds_concurrency ds_env
++
 +    expand_node :: ConcurrentWorkerEnv -> n -> IO (NodeRes v, [n])
 +    expand_node worker_env node = do
 +      result <- withConcurrency conc $
 +        runDownsweepM (setDownsweepWorkerEnv worker_env ds_env) (expand node)
 +      pure $ case result of
 +        NSkip                    -> (NSkip, [])
 +        NSuccess (val, new_work) -> (NSuccess val, new_work)
++
 +  concurrentTraversal_DF "downsweep_worker" conc (viewHscWorkerEnv $ ds_hsc_env ds_env)
 +    (fromMaybe mempty base_map) roots key expand_node
++
 +setDownsweepWorkerEnv :: ConcurrentWorkerEnv -> DownsweepEnv -> DownsweepEnv
 +setDownsweepWorkerEnv work_env env =
 +  env { ds_hsc_env = setHscWorkerEnv work_env (ds_hsc_env env) }
  {-
  Note [Downsweep Control Flow and Caching]
@@ -1963,70 +1899,10 @@ things, and that processing can often be costly (e.g. see `expandModuleSummary`)
  We leverage multiple threads in this traversal to expand more than one module
  at once, respecting -j<N> to mean we never expand more than N modules at once.
 -The parallel downsweep is all handled by `parDfsBuild` as follows:
+-
 -- We launch a thread for every module we discover that needs to be
 -  expanded in the `coordinator` thread, popping it from the worklist
 -- Every launched `worker` thread blocks waiting for a semaphore token
 -  (`withAbstractSem`) to respect -j<N>
 -- The main thread waits until both the worklist and pending list is
 -  cleared, atomically.
+-
 -STM is used crucially to guarantee e.g. we don't have race conditions
 -between taking from the worklist and writing to the pending list while
 -checking whether they are clear.
+-
 -Exceptions are bubbled up to the main thread. The "main" thread, which is
 -typically waiting for the worklist+pending lists to be clear, instead gets
 -unblocked by this exception (signaled in `exc_var`) and re-throws it.
 --}
 ---------------------------------------------------------------------------------
 --- * Concurrent utilities
 ---------------------------------------------------------------------------------
+-
 --- | Map an action over a list using the parallelism pipeline system.
 --- Create bundles of the list elems wrapped in a 'MakeAction' that uses
 --- 'withAbstractSem' to wait for a free slot, limiting the number of
 --- concurrently computed summaries to the value of the @-j@ option or the slots
 --- allocated by the job server, if that is used.
 ---
 --- The 'MakeAction' returns 'Maybe', which is not handled as an error, because
 --- 'runLoop' only sets it to 'Nothing' when an exception was thrown, so the
 --- result won't be read anyway here.
 ---
 --- To emulate the current behavior, we funnel exceptions past the concurrency
 --- barrier and rethrow the first one afterwards.
 -mapConcDS ::
 -  WorkerLimit ->
 -  Int {-^ Batch size -} ->
 -  MakeEnv ->
 -  (HscEnv -> a -> IO b) ->
 -  [a] ->
 -  IO ([b])
 -mapConcDS n_jobs bundle_size make_env run_action xs = do
 -  (actions, get_results) <- unzip <$> mapM action_and_result (zip [1..] bundles)
 -  runAllPipelines n_jobs make_env actions
 -  (sequence . catMaybes <$> sequence get_results) >>= \case
 -    Right results -> pure (concat results)
 -    Left exc -> throwIO exc
 -  where
 -    bundles = mk_bundles xs
+-
 -    mk_bundles = unfoldr \case
 -      [] -> Nothing
 -      ts -> Just (splitAt bundle_size ts)
+-
 -    action_and_result (log_queue_id, ts) = do
 -      res_var <- liftIO newEmptyMVar
 -      pure $! (MakeAction (action log_queue_id ts) res_var, readMVar res_var)
+-
 -    action log_queue_id target_bundle = do
 -      env@MakeEnv {compile_sem} <- ask
 -      lift $ lift $
 -        withAbstractSem compile_sem $
 -        withLoggerHsc log_queue_id env \ lcl_hsc_env ->
 -          MC.try (mapM (run_action lcl_hsc_env) target_bundle) >>= \case
 -            Left e | Just (_ :: SomeAsyncException) <- fromException e ->
 -              throwIO e
 -            a -> pure a
 +We use the concurrent scheduling abstraction from GHC.Driver.Concurrency
 +('concurrentTraversal_DF'). Each time we discover a new node, a worker is
 +spawned to expand it. After each expansion completes, the resulting children
 +nodes are pushed onto the worklist. With -j1 no threads are involved: each
 +expansion runs in sequence.
 +-}

compiler/GHC/Driver/Make.hs

@@ -63,8 +63,9 @@ import GHC.Driver.Env
  import GHC.Driver.Errors
  import GHC.Driver.Errors.Types
  import GHC.Driver.Main
 -import GHC.Driver.MakeSem
  import GHC.Driver.Downsweep
 +import GHC.Driver.Concurrency
 +import GHC.Driver.Config.Concurrency
  import GHC.Driver.MakeAction
  import GHC.Types.UnresolvedImport
@@ -156,22 +157,20 @@ depanal :: GhcMonad m =>
  depanal excluded_mods allow_dup_roots = do
      hsc_env <- getSession
      let sec = initSourceErrorContext (hsc_dflags hsc_env)
 -    (errs, mod_graph) <- depanalE mkUnknownDiagnostic Nothing excluded_mods allow_dup_roots
 +    (errs, mod_graph) <- depanalE excluded_mods allow_dup_roots
      if isEmptyMessages errs
        then pure mod_graph
        else throwErrors sec (fmap GhcDriverMessage errs)
  -- | Perform dependency analysis like in 'depanal'.
  -- In case of errors, the errors and an empty module graph are returned.
 -depanalE :: GhcMonad m =>     -- New for #17459
 -               (GhcMessage -> AnyGhcDiagnostic)
 -            -> Maybe Messager
 -            -> [ModuleName]      -- ^ excluded modules
 +depanalE :: GhcMonad m =>
 +               [ModuleName]   -- ^ excluded modules
              -> Bool           -- ^ allow duplicate roots
              -> m (DriverMessages, ModuleGraph)
 -depanalE diag_wrapper msg excluded_mods allow_dup_roots = do
 +depanalE excluded_mods allow_dup_roots = do
      hsc_env <- getSession
 -    (errs, mod_graph) <- depanalPartial diag_wrapper msg excluded_mods allow_dup_roots
 +    (errs, mod_graph) <- depanalPartial excluded_mods allow_dup_roots
      if isEmptyMessages errs
        then do
          hsc_env <- getSession
@@ -209,13 +208,11 @@ depanalE diag_wrapper msg excluded_mods allow_dup_roots = do
  -- new module graph.
  depanalPartial
      :: GhcMonad m
 -    => (GhcMessage -> AnyGhcDiagnostic)
 -    -> Maybe Messager
 -    -> [ModuleName]  -- ^ excluded modules
 +    => [ModuleName]  -- ^ excluded modules
      -> Bool          -- ^ allow duplicate roots
      -> m (DriverMessages, ModuleGraph)
      -- ^ possibly empty 'Bag' of errors and a module graph.
 -depanalPartial diag_wrapper msg excluded_mods allow_dup_roots = do
 +depanalPartial excluded_mods allow_dup_roots = do
    hsc_env <- getSession
    let
           targets = hsc_targets hsc_env
@@ -234,7 +231,7 @@ depanalPartial diag_wrapper msg excluded_mods allow_dup_roots = do
      liftIO $ flushFinderCaches (hsc_FC hsc_env) (hsc_unit_env hsc_env)
      (errs, mod_graph) <- liftIO $ downsweep
 -      hsc_env diag_wrapper msg (mgModSummaries old_graph) Nothing
 +      hsc_env (mgModSummaries old_graph) Nothing
        excluded_mods allow_dup_roots
      return (unionManyMessages errs, mod_graph)
@@ -438,7 +435,7 @@ loadWithCache :: GhcMonad m => Maybe ModIfaceCache -- ^ Instructions about how t
                              -> m SuccessFlag
  loadWithCache cache diag_wrapper how_much = do
      msg <- mkBatchMsg <$> getSession
 -    (errs, mod_graph) <- depanalE diag_wrapper (Just msg) [] False                        -- #17459
 +    (errs, mod_graph) <- depanalE [] False                        -- #17459
      success <- load' cache how_much diag_wrapper (Just msg) mod_graph
      hsc_env <- getSession
      let sec = initSourceErrorContext (hsc_dflags hsc_env)
@@ -840,14 +837,15 @@ The Algorithm
  a pair of an `IO a` action and a `MVar a`, where to place the result.
    The list is sorted topologically, so can be executed in order without fear of
    blocking.
 -* runPipelines takes this list and eventually passes it to runLoop which executes
 -  each action and places the result into the right MVar.
 -* The amount of parallelism is controlled by a semaphore. This is just used around the
 -  module compilation step, so that only the right number of modules are compiled at
 -  the same time which reduces overall memory usage and allocations.
 -* Each proper node has a LogQueue, which dictates where to send it's output.
 -* The LogQueue is placed into the LogQueueQueue when the action starts and a worker
 -  thread processes the LogQueueQueue printing logs for each module in a stable order.
 +* runPipelines spawns one worker per action ('GHC.Driver.Concurrency.mapConcurrentWorkers'),
 +  which executes the action and places the result into the right MVar.
 +* The amount of parallelism is controlled by a semaphore ('withMakeEnvConcurrency'). This is
 +  just used around the module compilation step, so that only the right number of
 +  modules are compiled at the same time which reduces overall memory usage and
 +  allocations.
 +* Each worker has a LogQueue, which dictates where to send its output. A log
 +  thread processes the LogQueues, printing logs for each module in a stable
 +  order (the order in which the actions were spawned).
  * The result variable for an action producing `a` is of type `Maybe a`, therefore
    it is still filled on a failure. If a module fails to compile, the
    failure is propagated through the whole module graph and any modules which didn't
@@ -1137,7 +1135,7 @@ interpretBuildPlan hug mhmi_cache old_hpt plan = do
            !build_deps = getDependencies (map gwib_mod deps) build_map
        let loop_action = withCurrentUnit loop_unit $ do
              !_ <- wait_deps build_deps
 -            hsc_env <- asks hsc_env
 +            hsc_env <- asks me_hsc_env
              let mns :: [ModuleName]
                  mns = mapMaybe (nodeKeyModName . gwib_mod) deps
@@ -1180,7 +1178,7 @@ interpretBuildPlan hug mhmi_cache old_hpt plan = do
  withCurrentUnit :: UnitId -> RunMakeM a -> RunMakeM a
  withCurrentUnit uid = do
 -  local (\env -> env { hsc_env = hscSetActiveUnitId uid (hsc_env env)})
 +  local (\env -> env { me_hsc_env = hscSetActiveUnitId uid (me_hsc_env env)})
  upsweep
      :: WorkerLimit -- ^ The number of workers we wish to run in parallel
@@ -1556,10 +1554,11 @@ executeInstantiationNode k n deps uid iu = do
          env <- ask
          -- Output of the logger is mediated by a central worker to
          -- avoid output interleaving
 -        msg <- asks env_messager
 -        wrapper <- asks diag_wrapper
 -        lift $ MaybeT $ withLoggerHsc k env $ \hsc_env ->
 -          let lcl_hsc_env = setHUG deps hsc_env
 +        msg <- asks me_messager
 +        wrapper <- asks me_diag_wrapper
 +        lift $ MaybeT $
 +          let hsc_env = me_hsc_env env
 +              lcl_hsc_env = setHUG deps hsc_env
            in wrapAction wrapper lcl_hsc_env $ do
              res <- upsweep_inst lcl_hsc_env msg k n uid iu
              cleanCurrentModuleTempFilesMaybe (hsc_logger hsc_env) (hsc_tmpfs hsc_env) (hsc_dflags hsc_env)
@@ -1582,13 +1581,13 @@ executeCompileNode :: Int
    -> ModuleNodeInfo
    -> RunMakeM HomeModInfo
  executeCompileNode k n !old_hmi hug mrehydrate_mods mni = do
 -  me@MakeEnv{..} <- ask
 +  make_env <- ask
    -- Rehydrate any dependencies if this module had a boot file or is a signature file.
 -  lift $ MaybeT (withAbstractSem compile_sem $ withLoggerHsc k me $ \hsc_env -> do
 +  lift $ MaybeT (withMakeEnvConcurrency make_env $ \hsc_env -> do
       hsc_env' <- liftIO $ maybeRehydrateBefore (setHUG hug hsc_env) mni fixed_mrehydrate_mods
       case mni of
 -       ModuleNodeCompile mod -> executeCompileNodeWithSource hsc_env' me  mod
 -       ModuleNodeFixed key loc -> executeCompileNodeFixed hsc_env' me key loc
 +       ModuleNodeCompile mod -> executeCompileNodeWithSource hsc_env' make_env mod
 +       ModuleNodeFixed key loc -> executeCompileNodeFixed hsc_env' make_env key loc
+     )
    where
@@ -1601,9 +1600,9 @@ executeCompileNode k n !old_hmi hug mrehydrate_mods mni = do
          _        -> mrehydrate_mods
      executeCompileNodeFixed :: HscEnv -> MakeEnv -> ModNodeKeyWithUid -> ModLocation -> IO (Maybe HomeModInfo)
 -    executeCompileNodeFixed hsc_env MakeEnv{diag_wrapper, env_messager} mod loc =
 -      wrapAction diag_wrapper hsc_env $ do
 -        forM_ env_messager $ \hscMessage -> hscMessage hsc_env (k, n) UpToDate (ModuleNode [] (ModuleNodeFixed mod loc))
 +    executeCompileNodeFixed hsc_env MakeEnv{me_diag_wrapper, me_messager} mod loc =
 +      wrapAction me_diag_wrapper hsc_env $ do
 +        forM_ me_messager $ \hscMessage -> hscMessage hsc_env (k, n) UpToDate (ModuleNode [] (ModuleNodeFixed mod loc))
          read_result <- readIface (hsc_hooks hsc_env) (hsc_logger hsc_env) (hsc_dflags hsc_env) (hsc_NC hsc_env) (mnkToModule mod) (ml_hi_file loc)
          let sec = initSourceErrorContext (hsc_dflags hsc_env)
          case read_result of
@@ -1619,7 +1618,7 @@ executeCompileNode k n !old_hmi hug mrehydrate_mods mni = do
              return (HomeModInfo iface details hm_linkable)
      executeCompileNodeWithSource :: HscEnv -> MakeEnv -> ModSummary -> IO (Maybe HomeModInfo)
 -    executeCompileNodeWithSource hsc_env MakeEnv{diag_wrapper, env_messager} mod = do
 +    executeCompileNodeWithSource hsc_env MakeEnv{me_diag_wrapper, me_messager} mod = do
       let -- Use the cached DynFlags which includes OPTIONS_GHC pragmas
           lcl_dynflags = ms_hspp_opts mod
       let lcl_hsc_env =
@@ -1628,8 +1627,8 @@ executeCompileNode k n !old_hmi hug mrehydrate_mods mni = do
               hsc_env
       -- Compile the module, locking with a semaphore to avoid too many modules
       -- being compiled at the same time leading to high memory usage.
 -     wrapAction diag_wrapper lcl_hsc_env $ do
 -      res <- upsweep_mod lcl_hsc_env env_messager old_hmi mod k n
 +     wrapAction me_diag_wrapper lcl_hsc_env $ do
 +      res <- upsweep_mod lcl_hsc_env me_messager old_hmi mod k n
        cleanCurrentModuleTempFilesMaybe (hsc_logger hsc_env) (hsc_tmpfs hsc_env) lcl_dynflags
        return res
@@ -1853,15 +1852,15 @@ Also closely related are
  -}
  executeLinkNode :: HomeUnitGraph -> (Int, Int) -> UnitId -> [NodeKey] -> RunMakeM ()
 -executeLinkNode hug kn@(k, _) uid deps = do
 +executeLinkNode hug kn uid deps = do
    withCurrentUnit uid $ do
      make_env@MakeEnv{..} <- ask
 -    let dflags = hsc_dflags hsc_env
 -        msg' = (\messager -> \recomp -> messager hsc_env kn recomp (LinkNode deps uid)) <$> env_messager
 +    let dflags = hsc_dflags me_hsc_env
 +        msg' = (\messager -> \recomp -> messager me_hsc_env kn recomp (LinkNode deps uid)) <$> me_messager
 -    linkresult <- lift $ MaybeT $ withAbstractSem compile_sem $ withLoggerHsc k make_env $ \lcl_hsc_env -> do
 +    linkresult <- lift $ MaybeT $ withMakeEnvConcurrency make_env $ \lcl_hsc_env -> do
                               let hsc_env' = setHUG hug lcl_hsc_env
 -                             wrapAction diag_wrapper hsc_env' $ do
 +                             wrapAction me_diag_wrapper hsc_env' $ do
                                 link (ghcLink dflags)
                                   hsc_env'
                                   True -- We already decided to link

compiler/GHC/Driver/MakeAction.hs

 -{-# LANGUAGE CPP #-}
  module GHC.Driver.MakeAction
    ( MakeAction(..)
    , RunMakeM
@@ -7,79 +6,41 @@ module GHC.Driver.MakeAction
    -- * Running the pipelines
    , runAllPipelines
    , runPipelines
 -  -- * Worker limit
 -  , WorkerLimit(..)
 -  , mkWorkerLimit
 -  , runWorkerLimit
    -- * Utility
 -  , withLoggerHsc
 -  , withParLog
 -  , withLocalTmpFS
 -  , withLocalTmpFSMake
 +  , withMakeEnvConcurrency
 +  , withWorkerLimitHsc
 +  , viewHscWorkerEnv
 +  , setHscWorkerEnv
    ) where
  import GHC.Prelude
 -import GHC.Driver.DynFlags
 -import GHC.Driver.Monad
 +import GHC.Driver.Concurrency
 +import GHC.Driver.Config.Concurrency
  import GHC.Driver.Env
  import GHC.Driver.Errors.Types
  import GHC.Driver.Messager
 -import GHC.Driver.MakeSem
+-
 -#if defined(wasm32_HOST_ARCH) || defined(javascript_HOST_ARCH)
 -import System.Semaphore
 -  ( SemaphoreIdentifier )
 -#else
 -import System.Semaphore
 -  ( SemaphoreError, SemaphoreIdentifier )
 -#endif
+-
 -#if !(defined(wasm32_HOST_ARCH) || defined(javascript_HOST_ARCH))
 -import GHC.Driver.Config.Diagnostic ( initDiagOpts, initPrintConfig )
 -import GHC.Driver.Errors ( printOrThrowDiagnostics )
 -import GHC.Types.Error ( singleMessage )
 -import GHC.Types.SrcLoc ( noSrcSpan )
 -import GHC.Utils.Error ( mkPlainMsgEnvelope )
 -#endif
 -import GHC.Utils.Logger
 -import GHC.Utils.TmpFs
 +import GHC.Driver.Monad
 -#if defined(wasm32_HOST_ARCH) || defined(javascript_HOST_ARCH)
 -import Control.Concurrent ( ThreadId, killThread, forkIOWithUnmask )
 -#else
 -import Control.Concurrent ( newQSem, waitQSem, signalQSem, ThreadId, killThread, forkIOWithUnmask )
 -#endif
  import qualified GHC.Conc as CC
  import Control.Concurrent.MVar
  import Control.Monad
  import qualified Control.Monad.Catch as MC
+-
 -#if defined(wasm32_HOST_ARCH) || defined(javascript_HOST_ARCH)
 -import GHC.Conc ( getNumProcessors )
 -#else
 -import GHC.Conc ( getNumProcessors, getNumCapabilities, setNumCapabilities )
 -#endif
 -import Control.Monad.Trans.Reader
 -import GHC.Driver.Pipeline.LogQueue
 -import Control.Concurrent.STM
  import Control.Monad.Trans.Maybe
 +import Control.Monad.Trans.Reader
  --------------------------------------------------------------------------------
  -- * MakeEnv and MakeAction
  --------------------------------------------------------------------------------
  -- | Environment used when compiling a module
 -data MakeEnv = MakeEnv { hsc_env :: !HscEnv -- The basic HscEnv which will be augmented for each module
 -                       , compile_sem :: !AbstractSem
 -                       -- Modify the environment for module k, with the supplied logger modification function.
 -                       -- For -j1, this wrapper doesn't do anything
 -                       -- For -jn, the wrapper initialised a log queue and then modifies the logger to pipe its output
 -                       --          into the log queue.
 -                       , withLogger :: forall a . Int -> ((Logger -> Logger) -> IO a) -> IO a
 -                       , env_messager :: !(Maybe Messager)
 -                       , diag_wrapper :: GhcMessage -> AnyGhcDiagnostic
 -                       }
 +data MakeEnv =
 +  MakeEnv
 +    { me_hsc_env      :: !HscEnv -- The basic HscEnv which will be augmented for each module
 +    , me_concurrency  :: !Concurrency
 +    , me_messager     :: !(Maybe Messager)
 +    , me_diag_wrapper :: GhcMessage -> AnyGhcDiagnostic
 +    }
  -- | Come up with a 'MakeEnv' based on the given 'WorkerLimit'.
  -- For -j1, it will be a trivial 'MakeEnv' not prepared for parallelism.
@@ -91,47 +52,14 @@ withMakeEnv
    -> Maybe Messager -- ^ Optional custom messager to use to report progress
    -> (MakeEnv -> IO r) -> IO r
  withMakeEnv worker_limit hsc_env diag_wrapper mHscMessager act =
 -  if isWorkerLimitSequential worker_limit
 -    then withSeqMakeEnv
 -    else withParMakeEnv
 -  where
 -    withSeqMakeEnv = do
 -      let seq_env = MakeEnv
 -            { hsc_env = hsc_env
 -            , withLogger = \_ k -> k id
 -            , compile_sem = AbstractSem (return ()) (return ())
 -            , env_messager = mHscMessager
 -            , diag_wrapper = diag_wrapper
 -            }
 -      act seq_env
+-
 -    withParMakeEnv = do
 -      -- A variable which we write to when an error has happened and we have to tell the
 -      -- logging thread to gracefully shut down.
 -      stopped_var <- newTVarIO False
 -      -- The queue of LogQueues which actions are able to write to. When an action starts it
 -      -- will add it's LogQueue into this queue.
 -      log_queue_queue_var <- newTVarIO newLogQueueQueue
 -      -- Thread which coordinates the printing of logs
 -      wait_log_thread <- logThread (hsc_logger hsc_env) stopped_var log_queue_queue_var
+-
+-
 -      -- Make the logger thread-safe, in case there is some output which isn't sent via the LogQueue.
 -      thread_safe_logger <- liftIO $ makeThreadSafe (hsc_logger hsc_env)
 -      let thread_safe_hsc_env = hsc_env { hsc_logger = thread_safe_logger }
+-
 -      runWorkerLimit (hsc_logger hsc_env) (hsc_dflags hsc_env) worker_limit $ \abstract_sem -> do
 -        let env = MakeEnv { hsc_env = thread_safe_hsc_env
 -                          , withLogger = withParLog log_queue_queue_var
 -                          , compile_sem = abstract_sem
 -                          , env_messager = mHscMessager
 -                          , diag_wrapper = diag_wrapper
 -                          }
 -        -- Reset the number of capabilities once the upsweep ends.
 -        r <- act env
 -        atomically $ writeTVar stopped_var True
 -        wait_log_thread
 -        pure r
 +  withWorkerLimitHsc hsc_env worker_limit $ \ conc hsc_env' ->
 +    act $
 +      MakeEnv
 +        { me_hsc_env       = hsc_env'
 +        , me_concurrency   = conc
 +        , me_messager      = mHscMessager
 +        , me_diag_wrapper  = diag_wrapper
 +        }
  -- ** MakeAction ---------------------------------------------------------------
@@ -139,9 +67,6 @@ data MakeAction = forall a . MakeAction !(RunMakeM a) !(MVar (Maybe a))
  type RunMakeM a = ReaderT MakeEnv (MaybeT IO) a
 -waitMakeAction :: MakeAction -> IO ()
 -waitMakeAction (MakeAction _ mvar) = () <$ readMVar mvar
+-
  --------------------------------------------------------------------------------
  -- * Running the pipelines
  --------------------------------------------------------------------------------
@@ -155,149 +80,51 @@ runPipelines
  runPipelines n_job hsc_env diag_wrapper mHscMessager all_pipelines = do
    liftIO $ label_self "main --make thread"
    withMakeEnv n_job hsc_env diag_wrapper mHscMessager $ \make_env -> do
 -    runAllPipelines n_job make_env all_pipelines
 +    runAllPipelines make_env all_pipelines
    where
      label_self :: String -> IO ()
      label_self thread_name = do
          self_tid <- CC.myThreadId
          CC.labelThread self_tid thread_name
 --- | Run the given actions and then wait for them all to finish.
 -runAllPipelines :: WorkerLimit -> MakeEnv -> [MakeAction] -> IO ()
 -runAllPipelines worker_limit env acts = do
 -  let single_worker = isWorkerLimitSequential worker_limit
 -      spawn_actions :: IO [ThreadId]
 -      spawn_actions = if single_worker
 -        then (:[]) <$> (forkIOWithUnmask $ \unmask -> void $ runLoop (\io -> io unmask) env acts)
 -        else runLoop forkIOWithUnmask env acts
+-
 -      kill_actions :: [ThreadId] -> IO ()
 -      kill_actions tids = mapM_ killThread tids
+-
 -  MC.bracket spawn_actions kill_actions $ \_ -> do
 -    mapM_ waitMakeAction acts
+-
 --- | Execute each action in order, limiting the amount of parallelism by the given
 --- semaphore.
 -runLoop :: (((forall a. IO a -> IO a) -> IO ()) -> IO a) -> MakeEnv -> [MakeAction] -> IO [a]
 -runLoop _ _env [] = return []
 -runLoop fork_thread env (MakeAction act res_var :acts) = do
+-
 -  -- withLocalTmpFs has to occur outside of fork to remain deterministic
 -  new_thread <- withLocalTmpFSMake env $ \lcl_env ->
 -    MC.mask_ $
 -      fork_thread $ \unmask -> (do
 -            mres <- (unmask $ run_pipeline lcl_env act)
 -                      `MC.onException` (putMVar res_var Nothing) -- Defensive: If there's an unhandled exception then still signal the failure.
 -            putMVar res_var mres)
 -  threads <- runLoop fork_thread env acts
 -  return (new_thread : threads)
 -  where
 -      run_pipeline :: MakeEnv -> RunMakeM a -> IO (Maybe a)
 -      run_pipeline env p = runMaybeT (runReaderT p env)
+-
 ---------------------------------------------------------------------------------
 --- * Worker Limit
 ---------------------------------------------------------------------------------
+-
 --- | This describes what we use to limit the number of jobs, either we limit it
 --- ourselves to a specific number or we have an external parallelism semaphore
 --- limit it for us.
 -data WorkerLimit
 -  = NumProcessorsLimit Int
 -  | JSemLimit
 -    SemaphoreIdentifier
 -      -- ^ Semaphore identifier from @-jsem@
 -  deriving Eq
+-
 -mkWorkerLimit :: DynFlags -> IO WorkerLimit
 -mkWorkerLimit dflags =
 -  case parMakeCount dflags of
 -    Nothing -> pure $ num_procs 1
 -    Just (ParMakeSemaphore h) -> pure (JSemLimit h)
 -    Just ParMakeNumProcessors -> num_procs <$> getNumProcessors
 -    Just (ParMakeThisMany n) -> pure $ num_procs n
 -  where
 -    num_procs x = NumProcessorsLimit (max 1 x)
+-
 -isWorkerLimitSequential :: WorkerLimit -> Bool
 -isWorkerLimitSequential (NumProcessorsLimit x) = x <= 1
 -isWorkerLimitSequential (JSemLimit {})         = False
+-
 -runWorkerLimit :: Logger -> DynFlags -> WorkerLimit -> (AbstractSem -> IO a) -> IO a
 -#if defined(wasm32_HOST_ARCH) || defined(javascript_HOST_ARCH)
 -runWorkerLimit _logger _dflags _ action = do
 -  lock <- newMVar ()
 -  action $ AbstractSem (takeMVar lock) (putMVar lock ())
 -#else
 -runWorkerLimit logger dflags worker_limit action = case worker_limit of
 -    NumProcessorsLimit n_jobs ->
 -      runNjobsAbstractSem n_jobs action
 -    JSemLimit sem_ident -> do
 -      result <- MC.try @_ @SemaphoreError $ runJSemAbstractSem sem_ident action
 -      case result of
 -        Right a -> return a
 -        Left err -> do
 -          let diag = DriverSemaphoreOpenFailure (checkBuildingCabalPackage dflags) err
 -              msg  = singleMessage $ mkPlainMsgEnvelope (initDiagOpts dflags) noSrcSpan diag
 -          printOrThrowDiagnostics logger (initPrintConfig dflags) (initDiagOpts dflags) (GhcDriverMessage <$> msg)
 -          runNjobsAbstractSem 1 action
 -#endif
+-
 -#if !(defined(wasm32_HOST_ARCH) || defined(javascript_HOST_ARCH))
 -runNjobsAbstractSem :: Int -> (AbstractSem -> IO a) -> IO a
 -runNjobsAbstractSem n_jobs action = do
 -  compile_sem <- newQSem n_jobs
 -  n_capabilities <- getNumCapabilities
 -  n_cpus <- getNumProcessors
 -  let
 -    asem = AbstractSem (waitQSem compile_sem) (signalQSem compile_sem)
 -    set_num_caps n = unless (n_capabilities /= 1) $ setNumCapabilities n
 -    updNumCapabilities =  do
 -      -- Setting number of capabilities more than
 -      -- CPU count usually leads to high userspace
 -      -- lock contention. #9221
 -      set_num_caps $ min n_jobs n_cpus
 -    resetNumCapabilities = set_num_caps n_capabilities
 -  MC.bracket_ updNumCapabilities resetNumCapabilities $ action asem
+-
 -#endif
 +-- | Run the given actions (assumed to be in dependency order) and wait for
 +-- them all to finish, rethrowing the first unhandled exception (in action order)
 +-- afterwards.
 +runAllPipelines :: MakeEnv -> [MakeAction] -> IO ()
 +runAllPipelines env acts =
 +  void $
 +    mapConcurrentWorkers "make_worker" (me_concurrency env) (viewHscWorkerEnv (me_hsc_env env))
 +      ( \ work_env (MakeAction act res_var) -> do
 +          let lcl_env = env { me_hsc_env = setHscWorkerEnv work_env (me_hsc_env env) }
 +          mres <- runMaybeT (runReaderT act lcl_env)
 +                    `MC.onException` putMVar res_var Nothing
 +          putMVar res_var mres )
 +      acts
  --------------------------------------------------------------------------------
  -- * Utility
  --------------------------------------------------------------------------------
 -withLoggerHsc :: Int -> MakeEnv -> (HscEnv -> IO a) -> IO a
 -withLoggerHsc k MakeEnv{withLogger, hsc_env} cont = do
 -  withLogger k $ \modifyLogger -> do
 -    let lcl_logger = modifyLogger (hsc_logger hsc_env)
 -        hsc_env' = hsc_env { hsc_logger = lcl_logger }
 -    -- Run continuation with modified logger
 -    cont hsc_env'
+-
 -withParLog :: TVar LogQueueQueue -> Int -> ((Logger -> Logger) -> IO b) -> IO b
 -withParLog lqq_var k cont = do
 -  let init_log = do
 -        -- Make a new log queue
 -        lq <- newLogQueue k
 -        -- Add it into the LogQueueQueue
 -        atomically $ initLogQueue lqq_var lq
 -        return lq
 -      finish_log lq = liftIO (finishLogQueue lq)
 -  MC.bracket init_log finish_log $ \lq -> cont (pushLogHook (const (parLogAction lq)))
+-
 -withLocalTmpFS :: TmpFs -> (TmpFs -> IO a) -> IO a
 -withLocalTmpFS tmpfs act = do
 -  let initialiser = do
 -        liftIO $ forkTmpFsFrom tmpfs
 -      finaliser tmpfs_local = do
 -        liftIO $ mergeTmpFsInto tmpfs_local tmpfs
 -       -- Add remaining files which weren't cleaned up into local tmp fs for
 -       -- clean-up later.
 -       -- Clear the logQueue if this node had it's own log queue
 -  MC.bracket initialiser finaliser act
+-
 -withLocalTmpFSMake :: MakeEnv -> (MakeEnv -> IO a) -> IO a
 -withLocalTmpFSMake env k =
 -  withLocalTmpFS (hsc_tmpfs (hsc_env env)) $ \lcl_tmpfs
 -    -> k (env { hsc_env = (hsc_env env) { hsc_tmpfs = lcl_tmpfs }})
 +-- | A version of 'withWorkerLimit' taking an 'HscEnv'.
 +withWorkerLimitHsc :: HscEnv -> WorkerLimit -> (Concurrency -> HscEnv -> IO a) -> IO a
 +withWorkerLimitHsc hsc_env limit k =
 +  withWorkerLimit (hsc_logger hsc_env) (hsc_tmpfs hsc_env)
 +    (semaphoreOpenFailureHandler (hsc_logger hsc_env) (hsc_dflags hsc_env))
 +    limit
 +    (\conc work_env -> k conc (setHscWorkerEnv work_env hsc_env))
++
 +-- | Like 'withConcurrency', but retrieving the 'Concurrency' and 'HscEnv' from
 +-- the 'MakeEnv'.
 +withMakeEnvConcurrency :: MakeEnv -> (HscEnv -> IO a) -> IO a
 +withMakeEnvConcurrency env cont =
 +  withConcurrency (me_concurrency env) (cont (me_hsc_env env))
++
 +-- | The local environment for a concurrent worker derived from an 'HscEnv'.
 +viewHscWorkerEnv :: HscEnv -> ConcurrentWorkerEnv
 +viewHscWorkerEnv hsc_env =
 +  ConcurrentWorkerEnv { cwe_logger = hsc_logger hsc_env, cwe_tmpfs = hsc_tmpfs hsc_env }
++
 +-- | Set the local concurrent worker environment within an 'HscEnv'.
 +setHscWorkerEnv :: ConcurrentWorkerEnv -> HscEnv -> HscEnv
 +setHscWorkerEnv (ConcurrentWorkerEnv { cwe_logger = logger, cwe_tmpfs = tmpfs }) hsc_env =
 +  hsc_env { hsc_logger = logger, hsc_tmpfs = tmpfs }

compiler/GHC/Driver/MakeSem.hs

@@ -39,6 +39,7 @@ import GHC.Utils.Json
  import System.Semaphore
    ( AbstractSem(..)
    , ClientSemaphore
 +  , SemaphoreError
    , SemaphoreIdentifier
    , SemaphoreToken
    , openSemaphore
@@ -534,18 +535,24 @@ makeJobserver sem_ident = do
  -- | Implement an abstract semaphore using a semaphore 'Jobserver'
  -- which queries the system semaphore of the given name for resources.
 +--
 +-- Returns 'Left' if the system semaphore could not be opened, in which case
 +-- the operation is not run at all. A 'SemaphoreError' arising after the
 +-- semaphore was successfully opened is thrown, not returned.
  runJSemAbstractSem :: SemaphoreIdentifier   -- ^ the semaphore identifier (from @-jsem@)
                     -> (AbstractSem -> IO a) -- ^ the operation to run
                                              -- which requires a semaphore
 -                   -> IO a
 +                   -> IO (Either SemaphoreError a)
  runJSemAbstractSem sem_ident action = MC.mask \ unmask -> do
 -  (abs, cleanup) <- makeJobserver sem_ident
 -  r <- try $ unmask $ action abs
 -  case r of
 -    Left (e1 :: MC.SomeException) -> do
 -      (_ :: Either MC.SomeException ()) <- MC.try cleanup
 -      MC.throwM e1
 -    Right x -> cleanup $> x
 +  MC.try @_ @SemaphoreError (makeJobserver sem_ident) >>= \case
 +    Left open_failure -> return (Left open_failure)
 +    Right (abs, cleanup) -> do
 +      r <- try $ unmask $ action abs
 +      case r of
 +        Left (e1 :: MC.SomeException) -> do
 +          (_ :: Either MC.SomeException ()) <- MC.try cleanup
 +          MC.throwM e1
 +        Right x -> cleanup $> Right x
  {- Note [Architecture of the Job Server]
  ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~

compiler/GHC/Utils/TmpFs.hs

@@ -6,6 +6,7 @@ module GHC.Utils.TmpFs
      , initTmpFs
      , forkTmpFsFrom
      , mergeTmpFsInto
 +    , withLocalTmpFS
      , PathsToClean(..)
      , emptyPathsToClean
      , TempFileLifetime(..)
@@ -157,6 +158,16 @@ mergeTmpFsInto src dst = do
      atomicModifyIORef' (tmp_files_to_clean dst) (\s -> (mergePathsToClean src_files s, ()))
      atomicModifyIORef' (tmp_subdirs_to_clean dst) (\s -> (mergePathsToClean src_subdirs s, ()))
 +-- | Run an action with a local 'TmpFs' forked from the given 'TmpFs'.
 +--
 +-- The remaining files of the local 'TmpFs' which weren't cleaned up by the
 +-- action are merged back into the given 'TmpFs', for clean-up later.
 +withLocalTmpFS :: TmpFs -> (TmpFs -> IO a) -> IO a
 +withLocalTmpFS tmpfs act =
 +    Exception.bracket
 +      (forkTmpFsFrom tmpfs)
 +      (\tmpfs_local -> mergeTmpFsInto tmpfs_local tmpfs)
 +      act
  cleanTempDirs :: Logger -> TmpFs -> IO ()
  cleanTempDirs logger tmpfs

compiler/ghc.cabal.in

@@ -487,11 +487,13 @@ Library
          GHC.Driver.ByteCode
          GHC.Driver.CmdLine
          GHC.Driver.CodeOutput
 +        GHC.Driver.Concurrency
          GHC.Driver.Config
          GHC.Driver.Config.Cmm
          GHC.Driver.Config.Cmm.Parser
          GHC.Driver.Config.CmmToAsm
          GHC.Driver.Config.CmmToLlvm
 +        GHC.Driver.Config.Concurrency
          GHC.Driver.Config.Core.Lint
          GHC.Driver.Config.Core.Lint.Interactive
          GHC.Driver.Config.Core.Opt.Arity

utils/haddock/haddock-api/src/Haddock/Interface.hs

@@ -172,7 +172,7 @@ createIfaces verbosity modules flags instIfaceMap = do
    _ <- setSessionDynFlags dflags''
    targets <- mapM (\(filePath, _) -> guessTarget filePath Nothing Nothing) hs_srcs
    setTargets targets
 -  (_errs, modGraph) <- depanalE mkUnknownDiagnostic (Just batchMsg) [] False
 +  (_errs, modGraph) <- depanalE [] False
    -- Create (if necessary) and load .hi-files. With --no-compilation this happens later.
    when (Flag_NoCompilation `notElem` flags) $ do