diff --git a/CHANGELOG.md b/CHANGELOG.md index 2e0ecaa..f16f479 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -11,6 +11,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - The amount of bytes read and written to a peer are now no longer reset after a reconnect (for outgoing connection). +- Renamed `connectionTxBytes` and `connectionRxBytes` on the peer stats struct + to `txBytes` and `rxBytes` to better express that they are no longer tied to + a single connection to the peer. ## [0.4.3] - 2024-03-15 diff --git a/docs/api.yaml b/docs/api.yaml index 104b609..64f8fd9 100644 --- a/docs/api.yaml +++ b/docs/api.yaml @@ -371,14 +371,14 @@ components: - 'connecting' - 'dead' example: alive - connectionTxBytes: - description: The amount of bytes transmitted on the current connection + txBytes: + description: The amount of bytes transmitted to this peer type: integer format: int64 minimum: 0 example: 464531564 - connectionRxBytes: - description: The amount of bytes received on the current connection + rxBytes: + description: The amount of bytes received from this peer type: integer format: int64 minimum: 0 diff --git a/src/peer.rs b/src/peer.rs index 88916db..6dcd51b 100644 --- a/src/peer.rs +++ b/src/peer.rs @@ -61,8 +61,7 @@ impl Peer { bytes_read: Arc, ) -> Result { // Wrap connection so we can get access to the counters. - let connection = - connection::Tracked::new(bytes_read.clone(), bytes_written.clone(), connection); + let connection = connection::Tracked::new(bytes_read, bytes_written, connection); // Data channel for peer let (to_peer_data, mut from_routing_data) = mpsc::unbounded_channel::(); @@ -78,8 +77,6 @@ impl Peer { to_peer_control, connection_identifier: connection.identifier()?, static_link_cost: connection.static_link_cost()?, - read: bytes_read, - written: bytes_written, death_notifier, alive: AtomicBool::new(true), }), @@ -259,16 +256,6 @@ impl Peer { inner: Arc::downgrade(&self.inner), } } - - /// Get the amount of bytes received from this `Peer`. - pub fn read(&self) -> u64 { - self.inner.read.load(Ordering::Relaxed) - } - - /// Get the amount of bytes written to this `Peer`. - pub fn written(&self) -> u64 { - self.inner.written.load(Ordering::Relaxed) - } } impl PeerRef { @@ -315,10 +302,6 @@ struct PeerInner { /// Static cost of using this link, to be added to the announced metric for routes through this /// Peer. static_link_cost: u16, - /// Amount of bytes received from the remote. - read: Arc, - /// Amount of bytes sent to the remote. - written: Arc, /// Channel to notify the connection of its decease. death_notifier: Arc, /// Keep track if the connection is alive. diff --git a/src/peer_manager.rs b/src/peer_manager.rs index de07bea..b08fd9d 100644 --- a/src/peer_manager.rs +++ b/src/peer_manager.rs @@ -13,7 +13,7 @@ use std::collections::hash_map::Entry; use std::collections::{HashMap, HashSet}; use std::fmt; use std::net::{IpAddr, SocketAddr}; -use std::sync::atomic::AtomicU64; +use std::sync::atomic::{AtomicU64, Ordering}; use std::sync::{Arc, Mutex}; use std::time::Duration; use tokio::net::TcpStream; @@ -101,10 +101,24 @@ pub struct PeerStats { pub pt: PeerType, /// State of the connection to this [`Peer`] pub connection_state: ConnectionState, - /// Amount of bytes transmitted on the current connection. - pub connection_tx_bytes: u64, - /// Amount of bytes received on the cureent conntection. - pub connection_rx_bytes: u64, + /// Amount of bytes transmitted to this [`Peer`]. + pub tx_bytes: u64, + /// Amount of bytes received from this [`Peer`]. + pub rx_bytes: u64, +} + +impl PeerInfo { + /// Return the amount of bytes read from this peer. + #[inline] + fn read(&self) -> u64 { + self.con_traffic.rx_bytes.load(Ordering::Relaxed) + } + + /// Return the amount of bytes written to this peer. + #[inline] + fn written(&self) -> u64 { + self.con_traffic.rx_bytes.load(Ordering::Relaxed) + } } /// Marker error to indicate a [`peer`](Endpoint) is already known. @@ -238,18 +252,19 @@ impl PeerManager { let peer_map = self.inner.peers.lock().unwrap(); let mut pi = Vec::with_capacity(peer_map.len()); for (endpoint, peer_info) in peer_map.iter() { - let (connection_state, connection_tx_bytes, connection_rx_bytes) = - match peer_info.pr.upgrade() { - None if !peer_info.connecting => (ConnectionState::Dead, 0, 0), - None => (ConnectionState::Connecting, 0, 0), - Some(peer) => (ConnectionState::Alive, peer.written(), peer.read()), - }; + let connection_state = if peer_info.connecting { + ConnectionState::Connecting + } else if peer_info.pr.alive() { + ConnectionState::Alive + } else { + ConnectionState::Dead + }; pi.push(PeerStats { endpoint: *endpoint, pt: peer_info.pt.clone(), connection_state, - connection_tx_bytes, - connection_rx_bytes, + tx_bytes: peer_info.written(), + rx_bytes: peer_info.read(), }); } pi