Git concurrency limit (#420)

src/steps/git.rs

@@ -1,19 +1,22 @@
+use std::collections::HashSet;
+use std::io;
+use std::path::{Path, PathBuf};
+use std::process::{Command, Output};
+
+use anyhow::Result;
+use console::style;
+use futures::stream::{iter, FuturesUnordered};
+use futures::StreamExt;
+use glob::{glob_with, MatchOptions};
+use log::{debug, error};
+use tokio::process::Command as AsyncCommand;
+use tokio::runtime;
+
 use crate::error::SkipStep;
 use crate::execution_context::ExecutionContext;
 use crate::executor::{CommandExt, RunType};
 use crate::terminal::print_separator;
 use crate::utils::{which, PathExt};
-use anyhow::Result;
-use console::style;
-use futures::future::try_join_all;
-use glob::{glob_with, MatchOptions};
-use log::{debug, error};
-use std::collections::HashSet;
-use std::io;
-use std::path::{Path, PathBuf};
-use std::process::{Command, Output};
-use tokio::process::Command as AsyncCommand;
-use tokio::runtime;
 
 #[cfg(windows)]
 static PATH_PREFIX: &str = "\\\\?\\";
@@ -189,7 +192,7 @@ impl Git {
             return Ok(());
         }
 
-        let futures: Vec<_> = repositories
+        let futures_iterator = repositories
             .repositories
             .iter()
             .filter(|repo| match has_remotes(git, repo) {
@@ -203,13 +206,19 @@ impl Git {
                 }
                 _ => true, // repo has remotes or command to check for remotes has failed. proceed to pull anyway.
             })
-            .map(|repo| pull_repository(repo.clone(), &git, ctx))
-            .collect();
+            .map(|repo| pull_repository(repo.clone(), &git, ctx));
+
+        let stream_of_futures = if let Some(limit) = ctx.config().git_concurrency_limit() {
+            iter(futures_iterator).buffer_unordered(limit).boxed()
+        } else {
+            futures_iterator.collect::<FuturesUnordered<_>>().boxed()
+        };
 
         let mut basic_rt = runtime::Runtime::new()?;
-        basic_rt.block_on(async { try_join_all(futures).await })?;
+        let results = basic_rt.block_on(async { stream_of_futures.collect::<Vec<Result<()>>>().await });
 
-        Ok(())
+        let error = results.into_iter().find(|r| r.is_err());
+        error.unwrap_or(Ok(()))
     }
 }