Implement progress like it was before

Except that now it should go towards zero on disconnect.

Author: Byron

src/cargo/sources/git/oxide.rs

@@ -1,12 +1,12 @@
 //! This module contains all code sporting `gitoxide` for operations on `git` repositories and it mirrors
 //! `utils` closely for now. One day it can be renamed into `utils` once `git2` isn't required anymore.
 
-use crate::util::{network, Progress};
+use crate::util::{human_readable_bytes, network, MetricsCounter, Progress};
 use crate::{CargoResult, Config};
 use git_repository as git;
-use std::sync::atomic::AtomicBool;
+use std::sync::atomic::{AtomicBool, Ordering};
 use std::sync::Arc;
-use std::time::Duration;
+use std::time::{Duration, Instant};
 
 /// For the time being, `repo_path` makes it easy to instantiate a gitoxide repo just for fetching.
 /// In future this may change to be the gitoxide repository itself.
@@ -49,22 +49,78 @@ pub fn with_retry_and_progress(
         .ok()
         .unwrap_or_default()
         .auto_deregister();
-    let should_interrupt = AtomicBool::new(false);
     let mut progress_bar = Progress::new("Fetch", config);
     std::thread::scope(move |s| {
         s.spawn({
             let root = Arc::downgrade(&progress_root);
             move || -> CargoResult<()> {
+                const READ_PACK_BYTES: [u8; 4] = *b"BWRB";
+                const DELTA_INDEX_OBJECTS: [u8; 4] = *b"IWIO";
+                const RESOLVE_OBJECTS: [u8; 4] = *b"IWRO";
+
+                // We choose `N=10` here to make a `300ms * 10slots ~= 3000ms`
+                // sliding window for tracking the data transfer rate (in bytes/s).
+                let mut last_update = Instant::now();
+                let mut counter = MetricsCounter::<10>::new(0, last_update);
+
                 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);
+                    let counters = tasks
+                        .iter()
+                        .find_map(|(_, t)| {
+                            (t.id == READ_PACK_BYTES).then(|| t.progress.as_ref().expect("set"))
+                        })
+                        .and_then(|read| {
+                            tasks.iter().find_map(|(_, t)| {
+                                (t.id == DELTA_INDEX_OBJECTS)
+                                    .then(|| (t.progress.as_ref().expect("set"), Some(read)))
+                            })
+                        })
+                        .or_else(|| {
+                            tasks.iter().find_map(|(_, t)| {
+                                (t.id == RESOLVE_OBJECTS)
+                                    .then(|| (t.progress.as_ref().expect("set"), None))
+                            })
+                        });
+                    let progress_info = match counters {
+                        Some((objs, None)) => {
+                            // Resolving deltas.
+                            let objects = objs.step.load(Ordering::Relaxed);
+                            let total_objects = objs.done_at.expect("known amount of objects");
+                            Some((
+                                objects,
+                                total_objects,
+                                format!(", ({}/{}) resolving deltas", objects, total_objects),
+                            ))
+                        }
+                        Some((objs, Some(read_pack))) => {
+                            // Receiving objects.
+                            let objects = objs.step.load(Ordering::Relaxed);
+                            let total_objects = objs.done_at.expect("known amount of objects");
+                            let received_bytes = read_pack.step.load(Ordering::Relaxed);
+
+                            let now = Instant::now();
+                            counter.add(received_bytes, now);
+                            last_update = now;
+                            let (rate, unit) = human_readable_bytes(counter.rate() as u64);
+                            Some((objects, total_objects, format!(", {:.2}{}/s", rate, unit)))
+                        }
+                        None => {
+                            counter = MetricsCounter::<10>::new(0, last_update);
+                            None
+                        }
+                    };
+                    if let Some((objects, total_objects, msg)) = progress_info {
+                        progress_bar.tick(objects, total_objects, &msg)?;
+                    }
                     std::thread::sleep(Duration::from_millis(300));
                 }
                 Ok(())
             }
         });
-        network::with_retry(config, || cb(&repo, &should_interrupt, &mut progress))
+        network::with_retry(config, || {
+            cb(&repo, &git::interrupt::IS_INTERRUPTED, &mut progress)
+        })
     })
 }

src/cargo/sources/git/utils.rs

@@ -867,7 +867,7 @@ pub fn fetch(
         oxide::with_retry_and_progress(
             &git2_repo.path().to_owned(),
             config,
-            &mut |repo, should_interrupt, progress| {
+            &mut |repo, should_interrupt, mut progress| {
                 // The `fetch` operation here may fail spuriously due to a corrupt
                 // repository. It could also fail, however, for a whole slew of other
                 // reasons (aka network related reasons). We want Cargo to automatically
@@ -889,7 +889,7 @@ pub fn fetch(
                             refspecs.iter().map(|s| s.as_str()),
                             git::remote::Direction::Fetch,
                         )?
-                        .connect(git::remote::Direction::Fetch, progress.add_child("fetch"))?
+                        .connect(git::remote::Direction::Fetch, &mut progress)?
                         .prepare_fetch(git::remote::ref_map::Options::default())?
                         .receive(should_interrupt);
                     let err = match res {