diff --git a/crates/concurrency/src/backpressure.rs b/crates/concurrency/src/backpressure.rs index 502d4fb0f..e692d7329 100644 --- a/crates/concurrency/src/backpressure.rs +++ b/crates/concurrency/src/backpressure.rs @@ -14,9 +14,8 @@ //! Shared backpressure policy type. //! -//! The runtime backpressure implementation (byte-watermark pipes and -//! monitors) lives in `rustfs/src/storage/backpressure.rs`; this module only -//! carries the watermark policy type that implementation shares. +//! This module only carries the watermark policy; the admission primitive it +//! projects into lives in `rustfs-io-core`. use rustfs_io_core::BackpressureConfig as CoreBackpressureConfig; diff --git a/docs/architecture/global-state-inventory.md b/docs/architecture/global-state-inventory.md index 8407ad864..b110374dd 100644 --- a/docs/architecture/global-state-inventory.md +++ b/docs/architecture/global-state-inventory.md @@ -105,7 +105,6 @@ inventory. Generic function-local names such as `CACHE`, `LOCK`, `INIT`, and | `USE_STARSHARD_CACHE`, `BUCKET_CACHE_SMALL`, `BUCKET_CACHE_LARGE` | `rustfs/src/storage/ecfs_extend.rs` | Cache or constant / owner-local cache | Bucket validation cache backend selection and cache storage stay private to the ECFS extension owner. | | `GLOBAL_SSE_DEK_PROVIDER`, `SSE_TEST_LOCK` | `rustfs/src/storage/sse.rs` | Owner-local cache / test state | SSE DEK provider cache and test serialization lock stay private to the SSE owner. | | `AUTH_FS` | `rustfs/src/storage/access.rs` | Cache or constant / owner-local cache | Authorization tag-condition lookup keeps its filesystem helper private to the access owner. | -| `LOCK_STATS` | `rustfs/src/storage/lock_optimizer.rs` | Process-global owner-local metrics | Lock optimization statistics stay private behind lock optimizer helper APIs. | | `DEADLOCK_DETECTOR` | `rustfs/src/storage/deadlock_detector.rs` | Process-global owner-local state | Deadlock detector lifecycle state stays private to the storage deadlock detector owner. | | `CONCURRENCY_MANAGER`, `ACTIVE_GET_REQUESTS`, `ACTIVE_PUT_REQUESTS` | `rustfs/src/storage/concurrency/*` | Process-global owner-local scheduler state | Storage concurrency manager and request counters remain inside the storage concurrency owner boundary. | | `GET_OBJECT_BUFFER_THRESHOLD_WARNED`, `GET_READER_STREAM_BUFFER_SIZE_OVERRIDE`, function-local `ENABLED`, `OBJECT_SEEK_SUPPORT_THRESHOLD`, `OBJECT_SEEK_SUPPORT_CONCURRENCY_THRESHOLDS` | `rustfs/src/app/object_usecase.rs` | Cache or constant / owner-local cache | Object GET/seek tuning caches and warning guards stay private to object usecase helpers. | diff --git a/rustfs/src/storage/backpressure.rs b/rustfs/src/storage/backpressure.rs deleted file mode 100644 index 9585ba8a8..000000000 --- a/rustfs/src/storage/backpressure.rs +++ /dev/null @@ -1,618 +0,0 @@ -// Copyright 2024 RustFS Team -// -// Licensed under the Apache License, Version 2.0 (the "License"); -// you may not use this file except in compliance with the License. -// You may obtain a copy of the License at -// -// http://www.apache.org/licenses/LICENSE-2.0 -// -// Unless required by applicable law or agreed to in writing, software -// distributed under the License is distributed on an "AS IS" BASIS, -// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -// See the License for the specific language governing permissions and -// limitations under the License. - -//! Backpressure Management for Object Data Transfer. -//! -//! This module provides backpressure-aware pipes for object data transfer, -//! preventing buffer overflow and memory exhaustion under high concurrency. - -//! # Key Features -//! -//! - Configurable buffer size with high/low watermarks -//! - Backpressure state monitoring and events -//! - Backpressure metrics emitted through the shared metrics pipeline -//! - Graceful handling of slow consumers -//! -//! # Architecture -//! -//! ```text -//! [Disk Reader] --> [BackpressurePipe] --> [HTTP Response] -//! | -//! v -//! [Buffer Monitor] -//! | -//! v -//! [High Watermark?] --> Apply Backpressure -//! ``` - -use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering}; -use std::time::{Duration, Instant}; -use tokio::io::{DuplexStream, duplex}; -use tracing::{debug, warn}; - -use metrics::counter; -use rustfs_concurrency::PipeBackpressurePolicy; -use rustfs_io_core::BackpressureConfig as CoreBackpressureConfig; - -/// Object-transfer duplex pipe backpressure policy. -#[derive(Debug, Clone, Copy)] -pub struct ObjectPipeBackpressurePolicy { - /// Buffer size in bytes (default 4MB). - pub buffer_size: usize, - /// High watermark percentage (default 80%). - /// When buffer usage exceeds this, backpressure is applied. - pub high_watermark: u32, - /// Low watermark percentage (default 50%). - /// When buffer usage drops below this after high watermark, backpressure is released. - pub low_watermark: u32, -} - -impl Default for ObjectPipeBackpressurePolicy { - fn default() -> Self { - Self { - buffer_size: rustfs_config::DEFAULT_OBJECT_DUPLEX_BUFFER_SIZE, - high_watermark: rustfs_config::DEFAULT_OBJECT_BACKPRESSURE_HIGH_WATERMARK, - low_watermark: rustfs_config::DEFAULT_OBJECT_BACKPRESSURE_LOW_WATERMARK, - } - } -} - -impl ObjectPipeBackpressurePolicy { - /// Load configuration from environment variables. - pub fn from_env() -> Self { - let buffer_size = rustfs_utils::get_env_usize( - rustfs_config::ENV_OBJECT_DUPLEX_BUFFER_SIZE, - rustfs_config::DEFAULT_OBJECT_DUPLEX_BUFFER_SIZE, - ); - let high_watermark = rustfs_utils::get_env_u32( - rustfs_config::ENV_OBJECT_BACKPRESSURE_HIGH_WATERMARK, - rustfs_config::DEFAULT_OBJECT_BACKPRESSURE_HIGH_WATERMARK, - ); - let low_watermark = rustfs_utils::get_env_u32( - rustfs_config::ENV_OBJECT_BACKPRESSURE_LOW_WATERMARK, - rustfs_config::DEFAULT_OBJECT_BACKPRESSURE_LOW_WATERMARK, - ); - - Self { - buffer_size, - high_watermark, - low_watermark, - } - } - - /// Calculate high watermark threshold in bytes. - pub fn high_watermark_bytes(&self) -> usize { - (self.buffer_size as u64 * self.high_watermark as u64 / 100) as usize - } - - /// Calculate low watermark threshold in bytes. - pub fn low_watermark_bytes(&self) -> usize { - (self.buffer_size as u64 * self.low_watermark as u64 / 100) as usize - } - - /// Project this object-transfer policy into the shared concurrency facade policy. - pub fn to_concurrency_policy(&self) -> PipeBackpressurePolicy { - PipeBackpressurePolicy { - buffer_size: self.buffer_size, - high_watermark: self.high_watermark, - low_watermark: self.low_watermark, - } - } - - /// Project this object-transfer policy into the reusable io-core admission config. - pub fn to_core_config(&self) -> CoreBackpressureConfig { - self.to_concurrency_policy().to_core_config() - } -} - -/// Backpressure state. -#[derive(Debug, Clone, Copy, PartialEq, Eq)] -pub enum BackpressureState { - /// Normal operation, buffer usage is below high watermark. - Normal, - /// Buffer usage is above high watermark, backpressure should be applied. - HighWatermark, - /// Backpressure is actively being applied to the producer. - BackpressureApplied, -} - -impl std::fmt::Display for BackpressureState { - fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { - match self { - BackpressureState::Normal => write!(f, "normal"), - BackpressureState::HighWatermark => write!(f, "high_watermark"), - BackpressureState::BackpressureApplied => write!(f, "backpressure_applied"), - } - } -} - -/// Compact metadata snapshot for object-transfer backpressure pipes. -#[derive(Debug, Clone, Copy, PartialEq, Eq)] -pub struct BackpressurePipeMeta { - /// Buffer capacity in bytes. - pub buffer_capacity: usize, - /// Current backpressure state. - pub state: BackpressureState, - /// Age of the pipe since creation. - pub age: Duration, -} - -/// Compact metadata snapshot for the lightweight backpressure monitor. -#[derive(Debug, Clone, Copy, PartialEq)] -pub struct BackpressureMonitorMeta { - /// Buffer capacity in bytes. - pub buffer_capacity: usize, - /// Current buffer usage percentage. - pub usage_percent: f32, - /// Current backpressure state. - pub state: BackpressureState, -} - -fn calculate_usage_percent(usage: usize, capacity: usize) -> f32 { - if capacity > 0 { - (usage as f32 / capacity as f32) * 100.0 - } else { - 0.0 - } -} - -fn apply_watermark_transition( - in_high_watermark: &AtomicBool, - usage: usize, - high: usize, - low: usize, -) -> (BackpressureState, bool) { - let current = in_high_watermark.load(Ordering::Acquire); - let next_state = if usage >= high { - BackpressureState::HighWatermark - } else if usage <= low { - BackpressureState::Normal - } else if current { - BackpressureState::HighWatermark - } else { - BackpressureState::Normal - }; - let next_is_high = matches!(next_state, BackpressureState::HighWatermark); - let changed = in_high_watermark.swap(next_is_high, Ordering::AcqRel) != next_is_high; - (next_state, changed) -} - -fn saturating_sub_atomic(value: &AtomicUsize, delta: usize) { - value - .fetch_update(Ordering::AcqRel, Ordering::Acquire, |current| Some(current.saturating_sub(delta))) - .ok(); -} - -/// A backpressure-aware pipe wrapping tokio's duplex. -/// -/// This provides monitoring and events for backpressure conditions -/// while maintaining compatibility with the standard duplex interface. -pub struct BackpressurePipe { - /// Reader end of the duplex pipe. - reader: DuplexStream, - /// Writer end of the duplex pipe. - writer: DuplexStream, - /// Configuration. - config: ObjectPipeBackpressurePolicy, - /// Current buffer usage (approximate, updated on write). - buffer_usage: AtomicUsize, - /// Current backpressure state. - state: AtomicBool, // true = in high watermark state - /// Total bytes written. - total_written: AtomicUsize, - /// Total bytes read. - total_read: AtomicUsize, - /// Cached high watermark threshold in bytes. - high_watermark_bytes: usize, - /// Cached low watermark threshold in bytes. - low_watermark_bytes: usize, - /// Pipe creation timestamp. - created_at: Instant, -} - -impl BackpressurePipe { - /// Create a new backpressure-aware pipe with default configuration. - pub fn new() -> Self { - Self::with_config(ObjectPipeBackpressurePolicy::from_env()) - } - - /// Create a new backpressure-aware pipe with custom configuration. - pub fn with_config(config: ObjectPipeBackpressurePolicy) -> Self { - let policy = config.to_concurrency_policy(); - let (reader, writer) = duplex(policy.buffer_size); - let high_watermark_bytes = policy.high_watermark_bytes(); - let low_watermark_bytes = policy.low_watermark_bytes(); - - debug!( - buffer_size = config.buffer_size, - high_watermark = config.high_watermark, - low_watermark = config.low_watermark, - high_watermark_bytes, - low_watermark_bytes, - "Created backpressure pipe" - ); - - Self { - reader, - writer, - config, - buffer_usage: AtomicUsize::new(0), - state: AtomicBool::new(false), - total_written: AtomicUsize::new(0), - total_read: AtomicUsize::new(0), - high_watermark_bytes, - low_watermark_bytes, - created_at: Instant::now(), - } - } - - /// Take the reader end of the pipe (consumes self). - pub fn into_reader(self) -> DuplexStream { - self.reader - } - - /// Take the writer end of the pipe (consumes self). - pub fn into_writer(self) -> DuplexStream { - self.writer - } - - /// Split into reader and writer (consumes self). - pub fn split(self) -> (DuplexStream, DuplexStream) { - (self.reader, self.writer) - } - - /// Get current backpressure state. - pub fn state(&self) -> BackpressureState { - if self.state.load(Ordering::Acquire) { - BackpressureState::BackpressureApplied - } else { - BackpressureState::Normal - } - } - - /// Get a compact metadata snapshot for the pipe. - pub fn meta(&self) -> BackpressurePipeMeta { - BackpressurePipeMeta { - buffer_capacity: self.config.buffer_size, - state: self.state(), - age: self.age(), - } - } - - /// Get the age of this pipe. - pub fn age(&self) -> Duration { - self.created_at.elapsed() - } - - /// Get current buffer usage. - pub fn usage(&self) -> usize { - self.buffer_usage.load(Ordering::Acquire) - } - - /// Record bytes written (call after successful write). - pub fn record_write(&self, bytes: usize) { - self.total_written.fetch_add(bytes, Ordering::Relaxed); - self.buffer_usage.fetch_add(bytes, Ordering::Release); - self.update_watermark_state(); - } - - /// Record bytes read (call after successful read). - pub fn record_read(&self, bytes: usize) { - self.total_read.fetch_add(bytes, Ordering::Relaxed); - saturating_sub_atomic(&self.buffer_usage, bytes); - self.update_watermark_state(); - } - - /// Update watermark state and emit transition signals. - fn update_watermark_state(&self) { - let usage = self.buffer_usage.load(Ordering::Acquire); - let usage_percent = calculate_usage_percent(usage, self.config.buffer_size) as u32; - let (next_state, changed) = - apply_watermark_transition(&self.state, usage, self.high_watermark_bytes, self.low_watermark_bytes); - - if changed { - match next_state { - BackpressureState::HighWatermark => { - counter!("rustfs_backpressure_events_total", "state" => "high_watermark").increment(1); - - warn!( - buffer_usage = usage, - buffer_capacity = self.config.buffer_size, - usage_percent, - high_watermark = self.config.high_watermark, - "Backpressure: high watermark reached" - ); - } - BackpressureState::Normal => { - counter!("rustfs_backpressure_events_total", "state" => "normal").increment(1); - - debug!( - buffer_usage = usage, - buffer_capacity = self.config.buffer_size, - usage_percent, - low_watermark = self.config.low_watermark, - "Backpressure: returned to normal" - ); - } - BackpressureState::BackpressureApplied => {} - } - } - } - - /// Get total bytes written. - pub fn total_written(&self) -> usize { - self.total_written.load(Ordering::Relaxed) - } - - /// Get total bytes read. - pub fn total_read(&self) -> usize { - self.total_read.load(Ordering::Relaxed) - } - - /// Get buffer capacity. - pub fn capacity(&self) -> usize { - self.config.buffer_size - } -} - -impl Default for BackpressurePipe { - fn default() -> Self { - Self::new() - } -} - -/// A simple wrapper that provides backpressure monitoring for duplex streams. -/// -/// This is a lighter-weight alternative to `BackpressurePipe` that doesn't -/// wrap the streams but provides monitoring capabilities. -pub struct BackpressureMonitor { - /// Configuration. - config: ObjectPipeBackpressurePolicy, - /// Current buffer usage. - buffer_usage: AtomicUsize, - /// In high watermark state. - in_high_watermark: AtomicBool, - /// Cached high watermark threshold in bytes. - high_watermark_bytes: usize, - /// Cached low watermark threshold in bytes. - low_watermark_bytes: usize, -} - -impl BackpressureMonitor { - /// Create a new monitor with default configuration. - pub fn new() -> Self { - Self::with_config(ObjectPipeBackpressurePolicy::from_env()) - } - - /// Create a new monitor with custom configuration. - pub fn with_config(config: ObjectPipeBackpressurePolicy) -> Self { - let policy = config.to_concurrency_policy(); - let high_watermark_bytes = policy.high_watermark_bytes(); - let low_watermark_bytes = policy.low_watermark_bytes(); - Self { - config, - buffer_usage: AtomicUsize::new(0), - in_high_watermark: AtomicBool::new(false), - high_watermark_bytes, - low_watermark_bytes, - } - } - - /// Record bytes added to buffer. - pub fn on_write(&self, bytes: usize) -> BackpressureState { - self.buffer_usage.fetch_add(bytes, Ordering::Release); - self.update_state() - } - - /// Record bytes removed from buffer. - pub fn on_read(&self, bytes: usize) -> BackpressureState { - saturating_sub_atomic(&self.buffer_usage, bytes); - self.update_state() - } - - /// Get current state. - pub fn state(&self) -> BackpressureState { - if self.in_high_watermark.load(Ordering::Acquire) { - BackpressureState::HighWatermark - } else { - BackpressureState::Normal - } - } - - /// Get current buffer usage. - pub fn usage(&self) -> usize { - self.buffer_usage.load(Ordering::Acquire) - } - - /// Get usage percentage. - pub fn usage_percent(&self) -> f32 { - let usage = self.buffer_usage.load(Ordering::Acquire); - calculate_usage_percent(usage, self.config.buffer_size) - } - - /// Get a compact metadata snapshot for the monitor. - pub fn meta(&self) -> BackpressureMonitorMeta { - let usage = self.buffer_usage.load(Ordering::Acquire); - BackpressureMonitorMeta { - buffer_capacity: self.config.buffer_size, - usage_percent: calculate_usage_percent(usage, self.config.buffer_size), - state: self.state(), - } - } - - /// Update state based on current usage. - fn update_state(&self) -> BackpressureState { - let usage = self.buffer_usage.load(Ordering::Acquire); - let usage_percent = calculate_usage_percent(usage, self.config.buffer_size) as u32; - let (next_state, changed) = - apply_watermark_transition(&self.in_high_watermark, usage, self.high_watermark_bytes, self.low_watermark_bytes); - - if matches!(next_state, BackpressureState::HighWatermark) { - if changed { - counter!("rustfs_backpressure_events_total", "state" => "high_watermark").increment(1); - - debug!(usage_percent, "Backpressure: entered high watermark"); - } - BackpressureState::HighWatermark - } else { - if changed { - counter!("rustfs_backpressure_events_total", "state" => "normal").increment(1); - - debug!(usage_percent, "Backpressure: returned to normal"); - } - BackpressureState::Normal - } - } -} - -impl Default for BackpressureMonitor { - fn default() -> Self { - Self::new() - } -} - -#[cfg(test)] -#[allow(unused_imports)] -mod tests { - use super::{BackpressureMonitor, BackpressurePipe, BackpressureState, ObjectPipeBackpressurePolicy}; - - #[test] - fn test_backpressure_config_default() { - let config = ObjectPipeBackpressurePolicy::default(); - assert_eq!(config.buffer_size, 4 * 1024 * 1024); - assert_eq!(config.high_watermark, 80); - assert_eq!(config.low_watermark, 50); - } - - #[test] - fn test_backpressure_config_watermarks() { - let config = ObjectPipeBackpressurePolicy { - buffer_size: 1000, - high_watermark: 80, - low_watermark: 50, - }; - assert_eq!(config.high_watermark_bytes(), 800); - assert_eq!(config.low_watermark_bytes(), 500); - } - - #[test] - fn test_backpressure_policy_projects_to_concurrency_and_core_config() { - let config = ObjectPipeBackpressurePolicy { - buffer_size: 2000, - high_watermark: 75, - low_watermark: 40, - }; - let concurrency = config.to_concurrency_policy(); - let core = config.to_core_config(); - - assert_eq!(concurrency.buffer_size, config.buffer_size); - assert_eq!(concurrency.high_watermark, config.high_watermark); - assert_eq!(concurrency.low_watermark, config.low_watermark); - assert_eq!(core.high_water_mark, 0.75); - assert_eq!(core.low_water_mark, 0.40); - assert!(core.enabled); - } - - #[test] - fn test_backpressure_pipe_consumes_concurrency_policy_thresholds() { - let config = ObjectPipeBackpressurePolicy { - buffer_size: 2000, - high_watermark: 75, - low_watermark: 40, - }; - let concurrency = config.to_concurrency_policy(); - let pipe = BackpressurePipe::with_config(config); - - assert_eq!(pipe.capacity(), concurrency.buffer_size); - assert_eq!(pipe.high_watermark_bytes, concurrency.high_watermark_bytes()); - assert_eq!(pipe.low_watermark_bytes, concurrency.low_watermark_bytes()); - } - - #[test] - fn test_backpressure_monitor_consumes_concurrency_policy_thresholds() { - let config = ObjectPipeBackpressurePolicy { - buffer_size: 2000, - high_watermark: 75, - low_watermark: 40, - }; - let concurrency = config.to_concurrency_policy(); - let monitor = BackpressureMonitor::with_config(config); - - assert_eq!(monitor.meta().buffer_capacity, concurrency.buffer_size); - assert_eq!(monitor.high_watermark_bytes, concurrency.high_watermark_bytes()); - assert_eq!(monitor.low_watermark_bytes, concurrency.low_watermark_bytes()); - } - - #[test] - fn test_backpressure_state_display() { - assert_eq!(format!("{}", BackpressureState::Normal), "normal"); - assert_eq!(format!("{}", BackpressureState::HighWatermark), "high_watermark"); - assert_eq!(format!("{}", BackpressureState::BackpressureApplied), "backpressure_applied"); - } - - #[test] - fn test_backpressure_monitor() { - let config = ObjectPipeBackpressurePolicy { - buffer_size: 1000, - high_watermark: 80, - low_watermark: 50, - }; - let monitor = BackpressureMonitor::with_config(config); - - // Initially normal - assert_eq!(monitor.state(), BackpressureState::Normal); - assert_eq!(monitor.meta().buffer_capacity, 1000); - assert_eq!(monitor.meta().usage_percent, 0.0); - - // Write to reach high watermark - let state = monitor.on_write(850); - assert_eq!(state, BackpressureState::HighWatermark); - assert_eq!(monitor.meta().usage_percent, 85.0); - - // Read to go below low watermark - let state = monitor.on_read(400); - assert_eq!(state, BackpressureState::Normal); - assert_eq!(monitor.meta().usage_percent, 45.0); - } - - #[tokio::test] - async fn test_backpressure_pipe_creation() { - let pipe = BackpressurePipe::new(); - assert_eq!(pipe.capacity(), 4 * 1024 * 1024); - assert_eq!(pipe.state(), BackpressureState::Normal); - assert_eq!(pipe.meta().buffer_capacity, 4 * 1024 * 1024); - assert!(pipe.meta().age <= pipe.age()); - } - - #[test] - fn test_backpressure_pipe_state_transitions() { - let config = ObjectPipeBackpressurePolicy { - buffer_size: 1000, - high_watermark: 80, - low_watermark: 50, - }; - let pipe = BackpressurePipe::with_config(config); - - assert_eq!(pipe.state(), BackpressureState::Normal); - assert_eq!(pipe.meta().state, BackpressureState::Normal); - - pipe.record_write(850); - assert_eq!(pipe.state(), BackpressureState::BackpressureApplied); - assert_eq!(pipe.meta().state, BackpressureState::BackpressureApplied); - - pipe.record_read(400); - assert_eq!(pipe.state(), BackpressureState::Normal); - assert_eq!(pipe.meta().state, BackpressureState::Normal); - } -} diff --git a/rustfs/src/storage/concurrent_fix_test.rs b/rustfs/src/storage/concurrent_fix_test.rs index 2fedeced1..a80865976 100644 --- a/rustfs/src/storage/concurrent_fix_test.rs +++ b/rustfs/src/storage/concurrent_fix_test.rs @@ -14,17 +14,15 @@ //! Integration tests for concurrent request fix. //! -//! These tests verify that the timeout, backpressure, and deadlock detection -//! mechanisms work correctly under high concurrency scenarios. +//! These tests verify that the timeout and deadlock detection mechanisms work +//! correctly under high concurrency scenarios. #[cfg(test)] mod tests { - use crate::storage::backpressure::{BackpressureMonitor, BackpressureState, ObjectPipeBackpressurePolicy}; use crate::storage::concurrency::{IoLoadLevel, IoPriority}; use crate::storage::deadlock_detector::{ DeadlockDetector, LockInfo, LockType, RequestHangDetectionPolicy, RequestResourceTracker, }; - use crate::storage::lock_optimizer::{LockOptimizeConfig, LockOptimizer, LockStats}; use crate::storage::timeout_wrapper::{GetObjectTimeoutPolicy, RequestTimeoutWrapper, TimedGetObjectResult}; use std::time::Duration; @@ -114,82 +112,6 @@ mod tests { } } - // ============================================ - // Backpressure Tests - // ============================================ - - #[test] - fn test_backpressure_config_defaults() { - let config = ObjectPipeBackpressurePolicy::default(); - assert_eq!(config.buffer_size, 4 * 1024 * 1024); // 4MB - assert_eq!(config.high_watermark, 80); - assert_eq!(config.low_watermark, 50); - } - - #[test] - fn test_backpressure_monitor_state_transitions() { - let config = ObjectPipeBackpressurePolicy { - buffer_size: 1000, - high_watermark: 80, - low_watermark: 50, - }; - let monitor = BackpressureMonitor::with_config(config); - - // Initially normal - assert_eq!(monitor.state(), BackpressureState::Normal); - - // Write to reach high watermark - let state = monitor.on_write(850); - assert_eq!(state, BackpressureState::HighWatermark); - - // Read to go below low watermark - let state = monitor.on_read(400); - assert_eq!(state, BackpressureState::Normal); - } - - #[test] - fn test_backpressure_usage_percent() { - let config = ObjectPipeBackpressurePolicy { - buffer_size: 1000, - high_watermark: 80, - low_watermark: 50, - }; - let monitor = BackpressureMonitor::with_config(config); - - monitor.on_write(500); - assert!((monitor.usage_percent() - 50.0).abs() < 1.0); - } - - // ============================================ - // Lock Optimizer Tests - // ============================================ - - #[test] - fn test_lock_optimize_config_defaults() { - let config = LockOptimizeConfig::default(); - assert!(config.enabled); - assert_eq!(config.acquire_timeout, Duration::from_secs(5)); - } - - #[test] - fn test_lock_stats_tracking() { - let stats = LockStats::new(); - - stats.record_acquire(); - stats.record_early_release(Duration::from_millis(100)); - stats.record_early_release(Duration::from_millis(200)); - - assert_eq!(stats.locks_acquired.load(std::sync::atomic::Ordering::Relaxed), 1); - assert_eq!(stats.locks_released_early.load(std::sync::atomic::Ordering::Relaxed), 2); - assert_eq!(stats.max_hold_time(), Duration::from_millis(200)); - } - - #[test] - fn test_lock_optimizer_creation() { - let optimizer = LockOptimizer::new(); - assert!(optimizer.is_enabled()); - } - // ============================================ // I/O Priority Tests // ============================================ diff --git a/rustfs/src/storage/lock_optimizer.rs b/rustfs/src/storage/lock_optimizer.rs deleted file mode 100644 index 3636c50c3..000000000 --- a/rustfs/src/storage/lock_optimizer.rs +++ /dev/null @@ -1,458 +0,0 @@ -// Copyright 2024 RustFS Team -// -// Licensed under the Apache License, Version 2.0 (the "License"); -// you may not use this file except in compliance with the License. -// You may obtain a copy of the License at -// -// http://www.apache.org/licenses/LICENSE-2.0 -// -// Unless required by applicable law or agreed to in writing, software -// distributed under the License is distributed on an "AS IS" BASIS, -// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -// See the License for the specific language governing permissions and -// limitations under the License. - -//! Lock Optimization for GetObject Operations. -//! -//! This module provides optimized lock management for read operations, -//! reducing lock contention by releasing locks early (after metadata read) -//! rather than holding them for the entire data transfer duration. -//! -//! # Migration Note -//! -//! For new code, consider using `rustfs_io_core::LockOptimizer` which provides -//! the same core functionality with better separation of concerns. This module -//! remains for backward compatibility and storage-specific configuration. -//! -//! ```ignore -//! // Recommended: Use io-core directly -//! use rustfs_io_core::LockOptimizer; -//! let optimizer = LockOptimizer::with_defaults(); -//! ``` - -// Allow dead_code for public API that may be used by external modules or future features -//! # Key Features -//! -//! - Early lock release after metadata read -//! - Lock hold time monitoring -//! - Configurable optimization (can be disabled for debugging) -//! - Lock contention metrics emitted through the shared metrics pipeline -//! -//! # Architecture -//! -//! ```text -//! Traditional: [Acquire Lock] --> [Read Metadata] --> [Transfer Data] --> [Release Lock] -//! |<------------------ Lock Held ------------------>| -//! -//! Optimized: [Acquire Lock] --> [Read Metadata] --> [Release Lock] --> [Transfer Data] -//! |<- Lock Held ->| -//! ``` - -use std::sync::Arc; -use std::sync::atomic::{AtomicU64, Ordering}; -use std::time::{Duration, Instant}; -use tracing::debug; - -use metrics::histogram; - -/// Lock optimization configuration. -#[derive(Debug, Clone)] -pub struct LockOptimizeConfig { - /// Whether to enable lock optimization. - /// When enabled, read locks are released after metadata read. - /// When disabled, locks are held for the entire operation (traditional behavior). - pub enabled: bool, - /// Lock acquisition timeout. - pub acquire_timeout: Duration, -} - -impl Default for LockOptimizeConfig { - fn default() -> Self { - Self { - enabled: rustfs_config::DEFAULT_OBJECT_LOCK_OPTIMIZATION_ENABLE, - acquire_timeout: Duration::from_secs(rustfs_config::DEFAULT_OBJECT_LOCK_ACQUIRE_TIMEOUT), - } - } -} - -impl LockOptimizeConfig { - /// Load configuration from environment variables. - pub fn from_env() -> Self { - let enabled = rustfs_utils::get_env_bool( - rustfs_config::ENV_OBJECT_LOCK_OPTIMIZATION_ENABLE, - rustfs_config::DEFAULT_OBJECT_LOCK_OPTIMIZATION_ENABLE, - ); - let acquire_timeout = Duration::from_secs(rustfs_utils::get_env_u64( - rustfs_config::ENV_OBJECT_LOCK_ACQUIRE_TIMEOUT, - rustfs_config::DEFAULT_OBJECT_LOCK_ACQUIRE_TIMEOUT, - )); - - Self { - enabled, - acquire_timeout, - } - } -} - -/// Statistics for lock optimization monitoring. -#[derive(Debug, Default)] -pub struct LockStats { - /// Total locks acquired. - pub locks_acquired: AtomicU64, - /// Total locks released early. - pub locks_released_early: AtomicU64, - /// Total lock hold time in microseconds. - pub total_hold_time_us: AtomicU64, - /// Maximum lock hold time in microseconds. - pub max_hold_time_us: AtomicU64, -} - -impl LockStats { - /// Create new lock statistics. - pub fn new() -> Self { - Self::default() - } - - /// Record a lock acquisition. - pub fn record_acquire(&self) { - self.locks_acquired.fetch_add(1, Ordering::Relaxed); - } - - /// Record an early lock release. - pub fn record_early_release(&self, hold_time: Duration) { - self.locks_released_early.fetch_add(1, Ordering::Relaxed); - self.record_hold_time(hold_time); - } - - /// Record lock hold time. - fn record_hold_time(&self, hold_time: Duration) { - let hold_time_us = hold_time.as_micros() as u64; - self.total_hold_time_us.fetch_add(hold_time_us, Ordering::Relaxed); - - // Update max hold time - let mut current_max = self.max_hold_time_us.load(Ordering::Relaxed); - while hold_time_us > current_max { - match self - .max_hold_time_us - .compare_exchange_weak(current_max, hold_time_us, Ordering::Relaxed, Ordering::Relaxed) - { - Ok(_) => break, - Err(actual) => current_max = actual, - } - } - } - - /// Get average hold time. - pub fn avg_hold_time(&self) -> Duration { - let total = self.total_hold_time_us.load(Ordering::Relaxed); - let count = self.locks_released_early.load(Ordering::Relaxed); - total.checked_div(count).map(Duration::from_micros).unwrap_or(Duration::ZERO) - } - - /// Get maximum hold time. - pub fn max_hold_time(&self) -> Duration { - Duration::from_micros(self.max_hold_time_us.load(Ordering::Relaxed)) - } -} - -/// Global lock statistics. -static LOCK_STATS: std::sync::OnceLock> = std::sync::OnceLock::new(); - -/// Get global lock statistics. -pub fn get_lock_stats() -> Arc { - LOCK_STATS.get_or_init(|| Arc::new(LockStats::new())).clone() -} - -/// An optimized lock guard that supports early release. -/// -/// This wraps the actual lock guard and provides: -/// - Early release capability (before drop) -/// - Hold time tracking -/// - Metrics reporting -pub struct OptimizedLockGuard { - /// The underlying lock guard. - guard: Option, - /// When the lock was acquired. - acquire_time: Instant, - /// Whether the lock has been released. - released: bool, - /// Lock resource name (for logging). - resource: String, - /// Statistics reference. - stats: Arc, -} - -impl OptimizedLockGuard { - /// Create a new optimized lock guard. - pub fn new(guard: G, resource: impl Into) -> Self { - let stats = get_lock_stats(); - stats.record_acquire(); - - Self { - guard: Some(guard), - acquire_time: Instant::now(), - released: false, - resource: resource.into(), - stats, - } - } - - /// Get the lock hold time so far. - pub fn hold_time(&self) -> Duration { - self.acquire_time.elapsed() - } - - /// Check if the lock has been released. - pub fn is_released(&self) -> bool { - self.released - } - - /// Release the lock early (before drop). - /// - /// This is the key optimization: releasing the lock after - /// metadata read rather than waiting for the entire operation. - pub fn early_release(&mut self) { - if self.released { - return; - } - - let hold_time = self.hold_time(); - self.guard.take(); - self.released = true; - - self.stats.record_early_release(hold_time); - - histogram!("rustfs_lock_hold_duration_seconds").record(hold_time.as_secs_f64()); - - debug!( - resource = %self.resource, - hold_time_ms = hold_time.as_millis(), - "Lock released early (optimization active)" - ); - } - - /// Get a reference to the underlying guard. - pub fn as_ref(&self) -> Option<&G> { - if self.released { None } else { self.guard.as_ref() } - } -} - -impl Drop for OptimizedLockGuard { - fn drop(&mut self) { - if !self.released { - let hold_time = self.hold_time(); - self.guard.take(); - self.released = true; - - self.stats.record_early_release(hold_time); - - histogram!("rustfs_lock_hold_duration_seconds").record(hold_time.as_secs_f64()); - - debug!( - resource = %self.resource, - hold_time_ms = hold_time.as_millis(), - "Lock released on drop (normal release)" - ); - } - } -} - -/// A scope guard that releases a lock when it goes out of scope. -/// -/// This is a simpler version of OptimizedLockGuard for cases -/// where we just need RAII semantics without tracking. -pub struct LockScopeGuard { - guard: Option, -} - -impl LockScopeGuard { - /// Create a new scope guard. - pub fn new(guard: G) -> Self { - Self { guard: Some(guard) } - } - - /// Release the lock early. - pub fn release(&mut self) { - self.guard.take(); - } -} - -impl Drop for LockScopeGuard { - fn drop(&mut self) { - self.guard.take(); - } -} - -/// Helper for managing lock optimization in GetObject operations. -/// -/// This provides a clean interface for the common pattern: -/// 1. Acquire lock -/// 2. Read metadata -/// 3. Release lock (if optimization enabled) -/// 4. Transfer data (without lock) -pub struct LockOptimizer { - /// Configuration. - config: LockOptimizeConfig, -} - -impl LockOptimizer { - /// Create a new lock optimizer with default configuration. - pub fn new() -> Self { - Self { - config: LockOptimizeConfig::from_env(), - } - } - - /// Create a new lock optimizer with custom configuration. - pub fn with_config(config: LockOptimizeConfig) -> Self { - Self { config } - } - - /// Check if lock optimization is enabled. - pub fn is_enabled(&self) -> bool { - self.config.enabled - } - - /// Get the lock acquisition timeout. - pub fn acquire_timeout(&self) -> Duration { - self.config.acquire_timeout - } - - /// Wrap a lock guard for optimization. - pub fn wrap_guard(&self, guard: G, resource: impl Into) -> OptimizedLockGuard { - OptimizedLockGuard::new(guard, resource) - } - - /// Execute a metadata read operation with lock optimization. - /// - /// This is the main entry point for optimized lock usage: - /// - If optimization is enabled: lock is released after metadata_fn completes - /// - If optimization is disabled: lock is held until the returned guard is dropped - /// - /// # Arguments - /// - /// * `guard` - The lock guard to optimize - /// * `resource` - Resource name for logging - /// * `metadata_fn` - Function to read metadata while holding lock - /// - /// # Returns - /// - /// A tuple of (metadata result, optional guard to hold for later release) - pub async fn with_optimized_lock( - &self, - guard: G, - resource: impl Into, - metadata_fn: F, - ) -> (T, Option>) - where - F: FnOnce() -> Fut, - Fut: std::future::Future, - { - let resource = resource.into(); - let mut optimized = OptimizedLockGuard::new(guard, &resource); - - // Execute metadata read while holding lock - let result = metadata_fn().await; - - if self.config.enabled { - // Release lock early - optimized.early_release(); - (result, None) - } else { - // Keep lock for caller to release - (result, Some(optimized)) - } - } -} - -impl Default for LockOptimizer { - fn default() -> Self { - Self::new() - } -} - -/// Check if lock optimization is enabled globally. -pub fn is_lock_optimization_enabled() -> bool { - rustfs_utils::get_env_bool( - rustfs_config::ENV_OBJECT_LOCK_OPTIMIZATION_ENABLE, - rustfs_config::DEFAULT_OBJECT_LOCK_OPTIMIZATION_ENABLE, - ) -} - -#[cfg(test)] -#[allow(unused_imports)] -mod tests { - use super::{LockOptimizeConfig, LockOptimizer, LockStats, OptimizedLockGuard}; - use std::sync::Mutex; - use std::sync::atomic::Ordering; - use std::time::Duration; - - #[test] - fn test_lock_optimize_config_default() { - let config = LockOptimizeConfig::default(); - assert!(config.enabled); - assert_eq!(config.acquire_timeout, Duration::from_secs(5)); - } - - #[test] - fn test_lock_stats() { - let stats = LockStats::new(); - - stats.record_acquire(); - stats.record_early_release(Duration::from_millis(100)); - stats.record_early_release(Duration::from_millis(200)); - - assert_eq!(stats.locks_acquired.load(Ordering::Relaxed), 1); - assert_eq!(stats.locks_released_early.load(Ordering::Relaxed), 2); - assert_eq!(stats.max_hold_time(), Duration::from_millis(200)); - } - - #[test] - fn test_optimized_lock_guard() { - let guard = Mutex::new(42); - let locked = guard.lock().unwrap(); - let mut optimized = OptimizedLockGuard::new(locked, "test-resource"); - - assert!(!optimized.is_released()); - assert!(optimized.hold_time() < Duration::from_secs(1)); - - optimized.early_release(); - assert!(optimized.is_released()); - } - - #[test] - fn test_lock_optimizer() { - let optimizer = LockOptimizer::new(); - assert!(optimizer.is_enabled()); - } - - #[tokio::test] - async fn test_with_optimized_lock_enabled() { - let optimizer = LockOptimizer::new(); - let guard = Mutex::new(42); - let locked = guard.lock().unwrap(); - - let (result, returned_guard) = optimizer.with_optimized_lock(locked, "test-resource", || async { 100 }).await; - - assert_eq!(result, 100); - // With optimization enabled, guard should be None (released early) - assert!(returned_guard.is_none()); - } - - #[tokio::test] - async fn test_with_optimized_lock_disabled() { - let config = LockOptimizeConfig { - enabled: false, - acquire_timeout: Duration::from_secs(5), - }; - let optimizer = LockOptimizer::with_config(config); - let guard = Mutex::new(42); - let locked = guard.lock().unwrap(); - - let (result, returned_guard) = optimizer.with_optimized_lock(locked, "test-resource", || async { 100 }).await; - - assert_eq!(result, 100); - // With optimization disabled, guard should be Some (held for later) - assert!(returned_guard.is_some()); - } -} diff --git a/rustfs/src/storage/mod.rs b/rustfs/src/storage/mod.rs index 58b6a0ce9..e0817da90 100644 --- a/rustfs/src/storage/mod.rs +++ b/rustfs/src/storage/mod.rs @@ -13,12 +13,10 @@ // limitations under the License. pub mod access; -pub mod backpressure; pub mod concurrency; pub mod deadlock_detector; pub mod ecfs; pub(crate) mod helper; -pub mod lock_optimizer; pub mod options; pub mod request_context; pub mod rpc;