From 1aacdfa08605aaac5caaff6f101656647397bf4b Mon Sep 17 00:00:00 2001 From: Johnathan Corgan Date: Sat, 11 Jul 2026 04:30:19 +0000 Subject: [PATCH] transport: share connection-pool counters via PoolCounters TcpStats and TorStats each carried an identical pair of pool_inbound/ pool_outbound atomics plus the same five record_pool_* / pool_inbound_count methods over them. Move the counter logic into a shared PoolCounters struct (new transport/stats_common.rs) embedded as a 'pool' field in both. The public record_pool_* methods stay as thin delegators so all call sites and the flat pool_inbound/pool_outbound snapshot fields are unchanged; only the duplicated atomic bookkeeping is now single-sourced. --- src/transport/mod.rs | 3 ++ src/transport/stats_common.rs | 56 +++++++++++++++++++++++++++++++++++ src/transport/tcp/stats.rs | 28 ++++++++---------- src/transport/tor/stats.rs | 29 +++++++++--------- 4 files changed, 86 insertions(+), 30 deletions(-) create mode 100644 src/transport/stats_common.rs diff --git a/src/transport/mod.rs b/src/transport/mod.rs index dadc520..9dacc17 100644 --- a/src/transport/mod.rs +++ b/src/transport/mod.rs @@ -34,6 +34,9 @@ use tor::TorTransport; use tor::control::TorMonitoringInfo; use udp::UdpTransport; +mod stats_common; +pub(crate) use stats_common::PoolCounters; + mod types; pub use types::*; diff --git a/src/transport/stats_common.rs b/src/transport/stats_common.rs new file mode 100644 index 0000000..c5e20c8 --- /dev/null +++ b/src/transport/stats_common.rs @@ -0,0 +1,56 @@ +//! Statistics helpers shared across transports. +//! +//! Counter logic that is identical across multiple transports lives here +//! rather than being copied into each transport's `stats.rs`. + +use portable_atomic::{AtomicU64, Ordering}; + +/// Inbound/outbound connection-pool occupancy counters. +/// +/// The connection-oriented transports (TCP, Tor) each track how many inbound +/// and outbound connections are currently held in their pool; the inbound +/// count drives the `max_inbound_connections` admission gate. The counter +/// logic is identical, so it is defined once here and embedded in each +/// transport's stats struct. +#[derive(Default)] +pub struct PoolCounters { + inbound: AtomicU64, + outbound: AtomicU64, +} + +impl PoolCounters { + /// Create counters at zero. + pub fn new() -> Self { + Self::default() + } + + /// Increment the inbound pool count (called on accept). + pub fn record_inbound_added(&self) { + self.inbound.fetch_add(1, Ordering::Relaxed); + } + + /// Decrement the inbound pool count (called on inbound receive-loop exit). + pub fn record_inbound_removed(&self) { + self.inbound.fetch_sub(1, Ordering::Relaxed); + } + + /// Increment the outbound pool count (called on connect-on-send / promote). + pub fn record_outbound_added(&self) { + self.outbound.fetch_add(1, Ordering::Relaxed); + } + + /// Decrement the outbound pool count (called on outbound receive-loop exit). + pub fn record_outbound_removed(&self) { + self.outbound.fetch_sub(1, Ordering::Relaxed); + } + + /// Load the current inbound pool count for the admission gate. + pub fn inbound_count(&self) -> u64 { + self.inbound.load(Ordering::Relaxed) + } + + /// Load the current outbound pool count. + pub fn outbound_count(&self) -> u64 { + self.outbound.load(Ordering::Relaxed) + } +} diff --git a/src/transport/tcp/stats.rs b/src/transport/tcp/stats.rs index 86785f8..62340e4 100644 --- a/src/transport/tcp/stats.rs +++ b/src/transport/tcp/stats.rs @@ -4,6 +4,8 @@ use portable_atomic::{AtomicU64, Ordering}; use serde::Serialize; +use crate::transport::PoolCounters; + /// Statistics for a TCP transport instance. /// /// Uses atomic counters for lock-free updates from per-connection @@ -21,12 +23,9 @@ pub struct TcpStats { pub connections_rejected: AtomicU64, pub connect_timeouts: AtomicU64, pub connect_refused: AtomicU64, - /// Current number of inbound (accepted) connections held in the pool. - /// Drives the `max_inbound_connections` admission check. - pub pool_inbound: AtomicU64, - /// Current number of outbound (connect-on-send / promoted) connections - /// held in the pool. Independent of the inbound cap. - pub pool_outbound: AtomicU64, + /// Inbound/outbound connection-pool occupancy. Inbound drives the + /// `max_inbound_connections` admission check. + pub pool: PoolCounters, } impl TcpStats { @@ -45,8 +44,7 @@ impl TcpStats { connections_rejected: AtomicU64::new(0), connect_timeouts: AtomicU64::new(0), connect_refused: AtomicU64::new(0), - pool_inbound: AtomicU64::new(0), - pool_outbound: AtomicU64::new(0), + pool: PoolCounters::new(), } } @@ -104,27 +102,27 @@ impl TcpStats { /// Increment the inbound pool count (called on accept). pub fn record_pool_inbound_added(&self) { - self.pool_inbound.fetch_add(1, Ordering::Relaxed); + self.pool.record_inbound_added(); } /// Decrement the inbound pool count (called on inbound receive-loop exit). pub fn record_pool_inbound_removed(&self) { - self.pool_inbound.fetch_sub(1, Ordering::Relaxed); + self.pool.record_inbound_removed(); } /// Increment the outbound pool count (called on connect-on-send / promote). pub fn record_pool_outbound_added(&self) { - self.pool_outbound.fetch_add(1, Ordering::Relaxed); + self.pool.record_outbound_added(); } /// Decrement the outbound pool count (called on outbound receive-loop exit). pub fn record_pool_outbound_removed(&self) { - self.pool_outbound.fetch_sub(1, Ordering::Relaxed); + self.pool.record_outbound_removed(); } /// Load the current inbound pool count for the admission gate. pub fn pool_inbound_count(&self) -> u64 { - self.pool_inbound.load(Ordering::Relaxed) + self.pool.inbound_count() } /// Take a snapshot of all counters. @@ -142,8 +140,8 @@ impl TcpStats { connections_rejected: self.connections_rejected.load(Ordering::Relaxed), connect_timeouts: self.connect_timeouts.load(Ordering::Relaxed), connect_refused: self.connect_refused.load(Ordering::Relaxed), - pool_inbound: self.pool_inbound.load(Ordering::Relaxed), - pool_outbound: self.pool_outbound.load(Ordering::Relaxed), + pool_inbound: self.pool.inbound_count(), + pool_outbound: self.pool.outbound_count(), } } } diff --git a/src/transport/tor/stats.rs b/src/transport/tor/stats.rs index 1838713..87a1790 100644 --- a/src/transport/tor/stats.rs +++ b/src/transport/tor/stats.rs @@ -4,6 +4,8 @@ use portable_atomic::{AtomicU64, Ordering}; use serde::Serialize; +use crate::transport::PoolCounters; + /// Statistics for a Tor transport instance. /// /// Uses atomic counters for lock-free updates from per-connection @@ -23,12 +25,10 @@ pub struct TorStats { pub connections_accepted: AtomicU64, pub connections_rejected: AtomicU64, pub control_errors: AtomicU64, - /// Current number of inbound (accepted via onion service) connections - /// held in the pool. Drives the `max_inbound_connections` admission check. - pub pool_inbound: AtomicU64, - /// Current number of outbound (SOCKS5 connect) connections held in - /// the pool. Independent of the inbound cap. - pub pool_outbound: AtomicU64, + /// Inbound (accepted via onion service) / outbound (SOCKS5 connect) + /// connection-pool occupancy. Inbound drives the + /// `max_inbound_connections` admission check. + pub pool: PoolCounters, } impl TorStats { @@ -49,8 +49,7 @@ impl TorStats { connections_accepted: AtomicU64::new(0), connections_rejected: AtomicU64::new(0), control_errors: AtomicU64::new(0), - pool_inbound: AtomicU64::new(0), - pool_outbound: AtomicU64::new(0), + pool: PoolCounters::new(), } } @@ -118,27 +117,27 @@ impl TorStats { /// Increment the inbound pool count (called on accept). pub fn record_pool_inbound_added(&self) { - self.pool_inbound.fetch_add(1, Ordering::Relaxed); + self.pool.record_inbound_added(); } /// Decrement the inbound pool count (called on inbound receive-loop exit). pub fn record_pool_inbound_removed(&self) { - self.pool_inbound.fetch_sub(1, Ordering::Relaxed); + self.pool.record_inbound_removed(); } /// Increment the outbound pool count (called on SOCKS5-connect promote). pub fn record_pool_outbound_added(&self) { - self.pool_outbound.fetch_add(1, Ordering::Relaxed); + self.pool.record_outbound_added(); } /// Decrement the outbound pool count (called on outbound receive-loop exit). pub fn record_pool_outbound_removed(&self) { - self.pool_outbound.fetch_sub(1, Ordering::Relaxed); + self.pool.record_outbound_removed(); } /// Load the current inbound pool count for the admission gate. pub fn pool_inbound_count(&self) -> u64 { - self.pool_inbound.load(Ordering::Relaxed) + self.pool.inbound_count() } /// Take a snapshot of all counters. @@ -158,8 +157,8 @@ impl TorStats { connections_accepted: self.connections_accepted.load(Ordering::Relaxed), connections_rejected: self.connections_rejected.load(Ordering::Relaxed), control_errors: self.control_errors.load(Ordering::Relaxed), - pool_inbound: self.pool_inbound.load(Ordering::Relaxed), - pool_outbound: self.pool_outbound.load(Ordering::Relaxed), + pool_inbound: self.pool.inbound_count(), + pool_outbound: self.pool.outbound_count(), } } }