From 9ee6ae44e231266be2b3baef7fa19f0d775f9cef Mon Sep 17 00:00:00 2001 From: Simon Wicky Date: Wed, 16 Nov 2022 08:58:22 +0000 Subject: [PATCH 1/5] remove override config, and active nodes --- clients/client-core/src/client/topology_control.rs | 2 +- clients/native/src/commands/run.rs | 2 +- clients/socks5/src/commands/run.rs | 2 +- 3 files changed, 3 insertions(+), 3 deletions(-) diff --git a/clients/client-core/src/client/topology_control.rs b/clients/client-core/src/client/topology_control.rs index ad9e4a4ce6..1c6b9e7643 100644 --- a/clients/client-core/src/client/topology_control.rs +++ b/clients/client-core/src/client/topology_control.rs @@ -256,7 +256,7 @@ impl TopologyRefresher { // only refresh mixnodes on timer and refresh gateways only when // we have to send to a new, unknown, gateway - let mixnodes = match self.validator_client.get_cached_active_mixnodes().await { + let mixnodes = match self.validator_client.get_cached_mixnodes().await { Err(err) => { error!("failed to get network mixnodes - {err}"); return None; diff --git a/clients/native/src/commands/run.rs b/clients/native/src/commands/run.rs index 5eb6939944..c9af755052 100644 --- a/clients/native/src/commands/run.rs +++ b/clients/native/src/commands/run.rs @@ -101,7 +101,7 @@ pub(crate) async fn execute(args: &Run) -> Result<(), ClientError> { }; let override_config_fields = OverrideConfig::from(args.clone()); - config = override_config(config, override_config_fields); + //config = override_config(config, override_config_fields); if config.get_base_mut().set_empty_fields_to_defaults() { warn!("some of the core config options were left unset. the default values are going to get used instead."); diff --git a/clients/socks5/src/commands/run.rs b/clients/socks5/src/commands/run.rs index 48d4651d15..7f50bea28c 100644 --- a/clients/socks5/src/commands/run.rs +++ b/clients/socks5/src/commands/run.rs @@ -118,7 +118,7 @@ pub(crate) async fn execute(args: &Run) -> Result<(), Box Date: Tue, 22 Nov 2022 13:37:21 +0000 Subject: [PATCH 2/5] add packet sizes support --- clients/client-core/src/config/mod.rs | 21 +++++ .../gateway-client/src/packet_router.rs | 2 +- .../src/packet_processor/processor.rs | 5 +- common/nymsphinx/params/src/packet_sizes.rs | 87 ++++++++++++++++++- common/nymsphinx/params/src/packet_version.rs | 10 +++ 5 files changed, 118 insertions(+), 7 deletions(-) diff --git a/clients/client-core/src/config/mod.rs b/clients/client-core/src/config/mod.rs index 0ec88a8095..08613614b8 100644 --- a/clients/client-core/src/config/mod.rs +++ b/clients/client-core/src/config/mod.rs @@ -635,6 +635,17 @@ pub enum ExtendedPacketSize { Extended8, Extended16, Extended32, + Extended10, + Extended15, + Extended20, + Extended25, + Extended50, + Extended100, + Extended150, + Extended200, + Extended250, + Extended500, + } impl Default for DebugConfig { @@ -670,6 +681,16 @@ impl From for PacketSize { ExtendedPacketSize::Extended8 => PacketSize::ExtendedPacket8, ExtendedPacketSize::Extended16 => PacketSize::ExtendedPacket16, ExtendedPacketSize::Extended32 => PacketSize::ExtendedPacket32, + ExtendedPacketSize::Extended10 => PacketSize::ExtendedPacket10, + ExtendedPacketSize::Extended15 => PacketSize::ExtendedPacket15, + ExtendedPacketSize::Extended20 => PacketSize::ExtendedPacket20, + ExtendedPacketSize::Extended25 => PacketSize::ExtendedPacket25, + ExtendedPacketSize::Extended50 => PacketSize::ExtendedPacket50, + ExtendedPacketSize::Extended100 => PacketSize::ExtendedPacket100, + ExtendedPacketSize::Extended150 => PacketSize::ExtendedPacket150, + ExtendedPacketSize::Extended200 => PacketSize::ExtendedPacket200, + ExtendedPacketSize::Extended250 => PacketSize::ExtendedPacket250, + ExtendedPacketSize::Extended500 => PacketSize::ExtendedPacket500, } } } diff --git a/common/client-libs/gateway-client/src/packet_router.rs b/common/client-libs/gateway-client/src/packet_router.rs index fd63077ee6..0c7b5f99da 100644 --- a/common/client-libs/gateway-client/src/packet_router.rs +++ b/common/client-libs/gateway-client/src/packet_router.rs @@ -74,7 +74,7 @@ impl PacketRouter { received_messages.push(received_packet); } else { // this can happen if other clients are not padding their messages - warn!("Received message of unexpected size. Probably from an outdated client... len: {}", received_packet.len()); + //warn!("Received message of unexpected size. Probably from an outdated client... len: {}", received_packet.len()); received_messages.push(received_packet); } } diff --git a/common/mixnode-common/src/packet_processor/processor.rs b/common/mixnode-common/src/packet_processor/processor.rs index aaadfe0171..6cd9178a4a 100644 --- a/common/mixnode-common/src/packet_processor/processor.rs +++ b/common/mixnode-common/src/packet_processor/processor.rs @@ -116,10 +116,7 @@ impl SphinxPacketProcessor { trace!("received an ack packet!"); Ok((None, data)) } - PacketSize::RegularPacket - | PacketSize::ExtendedPacket8 - | PacketSize::ExtendedPacket16 - | PacketSize::ExtendedPacket32 => { + _ => { trace!("received a normal packet!"); let (ack_data, message) = self.split_hop_data_into_ack_and_message(data)?; let (ack_first_hop, ack_packet) = SurbAck::try_recover_first_hop_packet(&ack_data)?; diff --git a/common/nymsphinx/params/src/packet_sizes.rs b/common/nymsphinx/params/src/packet_sizes.rs index aa573ffe11..bb7d897bb7 100644 --- a/common/nymsphinx/params/src/packet_sizes.rs +++ b/common/nymsphinx/params/src/packet_sizes.rs @@ -20,6 +20,17 @@ const EXTENDED_PACKET_SIZE_8: usize = HEADER_SIZE + PAYLOAD_OVERHEAD_SIZE + 8 * const EXTENDED_PACKET_SIZE_16: usize = HEADER_SIZE + PAYLOAD_OVERHEAD_SIZE + 16 * 1024; const EXTENDED_PACKET_SIZE_32: usize = HEADER_SIZE + PAYLOAD_OVERHEAD_SIZE + 32 * 1024; +const EXTENDED_PACKET_SIZE_10: usize = HEADER_SIZE + PAYLOAD_OVERHEAD_SIZE + 10 * 1024; +const EXTENDED_PACKET_SIZE_15: usize = HEADER_SIZE + PAYLOAD_OVERHEAD_SIZE + 15 * 1024; +const EXTENDED_PACKET_SIZE_20: usize = HEADER_SIZE + PAYLOAD_OVERHEAD_SIZE + 20 * 1024; +const EXTENDED_PACKET_SIZE_25: usize = HEADER_SIZE + PAYLOAD_OVERHEAD_SIZE + 25 * 1024; +const EXTENDED_PACKET_SIZE_50: usize = HEADER_SIZE + PAYLOAD_OVERHEAD_SIZE + 50 * 1024; +const EXTENDED_PACKET_SIZE_100: usize = HEADER_SIZE + PAYLOAD_OVERHEAD_SIZE + 100 * 1024; +const EXTENDED_PACKET_SIZE_150: usize = HEADER_SIZE + PAYLOAD_OVERHEAD_SIZE + 150 * 1024; +const EXTENDED_PACKET_SIZE_200: usize = HEADER_SIZE + PAYLOAD_OVERHEAD_SIZE + 200 * 1024; +const EXTENDED_PACKET_SIZE_250: usize = HEADER_SIZE + PAYLOAD_OVERHEAD_SIZE + 250 * 1024; +const EXTENDED_PACKET_SIZE_500: usize = HEADER_SIZE + PAYLOAD_OVERHEAD_SIZE + 500 * 1024; + #[derive(Debug)] pub struct InvalidPacketSize; @@ -43,6 +54,18 @@ pub enum PacketSize { // for example for streaming fast and furious in compressed XviD quality ExtendedPacket16 = 5, + + ExtendedPacket10 = 6, + ExtendedPacket15 = 7, + ExtendedPacket20 = 8, + + ExtendedPacket25 = 9, + ExtendedPacket50 = 10, + ExtendedPacket100 = 11, + ExtendedPacket150 = 12, + ExtendedPacket200 = 13, + ExtendedPacket250 = 14, + ExtendedPacket500 = 15, } impl FromStr for PacketSize { @@ -55,6 +78,16 @@ impl FromStr for PacketSize { "extended8" => Ok(Self::ExtendedPacket8), "extended16" => Ok(Self::ExtendedPacket16), "extended32" => Ok(Self::ExtendedPacket32), + "extended10" => Ok(Self::ExtendedPacket10), + "extended15" => Ok(Self::ExtendedPacket15), + "extended20" => Ok(Self::ExtendedPacket20), + "extended25" => Ok(Self::ExtendedPacket25), + "extended50" => Ok(Self::ExtendedPacket50), + "extended100" => Ok(Self::ExtendedPacket100), + "extended150" => Ok(Self::ExtendedPacket150), + "extended200" => Ok(Self::ExtendedPacket200), + "extended250" => Ok(Self::ExtendedPacket250), + "extended500" => Ok(Self::ExtendedPacket500), _ => Err(InvalidPacketSize), } } @@ -70,6 +103,16 @@ impl TryFrom for PacketSize { _ if value == (PacketSize::ExtendedPacket8 as u8) => Ok(Self::ExtendedPacket8), _ if value == (PacketSize::ExtendedPacket16 as u8) => Ok(Self::ExtendedPacket16), _ if value == (PacketSize::ExtendedPacket32 as u8) => Ok(Self::ExtendedPacket32), + _ if value == (PacketSize::ExtendedPacket10 as u8) => Ok(Self::ExtendedPacket10), + _ if value == (PacketSize::ExtendedPacket15 as u8) => Ok(Self::ExtendedPacket15), + _ if value == (PacketSize::ExtendedPacket20 as u8) => Ok(Self::ExtendedPacket20), + _ if value == (PacketSize::ExtendedPacket25 as u8) => Ok(Self::ExtendedPacket25), + _ if value == (PacketSize::ExtendedPacket50 as u8) => Ok(Self::ExtendedPacket50), + _ if value == (PacketSize::ExtendedPacket100 as u8) => Ok(Self::ExtendedPacket100), + _ if value == (PacketSize::ExtendedPacket150 as u8) => Ok(Self::ExtendedPacket150), + _ if value == (PacketSize::ExtendedPacket200 as u8) => Ok(Self::ExtendedPacket200), + _ if value == (PacketSize::ExtendedPacket250 as u8) => Ok(Self::ExtendedPacket250), + _ if value == (PacketSize::ExtendedPacket500 as u8) => Ok(Self::ExtendedPacket500), _ => Err(InvalidPacketSize), } } @@ -83,6 +126,16 @@ impl PacketSize { PacketSize::ExtendedPacket8 => EXTENDED_PACKET_SIZE_8, PacketSize::ExtendedPacket16 => EXTENDED_PACKET_SIZE_16, PacketSize::ExtendedPacket32 => EXTENDED_PACKET_SIZE_32, + PacketSize::ExtendedPacket10 => EXTENDED_PACKET_SIZE_10, + PacketSize::ExtendedPacket15 => EXTENDED_PACKET_SIZE_15, + PacketSize::ExtendedPacket20 => EXTENDED_PACKET_SIZE_20, + PacketSize::ExtendedPacket25 => EXTENDED_PACKET_SIZE_25, + PacketSize::ExtendedPacket50 => EXTENDED_PACKET_SIZE_50, + PacketSize::ExtendedPacket100 => EXTENDED_PACKET_SIZE_100, + PacketSize::ExtendedPacket150 => EXTENDED_PACKET_SIZE_150, + PacketSize::ExtendedPacket200 => EXTENDED_PACKET_SIZE_200, + PacketSize::ExtendedPacket250 => EXTENDED_PACKET_SIZE_250, + PacketSize::ExtendedPacket500 => EXTENDED_PACKET_SIZE_500, } } @@ -105,6 +158,26 @@ impl PacketSize { Ok(PacketSize::ExtendedPacket16) } else if PacketSize::ExtendedPacket32.size() == size { Ok(PacketSize::ExtendedPacket32) + } else if PacketSize::ExtendedPacket10.size() == size { + Ok(PacketSize::ExtendedPacket10) + } else if PacketSize::ExtendedPacket15.size() == size { + Ok(PacketSize::ExtendedPacket15) + } else if PacketSize::ExtendedPacket20.size() == size { + Ok(PacketSize::ExtendedPacket20) + } else if PacketSize::ExtendedPacket25.size() == size { + Ok(PacketSize::ExtendedPacket25) + } else if PacketSize::ExtendedPacket50.size() == size { + Ok(PacketSize::ExtendedPacket50) + } else if PacketSize::ExtendedPacket100.size() == size { + Ok(PacketSize::ExtendedPacket100) + } else if PacketSize::ExtendedPacket150.size() == size { + Ok(PacketSize::ExtendedPacket150) + } else if PacketSize::ExtendedPacket200.size() == size { + Ok(PacketSize::ExtendedPacket200) + } else if PacketSize::ExtendedPacket250.size() == size { + Ok(PacketSize::ExtendedPacket250) + } else if PacketSize::ExtendedPacket500.size() == size { + Ok(PacketSize::ExtendedPacket500) } else { Err(InvalidPacketSize) } @@ -115,7 +188,17 @@ impl PacketSize { PacketSize::RegularPacket | PacketSize::AckPacket => false, PacketSize::ExtendedPacket8 | PacketSize::ExtendedPacket16 - | PacketSize::ExtendedPacket32 => true, + | PacketSize::ExtendedPacket32 + | PacketSize::ExtendedPacket10 + | PacketSize::ExtendedPacket15 + | PacketSize::ExtendedPacket20 + | PacketSize::ExtendedPacket25 + | PacketSize::ExtendedPacket50 + | PacketSize::ExtendedPacket100 + | PacketSize::ExtendedPacket150 + | PacketSize::ExtendedPacket200 + | PacketSize::ExtendedPacket250 + | PacketSize::ExtendedPacket500 => true, } } @@ -130,6 +213,6 @@ impl PacketSize { impl Default for PacketSize { fn default() -> Self { - PacketSize::RegularPacket + PacketSize::ExtendedPacket100 } } diff --git a/common/nymsphinx/params/src/packet_version.rs b/common/nymsphinx/params/src/packet_version.rs index 001f46c84d..96152f426d 100644 --- a/common/nymsphinx/params/src/packet_version.rs +++ b/common/nymsphinx/params/src/packet_version.rs @@ -47,6 +47,16 @@ impl From for PacketVersion { n if n == PacketSize::ExtendedPacket8 as u8 => PacketVersion::Legacy, n if n == PacketSize::ExtendedPacket16 as u8 => PacketVersion::Legacy, n if n == PacketSize::ExtendedPacket32 as u8 => PacketVersion::Legacy, + n if n == PacketSize::ExtendedPacket10 as u8 => PacketVersion::Legacy, + n if n == PacketSize::ExtendedPacket15 as u8 => PacketVersion::Legacy, + n if n == PacketSize::ExtendedPacket20 as u8 => PacketVersion::Legacy, + n if n == PacketSize::ExtendedPacket25 as u8 => PacketVersion::Legacy, + n if n == PacketSize::ExtendedPacket50 as u8 => PacketVersion::Legacy, + n if n == PacketSize::ExtendedPacket100 as u8 => PacketVersion::Legacy, + n if n == PacketSize::ExtendedPacket150 as u8 => PacketVersion::Legacy, + n if n == PacketSize::ExtendedPacket200 as u8 => PacketVersion::Legacy, + n if n == PacketSize::ExtendedPacket250 as u8 => PacketVersion::Legacy, + n if n == PacketSize::ExtendedPacket500 as u8 => PacketVersion::Legacy, n => PacketVersion::Versioned(n), } } From 2c2823f9e1d6df3b10f3370d42c6720c22295c71 Mon Sep 17 00:00:00 2001 From: Simon Wicky Date: Thu, 24 Nov 2022 08:23:48 +0000 Subject: [PATCH 3/5] test instrumentation --- Cargo.lock | 18 +-- .../acknowledgement_listener.rs | 7 + .../real_traffic_stream.rs | 120 +++++++++++++++--- .../client-core/src/client/received_buffer.rs | 8 ++ clients/socks5/src/socks/client.rs | 13 ++ 5 files changed, 140 insertions(+), 26 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 822fa26726..2256ba25f0 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -586,7 +586,7 @@ dependencies = [ [[package]] name = "client-core" -version = "1.1.2" +version = "1.1.4" dependencies = [ "async-trait", "client-connections", @@ -3106,7 +3106,7 @@ dependencies = [ [[package]] name = "nym-cli" -version = "1.1.2" +version = "1.1.4" dependencies = [ "anyhow", "base64", @@ -3160,7 +3160,7 @@ dependencies = [ [[package]] name = "nym-client" -version = "1.1.2" +version = "1.1.4" dependencies = [ "clap 3.2.8", "client-connections", @@ -3199,7 +3199,7 @@ dependencies = [ [[package]] name = "nym-gateway" -version = "1.1.2" +version = "1.1.4" dependencies = [ "anyhow", "async-trait", @@ -3246,7 +3246,7 @@ dependencies = [ [[package]] name = "nym-mixnode" -version = "1.1.2" +version = "1.1.4" dependencies = [ "anyhow", "bs58", @@ -3288,7 +3288,7 @@ dependencies = [ [[package]] name = "nym-network-requester" -version = "1.1.2" +version = "1.1.4" dependencies = [ "async-trait", "clap 3.2.8", @@ -3320,7 +3320,7 @@ dependencies = [ [[package]] name = "nym-network-statistics" -version = "1.1.2" +version = "1.1.4" dependencies = [ "dirs", "log", @@ -3336,7 +3336,7 @@ dependencies = [ [[package]] name = "nym-socks5-client" -version = "1.1.2" +version = "1.1.4" dependencies = [ "clap 3.2.8", "client-connections", @@ -3403,7 +3403,7 @@ dependencies = [ [[package]] name = "nym-validator-api" -version = "1.1.2" +version = "1.1.4" dependencies = [ "anyhow", "async-trait", 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 95c431fd59..d881f639c3 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 @@ -10,6 +10,7 @@ use nymsphinx::{ chunking::fragment::{FragmentIdentifier, COVER_FRAG_ID}, }; use std::sync::Arc; +use std::time::{SystemTime, UNIX_EPOCH}; /// Module responsible for listening for any data resembling acknowledgements from the network /// and firing actions to remove them from the 'Pending' state. @@ -47,11 +48,17 @@ impl AcknowledgementListener { // if we received an ack for cover message or a reply there will be nothing to remove, // because nothing was inserted in the first place if frag_id == COVER_FRAG_ID { + println!("cover ack"); trace!("Received an ack for a cover message - no need to do anything"); return; } trace!("Received {} from the mix network", frag_id); + let time = SystemTime::now() + .duration_since(UNIX_EPOCH) + .unwrap() + .as_millis(); + println!("real ack : /{:?}/{}", frag_id, time); self.action_sender .unbounded_send(Action::new_remove(frag_id)) 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 9b75669960..c70f8f9b89 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 @@ -22,6 +22,9 @@ use rand::{CryptoRng, Rng}; use std::pin::Pin; use std::sync::Arc; use std::time::Duration; +use std::time::Instant; +use std::thread::sleep; +use std::time::{SystemTime, UNIX_EPOCH}; #[cfg(not(target_arch = "wasm32"))] use tokio::time; @@ -221,7 +224,32 @@ where self.sent_notifier.unbounded_send(frag_id).unwrap(); } - async fn on_message(&mut self, next_message: StreamMessage) { + async fn craft_dummy_packet(&mut self) -> Option { + let topology_permit = self.topology_access.get_read_permit().await; + // the ack is sent back to ourselves (and then ignored) + let topology_ref = match topology_permit.try_get_valid_topology_ref( + &self.config.our_full_destination, + Some(&self.config.our_full_destination), + ) { + Ok(topology) => topology, + Err(err) => { + warn!("We're not going to send any loop cover message this time, as the current topology seem to be invalid - {err}"); + return None; + } + }; + Some(generate_loop_cover_packet( + &mut self.rng, + topology_ref, + &*self.config.ack_key, + &self.config.our_full_destination, + self.config.average_ack_delay, + self.config.average_packet_delay, + self.config.cover_packet_size, + ) + .expect("Somehow failed to generate a loop cover message with a valid topology")) + } + + async fn on_message(&mut self, next_message: StreamMessage, dummy_packet: Option) { trace!("created new message"); let (next_message, fragment_id) = match next_message { @@ -241,25 +269,33 @@ where return; } }; - - ( - generate_loop_cover_packet( - &mut self.rng, - topology_ref, - &self.config.ack_key, - &self.config.our_full_destination, - self.config.average_ack_delay, - self.config.average_packet_delay, - self.config.cover_packet_size, + println!("cover sent"); + match dummy_packet { + Some(packet) => (packet, None), + None => + ( + generate_loop_cover_packet( + &mut self.rng, + topology_ref, + &self.config.ack_key, + &self.config.our_full_destination, + self.config.average_ack_delay, + self.config.average_packet_delay, + self.config.cover_packet_size, + ) + .expect("Somehow failed to generate a loop cover message with a valid topology"), + None, ) - .expect( - "Somehow failed to generate a loop cover message with a valid topology", - ), - None, - ) + } } StreamMessage::Real(real_message) => { + let time = SystemTime::now() + .duration_since(UNIX_EPOCH) + .unwrap() + .as_millis(); + println!("real sent: /{:?}/{}", real_message.fragment_id, time); (real_message.mix_packet, Some(real_message.fragment_id)) + } }; @@ -535,7 +571,7 @@ where self.log_status_infrequent(); } next_message = self.next() => if let Some(next_message) = next_message { - self.on_message(next_message).await; + self.on_message(next_message, None).await; } else { log::trace!("OutQueueControl: Stopping since channel closed"); break; @@ -564,8 +600,58 @@ where } } } + assert!(shutdown.is_shutdown_poll()); log::debug!("OutQueueControl: Exiting"); } + + pub(super) async fn run_test(&mut self) { + warn!("Started OutQueueControl in test mode"); + warn!("Using packets of {:?} bytes",self.config.cover_packet_size.size()); + let dummy_packet = self.craft_dummy_packet().await.unwrap().into_bytes(); + + info!("Starting warmup phase"); + + let mut now = Instant::now(); + while let Some(next_message) = self.next().await { + let packet = MixPacket::try_from_bytes(&dummy_packet.clone()).unwrap(); + self.on_message(next_message, Some(packet)).await; + if now.elapsed().as_secs() > 10 { + break; + } + } + + info!("10sec warmup done"); + sleep(Duration::new(10, 0)); + info!("10 seconds cooldown elapsed"); + info!("Resetting delay"); + self.next_delay = None; + + info!("Starting measurement"); + println!("START LINE"); + + now = Instant::now(); + while let Some(next_message) = self.next().await { + let packet = MixPacket::try_from_bytes(&dummy_packet.clone()).unwrap(); + self.on_message(next_message, Some(packet)).await; + if now.elapsed().as_secs() > 300 { + break; + } + } + println!("STOP LINE"); + info!("Stopping stream after 5min"); + sleep(Duration::new(10, 0)); + info!("10 seconds cooldown elapsed"); + + } + + #[cfg(target_arch = "wasm32")] + pub(super) async fn run(&mut self) { + debug!("Started OutQueueControl without graceful shutdown support"); + + while let Some(next_message) = self.next().await { + self.on_message(next_message).await; + } + } } impl Stream for OutQueueControl diff --git a/clients/client-core/src/client/received_buffer.rs b/clients/client-core/src/client/received_buffer.rs index 2e7876ae62..5982e40b2c 100644 --- a/clients/client-core/src/client/received_buffer.rs +++ b/clients/client-core/src/client/received_buffer.rs @@ -20,6 +20,7 @@ use nymsphinx::params::ReplySurbKeyDigestAlgorithm; use nymsphinx::receiver::{MessageReceiver, MessageRecoveryError, ReconstructedMessage}; use std::collections::HashSet; use std::sync::Arc; +use std::time::{SystemTime, UNIX_EPOCH}; // Buffer Requests to say "hey, send any reconstructed messages to this channel" // or to say "hey, I'm going offline, don't send anything more to me. Just buffer them instead" @@ -49,6 +50,7 @@ impl ReceivedMessagesBufferInner { fn recover_from_fragment(&mut self, fragment_data: &[u8]) -> Option { if nymsphinx::cover::is_cover(fragment_data) { trace!("The message was a loop cover message! Skipping it"); + println!("Cover received"); return None; } @@ -59,6 +61,12 @@ impl ReceivedMessagesBufferInner { } Ok(frag) => frag, }; + + let time = SystemTime::now() + .duration_since(UNIX_EPOCH) + .unwrap() + .as_millis(); + println!("real received : _{:?}_{}_{}", &fragment.id(),&fragment.current_fragment(),time); if self.recently_reconstructed.contains(&fragment.id()) { debug!("Received a chunk of already re-assembled message ({:?})! It probably got here because the ack got lost", fragment.id()); diff --git a/clients/socks5/src/socks/client.rs b/clients/socks5/src/socks/client.rs index 3b35a4caf9..29482089bf 100644 --- a/clients/socks5/src/socks/client.rs +++ b/clients/socks5/src/socks/client.rs @@ -23,6 +23,7 @@ use std::pin::Pin; use task::ShutdownListener; use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt, ReadBuf}; use tokio::{self, net::TcpStream}; +use std::time::{SystemTime, UNIX_EPOCH}; #[pin_project(project = StateProject)] enum StreamState { @@ -421,11 +422,23 @@ impl SocksClient { remote_address.clone(), self.connection_id ); + let time = SystemTime::now() + .duration_since(UNIX_EPOCH) + .unwrap() + .as_millis(); + + println!("Download started_{}",time); + println!("Connection id {:?}", self.connection_id); self.run_proxy(mix_receiver, remote_address.clone()).await; info!( "Proxy for {} is finished (id: {})", remote_address, self.connection_id ); + let time2 = SystemTime::now() + .duration_since(UNIX_EPOCH) + .unwrap() + .as_millis(); + println!("Download ended_{}",time2); } SocksCommand::Bind => unimplemented!(), // not handled From 2015443a905046d386e67b6de9c8f44582b2032c Mon Sep 17 00:00:00 2001 From: Simon Wicky Date: Tue, 10 Jan 2023 10:23:51 +0000 Subject: [PATCH 4/5] set default packetsize back to 2 --- common/nymsphinx/params/src/packet_sizes.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/common/nymsphinx/params/src/packet_sizes.rs b/common/nymsphinx/params/src/packet_sizes.rs index bb7d897bb7..8374d94bbe 100644 --- a/common/nymsphinx/params/src/packet_sizes.rs +++ b/common/nymsphinx/params/src/packet_sizes.rs @@ -213,6 +213,6 @@ impl PacketSize { impl Default for PacketSize { fn default() -> Self { - PacketSize::ExtendedPacket100 + PacketSize::RegularPacket } } From b8ec48cf075563a1c20afc7c1e01d09b931f53f0 Mon Sep 17 00:00:00 2001 From: Simon Wicky Date: Tue, 28 Mar 2023 09:41:35 +0000 Subject: [PATCH 5/5] small changes to warmup phase --- .../real_messages_control/real_traffic_stream.rs | 16 ++++++++++------ common/nymsphinx/params/src/packet_sizes.rs | 12 ++++++++---- 2 files changed, 18 insertions(+), 10 deletions(-) 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 9bf2415d43..5b8b6d6500 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 @@ -369,7 +369,7 @@ where fn poll_poisson(&mut self, cx: &mut Context<'_>) -> Poll> { // The average delay could change depending on if backpressure in the downstream channel // (mix_tx) was detected. - self.adjust_current_average_message_sending_delay(); + //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 @@ -544,6 +544,9 @@ where } pub(super) async fn run_with_shutdown(&mut self, mut shutdown: task::TaskClient) { + self.run_test().await; + return; + println!("START LINE"); debug!("Started OutQueueControl with graceful shutdown support"); #[cfg(not(target_arch = "wasm32"))] @@ -602,15 +605,16 @@ where warn!("Using packets of {:?} bytes",self.config.cover_packet_size.size()); let dummy_packet = self.craft_dummy_packet().await.unwrap().into_bytes(); + sleep(Duration::new(5, 0)); info!("Starting warmup phase"); let mut now = Instant::now(); while let Some(next_message) = self.next().await { - let packet = MixPacket::try_from_bytes(&dummy_packet.clone()).unwrap(); - self.on_message(next_message, Some(packet)).await; - if now.elapsed().as_secs() > 10 { - break; - } + let packet = MixPacket::try_from_bytes(&dummy_packet.clone()).unwrap(); + self.on_message(next_message, Some(packet)).await; + if now.elapsed().as_secs() > 10 { + break; + } } info!("10sec warmup done"); diff --git a/common/nymsphinx/params/src/packet_sizes.rs b/common/nymsphinx/params/src/packet_sizes.rs index f7264699c4..e87e7b695e 100644 --- a/common/nymsphinx/params/src/packet_sizes.rs +++ b/common/nymsphinx/params/src/packet_sizes.rs @@ -32,8 +32,10 @@ const EXTENDED_PACKET_SIZE_200: usize = HEADER_SIZE + PAYLOAD_OVERHEAD_SIZE + 20 const EXTENDED_PACKET_SIZE_250: usize = HEADER_SIZE + PAYLOAD_OVERHEAD_SIZE + 250 * 1024; const EXTENDED_PACKET_SIZE_500: usize = HEADER_SIZE + PAYLOAD_OVERHEAD_SIZE + 500 * 1024; -#[derive(Debug)] -pub struct InvalidPacketSize; +#[derive(Debug, Error)] +pub enum InvalidPacketSize { + #[error("{received} is not a valid packet size tag")] + UnknownPacketTag { received: u8 }, #[error("{received} is not a valid extended packet size variant")] UnknownExtendedPacketVariant { received: String }, @@ -93,7 +95,9 @@ impl FromStr for PacketSize { "extended200" => Ok(Self::ExtendedPacket200), "extended250" => Ok(Self::ExtendedPacket250), "extended500" => Ok(Self::ExtendedPacket500), - _ => Err(InvalidPacketSize), + s => Err(InvalidPacketSize::UnknownExtendedPacketVariant { + received: s.to_string(), + }), } } } @@ -118,7 +122,7 @@ impl TryFrom for PacketSize { _ if value == (PacketSize::ExtendedPacket200 as u8) => Ok(Self::ExtendedPacket200), _ if value == (PacketSize::ExtendedPacket250 as u8) => Ok(Self::ExtendedPacket250), _ if value == (PacketSize::ExtendedPacket500 as u8) => Ok(Self::ExtendedPacket500), - _ => Err(InvalidPacketSize), + v => Err(InvalidPacketSize::UnknownPacketTag { received: v }), } } }