fix(downloads): honor the RUSTUP_CONCURRENT_DOWNLOADS by always having "n" concurrent downloads

Using this environment variable with a value between 2 and 5 would mean
that concurrency was not being totally maximized as there could be less
than n futures running at a some point in time.
See:
https://docs.rs/futures-util/0.3.31/futures_util/stream/trait.StreamExt.html#method.buffered

Adding a semaphore allows US to control how many futures are running at
a time, fixing this problem and ensuring that the env var is always
honored and the downloads are maximizing concurrency.

Author: FranciscoTGouveia

src/cli/rustup_mode.rs

@@ -6,6 +6,7 @@ use std::{
     path::{Path, PathBuf},
     process::ExitStatus,
     str::FromStr,
+    sync::Arc,
     time::Duration,
 };
 
@@ -16,6 +17,7 @@ use console::style;
 use futures_util::stream::StreamExt;
 use indicatif::{MultiProgress, ProgressBar, ProgressStyle};
 use itertools::Itertools;
+use tokio::sync::Semaphore;
 use tracing::{info, trace, warn};
 use tracing_subscriber::{EnvFilter, Registry, reload::Handle};
 
@@ -799,13 +801,15 @@ 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 = cfg.process.concurrent_downloads().unwrap_or(channels.len());
+    let channels_len = 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 {
         let multi_progress_bars =
             MultiProgress::with_draw_target(cfg.process.progress_draw_target());
+        let semaphore = Arc::new(Semaphore::new(num_channels));
         let channels = tokio_stream::iter(channels.into_iter()).map(|(name, distributable)| {
             let pb = multi_progress_bars.add(ProgressBar::new(1));
             pb.set_style(
@@ -815,7 +819,10 @@ async fn check_updates(cfg: &Cfg<'_>, opts: CheckOpts) -> Result<utils::ExitCode
             );
             pb.set_message(format!("{name}"));
             pb.enable_steady_tick(Duration::from_millis(100));
+
+            let sem = semaphore.clone();
             async move {
+                let _permit = sem.acquire().await.unwrap();
                 let current_version = distributable.show_version()?;
                 let dist_version = distributable.show_dist_version().await?;
                 let mut update_a = false;
@@ -867,7 +874,7 @@ async fn check_updates(cfg: &Cfg<'_>, opts: CheckOpts) -> Result<utils::ExitCode
         // `indicatif`.
         let channels = if !multi_progress_bars.is_hidden() {
             channels
-                .buffer_unordered(num_channels)
+                .buffer_unordered(channels_len)
                 .collect::<Vec<_>>()
                 .await
         } else {

src/dist/manifestation.rs

@@ -8,6 +8,8 @@ use std::path::Path;
 
 use anyhow::{Context, Result, anyhow, bail};
 use futures_util::stream::StreamExt;
+use std::sync::Arc;
+use tokio::sync::Semaphore;
 use tracing::info;
 
 use crate::dist::component::{
@@ -154,10 +156,11 @@ 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());
+            .unwrap_or(components_len);
 
         const DEFAULT_MAX_RETRIES: usize = 3;
         let max_retries: usize = download_cfg
@@ -177,23 +180,29 @@ impl Manifestation {
             ));
         }
 
+        let semaphore = Arc::new(Semaphore::new(num_channels));
         let component_stream =
             tokio_stream::iter(components.into_iter()).map(|(component, format, url, hash)| {
-                self.download_component(
-                    component,
-                    format,
-                    url,
-                    hash,
-                    altered,
-                    tmp_cx,
-                    download_cfg,
-                    max_retries,
-                    new_manifest,
-                )
+                let sem = semaphore.clone();
+                async move {
+                    let _permit = sem.acquire().await.unwrap();
+                    self.download_component(
+                        component,
+                        format,
+                        url,
+                        hash,
+                        altered,
+                        tmp_cx,
+                        download_cfg,
+                        max_retries,
+                        new_manifest,
+                    )
+                    .await
+                }
             });
         if num_channels > 0 {
             let results = component_stream
-                .buffered(num_channels)
+                .buffered(components_len)
                 .collect::<Vec<_>>()
                 .await;
             for result in results {