sheaf pushed to branch wip/romes/27514 at Glasgow Haskell Compiler / GHC Commits: 04ba9450 by sheaf at 2026-08-19T12:12:07+02:00 Concurrent workers: determinism, avoid duplicated work - - - - - 3 changed files: - compiler/GHC/Driver/Concurrency.hs - compiler/GHC/Driver/Downsweep.hs - compiler/GHC/Driver/Pipeline/LogQueue.hs Changes: ===================================== compiler/GHC/Driver/Concurrency.hs ===================================== @@ -12,7 +12,8 @@ module GHC.Driver.Concurrency -- * Concurrent worker scheduling , ConcurrentWorkerEnv(..) , mapConcurrentWorkers - , concurrentTraversal_DF + , NodeExpander(..) + , concurrentTraversal ) where @@ -20,12 +21,16 @@ import GHC.Prelude import GHC.Driver.MakeSem import GHC.Driver.Pipeline.LogQueue - ( LogQueueQueue, finishLogQueue, initLogQueue, logThread - , newLogQueue, newLogQueueQueue, parLogAction ) + ( LogQueue, LogQueueQueue, finishLogQueue, initLogQueue, logThread + , newLogQueue, newLogQueueQueue, parLogAction, printLogs ) import GHC.Utils.Logger ( Logger, makeThreadSafe, pushLogHook ) +import GHC.Utils.Misc + ( HasDebugCallStack ) +import GHC.Utils.Outputable + ( Outputable(..), text, (<+>) ) import GHC.Utils.Panic - ( panic ) + ( massertPpr, pprPanic ) import GHC.Utils.TmpFs ( TmpFs, forkTmpFsFrom, mergeTmpFsInto, withLocalTmpFS ) @@ -36,31 +41,30 @@ import System.Semaphore import Control.Concurrent ( ThreadId, forkIOWithUnmask, killThread, myThreadId ) import Control.Concurrent.MVar - ( MVar, newEmptyMVar, newMVar, putMVar, takeMVar ) + ( MVar, 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 ) + , newQSem, signalQSem, waitQSem ) 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 ) + ( TVar, atomically, check, modifyTVar', newTVarIO, readTVar, readTVarIO + , writeTVar ) import Control.Exception ( AsyncException(ThreadKilled), SomeAsyncException, SomeException - , finally, fromException, mask, mask_, onException + , catch, finally, fromException, mask, mask_, onException , throwIO, try, uninterruptibleMask_ ) import Control.Monad - ( replicateM ) + ( unless ) import Data.Foldable - ( for_ ) + ( for_, traverse_ ) import Data.IORef - ( IORef, atomicModifyIORef', modifyIORef', newIORef, readIORef, writeIORef ) + ( IORef, atomicModifyIORef', modifyIORef', newIORef, readIORef ) import qualified Data.Map as Map import qualified Data.Sequence as Seq import qualified Data.Set as Set @@ -211,72 +215,163 @@ withWorkerLimit logger tmpfs report_semaphore_failure limit action action conc parent_work_env -------------------------------------------------------------------------------- --- * Scheduling concurrent workers +-- * Monotone data structures -------------------------------------------------------------------------------- --- | Internal scheduler abstraction with two capabilities: +-- | A map that only ever grows, and whose entries are written at most once. +newtype MonotoneMap k v = MonotoneMap ( IORef ( Map.Map k v ) ) + +newMonotoneMap :: Map.Map k v -> IO ( MonotoneMap k v ) +newMonotoneMap initial = MonotoneMap <$> newIORef initial + +-- | The outcome of inserting into a 'MonotoneMap' or a 'MonotoneSet'. +data InsertionResult + -- | The key was absent before the insertion. + = Inserted + -- | The key was already present; the container is unchanged. + | AlreadyPresent + +-- | Write an entry into a 'MonotoneMap' unless the key is already present. +insertMonotoneMap :: Ord k => MonotoneMap k v -> k -> v -> IO InsertionResult +insertMonotoneMap ( MonotoneMap ref ) k v = + atomicModifyIORef' ref \ m -> + case Map.insertLookupWithKey ( \ _ _ old -> old ) k v m of + ( Nothing , m' ) -> ( m', Inserted ) + ( Just _ , _ ) -> ( m , AlreadyPresent ) + +-- | Write a new entry into a 'MonotoneMap'. -- --- - 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. - } +-- Panics if the entry is already present. +insertMonotoneMap_new + :: ( HasDebugCallStack, Ord k, Outputable k ) + => MonotoneMap k v -> k -> v -> IO () +insertMonotoneMap_new mm k v = + insertMonotoneMap mm k v >>= \case + Inserted -> pure () + AlreadyPresent -> pprPanic "monotone map: duplicate key" $ ppr k + +-- | The contents of a monotone map. +freezeMonotoneMap :: MonotoneMap k v -> IO ( Map.Map k v ) +freezeMonotoneMap ( MonotoneMap ref ) = readIORef ref + +-- | A set that only ever grows. +newtype MonotoneSet k = MonotoneSet ( IORef ( Set.Set k ) ) + +newMonotoneSet :: Set.Set k -> IO ( MonotoneSet k ) +newMonotoneSet initial = MonotoneSet <$> newIORef initial + +-- | Add an element, unless it is already present. +insertMonotoneSet :: Ord k => MonotoneSet k -> k -> IO InsertionResult +insertMonotoneSet ( MonotoneSet ref ) k = + atomicModifyIORef' ref \ s -> + if k `Set.member` s + then ( s , AlreadyPresent ) + else ( Set.insert k s , Inserted ) + +-------------------------------------------------------------------------------- +-- * Pools of concurrent workers +-------------------------------------------------------------------------------- --- | Internal implementation of a concurrent worker scheduler. +-- | The order in which logging should happen when using concurrent workers. +data LogOrder + -- | Log as we go. + -- + -- Only valid when workers are spawned in a deterministic order. + = LogAsWeGo + -- | Accumulate logs per worker. Once all work is done, sort the logs + -- before proceeding. + -- + -- Used when workers may be spawned in a non-deterministic order. + | SortLogs + +-- | A pool of concurrent workers with a given worker key type, supporting two +-- operations: -- --- Usage of this function requires the following: +-- - spawning a new worker, +-- - waiting on all workers to finish. -- --- - 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 +-- See Note [Deterministic concurrent workers]. +data WorkerPool worker_key = + WorkerPool + { spawnWorker :: worker_key -> ( ConcurrentWorkerEnv -> IO () ) -> IO () + -- ^ Spawn one worker with the given worker key. + -- + -- May be called from inside another 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. + -- + -- An exception escaping the action stops further workers from being + -- spawned, and is rethrown by 'waitForWorkers'. + , waitForWorkers :: IO () + -- ^ Wait until all workers are done, throwing an exception if any + -- worker failed (which exception is thrown is not deterministic). + } + +-- | Internal implementation of a pool of concurrent workers. +run_pool + :: forall worker_key a + . ( HasDebugCallStack, Ord worker_key, Outputable worker_key ) + => String -- ^ thread label for workers + -> LogOrder -> Concurrency -> ConcurrentWorkerEnv - -> (Scheduler r -> IO a) - -- ^ worker action + -> ( WorkerPool worker_key -> IO a ) -> IO a -run_schedule worker_label conc parent_work_env withScheduler = +run_pool worker_label log_order conc parent_work_env withPool = case conc of Serial -> do - results_var <- newIORef Seq.empty + queued_var <- newIORef $ Seq.empty @( worker_key, ConcurrentWorkerEnv -> IO () ) + logs_var <- newMonotoneMap $ Map.empty @worker_key @LogQueue + let - spawnWorker :: (ConcurrentWorkerEnv -> IO r) -> IO () - spawnWorker action = do - res <- try @SomeException $ + spawnWorker :: worker_key -> ( ConcurrentWorkerEnv -> IO () ) -> IO () + spawnWorker worker_key action = + modifyIORef' queued_var ( Seq.|> ( worker_key, action ) ) + + run_worker :: worker_key -> ( ConcurrentWorkerEnv -> IO () ) -> IO () + run_worker worker_key action = case log_order of + LogAsWeGo -> 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 } + SortLogs -> do + -- Use a log queue for consistency with the concurrent case. + lq <- newLogQueue + insertMonotoneMap_new logs_var worker_key lq + let + worker_work_env :: ConcurrentWorkerEnv + worker_work_env = + parent_work_env + { cwe_logger = pushLogHook ( const ( parLogAction lq ) ) + ( cwe_logger parent_work_env ) } + workerEnv_withLocalTmpFS worker_work_env action + `finally` finishLogQueue lq + + waitForWorkers :: IO () + waitForWorkers = do + next <- atomicModifyIORef' queued_var \ queued -> + case queued of + work Seq.:<| rest -> ( rest , Just work ) + Seq.Empty -> ( queued, Nothing ) + case next of + Nothing -> pure () + Just ( worker_key, action ) -> + run_worker worker_key action *> waitForWorkers + + print_logs :: IO () + print_logs = do + logs <- freezeMonotoneMap logs_var + for_ ( Map.elems logs ) $ printLogs ( cwe_logger parent_work_env ) + + withPool ( WorkerPool { spawnWorker, waitForWorkers } ) + `finally` print_logs 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)) + worker_tids_var <- newTVarIO $ Set.empty @ThreadId + failure_var <- newTVarIO $ Nothing @SomeException + logs_var <- newMonotoneMap $ Map.empty @worker_key @LogQueue + last_spawned_var <- newIORef $ Nothing @worker_key let wait_for_workers :: IO () @@ -291,92 +386,118 @@ run_schedule worker_label conc parent_work_env withScheduler = 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 + -- Record a worker failure, preventing any further work from starting. + record_failure :: SomeException -> IO () + record_failure e = + atomically $ modifyTVar' failure_var \ failure -> + case failure of + Nothing -> Just e + Just {} -> failure - let + waitForWorkers :: IO () + waitForWorkers = do + wait_for_workers + traverse_ throwIO =<< readTVarIO failure_var + + -- Create the log queue of a worker, ordering it according to the worker key. + new_worker_log_queue :: worker_key -> IO LogQueue + new_worker_log_queue worker_key = do + lq <- newLogQueue + case log_order of + LogAsWeGo -> do + last_spawned <- + atomicModifyIORef' last_spawned_var \ last_spawned -> + ( Just worker_key, last_spawned ) + massertPpr ( all ( < worker_key ) last_spawned ) $ + text "run_pool: LogAsWeGo workers spawned out of order:" + <+> ppr last_spawned <+> text "then" <+> ppr worker_key + job_id <- atomicModifyIORef' ce_next_log_queue_id \ n -> ( n + 1, n ) + atomically $ initLogQueue ce_log_queue_queue job_id lq + SortLogs -> + insertMonotoneMap_new logs_var worker_key lq + pure lq + + -- Hand the log queues over for printing, in worker key order. + release_queued_logs :: IO () + release_queued_logs = do + queued <- freezeMonotoneMap logs_var + unless ( Map.null queued ) do + first_id <- + atomicModifyIORef' ce_next_log_queue_id \ n -> + ( n + Map.size queued, n ) + atomically $ + for_ ( zip [ first_id .. ] ( Map.elems queued ) ) \ ( job_id, lq ) -> + initLogQueue ce_log_queue_queue job_id lq + + spawnWorker :: worker_key -> ( ConcurrentWorkerEnv -> IO () ) -> IO () + spawnWorker worker_key action = mask_ do + failure <- readTVarIO failure_var + case failure of + -- A worker has failed: don't start any more work. + Just {} -> pure () + Nothing -> do + + -- TmpFs + lcl_tmpfs <- forkTmpFsFrom ( cwe_tmpfs parent_work_env ) + + -- LogQueue + lq <- new_worker_log_queue worker_key - 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 + 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 ) + } + + -- 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 + + handle_worker_exception :: SomeException -> IO () + handle_worker_exception e + -- Worker is being cancelled: not a failure to report. + | Just ThreadKilled <- fromException e + = pure () + | otherwise + = record_failure e + + run_worker :: ( forall b. IO b -> IO b ) -> IO () + run_worker unmask = do + tid <- myThreadId + labelThread tid worker_label + ( unmask ( action worker_work_env ) + `catch` handle_worker_exception ) + `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 + + ( `finally` release_queued_logs ) $ + mask \ restore -> do + result <- restore ( withPool $ WorkerPool { spawnWorker, waitForWorkers } ) + `onException` cancel_workers + restore wait_for_workers `onException` cancel_workers + pure result -------------------------------------------------------------------------------- -- * Derived scheduling functionality @@ -387,7 +508,9 @@ run_schedule worker_label conc parent_work_env withScheduler = -- Workers run to completion (no early abort); the first exception -- (in input order) is rethrown at the end. mapConcurrentWorkers - :: String -- ^ thread label for workers + :: forall a b + . HasDebugCallStack + => String -- ^ thread label for workers -> Concurrency -> ConcurrentWorkerEnv -> (ConcurrentWorkerEnv -> a -> IO b) @@ -398,67 +521,111 @@ mapConcurrentWorkers -> [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) + -- LogAsWeGo: workers are keyed by their position in the input list and + -- spawned in that same order, so their output can be printed as it is produced. + run_pool worker_label LogAsWeGo conc work_env \ pool -> do + results <- newMonotoneMap $ Map.empty @Int @( Either SomeException b ) + for_ ( zip [ 0 .. ] xs ) \ ( i, x ) -> + spawnWorker pool i \ worker_env -> do + res <- try @SomeException $ f worker_env x + case res of + Left e + -- Take care to avoid swallowing async exceptions. + | Just _ <- fromException @SomeAsyncException e + -> throwIO e + _ -> insertMonotoneMap_new results i res + waitForWorkers pool + all_results <- freezeMonotoneMap results + massertPpr ( Map.size all_results == length xs ) $ + text "mapConcurrentWorkers: missing results" + either throwIO pure $ sequence $ Map.elems all_results + +-- | How to expand a node in a graph for 'concurrentTraversal'. +data NodeExpander k n v = + NodeExpander + { nodeKey :: n -> k + -- ^ The identity of a node. + , expandNode :: ConcurrentWorkerEnv -> n -> IO ( v, [n] ) + -- ^ Expand a node into its result and the children to visit next. + -- + -- To guarantee determinism, the children must be a pure function of the + -- input, and IO effects must not observably depend on the order in + -- which nodes are expanded. + -- + -- NB: workers do not hold semaphore tokens by default; use + -- 'withConcurrency' to acquire one + } --- | Depth-first traversal with on-the-fly expansion of nodes. +-- | Deterministically traverse a graph whose nodes are discovered as they are +-- expanded. -- --- 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 +-- Fails fast: once a worker throws an exception, no further work is started, +-- and the exception is rethrown once the outstanding workers finish. +concurrentTraversal + :: forall k n v + . ( HasDebugCallStack, Ord k, Outputable k ) => String -- ^ thread label for workers -> Concurrency -> ConcurrentWorkerEnv - -> Map.Map k r + -> NodeExpander k n v + -> Map.Map k v -- ^ 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 + -> IO ( Map.Map k v ) +concurrentTraversal + worker_label conc work_env + ( NodeExpander { nodeKey, expandNode } ) + base_map roots + = + -- SortLogs: nodes are discovered in an order that depends on the schedule, + -- so the workers' logs must be ordered before being printed. + run_pool worker_label SortLogs conc work_env \ pool -> do + + -- The keys whose expansion has been started. + claims <- newMonotoneSet $ Map.keysSet base_map + results <- newMonotoneMap base_map + 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 + discover :: n -> IO () + discover node = + -- Claim the work for this node to avoid any other worker duplicating it. + insertMonotoneSet claims key >>= \case + AlreadyPresent -> pure () + Inserted -> + spawnWorker pool key \ worker_env -> do + ( result, children ) <- expandNode worker_env node + insertMonotoneMap_new results key result + for_ children discover 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 + key = nodeKey node + + for_ roots discover + waitForWorkers pool + freezeMonotoneMap results + +{- Note [Deterministic concurrent workers] +~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ +To provide deterministic output when doing graph traversal (as in downsweep) +despite using concurrent workers, we ensure that nothing can observe the order +in which workers do their work: + + 1. Any chunk of work is performed at most once: every worker atomically claims + ownership of the work it is going to do before it starts that work. + + 2. Workers report results by writing to a 'MonotoneMap', whose entries are + written at most once. Other outputs (such as logging output) is accumulated + in a deterministic order and reported at the end. + +This scheme allows us to retain maximum concurrency: it allows new edges to be +discovered by any worker and immediately processed. + +For this scheme to provide deterministic output, we require that: + + * The expansion of a node is a pure function of the node. + * The work itself should not observably depend on when it was run. + +Failure is not deterministic: which worker's exception is reported depends on +the schedule. When deterministic error messages are desired, the workers should +return an error value instead (as 'mapConcurrentWorkers' does). +-} ===================================== compiler/GHC/Driver/Downsweep.hs ===================================== @@ -174,9 +174,8 @@ incrementally constructing a ModuleGraph using the GHC API; See #27054). So `downsweep` takes a `Maybe ModuleGraph` as one of its arguments. Downsweep iteratively *expands* each so-called 'DownsweepNode' into a list of -its dependencies, and recursively traverses all reachable nodes in a parallel -non-det-depth-first order using 'parDfsBuild'. A 'DownsweepNode' is *expanded* -by 'dsNodeExpand': +its dependencies, and recursively traverses all reachable nodes concurrently +using 'parBuild'. A 'DownsweepNode' is *expanded* by 'dsNodeExpand': dsNodeExpand :: DownsweepNode -> DownsweepM (NodeRes (ModuleGraphNode, [DownsweepNode])) @@ -591,7 +590,7 @@ loopModuleNodeInfos :: M.Map NodeKey (NodeRes ModuleGraphNode) -> [ModuleNodeInf loopUnits :: M.Map NodeKey (NodeRes ModuleGraphNode) -> UnitId -> [UnitId] -> DownsweepM (M.Map NodeKey (NodeRes ModuleGraphNode)) loopInstantiations :: M.Map NodeKey (NodeRes ModuleGraphNode) -> [(UnitId, InstantiatedUnit)] -> DownsweepM (M.Map NodeKey (NodeRes ModuleGraphNode)) loopFromInteractive :: M.Map NodeKey (NodeRes ModuleGraphNode) -> Module -> [InteractiveImport] -> DownsweepM (M.Map NodeKey (NodeRes ModuleGraphNode)) -loopDownsweepNodes base_map nodes = parDfsBuild (Just base_map) nodes dsNodeInfoKey dsNodeExpand +loopDownsweepNodes base_map nodes = parBuild (Just base_map) nodes dsNodeInfoKey dsNodeExpand loopModuleNodeInfos base_map = loopDownsweepNodes base_map . map DSMod loopUnits base_map homud = loopDownsweepNodes base_map . map (DSUnit homud) loopInstantiations base_map = loopDownsweepNodes base_map . map (uncurry DSInst) @@ -1732,10 +1731,10 @@ getPreprocessedImports hsc_env src_fn mb_phase maybe_buf = do return PreprocessedImports {..} -------------------------------------------------------------------------------- --- * Generic traversal of iteratively-built graph: parDfsBuild +-- * Generic traversal of iteratively-built graph: parBuild -------------------------------------------------------------------------------- --- | The result of expanding a node in 'parDfsBuild'. +-- | The result of expanding a node in 'parBuild'. data NodeRes v -- | Computed the node payload successfully = NSuccess v @@ -1748,20 +1747,18 @@ data NodeRes v -- abort. | NSkip --- | In a parallel non-det-depth-first order, and starting from the given roots, traverse a --- graph by iteratively expanding a node into a payload and a list of children --- nodes to visit next. +-- | Starting from the given roots, traverse a graph by iteratively expanding a +-- node into a payload and a list of children nodes to visit next, expanding +-- nodes concurrently. -- --- A node is NEVER visited/expanded more than once, as long as the node key --- @k@, computed from the node @n@, uniquely identifies that node. --- --- The first argument @base_map@ is the starting set of already visited nodes --- (these nodes won't be expanded again!). +-- A node is NEVER visited/expanded more than once: nodes are identified by +-- their key @k@, and the first node discovered under a key is the one that is +-- expanded. -- -- The result is a mapping from the key of every node transitively reachable -- from the root nodes (inclusively) to the payload returned by expanding that -- node. The result includes the previously visited nodes given in @base_map@, --- s.t. @parDfsBuild base_map [] _ _ == base_map@. +-- s.t. @parBuild base_map [] _ _ _ == base_map@. -- -- The @expand@ function returns a 'NodeRes'. See the 'NodeRes' documentation -- for more information about each result type. @@ -1771,7 +1768,7 @@ data NodeRes v -- See Note [Parallel Downsweep] for more information about how parallelism is -- achieved, and See Note [Downsweep Control Flow and Caching] for information -- about the various caches used. -parDfsBuild :: forall k v n. Ord k +parBuild :: forall k v n. (Ord k, Outputable k) => Maybe (Map.Map k (NodeRes v)) -- ^ Base map, existing results. We won't re-expand any of the nodes -- already present in this map. @@ -1785,22 +1782,26 @@ parDfsBuild :: forall k v n. Ord k -> DownsweepM (Map.Map k (NodeRes v)) -- ^ 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 +parBuild base_map roots nodeKey expand = ReaderT $ \ds_env -> do let conc :: Concurrency conc = ds_concurrency ds_env - expand_node :: ConcurrentWorkerEnv -> n -> IO (NodeRes v, [n]) - expand_node worker_env node = do + expandNode :: ConcurrentWorkerEnv -> n -> IO (NodeRes v, [n]) + expandNode 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 + concurrentTraversal "downsweep_worker" + conc + (viewHscWorkerEnv $ ds_hsc_env ds_env) + (NodeExpander { nodeKey, expandNode }) + (fromMaybe mempty base_map) + roots setDownsweepWorkerEnv :: ConcurrentWorkerEnv -> DownsweepEnv -> DownsweepEnv setDownsweepWorkerEnv work_env env = @@ -1809,7 +1810,7 @@ setDownsweepWorkerEnv work_env env = {- Note [Downsweep Control Flow and Caching] ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ -The control flow of downsweep is extracted into a single function `parDfsBuild`, +The control flow of downsweep is extracted into a single function `parBuild`, which takes care of iteratively expanding and traversing all nodes of the in-construction module graph necessary to build a full `ModuleGraph` at the end. @@ -1818,7 +1819,7 @@ There are three levels of caching going on, all of which are necessary to make sure we don't do repeated work (notably, we NEVER summarise the same module twice). -1. `parDfsBuild` accumulates the final module graph and never revisits the +1. `parBuild` accumulates the final module graph and never revisits the same node of the module graph. Cache is keyed by the final `ModuleGraph`s `NodeKey`s. @@ -1900,9 +1901,12 @@ 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. -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. +We use the concurrent traversal abstraction from GHC.Driver.Concurrency +('concurrentTraversal'). Each time we discover a new node, a worker is spawned +to expand it; a worker spawns the workers for the children it discovers itself, +so an expansion never waits on an unrelated one. With -j1 no threads are +involved: each expansion runs in sequence. + +The traversal is deterministic despite the fact that the order in which workers +are spawned is not: see Note [Deterministic concurrent workers] in GHC.Driver.Concurrency. -} ===================================== compiler/GHC/Driver/Pipeline/LogQueue.hs ===================================== @@ -4,6 +4,7 @@ module GHC.Driver.Pipeline.LogQueue ( LogQueue(..) , finishLogQueue , writeLogQueue , parLogAction + , printLogs , LogQueueQueue(..) , initLogQueue @@ -25,19 +26,18 @@ import Control.Monad -- LogQueue Abstraction --- | Each module is given a unique 'LogQueue' to redirect compilation messages --- to. A 'Nothing' value contains the result of compilation, and denotes the --- end of the message queue. -data LogQueue = LogQueue { logQueueId :: !Int - , logQueueMessages :: !(IORef [Maybe (MessageClass, SrcSpan, SDoc, LogFlags)]) +-- | A buffer of compilation messages produced by one worker. +-- +-- A 'Nothing' value denotes the end of the message queue. +data LogQueue = LogQueue { logQueueMessages :: !(IORef [Maybe (MessageClass, SrcSpan, SDoc, LogFlags)]) , logQueueSemaphore :: !(MVar ()) } -newLogQueue :: Int -> IO LogQueue -newLogQueue n = do +newLogQueue :: IO LogQueue +newLogQueue = do mqueue <- newIORef [] sem <- newMVar () - return (LogQueue n mqueue sem) + return (LogQueue mqueue sem) finishLogQueue :: LogQueue -> IO () finishLogQueue lq = do @@ -50,7 +50,7 @@ writeLogQueue lq msg = do -- | Internal helper for writing log messages writeLogQueueInternal :: LogQueue -> Maybe (MessageClass,SrcSpan,SDoc, LogFlags) -> IO () -writeLogQueueInternal (LogQueue _n ref sem) msg = do +writeLogQueueInternal (LogQueue ref sem) msg = do atomicModifyIORef' ref $ \msgs -> (msg:msgs,()) _ <- tryPutMVar sem () return () @@ -61,9 +61,11 @@ parLogAction :: LogQueue -> LogAction parLogAction log_queue log_flags !msgClass !srcSpan !msg = writeLogQueue log_queue (msgClass,srcSpan,msg, log_flags) --- Print each message from the log_queue using the global logger +-- | Print each message from the log queue using the given logger. +-- +-- Blocks until the queue has been finished with 'finishLogQueue'. printLogs :: Logger -> LogQueue -> IO () -printLogs !logger (LogQueue _n ref sem) = read_msgs +printLogs !logger (LogQueue ref sem) = read_msgs where read_msgs = do takeMVar sem msgs <- atomicModifyIORef' ref $ \xs -> ([], reverse xs) @@ -84,11 +86,23 @@ data LogQueueQueue = LogQueueQueue Int (IM.IntMap LogQueue) newLogQueueQueue :: LogQueueQueue newLogQueueQueue = LogQueueQueue 1 IM.empty -addToQueueQueue :: LogQueue -> LogQueueQueue -> LogQueueQueue -addToQueueQueue lq (LogQueueQueue n im) = LogQueueQueue n (IM.insert (logQueueId lq) lq im) - -initLogQueue :: TVar LogQueueQueue -> LogQueue -> STM () -initLogQueue lqq lq = modifyTVar lqq (addToQueueQueue lq) +addToQueueQueue + :: Int -- ^ 1-indexed position in which to add + -> LogQueue + -> LogQueueQueue + -> LogQueueQueue +addToQueueQueue i lq (LogQueueQueue n im) = LogQueueQueue n (IM.insert i lq im) + +-- | Hand a log queue to the log thread, to be printed at the given position. +-- +-- Positions must be contiguous: the log thread prints position @n@ only after +-- every position below @n@ is done. +initLogQueue + :: TVar LogQueueQueue + -> Int -- ^ position in the 'LogQueueQueue' in which to insert the 'LogQueue' + -> LogQueue + -> STM () +initLogQueue lqq i lq = modifyTVar lqq (addToQueueQueue i lq) -- | Return all items in the queue in ascending order allLogQueues :: LogQueueQueue -> [LogQueue] View it on GitLab: https://gitlab.haskell.org/ghc/ghc/-/commit/04ba945021e675d73807501cf132cec7... -- View it on GitLab: https://gitlab.haskell.org/ghc/ghc/-/commit/04ba945021e675d73807501cf132cec7... 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