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

revert: remove BlockReadable trait and chunk-based I/O from rio, ecstore, disk layers (#2533)

Browse Source
This commit is contained in:
安正超
2026-04-14 19:09:02 +08:00
committed by GitHub
parent 2a3d127a86
commit 6d476ae9a1
22 changed files with 426 additions and 2807 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/ecstore/src/data_movement.rs
+1 -6
View File
@@ -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<R: AsyncRead + Unpin + Send + Sync> TryGetIndex for IndexedDataMovementRead
}
}
impl<R: AsyncRead + Unpin + Send + Sync> BlockReadable for IndexedDataMovementReader<R> {
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<Index> {
let bytes = index?;
let mut decoded = Index::new();
crates/ecstore/src/disk/disk_store.rs
-9
View File
@@ -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<BoxChunkStream> {
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<crate::disk::FileWriter> {
self.track_disk_health(|| async { self.disk.append_file(volume, path).await }, Duration::ZERO)
.await
crates/ecstore/src/disk/local.rs
+91 -783
View File
@@ -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<Uuid>,
@@ -109,410 +97,6 @@ pub struct LocalDisk {
exit_signal: Option<tokio::sync::broadcast::Sender<()>>,
}
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<BytesPool> = 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<usize> = 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<Bytes> {
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<MappedChunk> {
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<IoChunk> {
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<IoChunk> {
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<IoChunk> {
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<IoChunk> {
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<Bytes> {
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<BoxChunkStream> {
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, DiskError>(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
crates/ecstore/src/disk/mod.rs
-12
View File
@@ -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<BoxChunkStream> {
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<FileWriter> {
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<Bytes>;
/// 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<BoxChunkStream>;
async fn append_file(&self, volume: &str, path: &str) -> Result<FileWriter>;
async fn create_file(&self, origvolume: &str, volume: &str, path: &str, file_size: i64) -> Result<FileWriter>;
// ReadFileStream
crates/ecstore/src/erasure_coding/bitrot.rs
-68
View File
@@ -172,52 +172,6 @@ where
}
}
impl BitrotWriter<CustomWriter> {
fn write_inline_sync(&mut self, buf: &[u8]) -> std::io::Result<usize> {
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<W>(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<usize> {
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<Vec<u8>> {
match self.writer_type {
crates/ecstore/src/erasure_coding/decode.rs
-235
View File
@@ -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<Vec<u8>>], 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<Vec<u8>>],
data_blocks: usize,
mut offset: usize,
length: usize,
) -> io::Result<Vec<IoChunk>> {
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<W>(
writer: &mut W,
@@ -276,134 +213,6 @@ where
Ok(total_written)
}
pub(crate) struct ErasureChunkDecoder<R> {
erasure: Erasure,
reader: ParallelReader<R>,
offset: usize,
length: usize,
start_block: usize,
end_block: usize,
current_block: usize,
written: usize,
healable_error: Option<Error>,
finished: bool,
}
impl<R> ErasureChunkDecoder<R>
where
R: AsyncRead + Unpin + Send + Sync,
{
pub(crate) fn new(
erasure: Erasure,
readers: Vec<Option<BitrotReader<R>>>,
offset: usize,
length: usize,
total_length: usize,
) -> io::Result<Self> {
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<Option<Vec<IoChunk>>> {
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::<usize>();
Ok(Some(chunks))
}
pub(crate) fn written(&self) -> usize {
self.written
}
pub(crate) fn finish_error(&self) -> Option<io::Error> {
if self.written < self.length {
Some(Error::LessData.into())
} else {
None
}
}
pub(crate) fn take_healable_error(&mut self) -> Option<Error> {
self.healable_error.take()
}
}
pub(crate) type ReconstructedChunkDecoder<R> = ErasureChunkDecoder<R>;
impl Erasure {
pub async fn decode<W, R>(
&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<Cursor<Vec<u8>>> {
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::<Vec<_>>();
assert_eq!(collected, original);
assert!(decoder.next_chunks().await.unwrap().is_none());
assert_eq!(decoder.written(), original.len());
assert!(decoder.finish_error().is_none());
}
}
crates/ecstore/src/erasure_coding/encode.rs
+71 -308
View File
@@ -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<Option<Error>>,
}
pub(crate) struct BlockAssembler<R> {
reader: R,
block_buffer: EncodeBlockBuffer,
total_bytes: usize,
}
impl<R> BlockAssembler<R>
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<Option<Vec<u8>>> {
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<Erasure>,
buffer_pool: EncodedShardBufferPool,
}
impl ErasureChunkEncoder {
pub(crate) async fn new(erasure: Arc<Erasure>) -> 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<EncodedShardBlock> {
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<Erasure>,
write_quorum: usize,
}
impl ErasureWritePipeline {
pub(crate) fn new(erasure: Arc<Erasure>, write_quorum: usize) -> Self {
Self { erasure, write_quorum }
}
pub(crate) async fn run<R>(&self, reader: R, writers: &mut [Option<BitrotWriterWrapper>]) -> std::io::Result<(R, usize)>
where
R: AsyncRead + BlockReadable + Send + Sync + Unpin + 'static,
{
let (tx, mut rx) = mpsc::channel::<EncodedShardBlock>(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<Bytes> {
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<BitrotWriterWrapper>], write_quorum: usize) -> Self {
let length = writers.len();
@@ -201,10 +42,10 @@ impl<'a> MultiWriter<'a> {
}
}
async fn write_shard(writer_opt: &mut Option<BitrotWriterWrapper>, err: &mut Option<Error>, shard: Bytes) {
async fn write_shard(writer_opt: &mut Option<BitrotWriterWrapper>, err: &mut Option<Error>, 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<BitrotWriterWrapper>, err: &mut Option<Error>, 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<T>(&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<Bytes>) -> 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<T>(&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("<nil>".to_string(), |e| e.to_string()))
.collect::<Vec<_>>()
.join(", ")
)))
}
async fn shutdown_writer(writer_opt: &mut Option<BitrotWriterWrapper>, err: &mut Option<Error>) {
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<BitrotWriterWrapper>, err: &mut Option<Error>) {
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("<nil>".to_string(), |e| e.to_string()))
.collect::<Vec<_>>()
.join(", ")
)))
}
}
impl Erasure {
pub async fn encode<R>(
self: Arc<Self>,
reader: R,
mut reader: R,
writers: &mut [Option<BitrotWriterWrapper>],
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::<Vec<Bytes>>(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;
}
}
crates/ecstore/src/erasure_coding/erasure.rs
+26 -253
View File
@@ -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<u8>,
}
impl EncodeBlockBuffer {
pub(crate) fn new(block_size: usize) -> Self {
Self {
buf: vec![0u8; block_size],
}
}
pub(crate) async fn read_from<R>(&mut self, reader: &mut R) -> io::Result<usize>
where
R: AsyncRead + Send + Sync + Unpin,
{
rustfs_utils::read_full(&mut *reader, &mut self.buf).await
}
pub(crate) async fn read_from_block<R>(&mut self, reader: &mut R) -> io::Result<usize>
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<Item = Bytes> + '_ {
(0..self.shard_count).map(|idx| self.shard(idx))
}
pub fn into_vec(self) -> Vec<Bytes> {
(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<Mutex<Vec<BytesMut>>>,
}
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<Vec<Bytes>> {
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<EncodedShardBlock> {
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<EncodedShardBlock> {
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<EncodedShardBlock> {
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<EncodedShardBlock, std::io::Error>) -> Future<Output=Result<Option<EncodedShardBlock>, E>> + Send
/// * `mut on_block` - Async callback that receives encoded blocks and returns a Result
/// * `F` - Callback type: FnMut(Result<Vec<Bytes>, std::io::Error>) -> Future<Output=Result<(), E>> + 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<usize, E>
where
R: AsyncRead + Send + Sync + Unpin,
F: FnMut(std::io::Result<EncodedShardBlock>) -> Fut + Send,
Fut: std::future::Future<Output = Result<Option<EncodedShardBlock>, E>> + Send,
F: FnMut(std::io::Result<Vec<Bytes>>) -> Fut + Send,
Fut: std::future::Future<Output = Result<(), E>> + 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<Option<Vec<u8>>> = 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<Option<Vec<u8>>> = 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<Option<Vec<u8>>> = 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);
}
}
}
crates/ecstore/src/erasure_coding/heal.rs
+1 -1
View File
@@ -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(())
crates/ecstore/src/rpc/remote_disk.rs
+1 -30
View File
@@ -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<BoxChunkStream> {
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<FileWriter> {
info!("append_file {}/{}", volume, path);
crates/ecstore/src/set_disk/read.rs
+12 -224
View File
@@ -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<UnboundedReceiver<io::Result<IoChunk>>>,
}
impl ChannelChunkStream {
fn new(receiver: UnboundedReceiver<io::Result<IoChunk>>) -> Self {
Self {
receiver: Mutex::new(receiver),
}
}
}
impl Stream for ChannelChunkStream {
type Item = io::Result<IoChunk>;
fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
let mut receiver = self.receiver.lock().unwrap();
receiver.poll_recv(cx)
}
}
struct ChannelChunkWriter {
sender: UnboundedSender<io::Result<IoChunk>>,
buffer: BytesMut,
chunk_size: usize,
}
impl ChannelChunkWriter {
fn new(sender: UnboundedSender<io::Result<IoChunk>>, 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<std::result::Result<usize, io::Error>> {
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<std::result::Result<(), io::Error>> {
Poll::Ready(self.push_buffer())
}
fn poll_shutdown(mut self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<std::result::Result<(), io::Error>> {
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,
crates/io-metrics/src/lib.rs
+30
View File
@@ -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::*;
crates/object-io/src/put.rs
+1 -126
View File
@@ -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<S, B> PutObjectReducedCopyReader<S, B> {
buf.advance(copied);
copied
}
async fn read_into_slice<E>(&mut self, buf: &mut [u8]) -> std::io::Result<usize>
where
S: Stream<Item = Result<B, E>> + 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<S, B, E> AsyncRead for PutObjectReducedCopyReader<S, B>
@@ -371,17 +313,6 @@ where
}
}
impl<S, B, E> BlockReadable for PutObjectReducedCopyReader<S, B>
where
S: Stream<Item = Result<B, E>> + 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<S, B> EtagResolvable for PutObjectReducedCopyReader<S, B> {}
impl<S, B> HashReaderDetector for PutObjectReducedCopyReader<S, B> {}
@@ -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, &'static str>(Bytes::from_static(b"ab")),
Ok::<Bytes, &'static str>(Bytes::from_static(b"cd")),
Ok::<Bytes, &'static str>(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, &'static str>(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, &'static str>(Bytes::from_static(b"legacy"))]);
crates/rio/src/checksum.rs
+30 -22
View File
@@ -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==");
}
}
crates/rio/src/compress_reader.rs
-94
View File
@@ -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<R> TryGetIndex for CompressReader<R> {
@@ -196,50 +170,6 @@ where
delegate_reader_capabilities_generic_no_index!(CompressReader<R>, inner);
impl<R> BlockReadable for CompressReader<R>
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);
}
}
crates/rio/src/encrypt_reader.rs
+43 -139
View File
@@ -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<Vec<u8>> {
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<R> EncryptReader<R>
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<Vec<u8>> {
let nonce = derive_block_nonce(&self.base_nonce, self.block_index);
encrypt_segment_bytes(&self.cipher, &nonce, plaintext)
}
}
impl<R> AsyncRead for EncryptReader<R>
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<R> BlockReadable for EncryptReader<R>
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;
crates/rio/src/etag_reader.rs
+1 -51
View File
@@ -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<R> BlockReadable for EtagReader<R>
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";
crates/rio/src/hardlimit_reader.rs
-57
View File
@@ -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<R>, inner);
impl<R> BlockReadable for HardLimitReader<R>
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);
}
}
crates/rio/src/hash_reader.rs
+78 -162
View File
@@ -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<Option<Checksum>> {
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<usize> {
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<std::io::Result<()>> {
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<std::io::Result<()>> {
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";
crates/rio/src/http_reader.rs
+36 -68
View File
@@ -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<Box<dyn Stream<Item = std::io::Result<Bytes>> + Send + Sync>>;
async fn request_http_byte_stream(
url: String,
method: Method,
headers: HeaderMap,
body: Option<Vec<u8>>,
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<Vec<u8>>,
) -> io::Result<HttpByteStream> {
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<Vec<u8>>,
_read_buf_size: usize,
) -> io::Result<Self> {
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,
crates/rio/src/lib.rs
+2 -138
View File
@@ -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<Box<dyn Future<Output = std::io::Result<usize>> + 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<T> ReadStream for T where T: tokio::io::AsyncRead + Unpin + Send + Sync {}
pub trait ReaderCapabilities: EtagResolvable + HashReaderDetector + TryGetIndex {}
impl<T> ReaderCapabilities for T where T: EtagResolvable + HashReaderDetector + TryGetIndex {}
pub trait Reader: ReadStream + ReaderCapabilities + BlockReadable {}
impl<T> Reader for T where T: ReadStream + ReaderCapabilities + BlockReadable {}
pub trait Reader: ReadStream + ReaderCapabilities {}
impl<T> Reader for T where T: ReadStream + ReaderCapabilities {}
pub type DynReader = Box<dyn Reader>;
@@ -183,42 +154,6 @@ pub trait HashReaderDetector {
}
}
impl<R> BlockReadable for crate::WarpReader<R>
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<R> BlockReadable for tokio::io::BufReader<R>
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<R> BlockReadable for crate::LimitReader<R>
where
R: Reader,
{
fn read_block<'a>(&'a mut self, buf: &'a mut [u8]) -> BoxReadBlockFuture<'a> {
read_block_via_async_read(self, buf)
}
}
impl<R> BlockReadable for crate::DecryptReader<R>
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<R>(reader: R) -> DynReader
where
R: Reader + 'static,
@@ -263,74 +198,3 @@ where
self.as_ref().try_get_index()
}
}
impl<T> BlockReadable for Box<T>
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<u8>),
Error(ErrorKind),
Eof,
}
struct StepReader {
steps: VecDeque<ReadStep>,
}
impl StepReader {
fn new(steps: impl IntoIterator<Item = ReadStep>) -> 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<io::Result<()>> {
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);
}
}
rustfs/src/app/object_usecase/put_object_flow.rs
+2 -21
View File
@@ -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