diff --git a/common/protos/src/generated/flatbuffers_generated/models.rs b/common/protos/src/generated/flatbuffers_generated/models.rs index e4949fdcf..d55f1a98d 100644 --- a/common/protos/src/generated/flatbuffers_generated/models.rs +++ b/common/protos/src/generated/flatbuffers_generated/models.rs @@ -29,7 +29,7 @@ pub mod models { #[inline] unsafe fn follow(buf: &'a [u8], loc: usize) -> Self::Inner { Self { - _tab: flatbuffers::Table::new(buf, loc), + _tab: unsafe { flatbuffers::Table::new(buf, loc) }, } } } diff --git a/crates/filemeta/src/fileinfo.rs b/crates/filemeta/src/fileinfo.rs index 6ff41e6f1..44da0d23d 100644 --- a/crates/filemeta/src/fileinfo.rs +++ b/crates/filemeta/src/fileinfo.rs @@ -312,7 +312,7 @@ impl FileInfo { /// Check if the object is remote (transitioned to another tier) pub fn is_remote(&self) -> bool { - !self.transition_tier.as_ref().map_or(true, |s| s.is_empty()) + !self.transition_tier.as_ref().is_none_or(|s| s.is_empty()) } /// Get the data directory for this object diff --git a/crates/rio/src/lib.rs b/crates/rio/src/lib.rs index 86d82ec95..1eec035b2 100644 --- a/crates/rio/src/lib.rs +++ b/crates/rio/src/lib.rs @@ -21,6 +21,7 @@ pub use hash_reader::*; pub mod compress; pub mod reader; +pub use reader::WarpReader; mod writer; use tokio::io::{AsyncRead, BufReader}; diff --git a/crates/rio/src/reader.rs b/crates/rio/src/reader.rs index 8b1378917..88ed8b312 100644 --- a/crates/rio/src/reader.rs +++ b/crates/rio/src/reader.rs @@ -1 +1,27 @@ +use std::pin::Pin; +use std::task::{Context, Poll}; +use tokio::io::{AsyncRead, ReadBuf}; +use crate::{EtagResolvable, HashReaderDetector, Reader}; + +pub struct WarpReader { + inner: R, +} + +impl WarpReader { + pub fn new(inner: R) -> Self { + Self { inner } + } +} + +impl AsyncRead for WarpReader { + fn poll_read(mut self: Pin<&mut Self>, cx: &mut Context<'_>, buf: &mut ReadBuf<'_>) -> Poll> { + Pin::new(&mut self.inner).poll_read(cx, buf) + } +} + +impl HashReaderDetector for WarpReader {} + +impl EtagResolvable for WarpReader {} + +impl Reader for WarpReader {} diff --git a/crates/rio/src/writer.rs b/crates/rio/src/writer.rs index 686b5a132..d81bf015d 100644 --- a/crates/rio/src/writer.rs +++ b/crates/rio/src/writer.rs @@ -1,8 +1,6 @@ use std::io::Cursor; use std::pin::Pin; -use std::task::{Context, Poll}; use tokio::io::AsyncWrite; -use tokio::io::AsyncWriteExt; use crate::HttpWriter; @@ -92,77 +90,3 @@ impl AsyncWrite for Writer { } } } - -/// WriterAll wraps a Writer and ensures each write writes the entire buffer (like write_all). -pub struct WriterAll { - inner: W, -} - -impl WriterAll { - pub fn new(inner: W) -> Self { - Self { inner } - } - - /// Write the entire buffer, like write_all. - pub async fn write_all(&mut self, mut buf: &[u8]) -> std::io::Result<()> { - while !buf.is_empty() { - let n = self.inner.write(buf).await?; - if n == 0 { - return Err(std::io::Error::new(std::io::ErrorKind::WriteZero, "failed to write whole buffer")); - } - buf = &buf[n..]; - } - Ok(()) - } - - /// Get a mutable reference to the inner writer. - pub fn get_mut(&mut self) -> &mut W { - &mut self.inner - } -} - -impl AsyncWrite for WriterAll { - fn poll_write(mut self: Pin<&mut Self>, cx: &mut Context<'_>, mut buf: &[u8]) -> Poll> { - let mut total_written = 0; - while !buf.is_empty() { - // Safety: W: Unpin - let inner_pin = Pin::new(&mut self.inner); - match inner_pin.poll_write(cx, buf) { - Poll::Ready(Ok(0)) => { - if total_written == 0 { - return Poll::Ready(Ok(0)); - } else { - return Poll::Ready(Ok(total_written)); - } - } - Poll::Ready(Ok(n)) => { - total_written += n; - buf = &buf[n..]; - } - Poll::Ready(Err(e)) => { - if total_written == 0 { - return Poll::Ready(Err(e)); - } else { - return Poll::Ready(Ok(total_written)); - } - } - Poll::Pending => { - if total_written == 0 { - return Poll::Pending; - } else { - return Poll::Ready(Ok(total_written)); - } - } - } - } - Poll::Ready(Ok(total_written)) - } - - fn poll_flush(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { - Pin::new(&mut self.inner).poll_flush(cx) - } - - fn poll_shutdown(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { - Pin::new(&mut self.inner).poll_shutdown(cx) - } -} diff --git a/ecstore/src/disk/error.rs b/ecstore/src/disk/error.rs index 4c146a62f..f757895f3 100644 --- a/ecstore/src/disk/error.rs +++ b/ecstore/src/disk/error.rs @@ -253,11 +253,11 @@ impl From for DiskError { } } -impl Into for DiskError { - fn into(self) -> protos::proto_gen::node_service::Error { +impl From for protos::proto_gen::node_service::Error { + fn from(e: DiskError) -> Self { protos::proto_gen::node_service::Error { - code: self.to_u32(), - error_info: self.to_string(), + code: e.to_u32(), + error_info: e.to_string(), } } } diff --git a/ecstore/src/disk/error_reduce.rs b/ecstore/src/disk/error_reduce.rs index 3d08a3719..f25dd28a2 100644 --- a/ecstore/src/disk/error_reduce.rs +++ b/ecstore/src/disk/error_reduce.rs @@ -29,11 +29,7 @@ pub fn reduce_read_quorum_errs(errors: &[Option], ignored_errs: &[Error], pub fn reduce_quorum_errs(errors: &[Option], ignored_errs: &[Error], quorun: usize, quorun_err: Error) -> Option { let (max_count, err) = reduce_errs(errors, ignored_errs); - if max_count >= quorun { - err - } else { - Some(quorun_err) - } + if max_count >= quorun { err } else { Some(quorun_err) } } pub fn reduce_errs(errors: &[Option], ignored_errs: &[Error]) -> (usize, Option) { @@ -59,7 +55,7 @@ pub fn reduce_errs(errors: &[Option], ignored_errs: &[Error]) -> (usize, match (e1.to_string().as_str(), e2.to_string().as_str()) { ("nil", _) => std::cmp::Ordering::Greater, (_, "nil") => std::cmp::Ordering::Less, - (a, b) => a.cmp(&b), + (a, b) => a.cmp(b), } } else { count_cmp diff --git a/ecstore/src/disk/local.rs b/ecstore/src/disk/local.rs index 75751ec1f..ff0299af0 100644 --- a/ecstore/src/disk/local.rs +++ b/ecstore/src/disk/local.rs @@ -10,7 +10,6 @@ use super::{endpoint::Endpoint, error::DiskError, format::FormatV3}; use crate::bucket::metadata_sys::{self}; use crate::bucket::versioning::VersioningApi; use crate::bucket::versioning_sys::BucketVersioningSys; -use crate::disk::STORAGE_FORMAT_FILE; use crate::disk::error::FileAccessDeniedWithContext; use crate::disk::error_conv::{to_access_error, to_file_error, to_unformatted_disk_error, to_volume_error}; use crate::disk::fs::{ @@ -19,8 +18,9 @@ use crate::disk::fs::{ use crate::disk::os::{check_path_length, is_empty_dir}; use crate::disk::{ CHECK_PART_FILE_CORRUPT, CHECK_PART_FILE_NOT_FOUND, CHECK_PART_SUCCESS, CHECK_PART_UNKNOWN, CHECK_PART_VOLUME_NOT_FOUND, - conv_part_err_to_int, + FileReader, conv_part_err_to_int, }; +use crate::disk::{FileWriter, STORAGE_FORMAT_FILE}; use crate::global::{GLOBAL_IsErasureSD, GLOBAL_RootDiskThreshold}; use crate::heal::data_scanner::{ ScannerItem, ShouldSleepFn, SizeSummary, lc_has_active_rules, rep_has_active_rules, scan_data_folder, @@ -30,7 +30,6 @@ use crate::heal::data_usage_cache::{DataUsageCache, DataUsageEntry}; use crate::heal::error::{ERR_IGNORE_FILE_CONTRIB, ERR_SKIP_FILE}; use crate::heal::heal_commands::{HealScanMode, HealingTracker}; use crate::heal::heal_ops::HEALING_TRACKER_FILENAME; -use crate::io::FileWriter; use crate::new_object_layer_fn; use crate::store_api::{ObjectInfo, StorageAPI}; use crate::utils::os::get_info; @@ -45,7 +44,7 @@ use rustfs_filemeta::{ Cache, FileInfo, FileInfoOpts, FileMeta, MetaCacheEntry, MetacacheWriter, Opts, RawFileInfo, UpdateFn, get_file_info, read_xl_meta_no_data, }; -use rustfs_rio::{Reader, bitrot_verify}; +use rustfs_rio::bitrot_verify; use rustfs_utils::HashAlgorithm; use std::collections::{HashMap, HashSet}; use std::fmt::Debug; @@ -1299,7 +1298,7 @@ impl DiskAPI for LocalDisk { } } - remove_std(&dst_file_path).map_err(|e| to_file_error(e))?; + remove_std(&dst_file_path).map_err(to_file_error)?; } rename_all(&src_file_path, &dst_file_path, &dst_volume_dir).await?; @@ -1356,11 +1355,11 @@ impl DiskAPI for LocalDisk { if let Some(meta) = meta_op { if !meta.is_dir() { - return Err(DiskError::FileAccessDenied.into()); + return Err(DiskError::FileAccessDenied); } } - remove(&dst_file_path).await.map_err(|e| to_file_error(e))?; + remove(&dst_file_path).await.map_err(to_file_error)?; } rename_all(&src_file_path, &dst_file_path, &dst_volume_dir).await?; @@ -1425,7 +1424,7 @@ impl DiskAPI for LocalDisk { // TODO: io verifier #[tracing::instrument(level = "debug", skip(self))] - async fn read_file(&self, volume: &str, path: &str) -> Result> { + async fn read_file(&self, volume: &str, path: &str) -> Result { // warn!("disk read_file: volume: {}, path: {}", volume, path); let volume_dir = self.get_bucket_path(volume)?; if !skip_access_checks(volume) { @@ -1443,7 +1442,7 @@ impl DiskAPI for LocalDisk { } #[tracing::instrument(level = "debug", skip(self))] - async fn read_file_stream(&self, volume: &str, path: &str, offset: usize, length: usize) -> Result> { + async fn read_file_stream(&self, volume: &str, path: &str, offset: usize, length: usize) -> Result { // warn!( // "disk read_file_stream: volume: {}, path: {}, offset: {}, length: {}", // volume, path, offset, length @@ -1615,7 +1614,7 @@ impl DiskAPI for LocalDisk { let e: DiskError = to_file_error(e).into(); if e != DiskError::FileNotFound { - return Err(e.into()); + return Err(e); } None diff --git a/ecstore/src/disk/mod.rs b/ecstore/src/disk/mod.rs index 17dabe6ae..6c613e08e 100644 --- a/ecstore/src/disk/mod.rs +++ b/ecstore/src/disk/mod.rs @@ -17,13 +17,10 @@ pub const FORMAT_CONFIG_FILE: &str = "format.json"; pub const STORAGE_FORMAT_FILE: &str = "xl.meta"; pub const STORAGE_FORMAT_FILE_BACKUP: &str = "xl.meta.bkp"; -use crate::{ - heal::{ - data_scanner::ShouldSleepFn, - data_usage_cache::{DataUsageCache, DataUsageEntry}, - heal_commands::{HealScanMode, HealingTracker}, - }, - io::FileWriter, +use crate::heal::{ + data_scanner::ShouldSleepFn, + data_usage_cache::{DataUsageCache, DataUsageEntry}, + heal_commands::{HealScanMode, HealingTracker}, }; use endpoint::Endpoint; use error::DiskError; @@ -32,16 +29,21 @@ use local::LocalDisk; use madmin::info_commands::DiskMetrics; use remote::RemoteDisk; use rustfs_filemeta::{FileInfo, RawFileInfo}; -use rustfs_rio::Reader; use serde::{Deserialize, Serialize}; use std::{fmt::Debug, path::PathBuf, sync::Arc}; use time::OffsetDateTime; -use tokio::{io::AsyncWrite, sync::mpsc::Sender}; +use tokio::{ + io::{AsyncRead, AsyncWrite}, + sync::mpsc::Sender, +}; use tracing::warn; use uuid::Uuid; pub type DiskStore = Arc; +pub type FileReader = Box; +pub type FileWriter = Box; + #[derive(Debug)] pub enum Disk { Local(Box), @@ -277,7 +279,7 @@ impl DiskAPI for Disk { } #[tracing::instrument(skip(self))] - async fn read_file(&self, volume: &str, path: &str) -> Result> { + async fn read_file(&self, volume: &str, path: &str) -> Result { match self { Disk::Local(local_disk) => local_disk.read_file(volume, path).await, Disk::Remote(remote_disk) => remote_disk.read_file(volume, path).await, @@ -285,7 +287,7 @@ impl DiskAPI for Disk { } #[tracing::instrument(skip(self))] - async fn read_file_stream(&self, volume: &str, path: &str, offset: usize, length: usize) -> Result> { + async fn read_file_stream(&self, volume: &str, path: &str, offset: usize, length: usize) -> Result { match self { Disk::Local(local_disk) => local_disk.read_file_stream(volume, path, offset, length).await, Disk::Remote(remote_disk) => remote_disk.read_file_stream(volume, path, offset, length).await, @@ -485,8 +487,8 @@ pub trait DiskAPI: Debug + Send + Sync + 'static { // File operations. // 读目录下的所有文件、目录 async fn list_dir(&self, origvolume: &str, volume: &str, dir_path: &str, count: i32) -> Result>; - async fn read_file(&self, volume: &str, path: &str) -> Result>; - async fn read_file_stream(&self, volume: &str, path: &str, offset: usize, length: usize) -> Result>; + async fn read_file(&self, volume: &str, path: &str) -> Result; + async fn read_file_stream(&self, volume: &str, path: &str, offset: usize, length: usize) -> Result; async fn append_file(&self, volume: &str, path: &str) -> Result; async fn create_file(&self, origvolume: &str, volume: &str, path: &str, file_size: usize) -> Result; // ReadFileStream diff --git a/ecstore/src/disk/os.rs b/ecstore/src/disk/os.rs index 8496cad4f..d21e88b5c 100644 --- a/ecstore/src/disk/os.rs +++ b/ecstore/src/disk/os.rs @@ -108,7 +108,7 @@ pub async fn rename_all( ) -> Result<()> { reliable_rename(src_file_path, dst_file_path.as_ref(), base_dir) .await - .map_err(|e| to_file_error(e))?; + .map_err(to_file_error)?; Ok(()) } diff --git a/ecstore/src/disk/remote.rs b/ecstore/src/disk/remote.rs index 8818c6321..8cca548df 100644 --- a/ecstore/src/disk/remote.rs +++ b/ecstore/src/disk/remote.rs @@ -13,30 +13,33 @@ use protos::{ }; use rmp_serde::Serializer; use rustfs_filemeta::{FileInfo, MetaCacheEntry, MetacacheWriter, RawFileInfo}; -use rustfs_rio::{HttpReader, Reader}; +use rustfs_rio::{HttpReader, HttpWriter}; use serde::Serialize; use tokio::{ io::AsyncWrite, sync::mpsc::{self, Sender}, }; -use tokio_stream::{wrappers::ReceiverStream, StreamExt}; +use tokio_stream::{StreamExt, wrappers::ReceiverStream}; use tonic::Request; use tracing::info; use uuid::Uuid; use super::error::{Error, Result}; use super::{ - endpoint::Endpoint, CheckPartsResp, DeleteOptions, DiskAPI, DiskInfo, DiskInfoOptions, DiskLocation, DiskOption, - FileInfoVersions, ReadMultipleReq, ReadMultipleResp, ReadOptions, RenameDataResp, UpdateMetadataOpts, VolumeInfo, - WalkDirOptions, + CheckPartsResp, DeleteOptions, DiskAPI, DiskInfo, DiskInfoOptions, DiskLocation, DiskOption, FileInfoVersions, + ReadMultipleReq, ReadMultipleResp, ReadOptions, RenameDataResp, UpdateMetadataOpts, VolumeInfo, WalkDirOptions, + endpoint::Endpoint, }; -use crate::heal::{ - data_scanner::ShouldSleepFn, - data_usage_cache::{DataUsageCache, DataUsageEntry}, - heal_commands::{HealScanMode, HealingTracker}, +use crate::{ + disk::{FileReader, FileWriter}, + heal::{ + data_scanner::ShouldSleepFn, + data_usage_cache::{DataUsageCache, DataUsageEntry}, + heal_commands::{HealScanMode, HealingTracker}, + }, }; -use crate::io::{FileWriter, HttpFileWriter}; + use protos::proto_gen::node_service::RenamePartRequst; #[derive(Debug)] @@ -552,7 +555,7 @@ impl DiskAPI for RemoteDisk { } #[tracing::instrument(level = "debug", skip(self))] - async fn read_file(&self, volume: &str, path: &str) -> Result> { + async fn read_file(&self, volume: &str, path: &str) -> Result { info!("read_file {}/{}", volume, path); let url = format!( @@ -569,7 +572,7 @@ impl DiskAPI for RemoteDisk { } #[tracing::instrument(level = "debug", skip(self))] - async fn read_file_stream(&self, volume: &str, path: &str, offset: usize, length: usize) -> Result> { + async fn read_file_stream(&self, volume: &str, path: &str, offset: usize, length: usize) -> Result { info!("read_file_stream {}/{}/{}", self.endpoint.to_string(), volume, path); let url = format!( "{}/rustfs/rpc/read_file_stream?disk={}&volume={}&path={}&offset={}&length={}", @@ -587,27 +590,35 @@ impl DiskAPI for RemoteDisk { #[tracing::instrument(level = "debug", skip(self))] async fn append_file(&self, volume: &str, path: &str) -> Result { info!("append_file {}/{}", volume, path); - Ok(Box::new(HttpFileWriter::new( - self.endpoint.grid_host().as_str(), - self.endpoint.to_string().as_str(), - volume, - path, - 0, + + let url = format!( + "{}/rustfs/rpc/put_file_stream?disk={}&volume={}&path={}&append={}&size={}", + self.endpoint.grid_host(), + urlencoding::encode(self.endpoint.to_string().as_str()), + urlencoding::encode(volume), + urlencoding::encode(path), true, - )?)) + 0 + ); + + Ok(Box::new(HttpWriter::new(url, Method::PUT, HeaderMap::new()).await?)) } #[tracing::instrument(level = "debug", skip(self))] async fn create_file(&self, _origvolume: &str, volume: &str, path: &str, file_size: usize) -> Result { info!("create_file {}/{}/{}", self.endpoint.to_string(), volume, path); - Ok(Box::new(HttpFileWriter::new( - self.endpoint.grid_host().as_str(), - self.endpoint.to_string().as_str(), - volume, - path, - file_size, + + let url = format!( + "{}/rustfs/rpc/put_file_stream?disk={}&volume={}&path={}&append={}&size={}", + self.endpoint.grid_host(), + urlencoding::encode(self.endpoint.to_string().as_str()), + urlencoding::encode(volume), + urlencoding::encode(path), false, - )?)) + file_size + ); + + Ok(Box::new(HttpWriter::new(url, Method::PUT, HeaderMap::new()).await?)) } #[tracing::instrument(level = "debug", skip(self))] diff --git a/ecstore/src/io.rs b/ecstore/src/io.rs index e0fd2106e..d33cb8e52 100644 --- a/ecstore/src/io.rs +++ b/ecstore/src/io.rs @@ -6,9 +6,9 @@ use md5::Md5; use pin_project_lite::pin_project; use std::io; use std::pin::Pin; -use std::task::ready; use std::task::Context; use std::task::Poll; +use std::task::ready; use tokio::io::AsyncRead; use tokio::io::AsyncWrite; use tokio::io::ReadBuf; diff --git a/ecstore/src/lib.rs b/ecstore/src/lib.rs index 1c27f5930..17bfe8a67 100644 --- a/ecstore/src/lib.rs +++ b/ecstore/src/lib.rs @@ -13,7 +13,7 @@ pub mod error; // pub mod file_meta_inline; pub mod global; pub mod heal; -pub mod io; +// pub mod io; // pub mod metacache; pub mod metrics_realtime; pub mod notification_sys; diff --git a/ecstore/src/peer.rs b/ecstore/src/peer.rs index e203354c7..71911a857 100644 --- a/ecstore/src/peer.rs +++ b/ecstore/src/peer.rs @@ -1,9 +1,9 @@ use crate::disk::error::{Error, Result}; -use crate::disk::error_reduce::{is_all_buckets_not_found, reduce_write_quorum_errs, BUCKET_OP_IGNORED_ERRS}; +use crate::disk::error_reduce::{BUCKET_OP_IGNORED_ERRS, is_all_buckets_not_found, reduce_write_quorum_errs}; use crate::disk::{DiskAPI, DiskStore}; use crate::global::GLOBAL_LOCAL_DISK_MAP; use crate::heal::heal_commands::{ - HealOpts, DRIVE_STATE_CORRUPT, DRIVE_STATE_MISSING, DRIVE_STATE_OFFLINE, DRIVE_STATE_OK, HEAL_ITEM_BUCKET, + DRIVE_STATE_CORRUPT, DRIVE_STATE_MISSING, DRIVE_STATE_OFFLINE, DRIVE_STATE_OK, HEAL_ITEM_BUCKET, HealOpts, }; use crate::heal::heal_ops::RUSTFS_RESERVED_BUCKET; use crate::store::all_local_disk; @@ -137,7 +137,7 @@ impl S3PeerSys { return Ok(heal_bucket_results.read().await[i].clone()); } } - Err(Error::VolumeNotFound.into()) + Err(Error::VolumeNotFound) } pub async fn make_bucket(&self, bucket: &str, opts: &MakeBucketOptions) -> Result<()> { @@ -771,7 +771,7 @@ pub async fn heal_bucket_local(bucket: &str, opts: &HealOpts) -> Result>(); + let errs_clone = errs.iter().map(|e| e.clone()).collect::>(); futures.push(async move { if bs_clone.read().await[idx] == DRIVE_STATE_MISSING { info!("bucket not find, will recreate"); @@ -785,7 +785,7 @@ pub async fn heal_bucket_local(bucket: &str, opts: &HealOpts) -> Result>, duration: Duration) -> Result { - if !self.pools.get(idx).is_some_and(|v| v.decommission.is_some()) { + if self.pools.get(idx).is_none_or(|v| v.decommission.is_none()) { return Err(Error::other("InvalidArgument")); } @@ -882,7 +882,7 @@ impl ECStore { pool: Arc, bi: DecomBucketInfo, ) -> Result<()> { - let wk = Workers::new(pool.disk_set.len() * 2).map_err(|v| Error::other(v))?; + let wk = Workers::new(pool.disk_set.len() * 2).map_err(Error::other)?; // let mut vc = None; // replication diff --git a/ecstore/src/rebalance.rs b/ecstore/src/rebalance.rs index a99cac2dd..11cea68c4 100644 --- a/ecstore/src/rebalance.rs +++ b/ecstore/src/rebalance.rs @@ -2,18 +2,18 @@ use std::io::Cursor; use std::sync::Arc; use std::time::SystemTime; -use crate::cache_value::metacache_set::{list_path_raw, ListPathRawOptions}; +use crate::StorageAPI; +use crate::cache_value::metacache_set::{ListPathRawOptions, list_path_raw}; use crate::config::com::{read_config_with_metadata, save_config_with_opts}; use crate::disk::error::DiskError; -use crate::error::{is_err_data_movement_overwrite, is_err_object_not_found, is_err_version_not_found}; use crate::error::{Error, Result}; +use crate::error::{is_err_data_movement_overwrite, is_err_object_not_found, is_err_version_not_found}; use crate::global::get_global_endpoints; use crate::pools::ListCallback; use crate::set_disk::SetDisks; use crate::store::ECStore; use crate::store_api::{CompletePart, GetObjectReader, ObjectIO, ObjectOptions, PutObjReader}; use crate::utils::path::encode_dir_object; -use crate::StorageAPI; use common::defer; use http::HeaderMap; use rustfs_filemeta::{FileInfo, MetaCacheEntries, MetaCacheEntry, MetadataResolutionParams}; @@ -500,7 +500,7 @@ impl ECStore { if get_global_endpoints() .as_ref() .get(idx) - .map_or(true, |v| v.endpoints.as_ref().first().map_or(true, |e| e.is_local)) + .is_none_or(|v| v.endpoints.as_ref().first().map_or(true, |e| e.is_local)) { warn!("start_rebalance: pool {} is not local, skipping", idx); continue; @@ -957,7 +957,7 @@ impl ECStore { let pool = self.pools[pool_index].clone(); - let wk = Workers::new(pool.disk_set.len() * 2).map_err(|v| Error::other(v))?; + let wk = Workers::new(pool.disk_set.len() * 2).map_err(Error::other)?; for (set_idx, set) in pool.disk_set.iter().enumerate() { wk.clone().take().await; diff --git a/ecstore/src/set_disk.rs b/ecstore/src/set_disk.rs index e2f07796d..ddf392a1b 100644 --- a/ecstore/src/set_disk.rs +++ b/ecstore/src/set_disk.rs @@ -30,7 +30,6 @@ use crate::{ }, heal_ops::BG_HEALING_UUID, }, - io::READ_BUFFER_SIZE, store_api::{ BucketInfo, BucketOptions, CompletePart, DeleteBucketOptions, DeletedObject, GetObjectReader, HTTPRangeSpec, ListMultipartsInfo, ListObjectsV2Info, MakeBucketOptions, MultipartInfo, MultipartUploadResult, ObjectIO, ObjectInfo, @@ -77,14 +76,13 @@ use std::time::SystemTime; use std::{ collections::{HashMap, HashSet}, io::{Cursor, Write}, - mem::replace, path::Path, sync::Arc, time::Duration, }; use time::OffsetDateTime; use tokio::{ - io::{AsyncWrite, empty}, + io::AsyncWrite, sync::{RwLock, broadcast}, }; use tokio::{ @@ -97,6 +95,8 @@ use tracing::{debug, info, warn}; use uuid::Uuid; use workers::workers::Workers; +pub const DEFAULT_READ_BUFFER_SIZE: usize = 1024 * 1024; + #[derive(Debug)] pub struct SetDisks { pub lockers: Vec, @@ -146,12 +146,10 @@ impl SetDisks { disks.shuffle(&mut rng); - let disks = disks + disks .into_iter() .filter(|v| v.as_ref().is_some_and(|d| d.is_local())) - .collect(); - - disks + .collect() } pub async fn get_online_disks_with_healing(&self, incl_healing: bool) -> (Vec, bool) { @@ -248,12 +246,10 @@ impl SetDisks { disks.shuffle(&mut rng); - let disks = disks + disks .into_iter() .filter(|v| v.as_ref().is_some_and(|d| d.is_local())) - .collect(); - - disks + .collect() } fn default_write_quorum(&self) -> usize { let mut data_count = self.set_drive_count - self.default_parity_count; @@ -446,7 +442,7 @@ impl SetDisks { let errs: Vec> = join_all(futures) .await .into_iter() - .map(|e| e.unwrap_or_else(|_| Some(DiskError::Unexpected))) + .map(|e| e.unwrap_or(Some(DiskError::Unexpected))) .collect(); if let Some(err) = reduce_write_quorum_errs(&errs, OBJECT_OP_IGNORED_ERRS, write_quorum) { @@ -1890,7 +1886,11 @@ impl SetDisks { till_offset, ) .await?; - let reader = BitrotReader::new(rd, erasure.shard_size(), HashAlgorithm::HighwayHash256); + let reader = BitrotReader::new( + Box::new(rustfs_rio::WarpReader::new(rd)), + erasure.shard_size(), + HashAlgorithm::HighwayHash256, + ); readers.push(Some(reader)); errors.push(None); } else { @@ -1928,9 +1928,10 @@ impl SetDisks { // "read part {} part_offset {},part_length {},part_size {} ", // part_number, part_offset, part_length, part_size // ); - let (written, mut err) = erasure.decode(writer, readers, part_offset, part_length, part_size).await; + let (written, err) = erasure.decode(writer, readers, part_offset, part_length, part_size).await; if let Some(e) = err { let de_err: DiskError = e.into(); + let mut has_err = true; if written == part_length { match de_err { DiskError::FileNotFound | DiskError::FileCorrupt => { @@ -1947,14 +1948,16 @@ impl SetDisks { ..Default::default() }) .await; - err = None; + has_err = false; } _ => {} } } - error!("erasure.decode err {} {:?}", written, &de_err); - return Err(de_err.into()); + if has_err { + error!("erasure.decode err {} {:?}", written, &de_err); + return Err(de_err.into()); + } } // debug!("ec decode {} writed size {}", part_number, n); @@ -2500,27 +2503,35 @@ impl SetDisks { if let Some(ref data) = metadata.data { let rd = Cursor::new(data.clone()); - let reader = BitrotReader::new( - Box::new(rd), - erasure.shard_size(), - HashAlgorithm::HighwayHash256, - ); + let reader = + BitrotReader::new(Box::new(rd), erasure.shard_size(), checksum_algo.clone()); readers.push(Some(reader)); // errors.push(None); } else { + let length = + till_offset.div_ceil(erasure.shard_size()) * checksum_algo.size() + till_offset; let rd = match disk - .read_file(bucket, &format!("{}/{}/part.{}", object, src_data_dir, part.number)) + .read_file_stream( + bucket, + &format!("{}/{}/part.{}", object, src_data_dir, part.number), + 0, + length, + ) .await { Ok(rd) => rd, Err(e) => { // errors.push(Some(e.into())); + error!("heal_object read_file err: {:?}", e); writers.push(None); continue; } }; - let reader = - BitrotReader::new(rd, erasure.shard_size(), HashAlgorithm::HighwayHash256); + let reader = BitrotReader::new( + Box::new(rustfs_rio::WarpReader::new(rd)), + erasure.shard_size(), + HashAlgorithm::HighwayHash256, + ); readers.push(Some(reader)); // errors.push(None); } @@ -3694,7 +3705,7 @@ impl SetDisks { let errs = join_all(futures).await.into_iter().map(|v| v.err()).collect::>(); - if let Some(err) = reduce_write_quorum_errs(&errs, OBJECT_OP_IGNORED_ERRS.as_ref(), write_quorum) { + if let Some(err) = reduce_write_quorum_errs(&errs, OBJECT_OP_IGNORED_ERRS, write_quorum) { return Err(err); } @@ -3749,7 +3760,7 @@ impl ObjectIO for SetDisks { // TODO: remote - let (rd, wd) = tokio::io::duplex(READ_BUFFER_SIZE); + let (rd, wd) = tokio::io::duplex(DEFAULT_READ_BUFFER_SIZE); let (reader, offset, length) = GetObjectReader::new(Box::new(rd), range, &object_info, opts, &h)?; @@ -5950,7 +5961,7 @@ mod tests { metadata.insert("etag".to_string(), "test-etag".to_string()); let file_info = FileInfo { - metadata: metadata, + metadata, ..Default::default() }; let parts_metadata = vec![file_info]; diff --git a/ecstore/src/store.rs b/ecstore/src/store.rs index 7e6edc497..1e1c97fbd 100644 --- a/ecstore/src/store.rs +++ b/ecstore/src/store.rs @@ -49,7 +49,6 @@ use glob::Pattern; use http::HeaderMap; use lazy_static::lazy_static; use madmin::heal_commands::HealResultItem; -use rand::Rng; use rustfs_filemeta::MetaCacheEntry; use s3s::dto::{BucketVersioningStatus, ObjectLockConfiguration, ObjectLockEnabled, VersioningConfiguration}; use std::cmp::Ordering; diff --git a/rustfs/src/admin/rpc.rs b/rustfs/src/admin/rpc.rs index 46959489b..19fe84d0e 100644 --- a/rustfs/src/admin/rpc.rs +++ b/rustfs/src/admin/rpc.rs @@ -3,7 +3,7 @@ use super::router::Operation; use super::router::S3Router; use crate::storage::ecfs::bytes_stream; use ecstore::disk::DiskAPI; -use ecstore::io::READ_BUFFER_SIZE; +use ecstore::set_disk::DEFAULT_READ_BUFFER_SIZE; use ecstore::store::find_local_disk; use futures::TryStreamExt; use http::StatusCode; @@ -72,7 +72,7 @@ impl Operation for ReadFile { Ok(S3Response::new(( StatusCode::OK, Body::from(StreamingBlob::wrap(bytes_stream( - ReaderStream::with_capacity(file, READ_BUFFER_SIZE), + ReaderStream::with_capacity(file, DEFAULT_READ_BUFFER_SIZE), query.length, ))), ))) diff --git a/rustfs/src/storage/ecfs.rs b/rustfs/src/storage/ecfs.rs index e0e26bfa4..c45de3009 100644 --- a/rustfs/src/storage/ecfs.rs +++ b/rustfs/src/storage/ecfs.rs @@ -30,8 +30,8 @@ use ecstore::bucket::tagging::decode_tags; use ecstore::bucket::tagging::encode_tags; use ecstore::bucket::versioning_sys::BucketVersioningSys; use ecstore::error::StorageError; -use ecstore::io::READ_BUFFER_SIZE; use ecstore::new_object_layer_fn; +use ecstore::set_disk::DEFAULT_READ_BUFFER_SIZE; use ecstore::store_api::BucketOptions; use ecstore::store_api::CompletePart; use ecstore::store_api::DeleteBucketOptions; @@ -575,7 +575,7 @@ impl S3 for FS { let last_modified = info.mod_time.map(Timestamp::from); let body = Some(StreamingBlob::wrap(bytes_stream( - ReaderStream::with_capacity(reader.stream, READ_BUFFER_SIZE), + ReaderStream::with_capacity(reader.stream, DEFAULT_READ_BUFFER_SIZE), info.size, ))); diff --git a/s3select/api/src/object_store.rs b/s3select/api/src/object_store.rs index e70a8b70d..d62c99bcf 100644 --- a/s3select/api/src/object_store.rs +++ b/s3select/api/src/object_store.rs @@ -2,17 +2,16 @@ use async_trait::async_trait; use bytes::Bytes; use chrono::Utc; use common::DEFAULT_DELIMITER; -use ecstore::io::READ_BUFFER_SIZE; +use ecstore::StorageAPI; use ecstore::new_object_layer_fn; +use ecstore::set_disk::DEFAULT_READ_BUFFER_SIZE; use ecstore::store::ECStore; use ecstore::store_api::ObjectIO; use ecstore::store_api::ObjectOptions; -use ecstore::StorageAPI; use futures::pin_mut; use futures::{Stream, StreamExt}; use futures_core::stream::BoxStream; use http::HeaderMap; -use object_store::path::Path; use object_store::Attributes; use object_store::GetOptions; use object_store::GetResult; @@ -24,16 +23,17 @@ use object_store::PutMultipartOpts; use object_store::PutOptions; use object_store::PutPayload; use object_store::PutResult; +use object_store::path::Path; use object_store::{Error as o_Error, Result}; use pin_project_lite::pin_project; +use s3s::S3Result; use s3s::dto::SelectObjectContentInput; use s3s::s3_error; -use s3s::S3Result; use std::ops::Range; use std::pin::Pin; use std::sync::Arc; -use std::task::ready; use std::task::Poll; +use std::task::ready; use tokio::io::AsyncRead; use tokio_util::io::ReaderStream; use tracing::info; @@ -117,14 +117,21 @@ impl ObjectStore for EcObjectStore { let payload = if self.need_convert { object_store::GetResultPayload::Stream( bytes_stream( - ReaderStream::with_capacity(ConvertStream::new(reader.stream, self.delimiter.clone()), READ_BUFFER_SIZE), + ReaderStream::with_capacity( + ConvertStream::new(reader.stream, self.delimiter.clone()), + DEFAULT_READ_BUFFER_SIZE, + ), reader.object_info.size, ) .boxed(), ) } else { object_store::GetResultPayload::Stream( - bytes_stream(ReaderStream::with_capacity(reader.stream, READ_BUFFER_SIZE), reader.object_info.size).boxed(), + bytes_stream( + ReaderStream::with_capacity(reader.stream, DEFAULT_READ_BUFFER_SIZE), + reader.object_info.size, + ) + .boxed(), ) }; Ok(GetResult { diff --git a/scripts/run.sh b/scripts/run.sh index fc1e65174..62b022e8b 100755 --- a/scripts/run.sh +++ b/scripts/run.sh @@ -20,7 +20,7 @@ mkdir -p ./target/volume/test{0..4} if [ -z "$RUST_LOG" ]; then export RUST_BACKTRACE=1 # export RUST_LOG="rustfs=debug,ecstore=debug,s3s=debug,iam=debug" - export RUST_LOG="rustfs=info,ecstore=info,s3s=debug" + export RUST_LOG="s3s=debug" fi # export RUSTFS_ERASURE_SET_DRIVE_COUNT=5