From d02e248328bcdfc3ee8e4f6118586db3a2475f99 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?J=C4=99drzej=20Stuczy=C5=84ski?= Date: Thu, 2 Apr 2020 15:38:21 +0100 Subject: [PATCH] Bugfix/closing tcp client connections on drop (#167) * Filtering out early eof errors * Wrapping ConnectionManager in Abortable * Decreased pathchecker logging level * Client aborting all connection futures on drop * Moved AbortHandle to connection_managers HashMap to couple them closer together --- .../src/connection_manager/mod.rs | 44 ++++++++++++------- common/clients/multi-tcp-client/src/lib.rs | 31 +++++++++---- common/healthcheck/src/path_check.rs | 2 +- mixnode/src/node/listener.rs | 16 +++++-- .../src/provider/client_handling/listener.rs | 14 ++++-- .../src/provider/mix_handling/listener.rs | 16 +++++-- 6 files changed, 86 insertions(+), 37 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 8a568e6ac9..57a56bb810 100644 --- a/common/clients/multi-tcp-client/src/connection_manager/mod.rs +++ b/common/clients/multi-tcp-client/src/connection_manager/mod.rs @@ -1,6 +1,7 @@ use crate::connection_manager::reconnector::ConnectionReconnector; use crate::connection_manager::writer::ConnectionWriter; use futures::channel::{mpsc, oneshot}; +use futures::future::{abortable, AbortHandle}; use futures::task::Poll; use futures::{AsyncWriteExt, StreamExt}; use log::*; @@ -34,6 +35,12 @@ pub(crate) struct ConnectionManager<'a> { state: ConnectionState<'a>, } +impl<'a> Drop for ConnectionManager<'a> { + fn drop(&mut self) { + debug!("Connection manager to {:?} is being dropped", self.address) + } +} + impl<'a> ConnectionManager<'static> { pub(crate) async fn new( address: SocketAddr, @@ -68,24 +75,29 @@ impl<'a> ConnectionManager<'static> { } } - /// 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 - ); - } + async fn run(mut self) { + 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 + } + } + + /// consumes Self and returns channel for communication as well as an `AbortHandle` + pub(crate) fn start_abortable(self, handle: &Handle) -> (ConnectionManagerSender, AbortHandle) { + let sender_clone = self.conn_tx.clone(); + let (abort_fut, abort_handle) = abortable(self.run()); + + handle.spawn(async move { abort_fut.await }); + + (sender_clone, abort_handle) } async fn handle_new_message(&mut self, msg: Vec) -> io::Result<()> { diff --git a/common/clients/multi-tcp-client/src/lib.rs b/common/clients/multi-tcp-client/src/lib.rs index 9de04f3469..38c0b88511 100644 --- a/common/clients/multi-tcp-client/src/lib.rs +++ b/common/clients/multi-tcp-client/src/lib.rs @@ -1,5 +1,6 @@ use crate::connection_manager::{ConnectionManager, ConnectionManagerSender}; use futures::channel::oneshot; +use futures::future::AbortHandle; use log::*; use std::collections::HashMap; use std::io; @@ -28,7 +29,7 @@ impl Config { pub struct Client { runtime_handle: Handle, - connections_managers: HashMap, + connections_managers: HashMap, maximum_reconnection_backoff: Duration, initial_reconnection_backoff: Duration, } @@ -46,14 +47,19 @@ impl Client { } } - async fn start_new_connection_manager(&self, address: SocketAddr) -> ConnectionManagerSender { - ConnectionManager::new( + async fn start_new_connection_manager( + &mut self, + address: SocketAddr, + ) -> (ConnectionManagerSender, AbortHandle) { + let (sender, abort_handle) = ConnectionManager::new( address, self.initial_reconnection_backoff, self.maximum_reconnection_backoff, ) .await - .start(&self.runtime_handle) + .start_abortable(&self.runtime_handle); + + (sender, abort_handle) } // if wait_for_response is set to true, we will get information about any possible IO errors @@ -71,24 +77,33 @@ impl Client { address ); - let new_manager_sender = self.start_new_connection_manager(address).await; + let (new_manager_sender, abort_handle) = + self.start_new_connection_manager(address).await; self.connections_managers - .insert(address, new_manager_sender); + .insert(address, (new_manager_sender, abort_handle)); } 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(); + manager.0.unbounded_send((message, Some(res_tx))).unwrap(); res_rx.await.unwrap() } else { - manager.unbounded_send((message, None)).unwrap(); + manager.0.unbounded_send((message, None)).unwrap(); Ok(()) } } } +impl Drop for Client { + fn drop(&mut self) { + for (_, abort_handle) in self.connections_managers.values() { + abort_handle.abort() + } + } +} + #[cfg(test)] mod tests { use super::*; diff --git a/common/healthcheck/src/path_check.rs b/common/healthcheck/src/path_check.rs index 4e0c616595..f565e12e7f 100644 --- a/common/healthcheck/src/path_check.rs +++ b/common/healthcheck/src/path_check.rs @@ -183,7 +183,7 @@ impl PathChecker { return; } - debug!("Checking path: {:?} ({})", path, iteration); + trace!("Checking path: {:?} ({})", path, iteration); let path_identifier = PathChecker::unique_path_key(path, self.check_id, iteration); // check if there is even any point in sending the packet diff --git a/mixnode/src/node/listener.rs b/mixnode/src/node/listener.rs index 05f7b554bb..76d6e13b2c 100644 --- a/mixnode/src/node/listener.rs +++ b/mixnode/src/node/listener.rs @@ -48,6 +48,8 @@ async fn process_socket_connection( return; } Ok(n) => { + // If I understand it correctly, this if should never be executed as if `read_exact` + // does not fill buffer, it will throw UnexpectedEof? if n != sphinx::PACKET_SIZE { warn!("read data of different length than expected sphinx packet size - {} (expected {})", n, sphinx::PACKET_SIZE); continue; @@ -63,10 +65,16 @@ async fn process_socket_connection( )) } Err(e) => { - warn!( - "failed to read from socket. Closing the connection; err = {:?}", - e - ); + if e.kind() == io::ErrorKind::UnexpectedEof { + debug!("Read buffer was not fully filled. Most likely the client ({:?}) closed the connection.\ + Also closing the connection on this end.", socket.peer_addr()) + } else { + warn!( + "failed to read from socket (source: {:?}). Closing the connection; err = {:?}", + socket.peer_addr(), + e + ); + } return; } }; diff --git a/sfw-provider/src/provider/client_handling/listener.rs b/sfw-provider/src/provider/client_handling/listener.rs index 260c947b07..1508efa232 100644 --- a/sfw-provider/src/provider/client_handling/listener.rs +++ b/sfw-provider/src/provider/client_handling/listener.rs @@ -65,10 +65,16 @@ async fn process_socket_connection( // plus realistically it wouldn't really introduce any speed up Ok(n) => process_request(&mut socket, &buf[0..n], &mut request_processor).await, Err(e) => { - warn!( - "failed to read from socket. Closing the connection; err = {:?}", - e - ); + if e.kind() == io::ErrorKind::UnexpectedEof { + debug!("Read buffer was not fully filled. Most likely the client ({:?}) closed the connection.\ + Also closing the connection on this end.", socket.peer_addr()) + } else { + warn!( + "failed to read from socket (source: {:?}). Closing the connection; err = {:?}", + socket.peer_addr(), + e + ); + } return; } }; diff --git a/sfw-provider/src/provider/mix_handling/listener.rs b/sfw-provider/src/provider/mix_handling/listener.rs index 8bcc13fe7e..f93eba48d9 100644 --- a/sfw-provider/src/provider/mix_handling/listener.rs +++ b/sfw-provider/src/provider/mix_handling/listener.rs @@ -36,6 +36,8 @@ async fn process_socket_connection( return; } Ok(n) => { + // If I understand it correctly, this if should never be executed as if `read_exact` + // does not fill buffer, it will throw UnexpectedEof? if n != sphinx::PACKET_SIZE { warn!("read data of different length than expected sphinx packet size - {} (expected {})", n, sphinx::PACKET_SIZE); continue; @@ -45,10 +47,16 @@ async fn process_socket_connection( tokio::spawn(process_received_packet(buf, packet_processor.clone())) } Err(e) => { - warn!( - "failed to read from socket. Closing the connection; err = {:?}", - e - ); + if e.kind() == io::ErrorKind::UnexpectedEof { + debug!("Read buffer was not fully filled. Most likely the client ({:?}) closed the connection.\ + Also closing the connection on this end.", socket.peer_addr()) + } else { + warn!( + "failed to read from socket (source: {:?}). Closing the connection; err = {:?}", + socket.peer_addr(), + e + ); + } return; } };