Re: [pbs-devel] [PATCH proxmox v2 2/4] http: drop factored out rate limiter implementation

[Hannes Laimer <h.laimer@proxmox.com>]

Reviewed-by: Hannes Laimer <h.laimer@proxmox.com>
Tested-by: Hannes Laimer <h.laimer@proxmox.com>

On 9/16/25 14:42, Christian Ebner wrote:
> The rate limiter implementation has been moved together with the shared
> rate limiter into a dedicated crate. Depend on that and drop the now
> dead code from the proxmox-http crate.
> 
> Signed-off-by: Christian Ebner <c.ebner@proxmox.com>
> ---
> Changes since version 1:
> - not present in previous version
> 
>   Cargo.toml                              |   1 +
>   proxmox-http/Cargo.toml                 |   4 +-
>   proxmox-http/debian/control             |  19 +--
>   proxmox-http/src/client/connector.rs    |   3 +-
>   proxmox-http/src/lib.rs                 |   5 -
>   proxmox-http/src/rate_limited_stream.rs |   2 +-
>   proxmox-http/src/rate_limiter.rs        | 214 ------------------------
>   7 files changed, 8 insertions(+), 240 deletions(-)
>   delete mode 100644 proxmox-http/src/rate_limiter.rs
> 
> diff --git a/Cargo.toml b/Cargo.toml
> index bde32b17..f62b9882 100644
> --- a/Cargo.toml
> +++ b/Cargo.toml
> @@ -153,6 +153,7 @@ proxmox-login = { version = "1.0.0", path = "proxmox-login" }
>   proxmox-network-types = { version = "0.1.0", path = "proxmox-network-types" }
>   proxmox-product-config = { version = "1.0.0", path = "proxmox-product-config" }
>   proxmox-config-digest = { version = "1.0.0", path = "proxmox-config-digest" }
> +proxmox-rate-limiter = { version = "1.0.0", path = "proxmox-rate-limiter" }
>   proxmox-rest-server = { version = "1.0.0", path = "proxmox-rest-server" }
>   proxmox-router = { version = "3.2.2", path = "proxmox-router" }
>   proxmox-s3-client = { version = "1.2", path = "proxmox-s3-client" }
> diff --git a/proxmox-http/Cargo.toml b/proxmox-http/Cargo.toml
> index 2f83cf5f..0b362bef 100644
> --- a/proxmox-http/Cargo.toml
> +++ b/proxmox-http/Cargo.toml
> @@ -32,6 +32,7 @@ url = { workspace = true, optional = true }
>   
>   proxmox-async = { workspace = true, optional = true }
>   proxmox-base64 = { workspace = true, optional = true }
> +proxmox-rate-limiter = { workspace = true, optional = true, features = [ "rate-limiter" ] }
>   proxmox-sys = { workspace = true, optional = true }
>   proxmox-io = { workspace = true, optional = true }
>   proxmox-lang = { workspace = true, optional = true }
> @@ -53,7 +54,6 @@ body = [
>       "dep:sync_wrapper",
>       "sync_wrapper?/futures",
>   ]
> -rate-limiter = ["dep:hyper"]
>   rate-limited-stream = [
>       "dep:tokio",
>       "dep:hyper-util",
> @@ -61,7 +61,7 @@ rate-limited-stream = [
>       "hyper-util?/client-legacy",
>       "hyper-util?/http1",
>       "tokio?/time",
> -    "rate-limiter",
> +    "dep:proxmox-rate-limiter",
>   ]
>   client = [
>       "dep:bytes",
> diff --git a/proxmox-http/debian/control b/proxmox-http/debian/control
> index f86e58d1..ae3af111 100644
> --- a/proxmox-http/debian/control
> +++ b/proxmox-http/debian/control
> @@ -29,7 +29,6 @@ Suggests:
>    librust-proxmox-http+http-helpers-dev (= ${binary:Version}),
>    librust-proxmox-http+proxmox-async-dev (= ${binary:Version}),
>    librust-proxmox-http+rate-limited-stream-dev (= ${binary:Version}),
> - librust-proxmox-http+rate-limiter-dev (= ${binary:Version}),
>    librust-proxmox-http+websocket-dev (= ${binary:Version})
>   Provides:
>    librust-proxmox-http+default-dev (= ${binary:Version}),
> @@ -172,12 +171,13 @@ Multi-Arch: same
>   Depends:
>    ${misc:Depends},
>    librust-proxmox-http-dev (= ${binary:Version}),
> - librust-proxmox-http+rate-limiter-dev (= ${binary:Version}),
>    librust-hyper-util-0.1+client-dev (>= 0.1.12-~~),
>    librust-hyper-util-0.1+client-legacy-dev (>= 0.1.12-~~),
>    librust-hyper-util-0.1+default-dev (>= 0.1.12-~~),
>    librust-hyper-util-0.1+http1-dev (>= 0.1.12-~~),
>    librust-hyper-util-0.1+http2-dev (>= 0.1.12-~~),
> + librust-proxmox-rate-limiter-1+default-dev,
> + librust-proxmox-rate-limiter-1+rate-limiter-dev,
>    librust-tokio-1+default-dev (>= 1.6-~~),
>    librust-tokio-1+time-dev (>= 1.6-~~)
>   Provides:
> @@ -188,21 +188,6 @@ Description: Proxmox HTTP library - feature "rate-limited-stream"
>    This metapackage enables feature "rate-limited-stream" for the Rust proxmox-
>    http crate, by pulling in any additional dependencies needed by that feature.
>   
> -Package: librust-proxmox-http+rate-limiter-dev
> -Architecture: any
> -Multi-Arch: same
> -Depends:
> - ${misc:Depends},
> - librust-proxmox-http-dev (= ${binary:Version}),
> - librust-hyper-1+default-dev
> -Provides:
> - librust-proxmox-http-1+rate-limiter-dev (= ${binary:Version}),
> - librust-proxmox-http-1.0+rate-limiter-dev (= ${binary:Version}),
> - librust-proxmox-http-1.0.3+rate-limiter-dev (= ${binary:Version})
> -Description: Proxmox HTTP library - feature "rate-limiter"
> - This metapackage enables feature "rate-limiter" for the Rust proxmox-http
> - crate, by pulling in any additional dependencies needed by that feature.
> -
>   Package: librust-proxmox-http+websocket-dev
>   Architecture: any
>   Multi-Arch: same
> diff --git a/proxmox-http/src/client/connector.rs b/proxmox-http/src/client/connector.rs
> index 1600d47c..d5d85cb9 100644
> --- a/proxmox-http/src/client/connector.rs
> +++ b/proxmox-http/src/client/connector.rs
> @@ -13,13 +13,14 @@ use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt};
>   use tokio::net::TcpStream;
>   use tokio_openssl::SslStream;
>   
> +use proxmox_rate_limiter::ShareableRateLimit;
>   use proxmox_sys::linux::socket::set_tcp_keepalive;
>   
>   use crate::proxy_config::ProxyConfig;
>   use crate::uri::build_authority;
>   
>   use super::tls::MaybeTlsStream;
> -use crate::{RateLimitedStream, ShareableRateLimit};
> +use crate::RateLimitedStream;
>   
>   type SharedRateLimit = Arc<dyn ShareableRateLimit>;
>   
> diff --git a/proxmox-http/src/lib.rs b/proxmox-http/src/lib.rs
> index 8b6953b0..7f6f67f8 100644
> --- a/proxmox-http/src/lib.rs
> +++ b/proxmox-http/src/lib.rs
> @@ -26,11 +26,6 @@ mod client_trait;
>   #[cfg(feature = "client-trait")]
>   pub use client_trait::HttpClient;
>   
> -#[cfg(feature = "rate-limiter")]
> -mod rate_limiter;
> -#[cfg(feature = "rate-limiter")]
> -pub use rate_limiter::{RateLimit, RateLimiter, RateLimiterVec, ShareableRateLimit};
> -
>   #[cfg(feature = "rate-limited-stream")]
>   mod rate_limited_stream;
>   #[cfg(feature = "rate-limited-stream")]
> diff --git a/proxmox-http/src/rate_limited_stream.rs b/proxmox-http/src/rate_limited_stream.rs
> index e9308a47..2253bef8 100644
> --- a/proxmox-http/src/rate_limited_stream.rs
> +++ b/proxmox-http/src/rate_limited_stream.rs
> @@ -11,7 +11,7 @@ use tokio::time::Sleep;
>   
>   use std::task::{Context, Poll};
>   
> -use super::{RateLimiter, ShareableRateLimit};
> +use proxmox_rate_limiter::{RateLimiter, ShareableRateLimit};
>   
>   type SharedRateLimit = Arc<dyn ShareableRateLimit>;
>   
> diff --git a/proxmox-http/src/rate_limiter.rs b/proxmox-http/src/rate_limiter.rs
> deleted file mode 100644
> index 945c77a6..00000000
> --- a/proxmox-http/src/rate_limiter.rs
> +++ /dev/null
> @@ -1,214 +0,0 @@
> -use std::convert::TryInto;
> -use std::time::{Duration, Instant};
> -
> -use anyhow::{bail, Error};
> -
> -/// Rate limiter interface.
> -pub trait RateLimit {
> -    /// Update rate and bucket size
> -    fn update_rate(&mut self, rate: u64, bucket_size: u64);
> -
> -    /// Returns the overall traffic (since started)
> -    fn traffic(&self) -> u64;
> -
> -    /// Register traffic, returning a proposed delay to reach the
> -    /// expected rate.
> -    fn register_traffic(&mut self, current_time: Instant, data_len: u64) -> Duration;
> -}
> -
> -/// Like [`RateLimit`], but does not require self to be mutable.
> -///
> -/// This is useful for types providing internal mutability (Mutex).
> -pub trait ShareableRateLimit: Send + Sync {
> -    fn update_rate(&self, rate: u64, bucket_size: u64);
> -    fn traffic(&self) -> u64;
> -    fn register_traffic(&self, current_time: Instant, data_len: u64) -> Duration;
> -}
> -
> -/// IMPORTANT: We use this struct in shared memory, so please do not
> -/// change/modify the layout (do not add fields)
> -#[derive(Clone)]
> -#[repr(C)]
> -struct TbfState {
> -    traffic: u64, // overall traffic
> -    last_update: Instant,
> -    consumed_tokens: u64,
> -}
> -
> -impl TbfState {
> -    const NO_DELAY: Duration = Duration::from_millis(0);
> -
> -    fn refill_bucket(&mut self, rate: u64, current_time: Instant) {
> -        let time_diff = match current_time.checked_duration_since(self.last_update) {
> -            Some(duration) => duration.as_nanos(),
> -            None => return,
> -        };
> -
> -        if time_diff == 0 {
> -            return;
> -        }
> -
> -        self.last_update = current_time;
> -
> -        let allowed_traffic = ((time_diff.saturating_mul(rate as u128)) / 1_000_000_000)
> -            .try_into()
> -            .unwrap_or(u64::MAX);
> -
> -        self.consumed_tokens = self.consumed_tokens.saturating_sub(allowed_traffic);
> -    }
> -
> -    fn register_traffic(
> -        &mut self,
> -        rate: u64,
> -        bucket_size: u64,
> -        current_time: Instant,
> -        data_len: u64,
> -    ) -> Duration {
> -        self.refill_bucket(rate, current_time);
> -
> -        self.traffic += data_len;
> -        self.consumed_tokens += data_len;
> -
> -        if self.consumed_tokens <= bucket_size {
> -            return Self::NO_DELAY;
> -        }
> -        Duration::from_nanos(
> -            (self.consumed_tokens - bucket_size).saturating_mul(1_000_000_000) / rate,
> -        )
> -    }
> -}
> -
> -/// Token bucket based rate limiter
> -///
> -/// IMPORTANT: We use this struct in shared memory, so please do not
> -/// change/modify the layout (do not add fields)
> -#[repr(C)]
> -pub struct RateLimiter {
> -    rate: u64,        // tokens/second
> -    bucket_size: u64, // TBF bucket size
> -    state: TbfState,
> -}
> -
> -impl RateLimiter {
> -    /// Creates a new instance, using [Instant::now] as start time.
> -    pub fn new(rate: u64, bucket_size: u64) -> Self {
> -        let start_time = Instant::now();
> -        Self::with_start_time(rate, bucket_size, start_time)
> -    }
> -
> -    /// Creates a new instance with specified `rate`, `bucket_size` and `start_time`.
> -    pub fn with_start_time(rate: u64, bucket_size: u64, start_time: Instant) -> Self {
> -        Self {
> -            rate,
> -            bucket_size,
> -            state: TbfState {
> -                traffic: 0,
> -                last_update: start_time,
> -                // start with empty bucket (all tokens consumed)
> -                consumed_tokens: bucket_size,
> -            },
> -        }
> -    }
> -}
> -
> -impl RateLimit for RateLimiter {
> -    fn update_rate(&mut self, rate: u64, bucket_size: u64) {
> -        self.rate = rate;
> -
> -        if bucket_size < self.bucket_size && self.state.consumed_tokens > bucket_size {
> -            self.state.consumed_tokens = bucket_size; // start again
> -        }
> -
> -        self.bucket_size = bucket_size;
> -    }
> -
> -    fn traffic(&self) -> u64 {
> -        self.state.traffic
> -    }
> -
> -    fn register_traffic(&mut self, current_time: Instant, data_len: u64) -> Duration {
> -        self.state
> -            .register_traffic(self.rate, self.bucket_size, current_time, data_len)
> -    }
> -}
> -
> -impl<R: RateLimit + Send> ShareableRateLimit for std::sync::Mutex<R> {
> -    fn update_rate(&self, rate: u64, bucket_size: u64) {
> -        self.lock().unwrap().update_rate(rate, bucket_size);
> -    }
> -
> -    fn traffic(&self) -> u64 {
> -        self.lock().unwrap().traffic()
> -    }
> -
> -    fn register_traffic(&self, current_time: Instant, data_len: u64) -> Duration {
> -        self.lock()
> -            .unwrap()
> -            .register_traffic(current_time, data_len)
> -    }
> -}
> -
> -/// Array of rate limiters.
> -///
> -/// A group of rate limiters with same configuration.
> -pub struct RateLimiterVec {
> -    rate: u64,        // tokens/second
> -    bucket_size: u64, // TBF bucket size
> -    state: Vec<TbfState>,
> -}
> -
> -impl RateLimiterVec {
> -    /// Creates a new instance, using [Instant::now] as start time.
> -    pub fn new(group_size: usize, rate: u64, bucket_size: u64) -> Self {
> -        let start_time = Instant::now();
> -        Self::with_start_time(group_size, rate, bucket_size, start_time)
> -    }
> -
> -    /// Creates a new instance with specified `rate`, `bucket_size` and `start_time`.
> -    pub fn with_start_time(
> -        group_size: usize,
> -        rate: u64,
> -        bucket_size: u64,
> -        start_time: Instant,
> -    ) -> Self {
> -        let state = TbfState {
> -            traffic: 0,
> -            last_update: start_time,
> -            // start with empty bucket (all tokens consumed)
> -            consumed_tokens: bucket_size,
> -        };
> -        Self {
> -            rate,
> -            bucket_size,
> -            state: vec![state; group_size],
> -        }
> -    }
> -
> -    #[allow(clippy::len_without_is_empty)]
> -    /// Return the number of TBF entries (group_size)
> -    pub fn len(&self) -> usize {
> -        self.state.len()
> -    }
> -
> -    /// Traffic for the specified index
> -    pub fn traffic(&self, index: usize) -> Result<u64, Error> {
> -        if index >= self.state.len() {
> -            bail!("RateLimiterVec::traffic - index out of range");
> -        }
> -        Ok(self.state[index].traffic)
> -    }
> -
> -    /// Register traffic at the specified index
> -    pub fn register_traffic(
> -        &mut self,
> -        index: usize,
> -        current_time: Instant,
> -        data_len: u64,
> -    ) -> Result<Duration, Error> {
> -        if index >= self.state.len() {
> -            bail!("RateLimiterVec::register_traffic - index out of range");
> -        }
> -
> -        Ok(self.state[index].register_traffic(self.rate, self.bucket_size, current_time, data_len))
> -    }
> -}



_______________________________________________
pbs-devel mailing list
pbs-devel@lists.proxmox.com
https://lists.proxmox.com/cgi-bin/mailman/listinfo/pbs-devel