feat(downloads): introduce RUSTUP_CONCURRENT_DOWNLOADS to control concurrency

Author: FranciscoTGouveia

doc/user-guide/src/environment-variables.md

@@ -69,6 +69,9 @@
 - `RUSTUP_DOWNLOAD_TIMEOUT` *unstable* (default: 180). Allows to override the default
   timeout (in seconds) for downloading components.
 
+- `RUSTUP_CONCURRENT_DOWNLOADS` *unstable* (default: the number of components to download). Controls the number of
+  downloads made concurrently.
+
 [directive syntax]: https://docs.rs/tracing-subscriber/latest/tracing_subscriber/filter/struct.EnvFilter.html#directives
 [dc]: https://docs.docker.com/storage/storagedriver/overlayfs-driver/#modifying-files-or-directories
 [override]: overrides.md

src/cli/rustup_mode.rs

@@ -3,6 +3,7 @@ use std::{
     env::consts::EXE_SUFFIX,
     fmt,
     io::{self, Write},
+    num::NonZeroU64,
     path::{Path, PathBuf},
     process::ExitStatus,
     str::FromStr,
@@ -799,7 +800,13 @@ async fn check_updates(cfg: &Cfg<'_>, opts: CheckOpts) -> Result<utils::ExitCode
     let use_colors = matches!(t.color_choice(), ColorChoice::Auto | ColorChoice::Always);
     let mut update_available = false;
     let channels = cfg.list_channels()?;
-    let num_channels = channels.len();
+    let num_channels = match cfg.process.var("RUSTUP_CONCURRENT_DOWNLOADS") {
+        Ok(s) => NonZeroU64::from_str(&s).context(
+            "invalid value in RUSTUP_CONCURRENT_DOWNLOADS -- must be a natural number greater than zero"
+        )?.get() as usize,
+        Err(_) => 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 {

src/dist/manifestation.rs

@@ -4,7 +4,9 @@
 #[cfg(test)]
 mod tests;
 
+use std::num::NonZeroU64;
 use std::path::Path;
+use std::str::FromStr;
 
 use anyhow::{Context, Result, anyhow, bail};
 use futures_util::stream::StreamExt;
@@ -154,7 +156,12 @@ impl Manifestation {
         let mut things_to_install: Vec<(Component, CompressionKind, File)> = Vec::new();
         let mut things_downloaded: Vec<String> = Vec::new();
         let components = update.components_urls_and_hashes(new_manifest)?;
-        let components_len = components.len();
+        let num_channels = match download_cfg.process.var("RUSTUP_CONCURRENT_DOWNLOADS") {
+            Ok(s) => NonZeroU64::from_str(&s).context(
+                "invalid value in RUSTUP_CONCURRENT_DOWNLOADS -- must be a natural number greater than zero"
+            )?.get() as usize,
+            Err(_) => components.len(),
+        };
 
         const DEFAULT_MAX_RETRIES: usize = 3;
         let max_retries: usize = download_cfg
@@ -188,9 +195,9 @@ impl Manifestation {
                     new_manifest,
                 )
             });
-        if components_len > 0 {
+        if num_channels > 0 {
             let results = component_stream
-                .buffered(components_len)
+                .buffered(num_channels)
                 .collect::<Vec<_>>()
                 .await;
             for result in results {