diff --git a/CHANGELOG.md b/CHANGELOG.md index 5acf6497d4..369c290aab 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,8 +4,16 @@ Post 1.0.0 release, the changelog format is based on [Keep a Changelog](https:// ## Unreleased +### Added + - binaries: add `-c` shortform for `--config-env-file` +### Changed + +- clients: add concept of transmission lanes to better handle multiple data streams ([#1720]) + +[#1720]: https://github.com/nymtech/nym/pull/1720 + ## [v1.1.0](https://github.com/nymtech/nym/tree/v1.1.0) (2022-11-09) diff --git a/Cargo.lock b/Cargo.lock index 184129df61..b83de454cf 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -576,10 +576,18 @@ dependencies = [ "os_str_bytes", ] +[[package]] +name = "client-connections" +version = "0.1.0" +dependencies = [ + "futures", +] + [[package]] name = "client-core" version = "1.1.0" dependencies = [ + "client-connections", "config", "crypto", "dirs", @@ -3125,6 +3133,7 @@ name = "nym-client" version = "1.1.0" dependencies = [ "clap 3.2.8", + "client-connections", "client-core", "coconut-interface", "completions", @@ -3253,6 +3262,7 @@ version = "1.1.0" dependencies = [ "async-trait", "clap 3.2.8", + "client-connections", "completions", "dirs", "futures", @@ -3299,6 +3309,7 @@ name = "nym-socks5-client" version = "1.1.0" dependencies = [ "clap 3.2.8", + "client-connections", "client-core", "coconut-interface", "completions", @@ -4123,6 +4134,7 @@ name = "proxy-helpers" version = "0.1.0" dependencies = [ "bytes", + "client-connections", "futures", "log", "ordered-buffer", diff --git a/Cargo.toml b/Cargo.toml index 4866366280..4ab19fc70e 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -26,6 +26,7 @@ members = [ "common/client-libs/gateway-client", "common/client-libs/mixnet-client", "common/client-libs/validator-client", + "common/client-connections", "common/coconut-interface", "common/commands", "common/config", diff --git a/clients/client-core/Cargo.toml b/clients/client-core/Cargo.toml index 075c46fd29..e296a30591 100644 --- a/clients/client-core/Cargo.toml +++ b/clients/client-core/Cargo.toml @@ -14,11 +14,13 @@ log = "0.4" rand = { version = "0.7.3", features = ["wasm-bindgen"] } serde = { version = "1.0", features = ["derive"] } sled = { version = "0.34", optional = true } +tap = "1.0.1" thiserror = "1.0.34" url = { version ="2.2", features = ["serde"] } # internal config = { path = "../../common/config" } +client-connections = { path = "../../common/client-connections" } crypto = { path = "../../common/crypto" } gateway-client = { path = "../../common/client-libs/gateway-client" } #gateway-client = { path = "../../common/client-libs/gateway-client", default-features = false, features = ["wasm", "coconut"] } @@ -28,7 +30,6 @@ nymsphinx = { path = "../../common/nymsphinx" } pemstore = { path = "../../common/pemstore" } topology = { path = "../../common/topology" } validator-client = { path = "../../common/client-libs/validator-client", default-features = false } -tap = "1.0.1" tokio = { version = "1.21.2", features = ["time", "macros"]} @@ -56,4 +57,5 @@ tempfile = "3.1.0" default = ["reply-surb"] wasm = ["gateway-client/wasm"] coconut = ["gateway-client/coconut", "gateway-requests/coconut"] -reply-surb = ["sled"] \ No newline at end of file +reply-surb = ["sled"] + diff --git a/clients/client-core/src/client/cover_traffic_stream.rs b/clients/client-core/src/client/cover_traffic_stream.rs index 84fd277634..ee3b7851cb 100644 --- a/clients/client-core/src/client/cover_traffic_stream.rs +++ b/clients/client-core/src/client/cover_traffic_stream.rs @@ -178,6 +178,10 @@ impl LoopCoverTrafficStream { // This isn't a problem, if the channel is full means we're already sending the // max amount of messages downstream can handle. log::debug!("Failed to send cover message - channel full"); + // However it's still useful to alert the user that the gateway or the link to + // the gateway can't keep up. Either due to insufficient bandwidth on the + // client side, or that the gateway is overloaded. + log::warn!("Failed to send: gateway appears to not keep up"); } TrySendError::Closed(_) => { log::warn!("Failed to send cover message - channel closed"); diff --git a/clients/client-core/src/client/inbound_messages.rs b/clients/client-core/src/client/inbound_messages.rs index 136ea1d3c4..3cc71d4460 100644 --- a/clients/client-core/src/client/inbound_messages.rs +++ b/clients/client-core/src/client/inbound_messages.rs @@ -1,3 +1,4 @@ +use client_connections::TransmissionLane; use futures::channel::mpsc; use nymsphinx::addressing::clients::Recipient; use nymsphinx::anonymous_replies::ReplySurb; @@ -11,6 +12,7 @@ pub enum InputMessage { recipient: Recipient, data: Vec, with_reply_surb: bool, + lane: TransmissionLane, }, Reply { reply_surb: ReplySurb, @@ -19,11 +21,17 @@ pub enum InputMessage { } impl InputMessage { - pub fn new_fresh(recipient: Recipient, data: Vec, with_reply_surb: bool) -> Self { + pub fn new_fresh( + recipient: Recipient, + data: Vec, + with_reply_surb: bool, + lane: TransmissionLane, + ) -> Self { InputMessage::Fresh { recipient, data, with_reply_surb, + lane, } } diff --git a/clients/client-core/src/client/real_messages_control/acknowledgement_control/acknowledgement_listener.rs b/clients/client-core/src/client/real_messages_control/acknowledgement_control/acknowledgement_listener.rs index 2f9a6053ec..9b285d8444 100644 --- a/clients/client-core/src/client/real_messages_control/acknowledgement_control/acknowledgement_listener.rs +++ b/clients/client-core/src/client/real_messages_control/acknowledgement_control/acknowledgement_listener.rs @@ -33,7 +33,7 @@ impl AcknowledgementListener { } async fn on_ack(&mut self, ack_content: Vec) { - debug!("Received an ack"); + trace!("Received an ack"); let frag_id = match recover_identifier(&self.ack_key, &ack_content) .map(FragmentIdentifier::try_from_bytes) { 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 57e1162a72..20cbc10de5 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 @@ -8,6 +8,7 @@ use crate::client::{ real_messages_control::real_traffic_stream::{BatchRealMessageSender, RealMessage}, topology_control::TopologyAccessor, }; +use client_connections::TransmissionLane; use futures::StreamExt; use log::*; use nymsphinx::anonymous_replies::ReplySurb; @@ -104,6 +105,7 @@ where content: Vec, with_reply_surb: bool, ) -> Option> { + log::trace!("handling msg size: {}", content.len()); let topology_permit = self.topology_access.get_read_permit().await; let topology = match topology_permit .try_get_valid_topology_ref(&self.ack_recipient, Some(&recipient)) @@ -164,26 +166,30 @@ where } async fn on_input_message(&mut self, msg: InputMessage) { - let real_messages = match msg { + let (real_messages, lane) = match msg { InputMessage::Fresh { recipient, data, with_reply_surb, - } => { + lane, + } => ( self.handle_fresh_message(recipient, data, with_reply_surb) + .await, + lane, + ), + InputMessage::Reply { reply_surb, data } => ( + self.handle_reply(reply_surb, data) .await - } - InputMessage::Reply { reply_surb, data } => self - .handle_reply(reply_surb, data) - .await - .map(|message| vec![message]), + .map(|message| vec![message]), + TransmissionLane::Reply, + ), }; // 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) + .unbounded_send((real_messages, lane)) .unwrap(); } } 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..8a26809c79 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 @@ -1,17 +1,21 @@ // Copyright 2021 - Nym Technologies SA // SPDX-License-Identifier: Apache-2.0 -use super::action_controller::{Action, ActionSender}; -use super::PendingAcknowledgement; -use super::RetransmissionRequestReceiver; +use super::{ + action_controller::{Action, ActionSender}, + PendingAcknowledgement, RetransmissionRequestReceiver, +}; use crate::client::{ real_messages_control::real_traffic_stream::{BatchRealMessageSender, RealMessage}, topology_control::TopologyAccessor, }; + +use client_connections::TransmissionLane; use futures::StreamExt; use log::*; -use nymsphinx::preparer::MessagePreparer; -use nymsphinx::{acknowledgements::AckKey, addressing::clients::Recipient}; +use nymsphinx::{ + acknowledgements::AckKey, addressing::clients::Recipient, preparer::MessagePreparer, +}; use rand::{CryptoRng, Rng}; use std::sync::{Arc, Weak}; @@ -113,10 +117,10 @@ where // send to `OutQueueControl` to eventually send to the mix network self.real_message_sender - .unbounded_send(vec![RealMessage::new( - prepared_fragment.mix_packet, - frag_id, - )]) + .unbounded_send(( + vec![RealMessage::new(prepared_fragment.mix_packet, frag_id)], + TransmissionLane::Retransmission, + )) .unwrap(); } 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 d97e27bdc3..a7c03dc1c0 100644 --- a/clients/client-core/src/client/real_messages_control/mod.rs +++ b/clients/client-core/src/client/real_messages_control/mod.rs @@ -14,6 +14,7 @@ use crate::client::{ topology_control::TopologyAccessor, }; use crate::spawn_future; +use client_connections::ClosedConnectionReceiver; use futures::channel::mpsc; use gateway_client::AcknowledgementReceiver; use log::*; @@ -110,6 +111,7 @@ impl RealMessagesController { mix_sender: BatchMixMessageSender, topology_access: TopologyAccessor, #[cfg(feature = "reply-surb")] reply_key_storage: ReplyKeyStorage, + closed_connection_rx: ClosedConnectionReceiver, ) -> Self { let rng = OsRng; @@ -159,6 +161,7 @@ impl RealMessagesController { rng, config.self_recipient, topology_access, + closed_connection_rx, ); RealMessagesController { 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 2377839975..a42c24892e 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,6 +4,7 @@ use crate::client::mix_traffic::BatchMixMessageSender; use crate::client::real_messages_control::acknowledgement_control::SentPacketNotificationSender; use crate::client::topology_control::TopologyAccessor; +use client_connections::{ClosedConnectionReceiver, ConnectionId, TransmissionLane}; use futures::channel::mpsc; use futures::task::{Context, Poll}; use futures::{Future, Stream, StreamExt}; @@ -16,7 +17,6 @@ use nymsphinx::forwarding::packet::MixPacket; use nymsphinx::params::PacketSize; use nymsphinx::utils::sample_poisson_duration; use rand::{CryptoRng, Rng}; -use std::collections::VecDeque; use std::pin::Pin; use std::sync::Arc; use std::time::Duration; @@ -27,22 +27,22 @@ use tokio::time; #[cfg(target_arch = "wasm32")] use wasm_timer; -// The minimum time between increasing the average delay between packets. If we hit the ceiling in -// the available buffer space we want to take somewhat swift action, but we still need to give a -// short time to give the channel a chance reduce pressure. -const INCREASE_DELAY_MIN_CHANGE_INTERVAL_SECS: u64 = 1; -// The minimum time between decreasing the average delay between packets. We don't want to change -// to quickly to keep things somewhat stable. Also there are buffers downstreams meaning we need to -// wait a little to see the effect before we decrease further. -const DECREASE_DELAY_MIN_CHANGE_INTERVAL_SECS: u64 = 30; -// If we enough time passes without any sign of backpressure in the channel, we can consider -// lowering the average delay. The goal is to keep somewhat stable, rather than maxing out -// bandwidth at all times. -const ACCEPTABLE_TIME_WITHOUT_BACKPRESSURE_SECS: u64 = 30; -// The maximum multiplier we apply to the base average Poisson delay. -const MAX_DELAY_MULTIPLIER: u32 = 6; -// The minium multiplier we apply to the base average Poisson delay. -const MIN_DELAY_MULTIPLIER: u32 = 1; +use self::{ + sending_delay_controller::SendingDelayController, transmission_buffer::TransmissionBuffer, +}; + +mod sending_delay_controller; +mod transmission_buffer; + +#[cfg(not(target_arch = "wasm32"))] +fn get_time_now() -> time::Instant { + time::Instant::now() +} + +#[cfg(target_arch = "wasm32")] +fn get_time_now() -> wasm_timer::Instant { + wasm_timer::Instant::now() +} /// Configurable parameters of the `OutQueueControl` pub(crate) struct Config { @@ -85,101 +85,6 @@ impl Config { } } -struct SendingDelayController { - /// Multiply the average sending delay. - /// This is normally set to unity, but if we detect backpressure we increase this - /// multiplier. We use discrete steps. - current_multiplier: u32, - - /// Maximum delay multiplier - upper_bound: u32, - - /// Minimum delay multiplier - lower_bound: u32, - - /// To make sure we don't change the multiplier to fast, we limit a change to some duration - #[cfg(not(target_arch = "wasm32"))] - time_when_changed: time::Instant, - - #[cfg(target_arch = "wasm32")] - time_when_changed: wasm_timer::Instant, - - /// If we have a long enough time without any backpressure detected we try reducing the sending - /// delay multiplier - #[cfg(not(target_arch = "wasm32"))] - time_when_backpressure_detected: time::Instant, - - #[cfg(target_arch = "wasm32")] - time_when_backpressure_detected: wasm_timer::Instant, -} - -#[cfg(not(target_arch = "wasm32"))] -fn get_time_now() -> time::Instant { - time::Instant::now() -} - -#[cfg(target_arch = "wasm32")] -fn get_time_now() -> wasm_timer::Instant { - wasm_timer::Instant::now() -} - -impl SendingDelayController { - fn new(lower_bound: u32, upper_bound: u32) -> Self { - assert!(lower_bound <= upper_bound); - let now = get_time_now(); - SendingDelayController { - current_multiplier: MIN_DELAY_MULTIPLIER, - upper_bound, - lower_bound, - time_when_changed: now, - time_when_backpressure_detected: now, - } - } - - fn current_multiplier(&self) -> u32 { - self.current_multiplier - } - - fn increase_delay_multiplier(&mut self) { - self.current_multiplier = - (self.current_multiplier + 1).clamp(self.lower_bound, self.upper_bound); - self.time_when_changed = get_time_now(); - log::debug!( - "Increasing sending delay multiplier to: {}", - self.current_multiplier - ); - } - - fn decrease_delay_multiplier(&mut self) { - self.current_multiplier = - (self.current_multiplier - 1).clamp(self.lower_bound, self.upper_bound); - self.time_when_changed = get_time_now(); - log::debug!( - "Decreasing sending delay multiplier to: {}", - self.current_multiplier - ); - } - - fn record_backpressure_detected(&mut self) { - self.time_when_backpressure_detected = get_time_now(); - } - - fn not_increased_delay_recently(&self) -> bool { - get_time_now() - > self.time_when_changed + Duration::from_secs(INCREASE_DELAY_MIN_CHANGE_INTERVAL_SECS) - } - - fn is_sending_reliable(&self) -> bool { - let now = get_time_now(); - let delay_change_interval = Duration::from_secs(DECREASE_DELAY_MIN_CHANGE_INTERVAL_SECS); - let acceptable_time_without_backpressure = - Duration::from_secs(ACCEPTABLE_TIME_WITHOUT_BACKPRESSURE_SECS); - - now > self.time_when_backpressure_detected + acceptable_time_without_backpressure - && now > self.time_when_changed + delay_change_interval - } -} - pub(crate) struct OutQueueControl where R: CryptoRng + Rng, @@ -203,7 +108,7 @@ where // To make sure we don't overload the mix_tx channel, we limit the rate we are pushing // messages. - sending_rate_controller: SendingDelayController, + sending_delay_controller: SendingDelayController, /// Channel used for sending prepared sphinx packets to `MixTrafficController` that sends them /// out to the network without any further delays. @@ -222,8 +127,13 @@ where /// Accessor to the common instance of network topology. topology_access: TopologyAccessor, - /// Buffer containing all real messages received. It is first exhausted before more are pulled. - received_buffer: VecDeque, + /// Buffer containing all incoming real messages keyed by transmission lane, that we will send + /// out to the mixnet. + transmission_buffer: TransmissionBuffer, + + /// Incoming channel for being notified of closed connections, so that we can close lanes + /// corresponding to connections. To avoid sending traffic unnecessary + closed_connection_rx: ClosedConnectionReceiver, } pub(crate) struct RealMessage { @@ -242,8 +152,9 @@ impl RealMessage { // messages are already prepared, etc. the real point of it is to forward it to mix_traffic // after sufficient delay -pub(crate) type BatchRealMessageSender = mpsc::UnboundedSender>; -type BatchRealMessageReceiver = mpsc::UnboundedReceiver>; +pub(crate) type BatchRealMessageSender = + mpsc::UnboundedSender<(Vec, TransmissionLane)>; +type BatchRealMessageReceiver = mpsc::UnboundedReceiver<(Vec, TransmissionLane)>; pub(crate) enum StreamMessage { Cover, @@ -266,22 +177,21 @@ where rng: R, our_full_destination: Recipient, topology_access: TopologyAccessor, + closed_connection_rx: ClosedConnectionReceiver, ) -> Self { OutQueueControl { config, ack_key, sent_notifier, next_delay: None, - sending_rate_controller: SendingDelayController::new( - MIN_DELAY_MULTIPLIER, - MAX_DELAY_MULTIPLIER, - ), + sending_delay_controller: Default::default(), mix_tx, real_receiver, our_full_destination, rng, topology_access, - received_buffer: VecDeque::with_capacity(0), // we won't be putting any data into this guy directly + transmission_buffer: Default::default(), + closed_connection_rx, } } @@ -346,6 +256,10 @@ where self.sent_notify(fragment_id); } + // In addition to closing connections on receiving messages throught closed_connection_rx, + // also close connections when sufficiently stale. + self.transmission_buffer.prune_stale_connections(); + // 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 // JS2: Basically it was the case that with high enough rate, the stream had already a next value @@ -357,35 +271,41 @@ where tokio::task::yield_now().await; } + fn on_close_connection(&mut self, connection_id: ConnectionId) { + log::debug!("Removing lane for connection: {connection_id}"); + self.transmission_buffer + .remove(&TransmissionLane::ConnectionId(connection_id)); + } + fn current_average_message_sending_delay(&self) -> Duration { self.config.average_message_sending_delay - * self.sending_rate_controller.current_multiplier() + * self.sending_delay_controller.current_multiplier() } fn adjust_current_average_message_sending_delay(&mut self) { let used_slots = self.mix_tx.max_capacity() - self.mix_tx.capacity(); log::trace!( "used_slots: {used_slots}, current_multiplier: {}", - self.sending_rate_controller.current_multiplier() + self.sending_delay_controller.current_multiplier() ); // Even just a single used slot is enough to signal backpressure if used_slots > 0 { log::trace!("Backpressure detected"); - self.sending_rate_controller.record_backpressure_detected(); + self.sending_delay_controller.record_backpressure_detected(); } // If the buffer is running out, slow down the sending rate if self.mix_tx.capacity() == 0 - && self.sending_rate_controller.not_increased_delay_recently() + && self.sending_delay_controller.not_increased_delay_recently() { - self.sending_rate_controller.increase_delay_multiplier(); + self.sending_delay_controller.increase_delay_multiplier(); } // Very carefully step up the sending rate in case it seems like we can solidly handle the // current rate. - if self.sending_rate_controller.is_sending_reliable() { - self.sending_rate_controller.decrease_delay_multiplier(); + if self.sending_delay_controller.is_sending_reliable() { + self.sending_delay_controller.decrease_delay_multiplier(); } } @@ -395,6 +315,13 @@ where self.adjust_current_average_message_sending_delay(); let avg_delay = self.current_average_message_sending_delay(); + // Start by checking if we have any incoming messages about closed connections + // NOTE: this feels a bit iffy, the `OutQueueControl` is getting ripe for a rewrite to + // something simpler. + if let Poll::Ready(Some(id)) = Pin::new(&mut self.closed_connection_rx).poll_next(cx) { + self.on_close_connection(id); + } + 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() { @@ -419,28 +346,35 @@ where next_delay.as_mut().reset(next_poisson_delay); } - // check if we have anything immediately available - if let Some(real_available) = self.received_buffer.pop_front() { - return Poll::Ready(Some(StreamMessage::Real(Box::new(real_available)))); - } - - // decide what kind of message to send + // On every iteration we get new messages from upstream. Given that these come bunched + // in `Vec`, this ensures that on average we will fetch messages faster than we can + // send, which is a condition for being able to multiplex sphinx packets from multiple + // data streams. match Pin::new(&mut self.real_receiver).poll_next(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), - // if there are more messages available, return first one and store the rest - Poll::Ready(Some(real_messages)) => { - self.received_buffer = real_messages.into(); - // we MUST HAVE received at least ONE message - Poll::Ready(Some(StreamMessage::Real(Box::new( - self.received_buffer.pop_front().unwrap(), - )))) + Poll::Ready(Some((real_messages, conn_id))) => { + log::trace!("handling real_messages: size: {}", real_messages.len()); + + self.transmission_buffer.store(&conn_id, real_messages); + let real_next = self + .transmission_buffer + .pop_next_message_at_random() + .expect("we just added one"); + + Poll::Ready(Some(StreamMessage::Real(Box::new(real_next)))) } - // otherwise construct a dummy one - Poll::Pending => Poll::Ready(Some(StreamMessage::Cover)), + Poll::Pending => { + if let Some(real_next) = self.transmission_buffer.pop_next_message_at_random() { + Poll::Ready(Some(StreamMessage::Real(Box::new(real_next)))) + } else { + // otherwise construct a dummy one + Poll::Ready(Some(StreamMessage::Cover)) + } + } } } else { // we never set an initial delay - let's do it now @@ -462,14 +396,9 @@ where } fn poll_immediate(&mut self, cx: &mut Context<'_>) -> Poll> { - // check if we have anything immediately available - if let Some(real_available) = self.received_buffer.pop_front() { - // if there are more messages immediately available, notify the runtime - // because we should be polled again - if !self.received_buffer.is_empty() { - cx.waker().wake_by_ref() - } - return Poll::Ready(Some(StreamMessage::Real(Box::new(real_available)))); + // Start by checking if we have any incoming messages about closed connections + if let Poll::Ready(Some(id)) = Pin::new(&mut self.closed_connection_rx).poll_next(cx) { + self.on_close_connection(id); } match Pin::new(&mut self.real_receiver).poll_next(cx) { @@ -477,17 +406,26 @@ where // (and whoever is using the stream should panic) Poll::Ready(None) => Poll::Ready(None), - // if there are more messages available, return first one and store the rest - Poll::Ready(Some(real_messages)) => { - self.received_buffer = real_messages.into(); - // we MUST HAVE received at least ONE message - Poll::Ready(Some(StreamMessage::Real(Box::new( - self.received_buffer.pop_front().unwrap(), - )))) + Poll::Ready(Some((real_messages, conn_id))) => { + log::trace!("handling real_messages: size: {}", real_messages.len()); + + // First store what we got for the given connection id + self.transmission_buffer.store(&conn_id, real_messages); + let real_next = self + .transmission_buffer + .pop_next_message_at_random() + .expect("we just added one"); + + Poll::Ready(Some(StreamMessage::Real(Box::new(real_next)))) } - // if there's nothing, then there's nothing - Poll::Pending => Poll::Pending, + Poll::Pending => { + if let Some(real_next) = self.transmission_buffer.pop_next_message_at_random() { + Poll::Ready(Some(StreamMessage::Real(Box::new(real_next)))) + } else { + Poll::Pending + } + } } } @@ -502,24 +440,47 @@ where } } + #[cfg(not(target_arch = "wasm32"))] + fn log_status(&self) { + let packets = self.transmission_buffer.total_size(); + let backlog = self.transmission_buffer.total_size_in_bytes() as f64 / 1024.0; + let lanes = self.transmission_buffer.num_lanes(); + let mult = self.sending_delay_controller.current_multiplier(); + let delay = self.current_average_message_sending_delay().as_millis(); + if self.config.disable_poisson_packet_distribution { + log::info!( + "Status: {lanes} lanes, backlog: {:.2} kiB ({packets}), no delay", + backlog + ); + } else { + log::info!( + "Status: {lanes} lanes, backlog: {:.2} kiB ({packets}), avg delay: {}ms ({mult})", + backlog, + delay + ); + } + } + #[cfg(not(target_arch = "wasm32"))] pub(super) async fn run_with_shutdown(&mut self, mut shutdown: task::ShutdownListener) { debug!("Started OutQueueControl with graceful shutdown support"); + let mut status_timer = tokio::time::interval(Duration::from_secs(1)); + while !shutdown.is_shutdown() { tokio::select! { biased; _ = shutdown.recv() => { log::trace!("OutQueueControl: Received shutdown"); } - next_message = self.next() => match next_message { - Some(next_message) => { - self.on_message(next_message).await; - }, - None => { - log::trace!("OutQueueControl: Stopping since channel closed"); - break; - } + _ = status_timer.tick() => { + self.log_status(); + } + next_message = self.next() => if let Some(next_message) = next_message { + self.on_message(next_message).await; + } else { + log::trace!("OutQueueControl: Stopping since channel closed"); + break; } } } diff --git a/clients/client-core/src/client/real_messages_control/real_traffic_stream/sending_delay_controller.rs b/clients/client-core/src/client/real_messages_control/real_traffic_stream/sending_delay_controller.rs new file mode 100644 index 0000000000..c90f7b4983 --- /dev/null +++ b/clients/client-core/src/client/real_messages_control/real_traffic_stream/sending_delay_controller.rs @@ -0,0 +1,124 @@ +// Copyright 2021 - Nym Technologies SA +// SPDX-License-Identifier: Apache-2.0 + +use super::get_time_now; +use std::time::Duration; + +#[cfg(not(target_arch = "wasm32"))] +use tokio::time; +#[cfg(target_arch = "wasm32")] +use wasm_timer; + +// The minimum time between increasing the average delay between packets. If we hit the ceiling in +// the available buffer space we want to take somewhat swift action, but we still need to give a +// short time to give the channel a chance reduce pressure. +const INCREASE_DELAY_MIN_CHANGE_INTERVAL_SECS: u64 = 1; +// The minimum time between decreasing the average delay between packets. We don't want to change +// to quickly to keep things somewhat stable. Also there are buffers downstreams meaning we need to +// wait a little to see the effect before we decrease further. +const DECREASE_DELAY_MIN_CHANGE_INTERVAL_SECS: u64 = 30; +// If we enough time passes without any sign of backpressure in the channel, we can consider +// lowering the average delay. The goal is to keep somewhat stable, rather than maxing out +// bandwidth at all times. +const ACCEPTABLE_TIME_WITHOUT_BACKPRESSURE_SECS: u64 = 30; +// The maximum multiplier we apply to the base average Poisson delay. +const MAX_DELAY_MULTIPLIER: u32 = 6; +// The minium multiplier we apply to the base average Poisson delay. +const MIN_DELAY_MULTIPLIER: u32 = 1; + +pub(crate) struct SendingDelayController { + /// Multiply the average sending delay. + /// This is normally set to unity, but if we detect backpressure we increase this + /// multiplier. We use discrete steps. + current_multiplier: u32, + + /// Maximum delay multiplier + upper_bound: u32, + + /// Minimum delay multiplier + lower_bound: u32, + + /// To make sure we don't change the multiplier to fast, we limit a change to some duration + #[cfg(not(target_arch = "wasm32"))] + time_when_changed: time::Instant, + + #[cfg(target_arch = "wasm32")] + time_when_changed: wasm_timer::Instant, + + /// If we have a long enough time without any backpressure detected we try reducing the sending + /// delay multiplier + #[cfg(not(target_arch = "wasm32"))] + time_when_backpressure_detected: time::Instant, + + #[cfg(target_arch = "wasm32")] + time_when_backpressure_detected: wasm_timer::Instant, +} + +impl Default for SendingDelayController { + fn default() -> Self { + SendingDelayController::new(MIN_DELAY_MULTIPLIER, MAX_DELAY_MULTIPLIER) + } +} + +impl SendingDelayController { + pub(crate) fn new(lower_bound: u32, upper_bound: u32) -> Self { + assert!(lower_bound <= upper_bound); + let now = get_time_now(); + SendingDelayController { + current_multiplier: MIN_DELAY_MULTIPLIER, + upper_bound, + lower_bound, + time_when_changed: now, + time_when_backpressure_detected: now, + } + } + + pub(crate) fn current_multiplier(&self) -> u32 { + self.current_multiplier + } + + pub(crate) fn increase_delay_multiplier(&mut self) { + if self.current_multiplier < self.upper_bound { + self.current_multiplier = + (self.current_multiplier + 1).clamp(self.lower_bound, self.upper_bound); + self.time_when_changed = get_time_now(); + log::debug!( + "Increasing sending delay multiplier to: {}", + self.current_multiplier + ); + } else { + log::warn!("Trying to increase delay multipler higher than allowed"); + } + } + + pub(crate) fn decrease_delay_multiplier(&mut self) { + if self.current_multiplier > self.lower_bound { + self.current_multiplier = + (self.current_multiplier - 1).clamp(self.lower_bound, self.upper_bound); + self.time_when_changed = get_time_now(); + log::debug!( + "Decreasing sending delay multiplier to: {}", + self.current_multiplier + ); + } + } + + pub(crate) fn record_backpressure_detected(&mut self) { + self.time_when_backpressure_detected = get_time_now(); + } + + pub(crate) fn not_increased_delay_recently(&self) -> bool { + get_time_now() + > self.time_when_changed + Duration::from_secs(INCREASE_DELAY_MIN_CHANGE_INTERVAL_SECS) + } + + pub(crate) fn is_sending_reliable(&self) -> bool { + let now = get_time_now(); + let delay_change_interval = Duration::from_secs(DECREASE_DELAY_MIN_CHANGE_INTERVAL_SECS); + let acceptable_time_without_backpressure = + Duration::from_secs(ACCEPTABLE_TIME_WITHOUT_BACKPRESSURE_SECS); + + now > self.time_when_backpressure_detected + acceptable_time_without_backpressure + && now > self.time_when_changed + delay_change_interval + } +} 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 new file mode 100644 index 0000000000..2510a65e05 --- /dev/null +++ b/clients/client-core/src/client/real_messages_control/real_traffic_stream/transmission_buffer.rs @@ -0,0 +1,207 @@ +// Copyright 2022 - Nym Technologies SA +// SPDX-License-Identifier: Apache-2.0 + +use client_connections::TransmissionLane; +use rand::seq::SliceRandom; +use std::{ + collections::{HashMap, HashSet, VecDeque}, + time::Duration, +}; + +#[cfg(not(target_arch = "wasm32"))] +use tokio::time; + +#[cfg(target_arch = "wasm32")] +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; +// As a way of prune connections we also check for timeouts. +const MSG_CONSIDERED_STALE_AFTER_SECS: u64 = 10 * 60; + +#[derive(Default)] +pub(crate) struct TransmissionBuffer { + buffer: HashMap, +} + +impl TransmissionBuffer { + #[allow(unused)] + pub(crate) fn is_empty(&self) -> bool { + self.buffer.is_empty() + } + + pub(crate) fn remove(&mut self, lane: &TransmissionLane) -> Option { + self.buffer.remove(lane) + } + + #[cfg(not(target_arch = "wasm32"))] + pub(crate) fn num_lanes(&self) -> usize { + self.buffer.keys().count() + } + + #[allow(unused)] + pub(crate) fn connections(&self) -> HashSet { + self.buffer + .keys() + .filter_map(|lane| match lane { + TransmissionLane::ConnectionId(id) => Some(id), + _ => None, + }) + .copied() + .collect() + } + + #[cfg(not(target_arch = "wasm32"))] + pub(crate) fn total_size(&self) -> usize { + self.buffer.values().map(LaneBufferEntry::len).sum() + } + + #[cfg(not(target_arch = "wasm32"))] + pub(crate) fn total_size_in_bytes(&self) -> usize { + self.buffer + .values() + .map(|lane_buffer_entry| { + lane_buffer_entry + .real_messages + .iter() + .map(|real_message| real_message.mix_packet.sphinx_packet().len()) + .sum::() + }) + .sum() + } + + fn get_oldest_set(&self) -> Vec { + let mut buffer: Vec<_> = self + .buffer + .iter() + .map(|(k, v)| (k, v.messages_transmitted)) + .collect(); + buffer.sort_by_key(|v| v.1); + buffer + .iter() + .rev() + .map(|(k, _)| *k) + .take(OLDEST_LANE_SET_SIZE) + .copied() + .collect() + } + + pub(crate) fn store(&mut self, lane: &TransmissionLane, real_messages: Vec) { + if let Some(lane_buffer_entry) = self.buffer.get_mut(lane) { + lane_buffer_entry.append(real_messages); + } else { + self.buffer + .insert(*lane, LaneBufferEntry::new(real_messages)); + } + } + + fn pick_random_lane(&self) -> Option<&TransmissionLane> { + let lanes: Vec<&TransmissionLane> = self.buffer.keys().collect(); + lanes.choose(&mut rand::thread_rng()).copied() + } + + fn pick_random_small_lane(&self) -> Option<&TransmissionLane> { + let lanes: Vec<&TransmissionLane> = self + .buffer + .iter() + .filter(|(_, v)| v.is_small()) + .map(|(k, _)| k) + .collect(); + lanes.choose(&mut rand::thread_rng()).copied() + } + + fn pick_random_old_lane(&self) -> Option { + let lanes = self.get_oldest_set(); + lanes.choose(&mut rand::thread_rng()).copied() + } + + fn pop_front_from_lane(&mut self, lane: &TransmissionLane) -> Option { + let real_msgs_queued = self.buffer.get_mut(lane)?; + let real_next = real_msgs_queued.pop_front()?; + real_msgs_queued.messages_transmitted += 1; + if real_msgs_queued.is_empty() { + self.buffer.remove(lane); + } + Some(real_next) + } + + pub(crate) fn pop_next_message_at_random(&mut self) -> Option { + if self.buffer.is_empty() { + return None; + } + + // Very basic heuristic where we prioritize according to small lanes first, the older lanes + // to try to finish lanes when possible, then the rest. + let lane = if let Some(small_lane) = self.pick_random_small_lane() { + *small_lane + } else if let Some(old_lane) = self.pick_random_old_lane() { + old_lane + } else { + *self.pick_random_lane()? + }; + + log::trace!("picking to send from lane: {:?}", lane); + self.pop_front_from_lane(&lane) + } + + pub(crate) fn prune_stale_connections(&mut self) { + let stale_entries: Vec<_> = self + .buffer + .iter() + .filter_map(|(lane, entry)| if entry.is_stale() { Some(lane) } else { None }) + .copied() + .collect(); + + for lane in stale_entries { + self.remove(&lane); + } + } +} + +pub(crate) struct LaneBufferEntry { + pub real_messages: VecDeque, + pub messages_transmitted: usize, + #[cfg(not(target_arch = "wasm32"))] + pub time_for_last_activity: time::Instant, + #[cfg(target_arch = "wasm32")] + pub time_for_last_activity: wasm_timer::Instant, +} + +impl LaneBufferEntry { + fn new(real_messages: Vec) -> Self { + LaneBufferEntry { + real_messages: real_messages.into(), + messages_transmitted: 0, + time_for_last_activity: get_time_now(), + } + } + + fn append(&mut self, real_messages: Vec) { + self.real_messages.append(&mut real_messages.into()); + self.time_for_last_activity = get_time_now(); + } + + fn pop_front(&mut self) -> Option { + self.real_messages.pop_front() + } + + fn is_small(&self) -> bool { + self.real_messages.len() < 100 + } + + fn is_stale(&self) -> bool { + get_time_now() - self.time_for_last_activity + > Duration::from_secs(MSG_CONSIDERED_STALE_AFTER_SECS) + } + + #[cfg(not(target_arch = "wasm32"))] + fn len(&self) -> usize { + self.real_messages.len() + } + + fn is_empty(&self) -> bool { + self.real_messages.is_empty() + } +} diff --git a/clients/client-core/src/client/received_buffer.rs b/clients/client-core/src/client/received_buffer.rs index 9830f0a869..fa61b9b9a0 100644 --- a/clients/client-core/src/client/received_buffer.rs +++ b/clients/client-core/src/client/received_buffer.rs @@ -208,7 +208,7 @@ impl ReceivedMessagesBuffer { } async fn handle_new_received(&mut self, msgs: Vec>) { - debug!( + trace!( "Processing {:?} new message that might get added to the buffer!", msgs.len() ); diff --git a/clients/native/Cargo.toml b/clients/native/Cargo.toml index 95e057adb1..5da490d504 100644 --- a/clients/native/Cargo.toml +++ b/clients/native/Cargo.toml @@ -33,6 +33,7 @@ tokio-tungstenite = "0.14" # websocket ## internal client-core = { path = "../client-core" } +client-connections = { path = "../../common/client-connections" } coconut-interface = { path = "../../common/coconut-interface", optional = true } config = { path = "../../common/config" } completions = { path = "../../common/completions" } diff --git a/clients/native/examples/websocket_binarysend.rs b/clients/native/examples/websocket_binarysend.rs index 7ddf53e913..160a9bf009 100644 --- a/clients/native/examples/websocket_binarysend.rs +++ b/clients/native/examples/websocket_binarysend.rs @@ -43,6 +43,7 @@ async fn send_file_with_reply() { recipient, message: read_data, with_reply_surb: true, + connection_id: 0, }; println!("sending content of 'dummy_file' over the mix network..."); @@ -91,6 +92,7 @@ async fn send_file_without_reply() { recipient, message: read_data, with_reply_surb: false, + connection_id: 0, }; println!("sending content of 'dummy_file' over the mix network..."); diff --git a/clients/native/src/client/mod.rs b/clients/native/src/client/mod.rs index fd27282841..df77a7fb77 100644 --- a/clients/native/src/client/mod.rs +++ b/clients/native/src/client/mod.rs @@ -1,6 +1,7 @@ // Copyright 2021 - Nym Technologies SA // SPDX-License-Identifier: Apache-2.0 +use client_connections::{ClosedConnectionReceiver, ClosedConnectionSender, TransmissionLane}; use client_core::client::cover_traffic_stream::LoopCoverTrafficStream; use client_core::client::inbound_messages::{ InputMessage, InputMessageReceiver, InputMessageSender, @@ -110,6 +111,7 @@ impl NymClient { stream.start_with_shutdown(shutdown); } + #[allow(clippy::too_many_arguments)] fn start_real_traffic_controller( &self, topology_accessor: TopologyAccessor, @@ -117,6 +119,7 @@ impl NymClient { ack_receiver: AcknowledgementReceiver, input_receiver: InputMessageReceiver, mix_sender: BatchMixMessageSender, + closed_connection_rx: ClosedConnectionReceiver, shutdown: ShutdownListener, ) { let mut controller_config = real_messages_control::Config::new( @@ -146,6 +149,7 @@ impl NymClient { mix_sender, topology_accessor, reply_key_storage, + closed_connection_rx, ) .start_with_shutdown(shutdown); } @@ -279,11 +283,16 @@ impl NymClient { &self, buffer_requester: ReceivedBufferRequestSender, msg_input: InputMessageSender, + closed_connection_tx: ClosedConnectionSender, ) { info!("Starting websocket listener..."); - let websocket_handler = - websocket::Handler::new(msg_input, buffer_requester, self.as_mix_recipient()); + let websocket_handler = websocket::Handler::new( + msg_input, + closed_connection_tx, + buffer_requester, + self.as_mix_recipient(), + ); websocket::Listener::new(self.config.get_listening_port()).start(websocket_handler); } @@ -292,7 +301,8 @@ impl NymClient { /// It's untested and there are absolutely no guarantees about it (but seems to have worked /// well enough in local tests) pub fn send_message(&mut self, recipient: Recipient, message: Vec, with_reply_surb: bool) { - let input_msg = InputMessage::new_fresh(recipient, message, with_reply_surb); + let lane = TransmissionLane::General; + let input_msg = InputMessage::new_fresh(recipient, message, with_reply_surb, lane); self.input_tx .as_ref() @@ -403,12 +413,17 @@ impl NymClient { let sphinx_message_sender = Self::start_mix_traffic_controller(gateway_client, shutdown.subscribe()); + // Channels that the websocket listener can use to signal downstream to the real traffic + // controller that connections are closed. + let (closed_connection_tx, closed_connection_rx) = mpsc::unbounded(); + self.start_real_traffic_controller( shared_topology_accessor.clone(), reply_key_storage, ack_receiver, input_receiver, sphinx_message_sender.clone(), + closed_connection_rx, shutdown.subscribe(), ); @@ -425,9 +440,11 @@ impl NymClient { } match self.config.get_socket_type() { - SocketType::WebSocket => { - self.start_websocket_listener(received_buffer_request_sender, input_sender) - } + SocketType::WebSocket => self.start_websocket_listener( + received_buffer_request_sender, + input_sender, + closed_connection_tx, + ), SocketType::None => { // if we did not start the socket, it means we're running (supposedly) in the native mode // and hence we should announce 'ourselves' to the buffer diff --git a/clients/native/src/websocket/handler.rs b/clients/native/src/websocket/handler.rs index fe23a518ab..f43fbca4c8 100644 --- a/clients/native/src/websocket/handler.rs +++ b/clients/native/src/websocket/handler.rs @@ -1,6 +1,7 @@ // Copyright 2021 - Nym Technologies SA // SPDX-License-Identifier: Apache-2.0 +use client_connections::{ClosedConnectionSender, TransmissionLane}; use client_core::client::{ inbound_messages::{InputMessage, InputMessageSender}, received_buffer::{ @@ -34,6 +35,7 @@ impl Default for ReceivedResponseType { pub(crate) struct Handler { msg_input: InputMessageSender, + closed_connection_tx: ClosedConnectionSender, buffer_requester: ReceivedBufferRequestSender, self_full_address: Recipient, socket: Option>, @@ -45,6 +47,7 @@ impl Clone for Handler { fn clone(&self) -> Self { Handler { msg_input: self.msg_input.clone(), + closed_connection_tx: self.closed_connection_tx.clone(), buffer_requester: self.buffer_requester.clone(), self_full_address: self.self_full_address, socket: None, @@ -64,11 +67,13 @@ impl Drop for Handler { impl Handler { pub(crate) fn new( msg_input: InputMessageSender, + closed_connection_tx: ClosedConnectionSender, buffer_requester: ReceivedBufferRequestSender, self_full_address: Recipient, ) -> Self { Handler { msg_input, + closed_connection_tx, buffer_requester, self_full_address, socket: None, @@ -81,9 +86,11 @@ impl Handler { recipient: Recipient, message: Vec, with_reply_surb: bool, + connection_id: u64, ) -> Option { // the ack control is now responsible for chunking, etc. - let input_msg = InputMessage::new_fresh(recipient, message, with_reply_surb); + let lane = TransmissionLane::ConnectionId(connection_id); + let input_msg = InputMessage::new_fresh(recipient, message, with_reply_surb, lane); self.msg_input.unbounded_send(input_msg).unwrap(); None @@ -104,18 +111,27 @@ impl Handler { ServerResponse::SelfAddress(self.self_full_address) } + fn handle_closed_connection(&self, connection_id: u64) -> Option { + self.closed_connection_tx + .unbounded_send(connection_id) + .unwrap(); + None + } + fn handle_request(&mut self, request: ClientRequest) -> Option { match request { ClientRequest::Send { recipient, message, with_reply_surb, - } => self.handle_send(recipient, message, with_reply_surb), + connection_id, + } => self.handle_send(recipient, message, with_reply_surb, connection_id), ClientRequest::Reply { message, reply_surb, } => self.handle_reply(reply_surb, message), ClientRequest::SelfAddress => Some(self.handle_self_address()), + ClientRequest::ClosedConnection(id) => self.handle_closed_connection(id), } } diff --git a/clients/native/websocket-requests/src/requests.rs b/clients/native/websocket-requests/src/requests.rs index d84349bc68..ea44059250 100644 --- a/clients/native/websocket-requests/src/requests.rs +++ b/clients/native/websocket-requests/src/requests.rs @@ -20,6 +20,9 @@ pub const REPLY_REQUEST_TAG: u8 = 0x01; /// Value tag representing [`SelfAddress`] variant of the [`ClientRequest`] pub const SELF_ADDRESS_REQUEST_TAG: u8 = 0x02; +/// Value tag representing [`ClosedConnection`] variant of the [`ClientRequest`] +pub const CLOSED_CONNECTION_REQUEST_TAG: u8 = 0x03; + #[allow(non_snake_case)] #[derive(Debug)] pub enum ClientRequest { @@ -28,23 +31,32 @@ pub enum ClientRequest { message: Vec, // Perhaps we could change it to a number to indicate how many reply_SURBs we want to include? with_reply_surb: bool, + connection_id: u64, }, Reply { message: Vec, reply_surb: ReplySurb, }, SelfAddress, + ClosedConnection(u64), } // we could have been parsing it directly TryFrom, but we want to retain // information about whether it came from binary or text to send appropriate response back impl ClientRequest { - // SEND_REQUEST_TAG || with_surb || recipient || data_len || data - fn serialize_send(recipient: Recipient, data: Vec, with_reply_surb: bool) -> Vec { + // SEND_REQUEST_TAG || with_surb || recipient || conn_id || data_len || data + fn serialize_send( + recipient: Recipient, + data: Vec, + with_reply_surb: bool, + connection_id: u64, + ) -> Vec { let data_len_bytes = (data.len() as u64).to_be_bytes(); + let conn_id_bytes = connection_id.to_be_bytes(); std::iter::once(SEND_REQUEST_TAG) .chain(std::iter::once(with_reply_surb as u8)) .chain(recipient.to_bytes().iter().cloned()) // will not be length prefixed because the length is constant + .chain(conn_id_bytes.iter().cloned()) .chain(data_len_bytes.iter().cloned()) .chain(data.into_iter()) .collect() @@ -86,9 +98,15 @@ impl ClientRequest { } }; - let data_len_bytes = &b[2 + Recipient::LEN..2 + Recipient::LEN + size_of::()]; + let mut connection_id_bytes = [0u8; size_of::()]; + connection_id_bytes + .copy_from_slice(&b[2 + Recipient::LEN..2 + Recipient::LEN + size_of::()]); + let connection_id = u64::from_be_bytes(connection_id_bytes); + + let data_len_bytes = + &b[2 + Recipient::LEN + size_of::()..2 + Recipient::LEN + 2 * size_of::()]; let data_len = u64::from_be_bytes(data_len_bytes.try_into().unwrap()); - let data = &b[2 + Recipient::LEN + size_of::()..]; + let data = &b[2 + Recipient::LEN + 2 * size_of::()..]; if data.len() as u64 != data_len { return Err(error::Error::new( ErrorKind::MalformedRequest, @@ -104,6 +122,7 @@ impl ClientRequest { with_reply_surb, recipient, message: data.to_vec(), + connection_id, }) } @@ -202,13 +221,34 @@ impl ClientRequest { ClientRequest::SelfAddress } + // CLOSED_CONNECTION_REQUEST_TAG + fn serialize_closed_connection(connection_id: u64) -> Vec { + let conn_id_bytes = connection_id.to_be_bytes(); + std::iter::once(CLOSED_CONNECTION_REQUEST_TAG) + .chain(conn_id_bytes.iter().cloned()) + .collect() + } + + // CLOSED_CONNECTION_REQUEST_TAG + fn deserialize_closed_connection(b: &[u8]) -> Self { + // this MUST match because it was called by 'deserialize' + debug_assert_eq!(b[0], CLOSED_CONNECTION_REQUEST_TAG); + + let mut connection_id_bytes = [0u8; size_of::()]; + connection_id_bytes.copy_from_slice(&b[1..=size_of::()]); + let connection_id = u64::from_be_bytes(connection_id_bytes); + + ClientRequest::ClosedConnection(connection_id) + } + pub fn serialize(self) -> Vec { match self { ClientRequest::Send { recipient, message, with_reply_surb, - } => Self::serialize_send(recipient, message, with_reply_surb), + connection_id, + } => Self::serialize_send(recipient, message, with_reply_surb, connection_id), ClientRequest::Reply { message, @@ -216,6 +256,8 @@ impl ClientRequest { } => Self::serialize_reply(message, reply_surb), ClientRequest::SelfAddress => Self::serialize_self_address(), + + ClientRequest::ClosedConnection(id) => Self::serialize_closed_connection(id), } } @@ -245,6 +287,7 @@ impl ClientRequest { SEND_REQUEST_TAG => Self::deserialize_send(b), REPLY_REQUEST_TAG => Self::deserialize_reply(b), SELF_ADDRESS_REQUEST_TAG => Ok(Self::deserialize_self_address(b)), + CLOSED_CONNECTION_REQUEST_TAG => Ok(Self::deserialize_closed_connection(b)), n => Err(error::Error::new( ErrorKind::UnknownRequest, format!("type {}", n), @@ -280,6 +323,7 @@ mod tests { recipient, message: b"foomp".to_vec(), with_reply_surb: false, + connection_id: 42, }; let bytes = send_request_no_surb.serialize(); @@ -289,10 +333,12 @@ mod tests { recipient, message, with_reply_surb, + connection_id, } => { assert_eq!(recipient.to_string(), recipient_string); assert_eq!(message, b"foomp".to_vec()); - assert!(!with_reply_surb) + assert!(!with_reply_surb); + assert_eq!(connection_id, 42) } _ => unreachable!(), } @@ -301,6 +347,7 @@ mod tests { recipient, message: b"foomp".to_vec(), with_reply_surb: true, + connection_id: 213, }; let bytes = send_request_surb.serialize(); @@ -310,10 +357,12 @@ mod tests { recipient, message, with_reply_surb, + connection_id, } => { assert_eq!(recipient.to_string(), recipient_string); assert_eq!(message, b"foomp".to_vec()); - assert!(with_reply_surb) + assert!(with_reply_surb); + assert_eq!(connection_id, 213) } _ => unreachable!(), } @@ -352,4 +401,15 @@ mod tests { _ => unreachable!(), } } + + #[test] + fn close_connection_request_serialization_works() { + let close_connection_request = ClientRequest::ClosedConnection(42); + let bytes = close_connection_request.serialize(); + let recovered = ClientRequest::deserialize(&bytes).unwrap(); + match recovered { + ClientRequest::ClosedConnection(id) => assert_eq!(id, 42), + _ => unreachable!(), + } + } } diff --git a/clients/native/websocket-requests/src/text.rs b/clients/native/websocket-requests/src/text.rs index 812774bd32..5a9f9dcf01 100644 --- a/clients/native/websocket-requests/src/text.rs +++ b/clients/native/websocket-requests/src/text.rs @@ -20,6 +20,7 @@ pub(super) enum ClientRequestText { message: String, recipient: String, with_reply_surb: bool, + connection_id: u64, }, SelfAddress, #[serde(rename_all = "camelCase")] @@ -46,6 +47,7 @@ impl TryInto for ClientRequestText { message, recipient, with_reply_surb, + connection_id, } => { let message_bytes = message.into_bytes(); let recipient = Recipient::try_from_base58_string(recipient).map_err(|err| { @@ -56,6 +58,7 @@ impl TryInto for ClientRequestText { message: message_bytes, recipient, with_reply_surb, + connection_id, }) } ClientRequestText::SelfAddress => Ok(ClientRequest::SelfAddress), diff --git a/clients/socks5/Cargo.toml b/clients/socks5/Cargo.toml index 87417bbd4a..ee103e7fda 100644 --- a/clients/socks5/Cargo.toml +++ b/clients/socks5/Cargo.toml @@ -26,6 +26,7 @@ url = "2.2" # internal client-core = { path = "../client-core" } +client-connections = { path = "../../common/client-connections" } coconut-interface = { path = "../../common/coconut-interface", optional = true } config = { path = "../../common/config" } completions = { path = "../../common/completions" } diff --git a/clients/socks5/src/client/mod.rs b/clients/socks5/src/client/mod.rs index 6992f688d4..5b8530e6d0 100644 --- a/clients/socks5/src/client/mod.rs +++ b/clients/socks5/src/client/mod.rs @@ -9,6 +9,7 @@ use crate::socks::{ authentication::{AuthenticationMethods, Authenticator, User}, server::SphinxSocksServer, }; +use client_connections::{ClosedConnectionReceiver, ClosedConnectionSender}; use client_core::client::cover_traffic_stream::LoopCoverTrafficStream; use client_core::client::inbound_messages::{ InputMessage, InputMessageReceiver, InputMessageSender, @@ -110,6 +111,7 @@ impl NymClient { stream.start_with_shutdown(shutdown); } + #[allow(clippy::too_many_arguments)] fn start_real_traffic_controller( &self, topology_accessor: TopologyAccessor, @@ -117,6 +119,7 @@ impl NymClient { ack_receiver: AcknowledgementReceiver, input_receiver: InputMessageReceiver, mix_sender: BatchMixMessageSender, + closed_connection_rx: ClosedConnectionReceiver, shutdown: ShutdownListener, ) { let mut controller_config = client_core::client::real_messages_control::Config::new( @@ -146,6 +149,7 @@ impl NymClient { mix_sender, topology_accessor, reply_key_storage, + closed_connection_rx, ) .start_with_shutdown(shutdown); } @@ -279,6 +283,7 @@ impl NymClient { &self, buffer_requester: ReceivedBufferRequestSender, msg_input: InputMessageSender, + closed_connection_tx: ClosedConnectionSender, shutdown: ShutdownListener, ) { info!("Starting socks5 listener..."); @@ -293,7 +298,11 @@ impl NymClient { self.as_mix_recipient(), shutdown, ); - tokio::spawn(async move { sphinx_socks.serve(msg_input, buffer_requester).await }); + tokio::spawn(async move { + sphinx_socks + .serve(msg_input, buffer_requester, closed_connection_tx) + .await + }); } /// blocking version of `start` method. Will run forever (or until SIGINT is sent) @@ -396,12 +405,17 @@ impl NymClient { let sphinx_message_sender = Self::start_mix_traffic_controller(gateway_client, shutdown.subscribe()); + // Channel for announcing closed (socks5) connections by the controller. + // This will be forwarded to `OutQueueControl` + let (closed_connection_tx, closed_connection_rx) = mpsc::unbounded(); + self.start_real_traffic_controller( shared_topology_accessor.clone(), reply_key_storage, ack_receiver, input_receiver, sphinx_message_sender.clone(), + closed_connection_rx, shutdown.subscribe(), ); @@ -420,6 +434,7 @@ impl NymClient { self.start_socks5_listener( received_buffer_request_sender, input_sender, + closed_connection_tx, shutdown.subscribe(), ); diff --git a/clients/socks5/src/socks/client.rs b/clients/socks5/src/socks/client.rs index 71ad3514ec..5188f7a15d 100644 --- a/clients/socks5/src/socks/client.rs +++ b/clients/socks5/src/socks/client.rs @@ -4,8 +4,8 @@ use super::authentication::{AuthenticationMethods, Authenticator, User}; use super::request::{SocksCommand, SocksRequest}; use super::types::{ResponseCode, SocksProxyError}; use super::{RESERVED, SOCKS_VERSION}; -use client_core::client::inbound_messages::InputMessage; -use client_core::client::inbound_messages::InputMessageSender; +use client_connections::TransmissionLane; +use client_core::client::inbound_messages::{InputMessage, InputMessageSender}; use futures::channel::mpsc; use futures::task::{Context, Poll}; use log::*; @@ -226,17 +226,21 @@ impl SocksClient { } } - async fn send_connect_to_mixnet(&mut self, remote_address: RemoteAddress) { + fn send_connect_to_mixnet(&mut self, remote_address: RemoteAddress) { let req = Request::new_connect(self.connection_id, remote_address, self.self_address); let msg = Message::Request(req); - let input_message = InputMessage::new_fresh(self.service_provider, msg.into_bytes(), false); + let input_message = InputMessage::new_fresh( + self.service_provider, + msg.into_bytes(), + false, + TransmissionLane::ConnectionId(self.connection_id), + ); self.input_sender.unbounded_send(input_message).unwrap(); } async fn run_proxy(&mut self, conn_receiver: ConnectionReceiver, remote_proxy_target: String) { - self.send_connect_to_mixnet(remote_proxy_target.clone()) - .await; + self.send_connect_to_mixnet(remote_proxy_target.clone()); let stream = self.stream.run_proxy(); let local_stream_remote = stream @@ -259,7 +263,8 @@ impl SocksClient { .run(move |conn_id, read_data, socket_closed| { let provider_request = Request::new_send(conn_id, read_data, socket_closed); let provider_message = Message::Request(provider_request); - InputMessage::new_fresh(recipient, provider_message.into_bytes(), false) + let lane = TransmissionLane::ConnectionId(conn_id); + InputMessage::new_fresh(recipient, provider_message.into_bytes(), false, lane) }) .await .into_inner(); diff --git a/clients/socks5/src/socks/server.rs b/clients/socks5/src/socks/server.rs index c8feebbfb1..11912d0a0d 100644 --- a/clients/socks5/src/socks/server.rs +++ b/clients/socks5/src/socks/server.rs @@ -4,6 +4,7 @@ use super::{ mixnet_responses::MixnetResponseListener, types::{ResponseCode, SocksProxyError}, }; +use client_connections::ClosedConnectionSender; use client_core::client::{ inbound_messages::InputMessageSender, received_buffer::ReceivedBufferRequestSender, }; @@ -51,13 +52,14 @@ impl SphinxSocksServer { &mut self, input_sender: InputMessageSender, buffer_requester: ReceivedBufferRequestSender, + closed_connection_tx: ClosedConnectionSender, ) -> Result<(), SocksProxyError> { let listener = TcpListener::bind(self.listening_address).await.unwrap(); info!("Serving Connections..."); // controller for managing all active connections let (mut active_streams_controller, controller_sender) = - Controller::new(self.shutdown.clone()); + Controller::new(closed_connection_tx, self.shutdown.clone()); tokio::spawn(async move { active_streams_controller.run().await; }); diff --git a/clients/webassembly/Cargo.toml b/clients/webassembly/Cargo.toml index 9f729a660b..b296e29959 100644 --- a/clients/webassembly/Cargo.toml +++ b/clients/webassembly/Cargo.toml @@ -29,6 +29,7 @@ url = "2.2" # internal client-core = { path = "../client-core", default-features = false, features = ["wasm"] } +client-connections = { path = "../../common/client-connections" } coconut-interface = { path = "../../common/coconut-interface", optional = true } credentials = { path = "../../common/credentials", optional = true } crypto = { path = "../../common/crypto" } @@ -59,4 +60,4 @@ wasm-opt = true [profile.release] lto = true -opt-level = 'z' \ No newline at end of file +opt-level = 'z' diff --git a/clients/webassembly/src/client/mod.rs b/clients/webassembly/src/client/mod.rs index 7008805186..7aed6c9a3b 100644 --- a/clients/webassembly/src/client/mod.rs +++ b/clients/webassembly/src/client/mod.rs @@ -2,6 +2,7 @@ // SPDX-License-Identifier: Apache-2.0 use self::config::Config; +use client_connections::{ClosedConnectionReceiver, TransmissionLane}; use client_core::client::{ cover_traffic_stream::LoopCoverTrafficStream, inbound_messages::{InputMessage, InputMessageReceiver, InputMessageSender}, @@ -127,6 +128,7 @@ impl NymClient { ack_receiver: AcknowledgementReceiver, input_receiver: InputMessageReceiver, mix_sender: BatchMixMessageSender, + closed_connection_rx: ClosedConnectionReceiver, ) { let mut controller_config = real_messages_control::Config::new( self.key_manager.ack_key(), @@ -151,6 +153,7 @@ impl NymClient { input_receiver, mix_sender, topology_accessor, + closed_connection_rx, ) .start(); } @@ -327,6 +330,10 @@ impl NymClient { let (ack_sender, ack_receiver) = mpsc::unbounded(); let shared_topology_accessor = TopologyAccessor::new(); + // Channel that the real traffix controller can listed to for closing connections. + // Currently unused in the wasm client. + let (_closed_connection_tx, closed_connection_rx) = mpsc::unbounded(); + // the components are started in very specific order. Unless you know what you are doing, // do not change that. self.start_topology_refresher(shared_topology_accessor.clone()) @@ -351,6 +358,7 @@ impl NymClient { ack_receiver, input_receiver, sphinx_message_sender.clone(), + closed_connection_rx, ); if !self.config.debug.disable_loop_cover_traffic_stream { @@ -376,8 +384,9 @@ impl NymClient { console_log!("Sending {} bytes to {}", message.len(), recipient); let recipient = Recipient::try_from_base58_string(recipient).unwrap(); + let lane = TransmissionLane::General; - let input_msg = InputMessage::new_fresh(recipient, message, false); + let input_msg = InputMessage::new_fresh(recipient, message, false, lane); self.input_tx .as_ref() diff --git a/common/client-connections/Cargo.toml b/common/client-connections/Cargo.toml new file mode 100644 index 0000000000..067621c28f --- /dev/null +++ b/common/client-connections/Cargo.toml @@ -0,0 +1,9 @@ +[package] +name = "client-connections" +version = "0.1.0" +edition = "2021" + +# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html + +[dependencies] +futures = "0.3" diff --git a/common/client-connections/src/lib.rs b/common/client-connections/src/lib.rs new file mode 100644 index 0000000000..2aabce78b0 --- /dev/null +++ b/common/client-connections/src/lib.rs @@ -0,0 +1,21 @@ +// Copyright 2022 - Nym Technologies SA +// SPDX-License-Identifier: Apache-2.0 + +use futures::channel::mpsc; + +pub type ConnectionId = u64; + +#[derive(Copy, Clone, Debug, Hash, PartialEq, Eq)] +pub enum TransmissionLane { + General, + Reply, + Retransmission, + Control, + ConnectionId(ConnectionId), +} + +/// Announce connections that are closed, for whoever is interested. +/// One usecase is that the network-requester and socks5-client wants to know about this, so that +/// they can forward this to the `OutQueueControl` (via `ClientRequest` for the network-requester) +pub type ClosedConnectionSender = mpsc::UnboundedSender; +pub type ClosedConnectionReceiver = mpsc::UnboundedReceiver; diff --git a/common/socks5/proxy-helpers/Cargo.toml b/common/socks5/proxy-helpers/Cargo.toml index fd49c094ef..2ecd2a0507 100644 --- a/common/socks5/proxy-helpers/Cargo.toml +++ b/common/socks5/proxy-helpers/Cargo.toml @@ -14,8 +14,11 @@ tokio-util = { version = "0.7.3", features = [ "io" ] } # reason for getting thi # In the long run, the dependency should probably get removed in favour of pure-tokio implementation, but for time being it's fine. futures = "0.3" log = "0.4" -socks5-requests = { path = "../requests" } + +# internal +client-connections = { path = "../../client-connections" } ordered-buffer = { path = "../ordered-buffer" } +socks5-requests = { path = "../requests" } task = { path = "../../task" } [dev-dependencies] diff --git a/common/socks5/proxy-helpers/src/connection_controller.rs b/common/socks5/proxy-helpers/src/connection_controller.rs index 1a9b13cc29..16b370058d 100644 --- a/common/socks5/proxy-helpers/src/connection_controller.rs +++ b/common/socks5/proxy-helpers/src/connection_controller.rs @@ -1,6 +1,7 @@ // Copyright 2021 - Nym Technologies SA // SPDX-License-Identifier: Apache-2.0 +use client_connections::ClosedConnectionSender; use futures::channel::mpsc; use futures::StreamExt; use log::*; @@ -73,6 +74,9 @@ pub struct Controller { // to avoid memory issues recently_closed: HashSet, + // Broadcast closed connections + closed_connection_tx: ClosedConnectionSender, + // TODO: this can potentially be abused to ddos and kill provider. Not sure at this point // how to handle it more gracefully @@ -84,13 +88,17 @@ pub struct Controller { } impl Controller { - pub fn new(shutdown: ShutdownListener) -> (Self, ControllerSender) { + pub fn new( + closed_connection_tx: ClosedConnectionSender, + shutdown: ShutdownListener, + ) -> (Self, ControllerSender) { let (sender, receiver) = mpsc::unbounded(); ( Controller { active_connections: HashMap::new(), receiver, recently_closed: HashSet::new(), + closed_connection_tx, pending_messages: HashMap::new(), shutdown, }, @@ -127,6 +135,9 @@ impl Controller { ) } self.recently_closed.insert(conn_id); + + // Announce closed connections, currently used by the `OutQueueControl`. + self.closed_connection_tx.unbounded_send(conn_id).unwrap(); } fn send_to_connection(&mut self, conn_id: ConnectionId, payload: Vec, is_closed: bool) { @@ -172,11 +183,15 @@ impl Controller { pending.push((payload, is_closed)); } else if !is_closed { error!( - "Tried to write to closed connection ({} bytes were 'lost)", + "Tried to write to closed connection {} ({} bytes were 'lost)", + conn_id, payload.len() ); } else { - debug!("Tried to write to closed connection, but remote is already closed") + debug!( + "Tried to write to closed connection {}, but remote is already closed", + conn_id + ) } } diff --git a/common/socks5/proxy-helpers/src/proxy_runner/inbound.rs b/common/socks5/proxy-helpers/src/proxy_runner/inbound.rs index 5efc297ce7..e116ba1dc5 100644 --- a/common/socks5/proxy-helpers/src/proxy_runner/inbound.rs +++ b/common/socks5/proxy-helpers/src/proxy_runner/inbound.rs @@ -63,6 +63,10 @@ where // if we're sending through the mixnet increase the sequence number... let ordered_msg = message_sender.wrap_message(read_data.to_vec()).into_bytes(); + log::trace!( + "pushing data down the input sender: size: {}", + ordered_msg.len() + ); mix_sender .unbounded_send(adapter_fn(connection_id, ordered_msg, is_finished)) .unwrap(); diff --git a/nym-connect/Cargo.lock b/nym-connect/Cargo.lock index 0d0022042e..c60bb9b66c 100644 --- a/nym-connect/Cargo.lock +++ b/nym-connect/Cargo.lock @@ -594,10 +594,18 @@ dependencies = [ "os_str_bytes", ] +[[package]] +name = "client-connections" +version = "0.1.0" +dependencies = [ + "futures", +] + [[package]] name = "client-core" version = "1.0.1" dependencies = [ + "client-connections", "config", "crypto", "dirs", @@ -3250,6 +3258,7 @@ name = "nym-socks5-client" version = "1.1.0" dependencies = [ "clap", + "client-connections", "client-core", "completions", "config", @@ -4034,6 +4043,7 @@ name = "proxy-helpers" version = "0.1.0" dependencies = [ "bytes", + "client-connections", "futures", "log", "ordered-buffer", diff --git a/service-providers/network-requester/Cargo.toml b/service-providers/network-requester/Cargo.toml index 7ea81c1aa3..eb3cc1d229 100644 --- a/service-providers/network-requester/Cargo.toml +++ b/service-providers/network-requester/Cargo.toml @@ -28,9 +28,10 @@ tokio-tungstenite = "0.17.2" # internal +client-connections = { path = "../../common/client-connections" } +completions = { path = "../../common/completions" } network-defaults = { path = "../../common/network-defaults" } nymsphinx = { path = "../../common/nymsphinx" } -completions = { path = "../../common/completions" } logging = { path = "../../common/logging"} ordered-buffer = {path = "../../common/socks5/ordered-buffer"} proxy-helpers = { path = "../../common/socks5/proxy-helpers" } diff --git a/service-providers/network-requester/src/core.rs b/service-providers/network-requester/src/core.rs index 0c50a665fb..c9333d6bf6 100644 --- a/service-providers/network-requester/src/core.rs +++ b/service-providers/network-requester/src/core.rs @@ -7,6 +7,7 @@ use crate::error::NetworkRequesterError; use crate::statistics::ServiceStatisticsCollector; use crate::websocket; use crate::websocket::TSWebsocketStream; +use client_connections::ClosedConnectionReceiver; use futures::channel::mpsc; use futures::stream::{SplitSink, SplitStream}; use futures::{SinkExt, StreamExt}; @@ -68,32 +69,50 @@ impl ServiceProvider { mut websocket_writer: SplitSink, mut mix_reader: mpsc::UnboundedReceiver<(Socks5Message, Recipient)>, stats_collector: Option, + mut closed_connection_rx: ClosedConnectionReceiver, ) { - // TODO: wire SURBs in here once they're available - while let Some((msg, return_address)) = mix_reader.next().await { - if let Some(stats_collector) = stats_collector.as_ref() { - if let Some(remote_addr) = stats_collector - .connected_services - .read() - .await - .get(&msg.conn_id()) - { - stats_collector - .response_stats_data - .write() - .await - .processed(remote_addr, msg.size() as u32); + loop { + tokio::select! { + // TODO: wire SURBs in here once they're available + socks5_msg = mix_reader.next() => { + if let Some((msg, return_address)) = socks5_msg { + if let Some(stats_collector) = stats_collector.as_ref() { + if let Some(remote_addr) = stats_collector + .connected_services + .read() + .await + .get(&msg.conn_id()) + { + stats_collector + .response_stats_data + .write() + .await + .processed(remote_addr, msg.size() as u32); + } + } + let conn_id = msg.conn_id(); + + // make 'request' to native-websocket client + let response_message = ClientRequest::Send { + recipient: return_address, + message: msg.into_bytes(), + with_reply_surb: false, + connection_id: conn_id, + }; + + let message = Message::Binary(response_message.serialize()); + websocket_writer.send(message).await.unwrap(); + } else { + log::error!("Exiting: channel closed!"); + break; + } + }, + Some(id) = closed_connection_rx.next() => { + let msg = ClientRequest::ClosedConnection(id); + let ws_msg = Message::Binary(msg.serialize()); + websocket_writer.send(ws_msg).await.unwrap(); } } - // make 'request' to native-websocket client - let response_message = ClientRequest::Send { - recipient: return_address, - message: msg.into_bytes(), - with_reply_surb: false, - }; - - let message = Message::Binary(response_message.serialize()); - websocket_writer.send(message).await.unwrap(); } } @@ -309,12 +328,20 @@ impl ServiceProvider { let (mix_input_sender, mix_input_receiver) = mpsc::unbounded::<(Socks5Message, Recipient)>(); + // Used to notify tasks to shutdown. Not all tasks fully supports this (yet). + let shutdown = task::ShutdownNotifier::default(); + + // Channel for announcing closed (socks5) connections by the controller. + // The `mixnet_response_listener` will forward this info to the client using a + // `ClientRequest`. + let (closed_connection_tx, closed_connection_rx) = mpsc::unbounded(); + // Controller for managing all active connections. // We provide it with a ShutdownListener since it requires it, even though for the network // requester shutdown signalling is not yet fully implemented. - let shutdown = task::ShutdownNotifier::default(); let (mut active_connections_controller, mut controller_sender) = - Controller::new(shutdown.subscribe()); + Controller::new(closed_connection_tx, shutdown.subscribe()); + tokio::spawn(async move { active_connections_controller.run().await; }); @@ -341,6 +368,7 @@ impl ServiceProvider { websocket_writer, mix_input_receiver, stats_collector_clone, + closed_connection_rx, ) .await; });