fix

Cargo.lock

@@ -705,7 +705,7 @@ dependencies = [
  "http 0.2.12",
  "http 1.3.1",
  "http-body 0.4.6",
- "lru",
+ "lru 0.12.5",
  "percent-encoding",
  "regex-lite",
  "sha2 0.10.9",
@@ -4881,6 +4881,15 @@ dependencies = [
  "hashbrown 0.15.5",
 ]
 
+[[package]]
+name = "lru"
+version = "0.16.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "96051b46fc183dc9cd4a223960ef37b9af631b55191852a8274bfef064cda20f"
+dependencies = [
+ "hashbrown 0.16.1",
+]
+
 [[package]]
 name = "lru-slab"
 version = "0.1.2"
@@ -6949,7 +6958,7 @@ dependencies = [
  "hyper-util",
  "jemalloc_pprof",
  "libsystemd",
- "lru",
+ "lru 0.16.2",
  "matchit 0.9.0",
  "md5",
  "metrics",

Cargo.toml

@@ -196,6 +196,7 @@ lazy_static = "1.5.0"
 libc = "0.2.177"
 libsystemd = "0.7.2"
 local-ip-address = "0.6.5"
+lru = "0.16.2"
 lz4 = "1.28.1"
 matchit = "0.9.0"
 md-5 = "0.11.0-rc.3"

rustfs/Cargo.toml

@@ -107,7 +107,7 @@ clap = { workspace = true }
 const-str = { workspace = true }
 datafusion = { workspace = true }
 hex-simd.workspace = true
-lru = "0.12"
+lru = { workspace = true }
 matchit = { workspace = true }
 md5.workspace = true
 mime_guess = { workspace = true }

rustfs/src/storage/concurrency.rs

@@ -34,7 +34,7 @@ use rustfs_config::{KI_B, MI_B};
 use std::sync::atomic::{AtomicUsize, Ordering};
 use std::sync::{Arc, LazyLock};
 use std::time::{Duration, Instant};
-use tokio::sync::{Semaphore, RwLock};
+use tokio::sync::{RwLock, Semaphore};
 
 /// Global concurrent request counter for adaptive buffer sizing
 static ACTIVE_GET_REQUESTS: AtomicUsize = AtomicUsize::new(0);
@@ -81,7 +81,7 @@ impl GetObjectGuard {
 impl Drop for GetObjectGuard {
     fn drop(&mut self) {
         ACTIVE_GET_REQUESTS.fetch_sub(1, Ordering::Relaxed);
-        
+
         // Record metrics for monitoring
         #[cfg(feature = "metrics")]
         {
@@ -113,19 +113,19 @@ impl Drop for GetObjectGuard {
 /// Optimized buffer size in bytes for the current concurrency level
 pub fn get_concurrency_aware_buffer_size(file_size: i64, base_buffer_size: usize) -> usize {
     let concurrent_requests = ACTIVE_GET_REQUESTS.load(Ordering::Relaxed);
-    
+
     // Record concurrent request metrics
     #[cfg(feature = "metrics")]
     {
         use metrics::gauge;
         gauge!("rustfs_concurrent_get_requests").set(concurrent_requests as f64);
     }
-    
+
     // For low concurrency, use the base buffer size for maximum throughput
     if concurrent_requests <= 1 {
         return base_buffer_size;
     }
-    
+
     // Calculate adaptive multiplier based on concurrency level
     let adaptive_multiplier = if concurrent_requests <= 2 {
         // Low concurrency (1-2): use full buffer for maximum throughput
@@ -140,23 +140,23 @@ pub fn get_concurrency_aware_buffer_size(file_size: i64, base_buffer_size: usize
         // Very high concurrency (>8): minimize memory per request (40% of base)
         0.4
     };
-    
+
     // Calculate the adjusted buffer size
     let adjusted_size = (base_buffer_size as f64 * adaptive_multiplier) as usize;
-    
+
     // Ensure we stay within reasonable bounds
     let min_buffer = if file_size > 0 && file_size < 100 * KI_B as i64 {
         32 * KI_B // For very small files, use minimum buffer
     } else {
         64 * KI_B // Standard minimum buffer size
     };
-    
+
     let max_buffer = if concurrent_requests > HIGH_CONCURRENCY_THRESHOLD {
         256 * KI_B // Cap at 256KB for high concurrency
     } else {
         MI_B // Cap at 1MB for lower concurrency
     };
-    
+
     adjusted_size.clamp(min_buffer, max_buffer)
 }
 
@@ -177,7 +177,7 @@ struct HotObjectCache {
 }
 
 /// A cached object with metadata
-#[derive(Debug, Clone)]
+#[derive(Debug)]
 struct CachedObject {
     /// The object data
     data: Arc<Vec<u8>>,
@@ -199,50 +199,50 @@ impl HotObjectCache {
             cache: RwLock::new(lru::LruCache::new(std::num::NonZeroUsize::new(1000).unwrap())),
         }
     }
-    
+
     /// Try to get an object from cache
     async fn get(&self, key: &str) -> Option<Arc<Vec<u8>>> {
         let mut cache = self.cache.write().await;
-        
+
         if let Some(cached) = cache.get(key) {
             cached.hit_count.fetch_add(1, Ordering::Relaxed);
-            
+
             #[cfg(feature = "metrics")]
             {
                 use metrics::counter;
                 counter!("rustfs_object_cache_hits").increment(1);
             }
-            
+
             return Some(Arc::clone(&cached.data));
         }
-        
+
         #[cfg(feature = "metrics")]
         {
             use metrics::counter;
             counter!("rustfs_object_cache_misses").increment(1);
         }
-        
+
         None
     }
-    
+
     /// Put an object into cache if it's small enough
     async fn put(&self, key: String, data: Vec<u8>) {
         let size = data.len();
-        
+
         // Only cache objects smaller than max_object_size
         if size > self.max_object_size {
             return;
         }
-        
+
         let cached_obj = Arc::new(CachedObject {
             data: Arc::new(data),
             cached_at: Instant::now(),
             size,
             hit_count: AtomicUsize::new(0),
         });
-        
+
         let mut cache = self.cache.write().await;
-        
+
         // Evict items if cache is too large
         // Note: We load the current_size inside the write lock to avoid race conditions
         let mut current = self.current_size.load(Ordering::Relaxed);
@@ -254,11 +254,11 @@ impl HotObjectCache {
                 break;
             }
         }
-        
+
         cache.put(key, cached_obj);
         current += size;
         self.current_size.store(current, Ordering::Relaxed);
-        
+
         #[cfg(feature = "metrics")]
         {
             use metrics::{counter, gauge};
@@ -266,14 +266,14 @@ impl HotObjectCache {
             gauge!("rustfs_object_cache_size_bytes").set(self.current_size.load(Ordering::Relaxed) as f64);
         }
     }
-    
+
     /// Clear the cache
     async fn clear(&self) {
         let mut cache = self.cache.write().await;
         cache.clear();
         self.current_size.store(0, Ordering::Relaxed);
     }
-    
+
     /// Get cache statistics
     async fn stats(&self) -> CacheStats {
         let cache = self.cache.read().await;
@@ -319,24 +319,24 @@ impl ConcurrencyManager {
             disk_read_semaphore: Arc::new(Semaphore::new(64)),
         }
     }
-    
+
     /// Start tracking a new GetObject request
     ///
     /// Returns a guard that automatically decrements the counter when dropped
     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 if it's eligible
     pub async fn cache_object(&self, key: String, data: Vec<u8>) {
         self.cache.put(key, data).await
     }
-    
+
     /// Acquire a disk read permit
     ///
     /// This ensures we don't overwhelm the disk with too many concurrent reads
@@ -352,12 +352,12 @@ impl ConcurrencyManager {
             .await
             .expect("Failed to acquire disk read permit: semaphore is closed. This indicates a serious internal error.")
     }
-    
+
     /// Get cache statistics
     pub async fn cache_stats(&self) -> CacheStats {
         self.cache.stats().await
     }
-    
+
     /// Clear the cache
     pub async fn clear_cache(&self) {
         self.cache.clear().await
@@ -378,125 +378,125 @@ pub fn get_concurrency_manager() -> &'static ConcurrencyManager {
 #[cfg(test)]
 mod tests {
     use super::*;
-    
+
     #[test]
     fn test_concurrent_request_tracking() {
         assert_eq!(GetObjectGuard::concurrent_requests(), 0);
-        
+
         let _guard1 = GetObjectGuard::new();
         assert_eq!(GetObjectGuard::concurrent_requests(), 1);
-        
+
         let _guard2 = GetObjectGuard::new();
         assert_eq!(GetObjectGuard::concurrent_requests(), 2);
-        
+
         drop(_guard1);
         assert_eq!(GetObjectGuard::concurrent_requests(), 1);
-        
+
         drop(_guard2);
         assert_eq!(GetObjectGuard::concurrent_requests(), 0);
     }
-    
+
     #[test]
     fn test_adaptive_buffer_sizing() {
         // Reset counter
         ACTIVE_GET_REQUESTS.store(0, Ordering::Relaxed);
-        
+
         let base_size = 256 * KI_B;
-        
+
         // Low concurrency: use base size
         ACTIVE_GET_REQUESTS.store(1, Ordering::Relaxed);
         assert_eq!(get_concurrency_aware_buffer_size(10 * MI_B as i64, base_size), base_size);
-        
+
         // Medium concurrency: reduce to 75%
         ACTIVE_GET_REQUESTS.store(3, Ordering::Relaxed);
         let result = get_concurrency_aware_buffer_size(10 * MI_B as i64, base_size);
         assert!(result < base_size);
         assert!(result >= base_size / 2);
-        
+
         // High concurrency: reduce to 50%
         ACTIVE_GET_REQUESTS.store(6, Ordering::Relaxed);
         let result = get_concurrency_aware_buffer_size(10 * MI_B as i64, base_size);
         assert!(result <= base_size / 2);
         assert!(result >= base_size / 3);
-        
+
         // Very high concurrency: reduce to 40%
         ACTIVE_GET_REQUESTS.store(10, Ordering::Relaxed);
         let result = get_concurrency_aware_buffer_size(10 * MI_B as i64, base_size);
         assert!(result <= base_size / 2);
         assert!(result >= 64 * KI_B); // Should stay above minimum
-        
+
         // Reset for other tests
         ACTIVE_GET_REQUESTS.store(0, Ordering::Relaxed);
     }
-    
+
     #[tokio::test]
     async fn test_hot_object_cache() {
         let cache = HotObjectCache::new();
-        
+
         // Cache a small object
         let data = vec![1u8; 1024];
         cache.put("test-key".to_string(), data.clone()).await;
-        
+
         // Retrieve it
         let cached = cache.get("test-key").await;
         assert!(cached.is_some());
         assert_eq!(*cached.unwrap(), data);
-        
+
         // Try to get non-existent key
         let missing = cache.get("missing-key").await;
         assert!(missing.is_none());
-        
+
         // Cache too large object (> 10MB)
         let large_data = vec![1u8; 11 * MI_B];
         cache.put("large-key".to_string(), large_data).await;
-        
+
         // Should not be cached
         let not_cached = cache.get("large-key").await;
         assert!(not_cached.is_none());
     }
-    
+
     #[tokio::test]
     async fn test_cache_eviction() {
         let cache = HotObjectCache::new();
-        
+
         // Fill cache with small objects
         for i in 0..20 {
             let data = vec![1u8; 6 * MI_B]; // 6MB each
             cache.put(format!("key-{}", i), data).await;
         }
-        
+
         // Check that old items were evicted
         let stats = cache.stats().await;
         assert!(stats.size <= cache.max_cache_size);
-        
+
         // First keys should be evicted
         let first = cache.get("key-0").await;
         assert!(first.is_none());
-        
+
         // Recent keys should still be there
         let recent = cache.get("key-19").await;
         assert!(recent.is_some());
     }
-    
+
     #[test]
     fn test_concurrency_manager_creation() {
         let manager = ConcurrencyManager::new();
         assert_eq!(manager.disk_read_semaphore.available_permits(), 64);
     }
-    
+
     #[tokio::test]
     async fn test_disk_read_permits() {
         let manager = ConcurrencyManager::new();
-        
+
         let permit1 = manager.acquire_disk_read_permit().await;
         assert_eq!(manager.disk_read_semaphore.available_permits(), 63);
-        
+
         let permit2 = manager.acquire_disk_read_permit().await;
         assert_eq!(manager.disk_read_semaphore.available_permits(), 62);
-        
+
         drop(permit1);
         assert_eq!(manager.disk_read_semaphore.available_permits(), 63);
-        
+
         drop(permit2);
         assert_eq!(manager.disk_read_semaphore.available_permits(), 64);
     }

rustfs/src/storage/concurrent_get_object_test.rs

@@ -0,0 +1,276 @@
+// 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 GetObject performance optimization
+
+#[cfg(test)]
+mod tests {
+    use crate::storage::concurrency::{ConcurrencyManager, GetObjectGuard, get_concurrency_aware_buffer_size};
+    use rustfs_config::MI_B;
+    use std::time::Duration;
+    use tokio::time::Instant;
+
+    /// Test that concurrent requests are tracked correctly
+    #[tokio::test]
+    async fn test_concurrent_request_tracking() {
+        // Start with no active requests
+        let initial = GetObjectGuard::concurrent_requests();
+
+        // Create guards to simulate concurrent requests
+        let guard1 = ConcurrencyManager::track_request();
+        assert_eq!(GetObjectGuard::concurrent_requests(), initial + 1);
+
+        let guard2 = ConcurrencyManager::track_request();
+        assert_eq!(GetObjectGuard::concurrent_requests(), initial + 2);
+
+        let guard3 = ConcurrencyManager::track_request();
+        assert_eq!(GetObjectGuard::concurrent_requests(), initial + 3);
+
+        // Drop guards and verify count decreases
+        drop(guard1);
+        tokio::time::sleep(Duration::from_millis(10)).await;
+        assert_eq!(GetObjectGuard::concurrent_requests(), initial + 2);
+
+        drop(guard2);
+        tokio::time::sleep(Duration::from_millis(10)).await;
+        assert_eq!(GetObjectGuard::concurrent_requests(), initial + 1);
+
+        drop(guard3);
+        tokio::time::sleep(Duration::from_millis(10)).await;
+        assert_eq!(GetObjectGuard::concurrent_requests(), initial);
+    }
+
+    /// Test adaptive buffer sizing under different concurrency levels
+    #[tokio::test]
+    async fn test_adaptive_buffer_sizing() {
+        let file_size = 32 * MI_B as i64;
+        let base_buffer = 256 * 1024; // 256KB base
+
+        // Simulate different concurrency levels
+        let test_cases = vec![
+            (1, 1.0, "Very low concurrency: should use full buffer"),
+            (2, 1.0, "Low concurrency: should use full buffer"),
+            (3, 0.75, "Medium concurrency: should reduce to 75%"),
+            (6, 0.5, "High concurrency: should reduce to 50%"),
+            (10, 0.4, "Very high concurrency: should reduce to 40%"),
+        ];
+
+        for (concurrent_requests, expected_multiplier, description) in test_cases {
+            // Simulate concurrent requests
+            let mut guards = Vec::new();
+            for _ in 0..concurrent_requests {
+                guards.push(ConcurrencyManager::track_request());
+            }
+
+            tokio::time::sleep(Duration::from_millis(10)).await;
+
+            let buffer_size = get_concurrency_aware_buffer_size(file_size, base_buffer);
+            let actual_multiplier = buffer_size as f64 / base_buffer as f64;
+
+            println!(
+                "{}: {} requests, buffer {} bytes, multiplier {:.2}",
+                description, concurrent_requests, buffer_size, actual_multiplier
+            );
+
+            // Allow some tolerance for rounding
+            assert!(
+                (actual_multiplier - expected_multiplier).abs() < 0.15,
+                "{} - Expected multiplier {:.2}, got {:.2}",
+                description,
+                expected_multiplier,
+                actual_multiplier
+            );
+
+            // Cleanup
+            drop(guards);
+            tokio::time::sleep(Duration::from_millis(10)).await;
+        }
+    }
+
+    /// Test that buffer size stays within reasonable bounds
+    #[tokio::test]
+    async fn test_buffer_size_bounds() {
+        let base_buffer = 512 * 1024; // 512KB
+
+        // Test with extreme concurrency
+        let mut guards = Vec::new();
+        for _ in 0..100 {
+            guards.push(ConcurrencyManager::track_request());
+        }
+
+        tokio::time::sleep(Duration::from_millis(10)).await;
+
+        let buffer_size = get_concurrency_aware_buffer_size(10 * MI_B as i64, base_buffer);
+
+        // Should not go below 64KB
+        assert!(buffer_size >= 64 * 1024, "Buffer size too small: {}", buffer_size);
+
+        // Should not exceed 1MB for high concurrency
+        assert!(buffer_size <= MI_B, "Buffer size too large: {}", buffer_size);
+
+        drop(guards);
+    }
+
+    /// Benchmark concurrent request handling
+    #[tokio::test]
+    async fn bench_concurrent_requests() {
+        let concurrency_levels = vec![1, 2, 4, 8, 16];
+
+        for concurrency in concurrency_levels {
+            let start = Instant::now();
+            let mut handles = Vec::new();
+
+            for _ in 0..concurrency {
+                let handle = tokio::spawn(async {
+                    let _guard = ConcurrencyManager::track_request();
+
+                    // Simulate some work (e.g., reading a file)
+                    tokio::time::sleep(Duration::from_millis(10)).await;
+
+                    _guard.elapsed()
+                });
+
+                handles.push(handle);
+            }
+
+            // Wait for all to complete
+            let mut durations = Vec::new();
+            for handle in handles {
+                if let Ok(duration) = handle.await {
+                    durations.push(duration);
+                }
+            }
+
+            let total_elapsed = start.elapsed();
+            let avg_duration = durations.iter().sum::<Duration>() / durations.len() as u32;
+
+            println!(
+                "Concurrency {}: total={}ms, avg={}ms, max={}ms",
+                concurrency,
+                total_elapsed.as_millis(),
+                avg_duration.as_millis(),
+                durations.iter().max().unwrap().as_millis()
+            );
+        }
+    }
+
+    /// Test disk I/O permit acquisition
+    #[tokio::test]
+    async fn test_disk_io_permits() {
+        let manager = ConcurrencyManager::new();
+
+        // Acquire multiple permits
+        let permit1 = manager.acquire_disk_read_permit().await;
+        let permit2 = manager.acquire_disk_read_permit().await;
+
+        // Drop permits
+        drop(permit1);
+        drop(permit2);
+
+        // Should be able to acquire again
+        let _permit3 = manager.acquire_disk_read_permit().await;
+    }
+
+    /// Test cache behavior with manager
+    #[tokio::test]
+    async fn test_cache_operations() {
+        let manager = ConcurrencyManager::new();
+
+        // Initially empty cache
+        let stats = manager.cache_stats().await;
+        assert_eq!(stats.entries, 0);
+        assert_eq!(stats.size, 0);
+
+        // Cache a small object
+        let key = "test/object1".to_string();
+        let data = vec![1u8; 1024 * 1024]; // 1MB
+        manager.cache_object(key.clone(), data.clone()).await;
+
+        // Verify it was cached
+        let cached = manager.get_cached(&key).await;
+        assert!(cached.is_some());
+        assert_eq!(*cached.unwrap(), data);
+
+        // Verify stats updated
+        let stats = manager.cache_stats().await;
+        assert_eq!(stats.entries, 1);
+        assert!(stats.size >= data.len());
+
+        // Try to get non-existent key
+        let missing = manager.get_cached("missing/key").await;
+        assert!(missing.is_none());
+
+        // Clear cache
+        manager.clear_cache().await;
+        let stats = manager.cache_stats().await;
+        assert_eq!(stats.entries, 0);
+        assert_eq!(stats.size, 0);
+    }
+
+    /// Test that large objects are not cached
+    #[tokio::test]
+    async fn test_large_object_not_cached() {
+        let manager = ConcurrencyManager::new();
+
+        // Try to cache a large object (> 10MB)
+        let key = "test/large".to_string();
+        let large_data = vec![1u8; 15 * MI_B]; // 15MB
+
+        manager.cache_object(key.clone(), large_data).await;
+
+        // Should not be cached
+        let cached = manager.get_cached(&key).await;
+        assert!(cached.is_none());
+
+        // Cache stats should still be empty
+        let stats = manager.cache_stats().await;
+        assert_eq!(stats.entries, 0);
+    }
+
+    /// Test cache eviction under memory pressure
+    #[tokio::test]
+    async fn test_cache_eviction() {
+        let manager = ConcurrencyManager::new();
+
+        // Cache multiple objects until we exceed the limit
+        let object_size = 6 * MI_B; // 6MB each
+        let num_objects = 20; // Total 120MB > 100MB limit
+
+        for i in 0..num_objects {
+            let key = format!("test/object{}", i);
+            let data = vec![1u8; object_size];
+            manager.cache_object(key, data).await;
+        }
+
+        // Verify cache size is within limit
+        let stats = manager.cache_stats().await;
+        assert!(stats.size <= stats.max_size, "Cache size {} exceeded max {}", stats.size, stats.max_size);
+
+        // Some objects should have been evicted
+        assert!(
+            stats.entries < num_objects,
+            "Expected eviction, but all {} objects are still cached",
+            stats.entries
+        );
+
+        // First objects should be evicted (LRU)
+        let first = manager.get_cached("test/object0").await;
+        assert!(first.is_none(), "First object should have been evicted");
+
+        // Recent objects should still be there
+        let recent_key = format!("test/object{}", num_objects - 1);
+        let recent = manager.get_cached(&recent_key).await;
+        assert!(recent.is_some(), "Recent object should still be cached");
+    }
+}

rustfs/src/storage/ecfs.rs

@@ -17,7 +17,9 @@ use crate::config::workload_profiles::{
     RustFSBufferConfig, WorkloadProfile, get_global_buffer_config, is_buffer_profile_enabled,
 };
 use crate::error::ApiError;
-use crate::storage::concurrency::{ConcurrencyManager, GetObjectGuard, get_concurrency_aware_buffer_size, get_concurrency_manager};
+use crate::storage::concurrency::{
+    ConcurrencyManager, GetObjectGuard, get_concurrency_aware_buffer_size, get_concurrency_manager,
+};
 use crate::storage::entity;
 use crate::storage::helper::OperationHelper;
 use crate::storage::options::{filter_object_metadata, get_content_sha256};
@@ -65,7 +67,7 @@ use rustfs_ecstore::{
     disk::{error::DiskError, error_reduce::is_all_buckets_not_found},
     error::{StorageError, is_err_bucket_not_found, is_err_object_not_found, is_err_version_not_found},
     new_object_layer_fn,
-    set_disk::{DEFAULT_READ_BUFFER_SIZE, MAX_PARTS_COUNT, is_valid_storage_class},
+    set_disk::{MAX_PARTS_COUNT, is_valid_storage_class},
     store_api::{
         BucketOptions,
         CompletePart,
@@ -1614,12 +1616,9 @@ impl S3 for FS {
         // Track this request for concurrency-aware optimizations
         let _request_guard = ConcurrencyManager::track_request();
         let concurrent_requests = GetObjectGuard::concurrent_requests();
-        
-        debug!(
-            "GetObject request started with {} concurrent requests",
-            concurrent_requests
-        );
-        
+
+        debug!("GetObject request started with {} concurrent requests", concurrent_requests);
+
         let mut helper = OperationHelper::new(&req, EventName::ObjectAccessedGet, "s3:GetObject");
         // mc get 3
 
@@ -1639,24 +1638,24 @@ impl S3 for FS {
         // Try to get from cache for small, frequently accessed objects
         let manager = get_concurrency_manager();
         let cache_key = format!("{}/{}", bucket, key);
-        
+
         // Only attempt cache lookup for objects without range/part requests
         if part_number.is_none() && range.is_none() {
             if let Some(cached_data) = manager.get_cached(&cache_key).await {
                 debug!("Serving object from cache: {}", cache_key);
-                
+
                 // Build response from cached data
                 let body = Some(StreamingBlob::wrap(futures::stream::once(async move {
                     Ok(bytes::Bytes::from((*cached_data).clone()))
                 })));
-                
+
                 let output = GetObjectOutput {
                     body,
                     content_length: Some(cached_data.len() as i64),
                     accept_ranges: Some("bytes".to_string()),
                     ..Default::default()
                 };
-                
+
                 return Ok(S3Response::new(output));
             }
         }
@@ -1700,7 +1699,7 @@ impl S3 for FS {
 
         // Acquire disk read permit to prevent I/O saturation under high concurrency
         let _disk_permit = manager.acquire_disk_read_permit().await;
-        
+
         let reader = store
             .get_object_reader(bucket.as_str(), key.as_str(), rs.clone(), h, &opts)
             .await
@@ -1933,19 +1932,19 @@ impl S3 for FS {
         // This adapts based on the number of concurrent GetObject requests
         let base_buffer_size = get_buffer_size_opt_in(response_content_length);
         let optimal_buffer_size = get_concurrency_aware_buffer_size(response_content_length, base_buffer_size);
-        
+
         debug!(
             "GetObject buffer sizing: file_size={}, base={}, optimal={}, concurrent_requests={}",
-            response_content_length,
-            base_buffer_size,
-            optimal_buffer_size,
-            concurrent_requests
+            response_content_length, base_buffer_size, optimal_buffer_size, concurrent_requests
         );
-        
+
         // For SSE-C encrypted objects, don't use bytes_stream to limit the stream
         // because DecryptReader needs to read all encrypted data to produce decrypted output
         let body = if stored_sse_algorithm.is_some() || managed_encryption_applied {
-            info!("Managed SSE: Using unlimited stream for decryption with buffer size {}", optimal_buffer_size);
+            info!(
+                "Managed SSE: Using unlimited stream for decryption with buffer size {}",
+                optimal_buffer_size
+            );
             Some(StreamingBlob::wrap(ReaderStream::with_capacity(final_stream, optimal_buffer_size)))
         } else {
             Some(StreamingBlob::wrap(bytes_stream(
@@ -1953,23 +1952,24 @@ impl S3 for FS {
                 response_content_length as usize,
             )))
         };
-        
+
         // TODO: Implement proper streaming cache for small objects
         // For small objects (<= 10MB) with no range/part request, we could cache for future requests.
         // However, this requires refactoring to capture the stream data while serving the response.
         // Current implementation only provides cache lookup (lines 1644-1662), not cache insertion.
-        // 
+        //
         // Potential approaches:
         // 1. Use a TeeReader to duplicate the stream - one copy to response, one to cache
         // 2. Pre-load small objects into memory before streaming (impacts first request latency)
         // 3. Background task to cache after response completes (requires stream copy)
         //
         // For now, cache can be manually populated via admin API or future background processes.
-        if response_content_length <= 10 * 1024 * 1024 
-            && part_number.is_none() 
-            && rs.is_none() 
-            && !managed_encryption_applied 
-            && stored_sse_algorithm.is_none() {
+        if response_content_length <= 10 * 1024 * 1024
+            && part_number.is_none()
+            && rs.is_none()
+            && !managed_encryption_applied
+            && stored_sse_algorithm.is_none()
+        {
             debug!(
                 "Object {} is eligible for caching (size: {} bytes) but streaming cache not yet implemented",
                 cache_key, response_content_length
@@ -5221,6 +5221,7 @@ pub(crate) async fn has_replication_rules(bucket: &str, objects: &[ObjectToDelet
 mod tests {
     use super::*;
     use rustfs_config::MI_B;
+    use rustfs_ecstore::set_disk::DEFAULT_READ_BUFFER_SIZE;
 
     #[test]
     fn test_fs_creation() {

rustfs/src/storage/mod.rs

@@ -19,3 +19,5 @@ pub(crate) mod entity;
 pub(crate) mod helper;
 pub mod options;
 pub mod tonic_service;
+
+mod concurrent_get_object_test;

rustfs/tests/concurrent_get_object_test.rs

@@ -1,267 +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.
-
-//! Integration tests for concurrent GetObject performance optimization
-
-use rustfs::storage::concurrency::{ConcurrencyManager, GetObjectGuard, get_concurrency_aware_buffer_size};
-use rustfs_config::MI_B;
-use std::time::Duration;
-use tokio::time::Instant;
-
-/// Test that concurrent requests are tracked correctly
-#[tokio::test]
-async fn test_concurrent_request_tracking() {
-    // Start with no active requests
-    let initial = GetObjectGuard::concurrent_requests();
-    
-    // Create guards to simulate concurrent requests
-    let guard1 = ConcurrencyManager::track_request();
-    assert_eq!(GetObjectGuard::concurrent_requests(), initial + 1);
-    
-    let guard2 = ConcurrencyManager::track_request();
-    assert_eq!(GetObjectGuard::concurrent_requests(), initial + 2);
-    
-    let guard3 = ConcurrencyManager::track_request();
-    assert_eq!(GetObjectGuard::concurrent_requests(), initial + 3);
-    
-    // Drop guards and verify count decreases
-    drop(guard1);
-    tokio::time::sleep(Duration::from_millis(10)).await;
-    assert_eq!(GetObjectGuard::concurrent_requests(), initial + 2);
-    
-    drop(guard2);
-    tokio::time::sleep(Duration::from_millis(10)).await;
-    assert_eq!(GetObjectGuard::concurrent_requests(), initial + 1);
-    
-    drop(guard3);
-    tokio::time::sleep(Duration::from_millis(10)).await;
-    assert_eq!(GetObjectGuard::concurrent_requests(), initial);
-}
-
-/// Test adaptive buffer sizing under different concurrency levels
-#[tokio::test]
-async fn test_adaptive_buffer_sizing() {
-    let file_size = 32 * MI_B as i64;
-    let base_buffer = 256 * 1024; // 256KB base
-    
-    // Simulate different concurrency levels
-    let test_cases = vec![
-        (1, 1.0, "Very low concurrency: should use full buffer"),
-        (2, 1.0, "Low concurrency: should use full buffer"),
-        (3, 0.75, "Medium concurrency: should reduce to 75%"),
-        (6, 0.5, "High concurrency: should reduce to 50%"),
-        (10, 0.4, "Very high concurrency: should reduce to 40%"),
-    ];
-    
-    for (concurrent_requests, expected_multiplier, description) in test_cases {
-        // Simulate concurrent requests
-        let mut guards = Vec::new();
-        for _ in 0..concurrent_requests {
-            guards.push(ConcurrencyManager::track_request());
-        }
-        
-        tokio::time::sleep(Duration::from_millis(10)).await;
-        
-        let buffer_size = get_concurrency_aware_buffer_size(file_size, base_buffer);
-        let actual_multiplier = buffer_size as f64 / base_buffer as f64;
-        
-        println!("{}: {} requests, buffer {} bytes, multiplier {:.2}",
-                 description, concurrent_requests, buffer_size, actual_multiplier);
-        
-        // Allow some tolerance for rounding
-        assert!(
-            (actual_multiplier - expected_multiplier).abs() < 0.15,
-            "{} - Expected multiplier {:.2}, got {:.2}",
-            description, expected_multiplier, actual_multiplier
-        );
-        
-        // Cleanup
-        drop(guards);
-        tokio::time::sleep(Duration::from_millis(10)).await;
-    }
-}
-
-/// Test that buffer size stays within reasonable bounds
-#[tokio::test]
-async fn test_buffer_size_bounds() {
-    let base_buffer = 512 * 1024; // 512KB
-    
-    // Test with extreme concurrency
-    let mut guards = Vec::new();
-    for _ in 0..100 {
-        guards.push(ConcurrencyManager::track_request());
-    }
-    
-    tokio::time::sleep(Duration::from_millis(10)).await;
-    
-    let buffer_size = get_concurrency_aware_buffer_size(10 * MI_B as i64, base_buffer);
-    
-    // Should not go below 64KB
-    assert!(buffer_size >= 64 * 1024, "Buffer size too small: {}", buffer_size);
-    
-    // Should not exceed 1MB for high concurrency
-    assert!(buffer_size <= MI_B, "Buffer size too large: {}", buffer_size);
-    
-    drop(guards);
-}
-
-/// Benchmark concurrent request handling
-#[tokio::test]
-async fn bench_concurrent_requests() {
-    let concurrency_levels = vec![1, 2, 4, 8, 16];
-    
-    for concurrency in concurrency_levels {
-        let start = Instant::now();
-        let mut handles = Vec::new();
-        
-        for _ in 0..concurrency {
-            let handle = tokio::spawn(async {
-                let _guard = ConcurrencyManager::track_request();
-                
-                // Simulate some work (e.g., reading a file)
-                tokio::time::sleep(Duration::from_millis(10)).await;
-                
-                _guard.elapsed()
-            });
-            
-            handles.push(handle);
-        }
-        
-        // Wait for all to complete
-        let mut durations = Vec::new();
-        for handle in handles {
-            if let Ok(duration) = handle.await {
-                durations.push(duration);
-            }
-        }
-        
-        let total_elapsed = start.elapsed();
-        let avg_duration = durations.iter().sum::<Duration>() / durations.len() as u32;
-        
-        println!(
-            "Concurrency {}: total={}ms, avg={}ms, max={}ms",
-            concurrency,
-            total_elapsed.as_millis(),
-            avg_duration.as_millis(),
-            durations.iter().max().unwrap().as_millis()
-        );
-    }
-}
-
-/// Test disk I/O permit acquisition
-#[tokio::test]
-async fn test_disk_io_permits() {
-    let manager = ConcurrencyManager::new();
-    
-    // Acquire multiple permits
-    let permit1 = manager.acquire_disk_read_permit().await;
-    let permit2 = manager.acquire_disk_read_permit().await;
-    
-    // Drop permits
-    drop(permit1);
-    drop(permit2);
-    
-    // Should be able to acquire again
-    let _permit3 = manager.acquire_disk_read_permit().await;
-}
-
-/// Test cache behavior with manager
-#[tokio::test]
-async fn test_cache_operations() {
-    let manager = ConcurrencyManager::new();
-    
-    // Initially empty cache
-    let stats = manager.cache_stats().await;
-    assert_eq!(stats.entries, 0);
-    assert_eq!(stats.size, 0);
-    
-    // Cache a small object
-    let key = "test/object1".to_string();
-    let data = vec![1u8; 1024 * 1024]; // 1MB
-    manager.cache_object(key.clone(), data.clone()).await;
-    
-    // Verify it was cached
-    let cached = manager.get_cached(&key).await;
-    assert!(cached.is_some());
-    assert_eq!(*cached.unwrap(), data);
-    
-    // Verify stats updated
-    let stats = manager.cache_stats().await;
-    assert_eq!(stats.entries, 1);
-    assert!(stats.size >= data.len());
-    
-    // Try to get non-existent key
-    let missing = manager.get_cached("missing/key").await;
-    assert!(missing.is_none());
-    
-    // Clear cache
-    manager.clear_cache().await;
-    let stats = manager.cache_stats().await;
-    assert_eq!(stats.entries, 0);
-    assert_eq!(stats.size, 0);
-}
-
-/// Test that large objects are not cached
-#[tokio::test]
-async fn test_large_object_not_cached() {
-    let manager = ConcurrencyManager::new();
-    
-    // Try to cache a large object (> 10MB)
-    let key = "test/large".to_string();
-    let large_data = vec![1u8; 15 * MI_B]; // 15MB
-    
-    manager.cache_object(key.clone(), large_data).await;
-    
-    // Should not be cached
-    let cached = manager.get_cached(&key).await;
-    assert!(cached.is_none());
-    
-    // Cache stats should still be empty
-    let stats = manager.cache_stats().await;
-    assert_eq!(stats.entries, 0);
-}
-
-/// Test cache eviction under memory pressure
-#[tokio::test]
-async fn test_cache_eviction() {
-    let manager = ConcurrencyManager::new();
-    
-    // Cache multiple objects until we exceed the limit
-    let object_size = 6 * MI_B; // 6MB each
-    let num_objects = 20; // Total 120MB > 100MB limit
-    
-    for i in 0..num_objects {
-        let key = format!("test/object{}", i);
-        let data = vec![1u8; object_size];
-        manager.cache_object(key, data).await;
-    }
-    
-    // Verify cache size is within limit
-    let stats = manager.cache_stats().await;
-    assert!(stats.size <= stats.max_size, 
-            "Cache size {} exceeded max {}", stats.size, stats.max_size);
-    
-    // Some objects should have been evicted
-    assert!(stats.entries < num_objects,
-            "Expected eviction, but all {} objects are still cached", stats.entries);
-    
-    // First objects should be evicted (LRU)
-    let first = manager.get_cached("test/object0").await;
-    assert!(first.is_none(), "First object should have been evicted");
-    
-    // Recent objects should still be there
-    let recent_key = format!("test/object{}", num_objects - 1);
-    let recent = manager.get_cached(&recent_key).await;
-    assert!(recent.is_some(), "Recent object should still be cached");
-}