git: fetch multiple roots in parallel

Submit several tasks using the standard bounded executor, then wait for their completion.

Execute tasks under a common progress indicator, to make sure that all tasks are cancelled when one of them is.

The number of threads to execute is to be defined more precisely. Initially the number of cores is taken, but it doesn't seem to be obviously good enough default for a networking operation.

Make two implementations of RepoResult: one for fair per-remote result, another - for general error happening during the fetch. This could be solved by operating the pair of GitRepository & GitRemote, but the latter appeared to be less handy.

This fixes IDEA-160208

platform/util/resources/misc/registry.properties

@@ -1209,6 +1209,12 @@ git.log.fix.merge.commits.parents.order=false
 git.log.fix.merge.commits.parents.order.description=Find merge commits where a remote branch was merged into local before pushing\
    and fix parents order for such commits in order to display them more naturally in the commits graph.
 git.log.fix.merge.commits.parents.order.restartRequired=true
+git.parallel.fetch.threads=-3
+git.parallel.fetch.threads.description=Number of threads to call git fetch in parallel.\
+  Set to 1 to execute sequentially.\
+  Set to -1 to use the number of CPU cores.\
+  Set to -2 to start as many threads, as there are repositories in the project.\
+  Set to -3 to start as many threads, as there are repositories, but not more than 2 * number of cores.
 
 hg4idea.process.ignored=true
 hg4idea.process.ignored.description=Process and highlight mercurial ignored files. \

plugins/git4idea/src/git4idea/fetch/GitFetchSupportImpl.kt

@@ -20,15 +20,22 @@ import com.intellij.dvcs.MultiMessage
 import com.intellij.dvcs.MultiRootMessage
 import com.intellij.notification.NotificationType
 import com.intellij.openapi.diagnostic.logger
+import com.intellij.openapi.progress.EmptyProgressIndicator
+import com.intellij.openapi.progress.ProcessCanceledException
 import com.intellij.openapi.progress.ProgressManager
 import com.intellij.openapi.project.Project
+import com.intellij.openapi.util.registry.Registry
 import com.intellij.openapi.vcs.VcsNotifier
 import com.intellij.openapi.vcs.VcsNotifier.STANDARD_NOTIFICATION
+import com.intellij.util.concurrency.AppExecutorUtil
 import git4idea.GitUtil.*
 import git4idea.commands.Git
 import git4idea.repo.GitRemote
 import git4idea.repo.GitRemote.ORIGIN
 import git4idea.repo.GitRepository
+import java.util.concurrent.CancellationException
+import java.util.concurrent.ExecutionException
+import java.util.concurrent.Future
 import java.util.regex.Pattern
 
 private val LOG = logger<GitFetchSupportImpl>()
