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

Commits:

3 changed files:

Changes:

  • changelog.d/parallel-downsweep
    1 
    +section: compiler
    2 
    +synopsis: Parallelize the downsweep/module-discovery pass
    3 
    +issues: #27514
    4 
    +mrs: !16394
    5 
    +description: {
    6 
    +    Parallelize the downsweep pass s.t. processing and discovering the module
    7 
    +    graph can be done in parallel (parallelizing work like pre-processing CPP
    8 
    +    in modules) according to the -j<N> flag used. Using Cabal as an example
    9 
    +    with -j8, parallel downsweep was 2x faster (from 2s to 1s in downsweep time).
    10 
    +}
    11 
    +

  • compiler/GHC/Driver/Downsweep.hs
    ... ... @@ -62,7 +62,7 @@ import GHC.Data.OsPath ( OsPath, unsafeEncodeUtf )
    62 62 
     import GHC.Data.StringBuffer
    63 63 
     import GHC.Data.Graph.Directed.Reachability
    64 64
    65 
    -import GHC.Utils.Exception ( throwIO, SomeAsyncException )
    65 
    +import GHC.Utils.Exception ( throwIO, SomeAsyncException, AsyncException (..) )
    66 66 
     import GHC.Utils.Outputable
    67 67 
     import GHC.Utils.Panic
    68 68 
     import GHC.Utils.Misc
    ... ... @@ -112,8 +112,11 @@ import Control.Monad.Trans.Reader
    112 112 
     import qualified Data.Map.Strict as M
    113 113 
     import Control.Monad.Trans.Class
    114 114 
     import System.IO.Unsafe (unsafeInterleaveIO)
    115 
    -import Data.IORef
    116 115 
     import qualified Data.List.NonEmpty as NE
    116 
    +import Control.Concurrent
    117 
    +import Control.Concurrent.STM.TQueue
    118 
    +import Control.Concurrent.STM
    119 
    +import Control.Applicative
    117 120
    118 121 
     {-
    119 122 
     Note [The ModuleGraph]
    ... ... @@ -176,8 +179,9 @@ incrementally constructing a ModuleGraph using the GHC API; See #27054). So
    176 179 
     `downsweep` takes a `Maybe ModuleGraph` as one of its arguments.
    177 180
    178 181 
     Downsweep iteratively *expands* each so-called 'DownsweepNode' into a list of
    179 
    -its dependencies, and recursively traverses all reachable nodes in a
    180 
    -depth-first order using 'dfsBuild'. A 'DownsweepNode' is *expanded* by 'dsNodeExpand':
    182 
    +its dependencies, and recursively traverses all reachable nodes in a parallel
    183 
    +non-det-depth-first order using 'parDfsBuild'. A 'DownsweepNode' is *expanded*
    184 
    +by 'dsNodeExpand':
    181 185
    182 186 
       dsNodeExpand :: DownsweepNode -> DownsweepM (NodeRes (ModuleGraphNode, [DownsweepNode]))
    183 187
    ... ... @@ -256,38 +260,48 @@ downsweep :: HscEnv
    256 260 
                     -- which case there can be repeats
    257 261 
     downsweep hsc_env diag_wrapper msg old_summaries maybe_base_graph excl_mods allow_dup_roots = do
    258 262 
       n_jobs     <- mkWorkerLimit (hsc_dflags hsc_env)
    259 
    -  summ_cache <- newIORef (mkModSummaryCache (zip old_summaries (repeat SummOld)))
    260 
    -  imps_cache <- newIORef Map.empty
    261 
    -  (root_errs, root_summaries) <- rootSummariesParallel n_jobs hsc_env diag_wrapper msg
    262 
    -                                   (getRootSummary excl_mods summ_cache imps_cache)
    263 
    -  let closure_errs = checkHomeUnitsClosed unit_env
    264 
    -      unit_env = hsc_unit_env hsc_env
    265 
    -
    266 
    -      all_errs = closure_errs ++ root_errs
    267 
    -
    268 
    -  case all_errs of
    269 
    -    [] -> do
    270 
    -       (downsweep_errs, downsweep_nodes) <-
    271 
    -          downsweepFromRootNodes hsc_env summ_cache imps_cache maybe_base_graph
    272 
    -            excl_mods allow_dup_roots DownsweepUseCompile (map ModuleNodeCompile root_summaries) []
    273 
    -
    274 
    -       let (other_errs, unit_nodes) = partitionEithers $
    275 
    -              HUG.unitEnv_foldWithKey (\nodes uid hue -> nodes ++ unitModuleNodes downsweep_nodes uid hue) []
    276 
    -                                      (hsc_HUG hsc_env)
    277 
    -
    278 
    -       let all_nodes = downsweep_nodes ++ unit_nodes
    279 
    -       let all_errs = downsweep_errs ++ other_errs
    280 
    -
    281 
    -       let logger = hsc_logger hsc_env
    282 
    -           tmpfs = hsc_tmpfs hsc_env
    283 
    -       -- if we have been passed -fno-code, we enable code generation
    284 
    -       -- for dependencies of modules that have -XTemplateHaskell,
    285 
    -       -- otherwise those modules will fail to compile.
    286 
    -       -- See Note [-fno-code mode] #8025
    287 
    -       th_configured_nodes <- enableCodeGenForTH logger tmpfs unit_env all_nodes
    288 
    -
    289 
    -       return (all_errs, th_configured_nodes)
    290 
    -    _  -> return (all_errs, emptyMG)
    263 
    +  summ_cache <- newMVar (mkModSummaryCache (zip old_summaries (repeat SummOld)))
    264 
    +  imps_cache <- newMVar Map.empty
    265 
    +  withMakeEnv n_jobs hsc_env diag_wrapper msg $ \make_env -> do
    266 
    +    (root_errs, root_summaries) <- rootSummariesParallel n_jobs make_env (hsc_targets hsc_env)
    267 
    +                                     (getRootSummary excl_mods summ_cache imps_cache)
    268 
    +    let closure_errs = checkHomeUnitsClosed unit_env
    269 
    +        unit_env = hsc_unit_env hsc_env
    270 
    +
    271 
    +        all_errs = closure_errs ++ root_errs
    272 
    +
    273 
    +    case all_errs of
    274 
    +      [] -> do
    275 
    +         let env = DownsweepEnv
    276 
    +               { ds_hsc_env         = hsc_env
    277 
    +               , ds_summaries_cache = summ_cache
    278 
    +               , ds_imports_cache   = imps_cache
    279 
    +               , ds_mode            = DownsweepUseCompile
    280 
    +               , ds_excl_mods       = excl_mods
    281 
    +               , ds_n_jobs          = n_jobs
    282 
    +               , ds_make_env        = make_env
    283 
    +               }
    284 
    +         (downsweep_errs, downsweep_nodes) <- runDownsweepM env $
    285 
    +            downsweepFromRootNodes maybe_base_graph allow_dup_roots
    286 
    +              (map ModuleNodeCompile root_summaries) []
    287 
    +
    288 
    +         let (other_errs, unit_nodes) = partitionEithers $
    289 
    +                HUG.unitEnv_foldWithKey (\nodes uid hue -> nodes ++ unitModuleNodes downsweep_nodes uid hue) []
    290 
    +                                        (hsc_HUG hsc_env)
    291 
    +
    292 
    +         let all_nodes = downsweep_nodes ++ unit_nodes
    293 
    +         let all_errs = downsweep_errs ++ other_errs
    294 
    +
    295 
    +         let logger = hsc_logger hsc_env
    296 
    +             tmpfs = hsc_tmpfs hsc_env
    297 
    +         -- if we have been passed -fno-code, we enable code generation
    298 
    +         -- for dependencies of modules that have -XTemplateHaskell,
    299 
    +         -- otherwise those modules will fail to compile.
    300 
    +         -- See Note [-fno-code mode] #8025
    301 
    +         th_configured_nodes <- enableCodeGenForTH logger tmpfs unit_env all_nodes
    302 
    +
    303 
    +         return (all_errs, th_configured_nodes)
    304 
    +      _  -> return (all_errs, emptyMG)
    291 305 
       where
    292 306 
         -- Dependencies arising on a unit (backpack and module linking deps)
    293 307 
         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
    330 344 
     downsweepThunk :: HscEnv -> ModSummary -> IO ModuleGraph
    331 345 
     downsweepThunk hsc_env mod_summary = unsafeInterleaveIO $ do
    332 346 
       debugTraceMsg (hsc_logger hsc_env) 3 $ text "Computing Module Graph thunk..."
    333 
    -  summs <- newIORef (mkModSummaryCache [(mod_summary,SummOld)])
    334 
    -  imps  <- newIORef mempty
    335 
    -  ~(errs, mg) <- downsweepFromRootNodes hsc_env summs imps Nothing [] True DownsweepUseFixed [ModuleNodeCompile mod_summary] []
    336 
    -  let dflags = hsc_dflags hsc_env
    337 
    -  liftIO $ printOrThrowDiagnostics (hsc_logger hsc_env)
    338 
    -                                   (initPrintConfig dflags)
    339 
    -                                   (initDiagOpts dflags)
    340 
    -                                   (GhcDriverMessage <$> unionManyMessages errs)
    341 
    -  return (mkModuleGraph mg)
    347 
    +  njobs <- mkWorkerLimit (hsc_dflags hsc_env)
    348 
    +  summs <- newMVar (mkModSummaryCache [(mod_summary,SummOld)])
    349 
    +  imps  <- newMVar mempty
    350 
    +  withMakeEnv njobs hsc_env mkUnknownDiagnostic Nothing $ \make_env -> do
    351 
    +    let env = DownsweepEnv
    352 
    +          { ds_hsc_env         = hsc_env
    353 
    +          , ds_summaries_cache = summs
    354 
    +          , ds_imports_cache   = imps
    355 
    +          , ds_mode            = DownsweepUseFixed
    356 
    +          , ds_excl_mods       = []
    357 
    +          , ds_n_jobs          = njobs
    358 
    +          , ds_make_env        = make_env
    359 
    +          }
    360 
    +    ~(errs, mg) <- runDownsweepM env $
    361 
    +      downsweepFromRootNodes Nothing True
    362 
    +        [ModuleNodeCompile mod_summary] []
    363 
    +    let dflags = hsc_dflags hsc_env
    364 
    +    liftIO $ printOrThrowDiagnostics (hsc_logger hsc_env)
    365 
    +                                     (initPrintConfig dflags)
    366 
    +                                     (initDiagOpts dflags)
    367 
    +                                     (GhcDriverMessage <$> unionManyMessages errs)
    368 
    +    return (mkModuleGraph mg)
    342 369
    343 370 
     -- | Construct a module graph starting from the interactive context.
    344 371 
     -- Produces, a thunk, which when forced will perform the downsweep.
    ... ... @@ -362,13 +389,23 @@ downsweepInteractiveImports hsc_env ic = unsafeInterleaveIO $ do
    362 389 
       -- :load. Any home package modules need to already be in here.
    363 390 
       let cached_nodes = Map.fromList [ (mkNodeKey n, NSuccess n) | n <- mg_mss (hsc_mod_graph hsc_env) ]
    364 391
    365 
    -  summ_cache <- newIORef mempty
    366 
    -  imps_cache <- newIORef mempty
    367 
    -  let env = DownsweepEnv hsc_env DownsweepUseFixed{-or DownsweepUseCompile?-} summ_cache imps_cache []
    368 
    -  graph <- runDownsweepM env do
    369 
    -    loopFromInteractive cached_nodes interactive_mn imps
    370 
    -  let all_nodes  = [s | NSuccess s <- M.elems graph ]
    371 
    -  return $ mkModuleGraph all_nodes
    392 
    +  n_jobs     <- mkWorkerLimit (hsc_dflags hsc_env)
    393 
    +  summ_cache <- newMVar mempty
    394 
    +  imps_cache <- newMVar mempty
    395 
    +  withMakeEnv n_jobs hsc_env mkUnknownDiagnostic Nothing $ \make_env -> do
    396 
    +    let env = DownsweepEnv
    397 
    +          { ds_hsc_env         = hsc_env
    398 
    +          , ds_mode            = DownsweepUseFixed{-or DownsweepUseCompile?-}
    399 
    +          , ds_summaries_cache = summ_cache
    400 
    +          , ds_imports_cache   = imps_cache
    401 
    +          , ds_excl_mods       = []
    402 
    +          , ds_n_jobs          = n_jobs
    403 
    +          , ds_make_env        = make_env
    404 
    +          }
    405 
    +    graph <- runDownsweepM env do
    406 
    +      loopFromInteractive cached_nodes interactive_mn imps
    407 
    +    let all_nodes  = [s | NSuccess s <- M.elems graph ]
    408 
    +    return $ mkModuleGraph all_nodes
    372 409
    373 410 
     -- | Create a module graph from a list of installed modules.
    374 411 
     -- This is used by the loader when we need to load modules but there
    ... ... @@ -396,26 +433,38 @@ downsweepInstalledModules hsc_env mods = do
    396 433 
                 -- already know that we can find the modules we need to load.
    397 434 
                 _ -> throwGhcException $ ProgramError $ showSDoc (hsc_dflags hsc_env) $ text "downsweepInstalledModules: Could not find installed module" <+> ppr i
    398 435
    436 
    +    njobs <- mkWorkerLimit (hsc_dflags hsc_env)
    399 437 
         nodes <- mapM process installed_mods
    400 
    -    summs <- newIORef mempty
    401 
    -    imps  <- newIORef mempty
    402 
    -    (errs, mg) <- downsweepFromRootNodes hsc_env summs imps Nothing [] True DownsweepUseFixed nodes external_uids
    438 
    +    summs <- newMVar mempty
    439 
    +    imps  <- newMVar mempty
    440 
    +    withMakeEnv njobs hsc_env mkUnknownDiagnostic Nothing $ \make_env -> do
    441 
    +      let env = DownsweepEnv
    442 
    +            { ds_hsc_env         = hsc_env
    443 
    +            , ds_summaries_cache = summs
    444 
    +            , ds_imports_cache   = imps
    445 
    +            , ds_mode            = DownsweepUseFixed
    446 
    +            , ds_excl_mods       = []
    447 
    +            , ds_n_jobs          = njobs
    448 
    +            , ds_make_env        = make_env
    449 
    +            }
    450 
    +      (errs, mg) <- runDownsweepM env $
    451 
    +        downsweepFromRootNodes Nothing True nodes external_uids
    403 452
    404 
    -    -- Similarly here, we should really not get any errors, but print them out if we do.
    405 
    -    let dflags = hsc_dflags hsc_env
    406 
    -    liftIO $ printOrThrowDiagnostics (hsc_logger hsc_env)
    407 
    -                                     (initPrintConfig dflags)
    408 
    -                                     (initDiagOpts dflags)
    409 
    -                                     (GhcDriverMessage <$> unionManyMessages errs)
    453 
    +      -- Similarly here, we should really not get any errors, but print them out if we do.
    454 
    +      let dflags = hsc_dflags hsc_env
    455 
    +      liftIO $ printOrThrowDiagnostics (hsc_logger hsc_env)
    456 
    +                                       (initPrintConfig dflags)
    457 
    +                                       (initDiagOpts dflags)
    458 
    +                                       (GhcDriverMessage <$> unionManyMessages errs)
    410 459
    411 
    -    return (mkModuleGraph mg)
    460 
    +      return (mkModuleGraph mg)
    412 461
    413 462 
     -----------------------------------------------------------------------------
    414 463 
     -- * Orchestrator: downsweepFromRootNodes
    415 464 
     -----------------------------------------------------------------------------
    416 465
    417 
    -type ModSummaryCache = IORef ModSummaryCacheMap
    418 
    -type ImportsCache    = IORef ImportsCacheMap
    466 
    +type ModSummaryCache = MVar ModSummaryCacheMap
    467 
    +type ImportsCache    = MVar ImportsCacheMap
    419 468
    420 469 
     -- | A cache from file paths to the already summarised modules. The same file
    421 470 
     -- can be used in multiple units so the map is actually also keyed by which
    ... ... @@ -450,30 +499,26 @@ data DownsweepMode = DownsweepUseCompile | DownsweepUseFixed
    450 499 
     -- 'UnitId's.
    451 500 
     -- This function will start at the given roots, and traverse downwards to find
    452 501 
     -- all the dependencies, all the way to the leaf units.
    453 
    -downsweepFromRootNodes :: HscEnv
    454 
    -                  -> ModSummaryCache
    455 
    -                  -> ImportsCache
    456 
    -                  -> Maybe ModuleGraph
    457 
    -                  -> [ModuleName]
    458 
    -                  -> Bool
    459 
    -                  -> DownsweepMode -- ^ Whether to create fixed or compile nodes for dependencies
    460 
    -                  -> [ModuleNodeInfo] -- ^ The starting ModuleNodeInfo
    461 
    -                  -> [UnitId] -- ^ The starting units
    462 
    -                  -> IO ([DriverMessages], [ModuleGraphNode])
    463 
    -downsweepFromRootNodes hsc_env summ_cache imps_cache maybe_base_graph excl_mods allow_dup_roots mode root_nodes root_uids = do
    502 
    +downsweepFromRootNodes
    503 
    +  :: Maybe ModuleGraph
    504 
    +  -> Bool
    505 
    +  -> [ModuleNodeInfo] -- ^ The starting ModuleNodeInfo
    506 
    +  -> [UnitId] -- ^ The starting units
    507 
    +  -> DownsweepM ([DriverMessages], [ModuleGraphNode])
    508 
    +downsweepFromRootNodes maybe_base_graph allow_dup_roots root_nodes root_uids =
    509 
    +  ReaderT $ \env@DownsweepEnv{..} -> do
    464 510 
          when (not allow_dup_roots) $
    465 511 
            case root_duplicates of
    466 512 
              []           -> return ()
    467 
    -         (dup_root:_) -> multiRootsErr sec dup_root
    468 
    -     modifyImpsCache imps_cache (`M.union` mkRootMap root_nodes) -- add root nodes to imports cache
    469 
    -     let env = DownsweepEnv hsc_env mode summ_cache imps_cache excl_mods
    470 
    -     deps' <- runDownsweepM env  $ do
    513 
    +         (dup_root:_) -> multiRootsErr (sec ds_hsc_env) dup_root
    514 
    +     modifyImpsCache ds_imports_cache (`M.union` mkRootMap root_nodes) -- add root nodes to imports cache
    515 
    +     deps' <- runDownsweepM env $ do
    471 516 
             let base_nodes = maybe M.empty moduleGraphNodeMap maybe_base_graph
    472 517 
             module_deps <- loopModuleNodeInfos base_nodes root_nodes
    473 
    -        all_deps    <- loopUnits module_deps (hscActiveUnitId hsc_env) root_uids
    474 
    -        deps'       <- loopInstantiations all_deps (getHomeUnitInstantiations hsc_env)
    518 
    +        all_deps    <- loopUnits module_deps (hscActiveUnitId ds_hsc_env) root_uids
    519 
    +        deps'       <- loopInstantiations all_deps (getHomeUnitInstantiations ds_hsc_env)
    475 520 
             return deps'
    476 
    -     f_cache <- readIORef summ_cache
    521 
    +     f_cache <- readMVar ds_summaries_cache
    477 522 
          let downsweep_errs = lefts (M.elems f_cache)
    478 523 
              downsweep_nodes = [ s | NSuccess s <- M.elems deps' ]
    479 524
    ... ... @@ -501,7 +546,7 @@ downsweepFromRootNodes hsc_env summ_cache imps_cache maybe_base_graph excl_mods
    501 546 
         moduleGraphNodeMap graph
    502 547 
             = M.fromList [(mkNodeKey node, NSuccess node) | node <- mgModSummaries' graph]
    503 548
    504 
    -    sec = initSourceErrorContext (hsc_dflags hsc_env)
    549 
    +    sec hsc_env = initSourceErrorContext (hsc_dflags hsc_env)
    505 550
    506 551 
     --------------------------------------------------------------------------------
    507 552 
     -- ** 'DownsweepM'
    ... ... @@ -509,11 +554,14 @@ downsweepFromRootNodes hsc_env summ_cache imps_cache maybe_base_graph excl_mods
    509 554
    510 555 
     type DownsweepM a = ReaderT DownsweepEnv IO a
    511 556 
     data DownsweepEnv = DownsweepEnv {
    512 
    -      downsweep_hsc_env :: HscEnv
    513 
    -    , _downsweep_mode :: DownsweepMode
    514 
    -    , _downsweep_summaries_cache :: ModSummaryCache
    515 
    -    , downsweep_imports_cache :: ImportsCache
    516 
    -    , _downsweep_excl_mods :: [ModuleName]
    557 
    +      ds_hsc_env         :: HscEnv
    558 
    +    , ds_mode            :: DownsweepMode
    559 
    +      -- ^ Whether to create fixed or compile nodes for dependencies
    560 
    +    , ds_summaries_cache :: ModSummaryCache
    561 
    +    , ds_imports_cache   :: ImportsCache
    562 
    +    , ds_excl_mods       :: [ModuleName]
    563 
    +    , ds_n_jobs          :: WorkerLimit
    564 
    +    , ds_make_env        :: MakeEnv
    517 565 
     }
    518 566
    519 567 
     mkModSummaryCache :: [(ModSummary, SummProvenance)] -> ModSummaryCacheMap
    ... ... @@ -529,8 +577,8 @@ addModSummaryCache ms pr fe = upd_fe fe
    529 577
    530 578 
     modifySummCache :: ModSummaryCache -> (ModSummaryCacheMap -> ModSummaryCacheMap) -> IO ()
    531 579 
     modifyImpsCache :: ImportsCache    -> (ImportsCacheMap    -> ImportsCacheMap)    -> IO ()
    532 
    -modifySummCache r f = atomicModifyIORef' r (\c -> (f c, ()))
    533 
    -modifyImpsCache r f = atomicModifyIORef' r (\c -> (f c, ()))
    580 
    +modifySummCache v f = modifyMVar v (\c -> let !r = f c in pure (r, ()))
    581 
    +modifyImpsCache v f = modifyMVar v (\c -> let !r = f c in pure (r, ()))
    534 582
    535 583 
     -- | A cache from a module import (in given home unit context, with a package
    536 584 
     -- qualifier, and the imported module name (with or without SOURCE)) to the
    ... ... @@ -553,7 +601,7 @@ loopModuleNodeInfos :: M.Map NodeKey (NodeRes ModuleGraphNode) -> [ModuleNodeInf
    553 601 
     loopUnits           :: M.Map NodeKey (NodeRes ModuleGraphNode) -> UnitId -> [UnitId]            -> DownsweepM (M.Map NodeKey (NodeRes ModuleGraphNode))
    554 602 
     loopInstantiations  :: M.Map NodeKey (NodeRes ModuleGraphNode) -> [(UnitId, InstantiatedUnit)]  -> DownsweepM (M.Map NodeKey (NodeRes ModuleGraphNode))
    555 603 
     loopFromInteractive :: M.Map NodeKey (NodeRes ModuleGraphNode) -> Module -> [InteractiveImport] -> DownsweepM (M.Map NodeKey (NodeRes ModuleGraphNode))
    556 
    -loopDownsweepNodes  base_map nodes = dfsBuild (Just base_map) nodes dsNodeInfoKey dsNodeExpand
    604 
    +loopDownsweepNodes  base_map nodes = parDfsBuild (Just base_map) nodes dsNodeInfoKey dsNodeExpand
    557 605 
     loopModuleNodeInfos base_map       = loopDownsweepNodes base_map . map DSMod
    558 606 
     loopUnits           base_map homud = loopDownsweepNodes base_map . map (DSUnit homud)
    559 607 
     loopInstantiations  base_map       = loopDownsweepNodes base_map . map (uncurry DSInst)
    ... ... @@ -617,7 +665,7 @@ dsNodeExpand = \case
    617 665
    618 666 
     expandModuleSummary :: ModSummary -> DownsweepM (NodeRes (ModuleGraphNode, [DownsweepNode]))
    619 667 
     expandModuleSummary ms = do -- Didn't work out what the imports mean yet, now do that.
    620 
    -    hsc_env <- asks downsweep_hsc_env
    668 
    +    hsc_env <- asks ds_hsc_env
    621 669 
         let home_uid  = ms_unitid ms
    622 670 
             home_unit = ue_unitHomeUnit home_uid (hsc_unit_env hsc_env)
    623 671 
         (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
    652 700 
             FoundHomeWithError (_uid, _e) -> return
    653 701 
               ( Nothing, [] )
    654 702 
               -- the error @e@ is already stored in the summarisation cache,
    655 
    -          -- (the IORef in DownsweepM) and will get reported at the end.
    703 
    +          -- (the MVar in DownsweepM) and will get reported at the end.
    656 704 
             FoundHome s -> return
    657 705 
               -- MP: This assumes that we can only instantiate non home units, which is probably fair enough for now.
    658 706 
               ( 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
    673 721 
     -- NB: If you ever reach a Fixed node, everything under that also must be fixed.
    674 722 
     expandFixedModuleNode :: ModNodeKeyWithUid -> ModLocation -> DownsweepM (NodeRes (ModuleGraphNode, [DownsweepNode]))
    675 723 
     expandFixedModuleNode key loc = do
    676 
    -    hsc_env <- asks downsweep_hsc_env
    724 
    +    hsc_env <- asks ds_hsc_env
    677 725 
         -- MP: TODO, we should just read the dependency info from the interface rather than either
    678 726 
         -- a. Loading the whole thing into the EPS (this might never nececssary and causes lots of things to be permanently loaded into memory)
    679 727 
         -- 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
    732 780 
     expandUnitNode node_uid home_context_uid = do
    733 781 
         -- Set active unit so that looking loopUnit finds the correct
    734 782 
         -- -package flags in the unit state.
    735 
    -    hsc_env <- asks downsweep_hsc_env
    783 
    +    hsc_env <- asks ds_hsc_env
    736 784 
         let lcl_hsc_env = hscSetActiveUnitId home_context_uid hsc_env
    737 785 
         case unitDepends <$> lookupUnitId (hsc_units lcl_hsc_env) node_uid of
    738 786 
           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
    745 793
    746 794 
     expandInteractiveImports :: Module -> [InteractiveImport] -> DownsweepM (NodeRes (ModuleGraphNode, [DownsweepNode]))
    747 795 
     expandInteractiveImports imod imps = do
    748 
    -  hsc_env    <- asks downsweep_hsc_env
    749 
    -  imps_cache <- asks downsweep_imports_cache
    796 
    +  hsc_env    <- asks ds_hsc_env
    797 
    +  imps_cache <- asks ds_imports_cache
    750 798
    751 799 
       let
    752 800 
         -- A simple edge to a module from the same home unit
    ... ... @@ -807,13 +855,13 @@ downsweepSummarise :: HomeUnit
    807 855 
                        -> Maybe (StringBuffer, UTCTime)
    808 856 
                        -> DownsweepM SummariseResult
    809 857 
     downsweepSummarise home_unit imp maybe_buf = do
    810 
    -  DownsweepEnv hsc_env mode summaries_cache_ref imports_cache_ref excl_mods <- ask
    811 
    -  liftIO $ case mode of
    858 
    +  DownsweepEnv{..} <- ask
    859 
    +  liftIO $ case ds_mode of
    812 860 
         DownsweepUseCompile ->
    813 
    -      summariseModule hsc_env home_unit summaries_cache_ref imports_cache_ref
    814 
    -                      imp maybe_buf excl_mods
    861 
    +      summariseModule ds_hsc_env home_unit ds_summaries_cache ds_imports_cache
    862 
    +                      imp maybe_buf ds_excl_mods
    815 863 
         DownsweepUseFixed ->
    816 
    -      summariseModuleInterface hsc_env home_unit imports_cache_ref imp excl_mods
    864 
    +      summariseModuleInterface ds_hsc_env home_unit ds_imports_cache imp ds_excl_mods
    817 865
    818 866 
     multiRootsErr :: SourceErrorContext -> NE.NonEmpty ModuleNodeInfo -> IO ()
    819 867 
     multiRootsErr sec (summ1 NE.:| summs)
    ... ... @@ -878,56 +926,15 @@ getRootSummary excl_mods summ_cache imports_cache hsc_env target
    878 926 
           rootLoc = mkGeneralSrcSpan (fsLit "<command line>")
    879 927 
           dflags = homeUnitEnv_dflags (ue_findHomeUnitEnv uid (hsc_unit_env hsc_env))
    880 928
    881 
    --- | Execute 'getRootSummary' for the 'Target's using the parallelism pipeline
    882 
    --- system.
    883 
    --- Create bundles of 'Target's wrapped in a 'MakeAction' that uses
    884 
    --- 'withAbstractSem' to wait for a free slot, limiting the number of
    885 
    --- concurrently computed summaries to the value of the @-j@ option or the slots
    886 
    --- allocated by the job server, if that is used.
    887 
    ---
    888 
    --- The 'MakeAction' returns 'Maybe', which is not handled as an error, because
    889 
    --- 'runLoop' only sets it to 'Nothing' when an exception was thrown, so the
    890 
    --- result won't be read anyway here.
    891 
    ---
    892 
    --- To emulate the current behavior, we funnel exceptions past the concurrency
    893 
    --- barrier and rethrow the first one afterwards.
    894 
    -rootSummariesParallel ::
    895 
    -  WorkerLimit ->
    896 
    -  HscEnv ->
    897 
    -  (GhcMessage -> AnyGhcDiagnostic) ->
    898 
    -  Maybe Messager ->
    899 
    -  (HscEnv -> Target -> IO (Either DriverMessages ModSummary)) ->
    900 
    -  IO ([DriverMessages], [ModSummary])
    901 
    -rootSummariesParallel n_jobs hsc_env diag_wrapper msg get_summary = do
    902 
    -  (actions, get_results) <- unzip <$> mapM action_and_result (zip [1..] bundles)
    903 
    -  runPipelines n_jobs hsc_env diag_wrapper msg actions
    904 
    -  (sequence . catMaybes <$> sequence get_results) >>= \case
    905 
    -    Right results -> pure (partitionEithers (concat results))
    906 
    -    Left exc -> throwIO exc
    907 
    -  where
    908 
    -    bundles = mk_bundles targets
    909 
    -
    910 
    -    mk_bundles = unfoldr \case
    911 
    -      [] -> Nothing
    912 
    -      ts -> Just (splitAt bundle_size ts)
    913 
    -
    914 
    -    bundle_size = 20
    915 
    -
    916 
    -    targets = hsc_targets hsc_env
    917 
    -
    918 
    -    action_and_result (log_queue_id, ts) = do
    919 
    -      res_var <- liftIO newEmptyMVar
    920 
    -      pure $! (MakeAction (action log_queue_id ts) res_var, readMVar res_var)
    921 
    -
    922 
    -    action log_queue_id target_bundle = do
    923 
    -      env@MakeEnv {compile_sem} <- ask
    924 
    -      lift $ lift $
    925 
    -        withAbstractSem compile_sem $
    926 
    -        withLoggerHsc log_queue_id env \ lcl_hsc_env ->
    927 
    -          MC.try (mapM (get_summary lcl_hsc_env) target_bundle) >>= \case
    928 
    -            Left e | Just (_ :: SomeAsyncException) <- fromException e ->
    929 
    -              throwIO e
    930 
    -            a -> pure a
    929 
    +-- | Execute 'getRootSummary' for the 'Target's using the parallelism pipeline system.
    930 
    +rootSummariesParallel
    931 
    +  :: WorkerLimit -> MakeEnv -> [Target]
    932 
    +  -> (HscEnv -> Target -> IO (Either DriverMessages ModSummary))
    933 
    +  -> IO ([DriverMessages], [ModSummary])
    934 
    +rootSummariesParallel n_jobs make_env targets get_summary = do
    935 
    +  partitionEithers <$> mapConcDS n_jobs bundle_size make_env get_summary targets
    936 
    +    where
    937 
    +      bundle_size = 20
    931 938
    932 939 
     --------------------------------------------------------------------------------
    933 940 
     -- * Check/validate properties and error out
    ... ... @@ -1325,7 +1332,7 @@ summariseFile
    1325 1332 
             -> IO (Either DriverMessages ModSummary)
    1326 1333
    1327 1334 
     summariseFile hsc_env' home_unit summ_cache_ref src_fn mb_phase maybe_buf
    1328 
    -   = do file_summ_cache <- readIORef summ_cache_ref
    1335 
    +   = do file_summ_cache <- readMVar summ_cache_ref
    1329 1336 
             case M.lookup (homeUnitId home_unit, src_fn_os) file_summ_cache of
    1330 1337 
               Just (Right (chd_summary, SummFresh)) ->
    1331 1338 
                 -- Fresh: use it straight away
    ... ... @@ -1505,7 +1512,7 @@ summariseModuleDispatch k hsc_env' imps_cache_ref home_unit imp excl_mods
    1505 1512
    1506 1513 
         find_it :: IO SummariseResult
    1507 1514 
         find_it = do
    1508 
    -      imps_cache <- readIORef imps_cache_ref
    1515 
    +      imps_cache <- readMVar imps_cache_ref
    1509 1516 
           case M.lookup cache_key imps_cache of
    1510 1517 
             Just result -> return result
    1511 1518 
             Nothing -> do
    ... ... @@ -1547,7 +1554,7 @@ summariseModuleWithSource home_unit summ_cache_ref is_boot maybe_buf hsc_env loc
    1547 1554 
         -- Adjust location to point to the hs-boot source file,
    1548 1555 
         -- hi file, object file, when is_boot says so
    1549 1556 
         let src_fn = expectJust (ml_hs_file location)
    1550 
    -    summ_cache <- readIORef summ_cache_ref
    1557 
    +    summ_cache <- readMVar summ_cache_ref
    1551 1558
    1552 1559 
         -- Reject the cache result if the module name doesn't match the inferred
    1553 1560 
         -- module name based on the file name.
    ... ... @@ -1722,10 +1729,10 @@ getPreprocessedImports hsc_env src_fn mb_phase maybe_buf = do
    1722 1729 
       return PreprocessedImports {..}
    1723 1730
    1724 1731 
     --------------------------------------------------------------------------------
    1725 
    --- * Generic traversal of iteratively-built graph: dfsBuild
    1732 
    +-- * Generic traversal of iteratively-built graph: parDfsBuild
    1726 1733 
     --------------------------------------------------------------------------------
    1727 1734
    1728 
    --- | The result of expanding a node in 'dfsBuild'.
    1735 
    +-- | The result of expanding a node in 'parDfsBuild'.
    1729 1736 
     data NodeRes v
    1730 1737 
       -- | Computed the node payload successfully
    1731 1738 
       = NSuccess v
    ... ... @@ -1738,7 +1745,7 @@ data NodeRes v
    1738 1745 
       -- abort.
    1739 1746 
       | NSkip
    1740 1747
    1741 
    --- | In a depth-first order, and starting from the given roots, traverse a
    1748 
    +-- | In a parallel non-det-depth-first order, and starting from the given roots, traverse a
    1742 1749 
     -- graph by iteratively expanding a node into a payload and a list of children
    1743 1750 
     -- nodes to visit next.
    1744 1751 
     --
    ... ... @@ -1751,18 +1758,17 @@ data NodeRes v
    1751 1758 
     -- The result is a mapping from the key of every node transitively reachable
    1752 1759 
     -- from the root nodes (inclusively) to the payload returned by expanding that
    1753 1760 
     -- node. The result includes the previously visited nodes given in @base_map@,
    1754 
    --- s.t. @dfsBuild base_map [] _ _ == base_map@.
    1761 
    +-- s.t. @parDfsBuild base_map [] _ _ == base_map@.
    1755 1762 
     --
    1756 1763 
     -- The @expand@ function returns an 'NResult'. See the 'NResult' documentation
    1757 1764 
     -- for more information about each result type.
    1758 1765 
     --
    1759 
    --- Error handling and exiting early can be achieved by selecting a @Monad m@
    1760 
    --- accordingly, such as @Control.Monad.Except.Except@
    1761 
    ---
    1762 1766 
     -- Example usage: @n@ is instanced to @DownsweepNode@, @k@ is @NodeKey@, and @v@ is @ModuleNodeEdge@.
    1763 1767 
     --
    1764 
    --- See also Note [Downsweep Control Flow and Caching]
    1765 
    -dfsBuild :: (Ord k, Monad m)
    1768 
    +-- See Note [Parallel Downsweep] for more information about how parallelism is
    1769 
    +-- achieved, and See Note [Downsweep Control Flow and Caching] for information
    1770 
    +-- about the various caches used.
    1771 
    +parDfsBuild :: Ord k
    1766 1772 
              => Maybe (Map.Map k (NodeRes v))
    1767 1773 
              -- ^ Base map, existing results. We won't re-expand any of the nodes
    1768 1774 
              -- already present in this map.
    ... ... @@ -1770,34 +1776,91 @@ dfsBuild :: (Ord k, Monad m)
    1770 1776 
              -- ^ The root nodes from where to start traversal
    1771 1777 
              -> (n -> k)
    1772 1778 
              -- ^ Compute the key which uniquely identifies this node
    1773 
    -         -> (n -> m (NodeRes (v,[n])))
    1779 
    +         -> (n -> DownsweepM (NodeRes (v,[n])))
    1774 1780 
              -- ^ Expand this node into its payload result and into the list of
    1775 1781 
              -- children nodes to visit next.
    1776 
    -         -> m (Map.Map k (NodeRes v))
    1782 
    +         -> DownsweepM (Map.Map k (NodeRes v))
    1777 1783 
              -- ^ The result accumulates the payload of expanding the root nodes
    1778 1784 
              -- and all nodes transitively reachable from those roots.
    1779 
    -dfsBuild base_map roots key expand = go roots (fromMaybe Map.empty base_map)
    1785 
    +parDfsBuild base_map roots key expand = ReaderT $ \ds_env -> do
    1786 
    +    exc_var     <- newTVarIO Nothing -- signal this var when there's an exception
    1787 
    +    visited_var <- newTVarIO (fromMaybe Map.empty base_map)
    1788 
    +    pending     <- newTVarIO Set.empty
    1789 
    +    worklist    <- newTQueueIO
    1790 
    +
    1791 
    +    coord_tid   <- forkIO $
    1792 
    +      coordinator ds_env exc_var visited_var worklist pending
    1793 
    +        `MC.catch` \case
    1794 
    +          (e::MC.SomeException)
    1795 
    +            -- exit cleanly when killed
    1796 
    +            | Just ThreadKilled <- fromException e -> return ()
    1797 
    +            -- if the coordinator somehow else crashes,
    1798 
    +            -- signal the exc_var for the main thread to throw it
    1799 
    +            | otherwise -> atomically (modifyTVar' exc_var (<|> Just e))
    1800 
    +
    1801 
    +    mapM_ (atomically . writeTQueue worklist) roots
    1802 
    +
    1803 
    +    -- this txn retries until all work is done or there is an exception
    1804 
    +    mb_exc <- atomically $ do
    1805 
    +      readTVar exc_var >>= \case
    1806 
    +        Just e -> return (Just e)
    1807 
    +        Nothing -> do
    1808 
    +          empty_worklist <- isEmptyTQueue worklist
    1809 
    +          empty_pending  <- Set.null <$> readTVar pending
    1810 
    +          unless (empty_worklist && empty_pending) retry
    1811 
    +          return Nothing
    1812 
    +
    1813 
    +    killThread coord_tid
    1814 
    +    case mb_exc of
    1815 
    +      Just e  -> throwIO e
    1816 
    +      Nothing -> readTVarIO visited_var
    1817 
    +
    1780 1818 
       where
    1781 
    -    go []     visited = pure visited
    1782 
    -    go (s:ss) visited
    1783 
    -      | k `Map.member` visited
    1784 
    -      = go ss visited
    1785 
    -      | otherwise
    1786 
    -      = do r <- expand s
    1787 
    -           case r of
    1788 
    -             NSkip ->
    1789 
    -               go ss
    1790 
    -                  (Map.insert k NSkip        visited) -- Skip!
    1791 
    -             NSuccess (v,ns) ->
    1792 
    -               go (ns ++ ss)
    1793 
    -                  (Map.insert k (NSuccess v) visited)
    1794 
    -      where
    1795 
    -        k = key s
    1819 
    +    coordinator ds_env exc_var visvar worklist pendvar = forever $ do
    1820 
    +      mb_node_to_expand <- atomically $ do
    1821 
    +        node <- readTQueue worklist
    1822 
    +        let k = key node
    1823 
    +
    1824 
    +        visited <- readTVar visvar
    1825 
    +        pending <- readTVar pendvar
    1826 
    +
    1827 
    +        if (k `Set.member` pending || k `Map.member` visited)
    1828 
    +          then return Nothing
    1829 
    +          else do
    1830 
    +            -- must add to pending in the same transaction as worklist dequeue,
    1831 
    +            -- otherwise the main thread may find both the worklist and pending
    1832 
    +            -- lists empty and exit prematurely.
    1833 
    +            modifyTVar' pendvar (Set.insert k)
    1834 
    +            return (Just (k, node))
    1835 
    +
    1836 
    +      case mb_node_to_expand of
    1837 
    +        Nothing        -> return ()
    1838 
    +        Just (k, node) -> void $ do
    1839 
    +          withLocalTmpFSMake (ds_make_env ds_env) $ \make_env ->
    1840 
    +            forkIOWithUnmask $ \unmask ->
    1841 
    +              unmask (worker ds_env{ds_make_env = make_env} visvar worklist pendvar k node)
    1842 
    +                -- write exception in the worker to the main thread
    1843 
    +                `MC.catch` \(e::MC.SomeException) ->
    1844 
    +                  atomically (modifyTVar' exc_var (<|> Just e))
    1845 
    +
    1846 
    +    worker ds_env@DownsweepEnv{..} visvar worklist pendvar k node =
    1847 
    +      withAbstractSem (compile_sem ds_make_env) $ do
    1848 
    +        r <- runDownsweepM ds_env $
    1849 
    +             expand node -- do the main work!
    1850 
    +
    1851 
    +        case r of
    1852 
    +          NSkip ->
    1853 
    +            atomically $ modifyTVar' visvar (Map.insert k NSkip)
    1854 
    +          NSuccess (v,ns) -> do
    1855 
    +            atomically $ modifyTVar' visvar (Map.insert k (NSuccess v))
    1856 
    +            mapM_ (atomically . writeTQueue worklist) ns
    1857 
    +
    1858 
    +        atomically $ modifyTVar' pendvar (Set.delete k)
    1796 1859
    1797 1860 
     {-
    1798 1861 
     Note [Downsweep Control Flow and Caching]
    1799 1862 
     ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
    1800 
    -The control flow of downsweep is extracted into a single function `dfsBuild`,
    1863 
    +The control flow of downsweep is extracted into a single function `parDfsBuild`,
    1801 1864 
     which takes care of iteratively expanding and traversing all nodes of the
    1802 1865 
     in-construction module graph necessary to build a full `ModuleGraph` at the
    1803 1866 
     end.
    ... ... @@ -1806,7 +1869,7 @@ There are three levels of caching going on, all of which are necessary to make
    1806 1869 
     sure we don't do repeated work (notably, we NEVER summarise the same module
    1807 1870 
     twice).
    1808 1871
    1809 
    -1. `dfsBuild` accumulates the final module graph and never revisits the
    1872 
    +1. `parDfsBuild` accumulates the final module graph and never revisits the
    1810 1873 
        same node of the module graph. Cache is keyed by the final
    1811 1874 
        `ModuleGraph`s `NodeKey`s.
    1812 1875
    ... ... @@ -1874,6 +1937,83 @@ twice).
    1874 1937
    1875 1938 
        See tests T27461a and T27461b.
    1876 1939
    1877 
    -See also Note [Downsweep: building and maintaining the module graph] and
    1878 
    -Note [The ModuleGraph].
    1940 
    +See also Note [Downsweep: building and maintaining the module graph] and Note [The ModuleGraph].
    1941 
    +
    1942 
    +
    1943 
    +Note [Parallel Downsweep]
    1944 
    +~~~~~~~~~~~~~~~~~~~~~~~~~
    1945 
    +Downsweep traverses the modules iteratively to discover the module graph
    1946 
    +structure (see Note [Downsweep: building and maintaining the module graph])
    1947 
    +
    1948 
    +Each module has to be expanded/processed to discover dependencies amongst other
    1949 
    +things, and that processing can often be costly (e.g. see `expandModuleSummary`).
    1950 
    +
    1951 
    +We leverage multiple threads in this traversal to expand more than one module
    1952 
    +at once, respecting -j<N> to mean we never expand more than N modules at once.
    1953 
    +The parallel downsweep is all handled by `parDfsBuild` as follows:
    1954 
    +
    1955 
    +- We launch a thread for every module we discover that needs to be
    1956 
    +  expanded in the `coordinator` thread, popping it from the worklist
    1957 
    +- Every launched `worker` thread blocks waiting for a semaphore token
    1958 
    +  (`withAbstractSem`) to respect -j<N>
    1959 
    +- The main thread waits until both the worklist and pending list is
    1960 
    +  cleared, atomically.
    1961 
    +
    1962 
    +STM is used crucially to guarantee e.g. we don't have race conditions
    1963 
    +between taking from the worklist and writing to the pending list while
    1964 
    +checking whether they are clear.
    1965 
    +
    1966 
    +Exceptions are bubbled up to the main thread. The "main" thread, which is
    1967 
    +typically waiting for the worklist+pending lists to be clear, instead gets
    1968 
    +unblocked by this exception (signaled in `exc_var`) and re-throws it.
    1879 1969 
     -}
    1970 
    +
    1971 
    +--------------------------------------------------------------------------------
    1972 
    +-- * Concurrent utilities
    1973 
    +--------------------------------------------------------------------------------
    1974 
    +
    1975 
    +-- | Map an action over a list using the parallelism pipeline system.
    1976 
    +-- Create bundles of the list elems wrapped in a 'MakeAction' that uses
    1977 
    +-- 'withAbstractSem' to wait for a free slot, limiting the number of
    1978 
    +-- concurrently computed summaries to the value of the @-j@ option or the slots
    1979 
    +-- allocated by the job server, if that is used.
    1980 
    +--
    1981 
    +-- The 'MakeAction' returns 'Maybe', which is not handled as an error, because
    1982 
    +-- 'runLoop' only sets it to 'Nothing' when an exception was thrown, so the
    1983 
    +-- result won't be read anyway here.
    1984 
    +--
    1985 
    +-- To emulate the current behavior, we funnel exceptions past the concurrency
    1986 
    +-- barrier and rethrow the first one afterwards.
    1987 
    +mapConcDS ::
    1988 
    +  WorkerLimit ->
    1989 
    +  Int {-^ Batch size -} ->
    1990 
    +  MakeEnv ->
    1991 
    +  (HscEnv -> a -> IO b) ->
    1992 
    +  [a] ->
    1993 
    +  IO ([b])
    1994 
    +mapConcDS n_jobs bundle_size make_env run_action xs = do
    1995 
    +  (actions, get_results) <- unzip <$> mapM action_and_result (zip [1..] bundles)
    1996 
    +  runAllPipelines n_jobs make_env actions
    1997 
    +  (sequence . catMaybes <$> sequence get_results) >>= \case
    1998 
    +    Right results -> pure (concat results)
    1999 
    +    Left exc -> throwIO exc
    2000 
    +  where
    2001 
    +    bundles = mk_bundles xs
    2002 
    +
    2003 
    +    mk_bundles = unfoldr \case
    2004 
    +      [] -> Nothing
    2005 
    +      ts -> Just (splitAt bundle_size ts)
    2006 
    +
    2007 
    +    action_and_result (log_queue_id, ts) = do
    2008 
    +      res_var <- liftIO newEmptyMVar
    2009 
    +      pure $! (MakeAction (action log_queue_id ts) res_var, readMVar res_var)
    2010 
    +
    2011 
    +    action log_queue_id target_bundle = do
    2012 
    +      env@MakeEnv {compile_sem} <- ask
    2013 
    +      lift $ lift $
    2014 
    +        withAbstractSem compile_sem $
    2015 
    +        withLoggerHsc log_queue_id env \ lcl_hsc_env ->
    2016 
    +          MC.try (mapM (run_action lcl_hsc_env) target_bundle) >>= \case
    2017 
    +            Left e | Just (_ :: SomeAsyncException) <- fromException e ->
    2018 
    +              throwIO e
    2019 
    +            a -> pure a

  • compiler/GHC/Driver/MakeAction.hs
    1 1 
     {-# LANGUAGE CPP #-}
    2 2 
     module GHC.Driver.MakeAction
    3 3 
       ( MakeAction(..)
    4 
    -  , MakeEnv(..)
    5 4 
       , RunMakeM
    5 
    +  , MakeEnv(..)
    6 
    +  , withMakeEnv
    6 7 
       -- * Running the pipelines
    7 8 
       , runAllPipelines
    8 
    -  , runParPipelines
    9 
    -  , runSeqPipelines
    10 9 
       , runPipelines
    11 10 
       -- * Worker limit
    12 11 
       , WorkerLimit(..)
    ... ... @@ -66,31 +65,9 @@ import GHC.Driver.Pipeline.LogQueue
    66 65 
     import Control.Concurrent.STM
    67 66 
     import Control.Monad.Trans.Maybe
    68 67
    69 
    --- Executing the pipelines
    70 
    -
    71 
    -mkWorkerLimit :: DynFlags -> IO WorkerLimit
    72 
    -mkWorkerLimit dflags =
    73 
    -  case parMakeCount dflags of
    74 
    -    Nothing -> pure $ num_procs 1
    75 
    -    Just (ParMakeSemaphore h) -> pure (JSemLimit h)
    76 
    -    Just ParMakeNumProcessors -> num_procs <$> getNumProcessors
    77 
    -    Just (ParMakeThisMany n) -> pure $ num_procs n
    78 
    -  where
    79 
    -    num_procs x = NumProcessorsLimit (max 1 x)
    80 
    -
    81 
    -isWorkerLimitSequential :: WorkerLimit -> Bool
    82 
    -isWorkerLimitSequential (NumProcessorsLimit x) = x <= 1
    83 
    -isWorkerLimitSequential (JSemLimit {})         = False
    84 
    -
    85 
    --- | This describes what we use to limit the number of jobs, either we limit it
    86 
    --- ourselves to a specific number or we have an external parallelism semaphore
    87 
    --- limit it for us.
    88 
    -data WorkerLimit
    89 
    -  = NumProcessorsLimit Int
    90 
    -  | JSemLimit
    91 
    -    SemaphoreIdentifier
    92 
    -      -- ^ Semaphore identifier from @-jsem@
    93 
    -  deriving Eq
    68 
    +--------------------------------------------------------------------------------
    69 
    +-- * MakeEnv and MakeAction
    70 
    +--------------------------------------------------------------------------------
    94 71
    95 72 
     -- | Environment used when compiling a module
    96 73 
     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
    104 81 
                            , diag_wrapper :: GhcMessage -> AnyGhcDiagnostic
    105 82 
                            }
    106 83
    84 
    +-- | Come up with a 'MakeEnv' based on the given 'WorkerLimit'.
    85 
    +-- For -j1, it will be a trivial 'MakeEnv' not prepared for parallelism.
    86 
    +-- For -jn, it can be used from multiple threads (e.g. in runAllPipelines when -jN)
    87 
    +withMakeEnv
    88 
    +  :: WorkerLimit    -- ^ How to limit work parallelism
    89 
    +  -> HscEnv         -- ^ The basic HscEnv which is augmented with specific info for each module
    90 
    +  -> (GhcMessage -> AnyGhcDiagnostic)
    91 
    +  -> Maybe Messager -- ^ Optional custom messager to use to report progress
    92 
    +  -> (MakeEnv -> IO r) -> IO r
    93 
    +withMakeEnv worker_limit hsc_env diag_wrapper mHscMessager act =
    94 
    +  if isWorkerLimitSequential worker_limit
    95 
    +    then withSeqMakeEnv
    96 
    +    else withParMakeEnv
    97 
    +  where
    98 
    +    withSeqMakeEnv = do
    99 
    +      let seq_env = MakeEnv
    100 
    +            { hsc_env = hsc_env
    101 
    +            , withLogger = \_ k -> k id
    102 
    +            , compile_sem = AbstractSem (return ()) (return ())
    103 
    +            , env_messager = mHscMessager
    104 
    +            , diag_wrapper = diag_wrapper
    105 
    +            }
    106 
    +      act seq_env
    107 
    +
    108 
    +    withParMakeEnv = do
    109 
    +      -- A variable which we write to when an error has happened and we have to tell the
    110 
    +      -- logging thread to gracefully shut down.
    111 
    +      stopped_var <- newTVarIO False
    112 
    +      -- The queue of LogQueues which actions are able to write to. When an action starts it
    113 
    +      -- will add it's LogQueue into this queue.
    114 
    +      log_queue_queue_var <- newTVarIO newLogQueueQueue
    115 
    +      -- Thread which coordinates the printing of logs
    116 
    +      wait_log_thread <- logThread (hsc_logger hsc_env) stopped_var log_queue_queue_var
    117 
    +
    118 
    +
    119 
    +      -- Make the logger thread-safe, in case there is some output which isn't sent via the LogQueue.
    120 
    +      thread_safe_logger <- liftIO $ makeThreadSafe (hsc_logger hsc_env)
    121 
    +      let thread_safe_hsc_env = hsc_env { hsc_logger = thread_safe_logger }
    122 
    +
    123 
    +      runWorkerLimit (hsc_logger hsc_env) (hsc_dflags hsc_env) worker_limit $ \abstract_sem -> do
    124 
    +        let env = MakeEnv { hsc_env = thread_safe_hsc_env
    125 
    +                          , withLogger = withParLog log_queue_queue_var
    126 
    +                          , compile_sem = abstract_sem
    127 
    +                          , env_messager = mHscMessager
    128 
    +                          , diag_wrapper = diag_wrapper
    129 
    +                          }
    130 
    +        -- Reset the number of capabilities once the upsweep ends.
    131 
    +        r <- act env
    132 
    +        atomically $ writeTVar stopped_var True
    133 
    +        wait_log_thread
    134 
    +        pure r
    135 
    +
    136 
    +-- ** MakeAction ---------------------------------------------------------------
    107 137
    108 
    -label_self :: String -> IO ()
    109 
    -label_self thread_name = do
    110 
    -    self_tid <- CC.myThreadId
    111 
    -    CC.labelThread self_tid thread_name
    138 
    +data MakeAction = forall a . MakeAction !(RunMakeM a) !(MVar (Maybe a))
    139 
    +
    140 
    +type RunMakeM a = ReaderT MakeEnv (MaybeT IO) a
    112 141
    142 
    +waitMakeAction :: MakeAction -> IO ()
    143 
    +waitMakeAction (MakeAction _ mvar) = () <$ readMVar mvar
    113 144
    114 
    -runPipelines :: WorkerLimit -> HscEnv -> (GhcMessage -> AnyGhcDiagnostic) -> Maybe Messager -> [MakeAction] -> IO ()
    115 
    --- Don't even initialise plugins if there are no pipelines
    145 
    +--------------------------------------------------------------------------------
    146 
    +-- * Running the pipelines
    147 
    +--------------------------------------------------------------------------------
    148 
    +
    149 
    +-- | Build and run a pipeline using the given worker limit for parallelism
    150 
    +runPipelines
    151 
    +  :: WorkerLimit -> HscEnv
    152 
    +  -> (GhcMessage -> AnyGhcDiagnostic) -> Maybe Messager
    153 
    +  -> [MakeAction] -- ^ The build plan for all the module nodes
    154 
    +  -> IO ()
    116 155 
     runPipelines n_job hsc_env diag_wrapper mHscMessager all_pipelines = do
    117 156 
       liftIO $ label_self "main --make thread"
    118 
    -  case n_job of
    119 
    -    NumProcessorsLimit n | n <= 1 -> runSeqPipelines hsc_env diag_wrapper mHscMessager all_pipelines
    120 
    -    _n -> runParPipelines n_job hsc_env diag_wrapper mHscMessager all_pipelines
    121 
    -
    122 
    -runSeqPipelines :: HscEnv -> (GhcMessage -> AnyGhcDiagnostic) -> Maybe Messager -> [MakeAction] -> IO ()
    123 
    -runSeqPipelines plugin_hsc_env diag_wrapper mHscMessager all_pipelines =
    124 
    -  let env = MakeEnv { hsc_env = plugin_hsc_env
    125 
    -                    , withLogger = \_ k -> k id
    126 
    -                    , compile_sem = AbstractSem (return ()) (return ())
    127 
    -                    , env_messager = mHscMessager
    128 
    -                    , diag_wrapper = diag_wrapper
    129 
    -                    }
    130 
    -  in runAllPipelines (NumProcessorsLimit 1) env all_pipelines
    157 
    +  withMakeEnv n_job hsc_env diag_wrapper mHscMessager $ \make_env -> do
    158 
    +    runAllPipelines n_job make_env all_pipelines
    159 
    +  where
    160 
    +    label_self :: String -> IO ()
    161 
    +    label_self thread_name = do
    162 
    +        self_tid <- CC.myThreadId
    163 
    +        CC.labelThread self_tid thread_name
    131 164
    132 
    -#if !(defined(wasm32_HOST_ARCH) || defined(javascript_HOST_ARCH))
    133 
    -runNjobsAbstractSem :: Int -> (AbstractSem -> IO a) -> IO a
    134 
    -runNjobsAbstractSem n_jobs action = do
    135 
    -  compile_sem <- newQSem n_jobs
    136 
    -  n_capabilities <- getNumCapabilities
    137 
    -  n_cpus <- getNumProcessors
    138 
    -  let
    139 
    -    asem = AbstractSem (waitQSem compile_sem) (signalQSem compile_sem)
    140 
    -    set_num_caps n = unless (n_capabilities /= 1) $ setNumCapabilities n
    141 
    -    updNumCapabilities =  do
    142 
    -      -- Setting number of capabilities more than
    143 
    -      -- CPU count usually leads to high userspace
    144 
    -      -- lock contention. #9221
    145 
    -      set_num_caps $ min n_jobs n_cpus
    146 
    -    resetNumCapabilities = set_num_caps n_capabilities
    147 
    -  MC.bracket_ updNumCapabilities resetNumCapabilities $ action asem
    165 
    +-- | Run the given actions and then wait for them all to finish.
    166 
    +runAllPipelines :: WorkerLimit -> MakeEnv -> [MakeAction] -> IO ()
    167 
    +runAllPipelines worker_limit env acts = do
    168 
    +  let single_worker = isWorkerLimitSequential worker_limit
    169 
    +      spawn_actions :: IO [ThreadId]
    170 
    +      spawn_actions = if single_worker
    171 
    +        then (:[]) <$> (forkIOWithUnmask $ \unmask -> void $ runLoop (\io -> io unmask) env acts)
    172 
    +        else runLoop forkIOWithUnmask env acts
    148 173
    149 
    -#endif
    174 
    +      kill_actions :: [ThreadId] -> IO ()
    175 
    +      kill_actions tids = mapM_ killThread tids
    176 
    +
    177 
    +  MC.bracket spawn_actions kill_actions $ \_ -> do
    178 
    +    mapM_ waitMakeAction acts
    179 
    +
    180 
    +-- | Execute each action in order, limiting the amount of parallelism by the given
    181 
    +-- semaphore.
    182 
    +runLoop :: (((forall a. IO a -> IO a) -> IO ()) -> IO a) -> MakeEnv -> [MakeAction] -> IO [a]
    183 
    +runLoop _ _env [] = return []
    184 
    +runLoop fork_thread env (MakeAction act res_var :acts) = do
    185 
    +
    186 
    +  -- withLocalTmpFs has to occur outside of fork to remain deterministic
    187 
    +  new_thread <- withLocalTmpFSMake env $ \lcl_env ->
    188 
    +    fork_thread $ \unmask -> (do
    189 
    +            mres <- (unmask $ run_pipeline lcl_env act)
    190 
    +                      `MC.onException` (putMVar res_var Nothing) -- Defensive: If there's an unhandled exception then still signal the failure.
    191 
    +            putMVar res_var mres)
    192 
    +  threads <- runLoop fork_thread env acts
    193 
    +  return (new_thread : threads)
    194 
    +  where
    195 
    +      run_pipeline :: MakeEnv -> RunMakeM a -> IO (Maybe a)
    196 
    +      run_pipeline env p = runMaybeT (runReaderT p env)
    197 
    +
    198 
    +--------------------------------------------------------------------------------
    199 
    +-- * Worker Limit
    200 
    +--------------------------------------------------------------------------------
    201 
    +
    202 
    +-- | This describes what we use to limit the number of jobs, either we limit it
    203 
    +-- ourselves to a specific number or we have an external parallelism semaphore
    204 
    +-- limit it for us.
    205 
    +data WorkerLimit
    206 
    +  = NumProcessorsLimit Int
    207 
    +  | JSemLimit
    208 
    +    SemaphoreIdentifier
    209 
    +      -- ^ Semaphore identifier from @-jsem@
    210 
    +  deriving Eq
    211 
    +
    212 
    +mkWorkerLimit :: DynFlags -> IO WorkerLimit
    213 
    +mkWorkerLimit dflags =
    214 
    +  case parMakeCount dflags of
    215 
    +    Nothing -> pure $ num_procs 1
    216 
    +    Just (ParMakeSemaphore h) -> pure (JSemLimit h)
    217 
    +    Just ParMakeNumProcessors -> num_procs <$> getNumProcessors
    218 
    +    Just (ParMakeThisMany n) -> pure $ num_procs n
    219 
    +  where
    220 
    +    num_procs x = NumProcessorsLimit (max 1 x)
    221 
    +
    222 
    +isWorkerLimitSequential :: WorkerLimit -> Bool
    223 
    +isWorkerLimitSequential (NumProcessorsLimit x) = x <= 1
    224 
    +isWorkerLimitSequential (JSemLimit {})         = False
    150 225
    151 226 
     runWorkerLimit :: Logger -> DynFlags -> WorkerLimit -> (AbstractSem -> IO a) -> IO a
    152 227 
     #if defined(wasm32_HOST_ARCH) || defined(javascript_HOST_ARCH)
    ... ... @@ -168,41 +243,28 @@ runWorkerLimit logger dflags worker_limit action = case worker_limit of
    168 243 
               runNjobsAbstractSem 1 action
    169 244 
     #endif
    170 245
    171 
    --- | Build and run a pipeline
    172 
    -runParPipelines :: WorkerLimit -- ^ How to limit work parallelism
    173 
    -             -> HscEnv         -- ^ The basic HscEnv which is augmented with specific info for each module
    174 
    -             -> (GhcMessage -> AnyGhcDiagnostic)
    175 
    -             -> Maybe Messager   -- ^ Optional custom messager to use to report progress
    176 
    -             -> [MakeAction]  -- ^ The build plan for all the module nodes
    177 
    -             -> IO ()
    178 
    -runParPipelines worker_limit plugin_hsc_env diag_wrapper mHscMessager all_pipelines = do
    179 
    -
    180 
    -
    181 
    -  -- A variable which we write to when an error has happened and we have to tell the
    182 
    -  -- logging thread to gracefully shut down.
    183 
    -  stopped_var <- newTVarIO False
    184 
    -  -- The queue of LogQueues which actions are able to write to. When an action starts it
    185 
    -  -- will add it's LogQueue into this queue.
    186 
    -  log_queue_queue_var <- newTVarIO newLogQueueQueue
    187 
    -  -- Thread which coordinates the printing of logs
    188 
    -  wait_log_thread <- logThread (hsc_logger plugin_hsc_env) stopped_var log_queue_queue_var
    189 
    -
    190 
    -
    191 
    -  -- Make the logger thread-safe, in case there is some output which isn't sent via the LogQueue.
    192 
    -  thread_safe_logger <- liftIO $ makeThreadSafe (hsc_logger plugin_hsc_env)
    193 
    -  let thread_safe_hsc_env = plugin_hsc_env { hsc_logger = thread_safe_logger }
    194 
    -
    195 
    -  runWorkerLimit (hsc_logger plugin_hsc_env) (hsc_dflags plugin_hsc_env) worker_limit $ \abstract_sem -> do
    196 
    -    let env = MakeEnv { hsc_env = thread_safe_hsc_env
    197 
    -                      , withLogger = withParLog log_queue_queue_var
    198 
    -                      , compile_sem = abstract_sem
    199 
    -                      , env_messager = mHscMessager
    200 
    -                      , diag_wrapper = diag_wrapper
    201 
    -                      }
    202 
    -    -- Reset the number of capabilities once the upsweep ends.
    203 
    -    runAllPipelines worker_limit env all_pipelines
    204 
    -    atomically $ writeTVar stopped_var True
    205 
    -    wait_log_thread
    246 
    +#if !(defined(wasm32_HOST_ARCH) || defined(javascript_HOST_ARCH))
    247 
    +runNjobsAbstractSem :: Int -> (AbstractSem -> IO a) -> IO a
    248 
    +runNjobsAbstractSem n_jobs action = do
    249 
    +  compile_sem <- newQSem n_jobs
    250 
    +  n_capabilities <- getNumCapabilities
    251 
    +  n_cpus <- getNumProcessors
    252 
    +  let
    253 
    +    asem = AbstractSem (waitQSem compile_sem) (signalQSem compile_sem)
    254 
    +    set_num_caps n = unless (n_capabilities /= 1) $ setNumCapabilities n
    255 
    +    updNumCapabilities =  do
    256 
    +      -- Setting number of capabilities more than
    257 
    +      -- CPU count usually leads to high userspace
    258 
    +      -- lock contention. #9221
    259 
    +      set_num_caps $ min n_jobs n_cpus
    260 
    +    resetNumCapabilities = set_num_caps n_capabilities
    261 
    +  MC.bracket_ updNumCapabilities resetNumCapabilities $ action asem
    262 
    +
    263 
    +#endif
    264 
    +
    265 
    +--------------------------------------------------------------------------------
    266 
    +-- * Utility
    267 
    +--------------------------------------------------------------------------------
    206 268
    207 269 
     withLoggerHsc :: Int -> MakeEnv -> (HscEnv -> IO a) -> IO a
    208 270 
     withLoggerHsc k MakeEnv{withLogger, hsc_env} cont = do
    ... ... @@ -238,44 +300,3 @@ withLocalTmpFSMake :: MakeEnv -> (MakeEnv -> IO a) -> IO a
    238 300 
     withLocalTmpFSMake env k =
    239 301 
       withLocalTmpFS (hsc_tmpfs (hsc_env env)) $ \lcl_tmpfs
    240 302 
         -> k (env { hsc_env = (hsc_env env) { hsc_tmpfs = lcl_tmpfs }})
    241 
    -
    242 
    -
    243 
    --- | Run the given actions and then wait for them all to finish.
    244 
    -runAllPipelines :: WorkerLimit -> MakeEnv -> [MakeAction] -> IO ()
    245 
    -runAllPipelines worker_limit env acts = do
    246 
    -  let single_worker = isWorkerLimitSequential worker_limit
    247 
    -      spawn_actions :: IO [ThreadId]
    248 
    -      spawn_actions = if single_worker
    249 
    -        then (:[]) <$> (forkIOWithUnmask $ \unmask -> void $ runLoop (\io -> io unmask) env acts)
    250 
    -        else runLoop forkIOWithUnmask env acts
    251 
    -
    252 
    -      kill_actions :: [ThreadId] -> IO ()
    253 
    -      kill_actions tids = mapM_ killThread tids
    254 
    -
    255 
    -  MC.bracket spawn_actions kill_actions $ \_ -> do
    256 
    -    mapM_ waitMakeAction acts
    257 
    -
    258 
    --- | Execute each action in order, limiting the amount of parallelism by the given
    259 
    --- semaphore.
    260 
    -runLoop :: (((forall a. IO a -> IO a) -> IO ()) -> IO a) -> MakeEnv -> [MakeAction] -> IO [a]
    261 
    -runLoop _ _env [] = return []
    262 
    -runLoop fork_thread env (MakeAction act res_var :acts) = do
    263 
    -
    264 
    -  -- withLocalTmpFs has to occur outside of fork to remain deterministic
    265 
    -  new_thread <- withLocalTmpFSMake env $ \lcl_env ->
    266 
    -    fork_thread $ \unmask -> (do
    267 
    -            mres <- (unmask $ run_pipeline lcl_env act)
    268 
    -                      `MC.onException` (putMVar res_var Nothing) -- Defensive: If there's an unhandled exception then still signal the failure.
    269 
    -            putMVar res_var mres)
    270 
    -  threads <- runLoop fork_thread env acts
    271 
    -  return (new_thread : threads)
    272 
    -  where
    273 
    -      run_pipeline :: MakeEnv -> RunMakeM a -> IO (Maybe a)
    274 
    -      run_pipeline env p = runMaybeT (runReaderT p env)
    275 
    -
    276 
    -type RunMakeM a = ReaderT MakeEnv (MaybeT IO) a
    277 
    -
    278 
    -data MakeAction = forall a . MakeAction !(RunMakeM a) !(MVar (Maybe a))
    279 
    -
    280 
    -waitMakeAction :: MakeAction -> IO ()
    281 
    -waitMakeAction (MakeAction _ mvar) = () <$ readMVar mvar

    • —
    View it on GitLab.
    You're receiving this email because of your account on gitlab.haskell.org. Manage all notifications · Help Notification message regarding https://gitlab.haskell.org/ghc/ghc/-/compare/e53ad1ffb0bc4268d7cdb10dc7cb4f1156a8508e...17298acd31bf34b73d540c3b1662bc21c01638c9 at 1786639973