diff --git a/crates/ecstore/src/data_movement.rs b/crates/ecstore/src/data_movement.rs index 0e1fcbf0b..e3fa89b1e 100644 --- a/crates/ecstore/src/data_movement.rs +++ b/crates/ecstore/src/data_movement.rs @@ -16,7 +16,7 @@ use crate::error::{Error, Result}; use crate::store::ECStore; use crate::store_api::{CompletePart, GetObjectReader, MultipartOperations, ObjectIO, ObjectInfo, ObjectOptions, PutObjReader}; use bytes::Bytes; -use rustfs_rio::{BlockReadable, BoxReadBlockFuture, EtagResolvable, HashReader, HashReaderDetector, Index, TryGetIndex}; +use rustfs_rio::{EtagResolvable, HashReader, HashReaderDetector, Index, TryGetIndex}; use std::io::Cursor; use std::pin::Pin; use std::sync::{ @@ -54,11 +54,6 @@ impl TryGetIndex for IndexedDataMovementRead } } -impl BlockReadable for IndexedDataMovementReader { - fn read_block<'a>(&'a mut self, buf: &'a mut [u8]) -> BoxReadBlockFuture<'a> { - Box::pin(rustfs_utils::read_full(self, buf)) - } -} pub fn decode_part_index(index: Option<&Bytes>) -> Option { let bytes = index?; let mut decoded = Index::new(); diff --git a/crates/ecstore/src/disk/disk_store.rs b/crates/ecstore/src/disk/disk_store.rs index 347e79c72..958d2c07a 100644 --- a/crates/ecstore/src/disk/disk_store.rs +++ b/crates/ecstore/src/disk/disk_store.rs @@ -21,7 +21,6 @@ use crate::disk::{ use crate::global::GLOBAL_LOCAL_DISK_ID_MAP; use bytes::Bytes; use rustfs_filemeta::{FileInfo, ObjectPartInfo, RawFileInfo}; -use rustfs_io_core::BoxChunkStream; use std::{ path::PathBuf, sync::{ @@ -739,14 +738,6 @@ impl DiskAPI for LocalDiskWrapper { .await } - async fn read_file_chunks(&self, volume: &str, path: &str, offset: usize, length: usize) -> Result { - self.track_disk_health( - || async { self.disk.read_file_chunks(volume, path, offset, length).await }, - get_max_timeout_duration(), - ) - .await - } - async fn append_file(&self, volume: &str, path: &str) -> Result { self.track_disk_health(|| async { self.disk.append_file(volume, path).await }, Duration::ZERO) .await diff --git a/crates/ecstore/src/disk/local.rs b/crates/ecstore/src/disk/local.rs index 66c1786d5..fca5282ad 100644 --- a/crates/ecstore/src/disk/local.rs +++ b/crates/ecstore/src/disk/local.rs @@ -30,19 +30,12 @@ use crate::disk::{ }; use crate::erasure_coding::bitrot_verify; use crate::global::{GLOBAL_IsErasureSD, GLOBAL_RootDiskThreshold}; -use bytes::{Bytes, BytesMut}; -use futures_util::{StreamExt, stream}; +use bytes::Bytes; use parking_lot::RwLock as ParkingLotRwLock; -use rustfs_config::{ - DEFAULT_OBJECT_ZERO_COPY_ENABLE, DEFAULT_OBJECT_ZERO_COPY_MAX_ACTIVE_MMAP_BYTES, DEFAULT_OBJECT_ZERO_COPY_MMAP_WINDOW_BYTES, - DEFAULT_OBJECT_ZERO_COPY_MODE, ENV_OBJECT_ZERO_COPY_ENABLE, ENV_OBJECT_ZERO_COPY_MAX_ACTIVE_MMAP_BYTES, - ENV_OBJECT_ZERO_COPY_MMAP_WINDOW_BYTES, ENV_OBJECT_ZERO_COPY_MODE, -}; use rustfs_filemeta::{ Cache, FileInfo, FileInfoOpts, FileMeta, MetaCacheEntry, MetacacheWriter, ObjectPartInfo, Opts, RawFileInfo, UpdateFn, get_file_info, read_xl_meta_no_data, }; -use rustfs_io_core::{BoxChunkStream, BytesPool, IoChunk, MappedChunk, PooledChunk}; use rustfs_utils::HashAlgorithm; use rustfs_utils::os::get_info; use rustfs_utils::path::{ @@ -53,7 +46,7 @@ use std::collections::HashMap; use std::collections::HashSet; use std::fmt::Debug; use std::io::SeekFrom; -use std::sync::atomic::{AtomicU32, AtomicUsize, Ordering}; +use std::sync::atomic::{AtomicU32, Ordering}; use std::sync::{Arc, OnceLock}; use std::time::Duration; use std::{ @@ -68,11 +61,6 @@ use tokio::time::interval; use tracing::{debug, error, info, warn}; use uuid::Uuid; -#[cfg(test)] -use serial_test::serial; -#[cfg(test)] -use temp_env::with_var; - #[derive(Debug, Clone)] pub struct FormatInfo { pub id: Option, @@ -109,410 +97,6 @@ pub struct LocalDisk { exit_signal: Option>, } -const LOCAL_CHUNK_FAST_PATH_MIN_BYTES: usize = 64 * 1024; -const LOCAL_DISK_POOLED_SOURCE_FALLBACK: &str = "fallback"; -const LOCAL_DISK_POOLED_SOURCE_COMPAT_COLLECT: &str = "compat_collect"; -const LOCAL_DISK_POOLED_SOURCE_COMPAT_DIRECT: &str = "compat_direct"; -const ACTIVE_MMAP_WINDOW_BUDGET_EXCEEDED_MESSAGE: &str = "active mmap window budget exceeded"; - -#[cfg(unix)] -const LOCAL_CHUNK_COMPAT_MAX_MAPPED_WINDOWS: usize = 1; - -static LOCAL_CHUNK_FALLBACK_POOL: OnceLock = OnceLock::new(); - -fn local_chunk_fallback_pool() -> &'static BytesPool { - LOCAL_CHUNK_FALLBACK_POOL.get_or_init(BytesPool::new_tiered) -} - -#[derive(Debug, Clone, Copy, PartialEq, Eq)] -enum LocalChunkZeroCopyMode { - Off, - Conservative, - Balanced, - Aggressive, -} - -impl LocalChunkZeroCopyMode { - fn from_env() -> Self { - match rustfs_utils::get_env_str(ENV_OBJECT_ZERO_COPY_MODE, DEFAULT_OBJECT_ZERO_COPY_MODE) - .trim() - .to_ascii_lowercase() - .as_str() - { - "off" => Self::Off, - "conservative" => Self::Conservative, - "aggressive" => Self::Aggressive, - _ => Self::Balanced, - } - } - - fn effective() -> Self { - if !rustfs_utils::get_env_bool(ENV_OBJECT_ZERO_COPY_ENABLE, DEFAULT_OBJECT_ZERO_COPY_ENABLE) { - return Self::Off; - } - - Self::from_env() - } - - const fn fast_path_min_bytes(self) -> usize { - match self { - Self::Aggressive => 1, - Self::Off | Self::Conservative | Self::Balanced => LOCAL_CHUNK_FAST_PATH_MIN_BYTES, - } - } - - const fn allows_multi_window(self) -> bool { - matches!(self, Self::Balanced | Self::Aggressive) - } - - const fn is_disabled(self) -> bool { - matches!(self, Self::Off) - } -} - -#[cfg(unix)] -static ACTIVE_LOCAL_MMAP_BYTES: AtomicUsize = AtomicUsize::new(0); - -#[cfg(unix)] -#[derive(Debug)] -struct ActiveMmapWindow { - mmap: memmap2::Mmap, - accounted_len: usize, -} - -#[cfg(unix)] -impl AsRef<[u8]> for ActiveMmapWindow { - fn as_ref(&self) -> &[u8] { - &self.mmap[..] - } -} - -#[cfg(unix)] -impl Drop for ActiveMmapWindow { - fn drop(&mut self) { - let remaining = ACTIVE_LOCAL_MMAP_BYTES - .fetch_sub(self.accounted_len, Ordering::AcqRel) - .saturating_sub(self.accounted_len); - rustfs_io_metrics::record_local_disk_active_mmap_bytes(remaining); - } -} - -#[cfg(unix)] -#[allow(unsafe_code)] -fn mmap_page_size() -> usize { - static PAGE_SIZE: OnceLock = OnceLock::new(); - - *PAGE_SIZE.get_or_init(|| { - let page_size = unsafe { libc::sysconf(libc::_SC_PAGESIZE) }; - if page_size <= 0 { 4096 } else { page_size as usize } - }) -} - -#[cfg(unix)] -fn configured_local_chunk_window_bytes() -> usize { - let page_size = mmap_page_size(); - rustfs_utils::get_env_usize(ENV_OBJECT_ZERO_COPY_MMAP_WINDOW_BYTES, DEFAULT_OBJECT_ZERO_COPY_MMAP_WINDOW_BYTES) - .max(page_size) - .div_ceil(page_size) - * page_size -} - -#[cfg(unix)] -fn configured_local_chunk_max_active_mmap_bytes() -> usize { - rustfs_utils::get_env_usize(ENV_OBJECT_ZERO_COPY_MAX_ACTIVE_MMAP_BYTES, DEFAULT_OBJECT_ZERO_COPY_MAX_ACTIVE_MMAP_BYTES) - .max(configured_local_chunk_window_bytes()) -} - -#[cfg(unix)] -fn should_prefer_pooled_zero_copy_compat(mode: LocalChunkZeroCopyMode, length: usize, window_bytes: usize) -> bool { - if mode.is_disabled() || !mode.allows_multi_window() || length < mode.fast_path_min_bytes() || window_bytes == 0 { - return false; - } - - length.div_ceil(window_bytes) > LOCAL_CHUNK_COMPAT_MAX_MAPPED_WINDOWS -} - -fn fallback_reason_for_local_mmap_error(err: &DiskError) -> rustfs_io_metrics::FallbackReason { - match err { - DiskError::Io(io_error) if io_error.to_string().contains(ACTIVE_MMAP_WINDOW_BUDGET_EXCEEDED_MESSAGE) => { - rustfs_io_metrics::FallbackReason::WindowLimitExceeded - } - _ => rustfs_io_metrics::FallbackReason::MmapUnavailable, - } -} - -#[cfg(unix)] -fn try_reserve_active_mmap_bytes(accounted_len: usize, max_active_bytes: usize) -> bool { - loop { - let current = ACTIVE_LOCAL_MMAP_BYTES.load(Ordering::Acquire); - let Some(next) = current.checked_add(accounted_len) else { - return false; - }; - if next > max_active_bytes { - return false; - } - - if ACTIVE_LOCAL_MMAP_BYTES - .compare_exchange_weak(current, next, Ordering::AcqRel, Ordering::Acquire) - .is_ok() - { - rustfs_io_metrics::record_local_disk_active_mmap_bytes(next); - return true; - } - } -} - -#[cfg(unix)] -#[allow(unsafe_code)] -fn map_file_region_bytes(file_path: &Path, offset: usize, length: usize, max_active_bytes: usize) -> Result { - use memmap2::MmapOptions; - - let aligned_offset = offset / mmap_page_size() * mmap_page_size(); - let logical_offset = offset - aligned_offset; - let map_length = logical_offset.checked_add(length).ok_or(DiskError::FileCorrupt)?; - let visible_end = logical_offset.checked_add(length).ok_or(DiskError::FileCorrupt)?; - if !try_reserve_active_mmap_bytes(map_length, max_active_bytes) { - return Err(DiskError::other(ACTIVE_MMAP_WINDOW_BUDGET_EXCEEDED_MESSAGE)); - } - let file = std::fs::File::open(file_path).map_err(DiskError::from)?; - - let mmap_result = - unsafe { MmapOptions::new().offset(aligned_offset as u64).len(map_length).map(&file) }.map_err(DiskError::other); - let mmap = match mmap_result { - Ok(mmap) => mmap, - Err(err) => { - let remaining = ACTIVE_LOCAL_MMAP_BYTES - .fetch_sub(map_length, Ordering::AcqRel) - .saturating_sub(map_length); - rustfs_io_metrics::record_local_disk_active_mmap_bytes(remaining); - return Err(err); - } - }; - let bytes = Bytes::from_owner(ActiveMmapWindow { - mmap, - accounted_len: map_length, - }); - - Ok(bytes.slice(logical_offset..visible_end)) -} - -#[cfg(unix)] -#[allow(unsafe_code)] -fn map_file_region_chunk(file_path: &Path, offset: usize, length: usize, max_active_bytes: usize) -> Result { - let bytes = map_file_region_bytes(file_path, offset, length, max_active_bytes)?; - MappedChunk::new(bytes, 0, length).map_err(DiskError::other) -} - -#[cfg(unix)] -#[derive(Debug)] -struct LocalMappedChunkStreamState { - file_path: PathBuf, - next_offset: usize, - remaining: usize, - window_bytes: usize, -} - -async fn read_file_pooled_chunk_from_path(file_path: PathBuf, offset: usize, length: usize) -> std::io::Result { - read_file_pooled_chunk_from_path_with_source(file_path, offset, length, LOCAL_DISK_POOLED_SOURCE_FALLBACK).await -} - -async fn read_file_pooled_chunk_from_path_with_source( - file_path: PathBuf, - offset: usize, - length: usize, - metric_source: &'static str, -) -> std::io::Result { - let mut file = File::open(file_path).await?; - if offset > 0 { - file.seek(SeekFrom::Start(offset as u64)).await?; - } - - let mut buffer = local_chunk_fallback_pool().acquire_buffer(length).await; - buffer.resize(length, 0); - file.read_exact(&mut buffer[..length]).await?; - rustfs_io_metrics::record_local_disk_pooled_chunk(metric_source, length); - Ok(IoChunk::Pooled(PooledChunk::new(buffer, length).map_err(std::io::Error::other)?)) -} - -async fn prepare_read_file_request(disk: &LocalDisk, volume: &str, path: &str) -> Result<(PathBuf, PathBuf, Metadata)> { - let volume_dir = disk.get_bucket_path(volume)?; - if !skip_access_checks(volume) { - access(&volume_dir) - .await - .map_err(|e| to_access_error(e, DiskError::VolumeAccessDenied))?; - } - - let file_path = disk.get_object_path(volume, path)?; - check_path_length(file_path.to_string_lossy().as_ref())?; - - let file_path_clone = file_path.clone(); - let meta = tokio::task::spawn_blocking(move || std::fs::metadata(&file_path_clone).map_err(DiskError::from)) - .await - .map_err(DiskError::from)??; - - Ok((volume_dir, file_path, meta)) -} - -fn validate_read_file_bounds(meta: &Metadata, offset: usize, length: usize) -> Result<()> { - let end_offset = offset.checked_add(length).ok_or(DiskError::FileCorrupt)?; - if meta.len() < end_offset as u64 { - error!( - "read_file: file size is less than offset + length {} + {} = {}", - offset, - length, - meta.len() - ); - return Err(DiskError::FileCorrupt); - } - - Ok(()) -} - -#[cfg(unix)] -fn build_lazy_mapped_chunk_stream( - file_path: PathBuf, - offset: usize, - length: usize, - window_bytes: usize, - max_active_bytes: usize, -) -> BoxChunkStream { - let state = LocalMappedChunkStreamState { - file_path, - next_offset: offset, - remaining: length, - window_bytes, - }; - - Box::pin(stream::unfold(Some(state), move |state| async move { - let mut state = match state { - Some(state) => state, - None => return None, - }; - - if state.remaining == 0 { - return None; - } - - let visible_len = state.remaining.min(state.window_bytes); - let window_offset = state.next_offset; - let file_path = state.file_path.clone(); - let mmap_result = - tokio::task::spawn_blocking(move || map_file_region_chunk(&file_path, window_offset, visible_len, max_active_bytes)) - .await; - - match mmap_result { - Ok(Ok(chunk)) => { - state.next_offset += visible_len; - state.remaining -= visible_len; - let next_state = if state.remaining == 0 { None } else { Some(state) }; - Some((Ok(IoChunk::Mapped(chunk)), next_state)) - } - Ok(Err(err)) => { - rustfs_io_metrics::record_io_fallback( - rustfs_io_metrics::IoStage::LocalDiskChunk, - fallback_reason_for_local_mmap_error(&err), - ); - debug!( - error = %err, - offset = window_offset, - len = visible_len, - "local disk lazy mmap window failed, falling back to buffered remainder" - ); - let fallback = - read_file_pooled_chunk_from_path(state.file_path.clone(), state.next_offset, state.remaining).await; - Some((fallback, None)) - } - Err(err) => { - rustfs_io_metrics::record_io_fallback( - rustfs_io_metrics::IoStage::LocalDiskChunk, - rustfs_io_metrics::FallbackReason::MmapUnavailable, - ); - debug!( - error = %err, - offset = window_offset, - len = visible_len, - "local disk lazy mmap task failed, falling back to buffered remainder" - ); - let fallback = - read_file_pooled_chunk_from_path(state.file_path.clone(), state.next_offset, state.remaining).await; - Some((fallback, None)) - } - } - })) -} - -async fn read_file_pooled_chunk_fallback( - disk: &LocalDisk, - volume_dir: &Path, - file_path: PathBuf, - offset: usize, - length: usize, -) -> Result { - read_file_pooled_chunk_fallback_with_source(disk, volume_dir, file_path, offset, length, LOCAL_DISK_POOLED_SOURCE_FALLBACK) - .await -} - -async fn read_file_pooled_chunk_fallback_with_source( - disk: &LocalDisk, - volume_dir: &Path, - file_path: PathBuf, - offset: usize, - length: usize, - metric_source: &'static str, -) -> Result { - let mut f = disk.open_file(file_path, O_RDONLY, volume_dir).await?; - - if offset > 0 { - f.seek(SeekFrom::Start(offset as u64)).await?; - } - - let mut buffer = local_chunk_fallback_pool().acquire_buffer(length).await; - buffer.resize(length, 0); - f.read_exact(&mut buffer[..length]).await?; - rustfs_io_metrics::record_local_disk_pooled_chunk(metric_source, length); - Ok(IoChunk::Pooled(PooledChunk::new(buffer, length).map_err(DiskError::other)?)) -} - -async fn collect_chunk_stream_bytes(mut stream: BoxChunkStream, expected_len: usize) -> Result { - let Some(first) = stream.next().await else { - return Ok(Bytes::new()); - }; - let first = first.map_err(DiskError::from)?; - let first_len = first.len(); - if matches!(first, IoChunk::Pooled(_)) { - rustfs_io_metrics::record_local_disk_pooled_chunk(LOCAL_DISK_POOLED_SOURCE_COMPAT_COLLECT, first_len); - } - let first_bytes = first.as_bytes(); - - let Some(second) = stream.next().await else { - return Ok(first_bytes); - }; - let second = second.map_err(DiskError::from)?; - let second_len = second.len(); - if matches!(second, IoChunk::Pooled(_)) { - rustfs_io_metrics::record_local_disk_pooled_chunk(LOCAL_DISK_POOLED_SOURCE_COMPAT_COLLECT, second_len); - } - let mut chunk_count = 2usize; - let mut total_bytes = first_len + second_len; - let mut buffer = BytesMut::with_capacity(expected_len); - buffer.extend_from_slice(first_bytes.as_ref()); - buffer.extend_from_slice(second.as_bytes().as_ref()); - - while let Some(chunk) = stream.next().await { - let chunk = chunk.map_err(DiskError::from)?; - let chunk_len = chunk.len(); - if matches!(chunk, IoChunk::Pooled(_)) { - rustfs_io_metrics::record_local_disk_pooled_chunk(LOCAL_DISK_POOLED_SOURCE_COMPAT_COLLECT, chunk_len); - } - chunk_count += 1; - total_bytes += chunk_len; - buffer.extend_from_slice(chunk.as_bytes().as_ref()); - } - - rustfs_io_metrics::record_local_disk_compat_collect(chunk_count, total_bytes); - Ok(buffer.freeze()) -} - impl Drop for LocalDisk { fn drop(&mut self) { if let Some(exit_signal) = self.exit_signal.take() { @@ -2261,96 +1845,105 @@ impl DiskAPI for LocalDisk { use std::time::Instant; let start = Instant::now(); - #[cfg(unix)] - { - let zero_copy_mode = LocalChunkZeroCopyMode::effective(); - let window_bytes = configured_local_chunk_window_bytes(); - if should_prefer_pooled_zero_copy_compat(zero_copy_mode, length, window_bytes) { - let (volume_dir, file_path, meta) = prepare_read_file_request(self, volume, path).await?; - validate_read_file_bounds(&meta, offset, length)?; - let chunk = read_file_pooled_chunk_fallback_with_source( - self, - &volume_dir, - file_path, - offset, - length, - LOCAL_DISK_POOLED_SOURCE_COMPAT_DIRECT, - ) - .await?; - let bytes = collect_chunk_stream_bytes(Box::pin(stream::iter(vec![Ok(chunk)])), length).await?; - debug!( - size = bytes.len(), - duration_ms = start.elapsed().as_secs_f64() * 1000.0, - "chunk_compat_read_pooled_success" - ); - return Ok(bytes); - } + let volume_dir = self.get_bucket_path(volume)?; + if !skip_access_checks(volume) { + access(&volume_dir) + .await + .map_err(|e| to_access_error(e, DiskError::VolumeAccessDenied))?; } - let bytes = collect_chunk_stream_bytes(self.read_file_chunks(volume, path, offset, length).await?, length).await?; - debug!( - size = bytes.len(), - duration_ms = start.elapsed().as_secs_f64() * 1000.0, - "chunk_compat_read_success" - ); - Ok(bytes) - } + let file_path = self.get_object_path(volume, path)?; + check_path_length(file_path.to_string_lossy().as_ref())?; - #[allow(unsafe_code)] - #[tracing::instrument(level = "debug", skip(self))] - async fn read_file_chunks(&self, volume: &str, path: &str, offset: usize, length: usize) -> Result { - let (volume_dir, file_path, meta) = prepare_read_file_request(self, volume, path).await?; - validate_read_file_bounds(&meta, offset, length)?; + // Verify file exists and get metadata + let file_path_clone = file_path.clone(); + let meta = tokio::task::spawn_blocking(move || std::fs::metadata(&file_path_clone).map_err(DiskError::from)) + .await + .map_err(DiskError::from)??; - let zero_copy_mode = LocalChunkZeroCopyMode::effective(); - if zero_copy_mode.is_disabled() { - rustfs_io_metrics::record_io_fallback( - rustfs_io_metrics::IoStage::LocalDiskChunk, - rustfs_io_metrics::FallbackReason::MmapDisabled, - ); - let chunk = read_file_pooled_chunk_fallback(self, &volume_dir, file_path, offset, length).await?; - return Ok(Box::pin(stream::iter(vec![Ok(chunk)]))); - } - - if length < zero_copy_mode.fast_path_min_bytes() { - rustfs_io_metrics::record_io_fallback( - rustfs_io_metrics::IoStage::LocalDiskChunk, - rustfs_io_metrics::FallbackReason::SmallObject, - ); - let chunk = read_file_pooled_chunk_fallback(self, &volume_dir, file_path, offset, length).await?; - return Ok(Box::pin(stream::iter(vec![Ok(chunk)]))); - } - - #[cfg(unix)] - { - let window_bytes = configured_local_chunk_window_bytes(); - - if !zero_copy_mode.allows_multi_window() && length > window_bytes { - rustfs_io_metrics::record_io_fallback( - rustfs_io_metrics::IoStage::LocalDiskChunk, - rustfs_io_metrics::FallbackReason::WindowLimitExceeded, - ); - let chunk = read_file_pooled_chunk_fallback(self, &volume_dir, file_path, offset, length).await?; - return Ok(Box::pin(stream::iter(vec![Ok(chunk)]))); - } - - return Ok(build_lazy_mapped_chunk_stream( - file_path, + let end_offset = offset.checked_add(length).ok_or(DiskError::FileCorrupt)?; + if meta.len() < end_offset as u64 { + error!( + "read_file_zero_copy: file size is less than offset + length {} + {} = {}", offset, length, - window_bytes, - configured_local_chunk_max_active_mmap_bytes(), - )); + meta.len() + ); + return Err(DiskError::FileCorrupt); } + // Unix: use mmap to read the data (copies into Bytes for safe ownership) + // Non-Unix: fall back to efficient read + #[cfg(unix)] + { + use memmap2::MmapOptions; + let file_path_clone = file_path.clone(); + + let bytes = tokio::task::spawn_blocking(move || { + let file = std::fs::File::open(&file_path_clone).map_err(DiskError::from)?; + + // mmap offsets on Unix must be page-size aligned. Align the + // mapping down to the nearest page boundary, then slice out the + // originally requested logical range. + let page_size = unsafe { libc::sysconf(libc::_SC_PAGESIZE) }; + if page_size <= 0 { + return Err(DiskError::other("failed to determine system page size")); + } + let page_size = page_size as u64; + let offset_u64 = offset as u64; + let aligned_offset = offset_u64 - (offset_u64 % page_size); + let logical_offset = (offset_u64 - aligned_offset) as usize; + let map_len = logical_offset + .checked_add(length) + .ok_or_else(|| DiskError::other("mmap length overflow"))?; + + // SAFETY: The file is opened as read-only, and we're mapping a region + // that we've already verified exists and is within file bounds. The + // file offset passed to mmap is page-size aligned as required on Unix. + let mmap = + unsafe { MmapOptions::new().offset(aligned_offset).len(map_len).map(&file) }.map_err(DiskError::other)?; + + // Copy only the requested logical range into a Bytes buffer. This + // avoids undefined behavior from treating OS-managed mmap memory as + // allocator-managed Vec storage, at the cost of an extra copy. + let end = logical_offset + .checked_add(length) + .ok_or_else(|| DiskError::other("mmap slice length overflow"))?; + Ok::(Bytes::copy_from_slice(&mmap[logical_offset..end])) + }) + .await + .map_err(DiskError::from)??; + + // Log successful mmap read metrics + let duration_ms = start.elapsed().as_secs_f64() * 1000.0; + + // Record mmap read metrics + rustfs_io_metrics::record_zero_copy_read(length, duration_ms); + + debug!(size = length, duration_ms = duration_ms, "mmap_read_success"); + + return Ok(bytes); + } + + // Non-Unix fallback: efficient read into Bytes #[cfg(not(unix))] { - rustfs_io_metrics::record_io_fallback( - rustfs_io_metrics::IoStage::LocalDiskChunk, - rustfs_io_metrics::FallbackReason::MmapUnavailable, - ); - let chunk = read_file_pooled_chunk_fallback(self, &volume_dir, file_path, offset, length).await?; - Ok(Box::pin(stream::iter(vec![Ok(chunk)]))) + // Record zero-copy fallback + rustfs_io_metrics::record_zero_copy_fallback("non_unix_platform"); + + debug!(reason = "non_unix_platform", "zero_copy_fallback"); + + let mut f = self.open_file(file_path, O_RDONLY, volume_dir).await?; + + if offset > 0 { + f.seek(SeekFrom::Start(offset as u64)).await?; + } + + let mut buffer = Vec::with_capacity(length); + buffer.resize(length, 0); + f.read_exact(&mut buffer).await?; + + Ok(Bytes::from(buffer)) } } @@ -3109,7 +2702,6 @@ async fn get_disk_info(drive_path: PathBuf) -> Result<(rustfs_utils::os::DiskInf #[cfg(test)] mod test { use super::*; - use futures_util::StreamExt; #[tokio::test] async fn test_skip_access_checks() { @@ -3450,22 +3042,6 @@ mod test { assert!(matches!(result, Err(DiskError::FileCorrupt))); } - #[tokio::test] - async fn test_read_file_zero_copy_supports_non_zero_offset() { - use tempfile::tempdir; - - let dir = tempdir().unwrap(); - let endpoint = Endpoint::try_from(dir.path().to_str().unwrap()).unwrap(); - let disk = LocalDisk::new(&endpoint, false).await.unwrap(); - - disk.make_volume("test-volume").await.unwrap(); - let content = Bytes::from_static(b"0123456789abcdef"); - disk.write_all("test-volume", "test-file.txt", content.clone()).await.unwrap(); - - let result = disk.read_file_zero_copy("test-volume", "test-file.txt", 3, 7).await.unwrap(); - assert_eq!(result, Bytes::from_static(b"3456789")); - } - #[test] fn test_is_valid_volname() { // Valid volume names (length >= 3) @@ -3563,274 +3139,6 @@ mod test { let _ = fs::remove_file(test_file).await; } - #[tokio::test] - #[serial] - async fn test_read_file_chunks_returns_pooled_chunk_for_local_fallback() { - let dir = tempfile::tempdir().unwrap(); - let bucket = "chunk-bucket"; - let object = "obj.txt"; - let content = b"chunk-data"; - - fs::create_dir_all(dir.path().join(bucket)).await.unwrap(); - fs::write(dir.path().join(bucket).join(object), content).await.unwrap(); - - let endpoint = Endpoint::try_from(dir.path().to_str().unwrap()).unwrap(); - let disk = LocalDisk::new(&endpoint, false).await.unwrap(); - - let mut stream = disk.read_file_chunks(bucket, object, 0, content.len()).await.unwrap(); - let first = stream.next().await.unwrap().unwrap(); - assert!(matches!(first, IoChunk::Pooled(_))); - assert_eq!(first.as_bytes(), Bytes::from_static(content)); - assert!(stream.next().await.is_none()); - } - - #[tokio::test] - #[serial] - async fn test_read_file_chunks_prefers_mapped_chunk_when_eligible() { - let dir = tempfile::tempdir().unwrap(); - let bucket = "chunk-bucket"; - let object = "obj-large.txt"; - let content = vec![7u8; LOCAL_CHUNK_FAST_PATH_MIN_BYTES]; - - fs::create_dir_all(dir.path().join(bucket)).await.unwrap(); - fs::write(dir.path().join(bucket).join(object), &content).await.unwrap(); - - let endpoint = Endpoint::try_from(dir.path().to_str().unwrap()).unwrap(); - let disk = LocalDisk::new(&endpoint, false).await.unwrap(); - - let mut stream = disk.read_file_chunks(bucket, object, 0, content.len()).await.unwrap(); - let first = stream.next().await.unwrap().unwrap(); - #[cfg(unix)] - assert!(matches!(first, IoChunk::Mapped(_))); - #[cfg(not(unix))] - assert!(matches!(first, IoChunk::Pooled(_))); - assert_eq!(first.as_bytes(), Bytes::from(content)); - assert!(stream.next().await.is_none()); - } - - #[tokio::test] - #[serial] - async fn test_read_file_chunks_falls_back_to_pooled_for_small_object() { - let dir = tempfile::tempdir().unwrap(); - let bucket = "chunk-bucket"; - let object = "obj-small.txt"; - let content = b"small-object"; - - fs::create_dir_all(dir.path().join(bucket)).await.unwrap(); - fs::write(dir.path().join(bucket).join(object), content).await.unwrap(); - - let endpoint = Endpoint::try_from(dir.path().to_str().unwrap()).unwrap(); - let disk = LocalDisk::new(&endpoint, false).await.unwrap(); - - let mut stream = disk.read_file_chunks(bucket, object, 0, content.len()).await.unwrap(); - let first = stream.next().await.unwrap().unwrap(); - assert!(matches!(first, IoChunk::Pooled(_))); - assert_eq!(first.as_bytes(), Bytes::from_static(content)); - assert!(stream.next().await.is_none()); - } - - #[tokio::test] - #[serial] - async fn test_read_file_chunks_supports_non_zero_offset() { - let dir = tempfile::tempdir().unwrap(); - let bucket = "chunk-bucket"; - let object = "obj-offset.txt"; - let content = vec![3u8; LOCAL_CHUNK_FAST_PATH_MIN_BYTES + 16]; - - fs::create_dir_all(dir.path().join(bucket)).await.unwrap(); - fs::write(dir.path().join(bucket).join(object), &content).await.unwrap(); - - let endpoint = Endpoint::try_from(dir.path().to_str().unwrap()).unwrap(); - let disk = LocalDisk::new(&endpoint, false).await.unwrap(); - - let mut stream = disk - .read_file_chunks(bucket, object, 1, LOCAL_CHUNK_FAST_PATH_MIN_BYTES) - .await - .unwrap(); - let first = stream.next().await.unwrap().unwrap(); - #[cfg(unix)] - assert!(matches!(first, IoChunk::Mapped(_))); - #[cfg(not(unix))] - assert!(matches!(first, IoChunk::Pooled(_))); - assert_eq!(first.as_bytes(), Bytes::copy_from_slice(&content[1..1 + LOCAL_CHUNK_FAST_PATH_MIN_BYTES])); - assert!(stream.next().await.is_none()); - } - - #[cfg(unix)] - #[tokio::test] - #[serial] - async fn test_read_file_chunks_splits_large_reads_into_multiple_windows() { - let dir = tempfile::tempdir().unwrap(); - let bucket = "chunk-bucket"; - let object = "obj-windowed.txt"; - let content = vec![5u8; DEFAULT_OBJECT_ZERO_COPY_MMAP_WINDOW_BYTES + 32]; - - fs::create_dir_all(dir.path().join(bucket)).await.unwrap(); - fs::write(dir.path().join(bucket).join(object), &content).await.unwrap(); - - let endpoint = Endpoint::try_from(dir.path().to_str().unwrap()).unwrap(); - let disk = LocalDisk::new(&endpoint, false).await.unwrap(); - - let mut stream = disk.read_file_chunks(bucket, object, 0, content.len()).await.unwrap(); - let first = stream.next().await.unwrap().unwrap(); - let second = stream.next().await.unwrap().unwrap(); - - assert!(matches!(first, IoChunk::Mapped(_))); - assert!(matches!(second, IoChunk::Mapped(_))); - assert_eq!(first.len(), DEFAULT_OBJECT_ZERO_COPY_MMAP_WINDOW_BYTES); - assert_eq!(second.len(), 32); - assert_eq!(first.as_bytes(), Bytes::from(vec![5u8; DEFAULT_OBJECT_ZERO_COPY_MMAP_WINDOW_BYTES])); - assert_eq!(second.as_bytes(), Bytes::from(vec![5u8; 32])); - assert!(stream.next().await.is_none()); - } - - #[cfg(unix)] - #[tokio::test] - #[serial] - async fn test_read_file_zero_copy_collects_multi_window_chunk_stream() { - let dir = tempfile::tempdir().unwrap(); - let bucket = "chunk-bucket"; - let object = "obj-zero-copy-compat.txt"; - let content = vec![6u8; DEFAULT_OBJECT_ZERO_COPY_MMAP_WINDOW_BYTES + 48]; - - fs::create_dir_all(dir.path().join(bucket)).await.unwrap(); - fs::write(dir.path().join(bucket).join(object), &content).await.unwrap(); - - let endpoint = Endpoint::try_from(dir.path().to_str().unwrap()).unwrap(); - let disk = LocalDisk::new(&endpoint, false).await.unwrap(); - - let bytes = disk.read_file_zero_copy(bucket, object, 0, content.len()).await.unwrap(); - assert_eq!(bytes, Bytes::from(content)); - } - - #[cfg(unix)] - #[test] - #[serial] - fn test_read_file_chunks_lazy_windows_reuse_single_window_budget() { - let page_size = mmap_page_size(); - let window_bytes = page_size.to_string(); - let max_active_bytes = page_size.to_string(); - - with_var(ENV_OBJECT_ZERO_COPY_ENABLE, Some("true"), || { - with_var(ENV_OBJECT_ZERO_COPY_MODE, Some("aggressive"), || { - with_var(ENV_OBJECT_ZERO_COPY_MMAP_WINDOW_BYTES, Some(window_bytes.clone()), || { - with_var(ENV_OBJECT_ZERO_COPY_MAX_ACTIVE_MMAP_BYTES, Some(max_active_bytes.clone()), || { - ACTIVE_LOCAL_MMAP_BYTES.store(0, Ordering::Release); - - let runtime = tokio::runtime::Runtime::new().unwrap(); - runtime.block_on(async { - let dir = tempfile::tempdir().unwrap(); - let bucket = "chunk-bucket"; - let object = "obj-budgeted.txt"; - let content = vec![9u8; page_size * 2]; - - fs::create_dir_all(dir.path().join(bucket)).await.unwrap(); - fs::write(dir.path().join(bucket).join(object), &content).await.unwrap(); - - let endpoint = Endpoint::try_from(dir.path().to_str().unwrap()).unwrap(); - let disk = LocalDisk::new(&endpoint, false).await.unwrap(); - - let mut stream = disk.read_file_chunks(bucket, object, 0, content.len()).await.unwrap(); - let first = stream.next().await.unwrap().unwrap(); - assert!(matches!(first, IoChunk::Mapped(_))); - assert_eq!(first.len(), page_size); - assert_eq!(ACTIVE_LOCAL_MMAP_BYTES.load(Ordering::Acquire), page_size); - - drop(first); - assert_eq!(ACTIVE_LOCAL_MMAP_BYTES.load(Ordering::Acquire), 0); - - let second = stream.next().await.unwrap().unwrap(); - assert!(matches!(second, IoChunk::Mapped(_))); - assert_eq!(second.len(), page_size); - assert_eq!(ACTIVE_LOCAL_MMAP_BYTES.load(Ordering::Acquire), page_size); - - drop(second); - assert_eq!(ACTIVE_LOCAL_MMAP_BYTES.load(Ordering::Acquire), 0); - assert!(stream.next().await.is_none()); - }); - - assert_eq!(ACTIVE_LOCAL_MMAP_BYTES.load(Ordering::Acquire), 0); - }); - }); - }); - }); - } - - #[test] - #[serial] - fn test_local_chunk_zero_copy_mode_respects_enable_and_mode_env() { - with_var(ENV_OBJECT_ZERO_COPY_ENABLE, Some("false"), || { - assert_eq!(LocalChunkZeroCopyMode::effective(), LocalChunkZeroCopyMode::Off); - }); - - with_var(ENV_OBJECT_ZERO_COPY_ENABLE, Some("true"), || { - with_var(ENV_OBJECT_ZERO_COPY_MODE, Some("aggressive"), || { - assert_eq!(LocalChunkZeroCopyMode::effective(), LocalChunkZeroCopyMode::Aggressive); - }); - }); - } - - #[cfg(unix)] - #[test] - #[serial] - fn test_should_prefer_pooled_zero_copy_compat_for_multi_window_requests() { - let window_bytes = 1024; - let balanced_multi_window_len = LOCAL_CHUNK_FAST_PATH_MIN_BYTES.max(window_bytes * 2); - - assert!(!should_prefer_pooled_zero_copy_compat( - LocalChunkZeroCopyMode::Off, - window_bytes * 2, - window_bytes - )); - assert!(!should_prefer_pooled_zero_copy_compat( - LocalChunkZeroCopyMode::Conservative, - window_bytes * 2, - window_bytes - )); - assert!(!should_prefer_pooled_zero_copy_compat( - LocalChunkZeroCopyMode::Balanced, - window_bytes, - window_bytes - )); - assert!(should_prefer_pooled_zero_copy_compat( - LocalChunkZeroCopyMode::Balanced, - balanced_multi_window_len, - window_bytes - )); - assert!(should_prefer_pooled_zero_copy_compat( - LocalChunkZeroCopyMode::Aggressive, - window_bytes * 2, - window_bytes - )); - } - - #[test] - fn test_fallback_reason_for_local_mmap_error_distinguishes_budget_limit() { - let budget_err = DiskError::other(ACTIVE_MMAP_WINDOW_BUDGET_EXCEEDED_MESSAGE); - let generic_err = DiskError::other("mmap failed"); - - assert_eq!( - fallback_reason_for_local_mmap_error(&budget_err), - rustfs_io_metrics::FallbackReason::WindowLimitExceeded - ); - assert_eq!( - fallback_reason_for_local_mmap_error(&generic_err), - rustfs_io_metrics::FallbackReason::MmapUnavailable - ); - } - - #[cfg(unix)] - #[test] - #[serial] - fn test_configured_local_chunk_window_bytes_aligns_to_page_size() { - with_var(ENV_OBJECT_ZERO_COPY_MMAP_WINDOW_BYTES, Some("12345"), || { - let page_size = mmap_page_size(); - let window_bytes = configured_local_chunk_window_bytes(); - assert!(window_bytes >= 12345); - assert_eq!(window_bytes % page_size, 0); - }); - } - #[test] fn test_is_root_path() { // Unix root path diff --git a/crates/ecstore/src/disk/mod.rs b/crates/ecstore/src/disk/mod.rs index 3bd2857c6..b00a036ce 100644 --- a/crates/ecstore/src/disk/mod.rs +++ b/crates/ecstore/src/disk/mod.rs @@ -41,7 +41,6 @@ use error::DiskError; use error::{Error, Result}; use local::LocalDisk; use rustfs_filemeta::{FileInfo, ObjectPartInfo, RawFileInfo}; -use rustfs_io_core::BoxChunkStream; use rustfs_madmin::info_commands::DiskMetrics; use serde::{Deserialize, Serialize}; use std::{fmt::Debug, path::PathBuf, sync::Arc}; @@ -296,14 +295,6 @@ impl DiskAPI for Disk { } } - #[tracing::instrument(skip(self))] - async fn read_file_chunks(&self, volume: &str, path: &str, offset: usize, length: usize) -> Result { - match self { - Disk::Local(local_disk) => local_disk.read_file_chunks(volume, path, offset, length).await, - Disk::Remote(remote_disk) => remote_disk.read_file_chunks(volume, path, offset, length).await, - } - } - #[tracing::instrument(skip(self))] async fn append_file(&self, volume: &str, path: &str) -> Result { match self { @@ -514,9 +505,6 @@ pub trait DiskAPI: Debug + Send + Sync + 'static { /// On other platforms, falls back to efficient read operations. async fn read_file_zero_copy(&self, volume: &str, path: &str, offset: usize, length: usize) -> Result; - /// Chunk-based file read compatibility layer for the zero-copy data plane. - async fn read_file_chunks(&self, volume: &str, path: &str, offset: usize, length: usize) -> Result; - async fn append_file(&self, volume: &str, path: &str) -> Result; async fn create_file(&self, origvolume: &str, volume: &str, path: &str, file_size: i64) -> Result; // ReadFileStream diff --git a/crates/ecstore/src/erasure_coding/bitrot.rs b/crates/ecstore/src/erasure_coding/bitrot.rs index 7a02abf81..01e2f4c3c 100644 --- a/crates/ecstore/src/erasure_coding/bitrot.rs +++ b/crates/ecstore/src/erasure_coding/bitrot.rs @@ -172,52 +172,6 @@ where } } -impl BitrotWriter { - fn write_inline_sync(&mut self, buf: &[u8]) -> std::io::Result { - if buf.is_empty() { - return Ok(0); - } - - if self.finished { - return Err(std::io::Error::new(std::io::ErrorKind::InvalidInput, "bitrot writer already finished")); - } - - if buf.len() > self.shard_size { - return Err(std::io::Error::new( - std::io::ErrorKind::InvalidInput, - format!("data size {} exceeds shard size {}", buf.len(), self.shard_size), - )); - } - - if buf.len() < self.shard_size { - self.finished = true; - } - - match &mut self.inner { - CustomWriter::InlineBuffer(data) => { - if self.hash_algo.size() > 0 { - let hash = self.hash_algo.hash_encode(buf); - if hash.as_ref().is_empty() { - error!("bitrot writer write hash error: hash is empty"); - return Err(std::io::Error::new(std::io::ErrorKind::InvalidInput, "hash is empty")); - } - data.extend_from_slice(hash.as_ref()); - } - data.extend_from_slice(buf); - Ok(buf.len()) - } - CustomWriter::Other(_) => Err(std::io::Error::other("inline sync write requires inline buffer writer")), - } - } - - fn shutdown_inline_sync(&mut self) -> std::io::Result<()> { - match self.inner { - CustomWriter::InlineBuffer(_) => Ok(()), - CustomWriter::Other(_) => Err(std::io::Error::other("inline sync shutdown requires inline buffer writer")), - } - } -} - async fn write_all_vectored(writer: &mut W, hash: &[u8], data: &[u8]) -> std::io::Result<()> where W: AsyncWrite + Unpin, @@ -326,10 +280,6 @@ impl CustomWriter { Self::Other(_) => None, } } - - pub fn is_inline_buffer(&self) -> bool { - matches!(self, Self::InlineBuffer(_)) - } } impl AsyncWrite for CustomWriter { @@ -447,24 +397,6 @@ impl BitrotWriterWrapper { self.bitrot_writer.shutdown().await } - pub fn is_inline_buffer(&self) -> bool { - matches!(self.writer_type, WriterType::InlineBuffer) - } - - pub fn write_inline_sync(&mut self, buf: &[u8]) -> std::io::Result { - if !self.is_inline_buffer() { - return Err(std::io::Error::other("inline sync write requires inline buffer writer")); - } - self.bitrot_writer.write_inline_sync(buf) - } - - pub fn shutdown_inline_sync(&mut self) -> std::io::Result<()> { - if !self.is_inline_buffer() { - return Err(std::io::Error::other("inline sync shutdown requires inline buffer writer")); - } - self.bitrot_writer.shutdown_inline_sync() - } - /// Extract the inline buffer data, consuming the wrapper pub fn into_inline_data(self) -> Option> { match self.writer_type { diff --git a/crates/ecstore/src/erasure_coding/decode.rs b/crates/ecstore/src/erasure_coding/decode.rs index e0a5790ae..9b17a03ca 100644 --- a/crates/ecstore/src/erasure_coding/decode.rs +++ b/crates/ecstore/src/erasure_coding/decode.rs @@ -17,7 +17,6 @@ use crate::disk::error_reduce::reduce_errs; use crate::erasure_coding::{BitrotReader, Erasure}; use futures::stream::{FuturesUnordered, StreamExt}; use pin_project_lite::pin_project; -use rustfs_io_core::{IoChunk, PooledChunk}; use std::io; use std::io::ErrorKind; use tokio::io::AsyncRead; @@ -156,68 +155,6 @@ fn get_data_block_len(shards: &[Option>], data_blocks: usize) -> usize { size } -fn block_window( - offset: usize, - length: usize, - block_size: usize, - block_index: usize, - start_block: usize, - end_block: usize, -) -> (usize, usize) { - let end_remainder = offset.saturating_add(length) % block_size; - if start_block == end_block { - (offset % block_size, length) - } else if block_index == start_block { - (offset % block_size, block_size - (offset % block_size)) - } else if block_index == end_block { - (0, if end_remainder == 0 { block_size } else { end_remainder }) - } else { - (0, block_size) - } -} - -fn take_data_blocks_as_chunks( - shards: &mut [Option>], - data_blocks: usize, - mut offset: usize, - length: usize, -) -> io::Result> { - if get_data_block_len(shards, data_blocks) < length { - error!("take_data_blocks_as_chunks get_data_block_len < length"); - return Err(io::Error::new(ErrorKind::UnexpectedEof, "Not enough data blocks to write")); - } - - let mut chunks = Vec::new(); - let mut remaining = length; - for block_op in shards.iter_mut().take(data_blocks) { - let Some(block) = block_op.take() else { - error!("take_data_blocks_as_chunks block_op.is_none()"); - return Err(io::Error::new(ErrorKind::UnexpectedEof, "Missing data block")); - }; - - if offset >= block.len() { - offset -= block.len(); - continue; - } - - let start = offset; - offset = 0; - let take = (block.len() - start).min(remaining); - let chunk = if start == 0 && take == block.len() { - IoChunk::Pooled(PooledChunk::from_vec(block)) - } else { - IoChunk::Pooled(PooledChunk::from_vec(block).slice(start, take)?) - }; - chunks.push(chunk); - remaining -= take; - if remaining == 0 { - break; - } - } - - Ok(chunks) -} - /// Write data blocks from encoded blocks to target, supporting offset and length async fn write_data_blocks( writer: &mut W, @@ -276,134 +213,6 @@ where Ok(total_written) } -pub(crate) struct ErasureChunkDecoder { - erasure: Erasure, - reader: ParallelReader, - offset: usize, - length: usize, - start_block: usize, - end_block: usize, - current_block: usize, - written: usize, - healable_error: Option, - finished: bool, -} - -impl ErasureChunkDecoder -where - R: AsyncRead + Unpin + Send + Sync, -{ - pub(crate) fn new( - erasure: Erasure, - readers: Vec>>, - offset: usize, - length: usize, - total_length: usize, - ) -> io::Result { - if readers.len() != erasure.data_shards + erasure.parity_shards { - return Err(io::Error::new(ErrorKind::InvalidInput, "Invalid number of readers")); - } - - let end_offset = offset - .checked_add(length) - .ok_or_else(|| io::Error::new(ErrorKind::InvalidInput, "offset + length exceeds total length"))?; - if end_offset > total_length { - return Err(io::Error::new(ErrorKind::InvalidInput, "offset + length exceeds total length")); - } - - let start_block = offset / erasure.block_size; - let end_block = if length == 0 { - start_block - } else { - end_offset.saturating_sub(1) / erasure.block_size - }; - let reader = ParallelReader::new(readers, erasure.clone(), offset, total_length); - - Ok(Self { - erasure, - reader, - offset, - length, - start_block, - end_block, - current_block: start_block, - written: 0, - healable_error: None, - finished: length == 0, - }) - } - - pub(crate) async fn next_chunks(&mut self) -> io::Result>> { - if self.finished { - return Ok(None); - } - - if self.current_block > self.end_block { - self.finished = true; - return Ok(None); - } - - let block_index = self.current_block; - self.current_block += 1; - - let (block_offset, block_length) = block_window( - self.offset, - self.length, - self.erasure.block_size, - block_index, - self.start_block, - self.end_block, - ); - if block_length == 0 { - self.finished = true; - return Ok(None); - } - - let (mut shards, errs) = self.reader.read().await; - - if self.healable_error.is_none() - && let (_, Some(err)) = reduce_errs(&errs, &[]) - && (err == Error::FileNotFound || err == Error::FileCorrupt) - { - self.healable_error = Some(err); - } - - if !self.reader.can_decode(&shards) { - self.finished = true; - error!("reconstructed chunk decoder can_decode errs: {:?}", &errs); - return Err(Error::ErasureReadQuorum.into()); - } - - if let Err(err) = self.erasure.decode_data(&mut shards) { - self.finished = true; - error!("reconstructed chunk decoder decode_data err: {:?}", err); - return Err(err); - } - - let chunks = take_data_blocks_as_chunks(&mut shards, self.erasure.data_shards, block_offset, block_length)?; - self.written += chunks.iter().map(IoChunk::len).sum::(); - Ok(Some(chunks)) - } - - pub(crate) fn written(&self) -> usize { - self.written - } - - pub(crate) fn finish_error(&self) -> Option { - if self.written < self.length { - Some(Error::LessData.into()) - } else { - None - } - } - - pub(crate) fn take_healable_error(&mut self) -> Option { - self.healable_error.take() - } -} - -pub(crate) type ReconstructedChunkDecoder = ErasureChunkDecoder; - impl Erasure { pub async fn decode( &self, @@ -511,7 +320,6 @@ mod tests { disk::error::DiskError, erasure_coding::{BitrotReader, BitrotWriter}, }; - use bytes::Bytes; use rustfs_utils::HashAlgorithm; use std::io::Cursor; @@ -652,47 +460,4 @@ mod tests { let reader_cursor = Cursor::new(buf); BitrotReader::new(reader_cursor, shard_size, hash_algo.clone(), false) } - - async fn create_bitrot_reader_from_shard( - shard: Bytes, - shard_size: usize, - hash_algo: &HashAlgorithm, - ) -> BitrotReader>> { - let writer = Cursor::new(Vec::new()); - let mut writer = BitrotWriter::new(writer, shard_size, hash_algo.clone()); - writer.write(shard.as_ref()).await.unwrap(); - let reader_cursor = Cursor::new(writer.into_inner().into_inner()); - BitrotReader::new(reader_cursor, shard_size, hash_algo.clone(), false) - } - - #[tokio::test] - async fn test_erasure_chunk_decoder_reconstructs_missing_data_shard_as_pooled_chunks() { - let erasure = Erasure::new(2, 1, 4); - let original = b"abcd"; - let encoded = erasure.encode_data(original).unwrap(); - let shard_size = erasure.shard_size(); - let hash_algo = HashAlgorithm::None; - - let readers = vec![ - None, - Some(create_bitrot_reader_from_shard(encoded[1].clone(), shard_size, &hash_algo).await), - Some(create_bitrot_reader_from_shard(encoded[2].clone(), shard_size, &hash_algo).await), - ]; - - let mut decoder = ErasureChunkDecoder::new(erasure, readers, 0, original.len(), original.len()).unwrap(); - let first_batch = decoder.next_chunks().await.unwrap().unwrap(); - assert!( - first_batch.iter().all(|chunk| matches!(chunk, IoChunk::Pooled(_))), - "reconstructed decoder should produce pooled chunks" - ); - let collected = first_batch - .into_iter() - .flat_map(|chunk| chunk.as_bytes().to_vec()) - .collect::>(); - - assert_eq!(collected, original); - assert!(decoder.next_chunks().await.unwrap().is_none()); - assert_eq!(decoder.written(), original.len()); - assert!(decoder.finish_error().is_none()); - } } diff --git a/crates/ecstore/src/erasure_coding/encode.rs b/crates/ecstore/src/erasure_coding/encode.rs index 86b4a6930..3b2c1841a 100644 --- a/crates/ecstore/src/erasure_coding/encode.rs +++ b/crates/ecstore/src/erasure_coding/encode.rs @@ -17,12 +17,11 @@ use crate::disk::error_reduce::count_errs; use crate::disk::error_reduce::{OBJECT_OP_IGNORED_ERRS, reduce_write_quorum_errs}; use crate::erasure_coding::BitrotWriterWrapper; use crate::erasure_coding::Erasure; -use crate::erasure_coding::erasure::{EncodeBlockBuffer, EncodedShardBlock, EncodedShardBufferPool}; use bytes::Bytes; use futures::StreamExt; use futures::stream::FuturesUnordered; -use rustfs_rio::BlockReadable; use std::sync::Arc; +use std::vec; use tokio::io::AsyncRead; use tokio::sync::mpsc; use tracing::error; @@ -33,164 +32,6 @@ pub(crate) struct MultiWriter<'a> { errs: Vec>, } -pub(crate) struct BlockAssembler { - reader: R, - block_buffer: EncodeBlockBuffer, - total_bytes: usize, -} - -impl BlockAssembler -where - R: AsyncRead + BlockReadable + Send + Sync + Unpin + 'static, -{ - pub(crate) fn new(reader: R, block_size: usize) -> Self { - Self { - reader, - block_buffer: EncodeBlockBuffer::new(block_size), - total_bytes: 0, - } - } - - pub(crate) async fn next_block(&mut self) -> std::io::Result>> { - match self.block_buffer.read_from_block(&mut self.reader).await { - Ok(n) if n > 0 => { - self.total_bytes += n; - Ok(Some(self.block_buffer.filled(n).to_vec())) - } - Ok(_) => Ok(None), - Err(e) if e.kind() == std::io::ErrorKind::UnexpectedEof => { - if let Some(inner) = e.get_ref() - && rustfs_rio::is_checksum_mismatch(inner) - { - return Err(std::io::Error::new(std::io::ErrorKind::InvalidData, e.to_string())); - } - Ok(None) - } - Err(e) => Err(e), - } - } - - pub(crate) fn total_bytes(&self) -> usize { - self.total_bytes - } - - pub(crate) fn into_inner(self) -> R { - self.reader - } -} - -#[derive(Clone)] -pub(crate) struct ErasureChunkEncoder { - erasure: Arc, - buffer_pool: EncodedShardBufferPool, -} - -impl ErasureChunkEncoder { - pub(crate) async fn new(erasure: Arc) -> Self { - let reusable_capacity = erasure.shard_size() * erasure.total_shard_count(); - Self { - erasure, - buffer_pool: EncodedShardBufferPool::with_prefill(reusable_capacity, 2).await, - } - } - - pub(crate) async fn encode_block(&self, block: &[u8]) -> std::io::Result { - let reusable_buffer = self.buffer_pool.acquire().await; - self.erasure.encode_data_block_with_buffer(block, reusable_buffer) - } - - pub(crate) async fn release(&self, block: EncodedShardBlock) { - self.buffer_pool.release(block).await; - } -} - -pub(crate) struct ErasureWritePipeline { - erasure: Arc, - write_quorum: usize, -} - -impl ErasureWritePipeline { - pub(crate) fn new(erasure: Arc, write_quorum: usize) -> Self { - Self { erasure, write_quorum } - } - - pub(crate) async fn run(&self, reader: R, writers: &mut [Option]) -> std::io::Result<(R, usize)> - where - R: AsyncRead + BlockReadable + Send + Sync + Unpin + 'static, - { - let (tx, mut rx) = mpsc::channel::(8); - let producer = ErasureChunkEncoder::new(self.erasure.clone()).await; - let writer_pool = producer.clone(); - let block_size = self.erasure.block_size; - - let task = tokio::spawn(async move { - let mut assembler = BlockAssembler::new(reader, block_size); - while let Some(block) = assembler.next_block().await? { - let res = producer.encode_block(&block).await?; - if let Err(err) = tx.send(res).await { - return Err(std::io::Error::other(format!("Failed to send encoded data : {err}"))); - } - } - - let total = assembler.total_bytes(); - Ok((assembler.into_inner(), total)) - }); - - let mut writers = MultiWriter::new(writers, self.write_quorum); - let mut write_err = None; - - while let Some(block) = rx.recv().await { - if block.is_empty() { - break; - } - let write_result = writers.write(&block).await; - writer_pool.release(block).await; - if let Err(err) = write_result { - write_err = Some(err); - break; - } - } - - if let Some(err) = write_err { - task.abort(); - let _ = task.await; - if let Err(shutdown_err) = writers.shutdown().await { - error!("failed to shutdown erasure writers after write error: {:?}", shutdown_err); - } - return Err(err); - } - - let (reader, total) = task.await??; - writers.shutdown().await?; - Ok((reader, total)) - } -} - -pub(crate) trait ShardSource { - fn shard_count(&self) -> usize; - fn shard(&self, idx: usize) -> Bytes; -} - -impl ShardSource for EncodedShardBlock { - fn shard_count(&self) -> usize { - self.shard_count() - } - - fn shard(&self, idx: usize) -> Bytes { - self.shard(idx) - } -} - -impl ShardSource for Vec { - fn shard_count(&self) -> usize { - self.len() - } - - fn shard(&self, idx: usize) -> Bytes { - self[idx].clone() - } -} - impl<'a> MultiWriter<'a> { pub fn new(writers: &'a mut [Option], write_quorum: usize) -> Self { let length = writers.len(); @@ -201,10 +42,10 @@ impl<'a> MultiWriter<'a> { } } - async fn write_shard(writer_opt: &mut Option, err: &mut Option, shard: Bytes) { + async fn write_shard(writer_opt: &mut Option, err: &mut Option, shard: &Bytes) { match writer_opt { Some(writer) => { - match writer.write(&shard).await { + match writer.write(shard).await { Ok(n) => { if n < shard.len() { *err = Some(Error::ShortWrite); @@ -224,40 +65,16 @@ impl<'a> MultiWriter<'a> { } } - fn write_shard_inline(writer_opt: &mut Option, err: &mut Option, shard: Bytes) { - match writer_opt { - Some(writer) => match writer.write_inline_sync(&shard) { - Ok(n) => { - if n < shard.len() { - *err = Some(Error::ShortWrite); - *writer_opt = None; - } else { - *err = None; - } - } - Err(e) => { - *err = Some(Error::from(e)); - } - }, - None => { - *err = Some(Error::DiskNotFound); - } - } - } - - pub async fn write(&mut self, data: &T) -> std::io::Result<()> - where - T: ShardSource, - { - assert_eq!(data.shard_count(), self.writers.len()); + pub async fn write(&mut self, data: Vec) -> std::io::Result<()> { + assert_eq!(data.len(), self.writers.len()); { let mut futures = FuturesUnordered::new(); - for (idx, (writer_opt, err)) in self.writers.iter_mut().zip(self.errs.iter_mut()).enumerate() { + for ((writer_opt, err), shard) in self.writers.iter_mut().zip(self.errs.iter_mut()).zip(data.iter()) { if err.is_some() { continue; // Skip if we already have an error for this writer } - futures.push(Self::write_shard(writer_opt, err, data.shard(idx))); + futures.push(Self::write_shard(writer_opt, err, shard)); } while let Some(()) = futures.next().await {} } @@ -295,45 +112,6 @@ impl<'a> MultiWriter<'a> { ))) } - pub fn write_inline(&mut self, data: &T) -> std::io::Result<()> - where - T: ShardSource, - { - assert_eq!(data.shard_count(), self.writers.len()); - - for (idx, (writer_opt, err)) in self.writers.iter_mut().zip(self.errs.iter_mut()).enumerate() { - if err.is_some() { - continue; - } - Self::write_shard_inline(writer_opt, err, data.shard(idx)); - } - - let nil_count = self.errs.iter().filter(|&e| e.is_none()).count(); - if nil_count >= self.write_quorum { - return Ok(()); - } - - if let Some(write_err) = reduce_write_quorum_errs(&self.errs, OBJECT_OP_IGNORED_ERRS, self.write_quorum) { - return Err(std::io::Error::other(format!( - "Failed to write inline data: {} (offline-disks={}/{})", - write_err, - count_errs(&self.errs, &Error::DiskNotFound), - self.writers.len() - ))); - } - - Err(std::io::Error::other(format!( - "Failed to write inline data: (offline-disks={}/{}): {}", - count_errs(&self.errs, &Error::DiskNotFound), - self.writers.len(), - self.errs - .iter() - .map(|e| e.as_ref().map_or("".to_string(), |e| e.to_string())) - .collect::>() - .join(", ") - ))) - } - async fn shutdown_writer(writer_opt: &mut Option, err: &mut Option) { match writer_opt { Some(writer) => match writer.shutdown().await { @@ -351,23 +129,6 @@ impl<'a> MultiWriter<'a> { } } - fn shutdown_writer_inline(writer_opt: &mut Option, err: &mut Option) { - match writer_opt { - Some(writer) => match writer.shutdown_inline_sync() { - Ok(()) => { - *err = None; - } - Err(e) => { - *err = Some(Error::from(e)); - *writer_opt = None; - } - }, - None => { - *err = Some(Error::DiskNotFound); - } - } - } - pub async fn shutdown(&mut self) -> std::io::Result<()> { { let mut futures = FuturesUnordered::new(); @@ -412,53 +173,80 @@ impl<'a> MultiWriter<'a> { .join(", ") ))) } - - pub fn shutdown_inline(&mut self) -> std::io::Result<()> { - for (writer_opt, err) in self.writers.iter_mut().zip(self.errs.iter_mut()) { - if err.is_some() { - continue; - } - Self::shutdown_writer_inline(writer_opt, err); - } - - let nil_count = self.errs.iter().filter(|&e| e.is_none()).count(); - if nil_count >= self.write_quorum { - return Ok(()); - } - - if let Some(write_err) = reduce_write_quorum_errs(&self.errs, OBJECT_OP_IGNORED_ERRS, self.write_quorum) { - return Err(std::io::Error::other(format!( - "Failed to shutdown inline writers: {} (offline-disks={}/{})", - write_err, - count_errs(&self.errs, &Error::DiskNotFound), - self.writers.len() - ))); - } - - Err(std::io::Error::other(format!( - "Failed to shutdown inline writers: (offline-disks={}/{}): {}", - count_errs(&self.errs, &Error::DiskNotFound), - self.writers.len(), - self.errs - .iter() - .map(|e| e.as_ref().map_or("".to_string(), |e| e.to_string())) - .collect::>() - .join(", ") - ))) - } } impl Erasure { pub async fn encode( self: Arc, - reader: R, + mut reader: R, writers: &mut [Option], quorum: usize, ) -> std::io::Result<(R, usize)> where - R: AsyncRead + BlockReadable + Send + Sync + Unpin + 'static, + R: AsyncRead + Send + Sync + Unpin + 'static, { - ErasureWritePipeline::new(self, quorum).run(reader, writers).await + let (tx, mut rx) = mpsc::channel::>(8); + + let task = tokio::spawn(async move { + let block_size = self.block_size; + let mut total = 0; + let mut buf = vec![0u8; block_size]; + loop { + match rustfs_utils::read_full(&mut reader, &mut buf).await { + Ok(n) if n > 0 => { + total += n; + let res = self.encode_data(&buf[..n])?; + if let Err(err) = tx.send(res).await { + return Err(std::io::Error::other(format!("Failed to send encoded data : {err}"))); + } + } + Ok(_) => { + break; + } + Err(e) if e.kind() == std::io::ErrorKind::UnexpectedEof => { + // Check if the inner error is a checksum mismatch - if so, propagate it + if let Some(inner) = e.get_ref() + && rustfs_rio::is_checksum_mismatch(inner) + { + return Err(std::io::Error::new(std::io::ErrorKind::InvalidData, e.to_string())); + } + break; + } + Err(e) => { + return Err(e); + } + } + } + + Ok((reader, total)) + }); + + let mut writers = MultiWriter::new(writers, quorum); + + let mut write_err = None; + + while let Some(block) = rx.recv().await { + if block.is_empty() { + break; + } + if let Err(err) = writers.write(block).await { + write_err = Some(err); + break; + } + } + + if let Some(err) = write_err { + task.abort(); + let _ = task.await; + if let Err(shutdown_err) = writers.shutdown().await { + error!("failed to shutdown erasure writers after write error: {:?}", shutdown_err); + } + return Err(err); + } + + let (reader, total) = task.await??; + writers.shutdown().await?; + Ok((reader, total)) } } @@ -467,7 +255,6 @@ mod tests { use super::*; use crate::erasure_coding::{BitrotWriterWrapper, CustomWriter}; use rustfs_utils::HashAlgorithm; - use std::io::Cursor; use std::pin::Pin; use std::sync::{Arc, Mutex}; use std::task::{Context, Poll}; @@ -523,28 +310,4 @@ mod tests { assert_eq!(written, b"small payload".len()); assert!(!committed.lock().unwrap().is_empty()); } - - #[tokio::test] - async fn block_assembler_splits_input_into_erasure_blocks() { - let reader = tokio::io::BufReader::new(Cursor::new(b"abcdefghijkl".to_vec())); - let mut assembler = BlockAssembler::new(reader, 4); - - assert_eq!(assembler.next_block().await.unwrap(), Some(b"abcd".to_vec())); - assert_eq!(assembler.next_block().await.unwrap(), Some(b"efgh".to_vec())); - assert_eq!(assembler.next_block().await.unwrap(), Some(b"ijkl".to_vec())); - assert_eq!(assembler.next_block().await.unwrap(), None); - assert_eq!(assembler.total_bytes(), 12); - } - - #[tokio::test] - async fn erasure_chunk_encoder_produces_full_shard_block() { - let erasure = Arc::new(Erasure::new(2, 1, 4)); - let encoder = ErasureChunkEncoder::new(erasure.clone()).await; - let block = encoder.encode_block(b"abcd").await.unwrap(); - - assert_eq!(block.shard_count(), 3); - assert_eq!(block.shard(0).len(), erasure.shard_size()); - - encoder.release(block).await; - } } diff --git a/crates/ecstore/src/erasure_coding/erasure.rs b/crates/ecstore/src/erasure_coding/erasure.rs index 63a1350d2..c83f1e028 100644 --- a/crates/ecstore/src/erasure_coding/erasure.rs +++ b/crates/ecstore/src/erasure_coding/erasure.rs @@ -19,131 +19,12 @@ use bytes::{Bytes, BytesMut}; use reed_solomon_erasure::galois_8::ReedSolomon; use reed_solomon_simd; -use rustfs_rio::BlockReadable; use smallvec::SmallVec; use std::io; -use std::sync::Arc; use tokio::io::AsyncRead; -use tokio::sync::Mutex; use tracing::warn; use uuid::Uuid; -pub(crate) struct EncodeBlockBuffer { - buf: Vec, -} - -impl EncodeBlockBuffer { - pub(crate) fn new(block_size: usize) -> Self { - Self { - buf: vec![0u8; block_size], - } - } - - pub(crate) async fn read_from(&mut self, reader: &mut R) -> io::Result - where - R: AsyncRead + Send + Sync + Unpin, - { - rustfs_utils::read_full(&mut *reader, &mut self.buf).await - } - - pub(crate) async fn read_from_block(&mut self, reader: &mut R) -> io::Result - where - R: BlockReadable + Send + Sync + Unpin, - { - reader.read_block(&mut self.buf).await - } - - pub(crate) fn filled(&self, len: usize) -> &[u8] { - &self.buf[..len] - } -} - -pub struct EncodedShardBlock { - data: Bytes, - shard_size: usize, - shard_count: usize, -} - -impl EncodedShardBlock { - pub(crate) fn new(data: Bytes, shard_size: usize, shard_count: usize) -> Self { - Self { - data, - shard_size, - shard_count, - } - } - - pub fn shard_count(&self) -> usize { - self.shard_count - } - - pub fn len(&self) -> usize { - self.shard_count - } - - pub fn is_empty(&self) -> bool { - self.shard_count == 0 - } - - pub fn shard(&self, idx: usize) -> Bytes { - let start = idx * self.shard_size; - let end = start + self.shard_size; - self.data.slice(start..end) - } - - pub fn iter(&self) -> impl Iterator + '_ { - (0..self.shard_count).map(|idx| self.shard(idx)) - } - - pub fn into_vec(self) -> Vec { - (0..self.shard_count).map(|idx| self.shard(idx)).collect() - } - - pub fn into_reusable_buffer(self) -> BytesMut { - match self.data.try_into_mut() { - Ok(mut buf) => { - buf.clear(); - buf - } - Err(data) => BytesMut::with_capacity(data.len()), - } - } -} - -#[derive(Clone)] -pub(crate) struct EncodedShardBufferPool { - capacity: usize, - free: Arc>>, -} - -impl EncodedShardBufferPool { - pub(crate) async fn with_prefill(capacity: usize, initial: usize) -> Self { - let mut free = Vec::with_capacity(initial); - for _ in 0..initial { - free.push(BytesMut::with_capacity(capacity)); - } - - Self { - capacity, - free: Arc::new(Mutex::new(free)), - } - } - - pub(crate) async fn acquire(&self) -> BytesMut { - let mut free = self.free.lock().await; - free.pop().unwrap_or_else(|| BytesMut::with_capacity(self.capacity)) - } - - pub(crate) async fn release(&self, block: EncodedShardBlock) { - let mut free = self.free.lock().await; - let mut buf = block.into_reusable_buffer(); - if buf.capacity() < self.capacity { - buf.reserve(self.capacity - buf.capacity()); - } - free.push(buf); - } -} - /// Legacy calc_shard_size formula: (block_size.div_ceil(data_shards) + 1) & !1 /// Matches main branch and filemeta::ErasureInfo for old-version files. pub fn calc_shard_size_legacy(block_size: usize, data_shards: usize) -> usize { @@ -470,25 +351,6 @@ impl Erasure { /// A vector of encoded shards as `Bytes`. #[tracing::instrument(level = "debug", skip_all, fields(data_len=data.len()))] pub fn encode_data(&self, data: &[u8]) -> io::Result> { - Ok(self.encode_data_block(data)?.into_vec()) - } - - /// Encode one logical block into an `EncodedShardBlock` using a caller-provided backing buffer. - /// - /// This is the explicit reuse-oriented variant for non-hot paths that want to - /// thread a reusable `BytesMut` across multiple encode calls. - #[tracing::instrument(level = "debug", skip_all, fields(data_len=data.len()))] - pub fn encode_data_with_buffer(&self, data: &[u8], data_buffer: BytesMut) -> io::Result { - self.encode_data_block_with_buffer(data, data_buffer) - } - - #[tracing::instrument(level = "debug", skip_all, fields(data_len=data.len()))] - pub(crate) fn encode_data_block(&self, data: &[u8]) -> io::Result { - self.encode_data_block_with_buffer(data, BytesMut::with_capacity(self.shard_size() * self.total_shard_count())) - } - - #[tracing::instrument(level = "debug", skip_all, fields(data_len=data.len()))] - pub(crate) fn encode_data_block_with_buffer(&self, data: &[u8], mut data_buffer: BytesMut) -> io::Result { let shard_size_fn = if self.uses_legacy { calc_shard_size_legacy } else { @@ -497,10 +359,7 @@ impl Erasure { let per_shard_size = shard_size_fn(data.len(), self.data_shards); let need_total_size = per_shard_size * self.total_shard_count(); - data_buffer.clear(); - if data_buffer.capacity() < need_total_size { - data_buffer.reserve(need_total_size - data_buffer.capacity()); - } + let mut data_buffer = BytesMut::with_capacity(need_total_size); data_buffer.extend_from_slice(data); data_buffer.resize(need_total_size, 0u8); @@ -523,7 +382,14 @@ impl Erasure { } // Zero-copy split, all shards reference data_buffer - Ok(EncodedShardBlock::new(data_buffer.freeze(), per_shard_size, self.total_shard_count())) + let mut data_buffer = data_buffer.freeze(); + let mut shards = Vec::with_capacity(self.total_shard_count()); + for _ in 0..self.total_shard_count() { + let shard = data_buffer.split_to(per_shard_size); + shards.push(shard); + } + + Ok(shards) } /// Decode and reconstruct missing shards in-place. @@ -612,8 +478,8 @@ impl Erasure { /// /// # Arguments /// * `reader` - An async reader implementing AsyncRead + Send + Sync + Unpin - /// * `mut on_block` - Async callback that receives encoded blocks and returns the block for reuse - /// * `F` - Callback type: FnMut(Result) -> Future, E>> + Send + /// * `mut on_block` - Async callback that receives encoded blocks and returns a Result + /// * `F` - Callback type: FnMut(Result, std::io::Error>) -> Future> + Send /// * `Fut` - Future type returned by the callback /// * `E` - Error type returned by the callback /// * `R` - Reader type implementing AsyncRead + Send + Sync + Unpin @@ -630,24 +496,19 @@ impl Erasure { ) -> Result where R: AsyncRead + Send + Sync + Unpin, - F: FnMut(std::io::Result) -> Fut + Send, - Fut: std::future::Future, E>> + Send, + F: FnMut(std::io::Result>) -> Fut + Send, + Fut: std::future::Future> + Send, { let block_size = self.block_size; let mut total = 0; - let mut block_buffer = EncodeBlockBuffer::new(block_size); - let reusable_capacity = self.shard_size() * self.total_shard_count(); - let buffer_pool = EncodedShardBufferPool::with_prefill(reusable_capacity, 1).await; + let mut buf = vec![0u8; block_size]; loop { - match block_buffer.read_from(&mut *reader).await { + match rustfs_utils::read_full(&mut *reader, &mut buf).await { Ok(n) if n > 0 => { warn!("encode_stream_callback_async read n={}", n); total += n; - let reusable_buffer = buffer_pool.acquire().await; - let res = self.encode_data_block_with_buffer(block_buffer.filled(n), reusable_buffer); - if let Some(block) = on_block(res).await? { - buffer_pool.release(block).await; - } + let res = self.encode_data(&buf[..n]); + on_block(res).await? } Ok(_) => { warn!("encode_stream_callback_async read unexpected ok"); @@ -659,7 +520,7 @@ impl Erasure { } Err(e) => { warn!("encode_stream_callback_async read error={:?}", e); - let _ = on_block(Err(e)).await?; + on_block(Err(e)).await?; break; } } @@ -885,8 +746,8 @@ mod tests { let tx = tx.clone(); async move { let shards = res.unwrap(); - tx.send(shards.iter().collect()).await.unwrap(); - Ok(Some(shards)) + tx.send(shards).await.unwrap(); + Ok(()) } }) .await @@ -898,36 +759,6 @@ mod tests { assert_eq!(collected_shards.len(), data_shards + parity_shards); } - #[test] - fn test_encode_data_with_buffer_supports_explicit_reuse() { - let erasure = Erasure::new(4, 2, 1024); - let reusable_capacity = erasure.shard_size() * erasure.total_shard_count(); - - let first_data = b"explicit reusable buffer path".repeat(32); - let first_block = erasure - .encode_data_with_buffer(&first_data, BytesMut::with_capacity(reusable_capacity)) - .expect("first encode should succeed"); - let reusable_buffer = first_block.into_reusable_buffer(); - assert!(reusable_buffer.capacity() >= reusable_capacity); - - let second_data = b"second encode through same reusable buffer".repeat(24); - let second_block = erasure - .encode_data_with_buffer(&second_data, reusable_buffer) - .expect("second encode should succeed"); - - let mut shards_opt: Vec>> = second_block.iter().map(|shard| Some(shard.to_vec())).collect(); - shards_opt[1] = None; - shards_opt[5] = None; - erasure.decode_data(&mut shards_opt).expect("decode should succeed"); - - let mut recovered = Vec::new(); - for shard in shards_opt.iter().take(erasure.data_shards) { - recovered.extend_from_slice(shard.as_ref().expect("data shard should exist after decode")); - } - recovered.truncate(second_data.len()); - assert_eq!(&recovered, &second_data); - } - #[tokio::test] async fn test_encode_stream_callback_async_channel_decode() { use std::io::Cursor; @@ -954,8 +785,8 @@ mod tests { let tx = tx.clone(); async move { let shards = res.unwrap(); - tx.send(shards.iter().collect()).await.unwrap(); - Ok(Some(shards)) + tx.send(shards).await.unwrap(); + Ok(()) } }) .await @@ -968,8 +799,8 @@ mod tests { // Test decode using the old API that operates in-place let mut decode_input: Vec>> = vec![None; data_shards + parity_shards]; - for (i, shard) in shards.iter().enumerate().take(data_shards) { - decode_input[i] = Some(shard.to_vec()); + for i in 0..data_shards { + decode_input[i] = Some(shards[i].to_vec()); } erasure.decode_data(&mut decode_input).unwrap(); @@ -1366,8 +1197,8 @@ mod tests { let tx = tx.clone(); async move { let shards = res.unwrap(); - tx.send(shards.iter().collect()).await.unwrap(); - Ok(Some(shards)) + tx.send(shards).await.unwrap(); + Ok(()) } }) .await @@ -1401,63 +1232,5 @@ mod tests { recovered.truncate(data_clone.len()); assert_eq!(&recovered, &data_clone); } - - #[tokio::test] - #[ignore] - async fn stress_simd_stream_callback_reuses_backing_buffers_across_many_blocks() { - use std::io::Cursor; - use std::sync::Arc; - use std::sync::atomic::{AtomicUsize, Ordering}; - use tokio::sync::Mutex; - - let data_shards = 4; - let parity_shards = 2; - let block_size = 1024; - let erasure = Arc::new(Erasure::new(data_shards, parity_shards, block_size)); - - let sample = - b"SIMD stress callback test payload that intentionally spans many blocks to exercise reusable backing buffers."; - let data = sample.repeat((4 * 1024 * 1024 / sample.len()).max(1)); - let data_clone = data.clone(); - let mut reader = Cursor::new(data); - - let recovered = Arc::new(Mutex::new(Vec::with_capacity(data_clone.len()))); - let block_count = Arc::new(AtomicUsize::new(0)); - let erasure_for_callback = erasure.clone(); - let recovered_for_callback = recovered.clone(); - let block_count_for_callback = block_count.clone(); - - erasure - .clone() - .encode_stream_callback_async::<_, _, (), _>(&mut reader, move |res| { - let erasure_for_callback = erasure_for_callback.clone(); - let recovered_for_callback = recovered_for_callback.clone(); - let block_count_for_callback = block_count_for_callback.clone(); - async move { - let shards = res.unwrap(); - block_count_for_callback.fetch_add(1, Ordering::Relaxed); - - let mut shards_opt: Vec>> = shards.iter().map(|b| Some(b.to_vec())).collect(); - shards_opt[1] = None; - shards_opt[5] = None; - erasure_for_callback.decode_data(&mut shards_opt).unwrap(); - - let mut recovered = recovered_for_callback.lock().await; - for shard in shards_opt.iter().take(data_shards) { - recovered.extend_from_slice(shard.as_ref().unwrap()); - } - - Ok(Some(shards)) - } - }) - .await - .unwrap(); - - assert!(block_count.load(Ordering::Relaxed) > 1024); - - let mut recovered = recovered.lock().await; - recovered.truncate(data_clone.len()); - assert_eq!(&*recovered, &data_clone); - } } } diff --git a/crates/ecstore/src/erasure_coding/heal.rs b/crates/ecstore/src/erasure_coding/heal.rs index 66b6d0802..422ae7dac 100644 --- a/crates/ecstore/src/erasure_coding/heal.rs +++ b/crates/ecstore/src/erasure_coding/heal.rs @@ -77,7 +77,7 @@ impl super::Erasure { let available_writers = writers.iter().filter(|w| w.is_some()).count(); let write_quorum = available_writers.max(1); // At least 1 writer must succeed let mut writers = MultiWriter::new(writers, write_quorum); - writers.write(&shards).await?; + writers.write(shards).await?; } Ok(()) diff --git a/crates/ecstore/src/rpc/remote_disk.rs b/crates/ecstore/src/rpc/remote_disk.rs index 781091ac2..dd7945f60 100644 --- a/crates/ecstore/src/rpc/remote_disk.rs +++ b/crates/ecstore/src/rpc/remote_disk.rs @@ -30,10 +30,8 @@ use crate::{ }; use bytes::Bytes; use futures::lock::Mutex; -use futures_util::StreamExt; use http::{HeaderMap, HeaderValue, Method, header::CONTENT_TYPE}; use rustfs_filemeta::{FileInfo, ObjectPartInfo, RawFileInfo}; -use rustfs_io_core::{BoxChunkStream, IoChunk}; use rustfs_protos::proto_gen::node_service::RenamePartRequest; use rustfs_protos::proto_gen::node_service::{ CheckPartsRequest, DeletePathsRequest, DeleteRequest, DeleteVersionRequest, DeleteVersionsRequest, DeleteVolumeRequest, @@ -42,7 +40,7 @@ use rustfs_protos::proto_gen::node_service::{ RenameFileRequest, StatVolumeRequest, UpdateMetadataRequest, VerifyFileRequest, WriteAllRequest, WriteMetadataRequest, node_service_client::NodeServiceClient, }; -use rustfs_rio::{HttpReader, HttpWriter, open_http_byte_stream}; +use rustfs_rio::{HttpReader, HttpWriter}; use serde::{Serialize, de::DeserializeOwned}; use std::{ io::Cursor, @@ -1073,33 +1071,6 @@ impl DiskAPI for RemoteDisk { Ok(Bytes::from(buffer)) } - #[tracing::instrument(level = "debug", skip(self))] - async fn read_file_chunks(&self, volume: &str, path: &str, offset: usize, length: usize) -> Result { - if self.health.is_faulty() { - return Err(DiskError::FaultyDisk); - } - let disk = self.disk_ref().await; - - let url = format!( - "{}/rustfs/rpc/read_file_stream?disk={}&volume={}&path={}&offset={}&length={}", - self.endpoint.grid_host(), - urlencoding::encode(&disk), - urlencoding::encode(volume), - urlencoding::encode(path), - offset, - length - ); - - let mut headers = HeaderMap::new(); - headers.insert(CONTENT_TYPE, HeaderValue::from_static("application/json")); - build_auth_headers(&url, &Method::GET, &mut headers); - - let stream = open_http_byte_stream(url, Method::GET, headers, None) - .await? - .map(|result| result.map(IoChunk::Shared)); - Ok(Box::pin(stream)) - } - #[tracing::instrument(level = "debug", skip(self))] async fn append_file(&self, volume: &str, path: &str) -> Result { info!("append_file {}/{}", volume, path); diff --git a/crates/ecstore/src/set_disk/read.rs b/crates/ecstore/src/set_disk/read.rs index 1c5f240c6..1efc663d0 100644 --- a/crates/ecstore/src/set_disk/read.rs +++ b/crates/ecstore/src/set_disk/read.rs @@ -13,205 +13,7 @@ // limitations under the License. use super::*; -use crate::erasure_coding::{calc_shard_size, calc_shard_size_legacy}; -use bytes::BytesMut; -use futures_util::Stream; use rustfs_config::{DEFAULT_OBJECT_ZERO_COPY_ENABLE, ENV_OBJECT_ZERO_COPY_ENABLE}; -use rustfs_io_core::IoChunk; -use std::io; -use std::pin::Pin; -use std::sync::Mutex; -use std::task::{Context, Poll}; -use tokio::sync::mpsc::{UnboundedReceiver, UnboundedSender, unbounded_channel}; -use tokio_util::io::ReaderStream; - -struct ChannelChunkStream { - receiver: Mutex>>, -} - -impl ChannelChunkStream { - fn new(receiver: UnboundedReceiver>) -> Self { - Self { - receiver: Mutex::new(receiver), - } - } -} - -impl Stream for ChannelChunkStream { - type Item = io::Result; - - fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { - let mut receiver = self.receiver.lock().unwrap(); - receiver.poll_recv(cx) - } -} - -struct ChannelChunkWriter { - sender: UnboundedSender>, - buffer: BytesMut, - chunk_size: usize, -} - -impl ChannelChunkWriter { - fn new(sender: UnboundedSender>, chunk_size: usize) -> Self { - Self { - sender, - buffer: BytesMut::with_capacity(chunk_size.max(1)), - chunk_size: chunk_size.max(1), - } - } - - fn push_buffer(&mut self) -> io::Result<()> { - if self.buffer.is_empty() { - return Ok(()); - } - - let bytes = self.buffer.split().freeze(); - self.sender - .send(Ok(IoChunk::Shared(bytes))) - .map_err(|_| io::Error::new(io::ErrorKind::BrokenPipe, "chunk stream receiver dropped")) - } - - fn finish(&mut self) -> io::Result<()> { - self.push_buffer() - } - - fn send_error(&self, err: io::Error) { - let _ = self.sender.send(Err(err)); - } -} - -impl AsyncWrite for ChannelChunkWriter { - fn poll_write(mut self: Pin<&mut Self>, _cx: &mut Context<'_>, buf: &[u8]) -> Poll> { - self.buffer.extend_from_slice(buf); - let chunk_size = self.chunk_size; - while self.buffer.len() >= chunk_size { - let chunk = self.buffer.split_to(chunk_size).freeze(); - if self.sender.send(Ok(IoChunk::Shared(chunk))).is_err() { - return Poll::Ready(Err(io::Error::new(io::ErrorKind::BrokenPipe, "chunk stream receiver dropped"))); - } - } - - Poll::Ready(Ok(buf.len())) - } - - fn poll_flush(mut self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll> { - Poll::Ready(self.push_buffer()) - } - - fn poll_shutdown(mut self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll> { - Poll::Ready(self.finish()) - } -} - -fn multipart_logical_part_size(fi: &FileInfo, part_index: usize) -> usize { - let part = &fi.parts[part_index]; - if part.actual_size > 0 { - part.actual_size as usize - } else { - part.size - } -} - -fn multipart_logical_total_size(fi: &FileInfo) -> usize { - if fi.parts.is_empty() { - return fi.size.max(0) as usize; - } - - fi.parts - .iter() - .map(|part| { - if part.actual_size > 0 { - part.actual_size as usize - } else { - part.size - } - }) - .sum() -} - -fn multipart_stored_part_size(fi: &FileInfo, part_index: usize) -> usize { - fi.parts[part_index].size -} - -fn multipart_stored_total_size(fi: &FileInfo) -> usize { - if fi.parts.is_empty() { - return fi.size.max(0) as usize; - } - - fi.parts.iter().map(|part| part.size).sum() -} - -fn multipart_to_logical_part_offset(fi: &FileInfo, offset: usize) -> Result<(usize, usize)> { - if offset == 0 { - return Ok((0, 0)); - } - - let mut part_offset = offset; - for (i, _) in fi.parts.iter().enumerate() { - let logical_part_size = multipart_logical_part_size(fi, i); - if part_offset < logical_part_size { - return Ok((i, part_offset)); - } - - part_offset -= logical_part_size; - } - - Err(Error::other("part not found")) -} - -fn multipart_to_stored_part_offset(fi: &FileInfo, offset: usize) -> Result<(usize, usize)> { - if offset == 0 { - return Ok((0, 0)); - } - - let mut part_offset = offset; - for (i, part) in fi.parts.iter().enumerate() { - if part_offset < part.size { - return Ok((i, part_offset)); - } - - part_offset -= part.size; - } - - Err(Error::other("part not found")) -} - -fn block_window( - offset: usize, - length: usize, - block_size: usize, - block_index: usize, - start_block: usize, - end_block: usize, -) -> (usize, usize) { - let end_remainder = offset.saturating_add(length) % block_size; - if start_block == end_block { - (offset % block_size, length) - } else if block_index == start_block { - (offset % block_size, block_size - (offset % block_size)) - } else if block_index == end_block { - (0, if end_remainder == 0 { block_size } else { end_remainder }) - } else { - (0, block_size) - } -} - -fn direct_block_shard_size( - total_size: usize, - block_size: usize, - data_shards: usize, - block_index: usize, - uses_legacy: bool, -) -> usize { - let block_start = block_index.saturating_mul(block_size); - let logical_block_size = total_size.saturating_sub(block_start).min(block_size); - if uses_legacy { - calc_shard_size_legacy(logical_block_size, data_shards) - } else { - calc_shard_size(logical_block_size, data_shards) - } -} impl SetDisks { pub(super) async fn read_parts( @@ -781,13 +583,12 @@ impl SetDisks { debug!(bucket, object, requested_length = length, offset, "get_object_with_fileinfo start"); let (disks, files) = Self::shuffle_disks_and_parts_metadata_by_index(disks, &files, &fi); - let logical_total_size = multipart_logical_total_size(&fi); - let use_stored_part_sizes = length > logical_total_size as i64; - let total_size = if use_stored_part_sizes { - multipart_stored_total_size(&fi) - } else { - logical_total_size - }; + 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); + return Err(Error::other("offset out of range")); + } let length = if length < 0 { total_size - offset } else { length as usize }; @@ -796,27 +597,19 @@ impl SetDisks { return Err(Error::other("offset out of range")); }; - if offset > total_size || end_offset_exclusive > total_size { + if end_offset_exclusive > total_size { error!("get_object_with_fileinfo offset out of range: {}, total_size: {}", offset, total_size); return Err(Error::other("offset out of range")); } - let (part_index, mut part_offset) = if use_stored_part_sizes { - multipart_to_stored_part_offset(&fi, offset)? - } else { - multipart_to_logical_part_offset(&fi, offset)? - }; + let (part_index, mut part_offset) = fi.to_part_offset(offset)?; let mut end_offset = offset; if length > 0 { - end_offset = end_offset_exclusive - 1; + end_offset += length - 1 } - let (last_part_index, last_part_relative_offset) = if use_stored_part_sizes { - multipart_to_stored_part_offset(&fi, end_offset)? - } else { - multipart_to_logical_part_offset(&fi, end_offset)? - }; + let (last_part_index, last_part_relative_offset) = fi.to_part_offset(end_offset)?; debug!( bucket, @@ -848,11 +641,7 @@ impl SetDisks { } let part_number = fi.parts[current_part].number; - let part_size = if use_stored_part_sizes { - multipart_stored_part_size(&fi, current_part) - } else { - multipart_logical_part_size(&fi, current_part) - }; + let part_size = fi.parts[current_part].size; let mut part_length = part_size - part_offset; if part_length > (length - total_read) { part_length = length - total_read @@ -891,7 +680,6 @@ impl SetDisks { let mut readers = Vec::with_capacity(disks.len()); let mut errors = Vec::with_capacity(disks.len()); - let shard_length = till_offset.saturating_sub(read_offset); for (idx, disk_op) in disks.iter().enumerate() { match create_bitrot_reader( files[idx].data.as_deref(), @@ -899,7 +687,7 @@ impl SetDisks { bucket, &format!("{}/{}/part.{}", object, files[idx].data_dir.unwrap_or_default(), part_number), read_offset, - shard_length, + till_offset.saturating_sub(read_offset), erasure.shard_size(), checksum_algo.clone(), skip_verify_bitrot, diff --git a/crates/io-metrics/src/lib.rs b/crates/io-metrics/src/lib.rs index d2ff0128e..bdfa0020d 100644 --- a/crates/io-metrics/src/lib.rs +++ b/crates/io-metrics/src/lib.rs @@ -855,6 +855,36 @@ pub fn record_io_latency_p99(latency_ms: f64) { gauge!("rustfs.io.latency.p99.ms").set(latency_ms); } +// ============================================================================ +// Zero-Copy I/O Metrics +// ============================================================================ + +/// Record a successful zero-copy read operation (e.g., mmap). +/// +/// # Arguments +/// +/// * `size_bytes` - Size of the data read in bytes +/// * `duration_ms` - Time taken for the read operation in milliseconds +#[inline(always)] +pub fn record_zero_copy_read(size_bytes: usize, duration_ms: f64) { + counter!("rustfs.zero_copy.reads.total").increment(1); + histogram!("rustfs.zero_copy.read.size.bytes").record(size_bytes as f64); + histogram!("rustfs.zero_copy.read.duration.ms").record(duration_ms); +} + +/// Record a fallback from zero-copy to regular read. +/// +/// This happens when zero-copy read fails (e.g., mmap not available, +/// file too large, etc.) and the system falls back to regular I/O. +/// +/// # Arguments +/// +/// * `reason` - Reason for the fallback (e.g., "mmap_unavailable", "file_too_large") +#[inline(always)] +pub fn record_zero_copy_fallback(reason: &str) { + counter!("rustfs.zero_copy.fallback.total", "reason" => reason.to_string()).increment(1); +} + #[cfg(test)] mod tests { use super::*; diff --git a/crates/object-io/src/put.rs b/crates/object-io/src/put.rs index 7085a6468..15266ee6a 100644 --- a/crates/object-io/src/put.rs +++ b/crates/object-io/src/put.rs @@ -17,9 +17,7 @@ use futures_util::{Stream, StreamExt}; use http::HeaderMap; use rustfs_ecstore::compress::{MIN_COMPRESSIBLE_SIZE, is_compressible}; use rustfs_ecstore::store_api::ObjectOptions; -use rustfs_rio::{ - BlockReadable, BoxReadBlockFuture, Checksum, EtagResolvable, HashReader, HashReaderDetector, Reader, TryGetIndex, WarpReader, -}; +use rustfs_rio::{Checksum, EtagResolvable, HashReader, HashReaderDetector, Reader, TryGetIndex, WarpReader}; use rustfs_utils::http::AMZ_SERVER_SIDE_ENCRYPTION_CUSTOMER_ALGORITHM; use rustfs_utils::http::headers::{ AMZ_DECODED_CONTENT_LENGTH, AMZ_MINIO_SNOWBALL_IGNORE_DIRS, AMZ_MINIO_SNOWBALL_IGNORE_ERRORS, AMZ_MINIO_SNOWBALL_PREFIX, @@ -244,62 +242,6 @@ impl PutObjectReducedCopyReader { buf.advance(copied); copied } - - async fn read_into_slice(&mut self, buf: &mut [u8]) -> std::io::Result - where - S: Stream> + Unpin, - B: Buf + Unpin, - E: std::fmt::Display, - { - if let Some(err) = self.pending_error.take() { - return Err(err); - } - - if buf.is_empty() { - return Ok(0); - } - - let mut copied = 0; - - loop { - if copied == buf.len() { - return Ok(copied); - } - - if let Some(chunk) = self.current_chunk.as_mut() { - if !chunk.has_remaining() { - self.current_chunk = None; - continue; - } - - copied += Self::copy_chunk_into_slice(chunk, &mut buf[copied..]); - if chunk.has_remaining() || copied == buf.len() { - return Ok(copied); - } - - self.current_chunk = None; - continue; - } - - match self.body.next().await { - Some(Ok(chunk)) => { - if chunk.remaining() == 0 { - continue; - } - self.current_chunk = Some(chunk); - } - Some(Err(err)) => { - let err = std::io::Error::other(err.to_string()); - if copied > 0 { - self.pending_error = Some(err); - return Ok(copied); - } - return Err(err); - } - None => return Ok(copied), - } - } - } } impl AsyncRead for PutObjectReducedCopyReader @@ -371,17 +313,6 @@ where } } -impl BlockReadable for PutObjectReducedCopyReader -where - S: Stream> + Unpin + Send + Sync, - B: Buf + Unpin + Send + Sync, - E: std::fmt::Display + Send + Sync, -{ - fn read_block<'a>(&'a mut self, buf: &'a mut [u8]) -> BoxReadBlockFuture<'a> { - Box::pin(async move { self.read_into_slice(buf).await }) - } -} - impl EtagResolvable for PutObjectReducedCopyReader {} impl HashReaderDetector for PutObjectReducedCopyReader {} @@ -1444,62 +1375,6 @@ mod tests { assert!(err.to_string().contains("boom")); } - #[tokio::test] - async fn build_put_object_reduced_copy_reader_supports_direct_block_reads() { - let stream = futures_util::stream::iter([ - Ok::(Bytes::from_static(b"ab")), - Ok::(Bytes::from_static(b"cd")), - Ok::(Bytes::from_static(b"ef")), - ]); - - let mut reader = build_put_object_reduced_copy_reader(PutObjectReducedCopyIngress::from_stream( - stream, - PutObjectCompatIngressPlan { - kind: PutObjectCompatIngressKind::BufferedStreamCompat, - buffer_size: 16, - }, - )); - - let mut first = [0_u8; 4]; - let mut second = [0_u8; 4]; - assert_eq!(reader.read_block(&mut first).await.unwrap(), 4); - assert_eq!(&first, b"abcd"); - assert_eq!(reader.read_block(&mut second).await.unwrap(), 2); - assert_eq!(&second[..2], b"ef"); - assert_eq!(reader.read_block(&mut second).await.unwrap(), 0); - } - - #[tokio::test] - async fn build_put_object_plain_hash_stage_reports_reduced_copy_candidate_boundary() { - let stream = futures_util::stream::iter([Ok::(Bytes::from_static(b"plain"))]); - let mut headers = HeaderMap::new(); - headers.insert("content-type", HeaderValue::from_static("application/octet-stream")); - let plan = plan_put_object_body(2 * 1024 * 1024, &headers, "large.bin", 32); - - let stage = build_put_object_plain_hash_stage( - build_put_object_ingress_source(stream, plan), - PutObjectLegacyHashValues::default(), - PutObjectLegacyHashStagePlan { - size: 5, - actual_size: 5, - apply_s3_checksum: false, - ignore_s3_checksum_value: false, - }, - &headers, - None, - ) - .unwrap(); - - assert_eq!(stage.ingress_kind, PutObjectIngressKind::ReducedCopyCandidate); - - let mut reader = stage.reader; - let mut buf = [0_u8; 8]; - let n = reader.read_block(&mut buf).await.unwrap(); - assert_eq!(n, 5); - assert_eq!(&buf[..n], b"plain"); - assert_eq!(reader.read_block(&mut buf).await.unwrap(), 0); - } - #[tokio::test] async fn build_put_object_plain_hash_stage_preserves_legacy_compat_boundary() { let stream = futures_util::stream::iter([Ok::(Bytes::from_static(b"legacy"))]); diff --git a/crates/rio/src/checksum.rs b/crates/rio/src/checksum.rs index b09916bf8..ef4b85a6d 100644 --- a/crates/rio/src/checksum.rs +++ b/crates/rio/src/checksum.rs @@ -1138,8 +1138,36 @@ fn crc64_combine(poly: u64, crc1: u64, crc2: u64, len2: i64) -> u64 { #[cfg(test)] mod tests { - use super::{AMZ_SDK_CHECKSUM_ALGORITHM, Checksum, ChecksumType, get_content_checksum_direct}; - use http::{HeaderMap, HeaderValue}; + use super::{Checksum, ChecksumType}; + use http::HeaderMap; + + #[test] + fn algorithm_from_headers_parses_amz_checksum_algorithm() { + let mut headers = HeaderMap::new(); + headers.insert("x-amz-checksum-algorithm", "SHA256".parse().expect("valid header value")); + assert_eq!(ChecksumType::algorithm_from_headers(&headers), Some("SHA256")); + } + + #[test] + fn algorithm_from_headers_falls_back_to_sdk_header() { + let mut headers = HeaderMap::new(); + headers.insert("x-amz-sdk-checksum-algorithm", "CRC32C".parse().expect("valid header value")); + assert_eq!(ChecksumType::algorithm_from_headers(&headers), Some("CRC32C")); + } + + #[test] + fn algorithm_from_headers_prefers_amz_over_sdk() { + let mut headers = HeaderMap::new(); + headers.insert("x-amz-checksum-algorithm", "SHA256".parse().expect("valid header value")); + headers.insert("x-amz-sdk-checksum-algorithm", "CRC32C".parse().expect("valid header value")); + assert_eq!(ChecksumType::algorithm_from_headers(&headers), Some("SHA256")); + } + + #[test] + fn algorithm_from_headers_returns_none_when_missing() { + let headers = HeaderMap::new(); + assert_eq!(ChecksumType::algorithm_from_headers(&headers), None); + } #[test] fn crc64_nvme_add_part_matches_full_object_checksum() { @@ -1194,24 +1222,4 @@ mod tests { assert_eq!(combined.encoded, expected.encoded); assert_eq!(combined.raw, expected.raw); } - - #[test] - fn checksum_type_from_header_supports_sdk_checksum_algorithm_header() { - let mut headers = HeaderMap::new(); - headers.insert(AMZ_SDK_CHECKSUM_ALGORITHM, HeaderValue::from_static("CRC32")); - - assert_eq!(ChecksumType::from_header(&headers), ChecksumType::CRC32); - } - - #[test] - fn get_content_checksum_direct_supports_sdk_checksum_algorithm_header() { - let mut headers = HeaderMap::new(); - headers.insert(AMZ_SDK_CHECKSUM_ALGORITHM, HeaderValue::from_static("CRC32")); - headers.insert("x-amz-checksum-crc32", HeaderValue::from_static("nct/nQ==")); - - let (checksum_type, checksum_value) = get_content_checksum_direct(&headers); - - assert_eq!(checksum_type, ChecksumType::CRC32); - assert_eq!(checksum_value, "nct/nQ=="); - } } diff --git a/crates/rio/src/compress_reader.rs b/crates/rio/src/compress_reader.rs index e0d961234..418373a89 100644 --- a/crates/rio/src/compress_reader.rs +++ b/crates/rio/src/compress_reader.rs @@ -13,7 +13,6 @@ // limitations under the License. use crate::compress_index::{Index, TryGetIndex}; -use crate::{BlockReadable, BoxReadBlockFuture, Reader}; use pin_project_lite::pin_project; use rustfs_utils::compress::{CompressionAlgorithm, compress_block, decompress_block}; use rustfs_utils::{put_uvarint, uvarint}; @@ -86,31 +85,6 @@ where read_buffer: vec![0u8; block_size], } } - - fn copy_buffered(&mut self, buf: &mut [u8]) -> usize { - if self.pos >= self.buffer.len() || buf.is_empty() { - return 0; - } - - let to_copy = min(buf.len(), self.buffer.len() - self.pos); - buf[..to_copy].copy_from_slice(&self.buffer[self.pos..self.pos + to_copy]); - self.pos += to_copy; - if self.pos == self.buffer.len() { - self.buffer.clear(); - self.pos = 0; - } - to_copy - } - - fn queue_compressed_block(&mut self, uncompressed_data: &[u8]) -> io::Result<()> { - let out = build_compressed_block(uncompressed_data, self.compression_algorithm); - self.written += out.len(); - self.uncomp_written += uncompressed_data.len(); - self.index.add(self.written as i64, self.uncomp_written as i64)?; - self.buffer = out; - self.pos = 0; - Ok(()) - } } impl TryGetIndex for CompressReader { @@ -196,50 +170,6 @@ where delegate_reader_capabilities_generic_no_index!(CompressReader, inner); -impl BlockReadable for CompressReader -where - R: Reader, -{ - fn read_block<'a>(&'a mut self, buf: &'a mut [u8]) -> BoxReadBlockFuture<'a> { - Box::pin(async move { - if buf.is_empty() { - return Ok(0); - } - - let mut written = self.copy_buffered(buf); - while written < buf.len() { - if self.done { - break; - } - - self.temp_buffer.resize(self.block_size, 0); - let n = { - let inner = &mut self.inner; - let temp = &mut self.temp_buffer[..self.block_size]; - match inner.read_block(temp).await { - Ok(n) => n, - Err(err) if err.kind() == io::ErrorKind::UnexpectedEof => 0, - Err(err) => return Err(err), - } - }; - - if n == 0 { - self.done = true; - self.temp_buffer.clear(); - break; - } - - let block = self.temp_buffer[..n].to_vec(); - self.temp_buffer.clear(); - self.queue_compressed_block(&block)?; - written += self.copy_buffered(&mut buf[written..]); - } - - Ok(written) - }) - } -} - pin_project! { /// A reader wrapper that decompresses data on the fly using DEFLATE algorithm. /// Header format: @@ -460,7 +390,6 @@ fn build_compressed_block(uncompressed_data: &[u8], compression_algorithm: Compr #[cfg(test)] mod tests { use super::*; - use crate::{BlockReadable, WarpReader}; use rand::RngExt; use std::io::Cursor; use tokio::io::{AsyncReadExt, BufReader}; @@ -550,27 +479,4 @@ mod tests { assert_eq!(&decompressed, &data); } - - #[tokio::test] - async fn test_compress_reader_read_block_round_trips() { - let data = b"hello world, hello world, hello world!"; - let reader = Cursor::new(data.to_vec()); - let mut compress_reader = CompressReader::new(WarpReader::new(reader), CompressionAlgorithm::Gzip); - let mut compressed = Vec::new(); - let mut buf = [0u8; 19]; - - loop { - let n = compress_reader.read_block(&mut buf).await.unwrap(); - if n == 0 { - break; - } - compressed.extend_from_slice(&buf[..n]); - } - - let mut decompress_reader = DecompressReader::new(Cursor::new(compressed), CompressionAlgorithm::Gzip); - let mut decompressed = Vec::new(); - decompress_reader.read_to_end(&mut decompressed).await.unwrap(); - - assert_eq!(&decompressed, data); - } } diff --git a/crates/rio/src/encrypt_reader.rs b/crates/rio/src/encrypt_reader.rs index d815a75d3..80e37d9f3 100644 --- a/crates/rio/src/encrypt_reader.rs +++ b/crates/rio/src/encrypt_reader.rs @@ -13,7 +13,6 @@ // limitations under the License. use crate::compress_index::{Index, TryGetIndex}; -use crate::{BlockReadable, BoxReadBlockFuture, Reader}; use aes_gcm::aead::Aead; use aes_gcm::{Aes256Gcm, KeyInit, Nonce}; use pin_project_lite::pin_project; @@ -60,69 +59,6 @@ where } } -fn encrypt_segment_bytes(cipher: &Aes256Gcm, nonce_bytes: &[u8; 12], plaintext: &[u8]) -> std::io::Result> { - let nonce = Nonce::try_from(nonce_bytes.as_slice()).map_err(|_| Error::other("invalid nonce length"))?; - let plaintext_len = plaintext.len(); - let crc = { - let mut hasher = crc_fast::Digest::new(crc_fast::CrcAlgorithm::Crc32IsoHdlc); - hasher.update(plaintext); - hasher.finalize() as u32 - }; - let ciphertext = cipher - .encrypt(&nonce, plaintext) - .map_err(|e| Error::other(format!("encrypt error: {e}")))?; - let int_len = put_uvarint_len(plaintext_len as u64); - let clen = int_len + ciphertext.len() + 4; - let mut header = [0u8; 8]; - header[0] = 0x00; - header[1] = (clen & 0xFF) as u8; - header[2] = ((clen >> 8) & 0xFF) as u8; - header[3] = ((clen >> 16) & 0xFF) as u8; - header[4] = (crc & 0xFF) as u8; - header[5] = ((crc >> 8) & 0xFF) as u8; - header[6] = ((crc >> 16) & 0xFF) as u8; - header[7] = ((crc >> 24) & 0xFF) as u8; - debug!( - "encrypt block header typ=0 len={} header={:?} plaintext_len={} ciphertext_len={}", - clen, - header, - plaintext_len, - ciphertext.len() - ); - let mut out = Vec::with_capacity(8 + int_len + ciphertext.len()); - out.extend_from_slice(&header); - let mut plaintext_len_buf = vec![0u8; int_len]; - put_uvarint(&mut plaintext_len_buf, plaintext_len as u64); - out.extend_from_slice(&plaintext_len_buf); - out.extend_from_slice(&ciphertext); - Ok(out) -} - -impl EncryptReader -where - R: Reader, -{ - fn copy_buffered(&mut self, buf: &mut [u8]) -> usize { - if self.buffer_pos >= self.buffer.len() || buf.is_empty() { - return 0; - } - - let to_copy = buf.len().min(self.buffer.len() - self.buffer_pos); - buf[..to_copy].copy_from_slice(&self.buffer[self.buffer_pos..self.buffer_pos + to_copy]); - self.buffer_pos += to_copy; - if self.buffer_pos == self.buffer.len() { - self.buffer.clear(); - self.buffer_pos = 0; - } - to_copy - } - - fn encrypt_segment(&self, plaintext: &[u8]) -> std::io::Result> { - let nonce = derive_block_nonce(&self.base_nonce, self.block_index); - encrypt_segment_bytes(&self.cipher, &nonce, plaintext) - } -} - impl AsyncRead for EncryptReader where R: AsyncRead + Unpin + Send + Sync, @@ -161,8 +97,49 @@ where *this.buffer_pos += to_copy; Poll::Ready(Ok(())) } else { + // Encrypt the chunk let block_nonce = derive_block_nonce(this.base_nonce, *this.block_index); - *this.buffer = encrypt_segment_bytes(this.cipher, &block_nonce, &this.read_buffer[..n])?; + let nonce = Nonce::try_from(block_nonce.as_slice()).map_err(|_| Error::other("invalid nonce length"))?; + let plaintext = &this.read_buffer[..n]; + let plaintext_len = plaintext.len(); + let crc = { + let mut hasher = crc_fast::Digest::new(crc_fast::CrcAlgorithm::Crc32IsoHdlc); + hasher.update(plaintext); + hasher.finalize() as u32 + }; + let ciphertext = this + .cipher + .encrypt(&nonce, plaintext) + .map_err(|e| Error::other(format!("encrypt error: {e}")))?; + let int_len = put_uvarint_len(plaintext_len as u64); + let clen = int_len + ciphertext.len() + 4; + // Header: 8 bytes + // 0: type (0 = encrypted, 0xFF = end) + // 1-3: length (little endian u24, ciphertext length) + // 4-7: CRC32 of plaintext (little endian u32) + let mut header = [0u8; 8]; + header[0] = 0x00; // 0 = encrypted + header[1] = (clen & 0xFF) as u8; + header[2] = ((clen >> 8) & 0xFF) as u8; + header[3] = ((clen >> 16) & 0xFF) as u8; + header[4] = (crc & 0xFF) as u8; + header[5] = ((crc >> 8) & 0xFF) as u8; + header[6] = ((crc >> 16) & 0xFF) as u8; + header[7] = ((crc >> 24) & 0xFF) as u8; + debug!( + "encrypt block header typ=0 len={} header={:?} plaintext_len={} ciphertext_len={}", + clen, + header, + plaintext_len, + ciphertext.len() + ); + let mut out = Vec::with_capacity(8 + int_len + ciphertext.len()); + out.extend_from_slice(&header); + let mut plaintext_len_buf = [0u8; 10]; + let encoded_len = put_uvarint(&mut plaintext_len_buf, plaintext_len as u64); + out.extend_from_slice(&plaintext_len_buf[..encoded_len]); + out.extend_from_slice(&ciphertext); + *this.buffer = out; *this.buffer_pos = 0; *this.block_index += 1; let to_copy = std::cmp::min(buf.remaining(), this.buffer.len()); @@ -187,50 +164,6 @@ where } } -impl BlockReadable for EncryptReader -where - R: Reader, -{ - fn read_block<'a>(&'a mut self, buf: &'a mut [u8]) -> BoxReadBlockFuture<'a> { - Box::pin(async move { - if buf.is_empty() { - return Ok(0); - } - - let mut written = self.copy_buffered(buf); - while written < buf.len() { - if self.finished { - break; - } - - let mut plaintext = vec![0u8; 8 * 1024]; - let n = match self.inner.read_block(&mut plaintext).await { - Ok(n) => n, - Err(err) if err.kind() == std::io::ErrorKind::UnexpectedEof => 0, - Err(err) => return Err(err), - }; - if n == 0 { - self.buffer = [0xFF, 0, 0, 0, 0, 0, 0, 0].to_vec(); - self.buffer_pos = 0; - self.finished = true; - } else { - self.buffer = self.encrypt_segment(&plaintext[..n])?; - self.buffer_pos = 0; - } - - let copied = self.copy_buffered(&mut buf[written..]); - written += copied; - - if copied == 0 && self.finished { - break; - } - } - - Ok(written) - }) - } -} - pin_project! { /// A reader wrapper that decrypts data on the fly using AES-256-GCM. /// This is a demonstration. For production, use a secure and audited crypto library. @@ -553,7 +486,7 @@ mod tests { use std::pin::Pin; use std::task::{Context, Poll}; - use crate::{BlockReadable, HardLimitReader, WarpReader}; + use crate::HardLimitReader; use super::*; use futures::StreamExt; @@ -741,35 +674,6 @@ mod tests { assert_eq!(&decrypted, &data); } - #[tokio::test] - async fn test_encrypt_reader_read_block_round_trips() { - let data = b"hello sse encrypt via blocks"; - let mut key = [0u8; 32]; - let mut nonce = [0u8; 12]; - rand::rng().fill_bytes(&mut key); - rand::rng().fill_bytes(&mut nonce); - - let reader = Cursor::new(data.to_vec()); - let mut encrypt_reader = EncryptReader::new(WarpReader::new(reader), key, nonce); - let mut encrypted = Vec::new(); - let mut buf = [0u8; 17]; - - loop { - let n = encrypt_reader.read_block(&mut buf).await.unwrap(); - if n == 0 { - break; - } - encrypted.extend_from_slice(&buf[..n]); - } - - let reader = Cursor::new(encrypted); - let mut decrypt_reader = DecryptReader::new(WarpReader::new(reader), key, nonce); - let mut decrypted = Vec::new(); - decrypt_reader.read_to_end(&mut decrypted).await.unwrap(); - - assert_eq!(&decrypted, data); - } - #[tokio::test] async fn test_decrypt_reader_large_with_small_chunks() { let size = 1024 * 1024; diff --git a/crates/rio/src/etag_reader.rs b/crates/rio/src/etag_reader.rs index d6f410ae9..ba1638069 100644 --- a/crates/rio/src/etag_reader.rs +++ b/crates/rio/src/etag_reader.rs @@ -13,7 +13,7 @@ // limitations under the License. use crate::compress_index::{Index, TryGetIndex}; -use crate::{BlockReadable, BoxReadBlockFuture, EtagResolvable, HashReaderDetector, HashReaderMut, Reader}; +use crate::{EtagResolvable, HashReaderDetector, HashReaderMut}; use md5::{Digest, Md5}; use pin_project_lite::pin_project; use std::pin::Pin; @@ -135,41 +135,9 @@ where } } -impl BlockReadable for EtagReader -where - R: Reader, -{ - fn read_block<'a>(&'a mut self, buf: &'a mut [u8]) -> BoxReadBlockFuture<'a> { - Box::pin(async move { - let n = match self.inner.read_block(buf).await { - Ok(n) => n, - Err(err) if err.kind() == std::io::ErrorKind::UnexpectedEof => 0, - Err(err) => return Err(err), - }; - if n > 0 { - self.md5.update(&buf[..n]); - return Ok(n); - } - - self.finished = true; - if let Some(checksum) = &self.checksum { - let etag = self.md5.clone().finalize().to_vec(); - let etag_hex = hex_simd::encode_to_string(etag, hex_simd::AsciiCase::Lower); - if checksum != &etag_hex { - error!("Checksum mismatch, expected={:?}, actual={:?}", checksum, etag_hex); - return Err(std::io::Error::new(std::io::ErrorKind::InvalidData, "Checksum mismatch")); - } - } - - Ok(0) - }) - } -} - #[cfg(test)] mod tests { use super::*; - use crate::{BlockReadable, WarpReader}; use rand::RngExt; use std::io::Cursor; use tokio::io::{AsyncReadExt, BufReader}; @@ -212,24 +180,6 @@ mod tests { assert_eq!(etag, Some(expected)); } - #[tokio::test] - async fn test_etag_reader_read_block_updates_checksum() { - let data = b"hello world"; - let mut hasher = Md5::new(); - hasher.update(data); - let expected = faster_hex::hex_string(hasher.finalize().as_slice()).to_string(); - let reader = BufReader::new(&data[..]); - let reader = Box::new(WarpReader::new(reader)); - let mut etag_reader = EtagReader::new(reader, Some(expected.clone())); - - let mut buf = [0_u8; 32]; - let n = etag_reader.read_block(&mut buf).await.unwrap(); - assert_eq!(n, data.len()); - assert_eq!(&buf[..n], data); - assert_eq!(etag_reader.read_block(&mut buf).await.unwrap(), 0); - assert_eq!(etag_reader.try_resolve_etag(), Some(expected)); - } - #[tokio::test] async fn test_etag_reader_multiple_get() { let data = b"abc123"; diff --git a/crates/rio/src/hardlimit_reader.rs b/crates/rio/src/hardlimit_reader.rs index c74f37d89..e50b052f5 100644 --- a/crates/rio/src/hardlimit_reader.rs +++ b/crates/rio/src/hardlimit_reader.rs @@ -12,7 +12,6 @@ // See the License for the specific language governing permissions and // limitations under the License. -use crate::{BlockReadable, BoxReadBlockFuture, Reader}; use pin_project_lite::pin_project; use std::io::{Error, Result}; use std::pin::Pin; @@ -62,51 +61,11 @@ where delegate_reader_capabilities_generic!(HardLimitReader, inner); -impl BlockReadable for HardLimitReader -where - R: Reader, -{ - fn read_block<'a>(&'a mut self, buf: &'a mut [u8]) -> BoxReadBlockFuture<'a> { - Box::pin(async move { - if self.remaining < 0 { - return Err(Error::other("input provided more bytes than specified")); - } - - let max_len = match usize::try_from(self.remaining) { - Ok(remaining) => remaining.min(buf.len()), - Err(_) => buf.len(), - }; - - if max_len == 0 { - let mut probe = [0_u8; 1]; - match self.inner.read_block(&mut probe).await { - Ok(0) => return Ok(0), - Ok(n) => { - self.remaining -= n as i64; - return Err(Error::other("input provided more bytes than specified")); - } - Err(err) if err.kind() == std::io::ErrorKind::UnexpectedEof => return Ok(0), - Err(err) => return Err(err), - } - } - - let n = self.inner.read_block(&mut buf[..max_len]).await?; - self.remaining -= n as i64; - if self.remaining < 0 { - return Err(Error::other("input provided more bytes than specified")); - } - - Ok(n) - }) - } -} - #[cfg(test)] mod tests { use std::vec; use super::*; - use crate::{BlockReadable, WarpReader}; use rustfs_utils::read_full; use tokio::io::{AsyncReadExt, BufReader}; @@ -169,20 +128,4 @@ mod tests { assert_eq!(n, 0); assert_eq!(&buf, data); } - - #[tokio::test] - async fn test_hardlimit_reader_read_block_enforces_limit() { - let data = b"abcdef"; - let reader = BufReader::new(&data[..]); - let reader = Box::new(WarpReader::new(reader)); - let mut hardlimit = HardLimitReader::new(reader, 3); - - let mut buf = [0_u8; 8]; - let n = hardlimit.read_block(&mut buf).await.unwrap(); - assert_eq!(n, 3); - assert_eq!(&buf[..n], b"abc"); - - let err = hardlimit.read_block(&mut buf).await.unwrap_err(); - assert_eq!(err.kind(), std::io::ErrorKind::Other); - } } diff --git a/crates/rio/src/hash_reader.rs b/crates/rio/src/hash_reader.rs index 092eb3786..98fb70037 100644 --- a/crates/rio/src/hash_reader.rs +++ b/crates/rio/src/hash_reader.rs @@ -90,10 +90,7 @@ use crate::ChecksumType; use crate::Sha256Hasher; use crate::compress_index::{Index, TryGetIndex}; use crate::get_content_checksum; -use crate::{ - BlockReadable, BoxReadBlockFuture, DynReader, EtagReader, EtagResolvable, HardLimitReader, HashReaderDetector, WarpReader, - boxed_reader, wrap_reader, -}; +use crate::{DynReader, EtagReader, EtagResolvable, HardLimitReader, HashReaderDetector, WarpReader, boxed_reader, wrap_reader}; use base64::Engine; use base64::engine::general_purpose; use http::HeaderMap; @@ -469,97 +466,6 @@ impl HashReader { } map } - - pub fn finalize_content_hash(&mut self) -> std::io::Result> { - self.finish_checksum_validation()?; - Ok(self.content_hash.clone()) - } - - fn update_read_state(&mut self, data: &[u8]) -> std::io::Result<()> { - self.bytes_read += data.len() as u64; - - if data.is_empty() { - return Ok(()); - } - - if let Some(hasher) = self.content_sha256_hasher.as_mut() { - hasher.write_all(data)?; - } - - if let Some(hasher) = self.content_hasher.as_mut() { - hasher.write_all(data)?; - } - - Ok(()) - } - - fn finish_checksum_validation(&mut self) -> std::io::Result<()> { - if self.checksum_on_finish { - return Ok(()); - } - - if let (Some(mut hasher), Some(expected_sha256)) = (self.content_sha256_hasher.take(), self.content_sha256.as_ref()) { - let sha256 = hex_simd::encode_to_string(hasher.finalize(), hex_simd::AsciiCase::Lower); - if sha256 != *expected_sha256 { - error!("SHA256 mismatch, expected={:?}, actual={:?}", expected_sha256, sha256); - return Err(std::io::Error::new(std::io::ErrorKind::InvalidData, "SHA256 mismatch")); - } - } - - if let Some(mut expected_content_hash) = self.content_hash.clone() - && let Some(mut hasher) = self.content_hasher.take() - { - if expected_content_hash.checksum_type.trailing() - && let Some(trailer) = self.trailer_s3s.as_ref() - && let Some(Some(checksum_str)) = trailer.read(|headers| { - expected_content_hash - .checksum_type - .key() - .and_then(|key| headers.get(key).and_then(|value| value.to_str().ok().map(|s| s.to_string()))) - }) - { - expected_content_hash.encoded = checksum_str; - expected_content_hash.raw = general_purpose::STANDARD - .decode(&expected_content_hash.encoded) - .map_err(|_| std::io::Error::other("Invalid base64 checksum"))?; - - if expected_content_hash.raw.is_empty() { - return Err(std::io::Error::other("Content hash mismatch")); - } - } - - let content_hash = hasher.finalize(); - if expected_content_hash.encoded.is_empty() { - expected_content_hash.raw = content_hash.clone(); - expected_content_hash.encoded = general_purpose::STANDARD.encode(&content_hash); - self.content_hash = Some(expected_content_hash); - } else if content_hash != expected_content_hash.raw { - let expected_hex = hex_simd::encode_to_string(&expected_content_hash.raw, hex_simd::AsciiCase::Lower); - let actual_hex = hex_simd::encode_to_string(content_hash, hex_simd::AsciiCase::Lower); - error!( - "Content hash mismatch, type={:?}, encoded={:?}, expected={:?}, actual={:?}", - expected_content_hash.checksum_type, expected_content_hash.encoded, expected_hex, actual_hex - ); - let checksum_err = crate::errors::ChecksumMismatch { - want: expected_hex, - got: actual_hex, - }; - return Err(std::io::Error::new(std::io::ErrorKind::InvalidData, checksum_err)); - } - } - - self.checksum_on_finish = true; - Ok(()) - } - - pub async fn read_block(&mut self, buf: &mut [u8]) -> std::io::Result { - let n = self.inner.read_block(buf).await?; - self.update_read_state(&buf[..n])?; - if n == 0 { - self.finish_checksum_validation()?; - } - Ok(n) - } } impl HashReaderMut for HashReader { @@ -619,23 +525,90 @@ impl HashReaderMut for HashReader { impl AsyncRead for HashReader { fn poll_read(self: Pin<&mut Self>, cx: &mut Context<'_>, buf: &mut ReadBuf<'_>) -> Poll> { - let this = self.get_mut(); + let this = self.project(); let before = buf.filled().len(); - match Pin::new(&mut this.inner).poll_read(cx, buf) { + match this.inner.poll_read(cx, buf) { Poll::Pending => Poll::Pending, Poll::Ready(Ok(())) => { let data = &buf.filled()[before..]; let filled = data.len(); - if let Err(e) = this.update_read_state(data) { - error!("hash reader state update error, error={:?}", e); - return Poll::Ready(Err(std::io::Error::other(e))); + + *this.bytes_read += filled as u64; + + if filled > 0 { + // Update SHA256 hasher + if let Some(hasher) = this.content_sha256_hasher + && let Err(e) = hasher.write_all(data) + { + error!("SHA256 hasher write error, error={:?}", e); + return Poll::Ready(Err(std::io::Error::other(e))); + } + + // Update content hasher + if let Some(hasher) = this.content_hasher + && let Err(e) = hasher.write_all(data) + { + return Poll::Ready(Err(std::io::Error::other(e))); + } } - if filled == 0 - && let Err(e) = this.finish_checksum_validation() - { - return Poll::Ready(Err(e)); + if filled == 0 && !*this.checksum_on_finish { + // check SHA256 + if let (Some(hasher), Some(expected_sha256)) = (this.content_sha256_hasher, this.content_sha256) { + let sha256 = hex_simd::encode_to_string(hasher.finalize(), hex_simd::AsciiCase::Lower); + if sha256 != *expected_sha256 { + error!("SHA256 mismatch, expected={:?}, actual={:?}", expected_sha256, sha256); + return Poll::Ready(Err(std::io::Error::new(std::io::ErrorKind::InvalidData, "SHA256 mismatch"))); + } + } + + // check content hasher + if let (Some(hasher), Some(expected_content_hash)) = + (this.content_hasher.as_mut(), this.content_hash.as_mut()) + { + if expected_content_hash.checksum_type.trailing() + && let Some(trailer) = this.trailer_s3s.as_ref() + && let Some(Some(checksum_str)) = trailer.read(|headers| { + expected_content_hash + .checksum_type + .key() + .and_then(|key| headers.get(key).and_then(|value| value.to_str().ok().map(|s| s.to_string()))) + }) + { + expected_content_hash.encoded = checksum_str; + expected_content_hash.raw = general_purpose::STANDARD + .decode(&expected_content_hash.encoded) + .map_err(|_| std::io::Error::other("Invalid base64 checksum"))?; + + if expected_content_hash.raw.is_empty() { + return Poll::Ready(Err(std::io::Error::other("Content hash mismatch"))); + } + } + + let content_hash = hasher.finalize(); + + if expected_content_hash.encoded.is_empty() { + expected_content_hash.raw = content_hash.clone(); + expected_content_hash.encoded = general_purpose::STANDARD.encode(&content_hash); + *this.content_hash = Some(expected_content_hash.clone()); + } else if content_hash != expected_content_hash.raw { + let expected_hex = hex_simd::encode_to_string(&expected_content_hash.raw, hex_simd::AsciiCase::Lower); + let actual_hex = hex_simd::encode_to_string(content_hash, hex_simd::AsciiCase::Lower); + error!( + "Content hash mismatch, type={:?}, encoded={:?}, expected={:?}, actual={:?}", + expected_content_hash.checksum_type, expected_content_hash.encoded, expected_hex, actual_hex + ); + // Use ChecksumMismatch error so that API layer can return BadDigest + let checksum_err = crate::errors::ChecksumMismatch { + want: expected_hex, + got: actual_hex, + }; + return Poll::Ready(Err(std::io::Error::new(std::io::ErrorKind::InvalidData, checksum_err))); + } + } + + *this.checksum_on_finish = true; } Poll::Ready(Ok(())) } @@ -673,39 +646,13 @@ impl TryGetIndex for HashReader { } } -impl BlockReadable for HashReader { - fn read_block<'a>(&'a mut self, buf: &'a mut [u8]) -> BoxReadBlockFuture<'a> { - Box::pin(async move { self.read_block(buf).await }) - } -} - #[cfg(test)] mod tests { use super::*; use crate::{DecryptReader, EncryptReader, encrypt_reader, wrap_reader}; use rand::RngExt; use std::io::Cursor; - use std::pin::Pin; - use std::task::{Context, Poll}; - use tokio::io::{AsyncRead, AsyncReadExt, BufReader, ReadBuf}; - - struct UnexpectedEofReader; - - impl AsyncRead for UnexpectedEofReader { - fn poll_read(self: Pin<&mut Self>, _cx: &mut Context<'_>, _buf: &mut ReadBuf<'_>) -> Poll> { - Poll::Ready(Ok(())) - } - } - - impl BlockReadable for UnexpectedEofReader { - fn read_block<'a>(&'a mut self, _buf: &'a mut [u8]) -> BoxReadBlockFuture<'a> { - Box::pin(async { Err(std::io::Error::new(std::io::ErrorKind::UnexpectedEof, "synthetic unexpected eof")) }) - } - } - - impl EtagResolvable for UnexpectedEofReader {} - impl HashReaderDetector for UnexpectedEofReader {} - impl TryGetIndex for UnexpectedEofReader {} + use tokio::io::{AsyncReadExt, BufReader}; #[tokio::test] async fn test_hashreader_wrapping_logic() { @@ -818,37 +765,6 @@ mod tests { assert_eq!(buf, data); } - #[tokio::test] - async fn test_hashreader_read_block_reads_full_and_tail_block() { - let data = b"hello block reader"; - let reader = BufReader::new(Cursor::new(&data[..])); - let reader = Box::new(WarpReader::new(reader)); - let mut hash_reader = HashReader::new(reader, data.len() as i64, data.len() as i64, None, None, false).unwrap(); - - let mut first = [0_u8; 8]; - let n1 = hash_reader.read_block(&mut first).await.unwrap(); - assert_eq!(n1, 8); - assert_eq!(&first[..n1], b"hello bl"); - - let mut second = [0_u8; 32]; - let n2 = hash_reader.read_block(&mut second).await.unwrap(); - assert_eq!(n2, data.len() - n1); - assert_eq!(&second[..n2], b"ock reader"); - - let n3 = hash_reader.read_block(&mut second).await.unwrap(); - assert_eq!(n3, 0); - } - - #[tokio::test] - async fn test_hashreader_read_block_propagates_unexpected_eof() { - let mut hash_reader = HashReader::new(Box::new(UnexpectedEofReader), 0, 0, None, None, true).unwrap(); - let mut buf = [0_u8; 8]; - - let err = hash_reader.read_block(&mut buf).await.unwrap_err(); - - assert_eq!(err.kind(), std::io::ErrorKind::UnexpectedEof); - } - #[tokio::test] async fn test_hashreader_new_logic() { let data = b"test data"; diff --git a/crates/rio/src/http_reader.rs b/crates/rio/src/http_reader.rs index b207fbc95..39ef43ecd 100644 --- a/crates/rio/src/http_reader.rs +++ b/crates/rio/src/http_reader.rs @@ -23,6 +23,7 @@ use rustfs_utils::get_env_opt_str; use std::io::IoSlice; use std::io::{self, Error}; use std::net::IpAddr; +use std::ops::Not as _; use std::pin::Pin; use std::sync::LazyLock; use std::task::{Context, Poll}; @@ -136,71 +137,6 @@ fn get_http_client(url: &str) -> Client { CLIENT.clone() } -type HttpByteStream = Pin> + Send + Sync>>; - -async fn request_http_byte_stream( - url: String, - method: Method, - headers: HeaderMap, - body: Option>, - meter_stream_recv_bytes: bool, -) -> io::Result<(bool, HttpByteStream)> { - let track_internode_metrics = is_internode_rpc_url(&url); - let client = get_http_client(&url); - let mut request: RequestBuilder = client.request(method, url.clone()).headers(headers); - if let Some(body) = body { - request = request.body(body); - } - - let resp = request.send().await.map_err(|e| { - if track_internode_metrics { - global_internode_metrics().record_error(); - } - Error::other(format!("HttpReader HTTP request error: {e}")) - })?; - - if !resp.status().is_success() { - if track_internode_metrics { - global_internode_metrics().record_error(); - } - return Err(Error::other(format!( - "HttpReader HTTP request failed with non-200 status {}", - resp.status() - ))); - } - - if track_internode_metrics { - global_internode_metrics().record_outgoing_request(); - } - - let stream = resp - .bytes_stream() - .map_ok(move |bytes| { - if track_internode_metrics && meter_stream_recv_bytes { - global_internode_metrics().record_recv_bytes(bytes.len()); - } - bytes - }) - .map_err(move |e| { - if track_internode_metrics { - global_internode_metrics().record_error(); - } - Error::other(format!("HttpReader stream error: {e}")) - }); - - Ok((track_internode_metrics, Box::pin(stream))) -} - -pub async fn open_http_byte_stream( - url: String, - method: Method, - headers: HeaderMap, - body: Option>, -) -> io::Result { - let (_track_internode_metrics, stream) = request_http_byte_stream(url, method, headers, body, true).await?; - Ok(stream) -} - pin_project! { pub struct HttpReader { url:String, @@ -225,11 +161,43 @@ impl HttpReader { body: Option>, _read_buf_size: usize, ) -> io::Result { - let (track_internode_metrics, stream) = - request_http_byte_stream(url.clone(), method.clone(), headers.clone(), body, false).await?; + let track_internode_metrics = is_internode_rpc_url(&url); + let client = get_http_client(&url); + let mut request: RequestBuilder = client.request(method.clone(), url.clone()).headers(headers.clone()); + if let Some(body) = body { + request = request.body(body); + } + + let resp = request.send().await.map_err(|e| { + if track_internode_metrics { + global_internode_metrics().record_error(); + } + Error::other(format!("HttpReader HTTP request error: {e}")) + })?; + + if resp.status().is_success().not() { + if track_internode_metrics { + global_internode_metrics().record_error(); + } + return Err(Error::other(format!( + "HttpReader HTTP request failed with non-200 status {}", + resp.status() + ))); + } + + if track_internode_metrics { + global_internode_metrics().record_outgoing_request(); + } + + let stream = resp.bytes_stream().map_err(move |e| { + if track_internode_metrics { + global_internode_metrics().record_error(); + } + Error::other(format!("HttpReader stream error: {e}")) + }); Ok(Self { - inner: StreamReader::new(stream), + inner: StreamReader::new(Box::pin(stream)), url, method, headers, diff --git a/crates/rio/src/lib.rs b/crates/rio/src/lib.rs index 6e9040382..9663f133d 100644 --- a/crates/rio/src/lib.rs +++ b/crates/rio/src/lib.rs @@ -15,10 +15,6 @@ // Default encryption block size - aligned with system default read buffer size (1MB) pub const DEFAULT_ENCRYPTION_BLOCK_SIZE: usize = 1024 * 1024; -use std::future::Future; -use std::pin::Pin; -use tokio::io::AsyncReadExt; - macro_rules! delegate_reader_capabilities_generic { ($name:ident<$inner_ty:ident>, $inner:ident) => { impl<$inner_ty> crate::EtagResolvable for $name<$inner_ty> @@ -118,39 +114,14 @@ pub use compress_index::{Index, TryGetIndex}; mod etag; -pub type BoxReadBlockFuture<'a> = Pin> + Send + 'a>>; - -pub trait BlockReadable { - fn read_block<'a>(&'a mut self, buf: &'a mut [u8]) -> BoxReadBlockFuture<'a>; -} - -fn read_block_via_async_read<'a, R>(reader: &'a mut R, buf: &'a mut [u8]) -> BoxReadBlockFuture<'a> -where - R: tokio::io::AsyncRead + Unpin + Send + Sync + 'a, -{ - Box::pin(async move { - let mut total = 0; - - while total < buf.len() { - match reader.read(&mut buf[total..]).await { - Ok(0) => return Ok(total), - Ok(n) => total += n, - Err(err) => return Err(err), - } - } - - Ok(total) - }) -} - pub trait ReadStream: tokio::io::AsyncRead + Unpin + Send + Sync {} impl ReadStream for T where T: tokio::io::AsyncRead + Unpin + Send + Sync {} pub trait ReaderCapabilities: EtagResolvable + HashReaderDetector + TryGetIndex {} impl ReaderCapabilities for T where T: EtagResolvable + HashReaderDetector + TryGetIndex {} -pub trait Reader: ReadStream + ReaderCapabilities + BlockReadable {} -impl Reader for T where T: ReadStream + ReaderCapabilities + BlockReadable {} +pub trait Reader: ReadStream + ReaderCapabilities {} +impl Reader for T where T: ReadStream + ReaderCapabilities {} pub type DynReader = Box; @@ -183,42 +154,6 @@ pub trait HashReaderDetector { } } -impl BlockReadable for crate::WarpReader -where - R: tokio::io::AsyncRead + Unpin + Send + Sync, -{ - fn read_block<'a>(&'a mut self, buf: &'a mut [u8]) -> BoxReadBlockFuture<'a> { - read_block_via_async_read(self, buf) - } -} - -impl BlockReadable for tokio::io::BufReader -where - R: tokio::io::AsyncRead + Unpin + Send + Sync, -{ - fn read_block<'a>(&'a mut self, buf: &'a mut [u8]) -> BoxReadBlockFuture<'a> { - read_block_via_async_read(self, buf) - } -} - -impl BlockReadable for crate::LimitReader -where - R: Reader, -{ - fn read_block<'a>(&'a mut self, buf: &'a mut [u8]) -> BoxReadBlockFuture<'a> { - read_block_via_async_read(self, buf) - } -} - -impl BlockReadable for crate::DecryptReader -where - R: Reader, -{ - fn read_block<'a>(&'a mut self, buf: &'a mut [u8]) -> BoxReadBlockFuture<'a> { - read_block_via_async_read(self, buf) - } -} - pub fn boxed_reader(reader: R) -> DynReader where R: Reader + 'static, @@ -263,74 +198,3 @@ where self.as_ref().try_get_index() } } - -impl BlockReadable for Box -where - T: BlockReadable + ?Sized, -{ - fn read_block<'a>(&'a mut self, buf: &'a mut [u8]) -> BoxReadBlockFuture<'a> { - self.as_mut().read_block(buf) - } -} - -#[cfg(test)] -mod tests { - use super::*; - use std::collections::VecDeque; - use std::io::{self, ErrorKind}; - use std::pin::Pin; - use std::task::{Context, Poll}; - use tokio::io::{AsyncRead, ReadBuf}; - - enum ReadStep { - Data(Vec), - Error(ErrorKind), - Eof, - } - - struct StepReader { - steps: VecDeque, - } - - impl StepReader { - fn new(steps: impl IntoIterator) -> Self { - Self { - steps: steps.into_iter().collect(), - } - } - } - - impl AsyncRead for StepReader { - fn poll_read(mut self: Pin<&mut Self>, _cx: &mut Context<'_>, buf: &mut ReadBuf<'_>) -> Poll> { - match self.steps.pop_front().unwrap_or(ReadStep::Eof) { - ReadStep::Data(data) => { - buf.put_slice(&data); - Poll::Ready(Ok(())) - } - ReadStep::Error(kind) => Poll::Ready(Err(io::Error::new(kind, "synthetic read failure"))), - ReadStep::Eof => Poll::Ready(Ok(())), - } - } - } - - #[tokio::test] - async fn test_read_block_via_async_read_preserves_midstream_error_kind() { - let reader = StepReader::new([ReadStep::Data(b"ab".to_vec()), ReadStep::Error(ErrorKind::ConnectionReset)]); - let mut reader = WarpReader::new(reader); - let mut buf = [0_u8; 4]; - - let err = reader.read_block(&mut buf).await.unwrap_err(); - - assert_eq!(err.kind(), ErrorKind::ConnectionReset); - } - - #[tokio::test] - async fn test_read_block_via_async_read_returns_zero_on_initial_eof() { - let mut reader = WarpReader::new(StepReader::new([ReadStep::Eof])); - let mut buf = [0_u8; 4]; - - let n = reader.read_block(&mut buf).await.unwrap(); - - assert_eq!(n, 0); - } -} diff --git a/rustfs/src/app/object_usecase/put_object_flow.rs b/rustfs/src/app/object_usecase/put_object_flow.rs index be4025700..105685b4a 100644 --- a/rustfs/src/app/object_usecase/put_object_flow.rs +++ b/rustfs/src/app/object_usecase/put_object_flow.rs @@ -22,7 +22,7 @@ use rustfs_object_io::put::{ apply_trailing_checksums, build_put_object_ingress_source, build_put_object_legacy_hash_stage, build_put_object_plain_hash_stage, plan_put_object_body_with_transforms, resolve_put_transformed_fallback_reason, }; -use rustfs_rio::{BlockReadable, BoxReadBlockFuture, EtagResolvable, HashReaderDetector, TryGetIndex}; +use rustfs_rio::{EtagResolvable, HashReaderDetector, TryGetIndex}; use rustfs_utils::http::headers::AMZ_TRAILER; const DEFAULT_SMALL_PUT_EAGER_MAX_BYTES: i64 = 1024 * 1024; @@ -145,22 +145,6 @@ impl AsyncRead for PooledBufferReader { } } -impl BlockReadable for PooledBufferReader { - fn read_block<'a>(&'a mut self, buf: &'a mut [u8]) -> BoxReadBlockFuture<'a> { - Box::pin(async move { - let remaining = &self.buffer[self.position..]; - if remaining.is_empty() { - return Ok(0); - } - - let to_copy = remaining.len().min(buf.len()); - buf[..to_copy].copy_from_slice(&remaining[..to_copy]); - self.position += to_copy; - Ok(to_copy) - }) - } -} - impl EtagResolvable for PooledBufferReader {} impl HashReaderDetector for PooledBufferReader {} @@ -480,10 +464,7 @@ impl DefaultObjectUsecase { if stage.ingress_kind == PutObjectIngressKind::ReducedCopyCandidate { plain_reduced_copy_stage = true; - debug!( - "Plain PUT is using the reduced-copy Reader + BlockReadable hash path (bucket={}, key={})", - bucket, key - ); + debug!("Plain PUT is using the reduced-copy Reader hash path (bucket={}, key={})", bucket, key); } stage