From 790220039bbdb4eea63c84b2fac140f3ce2f8270 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?J=C4=99drzej=20Stuczy=C5=84ski?= Date: Fri, 24 Mar 2023 18:10:46 +0000 Subject: [PATCH 1/6] added read deadline to AvailableReader --- .../proxy-helpers/src/available_reader.rs | 106 +++++++++++++----- 1 file changed, 76 insertions(+), 30 deletions(-) diff --git a/common/socks5/proxy-helpers/src/available_reader.rs b/common/socks5/proxy-helpers/src/available_reader.rs index 70d968f968..018486e68b 100644 --- a/common/socks5/proxy-helpers/src/available_reader.rs +++ b/common/socks5/proxy-helpers/src/available_reader.rs @@ -1,8 +1,9 @@ -// Copyright 2021 - Nym Technologies SA +// Copyright 2021-2023 - Nym Technologies SA // SPDX-License-Identifier: Apache-2.0 use bytes::{BufMut, Bytes, BytesMut}; use futures::Stream; +use log::error; use std::cell::RefCell; use std::future::Future; use std::io; @@ -15,6 +16,7 @@ use tokio_util::io::poll_read_buf; const MAX_READ_AMOUNT: usize = 500 * 1000; // 0.5MB const GRACE_DURATION: Duration = Duration::from_millis(1); +const READ_TIMEOUT: Duration = Duration::from_millis(10); pub struct AvailableReader<'a, R: AsyncRead + Unpin> { // TODO: come up with a way to avoid using RefCell (not sure if possible though due to having to @@ -22,6 +24,7 @@ pub struct AvailableReader<'a, R: AsyncRead + Unpin> { buf: RefCell, inner: RefCell<&'a mut R>, grace_period: Option>>, + read_deadline: Option>>, } impl<'a, R> AvailableReader<'a, R> @@ -31,10 +34,58 @@ where const BUF_INCREMENT: usize = 4096; pub fn new(reader: &'a mut R) -> Self { + // pub fn new(reader: &'a mut R, buffer_size: usize) -> Self { AvailableReader { buf: RefCell::new(BytesMut::with_capacity(Self::BUF_INCREMENT)), inner: RefCell::new(reader), - grace_period: Some(Box::pin(sleep(GRACE_DURATION))), + grace_period: None, + read_deadline: None, + } + } + + fn return_buf(mut self: Pin<&mut Self>) -> Poll::Item>> { + // reset timeouts so the poll wouldn't be accidentally called + self.grace_period = None; + self.read_deadline = None; + + let buf = self.buf.borrow_mut().split(); + // there's no data in the buffer, it means the underlying source is done + if buf.is_empty() { + Poll::Ready(None) + } else { + Poll::Ready(Some(Ok(buf.freeze()))) + } + } + + fn poll_read_deadline(&mut self, cx: &mut Context<'_>) -> Poll<()> { + match self.read_deadline.as_mut() { + Some(read_deadline) => Pin::new(read_deadline).poll(cx), + None => { + // this is the first time we're polling this reader - set the deadline + self.read_deadline = Some(Box::pin(sleep(READ_TIMEOUT))); + Poll::Pending + } + } + } + + fn poll_grace_period(&mut self, cx: &mut Context<'_>) -> Poll<()> { + match self.grace_period.as_mut() { + Some(grace_period) => Pin::new(grace_period).poll(cx), + None => { + // this is the first time we're polling this reader - set the grace period + self.grace_period = Some(Box::pin(sleep(GRACE_DURATION))); + Poll::Pending + } + } + } + + fn reset_grace_period(&mut self) { + if let Some(grace_period) = self.grace_period.as_mut() { + let now = Instant::now(); + grace_period.as_mut().reset(now + GRACE_DURATION); + } else { + // this branch should be impossible! + error!("reached an impossible branch - attempted to reset a non-existent grace period timeout") } } } @@ -55,50 +106,46 @@ impl<'a, R: AsyncRead + Unpin> Stream for AvailableReader<'a, R> { self.buf.borrow_mut().deref_mut(), ); + let deadline_poll_res = self.poll_read_deadline(cx); + let grace_period_poll_res = self.poll_grace_period(cx); + match poll_res { Poll::Pending => { - // there's nothing for us here, just return whatever we have (assuming we read anything!) + // there's nothing for us here, just return whatever we have + // (assuming we read anything and we SHOULD HAVE read something) if self.buf.borrow().is_empty() { + // this case shouldn't be possible - if something woke the future we MUST HAVE HAD some data + // (it's because we never set the sleeps until the first call, meaning some data was already there) + error!("AvailableReader got woken with not data to return!"); + Poll::Pending + } else if grace_period_poll_res.is_pending() && deadline_poll_res.is_pending() { + // if neither timeout was reached yet, wait a bit more Poll::Pending } else { - // if exists - check grace period - if let Some(grace_period) = self.grace_period.as_mut() { - if Pin::new(grace_period).poll(cx).is_pending() { - return Poll::Pending; - } - } - - let buf = self.buf.replace(BytesMut::new()); - Poll::Ready(Some(Ok(buf.freeze()))) + // otherwise return what we managed to read (and reset timeouts) + self.return_buf() } } Poll::Ready(Err(err)) => Poll::Ready(Some(Err(err))), Poll::Ready(Ok(n)) => { - // if exists - reset grace period - if let Some(grace_period) = self.grace_period.as_mut() { - let now = Instant::now(); - grace_period.as_mut().reset(now + GRACE_DURATION); - } + self.reset_grace_period(); - // if we read a non-0 amount, we're not done yet! if n == 0 { - let buf = self.buf.replace(BytesMut::new()); - if !buf.is_empty() { - Poll::Ready(Some(Ok(buf.freeze()))) - } else { - Poll::Ready(None) - } + // if we read 0 bytes, it means the underlying source is done, so return what we have + // (note, this might get called twice if we already had something in the buffer) + self.return_buf() } else { // tell the waker we should be polled again! cx.waker().wake_by_ref(); - // if we reached our maximum amount - return it + // if we reached our maximum amount or we've been trying to read the data for too long + // return what we have let read_bytes_len = self.buf.borrow().len(); - if read_bytes_len >= MAX_READ_AMOUNT { - let buf = self.buf.replace(BytesMut::new()); - return Poll::Ready(Some(Ok(buf.freeze()))); + if read_bytes_len >= MAX_READ_AMOUNT || deadline_poll_res.is_ready() { + self.return_buf() + } else { + Poll::Pending } - Poll::Pending } } } @@ -154,7 +201,6 @@ mod tests { assert_eq!(read_data, first_data_chunk); assert_pending!(poll!(available_reader.next())); - // before dropping the mock, we need to empty it let mut buf = vec![0u8; second_data_chunk.len()]; assert_eq!(reader_mock.read(&mut buf).await.unwrap(), 100); From 47f5b4ceac61cd0c90bfc0e933613c6105bbf3a6 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?J=C4=99drzej=20Stuczy=C5=84ski?= Date: Mon, 27 Mar 2023 10:50:16 +0100 Subject: [PATCH 2/6] limit the maximum buffer size of AvailableReader by PacketSize of our mix packets --- common/socks5-client-core/src/lib.rs | 10 ++++++++++ common/socks5-client-core/src/socks/client.rs | 7 +++++++ .../proxy-helpers/src/available_reader.rs | 17 +++++++++-------- .../proxy-helpers/src/proxy_runner/inbound.rs | 5 ++++- .../proxy-helpers/src/proxy_runner/mod.rs | 7 ++++++- service-providers/network-requester/src/core.rs | 16 ++++++++++++++++ .../network-requester/src/socks5/tcp.rs | 5 +++++ 7 files changed, 57 insertions(+), 10 deletions(-) diff --git a/common/socks5-client-core/src/lib.rs b/common/socks5-client-core/src/lib.rs index daea560837..aa85a8a00f 100644 --- a/common/socks5-client-core/src/lib.rs +++ b/common/socks5-client-core/src/lib.rs @@ -23,6 +23,7 @@ use nym_client_core::client::key_manager::KeyManager; use nym_client_core::config::persistence::key_pathfinder::ClientKeyPathfinder; use nym_credential_storage::storage::Storage; use nym_sphinx::addressing::clients::Recipient; +use nym_sphinx::params::PacketSize; use nym_task::{TaskClient, TaskManager}; use nym_validator_client::nyxd::QueryNyxdClient; use nym_validator_client::Client; @@ -126,6 +127,14 @@ impl NymClient { .. } = client_status; + // FIXME: use correct value from the config once https://github.com/nymtech/nym/pull/3217 + // is merged + let packet_size = config + .get_base() + .get_use_extended_packet_size() + .map(Into::into) + .unwrap_or(PacketSize::RegularPacket); + let authenticator = Authenticator::new(auth_methods, allowed_users); let mut sphinx_socks = SphinxSocksServer::new( socks5_config.get_listening_port(), @@ -134,6 +143,7 @@ impl NymClient { self_address, shared_lane_queue_lengths, socks::client::Config::new( + packet_size, socks5_config.get_provider_interface_version(), socks5_config.get_socks5_protocol_version(), socks5_config.get_send_anonymously(), diff --git a/common/socks5-client-core/src/socks/client.rs b/common/socks5-client-core/src/socks/client.rs index e6f19c62d4..d1d4b10697 100644 --- a/common/socks5-client-core/src/socks/client.rs +++ b/common/socks5-client-core/src/socks/client.rs @@ -17,6 +17,7 @@ use nym_socks5_requests::{ ConnectionId, RemoteAddress, Socks5ProtocolVersion, Socks5ProviderRequest, Socks5Request, }; use nym_sphinx::addressing::clients::Recipient; +use nym_sphinx::params::PacketSize; use nym_task::connections::{LaneQueueLengths, TransmissionLane}; use nym_task::TaskClient; use pin_project::pin_project; @@ -131,6 +132,7 @@ impl AsyncWrite for StreamState { #[derive(Debug, Copy, Clone)] pub(crate) struct Config { + biggest_packet_size: PacketSize, provider_interface_version: ProviderInterfaceVersion, socks5_protocol_version: Socks5ProtocolVersion, use_surbs_for_responses: bool, @@ -140,6 +142,7 @@ pub(crate) struct Config { impl Config { pub(crate) fn new( + biggest_packet_size: PacketSize, provider_interface_version: ProviderInterfaceVersion, socks5_protocol_version: Socks5ProtocolVersion, use_surbs_for_responses: bool, @@ -147,6 +150,7 @@ impl Config { per_request_surbs: u32, ) -> Self { Self { + biggest_packet_size, provider_interface_version, socks5_protocol_version, use_surbs_for_responses, @@ -410,6 +414,9 @@ impl SocksClient { remote_proxy_target, conn_receiver, input_sender, + // FIXME: this does NOT include overhead due to acks or chunking + // (so actual true plaintext is smaller) + self.config.biggest_packet_size.plaintext_size(), connection_id, Some(self.lane_queue_lengths.clone()), self.shutdown_listener.clone(), diff --git a/common/socks5/proxy-helpers/src/available_reader.rs b/common/socks5/proxy-helpers/src/available_reader.rs index 018486e68b..1f169678ac 100644 --- a/common/socks5/proxy-helpers/src/available_reader.rs +++ b/common/socks5/proxy-helpers/src/available_reader.rs @@ -14,7 +14,7 @@ use tokio::io::AsyncRead; use tokio::time::{sleep, Duration, Instant, Sleep}; use tokio_util::io::poll_read_buf; -const MAX_READ_AMOUNT: usize = 500 * 1000; // 0.5MB +const DEFAULT_MAX_READ_AMOUNT: usize = 128 * 1024; // 128kB const GRACE_DURATION: Duration = Duration::from_millis(1); const READ_TIMEOUT: Duration = Duration::from_millis(10); @@ -25,6 +25,7 @@ pub struct AvailableReader<'a, R: AsyncRead + Unpin> { inner: RefCell<&'a mut R>, grace_period: Option>>, read_deadline: Option>>, + max_read: usize, } impl<'a, R> AvailableReader<'a, R> @@ -33,11 +34,11 @@ where { const BUF_INCREMENT: usize = 4096; - pub fn new(reader: &'a mut R) -> Self { - // pub fn new(reader: &'a mut R, buffer_size: usize) -> Self { + pub fn new(reader: &'a mut R, max_read: Option) -> Self { AvailableReader { buf: RefCell::new(BytesMut::with_capacity(Self::BUF_INCREMENT)), inner: RefCell::new(reader), + max_read: max_read.unwrap_or(DEFAULT_MAX_READ_AMOUNT), grace_period: None, read_deadline: None, } @@ -141,7 +142,7 @@ impl<'a, R: AsyncRead + Unpin> Stream for AvailableReader<'a, R> { // if we reached our maximum amount or we've been trying to read the data for too long // return what we have let read_bytes_len = self.buf.borrow().len(); - if read_bytes_len >= MAX_READ_AMOUNT || deadline_poll_res.is_ready() { + if read_bytes_len >= self.max_read || deadline_poll_res.is_ready() { self.return_buf() } else { Poll::Pending @@ -166,7 +167,7 @@ mod tests { let data = vec![42u8; 100]; let mut reader = Cursor::new(data.clone()); - let mut available_reader = AvailableReader::new(&mut reader); + let mut available_reader = AvailableReader::new(&mut reader, None); let read_data = available_reader.next().await.unwrap().unwrap(); assert_eq!(read_data, data); @@ -178,7 +179,7 @@ mod tests { let data = vec![42u8; AvailableReader::>>::BUF_INCREMENT + 100]; let mut reader = Cursor::new(data.clone()); - let mut available_reader = AvailableReader::new(&mut reader); + let mut available_reader = AvailableReader::new(&mut reader, None); let read_data = available_reader.next().await.unwrap().unwrap(); assert_eq!(read_data, data); @@ -196,7 +197,7 @@ mod tests { .read(&second_data_chunk) .build(); - let mut available_reader = AvailableReader::new(&mut reader_mock); + let mut available_reader = AvailableReader::new(&mut reader_mock, None); let read_data = available_reader.next().await.unwrap().unwrap(); assert_eq!(read_data, first_data_chunk); @@ -215,7 +216,7 @@ mod tests { .read(&data) .build(); - let mut available_reader = AvailableReader::new(&mut reader_mock); + let mut available_reader = AvailableReader::new(&mut reader_mock, None); let read_data = available_reader.next().await.unwrap().unwrap(); assert_eq!(read_data, data); diff --git a/common/socks5/proxy-helpers/src/proxy_runner/inbound.rs b/common/socks5/proxy-helpers/src/proxy_runner/inbound.rs index 548d1a0971..584ace090f 100644 --- a/common/socks5/proxy-helpers/src/proxy_runner/inbound.rs +++ b/common/socks5/proxy-helpers/src/proxy_runner/inbound.rs @@ -167,6 +167,7 @@ pub(super) async fn run_inbound( remote_source_address: String, connection_id: ConnectionId, mix_sender: MixProxySender, + available_plaintext_per_mix_packet: usize, adapter_fn: F, shutdown_notify: Arc, lane_queue_lengths: Option, @@ -176,7 +177,9 @@ where F: Fn(ConnectionId, Vec, bool) -> S + Send + 'static, S: Debug, { - let mut available_reader = AvailableReader::new(&mut reader); + // TODO: this multiplication by 4 is completely arbitrary here + let mut available_reader = + AvailableReader::new(&mut reader, Some(available_plaintext_per_mix_packet * 4)); let mut message_sender = OrderedMessageSender::new(); let shutdown_future = shutdown_notify.notified().then(|_| sleep(SHUTDOWN_TIMEOUT)); diff --git a/common/socks5/proxy-helpers/src/proxy_runner/mod.rs b/common/socks5/proxy-helpers/src/proxy_runner/mod.rs index c9d925ad4f..c8978141bf 100644 --- a/common/socks5/proxy-helpers/src/proxy_runner/mod.rs +++ b/common/socks5/proxy-helpers/src/proxy_runner/mod.rs @@ -1,4 +1,4 @@ -// Copyright 2021 - Nym Technologies SA +// Copyright 2021-2023 - Nym Technologies SA // SPDX-License-Identifier: Apache-2.0 use crate::connection_controller::ConnectionReceiver; @@ -49,6 +49,8 @@ pub struct ProxyRunner { connection_id: ConnectionId, lane_queue_lengths: Option, + available_plaintext_per_mix_packet: usize, + // Listens to shutdown commands from higher up shutdown_listener: TaskClient, } @@ -64,6 +66,7 @@ where remote_source_address: String, mix_receiver: ConnectionReceiver, mix_sender: MixProxySender, + available_plaintext_per_mix_packet: usize, connection_id: ConnectionId, lane_queue_lengths: Option, shutdown_listener: TaskClient, @@ -76,6 +79,7 @@ where remote_source_address, connection_id, lane_queue_lengths, + available_plaintext_per_mix_packet, shutdown_listener, } } @@ -96,6 +100,7 @@ where self.remote_source_address.clone(), self.connection_id, self.mix_sender.clone(), + self.available_plaintext_per_mix_packet, adapter_fn, Arc::clone(&shutdown_notify), self.lane_queue_lengths.clone(), diff --git a/service-providers/network-requester/src/core.rs b/service-providers/network-requester/src/core.rs index d8665a1306..39060e9e55 100644 --- a/service-providers/network-requester/src/core.rs +++ b/service-providers/network-requester/src/core.rs @@ -29,6 +29,7 @@ use nym_socks5_requests::{ }; use nym_sphinx::addressing::clients::Recipient; use nym_sphinx::anonymous_replies::requests::AnonymousSenderTag; +use nym_sphinx::params::PacketSize; use nym_statistics_common::collector::StatisticsSender; use nym_task::connections::LaneQueueLengths; use nym_task::{TaskClient, TaskManager}; @@ -57,6 +58,8 @@ pub struct NRServiceProviderBuilder { } struct NRServiceProvider { + config: Config, + outbound_request_filter: OutboundRequestFilter, open_proxy: bool, mixnet_client: nym_sdk::mixnet::MixnetClient, @@ -242,6 +245,7 @@ impl NRServiceProviderBuilder { start_allowed_list_reloader(self.allowed_hosts, shutdown.subscribe()).await; let service_provider = NRServiceProvider { + config: self.config, outbound_request_filter: self.outbound_request_filter, open_proxy: self.open_proxy, mixnet_client, @@ -329,6 +333,7 @@ impl NRServiceProvider { connection_id: ConnectionId, remote_addr: String, return_address: reply::MixnetAddress, + biggest_packet_size: PacketSize, controller_sender: ControllerSender, mix_input_sender: MixProxySender, lane_queue_lengths: LaneQueueLengths, @@ -385,6 +390,7 @@ impl NRServiceProvider { // run the proxy on the connection conn.run_proxy( remote_version, + biggest_packet_size, mix_receiver, mix_input_sender, lane_queue_lengths, @@ -437,6 +443,15 @@ impl NRServiceProvider { return; } + // FIXME: use correct value from the config once https://github.com/nymtech/nym/pull/3217 + // is merged + let packet_size = self + .config + .get_base() + .get_use_extended_packet_size() + .map(Into::into) + .unwrap_or(PacketSize::RegularPacket); + let controller_sender_clone = self.controller_sender.clone(); let mix_input_sender_clone = self.mix_input_sender.clone(); let lane_queue_lengths_clone = self.mixnet_client.shared_lane_queue_lengths(); @@ -449,6 +464,7 @@ impl NRServiceProvider { conn_id, remote_addr, return_address, + packet_size, controller_sender_clone, mix_input_sender_clone, lane_queue_lengths_clone, diff --git a/service-providers/network-requester/src/socks5/tcp.rs b/service-providers/network-requester/src/socks5/tcp.rs index 18281b0fb7..2a50bbf3a3 100644 --- a/service-providers/network-requester/src/socks5/tcp.rs +++ b/service-providers/network-requester/src/socks5/tcp.rs @@ -7,6 +7,7 @@ use nym_service_providers_common::interface::RequestVersion; use nym_socks5_proxy_helpers::connection_controller::ConnectionReceiver; use nym_socks5_proxy_helpers::proxy_runner::{MixProxySender, ProxyRunner}; use nym_socks5_requests::{ConnectionId, RemoteAddress, Socks5Request}; +use nym_sphinx::params::PacketSize; use nym_task::connections::LaneQueueLengths; use nym_task::TaskClient; use std::io; @@ -42,6 +43,7 @@ impl Connection { pub(crate) async fn run_proxy( &mut self, remote_version: RequestVersion, + biggest_packet_size: PacketSize, mix_receiver: ConnectionReceiver, mix_sender: MixProxySender, lane_queue_lengths: LaneQueueLengths, @@ -57,6 +59,9 @@ impl Connection { remote_source_address, mix_receiver, mix_sender, + // FIXME: this does NOT include overhead due to acks or chunking + // (so actual true plaintext is smaller) + biggest_packet_size.plaintext_size(), connection_id, Some(lane_queue_lengths), shutdown, From 5d7be89edb248615a92116f53e0623dfdf9a9a35 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?J=C4=99drzej=20Stuczy=C5=84ski?= Date: Tue, 28 Mar 2023 10:11:56 +0100 Subject: [PATCH 3/6] replaced inner implementation with tokio's 'ReaderStream' --- .../proxy-helpers/src/available_reader.rs | 245 +----------------- .../proxy-helpers/src/proxy_runner/inbound.rs | 2 + 2 files changed, 15 insertions(+), 232 deletions(-) diff --git a/common/socks5/proxy-helpers/src/available_reader.rs b/common/socks5/proxy-helpers/src/available_reader.rs index 1f169678ac..e8d4c4205a 100644 --- a/common/socks5/proxy-helpers/src/available_reader.rs +++ b/common/socks5/proxy-helpers/src/available_reader.rs @@ -1,255 +1,36 @@ // Copyright 2021-2023 - Nym Technologies SA // SPDX-License-Identifier: Apache-2.0 -use bytes::{BufMut, Bytes, BytesMut}; +use bytes::Bytes; use futures::Stream; -use log::error; -use std::cell::RefCell; -use std::future::Future; use std::io; -use std::ops::DerefMut; use std::pin::Pin; use std::task::{Context, Poll}; use tokio::io::AsyncRead; -use tokio::time::{sleep, Duration, Instant, Sleep}; -use tokio_util::io::poll_read_buf; -const DEFAULT_MAX_READ_AMOUNT: usize = 128 * 1024; // 128kB -const GRACE_DURATION: Duration = Duration::from_millis(1); -const READ_TIMEOUT: Duration = Duration::from_millis(10); +// note, min_capacity doesn't mean we're going to always read at least this amount of data, +// it defines the smallest allowed (by yours truly) upper bound +const MIN_CAPACITY: usize = 16 * 1024; +const DEFAULT_CAPACITY: usize = 64 * 1024; -pub struct AvailableReader<'a, R: AsyncRead + Unpin> { - // TODO: come up with a way to avoid using RefCell (not sure if possible though due to having to - // mutably borrow both inner reader and buffer at the same time) - buf: RefCell, - inner: RefCell<&'a mut R>, - grace_period: Option>>, - read_deadline: Option>>, - max_read: usize, +pub struct AvailableReader { + inner: tokio_util::io::ReaderStream, } -impl<'a, R> AvailableReader<'a, R> -where - R: AsyncRead + Unpin, -{ - const BUF_INCREMENT: usize = 4096; +impl AvailableReader { + pub fn new(reader: R, capacity: Option) -> Self { + let capacity = capacity.unwrap_or(DEFAULT_CAPACITY).max(MIN_CAPACITY); - pub fn new(reader: &'a mut R, max_read: Option) -> Self { AvailableReader { - buf: RefCell::new(BytesMut::with_capacity(Self::BUF_INCREMENT)), - inner: RefCell::new(reader), - max_read: max_read.unwrap_or(DEFAULT_MAX_READ_AMOUNT), - grace_period: None, - read_deadline: None, - } - } - - fn return_buf(mut self: Pin<&mut Self>) -> Poll::Item>> { - // reset timeouts so the poll wouldn't be accidentally called - self.grace_period = None; - self.read_deadline = None; - - let buf = self.buf.borrow_mut().split(); - // there's no data in the buffer, it means the underlying source is done - if buf.is_empty() { - Poll::Ready(None) - } else { - Poll::Ready(Some(Ok(buf.freeze()))) - } - } - - fn poll_read_deadline(&mut self, cx: &mut Context<'_>) -> Poll<()> { - match self.read_deadline.as_mut() { - Some(read_deadline) => Pin::new(read_deadline).poll(cx), - None => { - // this is the first time we're polling this reader - set the deadline - self.read_deadline = Some(Box::pin(sleep(READ_TIMEOUT))); - Poll::Pending - } - } - } - - fn poll_grace_period(&mut self, cx: &mut Context<'_>) -> Poll<()> { - match self.grace_period.as_mut() { - Some(grace_period) => Pin::new(grace_period).poll(cx), - None => { - // this is the first time we're polling this reader - set the grace period - self.grace_period = Some(Box::pin(sleep(GRACE_DURATION))); - Poll::Pending - } - } - } - - fn reset_grace_period(&mut self) { - if let Some(grace_period) = self.grace_period.as_mut() { - let now = Instant::now(); - grace_period.as_mut().reset(now + GRACE_DURATION); - } else { - // this branch should be impossible! - error!("reached an impossible branch - attempted to reset a non-existent grace period timeout") + inner: tokio_util::io::ReaderStream::with_capacity(reader, capacity), } } } -impl<'a, R: AsyncRead + Unpin> Stream for AvailableReader<'a, R> { +impl Stream for AvailableReader { type Item = io::Result; fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { - // if we have no space in buffer left - expand it - if !self.buf.borrow().has_remaining_mut() { - self.buf.borrow_mut().reserve(Self::BUF_INCREMENT); - } - - // note: poll_read_buf calls `buf.advance_mut(n)` - let poll_res = poll_read_buf( - Pin::new(self.inner.borrow_mut().deref_mut()), - cx, - self.buf.borrow_mut().deref_mut(), - ); - - let deadline_poll_res = self.poll_read_deadline(cx); - let grace_period_poll_res = self.poll_grace_period(cx); - - match poll_res { - Poll::Pending => { - // there's nothing for us here, just return whatever we have - // (assuming we read anything and we SHOULD HAVE read something) - if self.buf.borrow().is_empty() { - // this case shouldn't be possible - if something woke the future we MUST HAVE HAD some data - // (it's because we never set the sleeps until the first call, meaning some data was already there) - error!("AvailableReader got woken with not data to return!"); - Poll::Pending - } else if grace_period_poll_res.is_pending() && deadline_poll_res.is_pending() { - // if neither timeout was reached yet, wait a bit more - Poll::Pending - } else { - // otherwise return what we managed to read (and reset timeouts) - self.return_buf() - } - } - Poll::Ready(Err(err)) => Poll::Ready(Some(Err(err))), - Poll::Ready(Ok(n)) => { - self.reset_grace_period(); - - if n == 0 { - // if we read 0 bytes, it means the underlying source is done, so return what we have - // (note, this might get called twice if we already had something in the buffer) - self.return_buf() - } else { - // tell the waker we should be polled again! - cx.waker().wake_by_ref(); - - // if we reached our maximum amount or we've been trying to read the data for too long - // return what we have - let read_bytes_len = self.buf.borrow().len(); - if read_bytes_len >= self.max_read || deadline_poll_res.is_ready() { - self.return_buf() - } else { - Poll::Pending - } - } - } - } + Pin::new(&mut self.inner).poll_next(cx) } } - -#[cfg(test)] -mod tests { - use super::*; - use futures::{poll, StreamExt}; - use std::io::Cursor; - use std::time::Duration; - use tokio::io::AsyncReadExt; - use tokio_test::assert_pending; - - #[tokio::test] - async fn available_reader_reads_all_available_data_smaller_than_its_buf() { - let data = vec![42u8; 100]; - let mut reader = Cursor::new(data.clone()); - - let mut available_reader = AvailableReader::new(&mut reader, None); - let read_data = available_reader.next().await.unwrap().unwrap(); - - assert_eq!(read_data, data); - assert!(available_reader.next().await.is_none()); - } - - #[tokio::test] - async fn available_reader_reads_all_available_data_bigger_than_its_buf() { - let data = vec![42u8; AvailableReader::>>::BUF_INCREMENT + 100]; - let mut reader = Cursor::new(data.clone()); - - let mut available_reader = AvailableReader::new(&mut reader, None); - let read_data = available_reader.next().await.unwrap().unwrap(); - - assert_eq!(read_data, data); - assert!(available_reader.next().await.is_none()); - } - - #[tokio::test] - async fn available_reader_will_not_wait_for_more_data_if_it_already_has_some() { - let first_data_chunk = vec![42u8; 100]; - let second_data_chunk = vec![123u8; 100]; - - let mut reader_mock = tokio_test::io::Builder::new() - .read(&first_data_chunk) - .wait(Duration::from_millis(100)) // delay is irrelevant, what matters is that we don't get everything immediately - .read(&second_data_chunk) - .build(); - - let mut available_reader = AvailableReader::new(&mut reader_mock, None); - let read_data = available_reader.next().await.unwrap().unwrap(); - - assert_eq!(read_data, first_data_chunk); - assert_pending!(poll!(available_reader.next())); - // before dropping the mock, we need to empty it - let mut buf = vec![0u8; second_data_chunk.len()]; - assert_eq!(reader_mock.read(&mut buf).await.unwrap(), 100); - } - - #[tokio::test] - async fn available_reader_will_wait_for_more_data_if_it_doesnt_have_anything() { - let data = vec![42u8; 100]; - - let mut reader_mock = tokio_test::io::Builder::new() - .wait(Duration::from_millis(100)) - .read(&data) - .build(); - - let mut available_reader = AvailableReader::new(&mut reader_mock, None); - let read_data = available_reader.next().await.unwrap().unwrap(); - - assert_eq!(read_data, data); - assert!(available_reader.next().await.is_none()); - } - - // perhaps the issue of tokio io builder will be resolved in tokio 0.3? - // #[tokio::test] - // async fn available_reader_will_wait_for_more_data_if_its_within_grace_period() { - // let first_data_chunk = vec![42u8; 100]; - // let second_data_chunk = vec![123u8; 100]; - // - // let combined_chunks: Vec<_> = first_data_chunk - // .iter() - // .cloned() - // .chain(second_data_chunk.iter().cloned()) - // .collect(); - // - // let mut reader_mock = tokio_test::io::Builder::new() - // .read(&first_data_chunk) - // .wait(Duration::from_millis(2)) - // .read(&second_data_chunk) - // .build(); - // - // let mut available_reader = AvailableReader { - // buf: RefCell::new(BytesMut::with_capacity(4096)), - // inner: RefCell::new(&mut reader_mock), - // grace_period: Some(delay_for(Duration::from_millis(5))), - // }; - // - // let read_data = available_reader.next().await.unwrap().unwrap(); - // - // assert_eq!(read_data, combined_chunks); - // assert!(available_reader.next().await.is_none()) - // } -} diff --git a/common/socks5/proxy-helpers/src/proxy_runner/inbound.rs b/common/socks5/proxy-helpers/src/proxy_runner/inbound.rs index 584ace090f..e84ffae546 100644 --- a/common/socks5/proxy-helpers/src/proxy_runner/inbound.rs +++ b/common/socks5/proxy-helpers/src/proxy_runner/inbound.rs @@ -61,6 +61,8 @@ where None => (Default::default(), true), }; + error!("{} bytes", read_data.len()); + debug!( target: &*format!("({connection_id}) socks5 inbound"), "[{} bytes]\t{} → local → mixnet → remote → {}. Local closed: {}", From fd4930b198d929b7f9e7ba5c9924e61a9f94795b Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?J=C4=99drzej=20Stuczy=C5=84ski?= Date: Thu, 30 Mar 2023 14:34:20 +0100 Subject: [PATCH 4/6] removed old leftover log statement --- common/socks5/proxy-helpers/src/proxy_runner/inbound.rs | 1 - 1 file changed, 1 deletion(-) diff --git a/common/socks5/proxy-helpers/src/proxy_runner/inbound.rs b/common/socks5/proxy-helpers/src/proxy_runner/inbound.rs index e84ffae546..9b4da73d91 100644 --- a/common/socks5/proxy-helpers/src/proxy_runner/inbound.rs +++ b/common/socks5/proxy-helpers/src/proxy_runner/inbound.rs @@ -61,7 +61,6 @@ where None => (Default::default(), true), }; - error!("{} bytes", read_data.len()); debug!( target: &*format!("({connection_id}) socks5 inbound"), From 9d23766288c93a021515c57f5cb7b919fa9dbe32 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?J=C4=99drzej=20Stuczy=C5=84ski?= Date: Mon, 24 Apr 2023 09:40:12 +0100 Subject: [PATCH 5/6] updated used packet size --- common/socks5-client-core/src/lib.rs | 25 +++++++++---------- sdk/rust/nym-sdk/src/mixnet/client.rs | 2 ++ .../network-requester/src/core.rs | 12 +++------ 3 files changed, 18 insertions(+), 21 deletions(-) diff --git a/common/socks5-client-core/src/lib.rs b/common/socks5-client-core/src/lib.rs index aa85a8a00f..da5f403005 100644 --- a/common/socks5-client-core/src/lib.rs +++ b/common/socks5-client-core/src/lib.rs @@ -7,15 +7,10 @@ use crate::socks::{ authentication::{AuthenticationMethods, Authenticator, User}, server::SphinxSocksServer, }; - use futures::channel::mpsc; use futures::StreamExt; use log::*; use nym_bandwidth_controller::BandwidthController; -#[cfg(target_os = "android")] -use nym_client_core::client::base_client::helpers::setup_empty_reply_surb_backend; -#[cfg(not(target_os = "android"))] -use nym_client_core::client::base_client::non_wasm_helpers; use nym_client_core::client::base_client::{ BaseClientBuilder, ClientInput, ClientOutput, ClientState, }; @@ -23,12 +18,17 @@ use nym_client_core::client::key_manager::KeyManager; use nym_client_core::config::persistence::key_pathfinder::ClientKeyPathfinder; use nym_credential_storage::storage::Storage; use nym_sphinx::addressing::clients::Recipient; -use nym_sphinx::params::PacketSize; use nym_task::{TaskClient, TaskManager}; use nym_validator_client::nyxd::QueryNyxdClient; use nym_validator_client::Client; use std::error::Error; +#[cfg(target_os = "android")] +use nym_client_core::client::base_client::helpers::setup_empty_reply_surb_backend; +#[cfg(not(target_os = "android"))] +use nym_client_core::client::base_client::non_wasm_helpers; +use nym_client_core::config::DebugConfig; + pub mod config; pub mod error; pub mod socks; @@ -103,6 +103,7 @@ impl NymClient { pub fn start_socks5_listener( socks5_config: &Socks5, + debug_config: DebugConfig, client_input: ClientInput, client_output: ClientOutput, client_status: ClientState, @@ -127,13 +128,10 @@ impl NymClient { .. } = client_status; - // FIXME: use correct value from the config once https://github.com/nymtech/nym/pull/3217 - // is merged - let packet_size = config - .get_base() - .get_use_extended_packet_size() - .map(Into::into) - .unwrap_or(PacketSize::RegularPacket); + let packet_size = debug_config + .traffic + .secondary_packet_size + .unwrap_or(debug_config.traffic.primary_packet_size); let authenticator = Authenticator::new(auth_methods, allowed_users); let mut sphinx_socks = SphinxSocksServer::new( @@ -265,6 +263,7 @@ impl NymClient { Self::start_socks5_listener( self.config.get_socks5(), + *self.config.get_debug_settings(), client_input, client_output, client_state, diff --git a/sdk/rust/nym-sdk/src/mixnet/client.rs b/sdk/rust/nym-sdk/src/mixnet/client.rs index d37dd42a56..a685e39fbd 100644 --- a/sdk/rust/nym-sdk/src/mixnet/client.rs +++ b/sdk/rust/nym-sdk/src/mixnet/client.rs @@ -506,6 +506,7 @@ where .socks5_config .clone() .ok_or(Error::Socks5Config { set: false })?; + let debug_config = self.config.debug_config; let (mut started_client, nym_address) = self.connect_to_mixnet_common().await?; let (socks5_status_tx, mut socks5_status_rx) = mpsc::channel(128); @@ -515,6 +516,7 @@ where nym_socks5_client_core::NymClient::start_socks5_listener( &socks5_config, + debug_config, client_input, client_output, client_state.clone(), diff --git a/service-providers/network-requester/src/core.rs b/service-providers/network-requester/src/core.rs index 39060e9e55..3847585f8d 100644 --- a/service-providers/network-requester/src/core.rs +++ b/service-providers/network-requester/src/core.rs @@ -443,14 +443,10 @@ impl NRServiceProvider { return; } - // FIXME: use correct value from the config once https://github.com/nymtech/nym/pull/3217 - // is merged - let packet_size = self - .config - .get_base() - .get_use_extended_packet_size() - .map(Into::into) - .unwrap_or(PacketSize::RegularPacket); + let traffic_config = self.config.get_base().get_debug_config().traffic; + let packet_size = traffic_config + .secondary_packet_size + .unwrap_or(traffic_config.primary_packet_size); let controller_sender_clone = self.controller_sender.clone(); let mix_input_sender_clone = self.mix_input_sender.clone(); From f822d3db7bb3ab9f185b14b5fa873da9cd9b5698 Mon Sep 17 00:00:00 2001 From: Tommy Verrall Date: Mon, 24 Apr 2023 15:54:35 +0200 Subject: [PATCH 6/6] cargo fmt --- common/socks5/proxy-helpers/src/proxy_runner/inbound.rs | 1 - 1 file changed, 1 deletion(-) diff --git a/common/socks5/proxy-helpers/src/proxy_runner/inbound.rs b/common/socks5/proxy-helpers/src/proxy_runner/inbound.rs index 9b4da73d91..584ace090f 100644 --- a/common/socks5/proxy-helpers/src/proxy_runner/inbound.rs +++ b/common/socks5/proxy-helpers/src/proxy_runner/inbound.rs @@ -61,7 +61,6 @@ where None => (Default::default(), true), }; - debug!( target: &*format!("({connection_id}) socks5 inbound"), "[{} bytes]\t{} → local → mixnet → remote → {}. Local closed: {}",