diff --git a/crates/config/src/constants/object.rs b/crates/config/src/constants/object.rs index d4527fad6..28222c824 100644 --- a/crates/config/src/constants/object.rs +++ b/crates/config/src/constants/object.rs @@ -73,15 +73,36 @@ pub const DEFAULT_OBJECT_MAX_CONCURRENT_DISK_READS: usize = 64; /// - Example: `export RUSTFS_OBJECT_DISK_PERMIT_WAIT_TIMEOUT=5` pub const ENV_OBJECT_DISK_PERMIT_WAIT_TIMEOUT: &str = "RUSTFS_OBJECT_DISK_PERMIT_WAIT_TIMEOUT"; -/// Maximum time a GET request waits for a disk read permit (seconds). +/// Maximum time a GET request waits for a primary disk read permit (seconds). /// /// Permits are held for the whole response body transfer, so slow clients can /// occupy all of them while the disks sit idle. Instead of stalling until the -/// request-level timeout fires, a GET that waits longer than this proceeds -/// without a permit (degraded pass-through) and the bypass is counted in -/// metrics/logs. Set to 0 to wait indefinitely (previous behavior). +/// request-level timeout fires, a GET that waits longer than this falls through +/// to a bounded degraded admission lane; if that lane is also full the request +/// is rejected with `SlowDown`/503 rather than proceeding without any permit. +/// Set to 0 to wait on the primary lane indefinitely (never degrade or reject). pub const DEFAULT_OBJECT_DISK_PERMIT_WAIT_TIMEOUT: u64 = 5; +/// Environment variable for the bounded degraded disk-read admission lane size. +/// - Purpose: Cap how many GETs may proceed after the primary disk-read permit +/// pool is saturated, giving a hard upper bound on concurrent disk-active +/// reads (primary cap + degraded cap) instead of an unbounded pass-through. +/// - Unit: request count (usize). `0` means "mirror the primary cap", so the +/// absolute hard cap defaults to twice the primary disk-read cap. +/// - Example: `export RUSTFS_OBJECT_DISK_DEGRADED_READ_CAP=16` +pub const ENV_OBJECT_DISK_DEGRADED_READ_CAP: &str = "RUSTFS_OBJECT_DISK_DEGRADED_READ_CAP"; + +/// Size of the bounded degraded disk-read admission lane. +/// +/// When the primary disk-read permit pool is saturated and a GET exceeds +/// [`DEFAULT_OBJECT_DISK_PERMIT_WAIT_TIMEOUT`], it may take one permit from this +/// bounded overflow lane instead of reading without any admission token. The +/// total number of GETs performing disk-active reads is therefore hard-capped at +/// `primary_cap + degraded_cap`; beyond that a GET is rejected with `SlowDown`. +/// The default `0` mirrors the primary cap, so the hard cap is twice the primary +/// disk-read concurrency. +pub const DEFAULT_OBJECT_DISK_DEGRADED_READ_CAP: usize = 0; + /// Skip bitrot hash verification on GetObject reads. /// /// When enabled, GetObject reads skip the per-shard hash diff --git a/rustfs/src/app/object_usecase.rs b/rustfs/src/app/object_usecase.rs index 4d44218d4..0e55fba5a 100644 --- a/rustfs/src/app/object_usecase.rs +++ b/rustfs/src/app/object_usecase.rs @@ -53,8 +53,8 @@ use super::storage_api::object_usecase::bucket::{ }; use super::storage_api::object_usecase::compression::{MIN_DISK_COMPRESSIBLE_SIZE, is_disk_compressible}; use super::storage_api::object_usecase::concurrency::{ - self, ConcurrencyManager, GetObjectGuard, PutObjectGuard, get_concurrency_aware_buffer_size, get_concurrency_manager, - get_put_concurrency_aware_buffer_size, + self, ConcurrencyManager, DiskReadAdmission, GetObjectGuard, PutObjectGuard, get_concurrency_aware_buffer_size, + get_concurrency_manager, get_put_concurrency_aware_buffer_size, }; #[cfg(test)] use super::storage_api::object_usecase::contract::http::HTTPPreconditions; @@ -2871,29 +2871,44 @@ impl DefaultObjectUsecase { ) -> S3Result { let permit_wait_start = std::time::Instant::now(); let permit_wait_timeout = Self::disk_permit_wait_timeout(); - // Permits are held for the whole body transfer, so slow clients can - // pin all of them while disks are idle. Bound the wait and degrade to - // a permit-less read instead of stalling into the request timeout. - let disk_permit = if permit_wait_timeout.is_zero() { - Some( - manager - .acquire_owned_disk_read_permit() - .await - .map_err(|_| s3_error!(InternalError, "disk read semaphore closed"))?, - ) - } else { - match tokio::time::timeout(permit_wait_timeout, manager.acquire_owned_disk_read_permit()).await { - Ok(permit) => Some(permit.map_err(|_| s3_error!(InternalError, "disk read semaphore closed"))?), - Err(_) => { - metrics::counter!("rustfs.get_object.disk_permit.bypass.total").increment(1); - warn!( - bucket = %bucket, - key = %key, - wait_ms = permit_wait_start.elapsed().as_millis() as u64, - "GetObject proceeding without disk read permit after bounded wait" - ); - None - } + // Permits are held for the whole body transfer, so slow clients can pin + // all of them while disks are idle. Bound the wait on the primary pool + // and, on timeout, admit from a bounded degraded overflow lane. Total + // concurrent disk-active GETs are hard-capped at + // `primary_cap + degraded_cap`; once that cap is reached we reject with + // `SlowDown` instead of reading without any admission token. Never + // proceed permit-less. + let disk_permit = match manager + .admit_disk_read(permit_wait_timeout) + .await + .map_err(|_| s3_error!(InternalError, "disk read semaphore closed"))? + { + DiskReadAdmission::Primary(permit) => Some(permit), + // Throttling disabled by config (primary cap 0): proceed without an + // admission token. Not a saturation bypass. + DiskReadAdmission::Unbounded => None, + DiskReadAdmission::Degraded(permit) => { + metrics::counter!("rustfs.get_object.disk_permit.degraded.total").increment(1); + warn!( + bucket = %bucket, + key = %key, + wait_ms = permit_wait_start.elapsed().as_millis() as u64, + "GetObject admitted into bounded degraded disk-read lane after primary pool saturation" + ); + Some(permit) + } + DiskReadAdmission::Rejected => { + metrics::counter!("rustfs.get_object.disk_permit.hard_reject.total").increment(1); + warn!( + bucket = %bucket, + key = %key, + wait_ms = permit_wait_start.elapsed().as_millis() as u64, + "GetObject rejected: disk-read hard concurrency cap reached" + ); + return Err(s3_error!( + SlowDown, + "disk read concurrency limit reached, please reduce your request rate" + )); } }; let permit_wait_duration = permit_wait_start.elapsed(); diff --git a/rustfs/src/app/storage_api.rs b/rustfs/src/app/storage_api.rs index b43b8442b..22e581340 100644 --- a/rustfs/src/app/storage_api.rs +++ b/rustfs/src/app/storage_api.rs @@ -825,8 +825,8 @@ pub(crate) mod bucket { pub(crate) mod concurrency { pub(crate) use crate::storage::storage_api::concurrency_consumer::{ - ConcurrencyManager, GetObjectGuard, IoQueueStatus, IoStrategy, PutObjectGuard, get_concurrency_aware_buffer_size, - get_concurrency_manager, get_put_concurrency_aware_buffer_size, + ConcurrencyManager, DiskReadAdmission, GetObjectGuard, IoQueueStatus, IoStrategy, PutObjectGuard, + get_concurrency_aware_buffer_size, get_concurrency_manager, get_put_concurrency_aware_buffer_size, }; } diff --git a/rustfs/src/storage/concurrency/manager.rs b/rustfs/src/storage/concurrency/manager.rs index aabe85268..3794e76d6 100644 --- a/rustfs/src/storage/concurrency/manager.rs +++ b/rustfs/src/storage/concurrency/manager.rs @@ -41,6 +41,12 @@ pub(crate) static CONCURRENCY_MANAGER: LazyLock = LazyLock:: pub struct ConcurrencyManager { /// Semaphore to limit concurrent disk reads disk_read_semaphore: Arc, + /// Bounded overflow lane for GETs that time out waiting on the primary + /// disk-read permit pool. Admitting from this lane instead of reading + /// without any permit gives a hard upper bound on concurrent disk-active + /// reads (`primary cap + degraded cap`); when it is also full a GET is + /// rejected with `SlowDown` rather than proceeding unbounded. + degraded_read_semaphore: Arc, /// I/O load metrics for adaptive strategy calculation io_metrics: Arc>, /// I/O priority queue for request scheduling @@ -87,6 +93,27 @@ impl std::fmt::Debug for ConcurrencyManager { .finish() } } +/// Outcome of [`ConcurrencyManager::admit_disk_read`]. +/// +/// `Primary`/`Degraded` both carry an owned permit that must be held for the +/// full body transfer; `Rejected` means the hard concurrency cap was reached and +/// the caller must fail the GET with `SlowDown`/503 instead of reading without a +/// permit. +#[derive(Debug)] +pub enum DiskReadAdmission { + /// Admitted from the primary disk-read permit pool. + Primary(tokio::sync::OwnedSemaphorePermit), + /// Admitted from the bounded degraded overflow lane after the primary pool + /// stayed saturated past the configured wait. + Degraded(tokio::sync::OwnedSemaphorePermit), + /// Disk-read throttling is disabled (primary cap configured to `0`): the GET + /// proceeds without an admission token. This is the only permit-less path + /// and is an explicit operator opt-out, not a saturation bypass. + Unbounded, + /// Hard concurrency cap reached; the caller must reject with `SlowDown`. + Rejected, +} + #[allow(dead_code)] impl ConcurrencyManager { /// Create a new concurrency manager with default settings @@ -99,6 +126,17 @@ impl ConcurrencyManager { let max_disk_reads = scheduler_config.max_concurrent_reads; + // Bounded degraded admission lane. A configured `0` mirrors the primary + // cap, so the absolute hard cap on concurrent disk-active reads defaults + // to twice the primary disk-read concurrency. + let degraded_read_cap = { + let configured = rustfs_utils::get_env_usize( + rustfs_config::ENV_OBJECT_DISK_DEGRADED_READ_CAP, + rustfs_config::DEFAULT_OBJECT_DISK_DEGRADED_READ_CAP, + ); + if configured == 0 { max_disk_reads } else { configured } + }; + // Detect storage media let storage_media = detect_storage_media(scheduler_config.storage_detection_enabled, &scheduler_config.storage_media_override); @@ -129,6 +167,7 @@ impl ConcurrencyManager { Self { disk_read_semaphore: Arc::new(Semaphore::new(max_disk_reads)), + degraded_read_semaphore: Arc::new(Semaphore::new(degraded_read_cap)), io_metrics: Arc::new(Mutex::new(IoLoadMetrics::new(scheduler_config.load_sample_window))), priority_queue: Arc::new(IoPriorityQueue::new(queue_config)), bytes_pool: Arc::new(BytesPool::new_tiered()), @@ -140,6 +179,20 @@ impl ConcurrencyManager { } } + /// Build a manager with explicit disk-read admission caps for tests. + /// + /// Overrides only the primary/degraded semaphores and the snapshot cap so + /// admission behavior can be exercised deterministically under a paused + /// virtual clock, without depending on process-wide environment variables. + #[cfg(test)] + pub(crate) fn with_disk_read_caps_for_test(primary: usize, degraded: usize) -> Self { + let mut manager = Self::new(); + manager.disk_read_semaphore = Arc::new(Semaphore::new(primary)); + manager.degraded_read_semaphore = Arc::new(Semaphore::new(degraded)); + manager.scheduler_config.max_concurrent_reads = primary; + manager + } + /// Track a GetObject request pub fn track_request() -> GetObjectGuard { GetObjectGuard::new() @@ -177,6 +230,54 @@ impl ConcurrencyManager { self.disk_read_semaphore.clone().acquire_owned().await } + /// Admit a GET disk read under a hard concurrency cap. + /// + /// The permit is held for the whole response body transfer, so this is the + /// single admission boundary that bounds concurrent disk-active reads: + /// + /// - `primary_wait == 0` waits on the primary permit pool indefinitely and + /// always yields a [`DiskReadAdmission::Primary`] permit (no degrade, no + /// reject); this preserves the "wait forever" opt-out. + /// - Otherwise it waits up to `primary_wait` for a primary permit. On + /// timeout it takes one permit from the bounded degraded overflow lane + /// without blocking. If that lane is also full the request is + /// [`DiskReadAdmission::Rejected`] — it must surface as `SlowDown`/503 + /// rather than reading without any admission token. + /// + /// Total GETs holding a permit therefore never exceed + /// `primary_cap + degraded_cap`. + pub async fn admit_disk_read(&self, primary_wait: Duration) -> Result { + // Primary cap of 0 means disk-read throttling is disabled (matches the + // `AdmissionState::Disabled` snapshot semantics). Serve without an + // admission token rather than rejecting every GET; this preserves the + // pre-existing "permits disabled" behavior and is the only permit-less + // path. + if self.scheduler_config.max_concurrent_reads == 0 { + return Ok(DiskReadAdmission::Unbounded); + } + + if primary_wait.is_zero() { + let permit = self.disk_read_semaphore.clone().acquire_owned().await?; + return Ok(DiskReadAdmission::Primary(permit)); + } + + match tokio::time::timeout(primary_wait, self.disk_read_semaphore.clone().acquire_owned()).await { + Ok(permit) => Ok(DiskReadAdmission::Primary(permit?)), + Err(_) => { + // Primary pool saturated within the wait window. Do not read + // without a permit; fall through to the bounded degraded lane, + // and reject once the hard cap is reached. + match self.degraded_read_semaphore.clone().try_acquire_owned() { + Ok(permit) => Ok(DiskReadAdmission::Degraded(permit)), + // NoPermits => hard cap reached; Closed never happens for a + // semaphore we never close. Either way, reject rather than + // bypass admission. + Err(_) => Ok(DiskReadAdmission::Rejected), + } + } + } + } + // ============================================ // Adaptive I/O Strategy Methods // ============================================ @@ -1035,4 +1136,151 @@ mod integration_tests { assert_eq!(small_file_strategy.priority, IoPriority::High); assert_eq!(large_file_strategy.priority, IoPriority::Low); } + + // ============================================ + // Bounded disk-read admission (backlog#1317) + // ============================================ + + use super::DiskReadAdmission; + use std::sync::atomic::{AtomicUsize, Ordering as AtomicOrdering}; + + /// Primary pool saturation degrades into the bounded lane, and the hard cap + /// rejects with `Rejected` instead of admitting a permit-less read. Dropping + /// a permit frees the token so the next admission succeeds again. + #[tokio::test(start_paused = true)] + async fn test_admit_disk_read_degrades_then_hard_rejects() { + let manager = ConcurrencyManager::with_disk_read_caps_for_test(1, 1); + + // Primary permit granted immediately. + let primary = match manager.admit_disk_read(Duration::from_millis(100)).await.unwrap() { + DiskReadAdmission::Primary(permit) => permit, + other => panic!("expected primary admission, got {other:?}"), + }; + + // Primary pool saturated: next admission waits out the primary timeout + // and falls through to the bounded degraded lane. + let degraded = match manager.admit_disk_read(Duration::from_millis(100)).await.unwrap() { + DiskReadAdmission::Degraded(permit) => permit, + other => panic!("expected degraded admission, got {other:?}"), + }; + + // Both lanes full: hard cap reached, must reject rather than bypass. + match manager.admit_disk_read(Duration::from_millis(100)).await.unwrap() { + DiskReadAdmission::Rejected => {} + other => panic!("expected hard rejection, got {other:?}"), + } + + // Releasing the degraded permit re-opens exactly one degraded slot. + drop(degraded); + match manager.admit_disk_read(Duration::from_millis(100)).await.unwrap() { + DiskReadAdmission::Degraded(_permit) => {} + other => panic!("expected degraded admission after release, got {other:?}"), + } + + // Releasing the primary permit re-opens the primary lane immediately + // (no timeout wait), proving EOF/drop returns the token. + drop(primary); + match manager.admit_disk_read(Duration::from_millis(100)).await.unwrap() { + DiskReadAdmission::Primary(_permit) => {} + other => panic!("expected primary admission after release, got {other:?}"), + } + } + + /// With max primary = 1 and degraded = 1, 100 concurrent GETs never hold + /// more than `1 + 1` admission tokens simultaneously. Reverting the hard cap + /// (unbounded bypass) would let this exceed 2 and fail the test. + #[tokio::test(start_paused = true)] + async fn test_admit_disk_read_hard_caps_concurrent_admissions() { + let manager = std::sync::Arc::new(ConcurrencyManager::with_disk_read_caps_for_test(1, 1)); + let in_flight = std::sync::Arc::new(AtomicUsize::new(0)); + let max_in_flight = std::sync::Arc::new(AtomicUsize::new(0)); + let admitted = std::sync::Arc::new(AtomicUsize::new(0)); + let rejected = std::sync::Arc::new(AtomicUsize::new(0)); + + let mut handles = Vec::with_capacity(100); + for _ in 0..100 { + let manager = manager.clone(); + let in_flight = in_flight.clone(); + let max_in_flight = max_in_flight.clone(); + let admitted = admitted.clone(); + let rejected = rejected.clone(); + handles.push(tokio::spawn(async move { + match manager.admit_disk_read(Duration::from_millis(50)).await.unwrap() { + DiskReadAdmission::Primary(permit) | DiskReadAdmission::Degraded(permit) => { + let current = in_flight.fetch_add(1, AtomicOrdering::SeqCst) + 1; + max_in_flight.fetch_max(current, AtomicOrdering::SeqCst); + admitted.fetch_add(1, AtomicOrdering::SeqCst); + // Hold the admission token across an await point, as the + // real body transfer does, then release it. + tokio::time::sleep(Duration::from_millis(10)).await; + in_flight.fetch_sub(1, AtomicOrdering::SeqCst); + drop(permit); + } + DiskReadAdmission::Rejected => { + rejected.fetch_add(1, AtomicOrdering::SeqCst); + } + DiskReadAdmission::Unbounded => { + unreachable!("throttling is enabled (primary cap = 1) in this test"); + } + } + })); + } + + for handle in handles { + handle.await.unwrap(); + } + + // The invariant the hard cap guarantees: never more than + // primary + degraded = 2 GETs admitted at once. + assert!( + max_in_flight.load(AtomicOrdering::SeqCst) <= 2, + "max concurrent admissions {} exceeded the hard cap of 2", + max_in_flight.load(AtomicOrdering::SeqCst) + ); + // Every request is accounted for: admitted or explicitly rejected, never + // a permit-less bypass. + assert_eq!(admitted.load(AtomicOrdering::SeqCst) + rejected.load(AtomicOrdering::SeqCst), 100); + assert!( + rejected.load(AtomicOrdering::SeqCst) > 0, + "expected some hard rejections under saturation" + ); + } + + /// A primary cap of 0 (throttling disabled) serves every GET without an + /// admission token instead of rejecting them, preserving the pre-existing + /// "disk read permits disabled" behavior. + #[tokio::test(start_paused = true)] + async fn test_admit_disk_read_disabled_cap_is_unbounded() { + let manager = ConcurrencyManager::with_disk_read_caps_for_test(0, 0); + for _ in 0..10 { + match manager.admit_disk_read(Duration::from_millis(50)).await.unwrap() { + DiskReadAdmission::Unbounded => {} + other => panic!("expected unbounded admission when throttling disabled, got {other:?}"), + } + } + } + + /// `primary_wait == 0` preserves the wait-forever opt-out: it always yields + /// a primary permit and never degrades or rejects. + #[tokio::test] + async fn test_admit_disk_read_zero_wait_waits_on_primary() { + let manager = ConcurrencyManager::with_disk_read_caps_for_test(1, 1); + let held = match manager.admit_disk_read(Duration::ZERO).await.unwrap() { + DiskReadAdmission::Primary(permit) => permit, + other => panic!("expected primary admission, got {other:?}"), + }; + + // A second zero-wait admission blocks on the primary pool rather than + // degrading; it completes only once the first permit is released. + let manager2 = manager.clone(); + let waiter = tokio::spawn(async move { manager2.admit_disk_read(Duration::ZERO).await.unwrap() }); + tokio::task::yield_now().await; + assert!(!waiter.is_finished(), "zero-wait admission must block on the primary lane"); + + drop(held); + match waiter.await.unwrap() { + DiskReadAdmission::Primary(_permit) => {} + other => panic!("expected primary admission after release, got {other:?}"), + } + } } diff --git a/rustfs/src/storage/concurrency/mod.rs b/rustfs/src/storage/concurrency/mod.rs index ff8e7d57c..110e14cf6 100644 --- a/rustfs/src/storage/concurrency/mod.rs +++ b/rustfs/src/storage/concurrency/mod.rs @@ -54,7 +54,7 @@ pub use io_schedule::{ pub use request_guard::{GetObjectGuard, PutObjectGuard}; // Concurrency manager -pub use manager::ConcurrencyManager; +pub use manager::{ConcurrencyManager, DiskReadAdmission}; // ============================================ // New Module Re-exports (for gradual migration) diff --git a/rustfs/src/storage/storage_api.rs b/rustfs/src/storage/storage_api.rs index d573af5fd..d1afb3b68 100644 --- a/rustfs/src/storage/storage_api.rs +++ b/rustfs/src/storage/storage_api.rs @@ -114,8 +114,8 @@ pub(crate) mod access_consumer { pub(crate) mod concurrency_consumer { pub(crate) use super::super::concurrency::{ - ConcurrencyManager, GetObjectGuard, IoQueueStatus, IoStrategy, PutObjectGuard, get_concurrency_aware_buffer_size, - get_concurrency_manager, get_put_concurrency_aware_buffer_size, + ConcurrencyManager, DiskReadAdmission, GetObjectGuard, IoQueueStatus, IoStrategy, PutObjectGuard, + get_concurrency_aware_buffer_size, get_concurrency_manager, get_put_concurrency_aware_buffer_size, }; }