gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-07-26 08:18:18 +00:00
Code Issues Packages Projects Releases Wiki Activity

refactor: converge storage io hot paths (#3029)

Browse Source 
* refactor(issue-633): clarify layered io control policies

* refactor(issue-633): consolidate timeout and deadlock layers

* refactor(issue-633): align storage backpressure metadata

* refactor(issue-633): unify storage backpressure transitions

* refactor(issue-633): simplify watermark transition API

* test(issue-633): add storage backpressure transition test

* refactor(issue-633): align storage pipe meta shape

* refactor(issue-633): enrich storage monitor metadata

* refactor(issue-633): finalize storage backpressure convergence

* refactor(issue-633): complete scheduler layer convergence

* refactor(issue-633): reduce concurrency facade config duplication

* refactor(issue-633): migrate storage callsites to final policy names

* chore(issue-633): apply final pre-commit normalization

* refactor(issue-633): unify timeout wrapper dynamic size path

* refactor(issue-633): make concurrency policies copyable

* refactor(issue-633): converge storage io hot paths

* fix(issue-633): honor storage timeout min bound

* fix(storage): avoid timeout calc panic on huge sizes

* refactor(storage): consolidate timeout checks and test attrs

* fix(storage): harden io scheduler core config mapping

* refactor(storage): eliminate patch-on-patch patterns and dead code

- Remove trivial accessor methods on ConcurrencyConfig that just return pub fields
- Remove dead BackpressureEvent/BackpressureEventType types from concurrency crate
- Fix io_schedule test using wrong constructor (from_core_config -> from_scheduler_config)
- Update manager.rs to use config fields directly instead of removed accessors

* fix: adopt review feedback for config guards

* test: remove needless struct update defaults

* fix: harden timeout policy and preserve api alias
This commit is contained in:
houseme
2026-05-20 20:00:23 +08:00
committed by GitHub
parent 45b04316b9
commit 375482a4b6
14 changed files with 988 additions and 702 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/concurrency/src/backpressure.rs
+72 -51
View File
@@ -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<CoreBackpressureMonitor>,
}
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<CoreBackpressureMonitor> {
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<CoreBackpressureMonitor>,
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<CoreBackpressureMonitor>) -> Self {
fn new(config: PipeBackpressurePolicy, monitor: Arc<CoreBackpressureMonitor>) -> 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);
}
}
crates/concurrency/src/config.rs
+70 -84
View File
@@ -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::<u64>()
{
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::<u64>()
{
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::<usize>()
{
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::<usize>()
{
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();
crates/concurrency/src/deadlock.rs
+47 -17
View File
@@ -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<CoreDeadlockDetector>,
running: Arc<tokio::sync::Mutex<bool>>,
}
@@ -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<String, Vec<String>> {
&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));
}
}
crates/concurrency/src/lib.rs
+8 -8
View File
@@ -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};
}
crates/concurrency/src/manager.rs
+7 -24
View File
@@ -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;
}
}
crates/concurrency/src/scheduler.rs
+47 -14
View File
@@ -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<CoreIoScheduler>,
}
@@ -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<CoreIoScheduler> {
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<CoreIoScheduler>,
}
impl IoStrategy {
fn new(config: SchedulerConfig, scheduler: Arc<CoreIoScheduler>) -> Self {
fn new(config: SchedulerPolicy, scheduler: Arc<CoreIoScheduler>) -> 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);
crates/concurrency/src/timeout.rs
+89 -17
View File
@@ -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<Duration>) -> 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);
rustfs/src/app/object_usecase.rs
+75 -43
View File
@@ -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<GetObjectBootstrap> {
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<GetObjectIoPlanning<'a>> {
@@ -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<GetObjectInput>,
manager: &'a ConcurrencyManager,
wrapper: &RequestTimeoutWrapper,
timeout_config: &TimeoutConfig,
timeout_config: &GetObjectTimeoutPolicy,
bucket: &str,
key: &str,
rs: Option<HTTPRangeSpec>,
@@ -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,
rustfs/src/storage/backpressure.rs
+201 -140
View File
@@ -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<AtomicUsize>,
buffer_usage: AtomicUsize,
/// Current backpressure state.
state: Arc<AtomicBool>, // true = in high watermark state
state: AtomicBool, // true = in high watermark state
/// Total bytes written.
total_written: Arc<AtomicUsize>,
total_written: AtomicUsize,
/// Total bytes read.
total_read: Arc<AtomicUsize>,
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<AtomicUsize>,
buffer_usage: AtomicUsize,
/// In high watermark state.
in_high_watermark: Arc<AtomicBool>,
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);
}
}
rustfs/src/storage/concurrency/io_schedule.rs
+161 -47
View File
@@ -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<T> IoPriorityQueue<T> {
@@ -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
rustfs/src/storage/concurrency/manager.rs
+2 -8
View File
@@ -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)),
rustfs/src/storage/concurrent_fix_test.rs
+12 -12
View File
@@ -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()
};
rustfs/src/storage/deadlock_detector.rs
+27 -16
View File
@@ -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<RwLock<HashMap<RequestId, RequestResourceTracker>>>,
/// 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<RwLock<HashMap<RequestId, RequestResourceTracker>>>,
config: &DeadlockDetectorConfig,
config: &RequestHangDetectionPolicy,
deadlocks_detected: &Arc<AtomicU64>,
) {
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<Vec<RequestId>> {
// 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<Arc<DeadlockDetector>> = std::sync
pub fn get_deadlock_detector() -> Arc<DeadlockDetector> {
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()
};
rustfs/src/storage/timeout_wrapper.rs
+170 -221
View File
@@ -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<T, E> {
}
/// 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<u64>,
/// 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<String>) -> 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<String>) -> 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<u64>) -> 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<u64>) -> 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<u64>) -> Duration {
self.config.get_timeout_for_operation(operation_size)
pub fn get_timeout(&self, operation_size_hint: Option<u64>) -> Duration {
self.config
.get_timeout_for_operation(self.effective_operation_size(operation_size_hint))
}
fn new_with_parts(
config: GetObjectTimeoutPolicy,
operation_size: Option<u64>,
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<u64>) -> Option<u64> {
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<Output = Result<T, E>>,
{
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<F, Fut, T, E>(
self,
context: Option<(String, String)>,
operation: F,
) -> TimedGetObjectResult<T, E>
where
F: FnOnce(CancellationToken) -> Fut,
Fut: std::future::Future<Output = Result<T, E>>,
{
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);
}
}