summaryrefslogtreecommitdiffstats
path: root/src/transport/traffic.rs
diff options
context:
space:
mode:
Diffstat (limited to 'src/transport/traffic.rs')
-rw-r--r--src/transport/traffic.rs143
1 files changed, 100 insertions, 43 deletions
diff --git a/src/transport/traffic.rs b/src/transport/traffic.rs
index 9dcd91a..64595cb 100644
--- a/src/transport/traffic.rs
+++ b/src/transport/traffic.rs
@@ -1,12 +1,12 @@
//! Messages and bytes exchanged with each peer, by kind: counted where every message
//! passes anyway, at the transport. No messages of our own.
//!
-//! Bytes are payload bytes, what the Mycelium message carries, without its framing,
-//! the topic, or the base64 of its HTTP API. Counters only grow (from the start of the
-//! process); rates are for whoever reads them.
+//! Bytes are payload bytes, what a message carries, without the transport's framing
+//! (for Mycelium: the topic, or the base64 of its HTTP API). Counters only grow (from
+//! the start of the process); rates are for whoever reads them.
//!
-//! A message dropped by the debug firewall still counts as sent (the firewall plays a
-//! lossy network) but not as received.
+//! A message dropped by the debug firewall (in the paths layer, below) still counts as
+//! sent (the firewall plays a lossy network) but not as received.
use std::collections::HashMap;
use std::sync::{Arc, Mutex, MutexGuard};
@@ -14,14 +14,14 @@ use std::sync::{Arc, Mutex, MutexGuard};
use anyhow::Result;
use async_trait::async_trait;
use serde::Serialize;
-use utoipa::ToSchema;
+use crate::locks::MutexExt;
use crate::transport::{EncodedMessage, MessageHandler, MessageKind, ParsedMessage, Transport};
-#[derive(Debug, Default, Clone, Copy, Serialize, ToSchema)]
-pub struct Counter {
- pub messages: u64,
- pub bytes: u64,
+#[derive(Debug, Default, Clone, Copy, Serialize)]
+pub(crate) struct Counter {
+ pub(crate) messages: u64,
+ pub(crate) bytes: u64,
}
impl Counter {
@@ -31,15 +31,20 @@ impl Counter {
}
}
-#[derive(Debug, Default, Clone, Copy, Serialize, ToSchema)]
+#[derive(Debug, Default, Clone, Copy, Serialize)]
#[serde(rename_all = "camelCase")]
-pub struct ByKind {
- pub hello: Counter,
- pub hello_reply: Counter,
- pub sync: Counter,
- pub ping: Counter,
- pub pong: Counter,
- pub blob: Counter,
+pub(crate) struct ByKind {
+ pub(crate) hello: Counter,
+ pub(crate) hello_reply: Counter,
+ pub(crate) sync: Counter,
+ pub(crate) status: Counter,
+ pub(crate) status_reply: Counter,
+ pub(crate) blob: Counter,
+ /// WHO and ME (`transport::identity`), counted by the identity layer under the
+ /// address they went to or came from.
+ pub(crate) identity: Counter,
+ /// PING and PONG (`transport::paths`), counted by the paths layer under the address.
+ pub(crate) paths: Counter,
}
impl ByKind {
@@ -48,21 +53,23 @@ impl ByKind {
MessageKind::Hello => &mut self.hello,
MessageKind::HelloReply => &mut self.hello_reply,
MessageKind::Sync => &mut self.sync,
- MessageKind::Ping => &mut self.ping,
- MessageKind::Pong => &mut self.pong,
+ MessageKind::Status => &mut self.status,
+ MessageKind::StatusReply => &mut self.status_reply,
MessageKind::Blob => &mut self.blob,
+ MessageKind::Who | MessageKind::Me => &mut self.identity,
+ MessageKind::Ping | MessageKind::Pong => &mut self.paths,
}
}
}
-#[derive(Debug, Default, Clone, Copy, Serialize, ToSchema)]
+#[derive(Debug, Default, Clone, Copy, Serialize)]
#[serde(rename_all = "camelCase")]
-pub struct PeerTraffic {
- /// Handed to Mycelium (it accepted them).
- pub sent: ByKind,
- pub received: ByKind,
- /// Sends Mycelium refused (e.g. no route to the peer); not counted as sent.
- pub refused: ByKind,
+pub(crate) struct PeerTraffic {
+ /// Handed to the transport (it accepted them).
+ pub(crate) sent: ByKind,
+ pub(crate) received: ByKind,
+ /// Sends the transport refused (e.g. no route to the peer); not counted as sent.
+ pub(crate) refused: ByKind,
}
#[derive(Debug, Default)]
@@ -73,15 +80,13 @@ struct Table {
}
#[derive(Debug, Default)]
-pub struct Traffic {
+pub(crate) struct Traffic {
table: Mutex<Table>,
}
impl Traffic {
fn lock(&self) -> MutexGuard<'_, Table> {
- self.table
- .lock()
- .unwrap_or_else(|poisoned| poisoned.into_inner())
+ self.table.lock_or_recover()
}
fn count(&self, peer: &str, bytes: usize, pick: impl Fn(&mut PeerTraffic) -> &mut Counter) {
@@ -94,41 +99,41 @@ impl Traffic {
}
}
- fn sent(&self, peer: &str, kind: MessageKind, bytes: usize) {
+ pub(crate) fn sent(&self, peer: &str, kind: MessageKind, bytes: usize) {
self.count(peer, bytes, |t| t.sent.get_mut(kind));
}
- fn refused(&self, peer: &str, kind: MessageKind, bytes: usize) {
+ pub(crate) fn refused(&self, peer: &str, kind: MessageKind, bytes: usize) {
self.count(peer, bytes, |t| t.refused.get_mut(kind));
}
- fn received(&self, peer: &str, kind: MessageKind, bytes: usize) {
+ pub(crate) fn received(&self, peer: &str, kind: MessageKind, bytes: usize) {
self.count(peer, bytes, |t| t.received.get_mut(kind));
}
- pub fn peer(&self, peer: &str) -> Option<PeerTraffic> {
+ pub(crate) fn peer(&self, peer: &str) -> Option<PeerTraffic> {
self.lock().peers.get(peer).copied()
}
- pub fn total(&self) -> PeerTraffic {
+ pub(crate) fn total(&self) -> PeerTraffic {
self.lock().total
}
/// Drops the counters of peers not kept (anyone can send us messages; the totals
/// keep their share).
- pub fn retain(&self, mut keep: impl FnMut(&str) -> bool) {
+ pub(crate) fn retain(&self, mut keep: impl FnMut(&str) -> bool) {
self.lock().peers.retain(|peer, _| keep(peer));
}
}
/// Counts what passes through `inner`.
-pub struct CountingTransport {
+pub(crate) struct CountingTransport {
inner: Arc<dyn Transport>,
traffic: Arc<Traffic>,
}
impl CountingTransport {
- pub fn new(inner: Arc<dyn Transport>, traffic: Arc<Traffic>) -> Self {
+ pub(crate) fn new(inner: Arc<dyn Transport>, traffic: Arc<Traffic>) -> Self {
Self { inner, traffic }
}
}
@@ -139,6 +144,58 @@ impl Transport for CountingTransport {
self.inner.local_id()
}
+ fn is_peer_id(&self, id: &str) -> bool {
+ self.inner.is_peer_id(id)
+ }
+
+ fn delivery_limit(&self) -> std::time::Duration {
+ self.inner.delivery_limit()
+ }
+
+ fn local_addresses(&self) -> Vec<String> {
+ self.inner.local_addresses()
+ }
+
+ fn addresses_of(&self, peer: &str) -> Vec<String> {
+ self.inner.addresses_of(peer)
+ }
+
+ fn learn_addresses(&self, peer: &str, addresses: &[String]) {
+ self.inner.learn_addresses(peer, addresses)
+ }
+
+ fn can_reach(&self, peer: &str) -> bool {
+ self.inner.can_reach(peer)
+ }
+
+ fn discovered_seeds(&self) -> Vec<String> {
+ self.inner.discovered_seeds()
+ }
+
+ fn transports(&self) -> Vec<String> {
+ self.inner.transports()
+ }
+
+ fn destination(&self, peer: &str) -> Option<crate::transport::Destination> {
+ self.inner.destination(peer)
+ }
+
+ fn connection(&self, target: &str) -> Option<crate::transport::Connection> {
+ self.inner.connection(target)
+ }
+
+ fn stats(&self) -> crate::transport::stats::TransportStats {
+ self.inner.stats()
+ }
+
+ fn answers_pings(&self) -> bool {
+ self.inner.answers_pings()
+ }
+
+ fn rtt(&self, peer: &str) -> Option<std::time::Duration> {
+ self.inner.rtt(peer)
+ }
+
async fn resolve_seed(&self, seed: &str) -> Result<String> {
self.inner.resolve_seed(seed).await
}
@@ -181,20 +238,20 @@ mod tests {
#[test]
fn totals_survive_retain() {
let traffic = Traffic::default();
- traffic.sent("a", MessageKind::Ping, 100);
+ traffic.sent("a", MessageKind::Status, 100);
traffic.sent("b", MessageKind::Sync, 1000);
- traffic.received("b", MessageKind::Pong, 90);
+ traffic.received("b", MessageKind::StatusReply, 90);
traffic.refused("a", MessageKind::Sync, 500);
let b = traffic.peer("b").unwrap();
assert_eq!(b.sent.sync.messages, 1);
- assert_eq!(b.received.pong.bytes, 90);
+ assert_eq!(b.received.status_reply.bytes, 90);
traffic.retain(|peer| peer == "b");
assert!(traffic.peer("a").is_none());
let total = traffic.total();
- assert_eq!(total.sent.ping.bytes, 100);
+ assert_eq!(total.sent.status.bytes, 100);
assert_eq!(total.sent.sync.bytes, 1000);
assert_eq!(total.refused.sync.bytes, 500);
}
generated by cgit v1.3.1 (git 2.54.0) at 2026-10-28 07:24:47 +0000