fixup! Synchronize VCS repos concurrently

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

@@ -466,6 +466,7 @@ installAction flags@NixStyleFlags{extraFlags, configFlags, installFlags, project
         fetchAndReadSourcePackages
           verbosity
           distDirLayout
+          compiler
           (projectConfigShared config)
           (projectConfigBuildOnly config)
           [ProjectPackageRemoteTarball uri | uri <- uris]

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

@@ -31,6 +31,11 @@ module Distribution.Client.JobControl
   , Lock
   , newLock
   , criticalSection
+
+  -- * Higher level utils
+  , newJobControlFromParStrat
+  , withJobControl
+  , mapConcurrentWithJobs
   ) where
 
 import Distribution.Client.Compat.Prelude
@@ -40,11 +45,14 @@ import Control.Concurrent (forkIO, forkIOWithUnmask, threadDelay)
 import Control.Concurrent.MVar
 import Control.Concurrent.STM (STM, TVar, atomically, modifyTVar', newTVarIO, readTVar)
 import Control.Concurrent.STM.TChan
-import Control.Exception (bracket_, mask_, try)
+import Control.Exception (bracket_, mask_, try, bracket)
 import Control.Monad (forever, replicateM_)
 import Distribution.Client.Compat.Semaphore
+import Distribution.Client.Utils (numberOfProcessors)
 import Distribution.Compat.Stack
+import Distribution.Simple.Compiler
 import Distribution.Simple.Utils
+import Distribution.Types.ParStrat
 import System.Semaphore
 
 -- | A simple concurrency abstraction. Jobs can be spawned and can complete
@@ -262,3 +270,34 @@ newLock = fmap Lock $ newMVar ()
 
 criticalSection :: Lock -> IO a -> IO a
 criticalSection (Lock lck) act = bracket_ (takeMVar lck) (putMVar lck ()) act
+
+--------------------------------------------------------------------------------
+-- More high level utils
+--------------------------------------------------------------------------------
+
+newJobControlFromParStrat :: Verbosity -> Compiler
+                          -> ParStratInstall -- ^ The parallel strategy
+                          -> Maybe Int       -- ^ A cap on the number of jobs (e.g. to force a maximum of 2 concurrent downloads despite a -j8 parallel strategy)
+                          -> IO (JobControl IO a)
+newJobControlFromParStrat verbosity compiler parStrat numJobsCap = case parStrat of
+    Serial -> newSerialJobControl
+    NumJobs n -> newParallelJobControl (capJobs (fromMaybe numberOfProcessors n))
+    UseSem n ->
+      if jsemSupported compiler
+        then newSemaphoreJobControl verbosity (capJobs n)
+        else do
+          warn verbosity "-jsem is not supported by the selected compiler, falling back to normal parallelism control."
+          newParallelJobControl (capJobs n)
+  where
+    capJobs n = min (fromMaybe maxBound numJobsCap) n
+
+withJobControl :: IO (JobControl IO a) -> (JobControl IO a -> IO b) -> IO b
+withJobControl mkJC = bracket mkJC cleanupJobControl
+
+-- | Concurrently execute actions on a list using the given JobControl.
+-- The maximum number of concurrent jobs is tied to the JobControl instance.
+-- The resulting list does /not/ preserve the original order!
+mapConcurrentWithJobs :: JobControl IO b -> (a -> IO b) -> [a] -> IO [b]
+mapConcurrentWithJobs jobControl f xs = do
+  traverse_ (spawnJob jobControl . f) xs
+  traverse (const $ collectJob jobControl) xs

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

@@ -74,7 +74,6 @@ import Distribution.Package
 import Distribution.Simple.Compiler
   ( Compiler
   , PackageDB (..)
-  , jsemSupported
   )
 import Distribution.Simple.Program
 import qualified Distribution.Simple.Register as Cabal
@@ -92,7 +91,7 @@ import qualified Data.Set as Set
 
 import qualified Text.PrettyPrint as Disp
 
-import Control.Exception (assert, bracket, handle)
+import Control.Exception (assert, handle)
 import System.Directory (doesDirectoryExist, doesFileExist, renameDirectory)
 import System.FilePath (makeRelative, normalise, takeDirectory, (<.>), (</>))
 import System.Semaphore (SemaphoreName (..))
@@ -102,7 +101,6 @@ import Distribution.Simple.Flag (fromFlagOrDefault)
 
 import Distribution.Client.ProjectBuilding.PackageFileMonitor
 import Distribution.Client.ProjectBuilding.UnpackedPackage (annotateFailureNoLog, buildAndInstallUnpackedPackage, buildInplaceUnpackedPackage)
-import Distribution.Client.Utils (numberOfProcessors)
 
 ------------------------------------------------------------------------------
 
@@ -359,17 +357,6 @@ rebuildTargets
     }
     | fromFlagOrDefault False (projectConfigOfflineMode config) && not (null packagesToDownload) = return offlineError
     | otherwise = do
