use parking_lot::Mutex;
use std::sync::Arc;
use std::time::{Duration, Instant};
use thiserror::Error;
#[derive(Error, Debug, Clone)]
pub enum ThrottleError {
#[error("Bandwidth limit exceeded, retry after {0:?}")]
RateLimitExceeded(Duration),
#[error("Invalid throttle configuration: {0}")]
InvalidConfig(String),
#[error("Throttle disabled")]
Disabled,
}
#[derive(Debug, Clone)]
pub struct ThrottleConfig {
pub max_upload_bytes_per_sec: Option<u64>,
pub max_download_bytes_per_sec: Option<u64>,
pub burst_size_bytes: Option<u64>,
pub enabled: bool,
pub refill_interval: Duration,
}
impl Default for ThrottleConfig {
fn default() -> Self {
Self {
max_upload_bytes_per_sec: None,
max_download_bytes_per_sec: None,
burst_size_bytes: None,
enabled: false,
refill_interval: Duration::from_millis(100), }
}
}
impl ThrottleConfig {
pub fn mobile() -> Self {
Self {
max_upload_bytes_per_sec: Some(1_000_000), max_download_bytes_per_sec: Some(5_000_000), burst_size_bytes: Some(2_000_000), enabled: true,
refill_interval: Duration::from_millis(100),
}
}
pub fn iot() -> Self {
Self {
max_upload_bytes_per_sec: Some(128_000), max_download_bytes_per_sec: Some(512_000), burst_size_bytes: Some(256_000), enabled: true,
refill_interval: Duration::from_millis(100),
}
}
pub fn low_power() -> Self {
Self {
max_upload_bytes_per_sec: Some(64_000), max_download_bytes_per_sec: Some(256_000), burst_size_bytes: Some(128_000), enabled: true,
refill_interval: Duration::from_millis(200), }
}
pub fn validate(&self) -> Result<(), ThrottleError> {
if self.refill_interval.is_zero() {
return Err(ThrottleError::InvalidConfig(
"Refill interval must be > 0".to_string(),
));
}
if let Some(burst) = self.burst_size_bytes {
if burst == 0 {
return Err(ThrottleError::InvalidConfig(
"Burst size must be > 0 if specified".to_string(),
));
}
}
Ok(())
}
}
#[derive(Debug)]
struct TokenBucket {
tokens: f64,
capacity: f64,
refill_rate: f64,
last_refill: Instant,
refill_interval: Duration,
}
impl TokenBucket {
fn new(rate_bytes_per_sec: u64, burst_bytes: u64, refill_interval: Duration) -> Self {
Self {
tokens: burst_bytes as f64,
capacity: burst_bytes as f64,
refill_rate: rate_bytes_per_sec as f64,
last_refill: Instant::now(),
refill_interval,
}
}
fn refill(&mut self) {
let now = Instant::now();
let elapsed = now.duration_since(self.last_refill);
if elapsed >= self.refill_interval {
let tokens_to_add = self.refill_rate * elapsed.as_secs_f64();
self.tokens = (self.tokens + tokens_to_add).min(self.capacity);
self.last_refill = now;
}
}
fn consume(&mut self, bytes: u64) -> Result<(), Duration> {
self.refill();
if self.tokens >= bytes as f64 {
self.tokens -= bytes as f64;
Ok(())
} else {
let tokens_needed = bytes as f64 - self.tokens;
let wait_time = Duration::from_secs_f64(tokens_needed / self.refill_rate);
Err(wait_time)
}
}
fn available_tokens(&mut self) -> u64 {
self.refill();
self.tokens as u64
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum TrafficDirection {
Upload,
Download,
}
#[derive(Clone)]
pub struct BandwidthThrottle {
config: ThrottleConfig,
upload_bucket: Arc<Mutex<Option<TokenBucket>>>,
download_bucket: Arc<Mutex<Option<TokenBucket>>>,
}
impl BandwidthThrottle {
pub fn new(config: ThrottleConfig) -> Result<Self, ThrottleError> {
config.validate()?;
let upload_bucket = config.max_upload_bytes_per_sec.map(|rate| {
let burst = config.burst_size_bytes.unwrap_or(rate * 2);
TokenBucket::new(rate, burst, config.refill_interval)
});
let download_bucket = config.max_download_bytes_per_sec.map(|rate| {
let burst = config.burst_size_bytes.unwrap_or(rate * 2);
TokenBucket::new(rate, burst, config.refill_interval)
});
Ok(Self {
config: config.clone(),
upload_bucket: Arc::new(Mutex::new(upload_bucket)),
download_bucket: Arc::new(Mutex::new(download_bucket)),
})
}
pub fn check_and_consume(
&self,
direction: TrafficDirection,
bytes: u64,
) -> Result<(), ThrottleError> {
if !self.config.enabled {
return Err(ThrottleError::Disabled);
}
let bucket = match direction {
TrafficDirection::Upload => &self.upload_bucket,
TrafficDirection::Download => &self.download_bucket,
};
let mut guard = bucket.lock();
if let Some(bucket) = guard.as_mut() {
bucket
.consume(bytes)
.map_err(ThrottleError::RateLimitExceeded)
} else {
Ok(())
}
}
pub fn available_bandwidth(&self, direction: TrafficDirection) -> Option<u64> {
if !self.config.enabled {
return None;
}
let bucket = match direction {
TrafficDirection::Upload => &self.upload_bucket,
TrafficDirection::Download => &self.download_bucket,
};
let mut guard = bucket.lock();
guard.as_mut().map(|b| b.available_tokens())
}
pub fn set_enabled(&mut self, enabled: bool) {
Arc::make_mut(&mut Arc::new(self.config.clone())).enabled = enabled;
}
pub fn is_enabled(&self) -> bool {
self.config.enabled
}
pub fn config(&self) -> &ThrottleConfig {
&self.config
}
pub fn update_config(&mut self, config: ThrottleConfig) -> Result<(), ThrottleError> {
config.validate()?;
let upload_bucket = config.max_upload_bytes_per_sec.map(|rate| {
let burst = config.burst_size_bytes.unwrap_or(rate * 2);
TokenBucket::new(rate, burst, config.refill_interval)
});
let download_bucket = config.max_download_bytes_per_sec.map(|rate| {
let burst = config.burst_size_bytes.unwrap_or(rate * 2);
TokenBucket::new(rate, burst, config.refill_interval)
});
*self.upload_bucket.lock() = upload_bucket;
*self.download_bucket.lock() = download_bucket;
self.config = config;
Ok(())
}
}
#[derive(Debug, Clone, Default)]
pub struct ThrottleStats {
pub upload_bytes_allowed: u64,
pub upload_bytes_throttled: u64,
pub download_bytes_allowed: u64,
pub download_bytes_throttled: u64,
pub upload_throttle_count: u64,
pub download_throttle_count: u64,
}
#[cfg(test)]
mod tests {
use super::*;
use std::thread;
#[test]
fn test_throttle_config_default() {
let config = ThrottleConfig::default();
assert!(!config.enabled);
assert!(config.max_upload_bytes_per_sec.is_none());
assert!(config.max_download_bytes_per_sec.is_none());
}
#[test]
fn test_throttle_config_mobile() {
let config = ThrottleConfig::mobile();
assert!(config.enabled);
assert_eq!(config.max_upload_bytes_per_sec, Some(1_000_000));
assert_eq!(config.max_download_bytes_per_sec, Some(5_000_000));
assert!(config.validate().is_ok());
}
#[test]
fn test_throttle_config_iot() {
let config = ThrottleConfig::iot();
assert!(config.enabled);
assert_eq!(config.max_upload_bytes_per_sec, Some(128_000));
assert_eq!(config.max_download_bytes_per_sec, Some(512_000));
assert!(config.validate().is_ok());
}
#[test]
fn test_throttle_config_low_power() {
let config = ThrottleConfig::low_power();
assert!(config.enabled);
assert_eq!(config.max_upload_bytes_per_sec, Some(64_000));
assert_eq!(config.max_download_bytes_per_sec, Some(256_000));
assert!(config.validate().is_ok());
}
#[test]
fn test_throttle_disabled() {
let config = ThrottleConfig::default();
let throttle =
BandwidthThrottle::new(config).expect("test: default config should create throttle");
let result = throttle.check_and_consume(TrafficDirection::Upload, 1000);
assert!(matches!(result, Err(ThrottleError::Disabled)));
}
#[test]
fn test_throttle_upload_within_limit() {
let config = ThrottleConfig {
enabled: true,
max_upload_bytes_per_sec: Some(1000),
burst_size_bytes: Some(2000),
..Default::default()
};
let throttle = BandwidthThrottle::new(config)
.expect("test: upload within limit config should create throttle");
let result = throttle.check_and_consume(TrafficDirection::Upload, 1500);
assert!(result.is_ok());
}
#[test]
fn test_throttle_upload_exceeds_limit() {
let config = ThrottleConfig {
enabled: true,
max_upload_bytes_per_sec: Some(1000),
burst_size_bytes: Some(2000),
..Default::default()
};
let throttle = BandwidthThrottle::new(config)
.expect("test: upload exceeds limit config should create throttle");
let _ = throttle.check_and_consume(TrafficDirection::Upload, 2000);
let result = throttle.check_and_consume(TrafficDirection::Upload, 100);
assert!(matches!(result, Err(ThrottleError::RateLimitExceeded(_))));
}
#[test]
fn test_throttle_download_within_limit() {
let config = ThrottleConfig {
enabled: true,
max_download_bytes_per_sec: Some(5000),
burst_size_bytes: Some(10000),
..Default::default()
};
let throttle = BandwidthThrottle::new(config)
.expect("test: download within limit config should create throttle");
let result = throttle.check_and_consume(TrafficDirection::Download, 8000);
assert!(result.is_ok());
}
#[test]
fn test_throttle_refill() {
let config = ThrottleConfig {
enabled: true,
max_upload_bytes_per_sec: Some(1000),
burst_size_bytes: Some(1000),
refill_interval: Duration::from_millis(100),
..Default::default()
};
let throttle =
BandwidthThrottle::new(config).expect("test: refill config should create throttle");
let _ = throttle.check_and_consume(TrafficDirection::Upload, 1000);
thread::sleep(Duration::from_millis(150));
let available = throttle.available_bandwidth(TrafficDirection::Upload);
assert!(available.is_some());
assert!(available.expect("test: bandwidth should be available after refill") > 0);
}
#[test]
fn test_throttle_available_bandwidth() {
let config = ThrottleConfig {
enabled: true,
max_upload_bytes_per_sec: Some(1000),
burst_size_bytes: Some(2000),
..Default::default()
};
let throttle = BandwidthThrottle::new(config)
.expect("test: available bandwidth config should create throttle");
let available = throttle.available_bandwidth(TrafficDirection::Upload);
assert_eq!(available, Some(2000)); }
#[test]
fn test_throttle_independent_directions() {
let config = ThrottleConfig {
enabled: true,
max_upload_bytes_per_sec: Some(1000),
max_download_bytes_per_sec: Some(5000),
burst_size_bytes: Some(2000),
..Default::default()
};
let throttle = BandwidthThrottle::new(config)
.expect("test: independent directions config should create throttle");
let _ = throttle.check_and_consume(TrafficDirection::Upload, 2000);
let result = throttle.check_and_consume(TrafficDirection::Download, 2000);
assert!(result.is_ok());
}
#[test]
fn test_throttle_update_config() {
let config = ThrottleConfig {
enabled: true,
max_upload_bytes_per_sec: Some(1000),
..Default::default()
};
let mut throttle = BandwidthThrottle::new(config)
.expect("test: update config initial config should create throttle");
let new_config = ThrottleConfig {
enabled: true,
max_upload_bytes_per_sec: Some(5000),
burst_size_bytes: Some(10000),
..Default::default()
};
throttle
.update_config(new_config)
.expect("test: update_config with valid new config should succeed");
let available = throttle.available_bandwidth(TrafficDirection::Upload);
assert_eq!(available, Some(10000));
}
#[test]
fn test_throttle_config_validation() {
let config = ThrottleConfig {
refill_interval: Duration::from_secs(0),
..Default::default()
};
let result = BandwidthThrottle::new(config);
assert!(matches!(result, Err(ThrottleError::InvalidConfig(_))));
}
#[test]
fn test_throttle_no_limit_direction() {
let config = ThrottleConfig {
enabled: true,
max_upload_bytes_per_sec: Some(1000),
..Default::default()
};
let throttle = BandwidthThrottle::new(config)
.expect("test: no limit direction config should create throttle");
let result = throttle.check_and_consume(TrafficDirection::Download, 1_000_000);
assert!(result.is_ok());
}
}