Adjusting default settings

clients/client-core/src/client/real_messages_control/real_traffic_stream.rs

@@ -223,50 +223,6 @@ where
         tokio::task::yield_now().await;
     }
 
-    #[cfg(not(target_arch = "wasm32"))]
-    pub(super) async fn run_with_shutdown(&mut self, mut shutdown: task::ShutdownListener) {
-        debug!("Started OutQueueControl with graceful shutdown support");
-
-        // we should set initial delay only when we actually start the stream
-        let sampled =
-            sample_poisson_duration(&mut self.rng, self.config.average_message_sending_delay);
-        self.next_delay = Box::pin(time::sleep(sampled));
-
-        while !shutdown.is_shutdown() {
-            tokio::select! {
-                biased;
-                _ = shutdown.recv() => {
-                    log::trace!("OutQueueControl: Received shutdown");
-                }
-                next_message = self.next() => match next_message {
-                    Some(next_message) => {
-                        self.on_message(next_message).await;
-                    },
-                    None => {
-                        log::trace!("OutQueueControl: Stopping since channel closed");
-                        break;
-                    }
-                }
-            }
-        }
-        assert!(shutdown.is_shutdown_poll());
-        log::debug!("OutQueueControl: Exiting");
-    }
-
-    #[cfg(target_arch = "wasm32")]
-    pub(super) async fn run(&mut self) {
-        debug!("Started OutQueueControl without graceful shutdown support");
-
-        // we should set initial delay only when we actually start the stream
-        let sampled =
-            sample_poisson_duration(&mut self.rng, self.config.average_message_sending_delay);
-        self.next_delay = Box::pin(wasm_timer::Delay::new(sampled));
-
-        while let Some(next_message) = self.next().await {
-            self.on_message(next_message).await;
-        }
-    }
-
     fn poll_poisson(&mut self, cx: &mut Context<'_>) -> Poll<Option<StreamMessage>> {
         if let Some(ref mut next_delay) = &mut self.next_delay {
             // it is not yet time to return a message
@@ -277,13 +233,21 @@ where
             // we know it's time to send a message, so let's prepare delay for the next one
             // Get the `now` by looking at the current `delay` deadline
             let avg_delay = self.config.average_message_sending_delay;
-            let now = next_delay.deadline();
             let next_poisson_delay = sample_poisson_duration(&mut self.rng, avg_delay);
 
             // The next interval value is `next_poisson_delay` after the one that just
             // yielded.
-            let next = now + next_poisson_delay;
-            next_delay.as_mut().reset(next);
+            #[cfg(not(target_arch = "wasm32"))]
+            {
+                let now = next_delay.deadline();
+                let next = now + next_poisson_delay;
+                next_delay.as_mut().reset(next);
+            }
+
+            #[cfg(target_arch = "wasm32")]
+            {
+                next_delay.as_mut().reset(next_poisson_delay);
+            }
 
             // check if we have anything immediately available
             if let Some(real_available) = self.received_buffer.pop_front() {
@@ -312,10 +276,16 @@ where
             // we never set an initial delay - let's do it now
             cx.waker().wake_by_ref();
 
-            self.next_delay = Some(Box::pin(time::sleep(sample_poisson_duration(
-                &mut self.rng,
-                self.config.average_message_sending_delay,
-            ))));
+            let sampled =
+                sample_poisson_duration(&mut self.rng, self.config.average_message_sending_delay);
+
+            #[cfg(not(target_arch = "wasm32"))]
+            let next_delay = Box::pin(time::sleep(sampled));
+
+            #[cfg(target_arch = "wasm32")]
+            let next_delay = Box::pin(wasm_timer::Delay::new(sampled));
+
+            self.next_delay = Some(next_delay);
 
             Poll::Pending
         }
@@ -361,6 +331,40 @@ where
             self.poll_poisson(cx)
         }
     }
