From 740106d7595c4e2722ca2c8d0f6ee34f751f06d2 Mon Sep 17 00:00:00 2001 From: Fintan Halpenny Date: Mon, 29 Apr 2024 12:19:08 +0100 Subject: [PATCH] cli: upload-pack events Introduce an `UploadPack` type for keeping track of relevant data from upload-pack events and displaying progress to the terminal. It keeps track of the number of remotes that are being uploaded to as well as the throughput of transmitted data. The `Progress` events are logged rather than being displayed since they don't provide any useful details to the user. Signed-off-by: Fintan Halpenny X-Clacks-Overhead: GNU Terry Pratchett --- radicle-cli/src/commands/init.rs | 14 +++++-- radicle-cli/src/terminal.rs | 1 + radicle-cli/src/terminal/format.rs | 26 +++++++++++++ radicle-cli/src/terminal/upload_pack.rs | 50 +++++++++++++++++++++++++ radicle-node/src/worker/upload_pack.rs | 33 ++++++++++------ radicle/src/node/events/upload_pack.rs | 19 ++++++++++ 6 files changed, 129 insertions(+), 14 deletions(-) create mode 100644 radicle-cli/src/terminal/upload_pack.rs diff --git a/radicle-cli/src/commands/init.rs b/radicle-cli/src/commands/init.rs index f6b8cc66..2c8e3918 100644 --- a/radicle-cli/src/commands/init.rs +++ b/radicle-cli/src/commands/init.rs @@ -395,6 +395,8 @@ fn sync( spinner.message("Syncing.."); let mut replicas = HashSet::new(); + // Start upload pack as None and set it if we encounter an event + let mut upload_pack = term::upload_pack::UploadPack::new(); for e in events { match e { @@ -413,21 +415,27 @@ fn sync( remote, progress, })) if rid == rid_ => { - spinner.message(format!("Uploading {rid} to {remote} {progress}")); + log::debug!("Upload progress for {remote}: {progress}"); } + Ok(Event::UploadPack(UploadPack::PackProgress { + rid: rid_, + remote, + transmitted, + })) if rid == rid_ => spinner.message(upload_pack.transmitted(remote, transmitted)), Ok(Event::UploadPack(UploadPack::Done { rid: rid_, remote, status, })) if rid == rid_ => { - spinner.message(format!("Upload done for {rid} to {remote}: {status}")); + log::debug!("Upload done for {rid} to {remote} with status: {status}"); + spinner.message(upload_pack.done(&remote)); } Ok(Event::UploadPack(UploadPack::Error { rid: rid_, remote, err, })) if rid == rid_ => { - spinner.message(format!("Upload error for {rid} to {remote}: {err}")); + term::warning(format!("Upload error for {rid} to {remote}: {err}")); } Ok(_) => { // Some other irrelevant event received. diff --git a/radicle-cli/src/terminal.rs b/radicle-cli/src/terminal.rs index a708e002..5e351ec3 100644 --- a/radicle-cli/src/terminal.rs +++ b/radicle-cli/src/terminal.rs @@ -8,6 +8,7 @@ pub mod highlight; pub mod issue; pub mod json; pub mod patch; +pub mod upload_pack; use std::ffi::OsString; use std::process; diff --git a/radicle-cli/src/terminal/format.rs b/radicle-cli/src/terminal/format.rs index 74667719..a8bf1690 100644 --- a/radicle-cli/src/terminal/format.rs +++ b/radicle-cli/src/terminal/format.rs @@ -82,6 +82,22 @@ pub fn timestamp(time: impl Into) -> Paint { Paint::new(fmt.convert(duration.into())) } +pub fn bytes(size: usize) -> Paint { + const KB: usize = 1024; + const MB: usize = 1024usize.pow(2); + const GB: usize = 1024usize.pow(3); + let size = if size < KB { + format!("{size} B") + } else if size < MB { + format!("{} KiB", size / KB) + } else if size < GB { + format!("{} MiB", size / MB) + } else { + format!("{} GiB", size / GB) + }; + Paint::new(size) +} + /// Format a ref update. pub fn ref_update(update: &RefUpdate) -> Paint<&'static str> { match update { @@ -358,4 +374,14 @@ mod test { let res = strip_comments(test); assert_eq!(exp, res); } + + #[test] + fn test_bytes() { + assert_eq!(bytes(1023).to_string(), "1023 B"); + assert_eq!(bytes(1024).to_string(), "1 KiB"); + assert_eq!(bytes(1024 * 9).to_string(), "9 KiB"); + assert_eq!(bytes(1024usize.pow(2)).to_string(), "1 MiB"); + assert_eq!(bytes(1024usize.pow(2) * 56).to_string(), "56 MiB"); + assert_eq!(bytes(1024usize.pow(3) * 1024).to_string(), "1024 GiB"); + } } diff --git a/radicle-cli/src/terminal/upload_pack.rs b/radicle-cli/src/terminal/upload_pack.rs new file mode 100644 index 00000000..6e3db7bf --- /dev/null +++ b/radicle-cli/src/terminal/upload_pack.rs @@ -0,0 +1,50 @@ +use std::collections::BTreeSet; +use std::time::Instant; + +use crate::terminal::format; +use radicle::node::NodeId; + +/// Keeps track of upload-pack progress for displaying to the terminal. +pub struct UploadPack { + /// Keep track of which remotes are being uploaded to, removing any that + /// have completed. + remotes: BTreeSet, + /// Keep track of how long we've been transmitting for to calculate + /// throughput. + timer: Instant, +} + +impl Default for UploadPack { + fn default() -> Self { + Self::new() + } +} + +impl UploadPack { + /// Construct an empty set of spinners. + pub fn new() -> Self { + Self { + remotes: BTreeSet::new(), + timer: Instant::now(), + } + } + + /// Display the number of peers, the total transmitted bytes, and the + /// throughput. + pub fn transmitted(&mut self, remote: NodeId, transmitted: usize) -> String { + self.remotes.insert(remote); + let throughput = transmitted as f64 / self.timer.elapsed().as_secs_f64(); + let throughput = format::bytes(throughput.floor() as usize); + let n = self.remotes.len(); + let transmitted = format::bytes(transmitted); + format!("Uploading to {n} peer(s) ({transmitted} | {throughput:.2}/s)") + } + + /// Display which remote has completed upload-pack and how many are + /// remaining. + pub fn done(&mut self, remote: &NodeId) -> String { + self.remotes.remove(remote); + let n = self.remotes.len(); + format!("Uploaded to {remote}, {n} peer(s) remaining..") + } +} diff --git a/radicle-node/src/worker/upload_pack.rs b/radicle-node/src/worker/upload_pack.rs index 731b278d..d448bd8c 100644 --- a/radicle-node/src/worker/upload_pack.rs +++ b/radicle-node/src/worker/upload_pack.rs @@ -83,12 +83,8 @@ where let mut stdin = child.stdin.take().unwrap(); let mut stdout = io::BufReader::new(child.stdout.take().unwrap()); - let mut reporter = Reporter { - rid: header.repo, - remote, - emitter: emitter.clone(), - send, - }; + let mut reporter = Reporter::new(header.repo, remote, emitter.clone(), send); + thread::scope(|s| { thread::spawn_scoped(nid, "upload-pack", s, || { // N.b. we indefinitely copy stdout to the sender, @@ -152,17 +148,32 @@ struct Reporter { remote: NodeId, emitter: Emitter, send: W, + total: usize, } impl Reporter { - fn emit(&self, buf: &[u8]) { - if let Some(progress) = Self::as_upload_pack_progress(buf) { - log::trace!(target: "worker", "upload-pack progress: {progress}"); - self.emitter - .emit(events::UploadPack::write(self.rid, self.remote, progress).into()); + fn new(rid: RepoId, remote: NodeId, emitter: Emitter, send: W) -> Self { + Self { + rid, + remote, + emitter, + send, + total: 0, } } + fn emit(&mut self, buf: &[u8]) { + let event = match Self::as_upload_pack_progress(buf) { + Some(progress) => events::UploadPack::write(self.rid, self.remote, progress), + None => { + self.total += buf.len(); + events::UploadPack::pack_progress(self.rid, self.remote, self.total) + } + }; + log::trace!(target: "worker", "upload-pack progress: {event:?}"); + self.emitter.emit(event.into()) + } + fn as_upload_pack_progress(buf: &[u8]) -> Option { use events::upload_pack::Progress::*; let gix_protocol::RemoteProgress { diff --git a/radicle/src/node/events/upload_pack.rs b/radicle/src/node/events/upload_pack.rs index 284947fe..c9d16275 100644 --- a/radicle/src/node/events/upload_pack.rs +++ b/radicle/src/node/events/upload_pack.rs @@ -29,6 +29,7 @@ pub enum UploadPack { /// The progress metadata of the upload-pack. progress: Progress, }, + /// An error occurred during the upload-pack process. Error { /// The repository being fetched. rid: RepoId, @@ -37,9 +38,27 @@ pub enum UploadPack { /// The error that occurred during the upload-pack. err: String, }, + /// The upload-pack packfile transmission progress. + PackProgress { + /// The repository being fetched. + rid: RepoId, + /// The node being fetched from. + remote: NodeId, + /// The total number of bytes transmitted. + transmitted: usize, + }, } impl UploadPack { + /// Construct a `UploadPack::PackProgress` event. + pub fn pack_progress(rid: RepoId, remote: NodeId, transmitted: usize) -> Self { + Self::PackProgress { + rid, + remote, + transmitted, + } + } + /// Construct a `UploadPack::Write` event. pub fn write(rid: RepoId, remote: NodeId, progress: Progress) -> Self { Self::Write {