dist: drop unnecessary Notifier layer

Author: djc

src/cli/self_update.rs

@@ -1279,7 +1279,7 @@ pub(crate) async fn prepare_update(dl_cfg: &DownloadCfg<'_>) -> Result<Option<Pa
         &download_url,
         &setup_path,
         None,
-        &dl_cfg.notifier,
+        &dl_cfg.tracker,
         dl_cfg.process,
     )
     .await?;
@@ -1305,7 +1305,7 @@ async fn get_available_rustup_version(dl_cfg: &DownloadCfg<'_>) -> Result<String
         &release_file_url,
         &release_file,
         None,
-        &dl_cfg.notifier,
+        &dl_cfg.tracker,
         dl_cfg.process,
     )
     .await?;

src/cli/self_update/windows.rs

@@ -277,7 +277,7 @@ pub(crate) async fn try_install_msvc(
         &visual_studio_url,
         &visual_studio,
         None,
-        &dl_cfg.notifier,
+        &dl_cfg.tracker,
         dl_cfg.process,
     )
     .await?;

src/diskio/mod.rs

@@ -66,7 +66,7 @@ use std::{fmt::Debug, fs::OpenOptions};
 
 use anyhow::Result;
 
-use crate::dist::download::Notifier;
+use crate::dist::download::DownloadTracker;
 use crate::process::Process;
 
 /// Carries the implementation specific data for complete file transfers into the executor.
@@ -443,13 +443,13 @@ pub(crate) fn create_dir<P: AsRef<Path>>(path: P) -> io::Result<()> {
 
 /// Get the executor for disk IO.
 pub(crate) fn get_executor<'a>(
-    notifier: Option<&'a Notifier>,
+    tracker: Option<&'a DownloadTracker>,
     ram_budget: usize,
     process: &Process,
 ) -> anyhow::Result<Box<dyn Executor + 'a>> {
     // If this gets lots of use, consider exposing via the config file.
     Ok(match process.io_thread_count()? {
         0 | 1 => Box::new(immediate::ImmediateUnpacker::new()),
-        n => Box::new(threaded::Threaded::new(notifier, n, ram_budget)),
+        n => Box::new(threaded::Threaded::new(tracker, n, ram_budget)),
     })
 }

src/diskio/threaded.rs

@@ -15,7 +15,7 @@ use sharded_slab::pool::{OwnedRef, OwnedRefMut};
 use tracing::debug;
 
 use super::{CompletedIo, Executor, Item, perform};
