bugfix: drop tasks to connections closed by remote (#3190)

* applied patch #3187

* applied the same concept to the verloc listener
This commit is contained in:
Jędrzej Stuczyński
2023-03-23 11:09:43 +00:00
committed by GitHub
parent a7610a7a88
commit 5e04f48500
3 changed files with 22 additions and 16 deletions
+14 -10
View File
@@ -99,17 +99,24 @@ impl ConnectionHandler {
let mut framed_conn = Framed::new(conn, EchoPacketCodec);
while !shutdown_listener.is_shutdown() {
tokio::select! {
Some(echo_packet) = framed_conn.next() => {
biased;
_ = shutdown_listener.recv() => {
trace!("ConnectionHandler: Shutdown received");
}
maybe_echo_packet = framed_conn.next() => {
// handle echo packet
let reply_packet = match echo_packet {
Ok(echo_packet) => self.handle_echo_packet(echo_packet),
Err(err) => {
error!(
"The socket connection got corrupted with error: {}. Closing the socket",
err
let reply_packet = match maybe_echo_packet {
Some(Ok(echo_packet)) => self.handle_echo_packet(echo_packet),
Some(Err(err)) => {
error!(
"The socket connection got corrupted with error: {err}. Closing the socket",
);
return;
}
None => {
error!("The socket connection got terminated by the remote!");
return;
}
};
// write back the reply (note the lack of framing)
@@ -125,9 +132,6 @@ impl ConnectionHandler {
return;
}
},
_ = shutdown_listener.recv() => {
trace!("ConnectionHandler: Shutdown received");
}
}
}
}
@@ -189,9 +189,9 @@ impl<St: Storage> ConnectionHandler<St> {
_ = shutdown.recv() => {
log::trace!("ConnectionHandler: received shutdown");
}
Some(framed_sphinx_packet) = framed_conn.next() => {
framed_sphinx_packet = framed_conn.next() => {
match framed_sphinx_packet {
Ok(framed_sphinx_packet) => {
Some(Ok(framed_sphinx_packet)) => {
// TODO: benchmark spawning tokio task with full processing vs just processing it
// synchronously under higher load in single and multi-threaded situation.
@@ -200,12 +200,13 @@ impl<St: Storage> ConnectionHandler<St> {
// that change would only slow things down
self.handle_received_packet(framed_sphinx_packet).await;
}
Err(err) => {
Some(Err(err)) => {
error!(
"The socket connection got corrupted with error: {err}. Closing the socket",
);
return;
}
None => break, // stream got closed by remote
}
}
}
@@ -85,9 +85,9 @@ impl ConnectionHandler {
_ = shutdown.recv() => {
log::trace!("ConnectionHandler: received shutdown");
}
Some(framed_sphinx_packet) = framed_conn.next() => {
framed_sphinx_packet = framed_conn.next() => {
match framed_sphinx_packet {
Ok(framed_sphinx_packet) => {
Some(Ok(framed_sphinx_packet)) => {
// TODO: benchmark spawning tokio task with full processing vs just processing it
// synchronously (without delaying inside of course,
// delay is moved to a global DelayQueue)
@@ -98,12 +98,13 @@ impl ConnectionHandler {
// that change would only slow things down
self.handle_received_packet(framed_sphinx_packet);
}
Err(err) => {
Some(Err(err)) => {
error!(
"The socket connection got corrupted with error: {err}. Closing the socket",
);
return;
}
None => break, // stream got closed by remote
}
},
}