feat: show bytes in human readable format

CHANGELOG.md

@@ -42,6 +42,9 @@
 
 ### Changed
 
+- [#3819](https://github.com/ChainSafe/forest/pull/3819) Make progress messages
+  more human-readable.
+
 ### Removed
 
 ### Fixed

Cargo.toml

@@ -82,6 +82,7 @@ hex = { version = "0.4", features = ["serde"] }
 http = "1.0"
 http0 = { package = "http", version = "0.2" }
 human-repr = "1.0"
+human_bytes = "0.4"
 humantime = "2.1.0"
 indexmap = { version = "2.1", features = ["serde"] }
 indicatif = { version = "0.17.6", features = ["tokio"] }

src/utils/io/progress_log.rs

@@ -33,9 +33,9 @@
 //! ```
 //! # Future work
 //! - Add and move progressively to new API (Iterator, Streams), and removed deprecated usage of [`WithProgressRaw`]
-//! - Add support for bytes measure
-//! - Add a more accurate ETA, progress speed, etc
+//! - Add a more accurate ETA etc
 
+use human_bytes::human_bytes;
 use humantime::format_duration;
 use std::time::{Duration, Instant};
 
@@ -76,33 +76,55 @@ impl<R: tokio::io::AsyncRead> tokio::io::AsyncRead for WithProgress<R> {
 }
 
 impl<S> WithProgress<S> {
-    pub fn wrap_async_read(message: &str, read: S, _total_items: u64) -> WithProgress<S> {
+    pub fn wrap_async_read(message: &str, read: S, total_items: u64) -> WithProgress<S> {
         WithProgress {
             inner: read,
-            progress: Progress::new(message),
+            progress: Progress::new(message).with_total(total_items),
         }
     }
+
+    pub fn bytes(mut self) -> Self {
+        self.progress.item_type = ItemType::Bytes;
+        self
+    }
 }
 
 #[derive(Debug, Clone)]
 struct Progress {
     completed_items: u64,
+    total_items: Option<u64>,
+    last_logged_items: u64,
     start: Instant,
     last_logged: Instant,
     message: String,
+    item_type: ItemType,
+}
+
+#[derive(Debug, Clone, Copy)]
+enum ItemType {
+    Bytes,
+    Items,
 }
 
 impl Progress {
     fn new(message: &str) -> Self {
         let now = Instant::now();
         Self {
             completed_items: 0,
+            last_logged_items: 0,
+            total_items: None,
             start: now,
             last_logged: now,
             message: message.into(),
+            item_type: ItemType::Items,
         }
     }
 
+    fn with_total(mut self, total: u64) -> Self {
+        self.total_items = Some(total);
+        self
+    }
+
     fn inc(&mut self, value: u64) {
         self.completed_items += value;
 
@@ -115,18 +137,56 @@ impl Progress {
         self.emit_log_if_required();
     }
 
+    // Example output:
+    //
+    // Bytes, with total: 12.4 MiB / 1.2 GiB, 1%, 1.5 MiB/s, elapsed time: 8m 12s
+    // Bytes, without total: 12.4 MiB, 1.5 MiB/s, elapsed time: 8m 12s
+    // Items, with total: 12 / 1200, 1%, 1.5 items/s, elapsed time: 8m 12s
+    // Items, without total: 12, 1.5 items/s, elapsed time: 8m 12s
+    fn msg(&self, now: Instant) -> String {
+        let message = &self.message;
+        let elapsed_secs = (now - self.start).as_secs_f64();
+        let elapsed_duration = format_duration(Duration::from_secs(elapsed_secs as u64));
+
+        let at = match self.item_type {
+            ItemType::Bytes => human_bytes(self.completed_items as f64),
+            ItemType::Items => self.completed_items.to_string(),
+        };
+
+        let total = if let Some(total) = self.total_items {
+            let mut output = String::new();
+            if total > 0 {
+                output += " / ";
+                output += &match self.item_type {
+                    ItemType::Bytes => human_bytes(total as f64),
+                    ItemType::Items => total.to_string(),
+                };
+                output += &format!(", {:0}%", self.completed_items * 100 / total);
+            }
+            output
+        } else {
+            String::new()
+        };
+
+        let diff = self.completed_items - self.last_logged_items;
+        let speed = match self.item_type {
+            ItemType::Bytes => format!("{}/s", human_bytes(diff as f64)),
+            ItemType::Items => format!("{diff} items/s"),
+        };
+
+        format!("{message} {at}{total}, {speed}, elapsed time: {elapsed_duration}")
+    }
+
     fn emit_log_if_required(&mut self) {
         let now = Instant::now();
         if (now - self.last_logged) > UPDATE_FREQUENCY {
-            let elapsed_secs = (now - self.start).as_secs_f64();
-            let elapsed_duration = format_duration(Duration::from_secs(elapsed_secs as u64));
-
             tracing::info!(
                 target: "forest::progress",
-                "{} {} (elapsed time: {})",
-                self.message, self.completed_items, elapsed_duration
+                "{}",
+                self.msg(now)
             );
             self.last_logged = now;
+            self.last_logged_items = self.completed_items;
         }
     }
 }
@@ -151,3 +211,64 @@ impl WithProgressRaw {
         self.sync.lock().progress.set(value);
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn test_progress_msg_bytes() {
+        let mut progress = Progress::new("test");
+        let now = progress.start;
+        progress.item_type = ItemType::Bytes;
+        progress.total_items = Some(1024 * 1024 * 1024);
+        progress.set(1024 * 1024 * 1024);
+        progress.last_logged_items = 1024 * 1024 * 1024 / 2;
+        assert_eq!(
+            progress.msg(now),
+            "test 1 GiB / 1 GiB, 100%, 512 MiB/s, elapsed time: 0s"
+        );
+
+        progress.set(1024 * 1024 * 1024 / 2);
+        progress.last_logged_items = 1024 * 1024 * 1024 / 3;
+        assert_eq!(
+            progress.msg(now + Duration::from_secs(125)),
+            "test 512 MiB / 1 GiB, 50%, 170.7 MiB/s, elapsed time: 2m 5s"
+        );
+
+        progress.set(1024 * 1024 * 1024 / 10);
+        progress.last_logged_items = 1024 * 1024 * 1024 / 11;
+        assert_eq!(
+            progress.msg(now + Duration::from_secs(10)),
+            "test 102.4 MiB / 1 GiB, 9%, 9.3 MiB/s, elapsed time: 10s"
+        );
+    }
+
+    #[test]
+    fn test_progress_msg_items() {
+        let mut progress = Progress::new("test");
+        let now = progress.start;
+        progress.item_type = ItemType::Items;
+        progress.total_items = Some(1024);
+        progress.set(1024);
+        progress.last_logged_items = 1024 / 2;
+        assert_eq!(
+            progress.msg(now),
+            "test 1024 / 1024, 100%, 512 items/s, elapsed time: 0s"
+        );
+
+        progress.set(1024 / 2);
+        progress.last_logged_items = 1024 / 3;
+        assert_eq!(
+            progress.msg(now + Duration::from_secs(125)),
+            "test 512 / 1024, 50%, 171 items/s, elapsed time: 2m 5s"
+        );
+
+        progress.set(1024 / 10);
+        progress.last_logged_items = 1024 / 11;
+        assert_eq!(
+            progress.msg(now + Duration::from_secs(10)),
+            "test 102 / 1024, 9%, 9 items/s, elapsed time: 10s"
+        );
+    }
+}

src/utils/net.rs

@@ -88,11 +88,9 @@ pub async fn reader(location: &str) -> anyhow::Result<impl AsyncBufRead> {
         }
     };
 
-    Ok(tokio::io::BufReader::new(WithProgress::wrap_async_read(
-        "Loading",
-        stream,
-        content_length,
-    )))
+    Ok(tokio::io::BufReader::new(
+        WithProgress::wrap_async_read("Loading", stream, content_length).bytes(),
+    ))
 }
 
 pub async fn http_get(url: &Url) -> anyhow::Result<Response> {