sheaf pushed to branch wip/romes/27514 at Glasgow Haskell Compiler / GHC Commits: 566d789d by sheaf at 2026-08-14T23:26:51+02:00 Concurrent worker abstraction for downsweep/make This commit introduces GHC.Driver.Concurrency which provides a high-level interface over scheduling concurrent workers, used by downsweep and --make. Summary of changes: - The ad-hoc plumbing of AbstractSem is replaced by the dedicated 'data Concurrency = Serial | Concurrent ConcurrencyEnv'. This avoids a footgun in which one could try to use an 'AbstractSem' as a lock in the serial case. - We no longer wastefully re-run an entire action upon a semaphore failure. The "fallback to -j1" logic is preserved, but it only applies to the initial attempt at opening a semaphore, not on late semaphore errors that occur partway through a lengthy computation. - Logger threads are now properly cleaned up on exception. - Drop the 'GhcMessage -> AnyGhcDiagnostic' and 'Maybe Messager' arguments to 'depanalE', 'depanalPartial' and 'downsweep', which were all dead in practice. - - - - - 9 changed files: - + compiler/GHC/Driver/Concurrency.hs - + compiler/GHC/Driver/Config/Concurrency.hs - compiler/GHC/Driver/Downsweep.hs - compiler/GHC/Driver/Make.hs - compiler/GHC/Driver/MakeAction.hs - compiler/GHC/Driver/MakeSem.hs - compiler/GHC/Utils/TmpFs.hs - compiler/ghc.cabal.in - utils/haddock/haddock-api/src/Haddock/Interface.hs Changes: ===================================== compiler/GHC/Driver/Concurrency.hs ===================================== @@ -0,0 +1,464 @@ +{-# LANGUAGE CPP #-} + +{-# LANGUAGE BlockArguments #-} + +module GHC.Driver.Concurrency + ( -- * Worker limit and concurrency + WorkerLimit(..) + , isWorkerLimitSequential + , withWorkerLimit + , Concurrency + , withConcurrency + -- * Concurrent worker scheduling + , ConcurrentWorkerEnv(..) + , mapConcurrentWorkers + , concurrentTraversal_DF + ) + where + +import GHC.Prelude + +import GHC.Driver.MakeSem +import GHC.Driver.Pipeline.LogQueue + ( LogQueueQueue, finishLogQueue, initLogQueue, logThread + , newLogQueue, newLogQueueQueue, parLogAction ) +import GHC.Utils.Logger + ( Logger, makeThreadSafe, pushLogHook ) +import GHC.Utils.Panic + ( panic ) +import GHC.Utils.TmpFs + ( TmpFs, forkTmpFsFrom, mergeTmpFsInto, withLocalTmpFS ) + +import System.Semaphore + ( SemaphoreError, SemaphoreIdentifier ) + +#if defined(wasm32_HOST_ARCH) || defined(javascript_HOST_ARCH) +import Control.Concurrent + ( ThreadId, forkIOWithUnmask, killThread, myThreadId ) +import Control.Concurrent.MVar + ( MVar, newEmptyMVar, newMVar, putMVar, takeMVar ) +import GHC.Conc + ( labelThread ) +#else +import Control.Concurrent + ( ThreadId, forkIOWithUnmask, killThread, myThreadId + , newQSem, signalQSem, waitQSem, MVar, takeMVar, putMVar, newEmptyMVar ) +import Control.Monad + ( unless ) +import qualified Control.Monad.Catch as MC +import GHC.Conc + ( getNumCapabilities, getNumProcessors, labelThread, setNumCapabilities ) +#endif +import Control.Concurrent.STM + ( TVar, atomically, check, modifyTVar', newTVarIO, readTVar, writeTVar ) +import Control.Exception + ( AsyncException(ThreadKilled), SomeAsyncException, SomeException + , finally, fromException, mask, mask_, onException + , throwIO, try, uninterruptibleMask_ ) +import Control.Monad + ( replicateM ) +import Data.Foldable + ( for_ ) +import Data.IORef + ( IORef, atomicModifyIORef', modifyIORef', newIORef, readIORef, writeIORef ) +import qualified Data.Map as Map +import qualified Data.Sequence as Seq +import qualified Data.Set as Set + +-------------------------------------------------------------------------------- +-- * Worker Limit +-------------------------------------------------------------------------------- + +-- | A limit on the number of concurrent worker threads. +data WorkerLimit + -- | Fixed concurrent worker count limit @-jN@ + = NumProcessorsLimit Int + -- | The concurrent worker count is limited by a @-jsem@ semaphore + | JSemLimit + SemaphoreIdentifier + -- ^ Semaphore identifier (from the @semaphore-compat@ library) + deriving Eq + +isWorkerLimitSequential :: WorkerLimit -> Bool +isWorkerLimitSequential (NumProcessorsLimit x) = x <= 1 +isWorkerLimitSequential (JSemLimit {}) = False + +runWorkerLimit + :: (SemaphoreError -> IO ()) + -- ^ report failure when opening the @-jsem@ semaphore + -- (after which we fall back to running with a single job) + -> WorkerLimit -> (AbstractSem -> IO a) -> IO a +#if defined(wasm32_HOST_ARCH) || defined(javascript_HOST_ARCH) +runWorkerLimit _report_semaphore_failure _ action = do + lock <- newMVar () + action $ AbstractSem (takeMVar lock) (putMVar lock ()) +#else +runWorkerLimit report_semaphore_failure worker_limit action = case worker_limit of + NumProcessorsLimit n_jobs -> + runNjobsAbstractSem n_jobs action + JSemLimit sem_ident -> + runJSemAbstractSem sem_ident action >>= \case + Right a -> return a + Left err -> do + report_semaphore_failure err + runNjobsAbstractSem 1 action +#endif + +#if !(defined(wasm32_HOST_ARCH) || defined(javascript_HOST_ARCH)) +runNjobsAbstractSem :: Int -> (AbstractSem -> IO a) -> IO a +runNjobsAbstractSem n_jobs action = do + compile_sem <- newQSem n_jobs + n_capabilities <- getNumCapabilities + n_cpus <- getNumProcessors + let + asem = AbstractSem (waitQSem compile_sem) (signalQSem compile_sem) + set_num_caps n = unless (n_capabilities /= 1) $ setNumCapabilities n + updNumCapabilities = do + -- Setting number of capabilities more than + -- CPU count usually leads to high userspace + -- lock contention. #9221 + set_num_caps $ min n_jobs n_cpus + resetNumCapabilities = set_num_caps n_capabilities + MC.bracket_ updNumCapabilities resetNumCapabilities $ action asem + +#endif + +-------------------------------------------------------------------------------- +-- * Workers +-------------------------------------------------------------------------------- + +data Concurrency + = Serial + | Concurrent !ConcurrencyEnv + +-- | Run an action with the given concurrency control (serial or concurrent). +withConcurrency :: Concurrency -> IO a -> IO a +withConcurrency conc act = + case conc of + Serial -> act + Concurrent ( ConcurrencyEnv { ce_semaphore } ) -> + withAbstractSem ce_semaphore act + +data ConcurrencyEnv = + ConcurrencyEnv + { ce_semaphore :: !AbstractSem + , ce_log_queue_queue :: !( TVar LogQueueQueue ) + , ce_next_log_queue_id :: !( IORef Int ) + } + +-- | The local environment of a worker thread that may be scheduled concurrently. +data ConcurrentWorkerEnv = ConcurrentWorkerEnv + { cwe_logger :: !Logger + , cwe_tmpfs :: !TmpFs + } + +-- | Run an action with a local 'TmpFs', merging in the resulting temporary file +-- accumulator into the parent afterwards. +workerEnv_withLocalTmpFS :: ConcurrentWorkerEnv -> (ConcurrentWorkerEnv -> IO a) -> IO a +workerEnv_withLocalTmpFS env use = + withLocalTmpFS (cwe_tmpfs env) \ lcl_tmpfs -> + use env { cwe_tmpfs = lcl_tmpfs } + +-- | Run an action either serially or concurrently based on the provided +-- 'WorkerLimit'. +withWorkerLimit + :: Logger + -> TmpFs + -> (SemaphoreError -> IO ()) + -- ^ report a failure to open the @-jsem@ semaphore + -- (after which we fall back to running with a single job) + -> WorkerLimit + -> (Concurrency -> ConcurrentWorkerEnv -> IO a) -- ^ action to run + -> IO a +withWorkerLimit logger tmpfs report_semaphore_failure limit action + | isWorkerLimitSequential limit + = action Serial $ + ConcurrentWorkerEnv + { cwe_logger = logger + , cwe_tmpfs = tmpfs + } + | otherwise + = do + safe_logger <- makeThreadSafe logger + lqq_var <- newTVarIO newLogQueueQueue + stopped_var <- newTVarIO False + wait_log_thread <- logThread safe_logger stopped_var lqq_var + next_logq_var <- newIORef 1 + + let + stop_logging :: IO () + stop_logging = do + atomically $ writeTVar stopped_var True + wait_log_thread + + parent_work_env :: ConcurrentWorkerEnv + parent_work_env = + ConcurrentWorkerEnv + { cwe_logger = safe_logger + , cwe_tmpfs = tmpfs + } + + ( `finally` stop_logging ) $ + runWorkerLimit report_semaphore_failure limit \ sem -> do + let + conc = + Concurrent $ + ConcurrencyEnv + { ce_semaphore = sem + , ce_log_queue_queue = lqq_var + , ce_next_log_queue_id = next_logq_var + } + action conc parent_work_env + +-------------------------------------------------------------------------------- +-- * Scheduling concurrent workers +-------------------------------------------------------------------------------- + +-- | Internal scheduler abstraction with two capabilities: +-- +-- - spawn a new worker thread +-- - wait for a worker thread to complete +data Scheduler r = Scheduler + { spawnWorker :: (ConcurrentWorkerEnv -> IO r) -> IO () + -- ^ Spawn one concurrent worker. + -- + -- The worker does not hold a token of the concurrency semaphore: the + -- worker action should use 'withConcurrency' around the work whose + -- concurrency should be limited. + , awaitWorker :: IO (Either SomeException r) + -- ^ Wait for one worker to complete. + -- + -- Will crash if there are no outstanding workers. + } + +-- | Internal implementation of a concurrent worker scheduler. +-- +-- Usage of this function requires the following: +-- +-- - all spawn/await actions are performed by a single thread, +-- - we never wait for more workers than were spawned, +-- - no worker outlives 'run_schedule'. +run_schedule + :: forall r a + . String + -- ^ thread label for workers + -> Concurrency + -> ConcurrentWorkerEnv + -> (Scheduler r -> IO a) + -- ^ worker action + -> IO a +run_schedule worker_label conc parent_work_env withScheduler = + case conc of + + Serial -> do + results_var <- newIORef Seq.empty + let + spawnWorker :: (ConcurrentWorkerEnv -> IO r) -> IO () + spawnWorker action = do + res <- try @SomeException $ + workerEnv_withLocalTmpFS parent_work_env action + case res of + Left e + | Just _ <- fromException @SomeAsyncException e + -> throwIO e + _ -> modifyIORef' results_var (Seq.|> res) + + awaitWorker :: IO (Either SomeException r) + awaitWorker = + readIORef results_var >>= \case + res Seq.:<| rest -> do + writeIORef results_var rest + pure res + Seq.Empty -> + panic "run_schedule: no outstanding job" + + withScheduler $ Scheduler { spawnWorker, awaitWorker } + + Concurrent ( ConcurrencyEnv { ce_next_log_queue_id, ce_log_queue_queue } ) -> do + worker_tids_var <- newTVarIO $ Set.empty @ThreadId + all_results_vars_var <- newIORef $ Seq.empty @(MVar (Either SomeException r)) + + let + wait_for_workers :: IO () + wait_for_workers = + atomically $ + check . Set.null =<< readTVar worker_tids_var + + cancel_workers :: IO () + cancel_workers = do + uninterruptibleMask_ do + tids <- atomically $ readTVar worker_tids_var + for_ tids killThread + wait_for_workers + + awaitWorker :: IO (Either SomeException r) + awaitWorker = + readIORef all_results_vars_var >>= \case + first_worker_res_var Seq.:<| rest -> do + writeIORef all_results_vars_var rest + -- block on the earliest-spawned outstanding worker + takeMVar first_worker_res_var + Seq.Empty -> + panic "run_schedule: no outstanding job" + + spawnWorker :: (ConcurrentWorkerEnv -> IO r) -> IO () + spawnWorker action = mask_ do + + worker_res_var <- newEmptyMVar + + -- TmpFs + lcl_tmpfs <- forkTmpFsFrom (cwe_tmpfs parent_work_env) + + -- LogQueue + lq <- do + job_id <- atomicModifyIORef' ce_next_log_queue_id \n -> (n + 1, n) + lq <- newLogQueue job_id + atomically $ initLogQueue ce_log_queue_queue lq + pure lq + + let + + worker_work_env :: ConcurrentWorkerEnv + worker_work_env = + parent_work_env + { cwe_tmpfs = lcl_tmpfs + , cwe_logger = pushLogHook (const (parLogAction lq)) + (cwe_logger parent_work_env) + } + + -- Run a worker action and record its result. + run_worker_and_record :: IO r -> IO () + run_worker_and_record worker_action = do + res <- try @SomeException worker_action + case res of + Left e + -- Worker is being cancelled: don't record anything. + | Just ThreadKilled <- fromException e + -> pure () + _ -> putMVar worker_res_var res + + -- Record that a worker thread is done. + mark_worker_done :: ThreadId -> IO () + mark_worker_done tid = + uninterruptibleMask_ do + -- Uninterruptible: the deletion below /must/ occur. + -- An uninterruptible mask is OK as we only ever block for (GAP) below. + mergeTmpFsInto lcl_tmpfs $ cwe_tmpfs parent_work_env + finishLogQueue lq + atomically do + tids <- readTVar worker_tids_var + check $ tid `Set.member` tids + -- Ensure we never end up with a dead ThreadId in 'worker_tids_var' + -- (if the worker thread finishes before the parent thread has + -- the time to add its ThreadId to 'worker_tids_var'). + + writeTVar worker_tids_var $ Set.delete tid tids + + run_worker :: (forall b. IO b -> IO b) -> IO () + run_worker unmask = do + tid <- myThreadId + labelThread tid worker_label + let + worker_action :: IO r + worker_action = unmask $ action worker_work_env + + run_worker_and_record worker_action `finally` + mark_worker_done tid + + worker_tid <- + forkIOWithUnmask run_worker + `onException` finishLogQueue lq + -- Very short (GAP) between forking the thread and recording its ThreadId. + atomically $ modifyTVar' worker_tids_var $ Set.insert worker_tid + modifyIORef' all_results_vars_var (Seq.|> worker_res_var) + + mask \ restore -> do + result <- restore (withScheduler $ Scheduler { spawnWorker, awaitWorker }) + `onException` cancel_workers + restore wait_for_workers `onException` cancel_workers + pure result + +-------------------------------------------------------------------------------- +-- * Derived scheduling functionality +-------------------------------------------------------------------------------- + +-- | Map a worker action over the input list with the given concurrency control. +-- +-- Workers run to completion (no early abort); the first exception +-- (in input order) is rethrown at the end. +mapConcurrentWorkers + :: String -- ^ thread label for workers + -> Concurrency + -> ConcurrentWorkerEnv + -> (ConcurrentWorkerEnv -> a -> IO b) + -- ^ individual worker action + -- + -- NB: workers do not hold semaphore tokens by default; use + -- 'withConcurrency' to acquire one + -> [a] + -> IO [b] +mapConcurrentWorkers worker_label conc work_env f xs = + run_schedule worker_label conc work_env \ scheduler -> do + for_ xs \ x -> spawnWorker scheduler \ worker_env -> f worker_env x + results <- replicateM (length xs) (awaitWorker scheduler) + either throwIO pure (sequence results) + +-- | Depth-first traversal with on-the-fly expansion of nodes. +-- +-- Each expansion step is handled by a worker thread under the given +-- concurrency control. +-- +-- Deterministic: expansions are consumed in the order the nodes were +-- discovered, so the traversal is a function of the node graph alone. +-- +-- Fails fast: the first worker exception cancels the outstanding workers and +-- is rethrown. +concurrentTraversal_DF + :: forall k n r + . Ord k + => String -- ^ thread label for workers + -> Concurrency + -> ConcurrentWorkerEnv + -> Map.Map k r + -- ^ results known ahead of time (no expansion needed) + -> [n] + -- ^ root nodes + -> (n -> k) + -- ^ node key from node + -> (ConcurrentWorkerEnv -> n -> IO (r, [n])) + -- ^ worker action: expand a node into its result and the children to visit next + -- + -- NB: workers do not hold semaphore tokens by default; use + -- 'withConcurrency' to acquire one + -> IO (Map.Map k r) +concurrentTraversal_DF worker_label conc work_env base_map roots key expand = + run_schedule worker_label conc work_env \ scheduler -> do + let + expand_node :: n -> ConcurrentWorkerEnv -> IO (k, (r, [n])) + expand_node node worker_env = do + res <- expand worker_env node + pure (key node, res) + + go + :: Map.Map k r -- expanded nodes and their results + -> Set.Set k -- nodes currently being expanded + -> [n] -- discovered nodes, to expand next + -> IO (Map.Map k r) + go !visited !pending (node : worklist) + | k `Set.member` pending || k `Map.member` visited + = go visited pending worklist + | otherwise + = do spawnWorker scheduler (expand_node node) + go visited (Set.insert k pending) worklist + where + k = key node + go visited pending [] + | Set.null pending + = pure visited + | otherwise + = awaitWorker scheduler >>= \case + Left e -> throwIO e + Right (k, (result, children)) -> + go (Map.insert k result visited) (Set.delete k pending) children + + go base_map Set.empty roots ===================================== compiler/GHC/Driver/Config/Concurrency.hs ===================================== @@ -0,0 +1,41 @@ +-- | Subsystem configuration for 'GHC.Driver.Concurrency'. +module GHC.Driver.Config.Concurrency + ( mkWorkerLimit + , semaphoreOpenFailureHandler + ) where + +import GHC.Prelude + +import GHC.Driver.Concurrency +import GHC.Driver.Config.Diagnostic ( initDiagOpts, initPrintConfig ) +import GHC.Driver.DynFlags +import GHC.Driver.Errors ( printOrThrowDiagnostics ) +import GHC.Driver.Errors.Types + +import GHC.Types.Error ( singleMessage ) +import GHC.Types.SrcLoc ( noSrcSpan ) +import GHC.Utils.Error ( mkPlainMsgEnvelope ) +import GHC.Utils.Logger ( Logger ) + +import GHC.Conc ( getNumProcessors ) +import System.Semaphore ( SemaphoreError ) + +-------------------------------------------------------------------------------- + +-- | Compute the 'WorkerLimit' from the @-j@\/@-jsem@ flags. +mkWorkerLimit :: DynFlags -> IO WorkerLimit +mkWorkerLimit dflags = + case parMakeCount dflags of + Nothing -> pure $ num_procs 1 + Just (ParMakeSemaphore h) -> pure (JSemLimit h) + Just ParMakeNumProcessors -> num_procs <$> getNumProcessors + Just (ParMakeThisMany n) -> pure $ num_procs n + where + num_procs x = NumProcessorsLimit (max 1 x) + +-- | Report that the semaphore specified using the @-jsem@ flag could not be opened. +semaphoreOpenFailureHandler :: Logger -> DynFlags -> SemaphoreError -> IO () +semaphoreOpenFailureHandler logger dflags err = do + let diag = DriverSemaphoreOpenFailure (checkBuildingCabalPackage dflags) err + msg = singleMessage $ mkPlainMsgEnvelope (initDiagOpts dflags) noSrcSpan diag + printOrThrowDiagnostics logger (initPrintConfig dflags) (initDiagOpts dflags) (GhcDriverMessage <$> msg) ===================================== compiler/GHC/Driver/Downsweep.hs ===================================== @@ -40,9 +40,9 @@ import GHC.Driver.Monad import GHC.Driver.Env import GHC.Driver.Errors import GHC.Driver.Errors.Types -import GHC.Driver.Messager -import GHC.Driver.MakeSem +import GHC.Driver.Concurrency import GHC.Driver.MakeAction +import GHC.Driver.Config.Concurrency import GHC.Driver.Config.Diagnostic import GHC.Driver.Ppr @@ -64,7 +64,6 @@ import GHC.Data.OsPath ( OsPath, unsafeEncodeUtf ) import GHC.Data.StringBuffer import GHC.Data.Graph.Directed.Reachability -import GHC.Utils.Exception ( throwIO, SomeAsyncException, AsyncException (..) ) import GHC.Utils.Outputable import GHC.Utils.Panic import GHC.Utils.Misc @@ -101,7 +100,6 @@ import qualified Data.Set as Set import Control.Concurrent.MVar import Control.Monad import Control.Monad.Trans.Except ( ExceptT(..), runExceptT, throwE ) -import qualified Control.Monad.Catch as MC import Data.Maybe import Data.List (partition) import Data.Time @@ -112,13 +110,8 @@ import System.FilePath import Control.Monad.Trans.Reader import qualified Data.Map.Strict as M -import Control.Monad.Trans.Class import System.IO.Unsafe (unsafeInterleaveIO) import qualified Data.List.NonEmpty as NE -import Control.Concurrent -import Control.Concurrent.STM.TQueue -import Control.Concurrent.STM -import Control.Applicative {- Note [The ModuleGraph] @@ -245,8 +238,6 @@ See Note [The ModuleGraph] for an overview when we do downsweep. -- -- See also Note [The ModuleGraph] downsweep :: HscEnv - -> (GhcMessage -> AnyGhcDiagnostic) - -> Maybe Messager -> [ModSummary] -- ^ Old summaries -> Maybe ModuleGraph @@ -260,13 +251,14 @@ downsweep :: HscEnv -- The non-error elements of the returned list all have distinct -- (Modules, IsBoot) identifiers, unless the Bool is true in -- which case there can be repeats -downsweep hsc_env diag_wrapper msg old_summaries maybe_base_graph excl_mods allow_dup_roots = do +downsweep hsc_env old_summaries maybe_base_graph excl_mods allow_dup_roots = do n_jobs <- mkWorkerLimit (hsc_dflags hsc_env) summ_cache <- newMVar (mkModSummaryCache (zip old_summaries (repeat SummOld))) imps_cache <- newMVar Map.empty - withMakeEnv n_jobs hsc_env diag_wrapper msg $ \make_env -> do - (root_errs, root_summaries) <- rootSummariesParallel n_jobs make_env (hsc_targets hsc_env) - (getRootSummary excl_mods summ_cache imps_cache) + withWorkerLimitHsc hsc_env n_jobs $ \conc hsc_env' -> do + (root_errs, root_summaries) <- + rootSummariesParallel conc hsc_env' (hsc_targets hsc_env) + (getRootSummary excl_mods summ_cache imps_cache) let closure_errs = checkHomeUnitsClosed unit_env unit_env = hsc_unit_env hsc_env @@ -275,13 +267,12 @@ downsweep hsc_env diag_wrapper msg old_summaries maybe_base_graph excl_mods allo case all_errs of [] -> do let env = DownsweepEnv - { ds_hsc_env = hsc_env + { ds_hsc_env = hsc_env' , ds_summaries_cache = summ_cache , ds_imports_cache = imps_cache , ds_mode = DownsweepUseCompile , ds_excl_mods = excl_mods - , ds_n_jobs = n_jobs - , ds_make_env = make_env + , ds_concurrency = conc } (downsweep_errs, downsweep_nodes) <- runDownsweepM env $ downsweepFromRootNodes maybe_base_graph allow_dup_roots @@ -349,15 +340,14 @@ downsweepThunk hsc_env mod_summary = unsafeInterleaveIO $ do njobs <- mkWorkerLimit (hsc_dflags hsc_env) summs <- newMVar (mkModSummaryCache [(mod_summary,SummOld)]) imps <- newMVar mempty - withMakeEnv njobs hsc_env mkUnknownDiagnostic Nothing $ \make_env -> do + withWorkerLimitHsc hsc_env njobs $ \conc hsc_env' -> do let env = DownsweepEnv - { ds_hsc_env = hsc_env + { ds_hsc_env = hsc_env' , ds_summaries_cache = summs , ds_imports_cache = imps , ds_mode = DownsweepUseFixed , ds_excl_mods = [] - , ds_n_jobs = njobs - , ds_make_env = make_env + , ds_concurrency = conc } ~(errs, mg) <- runDownsweepM env $ downsweepFromRootNodes Nothing True @@ -394,15 +384,14 @@ downsweepInteractiveImports hsc_env ic = unsafeInterleaveIO $ do n_jobs <- mkWorkerLimit (hsc_dflags hsc_env) summ_cache <- newMVar mempty imps_cache <- newMVar mempty - withMakeEnv n_jobs hsc_env mkUnknownDiagnostic Nothing $ \make_env -> do + withWorkerLimitHsc hsc_env n_jobs $ \conc hsc_env' -> do let env = DownsweepEnv - { ds_hsc_env = hsc_env + { ds_hsc_env = hsc_env' , ds_mode = DownsweepUseFixed{-or DownsweepUseCompile?-} , ds_summaries_cache = summ_cache , ds_imports_cache = imps_cache , ds_excl_mods = [] - , ds_n_jobs = n_jobs - , ds_make_env = make_env + , ds_concurrency = conc } graph <- runDownsweepM env do loopFromInteractive cached_nodes interactive_mn imps @@ -439,15 +428,14 @@ downsweepInstalledModules hsc_env mods = do nodes <- mapM process installed_mods summs <- newMVar mempty imps <- newMVar mempty - withMakeEnv njobs hsc_env mkUnknownDiagnostic Nothing $ \make_env -> do + withWorkerLimitHsc hsc_env njobs $ \conc hsc_env' -> do let env = DownsweepEnv - { ds_hsc_env = hsc_env + { ds_hsc_env = hsc_env' , ds_summaries_cache = summs , ds_imports_cache = imps , ds_mode = DownsweepUseFixed , ds_excl_mods = [] - , ds_n_jobs = njobs - , ds_make_env = make_env + , ds_concurrency = conc } (errs, mg) <- runDownsweepM env $ downsweepFromRootNodes Nothing True nodes external_uids @@ -562,8 +550,8 @@ data DownsweepEnv = DownsweepEnv { , ds_summaries_cache :: ModSummaryCache , ds_imports_cache :: ImportsCache , ds_excl_mods :: [ModuleName] - , ds_n_jobs :: WorkerLimit - , ds_make_env :: MakeEnv + , ds_concurrency :: Concurrency + -- ^ The concurrency to use for downsweep } mkModSummaryCache :: [(ModSummary, SummProvenance)] -> ModSummaryCacheMap @@ -928,15 +916,28 @@ getRootSummary excl_mods summ_cache imports_cache hsc_env target rootLoc = mkGeneralSrcSpan (fsLit "<command line>") dflags = homeUnitEnv_dflags (ue_findHomeUnitEnv uid (hsc_unit_env hsc_env)) --- | Execute 'getRootSummary' for the 'Target's using the parallelism pipeline system. +-- | Execute 'getRootSummary' for the 'Target's in bundles, spawning one +-- worker per bundle. The number of bundles processed at once is limited by +-- the given 'Concurrency'. rootSummariesParallel - :: WorkerLimit -> MakeEnv -> [Target] + :: Concurrency -> HscEnv -> [Target] -> (HscEnv -> Target -> IO (Either DriverMessages ModSummary)) -> IO ([DriverMessages], [ModSummary]) -rootSummariesParallel n_jobs make_env targets get_summary = do - partitionEithers <$> mapConcDS n_jobs bundle_size make_env get_summary targets - where - bundle_size = 20 +rootSummariesParallel conc hsc_env targets get_summary = do + results <- + mapConcurrentWorkers "root_summary_worker" conc (viewHscWorkerEnv hsc_env) + ( \ work_env bundle -> + withConcurrency conc $ + mapM (get_summary (setHscWorkerEnv work_env hsc_env)) bundle ) + bundles + pure $ partitionEithers (concat results) + where + bundle_size = 20 + + bundles = mk_bundles targets + mk_bundles = unfoldr \case + [] -> Nothing + ts -> Just (splitAt bundle_size ts) -------------------------------------------------------------------------------- -- * Check/validate properties and error out @@ -1762,7 +1763,7 @@ data NodeRes v -- node. The result includes the previously visited nodes given in @base_map@, -- s.t. @parDfsBuild base_map [] _ _ == base_map@. -- --- The @expand@ function returns an 'NResult'. See the 'NResult' documentation +-- The @expand@ function returns a 'NodeRes'. See the 'NodeRes' documentation -- for more information about each result type. -- -- Example usage: @n@ is instanced to @DownsweepNode@, @k@ is @NodeKey@, and @v@ is @ModuleNodeEdge@. @@ -1785,90 +1786,25 @@ parDfsBuild :: forall k v n. Ord k -- ^ The result accumulates the payload of expanding the root nodes -- and all nodes transitively reachable from those roots. parDfsBuild base_map roots key expand = ReaderT $ \ds_env -> do - exc_var <- newTVarIO $ Nothing @MC.SomeException - visited_var <- newTVarIO $ fromMaybe Map.empty base_map - pending <- newTVarIO $ Set.empty @k - worklist <- newTQueueIO @n - threads <- newTVarIO [] - - coord_tid <- forkIO $ - coordinator ds_env exc_var visited_var worklist pending threads - `MC.catch` \case - (e::MC.SomeException) - -- exit cleanly when killed - | Just ThreadKilled <- fromException e -> return () - -- if the coordinator somehow else crashes, - -- signal the exc_var for the main thread to throw it - | otherwise -> atomically (modifyTVar' exc_var (<|> Just e)) - - atomically $ mapM_ (writeTQueue worklist) roots - - mb_exc <- wait_done exc_var worklist pending - `MC.finally` do - killThread coord_tid - mapM_ killThread =<< readTVarIO threads - - case mb_exc of - Just e -> throwIO e - Nothing -> readTVarIO visited_var - where - wait_done exc_var worklist pending = - -- this txn retries until all work is done or an exception is signaled - atomically $ do - readTVar exc_var >>= \case - Just e -> return (Just e) - Nothing -> do - empty_worklist <- isEmptyTQueue worklist - empty_pending <- Set.null <$> readTVar pending - check (empty_worklist && empty_pending) - return Nothing - - coordinator ds_env exc_var visvar worklist pendvar threads = forever $ do - mb_node_to_expand <- atomically $ do - node <- readTQueue worklist - let k = key node - - visited <- readTVar visvar - pending <- readTVar pendvar - - if (k `Set.member` pending || k `Map.member` visited) - then return Nothing - else do - -- must add to pending in the same transaction as worklist dequeue, - -- otherwise the main thread may find both the worklist and pending - -- lists empty and exit prematurely. - modifyTVar' pendvar (Set.insert k) - return (Just (k, node)) - - case mb_node_to_expand of - Nothing -> return () - Just (k, node) -> do - tid <- MC.mask_ $ forkIOWithUnmask $ \unmask -> - unmask (withLocalTmpFSMake (ds_make_env ds_env) $ \make_env -> - worker ds_env{ds_make_env = make_env} visvar worklist pendvar k node) - `MC.catch` \case - e | Just (_ :: SomeAsyncException) <- fromException e - -> throwIO e -- async exceptions like KillThread get thrown - | otherwise -- exceptions in workers are written for main thread - -> atomically (modifyTVar' exc_var (<|> Just e)) - - atomically $ modifyTVar' threads (tid:) - - worker ds_env@DownsweepEnv{..} visvar worklist pendvar k node = - withAbstractSem (compile_sem ds_make_env) $ do - r <- runDownsweepM ds_env $ - expand node -- do the main work! - - atomically $ do - case r of - NSkip -> - modifyTVar' visvar (Map.insert k NSkip) - NSuccess (v,ns) -> do - modifyTVar' visvar (Map.insert k (NSuccess v)) - mapM_ (writeTQueue worklist) ns - - modifyTVar' pendvar (Set.delete k) + let + conc :: Concurrency + conc = ds_concurrency ds_env + + expand_node :: ConcurrentWorkerEnv -> n -> IO (NodeRes v, [n]) + expand_node worker_env node = do + result <- withConcurrency conc $ + runDownsweepM (setDownsweepWorkerEnv worker_env ds_env) (expand node) + pure $ case result of + NSkip -> (NSkip, []) + NSuccess (val, new_work) -> (NSuccess val, new_work) + + concurrentTraversal_DF "downsweep_worker" conc (viewHscWorkerEnv $ ds_hsc_env ds_env) + (fromMaybe mempty base_map) roots key expand_node + +setDownsweepWorkerEnv :: ConcurrentWorkerEnv -> DownsweepEnv -> DownsweepEnv +setDownsweepWorkerEnv work_env env = + env { ds_hsc_env = setHscWorkerEnv work_env (ds_hsc_env env) } {- Note [Downsweep Control Flow and Caching] @@ -1963,70 +1899,10 @@ things, and that processing can often be costly (e.g. see `expandModuleSummary`) We leverage multiple threads in this traversal to expand more than one module at once, respecting -j<N> to mean we never expand more than N modules at once. -The parallel downsweep is all handled by `parDfsBuild` as follows: - -- We launch a thread for every module we discover that needs to be - expanded in the `coordinator` thread, popping it from the worklist -- Every launched `worker` thread blocks waiting for a semaphore token - (`withAbstractSem`) to respect -j<N> -- The main thread waits until both the worklist and pending list is - cleared, atomically. - -STM is used crucially to guarantee e.g. we don't have race conditions -between taking from the worklist and writing to the pending list while -checking whether they are clear. - -Exceptions are bubbled up to the main thread. The "main" thread, which is -typically waiting for the worklist+pending lists to be clear, instead gets -unblocked by this exception (signaled in `exc_var`) and re-throws it. --} --------------------------------------------------------------------------------- --- * Concurrent utilities --------------------------------------------------------------------------------- - --- | Map an action over a list using the parallelism pipeline system. --- Create bundles of the list elems wrapped in a 'MakeAction' that uses --- 'withAbstractSem' to wait for a free slot, limiting the number of --- concurrently computed summaries to the value of the @-j@ option or the slots --- allocated by the job server, if that is used. --- --- The 'MakeAction' returns 'Maybe', which is not handled as an error, because --- 'runLoop' only sets it to 'Nothing' when an exception was thrown, so the --- result won't be read anyway here. --- --- To emulate the current behavior, we funnel exceptions past the concurrency --- barrier and rethrow the first one afterwards. -mapConcDS :: - WorkerLimit -> - Int {-^ Batch size -} -> - MakeEnv -> - (HscEnv -> a -> IO b) -> - [a] -> - IO ([b]) -mapConcDS n_jobs bundle_size make_env run_action xs = do - (actions, get_results) <- unzip <$> mapM action_and_result (zip [1..] bundles) - runAllPipelines n_jobs make_env actions - (sequence . catMaybes <$> sequence get_results) >>= \case - Right results -> pure (concat results) - Left exc -> throwIO exc - where - bundles = mk_bundles xs - - mk_bundles = unfoldr \case - [] -> Nothing - ts -> Just (splitAt bundle_size ts) - - action_and_result (log_queue_id, ts) = do - res_var <- liftIO newEmptyMVar - pure $! (MakeAction (action log_queue_id ts) res_var, readMVar res_var) - - action log_queue_id target_bundle = do - env@MakeEnv {compile_sem} <- ask - lift $ lift $ - withAbstractSem compile_sem $ - withLoggerHsc log_queue_id env \ lcl_hsc_env -> - MC.try (mapM (run_action lcl_hsc_env) target_bundle) >>= \case - Left e | Just (_ :: SomeAsyncException) <- fromException e -> - throwIO e - a -> pure a +We use the concurrent scheduling abstraction from GHC.Driver.Concurrency +('concurrentTraversal_DF'). Each time we discover a new node, a worker is +spawned to expand it. After each expansion completes, the resulting children +nodes are pushed onto the worklist. With -j1 no threads are involved: each +expansion runs in sequence. +-} ===================================== compiler/GHC/Driver/Make.hs ===================================== @@ -63,8 +63,9 @@ import GHC.Driver.Env import GHC.Driver.Errors import GHC.Driver.Errors.Types import GHC.Driver.Main -import GHC.Driver.MakeSem import GHC.Driver.Downsweep +import GHC.Driver.Concurrency +import GHC.Driver.Config.Concurrency import GHC.Driver.MakeAction import GHC.Types.UnresolvedImport @@ -156,22 +157,20 @@ depanal :: GhcMonad m => depanal excluded_mods allow_dup_roots = do hsc_env <- getSession let sec = initSourceErrorContext (hsc_dflags hsc_env) - (errs, mod_graph) <- depanalE mkUnknownDiagnostic Nothing excluded_mods allow_dup_roots + (errs, mod_graph) <- depanalE excluded_mods allow_dup_roots if isEmptyMessages errs then pure mod_graph else throwErrors sec (fmap GhcDriverMessage errs) -- | Perform dependency analysis like in 'depanal'. -- In case of errors, the errors and an empty module graph are returned. -depanalE :: GhcMonad m => -- New for #17459 - (GhcMessage -> AnyGhcDiagnostic) - -> Maybe Messager - -> [ModuleName] -- ^ excluded modules +depanalE :: GhcMonad m => + [ModuleName] -- ^ excluded modules -> Bool -- ^ allow duplicate roots -> m (DriverMessages, ModuleGraph) -depanalE diag_wrapper msg excluded_mods allow_dup_roots = do +depanalE excluded_mods allow_dup_roots = do hsc_env <- getSession - (errs, mod_graph) <- depanalPartial diag_wrapper msg excluded_mods allow_dup_roots + (errs, mod_graph) <- depanalPartial excluded_mods allow_dup_roots if isEmptyMessages errs then do hsc_env <- getSession @@ -209,13 +208,11 @@ depanalE diag_wrapper msg excluded_mods allow_dup_roots = do -- new module graph. depanalPartial :: GhcMonad m - => (GhcMessage -> AnyGhcDiagnostic) - -> Maybe Messager - -> [ModuleName] -- ^ excluded modules + => [ModuleName] -- ^ excluded modules -> Bool -- ^ allow duplicate roots -> m (DriverMessages, ModuleGraph) -- ^ possibly empty 'Bag' of errors and a module graph. -depanalPartial diag_wrapper msg excluded_mods allow_dup_roots = do +depanalPartial excluded_mods allow_dup_roots = do hsc_env <- getSession let targets = hsc_targets hsc_env @@ -234,7 +231,7 @@ depanalPartial diag_wrapper msg excluded_mods allow_dup_roots = do liftIO $ flushFinderCaches (hsc_FC hsc_env) (hsc_unit_env hsc_env) (errs, mod_graph) <- liftIO $ downsweep - hsc_env diag_wrapper msg (mgModSummaries old_graph) Nothing + hsc_env (mgModSummaries old_graph) Nothing excluded_mods allow_dup_roots return (unionManyMessages errs, mod_graph) @@ -438,7 +435,7 @@ loadWithCache :: GhcMonad m => Maybe ModIfaceCache -- ^ Instructions about how t -> m SuccessFlag loadWithCache cache diag_wrapper how_much = do msg <- mkBatchMsg <$> getSession - (errs, mod_graph) <- depanalE diag_wrapper (Just msg) [] False -- #17459 + (errs, mod_graph) <- depanalE [] False -- #17459 success <- load' cache how_much diag_wrapper (Just msg) mod_graph hsc_env <- getSession let sec = initSourceErrorContext (hsc_dflags hsc_env) @@ -840,14 +837,15 @@ The Algorithm a pair of an `IO a` action and a `MVar a`, where to place the result. The list is sorted topologically, so can be executed in order without fear of blocking. -* runPipelines takes this list and eventually passes it to runLoop which executes - each action and places the result into the right MVar. -* The amount of parallelism is controlled by a semaphore. This is just used around the - module compilation step, so that only the right number of modules are compiled at - the same time which reduces overall memory usage and allocations. -* Each proper node has a LogQueue, which dictates where to send it's output. -* The LogQueue is placed into the LogQueueQueue when the action starts and a worker - thread processes the LogQueueQueue printing logs for each module in a stable order. +* runPipelines spawns one worker per action ('GHC.Driver.Concurrency.mapConcurrentWorkers'), + which executes the action and places the result into the right MVar. +* The amount of parallelism is controlled by a semaphore ('withMakeEnvConcurrency'). This is + just used around the module compilation step, so that only the right number of + modules are compiled at the same time which reduces overall memory usage and + allocations. +* Each worker has a LogQueue, which dictates where to send its output. A log + thread processes the LogQueues, printing logs for each module in a stable + order (the order in which the actions were spawned). * The result variable for an action producing `a` is of type `Maybe a`, therefore it is still filled on a failure. If a module fails to compile, the failure is propagated through the whole module graph and any modules which didn't @@ -1137,7 +1135,7 @@ interpretBuildPlan hug mhmi_cache old_hpt plan = do !build_deps = getDependencies (map gwib_mod deps) build_map let loop_action = withCurrentUnit loop_unit $ do !_ <- wait_deps build_deps - hsc_env <- asks hsc_env + hsc_env <- asks me_hsc_env let mns :: [ModuleName] mns = mapMaybe (nodeKeyModName . gwib_mod) deps @@ -1180,7 +1178,7 @@ interpretBuildPlan hug mhmi_cache old_hpt plan = do withCurrentUnit :: UnitId -> RunMakeM a -> RunMakeM a withCurrentUnit uid = do - local (\env -> env { hsc_env = hscSetActiveUnitId uid (hsc_env env)}) + local (\env -> env { me_hsc_env = hscSetActiveUnitId uid (me_hsc_env env)}) upsweep :: WorkerLimit -- ^ The number of workers we wish to run in parallel @@ -1556,10 +1554,11 @@ executeInstantiationNode k n deps uid iu = do env <- ask -- Output of the logger is mediated by a central worker to -- avoid output interleaving - msg <- asks env_messager - wrapper <- asks diag_wrapper - lift $ MaybeT $ withLoggerHsc k env $ \hsc_env -> - let lcl_hsc_env = setHUG deps hsc_env + msg <- asks me_messager + wrapper <- asks me_diag_wrapper + lift $ MaybeT $ + let hsc_env = me_hsc_env env + lcl_hsc_env = setHUG deps hsc_env in wrapAction wrapper lcl_hsc_env $ do res <- upsweep_inst lcl_hsc_env msg k n uid iu cleanCurrentModuleTempFilesMaybe (hsc_logger hsc_env) (hsc_tmpfs hsc_env) (hsc_dflags hsc_env) @@ -1582,13 +1581,13 @@ executeCompileNode :: Int -> ModuleNodeInfo -> RunMakeM HomeModInfo executeCompileNode k n !old_hmi hug mrehydrate_mods mni = do - me@MakeEnv{..} <- ask + make_env <- ask -- Rehydrate any dependencies if this module had a boot file or is a signature file. - lift $ MaybeT (withAbstractSem compile_sem $ withLoggerHsc k me $ \hsc_env -> do + lift $ MaybeT (withMakeEnvConcurrency make_env $ \hsc_env -> do hsc_env' <- liftIO $ maybeRehydrateBefore (setHUG hug hsc_env) mni fixed_mrehydrate_mods case mni of - ModuleNodeCompile mod -> executeCompileNodeWithSource hsc_env' me mod - ModuleNodeFixed key loc -> executeCompileNodeFixed hsc_env' me key loc + ModuleNodeCompile mod -> executeCompileNodeWithSource hsc_env' make_env mod + ModuleNodeFixed key loc -> executeCompileNodeFixed hsc_env' make_env key loc ) where @@ -1601,9 +1600,9 @@ executeCompileNode k n !old_hmi hug mrehydrate_mods mni = do _ -> mrehydrate_mods executeCompileNodeFixed :: HscEnv -> MakeEnv -> ModNodeKeyWithUid -> ModLocation -> IO (Maybe HomeModInfo) - executeCompileNodeFixed hsc_env MakeEnv{diag_wrapper, env_messager} mod loc = - wrapAction diag_wrapper hsc_env $ do - forM_ env_messager $ \hscMessage -> hscMessage hsc_env (k, n) UpToDate (ModuleNode [] (ModuleNodeFixed mod loc)) + executeCompileNodeFixed hsc_env MakeEnv{me_diag_wrapper, me_messager} mod loc = + wrapAction me_diag_wrapper hsc_env $ do + forM_ me_messager $ \hscMessage -> hscMessage hsc_env (k, n) UpToDate (ModuleNode [] (ModuleNodeFixed mod loc)) read_result <- readIface (hsc_hooks hsc_env) (hsc_logger hsc_env) (hsc_dflags hsc_env) (hsc_NC hsc_env) (mnkToModule mod) (ml_hi_file loc) let sec = initSourceErrorContext (hsc_dflags hsc_env) case read_result of @@ -1619,7 +1618,7 @@ executeCompileNode k n !old_hmi hug mrehydrate_mods mni = do return (HomeModInfo iface details hm_linkable) executeCompileNodeWithSource :: HscEnv -> MakeEnv -> ModSummary -> IO (Maybe HomeModInfo) - executeCompileNodeWithSource hsc_env MakeEnv{diag_wrapper, env_messager} mod = do + executeCompileNodeWithSource hsc_env MakeEnv{me_diag_wrapper, me_messager} mod = do let -- Use the cached DynFlags which includes OPTIONS_GHC pragmas lcl_dynflags = ms_hspp_opts mod let lcl_hsc_env = @@ -1628,8 +1627,8 @@ executeCompileNode k n !old_hmi hug mrehydrate_mods mni = do hsc_env -- Compile the module, locking with a semaphore to avoid too many modules -- being compiled at the same time leading to high memory usage. - wrapAction diag_wrapper lcl_hsc_env $ do - res <- upsweep_mod lcl_hsc_env env_messager old_hmi mod k n + wrapAction me_diag_wrapper lcl_hsc_env $ do + res <- upsweep_mod lcl_hsc_env me_messager old_hmi mod k n cleanCurrentModuleTempFilesMaybe (hsc_logger hsc_env) (hsc_tmpfs hsc_env) lcl_dynflags return res @@ -1853,15 +1852,15 @@ Also closely related are -} executeLinkNode :: HomeUnitGraph -> (Int, Int) -> UnitId -> [NodeKey] -> RunMakeM () -executeLinkNode hug kn@(k, _) uid deps = do +executeLinkNode hug kn uid deps = do withCurrentUnit uid $ do make_env@MakeEnv{..} <- ask - let dflags = hsc_dflags hsc_env - msg' = (\messager -> \recomp -> messager hsc_env kn recomp (LinkNode deps uid)) <$> env_messager + let dflags = hsc_dflags me_hsc_env + msg' = (\messager -> \recomp -> messager me_hsc_env kn recomp (LinkNode deps uid)) <$> me_messager - linkresult <- lift $ MaybeT $ withAbstractSem compile_sem $ withLoggerHsc k make_env $ \lcl_hsc_env -> do + linkresult <- lift $ MaybeT $ withMakeEnvConcurrency make_env $ \lcl_hsc_env -> do let hsc_env' = setHUG hug lcl_hsc_env - wrapAction diag_wrapper hsc_env' $ do + wrapAction me_diag_wrapper hsc_env' $ do link (ghcLink dflags) hsc_env' True -- We already decided to link ===================================== compiler/GHC/Driver/MakeAction.hs ===================================== @@ -1,4 +1,3 @@ -{-# LANGUAGE CPP #-} module GHC.Driver.MakeAction ( MakeAction(..) , RunMakeM @@ -7,79 +6,41 @@ module GHC.Driver.MakeAction -- * Running the pipelines , runAllPipelines , runPipelines - -- * Worker limit - , WorkerLimit(..) - , mkWorkerLimit - , runWorkerLimit -- * Utility - , withLoggerHsc - , withParLog - , withLocalTmpFS - , withLocalTmpFSMake + , withMakeEnvConcurrency + , withWorkerLimitHsc + , viewHscWorkerEnv + , setHscWorkerEnv ) where import GHC.Prelude -import GHC.Driver.DynFlags -import GHC.Driver.Monad +import GHC.Driver.Concurrency +import GHC.Driver.Config.Concurrency import GHC.Driver.Env import GHC.Driver.Errors.Types import GHC.Driver.Messager -import GHC.Driver.MakeSem - -#if defined(wasm32_HOST_ARCH) || defined(javascript_HOST_ARCH) -import System.Semaphore - ( SemaphoreIdentifier ) -#else -import System.Semaphore - ( SemaphoreError, SemaphoreIdentifier ) -#endif - -#if !(defined(wasm32_HOST_ARCH) || defined(javascript_HOST_ARCH)) -import GHC.Driver.Config.Diagnostic ( initDiagOpts, initPrintConfig ) -import GHC.Driver.Errors ( printOrThrowDiagnostics ) -import GHC.Types.Error ( singleMessage ) -import GHC.Types.SrcLoc ( noSrcSpan ) -import GHC.Utils.Error ( mkPlainMsgEnvelope ) -#endif -import GHC.Utils.Logger -import GHC.Utils.TmpFs +import GHC.Driver.Monad -#if defined(wasm32_HOST_ARCH) || defined(javascript_HOST_ARCH) -import Control.Concurrent ( ThreadId, killThread, forkIOWithUnmask ) -#else -import Control.Concurrent ( newQSem, waitQSem, signalQSem, ThreadId, killThread, forkIOWithUnmask ) -#endif import qualified GHC.Conc as CC import Control.Concurrent.MVar import Control.Monad import qualified Control.Monad.Catch as MC - -#if defined(wasm32_HOST_ARCH) || defined(javascript_HOST_ARCH) -import GHC.Conc ( getNumProcessors ) -#else -import GHC.Conc ( getNumProcessors, getNumCapabilities, setNumCapabilities ) -#endif -import Control.Monad.Trans.Reader -import GHC.Driver.Pipeline.LogQueue -import Control.Concurrent.STM import Control.Monad.Trans.Maybe +import Control.Monad.Trans.Reader -------------------------------------------------------------------------------- -- * MakeEnv and MakeAction -------------------------------------------------------------------------------- -- | Environment used when compiling a module -data MakeEnv = MakeEnv { hsc_env :: !HscEnv -- The basic HscEnv which will be augmented for each module - , compile_sem :: !AbstractSem - -- Modify the environment for module k, with the supplied logger modification function. - -- For -j1, this wrapper doesn't do anything - -- For -jn, the wrapper initialised a log queue and then modifies the logger to pipe its output - -- into the log queue. - , withLogger :: forall a . Int -> ((Logger -> Logger) -> IO a) -> IO a - , env_messager :: !(Maybe Messager) - , diag_wrapper :: GhcMessage -> AnyGhcDiagnostic - } +data MakeEnv = + MakeEnv + { me_hsc_env :: !HscEnv -- The basic HscEnv which will be augmented for each module + , me_concurrency :: !Concurrency + , me_messager :: !(Maybe Messager) + , me_diag_wrapper :: GhcMessage -> AnyGhcDiagnostic + } -- | Come up with a 'MakeEnv' based on the given 'WorkerLimit'. -- For -j1, it will be a trivial 'MakeEnv' not prepared for parallelism. @@ -91,47 +52,14 @@ withMakeEnv -> Maybe Messager -- ^ Optional custom messager to use to report progress -> (MakeEnv -> IO r) -> IO r withMakeEnv worker_limit hsc_env diag_wrapper mHscMessager act = - if isWorkerLimitSequential worker_limit - then withSeqMakeEnv - else withParMakeEnv - where - withSeqMakeEnv = do - let seq_env = MakeEnv - { hsc_env = hsc_env - , withLogger = \_ k -> k id - , compile_sem = AbstractSem (return ()) (return ()) - , env_messager = mHscMessager - , diag_wrapper = diag_wrapper - } - act seq_env - - withParMakeEnv = do - -- A variable which we write to when an error has happened and we have to tell the - -- logging thread to gracefully shut down. - stopped_var <- newTVarIO False - -- The queue of LogQueues which actions are able to write to. When an action starts it - -- will add it's LogQueue into this queue. - log_queue_queue_var <- newTVarIO newLogQueueQueue - -- Thread which coordinates the printing of logs - wait_log_thread <- logThread (hsc_logger hsc_env) stopped_var log_queue_queue_var - - - -- Make the logger thread-safe, in case there is some output which isn't sent via the LogQueue. - thread_safe_logger <- liftIO $ makeThreadSafe (hsc_logger hsc_env) - let thread_safe_hsc_env = hsc_env { hsc_logger = thread_safe_logger } - - runWorkerLimit (hsc_logger hsc_env) (hsc_dflags hsc_env) worker_limit $ \abstract_sem -> do - let env = MakeEnv { hsc_env = thread_safe_hsc_env - , withLogger = withParLog log_queue_queue_var - , compile_sem = abstract_sem - , env_messager = mHscMessager - , diag_wrapper = diag_wrapper - } - -- Reset the number of capabilities once the upsweep ends. - r <- act env - atomically $ writeTVar stopped_var True - wait_log_thread - pure r + withWorkerLimitHsc hsc_env worker_limit $ \ conc hsc_env' -> + act $ + MakeEnv + { me_hsc_env = hsc_env' + , me_concurrency = conc + , me_messager = mHscMessager + , me_diag_wrapper = diag_wrapper + } -- ** MakeAction --------------------------------------------------------------- @@ -139,9 +67,6 @@ data MakeAction = forall a . MakeAction !(RunMakeM a) !(MVar (Maybe a)) type RunMakeM a = ReaderT MakeEnv (MaybeT IO) a -waitMakeAction :: MakeAction -> IO () -waitMakeAction (MakeAction _ mvar) = () <$ readMVar mvar - -------------------------------------------------------------------------------- -- * Running the pipelines -------------------------------------------------------------------------------- @@ -155,149 +80,51 @@ runPipelines runPipelines n_job hsc_env diag_wrapper mHscMessager all_pipelines = do liftIO $ label_self "main --make thread" withMakeEnv n_job hsc_env diag_wrapper mHscMessager $ \make_env -> do - runAllPipelines n_job make_env all_pipelines + runAllPipelines make_env all_pipelines where label_self :: String -> IO () label_self thread_name = do self_tid <- CC.myThreadId CC.labelThread self_tid thread_name --- | Run the given actions and then wait for them all to finish. -runAllPipelines :: WorkerLimit -> MakeEnv -> [MakeAction] -> IO () -runAllPipelines worker_limit env acts = do - let single_worker = isWorkerLimitSequential worker_limit - spawn_actions :: IO [ThreadId] - spawn_actions = if single_worker - then (:[]) <$> (forkIOWithUnmask $ \unmask -> void $ runLoop (\io -> io unmask) env acts) - else runLoop forkIOWithUnmask env acts - - kill_actions :: [ThreadId] -> IO () - kill_actions tids = mapM_ killThread tids - - MC.bracket spawn_actions kill_actions $ \_ -> do - mapM_ waitMakeAction acts - --- | Execute each action in order, limiting the amount of parallelism by the given --- semaphore. -runLoop :: (((forall a. IO a -> IO a) -> IO ()) -> IO a) -> MakeEnv -> [MakeAction] -> IO [a] -runLoop _ _env [] = return [] -runLoop fork_thread env (MakeAction act res_var :acts) = do - - -- withLocalTmpFs has to occur outside of fork to remain deterministic - new_thread <- withLocalTmpFSMake env $ \lcl_env -> - MC.mask_ $ - fork_thread $ \unmask -> (do - mres <- (unmask $ run_pipeline lcl_env act) - `MC.onException` (putMVar res_var Nothing) -- Defensive: If there's an unhandled exception then still signal the failure. - putMVar res_var mres) - threads <- runLoop fork_thread env acts - return (new_thread : threads) - where - run_pipeline :: MakeEnv -> RunMakeM a -> IO (Maybe a) - run_pipeline env p = runMaybeT (runReaderT p env) - --------------------------------------------------------------------------------- --- * Worker Limit --------------------------------------------------------------------------------- - --- | This describes what we use to limit the number of jobs, either we limit it --- ourselves to a specific number or we have an external parallelism semaphore --- limit it for us. -data WorkerLimit - = NumProcessorsLimit Int - | JSemLimit - SemaphoreIdentifier - -- ^ Semaphore identifier from @-jsem@ - deriving Eq - -mkWorkerLimit :: DynFlags -> IO WorkerLimit -mkWorkerLimit dflags = - case parMakeCount dflags of - Nothing -> pure $ num_procs 1 - Just (ParMakeSemaphore h) -> pure (JSemLimit h) - Just ParMakeNumProcessors -> num_procs <$> getNumProcessors - Just (ParMakeThisMany n) -> pure $ num_procs n - where - num_procs x = NumProcessorsLimit (max 1 x) - -isWorkerLimitSequential :: WorkerLimit -> Bool -isWorkerLimitSequential (NumProcessorsLimit x) = x <= 1 -isWorkerLimitSequential (JSemLimit {}) = False - -runWorkerLimit :: Logger -> DynFlags -> WorkerLimit -> (AbstractSem -> IO a) -> IO a -#if defined(wasm32_HOST_ARCH) || defined(javascript_HOST_ARCH) -runWorkerLimit _logger _dflags _ action = do - lock <- newMVar () - action $ AbstractSem (takeMVar lock) (putMVar lock ()) -#else -runWorkerLimit logger dflags worker_limit action = case worker_limit of - NumProcessorsLimit n_jobs -> - runNjobsAbstractSem n_jobs action - JSemLimit sem_ident -> do - result <- MC.try @_ @SemaphoreError $ runJSemAbstractSem sem_ident action - case result of - Right a -> return a - Left err -> do - let diag = DriverSemaphoreOpenFailure (checkBuildingCabalPackage dflags) err - msg = singleMessage $ mkPlainMsgEnvelope (initDiagOpts dflags) noSrcSpan diag - printOrThrowDiagnostics logger (initPrintConfig dflags) (initDiagOpts dflags) (GhcDriverMessage <$> msg) - runNjobsAbstractSem 1 action -#endif - -#if !(defined(wasm32_HOST_ARCH) || defined(javascript_HOST_ARCH)) -runNjobsAbstractSem :: Int -> (AbstractSem -> IO a) -> IO a -runNjobsAbstractSem n_jobs action = do - compile_sem <- newQSem n_jobs - n_capabilities <- getNumCapabilities - n_cpus <- getNumProcessors - let - asem = AbstractSem (waitQSem compile_sem) (signalQSem compile_sem) - set_num_caps n = unless (n_capabilities /= 1) $ setNumCapabilities n - updNumCapabilities = do - -- Setting number of capabilities more than - -- CPU count usually leads to high userspace - -- lock contention. #9221 - set_num_caps $ min n_jobs n_cpus - resetNumCapabilities = set_num_caps n_capabilities - MC.bracket_ updNumCapabilities resetNumCapabilities $ action asem - -#endif +-- | Run the given actions (assumed to be in dependency order) and wait for +-- them all to finish, rethrowing the first unhandled exception (in action order) +-- afterwards. +runAllPipelines :: MakeEnv -> [MakeAction] -> IO () +runAllPipelines env acts = + void $ + mapConcurrentWorkers "make_worker" (me_concurrency env) (viewHscWorkerEnv (me_hsc_env env)) + ( \ work_env (MakeAction act res_var) -> do + let lcl_env = env { me_hsc_env = setHscWorkerEnv work_env (me_hsc_env env) } + mres <- runMaybeT (runReaderT act lcl_env) + `MC.onException` putMVar res_var Nothing + putMVar res_var mres ) + acts -------------------------------------------------------------------------------- -- * Utility -------------------------------------------------------------------------------- -withLoggerHsc :: Int -> MakeEnv -> (HscEnv -> IO a) -> IO a -withLoggerHsc k MakeEnv{withLogger, hsc_env} cont = do - withLogger k $ \modifyLogger -> do - let lcl_logger = modifyLogger (hsc_logger hsc_env) - hsc_env' = hsc_env { hsc_logger = lcl_logger } - -- Run continuation with modified logger - cont hsc_env' - -withParLog :: TVar LogQueueQueue -> Int -> ((Logger -> Logger) -> IO b) -> IO b -withParLog lqq_var k cont = do - let init_log = do - -- Make a new log queue - lq <- newLogQueue k - -- Add it into the LogQueueQueue - atomically $ initLogQueue lqq_var lq - return lq - finish_log lq = liftIO (finishLogQueue lq) - MC.bracket init_log finish_log $ \lq -> cont (pushLogHook (const (parLogAction lq))) - -withLocalTmpFS :: TmpFs -> (TmpFs -> IO a) -> IO a -withLocalTmpFS tmpfs act = do - let initialiser = do - liftIO $ forkTmpFsFrom tmpfs - finaliser tmpfs_local = do - liftIO $ mergeTmpFsInto tmpfs_local tmpfs - -- Add remaining files which weren't cleaned up into local tmp fs for - -- clean-up later. - -- Clear the logQueue if this node had it's own log queue - MC.bracket initialiser finaliser act - -withLocalTmpFSMake :: MakeEnv -> (MakeEnv -> IO a) -> IO a -withLocalTmpFSMake env k = - withLocalTmpFS (hsc_tmpfs (hsc_env env)) $ \lcl_tmpfs - -> k (env { hsc_env = (hsc_env env) { hsc_tmpfs = lcl_tmpfs }}) +-- | A version of 'withWorkerLimit' taking an 'HscEnv'. +withWorkerLimitHsc :: HscEnv -> WorkerLimit -> (Concurrency -> HscEnv -> IO a) -> IO a +withWorkerLimitHsc hsc_env limit k = + withWorkerLimit (hsc_logger hsc_env) (hsc_tmpfs hsc_env) + (semaphoreOpenFailureHandler (hsc_logger hsc_env) (hsc_dflags hsc_env)) + limit + (\conc work_env -> k conc (setHscWorkerEnv work_env hsc_env)) + +-- | Like 'withConcurrency', but retrieving the 'Concurrency' and 'HscEnv' from +-- the 'MakeEnv'. +withMakeEnvConcurrency :: MakeEnv -> (HscEnv -> IO a) -> IO a +withMakeEnvConcurrency env cont = + withConcurrency (me_concurrency env) (cont (me_hsc_env env)) + +-- | The local environment for a concurrent worker derived from an 'HscEnv'. +viewHscWorkerEnv :: HscEnv -> ConcurrentWorkerEnv +viewHscWorkerEnv hsc_env = + ConcurrentWorkerEnv { cwe_logger = hsc_logger hsc_env, cwe_tmpfs = hsc_tmpfs hsc_env } + +-- | Set the local concurrent worker environment within an 'HscEnv'. +setHscWorkerEnv :: ConcurrentWorkerEnv -> HscEnv -> HscEnv +setHscWorkerEnv (ConcurrentWorkerEnv { cwe_logger = logger, cwe_tmpfs = tmpfs }) hsc_env = + hsc_env { hsc_logger = logger, hsc_tmpfs = tmpfs } ===================================== compiler/GHC/Driver/MakeSem.hs ===================================== @@ -39,6 +39,7 @@ import GHC.Utils.Json import System.Semaphore ( AbstractSem(..) , ClientSemaphore + , SemaphoreError , SemaphoreIdentifier , SemaphoreToken , openSemaphore @@ -534,18 +535,24 @@ makeJobserver sem_ident = do -- | Implement an abstract semaphore using a semaphore 'Jobserver' -- which queries the system semaphore of the given name for resources. +-- +-- Returns 'Left' if the system semaphore could not be opened, in which case +-- the operation is not run at all. A 'SemaphoreError' arising after the +-- semaphore was successfully opened is thrown, not returned. runJSemAbstractSem :: SemaphoreIdentifier -- ^ the semaphore identifier (from @-jsem@) -> (AbstractSem -> IO a) -- ^ the operation to run -- which requires a semaphore - -> IO a + -> IO (Either SemaphoreError a) runJSemAbstractSem sem_ident action = MC.mask \ unmask -> do - (abs, cleanup) <- makeJobserver sem_ident - r <- try $ unmask $ action abs - case r of - Left (e1 :: MC.SomeException) -> do - (_ :: Either MC.SomeException ()) <- MC.try cleanup - MC.throwM e1 - Right x -> cleanup $> x + MC.try @_ @SemaphoreError (makeJobserver sem_ident) >>= \case + Left open_failure -> return (Left open_failure) + Right (abs, cleanup) -> do + r <- try $ unmask $ action abs + case r of + Left (e1 :: MC.SomeException) -> do + (_ :: Either MC.SomeException ()) <- MC.try cleanup + MC.throwM e1 + Right x -> cleanup $> Right x {- Note [Architecture of the Job Server] ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ ===================================== compiler/GHC/Utils/TmpFs.hs ===================================== @@ -6,6 +6,7 @@ module GHC.Utils.TmpFs , initTmpFs , forkTmpFsFrom , mergeTmpFsInto + , withLocalTmpFS , PathsToClean(..) , emptyPathsToClean , TempFileLifetime(..) @@ -157,6 +158,16 @@ mergeTmpFsInto src dst = do atomicModifyIORef' (tmp_files_to_clean dst) (\s -> (mergePathsToClean src_files s, ())) atomicModifyIORef' (tmp_subdirs_to_clean dst) (\s -> (mergePathsToClean src_subdirs s, ())) +-- | Run an action with a local 'TmpFs' forked from the given 'TmpFs'. +-- +-- The remaining files of the local 'TmpFs' which weren't cleaned up by the +-- action are merged back into the given 'TmpFs', for clean-up later. +withLocalTmpFS :: TmpFs -> (TmpFs -> IO a) -> IO a +withLocalTmpFS tmpfs act = + Exception.bracket + (forkTmpFsFrom tmpfs) + (\tmpfs_local -> mergeTmpFsInto tmpfs_local tmpfs) + act cleanTempDirs :: Logger -> TmpFs -> IO () cleanTempDirs logger tmpfs ===================================== compiler/ghc.cabal.in ===================================== @@ -487,11 +487,13 @@ Library GHC.Driver.ByteCode GHC.Driver.CmdLine GHC.Driver.CodeOutput + GHC.Driver.Concurrency GHC.Driver.Config GHC.Driver.Config.Cmm GHC.Driver.Config.Cmm.Parser GHC.Driver.Config.CmmToAsm GHC.Driver.Config.CmmToLlvm + GHC.Driver.Config.Concurrency GHC.Driver.Config.Core.Lint GHC.Driver.Config.Core.Lint.Interactive GHC.Driver.Config.Core.Opt.Arity ===================================== utils/haddock/haddock-api/src/Haddock/Interface.hs ===================================== @@ -172,7 +172,7 @@ createIfaces verbosity modules flags instIfaceMap = do _ <- setSessionDynFlags dflags'' targets <- mapM (\(filePath, _) -> guessTarget filePath Nothing Nothing) hs_srcs setTargets targets - (_errs, modGraph) <- depanalE mkUnknownDiagnostic (Just batchMsg) [] False + (_errs, modGraph) <- depanalE [] False -- Create (if necessary) and load .hi-files. With --no-compilation this happens later. when (Flag_NoCompilation `notElem` flags) $ do View it on GitLab: https://gitlab.haskell.org/ghc/ghc/-/commit/566d789d5522489eb017e9999f0b7bde... -- View it on GitLab: https://gitlab.haskell.org/ghc/ghc/-/commit/566d789d5522489eb017e9999f0b7bde... You're receiving this email because of your account on gitlab.haskell.org. Manage all notifications: https://gitlab.haskell.org/-/profile/notifications | Help: https://gitlab.haskell.org/help