From 5f1a475c5623352b47b725400405ee5929b65fd5 Mon Sep 17 00:00:00 2001 From: houseme Date: Fri, 10 Jul 2026 20:34:45 +0800 Subject: [PATCH] perf(ecstore): stop zeroing pooled shard buffers on the GET path (backlog#1159) (#4681) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `ShardBufferPool::take` handed out a `resize(len, 0)`-ed buffer, and the reader then overwrote every byte of it. CPU profiling of a cached 1 MiB GET (device reads = 0, so all cost is CPU) attributed 4.81% of the whole server to that memset — a buffer pool exists to reuse an allocation, and memsetting it gives the saving straight back. The zeroing was load-bearing only because `BitrotReader::read` takes `&mut [u8]`, which must be initialized. But the reader never reads what the caller put there, and never returns a partially filled buffer: both the hashed and the no-hash path either fill the whole shard or fail with UnexpectedEof, and a hash mismatch is an error rather than a short read. So the initialization bought nothing observable. Add `BitrotReader::read_appending(&mut Vec, want)`, which appends into the buffer's spare capacity instead of demanding initialized bytes: * hashed path — unchanged single copy, `extend_from_slice(data)` in place of `copy_from_slice` into a pre-zeroed buffer, and only after the hash verifies, so corrupt bytes never reach the caller's buffer; * no-hash path — `read_buf` writes straight into the spare capacity and advances the length only over bytes the reader actually wrote, so an uninitialized tail can never be exposed. `ShardBufferPool::take` now yields an empty buffer with capacity, and `read_shard` no longer needs to `truncate`. `read` keeps its old signature for the remaining callers. Four tests gate the contract rather than the call: * `read_appending` is byte-for-byte identical to `read` on both paths; * a truncated shard is UnexpectedEof, never a partially filled buffer; * bytes that fail the bitrot hash never reach the caller's buffer; * `want > shard_size` is rejected; plus the pool test now asserts the allocation is reused (same pointer) and never zeroed. Verified: `erasure::` 213 passed, 0 failed; on a real Linux host `erasure::` 209 and `disk::local::` 143 pass serially, and the failures seen in a parallel full-suite run reproduce identically on unmodified main (they are ENOSPC from a full root filesystem plus pre-existing flakes). Not claimed: an end-to-end throughput number. The A/B on the bench host was too noisy to attribute (one rep pair was not fully cached, and its root filesystem filled mid-run); what is measured is that the removed memset was 4.81% of GET CPU in the pre-change profile. Co-authored-by: heihutu --- crates/ecstore/src/erasure/codec/workspace.rs | 39 ++-- crates/ecstore/src/erasure/coding/bitrot.rs | 170 ++++++++++++++++++ crates/ecstore/src/erasure/coding/decode.rs | 12 +- 3 files changed, 204 insertions(+), 17 deletions(-) diff --git a/crates/ecstore/src/erasure/codec/workspace.rs b/crates/ecstore/src/erasure/codec/workspace.rs index f0d6af58a..0ed30b245 100644 --- a/crates/ecstore/src/erasure/codec/workspace.rs +++ b/crates/ecstore/src/erasure/codec/workspace.rs @@ -49,15 +49,24 @@ impl ShardBufferPool { } } - // Returned bytes may contain stale scratch data and must be overwritten before use. + /// An **empty** buffer with room for at least `len` bytes. The caller fills it + /// by appending (see `BitrotReader::read_appending`), so the pool never has to + /// initialize the bytes it hands out. + /// + /// Zeroing here was pure waste: `read_appending` overwrites every byte it + /// returns, and a short read is an error rather than a partial buffer, so no + /// caller ever observes a byte the reader did not write. At 1 MiB shards the + /// `resize(len, 0)` was ~4.8% of GET CPU (rustfs/backlog#1159) — a buffer pool + /// exists to reuse an allocation, and memsetting it gives that saving straight + /// back. #[inline] pub(crate) fn take(&mut self, index: usize, len: usize) -> Vec { self.ensure_slots(index + 1); let mut buf = self.buffers[index].take().unwrap_or_else(|| Vec::with_capacity(len)); + buf.clear(); if buf.capacity() < len { - buf.reserve_exact(len - buf.capacity()); + buf.reserve_exact(len - buf.len()); } - buf.resize(len, 0); buf } @@ -77,21 +86,26 @@ impl ShardBufferPool { mod tests { use super::*; + /// `take` hands out capacity, never length: the caller appends every byte it + /// will read back. Reusing a slot must keep the allocation and must not memset + /// it (rustfs/backlog#1159). #[test] - fn shard_buffer_pool_reuses_slot_without_clearing() { + fn shard_buffer_pool_reuses_the_allocation_and_never_zeroes_it() { let mut pool = ShardBufferPool::new(2); let mut buf = pool.take(1, 16); - assert_eq!(buf.len(), 16); - buf[0] = 42; + assert_eq!(buf.len(), 0, "take yields an empty buffer; the caller appends"); + assert!(buf.capacity() >= 16); + buf.extend_from_slice(&[42u8; 16]); let capacity = buf.capacity(); + let ptr = buf.as_ptr(); pool.put(1, buf); assert_eq!(pool.stored_capacity(1), Some(capacity)); let reused = pool.take(1, 8); - assert_eq!(reused.len(), 8); - assert!(reused.capacity() >= capacity); - assert_eq!(reused[0], 42); + assert_eq!(reused.len(), 0); + assert!(reused.capacity() >= capacity, "the allocation must be reused, not reallocated"); + assert_eq!(reused.as_ptr(), ptr, "same allocation"); } #[test] @@ -99,7 +113,8 @@ mod tests { let mut pool = ShardBufferPool::new(0); let buf = pool.take(3, 4); - assert_eq!(buf.len(), 4); + assert_eq!(buf.len(), 0); + assert!(buf.capacity() >= 4); assert_eq!(pool.buffers.len(), 4); } @@ -112,7 +127,7 @@ mod tests { pool.put(0, Vec::with_capacity(2)); let grown = pool.take(0, 8); - assert_eq!(grown.len(), 8); - assert!(grown.capacity() >= 8); + assert_eq!(grown.len(), 0); + assert!(grown.capacity() >= 8, "a too-small reused slot must grow to the requested capacity"); } } diff --git a/crates/ecstore/src/erasure/coding/bitrot.rs b/crates/ecstore/src/erasure/coding/bitrot.rs index 038ed6d84..ee6915ff5 100644 --- a/crates/ecstore/src/erasure/coding/bitrot.rs +++ b/crates/ecstore/src/erasure/coding/bitrot.rs @@ -132,6 +132,84 @@ where Ok(out.len()) } + /// Same contract as [`Self::read`], but **appends** `want` bytes into `out`'s + /// spare capacity instead of demanding an initialized `&mut [u8]` + /// (rustfs/backlog#1159). + /// + /// `read` forced its caller to hand over a zeroed buffer purely to satisfy + /// `&mut [u8]`, and every one of those bytes was then overwritten. Because a + /// short read is an error here (never a partially filled buffer), `out` ends + /// up holding exactly the bytes this reader produced, so nothing the reader + /// did not write is ever observable — the zeroing bought nothing. + /// + /// On return `out.len()` has grown by exactly the returned count. + pub async fn read_appending(&mut self, out: &mut Vec, want: usize) -> std::io::Result { + use bytes::BufMut as _; + use tokio::io::AsyncReadExt as _; + + self.last_verify_duration = Duration::ZERO; + if want > self.shard_size { + return Err(std::io::Error::new( + std::io::ErrorKind::InvalidInput, + format!("data size {want} exceeds shard size {}", self.shard_size), + )); + } + out.reserve(want); + let hash_size = self.hash_algo.size(); + + // No-hash path: read straight into `out`'s spare capacity. `read_buf` + // advances the length only over bytes the reader actually wrote, so an + // uninitialized tail can never be exposed. + if hash_size == 0 { + let start = out.len(); + while out.len() - start < want { + let remaining = want - (out.len() - start); + let n = self + .inner + .read_buf(&mut (&mut *out).limit(remaining)) + .await + .inspect_err(|e| { + error!("bitrot reader read error: {}", e); + })?; + if n == 0 { + break; + } + } + return self.finish_len(out.len() - start, want); + } + + // Hashed path: identical to `read` — one pass pulls `[hash][data]` into + // the scratch buffer — except the shard lands in `out` by `extend_from_slice` + // rather than `copy_from_slice` into a pre-zeroed buffer. Same single copy. + let need = hash_size + want; + if self.buf.len() < need { + self.buf.resize(need, 0); + } + let filled = fill(&mut self.inner, &mut self.buf[..need]).await?; + if filled < need { + let got_data = filled.saturating_sub(hash_size); + error!("bitrot reader short shard read: id={} got {} of {} bytes", self.id, got_data, want); + return Err(std::io::Error::new( + std::io::ErrorKind::UnexpectedEof, + format!("short shard read: got {got_data} of {want} bytes"), + )); + } + + let (hash, data) = self.buf[..need].split_at(hash_size); + if !self.skip_verify { + let verify_start = std::time::Instant::now(); + let actual_hash = self.hash_algo.hash_encode(data); + self.last_verify_duration = verify_start.elapsed(); + if actual_hash.as_ref() != hash { + error!("bitrot reader hash mismatch, id={} data_len={}, out_len={}", self.id, data.len(), want); + return Err(std::io::Error::new(std::io::ErrorKind::InvalidData, "bitrot hash mismatch")); + } + } + // Only after verification: a corrupt shard must not reach the caller. + out.extend_from_slice(data); + Ok(want) + } + /// Map a completed no-hash read to the shared short-shard contract: a full /// buffer returns its length, a short read is UnexpectedEof (backlog#799 B2). fn finish_len(&self, data_len: usize, want: usize) -> std::io::Result { @@ -1152,4 +1230,96 @@ mod tests { let err = r.read(&mut out).await.expect_err("truncated hash must error"); assert_eq!(err.kind(), std::io::ErrorKind::UnexpectedEof); } + + /// `read_appending` must be byte-for-byte identical to `read`, for both the + /// hashed and the no-hash path (rustfs/backlog#1159). It exists so callers can + /// hand over an *uninitialized* buffer; if it ever diverged from `read`, the + /// GET path would silently return different bytes. + #[tokio::test] + async fn read_appending_matches_read_for_both_paths() { + for algo in [HashAlgorithm::HighwayHash256, HashAlgorithm::None] { + const SHARD: usize = 4096; + let data: Vec = (0..SHARD).map(|i| (i * 31 + 7) as u8).collect(); + + let mut encoded = Vec::new(); + let mut w = BitrotWriter::new(&mut encoded, SHARD, algo.clone()); + w.write(&data).await.expect("write shard"); + + let mut via_read = vec![0u8; SHARD]; + let n1 = BitrotReader::new(Cursor::new(encoded.clone()), SHARD, algo.clone(), false) + .read(&mut via_read) + .await + .expect("read"); + + // A buffer with only capacity — no initialized bytes at all. + let mut via_append: Vec = Vec::with_capacity(SHARD); + let n2 = BitrotReader::new(Cursor::new(encoded), SHARD, algo.clone(), false) + .read_appending(&mut via_append, SHARD) + .await + .expect("read_appending"); + + assert_eq!(n1, n2, "{algo:?}: both must report the same length"); + assert_eq!(via_append.len(), n2, "{algo:?}: the buffer grows by exactly n"); + assert_eq!(via_read, via_append, "{algo:?}: bytes must be identical"); + assert_eq!(via_append, data, "{algo:?}: and equal to what was written"); + } + } + + /// A truncated shard must be an error, never a partially filled buffer — that + /// contract is what lets the pool hand out uninitialized capacity. + #[tokio::test] + async fn read_appending_rejects_a_short_shard_instead_of_returning_partial_bytes() { + for algo in [HashAlgorithm::HighwayHash256, HashAlgorithm::None] { + const SHARD: usize = 4096; + let data = vec![9u8; SHARD]; + let mut encoded = Vec::new(); + let mut w = BitrotWriter::new(&mut encoded, SHARD, algo.clone()); + w.write(&data).await.expect("write shard"); + encoded.truncate(encoded.len() - 1); + + let mut out: Vec = Vec::with_capacity(SHARD); + let err = BitrotReader::new(Cursor::new(encoded), SHARD, algo.clone(), false) + .read_appending(&mut out, SHARD) + .await + .expect_err("a truncated shard must not succeed"); + assert_eq!(err.kind(), std::io::ErrorKind::UnexpectedEof, "{algo:?}"); + assert!( + out.len() < SHARD, + "{algo:?}: a failed read must not claim a full shard; the caller drops the buffer" + ); + } + } + + /// A corrupt shard must fail verification, and the corrupt bytes must NOT be + /// appended: `read_appending` writes into a buffer the caller may recycle. + #[tokio::test] + async fn read_appending_does_not_expose_bytes_that_fail_the_hash() { + const SHARD: usize = 4096; + let algo = HashAlgorithm::HighwayHash256; + let data = vec![3u8; SHARD]; + let mut encoded = Vec::new(); + let mut w = BitrotWriter::new(&mut encoded, SHARD, algo.clone()); + w.write(&data).await.expect("write shard"); + let last = encoded.len() - 1; + encoded[last] ^= 0xff; + + let mut out: Vec = Vec::with_capacity(SHARD); + let err = BitrotReader::new(Cursor::new(encoded), SHARD, algo, false) + .read_appending(&mut out, SHARD) + .await + .expect_err("a corrupt shard must not verify"); + assert_eq!(err.kind(), std::io::ErrorKind::InvalidData); + assert!(out.is_empty(), "corrupt bytes must never reach the caller's buffer"); + } + + #[tokio::test] + async fn read_appending_rejects_a_want_larger_than_the_shard() { + let algo = HashAlgorithm::HighwayHash256; + let mut out: Vec = Vec::new(); + let err = BitrotReader::new(Cursor::new(Vec::new()), 16, algo, false) + .read_appending(&mut out, 17) + .await + .expect_err("want > shard_size must be rejected"); + assert_eq!(err.kind(), std::io::ErrorKind::InvalidInput); + } } diff --git a/crates/ecstore/src/erasure/coding/decode.rs b/crates/ecstore/src/erasure/coding/decode.rs index 2b97c8494..11a91ad85 100644 --- a/crates/ecstore/src/erasure/coding/decode.rs +++ b/crates/ecstore/src/erasure/coding/decode.rs @@ -275,13 +275,15 @@ where let role = shard_role(index, data_shards); if let Some(reader) = reader { Box::pin(async move { - let mut buf = recycled_buf.unwrap_or_else(|| vec![0; shard_size]); - debug_assert_eq!(buf.len(), shard_size); + // Capacity, not length: `read_appending` writes every byte it returns, so + // the buffer never needs zeroing first (rustfs/backlog#1159). + let mut buf = recycled_buf.unwrap_or_else(|| Vec::with_capacity(shard_size)); + buf.clear(); let read_start = metrics_path.map(|_| Instant::now()); let read_result = if read_timeout.is_zero() { - reader.read(&mut buf).await + reader.read_appending(&mut buf, shard_size).await } else { - match tokio::time::timeout(read_timeout, reader.read(&mut buf)).await { + match tokio::time::timeout(read_timeout, reader.read_appending(&mut buf, shard_size)).await { Ok(result) => result, Err(_) => { let timeout_error = io::Error::new(ErrorKind::TimedOut, "shard read timed out"); @@ -306,7 +308,7 @@ where match read_result { Ok(n) => { - buf.truncate(n); + debug_assert_eq!(buf.len(), n, "read_appending must grow the buffer by exactly n"); if let Some(path) = metrics_path { rustfs_io_metrics::record_get_object_shard_read_observation( path,