Merge remote-tracking branch 'origin/simon/instrumented' into simon/instrumented

clients/client-core/src/client/real_messages_control/acknowledgement_control/acknowledgement_listener.rs

@@ -10,6 +10,7 @@ use nym_sphinx::{
     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))

clients/client-core/src/client/real_messages_control/real_traffic_stream.rs

@@ -24,6 +24,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;
@@ -212,7 +215,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<MixPacket> {
+        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<MixPacket>) {
         trace!("created new message");
 
         let (next_message, fragment_id) = match next_message {
@@ -232,25 +260,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))
+
             }
         };
 
@@ -502,6 +538,9 @@ where
     }
 
     pub(super) async fn run_with_shutdown(&mut self, mut shutdown: nym_task::TaskClient) {
+        self.run_test().await;
+        return;
+        println!("START LINE");
         debug!("Started OutQueueControl with graceful shutdown support");
 
         #[cfg(not(target_arch = "wasm32"))]
@@ -522,7 +561,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;
@@ -549,8 +588,59 @@ 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();
+
+        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;
+           }
+        }
+
+        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<R> Stream for OutQueueControl<R>

clients/client-core/src/client/received_buffer.rs

@@ -20,6 +20,7 @@ use nym_sphinx::params::ReplySurbKeyDigestAlgorithm;
 use nym_sphinx::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<NymMessage> {
         if nym_sphinx::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());

clients/socks5/src/socks/client.rs

@@ -26,6 +26,7 @@ use std::net::SocketAddr;
 use std::pin::Pin;
 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 {
@@ -484,11 +485,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

common/nymsphinx/params/src/packet_sizes.rs

@@ -20,6 +20,7 @@ const ACK_PACKET_SIZE: usize = HEADER_SIZE + PAYLOAD_OVERHEAD_SIZE + ACK_IV_SIZE
 const EXTENDED_PACKET_SIZE_8: usize = HEADER_SIZE + PAYLOAD_OVERHEAD_SIZE + 8 * 1024;
 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;