diff --git a/crates/concurrency/src/backpressure.rs b/crates/concurrency/src/backpressure.rs index 077f4e782..fd0cecbc4 100644 --- a/crates/concurrency/src/backpressure.rs +++ b/crates/concurrency/src/backpressure.rs @@ -14,15 +14,17 @@ //! Backpressure management -use rustfs_io_core::{BackpressureMonitor as CoreBackpressureMonitor, BackpressureState}; +use rustfs_io_core::{ + BackpressureConfig as CoreBackpressureConfig, BackpressureMonitor as CoreBackpressureMonitor, BackpressureState, +}; use rustfs_io_metrics::backpressure_metrics; use std::sync::Arc; use std::time::Instant; use tokio::io::{DuplexStream, duplex}; -/// Backpressure configuration -#[derive(Debug, Clone)] -pub struct BackpressureConfig { +/// Facade policy for duplex-pipe watermark backpressure. +#[derive(Debug, Clone, Copy)] +pub struct PipeBackpressurePolicy { /// Buffer size in bytes pub buffer_size: usize, /// High watermark percentage @@ -31,7 +33,7 @@ pub struct BackpressureConfig { pub low_watermark: u32, } -impl Default for BackpressureConfig { +impl Default for PipeBackpressurePolicy { fn default() -> Self { Self { buffer_size: 4 * 1024 * 1024, // 4MB @@ -41,7 +43,7 @@ impl Default for BackpressureConfig { } } -impl BackpressureConfig { +impl PipeBackpressurePolicy { /// 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 @@ -51,42 +53,59 @@ impl BackpressureConfig { pub fn low_watermark_bytes(&self) -> usize { (self.buffer_size as u64 * self.low_watermark as u64 / 100) as usize } + + /// Convert the facade policy into the reusable io-core admission-pressure config. + /// + /// The concurrency layer still owns duplex buffer sizing, but the shared + /// overload/admission primitive lives in `io-core`. + pub fn to_core_config(&self) -> CoreBackpressureConfig { + CoreBackpressureConfig { + max_concurrent: 32, + high_water_mark: self.high_watermark as f64 / 100.0, + low_water_mark: self.low_watermark as f64 / 100.0, + cooldown: std::time::Duration::from_millis(100), + enabled: true, + } + } } /// Backpressure manager pub struct BackpressureManager { - config: BackpressureConfig, + config: PipeBackpressurePolicy, + core_config: CoreBackpressureConfig, monitor: Arc, } impl BackpressureManager { /// Create a new backpressure manager pub fn new(buffer_size: usize, high_watermark: u32, low_watermark: u32) -> Self { - let config = BackpressureConfig { + Self::from_policy(PipeBackpressurePolicy { buffer_size, high_watermark, low_watermark, - }; - - let core_config = rustfs_io_core::BackpressureConfig { - max_concurrent: 32, - high_water_mark: high_watermark as f64 / 100.0, - low_water_mark: low_watermark as f64 / 100.0, - cooldown: std::time::Duration::from_millis(100), - enabled: true, - }; + }) + } + /// Create a new backpressure manager from the facade policy type. + pub fn from_policy(config: PipeBackpressurePolicy) -> Self { + let core_config = config.to_core_config(); Self { config, + core_config: core_config.clone(), monitor: Arc::new(CoreBackpressureMonitor::new(core_config)), } } /// Get the configuration - pub fn config(&self) -> &BackpressureConfig { + pub fn config(&self) -> &PipeBackpressurePolicy { &self.config } + /// Get the derived io-core admission-pressure configuration. + pub fn core_config(&self) -> &CoreBackpressureConfig { + &self.core_config + } + /// Get the monitor pub fn monitor(&self) -> Arc { self.monitor.clone() @@ -94,7 +113,7 @@ impl BackpressureManager { /// Create a backpressure pipe pub fn create_pipe(&self) -> BackpressurePipe { - BackpressurePipe::new(self.config.clone(), self.monitor.clone()) + BackpressurePipe::new(self.config, self.monitor.clone()) } /// Get current state @@ -112,13 +131,24 @@ impl BackpressureManager { pub struct BackpressurePipe { reader: DuplexStream, writer: DuplexStream, - config: BackpressureConfig, + config: PipeBackpressurePolicy, monitor: Arc, created_at: Instant, } +/// Shared pipe metadata snapshot for facade-level backpressure pipes. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct BackpressurePipeMeta { + /// Configured duplex buffer capacity in bytes. + pub buffer_capacity: usize, + /// Current backpressure state reported by the shared core monitor. + pub state: BackpressureState, + /// Age of the pipe since creation. + pub age: std::time::Duration, +} + impl BackpressurePipe { - fn new(config: BackpressureConfig, monitor: Arc) -> Self { + fn new(config: PipeBackpressurePolicy, monitor: Arc) -> Self { let (reader, writer) = duplex(config.buffer_size); Self { @@ -146,7 +176,7 @@ impl BackpressurePipe { } /// Get the configuration - pub fn config(&self) -> &BackpressureConfig { + pub fn config(&self) -> &PipeBackpressurePolicy { &self.config } @@ -160,6 +190,15 @@ impl BackpressurePipe { self.created_at.elapsed() } + /// 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(), + } + } + /// Check if should apply backpressure pub fn should_apply_backpressure(&self) -> bool { let should = self.monitor.should_apply_backpressure(); @@ -170,45 +209,26 @@ impl BackpressurePipe { } } -/// Backpressure event -#[allow(dead_code)] -#[derive(Debug, Clone)] -pub struct BackpressureEvent { - /// Event timestamp - pub timestamp: Instant, - /// Event type - pub event_type: BackpressureEventType, - /// Buffer usage - pub buffer_usage: usize, - /// Buffer capacity - pub buffer_capacity: usize, -} - -/// Backpressure event type -#[allow(dead_code)] -#[derive(Debug, Clone, Copy)] -pub enum BackpressureEventType { - /// High watermark reached - HighWatermarkReached, - /// High watermark exited - HighWatermarkExited, - /// Backpressure applied - BackpressureApplied, - /// Backpressure released - BackpressureReleased, -} - #[cfg(test)] mod tests { use super::*; #[test] fn test_backpressure_config() { - let config = BackpressureConfig::default(); + let config = PipeBackpressurePolicy::default(); assert_eq!(config.buffer_size, 4 * 1024 * 1024); assert!(config.high_watermark > config.low_watermark); } + #[test] + fn test_backpressure_policy_to_core_config() { + let policy = PipeBackpressurePolicy::default(); + let core = policy.to_core_config(); + assert_eq!(core.high_water_mark, policy.high_watermark as f64 / 100.0); + assert_eq!(core.low_water_mark, policy.low_watermark as f64 / 100.0); + assert!(core.enabled); + } + #[test] fn test_backpressure_manager() { let manager = BackpressureManager::new(1024, 80, 50); @@ -220,5 +240,6 @@ mod tests { let manager = BackpressureManager::new(1024, 80, 50); let pipe = manager.create_pipe(); assert_eq!(pipe.state(), BackpressureState::Normal); + assert_eq!(pipe.meta().buffer_capacity, 1024); } } diff --git a/crates/concurrency/src/config.rs b/crates/concurrency/src/config.rs index 00bca9ffe..6fe7e1733 100644 --- a/crates/concurrency/src/config.rs +++ b/crates/concurrency/src/config.rs @@ -14,6 +14,10 @@ //! Configuration for concurrency management +use crate::{ + backpressure::PipeBackpressurePolicy, deadlock::DeadlockMonitorPolicy, scheduler::SchedulerPolicy, + timeout::TimeoutManagerPolicy, +}; use std::time::Duration; /// Feature flags for concurrency modules @@ -72,84 +76,39 @@ impl ConcurrencyFeatures { } } +/// Facade policy for lock manager behavior. +#[derive(Debug, Clone, Copy)] +pub struct LockManagerPolicy { + /// Enable lock optimization. + pub enabled: bool, + /// Lock acquisition timeout. + pub acquire_timeout: Duration, +} + +impl Default for LockManagerPolicy { + fn default() -> Self { + Self { + enabled: true, + acquire_timeout: Duration::from_secs(5), + } + } +} + /// Main configuration for concurrency management -#[derive(Debug, Clone)] +#[derive(Debug, Clone, Default)] pub struct ConcurrencyConfig { /// Feature flags pub features: ConcurrencyFeatures, - - // Timeout configuration - /// Default timeout duration - pub default_timeout: Duration, - /// Maximum timeout duration - pub max_timeout: Duration, - /// Enable dynamic timeout - pub enable_dynamic_timeout: bool, - - // Lock configuration - /// Enable lock optimization - pub enable_lock_optimization: bool, - /// Lock acquisition timeout - pub lock_acquire_timeout: Duration, - - // Deadlock configuration - /// Enable deadlock detection - pub enable_deadlock_detection: bool, - /// Deadlock check interval - pub deadlock_check_interval: Duration, - /// Hang threshold - pub hang_threshold: Duration, - - // Backpressure configuration - /// Buffer size for backpressure - pub backpressure_buffer_size: usize, - /// High watermark percentage - pub high_watermark: u32, - /// Low watermark percentage - pub low_watermark: u32, - - // Scheduler configuration - /// Base buffer size for I/O - pub io_buffer_size: usize, - /// Maximum buffer size - pub max_buffer_size: usize, - /// High priority size threshold - pub high_priority_threshold: usize, - /// Low priority size threshold - pub low_priority_threshold: usize, -} - -impl Default for ConcurrencyConfig { - fn default() -> Self { - Self { - features: ConcurrencyFeatures::default(), - - // Timeout defaults - default_timeout: Duration::from_secs(30), - max_timeout: Duration::from_secs(300), - enable_dynamic_timeout: true, - - // Lock defaults - enable_lock_optimization: true, - lock_acquire_timeout: Duration::from_secs(5), - - // Deadlock defaults - enable_deadlock_detection: false, - deadlock_check_interval: Duration::from_secs(10), - hang_threshold: Duration::from_secs(60), - - // Backpressure defaults - backpressure_buffer_size: 4 * 1024 * 1024, // 4MB - high_watermark: 80, - low_watermark: 50, - - // Scheduler defaults - io_buffer_size: 64 * 1024, // 64KB - max_buffer_size: 4 * 1024 * 1024, // 4MB - high_priority_threshold: 1024 * 1024, // 1MB - low_priority_threshold: 10 * 1024 * 1024, // 10MB - } - } + /// Timeout facade policy. + pub timeout_policy: TimeoutManagerPolicy, + /// Lock facade policy. + pub lock_policy: LockManagerPolicy, + /// Deadlock facade policy. + pub deadlock_policy: DeadlockMonitorPolicy, + /// Backpressure facade policy. + pub backpressure_policy: PipeBackpressurePolicy, + /// Scheduler facade policy. + pub scheduler_policy: SchedulerPolicy, } impl ConcurrencyConfig { @@ -161,25 +120,25 @@ impl ConcurrencyConfig { if let Ok(val) = std::env::var("RUSTFS_TIMEOUT_DEFAULT") && let Ok(secs) = val.parse::() { - config.default_timeout = Duration::from_secs(secs); + config.timeout_policy.default_timeout = Duration::from_secs(secs); } if let Ok(val) = std::env::var("RUSTFS_TIMEOUT_MAX") && let Ok(secs) = val.parse::() { - config.max_timeout = Duration::from_secs(secs); + config.timeout_policy.max_timeout = Duration::from_secs(secs); } if let Ok(val) = std::env::var("RUSTFS_BACKPRESSURE_BUFFER_SIZE") && let Ok(size) = val.parse::() { - config.backpressure_buffer_size = size; + config.backpressure_policy.buffer_size = size; } if let Ok(val) = std::env::var("RUSTFS_IO_BUFFER_SIZE") && let Ok(size) = val.parse::() { - config.io_buffer_size = size; + config.scheduler_policy.base_buffer_size = size; } config @@ -187,18 +146,25 @@ impl ConcurrencyConfig { /// Validate configuration pub fn validate(&self) -> Result<(), ConfigError> { - if self.default_timeout > self.max_timeout { + if self.timeout_policy.default_timeout > self.timeout_policy.max_timeout { return Err(ConfigError::InvalidTimeout("default_timeout cannot exceed max_timeout".to_string())); } + if self.timeout_policy.min_timeout > self.timeout_policy.max_timeout { + return Err(ConfigError::InvalidTimeout("min_timeout cannot exceed max_timeout".to_string())); + } - if self.high_watermark <= self.low_watermark || self.high_watermark > 100 { + if self.backpressure_policy.high_watermark <= self.backpressure_policy.low_watermark + || self.backpressure_policy.high_watermark > 100 + { return Err(ConfigError::InvalidBackpressure( "high_watermark must be > low_watermark and <= 100".to_string(), )); } - if self.io_buffer_size > self.max_buffer_size { - return Err(ConfigError::InvalidScheduler("io_buffer_size cannot exceed max_buffer_size".to_string())); + if self.scheduler_policy.base_buffer_size > self.scheduler_policy.max_buffer_size { + return Err(ConfigError::InvalidScheduler( + "base_buffer_size cannot exceed max_buffer_size".to_string(), + )); } Ok(()) @@ -235,8 +201,12 @@ mod tests { #[test] fn test_invalid_timeout() { let config = ConcurrencyConfig { - default_timeout: Duration::from_secs(100), - max_timeout: Duration::from_secs(50), + timeout_policy: TimeoutManagerPolicy { + default_timeout: Duration::from_secs(100), + max_timeout: Duration::from_secs(50), + enable_dynamic: true, + ..Default::default() + }, ..Default::default() }; assert!( @@ -245,6 +215,22 @@ mod tests { ); } + #[test] + fn test_invalid_min_timeout() { + let config = ConcurrencyConfig { + timeout_policy: TimeoutManagerPolicy { + min_timeout: Duration::from_secs(100), + max_timeout: Duration::from_secs(50), + ..Default::default() + }, + ..Default::default() + }; + assert!( + config.validate().is_err(), + "validate() should return an error when min_timeout > max_timeout" + ); + } + #[test] fn test_features() { let features = ConcurrencyFeatures::all(); diff --git a/crates/concurrency/src/deadlock.rs b/crates/concurrency/src/deadlock.rs index 20816a1dd..e291ec352 100644 --- a/crates/concurrency/src/deadlock.rs +++ b/crates/concurrency/src/deadlock.rs @@ -14,15 +14,15 @@ //! Deadlock detection management -use rustfs_io_core::{DeadlockDetector as CoreDeadlockDetector, LockType}; +use rustfs_io_core::{DeadlockDetector as CoreDeadlockDetector, DeadlockDetectorConfig as CoreDeadlockConfig, LockType}; use rustfs_io_metrics::deadlock_metrics; use std::collections::HashMap; use std::sync::Arc; use std::time::{Duration, Instant}; -/// Deadlock configuration -#[derive(Debug, Clone)] -pub struct DeadlockConfig { +/// Facade policy for the concurrency-layer deadlock monitor. +#[derive(Debug, Clone, Copy)] +pub struct DeadlockMonitorPolicy { /// Enable deadlock detection pub enabled: bool, /// Check interval @@ -31,7 +31,7 @@ pub struct DeadlockConfig { pub hang_threshold: Duration, } -impl Default for DeadlockConfig { +impl Default for DeadlockMonitorPolicy { fn default() -> Self { Self { enabled: false, @@ -41,9 +41,20 @@ impl Default for DeadlockConfig { } } +impl DeadlockMonitorPolicy { + /// Convert the facade policy into the reusable io-core deadlock config. + pub fn to_core_config(&self) -> CoreDeadlockConfig { + CoreDeadlockConfig { + enabled: self.enabled, + detection_interval: self.check_interval, + max_hold_time: self.hang_threshold, + } + } +} + /// Deadlock manager pub struct DeadlockManager { - config: DeadlockConfig, + config: DeadlockMonitorPolicy, detector: Arc, running: Arc>, } @@ -51,18 +62,16 @@ pub struct DeadlockManager { impl DeadlockManager { /// Create a new deadlock manager pub fn new(enabled: bool, check_interval: Duration, hang_threshold: Duration) -> Self { - let config = DeadlockConfig { + Self::from_policy(DeadlockMonitorPolicy { enabled, check_interval, hang_threshold, - }; - - let core_config = rustfs_io_core::DeadlockDetectorConfig { - enabled, - detection_interval: check_interval, - max_hold_time: hang_threshold, - }; + }) + } + /// Create a new deadlock manager from the facade policy type. + pub fn from_policy(config: DeadlockMonitorPolicy) -> Self { + let core_config = config.to_core_config(); Self { config, detector: Arc::new(CoreDeadlockDetector::new(core_config)), @@ -71,7 +80,7 @@ impl DeadlockManager { } /// Get the configuration - pub fn config(&self) -> &DeadlockConfig { + pub fn config(&self) -> &DeadlockMonitorPolicy { &self.config } @@ -129,7 +138,11 @@ impl DeadlockManager { } } -/// Request tracker for tracking resources +/// Lightweight compatibility wrapper for request-scoped deadlock bookkeeping. +/// +/// This type intentionally stays minimal in the concurrency layer. Rich +/// request-level lock/resource diagnostics belong to +/// `rustfs::storage::deadlock_detector::RequestResourceTracker`. pub struct RequestTracker { request_id: String, description: String, @@ -174,6 +187,11 @@ impl RequestTracker { deadlock_metrics::record_lock_acquisition("read"); } + /// Return a read-only view of tracked resource names. + pub fn resources(&self) -> &HashMap> { + &self.resources + } + /// Record a lock release pub fn record_lock_release(&mut self, lock_id: u64) { self.detector.record_release(lock_id); @@ -196,12 +214,24 @@ mod tests { assert!(!manager.config().enabled); } + #[test] + fn test_deadlock_policy_to_core_config() { + let policy = DeadlockMonitorPolicy::default(); + let core = policy.to_core_config(); + assert_eq!(core.enabled, policy.enabled); + assert_eq!(core.detection_interval, policy.check_interval); + assert_eq!(core.max_hold_time, policy.hang_threshold); + } + #[tokio::test] async fn test_request_tracker() { let manager = DeadlockManager::new(true, Duration::from_secs(10), Duration::from_secs(60)); - let tracker = manager.track_request("req-1".to_string(), "test request".to_string()); + let mut tracker = manager.track_request("req-1".to_string(), "test request".to_string()); + let lock_id = manager.register_lock(LockType::Mutex); + tracker.record_lock_acquire(lock_id, "bucket/key".to_string()); assert_eq!(tracker.request_id(), "req-1"); assert_eq!(tracker.description(), "test request"); + assert_eq!(tracker.resources().get("locks").map(Vec::len), Some(1)); } } diff --git a/crates/concurrency/src/lib.rs b/crates/concurrency/src/lib.rs index 6d0b5ff8e..cefe3ecba 100644 --- a/crates/concurrency/src/lib.rs +++ b/crates/concurrency/src/lib.rs @@ -128,19 +128,19 @@ pub mod workers; // Public module exports with feature gates #[cfg(feature = "timeout")] -pub use timeout::{TimeoutConfig, TimeoutGuard, TimeoutManager}; +pub use timeout::{TimeoutGuard, TimeoutManager, TimeoutManagerPolicy}; #[cfg(feature = "lock")] pub use lock::{LockConfig, LockManager, LockScopeGuard, OptimizedLockGuard}; #[cfg(feature = "deadlock")] -pub use deadlock::{DeadlockConfig, DeadlockManager, RequestTracker}; +pub use deadlock::{DeadlockManager, DeadlockMonitorPolicy, RequestTracker}; #[cfg(feature = "backpressure")] -pub use backpressure::{BackpressureConfig, BackpressureManager, BackpressurePipe}; +pub use backpressure::{BackpressureManager, BackpressurePipe, PipeBackpressurePolicy}; #[cfg(feature = "scheduler")] -pub use scheduler::{IoStrategy, SchedulerConfig, SchedulerManager}; +pub use scheduler::{IoStrategy, SchedulerManager, SchedulerPolicy}; // Configuration mod config; @@ -155,19 +155,19 @@ pub mod prelude { //! Prelude module for convenient imports #[cfg(feature = "timeout")] - pub use crate::timeout::{TimeoutConfig, TimeoutGuard, TimeoutManager}; + pub use crate::timeout::{TimeoutGuard, TimeoutManager, TimeoutManagerPolicy}; #[cfg(feature = "lock")] pub use crate::lock::{LockConfig, LockManager, LockScopeGuard, OptimizedLockGuard}; #[cfg(feature = "deadlock")] - pub use crate::deadlock::{DeadlockConfig, DeadlockManager, RequestTracker}; + pub use crate::deadlock::{DeadlockManager, DeadlockMonitorPolicy, RequestTracker}; #[cfg(feature = "backpressure")] - pub use crate::backpressure::{BackpressureConfig, BackpressureManager, BackpressurePipe}; + pub use crate::backpressure::{BackpressureManager, BackpressurePipe, PipeBackpressurePolicy}; #[cfg(feature = "scheduler")] - pub use crate::scheduler::{IoStrategy, SchedulerConfig, SchedulerManager}; + pub use crate::scheduler::{IoStrategy, SchedulerManager, SchedulerPolicy}; pub use crate::{ConcurrencyConfig, ConcurrencyFeatures, ConcurrencyManager}; } diff --git a/crates/concurrency/src/manager.rs b/crates/concurrency/src/manager.rs index f52126201..6259a7ff6 100644 --- a/crates/concurrency/src/manager.rs +++ b/crates/concurrency/src/manager.rs @@ -85,39 +85,22 @@ impl ConcurrencyManager { Self { #[cfg(feature = "timeout")] - timeout: Arc::new(crate::timeout::TimeoutManager::new( - config.default_timeout, - config.max_timeout, - config.enable_dynamic_timeout, - )), + timeout: Arc::new(crate::timeout::TimeoutManager::from_policy(config.timeout_policy)), #[cfg(feature = "lock")] lock: Arc::new(crate::lock::LockManager::new( - config.enable_lock_optimization, - config.lock_acquire_timeout, + config.lock_policy.enabled, + config.lock_policy.acquire_timeout, )), #[cfg(feature = "deadlock")] - deadlock: Arc::new(crate::deadlock::DeadlockManager::new( - config.enable_deadlock_detection, - config.deadlock_check_interval, - config.hang_threshold, - )), + deadlock: Arc::new(crate::deadlock::DeadlockManager::from_policy(config.deadlock_policy)), #[cfg(feature = "backpressure")] - backpressure: Arc::new(crate::backpressure::BackpressureManager::new( - config.backpressure_buffer_size, - config.high_watermark, - config.low_watermark, - )), + backpressure: Arc::new(crate::backpressure::BackpressureManager::from_policy(config.backpressure_policy)), #[cfg(feature = "scheduler")] - scheduler: Arc::new(crate::scheduler::SchedulerManager::new( - config.io_buffer_size, - config.max_buffer_size, - config.high_priority_threshold, - config.low_priority_threshold, - )), + scheduler: Arc::new(crate::scheduler::SchedulerManager::from_policy(config.scheduler_policy)), config, } @@ -244,7 +227,7 @@ impl ConcurrencyManager { pub async fn start(&self) { #[cfg(feature = "deadlock")] { - if self.config.enable_deadlock_detection { + if self.config.deadlock_policy.enabled { self.deadlock.start().await; } } diff --git a/crates/concurrency/src/scheduler.rs b/crates/concurrency/src/scheduler.rs index 0f19eb56d..86c5b3dce 100644 --- a/crates/concurrency/src/scheduler.rs +++ b/crates/concurrency/src/scheduler.rs @@ -22,9 +22,9 @@ use rustfs_io_metrics::io_metrics; use std::sync::Arc; use std::time::Duration; -/// Scheduler configuration -#[derive(Debug, Clone)] -pub struct SchedulerConfig { +/// Facade policy for the concurrency-layer scheduler manager. +#[derive(Debug, Clone, Copy)] +pub struct SchedulerPolicy { /// Base buffer size pub base_buffer_size: usize, /// Maximum buffer size @@ -35,7 +35,7 @@ pub struct SchedulerConfig { pub low_priority_threshold: usize, } -impl Default for SchedulerConfig { +impl Default for SchedulerPolicy { fn default() -> Self { Self { base_buffer_size: 64 * 1024, // 64KB @@ -46,9 +46,23 @@ impl Default for SchedulerConfig { } } +impl SchedulerPolicy { + /// Convert facade policy to io-core scheduler config. + pub fn to_core_config(&self) -> rustfs_io_core::IoSchedulerConfig { + rustfs_io_core::IoSchedulerConfig { + base_buffer_size: self.base_buffer_size, + max_buffer_size: self.max_buffer_size, + high_priority_size_threshold: self.high_priority_threshold, + low_priority_size_threshold: self.low_priority_threshold, + ..rustfs_io_core::IoSchedulerConfig::default() + } + } +} + /// Scheduler manager pub struct SchedulerManager { - config: SchedulerConfig, + config: SchedulerPolicy, + core_config: rustfs_io_core::IoSchedulerConfig, scheduler: Arc, } @@ -60,26 +74,35 @@ impl SchedulerManager { high_priority_threshold: usize, low_priority_threshold: usize, ) -> Self { - let config = SchedulerConfig { + Self::from_policy(SchedulerPolicy { base_buffer_size, max_buffer_size, high_priority_threshold, low_priority_threshold, - }; + }) + } - let core_config = rustfs_io_core::IoSchedulerConfig::default(); + /// Create a scheduler manager from facade policy. + pub fn from_policy(config: SchedulerPolicy) -> Self { + let core_config = config.to_core_config(); Self { config, + core_config: core_config.clone(), scheduler: Arc::new(CoreIoScheduler::new(core_config)), } } /// Get the configuration - pub fn config(&self) -> &SchedulerConfig { + pub fn config(&self) -> &SchedulerPolicy { &self.config } + /// Get the derived io-core scheduler config. + pub fn core_config(&self) -> &rustfs_io_core::IoSchedulerConfig { + &self.core_config + } + /// Get the scheduler pub fn scheduler(&self) -> Arc { self.scheduler.clone() @@ -87,7 +110,7 @@ impl SchedulerManager { /// Create an I/O strategy pub fn create_strategy(&self) -> IoStrategy { - IoStrategy::new(self.config.clone(), self.scheduler.clone()) + IoStrategy::new(self.config, self.scheduler.clone()) } /// Calculate buffer size @@ -111,12 +134,12 @@ impl SchedulerManager { /// I/O strategy pub struct IoStrategy { - config: SchedulerConfig, + config: SchedulerPolicy, scheduler: Arc, } impl IoStrategy { - fn new(config: SchedulerConfig, scheduler: Arc) -> Self { + fn new(config: SchedulerPolicy, scheduler: Arc) -> Self { Self { config, scheduler } } @@ -191,7 +214,7 @@ impl IoStrategy { } /// Get the configuration - pub fn config(&self) -> &SchedulerConfig { + pub fn config(&self) -> &SchedulerPolicy { &self.config } } @@ -202,10 +225,20 @@ mod tests { #[test] fn test_scheduler_config() { - let config = SchedulerConfig::default(); + let config = SchedulerPolicy::default(); assert!(config.base_buffer_size < config.max_buffer_size); } + #[test] + fn test_scheduler_policy_to_core_config() { + let policy = SchedulerPolicy::default(); + let core = policy.to_core_config(); + assert_eq!(core.base_buffer_size, policy.base_buffer_size); + assert_eq!(core.max_buffer_size, policy.max_buffer_size); + assert_eq!(core.high_priority_size_threshold, policy.high_priority_threshold); + assert_eq!(core.low_priority_size_threshold, policy.low_priority_threshold); + } + #[test] fn test_scheduler_manager() { let manager = SchedulerManager::new(1024, 4096, 512, 2048); diff --git a/crates/concurrency/src/timeout.rs b/crates/concurrency/src/timeout.rs index 52f730903..650cc91dc 100644 --- a/crates/concurrency/src/timeout.rs +++ b/crates/concurrency/src/timeout.rs @@ -14,60 +14,101 @@ //! Timeout management for operations -use rustfs_io_core::{TimeoutError, calculate_adaptive_timeout}; +use rustfs_io_core::{TimeoutConfig as CoreTimeoutConfig, TimeoutError, calculate_adaptive_timeout}; use std::time::{Duration, Instant}; use tokio_util::sync::CancellationToken; -/// Timeout configuration -#[derive(Debug, Clone)] -pub struct TimeoutConfig { +/// Facade policy for the concurrency-layer timeout manager. +#[derive(Debug, Clone, Copy)] +pub struct TimeoutManagerPolicy { /// Default timeout duration pub default_timeout: Duration, /// Maximum timeout duration pub max_timeout: Duration, + /// Minimum timeout floor (prevents dynamic calculation from going too low). + pub min_timeout: Duration, /// Enable dynamic timeout calculation pub enable_dynamic: bool, } -impl Default for TimeoutConfig { +impl Default for TimeoutManagerPolicy { fn default() -> Self { Self { default_timeout: Duration::from_secs(30), max_timeout: Duration::from_secs(300), + min_timeout: Duration::from_secs(5), enable_dynamic: true, } } } +impl TimeoutManagerPolicy { + /// Convert the facade policy into the reusable io-core timeout configuration. + /// + /// This keeps the concurrency layer explicitly wired to the shared core + /// timeout primitives without changing the facade's public behavior. + pub fn to_core_config(&self) -> CoreTimeoutConfig { + CoreTimeoutConfig { + base_timeout: self.default_timeout, + timeout_per_mb: Duration::ZERO, + max_timeout: self.max_timeout, + min_timeout: self.min_timeout, + get_object_timeout: self.default_timeout, + put_object_timeout: self.max_timeout, + list_objects_timeout: self.default_timeout, + enable_dynamic_timeout: self.enable_dynamic, + } + } +} + /// Timeout manager pub struct TimeoutManager { - config: TimeoutConfig, + config: TimeoutManagerPolicy, + core_config: CoreTimeoutConfig, } impl TimeoutManager { /// Create a new timeout manager pub fn new(default_timeout: Duration, max_timeout: Duration, enable_dynamic: bool) -> Self { - Self { - config: TimeoutConfig { - default_timeout, - max_timeout, - enable_dynamic, - }, - } + let min_timeout = default_timeout.min(max_timeout); + Self::from_policy(TimeoutManagerPolicy { + default_timeout, + max_timeout, + min_timeout, + enable_dynamic, + }) + } + + /// Create a new timeout manager from the facade policy type. + pub fn from_policy(config: TimeoutManagerPolicy) -> Self { + let config = TimeoutManagerPolicy { + // Guard clamp(min, max) from panic when callers provide an + // out-of-order policy (or very small max_timeout). + min_timeout: config.min_timeout.min(config.max_timeout), + ..config + }; + let core_config = config.to_core_config(); + Self { config, core_config } } /// Get the configuration - pub fn config(&self) -> &TimeoutConfig { + pub fn config(&self) -> &TimeoutManagerPolicy { &self.config } + /// Get the derived io-core timeout configuration. + pub fn core_config(&self) -> &CoreTimeoutConfig { + &self.core_config + } + /// Calculate timeout for a given size pub fn calculate_timeout(&self, size: u64, _history: &[Duration]) -> Duration { if !self.config.enable_dynamic { return self.config.default_timeout; } - calculate_adaptive_timeout(self.config.default_timeout, None, 0, size).min(self.config.max_timeout) + calculate_adaptive_timeout(self.core_config.base_timeout, None, 0, size) + .clamp(self.core_config.min_timeout, self.core_config.max_timeout) } /// Wrap an operation with timeout control @@ -87,7 +128,7 @@ impl TimeoutManager { /// Create a timeout guard for manual timeout control pub fn create_guard(&self, timeout: Option) -> TimeoutGuard { - TimeoutGuard::new(timeout.unwrap_or(self.config.default_timeout)) + TimeoutGuard::new(timeout.unwrap_or(self.core_config.base_timeout)) } } @@ -134,10 +175,41 @@ mod tests { #[test] fn test_timeout_config() { - let config = TimeoutConfig::default(); + let config = TimeoutManagerPolicy::default(); assert!(config.default_timeout < config.max_timeout); } + #[test] + fn test_timeout_policy_to_core_config() { + let policy = TimeoutManagerPolicy::default(); + let core = policy.to_core_config(); + assert_eq!(core.base_timeout, policy.default_timeout); + assert_eq!(core.max_timeout, policy.max_timeout); + assert_eq!(core.min_timeout, policy.min_timeout); + assert_eq!(core.get_object_timeout, policy.default_timeout); + assert!(core.enable_dynamic_timeout); + } + + #[test] + fn test_timeout_manager_new_sanitizes_min_timeout_with_small_max_timeout() { + let manager = TimeoutManager::new(Duration::from_secs(1), Duration::from_secs(1), true); + let timeout = manager.calculate_timeout(1024, &[]); + assert_eq!(timeout, Duration::from_secs(1)); + } + + #[test] + fn test_timeout_manager_from_policy_sanitizes_min_timeout() { + let manager = TimeoutManager::from_policy(TimeoutManagerPolicy { + default_timeout: Duration::from_secs(30), + max_timeout: Duration::from_secs(1), + min_timeout: Duration::from_secs(5), + enable_dynamic: true, + }); + + assert_eq!(manager.config().min_timeout, Duration::from_secs(1)); + assert_eq!(manager.core_config().min_timeout, Duration::from_secs(1)); + } + #[tokio::test] async fn test_wrap_operation_success() { let manager = TimeoutManager::new(Duration::from_secs(5), Duration::from_secs(10), true); diff --git a/rustfs/src/app/object_usecase.rs b/rustfs/src/app/object_usecase.rs index 5a157fe90..56e9a881a 100644 --- a/rustfs/src/app/object_usecase.rs +++ b/rustfs/src/app/object_usecase.rs @@ -32,7 +32,7 @@ use crate::storage::options::{ use crate::storage::request_context::spawn_traced; use crate::storage::s3_api::multipart::parse_list_parts_params; use crate::storage::sse::{SSEType, build_ssec_read_headers, encryption_material_to_metadata, map_get_object_reader_error}; -use crate::storage::timeout_wrapper::{RequestTimeoutWrapper, TimeoutConfig}; +use crate::storage::timeout_wrapper::{GetObjectTimeoutPolicy, RequestTimeoutWrapper}; use crate::storage::*; use bytes::Bytes; use datafusion::arrow::{ @@ -159,7 +159,7 @@ impl Drop for DeadlockRequestGuard { } struct GetObjectBootstrap { - timeout_config: TimeoutConfig, + timeout_config: GetObjectTimeoutPolicy, wrapper: RequestTimeoutWrapper, request_start: std::time::Instant, request_guard: GetObjectGuard, @@ -217,6 +217,12 @@ struct GetObjectOutputContext { optimal_buffer_size: usize, } +enum GetObjectTimeoutStage { + BeforeProcessing, + DiskPermitWait { permit_wait_duration: Duration }, + BeforeRead, +} + async fn enqueue_transitioned_delete_cleanup(bucket: &str, object: &str, opts: &ObjectOptions, existing: Option<&ObjectInfo>) { let Some(existing) = existing else { return; @@ -443,14 +449,14 @@ fn should_buffer_get_object_in_memory_with_threshold( #[cfg(test)] mod deadlock_request_guard_tests { use super::DeadlockRequestGuard; - use crate::storage::deadlock_detector::{DeadlockDetector, DeadlockDetectorConfig}; + use crate::storage::deadlock_detector::{DeadlockDetector, RequestHangDetectionPolicy}; use std::sync::Arc; #[test] fn deadlock_request_guard_unregisters_on_drop() { - let detector = Arc::new(DeadlockDetector::new(DeadlockDetectorConfig { + let detector = Arc::new(DeadlockDetector::new(RequestHangDetectionPolicy { enabled: true, - ..DeadlockDetectorConfig::default() + ..RequestHangDetectionPolicy::default() })); let request_id = "test-request-id".to_string(); @@ -1112,7 +1118,7 @@ impl DefaultObjectUsecase { } fn init_get_object_bootstrap(bucket: &str, key: &str, request_id: &str) -> S3Result { - let timeout_config = TimeoutConfig::from_env(); + let timeout_config = GetObjectTimeoutPolicy::from_env(); let wrapper = RequestTimeoutWrapper::with_request_id(timeout_config.clone(), request_id.to_string()); let request_start = std::time::Instant::now(); let request_guard = ConcurrencyManager::track_request(); @@ -1123,16 +1129,7 @@ impl DefaultObjectUsecase { deadlock_detector.register_request(&request_id, format!("GetObject {bucket}/{key}")); let deadlock_request_guard = DeadlockRequestGuard::new(deadlock_detector, request_id); - if wrapper.is_timeout() { - warn!( - bucket = %bucket, - key = %key, - timeout_secs = timeout_config.get_object_timeout.as_secs(), - elapsed_ms = wrapper.elapsed().as_millis(), - "GetObject request timed out before processing" - ); - return Err(s3_error!(InternalError, "Request timeout before processing")); - } + Self::ensure_get_object_not_timed_out(&wrapper, &timeout_config, bucket, key, GetObjectTimeoutStage::BeforeProcessing)?; rustfs_io_metrics::record_get_object_request_start(concurrent_requests); @@ -1154,7 +1151,7 @@ impl DefaultObjectUsecase { async fn acquire_get_object_io_planning<'a>( manager: &'a ConcurrencyManager, wrapper: &RequestTimeoutWrapper, - timeout_config: &TimeoutConfig, + timeout_config: &GetObjectTimeoutPolicy, bucket: &str, key: &str, ) -> S3Result> { @@ -1165,19 +1162,13 @@ impl DefaultObjectUsecase { .map_err(|_| s3_error!(InternalError, "disk read semaphore closed"))?; let permit_wait_duration = permit_wait_start.elapsed(); - if wrapper.is_timeout() { - warn!( - bucket = %bucket, - key = %key, - wait_ms = permit_wait_duration.as_millis(), - timeout_secs = timeout_config.get_object_timeout.as_secs(), - elapsed_ms = wrapper.elapsed().as_millis(), - "GetObject request timed out while waiting for disk permit" - ); - - rustfs_io_metrics::record_get_object_timeout(Some("disk_permit"), Some(wrapper.elapsed().as_secs_f64())); - return Err(s3_error!(InternalError, "Request timeout while waiting for disk permit")); - } + Self::ensure_get_object_not_timed_out( + wrapper, + timeout_config, + bucket, + key, + GetObjectTimeoutStage::DiskPermitWait { permit_wait_duration }, + )?; let queue_status = manager.io_queue_status(); let queue_snapshot = GetObjectQueueSnapshot::from_available_permits( @@ -1199,17 +1190,7 @@ impl DefaultObjectUsecase { rustfs_io_metrics::record_io_queue_congestion(); } - if wrapper.is_timeout() { - warn!( - bucket = %bucket, - key = %key, - timeout_secs = timeout_config.get_object_timeout.as_secs(), - elapsed_ms = wrapper.elapsed().as_millis(), - "GetObject request timed out before reading object" - ); - rustfs_io_metrics::record_get_object_timeout(Some("before_read"), Some(wrapper.elapsed().as_secs_f64())); - return Err(s3_error!(InternalError, "Request timeout before reading object")); - } + Self::ensure_get_object_not_timed_out(wrapper, timeout_config, bucket, key, GetObjectTimeoutStage::BeforeRead)?; Ok(GetObjectIoPlanning { _disk_permit: disk_permit, @@ -1274,7 +1255,7 @@ impl DefaultObjectUsecase { req: &S3Request, manager: &'a ConcurrencyManager, wrapper: &RequestTimeoutWrapper, - timeout_config: &TimeoutConfig, + timeout_config: &GetObjectTimeoutPolicy, bucket: &str, key: &str, rs: Option, @@ -2024,7 +2005,7 @@ impl DefaultObjectUsecase { fn finalize_get_object_completion( wrapper: &RequestTimeoutWrapper, - timeout_config: &TimeoutConfig, + timeout_config: &GetObjectTimeoutPolicy, total_duration: Duration, response_content_length: i64, optimal_buffer_size: usize, @@ -2052,6 +2033,57 @@ impl DefaultObjectUsecase { ); } + fn ensure_get_object_not_timed_out( + wrapper: &RequestTimeoutWrapper, + timeout_config: &GetObjectTimeoutPolicy, + bucket: &str, + key: &str, + stage: GetObjectTimeoutStage, + ) -> S3Result<()> { + if !wrapper.is_timeout() { + return Ok(()); + } + + let timeout_secs = timeout_config.get_object_timeout.as_secs(); + let elapsed_ms = wrapper.elapsed().as_millis(); + + match stage { + GetObjectTimeoutStage::BeforeProcessing => { + warn!( + bucket = %bucket, + key = %key, + timeout_secs, + elapsed_ms, + "GetObject request timed out before processing" + ); + Err(s3_error!(InternalError, "Request timeout before processing")) + } + GetObjectTimeoutStage::DiskPermitWait { permit_wait_duration } => { + warn!( + bucket = %bucket, + key = %key, + wait_ms = permit_wait_duration.as_millis(), + timeout_secs, + elapsed_ms, + "GetObject request timed out while waiting for disk permit" + ); + rustfs_io_metrics::record_get_object_timeout(Some("disk_permit"), Some(wrapper.elapsed().as_secs_f64())); + Err(s3_error!(InternalError, "Request timeout while waiting for disk permit")) + } + GetObjectTimeoutStage::BeforeRead => { + warn!( + bucket = %bucket, + key = %key, + timeout_secs, + elapsed_ms, + "GetObject request timed out before reading object" + ); + rustfs_io_metrics::record_get_object_timeout(Some("before_read"), Some(wrapper.elapsed().as_secs_f64())); + Err(s3_error!(InternalError, "Request timeout before reading object")) + } + } + } + async fn finalize_get_object_response( helper: OperationHelper, bucket: &str, diff --git a/rustfs/src/storage/backpressure.rs b/rustfs/src/storage/backpressure.rs index f6ea4be1a..942e31d1c 100644 --- a/rustfs/src/storage/backpressure.rs +++ b/rustfs/src/storage/backpressure.rs @@ -17,9 +17,6 @@ //! This module provides backpressure-aware pipes for object data transfer, //! preventing buffer overflow and memory exhaustion under high concurrency. -// Allow dead_code for public API that may be used by external modules or future features -#![allow(dead_code)] -//! //! # Key Features //! //! - Configurable buffer size with high/low watermarks @@ -39,17 +36,16 @@ //! [High Watermark?] --> Apply Backpressure //! ``` -use std::sync::Arc; use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering}; -use std::time::Instant; +use std::time::{Duration, Instant}; use tokio::io::{DuplexStream, duplex}; use tracing::{debug, warn}; use metrics::counter; -/// Backpressure pipe configuration. -#[derive(Debug, Clone)] -pub struct BackpressureConfig { +/// 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%). @@ -60,7 +56,7 @@ pub struct BackpressureConfig { pub low_watermark: u32, } -impl Default for BackpressureConfig { +impl Default for ObjectPipeBackpressurePolicy { fn default() -> Self { Self { buffer_size: rustfs_config::DEFAULT_OBJECT_DUPLEX_BUFFER_SIZE, @@ -70,7 +66,7 @@ impl Default for BackpressureConfig { } } -impl BackpressureConfig { +impl ObjectPipeBackpressurePolicy { /// Load configuration from environment variables. pub fn from_env() -> Self { let buffer_size = rustfs_utils::get_env_usize( @@ -125,45 +121,61 @@ impl std::fmt::Display for BackpressureState { } } -/// Backpressure event for monitoring. -#[derive(Debug, Clone)] -pub struct BackpressureEvent { - /// Event timestamp. - pub timestamp: Instant, - /// Event type. - pub event_type: BackpressureEventType, - /// Buffer usage at event time. - pub buffer_usage: usize, - /// Buffer capacity. - pub buffer_capacity: usize, - /// Usage percentage. - pub usage_percent: f32, -} - -/// Backpressure event type. -#[derive(Debug, Clone, Copy)] -pub enum BackpressureEventType { - /// Entered high watermark state. - HighWatermarkReached, - /// Exited high watermark state (backpressure released). - HighWatermarkExited, - /// Backpressure applied to producer. - BackpressureApplied, - /// Backpressure released. - BackpressureReleased, -} - -/// Snapshot of backpressure state. -#[derive(Debug, Clone)] -pub struct BackpressureSnapshot { +/// 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 buffer usage in bytes (approximate). - pub buffer_used: usize, - /// Usage percentage. - pub usage_percent: f32, - /// Current state. + /// 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. @@ -176,33 +188,41 @@ pub struct BackpressurePipe { /// Writer end of the duplex pipe. writer: DuplexStream, /// Configuration. - config: BackpressureConfig, + config: ObjectPipeBackpressurePolicy, /// Current buffer usage (approximate, updated on write). - buffer_usage: Arc, + buffer_usage: AtomicUsize, /// Current backpressure state. - state: Arc, // true = in high watermark state + state: AtomicBool, // true = in high watermark state /// Total bytes written. - total_written: Arc, + total_written: AtomicUsize, /// Total bytes read. - total_read: Arc, + 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(BackpressureConfig::from_env()) + Self::with_config(ObjectPipeBackpressurePolicy::from_env()) } /// Create a new backpressure-aware pipe with custom configuration. - pub fn with_config(config: BackpressureConfig) -> Self { + pub fn with_config(config: ObjectPipeBackpressurePolicy) -> Self { let (reader, writer) = duplex(config.buffer_size); + let high_watermark_bytes = config.high_watermark_bytes(); + let low_watermark_bytes = config.low_watermark_bytes(); debug!( buffer_size = config.buffer_size, high_watermark = config.high_watermark, low_watermark = config.low_watermark, - high_watermark_bytes = config.high_watermark_bytes(), - low_watermark_bytes = config.low_watermark_bytes(), + high_watermark_bytes, + low_watermark_bytes, "Created backpressure pipe" ); @@ -210,10 +230,13 @@ impl BackpressurePipe { reader, writer, config, - buffer_usage: Arc::new(AtomicUsize::new(0)), - state: Arc::new(AtomicBool::new(false)), - total_written: Arc::new(AtomicUsize::new(0)), - total_read: Arc::new(AtomicUsize::new(0)), + 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(), } } @@ -234,81 +257,79 @@ impl BackpressurePipe { /// Get current backpressure state. pub fn state(&self) -> BackpressureState { - if self.state.load(Ordering::Relaxed) { + if self.state.load(Ordering::Acquire) { BackpressureState::BackpressureApplied } else { BackpressureState::Normal } } - /// Get current buffer usage snapshot. - pub fn snapshot(&self) -> BackpressureSnapshot { - let buffer_used = self.buffer_usage.load(Ordering::Relaxed); - let usage_percent = if self.config.buffer_size > 0 { - (buffer_used as f64 / self.config.buffer_size as f64) * 100.0 - } else { - 0.0 - }; - - BackpressureSnapshot { + /// Get a compact metadata snapshot for the pipe. + pub fn meta(&self) -> BackpressurePipeMeta { + BackpressurePipeMeta { buffer_capacity: self.config.buffer_size, - buffer_used, - usage_percent: usage_percent as f32, 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::Relaxed); - self.check_high_watermark(); + 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); - self.buffer_usage.fetch_sub(bytes, Ordering::Relaxed); - self.check_low_watermark(); + saturating_sub_atomic(&self.buffer_usage, bytes); + self.update_watermark_state(); } - /// Check if high watermark is reached. - fn check_high_watermark(&self) { - let usage = self.buffer_usage.load(Ordering::Relaxed); - let threshold = self.config.high_watermark_bytes(); + /// 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 usage >= threshold && !self.state.load(Ordering::Relaxed) { - self.state.store(true, Ordering::Relaxed); + if changed { + match next_state { + BackpressureState::HighWatermark => { + counter!("rustfs_backpressure_events_total", "state" => "high_watermark").increment(1); - 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); - warn!( - buffer_usage = usage, - buffer_capacity = self.config.buffer_size, - usage_percent = (usage as f64 / self.config.buffer_size as f64 * 100.0) as u32, - high_watermark = self.config.high_watermark, - "Backpressure: high watermark reached" - ); - } - } - - /// Check if low watermark is reached (backpressure can be released). - fn check_low_watermark(&self) { - let usage = self.buffer_usage.load(Ordering::Relaxed); - let threshold = self.config.low_watermark_bytes(); - - if usage <= threshold && self.state.load(Ordering::Relaxed) { - self.state.store(false, Ordering::Relaxed); - - counter!("rustfs_backpressure_events_total", "state" => "normal").increment(1); - - debug!( - buffer_usage = usage, - buffer_capacity = self.config.buffer_size, - usage_percent = (usage as f64 / self.config.buffer_size as f64 * 100.0) as u32, - low_watermark = self.config.low_watermark, - "Backpressure: returned to normal" - ); + debug!( + buffer_usage = usage, + buffer_capacity = self.config.buffer_size, + usage_percent, + low_watermark = self.config.low_watermark, + "Backpressure: returned to normal" + ); + } + BackpressureState::BackpressureApplied => {} + } } } @@ -340,43 +361,51 @@ impl Default for BackpressurePipe { /// wrap the streams but provides monitoring capabilities. pub struct BackpressureMonitor { /// Configuration. - config: BackpressureConfig, + config: ObjectPipeBackpressurePolicy, /// Current buffer usage. - buffer_usage: Arc, + buffer_usage: AtomicUsize, /// In high watermark state. - in_high_watermark: Arc, + 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(BackpressureConfig::from_env()) + Self::with_config(ObjectPipeBackpressurePolicy::from_env()) } /// Create a new monitor with custom configuration. - pub fn with_config(config: BackpressureConfig) -> Self { + pub fn with_config(config: ObjectPipeBackpressurePolicy) -> Self { + let high_watermark_bytes = config.high_watermark_bytes(); + let low_watermark_bytes = config.low_watermark_bytes(); Self { config, - buffer_usage: Arc::new(AtomicUsize::new(0)), - in_high_watermark: Arc::new(AtomicBool::new(false)), + 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::Relaxed); + self.buffer_usage.fetch_add(bytes, Ordering::Release); self.update_state() } /// Record bytes removed from buffer. pub fn on_read(&self, bytes: usize) -> BackpressureState { - self.buffer_usage.fetch_sub(bytes, Ordering::Relaxed); + 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::Relaxed) { + if self.in_high_watermark.load(Ordering::Acquire) { BackpressureState::HighWatermark } else { BackpressureState::Normal @@ -385,41 +414,46 @@ impl BackpressureMonitor { /// Get current buffer usage. pub fn usage(&self) -> usize { - self.buffer_usage.load(Ordering::Relaxed) + self.buffer_usage.load(Ordering::Acquire) } /// Get usage percentage. pub fn usage_percent(&self) -> f32 { - let usage = self.buffer_usage.load(Ordering::Relaxed); - if self.config.buffer_size > 0 { - (usage as f32 / self.config.buffer_size as f32) * 100.0 - } else { - 0.0 + 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::Relaxed); - let high = self.config.high_watermark_bytes(); - let low = self.config.low_watermark_bytes(); + 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 usage >= high { - if !self.in_high_watermark.swap(true, Ordering::Relaxed) { + if matches!(next_state, BackpressureState::HighWatermark) { + if changed { counter!("rustfs_backpressure_events_total", "state" => "high_watermark").increment(1); - debug!(usage_percent = self.usage_percent() as u32, "Backpressure: entered high watermark"); + debug!(usage_percent, "Backpressure: entered high watermark"); } BackpressureState::HighWatermark - } else if usage <= low { - if self.in_high_watermark.swap(false, Ordering::Relaxed) { + } else { + if changed { counter!("rustfs_backpressure_events_total", "state" => "normal").increment(1); - debug!(usage_percent = self.usage_percent() as u32, "Backpressure: returned to normal"); + debug!(usage_percent, "Backpressure: returned to normal"); } BackpressureState::Normal - } else { - self.state() } } } @@ -436,7 +470,7 @@ mod tests { #[test] fn test_backpressure_config_default() { - let config = BackpressureConfig::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); @@ -444,7 +478,7 @@ mod tests { #[test] fn test_backpressure_config_watermarks() { - let config = BackpressureConfig { + let config = ObjectPipeBackpressurePolicy { buffer_size: 1000, high_watermark: 80, low_watermark: 50, @@ -462,7 +496,7 @@ mod tests { #[test] fn test_backpressure_monitor() { - let config = BackpressureConfig { + let config = ObjectPipeBackpressurePolicy { buffer_size: 1000, high_watermark: 80, low_watermark: 50, @@ -471,14 +505,18 @@ mod tests { // 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] @@ -486,5 +524,28 @@ mod tests { 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/concurrency/io_schedule.rs b/rustfs/src/storage/concurrency/io_schedule.rs index 3d7a271c5..ff8eb3268 100644 --- a/rustfs/src/storage/concurrency/io_schedule.rs +++ b/rustfs/src/storage/concurrency/io_schedule.rs @@ -32,6 +32,7 @@ use rustfs_config::{KI_B, MI_B}; use rustfs_io_core::io_profile::{AccessPattern, StorageMedia, StorageProfile}; +use rustfs_io_core::{IoPriorityQueueConfig as CoreIoPriorityQueueConfig, IoSchedulerConfig as CoreIoSchedulerConfig}; use rustfs_io_metrics::bandwidth::{BandwidthSnapshot, BandwidthTier}; use std::sync::atomic::{AtomicU64, AtomicUsize, Ordering}; use std::time::Duration; @@ -129,8 +130,8 @@ impl IoPriority { pub fn from_size(size: i64) -> Self { Self::from_size_with_thresholds( size, - IoSchedulerConfig::default().high_priority_size_threshold, - IoSchedulerConfig::default().low_priority_size_threshold, + rustfs_config::DEFAULT_OBJECT_IO_HIGH_PRIORITY_SIZE_THRESHOLD, + rustfs_config::DEFAULT_OBJECT_IO_LOW_PRIORITY_SIZE_THRESHOLD, ) } @@ -380,6 +381,32 @@ impl IoSchedulerConfig { ), } } + + /// Convert storage-layer scheduler config to io-core scheduler config. + /// + /// This keeps storage-specific policy loading in place while allowing + /// core scheduling components to consume a normalized shared config shape. + pub fn to_core_config(&self) -> CoreIoSchedulerConfig { + CoreIoSchedulerConfig { + max_concurrent_reads: self.max_concurrent_reads, + high_priority_size_threshold: self.high_priority_size_threshold, + low_priority_size_threshold: self.low_priority_size_threshold, + queue_high_capacity: self.queue_high_capacity, + queue_normal_capacity: self.queue_normal_capacity, + queue_low_capacity: self.queue_low_capacity, + starvation_prevention_interval_ms: self.starvation_prevention_interval_ms, + starvation_threshold_secs: self.starvation_threshold_secs, + load_sample_window: self.load_sample_window, + load_high_threshold_ms: self.load_high_threshold_ms, + load_low_threshold_ms: self.load_low_threshold_ms, + enable_priority: self.enable_priority, + storage_detection_enabled: self.storage_detection_enabled, + base_buffer_size: rustfs_config::DEFAULT_OBJECT_IO_BUFFER_SIZE, + max_buffer_size: MI_B, + min_buffer_size: 32 * KI_B, + ..Default::default() + } + } } /// I/O queue status for monitoring. @@ -1271,27 +1298,38 @@ impl IoLoadMetrics { self.observation_count.load(Ordering::Relaxed) } } -pub fn get_concurrency_aware_buffer_size(file_size: i64, base_buffer_size: usize) -> usize { - let concurrent_requests = ACTIVE_GET_REQUESTS.load(Ordering::Relaxed); +#[derive(Debug, Clone, Copy)] +struct ConcurrencyThresholds { + medium: usize, + high: usize, +} - // Record concurrent request metrics - { - use metrics::gauge; - gauge!("rustfs_concurrent_get_requests").set(concurrent_requests as f64); +fn load_concurrency_thresholds() -> ConcurrencyThresholds { + ConcurrencyThresholds { + medium: rustfs_utils::get_env_usize( + rustfs_config::ENV_OBJECT_MEDIUM_CONCURRENCY_THRESHOLD, + rustfs_config::DEFAULT_OBJECT_MEDIUM_CONCURRENCY_THRESHOLD, + ), + high: rustfs_utils::get_env_usize( + rustfs_config::ENV_OBJECT_HIGH_CONCURRENCY_THRESHOLD, + rustfs_config::DEFAULT_OBJECT_HIGH_CONCURRENCY_THRESHOLD, + ), } +} + +fn compute_concurrency_aware_buffer_size( + file_size: i64, + base_buffer_size: usize, + concurrent_requests: usize, + thresholds: ConcurrencyThresholds, +) -> usize { + let medium_threshold = thresholds.medium; + let high_threshold = thresholds.high; // For low concurrency, use the base buffer size for maximum throughput if concurrent_requests <= 1 { return base_buffer_size; } - let medium_threshold = rustfs_utils::get_env_usize( - rustfs_config::ENV_OBJECT_MEDIUM_CONCURRENCY_THRESHOLD, - rustfs_config::DEFAULT_OBJECT_MEDIUM_CONCURRENCY_THRESHOLD, - ); - let high_threshold = rustfs_utils::get_env_usize( - rustfs_config::ENV_OBJECT_HIGH_CONCURRENCY_THRESHOLD, - rustfs_config::DEFAULT_OBJECT_HIGH_CONCURRENCY_THRESHOLD, - ); // Calculate adaptive multiplier based on concurrency level let adaptive_multiplier = if concurrent_requests <= 2 { @@ -1327,6 +1365,18 @@ pub fn get_concurrency_aware_buffer_size(file_size: i64, base_buffer_size: usize adjusted_size.clamp(min_buffer, max_buffer) } +pub fn get_concurrency_aware_buffer_size(file_size: i64, base_buffer_size: usize) -> usize { + let concurrent_requests = ACTIVE_GET_REQUESTS.load(Ordering::Relaxed); + + // Record concurrent request metrics + { + use metrics::gauge; + gauge!("rustfs_concurrent_get_requests").set(concurrent_requests as f64); + } + + compute_concurrency_aware_buffer_size(file_size, base_buffer_size, concurrent_requests, load_concurrency_thresholds()) +} + /// Advanced concurrency-aware buffer sizing with file size optimization /// /// This enhanced version considers both concurrency level and file size patterns @@ -1353,6 +1403,7 @@ pub fn get_concurrency_aware_buffer_size(file_size: i64, base_buffer_size: usize /// ``` pub fn get_advanced_buffer_size(file_size: i64, base_buffer_size: usize, is_sequential: bool) -> usize { let concurrent_requests = ACTIVE_GET_REQUESTS.load(Ordering::Relaxed); + let thresholds = load_concurrency_thresholds(); // For very small files, use smaller buffers regardless of concurrency // Replace manual max/min chain with clamp @@ -1361,15 +1412,10 @@ pub fn get_advanced_buffer_size(file_size: i64, base_buffer_size: usize, is_sequ } // Base calculation from standard function - let standard_size = get_concurrency_aware_buffer_size(file_size, base_buffer_size); - let medium_threshold = rustfs_utils::get_env_usize( - rustfs_config::ENV_OBJECT_MEDIUM_CONCURRENCY_THRESHOLD, - rustfs_config::DEFAULT_OBJECT_MEDIUM_CONCURRENCY_THRESHOLD, - ); - let high_threshold = rustfs_utils::get_env_usize( - rustfs_config::ENV_OBJECT_HIGH_CONCURRENCY_THRESHOLD, - rustfs_config::DEFAULT_OBJECT_HIGH_CONCURRENCY_THRESHOLD, - ); + let standard_size = compute_concurrency_aware_buffer_size(file_size, base_buffer_size, concurrent_requests, thresholds); + + let medium_threshold = thresholds.medium; + let high_threshold = thresholds.high; // For sequential reads, we can be more aggressive with buffer sizes if is_sequential && concurrent_requests <= medium_threshold { return ((standard_size as f64 * 1.5) as usize).min(2 * MI_B); @@ -1523,6 +1569,28 @@ impl IoPriorityQueueConfig { ), } } + + /// Convert storage-layer queue config to io-core queue config. + pub fn to_core_config(&self) -> CoreIoPriorityQueueConfig { + CoreIoPriorityQueueConfig { + high_capacity: self.queue_high_capacity, + normal_capacity: self.queue_normal_capacity, + low_capacity: self.queue_low_capacity, + starvation_interval: Duration::from_millis(self.starvation_prevention_interval_ms), + starvation_threshold: Duration::from_secs(self.starvation_threshold_secs), + } + } + + /// Build queue config directly from a scheduler config. + pub fn from_scheduler_config(config: &IoSchedulerConfig) -> Self { + Self { + queue_high_capacity: config.queue_high_capacity, + queue_normal_capacity: config.queue_normal_capacity, + queue_low_capacity: config.queue_low_capacity, + starvation_prevention_interval_ms: config.starvation_prevention_interval_ms, + starvation_threshold_secs: config.starvation_threshold_secs, + } + } } impl IoPriorityQueue { @@ -1848,9 +1916,8 @@ pub fn get_buffer_size_opt_in(file_size: i64) -> usize { mod tests { use super::*; use serial_test::serial; - use tokio::test; - #[test] + #[tokio::test] #[serial] async fn test_io_priority_queue_basic() { let config = IoPriorityQueueConfig::default(); @@ -1869,7 +1936,7 @@ mod tests { assert_eq!(queue.len().await, 3); } - #[test] + #[tokio::test] #[serial] async fn test_io_priority_queue_dequeue_order() { let config = IoPriorityQueueConfig::default(); @@ -1897,7 +1964,7 @@ mod tests { assert!(queue.is_empty().await); } - #[test] + #[tokio::test] #[serial] async fn test_io_priority_queue_status() { let config = IoPriorityQueueConfig::default(); @@ -1915,7 +1982,7 @@ mod tests { assert_eq!(status.low_priority_waiting, 1); } - #[test] + #[tokio::test] #[serial] async fn test_io_priority_queue_starvation_prevention() { let config = IoPriorityQueueConfig { @@ -1939,7 +2006,7 @@ mod tests { assert_eq!(priority, IoPriority::Normal); } - #[test] + #[tokio::test] #[serial] async fn test_io_priority_from_size() { // High priority: < 1MB @@ -1955,7 +2022,7 @@ mod tests { assert_eq!(IoPriority::from_size(100 * 1024 * 1024), IoPriority::Low); } - #[test] + #[tokio::test] #[serial] async fn test_io_load_level_from_wait_duration() { use std::time::Duration; @@ -1973,7 +2040,7 @@ mod tests { assert_eq!(IoLoadLevel::from_wait_duration(Duration::from_millis(300)), IoLoadLevel::Critical); } - #[test] + #[tokio::test] #[serial] async fn test_io_scheduler_config_default() { let config = IoSchedulerConfig::default(); @@ -1987,7 +2054,54 @@ mod tests { assert_eq!(config.starvation_threshold_secs, 5); } - #[test] + #[tokio::test] + #[serial] + async fn test_io_scheduler_config_to_core_config() { + let config = IoSchedulerConfig::default(); + let core = config.to_core_config(); + assert_eq!(core.max_concurrent_reads, config.max_concurrent_reads); + assert_eq!(core.high_priority_size_threshold, config.high_priority_size_threshold); + assert_eq!(core.low_priority_size_threshold, config.low_priority_size_threshold); + assert_eq!(core.queue_high_capacity, config.queue_high_capacity); + assert_eq!(core.queue_normal_capacity, config.queue_normal_capacity); + assert_eq!(core.queue_low_capacity, config.queue_low_capacity); + assert_eq!(core.load_high_threshold_ms, config.load_high_threshold_ms); + assert_eq!(core.load_low_threshold_ms, config.load_low_threshold_ms); + assert_eq!(core.enable_priority, config.enable_priority); + } + + #[tokio::test] + #[serial] + async fn test_io_priority_queue_config_to_core_config() { + let config = IoPriorityQueueConfig::default(); + let core = config.to_core_config(); + assert_eq!(core.high_capacity, config.queue_high_capacity); + assert_eq!(core.normal_capacity, config.queue_normal_capacity); + assert_eq!(core.low_capacity, config.queue_low_capacity); + assert_eq!(core.starvation_interval, Duration::from_millis(config.starvation_prevention_interval_ms)); + assert_eq!(core.starvation_threshold, Duration::from_secs(config.starvation_threshold_secs)); + } + + #[tokio::test] + #[serial] + async fn test_io_priority_queue_config_from_scheduler_config() { + let scheduler_config = IoSchedulerConfig { + queue_high_capacity: 128, + queue_normal_capacity: 256, + queue_low_capacity: 512, + starvation_prevention_interval_ms: 2000, + starvation_threshold_secs: 120, + ..Default::default() + }; + let config = IoPriorityQueueConfig::from_scheduler_config(&scheduler_config); + assert_eq!(config.queue_high_capacity, 128); + assert_eq!(config.queue_normal_capacity, 256); + assert_eq!(config.queue_low_capacity, 512); + assert_eq!(config.starvation_prevention_interval_ms, 2000); + assert_eq!(config.starvation_threshold_secs, 120); + } + + #[tokio::test] #[serial] async fn test_io_priority_metrics() { let metrics = IoPriorityMetrics::new(); @@ -2014,7 +2128,7 @@ mod tests { // Multi-Factor Strategy Tests // ============================================ - #[test] + #[tokio::test] #[serial] async fn test_multi_factor_strategy_nvme_sequential_low_load() { // NVMe + Sequential + Low load = maximum buffer size @@ -2041,7 +2155,7 @@ mod tests { assert_eq!(strategy.bandwidth_tier, BandwidthTier::High); } - #[test] + #[tokio::test] #[serial] async fn test_multi_factor_strategy_hdd_random_high_load() { // HDD + Random + High load = conservative buffer size @@ -2068,7 +2182,7 @@ mod tests { assert!(strategy.bandwidth_limited, "Low bandwidth should be marked"); } - #[test] + #[tokio::test] #[serial] async fn test_multi_factor_strategy_ssd_mixed_medium_load() { // SSD + Mixed + Medium load = moderate buffer @@ -2096,7 +2210,7 @@ mod tests { assert_eq!(strategy.access_pattern, AccessPattern::Mixed); } - #[test] + #[tokio::test] #[serial] async fn test_multi_factor_strategy_critical_load_disables_features() { // Any media + Critical load = minimal features @@ -2121,7 +2235,7 @@ mod tests { assert!(strategy.buffer_size < 200 * 1024, "Critical load should reduce buffer"); } - #[test] + #[tokio::test] #[serial] async fn test_multi_factor_strategy_buffer_cap_enforcement() { // Test that storage media caps are enforced @@ -2146,7 +2260,7 @@ mod tests { assert!(strategy.debug_info.buffer_cap_applied, "Buffer cap should be applied"); } - #[test] + #[tokio::test] #[serial] async fn test_multi_factor_strategy_bandwidth_low_reduces_buffer() { // Low bandwidth should reduce buffer @@ -2170,7 +2284,7 @@ mod tests { assert!(strategy.buffer_size < context.base_buffer_size, "Low bandwidth should reduce buffer"); } - #[test] + #[tokio::test] #[serial] async fn test_multi_factor_strategy_high_concurrency_reduction() { // High concurrency should reduce buffer @@ -2193,7 +2307,7 @@ mod tests { assert!(strategy.buffer_size < context.base_buffer_size, "High concurrency should reduce buffer"); } - #[test] + #[tokio::test] #[serial] async fn test_multi_factor_strategy_sequential_boost() { // Sequential reads should get boost @@ -2235,7 +2349,7 @@ mod tests { } } - #[test] + #[tokio::test] #[serial] async fn test_multi_factor_strategy_unknown_media_conservative() { // Unknown media should be conservative @@ -2261,7 +2375,7 @@ mod tests { ); } - #[test] + #[tokio::test] #[serial] async fn test_multi_factor_strategy_priority_classification() { // Test priority classification based on file size @@ -2308,7 +2422,7 @@ mod tests { assert_eq!(large_strategy.priority, IoPriority::Low); } - #[test] + #[tokio::test] #[serial] async fn test_multi_factor_strategy_readahead_decision_matrix() { // Test readahead enable/disable logic @@ -2394,7 +2508,7 @@ mod tests { } } - #[test] + #[tokio::test] #[serial] async fn test_multi_factor_strategy_buffer_multiplier_stages() { // Test that all multiplier stages are applied @@ -2429,7 +2543,7 @@ mod tests { assert!(strategy.should_reduce_for_bandwidth); } - #[test] + #[tokio::test] #[serial] async fn test_multi_factor_strategy_compatibility_path() { // Test that compatibility path (from_wait_duration) still works diff --git a/rustfs/src/storage/concurrency/manager.rs b/rustfs/src/storage/concurrency/manager.rs index 3a83644c9..c847cf5c5 100644 --- a/rustfs/src/storage/concurrency/manager.rs +++ b/rustfs/src/storage/concurrency/manager.rs @@ -121,14 +121,8 @@ impl ConcurrencyManager { // Keep 1000 samples for P95/P99 calculation let metrics_collector = Arc::new(MetricsCollector::new(performance_metrics, 1000)); - // Build priority queue config - let queue_config = IoPriorityQueueConfig { - queue_high_capacity: scheduler_config.queue_high_capacity, - queue_normal_capacity: scheduler_config.queue_normal_capacity, - queue_low_capacity: scheduler_config.queue_low_capacity, - starvation_prevention_interval_ms: scheduler_config.starvation_prevention_interval_ms, - starvation_threshold_secs: scheduler_config.starvation_threshold_secs, - }; + // Build queue config directly from scheduler config. + let queue_config = IoPriorityQueueConfig::from_scheduler_config(&scheduler_config); Self { disk_read_semaphore: Arc::new(Semaphore::new(max_disk_reads)), diff --git a/rustfs/src/storage/concurrent_fix_test.rs b/rustfs/src/storage/concurrent_fix_test.rs index b64c4aaa3..2fedeced1 100644 --- a/rustfs/src/storage/concurrent_fix_test.rs +++ b/rustfs/src/storage/concurrent_fix_test.rs @@ -19,13 +19,13 @@ #[cfg(test)] mod tests { - use crate::storage::backpressure::{BackpressureConfig, BackpressureMonitor, BackpressureState}; + use crate::storage::backpressure::{BackpressureMonitor, BackpressureState, ObjectPipeBackpressurePolicy}; use crate::storage::concurrency::{IoLoadLevel, IoPriority}; use crate::storage::deadlock_detector::{ - DeadlockDetector, DeadlockDetectorConfig, LockInfo, LockType, RequestResourceTracker, + DeadlockDetector, LockInfo, LockType, RequestHangDetectionPolicy, RequestResourceTracker, }; use crate::storage::lock_optimizer::{LockOptimizeConfig, LockOptimizer, LockStats}; - use crate::storage::timeout_wrapper::{RequestTimeoutWrapper, TimedGetObjectResult, TimeoutConfig}; + use crate::storage::timeout_wrapper::{GetObjectTimeoutPolicy, RequestTimeoutWrapper, TimedGetObjectResult}; use std::time::Duration; // ============================================ @@ -34,7 +34,7 @@ mod tests { #[tokio::test] async fn test_timeout_wrapper_completes_within_timeout() { - let config = TimeoutConfig { + let config = GetObjectTimeoutPolicy { get_object_timeout: Duration::from_secs(5), ..Default::default() }; @@ -52,7 +52,7 @@ mod tests { #[tokio::test] async fn test_timeout_wrapper_times_out() { - let config = TimeoutConfig { + let config = GetObjectTimeoutPolicy { get_object_timeout: Duration::from_millis(50), ..Default::default() }; @@ -76,7 +76,7 @@ mod tests { #[tokio::test] async fn test_timeout_wrapper_returns_error() { - let config = TimeoutConfig { + let config = GetObjectTimeoutPolicy { get_object_timeout: Duration::from_secs(5), ..Default::default() }; @@ -94,7 +94,7 @@ mod tests { #[tokio::test] async fn test_timeout_wrapper_disabled() { - let config = TimeoutConfig { + let config = GetObjectTimeoutPolicy { get_object_timeout: Duration::ZERO, ..Default::default() }; @@ -120,7 +120,7 @@ mod tests { #[test] fn test_backpressure_config_defaults() { - let config = BackpressureConfig::default(); + 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); @@ -128,7 +128,7 @@ mod tests { #[test] fn test_backpressure_monitor_state_transitions() { - let config = BackpressureConfig { + let config = ObjectPipeBackpressurePolicy { buffer_size: 1000, high_watermark: 80, low_watermark: 50, @@ -149,7 +149,7 @@ mod tests { #[test] fn test_backpressure_usage_percent() { - let config = BackpressureConfig { + let config = ObjectPipeBackpressurePolicy { buffer_size: 1000, high_watermark: 80, low_watermark: 50, @@ -229,7 +229,7 @@ mod tests { #[test] fn test_deadlock_detector_config_defaults() { - let config = DeadlockDetectorConfig::default(); + let config = RequestHangDetectionPolicy::default(); assert!(!config.enabled); // Disabled by default assert_eq!(config.check_interval, Duration::from_secs(5)); assert_eq!(config.hang_threshold, Duration::from_secs(10)); @@ -255,7 +255,7 @@ mod tests { #[test] fn test_deadlock_detector_registration() { - let config = DeadlockDetectorConfig { + let config = RequestHangDetectionPolicy { enabled: true, ..Default::default() }; diff --git a/rustfs/src/storage/deadlock_detector.rs b/rustfs/src/storage/deadlock_detector.rs index cde095f9e..0e35a332c 100644 --- a/rustfs/src/storage/deadlock_detector.rs +++ b/rustfs/src/storage/deadlock_detector.rs @@ -40,9 +40,9 @@ //! # Usage //! //! ```ignore -//! use crate::storage::deadlock_detector::{DeadlockDetector, DeadlockDetectorConfig}; +//! use crate::storage::deadlock_detector::{DeadlockDetector, RequestHangDetectionPolicy}; //! -//! let config = DeadlockDetectorConfig::from_env(); +//! let config = RequestHangDetectionPolicy::from_env(); //! let detector = DeadlockDetector::new(config); //! detector.start(); //! @@ -66,6 +66,7 @@ use tokio::sync::broadcast; use tracing::{debug, error, warn}; use metrics::counter; +use rustfs_io_core::DeadlockDetectorConfig as CoreDeadlockConfig; /// Request identifier type. pub type RequestId = String; @@ -73,9 +74,9 @@ pub type RequestId = String; /// Lock identifier type. pub type LockId = String; -/// Deadlock detector configuration. +/// Request-level hang and deadlock diagnosis policy. #[derive(Debug, Clone)] -pub struct DeadlockDetectorConfig { +pub struct RequestHangDetectionPolicy { /// Whether deadlock detection is enabled. pub enabled: bool, /// Detection check interval. @@ -86,7 +87,7 @@ pub struct DeadlockDetectorConfig { pub capture_backtrace: bool, } -impl Default for DeadlockDetectorConfig { +impl Default for RequestHangDetectionPolicy { fn default() -> Self { Self { enabled: rustfs_config::DEFAULT_OBJECT_DEADLOCK_DETECTION_ENABLE, @@ -97,7 +98,7 @@ impl Default for DeadlockDetectorConfig { } } -impl DeadlockDetectorConfig { +impl RequestHangDetectionPolicy { /// Load configuration from environment variables. pub fn from_env() -> Self { let enabled = rustfs_utils::get_env_bool( @@ -120,6 +121,15 @@ impl DeadlockDetectorConfig { capture_backtrace: false, } } + + /// Convert the request-level policy into the shared io-core deadlock config. + pub fn to_core_config(&self) -> CoreDeadlockConfig { + CoreDeadlockConfig { + enabled: self.enabled, + detection_interval: self.check_interval, + max_hold_time: self.hang_threshold, + } + } } /// Lock information for tracking. @@ -247,7 +257,7 @@ pub struct ResourceUsage { /// Deadlock detector. pub struct DeadlockDetector { /// Configuration. - config: DeadlockDetectorConfig, + config: RequestHangDetectionPolicy, /// Active request trackers. requests: Arc>>, /// Detection task handle. @@ -262,7 +272,7 @@ pub struct DeadlockDetector { impl DeadlockDetector { /// Create a new deadlock detector. - pub fn new(config: DeadlockDetectorConfig) -> Self { + pub fn new(config: RequestHangDetectionPolicy) -> Self { let (shutdown_tx, _) = broadcast::channel(1); Self { @@ -426,7 +436,7 @@ impl DeadlockDetector { /// Detect deadlock cycles in the lock wait graph. fn detect_cycle( requests: &Arc>>, - config: &DeadlockDetectorConfig, + config: &RequestHangDetectionPolicy, deadlocks_detected: &Arc, ) { let requests_guard = requests.read().unwrap(); @@ -499,13 +509,16 @@ impl DeadlockDetector { /// Find a cycle in the wait graph using DFS. fn find_cycle(edges: &[WaitGraphEdge]) -> Option> { - // Build adjacency list + if edges.is_empty() { + return None; + } + + // Build adjacency list: from -> [to] let mut graph: HashMap<&RequestId, Vec<&RequestId>> = HashMap::new(); for edge in edges { graph.entry(&edge.from).or_default().push(&edge.to); } - // DFS with path tracking let mut visited: HashSet<&RequestId> = HashSet::new(); let mut path: Vec<&RequestId> = Vec::new(); let mut path_set: HashSet<&RequestId> = HashSet::new(); @@ -514,7 +527,6 @@ impl DeadlockDetector { if visited.contains(start) { continue; } - if Self::dfs_find_cycle(start, &graph, &mut visited, &mut path, &mut path_set) { return Some(path.iter().map(|s| (*s).clone()).collect()); } @@ -543,7 +555,6 @@ impl DeadlockDetector { path.drain(0..cycle_start); return true; } - if !visited.contains(neighbor) && Self::dfs_find_cycle(neighbor, graph, visited, path, path_set) { return true; } @@ -569,7 +580,7 @@ static DEADLOCK_DETECTOR: std::sync::OnceLock> = std::sync pub fn get_deadlock_detector() -> Arc { DEADLOCK_DETECTOR .get_or_init(|| { - let config = DeadlockDetectorConfig::from_env(); + let config = RequestHangDetectionPolicy::from_env(); Arc::new(DeadlockDetector::new(config)) }) .clone() @@ -595,7 +606,7 @@ mod tests { #[test] fn test_deadlock_detector_config_default() { - let config = DeadlockDetectorConfig::default(); + let config = RequestHangDetectionPolicy::default(); assert!(!config.enabled); assert_eq!(config.check_interval, Duration::from_secs(5)); assert_eq!(config.hang_threshold, Duration::from_secs(10)); @@ -621,7 +632,7 @@ mod tests { #[test] fn test_deadlock_detector_registration() { - let config = DeadlockDetectorConfig { + let config = RequestHangDetectionPolicy { enabled: true, ..Default::default() }; diff --git a/rustfs/src/storage/timeout_wrapper.rs b/rustfs/src/storage/timeout_wrapper.rs index 5c814d3fa..c5024812b 100644 --- a/rustfs/src/storage/timeout_wrapper.rs +++ b/rustfs/src/storage/timeout_wrapper.rs @@ -33,23 +33,14 @@ //! //! - Configurable request-level timeout (default 30 seconds) //! - Automatic cancellation of sub-tasks on timeout -//! - Resource cleanup on timeout (locks, memory, file handles) -//! - Timeout metrics emitted through `rustfs-io-metrics` - -// Allow dead_code for public API that may be used by external modules or future features -#![allow(dead_code)] -//! -//! - Configurable request-level timeout (default 30 seconds) -//! - Automatic cancellation of sub-tasks on timeout -//! - Resource cleanup on timeout (locks, memory, file handles) //! - Timeout metrics emitted through `rustfs-io-metrics` //! //! # Usage //! //! ```ignore -//! use crate::storage::timeout_wrapper::{RequestTimeoutWrapper, TimeoutConfig}; +//! use crate::storage::timeout_wrapper::{GetObjectTimeoutPolicy, RequestTimeoutWrapper}; //! -//! let config = TimeoutConfig::from_env(); +//! let config = GetObjectTimeoutPolicy::from_env(); //! let wrapper = RequestTimeoutWrapper::new(config); //! //! match wrapper.execute_with_timeout(|cancel_token| async move { @@ -66,23 +57,13 @@ use std::time::{Duration, Instant}; use tokio_util::sync::CancellationToken; use tracing::{debug, warn}; -// Re-export types from rustfs_io_core for convenience - -/// Timeout configuration for GetObject requests. +/// Request-level timeout policy for GetObject. #[derive(Debug, Clone)] -pub struct TimeoutConfig { +pub struct GetObjectTimeoutPolicy { /// GetObject request overall timeout (default 30s). /// After this duration, the request is cancelled and returns 504. pub get_object_timeout: Duration, - /// Lock acquisition timeout (default 5s). - /// Time to wait for a lock before giving up. - pub lock_acquire_timeout: Duration, - - /// Disk read operation timeout (default 10s). - /// Individual disk read operations that exceed this are cancelled. - pub disk_read_timeout: Duration, - /// Enable dynamic timeout calculation based on object size pub enable_dynamic_timeout: bool, @@ -96,12 +77,14 @@ pub struct TimeoutConfig { pub max_timeout: Duration, } -impl Default for TimeoutConfig { +/// Backward-compatible alias for external callers using the previous name. +#[deprecated(note = "use GetObjectTimeoutPolicy instead")] +pub type TimeoutConfig = GetObjectTimeoutPolicy; + +impl Default for GetObjectTimeoutPolicy { fn default() -> Self { Self { get_object_timeout: Duration::from_secs(rustfs_config::DEFAULT_OBJECT_GET_TIMEOUT), - lock_acquire_timeout: Duration::from_secs(rustfs_config::DEFAULT_OBJECT_LOCK_ACQUIRE_TIMEOUT), - disk_read_timeout: Duration::from_secs(rustfs_config::DEFAULT_OBJECT_DISK_READ_TIMEOUT), enable_dynamic_timeout: rustfs_config::DEFAULT_OBJECT_DYNAMIC_TIMEOUT_ENABLE, bytes_per_second: rustfs_config::DEFAULT_OBJECT_BYTES_PER_SECOND, min_timeout: Duration::from_secs(rustfs_config::DEFAULT_OBJECT_MIN_TIMEOUT), @@ -110,21 +93,11 @@ impl Default for TimeoutConfig { } } -impl TimeoutConfig { +impl GetObjectTimeoutPolicy { /// Load configuration from environment variables. pub fn from_env() -> Self { let get_object_timeout = rustfs_utils::get_env_u64(rustfs_config::ENV_OBJECT_GET_TIMEOUT, rustfs_config::DEFAULT_OBJECT_GET_TIMEOUT); - let lock_acquire_timeout = rustfs_utils::get_env_u64( - rustfs_config::ENV_OBJECT_LOCK_ACQUIRE_TIMEOUT, - rustfs_config::DEFAULT_OBJECT_LOCK_ACQUIRE_TIMEOUT, - ); - let disk_read_timeout = rustfs_utils::get_env_u64( - rustfs_config::ENV_OBJECT_DISK_READ_TIMEOUT, - rustfs_config::DEFAULT_OBJECT_DISK_READ_TIMEOUT, - ); - - // Dynamic timeout settings let enable_dynamic_timeout = rustfs_utils::get_env_bool( rustfs_config::ENV_OBJECT_DYNAMIC_TIMEOUT_ENABLE, rustfs_config::DEFAULT_OBJECT_DYNAMIC_TIMEOUT_ENABLE, @@ -138,8 +111,6 @@ impl TimeoutConfig { Self { get_object_timeout: Duration::from_secs(get_object_timeout), - lock_acquire_timeout: Duration::from_secs(lock_acquire_timeout), - disk_read_timeout: Duration::from_secs(disk_read_timeout), enable_dynamic_timeout, bytes_per_second, min_timeout: Duration::from_secs(min_timeout_secs), @@ -158,12 +129,17 @@ impl TimeoutConfig { return self.get_object_timeout; } - // Calculate timeout based on expected transfer speed - // Add 50% buffer for network overhead and system load - let estimated_seconds = (object_size / self.bytes_per_second) * 3 / 2; - - // Ensure at least 1 second - let estimated_duration = Duration::from_secs(estimated_seconds.max(1)); + // Keep storage-layer dynamic-timeout semantics local so request policy + // bounds remain authoritative. We preserve the historical 1.5x envelope: + // object_size / (0.8 * bps) * 1.2 == object_size * 1.5 / bps. + // + // Use integer math to avoid float->Duration conversion panics on extreme + // values and keep behavior predictable under saturation. + let bytes_per_second = self.bytes_per_second.max(1); + let numerator = (object_size as u128).saturating_mul(3); + let denominator = (bytes_per_second as u128).saturating_mul(2); + let estimated_secs = numerator.checked_div(denominator).unwrap_or(u128::MAX); + let estimated_duration = Duration::from_secs(estimated_secs.min(u64::MAX as u128) as u64); // Clamp to min/max bounds estimated_duration @@ -220,59 +196,74 @@ pub enum TimedGetObjectResult { } /// Request timeout wrapper for async operations. -#[derive(Debug)] pub struct RequestTimeoutWrapper { /// Configuration. - config: TimeoutConfig, + config: GetObjectTimeoutPolicy, /// Request start time. start_time: Instant, + /// Optional operation size hint for dynamic timeout decisions. + operation_size: Option, /// Cancellation token for propagating cancellation to sub-tasks. cancel_token: CancellationToken, /// Request ID for logging/metrics. request_id: String, } +impl std::fmt::Debug for RequestTimeoutWrapper { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("RequestTimeoutWrapper") + .field("config", &self.config) + .field("elapsed", &self.elapsed()) + .field("request_id", &self.request_id) + .finish() + } +} + impl RequestTimeoutWrapper { /// Create a new timeout wrapper with the given configuration. /// /// Note: This uses a sentinel request_id. Prefer `with_request_id()` to pass /// the canonical request-id from `RequestContext`. - pub fn new(config: TimeoutConfig) -> Self { - Self { - config, - start_time: Instant::now(), - cancel_token: CancellationToken::new(), - request_id: "no-request-id".to_string(), - } + pub fn new(config: GetObjectTimeoutPolicy) -> Self { + Self::new_with_parts(config, None, CancellationToken::new(), "no-request-id".to_string()) } /// Create a new timeout wrapper with a specific request ID. - pub fn with_request_id(config: TimeoutConfig, request_id: impl Into) -> Self { - Self { - config, - start_time: Instant::now(), - cancel_token: CancellationToken::new(), - request_id: request_id.into(), - } + pub fn with_request_id(config: GetObjectTimeoutPolicy, request_id: impl Into) -> Self { + Self::new_with_parts(config, None, CancellationToken::new(), request_id.into()) } /// Create a new timeout wrapper with operation size for dynamic timeout calculation. /// /// Note: This uses a sentinel request_id. Prefer `with_request_id()` to pass /// the canonical request-id from `RequestContext`. - pub fn with_operation_size(config: TimeoutConfig, operation_size: Option) -> Self { - let _ = operation_size; - Self { - config, - start_time: Instant::now(), - cancel_token: CancellationToken::new(), - request_id: "no-request-id".to_string(), - } + pub fn with_operation_size(config: GetObjectTimeoutPolicy, operation_size: Option) -> Self { + Self::new_with_parts(config, operation_size, CancellationToken::new(), "no-request-id".to_string()) } /// Get the configured timeout for this operation - pub fn get_timeout(&self, operation_size: Option) -> Duration { - self.config.get_timeout_for_operation(operation_size) + pub fn get_timeout(&self, operation_size_hint: Option) -> Duration { + self.config + .get_timeout_for_operation(self.effective_operation_size(operation_size_hint)) + } + + fn new_with_parts( + config: GetObjectTimeoutPolicy, + operation_size: Option, + cancel_token: CancellationToken, + request_id: String, + ) -> Self { + Self { + config, + start_time: Instant::now(), + operation_size, + cancel_token, + request_id, + } + } + + fn effective_operation_size(&self, operation_size_hint: Option) -> Option { + operation_size_hint.or(self.operation_size) } /// Get the request ID. @@ -293,7 +284,7 @@ impl RequestTimeoutWrapper { /// Check if the timeout has been exceeded. pub fn is_timeout(&self) -> bool { - self.config.is_timeout_enabled() && self.elapsed() >= self.config.get_object_timeout + self.config.is_timeout_enabled() && self.start_time.elapsed() >= self.get_timeout(None) } /// Get elapsed time since the request started. @@ -312,8 +303,10 @@ impl RequestTimeoutWrapper { if !self.config.is_timeout_enabled() { return None; } - let timeout = self.config.get_timeout_for_operation(operation_size); - let remaining = timeout.saturating_sub(self.elapsed()); + let timeout = self + .config + .get_timeout_for_operation(self.effective_operation_size(operation_size)); + let remaining = timeout.saturating_sub(self.start_time.elapsed()); if remaining == Duration::ZERO { None } else { Some(remaining) } } @@ -322,8 +315,10 @@ impl RequestTimeoutWrapper { if !self.config.is_timeout_enabled() { return false; } - let timeout = self.config.get_timeout_for_operation(operation_size); - self.elapsed() >= timeout + let timeout = self + .config + .get_timeout_for_operation(self.effective_operation_size(operation_size)); + self.start_time.elapsed() >= timeout } /// Execute an async operation with timeout protection. @@ -343,95 +338,7 @@ impl RequestTimeoutWrapper { F: FnOnce(CancellationToken) -> Fut, Fut: std::future::Future>, { - if !self.config.is_timeout_enabled() { - // Timeout disabled, run without timeout - debug!( - request_id = %self.request_id, - "Timeout disabled, executing operation without timeout" - ); - return match operation(self.cancel_token).await { - Ok(result) => TimedGetObjectResult::Success(result), - Err(e) => TimedGetObjectResult::Error(e), - }; - } - - let timeout_duration = self.config.get_object_timeout; - let request_id = self.request_id.clone(); - let start_time = self.start_time; - - debug!( - request_id = %request_id, - timeout_secs = timeout_duration.as_secs(), - "Starting timed operation" - ); - - // Record start time for metrics - rustfs_io_metrics::record_get_object_request_started(); - - // Clone cancel_token for the operation, keep original for potential cancellation - let cancel_token_for_op = self.cancel_token.clone(); - - match tokio::time::timeout(timeout_duration, operation(cancel_token_for_op)).await { - Ok(Ok(result)) => { - // Operation completed successfully - let elapsed = start_time.elapsed(); - - rustfs_io_metrics::record_get_object_request_result("success", elapsed.as_secs_f64()); - - debug!( - request_id = %request_id, - elapsed_ms = elapsed.as_millis(), - "Operation completed successfully" - ); - - TimedGetObjectResult::Success(result) - } - Ok(Err(e)) => { - // Operation failed before timeout - let elapsed = start_time.elapsed(); - - rustfs_io_metrics::record_get_object_request_result("error", elapsed.as_secs_f64()); - - debug!( - request_id = %request_id, - elapsed_ms = elapsed.as_millis(), - "Operation failed with error" - ); - - TimedGetObjectResult::Error(e) - } - Err(_) => { - // Timeout occurred - let elapsed = start_time.elapsed(); - - // Cancel the operation - self.cancel_token.cancel(); - - rustfs_io_metrics::record_get_object_timeout(None, Some(elapsed.as_secs_f64())); - rustfs_io_metrics::record_get_object_request_result("timeout", elapsed.as_secs_f64()); - - warn!( - request_id = %request_id, - timeout_secs = timeout_duration.as_secs(), - elapsed_ms = elapsed.as_millis(), - "Operation timed out, cancellation signal sent" - ); - - TimedGetObjectResult::Timeout(TimeoutInfo { - request_id, - bucket: String::new(), - key: String::new(), - timeout_duration, - elapsed, - bytes_transferred: 0, - lock_hold_time: None, - disk_reads_completed: 0, - disk_reads_pending: 0, - object_size: None, - progress_percent: None, - }) - } - } + self.execute_with_timeout_internal(None, operation).await } /// Execute an async operation with timeout and context information. @@ -450,86 +357,81 @@ impl RequestTimeoutWrapper { { let bucket = bucket.into(); let key = key.into(); + self.execute_with_timeout_internal(Some((bucket, key)), operation).await + } - if !self.config.is_timeout_enabled() { - debug!( + async fn execute_with_timeout_internal( + self, + context: Option<(String, String)>, + operation: F, + ) -> TimedGetObjectResult + where + F: FnOnce(CancellationToken) -> Fut, + Fut: std::future::Future>, + { + let timeout_duration = self.get_timeout(None); + + // Build a tracing span that carries request context for all log events. + let span = match &context { + Some((bucket, key)) => tracing::info_span!( + "timeout_operation", request_id = %self.request_id, bucket = %bucket, key = %key, - "Timeout disabled, executing operation without timeout" - ); + ), + None => tracing::info_span!( + "timeout_operation", + request_id = %self.request_id, + ), + }; + let _guard = span.enter(); + + if !self.config.is_timeout_enabled() { + debug!("Timeout disabled, executing operation without timeout"); + return match operation(self.cancel_token).await { Ok(result) => TimedGetObjectResult::Success(result), Err(e) => TimedGetObjectResult::Error(e), }; } - let timeout_duration = self.config.get_object_timeout; - let request_id = self.request_id.clone(); - let start_time = self.start_time; - - debug!( - request_id = %request_id, - bucket = %bucket, - key = %key, - timeout_secs = timeout_duration.as_secs(), - "Starting timed operation" - ); + debug!(timeout_secs = timeout_duration.as_secs(), "Starting timed operation"); rustfs_io_metrics::record_get_object_request_started(); - // Clone cancel_token for the operation, keep original for potential cancellation let cancel_token_for_op = self.cancel_token.clone(); match tokio::time::timeout(timeout_duration, operation(cancel_token_for_op)).await { Ok(Ok(result)) => { - let elapsed = start_time.elapsed(); - + let elapsed = self.elapsed(); rustfs_io_metrics::record_get_object_request_result("success", elapsed.as_secs_f64()); - - debug!( - request_id = %request_id, - bucket = %bucket, - key = %key, - elapsed_ms = elapsed.as_millis(), - "Operation completed successfully" - ); + debug!(elapsed_ms = elapsed.as_millis(), "Operation completed successfully"); TimedGetObjectResult::Success(result) } Ok(Err(e)) => { - let elapsed = start_time.elapsed(); - + let elapsed = self.elapsed(); rustfs_io_metrics::record_get_object_request_result("error", elapsed.as_secs_f64()); - - debug!( - request_id = %request_id, - bucket = %bucket, - key = %key, - elapsed_ms = elapsed.as_millis(), - "Operation failed with error" - ); + debug!(elapsed_ms = elapsed.as_millis(), "Operation failed with error"); TimedGetObjectResult::Error(e) } Err(_) => { - let elapsed = start_time.elapsed(); + let elapsed = self.elapsed(); self.cancel_token.cancel(); rustfs_io_metrics::record_get_object_timeout(None, Some(elapsed.as_secs_f64())); rustfs_io_metrics::record_get_object_request_result("timeout", elapsed.as_secs_f64()); warn!( - request_id = %request_id, - bucket = %bucket, - key = %key, timeout_secs = timeout_duration.as_secs(), elapsed_ms = elapsed.as_millis(), "Operation timed out, cancellation signal sent" ); + let (bucket, key) = context.unwrap_or_default(); TimedGetObjectResult::Timeout(TimeoutInfo { - request_id, + request_id: self.request_id, bucket, key, timeout_duration, @@ -538,7 +440,7 @@ impl RequestTimeoutWrapper { lock_hold_time: None, disk_reads_completed: 0, disk_reads_pending: 0, - object_size: None, + object_size: self.operation_size, progress_percent: None, }) } @@ -566,18 +468,16 @@ mod tests { #[test] fn test_timeout_config_default() { - let config = TimeoutConfig::default(); + let config = GetObjectTimeoutPolicy::default(); assert_eq!(config.get_object_timeout, Duration::from_secs(30)); - assert_eq!(config.lock_acquire_timeout, Duration::from_secs(5)); - assert_eq!(config.disk_read_timeout, Duration::from_secs(10)); } #[test] fn test_timeout_config_is_enabled() { - let config = TimeoutConfig::default(); + let config = GetObjectTimeoutPolicy::default(); assert!(config.is_timeout_enabled()); - let disabled_config = TimeoutConfig { + let disabled_config = GetObjectTimeoutPolicy { get_object_timeout: Duration::ZERO, ..Default::default() }; @@ -586,7 +486,7 @@ mod tests { #[tokio::test] async fn test_timeout_wrapper_success() { - let config = TimeoutConfig { + let config = GetObjectTimeoutPolicy { get_object_timeout: Duration::from_secs(5), ..Default::default() }; @@ -604,7 +504,7 @@ mod tests { #[tokio::test] async fn test_timeout_wrapper_timeout() { - let config = TimeoutConfig { + let config = GetObjectTimeoutPolicy { get_object_timeout: Duration::from_millis(100), ..Default::default() }; @@ -627,7 +527,7 @@ mod tests { #[tokio::test] async fn test_timeout_wrapper_error() { - let config = TimeoutConfig { + let config = GetObjectTimeoutPolicy { get_object_timeout: Duration::from_secs(5), ..Default::default() }; @@ -645,7 +545,7 @@ mod tests { #[tokio::test] async fn test_timeout_wrapper_disabled() { - let config = TimeoutConfig { + let config = GetObjectTimeoutPolicy { get_object_timeout: Duration::ZERO, ..Default::default() }; @@ -681,7 +581,7 @@ mod tests { #[test] fn test_timeout_config_default_with_dynamic() { - let config = TimeoutConfig::default(); + let config = GetObjectTimeoutPolicy::default(); assert!(config.enable_dynamic_timeout); assert_eq!(config.bytes_per_second, rustfs_config::DEFAULT_OBJECT_BYTES_PER_SECOND); assert_eq!(config.min_timeout, Duration::from_secs(rustfs_config::DEFAULT_OBJECT_MIN_TIMEOUT)); @@ -690,7 +590,7 @@ mod tests { #[test] fn test_calculate_timeout_for_size() { - let config = TimeoutConfig::default(); + let config = GetObjectTimeoutPolicy::default(); // Test with small object (should use min timeout) let small_timeout = config.calculate_timeout_for_size(1024); // 1KB @@ -707,9 +607,39 @@ mod tests { assert!(huge_timeout <= Duration::from_secs(rustfs_config::DEFAULT_OBJECT_MAX_TIMEOUT)); } + #[test] + fn test_calculate_timeout_for_size_respects_small_min_timeout() { + let config = GetObjectTimeoutPolicy { + get_object_timeout: Duration::from_secs(30), + enable_dynamic_timeout: true, + bytes_per_second: 1024 * 1024, // 1MB/s + min_timeout: Duration::from_secs(1), + max_timeout: Duration::from_secs(30), + }; + + // Tiny object should still honor policy min_timeout (1s), + // instead of being raised by any external hard floor. + let timeout = config.calculate_timeout_for_size(1024); // 1KB + assert_eq!(timeout, Duration::from_secs(1)); + } + + #[test] + fn test_calculate_timeout_for_size_huge_object_does_not_panic_and_clamps() { + let config = GetObjectTimeoutPolicy { + get_object_timeout: Duration::from_secs(300), + enable_dynamic_timeout: true, + bytes_per_second: 1, + min_timeout: Duration::from_secs(1), + max_timeout: Duration::from_secs(300), + }; + + let timeout = config.calculate_timeout_for_size(u64::MAX); + assert_eq!(timeout, Duration::from_secs(300)); + } + #[test] fn test_timeout_with_dynamic_disabled() { - let config = TimeoutConfig { + let config = GetObjectTimeoutPolicy { enable_dynamic_timeout: false, ..Default::default() }; @@ -778,7 +708,7 @@ mod tests { #[test] fn test_should_timeout() { - let config = TimeoutConfig { + let config = GetObjectTimeoutPolicy { get_object_timeout: Duration::from_millis(100), ..Default::default() }; @@ -795,7 +725,7 @@ mod tests { #[test] fn test_should_timeout_with_size() { - let config = TimeoutConfig { + let config = GetObjectTimeoutPolicy { enable_dynamic_timeout: true, bytes_per_second: 1024, // 1KB/s min_timeout: Duration::from_secs(rustfs_config::DEFAULT_OBJECT_MIN_TIMEOUT), @@ -811,4 +741,23 @@ mod tests { // Large size should calculate longer timeout assert!(!wrapper.should_timeout(Some(10 * 1024 * 1024))); } + + #[test] + fn test_wrapper_operation_size_hint_is_applied() { + let config = GetObjectTimeoutPolicy { + get_object_timeout: Duration::from_secs(300), + enable_dynamic_timeout: true, + bytes_per_second: 1024 * 1024, + min_timeout: Duration::from_secs(1), + max_timeout: Duration::from_secs(300), + }; + let size = 100 * 1024 * 1024; + let wrapper = RequestTimeoutWrapper::with_operation_size(config.clone(), Some(size)); + + let expected_dynamic = config.get_timeout_for_operation(Some(size)); + let baseline_no_size = config.get_timeout_for_operation(None); + + assert_ne!(expected_dynamic, baseline_no_size); + assert_eq!(wrapper.get_timeout(None), expected_dynamic); + } }