From f3ed0bb11f6412821d64340bf1f6d289d59d2274 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jon=20H=C3=A4ggblad?= Date: Fri, 9 Sep 2022 15:00:13 +0200 Subject: [PATCH] gateway-client: disable shutdown signalling for wasm --- common/client-libs/gateway-client/Cargo.toml | 4 +- .../client-libs/gateway-client/src/client.rs | 50 ++++++++++++++----- .../gateway-client/src/packet_router.rs | 6 ++- .../gateway-client/src/socket_state.rs | 30 +++++++---- 4 files changed, 67 insertions(+), 23 deletions(-) diff --git a/common/client-libs/gateway-client/Cargo.toml b/common/client-libs/gateway-client/Cargo.toml index d4688e9428..fa651db447 100644 --- a/common/client-libs/gateway-client/Cargo.toml +++ b/common/client-libs/gateway-client/Cargo.toml @@ -27,7 +27,6 @@ gateway-requests = { path = "../../../gateway/gateway-requests" } network-defaults = { path = "../../network-defaults" } nymsphinx = { path = "../../nymsphinx" } pemstore = { path = "../../pemstore" } -task = { path = "../../task" } validator-client = { path = "../validator-client", optional = true } [dependencies.tungstenite] @@ -49,6 +48,9 @@ version = "0.14" [target."cfg(not(target_arch = \"wasm32\"))".dependencies.credential-storage] path = "../../credential-storage" +[target."cfg(not(target_arch = \"wasm32\"))".dependencies.task] +path = "../../task" + # wasm-only dependencies [target."cfg(target_arch = \"wasm32\")".dependencies.wasm-bindgen] version = "0.2" diff --git a/common/client-libs/gateway-client/src/client.rs b/common/client-libs/gateway-client/src/client.rs index 71eb795e65..2c6226a013 100644 --- a/common/client-libs/gateway-client/src/client.rs +++ b/common/client-libs/gateway-client/src/client.rs @@ -30,6 +30,7 @@ use rand::rngs::OsRng; use std::convert::TryFrom; use std::sync::Arc; use std::time::Duration; +#[cfg(not(target_arch = "wasm32"))] use task::ShutdownListener; use tungstenite::protocol::Message; @@ -67,6 +68,7 @@ pub struct GatewayClient { /// Delay between each subsequent reconnection attempt. reconnection_backoff: Duration, + #[cfg(not(target_arch = "wasm32"))] /// Listen to shutdown messages. shutdown: Option, } @@ -84,7 +86,7 @@ impl GatewayClient { ack_sender: AcknowledgementSender, response_timeout_duration: Duration, bandwidth_controller: Option>, - shutdown: Option, + #[cfg(not(target_arch = "wasm32"))] shutdown: Option, ) -> Self { GatewayClient { authenticated: false, @@ -96,12 +98,18 @@ impl GatewayClient { local_identity, shared_key, connection: SocketState::NotConnected, - packet_router: PacketRouter::new(ack_sender, mixnet_message_sender, shutdown.clone()), + packet_router: PacketRouter::new( + ack_sender, + mixnet_message_sender, + #[cfg(not(target_arch = "wasm32"))] + shutdown.clone(), + ), response_timeout_duration, bandwidth_controller, should_reconnect_on_failure: true, reconnection_attempts: DEFAULT_RECONNECTION_ATTEMPTS, reconnection_backoff: DEFAULT_RECONNECTION_BACKOFF, + #[cfg(not(target_arch = "wasm32"))] shutdown, } } @@ -129,7 +137,7 @@ impl GatewayClient { gateway_owner: String, local_identity: Arc, response_timeout_duration: Duration, - shutdown: Option, + #[cfg(not(target_arch = "wasm32"))] shutdown: Option, ) -> Self { use futures::channel::mpsc; @@ -137,7 +145,12 @@ impl GatewayClient { // perfectly fine here, because it's not meant to be used let (ack_tx, _) = mpsc::unbounded(); let (mix_tx, _) = mpsc::unbounded(); - let packet_router = PacketRouter::new(ack_tx, mix_tx, shutdown.clone()); + let packet_router = PacketRouter::new( + ack_tx, + mix_tx, + #[cfg(not(target_arch = "wasm32"))] + shutdown.clone(), + ); GatewayClient { authenticated: false, @@ -155,6 +168,7 @@ impl GatewayClient { should_reconnect_on_failure: false, reconnection_attempts: DEFAULT_RECONNECTION_ATTEMPTS, reconnection_backoff: DEFAULT_RECONNECTION_BACKOFF, + #[cfg(not(target_arch = "wasm32"))] shutdown, } } @@ -287,16 +301,27 @@ impl GatewayClient { let mut fused_timeout = timeout.fuse(); let mut fused_stream = conn.fuse(); - // Bit of an ugly workaround for selecting on an `Option`. The unwrap is lazy so we use - // this bool inside the `select` to guard against unwrapping in the `None` case. - let shutdown_is_some = self.shutdown.is_some(); - let shutdown_recv_lazy = async { self.shutdown.clone().unwrap().recv().await }; - tokio::pin!(shutdown_recv_lazy); + // Bit of an ugly workaround for selecting on an `Option` without having access to + // `tokio::select` + #[cfg(not(target_arch = "wasm32"))] + let shutdown = { + let m_shutdown = self.shutdown.clone(); + async { + if let Some(mut s) = m_shutdown { + s.recv().await + } + } + .fuse() + }; + #[cfg(not(target_arch = "wasm32"))] + tokio::pin!(shutdown); + + #[cfg(target_arch = "wasm32")] + let mut shutdown = Box::pin(async {}.fuse()); loop { - tokio::select! { - biased; - _ = &mut shutdown_recv_lazy, if shutdown_is_some => { + futures::select! { + _ = shutdown => { log::trace!("GatewayClient control response: Received shutdown"); log::debug!("GatewayClient control response: Exiting"); break Err(GatewayClientError::ConnectionClosedGatewayShutdown); @@ -745,6 +770,7 @@ impl GatewayClient { .as_ref() .expect("no shared key present even though we're authenticated!"), ), + #[cfg(not(target_arch = "wasm32"))] self.shutdown.clone(), ) } diff --git a/common/client-libs/gateway-client/src/packet_router.rs b/common/client-libs/gateway-client/src/packet_router.rs index 2605b291fb..13977efdc7 100644 --- a/common/client-libs/gateway-client/src/packet_router.rs +++ b/common/client-libs/gateway-client/src/packet_router.rs @@ -8,6 +8,7 @@ use futures::channel::mpsc; use log::*; use nymsphinx::addressing::nodes::MAX_NODE_ADDRESS_UNPADDED_LEN; use nymsphinx::params::packet_sizes::PacketSize; +#[cfg(not(target_arch = "wasm32"))] use task::ShutdownListener; use crate::error::GatewayClientError; @@ -22,6 +23,7 @@ pub type AcknowledgementReceiver = mpsc::UnboundedReceiver>>; pub struct PacketRouter { ack_sender: AcknowledgementSender, mixnet_message_sender: MixnetMessageSender, + #[cfg(not(target_arch = "wasm32"))] shutdown: Option, } @@ -29,11 +31,12 @@ impl PacketRouter { pub fn new( ack_sender: AcknowledgementSender, mixnet_message_sender: MixnetMessageSender, - shutdown: Option, + #[cfg(not(target_arch = "wasm32"))] shutdown: Option, ) -> Self { PacketRouter { ack_sender, mixnet_message_sender, + #[cfg(not(target_arch = "wasm32"))] shutdown, } } @@ -72,6 +75,7 @@ impl PacketRouter { if !received_messages.is_empty() { trace!("routing 'real'"); if let Err(err) = self.mixnet_message_sender.unbounded_send(received_messages) { + #[cfg(not(target_arch = "wasm32"))] if let Some(shutdown) = &mut self.shutdown { if shutdown.is_shutdown_poll() { // This should ideally not happen, but it's ok diff --git a/common/client-libs/gateway-client/src/socket_state.rs b/common/client-libs/gateway-client/src/socket_state.rs index e828f7dbbf..8194f7c8c7 100644 --- a/common/client-libs/gateway-client/src/socket_state.rs +++ b/common/client-libs/gateway-client/src/socket_state.rs @@ -11,6 +11,7 @@ use gateway_requests::registration::handshake::SharedKeys; use gateway_requests::BinaryResponse; use log::*; use std::sync::Arc; +#[cfg(not(target_arch = "wasm32"))] use task::ShutdownListener; use tungstenite::Message; @@ -90,7 +91,7 @@ impl PartiallyDelegated { conn: WsConn, packet_router: PacketRouter, shared_key: Arc, - shutdown: Option, + #[cfg(not(target_arch = "wasm32"))] shutdown: Option, ) -> Self { // when called for, it NEEDS TO yield back the stream so that we could merge it and // read control request responses. @@ -104,16 +105,27 @@ impl PartiallyDelegated { let mut fused_stream = (&mut stream).fuse(); let mut packet_router = packet_router; - // Bit of an ugly workaround for selecting on an `Option`. The unwrap is lazy so we use - // this bool inside the `select` to guard against unwrapping in the `None` case. - let shutdown_is_some = shutdown.is_some(); - let shutdown_recv_lazy = async { shutdown.unwrap().recv().await }; - tokio::pin!(shutdown_recv_lazy); + // Bit of an ugly workaround for selecting on an `Option` without having access to + // `tokio::select` + #[cfg(not(target_arch = "wasm32"))] + let shutdown = { + let m_shutdown = shutdown.clone(); + async { + if let Some(mut s) = m_shutdown { + s.recv().await + } + } + .fuse() + }; + #[cfg(not(target_arch = "wasm32"))] + tokio::pin!(shutdown); + + #[cfg(target_arch = "wasm32")] + let mut shutdown = Box::pin(async {}.fuse()); let ret_err = loop { - tokio::select! { - biased; - _ = &mut shutdown_recv_lazy, if shutdown_is_some => { + futures::select! { + _ = shutdown => { log::trace!("GatewayClient listener: Received shutdown"); log::debug!("GatewayClient listener: Exiting"); return;