From c1b6bb5c506d31383ffecaa6aa31cec48d4e2abe Mon Sep 17 00:00:00 2001 From: zonyitoo Date: Thu, 31 Dec 2020 22:28:26 +0800 Subject: [PATCH] compatible with MIPS and PowerPC for non-supported AtomicU64 --- .../src/local/loadbalancing/ping_balancer.rs | 8 ++-- .../src/local/loadbalancing/server_data.rs | 12 +++--- .../src/local/loadbalancing/server_stat.rs | 16 +++---- crates/shadowsocks-service/src/net/flow.rs | 43 ++++++++++++++++--- 4 files changed, 56 insertions(+), 23 deletions(-) diff --git a/crates/shadowsocks-service/src/local/loadbalancing/ping_balancer.rs b/crates/shadowsocks-service/src/local/loadbalancing/ping_balancer.rs index 835caac0..6f0d5547 100644 --- a/crates/shadowsocks-service/src/local/loadbalancing/ping_balancer.rs +++ b/crates/shadowsocks-service/src/local/loadbalancing/ping_balancer.rs @@ -176,7 +176,7 @@ impl PingBalancerContext { let old_best_idx = self.best_tcp_idx.load(Ordering::Acquire); let mut best_idx = 0; - let mut best_score = u64::MAX; + let mut best_score = u32::MAX; for (idx, server) in self.servers.iter().enumerate() { let score = server.tcp_score().score(); if score < best_score { @@ -205,7 +205,7 @@ impl PingBalancerContext { let old_best_idx = self.best_udp_idx.load(Ordering::Acquire); let mut best_idx = 0; - let mut best_score = u64::MAX; + let mut best_score = u32::MAX; for (idx, server) in self.servers.iter().enumerate() { let score = server.udp_score().score(); if score < best_score { @@ -448,7 +448,7 @@ impl PingChecker { } } - async fn check_delay(&self) -> io::Result { + async fn check_delay(&self) -> io::Result { let start = Instant::now(); // Send HTTP GET and read the first byte @@ -456,7 +456,7 @@ impl PingChecker { let res = time::timeout(timeout, self.check_request()).await; let elapsed = Instant::now() - start; - let elapsed = elapsed.as_secs() * 1000 + u64::from(elapsed.subsec_millis()); // Converted to ms + let elapsed = elapsed.as_secs() as u32 * 1000 + elapsed.subsec_millis(); // Converted to ms match res { Ok(Ok(..)) => { // Got the result ... record its time diff --git a/crates/shadowsocks-service/src/local/loadbalancing/server_data.rs b/crates/shadowsocks-service/src/local/loadbalancing/server_data.rs index 1a23f858..2752ac40 100644 --- a/crates/shadowsocks-service/src/local/loadbalancing/server_data.rs +++ b/crates/shadowsocks-service/src/local/loadbalancing/server_data.rs @@ -2,7 +2,7 @@ use std::{ fmt::{self, Debug}, - sync::atomic::{AtomicU64, Ordering}, + sync::atomic::{AtomicU32, Ordering}, }; use shadowsocks::ServerConfig; @@ -13,7 +13,7 @@ use super::server_stat::{Score, ServerStat}; /// Server's statistic score pub struct ServerScore { stat_data: Mutex, - score: AtomicU64, + score: AtomicU32, } impl ServerScore { @@ -21,17 +21,17 @@ impl ServerScore { pub fn new() -> ServerScore { ServerScore { stat_data: Mutex::new(ServerStat::new()), - score: AtomicU64::new(0), + score: AtomicU32::new(0), } } /// Get server's current statistic scores - pub fn score(&self) -> u64 { + pub fn score(&self) -> u32 { self.score.load(Ordering::Acquire) } /// Append a `Score` into statistic and recalculate score of the server - pub async fn push_score(&self, score: Score) -> u64 { + pub async fn push_score(&self, score: Score) -> u32 { let updated_score = { let mut stat = self.stat_data.lock().await; stat.push_score(score) @@ -41,7 +41,7 @@ impl ServerScore { } /// Report request failure of this server, which will eventually records an `Errored` score - pub async fn report_failure(&self) -> u64 { + pub async fn report_failure(&self) -> u32 { self.push_score(Score::Errored).await } } diff --git a/crates/shadowsocks-service/src/local/loadbalancing/server_stat.rs b/crates/shadowsocks-service/src/local/loadbalancing/server_stat.rs index bdd7d546..0682efbe 100644 --- a/crates/shadowsocks-service/src/local/loadbalancing/server_stat.rs +++ b/crates/shadowsocks-service/src/local/loadbalancing/server_stat.rs @@ -7,14 +7,14 @@ pub const DEFAULT_CHECK_INTERVAL_SEC: u64 = 6; /// Timeout of each probing pub const DEFAULT_CHECK_TIMEOUT_SEC: u64 = 2; // Latency shouldn't greater than 2 secs, that's too long -const MAX_SERVER_RTT: u64 = DEFAULT_CHECK_TIMEOUT_SEC * 1000; +const MAX_SERVER_RTT: u32 = DEFAULT_CHECK_TIMEOUT_SEC as u32 * 1000; const MAX_LATENCY_QUEUE_SIZE: usize = 99; /// Statistic score #[derive(Debug, Copy, Clone)] pub enum Score { /// Unified latency - Latency(u64), + Latency(u32), /// Request error Errored, } @@ -26,7 +26,7 @@ pub struct ServerStat { /// /// Use median instead of average time, /// because probing result may have some really bad cases - rtt: u64, + rtt: u32, /// Total_Fail / Total_Probe fail_rate: f64, /// Recently probe data @@ -57,7 +57,7 @@ impl ServerStat { } } - fn score(&self) -> u64 { + fn score(&self) -> u32 { // Normalize rtt let nrtt = self.rtt as f64 / MAX_SERVER_RTT as f64; @@ -76,11 +76,11 @@ impl ServerStat { let score = (nrtt * SCORE_RTT_WEIGHT + self.fail_rate * SCORE_FAIL_WEIGHT + nstdev * SCORE_STDEV_WEIGHT) / (SCORE_RTT_WEIGHT + SCORE_FAIL_WEIGHT + SCORE_STDEV_WEIGHT); - // Times 10000 converts to u64, for 0.0001 precision - (score * 10000.0) as u64 + // Times 10000 converts to u32, for 0.0001 precision + (score * 10000.0) as u32 } - pub fn push_score(&mut self, score: Score) -> u64 { + pub fn push_score(&mut self, score: Score) -> u32 { self.latency_queue.push_back(score); // Only records recently MAX_LATENCY_QUEUE_SIZE probe data @@ -91,7 +91,7 @@ impl ServerStat { self.recalculate_score() } - fn recalculate_score(&mut self) -> u64 { + fn recalculate_score(&mut self) -> u32 { if self.latency_queue.is_empty() { return self.score(); } diff --git a/crates/shadowsocks-service/src/net/flow.rs b/crates/shadowsocks-service/src/net/flow.rs index d4cd657a..08131c5c 100644 --- a/crates/shadowsocks-service/src/net/flow.rs +++ b/crates/shadowsocks-service/src/net/flow.rs @@ -1,39 +1,72 @@ //! Server flow statistic -use std::sync::atomic::{AtomicU64, Ordering}; +use std::sync::atomic::Ordering; + +#[cfg(not(any(target_arch = "mips", target_arch = "powerpc")))] +type FlowCounter = std::sync::atomic::AtomicU64; +#[cfg(any(target_arch = "mips", target_arch = "powerpc"))] +type FlowCounter = std::sync::atomic::AtomicU32; /// Connection flow statistic pub struct FlowStat { - tx: AtomicU64, - rx: AtomicU64, + tx: FlowCounter, + rx: FlowCounter, } impl FlowStat { /// Create an empty flow statistic pub fn new() -> FlowStat { FlowStat { - tx: AtomicU64::new(0), - rx: AtomicU64::new(0), + tx: FlowCounter::new(0), + rx: FlowCounter::new(0), } } /// Transmitted bytes count + #[cfg(not(any(target_arch = "mips", target_arch = "powerpc")))] pub fn tx(&self) -> u64 { self.tx.load(Ordering::Relaxed) } + /// Transmitted bytes count + #[cfg(any(target_arch = "mips", target_arch = "powerpc"))] + pub fn tx(&self) -> u64 { + self.tx.load(Ordering::Relaxed) as u64 + } + /// Increase transmitted bytes + #[cfg(not(any(target_arch = "mips", target_arch = "powerpc")))] pub fn incr_tx(&self, n: u64) { self.tx.fetch_add(n, Ordering::AcqRel); } + /// Increase transmitted bytes + #[cfg(any(target_arch = "mips", target_arch = "powerpc"))] + pub fn incr_tx(&self, n: u64) { + self.tx.fetch_add(n as u32, Ordering::AcqRel); + } + /// Received bytes count + #[cfg(not(any(target_arch = "mips", target_arch = "powerpc")))] pub fn rx(&self) -> u64 { self.rx.load(Ordering::Relaxed) } + /// Received bytes count + #[cfg(any(target_arch = "mips", target_arch = "powerpc"))] + pub fn rx(&self) -> u64 { + self.rx.load(Ordering::Relaxed) as u32 + } + /// Increase received bytes + #[cfg(not(any(target_arch = "mips", target_arch = "powerpc")))] pub fn incr_rx(&self, n: u64) { self.rx.fetch_add(n, Ordering::AcqRel); } + + /// Increase received bytes + #[cfg(any(target_arch = "mips", target_arch = "powerpc"))] + pub fn incr_rx(&self, n: u64) { + self.rx.fetch_add(n as u32, Ordering::AcqRel); + } }