diff --git a/crates/ecstore/src/erasure_codec/bridge.rs b/crates/ecstore/src/erasure_codec/bridge.rs new file mode 100644 index 000000000..133bacaf9 --- /dev/null +++ b/crates/ecstore/src/erasure_codec/bridge.rs @@ -0,0 +1,154 @@ +// 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 crate::erasure_coding::Erasure; +use std::io; + +pub(crate) trait DecodeWorkspace: Send + Sync + 'static { + fn shard_len(&self) -> usize; +} + +pub(crate) trait ErasureDecodeEngine: Send + Sync + 'static { + type Workspace: DecodeWorkspace; + + fn data_shards(&self) -> usize; + fn parity_shards(&self) -> usize; + fn block_size(&self) -> usize; + + fn supports_progressive_decode(&self) -> bool; + fn supports_aligned_shards(&self) -> bool; + + fn prepare_workspace(&self, shard_len: usize) -> io::Result; + + fn reconstruct_into(&self, shards: &mut [Option>], workspace: &mut Self::Workspace) -> io::Result<()>; +} + +fn data_shards_complete(shards: &[Option>], data_shards: usize) -> bool { + shards.len() >= data_shards && shards.iter().take(data_shards).all(Option::is_some) +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(crate) struct LegacyDecodeWorkspace { + shard_len: usize, +} + +impl LegacyDecodeWorkspace { + fn new(shard_len: usize) -> Self { + Self { shard_len } + } +} + +impl DecodeWorkspace for LegacyDecodeWorkspace { + fn shard_len(&self) -> usize { + self.shard_len + } +} + +#[derive(Clone)] +pub(crate) struct LegacyEcDecodeEngine { + erasure: Erasure, +} + +impl LegacyEcDecodeEngine { + pub(crate) fn new(erasure: Erasure) -> Self { + Self { erasure } + } +} + +impl ErasureDecodeEngine for LegacyEcDecodeEngine { + type Workspace = LegacyDecodeWorkspace; + + fn data_shards(&self) -> usize { + self.erasure.data_shards + } + + fn parity_shards(&self) -> usize { + self.erasure.parity_shards + } + + fn block_size(&self) -> usize { + self.erasure.block_size + } + + fn supports_progressive_decode(&self) -> bool { + false + } + + fn supports_aligned_shards(&self) -> bool { + false + } + + fn prepare_workspace(&self, shard_len: usize) -> io::Result { + Ok(LegacyDecodeWorkspace::new(shard_len)) + } + + fn reconstruct_into(&self, shards: &mut [Option>], _workspace: &mut Self::Workspace) -> io::Result<()> { + if data_shards_complete(shards, self.erasure.data_shards) { + return Ok(()); + } + + self.erasure.decode_data(shards) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn data_shards_complete_ignores_parity_slots() { + let shards = vec![Some(vec![1]), Some(vec![2]), None]; + + assert!(data_shards_complete(&shards, 2)); + } + + #[test] + fn data_shards_complete_requires_all_data_slots() { + let missing_data = vec![Some(vec![1]), None, Some(vec![3])]; + let short = vec![Some(vec![1])]; + + assert!(!data_shards_complete(&missing_data, 2)); + assert!(!data_shards_complete(&short, 2)); + } + + #[test] + fn legacy_decode_engine_reports_erasure_shape() { + let engine = LegacyEcDecodeEngine::new(Erasure::new(4, 2, 1 << 20)); + + assert_eq!(engine.data_shards(), 4); + assert_eq!(engine.parity_shards(), 2); + assert_eq!(engine.block_size(), 1 << 20); + assert!(!engine.supports_progressive_decode()); + assert!(!engine.supports_aligned_shards()); + } + + #[test] + fn legacy_decode_engine_reconstructs_missing_data_shard() { + let erasure = Erasure::new(4, 2, 16); + let encoded = erasure + .encode_data(b"codec bridge keeps current reconstruction behavior") + .expect("encode should succeed"); + let mut shards = encoded.into_iter().map(|shard| Some(shard.to_vec())).collect::>(); + shards[1] = None; + + let engine = LegacyEcDecodeEngine::new(erasure); + let mut workspace = engine.prepare_workspace(4).expect("workspace should be prepared"); + engine + .reconstruct_into(&mut shards, &mut workspace) + .expect("legacy engine should reconstruct through current erasure path"); + + assert_eq!(workspace.shard_len(), 4); + assert!(shards.iter().take(engine.data_shards()).all(Option::is_some)); + } +} diff --git a/crates/ecstore/src/erasure_codec/mod.rs b/crates/ecstore/src/erasure_codec/mod.rs new file mode 100644 index 000000000..cdd2dd043 --- /dev/null +++ b/crates/ecstore/src/erasure_codec/mod.rs @@ -0,0 +1,16 @@ +// 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. + +pub(crate) mod bridge; +pub(crate) mod workspace; diff --git a/crates/ecstore/src/erasure_codec/workspace.rs b/crates/ecstore/src/erasure_codec/workspace.rs new file mode 100644 index 000000000..405b89e42 --- /dev/null +++ b/crates/ecstore/src/erasure_codec/workspace.rs @@ -0,0 +1,88 @@ +// 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. + +#[derive(Debug, Default)] +pub(crate) struct ShardBufferPool { + buffers: Vec>>, +} + +impl ShardBufferPool { + #[inline] + pub(crate) fn new(slot_count: usize) -> Self { + let mut buffers = Vec::with_capacity(slot_count); + buffers.resize_with(slot_count, || None); + Self { buffers } + } + + #[inline] + pub(crate) fn ensure_slots(&mut self, slot_count: usize) { + if self.buffers.len() < slot_count { + self.buffers.resize_with(slot_count, || None); + } + } + + // Returned bytes may contain stale scratch data and must be overwritten before use. + #[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)); + if buf.capacity() < len { + buf.reserve_exact(len - buf.capacity()); + } + buf.resize(len, 0); + buf + } + + #[inline] + pub(crate) fn put(&mut self, index: usize, buf: Vec) { + self.ensure_slots(index + 1); + self.buffers[index] = Some(buf); + } + + #[cfg(test)] + fn stored_capacity(&self, index: usize) -> Option { + self.buffers.get(index).and_then(|buf| buf.as_ref().map(Vec::capacity)) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn shard_buffer_pool_reuses_slot_without_clearing() { + let mut pool = ShardBufferPool::new(2); + let mut buf = pool.take(1, 16); + assert_eq!(buf.len(), 16); + buf[0] = 42; + let capacity = buf.capacity(); + + 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); + } + + #[test] + fn shard_buffer_pool_grows_for_sparse_slot_indexes() { + let mut pool = ShardBufferPool::new(0); + let buf = pool.take(3, 4); + + assert_eq!(buf.len(), 4); + assert_eq!(pool.buffers.len(), 4); + } +} diff --git a/crates/ecstore/src/erasure_coding/decode.rs b/crates/ecstore/src/erasure_coding/decode.rs index 4d7faac7b..5b0198df8 100644 --- a/crates/ecstore/src/erasure_coding/decode.rs +++ b/crates/ecstore/src/erasure_coding/decode.rs @@ -14,16 +14,92 @@ use crate::disk::error::Error; use crate::disk::error_reduce::reduce_errs; +use crate::erasure_codec::workspace::ShardBufferPool; use crate::erasure_coding::{BitrotReader, Erasure}; +use crate::get_diagnostics::{ + GET_OBJECT_PATH_LEGACY_DUPLEX, GET_SHARD_READ_OUTCOME_ERROR, GET_SHARD_READ_OUTCOME_MISSING, GET_SHARD_READ_OUTCOME_SUCCESS, + GET_SHARD_ROLE_DATA, GET_SHARD_ROLE_PARITY, GET_STAGE_EMIT, GET_STAGE_RANGE, GET_STAGE_RECONSTRUCT, GET_STAGE_STRIPE_READ, + GET_STAGE_STRIPE_READ_FIRST_SHARD, GET_STAGE_STRIPE_READ_QUORUM, GetObjectFailureReason, classify_io_error, + record_get_object_pipeline_failure, +}; +use crate::set_disk::shard_source::{ShardStripeSource, StripeReadState}; use futures::stream::{FuturesUnordered, StreamExt}; use pin_project_lite::pin_project; +use std::future::Future; use std::io; use std::io::ErrorKind; +use std::pin::Pin; +use std::time::Instant; use tokio::io::AsyncRead; use tokio::io::AsyncWrite; use tokio::io::AsyncWriteExt; use tracing::error; +type ShardReadFuture<'a> = Pin, Error>)> + Send + 'a>>; + +fn shard_role(index: usize, data_shards: usize) -> &'static str { + if index < data_shards { + GET_SHARD_ROLE_DATA + } else { + GET_SHARD_ROLE_PARITY + } +} + +fn read_shard<'a, R>( + index: usize, + reader: &'a mut Option>, + recycled_buf: Option>, + shard_size: usize, + data_shards: usize, + metrics_path: Option<&'static str>, +) -> ShardReadFuture<'a> +where + R: AsyncRead + Unpin + Send + Sync + 'a, +{ + 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); + let read_start = Instant::now(); + match reader.read(&mut buf).await { + Ok(n) => { + buf.truncate(n); + if let Some(path) = metrics_path { + rustfs_io_metrics::record_get_object_shard_read( + path, + role, + GET_SHARD_READ_OUTCOME_SUCCESS, + n, + read_start.elapsed().as_secs_f64(), + ); + } + (index, Ok(buf)) + } + Err(e) => { + if let Some(path) = metrics_path { + rustfs_io_metrics::record_get_object_shard_read( + path, + role, + GET_SHARD_READ_OUTCOME_ERROR, + 0, + read_start.elapsed().as_secs_f64(), + ); + } + (index, Err(Error::from(e))) + } + } + }) + } else { + Box::pin(async move { + if let Some(path) = metrics_path { + rustfs_io_metrics::record_get_object_shard_read(path, role, GET_SHARD_READ_OUTCOME_MISSING, 0, 0.0); + } + (index, Err(Error::FileNotFound)) + }) + } +} + pin_project! { pub(crate) struct ParallelReader { #[pin] @@ -33,9 +109,10 @@ pub(crate) struct ParallelReader { 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>>, + metrics_path: Option<&'static str>, + // Request-scoped shard buffers keyed by shard index. Keeping ownership in + // `ParallelReader` avoids dropping unused parity/backup slot buffers between stripes. + buffers: ShardBufferPool, } } @@ -45,6 +122,16 @@ where { // Readers should handle disk errors before being passed in, ensuring each reader reaches the available number of BitrotReaders pub fn new(readers: Vec>>, e: Erasure, offset: usize, total_length: usize) -> Self { + Self::new_with_metrics_path(readers, e, offset, total_length, None) + } + + pub fn new_with_metrics_path( + readers: Vec>>, + e: Erasure, + offset: usize, + total_length: usize, + metrics_path: Option<&'static str>, + ) -> Self { let shard_size = e.shard_size(); let shard_file_size = e.shard_file_size(total_length as i64) as usize; @@ -59,7 +146,8 @@ where shard_file_size, data_shards: e.data_shards, total_shards: e.data_shards + e.parity_shards, - recycled: Vec::new(), + metrics_path, + buffers: ShardBufferPool::new(e.data_shards + e.parity_shards), } } } @@ -87,79 +175,127 @@ where let mut shards: Vec>> = 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); + self.buffers.ensure_slots(num_readers); - let mut futures = Vec::with_capacity(self.total_shards); let reader_iter: std::slice::IterMut<'_, Option>> = 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 = recycled_buf.unwrap_or_default(); - buf.resize(shard_size, 0); - match reader.read(&mut buf).await { - Ok(n) => { - buf.truncate(n); - (i, Ok(buf)) - } - Err(e) => (i, Err(Error::from(e))), - } - }) as std::pin::Pin, Error>)> + Send>> - } else { - // Return FileNotFound error when reader is None - Box::pin(async move { (i, Err(Error::FileNotFound)) }) - as std::pin::Pin, Error>)> + Send>> - }; - - futures.push(future); - } - - if futures.len() >= self.data_shards { - let mut fut_iter = futures.into_iter(); + if num_readers >= self.data_shards { + let mut reader_iter = reader_iter.enumerate(); let mut sets = FuturesUnordered::new(); + let stripe_read_start = Instant::now(); + let mut scheduled = 0usize; for _ in 0..self.data_shards { - if let Some(future) = fut_iter.next() { - sets.push(future); + if let Some((i, reader)) = reader_iter.next() { + // Only claim a request-scoped buffer when a shard will actually be read. + let recycled_buf = if reader.is_some() { + Some(self.buffers.take(i, shard_size)) + } else { + None + }; + scheduled += 1; + sets.push(read_shard(i, reader, recycled_buf, shard_size, self.data_shards, self.metrics_path)); } } let mut success = 0; + let mut completed = 0usize; + let mut failed = 0usize; + let mut first_shard_recorded = false; + let mut quorum_recorded = false; while let Some((i, result)) = sets.next().await { + completed += 1; + if !first_shard_recorded { + if let Some(path) = self.metrics_path { + rustfs_io_metrics::record_get_object_stage_duration( + path, + GET_STAGE_STRIPE_READ_FIRST_SHARD, + stripe_read_start.elapsed().as_secs_f64(), + ); + } + first_shard_recorded = true; + } + match result { Ok(v) => { shards[i] = Some(v); success += 1; } Err(e) => { + failed += 1; errs[i] = Some(e); - if let Some(future) = fut_iter.next() { - sets.push(future); + if let Some((next_i, next_reader)) = reader_iter.next() { + let recycled_buf = if next_reader.is_some() { + Some(self.buffers.take(next_i, shard_size)) + } else { + None + }; + scheduled += 1; + sets.push(read_shard( + next_i, + next_reader, + recycled_buf, + shard_size, + self.data_shards, + self.metrics_path, + )); } } } if success >= self.data_shards { + if let Some(path) = self.metrics_path { + rustfs_io_metrics::record_get_object_stage_duration( + path, + GET_STAGE_STRIPE_READ_QUORUM, + stripe_read_start.elapsed().as_secs_f64(), + ); + rustfs_io_metrics::record_get_object_shard_read_fanout(path, scheduled, completed, success, failed); + } + quorum_recorded = true; break; } } + + if !quorum_recorded && let Some(path) = self.metrics_path { + rustfs_io_metrics::record_get_object_stage_duration( + path, + GET_STAGE_STRIPE_READ_QUORUM, + stripe_read_start.elapsed().as_secs_f64(), + ); + rustfs_io_metrics::record_get_object_shard_read_fanout(path, scheduled, completed, success, failed); + } } (shards, errs) } + pub fn recycle_shards(&mut self, shards: &mut [Option>]) { + for (i, reader) in self.readers.iter().enumerate() { + if reader.is_some() + && let Some(buf) = shards.get_mut(i).and_then(Option::take) + { + self.buffers.put(i, buf); + } + } + } + pub fn can_decode(&self, shards: &[Option>]) -> bool { shards.iter().filter(|s| s.is_some()).count() >= self.data_shards } } +#[async_trait::async_trait] +impl ShardStripeSource for ParallelReader +where + R: AsyncRead + Unpin + Send + Sync, +{ + async fn read_next_stripe(&mut self) -> StripeReadState { + let read_quorum = self.data_shards; + let (shards, errors) = ParallelReader::read(self).await; + StripeReadState::from_parts(shards, errors, read_quorum) + } +} + /// Get the total length of data blocks fn get_data_block_len(shards: &[Option>], data_blocks: usize) -> usize { let mut size = 0; @@ -182,6 +318,15 @@ where W: tokio::io::AsyncWrite + Send + Sync + Unpin, { if en_blocks.len() < data_blocks { + let reason = GetObjectFailureReason::RangeOrLengthInvalid; + record_get_object_pipeline_failure(GET_STAGE_RANGE, reason); + error!( + data_blocks, + available_shards = en_blocks.len(), + stage = GET_STAGE_RANGE, + reason = reason.as_str(), + "Write data blocks received fewer shards than data blocks" + ); return Err(io::Error::new(ErrorKind::InvalidInput, "data block count exceeds available shards")); } @@ -190,10 +335,31 @@ where } let Some(required_len) = offset.checked_add(length) else { + let reason = GetObjectFailureReason::RangeOrLengthInvalid; + record_get_object_pipeline_failure(GET_STAGE_RANGE, reason); + error!( + offset, + length, + stage = GET_STAGE_RANGE, + reason = reason.as_str(), + "Write data blocks offset and length overflow" + ); return Err(io::Error::new(ErrorKind::InvalidInput, "offset + length overflows")); }; - if get_data_block_len(en_blocks, data_blocks) < required_len { - error!("write_data_blocks not enough data after offset"); + let data_len = get_data_block_len(en_blocks, data_blocks); + if data_len < required_len { + let reason = GetObjectFailureReason::ShortRead; + error!( + data_blocks, + data_len, + required_len, + offset, + length, + stage = GET_STAGE_EMIT, + reason = reason.as_str(), + "Write data blocks had insufficient data after offset" + ); + record_get_object_pipeline_failure(GET_STAGE_EMIT, reason); return Err(io::Error::new(ErrorKind::UnexpectedEof, "Not enough data blocks to write")); } @@ -202,7 +368,16 @@ where for block_op in &en_blocks[..data_blocks] { let Some(block) = block_op else { - error!("write_data_blocks block_op.is_none()"); + let reason = GetObjectFailureReason::ShortRead; + error!( + data_blocks, + offset, + length, + stage = GET_STAGE_EMIT, + reason = reason.as_str(), + "Write data blocks found a missing data shard" + ); + record_get_object_pipeline_failure(GET_STAGE_EMIT, reason); return Err(io::Error::new(ErrorKind::UnexpectedEof, "Missing data block")); }; @@ -215,10 +390,23 @@ where offset = 0; let write_len = write_left.min(block_slice.len()); - writer.write_all(&block_slice[..write_len]).await.map_err(|e| { - error!("write_data_blocks write_all err: {}", e); - e - })?; + let write_stage_start = Instant::now(); + if let Err(e) = writer.write_all(&block_slice[..write_len]).await { + rustfs_io_metrics::record_get_object_duplex_backpressure_duration(write_stage_start.elapsed().as_secs_f64()); + let reason = classify_io_error(&e); + record_get_object_pipeline_failure(GET_STAGE_EMIT, reason); + error!( + write_len, + total_written, + write_left, + stage = GET_STAGE_EMIT, + reason = reason.as_str(), + error = ?e, + "Write data blocks failed to emit bytes" + ); + return Err(e); + } + rustfs_io_metrics::record_get_object_duplex_backpressure_duration(write_stage_start.elapsed().as_secs_f64()); total_written += write_len; write_left -= write_len; @@ -228,7 +416,16 @@ where } } - error!("write_data_blocks loop exhausted with write_left>0"); + let reason = GetObjectFailureReason::ShortRead; + error!( + total_written, + write_left, + length, + stage = GET_STAGE_EMIT, + reason = reason.as_str(), + "Write data blocks exhausted data before completing requested length" + ); + record_get_object_pipeline_failure(GET_STAGE_EMIT, reason); Err(io::Error::new(ErrorKind::UnexpectedEof, "Not enough data blocks to write")) } @@ -246,13 +443,16 @@ impl Erasure { R: AsyncRead + Unpin + Send + Sync, { if readers.len() != self.data_shards + self.parity_shards { + record_get_object_pipeline_failure(GET_STAGE_RANGE, GetObjectFailureReason::RangeOrLengthInvalid); return (0, Some(io::Error::new(ErrorKind::InvalidInput, "Invalid number of readers"))); } let Some(end_offset) = offset.checked_add(length) else { + record_get_object_pipeline_failure(GET_STAGE_RANGE, GetObjectFailureReason::RangeOrLengthInvalid); return (0, Some(io::Error::new(ErrorKind::InvalidInput, "offset + length exceeds total length"))); }; if end_offset > total_length { + record_get_object_pipeline_failure(GET_STAGE_RANGE, GetObjectFailureReason::RangeOrLengthInvalid); return (0, Some(io::Error::new(ErrorKind::InvalidInput, "offset + length exceeds total length"))); } @@ -264,7 +464,13 @@ impl Erasure { let mut written = 0; - let mut reader = ParallelReader::new(readers, self.clone(), offset, total_length); + let mut reader = ParallelReader::new_with_metrics_path( + readers, + self.clone(), + offset, + total_length, + Some(GET_OBJECT_PATH_LEGACY_DUPLEX), + ); let start = offset / self.block_size; let end = end_offset.saturating_sub(1) / self.block_size; @@ -286,7 +492,13 @@ impl Erasure { break; } + let stripe_read_stage_start = Instant::now(); let (mut shards, errs) = reader.read().await; + rustfs_io_metrics::record_get_object_stage_duration( + "legacy_duplex", + "stripe_read", + stripe_read_stage_start.elapsed().as_secs_f64(), + ); if ret_err.is_none() && let (_, Some(err)) = reduce_errs(&errs, &[]) @@ -296,22 +508,77 @@ impl Erasure { } if !reader.can_decode(&shards) { - error!("erasure decode can_decode errs: {:?}", &errs); + let reason = GetObjectFailureReason::ReadQuorum; + error!( + data_shards = self.data_shards, + total_shards = self.data_shards + self.parity_shards, + available_shards = shards.iter().filter(|shard| shard.is_some()).count(), + block_offset, + block_length, + stage = GET_STAGE_STRIPE_READ, + reason = reason.as_str(), + errors = ?errs, + "Erasure decode could not gather enough shards" + ); + record_get_object_pipeline_failure(GET_STAGE_STRIPE_READ, reason); ret_err = Some(Error::ErasureReadQuorum.into()); break; } // Decode the shards + let reconstruct_stage_start = Instant::now(); if let Err(e) = self.decode_data(&mut shards) { - error!("erasure decode decode_data err: {:?}", e); + rustfs_io_metrics::record_get_object_stage_duration( + "legacy_duplex", + "reconstruct", + reconstruct_stage_start.elapsed().as_secs_f64(), + ); + let reason = GetObjectFailureReason::DecodeError; + error!( + data_shards = self.data_shards, + total_shards = self.data_shards + self.parity_shards, + block_offset, + block_length, + stage = GET_STAGE_RECONSTRUCT, + reason = reason.as_str(), + error = ?e, + "Erasure shard reconstruction failed" + ); + record_get_object_pipeline_failure(GET_STAGE_RECONSTRUCT, reason); ret_err = Some(e); break; } + rustfs_io_metrics::record_get_object_stage_duration( + "legacy_duplex", + "reconstruct", + reconstruct_stage_start.elapsed().as_secs_f64(), + ); + let emit_stage_start = Instant::now(); let n = match write_data_blocks(writer, &shards, self.data_shards, block_offset, block_length).await { - Ok(n) => n, + Ok(n) => { + rustfs_io_metrics::record_get_object_stage_duration( + "legacy_duplex", + "emit", + emit_stage_start.elapsed().as_secs_f64(), + ); + n + } Err(e) => { - error!("erasure decode write_data_blocks err: {:?}", e); + rustfs_io_metrics::record_get_object_stage_duration( + "legacy_duplex", + "emit", + emit_stage_start.elapsed().as_secs_f64(), + ); + error!( + block_offset, + block_length, + bytes_written = written, + stage = GET_STAGE_EMIT, + reason = classify_io_error(&e).as_str(), + error = ?e, + "Erasure decode failed to emit reconstructed data" + ); ret_err = Some(e); break; } @@ -319,16 +586,9 @@ 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; + // Hand active-reader buffers back so the next stripe can reuse them + // without retaining offline shard data that cannot be read again. + reader.recycle_shards(&mut shards); } if ret_err.is_some() { diff --git a/crates/ecstore/src/erasure_coding/decode_reader.rs b/crates/ecstore/src/erasure_coding/decode_reader.rs new file mode 100644 index 000000000..2b09e37c7 --- /dev/null +++ b/crates/ecstore/src/erasure_coding/decode_reader.rs @@ -0,0 +1,676 @@ +// 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 crate::disk::error::Error as DiskError; +use crate::erasure_codec::bridge::ErasureDecodeEngine; +use crate::get_diagnostics::{ + GET_OBJECT_PATH_CODEC_STREAMING, GET_READER_BUFFER_OUTPUT, GET_READER_BUFFER_PREFETCH, GET_READER_POLL_PENDING, + GET_READER_POLL_READY_DATA, GET_READER_POLL_READY_EMPTY, GET_READER_POLL_READY_ERROR, GET_READER_PREFETCH_DIRECT, + GET_READER_PREFETCH_EOF, GET_READER_PREFETCH_ERROR_DEFERRED, GET_READER_PREFETCH_ERROR_IMMEDIATE, GET_READER_PREFETCH_STORED, + GET_STAGE_DECODE, GET_STAGE_EMIT, GET_STAGE_FILL, GET_STAGE_OUTPUT_LOCK_WAIT, GET_STAGE_OUTPUT_POLL, GET_STAGE_RECONSTRUCT, + GET_STAGE_STRIPE_READ, +}; +use crate::set_disk::shard_source::{ShardStripeSource, StripeReadState}; +use std::io; +use std::io::ErrorKind; +use std::pin::Pin; +use std::sync::Mutex; +use std::task::{Context, Poll, ready}; +use std::time::Instant; +use tokio::io::{AsyncRead, ReadBuf}; +use tokio::task::JoinHandle; + +type FillTask = JoinHandle>; + +struct FillResult { + source: S, + workspace: W, + result: io::Result>>, +} + +pub(crate) struct ErasureDecodeReader +where + E: ErasureDecodeEngine, +{ + source: Option, + engine: E, + workspace: Option, + output_buf: Vec, + output_pos: usize, + prefetched_buf: Option>, + prefetch_error: Option, + prefetch_wait_started_at: Option, + remaining: usize, + // Bounded lookahead: at most one background stripe read/decode is in flight. + fill: Option>, +} + +impl ErasureDecodeReader +where + S: ShardStripeSource + Send + 'static, + E: ErasureDecodeEngine + Clone + Send + Sync + 'static, +{ + pub(crate) fn new(source: S, engine: E, total_length: usize) -> io::Result { + if engine.data_shards() == 0 { + return Err(io::Error::new(ErrorKind::InvalidInput, "erasure reader requires data shards")); + } + if engine.block_size() == 0 { + return Err(io::Error::new(ErrorKind::InvalidInput, "erasure reader requires non-zero block size")); + } + + let shard_len = engine.block_size().div_ceil(engine.data_shards()); + let workspace = engine.prepare_workspace(shard_len)?; + + Ok(Self { + source: Some(source), + engine, + workspace: Some(workspace), + output_buf: Vec::new(), + output_pos: 0, + prefetched_buf: None, + prefetch_error: None, + prefetch_wait_started_at: None, + remaining: total_length, + fill: None, + }) + } + + fn poll_fill_result(&mut self, cx: &mut Context<'_>) -> Poll>>> { + if self.fill.is_none() { + let Some(mut source) = self.source.take() else { + return Poll::Ready(Err(io::Error::new(ErrorKind::BrokenPipe, "erasure reader source missing"))); + }; + let Some(mut workspace) = self.workspace.take() else { + self.source = Some(source); + return Poll::Ready(Err(io::Error::new(ErrorKind::BrokenPipe, "erasure reader workspace missing"))); + }; + + let engine = self.engine.clone(); + let remaining = self.remaining; + self.fill = Some(tokio::spawn(async move { + let fill_stage_start = Instant::now(); + let stripe_read_stage_start = Instant::now(); + let state = source.read_next_stripe().await; + rustfs_io_metrics::record_get_object_stage_duration( + GET_OBJECT_PATH_CODEC_STREAMING, + GET_STAGE_STRIPE_READ, + stripe_read_stage_start.elapsed().as_secs_f64(), + ); + let decode_stage_start = Instant::now(); + let result = decode_stripe(&engine, &mut workspace, state, remaining); + rustfs_io_metrics::record_get_object_stage_duration( + GET_OBJECT_PATH_CODEC_STREAMING, + GET_STAGE_DECODE, + decode_stage_start.elapsed().as_secs_f64(), + ); + rustfs_io_metrics::record_get_object_stage_duration( + GET_OBJECT_PATH_CODEC_STREAMING, + GET_STAGE_FILL, + fill_stage_start.elapsed().as_secs_f64(), + ); + FillResult { + source, + workspace, + result, + } + })); + } + + let fill = self + .fill + .as_mut() + .ok_or_else(|| io::Error::new(ErrorKind::BrokenPipe, "erasure reader fill future missing"))?; + let fill_result = ready!(Pin::new(fill).poll(cx)); + let FillResult { + source, + workspace, + result, + } = match fill_result { + Ok(result) => result, + Err(err) => { + self.fill = None; + return Poll::Ready(Err(io::Error::other(format!("erasure reader fill task failed: {err}")))); + } + }; + + self.source = Some(source); + self.workspace = Some(workspace); + self.fill = None; + + match result { + Ok(Some(buf)) => { + if buf.is_empty() && self.remaining > 0 { + return Poll::Ready(Err(DiskError::LessData.into())); + } + rustfs_io_metrics::record_get_object_reader_stripe(GET_OBJECT_PATH_CODEC_STREAMING); + rustfs_io_metrics::record_get_object_reader_bytes(GET_OBJECT_PATH_CODEC_STREAMING, buf.len()); + self.remaining -= buf.len(); + Poll::Ready(Ok(Some(buf))) + } + Ok(None) => { + if self.remaining == 0 { + Poll::Ready(Ok(None)) + } else { + Poll::Ready(Err(DiskError::LessData.into())) + } + } + Err(err) => Poll::Ready(Err(err)), + } + } + + fn poll_prefetch(&mut self, cx: &mut Context<'_>) -> Poll> { + if self.prefetched_buf.is_some() || self.prefetch_error.is_some() || self.remaining == 0 { + return Poll::Ready(Ok(())); + } + + if self.prefetch_wait_started_at.is_none() { + self.prefetch_wait_started_at = Some(Instant::now()); + } + + let fill = match self.poll_fill_result(cx) { + Poll::Ready(result) => { + if let Some(started_at) = self.prefetch_wait_started_at.take() { + rustfs_io_metrics::record_get_object_reader_prefetch_wait( + GET_OBJECT_PATH_CODEC_STREAMING, + started_at.elapsed().as_secs_f64(), + ); + } + result + } + Poll::Pending => return Poll::Pending, + }; + + match fill { + Ok(Some(buf)) => { + if self.output_pos < self.output_buf.len() { + rustfs_io_metrics::record_get_object_reader_prefetch( + GET_OBJECT_PATH_CODEC_STREAMING, + GET_READER_PREFETCH_STORED, + ); + rustfs_io_metrics::record_get_object_reader_buffer( + GET_OBJECT_PATH_CODEC_STREAMING, + GET_READER_BUFFER_PREFETCH, + buf.len(), + ); + self.prefetched_buf = Some(buf); + } else { + rustfs_io_metrics::record_get_object_reader_prefetch( + GET_OBJECT_PATH_CODEC_STREAMING, + GET_READER_PREFETCH_DIRECT, + ); + rustfs_io_metrics::record_get_object_reader_buffer( + GET_OBJECT_PATH_CODEC_STREAMING, + GET_READER_BUFFER_OUTPUT, + buf.len(), + ); + self.output_buf = buf; + self.output_pos = 0; + } + Poll::Ready(Ok(())) + } + Ok(None) => { + rustfs_io_metrics::record_get_object_reader_prefetch(GET_OBJECT_PATH_CODEC_STREAMING, GET_READER_PREFETCH_EOF); + Poll::Ready(Ok(())) + } + Err(err) => { + if self.output_pos < self.output_buf.len() { + rustfs_io_metrics::record_get_object_reader_prefetch( + GET_OBJECT_PATH_CODEC_STREAMING, + GET_READER_PREFETCH_ERROR_DEFERRED, + ); + self.prefetch_error = Some(err); + Poll::Ready(Ok(())) + } else { + rustfs_io_metrics::record_get_object_reader_prefetch( + GET_OBJECT_PATH_CODEC_STREAMING, + GET_READER_PREFETCH_ERROR_IMMEDIATE, + ); + Poll::Ready(Err(err)) + } + } + } + } +} + +impl Drop for ErasureDecodeReader +where + E: ErasureDecodeEngine, +{ + fn drop(&mut self) { + if let Some(fill) = self.fill.take() { + fill.abort(); + } + } +} + +impl Unpin for ErasureDecodeReader where E: ErasureDecodeEngine {} + +impl AsyncRead for ErasureDecodeReader +where + S: ShardStripeSource + Send + 'static, + E: ErasureDecodeEngine + Clone + Send + Sync + 'static, +{ + fn poll_read(mut self: Pin<&mut Self>, cx: &mut Context<'_>, buf: &mut ReadBuf<'_>) -> Poll> { + loop { + if self.output_pos < self.output_buf.len() { + if self.prefetched_buf.is_none() + && self.prefetch_error.is_none() + && self.remaining > 0 + && let Poll::Ready(result) = self.poll_prefetch(cx) + { + result?; + } + + let available = &self.output_buf[self.output_pos..]; + let read_buf_remaining_before = buf.remaining(); + let output_remaining_before = available.len(); + let copy_len = available.len().min(buf.remaining()); + let copy_start = Instant::now(); + buf.put_slice(&available[..copy_len]); + self.output_pos += copy_len; + if copy_len > 0 { + rustfs_io_metrics::record_get_object_reader_copy( + GET_OBJECT_PATH_CODEC_STREAMING, + copy_len, + read_buf_remaining_before, + output_remaining_before, + copy_start.elapsed().as_secs_f64(), + ); + } + return Poll::Ready(Ok(())); + } + + if let Some(next_buf) = self.prefetched_buf.take() { + rustfs_io_metrics::record_get_object_reader_buffer( + GET_OBJECT_PATH_CODEC_STREAMING, + GET_READER_BUFFER_OUTPUT, + next_buf.len(), + ); + self.output_buf = next_buf; + self.output_pos = 0; + continue; + } + + if let Some(err) = self.prefetch_error.take() { + return Poll::Ready(Err(err)); + } + + if self.remaining == 0 { + return Poll::Ready(Ok(())); + } + + ready!(self.poll_prefetch(cx))?; + } + } +} + +pub(crate) struct SyncErasureDecodeReader { + inner: Mutex, +} + +impl SyncErasureDecodeReader { + pub(crate) fn new(inner: R) -> Self { + Self { + inner: Mutex::new(inner), + } + } +} + +impl AsyncRead for SyncErasureDecodeReader +where + R: AsyncRead + Unpin + Send, +{ + fn poll_read(self: Pin<&mut Self>, cx: &mut Context<'_>, buf: &mut ReadBuf<'_>) -> Poll> { + let lock_wait_start = Instant::now(); + let mut inner = match self.inner.lock() { + Ok(inner) => { + rustfs_io_metrics::record_get_object_stage_duration( + GET_OBJECT_PATH_CODEC_STREAMING, + GET_STAGE_OUTPUT_LOCK_WAIT, + lock_wait_start.elapsed().as_secs_f64(), + ); + inner + } + Err(_) => { + rustfs_io_metrics::record_get_object_stage_duration( + GET_OBJECT_PATH_CODEC_STREAMING, + GET_STAGE_OUTPUT_LOCK_WAIT, + lock_wait_start.elapsed().as_secs_f64(), + ); + return Poll::Ready(Err(io::Error::other("erasure decode reader lock poisoned"))); + } + }; + let read_buf_remaining_before = buf.remaining(); + let filled_before = buf.filled().len(); + let poll_start = Instant::now(); + let result = Pin::new(&mut *inner).poll_read(cx, buf); + let poll_duration = poll_start.elapsed().as_secs_f64(); + let filled_bytes = buf.filled().len().saturating_sub(filled_before); + let poll_outcome = match &result { + Poll::Ready(Ok(())) if filled_bytes > 0 => GET_READER_POLL_READY_DATA, + Poll::Ready(Ok(())) => GET_READER_POLL_READY_EMPTY, + Poll::Ready(Err(_)) => GET_READER_POLL_READY_ERROR, + Poll::Pending => GET_READER_POLL_PENDING, + }; + rustfs_io_metrics::record_get_object_stage_duration( + GET_OBJECT_PATH_CODEC_STREAMING, + GET_STAGE_OUTPUT_POLL, + poll_duration, + ); + rustfs_io_metrics::record_get_object_reader_poll( + GET_OBJECT_PATH_CODEC_STREAMING, + poll_outcome, + read_buf_remaining_before, + filled_bytes, + poll_duration, + ); + result + } +} + +fn decode_stripe( + engine: &E, + workspace: &mut E::Workspace, + state: StripeReadState, + remaining: usize, +) -> io::Result>> +where + E: ErasureDecodeEngine, +{ + if state.slots().is_empty() { + return Ok(None); + } + if !state.can_decode() { + return Err(DiskError::ErasureReadQuorum.into()); + } + + let reconstruct_stage_start = Instant::now(); + if state.data_shards_complete(engine.data_shards()) { + rustfs_io_metrics::record_get_object_stage_duration( + GET_OBJECT_PATH_CODEC_STREAMING, + GET_STAGE_RECONSTRUCT, + reconstruct_stage_start.elapsed().as_secs_f64(), + ); + let emit_stage_start = Instant::now(); + let output = emit_data_shards(&state, engine.data_shards(), engine.block_size(), remaining)?; + rustfs_io_metrics::record_get_object_stage_duration( + GET_OBJECT_PATH_CODEC_STREAMING, + GET_STAGE_EMIT, + emit_stage_start.elapsed().as_secs_f64(), + ); + return Ok(Some(output)); + } + + let (mut shards, _errs) = state.into_parts(); + if let Err(err) = engine.reconstruct_into(&mut shards, workspace) { + rustfs_io_metrics::record_get_object_stage_duration( + GET_OBJECT_PATH_CODEC_STREAMING, + GET_STAGE_RECONSTRUCT, + reconstruct_stage_start.elapsed().as_secs_f64(), + ); + return Err(err); + } + rustfs_io_metrics::record_get_object_stage_duration( + GET_OBJECT_PATH_CODEC_STREAMING, + GET_STAGE_RECONSTRUCT, + reconstruct_stage_start.elapsed().as_secs_f64(), + ); + + if shards.len() < engine.data_shards() { + return Err(io::Error::new( + ErrorKind::UnexpectedEof, + "decoded stripe has fewer shards than data shard count", + )); + } + + let emit_stage_start = Instant::now(); + let mut output = Vec::with_capacity(engine.block_size().min(remaining)); + for shard in shards.iter().take(engine.data_shards()) { + if output.len() >= remaining { + break; + } + let Some(shard) = shard else { + return Err(io::Error::new(ErrorKind::UnexpectedEof, "decoded stripe is missing a data shard")); + }; + let copy_len = shard.len().min(remaining - output.len()); + output.extend_from_slice(&shard[..copy_len]); + } + rustfs_io_metrics::record_get_object_stage_duration( + GET_OBJECT_PATH_CODEC_STREAMING, + GET_STAGE_EMIT, + emit_stage_start.elapsed().as_secs_f64(), + ); + + Ok(Some(output)) +} + +fn emit_data_shards(state: &StripeReadState, data_shards: usize, block_size: usize, remaining: usize) -> io::Result> { + let mut output = Vec::with_capacity(block_size.min(remaining)); + for index in 0..data_shards { + if output.len() >= remaining { + break; + } + let Some(slot) = state.slot_by_index(index) else { + return Err(io::Error::new(ErrorKind::UnexpectedEof, "decoded stripe is missing a data shard")); + }; + let Some(shard) = slot.data_bytes() else { + return Err(io::Error::new(ErrorKind::UnexpectedEof, "decoded stripe is missing a data shard")); + }; + let copy_len = shard.len().min(remaining - output.len()); + output.extend_from_slice(&shard[..copy_len]); + } + Ok(output) +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::erasure_codec::bridge::LegacyEcDecodeEngine; + use crate::erasure_coding::Erasure; + use crate::set_disk::shard_source::{ShardSlot, StripeReadState}; + use std::collections::VecDeque; + use std::sync::Arc; + use std::sync::atomic::{AtomicUsize, Ordering}; + use tokio::io::AsyncReadExt; + use tokio::task::yield_now; + use tokio::time::{Duration, timeout}; + + struct VecStripeSource { + stripes: VecDeque, + read_quorum: usize, + read_count: Option>, + } + + #[async_trait::async_trait] + impl ShardStripeSource for VecStripeSource { + async fn read_next_stripe(&mut self) -> StripeReadState { + if let Some(read_count) = &self.read_count { + read_count.fetch_add(1, Ordering::SeqCst); + } + self.stripes + .pop_front() + .unwrap_or_else(|| StripeReadState::new(Vec::new(), self.read_quorum)) + } + } + + fn source_from_data(erasure: &Erasure, data: &[u8], missing_indexes: &[usize]) -> VecStripeSource { + let read_quorum = erasure.data_shards; + let stripes = data + .chunks(erasure.block_size) + .map(|chunk| { + let shards = erasure + .encode_data(chunk) + .expect("test stripe should encode") + .into_iter() + .enumerate() + .map(|(index, shard)| { + if missing_indexes.contains(&index) { + None + } else { + Some(shard.to_vec()) + } + }) + .collect(); + StripeReadState::from_parts(shards, Vec::new(), read_quorum) + }) + .collect(); + + VecStripeSource { + stripes, + read_quorum, + read_count: None, + } + } + + async fn decode_all(erasure: Erasure, data: &[u8], missing_indexes: &[usize]) -> io::Result> { + let source = source_from_data(&erasure, data, missing_indexes); + let engine = LegacyEcDecodeEngine::new(erasure); + let mut reader = ErasureDecodeReader::new(source, engine, data.len())?; + let mut decoded = Vec::new(); + reader.read_to_end(&mut decoded).await?; + Ok(decoded) + } + + #[tokio::test] + async fn erasure_decode_reader_reads_single_stripe() { + let erasure = Erasure::new(4, 2, 64); + let data = b"single stripe decode reader output"; + + let decoded = decode_all(erasure, data, &[]) + .await + .expect("single stripe reader should decode"); + + assert_eq!(decoded, data); + } + + #[tokio::test] + async fn erasure_decode_reader_reads_multiple_stripes() { + let erasure = Erasure::new(4, 2, 32); + let data = (0..150u16).map(|value| value.to_le_bytes()[0]).collect::>(); + + let decoded = decode_all(erasure, &data, &[]) + .await + .expect("multi stripe reader should decode"); + + assert_eq!(decoded, data); + } + + #[tokio::test] + async fn erasure_decode_reader_stops_at_eof_for_empty_object() { + let erasure = Erasure::new(4, 2, 32); + let source = source_from_data(&erasure, &[], &[]); + let engine = LegacyEcDecodeEngine::new(erasure); + let mut reader = ErasureDecodeReader::new(source, engine, 0).expect("empty reader should be constructed"); + let mut decoded = Vec::new(); + + let read = reader + .read_to_end(&mut decoded) + .await + .expect("empty reader should finish without reading stripes"); + + assert_eq!(read, 0); + assert!(decoded.is_empty()); + } + + #[tokio::test] + async fn erasure_decode_reader_reconstructs_missing_data_shard() { + let erasure = Erasure::new(4, 2, 32); + let data = (0..120u16) + .map(|value| value.wrapping_mul(17).to_le_bytes()[0]) + .collect::>(); + + let decoded = decode_all(erasure, &data, &[1]) + .await + .expect("reader should reconstruct one missing data shard"); + + assert_eq!(decoded, data); + } + + #[tokio::test] + async fn erasure_decode_reader_reads_when_only_parity_shards_are_missing() { + let erasure = Erasure::new(4, 2, 32); + let data = (0..120u16) + .map(|value| value.wrapping_mul(11).to_le_bytes()[0]) + .collect::>(); + + let decoded = decode_all(erasure, &data, &[4, 5]) + .await + .expect("reader should emit complete data shards without parity reconstruction"); + + assert_eq!(decoded, data); + } + + #[test] + fn emit_data_shards_preserves_output_order_for_out_of_order_slots() { + let state = StripeReadState::new( + vec![ + ShardSlot::data(1, b"cd".to_vec()), + ShardSlot::data(0, b"ab".to_vec()), + ShardSlot::data(2, b"ef".to_vec()), + ], + 2, + ); + + let output = emit_data_shards(&state, 3, 6, 5).expect("out-of-order data slots should emit by shard index"); + + assert_eq!(output, b"abcde"); + } + + #[tokio::test] + async fn erasure_decode_reader_reports_short_source() { + let erasure = Erasure::new(4, 2, 32); + let source = VecStripeSource { + stripes: VecDeque::new(), + read_quorum: erasure.data_shards, + read_count: None, + }; + let engine = LegacyEcDecodeEngine::new(erasure); + let mut reader = ErasureDecodeReader::new(source, engine, 1).expect("reader should be constructed"); + let mut decoded = Vec::new(); + + let err = reader + .read_to_end(&mut decoded) + .await + .expect_err("reader should reject EOF before requested length"); + + assert_eq!(err.kind(), ErrorKind::Other); + assert!(decoded.is_empty()); + } + + #[tokio::test] + async fn erasure_decode_reader_prefetches_next_stripe_while_output_remains() { + let erasure = Erasure::new(4, 2, 32); + let data = (0..96u16) + .map(|value| value.wrapping_mul(3).to_le_bytes()[0]) + .collect::>(); + let read_count = Arc::new(AtomicUsize::new(0)); + let mut source = source_from_data(&erasure, &data, &[]); + source.read_count = Some(Arc::clone(&read_count)); + let engine = LegacyEcDecodeEngine::new(erasure); + let mut reader = ErasureDecodeReader::new(source, engine, data.len()).expect("reader should be constructed"); + let mut first_read = [0u8; 1]; + + let read = reader.read(&mut first_read).await.expect("first read should succeed"); + + assert_eq!(read, first_read.len()); + assert_eq!(first_read[0], data[0]); + timeout(Duration::from_secs(1), async { + while read_count.load(Ordering::SeqCst) < 2 { + yield_now().await; + } + }) + .await + .expect("reader should start reading the next stripe before the current output buffer is fully consumed"); + } +} diff --git a/crates/ecstore/src/erasure_coding/mod.rs b/crates/ecstore/src/erasure_coding/mod.rs index 56d4a5f0d..21d283c42 100644 --- a/crates/ecstore/src/erasure_coding/mod.rs +++ b/crates/ecstore/src/erasure_coding/mod.rs @@ -14,6 +14,7 @@ mod bitrot; pub mod decode; +pub mod decode_reader; pub mod encode; pub mod erasure; pub mod heal; diff --git a/crates/ecstore/src/get_diagnostics.rs b/crates/ecstore/src/get_diagnostics.rs new file mode 100644 index 000000000..0c375d869 --- /dev/null +++ b/crates/ecstore/src/get_diagnostics.rs @@ -0,0 +1,213 @@ +// 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 crate::disk::error::DiskError; +use crate::error::StorageError; +use std::io; + +pub(crate) const GET_OBJECT_PATH_CODEC_STREAMING: &str = "codec_streaming"; +pub(crate) const GET_OBJECT_PATH_EMPTY: &str = "empty"; +pub(crate) const GET_OBJECT_PATH_LEGACY_DUPLEX: &str = "legacy_duplex"; +pub(crate) const GET_OBJECT_PATH_REMOTE_TRANSITION: &str = "remote_transition"; + +pub(crate) const GET_STAGE_DECODE: &str = "decode"; +pub(crate) const GET_STAGE_EMIT: &str = "emit"; +pub(crate) const GET_STAGE_FILL: &str = "fill"; +pub(crate) const GET_STAGE_METADATA: &str = "metadata"; +pub(crate) const GET_STAGE_OUTPUT_LOCK_WAIT: &str = "output_lock_wait"; +pub(crate) const GET_STAGE_OUTPUT_POLL: &str = "output_poll"; +pub(crate) const GET_STAGE_RANGE: &str = "range"; +pub(crate) const GET_STAGE_READER_SETUP: &str = "reader_setup"; +pub(crate) const GET_STAGE_RECONSTRUCT: &str = "reconstruct"; +pub(crate) const GET_STAGE_STRIPE_READ: &str = "stripe_read"; +pub(crate) const GET_STAGE_STRIPE_READ_FIRST_SHARD: &str = "stripe_read_first_shard"; +pub(crate) const GET_STAGE_STRIPE_READ_QUORUM: &str = "stripe_read_quorum"; + +pub(crate) const GET_READER_BUFFER_OUTPUT: &str = "output"; +pub(crate) const GET_READER_BUFFER_PREFETCH: &str = "prefetch"; +pub(crate) const GET_READER_PREFETCH_DIRECT: &str = "direct"; +pub(crate) const GET_READER_PREFETCH_EOF: &str = "eof"; +pub(crate) const GET_READER_PREFETCH_ERROR_DEFERRED: &str = "error_deferred"; +pub(crate) const GET_READER_PREFETCH_ERROR_IMMEDIATE: &str = "error_immediate"; +pub(crate) const GET_READER_PREFETCH_STORED: &str = "stored"; +pub(crate) const GET_READER_POLL_PENDING: &str = "pending"; +pub(crate) const GET_READER_POLL_READY_DATA: &str = "ready_data"; +pub(crate) const GET_READER_POLL_READY_EMPTY: &str = "ready_empty"; +pub(crate) const GET_READER_POLL_READY_ERROR: &str = "ready_error"; +pub(crate) const GET_SHARD_READ_OUTCOME_ERROR: &str = "error"; +pub(crate) const GET_SHARD_READ_OUTCOME_MISSING: &str = "missing"; +pub(crate) const GET_SHARD_READ_OUTCOME_SUCCESS: &str = "success"; +pub(crate) const GET_SHARD_ROLE_DATA: &str = "data"; +pub(crate) const GET_SHARD_ROLE_PARITY: &str = "parity"; + +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub(crate) enum GetObjectFailureReason { + BitrotMismatch, + DecodeError, + DownstreamClosed, + Io, + RangeOrLengthInvalid, + ReadQuorum, + ShortRead, + Timeout, + Unknown, +} + +impl GetObjectFailureReason { + pub(crate) const fn as_str(self) -> &'static str { + match self { + Self::BitrotMismatch => "bitrot_mismatch", + Self::DecodeError => "decode_error", + Self::DownstreamClosed => "downstream_closed", + Self::Io => "io", + Self::RangeOrLengthInvalid => "range_or_length_invalid", + Self::ReadQuorum => "read_quorum", + Self::ShortRead => "short_read", + Self::Timeout => "timeout", + Self::Unknown => "unknown", + } + } +} + +pub(crate) fn classify_storage_error(err: &StorageError) -> GetObjectFailureReason { + match err { + StorageError::ErasureReadQuorum | StorageError::InsufficientReadQuorum(_, _) => GetObjectFailureReason::ReadQuorum, + StorageError::FileCorrupt => GetObjectFailureReason::BitrotMismatch, + StorageError::InvalidRangeSpec(_) => GetObjectFailureReason::RangeOrLengthInvalid, + StorageError::Io(io_err) => classify_io_error(io_err), + _ => GetObjectFailureReason::Unknown, + } +} + +pub(crate) fn classify_disk_error(err: &DiskError) -> GetObjectFailureReason { + match err { + DiskError::ErasureReadQuorum => GetObjectFailureReason::ReadQuorum, + DiskError::FileCorrupt | DiskError::PartMissingOrCorrupt => GetObjectFailureReason::BitrotMismatch, + DiskError::LessData => GetObjectFailureReason::ShortRead, + DiskError::Timeout => GetObjectFailureReason::Timeout, + DiskError::Io(io_err) => classify_io_error(io_err), + _ => GetObjectFailureReason::Unknown, + } +} + +pub(crate) fn classify_io_error(err: &io::Error) -> GetObjectFailureReason { + match err.kind() { + io::ErrorKind::BrokenPipe | io::ErrorKind::ConnectionReset => GetObjectFailureReason::DownstreamClosed, + io::ErrorKind::TimedOut => GetObjectFailureReason::Timeout, + io::ErrorKind::UnexpectedEof => GetObjectFailureReason::ShortRead, + io::ErrorKind::InvalidInput | io::ErrorKind::InvalidData => GetObjectFailureReason::RangeOrLengthInvalid, + _ => GetObjectFailureReason::Io, + } +} + +pub(crate) fn record_get_object_pipeline_failure(stage: &'static str, reason: GetObjectFailureReason) { + rustfs_io_metrics::record_get_object_pipeline_failure(stage, reason.as_str()); +} + +pub(crate) fn record_get_object_pipeline_failure_for_path( + path: &'static str, + stage: &'static str, + reason: GetObjectFailureReason, +) { + rustfs_io_metrics::record_get_object_pipeline_failure_for_path(path, stage, reason.as_str()); +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn classifies_storage_errors_for_get_pipeline() { + assert_eq!( + classify_storage_error(&StorageError::ErasureReadQuorum), + GetObjectFailureReason::ReadQuorum + ); + assert_eq!( + classify_storage_error(&StorageError::InsufficientReadQuorum("bucket".to_string(), "object".to_string(),)), + GetObjectFailureReason::ReadQuorum + ); + assert_eq!(classify_storage_error(&StorageError::FileCorrupt), GetObjectFailureReason::BitrotMismatch); + assert_eq!( + classify_storage_error(&StorageError::InvalidRangeSpec("bad range".to_string())), + GetObjectFailureReason::RangeOrLengthInvalid + ); + } + + #[test] + fn classifies_disk_errors_for_get_pipeline() { + assert_eq!(classify_disk_error(&DiskError::ErasureReadQuorum), GetObjectFailureReason::ReadQuorum); + assert_eq!(classify_disk_error(&DiskError::FileCorrupt), GetObjectFailureReason::BitrotMismatch); + assert_eq!( + classify_disk_error(&DiskError::PartMissingOrCorrupt), + GetObjectFailureReason::BitrotMismatch + ); + assert_eq!(classify_disk_error(&DiskError::LessData), GetObjectFailureReason::ShortRead); + assert_eq!(classify_disk_error(&DiskError::Timeout), GetObjectFailureReason::Timeout); + } + + #[test] + fn classifies_io_errors_for_get_pipeline() { + let cases = [ + (io::ErrorKind::BrokenPipe, GetObjectFailureReason::DownstreamClosed), + (io::ErrorKind::ConnectionReset, GetObjectFailureReason::DownstreamClosed), + (io::ErrorKind::TimedOut, GetObjectFailureReason::Timeout), + (io::ErrorKind::UnexpectedEof, GetObjectFailureReason::ShortRead), + (io::ErrorKind::InvalidInput, GetObjectFailureReason::RangeOrLengthInvalid), + (io::ErrorKind::InvalidData, GetObjectFailureReason::RangeOrLengthInvalid), + (io::ErrorKind::Other, GetObjectFailureReason::Io), + ]; + + for (kind, expected) in cases { + let err = io::Error::from(kind); + assert_eq!(classify_io_error(&err), expected, "kind={kind:?}"); + } + } + + #[test] + fn keeps_metric_labels_stable() { + assert_eq!(GetObjectFailureReason::ReadQuorum.as_str(), "read_quorum"); + assert_eq!(GetObjectFailureReason::ShortRead.as_str(), "short_read"); + assert_eq!(GetObjectFailureReason::DownstreamClosed.as_str(), "downstream_closed"); + assert_eq!(GetObjectFailureReason::BitrotMismatch.as_str(), "bitrot_mismatch"); + assert_eq!(GetObjectFailureReason::DecodeError.as_str(), "decode_error"); + assert_eq!(GET_READER_BUFFER_OUTPUT, "output"); + assert_eq!(GET_READER_BUFFER_PREFETCH, "prefetch"); + assert_eq!(GET_READER_PREFETCH_DIRECT, "direct"); + assert_eq!(GET_READER_PREFETCH_STORED, "stored"); + assert_eq!(GET_READER_PREFETCH_EOF, "eof"); + assert_eq!(GET_READER_PREFETCH_ERROR_DEFERRED, "error_deferred"); + assert_eq!(GET_READER_PREFETCH_ERROR_IMMEDIATE, "error_immediate"); + assert_eq!(GET_READER_POLL_PENDING, "pending"); + assert_eq!(GET_READER_POLL_READY_DATA, "ready_data"); + assert_eq!(GET_READER_POLL_READY_EMPTY, "ready_empty"); + assert_eq!(GET_READER_POLL_READY_ERROR, "ready_error"); + assert_eq!(GET_STAGE_DECODE, "decode"); + assert_eq!(GET_STAGE_EMIT, "emit"); + assert_eq!(GET_STAGE_FILL, "fill"); + assert_eq!(GET_STAGE_METADATA, "metadata"); + assert_eq!(GET_STAGE_OUTPUT_LOCK_WAIT, "output_lock_wait"); + assert_eq!(GET_STAGE_OUTPUT_POLL, "output_poll"); + assert_eq!(GET_STAGE_RANGE, "range"); + assert_eq!(GET_STAGE_READER_SETUP, "reader_setup"); + assert_eq!(GET_STAGE_RECONSTRUCT, "reconstruct"); + assert_eq!(GET_STAGE_STRIPE_READ, "stripe_read"); + assert_eq!(GET_STAGE_STRIPE_READ_FIRST_SHARD, "stripe_read_first_shard"); + assert_eq!(GET_STAGE_STRIPE_READ_QUORUM, "stripe_read_quorum"); + assert_eq!(GET_SHARD_READ_OUTCOME_ERROR, "error"); + assert_eq!(GET_SHARD_READ_OUTCOME_MISSING, "missing"); + assert_eq!(GET_SHARD_READ_OUTCOME_SUCCESS, "success"); + assert_eq!(GET_SHARD_ROLE_DATA, "data"); + assert_eq!(GET_SHARD_ROLE_PARITY, "parity"); + } +} diff --git a/crates/ecstore/src/lib.rs b/crates/ecstore/src/lib.rs index b26764a56..051fc0ddf 100644 --- a/crates/ecstore/src/lib.rs +++ b/crates/ecstore/src/lib.rs @@ -29,8 +29,10 @@ mod data_usage; mod disk; mod disks_layout; mod endpoints; +mod erasure_codec; mod erasure_coding; mod error; +mod get_diagnostics; mod global; pub(crate) mod layout; mod metrics_realtime; diff --git a/crates/ecstore/src/set_disk.rs b/crates/ecstore/src/set_disk.rs index 1756ab3f9..3305ec513 100644 --- a/crates/ecstore/src/set_disk.rs +++ b/crates/ecstore/src/set_disk.rs @@ -35,6 +35,10 @@ use crate::disk::{STORAGE_FORMAT_FILE, count_part_not_success}; use crate::erasure_coding; use crate::error::{Error, Result, is_err_version_not_found}; use crate::error::{GenericError, ObjectApiError, is_err_object_not_found}; +use crate::get_diagnostics::{ + GET_OBJECT_PATH_CODEC_STREAMING, GET_OBJECT_PATH_EMPTY, GET_OBJECT_PATH_LEGACY_DUPLEX, GET_OBJECT_PATH_REMOTE_TRANSITION, + GET_STAGE_EMIT, GET_STAGE_METADATA, classify_storage_error, record_get_object_pipeline_failure, +}; use crate::object_api::ObjectOptions; use crate::rpc::heal_bucket_local_on_disks; use crate::runtime_sources; @@ -282,6 +286,12 @@ pub fn get_duplex_buffer_size() -> usize { } const DISK_ONLINE_TIMEOUT: Duration = Duration::from_secs(1); const DISK_HEALTH_CACHE_TTL: Duration = Duration::from_millis(750); +const GET_OBJECT_METADATA_CACHE_TTL: Duration = Duration::from_millis(250); +const GET_OBJECT_METADATA_CACHE_MAX_ENTRIES: usize = 1024; +const ENV_RUSTFS_GET_CODEC_STREAMING_ENABLE: &str = "RUSTFS_GET_CODEC_STREAMING_ENABLE"; +const ENV_RUSTFS_GET_CODEC_STREAMING_MIN_SIZE: &str = "RUSTFS_GET_CODEC_STREAMING_MIN_SIZE"; +const DEFAULT_RUSTFS_GET_CODEC_STREAMING_ENABLE: bool = false; +const DEFAULT_RUSTFS_GET_CODEC_STREAMING_MIN_SIZE: usize = MI_B; static OBJECT_LOCK_DIAG_ENABLED: OnceLock = OnceLock::new(); mod heal; @@ -291,6 +301,7 @@ mod metadata; mod multipart; mod read; mod replication; +pub(crate) mod shard_source; mod write; /// Get lock acquire timeout from environment variable RUSTFS_LOCK_ACQUIRE_TIMEOUT (in seconds) @@ -346,6 +357,87 @@ pub fn is_deadlock_detection_enabled() -> bool { ) } +fn is_get_codec_streaming_enabled() -> bool { + rustfs_utils::get_env_bool(ENV_RUSTFS_GET_CODEC_STREAMING_ENABLE, DEFAULT_RUSTFS_GET_CODEC_STREAMING_ENABLE) +} + +fn get_codec_streaming_min_size() -> usize { + rustfs_utils::get_env_usize(ENV_RUSTFS_GET_CODEC_STREAMING_MIN_SIZE, DEFAULT_RUSTFS_GET_CODEC_STREAMING_MIN_SIZE) +} + +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +enum GetCodecStreamingDecision { + Use, + Fallback(GetCodecStreamingFallbackReason), +} + +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +enum GetCodecStreamingFallbackReason { + Disabled, + LockOptimizationDisabled, + Range, + BelowMinSize, + Encrypted, + Compressed, + Remote, + Multipart, + InvalidMinSize, +} + +impl GetCodecStreamingFallbackReason { + const fn as_str(self) -> &'static str { + match self { + Self::Disabled => "disabled", + Self::LockOptimizationDisabled => "lock_optimization_disabled", + Self::Range => "range", + Self::BelowMinSize => "below_min_size", + Self::Encrypted => "encrypted", + Self::Compressed => "compressed", + Self::Remote => "remote", + Self::Multipart => "multipart", + Self::InvalidMinSize => "invalid_min_size", + } + } +} + +fn get_codec_streaming_reader_decision( + range: &Option, + object_info: &ObjectInfo, + fi: &FileInfo, + lock_optimization_enabled: bool, +) -> GetCodecStreamingDecision { + if !is_get_codec_streaming_enabled() { + return GetCodecStreamingDecision::Fallback(GetCodecStreamingFallbackReason::Disabled); + } + if !lock_optimization_enabled { + return GetCodecStreamingDecision::Fallback(GetCodecStreamingFallbackReason::LockOptimizationDisabled); + } + if range.is_some() { + return GetCodecStreamingDecision::Fallback(GetCodecStreamingFallbackReason::Range); + } + + let Ok(min_size) = i64::try_from(get_codec_streaming_min_size()) else { + return GetCodecStreamingDecision::Fallback(GetCodecStreamingFallbackReason::InvalidMinSize); + }; + if object_info.size < min_size { + return GetCodecStreamingDecision::Fallback(GetCodecStreamingFallbackReason::BelowMinSize); + } + if object_info.is_encrypted() { + return GetCodecStreamingDecision::Fallback(GetCodecStreamingFallbackReason::Encrypted); + } + if object_info.is_compressed() { + return GetCodecStreamingDecision::Fallback(GetCodecStreamingFallbackReason::Compressed); + } + if object_info.is_remote() { + return GetCodecStreamingDecision::Fallback(GetCodecStreamingFallbackReason::Remote); + } + if fi.parts.len() != 1 { + return GetCodecStreamingDecision::Fallback(GetCodecStreamingFallbackReason::Multipart); + } + + GetCodecStreamingDecision::Use +} + /// Record a lock acquisition for deadlock detection. /// This records detailed lock information for deadlock analysis. /// Returns the lock_id for later release tracking. @@ -438,10 +530,41 @@ pub struct SetDisks { pub pool_index: usize, pub format: FormatV3, disk_health_cache: Arc>>>, + get_object_metadata_cache: Arc>>, pub lockers: Vec>, local_lock_manager: Arc, } +#[derive(Clone, Debug, Eq, Hash, PartialEq)] +struct GetObjectMetadataCacheKey { + bucket: String, + object: String, +} + +impl GetObjectMetadataCacheKey { + fn new(bucket: &str, object: &str) -> Self { + Self { + bucket: bucket.to_string(), + object: object.to_string(), + } + } +} + +#[derive(Clone, Debug)] +struct GetObjectMetadataCacheEntry { + created_at: Instant, + fi: FileInfo, + parts_metadata: Vec, + online_disks: Vec>, + read_quorum: usize, +} + +impl GetObjectMetadataCacheEntry { + fn is_fresh(&self) -> bool { + self.created_at.elapsed() <= GET_OBJECT_METADATA_CACHE_TTL + } +} + #[derive(Clone, Debug)] struct DiskHealthEntry { last_check: Instant, @@ -459,6 +582,13 @@ impl DiskHealthEntry { } impl SetDisks { + async fn invalidate_get_object_metadata_cache(&self, bucket: &str, object: &str) { + self.get_object_metadata_cache + .write() + .await + .remove(&GetObjectMetadataCacheKey::new(bucket, object)); + } + async fn acquire_read_lock_diag(&self, op: &'static str, bucket: &str, object: &str) -> Result { let diag_enabled = is_object_lock_diag_enabled(); let ns_lock = self.new_ns_lock(bucket, object).await?; @@ -564,6 +694,7 @@ impl SetDisks { format, set_endpoints, disk_health_cache: Arc::new(RwLock::new(Vec::new())), + get_object_metadata_cache: Arc::new(RwLock::new(HashMap::new())), lockers, local_lock_manager: runtime_sources::global_lock_manager(), }) @@ -942,10 +1073,18 @@ impl rustfs_storage_api::ObjectIO for SetDisks { 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 metadata_stage_start = Instant::now(); + let (fi, files, disks) = match self.get_object_fileinfo(bucket, object, opts, true).await { + Ok(result) => { + rustfs_io_metrics::record_get_object_metadata_phase_duration(metadata_stage_start.elapsed().as_secs_f64()); + result + } + Err(err) => { + rustfs_io_metrics::record_get_object_metadata_phase_duration(metadata_stage_start.elapsed().as_secs_f64()); + record_get_object_pipeline_failure(GET_STAGE_METADATA, classify_storage_error(&err)); + return 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 { @@ -965,6 +1104,7 @@ impl rustfs_storage_api::ObjectIO for SetDisks { // } if object_info.size == 0 { + rustfs_io_metrics::record_get_object_reader_path(GET_OBJECT_PATH_EMPTY); // if let Some(rs) = range { // let _ = rs.get_offset_length(object_info.size)?; // } @@ -976,7 +1116,13 @@ impl rustfs_storage_api::ObjectIO for SetDisks { return Ok(reader); } + let codec_streaming_decision = get_codec_streaming_reader_decision(&range, &object_info, &fi, lock_optimization_enabled); + if object_info.is_remote() { + if let GetCodecStreamingDecision::Fallback(reason) = codec_streaming_decision { + rustfs_io_metrics::record_get_object_codec_streaming_fallback(reason.as_str()); + } + rustfs_io_metrics::record_get_object_reader_path(GET_OBJECT_PATH_REMOTE_TRANSITION); let mut opts = opts.clone(); if object_info.parts.len() == 1 { opts.part_number = Some(1); @@ -1004,6 +1150,30 @@ impl rustfs_storage_api::ObjectIO for SetDisks { read_lock_guard }; + match codec_streaming_decision { + GetCodecStreamingDecision::Use => { + rustfs_io_metrics::record_get_object_reader_path(GET_OBJECT_PATH_CODEC_STREAMING); + let stream = Self::get_object_decode_reader_with_fileinfo( + bucket, + object, + fi, + files, + &disks, + self.set_index, + self.pool_index, + opts.skip_verify_bitrot, + ) + .await?; + let (reader, _offset, _length) = GetObjectReader::new(stream, range, &object_info, opts, &h).await?; + return Ok(reader); + } + GetCodecStreamingDecision::Fallback(reason) => { + rustfs_io_metrics::record_get_object_codec_streaming_fallback(reason.as_str()); + } + } + + rustfs_io_metrics::record_get_object_reader_path(GET_OBJECT_PATH_LEGACY_DUPLEX); + let duplex_buffer_size = get_duplex_buffer_size(); let (rd, wd) = tokio::io::duplex(duplex_buffer_size); debug!(bucket, object, duplex_buffer_size, "Created duplex pipe for object data transfer"); @@ -1041,13 +1211,22 @@ impl rustfs_storage_api::ObjectIO for SetDisks { ) .await { + let reason = classify_storage_error(&e); + record_get_object_pipeline_failure(GET_STAGE_EMIT, reason); error!( event = EVENT_SET_DISK_WRITE, component = LOG_COMPONENT_ECSTORE, subsystem = LOG_SUBSYSTEM_SET_DISK, bucket, object, + pool_index, + set_index, + offset, + requested_length = length, + skip_verify_bitrot = skip_verify, state = "read_pipeline_failed", + stage = GET_STAGE_EMIT, + reason = reason.as_str(), error = ?e, "Set disk object read pipeline failed" ); @@ -1059,6 +1238,8 @@ impl rustfs_storage_api::ObjectIO for SetDisks { #[tracing::instrument(skip(self, data,))] async fn put_object(&self, bucket: &str, object: &str, data: &mut PutObjReader, opts: &ObjectOptions) -> Result { + self.invalidate_get_object_metadata_cache(bucket, object).await; + let disks = self.get_disks_internal().await; let mut object_lock_guard = None; @@ -1501,6 +1682,10 @@ impl rustfs_storage_api::ObjectIO for SetDisks { ); } + if result.is_ok() { + self.invalidate_get_object_metadata_cache(bucket, object).await; + } + if issue3031_diag_enabled() { warn!( target: "rustfs_ecstore::set_disk", @@ -2127,6 +2312,8 @@ impl rustfs_storage_api::ObjectOperations for SetDisks { ) }; + self.invalidate_get_object_metadata_cache(dst_bucket, dst_object).await; + if dst_opts.http_preconditions.is_some() && let Some(err) = self.check_write_precondition(dst_bucket, dst_object, dst_opts).await { @@ -2239,6 +2426,8 @@ impl rustfs_storage_api::ObjectOperations for SetDisks { .map_err(|e| to_object_err(e.into(), vec![src_bucket, src_object]))?; } + self.invalidate_get_object_metadata_cache(src_bucket, src_object).await; + Ok(ObjectInfo::from_file_info( &fi, src_bucket, @@ -2292,6 +2481,10 @@ impl rustfs_storage_api::ObjectOperations for SetDisks { objects: Vec, opts: ObjectOptions, ) -> (Vec, Vec>) { + for object in &objects { + self.invalidate_get_object_metadata_cache(bucket, &object.object_name).await; + } + // Default return value let mut del_objects = vec![DeletedObject::default(); objects.len()]; @@ -2525,11 +2718,19 @@ impl rustfs_storage_api::ObjectOperations for SetDisks { self.release_dist_delete_object_locks_batch(dist_batch_lock_ids).await; } + for (object, err) in objects.iter().zip(del_errs.iter()) { + if err.is_none() { + self.invalidate_get_object_metadata_cache(bucket, &object.object_name).await; + } + } + (del_objects, del_errs) } #[tracing::instrument(skip(self))] async fn delete_object(&self, bucket: &str, object: &str, mut opts: ObjectOptions) -> Result { + self.invalidate_get_object_metadata_cache(bucket, object).await; + // Guard lock for single object delete let _lock_guard = if (!opts.delete_prefix || opts.delete_prefix_object) && !opts.no_lock { Some(self.acquire_write_lock_diag("delete_object", bucket, object).await?) @@ -2541,6 +2742,7 @@ impl rustfs_storage_api::ObjectOperations for SetDisks { .await .map_err(|e| to_object_err(e.into(), vec![bucket, object]))?; + self.get_object_metadata_cache.write().await.clear(); return Ok(ObjectInfo::default()); } @@ -2629,6 +2831,7 @@ impl rustfs_storage_api::ObjectOperations for SetDisks { let mut oi = ObjectInfo::from_file_info(&fi, bucket, object, opts.versioned || opts.version_suspended); oi.replication_decision = goi.replication_decision; + self.invalidate_get_object_metadata_cache(bucket, object).await; return Ok(oi); } @@ -2658,6 +2861,7 @@ impl rustfs_storage_api::ObjectOperations for SetDisks { let mut obj_info = ObjectInfo::from_file_info(&dfi, bucket, object, opts.versioned || opts.version_suspended); obj_info.size = goi.size; + self.invalidate_get_object_metadata_cache(bucket, object).await; Ok(obj_info) } @@ -2709,6 +2913,8 @@ impl rustfs_storage_api::ObjectOperations for SetDisks { #[tracing::instrument(skip(self))] async fn put_object_metadata(&self, bucket: &str, object: &str, opts: &ObjectOptions) -> Result { + self.invalidate_get_object_metadata_cache(bucket, object).await; + // TODO: nslock // Guard lock for metadata update @@ -2784,6 +2990,8 @@ impl rustfs_storage_api::ObjectOperations for SetDisks { .await .map_err(|e| to_object_err(e.into(), vec![bucket, object]))?; + self.invalidate_get_object_metadata_cache(bucket, object).await; + Ok(ObjectInfo::from_file_info(&fi, bucket, object, opts.versioned || opts.version_suspended)) } @@ -4030,6 +4238,8 @@ impl rustfs_storage_api::MultipartOperations for SetDisks { uploaded_parts: Vec, opts: &ObjectOptions, ) -> Result { + self.invalidate_get_object_metadata_cache(bucket, object).await; + let mut object_lock_guard = None; if opts.http_preconditions.is_some() { @@ -4449,6 +4659,8 @@ impl rustfs_storage_api::MultipartOperations for SetDisks { fi.is_latest = true; + self.invalidate_get_object_metadata_cache(bucket, object).await; + Ok(ObjectInfo::from_file_info(&fi, bucket, object, opts.versioned || opts.version_suspended)) } } diff --git a/crates/ecstore/src/set_disk/read.rs b/crates/ecstore/src/set_disk/read.rs index e8fbe4bc7..a5e35fdf9 100644 --- a/crates/ecstore/src/set_disk/read.rs +++ b/crates/ecstore/src/set_disk/read.rs @@ -13,8 +13,13 @@ // limitations under the License. use super::*; +use crate::get_diagnostics::{ + GET_OBJECT_PATH_CODEC_STREAMING, GET_STAGE_DECODE, GET_STAGE_RANGE, GET_STAGE_READER_SETUP, GetObjectFailureReason, + classify_disk_error, record_get_object_pipeline_failure, record_get_object_pipeline_failure_for_path, +}; use rustfs_config::{DEFAULT_OBJECT_ZERO_COPY_ENABLE, ENV_OBJECT_ZERO_COPY_ENABLE}; use std::future::Future; +use tokio::io::AsyncRead; use tokio::task::JoinSet; async fn collect_read_multiple_results( @@ -98,6 +103,55 @@ where } impl SetDisks { + async fn is_get_object_metadata_cache_enabled(&self, bucket: &str, opts: &ObjectOptions, read_data: bool) -> bool { + is_get_object_metadata_cache_request_eligible(bucket, opts, read_data) && !runtime_sources::setup_is_dist_erasure().await + } + + async fn cached_get_object_fileinfo(&self, bucket: &str, object: &str) -> Option { + let key = GetObjectMetadataCacheKey::new(bucket, object); + let cache = self.get_object_metadata_cache.read().await; + cache + .get(&key) + .filter(|entry| { + entry.is_fresh() && entry.online_disks.iter().filter(|disk| disk.is_some()).count() >= entry.read_quorum + }) + .cloned() + } + + async fn cache_get_object_fileinfo( + &self, + bucket: &str, + object: &str, + fi: &FileInfo, + parts_metadata: &[FileInfo], + online_disks: &[Option], + read_quorum: usize, + ) { + if fi.deleted || !fi.is_valid() { + return; + } + + let key = GetObjectMetadataCacheKey::new(bucket, object); + let mut cache = self.get_object_metadata_cache.write().await; + if cache.len() >= GET_OBJECT_METADATA_CACHE_MAX_ENTRIES { + cache.retain(|_, entry| entry.is_fresh()); + if cache.len() >= GET_OBJECT_METADATA_CACHE_MAX_ENTRIES { + cache.clear(); + } + } + + cache.insert( + key, + GetObjectMetadataCacheEntry { + created_at: Instant::now(), + fi: fi.clone(), + parts_metadata: parts_metadata.to_vec(), + online_disks: online_disks.to_vec(), + read_quorum, + }, + ); + } + pub(super) async fn read_parts( disks: &[Option], bucket: &str, @@ -617,6 +671,11 @@ impl SetDisks { opts: &ObjectOptions, read_data: bool, ) -> Result<(FileInfo, Vec, Vec>)> { + let use_metadata_cache = self.is_get_object_metadata_cache_enabled(bucket, opts, read_data).await; + if use_metadata_cache && let Some(cached) = self.cached_get_object_fileinfo(bucket, object).await { + return Ok((cached.fi, cached.parts_metadata, cached.online_disks)); + } + let disks = self.disks.read().await; let disks = disks.clone(); @@ -660,6 +719,9 @@ impl SetDisks { Some(self.set_index), // set_index )) .await; + } else if use_metadata_cache { + self.cache_get_object_fileinfo(bucket, object, &fi, &parts_metadata, &op_online_disks, read_quorum as usize) + .await; } // debug!("get_object_fileinfo pick fi {:?}", &fi); @@ -733,19 +795,56 @@ impl SetDisks { let total_size = fi.size as usize; if offset > total_size { - error!("get_object_with_fileinfo offset out of range: {}, total_size: {}", offset, total_size); + let reason = GetObjectFailureReason::RangeOrLengthInvalid; + record_get_object_pipeline_failure(GET_STAGE_RANGE, reason); + error!( + bucket, + object, + offset, + total_size, + requested_length = length, + stage = GET_STAGE_RANGE, + reason = reason.as_str(), + state = "range_or_length_invalid", + "GetObject range validation failed" + ); return Err(Error::other("offset out of range")); } let length = if length < 0 { total_size - offset } else { length as usize }; let Some(end_offset_exclusive) = offset.checked_add(length) else { - error!("get_object_with_fileinfo offset overflow: {}, length: {}", offset, length); + let reason = GetObjectFailureReason::RangeOrLengthInvalid; + record_get_object_pipeline_failure(GET_STAGE_RANGE, reason); + error!( + bucket, + object, + offset, + total_size, + requested_length = length, + stage = GET_STAGE_RANGE, + reason = reason.as_str(), + state = "range_or_length_invalid", + "GetObject range validation overflow" + ); return Err(Error::other("offset out of range")); }; if end_offset_exclusive > total_size { - error!("get_object_with_fileinfo offset out of range: {}, total_size: {}", offset, total_size); + let reason = GetObjectFailureReason::RangeOrLengthInvalid; + record_get_object_pipeline_failure(GET_STAGE_RANGE, reason); + error!( + bucket, + object, + offset, + total_size, + requested_length = length, + end_offset_exclusive, + stage = GET_STAGE_RANGE, + reason = reason.as_str(), + state = "range_or_length_invalid", + "GetObject range validation failed" + ); return Err(Error::other("offset out of range")); } @@ -825,6 +924,7 @@ impl SetDisks { // Default: enabled (true) for performance let use_zero_copy = rustfs_utils::get_env_bool(ENV_OBJECT_ZERO_COPY_ENABLE, DEFAULT_OBJECT_ZERO_COPY_ENABLE); + let reader_setup_stage_start = Instant::now(); let mut readers = Vec::with_capacity(disks.len()); let mut errors = Vec::with_capacity(disks.len()); for (idx, disk_op) in disks.iter().enumerate() { @@ -856,14 +956,51 @@ impl SetDisks { } } } + rustfs_io_metrics::record_get_object_shard_reader_setup_duration(reader_setup_stage_start.elapsed().as_secs_f64()); let nil_count = errors.iter().filter(|&e| e.is_none()).count(); if nil_count < erasure.data_shards { if let Some(read_err) = reduce_read_quorum_errs(&errors, OBJECT_OP_IGNORED_ERRS, erasure.data_shards) { - error!("create_bitrot_reader reduce_read_quorum_errs {:?}", &errors); + let reason = classify_disk_error(&read_err); + error!( + bucket, + object, + part_index = current_part, + part_number, + part_offset, + part_length, + read_offset, + till_offset, + total_shards = erasure.data_shards + erasure.parity_shards, + available_shards = nil_count, + data_shards = erasure.data_shards, + stage = GET_STAGE_READER_SETUP, + reason = reason.as_str(), + errors = ?errors, + "Create bitrot reader failed read quorum" + ); + record_get_object_pipeline_failure(GET_STAGE_READER_SETUP, reason); return Err(to_object_err(read_err.into(), vec![bucket, object])); } - error!("create_bitrot_reader not enough disks to read: {:?}", &errors); + let reason = GetObjectFailureReason::ReadQuorum; + error!( + bucket, + object, + part_index = current_part, + part_number, + part_offset, + part_length, + read_offset, + till_offset, + total_shards = erasure.data_shards + erasure.parity_shards, + available_shards = nil_count, + data_shards = erasure.data_shards, + stage = GET_STAGE_READER_SETUP, + reason = reason.as_str(), + errors = ?errors, + "Create bitrot reader did not have enough disks" + ); + record_get_object_pipeline_failure(GET_STAGE_READER_SETUP, reason); return Err(Error::other(format!("not enough disks to read: {errors:?}"))); } @@ -924,7 +1061,9 @@ impl SetDisks { // "read part {} part_offset {},part_length {},part_size {} ", // part_number, part_offset, part_length, part_size // ); + let decode_stage_start = Instant::now(); let (written, err) = erasure.decode(writer, readers, part_offset, part_length, part_size).await; + rustfs_io_metrics::record_get_object_decode_duration(decode_stage_start.elapsed().as_secs_f64()); debug!( bucket, object, @@ -968,7 +1107,21 @@ impl SetDisks { } if has_err { - error!("erasure.decode err {} {:?}", written, &de_err); + let reason = classify_disk_error(&de_err); + error!( + bucket, + object, + part_index = current_part, + part_number, + part_offset, + part_length, + bytes_written = written, + stage = GET_STAGE_DECODE, + reason = reason.as_str(), + error = ?de_err, + "Erasure decode failed during GetObject" + ); + record_get_object_pipeline_failure(GET_STAGE_DECODE, reason); return Err(de_err.into()); } } @@ -985,12 +1138,516 @@ impl SetDisks { Ok(()) } + + #[allow(clippy::too_many_arguments)] + pub(super) async fn get_object_decode_reader_with_fileinfo( + bucket: &str, + object: &str, + fi: FileInfo, + files: Vec, + disks: &[Option], + set_index: usize, + pool_index: usize, + skip_verify_bitrot: bool, + ) -> Result> { + let (disks, files) = Self::shuffle_disks_and_parts_metadata_by_index(disks, &files, &fi); + if fi.parts.len() != 1 { + return Err(Error::other("codec streaming reader only supports single-part plain objects")); + } + + 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 part = &fi.parts[0]; + let part_number = part.number; + let part_size = part.size; + let part_length = usize::try_from(fi.size).map_err(|_| Error::other("codec streaming reader object size is invalid"))?; + if part_length > part_size { + return Err(Error::other("codec streaming reader part length exceeds part size")); + } + + 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 + }; + let use_zero_copy = rustfs_utils::get_env_bool(ENV_OBJECT_ZERO_COPY_ENABLE, DEFAULT_OBJECT_ZERO_COPY_ENABLE); + let till_offset = erasure.shard_file_offset(0, part_length, part_size); + + let reader_setup_stage_start = Instant::now(); + 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), + 0, + till_offset, + 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(e) => { + readers.push(None); + errors.push(Some(e)); + } + } + } + rustfs_io_metrics::record_get_object_stage_duration( + GET_OBJECT_PATH_CODEC_STREAMING, + GET_STAGE_READER_SETUP, + reader_setup_stage_start.elapsed().as_secs_f64(), + ); + + let available_shards = errors.iter().filter(|err| err.is_none()).count(); + if available_shards < erasure.data_shards { + if let Some(read_err) = reduce_read_quorum_errs(&errors, OBJECT_OP_IGNORED_ERRS, erasure.data_shards) { + let reason = classify_disk_error(&read_err); + record_get_object_pipeline_failure_for_path(GET_OBJECT_PATH_CODEC_STREAMING, GET_STAGE_READER_SETUP, reason); + return Err(to_object_err(read_err.into(), vec![bucket, object])); + } + record_get_object_pipeline_failure_for_path( + GET_OBJECT_PATH_CODEC_STREAMING, + GET_STAGE_READER_SETUP, + GetObjectFailureReason::ReadQuorum, + ); + return Err(Error::other(format!("not enough disks to read: {errors:?}"))); + } + + let total_shards = erasure.data_shards + erasure.parity_shards; + let missing_shards = total_shards.saturating_sub(available_shards); + if missing_shards > 0 + && let Err(e) = + rustfs_common::heal_channel::send_heal_request(rustfs_common::heal_channel::create_heal_request_with_options( + bucket.to_string(), + Some(object.to_string()), + false, + Some(HealChannelPriority::Normal), + Some(pool_index), + Some(set_index), + )) + .await + { + warn!( + bucket, + object, + part_number, + error = %e, + "Failed to enqueue heal request for missing shards" + ); + } + + let source = erasure_coding::decode::ParallelReader::new_with_metrics_path( + readers, + erasure.clone(), + 0, + part_size, + Some(GET_OBJECT_PATH_CODEC_STREAMING), + ); + let engine = crate::erasure_codec::bridge::LegacyEcDecodeEngine::new(erasure); + let reader = erasure_coding::decode_reader::ErasureDecodeReader::new(source, engine, part_length)?; + Ok(Box::new(erasure_coding::decode_reader::SyncErasureDecodeReader::new(reader))) + } +} + +fn is_get_object_metadata_cache_request_eligible(bucket: &str, opts: &ObjectOptions, read_data: bool) -> bool { + read_data + && !opts.no_lock + && opts.version_id.is_none() + && !opts.versioned + && !opts.version_suspended + && !opts.incl_free_versions + && !opts.delete_marker + && opts.part_number.is_none() + && !opts.data_movement + && !opts.raw_data_movement_read + && !bucket.starts_with(RUSTFS_META_BUCKET) +} + +#[cfg(test)] +mod metadata_cache_tests { + use super::*; + + async fn new_metadata_cache_test_set() -> Arc { + SetDisks::new( + "metadata-cache-test".to_string(), + Arc::new(RwLock::new(Vec::new())), + 4, + 2, + 0, + 0, + Vec::new(), + FormatV3::new(1, 4), + Vec::new(), + ) + .await + } + + fn valid_test_fileinfo(object: &str) -> FileInfo { + let mut fi = FileInfo::new(object, 2, 2); + fi.volume = "bucket".to_string(); + fi.name = object.to_string(); + fi.size = 1; + fi.erasure.index = 1; + fi.metadata.insert("etag".to_string(), "etag-1".to_string()); + fi + } + + #[test] + fn get_object_metadata_cache_request_eligibility_is_conservative() { + let opts = ObjectOptions::default(); + assert!(is_get_object_metadata_cache_request_eligible("bucket", &opts, true)); + assert!(!is_get_object_metadata_cache_request_eligible("bucket", &opts, false)); + assert!(!is_get_object_metadata_cache_request_eligible(RUSTFS_META_BUCKET, &opts, true)); + + let mut opts = ObjectOptions { + no_lock: true, + ..Default::default() + }; + assert!(!is_get_object_metadata_cache_request_eligible("bucket", &opts, true)); + + opts = ObjectOptions { + version_id: Some("version".to_string()), + ..Default::default() + }; + assert!(!is_get_object_metadata_cache_request_eligible("bucket", &opts, true)); + + opts = ObjectOptions { + versioned: true, + ..Default::default() + }; + assert!(!is_get_object_metadata_cache_request_eligible("bucket", &opts, true)); + + opts = ObjectOptions { + version_suspended: true, + ..Default::default() + }; + assert!(!is_get_object_metadata_cache_request_eligible("bucket", &opts, true)); + + opts = ObjectOptions { + incl_free_versions: true, + ..Default::default() + }; + assert!(!is_get_object_metadata_cache_request_eligible("bucket", &opts, true)); + + opts = ObjectOptions { + delete_marker: true, + ..Default::default() + }; + assert!(!is_get_object_metadata_cache_request_eligible("bucket", &opts, true)); + + opts = ObjectOptions { + part_number: Some(1), + ..Default::default() + }; + assert!(!is_get_object_metadata_cache_request_eligible("bucket", &opts, true)); + + opts = ObjectOptions { + data_movement: true, + ..Default::default() + }; + assert!(!is_get_object_metadata_cache_request_eligible("bucket", &opts, true)); + + opts = ObjectOptions { + raw_data_movement_read: true, + ..Default::default() + }; + assert!(!is_get_object_metadata_cache_request_eligible("bucket", &opts, true)); + } + + #[tokio::test] + async fn get_object_metadata_cache_hit_returns_stored_metadata() { + let set = new_metadata_cache_test_set().await; + let fi = valid_test_fileinfo("object"); + let parts_metadata = vec![fi.clone()]; + let online_disks = Vec::new(); + + set.cache_get_object_fileinfo("bucket", "object", &fi, &parts_metadata, &online_disks, 0) + .await; + + let cached = set + .cached_get_object_fileinfo("bucket", "object") + .await + .expect("fresh cache entry should be returned"); + assert_eq!(cached.fi.name, "object"); + assert_eq!(cached.parts_metadata.len(), 1); + assert_eq!(cached.online_disks.len(), 0); + assert_eq!(cached.read_quorum, 0); + } + + #[tokio::test] + async fn get_object_metadata_cache_rejects_deleted_and_invalid_fileinfo() { + let set = new_metadata_cache_test_set().await; + + let mut deleted = valid_test_fileinfo("deleted-object"); + deleted.deleted = true; + set.cache_get_object_fileinfo("bucket", "deleted-object", &deleted, &[deleted.clone()], &[], 0) + .await; + assert!( + set.cached_get_object_fileinfo("bucket", "deleted-object").await.is_none(), + "deleted metadata must not be cached" + ); + + let invalid = FileInfo::default(); + set.cache_get_object_fileinfo("bucket", "invalid-object", &invalid, std::slice::from_ref(&invalid), &[], 0) + .await; + assert!( + set.cached_get_object_fileinfo("bucket", "invalid-object").await.is_none(), + "invalid metadata must not be cached" + ); + } + + #[tokio::test] + async fn get_object_metadata_cache_requires_cached_read_quorum() { + let set = new_metadata_cache_test_set().await; + let fi = valid_test_fileinfo("object"); + + set.get_object_metadata_cache.write().await.insert( + GetObjectMetadataCacheKey::new("bucket", "object"), + GetObjectMetadataCacheEntry { + created_at: Instant::now(), + fi: fi.clone(), + parts_metadata: vec![fi], + online_disks: vec![None], + read_quorum: 1, + }, + ); + + assert!( + set.cached_get_object_fileinfo("bucket", "object").await.is_none(), + "cache hit must be rejected when cached online disks cannot satisfy read quorum" + ); + } + + #[tokio::test] + async fn get_object_metadata_cache_rejects_stale_entries() { + let set = new_metadata_cache_test_set().await; + let fi = valid_test_fileinfo("object"); + + set.get_object_metadata_cache.write().await.insert( + GetObjectMetadataCacheKey::new("bucket", "object"), + GetObjectMetadataCacheEntry { + created_at: Instant::now() - GET_OBJECT_METADATA_CACHE_TTL - Duration::from_millis(1), + fi: fi.clone(), + parts_metadata: vec![fi], + online_disks: Vec::new(), + read_quorum: 0, + }, + ); + + assert!( + set.cached_get_object_fileinfo("bucket", "object").await.is_none(), + "stale cache entry must not be returned" + ); + } + + #[tokio::test] + async fn get_object_metadata_cache_invalidation_removes_object_entry() { + let set = new_metadata_cache_test_set().await; + let fi = valid_test_fileinfo("object"); + + set.cache_get_object_fileinfo("bucket", "object", &fi, std::slice::from_ref(&fi), &[], 0) + .await; + assert!(set.cached_get_object_fileinfo("bucket", "object").await.is_some()); + + set.invalidate_get_object_metadata_cache("bucket", "object").await; + assert!( + set.cached_get_object_fileinfo("bucket", "object").await.is_none(), + "explicit invalidation must remove the cached object metadata" + ); + } + + #[tokio::test] + async fn get_object_metadata_cache_prunes_when_capacity_is_reached() { + let set = new_metadata_cache_test_set().await; + let stale_fi = valid_test_fileinfo("stale-object"); + let fresh_fi = valid_test_fileinfo("fresh-object"); + { + let mut cache = set.get_object_metadata_cache.write().await; + for idx in 0..GET_OBJECT_METADATA_CACHE_MAX_ENTRIES { + cache.insert( + GetObjectMetadataCacheKey::new("bucket", &format!("stale-object-{idx}")), + GetObjectMetadataCacheEntry { + created_at: Instant::now() - GET_OBJECT_METADATA_CACHE_TTL - Duration::from_millis(1), + fi: stale_fi.clone(), + parts_metadata: vec![stale_fi.clone()], + online_disks: Vec::new(), + read_quorum: 0, + }, + ); + } + } + + set.cache_get_object_fileinfo("bucket", "fresh-object", &fresh_fi, std::slice::from_ref(&fresh_fi), &[], 0) + .await; + + let cache = set.get_object_metadata_cache.read().await; + assert_eq!(cache.len(), 1); + assert!(cache.contains_key(&GetObjectMetadataCacheKey::new("bucket", "fresh-object"))); + } } #[cfg(test)] mod tests { use super::*; + fn codec_streaming_test_fileinfo(size: i64, part_count: usize) -> FileInfo { + let mut fi = FileInfo::new("object", 4, 2); + fi.volume = "bucket".to_string(); + fi.name = "object".to_string(); + fi.size = size; + fi.metadata.insert("etag".to_string(), "etag-1".to_string()); + + let size = usize::try_from(size).expect("test object size should fit usize"); + let part_count = part_count.max(1); + let part_size = size.div_ceil(part_count); + for index in 0..part_count { + let remaining = size.saturating_sub(index * part_size); + let current_part_size = remaining.min(part_size); + fi.add_object_part( + index + 1, + format!("etag-{index}"), + current_part_size, + None, + i64::try_from(current_part_size).expect("test part size should fit i64"), + None, + None, + ); + } + + fi + } + + fn codec_streaming_test_object_info(fi: &FileInfo) -> ObjectInfo { + ObjectInfo::from_file_info(fi, "bucket", "object", false) + } + + #[test] + fn codec_streaming_reader_gate_is_conservative() { + temp_env::with_vars( + [ + (ENV_RUSTFS_GET_CODEC_STREAMING_ENABLE, Some("true")), + (ENV_RUSTFS_GET_CODEC_STREAMING_MIN_SIZE, Some("1")), + ], + || { + let fi = codec_streaming_test_fileinfo(1024, 1); + let object_info = codec_streaming_test_object_info(&fi); + assert_eq!( + get_codec_streaming_reader_decision(&None, &object_info, &fi, true), + GetCodecStreamingDecision::Use + ); + + assert_eq!( + get_codec_streaming_reader_decision(&None, &object_info, &fi, false), + GetCodecStreamingDecision::Fallback(GetCodecStreamingFallbackReason::LockOptimizationDisabled) + ); + + let range = Some(HTTPRangeSpec { + is_suffix_length: false, + start: 0, + end: 1, + }); + assert_eq!( + get_codec_streaming_reader_decision(&range, &object_info, &fi, true), + GetCodecStreamingDecision::Fallback(GetCodecStreamingFallbackReason::Range) + ); + + let multipart_fi = codec_streaming_test_fileinfo(1024, 2); + let multipart_object_info = codec_streaming_test_object_info(&multipart_fi); + assert_eq!( + get_codec_streaming_reader_decision(&None, &multipart_object_info, &multipart_fi, true), + GetCodecStreamingDecision::Fallback(GetCodecStreamingFallbackReason::Multipart) + ); + + let mut encrypted_fi = fi.clone(); + encrypted_fi + .metadata + .insert("x-amz-server-side-encryption".to_string(), "AES256".to_string()); + let encrypted = codec_streaming_test_object_info(&encrypted_fi); + assert_eq!( + get_codec_streaming_reader_decision(&None, &encrypted, &fi, true), + GetCodecStreamingDecision::Fallback(GetCodecStreamingFallbackReason::Encrypted) + ); + + let mut compressed_fi = fi.clone(); + insert_str(&mut compressed_fi.metadata, SUFFIX_COMPRESSION, "lz4".to_string()); + let compressed = codec_streaming_test_object_info(&compressed_fi); + assert_eq!( + get_codec_streaming_reader_decision(&None, &compressed, &fi, true), + GetCodecStreamingDecision::Fallback(GetCodecStreamingFallbackReason::Compressed) + ); + + let small_fi = codec_streaming_test_fileinfo(0, 1); + let small_object_info = codec_streaming_test_object_info(&small_fi); + assert_eq!( + get_codec_streaming_reader_decision(&None, &small_object_info, &small_fi, true), + GetCodecStreamingDecision::Fallback(GetCodecStreamingFallbackReason::BelowMinSize) + ); + + let mut remote_fi = fi; + remote_fi.transition_status = crate::bucket::lifecycle::lifecycle::TRANSITION_COMPLETE.to_string(); + let remote = codec_streaming_test_object_info(&remote_fi); + assert_eq!( + get_codec_streaming_reader_decision(&None, &remote, &remote_fi, true), + GetCodecStreamingDecision::Fallback(GetCodecStreamingFallbackReason::Remote) + ); + }, + ); + } + + #[test] + fn codec_streaming_fallback_metric_labels_are_stable() { + assert_eq!(GetCodecStreamingFallbackReason::Disabled.as_str(), "disabled"); + assert_eq!( + GetCodecStreamingFallbackReason::LockOptimizationDisabled.as_str(), + "lock_optimization_disabled" + ); + assert_eq!(GetCodecStreamingFallbackReason::Range.as_str(), "range"); + assert_eq!(GetCodecStreamingFallbackReason::BelowMinSize.as_str(), "below_min_size"); + assert_eq!(GetCodecStreamingFallbackReason::Encrypted.as_str(), "encrypted"); + assert_eq!(GetCodecStreamingFallbackReason::Compressed.as_str(), "compressed"); + assert_eq!(GetCodecStreamingFallbackReason::Remote.as_str(), "remote"); + assert_eq!(GetCodecStreamingFallbackReason::Multipart.as_str(), "multipart"); + assert_eq!(GetCodecStreamingFallbackReason::InvalidMinSize.as_str(), "invalid_min_size"); + } + + #[test] + fn codec_streaming_reader_gate_defaults_to_disabled() { + temp_env::with_vars( + [ + (ENV_RUSTFS_GET_CODEC_STREAMING_ENABLE, None::<&str>), + (ENV_RUSTFS_GET_CODEC_STREAMING_MIN_SIZE, Some("1")), + ], + || { + let fi = codec_streaming_test_fileinfo(1024, 1); + let object_info = codec_streaming_test_object_info(&fi); + + assert_eq!( + get_codec_streaming_reader_decision(&None, &object_info, &fi, true), + GetCodecStreamingDecision::Fallback(GetCodecStreamingFallbackReason::Disabled) + ); + }, + ); + } + #[tokio::test] async fn collect_read_multiple_results_fails_early_when_quorum_is_impossible() { let started = std::time::Instant::now(); diff --git a/crates/ecstore/src/set_disk/shard_source.rs b/crates/ecstore/src/set_disk/shard_source.rs new file mode 100644 index 000000000..156416db0 --- /dev/null +++ b/crates/ecstore/src/set_disk/shard_source.rs @@ -0,0 +1,183 @@ +// 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 crate::disk::error::Error; + +#[derive(Debug, Clone, PartialEq, Eq)] +pub(crate) struct ShardSlot { + index: usize, + data: Option>, + error: Option, +} + +impl ShardSlot { + pub(crate) fn new(index: usize, data: Option>, error: Option) -> Self { + Self { index, data, error } + } + + pub(crate) fn data(index: usize, data: Vec) -> Self { + Self::new(index, Some(data), None) + } + + pub(crate) fn missing(index: usize, error: Error) -> Self { + Self::new(index, None, Some(error)) + } + + pub(crate) fn index(&self) -> usize { + self.index + } + + pub(crate) fn has_data(&self) -> bool { + self.data.is_some() + } + + pub(crate) fn data_bytes(&self) -> Option<&[u8]> { + self.data.as_deref() + } + + pub(crate) fn error(&self) -> Option<&Error> { + self.error.as_ref() + } +} + +#[derive(Debug, Clone, PartialEq, Eq)] +pub(crate) struct StripeReadState { + slots: Vec, + read_quorum: usize, +} + +impl StripeReadState { + pub(crate) fn new(slots: Vec, read_quorum: usize) -> Self { + Self { slots, read_quorum } + } + + pub(crate) fn from_parts(shards: Vec>>, errors: Vec>, read_quorum: usize) -> Self { + let slot_count = shards.len().max(errors.len()); + let mut slots = Vec::with_capacity(slot_count); + let mut shards = shards.into_iter(); + let mut errors = errors.into_iter(); + for index in 0..slot_count { + slots.push(ShardSlot::new(index, shards.next().flatten(), errors.next().flatten())); + } + Self::new(slots, read_quorum) + } + + pub(crate) fn available_shards(&self) -> usize { + self.slots.iter().filter(|slot| slot.has_data()).count() + } + + pub(crate) fn can_decode(&self) -> bool { + self.available_shards() >= self.read_quorum + } + + pub(crate) fn slots(&self) -> &[ShardSlot] { + &self.slots + } + + pub(crate) fn slot_by_index(&self, index: usize) -> Option<&ShardSlot> { + if let Some(slot) = self.slots.get(index) + && slot.index == index + { + return Some(slot); + } + self.slots.iter().find(|slot| slot.index == index) + } + + pub(crate) fn data_shards_complete(&self, data_shards: usize) -> bool { + (0..data_shards).all(|index| self.slot_by_index(index).is_some_and(ShardSlot::has_data)) + } + + pub(crate) fn into_parts(self) -> (Vec>>, Vec>) { + let part_count = self.slots.iter().map(|slot| slot.index).max().map_or(0, |index| index + 1); + let mut shards = Vec::with_capacity(part_count); + shards.resize_with(part_count, || None); + let mut errors = Vec::with_capacity(part_count); + errors.resize_with(part_count, || None); + for slot in self.slots { + shards[slot.index] = slot.data; + errors[slot.index] = slot.error; + } + (shards, errors) + } +} + +#[async_trait::async_trait] +pub(crate) trait ShardStripeSource: Send { + async fn read_next_stripe(&mut self) -> StripeReadState; +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn stripe_read_state_tracks_decode_quorum() { + let state = StripeReadState::new( + vec![ + ShardSlot::data(0, vec![1]), + ShardSlot::missing(1, Error::FileNotFound), + ShardSlot::data(2, vec![2]), + ], + 2, + ); + + assert_eq!(state.available_shards(), 2); + assert!(state.can_decode()); + assert_eq!(state.slots()[1].index(), 1); + } + + #[test] + fn stripe_read_state_preserves_shards_and_errors() { + let state = StripeReadState::new(vec![ShardSlot::missing(1, Error::FileCorrupt), ShardSlot::data(0, vec![1, 2, 3])], 2); + + assert!(!state.can_decode()); + let (shards, errors) = state.into_parts(); + assert_eq!(shards, vec![Some(vec![1, 2, 3]), None]); + assert_eq!(errors, vec![None, Some(Error::FileCorrupt)]); + } + + #[test] + fn stripe_read_state_builds_slots_from_parallel_reader_parts() { + let state = + StripeReadState::from_parts(vec![Some(vec![1]), None, Some(vec![3])], vec![None, Some(Error::FileNotFound)], 2); + + assert!(state.can_decode()); + assert_eq!(state.slots()[1].index(), 1); + assert_eq!(state.slots()[1].error(), Some(&Error::FileNotFound)); + } + + #[test] + fn stripe_read_state_reports_complete_data_shards_without_parity() { + let state = StripeReadState::from_parts(vec![Some(vec![1]), Some(vec![2]), None], Vec::new(), 2); + + assert!(state.data_shards_complete(2)); + assert_eq!(state.slots()[0].data_bytes(), Some(&[1][..])); + assert_eq!(state.slot_by_index(1).and_then(ShardSlot::data_bytes), Some(&[2][..])); + } + + #[test] + fn stripe_read_state_rejects_missing_data_shard_for_complete_fast_path() { + let state = StripeReadState::from_parts(vec![Some(vec![1]), None, Some(vec![3])], Vec::new(), 2); + + assert!(!state.data_shards_complete(2)); + } + + #[test] + fn stripe_read_state_finds_out_of_order_slots_by_index() { + let state = StripeReadState::new(vec![ShardSlot::data(2, vec![3]), ShardSlot::data(0, vec![1])], 2); + + assert_eq!(state.slot_by_index(0).and_then(ShardSlot::data_bytes), Some(&[1][..])); + assert!(state.slot_by_index(1).is_none()); + } +} diff --git a/crates/ecstore/src/set_disk/write.rs b/crates/ecstore/src/set_disk/write.rs index 2704f140b..ca27521ee 100644 --- a/crates/ecstore/src/set_disk/write.rs +++ b/crates/ecstore/src/set_disk/write.rs @@ -500,6 +500,8 @@ impl SetDisks { return Ok(()); } + self.invalidate_get_object_metadata_cache(bucket, object).await; + let mut futures = Vec::with_capacity(disks.len()); let mut errs = Vec::with_capacity(disks.len()); @@ -531,6 +533,8 @@ impl SetDisks { return Err(err); } + self.invalidate_get_object_metadata_cache(bucket, object).await; + Ok(()) } diff --git a/crates/io-metrics/src/lib.rs b/crates/io-metrics/src/lib.rs index b6418d5eb..2b9303fe7 100644 --- a/crates/io-metrics/src/lib.rs +++ b/crates/io-metrics/src/lib.rs @@ -182,6 +182,16 @@ fn saturating_sub_atomic(counter: &AtomicU64, bytes: u64) -> u64 { } } +#[inline(always)] +fn usize_to_f64(value: usize) -> f64 { + value as f64 +} + +#[inline(always)] +fn i64_non_negative_to_f64(value: i64) -> f64 { + value.max(0) as f64 +} + #[derive(Clone, Copy, Debug, Eq, PartialEq)] enum TrackedMemoryGauge { GetObjectBufferedBytes, @@ -266,9 +276,44 @@ pub fn record_get_object_completion(total_duration_secs: f64, response_size_byte pub fn record_get_object_stream_strategy(strategy: &str, buffer_size_bytes: usize, response_size_bytes: i64) { counter!("rustfs_io_get_object_stream_strategy_total", "strategy" => strategy.to_string()).increment(1); histogram!("rustfs_io_get_object_stream_buffer_size_bytes", "strategy" => strategy.to_string()) - .record(buffer_size_bytes as f64); + .record(usize_to_f64(buffer_size_bytes)); histogram!("rustfs_io_get_object_stream_response_size_bytes", "strategy" => strategy.to_string()) - .record(response_size_bytes.max(0) as f64); + .record(i64_non_negative_to_f64(response_size_bytes)); +} + +/// Record the response-body handoff shape from a GetObject reader into the S3 streaming body. +#[inline(always)] +pub fn record_get_object_response_handoff( + strategy: &str, + buffer_source: &str, + buffer_size_bytes: usize, + response_size_bytes: i64, + duration_secs: f64, +) { + counter!( + "rustfs_io_get_object_response_handoff_total", + "strategy" => strategy.to_string(), + "buffer_source" => buffer_source.to_string() + ) + .increment(1); + histogram!( + "rustfs_io_get_object_response_handoff_buffer_size_bytes", + "strategy" => strategy.to_string(), + "buffer_source" => buffer_source.to_string() + ) + .record(usize_to_f64(buffer_size_bytes)); + histogram!( + "rustfs_io_get_object_response_handoff_response_size_bytes", + "strategy" => strategy.to_string(), + "buffer_source" => buffer_source.to_string() + ) + .record(i64_non_negative_to_f64(response_size_bytes)); + histogram!( + "rustfs_io_get_object_response_handoff_duration_seconds", + "strategy" => strategy.to_string(), + "buffer_source" => buffer_source.to_string() + ) + .record(duration_secs); } /// Record I/O queue congestion observation. @@ -301,6 +346,169 @@ pub fn record_get_object_io_state( counter!("rustfs_io_strategy_selected_total", "level" => load_level.to_string()).increment(1); } +/// Record GetObject phase duration for the current read path. +#[inline(always)] +pub fn record_get_object_stage_duration(path: &'static str, stage: &'static str, duration_secs: f64) { + histogram!("rustfs_io_get_object_stage_duration_seconds", "path" => path, "stage" => stage).record(duration_secs); +} + +/// Record the selected GetObject reader path. +#[inline(always)] +pub fn record_get_object_reader_path(path: &'static str) { + counter!("rustfs_io_get_object_reader_path_total", "path" => path).increment(1); +} + +/// Record why the codec streaming reader was not selected. +#[inline(always)] +pub fn record_get_object_codec_streaming_fallback(reason: &'static str) { + counter!("rustfs_io_get_object_codec_streaming_fallback_total", "reason" => reason).increment(1); +} + +/// Record one decoded reader stripe processed by a GetObject read path. +#[inline(always)] +pub fn record_get_object_reader_stripe(path: &'static str) { + counter!("rustfs_io_get_object_reader_stripes_total", "path" => path).increment(1); +} + +/// Record bytes emitted by a GetObject reader path. +#[inline(always)] +pub fn record_get_object_reader_bytes(path: &'static str, bytes: usize) { + let bytes = u64::try_from(bytes).unwrap_or(u64::MAX); + counter!("rustfs_io_get_object_reader_bytes_total", "path" => path).increment(bytes); +} + +/// Record one reader buffer produced by a GetObject read path. +#[inline(always)] +pub fn record_get_object_reader_buffer(path: &'static str, role: &'static str, bytes: usize) { + histogram!("rustfs_io_get_object_reader_buffer_bytes", "path" => path, "role" => role).record(usize_to_f64(bytes)); +} + +/// Record one copy from a GetObject reader's internal buffer into the downstream read buffer. +#[inline(always)] +pub fn record_get_object_reader_copy( + path: &'static str, + bytes: usize, + read_buf_remaining_before: usize, + output_remaining_before: usize, + duration_secs: f64, +) { + let bytes_counter = u64::try_from(bytes).unwrap_or(u64::MAX); + counter!("rustfs_io_get_object_reader_copy_chunks_total", "path" => path).increment(1); + counter!("rustfs_io_get_object_reader_copy_bytes_total", "path" => path).increment(bytes_counter); + histogram!("rustfs_io_get_object_reader_copy_bytes", "path" => path).record(usize_to_f64(bytes)); + histogram!("rustfs_io_get_object_reader_copy_read_buf_remaining_bytes", "path" => path) + .record(usize_to_f64(read_buf_remaining_before)); + histogram!("rustfs_io_get_object_reader_copy_output_remaining_bytes", "path" => path) + .record(usize_to_f64(output_remaining_before)); + histogram!("rustfs_io_get_object_reader_copy_duration_seconds", "path" => path).record(duration_secs); +} + +/// Record one downstream poll of the response-facing GetObject reader. +#[inline(always)] +pub fn record_get_object_reader_poll( + path: &'static str, + outcome: &'static str, + read_buf_remaining_before: usize, + filled_bytes: usize, + duration_secs: f64, +) { + let filled_bytes_counter = u64::try_from(filled_bytes).unwrap_or(u64::MAX); + counter!("rustfs_io_get_object_reader_poll_total", "path" => path, "outcome" => outcome).increment(1); + counter!("rustfs_io_get_object_reader_poll_filled_bytes_total", "path" => path, "outcome" => outcome) + .increment(filled_bytes_counter); + histogram!("rustfs_io_get_object_reader_poll_read_buf_remaining_bytes", "path" => path, "outcome" => outcome) + .record(usize_to_f64(read_buf_remaining_before)); + histogram!("rustfs_io_get_object_reader_poll_filled_bytes", "path" => path, "outcome" => outcome) + .record(usize_to_f64(filled_bytes)); + histogram!("rustfs_io_get_object_reader_poll_duration_seconds", "path" => path, "outcome" => outcome).record(duration_secs); +} + +/// Record a bounded prefetch outcome for a GetObject reader path. +#[inline(always)] +pub fn record_get_object_reader_prefetch(path: &'static str, outcome: &'static str) { + counter!("rustfs_io_get_object_reader_prefetch_total", "path" => path, "outcome" => outcome).increment(1); +} + +/// Record how long a GetObject reader spent waiting for a prefetch/fill result. +#[inline(always)] +pub fn record_get_object_reader_prefetch_wait(path: &'static str, duration_secs: f64) { + histogram!("rustfs_io_get_object_reader_prefetch_wait_seconds", "path" => path).record(duration_secs); +} + +/// Record one underlying shard read attempt for GetObject read-path attribution. +#[inline(always)] +pub fn record_get_object_shard_read( + path: &'static str, + role: &'static str, + outcome: &'static str, + bytes: usize, + duration_secs: f64, +) { + let bytes = u64::try_from(bytes).unwrap_or(u64::MAX); + counter!("rustfs_io_get_object_shard_read_total", "path" => path, "role" => role, "outcome" => outcome).increment(1); + counter!("rustfs_io_get_object_shard_read_bytes_total", "path" => path, "role" => role, "outcome" => outcome) + .increment(bytes); + histogram!("rustfs_io_get_object_shard_read_duration_seconds", "path" => path, "role" => role, "outcome" => outcome) + .record(duration_secs); +} + +#[inline(always)] +fn shard_read_fanout_to_f64(value: usize) -> f64 { + u32::try_from(value).map(f64::from).unwrap_or(f64::from(u32::MAX)) +} + +/// Record per-stripe shard-read fanout shape for GetObject read-path attribution. +#[inline(always)] +pub fn record_get_object_shard_read_fanout( + path: &'static str, + scheduled: usize, + completed: usize, + successful: usize, + failed: usize, +) { + histogram!("rustfs_io_get_object_shard_read_scheduled", "path" => path).record(shard_read_fanout_to_f64(scheduled)); + histogram!("rustfs_io_get_object_shard_read_completed", "path" => path).record(shard_read_fanout_to_f64(completed)); + histogram!("rustfs_io_get_object_shard_read_successful", "path" => path).record(shard_read_fanout_to_f64(successful)); + histogram!("rustfs_io_get_object_shard_read_failed", "path" => path).record(shard_read_fanout_to_f64(failed)); +} + +/// Record GetObject metadata resolution duration. +#[inline(always)] +pub fn record_get_object_metadata_phase_duration(duration_secs: f64) { + record_get_object_stage_duration("legacy_duplex", "metadata", duration_secs); +} + +/// Record GetObject shard reader setup duration. +#[inline(always)] +pub fn record_get_object_shard_reader_setup_duration(duration_secs: f64) { + record_get_object_stage_duration("legacy_duplex", "reader_setup", duration_secs); +} + +/// Record GetObject erasure decode duration. +#[inline(always)] +pub fn record_get_object_decode_duration(duration_secs: f64) { + record_get_object_stage_duration("legacy_duplex", "decode", duration_secs); +} + +/// Record GetObject downstream write wait while emitting decoded data. +#[inline(always)] +pub fn record_get_object_duplex_backpressure_duration(duration_secs: f64) { + record_get_object_stage_duration("legacy_duplex", "duplex_backpressure", duration_secs); +} + +/// Record GetObject read pipeline failures using bounded labels. +#[inline(always)] +pub fn record_get_object_pipeline_failure(stage: &'static str, reason: &'static str) { + counter!("rustfs_io_get_object_pipeline_failures_total", "path" => "legacy_duplex", "stage" => stage, "reason" => reason) + .increment(1); +} + +/// Record GetObject read pipeline failures for an explicit bounded path label. +#[inline(always)] +pub fn record_get_object_pipeline_failure_for_path(path: &'static str, stage: &'static str, reason: &'static str) { + counter!("rustfs_io_get_object_pipeline_failures_total", "path" => path, "stage" => stage, "reason" => reason).increment(1); +} + /// Record a zero-copy read operation. /// /// # Arguments @@ -895,6 +1103,29 @@ mod tests { record_get_object(50.0, 2048); } + #[test] + fn test_record_get_object_stage_metrics() { + record_get_object_stage_duration("s3_handler", "request_context", 0.001); + record_get_object_reader_path("codec_streaming"); + record_get_object_codec_streaming_fallback("range"); + record_get_object_reader_stripe("codec_streaming"); + record_get_object_reader_bytes("codec_streaming", 1024); + record_get_object_reader_buffer("codec_streaming", "output", 1024); + record_get_object_reader_copy("codec_streaming", 512, 8192, 1024, 0.0001); + record_get_object_reader_poll("codec_streaming", "ready_data", 8192, 512, 0.0002); + record_get_object_reader_prefetch("codec_streaming", "stored"); + record_get_object_reader_prefetch_wait("codec_streaming", 0.0002); + record_get_object_response_handoff("standard", "selected", 8192, 1024, 0.0001); + record_get_object_metadata_phase_duration(0.002); + record_get_object_shard_reader_setup_duration(0.003); + record_get_object_decode_duration(0.004); + record_get_object_duplex_backpressure_duration(0.005); + record_get_object_pipeline_failure("decode", "read_quorum"); + record_get_object_pipeline_failure_for_path("codec_streaming", "decode", "read_quorum"); + + assert!(0.005_f64.is_sign_positive()); + } + #[test] fn test_record_put_object() { record_put_object(200.0, 1024 * 1024, true); diff --git a/rustfs/src/app/object_usecase.rs b/rustfs/src/app/object_usecase.rs index 1ebcaaeab..e9fca1507 100644 --- a/rustfs/src/app/object_usecase.rs +++ b/rustfs/src/app/object_usecase.rs @@ -308,6 +308,27 @@ impl GetObjectStreamStrategy { const LARGE_SEQUENTIAL_GET_THRESHOLD_BYTES: i64 = 1024 * 1024 * 1024; const LARGE_SEQUENTIAL_GET_STREAM_BUFFER_CAP_BYTES: usize = 4 * MI_B; const LARGE_SEQUENTIAL_GET_READAHEAD_MULTIPLIER: usize = 2; +const ENV_RUSTFS_GET_READER_STREAM_BUFFER_SIZE: &str = "RUSTFS_GET_READER_STREAM_BUFFER_SIZE"; +const GET_READER_STREAM_BUFFER_SOURCE_SELECTED: &str = "selected"; +const GET_READER_STREAM_BUFFER_SOURCE_ENV_OVERRIDE: &str = "env_override"; + +fn get_reader_stream_buffer_size_override() -> Option { + static GET_READER_STREAM_BUFFER_SIZE_OVERRIDE: OnceLock> = OnceLock::new(); + *GET_READER_STREAM_BUFFER_SIZE_OVERRIDE.get_or_init(|| { + std::env::var(ENV_RUSTFS_GET_READER_STREAM_BUFFER_SIZE) + .ok() + .and_then(|value| value.parse::().ok()) + .filter(|value| *value > 0) + }) +} + +fn resolve_reader_stream_buffer_size(selected_size: usize, override_size: Option) -> (usize, &'static str) { + if let Some(override_size) = override_size.filter(|value| *value > 0) { + return (override_size, GET_READER_STREAM_BUFFER_SOURCE_ENV_OVERRIDE); + } + + (selected_size.max(1), GET_READER_STREAM_BUFFER_SOURCE_SELECTED) +} async fn enqueue_transitioned_delete_cleanup( store: Arc, @@ -1583,12 +1604,15 @@ impl DefaultObjectUsecase { where R: AsyncRead + Send + Sync + 'static, { - let expected = response_content_length.max(0) as usize; + let expected = usize::try_from(response_content_length.max(0)).unwrap_or(usize::MAX); + let (stream_buffer_size, buffer_source) = + resolve_reader_stream_buffer_size(stream_buffer_size, get_reader_stream_buffer_size_override()); rustfs_io_metrics::record_get_object_stream_strategy( stream_strategy.as_str(), stream_buffer_size, response_content_length, ); + let handoff_start = std::time::Instant::now(); #[cfg(feature = "tracing-chunk-debug")] let stream = { let mut emitted = 0usize; @@ -1609,7 +1633,15 @@ impl DefaultObjectUsecase { }; #[cfg(not(feature = "tracing-chunk-debug"))] let stream = ReaderStream::with_capacity(reader, stream_buffer_size); - Some(StreamingBlob::wrap(bytes_stream(stream, expected))) + let blob = StreamingBlob::wrap(bytes_stream(stream, expected)); + rustfs_io_metrics::record_get_object_response_handoff( + stream_strategy.as_str(), + buffer_source, + stream_buffer_size, + response_content_length, + handoff_start.elapsed().as_secs_f64(), + ); + Some(blob) } fn init_get_object_bootstrap(bucket: &str, key: &str, request_id: &str) -> S3Result { @@ -1759,7 +1791,13 @@ impl DefaultObjectUsecase { ) -> S3Result> { let h = req.headers.clone(); let io_planning = Self::acquire_get_object_io_planning(manager, wrapper, timeout_config, bucket, key).await?; + let store_lookup_start = std::time::Instant::now(); let store = get_validated_store(bucket).await?; + rustfs_io_metrics::record_get_object_stage_duration( + "s3_handler", + "store_lookup", + store_lookup_start.elapsed().as_secs_f64(), + ); let read_start = std::time::Instant::now(); let read_setup = @@ -1785,6 +1823,11 @@ impl DefaultObjectUsecase { .get_object_reader(bucket, key, rs.clone(), h, opts) .await .map_err(map_get_object_reader_error)?; + rustfs_io_metrics::record_get_object_stage_duration( + "s3_handler", + "store_reader_setup", + read_start.elapsed().as_secs_f64(), + ); let info = reader.object_info; @@ -2929,7 +2972,13 @@ impl DefaultObjectUsecase { let helper = OperationHelper::new(&req, EventName::ObjectAccessedGet, S3Operation::GetObject).suppress_event(); // mc get 3 + let request_context_start = std::time::Instant::now(); let request_context = Self::prepare_get_object_request_context(&req).await?; + rustfs_io_metrics::record_get_object_stage_duration( + "s3_handler", + "request_context", + request_context_start.elapsed().as_secs_f64(), + ); let GetObjectRequestContext { bucket, key, @@ -2974,7 +3023,14 @@ impl DefaultObjectUsecase { encryption_applied, } = read_setup; + let versioning_start = std::time::Instant::now(); let versioned = BucketVersioningSys::prefix_enabled(&bucket, &key).await; + rustfs_io_metrics::record_get_object_stage_duration( + "s3_handler", + "versioning_lookup", + versioning_start.elapsed().as_secs_f64(), + ); + let output_build_start = std::time::Instant::now(); let output_context = self .build_get_object_output_context( &req, @@ -3002,6 +3058,11 @@ impl DefaultObjectUsecase { versioned, ) .await?; + rustfs_io_metrics::record_get_object_stage_duration( + "s3_handler", + "output_build", + output_build_start.elapsed().as_secs_f64(), + ); let GetObjectOutputContext { output, event_info, @@ -5633,6 +5694,30 @@ mod tests { assert_eq!(small_buffer_size, 512 * 1024); } + #[test] + fn resolve_reader_stream_buffer_size_keeps_selected_default() { + let (buffer_size, source) = resolve_reader_stream_buffer_size(128 * 1024, None); + + assert_eq!(buffer_size, 128 * 1024); + assert_eq!(source, GET_READER_STREAM_BUFFER_SOURCE_SELECTED); + } + + #[test] + fn resolve_reader_stream_buffer_size_applies_positive_override() { + let (buffer_size, source) = resolve_reader_stream_buffer_size(128 * 1024, Some(MI_B)); + + assert_eq!(buffer_size, MI_B); + assert_eq!(source, GET_READER_STREAM_BUFFER_SOURCE_ENV_OVERRIDE); + } + + #[test] + fn resolve_reader_stream_buffer_size_ignores_zero_override() { + let (buffer_size, source) = resolve_reader_stream_buffer_size(128 * 1024, Some(0)); + + assert_eq!(buffer_size, 128 * 1024); + assert_eq!(source, GET_READER_STREAM_BUFFER_SOURCE_SELECTED); + } + #[test] fn should_use_zero_copy_rejects_encrypted_requests_with_sse_customer_algorithm() { let mut headers = HeaderMap::new(); diff --git a/scripts/run_get_codec_streaming_smoke.sh b/scripts/run_get_codec_streaming_smoke.sh new file mode 100755 index 000000000..baaed5563 --- /dev/null +++ b/scripts/run_get_codec_streaming_smoke.sh @@ -0,0 +1,401 @@ +#!/usr/bin/env bash +set -euo pipefail + +# Local GET benchmark harness for the experimental codec streaming read path. +# +# This script intentionally keeps the benchmark orchestration thin: +# - start one local RustFS server with controlled GET/codec/scanner env +# - run the existing enhanced object benchmark in GET mode +# - optionally run legacy and codec profiles back-to-back for A/B comparison + +SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" +PROJECT_ROOT="$(cd "${SCRIPT_DIR}/.." && pwd)" +ENHANCED_BENCH="${PROJECT_ROOT}/scripts/run_object_batch_bench_enhanced.sh" + +ADDRESS="127.0.0.1:19030" +ACCESS_KEY="rustfsadmin" +SECRET_KEY="rustfsadmin" +BUCKET="rustfs-get-codec-smoke" +REGION="us-east-1" +SIZES="1MiB,4MiB,10MiB" +CONCURRENCY=32 +DURATION="15s" +ROUNDS=3 +RETRY_PER_ROUND=1 +ROUND_COOLDOWN_SECS=0 +MODE="both" +OUT_DIR="" +RUSTFS_BIN="${PROJECT_ROOT}/target/release/rustfs" +WARP_BIN="warp" +CODEC_MIN_SIZE=1 +RUST_LOG="warn" +HEALTH_TIMEOUT_SECS=60 +DRY_RUN=false +SKIP_BUILD=false + +SERVER_PID="" + +usage() { + cat <<'USAGE' +Usage: + scripts/run_get_codec_streaming_smoke.sh [options] + +Purpose: + Start a local RustFS server and run GET warp benchmarks for the legacy read + path, the experimental codec streaming read path, or both. + +Core options: + --mode Which profile(s) to run (default: both) + --address RustFS listen address (default: 127.0.0.1:19030) + --bucket Benchmark bucket (default: rustfs-get-codec-smoke) + --sizes Object sizes (default: 1MiB,4MiB,10MiB) + --concurrency warp concurrency (default: 32) + --duration warp duration per round (default: 15s) + --rounds rounds per size (default: 3) + --retry-per-round failed-attempt retries per round (default: 1) + --round-cooldown-secs cooldown seconds after each completed round (default: 0) + --out-dir output directory (default: target/bench/get-codec-streaming-) + +Binary/options: + --rustfs-bin RustFS binary (default: target/release/rustfs) + --warp-bin warp binary (default: warp) + --codec-min-size RUSTFS_GET_CODEC_STREAMING_MIN_SIZE (default: 1) + --skip-build do not run cargo build --release -p rustfs + --dry-run print benchmark commands without starting RustFS + +Credentials: + --access-key RustFS access key (default: rustfsadmin) + --secret-key RustFS secret key (default: rustfsadmin) + --region S3 region (default: us-east-1) + +Output: + /legacy/warp/median_summary.csv + /codec/warp/median_summary.csv + /codec/warp/baseline_compare.csv when --mode both + //manifest.env + //rustfs.log + +Example: + scripts/run_get_codec_streaming_smoke.sh \ + --mode both --sizes 1MiB,4MiB,10MiB --concurrency 64 --duration 30s +USAGE +} + +log() { + printf '%s\n' "$*" +} + +die() { + echo "ERROR: $*" >&2 + exit 1 +} + +require_cmd() { + if ! command -v "$1" >/dev/null 2>&1; then + die "command not found: $1" + fi +} + +validate_positive_int() { + local value="$1" + local name="$2" + if ! [[ "$value" =~ ^[0-9]+$ ]] || [[ "$value" -le 0 ]]; then + die "$name must be a positive integer, got: $value" + fi +} + +validate_non_negative_int() { + local value="$1" + local name="$2" + if ! [[ "$value" =~ ^[0-9]+$ ]]; then + die "$name must be a non-negative integer, got: $value" + fi +} + +parse_args() { + while [[ $# -gt 0 ]]; do + case "$1" in + --mode) MODE="$2"; shift 2 ;; + --address) ADDRESS="$2"; shift 2 ;; + --bucket) BUCKET="$2"; shift 2 ;; + --sizes) SIZES="$2"; shift 2 ;; + --concurrency) CONCURRENCY="$2"; shift 2 ;; + --duration) DURATION="$2"; shift 2 ;; + --rounds) ROUNDS="$2"; shift 2 ;; + --retry-per-round) RETRY_PER_ROUND="$2"; shift 2 ;; + --round-cooldown-secs) ROUND_COOLDOWN_SECS="$2"; shift 2 ;; + --out-dir) OUT_DIR="$2"; shift 2 ;; + --rustfs-bin) RUSTFS_BIN="$2"; shift 2 ;; + --warp-bin) WARP_BIN="$2"; shift 2 ;; + --codec-min-size) CODEC_MIN_SIZE="$2"; shift 2 ;; + --access-key) ACCESS_KEY="$2"; shift 2 ;; + --secret-key) SECRET_KEY="$2"; shift 2 ;; + --region) REGION="$2"; shift 2 ;; + --skip-build) SKIP_BUILD=true; shift ;; + --dry-run) DRY_RUN=true; shift ;; + -h|--help) usage; exit 0 ;; + *) + usage >&2 + die "unknown arg: $1" + ;; + esac + done +} + +validate_args() { + case "$MODE" in + legacy|codec|both) ;; + *) die "--mode must be legacy, codec, or both" ;; + esac + + [[ -n "$ADDRESS" ]] || die "--address must not be empty" + [[ -n "$ACCESS_KEY" ]] || die "--access-key must not be empty" + [[ -n "$SECRET_KEY" ]] || die "--secret-key must not be empty" + [[ -n "$BUCKET" ]] || die "--bucket must not be empty" + [[ -n "$SIZES" ]] || die "--sizes must not be empty" + validate_positive_int "$CONCURRENCY" "--concurrency" + validate_positive_int "$ROUNDS" "--rounds" + validate_positive_int "$RETRY_PER_ROUND" "--retry-per-round" + validate_non_negative_int "$ROUND_COOLDOWN_SECS" "--round-cooldown-secs" + validate_positive_int "$CODEC_MIN_SIZE" "--codec-min-size" + validate_positive_int "$HEALTH_TIMEOUT_SECS" "--health-timeout-secs" + + [[ -x "$ENHANCED_BENCH" ]] || die "enhanced benchmark script is not executable: $ENHANCED_BENCH" + require_cmd curl + require_cmd git + + if [[ "$DRY_RUN" != "true" ]]; then + require_cmd cargo + fi +} + +setup_output() { + if [[ -z "$OUT_DIR" ]]; then + OUT_DIR="${PROJECT_ROOT}/target/bench/get-codec-streaming-$(date +%Y%m%d-%H%M%S)" + fi + mkdir -p "$OUT_DIR" +} + +build_rustfs_if_needed() { + if [[ "$DRY_RUN" == "true" || "$SKIP_BUILD" == "true" ]]; then + return + fi + + log "Building RustFS release binary..." + cargo build --release -p rustfs +} + +profile_codec_enabled() { + local profile="$1" + case "$profile" in + legacy) echo "false" ;; + codec) echo "true" ;; + *) die "unknown profile: $profile" ;; + esac +} + +profile_data_root() { + local profile="$1" + echo "${OUT_DIR}/${profile}/data" +} + +profile_volumes() { + local data_root="$1" + printf '%s/disk1 %s/disk2 %s/disk3 %s/disk4' "$data_root" "$data_root" "$data_root" "$data_root" +} + +endpoint_url() { + echo "http://${ADDRESS}" +} + +write_manifest() { + local profile="$1" + local profile_dir="$2" + local codec_enabled="$3" + local volumes="$4" + local git_head git_dirty_count + + git_head="$(git -C "$PROJECT_ROOT" rev-parse HEAD)" + git_dirty_count="$(git -C "$PROJECT_ROOT" status --porcelain | awk 'END { print NR + 0 }')" + + cat >"${profile_dir}/manifest.env" </dev/null 2>&1; then + kill "$SERVER_PID" >/dev/null 2>&1 || true + wait "$SERVER_PID" >/dev/null 2>&1 || true + fi + SERVER_PID="" + fi +} + +wait_for_health() { + local profile="$1" + local log_file="$2" + local health_url + health_url="$(endpoint_url)/health" + + for ((attempt = 1; attempt <= HEALTH_TIMEOUT_SECS; attempt++)); do + if curl -fsS --noproxy '*' --connect-timeout 2 --max-time 3 "$health_url" >/dev/null 2>&1; then + return + fi + if [[ -n "$SERVER_PID" ]] && ! kill -0 "$SERVER_PID" >/dev/null 2>&1; then + tail -n 80 "$log_file" >&2 || true + die "RustFS exited before health check passed for profile: $profile" + fi + sleep 1 + done + + tail -n 80 "$log_file" >&2 || true + die "RustFS health check timed out for profile: $profile" +} + +start_server() { + local profile="$1" + local profile_dir="${OUT_DIR}/${profile}" + local data_root + local volumes + local codec_enabled + local rustfs_log + + data_root="$(profile_data_root "$profile")" + volumes="$(profile_volumes "$data_root")" + codec_enabled="$(profile_codec_enabled "$profile")" + rustfs_log="${profile_dir}/rustfs.log" + + mkdir -p "${data_root}/disk1" "${data_root}/disk2" "${data_root}/disk3" "${data_root}/disk4" + write_manifest "$profile" "$profile_dir" "$codec_enabled" "$volumes" + + if [[ "$DRY_RUN" == "true" ]]; then + log "[DRY-RUN] start RustFS profile=${profile} endpoint=$(endpoint_url)" + log "[DRY-RUN] RUSTFS_VOLUMES=${volumes}" + return + fi + + [[ -x "$RUSTFS_BIN" ]] || die "RustFS binary is not executable: $RUSTFS_BIN" + + ( + export RUSTFS_ADDRESS="$ADDRESS" + export RUSTFS_ACCESS_KEY="$ACCESS_KEY" + export RUSTFS_SECRET_KEY="$SECRET_KEY" + export RUSTFS_VOLUMES="$volumes" + export RUSTFS_CONSOLE_ENABLE=false + export RUSTFS_GET_CODEC_STREAMING_ENABLE="$codec_enabled" + export RUSTFS_GET_CODEC_STREAMING_MIN_SIZE="$CODEC_MIN_SIZE" + export RUSTFS_REGION="$REGION" + export RUSTFS_RPC_SECRET="rustfs-get-codec-smoke-rpc-secret" + export RUSTFS_SCANNER_ENABLED=false + export RUSTFS_SCANNER_START_DELAY_SECS=3600 + export RUSTFS_SCANNER_CYCLE=3600 + export RUSTFS_UNSAFE_BYPASS_DISK_CHECK=true + export RUST_LOG + exec "$RUSTFS_BIN" + ) >"$rustfs_log" 2>&1 & + + SERVER_PID="$!" + wait_for_health "$profile" "$rustfs_log" +} + +run_bench() { + local profile="$1" + local baseline_csv="${2:-}" + local profile_dir="${OUT_DIR}/${profile}" + local bench_dir="${profile_dir}/warp" + local cmd=( + "$ENHANCED_BENCH" + --tool warp + --endpoint "$(endpoint_url)" + --access-key "$ACCESS_KEY" + --secret-key "$SECRET_KEY" + --bucket "$BUCKET" + --region "$REGION" + --warp-bin "$WARP_BIN" + --warp-mode get + --sizes "$SIZES" + --concurrency "$CONCURRENCY" + --duration "$DURATION" + --rounds "$ROUNDS" + --retry-per-round "$RETRY_PER_ROUND" + --round-cooldown-secs "$ROUND_COOLDOWN_SECS" + --out-dir "$bench_dir" + ) + + if [[ -n "$baseline_csv" ]]; then + cmd+=(--baseline-csv "$baseline_csv") + fi + if [[ "$DRY_RUN" == "true" ]]; then + cmd+=(--dry-run) + fi + + log "Running ${profile} GET benchmark..." + "${cmd[@]}" +} + +run_profile() { + local profile="$1" + local baseline_csv="${2:-}" + + stop_server + start_server "$profile" + run_bench "$profile" "$baseline_csv" + stop_server + + log "Median summary: ${OUT_DIR}/${profile}/warp/median_summary.csv" + if [[ -f "${OUT_DIR}/${profile}/warp/baseline_compare.csv" ]]; then + log "Baseline compare: ${OUT_DIR}/${profile}/warp/baseline_compare.csv" + fi +} + +main() { + parse_args "$@" + validate_args + setup_output + build_rustfs_if_needed + + trap stop_server EXIT INT TERM + + log "Output dir: $OUT_DIR" + case "$MODE" in + legacy) + run_profile legacy + ;; + codec) + run_profile codec + ;; + both) + run_profile legacy + run_profile codec "${OUT_DIR}/legacy/warp/median_summary.csv" + ;; + esac + + log "GET codec streaming smoke finished." +} + +main "$@" diff --git a/scripts/run_object_batch_bench_enhanced.sh b/scripts/run_object_batch_bench_enhanced.sh index 0aadf5722..4af1d1ad1 100755 --- a/scripts/run_object_batch_bench_enhanced.sh +++ b/scripts/run_object_batch_bench_enhanced.sh @@ -73,6 +73,7 @@ Enhanced options: --retry-per-round Retry count per failed round (default: 2) --retry-sleep-secs Sleep seconds between retries (default: 2) --cooldown-secs Sleep seconds between rounds/sizes (default: 0) + --round-cooldown-secs Compatibility alias for --cooldown-secs --baseline-csv Baseline median CSV to compare --extra-args Extra args appended to tool command, quoted as one string @@ -130,6 +131,7 @@ parse_args() { --retry-per-round) RETRY_PER_ROUND="$2"; shift 2 ;; --retry-sleep-secs) RETRY_SLEEP_SECS="$2"; shift 2 ;; --cooldown-secs) COOLDOWN_SECS="$2"; shift 2 ;; + --round-cooldown-secs) COOLDOWN_SECS="$2"; shift 2 ;; --baseline-csv) BASELINE_CSV="$2"; shift 2 ;; --extra-args) # shellcheck disable=SC2206 @@ -421,12 +423,13 @@ run_one_attempt() { run_size() { local size="$1" - local round success attempt rc + local round success attempt rc max_attempts + max_attempts=$((RETRY_PER_ROUND + 1)) for ((round=1; round<=ROUNDS; round++)); do success="no" - for ((attempt=1; attempt<=RETRY_PER_ROUND+1; attempt++)); do - echo "==== size=$size round=$round attempt=$attempt/${RETRY_PER_ROUND+1} ====" + for ((attempt=1; attempt<=max_attempts; attempt++)); do + echo "==== size=$size round=$round attempt=$attempt/$max_attempts ====" rc="$(run_one_attempt "$size" "$round" "$attempt")" if [[ "$rc" == "ok" || "$DRY_RUN" == "true" ]]; then success="yes"