[PATCH proxmox v7 01/34] s3-client: add persistent shared request counters for client

Christian Ebner
2026-04-01T13:49:12Z

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.

The counter mappings are conditionally constructed on client
instantiation by caller given configuration options.

Since multiple client instances might exist, accessing the API
concurrently and possibly from different processes, provide the
atomic counters via shared memory mapping. This follows along the
lines of the shared traffic limiter implementation. To reduce cache
line contention, atomic counters are aligned to half the standard
cache line size of 64-bytes.

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 | 37 ++++
proxmox-s3-client/src/lib.rs | 7 +-
.../src/shared_request_counters.rs | 205 ++++++++++++++++++
6 files changed, 255 insertions(+), 1 deletion(-)
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 e4392adb..40977728 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 7e3b325b..be36244d 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 df523c75..dadef2c6 100644
--- a/proxmox-s3-client/examples/s3_client.rs
+++ b/proxmox-s3-client/examples/s3_client.rs
@@ -41,6 +41,7 @@ async fn run() -> Result<(), anyhow::Error> {
provider_quirks: Vec::new(),
rate_limiter_config: None,
proxy_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 0943b1f5..f8aaa1af 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};

@@ -33,6 +34,7 @@ use crate::response_reader::{
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);
@@ -75,6 +77,16 @@ pub struct S3RateLimiterConfig {
burst_out: Option<u64>,
}

