dist: call DownloadTracker methods directly

Author: djc

src/diskio/mod.rs

@@ -66,7 +66,6 @@ use std::{fmt::Debug, fs::OpenOptions};
 
 use anyhow::Result;
 
-use crate::dist::download::DownloadTracker;
 use crate::process::Process;
 
 /// Carries the implementation specific data for complete file transfers into the executor.
@@ -443,13 +442,12 @@ pub(crate) fn create_dir<P: AsRef<Path>>(path: P) -> io::Result<()> {
 
 /// Get the executor for disk IO.
 pub(crate) fn get_executor<'a>(
-    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(tracker, n, ram_budget)),
+        n => Box::new(threaded::Threaded::new(n, ram_budget)),
     })
 }

src/diskio/test.rs

@@ -24,7 +24,7 @@ fn test_incremental_file(io_threads: &str) -> Result<()> {
 
     let mut written = 0;
     let mut file_finished = false;
-    let mut io_executor: Box<dyn Executor> = get_executor(None, 32 * 1024 * 1024, &tp.process)?;
+    let mut io_executor: Box<dyn Executor> = get_executor(32 * 1024 * 1024, &tp.process)?;
     let (item, mut sender) = Item::write_file_segmented(
         work_dir.path().join("scratch"),
         0o666,
@@ -90,7 +90,7 @@ fn test_complete_file(io_threads: &str) -> Result<()> {
     vars.insert("RUSTUP_IO_THREADS".to_string(), io_threads.to_string());
     let tp = TestProcess::with_vars(vars);
 
-    let mut io_executor: Box<dyn Executor> = get_executor(None, 32 * 1024 * 1024, &tp.process)?;
+    let mut io_executor: Box<dyn Executor> = get_executor(32 * 1024 * 1024, &tp.process)?;
     let mut chunk = io_executor.get_buffer(10);
     chunk.extend(b"0123456789");
     assert_eq!(chunk.len(), 10);

src/diskio/threaded.rs

@@ -15,7 +15,6 @@ use sharded_slab::pool::{OwnedRef, OwnedRefMut};
 use tracing::debug;
 
 use super::{CompletedIo, Executor, Item, perform};
-use crate::dist::download::{DownloadTracker, Notification};
 
 #[derive(Copy, Clone, Debug, Enum)]
 pub(crate) enum Bucket {
@@ -96,23 +95,18 @@ impl fmt::Debug for Pool {
     }
 }
 
-pub(crate) struct Threaded<'a> {
+pub(crate) struct Threaded {
     n_files: Arc<AtomicUsize>,
     pool: threadpool::ThreadPool,
-    tracker: Option<&'a DownloadTracker>,
     rx: Receiver<Task>,
     tx: Sender<Task>,
     vec_pools: EnumMap<Bucket, Pool>,
     ram_budget: usize,
 }
 
-impl<'a> Threaded<'a> {
+impl Threaded {
     /// Construct a new Threaded executor.
-    pub(crate) fn new(
-        tracker: Option<&'a DownloadTracker>,
-        thread_count: usize,
-        ram_budget: usize,
-    ) -> Self {
+    pub(crate) fn new(thread_count: usize, ram_budget: usize) -> Self {
         // Defaults to hardware thread count threads; this is suitable for
         // our needs as IO bound operations tend to show up as write latencies
         // rather than close latencies, so we don't need to look at
@@ -168,7 +162,6 @@ impl<'a> Threaded<'a> {
         Self {
             n_files: Arc::new(AtomicUsize::new(0)),
             pool,
-            tracker,
             rx,
             tx,
             vec_pools,
@@ -233,7 +226,7 @@ impl<'a> Threaded<'a> {
     }
 }
 
-impl Executor for Threaded<'_> {
+impl Executor for Threaded {
     fn dispatch(&self, item: Item) -> Box<dyn Iterator<Item = CompletedIo> + '_> {
         // Yield any completed work before accepting new work - keep memory
         // pressure under control
@@ -260,39 +253,22 @@ impl Executor for Threaded<'_> {
         // items, and the download tracker's progress is confounded with
         // 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(tracker) = self.tracker {
-            tracker.handle(Notification::DownloadFinished(None));
-            tracker.handle(Notification::DownloadContentLengthReceived(
-                prev_files as u64,
-                None,
-            ));
-        }
+        let prev_files = self.n_files.load(Ordering::Relaxed);
         if prev_files > 50 {
             debug!("{prev_files} deferred IO operations");
         }
-        let buf: Vec<u8> = vec![0; prev_files];
+
         // Cheap wrap-around correctness check - we have 20k files, more than
         // 32K means we subtracted from 0 somewhere.
         assert!(32767 > prev_files);
         let mut current_files = prev_files;
         while current_files != 0 {
             use std::thread::sleep;
             sleep(std::time::Duration::from_millis(100));
-            prev_files = current_files;
             current_files = self.n_files.load(Ordering::Relaxed);
-            let step_count = prev_files - current_files;
-            if let Some(tracker) = self.tracker {
-                tracker.handle(Notification::DownloadDataReceived(
-                    &buf[0..step_count],
-                    None,
-                ));
-            }
         }
         self.pool.join();
-        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
         // pool join, so this is race-free. It is possible that try_iter is safe
@@ -352,19 +328,19 @@ impl Executor for Threaded<'_> {
     }
 }
 
-impl Drop for Threaded<'_> {
+impl Drop for Threaded {
     fn drop(&mut self) {
         // We are not permitted to fail - consume but do not handle the items.
         self.join().for_each(drop);
     }
 }
 
-struct JoinIterator<'a, 'b> {
-    executor: &'a Threaded<'b>,
+struct JoinIterator<'a> {
+    executor: &'a Threaded,
     consume_sentinel: bool,
 }
 
-impl JoinIterator<'_, '_> {
+impl JoinIterator<'_> {
     fn inner<T: Iterator<Item = Task>>(&self, mut iter: T) -> Option<CompletedIo> {
         loop {
             let task_o = iter.next();
@@ -388,7 +364,7 @@ impl JoinIterator<'_, '_> {
     }
 }
 
-impl Iterator for JoinIterator<'_, '_> {
+impl Iterator for JoinIterator<'_> {
     type Item = CompletedIo;
 
     fn next(&mut self) -> Option<CompletedIo> {
@@ -400,12 +376,12 @@ impl Iterator for JoinIterator<'_, '_> {
     }
 }
 
-struct SubmitIterator<'a, 'b> {
-    executor: &'a Threaded<'b>,
+struct SubmitIterator<'a> {
+    executor: &'a Threaded,
     item: Cell<Option<Item>>,
 }
 
-impl Iterator for SubmitIterator<'_, '_> {
+impl Iterator for SubmitIterator<'_> {
     type Item = CompletedIo;
 
     fn next(&mut self) -> Option<CompletedIo> {

src/dist/component/package.rs

@@ -299,8 +299,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.tracker), unpack_ram, dl_cfg.process)?;
+    let mut io_executor: Box<dyn Executor> = get_executor(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

@@ -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.tracker)?;
+            let cached_result = file_hash(&target_file)?;
             if hash == cached_result {
                 debug!("reusing previously downloaded file");
                 debug!(url = url.as_ref(), "checksum passed");
@@ -216,36 +216,6 @@ impl DownloadTracker {
         }
     }
 
-    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);
-                }
-            }
-            Notification::DownloadDataReceived(data, url) => {
-                if let Some(url) = url {
-                    self.data_received(data.len(), url);
-                }
-            }
-            Notification::DownloadFinished(url) => {
-                if let Some(url) = url {
-                    self.download_finished(url);
-                }
-            }
-            Notification::DownloadFailed(url) => {
-                self.download_failed(url);
-                debug!("download failed");
-            }
-            Notification::DownloadingComponent(component, url) => {
-                self.create_progress_bar(component.to_owned(), url.to_owned());
-            }
-            Notification::RetryingDownload(url) => {
-                self.retrying_download(url);
-            }
-        }
-    }
-
     /// Creates a new ProgressBar for the given component.
     pub(crate) fn create_progress_bar(&self, component: String, url: String) {
         let pb = ProgressBar::hidden();
@@ -329,25 +299,9 @@ impl DownloadTracker {
     }
 }
 
-#[derive(Debug)]
-pub(crate) enum Notification<'a> {
-    /// The URL of the download is passed as the last argument, to allow us to track concurrent downloads.
-    DownloadingComponent(&'a str, &'a str),
-    RetryingDownload(&'a str),
-    /// Received the Content-Length of the to-be downloaded data with
-    /// the respective URL of the download (for tracking concurrent downloads).
-    DownloadContentLengthReceived(u64, Option<&'a str>),
-    /// Received some data.
-    DownloadDataReceived(&'a [u8], Option<&'a str>),
-    /// Download has finished.
-    DownloadFinished(Option<&'a str>),
-    /// Download has failed.
-    DownloadFailed(&'a str),
-}
-
-fn file_hash(path: &Path, tracker: &DownloadTracker) -> Result<String> {
+fn file_hash(path: &Path) -> Result<String> {
     let mut hasher = Sha256::new();
-    let mut downloaded = utils::FileReaderWithProgress::new_file(path, tracker)?;
+    let mut downloaded = utils::FileReaderWithProgress::new_file(path)?;
     use std::io::Read;
     let mut buf = vec![0; 32768];
     while let Ok(n) = downloaded.read(&mut buf) {

src/dist/manifestation.rs

@@ -17,7 +17,7 @@ use crate::dist::component::{
     Components, Package, TarGzPackage, TarXzPackage, TarZStdPackage, Transaction,
 };
 use crate::dist::config::Config;
-use crate::dist::download::{DownloadCfg, File, Notification};
+use crate::dist::download::{DownloadCfg, File};
 use crate::dist::manifest::{Component, CompressionKind, HashedBinary, Manifest, TargetedPackage};
 use crate::dist::prefix::InstallPrefix;
 #[cfg(test)]
@@ -177,12 +177,10 @@ impl Manifestation {
 
         info!("downloading component(s)");
         for bin in &components {
-            download_cfg
-                .tracker
-                .handle(Notification::DownloadingComponent(
-                    &bin.component.short_name(new_manifest),
-                    &bin.binary.url,
-                ));
+            download_cfg.tracker.create_progress_bar(
+                bin.component.short_name(new_manifest),
+                bin.binary.url.clone(),
+            );
         }
 
         let semaphore = Arc::new(Semaphore::new(concurrent_downloads));
@@ -277,8 +275,7 @@ impl Manifestation {
                 }
             }
 
-            let reader =
-                utils::FileReaderWithProgress::new_file(&installer_file, &download_cfg.tracker)?;
+            let reader = utils::FileReaderWithProgress::new_file(&installer_file)?;
             let package = match format {
                 CompressionKind::GZip => &TarGzPackage::new(reader, download_cfg)? as &dyn Package,
                 CompressionKind::XZ => &TarXzPackage::new(reader, download_cfg)?,
@@ -445,7 +442,7 @@ impl Manifestation {
 
         dl_cfg
             .tracker
-            .handle(Notification::DownloadingComponent("rust", &url));
+            .create_progress_bar("rust".to_owned(), url.clone());
 
         let dl = dl_cfg
             .download_and_check(&url, update_hash, ".tar.gz")
@@ -468,7 +465,7 @@ impl Manifestation {
         }
 
         // Install all the components in the installer
-        let reader = utils::FileReaderWithProgress::new_file(&installer_file, &dl_cfg.tracker)?;
+        let reader = utils::FileReaderWithProgress::new_file(&installer_file)?;
         let package: &dyn Package = &TarGzPackage::new(reader, dl_cfg)?;
         for component in package.components() {
             tx = package.install(&self.installation, &component, None, tx)?;
@@ -749,9 +746,7 @@ impl<'a> ComponentBinary<'a> {
                 match e.downcast_ref::<RustupError>() {
                     Some(RustupError::BrokenPartialFile)
                     | Some(RustupError::DownloadingFile { .. }) => {
-                        download_cfg
-                            .tracker
-                            .handle(Notification::RetryingDownload(url.as_str()));
+                        download_cfg.tracker.retrying_download(url.as_str());
                         true
                     }
                     _ => false,

src/download/mod.rs

@@ -26,11 +26,7 @@ use tracing::info;
 use tracing::warn;
 use url::Url;
 
-use crate::{
-    dist::download::{DownloadTracker, Notification},
-    errors::RustupError,
-    process::Process,
-};
+use crate::{dist::download::DownloadTracker, errors::RustupError, process::Process};
 
 #[cfg(test)]
 mod tests;
@@ -111,14 +107,9 @@ async fn download_file_(
 
         match msg {
             Event::DownloadContentLengthReceived(len) => {
-                tracker.handle(Notification::DownloadContentLengthReceived(
-                    len,
-                    Some(url.as_str()),
-                ));
-            }
-            Event::DownloadDataReceived(data) => {
-                tracker.handle(Notification::DownloadDataReceived(data, Some(url.as_str())));
+                tracker.content_length_received(len, url.as_str())
             }
+            Event::DownloadDataReceived(data) => tracker.data_received(data.len(), url.as_str()),
             Event::ResumingPartialDownload => debug!("resuming partial download"),
         }
 
@@ -219,10 +210,10 @@ async fn download_file_(
         .await;
 
     // The notification should only be sent if the download was successful (i.e. didn't timeout)
-    tracker.handle(match &res {
-        Ok(_) => Notification::DownloadFinished(Some(url.as_str())),
-        Err(_) => Notification::DownloadFailed(url.as_str()),
-    });
+    match &res {
+        Ok(_) => tracker.download_finished(url.as_str()),
+        Err(_) => tracker.download_failed(url.as_str()),
+    };
 
     res
 }