diff --git a/common/clients/directory-client/src/presence/mod.rs b/common/clients/directory-client/src/presence/mod.rs index f869a69e32..37d00fcb87 100644 --- a/common/clients/directory-client/src/presence/mod.rs +++ b/common/clients/directory-client/src/presence/mod.rs @@ -88,6 +88,7 @@ mod converting_mixnode_presence_into_topology_mixnode { let result: Result = mix_presence.try_into(); // assert!(result.is_err()) // This fails only for me. Why? + // ¯\_(ツ)_/¯ - works on my machine (and travis) } #[test] diff --git a/common/clients/multi-tcp-client/src/connection_manager/mod.rs b/common/clients/multi-tcp-client/src/connection_manager/mod.rs index bbd0f8bcd7..8a568e6ac9 100644 --- a/common/clients/multi-tcp-client/src/connection_manager/mod.rs +++ b/common/clients/multi-tcp-client/src/connection_manager/mod.rs @@ -1,21 +1,31 @@ use crate::connection_manager::reconnector::ConnectionReconnector; use crate::connection_manager::writer::ConnectionWriter; +use futures::channel::{mpsc, oneshot}; use futures::task::Poll; -use futures::AsyncWriteExt; +use futures::{AsyncWriteExt, StreamExt}; use log::*; use std::io; use std::net::SocketAddr; use std::time::Duration; +use tokio::runtime::Handle; mod reconnector; mod writer; +pub(crate) type ResponseSender = Option>>; + +pub(crate) type ConnectionManagerSender = mpsc::UnboundedSender<(Vec, ResponseSender)>; +type ConnectionManagerReceiver = mpsc::UnboundedReceiver<(Vec, ResponseSender)>; + enum ConnectionState<'a> { Writing(ConnectionWriter), Reconnecting(ConnectionReconnector<'a>), } pub(crate) struct ConnectionManager<'a> { + conn_tx: ConnectionManagerSender, + conn_rx: ConnectionManagerReceiver, + address: SocketAddr, maximum_reconnection_backoff: Duration, @@ -24,12 +34,14 @@ pub(crate) struct ConnectionManager<'a> { state: ConnectionState<'a>, } -impl<'a> ConnectionManager<'a> { +impl<'a> ConnectionManager<'static> { pub(crate) async fn new( address: SocketAddr, reconnection_backoff: Duration, maximum_reconnection_backoff: Duration, ) -> ConnectionManager<'a> { + let (conn_tx, conn_rx) = mpsc::unbounded(); + // based on initial connection we will either have a writer or a reconnector let state = match tokio::net::TcpStream::connect(address).await { Ok(conn) => ConnectionState::Writing(ConnectionWriter::new(conn)), @@ -47,6 +59,8 @@ impl<'a> ConnectionManager<'a> { }; ConnectionManager { + conn_tx, + conn_rx, address, maximum_reconnection_backoff, reconnection_backoff, @@ -54,7 +68,27 @@ impl<'a> ConnectionManager<'a> { } } - pub(crate) async fn send(&mut self, msg: &[u8]) -> io::Result<()> { + /// consumes Self and returns channel for communication + pub(crate) fn start(mut self, handle: &Handle) -> ConnectionManagerSender { + let sender_clone = self.conn_tx.clone(); + handle.spawn(async move { + while let Some(msg) = self.conn_rx.next().await { + let (msg_content, res_ch) = msg; + let res = self.handle_new_message(msg_content).await; + if let Some(res_ch) = res_ch { + if let Err(e) = res_ch.send(res) { + error!( + "failed to send response on the channel to the caller! - {:?}", + e + ); + } + } + } + }); + sender_clone + } + + async fn handle_new_message(&mut self, msg: Vec) -> io::Result<()> { if let ConnectionState::Reconnecting(conn_reconnector) = &mut self.state { // do a single poll rather than await for future to completely resolve let new_connection = match futures::poll!(conn_reconnector) { @@ -74,9 +108,12 @@ impl<'a> ConnectionManager<'a> { // we must be in writing state if we are here, either by being here from beginning or just // transitioning from reconnecting if let ConnectionState::Writing(conn_writer) = &mut self.state { - return match conn_writer.write_all(msg).await { + return match conn_writer.write_all(msg.as_ref()).await { // if we failed to write to connection we should reconnect // TODO: is this true? can we fail to write to a connection while it still remains open and valid? + + // TODO: change connection writer to somehow also poll for responses and + // change return type of this method from io::Result<> to io::Result> Ok(_) => Ok(()), Err(e) => { trace!("Creating connection reconnector!"); @@ -89,6 +126,7 @@ impl<'a> ConnectionManager<'a> { } }; }; - unreachable!() + + unreachable!(); } } diff --git a/common/clients/multi-tcp-client/src/lib.rs b/common/clients/multi-tcp-client/src/lib.rs index 34b64e77a5..43e64b11f2 100644 --- a/common/clients/multi-tcp-client/src/lib.rs +++ b/common/clients/multi-tcp-client/src/lib.rs @@ -1,87 +1,91 @@ -use crate::connection_manager::ConnectionManager; +use crate::connection_manager::{ConnectionManager, ConnectionManagerSender}; +use futures::channel::oneshot; use log::*; use std::collections::HashMap; use std::io; use std::net::SocketAddr; use std::time::Duration; +use tokio::runtime::Handle; mod connection_manager; pub struct Config { - initial_endpoints: Vec, initial_reconnection_backoff: Duration, maximum_reconnection_backoff: Duration, } impl Config { pub fn new( - initial_endpoints: Vec, initial_reconnection_backoff: Duration, maximum_reconnection_backoff: Duration, ) -> Self { Config { - initial_endpoints, initial_reconnection_backoff, maximum_reconnection_backoff, } } } -pub struct Client<'a> { - connections_managers: HashMap>, +pub struct Client { + runtime_handle: Handle, + connections_managers: HashMap, maximum_reconnection_backoff: Duration, initial_reconnection_backoff: Duration, } -impl<'a> Client<'a> { - pub async fn new(config: Config) -> Client<'a> { - let mut connections_managers = HashMap::new(); - for initial_endpoint in config.initial_endpoints { - connections_managers.insert( - initial_endpoint, - ConnectionManager::new( - initial_endpoint, - config.initial_reconnection_backoff, - config.maximum_reconnection_backoff, - ) - .await, - ); - } - +impl Client { + pub fn new(config: Config) -> Client { Client { - connections_managers, + // if the function is not called within tokio runtime context, this will panic + // but perhaps the code should be better structured to completely avoid this call + runtime_handle: Handle::try_current() + .expect("The client MUST BE used within tokio runtime context"), + connections_managers: HashMap::new(), initial_reconnection_backoff: config.maximum_reconnection_backoff, maximum_reconnection_backoff: config.initial_reconnection_backoff, } } - pub async fn send(&mut self, address: SocketAddr, message: &[u8]) -> io::Result<()> { + async fn start_new_connection_manager(&self, address: SocketAddr) -> ConnectionManagerSender { + ConnectionManager::new( + address, + self.initial_reconnection_backoff, + self.maximum_reconnection_backoff, + ) + .await + .start(&self.runtime_handle) + } + + // if wait_for_response is set to true, we will get information about any possible IO errors + // as well as (once implemented) received replies, however, this will also cause way longer + // waiting periods + pub async fn send( + &mut self, + address: SocketAddr, + message: Vec, + wait_for_response: bool, + ) -> io::Result<()> { if !self.connections_managers.contains_key(&address) { info!( "There is no existing connection to {:?} - it will be established now", address ); - // TODO: now we're blocking to establish TCP connection this need to be changed - // so that other connections could progress - let new_manager = ConnectionManager::new( - address, - self.initial_reconnection_backoff, - self.maximum_reconnection_backoff, - ) - .await; - - self.connections_managers.insert(address, new_manager); + let new_manager_sender = self.start_new_connection_manager(address).await; + self.connections_managers + .insert(address, new_manager_sender); } - // to optimize later by using channels and separate tokio tasks for each connection handler - // because right now say we want to write to addresses A and B - - // We have to wait until we're done dealing with A before we can do anything with B - self.connections_managers - .get_mut(&address) - .unwrap() - .send(&message) - .await + let manager = self.connections_managers.get_mut(&address).unwrap(); + + if wait_for_response { + let (res_tx, res_rx) = oneshot::channel(); + manager.unbounded_send((message, Some(res_tx))).unwrap(); + res_rx.await.unwrap() + } else { + manager.unbounded_send((message, None)).unwrap(); + Ok(()) + } } } @@ -146,22 +150,22 @@ mod tests { let mut rt = tokio::runtime::Runtime::new().unwrap(); let addr = "127.0.0.1:6000".parse().unwrap(); let reconnection_backoff = Duration::from_secs(1); - let client_config = - Config::new(vec![addr], reconnection_backoff, 10 * reconnection_backoff); + let client_config = Config::new(reconnection_backoff, 10 * reconnection_backoff); - let messages_to_send = vec![[1u8; SERVER_MSG_LEN], [2; SERVER_MSG_LEN]]; + let messages_to_send = vec![[1u8; SERVER_MSG_LEN].to_vec(), [2; SERVER_MSG_LEN].to_vec()]; let dummy_server = rt.block_on(DummyServer::new(addr)); let finished_dummy_server_future = rt.spawn(dummy_server.listen_until(&CLOSE_MESSAGE)); - let mut c = rt.block_on(Client::new(client_config)); + let mut c = rt.enter(|| Client::new(client_config)); for msg in &messages_to_send { - rt.block_on(c.send(addr, msg)).unwrap(); + rt.block_on(c.send(addr, msg.clone(), true)).unwrap(); } // kill server - rt.block_on(c.send(addr, &CLOSE_MESSAGE)).unwrap(); + rt.block_on(c.send(addr, CLOSE_MESSAGE.to_vec(), true)) + .unwrap(); let received_messages = rt .block_on(finished_dummy_server_future) .unwrap() @@ -170,17 +174,22 @@ mod tests { assert_eq!(received_messages, messages_to_send); // try to send - go into reconnection - let post_kill_message = [3u8; SERVER_MSG_LEN]; + let post_kill_message = [3u8; SERVER_MSG_LEN].to_vec(); // we are trying to send to killed server - assert!(rt.block_on(c.send(addr, &post_kill_message)).is_err()); + assert!(rt + .block_on(c.send(addr, post_kill_message.clone(), true)) + .is_err()); let new_dummy_server = rt.block_on(DummyServer::new(addr)); let new_server_future = rt.spawn(new_dummy_server.listen_until(&CLOSE_MESSAGE)); // keep sending after we leave reconnection backoff and reconnect loop { - if rt.block_on(c.send(addr, &post_kill_message)).is_ok() { + if rt + .block_on(c.send(addr, post_kill_message.clone(), true)) + .is_ok() + { break; } rt.block_on( @@ -189,7 +198,8 @@ mod tests { } // kill the server to ensure it actually got the message - rt.block_on(c.send(addr, &CLOSE_MESSAGE)).unwrap(); + rt.block_on(c.send(addr, CLOSE_MESSAGE.to_vec(), true)) + .unwrap(); let new_received_messages = rt.block_on(new_server_future).unwrap().get_received(); assert_eq!(post_kill_message.to_vec(), new_received_messages[0]); } @@ -199,21 +209,21 @@ mod tests { let mut rt = tokio::runtime::Runtime::new().unwrap(); let addr = "127.0.0.1:6001".parse().unwrap(); let reconnection_backoff = Duration::from_secs(2); - let client_config = - Config::new(vec![addr], reconnection_backoff, 10 * reconnection_backoff); + let client_config = Config::new(reconnection_backoff, 10 * reconnection_backoff); - let messages_to_send = vec![[1u8; SERVER_MSG_LEN], [2; SERVER_MSG_LEN]]; + let messages_to_send = vec![[1u8; SERVER_MSG_LEN].to_vec(), [2; SERVER_MSG_LEN].to_vec()]; let dummy_server = rt.block_on(DummyServer::new(addr)); let finished_dummy_server_future = rt.spawn(dummy_server.listen_until(&CLOSE_MESSAGE)); - let mut c = rt.block_on(Client::new(client_config)); + let mut c = rt.enter(|| Client::new(client_config)); for msg in &messages_to_send { - rt.block_on(c.send(addr, msg)).unwrap(); + rt.block_on(c.send(addr, msg.clone(), true)).unwrap(); } - rt.block_on(c.send(addr, &CLOSE_MESSAGE)).unwrap(); + rt.block_on(c.send(addr, CLOSE_MESSAGE.to_vec(), true)) + .unwrap(); // the server future should have already been resolved let received_messages = rt diff --git a/mixnode/src/node/mod.rs b/mixnode/src/node/mod.rs index 4ba30307da..20b7133408 100644 --- a/mixnode/src/node/mod.rs +++ b/mixnode/src/node/mod.rs @@ -87,16 +87,11 @@ impl MixNode { fn start_packet_forwarder(&mut self) -> mpsc::UnboundedSender<(SocketAddr, Vec)> { info!("Starting packet forwarder..."); - - // this can later be replaced with topology information - let initial_addresses = vec![]; - self.runtime - .block_on(packet_forwarding::PacketForwarder::new( - initial_addresses, - self.config.get_packet_forwarding_initial_backoff(), - self.config.get_packet_forwarding_maximum_backoff(), - )) - .start(self.runtime.handle()) + packet_forwarding::PacketForwarder::new( + self.config.get_packet_forwarding_initial_backoff(), + self.config.get_packet_forwarding_maximum_backoff(), + ) + .start(self.runtime.handle()) } pub fn run(&mut self) { diff --git a/mixnode/src/node/packet_forwarding.rs b/mixnode/src/node/packet_forwarding.rs index 8335322081..bdba773d84 100644 --- a/mixnode/src/node/packet_forwarding.rs +++ b/mixnode/src/node/packet_forwarding.rs @@ -1,24 +1,21 @@ use futures::channel::mpsc; use futures::StreamExt; -use log::*; use std::net::SocketAddr; use std::time::Duration; use tokio::runtime::Handle; -pub(crate) struct PacketForwarder<'a> { - tcp_client: multi_tcp_client::Client<'a>, +pub(crate) struct PacketForwarder { + tcp_client: multi_tcp_client::Client, conn_tx: mpsc::UnboundedSender<(SocketAddr, Vec)>, conn_rx: mpsc::UnboundedReceiver<(SocketAddr, Vec)>, } -impl PacketForwarder<'static> { - pub(crate) async fn new( - initial_endpoints: Vec, +impl PacketForwarder { + pub(crate) fn new( initial_reconnection_backoff: Duration, maximum_reconnection_backoff: Duration, - ) -> PacketForwarder<'static> { + ) -> PacketForwarder { let tcp_client_config = multi_tcp_client::Config::new( - initial_endpoints, initial_reconnection_backoff, maximum_reconnection_backoff, ); @@ -26,7 +23,7 @@ impl PacketForwarder<'static> { let (conn_tx, conn_rx) = mpsc::unbounded(); PacketForwarder { - tcp_client: multi_tcp_client::Client::new(tcp_client_config).await, + tcp_client: multi_tcp_client::Client::new(tcp_client_config), conn_tx, conn_rx, } @@ -37,10 +34,9 @@ impl PacketForwarder<'static> { let sender_channel = self.conn_tx.clone(); handle.spawn(async move { while let Some((address, packet)) = self.conn_rx.next().await { - match self.tcp_client.send(address, &packet).await { - Err(e) => warn!("Failed to forward packet to {:?} - {:?}", address, e), - Ok(_) => trace!("Forwarded packet to {:?}", address), - } + // as a mix node we don't care about responses, we just want to fire packets + // as quickly as possible + self.tcp_client.send(address, packet, false).await.unwrap(); // if we're not waiting for response, we MUST get an Ok } }); sender_channel diff --git a/nym-client/src/client/mix_traffic.rs b/nym-client/src/client/mix_traffic.rs index c71a7c3be6..0eccb82b73 100644 --- a/nym-client/src/client/mix_traffic.rs +++ b/nym-client/src/client/mix_traffic.rs @@ -18,45 +18,35 @@ impl MixMessage { } // TODO: put our TCP client here -pub(crate) struct MixTrafficController<'a> { - tcp_client: multi_tcp_client::Client<'a>, +pub(crate) struct MixTrafficController { + tcp_client: multi_tcp_client::Client, mix_rx: MixMessageReceiver, } -impl MixTrafficController<'static> { - pub(crate) async fn new( - initial_endpoints: Vec, +impl MixTrafficController { + pub(crate) fn new( initial_reconnection_backoff: Duration, maximum_reconnection_backoff: Duration, mix_rx: MixMessageReceiver, ) -> Self { let tcp_client_config = multi_tcp_client::Config::new( - initial_endpoints, initial_reconnection_backoff, maximum_reconnection_backoff, ); MixTrafficController { - tcp_client: multi_tcp_client::Client::new(tcp_client_config).await, + tcp_client: multi_tcp_client::Client::new(tcp_client_config), mix_rx, } } async fn on_message(&mut self, mix_message: MixMessage) { debug!("Got a mix_message for {:?}", mix_message.0); - match self - .tcp_client - .send(mix_message.0, &mix_message.1.to_bytes()) + self.tcp_client + // TODO: possibly we might want to get an actual result here at some point + .send(mix_message.0, mix_message.1.to_bytes(), false) .await - { - Ok(_) => trace!("sent a mix message"), - // TODO: should there be some kind of threshold of failed messages - // that if reached, the application blows? - Err(e) => error!( - "We failed to send the packet to {} - {:?}", - mix_message.0, e - ), - }; + .unwrap(); // if we're not waiting for response, we MUST get an Ok } pub(crate) async fn run(&mut self) { diff --git a/nym-client/src/client/mod.rs b/nym-client/src/client/mod.rs index d80f512850..2dc70e8357 100644 --- a/nym-client/src/client/mod.rs +++ b/nym-client/src/client/mod.rs @@ -212,16 +212,12 @@ impl NymClient { // controller for sending sphinx packets to mixnet (either real traffic or cover traffic) fn start_mix_traffic_controller(&mut self, mix_rx: MixMessageReceiver) { info!("Starting mix trafic controller..."); - // TODO: possible optimisation: set the initial endpoints to all known mixes from layer 1 - let initial_mix_endpoints = Vec::new(); - self.runtime - .block_on(MixTrafficController::new( - initial_mix_endpoints, - self.config.get_packet_forwarding_initial_backoff(), - self.config.get_packet_forwarding_maximum_backoff(), - mix_rx, - )) - .start(self.runtime.handle()); + MixTrafficController::new( + self.config.get_packet_forwarding_initial_backoff(), + self.config.get_packet_forwarding_maximum_backoff(), + mix_rx, + ) + .start(self.runtime.handle()); } fn start_socket_listener(