rustfs/crates/concurrency/src/manager.rs

// 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<crate::timeout::TimeoutManager>,

    #[cfg(feature = "lock")]
    lock: Arc<crate::lock::LockManager>,

    #[cfg(feature = "deadlock")]
    deadlock: Arc<crate::deadlock::DeadlockManager>,

    #[cfg(feature = "backpressure")]
    backpressure: Arc<crate::backpressure::BackpressureManager>,

    #[cfg(feature = "scheduler")]
    scheduler: Arc<crate::scheduler::SchedulerManager>,
}

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<Self, ConfigError> {
        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, ConfigError> {
        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<crate::timeout::TimeoutManager> {
        self.timeout.clone()
    }

    /// Get lock manager
    #[cfg(feature = "lock")]
    pub fn lock(&self) -> Arc<crate::lock::LockManager> {
        self.lock.clone()
    }

    /// Get deadlock manager
    #[cfg(feature = "deadlock")]
    pub fn deadlock(&self) -> Arc<crate::deadlock::DeadlockManager> {
        self.deadlock.clone()
    }

    /// Get backpressure manager
    #[cfg(feature = "backpressure")]
    pub fn backpressure(&self) -> Arc<crate::backpressure::BackpressureManager> {
        self.backpressure.clone()
    }

    /// Get scheduler manager
    #[cfg(feature = "scheduler")]
    pub fn scheduler(&self) -> Arc<crate::scheduler::SchedulerManager> {
        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();
    }
}