+
+    #[cfg(not(target_arch = "wasm32"))]
+    pub(super) async fn run_with_shutdown(&mut self, mut shutdown: task::ShutdownListener) {
+        debug!("Started OutQueueControl with graceful shutdown support");
+
+        while !shutdown.is_shutdown() {
+            tokio::select! {
+                biased;
+                _ = shutdown.recv() => {
+                    log::trace!("OutQueueControl: Received shutdown");
+                }
+                next_message = self.next() => match next_message {
+                    Some(next_message) => {
+                        self.on_message(next_message).await;
+                    },
+                    None => {
+                        log::trace!("OutQueueControl: Stopping since channel closed");
+                        break;
+                    }
+                }
+            }
+        }
+        assert!(shutdown.is_shutdown_poll());
+        log::debug!("OutQueueControl: Exiting");
+    }
+
+    #[cfg(target_arch = "wasm32")]
+    pub(super) async fn run(&mut self) {
+        debug!("Started OutQueueControl without graceful shutdown support");
+
+        while let Some(next_message) = self.next().await {
+            self.on_message(next_message).await;
+        }
+    }
 }
 
 impl<R> Stream for OutQueueControl<R>

clients/webassembly/js-example/worker.js

@@ -70,15 +70,14 @@ async function main() {
   // only really useful if you want to adjust some settings like traffic rate
   // (if not needed you can just pass a null)
   const debug = default_debug();
-  // note: we still have poisson distribution so, on average, we will be sending SOME packet every 20ms
-  // debug.disable_main_poisson_packet_distribution = true;
-  // debug.disable_loop_cover_traffic_stream = true;
+
+  debug.disable_main_poisson_packet_distribution = true;
+  debug.disable_loop_cover_traffic_stream = true;
   // debug.average_packet_delay_ms = BigInt(10);
   // debug.average_ack_delay_ms = BigInt(10);
   // debug.ack_wait_addition_ms = BigInt(3000);
   // debug.ack_wait_multiplier = 10;
 
-  // this is currently disabled, i.e. we'll keep using the topology we get at startup
   debug.topology_refresh_rate_ms = BigInt(60000)
 
   const config = new Config('my-awesome-wasm-client', validator, gatewayEndpoint, debug);

clients/webassembly/src/client/config.rs

@@ -93,6 +93,10 @@ pub struct Debug {
     /// Controls whether the dedicated loop cover traffic stream should be enabled.
     /// (and sending packets, on average, every [Self::loop_cover_traffic_average_delay_ms])
     pub disable_loop_cover_traffic_stream: bool,
+
+    /// Controls whether the main packet stream constantly produces packets according to the predefined
+    /// poisson distribution.
+    pub disable_main_poisson_packet_distribution: bool,
 }
 
 impl From<Debug> for ConfigDebug {
@@ -114,6 +118,8 @@ impl From<Debug> for ConfigDebug {
                 debug.topology_resolution_timeout_ms,
             ),
             disable_loop_cover_traffic_stream: debug.disable_loop_cover_traffic_stream,
+            disable_main_poisson_packet_distribution: debug
+                .disable_main_poisson_packet_distribution,
         }
     }
 }
@@ -133,6 +139,8 @@ impl From<ConfigDebug> for Debug {
             topology_refresh_rate_ms: debug.topology_refresh_rate.as_millis() as u64,
             topology_resolution_timeout_ms: debug.topology_resolution_timeout.as_millis() as u64,
             disable_loop_cover_traffic_stream: debug.disable_loop_cover_traffic_stream,
+            disable_main_poisson_packet_distribution: debug
+                .disable_main_poisson_packet_distribution,
         }
     }
 }

clients/webassembly/src/client/mod.rs

@@ -125,6 +125,7 @@ impl NymClient {
             self.config.debug.average_ack_delay,
             self.config.debug.message_sending_average_delay,
             self.config.debug.average_packet_delay,
+            self.config.debug.disable_main_poisson_packet_distribution,
             self.as_mix_recipient(),
         );