diff --git a/common/client-core/config-types/src/lib.rs b/common/client-core/config-types/src/lib.rs index 8347a38c0e..8eb95d62f1 100644 --- a/common/client-core/config-types/src/lib.rs +++ b/common/client-core/config-types/src/lib.rs @@ -45,6 +45,7 @@ const DEFAULT_COVER_TRAFFIC_PRIMARY_SIZE_RATIO: f64 = 0.70; // clients/client-core/src/client/replies/reply_storage/surb_storage.rs const DEFAULT_MINIMUM_REPLY_SURB_STORAGE_THRESHOLD: usize = 10; const DEFAULT_MAXIMUM_REPLY_SURB_STORAGE_THRESHOLD: usize = 200; +const DEFAULT_MINIMUM_REPLY_SURB_THRESHOLD_BUFFER: usize = 0; // define how much to request at once // clients/client-core/src/client/replies/reply_controller.rs @@ -621,6 +622,10 @@ pub struct ReplySurbs { /// Defines the maximum number of reply surbs the client wants to keep in its storage at any times. pub maximum_reply_surb_storage_threshold: usize, + /// Defines the soft threshold ontop of the minimum reply surb storage threshold for when the client + /// should proactively request additional reply surbs. + pub minimum_reply_surb_threshold_buffer: usize, + /// Defines the minimum number of reply surbs the client would request. pub minimum_reply_surb_request_size: u32, @@ -660,6 +665,7 @@ impl Default for ReplySurbs { ReplySurbs { minimum_reply_surb_storage_threshold: DEFAULT_MINIMUM_REPLY_SURB_STORAGE_THRESHOLD, maximum_reply_surb_storage_threshold: DEFAULT_MAXIMUM_REPLY_SURB_STORAGE_THRESHOLD, + minimum_reply_surb_threshold_buffer: DEFAULT_MINIMUM_REPLY_SURB_THRESHOLD_BUFFER, minimum_reply_surb_request_size: DEFAULT_MINIMUM_REPLY_SURB_REQUEST_SIZE, maximum_reply_surb_request_size: DEFAULT_MAXIMUM_REPLY_SURB_REQUEST_SIZE, maximum_allowed_reply_surb_request_size: DEFAULT_MAXIMUM_ALLOWED_SURB_REQUEST_SIZE, diff --git a/common/client-core/config-types/src/old/v5.rs b/common/client-core/config-types/src/old/v5.rs index df673a0986..872d107cc5 100644 --- a/common/client-core/config-types/src/old/v5.rs +++ b/common/client-core/config-types/src/old/v5.rs @@ -181,6 +181,7 @@ impl From for Config { maximum_reply_surb_age: value.debug.reply_surbs.maximum_reply_surb_age, maximum_reply_key_age: value.debug.reply_surbs.maximum_reply_key_age, surb_mix_hops: value.debug.reply_surbs.surb_mix_hops, + ..Default::default() }, ..Default::default() }, 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 2789228dcf..d20b8040aa 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 @@ -517,17 +517,25 @@ where use crate::error::ClientCoreStatusMessage; let packets = self.transmission_buffer.total_size(); - let backlog = self.transmission_buffer.total_size_in_bytes() as f64 / 1024.0; - let lanes = self.transmission_buffer.num_lanes(); + let lanes = self.transmission_buffer.lanes(); let mult = self.sending_delay_controller.current_multiplier(); let delay = self.current_average_message_sending_delay().as_millis(); + + let lane_status = lanes + .iter() + .map(|lane_name| { + let lane_length = self.transmission_buffer.lane_length(lane_name).unwrap_or(0); + format!("{lane_name:?}: {lane_length}") + }) + .collect::>() + .join(", "); + let status_str = if self.config.traffic.disable_main_poisson_packet_distribution { - format!("Packet backlog: {backlog:.2} kiB ({packets}), {lanes} lanes, no delay") + format!("Packet backlog: {lane_status}, no delay") } else { - format!( - "Packet backlog: {backlog:.2} kiB ({packets}), {lanes} lanes, avg delay: {delay}ms ({mult})" - ) + format!("Packet backlog: {lane_status}, avg delay: {delay}ms ({mult})") }; + if packets > 1000 { log::warn!("{status_str}"); } else if packets > 0 { diff --git a/common/client-core/src/client/replies/reply_controller/mod.rs b/common/client-core/src/client/replies/reply_controller/mod.rs index 085fbfa6e4..fb8cdc468b 100644 --- a/common/client-core/src/client/replies/reply_controller/mod.rs +++ b/common/client-core/src/client/replies/reply_controller/mod.rs @@ -102,6 +102,7 @@ where fragments: I, lane: TransmissionLane, ) { + trace!("buffering pending replies for {recipient}"); self.pending_replies .entry(*recipient) .or_insert_with(TransmissionBuffer::new) @@ -113,6 +114,7 @@ where recipient: &AnonymousSenderTag, fragments: Vec<(TransmissionLane, Fragment)>, ) { + trace!("re-inserting pending replies for {recipient}"); // the buffer should ALWAYS exist at this point, if it doesn't, it's a bug... self.pending_replies .entry(*recipient) @@ -125,6 +127,7 @@ where recipient: &AnonymousSenderTag, data: Vec>, ) { + trace!("re-inserting pending retransmissions for {recipient}"); // the underlying entry MUST exist as we've just got data from there let map_entry = self .pending_retransmissions @@ -142,7 +145,7 @@ where } fn should_request_more_surbs(&self, target: &AnonymousSenderTag) -> bool { - trace!("checking if we should request more surbs from {:?}", target); + trace!("checking if we should request more surbs from {target}"); let pending_queue_size = self .pending_replies @@ -158,11 +161,6 @@ where let total_queue = pending_queue_size + retransmission_queue; - // simple as that - there's absolutely nothing to retransmit - if total_queue == 0 { - return false; - } - let available_surbs = self .full_reply_storage .surbs_storage_ref() @@ -179,11 +177,27 @@ where .full_reply_storage .surbs_storage_ref() .max_surb_threshold(); + let min_surbs_threshold_buffer = + self.config.reply_surbs.minimum_reply_surb_threshold_buffer; - debug!("total queue size: {total_queue} = pending data {pending_queue_size} + pending retransmission {retransmission_queue}, available surbs: {available_surbs} pending surbs: {pending_surbs} threshold range: {min_surbs_threshold}..{max_surbs_threshold}"); + // After clearing the queue, we want to have at least `min_surbs_threshold` surbs available + // and reserved for requesting additional surbs, and in addition to that we also want to + // have `min_surbs_threshold_buffer` surbs available proactively. + let target_surbs_after_clearing_queue = min_surbs_threshold + min_surbs_threshold_buffer; - (pending_surbs + available_surbs) < max_surbs_threshold - && (pending_surbs + available_surbs) < (total_queue + min_surbs_threshold) + // Check if we have enough surbs to handle the total queue and maintain minimum thresholds + let total_required_surbs = total_queue + target_surbs_after_clearing_queue; + let total_available_surbs = pending_surbs + available_surbs; + + debug!("total queue size: {total_queue} = pending data {pending_queue_size} + pending retransmission {retransmission_queue}, available surbs: {available_surbs} pending surbs: {pending_surbs} threshold range: {min_surbs_threshold}..+{min_surbs_threshold_buffer}..{max_surbs_threshold}"); + + // We should request more surbs if: + // 1. We haven't hit the maximum surb threshold, and + // 2. We don't have enough surbs to handle the queue plus minimum thresholds + let is_below_max_threshold = total_available_surbs < max_surbs_threshold; + let is_below_required_surbs = total_available_surbs < total_required_surbs; + + is_below_max_threshold && is_below_required_surbs } async fn handle_send_reply( @@ -244,6 +258,10 @@ where &recipient_tag, ); warn!("failed to send reply to {recipient_tag}: {err}"); + info!( + "buffering {no_fragments} fragments for {recipient_tag}", + no_fragments = to_send.len() + ); self.insert_pending_replies(&recipient_tag, to_send, lane); } } @@ -251,6 +269,13 @@ where // if there's leftover data we didn't send because we didn't have enough (or any) surbs - buffer it if !fragments.is_empty() { + // Ideally we should have enough surbs above the minimum threshold to handle sending + // new replies without having to first request more surbs. That's why I'd like to log + // these cases as they might indicate a problem with the surb management. + debug!( + "buffering {no_fragments} fragments for {recipient_tag}", + no_fragments = fragments.len() + ); self.insert_pending_replies(&recipient_tag, fragments, lane); } @@ -265,6 +290,7 @@ where target: AnonymousSenderTag, amount: u32, ) -> Result<(), PreparationError> { + debug!("requesting {amount} additional reply surbs for {target}"); let reply_surb = self .full_reply_storage .surbs_storage_ref() @@ -686,7 +712,7 @@ where // it should take into consideration the average latency, sending rate and queue size. // it should request as many surbs as it takes to saturate its sending rate before next batch arrives async fn request_reply_surbs_for_queue_clearing(&mut self, target: AnonymousSenderTag) { - trace!("requesting surbs for queues clearing"); + trace!("requesting surbs for queue clearing"); let pending_queue_size = self .pending_replies @@ -700,17 +726,18 @@ where .map(|pending_queue| pending_queue.len()) .unwrap_or_default(); + let min_surbs_buffer = self.config.reply_surbs.minimum_reply_surb_threshold_buffer as u32; + let total_queue = (pending_queue_size + retransmission_queue) as u32; - if total_queue == 0 { - trace!("the pending queues for {:?} are already empty", target); - return; - } + // To proactively request additional surbs, we aim to have a buffer of extra surbs in our + // storage. + let total_queue_with_buffer = total_queue + min_surbs_buffer; let request_size = min( self.config.reply_surbs.maximum_reply_surb_request_size, max( - total_queue, + total_queue_with_buffer, self.config.reply_surbs.minimum_reply_surb_request_size, ), ); diff --git a/common/client-core/src/client/transmission_buffer.rs b/common/client-core/src/client/transmission_buffer.rs index d6ee86e861..e6eb7d6891 100644 --- a/common/client-core/src/client/transmission_buffer.rs +++ b/common/client-core/src/client/transmission_buffer.rs @@ -58,8 +58,8 @@ impl TransmissionBuffer { } #[cfg(not(target_arch = "wasm32"))] - pub(crate) fn num_lanes(&self) -> usize { - self.buffer.keys().count() + pub(crate) fn lanes(&self) -> Vec { + self.buffer.keys().cloned().collect() } pub(crate) fn lane_length(&self, lane: &TransmissionLane) -> Option { @@ -83,6 +83,7 @@ impl TransmissionBuffer { } #[cfg(not(target_arch = "wasm32"))] + #[allow(unused)] pub(crate) fn total_size_in_bytes(&self) -> usize where T: SizedData, diff --git a/common/wasm/client-core/src/config/mod.rs b/common/wasm/client-core/src/config/mod.rs index ecd7be7927..64b8be5d40 100644 --- a/common/wasm/client-core/src/config/mod.rs +++ b/common/wasm/client-core/src/config/mod.rs @@ -462,6 +462,10 @@ pub struct ReplySurbsWasm { /// Defines the maximum number of reply surbs the client wants to keep in its storage at any times. pub maximum_reply_surb_storage_threshold: usize, + /// Defines the soft threshold ontop of the minimum reply surb storage threshold for when the client + /// should proactively request additional reply surbs. + pub minimum_reply_surb_threshold_buffer: usize, + /// Defines the minimum number of reply surbs the client would request. pub minimum_reply_surb_request_size: u32, @@ -503,6 +507,7 @@ impl From for ConfigReplySurbs { ConfigReplySurbs { minimum_reply_surb_storage_threshold: reply_surbs.minimum_reply_surb_storage_threshold, maximum_reply_surb_storage_threshold: reply_surbs.maximum_reply_surb_storage_threshold, + minimum_reply_surb_threshold_buffer: reply_surbs.minimum_reply_surb_threshold_buffer, minimum_reply_surb_request_size: reply_surbs.minimum_reply_surb_request_size, maximum_reply_surb_request_size: reply_surbs.maximum_reply_surb_request_size, maximum_allowed_reply_surb_request_size: reply_surbs @@ -529,6 +534,7 @@ impl From for ReplySurbsWasm { ReplySurbsWasm { minimum_reply_surb_storage_threshold: reply_surbs.minimum_reply_surb_storage_threshold, maximum_reply_surb_storage_threshold: reply_surbs.maximum_reply_surb_storage_threshold, + minimum_reply_surb_threshold_buffer: reply_surbs.minimum_reply_surb_threshold_buffer, minimum_reply_surb_request_size: reply_surbs.minimum_reply_surb_request_size, maximum_reply_surb_request_size: reply_surbs.maximum_reply_surb_request_size, maximum_allowed_reply_surb_request_size: reply_surbs diff --git a/common/wasm/client-core/src/config/override.rs b/common/wasm/client-core/src/config/override.rs index 0a485e49c7..72da8bd43d 100644 --- a/common/wasm/client-core/src/config/override.rs +++ b/common/wasm/client-core/src/config/override.rs @@ -339,6 +339,11 @@ pub struct ReplySurbsWasmOverride { #[tsify(optional)] pub maximum_reply_surb_storage_threshold: Option, + /// Defines the soft threshold ontop of the minimum reply surb storage threshold for when the client + /// should proactively request additional reply surbs. + #[tsify(optional)] + pub minimum_reply_surb_threshold_buffer: Option, + /// Defines the minimum number of reply surbs the client would request. #[tsify(optional)] pub minimum_reply_surb_request_size: Option, @@ -386,6 +391,9 @@ impl From for ReplySurbsWasm { maximum_reply_surb_storage_threshold: value .maximum_reply_surb_storage_threshold .unwrap_or(def.maximum_reply_surb_storage_threshold), + minimum_reply_surb_threshold_buffer: value + .minimum_reply_surb_threshold_buffer + .unwrap_or(def.minimum_reply_surb_threshold_buffer), minimum_reply_surb_request_size: value .minimum_reply_surb_request_size .unwrap_or(def.minimum_reply_surb_request_size), diff --git a/nym-node/src/config/service_providers.rs b/nym-node/src/config/service_providers.rs index c870d1947a..b7d2f2cb77 100644 --- a/nym-node/src/config/service_providers.rs +++ b/nym-node/src/config/service_providers.rs @@ -3,12 +3,17 @@ use crate::config::authenticator::Authenticator; use crate::config::persistence::ServiceProvidersPaths; -use nym_client_core_config_types::DebugConfig as ClientDebugConfig; +use nym_client_core_config_types::{DebugConfig as ClientDebugConfig, ReplySurbs}; use nym_config::defaults::mainnet; use serde::{Deserialize, Serialize}; use std::path::Path; use url::Url; +// Default reply surb storage threshold. +// It is set higher than the default since the traffic is mostly a stream of data, +// and not a bursty request/response pattern. +pub const DEFAULT_IPR_MINIMUM_REPLY_SURB_THRESHOLD_BUFFER: usize = 50; + #[derive(Debug, Clone, Serialize, Deserialize)] #[serde(deny_unknown_fields)] pub struct ServiceProvidersConfig { @@ -145,10 +150,18 @@ pub struct IpPacketRouterDebug { impl Default for IpPacketRouterDebug { fn default() -> Self { + let client_debug = ClientDebugConfig { + reply_surbs: ReplySurbs { + minimum_reply_surb_threshold_buffer: + DEFAULT_IPR_MINIMUM_REPLY_SURB_THRESHOLD_BUFFER, + ..Default::default() + }, + ..Default::default() + }; IpPacketRouterDebug { enabled: true, disable_poisson_rate: true, - client_debug: Default::default(), + client_debug, } } }