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

@@ -799,7 +799,8 @@ 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 = cfg.process.concurrent_downloads().unwrap_or(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

@@ -154,7 +154,10 @@ 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 = download_cfg
+            .process
+            .concurrent_downloads()
+            .unwrap_or(components.len());
 
         const DEFAULT_MAX_RETRIES: usize = 3;
         let max_retries: usize = download_cfg
@@ -188,9 +191,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 {

src/process.rs

@@ -2,6 +2,7 @@ use std::ffi::OsString;
 use std::fmt::Debug;
 use std::io;
 use std::io::IsTerminal;
+use std::num::NonZeroU64;
 use std::path::PathBuf;
 use std::str::FromStr;
 #[cfg(feature = "test")]
@@ -167,6 +168,15 @@ impl Process {
             _ => ProgressDrawTarget::hidden(),
         }
     }
+
+    pub fn concurrent_downloads(&self) -> Option<usize> {
+        match self.var("RUSTUP_CONCURRENT_DOWNLOADS") {
+            Ok(s) => Some(NonZeroU64::from_str(&s).context(
+                "invalid value in RUSTUP_CONCURRENT_DOWNLOADS -- must be a natural number greater than zero"
+            ).ok()?.get() as usize),
+            Err(_) => None,
+        }
+    }
 }
 
 impl home::env::Env for Process {