diff --git a/common/mixnode-common/src/rtt_measurement/error.rs b/common/mixnode-common/src/rtt_measurement/error.rs index 2b4204ef1d..851957a912 100644 --- a/common/mixnode-common/src/rtt_measurement/error.rs +++ b/common/mixnode-common/src/rtt_measurement/error.rs @@ -32,6 +32,7 @@ pub enum RttError { UnexpectedConnectionFailureWrite(String, io::Error), UnexpectedConnectionFailureRead(String, io::Error), ConnectionReadTimeout(String), + ConnectionWriteTimeout(String), UnexpectedReplySequence, } @@ -72,6 +73,9 @@ impl Display for RttError { RttError::ConnectionReadTimeout(id) => { write!(f, "Timed out while trying to read reply packet from {}", id) } + RttError::ConnectionWriteTimeout(id) => { + write!(f, "Timed out while trying to write echo packet to {}", id) + } RttError::UnexpectedReplySequence => write!( f, "The received reply packet had an unexpected sequence number" diff --git a/common/mixnode-common/src/rtt_measurement/measurement.rs b/common/mixnode-common/src/rtt_measurement/measurement.rs index e7b301eb16..a4acfceb7c 100644 --- a/common/mixnode-common/src/rtt_measurement/measurement.rs +++ b/common/mixnode-common/src/rtt_measurement/measurement.rs @@ -21,13 +21,28 @@ use std::time::Duration; use tokio::sync::RwLock; pub struct AtomicVerlocResult { - inner: Arc>>, + inner: Arc>, +} + +#[derive(Debug, Clone, Serialize)] +pub struct VerlocResult { + total_tested: usize, + #[serde(with = "humantime_serde")] + run_started: Option, + #[serde(with = "humantime_serde")] + run_finished: Option, + results: Vec, } impl AtomicVerlocResult { pub(crate) fn new() -> Self { AtomicVerlocResult { - inner: Arc::new(RwLock::new(Vec::new())), + inner: Arc::new(RwLock::new(VerlocResult { + total_tested: 0, + run_started: None, + run_finished: None, + results: Vec::new(), + })), } } @@ -38,14 +53,30 @@ impl AtomicVerlocResult { } } - pub(crate) async fn update_results(&self, new_data: Vec) { + pub(crate) async fn reset_results(&self, new_tested: usize) { let mut write_permit = self.inner.write().await; - *write_permit = new_data; + write_permit.total_tested = new_tested; + write_permit.run_started = Some(std::time::SystemTime::now()); + write_permit.run_finished = None; + write_permit.results = Vec::new() + } + + pub(crate) async fn append_results(&self, mut new_data: Vec) { + let mut write_permit = self.inner.write().await; + write_permit.results.append(&mut new_data); + // make sure the data always stays in order. + // TODO: considering the front of the results is guaranteed to be sorted, should perhaps + // a non-default sorting algorithm be used? + write_permit.results.sort() + } + + pub(crate) async fn finish_measurements(&self) { + self.inner.write().await.run_finished = Some(std::time::SystemTime::now()); } // Considering that on every read we will need to clone data regardless, let's make our // lives simpler and clone it here rather than deal with lifetime of the permit - pub async fn clone_data(&self) -> Vec { + pub async fn clone_data(&self) -> VerlocResult { self.inner.read().await.clone() } } @@ -69,11 +100,10 @@ where impl Display for Verloc { fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result { - let identity = self.identity.to_base58_string(); if let Some(measurement) = self.latest_measurement { - write!(f, "{} - {}", identity, measurement) + write!(f, "{} - {}", self.identity, measurement) } else { - write!(f, "{} - COULD NOT MEASURE", identity) + write!(f, "{} - COULD NOT MEASURE", self.identity) } } } diff --git a/common/mixnode-common/src/rtt_measurement/mod.rs b/common/mixnode-common/src/rtt_measurement/mod.rs index 4e0738d9e7..2a351eef49 100644 --- a/common/mixnode-common/src/rtt_measurement/mod.rs +++ b/common/mixnode-common/src/rtt_measurement/mod.rs @@ -13,7 +13,7 @@ // limitations under the License. use crate::rtt_measurement::listener::PacketListener; -pub use crate::rtt_measurement::measurement::{AtomicVerlocResult, Verloc}; +pub use crate::rtt_measurement::measurement::{AtomicVerlocResult, Verloc, VerlocResult}; use crate::rtt_measurement::sender::{PacketSender, TestedNode}; use crypto::asymmetric::identity; use futures::stream::FuturesUnordered; @@ -39,6 +39,7 @@ pub const DEFAULT_MEASUREMENT_PORT: u16 = 1790; // by default all of those are overwritten by config data from mixnodes directly const DEFAULT_PACKETS_PER_NODE: usize = 100; const DEFAULT_PACKET_TIMEOUT: Duration = Duration::from_millis(1500); +const DEFAULT_CONNECTION_TIMEOUT: Duration = Duration::from_millis(5000); const DEFAULT_DELAY_BETWEEN_PACKETS: Duration = Duration::from_millis(50); const DEFAULT_BATCH_SIZE: usize = 50; const DEFAULT_TESTING_INTERVAL: Duration = Duration::from_secs(60 * 60 * 12); @@ -61,6 +62,9 @@ pub struct Config { /// Specifies maximum amount of time to wait for the reply packet to arrive before abandoning the test. packet_timeout: Duration, + /// Specifies maximum amount of time to wait for the connection to get established. + connection_timeout: Duration, + /// Specifies delay between subsequent test packets being sent (after receiving a reply). delay_between_packets: Duration, @@ -114,6 +118,10 @@ impl ConfigBuilder { self.0.packet_timeout = packet_timeout; self } + pub fn connection_timeout(mut self, connection_timeout: Duration) -> Self { + self.0.connection_timeout = connection_timeout; + self + } pub fn delay_between_packets(mut self, delay_between_packets: Duration) -> Self { self.0.delay_between_packets = delay_between_packets; self @@ -163,6 +171,7 @@ impl Default for ConfigBuilder { .unwrap(), packets_per_node: DEFAULT_PACKETS_PER_NODE, packet_timeout: DEFAULT_PACKET_TIMEOUT, + connection_timeout: DEFAULT_CONNECTION_TIMEOUT, delay_between_packets: DEFAULT_DELAY_BETWEEN_PACKETS, tested_nodes_batch_size: DEFAULT_BATCH_SIZE, testing_interval: DEFAULT_TESTING_INTERVAL, @@ -198,6 +207,7 @@ impl RttMeasurer { Arc::clone(&identity), config.packets_per_node, config.packet_timeout, + config.connection_timeout, config.delay_between_packets, )), packet_listener: Arc::new(PacketListener::new( @@ -222,10 +232,10 @@ impl RttMeasurer { tokio::spawn(packet_listener.run()) } - async fn perform_measurement(&self, nodes_to_test: Vec) -> Vec { - let mut results = Vec::with_capacity(nodes_to_test.len()); - + async fn perform_measurement(&self, nodes_to_test: Vec) { for chunk in nodes_to_test.chunks(self.config.tested_nodes_batch_size) { + let mut chunk_results = Vec::with_capacity(chunk.len()); + let mut measurement_chunk = chunk .iter() .map(|node| { @@ -261,18 +271,18 @@ impl RttMeasurer { } Ok(result) => Some(result), }; - results.push(Verloc::new(execution_result.1, measurement_result)); + chunk_results.push(Verloc::new(execution_result.1, measurement_result)); } - } - // finally sort the results - results.sort(); - results + // update the results vector with chunks as they become available (by default every 50 nodes) + self.results.append_results(chunk_results).await; + } } pub async fn run(&mut self) { self.start_listening(); loop { + info!(target: "verloc", "Starting verloc measurements"); // TODO: should we also measure gateways? let all_mixes = match self.validator_client.get_mix_nodes().await { Ok(nodes) => nodes, @@ -310,9 +320,15 @@ impl RttMeasurer { }) .collect::>(); - let results = self.perform_measurement(tested_nodes).await; - self.results.update_results(results).await; + // on start of each run remove old results + self.results.reset_results(tested_nodes.len()).await; + self.perform_measurement(tested_nodes).await; + + // write current time to "run finished" field + self.results.finish_measurements().await; + + info!(target: "verloc", "Finished performing verloc measurements. The next one will happen in {:?}", self.config.testing_interval); sleep(self.config.testing_interval).await } } diff --git a/common/mixnode-common/src/rtt_measurement/sender.rs b/common/mixnode-common/src/rtt_measurement/sender.rs index 4c0de920e7..4bbefd2dac 100644 --- a/common/mixnode-common/src/rtt_measurement/sender.rs +++ b/common/mixnode-common/src/rtt_measurement/sender.rs @@ -18,6 +18,7 @@ use crate::rtt_measurement::packet::{EchoPacket, ReplyPacket}; use crypto::asymmetric::identity; use log::*; use rand::{thread_rng, Rng}; +use std::io; use std::net::SocketAddr; use std::sync::Arc; use std::time::Duration; @@ -42,6 +43,7 @@ pub(crate) struct PacketSender { // timeout for receiving before sending new one packets_per_node: usize, packet_timeout: Duration, + connection_timeout: Duration, delay_between_packets: Duration, } @@ -50,12 +52,14 @@ impl PacketSender { identity: Arc, packets_per_node: usize, packet_timeout: Duration, + connection_timeout: Duration, delay_between_packets: Duration, ) -> Self { PacketSender { identity, packets_per_node, packet_timeout, + connection_timeout, delay_between_packets, } } @@ -76,11 +80,26 @@ impl PacketSender { self: Arc, tested_node: TestedNode, ) -> Result { - let mut conn = TcpStream::connect(tested_node.address) - .await - .map_err(|err| { - RttError::UnreachableNode(tested_node.identity.to_base58_string(), err) - })?; + let mut conn = match tokio::time::timeout( + self.connection_timeout, + TcpStream::connect(tested_node.address), + ) + .await + { + Err(_timeout) => { + return Err(RttError::UnreachableNode( + tested_node.identity.to_base58_string(), + io::ErrorKind::TimedOut.into(), + )) + } + Ok(Err(err)) => { + return Err(RttError::UnreachableNode( + tested_node.identity.to_base58_string(), + err, + )) + } + Ok(Ok(conn)) => conn, + }; let mut results = Vec::with_capacity(self.packets_per_node); @@ -90,16 +109,35 @@ impl PacketSender { let start = tokio::time::Instant::now(); // TODO: should we get the start time after or before actually sending the data? // there's going to definitely some scheduler and network stack bias here - if let Err(err) = conn.write_all(packet.to_bytes().as_ref()).await { - let identity_string = tested_node.identity.to_base58_string(); - error!( - "failed to write echo packet to {} - {}. Stopping the test.", - identity_string, err - ); - return Err(RttError::UnexpectedConnectionFailureWrite( - identity_string, - err, - )); + match tokio::time::timeout( + self.packet_timeout, + conn.write_all(packet.to_bytes().as_ref()), + ) + .await + { + Err(_timeout) => { + let identity_string = tested_node.identity.to_base58_string(); + debug!( + "failed to write echo packet to {} within {:?}. Stopping the test.", + identity_string, self.packet_timeout + ); + return Err(RttError::UnexpectedConnectionFailureWrite( + identity_string, + io::ErrorKind::TimedOut.into(), + )); + } + Ok(Err(err)) => { + let identity_string = tested_node.identity.to_base58_string(); + debug!( + "failed to write echo packet to {} - {}. Stopping the test.", + identity_string, err + ); + return Err(RttError::UnexpectedConnectionFailureWrite( + identity_string, + err, + )); + } + Ok(Ok(_)) => {} } // there's absolutely no need to put a codec on ReplyPackets as we know exactly @@ -107,7 +145,7 @@ impl PacketSender { let reply_packet_future = async { let mut buf = [0u8; ReplyPacket::SIZE]; if let Err(err) = conn.read_exact(&mut buf).await { - error!( + debug!( "failed to read reply packet from {} - {}. Stopping the test.", tested_node.identity.to_base58_string(), err @@ -123,10 +161,10 @@ impl PacketSender { let reply_packet = match tokio::time::timeout(self.packet_timeout, reply_packet_future).await { Ok(reply_packet) => reply_packet, - Err(_) => { + Err(_timeout) => { // TODO: should we continue regardless (with the rest of the packets, or abandon the whole thing?) // Note: if we decide to continue, it would increase the complexity of the whole thing - error!( + debug!( "failed to receive reply to our echo packet within {:?}. Stopping the test", self.packet_timeout ); @@ -141,7 +179,7 @@ impl PacketSender { // note that we cannot receive packets not in order as we are not sending a next packet until // we have received the previous one if reply_packet.base_sequence_number() != seq { - error!("Received reply packet with invalid sequence number! Got {} expected {}. Stopping the test", reply_packet.base_sequence_number(), seq); + debug!("Received reply packet with invalid sequence number! Got {} expected {}. Stopping the test", reply_packet.base_sequence_number(), seq); return Err(RttError::UnexpectedReplySequence); } diff --git a/mixnode/src/config/mod.rs b/mixnode/src/config/mod.rs index ccc34d8ca2..c0bd0d138a 100644 --- a/mixnode/src/config/mod.rs +++ b/mixnode/src/config/mod.rs @@ -27,6 +27,7 @@ pub const DEFAULT_MIXNET_CONTRACT_ADDRESS: &str = "hal1k0jntykt7e4g3y88ltc60czgj // 'RTT MEASUREMENT' const DEFAULT_PACKETS_PER_NODE: usize = 100; +const DEFAULT_CONNECTION_TIMEOUT: Duration = Duration::from_millis(5000); const DEFAULT_PACKET_TIMEOUT: Duration = Duration::from_millis(1500); const DEFAULT_DELAY_BETWEEN_PACKETS: Duration = Duration::from_millis(50); const DEFAULT_BATCH_SIZE: usize = 50; @@ -314,6 +315,11 @@ impl Config { pub fn get_measurement_packet_timeout(&self) -> Duration { self.rtt_measurement.packet_timeout } + + pub fn get_measurement_connection_timeout(&self) -> Duration { + self.rtt_measurement.connection_timeout + } + pub fn get_measurement_delay_between_packets(&self) -> Duration { self.rtt_measurement.delay_between_packets } @@ -450,6 +456,9 @@ pub struct RttMeasurement { /// Specifies number of echo packets sent to each node during a measurement run. packets_per_node: usize, + /// Specifies maximum amount of time to wait for the connection to get established. + connection_timeout: Duration, + /// Specifies maximum amount of time to wait for the reply packet to arrive before abandoning the test. packet_timeout: Duration, @@ -471,6 +480,7 @@ impl Default for RttMeasurement { fn default() -> Self { RttMeasurement { packets_per_node: DEFAULT_PACKETS_PER_NODE, + connection_timeout: DEFAULT_CONNECTION_TIMEOUT, packet_timeout: DEFAULT_PACKET_TIMEOUT, delay_between_packets: DEFAULT_DELAY_BETWEEN_PACKETS, tested_nodes_batch_size: DEFAULT_BATCH_SIZE, diff --git a/mixnode/src/node/http/verloc.rs b/mixnode/src/node/http/verloc.rs index e4934de418..9d71ee67a3 100644 --- a/mixnode/src/node/http/verloc.rs +++ b/mixnode/src/node/http/verloc.rs @@ -1,4 +1,4 @@ -use mixnode_common::rtt_measurement::{AtomicVerlocResult, Verloc}; +use mixnode_common::rtt_measurement::{AtomicVerlocResult, VerlocResult}; use rocket::State; use rocket_contrib::json::Json; @@ -17,7 +17,7 @@ impl VerlocState { /// Provides verifiable location (verloc) measurements for this mixnode - a list of the /// round-trip times, in milliseconds, for all other mixnodes that this node knows about. #[get("/verloc")] -pub(crate) async fn verloc(state: State<'_, VerlocState>) -> Json> { +pub(crate) async fn verloc(state: State<'_, VerlocState>) -> Json { // since it's impossible to get a mutable reference to the state, we can't cache any results outside the lock : ( Json(state.shared.clone_data().await) } diff --git a/mixnode/src/node/mod.rs b/mixnode/src/node/mod.rs index 3a1d4d15f5..ac076ae2bb 100644 --- a/mixnode/src/node/mod.rs +++ b/mixnode/src/node/mod.rs @@ -144,6 +144,7 @@ impl MixNode { let config = rtt_measurement::ConfigBuilder::new() .listening_address(listening_address) .packets_per_node(self.config.get_measurement_packets_per_node()) + .connection_timeout(self.config.get_measurement_connection_timeout()) .packet_timeout(self.config.get_measurement_packet_timeout()) .delay_between_packets(self.config.get_measurement_delay_between_packets()) .tested_nodes_batch_size(self.config.get_measurement_tested_nodes_batch_size())