GRIN/nym
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

WIP: backpressure in out queue control

Browse Source
This commit is contained in:
Jon Häggblad
2022-10-19 00:15:19 +02:00
parent 73f3552b52
commit 7058e7139d
13 changed files with 188 additions and 20 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Cargo.lock
Generated
+1
View File
@@ -3199,6 +3199,7 @@ dependencies = [
"colored",
"completions",
"config",
"console-subscriber",
"credentials",
"crypto",
"dashmap",
clients/client-core/src/client/cover_traffic_stream.rs
+4 -1
View File
@@ -176,7 +176,9 @@ impl LoopCoverTrafficStream<OsRng> {
// - 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<OsRng> {
}
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");
clients/client-core/src/client/mix_traffic.rs
+1
View File
@@ -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 => {
clients/client-core/src/client/real_messages_control/real_traffic_stream.rs
+158 -9
View File
@@ -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<Pin<Box<wasm_timer::Delay>>>,
/// 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<Item = StreamMessage> + '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<R> Stream for OutQueueControl<R>
where
R: CryptoRng + Rng + Unpin,
clients/client-core/src/lib.rs
+4 -4
View File
@@ -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<F>(future: F, task_name: &str)
clients/native/src/client/mod.rs
+1 -1
View File
@@ -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
clients/socks5/src/client/mod.rs
+1 -1
View File
@@ -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
clients/socks5/src/main.rs
+1 -1
View File
@@ -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());
common/client-libs/gateway-client/Cargo.toml
+2 -1
View File
@@ -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"
common/client-libs/gateway-client/src/client.rs
+7 -1
View File
@@ -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<MixPacket>,
) -> 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);
}
gateway/Cargo.toml
+1
View File
@@ -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 }
gateway/src/main.rs
+3
View File
@@ -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
gateway/src/node/mixnet_handling/receiver/connection_handler.rs
+4 -1
View File
@@ -120,7 +120,7 @@ impl<St: Storage> ConnectionHandler<St> {
fn forward_ack(&self, forward_ack: Option<MixPacket>, 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<St: Storage> ConnectionHandler<St> {
}
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<St: Storage> ConnectionHandler<St> {
// 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) => {