diff --git a/sdk/rust/nym-sdk/examples/tcp_proxy_multistream.rs b/sdk/rust/nym-sdk/examples/tcp_proxy_multistream.rs index 697ac2071c..c1cc012324 100644 --- a/sdk/rust/nym-sdk/examples/tcp_proxy_multistream.rs +++ b/sdk/rust/nym-sdk/examples/tcp_proxy_multistream.rs @@ -37,7 +37,7 @@ async fn main() -> anyhow::Result<()> { // to DEBUG to see the contents of the msg buffer, sphinx packet chunking, etc. tracing_subscriber::registry() .with(fmt::layer()) - .with(EnvFilter::new("nym_sdk::tcp_proxy=info")) + .with(EnvFilter::new("nym_sdk::tcp_proxy=warn")) .init(); let env_path = env::args().nth(2).expect("Env file not specified"); @@ -45,9 +45,9 @@ async fn main() -> anyhow::Result<()> { let listen_port = env::args().nth(3).expect("Port not specified"); - // Within the TcpProxyClient, individual client shutdown is triggered by the timeout. + // Within the TcpProxyClient, individual client shutdown is triggered by the timeout. The final argument is how many clients to keep in reserve in the client pool. let proxy_client = - tcp_proxy::NymProxyClient::new(server, "127.0.0.1", &listen_port, 45, Some(env), 5).await?; + tcp_proxy::NymProxyClient::new(server, "127.0.0.1", &listen_port, 45, Some(env), 2).await?; tokio::spawn(async move { proxy_client.run().await?; @@ -60,7 +60,7 @@ async fn main() -> anyhow::Result<()> { println!("done. sending bytes"); // In the info traces you will see the different session IDs being set up, one for each TcpStream. - for i in 0..5 { + for i in 0..8 { let conn_id = i; let local_tcp_addr = format!("127.0.0.1:{}", listen_port.clone()); tokio::spawn(async move { @@ -77,7 +77,7 @@ async fn main() -> anyhow::Result<()> { // Lets just send a bunch of messages to the server with variable delays between them, with a message and tcp connection ids to keep track of ordering on the server side (for illustrative purposes **only**; keeping track of anonymous replies is handled by the proxy under the hood with Single Use Reply Blocks (SURBs); for this illustration we want some kind of app-level message id, but irl most of the time you'll probably be parsing on e.g. the incoming response type instead) tokio::spawn(async move { - for i in 0..4 { + for i in 0..8 { let mut rng = SmallRng::from_entropy(); let delay: f64 = rng.gen_range(2.5..5.0); tokio::time::sleep(tokio::time::Duration::from_secs_f64(delay)).await; @@ -104,17 +104,8 @@ async fn main() -> anyhow::Result<()> { while let Some(Ok(bytes)) = framed_read.next().await { match bincode::deserialize::(&bytes) { Ok(msg) => { - // println!( - // "<< client received {}: {} bytes on conn {}", - // msg.message_id, - // msg.message_bytes.len(), - // msg.tcp_conn - // ); reply_counter += 1; - println!( - "<< connection {} received reply {}/4", - msg.tcp_conn, reply_counter - ); + println!("<< conn {} received {}/8", msg.tcp_conn, reply_counter); } Err(e) => { println!("<< client received something that wasn't an example message of {} bytes. error: {}", bytes.len(), e); @@ -137,7 +128,7 @@ async fn main() -> anyhow::Result<()> { // emulate a series of small messages followed by a closing larger one fn gen_bytes_fixed(i: usize) -> Vec { - let amounts = [10, 15, 50, 1000]; + let amounts = [10, 15, 50, 1000, 10, 15, 500, 2000]; let len = amounts[i]; let mut rng = rand::thread_rng(); (0..len).map(|_| rng.gen::()).collect() diff --git a/sdk/rust/nym-sdk/src/tcp_proxy/client_pool.rs b/sdk/rust/nym-sdk/src/tcp_proxy/client_pool.rs index d0f69ddfd0..90667c1be0 100644 --- a/sdk/rust/nym-sdk/src/tcp_proxy/client_pool.rs +++ b/sdk/rust/nym-sdk/src/tcp_proxy/client_pool.rs @@ -3,15 +3,11 @@ use anyhow::Result; use std::fmt; use std::sync::Arc; use tokio::sync::RwLock; -use tokio::sync::Semaphore; use tracing::{debug, info, warn}; -// Make a set # of clients (low default) -// Once a client is used, kill the client & remove it from the pool pub struct ClientPool { - clients: Arc>>>, - semaphore: Arc, - default_pool_size: usize, // default # of clients to have available in pool. If incoming tcp requests > this, make ephemeral client elsewhere + clients: Arc>>>, // collection of clients waiting to be used which are popped off in get_mixnet_client() + client_pool_reserve_number: usize, // default # of clients to have available in pool in reserve waiting for incoming connections } impl fmt::Debug for ClientPool { @@ -42,18 +38,20 @@ impl fmt::Debug for ClientPool { let mut debug_struct = f.debug_struct("Pool"); debug_struct - .field("default_pool_size", &self.default_pool_size) + .field( + "client_pool_reserve_number", + &self.client_pool_reserve_number, + ) .field("clients", &format_args!("[{}]", clients_debug)); debug_struct.finish() } } impl ClientPool { - pub fn new(default_pool_size: usize) -> Self { + pub fn new(client_pool_reserve_number: usize) -> Self { ClientPool { clients: Arc::new(RwLock::new(Vec::new())), - semaphore: Arc::new(Semaphore::new(default_pool_size)), - default_pool_size, + client_pool_reserve_number, } } @@ -61,23 +59,17 @@ impl ClientPool { loop { let spawned_clients = self.clients.read().await.len(); let addresses = self; - info!( + debug!( "Currently spawned clients: {}: {:?} ", spawned_clients, addresses ); - // TODO PROBLEM IS HERE: not updating / tracking the in use permits when grab_mixnet_client is called - info!( - "current avail permits {}", - self.semaphore.available_permits() - ); - info!( - "current in use permits {}", - self.default_pool_size - self.semaphore.available_permits() - ); - if spawned_clients == self.semaphore.available_permits() { + if spawned_clients >= self.client_pool_reserve_number { debug!("Got enough clients already: sleeping"); } else { - info!("Spawning new client"); + info!( + "Clients in reserve = {}, reserve amount = {}, spawning new client", + spawned_clients, self.client_pool_reserve_number + ); let client = loop { let net = NymNetworkDetails::new_from_env(); match MixnetClientBuilder::new_ephemeral() @@ -95,17 +87,13 @@ impl ClientPool { }; self.clients.write().await.push(Arc::new(client)); } - tokio::time::sleep(tokio::time::Duration::from_secs(1)).await; + tokio::time::sleep(tokio::time::Duration::from_millis(100)).await; } } pub async fn get_mixnet_client(&self) -> Option { - info!("Grabbing client from pool"); - let permit = self.semaphore.acquire().await; - info!("{permit:?}"); - info!("Available permits: {}", self.semaphore.available_permits()); + debug!("Grabbing client from pool"); let mut clients = self.clients.write().await; - // gain ownership of client, tracking with semaphore once its working to stop constantly renewing size of pool to default_pool_size and instead have pool be (default_pool_size - in use clients) to stop bloat clients .pop() .and_then(|arc_client| Arc::try_unwrap(arc_client).ok()) @@ -124,8 +112,7 @@ impl ClientPool { pub fn clone(&self) -> Self { Self { clients: Arc::clone(&self.clients), - semaphore: Arc::clone(&self.semaphore), - default_pool_size: *&self.default_pool_size, + client_pool_reserve_number: *&self.client_pool_reserve_number, } } } diff --git a/sdk/rust/nym-sdk/src/tcp_proxy/tcp_proxy_client.rs b/sdk/rust/nym-sdk/src/tcp_proxy/tcp_proxy_client.rs index 027d039169..0ebc0fa3ee 100644 --- a/sdk/rust/nym-sdk/src/tcp_proxy/tcp_proxy_client.rs +++ b/sdk/rust/nym-sdk/src/tcp_proxy/tcp_proxy_client.rs @@ -1,13 +1,11 @@ -use crate::mixnet::{ - IncludedSurbs, MixnetClient, MixnetClientBuilder, MixnetMessageSender, NymNetworkDetails, -}; -use std::{sync::Arc, time::Duration}; +use crate::mixnet::{IncludedSurbs, MixnetClientBuilder, MixnetMessageSender, NymNetworkDetails}; +use std::sync::Arc; #[path = "client_pool.rs"] mod client_pool; use client_pool::ClientPool; #[path = "utils.rs"] mod utils; -use anyhow::{bail, Result}; +use anyhow::Result; use dashmap::DashSet; use nym_network_defaults::setup_env; use nym_sphinx::addressing::Recipient; @@ -85,6 +83,7 @@ impl NymProxyClient { }); // TODO add 'ready' marker for consuming code + loop { if DEFAULT_CLIENT_POOL_SIZE == 1 && self.conn_pool.get_client_count().await == 1 || self.conn_pool.get_client_count().await >= DEFAULT_CLIENT_POOL_SIZE / 2 @@ -133,13 +132,18 @@ impl NymProxyClient { client } None => { - info!("IF YOU SEE THIS not enough clients in pool, creating ephemeral client"); + warn!("Not enough clients in pool, creating ephemeral client"); let net = NymNetworkDetails::new_from_env(); - MixnetClientBuilder::new_ephemeral() + let client = MixnetClientBuilder::new_ephemeral() .network_details(net) .build()? .connect_to_mixnet() - .await? + .await?; + warn!( + "Using {} for the moment, created outside of the connection pool", + client.nym_address() + ); + client } };