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

remove double accounting for moment

Browse Source
This commit is contained in:
mfahampshire
2024-11-26 15:30:48 +01:00
parent 6f4421dc4a
commit 5083b420ec
3 changed files with 154 additions and 151 deletions
Download Patch File Download Diff File Expand all files Collapse all files
sdk/rust/nym-sdk/examples/tcp_proxy_single_connection.rs
+1 -1
View File
@@ -60,7 +60,7 @@ async fn main() -> anyhow::Result<()> {
// We'll run the instance with a long timeout since we're sending everything down the same Tcp connection, so should be using a single session.
// Within the TcpProxyClient, individual client shutdown is triggered by the timeout.
let proxy_client =
tcp_proxy::NymProxyClient::new(*proxy_nym_addr, "127.0.0.1", &client_port, 5, Some(env), 2)
tcp_proxy::NymProxyClient::new(*proxy_nym_addr, "127.0.0.1", &client_port, 5, Some(env), 1)
.await?;
tokio::spawn(async move {
sdk/rust/nym-sdk/src/tcp_proxy/client_pool.rs
+136 -132
View File
@@ -12,8 +12,7 @@ use tracing::{debug, info, warn};
pub struct ClientPool {
clients: Arc<RwLock<Vec<Arc<MixnetClient>>>>,
semaphore: Arc<Semaphore>,
default_pool_size: usize, // default # of clients to have running simultaneously, otherwise make ephemeral
conn_count: Arc<AtomicUsize>, // the actual # of connections running, denoting an incoming tcp request that is matched with a nym client
default_pool_size: usize, // default # of clients to have available in pool. If incoming tcp requests > this, make ephemeral client elsewhere
}
impl fmt::Debug for ClientPool {
@@ -45,7 +44,7 @@ impl fmt::Debug for ClientPool {
let mut debug_struct = f.debug_struct("Pool");
debug_struct
.field("default_pool_size", &self.default_pool_size)
.field("connection count", &*self.conn_count)
// .field("connection count", &*self.conn_count)
.field("clients", &format_args!("[{}]", clients_debug));
debug_struct.finish()
@@ -58,7 +57,7 @@ impl ClientPool {
clients: Arc::new(RwLock::new(Vec::new())),
semaphore: Arc::new(Semaphore::new(default_pool_size)),
default_pool_size,
conn_count: Arc::new(AtomicUsize::new(0)),
// conn_count: Arc::new(AtomicUsize::new(0)),
}
}
@@ -70,8 +69,12 @@ impl ClientPool {
"Currently spawned clients: {}: {:?} ",
spawned_clients, addresses
);
info!(
"current avail permits {}",
self.semaphore.available_permits()
);
// TODO check this logic in situation with larger pool
// TODO FIX THIS
if spawned_clients == self.semaphore.available_permits() {
debug!("Got enough clients already: sleeping");
} else {
@@ -118,158 +121,159 @@ impl ClientPool {
self.clients.read().await.len()
}
pub fn get_conn_count(&self) -> usize {
self.conn_count.load(Ordering::SeqCst)
}
// pub fn get_conn_count(&self) -> usize {
// // self.conn_count.load(Ordering::SeqCst)
// self.default_pool_size - self.semaphore.available_permits()
// }
pub fn increment_conn_count(&self) {
self.conn_count.fetch_add(1, Ordering::SeqCst);
}
// pub fn increment_conn_count(&self) {
// self.conn_count.fetch_add(1, Ordering::SeqCst);
// }
pub fn decrement_conn_count(&self) -> Result<()> {
if self.get_conn_count() == 0 {
bail!("count already 0");
}
self.conn_count.fetch_sub(1, Ordering::SeqCst);
Ok(())
}
// pub fn decrement_conn_count(&self) -> Result<()> {
// if self.get_conn_count() == 0 {
// bail!("count already 0");
// }
// self.conn_count.fetch_sub(1, Ordering::SeqCst);
// Ok(())
// }
pub fn clone(&self) -> Self {
Self {
clients: Arc::clone(&self.clients),
semaphore: Arc::clone(&self.semaphore),
default_pool_size: *&self.default_pool_size,
conn_count: Arc::clone(&self.conn_count),
// conn_count: Arc::clone(&self.conn_count),
}
}
}
// TODO COVER ALL FNS
#[cfg(test)]
mod tests {
use super::*;
use anyhow::Result;
use std::thread;
// #[cfg(test)]
// mod tests {
// use super::*;
// use anyhow::Result;
// use std::thread;
#[test]
fn test_conn_count_increment_decrement() -> Result<()> {
let tracker = ClientPool::new(0);
tracker.increment_conn_count();
tracker.increment_conn_count();
assert_eq!(
tracker.get_conn_count(),
2,
"should be 2 after single increment"
);
tracker.decrement_conn_count()?;
assert_eq!(
tracker.get_conn_count(),
1,
"should be 1 after two increments and one decrement"
);
Ok(())
}
#[test]
fn test_clone() {
let tracker = ClientPool::new(1);
let tracker_clone = tracker.clone();
// #[test]
// fn test_conn_count_increment_decrement() -> Result<()> {
// let tracker = ClientPool::new(0);
// tracker.increment_conn_count();
// tracker.increment_conn_count();
// assert_eq!(
// tracker.get_conn_count(),
// 2,
// "should be 2 after single increment"
// );
// tracker.decrement_conn_count()?;
// assert_eq!(
// tracker.get_conn_count(),
// 1,
// "should be 1 after two increments and one decrement"
// );
// Ok(())
// }
// #[test]
// fn test_clone() {
// let tracker = ClientPool::new(1);
// let tracker_clone = tracker.clone();
tracker.increment_conn_count();
assert_eq!(
tracker_clone.get_conn_count(),
1,
"tracker clones should share the same count"
);
}
// tracker.increment_conn_count();
// assert_eq!(
// tracker_clone.get_conn_count(),
// 1,
// "tracker clones should share the same count"
// );
// }
#[test]
fn test_conn_count_multiple_threads() {
let tracker = ClientPool::new(0);
let mut handles = vec![];
// #[test]
// fn test_conn_count_multiple_threads() {
// let tracker = ClientPool::new(0);
// let mut handles = vec![];
for _ in 0..10 {
let thread_tracker = tracker.clone();
let handle = thread::spawn(move || {
thread_tracker.increment_conn_count();
thread::sleep(std::time::Duration::from_millis(10));
});
handles.push(handle);
}
// for _ in 0..10 {
// let thread_tracker = tracker.clone();
// let handle = thread::spawn(move || {
// thread_tracker.increment_conn_count();
// thread::sleep(std::time::Duration::from_millis(10));
// });
// handles.push(handle);
// }
for handle in handles {
handle.join().unwrap();
}
// for handle in handles {
// handle.join().unwrap();
// }
assert_eq!(
tracker.get_conn_count(),
10,
"should be 10 after 10 thread increments"
);
}
// assert_eq!(
// tracker.get_conn_count(),
// 10,
// "should be 10 after 10 thread increments"
// );
// }
#[test]
fn test_concurrent_increment_decrement() -> Result<()> {
let tracker = ClientPool::new(0);
let mut handles = vec![];
// #[test]
// fn test_concurrent_increment_decrement() -> Result<()> {
// let tracker = ClientPool::new(0);
// let mut handles = vec![];
for i in 0..10 {
let thread_tracker = tracker.clone();
let handle = thread::spawn(move || {
if i < 5 {
thread_tracker.increment_conn_count();
} else {
thread_tracker.decrement_conn_count().unwrap();
}
thread::sleep(std::time::Duration::from_millis(10));
});
handles.push(handle);
}
// for i in 0..10 {
// let thread_tracker = tracker.clone();
// let handle = thread::spawn(move || {
// if i < 5 {
// thread_tracker.increment_conn_count();
// } else {
// thread_tracker.decrement_conn_count().unwrap();
// }
// thread::sleep(std::time::Duration::from_millis(10));
// });
// handles.push(handle);
// }
for handle in handles {
handle.join().unwrap();
}
// for handle in handles {
// handle.join().unwrap();
// }
assert_eq!(
tracker.get_conn_count(),
0,
"should be 0 after equal increments and decrements"
);
Ok(())
}
// assert_eq!(
// tracker.get_conn_count(),
// 0,
// "should be 0 after equal increments and decrements"
// );
// Ok(())
// }
#[test]
#[should_panic]
fn test_zero_floor() {
let tracker = ClientPool::new(0);
tracker.decrement_conn_count().unwrap();
}
// #[test]
// #[should_panic]
// fn test_zero_floor() {
// let tracker = ClientPool::new(0);
// tracker.decrement_conn_count().unwrap();
// }
#[test]
fn test_stress() {
let tracker = ClientPool::new(0);
let mut handles = vec![];
let num_threads = 100;
// #[test]
// fn test_stress() {
// let tracker = ClientPool::new(0);
// let mut handles = vec![];
// let num_threads = 100;
for _ in 0..num_threads {
let thread_tracker = tracker.clone();
let handle = thread::spawn(move || {
for _ in 0..100 {
thread_tracker.increment_conn_count();
thread::sleep(std::time::Duration::from_micros(1));
thread_tracker.decrement_conn_count().unwrap();
}
});
handles.push(handle);
}
// for _ in 0..num_threads {
// let thread_tracker = tracker.clone();
// let handle = thread::spawn(move || {
// for _ in 0..100 {
// thread_tracker.increment_conn_count();
// thread::sleep(std::time::Duration::from_micros(1));
// thread_tracker.decrement_conn_count().unwrap();
// }
// });
// handles.push(handle);
// }
for handle in handles {
handle.join().unwrap();
}
// for handle in handles {
// handle.join().unwrap();
// }
assert_eq!(
tracker.get_conn_count(),
0,
"should return to 0 after all increments and decrements"
);
}
}
// assert_eq!(
// tracker.get_conn_count(),
// 0,
// "should return to 0 after all increments and decrements"
// );
// }
// }
sdk/rust/nym-sdk/src/tcp_proxy/tcp_proxy_client.rs
+17 -18
View File
@@ -75,15 +75,14 @@ impl NymProxyClient {
let client_maker = self.conn_pool.clone();
tokio::spawn(async move { client_maker.start().await.unwrap() });
let overall_counter = self.conn_pool.clone();
tokio::spawn(async move {
loop {
info!("Active connections: {}", overall_counter.get_conn_count());
tokio::time::sleep(Duration::from_secs(5)).await;
}
});
// let overall_counter = self.conn_pool.clone();
// tokio::spawn(async move {
// loop {
// info!("Active connections: {}", overall_counter.get_conn_count());
// tokio::time::sleep(Duration::from_secs(5)).await;
// }
// });
// if self.conn_pool.get_client_count().await >= DEFAULT_CLIENT_POOL_SIZE / 2 {
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
@@ -142,11 +141,11 @@ impl NymProxyClient {
}
};
conn_pool.increment_conn_count();
info!(
"New connection - current active connections: {}",
conn_pool.get_conn_count()
);
// conn_pool.increment_conn_count();
// info!(
// "New connection - current active connections: {}",
// conn_pool.get_conn_count()
// );
// Split our tcpstream into OwnedRead and OwnedWrite halves for concurrent read/writing
let (read, mut write) = stream.into_split();
@@ -245,11 +244,11 @@ impl NymProxyClient {
info!(" Closing write end of session: {}", session_id);
info!(" Triggering client shutdown");
conn_pool.disconnect_and_remove_client(client).await?;
conn_pool.clone().decrement_conn_count()?;
info!(
"Dropped connection - current active connections: {}",
conn_pool.get_conn_count()
);
// conn_pool.clone().decrement_conn_count()?;
// info!(
// "Dropped connection - current active connections: {}",
// conn_pool.get_conn_count()
// );
return Ok::<(), anyhow::Error>(())
}
}