diff --git a/crates/io-metrics/src/lib.rs b/crates/io-metrics/src/lib.rs index 0768f162f..b6418d5eb 100644 --- a/crates/io-metrics/src/lib.rs +++ b/crates/io-metrics/src/lib.rs @@ -261,6 +261,16 @@ pub fn record_get_object_completion(total_duration_secs: f64, response_size_byte histogram!("rustfs_io_get_object_buffer_size_bytes").record(buffer_size_bytes as f64); } +/// Record the streaming strategy chosen for a GetObject response body. +#[inline(always)] +pub fn record_get_object_stream_strategy(strategy: &str, buffer_size_bytes: usize, response_size_bytes: i64) { + counter!("rustfs_io_get_object_stream_strategy_total", "strategy" => strategy.to_string()).increment(1); + histogram!("rustfs_io_get_object_stream_buffer_size_bytes", "strategy" => strategy.to_string()) + .record(buffer_size_bytes as f64); + histogram!("rustfs_io_get_object_stream_response_size_bytes", "strategy" => strategy.to_string()) + .record(response_size_bytes.max(0) as f64); +} + /// Record I/O queue congestion observation. #[inline(always)] pub fn record_io_queue_congestion() { diff --git a/docs/operations/issue-713-gt1g-get-baseline-v1-summary-zh.md b/docs/operations/issue-713-gt1g-get-baseline-v1-summary-zh.md new file mode 100644 index 000000000..cd575a324 --- /dev/null +++ b/docs/operations/issue-713-gt1g-get-baseline-v1-summary-zh.md @@ -0,0 +1,87 @@ +# Issue #713 `>1GiB GET` baseline v1 小结 + +## 1. 执行范围 + +本轮是 `#713` 的第一轮 plain baseline，先不碰 encrypted / compressed，只看： + +1. `1GiB` +2. `2GiB` +3. `sequential` +4. `ranged_parallel` +5. `concurrency=1,4,8` +6. `range_workers=4` +7. `rounds=1` + +结果目录： + +1. `rustfs/target/bench/issue713-gt1g-get-baseline-v1-20260625T105149Z` + +## 2. 结果总表 + +### 2.1 `1GiB` + +`sequential` + +1. `c1`: `99.36 MiB/s`, `10807.0ms` +2. `c4`: `518.53 MiB/s`, `8267.5ms` +3. `c8`: `708.04 MiB/s`, `11362.4ms` + +`ranged_parallel` + +1. `c1`: `481.71 MiB/s`, `2229.0ms` +2. `c4`: `1536.66 MiB/s`, `2728.5ms` +3. `c8`: `1548.29 MiB/s`, `5010.9ms` + +### 2.2 `2GiB` + +`sequential` + +1. `c1`: `116.43 MiB/s`, `18444.0ms` +2. `c4`: `522.06 MiB/s`, `16442.5ms` +3. `c8`: `712.24 MiB/s`, `21842.1ms` + +`ranged_parallel` + +1. `c1`: `503.28 MiB/s`, `4267.0ms` +2. `c4`: `1817.59 MiB/s`, `4707.3ms` +3. `c8`: `1681.50 MiB/s`, `9502.1ms` + +## 3. 当前第一结论 + +本轮 plain baseline 可以先收敛出三点： + +1. `ranged_parallel` 在当前单机多盘场景下，明显优于单流 `sequential` +2. 这种优势在 `1GiB` 和 `2GiB` 两个点上都成立 +3. 即使 `sequential` 已经过了当前这轮 `>1GiB GET` buffer / readahead 优化，当前仍然没有追平 `4-way ranged_parallel` + +换句话说： + +1. 这轮服务端优化是必要的 +2. 但对当前这组 plain `>1GiB GET` workload 来说，客户端并行 range 下载仍然是更强的模式 + +## 4. 当前推荐 + +在这轮 baseline 之后，最自然的下一步是： + +1. 先补一轮更稳妥的确认性复测，只保留重点点位： + - `1GiB c1` + - `1GiB c4` + - `2GiB c1` + - `2GiB c4` +2. 如果结论仍稳定，再继续补： + - larger `range_workers` 对比 + - encrypted `>1GiB GET` + - compressed `>1GiB GET` + +## 5. 当前边界说明 + +这轮还不能直接下最终结论的地方： + +1. 只有 `1 round` +2. `ranged_parallel` 当前只测了 `range_workers=4` +3. 还没有拆 plain / encrypted / compressed + +因此当前更准确的表述是： + +1. plain `>1GiB GET` 的第一轮 baseline 已显示 `ranged_parallel` 明显领先 +2. 但还需要更小面复测来确认这个结论的稳定性 diff --git a/docs/operations/issue-713-gt1g-get-ops-guide-zh.md b/docs/operations/issue-713-gt1g-get-ops-guide-zh.md new file mode 100644 index 000000000..22936faf1 --- /dev/null +++ b/docs/operations/issue-713-gt1g-get-ops-guide-zh.md @@ -0,0 +1,258 @@ +# Issue #713 `>1GiB GET` 优化与验证手册 + +## 1. 目标 + +本文用于收口 `backlog#713` 当前这一轮 `>1GiB GET` 优化的实现点与验证方式。 + +当前这轮改动先解决三件事： + +1. `sequential hint` 真正参与 GET I/O 策略计算 +2. storage media buffer cap 不再被统一的 `1MiB` hard clamp 二次截断 +3. `>1GiB`、非 range、允许 readahead 的 streaming GET，会走更大的 stream buffer 策略 + +## 2. 当前代码行为变化 + +### 2.1 多因素调度 + +涉及： + +1. `rustfs/src/storage/concurrency/io_schedule.rs` + +当前行为： + +1. `IoSchedulingContext.is_sequential_hint=true` 时，会优先按 `Sequential` 模式参与 buffer / readahead 决策 +2. storage profile 提供的 media cap 现在可以真实生效 +3. 例如默认常量下： + - `NVMe`: `2MiB` + - `SSD`: `1MiB` + - `HDD`: `512KiB` + +### 2.2 GET body stream 策略 + +涉及： + +1. `rustfs/src/app/object_usecase.rs` + +当前行为： + +1. `optimal_buffer_size` 不再被额外 `min(base_buffer_size)` 压回去 +2. `>1GiB`、非 range、`enable_readahead=true` 的 streaming GET，会命中： + - `large_sequential_readahead` +3. 该路径当前把 `ReaderStream` buffer 扩到： + - `min(optimal_buffer_size * 2, 4MiB)` + +### 2.3 GET observability + +涉及： + +1. `crates/io-metrics/src/lib.rs` + +新增： + +1. `rustfs_io_get_object_stream_strategy_total{strategy=...}` +2. `rustfs_io_get_object_stream_buffer_size_bytes{strategy=...}` +3. `rustfs_io_get_object_stream_response_size_bytes{strategy=...}` + +当前策略标签： + +1. `standard` +2. `large_sequential_readahead` + +## 3. 推荐验证顺序 + +按下面顺序做最稳妥： + +1. 先准备 plain `1GiB` / `2GiB` 对象 +2. 先跑 `sequential` baseline +3. 再跑 `ranged_parallel` +4. 最后再决定是否继续补 encrypted / compressed + +## 4. 准备测试对象 + +直接使用： + +1. `scripts/prepare_gt1g_get_test_objects.sh` + +示例： + +```bash +bash scripts/prepare_gt1g_get_test_objects.sh \ + --endpoint http://127.0.0.1:9000 \ + --access-key rustfsadmin \ + --secret-key rustfsadmin \ + --bucket rustfs-bench \ + --prefix bench/issue713/plain +``` + +默认会准备： + +1. `1GiB=plain-1g.bin` +2. `2GiB=plain-2g.bin` + +默认最终对象路径： + +1. `s3://rustfs-bench/bench/issue713/plain/plain-1g.bin` +2. `s3://rustfs-bench/bench/issue713/plain/plain-2g.bin` + +## 5. 执行 GET 矩阵 + +直接使用： + +1. `scripts/run_gt1g_get_http_matrix.sh` + +这个脚本特点： + +1. 不把响应体落到本地磁盘 +2. 直接把 body 丢到 `/dev/null` +3. 用 `curl` 的 `size_download` 做字节校验和吞吐计算 +4. 同时支持： + - `sequential` + - `ranged_parallel` + +补充说明： + +1. 如果本机没有 `mc`，脚本会自动 fallback 到仓库内置的 `rustfs/tests/gt1g_get_benchmark_tool.rs` +2. fallback 模式下，结果目录会落在： + - `rustfs/target/bench/...` +3. 非 fallback 模式下，结果目录仍按脚本参数落在： + - `target/bench/...` + +### 5.1 baseline：sequential + ranged_parallel + +```bash +bash scripts/run_gt1g_get_http_matrix.sh \ + --endpoint http://127.0.0.1:9000 \ + --access-key rustfsadmin \ + --secret-key rustfsadmin \ + --bucket rustfs-bench \ + --objects '1GiB=bench/issue713/plain/plain-1g.bin,2GiB=bench/issue713/plain/plain-2g.bin' \ + --modes sequential,ranged_parallel \ + --concurrencies 1,4,8 \ + --range-workers 4 \ + --rounds 3 \ + --cooldown-secs 20 +``` + +### 5.2 只跑 sequential + +```bash +bash scripts/run_gt1g_get_http_matrix.sh \ + --endpoint http://127.0.0.1:9000 \ + --access-key rustfsadmin \ + --secret-key rustfsadmin \ + --bucket rustfs-bench \ + --objects '1GiB=bench/issue713/plain/plain-1g.bin,2GiB=bench/issue713/plain/plain-2g.bin' \ + --modes sequential \ + --concurrencies 1,4,8 \ + --rounds 3 \ + --cooldown-secs 20 +``` + +### 5.3 只跑 ranged_parallel + +```bash +bash scripts/run_gt1g_get_http_matrix.sh \ + --endpoint http://127.0.0.1:9000 \ + --access-key rustfsadmin \ + --secret-key rustfsadmin \ + --bucket rustfs-bench \ + --objects '1GiB=bench/issue713/plain/plain-1g.bin,2GiB=bench/issue713/plain/plain-2g.bin' \ + --modes ranged_parallel \ + --concurrencies 1,4,8 \ + --range-workers 4,8 \ + --rounds 3 \ + --cooldown-secs 20 +``` + +## 6. 输出目录 + +默认输出： + +```text +target/bench/gt1g-get-http-/ +``` + +重点看： + +1. `round_results.csv` +2. `median_summary.csv` +3. `logs/` + +`round_results.csv` 字段： + +1. `object_label` +2. `object_key` +3. `mode` +4. `range_workers` +5. `concurrency` +6. `round` +7. `status` +8. `total_bytes` +9. `elapsed_ms` +10. `throughput_bps` +11. `avg_client_latency_ms` + +## 7. 当前推荐观察点 + +### 7.1 sequential 是否命中大对象策略 + +查询： + +```promql +sum by (strategy) ( + increase(rustfs_io_get_object_stream_strategy_total[5m]) +) +``` + +预期： + +1. `>1GiB` 顺序 GET 应出现 `large_sequential_readahead` + +### 7.2 GET stream buffer 分布 + +```promql +histogram_quantile( + 0.95, + sum by (strategy, le) ( + rate(rustfs_io_get_object_stream_buffer_size_bytes_bucket[5m]) + ) +) +``` + +### 7.3 GET 端到端完成时长 + +```promql +histogram_quantile( + 0.95, + sum by (le) ( + rate(rustfs_io_get_object_total_duration_seconds_bucket[5m]) + ) +) +``` + +## 8. 当前判断建议 + +如果 sequential 在 `1GiB/2GiB`、`c1` 下仍然没有明显优于旧结果，优先检查： + +1. `large_sequential_readahead` 是否真的命中 +2. buffer size 是否仍停在 `1MiB` +3. 请求是否实际是 range / encrypted / compressed，从而没走当前优化路径 + +如果 ranged_parallel 明显优于 sequential，再继续补： + +1. 客户端并行分片数建议 +2. plain 与 encrypted/compressed 的拆分对比 + +## 9. 当前边界 + +这轮实现还没有直接覆盖： + +1. encrypted `>1GiB GET` 的单独 read-ahead 策略 +2. compressed `>1GiB GET` 的专门 index / range 成本优化 +3. server-side 主动预取缓存 + +因此当前更适合先做： + +1. plain sequential vs ranged_parallel 基线 +2. 命中策略与吞吐结果收敛 +3. 再决定下一刀是否继续进 encrypted / compressed diff --git a/rustfs/src/app/object_usecase.rs b/rustfs/src/app/object_usecase.rs index 07ece3f83..052550aed 100644 --- a/rustfs/src/app/object_usecase.rs +++ b/rustfs/src/app/object_usecase.rs @@ -87,6 +87,7 @@ use md5::Context as Md5Context; use metrics::{counter, histogram}; use pin_project_lite::pin_project; use rustfs_concurrency::GetObjectQueueSnapshot; +use rustfs_config::MI_B; use rustfs_filemeta::{ REPLICATE_INCOMING_DELETE, ReplicateDecision, ReplicateTargetDecision, ReplicationState, ReplicationStatusType, ReplicationType, RestoreStatusOps, VersionPurgeStatusType, parse_restore_obj_status, replication_statuses_map, @@ -267,6 +268,7 @@ struct GetObjectStrategyContext { #[allow(dead_code)] io_strategy: concurrency::IoStrategy, optimal_buffer_size: usize, + enable_readahead: bool, } struct GetObjectOutputContext { @@ -282,6 +284,25 @@ enum GetObjectTimeoutStage { BeforeRead, } +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +enum GetObjectStreamStrategy { + Standard, + LargeSequentialReadahead, +} + +impl GetObjectStreamStrategy { + fn as_str(self) -> &'static str { + match self { + Self::Standard => "standard", + Self::LargeSequentialReadahead => "large_sequential_readahead", + } + } +} + +const LARGE_SEQUENTIAL_GET_THRESHOLD_BYTES: i64 = 1024 * 1024 * 1024; +const LARGE_SEQUENTIAL_GET_STREAM_BUFFER_CAP_BYTES: usize = 4 * MI_B; +const LARGE_SEQUENTIAL_GET_READAHEAD_MULTIPLIER: usize = 2; + async fn enqueue_transitioned_delete_cleanup( store: Arc, bucket: &str, @@ -1530,15 +1551,42 @@ impl DefaultObjectUsecase { ))) } - fn build_reader_blob(reader: R, response_content_length: i64, optimal_buffer_size: usize) -> Option + fn select_stream_buffer_strategy( + response_content_length: i64, + optimal_buffer_size: usize, + enable_readahead: bool, + has_range: bool, + ) -> (usize, GetObjectStreamStrategy) { + if enable_readahead && !has_range && response_content_length >= LARGE_SEQUENTIAL_GET_THRESHOLD_BYTES { + let expanded_buffer_size = optimal_buffer_size + .saturating_mul(LARGE_SEQUENTIAL_GET_READAHEAD_MULTIPLIER) + .min(LARGE_SEQUENTIAL_GET_STREAM_BUFFER_CAP_BYTES) + .max(optimal_buffer_size); + return (expanded_buffer_size, GetObjectStreamStrategy::LargeSequentialReadahead); + } + + (optimal_buffer_size, GetObjectStreamStrategy::Standard) + } + + fn build_reader_blob( + reader: R, + response_content_length: i64, + stream_buffer_size: usize, + stream_strategy: GetObjectStreamStrategy, + ) -> Option where R: AsyncRead + Send + Sync + 'static, { let expected = response_content_length.max(0) as usize; + rustfs_io_metrics::record_get_object_stream_strategy( + stream_strategy.as_str(), + stream_buffer_size, + response_content_length, + ); #[cfg(feature = "tracing-chunk-debug")] let stream = { let mut emitted = 0usize; - ReaderStream::with_capacity(reader, optimal_buffer_size).inspect(move |item| match item { + ReaderStream::with_capacity(reader, stream_buffer_size).inspect(move |item| match item { Ok(bytes) => { emitted += bytes.len(); tracing::debug!(emitted, expected, chunk_len = bytes.len(), "GetObject ReaderStream emitted bytes"); @@ -1554,7 +1602,7 @@ impl DefaultObjectUsecase { }) }; #[cfg(not(feature = "tracing-chunk-debug"))] - let stream = ReaderStream::with_capacity(reader, optimal_buffer_size); + let stream = ReaderStream::with_capacity(reader, stream_buffer_size); Some(StreamingBlob::wrap(bytes_stream(stream, expected))) } @@ -1864,7 +1912,11 @@ impl DefaultObjectUsecase { queue_status: &concurrency::IoQueueStatus, concurrent_requests: usize, ) -> GetObjectStrategyContext { - let base_buffer_size = self.base_buffer_size(); + let base_buffer_size = if response_content_length > 0 { + get_buffer_size_opt_in(response_content_length) + } else { + self.base_buffer_size() + }; let is_sequential_hint = if rs.is_none() { true @@ -1932,9 +1984,8 @@ impl DefaultObjectUsecase { "I/O priority finalized with actual request size" ); - let base_buffer_size = get_buffer_size_opt_in(response_content_length); let optimal_buffer_size = if io_strategy.buffer_size > 0 { - io_strategy.buffer_size.min(base_buffer_size) + io_strategy.buffer_size } else { get_concurrency_aware_buffer_size(response_content_length, base_buffer_size) }; @@ -1943,10 +1994,12 @@ impl DefaultObjectUsecase { "GetObject buffer sizing: file_size={}, base={}, optimal={}, concurrent_requests={}, io_strategy={:?}", response_content_length, base_buffer_size, optimal_buffer_size, concurrent_requests, io_strategy.load_level ); + let enable_readahead = io_strategy.enable_readahead; GetObjectStrategyContext { io_strategy, optimal_buffer_size, + enable_readahead, } } @@ -1993,6 +2046,7 @@ impl DefaultObjectUsecase { info: &ObjectInfo, response_content_length: i64, optimal_buffer_size: usize, + enable_readahead: bool, part_number: Option, has_range: bool, encryption_applied: bool, @@ -2023,7 +2077,14 @@ impl DefaultObjectUsecase { } debug!(buffer_size = optimal_buffer_size, "Encrypted object uses streaming decrypt path"); - return Ok(Self::build_reader_blob(final_stream, response_content_length, optimal_buffer_size)); + let (stream_buffer_size, stream_strategy) = + Self::select_stream_buffer_strategy(response_content_length, optimal_buffer_size, enable_readahead, has_range); + return Ok(Self::build_reader_blob( + final_stream, + response_content_length, + stream_buffer_size, + stream_strategy, + )); } let should_provide_seek_support = @@ -2049,7 +2110,14 @@ impl DefaultObjectUsecase { } } - Ok(Self::build_reader_blob(final_stream, response_content_length, optimal_buffer_size)) + let (stream_buffer_size, stream_strategy) = + Self::select_stream_buffer_strategy(response_content_length, optimal_buffer_size, enable_readahead, has_range); + Ok(Self::build_reader_blob( + final_stream, + response_content_length, + stream_buffer_size, + stream_strategy, + )) } fn put_object_execution_context(req: &S3Request) -> (EventName, QuotaOperation, &'static str) { @@ -2754,6 +2822,7 @@ impl DefaultObjectUsecase { let GetObjectStrategyContext { io_strategy: _, optimal_buffer_size, + enable_readahead, } = strategy; let body = Self::build_get_object_body( @@ -2761,6 +2830,7 @@ impl DefaultObjectUsecase { &info, response_content_length, optimal_buffer_size, + enable_readahead, part_number, rs.is_some(), encryption_applied, @@ -5487,6 +5557,7 @@ mod tests { &info, 18_i64 * 1024 * 1024 * 1024, 128 * 1024, + true, None, false, false, @@ -5518,6 +5589,7 @@ mod tests { &info, 18_i64 * 1024 * 1024 * 1024, 128 * 1024, + true, None, false, true, @@ -5533,6 +5605,28 @@ mod tests { ); } + #[test] + fn select_stream_buffer_strategy_expands_large_sequential_gets() { + let (buffer_size, strategy) = + DefaultObjectUsecase::select_stream_buffer_strategy(2_i64 * 1024 * 1024 * 1024, 2 * MI_B, true, false); + + assert_eq!(strategy, GetObjectStreamStrategy::LargeSequentialReadahead); + assert_eq!(buffer_size, 4 * MI_B); + } + + #[test] + fn select_stream_buffer_strategy_keeps_ranges_and_small_gets_standard() { + let (range_buffer_size, range_strategy) = + DefaultObjectUsecase::select_stream_buffer_strategy(2_i64 * 1024 * 1024 * 1024, 2 * MI_B, true, true); + assert_eq!(range_strategy, GetObjectStreamStrategy::Standard); + assert_eq!(range_buffer_size, 2 * MI_B); + + let (small_buffer_size, small_strategy) = + DefaultObjectUsecase::select_stream_buffer_strategy(64 * 1024 * 1024, 512 * 1024, true, false); + assert_eq!(small_strategy, GetObjectStreamStrategy::Standard); + assert_eq!(small_buffer_size, 512 * 1024); + } + #[test] fn should_use_zero_copy_rejects_encrypted_requests_with_sse_customer_algorithm() { let mut headers = HeaderMap::new(); diff --git a/rustfs/src/storage/concurrency/io_schedule.rs b/rustfs/src/storage/concurrency/io_schedule.rs index a62eb5843..afaf59ce3 100644 --- a/rustfs/src/storage/concurrency/io_schedule.rs +++ b/rustfs/src/storage/concurrency/io_schedule.rs @@ -889,6 +889,11 @@ impl IoStrategy { // Stage 1: Start with base buffer size let mut buffer_size; let mut buffer_multiplier = 1.0; + let effective_access_pattern = if context.is_sequential_hint { + AccessPattern::Sequential + } else { + context.access_pattern + }; // Stage 2: Apply load level reduction based on permit wait let load_level = IoLoadLevel::from_wait_duration_with_thresholds( @@ -924,7 +929,7 @@ impl IoStrategy { ); // Stage 5: Apply access pattern adjustments - let pattern_multiplier = match context.access_pattern { + let pattern_multiplier = match effective_access_pattern { AccessPattern::Sequential => storage_profile.sequential_boost_multiplier, AccessPattern::Random => storage_profile.random_penalty_multiplier, AccessPattern::Mixed => 1.0, @@ -964,7 +969,7 @@ impl IoStrategy { // Apply final clamp (safety bounds) let clamp_min = 32 * KI_B; - let clamp_max = MI_B; + let clamp_max = buffer_cap.max(MI_B); #[cfg(feature = "io-scheduler-debug")] let clamp_min_applied = buffer_size < clamp_min; #[cfg(feature = "io-scheduler-debug")] @@ -982,7 +987,7 @@ impl IoStrategy { }; // Apply access pattern override - let readahead_disabled_by_pattern = matches!(context.access_pattern, AccessPattern::Random); + let readahead_disabled_by_pattern = matches!(effective_access_pattern, AccessPattern::Random); if readahead_disabled_by_pattern { should_enable_readahead = false; #[cfg(feature = "io-scheduler-debug")] @@ -993,7 +998,7 @@ impl IoStrategy { // Apply concurrency override let readahead_disabled_by_concurrency = context.concurrent_requests >= config.random_readahead_disable_concurrency; - if readahead_disabled_by_concurrency && matches!(context.access_pattern, AccessPattern::Random) { + if readahead_disabled_by_concurrency && matches!(effective_access_pattern, AccessPattern::Random) { should_enable_readahead = false; #[cfg(feature = "io-scheduler-debug")] { @@ -1039,7 +1044,7 @@ impl IoStrategy { let core = IoStrategyCore { // ===== Basic Configuration ===== storage_media: context.storage_media, - access_pattern: context.access_pattern, + access_pattern: effective_access_pattern, request_size: context.file_size, base_buffer_size: context.base_buffer_size, buffer_cap, @@ -1055,7 +1060,7 @@ impl IoStrategy { observed_bandwidth_bps: context.observed_bandwidth_bps, bandwidth_tier, bandwidth_limited, - sequential_detected: matches!(context.access_pattern, AccessPattern::Sequential), + sequential_detected: matches!(effective_access_pattern, AccessPattern::Sequential), // ===== Decision Flags ===== storage_profile, @@ -1065,8 +1070,8 @@ impl IoStrategy { // ===== Tuning Multipliers ===== final_multiplier: buffer_multiplier, - should_throttle_random_io: matches!(context.access_pattern, AccessPattern::Random), - should_expand_for_sequential: matches!(context.access_pattern, AccessPattern::Sequential), + should_throttle_random_io: matches!(effective_access_pattern, AccessPattern::Random), + should_expand_for_sequential: matches!(effective_access_pattern, AccessPattern::Sequential), should_reduce_for_concurrency: concurrency_multiplier < 1.0, should_reduce_for_bandwidth: bandwidth_limited, should_disable_readahead: !enable_readahead, @@ -1112,7 +1117,7 @@ impl IoStrategy { // ===== State Labels ===== load_level_label: load_level_clone.as_str(), - pattern_label: context.access_pattern.as_str(), + pattern_label: effective_access_pattern.as_str(), media_label: match context.storage_media { StorageMedia::Nvme => "nvme", StorageMedia::Ssd => "ssd", @@ -1143,8 +1148,8 @@ impl IoStrategy { read_size_known: context.file_size > 0, // ===== Decision Tracking ===== - random_penalty_applied: matches!(context.access_pattern, AccessPattern::Random), - sequential_boost_applied: matches!(context.access_pattern, AccessPattern::Sequential), + random_penalty_applied: matches!(effective_access_pattern, AccessPattern::Random), + sequential_boost_applied: matches!(effective_access_pattern, AccessPattern::Sequential), buffer_cap_applied, clamp_min_applied, clamp_max_applied, @@ -2265,13 +2270,38 @@ mod tests { let config = IoSchedulerConfig::default(); let strategy = IoStrategy::from_context_with_config(&context, &config); - // Should be capped at NVMe buffer cap and 1MB max - assert!(strategy.buffer_size <= MI_B, "Should be capped at 1MB max"); + // Should be capped at the configured NVMe media cap. + assert_eq!(strategy.buffer_size, rustfs_config::DEFAULT_OBJECT_IO_NVME_BUFFER_CAP); #[cfg(feature = "io-scheduler-debug")] assert!(strategy.debug_info.buffer_cap_applied, "Buffer cap should be applied"); } + #[tokio::test] + #[serial] + async fn test_multi_factor_strategy_applies_sequential_hint_when_pattern_unknown() { + let context = IoSchedulingContext { + file_size: 2 * 1024 * 1024 * 1024, // 2GiB + base_buffer_size: 1024 * 1024, // 1MiB base + permit_wait_duration: Duration::from_millis(1), // Low load + is_sequential_hint: true, + access_pattern: AccessPattern::Unknown, + storage_media: StorageMedia::Nvme, + observed_bandwidth_bps: Some(1000 * 1024 * 1024), + concurrent_requests: 1, + }; + + let config = IoSchedulerConfig::default(); + let strategy = IoStrategy::from_context_with_config(&context, &config); + + assert_eq!(strategy.access_pattern, AccessPattern::Sequential); + assert!(strategy.enable_readahead, "Sequential hint should keep readahead enabled"); + assert!( + strategy.buffer_size > context.base_buffer_size, + "Sequential hint should allow a larger buffer for large GETs" + ); + } + #[tokio::test] #[serial] async fn test_multi_factor_strategy_bandwidth_low_reduces_buffer() { diff --git a/rustfs/tests/gt1g_get_benchmark_tool.rs b/rustfs/tests/gt1g_get_benchmark_tool.rs new file mode 100644 index 000000000..72a6b03d7 --- /dev/null +++ b/rustfs/tests/gt1g_get_benchmark_tool.rs @@ -0,0 +1,526 @@ +use anyhow::{Context, Result, anyhow}; +use aws_config::BehaviorVersion; +use aws_config::meta::region::RegionProviderChain; +use aws_sdk_s3::Client; +use aws_sdk_s3::config::{Builder as S3ConfigBuilder, Credentials, Region}; +use aws_sdk_s3::primitives::ByteStream; +use std::env; +use std::path::PathBuf; +use std::time::Instant; +use tokio::fs::OpenOptions; +use tokio::io::AsyncWriteExt; +use tokio::time::{Duration, sleep}; + +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +enum ToolAction { + Prepare, + Bench, +} + +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +enum BenchMode { + Sequential, + RangedParallel, +} + +impl BenchMode { + fn as_str(self) -> &'static str { + match self { + Self::Sequential => "sequential", + Self::RangedParallel => "ranged_parallel", + } + } +} + +#[derive(Clone, Debug)] +struct ObjectSpec { + label: String, + key: String, + size_bytes: i64, +} + +#[derive(Clone, Debug)] +struct ToolSettings { + action: ToolAction, + endpoint: String, + access_key: String, + secret_key: String, + bucket: String, + region: String, + out_dir: PathBuf, + objects: Vec, + modes: Vec, + concurrencies: Vec, + range_workers: Vec, + rounds: usize, + cooldown_secs: u64, + force: bool, +} + +type BenchGroupKey = (String, String, String, usize, usize); +type BenchGroupValue = (bool, u64, f64, f64, f64); + +fn parse_action() -> Result { + match env::var("GT1G_GET_ACTION") + .unwrap_or_else(|_| "bench".to_string()) + .trim() + .to_ascii_lowercase() + .as_str() + { + "prepare" => Ok(ToolAction::Prepare), + "bench" => Ok(ToolAction::Bench), + other => Err(anyhow!("unsupported GT1G_GET_ACTION: {other}")), + } +} + +fn parse_size_label_to_bytes(label: &str) -> Result { + let normalized = label.trim(); + if let Some(raw) = normalized.strip_suffix("GiB") { + let value = raw.trim().parse::()?; + return Ok(value * 1024 * 1024 * 1024); + } + if let Some(raw) = normalized.strip_suffix("MiB") { + let value = raw.trim().parse::()?; + return Ok(value * 1024 * 1024); + } + if let Some(raw) = normalized.strip_suffix("KiB") { + let value = raw.trim().parse::()?; + return Ok(value * 1024); + } + Err(anyhow!("unsupported size label: {label}")) +} + +fn parse_object_specs(raw: &str) -> Result> { + raw.split(',') + .map(str::trim) + .filter(|entry| !entry.is_empty()) + .map(|entry| { + let (label, key) = entry.split_once('=').ok_or_else(|| anyhow!("invalid object spec: {entry}"))?; + Ok(ObjectSpec { + label: label.trim().to_string(), + key: key.trim().to_string(), + size_bytes: parse_size_label_to_bytes(label.trim())?, + }) + }) + .collect() +} + +fn parse_csv_usize(raw: &str, field: &str) -> Result> { + let values: Vec = raw + .split(',') + .map(str::trim) + .filter(|entry| !entry.is_empty()) + .map(|entry| { + entry + .parse::() + .with_context(|| format!("invalid {field} value: {entry}")) + }) + .collect::>>()?; + if values.is_empty() { + return Err(anyhow!("{field} cannot be empty")); + } + Ok(values) +} + +fn parse_modes(raw: &str) -> Result> { + let modes: Vec = raw + .split(',') + .map(str::trim) + .filter(|entry| !entry.is_empty()) + .map(|entry| match entry { + "sequential" => Ok(BenchMode::Sequential), + "ranged_parallel" => Ok(BenchMode::RangedParallel), + other => Err(anyhow!("unsupported GT1G_GET_MODES value: {other}")), + }) + .collect::>>()?; + if modes.is_empty() { + return Err(anyhow!("GT1G_GET_MODES cannot be empty")); + } + Ok(modes) +} + +impl ToolSettings { + fn from_env() -> Result { + let action = parse_action()?; + let endpoint = env::var("GT1G_GET_ENDPOINT").context("missing GT1G_GET_ENDPOINT")?; + let access_key = env::var("GT1G_GET_ACCESS_KEY").context("missing GT1G_GET_ACCESS_KEY")?; + let secret_key = env::var("GT1G_GET_SECRET_KEY").context("missing GT1G_GET_SECRET_KEY")?; + let bucket = env::var("GT1G_GET_BUCKET").context("missing GT1G_GET_BUCKET")?; + let region = env::var("GT1G_GET_REGION").unwrap_or_else(|_| "us-east-1".to_string()); + let out_dir = env::var("GT1G_GET_OUT_DIR") + .map(PathBuf::from) + .unwrap_or_else(|_| PathBuf::from(format!("target/bench/gt1g-get-tool-{}", chrono_like_timestamp()))); + let objects = parse_object_specs(&env::var("GT1G_GET_OBJECTS").context("missing GT1G_GET_OBJECTS")?)?; + let modes = parse_modes(&env::var("GT1G_GET_MODES").unwrap_or_else(|_| "sequential,ranged_parallel".to_string()))?; + let concurrencies = parse_csv_usize( + &env::var("GT1G_GET_CONCURRENCIES").unwrap_or_else(|_| "1,4,8".to_string()), + "GT1G_GET_CONCURRENCIES", + )?; + let range_workers = parse_csv_usize( + &env::var("GT1G_GET_RANGE_WORKERS").unwrap_or_else(|_| "4".to_string()), + "GT1G_GET_RANGE_WORKERS", + )?; + let rounds = env::var("GT1G_GET_ROUNDS") + .unwrap_or_else(|_| "3".to_string()) + .parse::() + .context("invalid GT1G_GET_ROUNDS")?; + let cooldown_secs = env::var("GT1G_GET_COOLDOWN_SECS") + .unwrap_or_else(|_| "15".to_string()) + .parse::() + .context("invalid GT1G_GET_COOLDOWN_SECS")?; + let force = env::var("GT1G_GET_FORCE") + .ok() + .map(|raw| matches!(raw.as_str(), "1" | "true" | "TRUE" | "yes" | "YES")) + .unwrap_or(false); + + Ok(Self { + action, + endpoint, + access_key, + secret_key, + bucket, + region, + out_dir, + objects, + modes, + concurrencies, + range_workers, + rounds, + cooldown_secs, + force, + }) + } +} + +fn chrono_like_timestamp() -> String { + use std::time::{SystemTime, UNIX_EPOCH}; + let now = SystemTime::now().duration_since(UNIX_EPOCH).unwrap_or_default().as_secs(); + now.to_string() +} + +async fn build_client(settings: &ToolSettings) -> Result { + let region = Region::new(settings.region.clone()); + let region_provider = RegionProviderChain::first_try(Some(region.clone())); + let shared_config = aws_config::defaults(BehaviorVersion::latest()) + .region(region_provider) + .credentials_provider(Credentials::new( + settings.access_key.clone(), + settings.secret_key.clone(), + None, + None, + "issue713-gt1g-get-tool", + )) + .load() + .await; + + let s3_config = S3ConfigBuilder::from(&shared_config) + .endpoint_url(settings.endpoint.clone()) + .force_path_style(true) + .region(region) + .build(); + Ok(Client::from_conf(s3_config)) +} + +async fn ensure_bucket(client: &Client, bucket: &str) -> Result<()> { + if client.head_bucket().bucket(bucket).send().await.is_ok() { + return Ok(()); + } + client.create_bucket().bucket(bucket).send().await?; + Ok(()) +} + +async fn object_matches_size(client: &Client, bucket: &str, object: &ObjectSpec) -> Result { + let out = match client.head_object().bucket(bucket).key(&object.key).send().await { + Ok(out) => out, + Err(_) => return Ok(false), + }; + Ok(out.content_length().unwrap_or_default() == object.size_bytes) +} + +async fn create_sparse_file(path: &PathBuf, size_bytes: i64) -> Result<()> { + let file = OpenOptions::new() + .create(true) + .truncate(true) + .write(true) + .open(path) + .await + .with_context(|| format!("failed to open sparse file: {}", path.display()))?; + file.set_len(size_bytes as u64) + .await + .with_context(|| format!("failed to set sparse file length: {}", path.display()))?; + Ok(()) +} + +async fn prepare_objects(settings: &ToolSettings, client: &Client) -> Result<()> { + ensure_bucket(client, &settings.bucket).await?; + tokio::fs::create_dir_all(&settings.out_dir).await?; + let tmp_root = settings.out_dir.join("prepare-tmp"); + tokio::fs::create_dir_all(&tmp_root).await?; + + for object in &settings.objects { + if !settings.force && object_matches_size(client, &settings.bucket, object).await? { + println!("skip existing: s3://{}/{}", settings.bucket, object.key); + continue; + } + + let local_path = tmp_root.join(object.key.replace('/', "_")); + create_sparse_file(&local_path, object.size_bytes).await?; + println!("uploading: s3://{}/{} ({})", settings.bucket, object.key, object.label); + client + .put_object() + .bucket(&settings.bucket) + .key(&object.key) + .body(ByteStream::from_path(&local_path).await?) + .send() + .await + .with_context(|| format!("failed to upload {}", object.key))?; + let _ = tokio::fs::remove_file(&local_path).await; + } + + Ok(()) +} + +async fn drain_body(mut body: ByteStream) -> Result { + let mut total = 0u64; + while let Some(chunk) = body.try_next().await? { + total += chunk.len() as u64; + } + Ok(total) +} + +async fn run_sequential_task(client: Client, bucket: String, key: String, expected_size: i64) -> Result<(u64, u128)> { + let started = Instant::now(); + let output = client.get_object().bucket(bucket).key(key).send().await?; + let bytes = drain_body(output.body).await?; + if bytes != expected_size as u64 { + return Err(anyhow!("downloaded bytes mismatch: expected {}, got {}", expected_size, bytes)); + } + Ok((bytes, started.elapsed().as_millis())) +} + +async fn run_range_worker(client: Client, bucket: String, key: String, start: i64, end: i64) -> Result { + let range = format!("bytes={start}-{end}"); + let output = client.get_object().bucket(bucket).key(key).range(range).send().await?; + let bytes = drain_body(output.body).await?; + Ok(bytes) +} + +async fn run_ranged_parallel_task( + client: Client, + bucket: String, + key: String, + expected_size: i64, + range_workers: usize, +) -> Result<(u64, u128)> { + let started = Instant::now(); + let chunk_size = (expected_size + range_workers as i64 - 1) / range_workers as i64; + let mut join_set = tokio::task::JoinSet::new(); + + for worker_index in 0..range_workers { + let start = worker_index as i64 * chunk_size; + if start >= expected_size { + continue; + } + let end = (start + chunk_size - 1).min(expected_size - 1); + join_set.spawn(run_range_worker(client.clone(), bucket.clone(), key.clone(), start, end)); + } + + let mut total = 0u64; + while let Some(result) = join_set.join_next().await { + total += result??; + } + if total != expected_size as u64 { + return Err(anyhow!("ranged download bytes mismatch: expected {}, got {}", expected_size, total)); + } + Ok((total, started.elapsed().as_millis())) +} + +async fn run_bench(settings: &ToolSettings, client: &Client) -> Result<()> { + tokio::fs::create_dir_all(&settings.out_dir).await?; + let round_csv = settings.out_dir.join("round_results.csv"); + let median_csv = settings.out_dir.join("median_summary.csv"); + tokio::fs::write( + &round_csv, + "object_label,object_key,mode,range_workers,concurrency,round,status,total_bytes,elapsed_ms,throughput_bps,avg_client_latency_ms\n", + ) + .await?; + tokio::fs::write( + &median_csv, + "object_label,object_key,mode,range_workers,concurrency,successful_rounds,failed_rounds,median_total_bytes,median_elapsed_ms,median_throughput_bps,median_avg_client_latency_ms\n", + ) + .await?; + + let mut rows = Vec::new(); + for object in &settings.objects { + for &mode in &settings.modes { + for &concurrency in &settings.concurrencies { + let worker_values = if mode == BenchMode::Sequential { + vec![1usize] + } else { + settings.range_workers.clone() + }; + + for range_workers in worker_values { + for round in 1..=settings.rounds { + let started = Instant::now(); + let mut join_set = tokio::task::JoinSet::new(); + for _ in 0..concurrency { + let task_client = client.clone(); + let bucket = settings.bucket.clone(); + let key = object.key.clone(); + let size = object.size_bytes; + if mode == BenchMode::Sequential { + join_set.spawn(run_sequential_task(task_client, bucket, key, size)); + } else { + join_set.spawn(run_ranged_parallel_task(task_client, bucket, key, size, range_workers)); + } + } + + let mut status = "ok".to_string(); + let mut total_bytes = 0u64; + let mut latencies_ms = Vec::new(); + + while let Some(result) = join_set.join_next().await { + match result { + Ok(Ok((bytes, elapsed_ms))) => { + total_bytes += bytes; + latencies_ms.push(elapsed_ms); + } + Ok(Err(err)) => { + status = format!("failed:{err}"); + } + Err(err) => { + status = format!("failed:{err}"); + } + } + } + + let elapsed_ms = started.elapsed().as_millis(); + let avg_client_latency_ms = if latencies_ms.is_empty() { + 0.0 + } else { + latencies_ms.iter().sum::() as f64 / latencies_ms.len() as f64 + }; + let throughput_bps = if elapsed_ms == 0 { + 0.0 + } else { + total_bytes as f64 / (elapsed_ms as f64 / 1000.0) + }; + + let row = format!( + "{},{},{},{},{},{},{},{},{},{:.6},{:.3}\n", + object.label, + object.key, + mode.as_str(), + range_workers, + concurrency, + round, + status, + total_bytes, + elapsed_ms, + throughput_bps, + avg_client_latency_ms + ); + tokio::fs::OpenOptions::new() + .append(true) + .open(&round_csv) + .await? + .write_all(row.as_bytes()) + .await?; + rows.push(( + object.label.clone(), + object.key.clone(), + mode.as_str().to_string(), + range_workers, + concurrency, + status, + total_bytes, + elapsed_ms as f64, + throughput_bps, + avg_client_latency_ms, + )); + + if settings.cooldown_secs > 0 && round < settings.rounds { + sleep(Duration::from_secs(settings.cooldown_secs)).await; + } + } + } + } + } + } + + use std::collections::BTreeMap; + let mut grouped: BTreeMap> = BTreeMap::new(); + for (label, key, mode, range_workers, concurrency, status, total_bytes, elapsed_ms, throughput_bps, avg_latency_ms) in rows { + grouped + .entry((label, key, mode, range_workers, concurrency)) + .or_default() + .push((status == "ok", total_bytes, elapsed_ms, throughput_bps, avg_latency_ms)); + } + + for ((label, key, mode, range_workers, concurrency), values) in grouped { + let successful_rounds = values.iter().filter(|v| v.0).count(); + let failed_rounds = values.len().saturating_sub(successful_rounds); + let mut bytes = Vec::new(); + let mut elapsed = Vec::new(); + let mut throughput = Vec::new(); + let mut latency = Vec::new(); + for (ok, total_bytes, elapsed_ms, throughput_bps, avg_latency_ms) in values { + if ok { + bytes.push(total_bytes as f64); + elapsed.push(elapsed_ms); + throughput.push(throughput_bps); + latency.push(avg_latency_ms); + } + } + + let median = |input: &mut Vec| -> String { + if input.is_empty() { + return "N/A".to_string(); + } + input.sort_by(|a, b| a.partial_cmp(b).unwrap_or(std::cmp::Ordering::Equal)); + let n = input.len(); + if n % 2 == 1 { + format!("{:.6}", input[n / 2]) + } else { + format!("{:.6}", (input[n / 2 - 1] + input[n / 2]) / 2.0) + } + }; + + let row = format!( + "{},{},{},{},{},{},{},{},{},{},{}\n", + label, + key, + mode, + range_workers, + concurrency, + successful_rounds, + failed_rounds, + median(&mut bytes), + median(&mut elapsed), + median(&mut throughput), + median(&mut latency) + ); + tokio::fs::OpenOptions::new() + .append(true) + .open(&median_csv) + .await? + .write_all(row.as_bytes()) + .await?; + } + Ok(()) +} + +#[tokio::test] +#[ignore] +async fn gt1g_get_benchmark_tool() -> Result<()> { + let settings = ToolSettings::from_env()?; + let client = build_client(&settings).await?; + + match settings.action { + ToolAction::Prepare => prepare_objects(&settings, &client).await?, + ToolAction::Bench => run_bench(&settings, &client).await?, + } + + Ok(()) +} diff --git a/scripts/prepare_gt1g_get_test_objects.sh b/scripts/prepare_gt1g_get_test_objects.sh new file mode 100755 index 000000000..cfd260aab --- /dev/null +++ b/scripts/prepare_gt1g_get_test_objects.sh @@ -0,0 +1,249 @@ +#!/usr/bin/env bash +set -euo pipefail + +ENDPOINT="" +ACCESS_KEY="" +SECRET_KEY="" +BUCKET="" +PREFIX="bench/issue713" +OBJECTS="1GiB=plain-1g.bin,2GiB=plain-2g.bin" +REGION="us-east-1" +MC_BIN="${MC_BIN:-mc}" +FORCE=false +INSECURE=false +DRY_RUN=false + +usage() { + cat <<'USAGE' +Usage: + scripts/prepare_gt1g_get_test_objects.sh \ + --endpoint --access-key --secret-key --bucket [options] + +Required: + --endpoint + --access-key + --secret-key + --bucket + +Options: + --prefix Default: bench/issue713 + --objects Default: 1GiB=plain-1g.bin,2GiB=plain-2g.bin + Format: size=object-name,size=object-name + --region Default: us-east-1 + --mc-bin Default: mc + --force Re-upload even if object already exists + --insecure Allow insecure TLS + --dry-run + -h, --help + +Examples: + scripts/prepare_gt1g_get_test_objects.sh \ + --endpoint http://127.0.0.1:9000 \ + --access-key rustfsadmin \ + --secret-key rustfsadmin \ + --bucket rustfs-bench \ + --prefix bench/issue713/plain +USAGE +} + +arg_value() { + local flag="$1" + local value="${2:-}" + if [[ -z "$value" || "$value" == --* ]]; then + echo "ERROR: missing value for $flag" >&2 + exit 1 + fi + printf '%s\n' "$value" +} + +require_cmd() { + if ! command -v "$1" >/dev/null 2>&1; then + echo "ERROR: command not found: $1" >&2 + exit 1 + fi +} + +run_rust_helper_fallback() { + local objects_arg="" + local helper_out_dir="$OUT_DIR" + local raw_spec + IFS=',' read -r -a object_specs <<< "$OBJECTS" + for raw_spec in "${object_specs[@]}"; do + local spec size object_name + spec="$(echo "$raw_spec" | awk '{$1=$1;print}')" + [[ -z "$spec" ]] && continue + size="${spec%%=*}" + object_name="${spec#*=}" + objects_arg+="${size}=${PREFIX%/}/${object_name}," + done + objects_arg="${objects_arg%,}" + + if [[ -n "$helper_out_dir" && "$helper_out_dir" != /* ]]; then + helper_out_dir="$PWD/$helper_out_dir" + fi + + echo "mc not found; falling back to rustfs/tests/gt1g_get_benchmark_tool.rs" + GT1G_GET_ACTION=prepare \ + GT1G_GET_ENDPOINT="$ENDPOINT" \ + GT1G_GET_ACCESS_KEY="$ACCESS_KEY" \ + GT1G_GET_SECRET_KEY="$SECRET_KEY" \ + GT1G_GET_BUCKET="$BUCKET" \ + GT1G_GET_REGION="$REGION" \ + GT1G_GET_OBJECTS="$objects_arg" \ + GT1G_GET_OUT_DIR="${helper_out_dir:-$PWD/target/bench/issue713-prepare-helper}" \ + GT1G_GET_FORCE="$([[ "$FORCE" == "true" ]] && echo true || echo false)" \ + cargo test -p rustfs --test gt1g_get_benchmark_tool gt1g_get_benchmark_tool -- --ignored --nocapture +} + +parse_args() { + while [[ $# -gt 0 ]]; do + case "$1" in + --endpoint) ENDPOINT="$(arg_value "$1" "${2:-}")"; shift 2 ;; + --access-key) ACCESS_KEY="$(arg_value "$1" "${2:-}")"; shift 2 ;; + --secret-key) SECRET_KEY="$(arg_value "$1" "${2:-}")"; shift 2 ;; + --bucket) BUCKET="$(arg_value "$1" "${2:-}")"; shift 2 ;; + --prefix) PREFIX="$(arg_value "$1" "${2:-}")"; shift 2 ;; + --objects) OBJECTS="$(arg_value "$1" "${2:-}")"; shift 2 ;; + --region) REGION="$(arg_value "$1" "${2:-}")"; shift 2 ;; + --mc-bin) MC_BIN="$(arg_value "$1" "${2:-}")"; shift 2 ;; + --force) FORCE=true; shift ;; + --insecure) INSECURE=true; shift ;; + --dry-run) DRY_RUN=true; shift ;; + -h|--help) usage; exit 0 ;; + *) + echo "ERROR: unknown arg: $1" >&2 + usage + exit 1 + ;; + esac + done +} + +validate_args() { + if [[ -z "$ENDPOINT" || -z "$ACCESS_KEY" || -z "$SECRET_KEY" || -z "$BUCKET" ]]; then + echo "ERROR: --endpoint, --access-key, --secret-key, and --bucket are required" >&2 + exit 1 + fi +} + +size_to_bytes() { + local size="$1" + case "$size" in + *GiB) + local n="${size%GiB}" + echo $((n * 1024 * 1024 * 1024)) + ;; + *MiB) + local n="${size%MiB}" + echo $((n * 1024 * 1024)) + ;; + *KiB) + local n="${size%KiB}" + echo $((n * 1024)) + ;; + *B) + echo "${size%B}" + ;; + *) + echo "ERROR" + ;; + esac +} + +create_sparse_file() { + local file_path="$1" + local bytes="$2" + if command -v mkfile >/dev/null 2>&1; then + mkfile -n "$bytes" "$file_path" + else + truncate -s "$bytes" "$file_path" + fi +} + +object_exists() { + local alias_path="$1" + local -a cmd=("$MC_BIN" stat "$alias_path") + if [[ "$INSECURE" == "true" ]]; then + cmd=("$MC_BIN" --insecure stat "$alias_path") + fi + + "${cmd[@]}" >/dev/null 2>&1 +} + +main() { + parse_args "$@" + validate_args + + if ! command -v "$MC_BIN" >/dev/null 2>&1; then + run_rust_helper_fallback + exit 0 + fi + + local tmp_root + tmp_root="$(mktemp -d "${TMPDIR:-/tmp}/issue713-gt1g-get.XXXXXX")" + trap 'rm -rf "$tmp_root"' EXIT + + local mc_config_dir="$tmp_root/mc" + mkdir -p "$mc_config_dir" + + local -a mc_alias_cmd=("$MC_BIN" --config-dir "$mc_config_dir") + if [[ "$INSECURE" == "true" ]]; then + mc_alias_cmd+=("--insecure") + fi + + if [[ "$DRY_RUN" == "true" ]]; then + echo "[DRY-RUN] ${mc_alias_cmd[*]} alias set issue713 ${ENDPOINT} REDACTED REDACTED" + else + "${mc_alias_cmd[@]}" alias set issue713 "$ENDPOINT" "$ACCESS_KEY" "$SECRET_KEY" >/dev/null + fi + + IFS=',' read -r -a object_specs <<< "$OBJECTS" + for raw_spec in "${object_specs[@]}"; do + local spec size object_name bytes local_file object_key alias_path + spec="$(echo "$raw_spec" | awk '{$1=$1;print}')" + [[ -z "$spec" ]] && continue + + size="${spec%%=*}" + object_name="${spec#*=}" + if [[ -z "$size" || -z "$object_name" || "$size" == "$object_name" ]]; then + echo "ERROR: invalid object spec: $spec" >&2 + exit 1 + fi + + bytes="$(size_to_bytes "$size")" + if [[ "$bytes" == "ERROR" ]]; then + echo "ERROR: unsupported size label: $size" >&2 + exit 1 + fi + + object_key="${PREFIX%/}/${object_name}" + alias_path="issue713/${BUCKET}/${object_key}" + + if [[ "$FORCE" != "true" && "$DRY_RUN" != "true" ]] && object_exists "$alias_path"; then + echo "skip existing: s3://${BUCKET}/${object_key}" + continue + fi + + local_file="${tmp_root}/${object_name}" + create_sparse_file "$local_file" "$bytes" + + if [[ "$DRY_RUN" == "true" ]]; then + echo "[DRY-RUN] create sparse file ${local_file} (${bytes} bytes)" + echo "[DRY-RUN] ${mc_alias_cmd[*]} cp ${local_file} ${alias_path}" + else + echo "uploading: s3://${BUCKET}/${object_key} (${size})" + "${mc_alias_cmd[@]}" cp "$local_file" "$alias_path" >/dev/null + fi + done + + echo "prepared objects:" + for raw_spec in "${object_specs[@]}"; do + local spec object_name + spec="$(echo "$raw_spec" | awk '{$1=$1;print}')" + [[ -z "$spec" ]] && continue + object_name="${spec#*=}" + echo " s3://${BUCKET}/${PREFIX%/}/${object_name}" + done +} + +main "$@" diff --git a/scripts/run_gt1g_get_http_matrix.sh b/scripts/run_gt1g_get_http_matrix.sh new file mode 100755 index 000000000..e3c33770b --- /dev/null +++ b/scripts/run_gt1g_get_http_matrix.sh @@ -0,0 +1,501 @@ +#!/usr/bin/env bash +set -euo pipefail + +ENDPOINT="" +ACCESS_KEY="" +SECRET_KEY="" +BUCKET="" +OBJECTS="" +REGION="us-east-1" +MODES="sequential,ranged_parallel" +CONCURRENCIES="1,4,8" +RANGE_WORKERS="4" +ROUNDS=3 +COOLDOWN_SECS=15 +PRESIGN_EXPIRY="2h" +OUT_DIR="" +MC_BIN="${MC_BIN:-mc}" +CURL_BIN="${CURL_BIN:-curl}" +JQ_BIN="${JQ_BIN:-jq}" +INSECURE=false +DRY_RUN=false + +usage() { + cat <<'USAGE' +Usage: + scripts/run_gt1g_get_http_matrix.sh \ + --endpoint --access-key --secret-key \ + --bucket --objects [options] + +Required: + --endpoint + --access-key + --secret-key + --bucket + --objects Format: label=object-key,label=object-key + +Options: + --region Default: us-east-1 + --modes Default: sequential,ranged_parallel + --concurrencies Default: 1,4,8 + --range-workers Default: 4 + Used only by ranged_parallel mode + --rounds Default: 3 + --cooldown-secs Default: 15 + --presign-expiry Default: 2h + --out-dir Default: target/bench/gt1g-get-http- + --mc-bin Default: mc + --curl-bin Default: curl + --jq-bin Default: jq + --insecure Allow insecure TLS + --dry-run + -h, --help + +Examples: + scripts/run_gt1g_get_http_matrix.sh \ + --endpoint http://127.0.0.1:9000 \ + --access-key rustfsadmin \ + --secret-key rustfsadmin \ + --bucket rustfs-bench \ + --objects '1GiB=bench/issue713/plain-1g.bin,2GiB=bench/issue713/plain-2g.bin' +USAGE +} + +arg_value() { + local flag="$1" + local value="${2:-}" + if [[ -z "$value" || "$value" == --* ]]; then + echo "ERROR: missing value for $flag" >&2 + exit 1 + fi + printf '%s\n' "$value" +} + +require_cmd() { + if ! command -v "$1" >/dev/null 2>&1; then + echo "ERROR: command not found: $1" >&2 + exit 1 + fi +} + +run_rust_helper_fallback() { + local helper_out_dir="$OUT_DIR" + if [[ -n "$helper_out_dir" && "$helper_out_dir" != /* ]]; then + helper_out_dir="$PWD/$helper_out_dir" + fi + + echo "mc not found; falling back to rustfs/tests/gt1g_get_benchmark_tool.rs" + + GT1G_GET_ACTION=bench \ + GT1G_GET_ENDPOINT="$ENDPOINT" \ + GT1G_GET_ACCESS_KEY="$ACCESS_KEY" \ + GT1G_GET_SECRET_KEY="$SECRET_KEY" \ + GT1G_GET_BUCKET="$BUCKET" \ + GT1G_GET_REGION="$REGION" \ + GT1G_GET_OBJECTS="$OBJECTS" \ + GT1G_GET_MODES="$MODES" \ + GT1G_GET_CONCURRENCIES="$CONCURRENCIES" \ + GT1G_GET_RANGE_WORKERS="$RANGE_WORKERS" \ + GT1G_GET_ROUNDS="$ROUNDS" \ + GT1G_GET_COOLDOWN_SECS="$COOLDOWN_SECS" \ + GT1G_GET_OUT_DIR="$helper_out_dir" \ + cargo test -p rustfs --test gt1g_get_benchmark_tool gt1g_get_benchmark_tool -- --ignored --nocapture +} + +parse_args() { + while [[ $# -gt 0 ]]; do + case "$1" in + --endpoint) ENDPOINT="$(arg_value "$1" "${2:-}")"; shift 2 ;; + --access-key) ACCESS_KEY="$(arg_value "$1" "${2:-}")"; shift 2 ;; + --secret-key) SECRET_KEY="$(arg_value "$1" "${2:-}")"; shift 2 ;; + --bucket) BUCKET="$(arg_value "$1" "${2:-}")"; shift 2 ;; + --objects) OBJECTS="$(arg_value "$1" "${2:-}")"; shift 2 ;; + --region) REGION="$(arg_value "$1" "${2:-}")"; shift 2 ;; + --modes) MODES="$(arg_value "$1" "${2:-}")"; shift 2 ;; + --concurrencies) CONCURRENCIES="$(arg_value "$1" "${2:-}")"; shift 2 ;; + --range-workers) RANGE_WORKERS="$(arg_value "$1" "${2:-}")"; shift 2 ;; + --rounds) ROUNDS="$(arg_value "$1" "${2:-}")"; shift 2 ;; + --cooldown-secs) COOLDOWN_SECS="$(arg_value "$1" "${2:-}")"; shift 2 ;; + --presign-expiry) PRESIGN_EXPIRY="$(arg_value "$1" "${2:-}")"; shift 2 ;; + --out-dir) OUT_DIR="$(arg_value "$1" "${2:-}")"; shift 2 ;; + --mc-bin) MC_BIN="$(arg_value "$1" "${2:-}")"; shift 2 ;; + --curl-bin) CURL_BIN="$(arg_value "$1" "${2:-}")"; shift 2 ;; + --jq-bin) JQ_BIN="$(arg_value "$1" "${2:-}")"; shift 2 ;; + --insecure) INSECURE=true; shift ;; + --dry-run) DRY_RUN=true; shift ;; + -h|--help) usage; exit 0 ;; + *) + echo "ERROR: unknown arg: $1" >&2 + usage + exit 1 + ;; + esac + done +} + +validate_positive_int() { + local value="$1" + local label="$2" + if ! [[ "$value" =~ ^[0-9]+$ ]] || (( value <= 0 )); then + echo "ERROR: $label must be a positive integer, got: $value" >&2 + exit 1 + fi +} + +validate_nonnegative_int() { + local value="$1" + local label="$2" + if ! [[ "$value" =~ ^[0-9]+$ ]]; then + echo "ERROR: $label must be a nonnegative integer, got: $value" >&2 + exit 1 + fi +} + +trim() { + echo "$1" | awk '{$1=$1;print}' +} + +setup_output() { + if [[ -z "$OUT_DIR" ]]; then + OUT_DIR="target/bench/gt1g-get-http-$(date -u +%Y%m%dT%H%M%SZ)" + fi + LOG_DIR="${OUT_DIR}/logs" + mkdir -p "$LOG_DIR" + ROUND_CSV="${OUT_DIR}/round_results.csv" + MEDIAN_CSV="${OUT_DIR}/median_summary.csv" + cat > "$ROUND_CSV" <<'EOF' +object_label,object_key,mode,range_workers,concurrency,round,status,total_bytes,elapsed_ms,throughput_bps,avg_client_latency_ms,log_file +EOF + cat > "$MEDIAN_CSV" <<'EOF' +object_label,object_key,mode,range_workers,concurrency,successful_rounds,failed_rounds,median_total_bytes,median_elapsed_ms,median_throughput_bps,median_avg_client_latency_ms +EOF + cat > "${OUT_DIR}/artifact_layout.txt" <<'EOF' +round_results.csv +median_summary.csv +artifact_layout.txt +logs/ +-