Rodrigo Mesquita pushed to branch wip/romes/27514 at Glasgow Haskell Compiler / GHC Commits: 8e5ba737 by Rodrigo Mesquita at 2026-08-13T17:32:18+01:00 Refactor GHC.Driver.MakeAction Pull out of 'runParPipelines' the logic for creating a 'MakeEnv' ready to be used by multiple threads, as that will be useful for downsweep as well (which doesn't fit the 'runPipelines' flow), rather than being just for upsweep. The code is moved and re-structured to match the export list, simplify the sequentiality checks previously both in 'runPipelines' and 'runAllPipelines', which were weirdly similar and confusing; into the two part step where we need these checks: (1) to construct the MakeEnv, (2) to run the MakeActions in parallel. These two steps are separate and used to be too mixed up. Some additional little simplifications or clean ups here and there. - - - - - 8128c6c9 by Rodrigo Mesquita at 2026-08-13T17:32:18+01:00 Refactor GHC.Driver.Downsweep in preparation for parallel downsweep Pure refactor to improve the code to facilitate implementing parallel downsweep in the next commit. This commit puts a MakeEnv into the DownsweepEnv, gives the fields proper names and uses RecordWildcards to simplify, rather than passing around all diagnostic wrappers, driver-message-things and using 10s of positional fields. No behavior changes here! - - - - - 17298acd by Rodrigo Mesquita at 2026-08-13T17:52:36+01:00 Parallelize downsweep traversal Parallelize the downsweep pass s.t. processing and discovering the module graph can be done in parallel (parallelizing work like pre-processing CPP in modules) according to the -j<N> flag used. Using Cabal as an example with -j8, parallel downsweep was 2x faster (from 2s to 1s in downsweep time). The parallel downsweep is all implemented in the previous `dfsBuild` (now named `parDfsBuild`): - We launch a thread for every module we discover that needs to be expanded next, in the `coordinator` thread - 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. 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, which is unblocked and re-throws the exception signaled in `exc_var`, mimicking the previous behavior of `dfsBuild`. See also Note [Parallel Downsweep] Fixes #27514 - - - - - 3 changed files: - + changelog.d/parallel-downsweep - compiler/GHC/Driver/Downsweep.hs - compiler/GHC/Driver/MakeAction.hs Changes: ===================================== changelog.d/parallel-downsweep ===================================== @@ -0,0 +1,11 @@ +section: compiler +synopsis: Parallelize the downsweep/module-discovery pass +issues: #27514 +mrs: !16394 +description: { + Parallelize the downsweep pass s.t. processing and discovering the module + graph can be done in parallel (parallelizing work like pre-processing CPP + in modules) according to the -j<N> flag used. Using Cabal as an example + with -j8, parallel downsweep was 2x faster (from 2s to 1s in downsweep time). +} + ===================================== compiler/GHC/Driver/Downsweep.hs ===================================== @@ -62,7 +62,7 @@ import GHC.Data.OsPath ( OsPath, unsafeEncodeUtf ) import GHC.Data.StringBuffer import GHC.Data.Graph.Directed.Reachability -import GHC.Utils.Exception ( throwIO, SomeAsyncException ) +import GHC.Utils.Exception ( throwIO, SomeAsyncException, AsyncException (..) ) import GHC.Utils.Outputable import GHC.Utils.Panic import GHC.Utils.Misc @@ -112,8 +112,11 @@ import Control.Monad.Trans.Reader import qualified Data.Map.Strict as M import Control.Monad.Trans.Class import System.IO.Unsafe (unsafeInterleaveIO) -import Data.IORef 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] @@ -176,8 +179,9 @@ incrementally constructing a ModuleGraph using the GHC API; See #27054). So `downsweep` takes a `Maybe ModuleGraph` as one of its arguments. Downsweep iteratively *expands* each so-called 'DownsweepNode' into a list of -its dependencies, and recursively traverses all reachable nodes in a -depth-first order using 'dfsBuild'. A 'DownsweepNode' is *expanded* by 'dsNodeExpand': +its dependencies, and recursively traverses all reachable nodes in a parallel +non-det-depth-first order using 'parDfsBuild'. A 'DownsweepNode' is *expanded* +by 'dsNodeExpand': dsNodeExpand :: DownsweepNode -> DownsweepM (NodeRes (ModuleGraphNode, [DownsweepNode])) @@ -256,38 +260,48 @@ downsweep :: HscEnv -- which case there can be repeats downsweep hsc_env diag_wrapper msg old_summaries maybe_base_graph excl_mods allow_dup_roots = do n_jobs <- mkWorkerLimit (hsc_dflags hsc_env) - summ_cache <- newIORef (mkModSummaryCache (zip old_summaries (repeat SummOld))) - imps_cache <- newIORef Map.empty - (root_errs, root_summaries) <- rootSummariesParallel n_jobs hsc_env diag_wrapper msg - (getRootSummary excl_mods summ_cache imps_cache) - let closure_errs = checkHomeUnitsClosed unit_env - unit_env = hsc_unit_env hsc_env - - all_errs = closure_errs ++ root_errs - - case all_errs of - [] -> do - (downsweep_errs, downsweep_nodes) <- - downsweepFromRootNodes hsc_env summ_cache imps_cache maybe_base_graph - excl_mods allow_dup_roots DownsweepUseCompile (map ModuleNodeCompile root_summaries) [] - - let (other_errs, unit_nodes) = partitionEithers $ - HUG.unitEnv_foldWithKey (\nodes uid hue -> nodes ++ unitModuleNodes downsweep_nodes uid hue) [] - (hsc_HUG hsc_env) - - let all_nodes = downsweep_nodes ++ unit_nodes - let all_errs = downsweep_errs ++ other_errs - - let logger = hsc_logger hsc_env - tmpfs = hsc_tmpfs hsc_env - -- if we have been passed -fno-code, we enable code generation - -- for dependencies of modules that have -XTemplateHaskell, - -- otherwise those modules will fail to compile. - -- See Note [-fno-code mode] #8025 - th_configured_nodes <- enableCodeGenForTH logger tmpfs unit_env all_nodes - - return (all_errs, th_configured_nodes) - _ -> return (all_errs, emptyMG) + 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) + let closure_errs = checkHomeUnitsClosed unit_env + unit_env = hsc_unit_env hsc_env + + all_errs = closure_errs ++ root_errs + + case all_errs of + [] -> do + let env = DownsweepEnv + { 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 + } + (downsweep_errs, downsweep_nodes) <- runDownsweepM env $ + downsweepFromRootNodes maybe_base_graph allow_dup_roots + (map ModuleNodeCompile root_summaries) [] + + let (other_errs, unit_nodes) = partitionEithers $ + HUG.unitEnv_foldWithKey (\nodes uid hue -> nodes ++ unitModuleNodes downsweep_nodes uid hue) [] + (hsc_HUG hsc_env) + + let all_nodes = downsweep_nodes ++ unit_nodes + let all_errs = downsweep_errs ++ other_errs + + let logger = hsc_logger hsc_env + tmpfs = hsc_tmpfs hsc_env + -- if we have been passed -fno-code, we enable code generation + -- for dependencies of modules that have -XTemplateHaskell, + -- otherwise those modules will fail to compile. + -- See Note [-fno-code mode] #8025 + th_configured_nodes <- enableCodeGenForTH logger tmpfs unit_env all_nodes + + return (all_errs, th_configured_nodes) + _ -> return (all_errs, emptyMG) where -- Dependencies arising on a unit (backpack and module linking deps) unitModuleNodes :: [ModuleGraphNode] -> UnitId -> HomeUnitEnv -> [Either (Messages DriverMessage) ModuleGraphNode] @@ -330,15 +344,28 @@ downsweep hsc_env diag_wrapper msg old_summaries maybe_base_graph excl_mods allo downsweepThunk :: HscEnv -> ModSummary -> IO ModuleGraph downsweepThunk hsc_env mod_summary = unsafeInterleaveIO $ do debugTraceMsg (hsc_logger hsc_env) 3 $ text "Computing Module Graph thunk..." - summs <- newIORef (mkModSummaryCache [(mod_summary,SummOld)]) - imps <- newIORef mempty - ~(errs, mg) <- downsweepFromRootNodes hsc_env summs imps Nothing [] True DownsweepUseFixed [ModuleNodeCompile mod_summary] [] - let dflags = hsc_dflags hsc_env - liftIO $ printOrThrowDiagnostics (hsc_logger hsc_env) - (initPrintConfig dflags) - (initDiagOpts dflags) - (GhcDriverMessage <$> unionManyMessages errs) - return (mkModuleGraph mg) + njobs <- mkWorkerLimit (hsc_dflags hsc_env) + summs <- newMVar (mkModSummaryCache [(mod_summary,SummOld)]) + imps <- newMVar mempty + withMakeEnv njobs hsc_env mkUnknownDiagnostic Nothing $ \make_env -> do + let env = DownsweepEnv + { 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 + } + ~(errs, mg) <- runDownsweepM env $ + downsweepFromRootNodes Nothing True + [ModuleNodeCompile mod_summary] [] + let dflags = hsc_dflags hsc_env + liftIO $ printOrThrowDiagnostics (hsc_logger hsc_env) + (initPrintConfig dflags) + (initDiagOpts dflags) + (GhcDriverMessage <$> unionManyMessages errs) + return (mkModuleGraph mg) -- | Construct a module graph starting from the interactive context. -- Produces, a thunk, which when forced will perform the downsweep. @@ -362,13 +389,23 @@ downsweepInteractiveImports hsc_env ic = unsafeInterleaveIO $ do -- :load. Any home package modules need to already be in here. let cached_nodes = Map.fromList [ (mkNodeKey n, NSuccess n) | n <- mg_mss (hsc_mod_graph hsc_env) ] - summ_cache <- newIORef mempty - imps_cache <- newIORef mempty - let env = DownsweepEnv hsc_env DownsweepUseFixed{-or DownsweepUseCompile?-} summ_cache imps_cache [] - graph <- runDownsweepM env do - loopFromInteractive cached_nodes interactive_mn imps - let all_nodes = [s | NSuccess s <- M.elems graph ] - return $ mkModuleGraph all_nodes + 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 + let env = DownsweepEnv + { 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 + } + graph <- runDownsweepM env do + loopFromInteractive cached_nodes interactive_mn imps + let all_nodes = [s | NSuccess s <- M.elems graph ] + return $ mkModuleGraph all_nodes -- | Create a module graph from a list of installed modules. -- This is used by the loader when we need to load modules but there @@ -396,26 +433,38 @@ downsweepInstalledModules hsc_env mods = do -- already know that we can find the modules we need to load. _ -> throwGhcException $ ProgramError $ showSDoc (hsc_dflags hsc_env) $ text "downsweepInstalledModules: Could not find installed module" <+> ppr i + njobs <- mkWorkerLimit (hsc_dflags hsc_env) nodes <- mapM process installed_mods - summs <- newIORef mempty - imps <- newIORef mempty - (errs, mg) <- downsweepFromRootNodes hsc_env summs imps Nothing [] True DownsweepUseFixed nodes external_uids + summs <- newMVar mempty + imps <- newMVar mempty + withMakeEnv njobs hsc_env mkUnknownDiagnostic Nothing $ \make_env -> do + let env = DownsweepEnv + { 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 + } + (errs, mg) <- runDownsweepM env $ + downsweepFromRootNodes Nothing True nodes external_uids - -- Similarly here, we should really not get any errors, but print them out if we do. - let dflags = hsc_dflags hsc_env - liftIO $ printOrThrowDiagnostics (hsc_logger hsc_env) - (initPrintConfig dflags) - (initDiagOpts dflags) - (GhcDriverMessage <$> unionManyMessages errs) + -- Similarly here, we should really not get any errors, but print them out if we do. + let dflags = hsc_dflags hsc_env + liftIO $ printOrThrowDiagnostics (hsc_logger hsc_env) + (initPrintConfig dflags) + (initDiagOpts dflags) + (GhcDriverMessage <$> unionManyMessages errs) - return (mkModuleGraph mg) + return (mkModuleGraph mg) ----------------------------------------------------------------------------- -- * Orchestrator: downsweepFromRootNodes ----------------------------------------------------------------------------- -type ModSummaryCache = IORef ModSummaryCacheMap -type ImportsCache = IORef ImportsCacheMap +type ModSummaryCache = MVar ModSummaryCacheMap +type ImportsCache = MVar ImportsCacheMap -- | A cache from file paths to the already summarised modules. The same file -- can be used in multiple units so the map is actually also keyed by which @@ -450,30 +499,26 @@ data DownsweepMode = DownsweepUseCompile | DownsweepUseFixed -- 'UnitId's. -- This function will start at the given roots, and traverse downwards to find -- all the dependencies, all the way to the leaf units. -downsweepFromRootNodes :: HscEnv - -> ModSummaryCache - -> ImportsCache - -> Maybe ModuleGraph - -> [ModuleName] - -> Bool - -> DownsweepMode -- ^ Whether to create fixed or compile nodes for dependencies - -> [ModuleNodeInfo] -- ^ The starting ModuleNodeInfo - -> [UnitId] -- ^ The starting units - -> IO ([DriverMessages], [ModuleGraphNode]) -downsweepFromRootNodes hsc_env summ_cache imps_cache maybe_base_graph excl_mods allow_dup_roots mode root_nodes root_uids = do +downsweepFromRootNodes + :: Maybe ModuleGraph + -> Bool + -> [ModuleNodeInfo] -- ^ The starting ModuleNodeInfo + -> [UnitId] -- ^ The starting units + -> DownsweepM ([DriverMessages], [ModuleGraphNode]) +downsweepFromRootNodes maybe_base_graph allow_dup_roots root_nodes root_uids = + ReaderT $ \env@DownsweepEnv{..} -> do when (not allow_dup_roots) $ case root_duplicates of [] -> return () - (dup_root:_) -> multiRootsErr sec dup_root - modifyImpsCache imps_cache (`M.union` mkRootMap root_nodes) -- add root nodes to imports cache - let env = DownsweepEnv hsc_env mode summ_cache imps_cache excl_mods - deps' <- runDownsweepM env $ do + (dup_root:_) -> multiRootsErr (sec ds_hsc_env) dup_root + modifyImpsCache ds_imports_cache (`M.union` mkRootMap root_nodes) -- add root nodes to imports cache + deps' <- runDownsweepM env $ do let base_nodes = maybe M.empty moduleGraphNodeMap maybe_base_graph module_deps <- loopModuleNodeInfos base_nodes root_nodes - all_deps <- loopUnits module_deps (hscActiveUnitId hsc_env) root_uids - deps' <- loopInstantiations all_deps (getHomeUnitInstantiations hsc_env) + all_deps <- loopUnits module_deps (hscActiveUnitId ds_hsc_env) root_uids + deps' <- loopInstantiations all_deps (getHomeUnitInstantiations ds_hsc_env) return deps' - f_cache <- readIORef summ_cache + f_cache <- readMVar ds_summaries_cache let downsweep_errs = lefts (M.elems f_cache) downsweep_nodes = [ s | NSuccess s <- M.elems deps' ] @@ -501,7 +546,7 @@ downsweepFromRootNodes hsc_env summ_cache imps_cache maybe_base_graph excl_mods moduleGraphNodeMap graph = M.fromList [(mkNodeKey node, NSuccess node) | node <- mgModSummaries' graph] - sec = initSourceErrorContext (hsc_dflags hsc_env) + sec hsc_env = initSourceErrorContext (hsc_dflags hsc_env) -------------------------------------------------------------------------------- -- ** 'DownsweepM' @@ -509,11 +554,14 @@ downsweepFromRootNodes hsc_env summ_cache imps_cache maybe_base_graph excl_mods type DownsweepM a = ReaderT DownsweepEnv IO a data DownsweepEnv = DownsweepEnv { - downsweep_hsc_env :: HscEnv - , _downsweep_mode :: DownsweepMode - , _downsweep_summaries_cache :: ModSummaryCache - , downsweep_imports_cache :: ImportsCache - , _downsweep_excl_mods :: [ModuleName] + ds_hsc_env :: HscEnv + , ds_mode :: DownsweepMode + -- ^ Whether to create fixed or compile nodes for dependencies + , ds_summaries_cache :: ModSummaryCache + , ds_imports_cache :: ImportsCache + , ds_excl_mods :: [ModuleName] + , ds_n_jobs :: WorkerLimit + , ds_make_env :: MakeEnv } mkModSummaryCache :: [(ModSummary, SummProvenance)] -> ModSummaryCacheMap @@ -529,8 +577,8 @@ addModSummaryCache ms pr fe = upd_fe fe modifySummCache :: ModSummaryCache -> (ModSummaryCacheMap -> ModSummaryCacheMap) -> IO () modifyImpsCache :: ImportsCache -> (ImportsCacheMap -> ImportsCacheMap) -> IO () -modifySummCache r f = atomicModifyIORef' r (\c -> (f c, ())) -modifyImpsCache r f = atomicModifyIORef' r (\c -> (f c, ())) +modifySummCache v f = modifyMVar v (\c -> let !r = f c in pure (r, ())) +modifyImpsCache v f = modifyMVar v (\c -> let !r = f c in pure (r, ())) -- | A cache from a module import (in given home unit context, with a package -- qualifier, and the imported module name (with or without SOURCE)) to the @@ -553,7 +601,7 @@ loopModuleNodeInfos :: M.Map NodeKey (NodeRes ModuleGraphNode) -> [ModuleNodeInf loopUnits :: M.Map NodeKey (NodeRes ModuleGraphNode) -> UnitId -> [UnitId] -> DownsweepM (M.Map NodeKey (NodeRes ModuleGraphNode)) loopInstantiations :: M.Map NodeKey (NodeRes ModuleGraphNode) -> [(UnitId, InstantiatedUnit)] -> DownsweepM (M.Map NodeKey (NodeRes ModuleGraphNode)) loopFromInteractive :: M.Map NodeKey (NodeRes ModuleGraphNode) -> Module -> [InteractiveImport] -> DownsweepM (M.Map NodeKey (NodeRes ModuleGraphNode)) -loopDownsweepNodes base_map nodes = dfsBuild (Just base_map) nodes dsNodeInfoKey dsNodeExpand +loopDownsweepNodes base_map nodes = parDfsBuild (Just base_map) nodes dsNodeInfoKey dsNodeExpand loopModuleNodeInfos base_map = loopDownsweepNodes base_map . map DSMod loopUnits base_map homud = loopDownsweepNodes base_map . map (DSUnit homud) loopInstantiations base_map = loopDownsweepNodes base_map . map (uncurry DSInst) @@ -617,7 +665,7 @@ dsNodeExpand = \case expandModuleSummary :: ModSummary -> DownsweepM (NodeRes (ModuleGraphNode, [DownsweepNode])) expandModuleSummary ms = do -- Didn't work out what the imports mean yet, now do that. - hsc_env <- asks downsweep_hsc_env + hsc_env <- asks ds_hsc_env let home_uid = ms_unitid ms home_unit = ue_unitHomeUnit home_uid (hsc_unit_env hsc_env) (final_deps, todo) <- unzip <$> mapM (expandModImport home_uid home_unit) (calcDeps ms) @@ -652,7 +700,7 @@ expandModuleSummary ms = do -- Didn't work out what the imports mean yet, now do FoundHomeWithError (_uid, _e) -> return ( Nothing, [] ) -- the error @e@ is already stored in the summarisation cache, - -- (the IORef in DownsweepM) and will get reported at the end. + -- (the MVar in DownsweepM) and will get reported at the end. FoundHome s -> return -- MP: This assumes that we can only instantiate non home units, which is probably fair enough for now. ( Just $ mkModuleEdge lvl (NodeKey_Module (mnKey s)) @@ -673,7 +721,7 @@ expandModuleSummary ms = do -- Didn't work out what the imports mean yet, now do -- NB: If you ever reach a Fixed node, everything under that also must be fixed. expandFixedModuleNode :: ModNodeKeyWithUid -> ModLocation -> DownsweepM (NodeRes (ModuleGraphNode, [DownsweepNode])) expandFixedModuleNode key loc = do - hsc_env <- asks downsweep_hsc_env + hsc_env <- asks ds_hsc_env -- MP: TODO, we should just read the dependency info from the interface rather than either -- a. Loading the whole thing into the EPS (this might never nececssary and causes lots of things to be permanently loaded into memory) -- b. Loading the whole interface into a buffer before discarding it. (wasted allocation and deserialisation) @@ -732,7 +780,7 @@ expandUnitNode :: UnitId {-^ @node_uid@ -} -> UnitId {-^ Home unit from where @n expandUnitNode node_uid home_context_uid = do -- Set active unit so that looking loopUnit finds the correct -- -package flags in the unit state. - hsc_env <- asks downsweep_hsc_env + hsc_env <- asks ds_hsc_env let lcl_hsc_env = hscSetActiveUnitId home_context_uid hsc_env case unitDepends <$> lookupUnitId (hsc_units lcl_hsc_env) node_uid of Just us -> pure $ NSuccess ((UnitNode us node_uid), map (\u -> DSUnit{node_uid=u, home_context_uid{-inherit-}}) us) @@ -745,8 +793,8 @@ expandInstantiatedUnit iud home_uid = pure $ NSuccess expandInteractiveImports :: Module -> [InteractiveImport] -> DownsweepM (NodeRes (ModuleGraphNode, [DownsweepNode])) expandInteractiveImports imod imps = do - hsc_env <- asks downsweep_hsc_env - imps_cache <- asks downsweep_imports_cache + hsc_env <- asks ds_hsc_env + imps_cache <- asks ds_imports_cache let -- A simple edge to a module from the same home unit @@ -807,13 +855,13 @@ downsweepSummarise :: HomeUnit -> Maybe (StringBuffer, UTCTime) -> DownsweepM SummariseResult downsweepSummarise home_unit imp maybe_buf = do - DownsweepEnv hsc_env mode summaries_cache_ref imports_cache_ref excl_mods <- ask - liftIO $ case mode of + DownsweepEnv{..} <- ask + liftIO $ case ds_mode of DownsweepUseCompile -> - summariseModule hsc_env home_unit summaries_cache_ref imports_cache_ref - imp maybe_buf excl_mods + summariseModule ds_hsc_env home_unit ds_summaries_cache ds_imports_cache + imp maybe_buf ds_excl_mods DownsweepUseFixed -> - summariseModuleInterface hsc_env home_unit imports_cache_ref imp excl_mods + summariseModuleInterface ds_hsc_env home_unit ds_imports_cache imp ds_excl_mods multiRootsErr :: SourceErrorContext -> NE.NonEmpty ModuleNodeInfo -> IO () multiRootsErr sec (summ1 NE.:| summs) @@ -878,56 +926,15 @@ 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. --- Create bundles of 'Target's 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. -rootSummariesParallel :: - WorkerLimit -> - HscEnv -> - (GhcMessage -> AnyGhcDiagnostic) -> - Maybe Messager -> - (HscEnv -> Target -> IO (Either DriverMessages ModSummary)) -> - IO ([DriverMessages], [ModSummary]) -rootSummariesParallel n_jobs hsc_env diag_wrapper msg get_summary = do - (actions, get_results) <- unzip <$> mapM action_and_result (zip [1..] bundles) - runPipelines n_jobs hsc_env diag_wrapper msg actions - (sequence . catMaybes <$> sequence get_results) >>= \case - Right results -> pure (partitionEithers (concat results)) - Left exc -> throwIO exc - where - bundles = mk_bundles targets - - mk_bundles = unfoldr \case - [] -> Nothing - ts -> Just (splitAt bundle_size ts) - - bundle_size = 20 - - targets = hsc_targets hsc_env - - 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 (get_summary lcl_hsc_env) target_bundle) >>= \case - Left e | Just (_ :: SomeAsyncException) <- fromException e -> - throwIO e - a -> pure a +-- | Execute 'getRootSummary' for the 'Target's using the parallelism pipeline system. +rootSummariesParallel + :: WorkerLimit -> MakeEnv -> [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 -------------------------------------------------------------------------------- -- * Check/validate properties and error out @@ -1325,7 +1332,7 @@ summariseFile -> IO (Either DriverMessages ModSummary) summariseFile hsc_env' home_unit summ_cache_ref src_fn mb_phase maybe_buf - = do file_summ_cache <- readIORef summ_cache_ref + = do file_summ_cache <- readMVar summ_cache_ref case M.lookup (homeUnitId home_unit, src_fn_os) file_summ_cache of Just (Right (chd_summary, SummFresh)) -> -- Fresh: use it straight away @@ -1505,7 +1512,7 @@ summariseModuleDispatch k hsc_env' imps_cache_ref home_unit imp excl_mods find_it :: IO SummariseResult find_it = do - imps_cache <- readIORef imps_cache_ref + imps_cache <- readMVar imps_cache_ref case M.lookup cache_key imps_cache of Just result -> return result Nothing -> do @@ -1547,7 +1554,7 @@ summariseModuleWithSource home_unit summ_cache_ref is_boot maybe_buf hsc_env loc -- Adjust location to point to the hs-boot source file, -- hi file, object file, when is_boot says so let src_fn = expectJust (ml_hs_file location) - summ_cache <- readIORef summ_cache_ref + summ_cache <- readMVar summ_cache_ref -- Reject the cache result if the module name doesn't match the inferred -- module name based on the file name. @@ -1722,10 +1729,10 @@ getPreprocessedImports hsc_env src_fn mb_phase maybe_buf = do return PreprocessedImports {..} -------------------------------------------------------------------------------- --- * Generic traversal of iteratively-built graph: dfsBuild +-- * Generic traversal of iteratively-built graph: parDfsBuild -------------------------------------------------------------------------------- --- | The result of expanding a node in 'dfsBuild'. +-- | The result of expanding a node in 'parDfsBuild'. data NodeRes v -- | Computed the node payload successfully = NSuccess v @@ -1738,7 +1745,7 @@ data NodeRes v -- abort. | NSkip --- | In a depth-first order, and starting from the given roots, traverse a +-- | In a parallel non-det-depth-first order, and starting from the given roots, traverse a -- graph by iteratively expanding a node into a payload and a list of children -- nodes to visit next. -- @@ -1751,18 +1758,17 @@ data NodeRes v -- The result is a mapping from the key of every node transitively reachable -- from the root nodes (inclusively) to the payload returned by expanding that -- node. The result includes the previously visited nodes given in @base_map@, --- s.t. @dfsBuild base_map [] _ _ == base_map@. +-- s.t. @parDfsBuild base_map [] _ _ == base_map@. -- -- The @expand@ function returns an 'NResult'. See the 'NResult' documentation -- for more information about each result type. -- --- Error handling and exiting early can be achieved by selecting a @Monad m@ --- accordingly, such as @Control.Monad.Except.Except@ --- -- Example usage: @n@ is instanced to @DownsweepNode@, @k@ is @NodeKey@, and @v@ is @ModuleNodeEdge@. -- --- See also Note [Downsweep Control Flow and Caching] -dfsBuild :: (Ord k, Monad m) +-- See Note [Parallel Downsweep] for more information about how parallelism is +-- achieved, and See Note [Downsweep Control Flow and Caching] for information +-- about the various caches used. +parDfsBuild :: Ord k => Maybe (Map.Map k (NodeRes v)) -- ^ Base map, existing results. We won't re-expand any of the nodes -- already present in this map. @@ -1770,34 +1776,91 @@ dfsBuild :: (Ord k, Monad m) -- ^ The root nodes from where to start traversal -> (n -> k) -- ^ Compute the key which uniquely identifies this node - -> (n -> m (NodeRes (v,[n]))) + -> (n -> DownsweepM (NodeRes (v,[n]))) -- ^ Expand this node into its payload result and into the list of -- children nodes to visit next. - -> m (Map.Map k (NodeRes v)) + -> DownsweepM (Map.Map k (NodeRes v)) -- ^ The result accumulates the payload of expanding the root nodes -- and all nodes transitively reachable from those roots. -dfsBuild base_map roots key expand = go roots (fromMaybe Map.empty base_map) +parDfsBuild base_map roots key expand = ReaderT $ \ds_env -> do + exc_var <- newTVarIO Nothing -- signal this var when there's an exception + visited_var <- newTVarIO (fromMaybe Map.empty base_map) + pending <- newTVarIO Set.empty + worklist <- newTQueueIO + + coord_tid <- forkIO $ + coordinator ds_env exc_var visited_var worklist pending + `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)) + + mapM_ (atomically . writeTQueue worklist) roots + + -- this txn retries until all work is done or there is an exception + mb_exc <- atomically $ do + readTVar exc_var >>= \case + Just e -> return (Just e) + Nothing -> do + empty_worklist <- isEmptyTQueue worklist + empty_pending <- Set.null <$> readTVar pending + unless (empty_worklist && empty_pending) retry + return Nothing + + killThread coord_tid + case mb_exc of + Just e -> throwIO e + Nothing -> readTVarIO visited_var + where - go [] visited = pure visited - go (s:ss) visited - | k `Map.member` visited - = go ss visited - | otherwise - = do r <- expand s - case r of - NSkip -> - go ss - (Map.insert k NSkip visited) -- Skip! - NSuccess (v,ns) -> - go (ns ++ ss) - (Map.insert k (NSuccess v) visited) - where - k = key s + coordinator ds_env exc_var visvar worklist pendvar = 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) -> void $ do + withLocalTmpFSMake (ds_make_env ds_env) $ \make_env -> + forkIOWithUnmask $ \unmask -> + unmask (worker ds_env{ds_make_env = make_env} visvar worklist pendvar k node) + -- write exception in the worker to the main thread + `MC.catch` \(e::MC.SomeException) -> + atomically (modifyTVar' exc_var (<|> Just e)) + + 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! + + case r of + NSkip -> + atomically $ modifyTVar' visvar (Map.insert k NSkip) + NSuccess (v,ns) -> do + atomically $ modifyTVar' visvar (Map.insert k (NSuccess v)) + mapM_ (atomically . writeTQueue worklist) ns + + atomically $ modifyTVar' pendvar (Set.delete k) {- Note [Downsweep Control Flow and Caching] ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ -The control flow of downsweep is extracted into a single function `dfsBuild`, +The control flow of downsweep is extracted into a single function `parDfsBuild`, which takes care of iteratively expanding and traversing all nodes of the in-construction module graph necessary to build a full `ModuleGraph` at the end. @@ -1806,7 +1869,7 @@ There are three levels of caching going on, all of which are necessary to make sure we don't do repeated work (notably, we NEVER summarise the same module twice). -1. `dfsBuild` accumulates the final module graph and never revisits the +1. `parDfsBuild` accumulates the final module graph and never revisits the same node of the module graph. Cache is keyed by the final `ModuleGraph`s `NodeKey`s. @@ -1874,6 +1937,83 @@ twice). See tests T27461a and T27461b. -See also Note [Downsweep: building and maintaining the module graph] and -Note [The ModuleGraph]. +See also Note [Downsweep: building and maintaining the module graph] and Note [The ModuleGraph]. + + +Note [Parallel Downsweep] +~~~~~~~~~~~~~~~~~~~~~~~~~ +Downsweep traverses the modules iteratively to discover the module graph +structure (see Note [Downsweep: building and maintaining the module graph]) + +Each module has to be expanded/processed to discover dependencies amongst other +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 ===================================== compiler/GHC/Driver/MakeAction.hs ===================================== @@ -1,12 +1,11 @@ {-# LANGUAGE CPP #-} module GHC.Driver.MakeAction ( MakeAction(..) - , MakeEnv(..) , RunMakeM + , MakeEnv(..) + , withMakeEnv -- * Running the pipelines , runAllPipelines - , runParPipelines - , runSeqPipelines , runPipelines -- * Worker limit , WorkerLimit(..) @@ -66,31 +65,9 @@ import GHC.Driver.Pipeline.LogQueue import Control.Concurrent.STM import Control.Monad.Trans.Maybe --- Executing the pipelines - -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 - --- | 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 +-------------------------------------------------------------------------------- +-- * 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 @@ -104,49 +81,147 @@ data MakeEnv = MakeEnv { hsc_env :: !HscEnv -- The basic HscEnv which will be au , 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. +-- For -jn, it can be used from multiple threads (e.g. in runAllPipelines when -jN) +withMakeEnv + :: WorkerLimit -- ^ How to limit work parallelism + -> HscEnv -- ^ The basic HscEnv which is augmented with specific info for each module + -> (GhcMessage -> AnyGhcDiagnostic) + -> 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 + +-- ** MakeAction --------------------------------------------------------------- -label_self :: String -> IO () -label_self thread_name = do - self_tid <- CC.myThreadId - CC.labelThread self_tid thread_name +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 -runPipelines :: WorkerLimit -> HscEnv -> (GhcMessage -> AnyGhcDiagnostic) -> Maybe Messager -> [MakeAction] -> IO () --- Don't even initialise plugins if there are no pipelines +-------------------------------------------------------------------------------- +-- * Running the pipelines +-------------------------------------------------------------------------------- + +-- | Build and run a pipeline using the given worker limit for parallelism +runPipelines + :: WorkerLimit -> HscEnv + -> (GhcMessage -> AnyGhcDiagnostic) -> Maybe Messager + -> [MakeAction] -- ^ The build plan for all the module nodes + -> IO () runPipelines n_job hsc_env diag_wrapper mHscMessager all_pipelines = do liftIO $ label_self "main --make thread" - case n_job of - NumProcessorsLimit n | n <= 1 -> runSeqPipelines hsc_env diag_wrapper mHscMessager all_pipelines - _n -> runParPipelines n_job hsc_env diag_wrapper mHscMessager all_pipelines - -runSeqPipelines :: HscEnv -> (GhcMessage -> AnyGhcDiagnostic) -> Maybe Messager -> [MakeAction] -> IO () -runSeqPipelines plugin_hsc_env diag_wrapper mHscMessager all_pipelines = - let env = MakeEnv { hsc_env = plugin_hsc_env - , withLogger = \_ k -> k id - , compile_sem = AbstractSem (return ()) (return ()) - , env_messager = mHscMessager - , diag_wrapper = diag_wrapper - } - in runAllPipelines (NumProcessorsLimit 1) env all_pipelines + withMakeEnv n_job hsc_env diag_wrapper mHscMessager $ \make_env -> do + runAllPipelines n_job make_env all_pipelines + where + label_self :: String -> IO () + label_self thread_name = do + self_tid <- CC.myThreadId + CC.labelThread self_tid thread_name -#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 +-- | 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 -#endif + 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 -> + 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) @@ -168,41 +243,28 @@ runWorkerLimit logger dflags worker_limit action = case worker_limit of runNjobsAbstractSem 1 action #endif --- | Build and run a pipeline -runParPipelines :: WorkerLimit -- ^ How to limit work parallelism - -> HscEnv -- ^ The basic HscEnv which is augmented with specific info for each module - -> (GhcMessage -> AnyGhcDiagnostic) - -> Maybe Messager -- ^ Optional custom messager to use to report progress - -> [MakeAction] -- ^ The build plan for all the module nodes - -> IO () -runParPipelines worker_limit plugin_hsc_env diag_wrapper mHscMessager all_pipelines = 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 plugin_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 plugin_hsc_env) - let thread_safe_hsc_env = plugin_hsc_env { hsc_logger = thread_safe_logger } - - runWorkerLimit (hsc_logger plugin_hsc_env) (hsc_dflags plugin_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. - runAllPipelines worker_limit env all_pipelines - atomically $ writeTVar stopped_var True - wait_log_thread +#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 + +-------------------------------------------------------------------------------- +-- * Utility +-------------------------------------------------------------------------------- withLoggerHsc :: Int -> MakeEnv -> (HscEnv -> IO a) -> IO a withLoggerHsc k MakeEnv{withLogger, hsc_env} cont = do @@ -238,44 +300,3 @@ 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 }}) - - --- | 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 -> - 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) - -type RunMakeM a = ReaderT MakeEnv (MaybeT IO) a - -data MakeAction = forall a . MakeAction !(RunMakeM a) !(MVar (Maybe a)) - -waitMakeAction :: MakeAction -> IO () -waitMakeAction (MakeAction _ mvar) = () <$ readMVar mvar View it on GitLab: https://gitlab.haskell.org/ghc/ghc/-/compare/e53ad1ffb0bc4268d7cdb10dc7cb4f1... -- View it on GitLab: https://gitlab.haskell.org/ghc/ghc/-/compare/e53ad1ffb0bc4268d7cdb10dc7cb4f1... 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