fix(object): Fix concurrent request hang issue in S3 range read workloads (#2251)

Co-authored-by: heihutu <heihutu@gmail.com>
This commit is contained in:
houseme
2026-03-23 09:55:08 +08:00
committed by GitHub
parent ff40e2bc79
commit 6cb094e30a
20 changed files with 6451 additions and 2163 deletions
Generated
+2
View File
@@ -7270,6 +7270,7 @@ dependencies = [
"base64 0.22.1",
"base64-simd",
"bytes",
"chrono",
"clap",
"const-str",
"datafusion",
@@ -7500,6 +7501,7 @@ dependencies = [
"hyper-util",
"lazy_static",
"md-5 0.11.0-rc.5",
"metrics",
"moka",
"num_cpus",
"parking_lot 0.12.5",
+262
View File
@@ -180,3 +180,265 @@ pub const ENV_OBJECT_GET_SKIP_BITROT_VERIFY: &str = "RUSTFS_OBJECT_GET_SKIP_BITR
/// Default: bitrot verification is enabled on GetObject reads (do not skip).
pub const DEFAULT_OBJECT_GET_SKIP_BITROT_VERIFY: bool = false;
// =============================================================================
// Concurrent Request Fix - Timeout and Backpressure Configuration
// =============================================================================
/// Environment variable for GetObject request timeout in seconds.
///
/// When a GetObject request exceeds this duration, it will be cancelled
/// and return a 504 Gateway Timeout error. This prevents requests from
/// hanging indefinitely due to deadlocks or resource exhaustion.
///
/// Default: 30 seconds (can be overridden by `RUSTFS_OBJECT_GET_TIMEOUT`).
/// Set to 0 to disable timeout (not recommended for production).
pub const ENV_OBJECT_GET_TIMEOUT: &str = "RUSTFS_OBJECT_GET_TIMEOUT";
/// Default GetObject request timeout in seconds.
///
/// This value balances between allowing large object transfers to complete
/// and preventing indefinite hangs. For 20-26MB objects with concurrent
/// range reads, 30 seconds should be sufficient under normal conditions.
pub const DEFAULT_OBJECT_GET_TIMEOUT: u64 = 30;
/// Environment variable for disk read operation timeout in seconds.
///
/// Individual disk read operations that exceed this duration will be
/// cancelled and treated as failures. This helps detect slow or hung
/// disks without waiting indefinitely.
///
/// Default: 10 seconds (can be overridden by `RUSTFS_OBJECT_DISK_READ_TIMEOUT`).
pub const ENV_OBJECT_DISK_READ_TIMEOUT: &str = "RUSTFS_OBJECT_DISK_READ_TIMEOUT";
/// Default disk read timeout in seconds.
pub const DEFAULT_OBJECT_DISK_READ_TIMEOUT: u64 = 10;
/// Environment variable for duplex pipe buffer size in bytes.
///
/// The duplex pipe connects the disk read task to the HTTP response stream.
/// A larger buffer reduces backpressure but increases memory usage.
/// For large objects (20-26MB), a 4MB buffer provides good throughput.
///
/// Default: 4194304 (4 MB, can be overridden by `RUSTFS_OBJECT_DUPLEX_BUFFER_SIZE`).
/// Minimum recommended: 1048576 (1 MB).
pub const ENV_OBJECT_DUPLEX_BUFFER_SIZE: &str = "RUSTFS_OBJECT_DUPLEX_BUFFER_SIZE";
/// Default duplex buffer size: 4 MB.
///
/// This is 4x larger than the original 1 MB buffer, providing better
/// handling of large objects and reducing backpressure-related hangs.
pub const DEFAULT_OBJECT_DUPLEX_BUFFER_SIZE: usize = 4 * 1024 * 1024;
/// Environment variable for I/O buffer size in bytes.
///
/// This controls the buffer size used for individual I/O operations.
/// A larger buffer improves throughput for sequential reads but may
/// increase latency for small random reads.
///
/// Default: 131072 (128 KB, can be overridden by `RUSTFS_OBJECT_IO_BUFFER_SIZE`).
pub const ENV_OBJECT_IO_BUFFER_SIZE: &str = "RUSTFS_OBJECT_IO_BUFFER_SIZE";
/// Default I/O buffer size: 128 KB.
pub const DEFAULT_OBJECT_IO_BUFFER_SIZE: usize = 128 * 1024;
/// Environment variable to enable/disable lock optimization.
///
/// When enabled, read locks are released immediately after metadata
/// is read, rather than being held for the entire data transfer.
/// This significantly reduces lock contention under high concurrency.
///
/// Default: true (enabled, can be overridden by `RUSTFS_OBJECT_LOCK_OPTIMIZATION_ENABLE`).
pub const ENV_OBJECT_LOCK_OPTIMIZATION_ENABLE: &str = "RUSTFS_OBJECT_LOCK_OPTIMIZATION_ENABLE";
/// Default: lock optimization is enabled.
pub const DEFAULT_OBJECT_LOCK_OPTIMIZATION_ENABLE: bool = true;
/// Environment variable to enable/disable priority-based I/O scheduling.
///
/// When enabled, smaller requests (< 1MB) are given higher priority
/// than larger requests (> 10MB), preventing "starvation" of small
/// requests by large ones.
///
/// Default: true (enabled, can be overridden by `RUSTFS_OBJECT_PRIORITY_SCHEDULING_ENABLE`).
pub const ENV_OBJECT_PRIORITY_SCHEDULING_ENABLE: &str = "RUSTFS_OBJECT_PRIORITY_SCHEDULING_ENABLE";
/// Default: priority scheduling is enabled.
pub const DEFAULT_OBJECT_PRIORITY_SCHEDULING_ENABLE: bool = true;
/// Environment variable to enable/disable deadlock detection.
///
/// When enabled, the system monitors active requests and detects
/// potential deadlock situations (circular lock wait chains).
/// This has some performance overhead and is intended for debugging.
///
/// Default: false (disabled, can be overridden by `RUSTFS_OBJECT_DEADLOCK_DETECTION_ENABLE`).
pub const ENV_OBJECT_DEADLOCK_DETECTION_ENABLE: &str = "RUSTFS_OBJECT_DEADLOCK_DETECTION_ENABLE";
/// Default: deadlock detection is disabled for performance.
pub const DEFAULT_OBJECT_DEADLOCK_DETECTION_ENABLE: bool = false;
/// Environment variable for deadlock detection check interval in seconds.
///
/// How often the deadlock detector analyzes the lock wait graph.
/// More frequent checks detect deadlocks faster but use more CPU.
///
/// Default: 5 seconds (can be overridden by `RUSTFS_OBJECT_DEADLOCK_CHECK_INTERVAL`).
pub const ENV_OBJECT_DEADLOCK_CHECK_INTERVAL: &str = "RUSTFS_OBJECT_DEADLOCK_CHECK_INTERVAL";
/// Default deadlock check interval: 5 seconds.
pub const DEFAULT_OBJECT_DEADLOCK_CHECK_INTERVAL: u64 = 5;
/// Environment variable for deadlock detection hang threshold in seconds.
///
/// Requests that have been running longer than this threshold are
/// considered "potentially hung" and included in deadlock analysis.
///
/// Default: 10 seconds (can be overridden by `RUSTFS_OBJECT_DEADLOCK_HANG_THRESHOLD`).
pub const ENV_OBJECT_DEADLOCK_HANG_THRESHOLD: &str = "RUSTFS_OBJECT_DEADLOCK_HANG_THRESHOLD";
/// Default hang threshold: 10 seconds.
pub const DEFAULT_OBJECT_DEADLOCK_HANG_THRESHOLD: u64 = 10;
/// Environment variable for backpressure high watermark percentage.
///
/// When buffer usage exceeds this percentage, the system enters
/// "high watermark" state and may apply backpressure to producers.
///
/// Default: 80 (80%, can be overridden by `RUSTFS_OBJECT_BACKPRESSURE_HIGH_WATERMARK`).
pub const ENV_OBJECT_BACKPRESSURE_HIGH_WATERMARK: &str = "RUSTFS_OBJECT_BACKPRESSURE_HIGH_WATERMARK";
/// Default high watermark: 80%.
pub const DEFAULT_OBJECT_BACKPRESSURE_HIGH_WATERMARK: u32 = 80;
/// Environment variable for backpressure low watermark percentage.
///
/// When buffer usage drops below this percentage after being in
/// high watermark state, backpressure is released.
///
/// Default: 50 (50%, can be overridden by `RUSTFS_OBJECT_BACKPRESSURE_LOW_WATERMARK`).
pub const ENV_OBJECT_BACKPRESSURE_LOW_WATERMARK: &str = "RUSTFS_OBJECT_BACKPRESSURE_LOW_WATERMARK";
/// Default low watermark: 50%.
pub const DEFAULT_OBJECT_BACKPRESSURE_LOW_WATERMARK: u32 = 50;
/// Environment variable for lock acquisition timeout in seconds.
///
/// When a lock cannot be acquired within this duration, the operation
/// will fail with a timeout error. This prevents indefinite waiting
/// for locks that may never be released due to deadlocks.
///
/// Default: 5 seconds (can be overridden by `RUSTFS_OBJECT_LOCK_ACQUIRE_TIMEOUT`).
pub const ENV_OBJECT_LOCK_ACQUIRE_TIMEOUT: &str = "RUSTFS_OBJECT_LOCK_ACQUIRE_TIMEOUT";
/// Default lock acquisition timeout: 5 seconds.
pub const DEFAULT_OBJECT_LOCK_ACQUIRE_TIMEOUT: u64 = 5;
// ============================================================================
// I/O priority scheduling configuration
// ============================================================================
/// Environment variable for I/O high priority size threshold in bytes.
///
/// Requests smaller than this threshold are classified as high priority.
/// High priority requests are processed first to prevent starvation of small requests.
///
/// Default: 1048576 (1 MB, can be overridden by `RUSTFS_OBJECT_IO_HIGH_PRIORITY_SIZE_THRESHOLD`).
pub const ENV_OBJECT_IO_HIGH_PRIORITY_SIZE_THRESHOLD: &str = "RUSTFS_OBJECT_IO_HIGH_PRIORITY_SIZE_THRESHOLD";
/// Default high priority size threshold: 1 MB.
pub const DEFAULT_OBJECT_IO_HIGH_PRIORITY_SIZE_THRESHOLD: usize = 1024 * 1024;
/// Environment variable for I/O low priority size threshold in bytes.
///
/// Requests larger than this threshold are classified as low priority.
/// Low priority requests are processed last to avoid blocking small requests.
///
/// Default: 104857600 (100 MB, can be overridden by `RUSTFS_OBJECT_IO_LOW_PRIORITY_SIZE_THRESHOLD`).
pub const ENV_OBJECT_IO_LOW_PRIORITY_SIZE_THRESHOLD: &str = "RUSTFS_OBJECT_IO_LOW_PRIORITY_SIZE_THRESHOLD";
/// Default low priority size threshold: 100 MB.
pub const DEFAULT_OBJECT_IO_LOW_PRIORITY_SIZE_THRESHOLD: usize = 100 * 1024 * 1024;
/// Environment variable for high priority queue capacity.
///
/// Maximum number of requests that can be queued in the high priority queue.
///
/// Default: 32 (can be overridden by `RUSTFS_OBJECT_IO_QUEUE_HIGH_CAPACITY`).
pub const ENV_OBJECT_IO_QUEUE_HIGH_CAPACITY: &str = "RUSTFS_OBJECT_IO_QUEUE_HIGH_CAPACITY";
/// Default high priority queue capacity: 32.
pub const DEFAULT_OBJECT_IO_QUEUE_HIGH_CAPACITY: usize = 32;
/// Environment variable for normal priority queue capacity.
///
/// Maximum number of requests that can be queued in the normal priority queue.
///
/// Default: 64 (can be overridden by `RUSTFS_OBJECT_IO_QUEUE_NORMAL_CAPACITY`).
pub const ENV_OBJECT_IO_QUEUE_NORMAL_CAPACITY: &str = "RUSTFS_OBJECT_IO_QUEUE_NORMAL_CAPACITY";
/// Default normal priority queue capacity: 64.
pub const DEFAULT_OBJECT_IO_QUEUE_NORMAL_CAPACITY: usize = 64;
/// Environment variable for low priority queue capacity.
///
/// Maximum number of requests that can be queued in the low priority queue.
///
/// Default: 16 (can be overridden by `RUSTFS_OBJECT_IO_QUEUE_LOW_CAPACITY`).
pub const ENV_OBJECT_IO_QUEUE_LOW_CAPACITY: &str = "RUSTFS_OBJECT_IO_QUEUE_LOW_CAPACITY";
/// Default low priority queue capacity: 16.
pub const DEFAULT_OBJECT_IO_QUEUE_LOW_CAPACITY: usize = 16;
/// Environment variable for starvation prevention check interval in milliseconds.
///
/// How often the system checks for starving low-priority requests.
/// When a low-priority request has been waiting longer than the starvation threshold,
/// it is promoted to normal priority.
///
/// Default: 100 ms (can be overridden by `RUSTFS_OBJECT_IO_STARVATION_PREVENTION_INTERVAL`).
pub const ENV_OBJECT_IO_STARVATION_PREVENTION_INTERVAL: &str = "RUSTFS_OBJECT_IO_STARVATION_PREVENTION_INTERVAL";
/// Default starvation prevention interval: 100 ms.
pub const DEFAULT_OBJECT_IO_STARVATION_PREVENTION_INTERVAL: u64 = 100;
/// Environment variable for starvation threshold in seconds.
///
/// Maximum time a low-priority request can wait before being promoted to normal priority.
/// This prevents indefinite starvation of low-priority requests.
///
/// Default: 5 seconds (can be overridden by `RUSTFS_OBJECT_IO_STARVATION_THRESHOLD_SECS`).
pub const ENV_OBJECT_IO_STARVATION_THRESHOLD_SECS: &str = "RUSTFS_OBJECT_IO_STARVATION_THRESHOLD_SECS";
/// Default starvation threshold: 5 seconds.
pub const DEFAULT_OBJECT_IO_STARVATION_THRESHOLD_SECS: u64 = 5;
/// Environment variable for load sampling window size.
///
/// Number of recent samples used to calculate I/O load metrics.
///
/// Default: 100 samples (can be overridden by `RUSTFS_OBJECT_IO_LOAD_SAMPLE_WINDOW`).
pub const ENV_OBJECT_IO_LOAD_SAMPLE_WINDOW: &str = "RUSTFS_OBJECT_IO_LOAD_SAMPLE_WINDOW";
/// Default load sampling window: 100 samples.
pub const DEFAULT_OBJECT_IO_LOAD_SAMPLE_WINDOW: usize = 100;
/// Environment variable for high load wait time threshold in milliseconds.
///
/// When average wait time exceeds this threshold, the system is considered to be under high load.
///
/// Default: 50 ms (can be overridden by `RUSTFS_OBJECT_IO_LOAD_HIGH_THRESHOLD_MS`).
pub const ENV_OBJECT_IO_LOAD_HIGH_THRESHOLD_MS: &str = "RUSTFS_OBJECT_IO_LOAD_HIGH_THRESHOLD_MS";
/// Default high load threshold: 50 ms.
pub const DEFAULT_OBJECT_IO_LOAD_HIGH_THRESHOLD_MS: u64 = 50;
/// Environment variable for low load wait time threshold in milliseconds.
///
/// When average wait time is below this threshold, the system is considered to be under low load.
///
/// Default: 10 ms (can be overridden by `RUSTFS_OBJECT_IO_LOAD_LOW_THRESHOLD_MS`).
pub const ENV_OBJECT_IO_LOAD_LOW_THRESHOLD_MS: &str = "RUSTFS_OBJECT_IO_LOAD_LOW_THRESHOLD_MS";
/// Default low load threshold: 10 ms.
pub const DEFAULT_OBJECT_IO_LOAD_LOW_THRESHOLD_MS: u64 = 10;
+3
View File
@@ -116,6 +116,9 @@ faster-hex = { workspace = true }
ratelimit = { workspace = true }
aws-smithy-http-client.workspace = true
# Observability and Metrics
metrics = { workspace = true }
[dev-dependencies]
tokio = { workspace = true, features = ["rt-multi-thread", "macros"] }
criterion = { workspace = true, features = ["html_reports"] }
+149 -25
View File
@@ -121,6 +121,21 @@ use uuid::Uuid;
pub const DEFAULT_READ_BUFFER_SIZE: usize = MI_B; // 1 MiB = 1024 * 1024;
pub const MAX_PARTS_COUNT: usize = 10000;
/// Get the duplex buffer size from environment variable or use default.
///
/// This function reads `RUSTFS_DUPLEX_BUFFER_SIZE` environment variable
/// to allow runtime configuration of the duplex pipe buffer size.
/// A larger buffer (e.g., 4MB) helps prevent backpressure-related hangs
/// when reading large objects (20-26MB) under high concurrency.
///
/// Default: 4MB (4 * 1024 * 1024 bytes)
pub fn get_duplex_buffer_size() -> usize {
rustfs_utils::get_env_usize(
rustfs_config::ENV_OBJECT_DUPLEX_BUFFER_SIZE,
rustfs_config::DEFAULT_OBJECT_DUPLEX_BUFFER_SIZE,
)
}
const DISK_ONLINE_TIMEOUT: Duration = Duration::from_secs(1);
const DISK_HEALTH_CACHE_TTL: Duration = Duration::from_millis(750);
@@ -136,7 +151,72 @@ mod write;
/// Get lock acquire timeout from environment variable RUSTFS_LOCK_ACQUIRE_TIMEOUT (in seconds)
/// Defaults to 30 seconds if not set or invalid
pub fn get_lock_acquire_timeout() -> Duration {
Duration::from_secs(rustfs_utils::get_env_u64("RUSTFS_LOCK_ACQUIRE_TIMEOUT", 5))
Duration::from_secs(rustfs_utils::get_env_u64(
rustfs_config::ENV_OBJECT_LOCK_ACQUIRE_TIMEOUT,
rustfs_config::DEFAULT_OBJECT_LOCK_ACQUIRE_TIMEOUT,
))
}
/// Check if lock optimization is enabled.
/// When enabled, read locks are released after metadata read instead of
/// being held for the entire data transfer duration.
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,
)
}
/// Check if deadlock detection is enabled.
/// When enabled, lock operations are recorded for deadlock analysis.
pub fn is_deadlock_detection_enabled() -> bool {
rustfs_utils::get_env_bool(
rustfs_config::ENV_OBJECT_DEADLOCK_DETECTION_ENABLE,
rustfs_config::DEFAULT_OBJECT_DEADLOCK_DETECTION_ENABLE,
)
}
/// Record a lock acquisition for deadlock detection.
/// This records detailed lock information for deadlock analysis.
/// Returns the lock_id for later release tracking.
#[inline]
fn record_lock_acquire(bucket: &str, object: &str, lock_type: &str) -> String {
let lock_id = format!("{}:{}", bucket, object);
if !is_deadlock_detection_enabled() {
return lock_id;
}
let request_id = format!("get-{}-{}", bucket, object);
let resource = format!("{}/{}", bucket, object);
// Log with structured fields for analysis
debug!(
request_id = %request_id,
lock_id = %lock_id,
lock_type = %lock_type,
resource = %resource,
"Lock acquired for deadlock tracking"
);
lock_id
}
/// Record a lock release for deadlock detection.
#[inline]
fn record_lock_release(bucket: &str, object: &str, lock_id: &str, lock_type: &str) {
if !is_deadlock_detection_enabled() {
return;
}
let request_id = format!("get-{}-{}", bucket, object);
debug!(
request_id = %request_id,
lock_id = %lock_id,
lock_type = %lock_type,
"Lock released for deadlock tracking"
);
}
fn build_tiered_decommission_file_info(
@@ -451,20 +531,44 @@ impl ObjectIO for SetDisks {
h: HeaderMap,
opts: &ObjectOptions,
) -> Result<GetObjectReader> {
// Check if lock optimization is enabled
// When enabled, read locks are released after metadata read
let lock_optimization_enabled = is_lock_optimization_enabled();
// Acquire a shared read-lock early to protect read consistency
let read_lock_guard = if !opts.no_lock {
Some(
self.new_ns_lock(bucket, object)
.await?
.get_read_lock(get_lock_acquire_timeout())
.await
.map_err(|e| {
Error::other(format!(
"Failed to acquire read lock: {}",
self.format_lock_error_from_error(bucket, object, "read", &e)
))
})?,
)
let acquire_start = Instant::now();
// Record lock wait for deadlock detection
if is_deadlock_detection_enabled() {
debug!(
lock_id = format!("{}:{}", bucket, object),
lock_type = "read",
resource = format!("{}/{}", bucket, object),
"Waiting for read lock"
);
}
let guard = self
.new_ns_lock(bucket, object)
.await?
.get_read_lock(get_lock_acquire_timeout())
.await
.map_err(|e| {
Error::other(format!(
"Failed to acquire read lock: {}",
self.format_lock_error_from_error(bucket, object, "read", &e)
))
})?;
// Record lock acquisition for deadlock detection
let _lock_id = record_lock_acquire(bucket, object, "read");
// Record lock statistics
metrics::counter!("rustfs.lock.acquire.total", "type" => "read").increment(1);
metrics::histogram!("rustfs.lock.acquire.duration.seconds").record(acquire_start.elapsed().as_secs_f64());
Some(guard)
} else {
None
};
@@ -512,7 +616,28 @@ impl ObjectIO for SetDisks {
return Ok(gr);
}
let (rd, wd) = tokio::io::duplex(DEFAULT_READ_BUFFER_SIZE);
// Lock optimization: release read lock after metadata read if enabled
// This reduces lock contention by not holding the lock during data transfer
let read_lock_guard = if lock_optimization_enabled {
// Record lock release for deadlock detection
if read_lock_guard.is_some() {
let lock_id = format!("{}:{}", bucket, object);
record_lock_release(bucket, object, &lock_id, "read");
// Record early lock release statistics
metrics::counter!("rustfs.lock.release.early.total", "type" => "read").increment(1);
}
// Explicitly drop the lock guard to release the lock early
drop(read_lock_guard);
debug!(bucket, object, "Lock optimization: released read lock after metadata read");
None
} else {
read_lock_guard
};
let duplex_buffer_size = get_duplex_buffer_size();
let (rd, wd) = tokio::io::duplex(duplex_buffer_size);
debug!(bucket, object, duplex_buffer_size, "Created duplex pipe for object data transfer");
let (reader, offset, length) = GetObjectReader::new(Box::new(rd), range, &object_info, opts, &h)?;
@@ -523,9 +648,10 @@ impl ObjectIO for SetDisks {
let pool_index = self.pool_index;
let skip_verify = opts.skip_verify_bitrot;
// Move the read-lock guard into the task so it lives for the duration of the read
// Note: when lock optimization is enabled, read_lock_guard is None
// let _guard_to_hold = _read_lock_guard; // moved into closure below
tokio::spawn(async move {
let _guard = read_lock_guard; // keep guard alive until task ends
let _guard = read_lock_guard; // keep guard alive until task ends (None if optimization enabled)
let mut writer = wd;
if let Err(e) = Self::get_object_with_fileinfo(
&bucket,
@@ -751,7 +877,7 @@ impl ObjectIO for SetDisks {
pfi.metadata = user_defined.clone();
if is_inline_buffer {
if let Some(writer) = writers[i].take() {
pfi.data = Some(writer.into_inline_data().map(bytes::Bytes::from).unwrap_or_default());
pfi.data = Some(writer.into_inline_data().map(Bytes::from).unwrap_or_default());
}
pfi.set_inline_data();
@@ -1093,7 +1219,7 @@ impl ObjectOperations for SetDisks {
let mut unique_objects: HashSet<String> = HashSet::new();
for dobj in &objects {
if unique_objects.insert(dobj.object_name.clone()) {
batch = batch.add_write_lock(rustfs_lock::ObjectKey::new(bucket, dobj.object_name.clone()));
batch = batch.add_write_lock(ObjectKey::new(bucket, dobj.object_name.clone()));
}
}
@@ -2757,8 +2883,7 @@ impl MultipartOperations for SetDisks {
// Build a lookup map for O(1) part resolution instead of O(n) find() in the loop
// This optimizes from O(n^2) to O(n) when processing many parts
use std::collections::HashMap;
let part_lookup: HashMap<usize, &rustfs_filemeta::ObjectPartInfo> =
curr_fi.parts.iter().map(|part| (part.number, part)).collect();
let part_lookup: HashMap<usize, &ObjectPartInfo> = curr_fi.parts.iter().map(|part| (part.number, part)).collect();
for (i, p) in uploaded_parts.iter().enumerate() {
let Some(ext_part) = part_lookup.get(&p.part_num) else {
@@ -3445,12 +3570,11 @@ async fn disks_with_all_parts(
if (meta.data.is_some() || meta.size == 0) && !meta.parts.is_empty() {
if let Some(data) = &meta.data {
let checksum_info = meta.erasure.get_checksum_info(meta.parts[0].number);
let checksum_algo =
if meta.uses_legacy_checksum && checksum_info.algorithm == rustfs_utils::HashAlgorithm::HighwayHash256S {
rustfs_utils::HashAlgorithm::HighwayHash256SLegacy
} else {
checksum_info.algorithm
};
let checksum_algo = if meta.uses_legacy_checksum && checksum_info.algorithm == HashAlgorithm::HighwayHash256S {
HashAlgorithm::HighwayHash256SLegacy
} else {
checksum_info.algorithm
};
let data_len = data.len();
let verify_err = bitrot_verify(
Box::new(Cursor::new(data.clone())),
+1
View File
@@ -106,6 +106,7 @@ serde_urlencoded = { workspace = true }
# Cryptography and Security
rustls = { workspace = true }
subtle = { workspace = true }
chrono = { workspace = true }
jiff = { workspace = true }
time = { workspace = true, features = ["parsing", "formatting", "serde"] }
+131 -1
View File
@@ -30,6 +30,7 @@ use crate::storage::options::{
};
use crate::storage::s3_api::multipart::parse_list_parts_params;
use crate::storage::s3_api::{acl, restore, select};
use crate::storage::timeout_wrapper::{RequestTimeoutWrapper, TimeoutConfig};
use crate::storage::*;
use bytes::Bytes;
use datafusion::arrow::{
@@ -930,12 +931,37 @@ impl DefaultObjectUsecase {
let _ = context.object_store();
}
// Create timeout wrapper for enhanced timeout tracking
let timeout_config = TimeoutConfig::from_env();
let wrapper =
RequestTimeoutWrapper::with_request_id(timeout_config.clone(), format!("get-{}-{}", req.input.bucket, req.input.key));
// Get cancellation token for cooperative cancellation
let _cancel_token = wrapper.cancel_token();
let request_start = std::time::Instant::now();
// Track this request for concurrency-aware optimizations
let _request_guard = ConcurrencyManager::track_request();
let concurrent_requests = GetObjectGuard::concurrent_requests();
// Register with deadlock detector if enabled
let deadlock_detector = crate::storage::deadlock_detector::get_deadlock_detector();
let request_id = wrapper.request_id().to_string();
deadlock_detector.register_request(&request_id, format!("GetObject {}/{}", req.input.bucket, req.input.key));
// Check for request timeout before proceeding
if wrapper.is_timeout() {
warn!(
bucket = %req.input.bucket,
key = %req.input.key,
timeout_secs = timeout_config.get_object_timeout.as_secs(),
elapsed_ms = wrapper.elapsed().as_millis(),
"GetObject request timed out before processing"
);
deadlock_detector.unregister_request(&request_id);
return Err(s3_error!(InternalError, "Request timeout before processing"));
}
#[cfg(feature = "metrics")]
{
use metrics::{counter, gauge};
@@ -943,7 +969,10 @@ impl DefaultObjectUsecase {
gauge!("rustfs.concurrent.get.object.requests").set(concurrent_requests as f64);
}
debug!("GetObject request started with {} concurrent requests", concurrent_requests);
debug!(
"GetObject request started with {} concurrent requests, timeout={:?}",
concurrent_requests, timeout_config.get_object_timeout
);
let mut helper = OperationHelper::new(&req, EventName::ObjectAccessedGet, S3Operation::GetObject);
// mc get 3
@@ -1111,18 +1140,86 @@ impl DefaultObjectUsecase {
.map_err(|_| s3_error!(InternalError, "disk read semaphore closed"))?;
let permit_wait_duration = permit_wait_start.elapsed();
// Check timeout after acquiring permit
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"
);
deadlock_detector.unregister_request(&request_id);
#[cfg(feature = "metrics")]
metrics::counter!("rustfs.get.object.timeout.total", "stage" => "disk_permit").increment(1);
return Err(s3_error!(InternalError, "Request timeout while waiting for disk permit"));
}
// Monitor I/O queue status for congestion detection
let queue_status = manager.io_queue_status();
let queue_utilization = if queue_status.total_permits > 0 {
(queue_status.permits_in_use as f64 / queue_status.total_permits as f64) * 100.0
} else {
0.0
};
// Log warning if queue is congested (> 80% utilization)
if queue_utilization > 80.0 {
warn!(
bucket = %bucket,
key = %key,
queue_utilization = format!("{:.1}%", queue_utilization),
permits_in_use = queue_status.permits_in_use,
total_permits = queue_status.total_permits,
"I/O queue congestion detected"
);
#[cfg(feature = "metrics")]
{
use metrics::counter;
counter!("rustfs.io.queue.congestion.total").increment(1);
}
}
// Calculate adaptive I/O strategy from permit wait time
// This adjusts buffer sizes, read-ahead, and caching behavior based on load
// Use 256KB as the base buffer size for strategy calculation
let base_buffer_size = self.base_buffer_size();
let io_strategy = manager.calculate_io_strategy(permit_wait_duration, base_buffer_size);
// Determine I/O priority based on request size (for priority scheduling)
// Small requests (< 1MB) get high priority, large requests (> 10MB) get low priority
let io_priority = manager.get_io_priority(io_strategy.buffer_size as i64);
// Log priority information for observability
if manager.is_priority_scheduling_enabled() {
debug!(
bucket = %bucket,
key = %key,
priority = %io_priority,
request_size = io_strategy.buffer_size,
"I/O priority assigned"
);
#[cfg(feature = "metrics")]
{
use metrics::counter;
counter!("rustfs.io.priority.assigned.total", "priority" => io_priority.as_str()).increment(1);
}
}
// Record detailed I/O metrics for monitoring
#[cfg(feature = "metrics")]
{
use metrics::{counter, gauge, histogram};
// Record permit wait time histogram
histogram!("rustfs.disk.permit.wait.duration.seconds").record(permit_wait_duration.as_secs_f64());
// Record I/O queue utilization
gauge!("rustfs.io.queue.utilization").set(queue_utilization);
gauge!("rustfs.io.queue.permits_in_use").set(queue_status.permits_in_use as f64);
gauge!("rustfs.io.queue.permits_available")
.set(queue_status.total_permits.saturating_sub(queue_status.permits_in_use) as f64);
// Record current load level as gauge (0=Low, 1=Medium, 2=High, 3=Critical)
let load_level_value = match io_strategy.load_level {
crate::storage::concurrency::IoLoadLevel::Low => 0.0,
@@ -1135,6 +1232,8 @@ impl DefaultObjectUsecase {
gauge!("rustfs.io.buffer.multiplier").set(io_strategy.buffer_multiplier);
// Count strategy selections by load level
counter!("rustfs.io.strategy.selected", "level" => format!("{:?}", io_strategy.load_level)).increment(1);
// Record I/O priority
counter!("rustfs.io.priority.assigned", "priority" => io_priority.as_str()).increment(1);
}
// Log strategy details at debug level for troubleshooting
@@ -1144,9 +1243,25 @@ impl DefaultObjectUsecase {
buffer_size = io_strategy.buffer_size,
readahead = io_strategy.enable_readahead,
cache_wb = io_strategy.cache_writeback_enabled,
priority = io_priority.as_str(),
"Adaptive I/O strategy calculated"
);
// Check timeout before reading object
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"
);
deadlock_detector.unregister_request(&request_id);
#[cfg(feature = "metrics")]
metrics::counter!("rustfs.get.object.timeout.total", "stage" => "before_read").increment(1);
return Err(s3_error!(InternalError, "Request timeout before reading object"));
}
let reader = store
.get_object_reader(bucket.as_str(), key.as_str(), rs.clone(), h, &opts)
.await
@@ -1485,11 +1600,26 @@ impl DefaultObjectUsecase {
histogram!("get.object.buffer.size.bytes").record(optimal_buffer_size as f64);
}
// Check for timeout before returning
if wrapper.is_timeout() {
warn!(
"GetObject request exceeded timeout: key={} duration={:?} timeout={:?}",
cache_key,
wrapper.elapsed(),
timeout_config.get_object_timeout
);
#[cfg(feature = "metrics")]
counter!("rustfs.get.object.timeout.total").increment(1);
}
debug!(
"GetObject completed: key={} size={} duration={:?} buffer={}",
cache_key, response_content_length, total_duration, optimal_buffer_size
);
// Unregister from deadlock detector
deadlock_detector.unregister_request(&request_id);
let response = wrap_response_with_cors(&bucket, &req.method, &req.headers, output).await;
let result = Ok(response);
let _ = helper.complete(&result);
+9
View File
@@ -452,6 +452,15 @@ async fn run(config: config::Config) -> Result<()> {
Err(e) => error!(target: "rustfs::main::run","Failed to start audit system: {}", e),
}
// Initialize deadlock detector if enabled
let detector = crate::storage::deadlock_detector::get_deadlock_detector();
if detector.is_enabled() {
detector.start();
info!(target: "rustfs::main::run","Deadlock detector started successfully.");
} else {
info!(target: "rustfs::main::run","Deadlock detector disabled.");
}
let buckets_list = store
.list_bucket(&BucketOptions {
no_metadata: true,
+495
View File
@@ -0,0 +1,495 @@
// 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.
// 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
//! - Backpressure state monitoring and events
//! - Prometheus metrics for backpressure events
//! - Graceful handling of slow consumers
//!
//! # Architecture
//!
//! ```text
//! [Disk Reader] --> [BackpressurePipe] --> [HTTP Response]
//! |
//! v
//! [Buffer Monitor]
//! |
//! v
//! [High Watermark?] --> Apply Backpressure
//! ```
use std::sync::Arc;
use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
use std::time::Instant;
use tokio::io::{DuplexStream, duplex};
use tracing::{debug, warn};
#[cfg(feature = "metrics")]
use metrics::counter;
/// Backpressure pipe configuration.
#[derive(Debug, Clone)]
pub struct BackpressureConfig {
/// 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 BackpressureConfig {
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 BackpressureConfig {
/// 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
}
}
/// 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"),
}
}
}
/// 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 {
/// 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.
pub state: BackpressureState,
}
/// 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: BackpressureConfig,
/// Current buffer usage (approximate, updated on write).
buffer_usage: Arc<AtomicUsize>,
/// Current backpressure state.
state: Arc<AtomicBool>, // true = in high watermark state
/// Total bytes written.
total_written: Arc<AtomicUsize>,
/// Total bytes read.
total_read: Arc<AtomicUsize>,
}
impl BackpressurePipe {
/// Create a new backpressure-aware pipe with default configuration.
pub fn new() -> Self {
Self::with_config(BackpressureConfig::from_env())
}
/// Create a new backpressure-aware pipe with custom configuration.
pub fn with_config(config: BackpressureConfig) -> Self {
let (reader, writer) = duplex(config.buffer_size);
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(),
"Created backpressure pipe"
);
Self {
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)),
}
}
/// 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::Relaxed) {
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 {
buffer_capacity: self.config.buffer_size,
buffer_used,
usage_percent: usage_percent as f32,
state: self.state(),
}
}
/// 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();
}
/// 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();
}
/// 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();
if usage >= threshold && !self.state.load(Ordering::Relaxed) {
self.state.store(true, Ordering::Relaxed);
#[cfg(feature = "metrics")]
counter!("rustfs.backpressure.events.total", "state" => "high_watermark").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);
#[cfg(feature = "metrics")]
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"
);
}
}
/// 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: BackpressureConfig,
/// Current buffer usage.
buffer_usage: Arc<AtomicUsize>,
/// In high watermark state.
in_high_watermark: Arc<AtomicBool>,
}
impl BackpressureMonitor {
/// Create a new monitor with default configuration.
pub fn new() -> Self {
Self::with_config(BackpressureConfig::from_env())
}
/// Create a new monitor with custom configuration.
pub fn with_config(config: BackpressureConfig) -> Self {
Self {
config,
buffer_usage: Arc::new(AtomicUsize::new(0)),
in_high_watermark: Arc::new(AtomicBool::new(false)),
}
}
/// Record bytes added to buffer.
pub fn on_write(&self, bytes: usize) -> BackpressureState {
self.buffer_usage.fetch_add(bytes, Ordering::Relaxed);
self.update_state()
}
/// Record bytes removed from buffer.
pub fn on_read(&self, bytes: usize) -> BackpressureState {
self.buffer_usage.fetch_sub(bytes, Ordering::Relaxed);
self.update_state()
}
/// Get current state.
pub fn state(&self) -> BackpressureState {
if self.in_high_watermark.load(Ordering::Relaxed) {
BackpressureState::HighWatermark
} else {
BackpressureState::Normal
}
}
/// Get current buffer usage.
pub fn usage(&self) -> usize {
self.buffer_usage.load(Ordering::Relaxed)
}
/// 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
}
}
/// 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();
if usage >= high {
if !self.in_high_watermark.swap(true, Ordering::Relaxed) {
#[cfg(feature = "metrics")]
counter!("rustfs.backpressure.events.total", "state" => "high_watermark").increment(1);
debug!(usage_percent = self.usage_percent() as u32, "Backpressure: entered high watermark");
}
BackpressureState::HighWatermark
} else if usage <= low {
if self.in_high_watermark.swap(false, Ordering::Relaxed) {
#[cfg(feature = "metrics")]
counter!("rustfs.backpressure.events.total", "state" => "normal").increment(1);
debug!(usage_percent = self.usage_percent() as u32, "Backpressure: returned to normal");
}
BackpressureState::Normal
} else {
self.state()
}
}
}
impl Default for BackpressureMonitor {
fn default() -> Self {
Self::new()
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn test_backpressure_config_default() {
let config = BackpressureConfig::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 = BackpressureConfig {
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_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 = BackpressureConfig {
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);
}
#[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);
}
}
File diff suppressed because it is too large Load Diff
File diff suppressed because it is too large Load Diff
+717
View File
@@ -0,0 +1,717 @@
// 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.
//! Concurrency manager for coordinating concurrent GetObject requests.
use super::io_schedule::{
IoLoadLevel, IoLoadMetrics, IoPriority, IoPriorityQueue, IoPriorityQueueConfig, IoQueueStatus, IoStrategy,
get_advanced_buffer_size,
};
use super::object_cache::{CacheStats, CachedGetObject, CachedObject, HotObjectCache};
use super::request_guard::GetObjectGuard;
use rustfs_config::{KI_B, MI_B};
use std::sync::{Arc, LazyLock, Mutex};
use std::time::Duration;
use tokio::sync::Semaphore;
use tracing::debug;
/// Global concurrency manager instance
pub(crate) static CONCURRENCY_MANAGER: LazyLock<ConcurrencyManager> = LazyLock::new(ConcurrencyManager::new);
#[derive(Clone)]
pub struct ConcurrencyManager {
/// Hot object cache for frequently accessed objects
cache: Arc<HotObjectCache>,
/// Semaphore to limit concurrent disk reads
disk_read_semaphore: Arc<Semaphore>,
/// Whether object caching is enabled (from RUSTFS_OBJECT_CACHE_ENABLE env var)
cache_enabled: bool,
/// I/O load metrics for adaptive strategy calculation
io_metrics: Arc<Mutex<IoLoadMetrics>>,
/// I/O priority queue for request scheduling
#[allow(dead_code)]
priority_queue: Arc<IoPriorityQueue<()>>,
}
impl std::fmt::Debug for ConcurrencyManager {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
use std::sync::atomic::Ordering;
let io_metrics_info = if let Ok(metrics) = self.io_metrics.lock() {
format!("avg_wait={:?}, observations={}", metrics.average_wait(), metrics.observation_count())
} else {
"locked".to_string()
};
f.debug_struct("ConcurrencyManager")
.field("active_requests", &super::io_schedule::ACTIVE_GET_REQUESTS.load(Ordering::Relaxed))
.field("disk_read_permits", &self.disk_read_semaphore.available_permits())
.field("io_metrics", &io_metrics_info)
.finish()
}
}
#[allow(dead_code)]
impl ConcurrencyManager {
/// Create a new concurrency manager with default settings
///
/// Reads configuration from environment variables:
/// - `RUSTFS_OBJECT_CACHE_ENABLE`: Enable/disable object caching (default: false)
pub fn new() -> Self {
let cache_enabled =
rustfs_utils::get_env_bool(rustfs_config::ENV_OBJECT_CACHE_ENABLE, rustfs_config::DEFAULT_OBJECT_CACHE_ENABLE);
let max_disk_reads = rustfs_utils::get_env_usize(
rustfs_config::ENV_OBJECT_MAX_CONCURRENT_DISK_READS,
rustfs_config::DEFAULT_OBJECT_MAX_CONCURRENT_DISK_READS,
);
Self {
cache: Arc::new(HotObjectCache::new()),
disk_read_semaphore: Arc::new(Semaphore::new(max_disk_reads)),
cache_enabled,
io_metrics: Arc::new(Mutex::new(IoLoadMetrics::new(100))), // Keep last 100 observations
priority_queue: Arc::new(IoPriorityQueue::new(IoPriorityQueueConfig::default())),
}
}
/// Check if object caching is enabled
///
/// Returns true if the `RUSTFS_OBJECT_CACHE_ENABLE` environment variable
/// is set to "true" (case-insensitive). When disabled, cache lookups and
/// writebacks are skipped, reducing memory usage at the cost of repeated
/// disk reads for the same objects.
///
/// # Returns
///
/// `true` if caching is enabled, `false` otherwise
pub fn is_cache_enabled(&self) -> bool {
self.cache_enabled
}
/// Track a GetObject request
pub fn track_request() -> GetObjectGuard {
GetObjectGuard::new()
}
/// Try to get an object from cache
pub async fn get_cached(&self, key: &str) -> Option<Arc<Vec<u8>>> {
self.cache.get(key).await
}
/// Cache an object for future retrievals
pub async fn cache_object(&self, key: String, data: Vec<u8>) {
let size = data.len();
let cached_obj = Arc::new(CachedObject::new_with_size(data, size));
self.cache.put(key, cached_obj).await;
}
/// Acquire a permit to perform a disk read operation
///
/// This ensures we don't overwhelm the disk subsystem with too many
/// concurrent reads, which can cause performance degradation.
pub async fn acquire_disk_read_permit(&self) -> Result<tokio::sync::SemaphorePermit<'_>, tokio::sync::AcquireError> {
self.disk_read_semaphore.acquire().await
}
// ============================================
// Adaptive I/O Strategy Methods
// ============================================
/// Record a disk permit wait observation for load tracking.
///
/// This method updates the rolling metrics used to calculate adaptive I/O
/// strategies. Should be called after each disk permit acquisition.
///
/// # Arguments
///
/// * `wait_duration` - Time spent waiting for the disk read permit
pub fn record_permit_wait(&self, wait_duration: Duration) {
if let Ok(mut metrics) = self.io_metrics.lock() {
metrics.record(wait_duration);
}
// Record histogram metric for Prometheus
#[cfg(all(feature = "metrics", not(test)))]
{
use metrics::histogram;
histogram!("rustfs.disk.permit.wait.duration.seconds").record(wait_duration.as_secs_f64());
}
}
/// Calculate an adaptive I/O strategy based on disk permit wait time.
///
/// This method analyzes the permit wait duration to determine the current
/// I/O load level and returns optimized parameters for the read operation.
///
/// # Arguments
///
/// * `permit_wait_duration` - Time spent waiting for disk read permit
/// * `base_buffer_size` - Base buffer size from workload configuration
///
/// # Returns
///
/// An `IoStrategy` containing optimized I/O parameters.
///
/// # Example
///
/// ```ignore
/// let permit_wait_start = Instant::now();
/// let _permit = manager.acquire_disk_read_permit().await;
/// let permit_wait_duration = permit_wait_start.elapsed();
///
/// let strategy = manager.calculate_io_strategy(permit_wait_duration, 256 * 1024);
/// let optimal_buffer = strategy.buffer_size;
/// ```
pub fn calculate_io_strategy(&self, permit_wait_duration: Duration, base_buffer_size: usize) -> IoStrategy {
// Record the observation for future smoothing
self.record_permit_wait(permit_wait_duration);
// Calculate strategy from the current wait duration
IoStrategy::from_wait_duration(permit_wait_duration, base_buffer_size)
}
/// Get the smoothed I/O load level based on recent observations.
///
/// This uses the rolling window of permit wait times to provide a more
/// stable estimate of the current load level, reducing oscillation from
/// transient spikes.
///
/// # Returns
///
/// The smoothed `IoLoadLevel` based on average recent wait times.
pub fn smoothed_load_level(&self) -> IoLoadLevel {
if let Ok(metrics) = self.io_metrics.lock() {
metrics.smoothed_load_level()
} else {
IoLoadLevel::Medium // Default to medium if lock fails
}
}
/// Get I/O load statistics for monitoring.
///
/// Returns statistics about recent disk permit wait times for
/// monitoring dashboards and capacity planning.
///
/// # Returns
///
/// A tuple of (average_wait, p95_wait, max_wait, observation_count)
pub fn io_load_stats(&self) -> (Duration, Duration, Duration, u64) {
if let Ok(metrics) = self.io_metrics.lock() {
(
metrics.average_wait(),
metrics.p95_wait(),
metrics.max_wait(),
metrics.observation_count(),
)
} else {
(Duration::ZERO, Duration::ZERO, Duration::ZERO, 0)
}
}
/// Get the recommended buffer size based on current I/O load.
///
/// This is a convenience method that combines load level detection with
/// buffer size calculation. Uses the smoothed load level for stability.
///
/// # Arguments
///
/// * `base_buffer_size` - Base buffer size from workload configuration
///
/// # Returns
///
/// Recommended buffer size in bytes.
pub fn adaptive_buffer_size(&self, base_buffer_size: usize) -> usize {
let load_level = self.smoothed_load_level();
let multiplier = match load_level {
IoLoadLevel::Low => 1.0,
IoLoadLevel::Medium => 0.75,
IoLoadLevel::High => 0.5,
IoLoadLevel::Critical => 0.4,
};
let buffer_size = ((base_buffer_size as f64) * multiplier) as usize;
buffer_size.clamp(32 * KI_B, MI_B)
}
/// Get cache statistics
pub async fn cache_stats(&self) -> CacheStats {
self.cache.stats().await
}
/// Clear all cached objects
pub async fn clear_cache(&self) {
self.cache.clear().await;
}
/// Check if a key is cached
pub async fn is_cached(&self, key: &str) -> bool {
self.cache.contains(key).await
}
/// Get multiple cached objects in a single operation
pub async fn get_cached_batch(&self, keys: &[String]) -> Vec<Option<Arc<Vec<u8>>>> {
self.cache.get_batch(keys).await
}
/// Remove a specific object from cache
pub async fn remove_cached(&self, key: &str) -> bool {
self.cache.remove(key).await
}
/// Get the most frequently accessed keys
pub async fn get_hot_keys(&self, limit: usize) -> Vec<(String, u64)> {
self.cache.get_hot_keys(limit).await
}
/// Get cache hit rate percentage
pub fn cache_hit_rate(&self) -> f64 {
self.cache.hit_rate()
}
/// Warm up cache with frequently accessed objects
///
/// This can be called during server startup or maintenance windows
/// to pre-populate the cache with known hot objects.
pub async fn warm_cache(&self, objects: Vec<(String, Vec<u8>)>) {
self.cache.warm(objects).await;
}
/// Get optimized buffer size for a request
///
/// This wraps the advanced buffer sizing logic and makes it accessible
/// through the concurrency manager interface.
pub fn buffer_size(&self, file_size: i64, base: usize, sequential: bool) -> usize {
get_advanced_buffer_size(file_size, base, sequential)
}
// ============================================
// Response Cache Methods (CachedGetObject)
// ============================================
/// Get a cached GetObject response with full metadata
///
/// This method retrieves a complete GetObject response from the response cache,
/// including body data and all response metadata (e_tag, last_modified, content_type, etc.).
///
/// # Arguments
///
/// * `key` - Cache key in the format "{bucket}/{key}" or "{bucket}/{key}?versionId={version_id}"
///
/// # Returns
///
/// * `Some(Arc<CachedGetObject>)` - Cached response data if found and not expired
/// * `None` - Cache miss
///
/// # Example
///
/// ```ignore
/// let cache_key = format!("{}/{}", bucket, key);
/// if let Some(cached) = manager.get_cached_object(&cache_key).await {
/// // Build response from cached data
/// let output = GetObjectOutput {
/// body: Some(StreamingBlob::from(cached.body.clone())),
/// content_length: Some(cached.content_length),
/// e_tag: cached.e_tag.clone(),
/// last_modified: cached.last_modified.as_ref().map(|s| parse_rfc3339(s)),
/// ..Default::default()
/// };
/// }
/// ```
pub async fn get_cached_object(&self, key: &str) -> Option<Arc<CachedGetObject>> {
self.cache.get_response(key).await
}
/// Cache a complete GetObject response for future retrievals
///
/// This method caches a complete GetObject response including body and all metadata.
/// Objects larger than the maximum cache size (10MB by default) or empty objects
/// are not cached.
///
/// # Arguments
///
/// * `key` - Cache key in the format "{bucket}/{key}" or "{bucket}/{key}?versionId={version_id}"
/// * `response` - The complete cached response to store
///
/// # Example
///
/// ```ignore
/// let cached = CachedGetObject {
/// body: Bytes::from(data),
/// content_length: data.len() as i64,
/// content_type: Some("application/octet-stream".to_string()),
/// e_tag: Some("\"abc123\"".to_string()),
/// last_modified: Some("2024-01-01T00:00:00Z".to_string()),
/// ..Default::default()
/// };
/// manager.put_cached_object(cache_key, cached).await;
/// ```
pub async fn put_cached_object(&self, key: String, response: CachedGetObject) {
self.cache.put_response(key, response).await;
}
/// Invalidate cache entries for a specific object
///
/// This method removes both simple byte cache and response cache entries
/// for the given key. Should be called after write operations (put_object,
/// copy_object, delete_object, etc.) to prevent stale data from being served.
///
/// # Arguments
///
/// * `key` - Cache key to invalidate (e.g., "{bucket}/{key}")
///
/// # Example
///
/// ```ignore
/// // After put_object succeeds
/// let cache_key = format!("{}/{}", bucket, key);
/// manager.invalidate_cache(&cache_key).await;
/// ```
pub async fn invalidate_cache(&self, key: &str) {
self.cache.invalidate(key).await;
}
/// Invalidate cache entries for an object and its latest version
///
/// For versioned buckets, this invalidates both:
/// - The specific version key: "{bucket}/{key}?versionId={version_id}"
/// - The latest version key: "{bucket}/{key}"
///
/// This ensures that after a write/delete, clients don't receive stale data.
/// Should be called after any write operation that modifies object data or creates
/// new versions.
///
/// # Arguments
///
/// * `bucket` - Bucket name
/// * `key` - Object key
/// * `version_id` - Optional version ID (if None, only invalidates the base key)
///
/// # Example
///
/// ```ignore
/// // After delete_object with version
/// manager.invalidate_cache_versioned(&bucket, &key, Some(&version_id)).await;
///
/// // After put_object (invalidates latest)
/// manager.invalidate_cache_versioned(&bucket, &key, None).await;
/// ```
pub async fn invalidate_cache_versioned(&self, bucket: &str, key: &str, version_id: Option<&str>) {
self.cache.invalidate_versioned(bucket, key, version_id).await;
}
/// Generate a cache key for an object
///
/// Creates a cache key in the appropriate format based on whether a version ID
/// is specified. For versioned requests, uses "{bucket}/{key}?versionId={version_id}".
/// For non-versioned requests, uses "{bucket}/{key}".
///
/// # Arguments
///
/// * `bucket` - Bucket name
/// * `key` - Object key
/// * `version_id` - Optional version ID
///
/// # Returns
///
/// Cache key string
pub fn make_cache_key(bucket: &str, key: &str, version_id: Option<&str>) -> String {
match version_id {
Some(vid) => format!("{bucket}/{key}?versionId={vid}"),
None => format!("{bucket}/{key}"),
}
}
/// Get maximum cacheable object size
///
/// Returns the maximum size in bytes for objects that can be cached.
/// Objects larger than this size are not cached to prevent memory exhaustion.
pub fn max_object_size(&self) -> usize {
self.cache.max_object_size()
}
// ============================================
// Priority-Based I/O Scheduling Methods
// ============================================
/// Get I/O priority for a request based on its size.
///
/// This enables priority-based scheduling where small requests
/// are processed before large requests to prevent starvation.
///
/// # Arguments
///
/// * `request_size` - Size of the request in bytes (-1 if unknown)
///
/// # Returns
///
/// Priority level (High, Normal, or Low)
pub fn get_io_priority(&self, request_size: i64) -> IoPriority {
if request_size < 0 {
// Unknown size, use normal priority
IoPriority::Normal
} else {
IoPriority::from_size(request_size)
}
}
/// Check if priority scheduling is enabled.
pub fn is_priority_scheduling_enabled(&self) -> bool {
rustfs_utils::get_env_bool(
rustfs_config::ENV_OBJECT_PRIORITY_SCHEDULING_ENABLE,
rustfs_config::DEFAULT_OBJECT_PRIORITY_SCHEDULING_ENABLE,
)
}
/// Get current I/O queue status for monitoring.
///
/// Returns information about permit usage and waiting requests.
pub fn io_queue_status(&self) -> IoQueueStatus {
let total_permits = rustfs_utils::get_env_usize(
rustfs_config::ENV_OBJECT_MAX_CONCURRENT_DISK_READS,
rustfs_config::DEFAULT_OBJECT_MAX_CONCURRENT_DISK_READS,
);
let permits_in_use = total_permits.saturating_sub(self.disk_read_semaphore.available_permits());
IoQueueStatus {
total_permits,
permits_in_use,
high_priority_waiting: 0, // Would need additional tracking
normal_priority_waiting: 0,
low_priority_waiting: 0,
high_priority_processed: 0,
normal_priority_processed: 0,
low_priority_processed: 0,
starvation_events: 0,
}
}
/// Acquire a disk read permit with priority awareness.
///
/// When priority scheduling is enabled, this method logs the priority
/// for observability. The actual acquisition uses the same semaphore
/// but priority information is used for monitoring.
///
/// # Arguments
///
/// * `priority` - Priority level for this request
///
/// # Returns
///
/// Semaphore permit on success, error on failure
pub async fn acquire_priority_permit(
&self,
priority: IoPriority,
) -> Result<tokio::sync::SemaphorePermit<'_>, tokio::sync::AcquireError> {
#[cfg(feature = "metrics")]
{
use metrics::counter;
counter!("rustfs.disk.read.queue.total", "priority" => priority.as_str()).increment(1);
}
debug!(
priority = %priority,
available_permits = self.disk_read_semaphore.available_permits(),
"Acquiring disk read permit"
);
self.disk_read_semaphore.acquire().await
}
/// Get the global concurrency manager instance.
pub fn global() -> &'static Self {
&CONCURRENCY_MANAGER
}
}
impl Default for ConcurrencyManager {
fn default() -> Self {
Self::new()
}
}
// ============================================
// Integration Tests for ConcurrencyManager
// ============================================
#[cfg(test)]
mod integration_tests {
use super::*;
use bytes::Bytes;
use serial_test::serial;
#[tokio::test]
#[serial]
async fn test_concurrency_manager_priority_queue_integration() {
let manager = ConcurrencyManager::new();
// Test priority determination
let small_size = 100 * 1024; // 100KB
let large_size = 200 * 1024 * 1024; // 200MB
let small_priority = manager.get_io_priority(small_size as i64);
let large_priority = manager.get_io_priority(large_size as i64);
assert_eq!(small_priority, IoPriority::High);
assert_eq!(large_priority, IoPriority::Low);
}
#[tokio::test]
#[serial]
async fn test_concurrency_manager_cache_operations() {
let manager = ConcurrencyManager::new();
// Test cache put and get
let obj = CachedGetObject::new(Bytes::from("test data"), 9)
.with_content_type("text/plain".to_string())
.with_e_tag("\"abc123\"".to_string());
manager.put_cached_object("test-key".to_string(), obj).await;
let cached = manager.get_cached_object("test-key").await;
assert!(cached.is_some());
let cached_obj = cached.unwrap();
assert_eq!(cached_obj.content_type, Some("text/plain".to_string()));
assert_eq!(cached_obj.e_tag, Some("\"abc123\"".to_string()));
}
#[tokio::test]
#[serial]
async fn test_concurrency_manager_cache_stats() {
let manager = ConcurrencyManager::new();
// Add some objects
for i in 0..5 {
let obj = CachedGetObject::new(Bytes::from(format!("data{}", i)), 5);
manager.put_cached_object(format!("key{}", i), obj).await;
}
// Get stats
let stats = manager.cache_stats().await;
assert!(stats.entries >= 5);
}
#[tokio::test]
#[serial]
async fn test_concurrency_manager_io_queue_status() {
let manager = ConcurrencyManager::new();
let status = manager.io_queue_status();
// Initial state should have no waiting requests
assert_eq!(status.high_priority_waiting, 0);
assert_eq!(status.normal_priority_waiting, 0);
assert_eq!(status.low_priority_waiting, 0);
}
#[tokio::test]
#[serial]
async fn test_concurrency_manager_disk_read_permit() {
let manager = ConcurrencyManager::new();
// Acquire permit
let permit = manager.acquire_disk_read_permit().await;
assert!(permit.is_ok());
// Permit should be released when dropped
drop(permit);
// Should be able to acquire again
let permit2 = manager.acquire_disk_read_permit().await;
assert!(permit2.is_ok());
}
#[tokio::test]
#[serial]
async fn test_concurrency_manager_priority_scheduling() {
let manager = ConcurrencyManager::new();
// Test if priority scheduling is enabled
let enabled = manager.is_priority_scheduling_enabled();
assert!(enabled); // Should be enabled by default
// Test priority determination for different sizes
assert_eq!(manager.get_io_priority(500 * 1024), IoPriority::High); // 500KB
assert_eq!(manager.get_io_priority(5 * 1024 * 1024), IoPriority::Normal); // 5MB
assert_eq!(manager.get_io_priority(50 * 1024 * 1024), IoPriority::Low); // 50MB
}
#[tokio::test]
#[serial]
async fn test_concurrency_manager_cache_invalidation() {
let manager = ConcurrencyManager::new();
// Add an object
let obj = CachedGetObject::new(Bytes::from("test"), 4);
manager.put_cached_object("test-key".to_string(), obj).await;
// Verify it's cached
assert!(manager.is_cached("test-key").await);
// Invalidate
manager.invalidate_cache("test-key").await;
// Should not be cached anymore
assert!(!manager.is_cached("test-key").await);
}
#[tokio::test]
#[serial]
async fn test_concurrency_manager_cache_clear() {
let manager = ConcurrencyManager::new();
// Add multiple objects
for i in 0..10 {
let obj = CachedGetObject::new(Bytes::from(format!("data{}", i)), 5);
manager.put_cached_object(format!("key{}", i), obj).await;
}
// Clear cache
manager.clear_cache().await;
// Verify all are removed
let stats = manager.cache_stats().await;
assert_eq!(stats.entries, 0);
}
#[tokio::test]
#[serial]
async fn test_concurrency_manager_io_strategy() {
let manager = ConcurrencyManager::new();
// Test I/O strategy calculation
let low_wait = Duration::from_millis(5);
let high_wait = Duration::from_millis(100);
let strategy_low = manager.calculate_io_strategy(low_wait, 128 * 1024);
let strategy_high = manager.calculate_io_strategy(high_wait, 128 * 1024);
// Under low load, should use larger buffers
assert!(strategy_low.buffer_size >= strategy_high.buffer_size);
}
#[tokio::test]
#[serial]
async fn test_concurrency_manager_adaptive_buffer_size() {
let manager = ConcurrencyManager::new();
let base_size = 128 * 1024; // 128KB
// Test adaptive buffer sizing
let size1 = manager.adaptive_buffer_size(base_size);
// Should return a reasonable buffer size
assert!(size1 > 0);
assert!(size1 <= 2 * 1024 * 1024); // Not more than 2MB
}
}
+57
View File
@@ -0,0 +1,57 @@
// 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.
//! Concurrency optimization module for high-performance object retrieval.
// Sub-modules
pub mod io_schedule;
pub mod manager;
pub mod object_cache;
pub mod request_guard;
// ============================================
// Public API Re-exports
// ============================================
// I/O scheduling types
#[allow(unused_imports)]
pub use io_schedule::{
IO_PRIORITY_METRICS, IoLoadLevel, IoPriority, IoPriorityMetrics, IoPriorityQueue, IoPriorityQueueConfig, IoQueueStatus,
IoStrategy, get_advanced_buffer_size, get_concurrency_aware_buffer_size,
};
// Request tracking
pub use request_guard::GetObjectGuard;
// Cache types
#[allow(unused_imports)]
pub use object_cache::{CacheHealthStatus, CacheStats, CachedGetObject};
// Concurrency manager
pub use manager::ConcurrencyManager;
// ============================================
// Helper Functions
// ============================================
/// Get the global concurrency manager instance.
pub fn get_concurrency_manager() -> &'static ConcurrencyManager {
ConcurrencyManager::global()
}
/// Reset the active get requests counter (for testing).
#[allow(dead_code)]
pub fn reset_active_get_requests() {
io_schedule::ACTIVE_GET_REQUESTS.store(0, std::sync::atomic::Ordering::Relaxed);
}
File diff suppressed because it is too large Load Diff
@@ -0,0 +1,107 @@
// 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.
//! Request tracking module with RAII guard for concurrent request management.
use std::sync::atomic::Ordering;
use std::time::Instant;
use super::io_schedule::ACTIVE_GET_REQUESTS;
/// RAII guard for tracking active GetObject requests.
#[derive(Debug)]
pub struct GetObjectGuard {
start_time: Instant,
}
impl GetObjectGuard {
/// Create a new guard and increment the concurrent request counter.
pub fn new() -> Self {
ACTIVE_GET_REQUESTS.fetch_add(1, Ordering::Relaxed);
#[cfg(all(feature = "metrics", not(test)))]
if !std::thread::panicking() {
use metrics::counter;
counter!("rustfs.get.object.requests.started").increment(1);
}
Self {
start_time: Instant::now(),
}
}
/// Get the elapsed time since this guard was created.
pub fn elapsed(&self) -> std::time::Duration {
self.start_time.elapsed()
}
/// Get the current number of concurrent GetObject requests.
pub fn concurrent_count() -> usize {
ACTIVE_GET_REQUESTS.load(Ordering::Relaxed)
}
/// Get the current number of concurrent requests (alias for concurrent_count).
pub fn concurrent_requests() -> usize {
Self::concurrent_count()
}
}
impl Default for GetObjectGuard {
fn default() -> Self {
Self::new()
}
}
impl Drop for GetObjectGuard {
fn drop(&mut self) {
if let Err(previous) =
ACTIVE_GET_REQUESTS.fetch_update(Ordering::Relaxed, Ordering::Relaxed, |current| current.checked_sub(1))
{
debug_assert_eq!(
previous, 0,
"ACTIVE_GET_REQUESTS underflow attempt in GetObjectGuard::drop; previous value = {}",
previous
);
}
#[cfg(all(feature = "metrics", not(test)))]
if !std::thread::panicking() {
use metrics::{counter, histogram};
counter!("rustfs.get.object.requests.completed").increment(1);
histogram!("rustfs.get.object.duration.seconds").record(self.elapsed().as_secs_f64());
}
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn test_guard_increments_counter() {
let initial = GetObjectGuard::concurrent_count();
{
let _guard = GetObjectGuard::new();
assert_eq!(GetObjectGuard::concurrent_count(), initial + 1);
}
assert_eq!(GetObjectGuard::concurrent_count(), initial);
}
#[test]
fn test_guard_elapsed() {
let guard = GetObjectGuard::new();
std::thread::sleep(std::time::Duration::from_millis(10));
assert!(guard.elapsed().as_millis() >= 10);
}
}
+338
View File
@@ -0,0 +1,338 @@
// 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.
//! Integration tests for concurrent request fix.
//!
//! These tests verify that the timeout, backpressure, and deadlock detection
//! mechanisms work correctly under high concurrency scenarios.
#[cfg(test)]
mod tests {
use crate::storage::backpressure::{BackpressureConfig, BackpressureMonitor, BackpressureState};
use crate::storage::concurrency::{IoLoadLevel, IoPriority};
use crate::storage::deadlock_detector::{
DeadlockDetector, DeadlockDetectorConfig, LockInfo, LockType, RequestResourceTracker,
};
use crate::storage::lock_optimizer::{LockOptimizeConfig, LockOptimizer, LockStats};
use crate::storage::timeout_wrapper::{RequestTimeoutWrapper, TimedGetObjectResult, TimeoutConfig};
use std::time::Duration;
// ============================================
// Timeout Wrapper Tests
// ============================================
#[tokio::test]
async fn test_timeout_wrapper_completes_within_timeout() {
let config = TimeoutConfig {
get_object_timeout: Duration::from_secs(5),
..Default::default()
};
let wrapper = RequestTimeoutWrapper::new(config);
let result = wrapper
.execute_with_timeout(|_token| async move { Ok::<i32, String>(42) })
.await;
match result {
TimedGetObjectResult::Success(value) => assert_eq!(value, 42),
_ => panic!("Expected Success, got {:?}", result),
}
}
#[tokio::test]
async fn test_timeout_wrapper_times_out() {
let config = TimeoutConfig {
get_object_timeout: Duration::from_millis(50),
..Default::default()
};
let wrapper = RequestTimeoutWrapper::new(config);
let result = wrapper
.execute_with_timeout(|_token| async move {
// Simulate a slow operation
tokio::time::sleep(Duration::from_secs(10)).await;
Ok::<i32, String>(42)
})
.await;
match result {
TimedGetObjectResult::Timeout(info) => {
assert!(info.elapsed >= Duration::from_millis(50));
}
_ => panic!("Expected Timeout, got {:?}", result),
}
}
#[tokio::test]
async fn test_timeout_wrapper_returns_error() {
let config = TimeoutConfig {
get_object_timeout: Duration::from_secs(5),
..Default::default()
};
let wrapper = RequestTimeoutWrapper::new(config);
let result = wrapper
.execute_with_timeout(|_token| async move { Err::<i32, String>("test error".to_string()) })
.await;
match result {
TimedGetObjectResult::Error(e) => assert_eq!(e, "test error"),
_ => panic!("Expected Error, got {:?}", result),
}
}
#[tokio::test]
async fn test_timeout_wrapper_disabled() {
let config = TimeoutConfig {
get_object_timeout: Duration::ZERO,
..Default::default()
};
let wrapper = RequestTimeoutWrapper::new(config);
// This would timeout if enabled
let result = wrapper
.execute_with_timeout(|_token| async move {
tokio::time::sleep(Duration::from_millis(100)).await;
Ok::<i32, String>(42)
})
.await;
match result {
TimedGetObjectResult::Success(value) => assert_eq!(value, 42),
_ => panic!("Expected Success when timeout disabled, got {:?}", result),
}
}
// ============================================
// Backpressure Tests
// ============================================
#[test]
fn test_backpressure_config_defaults() {
let config = BackpressureConfig::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 = BackpressureConfig {
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 = BackpressureConfig {
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
// ============================================
#[test]
fn test_io_priority_from_size() {
// Small request (< 1MB) -> High priority
assert_eq!(IoPriority::from_size(100 * 1024), IoPriority::High);
assert_eq!(IoPriority::from_size(512 * 1024), IoPriority::High);
// Medium request (1MB - 10MB) -> Normal priority
assert_eq!(IoPriority::from_size(2 * 1024 * 1024), IoPriority::Normal);
assert_eq!(IoPriority::from_size(5 * 1024 * 1024), IoPriority::Normal);
// Large request (> 10MB) -> Low priority
assert_eq!(IoPriority::from_size(20 * 1024 * 1024), IoPriority::Low);
assert_eq!(IoPriority::from_size(100 * 1024 * 1024), IoPriority::Low);
}
#[test]
fn test_io_priority_ordering() {
assert!(IoPriority::High < IoPriority::Normal);
assert!(IoPriority::Normal < IoPriority::Low);
}
#[test]
fn test_io_load_level_from_wait() {
assert_eq!(IoLoadLevel::from_wait_duration(Duration::from_millis(5)), IoLoadLevel::Low);
assert_eq!(IoLoadLevel::from_wait_duration(Duration::from_millis(30)), IoLoadLevel::Medium);
assert_eq!(IoLoadLevel::from_wait_duration(Duration::from_millis(100)), IoLoadLevel::High);
assert_eq!(IoLoadLevel::from_wait_duration(Duration::from_millis(500)), IoLoadLevel::Critical);
}
// ============================================
// Deadlock Detector Tests
// ============================================
#[test]
fn test_deadlock_detector_config_defaults() {
let config = DeadlockDetectorConfig::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));
}
#[test]
fn test_request_resource_tracker() {
let tracker = RequestResourceTracker::new("req-1".to_string(), "GetObject bucket/key");
assert!(tracker.elapsed() < Duration::from_secs(1));
assert!(!tracker.is_hung(Duration::from_secs(10)));
}
#[test]
fn test_lock_info_creation() {
let lock = LockInfo {
id: "lock-1".to_string(),
lock_type: LockType::Read,
resource: "bucket/key".to_string(),
acquire_time: std::time::Instant::now(),
};
assert_eq!(format!("{}", lock.lock_type), "read");
}
#[test]
fn test_deadlock_detector_registration() {
let config = DeadlockDetectorConfig {
enabled: true,
..Default::default()
};
let detector = DeadlockDetector::new(config);
detector.register_request("req-1", "Test request");
assert_eq!(detector.tracked_count(), 1);
detector.register_request("req-2", "Another request");
assert_eq!(detector.tracked_count(), 2);
detector.unregister_request("req-1");
assert_eq!(detector.tracked_count(), 1);
detector.unregister_request("req-2");
assert_eq!(detector.tracked_count(), 0);
}
// ============================================
// Integration Tests
// ============================================
#[tokio::test]
async fn test_concurrent_requests_with_timeout() {
use std::sync::Arc;
use tokio::sync::Semaphore;
let semaphore = Arc::new(Semaphore::new(10));
let mut handles = vec![];
// Spawn 20 concurrent "requests"
for i in 0..20 {
let sem = semaphore.clone();
let handle = tokio::spawn(async move {
let _permit = sem.acquire().await.unwrap();
// Simulate some work
tokio::time::sleep(Duration::from_millis(10)).await;
i
});
handles.push(handle);
}
// Wait for all to complete
let results: Vec<_> = futures::future::join_all(handles).await;
assert_eq!(results.len(), 20);
// All should complete successfully
for result in results {
assert!(result.is_ok());
}
}
#[test]
fn test_priority_scheduling_fairness() {
// Simulate priority scheduling
let mut high_priority_count = 0;
let mut normal_priority_count = 0;
let mut low_priority_count = 0;
// Simulate 100 requests of various sizes
for i in 0..100 {
let size = match i % 3 {
0 => 100 * 1024, // Small: 100KB
1 => 5 * 1024 * 1024, // Medium: 5MB
_ => 50 * 1024 * 1024, // Large: 50MB
};
match IoPriority::from_size(size) {
IoPriority::High => high_priority_count += 1,
IoPriority::Normal => normal_priority_count += 1,
IoPriority::Low => low_priority_count += 1,
}
}
// Each priority should have roughly 1/3 of requests
assert_eq!(high_priority_count, 34);
assert_eq!(normal_priority_count, 33);
assert_eq!(low_priority_count, 33);
}
}
+673
View File
@@ -0,0 +1,673 @@
// 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.
//! Deadlock Detection for Concurrent Request Monitoring.
//!
//! This module provides deadlock detection capabilities for diagnosing
//! hanging requests and lock contention issues in production systems.
//!
//! # Key Features
//!
//! - Request resource tracking (locks, memory, file handles)
//! - Lock wait graph analysis for cycle detection
//! - Configurable detection interval and hang threshold
//! - Prometheus metrics for deadlock events
//! - Detailed diagnostic logging
//!
//! # Usage
//!
//! ```ignore
//! use crate::storage::deadlock_detector::{DeadlockDetector, DeadlockDetectorConfig};
//!
//! let config = DeadlockDetectorConfig::from_env();
//! let detector = DeadlockDetector::new(config);
//! detector.start();
//!
//! // Track a request
//! let request_id = detector.register_request();
//! detector.record_lock_acquire(request_id, lock_info);
//!
//! // ... request processing ...
//!
//! detector.unregister_request(request_id);
//! ```
// Allow dead_code for public API that may be used by external modules or future features
#![allow(dead_code)]
use std::collections::{HashMap, HashSet};
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
use std::sync::{Arc, Mutex, RwLock};
use std::time::{Duration, Instant};
use tokio::sync::broadcast;
use tracing::{debug, error, warn};
#[cfg(feature = "metrics")]
use metrics::counter;
/// Request identifier type.
pub type RequestId = String;
/// Lock identifier type.
pub type LockId = String;
/// Deadlock detector configuration.
#[derive(Debug, Clone)]
pub struct DeadlockDetectorConfig {
/// Whether deadlock detection is enabled.
pub enabled: bool,
/// Detection check interval.
pub check_interval: Duration,
/// Hang threshold - requests running longer than this are considered potentially hung.
pub hang_threshold: Duration,
/// Whether to capture backtraces (expensive, only for debugging).
pub capture_backtrace: bool,
}
impl Default for DeadlockDetectorConfig {
fn default() -> Self {
Self {
enabled: rustfs_config::DEFAULT_OBJECT_DEADLOCK_DETECTION_ENABLE,
check_interval: Duration::from_secs(rustfs_config::DEFAULT_OBJECT_DEADLOCK_CHECK_INTERVAL),
hang_threshold: Duration::from_secs(rustfs_config::DEFAULT_OBJECT_DEADLOCK_HANG_THRESHOLD),
capture_backtrace: false,
}
}
}
impl DeadlockDetectorConfig {
/// Load configuration from environment variables.
pub fn from_env() -> Self {
let enabled = rustfs_utils::get_env_bool(
rustfs_config::ENV_OBJECT_DEADLOCK_DETECTION_ENABLE,
rustfs_config::DEFAULT_OBJECT_DEADLOCK_DETECTION_ENABLE,
);
let check_interval = Duration::from_secs(rustfs_utils::get_env_u64(
rustfs_config::ENV_OBJECT_DEADLOCK_CHECK_INTERVAL,
rustfs_config::DEFAULT_OBJECT_DEADLOCK_CHECK_INTERVAL,
));
let hang_threshold = Duration::from_secs(rustfs_utils::get_env_u64(
rustfs_config::ENV_OBJECT_DEADLOCK_HANG_THRESHOLD,
rustfs_config::DEFAULT_OBJECT_DEADLOCK_HANG_THRESHOLD,
));
Self {
enabled,
check_interval,
hang_threshold,
capture_backtrace: false,
}
}
}
/// Lock information for tracking.
#[derive(Debug, Clone)]
pub struct LockInfo {
/// Lock identifier.
pub id: LockId,
/// Lock type (read/write).
pub lock_type: LockType,
/// Resource being locked (bucket/key).
pub resource: String,
/// When the lock was acquired.
pub acquire_time: Instant,
}
/// Lock type.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum LockType {
/// Read lock (shared).
Read,
/// Write lock (exclusive).
Write,
}
impl std::fmt::Display for LockType {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
LockType::Read => write!(f, "read"),
LockType::Write => write!(f, "write"),
}
}
}
/// Resource type being tracked.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
pub enum ResourceType {
/// Lock resource.
Lock,
/// Memory resource.
Memory,
/// File handle.
FileHandle,
/// I/O permit.
IoPermit,
}
/// Request resource tracker.
#[derive(Debug, Clone)]
pub struct RequestResourceTracker {
/// Request ID.
pub request_id: RequestId,
/// When the request started.
pub start_time: Instant,
/// Locks currently held by this request.
pub held_locks: Vec<LockInfo>,
/// Lock this request is waiting for (if any).
pub waiting_lock: Option<LockInfo>,
/// Resources held by this request.
pub resources: HashMap<ResourceType, usize>,
/// Request description (e.g., "GetObject bucket/key").
pub description: String,
}
impl RequestResourceTracker {
/// Create a new tracker for a request.
pub fn new(request_id: RequestId, description: impl Into<String>) -> Self {
Self {
request_id,
start_time: Instant::now(),
held_locks: Vec::new(),
waiting_lock: None,
resources: HashMap::new(),
description: description.into(),
}
}
/// Get the elapsed time since the request started.
pub fn elapsed(&self) -> Duration {
self.start_time.elapsed()
}
/// Check if this request is potentially hung.
pub fn is_hung(&self, threshold: Duration) -> bool {
self.elapsed() > threshold
}
}
/// Deadlock detection result.
#[derive(Debug, Clone)]
pub struct DeadlockInfo {
/// Time of detection.
pub detected_at: Instant,
/// Requests involved in the deadlock cycle.
pub cycle: Vec<RequestId>,
/// Lock wait graph showing the deadlock.
pub wait_graph: Vec<WaitGraphEdge>,
/// Resource usage at detection time.
pub resource_usage: ResourceUsage,
}
/// Edge in the lock wait graph.
#[derive(Debug, Clone)]
pub struct WaitGraphEdge {
/// Request that is waiting.
pub from: RequestId,
/// Request that is blocking (holds the lock).
pub to: RequestId,
/// The lock being waited for.
pub lock_id: LockId,
}
/// Resource usage snapshot.
#[derive(Debug, Clone, Default)]
pub struct ResourceUsage {
/// Total memory used (bytes).
pub memory_bytes: usize,
/// Total file handles open.
pub open_files: usize,
/// Total active requests.
pub active_requests: usize,
/// Total locks held.
pub locks_held: usize,
}
/// Deadlock detector.
pub struct DeadlockDetector {
/// Configuration.
config: DeadlockDetectorConfig,
/// Active request trackers.
requests: Arc<RwLock<HashMap<RequestId, RequestResourceTracker>>>,
/// Detection task handle.
detector_task: Arc<Mutex<Option<tokio::task::JoinHandle<()>>>>,
/// Shutdown signal.
shutdown_tx: broadcast::Sender<()>,
/// Total deadlocks detected.
deadlocks_detected: Arc<AtomicU64>,
/// Is currently running.
running: Arc<AtomicBool>,
}
impl DeadlockDetector {
/// Create a new deadlock detector.
pub fn new(config: DeadlockDetectorConfig) -> Self {
let (shutdown_tx, _) = broadcast::channel(1);
Self {
config,
requests: Arc::new(RwLock::new(HashMap::new())),
detector_task: Arc::new(Mutex::new(None)),
shutdown_tx,
deadlocks_detected: Arc::new(AtomicU64::new(0)),
running: Arc::new(AtomicBool::new(false)),
}
}
/// Check if detection is enabled.
pub fn is_enabled(&self) -> bool {
self.config.enabled
}
/// Start the detection task.
pub fn start(&self) {
if !self.config.enabled {
debug!("Deadlock detection is disabled");
return;
}
if self.running.swap(true, Ordering::Relaxed) {
debug!("Deadlock detector already running");
return;
}
let requests = self.requests.clone();
let config = self.config.clone();
let deadlocks_detected = self.deadlocks_detected.clone();
let mut shutdown_rx = self.shutdown_tx.subscribe();
let running = self.running.clone();
let handle = tokio::spawn(async move {
debug!(
check_interval_secs = config.check_interval.as_secs(),
hang_threshold_secs = config.hang_threshold.as_secs(),
"Deadlock detector started"
);
loop {
tokio::select! {
_ = tokio::time::sleep(config.check_interval) => {
Self::detect_cycle(&requests, &config, &deadlocks_detected);
}
_ = shutdown_rx.recv() => {
debug!("Deadlock detector shutting down");
break;
}
}
}
running.store(false, Ordering::Relaxed);
});
*self.detector_task.lock().unwrap() = Some(handle);
}
/// Stop the detection task.
pub fn stop(&self) {
let _ = self.shutdown_tx.send(());
if let Some(handle) = self.detector_task.lock().unwrap().take() {
// Don't await the handle as we're in a non-async context
handle.abort();
}
self.running.store(false, Ordering::Relaxed);
}
/// Register a new request for tracking.
pub fn register_request(&self, request_id: impl Into<String>, description: impl Into<String>) {
if !self.config.enabled {
return;
}
let request_id = request_id.into();
let tracker = RequestResourceTracker::new(request_id.clone(), description);
self.requests.write().unwrap().insert(request_id.clone(), tracker);
debug!(request_id = %request_id, "Request registered for deadlock tracking");
}
/// Unregister a request (it completed or was cancelled).
pub fn unregister_request(&self, request_id: &str) {
if !self.config.enabled {
return;
}
self.requests.write().unwrap().remove(request_id);
debug!(request_id = %request_id, "Request unregistered from deadlock tracking");
}
/// Record a lock acquisition.
pub fn record_lock_acquire(&self, request_id: &str, lock: LockInfo) {
if !self.config.enabled {
return;
}
if let Some(tracker) = self.requests.write().unwrap().get_mut(request_id) {
tracker.held_locks.push(lock);
}
}
/// Record a lock release.
pub fn record_lock_release(&self, request_id: &str, lock_id: &LockId) {
if !self.config.enabled {
return;
}
if let Some(tracker) = self.requests.write().unwrap().get_mut(request_id) {
tracker.held_locks.retain(|l| &l.id != lock_id);
}
}
/// Record that a request is waiting for a lock.
pub fn record_lock_wait(&self, request_id: &str, lock: LockInfo) {
if !self.config.enabled {
return;
}
if let Some(tracker) = self.requests.write().unwrap().get_mut(request_id) {
tracker.waiting_lock = Some(lock);
}
}
/// Clear the waiting lock (acquired or gave up).
pub fn clear_lock_wait(&self, request_id: &str) {
if !self.config.enabled {
return;
}
if let Some(tracker) = self.requests.write().unwrap().get_mut(request_id) {
tracker.waiting_lock = None;
}
}
/// Record resource usage.
pub fn record_resource(&self, request_id: &str, resource_type: ResourceType, amount: usize) {
if !self.config.enabled {
return;
}
if let Some(tracker) = self.requests.write().unwrap().get_mut(request_id) {
tracker.resources.insert(resource_type, amount);
}
}
/// Get current number of tracked requests.
pub fn tracked_count(&self) -> usize {
self.requests.read().unwrap().len()
}
/// Get total deadlocks detected.
pub fn total_detected(&self) -> u64 {
self.deadlocks_detected.load(Ordering::Relaxed)
}
/// Detect deadlock cycles in the lock wait graph.
fn detect_cycle(
requests: &Arc<RwLock<HashMap<RequestId, RequestResourceTracker>>>,
config: &DeadlockDetectorConfig,
deadlocks_detected: &Arc<AtomicU64>,
) {
let requests_guard = requests.read().unwrap();
// Find hung requests
let hung_requests: Vec<_> = requests_guard.values().filter(|r| r.is_hung(config.hang_threshold)).collect();
if hung_requests.is_empty() {
return;
}
// Build lock wait graph
// Edge: request A -> request B means A is waiting for a lock that B holds
let mut wait_graph: Vec<WaitGraphEdge> = Vec::new();
for waiting in &hung_requests {
if let Some(waiting_for) = &waiting.waiting_lock {
// Find who holds this lock
for holding in &hung_requests {
if holding.request_id == waiting.request_id {
continue;
}
if holding.held_locks.iter().any(|l| l.id == waiting_for.id) {
wait_graph.push(WaitGraphEdge {
from: waiting.request_id.clone(),
to: holding.request_id.clone(),
lock_id: waiting_for.id.clone(),
});
}
}
}
}
// Detect cycles using DFS
if let Some(cycle) = Self::find_cycle(&wait_graph) {
deadlocks_detected.fetch_add(1, Ordering::Relaxed);
#[cfg(feature = "metrics")]
counter!("rustfs.deadlock.detected.total").increment(1);
// Log detailed deadlock information
error!(
cycle = ?cycle,
wait_graph = ?wait_graph,
hung_requests_count = hung_requests.len(),
"Deadlock detected: circular lock wait chain found"
);
// Log each request in the cycle
for request_id in &cycle {
if let Some(tracker) = requests_guard.get(request_id) {
warn!(
request_id = %request_id,
description = %tracker.description,
elapsed_secs = tracker.elapsed().as_secs(),
held_locks = ?tracker.held_locks.iter().map(|l| &l.id).collect::<Vec<_>>(),
waiting_lock = ?tracker.waiting_lock.as_ref().map(|l| &l.id),
"Request in deadlock cycle"
);
}
}
} else {
// No cycle, but log hung requests for diagnosis
debug!(
hung_requests_count = hung_requests.len(),
wait_graph_edges = wait_graph.len(),
"Hung requests detected but no deadlock cycle"
);
}
}
/// Find a cycle in the wait graph using DFS.
fn find_cycle(edges: &[WaitGraphEdge]) -> Option<Vec<RequestId>> {
// Build adjacency list
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();
for start in graph.keys() {
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());
}
}
None
}
/// DFS helper for cycle detection.
fn dfs_find_cycle<'a>(
node: &'a RequestId,
graph: &HashMap<&'a RequestId, Vec<&'a RequestId>>,
visited: &mut HashSet<&'a RequestId>,
path: &mut Vec<&'a RequestId>,
path_set: &mut HashSet<&'a RequestId>,
) -> bool {
visited.insert(node);
path.push(node);
path_set.insert(node);
if let Some(neighbors) = graph.get(&node) {
for neighbor in neighbors {
if path_set.contains(neighbor) {
// Found cycle - trim path to just the cycle
let cycle_start = path.iter().position(|n| *n == *neighbor).unwrap();
path.drain(0..cycle_start);
return true;
}
if !visited.contains(neighbor) && Self::dfs_find_cycle(neighbor, graph, visited, path, path_set) {
return true;
}
}
}
path.pop();
path_set.remove(node);
false
}
}
impl Drop for DeadlockDetector {
fn drop(&mut self) {
self.stop();
}
}
/// Global deadlock detector instance.
static DEADLOCK_DETECTOR: std::sync::OnceLock<Arc<DeadlockDetector>> = std::sync::OnceLock::new();
/// Get the global deadlock detector.
pub fn get_deadlock_detector() -> Arc<DeadlockDetector> {
DEADLOCK_DETECTOR
.get_or_init(|| {
let config = DeadlockDetectorConfig::from_env();
Arc::new(DeadlockDetector::new(config))
})
.clone()
}
/// Initialize and start the global deadlock detector.
pub fn init_deadlock_detector() {
let detector = get_deadlock_detector();
detector.start();
}
/// Check if deadlock detection is enabled.
pub fn is_deadlock_detection_enabled() -> bool {
rustfs_utils::get_env_bool(
rustfs_config::ENV_OBJECT_DEADLOCK_DETECTION_ENABLE,
rustfs_config::DEFAULT_OBJECT_DEADLOCK_DETECTION_ENABLE,
)
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn test_deadlock_detector_config_default() {
let config = DeadlockDetectorConfig::default();
assert!(!config.enabled);
assert_eq!(config.check_interval, Duration::from_secs(5));
assert_eq!(config.hang_threshold, Duration::from_secs(10));
}
#[test]
fn test_request_resource_tracker() {
let tracker = RequestResourceTracker::new("req-1".to_string(), "GetObject bucket/key");
assert!(tracker.elapsed() < Duration::from_secs(1));
assert!(!tracker.is_hung(Duration::from_secs(10)));
}
#[test]
fn test_lock_info() {
let lock = LockInfo {
id: "lock-1".to_string(),
lock_type: LockType::Read,
resource: "bucket/key".to_string(),
acquire_time: Instant::now(),
};
assert_eq!(format!("{}", lock.lock_type), "read");
}
#[test]
fn test_deadlock_detector_registration() {
let config = DeadlockDetectorConfig {
enabled: true,
..Default::default()
};
let detector = DeadlockDetector::new(config);
detector.register_request("req-1", "Test request");
assert_eq!(detector.tracked_count(), 1);
detector.unregister_request("req-1");
assert_eq!(detector.tracked_count(), 0);
}
#[test]
fn test_cycle_detection() {
// Create a simple cycle: A -> B -> C -> A
let edges = vec![
WaitGraphEdge {
from: "A".to_string(),
to: "B".to_string(),
lock_id: "lock-1".to_string(),
},
WaitGraphEdge {
from: "B".to_string(),
to: "C".to_string(),
lock_id: "lock-2".to_string(),
},
WaitGraphEdge {
from: "C".to_string(),
to: "A".to_string(),
lock_id: "lock-3".to_string(),
},
];
let cycle = DeadlockDetector::find_cycle(&edges);
assert!(cycle.is_some());
let cycle = cycle.unwrap();
assert!(cycle.len() >= 2);
}
#[test]
fn test_no_cycle() {
// No cycle: A -> B -> C
let edges = vec![
WaitGraphEdge {
from: "A".to_string(),
to: "B".to_string(),
lock_id: "lock-1".to_string(),
},
WaitGraphEdge {
from: "B".to_string(),
to: "C".to_string(),
lock_id: "lock-2".to_string(),
},
];
let cycle = DeadlockDetector::find_cycle(&edges);
assert!(cycle.is_none());
}
}
+451
View File
@@ -0,0 +1,451 @@
// 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.
// Allow dead_code for public API that may be used by external modules or future features
#![allow(dead_code)]
//! # Key Features
//!
//! - Early lock release after metadata read
//! - Lock hold time monitoring
//! - Configurable optimization (can be disabled for debugging)
//! - Prometheus metrics for lock contention analysis
//!
//! # 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;
#[cfg(feature = "metrics")]
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);
if count > 0 {
Duration::from_micros(total / count)
} else {
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);
#[cfg(feature = "metrics")]
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);
#[cfg(feature = "metrics")]
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)]
mod tests {
use super::*;
use std::sync::Mutex;
#[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());
}
}
+6
View File
@@ -13,17 +13,23 @@
// limitations under the License.
pub mod access;
pub mod backpressure;
pub mod concurrency;
pub mod deadlock_detector;
pub mod ecfs;
pub(crate) mod entity;
pub(crate) mod helper;
pub mod lock_optimizer;
pub mod options;
pub(crate) mod readers;
pub mod rpc;
pub(crate) mod s3_api;
mod sse;
pub mod timeout_wrapper;
pub mod tonic_service;
#[cfg(test)]
mod concurrent_fix_test;
#[cfg(test)]
mod concurrent_get_object_test;
mod ecfs_extend;
+582
View File
@@ -0,0 +1,582 @@
// 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.
//! Request Timeout Wrapper for GetObject operations.
//!
//! This module provides timeout protection for GetObject requests to prevent
//! indefinite hangs caused by deadlocks, resource exhaustion, or slow I/O.
// 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)
//! - Prometheus metrics for timeout monitoring
//!
//! # Usage
//!
//! ```ignore
//! use crate::storage::timeout_wrapper::{RequestTimeoutWrapper, TimeoutConfig};
//!
//! let config = TimeoutConfig::from_env();
//! let wrapper = RequestTimeoutWrapper::new(config);
//!
//! match wrapper.execute_with_timeout(|cancel_token| async move {
//! // Your async operation here
//! Ok(result)
//! }).await {
//! TimedGetObjectResult::Success(result) => { /* handle success */ }
//! TimedGetObjectResult::Timeout(info) => { /* handle timeout */ }
//! TimedGetObjectResult::Error(e) => { /* handle error */ }
//! }
//! ```
use std::time::{Duration, Instant};
use tokio_util::sync::CancellationToken;
use tracing::{debug, warn};
#[cfg(feature = "metrics")]
use metrics::{counter, histogram};
/// Timeout configuration for GetObject requests.
#[derive(Debug, Clone)]
pub struct TimeoutConfig {
/// 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,
}
impl Default for TimeoutConfig {
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),
}
}
}
impl TimeoutConfig {
/// 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,
);
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),
}
}
/// Check if timeout is enabled (timeout > 0).
pub fn is_timeout_enabled(&self) -> bool {
self.get_object_timeout > Duration::ZERO
}
}
/// Information about a timeout event.
#[derive(Debug, Clone)]
pub struct TimeoutInfo {
/// Request ID for correlation.
pub request_id: String,
/// Bucket name.
pub bucket: String,
/// Object key.
pub key: String,
/// Configured timeout duration.
pub timeout_duration: Duration,
/// Actual elapsed time before timeout.
pub elapsed: Duration,
/// Number of bytes transferred before timeout.
pub bytes_transferred: u64,
/// Lock hold time before timeout.
pub lock_hold_time: Option<Duration>,
/// Number of disk reads completed.
pub disk_reads_completed: u32,
/// Number of disk reads pending.
pub disk_reads_pending: u32,
}
/// Result of a timed GetObject operation.
#[derive(Debug)]
pub enum TimedGetObjectResult<T, E> {
/// Operation completed successfully within timeout.
Success(T),
/// Operation timed out and was cancelled.
Timeout(TimeoutInfo),
/// Operation failed with an error (before timeout).
Error(E),
}
/// Request timeout wrapper for async operations.
#[derive(Debug)]
pub struct RequestTimeoutWrapper {
/// Configuration.
config: TimeoutConfig,
/// Request start time.
start_time: Instant,
/// Cancellation token for propagating cancellation to sub-tasks.
cancel_token: CancellationToken,
/// Request ID for logging/metrics.
request_id: String,
}
impl RequestTimeoutWrapper {
/// Create a new timeout wrapper with the given configuration.
pub fn new(config: TimeoutConfig) -> Self {
Self {
config,
start_time: Instant::now(),
cancel_token: CancellationToken::new(),
request_id: format!("req-{}", &uuid::Uuid::new_v4().to_string()[..8]),
}
}
/// 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(),
}
}
/// Get the request ID.
pub fn request_id(&self) -> &str {
&self.request_id
}
/// Get the cancellation token.
/// This can be cloned and passed to sub-tasks for cooperative cancellation.
pub fn cancel_token(&self) -> CancellationToken {
self.cancel_token.clone()
}
/// Check if the operation has been cancelled.
pub fn is_cancelled(&self) -> bool {
self.cancel_token.is_cancelled()
}
/// 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
}
/// Get elapsed time since the request started.
pub fn elapsed(&self) -> Duration {
self.start_time.elapsed()
}
/// Get remaining time before timeout.
/// Returns None if timeout is disabled or already exceeded.
pub fn remaining_time(&self) -> Option<Duration> {
if !self.config.is_timeout_enabled() {
return None;
}
let remaining = self.config.get_object_timeout.saturating_sub(self.elapsed());
if remaining == Duration::ZERO { None } else { Some(remaining) }
}
/// Execute an async operation with timeout protection.
///
/// The operation receives a `CancellationToken` that it can use to:
/// - Check for cancellation: `token.is_cancelled()`
/// - Pass to sub-tasks for propagation
/// - Bind to futures: `future.until_cancelled(token)`
///
/// # Returns
///
/// - `TimedGetObjectResult::Success(T)` if the operation completed within timeout
/// - `TimedGetObjectResult::Timeout(TimeoutInfo)` if the operation timed out
/// - `TimedGetObjectResult::Error(E)` if the operation failed
pub async fn execute_with_timeout<F, Fut, T, E>(self, operation: F) -> TimedGetObjectResult<T, E>
where
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
#[cfg(feature = "metrics")]
counter!("rustfs.get.object.requests.started").increment(1);
// 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();
#[cfg(feature = "metrics")]
{
counter!("rustfs.get.object.requests.completed").increment(1);
histogram!("rustfs.get.object.duration.seconds").record(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();
#[cfg(feature = "metrics")]
{
counter!("rustfs.get.object.requests.failed").increment(1);
histogram!("rustfs.get.object.duration.seconds").record(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();
#[cfg(feature = "metrics")]
{
counter!("rustfs.get.object.timeout.total").increment(1);
histogram!("rustfs.get.object.duration.seconds").record(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,
})
}
}
}
/// Execute an async operation with timeout and context information.
///
/// This is an extended version of `execute_with_timeout` that includes
/// bucket and key information for better timeout logging.
pub async fn execute_with_timeout_and_context<F, Fut, T, E>(
self,
bucket: impl Into<String>,
key: impl Into<String>,
operation: F,
) -> TimedGetObjectResult<T, E>
where
F: FnOnce(CancellationToken) -> Fut,
Fut: std::future::Future<Output = Result<T, E>>,
{
let bucket = bucket.into();
let key = key.into();
if !self.config.is_timeout_enabled() {
debug!(
request_id = %self.request_id,
bucket = %bucket,
key = %key,
"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"
);
#[cfg(feature = "metrics")]
counter!("rustfs.get.object.requests.started").increment(1);
// 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();
#[cfg(feature = "metrics")]
{
counter!("rustfs.get.object.requests.completed").increment(1);
histogram!("rustfs.get.object.duration.seconds").record(elapsed.as_secs_f64());
}
debug!(
request_id = %request_id,
bucket = %bucket,
key = %key,
elapsed_ms = elapsed.as_millis(),
"Operation completed successfully"
);
TimedGetObjectResult::Success(result)
}
Ok(Err(e)) => {
let elapsed = start_time.elapsed();
#[cfg(feature = "metrics")]
{
counter!("rustfs.get.object.requests.failed").increment(1);
histogram!("rustfs.get.object.duration.seconds").record(elapsed.as_secs_f64());
}
debug!(
request_id = %request_id,
bucket = %bucket,
key = %key,
elapsed_ms = elapsed.as_millis(),
"Operation failed with error"
);
TimedGetObjectResult::Error(e)
}
Err(_) => {
let elapsed = start_time.elapsed();
self.cancel_token.cancel();
#[cfg(feature = "metrics")]
{
counter!("rustfs.get.object.timeout.total").increment(1);
histogram!("rustfs.get.object.duration.seconds").record(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"
);
TimedGetObjectResult::Timeout(TimeoutInfo {
request_id,
bucket,
key,
timeout_duration,
elapsed,
bytes_transferred: 0,
lock_hold_time: None,
disk_reads_completed: 0,
disk_reads_pending: 0,
})
}
}
}
}
/// Get the duplex buffer size from environment or default.
pub fn get_duplex_buffer_size() -> usize {
rustfs_utils::get_env_usize(
rustfs_config::ENV_OBJECT_DUPLEX_BUFFER_SIZE,
rustfs_config::DEFAULT_OBJECT_DUPLEX_BUFFER_SIZE,
)
}
/// Get the I/O buffer size from environment or default.
pub fn get_io_buffer_size() -> usize {
rustfs_utils::get_env_usize(rustfs_config::ENV_OBJECT_IO_BUFFER_SIZE, rustfs_config::DEFAULT_OBJECT_IO_BUFFER_SIZE)
}
#[cfg(test)]
mod tests {
use super::*;
use std::time::Duration;
#[test]
fn test_timeout_config_default() {
let config = TimeoutConfig::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();
assert!(config.is_timeout_enabled());
let disabled_config = TimeoutConfig {
get_object_timeout: Duration::ZERO,
..Default::default()
};
assert!(!disabled_config.is_timeout_enabled());
}
#[tokio::test]
async fn test_timeout_wrapper_success() {
let config = TimeoutConfig {
get_object_timeout: Duration::from_secs(5),
..Default::default()
};
let wrapper = RequestTimeoutWrapper::new(config);
let result = wrapper
.execute_with_timeout(|_token| async move { Ok::<i32, String>(42) })
.await;
match result {
TimedGetObjectResult::Success(value) => assert_eq!(value, 42),
_ => panic!("Expected Success result"),
}
}
#[tokio::test]
async fn test_timeout_wrapper_timeout() {
let config = TimeoutConfig {
get_object_timeout: Duration::from_millis(100),
..Default::default()
};
let wrapper = RequestTimeoutWrapper::new(config);
let result = wrapper
.execute_with_timeout(|_token| async move {
tokio::time::sleep(Duration::from_secs(10)).await;
Ok::<i32, String>(42)
})
.await;
match result {
TimedGetObjectResult::Timeout(info) => {
assert!(info.elapsed >= Duration::from_millis(100));
}
_ => panic!("Expected Timeout result"),
}
}
#[tokio::test]
async fn test_timeout_wrapper_error() {
let config = TimeoutConfig {
get_object_timeout: Duration::from_secs(5),
..Default::default()
};
let wrapper = RequestTimeoutWrapper::new(config);
let result = wrapper
.execute_with_timeout(|_token| async move { Err::<i32, String>("test error".to_string()) })
.await;
match result {
TimedGetObjectResult::Error(e) => assert_eq!(e, "test error"),
_ => panic!("Expected Error result"),
}
}
#[tokio::test]
async fn test_timeout_wrapper_disabled() {
let config = TimeoutConfig {
get_object_timeout: Duration::ZERO,
..Default::default()
};
let wrapper = RequestTimeoutWrapper::new(config);
// This would timeout if timeout was enabled
let result = wrapper
.execute_with_timeout(|_token| async move {
tokio::time::sleep(Duration::from_millis(200)).await;
Ok::<i32, String>(42)
})
.await;
match result {
TimedGetObjectResult::Success(value) => assert_eq!(value, 42),
_ => panic!("Expected Success result when timeout is disabled"),
}
}
#[test]
fn test_get_duplex_buffer_size() {
// Should return default (4MB) when env var not set
let size = get_duplex_buffer_size();
assert_eq!(size, 4 * 1024 * 1024);
}
#[test]
fn test_get_io_buffer_size() {
// Should return default (128KB) when env var not set
let size = get_io_buffer_size();
assert_eq!(size, 128 * 1024);
}
}
Executable → Regular
+91 -7
View File
@@ -53,7 +53,7 @@ export RUSTFS_CONSOLE_ADDRESS=":9001"
# export RUSTFS_TLS_PATH="./deploy/certs"
# Observability related configuration
export RUSTFS_OBS_ENDPOINT=http://localhost:4318 # OpenTelemetry Collector address
#export RUSTFS_OBS_ENDPOINT=http://localhost:4318 # OpenTelemetry Collector address
# RustFS OR OTEL exporter configuration
#export RUSTFS_OBS_TRACE_ENDPOINT=http://localhost:4318/v1/traces # OpenTelemetry Collector trace address http://localhost:4318/v1/traces
#export OTEL_EXPORTER_OTLP_TRACES_ENDPOINT=http://localhost:14318/v1/traces
@@ -215,14 +215,98 @@ export RUSTFS_TRUST_SYSTEM_CA=true
# export RUSTFS_FTPS_ADDRESS="0.0.0.0:8022"
# export RUSTFS_FTPS_CERTS_DIR="${current_dir}/deploy/certs/ftps"
# Use default timeout (60 seconds)
# No environment variable needed
# ============================================
# Concurrent Request Optimization Configuration
# ============================================
# These settings optimize GetObject performance under high concurrency.
# Most features are enabled by default with sensible values.
# Uncomment and adjust based on your scenario.
# Increase timeout for high-latency network storage
#export RUSTFS_LOCK_ACQUIRE_TIMEOUT=120
# --- Default Configuration (Recommended for most cases) ---
# Request timeout: 30 seconds (prevents indefinite hangs)
export RUSTFS_OBJECT_GET_TIMEOUT=30
# Disk read timeout: 10 seconds
export RUSTFS_OBJECT_DISK_READ_TIMEOUT=10
# Lock acquire timeout: 5 seconds
export RUSTFS_OBJECT_LOCK_ACQUIRE_TIMEOUT=5
# Duplex buffer size: 4MB (4x larger than original 1MB)
export RUSTFS_OBJECT_DUPLEX_BUFFER_SIZE=4194304
# I/O buffer size: 128KB
export RUSTFS_OBJECT_IO_BUFFER_SIZE=131072
# Max concurrent disk reads: 64
export RUSTFS_OBJECT_MAX_CONCURRENT_DISK_READS=64
# Lock optimization: release read lock after metadata read
export RUSTFS_OBJECT_LOCK_OPTIMIZATION_ENABLE=true
# Priority scheduling: small requests get higher priority
export RUSTFS_OBJECT_PRIORITY_SCHEDULING_ENABLE=true
# Deadlock detection: disabled by default (has performance overhead)
# export RUSTFS_OBJECT_DEADLOCK_DETECTION_ENABLE=false
# Reduce timeout for low-latency local storage
export RUSTFS_LOCK_ACQUIRE_TIMEOUT=30
# --- Scenario 1: Home NAS / Small Storage Server ---
# Hardware: 4-8 cores, 8-16GB RAM, 1-4 HDDs, 1Gbps network
# Typical concurrency: 5-20 requests
# export RUSTFS_OBJECT_GET_TIMEOUT=30
# export RUSTFS_OBJECT_DUPLEX_BUFFER_SIZE=4194304
# export RUSTFS_OBJECT_MAX_CONCURRENT_DISK_READS=64
# export RUSTFS_OBJECT_LOCK_OPTIMIZATION_ENABLE=true
# export RUSTFS_OBJECT_PRIORITY_SCHEDULING_ENABLE=true
# --- Scenario 2: Medium Enterprise Storage ---
# Hardware: 8-16 cores, 32-64GB RAM, 4-12 HDDs/SSDs, 10Gbps network
# Typical concurrency: 20-100 requests
# export RUSTFS_OBJECT_GET_TIMEOUT=60
# export RUSTFS_OBJECT_DUPLEX_BUFFER_SIZE=8388608
# export RUSTFS_OBJECT_IO_BUFFER_SIZE=262144
# export RUSTFS_OBJECT_MAX_CONCURRENT_DISK_READS=128
# export RUSTFS_OBJECT_HIGH_CONCURRENCY_THRESHOLD=16
# export RUSTFS_OBJECT_LOCK_OPTIMIZATION_ENABLE=true
# export RUSTFS_OBJECT_PRIORITY_SCHEDULING_ENABLE=true
# --- Scenario 3: Large Enterprise / Cloud Storage ---
# Hardware: 32+ cores, 128+GB RAM, NVMe SSD array, 25-100Gbps network
# Typical concurrency: 100-1000 requests
# export RUSTFS_OBJECT_GET_TIMEOUT=120
# export RUSTFS_OBJECT_DUPLEX_BUFFER_SIZE=16777216
# export RUSTFS_OBJECT_IO_BUFFER_SIZE=524288
# export RUSTFS_OBJECT_MAX_CONCURRENT_DISK_READS=256
# export RUSTFS_OBJECT_HIGH_CONCURRENCY_THRESHOLD=32
# export RUSTFS_OBJECT_LOCK_OPTIMIZATION_ENABLE=true
# export RUSTFS_OBJECT_PRIORITY_SCHEDULING_ENABLE=true
# --- Scenario 4: Kopia Backup Optimized ---
# Problem: Kopia sends 20-30 concurrent range requests for 20-26MB objects
# Solution: Larger buffer, lock optimization, priority scheduling
# export RUSTFS_OBJECT_GET_TIMEOUT=45
# export RUSTFS_OBJECT_DUPLEX_BUFFER_SIZE=8388608
# export RUSTFS_OBJECT_MAX_CONCURRENT_DISK_READS=64
# export RUSTFS_OBJECT_LOCK_OPTIMIZATION_ENABLE=true
# export RUSTFS_OBJECT_PRIORITY_SCHEDULING_ENABLE=true
# export RUSTFS_OBJECT_DEADLOCK_DETECTION_ENABLE=true # Enable for debugging
# --- Scenario 5: Low Power / Embedded Storage ---
# Hardware: 2-4 cores (ARM/x86), 2-4GB RAM, SD card/eMMC, 100Mbps-1Gbps
# Typical concurrency: 1-5 requests
# export RUSTFS_OBJECT_GET_TIMEOUT=60
# export RUSTFS_OBJECT_DISK_READ_TIMEOUT=20
# export RUSTFS_OBJECT_DUPLEX_BUFFER_SIZE=2097152
# export RUSTFS_OBJECT_IO_BUFFER_SIZE=65536
# export RUSTFS_OBJECT_MAX_CONCURRENT_DISK_READS=16
# export RUSTFS_OBJECT_HIGH_CONCURRENCY_THRESHOLD=4
# export RUSTFS_OBJECT_LOCK_OPTIMIZATION_ENABLE=true
# export RUSTFS_OBJECT_PRIORITY_SCHEDULING_ENABLE=true
# --- Scenario 6: Debugging / Troubleshooting ---
# Enable all diagnostic features
# export RUSTFS_OBJECT_GET_TIMEOUT=15
# export RUSTFS_OBJECT_DEADLOCK_DETECTION_ENABLE=true
# export RUSTFS_OBJECT_DEADLOCK_CHECK_INTERVAL=3
# export RUSTFS_OBJECT_DEADLOCK_HANG_THRESHOLD=5
# --- Backpressure Configuration ---
# High watermark: trigger backpressure when buffer usage exceeds this percentage
export RUSTFS_BACKPRESSURE_HIGH_WATERMARK=80
# Low watermark: release backpressure when buffer usage drops below this percentage
export RUSTFS_BACKPRESSURE_LOW_WATERMARK=50
if [ -n "$1" ]; then
export RUSTFS_VOLUMES="$1"