-        -- Concurrency control: create the job controller and concurrency limits
-        -- for downloading, building and installing.
-        mkJobControl <- case buildSettingNumJobs of
-          Serial -> newSerialJobControl
-          NumJobs n -> newParallelJobControl (fromMaybe numberOfProcessors n)
-          UseSem n ->
-            if jsemSupported compiler
-              then newSemaphoreJobControl verbosity n
-              else do
-                warn verbosity "-jsem is not supported by the selected compiler, falling back to normal parallelism control."
-                newParallelJobControl n
         registerLock <- newLock -- serialise registration
         cacheLock <- newLock -- serialise access to setup exe cache
         -- TODO: [code cleanup] eliminate setup exe cache
@@ -384,7 +371,9 @@ rebuildTargets
         createDirectoryIfMissingVerbose verbosity True distTempDirectory
         traverse_ (createPackageDBIfMissing verbosity compiler progdb) packageDBsToUse
 
-        bracket (pure mkJobControl) cleanupJobControl $ \jobControl -> do
+        -- Concurrency control: create the job controller and concurrency limits
+        -- for downloading, building and installing.
+        withJobControl (newJobControlFromParStrat verbosity compiler buildSettingNumJobs Nothing) $ \jobControl -> do
           -- Before traversing the install plan, preemptively find all packages that
           -- will need to be downloaded and start downloading them.
           asyncDownloadPackages
@@ -395,7 +384,7 @@ rebuildTargets
             $ \downloadMap ->
               -- For each package in the plan, in dependency order, but in parallel...
               InstallPlan.execute
-                mkJobControl
+                jobControl
                 keepGoing
                 (BuildFailure Nothing . DependentFailed . packageId)
                 installPlan

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

@@ -51,6 +51,7 @@ module Distribution.Client.ProjectConfig
   , resolveSolverSettings
   , BuildTimeSettings (..)
   , resolveBuildTimeSettings
+  , resolveNumJobsSetting
 
     -- * Checking configuration
   , checkBadPerPackageCompilerPaths
@@ -61,6 +62,7 @@ import Distribution.Client.Compat.Prelude
 import Text.PrettyPrint (nest, render, text, vcat)
 import Prelude ()
 
+import Distribution.Client.JobControl
 import Distribution.Client.Glob
   ( isTrivialRootedGlob
   )
@@ -424,12 +426,7 @@ resolveBuildTimeSettings
       -- buildSettingLogVerbosity  -- defined below, more complicated
       buildSettingBuildReports = fromFlag projectConfigBuildReports
       buildSettingSymlinkBinDir = flagToList projectConfigSymlinkBinDir
-      buildSettingNumJobs =
-        if fromFlag projectConfigUseSemaphore
-          then UseSem (determineNumJobs projectConfigNumJobs)
-          else case (determineNumJobs projectConfigNumJobs) of
-            1 -> Serial
-            n -> NumJobs (Just n)
+      buildSettingNumJobs = resolveNumJobsSetting projectConfigUseSemaphore projectConfigNumJobs
       buildSettingKeepGoing = fromFlag projectConfigKeepGoing
       buildSettingOfflineMode = fromFlag projectConfigOfflineMode
       buildSettingKeepTempFiles = fromFlag projectConfigKeepTempFiles
@@ -525,6 +522,15 @@ resolveBuildTimeSettings
         | isParallelBuild buildSettingNumJobs = False
         | otherwise = False
 
+-- | Determine the number of jobs (ParStrat) from the project config
+resolveNumJobsSetting :: Flag Bool -> Flag (Maybe Int) -> ParStratX Int
+resolveNumJobsSetting projectConfigUseSemaphore projectConfigNumJobs =
+  if fromFlag projectConfigUseSemaphore
+    then UseSem (determineNumJobs projectConfigNumJobs)
+    else case (determineNumJobs projectConfigNumJobs) of
+      1 -> Serial
+      n -> NumJobs (Just n)
+
 ---------------------------------------------
 -- Reading and writing project config files
 --
@@ -1156,13 +1162,15 @@ mplusMaybeT ma mb = do
 fetchAndReadSourcePackages
   :: Verbosity
   -> DistDirLayout
+  -> Compiler
   -> ProjectConfigShared
   -> ProjectConfigBuildOnly
   -> [ProjectPackageLocation]
   -> Rebuild [PackageSpecifier (SourcePackage UnresolvedPkgLoc)]
 fetchAndReadSourcePackages
   verbosity
   distDirLayout
+  compiler
   projectConfigShared
   projectConfigBuildOnly
   pkgLocations = do
