diff --git a/src/transport/ethernet/stats.rs b/src/transport/ethernet/stats.rs index c20492f..0346232 100644 --- a/src/transport/ethernet/stats.rs +++ b/src/transport/ethernet/stats.rs @@ -2,6 +2,8 @@ use portable_atomic::{AtomicU64, Ordering}; +use serde::Serialize; + /// Statistics for an Ethernet transport instance. /// /// Uses atomic counters for lock-free updates from the receive loop @@ -92,7 +94,7 @@ impl Default for EthernetStats { } /// Point-in-time snapshot of Ethernet stats (non-atomic, copyable). -#[derive(Clone, Debug, Default)] +#[derive(Clone, Debug, Default, Serialize)] pub struct EthernetStatsSnapshot { pub frames_sent: u64, pub frames_recv: u64, diff --git a/src/transport/tcp/stream.rs b/src/transport/framing.rs similarity index 99% rename from src/transport/tcp/stream.rs rename to src/transport/framing.rs index 626b76e..44f6ca3 100644 --- a/src/transport/tcp/stream.rs +++ b/src/transport/framing.rs @@ -3,8 +3,8 @@ //! Recovers FIPS packet boundaries from a TCP byte stream using the //! existing 4-byte FMP common prefix `[ver+phase:1][flags:1][payload_len:2 LE]`. //! -//! This module is deliberately separate from the TCP transport so it can -//! be reused by the future Tor transport. +//! This module is deliberately separate from any single transport so it can +//! be shared by the stream-oriented transports (TCP, Tor, Nym). use tokio::io::{AsyncRead, AsyncReadExt}; diff --git a/src/transport/mod.rs b/src/transport/mod.rs index 5128ce8..1176215 100644 --- a/src/transport/mod.rs +++ b/src/transport/mod.rs @@ -34,6 +34,11 @@ use tor::TorTransport; use tor::control::TorMonitoringInfo; use udp::UdpTransport; +pub(crate) mod framing; + +mod stats_common; +pub(crate) use stats_common::PoolCounters; + mod types; pub use types::*; @@ -1022,19 +1027,7 @@ impl TransportHandle { } #[cfg(unix)] TransportHandle::Ethernet(t) => { - let snap = t.stats().snapshot(); - serde_json::json!({ - "frames_sent": snap.frames_sent, - "frames_recv": snap.frames_recv, - "bytes_sent": snap.bytes_sent, - "bytes_recv": snap.bytes_recv, - "send_errors": snap.send_errors, - "recv_errors": snap.recv_errors, - "beacons_sent": snap.beacons_sent, - "beacons_recv": snap.beacons_recv, - "frames_too_short": snap.frames_too_short, - "frames_too_long": snap.frames_too_long, - }) + serde_json::to_value(t.stats().snapshot()).unwrap_or_default() } TransportHandle::Tcp(t) => { serde_json::to_value(t.stats().snapshot()).unwrap_or_default() diff --git a/src/transport/nym/mod.rs b/src/transport/nym/mod.rs index a3190b2..becffb4 100644 --- a/src/transport/nym/mod.rs +++ b/src/transport/nym/mod.rs @@ -9,7 +9,7 @@ //! //! Outbound-only: connects to remote TCP peers through the local //! nym-socks5-client SOCKS5 proxy. Like the Tor transport, reuses FMP -//! stream framing from `tcp::stream` and follows the same connection +//! stream framing from `transport::framing` and follows the same connection //! pool pattern. No inbound service is supported. pub mod stats; @@ -22,7 +22,7 @@ use super::{ TransportError, TransportId, TransportState, TransportType, }; use crate::config::NymConfig; -use crate::transport::tcp::stream::read_fmp_packet; +use crate::transport::framing::read_fmp_packet; use stats::NymStats; use futures::FutureExt; 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/mod.rs b/src/transport/tcp/mod.rs index 74529b4..e4e94bb 100644 --- a/src/transport/tcp/mod.rs +++ b/src/transport/tcp/mod.rs @@ -23,7 +23,6 @@ //! TCP stream and the receiver uses phase-dependent size computation. pub mod stats; -pub mod stream; use super::resolve_socket_addr; use super::{ @@ -31,8 +30,8 @@ use super::{ TransportError, TransportId, TransportState, TransportType, }; use crate::config::TcpConfig; +use crate::transport::framing::read_fmp_packet; use stats::TcpStats; -use stream::read_fmp_packet; use futures::FutureExt; use socket2::TcpKeepalive; 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/mod.rs b/src/transport/tor/mod.rs index 9dd50e8..5eae40f 100644 --- a/src/transport/tor/mod.rs +++ b/src/transport/tor/mod.rs @@ -14,8 +14,8 @@ //! ## Architecture //! //! Like TCP, each peer has its own connection. The transport reuses FMP -//! stream framing from `tcp::stream` and follows the same connection pool -//! pattern as the TCP transport. Inbound connections arrive via a local +//! stream framing from `transport::framing` and follows the same connection +//! pool pattern as the TCP transport. Inbound connections arrive via a local //! TCP listener that the Tor daemon forwards onion service traffic to. pub mod control; @@ -31,7 +31,7 @@ use super::{ TransportError, TransportId, TransportState, TransportType, }; use crate::config::TorConfig; -use crate::transport::tcp::stream::read_fmp_packet; +use crate::transport::framing::read_fmp_packet; use control::{ControlAuth, TorControlClient, TorMonitoringInfo}; use stats::TorStats; 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(), } } }