diff --git a/ARCHITECTURE.md b/ARCHITECTURE.md index f22d1c018..e4362b727 100644 --- a/ARCHITECTURE.md +++ b/ARCHITECTURE.md @@ -141,7 +141,7 @@ Depth 8 — TOP: | `io-metrics` | 4.5K | I/O operation metrics and counters | | `rio` | 6.9K | Composable reader chain (encrypt → compress → hash → limit) | | `object-io` | 2.4K | High-level object read/write using rio + ecstore | -| `concurrency` | 1.8K | Concurrency control wrappers over io-core | +| `concurrency` | 0.8K | Shared concurrency contract types: workload admission snapshots, worker-slot pool, policy types (runtime control lives in `rustfs/src/storage`) | **Storage Engine:** diff --git a/Cargo.lock b/Cargo.lock index 65d82ea23..736d5adb6 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -9027,12 +9027,9 @@ version = "1.0.0-beta.8" dependencies = [ "insta", "rustfs-io-core", - "rustfs-io-metrics", "serde", "serde_json", - "thiserror 2.0.18", "tokio", - "tokio-util", "tracing", ] diff --git a/crates/concurrency/Cargo.toml b/crates/concurrency/Cargo.toml index 0b4e329c8..59e75776f 100644 --- a/crates/concurrency/Cargo.toml +++ b/crates/concurrency/Cargo.toml @@ -6,22 +6,17 @@ license.workspace = true repository.workspace = true rust-version.workspace = true homepage.workspace = true -description = "Concurrency management for RustFS - timeout, locking, backpressure, and I/O scheduling" -keywords = ["rustfs", "concurrency", "timeout", "backpressure", "scheduling"] +description = "Shared concurrency contract types for RustFS - workload admission, queue snapshots, worker pools, and policy types" +keywords = ["rustfs", "concurrency", "admission", "backpressure", "workers"] categories = ["concurrency", "filesystem"] [dependencies] # Internal crates rustfs-io-core = { workspace = true } -rustfs-io-metrics = { workspace = true } serde = { workspace = true } # Async runtime -tokio = { workspace = true, features = ["sync", "time", "rt"] } -tokio-util = { workspace = true } - -# Error handling -thiserror = { workspace = true } +tokio = { workspace = true, features = ["sync"] } # Logging tracing = { workspace = true } @@ -29,18 +24,4 @@ tracing = { workspace = true } [dev-dependencies] insta = { workspace = true } serde_json = { workspace = true } -tokio = { workspace = true, features = ["test-util","macros","rt-multi-thread"] } - -[features] -default = ["timeout", "lock", "deadlock", "backpressure", "scheduler"] - -# Feature modules -timeout = [] -lock = [] -deadlock = [] -backpressure = [] -scheduler = [] - -[package.metadata.docs.rs] -all-features = true -rustdoc-args = ["--cfg", "docsrs"] +tokio = { workspace = true, features = ["test-util", "macros", "rt-multi-thread"] } diff --git a/crates/concurrency/src/backpressure.rs b/crates/concurrency/src/backpressure.rs index f846b64d4..502d4fb0f 100644 --- a/crates/concurrency/src/backpressure.rs +++ b/crates/concurrency/src/backpressure.rs @@ -12,17 +12,15 @@ // See the License for the specific language governing permissions and // limitations under the License. -//! Backpressure management +//! Shared backpressure policy type. +//! +//! The runtime backpressure implementation (byte-watermark pipes and +//! monitors) lives in `rustfs/src/storage/backpressure.rs`; this module only +//! carries the watermark policy type that implementation shares. -use rustfs_io_core::{ - BackpressureConfig as CoreBackpressureConfig, BackpressureMonitor as CoreBackpressureMonitor, BackpressureState, -}; -use rustfs_io_metrics::backpressure_metrics; -use std::sync::Arc; -use std::time::Instant; -use tokio::io::{DuplexStream, duplex}; +use rustfs_io_core::BackpressureConfig as CoreBackpressureConfig; -/// Facade policy for duplex-pipe watermark backpressure. +/// Watermark policy for duplex-pipe backpressure. #[derive(Debug, Clone, Copy)] pub struct PipeBackpressurePolicy { /// Buffer size in bytes @@ -54,9 +52,9 @@ impl PipeBackpressurePolicy { (self.buffer_size as u64 * self.low_watermark as u64 / 100) as usize } - /// Convert the facade policy into the reusable io-core admission-pressure config. + /// Convert the policy into the reusable io-core admission-pressure config. /// - /// The concurrency layer still owns duplex buffer sizing, but the shared + /// The caller still owns duplex buffer sizing, but the shared /// overload/admission primitive lives in `io-core`. pub fn to_core_config(&self) -> CoreBackpressureConfig { CoreBackpressureConfig { @@ -69,157 +67,28 @@ impl PipeBackpressurePolicy { } } -/// Backpressure manager -pub struct BackpressureManager { - config: PipeBackpressurePolicy, - core_config: CoreBackpressureConfig, - monitor: Arc, -} - -impl BackpressureManager { - /// Create a new backpressure manager - pub fn new(buffer_size: usize, high_watermark: u32, low_watermark: u32) -> Self { - Self::from_policy(PipeBackpressurePolicy { - buffer_size, - high_watermark, - low_watermark, - }) - } - - /// Create a new backpressure manager from the facade policy type. - pub fn from_policy(config: PipeBackpressurePolicy) -> Self { - let core_config = config.to_core_config(); - Self { - config, - core_config: core_config.clone(), - monitor: Arc::new(CoreBackpressureMonitor::new(core_config)), - } - } - - /// Get the configuration - pub fn config(&self) -> &PipeBackpressurePolicy { - &self.config - } - - /// Get the derived io-core admission-pressure configuration. - pub fn core_config(&self) -> &CoreBackpressureConfig { - &self.core_config - } - - /// Get the monitor - pub fn monitor(&self) -> Arc { - self.monitor.clone() - } - - /// Create a backpressure pipe - pub fn create_pipe(&self) -> BackpressurePipe { - BackpressurePipe::new(self.config, self.monitor.clone()) - } - - /// Get current state - pub fn state(&self) -> BackpressureState { - self.monitor.state() - } - - /// Check if backpressure is active - pub fn is_active(&self) -> bool { - self.monitor.is_active() - } -} - -/// Backpressure pipe wrapping tokio's duplex -pub struct BackpressurePipe { - reader: DuplexStream, - writer: DuplexStream, - config: PipeBackpressurePolicy, - monitor: Arc, - created_at: Instant, -} - -/// Shared pipe metadata snapshot for facade-level backpressure pipes. -#[derive(Debug, Clone, Copy, PartialEq, Eq)] -pub struct BackpressurePipeMeta { - /// Configured duplex buffer capacity in bytes. - pub buffer_capacity: usize, - /// Current backpressure state reported by the shared core monitor. - pub state: BackpressureState, - /// Age of the pipe since creation. - pub age: std::time::Duration, -} - -impl BackpressurePipe { - fn new(config: PipeBackpressurePolicy, monitor: Arc) -> Self { - let (reader, writer) = duplex(config.buffer_size); - - Self { - reader, - writer, - config, - monitor, - created_at: Instant::now(), - } - } - - /// Get the reader end - pub fn reader(&mut self) -> &mut DuplexStream { - &mut self.reader - } - - /// Get the writer end - pub fn writer(&mut self) -> &mut DuplexStream { - &mut self.writer - } - - /// Split into reader and writer - pub fn into_split(self) -> (DuplexStream, DuplexStream) { - (self.reader, self.writer) - } - - /// Get the configuration - pub fn config(&self) -> &PipeBackpressurePolicy { - &self.config - } - - /// Get current state - pub fn state(&self) -> BackpressureState { - self.monitor.state() - } - - /// Get the age of this pipe - pub fn age(&self) -> std::time::Duration { - self.created_at.elapsed() - } - - /// Get a compact metadata snapshot for the pipe. - pub fn meta(&self) -> BackpressurePipeMeta { - BackpressurePipeMeta { - buffer_capacity: self.config.buffer_size, - state: self.state(), - age: self.age(), - } - } - - /// Check if should apply backpressure - pub fn should_apply_backpressure(&self) -> bool { - let should = self.monitor.should_apply_backpressure(); - if should { - backpressure_metrics::record_backpressure_activation(); - } - should - } -} - #[cfg(test)] mod tests { use super::*; #[test] - fn test_backpressure_config() { + fn test_backpressure_policy_defaults() { let config = PipeBackpressurePolicy::default(); assert_eq!(config.buffer_size, 4 * 1024 * 1024); assert!(config.high_watermark > config.low_watermark); } + #[test] + fn test_backpressure_policy_watermark_bytes() { + let config = PipeBackpressurePolicy { + buffer_size: 1000, + high_watermark: 80, + low_watermark: 50, + }; + assert_eq!(config.high_watermark_bytes(), 800); + assert_eq!(config.low_watermark_bytes(), 500); + } + #[test] fn test_backpressure_policy_to_core_config() { let policy = PipeBackpressurePolicy::default(); @@ -228,33 +97,4 @@ mod tests { assert_eq!(core.low_water_mark, policy.low_watermark as f64 / 100.0); assert!(core.enabled); } - - #[test] - fn test_backpressure_manager() { - let manager = BackpressureManager::new(1024, 80, 50); - assert_eq!(manager.state(), BackpressureState::Normal); - } - - #[test] - fn test_backpressure_pipe() { - let manager = BackpressureManager::new(1024, 80, 50); - let pipe = manager.create_pipe(); - assert_eq!(pipe.state(), BackpressureState::Normal); - assert_eq!(pipe.meta().buffer_capacity, 1024); - } - - #[test] - fn test_backpressure_pipe_meta_is_read_only() { - let manager = BackpressureManager::new(2048, 80, 50); - let pipe = manager.create_pipe(); - let first = pipe.meta(); - let second = pipe.meta(); - - assert_eq!(first.buffer_capacity, 2048); - assert_eq!(first.state, BackpressureState::Normal); - assert_eq!(second.buffer_capacity, first.buffer_capacity); - assert_eq!(second.state, first.state); - assert!(second.age >= first.age); - assert_eq!(manager.state(), BackpressureState::Normal); - } } diff --git a/crates/concurrency/src/config.rs b/crates/concurrency/src/config.rs deleted file mode 100644 index 8f010c104..000000000 --- a/crates/concurrency/src/config.rs +++ /dev/null @@ -1,325 +0,0 @@ -// Copyright 2024 RustFS Team -// -// Licensed under the Apache License, Version 2.0 (the "License"); -// you may not use this file except in compliance with the License. -// You may obtain a copy of the License at -// -// http://www.apache.org/licenses/LICENSE-2.0 -// -// Unless required by applicable law or agreed to in writing, software -// distributed under the License is distributed on an "AS IS" BASIS, -// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -// See the License for the specific language governing permissions and -// limitations under the License. - -//! Configuration for concurrency management - -use crate::{ - backpressure::PipeBackpressurePolicy, deadlock::DeadlockMonitorPolicy, scheduler::SchedulerPolicy, - timeout::TimeoutManagerPolicy, -}; -use std::time::Duration; - -/// Feature flags for concurrency modules -#[derive(Debug, Clone, Copy, PartialEq, Eq)] -pub struct ConcurrencyFeatures { - /// Enable timeout control - pub timeout: bool, - /// Enable lock optimization - pub lock: bool, - /// Enable deadlock detection - pub deadlock: bool, - /// Enable backpressure management - pub backpressure: bool, - /// Enable I/O scheduling - pub scheduler: bool, -} - -impl Default for ConcurrencyFeatures { - fn default() -> Self { - Self { - timeout: cfg!(feature = "timeout"), - lock: cfg!(feature = "lock"), - deadlock: cfg!(feature = "deadlock"), - backpressure: cfg!(feature = "backpressure"), - scheduler: cfg!(feature = "scheduler"), - } - } -} - -impl ConcurrencyFeatures { - /// Create with all features enabled - pub fn all() -> Self { - Self { - timeout: true, - lock: true, - deadlock: true, - backpressure: true, - scheduler: true, - } - } - - /// Create with no features enabled - pub fn none() -> Self { - Self { - timeout: false, - lock: false, - deadlock: false, - backpressure: false, - scheduler: false, - } - } - - /// Check if any feature is enabled - pub fn any_enabled(&self) -> bool { - self.timeout || self.lock || self.deadlock || self.backpressure || self.scheduler - } -} - -/// Facade policy for lock manager behavior. -#[derive(Debug, Clone, Copy)] -pub struct LockManagerPolicy { - /// Enable lock optimization. - pub enabled: bool, - /// Lock acquisition timeout. - pub acquire_timeout: Duration, -} - -impl Default for LockManagerPolicy { - fn default() -> Self { - Self { - enabled: true, - acquire_timeout: Duration::from_secs(5), - } - } -} - -/// Main configuration for concurrency management -#[derive(Debug, Clone, Default)] -pub struct ConcurrencyConfig { - /// Feature flags - pub features: ConcurrencyFeatures, - /// Timeout facade policy. - pub timeout_policy: TimeoutManagerPolicy, - /// Lock facade policy. - pub lock_policy: LockManagerPolicy, - /// Deadlock facade policy. - pub deadlock_policy: DeadlockMonitorPolicy, - /// Backpressure facade policy. - pub backpressure_policy: PipeBackpressurePolicy, - /// Scheduler facade policy. - pub scheduler_policy: SchedulerPolicy, -} - -impl ConcurrencyConfig { - /// Create configuration from environment variables - pub fn from_env() -> Self { - let mut config = Self::default(); - - // Read from environment if available - if let Some(secs) = parse_env::("RUSTFS_TIMEOUT_DEFAULT") { - config.timeout_policy.default_timeout = Duration::from_secs(secs); - } - - if let Some(secs) = parse_env::("RUSTFS_TIMEOUT_MAX") { - config.timeout_policy.max_timeout = Duration::from_secs(secs); - } - - if let Some(size) = parse_env::("RUSTFS_BACKPRESSURE_BUFFER_SIZE") { - config.backpressure_policy.buffer_size = size; - } - - if let Some(size) = parse_env::("RUSTFS_IO_BUFFER_SIZE") { - config.scheduler_policy.base_buffer_size = size; - } - - config - } - - /// Validate configuration - pub fn validate(&self) -> Result<(), ConfigError> { - if self.timeout_policy.default_timeout.is_zero() || self.timeout_policy.max_timeout.is_zero() { - return Err(ConfigError::InvalidTimeout("timeouts must be > 0".to_string())); - } - if self.timeout_policy.default_timeout > self.timeout_policy.max_timeout { - return Err(ConfigError::InvalidTimeout("default_timeout cannot exceed max_timeout".to_string())); - } - if self.timeout_policy.min_timeout > self.timeout_policy.max_timeout { - return Err(ConfigError::InvalidTimeout("min_timeout cannot exceed max_timeout".to_string())); - } - - // A zero-capacity pipe (duplex(0)) never accepts writes, hanging producers forever. - if self.backpressure_policy.buffer_size == 0 { - return Err(ConfigError::InvalidBackpressure("buffer_size must be > 0".to_string())); - } - if self.backpressure_policy.high_watermark <= self.backpressure_policy.low_watermark - || self.backpressure_policy.high_watermark > 100 - { - return Err(ConfigError::InvalidBackpressure( - "high_watermark must be > low_watermark and <= 100".to_string(), - )); - } - - if self.scheduler_policy.base_buffer_size == 0 { - return Err(ConfigError::InvalidScheduler("base_buffer_size must be > 0".to_string())); - } - if self.scheduler_policy.base_buffer_size > self.scheduler_policy.max_buffer_size { - return Err(ConfigError::InvalidScheduler( - "base_buffer_size cannot exceed max_buffer_size".to_string(), - )); - } - // Ensure the derived io-core config also holds its invariants. - self.scheduler_policy - .to_core_config() - .validate() - .map_err(|e| ConfigError::InvalidScheduler(e.to_string()))?; - - Ok(()) - } -} - -/// Parse an environment variable, logging a warning instead of silently -/// falling back to the default when the value is set but unparsable. -fn parse_env(name: &str) -> Option { - let val = std::env::var(name).ok()?; - match val.parse() { - Ok(parsed) => Some(parsed), - Err(_) => { - tracing::warn!(env_var = name, value = %val, "ignoring unparsable environment variable"); - None - } - } -} - -/// Configuration error -#[allow(clippy::enum_variant_names)] -#[derive(Debug, Clone, thiserror::Error)] -pub enum ConfigError { - /// Invalid timeout configuration - #[error("Invalid timeout config: {0}")] - InvalidTimeout(String), - - /// Invalid backpressure configuration - #[error("Invalid backpressure config: {0}")] - InvalidBackpressure(String), - - /// Invalid scheduler configuration - #[error("Invalid scheduler config: {0}")] - InvalidScheduler(String), -} - -#[cfg(test)] -mod tests { - use super::*; - - #[test] - fn test_default_config() { - let config = ConcurrencyConfig::default(); - assert!(config.validate().is_ok()); - } - - #[test] - fn test_invalid_timeout() { - let config = ConcurrencyConfig { - timeout_policy: TimeoutManagerPolicy { - default_timeout: Duration::from_secs(100), - max_timeout: Duration::from_secs(50), - enable_dynamic: true, - ..Default::default() - }, - ..Default::default() - }; - assert!( - config.validate().is_err(), - "validate() should return an error when default_timeout > max_timeout" - ); - } - - #[test] - fn test_invalid_min_timeout() { - let config = ConcurrencyConfig { - timeout_policy: TimeoutManagerPolicy { - min_timeout: Duration::from_secs(100), - max_timeout: Duration::from_secs(50), - ..Default::default() - }, - ..Default::default() - }; - assert!( - config.validate().is_err(), - "validate() should return an error when min_timeout > max_timeout" - ); - } - - #[test] - fn test_zero_backpressure_buffer_size_rejected() { - let config = ConcurrencyConfig { - backpressure_policy: PipeBackpressurePolicy { - buffer_size: 0, - ..Default::default() - }, - ..Default::default() - }; - assert!( - matches!(config.validate(), Err(ConfigError::InvalidBackpressure(_))), - "validate() must reject buffer_size=0 (duplex(0) hangs all writes)" - ); - } - - #[test] - fn test_zero_timeouts_rejected() { - let config = ConcurrencyConfig { - timeout_policy: TimeoutManagerPolicy { - default_timeout: Duration::ZERO, - ..Default::default() - }, - ..Default::default() - }; - assert!(matches!(config.validate(), Err(ConfigError::InvalidTimeout(_)))); - - let config = ConcurrencyConfig { - timeout_policy: TimeoutManagerPolicy { - max_timeout: Duration::ZERO, - ..Default::default() - }, - ..Default::default() - }; - assert!(matches!(config.validate(), Err(ConfigError::InvalidTimeout(_)))); - } - - #[test] - fn test_zero_scheduler_base_buffer_size_rejected() { - let config = ConcurrencyConfig { - scheduler_policy: SchedulerPolicy { - base_buffer_size: 0, - ..Default::default() - }, - ..Default::default() - }; - assert!(matches!(config.validate(), Err(ConfigError::InvalidScheduler(_)))); - } - - #[test] - fn test_small_max_buffer_size_passes_core_validation() { - // max_buffer_size below the io-core default min_buffer_size (4KB) must - // still yield a valid core config (min lowered by to_core_config). - let config = ConcurrencyConfig { - scheduler_policy: SchedulerPolicy { - base_buffer_size: 1024, - max_buffer_size: 2048, - ..Default::default() - }, - ..Default::default() - }; - assert!(config.validate().is_ok()); - } - - #[test] - fn test_features() { - let features = ConcurrencyFeatures::all(); - assert!(features.any_enabled()); - - let features = ConcurrencyFeatures::none(); - assert!(!features.any_enabled()); - } -} diff --git a/crates/concurrency/src/deadlock.rs b/crates/concurrency/src/deadlock.rs index f55b4990e..539b1250e 100644 --- a/crates/concurrency/src/deadlock.rs +++ b/crates/concurrency/src/deadlock.rs @@ -12,15 +12,16 @@ // See the License for the specific language governing permissions and // limitations under the License. -//! Deadlock detection management +//! Shared deadlock-monitor policy type. +//! +//! The runtime request-hang / deadlock detection loop lives in +//! `rustfs/src/storage/deadlock_detector.rs`; this module only carries the +//! monitor policy type that implementation shares. -use rustfs_io_core::{DeadlockDetector as CoreDeadlockDetector, DeadlockDetectorConfig as CoreDeadlockConfig, LockType}; -use rustfs_io_metrics::deadlock_metrics; -use std::collections::HashMap; -use std::sync::Arc; -use std::time::{Duration, Instant}; +use rustfs_io_core::DeadlockDetectorConfig as CoreDeadlockConfig; +use std::time::Duration; -/// Facade policy for the concurrency-layer deadlock monitor. +/// Policy for the request-hang deadlock monitor. #[derive(Debug, Clone, Copy)] pub struct DeadlockMonitorPolicy { /// Enable deadlock detection @@ -42,7 +43,7 @@ impl Default for DeadlockMonitorPolicy { } impl DeadlockMonitorPolicy { - /// Convert the facade policy into the reusable io-core deadlock config. + /// Convert the policy into the reusable io-core deadlock config. pub fn to_core_config(&self) -> CoreDeadlockConfig { CoreDeadlockConfig { enabled: self.enabled, @@ -52,182 +53,14 @@ impl DeadlockMonitorPolicy { } } -/// Deadlock manager -pub struct DeadlockManager { - config: DeadlockMonitorPolicy, - detector: Arc, - running: Arc>, -} - -impl DeadlockManager { - /// Create a new deadlock manager - pub fn new(enabled: bool, check_interval: Duration, hang_threshold: Duration) -> Self { - Self::from_policy(DeadlockMonitorPolicy { - enabled, - check_interval, - hang_threshold, - }) - } - - /// Create a new deadlock manager from the facade policy type. - pub fn from_policy(config: DeadlockMonitorPolicy) -> Self { - let core_config = config.to_core_config(); - Self { - config, - detector: Arc::new(CoreDeadlockDetector::new(core_config)), - running: Arc::new(tokio::sync::Mutex::new(false)), - } - } - - /// Get the configuration - pub fn config(&self) -> &DeadlockMonitorPolicy { - &self.config - } - - /// Get the core detector - pub fn detector(&self) -> Arc { - self.detector.clone() - } - - /// Start the deadlock detection background task - pub async fn start(&self) { - if !self.config.enabled { - return; - } - - let mut running = self.running.lock().await; - if *running { - return; - } - *running = true; - drop(running); - - tracing::info!( - event = "deadlock_monitor.lifecycle", - component = "concurrency", - subsystem = "deadlock", - state = "started", - check_interval_ms = self.config.check_interval.as_millis(), - hang_threshold_ms = self.config.hang_threshold.as_millis(), - "deadlock monitor state changed" - ); - } - - /// Stop the deadlock detection - pub async fn stop(&self) { - let mut running = self.running.lock().await; - *running = false; - - tracing::info!( - event = "deadlock_monitor.lifecycle", - component = "concurrency", - subsystem = "deadlock", - state = "stopped", - check_interval_ms = self.config.check_interval.as_millis(), - hang_threshold_ms = self.config.hang_threshold.as_millis(), - "deadlock monitor state changed" - ); - } - - /// Create a request tracker - pub fn track_request(&self, request_id: String, description: String) -> RequestTracker { - RequestTracker::new(request_id, description, self.detector.clone()) - } - - /// Register a lock - pub fn register_lock(&self, lock_type: LockType) -> u64 { - self.detector.register_lock(lock_type) - } - - /// Unregister a lock - pub fn unregister_lock(&self, lock_id: u64) { - self.detector.unregister_lock(lock_id); - } - - /// Detect deadlock - pub fn detect_deadlock(&self) -> Option> { - let result = self.detector.detect_deadlock(); - if let Some(ref cycle) = result { - deadlock_metrics::record_deadlock_detected(cycle.len()); - } - result - } -} - -/// Lightweight compatibility wrapper for request-scoped deadlock bookkeeping. -/// -/// This type intentionally stays minimal in the concurrency layer. Rich -/// request-level lock/resource diagnostics belong to -/// `rustfs::storage::deadlock_detector::RequestResourceTracker`. -pub struct RequestTracker { - request_id: String, - description: String, - start_time: Instant, - resources: HashMap>, - detector: Arc, -} - -impl RequestTracker { - fn new(request_id: String, description: String, detector: Arc) -> Self { - let start_time = Instant::now(); - detector.register_request(&request_id, 1); // Use placeholder thread ID - - Self { - request_id, - description, - start_time, - resources: HashMap::new(), - detector, - } - } - - /// Get the request ID - pub fn request_id(&self) -> &str { - &self.request_id - } - - /// Get the description - pub fn description(&self) -> &str { - &self.description - } - - /// Get the elapsed time - pub fn elapsed(&self) -> Duration { - self.start_time.elapsed() - } - - /// Record a lock acquisition - pub fn record_lock_acquire(&mut self, lock_id: u64, resource: String) { - self.resources.entry("locks".to_string()).or_default().push(resource); - self.detector.record_acquire(lock_id, 1); // Use placeholder thread ID - deadlock_metrics::record_lock_acquisition("read"); - } - - /// Return a read-only view of tracked resource names. - pub fn resources(&self) -> &HashMap> { - &self.resources - } - - /// Record a lock release - pub fn record_lock_release(&mut self, lock_id: u64) { - self.detector.record_release(lock_id); - } -} - -impl Drop for RequestTracker { - fn drop(&mut self) { - self.detector.unregister_request(&self.request_id); - } -} - #[cfg(test)] mod tests { use super::*; #[test] - fn test_deadlock_manager_creation() { - let manager = DeadlockManager::new(false, Duration::from_secs(10), Duration::from_secs(60)); - assert!(!manager.config().enabled); + fn test_deadlock_policy_defaults_disabled() { + let policy = DeadlockMonitorPolicy::default(); + assert!(!policy.enabled); } #[test] @@ -238,16 +71,4 @@ mod tests { assert_eq!(core.detection_interval, policy.check_interval); assert_eq!(core.max_hold_time, policy.hang_threshold); } - - #[tokio::test] - async fn test_request_tracker() { - let manager = DeadlockManager::new(true, Duration::from_secs(10), Duration::from_secs(60)); - let mut tracker = manager.track_request("req-1".to_string(), "test request".to_string()); - let lock_id = manager.register_lock(LockType::Mutex); - tracker.record_lock_acquire(lock_id, "bucket/key".to_string()); - - assert_eq!(tracker.request_id(), "req-1"); - assert_eq!(tracker.description(), "test request"); - assert_eq!(tracker.resources().get("locks").map(Vec::len), Some(1)); - } } diff --git a/crates/concurrency/src/lib.rs b/crates/concurrency/src/lib.rs index e4bca7a98..7e6ba9706 100644 --- a/crates/concurrency/src/lib.rs +++ b/crates/concurrency/src/lib.rs @@ -12,167 +12,40 @@ // See the License for the specific language governing permissions and // limitations under the License. -//! # RustFS Concurrency Management +//! # RustFS Concurrency Contracts //! -//! This crate provides comprehensive concurrency management for RustFS, -//! including timeout control, lock optimization, deadlock detection, -//! backpressure management, and I/O scheduling. +//! Shared concurrency contract types for RustFS: //! -//! ## Features +//! - [`workload`]: workload admission snapshot/contract types consumed by +//! ecstore, heal, and the server for admission-aware throttling. +//! - [`GetObjectQueueSnapshot`]: disk permit queue usage snapshot for +//! GetObject orchestration. +//! - [`workers`]: a bounded semaphore-backed worker-slot pool. +//! - [`PipeBackpressurePolicy`] / [`DeadlockMonitorPolicy`]: policy types +//! shared with the runtime implementations. //! -//! All features are controlled by feature flags and can be enabled/disabled at compile time: -//! -//! - **timeout**: Dynamic timeout calculation based on data size and transfer rate -//! - **lock**: Early lock release to reduce contention -//! - **deadlock**: Request tracking and cycle detection -//! - **backpressure**: Buffer-based flow control -//! - **scheduler**: Adaptive buffer sizing and priority queuing -//! -//! ## Architecture -//! -//! ```text -//! rustfs-concurrency (Business Layer) -//! ├── timeout (Timeout Control) -//! ├── lock (Lock Optimization) -//! ├── deadlock (Deadlock Detection) -//! ├── backpressure (Backpressure Management) -//! └── scheduler (I/O Scheduling) -//! │ -//! ├── rustfs-io-core (Core Algorithms) -//! └── rustfs-io-metrics (Metrics Collection) -//! ``` -//! -//! ## Usage -//! -//! ```rust,no_run -//! use rustfs_concurrency::{ConcurrencyConfig, ConcurrencyManager}; -//! -//! # #[tokio::main] -//! # async fn main() { -//! // Create manager with all features enabled -//! let config = ConcurrencyConfig::default(); -//! let manager = ConcurrencyManager::new(config); -//! -//! // Start services -//! manager.start().await; -//! -//! // Use timeout control (if enabled) -//! if manager.is_timeout_enabled() { -//! let timeout_manager = manager.timeout(); -//! let _ = timeout_manager; -//! } -//! -//! // Use lock optimization (if enabled) -//! if manager.is_lock_enabled() { -//! let lock_manager = manager.lock(); -//! let _ = lock_manager; -//! } -//! -//! // Stop services -//! manager.stop().await; -//! # } -//! ``` +//! The actual runtime concurrency control — size-aware timeouts, byte-watermark +//! backpressure, request-hang/deadlock detection, and I/O scheduling — is +//! implemented in `rustfs/src/storage/*` on top of the `rustfs-io-core` +//! primitives. This crate only carries the data and contract types those +//! implementations share; it does not run any background tasks itself. #![deny(missing_docs)] #![deny(unsafe_code)] -#![cfg_attr(docsrs, feature(doc_cfg))] -// Re-export core types from io-core -pub use rustfs_io_core::{ - // Backpressure types - BackpressureConfig as CoreBackpressureConfig, - BackpressureMonitor as CoreBackpressureMonitor, - BackpressureState, +// Re-exported io-core type used by downstream timeout implementations. +pub use rustfs_io_core::OperationProgress; - // Deadlock types - DeadlockDetector as CoreDeadlockDetector, - IoLoadLevel, - IoLoadMetrics, - IoPriority, - // Scheduler types - IoScheduler, - IoSchedulingContext, - LockInfo, - LockOptimizer as CoreLockOptimizer, - - // Lock types - LockStats as CoreLockStats, - LockType, - // Timeout types - OperationProgress, - TimeoutError, - TimeoutStats, - WaitGraphEdge, - - calculate_adaptive_timeout, - estimate_bytes_per_second, -}; - -// Module declarations with feature gates -#[cfg(feature = "timeout")] -mod timeout; - -#[cfg(feature = "lock")] -mod lock; - -#[cfg(feature = "deadlock")] -mod deadlock; - -#[cfg(feature = "backpressure")] mod backpressure; - -#[cfg(feature = "scheduler")] -mod scheduler; - +mod deadlock; +mod queue; pub mod workers; pub mod workload; -// Public module exports with feature gates -#[cfg(feature = "timeout")] -pub use timeout::{TimeoutGuard, TimeoutManager, TimeoutManagerPolicy}; - -#[cfg(feature = "lock")] -pub use lock::{LockConfig, LockManager, LockScopeGuard, OptimizedLockGuard}; - -#[cfg(feature = "deadlock")] -pub use deadlock::{DeadlockManager, DeadlockMonitorPolicy, RequestTracker}; - -#[cfg(feature = "backpressure")] -pub use backpressure::{BackpressureManager, BackpressurePipe, PipeBackpressurePolicy}; - -#[cfg(feature = "scheduler")] -pub use scheduler::{IoStrategy, SchedulerManager, SchedulerPolicy}; - -// Configuration -mod config; -pub use config::{ConcurrencyConfig, ConcurrencyFeatures}; - -// Manager -mod manager; -pub use manager::{ConcurrencyManager, GetObjectQueueSnapshot}; +pub use backpressure::PipeBackpressurePolicy; +pub use deadlock::DeadlockMonitorPolicy; +pub use queue::GetObjectQueueSnapshot; pub use workload::{ AdmissionState, WorkloadAdmissionRegistrySnapshot, WorkloadAdmissionSnapshot, WorkloadAdmissionSnapshotProvider, WorkloadClass, }; - -// Prelude for convenient imports -pub mod prelude { - //! Prelude module for convenient imports - - #[cfg(feature = "timeout")] - pub use crate::timeout::{TimeoutGuard, TimeoutManager, TimeoutManagerPolicy}; - - #[cfg(feature = "lock")] - pub use crate::lock::{LockConfig, LockManager, LockScopeGuard, OptimizedLockGuard}; - - #[cfg(feature = "deadlock")] - pub use crate::deadlock::{DeadlockManager, DeadlockMonitorPolicy, RequestTracker}; - - #[cfg(feature = "backpressure")] - pub use crate::backpressure::{BackpressureManager, BackpressurePipe, PipeBackpressurePolicy}; - - #[cfg(feature = "scheduler")] - pub use crate::scheduler::{IoStrategy, SchedulerManager, SchedulerPolicy}; - - pub use crate::{AdmissionState, ConcurrencyConfig, ConcurrencyFeatures, ConcurrencyManager, WorkloadAdmissionSnapshot}; -} diff --git a/crates/concurrency/src/lock.rs b/crates/concurrency/src/lock.rs deleted file mode 100644 index 25a02aa24..000000000 --- a/crates/concurrency/src/lock.rs +++ /dev/null @@ -1,227 +0,0 @@ -// Copyright 2024 RustFS Team -// -// Licensed under the Apache License, Version 2.0 (the "License"); -// you may not use this file except in compliance with the License. -// You may obtain a copy of the License at -// -// http://www.apache.org/licenses/LICENSE-2.0 -// -// Unless required by applicable law or agreed to in writing, software -// distributed under the License is distributed on an "AS IS" BASIS, -// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -// See the License for the specific language governing permissions and -// limitations under the License. - -//! Lock optimization management - -use rustfs_io_core::{LockOptimizer as CoreLockOptimizer, LockStats}; -use rustfs_io_metrics::lock_metrics; -use std::sync::Arc; -use std::time::{Duration, Instant}; - -/// Lock configuration -#[derive(Debug, Clone)] -pub struct LockConfig { - /// Enable lock optimization - pub enabled: bool, - /// Lock acquisition timeout - pub acquire_timeout: Duration, -} - -impl Default for LockConfig { - fn default() -> Self { - Self { - enabled: true, - acquire_timeout: Duration::from_secs(5), - } - } -} - -/// Lock manager -pub struct LockManager { - config: LockConfig, - optimizer: Arc, -} - -impl LockManager { - /// Create a new lock manager - pub fn new(enabled: bool, acquire_timeout: Duration) -> Self { - let config = LockConfig { - enabled, - acquire_timeout, - }; - - let core_config = rustfs_io_core::LockOptimizeConfig { - enabled, - acquire_timeout, - max_hold_time_warning: Duration::from_millis(100), - adaptive_spin: true, - max_spin_iterations: 1000, - }; - - Self { - config, - optimizer: Arc::new(CoreLockOptimizer::new(core_config)), - } - } - - /// Get the configuration - pub fn config(&self) -> &LockConfig { - &self.config - } - - /// Get the core optimizer - pub fn optimizer(&self) -> Arc { - self.optimizer.clone() - } - - /// Get lock statistics - pub fn stats(&self) -> &LockStats { - self.optimizer.stats() - } - - /// Optimize a lock guard - pub fn optimize(&self, guard: G, resource: impl Into) -> OptimizedLockGuard { - OptimizedLockGuard::new(guard, resource, self.optimizer.clone()) - } - - /// Check if optimization is enabled - pub fn is_enabled(&self) -> bool { - self.config.enabled - } -} - -/// Optimized lock guard with early release support -pub struct OptimizedLockGuard { - guard: Option, - acquire_time: Instant, - released: bool, - resource: String, - optimizer: Arc, -} - -impl OptimizedLockGuard { - fn new(guard: G, resource: impl Into, optimizer: Arc) -> Self { - optimizer.on_acquire(); - lock_metrics::record_lock_optimization_enabled(optimizer.config().enabled); - - Self { - guard: Some(guard), - acquire_time: Instant::now(), - released: false, - resource: resource.into(), - optimizer, - } - } - - /// Get the lock hold time - 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 - pub fn early_release(&mut self) { - if self.released { - return; - } - - let hold_time = self.hold_time(); - self.guard.take(); - self.released = true; - - self.optimizer.on_release(hold_time); - lock_metrics::record_lock_hold_time(hold_time); - - tracing::debug!( - event = "lock_guard.release", - component = "concurrency", - subsystem = "lock", - release_mode = "early", - resource = %self.resource, - hold_time_ms = hold_time.as_millis(), - "lock guard released" - ); - } - - /// Get a reference to the underlying guard - pub fn as_ref(&self) -> Option<&G> { - if self.released { None } else { self.guard.as_ref() } - } -} - -impl Drop for OptimizedLockGuard { - fn drop(&mut self) { - if !self.released { - let hold_time = self.hold_time(); - self.guard.take(); - self.released = true; - - self.optimizer.on_release(hold_time); - lock_metrics::record_lock_hold_time(hold_time); - - tracing::debug!( - event = "lock_guard.release", - component = "concurrency", - subsystem = "lock", - release_mode = "drop", - resource = %self.resource, - hold_time_ms = hold_time.as_millis(), - "lock guard released" - ); - } - } -} - -/// Lock scope guard for RAII semantics -pub struct LockScopeGuard { - guard: Option, -} - -impl LockScopeGuard { - /// Create a new scope guard - pub fn new(guard: G) -> Self { - Self { guard: Some(guard) } - } - - /// Get a reference to the guard - pub fn as_ref(&self) -> Option<&G> { - self.guard.as_ref() - } -} - -impl Drop for LockScopeGuard { - fn drop(&mut self) { - self.guard.take(); - } -} - -#[cfg(test)] -mod tests { - use super::*; - use std::sync::Mutex; - - #[test] - fn test_lock_manager_creation() { - let manager = LockManager::new(true, Duration::from_secs(5)); - assert!(manager.is_enabled()); - } - - #[test] - fn test_optimized_lock_guard() { - let manager = LockManager::new(true, Duration::from_secs(5)); - let mutex = Mutex::new(42); - let guard = mutex.lock().unwrap(); - - let optimized = manager.optimize(guard, "test_resource"); - assert!(!optimized.is_released()); - - let mut optimized = optimized; - optimized.early_release(); - assert!(optimized.is_released()); - } -} diff --git a/crates/concurrency/src/manager.rs b/crates/concurrency/src/manager.rs deleted file mode 100644 index e302c52be..000000000 --- a/crates/concurrency/src/manager.rs +++ /dev/null @@ -1,397 +0,0 @@ -// Copyright 2024 RustFS Team -// -// Licensed under the Apache License, Version 2.0 (the "License"); -// you may not use this file except in compliance with the License. -// You may obtain a copy of the License at -// -// http://www.apache.org/licenses/LICENSE-2.0 -// -// Unless required by applicable law or agreed to in writing, software -// distributed under the License is distributed on an "AS IS" BASIS, -// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -// See the License for the specific language governing permissions and -// limitations under the License. - -//! Main concurrency manager - -use crate::config::{ConcurrencyConfig, ConfigError}; -use std::sync::Arc; - -/// Snapshot of disk permit queue usage for GetObject orchestration. -#[derive(Debug, Clone, Copy, PartialEq, Eq)] -pub struct GetObjectQueueSnapshot { - /// Total permits configured for disk reads. - pub total_permits: usize, - /// Permits currently in use. - pub permits_in_use: usize, -} - -impl GetObjectQueueSnapshot { - /// Create a queue snapshot from total and available permits. - pub fn from_available_permits(total_permits: usize, available_permits: usize) -> Self { - Self { - total_permits, - permits_in_use: total_permits.saturating_sub(available_permits), - } - } - - /// Return currently available permits. - pub fn permits_available(&self) -> usize { - self.total_permits.saturating_sub(self.permits_in_use) - } - - /// Return queue utilization percentage in the 0-100 range. - pub fn utilization_percent(&self) -> f64 { - if self.total_permits == 0 { - 0.0 - } else { - (self.permits_in_use as f64 / self.total_permits as f64) * 100.0 - } - } - - /// Return whether the queue is considered congested. - pub fn is_congested(&self, threshold_percent: f64) -> bool { - self.utilization_percent() > threshold_percent - } -} - -/// Main concurrency manager that provides access to all concurrency features -pub struct ConcurrencyManager { - config: ConcurrencyConfig, - - #[cfg(feature = "timeout")] - timeout: Arc, - - #[cfg(feature = "lock")] - lock: Arc, - - #[cfg(feature = "deadlock")] - deadlock: Arc, - - #[cfg(feature = "backpressure")] - backpressure: Arc, - - #[cfg(feature = "scheduler")] - scheduler: Arc, -} - -impl ConcurrencyManager { - fn build(config: ConcurrencyConfig) -> Self { - Self { - #[cfg(feature = "timeout")] - timeout: Arc::new(crate::timeout::TimeoutManager::from_policy(config.timeout_policy)), - - #[cfg(feature = "lock")] - lock: Arc::new(crate::lock::LockManager::new( - config.lock_policy.enabled, - config.lock_policy.acquire_timeout, - )), - - #[cfg(feature = "deadlock")] - deadlock: Arc::new(crate::deadlock::DeadlockManager::from_policy(config.deadlock_policy)), - - #[cfg(feature = "backpressure")] - backpressure: Arc::new(crate::backpressure::BackpressureManager::from_policy(config.backpressure_policy)), - - #[cfg(feature = "scheduler")] - scheduler: Arc::new(crate::scheduler::SchedulerManager::from_policy(config.scheduler_policy)), - - config, - } - } - - /// Try to create a new concurrency manager with the given configuration. - pub fn try_new(config: ConcurrencyConfig) -> Result { - config.validate()?; - Ok(Self::build(config)) - } - - /// Create a new concurrency manager with the given configuration. - /// - /// Invalid configurations are downgraded to the default configuration instead of - /// panicking so startup/runtime callers can remain fail-safe in production paths. - pub fn new(config: ConcurrencyConfig) -> Self { - match Self::try_new(config) { - Ok(manager) => manager, - Err(err) => { - tracing::warn!( - event = "concurrency_manager.invalid_config_fallback", - component = "concurrency", - subsystem = "manager", - error = %err, - "Invalid concurrency configuration detected; falling back to defaults" - ); - Self::build(ConcurrencyConfig::default()) - } - } - } - - /// Create with default configuration - pub fn with_defaults() -> Self { - Self::build(ConcurrencyConfig::default()) - } - - /// Try to create a manager from environment-derived configuration. - pub fn try_from_env() -> Result { - Self::try_new(ConcurrencyConfig::from_env()) - } - - /// Create from environment variables - pub fn from_env() -> Self { - match Self::try_from_env() { - Ok(manager) => manager, - Err(err) => { - tracing::warn!( - event = "concurrency_manager.invalid_env_fallback", - component = "concurrency", - subsystem = "manager", - error = %err, - "Invalid environment-derived concurrency configuration detected; falling back to defaults" - ); - Self::build(ConcurrencyConfig::default()) - } - } - } - - /// Get the configuration - pub fn config(&self) -> &ConcurrencyConfig { - &self.config - } - - // ============================================ - // Feature enable checks - // ============================================ - - /// Check if timeout feature is enabled - pub fn is_timeout_enabled(&self) -> bool { - #[cfg(feature = "timeout")] - { - self.config.features.timeout - } - #[cfg(not(feature = "timeout"))] - { - false - } - } - - /// Check if lock feature is enabled - pub fn is_lock_enabled(&self) -> bool { - #[cfg(feature = "lock")] - { - self.config.features.lock - } - #[cfg(not(feature = "lock"))] - { - false - } - } - - /// Check if deadlock feature is enabled - pub fn is_deadlock_enabled(&self) -> bool { - #[cfg(feature = "deadlock")] - { - self.config.features.deadlock - } - #[cfg(not(feature = "deadlock"))] - { - false - } - } - - /// Check if backpressure feature is enabled - pub fn is_backpressure_enabled(&self) -> bool { - #[cfg(feature = "backpressure")] - { - self.config.features.backpressure - } - #[cfg(not(feature = "backpressure"))] - { - false - } - } - - /// Check if scheduler feature is enabled - pub fn is_scheduler_enabled(&self) -> bool { - #[cfg(feature = "scheduler")] - { - self.config.features.scheduler - } - #[cfg(not(feature = "scheduler"))] - { - false - } - } - - // ============================================ - // Feature accessors - // ============================================ - - /// Get timeout manager - #[cfg(feature = "timeout")] - pub fn timeout(&self) -> Arc { - self.timeout.clone() - } - - /// Get lock manager - #[cfg(feature = "lock")] - pub fn lock(&self) -> Arc { - self.lock.clone() - } - - /// Get deadlock manager - #[cfg(feature = "deadlock")] - pub fn deadlock(&self) -> Arc { - self.deadlock.clone() - } - - /// Get backpressure manager - #[cfg(feature = "backpressure")] - pub fn backpressure(&self) -> Arc { - self.backpressure.clone() - } - - /// Get scheduler manager - #[cfg(feature = "scheduler")] - pub fn scheduler(&self) -> Arc { - self.scheduler.clone() - } - - // ============================================ - // Lifecycle management - // ============================================ - - /// Start all enabled services (e.g., deadlock detection background task) - pub async fn start(&self) { - #[cfg(feature = "deadlock")] - { - if self.config.deadlock_policy.enabled { - self.deadlock.start().await; - } - } - - tracing::info!( - event = "concurrency_manager.lifecycle", - component = "concurrency", - subsystem = "manager", - state = "started", - timeout_enabled = self.is_timeout_enabled(), - lock_enabled = self.is_lock_enabled(), - deadlock_enabled = self.is_deadlock_enabled(), - backpressure_enabled = self.is_backpressure_enabled(), - scheduler_enabled = self.is_scheduler_enabled(), - "concurrency manager state changed" - ); - } - - /// Stop all services - pub async fn stop(&self) { - #[cfg(feature = "deadlock")] - { - self.deadlock.stop().await; - } - - tracing::info!( - event = "concurrency_manager.lifecycle", - component = "concurrency", - subsystem = "manager", - state = "stopped", - "concurrency manager state changed" - ); - } -} - -impl Default for ConcurrencyManager { - fn default() -> Self { - Self::with_defaults() - } -} - -#[cfg(test)] -mod tests { - use super::*; - - #[test] - fn test_queue_snapshot() { - let snapshot = GetObjectQueueSnapshot::from_available_permits(64, 16); - assert_eq!(snapshot.permits_in_use, 48); - assert_eq!(snapshot.permits_available(), 16); - assert!(snapshot.is_congested(70.0)); - } - - #[test] - fn test_queue_snapshot_clamps_over_available_permits() { - let snapshot = GetObjectQueueSnapshot::from_available_permits(64, 96); - assert_eq!(snapshot.permits_in_use, 0); - assert_eq!(snapshot.permits_available(), 64); - assert_eq!(snapshot.utilization_percent(), 0.0); - assert!(!snapshot.is_congested(0.0)); - } - - #[test] - fn test_queue_snapshot_handles_zero_total_permits() { - let snapshot = GetObjectQueueSnapshot::from_available_permits(0, 0); - assert_eq!(snapshot.permits_in_use, 0); - assert_eq!(snapshot.permits_available(), 0); - assert_eq!(snapshot.utilization_percent(), 0.0); - assert!(!snapshot.is_congested(0.0)); - } - - #[test] - fn test_manager_creation() { - let manager = ConcurrencyManager::with_defaults(); - assert!(manager.config().validate().is_ok()); - } - - #[test] - fn test_try_new_returns_error_for_invalid_config() { - let config = ConcurrencyConfig { - timeout_policy: crate::TimeoutManagerPolicy { - default_timeout: std::time::Duration::from_secs(10), - max_timeout: std::time::Duration::from_secs(1), - ..Default::default() - }, - ..Default::default() - }; - - let result = ConcurrencyManager::try_new(config); - assert!(matches!(result, Err(ConfigError::InvalidTimeout(_)))); - } - - #[test] - fn test_new_falls_back_to_default_for_invalid_config() { - let config = ConcurrencyConfig { - timeout_policy: crate::TimeoutManagerPolicy { - default_timeout: std::time::Duration::from_secs(10), - max_timeout: std::time::Duration::from_secs(1), - ..Default::default() - }, - ..Default::default() - }; - - let manager = ConcurrencyManager::new(config); - assert!(manager.config().validate().is_ok()); - assert_eq!( - manager.config().timeout_policy.default_timeout, - ConcurrencyConfig::default().timeout_policy.default_timeout - ); - } - - #[tokio::test] - async fn test_manager_lifecycle() { - let manager = ConcurrencyManager::with_defaults(); - manager.start().await; - manager.stop().await; - } - - #[test] - fn test_feature_checks() { - let manager = ConcurrencyManager::with_defaults(); - - // These should return the feature flag status - let _ = manager.is_timeout_enabled(); - let _ = manager.is_lock_enabled(); - let _ = manager.is_deadlock_enabled(); - let _ = manager.is_backpressure_enabled(); - let _ = manager.is_scheduler_enabled(); - } -} diff --git a/crates/concurrency/src/queue.rs b/crates/concurrency/src/queue.rs new file mode 100644 index 000000000..39826d262 --- /dev/null +++ b/crates/concurrency/src/queue.rs @@ -0,0 +1,84 @@ +// 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. + +//! Disk permit queue snapshot for GetObject orchestration. + +/// Snapshot of disk permit queue usage for GetObject orchestration. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct GetObjectQueueSnapshot { + /// Total permits configured for disk reads. + pub total_permits: usize, + /// Permits currently in use. + pub permits_in_use: usize, +} + +impl GetObjectQueueSnapshot { + /// Create a queue snapshot from total and available permits. + pub fn from_available_permits(total_permits: usize, available_permits: usize) -> Self { + Self { + total_permits, + permits_in_use: total_permits.saturating_sub(available_permits), + } + } + + /// Return currently available permits. + pub fn permits_available(&self) -> usize { + self.total_permits.saturating_sub(self.permits_in_use) + } + + /// Return queue utilization percentage in the 0-100 range. + pub fn utilization_percent(&self) -> f64 { + if self.total_permits == 0 { + 0.0 + } else { + (self.permits_in_use as f64 / self.total_permits as f64) * 100.0 + } + } + + /// Return whether the queue is considered congested. + pub fn is_congested(&self, threshold_percent: f64) -> bool { + self.utilization_percent() > threshold_percent + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn test_queue_snapshot() { + let snapshot = GetObjectQueueSnapshot::from_available_permits(64, 16); + assert_eq!(snapshot.permits_in_use, 48); + assert_eq!(snapshot.permits_available(), 16); + assert!(snapshot.is_congested(70.0)); + } + + #[test] + fn test_queue_snapshot_clamps_over_available_permits() { + let snapshot = GetObjectQueueSnapshot::from_available_permits(64, 96); + assert_eq!(snapshot.permits_in_use, 0); + assert_eq!(snapshot.permits_available(), 64); + assert_eq!(snapshot.utilization_percent(), 0.0); + assert!(!snapshot.is_congested(0.0)); + } + + #[test] + fn test_queue_snapshot_handles_zero_total_permits() { + let snapshot = GetObjectQueueSnapshot::from_available_permits(0, 0); + assert_eq!(snapshot.permits_in_use, 0); + assert_eq!(snapshot.permits_available(), 0); + assert_eq!(snapshot.utilization_percent(), 0.0); + assert!(!snapshot.is_congested(0.0)); + } +} diff --git a/crates/concurrency/src/scheduler.rs b/crates/concurrency/src/scheduler.rs deleted file mode 100644 index fc892f10e..000000000 --- a/crates/concurrency/src/scheduler.rs +++ /dev/null @@ -1,292 +0,0 @@ -// Copyright 2024 RustFS Team -// -// Licensed under the Apache License, Version 2.0 (the "License"); -// you may not use this file except in compliance with the License. -// You may obtain a copy of the License at -// -// http://www.apache.org/licenses/LICENSE-2.0 -// -// Unless required by applicable law or agreed to in writing, software -// distributed under the License is distributed on an "AS IS" BASIS, -// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -// See the License for the specific language governing permissions and -// limitations under the License. - -//! I/O scheduler management - -use rustfs_io_core::{ - IoLoadLevel, IoPriority, IoScheduler as CoreIoScheduler, IoSchedulingContext, - io_profile::{AccessPattern, StorageMedia}, -}; -use rustfs_io_metrics::io_metrics; -use std::sync::Arc; -use std::time::Duration; - -/// Facade policy for the concurrency-layer scheduler manager. -#[derive(Debug, Clone, Copy)] -pub struct SchedulerPolicy { - /// Base buffer size - pub base_buffer_size: usize, - /// Maximum buffer size - pub max_buffer_size: usize, - /// High priority threshold - pub high_priority_threshold: usize, - /// Low priority threshold - pub low_priority_threshold: usize, -} - -impl Default for SchedulerPolicy { - fn default() -> Self { - Self { - base_buffer_size: 64 * 1024, // 64KB - max_buffer_size: 4 * 1024 * 1024, // 4MB - high_priority_threshold: 1024 * 1024, // 1MB - low_priority_threshold: 10 * 1024 * 1024, // 10MB - } - } -} - -impl SchedulerPolicy { - /// Convert facade policy to io-core scheduler config. - pub fn to_core_config(&self) -> rustfs_io_core::IoSchedulerConfig { - let default = rustfs_io_core::IoSchedulerConfig::default(); - rustfs_io_core::IoSchedulerConfig { - base_buffer_size: self.base_buffer_size, - max_buffer_size: self.max_buffer_size, - // Keep min <= base <= max: the io-core default min (4KB) would make - // `clamp(min, max)` panic for policies with a smaller max_buffer_size. - min_buffer_size: default.min_buffer_size.min(self.base_buffer_size).min(self.max_buffer_size), - high_priority_size_threshold: self.high_priority_threshold, - low_priority_size_threshold: self.low_priority_threshold, - ..default - } - } -} - -/// Scheduler manager -pub struct SchedulerManager { - config: SchedulerPolicy, - core_config: rustfs_io_core::IoSchedulerConfig, - scheduler: Arc, -} - -impl SchedulerManager { - /// Create a new scheduler manager - pub fn new( - base_buffer_size: usize, - max_buffer_size: usize, - high_priority_threshold: usize, - low_priority_threshold: usize, - ) -> Self { - Self::from_policy(SchedulerPolicy { - base_buffer_size, - max_buffer_size, - high_priority_threshold, - low_priority_threshold, - }) - } - - /// Create a scheduler manager from facade policy. - pub fn from_policy(config: SchedulerPolicy) -> Self { - let core_config = config.to_core_config(); - - Self { - config, - core_config: core_config.clone(), - scheduler: Arc::new(CoreIoScheduler::new(core_config)), - } - } - - /// Get the configuration - pub fn config(&self) -> &SchedulerPolicy { - &self.config - } - - /// Get the derived io-core scheduler config. - pub fn core_config(&self) -> &rustfs_io_core::IoSchedulerConfig { - &self.core_config - } - - /// Get the scheduler - pub fn scheduler(&self) -> Arc { - self.scheduler.clone() - } - - /// Create an I/O strategy - pub fn create_strategy(&self) -> IoStrategy { - IoStrategy::new(self.config, self.scheduler.clone()) - } - - /// Calculate buffer size - pub fn calculate_buffer_size( - &self, - file_size: i64, - media: StorageMedia, - pattern: AccessPattern, - load: IoLoadLevel, - concurrent: usize, - ) -> usize { - let strategy = self.create_strategy(); - strategy.calculate_buffer_size(file_size, media, pattern, load, concurrent) - } - - /// Get I/O priority - pub fn get_priority(&self, size: i64) -> IoPriority { - IoPriority::from_size(size, self.config.high_priority_threshold, self.config.low_priority_threshold) - } -} - -/// I/O strategy -pub struct IoStrategy { - config: SchedulerPolicy, - scheduler: Arc, -} - -impl IoStrategy { - fn new(config: SchedulerPolicy, scheduler: Arc) -> Self { - Self { config, scheduler } - } - - /// Calculate buffer size with multi-factor strategy - pub fn calculate_buffer_size( - &self, - file_size: i64, - media: StorageMedia, - pattern: AccessPattern, - load: IoLoadLevel, - concurrent: usize, - ) -> usize { - // Create scheduling context - let _ctx = IoSchedulingContext::new(file_size, self.config.base_buffer_size) - .with_sequential(matches!(pattern, AccessPattern::Sequential)) - .with_media(media); - - // Get base buffer size from core scheduler - let permit_wait = Duration::from_millis(10); // Default wait time - let is_sequential = matches!(pattern, AccessPattern::Sequential); - let core_strategy = self.scheduler.calculate_strategy(file_size, permit_wait, is_sequential); - let base_size = core_strategy.buffer_size; - - // Apply multi-factor adjustments - let adjusted_size = self.apply_adjustments(base_size, media, pattern, load, concurrent); - - // Record metrics - io_metrics::record_io_scheduler_decision(adjusted_size, load.as_str(), pattern.as_str()); - - adjusted_size.min(self.config.max_buffer_size) - } - - fn apply_adjustments( - &self, - base_size: usize, - media: StorageMedia, - pattern: AccessPattern, - load: IoLoadLevel, - concurrent: usize, - ) -> usize { - let mut size = base_size; - - // Media adjustment - size = match media { - StorageMedia::Nvme => (size as f64 * 1.5) as usize, - StorageMedia::Ssd => (size as f64 * 1.2) as usize, - StorageMedia::Hdd => size, - _ => size, - }; - - // Pattern adjustment - size = match pattern { - AccessPattern::Sequential => (size as f64 * 1.5) as usize, - AccessPattern::Random => (size as f64 * 0.5) as usize, - _ => size, - }; - - // Load adjustment - size = match load { - IoLoadLevel::Low => (size as f64 * 1.2) as usize, - IoLoadLevel::Medium => size, - IoLoadLevel::High => (size as f64 * 0.7) as usize, - IoLoadLevel::Critical => (size as f64 * 0.5) as usize, - }; - - // Concurrency adjustment - if concurrent > 10 { - size = (size as f64 * 0.8) as usize; - } - - size - } - - /// Get the configuration - pub fn config(&self) -> &SchedulerPolicy { - &self.config - } -} - -#[cfg(test)] -mod tests { - use super::*; - - #[test] - fn test_scheduler_config() { - let config = SchedulerPolicy::default(); - assert!(config.base_buffer_size < config.max_buffer_size); - } - - #[test] - fn test_scheduler_policy_to_core_config() { - let policy = SchedulerPolicy::default(); - let core = policy.to_core_config(); - assert_eq!(core.base_buffer_size, policy.base_buffer_size); - assert_eq!(core.max_buffer_size, policy.max_buffer_size); - assert_eq!(core.high_priority_size_threshold, policy.high_priority_threshold); - assert_eq!(core.low_priority_size_threshold, policy.low_priority_threshold); - } - - #[test] - fn test_scheduler_default_thresholds_remain_stable() { - let policy = SchedulerPolicy::default(); - assert_eq!(policy.base_buffer_size, 64 * 1024); - assert_eq!(policy.max_buffer_size, 4 * 1024 * 1024); - assert_eq!(policy.high_priority_threshold, 1024 * 1024); - assert_eq!(policy.low_priority_threshold, 10 * 1024 * 1024); - } - - #[test] - fn test_scheduler_priority_boundaries_remain_stable() { - let policy = SchedulerPolicy::default(); - let manager = SchedulerManager::from_policy(policy); - - assert_eq!(manager.get_priority(1_048_575), IoPriority::High); - assert_eq!(manager.get_priority(1_048_576), IoPriority::Normal); - assert_eq!(manager.get_priority(10_485_760), IoPriority::Normal); - assert_eq!(manager.get_priority(10_485_761), IoPriority::Low); - } - - #[test] - fn test_scheduler_manager() { - let manager = SchedulerManager::new(1024, 4096, 512, 2048); - let priority = manager.get_priority(100); - assert!(priority.is_high()); - } - - #[test] - fn test_small_max_buffer_size_does_not_panic() { - // Regression: max_buffer_size below the io-core default min (4KB) used to - // panic in the core clamp (min > max) on the first buffer-size calculation. - let manager = SchedulerManager::new(1024, 2048, 512, 2048); - let size = manager.calculate_buffer_size(1024, StorageMedia::Ssd, AccessPattern::Sequential, IoLoadLevel::Medium, 1); - assert!(size > 0 && size <= 2048); - assert!(manager.core_config().validate().is_ok()); - } - - #[test] - fn test_io_strategy() { - let manager = SchedulerManager::new(1024, 4096, 512, 2048); - let strategy = manager.create_strategy(); - - let size = strategy.calculate_buffer_size(1024 * 1024, StorageMedia::Ssd, AccessPattern::Sequential, IoLoadLevel::Low, 1); - - assert!(size > 0); - } -} diff --git a/crates/concurrency/src/timeout.rs b/crates/concurrency/src/timeout.rs deleted file mode 100644 index 650cc91dc..000000000 --- a/crates/concurrency/src/timeout.rs +++ /dev/null @@ -1,222 +0,0 @@ -// Copyright 2024 RustFS Team -// -// Licensed under the Apache License, Version 2.0 (the "License"); -// you may not use this file except in compliance with the License. -// You may obtain a copy of the License at -// -// http://www.apache.org/licenses/LICENSE-2.0 -// -// Unless required by applicable law or agreed to in writing, software -// distributed under the License is distributed on an "AS IS" BASIS, -// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -// See the License for the specific language governing permissions and -// limitations under the License. - -//! Timeout management for operations - -use rustfs_io_core::{TimeoutConfig as CoreTimeoutConfig, TimeoutError, calculate_adaptive_timeout}; -use std::time::{Duration, Instant}; -use tokio_util::sync::CancellationToken; - -/// Facade policy for the concurrency-layer timeout manager. -#[derive(Debug, Clone, Copy)] -pub struct TimeoutManagerPolicy { - /// Default timeout duration - pub default_timeout: Duration, - /// Maximum timeout duration - pub max_timeout: Duration, - /// Minimum timeout floor (prevents dynamic calculation from going too low). - pub min_timeout: Duration, - /// Enable dynamic timeout calculation - pub enable_dynamic: bool, -} - -impl Default for TimeoutManagerPolicy { - fn default() -> Self { - Self { - default_timeout: Duration::from_secs(30), - max_timeout: Duration::from_secs(300), - min_timeout: Duration::from_secs(5), - enable_dynamic: true, - } - } -} - -impl TimeoutManagerPolicy { - /// Convert the facade policy into the reusable io-core timeout configuration. - /// - /// This keeps the concurrency layer explicitly wired to the shared core - /// timeout primitives without changing the facade's public behavior. - pub fn to_core_config(&self) -> CoreTimeoutConfig { - CoreTimeoutConfig { - base_timeout: self.default_timeout, - timeout_per_mb: Duration::ZERO, - max_timeout: self.max_timeout, - min_timeout: self.min_timeout, - get_object_timeout: self.default_timeout, - put_object_timeout: self.max_timeout, - list_objects_timeout: self.default_timeout, - enable_dynamic_timeout: self.enable_dynamic, - } - } -} - -/// Timeout manager -pub struct TimeoutManager { - config: TimeoutManagerPolicy, - core_config: CoreTimeoutConfig, -} - -impl TimeoutManager { - /// Create a new timeout manager - pub fn new(default_timeout: Duration, max_timeout: Duration, enable_dynamic: bool) -> Self { - let min_timeout = default_timeout.min(max_timeout); - Self::from_policy(TimeoutManagerPolicy { - default_timeout, - max_timeout, - min_timeout, - enable_dynamic, - }) - } - - /// Create a new timeout manager from the facade policy type. - pub fn from_policy(config: TimeoutManagerPolicy) -> Self { - let config = TimeoutManagerPolicy { - // Guard clamp(min, max) from panic when callers provide an - // out-of-order policy (or very small max_timeout). - min_timeout: config.min_timeout.min(config.max_timeout), - ..config - }; - let core_config = config.to_core_config(); - Self { config, core_config } - } - - /// Get the configuration - pub fn config(&self) -> &TimeoutManagerPolicy { - &self.config - } - - /// Get the derived io-core timeout configuration. - pub fn core_config(&self) -> &CoreTimeoutConfig { - &self.core_config - } - - /// Calculate timeout for a given size - pub fn calculate_timeout(&self, size: u64, _history: &[Duration]) -> Duration { - if !self.config.enable_dynamic { - return self.config.default_timeout; - } - - calculate_adaptive_timeout(self.core_config.base_timeout, None, 0, size) - .clamp(self.core_config.min_timeout, self.core_config.max_timeout) - } - - /// Wrap an operation with timeout control - pub async fn wrap_operation(&self, operation: F, timeout: Option) -> Result - where - F: std::future::Future>, - E: Into, - { - let timeout = timeout.unwrap_or(self.config.default_timeout); - - match tokio::time::timeout(timeout, operation).await { - Ok(Ok(result)) => Ok(result), - Ok(Err(e)) => Err(e.into()), - Err(_) => Err(TimeoutError::TimedOut(timeout)), - } - } - - /// Create a timeout guard for manual timeout control - pub fn create_guard(&self, timeout: Option) -> TimeoutGuard { - TimeoutGuard::new(timeout.unwrap_or(self.core_config.base_timeout)) - } -} - -/// Timeout guard for manual timeout control -pub struct TimeoutGuard { - timeout: Duration, - start: Instant, - cancel_token: CancellationToken, -} - -impl TimeoutGuard { - fn new(timeout: Duration) -> Self { - Self { - timeout, - start: Instant::now(), - cancel_token: CancellationToken::new(), - } - } - - /// Check if timeout has elapsed - pub fn is_timed_out(&self) -> bool { - self.start.elapsed() > self.timeout - } - - /// Get remaining time - pub fn remaining(&self) -> Duration { - self.timeout.saturating_sub(self.start.elapsed()) - } - - /// Get the cancellation token - pub fn cancel_token(&self) -> CancellationToken { - self.cancel_token.clone() - } - - /// Cancel the operation - pub fn cancel(&self) { - self.cancel_token.cancel(); - } -} - -#[cfg(test)] -mod tests { - use super::*; - - #[test] - fn test_timeout_config() { - let config = TimeoutManagerPolicy::default(); - assert!(config.default_timeout < config.max_timeout); - } - - #[test] - fn test_timeout_policy_to_core_config() { - let policy = TimeoutManagerPolicy::default(); - let core = policy.to_core_config(); - assert_eq!(core.base_timeout, policy.default_timeout); - assert_eq!(core.max_timeout, policy.max_timeout); - assert_eq!(core.min_timeout, policy.min_timeout); - assert_eq!(core.get_object_timeout, policy.default_timeout); - assert!(core.enable_dynamic_timeout); - } - - #[test] - fn test_timeout_manager_new_sanitizes_min_timeout_with_small_max_timeout() { - let manager = TimeoutManager::new(Duration::from_secs(1), Duration::from_secs(1), true); - let timeout = manager.calculate_timeout(1024, &[]); - assert_eq!(timeout, Duration::from_secs(1)); - } - - #[test] - fn test_timeout_manager_from_policy_sanitizes_min_timeout() { - let manager = TimeoutManager::from_policy(TimeoutManagerPolicy { - default_timeout: Duration::from_secs(30), - max_timeout: Duration::from_secs(1), - min_timeout: Duration::from_secs(5), - enable_dynamic: true, - }); - - assert_eq!(manager.config().min_timeout, Duration::from_secs(1)); - assert_eq!(manager.core_config().min_timeout, Duration::from_secs(1)); - } - - #[tokio::test] - async fn test_wrap_operation_success() { - let manager = TimeoutManager::new(Duration::from_secs(5), Duration::from_secs(10), true); - - let result = manager.wrap_operation(async { Ok::<_, TimeoutError>(42) }, None).await; - - assert!(result.is_ok()); - assert_eq!(result.unwrap(), 42); - } -} diff --git a/docs/architecture/scheduler-baseline.md b/docs/architecture/scheduler-baseline.md index 2a5b97181..94144adc7 100644 --- a/docs/architecture/scheduler-baseline.md +++ b/docs/architecture/scheduler-baseline.md @@ -9,9 +9,8 @@ runtime-builder ownership. It does not define new behavior. | Surface | Current owner | Current responsibility | Migration boundary | |---|---|---|---| | `ConcurrencyManager` | `rustfs/src/storage/concurrency/manager.rs` | Owns the RustFS S3 read-path disk-read semaphore, I/O metrics, priority queue, storage media detection, access-pattern detection, and buffer strategy. | Keep request admission and I/O metrics behavior stable until a controller can consume the same state explicitly. | -| `SchedulerManager` | `crates/concurrency/src/scheduler.rs` | Provides a reusable facade over `rustfs-io-core::IoScheduler` and derives buffer/priority decisions from `SchedulerPolicy`. | Treat it as reusable library surface; do not assume the RustFS S3 read path has already moved to this facade. | -| `BackpressureManager` | `crates/concurrency/src/backpressure.rs` | Provides a reusable duplex-pipe backpressure facade over `rustfs-io-core::BackpressureMonitor`. | Keep pipe sizing and watermark policy separate from object-read disk semaphore admission. | -| RustFS backpressure monitor | `rustfs/src/storage/backpressure.rs` | Tracks object-pipe watermark state used by RustFS storage backpressure tests and helpers. | Preserve current state labels and watermark semantics when consolidating with reusable facades. | +| I/O scheduler core | `crates/io-core/src/scheduler.rs` | Owns the reusable buffer-size and priority algorithms consumed by the RustFS S3 read path (`rustfs/src/storage/concurrency/io_schedule.rs`). The former `SchedulerManager` facade in `rustfs-concurrency` was removed as zero-caller dead code (backlog#1025). | Treat `rustfs-io-core` as the reusable algorithm surface; the RustFS S3 read path owns its own scheduling wiring. | +| RustFS backpressure monitor | `rustfs/src/storage/backpressure.rs` | Tracks object-pipe watermark state used by RustFS storage backpressure tests and helpers, using the shared `PipeBackpressurePolicy` from `crates/concurrency/src/backpressure.rs`. The former `BackpressureManager`/`BackpressurePipe` facade in `rustfs-concurrency` was removed as zero-caller dead code (backlog#1025). | Preserve current state labels and watermark semantics; keep pipe sizing and watermark policy separate from object-read disk semaphore admission. | | `Workers` | `crates/concurrency/src/workers.rs` | Provides cooperative worker-slot admission with `take`, `give`, and `wait`; current background workflows use it for bounded set workers. | Preserve blocking/wakeup semantics and over-release clamping. | | Scanner cycle budget | `crates/scanner/src/scanner_budget.rs` | Cancels a child token when runtime, object-count, or directory-count budget is reached. | Preserve partial-cycle reason mapping and checkpoint accounting. | | Heal admission | `crates/heal/src/heal/manager.rs`, `crates/heal/src/heal/channel.rs`, `rustfs_common::heal_channel` | Owns priority queue admission, duplicate merge/drop/full results, active-task tracking, retry admission, and channel responses. | Preserve low-priority scanner behavior and high-priority escalation gates. | diff --git a/docs/architecture/workload-admission-contracts.md b/docs/architecture/workload-admission-contracts.md index 34f260022..15f7ce9e5 100644 --- a/docs/architecture/workload-admission-contracts.md +++ b/docs/architecture/workload-admission-contracts.md @@ -5,18 +5,18 @@ preservation and runtime workload-class contract slice. ## Preservation Coverage -The `rustfs-concurrency` tests pin the current reusable scheduler and -admission-facing behavior before later snapshot extraction: +The `rustfs-concurrency` tests pin the current reusable admission-facing +behavior before later snapshot extraction: - Worker slot over-release remains clamped by the configured worker limit. -- Scheduler default buffer and priority thresholds remain unchanged. -- Scheduler priority boundaries remain high below the high threshold, normal at - both thresholds, and low above the low threshold. -- Backpressure pipe metadata reads preserve buffer capacity and state without - mutating the manager state. - `GetObjectQueueSnapshot` preserves saturated, over-available, and zero-total permit semantics. +The former reusable scheduler and backpressure-pipe facades (and their +preservation tests) were removed as zero-caller dead code in backlog#1025; +scheduler buffer/priority behavior is now pinned by `rustfs-io-core` and +`rustfs/src/storage/concurrency` tests. + ## Workload Class Contract `WorkloadClass` defines the required future admission categories: diff --git a/scripts/check_logging_guardrails.sh b/scripts/check_logging_guardrails.sh index 00e73f432..2c3462693 100755 --- a/scripts/check_logging_guardrails.sh +++ b/scripts/check_logging_guardrails.sh @@ -54,8 +54,6 @@ checked_files=( "crates/scanner/src/scanner_io.rs" "crates/scanner/src/scanner_folder.rs" "crates/concurrency/src/workers.rs" - "crates/concurrency/src/manager.rs" - "crates/concurrency/src/lock.rs" "crates/concurrency/src/deadlock.rs" "crates/trusted-proxies/src/global.rs" "crates/trusted-proxies/src/config/loader.rs"