Implement git pulling with Tokio (#408)

Cargo.lock

@@ -39,6 +39,12 @@ version = "1.0.26"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "7825f6833612eb2414095684fcf6c635becf3ce97fe48cf6421321e93bfbd53c"
 
+[[package]]
+name = "arc-swap"
+version = "0.4.6"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "b585a98a234c46fc563103e9278c9391fde1f4e6850334da895d27edb9580f62"
+
 [[package]]
 name = "arrayref"
 version = "0.3.6"
@@ -488,6 +494,21 @@ version = "0.3.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "3dcaa9ae7725d12cdb85b3ad99a434db70b468c09ded17e012d86b5c1010f7a7"
 
+[[package]]
+name = "futures"
+version = "0.3.4"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "5c329ae8753502fb44ae4fc2b622fa2a94652c41e795143765ba0927f92ab780"
+dependencies = [
+ "futures-channel",
+ "futures-core",
+ "futures-executor",
+ "futures-io",
+ "futures-sink",
+ "futures-task",
+ "futures-util",
+]
+
 [[package]]
 name = "futures-channel"
 version = "0.3.4"
@@ -495,6 +516,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "f0c77d04ce8edd9cb903932b608268b3fffec4163dc053b3b402bf47eac1f1a8"
 dependencies = [
  "futures-core",
+ "futures-sink",
 ]
 
 [[package]]
@@ -503,6 +525,17 @@ version = "0.3.4"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "f25592f769825e89b92358db00d26f965761e094951ac44d3663ef25b7ac464a"
 
+[[package]]
+name = "futures-executor"
+version = "0.3.4"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "f674f3e1bcb15b37284a90cedf55afdba482ab061c407a9c0ebbd0f3109741ba"
+dependencies = [
+ "futures-core",
+ "futures-task",
+ "futures-util",
+]
+
 [[package]]
 name = "futures-io"
 version = "0.3.4"
@@ -539,9 +572,11 @@ version = "0.3.4"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "22766cf25d64306bedf0384da004d05c9974ab104fcc4528f1236181c18004c5"
 dependencies = [
+ "futures-channel",
  "futures-core",
  "futures-io",
  "futures-macro",
+ "futures-sink",
  "futures-task",
  "memchr",
  "pin-utils",
@@ -895,12 +930,35 @@ dependencies = [
  "kernel32-sys",
  "libc",
  "log",
- "miow",
+ "miow 0.2.1",
  "net2",
  "slab",
  "winapi 0.2.8",
 ]
 
+[[package]]
+name = "mio-named-pipes"
+version = "0.1.6"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "f5e374eff525ce1c5b7687c4cef63943e7686524a387933ad27ca7ec43779cb3"
+dependencies = [
+ "log",
+ "mio",
+ "miow 0.3.3",
+ "winapi 0.3.8",
+]
+
+[[package]]
+name = "mio-uds"
+version = "0.6.7"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "966257a94e196b11bb43aca423754d87429960a768de9414f3691d6957abf125"
+dependencies = [
+ "iovec",
+ "libc",
+ "mio",
+]
+
 [[package]]
 name = "miow"
 version = "0.2.1"
@@ -913,6 +971,16 @@ dependencies = [
  "ws2_32-sys",
 ]
 
+[[package]]
+name = "miow"
+version = "0.3.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "396aa0f2003d7df8395cb93e09871561ccc3e785f0acb369170e8cc74ddf9226"
+dependencies = [
+ "socket2",
+ "winapi 0.3.8",
+]
+
 [[package]]
 name = "native-tls"
 version = "0.2.4"
@@ -1600,6 +1668,16 @@ dependencies = [
  "dirs 2.0.2",
 ]
 
+[[package]]
+name = "signal-hook-registry"
+version = "1.2.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "94f478ede9f64724c5d173d7bb56099ec3e2d9fc2774aac65d34b8b890405f41"
+dependencies = [
+ "arc-swap",
+ "libc",
+]
+
 [[package]]
 name = "slab"
 version = "0.4.2"
@@ -1612,6 +1690,18 @@ version = "1.2.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "5c2fb2ec9bcd216a5b0d0ccf31ab17b5ed1d627960edff65bbe95d3ce221cefc"
 
+[[package]]
+name = "socket2"
+version = "0.3.12"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "03088793f677dce356f3ccc2edb1b314ad191ab702a5de3faf49304f7e104918"
+dependencies = [
+ "cfg-if",
+ "libc",
+ "redox_syscall",
+ "winapi 0.3.8",
+]
+
 [[package]]
 name = "spin"
 version = "0.5.2"
@@ -1804,11 +1894,16 @@ dependencies = [
  "fnv",
  "iovec",
  "lazy_static",
+ "libc",
  "memchr",
  "mio",
+ "mio-named-pipes",
+ "mio-uds",
  "num_cpus",
  "pin-project-lite",
+ "signal-hook-registry",
  "slab",
+ "winapi 0.3.8",
 ]
 
 [[package]]
@@ -1865,6 +1960,7 @@ dependencies = [
  "chrono",
  "console 0.9.2",
  "directories",
+ "futures",
  "glob",
  "lazy_static",
  "log",
@@ -1879,6 +1975,7 @@ dependencies = [
  "strum",
  "tempfile",
  "thiserror",
+ "tokio",
  "toml",
  "walkdir",
  "which",

Cargo.toml

@@ -30,6 +30,8 @@ thiserror = "1.0.9"
 anyhow = "1.0.25"
 tempfile = "3.1.0"
 cfg-if = "0.1.10"
+tokio = { version = "0.2", features = ["rt-core", "process"] }
+futures = "0.3"
 
 [target.'cfg(target_os = "macos")'.dependencies]
 notify-rust = "3.6.3"

src/error.rs

@@ -13,9 +13,6 @@ pub enum TopgradeError {
     #[error("Unknown Linux Distribution")]
     #[cfg(target_os = "linux")]
     UnknownLinuxDistribution,
-
-    #[error("A pull action was failed")]
-    PullFailed,
 }
 
 #[derive(Error, Debug)]

src/steps/git.rs

@@ -1,19 +1,19 @@
-use crate::error::{SkipStep, TopgradeError};
+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::io::Read;
 use std::path::{Path, PathBuf};
-use std::process::{Child, Command, Stdio};
+use std::process::{Command, Output};
-use std::thread::sleep;
+use tokio::process::Command as AsyncCommand;
-use std::time::Duration;
+use tokio::runtime;
 
 #[cfg(windows)]
 static PATH_PREFIX: &str = "\\\\?\\";
@@ -23,18 +23,77 @@ pub struct Git {
     git: Option<PathBuf>,
 }
 
-struct PullProcess {
-    child: Child,
-    repo: String,
-    before_revision: Option<String>,
-}
-
 pub struct Repositories<'a> {
     git: &'a Git,
     repositories: HashSet<String>,
     glob_match_options: MatchOptions,
 }
 
+fn check_output(output: Output) -> std::result::Result<(), String> {
+    if !(output.status.success()) {
+        let stderr = String::from_utf8(output.stderr).unwrap();
+        Err(stderr)
+    } else {
+        Ok(())
+    }
+}
+
+async fn pull_repository(repo: String, git: &PathBuf, ctx: &ExecutionContext<'_>) -> Result<()> {
+    let path = repo.to_string();
+    let before_revision = get_head_revision(git, &repo);
+
+    println!("{} {}", style("Pulling").cyan().bold(), path);
+
+    let mut command = AsyncCommand::new(git);
+
+    command.args(&["pull", "--ff-only"]).current_dir(&repo);
+
+    if let Some(extra_arguments) = ctx.config().git_arguments() {
+        command.args(extra_arguments.split_whitespace());
+    }
+
+    let pull_output = command.output().await?;
+    let submodule_output = AsyncCommand::new(git)
+        .args(&["submodule", "update", "--recursive"])
+        .current_dir(&repo)
+        .output()
+        .await?;
+    let result = check_output(pull_output).and_then(|_| check_output(submodule_output));
+
+    if let Err(message) = result {
+        println!("{} pulling {}", style("Failed").red().bold(), &repo);
+        print!("{}", message);
+    } else {
+        let after_revision = get_head_revision(&git, &repo);
+
+        match (&before_revision, &after_revision) {
+            (Some(before), Some(after)) if before != after => {
+                println!("{} {}:", style("Changed").yellow().bold(), &repo);
+
+                Command::new(&git)
+                    .current_dir(&repo)
+                    .args(&[
+                        "--no-pager",
+                        "log",
+                        "--no-decorate",
+                        "--oneline",
+                        &format!("{}..{}", before, after),
+                    ])
+                    .spawn()
+                    .unwrap()
+                    .wait()
+                    .unwrap();
+                println!();
+            }
+            _ => {
+                println!("{} {}", style("Up-to-date").green().bold(), &repo);
+            }
+        }
+    }
+
+    Ok(())
+}
+
 fn get_head_revision(git: &Path, repo: &str) -> Option<String> {
     Command::new(git)
         .args(&["rev-parse", "HEAD"])
@@ -130,7 +189,7 @@ impl Git {
             return Ok(());
         }
 
-        let mut processes: Vec<_> = repositories
+        let futures: Vec<_> = repositories
             .repositories
             .iter()
             .filter(|repo| match has_remotes(git, repo) {
@@ -144,87 +203,13 @@ impl Git {
                 }
                 _ => true, // repo has remotes or command to check for remotes has failed. proceed to pull anyway.
             })
-            .filter_map(|repo| {
+            .map(|repo| pull_repository(repo.clone(), &git, ctx))
-                let repo = repo.clone();
-                let path = repo.to_string();
-                let before_revision = get_head_revision(git, &repo);
-
-                println!("{} {}", style("Pulling").cyan().bold(), path);
-
-                let mut command = Command::new(git);
-
-                command.args(&["pull", "--ff-only"]).current_dir(&repo);
-
-                if let Some(extra_arguments) = ctx.config().git_arguments() {
-                    command.args(extra_arguments.split_whitespace());
-                }
-
-                command
-                    .stdout(Stdio::null())
-                    .stderr(Stdio::piped())
-                    .spawn()
-                    .map(|child| PullProcess {
-                        child,
-                        repo,
-                        before_revision,
-                    })
-                    .ok()
-            })
             .collect();
 
-        let mut success = true;
+        let mut basic_rt = runtime::Runtime::new()?;
-        while !processes.is_empty() {
+        basic_rt.block_on(async { try_join_all(futures).await })?;
-            let mut remaining_processes = Vec::<PullProcess>::with_capacity(processes.len());
-            for mut p in processes {
-                if let Some(status) = p.child.try_wait().unwrap() {
-                    if status.success() {
-                        let after_revision = get_head_revision(&git, &p.repo);
 
-                        match (&p.before_revision, &after_revision) {
+        Ok(())
-                            (Some(before), Some(after)) if before != after => {
-                                println!("{} {}:", style("Changed").yellow().bold(), &p.repo);
-
-                                Command::new(&git)
-                                    .current_dir(&p.repo)
-                                    .args(&[
-                                        "--no-pager",
-                                        "log",
-                                        "--no-decorate",
-                                        "--oneline",
-                                        &format!("{}..{}", before, after),
-                                    ])
-                                    .spawn()
-                                    .unwrap()
-                                    .wait()
-                                    .unwrap();
-                                println!();
-                            }
-                            _ => {
-                                println!("{} {}", style("Up-to-date").green().bold(), &p.repo);
-                            }
-                        }
-                    } else {
-                        success = false;
-                        println!("{} pulling {}", style("Failed").red().bold(), &p.repo);
-                        let mut stderr = String::new();
-                        if p.child.stderr.unwrap().read_to_string(&mut stderr).is_ok() {
-                            print!("{}", stderr);
-                        }
-                    }
-                } else {
-                    remaining_processes.push(p);
-                }
-            }
-
-            processes = remaining_processes;
-            sleep(Duration::from_millis(200));
-        }
-
-        if !success {
-            Err(TopgradeError::PullFailed.into())
-        } else {
-            Ok(())
-        }
     }
 }