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

Merge branch 'develop' of github.com:nymtech/nym into develop

Browse Source
This commit is contained in:
Dave
2020-11-19 18:07:07 +00:00
parent 4b34ed1f16 08cb39d0af
commit 455e735da4
5 changed files with 116 additions and 14 deletions
Download Patch File Download Diff File Expand all files Collapse all files
common/client-libs/validator-client/src/models/mixnode.rs
+8
View File
@@ -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<topology::mix::Node> for RegisteredMix {
network-monitor/src/main.rs
+34 -6
View File
@@ -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
}
network-monitor/src/monitor.rs
+6 -3
View File
@@ -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<Vec<Vec<u8>>>;
pub(crate) type MixnetSender = UnboundedSender<Vec<Vec<u8>>>;
@@ -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;
}
});
network-monitor/src/packet_sender.rs
+2 -1
View File
@@ -151,7 +151,7 @@ impl PacketSender {
async fn prepare_mix_packets(&mut self, test_mixes: Vec<TestMix>) -> Vec<MixPacket> {
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(())
}
}
network-monitor/src/tested_network/mod.rs
+66 -4
View File
@@ -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<MixPacket>,
mut mix_packets: Vec<MixPacket>,
) -> 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(())
}