From 98763d8a80e365d5c5696ceb8ca076e6ec9bbdf0 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jon=20H=C3=A4ggblad?= Date: Wed, 9 Nov 2022 12:46:59 +0100 Subject: [PATCH] prio --- .../input_message_listener.rs | 34 +++++++++++++++---- .../retransmission_request_listener.rs | 11 ++++-- .../src/client/real_messages_control/mod.rs | 2 +- .../real_traffic_stream.rs | 8 ++--- .../proxy-helpers/src/proxy_runner/inbound.rs | 2 ++ 5 files changed, 43 insertions(+), 14 deletions(-) diff --git a/clients/client-core/src/client/real_messages_control/acknowledgement_control/input_message_listener.rs b/clients/client-core/src/client/real_messages_control/acknowledgement_control/input_message_listener.rs index e4d3a7ba56..fb92def08d 100644 --- a/clients/client-core/src/client/real_messages_control/acknowledgement_control/input_message_listener.rs +++ b/clients/client-core/src/client/real_messages_control/acknowledgement_control/input_message_listener.rs @@ -15,6 +15,7 @@ use nymsphinx::preparer::MessagePreparer; use nymsphinx::{acknowledgements::AckKey, addressing::clients::Recipient}; use rand::{CryptoRng, Rng}; use std::sync::Arc; +use std::time::Duration; #[cfg(feature = "reply-surb")] use crate::client::reply_key_storage::ReplyKeyStorage; @@ -186,9 +187,9 @@ where // there's no point in trying to send nothing if let Some(real_messages) = real_messages { // tells real message sender (with the poisson timer) to send this to the mix network - self.real_message_sender - .unbounded_send(real_messages) - .unwrap(); + if self.real_message_sender.send(real_messages).await.is_err() { + panic!(); + } } } @@ -377,9 +378,30 @@ where // there's no point in trying to send nothing if let Some(real_messages) = real_messages { // tells real message sender (with the poisson timer) to send this to the mix network - self.real_message_sender - .unbounded_send(real_messages) - .unwrap(); + + log::info!("chunked into {} messages", real_messages.len()); + log::info!("current capacity: {}", self.real_message_sender.capacity()); + if real_messages.len() > 10 { + // only send if there is nothing in queue + log::info!("sending large message(s)"); + for msg in real_messages { + let msgs = vec![msg]; + loop { + if self.real_message_sender.capacity() > 2 { + if self.real_message_sender.send(msgs).await.is_err() { + panic!(); + } + break; + } + tokio::time::sleep(Duration::from_millis(50)).await; + } + } + log::error!("finished sending large message(s)"); + } else { + if self.real_message_sender.send(real_messages).await.is_err() { + panic!(); + } + } } } } diff --git a/clients/client-core/src/client/real_messages_control/acknowledgement_control/retransmission_request_listener.rs b/clients/client-core/src/client/real_messages_control/acknowledgement_control/retransmission_request_listener.rs index edd3b862bf..aa03988671 100644 --- a/clients/client-core/src/client/real_messages_control/acknowledgement_control/retransmission_request_listener.rs +++ b/clients/client-core/src/client/real_messages_control/acknowledgement_control/retransmission_request_listener.rs @@ -112,12 +112,17 @@ where .unwrap(); // send to `OutQueueControl` to eventually send to the mix network - self.real_message_sender - .unbounded_send(vec![RealMessage::new( + if self + .real_message_sender + .send(vec![RealMessage::new( prepared_fragment.mix_packet, frag_id, )]) - .unwrap(); + .await + .is_err() + { + panic!(); + } } #[cfg(not(target_arch = "wasm32"))] diff --git a/clients/client-core/src/client/real_messages_control/mod.rs b/clients/client-core/src/client/real_messages_control/mod.rs index 117e6d7f6a..67626e8dc2 100644 --- a/clients/client-core/src/client/real_messages_control/mod.rs +++ b/clients/client-core/src/client/real_messages_control/mod.rs @@ -123,7 +123,7 @@ impl RealMessagesController { ) -> Self { let rng = OsRng; - let (real_message_sender, real_message_receiver) = mpsc::unbounded(); + let (real_message_sender, real_message_receiver) = tokio::sync::mpsc::channel(3); let (sent_notifier_tx, sent_notifier_rx) = mpsc::unbounded(); let ack_controller_connectors = AcknowledgementControllerConnectors::new( diff --git a/clients/client-core/src/client/real_messages_control/real_traffic_stream.rs b/clients/client-core/src/client/real_messages_control/real_traffic_stream.rs index 54d98d5941..fa60f6a692 100644 --- a/clients/client-core/src/client/real_messages_control/real_traffic_stream.rs +++ b/clients/client-core/src/client/real_messages_control/real_traffic_stream.rs @@ -242,8 +242,8 @@ impl RealMessage { // messages are already prepared, etc. the real point of it is to forward it to mix_traffic // after sufficient delay -pub type BatchRealMessageSender = mpsc::UnboundedSender>; -type BatchRealMessageReceiver = mpsc::UnboundedReceiver>; +pub type BatchRealMessageSender = tokio::sync::mpsc::Sender>; +type BatchRealMessageReceiver = tokio::sync::mpsc::Receiver>; pub(crate) enum StreamMessage { Cover, @@ -425,7 +425,7 @@ where } // decide what kind of message to send - match Pin::new(&mut self.real_receiver).poll_next(cx) { + match Pin::new(&mut self.real_receiver).poll_recv(cx) { // in the case our real message channel stream was closed, we should also indicate we are closed // (and whoever is using the stream should panic) Poll::Ready(None) => Poll::Ready(None), @@ -472,7 +472,7 @@ where return Poll::Ready(Some(StreamMessage::Real(Box::new(real_available)))); } - match Pin::new(&mut self.real_receiver).poll_next(cx) { + match Pin::new(&mut self.real_receiver).poll_recv(cx) { // in the case our real message channel stream was closed, we should also indicate we are closed // (and whoever is using the stream should panic) Poll::Ready(None) => Poll::Ready(None), diff --git a/common/socks5/proxy-helpers/src/proxy_runner/inbound.rs b/common/socks5/proxy-helpers/src/proxy_runner/inbound.rs index 25bd32e621..7394fa6848 100644 --- a/common/socks5/proxy-helpers/src/proxy_runner/inbound.rs +++ b/common/socks5/proxy-helpers/src/proxy_runner/inbound.rs @@ -70,8 +70,10 @@ where // WIP(JON): here we do the chunking, and send to real_message_sender instead if let Some(chunker) = msg_chunker { + log::info!("({connection_id}): chunking and sending"); let msg = adapter_fn(connection_id, ordered_msg, is_finished); chunker.chunk(msg).await; + log::info!("({connection_id}): chunking and sending done"); } else { mix_sender .unbounded_send(adapter_fn(connection_id, ordered_msg, is_finished))