From 31056a90cb4ed64ea04e6ab90cc3b8416ca83b3c Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?J=C4=99drzej=20Stuczy=C5=84ski?= Date: Wed, 5 Oct 2022 09:49:23 +0100 Subject: [PATCH] WIP cleanup before machine switch --- Cargo.lock | 2 +- clients/client-core/Cargo.toml | 5 +- .../acknowledgement_listener.rs | 67 +++++---- .../action_controller.rs | 75 +++++----- .../input_message_listener.rs | 53 ++++--- .../acknowledgement_control/mod.rs | 136 ++++++++++-------- .../retransmission_request_listener.rs | 50 +++---- .../sent_notification_listener.rs | 54 +++---- .../src/client/real_messages_control/mod.rs | 49 +++---- .../real_traffic_stream.rs | 97 +++++++------ .../src/client/topology_control.rs | 39 ++--- clients/client-core/src/lib.rs | 2 - clients/webassembly/js-example/index.js | 5 + clients/webassembly/src/client/mod.rs | 5 +- clients/webassembly/src/lib.rs | 1 + 15 files changed, 332 insertions(+), 308 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 5e1f4813af..47a46dcfc6 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -651,7 +651,6 @@ dependencies = [ "config", "crypto", "dirs", - "fluvio-wasm-timer", "futures", "gateway-client", "gateway-requests", @@ -673,6 +672,7 @@ dependencies = [ "validator-client", "wasm-bindgen", "wasm-bindgen-futures", + "wasm-timer", ] [[package]] diff --git a/clients/client-core/Cargo.toml b/clients/client-core/Cargo.toml index 68a132193c..3226252751 100644 --- a/clients/client-core/Cargo.toml +++ b/clients/client-core/Cargo.toml @@ -37,8 +37,9 @@ version = "0.4" [target."cfg(target_arch = \"wasm32\")".dependencies.wasm-bindgen] version = "=0.2.78" -[target."cfg(target_arch = \"wasm32\")".dependencies.fluvio-wasm-timer] -version = "0.2.5" +[target."cfg(target_arch = \"wasm32\")".dependencies.wasm-timer] +git = "https://github.com/jstuczyn/wasm-timer" +rev = "2859241251dd00c54a9c99b74727b99bd7feffb4" [target."cfg(not(target_arch = \"wasm32\"))".dependencies.task] path = "../../common/task" diff --git a/clients/client-core/src/client/real_messages_control/acknowledgement_control/acknowledgement_listener.rs b/clients/client-core/src/client/real_messages_control/acknowledgement_control/acknowledgement_listener.rs index ff4a25e864..e62c888707 100644 --- a/clients/client-core/src/client/real_messages_control/acknowledgement_control/acknowledgement_listener.rs +++ b/clients/client-core/src/client/real_messages_control/acknowledgement_control/acknowledgement_listener.rs @@ -20,8 +20,6 @@ pub(super) struct AcknowledgementListener { ack_key: Arc, ack_receiver: AcknowledgementReceiver, action_sender: ActionSender, - #[cfg(not(target_arch = "wasm32"))] - shutdown: ShutdownListener, } impl AcknowledgementListener { @@ -29,14 +27,11 @@ impl AcknowledgementListener { ack_key: Arc, ack_receiver: AcknowledgementReceiver, action_sender: ActionSender, - #[cfg(not(target_arch = "wasm32"))] shutdown: ShutdownListener, ) -> Self { AcknowledgementListener { ack_key, ack_receiver, action_sender, - #[cfg(not(target_arch = "wasm32"))] - shutdown, } } @@ -71,39 +66,41 @@ impl AcknowledgementListener { .unwrap(); } - pub(super) async fn run(&mut self) { - debug!("Started AcknowledgementListener"); + async fn handle_ack_receiver_item(&mut self, item: Vec>) { + // realistically we would only be getting one ack at the time + for ack in item { + self.on_ack(ack).await; + } + } - // todo: make it non-wasm compatible - debug!("Started AcknowledgementListener"); - while let Some(acks) = self.ack_receiver.next().await { - // realistically we would only be getting one ack at the time - for ack in acks { - self.on_ack(ack).await; + #[cfg(not(target_arch = "wasm32"))] + pub(super) async fn run_with_shutdown(&mut self, mut shutdown: task::ShutdownListener) { + debug!("Started AcknowledgementListener with graceful shutdown support"); + + while !shutdown.is_shutdown() { + tokio::select! { + acks = self.ack_receiver.next() => match acks { + Some(acks) => self.handle_ack_receiver_item(acks).await, + None => { + log::trace!("AcknowledgementListener: Stopping since channel closed"); + break; + } + }, + _ = shutdown.recv() => { + log::trace!("AcknowledgementListener: Received shutdown"); + } } } - error!("TODO: error msg. Or maybe panic?") + assert!(shutdown.is_shutdown_poll()); + log::debug!("AcknowledgementListener: Exiting"); + } - // while !self.shutdown.is_shutdown() { - // tokio::select! { - // acks = self.ack_receiver.next() => match acks { - // Some(acks) => { - // // realistically we would only be getting one ack at the time - // for ack in acks { - // self.on_ack(ack).await; - // } - // }, - // None => { - // log::trace!("AcknowledgementListener: Stopping since channel closed"); - // break; - // } - // }, - // _ = self.shutdown.recv() => { - // log::trace!("AcknowledgementListener: Received shutdown"); - // } - // } - // } - // assert!(self.shutdown.is_shutdown_poll()); - // log::debug!("AcknowledgementListener: Exiting"); + #[cfg(target_arch = "wasm32")] + pub(super) async fn run(&mut self) { + debug!("Started AcknowledgementListener without graceful shutdown support"); + + while let Some(acks) = self.ack_receiver.next().await { + self.handle_ack_receiver_item(acks).await + } } } diff --git a/clients/client-core/src/client/real_messages_control/acknowledgement_control/action_controller.rs b/clients/client-core/src/client/real_messages_control/acknowledgement_control/action_controller.rs index 94a4582fd5..59d886bcaf 100644 --- a/clients/client-core/src/client/real_messages_control/acknowledgement_control/action_controller.rs +++ b/clients/client-core/src/client/real_messages_control/acknowledgement_control/action_controller.rs @@ -102,17 +102,12 @@ pub(super) struct ActionController { /// Channel for notifying `RetransmissionRequestListener` about expired acknowledgements. retransmission_sender: RetransmissionRequestSender, - - /// Listen for shutdown notifications - #[cfg(not(target_arch = "wasm32"))] - shutdown: ShutdownListener, } impl ActionController { pub(super) fn new( config: Config, retransmission_sender: RetransmissionRequestSender, - #[cfg(not(target_arch = "wasm32"))] shutdown: ShutdownListener, ) -> (Self, ActionSender) { let (sender, receiver) = mpsc::unbounded(); ( @@ -122,8 +117,6 @@ impl ActionController { pending_acks_timers: NonExhaustiveDelayQueue::new(), incoming_actions: receiver, retransmission_sender, - #[cfg(not(target_arch = "wasm32"))] - shutdown, }, sender, ) @@ -255,42 +248,46 @@ impl ActionController { } } - pub(super) async fn run(&mut self) { - // TODO: shutdown without wasm etc etc - loop { - // at some point there will be a global shutdown signal here as the third option + #[cfg(not(target_arch = "wasm32"))] + pub(super) async fn run_with_shutdown(&mut self, mut shutdown: task::ShutdownListener) { + debug!("Started ActionController with graceful shutdown support"); + + while !shutdown.is_shutdown() { + tokio::select! { + action = self.incoming_actions.next() => match action { + Some(action) => self.process_action(action), + None => { + log::trace!( + "ActionController: Stopping since incoming actions channel closed" + ); + break; + } + }, + expired_ack = self.pending_acks_timers.next() => match expired_ack { + Some(expired_ack) => self.handle_expired_ack_timer(expired_ack), + None => { + log::trace!("ActionController: Stopping since ack channel closed"); + break; + } + }, + _ = shutdown.recv() => { + log::trace!("ActionController: Received shutdown"); + } + } + } + assert!(shutdown.is_shutdown_poll()); + log::debug!("ActionController: Exiting"); + } + + #[cfg(target_arch = "wasm32")] + pub(super) async fn run(&mut self) { + debug!("Started ActionController without graceful shutdown support"); + + loop { tokio::select! { - // we NEVER expect for ANY sender to get dropped so unwrap here is fine action = self.incoming_actions.next() => self.process_action(action.unwrap()), - // pending ack queue Stream CANNOT return a `None` so unwrap here is fine expired_ack = self.pending_acks_timers.next() => self.handle_expired_ack_timer(expired_ack.unwrap()) } } - - // while !self.shutdown.is_shutdown() { - // tokio::select! { - // action = self.incoming_actions.next() => match action { - // Some(action) => self.process_action(action), - // None => { - // log::trace!( - // "ActionController: Stopping since incoming actions channel closed" - // ); - // break; - // } - // }, - // expired_ack = self.pending_acks_timers.next() => match expired_ack { - // Some(expired_ack) => self.handle_expired_ack_timer(expired_ack), - // None => { - // log::trace!("ActionController: Stopping since ack channel closed"); - // break; - // } - // }, - // _ = self.shutdown.recv() => { - // log::trace!("ActionController: Received shutdown"); - // } - // } - // } - // assert!(self.shutdown.is_shutdown_poll()); - // log::debug!("ActionController: Exiting"); } } diff --git a/clients/client-core/src/client/real_messages_control/acknowledgement_control/input_message_listener.rs b/clients/client-core/src/client/real_messages_control/acknowledgement_control/input_message_listener.rs index a33783cc41..2f4ec93bd3 100644 --- a/clients/client-core/src/client/real_messages_control/acknowledgement_control/input_message_listener.rs +++ b/clients/client-core/src/client/real_messages_control/acknowledgement_control/input_message_listener.rs @@ -38,8 +38,6 @@ where topology_access: TopologyAccessor, #[cfg(feature = "reply-surb")] reply_key_storage: ReplyKeyStorage, - #[cfg(not(target_arch = "wasm32"))] - shutdown: ShutdownListener, } impl InputMessageListener @@ -58,7 +56,6 @@ where real_message_sender: BatchRealMessageSender, topology_access: TopologyAccessor, #[cfg(feature = "reply-surb")] reply_key_storage: ReplyKeyStorage, - #[cfg(not(target_arch = "wasm32"))] shutdown: ShutdownListener, ) -> Self { InputMessageListener { ack_key, @@ -70,8 +67,6 @@ where topology_access, #[cfg(feature = "reply-surb")] reply_key_storage, - #[cfg(not(target_arch = "wasm32"))] - shutdown, } } @@ -193,31 +188,35 @@ where } } + #[cfg(not(target_arch = "wasm32"))] + pub(super) async fn run_with_shutdown(&mut self, mut shutdown: task::ShutdownListener) { + debug!("Started InputMessageListener with graceful shutdown support"); + + while !shutdown.is_shutdown() { + tokio::select! { + input_msg = self.input_receiver.next() => match input_msg { + Some(input_msg) => { + self.on_input_message(input_msg).await; + }, + None => { + log::trace!("InputMessageListener: Stopping since channel closed"); + break; + } + }, + _ = shutdown.recv() => { + log::trace!("InputMessageListener: Received shutdown"); + } + } + } + assert!(shutdown.is_shutdown_poll()); + log::debug!("InputMessageListener: Exiting"); + } + + #[cfg(target_arch = "wasm32")] pub(super) async fn run(&mut self) { - // TODO: shutdown without wasm etc etc - debug!("Started InputMessageListener"); + debug!("Started InputMessageListener without graceful shutdown support"); while let Some(input_msg) = self.input_receiver.next().await { self.on_input_message(input_msg).await; } - - // debug!("Started InputMessageListener"); - // while !self.shutdown.is_shutdown() { - // tokio::select! { - // input_msg = self.input_receiver.next() => match input_msg { - // Some(input_msg) => { - // self.on_input_message(input_msg).await; - // }, - // None => { - // log::trace!("InputMessageListener: Stopping since channel closed"); - // break; - // } - // }, - // _ = self.shutdown.recv() => { - // log::trace!("InputMessageListener: Received shutdown"); - // } - // } - // } - // assert!(self.shutdown.is_shutdown_poll()); - // log::debug!("InputMessageListener: Exiting"); } } diff --git a/clients/client-core/src/client/real_messages_control/acknowledgement_control/mod.rs b/clients/client-core/src/client/real_messages_control/acknowledgement_control/mod.rs index ac77009cb3..38766af3a0 100644 --- a/clients/client-core/src/client/real_messages_control/acknowledgement_control/mod.rs +++ b/clients/client-core/src/client/real_messages_control/acknowledgement_control/mod.rs @@ -146,11 +146,11 @@ pub(super) struct AcknowledgementController where R: CryptoRng + Rng, { - acknowledgement_listener: Option, - input_message_listener: Option>, - retransmission_request_listener: Option>, - sent_notification_listener: Option, - action_controller: Option, + acknowledgement_listener: AcknowledgementListener, + input_message_listener: InputMessageListener, + retransmission_request_listener: RetransmissionRequestListener, + sent_notification_listener: SentNotificationListener, + action_controller: ActionController, } impl AcknowledgementController @@ -166,18 +166,13 @@ where ack_recipient: Recipient, connectors: AcknowledgementControllerConnectors, #[cfg(feature = "reply-surb")] reply_key_storage: ReplyKeyStorage, - #[cfg(not(target_arch = "wasm32"))] shutdown: ShutdownListener, ) -> Self { let (retransmission_tx, retransmission_rx) = mpsc::unbounded(); let action_config = action_controller::Config::new(config.ack_wait_addition, config.ack_wait_multiplier); - let (action_controller, action_sender) = ActionController::new( - action_config, - retransmission_tx, - #[cfg(not(target_arch = "wasm32"))] - shutdown.clone(), - ); + let (action_controller, action_sender) = + ActionController::new(action_config, retransmission_tx); let message_preparer = MessagePreparer::new( rng, @@ -191,8 +186,6 @@ where Arc::clone(&ack_key), connectors.ack_receiver, action_sender.clone(), - #[cfg(not(target_arch = "wasm32"))] - shutdown.clone(), ); // will listen for any new messages from the client @@ -206,8 +199,6 @@ where topology_access.clone(), #[cfg(feature = "reply-surb")] reply_key_storage, - #[cfg(not(target_arch = "wasm32"))] - shutdown.clone(), ); // will listen for any ack timeouts and trigger retransmission @@ -219,72 +210,95 @@ where connectors.real_message_sender, retransmission_rx, topology_access, - #[cfg(not(target_arch = "wasm32"))] - shutdown.clone(), ); // will listen for events indicating the packet was sent through the network so that // the retransmission timer should be started. - let sent_notification_listener = SentNotificationListener::new( - connectors.sent_notifier, - action_sender, - #[cfg(not(target_arch = "wasm32"))] - shutdown, - ); + let sent_notification_listener = + SentNotificationListener::new(connectors.sent_notifier, action_sender); AcknowledgementController { - acknowledgement_listener: Some(acknowledgement_listener), - input_message_listener: Some(input_message_listener), - retransmission_request_listener: Some(retransmission_request_listener), - sent_notification_listener: Some(sent_notification_listener), - action_controller: Some(action_controller), + acknowledgement_listener, + input_message_listener, + retransmission_request_listener, + sent_notification_listener, + action_controller, } } - pub(super) async fn run(&mut self) { - let mut acknowledgement_listener = self.acknowledgement_listener.take().unwrap(); - let mut input_message_listener = self.input_message_listener.take().unwrap(); - let mut retransmission_request_listener = - self.retransmission_request_listener.take().unwrap(); - let mut sent_notification_listener = self.sent_notification_listener.take().unwrap(); - let mut action_controller = self.action_controller.take().unwrap(); + #[cfg(not(target_arch = "wasm32"))] + pub(super) fn start_with_shutdown(self, shutdown: task::ShutdownListener) { + let mut acknowledgement_listener = self.acknowledgement_listener; + let mut input_message_listener = self.input_message_listener; + let mut retransmission_request_listener = self.retransmission_request_listener; + let mut sent_notification_listener = self.sent_notification_listener; + let mut action_controller = self.action_controller; - // the below are log messages are errors as at the current stage we do not expect any of - // the task to ever finish. This will of course change once we introduce - // graceful shutdowns. - let ack_listener_fut = spawn_future(async move { - acknowledgement_listener.run().await; + let shutdown_handle = shutdown.clone(); + spawn_future(async move { + acknowledgement_listener + .run_with_shutdown(shutdown_handle) + .await; debug!("The acknowledgement listener has finished execution!"); }); - let input_listener_fut = spawn_future(async move { - input_message_listener.run().await; + + let shutdown_handle = shutdown.clone(); + spawn_future(async move { + input_message_listener + .run_with_shutdown(shutdown_handle) + .await; debug!("The input listener has finished execution!"); }); - let retransmission_req_fut = spawn_future(async move { - retransmission_request_listener.run().await; + + let shutdown_handle = shutdown.clone(); + spawn_future(async move { + retransmission_request_listener + .run_with_shutdown(shutdown_handle) + .await; debug!("The retransmission request listener has finished execution!"); }); - let sent_notification_fut = spawn_future(async move { - sent_notification_listener.run().await; + + let shutdown_handle = shutdown.clone(); + spawn_future(async move { + sent_notification_listener + .run_with_shutdown(shutdown_handle) + .await; debug!("The sent notification listener has finished execution!"); }); - let action_controller_fut = spawn_future(async move { - action_controller.run().await; + + spawn_future(async move { + action_controller.run_with_shutdown(shutdown).await; debug!("The controller has finished execution!"); }); - - // // technically we don't have to bring `AcknowledgementController` back to a valid state - // // but we can do it, so why not? Perhaps it might be useful if we wanted to allow - // // for restarts of certain modules without killing the entire process. - // self.acknowledgement_listener = Some(ack_listener_fut.await.unwrap()); - // self.input_message_listener = Some(input_listener_fut.await.unwrap()); - // self.retransmission_request_listener = Some(retransmission_req_fut.await.unwrap()); - // self.sent_notification_listener = Some(sent_notification_fut.await.unwrap()); - // self.action_controller = Some(action_controller_fut.await.unwrap()); } - #[allow(dead_code)] - pub(super) fn start(mut self) { - spawn_future(async move { self.run().await }) + #[cfg(target_arch = "wasm32")] + pub(super) fn start(self) { + let mut acknowledgement_listener = self.acknowledgement_listener; + let mut input_message_listener = self.input_message_listener; + let mut retransmission_request_listener = self.retransmission_request_listener; + let mut sent_notification_listener = self.sent_notification_listener; + let mut action_controller = self.action_controller; + + spawn_future(async move { + acknowledgement_listener.run().await; + error!("The acknowledgement listener has finished execution!"); + }); + spawn_future(async move { + input_message_listener.run().await; + error!("The input listener has finished execution!"); + }); + spawn_future(async move { + retransmission_request_listener.run().await; + error!("The retransmission request listener has finished execution!"); + }); + spawn_future(async move { + sent_notification_listener.run().await; + error!("The sent notification listener has finished execution!"); + }); + spawn_future(async move { + action_controller.run().await; + error!("The controller has finished execution!"); + }); } } diff --git a/clients/client-core/src/client/real_messages_control/acknowledgement_control/retransmission_request_listener.rs b/clients/client-core/src/client/real_messages_control/acknowledgement_control/retransmission_request_listener.rs index e42967d06f..371ba3d304 100644 --- a/clients/client-core/src/client/real_messages_control/acknowledgement_control/retransmission_request_listener.rs +++ b/clients/client-core/src/client/real_messages_control/acknowledgement_control/retransmission_request_listener.rs @@ -30,8 +30,6 @@ where real_message_sender: BatchRealMessageSender, request_receiver: RetransmissionRequestReceiver, topology_access: TopologyAccessor, - #[cfg(not(target_arch = "wasm32"))] - shutdown: ShutdownListener, } impl RetransmissionRequestListener @@ -47,7 +45,6 @@ where real_message_sender: BatchRealMessageSender, request_receiver: RetransmissionRequestReceiver, topology_access: TopologyAccessor, - #[cfg(not(target_arch = "wasm32"))] shutdown: ShutdownListener, ) -> Self { RetransmissionRequestListener { ack_key, @@ -57,8 +54,6 @@ where real_message_sender, request_receiver, topology_access, - #[cfg(not(target_arch = "wasm32"))] - shutdown, } } @@ -128,29 +123,34 @@ where .unwrap(); } - pub(super) async fn run(&mut self) { - debug!("Started RetransmissionRequestListener"); + #[cfg(not(target_arch = "wasm32"))] + pub(super) async fn run_with_shutdown(&mut self, mut shutdown: task::ShutdownListener) { + debug!("Started RetransmissionRequestListener with graceful shutdown support"); + + while !shutdown.is_shutdown() { + tokio::select! { + timed_out_ack = self.request_receiver.next() => match timed_out_ack { + Some(timed_out_ack) => self.on_retransmission_request(timed_out_ack).await, + None => { + log::trace!("RetransmissionRequestListener: Stopping since channel closed"); + break; + } + }, + _ = shutdown.recv() => { + log::trace!("RetransmissionRequestListener: Received shutdown"); + } + } + } + assert!(shutdown.is_shutdown_poll()); + log::debug!("RetransmissionRequestListener: Exiting"); + } + + #[cfg(target_arch = "wasm32")] + pub(super) async fn run(&mut self) { + debug!("Started RetransmissionRequestListener without graceful shutdown support"); - // TODO: shutdown without wasm etc etc while let Some(timed_out_ack) = self.request_receiver.next().await { self.on_retransmission_request(timed_out_ack).await; } - - // while !self.shutdown.is_shutdown() { - // tokio::select! { - // timed_out_ack = self.request_receiver.next() => match timed_out_ack { - // Some(timed_out_ack) => self.on_retransmission_request(timed_out_ack).await, - // None => { - // log::trace!("RetransmissionRequestListener: Stopping since channel closed"); - // break; - // } - // }, - // _ = self.shutdown.recv() => { - // log::trace!("RetransmissionRequestListener: Received shutdown"); - // } - // } - // } - // assert!(self.shutdown.is_shutdown_poll()); - // log::debug!("RetransmissionRequestListener: Exiting"); } } diff --git a/clients/client-core/src/client/real_messages_control/acknowledgement_control/sent_notification_listener.rs b/clients/client-core/src/client/real_messages_control/acknowledgement_control/sent_notification_listener.rs index c72769941f..ce516dc30d 100644 --- a/clients/client-core/src/client/real_messages_control/acknowledgement_control/sent_notification_listener.rs +++ b/clients/client-core/src/client/real_messages_control/acknowledgement_control/sent_notification_listener.rs @@ -17,21 +17,16 @@ use task::ShutdownListener; pub(super) struct SentNotificationListener { sent_notifier: SentPacketNotificationReceiver, action_sender: ActionSender, - #[cfg(not(target_arch = "wasm32"))] - shutdown: ShutdownListener, } impl SentNotificationListener { pub(super) fn new( sent_notifier: SentPacketNotificationReceiver, action_sender: ActionSender, - #[cfg(not(target_arch = "wasm32"))] shutdown: ShutdownListener, ) -> Self { SentNotificationListener { sent_notifier, action_sender, - #[cfg(not(target_arch = "wasm32"))] - shutdown, } } @@ -50,31 +45,36 @@ impl SentNotificationListener { .unwrap(); } - pub(super) async fn run(&mut self) { - debug!("Started SentNotificationListener"); + #[cfg(not(target_arch = "wasm32"))] + pub(super) async fn run_with_shutdown(&mut self, mut shutdown: task::ShutdownListener) { + debug!("Started SentNotificationListener with graceful shutdown support"); + + while !shutdown.is_shutdown() { + tokio::select! { + frag_id = self.sent_notifier.next() => match frag_id { + Some(frag_id) => { + self.on_sent_message(frag_id).await; + } + None => { + log::trace!("SentNotificationListener: Stopping since channel closed"); + break; + } + }, + _ = shutdown.recv() => { + log::trace!("SentNotificationListener: Received shutdown"); + } + } + } + assert!(shutdown.is_shutdown_poll()); + log::debug!("SentNotificationListener: Exiting"); + } + + #[cfg(target_arch = "wasm32")] + pub(super) async fn run(&mut self) { + debug!("Started SentNotificationListener without graceful shutdown support"); - // TODO: shutdown without wasm etc etc while let Some(frag_id) = self.sent_notifier.next().await { self.on_sent_message(frag_id).await; } - - // while !self.shutdown.is_shutdown() { - // tokio::select! { - // frag_id = self.sent_notifier.next() => match frag_id { - // Some(frag_id) => { - // self.on_sent_message(frag_id).await; - // } - // None => { - // log::trace!("SentNotificationListener: Stopping since channel closed"); - // break; - // } - // }, - // _ = self.shutdown.recv() => { - // log::trace!("SentNotificationListener: Received shutdown"); - // } - // } - // } - // assert!(self.shutdown.is_shutdown_poll()); - // log::debug!("SentNotificationListener: Exiting"); } } diff --git a/clients/client-core/src/client/real_messages_control/mod.rs b/clients/client-core/src/client/real_messages_control/mod.rs index decb40dc54..add17db17e 100644 --- a/clients/client-core/src/client/real_messages_control/mod.rs +++ b/clients/client-core/src/client/real_messages_control/mod.rs @@ -82,8 +82,8 @@ pub struct RealMessagesController where R: CryptoRng + Rng, { - out_queue_control: Option>, - ack_control: Option>, + out_queue_control: OutQueueControl, + ack_control: AcknowledgementController, } // obviously when we finally make shared rng that is on 'higher' level, this should become @@ -96,7 +96,6 @@ impl RealMessagesController { mix_sender: BatchMixMessageSender, topology_access: TopologyAccessor, #[cfg(feature = "reply-surb")] reply_key_storage: ReplyKeyStorage, - #[cfg(not(target_arch = "wasm32"))] shutdown: ShutdownListener, ) -> Self { let rng = OsRng; @@ -126,8 +125,6 @@ impl RealMessagesController { ack_controller_connectors, #[cfg(feature = "reply-surb")] reply_key_storage, - #[cfg(not(target_arch = "wasm32"))] - shutdown.clone(), ); let out_queue_config = real_traffic_stream::Config::new( @@ -145,40 +142,36 @@ impl RealMessagesController { rng, config.self_recipient, topology_access, - #[cfg(not(target_arch = "wasm32"))] - shutdown, ); RealMessagesController { - out_queue_control: Some(out_queue_control), - ack_control: Some(ack_control), + out_queue_control, + ack_control, } } - pub(super) async fn run(&mut self) { - let mut out_queue_control = self.out_queue_control.take().unwrap(); - let mut ack_control = self.ack_control.take().unwrap(); + #[cfg(not(target_arch = "wasm32"))] + pub fn start_with_shutdown(self, shutdown: task::ShutdownListener) { + let mut out_queue_control = self.out_queue_control; + let mut ack_control = self.ack_control; - // the below are log messages are errors as at the current stage we do not expect any of - // the task to ever finish. This will of course change once we introduce - // graceful shutdowns. - let out_queue_control_fut = spawn_future(async move { - out_queue_control.run_out_queue_control().await; + let shutdown_handle = shutdown.clone(); + spawn_future(async move { + out_queue_control.run_with_shutdown(shutdown_handle).await; debug!("The out queue controller has finished execution!"); }); - let ack_control_fut = spawn_future(async move { - ack_control.run().await; - debug!("The acknowledgement controller has finished execution!"); - }); - - // // technically we don't have to bring `RealMessagesController` back to a valid state - // // but we can do it, so why not? Perhaps it might be useful if we wanted to allow - // // for restarts of certain modules without killing the entire process. - // self.out_queue_control = Some(out_queue_control_fut.await.unwrap()); - // self.ack_control = Some(ack_control_fut.await.unwrap()); + ack_control.start_with_shutdown(shutdown); } + #[cfg(target_arch = "wasm32")] pub fn start(mut self) { - spawn_future(async move { self.run().await }) + let mut out_queue_control = self.out_queue_control; + let mut ack_control = self.ack_control; + + spawn_future(async move { + out_queue_control.run().await; + debug!("The out queue controller has finished execution!"); + }); + ack_control.start(); } } 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 01e7ddf3ac..deb4ccaad3 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 @@ -24,7 +24,7 @@ use std::time::Duration; use tokio::time; #[cfg(target_arch = "wasm32")] -use fluvio_wasm_timer as wasm_timer; +use wasm_timer; #[cfg(not(target_arch = "wasm32"))] use task::ShutdownListener; @@ -95,10 +95,6 @@ where /// Buffer containing all real messages received. It is first exhausted before more are pulled. received_buffer: VecDeque, - - /// Listens for shutdown signals - #[cfg(not(target_arch = "wasm32"))] - shutdown: ShutdownListener, } pub(crate) struct RealMessage { @@ -199,7 +195,6 @@ where rng: R, our_full_destination: Recipient, topology_access: TopologyAccessor, - #[cfg(not(target_arch = "wasm32"))] shutdown: ShutdownListener, ) -> Self { #[cfg(not(target_arch = "wasm32"))] let next_delay = Box::pin(time::sleep(Default::default())); @@ -218,8 +213,6 @@ where rng, topology_access, received_buffer: VecDeque::with_capacity(0), // we won't be putting any data into this guy directly - #[cfg(not(target_arch = "wasm32"))] - shutdown, } } @@ -295,49 +288,69 @@ where tokio::task::yield_now().await; } - // Send messages at certain rate and if no real traffic is available, send cover message. - async fn run_normal_out_queue(&mut self) { + // // Send messages at certain rate and if no real traffic is available, send cover message. + // async fn run_normal_out_queue(&mut self) { + // // 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); + // + // #[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 = next_delay; + // + // // TODO: fix it for non-wasm + // } + + // pub(crate) async fn run_out_queue_control(&mut self) { + // debug!("Starting out queue controller..."); + // self.run_normal_out_queue().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)); - #[cfg(not(target_arch = "wasm32"))] - let 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")] - let next_delay = Box::pin(wasm_timer::Delay::new(sampled)); + #[cfg(target_arch = "wasm32")] + pub(super) async fn run(&mut self) { + debug!("Started OutQueueControl without graceful shutdown support"); - self.next_delay = next_delay; + // 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)); - // TODO: fix it for non-wasm while let Some(next_message) = self.next().await { self.on_message(next_message).await; } - - // let mut shutdown = self.shutdown.clone(); - // 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"); - } - - pub(crate) async fn run_out_queue_control(&mut self) { - debug!("Starting out queue controller..."); - self.run_normal_out_queue().await } } diff --git a/clients/client-core/src/client/topology_control.rs b/clients/client-core/src/client/topology_control.rs index abad227b0a..7bebc34ea9 100644 --- a/clients/client-core/src/client/topology_control.rs +++ b/clients/client-core/src/client/topology_control.rs @@ -306,28 +306,33 @@ impl TopologyRefresher { self.topology_accessor.is_routable().await } - pub fn start(mut self, #[cfg(not(target_arch = "wasm32"))] mut shutdown: ShutdownListener) { - // TODO: usual non-wasm, etc, etc + #[cfg(not(target_arch = "wasm32"))] + pub async fn start_with_shutdown(mut self, mut shutdown: task::ShutdownListener) { + debug!("Started TopologyRefresher with graceful shutdown support"); + + while !shutdown.is_shutdown() { + tokio::select! { + _ = tokio::time::sleep(self.refresh_rate) => { + self.refresh().await; + }, + _ = shutdown.recv() => { + log::trace!("TopologyRefresher: Received shutdown"); + }, + } + } + assert!(shutdown.is_shutdown_poll()); + log::debug!("TopologyRefresher: Exiting"); + } + + #[cfg(target_arch = "wasm32")] + pub fn start(mut self) { + debug!("Started TopologyRefresher without graceful shutdown support"); spawn_future(async move { loop { - // TODO: this won't work in wasm - tokio::time::sleep(self.refresh_rate).await; + wasm_timer::Delay::new(self.refresh_rate).await; self.refresh().await; } - - // while !shutdown.is_shutdown() { - // tokio::select! { - // _ = tokio::time::sleep(self.refresh_rate) => { - // self.refresh().await; - // }, - // _ = shutdown.recv() => { - // log::trace!("TopologyRefresher: Received shutdown"); - // }, - // } - // } - // assert!(shutdown.is_shutdown_poll()); - // log::debug!("TopologyRefresher: Exiting"); }) } } diff --git a/clients/client-core/src/lib.rs b/clients/client-core/src/lib.rs index 9d23410a50..04e215daec 100644 --- a/clients/client-core/src/lib.rs +++ b/clients/client-core/src/lib.rs @@ -5,8 +5,6 @@ pub mod config; pub mod error; pub mod init; -// TODO: move those to separate lower modules and conditionally re-export them accordingly to make intellij happier about name clash - #[cfg(target_arch = "wasm32")] pub(crate) fn spawn_future(future: F) where diff --git a/clients/webassembly/js-example/index.js b/clients/webassembly/js-example/index.js index 936df556df..b0b4916ae5 100644 --- a/clients/webassembly/js-example/index.js +++ b/clients/webassembly/js-example/index.js @@ -61,6 +61,11 @@ 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(); + debug.loop_cover_traffic_average_delay_ms = BigInt(60_000); + // note: we still have poisson distribution so, on average, we will be sending SOME packet every 20ms + debug.message_sending_average_delay_ms = BigInt(20); + debug.average_packet_delay_ms = BigInt(10); + debug.average_ack_delay_ms = BigInt(10); const config = new Config("my-awesome-wasm-client", validator, gatewayEndpoint, debug) diff --git a/clients/webassembly/src/client/mod.rs b/clients/webassembly/src/client/mod.rs index 64f08eade8..755345de05 100644 --- a/clients/webassembly/src/client/mod.rs +++ b/clients/webassembly/src/client/mod.rs @@ -242,6 +242,7 @@ impl NymClient { } console_log!("Starting topology refresher..."); + // TODO: re-enable // topology_refresher.start(); } @@ -354,14 +355,14 @@ impl NymClient { // Right now it's impossible to have async exported functions to take `&mut self` rather than mut self // TODO: try Rc> approach? - pub async fn send_message(mut self, message: String, recipient: String) -> Self { + pub async fn send_message(self, message: String, recipient: String) -> Self { console_log!("Sending {} to {}", message, recipient); let message_bytes = message.into_bytes(); self.send_binary_message(message_bytes, recipient).await } - pub async fn send_binary_message(mut self, message: Vec, recipient: String) -> Self { + pub async fn send_binary_message(self, message: Vec, recipient: String) -> Self { console_log!("Sending {} bytes to {}", message.len(), recipient); let recipient = Recipient::try_from_base58_string(recipient).unwrap(); diff --git a/clients/webassembly/src/lib.rs b/clients/webassembly/src/lib.rs index 61e3002009..fd18bec107 100644 --- a/clients/webassembly/src/lib.rs +++ b/clients/webassembly/src/lib.rs @@ -5,6 +5,7 @@ use wasm_bindgen::prelude::*; #[cfg(target_arch = "wasm32")] mod client; +#[cfg(target_arch = "wasm32")] pub mod gateway_selector; #[wasm_bindgen]