From 08cb39d0af8963680b32f333e54e10dc8fa7933c Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?J=C4=99drzej=20Stuczy=C5=84ski?= Date: Thu, 19 Nov 2020 17:48:22 +0000 Subject: [PATCH] Feature/slowed down network monitor (#461) * Variable gateway sending rate * Changed interval ticker to delay * Extra comment * Performing delay after test run * Minor adjustements * Decreased sending rate * minor optimization in case of single packet to send. --- .../validator-client/src/models/mixnode.rs | 8 +++ network-monitor/src/main.rs | 40 +++++++++-- network-monitor/src/monitor.rs | 9 ++- network-monitor/src/packet_sender.rs | 3 +- network-monitor/src/tested_network/mod.rs | 70 +++++++++++++++++-- 5 files changed, 116 insertions(+), 14 deletions(-) diff --git a/common/client-libs/validator-client/src/models/mixnode.rs b/common/client-libs/validator-client/src/models/mixnode.rs index b72eabafe2..ad4ad62196 100644 --- a/common/client-libs/validator-client/src/models/mixnode.rs +++ b/common/client-libs/validator-client/src/models/mixnode.rs @@ -89,6 +89,14 @@ impl RegisteredMix { pub fn mix_host(&self) -> String { self.mix_info.node_info.mix_host.clone() } + + pub fn reputation(&self) -> i64 { + self.reputation + } + + pub fn layer(&self) -> u64 { + self.mix_info.layer + } } impl TryInto for RegisteredMix { diff --git a/network-monitor/src/main.rs b/network-monitor/src/main.rs index 9bc6dc0be1..1d644662b7 100644 --- a/network-monitor/src/main.rs +++ b/network-monitor/src/main.rs @@ -28,6 +28,7 @@ use packet_sender::PacketSender; use rand::rngs::OsRng; use std::sync::Arc; use std::time; +use std::time::Duration; use topology::{gateway, NymTopology}; mod chunker; @@ -45,6 +46,11 @@ const V4_TOPOLOGY_ARG: &str = "v4-topology-filepath"; const V6_TOPOLOGY_ARG: &str = "v6-topology-filepath"; const VALIDATOR_ARG: &str = "validator"; const DETAILED_REPORT_ARG: &str = "detailed-report"; +const GATEWAY_SENDING_RATE_ARG: &str = "gateway-rate"; + +const DEFAULT_VALIDATOR: &str = "http://testnet-validator1.nymtech.net:8081"; +const DEFAULT_GATEWAY_SENDING_RATE: usize = 20; +pub(crate) const TIME_CHUNK_SIZE: Duration = Duration::from_millis(50); fn parse_args<'a>() -> ArgMatches<'a> { App::new("Nym Network Monitor") @@ -52,24 +58,34 @@ fn parse_args<'a>() -> ArgMatches<'a> { .arg( Arg::with_name(V4_TOPOLOGY_ARG) .help("location of .json file containing IPv4 'good' network topology") + .long(V4_TOPOLOGY_ARG) .takes_value(true) .required(true), ) .arg( Arg::with_name(V6_TOPOLOGY_ARG) .help("location of .json file containing IPv6 'good' network topology") + .long(V6_TOPOLOGY_ARG) .takes_value(true) .required(true), ) .arg( Arg::with_name(VALIDATOR_ARG) .help("REST endpoint of the validator the monitor will grab nodes to test") + .long(VALIDATOR_ARG) .takes_value(true) - .required(true), ) .arg( Arg::with_name(DETAILED_REPORT_ARG) - .help("specifies whether a detailed report should be printed after each run"), + .help("specifies whether a detailed report should be printed after each run") + .long(DETAILED_REPORT_ARG) + , + ) + .arg(Arg::with_name(GATEWAY_SENDING_RATE_ARG) + .help("specifies maximum rate (in packets per second) of test packets being sent to gateway") + .takes_value(true) + .long(GATEWAY_SENDING_RATE_ARG) + .short("r") ) .get_matches() } @@ -85,8 +101,14 @@ async fn main() { let v4_topology = parse_topology_file(v4_topology_path); let v6_topology = parse_topology_file(v6_topology_path); - let validator_rest_uri = matches.value_of(VALIDATOR_ARG).unwrap(); + let validator_rest_uri = matches + .value_of(VALIDATOR_ARG) + .unwrap_or_else(|| DEFAULT_VALIDATOR); let detailed_report = matches.is_present(DETAILED_REPORT_ARG); + let sending_rate = matches + .value_of(GATEWAY_SENDING_RATE_ARG) + .map(|v| v.parse().unwrap()) + .unwrap_or_else(|| DEFAULT_GATEWAY_SENDING_RATE); check_if_up_to_date(&v4_topology, &v6_topology); setup_logging(); @@ -132,7 +154,8 @@ async fn main() { ); let gateway_client = new_gateway_client(gateway, identity_keypair, ack_sender, mixnet_sender); - let tested_network = new_tested_network(gateway_client, v4_topology, v6_topology).await; + let tested_network = + new_tested_network(gateway_client, v4_topology, v6_topology, sending_rate).await; let packet_sender = new_packet_sender( validator_client, @@ -148,10 +171,15 @@ async fn new_tested_network( gateway_client: GatewayClient, good_v4_topology: NymTopology, good_v6_topology: NymTopology, + max_sending_rate: usize, ) -> TestedNetwork { // TODO: possibly change that if it turns out we need two clients (v4 and v6) - let mut tested_network = - TestedNetwork::new_good(gateway_client, good_v4_topology, good_v6_topology); + let mut tested_network = TestedNetwork::new_good( + gateway_client, + good_v4_topology, + good_v6_topology, + max_sending_rate, + ); tested_network.start_gateway_client().await; tested_network } diff --git a/network-monitor/src/monitor.rs b/network-monitor/src/monitor.rs index 1586e68534..c2c373b668 100644 --- a/network-monitor/src/monitor.rs +++ b/network-monitor/src/monitor.rs @@ -15,7 +15,7 @@ use crate::{notifications::Notifier, packet_sender::PacketSender}; use futures::channel::mpsc::{UnboundedReceiver, UnboundedSender}; use log::*; -use tokio::time::{self, Duration}; +use tokio::time::{delay_for, Duration}; pub(crate) type MixnetReceiver = UnboundedReceiver>>; pub(crate) type MixnetSender = UnboundedSender>>; @@ -39,14 +39,17 @@ impl Monitor { }); tokio::spawn(async move { - let mut interval = time::interval(MONITOR_RUN_INTERVAL); loop { - interval.tick().await; info!(target: "Monitor", "Starting test run"); if let Err(err) = packet_sender.run_test().await { error!("Test run failed! - {:?}", err); } + + // only start delay after test run finished (note: this makes it so that + // test runs do not happen after EXACTLY MONITOR_RUN_INTERVAL, but at least + // it will be way less likely for multiple test runs to overlap each other) + delay_for(MONITOR_RUN_INTERVAL).await; } }); diff --git a/network-monitor/src/packet_sender.rs b/network-monitor/src/packet_sender.rs index 3a24f48e7b..e533dedbd4 100644 --- a/network-monitor/src/packet_sender.rs +++ b/network-monitor/src/packet_sender.rs @@ -151,7 +151,7 @@ impl PacketSender { async fn prepare_mix_packets(&mut self, test_mixes: Vec) -> Vec { let num_valid = test_mixes.iter().filter(|mix| mix.is_valid()).count(); - let mut mix_packets = Vec::with_capacity(num_valid); + let mut mix_packets = Vec::with_capacity(2 * num_valid); for test_mix in test_mixes { match test_mix { @@ -194,6 +194,7 @@ impl PacketSender { .unbounded_send(TestRunUpdate::DoneSending(self.nonce)) .expect("notifier has crashed!"); + info!(target: "Monitor", "Waiting for the test run to finish..."); Ok(()) } } diff --git a/network-monitor/src/tested_network/mod.rs b/network-monitor/src/tested_network/mod.rs index 5f2722a34d..4e4bf0fc1f 100644 --- a/network-monitor/src/tested_network/mod.rs +++ b/network-monitor/src/tested_network/mod.rs @@ -13,9 +13,12 @@ // limitations under the License. use crate::test_packet::{IpVersion, TestPacket}; +use crate::TIME_CHUNK_SIZE; use gateway_client::error::GatewayClientError; use gateway_client::GatewayClient; +use log::*; use nymsphinx::forwarding::packet::MixPacket; +use std::time::Duration; use topology::{mix, NymTopology}; pub(crate) mod good_topology; @@ -37,6 +40,7 @@ pub(crate) struct TestedNetwork { gateway_client: GatewayClient, good_v4_topology: NymTopology, good_v6_topology: NymTopology, + max_sending_rate: usize, } impl TestedNetwork { @@ -44,12 +48,14 @@ impl TestedNetwork { gateway_client: GatewayClient, good_v4_topology: NymTopology, good_v6_topology: NymTopology, + max_sending_rate: usize, ) -> Self { TestedNetwork { system_version: good_v4_topology.mixes()[&1][0].version.clone(), gateway_client, good_v4_topology, good_v6_topology, + max_sending_rate, } } @@ -66,11 +72,67 @@ impl TestedNetwork { pub(crate) async fn send_messages( &mut self, - mix_packets: Vec, + mut mix_packets: Vec, ) -> Result<(), GatewayClientError> { - self.gateway_client - .batch_send_mix_packets(mix_packets) - .await?; + info!(target: "MessageSender", "Got {} packets to send to gateway", mix_packets.len()); + // if we have fewer packets than our rate, just send it all + if mix_packets.len() <= self.max_sending_rate { + info!(target: "MessageSender", "Everything is going to get sent as one."); + self.gateway_client + .batch_send_mix_packets(mix_packets) + .await?; + } else { + let packets_per_time_chunk = + (self.max_sending_rate as f64 * TIME_CHUNK_SIZE.as_secs_f64()) as usize; + + info!( + target: "MessageSender", + "Going to send {} packets every {:?}", + packets_per_time_chunk, TIME_CHUNK_SIZE + ); + + let total_expected_time = + Duration::from_secs_f64(mix_packets.len() as f64 / self.max_sending_rate as f64); + info!(target: "MessageSender", + "With our rate of {} packets/s it should take around {:?} to send it all...", + self.max_sending_rate, total_expected_time + ); + + // TODO: is it perhaps possible to avoid so many reallocations here? + loop { + let mut retained = mix_packets.split_off(packets_per_time_chunk); + + let is_last = retained.len() < packets_per_time_chunk; + + debug!(target: "MessageSender", "Sending {} packets...", mix_packets.len()); + if mix_packets.len() == 1 { + self.gateway_client + .send_mix_packet(mix_packets.pop().unwrap()) + .await?; + } else { + self.gateway_client + .batch_send_mix_packets(mix_packets) + .await?; + } + + tokio::time::delay_for(TIME_CHUNK_SIZE).await; + + if is_last { + debug!(target: "MessageSender", "Sending {} packets...", retained.len()); + if retained.len() == 1 { + self.gateway_client + .send_mix_packet(retained.pop().unwrap()) + .await?; + } else { + self.gateway_client.batch_send_mix_packets(retained).await?; + } + break; + } + mix_packets = retained + } + info!(target: "MessageSender", "Done sending"); + } + Ok(()) }