synchronize source repos concurrently

Cabal/src/Distribution/Simple/Utils.hs

@@ -192,6 +192,8 @@ module Distribution.Simple.Utils
   , unintersperse
   , wrapText
   , wrapLine
+  , sequenceConcurrentlyBounded
+  , sequenceConcurrentlyBounded_
 
     -- * FilePath stuff
   , isAbsoluteOnAnyPlatform
@@ -235,6 +237,7 @@ import Data.Typeable
   ( cast
   )
 
+import Control.Concurrent
 import qualified Control.Exception as Exception
 import Data.Time.Clock.POSIX (POSIXTime, getPOSIXTime)
 import Distribution.Compat.Process (proc)
@@ -2025,3 +2028,45 @@ findHookedPackageDesc verbosity mbWorkDir dir = do
 
 buildInfoExt :: String
 buildInfoExt = ".buildinfo"
+
+sequenceConcurrentlyBounded :: Int -> [IO a] -> IO [a]
+sequenceConcurrentlyBounded n xs = do
+  sem <- newQSem (n - 1)
+  tid <- myThreadId
+  let
+    catchForMe x =
+      Exception.catches
+        x
+        [ Exception.Handler $ \e@(Exception.SomeAsyncException _) -> throwIO e
+        , Exception.Handler $ \e@(SomeException _) -> Exception.throwTo tid e
+        ]
+  Exception.mask $ \restore -> do
+    resultvars <- for xs $ \x -> do
+      var <- newEmptyMVar
+      _tid <- forkIO $ Exception.bracket_ (waitQSem sem) (signalQSem sem) $ catchForMe $ do
+        res <- restore x
+        True <- tryPutMVar var res
+        return ()
+      return var
+    Exception.bracket_ (signalQSem sem) (waitQSem sem) (traverse takeMVar resultvars)
+
+sequenceConcurrentlyBounded_ :: Int -> [IO a] -> IO ()
+sequenceConcurrentlyBounded_ n xs = do
+  sem <- newQSem (n - 1)
+  tid <- myThreadId
+  let
+    catchForMe x =
+      Exception.catches
+        x
+        [ Exception.Handler $ \e@(Exception.SomeAsyncException _) -> throwIO e
+        , Exception.Handler $ \e@(SomeException _) -> Exception.throwTo tid e
+        ]
+  Exception.mask $ \restore -> do
+    resultvars <- for xs $ \x -> do
+      var <- newEmptyMVar
+      _tid <- forkIO $ Exception.bracket_ (waitQSem sem) (signalQSem sem) $ catchForMe $ do
+        _ <- restore x
+        True <- tryPutMVar var ()
+        return ()
+      return var
+    Exception.bracket_ (signalQSem sem) (waitQSem sem) (traverse_ takeMVar resultvars)

cabal-install/src/Distribution/Client/ProjectConfig.hs

@@ -494,7 +494,7 @@ resolveBuildTimeSettings
           cabalLogsDirectory
             </> "$compiler"
             </> "$libname"
-            <.> "log"
+              <.> "log"
       givenTemplate = flagToMaybe projectConfigLogFile
 
       useDefaultTemplate
@@ -1245,10 +1245,10 @@ fetchAndReadSourcePackages
             preferredHttpTransport
       sequenceA
         [ fetchAndReadSourcePackageRemoteTarball
-          verbosity
-          distDirLayout
-          getTransport
-          uri
+            verbosity
+            distDirLayout
+            getTransport
+            uri
         | ProjectPackageRemoteTarball uri <- pkgLocations
         ]
 
@@ -1403,15 +1403,17 @@ syncAndReadSourcePackagesRemoteRepos
             ]
 
     let progPathExtra = fromNubList projectConfigProgPathExtra
+    let numJobs = 4 -- hardcoded for now
     getConfiguredVCS <- delayInitSharedResources $ \repoType ->
       let vcs = Map.findWithDefault (error $ "Unknown VCS: " ++ prettyShow repoType) repoType knownVCSs
        in configureVCS verbosity progPathExtra vcs
 
     concat
-      <$> sequenceA
+      <$> sequenceConcurrentlyBoundedRebuild
+        numJobs
         [ rerunIfChanged verbosity monitor repoGroup' $ do
-          vcs' <- getConfiguredVCS repoType
-          syncRepoGroupAndReadSourcePackages vcs' pathStem repoGroup'
+            vcs' <- getConfiguredVCS repoType
+            syncRepoGroupAndReadSourcePackages vcs' pathStem repoGroup'
         | repoGroup@((primaryRepo, repoType) : _) <- Map.elems reposByLocation
         , let repoGroup' = map fst repoGroup
               pathStem =

cabal-install/src/Distribution/Client/RebuildMonad.hs

@@ -56,6 +56,7 @@ module Distribution.Client.RebuildMonad
   , findFileWithExtensionMonitored
   , findFirstFileMonitored
   , findFileMonitored
+  , sequenceConcurrentlyBoundedRebuild
   ) where
 
 import Distribution.Client.Compat.Prelude
@@ -66,7 +67,7 @@ import Distribution.Client.Glob hiding (matchFileGlob)
 import qualified Distribution.Client.Glob as Glob (matchFileGlob)
 import Distribution.Simple.PreProcess.Types (Suffix (..))
 
-import Distribution.Simple.Utils (debug)
+import Distribution.Simple.Utils (debug, sequenceConcurrentlyBounded)
 
 import Control.Concurrent.MVar (MVar, modifyMVar, newMVar)
 import Control.Monad.Reader as Reader
@@ -330,3 +331,13 @@ findFileMonitored searchPath fileName =
     [ path </> fileName
     | path <- nub searchPath
     ]
+
+-- | Run multiple 'Rebuild' actions in parallel, collecting the final
+-- list of used files.
+sequenceConcurrentlyBoundedRebuild :: Int -> [Rebuild a] -> Rebuild [a]
+sequenceConcurrentlyBoundedRebuild n xs = do
+  root <- askRoot
+  results <- liftIO $ sequenceConcurrentlyBounded n (unRebuild root <$> xs)
+  for results $ \(a, files) -> do
+    monitorFiles files
+    return a

changelog.d/synchronize-source-repos-concurrently

@@ -0,0 +1,5 @@
+synopsis: Synchronize source repositories concurrently
+packages: cabal-install
+prs: #0000
+significance: significant
+