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

Bugfix/verloc fixes and adjustments (#618)

Browse Source 
* Using new display trait for identity key

* Establishing verloc connection with timeout

* Further decrease in log severity

* Writing echo packets with timeout

* Sender formatting

* ConnectionWriteTimeout error definition

* Writing verloc results in chunks

* Added run started and run finished fields to verloc

* Reordered the fields

* Storing the timestamps as options to indicate run in progress
This commit is contained in:
Jędrzej Stuczyński
2021-06-02 12:50:36 +01:00
committed by GitHub
parent 0dcb046576
commit 93e9dc5c1e
7 changed files with 139 additions and 40 deletions
Download Patch File Download Diff File Expand all files Collapse all files
common/mixnode-common/src/rtt_measurement/error.rs
+4
View File
@@ -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"
common/mixnode-common/src/rtt_measurement/measurement.rs
+38 -8
View File
@@ -21,13 +21,28 @@ use std::time::Duration;
use tokio::sync::RwLock;
pub struct AtomicVerlocResult {
inner: Arc<RwLock<Vec<Verloc>>>,
inner: Arc<RwLock<VerlocResult>>,
}
#[derive(Debug, Clone, Serialize)]
pub struct VerlocResult {
total_tested: usize,
#[serde(with = "humantime_serde")]
run_started: Option<std::time::SystemTime>,
#[serde(with = "humantime_serde")]
run_finished: Option<std::time::SystemTime>,
results: Vec<Verloc>,
}
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<Verloc>) {
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<Verloc>) {
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<Verloc> {
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)
}
}
}
common/mixnode-common/src/rtt_measurement/mod.rs
+27 -11
View File
@@ -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<TestedNode>) -> Vec<Verloc> {
let mut results = Vec::with_capacity(nodes_to_test.len());
async fn perform_measurement(&self, nodes_to_test: Vec<TestedNode>) {
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::<Vec<_>>();
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
}
}
common/mixnode-common/src/rtt_measurement/sender.rs
+57 -19
View File
@@ -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<identity::KeyPair>,
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<Self>,
tested_node: TestedNode,
) -> Result<Measurement, RttError> {
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);
}
mixnode/src/config/mod.rs
+10
View File
@@ -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,
mixnode/src/node/http/verloc.rs
+2 -2
View File
@@ -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<Vec<Verloc>> {
pub(crate) async fn verloc(state: State<'_, VerlocState>) -> Json<VerlocResult> {
// 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)
}
mixnode/src/node/mod.rs
+1
View File
@@ -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())