@@ -50,18 +57,19 @@ internal class GitFetchSupportImpl(val project: Project) : GitFetchSupport {
   }
 
   override fun fetch(repositories: Collection<GitRepository>): GitFetchResult {
-    val results = mutableMapOf<GitRepository, RepoResult>()
+    val remotesToFetch = mutableMapOf<GitRepository, GitRemote>()
     for (repository in repositories) {
       val remote = getDefaultRemoteToFetch(repository)
-      if (remote != null) {
-        val repoResult = withIndicator(repository) {
-          doFetch(repository, listOf(remote))
-        }
-        results[repository] = repoResult
-      }
+      if (remote != null) remotesToFetch[repository] = remote
       else LOG.info("No remote to fetch found in $repository")
     }
-    return resultOf(results)
+    return doFetch(remotesToFetch)
+  }
+
+  private fun doFetch(remotes: Map<GitRepository, GitRemote>): GitFetchResult {
+    val tasks = fetchInParallel(remotes)
+    val results = waitForFetchTasks(tasks)
+    return FetchResultImpl(project, results)
   }
 
   override fun fetch(repository: GitRepository, remote: GitRemote): GitFetchResult {
@@ -72,6 +80,61 @@ internal class GitFetchSupportImpl(val project: Project) : GitFetchSupport {
     return withIndicator(repository) { resultOf(mapOf(Pair(repository, doFetch(repository, remotes)))) }
   }
 
+  private fun fetchInParallel(remotes: Map<GitRepository, GitRemote>): Map<GitRepository, Future<RepoResult>> {
+    val tasks = mutableMapOf<GitRepository, Future<RepoResult>>()
+    val maxThreads = getMaxThreads(remotes.size)
+    LOG.debug("Fetching $remotes using $maxThreads threads")
+    val executor = AppExecutorUtil.createBoundedApplicationPoolExecutor("GitFetch pool", maxThreads)
+    val commonIndicator = ProgressManager.getInstance().progressIndicator ?: EmptyProgressIndicator()
+    for ((repository, remote) in remotes) {
+      LOG.debug("Fetching $remote in $repository")
+      val task: Future<RepoResult> = executor.submit<RepoResult> {
+        commonIndicator.checkCanceled()
+        lateinit var result: RepoResult
+        ProgressManager.getInstance().executeProcessUnderProgress({
+          commonIndicator.checkCanceled()
+          val fetchResult = doFetch(repository, remote)
+          result = resultOf(remote, fetchResult)
+        }, commonIndicator)
+        result
+      }
+      tasks[repository] = task
+    }
+    return tasks
+  }
+
+  private fun getMaxThreads(numberOfRemotes: Int): Int {
+    val maxThreads = Registry.intValue("git.parallel.fetch.threads")
+    return when {
+      maxThreads > 0 -> maxThreads
+      maxThreads == -1 -> Runtime.getRuntime().availableProcessors()
+      maxThreads == -2 -> numberOfRemotes
+      maxThreads == -3 -> Math.min(numberOfRemotes, Runtime.getRuntime().availableProcessors() * 2)
+      else -> 1
+    }
+  }
+
+  private fun waitForFetchTasks(tasks: Map<GitRepository, Future<RepoResult>>): Map<GitRepository, RepoResult> {
+    val results = mutableMapOf<GitRepository, RepoResult>()
+    for ((repository, task) in tasks) {
+      try {
+        results[repository] = task.get()
+      }
+      catch (e: CancellationException) {
+        throw ProcessCanceledException(e)
+      }
+      catch (e: InterruptedException) {
+        throw ProcessCanceledException(e)
+      }
+      catch (e: ExecutionException) {
+        if (e.cause is ProcessCanceledException) throw e.cause as ProcessCanceledException
+        results[repository] = ErrorRepoResult(e.message ?: "")
+        LOG.error(e)
+      }
+    }
+    return results
+  }
+
   private fun <T> withIndicator(repository: GitRepository, operation: () -> T): T {
     return withIndicator(getProgressTitle(repository), operation)
   }
@@ -96,7 +159,7 @@ internal class GitFetchSupportImpl(val project: Project) : GitFetchSupport {
     for (remote in remotes) {
       results[remote] = doFetch(repository, remote)
     }
-    return RepoResult(results)
+    return RepoResultPerRemote(results)
   }
 
   private fun doFetch(repository: GitRepository, remote: GitRemote): SingleRemoteResult {
@@ -111,9 +174,25 @@ internal class GitFetchSupportImpl(val project: Project) : GitFetchSupport {
     return if (matcher.matches()) matcher.group(1) else null
   }
 
+  private fun resultOf(remote: GitRemote, remoteResult: SingleRemoteResult): RepoResult {
+    return RepoResultPerRemote(mapOf(Pair(remote, remoteResult)))
+  }
+
   private fun resultOf(results: Map<GitRepository, RepoResult>) = FetchResultImpl(project, results)
 
-  private class RepoResult(val results: Map<GitRemote, SingleRemoteResult>) {
+  private interface RepoResult {
+    fun totallySuccessful(): Boolean
+    fun error(): String?
+    fun prunedRefs(): String
+  }
+
+  private class ErrorRepoResult(val error: String) : RepoResult {
+    override fun totallySuccessful() = false
+    override fun error() = error
+    override fun prunedRefs() = ""
+  }
+
+  private class RepoResultPerRemote(val results: Map<GitRemote, SingleRemoteResult>) : RepoResult {
     /*
        For simplicity, remote and repository results are merged separately.
        It means that they are not merged, if two repositories have two remotes,
@@ -121,9 +200,9 @@ internal class GitFetchSupportImpl(val project: Project) : GitFetchSupport {
        Such cases are rare, and can be handled when actual problem is reported.
      */
 
-    fun totallySuccessful() = results.values.all { it.success() }
+    override fun totallySuccessful() = results.values.all { it.success() }
 
-    fun error(): String? {
+    override fun error(): String? {
       val errorMessage = multiRemoteMessage()
       for ((remote, result) in results) {
         if (result.error != null) errorMessage.append(remote, result.error)
@@ -131,7 +210,7 @@ internal class GitFetchSupportImpl(val project: Project) : GitFetchSupport {
       return errorMessage.asString()
     }
 
-    fun prunedRefs(): String {
+    override fun prunedRefs(): String {
       val prunedRefs = multiRemoteMessage()
       for ((remote, result) in results) {
         if (result.prunedRefs.isNotEmpty()) prunedRefs.append(remote, result.prunedRefs.joinToString("\n"))