Re: [PATCH v1 2/6] s3-client: add persistent shared request counters for client

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

On 2/11/26 1:12 PM, Robert Obkircher wrote:
> 
> On 2/9/26 10:14, Christian Ebner wrote:
>> Implements atomic counters for api requests successfully send to the
>> S3 API endpoint via the client, accounting for individual requests
>> discriminating based on their method.
>>
>> Since multiple client instances might exists accessing the API
>> concurrently, even in different processes, provide the atomic
>> counters via shared memory mapping. This follows along the lines of
>> the shared traffic limiter implementation.
>>
>> The counter mappings are conditionally constructed on client
>> instantiation by caller give configuration options.
>>
>> Keep sequential ordering of the counters with the intent to use them
>> for statistics, soft limits and/or notifications within Proxmox
>> Backup Server.
>>
>> Signed-off-by: Christian Ebner <c.ebner@proxmox.com>
>> ---
>>   proxmox-s3-client/Cargo.toml                  |   4 +
>>   proxmox-s3-client/debian/control              |   2 +
>>   proxmox-s3-client/examples/s3_client.rs       |   1 +
>>   proxmox-s3-client/src/client.rs               |  48 +++++-
>>   proxmox-s3-client/src/lib.rs                  |   7 +-
>>   .../src/shared_request_counters.rs            | 152 ++++++++++++++++++
>>   6 files changed, 212 insertions(+), 2 deletions(-)
>>   create mode 100644 proxmox-s3-client/src/shared_request_counters.rs
>>
>> diff --git a/proxmox-s3-client/Cargo.toml b/proxmox-s3-client/Cargo.toml
>> index a50fa715..1e31bca4 100644
>> --- a/proxmox-s3-client/Cargo.toml
>> +++ b/proxmox-s3-client/Cargo.toml
>> @@ -38,7 +38,9 @@ proxmox-base64 = { workspace = true, optional = true }
>>   proxmox-http = { workspace = true, features = [ "body", "client", "client-trait" ], optional = true }
>>   proxmox-human-byte.workspace = true
>>   proxmox-rate-limiter = { workspace = true, features = [ "rate-limiter", "shared-rate-limiter" ], optional = true }
>> +proxmox-shared-memory = { workspace = true, optional = true }
>>   proxmox-schema = { workspace = true, features = [ "api-macro", "api-types" ] }
>> +proxmox-sys = { workspace = true, optional = true }
>>   proxmox-serde.workspace = true
>>   proxmox-time = {workspace = true, optional = true }
>>   
>> @@ -65,6 +67,8 @@ impl = [
>>       "dep:proxmox-base64",
>>       "dep:proxmox-http",
>>       "dep:proxmox-rate-limiter",
>> +    "dep:proxmox-shared-memory",
>> +    "dep:proxmox-sys",
>>       "dep:proxmox-time",
>>   ]
>>   
>> diff --git a/proxmox-s3-client/debian/control b/proxmox-s3-client/debian/control
>> index 33418881..a534a107 100644
>> --- a/proxmox-s3-client/debian/control
>> +++ b/proxmox-s3-client/debian/control
>> @@ -85,6 +85,8 @@ Depends:
>>    librust-proxmox-rate-limiter-1+default-dev,
>>    librust-proxmox-rate-limiter-1+rate-limiter-dev,
>>    librust-proxmox-rate-limiter-1+shared-rate-limiter-dev,
>> + librust-proxmox-shared-memory-1+default-dev,
>> + librust-proxmox-sys-1+default-dev,
>>    librust-proxmox-time-2+default-dev (>= 2.1.0-~~),
>>    librust-quick-xml-0.36+async-tokio-dev (>= 0.36.1-~~),
>>    librust-quick-xml-0.36+default-dev (>= 0.36.1-~~),
>> diff --git a/proxmox-s3-client/examples/s3_client.rs b/proxmox-s3-client/examples/s3_client.rs
>> index ca69971c..329de47a 100644
>> --- a/proxmox-s3-client/examples/s3_client.rs
>> +++ b/proxmox-s3-client/examples/s3_client.rs
>> @@ -40,6 +40,7 @@ async fn run() -> Result<(), anyhow::Error> {
>>           put_rate_limit: None,
>>           provider_quirks: Vec::new(),
>>           rate_limiter_config: None,
>> +        request_counter_config: None,
>>       };
>>   
>>       // Creating a client instance and connect to api endpoint
>> diff --git a/proxmox-s3-client/src/client.rs b/proxmox-s3-client/src/client.rs
>> index 83176b39..4a6a702b 100644
>> --- a/proxmox-s3-client/src/client.rs
>> +++ b/proxmox-s3-client/src/client.rs
>> @@ -1,5 +1,6 @@
>>   use std::path::{Path, PathBuf};
>>   use std::str::FromStr;
>> +use std::sync::atomic::Ordering;
>>   use std::sync::{Arc, Mutex};
>>   use std::time::{Duration, Instant};
>>   
>> @@ -32,6 +33,7 @@ use crate::response_reader::{
>>       CopyObjectResponse, DeleteObjectsResponse, GetObjectResponse, HeadObjectResponse,
>>       ListBucketsResponse, ListObjectsV2Response, PutObjectResponse, ResponseReader,
>>   };
>> +use crate::shared_request_counters::SharedRequestCounters;
>>   
>>   /// Default timeout for s3 api requests.
>>   pub const S3_HTTP_REQUEST_TIMEOUT: Duration = Duration::from_secs(30 * 60);
>> @@ -74,6 +76,21 @@ pub struct S3RateLimiterConfig {
>>       burst_out: Option<u64>,
>>   }
>>   
>> +/// Options for the s3 client's shared request counters
>> +pub struct S3RequestCounterOptions {
>> +    /// ID for the memory mapped file
>> +    pub id: String,
>> +    /// Base path for the shared memory mapped file
>> +    pub base_path: PathBuf,
>> +    /// User for the to be created shared memory mapped file and folders
>> +    pub user: User,
>> +}
>> +
>> +/// Configuration  for the s3 client's shared request counters
>> +pub struct S3RequestCounterConfig {
>> +    options: S3RequestCounterOptions,
>> +}
>> +
>>   /// Configuration options for client
>>   pub struct S3ClientOptions {
>>       /// Endpoint to access S3 object store.
>> @@ -100,6 +117,8 @@ pub struct S3ClientOptions {
>>       pub provider_quirks: Vec<ProviderQuirks>,
>>       /// Configuration options for the shared rate limiter.
>>       pub rate_limiter_config: Option<S3RateLimiterConfig>,
>> +    /// Configuration options for the client's shared request counters.
>> +    pub request_counter_config: Option<S3RequestCounterConfig>,
>>   }
>>   
>>   impl S3ClientOptions {
>> @@ -110,6 +129,7 @@ impl S3ClientOptions {
>>           bucket: Option<String>,
>>           common_prefix: String,
>>           rate_limiter_options: Option<S3RateLimiterOptions>,
>> +        request_counter_options: Option<S3RequestCounterOptions>,
>>       ) -> Self {
>>           let rate_limiter_config = rate_limiter_options.map(|options| S3RateLimiterConfig {
>>               options,
>> @@ -118,6 +138,8 @@ impl S3ClientOptions {
>>               rate_out: config.rate_out.map(|human_bytes| human_bytes.as_u64()),
>>               burst_out: config.burst_out.map(|human_bytes| human_bytes.as_u64()),
>>           });
>> +        let request_counter_config =
>> +            request_counter_options.map(|options| S3RequestCounterConfig { options });
>>           Self {
>>               endpoint: config.endpoint,
>>               port: config.port,
>> @@ -131,6 +153,7 @@ impl S3ClientOptions {
>>               put_rate_limit: config.put_rate_limit,
>>               provider_quirks: config.provider_quirks.unwrap_or_default(),
>>               rate_limiter_config,
>> +            request_counter_config,
>>           }
>>       }
>>   }
>> @@ -141,6 +164,7 @@ pub struct S3Client {
>>       options: S3ClientOptions,
>>       authority: Authority,
>>       put_rate_limiter: Option<Arc<Mutex<RateLimiter>>>,
>> +    request_counters: Option<Arc<SharedRequestCounters>>,
>>   }
>>   
>>   impl S3Client {
>> @@ -213,6 +237,21 @@ impl S3Client {
>>               }
>>           }
>>   
>> +        let request_counters = if let Some(config) = options.request_counter_config.as_ref() {
>> +            let path = config
>> +                .options
>> +                .base_path
>> +                .join(format!("{}.shmem", config.options.id));
>> +            let request_counters = SharedRequestCounters::open_shared_memory_mapped(
>> +                &path,
>> +                config.options.user.clone(),
>> +            )
>> +            .context("failed to mmap shared S3 request counters")?;
>> +            Some(Arc::new(request_counters))
>> +        } else {
>> +            None
>> +        };
>> +
>>           let client = Client::builder(TokioExecutor::new()).build::<_, Body>(https_connector);
>>   
>>           let authority_template = if let Some(port) = options.port {
>> @@ -241,6 +280,7 @@ impl S3Client {
>>               options,
>>               authority,
>>               put_rate_limiter,
>> +            request_counters,
>>           })
>>       }
>>   
>> @@ -392,7 +432,13 @@ impl S3Client {
>>               };
>>   
>>               match response {
>> -                Ok(Ok(response)) => return Ok(response),
>> +                Ok(Ok(response)) => {
>> +                    if let Some(counters) = self.request_counters.as_ref() {
>> +                        let _prev = counters.increment(parts.method.clone(), Ordering::SeqCst);
>> +                    }
>> +
>> +                    return Ok(response);
>> +                }
>>                   Ok(Err(err)) => {
>>                       if retry >= MAX_S3_HTTP_REQUEST_RETRY - 1 {
>>                           return Err(err.into());
>> diff --git a/proxmox-s3-client/src/lib.rs b/proxmox-s3-client/src/lib.rs
>> index d02fd0dc..ceee41a2 100644
>> --- a/proxmox-s3-client/src/lib.rs
>> +++ b/proxmox-s3-client/src/lib.rs
>> @@ -21,7 +21,8 @@ pub use aws_sign_v4::uri_decode;
>>   mod client;
>>   #[cfg(feature = "impl")]
>>   pub use client::{
>> -    S3Client, S3ClientOptions, S3PathPrefix, S3RateLimiterOptions, S3_HTTP_REQUEST_TIMEOUT,
>> +    S3Client, S3ClientOptions, S3PathPrefix, S3RateLimiterOptions, S3RequestCounterOptions,
>> +    S3_HTTP_REQUEST_TIMEOUT,
>>   };
>>   #[cfg(feature = "impl")]
>>   mod timestamps;
>> @@ -33,3 +34,7 @@ mod object_key;
>>   pub use object_key::S3ObjectKey;
>>   #[cfg(feature = "impl")]
>>   mod response_reader;
>> +#[cfg(feature = "impl")]
>> +mod shared_request_counters;
>> +#[cfg(feature = "impl")]
>> +pub use shared_request_counters::SharedRequestCounters;
>> diff --git a/proxmox-s3-client/src/shared_request_counters.rs b/proxmox-s3-client/src/shared_request_counters.rs
>> new file mode 100644
>> index 00000000..b236490b
>> --- /dev/null
>> +++ b/proxmox-s3-client/src/shared_request_counters.rs
>> @@ -0,0 +1,152 @@
>> +use std::mem::MaybeUninit;
>> +use std::path::Path;
>> +use std::sync::atomic::{AtomicU64, Ordering};
>> +
>> +use anyhow::{bail, Error};
>> +use hyper::http::method::Method;
>> +use nix::sys::stat::Mode;
>> +use nix::unistd::User;
>> +
>> +use proxmox_shared_memory::{Init, SharedMemory};
>> +use proxmox_sys::fs::CreateOptions;
>> +
>> +const MEMORY_PAGE_SIZE: usize = 4096;
>> +/// Generated via openssl::sha::sha256(b"Proxmox shared request counters v1.0")[0..8]
>> +const PROXMOX_SHARED_REQUEST_COUNTERS_1_0: [u8; 8] = [224, 110, 88, 252, 26, 77, 180, 5];
>> +
>> +#[repr(C)]
>> +#[derive(Default)]
>> +struct RequestCounters {
>> +    delete: AtomicU64,
>> +    get: AtomicU64,
>> +    head: AtomicU64,
>> +    post: AtomicU64,
>> +    put: AtomicU64,
> It probably doesn't matter (especially not with SeqCst), but aligning
> each field to 64 bytes would avoid false sharing.

Can you elaborate a bit on this? Not sure if I understand what the 
benefits for this would be.

>> +}
>> +
>> +impl Init for RequestCounters {
>> +    fn initialize(this: &mut MaybeUninit<Self>) {
>> +        unsafe {
>> +            let this = &mut *this.as_mut_ptr();
>> +            *this = RequestCounters::default();
> If the memory is actually uninitialized this is undefined behavior
> because `*this =` drops the previous value. The memory from mmap is
> initialized, but I think it would still be better to use ptr::write.

Okay, can adapt this accordingly, thanks!

>> +        }
>> +    }
>> +}
>> +
>> +impl RequestCounters {
>> +    /// Increment the counter for given method, following the provided memory ordering constrains.
>> +    ///
>> +    /// Returns the previously stored value.
>> +    pub fn increment(&self, method: Method, ordering: Ordering) -> u64 {
>> +        match method {
>> +            Method::DELETE => self.delete.fetch_add(1, ordering),
>> +            Method::GET => self.get.fetch_add(1, ordering),
>> +            Method::HEAD => self.head.fetch_add(1, ordering),
>> +            Method::POST => self.post.fetch_add(1, ordering),
>> +            Method::PUT => self.put.fetch_add(1, ordering),
>> +            _ => 0,
>> +        }
>> +    }
>> +
>> +    /// Load current counter state for given method, following the provided memory ordering constrains
>> +    pub fn load(&self, method: Method, ordering: Ordering) -> u64 {
>> +        match method {
>> +            Method::DELETE => self.delete.load(ordering),
>> +            Method::GET => self.get.load(ordering),
>> +            Method::HEAD => self.head.load(ordering),
>> +            Method::POST => self.post.load(ordering),
>> +            Method::PUT => self.put.load(ordering),
>> +            _ => 0,
>> +        }
>> +    }
>> +
>> +    /// Reset all counters, following the provided memory ordering constrains
>> +    pub fn reset(&self, ordering: Ordering) {
>> +        self.delete.store(0, ordering);
>> +        self.get.store(0, ordering);
>> +        self.head.store(0, ordering);
>> +        self.post.store(0, ordering);
>> +        self.put.store(0, ordering);
>> +    }
>> +}
>> +
>> +#[repr(C)]
>> +struct MappableRequestCounters {
>> +    magic: [u8; 8],
>> +    counters: RequestCounters,
>> +    _page_size_padding: [u8; MEMORY_PAGE_SIZE
>> +        - PROXMOX_SHARED_REQUEST_COUNTERS_1_0.len()
>> +        - std::mem::size_of::<RequestCounters>()],
> Why do we need to hard-code a possibly incorrect assumption about the
> page size? I know `SharedMemory` claims that the size needs to be a
> multiple of 4096, but I don't understand why.

This padding was chosen exactly because of that doc comment and since 
other users of `SharedMemory` do the same 4k page size padding.

Will however have a look at the implementation details there, now that 
you sparked my interest, maybe that sheds some light.

Or maybe you already have further details since you are working on the 
patches to make the code compatible with kernels with non-default page 
sizes?

> I guess this implicitly prevents unused padding at the end of the
> struct which would be problematic if we add a field and run two
> processes with different versions at the same time, but that doesn't
> explain the page size.

Allowing to extend with further fields in the future has proven to be 
beneficial, yes. E.g. for the token shadow config version as mentioned 
in 
https://lore.proxmox.com/pbs-devel/20260121151408.731516-2-s.rufinatscha@proxmox.com/T/

>> +}
>> +
>> +impl Init for MappableRequestCounters {
>> +    fn initialize(this: &mut MaybeUninit<Self>) {
>> +        unsafe {
>> +            let this = &mut *this.as_mut_ptr();
>> +            this.magic = PROXMOX_SHARED_REQUEST_COUNTERS_1_0;
>> +            this.counters = RequestCounters::default();
>> +        }
>> +    }
>> +
>> +    fn check_type_magic(this: &MaybeUninit<Self>) -> Result<(), Error> {
>> +        unsafe {
>> +            let this = &*this.as_ptr();
>> +            if this.magic != PROXMOX_SHARED_REQUEST_COUNTERS_1_0 {
>> +                bail!("incorrect magic number for request counters detected");
>> +            }
>> +            proxmox_shared_memory::check_subtype(&this.counters)?;
>> +        }
>> +        Ok(())
>> +    }
>> +}
>> +
>> +/// Atomic counters storing per-request method counts for the client.
>> +///
>> +/// If set, the counts can be filtered based on a path prefix.
>> +pub struct SharedRequestCounters {
>> +    shared_memory: SharedMemory<MappableRequestCounters>,
>> +}
>> +
>> +impl SharedRequestCounters {
>> +    /// Create a new shared counter instance.
>> +    ///
>> +    /// Opens or creates mmap file and accesses it via shared memory mapping.
>> +    pub fn open_shared_memory_mapped<P: AsRef<Path>>(path: P, user: User) -> Result<Self, Error> {
>> +        let path = path.as_ref();
>> +        if let Some(parent) = path.parent() {
>> +            let dir_opts = CreateOptions::new()
>> +                .perm(Mode::from_bits_truncate(0o770))
>> +                .owner(user.uid)
>> +                .group(user.gid);
>> +
>> +            proxmox_sys::fs::create_path(parent, Some(dir_opts), Some(dir_opts))?;
>> +        }
>> +
>> +        let file_opts = CreateOptions::new()
>> +            .perm(Mode::from_bits_truncate(0o660))
>> +            .owner(user.uid)
>> +            .group(user.gid);
>> +        let shared_memory = SharedMemory::open(path, file_opts)?;
>> +        Ok(Self { shared_memory })
>> +    }
>> +
>> +    /// Increment the counter for given method, following the provided memory ordering constrains
>> +    ///
>> +    /// Returns the previously stored value.
>> +    pub fn increment(&self, method: Method, ordering: Ordering) -> u64 {
>> +        self.shared_memory
>> +            .data()
>> +            .counters
>> +            .increment(method, ordering)
>> +    }
>> +
>> +    /// Load current counter state for given method, following the provided memory ordering constrains
>> +    pub fn load(&self, method: Method, ordering: Ordering) -> u64 {
>> +        self.shared_memory.data().counters.load(method, ordering)
>> +    }
>> +
>> +    /// Reset all counters, following the provided memory ordering constrains
>> +    pub fn reset(&self, ordering: Ordering) {
>> +        self.shared_memory.data().counters.reset(ordering)
>> +    }
>> +}