From c720481a45d0c08c012aa864c90f3cd0f667efae Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jon=20H=C3=A4ggblad?= Date: Tue, 13 Dec 2022 11:39:02 +0100 Subject: [PATCH] socks5: rework waiting in inbound.rs (#1880) * socks5: rework waiting in inbound.rs * changelog: add note * out queue control: reduce old set size to 4 --- CHANGELOG.md | 1 + .../transmission_buffer.rs | 2 +- .../proxy-helpers/src/proxy_runner/inbound.rs | 45 +++++++++++-------- 3 files changed, 29 insertions(+), 19 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index e6cf4c0a5a..5b03ddd8ba 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -15,6 +15,7 @@ Post 1.0.0 release, the changelog format is based on [Keep a Changelog](https:// - network-requester: fix bug where websocket connection disconnect resulted in success error code - clients: fix a few panics handling the gateway-client - mixnode, gateway, validator-api: Use mainnet values as defaults for URLs and mixnet contract ([#1884]) +- socks5: fixed bug where connections sometimes where closed too early [#1872]: https://github.com/nymtech/nym/pull/1872 [#1884]: https://github.com/nymtech/nym/pull/1884 diff --git a/clients/client-core/src/client/real_messages_control/real_traffic_stream/transmission_buffer.rs b/clients/client-core/src/client/real_messages_control/real_traffic_stream/transmission_buffer.rs index 20c7ce6cb7..1f16f472c1 100644 --- a/clients/client-core/src/client/real_messages_control/real_traffic_stream/transmission_buffer.rs +++ b/clients/client-core/src/client/real_messages_control/real_traffic_stream/transmission_buffer.rs @@ -17,7 +17,7 @@ use wasm_timer; use super::{get_time_now, RealMessage}; // The number of lanes included in the oldest set. Used when we need to prioritize traffic. -const OLDEST_LANE_SET_SIZE: usize = 5; +const OLDEST_LANE_SET_SIZE: usize = 4; // As a way of prune connections we also check for timeouts. const MSG_CONSIDERED_STALE_AFTER_SECS: u64 = 10 * 60; diff --git a/common/socks5/proxy-helpers/src/proxy_runner/inbound.rs b/common/socks5/proxy-helpers/src/proxy_runner/inbound.rs index f2ebc2cd9b..37403c33ad 100644 --- a/common/socks5/proxy-helpers/src/proxy_runner/inbound.rs +++ b/common/socks5/proxy-helpers/src/proxy_runner/inbound.rs @@ -77,16 +77,11 @@ where ordered_msg.len() ); - // If we are closing the socket, wait until the data has passed `OutQueueControl` and the lane - // is empty, otherwise just wait until we are reasonably close to finish sending as a way to - // throttle the incoming data. - if let Some(lane_queue_lengths) = lane_queue_lengths { - if is_finished { - wait_until_lane_empty(lane_queue_lengths, connection_id).await; - } else { - // We allow a bit of slack when this is not the last msg - wait_until_lane_almost_empty(lane_queue_lengths, connection_id).await; - } + // Before sending the data downstream, wait for the lane at the `OutQueueControl` is reasonably + // close to finishing. This is a way of pacing the sending application (backpressure). + if let Some(ref lane_queue_lengths) = lane_queue_lengths { + // We allow a bit of slack to try to keep the pipeline >0 + wait_until_lane_almost_empty(lane_queue_lengths, connection_id).await; } mix_sender @@ -95,16 +90,30 @@ where .expect("InputMessageReceiver has stopped receiving!"); if is_finished { - // technically we already informed it when we sent the message to mixnet above - debug!(target: &*format!("({}) socks5 inbound", connection_id), "The local socket is closed - won't receive any more data. Informing remote about that..."); + // After sending, if this is the last message, wait until we've actually transmitted the data + // in the `OutQueueControl` and the lane is empty. + if let Some(ref lane_queue_lengths) = lane_queue_lengths { + // This is basically an ugly workaround to make sure that we don't start waiting until + // the data that we pushed arrived at the OutQueueControl. + // This usually not a problem in the socks5-client, but for the network-requester this + // info is synced at up to every 500ms. + sleep(Duration::from_secs(2)).await; + wait_until_lane_empty(lane_queue_lengths, connection_id).await; + } + + // Technically we already informed it when we sent the message to mixnet above + debug!( + target: &*format!("({}) socks5 inbound", connection_id), + "The local socket is closed - won't receive any more data. Informing remote about that..." + ); } is_finished } -async fn wait_until_lane_empty(lane_queue_lengths: LaneQueueLengths, connection_id: u64) { +async fn wait_until_lane_empty(lane_queue_lengths: &LaneQueueLengths, connection_id: u64) { if tokio::time::timeout( - Duration::from_secs(2 * 60), + Duration::from_secs(4 * 60), wait_for_lane( lane_queue_lengths, connection_id, @@ -119,13 +128,13 @@ async fn wait_until_lane_empty(lane_queue_lengths: LaneQueueLengths, connection_ } } -async fn wait_until_lane_almost_empty(lane_queue_lengths: LaneQueueLengths, connection_id: u64) { +async fn wait_until_lane_almost_empty(lane_queue_lengths: &LaneQueueLengths, connection_id: u64) { if tokio::time::timeout( - Duration::from_secs(2 * 60), + Duration::from_secs(4 * 60), wait_for_lane( lane_queue_lengths, connection_id, - 10, + 30, Duration::from_millis(100), ), ) @@ -137,7 +146,7 @@ async fn wait_until_lane_almost_empty(lane_queue_lengths: LaneQueueLengths, conn } async fn wait_for_lane( - lane_queue_lengths: LaneQueueLengths, + lane_queue_lengths: &LaneQueueLengths, connection_id: u64, queue_length_threshold: usize, sleep_duration: Duration,