记录具体路径的丢包率

This commit is contained in:
lbl
2026-02-28 18:56:11 +08:00
parent 2aa423062e
commit 12f7eed8a5
6 changed files with 129 additions and 26 deletions
+14 -2
View File
@@ -67,14 +67,26 @@ impl VntApi {
pub fn peer_nat_info(&self, ip: &Ipv4Addr) -> Option<NatInfo> { pub fn peer_nat_info(&self, ip: &Ipv4Addr) -> Option<NatInfo> {
self.app_state.get_peer_info(ip).and_then(|v| v.nat_info) self.app_state.get_peer_info(ip).and_then(|v| v.nat_info)
} }
/// 获取指定 IP 的聚合丢包信息(所有路由合并)
pub fn packet_loss_info(&self, ip: &Ipv4Addr) -> Option<PacketLossInfo> { pub fn packet_loss_info(&self, ip: &Ipv4Addr) -> Option<PacketLossInfo> {
self.app_state.packet_loss_stats.get_loss_info(ip) self.app_state
.packet_loss_stats
.get_aggregated_loss_info(ip)
}
/// 获取指定 IP 的所有路由的丢包信息
pub fn packet_loss_info_by_routes(&self, ip: &Ipv4Addr) -> Vec<PacketLossInfo> {
self.app_state.packet_loss_stats.get_loss_info_by_ip(ip)
} }
pub fn all_packet_loss_info(&self) -> Vec<PacketLossInfo> { pub fn all_packet_loss_info(&self) -> Vec<PacketLossInfo> {
self.app_state.packet_loss_stats.get_all_loss_info() self.app_state.packet_loss_stats.get_all_loss_info()
} }
pub fn reset_packet_loss(&self, ip: &Ipv4Addr) { pub fn reset_packet_loss(&self, ip: &Ipv4Addr) {
self.app_state.packet_loss_stats.reset(ip) // 重置该 IP 的所有路由统计
for info in self.app_state.packet_loss_stats.get_loss_info_by_ip(ip) {
if let Some(route_key) = info.route_key {
self.app_state.packet_loss_stats.reset(ip, &route_key);
}
}
} }
pub fn reset_all_packet_loss(&self) { pub fn reset_all_packet_loss(&self) {
self.app_state.packet_loss_stats.reset_all() self.app_state.packet_loss_stats.reset_all()
+86 -16
View File
@@ -8,6 +8,7 @@ use crate::tunnel_core::server::transport::config::ProtocolAddress;
use ipnet::Ipv4Net; use ipnet::Ipv4Net;
use parking_lot::{Mutex, RwLock}; use parking_lot::{Mutex, RwLock};
use rust_p2p_core::nat::NatInfo; use rust_p2p_core::nat::NatInfo;
use rust_p2p_core::route::RouteKey;
use serde::{Deserialize, Serialize}; use serde::{Deserialize, Serialize};
use std::collections::HashMap; use std::collections::HashMap;
use std::net::Ipv4Addr; use std::net::Ipv4Addr;
@@ -108,45 +109,55 @@ impl TrafficStats {
#[derive(Clone, Default)] #[derive(Clone, Default)]
pub struct PacketLossStats { pub struct PacketLossStats {
inner: Arc<RwLock<HashMap<Ipv4Addr, Arc<Mutex<PingStats>>>>>, inner: Arc<RwLock<HashMap<(Ipv4Addr, RouteKey), Arc<Mutex<PingStats>>>>>,
} }
#[derive(Clone, Debug, Serialize, Deserialize)] #[derive(Clone, Debug, Serialize, Deserialize)]
pub struct PacketLossInfo { pub struct PacketLossInfo {
pub ip: Ipv4Addr, pub ip: Ipv4Addr,
#[serde(skip)]
pub route_key: Option<RouteKey>,
pub sent: u64, pub sent: u64,
pub received: u64, pub received: u64,
pub loss_rate: f64, pub loss_rate: f64,
} }
impl PacketLossStats { impl PacketLossStats {
fn get_or_create(&self, ip: Ipv4Addr) -> Arc<Mutex<PingStats>> { fn get_or_create(&self, ip: Ipv4Addr, route_key: RouteKey) -> Arc<Mutex<PingStats>> {
{ {
let read = self.inner.read(); let read = self.inner.read();
if let Some(stats) = read.get(&ip) { if let Some(stats) = read.get(&(ip, route_key)) {
return stats.clone(); return stats.clone();
} }
} }
let mut write = self.inner.write(); let mut write = self.inner.write();
write write
.entry(ip) .entry((ip, route_key))
.or_insert_with(|| Arc::new(Mutex::new(PingStats::default()))) .or_insert_with(|| Arc::new(Mutex::new(PingStats::default())))
.clone() .clone()
} }
pub fn record_sent(&self, ip: Ipv4Addr) { pub fn record_sent(&self, ip: Ipv4Addr, route_key: RouteKey) {
let stats = self.get_or_create(ip); let stats = self.get_or_create(ip, route_key);
stats.lock().sent += 1; stats.lock().sent += 1;
} }
pub fn record_received(&self, ip: Ipv4Addr) { pub fn record_received(&self, ip: Ipv4Addr, route_key: RouteKey) -> f64 {
let stats = self.get_or_create(ip); let stats = self.get_or_create(ip, route_key);
stats.lock().received += 1; let mut guard = stats.lock();
guard.received += 1;
// 计算并返回丢包率
if guard.sent > 0 {
1.0 - (guard.received as f64 / guard.sent as f64)
} else {
0.0
}
} }
pub fn get_loss_info(&self, ip: &Ipv4Addr) -> Option<PacketLossInfo> { pub fn get_loss_info(&self, ip: &Ipv4Addr, route_key: &RouteKey) -> Option<PacketLossInfo> {
let read = self.inner.read(); let read = self.inner.read();
read.get(ip).map(|stats| { read.get(&(*ip, *route_key)).map(|stats| {
let guard = stats.lock(); let guard = stats.lock();
let loss_rate = if guard.sent > 0 { let loss_rate = if guard.sent > 0 {
1.0 - (guard.received as f64 / guard.sent as f64) 1.0 - (guard.received as f64 / guard.sent as f64)
@@ -155,6 +166,7 @@ impl PacketLossStats {
}; };
PacketLossInfo { PacketLossInfo {
ip: *ip, ip: *ip,
route_key: Some(*route_key),
sent: guard.sent, sent: guard.sent,
received: guard.received, received: guard.received,
loss_rate, loss_rate,
@@ -162,10 +174,12 @@ impl PacketLossStats {
}) })
} }
pub fn get_all_loss_info(&self) -> Vec<PacketLossInfo> { /// 获取指定 IP 的所有路由的丢包信息
pub fn get_loss_info_by_ip(&self, ip: &Ipv4Addr) -> Vec<PacketLossInfo> {
let read = self.inner.read(); let read = self.inner.read();
read.iter() read.iter()
.map(|(ip, stats)| { .filter(|((addr, _), _)| addr == ip)
.map(|((addr, route_key), stats)| {
let guard = stats.lock(); let guard = stats.lock();
let loss_rate = if guard.sent > 0 { let loss_rate = if guard.sent > 0 {
1.0 - (guard.received as f64 / guard.sent as f64) 1.0 - (guard.received as f64 / guard.sent as f64)
@@ -173,7 +187,8 @@ impl PacketLossStats {
0.0 0.0
}; };
PacketLossInfo { PacketLossInfo {
ip: *ip, ip: *addr,
route_key: Some(*route_key),
sent: guard.sent, sent: guard.sent,
received: guard.received, received: guard.received,
loss_rate, loss_rate,
@@ -182,9 +197,64 @@ impl PacketLossStats {
.collect() .collect()
} }
pub fn reset(&self, ip: &Ipv4Addr) { /// 获取指定 IP 的聚合丢包信息(所有路由合并)
pub fn get_aggregated_loss_info(&self, ip: &Ipv4Addr) -> Option<PacketLossInfo> {
let read = self.inner.read(); let read = self.inner.read();
if let Some(stats) = read.get(ip) { let mut total_sent = 0u64;
let mut total_received = 0u64;
let mut found = false;
for ((addr, _), stats) in read.iter() {
if addr == ip {
found = true;
let guard = stats.lock();
total_sent += guard.sent;
total_received += guard.received;
}
}
if found {
let loss_rate = if total_sent > 0 {
1.0 - (total_received as f64 / total_sent as f64)
} else {
0.0
};
Some(PacketLossInfo {
ip: *ip,
route_key: None,
sent: total_sent,
received: total_received,
loss_rate,
})
} else {
None
}
}
pub fn get_all_loss_info(&self) -> Vec<PacketLossInfo> {
let read = self.inner.read();
read.iter()
.map(|((ip, route_key), stats)| {
let guard = stats.lock();
let loss_rate = if guard.sent > 0 {
1.0 - (guard.received as f64 / guard.sent as f64)
} else {
0.0
};
PacketLossInfo {
ip: *ip,
route_key: Some(*route_key),
sent: guard.sent,
received: guard.received,
loss_rate,
}
})
.collect()
}
pub fn reset(&self, ip: &Ipv4Addr, route_key: &RouteKey) {
let read = self.inner.read();
if let Some(stats) = read.get(&(*ip, *route_key)) {
*stats.lock() = PingStats::default(); *stats.lock() = PingStats::default();
} }
} }
+8 -2
View File
@@ -201,11 +201,17 @@ impl P2pInboundHandler {
let time = i64::from_be_bytes(net_packet.payload()[..8].try_into()?); let time = i64::from_be_bytes(net_packet.payload()[..8].try_into()?);
let now = crate::utils::time::now_ts_ms(); let now = crate::utils::time::now_ts_ms();
if now >= time { if now >= time {
// 记录接收并获取丢包率
let loss_rate_f64 = self
.packet_loss_stats
.record_received(ctx.src_ip, route_key);
// 转换为万分率
let loss_rate = (loss_rate_f64 * 10000.0).round() as u16;
self.route_table.add_route( self.route_table.add_route(
ctx.src_ip, ctx.src_ip,
Route::from(route_key, metric, (now - time) as _), Route::from_with_loss(route_key, metric, (now - time) as _, loss_rate),
); );
self.packet_loss_stats.record_received(ctx.src_ip);
} }
} }
} }
@@ -10,6 +10,8 @@ pub struct Route {
route_key: RouteKey, route_key: RouteKey,
metric: u8, metric: u8,
rtt: u32, rtt: u32,
/// 丢包率,万分率(0-1000010000 表示 100% 丢包)
loss_rate: u16,
} }
impl Route { impl Route {
pub fn from(route_key: RouteKey, metric: u8, rtt: u32) -> Self { pub fn from(route_key: RouteKey, metric: u8, rtt: u32) -> Self {
@@ -17,6 +19,15 @@ impl Route {
route_key, route_key,
metric, metric,
rtt, rtt,
loss_rate: 0,
}
}
pub fn from_with_loss(route_key: RouteKey, metric: u8, rtt: u32, loss_rate: u16) -> Self {
Self {
route_key,
metric,
rtt,
loss_rate,
} }
} }
pub fn from_default_rt(route_key: RouteKey, metric: u8) -> Self { pub fn from_default_rt(route_key: RouteKey, metric: u8) -> Self {
@@ -24,6 +35,7 @@ impl Route {
route_key, route_key,
metric, metric,
rtt: DEFAULT_RTT, rtt: DEFAULT_RTT,
loss_rate: 0,
} }
} }
pub fn route_key(&self) -> RouteKey { pub fn route_key(&self) -> RouteKey {
@@ -39,6 +51,9 @@ impl Route {
pub fn metric(&self) -> u8 { pub fn metric(&self) -> u8 {
self.metric self.metric
} }
pub fn loss_rate(&self) -> u16 {
self.loss_rate
}
} }
#[derive(Clone)] #[derive(Clone)]
@@ -133,12 +133,9 @@ pub async fn ping_all(
ping.set_dest_id(id.into()); ping.set_dest_id(id.into());
ping.set_payload(&crate::utils::time::now_ts_ms().to_be_bytes()) ping.set_payload(&crate::utils::time::now_ts_ms().to_be_bytes())
.unwrap(); .unwrap();
if socket_manager let route_key = route.route_key();
.send_to(ping, &route.route_key()) if socket_manager.send_to(ping, &route_key).await.is_ok() {
.await packet_loss_stats.record_sent(id, route_key);
.is_ok()
{
packet_loss_stats.record_sent(id);
} }
} }
tokio::time::sleep(Duration::from_millis(10)).await; tokio::time::sleep(Duration::from_millis(10)).await;
+3
View File
@@ -295,6 +295,7 @@ struct HttpRouteDetail {
protocol: String, protocol: String,
metric: u8, metric: u8,
rtt: u32, rtt: u32,
loss_rate: u16,
} }
#[derive(Serialize)] #[derive(Serialize)]
@@ -1032,6 +1033,7 @@ async fn get_peers(State(state): State<HttpAppState>) -> Json<ApiResponse<Vec<Ht
protocol: route.route_key().protocol().to_string(), protocol: route.route_key().protocol().to_string(),
metric: route.metric(), metric: route.metric(),
rtt: route.rtt(), rtt: route.rtt(),
loss_rate: route.loss_rate(),
}) })
}; };
@@ -1114,6 +1116,7 @@ async fn get_routes(State(state): State<HttpAppState>) -> Json<ApiResponse<Vec<H
protocol: v.route_key().protocol().to_string(), protocol: v.route_key().protocol().to_string(),
metric: v.metric(), metric: v.metric(),
rtt: v.rtt(), rtt: v.rtt(),
loss_rate: v.loss_rate(),
}) })
.collect(), .collect(),
}) })