From ff3bf369578dfd3489238cc5dde2023cc39b50fd Mon Sep 17 00:00:00 2001 From: zonyitoo Date: Sun, 23 Jun 2024 22:40:41 +0800 Subject: [PATCH] feat(local): PingBalancer print warning if server's fail_rate > 0.5 --- .../src/local/loadbalancing/ping_balancer.rs | 50 +++++++++------ .../src/local/loadbalancing/server_data.rs | 7 +- .../src/local/loadbalancing/server_stat.rs | 64 ++++++++++++------- 3 files changed, 78 insertions(+), 43 deletions(-) diff --git a/crates/shadowsocks-service/src/local/loadbalancing/ping_balancer.rs b/crates/shadowsocks-service/src/local/loadbalancing/ping_balancer.rs index 33f3fbe6..dc75b315 100644 --- a/crates/shadowsocks-service/src/local/loadbalancing/ping_balancer.rs +++ b/crates/shadowsocks-service/src/local/loadbalancing/ping_balancer.rs @@ -818,24 +818,36 @@ struct PingChecker { impl PingChecker { /// Checks server's score and update into `ServerScore` async fn check_update_score(self) { - let score = match self.check_delay().await { - Ok(d) => match self.server_type { - ServerType::Tcp => self.server.tcp_score().push_score(Score::Latency(d)).await, - ServerType::Udp => self.server.udp_score().push_score(Score::Latency(d)).await, - }, - // Penalty - Err(..) => match self.server_type { - ServerType::Tcp => self.server.tcp_score().push_score(Score::Errored).await, - ServerType::Udp => self.server.udp_score().push_score(Score::Errored).await, - }, + let server_score = match self.server_type { + ServerType::Tcp => self.server.tcp_score(), + ServerType::Udp => self.server.udp_score(), }; - trace!( - "updated remote {} server {} (score: {})", - self.server_type, - self.server.server_config().addr(), - score - ); + let score = match self.check_delay().await { + Ok(d) => server_score.push_score(Score::Latency(d)).await, + // Penalty + Err(..) => server_score.push_score(Score::Errored).await, + }; + + let stat_data = server_score.stat_data().await; + + if stat_data.fail_rate > 0.5 { + warn!( + "balancer: checked & updated remote {} server {} (score: {}), {:?}", + self.server_type, + ServerConfigFormatter::new(self.server.server_config()), + score, + stat_data, + ); + } else { + debug!( + "balancer: checked & updated remote {} server {} (score: {}), {:?}", + self.server_type, + ServerConfigFormatter::new(self.server.server_config()), + score, + stat_data, + ); + } } /// Detect TCP connectivity with Chromium [Network Portal Detection](https://www.chromium.org/chromium-os/chromiumos-design-docs/network-portal-detection) @@ -988,7 +1000,7 @@ impl PingChecker { trace!( "checked remote {} server {} latency with {} ms", self.server_type, - self.server.server_config().addr(), + ServerConfigFormatter::new(self.server.server_config()), elapsed ); Ok(elapsed) @@ -997,7 +1009,7 @@ impl PingChecker { debug!( "failed to check {} server {}, error: {}", self.server_type, - self.server.server_config().addr(), + ServerConfigFormatter::new(self.server.server_config()), err ); @@ -1011,7 +1023,7 @@ impl PingChecker { trace!( "checked remote {} server {} latency timeout, elapsed {} ms", self.server_type, - self.server.server_config().addr(), + ServerConfigFormatter::new(self.server.server_config()), elapsed ); diff --git a/crates/shadowsocks-service/src/local/loadbalancing/server_data.rs b/crates/shadowsocks-service/src/local/loadbalancing/server_data.rs index 6fe4a2cf..39e511ed 100644 --- a/crates/shadowsocks-service/src/local/loadbalancing/server_data.rs +++ b/crates/shadowsocks-service/src/local/loadbalancing/server_data.rs @@ -14,7 +14,7 @@ use tokio::sync::Mutex; use crate::{config::ServerInstanceConfig, local::context::ServiceContext}; -use super::server_stat::{Score, ServerStat}; +use super::server_stat::{Score, ServerStat, ServerStatData}; /// Server's statistic score pub struct ServerScore { @@ -53,6 +53,11 @@ impl ServerScore { pub async fn report_failure(&self) -> u32 { self.push_score(Score::Errored).await } + + /// Get statistic data + pub async fn stat_data(&self) -> ServerStatData { + self.stat_data.lock().await.data().clone() + } } impl Debug for ServerScore { diff --git a/crates/shadowsocks-service/src/local/loadbalancing/server_stat.rs b/crates/shadowsocks-service/src/local/loadbalancing/server_stat.rs index a4810a93..9091ac20 100644 --- a/crates/shadowsocks-service/src/local/loadbalancing/server_stat.rs +++ b/crates/shadowsocks-service/src/local/loadbalancing/server_stat.rs @@ -19,30 +19,37 @@ pub enum Score { Errored, } -/// Statistic of a remote server -#[derive(Debug)] -pub struct ServerStat { +/// Server statistic data +#[derive(Debug, Clone, Copy)] +pub struct ServerStatData { /// Median of latency time (in millisec) /// /// Use median instead of average time, /// because probing result may have some really bad cases - rtt: u32, + pub latency_median: u32, + /// Total_Fail / Total_Probe + pub fail_rate: f64, + /// Score's standard deviation + pub latency_stdev: f64, + /// Score's average + pub latency_mean: f64, +} + +/// Statistic of a remote server +#[derive(Debug)] +pub struct ServerStat { /// MAX server's RTT, normally the check timeout milliseconds max_server_rtt: u32, - /// Total_Fail / Total_Probe - fail_rate: f64, /// Recently probe data latency_queue: VecDeque<(Score, Instant)>, - /// Score's standard deviation - latency_stdev: f64, /// Score's standard deviation MAX max_latency_stdev: f64, - /// Score's average - latency_mean: f64, /// User's customized weight user_weight: f32, /// Checking window size check_window: Duration, + /// Statistic Data + data: ServerStatData, } fn max_latency_stdev(max_server_rtt: u32) -> f64 { @@ -58,25 +65,28 @@ impl ServerStat { pub fn new(user_weight: f32, max_server_rtt: u32, check_window: Duration) -> ServerStat { assert!((0.0..=1.0).contains(&user_weight)); + let max_latency_stdev = max_latency_stdev(max_server_rtt); ServerStat { - rtt: max_server_rtt, max_server_rtt, - fail_rate: 1.0, latency_queue: VecDeque::new(), - latency_stdev: 0.0, - max_latency_stdev: max_latency_stdev(max_server_rtt), - latency_mean: 0.0, + max_latency_stdev, user_weight, check_window, + data: ServerStatData { + latency_median: max_server_rtt, + fail_rate: 1.0, + latency_stdev: max_latency_stdev, + latency_mean: max_server_rtt as f64, + }, } } fn score(&self) -> u32 { // Normalize rtt - let nrtt = self.rtt as f64 / self.max_server_rtt as f64; + let nrtt = self.data.latency_median as f64 / self.max_server_rtt as f64; // Normalize stdev - let nstdev = self.latency_stdev / self.max_latency_stdev; + let nstdev = self.data.latency_stdev / self.max_latency_stdev; const SCORE_RTT_WEIGHT: f64 = 1.0; const SCORE_FAIL_WEIGHT: f64 = 3.0; @@ -92,7 +102,7 @@ impl ServerStat { // 2. The lower errored count, the better // 3. The lower latency's stdev, the better // 4. The higher user's weight, the better - let score = (nrtt * SCORE_RTT_WEIGHT + self.fail_rate * SCORE_FAIL_WEIGHT + nstdev * SCORE_STDEV_WEIGHT) + let score = (nrtt * SCORE_RTT_WEIGHT + self.data.fail_rate * SCORE_FAIL_WEIGHT + nstdev * SCORE_STDEV_WEIGHT) / (SCORE_RTT_WEIGHT + SCORE_FAIL_WEIGHT + SCORE_STDEV_WEIGHT) / user_weight as f64; @@ -132,15 +142,17 @@ impl ServerStat { } // Error rate - self.fail_rate = cerr as f64 / self.latency_queue.len() as f64; + self.data.fail_rate = cerr as f64 / self.latency_queue.len() as f64; + self.data.latency_stdev = self.max_latency_stdev; + self.data.latency_mean = self.max_server_rtt as f64; if !vlat.is_empty() { vlat.sort_unstable(); // Find median of latency let mid = vlat.len() / 2; - self.rtt = if vlat.len() % 2 == 0 { + self.data.latency_median = if vlat.len() % 2 == 0 { (vlat[mid] + vlat[mid - 1]) / 2 } else { vlat[mid] @@ -154,17 +166,23 @@ impl ServerStat { for s in &vlat { total_lat += *s; } - self.latency_mean = total_lat as f64 / n; + self.data.latency_mean = total_lat as f64 / n; let mut acc_diff = 0.0; for s in &vlat { - let diff = *s as f64 - self.latency_mean; + let diff = *s as f64 - self.data.latency_mean; acc_diff += diff * diff; } // Corrected Sample Standard Deviation - self.latency_stdev = ((1.0 / (n - 1.0)) * acc_diff).sqrt(); + self.data.latency_stdev = ((1.0 / (n - 1.0)) * acc_diff).sqrt(); + } else { + self.data.latency_mean = vlat[0] as f64; } } self.score() } + + pub fn data(&self) -> &ServerStatData { + &self.data + } }