diff --git a/rustfs/src/storage/concurrency.rs b/rustfs/src/storage/concurrency.rs index 5e24f2aca..21e0e300b 100644 --- a/rustfs/src/storage/concurrency.rs +++ b/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 diff --git a/rustfs/src/storage/ecfs.rs b/rustfs/src/storage/ecfs.rs index bd89ad0c1..96d36e7bc 100644 --- a/rustfs/src/storage/ecfs.rs +++ b/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 diff --git a/rustfs/tests/concurrent_get_object_test.rs b/rustfs/tests/concurrent_get_object_test.rs index fdf2ea97a..69fedf8b1 100644 --- a/rustfs/tests/concurrent_get_object_test.rs +++ b/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%"),