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

[Christian Ebner <c.ebner@proxmox.com>]

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))
-    }
-}
-- 
2.47.3



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