todo:put_object

ecstore/src/disk.rs

@@ -16,6 +16,7 @@ use tokio::{
     fs::{self, File},
     io::DuplexStream,
 };
+use tracing::debug;
 use uuid::Uuid;
 
 use crate::{
@@ -167,7 +168,10 @@ impl LocalDisk {
     /// This is done by first writing to a temporary location and then moving the file.
     pub(crate) async fn prepare_file_write<'a>(&self, path: &'a PathBuf) -> Result<FileWriter<'a>> {
         let tmp_path = self.get_object_path(RUSTFS_META_TMP_BUCKET, Uuid::new_v4().to_string().as_str())?;
-        let file = File::create(&path).await?;
+
+        debug!("prepare_file_write tmp_path:{:?}, path:{:?}", &tmp_path, &path);
+
+        let file = File::create(&tmp_path).await?;
         let writer = BufWriter::new(file);
         Ok(FileWriter {
             tmp_path,
@@ -308,14 +312,20 @@ impl DiskAPI for LocalDisk {
         Ok(())
     }
 
-    async fn CreateFile(&self, origvolume: &str, volume: &str, path: &str, fileSize: usize, mut r: DuplexStream) -> Result<()> {
+    async fn create_file(&self, origvolume: &str, volume: &str, path: &str, fileSize: usize, mut r: DuplexStream) -> Result<()> {
         let fpath = self.get_object_path(volume, path)?;
 
-        let mut writer = self.prepare_file_write(&fpath).await?;
+        debug!("CreateFile fpath: {:?}", fpath);
 
-        io::copy(&mut r, writer.writer()).await?;
+        if let Some(_dir_path) = fpath.parent() {
+            fs::create_dir_all(&_dir_path).await?;
+        }
 
-        writer.done().await?;
+        let file = File::create(&fpath).await?;
+
+        let mut writer = BufWriter::new(file);
+
+        io::copy(&mut r, &mut writer).await?;
 
         Ok(())
     }

ecstore/src/disk_api.rs

@@ -11,7 +11,7 @@ pub trait DiskAPI: Debug + Send + Sync + 'static {
     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 rename_file(&self, src_volume: &str, src_path: &str, dst_volume: &str, dst_path: &str) -> Result<()>;
-    async fn CreateFile(&self, origvolume: &str, volume: &str, path: &str, fileSize: usize, r: DuplexStream) -> Result<()>;
+    async fn create_file(&self, origvolume: &str, volume: &str, path: &str, fileSize: usize, r: DuplexStream) -> Result<()>;
 
     async fn make_volumes(&self, volume: Vec<&str>) -> Result<()>;
     async fn make_volume(&self, volume: &str) -> Result<()>;

ecstore/src/sets.rs

@@ -2,13 +2,13 @@ use std::sync::Arc;
 
 use anyhow::Result;
 
-use futures::{AsyncWrite, StreamExt};
+use futures::{future::join_all, AsyncWrite, StreamExt};
 use time::OffsetDateTime;
 use tracing::debug;
 use uuid::Uuid;
 
 use crate::{
-    disk::{self, DiskStore},
+    disk::{self, DiskStore, RUSTFS_META_TMP_BUCKET},
     endpoint::PoolEndpoints,
     erasure::Erasure,
     format::{DistributionAlgoVersion, FormatV3},
@@ -90,6 +90,10 @@ impl Sets {
             }
         }
     }
+
+    async fn rename_data(&self) -> Result<()> {
+        unimplemented!()
+    }
 }
 
 // #[derive(Debug)]
@@ -132,22 +136,27 @@ impl StorageAPI for Sets {
 
         let mut writers = Vec::with_capacity(disks.len());
 
+        let mut futures = Vec::with_capacity(disks.len());
+
+        let tmp_dir = Uuid::new_v4().to_string();
+
+        let tmp_object = format!("{}/{}/part.1", tmp_dir, fi.data_dir);
+
         for disk in shuffle_disks.iter() {
             let (reader, writer) = tokio::io::duplex(fi.erasure.block_size);
 
             let disk = disk.as_ref().unwrap().clone();
-            let bucket = bucket.to_string();
-            let object = object.to_string();
-            tokio::spawn(async move {
-                debug!("do createfile");
-                match disk
-                    .CreateFile("", bucket.as_str(), object.as_str(), data.content_length, reader)
+            let tmp_object = tmp_object.clone();
+
+            futures.push(async move {
+                disk.create_file("", RUSTFS_META_TMP_BUCKET, tmp_object.as_str(), data.content_length, reader)
                     .await
-                {
-                    Ok(_) => (),
-                    Err(e) => debug!("creatfile err :{:?}", e),
-                }
             });
+            // futures.push(tokio::spawn(async move {
+            //     debug!("do createfile");
+            //     disk.CreateFile("", bucket.as_str(), object.as_str(), data.content_length, reader)
+            //         .await;
+            // }));
 
             writers.push(writer);
         }
@@ -158,7 +167,29 @@ impl StorageAPI for Sets {
             .encode(data.stream, &mut writers, fi.erasure.block_size, data.content_length, write_quorum)
             .await?;
 
-        unimplemented!()
+        // close reader in create_file
+        drop(writers);
+
+        let mut errors = Vec::with_capacity(disks.len());
+
+        let results = join_all(futures).await;
+        for result in results {
+            match result {
+                Ok(_) => {
+                    errors.push(None);
+                }
+                Err(e) => {
+                    errors.push(Some(e));
+                }
+            }
+        }
+
+        debug!("CreateFile errs:{:?}", errors);
+
+        // TODO: reduceWriteQuorumErrs
+        // evalDisks
+
+        Ok(())
     }
 }
 

ecstore/src/store.rs

@@ -3,6 +3,7 @@ use std::collections::HashMap;
 use anyhow::{Error, Result};
 
 use s3s::{dto::StreamingBlob, Body};
+use tracing::debug;
 use uuid::Uuid;
 
 use crate::{
@@ -139,6 +140,7 @@ impl StorageAPI for ECStore {
         let object = utils::path::encode_dir_object(object);
 
         if self.single_pool() {
+            println!("put_object single_pool");
             self.pools[0].put_object(bucket, object.as_str(), data, opts).await?;
             return Ok(());
         }

ecstore/src/store_api.rs

@@ -5,6 +5,7 @@ use bytes::Bytes;
 use futures::Stream;
 use s3s::{dto::StreamingBlob, Body};
 use time::OffsetDateTime;
+use tracing::debug;
 
 pub const ERASURE_ALGORITHM: &str = "rs-vandermonde";
 pub const BLOCK_SIZE_V2: usize = 1048576; // 1M
@@ -38,7 +39,7 @@ impl FileInfo {
 
             let start = key_crc as usize % cardinality;
             for i in 1..=cardinality {
-                nums[i - 1] = 1 + ((start + 1) % cardinality);
+                nums[i - 1] = 1 + ((start + i) % cardinality);
             }
 
             nums

rustfs/src/storage/ecfs.rs

@@ -1,5 +1,6 @@
 use std::fmt::Debug;
 
+use ecstore::store_api::MakeBucketOptions;
 use ecstore::store_api::ObjectOptions;
 use ecstore::store_api::PutObjReader;
 use ecstore::store_api::StorageAPI;
@@ -44,6 +45,12 @@ impl S3 for FS {
     async fn create_bucket(&self, req: S3Request<CreateBucketInput>) -> S3Result<S3Response<CreateBucketOutput>> {
         let input = req.input;
 
+        try_!(
+            self.store
+                .make_bucket(&input.bucket, &MakeBucketOptions { force_create: true })
+                .await
+        );
+
         let output = CreateBucketOutput::default(); // TODO: handle other fields
         Ok(S3Response::new(output))
     }

scripts/run.sh

@@ -9,7 +9,7 @@ if [ -n "$1" ]; then
 fi
 
 if [ -z "$RUST_LOG" ]; then
-    export RUST_LOG="s3s-rustfs=debug,s3s=debug"
+    export RUST_LOG="s3s-rustfs=debug,ecstore=debug,s3s=debug"
 fi
 
 cargo run \