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

chore(storage): drop dead backpressure and lock-optimizer wrappers (#6195)

Browse Source 
Neither rustfs/src/storage/backpressure.rs nor rustfs/src/storage/lock_optimizer.rs had a production caller: their only non-self references were the pub mod lines in storage/mod.rs and a cfg(test) module, so the object transfer path never applied this backpressure and never took these lock shortcuts.

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