@@ -1199,7 +1207,9 @@ fetchAndReadSourcePackages
       syncAndReadSourcePackagesRemoteRepos
         verbosity
         distDirLayout
+        compiler
         projectConfigShared
+        projectConfigBuildOnly
         (fromFlag (projectConfigOfflineMode projectConfigBuildOnly))
         [repo | ProjectPackageRemoteRepo repo <- pkgLocations]
 
@@ -1316,16 +1326,23 @@ fetchAndReadSourcePackageRemoteTarball
 syncAndReadSourcePackagesRemoteRepos
   :: Verbosity
   -> DistDirLayout
+  -> Compiler
   -> ProjectConfigShared
+  -> ProjectConfigBuildOnly
   -> Bool
   -> [SourceRepoList]
   -> Rebuild [PackageSpecifier (SourcePackage UnresolvedPkgLoc)]
 syncAndReadSourcePackagesRemoteRepos
   verbosity
   DistDirLayout{distDownloadSrcDirectory}
+  compiler
   ProjectConfigShared
     { projectConfigProgPathExtra
     }
+  ProjectConfigBuildOnly
+    { projectConfigUseSemaphore
+    , projectConfigNumJobs
+    }
   offlineMode
   repos = do
     repos' <-
@@ -1351,7 +1368,7 @@ syncAndReadSourcePackagesRemoteRepos
        in configureVCS verbosity progPathExtra vcs
 
     concat
