Re: [pbs-devel] [PATCH proxmox v2 1/4] rate-limiter: add crate for traffic rate limiter implementations

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

comment inline, other than that consider this

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:
> Factors out the traffic rate limiter implementations currently tied
> to the proxmox-backup and proxmox-http crates to make them
> independent and easily reusable, e.g. for the s3-client
> implementation.
> 
> The shared rate limiter implementation from PBS relies on mmapping
> for state sharing, the file exposed having a predefined magic number.
> In order to be backwards compatible, leave the magic number as is and
> only adapt the constant name to be more generic, although the string
> the magic number is derived from is PBS specific.
> 
> Further, the user for file ownership and base path of the mmapped
> file are now passed as parameters during shared rate limiter
> instantiation.
> 
> Signed-off-by: Christian Ebner <c.ebner@proxmox.com>
> ---
> Changes since version 1:
> - move rate limiter implementations into dedicated crate instead of
>    proxmox-http
> - Adapt shared state file magic constant name to be generic
> 
>   Cargo.toml                                    |   1 +
>   proxmox-rate-limiter/Cargo.toml               |  29 +++
>   proxmox-rate-limiter/debian/changelog         |   5 +
>   proxmox-rate-limiter/debian/control           |  70 ++++++
>   proxmox-rate-limiter/debian/copyright         |  18 ++
>   proxmox-rate-limiter/debian/debcargo.toml     |   7 +
>   proxmox-rate-limiter/src/lib.rs               |  13 ++
>   proxmox-rate-limiter/src/rate_limiter.rs      | 214 ++++++++++++++++++
>   .../src/shared_rate_limiter.rs                | 130 +++++++++++
>   9 files changed, 487 insertions(+)
>   create mode 100644 proxmox-rate-limiter/Cargo.toml
>   create mode 100644 proxmox-rate-limiter/debian/changelog
>   create mode 100644 proxmox-rate-limiter/debian/control
>   create mode 100644 proxmox-rate-limiter/debian/copyright
>   create mode 100644 proxmox-rate-limiter/debian/debcargo.toml
>   create mode 100644 proxmox-rate-limiter/src/lib.rs
>   create mode 100644 proxmox-rate-limiter/src/rate_limiter.rs
>   create mode 100644 proxmox-rate-limiter/src/shared_rate_limiter.rs
> 
> diff --git a/Cargo.toml b/Cargo.toml
> index f149af65..bde32b17 100644
> --- a/Cargo.toml
> +++ b/Cargo.toml
> @@ -29,6 +29,7 @@ members = [
>       "proxmox-notify",
>       "proxmox-openid",
>       "proxmox-product-config",
> +    "proxmox-rate-limiter",
>       "proxmox-resource-scheduling",
>       "proxmox-rest-server",
>       "proxmox-router",
> diff --git a/proxmox-rate-limiter/Cargo.toml b/proxmox-rate-limiter/Cargo.toml
> new file mode 100644
> index 00000000..6d8d96cc
> --- /dev/null
> +++ b/proxmox-rate-limiter/Cargo.toml
> @@ -0,0 +1,29 @@
> +[package]
> +name = "proxmox-rate-limiter"
> +description = "Token bucket based traffic rate limiter implementation"
> +version = "1.0.0"
> +
> +authors.workspace = true
> +edition.workspace = true
> +exclude.workspace = true
> +homepage.workspace = true
> +license.workspace = true
> +repository.workspace = true
> +rust-version.workspace = true
> +
> +[dependencies]
> +anyhow.workspace = true
> +hyper = { workspace = true, optional = true }
> +nix = { workspace = true, optional = true }
> +
> +proxmox-shared-memory = { workspace = true, optional = true }
> +proxmox-sys = { workspace = true, optional = true }
> +
> +[features]
> +default = []
> +rate-limiter = ["dep:hyper"]
> +shared-rate-limiter = [
> +    "dep:nix",
> +    "dep:proxmox-shared-memory",
> +    "dep:proxmox-sys",
> +]
> diff --git a/proxmox-rate-limiter/debian/changelog b/proxmox-rate-limiter/debian/changelog
> new file mode 100644
> index 00000000..0bffa551
> --- /dev/null
> +++ b/proxmox-rate-limiter/debian/changelog
> @@ -0,0 +1,5 @@
> +rust-proxmox-rate-limiter (1.0.0-1) bookworm; urgency=medium
> +
> +  * initial packaging
> +
> + -- Proxmox Support Team <support@proxmox.com>  Tue, 16 Sep 2025 11:06:23 +0200
> diff --git a/proxmox-rate-limiter/debian/control b/proxmox-rate-limiter/debian/control
> new file mode 100644
> index 00000000..689fe02e
> --- /dev/null
> +++ b/proxmox-rate-limiter/debian/control
> @@ -0,0 +1,70 @@
> +Source: rust-proxmox-rate-limiter
> +Section: rust
> +Priority: optional
> +Build-Depends: debhelper-compat (= 13),
> + dh-sequence-cargo
> +Build-Depends-Arch: cargo:native <!nocheck>,
> + rustc:native (>= 1.82) <!nocheck>,
> + libstd-rust-dev <!nocheck>,
> + librust-anyhow-1+default-dev <!nocheck>
> +Maintainer: Proxmox Support Team <support@proxmox.com>
> +Standards-Version: 4.7.0
> +Vcs-Git: git://git.proxmox.com/git/proxmox.git
> +Vcs-Browser: https://git.proxmox.com/?p=proxmox.git
> +Homepage: https://proxmox.com
> +X-Cargo-Crate: proxmox-rate-limiter
> +Rules-Requires-Root: no
> +
> +Package: librust-proxmox-rate-limiter-dev
> +Architecture: any
> +Multi-Arch: same
> +Depends:
> + ${misc:Depends},
> + librust-anyhow-1+default-dev
> +Suggests:
> + librust-proxmox-rate-limiter+rate-limiter-dev (= ${binary:Version}),
> + librust-proxmox-rate-limiter+shared-rate-limiter-dev (= ${binary:Version})
> +Provides:
> + librust-proxmox-rate-limiter+default-dev (= ${binary:Version}),
> + librust-proxmox-rate-limiter-1-dev (= ${binary:Version}),
> + librust-proxmox-rate-limiter-1+default-dev (= ${binary:Version}),
> + librust-proxmox-rate-limiter-1.0-dev (= ${binary:Version}),
> + librust-proxmox-rate-limiter-1.0+default-dev (= ${binary:Version}),
> + librust-proxmox-rate-limiter-1.0.0-dev (= ${binary:Version}),
> + librust-proxmox-rate-limiter-1.0.0+default-dev (= ${binary:Version})
> +Description: Token bucket based traffic rate limiter implementation - Rust source code
> + Source code for Debianized Rust crate "proxmox-rate-limiter"
> +
> +Package: librust-proxmox-rate-limiter+rate-limiter-dev
> +Architecture: any
> +Multi-Arch: same
> +Depends:
> + ${misc:Depends},
> + librust-proxmox-rate-limiter-dev (= ${binary:Version}),
> + librust-hyper-1+default-dev
> +Provides:
> + librust-proxmox-rate-limiter-1+rate-limiter-dev (= ${binary:Version}),
> + librust-proxmox-rate-limiter-1.0+rate-limiter-dev (= ${binary:Version}),
> + librust-proxmox-rate-limiter-1.0.0+rate-limiter-dev (= ${binary:Version})
> +Description: Token bucket based traffic rate limiter implementation - feature "rate-limiter"
> + This metapackage enables feature "rate-limiter" for the Rust proxmox-rate-
> + limiter crate, by pulling in any additional dependencies needed by that
> + feature.
> +
> +Package: librust-proxmox-rate-limiter+shared-rate-limiter-dev
> +Architecture: any
> +Multi-Arch: same
> +Depends:
> + ${misc:Depends},
> + librust-proxmox-rate-limiter-dev (= ${binary:Version}),
> + librust-nix-0.29+default-dev,
> + librust-proxmox-shared-memory-1+default-dev,
> + librust-proxmox-sys-1+default-dev
> +Provides:
> + librust-proxmox-rate-limiter-1+shared-rate-limiter-dev (= ${binary:Version}),
> + librust-proxmox-rate-limiter-1.0+shared-rate-limiter-dev (= ${binary:Version}),
> + librust-proxmox-rate-limiter-1.0.0+shared-rate-limiter-dev (= ${binary:Version})
> +Description: Token bucket based traffic rate limiter implementation - feature "shared-rate-limiter"
> + This metapackage enables feature "shared-rate-limiter" for the Rust proxmox-
> + rate-limiter crate, by pulling in any additional dependencies needed by that
> + feature.
> diff --git a/proxmox-rate-limiter/debian/copyright b/proxmox-rate-limiter/debian/copyright
> new file mode 100644
> index 00000000..d6e3c304
> --- /dev/null
> +++ b/proxmox-rate-limiter/debian/copyright
> @@ -0,0 +1,18 @@
> +Format: https://www.debian.org/doc/packaging-manuals/copyright-format/1.0/
> +
> +Files:
> + *
> +Copyright: 2025 Proxmox Server Solutions GmbH <support@proxmox.com>
> +License: AGPL-3.0-or-later
> + This program is free software: you can redistribute it and/or modify it under
> + the terms of the GNU Affero General Public License as published by the Free
> + Software Foundation, either version 3 of the License, or (at your option) any
> + later version.
> + .
> + This program is distributed in the hope that it will be useful, but WITHOUT
> + ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS
> + FOR A PARTICULAR PURPOSE. See the GNU Affero General Public License for more
> + details.
> + .
> + You should have received a copy of the GNU Affero General Public License along
> + with this program. If not, see <https://www.gnu.org/licenses/>.
> diff --git a/proxmox-rate-limiter/debian/debcargo.toml b/proxmox-rate-limiter/debian/debcargo.toml
> new file mode 100644
> index 00000000..b7864cdb
> --- /dev/null
> +++ b/proxmox-rate-limiter/debian/debcargo.toml
> @@ -0,0 +1,7 @@
> +overlay = "."
> +crate_src_path = ".."
> +maintainer = "Proxmox Support Team <support@proxmox.com>"
> +
> +[source]
> +vcs_git = "git://git.proxmox.com/git/proxmox.git"
> +vcs_browser = "https://git.proxmox.com/?p=proxmox.git"
> diff --git a/proxmox-rate-limiter/src/lib.rs b/proxmox-rate-limiter/src/lib.rs
> new file mode 100644
> index 00000000..a8d7cfdd
> --- /dev/null
> +++ b/proxmox-rate-limiter/src/lib.rs
> @@ -0,0 +1,13 @@
> +//! Token bucket based traffic rate limiter implementations.
> +
> +#![cfg_attr(docsrs, feature(doc_cfg, doc_auto_cfg))]
> +
> +#[cfg(feature = "rate-limiter")]
> +mod rate_limiter;
> +#[cfg(feature = "rate-limiter")]
> +pub use rate_limiter::{RateLimit, RateLimiter, RateLimiterVec, ShareableRateLimit};
> +
> +#[cfg(feature = "shared-rate-limiter")]
> +mod shared_rate_limiter;
> +#[cfg(feature = "shared-rate-limiter")]
> +pub use shared_rate_limiter::SharedRateLimiter;
> diff --git a/proxmox-rate-limiter/src/rate_limiter.rs b/proxmox-rate-limiter/src/rate_limiter.rs
> new file mode 100644
> index 00000000..945c77a6
> --- /dev/null
> +++ b/proxmox-rate-limiter/src/rate_limiter.rs
> @@ -0,0 +1,214 @@
> +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,

the UI doesn't let you set 0, but if one edits the config directly the
`... / rate` is a problem. The scenario is a little far fetched, but 
someone might quickly want to test something and just assume 0 means
unlimited and just set it to that instead of removing the line.

Given that it is not super easy to run into this `.min(1)` is probably
enough here. Not sure if checking for 0 in the endpoint when setting the
limit makes sense.

> +        )
> +    }
> +}
> +
> +/// 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))
> +    }
> +}
> diff --git a/proxmox-rate-limiter/src/shared_rate_limiter.rs b/proxmox-rate-limiter/src/shared_rate_limiter.rs
> new file mode 100644
> index 00000000..2822e7ea
> --- /dev/null
> +++ b/proxmox-rate-limiter/src/shared_rate_limiter.rs
> @@ -0,0 +1,130 @@
> +//! Rate limiter designed for shared memory
> +
> +use std::mem::MaybeUninit;
> +use std::path::Path;
> +use std::time::{Duration, Instant};
> +
> +use anyhow::{bail, Error};
> +use nix::sys::stat::Mode;
> +use nix::unistd::User;
> +
> +use proxmox_shared_memory::{check_subtype, initialize_subtype};
> +use proxmox_shared_memory::{Init, SharedMemory, SharedMutex};
> +use proxmox_sys::fs::{create_path, CreateOptions};
> +
> +use crate::{RateLimit, RateLimiter, ShareableRateLimit};
> +
> +/// Magic number for shared rate limiter exposed file mappings
> +///
> +/// Generated by `openssl::sha::sha256(b"Proxmox Backup SharedRateLimiter v1.0")[0..8];`
> +/// Original magic number kept when factored out from the initial
> +/// PBS implementation for full backwards compatibility.
> +pub const PROXMOX_SHARED_RATE_LIMITER_MAGIC_1_0: [u8; 8] = [6, 58, 213, 96, 161, 122, 130, 117];
> +
> +// Wrap RateLimiter, so that we can provide an Init impl
> +#[repr(C)]
> +struct WrapLimiter(RateLimiter);
> +
> +impl Init for WrapLimiter {
> +    fn initialize(this: &mut MaybeUninit<Self>) {
> +        // default does not matter here, because we override later
> +        this.write(WrapLimiter(RateLimiter::new(1_000_000, 1_000_000)));
> +    }
> +}
> +
> +#[repr(C)]
> +struct SharedRateLimiterData {
> +    magic: [u8; 8],
> +    tbf: SharedMutex<WrapLimiter>,
> +    padding: [u8; 4096 - 104],
> +}
> +
> +impl Init for SharedRateLimiterData {
> +    fn initialize(this: &mut MaybeUninit<Self>) {
> +        unsafe {
> +            let me = &mut *this.as_mut_ptr();
> +            me.magic = PROXMOX_SHARED_RATE_LIMITER_MAGIC_1_0;
> +            initialize_subtype(&mut me.tbf);
> +        }
> +    }
> +
> +    fn check_type_magic(this: &MaybeUninit<Self>) -> Result<(), Error> {
> +        unsafe {
> +            let me = &*this.as_ptr();
> +            if me.magic != PROXMOX_SHARED_RATE_LIMITER_MAGIC_1_0 {
> +                bail!("SharedRateLimiterData: wrong magic number");
> +            }
> +            check_subtype(&me.tbf)?;
> +            Ok(())
> +        }
> +    }
> +}
> +
> +/// Rate limiter designed for shared memory ([SharedMemory])
> +///
> +/// The actual [RateLimiter] is protected by a [SharedMutex] and
> +/// implements [Init]. This way we can share the limiter between
> +/// different processes.
> +pub struct SharedRateLimiter {
> +    shmem: SharedMemory<SharedRateLimiterData>,
> +}
> +
> +impl SharedRateLimiter {
> +    /// Creates a new mmap'ed instance.
> +    ///
> +    /// Data is mapped in `<base_path>/<name>` using
> +    /// `TMPFS`.
> +    pub fn mmap_shmem<P: AsRef<Path>>(
> +        name: &str,
> +        rate: u64,
> +        burst: u64,
> +        user: User,
> +        base_path: P,
> +    ) -> Result<Self, Error> {
> +        let mut path = base_path.as_ref().to_path_buf();
> +
> +        let dir_opts = CreateOptions::new()
> +            .perm(Mode::from_bits_truncate(0o770))
> +            .owner(user.uid)
> +            .group(user.gid);
> +
> +        create_path(&path, Some(dir_opts), Some(dir_opts))?;
> +
> +        path.push(name);
> +
> +        let file_opts = CreateOptions::new()
> +            .perm(Mode::from_bits_truncate(0o660))
> +            .owner(user.uid)
> +            .group(user.gid);
> +
> +        let shmem: SharedMemory<SharedRateLimiterData> = SharedMemory::open(&path, file_opts)?;
> +
> +        shmem.data().tbf.lock().0.update_rate(rate, burst);
> +
> +        Ok(Self { shmem })
> +    }
> +}
> +
> +impl ShareableRateLimit for SharedRateLimiter {
> +    fn update_rate(&self, rate: u64, bucket_size: u64) {
> +        self.shmem
> +            .data()
> +            .tbf
> +            .lock()
> +            .0
> +            .update_rate(rate, bucket_size);
> +    }
> +
> +    fn traffic(&self) -> u64 {
> +        self.shmem.data().tbf.lock().0.traffic()
> +    }
> +
> +    fn register_traffic(&self, current_time: Instant, data_len: u64) -> Duration {
> +        self.shmem
> +            .data()
> +            .tbf
> +            .lock()
> +            .0
> +            .register_traffic(current_time, data_len)
> +    }
> +}



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