diff --git a/ecstore/src/chunk_stream.rs b/ecstore/src/chunk_stream.rs index f8ec02aeb..43a4dbce5 100644 --- a/ecstore/src/chunk_stream.rs +++ b/ecstore/src/chunk_stream.rs @@ -5,7 +5,6 @@ use s3s::StdError; use std::future::Future; use std::pin::Pin; use std::task::{Context, Poll}; -use tracing::debug; use transform_stream::AsyncTryStream; pub type SyncBoxFuture<'a, T> = Pin + Send + Sync + 'a>>; @@ -37,13 +36,13 @@ impl ChunkedStream { None => break, Some(Err(e)) => return Err(e), Some(Ok((data, remaining_bytes))) => { - debug!( - "content_length:{},readed_size:{}, read_data data:{}, remaining_bytes: {} ", - content_length, - readed_size, - data.len(), - remaining_bytes.len() - ); + // debug!( + // "content_length:{},readed_size:{}, read_data data:{}, remaining_bytes: {} ", + // content_length, + // readed_size, + // data.len(), + // remaining_bytes.len() + // ); prev_bytes = remaining_bytes; data @@ -53,17 +52,17 @@ impl ChunkedStream { for bytes in data { readed_size += bytes.len(); - println!("readed_size {}, content_length {}", readed_size, content_length,); + // debug!("readed_size {}, content_length {}", readed_size, content_length,); y.yield_ok(bytes).await; } if readed_size + prev_bytes.len() >= content_length { - println!( - "读完了 readed_size:{} + prev_bytes.len({}) == content_length {}", - readed_size, - prev_bytes.len(), - content_length, - ); + // debug!( + // "读完了 readed_size:{} + prev_bytes.len({}) == content_length {}", + // readed_size, + // prev_bytes.len(), + // content_length, + // ); // 填充0？ if !need_padding { @@ -81,7 +80,7 @@ impl ChunkedStream { } } - debug!("chunked stream exit"); + // debug!("chunked stream exit"); Ok(()) }) @@ -104,7 +103,7 @@ impl ChunkedStream { // 只执行一次 let mut push_data_bytes = |mut bytes: Bytes| { - debug!("read from body {} split per {}, prev_bytes: {}", bytes.len(), data_size, prev_bytes.len()); + // debug!("read from body {} split per {}, prev_bytes: {}", bytes.len(), data_size, prev_bytes.len()); if bytes.is_empty() { return None; @@ -117,24 +116,24 @@ impl ChunkedStream { // 合并上一次数据 if !prev_bytes.is_empty() { let need_size = data_size.wrapping_sub(prev_bytes.len()); - println!( - " 上一次有剩余{},从这一次中取{},共：{}", - prev_bytes.len(), - need_size, - prev_bytes.len() + need_size - ); + // debug!( + // " 上一次有剩余{},从这一次中取{},共：{}", + // prev_bytes.len(), + // need_size, + // prev_bytes.len() + need_size + // ); if bytes.len() >= need_size { let data = bytes.split_to(need_size); let mut combined = Vec::new(); combined.extend_from_slice(&prev_bytes); combined.extend_from_slice(&data); - debug!( - "取到的长度大于所需，取出需要的长度：{},与上一次合并得到：{}，bytes剩余：{}", - need_size, - combined.len(), - bytes.len(), - ); + // debug!( + // "取到的长度大于所需，取出需要的长度：{},与上一次合并得到：{}，bytes剩余：{}", + // need_size, + // combined.len(), + // bytes.len(), + // ); bytes_buffer.push(Bytes::from(combined)); } else { @@ -142,12 +141,12 @@ impl ChunkedStream { combined.extend_from_slice(&prev_bytes); combined.extend_from_slice(&bytes); - debug!( - "取到的长度小于所需，取出需要的长度：{},与上一次合并得到：{}，bytes剩余：{}，直接返回", - need_size, - combined.len(), - bytes.len(), - ); + // debug!( + // "取到的长度小于所需，取出需要的长度：{},与上一次合并得到：{}，bytes剩余：{}，直接返回", + // need_size, + // combined.len(), + // bytes.len(), + // ); return Some(Bytes::from(combined)); } diff --git a/ecstore/src/disk.rs b/ecstore/src/disk.rs index f5bbfde87..2b2966dd1 100644 --- a/ecstore/src/disk.rs +++ b/ecstore/src/disk.rs @@ -9,16 +9,13 @@ use bytes::Bytes; use futures::future::join_all; use path_absolutize::Absolutize; use time::OffsetDateTime; -use tokio::io::{self, AsyncWriteExt, BufWriter, ErrorKind}; -use tokio::{ - fs::{self, File}, - io::DuplexStream, -}; +use tokio::fs::{self, File}; +use tokio::io::ErrorKind; use tracing::debug; use uuid::Uuid; use crate::{ - disk_api::{DiskAPI, DiskError, ReadOptions, VolumeInfo}, + disk_api::{DiskAPI, DiskError, FileWriter, ReadOptions, VolumeInfo}, endpoint::{Endpoint, Endpoints}, file_meta::FileMeta, format::FormatV3, @@ -407,14 +404,7 @@ impl DiskAPI for LocalDisk { Ok(()) } - async fn create_file( - &self, - _origvolume: &str, - volume: &str, - path: &str, - _file_size: usize, - mut r: DuplexStream, - ) -> Result<()> { + async fn create_file(&self, _origvolume: &str, volume: &str, path: &str, _file_size: usize) -> Result { let fpath = self.get_object_path(volume, path)?; debug!("CreateFile fpath: {:?}", fpath); @@ -425,39 +415,44 @@ impl DiskAPI for LocalDisk { let file = File::create(&fpath).await?; - let mut writer = BufWriter::new(file); + Ok(FileWriter::new(file)) - io::copy(&mut r, &mut writer).await?; + // let mut writer = BufWriter::new(file); - Ok(()) + // io::copy(&mut r, &mut writer).await?; + + // Ok(()) } - async fn append_file(&self, volume: &str, path: &str, mut r: DuplexStream) -> Result<()> { + // async fn append_file(&self, volume: &str, path: &str, mut r: DuplexStream) -> Result { + async fn append_file(&self, volume: &str, path: &str) -> Result { let p = self.get_object_path(&volume, &path)?; debug!("append_file start {} {:?}", self.id(), &p); - // if let Some(dir_path) = p.parent() { - // fs::create_dir_all(&dir_path).await?; - // } + if let Some(dir_path) = p.parent() { + fs::create_dir_all(&dir_path).await?; + } // debug!("append_file open {} {:?}", self.id(), &p); - let mut file = File::options() + let file = File::options() .read(true) .create(true) .write(true) - // .append(true) + .append(true) .open(&p) .await?; - let mut writer = BufWriter::new(file); + Ok(FileWriter::new(file)) - io::copy(&mut r, &mut writer).await?; + // let mut writer = BufWriter::new(file); - debug!("append_file end {} {}", self.id(), path); + // io::copy(&mut r, &mut writer).await?; + + // debug!("append_file end {} {}", self.id(), path); // io::copy(&mut r, &mut file).await?; - Ok(()) + // Ok(()) } async fn rename_data(&self, src_volume: &str, src_path: &str, fi: &FileInfo, dst_volume: &str, dst_path: &str) -> Result<()> { diff --git a/ecstore/src/disk_api.rs b/ecstore/src/disk_api.rs index 56e1f4b4e..b3c0be12b 100644 --- a/ecstore/src/disk_api.rs +++ b/ecstore/src/disk_api.rs @@ -1,9 +1,9 @@ -use std::fmt::Debug; +use std::{fmt::Debug, pin::Pin}; use anyhow::{Error, Result}; use bytes::Bytes; use time::OffsetDateTime; -use tokio::io::DuplexStream; +use tokio::io::AsyncWrite; use uuid::Uuid; use crate::store_api::{FileInfo, RawFileInfo}; @@ -16,8 +16,8 @@ pub trait DiskAPI: Debug + Send + Sync + 'static { async fn read_all(&self, volume: &str, path: &str) -> Result; async fn write_all(&self, volume: &str, path: &str, data: Vec) -> Result<()>; async fn rename_file(&self, src_volume: &str, src_path: &str, dst_volume: &str, dst_path: &str) -> Result<()>; - async fn create_file(&self, origvolume: &str, volume: &str, path: &str, file_size: usize, r: DuplexStream) -> Result<()>; - async fn append_file(&self, volume: &str, path: &str, r: DuplexStream) -> Result<()>; + async fn create_file(&self, origvolume: &str, volume: &str, path: &str, file_size: usize) -> Result; + async fn append_file(&self, volume: &str, path: &str) -> Result; async fn rename_data( &self, src_volume: &str, @@ -53,6 +53,43 @@ pub struct ReadOptions { pub healing: bool, } +pub struct FileWriter { + pub inner: Pin>, +} + +impl AsyncWrite for FileWriter { + fn poll_write( + mut self: Pin<&mut Self>, + cx: &mut std::task::Context<'_>, + buf: &[u8], + ) -> std::task::Poll> { + Pin::new(&mut self.inner).poll_write(cx, buf) + } + + fn poll_flush( + mut self: Pin<&mut Self>, + cx: &mut std::task::Context<'_>, + ) -> std::task::Poll> { + Pin::new(&mut self.inner).poll_flush(cx) + } + + fn poll_shutdown( + mut self: Pin<&mut Self>, + cx: &mut std::task::Context<'_>, + ) -> std::task::Poll> { + Pin::new(&mut self.inner).poll_shutdown(cx) + } +} + +impl FileWriter { + pub fn new(inner: W) -> Self + where + W: AsyncWrite + Send + Sync + 'static, + { + Self { inner: Box::pin(inner) } + } +} + #[derive(Debug, thiserror::Error)] pub enum DiskError { #[error("file not found")] diff --git a/ecstore/src/erasure.rs b/ecstore/src/erasure.rs index f371cc2a5..1d01c6205 100644 --- a/ecstore/src/erasure.rs +++ b/ecstore/src/erasure.rs @@ -54,35 +54,35 @@ impl Erasure { for (i, w) in writers.iter_mut().enumerate() { total += blocks[i].len(); - debug!( - "{} {}-{} encode write {} , total:{}, readed:{}", - self.id, - idx, - i, - blocks[i].len(), - data_size, - total - ); + // debug!( + // "{} {}-{} encode write {} , total:{}, readed:{}", + // self.id, + // idx, + // i, + // blocks[i].len(), + // data_size, + // total + // ); - match w.write(blocks[i].as_ref()).await { + match w.write_all(blocks[i].as_ref()).await { Ok(_) => errs.push(None), Err(e) => errs.push(Some(e)), } } - debug!("{} encode_data write errs:{:?}", self.id, errs); - // TODO: reduceWriteQuorumErrs - for err in errs.iter() { - if err.is_some() { - return Err(Error::msg("message")); - } - } + // debug!("{} encode_data write errs:{:?}", self.id, errs); + // // TODO: reduceWriteQuorumErrs + // for err in errs.iter() { + // if err.is_some() { + // return Err(Error::msg("message")); + // } + // } } Err(e) => return Err(anyhow!(e)), } } - debug!("{} encode_data done {}", self.id, total); + // debug!("{} encode_data done {}", self.id, total); Ok(total) diff --git a/ecstore/src/lib.rs b/ecstore/src/lib.rs index caef7b39f..67cd5f451 100644 --- a/ecstore/src/lib.rs +++ b/ecstore/src/lib.rs @@ -16,4 +16,3 @@ pub mod store; pub mod store_api; mod store_init; mod utils; -mod writer; diff --git a/ecstore/src/sets.rs b/ecstore/src/sets.rs index b72b76ac1..8a64dab06 100644 --- a/ecstore/src/sets.rs +++ b/ecstore/src/sets.rs @@ -1,23 +1,15 @@ -use anyhow::{Error, Result}; -use futures::future::join_all; -use time::OffsetDateTime; -use tracing::debug; +use anyhow::Result; use uuid::Uuid; use crate::{ - disk::{DiskStore, RUSTFS_META_MULTIPART_BUCKET, RUSTFS_META_TMP_BUCKET}, + disk::DiskStore, endpoint::PoolEndpoints, - erasure::Erasure, format::{DistributionAlgoVersion, FormatV3}, set_disk::SetDisks, store_api::{ - BucketInfo, BucketOptions, FileInfo, MakeBucketOptions, MultipartUploadResult, ObjectOptions, PartInfo, PutObjReader, - StorageAPI, - }, - utils::{ - crypto::{base64_decode, base64_encode, hex, sha256}, - hash, + BucketInfo, BucketOptions, MakeBucketOptions, MultipartUploadResult, ObjectOptions, PartInfo, PutObjReader, StorageAPI, }, + utils::hash, }; #[derive(Debug)] diff --git a/ecstore/src/writer.rs b/ecstore/src/writer.rs deleted file mode 100644 index 48257dc05..000000000 --- a/ecstore/src/writer.rs +++ /dev/null @@ -1,90 +0,0 @@ -use std::{io, task::Poll}; - -use futures::{ready, Future}; -use tokio::io::{AsyncWrite, BufWriter}; -use tracing::debug; -use uuid::Uuid; - -use crate::disk::DiskStore; - -pub struct AppendWriter<'a> { - disk: DiskStore, - volume: &'a str, - path: &'a str, -} - -impl<'a> AppendWriter<'a> { - pub fn new(disk: DiskStore, volume: &'a str, path: &'a str) -> Self { - debug!("AppendWriter new {}: {}/{}", disk.id(), volume, path); - Self { disk, volume, path } - } - - async fn async_write(&self, buf: &[u8]) -> Result<(), std::io::Error> { - debug!("async_write {}: {}: {}", self.disk.id(), &self.path, buf.len()); - - // self.disk - // .append_file(&self.volume, &self.path, buf) - // .await - // .map_err(|e| std::io::Error::new(std::io::ErrorKind::Other, e))?; - Ok(()) - } -} - -impl<'a> AsyncWrite for AppendWriter<'a> { - fn poll_write( - self: std::pin::Pin<&mut Self>, - cx: &mut std::task::Context<'_>, - buf: &[u8], - ) -> std::task::Poll> { - let mut fut = Box::pin(self.async_write(buf)); - debug!("AsyncWrite poll_write {}, buf:{}", self.disk.id(), buf.len()); - - // while let Poll::Ready(e) = fut.as_mut().poll(cx) { - // let a = match e { - // Ok(_) => { - // debug!("Ready ok {}", self.disk.id()); - // Poll::Ready(Ok(buf.len())) - // } - // Err(e) => { - // debug!("Ready err {}", self.disk.id()); - // Poll::Ready(Err(e)) - // } - // }; - - // return a; - // } - - // Poll::Pending - - match fut.as_mut().poll(cx) { - Poll::Pending => { - debug!("Pending {}", self.disk.id()); - Poll::Pending - } - Poll::Ready(e) => match e { - Ok(_) => { - debug!("Ready ok {}", self.disk.id()); - Poll::Ready(Ok(buf.len())) - } - Err(e) => { - debug!("Ready err {}", self.disk.id()); - Poll::Ready(Err(e)) - } - }, - } - } - - fn poll_flush( - self: std::pin::Pin<&mut Self>, - _cx: &mut std::task::Context<'_>, - ) -> std::task::Poll> { - Poll::Ready(Ok(())) - } - - fn poll_shutdown( - self: std::pin::Pin<&mut Self>, - _cx: &mut std::task::Context<'_>, - ) -> std::task::Poll> { - Poll::Ready(Ok(())) - } -}