bug:ec.encode

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<Box<dyn Future<Output = T> + 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));
                 }

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<u8>) -> 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<FileInfo> {
         let file_path = self.get_object_path(volume, path)?;
         let file_dir = self.get_bucket_path(volume)?;

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<Bytes>;
-    async fn write_all(&self, volume: &str, path: &str, data: Bytes) -> Result<()>;
+    async fn write_all(&self, volume: &str, path: &str, data: Vec<u8>) -> 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<FileInfo>;
     async fn read_xl(&self, volume: &str, path: &str, read_data: bool) -> Result<RawFileInfo>;
 }
@@ -48,7 +49,7 @@ pub struct VolumeInfo {
 
 pub struct ReadOptions {
     pub read_data: bool,
-    // pub healing: bool,
+    pub healing: bool,
 }
 
 #[derive(Debug, thiserror::Error)]

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

ecstore/src/lib.rs

@@ -16,3 +16,4 @@ pub mod store;
 pub mod store_api;
 mod store_init;
 mod utils;
+mod writer;

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 {

ecstore/src/store_init.rs

@@ -245,12 +245,13 @@ async fn save_format_file(disk: &Option<DiskStore>, format: &Option<FormatV3>) -
 
     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?;

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<Result<usize, std::io::Error>> {
+        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<Result<(), std::io::Error>> {
+        Poll::Ready(Ok(()))
+    }
+
+    fn poll_shutdown(
+        self: std::pin::Pin<&mut Self>,
+        _cx: &mut std::task::Context<'_>,
+    ) -> std::task::Poll<Result<(), std::io::Error>> {
+        Poll::Ready(Ok(()))
+    }
+}

rustfs/src/storage/ecfs.rs

@@ -243,16 +243,30 @@ impl S3 for FS {
     async fn upload_part(&self, req: S3Request<UploadPartInput>) -> S3Result<S3Response<UploadPartOutput>> {
         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))