mirror of
https://github.com/rustfs/rustfs.git
synced 2026-07-26 16:28:15 +00:00
perf(ecstore): reuse erasure-decode shard buffers across stripes (#3482)
* perf(ecstore): reuse erasure-decode shard buffers across stripes ParallelReader::read allocated and zero-filled a fresh `vec![0u8; shard_size]` per shard on every erasure stripe. Erasure::decode builds one ParallelReader and loops stripes, so those buffers can be reused: BitrotReader::read overwrites buf[..n] and the caller truncates to n, so leftover bytes from the prior stripe are never observed. The decode loop now hands each stripe's shard buffers back to the reader, which reuses them (resized to shard_size) on the next stripe. The heal path, which consumes its buffers into Bytes, never hands them back and is unchanged. This path runs on every object GET. Streaming-decode benchmark (Erasure::decode end-to-end, median): 4+2 16MiB 1MiB-block: -9.6% verify-on, -23.8% verify-off 6+3 24MiB 1MiB-block: -9.7% verify-on, -22.5% verify-off 4+2 4MiB 64KiB-block: -5.5% verify-on, -16.3% verify-off Adds a streaming-decode benchmark covering the ParallelReader path and a regression test that decodes a multi-stripe object with missing data shards (so reconstructed buffers are recycled) with bitrot verification both on and off, asserting byte-exact output. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> * perf(ecstore): tighten decode buffer reuse and bench harness Address PR review feedback on the erasure-decode buffer-reuse change: - Only claim a recycled shard buffer inside the `Some(reader)` branch of `ParallelReader::read`, so the take is co-located with the read that uses it and missing shards no longer touch the recycle slot. - Move per-shard `BitrotReader` construction into Criterion `iter_batched` setup so reader/UUID construction stays out of the timed decode path, and assert decode success plus full output length each iteration. Re-measured with the cleaner harness (median, before -> after, all p < 0.05): verify-on -5.3%..-11.7%, verify-off -15.8%..-25.3%. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> * perf(ecstore): drop dead recycle slots and tidy decode bench Address the second PR review pass: - In `Erasure::decode`, clear recycled slots for missing-shard (None-reader) indices before handing buffers back. `ParallelReader::read` only reuses slots with an active reader, and `decode_data` always re-allocates the reconstructed shard, so retaining those buffers only held memory that could never be reused in degraded reads. - Move the output sink into the benchmark's `iter_batched` setup and validate decode success/output length once per config instead of inside the timed loop, so neither sink construction nor assertions touch the measurement. Re-measured with the final harness (median, before -> after, all p < 0.05): verify-on -5.2%..-9.8%, verify-off -19.4%..-25.8%. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --------- Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
This commit is contained in:
Ramakrishna Chilaka
committed by
GitHub
GitHub
parent
49c0f13120
commit
956bd19417
@@ -44,9 +44,12 @@
|
||||
//! - SIMD optimization for different shard sizes
|
||||
|
||||
use criterion::{BenchmarkId, Criterion, Throughput, criterion_group, criterion_main};
|
||||
use rustfs_ecstore::erasure_coding::{Erasure, calc_shard_size};
|
||||
use rustfs_ecstore::erasure_coding::{BitrotReader, BitrotWriter, Erasure, calc_shard_size};
|
||||
use rustfs_utils::HashAlgorithm;
|
||||
use std::hint::black_box;
|
||||
use std::io::Cursor;
|
||||
use std::time::Duration;
|
||||
use tokio::runtime::Runtime;
|
||||
|
||||
/// Benchmark configuration structure
|
||||
#[derive(Clone, Debug)]
|
||||
@@ -328,11 +331,113 @@ fn bench_memory_patterns(c: &mut Criterion) {
|
||||
group.finish();
|
||||
}
|
||||
|
||||
/// Benchmark: end-to-end streaming decode through `Erasure::decode`.
|
||||
///
|
||||
/// Unlike `bench_decode_performance` (which calls `decode_data` on in-memory
|
||||
/// shards), this drives the async `ParallelReader` path that reads each shard
|
||||
/// through a `BitrotReader` per erasure stripe. That is the path executed on
|
||||
/// every object GET, and the one where per-stripe shard buffers are allocated.
|
||||
fn bench_streaming_decode(c: &mut Criterion) {
|
||||
let rt = Runtime::new().unwrap();
|
||||
let hash_algo = HashAlgorithm::HighwayHash256;
|
||||
|
||||
// (data_shards, parity_shards, object_size, block_size)
|
||||
let configs = vec![
|
||||
(4usize, 2usize, 16 * 1024 * 1024usize, 1024 * 1024usize),
|
||||
(6, 3, 24 * 1024 * 1024, 1024 * 1024),
|
||||
(4, 2, 4 * 1024 * 1024, 64 * 1024),
|
||||
];
|
||||
|
||||
for (data_shards, parity_shards, data_size, block_size) in configs {
|
||||
let data = generate_test_data(data_size);
|
||||
let erasure = Erasure::new(data_shards, parity_shards, block_size);
|
||||
let shard_size = erasure.shard_size();
|
||||
let total_len = data.len();
|
||||
|
||||
// Pre-encode the object into per-shard bitrot streams once (setup).
|
||||
let total_shards = data_shards + parity_shards;
|
||||
let shard_bufs: Vec<Vec<u8>> = rt.block_on(async {
|
||||
let mut writers: Vec<BitrotWriter<Cursor<Vec<u8>>>> = (0..total_shards)
|
||||
.map(|_| BitrotWriter::new(Cursor::new(Vec::new()), shard_size, hash_algo.clone()))
|
||||
.collect();
|
||||
let mut off = 0;
|
||||
while off < data.len() {
|
||||
let end = (off + block_size).min(data.len());
|
||||
let shards = erasure.encode_data(&data[off..end]).unwrap();
|
||||
for (i, shard) in shards.iter().enumerate() {
|
||||
writers[i].write(shard).await.unwrap();
|
||||
}
|
||||
off = end;
|
||||
}
|
||||
writers.into_iter().map(|w| w.into_inner().into_inner()).collect()
|
||||
});
|
||||
|
||||
// verify=true mirrors the production default (bitrot verification on);
|
||||
// verify=false isolates the buffer-handling cost from the hash pass.
|
||||
for verify in [true, false] {
|
||||
let skip_verify = !verify;
|
||||
let mut group = c.benchmark_group("streaming_decode");
|
||||
group.throughput(Throughput::Bytes(total_len as u64));
|
||||
group.sample_size(10);
|
||||
group.measurement_time(Duration::from_secs(5));
|
||||
|
||||
let name = format!(
|
||||
"{}+{}_{}KB_{}KB-block_verify-{}",
|
||||
data_shards,
|
||||
parity_shards,
|
||||
data_size / 1024,
|
||||
block_size / 1024,
|
||||
verify
|
||||
);
|
||||
|
||||
// Validate decode once per config (untimed) so a regression in
|
||||
// success or output length fails the benchmark without putting
|
||||
// assertions inside the measured loop.
|
||||
{
|
||||
let readers: Vec<Option<BitrotReader<Cursor<&[u8]>>>> = shard_bufs
|
||||
.iter()
|
||||
.map(|buf| Some(BitrotReader::new(Cursor::new(buf.as_slice()), shard_size, hash_algo.clone(), skip_verify)))
|
||||
.collect();
|
||||
let mut sink = tokio::io::sink();
|
||||
let (written, err) = rt.block_on(erasure.decode(&mut sink, readers, 0, total_len, total_len));
|
||||
assert!(err.is_none(), "decode failed: {err:?}");
|
||||
assert_eq!(written, total_len, "decode wrote {written} of {total_len} bytes");
|
||||
}
|
||||
|
||||
group.bench_function(BenchmarkId::new("decode", name), |b| {
|
||||
b.iter_batched(
|
||||
|| {
|
||||
// Setup (untimed): per-shard readers positioned at the
|
||||
// object start, plus the no-op sink, so reader/UUID and
|
||||
// sink construction stay out of the timed decode path.
|
||||
let readers: Vec<Option<BitrotReader<Cursor<&[u8]>>>> = shard_bufs
|
||||
.iter()
|
||||
.map(|buf| {
|
||||
Some(BitrotReader::new(Cursor::new(buf.as_slice()), shard_size, hash_algo.clone(), skip_verify))
|
||||
})
|
||||
.collect();
|
||||
(readers, tokio::io::sink())
|
||||
},
|
||||
|(readers, mut sink)| {
|
||||
rt.block_on(async {
|
||||
let (written, _err) = erasure.decode(black_box(&mut sink), readers, 0, total_len, total_len).await;
|
||||
black_box(written);
|
||||
});
|
||||
},
|
||||
criterion::BatchSize::SmallInput,
|
||||
);
|
||||
});
|
||||
group.finish();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Benchmark group configuration
|
||||
criterion_group!(
|
||||
benches,
|
||||
bench_encode_performance,
|
||||
bench_decode_performance,
|
||||
bench_streaming_decode,
|
||||
bench_shard_size_impact,
|
||||
bench_coding_configurations,
|
||||
bench_memory_patterns
|
||||
|
||||
@@ -33,6 +33,9 @@ pub(crate) struct ParallelReader<R> {
|
||||
shard_file_size: usize,
|
||||
data_shards: usize,
|
||||
total_shards: usize,
|
||||
// Shard buffers handed back by the caller to be reused by the next `read`,
|
||||
// avoiding a per-stripe allocation + zero-fill on the object read hot path.
|
||||
recycled: Vec<Option<Vec<u8>>>,
|
||||
}
|
||||
}
|
||||
|
||||
@@ -56,6 +59,7 @@ where
|
||||
shard_file_size,
|
||||
data_shards: e.data_shards,
|
||||
total_shards: e.data_shards + e.parity_shards,
|
||||
recycled: Vec::new(),
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -83,12 +87,23 @@ where
|
||||
let mut shards: Vec<Option<Vec<u8>>> = vec![None; num_readers];
|
||||
let mut errs = vec![None; num_readers];
|
||||
|
||||
// Reuse the previous stripe's shard buffers instead of allocating and
|
||||
// zero-filling a fresh `vec![0u8; shard_size]` per shard every stripe.
|
||||
// `BitrotReader::read` overwrites `buf[..n]` and the caller truncates to
|
||||
// `n`, so leftover bytes from the prior stripe are never observed.
|
||||
let mut recycled = std::mem::take(&mut self.recycled);
|
||||
recycled.resize_with(num_readers, || None);
|
||||
|
||||
let mut futures = Vec::with_capacity(self.total_shards);
|
||||
let reader_iter: std::slice::IterMut<'_, Option<BitrotReader<R>>> = self.readers.iter_mut();
|
||||
for (i, reader) in reader_iter.enumerate() {
|
||||
let future = if let Some(reader) = reader {
|
||||
// Only claim a recycled buffer when a shard will actually be read
|
||||
// into it; missing shards are reconstructed by `decode_data`.
|
||||
let recycled_buf = recycled[i].take();
|
||||
Box::pin(async move {
|
||||
let mut buf = vec![0u8; shard_size];
|
||||
let mut buf = recycled_buf.unwrap_or_default();
|
||||
buf.resize(shard_size, 0);
|
||||
match reader.read(&mut buf).await {
|
||||
Ok(n) => {
|
||||
buf.truncate(n);
|
||||
@@ -303,6 +318,17 @@ impl Erasure {
|
||||
};
|
||||
|
||||
written += n;
|
||||
|
||||
// Hand this stripe's buffers back so the next `read` reuses them.
|
||||
// Only retain slots with an active reader; missing shards are always
|
||||
// re-allocated by `decode_data`, so retaining their buffers here would
|
||||
// hold memory that `read` can never reuse.
|
||||
for (i, r) in reader.readers.iter().enumerate() {
|
||||
if r.is_none() {
|
||||
shards[i] = None;
|
||||
}
|
||||
}
|
||||
reader.recycled = shards;
|
||||
}
|
||||
|
||||
if ret_err.is_some() {
|
||||
@@ -433,6 +459,69 @@ mod tests {
|
||||
}
|
||||
}
|
||||
|
||||
/// Guards the shard-buffer reuse in `ParallelReader`: a multi-stripe
|
||||
/// `Erasure::decode` that reconstructs missing data shards on every stripe
|
||||
/// must still return byte-exact output. Reconstructed and parity buffers are
|
||||
/// recycled into the next stripe, so this catches stale-byte leaks or a
|
||||
/// missing resize between stripes (including the short final stripe). Run
|
||||
/// with bitrot verification both on and off.
|
||||
#[tokio::test]
|
||||
async fn test_erasure_decode_with_missing_shards_reuses_buffers() {
|
||||
const DATA_SHARDS: usize = 4;
|
||||
const PARITY_SHARDS: usize = 2;
|
||||
const BLOCK_SIZE: usize = 64;
|
||||
|
||||
// 200 bytes => 3 full blocks + 1 partial: several stripes are decoded
|
||||
// through the same `ParallelReader`, so its buffers are reused.
|
||||
let total_data: Vec<u8> = (0..200u32).map(|i| i as u8).collect();
|
||||
let total_len = total_data.len();
|
||||
|
||||
let erasure = Erasure::new(DATA_SHARDS, PARITY_SHARDS, BLOCK_SIZE);
|
||||
let total_shards = DATA_SHARDS + PARITY_SHARDS;
|
||||
let shard_size = erasure.shard_size();
|
||||
let hash_algo = HashAlgorithm::HighwayHash256;
|
||||
|
||||
let mut shard_writers: Vec<BitrotWriter<Cursor<Vec<u8>>>> = (0..total_shards)
|
||||
.map(|_| BitrotWriter::new(Cursor::new(Vec::new()), shard_size, hash_algo.clone()))
|
||||
.collect();
|
||||
|
||||
let mut offset = 0;
|
||||
while offset < total_len {
|
||||
let end = (offset + BLOCK_SIZE).min(total_len);
|
||||
let shards = erasure.encode_data(&total_data[offset..end]).unwrap();
|
||||
for (i, shard) in shards.iter().enumerate() {
|
||||
shard_writers[i].write(shard).await.unwrap();
|
||||
}
|
||||
offset = end;
|
||||
}
|
||||
|
||||
let shard_bufs: Vec<Vec<u8>> = shard_writers.into_iter().map(|w| w.into_inner().into_inner()).collect();
|
||||
|
||||
// Drop two data shards: each stripe must be reconstructed from the
|
||||
// remaining data + parity shards, and those reconstructed buffers are
|
||||
// what gets recycled.
|
||||
let missing = [0usize, 2usize];
|
||||
for verify in [true, false] {
|
||||
let readers: Vec<Option<BitrotReader<Cursor<Vec<u8>>>>> = shard_bufs
|
||||
.iter()
|
||||
.enumerate()
|
||||
.map(|(i, buf)| {
|
||||
if missing.contains(&i) {
|
||||
None
|
||||
} else {
|
||||
Some(BitrotReader::new(Cursor::new(buf.clone()), shard_size, hash_algo.clone(), !verify))
|
||||
}
|
||||
})
|
||||
.collect();
|
||||
|
||||
let mut output = Vec::new();
|
||||
let (written, err) = erasure.decode(&mut output, readers, 0, total_len, total_len).await;
|
||||
assert!(err.is_none(), "verify={verify}: unexpected error: {err:?}");
|
||||
assert_eq!(written, total_len, "verify={verify}: short write");
|
||||
assert_eq!(output, total_data, "verify={verify}: reconstructed bytes mismatch");
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(feature = "rio-v2")]
|
||||
#[tokio::test]
|
||||
async fn test_erasure_decode_preserves_compressed_stream_near_block_boundary() {
|
||||
|
||||
Reference in New Issue
Block a user