-use crate::dist::download::{Notification, Notifier};
+use crate::dist::download::{DownloadTracker, Notification};
 
 #[derive(Copy, Clone, Debug, Enum)]
 pub(crate) enum Bucket {
@@ -99,7 +99,7 @@ impl fmt::Debug for Pool {
 pub(crate) struct Threaded<'a> {
     n_files: Arc<AtomicUsize>,
     pool: threadpool::ThreadPool,
-    notifier: Option<&'a Notifier>,
+    tracker: Option<&'a DownloadTracker>,
     rx: Receiver<Task>,
     tx: Sender<Task>,
     vec_pools: EnumMap<Bucket, Pool>,
@@ -109,7 +109,7 @@ pub(crate) struct Threaded<'a> {
 impl<'a> Threaded<'a> {
     /// Construct a new Threaded executor.
     pub(crate) fn new(
-        notify_handler: Option<&'a Notifier>,
+        tracker: Option<&'a DownloadTracker>,
         thread_count: usize,
         ram_budget: usize,
     ) -> Self {
@@ -168,7 +168,7 @@ impl<'a> Threaded<'a> {
         Self {
             n_files: Arc::new(AtomicUsize::new(0)),
             pool,
-            notifier: notify_handler,
+            tracker,
             rx,
             tx,
             vec_pools,
@@ -261,9 +261,9 @@ impl Executor for Threaded<'_> {
         // actual handling of data today, we synthesis a data buffer and
         // pretend to have bytes to deliver.
         let mut prev_files = self.n_files.load(Ordering::Relaxed);
-        if let Some(notifier) = self.notifier {
-            notifier.handle(Notification::DownloadFinished(None));
-            notifier.handle(Notification::DownloadContentLengthReceived(
+        if let Some(tracker) = self.tracker {
+            tracker.handle(Notification::DownloadFinished(None));
+            tracker.handle(Notification::DownloadContentLengthReceived(
                 prev_files as u64,
                 None,
             ));
@@ -282,16 +282,16 @@ impl Executor for Threaded<'_> {
             prev_files = current_files;
             current_files = self.n_files.load(Ordering::Relaxed);
             let step_count = prev_files - current_files;
-            if let Some(notifier) = self.notifier {
-                notifier.handle(Notification::DownloadDataReceived(
+            if let Some(tracker) = self.tracker {
+                tracker.handle(Notification::DownloadDataReceived(
                     &buf[0..step_count],
                     None,
                 ));
             }
         }
         self.pool.join();
-        if let Some(notifier) = self.notifier {
-            notifier.handle(Notification::DownloadFinished(None));
+        if let Some(tracker) = self.tracker {
+            tracker.handle(Notification::DownloadFinished(None));
         }
         // close the feedback channel so that blocking reads on it can
         // complete. send is atomic, and we know the threads completed from the

src/dist/component/package.rs

@@ -300,7 +300,7 @@ fn unpack_without_first_dir<R: Read>(
     };
     let unpack_ram = unpack_ram(IO_CHUNK_SIZE, effective_max_ram, dl_cfg);
     let mut io_executor: Box<dyn Executor> =
-        get_executor(Some(&dl_cfg.notifier), unpack_ram, dl_cfg.process)?;
+        get_executor(Some(&dl_cfg.tracker), unpack_ram, dl_cfg.process)?;
 
     let mut directories: HashMap<PathBuf, DirStatus> = HashMap::new();
     // Path is presumed to exist. Call it a precondition.

src/dist/download.rs

@@ -23,7 +23,7 @@ const UPDATE_HASH_LEN: usize = 20;
 pub struct DownloadCfg<'a> {
     pub tmp_cx: &'a temp::Context,
     pub download_dir: &'a PathBuf,
-    pub(crate) notifier: Notifier,
+    pub(crate) tracker: DownloadTracker,
     pub process: &'a Process,
 }
 
@@ -33,7 +33,7 @@ impl<'a> DownloadCfg<'a> {
         DownloadCfg {
             tmp_cx: &cfg.tmp_cx,
             download_dir: &cfg.download_dir,
-            notifier: Notifier::new(cfg.quiet, cfg.process),
+            tracker: DownloadTracker::new(cfg.quiet, cfg.process),
             process: cfg.process,
         }
     }
@@ -47,7 +47,7 @@ impl<'a> DownloadCfg<'a> {
         let target_file = self.download_dir.join(Path::new(hash));
 
         if target_file.exists() {
-            let cached_result = file_hash(&target_file, &self.notifier)?;
+            let cached_result = file_hash(&target_file, &self.tracker)?;
             if hash == cached_result {
                 debug!("reusing previously downloaded file");
                 debug!(url = url.as_ref(), "checksum passed");
@@ -76,7 +76,7 @@ impl<'a> DownloadCfg<'a> {
             &partial_file_path,
             Some(&mut hasher),
             true,
-            &self.notifier,
+            &self.tracker,
             self.process,
         )
         .await
@@ -125,7 +125,7 @@ impl<'a> DownloadCfg<'a> {
         let hash_url = utils::parse_url(&(url.to_owned() + ".sha256"))?;
         let hash_file = self.tmp_cx.new_file()?;
 
-        download_file(&hash_url, &hash_file, None, &self.notifier, self.process).await?;
+        download_file(&hash_url, &hash_file, None, &self.tracker, self.process).await?;
 
         utils::read_file("hash", &hash_file).map(|s| s[0..64].to_owned())
     }
@@ -166,7 +166,7 @@ impl<'a> DownloadCfg<'a> {
         let file = self.tmp_cx.new_file_with_ext("", ext)?;
 
         let mut hasher = Sha256::new();
-        download_file(&url, &file, Some(&mut hasher), &self.notifier, self.process).await?;
+        download_file(&url, &file, Some(&mut hasher), &self.tracker, self.process).await?;
         let actual_hash = format!("{:x}", hasher.finalize());
 
         if hash != actual_hash {
@@ -185,22 +185,6 @@ impl<'a> DownloadCfg<'a> {
     }
 }
 
-pub(crate) struct Notifier {
-    tracker: Mutex<DownloadTracker>,
-}
-
-impl Notifier {
-    pub(crate) fn new(quiet: bool, process: &Process) -> Self {
-        Self {
-            tracker: Mutex::new(DownloadTracker::new(!quiet, process)),
-        }
-    }
-
-    pub(crate) fn handle(&self, n: Notification<'_>) {
-        self.tracker.lock().unwrap().handle_notification(&n);
-    }
-}
-
 /// Tracks download progress and displays information about it to a terminal.
 ///
 /// *not* safe for tracking concurrent downloads yet - it is basically undefined
@@ -214,7 +198,7 @@ pub(crate) struct DownloadTracker {
     /// the message "retrying download" for at least a second.
     /// Without it, the progress bar would reappear immediately, not allowing the user to
     /// correctly see the message, before the progress bar starts again.
-    file_progress_bars: HashMap<String, (ProgressBar, Option<Instant>)>,
+    file_progress_bars: Mutex<HashMap<String, (ProgressBar, Option<Instant>)>>,
 }
 
 impl DownloadTracker {
@@ -228,12 +212,12 @@ impl DownloadTracker {
 
         Self {
             multi_progress_bars,
-            file_progress_bars: HashMap::new(),
+            file_progress_bars: Mutex::new(HashMap::new()),
         }
     }
 
-    pub(crate) fn handle_notification(&mut self, n: &Notification<'_>) {
-        match *n {
+    pub(crate) fn handle(&self, n: Notification<'_>) {
+        match n {
             Notification::DownloadContentLengthReceived(content_len, url) => {
                 if let Some(url) = url {
                     self.content_length_received(content_len, url);
@@ -263,7 +247,7 @@ impl DownloadTracker {
     }
 
     /// Creates a new ProgressBar for the given component.
-    pub(crate) fn create_progress_bar(&mut self, component: String, url: String) {
+    pub(crate) fn create_progress_bar(&self, component: String, url: String) {
         let pb = ProgressBar::hidden();
         pb.set_style(
             ProgressStyle::with_template(
@@ -274,20 +258,24 @@ impl DownloadTracker {
         );
         pb.set_message(component);
         self.multi_progress_bars.add(pb.clone());
-        self.file_progress_bars.insert(url, (pb, None));
+        self.file_progress_bars
+            .lock()
+            .unwrap()
+            .insert(url, (pb, None));
     }
 
     /// Sets the length for a new ProgressBar and gives it a style.
-    pub(crate) fn content_length_received(&mut self, content_len: u64, url: &str) {
-        if let Some((pb, _)) = self.file_progress_bars.get(url) {
+    pub(crate) fn content_length_received(&self, content_len: u64, url: &str) {
+        if let Some((pb, _)) = self.file_progress_bars.lock().unwrap().get(url) {
             pb.reset();
             pb.set_length(content_len);
         }
     }
 
     /// Notifies self that data of size `len` has been received.
-    pub(crate) fn data_received(&mut self, len: usize, url: &str) {
-        let Some((pb, retry_time)) = self.file_progress_bars.get_mut(url) else {
+    pub(crate) fn data_received(&self, len: usize, url: &str) {
+        let mut map = self.file_progress_bars.lock().unwrap();
+        let Some((pb, retry_time)) = map.get_mut(url) else {
             return;
         };
         pb.inc(len as u64);
@@ -305,8 +293,9 @@ impl DownloadTracker {
     }
 
     /// Notifies self that the download has finished.
-    pub(crate) fn download_finished(&mut self, url: &str) {
-        let Some((pb, _)) = self.file_progress_bars.get(url) else {
+    pub(crate) fn download_finished(&self, url: &str) {
+        let map = self.file_progress_bars.lock().unwrap();
+        let Some((pb, _)) = map.get(url) else {
             return;
         };
         pb.set_style(
@@ -317,8 +306,9 @@ impl DownloadTracker {
     }
 
     /// Notifies self that the download has failed.
-    pub(crate) fn download_failed(&mut self, url: &str) {
-        let Some((pb, _)) = self.file_progress_bars.get(url) else {
+    pub(crate) fn download_failed(&self, url: &str) {
+        let map = self.file_progress_bars.lock().unwrap();
+        let Some((pb, _)) = map.get(url) else {
             return;
         };
         pb.set_style(
@@ -329,8 +319,9 @@ impl DownloadTracker {
     }
 
     /// Notifies self that the download is being retried.
-    pub(crate) fn retrying_download(&mut self, url: &str) {
-        let Some((pb, retry_time)) = self.file_progress_bars.get_mut(url) else {
+    pub(crate) fn retrying_download(&self, url: &str) {
+        let mut map = self.file_progress_bars.lock().unwrap();
+        let Some((pb, retry_time)) = map.get_mut(url) else {
             return;
         };
         *retry_time = Some(Instant::now());
@@ -354,9 +345,9 @@ pub(crate) enum Notification<'a> {
     DownloadFailed(&'a str),
 }
 
-fn file_hash(path: &Path, notifier: &Notifier) -> Result<String> {
+fn file_hash(path: &Path, tracker: &DownloadTracker) -> Result<String> {
     let mut hasher = Sha256::new();
-    let mut downloaded = utils::FileReaderWithProgress::new_file(path, notifier)?;
+    let mut downloaded = utils::FileReaderWithProgress::new_file(path, tracker)?;
     use std::io::Read;
     let mut buf = vec![0; 32768];
     while let Ok(n) = downloaded.read(&mut buf) {

src/dist/manifestation.rs

@@ -178,7 +178,7 @@ impl Manifestation {
         info!("downloading component(s)");
         for bin in &components {
             download_cfg
-                .notifier
+                .tracker
                 .handle(Notification::DownloadingComponent(
                     &bin.component.short_name(new_manifest),
                     &bin.binary.url,
@@ -278,7 +278,7 @@ impl Manifestation {
             }
 
             let reader =
-                utils::FileReaderWithProgress::new_file(&installer_file, &download_cfg.notifier)?;
+                utils::FileReaderWithProgress::new_file(&installer_file, &download_cfg.tracker)?;
             let package = match format {
                 CompressionKind::GZip => &TarGzPackage::new(reader, download_cfg)? as &dyn Package,
                 CompressionKind::XZ => &TarXzPackage::new(reader, download_cfg)?,
@@ -444,7 +444,7 @@ impl Manifestation {
             .replace(DEFAULT_DIST_SERVER, dl_cfg.tmp_cx.dist_server.as_str());
 
         dl_cfg
-            .notifier
+            .tracker
             .handle(Notification::DownloadingComponent("rust", &url));
 
         let dl = dl_cfg
@@ -468,7 +468,7 @@ impl Manifestation {
         }
 
         // Install all the components in the installer
-        let reader = utils::FileReaderWithProgress::new_file(&installer_file, &dl_cfg.notifier)?;
+        let reader = utils::FileReaderWithProgress::new_file(&installer_file, &dl_cfg.tracker)?;
         let package: &dyn Package = &TarGzPackage::new(reader, dl_cfg)?;
         for component in package.components() {
             tx = package.install(&self.installation, &component, None, tx)?;
@@ -750,7 +750,7 @@ impl<'a> ComponentBinary<'a> {
                     Some(RustupError::BrokenPartialFile)
                     | Some(RustupError::DownloadingFile { .. }) => {
                         download_cfg
-                            .notifier
+                            .tracker
                             .handle(Notification::RetryingDownload(url.as_str()));
                         true
                     }

src/dist/manifestation/tests.rs

@@ -16,7 +16,7 @@ use url::Url;
 use crate::{
     dist::{
         DEFAULT_DIST_SERVER, Profile, TargetTriple, ToolchainDesc,
-        download::{DownloadCfg, Notifier},
+        download::{DownloadCfg, DownloadTracker},
         manifest::{Component, Manifest},
         manifestation::{Changes, Manifestation, UpdateStatus},
         prefix::InstallPrefix,
@@ -482,7 +482,7 @@ impl TestContext {
         let dl_cfg = DownloadCfg {
             tmp_cx: &self.tmp_cx,
             download_dir: &self.download_dir,
-            notifier: Notifier::new(false, &self.tp.process),
+            tracker: DownloadTracker::new(false, &self.tp.process),
             process: &self.tp.process,
         };
 
@@ -493,7 +493,7 @@ impl TestContext {
             &manifest_url,
             &manifest_file,
             None,
-            &dl_cfg.notifier,
+            &dl_cfg.tracker,
             dl_cfg.process,
         )
         .await?;