Load peer counters from peer manager

Don't load the counters from the peer itself, since they are now also
tracked on the peer manager. Additionally, this means we don't have to
save them on the peer itself anymore.

Signed-off-by: Lee Smet <lee.smet@hotmail.com>

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
 

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

src/peer.rs

@@ -61,8 +61,7 @@ impl Peer {
         bytes_read: Arc<AtomicU64>,
     ) -> Result<Self, io::Error> {
         // 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::<DataPacket>();
@@ -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<AtomicU64>,
-    /// Amount of bytes sent to the remote.
-    written: Arc<AtomicU64>,
     /// Channel to notify the connection of its decease.
     death_notifier: Arc<Notify>,
     /// Keep track if the connection is alive.

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