From 65b883ace919b64a70661b3225dc11091300f38c Mon Sep 17 00:00:00 2001 From: weisd Date: Wed, 3 Jul 2024 11:45:29 +0800 Subject: [PATCH] todo:put_object --- ecstore/src/disk.rs | 20 +++++++++---- ecstore/src/disk_api.rs | 2 +- ecstore/src/sets.rs | 57 +++++++++++++++++++++++++++++--------- ecstore/src/store.rs | 2 ++ ecstore/src/store_api.rs | 3 +- rustfs/src/storage/ecfs.rs | 7 +++++ scripts/run.sh | 2 +- 7 files changed, 72 insertions(+), 21 deletions(-) diff --git a/ecstore/src/disk.rs b/ecstore/src/disk.rs index b10a7d889..95f78a9e7 100644 --- a/ecstore/src/disk.rs +++ b/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> { 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(()) } diff --git a/ecstore/src/disk_api.rs b/ecstore/src/disk_api.rs index ebb6fe35f..476e0e52a 100644 --- a/ecstore/src/disk_api.rs +++ b/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; 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<()>; diff --git a/ecstore/src/sets.rs b/ecstore/src/sets.rs index 2cdd3536c..393bd8c18 100644 --- a/ecstore/src/sets.rs +++ b/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(()) } } diff --git a/ecstore/src/store.rs b/ecstore/src/store.rs index 7f90928af..da02f18b7 100644 --- a/ecstore/src/store.rs +++ b/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(()); } diff --git a/ecstore/src/store_api.rs b/ecstore/src/store_api.rs index 8c25e1eee..5ce0c67d3 100644 --- a/ecstore/src/store_api.rs +++ b/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 diff --git a/rustfs/src/storage/ecfs.rs b/rustfs/src/storage/ecfs.rs index ed34c81e3..a3fb35e13 100644 --- a/rustfs/src/storage/ecfs.rs +++ b/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) -> S3Result> { 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)) } diff --git a/scripts/run.sh b/scripts/run.sh index 16df7c98a..d0829fe6a 100755 --- a/scripts/run.sh +++ b/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 \