From 51ae70c02be2562ab53678047110cce383b28816 Mon Sep 17 00:00:00 2001 From: Jedrzej Stuczynski Date: Thu, 12 Mar 2020 11:16:00 +0000 Subject: [PATCH 01/10] Simple connection error listener --- .../src/connection_manager/error_reader.rs | 33 +++++++++++++++++++ 1 file changed, 33 insertions(+) create mode 100644 common/clients/multi-tcp-client/src/connection_manager/error_reader.rs diff --git a/common/clients/multi-tcp-client/src/connection_manager/error_reader.rs b/common/clients/multi-tcp-client/src/connection_manager/error_reader.rs new file mode 100644 index 0000000000..3996a2465f --- /dev/null +++ b/common/clients/multi-tcp-client/src/connection_manager/error_reader.rs @@ -0,0 +1,33 @@ +use futures::channel::mpsc; +use futures::StreamExt; +use log::*; +use std::io; +use std::net::SocketAddr; +use tokio::runtime::Handle; +use tokio::task::JoinHandle; + +type ConnectionErrorResponse = (SocketAddr, io::Result<()>); +type ConnectionErrorSender = mpsc::UnboundedSender; +type ConnectionErrorReceiver = mpsc::UnboundedReceiver; + +struct ConnectionErrorReader { + error_rx: ConnectionErrorReceiver, +} + +impl ConnectionErrorReader { + fn new(error_rx: ConnectionErrorReceiver) -> Self { + ConnectionErrorReader { error_rx } + } + + fn start(mut self, handle: &Handle) -> JoinHandle<()> { + handle.spawn(async move { + while let Some(err_res) = self.error_rx.next().await { + let (source, err) = err_res; + match err { + Ok(_) => trace!("packet to {} was sent successfully!", source.to_string()), + Err(e) => warn!("failed to send packet to {} - {:?}", source.to_string(), e), + } + } + }) + } +} From f00bc5a3d1f88c1cc59b4de771053fea1191a8ef Mon Sep 17 00:00:00 2001 From: Jedrzej Stuczynski Date: Thu, 12 Mar 2020 16:22:18 +0000 Subject: [PATCH 02/10] Moved the error reader level up --- .../multi-tcp-client/src/{connection_manager => }/error_reader.rs | 0 1 file changed, 0 insertions(+), 0 deletions(-) rename common/clients/multi-tcp-client/src/{connection_manager => }/error_reader.rs (100%) diff --git a/common/clients/multi-tcp-client/src/connection_manager/error_reader.rs b/common/clients/multi-tcp-client/src/error_reader.rs similarity index 100% rename from common/clients/multi-tcp-client/src/connection_manager/error_reader.rs rename to common/clients/multi-tcp-client/src/error_reader.rs From 8279efb684c732918f3b5c2eb99a48755fb796d8 Mon Sep 17 00:00:00 2001 From: Jedrzej Stuczynski Date: Thu, 12 Mar 2020 16:22:40 +0000 Subject: [PATCH 03/10] Exposed methods --- .../clients/multi-tcp-client/src/error_reader.rs | 15 ++++++++------- 1 file changed, 8 insertions(+), 7 deletions(-) diff --git a/common/clients/multi-tcp-client/src/error_reader.rs b/common/clients/multi-tcp-client/src/error_reader.rs index 3996a2465f..8a30197942 100644 --- a/common/clients/multi-tcp-client/src/error_reader.rs +++ b/common/clients/multi-tcp-client/src/error_reader.rs @@ -6,26 +6,27 @@ use std::net::SocketAddr; use tokio::runtime::Handle; use tokio::task::JoinHandle; -type ConnectionErrorResponse = (SocketAddr, io::Result<()>); -type ConnectionErrorSender = mpsc::UnboundedSender; -type ConnectionErrorReceiver = mpsc::UnboundedReceiver; +pub(crate) type ConnectionErrorResponse = (SocketAddr, io::Result<()>); +pub(crate) type ConnectionErrorSender = mpsc::UnboundedSender; +pub(crate) type ConnectionErrorReceiver = mpsc::UnboundedReceiver; -struct ConnectionErrorReader { +pub(crate) struct ConnectionErrorReader { error_rx: ConnectionErrorReceiver, } impl ConnectionErrorReader { - fn new(error_rx: ConnectionErrorReceiver) -> Self { + pub(crate) fn new(error_rx: ConnectionErrorReceiver) -> Self { ConnectionErrorReader { error_rx } } - fn start(mut self, handle: &Handle) -> JoinHandle<()> { + pub(crate) fn start(mut self, handle: &Handle) -> JoinHandle<()> { handle.spawn(async move { while let Some(err_res) = self.error_rx.next().await { let (source, err) = err_res; match err { - Ok(_) => trace!("packet to {} was sent successfully!", source.to_string()), + // Ok(_) => trace!("packet to {} was sent successfully!", source.to_string()), Err(e) => warn!("failed to send packet to {} - {:?}", source.to_string(), e), + Ok(_) => (), // right now we're not expecting to receive any 'Ok' responses } } }) From 7143ab9b53ffa96a3220ecd02a3ad9922e01e2fc Mon Sep 17 00:00:00 2001 From: Jedrzej Stuczynski Date: Thu, 12 Mar 2020 16:23:47 +0000 Subject: [PATCH 04/10] Changed connection manager to be accessed via a channel --- .../src/connection_manager/mod.rs | 76 ++-- common/clients/multi-tcp-client/src/lib.rs | 359 +++++++++--------- 2 files changed, 233 insertions(+), 202 deletions(-) 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..1a6b76ffad 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 crate::error_reader::ConnectionErrorSender; +use futures::channel::mpsc; 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 ConnectionManagerSender = mpsc::UnboundedSender>; +type ConnectionManagerReceiver = mpsc::UnboundedReceiver>; + enum ConnectionState<'a> { Writing(ConnectionWriter), Reconnecting(ConnectionReconnector<'a>), } pub(crate) struct ConnectionManager<'a> { + conn_tx: ConnectionManagerSender, + conn_rx: ConnectionManagerReceiver, + + errors_tx: ConnectionErrorSender, address: SocketAddr, maximum_reconnection_backoff: Duration, @@ -24,12 +34,15 @@ 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, + errors_tx: ConnectionErrorSender, ) -> 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 +60,9 @@ impl<'a> ConnectionManager<'a> { }; ConnectionManager { + conn_tx, + conn_rx, + errors_tx, address, maximum_reconnection_backoff, reconnection_backoff, @@ -54,41 +70,55 @@ 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 { + self.handle_new_message(msg).await; + } + }); + sender_clone + } + + async fn handle_new_message(&mut self, msg: Vec) { + info!("sending to {:?}", self.address); 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) { Poll::Pending => { - return Err(io::Error::new( - io::ErrorKind::BrokenPipe, - "connection is broken - reconnection is in progress", - )) + self.errors_tx + .unbounded_send(( + self.address, + Err(io::Error::new( + io::ErrorKind::BrokenPipe, + "connection is broken - reconnection is in progress", + )), + )) + .unwrap(); + return; } Poll::Ready(conn) => conn, }; - debug!("Managed to reconnect to {}!", self.address); + info!("Managed to reconnect to {}!", self.address); self.state = ConnectionState::Writing(ConnectionWriter::new(new_connection)); } // 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 { - // 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? - Ok(_) => Ok(()), - Err(e) => { - trace!("Creating connection reconnector!"); - self.state = ConnectionState::Reconnecting(ConnectionReconnector::new( - self.address, - self.reconnection_backoff, - self.maximum_reconnection_backoff, - )); - Err(e) - } - }; + if let Err(e) = conn_writer.write_all(msg.as_ref()).await { + info!("Creating connection reconnector!"); + self.state = ConnectionState::Reconnecting(ConnectionReconnector::new( + self.address, + self.reconnection_backoff, + self.maximum_reconnection_backoff, + )); + self.errors_tx + .unbounded_send((self.address, Err(e))) + .unwrap(); + } }; - unreachable!() } } diff --git a/common/clients/multi-tcp-client/src/lib.rs b/common/clients/multi-tcp-client/src/lib.rs index 34b64e77a5..3ebe74e895 100644 --- a/common/clients/multi-tcp-client/src/lib.rs +++ b/common/clients/multi-tcp-client/src/lib.rs @@ -1,226 +1,227 @@ -use crate::connection_manager::ConnectionManager; +use crate::connection_manager::{ConnectionManager, ConnectionManagerSender}; +use crate::error_reader::{ConnectionErrorReader, ConnectionErrorSender}; +use futures::channel::mpsc; use log::*; use std::collections::HashMap; -use std::io; use std::net::SocketAddr; use std::time::Duration; +use tokio::runtime::Handle; mod connection_manager; +mod error_reader; 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, + errors_tx: ConnectionErrorSender, + 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 async fn start_new(config: Config) -> Client { + let (errors_tx, errors_rx) = mpsc::unbounded(); + let errors_reader = ConnectionErrorReader::new(errors_rx); - Client { - connections_managers, + let client = Client { + // 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"), + errors_tx, + connections_managers: HashMap::new(), initial_reconnection_backoff: config.maximum_reconnection_backoff, maximum_reconnection_backoff: config.initial_reconnection_backoff, - } + }; + + errors_reader.start(&client.runtime_handle); + client } - 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, + self.errors_tx.clone(), + ) + .await + .start(&self.runtime_handle) + } + + pub async fn send(&mut self, address: SocketAddr, message: Vec) { 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 + .unbounded_send(message) + .unwrap(); } } -#[cfg(test)] -mod tests { - use super::*; - use std::str; - use std::time; - use tokio::prelude::*; - - const SERVER_MSG_LEN: usize = 16; - const CLOSE_MESSAGE: [u8; SERVER_MSG_LEN] = [0; SERVER_MSG_LEN]; - - struct DummyServer { - received_buf: Vec>, - listener: tokio::net::TcpListener, - } - - impl DummyServer { - async fn new(address: SocketAddr) -> Self { - DummyServer { - received_buf: Vec::new(), - listener: tokio::net::TcpListener::bind(address).await.unwrap(), - } - } - - fn get_received(&self) -> Vec> { - self.received_buf.clone() - } - - // this is only used in tests so slightly higher logging levels are fine - async fn listen_until(mut self, close_message: &[u8]) -> Self { - let (mut socket, _) = self.listener.accept().await.unwrap(); - loop { - let mut buf = [0u8; SERVER_MSG_LEN]; - match socket.read(&mut buf).await { - Ok(n) if n == 0 => { - info!("Remote connection closed"); - return self; - } - Ok(n) => { - info!("received ({}) - {:?}", n, str::from_utf8(buf[..n].as_ref())); - - if buf[..n].as_ref() == close_message { - info!("closing..."); - socket.shutdown(std::net::Shutdown::Both).unwrap(); - return self; - } else { - self.received_buf.push(buf[..n].to_vec()); - } - } - Err(e) => { - panic!("failed to read from socket; err = {:?}", e); - } - }; - } - } - } - - #[test] - fn client_reconnects_to_server_after_it_went_down() { - 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 messages_to_send = vec![[1u8; SERVER_MSG_LEN], [2; SERVER_MSG_LEN]]; - - 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)); - - for msg in &messages_to_send { - rt.block_on(c.send(addr, msg)).unwrap(); - } - - // kill server - rt.block_on(c.send(addr, &CLOSE_MESSAGE)).unwrap(); - let received_messages = rt - .block_on(finished_dummy_server_future) - .unwrap() - .get_received(); - - assert_eq!(received_messages, messages_to_send); - - // try to send - go into reconnection - let post_kill_message = [3u8; SERVER_MSG_LEN]; - - // we are trying to send to killed server - assert!(rt.block_on(c.send(addr, &post_kill_message)).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() { - break; - } - rt.block_on( - async move { tokio::time::delay_for(time::Duration::from_millis(50)).await }, - ); - } - - // kill the server to ensure it actually got the message - rt.block_on(c.send(addr, &CLOSE_MESSAGE)).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]); - } - - #[test] - fn server_receives_all_sent_messages_when_up() { - 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 messages_to_send = vec![[1u8; SERVER_MSG_LEN], [2; SERVER_MSG_LEN]]; - - 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)); - - for msg in &messages_to_send { - rt.block_on(c.send(addr, msg)).unwrap(); - } - - rt.block_on(c.send(addr, &CLOSE_MESSAGE)).unwrap(); - - // the server future should have already been resolved - let received_messages = rt - .block_on(finished_dummy_server_future) - .unwrap() - .get_received(); - - assert_eq!(received_messages, messages_to_send); - } -} +// #[cfg(test)] +// mod tests { +// use super::*; +// use std::str; +// use std::time; +// use tokio::prelude::*; +// +// const SERVER_MSG_LEN: usize = 16; +// const CLOSE_MESSAGE: [u8; SERVER_MSG_LEN] = [0; SERVER_MSG_LEN]; +// +// struct DummyServer { +// received_buf: Vec>, +// listener: tokio::net::TcpListener, +// } +// +// impl DummyServer { +// async fn new(address: SocketAddr) -> Self { +// DummyServer { +// received_buf: Vec::new(), +// listener: tokio::net::TcpListener::bind(address).await.unwrap(), +// } +// } +// +// fn get_received(&self) -> Vec> { +// self.received_buf.clone() +// } +// +// // this is only used in tests so slightly higher logging levels are fine +// async fn listen_until(mut self, close_message: &[u8]) -> Self { +// let (mut socket, _) = self.listener.accept().await.unwrap(); +// loop { +// let mut buf = [0u8; SERVER_MSG_LEN]; +// match socket.read(&mut buf).await { +// Ok(n) if n == 0 => { +// info!("Remote connection closed"); +// return self; +// } +// Ok(n) => { +// info!("received ({}) - {:?}", n, str::from_utf8(buf[..n].as_ref())); +// +// if buf[..n].as_ref() == close_message { +// info!("closing..."); +// socket.shutdown(std::net::Shutdown::Both).unwrap(); +// return self; +// } else { +// self.received_buf.push(buf[..n].to_vec()); +// } +// } +// Err(e) => { +// panic!("failed to read from socket; err = {:?}", e); +// } +// }; +// } +// } +// } +// +// #[test] +// fn client_reconnects_to_server_after_it_went_down() { +// 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 messages_to_send = vec![[1u8; SERVER_MSG_LEN], [2; SERVER_MSG_LEN]]; +// +// 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)); +// +// for msg in &messages_to_send { +// rt.block_on(c.send(addr, msg)).unwrap(); +// } +// +// // kill server +// rt.block_on(c.send(addr, &CLOSE_MESSAGE)).unwrap(); +// let received_messages = rt +// .block_on(finished_dummy_server_future) +// .unwrap() +// .get_received(); +// +// assert_eq!(received_messages, messages_to_send); +// +// // try to send - go into reconnection +// let post_kill_message = [3u8; SERVER_MSG_LEN]; +// +// // we are trying to send to killed server +// assert!(rt.block_on(c.send(addr, &post_kill_message)).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() { +// break; +// } +// rt.block_on( +// async move { tokio::time::delay_for(time::Duration::from_millis(50)).await }, +// ); +// } +// +// // kill the server to ensure it actually got the message +// rt.block_on(c.send(addr, &CLOSE_MESSAGE)).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]); +// } +// +// #[test] +// fn server_receives_all_sent_messages_when_up() { +// 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 messages_to_send = vec![[1u8; SERVER_MSG_LEN], [2; SERVER_MSG_LEN]]; +// +// 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)); +// +// for msg in &messages_to_send { +// rt.block_on(c.send(addr, msg)).unwrap(); +// } +// +// rt.block_on(c.send(addr, &CLOSE_MESSAGE)).unwrap(); +// +// // the server future should have already been resolved +// let received_messages = rt +// .block_on(finished_dummy_server_future) +// .unwrap() +// .get_received(); +// +// assert_eq!(received_messages, messages_to_send); +// } +// } From fe85fefb5b5ec2fe4c0d7efd7751f91b9db22b36 Mon Sep 17 00:00:00 2001 From: Jedrzej Stuczynski Date: Thu, 12 Mar 2020 16:24:04 +0000 Subject: [PATCH 05/10] Client and mixnode adjustments due to previous changes --- mixnode/src/node/mod.rs | 4 ---- mixnode/src/node/packet_forwarding.rs | 18 ++++++------------ nym-client/src/client/mix_traffic.rs | 26 +++++++------------------- nym-client/src/client/mod.rs | 3 --- 4 files changed, 13 insertions(+), 38 deletions(-) diff --git a/mixnode/src/node/mod.rs b/mixnode/src/node/mod.rs index c73aade059..54309f02c5 100644 --- a/mixnode/src/node/mod.rs +++ b/mixnode/src/node/mod.rs @@ -86,12 +86,8 @@ 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(), )) diff --git a/mixnode/src/node/packet_forwarding.rs b/mixnode/src/node/packet_forwarding.rs index 8335322081..6fb507225c 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> { +impl PacketForwarder { pub(crate) async fn new( - initial_endpoints: Vec, 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::start_new(tcp_client_config).await, conn_tx, conn_rx, } @@ -37,10 +34,7 @@ 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), - } + self.tcp_client.send(address, packet).await; } }); sender_channel diff --git a/nym-client/src/client/mix_traffic.rs b/nym-client/src/client/mix_traffic.rs index c71a7c3be6..1b27bb7d17 100644 --- a/nym-client/src/client/mix_traffic.rs +++ b/nym-client/src/client/mix_traffic.rs @@ -18,45 +18,33 @@ 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> { +impl MixTrafficController { pub(crate) async fn new( - initial_endpoints: Vec, 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::start_new(tcp_client_config).await, 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()) - .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 - ), - }; + self.tcp_client + .send(mix_message.0, mix_message.1.to_bytes()) + .await; } 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..dd889faf54 100644 --- a/nym-client/src/client/mod.rs +++ b/nym-client/src/client/mod.rs @@ -212,11 +212,8 @@ 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, From 1271d6c1f52e514d4d33d31633ba533a55024341 Mon Sep 17 00:00:00 2001 From: Jedrzej Stuczynski Date: Fri, 13 Mar 2020 09:36:14 +0000 Subject: [PATCH 06/10] Comment regarding future work on getting possible responses back to the callee --- common/clients/multi-tcp-client/src/error_reader.rs | 6 ++++++ 1 file changed, 6 insertions(+) diff --git a/common/clients/multi-tcp-client/src/error_reader.rs b/common/clients/multi-tcp-client/src/error_reader.rs index 8a30197942..ccff3435ed 100644 --- a/common/clients/multi-tcp-client/src/error_reader.rs +++ b/common/clients/multi-tcp-client/src/error_reader.rs @@ -14,6 +14,12 @@ pub(crate) struct ConnectionErrorReader { error_rx: ConnectionErrorReceiver, } +// TODO: do some benchmarking and reconsider changing 'global' ConnectionErrorReader to requests +// of (Message, ReturnOneShotChannel): (Vec, oneshot::Sender>) +// this way callee would always get response to his specific request directly as well as any errors. +// and if he doesn't care about it, he could simply close channel immediately. +// Alternatively change the signature to (Vec, Option>>), +// so in the case of a 'None', the sender won't even attempt writing errors or responses received impl ConnectionErrorReader { pub(crate) fn new(error_rx: ConnectionErrorReceiver) -> Self { ConnectionErrorReader { error_rx } From b99a52bc1ba95c491a9842bc0f7f2b5865307983 Mon Sep 17 00:00:00 2001 From: Jedrzej Stuczynski Date: Fri, 13 Mar 2020 12:18:54 +0000 Subject: [PATCH 07/10] Removed oversensitive logging messages --- .../clients/multi-tcp-client/src/connection_manager/mod.rs | 5 ++--- 1 file changed, 2 insertions(+), 3 deletions(-) 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 1a6b76ffad..da2df6a1dc 100644 --- a/common/clients/multi-tcp-client/src/connection_manager/mod.rs +++ b/common/clients/multi-tcp-client/src/connection_manager/mod.rs @@ -82,7 +82,6 @@ impl<'a> ConnectionManager<'static> { } async fn handle_new_message(&mut self, msg: Vec) { - info!("sending to {:?}", self.address); 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) { @@ -101,7 +100,7 @@ impl<'a> ConnectionManager<'static> { Poll::Ready(conn) => conn, }; - info!("Managed to reconnect to {}!", self.address); + debug!("Managed to reconnect to {}!", self.address); self.state = ConnectionState::Writing(ConnectionWriter::new(new_connection)); } @@ -109,7 +108,7 @@ impl<'a> ConnectionManager<'static> { // transitioning from reconnecting if let ConnectionState::Writing(conn_writer) = &mut self.state { if let Err(e) = conn_writer.write_all(msg.as_ref()).await { - info!("Creating connection reconnector!"); + debug!("Creating connection reconnector!"); self.state = ConnectionState::Reconnecting(ConnectionReconnector::new( self.address, self.reconnection_backoff, From 3b2461080cdf6b5c2818c3947c7210d1e7db9845 Mon Sep 17 00:00:00 2001 From: jstuczyn Date: Mon, 16 Mar 2020 17:04:26 +0000 Subject: [PATCH 08/10] Changed the multi-tcp client to have an option to wait for an error response --- .../src/connection_manager/mod.rs | 69 ++++---- .../multi-tcp-client/src/error_reader.rs | 40 ----- common/clients/multi-tcp-client/src/lib.rs | 160 +++++++++++++++--- mixnode/src/node/mod.rs | 11 +- mixnode/src/node/packet_forwarding.rs | 8 +- nym-client/src/client/mix_traffic.rs | 10 +- nym-client/src/client/mod.rs | 13 +- 7 files changed, 200 insertions(+), 111 deletions(-) delete mode 100644 common/clients/multi-tcp-client/src/error_reader.rs 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 da2df6a1dc..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,7 +1,6 @@ use crate::connection_manager::reconnector::ConnectionReconnector; use crate::connection_manager::writer::ConnectionWriter; -use crate::error_reader::ConnectionErrorSender; -use futures::channel::mpsc; +use futures::channel::{mpsc, oneshot}; use futures::task::Poll; use futures::{AsyncWriteExt, StreamExt}; use log::*; @@ -13,8 +12,10 @@ use tokio::runtime::Handle; mod reconnector; mod writer; -pub(crate) type ConnectionManagerSender = mpsc::UnboundedSender>; -type ConnectionManagerReceiver = mpsc::UnboundedReceiver>; +pub(crate) type ResponseSender = Option>>; + +pub(crate) type ConnectionManagerSender = mpsc::UnboundedSender<(Vec, ResponseSender)>; +type ConnectionManagerReceiver = mpsc::UnboundedReceiver<(Vec, ResponseSender)>; enum ConnectionState<'a> { Writing(ConnectionWriter), @@ -25,7 +26,6 @@ pub(crate) struct ConnectionManager<'a> { conn_tx: ConnectionManagerSender, conn_rx: ConnectionManagerReceiver, - errors_tx: ConnectionErrorSender, address: SocketAddr, maximum_reconnection_backoff: Duration, @@ -39,7 +39,6 @@ impl<'a> ConnectionManager<'static> { address: SocketAddr, reconnection_backoff: Duration, maximum_reconnection_backoff: Duration, - errors_tx: ConnectionErrorSender, ) -> ConnectionManager<'a> { let (conn_tx, conn_rx) = mpsc::unbounded(); @@ -62,7 +61,6 @@ impl<'a> ConnectionManager<'static> { ConnectionManager { conn_tx, conn_rx, - errors_tx, address, maximum_reconnection_backoff, reconnection_backoff, @@ -75,27 +73,30 @@ impl<'a> ConnectionManager<'static> { let sender_clone = self.conn_tx.clone(); handle.spawn(async move { while let Some(msg) = self.conn_rx.next().await { - self.handle_new_message(msg).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) { + 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) { Poll::Pending => { - self.errors_tx - .unbounded_send(( - self.address, - Err(io::Error::new( - io::ErrorKind::BrokenPipe, - "connection is broken - reconnection is in progress", - )), - )) - .unwrap(); - return; + return Err(io::Error::new( + io::ErrorKind::BrokenPipe, + "connection is broken - reconnection is in progress", + )) } Poll::Ready(conn) => conn, }; @@ -107,17 +108,25 @@ impl<'a> ConnectionManager<'static> { // 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 { - if let Err(e) = conn_writer.write_all(msg.as_ref()).await { - debug!("Creating connection reconnector!"); - self.state = ConnectionState::Reconnecting(ConnectionReconnector::new( - self.address, - self.reconnection_backoff, - self.maximum_reconnection_backoff, - )); - self.errors_tx - .unbounded_send((self.address, Err(e))) - .unwrap(); - } + 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!"); + self.state = ConnectionState::Reconnecting(ConnectionReconnector::new( + self.address, + self.reconnection_backoff, + self.maximum_reconnection_backoff, + )); + Err(e) + } + }; }; + + unreachable!(); } } diff --git a/common/clients/multi-tcp-client/src/error_reader.rs b/common/clients/multi-tcp-client/src/error_reader.rs deleted file mode 100644 index ccff3435ed..0000000000 --- a/common/clients/multi-tcp-client/src/error_reader.rs +++ /dev/null @@ -1,40 +0,0 @@ -use futures::channel::mpsc; -use futures::StreamExt; -use log::*; -use std::io; -use std::net::SocketAddr; -use tokio::runtime::Handle; -use tokio::task::JoinHandle; - -pub(crate) type ConnectionErrorResponse = (SocketAddr, io::Result<()>); -pub(crate) type ConnectionErrorSender = mpsc::UnboundedSender; -pub(crate) type ConnectionErrorReceiver = mpsc::UnboundedReceiver; - -pub(crate) struct ConnectionErrorReader { - error_rx: ConnectionErrorReceiver, -} - -// TODO: do some benchmarking and reconsider changing 'global' ConnectionErrorReader to requests -// of (Message, ReturnOneShotChannel): (Vec, oneshot::Sender>) -// this way callee would always get response to his specific request directly as well as any errors. -// and if he doesn't care about it, he could simply close channel immediately. -// Alternatively change the signature to (Vec, Option>>), -// so in the case of a 'None', the sender won't even attempt writing errors or responses received -impl ConnectionErrorReader { - pub(crate) fn new(error_rx: ConnectionErrorReceiver) -> Self { - ConnectionErrorReader { error_rx } - } - - pub(crate) fn start(mut self, handle: &Handle) -> JoinHandle<()> { - handle.spawn(async move { - while let Some(err_res) = self.error_rx.next().await { - let (source, err) = err_res; - match err { - // Ok(_) => trace!("packet to {} was sent successfully!", source.to_string()), - Err(e) => warn!("failed to send packet to {} - {:?}", source.to_string(), e), - Ok(_) => (), // right now we're not expecting to receive any 'Ok' responses - } - } - }) - } -} diff --git a/common/clients/multi-tcp-client/src/lib.rs b/common/clients/multi-tcp-client/src/lib.rs index 3ebe74e895..b650b773cd 100644 --- a/common/clients/multi-tcp-client/src/lib.rs +++ b/common/clients/multi-tcp-client/src/lib.rs @@ -1,14 +1,13 @@ use crate::connection_manager::{ConnectionManager, ConnectionManagerSender}; -use crate::error_reader::{ConnectionErrorReader, ConnectionErrorSender}; -use futures::channel::mpsc; +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; -mod error_reader; pub struct Config { initial_reconnection_backoff: Duration, @@ -29,30 +28,22 @@ impl Config { pub struct Client { runtime_handle: Handle, - errors_tx: ConnectionErrorSender, connections_managers: HashMap, maximum_reconnection_backoff: Duration, initial_reconnection_backoff: Duration, } impl Client { - pub async fn start_new(config: Config) -> Client { - let (errors_tx, errors_rx) = mpsc::unbounded(); - let errors_reader = ConnectionErrorReader::new(errors_rx); - - let client = Client { + pub fn new(config: Config) -> Client { + Client { // 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"), - errors_tx, connections_managers: HashMap::new(), initial_reconnection_backoff: config.maximum_reconnection_backoff, maximum_reconnection_backoff: config.initial_reconnection_backoff, - }; - - errors_reader.start(&client.runtime_handle); - client + } } async fn start_new_connection_manager(&self, address: SocketAddr) -> ConnectionManagerSender { @@ -60,13 +51,20 @@ impl Client { address, self.initial_reconnection_backoff, self.maximum_reconnection_backoff, - self.errors_tx.clone(), ) .await .start(&self.runtime_handle) } - pub async fn send(&mut self, address: SocketAddr, message: Vec) { + // 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", @@ -78,14 +76,134 @@ impl Client { .insert(address, new_manager_sender); } - self.connections_managers - .get_mut(&address) - .unwrap() - .unbounded_send(message) - .unwrap(); + 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(()) + } } } +#[cfg(test)] +mod tests { + use super::*; + use std::str; + use std::time; + use tokio::prelude::*; + + const SERVER_MSG_LEN: usize = 16; + const CLOSE_MESSAGE: [u8; SERVER_MSG_LEN] = [0; SERVER_MSG_LEN]; + + struct DummyServer { + received_buf: Vec>, + listener: tokio::net::TcpListener, + } + + impl DummyServer { + async fn new(address: SocketAddr) -> Self { + DummyServer { + received_buf: Vec::new(), + listener: tokio::net::TcpListener::bind(address).await.unwrap(), + } + } + + fn get_received(&self) -> Vec> { + self.received_buf.clone() + } + + // this is only used in tests so slightly higher logging levels are fine + async fn listen_until(mut self, close_message: &[u8]) -> Self { + let (mut socket, _) = self.listener.accept().await.unwrap(); + loop { + let mut buf = [0u8; SERVER_MSG_LEN]; + match socket.read(&mut buf).await { + Ok(n) if n == 0 => { + info!("Remote connection closed"); + return self; + } + Ok(n) => { + info!("received ({}) - {:?}", n, str::from_utf8(buf[..n].as_ref())); + + if buf[..n].as_ref() == close_message { + info!("closing..."); + socket.shutdown(std::net::Shutdown::Both).unwrap(); + return self; + } else { + self.received_buf.push(buf[..n].to_vec()); + } + } + Err(e) => { + panic!("failed to read from socket; err = {:?}", e); + } + }; + } + } + } + + #[test] + fn client_reconnects_to_server_after_it_went_down() { + 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(reconnection_backoff, 10 * reconnection_backoff); + + 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.enter(|| Client::new(client_config)); + + for msg in &messages_to_send { + rt.block_on(c.send(addr, msg.clone(), true)).unwrap(); + } + + // kill server + rt.block_on(c.send(addr, CLOSE_MESSAGE.to_vec(), true)) + .unwrap(); + let received_messages = rt + .block_on(finished_dummy_server_future) + .unwrap() + .get_received(); + + assert_eq!(received_messages, messages_to_send); + + // try to send - go into reconnection + 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.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.clone(), true)) + .is_ok() + { + break; + } + rt.block_on( + async move { tokio::time::delay_for(time::Duration::from_millis(50)).await }, + ); + } + + // kill the server to ensure it actually got the message + 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]); + } + // #[cfg(test)] // mod tests { // use super::*; diff --git a/mixnode/src/node/mod.rs b/mixnode/src/node/mod.rs index e41e837328..20b7133408 100644 --- a/mixnode/src/node/mod.rs +++ b/mixnode/src/node/mod.rs @@ -87,12 +87,11 @@ impl MixNode { fn start_packet_forwarder(&mut self) -> mpsc::UnboundedSender<(SocketAddr, Vec)> { info!("Starting packet forwarder..."); - self.runtime - .block_on(packet_forwarding::PacketForwarder::new( - 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 6fb507225c..bdba773d84 100644 --- a/mixnode/src/node/packet_forwarding.rs +++ b/mixnode/src/node/packet_forwarding.rs @@ -11,7 +11,7 @@ pub(crate) struct PacketForwarder { } impl PacketForwarder { - pub(crate) async fn new( + pub(crate) fn new( initial_reconnection_backoff: Duration, maximum_reconnection_backoff: Duration, ) -> PacketForwarder { @@ -23,7 +23,7 @@ impl PacketForwarder { let (conn_tx, conn_rx) = mpsc::unbounded(); PacketForwarder { - tcp_client: multi_tcp_client::Client::start_new(tcp_client_config).await, + tcp_client: multi_tcp_client::Client::new(tcp_client_config), conn_tx, conn_rx, } @@ -34,7 +34,9 @@ impl PacketForwarder { let sender_channel = self.conn_tx.clone(); handle.spawn(async move { while let Some((address, packet)) = self.conn_rx.next().await { - self.tcp_client.send(address, packet).await; + // 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 1b27bb7d17..0eccb82b73 100644 --- a/nym-client/src/client/mix_traffic.rs +++ b/nym-client/src/client/mix_traffic.rs @@ -24,7 +24,7 @@ pub(crate) struct MixTrafficController { } impl MixTrafficController { - pub(crate) async fn new( + pub(crate) fn new( initial_reconnection_backoff: Duration, maximum_reconnection_backoff: Duration, mix_rx: MixMessageReceiver, @@ -35,7 +35,7 @@ impl MixTrafficController { ); MixTrafficController { - tcp_client: multi_tcp_client::Client::start_new(tcp_client_config).await, + tcp_client: multi_tcp_client::Client::new(tcp_client_config), mix_rx, } } @@ -43,8 +43,10 @@ impl MixTrafficController { async fn on_message(&mut self, mix_message: MixMessage) { debug!("Got a mix_message for {:?}", mix_message.0); self.tcp_client - .send(mix_message.0, mix_message.1.to_bytes()) - .await; + // 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 + .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 dd889faf54..2dc70e8357 100644 --- a/nym-client/src/client/mod.rs +++ b/nym-client/src/client/mod.rs @@ -212,13 +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..."); - self.runtime - .block_on(MixTrafficController::new( - 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( From a44272be6057adbc273f25c70cf2a12324a53dae Mon Sep 17 00:00:00 2001 From: jstuczyn Date: Mon, 16 Mar 2020 17:04:37 +0000 Subject: [PATCH 09/10] Restored previously commented out tests --- common/clients/multi-tcp-client/src/lib.rs | 169 ++++----------------- 1 file changed, 30 insertions(+), 139 deletions(-) diff --git a/common/clients/multi-tcp-client/src/lib.rs b/common/clients/multi-tcp-client/src/lib.rs index b650b773cd..43e64b11f2 100644 --- a/common/clients/multi-tcp-client/src/lib.rs +++ b/common/clients/multi-tcp-client/src/lib.rs @@ -204,142 +204,33 @@ mod tests { assert_eq!(post_kill_message.to_vec(), new_received_messages[0]); } -// #[cfg(test)] -// mod tests { -// use super::*; -// use std::str; -// use std::time; -// use tokio::prelude::*; -// -// const SERVER_MSG_LEN: usize = 16; -// const CLOSE_MESSAGE: [u8; SERVER_MSG_LEN] = [0; SERVER_MSG_LEN]; -// -// struct DummyServer { -// received_buf: Vec>, -// listener: tokio::net::TcpListener, -// } -// -// impl DummyServer { -// async fn new(address: SocketAddr) -> Self { -// DummyServer { -// received_buf: Vec::new(), -// listener: tokio::net::TcpListener::bind(address).await.unwrap(), -// } -// } -// -// fn get_received(&self) -> Vec> { -// self.received_buf.clone() -// } -// -// // this is only used in tests so slightly higher logging levels are fine -// async fn listen_until(mut self, close_message: &[u8]) -> Self { -// let (mut socket, _) = self.listener.accept().await.unwrap(); -// loop { -// let mut buf = [0u8; SERVER_MSG_LEN]; -// match socket.read(&mut buf).await { -// Ok(n) if n == 0 => { -// info!("Remote connection closed"); -// return self; -// } -// Ok(n) => { -// info!("received ({}) - {:?}", n, str::from_utf8(buf[..n].as_ref())); -// -// if buf[..n].as_ref() == close_message { -// info!("closing..."); -// socket.shutdown(std::net::Shutdown::Both).unwrap(); -// return self; -// } else { -// self.received_buf.push(buf[..n].to_vec()); -// } -// } -// Err(e) => { -// panic!("failed to read from socket; err = {:?}", e); -// } -// }; -// } -// } -// } -// -// #[test] -// fn client_reconnects_to_server_after_it_went_down() { -// 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 messages_to_send = vec![[1u8; SERVER_MSG_LEN], [2; SERVER_MSG_LEN]]; -// -// 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)); -// -// for msg in &messages_to_send { -// rt.block_on(c.send(addr, msg)).unwrap(); -// } -// -// // kill server -// rt.block_on(c.send(addr, &CLOSE_MESSAGE)).unwrap(); -// let received_messages = rt -// .block_on(finished_dummy_server_future) -// .unwrap() -// .get_received(); -// -// assert_eq!(received_messages, messages_to_send); -// -// // try to send - go into reconnection -// let post_kill_message = [3u8; SERVER_MSG_LEN]; -// -// // we are trying to send to killed server -// assert!(rt.block_on(c.send(addr, &post_kill_message)).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() { -// break; -// } -// rt.block_on( -// async move { tokio::time::delay_for(time::Duration::from_millis(50)).await }, -// ); -// } -// -// // kill the server to ensure it actually got the message -// rt.block_on(c.send(addr, &CLOSE_MESSAGE)).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]); -// } -// -// #[test] -// fn server_receives_all_sent_messages_when_up() { -// 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 messages_to_send = vec![[1u8; SERVER_MSG_LEN], [2; SERVER_MSG_LEN]]; -// -// 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)); -// -// for msg in &messages_to_send { -// rt.block_on(c.send(addr, msg)).unwrap(); -// } -// -// rt.block_on(c.send(addr, &CLOSE_MESSAGE)).unwrap(); -// -// // the server future should have already been resolved -// let received_messages = rt -// .block_on(finished_dummy_server_future) -// .unwrap() -// .get_received(); -// -// assert_eq!(received_messages, messages_to_send); -// } -// } + #[test] + fn server_receives_all_sent_messages_when_up() { + 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(reconnection_backoff, 10 * reconnection_backoff); + + 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.enter(|| Client::new(client_config)); + + for msg in &messages_to_send { + rt.block_on(c.send(addr, msg.clone(), true)).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 + .block_on(finished_dummy_server_future) + .unwrap() + .get_received(); + + assert_eq!(received_messages, messages_to_send); + } +} From 81ac2b97f431763c16ee70df0a77656ff90fc74b Mon Sep 17 00:00:00 2001 From: jstuczyn Date: Mon, 16 Mar 2020 17:04:48 +0000 Subject: [PATCH 10/10] Added a very important note : ) --- common/clients/directory-client/src/presence/mod.rs | 1 + 1 file changed, 1 insertion(+) 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]