diff --git a/ecstore/src/chunk_stream.rs b/ecstore/src/chunk_stream.rs index 95bd3ba1a..5fb46625b 100644 --- a/ecstore/src/chunk_stream.rs +++ b/ecstore/src/chunk_stream.rs @@ -5,6 +5,7 @@ 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>>; @@ -36,6 +37,14 @@ 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() + ); + prev_bytes = remaining_bytes; data } @@ -75,6 +84,8 @@ impl ChunkedStream { } } + debug!("chunked stream exit"); + Ok(()) }) }); @@ -96,6 +107,8 @@ 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()); + if bytes.is_empty() { return None; } @@ -119,12 +132,12 @@ impl ChunkedStream { combined.extend_from_slice(&prev_bytes); combined.extend_from_slice(&data); - // println!( - // "取到的长度大于所需，取出需要的长度：{},与上一次合并得到：{}，bytes剩余：{}", - // need_size, - // combined.len(), - // bytes.len(), - // ); + debug!( + "取到的长度大于所需，取出需要的长度：{},与上一次合并得到：{}，bytes剩余：{}", + need_size, + combined.len(), + bytes.len(), + ); bytes_buffer.push(Bytes::from(combined)); } else { @@ -132,12 +145,12 @@ impl ChunkedStream { combined.extend_from_slice(&prev_bytes); combined.extend_from_slice(&bytes); - // println!( - // "取到的长度小于所需，取出需要的长度：{},与上一次合并得到：{}，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 316f726da..147630c03 100644 --- a/ecstore/src/disk.rs +++ b/ecstore/src/disk.rs @@ -9,7 +9,7 @@ use bytes::Bytes; use futures::future::join_all; use path_absolutize::Absolutize; use time::OffsetDateTime; -use tokio::io::{self, BufWriter, ErrorKind}; +use tokio::io::{self, AsyncWriteExt, BufWriter, ErrorKind}; use tokio::{ fs::{self, File}, io::DuplexStream, @@ -354,7 +354,7 @@ impl DiskAPI for LocalDisk { Ok(Bytes::from(data)) } - async fn write_all(&self, volume: &str, path: &str, data: Bytes) -> Result<()> { + async fn write_all(&self, volume: &str, path: &str, data: Vec) -> Result<()> { let p = self.get_object_path(&volume, &path)?; write_all_internal(p, data).await?; @@ -427,6 +427,22 @@ impl DiskAPI for LocalDisk { Ok(()) } + async fn append_file(&self, volume: &str, path: &str, mut r: DuplexStream) -> Result<()> { + let p = self.get_object_path(&volume, &path)?; + + let mut file = File::options() + .create(true) + .write(true) + .append(true) // 设置为追加模式 + .open(p) + .await?; + + let mut writer = BufWriter::new(file); + + io::copy(&mut r, &mut writer).await?; + + Ok(()) + } async fn rename_data(&self, src_volume: &str, src_path: &str, fi: &FileInfo, dst_volume: &str, dst_path: &str) -> Result<()> { let src_volume_path = self.get_bucket_path(&src_volume)?; @@ -574,7 +590,7 @@ impl DiskAPI for LocalDisk { volume: &str, path: &str, version_id: Uuid, - opts: ReadOptions, + opts: &ReadOptions, ) -> Result { let file_path = self.get_object_path(volume, path)?; let file_dir = self.get_bucket_path(volume)?; diff --git a/ecstore/src/disk_api.rs b/ecstore/src/disk_api.rs index ff2152adb..b1ea4a33d 100644 --- a/ecstore/src/disk_api.rs +++ b/ecstore/src/disk_api.rs @@ -13,9 +13,10 @@ pub trait DiskAPI: Debug + Send + Sync + 'static { fn is_local(&self) -> bool; async fn read_all(&self, volume: &str, path: &str) -> Result; - async fn write_all(&self, volume: &str, path: &str, data: Bytes) -> 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 rename_data( &self, src_volume: &str, @@ -36,7 +37,7 @@ pub trait DiskAPI: Debug + Send + Sync + 'static { volume: &str, path: &str, version_id: Uuid, - opts: ReadOptions, + opts: &ReadOptions, ) -> Result; async fn read_xl(&self, volume: &str, path: &str, read_data: bool) -> Result; } @@ -48,7 +49,7 @@ pub struct VolumeInfo { pub struct ReadOptions { pub read_data: bool, - // pub healing: bool, + pub healing: bool, } #[derive(Debug, thiserror::Error)] diff --git a/ecstore/src/erasure.rs b/ecstore/src/erasure.rs index 68c726bdb..7c7da1ff2 100644 --- a/ecstore/src/erasure.rs +++ b/ecstore/src/erasure.rs @@ -49,6 +49,8 @@ impl Erasure { for (i, w) in writers.iter_mut().enumerate() { total += blocks[i].len(); + debug!("encode write {}", blocks[i].len()); + match w.write_all(blocks[i].as_ref()).await { Ok(_) => errs.push(None), Err(e) => errs.push(Some(e)), diff --git a/ecstore/src/lib.rs b/ecstore/src/lib.rs index 67cd5f451..caef7b39f 100644 --- a/ecstore/src/lib.rs +++ b/ecstore/src/lib.rs @@ -16,3 +16,4 @@ pub mod store; pub mod store_api; mod store_init; mod utils; +mod writer; diff --git a/ecstore/src/store_api.rs b/ecstore/src/store_api.rs index 36543649f..57f595ed3 100644 --- a/ecstore/src/store_api.rs +++ b/ecstore/src/store_api.rs @@ -26,6 +26,18 @@ impl FileInfo { // TODO: when lifecycle false } + + pub fn write_quorum(&self, quorum: usize) -> usize { + if self.deleted { + return quorum; + } + + if self.erasure.data_blocks == self.erasure.parity_blocks { + return self.erasure.data_blocks + 1; + } + + self.erasure.data_blocks + } } impl Default for FileInfo { diff --git a/ecstore/src/store_init.rs b/ecstore/src/store_init.rs index 0f63501b6..3697ebdbd 100644 --- a/ecstore/src/store_init.rs +++ b/ecstore/src/store_init.rs @@ -245,12 +245,13 @@ async fn save_format_file(disk: &Option, format: &Option) - let format = format.as_ref().unwrap(); - let json_data = format.to_json().map(|data| Bytes::from(data))?; + let json_data = format.to_json()?; let tmpfile = Uuid::new_v4().to_string(); let disk = disk.as_ref().unwrap(); - disk.write_all(RUSTFS_META_BUCKET, tmpfile.as_str(), json_data).await?; + disk.write_all(RUSTFS_META_BUCKET, tmpfile.as_str(), json_data.into_bytes()) + .await?; disk.rename_file(RUSTFS_META_BUCKET, tmpfile.as_str(), RUSTFS_META_BUCKET, FORMAT_CONFIG_FILE) .await?; diff --git a/ecstore/src/writer.rs b/ecstore/src/writer.rs new file mode 100644 index 000000000..877b7a277 --- /dev/null +++ b/ecstore/src/writer.rs @@ -0,0 +1,54 @@ +use std::{io, task::Poll}; + +use futures::Future; +use tokio::io::AsyncWrite; +use tracing::debug; + +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 { + Self { disk, volume, path } + } + + async fn async_write(&self, buf: &[u8]) -> Result<(), std::io::Error> { + debug!("async_write {}: {}", &self.path, buf.len()); + + unimplemented!() + } +} + +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)); + + match fut.as_mut().poll(cx) { + Poll::Pending => Poll::Pending, + Poll::Ready(_) => Poll::Ready(Ok(buf.len())), + } + } + + 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(())) + } +} diff --git a/rustfs/src/storage/ecfs.rs b/rustfs/src/storage/ecfs.rs index 8f259f77f..baf6afb27 100644 --- a/rustfs/src/storage/ecfs.rs +++ b/rustfs/src/storage/ecfs.rs @@ -243,16 +243,30 @@ impl S3 for FS { async fn upload_part(&self, req: S3Request) -> S3Result> { let UploadPartInput { body, - // upload_id, - // part_number, + bucket, + key, + upload_id, + part_number, content_length, .. } = req.input; - let _body = body.ok_or_else(|| s3_error!(IncompleteBody))?; - let _content_length = content_length.ok_or_else(|| s3_error!(IncompleteBody))?; + let part_id = part_number as usize; + + // let upload_id = + + let body = body.ok_or_else(|| s3_error!(IncompleteBody))?; + let content_length = content_length.ok_or_else(|| s3_error!(IncompleteBody))?; // mc cp step 4 + let data = PutObjReader::new(body.into(), content_length as usize); + let opts = ObjectOptions::default(); + + try_!( + self.store + .put_object_part(&bucket, &key, &upload_id, part_id, data, &opts) + .await + ); let output = UploadPartOutput { ..Default::default() }; Ok(S3Response::new(output))