From 87423bfb8c0d7fc5fce8ce701e6f26ed7747fa8a Mon Sep 17 00:00:00 2001 From: Nugine Date: Sun, 15 Jun 2025 21:01:38 +0800 Subject: [PATCH 1/6] build(deps): update `bytes` --- Cargo.lock | 4 ++++ Cargo.toml | 2 +- 2 files changed, 5 insertions(+), 1 deletion(-) diff --git a/Cargo.lock b/Cargo.lock index 09812efc8..4a885a4ba 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1406,6 +1406,9 @@ name = "bytes" version = "1.10.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "d71b6127be86fdcfddb610f7182ac57211d4b18a3e9c82eb2d17662f2227ad6a" +dependencies = [ + "serde", +] [[package]] name = "bytes-utils" @@ -8357,6 +8360,7 @@ name = "rustfs-filemeta" version = "0.0.1" dependencies = [ "byteorder", + "bytes", "crc32fast", "criterion", "rmp", diff --git a/Cargo.toml b/Cargo.toml index 96a1b07f0..d9af6894d 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -75,7 +75,7 @@ axum-server = { version = "0.7.2", features = ["tls-rustls"] } backon = "1.5.1" base64-simd = "0.8.0" blake2 = "0.10.6" -bytes = "1.10.1" +bytes = { version = "1.10.1", features = ["serde"] } bytesize = "2.0.1" byteorder = "1.5.0" cfg-if = "1.0.0" From 3a567768c1e69443706bc03eac3c321caf3ff6c1 Mon Sep 17 00:00:00 2001 From: Nugine Date: Sun, 15 Jun 2025 21:01:38 +0800 Subject: [PATCH 2/6] refactor(filemeta): `ChecksumInfo` `hash` use `Bytes` --- crates/filemeta/Cargo.toml | 2 +- crates/filemeta/src/fileinfo.rs | 3 ++- ecstore/src/disk/local.rs | 2 +- ecstore/src/erasure_coding/bitrot.rs | 3 ++- 4 files changed, 6 insertions(+), 4 deletions(-) diff --git a/crates/filemeta/Cargo.toml b/crates/filemeta/Cargo.toml index 7f92441e6..6f5e581aa 100644 --- a/crates/filemeta/Cargo.toml +++ b/crates/filemeta/Cargo.toml @@ -15,7 +15,7 @@ time.workspace = true uuid = { workspace = true, features = ["v4", "fast-rng", "serde"] } tokio = { workspace = true, features = ["io-util", "macros", "sync"] } xxhash-rust = { version = "0.8.15", features = ["xxh64"] } - +bytes.workspace = true rustfs-utils = {workspace = true, features= ["hash"]} byteorder = "1.5.0" tracing.workspace = true diff --git a/crates/filemeta/src/fileinfo.rs b/crates/filemeta/src/fileinfo.rs index 1ae3c5afa..b9d754965 100644 --- a/crates/filemeta/src/fileinfo.rs +++ b/crates/filemeta/src/fileinfo.rs @@ -1,5 +1,6 @@ use crate::error::{Error, Result}; use crate::headers::RESERVED_METADATA_PREFIX_LOWER; +use bytes::Bytes; use rmp_serde::Serializer; use rustfs_utils::HashAlgorithm; use serde::Deserialize; @@ -36,7 +37,7 @@ pub struct ObjectPartInfo { pub struct ChecksumInfo { pub part_number: usize, pub algorithm: HashAlgorithm, - pub hash: Vec, + pub hash: Bytes, } #[derive(Debug, Serialize, Deserialize, PartialEq, Eq, PartialOrd, Default, Clone)] diff --git a/ecstore/src/disk/local.rs b/ecstore/src/disk/local.rs index 8d83006c3..25870964e 100644 --- a/ecstore/src/disk/local.rs +++ b/ecstore/src/disk/local.rs @@ -703,7 +703,7 @@ impl LocalDisk { let meta = file.metadata().await.map_err(to_file_error)?; let file_size = meta.len() as usize; - bitrot_verify(Box::new(file), file_size, part_size, algo, sum.to_vec(), shard_size) + bitrot_verify(Box::new(file), file_size, part_size, algo, bytes::Bytes::copy_from_slice(sum), shard_size) .await .map_err(to_file_error)?; diff --git a/ecstore/src/erasure_coding/bitrot.rs b/ecstore/src/erasure_coding/bitrot.rs index 367bf207c..63421d7be 100644 --- a/ecstore/src/erasure_coding/bitrot.rs +++ b/ecstore/src/erasure_coding/bitrot.rs @@ -1,3 +1,4 @@ +use bytes::Bytes; use pin_project_lite::pin_project; use rustfs_utils::{HashAlgorithm, read_full, write_all}; use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite}; @@ -174,7 +175,7 @@ pub async fn bitrot_verify( want_size: usize, part_size: usize, algo: HashAlgorithm, - _want: Vec, + _want: Bytes, // FIXME: useless parameter? mut shard_size: usize, ) -> std::io::Result<()> { let mut hash_buf = vec![0; algo.size()]; From 8309d2f8bea04ff62e171f51b6adb077d1fcf96b Mon Sep 17 00:00:00 2001 From: Nugine Date: Sun, 15 Jun 2025 21:01:38 +0800 Subject: [PATCH 3/6] refactor(filemeta): `FileInfo` `data` use `Bytes` --- crates/filemeta/src/fileinfo.rs | 4 ++-- crates/filemeta/src/filemeta.rs | 7 +++++-- ecstore/src/set_disk.rs | 5 +++-- 3 files changed, 10 insertions(+), 6 deletions(-) diff --git a/crates/filemeta/src/fileinfo.rs b/crates/filemeta/src/fileinfo.rs index b9d754965..69ccb34ce 100644 --- a/crates/filemeta/src/fileinfo.rs +++ b/crates/filemeta/src/fileinfo.rs @@ -168,13 +168,13 @@ pub struct FileInfo { pub mark_deleted: bool, // ReplicationState - Internal replication state to be passed back in ObjectInfo // pub replication_state: Option, // TODO: implement ReplicationState - pub data: Option>, + pub data: Option, pub num_versions: usize, pub successor_mod_time: Option, pub fresh: bool, pub idx: usize, // Combined checksum when object was uploaded - pub checksum: Option>, + pub checksum: Option, pub versioned: bool, } diff --git a/crates/filemeta/src/filemeta.rs b/crates/filemeta/src/filemeta.rs index 5ef8d3e2b..a9c5d86b0 100644 --- a/crates/filemeta/src/filemeta.rs +++ b/crates/filemeta/src/filemeta.rs @@ -419,7 +419,7 @@ impl FileMeta { if let Some(ref data) = fi.data { let key = vid.unwrap_or_default().to_string(); - self.data.replace(&key, data.clone())?; + self.data.replace(&key, data.to_vec())?; } let version = FileMetaVersion::from(fi); @@ -543,7 +543,10 @@ impl FileMeta { } if read_data { - fi.data = self.data.find(fi.version_id.unwrap_or_default().to_string().as_str())?; + fi.data = self + .data + .find(fi.version_id.unwrap_or_default().to_string().as_str())? + .map(bytes::Bytes::from); } fi.num_versions = self.versions.len(); diff --git a/ecstore/src/set_disk.rs b/ecstore/src/set_disk.rs index 0efbce668..05fa3893b 100644 --- a/ecstore/src/set_disk.rs +++ b/ecstore/src/set_disk.rs @@ -2594,7 +2594,8 @@ impl SetDisks { // if let Some(w) = writer.as_any().downcast_ref::() { // parts_metadata[index].data = Some(w.inline_data().to_vec()); // } - parts_metadata[index].data = Some(writer.into_inline_data().unwrap_or_default()); + parts_metadata[index].data = + Some(writer.into_inline_data().map(bytes::Bytes::from).unwrap_or_default()); } parts_metadata[index].set_inline_data(); } else { @@ -3920,7 +3921,7 @@ impl ObjectIO for SetDisks { for (i, fi) in parts_metadatas.iter_mut().enumerate() { if is_inline_buffer { if let Some(writer) = writers[i].take() { - fi.data = Some(writer.into_inline_data().unwrap_or_default()); + fi.data = Some(writer.into_inline_data().map(bytes::Bytes::from).unwrap_or_default()); } } From 3c5e20b633b7e4f141f80b9e0bae9674095390d5 Mon Sep 17 00:00:00 2001 From: Nugine Date: Sun, 15 Jun 2025 21:01:38 +0800 Subject: [PATCH 4/6] refactor(ecstore): `DiskAPI::rename_part` `meta` use `Bytes` --- ecstore/src/disk/local.rs | 5 +++-- ecstore/src/disk/mod.rs | 5 +++-- ecstore/src/disk/remote.rs | 5 +++-- ecstore/src/set_disk.rs | 5 +++-- rustfs/src/grpc.rs | 2 +- 5 files changed, 13 insertions(+), 9 deletions(-) diff --git a/ecstore/src/disk/local.rs b/ecstore/src/disk/local.rs index 25870964e..0771f0d88 100644 --- a/ecstore/src/disk/local.rs +++ b/ecstore/src/disk/local.rs @@ -38,6 +38,7 @@ use rustfs_utils::path::{ }; use crate::erasure_coding::bitrot_verify; +use bytes::Bytes; use common::defer; use path_absolutize::Absolutize; use rustfs_filemeta::{ @@ -1250,7 +1251,7 @@ impl DiskAPI for LocalDisk { } #[tracing::instrument(level = "debug", skip(self))] - async fn rename_part(&self, src_volume: &str, src_path: &str, dst_volume: &str, dst_path: &str, meta: Vec) -> Result<()> { + async fn rename_part(&self, src_volume: &str, src_path: &str, dst_volume: &str, dst_path: &str, meta: Bytes) -> Result<()> { let src_volume_dir = self.get_bucket_path(src_volume)?; let dst_volume_dir = self.get_bucket_path(dst_volume)?; if !skip_access_checks(src_volume) { @@ -1303,7 +1304,7 @@ impl DiskAPI for LocalDisk { rename_all(&src_file_path, &dst_file_path, &dst_volume_dir).await?; - self.write_all(dst_volume, format!("{}.meta", dst_path).as_str(), meta) + self.write_all(dst_volume, format!("{}.meta", dst_path).as_str(), meta.to_vec()) .await?; if let Some(parent) = src_file_path.parent() { diff --git a/ecstore/src/disk/mod.rs b/ecstore/src/disk/mod.rs index a6369808c..8fc016017 100644 --- a/ecstore/src/disk/mod.rs +++ b/ecstore/src/disk/mod.rs @@ -22,6 +22,7 @@ use crate::heal::{ data_usage_cache::{DataUsageCache, DataUsageEntry}, heal_commands::{HealScanMode, HealingTracker}, }; +use bytes::Bytes; use endpoint::Endpoint; use error::DiskError; use error::{Error, Result}; @@ -319,7 +320,7 @@ impl DiskAPI for Disk { } #[tracing::instrument(skip(self))] - async fn rename_part(&self, src_volume: &str, src_path: &str, dst_volume: &str, dst_path: &str, meta: Vec) -> Result<()> { + async fn rename_part(&self, src_volume: &str, src_path: &str, dst_volume: &str, dst_path: &str, meta: Bytes) -> Result<()> { match self { Disk::Local(local_disk) => local_disk.rename_part(src_volume, src_path, dst_volume, dst_path, meta).await, Disk::Remote(remote_disk) => { @@ -493,7 +494,7 @@ pub trait DiskAPI: Debug + Send + Sync + 'static { async fn create_file(&self, origvolume: &str, volume: &str, path: &str, file_size: usize) -> Result; // ReadFileStream async fn rename_file(&self, src_volume: &str, src_path: &str, dst_volume: &str, dst_path: &str) -> Result<()>; - async fn rename_part(&self, src_volume: &str, src_path: &str, dst_volume: &str, dst_path: &str, meta: Vec) -> Result<()>; + async fn rename_part(&self, src_volume: &str, src_path: &str, dst_volume: &str, dst_path: &str, meta: Bytes) -> Result<()>; async fn delete(&self, volume: &str, path: &str, opt: DeleteOptions) -> Result<()>; // VerifyFile async fn verify_file(&self, volume: &str, path: &str, fi: &FileInfo) -> Result; diff --git a/ecstore/src/disk/remote.rs b/ecstore/src/disk/remote.rs index 511022cc6..7160dd545 100644 --- a/ecstore/src/disk/remote.rs +++ b/ecstore/src/disk/remote.rs @@ -1,5 +1,6 @@ use std::path::PathBuf; +use bytes::Bytes; use futures::lock::Mutex; use http::{HeaderMap, Method}; use protos::{ @@ -649,7 +650,7 @@ impl DiskAPI for RemoteDisk { } #[tracing::instrument(skip(self))] - async fn rename_part(&self, src_volume: &str, src_path: &str, dst_volume: &str, dst_path: &str, meta: Vec) -> Result<()> { + async fn rename_part(&self, src_volume: &str, src_path: &str, dst_volume: &str, dst_path: &str, meta: Bytes) -> Result<()> { info!("rename_part {}/{}", src_volume, src_path); let mut client = node_service_time_out_client(&self.addr) .await @@ -660,7 +661,7 @@ impl DiskAPI for RemoteDisk { src_path: src_path.to_string(), dst_volume: dst_volume.to_string(), dst_path: dst_path.to_string(), - meta, + meta: meta.to_vec(), }); let response = client.rename_part(request).await?.into_inner(); diff --git a/ecstore/src/set_disk.rs b/ecstore/src/set_disk.rs index 05fa3893b..271cc4c9a 100644 --- a/ecstore/src/set_disk.rs +++ b/ecstore/src/set_disk.rs @@ -45,6 +45,7 @@ use crate::{ heal::data_scanner::{HEAL_DELETE_DANGLING, globalHealConfig}, store_api::ListObjectVersionsInfo, }; +use bytes::Bytes; use bytesize::ByteSize; use chrono::Utc; use futures::future::join_all; @@ -489,7 +490,7 @@ impl SetDisks { src_object: &str, dst_bucket: &str, dst_object: &str, - meta: Vec, + meta: Bytes, write_quorum: usize, ) -> disk::error::Result>> { let src_bucket = Arc::new(src_bucket.to_string()); @@ -4600,7 +4601,7 @@ impl StorageAPI for SetDisks { &tmp_part_path, RUSTFS_META_MULTIPART_BUCKET, &part_path, - fi_buff, + fi_buff.into(), write_quorum, ) .await?; diff --git a/rustfs/src/grpc.rs b/rustfs/src/grpc.rs index 6c155edc0..80b957909 100644 --- a/rustfs/src/grpc.rs +++ b/rustfs/src/grpc.rs @@ -446,7 +446,7 @@ impl Node for NodeService { &request.src_path, &request.dst_volume, &request.dst_path, - request.meta, + request.meta.into(), ) .await { From 2f3dbac59be6fd0701f09add4c6596f26ff6f6c8 Mon Sep 17 00:00:00 2001 From: Nugine Date: Sun, 15 Jun 2025 21:01:38 +0800 Subject: [PATCH 5/6] feat(ecstore): `LocalDisk::write_all_internal` use `InternalBuf` --- ecstore/src/disk/local.rs | 52 ++++++++++++++++++++++++--------------- 1 file changed, 32 insertions(+), 20 deletions(-) diff --git a/ecstore/src/disk/local.rs b/ecstore/src/disk/local.rs index 0771f0d88..5b163c5bb 100644 --- a/ecstore/src/disk/local.rs +++ b/ecstore/src/disk/local.rs @@ -83,6 +83,12 @@ impl FormatInfo { } } +/// A helper enum to handle internal buffer types for writing data. +pub enum InternalBuf<'a> { + Ref(&'a [u8]), + Owned(Bytes), +} + pub struct LocalDisk { pub root: PathBuf, pub format_path: PathBuf, @@ -596,8 +602,14 @@ impl LocalDisk { let volume_dir = self.get_bucket_path(volume)?; - self.write_all_private(volume, format!("{}/{}", path, STORAGE_FORMAT_FILE).as_str(), &buf, true, volume_dir) - .await?; + self.write_all_private( + volume, + format!("{}/{}", path, STORAGE_FORMAT_FILE).as_str(), + buf.into(), + true, + &volume_dir, + ) + .await?; Ok(()) } @@ -610,7 +622,8 @@ impl LocalDisk { let tmp_volume_dir = self.get_bucket_path(super::RUSTFS_META_TMP_BUCKET)?; let tmp_file_path = tmp_volume_dir.join(Path::new(Uuid::new_v4().to_string().as_str())); - self.write_all_internal(&tmp_file_path, buf, sync, tmp_volume_dir).await?; + self.write_all_internal(&tmp_file_path, InternalBuf::Ref(buf), sync, &tmp_volume_dir) + .await?; rename_all(tmp_file_path, file_path, volume_dir).await } @@ -624,47 +637,46 @@ impl LocalDisk { let volume_dir = self.get_bucket_path(volume)?; - self.write_all_private(volume, path, &data, true, volume_dir).await?; + self.write_all_private(volume, path, data.into(), true, &volume_dir).await?; Ok(()) } // write_all_private with check_path_length #[tracing::instrument(level = "debug", skip_all)] - pub async fn write_all_private( - &self, - volume: &str, - path: &str, - buf: &[u8], - sync: bool, - skip_parent: impl AsRef, - ) -> Result<()> { + pub async fn write_all_private(&self, volume: &str, path: &str, buf: Bytes, sync: bool, skip_parent: &Path) -> Result<()> { let volume_dir = self.get_bucket_path(volume)?; let file_path = volume_dir.join(Path::new(&path)); check_path_length(file_path.to_string_lossy().as_ref())?; - self.write_all_internal(file_path, buf, sync, skip_parent).await + self.write_all_internal(&file_path, InternalBuf::Owned(buf), sync, skip_parent) + .await } // write_all_internal do write file pub async fn write_all_internal( &self, - file_path: impl AsRef, - data: impl AsRef<[u8]>, + file_path: &Path, + data: InternalBuf<'_>, sync: bool, - skip_parent: impl AsRef, + skip_parent: &Path, ) -> Result<()> { let flags = O_CREATE | O_WRONLY | O_TRUNC; let mut f = { if sync { // TODO: suport sync - self.open_file(file_path.as_ref(), flags, skip_parent.as_ref()).await? + self.open_file(file_path, flags, skip_parent).await? } else { - self.open_file(file_path.as_ref(), flags, skip_parent.as_ref()).await? + self.open_file(file_path, flags, skip_parent).await? } }; - f.write_all(data.as_ref()).await.map_err(to_file_error)?; + let data: &[u8] = match &data { + InternalBuf::Ref(buf) => buf, + InternalBuf::Owned(buf) => buf.as_ref(), + }; + + f.write_all(data).await.map_err(to_file_error)?; Ok(()) } @@ -1691,7 +1703,7 @@ impl DiskAPI for LocalDisk { .write_all_private( dst_volume, format!("{}/{}/{}", &dst_path, &old_data_dir.to_string(), STORAGE_FORMAT_FILE).as_str(), - &dst_buf, + dst_buf.into(), true, &skip_parent, ) From 82cc1402c46ce189234b564f8b57de8c35c22ef2 Mon Sep 17 00:00:00 2001 From: Nugine Date: Sun, 15 Jun 2025 21:01:38 +0800 Subject: [PATCH 6/6] feat(ecstore): LocalDisk writes file by spawn_blocking --- ecstore/src/disk/local.rs | 21 +++++++++++++++------ 1 file changed, 15 insertions(+), 6 deletions(-) diff --git a/ecstore/src/disk/local.rs b/ecstore/src/disk/local.rs index 5b163c5bb..e13dbd549 100644 --- a/ecstore/src/disk/local.rs +++ b/ecstore/src/disk/local.rs @@ -671,12 +671,21 @@ impl LocalDisk { } }; - let data: &[u8] = match &data { - InternalBuf::Ref(buf) => buf, - InternalBuf::Owned(buf) => buf.as_ref(), - }; - - f.write_all(data).await.map_err(to_file_error)?; + match data { + InternalBuf::Ref(buf) => { + f.write_all(buf).await.map_err(to_file_error)?; + } + InternalBuf::Owned(buf) => { + // Reduce one copy by using the owned buffer directly. + // It may be more efficient for larger writes. + let mut f = f.into_std().await; + let task = tokio::task::spawn_blocking(move || { + use std::io::Write as _; + f.write_all(buf.as_ref()).map_err(to_file_error) + }); + task.await??; + } + } Ok(()) }