From 54d7b37f1024fba33bc448986fd90acb4ec7dffa Mon Sep 17 00:00:00 2001 From: mfahampshire Date: Fri, 3 Jan 2025 18:28:28 +0100 Subject: [PATCH] debugging failing lock quiring --- tools/echo-server/src/lib.rs | 13 +- .../mixnet-check-all-gateways/src/main.rs | 182 +++++++++--------- 2 files changed, 103 insertions(+), 92 deletions(-) diff --git a/tools/echo-server/src/lib.rs b/tools/echo-server/src/lib.rs index 1a706b17f4..c89131062c 100644 --- a/tools/echo-server/src/lib.rs +++ b/tools/echo-server/src/lib.rs @@ -2,6 +2,7 @@ // SPDX-License-Identifier: GPL-3.0-only use anyhow::Result; +use clap::builder::styling::Reset; use nym_crypto::asymmetric::ed25519; use nym_sdk::mixnet::Recipient; use nym_sdk::tcp_proxy; @@ -12,6 +13,7 @@ use tokio::io::AsyncWriteExt; use tokio::net::{TcpListener, TcpStream}; use tokio::sync::Mutex; use tokio::task; +use tokio::time::{timeout, Duration}; use tokio_stream::StreamExt; use tokio_util::sync::CancellationToken; use tracing::{debug, error, info, warn}; @@ -108,7 +110,7 @@ impl NymEchoServer { )); } _ = self.cancel_token.cancelled() => { - info!("Cancel token cancelled: {:?}", self.cancel_token.cancelled()); + info!("token cancelled, stopping handling streams"); break Ok(()); } } @@ -155,8 +157,15 @@ impl NymEchoServer { pub async fn disconnect(&self) { self.cancel_token.cancel(); + info!("token cancelled"); let client = Arc::clone(&self.client); - client.lock().await.disconnect().await; + info!("acquiring lock"); + if let Ok(guard) = timeout(Duration::from_secs(5), client.lock()).await { + guard.disconnect().await; + } else { + error!("Failed to acquire lock to trigger shutdown"); + // TODO somehow force aquire lock and kill + }; } pub async fn nym_address(&self) -> Recipient { diff --git a/tools/internal/mixnet-check-all-gateways/src/main.rs b/tools/internal/mixnet-check-all-gateways/src/main.rs index a308b032ef..77762eac89 100644 --- a/tools/internal/mixnet-check-all-gateways/src/main.rs +++ b/tools/internal/mixnet-check-all-gateways/src/main.rs @@ -21,6 +21,7 @@ use tokio::time::timeout; #[path = "utils.rs"] // TODO make these exportable from tcp_proxy module and then import from there, ditto with echo server lib mod utils; +use nym_sdk::client_pool::ClientPool; use std::time::{SystemTime, UNIX_EPOCH}; use tokio_util::sync::CancellationToken; use tracing::{debug, info, warn}; @@ -79,8 +80,6 @@ async fn main() -> Result<()> { let cancel_token = CancellationToken::new(); let watcher_token = cancel_token.clone(); - - // Cancel listener thread tokio::spawn(async move { signal::ctrl_c().await?; println!("CTRL_C received"); @@ -151,13 +150,13 @@ async fn main() -> Result<()> { let echo_addr = echo_server.nym_address().await; debug!("echo addr: {echo_addr}"); - // TODO change this to mpsc to do proper cancellation tokio::task::spawn(async move { loop { tokio::select! { _ = thread_echo_server_token.cancelled() => { info!("loop over; disconnecting echo server {}", echo_addr.clone()); - echo_server.disconnect().await; + echo_server.disconnect().await; // TODO fix this failing lock: can't aquire lock on the ProxyServer to kill it + info!("disconnected {}", echo_addr.clone()); break; } _ = echo_server.run() => {} @@ -170,100 +169,103 @@ async fn main() -> Result<()> { time::sleep(Duration::from_secs(5)).await; for entry_gw in entry_gw_keys.clone() { - let builder = mixnet::MixnetClientBuilder::new_ephemeral() - .request_gateway(entry_gw.clone()) - .build()?; + // let builder = mixnet::MixnetClientBuilder::new_ephemeral() + // .request_gateway(entry_gw.clone()) + // .build()?; - let mut client = match builder.connect_to_mixnet().await { - Ok(client) => client, - Err(err) => { - let res = TestResult { - entry_gw: entry_gw.clone(), - exit_gw: exit_gw.clone(), - error: TestError::Other(err.to_string()), - }; - info!("{res:#?}"); - results_vec.push(json!(res)); - continue; - } - }; + // let mut client = match builder.connect_to_mixnet().await { + // Ok(client) => client, + // Err(err) => { + // let res = TestResult { + // entry_gw: entry_gw.clone(), + // exit_gw: exit_gw.clone(), + // error: TestError::Other(err.to_string()), + // }; + // info!("{res:#?}"); + // results_vec.push(json!(res)); + // continue; + // } + // }; - let test_address = client.nym_address(); - info!("currently testing entry gateway: {test_address}"); + // let test_address = client.nym_address(); + // info!("currently testing entry gateway: {test_address}"); - // Has to be ProxiedMessage for the moment which is slightly annoying until I - // modify the ProxyServer to just stupidly echo back whatever it gets in a - // ReconstructedMessage format if it can't deseralise incoming traffic to a ProxiedMessage - let session_id = uuid::Uuid::new_v4(); - let message_id = 0; - let outgoing = ProxiedMessage::new( - Payload::Data(MESSAGE.as_bytes().to_vec()), - session_id, - message_id, - ); - let coded_message = bincode::serialize(&outgoing).unwrap(); + // // Has to be ProxiedMessage for the moment which is slightly annoying until I + // // modify the ProxyServer to just stupidly echo back whatever it gets in a + // // ReconstructedMessage format if it can't deseralise incoming traffic to a ProxiedMessage + // let session_id = uuid::Uuid::new_v4(); + // let message_id = 0; + // let outgoing = ProxiedMessage::new( + // Payload::Data(MESSAGE.as_bytes().to_vec()), + // session_id, + // message_id, + // ); + // let coded_message = bincode::serialize(&outgoing).unwrap(); - match client - .send_message(echo_addr, &coded_message, IncludedSurbs::Amount(10)) - .await - { - Ok(_) => { - debug!("Message sent"); - } - Err(err) => { - let res = TestResult { - entry_gw: entry_gw.clone(), - exit_gw: exit_gw.clone(), - error: TestError::Other(err.to_string()), - }; - info!("{res:#?}"); - results_vec.push(json!(res)); - continue; - } - }; + // match client + // .send_message(echo_addr, &coded_message, IncludedSurbs::Amount(10)) + // .await + // { + // Ok(_) => { + // debug!("Message sent"); + // } + // Err(err) => { + // let res = TestResult { + // entry_gw: entry_gw.clone(), + // exit_gw: exit_gw.clone(), + // error: TestError::Other(err.to_string()), + // }; + // info!("{res:#?}"); + // results_vec.push(json!(res)); + // continue; + // } + // }; - let res = match timeout(Duration::from_secs(TIMEOUT), client.next()).await { - Err(_timeout) => { - warn!("timed out"); - TestResult { - entry_gw, - exit_gw: exit_gw.clone(), - error: TestError::Timeout, - } - } - Ok(Some(received)) => { - let incoming: ProxiedMessage = bincode::deserialize(&received.message).unwrap(); - debug!("got echo: {:?}", incoming); - info!( - "sent message as lazy ref until I properly sort the utils for comparison: {:?}", - MESSAGE.as_bytes() - ); - info!("incoming message: {:?}", incoming.message); - // TODO check incoming is same as outgoing else make a MangledReply err type + ERR log - TestResult { - entry_gw, - exit_gw: exit_gw.clone(), - error: TestError::None, - } - } - Ok(None) => { - info!("failed to receive any message back..."); - TestResult { - entry_gw, - exit_gw: exit_gw.clone(), - error: TestError::NoMessage, - } - } - }; - info!("{res:#?}"); - results_vec.push(json!(res)); - debug!("disconnecting the client before shutting down..."); - client.disconnect().await; + // let res = match timeout(Duration::from_secs(TIMEOUT), client.next()).await { + // Err(_timeout) => { + // warn!("timed out"); + // TestResult { + // entry_gw, + // exit_gw: exit_gw.clone(), + // error: TestError::Timeout, + // } + // } + // Ok(Some(received)) => { + // let incoming: ProxiedMessage = bincode::deserialize(&received.message).unwrap(); + // debug!("got echo: {:?}", incoming); + // info!( + // "sent message as lazy ref until I properly sort the utils for comparison: {:?}", + // MESSAGE.as_bytes() + // ); + // info!("incoming message: {:?}", incoming.message); + // // TODO check incoming is same as outgoing else make a MangledReply err type + ERR log + // TestResult { + // entry_gw, + // exit_gw: exit_gw.clone(), + // error: TestError::None, + // } + // } + // Ok(None) => { + // info!("failed to receive any message back..."); + // TestResult { + // entry_gw, + // exit_gw: exit_gw.clone(), + // error: TestError::NoMessage, + // } + // } + // }; + // info!("{res:#?}"); + // results_vec.push(json!(res)); + // debug!("disconnecting the client before shutting down..."); + // client.disconnect().await; + + if Some(&entry_gw) == entry_gw_keys.last() { + echo_server_token.cancel(); + } } let json_array = json!(results_vec); debug!("{json_array}"); results.write_all(json_array.to_string().as_bytes())?; - echo_server_token.cancel(); } Ok(()) } @@ -281,7 +283,7 @@ fn filter_gateway_keys(json: &Value, key: &str) -> Result> { if let Some(nodes) = json["nodes"]["data"].as_array() { for node in nodes { if let Some(performance) = node.get("performance").and_then(|v| v.as_str()) { - let performance_value: f64 = performance.parse().unwrap_or(0.0); // TODO could make this a const @ top? + let performance_value: f64 = performance.parse().unwrap_or(0.0); // TODO make this configurable let inactive = node.get("role").and_then(|v| v.as_str()) == Some("Inactive");