make util::Progress threadsafe

Only then it will be possible to easily obtain just-in-time progress
information from `gitoxide`.

This also fixes a long-standing issue of transfer speeds never heading
to zero if `git2` stops calling the progress function.

Author: Byron

Cargo.toml

@@ -47,6 +47,7 @@ libgit2-sys = "0.14.0"
 memchr = "2.1.3"
 opener = "0.5"
 os_info = "3.5.0"
+parking_lot = "0.12.1"
 pathdiff = "0.2"
 percent-encoding = "2.0"
 rustfix = "0.6.0"

src/cargo/core/compiler/job_queue.rs

@@ -136,7 +136,7 @@ struct DrainState<'cfg> {
     documented: HashSet<PackageId>,
     scraped: HashSet<PackageId>,
     counts: HashMap<PackageId, usize>,
-    progress: Progress<'cfg>,
+    progress: Progress,
     next_id: u32,
     timings: Timings<'cfg>,
 

src/cargo/core/package.rs

@@ -329,7 +329,7 @@ pub struct Downloads<'a, 'cfg> {
     /// The next ID to use for creating a token (see `Download::token`).
     next: usize,
     /// Progress bar.
-    progress: RefCell<Option<Progress<'cfg>>>,
+    progress: RefCell<Option<Progress>>,
     /// Number of downloads that have successfully finished.
     downloads_finished: usize,
     /// Total bytes for all successfully downloaded packages.

src/cargo/core/shell.rs

@@ -72,7 +72,7 @@ impl fmt::Debug for Shell {
 /// A `Write`able object, either with or without color support
 enum ShellOut {
     /// A plain write object without color support
-    Write(Box<dyn Write>),
+    Write(Box<dyn Write + Send>),
     /// Color-enabled stdio, with information on whether color should be used
     Stream {
         stdout: StandardStream,
@@ -111,7 +111,7 @@ impl Shell {
     }
 
     /// Creates a shell from a plain writable object, with no color, and max verbosity.
-    pub fn from_write(out: Box<dyn Write>) -> Shell {
+    pub fn from_write(out: Box<dyn Write + Send>) -> Shell {
         Shell {
             output: ShellOut::Write(out),
             verbosity: Verbosity::Verbose,

src/cargo/ops/cargo_clean.rs

@@ -315,14 +315,14 @@ trait CleaningProgressBar {
     fn on_clean(&mut self) -> CargoResult<()>;
 }
 
-struct CleaningFolderBar<'cfg> {
-    bar: Progress<'cfg>,
+struct CleaningFolderBar {
+    bar: Progress,
     max: usize,
     cur: usize,
 }
 
-impl<'cfg> CleaningFolderBar<'cfg> {
-    fn new(cfg: &'cfg Config, max: usize) -> Self {
+impl CleaningFolderBar {
+    fn new(cfg: &Config, max: usize) -> Self {
         Self {
             bar: Progress::with_style("Cleaning", ProgressStyle::Percentage, cfg),
             max,
@@ -335,7 +335,7 @@ impl<'cfg> CleaningFolderBar<'cfg> {
     }
 }
 
-impl<'cfg> CleaningProgressBar for CleaningFolderBar<'cfg> {
+impl CleaningProgressBar for CleaningFolderBar {
     fn display_now(&mut self) -> CargoResult<()> {
         self.bar.tick_now(self.cur_progress(), self.max, "")
     }
@@ -346,16 +346,16 @@ impl<'cfg> CleaningProgressBar for CleaningFolderBar<'cfg> {
     }
 }
 
-struct CleaningPackagesBar<'cfg> {
-    bar: Progress<'cfg>,
+struct CleaningPackagesBar {
+    bar: Progress,
     max: usize,
     cur: usize,
     num_files_folders_cleaned: usize,
     package_being_cleaned: String,
 }
 
-impl<'cfg> CleaningPackagesBar<'cfg> {
-    fn new(cfg: &'cfg Config, max: usize) -> Self {
+impl CleaningPackagesBar {
+    fn new(cfg: &Config, max: usize) -> Self {
         Self {
             bar: Progress::with_style("Cleaning", ProgressStyle::Ratio, cfg),
             max,
@@ -384,7 +384,7 @@ impl<'cfg> CleaningPackagesBar<'cfg> {
     }
 }
 
-impl<'cfg> CleaningProgressBar for CleaningPackagesBar<'cfg> {
+impl CleaningProgressBar for CleaningPackagesBar {
     fn display_now(&mut self) -> CargoResult<()> {
         self.bar
             .tick_now(self.cur_progress(), self.max, &self.format_message())

src/cargo/sources/git/oxide.rs

@@ -50,14 +50,15 @@ pub fn with_retry_and_progress(
         .unwrap_or_default()
         .auto_deregister();
     let should_interrupt = AtomicBool::new(false);
-    let _progress_bar = Progress::new("Fetch", config);
+    let mut progress_bar = Progress::new("Fetch", config);
     std::thread::scope(move |s| {
         s.spawn({
             let root = Arc::downgrade(&progress_root);
             move || -> CargoResult<()> {
                 let mut tasks = Vec::with_capacity(10);
                 while let Some(root) = root.upgrade() {
                     root.sorted_snapshot(&mut tasks);
+                    progress_bar.tick(0, 10, "TBD")?;
                     // dbg!(&tasks);
                     std::thread::sleep(Duration::from_millis(300));
                 }

src/cargo/sources/registry/http_remote.rs

@@ -105,7 +105,7 @@ pub struct Downloads<'cfg> {
     /// The next ID to use for creating a token (see `Download::token`).
     next: usize,
     /// Progress bar.
-    progress: RefCell<Option<Progress<'cfg>>>,
+    progress: RefCell<Option<Progress>>,
     /// Number of downloads that have successfully finished.
     downloads_finished: usize,
     /// Number of times the caller has requested blocking. This is used for

src/cargo/util/config/mod.rs

@@ -49,6 +49,7 @@
 //! translate from `ConfigValue` and environment variables to the caller's
 //! desired type.
 
+use parking_lot::{Mutex, MutexGuard};
 use std::borrow::Cow;
 use std::cell::{RefCell, RefMut};
 use std::collections::hash_map::Entry::{Occupied, Vacant};
@@ -62,7 +63,7 @@ use std::io::{self, SeekFrom};
 use std::mem;
 use std::path::{Path, PathBuf};
 use std::str::FromStr;
-use std::sync::Once;
+use std::sync::{Arc, Once};
 use std::time::Instant;
 
 use self::ConfigValue as CV;
@@ -143,7 +144,7 @@ pub struct Config {
     /// The location of the user's Cargo home directory. OS-dependent.
     home_path: Filesystem,
     /// Information about how to write messages to the shell
-    shell: RefCell<Shell>,
+    shell: Arc<Mutex<Shell>>,
     /// A collection of configuration options
     values: LazyCell<HashMap<String, ConfigValue>>,
     /// A collection of configuration options from the credentials file
@@ -269,7 +270,7 @@ impl Config {
 
         Config {
             home_path: Filesystem::new(homedir),
-            shell: RefCell::new(shell),
+            shell: Arc::new(Mutex::new(shell)),
             cwd,
             search_stop_path: None,
             values: LazyCell::new(),
@@ -368,8 +369,13 @@ impl Config {
     }
 
     /// Gets a reference to the shell, e.g., for writing error messages.
-    pub fn shell(&self) -> RefMut<'_, Shell> {
-        self.shell.borrow_mut()
+    pub fn shell(&self) -> MutexGuard<'_, Shell> {
+        self.shell.lock()
+    }
+
+    /// Gets a shared reference to the shell, e.g., for writing error messages, for use when writing from threads.
+    pub fn shell_detached(&self) -> Arc<Mutex<Shell>> {
+        Arc::clone(&self.shell)
     }
 
     /// Gets the path to the `rustdoc` executable.
@@ -1261,9 +1267,7 @@ impl Config {
                 // --config path_to_file
                 let str_path = arg_as_path
                     .to_str()
-                    .ok_or_else(|| {
-                        anyhow::format_err!("config path {:?} is not utf-8", arg_as_path)
-                    })?
+                    .ok_or_else(|| format_err!("config path {:?} is not utf-8", arg_as_path))?
                     .to_string();
                 self._load_file(&self.cwd().join(&str_path), &mut seen, true, WhyLoad::Cli)
                     .with_context(|| format!("failed to load config from `{}`", str_path))?

src/cargo/util/progress.rs

@@ -1,15 +1,18 @@
+use parking_lot::Mutex;
 use std::cmp;
 use std::env;
+use std::sync::Arc;
 use std::time::{Duration, Instant};
 
 use crate::core::shell::Verbosity;
+use crate::core::Shell;
 use crate::util::config::ProgressWhen;
 use crate::util::{CargoResult, Config};
 use cargo_util::is_ci;
 use unicode_width::UnicodeWidthChar;
 
-pub struct Progress<'cfg> {
-    state: Option<State<'cfg>>,
+pub struct Progress {
+    state: Option<State>,
 }
 
 pub enum ProgressStyle {
@@ -23,8 +26,8 @@ struct Throttle {
     last_update: Instant,
 }
 
-struct State<'cfg> {
-    config: &'cfg Config,
+struct State {
+    shell: Arc<Mutex<Shell>>,
     format: Format,
     name: String,
     done: bool,
@@ -39,8 +42,8 @@ struct Format {
     max_print: usize,
 }
 
-impl<'cfg> Progress<'cfg> {
-    pub fn with_style(name: &str, style: ProgressStyle, cfg: &'cfg Config) -> Progress<'cfg> {
+impl Progress {
+    pub fn with_style(name: &str, style: ProgressStyle, cfg: &Config) -> Progress {
         // report no progress when -q (for quiet) or TERM=dumb are set
         // or if running on Continuous Integration service like Travis where the
         // output logs get mangled.
@@ -60,15 +63,15 @@ impl<'cfg> Progress<'cfg> {
         Progress::new_priv(name, style, cfg)
     }
 
-    fn new_priv(name: &str, style: ProgressStyle, cfg: &'cfg Config) -> Progress<'cfg> {
+    fn new_priv(name: &str, style: ProgressStyle, cfg: &Config) -> Progress {
         let progress_config = cfg.progress_config();
         let width = progress_config
             .width
             .or_else(|| cfg.shell().err_width().progress_max_width());
 
         Progress {
             state: width.map(|n| State {
-                config: cfg,
+                shell: cfg.shell_detached(),
                 format: Format {
                     style,
                     max_width: n,
@@ -93,7 +96,7 @@ impl<'cfg> Progress<'cfg> {
         self.state.is_some()
     }
 
-    pub fn new(name: &str, cfg: &'cfg Config) -> Progress<'cfg> {
+    pub fn new(name: &str, cfg: &Config) -> Progress {
         Self::with_style(name, ProgressStyle::Percentage, cfg)
     }
 
@@ -180,7 +183,7 @@ impl Throttle {
     }
 }
 
-impl<'cfg> State<'cfg> {
+impl State {
     fn tick(&mut self, cur: usize, max: usize, msg: &str) -> CargoResult<()> {
         if self.done {
             return Ok(());
@@ -215,8 +218,8 @@ impl<'cfg> State<'cfg> {
         }
 
         // Only update if the line has changed.
-        if self.config.shell().is_cleared() || self.last_line.as_ref() != Some(&line) {
-            let mut shell = self.config.shell();
+        if self.shell.lock().is_cleared() || self.last_line.as_ref() != Some(&line) {
+            let mut shell = self.shell.lock();
             shell.set_needs_clear(false);
             shell.status_header(&self.name)?;
             write!(shell.err(), "{}\r", line)?;
@@ -229,15 +232,15 @@ impl<'cfg> State<'cfg> {
 
     fn clear(&mut self) {
         // No need to clear if the progress is not currently being displayed.
-        if self.last_line.is_some() && !self.config.shell().is_cleared() {
-            self.config.shell().err_erase_line();
+        if self.last_line.is_some() && !self.shell.lock().is_cleared() {
+            self.shell.lock().err_erase_line();
             self.last_line = None;
         }
     }
 
     fn try_update_max_width(&mut self) {
         if self.fixed_width.is_none() {
-            if let Some(n) = self.config.shell().err_width().progress_max_width() {
+            if let Some(n) = self.shell.lock().err_width().progress_max_width() {
                 self.format.max_width = n;
             }
         }
@@ -323,7 +326,7 @@ impl Format {
     }
 }
 
-impl<'cfg> Drop for State<'cfg> {
+impl Drop for State {
     fn drop(&mut self) {
         self.clear();
     }