From 643f54024b5f2f459cd732200d6d797a26f0acf3 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jon=20H=C3=A4ggblad?= Date: Mon, 22 Jan 2024 09:00:11 +0100 Subject: [PATCH] Client packet counters (#4325) * WIP: put in some packet counters * ws packet counters * wip * Add static counters to client traffic stream * Tweak status log message * Add packet statistics control * fixup! Add static counters to client traffic stream * tweak log * Move the packet statistics control one level up * Redo packet stats control to collect locally * Switch loop cover traffic report over to new channel mechanism * Switch packet stats in real message stream to channel report * Finished switching over to channel reporting * Fix handle stats event * Log packets received * Tidy up * rustfmt * Add strongly typed stats reporter * Count cover packets as well * Log packet sizes sent * Also log recieved sizes --- .../client-core/src/client/base_client/mod.rs | 21 ++ .../src/client/cover_traffic_stream.rs | 12 +- common/client-core/src/client/mod.rs | 1 + .../src/client/packet_statistics_control.rs | 202 ++++++++++++++++++ .../acknowledgement_listener.rs | 13 +- .../acknowledgement_control/mod.rs | 3 + .../src/client/real_messages_control/mod.rs | 5 + .../real_traffic_stream.rs | 56 ++++- .../client-core/src/client/received_buffer.rs | 23 +- .../client-libs/gateway-client/src/client.rs | 5 +- 10 files changed, 331 insertions(+), 10 deletions(-) create mode 100644 common/client-core/src/client/packet_statistics_control.rs diff --git a/common/client-core/src/client/base_client/mod.rs b/common/client-core/src/client/base_client/mod.rs index 07d676a6f3..47b119261a 100644 --- a/common/client-core/src/client/base_client/mod.rs +++ b/common/client-core/src/client/base_client/mod.rs @@ -1,6 +1,7 @@ // Copyright 2022-2023 - Nym Technologies SA // SPDX-License-Identifier: Apache-2.0 +use super::packet_statistics_control::PacketStatisticsReporter; use super::received_buffer::ReceivedBufferMessage; use super::topology_control::geo_aware_provider::GeoAwareTopologyProvider; use crate::client::base_client::storage::gateway_details::GatewayDetailsStore; @@ -10,6 +11,7 @@ use crate::client::inbound_messages::{InputMessage, InputMessageReceiver, InputM use crate::client::key_manager::persistence::KeyStore; use crate::client::mix_traffic::transceiver::{GatewayReceiver, GatewayTransceiver, RemoteGateway}; use crate::client::mix_traffic::{BatchMixMessageSender, MixTrafficController}; +use crate::client::packet_statistics_control::PacketStatisticsControl; use crate::client::real_messages_control; use crate::client::real_messages_control::RealMessagesController; use crate::client::received_buffer::{ @@ -254,6 +256,7 @@ where self_address: Recipient, topology_accessor: TopologyAccessor, mix_tx: BatchMixMessageSender, + stats_tx: PacketStatisticsReporter, shutdown: TaskClient, ) { info!("Starting loop cover traffic stream..."); @@ -266,6 +269,7 @@ where topology_accessor, debug_config.traffic, debug_config.cover_traffic, + stats_tx, ); stream.start_with_shutdown(shutdown); @@ -285,6 +289,7 @@ where client_connection_rx: ConnectionCommandReceiver, shutdown: TaskClient, packet_type: PacketType, + stats_tx: PacketStatisticsReporter, ) { info!("Starting real traffic stream..."); @@ -299,6 +304,7 @@ where reply_controller_receiver, lane_queue_lengths, client_connection_rx, + stats_tx, ) .start_with_shutdown(shutdown, packet_type); } @@ -312,6 +318,7 @@ where reply_key_storage: SentReplyKeys, reply_controller_sender: ReplyControllerSender, shutdown: TaskClient, + packet_statistics_control: PacketStatisticsReporter, ) { info!("Starting received messages buffer controller..."); let controller: ReceivedMessagesBufferController = @@ -321,6 +328,7 @@ where mixnet_receiver, reply_key_storage, reply_controller_sender, + packet_statistics_control, ); controller.start_with_shutdown(shutdown) } @@ -506,6 +514,13 @@ where Ok(()) } + fn start_packet_statistics_control(shutdown: TaskClient) -> PacketStatisticsReporter { + info!("Starting packet statistics control..."); + let (packet_statistics_control, packet_stats_reporter) = PacketStatisticsControl::new(); + packet_statistics_control.start_with_shutdown(shutdown); + packet_stats_reporter + } + fn start_mix_traffic_controller( gateway_transceiver: Box, shutdown: TaskClient, @@ -633,6 +648,9 @@ where ) .await?; + let packet_stats_reporter = + Self::start_packet_statistics_control(shutdown.fork("packet_statistics_control")); + let gateway_packet_router = PacketRouter::new( ack_sender, mixnet_messages_sender, @@ -662,6 +680,7 @@ where reply_storage.key_storage(), reply_controller_sender.clone(), shutdown.fork("received_messages_buffer"), + packet_stats_reporter.clone(), ); // The message_sender is the transmitter for any component generating sphinx packets @@ -700,6 +719,7 @@ where client_connection_rx, shutdown.fork("real_traffic_controller"), self.config.debug.traffic.packet_type, + packet_stats_reporter.clone(), ); if !self @@ -714,6 +734,7 @@ where self_address, shared_topology_accessor.clone(), message_sender, + packet_stats_reporter, shutdown.fork("cover_traffic_stream"), ); } diff --git a/common/client-core/src/client/cover_traffic_stream.rs b/common/client-core/src/client/cover_traffic_stream.rs index c603926933..101a0a7754 100644 --- a/common/client-core/src/client/cover_traffic_stream.rs +++ b/common/client-core/src/client/cover_traffic_stream.rs @@ -2,6 +2,7 @@ // SPDX-License-Identifier: Apache-2.0 use crate::client::mix_traffic::BatchMixMessageSender; +use crate::client::packet_statistics_control::{PacketStatisticsEvent, PacketStatisticsReporter}; use crate::client::topology_control::TopologyAccessor; use crate::{config, spawn_future}; use futures::task::{Context, Poll}; @@ -61,6 +62,8 @@ where secondary_packet_size: Option, packet_type: PacketType, + + stats_tx: PacketStatisticsReporter, } impl Stream for LoopCoverTrafficStream @@ -97,7 +100,8 @@ where // obviously when we finally make shared rng that is on 'higher' level, this should become // generic `R` impl LoopCoverTrafficStream { - pub fn new( + #[allow(clippy::too_many_arguments)] + pub(crate) fn new( ack_key: Arc, average_ack_delay: Duration, mix_tx: BatchMixMessageSender, @@ -105,6 +109,7 @@ impl LoopCoverTrafficStream { topology_access: TopologyAccessor, traffic_config: config::Traffic, cover_config: config::CoverTraffic, + stats_tx: PacketStatisticsReporter, ) -> Self { let rng = OsRng; @@ -122,6 +127,7 @@ impl LoopCoverTrafficStream { primary_packet_size: traffic_config.primary_packet_size, secondary_packet_size: traffic_config.secondary_packet_size, packet_type: traffic_config.packet_type, + stats_tx, } } @@ -191,6 +197,10 @@ impl LoopCoverTrafficStream { log::warn!("Failed to send cover message - channel closed"); } } + } else { + self.stats_tx.report(PacketStatisticsEvent::CoverPacketSent( + cover_traffic_packet_size.size(), + )); } // TODO: I'm not entirely sure whether this is really required, because I'm not 100% diff --git a/common/client-core/src/client/mod.rs b/common/client-core/src/client/mod.rs index 789dd300de..9a3a3a33c9 100644 --- a/common/client-core/src/client/mod.rs +++ b/common/client-core/src/client/mod.rs @@ -7,6 +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 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 new file mode 100644 index 0000000000..824f24a30b --- /dev/null +++ b/common/client-core/src/client/packet_statistics_control.rs @@ -0,0 +1,202 @@ +use std::time::Duration; + +use crate::spawn_future; + +// Time interval between reporting packet statistics +const PACKET_REPORT_INTERVAL_SECS: u64 = 2; + +#[derive(Default, Debug)] +struct PacketStatistics { + // Sent + real_packets_sent: u64, + real_packets_sent_size: usize, + cover_packets_sent: u64, + cover_packets_sent_size: usize, + + // Received + real_packets_received: u64, + real_packets_received_size: usize, + cover_packets_received: u64, + cover_packets_received_size: usize, + + // Acks + total_acks_received: u64, + total_acks_received_size: usize, + real_acks_received: u64, + real_acks_received_size: usize, + cover_acks_received: u64, + cover_acks_received_size: usize, + + // Types of packets queued + // TODO: track the type sent instead + real_packets_queued: u64, + retransmissions_queued: u64, + reply_surbs_queued: u64, + additional_reply_surbs_queued: u64, +} + +impl PacketStatistics { + fn handle_event(&mut self, event: PacketStatisticsEvent) { + match event { + PacketStatisticsEvent::RealPacketSent(packet_size) => { + self.real_packets_sent += 1; + self.real_packets_sent_size += packet_size; + } + PacketStatisticsEvent::CoverPacketSent(packet_size) => { + self.cover_packets_sent += 1; + self.cover_packets_sent_size += packet_size; + } + PacketStatisticsEvent::RealPacketReceived(packet_size) => { + self.real_packets_received += 1; + self.real_packets_received_size += packet_size; + } + PacketStatisticsEvent::CoverPacketReceived(packet_size) => { + self.cover_packets_received += 1; + self.cover_packets_received_size += packet_size; + } + PacketStatisticsEvent::AckReceived(packet_size) => { + self.total_acks_received += 1; + self.total_acks_received_size += packet_size; + } + PacketStatisticsEvent::RealAckReceived(packet_size) => { + self.real_acks_received += 1; + self.real_acks_received_size += packet_size; + } + PacketStatisticsEvent::CoverAckReceived(packet_size) => { + self.cover_acks_received += 1; + self.cover_acks_received_size += packet_size; + } + PacketStatisticsEvent::RealPacketQueued => { + self.real_packets_queued += 1; + } + PacketStatisticsEvent::RetransmissionQueued => { + self.retransmissions_queued += 1; + } + PacketStatisticsEvent::ReplySurbRequestQueued => { + self.reply_surbs_queued += 1; + } + PacketStatisticsEvent::AdditionalReplySurbRequestQueued => { + self.additional_reply_surbs_queued += 1; + } + } + } +} + +pub(crate) enum PacketStatisticsEvent { + // The real packets sent. Recall that acks are sent by the gateway, so it's not included here. + RealPacketSent(usize), + // The cover packets sent + CoverPacketSent(usize), + + // Real packets received + RealPacketReceived(usize), + // Cover packets received + CoverPacketReceived(usize), + + // Ack of any type received. This is mostly used as a consistency check, and should be the sum + // of real and cover acks received. + AckReceived(usize), + // Out of the total acks received, this is the subset of those that were real + RealAckReceived(usize), + // Out of the total acks received, this is the subset of those that were for cover traffic + CoverAckReceived(usize), + + // Types of packets queued + RealPacketQueued, + RetransmissionQueued, + ReplySurbRequestQueued, + AdditionalReplySurbRequestQueued, +} + +type PacketStatisticsReceiver = tokio::sync::mpsc::UnboundedReceiver; + +#[derive(Clone)] +pub(crate) struct PacketStatisticsReporter { + stats_tx: tokio::sync::mpsc::UnboundedSender, +} + +impl PacketStatisticsReporter { + pub(crate) fn new(stats_tx: tokio::sync::mpsc::UnboundedSender) -> Self { + Self { stats_tx } + } + + pub(crate) fn report(&self, event: PacketStatisticsEvent) { + self.stats_tx.send(event).unwrap_or_else(|err| { + log::error!("Failed to report packet stat: {:?}", err); + }); + } +} + +pub(crate) struct PacketStatisticsControl { + // Incoming packet stats events from other tasks + stats_rx: PacketStatisticsReceiver, + stats: PacketStatistics, +} + +impl PacketStatisticsControl { + pub(crate) fn new() -> (Self, PacketStatisticsReporter) { + let (stats_tx, stats_rx) = tokio::sync::mpsc::unbounded_channel(); + ( + Self { + stats_rx, + stats: PacketStatistics::default(), + }, + PacketStatisticsReporter::new(stats_tx), + ) + } + + fn report_statistics(&self) { + log::trace!("packet statistics: {:?}", &self.stats); + log::info!( + "packets sent: {} (real: {}, cover: {}, retransmissions: {})", + self.stats.real_packets_sent + self.stats.cover_packets_sent, + self.stats.real_packets_sent, + self.stats.cover_packets_sent, + self.stats.retransmissions_queued, + ); + log::info!( + "packets received: {}, (real: {}, cover: {}, acks: {}, acks for cover: {})", + self.stats.real_packets_received + self.stats.cover_packets_received, + self.stats.real_packets_received, + self.stats.cover_packets_received, + self.stats.real_acks_received, + self.stats.cover_acks_received, + ); + } + + pub(crate) async fn run_with_shutdown(&mut self, mut shutdown: nym_task::TaskClient) { + log::debug!("Started PacketStatisticsControl with graceful shutdown support"); + + let interval = Duration::from_secs(PACKET_REPORT_INTERVAL_SECS); + let mut interval = tokio::time::interval(interval); + + loop { + tokio::select! { + stats_event = self.stats_rx.recv() => match stats_event { + Some(stats_event) => { + log::trace!("PacketStatisticsControl: Received stats event"); + self.stats.handle_event(stats_event); + }, + None => { + log::trace!("PacketStatisticsControl: stopping since stats channel was closed"); + break; + } + }, + _ = interval.tick() => { + self.report_statistics(); + } + _ = shutdown.recv_with_delay() => { + log::trace!("PacketStatisticsControl: Received shutdown"); + break; + }, + } + } + log::debug!("PacketStatisticsControl: Exiting"); + } + + pub(crate) fn start_with_shutdown(mut self, task_client: nym_task::TaskClient) { + spawn_future(async move { + self.run_with_shutdown(task_client).await; + }) + } +} diff --git a/common/client-core/src/client/real_messages_control/acknowledgement_control/acknowledgement_listener.rs b/common/client-core/src/client/real_messages_control/acknowledgement_control/acknowledgement_listener.rs index 0034e00567..d7876c64ab 100644 --- a/common/client-core/src/client/real_messages_control/acknowledgement_control/acknowledgement_listener.rs +++ b/common/client-core/src/client/real_messages_control/acknowledgement_control/acknowledgement_listener.rs @@ -1,6 +1,8 @@ // Copyright 2021 - Nym Technologies SA // SPDX-License-Identifier: Apache-2.0 +use crate::client::packet_statistics_control::{PacketStatisticsEvent, PacketStatisticsReporter}; + use super::action_controller::{AckActionSender, Action}; use futures::StreamExt; use log::*; @@ -17,6 +19,7 @@ pub(super) struct AcknowledgementListener { ack_key: Arc, ack_receiver: AcknowledgementReceiver, action_sender: AckActionSender, + stats_tx: PacketStatisticsReporter, } impl AcknowledgementListener { @@ -24,16 +27,21 @@ impl AcknowledgementListener { ack_key: Arc, ack_receiver: AcknowledgementReceiver, action_sender: AckActionSender, + stats_tx: PacketStatisticsReporter, ) -> Self { AcknowledgementListener { ack_key, ack_receiver, action_sender, + stats_tx, } } async fn on_ack(&mut self, ack_content: Vec) { trace!("Received an ack"); + self.stats_tx + .report(PacketStatisticsEvent::AckReceived(ack_content.len())); + let frag_id = match recover_identifier(&self.ack_key, &ack_content) .map(FragmentIdentifier::try_from_bytes) { @@ -48,11 +56,14 @@ impl AcknowledgementListener { // because nothing was inserted in the first place if frag_id == COVER_FRAG_ID { trace!("Received an ack for a cover message - no need to do anything"); + self.stats_tx + .report(PacketStatisticsEvent::CoverAckReceived(ack_content.len())); return; } trace!("Received {} from the mix network", frag_id); - + self.stats_tx + .report(PacketStatisticsEvent::RealAckReceived(ack_content.len())); self.action_sender .unbounded_send(Action::new_remove(frag_id)) .unwrap(); diff --git a/common/client-core/src/client/real_messages_control/acknowledgement_control/mod.rs b/common/client-core/src/client/real_messages_control/acknowledgement_control/mod.rs index d7a9eac0ab..ec35429b9f 100644 --- a/common/client-core/src/client/real_messages_control/acknowledgement_control/mod.rs +++ b/common/client-core/src/client/real_messages_control/acknowledgement_control/mod.rs @@ -8,6 +8,7 @@ use self::{ sent_notification_listener::SentNotificationListener, }; use crate::client::inbound_messages::InputMessageReceiver; +use crate::client::packet_statistics_control::PacketStatisticsReporter; use crate::client::real_messages_control::message_handler::MessageHandler; use crate::client::replies::reply_controller::ReplyControllerSender; use crate::spawn_future; @@ -208,6 +209,7 @@ where connectors: AcknowledgementControllerConnectors, message_handler: MessageHandler, reply_controller_sender: ReplyControllerSender, + stats_tx: PacketStatisticsReporter, ) -> Self { let (retransmission_tx, retransmission_rx) = mpsc::unbounded(); @@ -224,6 +226,7 @@ where Arc::clone(&ack_key), connectors.ack_receiver, connectors.ack_action_sender.clone(), + stats_tx, ); // will listen for any new messages from the client diff --git a/common/client-core/src/client/real_messages_control/mod.rs b/common/client-core/src/client/real_messages_control/mod.rs index 9c57798fe7..d4914f1440 100644 --- a/common/client-core/src/client/real_messages_control/mod.rs +++ b/common/client-core/src/client/real_messages_control/mod.rs @@ -35,6 +35,8 @@ use crate::client::replies::reply_controller; use crate::config; pub(crate) use acknowledgement_control::{AckActionSender, Action}; +use super::packet_statistics_control::PacketStatisticsReporter; + pub(crate) mod acknowledgement_control; pub(crate) mod message_handler; pub(crate) mod real_traffic_stream; @@ -143,6 +145,7 @@ impl RealMessagesController { reply_controller_receiver: ReplyControllerReceiver, lane_queue_lengths: LaneQueueLengths, client_connection_rx: ConnectionCommandReceiver, + stats_tx: PacketStatisticsReporter, ) -> Self { let rng = OsRng; @@ -181,6 +184,7 @@ impl RealMessagesController { ack_controller_connectors, message_handler.clone(), reply_controller_sender, + stats_tx.clone(), ); let reply_control = ReplyController::new( @@ -199,6 +203,7 @@ impl RealMessagesController { topology_access, lane_queue_lengths, client_connection_rx, + stats_tx, ); RealMessagesController { diff --git a/common/client-core/src/client/real_messages_control/real_traffic_stream.rs b/common/client-core/src/client/real_messages_control/real_traffic_stream.rs index 7874d758f7..ef91208f1b 100644 --- a/common/client-core/src/client/real_messages_control/real_traffic_stream.rs +++ b/common/client-core/src/client/real_messages_control/real_traffic_stream.rs @@ -3,6 +3,7 @@ use self::sending_delay_controller::SendingDelayController; use crate::client::mix_traffic::BatchMixMessageSender; +use crate::client::packet_statistics_control::{PacketStatisticsEvent, PacketStatisticsReporter}; use crate::client::real_messages_control::acknowledgement_control::SentPacketNotificationSender; use crate::client::topology_control::TopologyAccessor; use crate::client::transmission_buffer::TransmissionBuffer; @@ -113,6 +114,9 @@ where /// Report queue lengths so that upstream can backoff sending data, and keep connections open. lane_queue_lengths: LaneQueueLengths, + + /// Channel used for sending statistics events to `PacketStatisticsControl`. + stats_tx: PacketStatisticsReporter, } #[derive(Debug)] @@ -171,6 +175,7 @@ where topology_access: TopologyAccessor, lane_queue_lengths: LaneQueueLengths, client_connection_rx: ConnectionCommandReceiver, + stats_tx: PacketStatisticsReporter, ) -> Self { OutQueueControl { config, @@ -184,6 +189,7 @@ where transmission_buffer: TransmissionBuffer::new(), client_connection_rx, lane_queue_lengths, + stats_tx, } } @@ -214,7 +220,7 @@ where async fn on_message(&mut self, next_message: StreamMessage) { trace!("created new message"); - let (next_message, fragment_id) = match next_message { + let (next_message, fragment_id, packet_size) = match next_message { StreamMessage::Cover => { let cover_traffic_packet_size = self.loop_cover_message_size(); trace!("the next loop cover message will be put in a {cover_traffic_packet_size} packet"); @@ -250,15 +256,28 @@ where "Somehow failed to generate a loop cover message with a valid topology", ), None, + cover_traffic_packet_size.size(), ) } StreamMessage::Real(real_message) => { - (real_message.mix_packet, real_message.fragment_id) + let packet_size = real_message.packet_size(); + ( + real_message.mix_packet, + real_message.fragment_id, + packet_size, + ) } }; if let Err(err) = self.mix_tx.send(vec![next_message]).await { log::error!("Failed to send: {err}"); + } else { + let event = if fragment_id.is_some() { + PacketStatisticsEvent::RealPacketSent(packet_size) + } else { + PacketStatisticsEvent::CoverPacketSent(packet_size) + }; + self.stats_tx.report(event); } // notify ack controller about sending our message only after we actually managed to push it @@ -340,6 +359,28 @@ where let lane_length = self.transmission_buffer.lane_length(&lane); self.lane_queue_lengths.set(&lane, lane_length); + // This is the last step in the pipeline where we know the type of the message, so + // lets count the number of retransmissions and reply surb messages sent here. + let stat_event = match lane { + TransmissionLane::General => None, + TransmissionLane::ConnectionId(_) => None, + TransmissionLane::ReplySurbRequest => { + Some(PacketStatisticsEvent::ReplySurbRequestQueued) + } + TransmissionLane::AdditionalReplySurbs => { + Some(PacketStatisticsEvent::AdditionalReplySurbRequestQueued) + } + TransmissionLane::Retransmission => Some(PacketStatisticsEvent::RetransmissionQueued), + }; + if let Some(stat_event) = stat_event { + self.stats_tx.report(stat_event); + } + // To avoid comparing apples to oranges when presenting the fraction of packets that are + // retransmissions, we also need to keep track to the total number of real messages queued, + // even though we also track the actual number of messages sent later in the pipeline. + self.stats_tx + .report(PacketStatisticsEvent::RealPacketQueued); + Some(real_next) } @@ -433,6 +474,13 @@ where Poll::Ready(Some((real_messages, conn_id))) => { log::trace!("handling real_messages: size: {}", real_messages.len()); + // This is the last step in the pipeline where we know the type of the message, so + // lets count the number of retransmissions here. + if conn_id == TransmissionLane::Retransmission { + self.stats_tx + .report(PacketStatisticsEvent::RetransmissionQueued); + } + // First store what we got for the given connection id self.transmission_buffer.store(&conn_id, real_messages); let real_next = self.pop_next_message().expect("we just added one"); @@ -471,10 +519,10 @@ where let mult = self.sending_delay_controller.current_multiplier(); let delay = self.current_average_message_sending_delay().as_millis(); let status_str = if self.config.traffic.disable_main_poisson_packet_distribution { - format!("Status: {lanes} lanes, backlog: {backlog:.2} kiB ({packets}), no delay") + format!("Packet backlog: {backlog:.2} kiB ({packets}), {lanes} lanes, no delay") } else { format!( - "Status: {lanes} lanes, backlog: {backlog:.2} kiB ({packets}), avg delay: {delay}ms ({mult})" + "Packet backlog: {backlog:.2} kiB ({packets}), {lanes} lanes, avg delay: {delay}ms ({mult})" ) }; if packets > 1000 { diff --git a/common/client-core/src/client/received_buffer.rs b/common/client-core/src/client/received_buffer.rs index 3421ed86ff..fe6887b01f 100644 --- a/common/client-core/src/client/received_buffer.rs +++ b/common/client-core/src/client/received_buffer.rs @@ -1,8 +1,10 @@ // Copyright 2021 - Nym Technologies SA // SPDX-License-Identifier: Apache-2.0 -use crate::client::replies::reply_controller::ReplyControllerSender; -use crate::client::replies::reply_storage::SentReplyKeys; +use crate::client::{ + packet_statistics_control::{PacketStatisticsEvent, PacketStatisticsReporter}, + replies::{reply_controller::ReplyControllerSender, reply_storage::SentReplyKeys}, +}; use crate::spawn_future; use futures::channel::mpsc; use futures::lock::Mutex; @@ -43,15 +45,26 @@ struct ReceivedMessagesBufferInner { // but perhaps it should be changed to include timestamps of when the message was reconstructed // and every now and then remove ids older than X recently_reconstructed: HashSet, + + stats_tx: PacketStatisticsReporter, } impl ReceivedMessagesBufferInner { fn recover_from_fragment(&mut self, fragment_data: &[u8]) -> Option { if nym_sphinx::cover::is_cover(fragment_data) { trace!("The message was a loop cover message! Skipping it"); + self.stats_tx + .report(PacketStatisticsEvent::CoverPacketReceived( + fragment_data.len(), + )); return None; } + self.stats_tx + .report(PacketStatisticsEvent::RealPacketReceived( + fragment_data.len(), + )); + let fragment = match self.message_receiver.recover_fragment(fragment_data) { Err(err) => { warn!("failed to recover fragment from raw data: {err}. The whole underlying message might be corrupted and unrecoverable!"); @@ -141,6 +154,7 @@ impl ReceivedMessagesBuffer { local_encryption_keypair: Arc, reply_key_storage: SentReplyKeys, reply_controller_sender: ReplyControllerSender, + stats_tx: PacketStatisticsReporter, ) -> Self { ReceivedMessagesBuffer { inner: Arc::new(Mutex::new(ReceivedMessagesBufferInner { @@ -149,6 +163,7 @@ impl ReceivedMessagesBuffer { message_receiver: R::new(), message_sender: None, recently_reconstructed: HashSet::new(), + stats_tx, })), reply_key_storage, reply_controller_sender, @@ -353,7 +368,7 @@ impl ReceivedMessagesBuffer { }; if let Some(completed) = completed_message { - info!("received {completed}"); + debug!("received {completed}"); completed_messages.push(completed) } } @@ -480,11 +495,13 @@ impl ReceivedMessagesBufferControll mixnet_packet_receiver: MixnetMessageReceiver, reply_key_storage: SentReplyKeys, reply_controller_sender: ReplyControllerSender, + packet_statistics_reporter: PacketStatisticsReporter, ) -> Self { let received_buffer = ReceivedMessagesBuffer::new( local_encryption_keypair, reply_key_storage, reply_controller_sender, + packet_statistics_reporter, ); ReceivedMessagesBufferController { diff --git a/common/client-libs/gateway-client/src/client.rs b/common/client-libs/gateway-client/src/client.rs index f65e61c29f..f8803bcc12 100644 --- a/common/client-libs/gateway-client/src/client.rs +++ b/common/client-libs/gateway-client/src/client.rs @@ -354,7 +354,10 @@ impl GatewayClient { msg: Message, ) -> Result<(), GatewayClientError> { match self.connection { - SocketState::Available(ref mut conn) => Ok(conn.send(msg).await?), + SocketState::Available(ref mut conn) => { + conn.send(msg).await?; + Ok(()) + } SocketState::PartiallyDelegated(ref mut partially_delegated) => { if let Err(err) = partially_delegated.send_without_response(msg).await { error!("failed to send message without response - {err}...");