diff --git a/clients/client-core/src/client/real_messages_control/real_traffic_stream.rs b/clients/client-core/src/client/real_messages_control/real_traffic_stream.rs index cbf08e0975..2a72173846 100644 --- a/clients/client-core/src/client/real_messages_control/real_traffic_stream.rs +++ b/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> { 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 Stream for OutQueueControl diff --git a/clients/webassembly/js-example/worker.js b/clients/webassembly/js-example/worker.js index e35b24fe79..c1a3007802 100644 --- a/clients/webassembly/js-example/worker.js +++ b/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); diff --git a/clients/webassembly/src/client/config.rs b/clients/webassembly/src/client/config.rs index 131b198cf0..bd645e882e 100644 --- a/clients/webassembly/src/client/config.rs +++ b/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 for ConfigDebug { @@ -114,6 +118,8 @@ impl From 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 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, } } } diff --git a/clients/webassembly/src/client/mod.rs b/clients/webassembly/src/client/mod.rs index 40448305be..18c21d5439 100644 --- a/clients/webassembly/src/client/mod.rs +++ b/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(), );