+/// Configuration for the s3 client's shared request counters
+pub struct S3RequestCounterConfig {
+ /// 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 options for client
pub struct S3ClientOptions {
/// Endpoint to access S3 object store.
@@ -103,6 +115,8 @@ pub struct S3ClientOptions {
pub rate_limiter_config: Option<S3RateLimiterConfig>,
/// Proxy configuration to be used by the client.
pub proxy_config: Option<ProxyConfig>,
+ /// Configuration options for the client's shared request counters.
+ pub request_counter_config: Option<S3RequestCounterConfig>,
}

impl S3ClientOptions {
@@ -114,6 +128,7 @@ impl S3ClientOptions {
common_prefix: String,
rate_limiter_options: Option<S3RateLimiterOptions>,
proxy_config: Option<ProxyConfig>,
+ request_counter_config: Option<S3RequestCounterConfig>,
) -> Self {
let rate_limiter_config = rate_limiter_options.map(|options| S3RateLimiterConfig {
options,
@@ -136,6 +151,7 @@ impl S3ClientOptions {
provider_quirks: config.provider_quirks.unwrap_or_default(),
rate_limiter_config,
proxy_config,
+ request_counter_config,
}
}
}
@@ -146,6 +162,7 @@ pub struct S3Client {
options: S3ClientOptions,
authority: Authority,
put_rate_limiter: Option<Arc<Mutex<RateLimiter>>>,
+ request_counters: Option<Arc<SharedRequestCounters>>,
}

impl S3Client {
@@ -222,6 +239,20 @@ impl S3Client {
https_connector.set_proxy(proxy_config.clone());
}

+ let request_counters = if let Some(config) = options.request_counter_config.as_ref() {
+ let path = config
+ .base_path
+ .join(format!("{}.shmem", config.id));
+ let request_counters = SharedRequestCounters::open_shared_memory_mapped(
+ &path,
+ config.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 {
@@ -250,6 +281,7 @@ impl S3Client {
options,
authority,
put_rate_limiter,
+ request_counters,
})
}

@@ -408,6 +440,11 @@ impl S3Client {
};

if !do_retry || retry >= MAX_S3_HTTP_REQUEST_RETRY - 1 {
+ if let Some(counters) = self.request_counters.as_ref()
+ && response.is_ok()
+ {
+ let _prev = counters.increment(parts.method.clone(), Ordering::AcqRel);
+ }
return Ok(response?);
}

diff --git a/proxmox-s3-client/src/lib.rs b/proxmox-s3-client/src/lib.rs
index 5c8c4b08..41905e34 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, S3RequestCounterConfig,
+ S3_HTTP_REQUEST_TIMEOUT,
};
#[cfg(feature = "impl")]
mod timestamps;
@@ -35,3 +36,7 @@ pub use object_key::S3ObjectKey;
mod response_reader;
#[cfg(feature = "impl")]
pub use response_reader::DeleteObjectError;
+#[cfg(feature = "impl")]
+mod shared_request_counters;
+#[cfg(feature = "impl")]
+pub use shared_request_counters::{RequestCounterValues, 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..e6820a04
--- /dev/null
+++ b/proxmox-s3-client/src/shared_request_counters.rs
@@ -0,0 +1,205 @@
+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, align(32))]
+#[derive(Default)]
+/// AtomicU64 aligned to the half default cache line size of 64-bytes.
+struct AlignedAtomic(AtomicU64);
+
+#[repr(C, align(32))]
+#[derive(Default, PartialEq)]
+/// Mmapped file magic number aligned to half the default cache line size of 64-bytes.
+/// Facilitates the padding size calculation.
+struct AlignedMagic([u8; 8]);
+
+#[repr(C)]
+#[derive(Default)]
+// Ordering is chosen to bundle frequently expected counter updates with less
+// fequent ones. Ideally each counter would live in it's own cache line, but
+// that requires double the memory.
+struct RequestCounters {
+ // request count
+ get: AlignedAtomic,
+ delete: AlignedAtomic,
+ put: AlignedAtomic,
+ head: AlignedAtomic,
+ post: AlignedAtomic,
+}
+
+impl Init for RequestCounters {
+ fn initialize(this: &mut MaybeUninit<Self>) {
+ // safety: RequestCounters contains simple data types with no internal references.
+ this.write(RequestCounters::default());
+ }
+}
+
+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.0.fetch_add(1, ordering),
+ Method::GET => self.get.0.fetch_add(1, ordering),
+ Method::HEAD => self.head.0.fetch_add(1, ordering),
+ Method::POST => self.post.0.fetch_add(1, ordering),
+ Method::PUT => self.put.0.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.0.load(ordering),
+ Method::GET => self.get.0.load(ordering),
+ Method::HEAD => self.head.0.load(ordering),
+ Method::POST => self.post.0.load(ordering),
+ Method::PUT => self.put.0.load(ordering),
+ _ => 0,
+ }
+ }
+
+ /// Reset all counters, following the provided memory ordering constrains
+ ///
+ /// Returns the respective counter values before reset.
+ pub fn reset(&self, ordering: Ordering) -> RequestCounterValues {
+ RequestCounterValues {
+ delete: self.delete.0.swap(0, ordering),
+ get: self.get.0.swap(0, ordering),
+ head: self.head.0.swap(0, ordering),
+ post: self.post.0.swap(0, ordering),
+ put: self.put.0.swap(0, ordering),
+ }
+ }
+}
+
+/// Size of the padding to align the mmapped request counters to 4k default
+/// page size.
+const PADDING_SIZE: usize =
+ MEMORY_PAGE_SIZE - std::mem::size_of::<AlignedMagic>() - std::mem::size_of::<RequestCounters>();
+
+#[repr(C)]
+// Alignment is chosen to reduce cache line contention while keeping low
+// memory footprint.
+struct MappableRequestCounters {
+ magic: AlignedMagic,
+ counters: RequestCounters,
+ _page_size_padding: [u8; PADDING_SIZE],
+}
+
+impl Default for MappableRequestCounters {
+ fn default() -> Self {
+ Self {
+ magic: AlignedMagic(PROXMOX_SHARED_REQUEST_COUNTERS_1_0),
+ counters: RequestCounters::default(),
+ _page_size_padding: [0; PADDING_SIZE],
+ }
+ }
+}
+
+impl Init for MappableRequestCounters {
+ fn initialize(this: &mut MaybeUninit<Self>) {
+ // safety: MappableRequestCounters contains simple data types with no internal references.
+ this.write(MappableRequestCounters::default());
+ }
+
+ fn check_type_magic(this: &MaybeUninit<Self>) -> Result<(), Error> {
+ unsafe {
+ // safety: do not make assumptions about the object being initialized,
+ // use raw pointer offsets to check memory for expected contents.
+ let this_ptr = this.as_ptr();
+
+ let magic_ptr = std::ptr::addr_of!((*this_ptr).magic);
+ if *magic_ptr != AlignedMagic(PROXMOX_SHARED_REQUEST_COUNTERS_1_0) {
+ bail!("incorrect magic number for request counters detected");
+ }
+
+ let counters_ptr = std::ptr::addr_of!((*this_ptr).counters);
+ proxmox_shared_memory::check_subtype(&*counters_ptr)?;
+ }
+ 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_non_tmpfs(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
+ ///
+ /// Returns the respective counter values before reset.
+ pub fn reset(&self, ordering: Ordering) -> RequestCounterValues {
+ self.shared_memory.data().counters.reset(ordering)
+ }
+}
+
+/// Current value of the individual request counters.
+///
+/// The values for the different fields are not guaranteed to be synchronized.
+pub struct RequestCounterValues {
+ /// number of GET requests
+ pub get: u64,
+ /// number of DELETE requests
+ pub delete: u64,
+ /// number of PUT requests
+ pub put: u64,
+ /// number of HEAD requests
+ pub head: u64,
+ /// number of POST requests
+ pub post: u64,
+}
--
2.47.3