fix: address code review issues in concurrency module

- Fix race condition in cache size tracking by using consistent atomic operations within lock
- Correct buffer sizing logic: 1-2 requests use 100%, 3-4 use 75%, 5-8 use 50%, >8 use 40%
- Improve error message for semaphore acquire failure
- Document limitation of streaming cache implementation (not yet implemented)
- Add TODO for proper streaming cache with suggested approaches
- Update tests to match corrected buffer sizing thresholds

Co-authored-by: houseme <4829346+houseme@users.noreply.github.com>

rustfs/src/storage/concurrency.rs

@@ -127,14 +127,17 @@ pub fn get_concurrency_aware_buffer_size(file_size: i64, base_buffer_size: usize
     }
     
     // Calculate adaptive multiplier based on concurrency level
-    let adaptive_multiplier = if concurrent_requests <= MEDIUM_CONCURRENCY_THRESHOLD {
-        // Medium concurrency: slightly reduce buffer size (75% of base)
+    let adaptive_multiplier = if concurrent_requests <= 2 {
+        // Low concurrency (1-2): use full buffer for maximum throughput
+        1.0
+    } else if concurrent_requests <= MEDIUM_CONCURRENCY_THRESHOLD {
+        // Medium concurrency (3-4): slightly reduce buffer size (75% of base)
         0.75
     } else if concurrent_requests <= HIGH_CONCURRENCY_THRESHOLD {
-        // Higher concurrency: more aggressive reduction (50% of base)
+        // Higher concurrency (5-8): more aggressive reduction (50% of base)
         0.5
     } else {
-        // Very high concurrency: minimize memory per request (40% of base)
+        // Very high concurrency (>8): minimize memory per request (40% of base)
         0.4
     };
     
@@ -241,16 +244,20 @@ impl HotObjectCache {
         let mut cache = self.cache.write().await;
         
         // Evict items if cache is too large
-        while self.current_size.load(Ordering::Relaxed) + size > self.max_cache_size {
+        // Note: We load the current_size inside the write lock to avoid race conditions
+        let mut current = self.current_size.load(Ordering::Relaxed);
+        while current + size > self.max_cache_size {
             if let Some((_, evicted)) = cache.pop_lru() {
-                self.current_size.fetch_sub(evicted.size, Ordering::Relaxed);
+                current -= evicted.size;
+                self.current_size.store(current, Ordering::Relaxed);
             } else {
                 break;
             }
         }
         
         cache.put(key, cached_obj);
-        self.current_size.fetch_add(size, Ordering::Relaxed);
+        current += size;
+        self.current_size.store(current, Ordering::Relaxed);
         
         #[cfg(feature = "metrics")]
         {
@@ -333,8 +340,17 @@ impl ConcurrencyManager {
     /// Acquire a disk read permit
     ///
     /// This ensures we don't overwhelm the disk with too many concurrent reads
+    ///
+    /// # Panics
+    ///
+    /// This function will panic if the semaphore has been closed, which should never
+    /// happen in normal operation since the semaphore is owned by the ConcurrencyManager
+    /// which lives for the entire application lifetime.
     pub async fn acquire_disk_read_permit(&self) -> tokio::sync::SemaphorePermit<'_> {
-        self.disk_read_semaphore.acquire().await.expect("Semaphore closed")
+        self.disk_read_semaphore
+            .acquire()
+            .await
+            .expect("Failed to acquire disk read permit: semaphore is closed. This indicates a serious internal error.")
     }
     
     /// Get cache statistics

rustfs/src/storage/ecfs.rs

@@ -1954,16 +1954,26 @@ impl S3 for FS {
             )))
         };
         
-        // For small objects (<= 10MB) with no range/part request, cache for future requests
+        // 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() {
-            // Note: In production, we'd want to stream the data and cache it
-            // For now, we'll just note the intention. Actual implementation would
-            // require refactoring to capture the stream data.
-            debug!("Object {} is eligible for caching (size: {} bytes)", cache_key, response_content_length);
+            debug!(
+                "Object {} is eligible for caching (size: {} bytes) but streaming cache not yet implemented",
+                cache_key, response_content_length
+            );
         }
 
         // Extract SSE information from metadata for response

rustfs/tests/concurrent_get_object_test.rs

@@ -57,7 +57,8 @@ async fn test_adaptive_buffer_sizing() {
     
     // Simulate different concurrency levels
     let test_cases = vec![
-        (1, 1.0, "Low concurrency: should use full buffer"),
+        (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%"),