[Git][ghc/ghc][wip/romes/27514] Concurrent worker abstraction for downsweep/make

14 Aug 2026

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
=====================================
@@ -0,0 +1,464 @@
+{-# 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
=====================================
@@ -0,0 +1,41 @@
+-- | 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
=====================================
@@ -1,4 +1,3 @@
-{-# 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



View it on GitLab: https://gitlab.haskell.org/ghc/ghc/-/commit/566d789d5522489eb017e9999f0b7bde...

-- 
View it on GitLab: https://gitlab.haskell.org/ghc/ghc/-/commit/566d789d5522489eb017e9999f0b7bde...
You're receiving this email because of your account on gitlab.haskell.org. Manage all notifications: https://gitlab.haskell.org/-/profile/notifications | Help: https://gitlab.haskell.org/help

    

[Git][ghc/ghc][wip/romes/27514] Concurrent worker abstraction for downsweep/make

sheaf (＠sheaf)