-     <$> rerunConcurrentlyIfChanged verbosity
+     <$> rerunConcurrentlyIfChanged verbosity (newJobControlFromParStrat verbosity compiler parStrat maxNumFetchJobs)
           [ ( monitor
             , repoGroup'
             , do vcs' <- getConfiguredVCS repoType
@@ -1369,6 +1386,8 @@ syncAndReadSourcePackagesRemoteRepos
                 monitor = newFileMonitor (pathStem <.> "cache")
           ]
     where
+      maxNumFetchJobs = Just 2 -- try to keep this in sync with Distribution.Client.Install's numFetchJobs. 
+      parStrat = resolveNumJobsSetting projectConfigUseSemaphore projectConfigNumJobs
       syncRepoGroupAndReadSourcePackages
         :: VCS ConfiguredProgram
         -> FilePath

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

@@ -205,12 +205,10 @@ type ProjectConfigSkeleton = CondTree ConfVar [ProjectConfigPath] ProjectConfig
 singletonProjectConfigSkeleton :: ProjectConfig -> ProjectConfigSkeleton
 singletonProjectConfigSkeleton x = CondNode x mempty mempty
 
-instantiateProjectConfigSkeletonFetchingCompiler :: Monad m => m (OS, Arch, CompilerInfo) -> FlagAssignment -> ProjectConfigSkeleton -> m ProjectConfig
-instantiateProjectConfigSkeletonFetchingCompiler fetch flags skel
-  | null (toListOf traverseCondTreeV skel) = pure $ fst (ignoreConditions skel)
-  | otherwise = do
-      (os, arch, impl) <- fetch
-      pure $ instantiateProjectConfigSkeletonWithCompiler os arch impl flags skel
+instantiateProjectConfigSkeletonFetchingCompiler :: (OS, Arch, CompilerInfo) -> FlagAssignment -> ProjectConfigSkeleton -> ProjectConfig
+instantiateProjectConfigSkeletonFetchingCompiler (os, arch, impl) flags skel
+  | null (toListOf traverseCondTreeV skel) = fst (ignoreConditions skel)
+  | otherwise = instantiateProjectConfigSkeletonWithCompiler os arch impl flags skel
 
 instantiateProjectConfigSkeletonWithCompiler :: OS -> Arch -> CompilerInfo -> FlagAssignment -> ProjectConfigSkeleton -> ProjectConfig
 instantiateProjectConfigSkeletonWithCompiler os arch impl _flags skel = go $ mapTreeConds (fst . simplifyWithSysParams os arch impl) skel

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

@@ -381,17 +381,16 @@ rebuildProjectConfig
         $ do
           liftIO $ info verbosity "Project settings changed, reconfiguring..."
           projectConfigSkeleton <- phaseReadProjectConfig
-          let fetchCompiler = do
-                -- have to create the cache directory before configuring the compiler
-                liftIO $ createDirectoryIfMissingVerbose verbosity True distProjectCacheDirectory
-                (compiler, Platform arch os, _) <- configureCompiler verbosity distDirLayout (fst (PD.ignoreConditions projectConfigSkeleton) <> cliConfig)
-                pure (os, arch, compilerInfo compiler)
 
-          projectConfig <- instantiateProjectConfigSkeletonFetchingCompiler fetchCompiler mempty projectConfigSkeleton
+          -- have to create the cache directory before configuring the compiler
+          liftIO $ createDirectoryIfMissingVerbose verbosity True distProjectCacheDirectory
+          (compiler, Platform arch os, _) <- configureCompiler verbosity distDirLayout (fst (PD.ignoreConditions projectConfigSkeleton) <> cliConfig)
+
+          let projectConfig = instantiateProjectConfigSkeletonFetchingCompiler (os, arch, compilerInfo compiler) mempty projectConfigSkeleton
           when (projectConfigDistDir (projectConfigShared $ projectConfig) /= NoFlag) $
             liftIO $
               warn verbosity "The builddir option is not supported in project and config files. It will be ignored."
-          localPackages <- phaseReadLocalPackages (projectConfig <> cliConfig)
+          localPackages <- phaseReadLocalPackages compiler (projectConfig <> cliConfig)
           return (projectConfig, localPackages)
 
     sequence_
@@ -423,9 +422,11 @@ rebuildProjectConfig
       -- NOTE: These are all packages mentioned in the project configuration.
       -- Whether or not they will be considered local to the project will be decided by `shouldBeLocal`.
       phaseReadLocalPackages
-        :: ProjectConfig
+        :: Compiler
+        -> ProjectConfig
         -> Rebuild [PackageSpecifier UnresolvedSourcePackage]
       phaseReadLocalPackages
+        compiler
         projectConfig@ProjectConfig
           { projectConfigShared
           , projectConfigBuildOnly
@@ -440,6 +441,7 @@ rebuildProjectConfig
           fetchAndReadSourcePackages
             verbosity
             distDirLayout
+            compiler
             projectConfigShared
             projectConfigBuildOnly
             pkgLocations

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

@@ -65,12 +65,12 @@ import Prelude ()
 
 import Distribution.Client.FileMonitor
 import Distribution.Client.Glob hiding (matchFileGlob)
+import Distribution.Client.JobControl
 import qualified Distribution.Client.Glob as Glob (matchFileGlob)
 import Distribution.Simple.PreProcess.Types (Suffix (..))
 
 import Distribution.Simple.Utils (debug)
 
-import Control.Concurrent.Async
 import Control.Concurrent.MVar (MVar, modifyMVar, newMVar)
 import Control.Monad
 import Control.Monad.Reader as Reader
@@ -130,7 +130,7 @@ rerunIfChanged verbosity monitor key action = do
   -- rerunIfChanged is implemented in terms of rerunConcurrentlyIfChanged, but
   -- nothing concurrent will happen since the list of concurrent actions has a
   -- single value that will be waited for alone.
-  rerunConcurrentlyIfChanged verbosity [(monitor, key, action)] >>= \case
+  rerunConcurrentlyIfChanged verbosity newSerialJobControl [(monitor, key, action)] >>= \case
     [x] -> return x
     _   -> error "rerunIfChanged: impossible!"
 
@@ -140,9 +140,10 @@ rerunIfChanged verbosity monitor key action = do
 rerunConcurrentlyIfChanged
   :: (Binary a, Structured a, Binary b, Structured b)
   => Verbosity
+  -> IO (JobControl IO (b, [MonitorFilePath]))
   -> [(FileMonitor a b, a, Rebuild b)]
   -> Rebuild [b]
-rerunConcurrentlyIfChanged verbosity triples = do
+rerunConcurrentlyIfChanged verbosity mkJobControl triples = do
   rootDir <- askRoot
   dacts <- forM triples $ \(monitor, key, action) -> do
     let monitorName = takeFileName (fileMonitorCacheFile monitor)
@@ -175,10 +176,9 @@ rerunConcurrentlyIfChanged verbosity triples = do
             result
           return (result, files)
 
-  (results, files) <-
-    liftIO $
-    unzip <$>
-    mapConcurrently id dacts
+  (results, files) <- liftIO $
+    withJobControl mkJobControl $ \jobControl -> do
+      unzip <$> mapConcurrentWithJobs jobControl id dacts
   monitorFiles (concat files)
   return results
   where

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

@@ -381,7 +381,7 @@ withContextAndSelectors noTargets kind flags@NixStyleFlags{..} targetStrings glo
           createDirectoryIfMissingVerbose verbosity True (distProjectCacheDirectory $ distDirLayout ctx)
           (compiler, platform@(Platform arch os), _) <- runRebuild projectRoot $ configureCompiler verbosity (distDirLayout ctx) (fst (ignoreConditions projectCfgSkeleton) <> projectConfig ctx)
 
-          projectCfg <- instantiateProjectConfigSkeletonFetchingCompiler (pure (os, arch, compilerInfo compiler)) mempty projectCfgSkeleton
+          let projectCfg = instantiateProjectConfigSkeletonFetchingCompiler (os, arch, compilerInfo compiler) mempty projectCfgSkeleton
 
           let ctx' = ctx & lProjectConfig %~ (<> projectCfg)