feat: show bytes in human readable format (#3819)

Co-authored-by: Hubert <[email protected]>

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> {