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

revert: remove #2351 chunk I/O and object-io crate (phase 7) (#2543)

Browse Source 
Co-authored-by: houseme <housemecn@gmail.com>
This commit is contained in:
安正超
2026-04-15 10:54:41 +08:00
committed by GitHub
parent 16b9189e9b
commit 642d83f0e4
22 changed files with 2757 additions and 3815 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Cargo.lock
Generated
-26
View File
@@ -7730,7 +7730,6 @@ dependencies = [
"rustfs-metrics",
"rustfs-notify",
"rustfs-object-capacity",
"rustfs-object-io",
"rustfs-obs",
"rustfs-policy",
"rustfs-protocols",
@@ -8074,8 +8073,6 @@ name = "rustfs-io-core"
version = "0.0.5"
dependencies = [
"bytes",
"futures-core",
"futures-util",
"memmap2 0.9.10",
"rustfs-io-metrics",
"thiserror 2.0.18",
@@ -8264,29 +8261,6 @@ dependencies = [
"walkdir",
]
[[package]]
name = "rustfs-object-io"
version = "0.0.5"
dependencies = [
"astral-tokio-tar",
"atoi",
"bytes",
"futures-util",
"http 1.4.0",
"rustfs-ecstore",
"rustfs-io-core",
"rustfs-io-metrics",
"rustfs-rio",
"rustfs-s3select-api",
"rustfs-utils",
"s3s",
"thiserror 2.0.18",
"time",
"tokio",
"tokio-util",
"uuid",
]
[[package]]
name = "rustfs-obs"
version = "0.0.5"
Cargo.toml
-2
View File
@@ -52,7 +52,6 @@ members = [
"crates/workers", # Worker thread pools and task scheduling
"crates/io-metrics", # Zero-copy metrics collection for performance analysis
"crates/io-core", # Zero-copy core reader and writer implementations
"crates/object-io", # Object I/O policy and zero-copy helper primitives
"crates/zip", # ZIP file handling and compression
]
resolver = "3"
@@ -99,7 +98,6 @@ rustfs-metrics = { path = "crates/metrics", version = "0.0.5" }
rustfs-notify = { path = "crates/notify", version = "0.0.5" }
rustfs-io-metrics = { path = "crates/io-metrics", version = "0.0.5" }
rustfs-io-core = { path = "crates/io-core", version = "0.0.5" }
rustfs-object-io = { path = "crates/object-io", version = "0.0.5" }
rustfs-object-capacity = { path = "crates/object-capacity", version = "0.0.5" }
rustfs-obs = { path = "crates/obs", version = "0.0.5" }
rustfs-policy = { path = "crates/policy", version = "0.0.5" }
crates/ecstore/src/config/storageclass.rs
+12 -39
View File
@@ -150,7 +150,18 @@ impl Config {
return false;
}
shard_size as usize <= self.inline_shard_limit_bytes(versioned)
let shard_size = shard_size as usize;
let mut inline_block = DEFAULT_INLINE_BLOCK;
if self.initialized {
inline_block = self.inline_block;
}
if versioned {
shard_size <= inline_block / 8
} else {
shard_size <= inline_block
}
}
pub fn inline_block(&self) -> usize {
@@ -161,15 +172,6 @@ impl Config {
}
}
pub fn inline_shard_limit_bytes(&self, versioned: bool) -> usize {
let inline_block = self.inline_block();
if versioned { inline_block / 8 } else { inline_block }
}
pub fn inline_object_limit_bytes(&self, data_shards: usize, versioned: bool) -> usize {
self.inline_shard_limit_bytes(versioned).saturating_mul(data_shards.max(1))
}
pub fn capacity_optimized(&self) -> bool {
if !self.initialized {
false
@@ -334,32 +336,3 @@ pub fn validate_parity_inner(ss_parity: usize, rrs_parity: usize, set_drive_coun
}
Ok(())
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn inline_object_limit_matches_default_non_versioned_budget() {
let cfg = Config {
initialized: true,
inline_block: DEFAULT_INLINE_BLOCK,
..Default::default()
};
assert_eq!(cfg.inline_shard_limit_bytes(false), DEFAULT_INLINE_BLOCK);
assert_eq!(cfg.inline_object_limit_bytes(8, false), DEFAULT_INLINE_BLOCK * 8);
}
#[test]
fn inline_object_limit_scales_down_for_versioned_objects() {
let cfg = Config {
initialized: true,
inline_block: DEFAULT_INLINE_BLOCK,
..Default::default()
};
assert_eq!(cfg.inline_shard_limit_bytes(true), DEFAULT_INLINE_BLOCK / 8);
assert_eq!(cfg.inline_object_limit_bytes(8, true), DEFAULT_INLINE_BLOCK);
}
}
crates/ecstore/src/data_movement.rs
+4 -4
View File
@@ -64,7 +64,7 @@ pub fn decode_part_index(index: Option<&Bytes>) -> Option<Index> {
}
}
pub fn put_data_from_chunk(chunk: Vec<u8>, size: i64, actual_size: i64, index: Option<Index>) -> Result<PutObjReader> {
pub fn put_obj_reader_from_chunk(chunk: Vec<u8>, size: i64, actual_size: i64, index: Option<Index>) -> Result<PutObjReader> {
use sha2::{Digest, Sha256};
let sha256hex = if !chunk.is_empty() {
@@ -74,7 +74,7 @@ pub fn put_data_from_chunk(chunk: Vec<u8>, size: i64, actual_size: i64, index: O
};
let reader = IndexedDataMovementReader::new(Cursor::new(chunk), index);
let hash_reader = HashReader::from_reader(reader, size, actual_size, None, sha256hex, false)?;
let hash_reader = HashReader::from_stream(reader, size, actual_size, None, sha256hex, false)?;
Ok(PutObjReader::new(hash_reader))
}
@@ -172,7 +172,7 @@ pub(crate) async fn migrate_object(
let part_size = i64::try_from(part.size).map_err(|_| Error::other("part size overflow"))?;
let part_actual_size = if part.actual_size > 0 { part.actual_size } else { part_size };
let index = decode_part_index(part.index.as_ref());
let mut data = put_data_from_chunk(chunk, part_size, part_actual_size, index)?;
let mut data = put_obj_reader_from_chunk(chunk, part_size, part_actual_size, index)?;
let pi = match store
.put_object_part(
@@ -254,7 +254,7 @@ pub(crate) async fn migrate_object(
.first()
.and_then(|part| decode_part_index(part.index.as_ref()));
let reader = IndexedDataMovementReader::new(BufReader::new(rd.stream), index);
let hrd = HashReader::from_reader(reader, object_info.size, actual_size, object_info.etag.clone(), None, false)?;
let hrd = HashReader::from_stream(reader, object_info.size, actual_size, object_info.etag.clone(), None, false)?;
let mut data = PutObjReader::new(hrd);
if let Err(err) = store
crates/ecstore/src/set_disk.rs
+16 -167
View File
@@ -119,19 +119,18 @@ use tokio::{
time::{interval, timeout},
};
use tokio_util::sync::CancellationToken;
use tracing::error;
use tracing::{debug, info, warn};
use uuid::Uuid;
const ENV_RUSTFS_PUT_INLINE_OBJECT_MAX_BYTES: &str = "RUSTFS_PUT_INLINE_OBJECT_MAX_BYTES";
const ENV_RUSTFS_PUT_FORCE_DISABLE_INLINE: &str = "RUSTFS_PUT_FORCE_DISABLE_INLINE";
const SLOW_PUT_STORAGE_PHASE_DEBUG_THRESHOLD_MS: u64 = 100;
const SLOW_PUT_STORAGE_PHASE_WARN_THRESHOLD_MS: u64 = 1_000;
const SLOW_PUT_STORAGE_PHASE_ERROR_THRESHOLD_MS: u64 = 5_000;
pub const DEFAULT_READ_BUFFER_SIZE: usize = MI_B; // 1 MiB = 1024 * 1024;
pub const MAX_PARTS_COUNT: usize = 10000;
pub(crate) const RUSTFS_MULTIPART_BUCKET_KEY: &str = "x-rustfs-internal-multipart-bucket";
pub(crate) const RUSTFS_MULTIPART_OBJECT_KEY: &str = "x-rustfs-internal-multipart-object";
fn env_flag_enabled(name: &str) -> bool {
rustfs_utils::get_env_bool(name, false)
}
fn env_non_negative_usize(name: &str) -> Option<usize> {
rustfs_utils::get_env_opt_usize(name)
pub(crate) fn strip_internal_multipart_metadata(metadata: &mut HashMap<String, String>) {
metadata.remove(RUSTFS_MULTIPART_BUCKET_KEY);
metadata.remove(RUSTFS_MULTIPART_OBJECT_KEY);
}
fn capacity_scope_from_disks(disks: &[Option<DiskStore>]) -> CapacityScope {
@@ -164,65 +163,6 @@ fn record_capacity_scope_if_needed(scope_token: Option<Uuid>, disks: &[Option<Di
}
}
fn resolved_put_inline_buffer_enabled(object_size: i64, inline_by_topology: bool) -> bool {
if !inline_by_topology || object_size < 0 {
return false;
}
if env_flag_enabled(ENV_RUSTFS_PUT_FORCE_DISABLE_INLINE) {
return false;
}
env_non_negative_usize(ENV_RUSTFS_PUT_INLINE_OBJECT_MAX_BYTES)
.map(|value| usize::try_from(object_size).is_ok_and(|size| size <= value))
.unwrap_or(inline_by_topology)
}
fn log_put_storage_phase(
bucket: &str,
object: &str,
phase: &str,
elapsed: Duration,
object_size: i64,
inline_selected: bool,
write_quorum: usize,
) {
let duration_ms = elapsed.as_millis() as u64;
if duration_ms < SLOW_PUT_STORAGE_PHASE_DEBUG_THRESHOLD_MS {
return;
}
if duration_ms >= SLOW_PUT_STORAGE_PHASE_ERROR_THRESHOLD_MS {
error!(
phase,
duration_ms, object_size, inline_selected, write_quorum, bucket, object, "PUT storage phase is critically slow"
);
} else if duration_ms >= SLOW_PUT_STORAGE_PHASE_WARN_THRESHOLD_MS {
warn!(
phase,
duration_ms, object_size, inline_selected, write_quorum, bucket, object, "PUT storage phase is slow"
);
} else {
debug!(
phase,
duration_ms, object_size, inline_selected, write_quorum, bucket, object, "PUT storage phase exceeded debug threshold"
);
}
}
use tracing::error;
use tracing::{debug, info, warn};
use uuid::Uuid;
pub const DEFAULT_READ_BUFFER_SIZE: usize = MI_B; // 1 MiB = 1024 * 1024;
pub const MAX_PARTS_COUNT: usize = 10000;
pub(crate) const RUSTFS_MULTIPART_BUCKET_KEY: &str = "x-rustfs-internal-multipart-bucket";
pub(crate) const RUSTFS_MULTIPART_OBJECT_KEY: &str = "x-rustfs-internal-multipart-object";
pub(crate) fn strip_internal_multipart_metadata(metadata: &mut HashMap<String, String>) {
metadata.remove(RUSTFS_MULTIPART_BUCKET_KEY);
metadata.remove(RUSTFS_MULTIPART_OBJECT_KEY);
}
/// Get the duplex buffer size from environment variable or use default.
///
/// This function reads `RUSTFS_DUPLEX_BUFFER_SIZE` environment variable
@@ -872,18 +812,13 @@ impl ObjectIO for SetDisks {
let erasure = erasure_coding::Erasure::new(fi.erasure.data_blocks, fi.erasure.parity_blocks, fi.erasure.block_size);
let is_inline_buffer = {
let inline_by_topology = if let Some(sc) = GLOBAL_STORAGE_CLASS.get() {
if let Some(sc) = GLOBAL_STORAGE_CLASS.get() {
sc.should_inline(erasure.shard_file_size(data.size()), opts.versioned)
} else {
false
};
resolved_put_inline_buffer_enabled(data.size(), inline_by_topology)
}
};
if is_inline_buffer {
rustfs_io_metrics::record_put_inline_selected(data.size(), opts.versioned);
}
let writer_setup_start = Instant::now();
let mut writers = Vec::with_capacity(shuffle_disks.len());
let mut errors = Vec::with_capacity(shuffle_disks.len());
for disk_op in shuffle_disks.iter() {
@@ -917,15 +852,6 @@ impl ObjectIO for SetDisks {
writers.push(None);
}
}
log_put_storage_phase(
bucket,
object,
"writer_setup",
writer_setup_start.elapsed(),
data.size(),
is_inline_buffer,
write_quorum,
);
let nil_count = errors.iter().filter(|&e| e.is_none()).count();
if nil_count < write_quorum {
@@ -937,9 +863,6 @@ impl ObjectIO for SetDisks {
return Err(Error::other(format!("not enough disks to write: {errors:?}")));
}
let object_size = data.size();
let encode_write_start = Instant::now();
let stream = mem::replace(
&mut data.stream,
HashReader::from_stream(Cursor::new(Vec::new()), 0, 0, None, None, false)?,
@@ -948,30 +871,15 @@ impl ObjectIO for SetDisks {
let (reader, w_size) = match Arc::new(erasure).encode(stream, &mut writers, write_quorum).await {
Ok((r, w)) => (r, w),
Err(e) => {
log_put_storage_phase(
bucket,
object,
"encode_write",
encode_write_start.elapsed(),
object_size,
is_inline_buffer,
write_quorum,
);
error!("encode err {:?}", e);
return Err(e.into());
}
}; // TODO: delete temporary directory on error
let _ = mem::replace(&mut data.stream, reader);
log_put_storage_phase(
bucket,
object,
"encode_write",
encode_write_start.elapsed(),
object_size,
is_inline_buffer,
write_quorum,
);
// if let Err(err) = close_bitrot_writers(&mut writers).await {
// error!("close_bitrot_writers err {:?}", err);
// }
if (w_size as i64) < data.size() {
warn!("put_object write size < data.size(), w_size={}, data.size={}", w_size, data.size());
@@ -1047,7 +955,6 @@ impl ObjectIO for SetDisks {
drop(writers); // drop writers to close all files, this is to prevent FileAccessDenied errors when renaming data
let post_write_lock_start = Instant::now();
if !opts.no_lock && object_lock_guard.is_none() {
let ns_lock = self.new_ns_lock(bucket, object).await?;
object_lock_guard = Some(ns_lock.get_write_lock(get_lock_acquire_timeout()).await.map_err(|e| {
@@ -1057,17 +964,7 @@ impl ObjectIO for SetDisks {
))
})?);
}
log_put_storage_phase(
bucket,
object,
"post_write_lock",
post_write_lock_start.elapsed(),
data.size(),
is_inline_buffer,
write_quorum,
);
let finalize_start = Instant::now();
let (online_disks, _, op_old_dir) = Self::rename_data(
&shuffle_disks,
RUSTFS_META_TMP_BUCKET,
@@ -1077,18 +974,7 @@ impl ObjectIO for SetDisks {
object,
write_quorum,
)
.await
.inspect_err(|_| {
log_put_storage_phase(
bucket,
object,
"finalize",
finalize_start.elapsed(),
data.size(),
is_inline_buffer,
write_quorum,
);
})?;
.await?;
if let Some(old_dir) = op_old_dir {
self.commit_rename_data_dir(&online_disks, bucket, object, &old_dir.to_string(), write_quorum)
@@ -1098,15 +984,6 @@ impl ObjectIO for SetDisks {
drop(object_lock_guard); // drop object lock guard to release the lock
self.delete_all(RUSTFS_META_TMP_BUCKET, &tmp_dir).await?;
log_put_storage_phase(
bucket,
object,
"finalize",
finalize_start.elapsed(),
data.size(),
is_inline_buffer,
write_quorum,
);
for (i, op_disk) in online_disks.iter().enumerate() {
if let Some(disk) = op_disk
@@ -2037,9 +1914,6 @@ impl ObjectOperations for SetDisks {
if let Some(ref version_id) = opts.version_id {
fi.version_id = Uuid::parse_str(version_id).ok();
}
if let Some(checksum) = &opts.resolved_checksum {
fi.checksum = Some(checksum.clone());
}
self.update_object_meta(bucket, object, fi.clone(), &online_disks)
.await
@@ -4403,31 +4277,6 @@ mod tests {
}
}
#[test]
#[serial]
fn resolved_put_inline_buffer_enabled_honors_disable_env() {
temp_env::with_var(ENV_RUSTFS_PUT_FORCE_DISABLE_INLINE, Some("true"), || {
assert!(!resolved_put_inline_buffer_enabled(4096, true));
});
}
#[test]
#[serial]
fn resolved_put_inline_buffer_enabled_honors_max_bytes_override() {
temp_env::with_var(ENV_RUSTFS_PUT_INLINE_OBJECT_MAX_BYTES, Some("4096"), || {
assert!(resolved_put_inline_buffer_enabled(4096, true));
assert!(!resolved_put_inline_buffer_enabled(4097, true));
});
}
#[test]
#[serial]
fn resolved_put_inline_buffer_enabled_ignores_invalid_override() {
temp_env::with_var(ENV_RUSTFS_PUT_INLINE_OBJECT_MAX_BYTES, Some("invalid"), || {
assert!(resolved_put_inline_buffer_enabled(4096, true));
});
}
async fn current_setup_type() -> SetupType {
if is_dist_erasure().await {
SetupType::DistErasure
crates/ecstore/src/store_api/types.rs
-14
View File
@@ -70,7 +70,6 @@ pub struct ObjectOptions {
pub eval_metadata: Option<HashMap<String, String>>,
pub resolved_checksum: Option<Bytes>,
pub want_checksum: Option<Checksum>,
pub skip_verify_bitrot: bool,
pub capacity_scope_token: Option<Uuid>,
@@ -511,18 +510,6 @@ impl ObjectInfo {
})
.collect();
let actual_size = fi
.parts
.iter()
.map(|part| {
if part.actual_size > 0 {
part.actual_size
} else {
i64::try_from(part.size).unwrap_or_default()
}
})
.sum();
// TODO: part checksums
ObjectInfo {
@@ -535,7 +522,6 @@ impl ObjectInfo {
delete_marker: fi.deleted,
mod_time: fi.mod_time,
size: fi.size,
actual_size,
parts,
is_latest: fi.is_latest,
user_tags,
crates/io-core/Cargo.toml
-2
View File
@@ -29,14 +29,12 @@ workspace = true
[dependencies]
bytes = { workspace = true }
futures-core = { workspace = true }
thiserror = { workspace = true }
tokio = { workspace = true, features = ["io-util", "fs", "rt", "sync"] }
memmap2 = { workspace = true }
rustfs-io-metrics = { workspace = true }
[dev-dependencies]
futures-util = { workspace = true }
tokio = { workspace = true, features = ["rt-multi-thread", "macros"] }
[lib]
crates/io-core/src/adapter.rs
-124
View File
@@ -1,124 +0,0 @@
// Copyright 2024 RustFS Team
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
//! Compatibility adapter that exposes chunk streams as `AsyncRead`.
use crate::chunk::BoxChunkStream;
use bytes::Bytes;
use std::io;
use std::pin::Pin;
use std::task::{Context, Poll};
use tokio::io::{AsyncRead, ReadBuf};
/// `AsyncRead` adapter for boxed chunk streams.
pub struct ChunkStreamReader {
stream: BoxChunkStream,
current: Option<Bytes>,
offset: usize,
}
impl ChunkStreamReader {
#[must_use]
pub fn new(stream: BoxChunkStream) -> Self {
Self {
stream,
current: None,
offset: 0,
}
}
}
impl AsyncRead for ChunkStreamReader {
fn poll_read(mut self: Pin<&mut Self>, cx: &mut Context<'_>, buf: &mut ReadBuf<'_>) -> Poll<io::Result<()>> {
loop {
if let Some(current) = &self.current {
if self.offset < current.len() {
let remaining = &current[self.offset..];
let to_read = remaining.len().min(buf.remaining());
buf.put_slice(&remaining[..to_read]);
self.offset += to_read;
return Poll::Ready(Ok(()));
}
self.current = None;
self.offset = 0;
continue;
}
match self.stream.as_mut().poll_next(cx) {
Poll::Pending => return Poll::Pending,
Poll::Ready(Some(Ok(chunk))) => {
let next = chunk.as_bytes();
if next.is_empty() {
continue;
}
self.current = Some(next);
self.offset = 0;
}
Poll::Ready(Some(Err(err))) => return Poll::Ready(Err(err)),
Poll::Ready(None) => return Poll::Ready(Ok(())),
}
}
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::chunk::{IoChunk, MappedChunk, PooledChunk};
use bytes::Bytes;
use futures_util::stream;
use tokio::io::AsyncReadExt;
#[tokio::test]
async fn test_chunk_stream_reader_reads_single_chunk() {
let stream: BoxChunkStream = Box::pin(stream::iter(vec![Ok(IoChunk::Shared(Bytes::from_static(b"hello")))]));
let mut reader = ChunkStreamReader::new(stream);
let mut out = Vec::new();
reader.read_to_end(&mut out).await.unwrap();
assert_eq!(out, b"hello");
}
#[tokio::test]
async fn test_chunk_stream_reader_reads_multiple_chunks() {
let stream: BoxChunkStream = Box::pin(stream::iter(vec![
Ok(IoChunk::Shared(Bytes::from_static(b"he"))),
Ok(IoChunk::Mapped(MappedChunk::new(Bytes::from_static(b"llo!"), 0, 4).unwrap())),
Ok(IoChunk::Pooled(PooledChunk::from_bytes(Bytes::from_static(b" world")).unwrap())),
]));
let mut reader = ChunkStreamReader::new(stream);
let mut out = Vec::new();
reader.read_to_end(&mut out).await.unwrap();
assert_eq!(out, b"hello! world");
}
#[tokio::test]
async fn test_chunk_stream_reader_handles_empty_stream() {
let stream: BoxChunkStream = Box::pin(stream::iter(Vec::<io::Result<IoChunk>>::new()));
let mut reader = ChunkStreamReader::new(stream);
let mut out = Vec::new();
reader.read_to_end(&mut out).await.unwrap();
assert!(out.is_empty());
}
#[tokio::test]
async fn test_chunk_stream_reader_propagates_stream_error() {
let stream: BoxChunkStream = Box::pin(stream::iter(vec![Err(io::Error::other("chunk stream failure"))]));
let mut reader = ChunkStreamReader::new(stream);
let mut out = Vec::new();
let err = reader.read_to_end(&mut out).await.unwrap_err();
assert_eq!(err.kind(), io::ErrorKind::Other);
assert!(err.to_string().contains("chunk stream failure"));
}
}
crates/io-core/src/chunk.rs
-276
View File
@@ -1,276 +0,0 @@
// Copyright 2024 RustFS Team
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
//! Core chunk ownership abstractions for the zero-copy data plane.
use crate::pool::PooledBuffer;
use bytes::Bytes;
use futures_core::Stream;
use std::io;
use std::pin::Pin;
/// Boxed asynchronous stream of I/O chunks.
pub type BoxChunkStream = Pin<Box<dyn Stream<Item = io::Result<IoChunk>> + Send + Sync + 'static>>;
/// Source of chunked data.
pub trait ChunkSource {
fn into_chunk_stream(self) -> BoxChunkStream
where
Self: Sized;
}
/// Owned chunk variants used by the zero-copy data plane.
#[derive(Debug)]
pub enum IoChunk {
Shared(Bytes),
Mapped(MappedChunk),
Pooled(PooledChunk),
}
impl IoChunk {
/// Returns the visible length of this chunk.
#[must_use]
pub fn len(&self) -> usize {
match self {
Self::Shared(bytes) => bytes.len(),
Self::Mapped(chunk) => chunk.len(),
Self::Pooled(chunk) => chunk.len(),
}
}
/// Returns true when the chunk has no visible data.
#[must_use]
pub fn is_empty(&self) -> bool {
self.len() == 0
}
/// Returns a shared read-only view of the visible bytes.
#[must_use]
pub fn as_bytes(&self) -> Bytes {
match self {
Self::Shared(bytes) => bytes.clone(),
Self::Mapped(chunk) => chunk.as_bytes(),
Self::Pooled(chunk) => chunk.as_bytes(),
}
}
/// Returns a sliced view relative to the currently visible bytes.
pub fn slice(&self, offset: usize, len: usize) -> io::Result<Self> {
match self {
Self::Shared(bytes) => {
validate_slice_bounds(bytes.len(), offset, len)?;
Ok(Self::Shared(bytes.slice(offset..offset + len)))
}
Self::Mapped(chunk) => chunk.slice(offset, len).map(Self::Mapped),
Self::Pooled(chunk) => chunk.slice(offset, len).map(Self::Pooled),
}
}
}
/// Logical view into mapped file bytes.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct MappedChunk {
bytes: Bytes,
logical_offset: usize,
logical_len: usize,
}
impl MappedChunk {
pub fn new(bytes: Bytes, logical_offset: usize, logical_len: usize) -> io::Result<Self> {
validate_slice_bounds(bytes.len(), logical_offset, logical_len)?;
Ok(Self {
bytes,
logical_offset,
logical_len,
})
}
/// Returns the visible length of this mapped chunk.
#[must_use]
pub const fn len(&self) -> usize {
self.logical_len
}
/// Returns true when the chunk has no visible data.
#[must_use]
pub const fn is_empty(&self) -> bool {
self.logical_len == 0
}
/// Returns the visible bytes for this logical view.
#[must_use]
pub fn as_bytes(&self) -> Bytes {
self.bytes
.slice(self.logical_offset..self.logical_offset.saturating_add(self.logical_len))
}
/// Returns a sliced logical view relative to the current logical view.
pub fn slice(&self, offset: usize, len: usize) -> io::Result<Self> {
validate_slice_bounds(self.logical_len, offset, len)?;
Self::new(self.bytes.clone(), self.logical_offset + offset, len)
}
}
/// Placeholder pooled chunk variant for commit 4.
///
/// This is backed by a `PooledBuffer` and exposes a visible read-only window.
#[derive(Debug)]
pub struct PooledChunk {
bytes: Bytes,
}
#[derive(Debug)]
struct PooledChunkOwner {
buffer: PooledBuffer,
visible_len: usize,
}
impl AsRef<[u8]> for PooledChunkOwner {
fn as_ref(&self) -> &[u8] {
&self.buffer[..self.visible_len]
}
}
#[derive(Debug)]
struct DetachedVecChunkOwner {
bytes: Vec<u8>,
}
impl AsRef<[u8]> for DetachedVecChunkOwner {
fn as_ref(&self) -> &[u8] {
&self.bytes
}
}
impl PooledChunk {
pub fn new(buffer: PooledBuffer, len: usize) -> io::Result<Self> {
validate_slice_bounds(buffer.len(), 0, len)?;
Ok(Self {
bytes: Bytes::from_owner(PooledChunkOwner {
buffer,
visible_len: len,
}),
})
}
/// Convenience constructor for detached test and compatibility values.
pub fn from_bytes(bytes: Bytes) -> io::Result<Self> {
let len = bytes.len();
Self::new(PooledBuffer::from_bytes(bytes), len)
}
/// Detached constructor that takes ownership of an existing `Vec<u8>`
/// without introducing an additional copy.
pub fn from_vec(bytes: Vec<u8>) -> Self {
Self {
bytes: Bytes::from_owner(DetachedVecChunkOwner { bytes }),
}
}
/// Returns the visible length of this pooled chunk.
#[must_use]
pub fn len(&self) -> usize {
self.bytes.len()
}
/// Returns true when the chunk has no visible data.
#[must_use]
pub fn is_empty(&self) -> bool {
self.bytes.is_empty()
}
/// Returns the visible bytes for this pooled chunk.
#[must_use]
pub fn as_bytes(&self) -> Bytes {
self.bytes.clone()
}
/// Returns a sliced pooled chunk relative to the current visible view.
pub fn slice(&self, offset: usize, len: usize) -> io::Result<Self> {
validate_slice_bounds(self.bytes.len(), offset, len)?;
Ok(Self {
bytes: self.bytes.slice(offset..offset + len),
})
}
}
fn validate_slice_bounds(visible_len: usize, offset: usize, len: usize) -> io::Result<()> {
let end = offset
.checked_add(len)
.ok_or_else(|| io::Error::new(io::ErrorKind::InvalidInput, "chunk slice overflows"))?;
if end > visible_len {
return Err(io::Error::new(io::ErrorKind::InvalidInput, "chunk slice exceeds visible length"));
}
Ok(())
}
#[cfg(test)]
mod tests {
use super::*;
use crate::pool::BytesPool;
#[test]
fn test_shared_chunk_len_and_slice() {
let chunk = IoChunk::Shared(Bytes::from_static(b"abcdef"));
assert_eq!(chunk.len(), 6);
assert!(!chunk.is_empty());
assert_eq!(chunk.as_bytes(), Bytes::from_static(b"abcdef"));
assert_eq!(chunk.slice(1, 3).unwrap().as_bytes(), Bytes::from_static(b"bcd"));
}
#[test]
fn test_mapped_chunk_len_and_slice() {
let chunk = MappedChunk::new(Bytes::from_static(b"abcdefgh"), 2, 4).unwrap();
assert_eq!(chunk.len(), 4);
assert_eq!(chunk.as_bytes(), Bytes::from_static(b"cdef"));
assert_eq!(chunk.slice(1, 2).unwrap().as_bytes(), Bytes::from_static(b"de"));
}
#[test]
fn test_pooled_chunk_len_and_as_bytes() {
let chunk = PooledChunk::from_bytes(Bytes::from_static(b"hello")).unwrap();
assert_eq!(chunk.len(), 5);
assert_eq!(chunk.as_bytes(), Bytes::from_static(b"hello"));
assert_eq!(chunk.slice(1, 3).unwrap().as_bytes(), Bytes::from_static(b"ell"));
}
#[test]
fn test_io_chunk_as_bytes_for_all_variants() {
let shared = IoChunk::Shared(Bytes::from_static(b"s"));
let mapped = IoChunk::Mapped(MappedChunk::new(Bytes::from_static(b"mapped"), 0, 6).unwrap());
let pooled = IoChunk::Pooled(PooledChunk::from_bytes(Bytes::from_static(b"p")).unwrap());
assert_eq!(shared.as_bytes(), Bytes::from_static(b"s"));
assert_eq!(mapped.as_bytes(), Bytes::from_static(b"mapped"));
assert_eq!(pooled.as_bytes(), Bytes::from_static(b"p"));
}
#[tokio::test]
async fn test_pooled_chunk_keeps_owner_alive_until_last_view_drops() {
let pool = BytesPool::new_tiered();
let mut buffer = pool.acquire_buffer(16).await;
buffer.extend_from_slice(b"pooled-bytes");
let chunk = PooledChunk::new(buffer, "pooled-bytes".len()).unwrap();
let bytes = chunk.as_bytes();
assert_eq!(pool.available_buffers(), 0);
drop(chunk);
assert_eq!(pool.available_buffers(), 0);
assert_eq!(bytes, Bytes::from_static(b"pooled-bytes"));
drop(bytes);
assert_eq!(pool.available_buffers(), 1);
}
}
crates/io-core/src/lib.rs
-4
View File
@@ -46,10 +46,8 @@
//! let mut buffer = pool.acquire_buffer(8192).await;
//! ```
pub mod adapter;
pub mod backpressure;
pub mod bufreader_optimizer;
pub mod chunk;
pub mod config;
pub mod deadlock_detector;
pub mod direct_io;
@@ -70,9 +68,7 @@ pub use reader::{ZeroCopyObjectReader, ZeroCopyReadError};
pub use writer::{ZeroCopyObjectWriter, ZeroCopyWriteError};
// BufReader optimizer exports
pub use adapter::ChunkStreamReader;
pub use bufreader_optimizer::{BufReaderConfig, BufReaderOptimizer, BufReaderStats, BufferedSource};
pub use chunk::{BoxChunkStream, ChunkSource, IoChunk, MappedChunk, PooledChunk};
// Shared memory exports
pub use shared_memory::{ArcData, ArcMetadata, SharedMemoryConfig, SharedMemoryPool, SharedMemoryStats};
crates/io-core/src/pool.rs
+1 -41
View File
@@ -17,7 +17,7 @@
//! Migrated from rustfs-ecstore to provide unified buffer pooling
//! across rustfs and rustfs-ecstore without cyclic dependencies.
use bytes::{Bytes, BytesMut};
use bytes::BytesMut;
use std::mem::ManuallyDrop;
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::{Arc, Mutex};
@@ -108,7 +108,6 @@ pub struct BytesPoolMetrics {
/// A buffer managed by the BytesPool.
///
/// When dropped, the buffer is automatically returned to the pool for reuse.
#[derive(Debug)]
pub struct PooledBuffer {
/// The underlying buffer (ManuallyDrop to allow taking on drop)
pub buffer: ManuallyDrop<BytesMut>,
@@ -118,45 +117,6 @@ pub struct PooledBuffer {
_permit: Option<OwnedSemaphorePermit>,
}
impl PooledBuffer {
/// Create a detached pooled buffer from bytes.
///
/// This is primarily used for tests and transitional adapters where the
/// chunk abstraction needs a pool-shaped owner before a real pool-backed
/// producer exists.
#[must_use]
pub fn from_bytes(bytes: Bytes) -> Self {
Self {
buffer: ManuallyDrop::new(BytesMut::from(bytes.as_ref())),
tier: None,
_permit: None,
}
}
/// Current visible length of the underlying buffer.
#[must_use]
pub fn len(&self) -> usize {
self.buffer.len()
}
/// Total buffer capacity.
#[must_use]
pub fn capacity(&self) -> usize {
self.buffer.capacity()
}
/// Clear the visible contents while preserving capacity.
pub fn clear(&mut self) {
self.buffer.clear();
}
/// Returns true when the visible buffer is empty.
#[must_use]
pub fn is_empty(&self) -> bool {
self.buffer.is_empty()
}
}
/// BytesPool configuration.
///
/// Allows customization of buffer sizes and limits for each tier.
crates/io-metrics/src/lib.rs
+225 -401
View File
@@ -121,129 +121,8 @@ pub use config::{
// Re-exports for convenience
pub use collector::MetricsCollector;
pub use metric_names::data_plane;
pub use performance::PerformanceMetrics;
/// High-level request path selected for an I/O operation.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum IoPath {
Fast,
Legacy,
}
impl IoPath {
#[must_use]
pub const fn as_str(self) -> &'static str {
match self {
Self::Fast => "fast",
Self::Legacy => "legacy",
}
}
}
/// Effective copy mode observed for an I/O operation.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum CopyMode {
TrueZeroCopy,
SharedBytes,
SingleCopy,
Reconstructed,
Transformed,
}
impl CopyMode {
#[must_use]
pub const fn as_str(self) -> &'static str {
match self {
Self::TrueZeroCopy => "true_zero_copy",
Self::SharedBytes => "shared_bytes",
Self::SingleCopy => "single_copy",
Self::Reconstructed => "reconstructed",
Self::Transformed => "transformed",
}
}
}
/// Stage where a data plane decision or fallback happened.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum IoStage {
Unknown,
ReadSetup,
HttpBridge,
LocalDiskChunk,
RangeGuard,
PutTransform,
}
impl IoStage {
#[must_use]
pub const fn as_str(self) -> &'static str {
match self {
Self::Unknown => "unknown",
Self::ReadSetup => "read_setup",
Self::HttpBridge => "http_bridge",
Self::LocalDiskChunk => "local_disk_chunk",
Self::RangeGuard => "range_guard",
Self::PutTransform => "put_transform",
}
}
}
/// Reason why the data plane fell back from a preferred path.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum FallbackReason {
Unknown,
ProbeFailed,
MmapDisabled,
MmapUnavailable,
SmallObject,
WindowLimitExceeded,
UnalignedWindow,
RangeNotSupported,
EncryptionEnabled,
CompressionEnabled,
TransformEncryptionLegacy,
TransformCompressionLegacy,
TransformCompressionEncryptionLegacy,
ChunkBridgeUnavailable,
NonLocalBackend,
}
impl FallbackReason {
#[must_use]
pub const fn as_str(self) -> &'static str {
match self {
Self::Unknown => "unknown",
Self::ProbeFailed => "probe_failed",
Self::MmapDisabled => "mmap_disabled",
Self::MmapUnavailable => "mmap_unavailable",
Self::SmallObject => "small_object",
Self::WindowLimitExceeded => "window_limit_exceeded",
Self::UnalignedWindow => "unaligned_window",
Self::RangeNotSupported => "range_not_supported",
Self::EncryptionEnabled => "encryption_enabled",
Self::CompressionEnabled => "compression_enabled",
Self::TransformEncryptionLegacy => "transform_encryption_legacy",
Self::TransformCompressionLegacy => "transform_compression_legacy",
Self::TransformCompressionEncryptionLegacy => "transform_compression_encryption_legacy",
Self::ChunkBridgeUnavailable => "chunk_bridge_unavailable",
Self::NonLocalBackend => "non_local_backend",
}
}
}
#[inline(always)]
fn put_size_bucket_label(size_bytes: i64) -> &'static str {
match size_bytes {
..=0 => "unknown",
1..=16_384 => "le_16kib",
16_385..=65_536 => "le_64kib",
65_537..=262_144 => "le_256kib",
262_145..=1_048_576 => "le_1mib",
_ => "gt_1mib",
}
}
/// Record GetObject request start.
#[inline(always)]
pub fn record_get_object_request_start(concurrent_requests: usize) {
@@ -316,138 +195,40 @@ pub fn record_get_object_io_state(
counter!("rustfs_io_strategy_selected_total", "level" => load_level.to_string()).increment(1);
}
/// Record which request path was selected for an operation.
/// Record a zero-copy read operation.
///
/// # 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_io_path_selected(operation: &'static str, io_path: IoPath) {
counter!(
metric_names::data_plane::PATH_SELECTED_TOTAL,
"path" => operation,
"mode" => io_path.as_str()
)
.increment(1);
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 the effective copy mode for an operation.
/// Record memory copies avoided by using zero-copy.
///
/// # Arguments
///
/// * `bytes_saved` - Number of bytes that would have been copied without zero-copy
#[inline(always)]
pub fn record_io_copy_mode(operation: &'static str, copy_mode: CopyMode, size_bytes: usize) {
counter!(
metric_names::data_plane::COPY_MODE_BYTES_TOTAL,
"path" => operation,
"mode" => copy_mode.as_str()
)
.increment(size_bytes as u64);
pub fn record_memory_copy_saved(bytes_saved: usize) {
counter!("rustfs.zero_copy.memory.saved.bytes").increment(bytes_saved as u64);
}
/// Record a data plane fallback decision.
/// 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_io_fallback(stage: IoStage, reason: FallbackReason) {
counter!(
metric_names::data_plane::FALLBACK_TOTAL,
"stage" => stage.as_str(),
"reason" => reason.as_str()
)
.increment(1);
}
/// Record the currently active mmap bytes held by LocalDisk chunk streams.
#[inline(always)]
pub fn record_local_disk_active_mmap_bytes(active_bytes: usize) {
gauge!(metric_names::data_plane::LOCAL_DISK_ACTIVE_MMAP_BYTES).set(active_bytes as f64);
}
/// Record pooled chunk usage in LocalDisk compatibility paths.
#[inline(always)]
pub fn record_local_disk_pooled_chunk(source: &'static str, size_bytes: usize) {
counter!(
metric_names::data_plane::LOCAL_DISK_POOLED_CHUNKS_TOTAL,
"source" => source
)
.increment(1);
counter!(
metric_names::data_plane::LOCAL_DISK_POOLED_BYTES_TOTAL,
"source" => source
)
.increment(size_bytes as u64);
}
/// Record a compatibility chunk-stream aggregation performed by `read_file_zero_copy()`.
#[inline(always)]
pub fn record_local_disk_compat_collect(chunk_count: usize, total_bytes: usize) {
counter!(metric_names::data_plane::LOCAL_DISK_COMPAT_COLLECT_TOTAL).increment(1);
histogram!(metric_names::data_plane::LOCAL_DISK_COMPAT_COLLECT_CHUNKS).record(chunk_count as f64);
histogram!(metric_names::data_plane::LOCAL_DISK_COMPAT_COLLECT_BYTES).record(total_bytes as f64);
}
/// Record an attempted PUT fast path.
#[inline(always)]
pub fn record_put_object_attempted_fast_path(size_bytes: i64) {
counter!(metric_names::data_plane::PUT_FAST_PATH_ATTEMPTS_TOTAL).increment(1);
if size_bytes > 0 {
histogram!(metric_names::data_plane::PUT_FAST_PATH_ATTEMPT_SIZE_BYTES).record(size_bytes as f64);
}
}
/// Record which transformed PUT pipeline was selected.
#[inline(always)]
pub fn record_put_transform_selected(kind: &'static str, io_path: IoPath, size_bytes: usize) {
counter!(
metric_names::data_plane::PUT_TRANSFORM_SELECTED_TOTAL,
"kind" => kind,
"mode" => io_path.as_str()
)
.increment(1);
histogram!(
metric_names::data_plane::PUT_TRANSFORM_SIZE_BYTES,
"kind" => kind,
"mode" => io_path.as_str()
)
.record(size_bytes as f64);
}
/// Record PUT path selection with size-bucket context.
#[inline(always)]
pub fn record_put_path_selected(size_bytes: i64, io_path: IoPath) {
counter!(
"rustfs.s3.put_object.path.selected.total",
"mode" => io_path.as_str(),
"size_bucket" => put_size_bucket_label(size_bytes)
)
.increment(1);
}
/// Record PUT copy mode with size-bucket context.
#[inline(always)]
pub fn record_put_copy_mode(size_bytes: i64, copy_mode: CopyMode) {
counter!(
"rustfs.s3.put_object.copy_mode.total",
"mode" => copy_mode.as_str(),
"size_bucket" => put_size_bucket_label(size_bytes)
)
.increment(1);
}
/// Record PUT fallback with size-bucket context.
#[inline(always)]
pub fn record_put_fallback(size_bytes: i64, reason: FallbackReason) {
counter!(
"rustfs.s3.put_object.fallback.total",
"reason" => reason.as_str(),
"size_bucket" => put_size_bucket_label(size_bytes)
)
.increment(1);
}
/// Record inline-object selection for PUT with size-bucket context.
#[inline(always)]
pub fn record_put_inline_selected(size_bytes: i64, versioned: bool) {
counter!(
"rustfs.s3.put_object.inline.selected.total",
"versioned" => if versioned { "true" } else { "false" },
"size_bucket" => put_size_bucket_label(size_bytes)
)
.increment(1);
pub fn record_zero_copy_fallback(reason: &str) {
counter!("rustfs.zero_copy.fallback.total", "reason" => reason.to_string()).increment(1);
}
// ============================================================================
@@ -505,6 +286,41 @@ pub fn record_bytes_pool_hit_rate(tier: &str, hit_rate: f64) {
gauge!("rustfs.bytes.pool.hit.rate", "tier" => tier.to_string()).set(hit_rate * 100.0);
}
/// Record zero-copy write operation.
///
/// # Arguments
///
/// * `size_bytes` - Size of the data written in bytes
/// * `duration_ms` - Time taken for the write operation in milliseconds
#[inline(always)]
pub fn record_zero_copy_write(size_bytes: usize, duration_ms: f64) {
counter!("rustfs.zero_copy.write.total").increment(1);
histogram!("rustfs.zero_copy.write.size.bytes").record(size_bytes as f64);
histogram!("rustfs.zero_copy.write.duration.ms").record(duration_ms);
}
/// Record zero-copy write fallback.
///
/// This happens when zero-copy write fails and the system falls back to regular I/O.
///
/// # Arguments
///
/// * `reason` - Reason for the fallback
#[inline(always)]
pub fn record_zero_copy_write_fallback(reason: &str) {
counter!("rustfs.zero_copy.write.fallback.total", "reason" => reason.to_string()).increment(1);
}
/// Record bytes saved from zero-copy.
///
/// # Arguments
///
/// * `size_bytes` - Number of bytes saved from zero-copy
#[inline(always)]
pub fn record_bytes_saved(size_bytes: usize) {
counter!("rustfs.zero_copy.bytes.saved.total").increment(size_bytes as u64);
}
// ============================================================================
// S3 Operation Metrics (GetObject, PutObject, etc.)
// ============================================================================
@@ -534,33 +350,14 @@ pub fn record_get_object(duration_ms: f64, size_bytes: i64) {
///
/// * `duration_ms` - Operation duration in milliseconds
/// * `size_bytes` - Object size in bytes
/// * `zero_copy_enabled` - Legacy aggregate flag preserved for compatibility
///
/// Note: this function records aggregate S3 PUT metrics only. The definitive
/// outcome of request-level fast-path attempts must be tracked separately via
/// ADR 0001 data-plane helpers.
/// * `zero_copy_enabled` - Whether zero-copy was enabled for this operation
#[inline(always)]
pub fn record_put_object(duration_ms: f64, size_bytes: i64, zero_copy_enabled: bool) {
counter!("rustfs.s3.put_object.total").increment(1);
histogram!("rustfs.s3.put_object.duration.ms").record(duration_ms);
counter!(
"rustfs.s3.put_object.bucketed.total",
"size_bucket" => put_size_bucket_label(size_bytes)
)
.increment(1);
histogram!(
"rustfs.s3.put_object.bucketed.duration.ms",
"size_bucket" => put_size_bucket_label(size_bytes)
)
.record(duration_ms);
if size_bytes > 0 {
histogram!("rustfs.s3.put_object.size.bytes").record(size_bytes as f64);
histogram!(
"rustfs.s3.put_object.bucketed.size.bytes",
"size_bucket" => put_size_bucket_label(size_bytes)
)
.record(size_bytes as f64);
}
if zero_copy_enabled {
@@ -855,146 +652,15 @@ 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::*;
#[test]
fn test_io_path_as_str_values_stable() {
assert_eq!(IoPath::Fast.as_str(), "fast");
assert_eq!(IoPath::Legacy.as_str(), "legacy");
}
#[test]
fn test_copy_mode_as_str_values_stable() {
assert_eq!(CopyMode::TrueZeroCopy.as_str(), "true_zero_copy");
assert_eq!(CopyMode::SharedBytes.as_str(), "shared_bytes");
assert_eq!(CopyMode::SingleCopy.as_str(), "single_copy");
assert_eq!(CopyMode::Reconstructed.as_str(), "reconstructed");
assert_eq!(CopyMode::Transformed.as_str(), "transformed");
}
#[test]
fn test_fallback_reason_as_str_values_stable() {
assert_eq!(FallbackReason::Unknown.as_str(), "unknown");
assert_eq!(FallbackReason::ProbeFailed.as_str(), "probe_failed");
assert_eq!(FallbackReason::MmapDisabled.as_str(), "mmap_disabled");
assert_eq!(FallbackReason::MmapUnavailable.as_str(), "mmap_unavailable");
assert_eq!(FallbackReason::SmallObject.as_str(), "small_object");
assert_eq!(FallbackReason::WindowLimitExceeded.as_str(), "window_limit_exceeded");
assert_eq!(FallbackReason::UnalignedWindow.as_str(), "unaligned_window");
assert_eq!(FallbackReason::RangeNotSupported.as_str(), "range_not_supported");
assert_eq!(FallbackReason::EncryptionEnabled.as_str(), "encryption_enabled");
assert_eq!(FallbackReason::CompressionEnabled.as_str(), "compression_enabled");
assert_eq!(FallbackReason::TransformEncryptionLegacy.as_str(), "transform_encryption_legacy");
assert_eq!(FallbackReason::TransformCompressionLegacy.as_str(), "transform_compression_legacy");
assert_eq!(
FallbackReason::TransformCompressionEncryptionLegacy.as_str(),
"transform_compression_encryption_legacy"
);
assert_eq!(FallbackReason::ChunkBridgeUnavailable.as_str(), "chunk_bridge_unavailable");
assert_eq!(FallbackReason::NonLocalBackend.as_str(), "non_local_backend");
}
#[test]
fn test_record_io_path_selected() {
record_io_path_selected("get", IoPath::Fast);
record_io_path_selected("put", IoPath::Legacy);
}
#[test]
fn test_record_io_copy_mode() {
record_io_copy_mode("get", CopyMode::SharedBytes, 1024);
record_io_copy_mode("put", CopyMode::Transformed, 2048);
}
#[test]
fn test_record_io_fallback() {
record_io_fallback(IoStage::ReadSetup, FallbackReason::MmapUnavailable);
record_io_fallback(IoStage::HttpBridge, FallbackReason::ChunkBridgeUnavailable);
}
#[test]
fn test_record_local_disk_active_mmap_bytes() {
record_local_disk_active_mmap_bytes(4096);
record_local_disk_active_mmap_bytes(0);
}
#[test]
fn test_record_local_disk_pooled_chunk() {
record_local_disk_pooled_chunk("fallback", 4096);
record_local_disk_pooled_chunk("compat_collect", 8192);
}
#[test]
fn test_record_local_disk_compat_collect() {
record_local_disk_compat_collect(3, 16384);
}
#[test]
fn test_record_put_object_attempted_fast_path() {
record_put_object_attempted_fast_path(1024 * 1024);
record_put_object_attempted_fast_path(0);
}
#[test]
fn test_record_put_transform_selected() {
record_put_transform_selected("compression", IoPath::Fast, 2048);
record_put_transform_selected("compression_encryption", IoPath::Legacy, 4096);
}
#[test]
fn test_record_put_path_selected() {
record_put_path_selected(8 * 1024, IoPath::Fast);
record_put_path_selected(2 * 1024 * 1024, IoPath::Legacy);
}
#[test]
fn test_record_put_copy_mode() {
record_put_copy_mode(8 * 1024, CopyMode::SingleCopy);
record_put_copy_mode(512 * 1024, CopyMode::Transformed);
}
#[test]
fn test_record_put_fallback() {
record_put_fallback(32 * 1024, FallbackReason::CompressionEnabled);
record_put_fallback(2 * 1024 * 1024, FallbackReason::EncryptionEnabled);
}
#[test]
fn test_record_put_inline_selected() {
record_put_inline_selected(8 * 1024, false);
record_put_inline_selected(32 * 1024, true);
fn test_record_zero_copy_read() {
record_zero_copy_read(1024, 10.5);
record_memory_copy_saved(1024);
record_zero_copy_fallback("test");
}
#[test]
@@ -1005,6 +671,13 @@ mod tests {
record_bytes_pool_hit_rate("small", 0.85);
}
#[test]
fn test_record_zero_copy_write() {
record_zero_copy_write(1024, 10.5);
record_zero_copy_write_fallback("test");
record_bytes_saved(1024);
}
// S3 Operation Metrics Tests
#[test]
fn test_record_get_object() {
@@ -1109,6 +782,157 @@ mod tests {
}
}
// ============================================================================
// Zero-Copy Optimization Metrics (Phase 1 Extension)
// ============================================================================
pub mod bandwidth;
pub mod global_metrics;
pub mod metric_names;
pub use metric_names::zero_copy;
/// Record a zero-copy buffer operation.
///
/// This function records metrics for zero-copy buffer operations,
/// including the operation type and size.
#[inline(always)]
pub fn record_zero_copy_buffer_operation(operation: &str, size: usize) {
counter!(
zero_copy::BUFFER_OPERATIONS_TOTAL,
"operation" => operation.to_string()
)
.increment(1);
counter!(
zero_copy::BUFFER_BYTES_TOTAL,
"operation" => operation.to_string()
)
.increment(size as u64);
}
/// Record memory copy operations.
///
/// This function tracks the number and size of memory copies,
/// which should be minimized in zero-copy paths.
#[inline(always)]
pub fn record_memory_copy(count: u32, size: usize) {
counter!(zero_copy::MEMORY_COPY_TOTAL).increment(count as u64);
counter!(zero_copy::MEMORY_COPY_BYTES_TOTAL).increment(size as u64);
histogram!("rustfs_memory_copy_size_bytes").record(size as f64);
}
/// Record a shared reference operation.
///
/// This function tracks operations that create or use shared references
/// for zero-copy data sharing.
#[inline(always)]
pub fn record_shared_ref_operation(operation: &str) {
counter!(
zero_copy::SHARED_REF_OPERATIONS_TOTAL,
"operation" => operation.to_string()
)
.increment(1);
}
/// Record BufReader optimization.
///
/// This function tracks BufReader layer elimination and buffer size
/// adjustments.
#[inline(always)]
pub fn record_bufreader_optimization(layers_eliminated: u32, buffer_size: usize) {
counter!(zero_copy::BUFREADER_LAYERS_ELIMINATED_TOTAL).increment(layers_eliminated as u64);
histogram!(zero_copy::BUFREADER_BUFFER_SIZE_BYTES).record(buffer_size as f64);
}
/// Record Direct I/O operation.
///
/// This function tracks Direct I/O operations and their success/fallback
/// status.
#[inline(always)]
pub fn record_direct_io_operation(operation: &str, size: usize, success: bool) {
let status = if success { "success" } else { "fallback" };
counter!(
zero_copy::DIRECT_IO_OPERATIONS_TOTAL,
"operation" => operation.to_string(),
"status" => status.to_string()
)
.increment(1);
counter!(
zero_copy::DIRECT_IO_BYTES_TOTAL,
"operation" => operation.to_string(),
"status" => status.to_string()
)
.increment(size as u64);
}
/// Update zero-copy performance metrics.
///
/// This function updates gauge metrics for overall zero-copy performance.
#[inline(always)]
pub fn update_zero_copy_performance_metrics(copy_count: u32, throughput_mbps: f64, memory_saved: u64) {
gauge!(zero_copy::AVG_COPY_COUNT).set(copy_count as f64);
gauge!(zero_copy::THROUGHPUT_MBPS).set(throughput_mbps);
gauge!(zero_copy::MEMORY_SAVED_BYTES).set(memory_saved as f64);
}
// ============================================================================
// Zero-Copy Metrics Tests
// ============================================================================
#[cfg(test)]
mod zero_copy_tests {
use super::*;
#[test]
fn test_record_zero_copy_buffer_operation() {
// This test verifies the function compiles and runs
// Actual metric verification requires a metrics recorder
record_zero_copy_buffer_operation("read", 1024);
record_zero_copy_buffer_operation("write", 2048);
}
#[test]
fn test_record_memory_copy() {
record_memory_copy(1, 1024);
record_memory_copy(2, 2048);
}
#[test]
fn test_record_shared_ref_operation() {
record_shared_ref_operation("create");
record_shared_ref_operation("share");
}
#[test]
fn test_record_bufreader_optimization() {
record_bufreader_optimization(1, 8192);
record_bufreader_optimization(2, 65536);
}
#[test]
fn test_record_direct_io_operation() {
record_direct_io_operation("read", 4096, true);
record_direct_io_operation("write", 8192, false);
}
#[test]
fn test_update_zero_copy_performance_metrics() {
update_zero_copy_performance_metrics(2, 150.5, 1024 * 1024);
}
#[test]
fn test_metric_names() {
// Verify metric names are defined
assert!(!zero_copy::BUFFER_OPERATIONS_TOTAL.is_empty());
assert!(!zero_copy::MEMORY_COPY_TOTAL.is_empty());
assert!(!zero_copy::THROUGHPUT_MBPS.is_empty());
}
}
crates/io-metrics/src/metric_names.rs
+26 -29
View File
@@ -14,44 +14,41 @@
//! Metric name constants for consistent naming across the codebase.
/// Request-level data plane metric names introduced by ADR 0001.
pub mod data_plane {
/// Total number of selected request paths.
pub const PATH_SELECTED_TOTAL: &str = "rustfs.io.path.selected_total";
/// Zero-copy operation metric names.
pub mod zero_copy {
/// Total number of zero-copy buffer operations
pub const BUFFER_OPERATIONS_TOTAL: &str = "rustfs_zero_copy_buffer_operations_total";
/// Total bytes observed for a given effective copy mode.
pub const COPY_MODE_BYTES_TOTAL: &str = "rustfs.io.copy_mode.bytes_total";
/// Total bytes processed by zero-copy buffer operations
pub const BUFFER_BYTES_TOTAL: &str = "rustfs_zero_copy_buffer_bytes_total";
/// Total number of data plane fallbacks.
pub const FALLBACK_TOTAL: &str = "rustfs.io.zero_copy.fallback_total";
/// Total number of memory copies
pub const MEMORY_COPY_TOTAL: &str = "rustfs_memory_copy_total";
/// Current active local-disk mmap bytes held by chunk fast paths.
pub const LOCAL_DISK_ACTIVE_MMAP_BYTES: &str = "rustfs.io.local_disk.active_mmap.bytes";
/// Total bytes copied in memory
pub const MEMORY_COPY_BYTES_TOTAL: &str = "rustfs_memory_copy_bytes_total";
/// Total pooled chunks produced or consumed by LocalDisk compatibility paths.
pub const LOCAL_DISK_POOLED_CHUNKS_TOTAL: &str = "rustfs.io.local_disk.pooled_chunks.total";
/// Total number of shared reference operations
pub const SHARED_REF_OPERATIONS_TOTAL: &str = "rustfs_shared_ref_operations_total";
/// Total pooled bytes produced or consumed by LocalDisk compatibility paths.
pub const LOCAL_DISK_POOLED_BYTES_TOTAL: &str = "rustfs.io.local_disk.pooled_bytes.total";
/// Total number of BufReader layers eliminated
pub const BUFREADER_LAYERS_ELIMINATED_TOTAL: &str = "rustfs_bufreader_layers_eliminated_total";
/// Total number of compatibility chunk-stream aggregations performed for LocalDisk reads.
pub const LOCAL_DISK_COMPAT_COLLECT_TOTAL: &str = "rustfs.io.local_disk.compat_collect.total";
/// BufReader buffer size distribution
pub const BUFREADER_BUFFER_SIZE_BYTES: &str = "rustfs_bufreader_buffer_size_bytes";
/// Chunk count distribution for LocalDisk compatibility chunk aggregation.
pub const LOCAL_DISK_COMPAT_COLLECT_CHUNKS: &str = "rustfs.io.local_disk.compat_collect.chunks";
/// Total number of Direct I/O operations
pub const DIRECT_IO_OPERATIONS_TOTAL: &str = "rustfs_direct_io_operations_total";
/// Byte distribution for LocalDisk compatibility chunk aggregation.
pub const LOCAL_DISK_COMPAT_COLLECT_BYTES: &str = "rustfs.io.local_disk.compat_collect.bytes";
/// Total bytes processed by Direct I/O
pub const DIRECT_IO_BYTES_TOTAL: &str = "rustfs_direct_io_bytes_total";
/// Total number of attempted PUT fast paths.
pub const PUT_FAST_PATH_ATTEMPTS_TOTAL: &str = "rustfs.io.put.fast_path.attempts_total";
/// Average copy count per operation
pub const AVG_COPY_COUNT: &str = "rustfs_zero_copy_avg_copy_count";
/// Size distribution for attempted PUT fast paths.
pub const PUT_FAST_PATH_ATTEMPT_SIZE_BYTES: &str = "rustfs.io.put.fast_path.attempt.size.bytes";
/// Throughput in MB/s
pub const THROUGHPUT_MBPS: &str = "rustfs_zero_copy_throughput_mbps";
/// Total number of transformed PUT selections grouped by transform kind and ingress path.
pub const PUT_TRANSFORM_SELECTED_TOTAL: &str = "rustfs.io.put.transform.selected_total";
/// Size distribution for transformed PUT selections.
pub const PUT_TRANSFORM_SIZE_BYTES: &str = "rustfs.io.put.transform.size.bytes";
/// Memory saved by zero-copy in bytes
pub const MEMORY_SAVED_BYTES: &str = "rustfs_zero_copy_memory_saved_bytes";
}
crates/object-io/Cargo.toml
-36
View File
@@ -1,36 +0,0 @@
[package]
name = "rustfs-object-io"
version.workspace = true
edition.workspace = true
license.workspace = true
repository.workspace = true
rust-version.workspace = true
homepage.workspace = true
description = "Object I/O policy helpers and zero-copy support primitives for RustFS."
keywords.workspace = true
categories.workspace = true
[lints]
workspace = true
[dependencies]
atoi = { workspace = true }
bytes = { workspace = true }
futures-util = { workspace = true }
rustfs-ecstore = { workspace = true }
rustfs-io-core = { workspace = true }
rustfs-io-metrics = { workspace = true }
rustfs-rio = { workspace = true }
rustfs-s3select-api = { workspace = true }
rustfs-utils = { workspace = true ,features = ["http"]}
http = { workspace = true }
s3s.workspace = true
thiserror = { workspace = true }
time = { workspace = true, features = ["parsing", "formatting"] }
tokio = { workspace = true, features = ["io-util"] }
tokio-util = { workspace = true, features = ["io"] }
astral-tokio-tar = { workspace = true }
uuid = { workspace = true }
[lib]
doctest = false
crates/object-io/src/get.rs
-804
View File
@@ -1,804 +0,0 @@
// Copyright 2024 RustFS Team
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
use bytes::Bytes;
use http::HeaderMap;
use http::header::{CACHE_CONTROL, CONTENT_DISPOSITION, CONTENT_LANGUAGE};
use rustfs_ecstore::client::object_api_utils::to_s3s_etag;
use rustfs_ecstore::store_api::{HTTPRangeSpec, ObjectInfo};
use rustfs_rio::Reader;
use rustfs_s3select_api::object_store::bytes_stream;
use rustfs_utils::http::{AMZ_CHECKSUM_MODE, AMZ_CHECKSUM_TYPE};
use s3s::dto::{
ChecksumType, ContentType, GetObjectOutput, SSECustomerAlgorithm, SSECustomerKeyMD5, SSEKMSKeyId, ServerSideEncryption,
StreamingBlob, Timestamp,
};
use std::collections::HashMap;
use std::str::FromStr;
use std::sync::Arc;
#[cfg(test)]
use time::OffsetDateTime;
use tokio::io::{AsyncRead, AsyncSeek, ReadBuf};
use tokio_util::io::ReaderStream;
pub struct InMemoryAsyncReader {
cursor: std::io::Cursor<Bytes>,
}
impl InMemoryAsyncReader {
pub fn new(data: Bytes) -> Self {
Self {
cursor: std::io::Cursor::new(data),
}
}
}
impl AsyncRead for InMemoryAsyncReader {
fn poll_read(
mut self: std::pin::Pin<&mut Self>,
_cx: &mut std::task::Context<'_>,
buf: &mut ReadBuf<'_>,
) -> std::task::Poll<std::io::Result<()>> {
let unfilled = buf.initialize_unfilled();
let bytes_read = std::io::Read::read(&mut self.cursor, unfilled)?;
buf.advance(bytes_read);
std::task::Poll::Ready(Ok(()))
}
}
impl AsyncSeek for InMemoryAsyncReader {
fn start_seek(mut self: std::pin::Pin<&mut Self>, position: std::io::SeekFrom) -> std::io::Result<()> {
std::io::Seek::seek(&mut self.cursor, position)?;
Ok(())
}
fn poll_complete(self: std::pin::Pin<&mut Self>, _cx: &mut std::task::Context<'_>) -> std::task::Poll<std::io::Result<u64>> {
std::task::Poll::Ready(Ok(self.cursor.position()))
}
}
pub fn build_memory_blob(buf: Bytes, response_content_length: i64, optimal_buffer_size: usize) -> Option<StreamingBlob> {
let mem_reader = InMemoryAsyncReader::new(buf);
Some(StreamingBlob::wrap(bytes_stream(
ReaderStream::with_capacity(Box::new(mem_reader), optimal_buffer_size),
response_content_length as usize,
)))
}
#[derive(Clone)]
pub struct FrozenGetObjectBody {
body: Arc<Bytes>,
}
impl FrozenGetObjectBody {
pub fn new(body: Bytes) -> Self {
Self { body: Arc::new(body) }
}
pub fn shared_body(&self) -> &Arc<Bytes> {
&self.body
}
pub fn into_shared_body(self) -> Arc<Bytes> {
self.body
}
pub fn build_blob(&self, response_content_length: i64, optimal_buffer_size: usize) -> Option<StreamingBlob> {
build_memory_blob((*self.body).clone(), response_content_length, optimal_buffer_size)
}
}
pub fn build_reader_blob<R>(reader: R, response_content_length: i64, optimal_buffer_size: usize) -> Option<StreamingBlob>
where
R: AsyncRead + Send + Sync + 'static,
{
Some(StreamingBlob::wrap(bytes_stream(
ReaderStream::with_capacity(reader, optimal_buffer_size),
response_content_length as usize,
)))
}
pub fn get_object_sequential_hint(rs: Option<&HTTPRangeSpec>) -> bool {
if rs.is_none() {
true
} else if let Some(range_spec) = rs {
range_spec.start == 0 && !range_spec.is_suffix_length
} else {
false
}
}
pub struct GetObjectOutputContext {
pub output: GetObjectOutput,
pub event_info: ObjectInfo,
pub response_content_length: i64,
pub optimal_buffer_size: usize,
pub copy_mode_override: Option<rustfs_io_metrics::CopyMode>,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct GetObjectStrategyLayout {
pub is_sequential_hint: bool,
pub optimal_buffer_size: usize,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct GetObjectBodyPlanningInputs {
pub is_part_request: bool,
pub is_range_request: bool,
pub encryption_applied: bool,
pub response_size: i64,
}
pub enum GetObjectBodySource {
Reader(Box<dyn Reader>),
}
pub struct GetObjectReadSetup {
pub info: ObjectInfo,
pub event_info: ObjectInfo,
pub body_source: GetObjectBodySource,
pub rs: Option<HTTPRangeSpec>,
pub content_type: Option<ContentType>,
pub last_modified: Option<Timestamp>,
pub response_content_length: i64,
pub content_range: Option<String>,
pub server_side_encryption: Option<ServerSideEncryption>,
pub sse_customer_algorithm: Option<SSECustomerAlgorithm>,
pub sse_customer_key_md5: Option<SSECustomerKeyMD5>,
pub ssekms_key_id: Option<SSEKMSKeyId>,
pub encryption_applied: bool,
}
pub struct LegacyReadPlan {
pub rs: Option<HTTPRangeSpec>,
pub content_type: Option<ContentType>,
pub last_modified: Option<Timestamp>,
pub response_content_length: i64,
pub content_range: Option<String>,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum GetObjectBodyPlan {
BufferEncrypted,
BufferSeekable,
Stream,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct GetObjectDataPlaneMetricContract {
pub io_path: rustfs_io_metrics::IoPath,
pub copy_mode: rustfs_io_metrics::CopyMode,
}
impl GetObjectDataPlaneMetricContract {
pub fn disk(io_path: rustfs_io_metrics::IoPath, copy_mode: rustfs_io_metrics::CopyMode) -> Self {
Self { io_path, copy_mode }
}
}
pub struct GetObjectBodyMaterialization {
pub body: Option<StreamingBlob>,
pub plan: GetObjectBodyPlan,
}
#[derive(Default)]
pub struct GetObjectEncryptionState {
pub server_side_encryption: Option<ServerSideEncryption>,
pub sse_customer_algorithm: Option<SSECustomerAlgorithm>,
pub sse_customer_key_md5: Option<SSECustomerKeyMD5>,
pub ssekms_key_id: Option<SSEKMSKeyId>,
pub encryption_applied: bool,
pub response_content_length_override: Option<i64>,
}
#[derive(Debug, thiserror::Error)]
pub enum MaterializeGetObjectBodyError {
#[error("failed to read decrypted object: {0}")]
EncryptedRead(std::io::Error),
}
pub struct GetObjectFlowResult {
pub output: GetObjectOutput,
pub event_info: ObjectInfo,
pub version_id_for_event: String,
}
pub fn build_get_object_flow_result(
output: GetObjectOutput,
event_info: ObjectInfo,
version_id_for_event: String,
) -> GetObjectFlowResult {
GetObjectFlowResult {
output,
event_info,
version_id_for_event,
}
}
pub fn build_cors_wrapped_get_object_flow_result(
output_context: GetObjectOutputContext,
version_id_for_event: String,
) -> GetObjectFlowResult {
let GetObjectOutputContext {
output,
event_info,
response_content_length: _,
optimal_buffer_size: _,
copy_mode_override: _,
} = output_context;
build_get_object_flow_result(output, event_info, version_id_for_event)
}
#[derive(Debug, Default, Clone, PartialEq, Eq)]
pub struct GetObjectChecksums {
pub crc32: Option<String>,
pub crc32c: Option<String>,
pub sha1: Option<String>,
pub sha256: Option<String>,
pub crc64nvme: Option<String>,
pub checksum_type: Option<ChecksumType>,
}
fn decode_get_object_checksums(decrypted_checksums: HashMap<String, String>) -> GetObjectChecksums {
let mut checksums = GetObjectChecksums::default();
for (key, checksum) in decrypted_checksums {
if key == AMZ_CHECKSUM_TYPE {
checksums.checksum_type = Some(ChecksumType::from(checksum));
continue;
}
match rustfs_rio::ChecksumType::from_string(key.as_str()) {
rustfs_rio::ChecksumType::CRC32 => checksums.crc32 = Some(checksum),
rustfs_rio::ChecksumType::CRC32C => checksums.crc32c = Some(checksum),
rustfs_rio::ChecksumType::SHA1 => checksums.sha1 = Some(checksum),
rustfs_rio::ChecksumType::SHA256 => checksums.sha256 = Some(checksum),
rustfs_rio::ChecksumType::CRC64_NVME => checksums.crc64nvme = Some(checksum),
_ => (),
}
}
checksums
}
fn read_object_checksums(
info: &ObjectInfo,
headers: &HeaderMap,
part_number: Option<usize>,
) -> std::io::Result<GetObjectChecksums> {
let (decrypted_checksums, _is_multipart) = info
.decrypt_checksums(part_number.unwrap_or(0), headers)
.map_err(|e| std::io::Error::other(e.to_string()))?;
Ok(decode_get_object_checksums(decrypted_checksums))
}
pub fn build_get_object_checksums(
info: &ObjectInfo,
headers: &HeaderMap,
part_number: Option<usize>,
rs: Option<&HTTPRangeSpec>,
) -> std::io::Result<GetObjectChecksums> {
if let Some(checksum_mode) = headers.get(AMZ_CHECKSUM_MODE)
&& checksum_mode.to_str().unwrap_or_default() == "ENABLED"
&& rs.is_none()
{
return read_object_checksums(info, headers, part_number);
}
Ok(GetObjectChecksums::default())
}
pub fn build_output_version_id(versioned: bool, version_id: Option<&uuid::Uuid>) -> Option<String> {
if !versioned {
return None;
}
version_id.map(|vid| {
if *vid == uuid::Uuid::nil() {
"null".to_string()
} else {
vid.to_string()
}
})
}
pub fn plan_get_object_strategy_layout(
rs: Option<&HTTPRangeSpec>,
response_content_length: i64,
suggested_buffer_size: usize,
fallback_buffer_size: usize,
) -> GetObjectStrategyLayout {
let is_sequential_hint = get_object_sequential_hint(rs);
let optimal_buffer_size = if suggested_buffer_size > 0 {
suggested_buffer_size.min(fallback_buffer_size)
} else {
rustfs_io_core::get_concurrency_aware_buffer_size(response_content_length, fallback_buffer_size)
};
GetObjectStrategyLayout {
is_sequential_hint,
optimal_buffer_size,
}
}
pub fn plan_get_object_body(
planning_inputs: GetObjectBodyPlanningInputs,
seekable_object_size_threshold: usize,
) -> GetObjectBodyPlan {
let should_buffer_for_seek = planning_inputs.response_size > 0
&& planning_inputs.response_size <= seekable_object_size_threshold as i64
&& !planning_inputs.is_part_request
&& !planning_inputs.is_range_request;
if planning_inputs.encryption_applied && should_buffer_for_seek {
GetObjectBodyPlan::BufferEncrypted
} else if should_buffer_for_seek {
GetObjectBodyPlan::BufferSeekable
} else {
GetObjectBodyPlan::Stream
}
}
pub async fn materialize_get_object_body<R>(
mut final_stream: R,
plan: GetObjectBodyPlan,
response_content_length: i64,
optimal_buffer_size: usize,
) -> Result<GetObjectBodyMaterialization, MaterializeGetObjectBodyError>
where
R: AsyncRead + Send + Sync + Unpin + 'static,
{
match plan {
GetObjectBodyPlan::BufferEncrypted => {
let mut buf = Vec::with_capacity(response_content_length as usize);
tokio::io::AsyncReadExt::read_to_end(&mut final_stream, &mut buf)
.await
.map_err(MaterializeGetObjectBodyError::EncryptedRead)?;
let body = FrozenGetObjectBody::new(Bytes::from(buf));
Ok(GetObjectBodyMaterialization {
body: body.build_blob(response_content_length, optimal_buffer_size),
plan,
})
}
GetObjectBodyPlan::BufferSeekable => {
let mut buf = Vec::with_capacity(response_content_length as usize);
let body = match tokio::io::AsyncReadExt::read_to_end(&mut final_stream, &mut buf).await {
Ok(_) => FrozenGetObjectBody::new(Bytes::from(buf)).build_blob(response_content_length, optimal_buffer_size),
Err(_) => build_reader_blob(final_stream, response_content_length, optimal_buffer_size),
};
Ok(GetObjectBodyMaterialization { body, plan })
}
GetObjectBodyPlan::Stream => Ok(GetObjectBodyMaterialization {
body: build_reader_blob(final_stream, response_content_length, optimal_buffer_size),
plan,
}),
}
}
fn resolve_requested_range(
info: &ObjectInfo,
mut rs: Option<HTTPRangeSpec>,
part_number: Option<usize>,
) -> Option<HTTPRangeSpec> {
if let Some(part_number) = part_number
&& rs.is_none()
{
rs = HTTPRangeSpec::from_object_info(info, part_number);
}
rs
}
fn resolve_response_range(
total_size: i64,
rs: Option<HTTPRangeSpec>,
) -> std::io::Result<(Option<HTTPRangeSpec>, i64, Option<String>)> {
let Some(range_spec) = rs else {
return Ok((None, total_size, None));
};
let (start, length) = range_spec.get_offset_length(total_size)?;
let content_range = Some(format!("bytes {}-{}/{}", start, start as i64 + length - 1, total_size));
Ok((Some(range_spec), length, content_range))
}
pub fn plan_legacy_read(
info: &ObjectInfo,
rs: Option<HTTPRangeSpec>,
part_number: Option<usize>,
) -> std::io::Result<LegacyReadPlan> {
let content_type = info
.content_type
.as_ref()
.and_then(|content_type| ContentType::from_str(content_type).ok());
let last_modified = info.mod_time.map(Timestamp::from);
let rs = resolve_requested_range(info, rs, part_number);
let total_size = info.get_actual_size()?;
let (rs, response_content_length, content_range) = resolve_response_range(total_size, rs)?;
Ok(LegacyReadPlan {
rs,
content_type,
last_modified,
response_content_length,
content_range,
})
}
pub fn build_reader_read_setup(
info: ObjectInfo,
event_info: ObjectInfo,
final_stream: Box<dyn Reader>,
plan: LegacyReadPlan,
encryption_state: GetObjectEncryptionState,
) -> GetObjectReadSetup {
let LegacyReadPlan {
rs,
content_type,
last_modified,
response_content_length,
content_range,
} = plan;
let GetObjectEncryptionState {
server_side_encryption,
sse_customer_algorithm,
sse_customer_key_md5,
ssekms_key_id,
encryption_applied,
response_content_length_override,
} = encryption_state;
GetObjectReadSetup {
info,
event_info,
body_source: GetObjectBodySource::Reader(final_stream),
rs,
content_type,
last_modified,
response_content_length: response_content_length_override.unwrap_or(response_content_length),
content_range,
server_side_encryption,
sse_customer_algorithm,
sse_customer_key_md5,
ssekms_key_id,
encryption_applied,
}
}
#[allow(clippy::too_many_arguments)]
pub fn build_get_object_output_context(
body: Option<StreamingBlob>,
info: ObjectInfo,
event_info: ObjectInfo,
content_type: Option<ContentType>,
last_modified: Option<Timestamp>,
response_content_length: i64,
content_range: Option<String>,
server_side_encryption: Option<ServerSideEncryption>,
sse_customer_algorithm: Option<SSECustomerAlgorithm>,
sse_customer_key_md5: Option<SSECustomerKeyMD5>,
ssekms_key_id: Option<SSEKMSKeyId>,
checksums: &GetObjectChecksums,
filtered_metadata: Option<HashMap<String, String>>,
versioned: bool,
optimal_buffer_size: usize,
copy_mode_override: Option<rustfs_io_metrics::CopyMode>,
) -> GetObjectOutputContext {
let output_version_id = build_output_version_id(versioned, info.version_id.as_ref());
let output = build_get_object_output(
body,
&info,
content_type,
last_modified,
response_content_length,
content_range,
server_side_encryption,
sse_customer_algorithm,
sse_customer_key_md5,
ssekms_key_id,
checksums,
output_version_id,
filtered_metadata,
);
GetObjectOutputContext {
output,
event_info,
response_content_length,
optimal_buffer_size,
copy_mode_override,
}
}
#[allow(clippy::too_many_arguments)]
pub fn build_get_object_output(
body: Option<StreamingBlob>,
info: &ObjectInfo,
content_type: Option<ContentType>,
last_modified: Option<Timestamp>,
response_content_length: i64,
content_range: Option<String>,
server_side_encryption: Option<ServerSideEncryption>,
sse_customer_algorithm: Option<SSECustomerAlgorithm>,
sse_customer_key_md5: Option<SSECustomerKeyMD5>,
ssekms_key_id: Option<SSEKMSKeyId>,
checksums: &GetObjectChecksums,
output_version_id: Option<String>,
filtered_metadata: Option<HashMap<String, String>>,
) -> GetObjectOutput {
GetObjectOutput {
body,
content_length: Some(response_content_length),
last_modified,
expires: info.expires.map(Timestamp::from),
content_type,
cache_control: info.user_defined.get(CACHE_CONTROL.as_str()).cloned(),
content_disposition: info.user_defined.get(CONTENT_DISPOSITION.as_str()).cloned(),
content_encoding: info.content_encoding.clone(),
content_language: info.user_defined.get(CONTENT_LANGUAGE.as_str()).cloned(),
accept_ranges: Some("bytes".to_string()),
content_range,
e_tag: info.etag.as_ref().map(|etag| to_s3s_etag(etag)),
metadata: filtered_metadata,
server_side_encryption,
sse_customer_algorithm,
sse_customer_key_md5,
ssekms_key_id,
checksum_crc32: checksums.crc32.clone(),
checksum_crc32c: checksums.crc32c.clone(),
checksum_sha1: checksums.sha1.clone(),
checksum_sha256: checksums.sha256.clone(),
checksum_crc64nvme: checksums.crc64nvme.clone(),
checksum_type: checksums.checksum_type.clone(),
version_id: output_version_id,
..Default::default()
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn build_get_object_checksums_returns_default_when_mode_absent() {
let checksums = build_get_object_checksums(&ObjectInfo::default(), &HeaderMap::new(), None, None).unwrap();
assert_eq!(checksums, GetObjectChecksums::default());
}
#[test]
fn plan_legacy_read_uses_part_number_range_when_available() {
let mut info = ObjectInfo {
size: 12,
actual_size: 12,
content_type: Some("application/octet-stream".to_string()),
..Default::default()
};
info.parts = vec![Default::default(), Default::default()];
info.parts[0].number = 1;
info.parts[0].size = 5;
info.parts[0].actual_size = 5;
info.parts[1].number = 2;
info.parts[1].size = 7;
info.parts[1].actual_size = 7;
let plan = plan_legacy_read(&info, None, Some(2)).unwrap();
let rs = plan.rs.expect("range from part number");
assert_eq!(rs.start, 5);
assert_eq!(rs.end, 11);
assert_eq!(plan.response_content_length, 7);
assert_eq!(plan.content_range.as_deref(), Some("bytes 5-11/12"));
assert!(plan.content_type.is_some());
}
#[test]
fn build_reader_read_setup_uses_encryption_length_override() {
let plan = LegacyReadPlan {
rs: None,
content_type: None,
last_modified: None,
response_content_length: 16,
content_range: None,
};
let encryption_state = GetObjectEncryptionState {
encryption_applied: true,
response_content_length_override: Some(12),
..Default::default()
};
let reader = Box::new(rustfs_rio::WarpReader::new(tokio::io::empty())) as Box<dyn Reader>;
let setup = build_reader_read_setup(ObjectInfo::default(), ObjectInfo::default(), reader, plan, encryption_state);
assert!(setup.encryption_applied);
assert_eq!(setup.response_content_length, 12);
match setup.body_source {
GetObjectBodySource::Reader(_) => {}
}
}
#[test]
fn plan_get_object_body_buffers_seekable_small_plain_request() {
let plan = plan_get_object_body(
GetObjectBodyPlanningInputs {
is_part_request: false,
is_range_request: false,
encryption_applied: false,
response_size: 1024,
},
4096,
);
assert_eq!(plan, GetObjectBodyPlan::BufferSeekable);
}
#[test]
fn disk_metric_contract_preserves_io_labels() {
let contract =
GetObjectDataPlaneMetricContract::disk(rustfs_io_metrics::IoPath::Legacy, rustfs_io_metrics::CopyMode::SingleCopy);
assert_eq!(contract.io_path, rustfs_io_metrics::IoPath::Legacy);
assert_eq!(contract.copy_mode, rustfs_io_metrics::CopyMode::SingleCopy);
}
#[test]
fn plan_get_object_body_uses_encrypted_buffer_for_small_plain_request() {
let plan = plan_get_object_body(
GetObjectBodyPlanningInputs {
is_part_request: false,
is_range_request: false,
encryption_applied: true,
response_size: 1024,
},
4096,
);
assert_eq!(plan, GetObjectBodyPlan::BufferEncrypted);
}
#[test]
fn get_object_sequential_hint_distinguishes_prefix_and_suffix_ranges() {
assert!(get_object_sequential_hint(None));
assert!(get_object_sequential_hint(Some(&HTTPRangeSpec {
is_suffix_length: false,
start: 0,
end: -1,
})));
assert!(!get_object_sequential_hint(Some(&HTTPRangeSpec {
is_suffix_length: false,
start: 4,
end: 8,
})));
assert!(!get_object_sequential_hint(Some(&HTTPRangeSpec {
is_suffix_length: true,
start: 4,
end: -1,
})));
}
#[test]
fn plan_get_object_strategy_layout_caps_buffer_to_fallback() {
let layout = plan_get_object_strategy_layout(None, 1024, 8192, 4096);
assert!(layout.is_sequential_hint);
assert_eq!(layout.optimal_buffer_size, 4096);
}
#[test]
fn frozen_get_object_body_reuses_same_shared_bytes_for_memory_blob() {
let frozen = FrozenGetObjectBody::new(Bytes::from_static(b"abc"));
let shared = Arc::clone(frozen.shared_body());
assert_eq!(*shared, Bytes::from_static(b"abc"));
assert!(Arc::ptr_eq(&shared, frozen.shared_body()));
}
#[test]
fn build_output_version_id_maps_nil_uuid_to_null() {
let nil = uuid::Uuid::nil();
let version_id = build_output_version_id(true, Some(&nil));
assert_eq!(version_id.as_deref(), Some("null"));
}
#[test]
fn build_get_object_output_context_preserves_copy_mode_override() {
let info = ObjectInfo {
version_id: Some(uuid::Uuid::nil()),
..Default::default()
};
let output_context = build_get_object_output_context(
None,
info,
ObjectInfo::default(),
None,
None,
8,
None,
None,
None,
None,
None,
&GetObjectChecksums::default(),
None,
true,
4096,
Some(rustfs_io_metrics::CopyMode::Reconstructed),
);
assert_eq!(output_context.output.version_id.as_deref(), Some("null"));
assert_eq!(output_context.copy_mode_override, Some(rustfs_io_metrics::CopyMode::Reconstructed));
assert_eq!(output_context.optimal_buffer_size, 4096);
}
#[test]
fn build_get_object_output_preserves_http_metadata_like_cached_path() {
let mut info = ObjectInfo {
content_type: Some("application/octet-stream".to_string()),
content_encoding: Some("zstd".to_string()),
etag: Some("etag".to_string()),
expires: Some(OffsetDateTime::UNIX_EPOCH),
..Default::default()
};
info.user_defined
.insert("cache-control".to_string(), "max-age=3600".to_string());
info.user_defined
.insert("content-disposition".to_string(), "attachment; filename=\"bundle.zip\"".to_string());
info.user_defined.insert("content-language".to_string(), "en-US".to_string());
let output = build_get_object_output(
None,
&info,
info.content_type
.as_ref()
.and_then(|content_type| ContentType::from_str(content_type).ok()),
info.mod_time.map(Timestamp::from),
8,
None,
None,
None,
None,
None,
&GetObjectChecksums::default(),
None,
None,
);
assert_eq!(output.cache_control.as_deref(), Some("max-age=3600"));
assert_eq!(output.content_disposition.as_deref(), Some("attachment; filename=\"bundle.zip\""));
assert_eq!(output.content_language.as_deref(), Some("en-US"));
assert_eq!(output.content_encoding.as_deref(), Some("zstd"));
assert_eq!(output.expires, Some(Timestamp::from(OffsetDateTime::UNIX_EPOCH)));
}
#[test]
fn build_cors_wrapped_get_object_flow_result_uses_wrapped_mode() {
let result = build_cors_wrapped_get_object_flow_result(
GetObjectOutputContext {
output: GetObjectOutput::default(),
event_info: ObjectInfo::default(),
response_content_length: 1,
optimal_buffer_size: 1024,
copy_mode_override: None,
},
"vid".to_string(),
);
assert_eq!(result.version_id_for_event, "vid");
}
}
crates/object-io/src/lib.rs
-16
View File
@@ -1,16 +0,0 @@
// Copyright 2024 RustFS Team
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
pub mod get;
pub mod put;
crates/object-io/src/put.rs
-1439
View File
File diff suppressed because it is too large Load Diff
crates/rio/src/hash_reader.rs
+2 -25
View File
@@ -408,23 +408,6 @@ impl HashReader {
Ok(())
}
pub fn enable_auto_checksum(&mut self, checksum_type: ChecksumType) -> Result<(), std::io::Error> {
if !checksum_type.is_set() {
return Ok(());
}
let Some(hasher) = checksum_type.hasher() else {
return Err(std::io::Error::new(std::io::ErrorKind::InvalidData, "Invalid checksum type"));
};
self.content_hash = Some(Checksum {
checksum_type,
..Default::default()
});
self.content_hasher = Some(hasher);
Ok(())
}
pub fn checksum(&self) -> Option<Checksum> {
if self
.content_hash
@@ -564,9 +547,7 @@ impl AsyncRead for HashReader {
}
// check content hasher
if let (Some(hasher), Some(expected_content_hash)) =
(this.content_hasher.as_mut(), this.content_hash.as_mut())
{
if let (Some(hasher), Some(expected_content_hash)) = (this.content_hasher, this.content_hash) {
if expected_content_hash.checksum_type.trailing()
&& let Some(trailer) = this.trailer_s3s.as_ref()
&& let Some(Some(checksum_str)) = trailer.read(|headers| {
@@ -588,11 +569,7 @@ impl AsyncRead for HashReader {
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 {
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!(
rustfs/Cargo.toml
-1
View File
@@ -90,7 +90,6 @@ rustfs-utils = { workspace = true, features = ["full"] }
rustfs-zip = { workspace = true }
rustfs-io-core = { workspace = true }
rustfs-io-metrics = { workspace = true }
rustfs-object-io = { workspace = true }
rustfs-object-capacity = { workspace = true }
rustfs-concurrency = { workspace = true }
rustfs-scanner = { workspace = true }
rustfs/src/app/multipart_usecase.rs
+101 -64
View File
@@ -48,8 +48,7 @@ use rustfs_ecstore::set_disk::is_valid_storage_class;
use rustfs_ecstore::store_api::{CompletePart, HTTPRangeSpec, MultipartUploadResult, ObjectIO, ObjectOptions, PutObjReader};
use rustfs_ecstore::store_api::{MultipartOperations, ObjectOperations};
use rustfs_filemeta::{ReplicationStatusType, ReplicationType};
use rustfs_object_io::put::PutObjectChecksums;
use rustfs_rio::{CompressReader, HashReader, Reader, WarpReader};
use rustfs_rio::{CompressReader, HashReader};
use rustfs_s3_common::S3Operation;
use rustfs_targets::EventName;
use rustfs_utils::CompressionAlgorithm;
@@ -651,13 +650,6 @@ impl DefaultMultipartUsecase {
.map_err(ApiError::from)?;
let mut size = size.ok_or_else(|| s3_error!(UnexpectedContent))?;
let mut requested_checksum_type = rustfs_rio::ChecksumType::from_header(&req.headers);
if !requested_checksum_type.is_set()
&& let Some(checksum_algo) = fi.user_defined.get(rustfs_rio::RUSTFS_MULTIPART_CHECKSUM)
&& let Some(checksum_type) = fi.user_defined.get(rustfs_rio::RUSTFS_MULTIPART_CHECKSUM_TYPE)
{
requested_checksum_type = rustfs_rio::ChecksumType::from_string_with_obj_type(checksum_algo, checksum_type);
}
// Apply adaptive buffer sizing based on part size for optimal streaming performance.
// Uses workload profile configuration (enabled by default) to select appropriate buffer size.
@@ -670,8 +662,6 @@ impl DefaultMultipartUsecase {
let is_compressible = rustfs_utils::http::contains_key_str(&fi.user_defined, rustfs_utils::http::SUFFIX_COMPRESSION);
let mut reader: Box<dyn Reader> = Box::new(WarpReader::new(body));
let actual_size = size;
let mut md5hex = if let Some(base64_md5) = input.content_md5 {
@@ -685,32 +675,31 @@ impl DefaultMultipartUsecase {
let mut sha256hex = get_content_sha256_with_query(&req.headers, req.uri.query());
if is_compressible {
let mut hrd =
HashReader::new(reader, size, actual_size, md5hex.take(), sha256hex.take(), false).map_err(ApiError::from)?;
let mut reader = if is_compressible {
let mut hrd = HashReader::from_stream(body, size, actual_size, md5hex.take(), sha256hex.take(), false)
.map_err(ApiError::from)?;
if let Err(err) = hrd.add_checksum_from_s3s(&req.headers, req.trailing_headers.clone(), false) {
return Err(ApiError::from(err).into());
}
if requested_checksum_type.is_set() && hrd.checksum().is_none() {
hrd.enable_auto_checksum(requested_checksum_type).map_err(ApiError::from)?;
}
let compress_reader = CompressReader::new(hrd, CompressionAlgorithm::default());
reader = Box::new(compress_reader);
size = HashReader::SIZE_PRESERVE_LAYER;
md5hex = None;
sha256hex = None;
}
let mut reader = HashReader::new(reader, size, actual_size, md5hex, sha256hex, false).map_err(ApiError::from)?;
HashReader::from_reader(
CompressReader::new(hrd, CompressionAlgorithm::default()),
size,
actual_size,
None,
None,
false,
)
.map_err(ApiError::from)?
} else {
HashReader::from_stream(body, size, actual_size, md5hex, sha256hex, false).map_err(ApiError::from)?
};
if let Err(err) = reader.add_checksum_from_s3s(&req.headers, req.trailing_headers.clone(), size < 0) {
return Err(ApiError::from(err).into());
}
if requested_checksum_type.is_set() && reader.checksum().is_none() {
reader.enable_auto_checksum(requested_checksum_type).map_err(ApiError::from)?;
}
let has_ssec = sse_customer_algorithm.is_some();
// When SSE-C headers are present, skip managed-encryption metadata to avoid
@@ -760,8 +749,9 @@ impl DefaultMultipartUsecase {
let requested_kms_key_id = material.kms_key_id.clone();
let encrypted_reader = material.wrap_reader(reader);
reader = HashReader::new(encrypted_reader, HashReader::SIZE_PRESERVE_LAYER, actual_size, None, None, false)
.map_err(ApiError::from)?;
reader =
HashReader::from_reader(encrypted_reader, HashReader::SIZE_PRESERVE_LAYER, actual_size, None, None, false)
.map_err(ApiError::from)?;
fi.user_defined.extend(material.metadata);
@@ -777,13 +767,11 @@ impl DefaultMultipartUsecase {
.await
.map_err(ApiError::from)?;
let mut checksums = PutObjectChecksums {
crc32: input.checksum_crc32,
crc32c: input.checksum_crc32c,
sha1: input.checksum_sha1,
sha256: input.checksum_sha256,
crc64nvme: input.checksum_crc64nvme,
};
let mut checksum_crc32 = input.checksum_crc32;
let mut checksum_crc32c = input.checksum_crc32c;
let mut checksum_sha1 = input.checksum_sha1;
let mut checksum_sha256 = input.checksum_sha256;
let mut checksum_crc64nvme = input.checksum_crc64nvme;
if let Some(alg) = &input.checksum_algorithm
&& let Some(Some(checksum_str)) = req.trailing_headers.as_ref().map(|trailer| {
@@ -803,26 +791,25 @@ impl DefaultMultipartUsecase {
})
{
match alg.as_str() {
ChecksumAlgorithm::CRC32 => checksums.crc32 = checksum_str,
ChecksumAlgorithm::CRC32C => checksums.crc32c = checksum_str,
ChecksumAlgorithm::SHA1 => checksums.sha1 = checksum_str,
ChecksumAlgorithm::SHA256 => checksums.sha256 = checksum_str,
ChecksumAlgorithm::CRC64NVME => checksums.crc64nvme = checksum_str,
ChecksumAlgorithm::CRC32 => checksum_crc32 = checksum_str,
ChecksumAlgorithm::CRC32C => checksum_crc32c = checksum_str,
ChecksumAlgorithm::SHA1 => checksum_sha1 = checksum_str,
ChecksumAlgorithm::SHA256 => checksum_sha256 = checksum_str,
ChecksumAlgorithm::CRC64NVME => checksum_crc64nvme = checksum_str,
_ => (),
}
}
checksums.merge_from_map(&reader.as_hash_reader().content_crc());
let output = UploadPartOutput {
server_side_encryption: requested_sse,
ssekms_key_id: requested_kms_key_id,
sse_customer_algorithm,
sse_customer_key_md5,
checksum_crc32: checksums.crc32,
checksum_crc32c: checksums.crc32c,
checksum_sha1: checksums.sha1,
checksum_sha256: checksums.sha256,
checksum_crc64nvme: checksums.crc64nvme,
checksum_crc32,
checksum_crc32c,
checksum_sha1,
checksum_sha256,
checksum_crc64nvme,
e_tag: info.etag.map(|etag| to_s3s_etag(&etag)),
..Default::default()
};
@@ -1013,8 +1000,6 @@ impl DefaultMultipartUsecase {
let is_compressible = rustfs_utils::http::contains_key_str(&mp_info.user_defined, rustfs_utils::http::SUFFIX_COMPRESSION);
let mut reader: Box<dyn Reader> = Box::new(WarpReader::new(src_stream));
let src_decryption_request = DecryptionRequest {
bucket: &src_bucket,
key: &src_key,
@@ -1026,23 +1011,74 @@ impl DefaultMultipartUsecase {
etag: src_info.etag.as_deref(),
};
if let Some(material) = sse_decryption(src_decryption_request).await? {
reader = material.wrap_single_reader(reader);
if let Some(original) = material.original_size {
src_info.actual_size = original;
}
}
let actual_size = length;
let mut size = length;
if is_compressible {
let hrd = HashReader::new(reader, size, actual_size, None, None, false).map_err(ApiError::from)?;
reader = Box::new(CompressReader::new(hrd, CompressionAlgorithm::default()));
size = HashReader::SIZE_PRESERVE_LAYER;
}
let mut reader = match sse_decryption(src_decryption_request).await? {
Some(material) => {
if let Some(original) = material.original_size {
src_info.actual_size = original;
}
let mut reader = HashReader::new(reader, size, actual_size, None, None, false).map_err(ApiError::from)?;
if material.is_multipart {
let (decrypted_stream, plaintext_size) =
material.wrap_reader(src_stream, size).await.map_err(ApiError::from)?;
size = plaintext_size;
if is_compressible {
let hrd = HashReader::from_reader(decrypted_stream, size, actual_size, None, None, false)
.map_err(ApiError::from)?;
size = HashReader::SIZE_PRESERVE_LAYER;
HashReader::from_reader(
CompressReader::new(hrd, CompressionAlgorithm::default()),
size,
actual_size,
None,
None,
false,
)
.map_err(ApiError::from)?
} else {
HashReader::from_reader(decrypted_stream, size, actual_size, None, None, false).map_err(ApiError::from)?
}
} else if is_compressible {
let hrd =
HashReader::from_stream(material.wrap_single_reader(src_stream), size, actual_size, None, None, false)
.map_err(ApiError::from)?;
size = HashReader::SIZE_PRESERVE_LAYER;
HashReader::from_reader(
CompressReader::new(hrd, CompressionAlgorithm::default()),
size,
actual_size,
None,
None,
false,
)
.map_err(ApiError::from)?
} else {
HashReader::from_stream(material.wrap_single_reader(src_stream), size, actual_size, None, None, false)
.map_err(ApiError::from)?
}
}
None => {
if is_compressible {
let hrd =
HashReader::from_stream(src_stream, size, actual_size, None, None, false).map_err(ApiError::from)?;
size = HashReader::SIZE_PRESERVE_LAYER;
HashReader::from_reader(
CompressReader::new(hrd, CompressionAlgorithm::default()),
size,
actual_size,
None,
None,
false,
)
.map_err(ApiError::from)?
} else {
HashReader::from_stream(src_stream, size, actual_size, None, None, false).map_err(ApiError::from)?
}
}
};
let server_side_encryption = mp_info
.user_defined
@@ -1083,8 +1119,9 @@ impl DefaultMultipartUsecase {
let requested_kms_key_id = material.kms_key_id.clone();
let encrypted_reader = material.wrap_reader(reader);
reader = HashReader::new(encrypted_reader, HashReader::SIZE_PRESERVE_LAYER, actual_size, None, None, false)
.map_err(ApiError::from)?;
reader =
HashReader::from_reader(encrypted_reader, HashReader::SIZE_PRESERVE_LAYER, actual_size, None, None, false)
.map_err(ApiError::from)?;
mp_info.user_defined.extend(material.metadata);
rustfs/src/app/object_usecase.rs
+2369 -284
View File
File diff suppressed because it is too large Load Diff
rustfs/src/error.rs
+1 -17
View File
@@ -257,24 +257,8 @@ impl From<StorageError> for ApiError {
impl From<std::io::Error> for ApiError {
fn from(err: std::io::Error) -> Self {
// Check if the inner error is a StorageError (e.g. InvalidRangeSpec wrapped by object-io)
// Check if the error is a ChecksumMismatch (BadDigest)
if let Some(inner) = err.get_ref() {
if let Some(storage_error) = inner.downcast_ref::<StorageError>() {
let code = match storage_error {
StorageError::InvalidRangeSpec(_) => S3ErrorCode::InvalidRange,
_ => S3ErrorCode::InternalError,
};
let message = if code == S3ErrorCode::InternalError {
storage_error.to_string()
} else {
ApiError::error_code_to_message(&code)
};
return ApiError {
code,
message,
source: Some(Box::new(err)),
};
}
if inner.downcast_ref::<rustfs_rio::ChecksumMismatch>().is_some() {
return ApiError {
code: S3ErrorCode::BadDigest,