From 7058e7139d51eed40ff925bec2d97bf5bc231248 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jon=20H=C3=A4ggblad?= Date: Wed, 19 Oct 2022 00:15:19 +0200 Subject: [PATCH] WIP: backpressure in out queue control --- Cargo.lock | 1 + .../src/client/cover_traffic_stream.rs | 5 +- clients/client-core/src/client/mix_traffic.rs | 1 + .../real_traffic_stream.rs | 167 +++++++++++++++++- clients/client-core/src/lib.rs | 8 +- clients/native/src/client/mod.rs | 2 +- clients/socks5/src/client/mod.rs | 2 +- clients/socks5/src/main.rs | 2 +- common/client-libs/gateway-client/Cargo.toml | 3 +- .../client-libs/gateway-client/src/client.rs | 8 +- gateway/Cargo.toml | 1 + gateway/src/main.rs | 3 + .../receiver/connection_handler.rs | 5 +- 13 files changed, 188 insertions(+), 20 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 3d03bbfed9..190d90f90a 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3199,6 +3199,7 @@ dependencies = [ "colored", "completions", "config", + "console-subscriber", "credentials", "crypto", "dashmap", diff --git a/clients/client-core/src/client/cover_traffic_stream.rs b/clients/client-core/src/client/cover_traffic_stream.rs index 4a1c5c36a1..293dc9580e 100644 --- a/clients/client-core/src/client/cover_traffic_stream.rs +++ b/clients/client-core/src/client/cover_traffic_stream.rs @@ -176,7 +176,9 @@ impl LoopCoverTrafficStream { // - the receiver channel is closed // in either case there's no recovery and we can only panic //self.mix_tx.unbounded_send(vec![cover_message]).unwrap(); - self.mix_tx.try_send(vec![cover_message]).unwrap(); + if let Err(err) = self.mix_tx.try_send(vec![cover_message]) { + log::error!("Failed to send cover traffic: {}", err); + } // TODO: I'm not entirely sure whether this is really required, because I'm not 100% // sure how `yield_now()` works - whether it just notifies the scheduler or whether it @@ -209,6 +211,7 @@ impl LoopCoverTrafficStream { } next = self.next() => { if next.is_some() { + //log::debug!("loop cover traffic: got next msg to send to mix traffic"); self.on_new_message().await; } else { log::trace!("LoopCoverTrafficStream: Stopping since channel closed"); diff --git a/clients/client-core/src/client/mix_traffic.rs b/clients/client-core/src/client/mix_traffic.rs index 2670c34035..d73ad16899 100644 --- a/clients/client-core/src/client/mix_traffic.rs +++ b/clients/client-core/src/client/mix_traffic.rs @@ -77,6 +77,7 @@ impl MixTrafficController { tokio::select! { mix_packets = self.mix_rx.next() => match mix_packets { Some(mix_packets) => { + //log::debug!("received mix packet to send to gateway"); self.on_messages(mix_packets).await; }, None => { 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 94f7588d8d..2a5a11c75d 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 @@ -4,7 +4,7 @@ use crate::client::mix_traffic::BatchMixMessageSender; use crate::client::real_messages_control::acknowledgement_control::SentPacketNotificationSender; use crate::client::topology_control::TopologyAccessor; -use futures::channel::mpsc; +use futures::channel::mpsc::{self, TrySendError}; use futures::task::{Context, Poll}; use futures::{Future, Stream, StreamExt}; use log::*; @@ -20,6 +20,7 @@ use std::collections::VecDeque; use std::pin::Pin; use std::sync::Arc; use std::time::Duration; +use tokio::time::sleep; #[cfg(not(target_arch = "wasm32"))] use tokio::time; @@ -27,6 +28,8 @@ use tokio::time; #[cfg(target_arch = "wasm32")] use wasm_timer; +const REDUCE_DELAY_THRESHOLD: usize = 100; + /// Configurable parameters of the `OutQueueControl` pub(crate) struct Config { /// Average delay an acknowledgement packet is going to get delay at a single mixnode. @@ -89,6 +92,13 @@ where #[cfg(target_arch = "wasm32")] next_delay: Option>>, + /// The current message delay, only different from `average_message_sending_delay` if there has been + /// backpressure from the gateway client to slow down. + current_average_message_sending_delay: Duration, + + /// Count the number of successful sends to determine if we should reduce sending delay + number_of_consectivive_successful_sends: usize, + /// Channel used for sending prepared sphinx packets to `MixTrafficController` that sends them /// out to the network without any further delays. mix_tx: BatchMixMessageSender, @@ -151,11 +161,14 @@ where our_full_destination: Recipient, topology_access: TopologyAccessor, ) -> Self { + let current_average_message_sending_delay = config.average_message_sending_delay; OutQueueControl { config, ack_key, sent_notifier, next_delay: None, + current_average_message_sending_delay, + number_of_consectivive_successful_sends: 0, mix_tx, real_receiver, our_full_destination, @@ -173,10 +186,12 @@ where self.sent_notifier.unbounded_send(frag_id).unwrap(); } + #[allow(clippy::too_many_lines)] async fn on_message(&mut self, next_message: StreamMessage) { trace!("created new message"); - let next_message = match next_message { + //let next_message = match next_message { + match next_message { StreamMessage::Cover => { // TODO for way down the line: in very rare cases (during topology update) we might have // to wait a really tiny bit before actually obtaining the permit hence messing with our @@ -195,7 +210,7 @@ where } let topology_ref = topology_ref_option.unwrap(); - generate_loop_cover_packet( + let next_message = generate_loop_cover_packet( &mut self.rng, topology_ref, &self.ack_key, @@ -204,11 +219,104 @@ where self.config.average_packet_delay, self.config.cover_packet_size, ) - .expect("Somehow failed to generate a loop cover message with a valid topology") + .expect("Somehow failed to generate a loop cover message with a valid topology"); + + match self.mix_tx.try_send(vec![next_message]) { + Ok(_) => { + self.number_of_consectivive_successful_sends += 1; + if self.number_of_consectivive_successful_sends > REDUCE_DELAY_THRESHOLD { + self.number_of_consectivive_successful_sends = 0; + log::error!( + "old average_message_sending_delay: {:?}", + self.current_average_message_sending_delay + ); + self.current_average_message_sending_delay = + self.current_average_message_sending_delay.mul_f64(0.9); + log::error!( + "new average_message_sending_delay: {:?}", + self.current_average_message_sending_delay + ); + } + } + Err(err) => { + if err.is_full() { + self.number_of_consectivive_successful_sends = 0; + // Increase average send delay + log::error!( + "old average_message_sending_delay: {:?}", + self.current_average_message_sending_delay + ); + //self.current_average_message_sending_delay *= 2; + self.current_average_message_sending_delay = + self.current_average_message_sending_delay.mul_f64(1.2); + log::error!( + "new average_message_sending_delay: {:?}", + self.current_average_message_sending_delay + ); + sleep(Duration::from_millis(100)).await; + } else if err.is_disconnected() { + log::warn!("should not happen during normal operation!"); + } else { + } + } + }; } StreamMessage::Real(real_message) => { - self.sent_notify(real_message.fragment_id); - real_message.mix_packet + let RealMessage { + mix_packet: next_message, + fragment_id, + } = *real_message; + match self.mix_tx.try_send(vec![next_message]) { + Ok(_) => { + self.sent_notify(fragment_id); + self.number_of_consectivive_successful_sends += 1; + if self.number_of_consectivive_successful_sends > REDUCE_DELAY_THRESHOLD { + self.number_of_consectivive_successful_sends = 0; + log::error!( + "old average_message_sending_delay: {:?}", + self.current_average_message_sending_delay + ); + self.current_average_message_sending_delay = + self.current_average_message_sending_delay.mul_f64(0.9); + log::error!( + "new average_message_sending_delay: {:?}", + self.current_average_message_sending_delay + ); + } + } + Err(err) => { + if err.is_full() { + log::error!( + "Failed to send, channel full, will retry: {}", + fragment_id + ); + self.number_of_consectivive_successful_sends = 0; + // Re-queue at the front + let mut msg = err.into_inner(); + assert!(msg.len() == 1); + let msg = msg.pop().unwrap(); + let new_real_message = RealMessage::new(msg, real_message.fragment_id); + self.received_buffer.push_front(new_real_message); + + // Increase average send delay + log::error!( + "old average_message_sending_delay: {:?}", + self.current_average_message_sending_delay + ); + //self.current_average_message_sending_delay *= 2; + self.current_average_message_sending_delay = + self.current_average_message_sending_delay.mul_f64(1.2); + log::error!( + "new average_message_sending_delay: {:?}", + self.current_average_message_sending_delay + ); + sleep(Duration::from_millis(100)).await; + } else if err.is_disconnected() { + log::warn!("should not happen during normal operation!"); + } else { + } + } + }; } }; @@ -222,7 +330,7 @@ where // err.into_inner().len() // ); //} - self.mix_tx.try_send(vec![next_message]).unwrap(); + //self.mix_tx.try_send(vec![next_message]).unwrap(); // JS: Not entirely sure why or how it fixes stuff, but without the yield call, // the UnboundedReceiver [of mix_rx] will not get a chance to read anything @@ -239,12 +347,15 @@ where if let Some(ref mut next_delay) = &mut self.next_delay { // it is not yet time to return a message if next_delay.as_mut().poll(cx).is_pending() { + //log::debug!("poisson: pending"); return Poll::Pending; + //} else { + //log::debug!("poisson: not pending"); }; // we know it's time to send a message, so let's prepare delay for the next one // Get the `now` by looking at the current `delay` deadline - let avg_delay = self.config.average_message_sending_delay; + let avg_delay = self.current_average_message_sending_delay; let next_poisson_delay = sample_poisson_duration(&mut self.rng, avg_delay); // The next interval value is `next_poisson_delay` after the one that just @@ -263,6 +374,11 @@ where // check if we have anything immediately available if let Some(real_available) = self.received_buffer.pop_front() { + //log::debug!("real available"); + log::debug!( + "sending from received_buffer: {}", + self.received_buffer.len() + ); return Poll::Ready(Some(StreamMessage::Real(Box::new(real_available)))); } @@ -274,6 +390,7 @@ where // if there are more messages available, return first one and store the rest Poll::Ready(Some(real_messages)) => { + log::debug!("sending from real_receiver: {}", real_messages.len()); self.received_buffer = real_messages.into(); // we MUST HAVE received at least ONE message Poll::Ready(Some(StreamMessage::Real(Box::new( @@ -282,9 +399,13 @@ where } // otherwise construct a dummy one - Poll::Pending => Poll::Ready(Some(StreamMessage::Cover)), + Poll::Pending => { + //log::debug!("cover message"); + Poll::Ready(Some(StreamMessage::Cover)) + } } } else { + log::debug!("poisson: not send"); // we never set an initial delay - let's do it now cx.waker().wake_by_ref(); @@ -348,14 +469,29 @@ where pub(super) async fn run_with_shutdown(&mut self, mut shutdown: task::ShutdownListener) { debug!("Started OutQueueControl with graceful shutdown support"); + //let stream = self.delay_stream(); + //let stream = delay_stream(self.config.average_message_sending_delay); + //tokio::pin!(stream); + while !shutdown.is_shutdown() { tokio::select! { biased; _ = shutdown.recv() => { log::trace!("OutQueueControl: Received shutdown"); } + //next_message = stream.next() => match next_message { + // Some(next_message) => { + // log::debug!("out queue control: got next_message to send to mix traffic"); + // self.on_message(next_message).await; + // }, + // None => { + // log::trace!("OutQueueControl: Stopping since channel closed"); + // break; + // } + //} next_message = self.next() => match next_message { Some(next_message) => { + //log::debug!("out queue control: got next_message to send to mix traffic"); self.on_message(next_message).await; }, None => { @@ -379,6 +515,19 @@ where } } +//fn delay_stream(avg_delay: Duration) -> impl futures::Stream + 'static { +// let mut rng = rand::rngs::OsRng; +// +// futures::stream::unfold((rng, avg_delay), |(mut rng, avg_delay)| async { +// //let avg_delay = out_queue_control.config.average_message_sending_delay; +// let next_poisson_delay = sample_poisson_duration(&mut rng, avg_delay); +// time::sleep(next_poisson_delay).await; +// //let a = 1; +// let msg = StreamMessage::Cover; +// Some((msg, (rng, avg_delay))) +// }) +//} + impl Stream for OutQueueControl where R: CryptoRng + Rng + Unpin, diff --git a/clients/client-core/src/lib.rs b/clients/client-core/src/lib.rs index 315b22011d..508b5c685c 100644 --- a/clients/client-core/src/lib.rs +++ b/clients/client-core/src/lib.rs @@ -27,10 +27,10 @@ where F: Future + Send + 'static, F::Output: Send + 'static, { - tokio::task::Builder::default() - .name(task_name) - .spawn(future); - //tokio::spawn(future); + //tokio::task::Builder::default() + // .name(task_name) + // .spawn(future); + tokio::spawn(future); } //pub(crate) fn spawn_future_named(future: F, task_name: &str) diff --git a/clients/native/src/client/mod.rs b/clients/native/src/client/mod.rs index 6aa2516031..a8681a01d0 100644 --- a/clients/native/src/client/mod.rs +++ b/clients/native/src/client/mod.rs @@ -360,7 +360,7 @@ impl NymClient { // they are used by cover traffic stream and real traffic stream // sphinx_message_receiver is the receiver used by MixTrafficController that sends the actual traffic //let (sphinx_message_sender, sphinx_message_receiver) = mpsc::unbounded(); - let (sphinx_message_sender, sphinx_message_receiver) = mpsc::channel(300); + let (sphinx_message_sender, sphinx_message_receiver) = mpsc::channel(4); // unwrapped_sphinx_sender is the transmitter of mixnet messages received from the gateway // unwrapped_sphinx_receiver is the receiver for said messages - used by ReceivedMessagesBuffer diff --git a/clients/socks5/src/client/mod.rs b/clients/socks5/src/client/mod.rs index af9c1859e2..2603fd9ad8 100644 --- a/clients/socks5/src/client/mod.rs +++ b/clients/socks5/src/client/mod.rs @@ -349,7 +349,7 @@ impl NymClient { // they are used by cover traffic stream and real traffic stream // sphinx_message_receiver is the receiver used by MixTrafficController that sends the actual traffic //let (sphinx_message_sender, sphinx_message_receiver) = mpsc::unbounded(); - let (sphinx_message_sender, sphinx_message_receiver) = mpsc::channel(300); + let (sphinx_message_sender, sphinx_message_receiver) = mpsc::channel(4); // unwrapped_sphinx_sender is the transmitter of mixnet messages received from the gateway // unwrapped_sphinx_receiver is the receiver for said messages - used by ReceivedMessagesBuffer diff --git a/clients/socks5/src/main.rs b/clients/socks5/src/main.rs index f7a228afe6..f0967ffcd2 100644 --- a/clients/socks5/src/main.rs +++ b/clients/socks5/src/main.rs @@ -11,7 +11,7 @@ pub mod socks; #[tokio::main] async fn main() { // instrument tokio console subscriber needs RUSTFLAGS="--cfg tokio_unstable" at build time - console_subscriber::init(); + //console_subscriber::init(); setup_logging(); println!("{}", banner()); diff --git a/common/client-libs/gateway-client/Cargo.toml b/common/client-libs/gateway-client/Cargo.toml index 01557331fa..05ebb6e96c 100644 --- a/common/client-libs/gateway-client/Cargo.toml +++ b/common/client-libs/gateway-client/Cargo.toml @@ -35,7 +35,8 @@ default-features = false # non-wasm-only dependencies [target."cfg(not(target_arch = \"wasm32\"))".dependencies.tokio] version = "1.21.2" -features = ["macros", "rt", "net", "sync", "time"] +#features = ["macros", "rt", "net", "sync", "time"] +features = ["full", "tracing"] [target."cfg(not(target_arch = \"wasm32\"))".dependencies.tokio-stream] version = "0.1.9" diff --git a/common/client-libs/gateway-client/src/client.rs b/common/client-libs/gateway-client/src/client.rs index f4c2765a63..8e74dbace9 100644 --- a/common/client-libs/gateway-client/src/client.rs +++ b/common/client-libs/gateway-client/src/client.rs @@ -417,8 +417,12 @@ impl GatewayClient { msg: Message, ) -> Result<(), GatewayClientError> { match self.connection { - SocketState::Available(ref mut conn) => Ok(conn.send(msg).await?), + SocketState::Available(ref mut conn) => { + //log::debug!("SocketState::Available: conn.send()"); + Ok(conn.send(msg).await?) + }, SocketState::PartiallyDelegated(ref mut partially_delegated) => { + //log::debug!("SocketState::PartiallyDelegated: send_without_response()"); if let Err(err) = partially_delegated.send_without_response(msg).await { error!("failed to send message without response - {}...", err); // we must ensure we do not leave the task still active @@ -609,6 +613,7 @@ impl GatewayClient { &mut self, packets: Vec, ) -> Result<(), GatewayClientError> { + //log::debug!("batch_send_mix_packets: {}", packets.len()); if !self.authenticated { return Err(GatewayClientError::NotAuthenticated); } @@ -679,6 +684,7 @@ impl GatewayClient { &mut self, mix_packet: MixPacket, ) -> Result<(), GatewayClientError> { + //log::debug!("send_mix_packet"); if !self.authenticated { return Err(GatewayClientError::NotAuthenticated); } diff --git a/gateway/Cargo.toml b/gateway/Cargo.toml index 178ab70c59..be5773fe0f 100644 --- a/gateway/Cargo.toml +++ b/gateway/Cargo.toml @@ -49,6 +49,7 @@ tokio-stream = { version = "0.1.9", features = ["fs"] } tokio-tungstenite = "0.14" tokio-util = { version = "0.7.3", features = ["codec"] } url = { version = "2.2", features = ["serde"] } +console-subscriber = { version = "0.1.8"} # validator-api needs to be built with RUSTFLAGS="--cfg tokio_unstable" # internal coconut-interface = { path = "../common/coconut-interface", optional = true } diff --git a/gateway/src/main.rs b/gateway/src/main.rs index 0651b37237..fbdb0fa1a3 100644 --- a/gateway/src/main.rs +++ b/gateway/src/main.rs @@ -29,6 +29,9 @@ struct Cli { #[tokio::main] async fn main() { + // instrument tokio console subscriber needs RUSTFLAGS="--cfg tokio_unstable" at build time + //console_subscriber::init(); + setup_logging(); println!("{}", banner()); LONG_VERSION diff --git a/gateway/src/node/mixnet_handling/receiver/connection_handler.rs b/gateway/src/node/mixnet_handling/receiver/connection_handler.rs index 2424a363f0..508e75414a 100644 --- a/gateway/src/node/mixnet_handling/receiver/connection_handler.rs +++ b/gateway/src/node/mixnet_handling/receiver/connection_handler.rs @@ -120,7 +120,7 @@ impl ConnectionHandler { fn forward_ack(&self, forward_ack: Option, client_address: DestinationAddressBytes) { if let Some(forward_ack) = forward_ack { - trace!( + debug!( "Sending ack from packet for {} to {}", client_address, forward_ack.next_hop() @@ -131,6 +131,7 @@ impl ConnectionHandler { } async fn handle_processed_packet(&mut self, processed_final_hop: ProcessedFinalHop) { + log::debug!("Handle processed packet"); let client_address = processed_final_hop.destination; let message = processed_final_hop.message; let forward_ack = processed_final_hop.forward_ack; @@ -169,6 +170,8 @@ impl ConnectionHandler { // question: can it also be per connection vs global? // + log::debug!("Handle received packet"); + let processed_final_hop = match self.packet_processor.process_received(framed_sphinx_packet) { Err(e) => {