From 4c4dcb6f5e663bb25436a9527a28925206d9e42f Mon Sep 17 00:00:00 2001 From: Zhengchao An Date: Mon, 7 Sep 2026 12:13:30 +0800 Subject: [PATCH] fix(storage): queue multipart parts for foreground write permits (#7337) Multipart parts shared the 250 ms direct-PutObject wait on the foreground write permit pool, so SDK-default concurrency (many parts per upload in flight at once) was rejected wholesale with SlowDown at stock settings. Keep the pool that bounds in-flight bodies, but let parts wait in a bounded queue with their own timeout before body ingest, report the queue depth in the ForegroundWrite admission snapshot, and document the foreground write admission environment variables. --- crates/config/README.md | 31 +++ crates/config/src/constants/object.rs | 25 +- .../workload-admission-contracts.md | 2 +- rustfs/src/storage/concurrency/manager.rs | 261 ++++++++++++++++-- 4 files changed, 299 insertions(+), 20 deletions(-) diff --git a/crates/config/README.md b/crates/config/README.md index b5db390c6..74a25376b 100644 --- a/crates/config/README.md +++ b/crates/config/README.md @@ -130,6 +130,37 @@ Scanner cycle budget controls: - timeout returns S3 `SlowDown`, so clients should use normal SDK retry handling. - this is not a fdatasync or group-commit switch. Track fdatasync batching separately with `rustfs_s3_put_object_rename_fdatasync_batch_files`. +## Foreground write admission environment variables + +Large direct `PutObject` requests and multipart `UploadPart` requests share one +per-process permit pool that bounds how many bodies are ingested and written +concurrently. Small direct PUTs stay on the legacy path. + +- `RUSTFS_PUT_LARGE_FOREGROUND_ADMISSION_ENABLE` + - enables the default-on pool; `false` keeps only the soft request counter. + - default is `true`. +- `RUSTFS_PUT_LARGE_FOREGROUND_ADMISSION_LIMIT` + - permits in the pool; `0` derives half of `RUSTFS_OBJECT_MAX_CONCURRENT_DISK_READS`, clamped to `32`. + - default is `0` (32 permits at stock settings). +- `RUSTFS_PUT_LARGE_FOREGROUND_ADMISSION_MIN_SIZE_BYTES` + - smallest direct `PutObject` that takes a permit; unknown-size requests always do. + - default is `33554432` (32 MiB). +- `RUSTFS_PUT_LARGE_FOREGROUND_ADMISSION_WAIT_TIMEOUT_MS` + - how long a direct `PutObject` waits for a permit before returning S3 `SlowDown`. + - default is `250`. +- `RUSTFS_PUT_MULTIPART_FOREGROUND_ADMISSION_MIN_SIZE_BYTES` + - smallest `UploadPart` that takes a permit; `0` gates every part. + - default is `0`. +- `RUSTFS_PUT_MULTIPART_FOREGROUND_ADMISSION_WAIT_TIMEOUT_MS` + - how long an `UploadPart` waits in the bounded queue for a permit before returning S3 `SlowDown`; `0` rejects immediately when the pool is full. + - default is `30000`. Parts wait before body ingest, so SDK-default clients that send every part of an upload concurrently drain through the pool instead of failing. +- `RUSTFS_PUT_MULTIPART_FOREGROUND_ADMISSION_MAX_PENDING` + - maximum `UploadPart` requests waiting for a permit at once; parts beyond it return `SlowDown` without waiting. + - default is `0`, which derives 16 times the permit limit (512 at stock settings). +- `RUSTFS_PUT_FOREGROUND_ADMISSION_ENABLE`, `RUSTFS_PUT_FOREGROUND_ADMISSION_LIMIT`, `RUSTFS_PUT_FOREGROUND_ADMISSION_WAIT_TIMEOUT_MS` + - experimental strict gate that applies to every foreground write regardless of size and replaces the pool above when enabled. + - default is disabled; enabling it with limit `0` disables foreground write admission entirely. + ## Remote tier timeout environment variables - `RUSTFS_TIER_REMOTE_CONNECT_TIMEOUT_SECS` diff --git a/crates/config/src/constants/object.rs b/crates/config/src/constants/object.rs index 7ec459afd..ac4a585b7 100644 --- a/crates/config/src/constants/object.rs +++ b/crates/config/src/constants/object.rs @@ -365,13 +365,36 @@ pub const ENV_PUT_MULTIPART_FOREGROUND_ADMISSION_MIN_SIZE_BYTES: &str = "RUSTFS_PUT_MULTIPART_FOREGROUND_ADMISSION_MIN_SIZE_BYTES"; pub const DEFAULT_PUT_MULTIPART_FOREGROUND_ADMISSION_MIN_SIZE_BYTES: usize = 0; -/// Time in milliseconds an automatic foreground write waits for a permit. +/// Time in milliseconds an automatic foreground direct PutObject waits for a permit. /// /// A short wait smooths transient bursts while still returning S3 /// `SlowDown`/503 before body ingest when the node is already saturated. pub const ENV_PUT_LARGE_FOREGROUND_ADMISSION_WAIT_TIMEOUT_MS: &str = "RUSTFS_PUT_LARGE_FOREGROUND_ADMISSION_WAIT_TIMEOUT_MS"; pub const DEFAULT_PUT_LARGE_FOREGROUND_ADMISSION_WAIT_TIMEOUT_MS: u64 = 250; +/// Time in milliseconds a multipart UploadPart waits for a foreground write permit. +/// +/// SDK-default multipart clients send every part of an upload concurrently, so +/// a single node routinely sees several times more parts in flight than the +/// permit pool allows. Those parts have not ingested a body yet, so queueing +/// them costs a connection rather than memory or internode streams; the pool +/// still bounds the number of parts being written. The wait is long enough for +/// an ordinary queue to drain on modest hardware, and a part that cannot get a +/// permit within it fails with S3 `SlowDown`/503 for the client to retry. +/// `0` rejects immediately when the pool is full. +pub const ENV_PUT_MULTIPART_FOREGROUND_ADMISSION_WAIT_TIMEOUT_MS: &str = + "RUSTFS_PUT_MULTIPART_FOREGROUND_ADMISSION_WAIT_TIMEOUT_MS"; +pub const DEFAULT_PUT_MULTIPART_FOREGROUND_ADMISSION_WAIT_TIMEOUT_MS: u64 = 30_000; + +/// Maximum multipart UploadPart requests waiting for a foreground write permit per process. +/// +/// Parts beyond this queue depth are rejected with S3 `SlowDown`/503 without +/// waiting, so a genuinely saturated node still fails fast instead of holding +/// an unbounded set of connections open for the whole wait timeout. +/// `0` derives the depth from the permit limit. +pub const ENV_PUT_MULTIPART_FOREGROUND_ADMISSION_MAX_PENDING: &str = "RUSTFS_PUT_MULTIPART_FOREGROUND_ADMISSION_MAX_PENDING"; +pub const DEFAULT_PUT_MULTIPART_FOREGROUND_ADMISSION_MAX_PENDING: usize = 0; + const _: () = assert!(DEFAULT_PUT_LARGE_FOREGROUND_ADMISSION_ENABLE); /// Environment variable for minimum GetObject timeout in seconds. diff --git a/docs/architecture/workload-admission-contracts.md b/docs/architecture/workload-admission-contracts.md index 3d3295e00..cb007425e 100644 --- a/docs/architecture/workload-admission-contracts.md +++ b/docs/architecture/workload-admission-contracts.md @@ -12,7 +12,7 @@ | Class | Provider (`impl WorkloadAdmissionSnapshotProvider`) | `active` / `queued` / `limit` source | Reports `Unknown` when | |---|---|---|---| | `ForegroundRead` | `ConcurrencyManager` in `rustfs/src/storage/concurrency/manager.rs` (source of truth); re-exposed unchanged by the RustFS runtime provider | disk-read permits in use / `None` (the semaphore exposes no waiter count) / configured max concurrent disk reads | the storage registry has no entry | -| `ForegroundWrite` | `ConcurrencyManager` in `rustfs/src/storage/concurrency/manager.rs` (source of truth); re-exposed unchanged by the RustFS runtime provider | foreground-write permits in use or legacy active-write counter / `None` / configured or derived write-admission limit | the storage registry has no entry | +| `ForegroundWrite` | `ConcurrencyManager` in `rustfs/src/storage/concurrency/manager.rs` (source of truth); re-exposed unchanged by the RustFS runtime provider | foreground-write permits in use or legacy active-write counter / multipart parts waiting in the bounded admission queue (`None` for the strict and legacy policies) / configured or derived write-admission limit | the storage registry has no entry | | `Metadata` | `RustFsWorkloadAdmissionSnapshotProvider` in `rustfs/src/workload_admission.rs` | `Open` once the bucket metadata runtime handle exists; no counts | bucket metadata runtime not initialized | | `Scanner` | same | scanner active work-unit counter / none / configured set-scan limit when nonzero | scanner runtime not initialized | | `Repair` | same | heal active tasks / heal queue length / `None` (limits live behind the async heal manager state) | heal manager not initialized | diff --git a/rustfs/src/storage/concurrency/manager.rs b/rustfs/src/storage/concurrency/manager.rs index 68d450d65..d625b3b06 100644 --- a/rustfs/src/storage/concurrency/manager.rs +++ b/rustfs/src/storage/concurrency/manager.rs @@ -29,12 +29,17 @@ use rustfs_io_core::BytesPool; use rustfs_io_core::io_profile::{AccessPattern, IoPatternDetector, StorageMedia, detect_storage_media}; use rustfs_io_metrics::bandwidth::{BandwidthMonitor, BandwidthSnapshot}; use rustfs_io_metrics::{MetricsCollector, PerformanceMetrics}; +use std::sync::atomic::{AtomicUsize, Ordering}; use std::sync::{Arc, LazyLock, Mutex}; use std::time::Duration; use tokio::sync::{OwnedSemaphorePermit, Semaphore}; use tracing::debug; const DERIVED_LARGE_PUT_ADMISSION_LIMIT_MAX: usize = 32; +// A queued multipart part holds a connection but no body, so the queue can be +// several times deeper than the permit pool. Sixteen uploads sending sixteen +// parts each through one node fits inside the derived depth of 32 * 16. +const DERIVED_MULTIPART_ADMISSION_MAX_PENDING_FACTOR: usize = 16; // Framed S2 alone can retain one encoded and one decoded block of roughly // 4 MiB each, while other codecs have their own larger windows. Four keeps // useful request parallelism without scaling codec memory and CPU with clients. @@ -149,6 +154,28 @@ struct ForegroundWriteAdmissionGate { semaphore: Arc, limit: usize, wait_timeout: Duration, + /// Requests currently waiting in the bounded multipart queue. + pending: Arc, +} + +/// Reservation of one slot in the bounded multipart wait queue; released on +/// drop so a cancelled or timed-out waiter never leaks queue depth. +struct PendingSlot(Arc); + +impl PendingSlot { + fn reserve(pending: &Arc, max_pending: usize) -> Option { + if pending.fetch_add(1, Ordering::AcqRel) >= max_pending { + pending.fetch_sub(1, Ordering::AcqRel); + return None; + } + Some(Self(pending.clone())) + } +} + +impl Drop for PendingSlot { + fn drop(&mut self) { + self.0.fetch_sub(1, Ordering::AcqRel); + } } impl ForegroundWriteAdmissionGate { @@ -157,6 +184,7 @@ impl ForegroundWriteAdmissionGate { semaphore: Arc::new(Semaphore::new(limit)), limit, wait_timeout, + pending: Arc::new(AtomicUsize::new(0)), } } @@ -164,6 +192,35 @@ impl ForegroundWriteAdmissionGate { self.limit.saturating_sub(self.semaphore.available_permits()) } + fn pending(&self) -> usize { + self.pending.load(Ordering::Acquire) + } + + /// Admit through the same permit pool as [`Self::admit`], but let the + /// request wait in a bounded queue for `wait_timeout` instead of failing + /// on the gate's own short wait. A full queue rejects immediately. + async fn admit_queued( + &self, + wait_timeout: Duration, + max_pending: usize, + ) -> Result { + match self.semaphore.clone().try_acquire_owned() { + Ok(permit) => return Ok(ForegroundWriteAdmission::Admitted(permit)), + Err(tokio::sync::TryAcquireError::Closed) => return Ok(ForegroundWriteAdmission::Rejected), + Err(tokio::sync::TryAcquireError::NoPermits) => {} + } + if wait_timeout.is_zero() { + return Ok(ForegroundWriteAdmission::Rejected); + } + let Some(_slot) = PendingSlot::reserve(&self.pending, max_pending) else { + return Ok(ForegroundWriteAdmission::Rejected); + }; + match tokio::time::timeout(wait_timeout, self.semaphore.clone().acquire_owned()).await { + Ok(permit) => Ok(ForegroundWriteAdmission::Admitted(permit?)), + Err(_) => Ok(ForegroundWriteAdmission::Rejected), + } + } + async fn admit(&self) -> Result { if self.wait_timeout.is_zero() { return Ok(match self.semaphore.clone().try_acquire_owned() { @@ -194,6 +251,8 @@ enum ForegroundWriteAdmissionPolicy { gate: ForegroundWriteAdmissionGate, put_object_min_size_bytes: usize, multipart_part_min_size_bytes: usize, + multipart_wait_timeout: Duration, + multipart_max_pending: usize, }, } @@ -252,11 +311,24 @@ impl ForegroundWriteAdmissionPolicy { rustfs_config::ENV_PUT_LARGE_FOREGROUND_ADMISSION_WAIT_TIMEOUT_MS, rustfs_config::DEFAULT_PUT_LARGE_FOREGROUND_ADMISSION_WAIT_TIMEOUT_MS, )); + let multipart_wait_timeout = Duration::from_millis(rustfs_utils::get_env_u64( + rustfs_config::ENV_PUT_MULTIPART_FOREGROUND_ADMISSION_WAIT_TIMEOUT_MS, + rustfs_config::DEFAULT_PUT_MULTIPART_FOREGROUND_ADMISSION_WAIT_TIMEOUT_MS, + )); + let multipart_max_pending = derive_multipart_admission_max_pending( + rustfs_utils::get_env_usize( + rustfs_config::ENV_PUT_MULTIPART_FOREGROUND_ADMISSION_MAX_PENDING, + rustfs_config::DEFAULT_PUT_MULTIPART_FOREGROUND_ADMISSION_MAX_PENDING, + ), + large_limit, + ); Self::Large { gate: ForegroundWriteAdmissionGate::new(large_limit, wait_timeout), put_object_min_size_bytes, multipart_part_min_size_bytes, + multipart_wait_timeout, + multipart_max_pending, } } @@ -275,11 +347,25 @@ impl ForegroundWriteAdmissionPolicy { #[cfg(test)] fn large_for_test(enabled: bool, limit: usize, min_size_bytes: usize, wait_timeout: Duration) -> Self { + Self::large_with_multipart_queue_for_test(enabled, limit, min_size_bytes, wait_timeout, wait_timeout, 0) + } + + #[cfg(test)] + fn large_with_multipart_queue_for_test( + enabled: bool, + limit: usize, + min_size_bytes: usize, + wait_timeout: Duration, + multipart_wait_timeout: Duration, + multipart_max_pending: usize, + ) -> Self { if enabled && limit > 0 { Self::Large { gate: ForegroundWriteAdmissionGate::new(limit, wait_timeout), put_object_min_size_bytes: min_size_bytes, multipart_part_min_size_bytes: 0, + multipart_wait_timeout, + multipart_max_pending: derive_multipart_admission_max_pending(multipart_max_pending, limit), } } else { Self::LegacyCounterOnly @@ -298,35 +384,45 @@ impl ForegroundWriteAdmissionPolicy { gate, put_object_min_size_bytes, multipart_part_min_size_bytes, - } => { - let min_size_bytes = match kind { - ForegroundWriteAdmissionKind::PutObject => *put_object_min_size_bytes, - ForegroundWriteAdmissionKind::MultipartPart => *multipart_part_min_size_bytes, - }; - if should_gate_foreground_write(size, min_size_bytes) { + multipart_wait_timeout, + multipart_max_pending, + } => match kind { + ForegroundWriteAdmissionKind::PutObject if should_gate_foreground_write(size, *put_object_min_size_bytes) => { gate.admit().await - } else { - Ok(ForegroundWriteAdmission::Disabled) } - } + ForegroundWriteAdmissionKind::MultipartPart + if should_gate_foreground_write(size, *multipart_part_min_size_bytes) => + { + gate.admit_queued(*multipart_wait_timeout, *multipart_max_pending).await + } + _ => Ok(ForegroundWriteAdmission::Disabled), + }, } } fn snapshot(&self, legacy_limit: usize) -> WorkloadAdmissionSnapshot { match self { - Self::Disabled => put_admission_snapshot(0, 0, None), - Self::LegacyCounterOnly => put_admission_snapshot(PutObjectGuard::concurrent_count(), legacy_limit, None), + Self::Disabled => put_admission_snapshot(0, None, 0, None), + Self::LegacyCounterOnly => put_admission_snapshot(PutObjectGuard::concurrent_count(), None, legacy_limit, None), Self::Strict(gate) => { - put_admission_snapshot(gate.active(), gate.limit, Some("foreground write admission permits exhausted")) - } - Self::Large { gate, .. } => { - put_admission_snapshot(gate.active(), gate.limit, Some("large foreground write admission permits exhausted")) + put_admission_snapshot(gate.active(), None, gate.limit, Some("foreground write admission permits exhausted")) } + Self::Large { gate, .. } => put_admission_snapshot( + gate.active(), + Some(gate.pending()), + gate.limit, + Some("large foreground write admission permits exhausted"), + ), } } } -fn put_admission_snapshot(active: usize, limit: usize, hard_gate_reason: Option<&'static str>) -> WorkloadAdmissionSnapshot { +fn put_admission_snapshot( + active: usize, + queued: Option, + limit: usize, + hard_gate_reason: Option<&'static str>, +) -> WorkloadAdmissionSnapshot { let state = if limit == 0 { AdmissionState::Disabled } else if active >= limit { @@ -336,7 +432,7 @@ fn put_admission_snapshot(active: usize, limit: usize, hard_gate_reason: Option< }; let admission = - WorkloadAdmissionSnapshot::new(WorkloadClass::ForegroundWrite, state).with_counts(Some(active), None, Some(limit)); + WorkloadAdmissionSnapshot::new(WorkloadClass::ForegroundWrite, state).with_counts(Some(active), queued, Some(limit)); match state { AdmissionState::Disabled => admission.with_reason("foreground write admission disabled"), @@ -347,6 +443,13 @@ fn put_admission_snapshot(active: usize, limit: usize, hard_gate_reason: Option< } } +fn derive_multipart_admission_max_pending(configured_max_pending: usize, limit: usize) -> usize { + if configured_max_pending > 0 { + return configured_max_pending; + } + limit.saturating_mul(DERIVED_MULTIPART_ADMISSION_MAX_PENDING_FACTOR) +} + fn derive_large_put_admission_limit(configured_limit: usize, max_disk_reads: usize) -> usize { if configured_limit > 0 { return configured_limit; @@ -464,6 +567,24 @@ impl ConcurrencyManager { manager } + #[cfg(test)] + pub(crate) fn with_multipart_admission_queue_for_test( + limit: usize, + multipart_wait_timeout: Duration, + multipart_max_pending: usize, + ) -> Self { + let mut manager = Self::new(); + manager.foreground_write_admission_policy = ForegroundWriteAdmissionPolicy::large_with_multipart_queue_for_test( + true, + limit, + rustfs_config::DEFAULT_PUT_LARGE_FOREGROUND_ADMISSION_MIN_SIZE_BYTES, + Duration::ZERO, + multipart_wait_timeout, + multipart_max_pending, + ); + manager + } + #[cfg(test)] pub(crate) fn with_large_put_admission_for_test( enabled: bool, @@ -1108,7 +1229,7 @@ mod integration_tests { use super::super::request_guard::GetObjectGuard; use super::{ ConcurrencyManager, ForegroundWriteAdmission, SNOWBALL_ARCHIVE_DECODER_LIMIT, SNOWBALL_MEMBER_COMMIT_LIMIT, - SNOWBALL_STAGING_BYTES_LIMIT, derive_large_put_admission_limit, + SNOWBALL_STAGING_BYTES_LIMIT, derive_large_put_admission_limit, derive_multipart_admission_max_pending, }; use crate::storage::storage_api::concurrency_consumer::PutObjectGuard; use rustfs_concurrency::{AdmissionState, WorkloadAdmissionSnapshotProvider, WorkloadClass}; @@ -1524,6 +1645,110 @@ mod integration_tests { drop((first, second)); } + #[tokio::test(start_paused = true)] + #[serial] + async fn test_concurrency_manager_multipart_part_waits_for_released_permit() { + let manager = ConcurrencyManager::with_multipart_admission_queue_for_test(1, Duration::from_secs(30), 0); + let held = manager + .admit_multipart_part(8 * 1024 * 1024) + .await + .expect("first multipart part admission should acquire"); + assert!(matches!(held, ForegroundWriteAdmission::Admitted(_))); + + let waiter_manager = manager.clone(); + let waiter = tokio::spawn(async move { waiter_manager.admit_multipart_part(8 * 1024 * 1024).await }); + tokio::task::yield_now().await; + assert_eq!(manager.put_object_admission_snapshot().queued, Some(1)); + + tokio::time::advance(Duration::from_secs(5)).await; + drop(held); + + let admission = waiter + .await + .expect("multipart admission waiter task must not panic") + .expect("multipart admission gate must stay open"); + assert!(matches!(admission, ForegroundWriteAdmission::Admitted(_))); + assert_eq!(manager.put_object_admission_snapshot().queued, Some(0)); + } + + #[tokio::test(start_paused = true)] + #[serial] + async fn test_concurrency_manager_multipart_part_rejects_after_queue_wait_timeout() { + let manager = ConcurrencyManager::with_multipart_admission_queue_for_test(1, Duration::from_secs(30), 0); + let held = manager + .admit_multipart_part(1024) + .await + .expect("first multipart part admission should acquire"); + + let waiter_manager = manager.clone(); + let waiter = tokio::spawn(async move { waiter_manager.admit_multipart_part(1024).await }); + tokio::task::yield_now().await; + tokio::time::advance(Duration::from_secs(30)).await; + + let admission = waiter + .await + .expect("multipart admission waiter task must not panic") + .expect("multipart admission gate must stay open"); + assert!(matches!(admission, ForegroundWriteAdmission::Rejected)); + assert_eq!(manager.put_object_admission_snapshot().queued, Some(0)); + drop(held); + } + + #[tokio::test(start_paused = true)] + #[serial] + async fn test_concurrency_manager_multipart_part_rejects_immediately_when_queue_is_full() { + let manager = ConcurrencyManager::with_multipart_admission_queue_for_test(1, Duration::from_secs(30), 1); + let held = manager + .admit_multipart_part(1024) + .await + .expect("first multipart part admission should acquire"); + + let waiter_manager = manager.clone(); + let queued = tokio::spawn(async move { waiter_manager.admit_multipart_part(1024).await }); + tokio::task::yield_now().await; + assert_eq!(manager.put_object_admission_snapshot().queued, Some(1)); + + let overflow = manager + .admit_multipart_part(1024) + .await + .expect("full multipart queue should reject, not close"); + assert!(matches!(overflow, ForegroundWriteAdmission::Rejected)); + + drop(held); + let admission = queued + .await + .expect("queued multipart part task must not panic") + .expect("multipart admission gate must stay open"); + assert!(matches!(admission, ForegroundWriteAdmission::Admitted(_))); + } + + #[tokio::test(start_paused = true)] + #[serial] + async fn test_concurrency_manager_cancelled_multipart_waiter_releases_queue_slot() { + let manager = ConcurrencyManager::with_multipart_admission_queue_for_test(1, Duration::from_secs(30), 1); + let held = manager + .admit_multipart_part(1024) + .await + .expect("first multipart part admission should acquire"); + + let waiter_manager = manager.clone(); + let queued = tokio::spawn(async move { waiter_manager.admit_multipart_part(1024).await }); + tokio::task::yield_now().await; + assert_eq!(manager.put_object_admission_snapshot().queued, Some(1)); + queued.abort(); + let _ = queued.await; + + assert_eq!(manager.put_object_admission_snapshot().queued, Some(0)); + drop(held); + } + + #[test] + fn test_concurrency_manager_derives_multipart_admission_max_pending_from_limit() { + assert_eq!(derive_multipart_admission_max_pending(7, 32), 7); + assert_eq!(derive_multipart_admission_max_pending(0, 32), 512); + assert_eq!(derive_multipart_admission_max_pending(0, 1), 16); + } + #[test] fn test_concurrency_manager_derives_large_put_admission_limit_from_scheduler_cap() { assert_eq!(derive_large_put_admission_limit(7, 64), 7);