Add rate meter to terminal progress

src/progress/mod.rs

@@ -1,21 +1,30 @@
 pub mod filter;
 pub mod log;
+mod rate;
 pub mod terminal;
 
 pub trait Progressor: Send {
     fn run(&mut self);
 }
 
-fn human_bytes(bytes: u64) -> String {
+fn human_bytes(bytes: f64) -> String {
     let units = ["B", "kB", "MB", "GB", "TB"];
 
-    let mut value = bytes as f64;
+    let mut a = bytes;
-    let mut index = 0;
+    let mut i = 0;
-    while value >= 1024.0 && index < units.len() - 1 {
+    while a >= 1024.0 && i < units.len() - 1 {
-        value /= 1024.0;
+        a /= 1024.0;
-        index += 1;
+        i += 1;
     }
-    format!("{:.2} {:2}", value, units[index])
+    format!("{:.2} {:2}", a, units[i])
+}
+
+fn format_eta(seconds: f64) -> String {
+    let secs = seconds as u64;
+    let h = secs / 3600;
+    let m = (secs % 3600) / 60;
+    let s = secs % 60;
+    format!("{:02}:{:02}:{:02}", h, m, s)
 }
 
 pub enum BackupEvent {

src/progress/rate.rs

@@ -0,0 +1,74 @@
+use std::time::Instant;
+
+#[derive(Clone, Copy)]
+struct Sample {
+    at: Instant,
+    bytes: u64,
+}
+
+pub(crate) struct RateTracker {
+    samples: Vec<Sample>,
+    head: usize,
+    count: usize,
+    smoothed: f64,
+    alpha: f64,
+}
+
+impl RateTracker {
+    pub(crate) fn new(capacity: usize) -> Self {
+        RateTracker {
+            samples: vec![
+                Sample {
+                    at: Instant::now(),
+                    bytes: 0,
+                };
+                capacity
+            ],
+            head: 0,
+            count: 0,
+            smoothed: 0.0,
+            alpha: 0.05,
+        }
+    }
+
+    pub(crate) fn push(&mut self, bytes: u64) {
+        let sample = Sample {
+            at: Instant::now(),
+            bytes,
+        };
+        self.samples[self.head] = sample;
+        self.head = (self.head + 1) % self.samples.capacity();
+        self.count = (self.count + 1).min(self.samples.capacity())
+    }
+
+    pub(crate) fn rate(&mut self) -> Option<f64> {
+        if self.count < 2 {
+            return None;
+        }
+
+        let cap = self.samples.capacity();
+
+        let oldest_idx = (cap + self.head - self.count) % cap;
+        let newest_idx = (cap + self.head - 1) % cap;
+
+        let oldest = self.samples[oldest_idx];
+        let newest = self.samples[newest_idx];
+
+        let elapsed = newest.at.duration_since(oldest.at).as_secs_f64();
+        if elapsed == 0.0 {
+            return None;
+        }
+
+        // If for any reason oldest < newest, the subtraction will overflow
+        if newest.bytes < oldest.bytes {
+            return None;
+        }
+
+        let rate = (newest.bytes - oldest.bytes) as f64 / elapsed;
+
+        // exponentially weighted moving average to smooth those spikes
+        self.smoothed = self.alpha * rate + (1.0 - self.alpha) * self.smoothed;
+
+        Some(self.smoothed)
+    }
+}

src/progress/terminal.rs

@@ -1,9 +1,13 @@
+use crate::progress::format_eta;
+
+use super::rate::RateTracker;
 use super::{BackupEvent, human_bytes};
 use std::{io::Write, sync::mpsc::Receiver};
 
 pub struct Progressor {
     receiver: Receiver<BackupEvent>,
     estimated_size: u64,
+    rate_tracker: RateTracker,
 }
 
 impl super::Progressor for Progressor {
@@ -11,7 +15,7 @@ impl super::Progressor for Progressor {
         while let Ok(event) = self.receiver.recv() {
             match event {
                 BackupEvent::Estimate(size) => {
-                    println!("Estimated total backup size: {}", human_bytes(size));
+                    println!("Estimated total backup size: {}", human_bytes(size as f64));
                     self.estimated_size = size;
                 }
                 BackupEvent::StartingFullBackup {
@@ -53,15 +57,34 @@ impl super::Progressor for Progressor {
                             } else {
                                 0.0
                             };
+                            self.rate_tracker.push(bytes);
+                            let rate = self.rate_tracker.rate();
+                            let rate_display = match rate {
+                                None => String::from("[unknown]"),
+                                Some(0.0) => {
+                                    String::from("[stalled] ETA heat death of the universe")
+                                }
+                                Some(rate) => {
+                                    let eta = if bytes > total {
+                                        String::from("any moment now")
+                                    } else {
+                                        let eta = (total - bytes) as f64 / rate;
+                                        format_eta(eta)
+                                    };
+                                    format!("{}/s ETA {}", human_bytes(rate), eta)
+                                }
+                            };
+
                             print!(
-                                "{:>3.0}% {}/{} transferred\r",
+                                "\x1b[2K{:>3.0}% {}/{} @ {}\r",
                                 percent * 100.0,
-                                human_bytes(bytes),
+                                human_bytes(bytes as f64),
-                                human_bytes(total)
+                                human_bytes(total as f64),
+                                rate_display,
                             );
                         }
                         None => {
-                            print!("{} transferred\r", human_bytes(bytes));
+                            print!("\x1b[2K{} transferred\r", human_bytes(bytes as f64));
                         }
                     }
                     std::io::stdout().flush().ok();
@@ -82,6 +105,7 @@ impl Progressor {
         Self {
             receiver,
             estimated_size: 0,
+            rate_tracker: RateTracker::new(100),
         }
     }
 }