diff --git a/common/client-core/src/client/mod.rs b/common/client-core/src/client/mod.rs index 9a3a3a33c9..f8226da387 100644 --- a/common/client-core/src/client/mod.rs +++ b/common/client-core/src/client/mod.rs @@ -7,7 +7,7 @@ pub(crate) mod helpers; pub mod inbound_messages; pub mod key_manager; pub mod mix_traffic; -pub(crate) mod packet_statistics_control; +pub mod packet_statistics_control; pub mod real_messages_control; pub mod received_buffer; pub mod replies; diff --git a/common/client-core/src/client/packet_statistics_control.rs b/common/client-core/src/client/packet_statistics_control.rs index 60f0d36e6e..bfdca3c38a 100644 --- a/common/client-core/src/client/packet_statistics_control.rs +++ b/common/client-core/src/client/packet_statistics_control.rs @@ -1,3 +1,4 @@ +use core::fmt; use std::{ collections::VecDeque, time::{Duration, Instant}, @@ -189,29 +190,64 @@ impl std::ops::Sub for PacketStatistics { } } +pub struct MixnetBandwidthStatisticsEvent { + pub rates: PacketRates, +} + +impl MixnetBandwidthStatisticsEvent { + pub fn new(rates: PacketRates) -> Self { + Self { rates } + } +} + +impl nym_task::TaskStatusEvent for MixnetBandwidthStatisticsEvent { + fn as_any(&self) -> &dyn std::any::Any { + self + } +} + +impl fmt::Display for MixnetBandwidthStatisticsEvent { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + write!(f, "{}", self.rates.summary()) + } +} + #[derive(Debug, Clone)] -struct PacketRates { - real_packets_sent: f64, - real_packets_sent_size: f64, - cover_packets_sent: f64, - cover_packets_sent_size: f64, +pub struct PacketRates { + pub real_packets_sent: f64, + pub real_packets_sent_size: f64, + pub cover_packets_sent: f64, + pub cover_packets_sent_size: f64, - real_packets_received: f64, - real_packets_received_size: f64, - cover_packets_received: f64, - cover_packets_received_size: f64, + pub real_packets_received: f64, + pub real_packets_received_size: f64, + pub cover_packets_received: f64, + pub cover_packets_received_size: f64, - total_acks_received: f64, - total_acks_received_size: f64, - real_acks_received: f64, - real_acks_received_size: f64, - cover_acks_received: f64, - cover_acks_received_size: f64, + pub total_acks_received: f64, + pub total_acks_received_size: f64, + pub real_acks_received: f64, + pub real_acks_received_size: f64, + pub cover_acks_received: f64, + pub cover_acks_received_size: f64, - real_packets_queued: f64, - retransmissions_queued: f64, - reply_surbs_queued: f64, - additional_reply_surbs_queued: f64, + pub real_packets_queued: f64, + pub retransmissions_queued: f64, + pub reply_surbs_queued: f64, + pub additional_reply_surbs_queued: f64, +} + +impl fmt::Display for PacketRates { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + write!( + f, + "down: {}/s, up: {}/s (cover down: {}/s, cover up: {}/s)", + bibytes2(self.real_packets_received_size), + bibytes2(self.real_packets_sent_size), + bibytes2(self.cover_packets_received_size), + bibytes2(self.cover_packets_sent_size), + ) + } } impl From for PacketRates { @@ -456,11 +492,13 @@ impl PacketStatisticsControl { } } - fn report_rates(&self) { + fn report_rates(&self) -> Option { if let Some((_, rates)) = self.rates.back() { log::debug!("{}", rates.summary()); log::debug!("{}", rates.detailed_summary()); + return Some(rates.clone()); } + None } fn report_counters(&self) { @@ -499,7 +537,7 @@ impl PacketStatisticsControl { // IDEA: if there is a burst of acks, that could indicate tokio task starvation. } - pub(crate) async fn run_with_shutdown(&mut self, mut shutdown: nym_task::TaskClient) { + pub(crate) async fn run_with_shutdown(&mut self, mut task_client: nym_task::TaskClient) { log::debug!("Started PacketStatisticsControl with graceful shutdown support"); let report_interval = Duration::from_secs(PACKET_REPORT_INTERVAL_SECS); @@ -583,11 +621,16 @@ impl PacketStatisticsControl { self.update_rates(); } _ = report_interval.tick() => { - self.report_rates(); + let rates = self.report_rates(); self.check_for_notable_events(); self.report_counters(); + + // Report our current bandwidth used to e.g a GUI client + if let Some(rates) = rates { + task_client.send_status_msg(Box::new(MixnetBandwidthStatisticsEvent::new(rates))); + } } - _ = shutdown.recv_with_delay() => { + _ = task_client.recv_with_delay() => { log::trace!("PacketStatisticsControl: Received shutdown"); break; },