diff --git a/Cargo.lock b/Cargo.lock index a5cb9452f..257052b4e 100644 --- a/Cargo.lock +++ b/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", diff --git a/Cargo.toml b/Cargo.toml index 2c8158e40..fc9c2ab80 100644 --- a/Cargo.toml +++ b/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" diff --git a/rustfs/Cargo.toml b/rustfs/Cargo.toml index ec377ec2c..2a58125a8 100644 --- a/rustfs/Cargo.toml +++ b/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 } diff --git a/rustfs/src/storage/concurrency.rs b/rustfs/src/storage/concurrency.rs index 21e0e300b..418d26f03 100644 --- a/rustfs/src/storage/concurrency.rs +++ b/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>, @@ -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>> { 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) { 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>> { self.cache.get(key).await } - + /// Cache an object if it's eligible pub async fn cache_object(&self, key: String, data: Vec) { 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); } diff --git a/rustfs/src/storage/concurrent_get_object_test.rs b/rustfs/src/storage/concurrent_get_object_test.rs new file mode 100644 index 000000000..ae6ccfb87 --- /dev/null +++ b/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::() / 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"); + } +} diff --git a/rustfs/src/storage/ecfs.rs b/rustfs/src/storage/ecfs.rs index fda7aae8a..9279a72d8 100644 --- a/rustfs/src/storage/ecfs.rs +++ b/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() { diff --git a/rustfs/src/storage/mod.rs b/rustfs/src/storage/mod.rs index a81ce9226..a4c8d2dba 100644 --- a/rustfs/src/storage/mod.rs +++ b/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; diff --git a/rustfs/tests/concurrent_get_object_test.rs b/rustfs/tests/concurrent_get_object_test.rs deleted file mode 100644 index 69fedf8b1..000000000 --- a/rustfs/tests/concurrent_get_object_test.rs +++ /dev/null @@ -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::() / 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"); -}