diff --git a/ecstore/src/bitrot.rs b/ecstore/src/bitrot.rs index 21ceb5b67..b69ef859d 100644 --- a/ecstore/src/bitrot.rs +++ b/ecstore/src/bitrot.rs @@ -1,9 +1,5 @@ use crate::{ - disk::{ - error::DiskError, - io::{FileReader, FileWriter}, - Disk, DiskAPI, - }, + disk::{error::DiskError, Disk, DiskAPI, FileReader, FileWriter}, erasure::{ReadAt, Writer}, error::{Error, Result}, store_api::BitrotAlgorithm, @@ -488,23 +484,45 @@ pub async fn bitrot_verify( // } pub struct BitrotFileWriter { - pub inner: FileWriter, + inner: Option, hasher: Hasher, _shard_size: usize, + inline: bool, + inline_data: Vec, } impl BitrotFileWriter { - pub fn new(inner: FileWriter, algo: BitrotAlgorithm, _shard_size: usize) -> Self { + pub async fn new( + disk: Arc, + volume: &str, + path: &str, + inline: bool, + algo: BitrotAlgorithm, + _shard_size: usize, + ) -> Result { + let inner = if !inline { + Some(disk.create_file("", volume, path, 0).await?) + } else { + None + }; + let hasher = algo.new_hasher(); - Self { + + Ok(Self { inner, + inline, + inline_data: Vec::new(), hasher, _shard_size, - } + }) } - pub fn writer(&self) -> &FileWriter { - &self.inner + // pub fn writer(&self) -> &FileWriter { + // &self.inner + // } + + pub fn inline_data(&self) -> &[u8] { + &self.inline_data } } @@ -521,18 +539,43 @@ impl Writer for BitrotFileWriter { self.hasher.reset(); self.hasher.update(buf); let hash_bytes = self.hasher.clone().finalize(); - let _ = self.inner.write_all(&hash_bytes).await?; - let _ = self.inner.write_all(buf).await?; + + if let Some(f) = self.inner.as_mut() { + f.write_all(&hash_bytes).await?; + f.write_all(buf).await?; + } else { + self.inline_data.extend_from_slice(&hash_bytes); + self.inline_data.extend_from_slice(buf); + } + + Ok(()) + } + async fn close(&mut self) -> Result<()> { + if self.inline { + return Ok(()); + } + + if let Some(f) = self.inner.as_mut() { + f.shutdown().await?; + } Ok(()) } } -pub fn new_bitrot_filewriter(inner: FileWriter, algo: BitrotAlgorithm, shard_size: usize) -> BitrotWriter { - Box::new(BitrotFileWriter::new(inner, algo, shard_size)) +pub async fn new_bitrot_filewriter( + disk: Arc, + volume: &str, + path: &str, + inline: bool, + algo: BitrotAlgorithm, + shard_size: usize, +) -> Result { + let w = BitrotFileWriter::new(disk, volume, path, inline, algo, shard_size).await?; + + Ok(Box::new(w)) } -#[derive(Debug)] struct BitrotFileReader { disk: Arc, data: Option>, @@ -599,7 +642,7 @@ impl ReadAt for BitrotFileReader { let stream_offset = (offset / self.shard_size) * self.hasher.size() + offset; if let Some(data) = self.data.clone() { - self.reader = Some(FileReader::Buffer(Cursor::new(data))); + self.reader = Some(Box::new(Cursor::new(data))); } else { self.reader = Some( self.disk diff --git a/ecstore/src/disk/io.rs b/ecstore/src/disk/io.rs index 981c6ddde..c0360050e 100644 --- a/ecstore/src/disk/io.rs +++ b/ecstore/src/disk/io.rs @@ -1,169 +1,27 @@ use crate::error::Result; use futures::TryStreamExt; -use std::io::Cursor; use std::pin::Pin; use std::task::Poll; -use tokio::fs::File; -use tokio::io::{AsyncRead, AsyncWrite}; +use tokio::io::AsyncWrite; +use tokio::sync::oneshot; use tokio_util::io::ReaderStream; use tokio_util::io::StreamReader; use tracing::error; use tracing::warn; -#[derive(Debug)] -pub enum FileReader { - Local(File), - // Remote(RemoteFileReader), - Buffer(Cursor>), - Http(HttpFileReader), -} - -impl AsyncRead for FileReader { - #[tracing::instrument(level = "debug", skip(self, buf))] - fn poll_read( - mut self: Pin<&mut Self>, - cx: &mut std::task::Context<'_>, - buf: &mut tokio::io::ReadBuf<'_>, - ) -> std::task::Poll> { - match &mut *self { - Self::Local(reader) => Pin::new(reader).poll_read(cx, buf), - Self::Buffer(reader) => Pin::new(reader).poll_read(cx, buf), - Self::Http(reader) => Pin::new(reader).poll_read(cx, buf), - } - } -} - -#[derive(Debug)] -pub struct HttpFileReader { - // client: reqwest::Client, - // url: String, - // disk: String, - // volume: String, - // path: String, - // offset: usize, - // length: usize, - inner: tokio::io::DuplexStream, - // buf: Vec, - // pos: usize, -} - -impl HttpFileReader { - pub fn new(url: &str, disk: &str, volume: &str, path: &str, offset: usize, length: usize) -> Result { - warn!("http read start {}", path); - let url = url.to_owned(); - let disk = disk.to_owned(); - let volume = volume.to_owned(); - let path = path.to_owned(); - - // let (reader, mut writer) = tokio::io::simplex(1024); - let (reader, mut writer) = tokio::io::duplex(1024 * 1024 * 10); - - tokio::spawn(async move { - let client = reqwest::Client::new(); - let resp = match client - .get(format!( - "{}/rustfs/rpc/read_file_stream?disk={}&volume={}&path={}&offset={}&length={}", - url, - urlencoding::encode(&disk), - urlencoding::encode(&volume), - urlencoding::encode(&path), - offset, - length - )) - .send() - .await - .map_err(|e| std::io::Error::new(std::io::ErrorKind::Other, e)) - { - Ok(resp) => resp, - Err(err) => { - warn!("http file reader error: {}", err); - return; - } - }; - - let mut rd = StreamReader::new( - resp.bytes_stream() - .map_err(|e| std::io::Error::new(std::io::ErrorKind::Other, e)), - ); - - if let Err(err) = tokio::io::copy(&mut rd, &mut writer).await { - error!("http file reader copy error: {}", err); - }; - }); - Ok(Self { - // client: reqwest::Client::new(), - // url: url.to_string(), - // disk: disk.to_string(), - // volume: volume.to_string(), - // path: path.to_string(), - // offset, - // length, - inner: reader, - // buf: Vec::new(), - // pos: 0, - }) - } -} - -impl AsyncRead for HttpFileReader { - #[tracing::instrument(level = "debug", skip(self, buf))] - fn poll_read( - mut self: Pin<&mut Self>, - cx: &mut std::task::Context<'_>, - buf: &mut tokio::io::ReadBuf<'_>, - ) -> std::task::Poll> { - Pin::new(&mut self.inner).poll_read(cx, buf) - } -} - -#[derive(Debug)] -pub enum FileWriter { - Local(File), - Http(HttpFileWriter), - Buffer(Cursor>), -} - -impl AsyncWrite for FileWriter { - #[tracing::instrument(level = "debug", skip(self, buf))] - fn poll_write( - mut self: Pin<&mut Self>, - cx: &mut std::task::Context<'_>, - buf: &[u8], - ) -> Poll> { - match &mut *self { - Self::Local(writer) => Pin::new(writer).poll_write(cx, buf), - Self::Buffer(writer) => Pin::new(writer).poll_write(cx, buf), - Self::Http(writer) => Pin::new(writer).poll_write(cx, buf), - } - } - - #[tracing::instrument(level = "debug", skip(self))] - fn poll_flush(mut self: Pin<&mut Self>, cx: &mut std::task::Context<'_>) -> Poll> { - match &mut *self { - Self::Local(writer) => Pin::new(writer).poll_flush(cx), - Self::Buffer(writer) => Pin::new(writer).poll_flush(cx), - Self::Http(writer) => Pin::new(writer).poll_flush(cx), - } - } - - #[tracing::instrument(level = "debug", skip(self))] - fn poll_shutdown(mut self: Pin<&mut Self>, cx: &mut std::task::Context<'_>) -> Poll> { - match &mut *self { - Self::Local(writer) => Pin::new(writer).poll_shutdown(cx), - Self::Buffer(writer) => Pin::new(writer).poll_shutdown(cx), - Self::Http(writer) => Pin::new(writer).poll_shutdown(cx), - } - } -} +use super::FileReader; #[derive(Debug)] pub struct HttpFileWriter { wd: tokio::io::WriteHalf, + err_rx: oneshot::Receiver, } impl HttpFileWriter { pub fn new(url: &str, disk: &str, volume: &str, path: &str, size: usize, append: bool) -> Result { - let (rd, wd) = tokio::io::simplex(1024 * 1024 * 10); + let (rd, wd) = tokio::io::simplex(4096); + + let (err_tx, err_rx) = oneshot::channel::(); let body = reqwest::Body::wrap_stream(ReaderStream::new(rd)); @@ -187,18 +45,22 @@ impl HttpFileWriter { .body(body) .send() .await + .map_err(|e| std::io::Error::new(std::io::ErrorKind::Other, e)) { error!("HttpFileWriter put file err: {:?}", err); + + if let Err(er) = err_tx.send(err) { + error!("HttpFileWriter tx.send err: {:?}", er); + } // return; } - // TODO: handle response - - // debug!("http write done {}", path); + // error!("http write done {}", path); }); Ok(Self { wd, + err_rx, // client: reqwest::Client::new(), // url: url.to_string(), // disk: disk.to_string(), @@ -214,6 +76,10 @@ impl AsyncWrite for HttpFileWriter { cx: &mut std::task::Context<'_>, buf: &[u8], ) -> Poll> { + if let Ok(err) = self.as_mut().err_rx.try_recv() { + return Poll::Ready(Err(err)); + } + Pin::new(&mut self.wd).poll_write(cx, buf) } @@ -227,3 +93,29 @@ impl AsyncWrite for HttpFileWriter { Pin::new(&mut self.wd).poll_shutdown(cx) } } + +pub async fn new_http_reader( + url: &str, + disk: &str, + volume: &str, + path: &str, + offset: usize, + length: usize, +) -> Result { + let resp = reqwest::Client::new() + .get(format!( + "{}/rustfs/rpc/read_file_stream?disk={}&volume={}&path={}&offset={}&length={}", + url, + urlencoding::encode(disk), + urlencoding::encode(volume), + urlencoding::encode(path), + offset, + length + )) + .send() + .await?; + + let inner = StreamReader::new(resp.bytes_stream().map_err(std::io::Error::other)); + + Ok(Box::new(inner)) +} diff --git a/ecstore/src/disk/local.rs b/ecstore/src/disk/local.rs index 013be96aa..4690450cd 100644 --- a/ecstore/src/disk/local.rs +++ b/ecstore/src/disk/local.rs @@ -745,7 +745,7 @@ impl LocalDisk { let meta = file.metadata().await?; - bitrot_verify(FileReader::Local(file), meta.size() as usize, part_size, algo, sum.to_vec(), shard_size).await + bitrot_verify(Box::new(file), meta.size() as usize, part_size, algo, sum.to_vec(), shard_size).await } async fn scan_dir( @@ -1314,7 +1314,7 @@ impl DiskAPI for LocalDisk { let src_file_path = src_volume_dir.join(Path::new(src_path)); let dst_file_path = dst_volume_dir.join(Path::new(dst_path)); - warn!("rename_part src_file_path:{:?}, dst_file_path:{:?}", &src_file_path, &dst_file_path); + // warn!("rename_part src_file_path:{:?}, dst_file_path:{:?}", &src_file_path, &dst_file_path); check_path_length(src_file_path.to_string_lossy().as_ref())?; check_path_length(dst_file_path.to_string_lossy().as_ref())?; @@ -1471,7 +1471,7 @@ impl DiskAPI for LocalDisk { #[tracing::instrument(level = "debug", skip(self))] async fn create_file(&self, origvolume: &str, volume: &str, path: &str, _file_size: usize) -> Result { - warn!("disk create_file: origvolume: {}, volume: {}, path: {}", origvolume, volume, path); + // warn!("disk create_file: origvolume: {}, volume: {}, path: {}", origvolume, volume, path); if !origvolume.is_empty() { let origvolume_dir = self.get_bucket_path(origvolume)?; @@ -1495,7 +1495,7 @@ impl DiskAPI for LocalDisk { .await .map_err(os_err_to_file_err)?; - Ok(FileWriter::Local(f)) + Ok(Box::new(f)) // Ok(()) } @@ -1517,7 +1517,7 @@ impl DiskAPI for LocalDisk { let f = self.open_file(file_path, O_CREATE | O_APPEND | O_WRONLY, volume_dir).await?; - Ok(FileWriter::Local(f)) + Ok(Box::new(f)) } // TODO: io verifier @@ -1552,7 +1552,7 @@ impl DiskAPI for LocalDisk { } })?; - Ok(FileReader::Local(f)) + Ok(Box::new(f)) } #[tracing::instrument(level = "debug", skip(self))] @@ -1603,7 +1603,7 @@ impl DiskAPI for LocalDisk { f.seek(SeekFrom::Start(offset as u64)).await?; - Ok(FileReader::Local(f)) + Ok(Box::new(f)) } #[tracing::instrument(level = "debug", skip(self))] async fn list_dir(&self, origvolume: &str, volume: &str, dir_path: &str, count: i32) -> Result> { @@ -2291,6 +2291,9 @@ impl DiskAPI for LocalDisk { self.scanning.fetch_add(1, Ordering::SeqCst); defer!(|| { self.scanning.fetch_sub(1, Ordering::SeqCst) }); + // must befor metadata_sys + let Some(store) = new_object_layer_fn() else { return Err(Error::msg("errServerNotInitialized")) }; + // Check if the current bucket has replication configuration if let Ok((rcfg, _)) = metadata_sys::get_replication_config(&cache.info.name).await { if has_active_rules(&rcfg, "", true) { @@ -2298,7 +2301,6 @@ impl DiskAPI for LocalDisk { } } - let Some(store) = new_object_layer_fn() else { return Err(Error::msg("errServerNotInitialized")) }; let loc = self.get_disk_location(); let disks = store.get_disks(loc.pool_idx.unwrap(), loc.disk_idx.unwrap()).await?; let disk = Arc::new(LocalDisk::new(&self.endpoint(), false).await?); diff --git a/ecstore/src/disk/mod.rs b/ecstore/src/disk/mod.rs index dcf719f29..a708977ee 100644 --- a/ecstore/src/disk/mod.rs +++ b/ecstore/src/disk/mod.rs @@ -29,14 +29,16 @@ use crate::{ }; use endpoint::Endpoint; use error::DiskError; -use io::{FileReader, FileWriter}; use local::LocalDisk; use madmin::info_commands::DiskMetrics; use remote::RemoteDisk; use serde::{Deserialize, Serialize}; use std::{cmp::Ordering, 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::info; use tracing::warn; use uuid::Uuid; @@ -372,6 +374,9 @@ pub async fn new_disk(ep: &endpoint::Endpoint, opt: &DiskOption) -> Result; +pub type FileWriter = Box; + #[async_trait::async_trait] pub trait DiskAPI: Debug + Send + Sync + 'static { fn to_string(&self) -> String; diff --git a/ecstore/src/disk/remote.rs b/ecstore/src/disk/remote.rs index 8528ac1ee..17c4c9926 100644 --- a/ecstore/src/disk/remote.rs +++ b/ecstore/src/disk/remote.rs @@ -22,9 +22,9 @@ use tracing::info; use uuid::Uuid; use super::{ - endpoint::Endpoint, io::HttpFileReader, CheckPartsResp, DeleteOptions, DiskAPI, DiskInfo, DiskInfoOptions, DiskLocation, - DiskOption, FileInfoVersions, FileReader, FileWriter, ReadMultipleReq, ReadMultipleResp, ReadOptions, RenameDataResp, - UpdateMetadataOpts, VolumeInfo, WalkDirOptions, + endpoint::Endpoint, CheckPartsResp, DeleteOptions, DiskAPI, DiskInfo, DiskInfoOptions, DiskLocation, DiskOption, + FileInfoVersions, FileReader, FileWriter, ReadMultipleReq, ReadMultipleResp, ReadOptions, RenameDataResp, UpdateMetadataOpts, + VolumeInfo, WalkDirOptions, }; use crate::{ disk::error::DiskError, @@ -36,7 +36,10 @@ use crate::{ }, store_api::{FileInfo, RawFileInfo}, }; -use crate::{disk::io::HttpFileWriter, utils::proto_err_to_err}; +use crate::{ + disk::io::{new_http_reader, HttpFileWriter}, + utils::proto_err_to_err, +}; use crate::{disk::MetaCacheEntry, metacache::writer::MetacacheWriter}; use protos::proto_gen::node_service::RenamePartRequst; @@ -316,7 +319,7 @@ impl DiskAPI for RemoteDisk { #[tracing::instrument(level = "debug", skip(self))] async fn create_file(&self, _origvolume: &str, volume: &str, path: &str, file_size: usize) -> Result { info!("create_file"); - Ok(FileWriter::Http(HttpFileWriter::new( + Ok(Box::new(HttpFileWriter::new( self.endpoint.grid_host().as_str(), self.endpoint.to_string().as_str(), volume, @@ -329,7 +332,7 @@ impl DiskAPI for RemoteDisk { #[tracing::instrument(level = "debug", skip(self))] async fn append_file(&self, volume: &str, path: &str) -> Result { info!("append_file"); - Ok(FileWriter::Http(HttpFileWriter::new( + Ok(Box::new(HttpFileWriter::new( self.endpoint.grid_host().as_str(), self.endpoint.to_string().as_str(), volume, @@ -342,26 +345,20 @@ impl DiskAPI for RemoteDisk { #[tracing::instrument(level = "debug", skip(self))] async fn read_file(&self, volume: &str, path: &str) -> Result { info!("read_file"); - Ok(FileReader::Http(HttpFileReader::new( - self.endpoint.grid_host().as_str(), - self.endpoint.to_string().as_str(), - volume, - path, - 0, - 0, - )?)) + Ok(new_http_reader(self.endpoint.grid_host().as_str(), self.endpoint.to_string().as_str(), volume, path, 0, 0).await?) } #[tracing::instrument(level = "debug", skip(self))] async fn read_file_stream(&self, volume: &str, path: &str, offset: usize, length: usize) -> Result { - Ok(FileReader::Http(HttpFileReader::new( + Ok(new_http_reader( self.endpoint.grid_host().as_str(), self.endpoint.to_string().as_str(), volume, path, offset, length, - )?)) + ) + .await?) } async fn list_dir(&self, _origvolume: &str, volume: &str, _dir_path: &str, _count: i32) -> Result> { diff --git a/ecstore/src/erasure.rs b/ecstore/src/erasure.rs index f068b1098..0056d4bf6 100644 --- a/ecstore/src/erasure.rs +++ b/ecstore/src/erasure.rs @@ -6,7 +6,6 @@ use futures::future::join_all; use futures::{pin_mut, Stream, StreamExt}; use reed_solomon_erasure::galois_8::ReedSolomon; use std::any::Any; -use std::fmt::Debug; use std::io::ErrorKind; use tokio::io::DuplexStream; use tokio::io::{AsyncReadExt, AsyncWriteExt}; @@ -500,11 +499,10 @@ pub trait Writer { } #[async_trait::async_trait] -pub trait ReadAt: Debug { +pub trait ReadAt { async fn read_at(&mut self, offset: usize, length: usize) -> Result<(Vec, usize)>; } -#[derive(Debug)] pub struct ShardReader { readers: Vec>, // 磁盘 data_block_count: usize, // 总的分片数量 diff --git a/ecstore/src/set_disk.rs b/ecstore/src/set_disk.rs index d1ef1de4e..e91907bc2 100644 --- a/ecstore/src/set_disk.rs +++ b/ecstore/src/set_disk.rs @@ -14,7 +14,6 @@ use crate::{ endpoint::Endpoint, error::{is_all_not_found, DiskError}, format::FormatV3, - io::{FileReader, FileWriter}, new_disk, CheckPartsResp, DeleteOptions, DiskAPI, DiskInfo, DiskInfoOptions, DiskOption, DiskStore, FileInfoVersions, MetaCacheEntries, MetaCacheEntry, MetadataResolutionParams, ReadMultipleReq, ReadMultipleResp, ReadOptions, UpdateMetadataOpts, RUSTFS_META_BUCKET, RUSTFS_META_MULTIPART_BUCKET, RUSTFS_META_TMP_BUCKET, @@ -636,7 +635,7 @@ impl SetDisks { } fn get_upload_id_dir(bucket: &str, object: &str, upload_id: &str) -> String { - warn!("get_upload_id_dir upload_id {:?}", upload_id); + // warn!("get_upload_id_dir upload_id {:?}", upload_id); let upload_uuid = base64_decode(upload_id.as_bytes()) .and_then(|v| { @@ -2450,21 +2449,25 @@ impl SetDisks { for disk in out_dated_disks.iter() { if let Some(disk) = disk { - let filewriter = { - if is_inline_buffer { - FileWriter::Buffer(Cursor::new(Vec::new())) - } else { - let disk = disk.clone(); - let part_path = format!("{}/{}/part.{}", tmp_id, dst_data_dir, part.number); - disk.create_file("", RUSTFS_META_TMP_BUCKET, &part_path, 0).await? - } - }; + // let filewriter = { + // if is_inline_buffer { + // Box::new(Cursor::new(Vec::new())) + // } else { + // let disk = disk.clone(); + // let part_path = format!("{}/{}/part.{}", tmp_id, dst_data_dir, part.number); + // disk.create_file("", RUSTFS_META_TMP_BUCKET, &part_path, 0).await? + // } + // }; let writer = new_bitrot_filewriter( - filewriter, + disk.clone(), + RUSTFS_META_TMP_BUCKET, + format!("{}/{}/part.{}", tmp_id, dst_data_dir, part.number).as_str(), + is_inline_buffer, DEFAULT_BITROT_ALGO, erasure.shard_size(erasure.block_size), - ); + ) + .await?; writers.push(Some(writer)); } else { @@ -2500,9 +2503,7 @@ impl SetDisks { if is_inline_buffer { if let Some(ref writer) = writers[index] { if let Some(w) = writer.as_any().downcast_ref::() { - if let FileWriter::Buffer(buffer_writer) = w.writer() { - parts_metadata[index].data = Some(buffer_writer.clone().into_inner()); - } + parts_metadata[index].data = Some(w.inline_data().to_vec()); } } parts_metadata[index].set_inline_data(); @@ -3742,17 +3743,25 @@ impl ObjectIO for SetDisks { for disk_op in shuffle_disks.iter() { if let Some(disk) = disk_op { - let filewriter = { - if is_inline_buffer { - FileWriter::Buffer(Cursor::new(Vec::new())) - } else { - let disk = disk.clone(); + // let filewriter = { + // if is_inline_buffer { + // Box::new(Cursor::new(Vec::new())) + // } else { + // let disk = disk.clone(); - disk.create_file("", RUSTFS_META_TMP_BUCKET, &tmp_object, 0).await? - } - }; + // disk.create_file("", RUSTFS_META_TMP_BUCKET, &tmp_object, 0).await? + // } + // }; - let writer = new_bitrot_filewriter(filewriter, DEFAULT_BITROT_ALGO, erasure.shard_size(erasure.block_size)); + let writer = new_bitrot_filewriter( + disk.clone(), + RUSTFS_META_TMP_BUCKET, + &tmp_object, + is_inline_buffer, + DEFAULT_BITROT_ALGO, + erasure.shard_size(erasure.block_size), + ) + .await?; writers.push(Some(writer)); } else { @@ -3760,8 +3769,6 @@ impl ObjectIO for SetDisks { } } - warn!("put_object data.content_length {}", data.content_length); - // TODO: etag from header let mut etag_stream = EtagReader::new(&mut data.stream, None, None); @@ -3769,6 +3776,10 @@ impl ObjectIO for SetDisks { .encode(&mut etag_stream, &mut writers, data.content_length, write_quorum) .await?; // TODO: 出错,删除临时目录 + if let Err(err) = close_bitrot_writers(&mut writers).await { + error!("close_bitrot_writers err {:?}", err); + } + let etag = etag_stream.etag(); //TODO: userDefined @@ -3790,9 +3801,7 @@ impl ObjectIO for SetDisks { if is_inline_buffer { if let Some(ref writer) = writers[i] { if let Some(w) = writer.as_any().downcast_ref::() { - if let FileWriter::Buffer(buffer_writer) = w.writer() { - fi.data = Some(buffer_writer.clone().into_inner()); - } + fi.data = Some(w.inline_data().to_vec()); } } } @@ -4089,7 +4098,7 @@ impl StorageAPI for SetDisks { for errs in results.into_iter().flatten() { // TODO: handle err reduceWriteQuorumErrs - for err in errs.iter() { + for err in errs.iter().flatten() { warn!("result err {:?}", err); } } @@ -4327,10 +4336,18 @@ impl StorageAPI for SetDisks { for disk in disks.iter() { if let Some(disk) = disk { // let writer = disk.append_file(RUSTFS_META_TMP_BUCKET, &tmp_part_path).await?; - let filewriter = disk - .create_file("", RUSTFS_META_TMP_BUCKET, &tmp_part_path, data.content_length) - .await?; - let writer = new_bitrot_filewriter(filewriter, DEFAULT_BITROT_ALGO, erasure.shard_size(erasure.block_size)); + // let filewriter = disk + // .create_file("", RUSTFS_META_TMP_BUCKET, &tmp_part_path, data.content_length) + // .await?; + let writer = new_bitrot_filewriter( + disk.clone(), + RUSTFS_META_TMP_BUCKET, + &tmp_part_path, + false, + DEFAULT_BITROT_ALGO, + erasure.shard_size(erasure.block_size), + ) + .await?; writers.push(Some(writer)); } else { writers.push(None); @@ -4345,6 +4362,10 @@ impl StorageAPI for SetDisks { .encode(&mut etag_stream, &mut writers, data.content_length, write_quorum) .await?; + if let Err(err) = close_bitrot_writers(&mut writers).await { + error!("close_bitrot_writers err {:?}", err); + } + let mut etag = etag_stream.etag(); if let Some(ref tag) = opts.preserve_etag { @@ -5248,7 +5269,7 @@ async fn disks_with_all_parts( let checksum_info = meta.erasure.get_checksum_info(meta.parts[0].number); let data_len = data.len(); let verify_err = match bitrot_verify( - FileReader::Buffer(Cursor::new(data.clone())), + Box::new(Cursor::new(data.clone())), data_len, meta.erasure.shard_file_size(meta.size), checksum_info.algorithm, diff --git a/ecstore/src/store_api.rs b/ecstore/src/store_api.rs index c67d17813..091ee1e89 100644 --- a/ecstore/src/store_api.rs +++ b/ecstore/src/store_api.rs @@ -19,7 +19,7 @@ use time::OffsetDateTime; use uuid::Uuid; pub const ERASURE_ALGORITHM: &str = "rs-vandermonde"; -pub const BLOCK_SIZE_V2: usize = 1048576; // 1M +pub const BLOCK_SIZE_V2: usize = 1024 * 1024; // 1M pub const RESERVED_METADATA_PREFIX: &str = "X-Rustfs-Internal-"; pub const RESERVED_METADATA_PREFIX_LOWER: &str = "X-Rustfs-Internal-"; pub const RUSTFS_HEALING: &str = "X-Rustfs-Internal-healing"; diff --git a/rustfs/src/admin/rpc.rs b/rustfs/src/admin/rpc.rs index 9d2fda5b5..42bbb0ace 100644 --- a/rustfs/src/admin/rpc.rs +++ b/rustfs/src/admin/rpc.rs @@ -3,7 +3,6 @@ use super::router::Operation; use super::router::S3Router; use crate::storage::ecfs::bytes_stream; use common::error::Result; -use ecstore::disk::io::FileReader; use ecstore::disk::DiskAPI; use ecstore::store::find_local_disk; use futures::TryStreamExt; @@ -19,7 +18,6 @@ use s3s::S3Result; use serde_urlencoded::from_bytes; use tokio_util::io::ReaderStream; use tokio_util::io::StreamReader; -use tracing::warn; pub const RPC_PREFIX: &str = "/rustfs/rpc"; @@ -52,8 +50,6 @@ pub struct ReadFile {} #[async_trait::async_trait] impl Operation for ReadFile { async fn call(&self, req: S3Request, _params: Params<'_, '_>) -> S3Result> { - warn!("handle ReadFile"); - let query = { if let Some(query) = req.uri.query() { let input: ReadFileQuery = @@ -68,39 +64,15 @@ impl Operation for ReadFile { return Err(s3_error!(InvalidArgument, "disk not found")); }; - let file: FileReader = disk + let file = disk .read_file_stream(&query.volume, &query.path, query.offset, query.length) .await .map_err(|e| s3_error!(InternalError, "read file err {}", e))?; - let s = bytes_stream(ReaderStream::new(file), query.length); - - Ok(S3Response::new((StatusCode::OK, Body::from(StreamingBlob::wrap(s))))) - - // let querys = req.uri.query().map(|q| { - // let mut querys = HashMap::new(); - // for (k, v) in url::form_urlencoded::parse(q.as_bytes()) { - // println!("{}={}", k, v); - // querys.insert(k.to_string(), v.to_string()); - // } - // querys - // }); - - // // TODO: file_path from root - - // if let Some(file_path) = querys.and_then(|q| q.get("file_path").cloned()) { - // let file = fs::OpenOptions::new() - // .read(true) - // .open(file_path) - // .await - // .map_err(|e| S3Error::with_message(S3ErrorCode::InternalError, format!("open file err {}", e)))?; - - // let s = bytes_stream(ReaderStream::new(file), 0); - - // return Ok(S3Response::new((StatusCode::OK, Body::from(StreamingBlob::wrap(s))))); - // } - - // Ok(S3Response::new((StatusCode::BAD_REQUEST, Body::empty()))) + Ok(S3Response::new(( + StatusCode::OK, + Body::from(StreamingBlob::wrap(bytes_stream(ReaderStream::new(file), query.length))), + ))) } } @@ -117,8 +89,6 @@ pub struct PutFile {} #[async_trait::async_trait] impl Operation for PutFile { async fn call(&self, req: S3Request, _params: Params<'_, '_>) -> S3Result> { - warn!("handle PutFile"); - let query = { if let Some(query) = req.uri.query() { let input: PutFileQuery =