From dadfdacf146b848ae2db3a207c68db1fb6b468a1 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?J=C4=99drzej=20Stuczy=C5=84ski?= Date: Wed, 22 Mar 2023 15:53:03 +0000 Subject: [PATCH] weighted decision on cover traffic packet size ('real traffic stream') --- .../src/client/real_messages_control/mod.rs | 20 ++---- .../real_traffic_stream.rs | 67 +++++++++--------- .../client/replies/reply_controller/mod.rs | 69 ++++++++++--------- common/client-core/src/config/mod.rs | 4 +- 4 files changed, 81 insertions(+), 79 deletions(-) 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 9771b17908..bfb3c0e1fc 100644 --- a/common/client-core/src/client/real_messages_control/mod.rs +++ b/common/client-core/src/client/real_messages_control/mod.rs @@ -49,6 +49,9 @@ pub struct Config { /// Specifies all traffic related configuration options. traffic: config::Traffic, + /// Specifies all cover traffic related configuration options. + cover_traffic: config::CoverTraffic, + /// Specifies all acknowledgements related configuration options. acks: config::Acknowledgements, @@ -72,25 +75,15 @@ impl<'a> From<&'a Config> for real_traffic_stream::Config { Arc::clone(&cfg.ack_key), cfg.self_recipient, cfg.acks.average_ack_delay, - cfg.traffic.average_packet_delay, - cfg.traffic.message_sending_average_delay, - cfg.traffic.disable_main_poisson_packet_distribution, + cfg.traffic, + cfg.cover_traffic.cover_traffic_primary_size_ratio, ) - .with_custom_cover_packet_size(cfg.traffic.primary_packet_size) } } impl<'a> From<&'a Config> for reply_controller::Config { fn from(cfg: &'a Config) -> Self { - reply_controller::Config::new( - cfg.reply_surbs.minimum_reply_surb_request_size, - cfg.reply_surbs.maximum_reply_surb_request_size, - cfg.reply_surbs.maximum_allowed_reply_surb_request_size, - cfg.reply_surbs.maximum_reply_surb_rerequest_waiting_period, - cfg.reply_surbs.maximum_reply_surb_drop_waiting_period, - cfg.reply_surbs.maximum_reply_surb_age, - cfg.reply_surbs.maximum_reply_key_age, - ) + reply_controller::Config::new(cfg.reply_surbs) } } @@ -117,6 +110,7 @@ impl Config { ack_key, self_recipient, traffic: base_client_debug_config.traffic, + cover_traffic: base_client_debug_config.cover_traffic, acks: base_client_debug_config.acknowledgements, reply_surbs: base_client_debug_config.reply_surbs, } 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 4bcebe75d0..bd65ee7f59 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 @@ -6,6 +6,7 @@ use crate::client::mix_traffic::BatchMixMessageSender; use crate::client::real_messages_control::acknowledgement_control::SentPacketNotificationSender; use crate::client::topology_control::TopologyAccessor; use crate::client::transmission_buffer::TransmissionBuffer; +use crate::config; use futures::task::{Context, Poll}; use futures::{Future, Stream, StreamExt}; use log::*; @@ -27,7 +28,6 @@ use std::time::Duration; #[cfg(not(target_arch = "wasm32"))] use tokio::time; - #[cfg(target_arch = "wasm32")] use wasm_timer; @@ -44,18 +44,12 @@ pub(crate) struct Config { /// Average delay an acknowledgement packet is going to get delay at a single mixnode. average_ack_delay: Duration, - /// Average delay a data packet is going to get delay at a single mixnode. - average_packet_delay: Duration, + /// Defines all configuration options related to this traffic stream. + traffic: config::Traffic, - /// Average delay between sending subsequent packets. - average_message_sending_delay: Duration, - - /// Controls whether the stream constantly produces packets according to the predefined - /// poisson distribution. - disable_poisson_packet_distribution: bool, - - /// Predefined packet size used for the loop cover messages. - cover_packet_size: PacketSize, + /// Specifies the ratio of `primary_packet_size` to `secondary_packet_size` used in cover traffic. + /// Only applicable if `secondary_packet_size` is enabled. + cover_traffic_primary_size_ratio: f64, } impl Config { @@ -63,25 +57,17 @@ impl Config { ack_key: Arc, our_full_destination: Recipient, average_ack_delay: Duration, - average_packet_delay: Duration, - average_message_sending_delay: Duration, - disable_poisson_packet_distribution: bool, + traffic: config::Traffic, + cover_traffic_primary_size_ratio: f64, ) -> Self { Config { ack_key, our_full_destination, average_ack_delay, - average_packet_delay, - average_message_sending_delay, - disable_poisson_packet_distribution, - cover_packet_size: Default::default(), + traffic, + cover_traffic_primary_size_ratio, } } - - pub fn with_custom_cover_packet_size(mut self, packet_size: PacketSize) -> Self { - self.cover_packet_size = packet_size; - self - } } pub(crate) struct OutQueueControl @@ -212,11 +198,28 @@ where self.sent_notifier.unbounded_send(frag_id).unwrap(); } + fn loop_cover_message_size(&mut self) -> PacketSize { + let Some(secondary_packet_size) = self.config.traffic.secondary_packet_size else { + return self.config.traffic.primary_packet_size + }; + + if self + .rng + .gen_bool(self.config.cover_traffic_primary_size_ratio) + { + self.config.traffic.primary_packet_size + } else { + secondary_packet_size + } + } + async fn on_message(&mut self, next_message: StreamMessage) { trace!("created new message"); let (next_message, fragment_id) = match next_message { StreamMessage::Cover => { + let cover_traffic_packet_size = self.loop_cover_message_size(); + // TODO for way down the line: in very rare cases (during topology update) we might have // to wait a really tiny bit before actually obtaining the permit hence messing with our // poisson delay, but is it really a problem? @@ -240,8 +243,8 @@ where &self.config.ack_key, &self.config.our_full_destination, self.config.average_ack_delay, - self.config.average_packet_delay, - self.config.cover_packet_size, + self.config.traffic.average_packet_delay, + cover_traffic_packet_size, ) .expect( "Somehow failed to generate a loop cover message with a valid topology", @@ -286,7 +289,7 @@ where } fn current_average_message_sending_delay(&self) -> Duration { - self.config.average_message_sending_delay + self.config.traffic.message_sending_average_delay * self.sending_delay_controller.current_multiplier() } @@ -400,8 +403,10 @@ where // we never set an initial delay - let's do it now cx.waker().wake_by_ref(); - let sampled = - sample_poisson_duration(&mut self.rng, self.config.average_message_sending_delay); + let sampled = sample_poisson_duration( + &mut self.rng, + self.config.traffic.message_sending_average_delay, + ); #[cfg(not(target_arch = "wasm32"))] let next_delay = Box::pin(time::sleep(sampled)); @@ -452,7 +457,7 @@ where mut self: Pin<&mut Self>, cx: &mut Context<'_>, ) -> Poll> { - if self.config.disable_poisson_packet_distribution { + if self.config.traffic.disable_main_poisson_packet_distribution { self.poll_immediate(cx) } else { self.poll_poisson(cx) @@ -468,7 +473,7 @@ where let lanes = self.transmission_buffer.num_lanes(); let mult = self.sending_delay_controller.current_multiplier(); let delay = self.current_average_message_sending_delay().as_millis(); - let status_str = if self.config.disable_poisson_packet_distribution { + let status_str = if self.config.traffic.disable_main_poisson_packet_distribution { format!("Status: {lanes} lanes, backlog: {backlog:.2} kiB ({packets}), no delay") } else { format!( 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 267e2ada1f..21e1f5a9d1 100644 --- a/common/client-core/src/client/replies/reply_controller/mod.rs +++ b/common/client-core/src/client/replies/reply_controller/mod.rs @@ -22,38 +22,21 @@ use time::OffsetDateTime; use crate::client::helpers::new_interval_stream; use crate::client::transmission_buffer::TransmissionBuffer; +use crate::config; pub(crate) use requests::{ReplyControllerMessage, ReplyControllerReceiver, ReplyControllerSender}; pub mod requests; +// this is still left as a separate config so I wouldn't need to replace it everywhere +// plus its not unreasonable to think that we might need something outside config::ReplySurbs struct pub struct Config { - min_surb_request_size: u32, - max_surb_request_size: u32, - maximum_allowed_reply_surb_request_size: u32, - max_surb_rerequest_waiting_period: Duration, - max_surb_drop_waiting_period: Duration, - max_reply_surb_age: Duration, - max_reply_key_age: Duration, + reply_surbs: config::ReplySurbs, } impl Config { - pub(crate) fn new( - min_surb_request_size: u32, - max_surb_request_size: u32, - maximum_allowed_reply_surb_request_size: u32, - max_surb_rerequest_waiting_period: Duration, - max_surb_drop_waiting_period: Duration, - max_reply_surb_age: Duration, - max_reply_key_age: Duration, - ) -> Self { + pub(crate) fn new(reply_surbs_cfg: config::ReplySurbs) -> Self { Self { - min_surb_request_size, - max_surb_request_size, - maximum_allowed_reply_surb_request_size, - max_surb_rerequest_waiting_period, - max_surb_drop_waiting_period, - max_reply_surb_age, - max_reply_key_age, + reply_surbs: reply_surbs_cfg, } } } @@ -509,9 +492,17 @@ where } // 2. check whether the requested amount is within sane range - if amount > self.config.maximum_allowed_reply_surb_request_size { - warn!("The requested reply surb amount is larger than our maximum allowed ({amount} > {}). Lowering it to a more sane value...", self.config.maximum_allowed_reply_surb_request_size); - amount = self.config.maximum_allowed_reply_surb_request_size; + if amount + > self + .config + .reply_surbs + .maximum_allowed_reply_surb_request_size + { + warn!("The requested reply surb amount is larger than our maximum allowed ({amount} > {}). Lowering it to a more sane value...", self.config.reply_surbs.maximum_allowed_reply_surb_request_size); + amount = self + .config + .reply_surbs + .maximum_allowed_reply_surb_request_size; } // 3. construct and send the surbs away @@ -707,8 +698,11 @@ where } let request_size = min( - self.config.max_surb_request_size, - max(total_queue, self.config.min_surb_request_size), + self.config.reply_surbs.maximum_reply_surb_request_size, + max( + total_queue, + self.config.reply_surbs.minimum_reply_surb_request_size, + ), ); if let Err(err) = self @@ -744,9 +738,17 @@ where }; let diff = now - last_received_time; + let max_rerequest_wait = self + .config + .reply_surbs + .maximum_reply_surb_rerequest_waiting_period; + let max_drop_wait = self + .config + .reply_surbs + .maximum_reply_surb_drop_waiting_period; - if diff > self.config.max_surb_rerequest_waiting_period { - if diff > self.config.max_surb_drop_waiting_period { + if diff > max_rerequest_wait { + if diff > max_drop_wait { to_remove.push(*pending_reply_target) } else { debug!("We haven't received any surbs in {:?} from {pending_reply_target}. Going to explicitly ask for more", diff); @@ -793,7 +795,7 @@ where }; let diff = now - last_received_time; - if diff > self.config.max_reply_surb_age { + if diff > self.config.reply_surbs.maximum_reply_surb_age { info!("it's been {diff:?} since we last received any reply surb from {sender}. Going to remove all stored entries..."); to_remove_surbs.push(*sender); @@ -813,7 +815,7 @@ where let diff = now - sent_at; - if diff > self.config.max_reply_key_age { + if diff > self.config.reply_surbs.maximum_reply_key_age { debug!("it's been {diff:?} since we created this reply key. it's probably never going to get used, so we're going to purge it..."); to_remove_keys.push(*digest); } @@ -842,7 +844,8 @@ where let mut stale_inspection = new_interval_stream(polling_rate); // this is in the order of hours/days so we don't have to poll it that often - let polling_rate = Duration::from_secs(self.config.max_reply_surb_age.as_secs() / 10); + let polling_rate = + Duration::from_secs(self.config.reply_surbs.maximum_reply_surb_age.as_secs() / 10); let mut invalidation_inspection = new_interval_stream(polling_rate); while !shutdown.is_shutdown() { diff --git a/common/client-core/src/config/mod.rs b/common/client-core/src/config/mod.rs index ebcd604759..06490053c5 100644 --- a/common/client-core/src/config/mod.rs +++ b/common/client-core/src/config/mod.rs @@ -32,7 +32,7 @@ const DEFAULT_TOPOLOGY_RESOLUTION_TIMEOUT: Duration = Duration::from_millis(5_00 // bandwidth bridging protocol, we can come back to a smaller timeout value const DEFAULT_GATEWAY_RESPONSE_TIMEOUT: Duration = Duration::from_secs(5 * 60); -const DEFAULT_COVER_TRAFFIC_PRIMARY_SIZE_RATIO: f32 = 0.70; +const DEFAULT_COVER_TRAFFIC_PRIMARY_SIZE_RATIO: f64 = 0.70; // reply-surbs related: @@ -664,7 +664,7 @@ pub struct CoverTraffic { /// Specifies the ratio of `primary_packet_size` to `secondary_packet_size` used in cover traffic. /// Only applicable if `secondary_packet_size` is enabled. - pub cover_traffic_primary_size_ratio: f32, + pub cover_traffic_primary_size_ratio: f64, /// Controls whether the dedicated loop cover traffic stream should be enabled. /// (and sending packets, on average, every [Self::loop_cover_traffic_average_delay])