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

Debugging validator (#1198)

Browse Source 
* Checkpoint

* Replace Stream logic with StreamMap

* Ignore blacklisted mixnodes and gateways

* Moar logging

* Remove version checks

* Cleanup

* Some more cleanup
This commit is contained in:
Drazen Urch
2022-04-07 18:13:29 +02:00
committed by GitHub
parent c177f14073
commit 9aa5b98465
10 changed files with 84 additions and 237 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Cargo.lock
Generated
+1
View File
@@ -3100,6 +3100,7 @@ dependencies = [
"thiserror",
"time 0.3.7",
"tokio",
"tokio-stream",
"topology",
"url",
"validator-api-requests",
common/client-libs/gateway-client/src/packet_router.rs
+6 -1
View File
@@ -72,7 +72,12 @@ impl PacketRouter {
if !received_acks.is_empty() {
trace!("routing acks");
self.ack_sender.unbounded_send(received_acks).unwrap();
match self.ack_sender.unbounded_send(received_acks) {
Ok(_) => {}
Err(e) => {
error!("failed to send ack: {:?}", e);
}
};
}
}
}
common/client-libs/gateway-client/src/socket_state.rs
+1 -1
View File
@@ -74,7 +74,7 @@ impl PartiallyDelegated {
// This would also require NOT discarding any text responses here.
// TODO: those can return the "send confirmations" - perhaps it should be somehow worked around?
Message::Text(text) => debug!(
Message::Text(text) => trace!(
"received a text message - probably a response to some previous query! - {}",
text
),
validator-api/Cargo.toml
+1
View File
@@ -15,6 +15,7 @@ rust-version = "1.56"
# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
[dependencies]
tokio-stream = "0.1.8"
clap = "2.33.0"
dirs = "3.0"
dotenv = "0.15.0"
validator-api/src/network_monitor/gateways_reader.rs
+18 -158
View File
@@ -2,181 +2,41 @@
// SPDX-License-Identifier: Apache-2.0
use crypto::asymmetric::identity;
use futures::stream::Stream;
use futures::task::Context;
use gateway_client::{AcknowledgementReceiver, MixnetMessageReceiver};
use std::pin::Pin;
use std::task::{Poll, Waker};
use tokio_stream::StreamMap;
/// Constant used to determine maximum number of times the GatewayReader can poll. It basically
/// tries to solve the same problem that `FuturesUnordered` has: https://github.com/rust-lang/futures-rs/issues/2047
const YIELD_EVERY: usize = 32;
// TODO: Originally I set it to (identity::PublicKey, Vec<Vec<u8>>) and I definitely
// had a reason for doing so, but right now I can't remember what that was...
pub(crate) type GatewayMessages = Vec<Vec<u8>>;
pub(crate) struct GatewayChannel {
id: identity::PublicKey,
message_receiver: MixnetMessageReceiver,
ack_receiver: AcknowledgementReceiver,
is_closed: bool,
}
impl GatewayChannel {
pub(crate) fn new(
id: identity::PublicKey,
message_receiver: MixnetMessageReceiver,
ack_receiver: AcknowledgementReceiver,
) -> Self {
GatewayChannel {
id,
message_receiver,
ack_receiver,
is_closed: false,
}
}
}
impl Stream for GatewayChannel {
type Item = Vec<Vec<u8>>;
fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
if self.is_closed {
return Poll::Ready(None);
}
let mut polled = 0;
// empty the ack channel if anything is on it (we don't care about the content at all at the
// moment)
while let Poll::Ready(_ack) = Pin::new(&mut self.ack_receiver).poll_next(cx) {
polled += 1;
}
let item = futures::ready!(Pin::new(&mut self.message_receiver).poll_next(cx));
match item {
None => Poll::Ready(None),
// if we managed to get an item, try to also read additional ones
Some(mut messages) => {
while let Poll::Ready(new_item) = Pin::new(&mut self.message_receiver).poll_next(cx)
{
polled += 1;
match new_item {
None => {
self.is_closed = true;
cx.waker().wake_by_ref();
return Poll::Ready(Some(messages));
}
Some(mut additional_messages) => {
messages.append(&mut additional_messages);
// it is fine enough to use the same constant here as in the main GatewayReader
// as only a single channel, i.e. the main gateway will be capable
// of returning more than 2 values
if polled >= YIELD_EVERY {
cx.waker().wake_by_ref();
break;
}
}
}
}
Poll::Ready(Some(messages))
}
}
}
}
pub(crate) struct GatewaysReader {
latest_read: usize,
// channels: FuturesUnordered<GatewayChannel>,
channels: Vec<GatewayChannel>,
waker: Option<Waker>,
ack_map: StreamMap<String, AcknowledgementReceiver>,
stream_map: StreamMap<String, MixnetMessageReceiver>,
}
impl GatewaysReader {
pub(crate) fn new() -> Self {
GatewaysReader {
latest_read: 0,
channels: Vec::new(),
waker: None,
ack_map: StreamMap::new(),
stream_map: StreamMap::new(),
}
}
fn remove_nth(&mut self, i: usize) {
self.channels.remove(i);
pub fn stream_map(&mut self) -> &mut StreamMap<String, MixnetMessageReceiver> {
&mut self.stream_map
}
// todo: if we find that this method is called frequently, perhaps the vector should get
// replaced with different data structure
pub(crate) fn remove_by_key(&mut self, key: identity::PublicKey) {
match self.channels.iter().position(|item| item.id == key) {
Some(i) => {
self.channels.remove(i);
}
// this shouldn't ever get thrown, so perhaps a panic would be more in order?
None => error!(
"tried to remove gateway reader {} but it doesn't exist!",
key.to_base58_string()
),
}
}
fn poll_nth(
pub fn add_recievers(
&mut self,
cx: &mut Context<'_>,
i: usize,
) -> Option<Poll<Option<GatewayMessages>>> {
if let Poll::Ready(item) = Pin::new(&mut self.channels[i]).poll_next(cx) {
self.latest_read = i;
match item {
// Some(messages) => return Some(Poll::Ready(Some((self.channels[i].id, messages)))),
Some(messages) => return Some(Poll::Ready(Some(messages))),
// remove dead channel
None => self.remove_nth(i),
}
}
None
id: identity::PublicKey,
message_receiver: MixnetMessageReceiver,
ack_receiver: AcknowledgementReceiver,
) {
let channel_id = id.to_string();
self.stream_map.insert(channel_id.clone(), message_receiver);
self.ack_map.insert(channel_id, ack_receiver);
}
pub(crate) fn insert_channel(&mut self, channel: GatewayChannel) {
self.channels.push(channel);
if let Some(waker) = self.waker.take() {
waker.wake()
}
}
}
// TODO: not sure if this will scale well, but I don't know what would be a good alternative,
// perhaps try to somehow incorporate FuturesUnordered?
// also, perhaps reading should be done in parallel?
impl Stream for GatewaysReader {
// item represents gateway that returned messages alongside the actual messages
type Item = GatewayMessages;
fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
if self.latest_read >= self.channels.len() {
self.latest_read = 0;
}
// don't start reading from beginning each time to at least slightly help with the bias
for i in self.latest_read..self.channels.len() {
if let Some(item) = self.poll_nth(cx, i) {
return item;
}
}
for i in 0..self.latest_read {
if let Some(item) = self.poll_nth(cx, i) {
return item;
}
}
// if we have no channels available, store the waker to be woken when a new one is pushed
if self.channels.is_empty() {
self.waker = Some(cx.waker().clone())
}
Poll::Pending
pub fn remove_recievers(&mut self, id: &str) {
self.stream_map.remove(id);
self.ack_map.remove(id);
}
}
validator-api/src/network_monitor/monitor/gateways_pinger.rs
+17 -7
View File
@@ -44,14 +44,17 @@ impl GatewayPinger {
}
async fn ping_and_cleanup_all_gateways(&self) {
info!(target: "GatewayPinger", "Pinging all active gateways");
info!("Pinging all active gateways");
let lock_acquire_start = Instant::now();
let active_gateway_clients_guard = self.gateway_clients.lock().await;
trace!(target: "GatewayPinger", "Acquiring lock took {:?}", Instant::now().duration_since(lock_acquire_start));
trace!(
"Acquiring lock took {:?}",
Instant::now().duration_since(lock_acquire_start)
);
if active_gateway_clients_guard.is_empty() {
debug!(target: "GatewayPinger", "no gateways to ping");
debug!("no gateways to ping");
return;
}
@@ -89,7 +92,6 @@ impl GatewayPinger {
{
Err(_timeout) => {
warn!(
target: "GatewayPinger",
"we timed out trying to ping {} - assuming the connection is dead.",
active_client.gateway_identity().to_base58_string(),
);
@@ -97,7 +99,6 @@ impl GatewayPinger {
}
Ok(Err(err)) => {
warn!(
target: "GatewayPinger",
"failed to send ping message to gateway {} - {} - assuming the connection is dead.",
active_client.gateway_identity().to_base58_string(),
err,
@@ -112,25 +113,34 @@ impl GatewayPinger {
}
}
info!(
"Purging {} gateways, acquiring lock",
clients_to_purge.len()
);
// purge all dead connections
// reacquire the guard
let lock_acquire_start = Instant::now();
let mut active_gateway_clients_guard = self.gateway_clients.lock().await;
trace!(target: "GatewayPinger", "Acquiring lock took {:?}", Instant::now().duration_since(lock_acquire_start));
info!(
"Acquiring lock took {:?}",
Instant::now().duration_since(lock_acquire_start)
);
for gateway_id in clients_to_purge.into_iter() {
if let Some(removed_handle) = active_gateway_clients_guard.remove(&gateway_id) {
if !removed_handle.is_invalid().await {
info!("Handle is invalid, purging");
// it was not invalidated by the packet sender meaning it probably was some unbonded node
// that was never cleared
self.notify_connection_failure(gateway_id);
}
info!("Handle is not invalid, not purged")
}
}
let ping_end = Instant::now();
let time_taken = ping_end.duration_since(ping_start);
debug!(target: "GatewayPinger", "Pinging all active gateways took {:?}", time_taken);
info!("Pinging all active gateways took {:?}", time_taken);
}
pub(crate) async fn run(&self) {
validator-api/src/network_monitor/monitor/mod.rs
+13 -9
View File
@@ -177,9 +177,9 @@ impl Monitor {
for entry in results.iter() {
if *entry.1 == self.route_test_packets {
debug!("✔️ {} succeeded", entry.0)
info!("✔️ {} succeeded", entry.0)
} else {
debug!(
info!(
"❌️ {} failed ({}/{} received)",
entry.0, entry.1, self.route_test_packets
)
@@ -200,7 +200,7 @@ impl Monitor {
}
async fn prepare_test_routes(&mut self) -> Option<Vec<TestRoute>> {
info!(target: "Monitor", "Generating test routes...");
info!("Generating test routes...");
// keep track of nodes that should not be used for route construction
let mut blacklist = HashSet::new();
@@ -209,7 +209,7 @@ impl Monitor {
let mut current_attempt = 0;
// todo: tweak this to something more appropriate
let max_attempts = self.test_routes * 2;
let max_attempts = self.test_routes * 10;
'outer: loop {
if current_attempt >= max_attempts {
@@ -277,13 +277,12 @@ impl Monitor {
.set_new_test_nonce(self.test_nonce)
.await;
debug!("Sending packets to all gateways...");
info!("Sending packets to all gateways...");
self.packet_sender
.send_packets(prepared_packets.packets)
.await;
debug!(
target: "Monitor",
info!(
"Sending is over, waiting for {:?} before checking what we received",
self.packet_delivery_timeout
);
@@ -293,6 +292,8 @@ impl Monitor {
let received = self.received_processor.return_received().await;
let total_received = received.len();
info!("Test routes: {:?}", routes);
info!("Received {}/{} packets", total_received, total_sent);
let summary = self.summary_producer.produce_summary(
prepared_packets.tested_mixnodes,
@@ -310,11 +311,14 @@ impl Monitor {
}
async fn test_run(&mut self) {
info!(target: "Monitor", "Starting test run no. {}", self.test_nonce);
info!("Starting test run no. {}", self.test_nonce);
let start = Instant::now();
if let Some(test_routes) = self.prepare_test_routes().await {
debug!(target: "Monitor", "Determined reliable routes to test all other nodes against. : {:?}", test_routes);
info!(
"Determined reliable routes to test all other nodes against. : {:?}",
test_routes
);
self.test_network_against(&test_routes).await;
} else {
error!("We failed to construct sufficient number of test routes to test the network against")
validator-api/src/network_monitor/monitor/preparer.rs
+15 -51
View File
@@ -25,6 +25,7 @@ type Id = String;
type Owner = Addr;
#[derive(Clone)]
#[allow(dead_code)]
pub(crate) enum InvalidNode {
Outdated(Id, Owner, Version),
Malformed(Id, Owner),
@@ -188,8 +189,8 @@ impl PacketPreparer {
info!("Waiting for minimal topology to be online");
let initialisation_backoff = Duration::from_secs(30);
loop {
let gateways = self.validator_cache.gateways().await;
let mixnodes = self.validator_cache.rewarded_set().await;
let gateways = self.validator_cache.gateways_all().await;
let mixnodes = self.validator_cache.mixnodes_all().await;
if gateways.len() < minimum_full_routes {
info!(
@@ -201,7 +202,7 @@ impl PacketPreparer {
}
let mut layered_mixes = HashMap::new();
for mix in mixnodes.into_inner() {
for mix in mixnodes {
let layer = mix.layer;
let mixes = layered_mixes.entry(layer).or_insert_with(Vec::new);
mixes.push(mix)
@@ -229,7 +230,7 @@ impl PacketPreparer {
}
async fn all_mixnodes_and_gateways(&self) -> (Vec<MixNodeBond>, Vec<GatewayBond>) {
info!(target: "Monitor", "Obtaining network topology...");
info!("Obtaining network topology...");
let mixnodes = self.validator_cache.mixnodes_all().await;
let gateways = self.validator_cache.gateways_all().await;
@@ -239,9 +240,6 @@ impl PacketPreparer {
pub(crate) fn try_parse_mix_bond(&self, mix: &MixNodeBond) -> Result<mix::Node, String> {
let identity = mix.mix_node.identity_key.clone();
if !self.check_version_compatibility(&mix.mix_node.version) {
return Err(identity);
}
mix.try_into().map_err(|_| identity)
}
@@ -250,9 +248,6 @@ impl PacketPreparer {
gateway: &GatewayBond,
) -> Result<gateway::Node, String> {
let identity = gateway.gateway.identity_key.clone();
if !self.check_version_compatibility(&gateway.gateway.version) {
return Err(identity);
}
gateway.try_into().map_err(|_| identity)
}
@@ -270,19 +265,10 @@ impl PacketPreparer {
// separate mixes into layers for easier selection
let mut layered_mixes = HashMap::new();
for mix in mixnodes {
// filter out mixes on the blacklist
if blacklist.contains(&mix.mix_node.identity_key) {
continue;
}
let layer = mix.layer;
let mixes = layered_mixes.entry(layer).or_insert_with(Vec::new);
mixes.push(mix)
}
// filter out gateways on the blacklist
let gateways = gateways
.into_iter()
.filter(|gateway| !blacklist.contains(&gateway.gateway.identity_key))
.collect::<Vec<_>>();
// get all nodes from each layer...
let l1 = layered_mixes.get(&Layer::One)?;
@@ -297,9 +283,6 @@ impl PacketPreparer {
let rand_l3 = l3.choose_multiple(&mut rng, n).collect::<Vec<_>>();
let rand_gateways = gateways.choose_multiple(&mut rng, n).collect::<Vec<_>>();
// let rand_gateways = gateways.iter().filter(|g| g.gateway.host == "109.74.196.254".to_string()).collect::<Vec<_>>();
// let rand_gateways = gateways.iter().filter(|g| g.gateway.host == "185.3.94.33".to_string()).collect::<Vec<_>>();
// the unwrap on `min()` is fine as we know the iterator is not empty
let most_available = *[
rand_l1.len(),
@@ -312,9 +295,10 @@ impl PacketPreparer {
.unwrap();
if most_available == 0 {
// it's impossible to generate a single route
error!("Cannot construct test routes. No nodes or gateways available");
None
} else {
trace!("Generating test routes...");
let mut routes = Vec::new();
for i in 0..most_available {
let node_1 = match self.try_parse_mix_bond(rand_l1[i]) {
@@ -358,14 +342,11 @@ impl PacketPreparer {
gateway,
))
}
info!("{:?}", routes);
Some(routes)
}
}
fn check_version_compatibility(&self, node_version: &str) -> bool {
version_checker::is_minor_version_compatible(node_version, &self.system_version)
}
fn create_packet_sender(&self, gateway: &gateway::Node) -> Recipient {
Recipient::new(
self.self_public_identity,
@@ -404,14 +385,6 @@ impl PacketPreparer {
let mut parsed_nodes = Vec::new();
let mut invalid_nodes = Vec::new();
for mixnode in nodes {
if !self.check_version_compatibility(&mixnode.mix_node.version) {
invalid_nodes.push(InvalidNode::Outdated(
mixnode.mix_node.identity_key,
mixnode.owner,
mixnode.mix_node.version,
));
continue;
}
if let Ok(parsed_node) = (&mixnode).try_into() {
parsed_nodes.push(parsed_node)
} else {
@@ -431,14 +404,6 @@ impl PacketPreparer {
let mut parsed_nodes = Vec::new();
let mut invalid_nodes = Vec::new();
for gateway in nodes {
if !self.check_version_compatibility(&gateway.gateway.version) {
invalid_nodes.push(InvalidNode::Outdated(
gateway.gateway.identity_key,
gateway.owner,
gateway.gateway.version,
));
continue;
}
if let Ok(parsed_node) = (&gateway).try_into() {
parsed_nodes.push(parsed_node)
} else {
@@ -460,18 +425,17 @@ impl PacketPreparer {
// (remember that "idle" nodes are still part of that set)
// we don't care about other nodes, i.e. nodes that are bonded but will not get
// any reward during the current rewarding interval
let (rewarded_set, all_gateways) = self.all_mixnodes_and_gateways().await;
let (mixnodes, gateways) = self.all_mixnodes_and_gateways().await;
let (mixes, invalid_mixnodes) = self.filter_outdated_and_malformed_mixnodes(rewarded_set);
let (gateways, invalid_gateways) =
self.filter_outdated_and_malformed_gateways(all_gateways);
let (mixnodes, invalid_mixnodes) = self.filter_outdated_and_malformed_mixnodes(mixnodes);
let (gateways, invalid_gateways) = self.filter_outdated_and_malformed_gateways(gateways);
let tested_mixnodes = mixes.iter().map(|node| node.into()).collect::<Vec<_>>();
let tested_mixnodes = mixnodes.iter().map(|node| node.into()).collect::<Vec<_>>();
let tested_gateways = gateways.iter().map(|node| node.into()).collect::<Vec<_>>();
let packets_to_create = (test_routes.len() * self.per_node_test_packets)
* (tested_mixnodes.len() + tested_gateways.len());
info!(target: "TestPreparer", "Need to create {} mix packets", packets_to_create);
info!("Need to create {} mix packets", packets_to_create);
let mut all_gateway_packets = HashMap::new();
@@ -483,10 +447,10 @@ impl PacketPreparer {
let gateway_owner = test_route.gateway_owner();
// it's actually going to be a tiny bit more due to gateway testing, but it's a good enough approximation
let mut mix_packets = Vec::with_capacity(mixes.len() * self.per_node_test_packets);
let mut mix_packets = Vec::with_capacity(mixnodes.len() * self.per_node_test_packets);
// and for each mixnode...
for mixnode in &mixes {
for mixnode in &mixnodes {
let test_packet = TestPacket::from_mixnode(mixnode, test_route.id(), test_nonce);
let topology = test_route.substitute_mix(mixnode);
// produce n mix packets
validator-api/src/network_monitor/monitor/receiver.rs
+9 -5
View File
@@ -1,7 +1,7 @@
// Copyright 2021 - Nym Technologies SA <contact@nymtech.net>
// SPDX-License-Identifier: Apache-2.0
use crate::network_monitor::gateways_reader::{GatewayChannel, GatewayMessages, GatewaysReader};
use crate::network_monitor::gateways_reader::{GatewayMessages, GatewaysReader};
use crate::network_monitor::monitor::processor::ReceivedProcessorSender;
use crypto::asymmetric::identity;
use futures::channel::mpsc;
@@ -40,10 +40,12 @@ impl PacketReceiver {
fn process_gateway_update(&mut self, update: GatewayClientUpdate) {
match update {
GatewayClientUpdate::New(id, (message_receiver, ack_receiver)) => {
let channel = GatewayChannel::new(id, message_receiver, ack_receiver);
self.gateways_reader.insert_channel(channel);
self.gateways_reader
.add_recievers(id, message_receiver, ack_receiver);
}
GatewayClientUpdate::Failure(id) => {
self.gateways_reader.remove_recievers(&id.to_string());
}
GatewayClientUpdate::Failure(id) => self.gateways_reader.remove_by_key(id),
}
}
@@ -62,7 +64,9 @@ impl PacketReceiver {
// similarly gateway reader will never return a `None` as it's implemented
// as an infinite stream that returns Poll::Pending if it doesn't have anything
// to return
messages = self.gateways_reader.next() => self.process_gateway_messages(messages.unwrap()),
Some((_gateway_id, message)) = self.gateways_reader.stream_map().next() => {
self.process_gateway_messages(message)
}
}
}
}
validator-api/src/network_monitor/monitor/sender.rs
+3 -5
View File
@@ -232,14 +232,13 @@ impl PacketSender {
) -> Result<(), GatewayClientError> {
let gateway_id = client.gateway_identity().to_base58_string();
info!(
target: "MessageSender",
"Got {} packets to send to gateway {}",
mix_packets.len(),
gateway_id
);
if mix_packets.len() <= max_sending_rate {
debug!(target: "MessageSender","Everything is going to get sent as one.");
debug!("Everything is going to get sent as one.");
client.batch_send_mix_packets(mix_packets).await?;
} else {
let packets_per_time_chunk =
@@ -248,7 +247,6 @@ impl PacketSender {
let total_expected_time =
Duration::from_secs_f64(mix_packets.len() as f64 / max_sending_rate as f64);
info!(
target: "MessageSender",
"With our rate of {} packets/s it should take around {:?} to send it all to {} ...",
max_sending_rate, total_expected_time, gateway_id
);
@@ -268,7 +266,7 @@ impl PacketSender {
// this way we won't have to do reallocations in here as they're unavoidable when
// splitting a vector into multiple vectors
while let Some(retained) = split_off_vec(&mut mix_packets, packets_per_time_chunk) {
trace!(target: "MessageSender","Sending {} packets...", mix_packets.len());
trace!("Sending {} packets...", mix_packets.len());
if mix_packets.len() == 1 {
client.send_mix_packet(mix_packets.pop().unwrap()).await?;
@@ -280,7 +278,7 @@ impl PacketSender {
mix_packets = retained;
}
debug!(target: "MessageSender", "Done sending");
debug!("Done sending");
}
Ok(())