gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-07-26 16:28:15 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(get): remove GET chunk fast path (#2507)

Browse Source
This commit is contained in:
安正超
2026-04-12 23:12:23 +08:00
committed by GitHub
parent 0e697fa7e1
commit 458086dc80
22 changed files with 80 additions and 4746 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/config/src/constants/zero_copy.rs
-18
View File
@@ -17,24 +17,6 @@
//! This module defines environment variables and default values for zero-copy
//! read operations, which use memory mapping (mmap) to avoid data copying.
// =============================================================================
// GET Fast Path Configuration
// =============================================================================
/// Environment variable for the GetObject chunk fast path master switch.
///
/// When disabled, `GetObject` bypasses the chunk-streaming fast path entirely and
/// always uses the legacy reader path. This provides an operational stopgap for
/// regressions in the streaming data plane while keeping zero-copy internals
/// configurable independently for future opt-in validation.
pub const ENV_OBJECT_GET_CHUNK_FAST_PATH_ENABLE: &str = "RUSTFS_OBJECT_GET_CHUNK_FAST_PATH_ENABLE";
/// Default: GetObject chunk fast path is disabled.
///
/// The legacy reader path remains the safe default until the chunk-streaming
/// path has sufficient regression coverage for full-body delivery semantics.
pub const DEFAULT_OBJECT_GET_CHUNK_FAST_PATH_ENABLE: bool = false;
// =============================================================================
// Zero-Copy Configuration
// =============================================================================
crates/e2e_test/src/archive_download_integrity_test.rs
+2 -3
View File
@@ -750,11 +750,10 @@ mod tests {
#[tokio::test]
#[serial]
async fn test_presigned_get_and_reverse_proxy_preserve_multipart_bytes_with_fast_path()
-> Result<(), Box<dyn Error + Send + Sync>> {
async fn test_presigned_get_and_reverse_proxy_preserve_multipart_bytes() -> Result<(), Box<dyn Error + Send + Sync>> {
init_logging();
let mut env = RustFSTestEnvironment::new().await?;
start_rustfs_server_with_env(&mut env, &[("RUSTFS_OBJECT_GET_CHUNK_FAST_PATH_ENABLE", "true")]).await?;
env.start_rustfs_server(vec![]).await?;
env.create_test_bucket(MULTIPART_ARCHIVE_TEST_BUCKET).await?;
let client = env.create_s3_client();
crates/ecstore/Cargo.toml
-12
View File
@@ -140,17 +140,5 @@ harness = false
name = "comparison_benchmark"
harness = false
[[bench]]
name = "direct_chunk_benchmark"
harness = false
[[bench]]
name = "reconstructed_chunk_benchmark"
harness = false
[[bench]]
name = "bitrot_chunk_benchmark"
harness = false
[lib]
doctest = false
crates/ecstore/README.md
+1 -34
View File
@@ -34,40 +34,7 @@ For comprehensive documentation, examples, and usage guides, please visit the ma
## 📈 Benchmarks
ECStore ships several Criterion benchmarks under [`crates/ecstore/benches/`](./benches/).
### Direct Chunk Path
Use the direct chunk benchmark to compare the current slice-forwarding path against the previous assembled-copy path:
```bash
cargo bench -p rustfs-ecstore --bench direct_chunk_benchmark
```
To run only the end-to-end ECStore range-read benchmark:
```bash
cargo bench -p rustfs-ecstore --bench direct_chunk_benchmark ecstore_get_object_chunks
```
To run the reconstructed multi-disk range-read benchmark:
```bash
cargo bench -p rustfs-ecstore --bench reconstructed_chunk_benchmark
```
### Saved Comparison Points
Latest local measurements on this branch:
- `direct_chunk_path/slice_forwarding/single_block_aligned`: about `477 ns`
- `direct_chunk_path/assembled_copy/single_block_aligned`: about `3.25 us`
- `direct_chunk_path/slice_forwarding/multi_block_unaligned`: about `963 ns`
- `direct_chunk_path/assembled_copy/multi_block_unaligned`: about `7.25 us`
- `ecstore_get_object_chunks/drain/multi_disk_range`: about `644-654 us`, throughput about `2.86-2.90 GiB/s`
- `reconstructed_chunk_path/drain/multi_disk_missing_shard`: about `1.292-1.304 ms`, throughput about `1.43-1.45 GiB/s`
These numbers are intended as branch-local reference points. Re-run the benchmark on your target machine before treating them as a regression baseline.
ECStore ships Criterion benchmarks under [`crates/ecstore/benches/`](./benches/) for the remaining erasure-code and comparison workloads. Re-run benchmarks on your target machine before treating any local measurement as a regression baseline.
## 📄 License
crates/ecstore/benches/bitrot_chunk_benchmark.rs
-106
View File
@@ -1,106 +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.
use bytes::Bytes;
use criterion::{BenchmarkId, Criterion, Throughput, criterion_group, criterion_main};
use rustfs_ecstore::bitrot::decode_bitrot_chunk_source_for_bench;
use rustfs_io_core::IoChunk;
use rustfs_utils::HashAlgorithm;
use std::hint::black_box;
struct BitrotChunkBenchCase {
name: &'static str,
source_chunks: Vec<IoChunk>,
shard_size: usize,
expected_decoded_len: usize,
expected_copied: bool,
}
fn encode_shard(checksum_algo: HashAlgorithm, shard: &[u8]) -> Vec<u8> {
let mut encoded = Vec::with_capacity(checksum_algo.size() + shard.len());
encoded.extend_from_slice(checksum_algo.hash_encode(shard).as_ref());
encoded.extend_from_slice(shard);
encoded
}
fn bitrot_chunk_bench_cases() -> [BitrotChunkBenchCase; 2] {
let checksum_algo = HashAlgorithm::Md5;
let shard_one = b"abcd";
let shard_two = b"efgh";
let encoded_one = encode_shard(checksum_algo.clone(), shard_one);
let encoded_two = encode_shard(checksum_algo.clone(), shard_two);
let mut cross_chunk = Vec::with_capacity(encoded_one.len() + encoded_two.len());
cross_chunk.extend_from_slice(&encoded_one);
cross_chunk.extend_from_slice(&encoded_two);
let split = checksum_algo.size() + 2;
[
BitrotChunkBenchCase {
name: "aligned_multi_chunk_no_copy",
source_chunks: vec![
IoChunk::Shared(Bytes::from(encoded_one)),
IoChunk::Shared(Bytes::from(encoded_two)),
],
shard_size: shard_one.len(),
expected_decoded_len: shard_one.len() + shard_two.len(),
expected_copied: false,
},
BitrotChunkBenchCase {
name: "cross_chunk_frame_copy",
source_chunks: vec![
IoChunk::Shared(Bytes::copy_from_slice(&cross_chunk[..split])),
IoChunk::Shared(Bytes::copy_from_slice(&cross_chunk[split..])),
],
shard_size: shard_one.len(),
expected_decoded_len: shard_one.len() + shard_two.len(),
expected_copied: true,
},
]
}
fn bench_bitrot_chunk_decode(c: &mut Criterion) {
let checksum_algo = HashAlgorithm::Md5;
let mut group = c.benchmark_group("bitrot_chunk_decode");
group.sample_size(20);
for case in bitrot_chunk_bench_cases() {
let (decoded, copied) =
decode_bitrot_chunk_source_for_bench(&case.source_chunks, case.shard_size, checksum_algo.clone(), false)
.expect("decode bitrot source");
let decoded_len: usize = decoded.iter().map(IoChunk::len).sum();
assert_eq!(decoded_len, case.expected_decoded_len);
assert_eq!(copied, case.expected_copied);
group.throughput(Throughput::Bytes(case.expected_decoded_len as u64));
group.bench_with_input(BenchmarkId::new("decode", case.name), &case, |b, case| {
b.iter(|| {
let result = decode_bitrot_chunk_source_for_bench(
black_box(&case.source_chunks),
black_box(case.shard_size),
checksum_algo.clone(),
false,
)
.expect("decode bitrot source");
black_box(result);
});
});
}
group.finish();
}
criterion_group!(benches, bench_bitrot_chunk_decode);
criterion_main!(benches);
crates/ecstore/benches/direct_chunk_benchmark.rs
-390
View File
@@ -1,390 +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.
use bytes::{Bytes, BytesMut};
use criterion::{BenchmarkId, Criterion, Throughput, criterion_group, criterion_main};
use futures_util::StreamExt;
use futures_util::stream;
use http::HeaderMap;
use rustfs_ecstore::bucket::metadata_sys;
use rustfs_ecstore::disk::endpoint::Endpoint;
use rustfs_ecstore::endpoints::{EndpointServerPools, Endpoints, PoolEndpoints};
use rustfs_ecstore::global::{GLOBAL_LOCAL_DISK_ID_MAP, GLOBAL_LOCAL_DISK_MAP, GLOBAL_LOCAL_DISK_SET_DRIVES};
use rustfs_ecstore::set_disk::collect_direct_data_shard_chunks_for_benchmark;
use rustfs_ecstore::store::{ECStore, init_local_disks};
use rustfs_ecstore::store_api::{
BucketOperations, BucketOptions, ChunkNativePutData, GetObjectChunkCopyMode, HTTPRangeSpec, MakeBucketOptions, ObjectIO,
ObjectOptions,
};
use rustfs_io_core::{BoxChunkStream, IoChunk, MappedChunk};
use std::hint::black_box;
use std::net::SocketAddr;
use std::sync::Arc;
use std::sync::atomic::{AtomicU16, Ordering};
use tempfile::TempDir;
use tokio::runtime::Runtime;
use tokio_util::sync::CancellationToken;
#[derive(Clone)]
struct BenchCase {
name: &'static str,
data_shards: usize,
block_size: usize,
blocks: usize,
offset: usize,
length: usize,
}
fn bench_cases() -> [BenchCase; 2] {
[
BenchCase {
name: "single_block_aligned",
data_shards: 4,
block_size: 256 * 1024,
blocks: 1,
offset: 0,
length: 256 * 1024,
},
BenchCase {
name: "multi_block_unaligned",
data_shards: 4,
block_size: 256 * 1024,
blocks: 8,
offset: 123_457,
length: 2 * 256 * 1024 + 33_333,
},
]
}
#[derive(Clone)]
struct EcstoreBenchCase {
name: &'static str,
disk_count: usize,
payload_len: usize,
range: HTTPRangeSpec,
expected_copy_mode: GetObjectChunkCopyMode,
}
struct EcstoreBenchEnv {
_temp_dir: TempDir,
store: Arc<ECStore>,
bucket: String,
key: String,
range: HTTPRangeSpec,
opts: ObjectOptions,
expected_len: usize,
expected_copy_mode: GetObjectChunkCopyMode,
}
fn ecstore_bench_cases() -> [EcstoreBenchCase; 1] {
[EcstoreBenchCase {
name: "multi_disk_range",
disk_count: 4,
payload_len: 3 * 1024 * 1024 + 137,
range: HTTPRangeSpec {
is_suffix_length: false,
start: 123_457,
end: 2 * 1024 * 1024 + 33_333,
},
expected_copy_mode: expected_direct_copy_mode(),
}]
}
#[cfg(unix)]
const fn expected_direct_copy_mode() -> GetObjectChunkCopyMode {
GetObjectChunkCopyMode::TrueZeroCopy
}
#[cfg(not(unix))]
const fn expected_direct_copy_mode() -> GetObjectChunkCopyMode {
GetObjectChunkCopyMode::SharedBytes
}
fn clone_range_spec(range: &HTTPRangeSpec) -> HTTPRangeSpec {
HTTPRangeSpec {
is_suffix_length: range.is_suffix_length,
start: range.start,
end: range.end,
}
}
fn next_loopback_addr() -> SocketAddr {
static NEXT_PORT: AtomicU16 = AtomicU16::new(39013);
let port = NEXT_PORT.fetch_add(1, Ordering::Relaxed);
SocketAddr::from(([127, 0, 0, 1], port))
}
fn build_endpoint_pools(paths: &[std::path::PathBuf]) -> EndpointServerPools {
let mut endpoints = Vec::with_capacity(paths.len());
for (idx, disk_path) in paths.iter().enumerate() {
let mut endpoint = Endpoint::try_from(disk_path.to_str().expect("utf8 path")).expect("endpoint");
endpoint.set_pool_index(0);
endpoint.set_set_index(0);
endpoint.set_disk_index(idx);
endpoints.push(endpoint);
}
EndpointServerPools(vec![PoolEndpoints {
legacy: false,
set_count: 1,
drives_per_set: paths.len(),
endpoints: Endpoints::from(endpoints),
cmd_line: "bench".to_string(),
platform: format!("OS: {} | Arch: {}", std::env::consts::OS, std::env::consts::ARCH),
}])
}
async fn build_ecstore_bench_env(case: &EcstoreBenchCase) -> EcstoreBenchEnv {
let temp_dir = tempfile::tempdir().expect("tempdir");
let mut disk_paths = Vec::with_capacity(case.disk_count);
for idx in 0..case.disk_count {
let path = temp_dir.path().join(format!("disk{}", idx + 1));
tokio::fs::create_dir_all(&path).await.expect("create disk dir");
disk_paths.push(path);
}
let endpoint_pools = build_endpoint_pools(&disk_paths);
GLOBAL_LOCAL_DISK_MAP.write().await.clear();
GLOBAL_LOCAL_DISK_ID_MAP.write().await.clear();
GLOBAL_LOCAL_DISK_SET_DRIVES.write().await.clear();
init_local_disks(endpoint_pools.clone()).await.expect("init local disks");
let store = ECStore::new(next_loopback_addr(), endpoint_pools, CancellationToken::new())
.await
.expect("create ecstore");
let buckets = store
.list_bucket(&BucketOptions {
no_metadata: true,
..Default::default()
})
.await
.expect("list buckets")
.into_iter()
.map(|bucket| bucket.name)
.collect();
metadata_sys::init_bucket_metadata_sys(store.clone(), buckets).await;
let object_id = case.name.replace('_', "-");
let bucket = format!("bench-direct-{object_id}");
let key = format!("objects/{object_id}.bin");
store
.make_bucket(
&bucket,
&MakeBucketOptions {
versioning_enabled: true,
..Default::default()
},
)
.await
.expect("make bucket");
let payload: Vec<u8> = (0..case.payload_len).map(|idx| (idx % 251) as u8).collect();
let mut reader = ChunkNativePutData::from_vec(payload);
let put_info = store
.put_object(&bucket, &key, &mut reader, &ObjectOptions::default())
.await
.expect("put object");
if case.disk_count > 1 {
assert!(put_info.data_blocks > 1, "expected multi-data-shard object");
}
let (_, expected_len) = case.range.get_offset_length(case.payload_len as i64).expect("range length");
EcstoreBenchEnv {
_temp_dir: temp_dir,
store,
bucket,
key,
range: clone_range_spec(&case.range),
opts: ObjectOptions::default(),
expected_len: expected_len as usize,
expected_copy_mode: case.expected_copy_mode,
}
}
async fn run_ecstore_get_object_chunks_bench(env: &EcstoreBenchEnv) -> (GetObjectChunkCopyMode, usize, usize) {
let mut result = env
.store
.get_object_chunks(&env.bucket, &env.key, Some(clone_range_spec(&env.range)), HeaderMap::new(), &env.opts)
.await
.expect("get object chunks");
let copy_mode = result.copy_mode;
let mut total_len = 0usize;
let mut chunk_count = 0usize;
while let Some(chunk) = result.stream.next().await {
let chunk = chunk.expect("chunk");
total_len += chunk.len();
chunk_count += 1;
}
(copy_mode, total_len, chunk_count)
}
fn build_shard_bytes(case: &BenchCase) -> Vec<Vec<Bytes>> {
let total_len = case.block_size * case.blocks;
let payload: Vec<u8> = (0..total_len).map(|idx| (idx % 251) as u8).collect();
let mut shards = vec![Vec::with_capacity(case.blocks); case.data_shards];
for block in 0..case.blocks {
let block_start = block * case.block_size;
let block_slice = &payload[block_start..block_start + case.block_size];
let shard_width = case.block_size / case.data_shards;
for (shard_idx, shard) in shards.iter_mut().enumerate().take(case.data_shards) {
let shard_start = shard_idx * shard_width;
let shard_end = shard_start + shard_width;
shard.push(Bytes::copy_from_slice(&block_slice[shard_start..shard_end]));
}
}
shards
}
fn build_mapped_streams(shards: &[Vec<Bytes>]) -> Vec<BoxChunkStream> {
shards
.iter()
.map(|shard_chunks| {
let chunks: Vec<_> = shard_chunks
.iter()
.map(|chunk| {
let mapped = MappedChunk::new(chunk.clone(), 0, chunk.len()).expect("mapped chunk");
Ok::<IoChunk, std::io::Error>(IoChunk::Mapped(mapped))
})
.collect();
Box::pin(stream::iter(chunks)) as BoxChunkStream
})
.collect()
}
fn collect_old_assembly(
shards: &[Vec<Bytes>],
data_shards: usize,
block_size: usize,
offset: usize,
length: usize,
) -> Vec<IoChunk> {
if length == 0 {
return Vec::new();
}
let start_block = offset / block_size;
let end_block = offset.saturating_add(length - 1) / block_size;
let mut result = Vec::with_capacity(end_block - start_block + 1);
for block_index in start_block..=end_block {
let block_offset = if block_index == start_block { offset % block_size } else { 0 };
let block_length = if start_block == end_block {
length
} else if block_index == start_block {
block_size - (offset % block_size)
} else if block_index == end_block {
(offset + length) % block_size
} else {
block_size
};
if block_length == 0 {
break;
}
let mut block = BytesMut::with_capacity(block_length);
let mut write_left = block_length;
let mut skip = block_offset;
for shard in shards.iter().take(data_shards) {
let shard_chunk = &shard[block_index];
if skip >= shard_chunk.len() {
skip -= shard_chunk.len();
continue;
}
let available = &shard_chunk[skip..];
skip = 0;
let take = available.len().min(write_left);
block.extend_from_slice(&available[..take]);
write_left -= take;
if write_left == 0 {
break;
}
}
result.push(IoChunk::Shared(block.freeze()));
}
result
}
fn bench_direct_chunk_path(c: &mut Criterion) {
let runtime = Runtime::new().expect("tokio runtime");
let mut group = c.benchmark_group("direct_chunk_path");
for case in bench_cases() {
let shard_bytes = build_shard_bytes(&case);
group.throughput(Throughput::Bytes(case.length as u64));
group.bench_with_input(BenchmarkId::new("slice_forwarding", case.name), &case, |b, case| {
b.iter(|| {
let streams = build_mapped_streams(&shard_bytes);
let chunks = runtime
.block_on(collect_direct_data_shard_chunks_for_benchmark(
streams,
case.data_shards,
case.block_size,
case.blocks * case.block_size,
false,
case.offset,
case.length,
))
.expect("collect direct chunks");
black_box(chunks);
});
});
group.bench_with_input(BenchmarkId::new("assembled_copy", case.name), &case, |b, case| {
b.iter(|| {
let chunks = collect_old_assembly(&shard_bytes, case.data_shards, case.block_size, case.offset, case.length);
black_box(chunks);
});
});
}
group.finish();
}
fn bench_ecstore_get_object_chunks(c: &mut Criterion) {
let runtime = Runtime::new().expect("tokio runtime");
let mut group = c.benchmark_group("ecstore_get_object_chunks");
group.sample_size(10);
for case in ecstore_bench_cases() {
let env = runtime.block_on(build_ecstore_bench_env(&case));
let (copy_mode, total_len, _) = runtime.block_on(run_ecstore_get_object_chunks_bench(&env));
assert_eq!(copy_mode, env.expected_copy_mode);
assert_eq!(total_len, env.expected_len);
group.throughput(Throughput::Bytes(env.expected_len as u64));
group.bench_with_input(BenchmarkId::new("drain", case.name), &env, |b, env| {
b.iter(|| {
let result = runtime.block_on(run_ecstore_get_object_chunks_bench(env));
black_box(result);
});
});
}
group.finish();
}
criterion_group!(benches, bench_direct_chunk_path, bench_ecstore_get_object_chunks);
criterion_main!(benches);
crates/ecstore/benches/reconstructed_chunk_benchmark.rs
-393
View File
@@ -1,393 +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.
use criterion::{BenchmarkId, Criterion, Throughput, criterion_group, criterion_main};
use futures_util::StreamExt;
use http::HeaderMap;
use rustfs_ecstore::bucket::metadata_sys;
use rustfs_ecstore::disk::endpoint::Endpoint;
use rustfs_ecstore::endpoints::{EndpointServerPools, Endpoints, PoolEndpoints};
use rustfs_ecstore::global::{GLOBAL_LOCAL_DISK_ID_MAP, GLOBAL_LOCAL_DISK_MAP, GLOBAL_LOCAL_DISK_SET_DRIVES};
use rustfs_ecstore::store::{ECStore, init_local_disks};
use rustfs_ecstore::store_api::{
BucketOperations, BucketOptions, ChunkNativePutData, CompletePart, GetObjectChunkCopyMode, HTTPRangeSpec, MakeBucketOptions,
MultipartOperations, ObjectOperations, ObjectOptions,
};
use std::hint::black_box;
use std::net::SocketAddr;
use std::path::{Path, PathBuf};
use std::sync::Arc;
use std::sync::atomic::{AtomicU16, Ordering};
use tempfile::TempDir;
use tokio::runtime::Runtime;
use tokio_util::sync::CancellationToken;
#[derive(Clone)]
enum ReconstructedReadSpec {
PartNumber {
part_number: usize,
missing_part_name: &'static str,
},
Range {
start: u64,
end: u64,
missing_part_name: &'static str,
},
}
#[derive(Clone)]
struct ReconstructedBenchCase {
object_id: &'static str,
name: &'static str,
read_spec: ReconstructedReadSpec,
}
struct ReconstructedBenchEnv {
store: Arc<ECStore>,
bucket: String,
key: String,
range: HTTPRangeSpec,
opts: ObjectOptions,
expected_len: usize,
}
struct ReconstructedBenchSuite {
_temp_dir: TempDir,
store: Arc<ECStore>,
disk_paths: Vec<PathBuf>,
}
const MULTIPART_PART_ONE_LEN: usize = 5 * 1024 * 1024;
const MULTIPART_PART_TWO_LEN: usize = 5 * 1024 * 1024 + 137;
const MULTIPART_PART_THREE_LEN: usize = 1024 * 1024 + 77;
fn reconstructed_bench_cases() -> [ReconstructedBenchCase; 4] {
[
ReconstructedBenchCase {
object_id: "mp-part2",
name: "multi_disk_missing_shard_multipart_part2",
read_spec: ReconstructedReadSpec::PartNumber {
part_number: 2,
missing_part_name: "part.2",
},
},
ReconstructedBenchCase {
object_id: "mp-cross-range",
name: "multi_disk_missing_shard_multipart_cross_part_range",
read_spec: ReconstructedReadSpec::Range {
start: (MULTIPART_PART_ONE_LEN - 32 * 1024) as u64,
end: (MULTIPART_PART_ONE_LEN + 96 * 1024) as u64,
missing_part_name: "part.2",
},
},
ReconstructedBenchCase {
object_id: "mp-part3",
name: "multi_disk_missing_shard_multipart_part3",
read_spec: ReconstructedReadSpec::PartNumber {
part_number: 3,
missing_part_name: "part.3",
},
},
ReconstructedBenchCase {
object_id: "mp-cross-final-range",
name: "multi_disk_missing_shard_multipart_cross_final_part_range",
read_spec: ReconstructedReadSpec::Range {
start: (MULTIPART_PART_ONE_LEN + MULTIPART_PART_TWO_LEN - 32 * 1024) as u64,
end: (MULTIPART_PART_ONE_LEN + MULTIPART_PART_TWO_LEN + 96 * 1024) as u64,
missing_part_name: "part.3",
},
},
]
}
fn next_loopback_addr() -> SocketAddr {
static NEXT_PORT: AtomicU16 = AtomicU16::new(39113);
let port = NEXT_PORT.fetch_add(1, Ordering::Relaxed);
SocketAddr::from(([127, 0, 0, 1], port))
}
fn clone_range_spec(range: &HTTPRangeSpec) -> HTTPRangeSpec {
HTTPRangeSpec {
is_suffix_length: range.is_suffix_length,
start: range.start,
end: range.end,
}
}
fn build_endpoint_pools(paths: &[PathBuf]) -> EndpointServerPools {
let mut endpoints = Vec::with_capacity(paths.len());
for (idx, disk_path) in paths.iter().enumerate() {
let mut endpoint = Endpoint::try_from(disk_path.to_str().expect("utf8 path")).expect("endpoint");
endpoint.set_pool_index(0);
endpoint.set_set_index(0);
endpoint.set_disk_index(idx);
endpoints.push(endpoint);
}
EndpointServerPools(vec![PoolEndpoints {
legacy: false,
set_count: 1,
drives_per_set: paths.len(),
endpoints: Endpoints::from(endpoints),
cmd_line: "bench".to_string(),
platform: format!("OS: {} | Arch: {}", std::env::consts::OS, std::env::consts::ARCH),
}])
}
fn find_part_files(root: &Path, part_name: &str, out: &mut Vec<PathBuf>) {
let Ok(entries) = std::fs::read_dir(root) else {
return;
};
for entry in entries.flatten() {
let path = entry.path();
if path.is_dir() {
find_part_files(&path, part_name, out);
continue;
}
if path.file_name().and_then(|name| name.to_str()) == Some(part_name) {
out.push(path);
}
}
}
async fn remove_part_files(root: &Path, part_name: &str) -> Vec<(PathBuf, Vec<u8>)> {
let mut paths = Vec::new();
find_part_files(root, part_name, &mut paths);
let mut removed = Vec::with_capacity(paths.len());
for path in paths {
let content = tokio::fs::read(&path).await.expect("read part file before removal");
tokio::fs::remove_file(&path).await.expect("remove part file");
removed.push((path, content));
}
removed
}
async fn restore_part_files(files: Vec<(PathBuf, Vec<u8>)>) {
for (path, content) in files {
if let Some(parent) = path.parent() {
tokio::fs::create_dir_all(parent)
.await
.expect("ensure parent for part restore");
}
tokio::fs::write(&path, content).await.expect("restore part file");
}
}
async fn run_reconstructed_get_object_chunks(
store: &Arc<ECStore>,
bucket: &str,
key: &str,
range: &HTTPRangeSpec,
opts: &ObjectOptions,
) -> (GetObjectChunkCopyMode, usize, usize) {
let mut result = store
.get_object_chunks(bucket, key, Some(clone_range_spec(range)), HeaderMap::new(), opts)
.await
.expect("get object chunks");
let copy_mode = result.copy_mode;
let mut total_len = 0usize;
let mut chunk_count = 0usize;
while let Some(chunk) = result.stream.next().await {
let chunk = chunk.expect("chunk");
total_len += chunk.len();
chunk_count += 1;
}
(copy_mode, total_len, chunk_count)
}
async fn create_multipart_object(store: &Arc<ECStore>, bucket: &str, key: &str, parts: &[Vec<u8>]) -> usize {
let upload = store
.new_multipart_upload(bucket, key, &ObjectOptions::default())
.await
.expect("new multipart upload");
let mut completed_parts = Vec::with_capacity(parts.len());
for (idx, part) in parts.iter().enumerate() {
let mut reader = ChunkNativePutData::from_vec(part.clone());
let part_info = store
.put_object_part(bucket, key, &upload.upload_id, idx + 1, &mut reader, &ObjectOptions::default())
.await
.expect("put object part");
completed_parts.push(CompletePart {
part_num: idx + 1,
etag: part_info.etag,
..Default::default()
});
}
store
.clone()
.complete_multipart_upload(bucket, key, &upload.upload_id, completed_parts, &ObjectOptions::default())
.await
.expect("complete multipart upload");
parts.iter().map(Vec::len).sum()
}
async fn build_reconstructed_bench_suite() -> ReconstructedBenchSuite {
let temp_dir = tempfile::tempdir().expect("tempdir");
let disk_paths: Vec<_> = (1..=4).map(|idx| temp_dir.path().join(format!("disk{idx}"))).collect();
for disk_path in &disk_paths {
tokio::fs::create_dir_all(disk_path).await.expect("create disk dir");
}
let endpoint_pools = build_endpoint_pools(&disk_paths);
GLOBAL_LOCAL_DISK_MAP.write().await.clear();
GLOBAL_LOCAL_DISK_ID_MAP.write().await.clear();
GLOBAL_LOCAL_DISK_SET_DRIVES.write().await.clear();
init_local_disks(endpoint_pools.clone()).await.expect("init local disks");
let store = ECStore::new(next_loopback_addr(), endpoint_pools, CancellationToken::new())
.await
.expect("create ecstore");
let buckets = store
.list_bucket(&BucketOptions {
no_metadata: true,
..Default::default()
})
.await
.expect("list buckets")
.into_iter()
.map(|bucket| bucket.name)
.collect();
metadata_sys::init_bucket_metadata_sys(store.clone(), buckets).await;
ReconstructedBenchSuite {
_temp_dir: temp_dir,
store,
disk_paths,
}
}
async fn build_reconstructed_bench_env(suite: &ReconstructedBenchSuite, case: &ReconstructedBenchCase) -> ReconstructedBenchEnv {
let bucket = format!("bench-r-{}", case.object_id);
let key = format!("objects/{}.bin", case.object_id);
suite
.store
.make_bucket(
&bucket,
&MakeBucketOptions {
versioning_enabled: true,
..Default::default()
},
)
.await
.expect("make bucket");
let part_one: Vec<u8> = (0..MULTIPART_PART_ONE_LEN).map(|idx| (idx % 251) as u8).collect();
let part_two: Vec<u8> = (0..MULTIPART_PART_TWO_LEN).map(|idx| ((idx + 11) % 251) as u8).collect();
let part_three: Vec<u8> = (0..MULTIPART_PART_THREE_LEN).map(|idx| ((idx + 29) % 251) as u8).collect();
let payload_len = create_multipart_object(&suite.store, &bucket, &key, &[part_one, part_two, part_three]).await;
let info = suite
.store
.get_object_info(&bucket, &key, &ObjectOptions::default())
.await
.expect("get object info");
let (range, opts, missing_part_name) = match case.read_spec.clone() {
ReconstructedReadSpec::PartNumber {
part_number,
missing_part_name,
} => {
let range = HTTPRangeSpec::from_object_info(&info, part_number).expect("part_number range");
let opts = ObjectOptions {
part_number: Some(part_number),
..Default::default()
};
(range, opts, missing_part_name)
}
ReconstructedReadSpec::Range {
start,
end,
missing_part_name,
} => (
HTTPRangeSpec {
is_suffix_length: false,
start: start as i64,
end: end as i64,
},
ObjectOptions::default(),
missing_part_name,
),
};
let (_, expected_len) = range.get_offset_length(payload_len as i64).expect("range length");
let mut selected_reconstructed = false;
for disk_path in &suite.disk_paths {
let object_root = disk_path
.join(&bucket)
.join("objects")
.join(format!("{}.bin", case.object_id));
let removed_parts = remove_part_files(&object_root, missing_part_name).await;
if removed_parts.is_empty() {
continue;
}
let (copy_mode, total_len, _) = run_reconstructed_get_object_chunks(&suite.store, &bucket, &key, &range, &opts).await;
if copy_mode == GetObjectChunkCopyMode::Reconstructed && total_len == expected_len as usize {
selected_reconstructed = true;
break;
}
restore_part_files(removed_parts).await;
}
assert!(
selected_reconstructed,
"failed to select a missing shard placement that triggers reconstructed copy mode for benchmark case {}",
case.name
);
ReconstructedBenchEnv {
store: suite.store.clone(),
bucket,
key,
range: clone_range_spec(&range),
opts,
expected_len: expected_len as usize,
}
}
async fn run_reconstructed_get_object_chunks_bench(env: &ReconstructedBenchEnv) -> (GetObjectChunkCopyMode, usize, usize) {
run_reconstructed_get_object_chunks(&env.store, &env.bucket, &env.key, &env.range, &env.opts).await
}
fn bench_reconstructed_chunk_path(c: &mut Criterion) {
let runtime = Runtime::new().expect("tokio runtime");
let suite = runtime.block_on(build_reconstructed_bench_suite());
let mut group = c.benchmark_group("reconstructed_chunk_path");
group.sample_size(10);
for case in reconstructed_bench_cases() {
let env = runtime.block_on(build_reconstructed_bench_env(&suite, &case));
let (copy_mode, total_len, _) = runtime.block_on(run_reconstructed_get_object_chunks_bench(&env));
assert_eq!(copy_mode, GetObjectChunkCopyMode::Reconstructed);
assert_eq!(total_len, env.expected_len);
group.throughput(Throughput::Bytes(env.expected_len as u64));
group.bench_with_input(BenchmarkId::new("drain", case.name), &env, |b, env| {
b.iter(|| {
let result = runtime.block_on(run_reconstructed_get_object_chunks_bench(env));
black_box(result);
});
});
}
group.finish();
}
criterion_group!(benches, bench_reconstructed_chunk_path);
criterion_main!(benches);
crates/ecstore/run_benchmarks.sh
-17
View File
@@ -119,16 +119,6 @@ run_large_data_test() {
print_success "Large-dataset tests completed"
}
# Run direct chunk path benchmarks
run_direct_chunk_benchmark() {
print_info "📦 Starting direct chunk path benchmarks..."
echo "================================================"
cargo bench --bench direct_chunk_benchmark
print_success "Direct chunk path benchmarks completed"
}
# Generate comparison report
generate_comparison_report() {
print_info "📊 Generating performance report..."
@@ -178,7 +168,6 @@ show_help() {
echo " full Run the full benchmark suite"
echo " performance Run detailed performance tests"
echo " simd Run the SIMD-only tests"
echo " direct Run the direct chunk path benchmarks"
echo " large Run large-dataset tests"
echo " clean Remove previous results"
echo " help Show this help message"
@@ -188,7 +177,6 @@ show_help() {
echo " $0 performance # Detailed performance test"
echo " $0 full # Full benchmark suite"
echo " $0 simd # SIMD-only benchmark"
echo " $0 direct # Direct chunk path benchmark"
echo " $0 large # Large-dataset benchmark"
echo ""
echo "Features:"
@@ -252,11 +240,6 @@ main() {
run_simd_benchmark
generate_comparison_report
;;
"direct")
cleanup
run_direct_chunk_benchmark
generate_comparison_report
;;
"large")
cleanup
run_large_data_test
crates/ecstore/src/bitrot.rs
+3 -750
View File
@@ -14,12 +14,8 @@
use crate::disk::{self, DiskAPI as _, DiskStore, error::DiskError};
use crate::erasure_coding::{BitrotReader, BitrotWriterWrapper, CustomWriter};
use crate::store_api::{GetObjectChunkCopyMode, GetObjectChunkPath, GetObjectChunkResult};
use bytes::{Bytes, BytesMut};
use futures_util::{StreamExt, stream};
use rustfs_io_core::{BoxChunkStream, IoChunk};
use bytes::Bytes;
use rustfs_utils::HashAlgorithm;
use std::collections::VecDeque;
use std::io::Cursor;
use std::time::Instant;
use tokio::io::AsyncRead;
@@ -27,315 +23,6 @@ use tracing::debug;
const BITROT_READ_OPERATION: &str = "bitrot_read";
fn classify_chunk_copy_mode(source_direct: bool, copied: bool) -> GetObjectChunkCopyMode {
if copied {
GetObjectChunkCopyMode::SingleCopy
} else if source_direct {
GetObjectChunkCopyMode::TrueZeroCopy
} else {
GetObjectChunkCopyMode::SharedBytes
}
}
struct ChunkSpan {
bytes: Bytes,
chunk: IoChunk,
copied: bool,
}
fn take_contiguous_chunk_span(chunk: &IoChunk, offset: usize, len: usize) -> std::io::Result<ChunkSpan> {
match chunk {
IoChunk::Shared(bytes) => {
let bytes = bytes.slice(offset..offset + len);
Ok(ChunkSpan {
bytes: bytes.clone(),
chunk: IoChunk::Shared(bytes),
copied: false,
})
}
IoChunk::Mapped(mapped) => {
let chunk = IoChunk::Mapped(mapped.slice(offset, len)?);
let bytes = chunk.as_bytes();
Ok(ChunkSpan {
bytes,
chunk,
copied: false,
})
}
IoChunk::Pooled(pooled) => {
let chunk = IoChunk::Pooled(pooled.slice(offset, len)?);
let bytes = chunk.as_bytes();
Ok(ChunkSpan {
bytes,
chunk,
copied: false,
})
}
}
}
struct BitrotChunkSource {
source_stream: BoxChunkStream,
source_chunks: VecDeque<IoChunk>,
source_chunk_offset: usize,
source_buffered_bytes: usize,
source_done: bool,
}
struct BitrotChunkStreamState {
source: BitrotChunkSource,
decoded_remaining: usize,
trim_prefix: usize,
output_remaining: usize,
shard_size: usize,
checksum_algo: HashAlgorithm,
skip_verify: bool,
}
struct ChunkCursor<'a> {
chunks: &'a [IoChunk],
chunk_index: usize,
chunk_offset: usize,
consumed: usize,
total_len: usize,
}
impl<'a> ChunkCursor<'a> {
fn new(chunks: &'a [IoChunk]) -> Self {
Self {
chunks,
chunk_index: 0,
chunk_offset: 0,
consumed: 0,
total_len: chunks.iter().map(IoChunk::len).sum(),
}
}
fn remaining(&self) -> usize {
self.total_len.saturating_sub(self.consumed)
}
fn skip_empty_chunks(&mut self) {
while let Some(chunk) = self.chunks.get(self.chunk_index) {
if self.chunk_offset < chunk.len() {
break;
}
self.chunk_index += 1;
self.chunk_offset = 0;
}
}
fn advance(&mut self, len: usize) {
self.consumed += len;
self.chunk_offset += len;
self.skip_empty_chunks();
}
fn take_span(&mut self, len: usize) -> std::io::Result<ChunkSpan> {
self.skip_empty_chunks();
if self.remaining() < len {
return Err(std::io::Error::new(std::io::ErrorKind::UnexpectedEof, "truncated bitrot chunk source"));
}
let Some(chunk) = self.chunks.get(self.chunk_index) else {
return Err(std::io::Error::new(std::io::ErrorKind::UnexpectedEof, "missing bitrot chunk source"));
};
let available = chunk.len().saturating_sub(self.chunk_offset);
if len <= available {
let span = take_contiguous_chunk_span(chunk, self.chunk_offset, len)?;
self.advance(len);
return Ok(span);
}
let mut aggregate = BytesMut::with_capacity(len);
let mut remaining = len;
while remaining > 0 {
self.skip_empty_chunks();
let Some(chunk) = self.chunks.get(self.chunk_index) else {
return Err(std::io::Error::new(std::io::ErrorKind::UnexpectedEof, "truncated bitrot chunk source"));
};
let available = chunk.len().saturating_sub(self.chunk_offset);
let take = available.min(remaining);
aggregate.extend_from_slice(&chunk.as_bytes()[self.chunk_offset..self.chunk_offset + take]);
self.advance(take);
remaining -= take;
}
let bytes = aggregate.freeze();
Ok(ChunkSpan {
bytes: bytes.clone(),
chunk: IoChunk::Shared(bytes),
copied: true,
})
}
}
impl BitrotChunkSource {
fn new(source_stream: BoxChunkStream, source_chunks: VecDeque<IoChunk>, source_done: bool) -> Self {
let source_buffered_bytes = source_chunks.iter().map(IoChunk::len).sum();
Self {
source_stream,
source_chunks,
source_chunk_offset: 0,
source_buffered_bytes,
source_done,
}
}
fn skip_empty_chunks(&mut self) {
while let Some(chunk) = self.source_chunks.front() {
if self.source_chunk_offset < chunk.len() {
break;
}
self.source_chunks.pop_front();
self.source_chunk_offset = 0;
}
}
async fn fill(&mut self, min_bytes: usize) -> std::io::Result<()> {
while self.source_buffered_bytes < min_bytes && !self.source_done {
match self.source_stream.next().await {
Some(Ok(chunk)) => {
self.source_buffered_bytes += chunk.len();
self.source_chunks.push_back(chunk);
}
Some(Err(err)) => return Err(err),
None => self.source_done = true,
}
}
if self.source_buffered_bytes < min_bytes {
return Err(std::io::Error::new(std::io::ErrorKind::UnexpectedEof, "truncated bitrot chunk source"));
}
Ok(())
}
fn advance(&mut self, len: usize) {
self.source_buffered_bytes = self.source_buffered_bytes.saturating_sub(len);
self.source_chunk_offset += len;
self.skip_empty_chunks();
}
fn take_span(&mut self, len: usize) -> std::io::Result<ChunkSpan> {
self.skip_empty_chunks();
if self.source_buffered_bytes < len {
return Err(std::io::Error::new(std::io::ErrorKind::UnexpectedEof, "truncated bitrot chunk source"));
}
let Some(chunk) = self.source_chunks.front() else {
return Err(std::io::Error::new(std::io::ErrorKind::UnexpectedEof, "missing bitrot chunk source"));
};
let available = chunk.len().saturating_sub(self.source_chunk_offset);
if len <= available {
let span = take_contiguous_chunk_span(chunk, self.source_chunk_offset, len)?;
self.advance(len);
return Ok(span);
}
let mut aggregate = BytesMut::with_capacity(len);
let mut remaining = len;
while remaining > 0 {
self.skip_empty_chunks();
let Some(chunk) = self.source_chunks.front() else {
return Err(std::io::Error::new(std::io::ErrorKind::UnexpectedEof, "truncated bitrot chunk source"));
};
let available = chunk.len().saturating_sub(self.source_chunk_offset);
let take = available.min(remaining);
aggregate.extend_from_slice(&chunk.as_bytes()[self.source_chunk_offset..self.source_chunk_offset + take]);
self.advance(take);
remaining -= take;
}
let bytes = aggregate.freeze();
Ok(ChunkSpan {
bytes: bytes.clone(),
chunk: IoChunk::Shared(bytes),
copied: true,
})
}
}
impl BitrotChunkStreamState {
#[allow(clippy::too_many_arguments)]
fn new(
source_stream: BoxChunkStream,
source_chunks: VecDeque<IoChunk>,
source_done: bool,
decoded_remaining: usize,
trim_prefix: usize,
output_remaining: usize,
shard_size: usize,
checksum_algo: HashAlgorithm,
skip_verify: bool,
) -> Self {
Self {
source: BitrotChunkSource::new(source_stream, source_chunks, source_done),
decoded_remaining,
trim_prefix,
output_remaining,
shard_size,
checksum_algo,
skip_verify,
}
}
fn hash_size(&self) -> usize {
self.checksum_algo.size()
}
async fn next_verified_chunk(&mut self) -> std::io::Result<Option<IoChunk>> {
let hash_size = self.hash_size();
while self.output_remaining > 0 && self.decoded_remaining > 0 {
let data_len = self.shard_size.min(self.decoded_remaining);
let expected_hash = if hash_size > 0 {
self.source.fill(hash_size).await?;
Some(self.source.take_span(hash_size)?)
} else {
None
};
self.source.fill(data_len).await?;
let data_span = self.source.take_span(data_len)?;
if let Some(expected_hash) = expected_hash
&& !self.skip_verify
&& self.checksum_algo.hash_encode(data_span.bytes.as_ref()).as_ref() != expected_hash.bytes.as_ref()
{
return Err(std::io::Error::new(std::io::ErrorKind::InvalidData, "bitrot hash mismatch"));
}
self.decoded_remaining -= data_len;
if self.trim_prefix >= data_len {
self.trim_prefix -= data_len;
continue;
}
let start = self.trim_prefix;
self.trim_prefix = 0;
let take = (data_len - start).min(self.output_remaining);
self.output_remaining -= take;
let chunk = if start == 0 && take == data_len {
data_span.chunk
} else {
data_span.chunk.slice(start, take)?
};
if !chunk.is_empty() {
return Ok(Some(chunk));
}
}
Ok(None)
}
}
/// Create a BitrotReader from either inline data or disk file stream
///
/// # Parameters
@@ -399,9 +86,8 @@ pub async fn create_bitrot_reader(
let duration_ms = start.elapsed().as_secs_f64() * 1000.0;
rustfs_io_metrics::record_io_path_selected(BITROT_READ_OPERATION, rustfs_io_metrics::IoPath::Fast);
// `read_file_zero_copy()` returns a shared `Bytes` view, but it may still
// internally aggregate multiple chunk windows. The exact chunk-native copy
// mode is only preserved by `create_bitrot_chunk_stream()`.
// `read_file_zero_copy()` returns a shared `Bytes` view, which preserves the
// mmap-backed fast path without exposing chunk-native GET internals.
rustfs_io_metrics::record_io_copy_mode(
BITROT_READ_OPERATION,
rustfs_io_metrics::CopyMode::SharedBytes,
@@ -469,198 +155,6 @@ pub async fn create_bitrot_reader(
}
}
/// Create a chunk stream from bitrot-encoded data, preserving source chunk provenance when possible.
#[allow(clippy::too_many_arguments)]
pub async fn create_bitrot_chunk_stream(
inline_data: Option<&[u8]>,
disk: Option<&DiskStore>,
bucket: &str,
path: &str,
offset: usize,
length: usize,
total_data_size: usize,
shard_size: usize,
checksum_algo: HashAlgorithm,
skip_verify: bool,
use_zero_copy: bool,
) -> disk::error::Result<Option<GetObjectChunkResult>> {
let fetch_start = (offset / shard_size) * shard_size;
let fetch_end = (offset + length).div_ceil(shard_size) * shard_size;
let fetch_end = fetch_end.min(total_data_size);
let fetch_length = fetch_end.saturating_sub(fetch_start);
let trim_prefix = offset.saturating_sub(fetch_start);
let hash_size = checksum_algo.size();
let encoded_length = fetch_length.div_ceil(shard_size) * hash_size + fetch_length;
let encoded_offset = fetch_start.div_ceil(shard_size) * hash_size + fetch_start;
let mut source_done = false;
let (source_stream, mut prefetched_chunks, source_direct) = if let Some(data) = inline_data {
source_done = true;
let mut chunks = VecDeque::new();
chunks.push_back(IoChunk::Shared(
Bytes::copy_from_slice(data).slice(encoded_offset..encoded_offset + encoded_length),
));
let source_stream: BoxChunkStream = Box::pin(stream::empty::<std::io::Result<IoChunk>>());
(source_stream, chunks, false)
} else if let Some(disk) = disk {
if use_zero_copy {
let mut source_stream = disk.read_file_chunks(bucket, path, encoded_offset, encoded_length).await?;
let mut prefetched_chunks = VecDeque::new();
let mut direct = true;
while prefetched_chunks.len() < 2 {
let Some(chunk) = source_stream.next().await else {
source_done = true;
break;
};
let chunk = chunk?;
direct &= matches!(chunk, IoChunk::Mapped(_));
prefetched_chunks.push_back(chunk);
}
(source_stream, prefetched_chunks, direct)
} else {
source_done = true;
let bytes = disk.read_file_zero_copy(bucket, path, encoded_offset, encoded_length).await?;
let mut chunks = VecDeque::new();
chunks.push_back(IoChunk::Shared(bytes));
let source_stream: BoxChunkStream = Box::pin(stream::empty::<std::io::Result<IoChunk>>());
(source_stream, chunks, false)
}
} else {
return Ok(None);
};
let copied = predicted_stream_copy(encoded_length, shard_size, checksum_algo.size(), &prefetched_chunks, source_done);
let state = BitrotChunkStreamState::new(
source_stream,
std::mem::take(&mut prefetched_chunks),
source_done,
fetch_length,
trim_prefix,
length,
shard_size,
checksum_algo,
skip_verify,
);
let stream = stream::unfold(Some(state), |state| async move {
let mut state = match state {
Some(state) => state,
None => return None,
};
match state.next_verified_chunk().await {
Ok(Some(chunk)) => {
let next_state = if state.output_remaining == 0 { None } else { Some(state) };
Some((Ok::<IoChunk, std::io::Error>(chunk), next_state))
}
Ok(None) => None,
Err(err) => Some((Err(err), None)),
}
});
Ok(Some(GetObjectChunkResult {
stream: Box::pin(stream),
path: GetObjectChunkPath::Direct,
copy_mode: classify_chunk_copy_mode(source_direct, copied),
}))
}
fn predicted_stream_copy(
encoded_length: usize,
shard_size: usize,
hash_size: usize,
prefetched_chunks: &VecDeque<IoChunk>,
source_done: bool,
) -> bool {
if prefetched_chunks.is_empty() {
return false;
}
if source_done && prefetched_chunks.len() == 1 {
return false;
}
let full_frame_len = hash_size + shard_size;
if full_frame_len == 0 {
return false;
}
let first_window_len = prefetched_chunks.front().map(IoChunk::len).unwrap_or(encoded_length);
encoded_length > first_window_len && !first_window_len.is_multiple_of(full_frame_len)
}
fn trim_chunk_vec(chunks: Vec<IoChunk>, offset: usize, length: usize) -> std::io::Result<Vec<IoChunk>> {
let mut skip = offset;
let mut remaining = length;
let mut result = Vec::new();
for chunk in chunks {
if remaining == 0 {
break;
}
let chunk_len = chunk.len();
if skip >= chunk_len {
skip -= chunk_len;
continue;
}
let start = skip;
let take = (chunk_len - start).min(remaining);
result.push(chunk.slice(start, take)?);
remaining -= take;
skip = 0;
}
Ok(result)
}
fn decode_bitrot_chunk_source(
source_chunks: &[IoChunk],
shard_size: usize,
checksum_algo: HashAlgorithm,
skip_verify: bool,
) -> std::io::Result<(Vec<IoChunk>, bool)> {
let hash_size = checksum_algo.size();
let mut cursor = ChunkCursor::new(source_chunks);
let mut result = Vec::new();
let mut copied = false;
while cursor.remaining() > 0 {
let expected_hash = if hash_size > 0 {
Some(cursor.take_span(hash_size)?)
} else {
None
};
let data_len = shard_size.min(cursor.remaining());
if data_len == 0 {
break;
}
let data_span = cursor.take_span(data_len)?;
copied |= data_span.copied;
if let Some(expected_hash) = expected_hash {
copied |= expected_hash.copied;
if !skip_verify && checksum_algo.hash_encode(data_span.bytes.as_ref()).as_ref() != expected_hash.bytes.as_ref() {
return Err(std::io::Error::new(std::io::ErrorKind::InvalidData, "bitrot hash mismatch"));
}
}
result.push(data_span.chunk);
}
Ok((result, copied))
}
#[doc(hidden)]
pub fn decode_bitrot_chunk_source_for_bench(
source_chunks: &[IoChunk],
shard_size: usize,
checksum_algo: HashAlgorithm,
skip_verify: bool,
) -> std::io::Result<(Vec<IoChunk>, bool)> {
decode_bitrot_chunk_source(source_chunks, shard_size, checksum_algo, skip_verify)
}
/// Create a new BitrotWriterWrapper based on the provided parameters
///
/// # Parameters
@@ -705,7 +199,6 @@ pub async fn create_bitrot_writer(
#[cfg(test)]
mod tests {
use super::*;
use futures_util::StreamExt;
#[tokio::test]
async fn test_create_bitrot_reader_with_inline_data() {
@@ -756,246 +249,6 @@ mod tests {
assert!(result.unwrap().is_some());
}
#[tokio::test]
async fn test_create_bitrot_chunk_stream_with_inline_data() {
let shard_size = 4;
let checksum_algo = HashAlgorithm::HighwayHash256S;
let shard1 = b"abcd";
let shard2 = b"ef";
let mut encoded = Vec::new();
encoded.extend_from_slice(checksum_algo.hash_encode(shard1).as_ref());
encoded.extend_from_slice(shard1);
encoded.extend_from_slice(checksum_algo.hash_encode(shard2).as_ref());
encoded.extend_from_slice(shard2);
let mut stream = create_bitrot_chunk_stream(
Some(&encoded),
None,
"test-bucket",
"test-path",
0,
shard1.len() + shard2.len(),
shard1.len() + shard2.len(),
shard_size,
checksum_algo,
false,
false,
)
.await
.unwrap()
.unwrap()
.stream;
let mut collected = Vec::new();
while let Some(chunk) = stream.next().await {
collected.extend_from_slice(&chunk.unwrap().as_bytes());
}
assert_eq!(collected, b"abcdef");
}
#[tokio::test]
async fn test_create_bitrot_chunk_stream_detects_hash_mismatch() {
let shard_size = 4;
let checksum_algo = HashAlgorithm::HighwayHash256S;
let shard = b"abcd";
let mut encoded = Vec::new();
let mut bad_hash = checksum_algo.hash_encode(shard).as_ref().to_vec();
bad_hash[0] ^= 0xFF;
encoded.extend_from_slice(&bad_hash);
encoded.extend_from_slice(shard);
let result = create_bitrot_chunk_stream(
Some(&encoded),
None,
"test-bucket",
"test-path",
0,
shard.len(),
shard.len(),
shard_size,
checksum_algo,
false,
false,
)
.await;
let mut stream = result.unwrap().unwrap().stream;
let err = stream.next().await.unwrap().unwrap_err();
assert_eq!(err.kind(), std::io::ErrorKind::InvalidData);
assert!(err.to_string().contains("bitrot hash mismatch"));
}
#[tokio::test]
async fn test_create_bitrot_chunk_stream_trims_range_after_decode() {
let shard_size = 4;
let checksum_algo = HashAlgorithm::HighwayHash256S;
let shard1 = b"abcd";
let shard2 = b"efgh";
let mut encoded = Vec::new();
encoded.extend_from_slice(checksum_algo.hash_encode(shard1).as_ref());
encoded.extend_from_slice(shard1);
encoded.extend_from_slice(checksum_algo.hash_encode(shard2).as_ref());
encoded.extend_from_slice(shard2);
let mut stream = create_bitrot_chunk_stream(
Some(&encoded),
None,
"test-bucket",
"test-path",
1,
5,
shard1.len() + shard2.len(),
shard_size,
checksum_algo,
false,
false,
)
.await
.unwrap()
.unwrap()
.stream;
let mut collected = Vec::new();
while let Some(chunk) = stream.next().await {
collected.extend_from_slice(&chunk.unwrap().as_bytes());
}
assert_eq!(collected, b"bcdef");
}
#[test]
fn test_decode_bitrot_chunk_source_preserves_aligned_multi_chunk_slices() {
let shard_size = 4;
let checksum_algo = HashAlgorithm::Md5;
let shard1 = b"abcd";
let shard2 = b"efgh";
let mut encoded_chunk_one = Vec::new();
encoded_chunk_one.extend_from_slice(checksum_algo.hash_encode(shard1).as_ref());
encoded_chunk_one.extend_from_slice(shard1);
let mut encoded_chunk_two = Vec::new();
encoded_chunk_two.extend_from_slice(checksum_algo.hash_encode(shard2).as_ref());
encoded_chunk_two.extend_from_slice(shard2);
let source_chunks = vec![
IoChunk::Shared(Bytes::from(encoded_chunk_one)),
IoChunk::Shared(Bytes::from(encoded_chunk_two)),
];
let (decoded, copied) = decode_bitrot_chunk_source(&source_chunks, shard_size, checksum_algo, false).unwrap();
assert!(!copied, "frame-aligned multi-chunk source should not require aggregate copies");
assert_eq!(decoded.len(), 2);
assert_eq!(decoded[0].as_bytes(), Bytes::from_static(b"abcd"));
assert_eq!(decoded[1].as_bytes(), Bytes::from_static(b"efgh"));
}
#[test]
fn test_decode_bitrot_chunk_source_marks_cross_chunk_frame_as_copied() {
let shard_size = 4;
let checksum_algo = HashAlgorithm::Md5;
let shard1 = b"abcd";
let shard2 = b"efgh";
let hash1 = checksum_algo.hash_encode(shard1).as_ref().to_vec();
let hash2 = checksum_algo.hash_encode(shard2).as_ref().to_vec();
let mut encoded = Vec::new();
encoded.extend_from_slice(&hash1);
encoded.extend_from_slice(shard1);
encoded.extend_from_slice(&hash2);
encoded.extend_from_slice(shard2);
let split = hash1.len() + 2;
let source_chunks = vec![
IoChunk::Shared(Bytes::copy_from_slice(&encoded[..split])),
IoChunk::Shared(Bytes::copy_from_slice(&encoded[split..])),
];
let (decoded, copied) = decode_bitrot_chunk_source(&source_chunks, shard_size, checksum_algo, false).unwrap();
assert!(copied, "cross-chunk frame should be classified as requiring a copy");
assert_eq!(decoded.len(), 2);
assert_eq!(decoded[0].as_bytes(), Bytes::from_static(b"abcd"));
assert_eq!(decoded[1].as_bytes(), Bytes::from_static(b"efgh"));
}
#[test]
fn test_decode_bitrot_chunk_source_preserves_pooled_single_chunk_slice() {
let shard_size = 4;
let checksum_algo = HashAlgorithm::Md5;
let shard = b"abcd";
let mut encoded = Vec::new();
encoded.extend_from_slice(checksum_algo.hash_encode(shard).as_ref());
encoded.extend_from_slice(shard);
let source_chunks = vec![IoChunk::Pooled(rustfs_io_core::PooledChunk::from_vec(encoded))];
let (decoded, copied) = decode_bitrot_chunk_source(&source_chunks, shard_size, checksum_algo, false).unwrap();
assert!(!copied, "single pooled chunk slice should preserve provenance without copy");
assert_eq!(decoded.len(), 1);
assert!(matches!(&decoded[0], IoChunk::Pooled(_)));
assert_eq!(decoded[0].as_bytes(), Bytes::from_static(b"abcd"));
}
#[tokio::test]
async fn test_bitrot_chunk_source_marks_cross_chunk_take_as_copied() {
let source_stream: BoxChunkStream = Box::pin(stream::iter(vec![
Ok(IoChunk::Shared(Bytes::from_static(b"ab"))),
Ok(IoChunk::Shared(Bytes::from_static(b"cd"))),
]));
let mut source = BitrotChunkSource::new(source_stream, VecDeque::new(), false);
source.fill(4).await.expect("source fill should succeed");
let span = source.take_span(4).expect("cross-chunk take should succeed");
assert!(span.copied, "cross-chunk take should be classified as copied");
assert_eq!(span.bytes, Bytes::from_static(b"abcd"));
assert_eq!(span.chunk.as_bytes(), Bytes::from_static(b"abcd"));
}
#[tokio::test]
async fn test_bitrot_chunk_stream_state_yields_verified_prefix_before_later_truncation() {
let shard_size = 4;
let checksum_algo = HashAlgorithm::Md5;
let shard1 = b"abcd";
let shard2 = b"efgh";
let mut first_frame = Vec::new();
first_frame.extend_from_slice(checksum_algo.hash_encode(shard1).as_ref());
first_frame.extend_from_slice(shard1);
let mut second_frame_prefix = Vec::new();
second_frame_prefix.extend_from_slice(checksum_algo.hash_encode(shard2).as_ref());
second_frame_prefix.extend_from_slice(&shard2[..2]);
let source_stream: BoxChunkStream = Box::pin(stream::iter(vec![
Ok(IoChunk::Shared(Bytes::from(first_frame))),
Ok(IoChunk::Shared(Bytes::from(second_frame_prefix))),
]));
let mut state = BitrotChunkStreamState::new(
source_stream,
VecDeque::new(),
false,
shard1.len() + shard2.len(),
0,
shard1.len() + shard2.len(),
shard_size,
checksum_algo,
false,
);
let first = state.next_verified_chunk().await.unwrap().unwrap();
assert_eq!(first.as_bytes(), Bytes::from_static(b"abcd"));
let err = state.next_verified_chunk().await.unwrap_err();
assert_eq!(err.kind(), std::io::ErrorKind::UnexpectedEof);
assert!(err.to_string().contains("truncated bitrot chunk source"));
}
#[tokio::test]
async fn test_create_bitrot_reader_with_inline_offset_starts_at_requested_shard() {
let shard_size = 4;
crates/ecstore/src/set_disk.rs
-3
View File
@@ -249,9 +249,6 @@ mod read;
mod replication;
mod write;
#[doc(hidden)]
pub use read::collect_direct_data_shard_chunks_for_benchmark;
/// Get lock acquire timeout from environment variable RUSTFS_LOCK_ACQUIRE_TIMEOUT (in seconds)
/// Defaults to 30 seconds if not set or invalid
pub fn get_lock_acquire_timeout() -> Duration {
crates/ecstore/src/set_disk/read.rs
+2 -775
View File
@@ -13,14 +13,11 @@
// limitations under the License.
use super::*;
use crate::bitrot::create_bitrot_chunk_stream;
use crate::erasure_coding::decode::ErasureChunkDecoder;
use crate::erasure_coding::{calc_shard_size, calc_shard_size_legacy};
use crate::store_api::{GetObjectChunkCopyMode, GetObjectChunkPath, GetObjectChunkResult};
use bytes::BytesMut;
use futures_util::{Stream, StreamExt, stream};
use futures_util::Stream;
use rustfs_config::{DEFAULT_OBJECT_ZERO_COPY_ENABLE, ENV_OBJECT_ZERO_COPY_ENABLE};
use rustfs_io_core::{BoxChunkStream, IoChunk};
use rustfs_io_core::IoChunk;
use std::io;
use std::pin::Pin;
use std::sync::Mutex;
@@ -40,56 +37,6 @@ impl ChannelChunkStream {
}
}
struct DirectShardCursor {
stream: BoxChunkStream,
current_chunk: Option<IoChunk>,
current_offset: usize,
}
impl DirectShardCursor {
fn new(stream: BoxChunkStream) -> Self {
Self {
stream,
current_chunk: None,
current_offset: 0,
}
}
fn current_remaining(&self) -> usize {
self.current_chunk
.as_ref()
.map(|chunk| chunk.len().saturating_sub(self.current_offset))
.unwrap_or(0)
}
fn consume_current(&mut self, len: usize) {
self.current_offset += len;
if self.current_remaining() == 0 {
self.current_chunk = None;
self.current_offset = 0;
}
}
async fn ensure_chunk(&mut self, shard_index: usize) -> io::Result<bool> {
if self.current_chunk.is_some() && self.current_remaining() > 0 {
return Ok(true);
}
match self.stream.next().await {
Some(Ok(chunk)) => {
self.current_chunk = Some(chunk);
self.current_offset = 0;
Ok(true)
}
Some(Err(err)) => Err(err),
None => {
debug!(shard_index, "direct shard cursor reached EOF");
Ok(false)
}
}
}
}
impl Stream for ChannelChunkStream {
type Item = io::Result<IoChunk>;
@@ -157,17 +104,6 @@ impl AsyncWrite for ChannelChunkWriter {
}
}
fn merge_chunk_copy_mode(current: GetObjectChunkCopyMode, next: GetObjectChunkCopyMode) -> GetObjectChunkCopyMode {
use GetObjectChunkCopyMode::{Reconstructed, SharedBytes, SingleCopy, TrueZeroCopy};
match (current, next) {
(Reconstructed, _) | (_, Reconstructed) => Reconstructed,
(SingleCopy, _) | (_, SingleCopy) => SingleCopy,
(SharedBytes, _) | (_, SharedBytes) => SharedBytes,
(TrueZeroCopy, TrueZeroCopy) => TrueZeroCopy,
}
}
fn multipart_logical_part_size(fi: &FileInfo, part_index: usize) -> usize {
let part = &fi.parts[part_index];
if part.actual_size > 0 {
@@ -277,616 +213,7 @@ fn direct_block_shard_size(
}
}
#[allow(clippy::too_many_arguments)]
async fn send_direct_data_shard_chunks(
sender: UnboundedSender<io::Result<IoChunk>>,
shard_streams: Vec<BoxChunkStream>,
data_shards: usize,
block_size: usize,
total_size: usize,
uses_legacy: bool,
offset: usize,
length: usize,
) {
if length == 0 {
return;
}
let start_block = offset / block_size;
let end_block = offset.saturating_add(length.saturating_sub(1)) / block_size;
let mut shard_cursors = shard_streams.into_iter().map(DirectShardCursor::new).collect::<Vec<_>>();
for block_index in start_block..=end_block {
let (block_offset, block_length) = block_window(offset, length, block_size, block_index, start_block, end_block);
if block_length == 0 {
break;
}
let shard_block_size = direct_block_shard_size(total_size, block_size, data_shards, block_index, uses_legacy);
let mut write_left = block_length;
let mut skip = block_offset;
for (shard_index, shard_cursor) in shard_cursors.iter_mut().enumerate().take(data_shards) {
let mut shard_block_left = shard_block_size;
while shard_block_left > 0 {
let has_chunk = match shard_cursor.ensure_chunk(shard_index).await {
Ok(has_chunk) => has_chunk,
Err(err) => {
let _ = sender.send(Err(err));
return;
}
};
if !has_chunk {
let _ = sender.send(Err(io::Error::new(
io::ErrorKind::UnexpectedEof,
format!("missing chunk for data shard {shard_index}"),
)));
return;
}
let chunk = shard_cursor.current_chunk.as_ref().expect("chunk should exist after ensure");
let chunk_remaining = chunk.len().saturating_sub(shard_cursor.current_offset);
let take_from_chunk = chunk_remaining.min(shard_block_left);
if skip >= take_from_chunk {
skip -= take_from_chunk;
shard_cursor.consume_current(take_from_chunk);
shard_block_left -= take_from_chunk;
continue;
}
let start = shard_cursor.current_offset + skip;
let available = take_from_chunk.saturating_sub(skip);
let take = available.min(write_left);
let out_chunk = if start == shard_cursor.current_offset && take == take_from_chunk {
chunk.slice(start, take).expect("full remaining slice should succeed")
} else {
match chunk.slice(start, take) {
Ok(chunk) => chunk,
Err(err) => {
let _ = sender.send(Err(err));
return;
}
}
};
let consumed = skip + take;
skip = 0;
shard_cursor.consume_current(consumed);
shard_block_left -= consumed;
if sender.send(Ok(out_chunk)).is_err() {
return;
}
write_left -= take;
if write_left == 0 {
break;
}
}
if write_left == 0 {
break;
}
}
if write_left != 0 {
let _ = sender.send(Err(io::Error::new(
io::ErrorKind::UnexpectedEof,
"not enough decoded shard data for requested block",
)));
return;
}
}
}
#[doc(hidden)]
pub async fn collect_direct_data_shard_chunks_for_benchmark(
shard_streams: Vec<BoxChunkStream>,
data_shards: usize,
block_size: usize,
total_size: usize,
uses_legacy: bool,
offset: usize,
length: usize,
) -> io::Result<Vec<IoChunk>> {
let (tx, rx) = unbounded_channel();
send_direct_data_shard_chunks(tx, shard_streams, data_shards, block_size, total_size, uses_legacy, offset, length).await;
let mut stream = ChannelChunkStream::new(rx);
let mut chunks = Vec::new();
while let Some(chunk) = stream.next().await {
chunks.push(chunk?);
}
Ok(chunks)
}
#[allow(clippy::too_many_arguments)]
async fn build_reconstructed_part_stream(
bucket: &str,
object: &str,
part_number: usize,
part_offset: usize,
part_length: usize,
part_size: usize,
read_offset: usize,
till_offset: usize,
files: &[FileInfo],
disks: &[Option<DiskStore>],
erasure: &erasure_coding::Erasure,
checksum_algo: rustfs_utils::HashAlgorithm,
skip_verify_bitrot: bool,
use_zero_copy: bool,
) -> Result<Option<BoxChunkStream>> {
let shard_length = till_offset.saturating_sub(read_offset);
let mut readers = Vec::with_capacity(disks.len());
let mut errors = Vec::with_capacity(disks.len());
for (idx, disk_op) in disks.iter().enumerate() {
match create_bitrot_reader(
files[idx].data.as_deref(),
disk_op.as_ref(),
bucket,
&format!("{}/{}/part.{}", object, files[idx].data_dir.unwrap_or_default(), part_number),
read_offset,
shard_length,
erasure.shard_size(),
checksum_algo.clone(),
skip_verify_bitrot,
use_zero_copy,
)
.await
{
Ok(Some(reader)) => {
readers.push(Some(reader));
errors.push(None);
}
Ok(None) => {
readers.push(None);
errors.push(Some(DiskError::DiskNotFound));
}
Err(err) => {
readers.push(None);
errors.push(Some(err));
}
}
}
let available_shards = errors.iter().filter(|error| error.is_none()).count();
if available_shards < erasure.data_shards {
return Ok(None);
}
let missing_shards = readers.len().saturating_sub(available_shards);
if missing_shards > 0 {
debug!(
bucket,
object,
part_number,
missing_shards,
available_shards,
data_shards = erasure.data_shards,
parity_shards = erasure.parity_shards,
"using reconstructed part stream for missing shards"
);
}
let (tx, rx) = unbounded_channel();
let bucket = bucket.to_string();
let object = object.to_string();
let erasure = erasure.clone();
tokio::spawn(async move {
let mut decoder = match ErasureChunkDecoder::new(erasure, readers, part_offset, part_length, part_size) {
Ok(decoder) => decoder,
Err(err) => {
let _ = tx.send(Err(err));
return;
}
};
loop {
match decoder.next_chunks().await {
Ok(Some(chunks)) => {
for chunk in chunks {
if tx.send(Ok(chunk)).is_err() {
return;
}
}
}
Ok(None) => break,
Err(err) => {
let _ = tx.send(Err(err));
return;
}
}
}
if let Some(err) = decoder.finish_error() {
let _ = tx.send(Err(err));
return;
}
if let Some(disk_err) = decoder.take_healable_error() {
let allow_heal_only =
decoder.written() == part_length && matches!(disk_err, DiskError::FileNotFound | DiskError::FileCorrupt);
if !allow_heal_only {
let _ = tx.send(Err(io::Error::other(disk_err.to_string())));
return;
}
debug!(
bucket,
object,
part_number,
bytes_written = decoder.written(),
error = %disk_err,
"reconstructed part completed with healable shard error"
);
}
});
Ok(Some(Box::pin(ChannelChunkStream::new(rx))))
}
impl SetDisks {
#[tracing::instrument(level = "debug", skip(self, h, opts))]
pub(crate) async fn get_object_chunks(
&self,
bucket: &str,
object: &str,
range: Option<HTTPRangeSpec>,
h: HeaderMap,
opts: &ObjectOptions,
) -> Result<GetObjectChunkResult> {
let lock_optimization_enabled = is_lock_optimization_enabled();
let read_lock_guard = if !opts.no_lock {
let acquire_start = Instant::now();
if is_deadlock_detection_enabled() {
debug!(
lock_id = format!("{}:{}", bucket, object),
lock_type = "read",
resource = format!("{}/{}", bucket, object),
"Waiting for read lock"
);
}
let guard = self
.new_ns_lock(bucket, object)
.await?
.get_read_lock(get_lock_acquire_timeout())
.await
.map_err(|e| {
Error::other(format!(
"Failed to acquire read lock: {}",
self.format_lock_error_from_error(bucket, object, "read", &e)
))
})?;
let _lock_id = record_lock_acquire(bucket, object, "read");
metrics::counter!("rustfs.lock.acquire.total", "type" => "read").increment(1);
metrics::histogram!("rustfs.lock.acquire.duration.seconds").record(acquire_start.elapsed().as_secs_f64());
Some(guard)
} else {
None
};
let (fi, files, disks) = self
.get_object_fileinfo(bucket, object, opts, true)
.await
.map_err(|err| to_object_err(err, vec![bucket, object]))?;
let object_info = ObjectInfo::from_file_info(&fi, bucket, object, opts.versioned || opts.version_suspended);
if object_info.delete_marker {
if opts.version_id.is_none() {
return Err(to_object_err(Error::FileNotFound, vec![bucket, object]));
}
return Err(to_object_err(Error::MethodNotAllowed, vec![bucket, object]));
}
if object_info.size == 0 {
return Ok(GetObjectChunkResult {
stream: Box::pin(stream::iter(Vec::<io::Result<IoChunk>>::new())),
path: GetObjectChunkPath::Direct,
copy_mode: GetObjectChunkCopyMode::SharedBytes,
});
}
let (bridge_offset, bridge_length) = if fi.parts.is_empty() {
(0, fi.size)
} else {
let total_size = multipart_logical_total_size(&fi);
if let Some(range) = &range {
let (offset, length) = range
.get_offset_length(total_size as i64)
.map_err(|err| to_object_err(err, vec![bucket, object]))?;
(offset, length)
} else {
(0, total_size as i64)
}
};
if object_info.is_remote() {
let mut opts = opts.clone();
if object_info.parts.len() == 1 {
opts.part_number = Some(1);
}
let gr = get_transitioned_object_reader(bucket, object, &range, &h, &object_info, &opts).await?;
let stream = ReaderStream::new(gr.stream).map(|result| result.map(IoChunk::Shared));
return Ok(GetObjectChunkResult {
stream: Box::pin(stream),
path: GetObjectChunkPath::Bridge,
copy_mode: GetObjectChunkCopyMode::SingleCopy,
});
}
if fi.erasure.data_blocks > 0 {
let (disks, files) = Self::shuffle_disks_and_parts_metadata_by_index(&disks, &files, &fi);
let total_size = multipart_logical_total_size(&fi);
let requested_length = if let Some(range) = &range {
let (offset, length) = range
.get_offset_length(total_size as i64)
.map_err(|err| to_object_err(err, vec![bucket, object]))?;
(offset, length as usize)
} else {
(0, total_size)
};
let (part_index, mut part_offset) = multipart_to_logical_part_offset(&fi, requested_length.0)?;
let mut end_offset = requested_length.0;
if requested_length.1 > 0 {
end_offset += requested_length.1 - 1;
}
let (last_part_index, _) = multipart_to_logical_part_offset(&fi, end_offset)?;
let use_zero_copy = rustfs_utils::get_env_bool(ENV_OBJECT_ZERO_COPY_ENABLE, DEFAULT_OBJECT_ZERO_COPY_ENABLE);
let single_shard_file = if fi.erasure.data_blocks == 1 {
Some(
files
.first()
.ok_or_else(|| Error::other("single-shard multipart metadata missing"))?,
)
} else {
None
};
let single_shard_disk = if fi.erasure.data_blocks == 1 {
Some(
disks
.first()
.ok_or_else(|| Error::other("single-shard multipart disk slot missing"))?,
)
} else {
None
};
let mut part_streams = Vec::new();
let mut part_total_read = 0usize;
let mut merged_copy_mode = GetObjectChunkCopyMode::TrueZeroCopy;
for current_part in part_index..=last_part_index {
let part_number = fi.parts[current_part].number;
let part_size = multipart_logical_part_size(&fi, current_part);
let mut part_length = part_size - part_offset;
if part_length > (requested_length.1 - part_total_read) {
part_length = requested_length.1 - part_total_read;
}
let checksum_info = fi.erasure.get_checksum_info(part_number);
let checksum_algo =
if fi.uses_legacy_checksum && checksum_info.algorithm == rustfs_utils::HashAlgorithm::HighwayHash256S {
rustfs_utils::HashAlgorithm::HighwayHash256SLegacy
} else {
checksum_info.algorithm.clone()
};
if fi.erasure.data_blocks == 1 {
let single_shard_file = single_shard_file.expect("single-shard multipart metadata must exist");
let single_shard_disk = single_shard_disk.expect("single-shard multipart disk slot must exist");
let data_dir = single_shard_file
.data_dir
.as_ref()
.map(uuid::Uuid::to_string)
.unwrap_or_default();
let data_path = format!("{}/{}/part.{}", object, data_dir, part_number);
let chunk_result = create_bitrot_chunk_stream(
single_shard_file.data.as_deref(),
single_shard_disk.as_ref(),
bucket,
&data_path,
part_offset,
part_length,
part_size,
fi.erasure.shard_size(),
checksum_algo,
opts.skip_verify_bitrot,
use_zero_copy,
)
.await?;
let Some(chunk_result) = chunk_result else {
part_streams.clear();
break;
};
merged_copy_mode = merge_chunk_copy_mode(merged_copy_mode, chunk_result.copy_mode);
part_streams.push(chunk_result.stream);
} else {
let erasure = erasure_coding::Erasure::new_with_options(
fi.erasure.data_blocks,
fi.erasure.parity_blocks,
fi.erasure.block_size,
fi.uses_legacy_checksum,
);
let read_offset = (part_offset / erasure.block_size) * erasure.shard_size();
let till_offset = erasure.shard_file_offset(part_offset, part_length, part_size);
let shard_length = till_offset.saturating_sub(read_offset);
let shard_total_size = erasure.shard_file_size(part_size as i64) as usize;
let mut shard_streams = Vec::with_capacity(erasure.data_shards);
let mut part_copy_mode = GetObjectChunkCopyMode::TrueZeroCopy;
let mut needs_reconstruct = false;
for shard_index in 0..erasure.data_shards {
let data_path =
format!("{}/{}/part.{}", object, files[shard_index].data_dir.unwrap_or_default(), part_number);
let chunk_result = match create_bitrot_chunk_stream(
files[shard_index].data.as_deref(),
disks[shard_index].as_ref(),
bucket,
&data_path,
read_offset,
shard_length,
shard_total_size,
erasure.shard_size(),
checksum_algo.clone(),
opts.skip_verify_bitrot,
use_zero_copy,
)
.await
{
Ok(Some(chunk_result)) => chunk_result,
Ok(None) => {
needs_reconstruct = true;
shard_streams.clear();
break;
}
Err(err) => {
debug!(
bucket,
object,
part_number,
shard_index,
error = %err,
"multi-shard direct chunk path unavailable, falling back to decoded read path"
);
needs_reconstruct = true;
shard_streams.clear();
break;
}
};
part_copy_mode = merge_chunk_copy_mode(part_copy_mode, chunk_result.copy_mode);
shard_streams.push(chunk_result.stream);
}
if needs_reconstruct {
let reconstructed_stream = match build_reconstructed_part_stream(
bucket,
object,
part_number,
part_offset,
part_length,
part_size,
read_offset,
till_offset,
&files,
&disks,
&erasure,
checksum_algo,
opts.skip_verify_bitrot,
use_zero_copy,
)
.await?
{
Some(stream) => stream,
None => {
part_streams.clear();
break;
}
};
merged_copy_mode = merge_chunk_copy_mode(merged_copy_mode, GetObjectChunkCopyMode::Reconstructed);
part_streams.push(reconstructed_stream);
part_total_read += part_length;
part_offset = 0;
continue;
}
if shard_streams.len() != erasure.data_shards {
part_streams.clear();
break;
}
let (tx, rx) = unbounded_channel();
tokio::spawn(send_direct_data_shard_chunks(
tx,
shard_streams,
erasure.data_shards,
erasure.block_size,
part_size,
fi.uses_legacy_checksum,
part_offset,
part_length,
));
merged_copy_mode = merge_chunk_copy_mode(merged_copy_mode, part_copy_mode);
part_streams.push(Box::pin(ChannelChunkStream::new(rx)));
}
part_total_read += part_length;
part_offset = 0;
}
if !part_streams.is_empty() {
return Ok(GetObjectChunkResult {
stream: Box::pin(stream::iter(part_streams).flatten()),
path: GetObjectChunkPath::Direct,
copy_mode: merged_copy_mode,
});
}
}
let read_lock_guard = if lock_optimization_enabled {
if read_lock_guard.is_some() {
let lock_id = format!("{}:{}", bucket, object);
record_lock_release(bucket, object, &lock_id, "read");
metrics::counter!("rustfs.lock.release.early.total", "type" => "read").increment(1);
}
drop(read_lock_guard);
debug!(bucket, object, "Lock optimization: released read lock after metadata read");
None
} else {
read_lock_guard
};
let chunk_size = get_duplex_buffer_size();
let bucket = bucket.to_owned();
let object = object.to_owned();
let set_index = self.set_index;
let pool_index = self.pool_index;
let skip_verify = opts.skip_verify_bitrot;
let (tx, rx) = unbounded_channel();
tokio::spawn(async move {
let _guard = read_lock_guard;
let mut writer = ChannelChunkWriter::new(tx, chunk_size);
if let Err(err) = Self::get_object_with_fileinfo(
&bucket,
&object,
bridge_offset,
bridge_length,
&mut writer,
fi,
files,
&disks,
set_index,
pool_index,
skip_verify,
)
.await
{
error!("get_object_with_fileinfo {bucket}/{object} err {:?}", err);
writer.send_error(io::Error::other(err.to_string()));
}
if let Err(err) = writer.finish() {
debug!(bucket, object, error = %err, "failed to flush chunk writer");
}
});
Ok(GetObjectChunkResult {
stream: Box::pin(ChannelChunkStream::new(rx)),
path: GetObjectChunkPath::Bridge,
copy_mode: GetObjectChunkCopyMode::SingleCopy,
})
}
pub(super) async fn read_parts(
disks: &[Option<DiskStore>],
bucket: &str,
@@ -1724,103 +1051,3 @@ impl SetDisks {
Ok(())
}
}
#[cfg(test)]
mod tests {
use super::*;
use bytes::Bytes;
use futures_util::StreamExt;
#[tokio::test]
async fn send_direct_data_shard_chunks_reassembles_multi_block_range() {
let data_shards = 4;
let block_size = 16;
let shard_streams: Vec<BoxChunkStream> = vec![
Box::pin(stream::iter(vec![
Ok(IoChunk::Shared(Bytes::copy_from_slice(&[0, 1, 2, 3]))),
Ok(IoChunk::Shared(Bytes::copy_from_slice(&[16, 17, 18, 19]))),
])),
Box::pin(stream::iter(vec![
Ok(IoChunk::Shared(Bytes::copy_from_slice(&[4, 5, 6, 7]))),
Ok(IoChunk::Shared(Bytes::copy_from_slice(&[20, 21, 22, 23]))),
])),
Box::pin(stream::iter(vec![
Ok(IoChunk::Shared(Bytes::copy_from_slice(&[8, 9, 10, 11]))),
Ok(IoChunk::Shared(Bytes::copy_from_slice(&[24, 25, 26, 27]))),
])),
Box::pin(stream::iter(vec![
Ok(IoChunk::Shared(Bytes::copy_from_slice(&[12, 13, 14, 15]))),
Ok(IoChunk::Shared(Bytes::copy_from_slice(&[28, 29, 30, 31]))),
])),
];
let (tx, rx) = unbounded_channel();
send_direct_data_shard_chunks(tx, shard_streams, data_shards, block_size, 32, false, 3, 18).await;
let mut stream = ChannelChunkStream::new(rx);
let mut collected = Vec::new();
while let Some(chunk) = stream.next().await {
collected.extend_from_slice(&chunk.unwrap().as_bytes());
}
assert_eq!(collected, (3u8..21).collect::<Vec<_>>());
}
#[tokio::test]
async fn send_direct_data_shard_chunks_keeps_block_boundaries_with_cross_block_chunks() {
let data_shards = 4;
let block_size = 16;
let shard_streams: Vec<BoxChunkStream> = vec![
Box::pin(stream::iter(vec![Ok(IoChunk::Shared(Bytes::copy_from_slice(&[
0, 1, 2, 3, 16, 17, 18, 19,
])))])),
Box::pin(stream::iter(vec![Ok(IoChunk::Shared(Bytes::copy_from_slice(&[
4, 5, 6, 7, 20, 21, 22, 23,
])))])),
Box::pin(stream::iter(vec![Ok(IoChunk::Shared(Bytes::copy_from_slice(&[
8, 9, 10, 11, 24, 25, 26, 27,
])))])),
Box::pin(stream::iter(vec![Ok(IoChunk::Shared(Bytes::copy_from_slice(&[
12, 13, 14, 15, 28, 29, 30, 31,
])))])),
];
let (tx, rx) = unbounded_channel();
send_direct_data_shard_chunks(tx, shard_streams, data_shards, block_size, 32, false, 3, 18).await;
let mut stream = ChannelChunkStream::new(rx);
let mut collected = Vec::new();
while let Some(chunk) = stream.next().await {
collected.extend_from_slice(&chunk.unwrap().as_bytes());
}
assert_eq!(collected, (3u8..21).collect::<Vec<_>>());
}
#[tokio::test]
async fn send_direct_data_shard_chunks_keeps_final_full_block_when_length_is_block_aligned() {
let data_shards = 2;
let block_size = 16;
let shard_streams: Vec<BoxChunkStream> = vec![
Box::pin(stream::iter(vec![
Ok(IoChunk::Shared(Bytes::copy_from_slice(&[0, 1, 2, 3, 4, 5, 6, 7]))),
Ok(IoChunk::Shared(Bytes::copy_from_slice(&[16, 17, 18, 19, 20, 21, 22, 23]))),
])),
Box::pin(stream::iter(vec![
Ok(IoChunk::Shared(Bytes::copy_from_slice(&[8, 9, 10, 11, 12, 13, 14, 15]))),
Ok(IoChunk::Shared(Bytes::copy_from_slice(&[24, 25, 26, 27, 28, 29, 30, 31]))),
])),
];
let (tx, rx) = unbounded_channel();
send_direct_data_shard_chunks(tx, shard_streams, data_shards, block_size, 32, false, 0, 32).await;
let mut stream = ChannelChunkStream::new(rx);
let mut collected = Vec::new();
while let Some(chunk) = stream.next().await {
collected.extend_from_slice(&chunk.unwrap().as_bytes());
}
assert_eq!(collected, (0u8..32).collect::<Vec<_>>());
}
}
crates/ecstore/src/sets.rs
+1 -14
View File
@@ -15,7 +15,7 @@
use crate::disk::error_reduce::count_errs;
use crate::error::{Error, Result};
use crate::store_api::{GetObjectChunkResult, ListPartsInfo, ObjectInfoOrErr, WalkOptions};
use crate::store_api::{ListPartsInfo, ObjectInfoOrErr, WalkOptions};
use crate::{
disk::{
DiskAPI, DiskInfo, DiskOption, DiskStore,
@@ -287,19 +287,6 @@ impl Sets {
self.get_disks(self.get_hashed_set_index(key))
}
pub async fn get_object_chunks(
&self,
bucket: &str,
object: &str,
range: Option<HTTPRangeSpec>,
h: HeaderMap,
opts: &ObjectOptions,
) -> Result<GetObjectChunkResult> {
self.get_disks_by_key(object)
.get_object_chunks(bucket, object, range, h, opts)
.await
}
fn get_hashed_set_index(&self, input: &str) -> usize {
match self.distribution_algo {
DistributionAlgoVersion::V1 => crc_hash(input, self.disk_set.len()),
crates/ecstore/src/store.rs
+3 -17
View File
@@ -60,9 +60,9 @@ use crate::{
sets::Sets,
store_api::{
BucketInfo, BucketOperations, BucketOptions, ChunkNativePutData, CompletePart, DeleteBucketOptions, DeletedObject,
GetObjectChunkResult, GetObjectReader, HTTPRangeSpec, HealOperations, ListObjectsV2Info, ListOperations,
MakeBucketOptions, MultipartOperations, MultipartUploadResult, ObjectInfo, ObjectOperations, ObjectOptions,
ObjectToDelete, PartInfo, StorageAPI,
GetObjectReader, HTTPRangeSpec, HealOperations, ListObjectsV2Info, ListOperations, MakeBucketOptions,
MultipartOperations, MultipartUploadResult, ObjectInfo, ObjectOperations, ObjectOptions, ObjectToDelete, PartInfo,
StorageAPI,
},
store_init,
};
@@ -271,20 +271,6 @@ impl ObjectIO for ECStore {
}
}
impl ECStore {
#[instrument(level = "debug", skip(self))]
pub async fn get_object_chunks(
&self,
bucket: &str,
object: &str,
range: Option<HTTPRangeSpec>,
h: HeaderMap,
opts: &ObjectOptions,
) -> Result<GetObjectChunkResult> {
self.handle_get_object_chunks(bucket, object, range, h, opts).await
}
}
lazy_static! {
static ref enableObjcetLockConfig: ObjectLockConfiguration = ObjectLockConfiguration {
object_lock_enabled: Some(ObjectLockEnabled::from_static(ObjectLockEnabled::ENABLED)),
crates/ecstore/src/store/object.rs
-30
View File
@@ -13,8 +13,6 @@
// limitations under the License.
use super::*;
use crate::store_api::GetObjectChunkResult;
fn select_data_movement_target_pool(
existing_pool_idx: Result<usize>,
src_pool_idx: usize,
@@ -214,34 +212,6 @@ impl ECStore {
.await
}
#[instrument(level = "debug", skip(self))]
pub(super) async fn handle_get_object_chunks(
&self,
bucket: &str,
object: &str,
range: Option<HTTPRangeSpec>,
h: HeaderMap,
opts: &ObjectOptions,
) -> Result<GetObjectChunkResult> {
check_get_obj_args(bucket, object)?;
let object = encode_dir_object(object);
if self.single_pool() {
return self.pools[0].get_object_chunks(bucket, object.as_str(), range, h, opts).await;
}
let mut opts = opts.clone();
opts.no_lock = true;
let (_, idx) = self
.get_latest_accessible_object_info_with_idx(bucket, &object, &opts)
.await?;
self.pools[idx]
.get_object_chunks(bucket, object.as_str(), range, h, &opts)
.await
}
#[instrument(level = "debug", skip(self, data))]
pub(super) async fn handle_put_object(
&self,
crates/ecstore/src/store_api.rs
-1
View File
@@ -31,7 +31,6 @@ use rustfs_filemeta::{
RestoreStatusOps as _, VersionPurgeStatusType, parse_restore_obj_status, replication_statuses_map,
version_purge_statuses_map,
};
use rustfs_io_core::BoxChunkStream;
use rustfs_lock::NamespaceLockWrapper;
use rustfs_madmin::heal_commands::HealResultItem;
use rustfs_rio::Checksum;
crates/ecstore/src/store_api/readers.rs
-20
View File
@@ -189,26 +189,6 @@ pub struct GetObjectReader {
pub object_info: ObjectInfo,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum GetObjectChunkPath {
Direct,
Bridge,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum GetObjectChunkCopyMode {
TrueZeroCopy,
SharedBytes,
SingleCopy,
Reconstructed,
}
pub struct GetObjectChunkResult {
pub stream: BoxChunkStream,
pub path: GetObjectChunkPath,
pub copy_mode: GetObjectChunkCopyMode,
}
impl GetObjectReader {
#[tracing::instrument(level = "debug", skip(reader, rs, opts, _h))]
pub fn new(
crates/io-metrics/src/lib.rs
-3
View File
@@ -193,7 +193,6 @@ impl IoStage {
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum FallbackReason {
Unknown,
FeatureDisabled,
ProbeFailed,
MmapDisabled,
MmapUnavailable,
@@ -215,7 +214,6 @@ impl FallbackReason {
pub const fn as_str(self) -> &'static str {
match self {
Self::Unknown => "unknown",
Self::FeatureDisabled => "feature_disabled",
Self::ProbeFailed => "probe_failed",
Self::MmapDisabled => "mmap_disabled",
Self::MmapUnavailable => "mmap_unavailable",
@@ -932,7 +930,6 @@ mod tests {
#[test]
fn test_fallback_reason_as_str_values_stable() {
assert_eq!(FallbackReason::Unknown.as_str(), "unknown");
assert_eq!(FallbackReason::FeatureDisabled.as_str(), "feature_disabled");
assert_eq!(FallbackReason::ProbeFailed.as_str(), "probe_failed");
assert_eq!(FallbackReason::MmapDisabled.as_str(), "mmap_disabled");
assert_eq!(FallbackReason::MmapUnavailable.as_str(), "mmap_unavailable");
crates/object-io/src/get.rs
+1 -479
View File
@@ -13,14 +13,10 @@
// limitations under the License.
use bytes::Bytes;
use futures_util::StreamExt;
use http::HeaderMap;
use http::header::{CACHE_CONTROL, CONTENT_DISPOSITION, CONTENT_LANGUAGE};
use rustfs_ecstore::bucket::lifecycle::lifecycle::TRANSITION_COMPLETE;
use rustfs_ecstore::client::object_api_utils::to_s3s_etag;
use rustfs_ecstore::error::StorageError;
use rustfs_ecstore::store_api::{GetObjectChunkCopyMode, GetObjectChunkPath, GetObjectChunkResult, HTTPRangeSpec, ObjectInfo};
use rustfs_io_core::BoxChunkStream;
use rustfs_ecstore::store_api::{HTTPRangeSpec, ObjectInfo};
use rustfs_rio::Reader;
use rustfs_s3select_api::object_store::bytes_stream;
use rustfs_utils::http::{AMZ_CHECKSUM_MODE, AMZ_CHECKSUM_TYPE};
@@ -113,55 +109,6 @@ where
)))
}
pub fn build_chunk_blob(chunk_stream: BoxChunkStream) -> Option<StreamingBlob> {
Some(StreamingBlob::wrap(chunk_stream.map(|result| result.map(|chunk| chunk.as_bytes()))))
}
/// ADR-facing alias for the chunk-stream to HTTP body bridge.
pub fn build_chunk_http_body(chunk_stream: BoxChunkStream) -> Option<StreamingBlob> {
build_chunk_blob(chunk_stream)
}
pub fn map_chunk_copy_mode(copy_mode: GetObjectChunkCopyMode) -> rustfs_io_metrics::CopyMode {
match copy_mode {
GetObjectChunkCopyMode::TrueZeroCopy => rustfs_io_metrics::CopyMode::TrueZeroCopy,
GetObjectChunkCopyMode::SharedBytes => rustfs_io_metrics::CopyMode::SharedBytes,
GetObjectChunkCopyMode::SingleCopy => rustfs_io_metrics::CopyMode::SingleCopy,
GetObjectChunkCopyMode::Reconstructed => rustfs_io_metrics::CopyMode::Reconstructed,
}
}
pub fn chunk_body_data_plane_labels(
path: GetObjectChunkPath,
copy_mode: rustfs_io_metrics::CopyMode,
) -> (rustfs_io_metrics::IoPath, rustfs_io_metrics::CopyMode) {
(
match path {
GetObjectChunkPath::Direct | GetObjectChunkPath::Bridge => rustfs_io_metrics::IoPath::Fast,
},
copy_mode,
)
}
#[must_use]
pub const fn get_object_chunk_path_label(path: GetObjectChunkPath) -> &'static str {
match path {
GetObjectChunkPath::Direct => "direct",
GetObjectChunkPath::Bridge => "bridge",
}
}
pub fn get_object_chunk_fast_path_guard(
has_sse_customer_key: bool,
has_sse_customer_key_md5: bool,
) -> Result<(), ChunkReadFallback> {
if has_sse_customer_key || has_sse_customer_key_md5 {
return Err(ChunkReadFallback::read_setup(rustfs_io_metrics::FallbackReason::EncryptionEnabled));
}
Ok(())
}
pub fn get_object_sequential_hint(rs: Option<&HTTPRangeSpec>) -> bool {
if rs.is_none() {
true
@@ -196,11 +143,6 @@ pub struct GetObjectBodyPlanningInputs {
pub enum GetObjectBodySource {
Reader(Box<dyn Reader>),
Chunk {
stream: BoxChunkStream,
path: GetObjectChunkPath,
copy_mode: rustfs_io_metrics::CopyMode,
},
}
pub struct GetObjectReadSetup {
@@ -261,11 +203,6 @@ pub struct GetObjectEncryptionState {
pub response_content_length_override: Option<i64>,
}
pub struct ChunkReadSetupResult {
pub read_setup: GetObjectReadSetup,
pub io_path: rustfs_io_metrics::IoPath,
}
#[derive(Debug, thiserror::Error)]
pub enum MaterializeGetObjectBodyError {
#[error("failed to read decrypted object: {0}")]
@@ -590,55 +527,6 @@ pub fn build_get_object_output_context(
}
}
#[allow(clippy::too_many_arguments)]
fn build_chunk_read_setup(
info: ObjectInfo,
event_info: ObjectInfo,
path: GetObjectChunkPath,
copy_mode: rustfs_io_metrics::CopyMode,
stream: BoxChunkStream,
plan: ChunkReadPlan,
) -> GetObjectReadSetup {
let ChunkReadPlan {
rs,
content_type,
last_modified,
response_content_length,
content_range,
} = plan;
GetObjectReadSetup {
info,
event_info,
body_source: GetObjectBodySource::Chunk { stream, path, copy_mode },
rs,
content_type,
last_modified,
response_content_length,
content_range,
server_side_encryption: None,
sse_customer_algorithm: None,
sse_customer_key_md5: None,
ssekms_key_id: None,
encryption_applied: false,
}
}
pub fn finalize_chunk_read_setup(
info: ObjectInfo,
event_info: ObjectInfo,
chunk_result: GetObjectChunkResult,
plan: ChunkReadPlan,
) -> ChunkReadSetupResult {
let copy_mode = map_chunk_copy_mode(chunk_result.copy_mode);
let (io_path, _) = chunk_body_data_plane_labels(chunk_result.path, copy_mode);
ChunkReadSetupResult {
io_path,
read_setup: build_chunk_read_setup(info, event_info, chunk_result.path, copy_mode, chunk_result.stream, plan),
}
}
#[allow(clippy::too_many_arguments)]
pub fn build_get_object_output(
body: Option<StreamingBlob>,
@@ -684,348 +572,16 @@ pub fn build_get_object_output(
}
}
#[derive(Debug)]
pub struct ChunkReadPlan {
pub rs: Option<HTTPRangeSpec>,
pub content_type: Option<ContentType>,
pub last_modified: Option<Timestamp>,
pub response_content_length: i64,
pub content_range: Option<String>,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct ChunkReadFallback {
pub stage: rustfs_io_metrics::IoStage,
pub reason: rustfs_io_metrics::FallbackReason,
}
impl ChunkReadFallback {
pub const fn new(stage: rustfs_io_metrics::IoStage, reason: rustfs_io_metrics::FallbackReason) -> Self {
Self { stage, reason }
}
pub const fn read_setup(reason: rustfs_io_metrics::FallbackReason) -> Self {
Self::new(rustfs_io_metrics::IoStage::ReadSetup, reason)
}
pub const fn range_guard(reason: rustfs_io_metrics::FallbackReason) -> Self {
Self::new(rustfs_io_metrics::IoStage::RangeGuard, reason)
}
}
#[derive(Debug)]
pub enum ChunkReadDecision {
Eligible(ChunkReadPlan),
Fallback(ChunkReadFallback),
}
#[derive(Debug)]
pub enum ChunkReadPlanError {
NoSuchKey,
MethodNotAllowed,
Io(std::io::Error),
}
impl From<std::io::Error> for ChunkReadPlanError {
fn from(value: std::io::Error) -> Self {
Self::Io(value)
}
}
impl From<StorageError> for ChunkReadPlanError {
fn from(value: StorageError) -> Self {
Self::Io(std::io::Error::other(value.to_string()))
}
}
pub fn get_object_chunk_range_guard(rs: Option<&HTTPRangeSpec>) -> Result<(), ChunkReadFallback> {
let Some(range_spec) = rs else {
return Ok(());
};
let unsupported = if range_spec.is_suffix_length {
range_spec.end != -1
} else {
range_spec.start < 0 || range_spec.end < -1 || (range_spec.end != -1 && range_spec.end < range_spec.start)
};
if unsupported {
return Err(ChunkReadFallback::range_guard(rustfs_io_metrics::FallbackReason::RangeNotSupported));
}
Ok(())
}
pub fn plan_chunk_read(
info: &ObjectInfo,
version_id_missing: bool,
rs: Option<HTTPRangeSpec>,
part_number: Option<usize>,
) -> Result<ChunkReadDecision, ChunkReadPlanError> {
if info.delete_marker {
if version_id_missing {
return Err(ChunkReadPlanError::NoSuchKey);
}
return Err(ChunkReadPlanError::MethodNotAllowed);
}
let (_, is_compressed) = info.is_compressed_ok()?;
if is_compressed {
return Ok(ChunkReadDecision::Fallback(ChunkReadFallback::read_setup(
rustfs_io_metrics::FallbackReason::CompressionEnabled,
)));
}
if info.transitioned_object.status == TRANSITION_COMPLETE {
return Ok(ChunkReadDecision::Fallback(ChunkReadFallback::read_setup(
rustfs_io_metrics::FallbackReason::NonLocalBackend,
)));
}
let has_encryption_metadata = info.user_defined.contains_key("x-rustfs-encryption-key")
|| info.user_defined.contains_key("x-amz-server-side-encryption")
|| info
.user_defined
.contains_key("x-amz-server-side-encryption-customer-algorithm");
if has_encryption_metadata {
return Ok(ChunkReadDecision::Fallback(ChunkReadFallback::read_setup(
rustfs_io_metrics::FallbackReason::EncryptionEnabled,
)));
}
let rs = resolve_requested_range(info, rs, part_number);
if let Err(fallback) = get_object_chunk_range_guard(rs.as_ref()) {
return Ok(ChunkReadDecision::Fallback(fallback));
}
let content_type = info
.content_type
.as_ref()
.and_then(|content_type| ContentType::from_str(content_type).ok());
let last_modified = info.mod_time.map(Timestamp::from);
let total_size = info.get_actual_size()?;
let (rs, response_content_length, content_range) = resolve_response_range(total_size, rs)?;
Ok(ChunkReadDecision::Eligible(ChunkReadPlan {
rs,
content_type,
last_modified,
response_content_length,
content_range,
}))
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn map_chunk_copy_mode_uses_expected_metric_modes() {
assert_eq!(
map_chunk_copy_mode(GetObjectChunkCopyMode::TrueZeroCopy),
rustfs_io_metrics::CopyMode::TrueZeroCopy
);
assert_eq!(
map_chunk_copy_mode(GetObjectChunkCopyMode::SharedBytes),
rustfs_io_metrics::CopyMode::SharedBytes
);
assert_eq!(
map_chunk_copy_mode(GetObjectChunkCopyMode::SingleCopy),
rustfs_io_metrics::CopyMode::SingleCopy
);
assert_eq!(
map_chunk_copy_mode(GetObjectChunkCopyMode::Reconstructed),
rustfs_io_metrics::CopyMode::Reconstructed
);
}
#[test]
fn chunk_body_labels_keep_fast_path_and_copy_mode() {
let (path, copy_mode) = chunk_body_data_plane_labels(GetObjectChunkPath::Bridge, rustfs_io_metrics::CopyMode::SingleCopy);
assert_eq!(path, rustfs_io_metrics::IoPath::Fast);
assert_eq!(copy_mode, rustfs_io_metrics::CopyMode::SingleCopy);
}
#[test]
fn get_object_chunk_fast_path_guard_rejects_ssec_requests() {
let err = get_object_chunk_fast_path_guard(true, false).unwrap_err();
assert_eq!(
err,
ChunkReadFallback {
stage: rustfs_io_metrics::IoStage::ReadSetup,
reason: rustfs_io_metrics::FallbackReason::EncryptionEnabled,
}
);
}
#[test]
fn get_object_chunk_fast_path_guard_allows_plain_request() {
assert!(get_object_chunk_fast_path_guard(false, false).is_ok());
}
#[test]
fn get_object_chunk_range_guard_rejects_invalid_suffix_range() {
let err = get_object_chunk_range_guard(Some(&HTTPRangeSpec {
is_suffix_length: true,
start: 4,
end: 0,
}))
.unwrap_err();
assert_eq!(
err,
ChunkReadFallback {
stage: rustfs_io_metrics::IoStage::RangeGuard,
reason: rustfs_io_metrics::FallbackReason::RangeNotSupported,
}
);
}
#[test]
fn build_get_object_checksums_returns_default_when_mode_absent() {
let checksums = build_get_object_checksums(&ObjectInfo::default(), &HeaderMap::new(), None, None).unwrap();
assert_eq!(checksums, GetObjectChecksums::default());
}
#[test]
fn plan_chunk_read_returns_fallback_for_compressed_object() {
let mut info = ObjectInfo::default();
rustfs_utils::http::insert_str(
&mut info.user_defined,
rustfs_utils::http::SUFFIX_COMPRESSION,
rustfs_utils::CompressionAlgorithm::Zstd.to_string(),
);
let decision = plan_chunk_read(&info, true, None, None).unwrap();
assert!(
matches!(
decision,
ChunkReadDecision::Fallback(ChunkReadFallback {
stage: rustfs_io_metrics::IoStage::ReadSetup,
reason: rustfs_io_metrics::FallbackReason::CompressionEnabled,
})
),
"unexpected decision: {:?}",
decision
);
}
#[test]
fn plan_chunk_read_returns_fallback_for_transitioned_object() {
let mut info = ObjectInfo::default();
info.transitioned_object.status = TRANSITION_COMPLETE.to_string();
let decision = plan_chunk_read(&info, true, None, None).unwrap();
assert!(matches!(
decision,
ChunkReadDecision::Fallback(ChunkReadFallback {
stage: rustfs_io_metrics::IoStage::ReadSetup,
reason: rustfs_io_metrics::FallbackReason::NonLocalBackend,
})
));
}
#[test]
fn plan_chunk_read_returns_range_guard_fallback_for_invalid_range() {
let info = ObjectInfo {
size: 16,
actual_size: 16,
..Default::default()
};
let decision = plan_chunk_read(
&info,
true,
Some(HTTPRangeSpec {
is_suffix_length: false,
start: -1,
end: 4,
}),
None,
)
.unwrap();
assert!(matches!(
decision,
ChunkReadDecision::Fallback(ChunkReadFallback {
stage: rustfs_io_metrics::IoStage::RangeGuard,
reason: rustfs_io_metrics::FallbackReason::RangeNotSupported,
})
));
}
#[test]
fn plan_chunk_read_allows_suffix_range() {
let info = ObjectInfo {
size: 16,
actual_size: 16,
..Default::default()
};
let decision = plan_chunk_read(
&info,
true,
Some(HTTPRangeSpec {
is_suffix_length: true,
start: 4,
end: -1,
}),
None,
)
.unwrap();
let ChunkReadDecision::Eligible(plan) = decision else {
panic!("expected eligible plan");
};
let rs = plan.rs.expect("suffix range should be preserved");
assert!(rs.is_suffix_length);
assert_eq!(rs.start, 4);
assert_eq!(plan.response_content_length, 4);
assert_eq!(plan.content_range.as_deref(), Some("bytes 12-15/16"));
}
#[test]
fn plan_chunk_read_uses_part_number_range_when_available() {
let mut info = ObjectInfo {
size: 12,
actual_size: 12,
content_type: Some("application/octet-stream".to_string()),
..Default::default()
};
info.parts = vec![Default::default(), Default::default()];
info.parts[0].number = 1;
info.parts[0].size = 5;
info.parts[0].actual_size = 5;
info.parts[1].number = 2;
info.parts[1].size = 7;
info.parts[1].actual_size = 7;
let decision = plan_chunk_read(&info, true, None, Some(2)).unwrap();
let ChunkReadDecision::Eligible(plan) = decision else {
panic!("expected eligible plan");
};
let rs = plan.rs.expect("range from part number");
assert_eq!(rs.start, 5);
assert_eq!(rs.end, 11);
assert_eq!(plan.response_content_length, 7);
assert_eq!(plan.content_range.as_deref(), Some("bytes 5-11/12"));
assert!(plan.content_type.is_some());
}
#[test]
fn plan_chunk_read_returns_delete_marker_errors() {
let info = ObjectInfo {
delete_marker: true,
..Default::default()
};
let err = plan_chunk_read(&info, true, None, None).unwrap_err();
assert!(matches!(err, ChunkReadPlanError::NoSuchKey));
let err = plan_chunk_read(&info, false, None, None).unwrap_err();
assert!(matches!(err, ChunkReadPlanError::MethodNotAllowed));
}
#[test]
fn plan_legacy_read_uses_part_number_range_when_available() {
let mut info = ObjectInfo {
@@ -1074,7 +630,6 @@ mod tests {
assert_eq!(setup.response_content_length, 12);
match setup.body_source {
GetObjectBodySource::Reader(_) => {}
GetObjectBodySource::Chunk { .. } => panic!("expected reader body source"),
}
}
@@ -1246,37 +801,4 @@ mod tests {
assert_eq!(result.version_id_for_event, "vid");
}
#[test]
fn finalize_chunk_read_setup_preserves_body_source_and_io_path() {
let chunk_result = GetObjectChunkResult {
stream: Box::pin(futures_util::stream::empty::<std::io::Result<rustfs_io_core::IoChunk>>()),
path: GetObjectChunkPath::Direct,
copy_mode: GetObjectChunkCopyMode::Reconstructed,
};
let plan = ChunkReadPlan {
rs: Some(HTTPRangeSpec {
is_suffix_length: false,
start: 0,
end: 7,
}),
content_type: None,
last_modified: None,
response_content_length: 8,
content_range: Some("bytes 0-7/8".to_string()),
};
let result = finalize_chunk_read_setup(ObjectInfo::default(), ObjectInfo::default(), chunk_result, plan);
assert_eq!(result.io_path, rustfs_io_metrics::IoPath::Fast);
match result.read_setup.body_source {
GetObjectBodySource::Chunk { path, copy_mode, .. } => {
assert_eq!(path, GetObjectChunkPath::Direct);
assert_eq!(copy_mode, rustfs_io_metrics::CopyMode::Reconstructed);
}
GetObjectBodySource::Reader(_) => panic!("expected chunk body source"),
}
assert_eq!(result.read_setup.response_content_length, 8);
assert_eq!(result.read_setup.content_range.as_deref(), Some("bytes 0-7/8"));
}
}
rustfs/src/app/object_usecase.rs
-2
View File
@@ -18,8 +18,6 @@ mod get_object_flow;
mod get_object_zero_copy;
mod put_object_extract;
mod put_object_flow;
#[cfg(test)]
mod zero_copy_tests;
use self::get_object_flow::GetObjectBootstrap;
use crate::app::context::{AppContext, default_notify_interface, get_global_app_context};
rustfs/src/app/object_usecase/get_object_flow.rs
+62 -287
View File
@@ -14,20 +14,14 @@
use super::DeadlockRequestGuard;
use super::GetObjectRequestContext;
use super::get_object_zero_copy::{
GetObjectIoPlanning, GetObjectPreparedRead, prepare_get_object_read, prepare_get_object_read_execution,
};
use super::get_object_zero_copy::{GetObjectIoPlanning, GetObjectPreparedRead, prepare_get_object_read_execution};
use crate::error::ApiError;
use crate::storage::concurrency::{ConcurrencyManager, GetObjectGuard, get_buffer_size_opt_in};
use crate::storage::get_validated_store;
use crate::storage::options::filter_object_metadata;
use crate::storage::timeout_wrapper::{RequestTimeoutWrapper, TimeoutConfig};
use bytes::Bytes;
use futures_util::StreamExt;
use rustfs_ecstore::bucket::versioning_sys::BucketVersioningSys;
use rustfs_ecstore::error::StorageError;
use rustfs_ecstore::store_api::{HTTPRangeSpec, ObjectInfo};
use rustfs_io_core::BoxChunkStream;
use rustfs_object_io::get::{
GetObjectBodyPlan as ObjectIoGetObjectBodyPlan, GetObjectBodyPlanningInputs as ObjectIoGetObjectBodyPlanningInputs,
GetObjectBodySource, GetObjectDataPlaneMetricContract as ObjectIoGetObjectDataPlaneMetricContract, GetObjectFlowResult,
@@ -35,8 +29,6 @@ use rustfs_object_io::get::{
build_cors_wrapped_get_object_flow_result as object_io_build_cors_wrapped_get_object_flow_result,
build_get_object_checksums as object_io_build_get_object_checksums,
build_get_object_output_context as object_io_build_get_object_output_context,
build_memory_blob as object_io_build_memory_blob, chunk_body_data_plane_labels as object_io_chunk_body_data_plane_labels,
get_object_chunk_path_label as object_io_get_object_chunk_path_label,
materialize_get_object_body as object_io_materialize_get_object_body, plan_get_object_body as object_io_plan_get_object_body,
plan_get_object_strategy_layout as object_io_plan_get_object_strategy_layout,
};
@@ -54,95 +46,6 @@ pub(super) struct GetObjectBootstrap {
pub(super) _deadlock_request_guard: DeadlockRequestGuard,
}
#[derive(Debug)]
struct ChunkCommitMaterializationError {
source: std::io::Error,
streamed_bytes: usize,
}
fn build_chunk_materialization_length_error(actual: usize, expected: usize) -> std::io::Error {
let error_kind = if actual > expected {
std::io::ErrorKind::InvalidData
} else {
std::io::ErrorKind::UnexpectedEof
};
std::io::Error::new(
error_kind,
format!("chunk fast path produced {actual} bytes before response commit, expected {expected}"),
)
}
async fn materialize_chunk_stream_before_commit_with_threshold(
mut chunk_stream: BoxChunkStream,
response_content_length: i64,
optimal_buffer_size: usize,
in_memory_threshold_bytes: usize,
) -> Result<Option<StreamingBlob>, ChunkCommitMaterializationError> {
let expected_bytes = usize::try_from(response_content_length).map_err(|_| ChunkCommitMaterializationError {
source: std::io::Error::new(
std::io::ErrorKind::InvalidInput,
format!("negative response content length {response_content_length} for chunk fast path"),
),
streamed_bytes: 0,
})?;
// Objects larger than the in-memory threshold fall back to the legacy reader path
// rather than spooling to disk, to avoid exhausting local disk under concurrent large downloads.
if expected_bytes > in_memory_threshold_bytes {
return Err(ChunkCommitMaterializationError {
source: std::io::Error::other(format!(
"chunk fast path object size {expected_bytes} exceeds in-memory threshold \
{in_memory_threshold_bytes}; falling back to legacy reader"
)),
streamed_bytes: 0,
});
}
let mut buf = Vec::with_capacity(expected_bytes);
let mut streamed_bytes = 0usize;
while let Some(result) = chunk_stream.next().await {
let chunk = result.map_err(|source| ChunkCommitMaterializationError { source, streamed_bytes })?;
let bytes = chunk.as_bytes();
streamed_bytes = streamed_bytes.saturating_add(bytes.len());
if streamed_bytes > expected_bytes {
return Err(ChunkCommitMaterializationError {
source: build_chunk_materialization_length_error(streamed_bytes, expected_bytes),
streamed_bytes,
});
}
buf.extend_from_slice(bytes.as_ref());
}
if streamed_bytes != expected_bytes {
return Err(ChunkCommitMaterializationError {
source: build_chunk_materialization_length_error(streamed_bytes, expected_bytes),
streamed_bytes,
});
}
Ok(object_io_build_memory_blob(
Bytes::from(buf),
response_content_length,
optimal_buffer_size,
))
}
async fn materialize_chunk_stream_before_commit(
chunk_stream: BoxChunkStream,
response_content_length: i64,
optimal_buffer_size: usize,
) -> Result<Option<StreamingBlob>, ChunkCommitMaterializationError> {
materialize_chunk_stream_before_commit_with_threshold(
chunk_stream,
response_content_length,
optimal_buffer_size,
rustfs_config::DEFAULT_OBJECT_SEEK_SUPPORT_THRESHOLD,
)
.await
}
async fn build_get_object_body_adapter<R>(
final_stream: R,
bucket: &str,
@@ -340,14 +243,62 @@ pub(super) async fn build_get_object_output_context(
let bucket = &request_context.bucket;
let key = &request_context.key;
let part_number = request_context.part_number;
let mut active_read_setup = read_setup;
let GetObjectReadSetup {
info,
event_info,
body_source,
rs,
content_type,
last_modified,
response_content_length,
content_range,
server_side_encryption,
sse_customer_algorithm,
sse_customer_key_md5,
ssekms_key_id,
encryption_applied,
} = read_setup;
loop {
let GetObjectReadSetup {
let optimal_buffer_size = finalize_get_object_strategy_runtime(
request_context,
rs.as_ref(),
manager,
base_buffer_size,
&info,
response_content_length,
io_planning,
);
let GetObjectBodySource::Reader(final_stream) = body_source;
let body = build_get_object_body_adapter(
final_stream,
bucket,
key,
response_content_length,
optimal_buffer_size,
ObjectIoGetObjectBodyPlanningInputs {
is_part_request: part_number.is_some(),
is_range_request: rs.is_some(),
encryption_applied,
response_size: response_content_length,
},
)
.await?;
let metric_contract = ObjectIoGetObjectDataPlaneMetricContract::disk(
rustfs_io_metrics::IoPath::Legacy,
rustfs_io_metrics::CopyMode::SingleCopy,
);
let checksums = object_io_build_get_object_checksums(&info, &request_context.headers, part_number, rs.as_ref())
.map_err(ApiError::from)?;
let filtered_metadata = filter_object_metadata(&info.user_defined);
Ok((
object_io_build_get_object_output_context(
body,
info,
event_info,
body_source,
rs,
content_type,
last_modified,
response_content_length,
@@ -356,108 +307,14 @@ pub(super) async fn build_get_object_output_context(
sse_customer_algorithm,
sse_customer_key_md5,
ssekms_key_id,
encryption_applied,
} = active_read_setup;
let optimal_buffer_size = finalize_get_object_strategy_runtime(
request_context,
rs.as_ref(),
manager,
base_buffer_size,
&info,
response_content_length,
io_planning,
);
let (body, metric_contract) = match body_source {
GetObjectBodySource::Reader(final_stream) => {
let body = build_get_object_body_adapter(
final_stream,
bucket,
key,
response_content_length,
optimal_buffer_size,
ObjectIoGetObjectBodyPlanningInputs {
is_part_request: part_number.is_some(),
is_range_request: rs.is_some(),
encryption_applied,
response_size: response_content_length,
},
)
.await?;
let metric_contract = ObjectIoGetObjectDataPlaneMetricContract::disk(
rustfs_io_metrics::IoPath::Legacy,
rustfs_io_metrics::CopyMode::SingleCopy,
);
(body, metric_contract)
}
GetObjectBodySource::Chunk {
stream: chunk_stream,
path,
copy_mode,
} => {
let (io_path, copy_mode) = object_io_chunk_body_data_plane_labels(path, copy_mode);
match materialize_chunk_stream_before_commit(chunk_stream, response_content_length, optimal_buffer_size).await {
Ok(body) => (body, ObjectIoGetObjectDataPlaneMetricContract::disk(io_path, copy_mode)),
Err(err) => {
let path_label = object_io_get_object_chunk_path_label(path);
rustfs_io_metrics::record_io_fallback(
rustfs_io_metrics::IoStage::ReadSetup,
rustfs_io_metrics::FallbackReason::ProbeFailed,
);
rustfs_io_metrics::record_get_object_fast_path_probe_failed(
path_label,
copy_mode,
response_content_length,
);
warn!(
bucket = %request_context.bucket,
key = %request_context.key,
version_id = ?request_context.opts.version_id,
path = path_label,
copy_mode = copy_mode.as_str(),
promised_bytes = response_content_length,
materialized_bytes = err.streamed_bytes,
error = %err.source,
"GetObject chunk fast path full-body materialization failed before response commit"
);
let store = get_validated_store(&request_context.bucket).await?;
active_read_setup =
prepare_get_object_read(request_context, &store, manager, std::time::Instant::now()).await?;
continue;
}
}
}
};
let checksums = object_io_build_get_object_checksums(&info, &request_context.headers, part_number, rs.as_ref())
.map_err(ApiError::from)?;
let filtered_metadata = filter_object_metadata(&info.user_defined);
return Ok((
object_io_build_get_object_output_context(
body,
info,
event_info,
content_type,
last_modified,
response_content_length,
content_range,
server_side_encryption,
sse_customer_algorithm,
sse_customer_key_md5,
ssekms_key_id,
&checksums,
filtered_metadata,
versioned,
optimal_buffer_size,
Some(metric_contract.copy_mode),
),
metric_contract,
));
}
&checksums,
filtered_metadata,
versioned,
optimal_buffer_size,
Some(metric_contract.copy_mode),
),
metric_contract,
))
}
pub(super) async fn run_get_object_flow(
@@ -498,10 +355,8 @@ pub(super) async fn run_get_object_flow(
mod tests {
use super::get_object_strategy_range;
use super::*;
use futures_util::StreamExt;
use http::HeaderMap;
use rustfs_ecstore::store_api::ObjectOptions;
use rustfs_io_core::IoChunk;
fn sample_range(start: i64, end: i64) -> HTTPRangeSpec {
HTTPRangeSpec {
@@ -545,84 +400,4 @@ mod tests {
assert_eq!(strategy_range.start, 0);
assert_eq!(strategy_range.end, 511);
}
#[tokio::test]
async fn materialize_chunk_stream_before_commit_buffers_small_payload_in_memory() {
let chunk_stream: BoxChunkStream = Box::pin(futures_util::stream::iter(vec![
Ok(IoChunk::Shared(bytes::Bytes::from_static(b"hello"))),
Ok(IoChunk::Shared(bytes::Bytes::from_static(b" world"))),
]));
let mut body = materialize_chunk_stream_before_commit_with_threshold(chunk_stream, 11, 8 * 1024, 1024)
.await
.unwrap()
.unwrap();
let mut collected = Vec::new();
while let Some(chunk) = body.next().await {
collected.extend_from_slice(&chunk.unwrap());
}
assert_eq!(collected, b"hello world");
}
#[tokio::test]
async fn materialize_chunk_stream_before_commit_falls_back_for_large_payload() {
let chunk_stream: BoxChunkStream = Box::pin(futures_util::stream::iter(vec![
Ok(IoChunk::Shared(bytes::Bytes::from_static(b"hello"))),
Ok(IoChunk::Shared(bytes::Bytes::from_static(b" world"))),
]));
// When payload exceeds the in-memory threshold an error is returned so the
// caller can fall back to the legacy reader path rather than spooling to disk.
let err = materialize_chunk_stream_before_commit_with_threshold(chunk_stream, 11, 8 * 1024, 4)
.await
.unwrap_err();
assert_eq!(err.streamed_bytes, 0);
assert_eq!(err.source.kind(), std::io::ErrorKind::Other);
}
#[tokio::test]
async fn materialize_chunk_stream_before_commit_rejects_short_body() {
let chunk_stream: BoxChunkStream =
Box::pin(futures_util::stream::iter(vec![Ok(IoChunk::Shared(bytes::Bytes::from_static(b"hello")))]));
let err = materialize_chunk_stream_before_commit_with_threshold(chunk_stream, 11, 8 * 1024, 1024)
.await
.unwrap_err();
assert_eq!(err.streamed_bytes, 5);
assert_eq!(err.source.kind(), std::io::ErrorKind::UnexpectedEof);
}
#[tokio::test]
async fn materialize_chunk_stream_before_commit_rejects_long_body() {
let chunk_stream: BoxChunkStream = Box::pin(futures_util::stream::iter(vec![
Ok(IoChunk::Shared(bytes::Bytes::from_static(b"hello "))),
Ok(IoChunk::Shared(bytes::Bytes::from_static(b"world!"))),
]));
let err = materialize_chunk_stream_before_commit_with_threshold(chunk_stream, 11, 8 * 1024, 1024)
.await
.unwrap_err();
assert_eq!(err.streamed_bytes, 12);
assert_eq!(err.source.kind(), std::io::ErrorKind::InvalidData);
}
#[tokio::test]
async fn materialize_chunk_stream_before_commit_preserves_midstream_io_errors() {
let chunk_stream: BoxChunkStream = Box::pin(futures_util::stream::iter(vec![
Ok(IoChunk::Shared(bytes::Bytes::from_static(b"hello"))),
Err(std::io::Error::new(std::io::ErrorKind::BrokenPipe, "writer closed")),
]));
let err = materialize_chunk_stream_before_commit_with_threshold(chunk_stream, 11, 8 * 1024, 1024)
.await
.unwrap_err();
assert_eq!(err.streamed_bytes, 5);
assert_eq!(err.source.kind(), std::io::ErrorKind::BrokenPipe);
}
}
rustfs/src/app/object_usecase/get_object_zero_copy.rs
+5 -230
View File
@@ -20,56 +20,18 @@ use crate::storage::{
DecryptionRequest, check_preconditions, get_validated_store, sse_decryption, validate_sse_headers_for_read,
validate_ssec_for_read,
};
use futures_util::{StreamExt, stream};
use http::HeaderMap;
use rustfs_concurrency::GetObjectQueueSnapshot;
use rustfs_ecstore::store_api::{ObjectIO, ObjectOperations};
use rustfs_io_core::{BoxChunkStream, IoChunk};
use rustfs_ecstore::store_api::ObjectIO;
use rustfs_object_io::get::{
ChunkReadDecision, ChunkReadPlanError, GetObjectEncryptionState as ObjectIoGetObjectEncryptionState, GetObjectReadSetup,
build_reader_read_setup as object_io_build_reader_read_setup,
finalize_chunk_read_setup as object_io_finalize_chunk_read_setup,
get_object_chunk_fast_path_guard as object_io_get_object_chunk_fast_path_guard,
get_object_chunk_path_label as object_io_get_object_chunk_path_label, map_chunk_copy_mode as object_io_map_chunk_copy_mode,
plan_chunk_read as object_io_plan_chunk_read, plan_legacy_read as object_io_plan_legacy_read,
GetObjectEncryptionState as ObjectIoGetObjectEncryptionState, GetObjectReadSetup,
build_reader_read_setup as object_io_build_reader_read_setup, plan_legacy_read as object_io_plan_legacy_read,
};
use rustfs_rio::{Reader, WarpReader};
use s3s::{S3Error, S3ErrorCode, S3Result, s3_error};
use s3s::{S3Result, s3_error};
use std::time::Duration;
use tracing::{debug, warn};
fn get_object_chunk_fast_path_enabled() -> bool {
rustfs_utils::get_env_bool(
rustfs_config::ENV_OBJECT_GET_CHUNK_FAST_PATH_ENABLE,
rustfs_config::DEFAULT_OBJECT_GET_CHUNK_FAST_PATH_ENABLE,
)
}
async fn probe_chunk_stream_before_commit(
mut chunk_stream: BoxChunkStream,
response_content_length: i64,
) -> Result<BoxChunkStream, rustfs_io_metrics::FallbackReason> {
if response_content_length <= 0 {
return Ok(chunk_stream);
}
let mut prefetched = Vec::new();
loop {
match chunk_stream.next().await {
Some(Ok(chunk)) => {
let chunk_len = chunk.len();
prefetched.push(chunk);
if chunk_len > 0 {
let prefix = stream::iter(prefetched.into_iter().map(Ok::<IoChunk, std::io::Error>));
return Ok(Box::pin(prefix.chain(chunk_stream)));
}
}
Some(Err(_)) | None => return Err(rustfs_io_metrics::FallbackReason::ProbeFailed),
}
}
}
pub(super) struct GetObjectIoPlanning<'a> {
pub(super) _disk_permit: tokio::sync::SemaphorePermit<'a>,
pub(super) permit_wait_duration: Duration,
@@ -262,194 +224,7 @@ pub(super) async fn prepare_get_object_read_execution<'a>(
let io_planning =
acquire_get_object_io_planning(manager, wrapper, timeout_config, &request_context.bucket, &request_context.key).await?;
let store = get_validated_store(&request_context.bucket).await?;
let read_start = std::time::Instant::now();
let read_setup = if !get_object_chunk_fast_path_enabled() {
rustfs_io_metrics::record_io_fallback(
rustfs_io_metrics::IoStage::ReadSetup,
rustfs_io_metrics::FallbackReason::FeatureDisabled,
);
prepare_get_object_read(request_context, &store, manager, read_start).await?
} else {
match object_io_get_object_chunk_fast_path_guard(
request_context.sse_customer_key.is_some(),
request_context.sse_customer_key_md5.is_some(),
) {
Ok(()) => match prepare_get_object_chunk_read(request_context, &store, manager, read_start).await? {
Some(read_setup) => read_setup,
None => prepare_get_object_read(request_context, &store, manager, read_start).await?,
},
Err(fallback) => {
rustfs_io_metrics::record_io_fallback(fallback.stage, fallback.reason);
prepare_get_object_read(request_context, &store, manager, read_start).await?
}
}
};
let read_setup = prepare_get_object_read(request_context, &store, manager, std::time::Instant::now()).await?;
Ok(GetObjectPreparedRead { io_planning, read_setup })
}
pub(super) async fn prepare_get_object_chunk_read(
request_context: &GetObjectRequestContext,
store: &rustfs_ecstore::store::ECStore,
manager: &ConcurrencyManager,
read_start: std::time::Instant,
) -> S3Result<Option<GetObjectReadSetup>> {
let info = store
.get_object_info(&request_context.bucket, &request_context.key, &request_context.opts)
.await
.map_err(ApiError::from)?;
validate_sse_headers_for_read(&info.user_defined, &request_context.headers)?;
validate_ssec_for_read(
&info.user_defined,
request_context.sse_customer_key.as_ref(),
request_context.sse_customer_key_md5.as_ref(),
)?;
check_preconditions(&request_context.headers, &info)?;
let encrypted_object = info.user_defined.contains_key("x-rustfs-encryption-key")
|| info
.user_defined
.contains_key("x-amz-server-side-encryption-customer-algorithm");
if encrypted_object {
rustfs_io_metrics::record_io_fallback(
rustfs_io_metrics::IoStage::ReadSetup,
rustfs_io_metrics::FallbackReason::EncryptionEnabled,
);
return Ok(None);
}
let plan = match object_io_plan_chunk_read(
&info,
request_context.opts.version_id.is_none(),
request_context.rs.clone(),
request_context.part_number,
) {
Ok(ChunkReadDecision::Eligible(plan)) => plan,
Ok(ChunkReadDecision::Fallback(fallback)) => {
rustfs_io_metrics::record_io_fallback(fallback.stage, fallback.reason);
return Ok(None);
}
Err(ChunkReadPlanError::NoSuchKey) => return Err(S3Error::new(S3ErrorCode::NoSuchKey)),
Err(ChunkReadPlanError::MethodNotAllowed) => return Err(S3Error::new(S3ErrorCode::MethodNotAllowed)),
Err(ChunkReadPlanError::Io(err)) => return Err(ApiError::from(err).into()),
};
let rs = plan.rs.clone();
let response_content_length = plan.response_content_length;
let read_duration = read_start.elapsed();
manager.record_disk_operation(info.size as u64, read_duration, true).await;
let event_info = info.clone();
let chunk_result = match store
.get_object_chunks(
&request_context.bucket,
&request_context.key,
rs.clone(),
HeaderMap::new(),
&request_context.opts,
)
.await
.map_err(ApiError::from)
{
Ok(result) => result,
Err(_err) => {
rustfs_io_metrics::record_io_fallback(
rustfs_io_metrics::IoStage::HttpBridge,
rustfs_io_metrics::FallbackReason::ChunkBridgeUnavailable,
);
return Ok(None);
}
};
let path_label = object_io_get_object_chunk_path_label(chunk_result.path);
let copy_mode = object_io_map_chunk_copy_mode(chunk_result.copy_mode);
let chunk_result = match probe_chunk_stream_before_commit(chunk_result.stream, response_content_length).await {
Ok(stream) => rustfs_ecstore::store_api::GetObjectChunkResult {
stream,
path: chunk_result.path,
copy_mode: chunk_result.copy_mode,
},
Err(reason) => {
rustfs_io_metrics::record_io_fallback(rustfs_io_metrics::IoStage::ReadSetup, reason);
rustfs_io_metrics::record_get_object_fast_path_probe_failed(path_label, copy_mode, response_content_length);
warn!(
bucket = %request_context.bucket,
key = %request_context.key,
version_id = ?request_context.opts.version_id,
path = path_label,
copy_mode = copy_mode.as_str(),
promised_bytes = response_content_length,
fallback_reason = reason.as_str(),
"GetObject chunk fast path probe failed before response commit"
);
return Ok(None);
}
};
let setup_result = object_io_finalize_chunk_read_setup(info, event_info, chunk_result, plan);
rustfs_io_metrics::record_get_object_fast_path_selected(path_label, copy_mode, response_content_length);
rustfs_io_metrics::record_io_path_selected("get", setup_result.io_path);
Ok(Some(setup_result.read_setup))
}
#[cfg(test)]
mod tests {
use super::{get_object_chunk_fast_path_enabled, probe_chunk_stream_before_commit};
use bytes::Bytes;
use futures_util::{StreamExt, stream};
use rustfs_io_core::{BoxChunkStream, IoChunk};
#[test]
fn get_object_chunk_fast_path_defaults_to_disabled() {
temp_env::with_var_unset(rustfs_config::ENV_OBJECT_GET_CHUNK_FAST_PATH_ENABLE, || {
assert!(!get_object_chunk_fast_path_enabled());
});
}
#[test]
fn get_object_chunk_fast_path_can_be_explicitly_enabled() {
temp_env::with_var(rustfs_config::ENV_OBJECT_GET_CHUNK_FAST_PATH_ENABLE, Some("true"), || {
assert!(get_object_chunk_fast_path_enabled());
});
}
#[tokio::test]
async fn probe_chunk_stream_before_commit_preserves_prefetched_payload() {
let stream: BoxChunkStream = Box::pin(stream::iter(vec![
Ok(IoChunk::Shared(Bytes::from_static(b"hello "))),
Ok(IoChunk::Shared(Bytes::from_static(b"world"))),
]));
let mut probed = probe_chunk_stream_before_commit(stream, 11).await.unwrap();
let mut collected = Vec::new();
while let Some(chunk) = probed.next().await {
collected.extend_from_slice(chunk.unwrap().as_bytes().as_ref());
}
assert_eq!(collected, b"hello world");
}
#[tokio::test]
async fn probe_chunk_stream_before_commit_rejects_midstream_failure_before_first_chunk() {
let stream: BoxChunkStream = Box::pin(stream::iter(vec![Err(std::io::Error::other("probe failed"))]));
let err = match probe_chunk_stream_before_commit(stream, 1).await {
Ok(_) => panic!("expected probe failure"),
Err(err) => err,
};
assert_eq!(err, rustfs_io_metrics::FallbackReason::ProbeFailed);
}
#[tokio::test]
async fn probe_chunk_stream_before_commit_rejects_unexpected_empty_stream() {
let stream: BoxChunkStream = Box::pin(stream::empty());
let err = match probe_chunk_stream_before_commit(stream, 1).await {
Ok(_) => panic!("expected probe failure"),
Err(err) => err,
};
assert_eq!(err, rustfs_io_metrics::FallbackReason::ProbeFailed);
}
}
rustfs/src/app/object_usecase/zero_copy_tests.rs
-1162
View File
File diff suppressed because it is too large Load Diff