feat(updates): check for updates concurrently

This change implied the introduction of the `futures_util` crate.
Feedback is greately appreciated on the above.

To ensure that there are no hangs, a special "if" was needed to ensure
that we never try running `.buffered(0)`. Otherwise, this would cause an
hang (kudos to rami3l for helping me out with this).

To report progress in a concurrent way whilst maintaining order, the
`indicatf` crate was used.
Note that, if we are not in a *tty* the progress wil be hidden and will
instead be printed after every channels is checked for updates.

NB: since we are using `indicatif` to display progress, when running in
a *tty*, we can use `buffer_unordered` instead of `buffered`, as
`indicatif` already handles the output in the correct order.
A quick benchmark (100 runs + 10 warmpu runs in a 50Mbps connection)
showed that this small change yields a **1.2x** speedup.

Author: FranciscoTGouveia

Cargo.lock

@@ -53,6 +53,7 @@ enum-map = "2.5.0"
 env_proxy = { version = "0.4.1", optional = true }
 flate2 = { version = "1.1.1", default-features = false, features = ["zlib-rs"] }
 fs_at = "0.2.1"
+futures-util = "0.3.31"
 git-testament = "0.2"
 home = "0.5.4"
 indicatif = "0.18"

Cargo.toml

@@ -6,11 +6,15 @@ use std::{
     path::{Path, PathBuf},
     process::ExitStatus,
     str::FromStr,
+    time::Duration,
 };
 
 use anyhow::{Context, Error, Result, anyhow};
 use clap::{Args, CommandFactory, Parser, Subcommand, ValueEnum, builder::PossibleValue};
 use clap_complete::Shell;
+use console::style;
+use futures_util::stream::StreamExt;
+use indicatif::{MultiProgress, ProgressBar, ProgressDrawTarget, ProgressStyle};
 use itertools::Itertools;
 use tracing::{info, trace, warn};
 use tracing_subscriber::{EnvFilter, Registry, reload::Handle};
@@ -34,7 +38,7 @@ use crate::{
     install::{InstallMethod, UpdateStatus},
     process::{
         Process,
-        terminalsource::{self, ColorableTerminal},
+        terminalsource::{self, ColorChoice, ColorableTerminal},
     },
     toolchain::{
         CustomToolchainName, DistributableToolchain, LocalToolchainName,
@@ -792,45 +796,99 @@ async fn default_(
 }
 
 async fn check_updates(cfg: &Cfg<'_>, opts: CheckOpts) -> Result<utils::ExitCode> {
+    let t = cfg.process.stdout().terminal(cfg.process);
+    let is_a_tty = t.is_a_tty();
+    let use_colors = matches!(t.color_choice(), ColorChoice::Auto | ColorChoice::Always);
     let mut update_available = false;
-
-    let mut t = cfg.process.stdout().terminal(cfg.process);
     let channels = cfg.list_channels()?;
-
-    for channel in channels {
-        let (name, distributable) = channel;
-        let current_version = distributable.show_version()?;
-        let dist_version = distributable.show_dist_version().await?;
-        let _ = t.attr(terminalsource::Attr::Bold);
-        write!(t.lock(), "{name} - ")?;
-        match (current_version, dist_version) {
-            (None, None) => {
-                let _ = t.fg(terminalsource::Color::Red);
-                writeln!(t.lock(), "Cannot identify installed or update versions")?;
-            }
-            (Some(cv), None) => {
-                let _ = t.fg(terminalsource::Color::Green);
-                write!(t.lock(), "Up to date")?;
-                let _ = t.reset();
-                writeln!(t.lock(), " : {cv}")?;
+    let num_channels = channels.len();
+    // Ensure that `.buffered()` is never called with 0 as this will cause a hang.
+    // See: https://github.com/rust-lang/futures-rs/pull/1194#discussion_r209501774
+    if num_channels > 0 {
+        let multi_progress_bars = if is_a_tty {
+            MultiProgress::with_draw_target(ProgressDrawTarget::term_like(Box::new(t)))
+        } else {
+            MultiProgress::with_draw_target(ProgressDrawTarget::hidden())
+        };
+        let channels = tokio_stream::iter(channels.into_iter()).map(|(name, distributable)| {
+            let pb = multi_progress_bars.add(ProgressBar::new(1));
+            pb.set_style(
+                ProgressStyle::with_template("{msg:.bold} - Checking... {spinner:.green}")
+                    .unwrap()
+                    .tick_chars("⠁⠂⠄⡀⢀⠠⠐⠈ "),
+            );
+            pb.set_message(format!("{name}"));
+            pb.enable_steady_tick(Duration::from_millis(100));
+            async move {
+                let current_version = distributable.show_version()?;
+                let dist_version = distributable.show_dist_version().await?;
+                let mut update_a = false;
+
+                let mut styled_name = style(format!("{name} - "));
+                if use_colors {
+                    styled_name = styled_name.bold();
+                }
+                let message = match (current_version, dist_version) {
+                    (None, None) => {
+                        let mut m = style("Cannot identify installed or update versions");
+                        if use_colors {
+                            m = m.red().bold();
+                        }
+                        format!("{styled_name}{m}")
+                    }
+                    (Some(cv), None) => {
+                        let mut m = style("Up to date");
+                        if use_colors {
+                            m = m.green().bold();
+                        }
+                        format!("{styled_name}{m} : {cv}")
+                    }
+                    (Some(cv), Some(dv)) => {
+                        let mut m = style("Update available");
+                        if use_colors {
+                            m = m.yellow().bold();
+                        }
+                        update_a = true;
+                        format!("{styled_name}{m} : {cv} -> {dv}")
+                    }
+                    (None, Some(dv)) => {
+                        let mut m = style("Update available");
+                        if use_colors {
+                            m = m.yellow().bold();
+                        }
+                        update_a = true;
+                        format!("{styled_name}{m} : (Unknown version) -> {dv}")
+                    }
+                };
+                pb.set_style(ProgressStyle::with_template(message.as_str()).unwrap());
+                pb.finish();
+                Ok::<(bool, String), Error>((update_a, message))
             }
-            (Some(cv), Some(dv)) => {
+        });
+
+        // If we are running in a TTY, we can use `buffer_unordered` since
+        // displaying the output in the correct order is already handled by
+        // `indicatif`.
+        let channels = if is_a_tty {
+            channels
+                .buffer_unordered(num_channels)
+                .collect::<Vec<_>>()
+                .await
+        } else {
+            channels.buffered(num_channels).collect::<Vec<_>>().await
+        };
+
+        let t = cfg.process.stdout().terminal(cfg.process);
+        for result in channels {
+            let (update_a, message) = result?;
+            if update_a {
                 update_available = true;
-                let _ = t.fg(terminalsource::Color::Yellow);
-                write!(t.lock(), "Update available")?;
-                let _ = t.reset();
-                writeln!(t.lock(), " : {cv} -> {dv}")?;
             }
-            (None, Some(dv)) => {
-                update_available = true;
-                let _ = t.fg(terminalsource::Color::Yellow);
-                write!(t.lock(), "Update available")?;
-                let _ = t.reset();
-                writeln!(t.lock(), " : (Unknown version) -> {dv}")?;
+            if !is_a_tty {
+                writeln!(t.lock(), "{message}")?;
             }
         }
     }
-
     let self_update_mode = cfg.get_self_update_mode()?;
     // Priority: no-self-update feature > self_update_mode > no-self-update args.
     // Check for update only if rustup does **not** have the no-self-update feature,