weighted decision on cover traffic packet size ('real traffic stream')

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,
         }

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<AckKey>,
         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<R>
@@ -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<Option<StreamMessage>> {
-        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!(

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() {

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])