Merge pull request #475 from rustfs/nugine/refactor/use-bytes

refactor: use `Bytes` for data buffers

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",

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"

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

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<u8>,
+    pub hash: Bytes,
 }
 
 #[derive(Debug, Serialize, Deserialize, PartialEq, Eq, PartialOrd, Default, Clone)]
@@ -167,13 +168,13 @@ pub struct FileInfo {
     pub mark_deleted: bool,
     // ReplicationState - Internal replication state to be passed back in ObjectInfo
     // pub replication_state: Option<ReplicationState>, // TODO: implement ReplicationState
-    pub data: Option<Vec<u8>>,
+    pub data: Option<Bytes>,
     pub num_versions: usize,
     pub successor_mod_time: Option<OffsetDateTime>,
     pub fresh: bool,
     pub idx: usize,
     // Combined checksum when object was uploaded
-    pub checksum: Option<Vec<u8>>,
+    pub checksum: Option<Bytes>,
     pub versioned: bool,
 }
 

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();

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::{
@@ -82,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,
@@ -595,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)
+        self.write_all_private(
-            .await?;
+            volume,
+            format!("{}/{}", path, STORAGE_FORMAT_FILE).as_str(),
+            buf.into(),
+            true,
+            &volume_dir,
+        )
+        .await?;
 
         Ok(())
     }
@@ -609,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
     }
@@ -623,47 +637,55 @@ 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(
+    pub async fn write_all_private(&self, volume: &str, path: &str, buf: Bytes, sync: bool, skip_parent: &Path) -> Result<()> {
-        &self,
-        volume: &str,
-        path: &str,
-        buf: &[u8],
-        sync: bool,
-        skip_parent: impl AsRef<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<Path>,
+        file_path: &Path,
-        data: impl AsRef<[u8]>,
+        data: InternalBuf<'_>,
         sync: bool,
-        skip_parent: impl AsRef<Path>,
+        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)?;
+        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(())
     }
@@ -703,7 +725,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)?;
 
@@ -1250,7 +1272,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<u8>) -> 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 +1325,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() {
@@ -1690,7 +1712,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,
                     )

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<u8>) -> 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<FileWriter>;
     // 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<u8>) -> 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<CheckPartsResp>;

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<u8>) -> 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();

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<R: AsyncRead + Unpin + Send>(
     want_size: usize,
     part_size: usize,
     algo: HashAlgorithm,
-    _want: Vec<u8>,
+    _want: Bytes, // FIXME: useless parameter?
     mut shard_size: usize,
 ) -> std::io::Result<()> {
     let mut hash_buf = vec![0; algo.size()];

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<u8>,
+        meta: Bytes,
         write_quorum: usize,
     ) -> disk::error::Result<Vec<Option<DiskStore>>> {
         let src_bucket = Arc::new(src_bucket.to_string());
@@ -2594,7 +2595,8 @@ impl SetDisks {
                                             // if let Some(w) = writer.as_any().downcast_ref::<BitrotFileWriter>() {
                                             //     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 +3922,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());
                 }
             }
 
@@ -4599,7 +4601,7 @@ impl StorageAPI for SetDisks {
             &tmp_part_path,
             RUSTFS_META_MULTIPART_BUCKET,
             &part_path,
-            fi_buff,
+            fi_buff.into(),
             write_quorum,
         )
         .await?;

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
             {