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

first version working

Browse Source
This commit is contained in:
mfahampshire
2024-11-27 18:59:58 +01:00
parent 8c8e085796
commit c49f7f91ed
3 changed files with 36 additions and 54 deletions
Download Patch File Download Diff File Expand all files Collapse all files
sdk/rust/nym-sdk/examples/tcp_proxy_multistream.rs
+7 -16
View File
@@ -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::<ExampleMessage>(&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<u8> {
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::<u8>()).collect()
sdk/rust/nym-sdk/src/tcp_proxy/client_pool.rs
+17 -30
View File
@@ -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<RwLock<Vec<Arc<MixnetClient>>>>,
semaphore: Arc<Semaphore>,
default_pool_size: usize, // default # of clients to have available in pool. If incoming tcp requests > this, make ephemeral client elsewhere
clients: Arc<RwLock<Vec<Arc<MixnetClient>>>>, // 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<MixnetClient> {
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,
}
}
}
sdk/rust/nym-sdk/src/tcp_proxy/tcp_proxy_client.rs
+12 -8
View File
@@ -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
}
};