From 91c3fc2fe1cb2e0f0373b22e2a32d7e759d314af Mon Sep 17 00:00:00 2001 From: weisd Date: Fri, 5 Jul 2024 11:48:34 +0800 Subject: [PATCH 1/2] test:FileMeta --- ecstore/src/disk.rs | 47 ++++++++++++++++++++++++++++++++++-------- ecstore/src/erasure.rs | 7 +++++-- ecstore/src/sets.rs | 11 ++++++++-- 3 files changed, 52 insertions(+), 13 deletions(-) diff --git a/ecstore/src/disk.rs b/ecstore/src/disk.rs index 22a422689..ab1597b4b 100644 --- a/ecstore/src/disk.rs +++ b/ecstore/src/disk.rs @@ -341,32 +341,61 @@ impl DiskAPI for LocalDisk { let vol_path = self.get_bucket_path(&src_volume)?; check_volume_exists(&vol_path).await?; } + + let dst_volume_path = self.get_bucket_path(&dst_volume)?; if !skip_access_checks(&dst_volume) { - let vol_path = self.get_bucket_path(&dst_volume)?; - check_volume_exists(&vol_path).await?; + check_volume_exists(&dst_volume_path).await?; } let src_file_path = self.get_object_path(&src_volume, format!("{}/{}", &src_path, STORAGE_FORMAT_FILE).as_str())?; let dst_file_path = self.get_object_path(&dst_volume, format!("{}/{}", &dst_path, STORAGE_FORMAT_FILE).as_str())?; - // let mut data_dir = String::new(); - // if !fi.is_remote() { - // data_dir = utils::path::retain_slash(&fi.data_dir); - // } + let (src_data_path, dst_data_path) = { + let mut data_dir = String::new(); + if !fi.is_remote() { + data_dir = utils::path::retain_slash(fi.data_dir.to_string().as_str()); + } - // if !data_dir.is_empty() {} + if !data_dir.is_empty() { + let src_data_path = self.get_object_path( + &src_volume, + utils::path::retain_slash(format!("{}/{}", &src_path, data_dir).as_str()).as_str(), + )?; + let dst_data_path = self.get_object_path(&dst_volume, format!("{}/{}", &dst_path, data_dir).as_str())?; + + (src_data_path, dst_data_path) + } else { + (PathBuf::new(), PathBuf::new()) + } + }; let curreng_data_path = self.get_object_path(&dst_volume, &dst_path); - let meta = FileMeta::new(); + let mut meta = FileMeta::new(); let (dst_buf, _) = read_file_exists(&dst_file_path).await?; + + let mut skipParent = dst_volume_path.as_path(); + if !&dst_buf.is_empty() { + skipParent = skipParent.parent().unwrap_or(Path::new("/")); + } + if !dst_buf.is_empty() { + meta = match FileMeta::unmarshal(dst_buf) { + Ok(m) => m, + Err(e) => FileMeta::new(), + } // xl.load // meta.from(dst_buf); } - unimplemented!() + meta.add_version(fi.clone())?; + + let fm_data = meta.marshal_msg()?; + + fs::write(&src_file_path, fm_data).await?; + + Ok(()) } async fn make_volumes(&self, volumes: Vec<&str>) -> Result<()> { diff --git a/ecstore/src/erasure.rs b/ecstore/src/erasure.rs index 2d9cffbed..b560a443a 100644 --- a/ecstore/src/erasure.rs +++ b/ecstore/src/erasure.rs @@ -32,12 +32,13 @@ impl Erasure { block_size: usize, data_size: usize, write_quorum: usize, - ) -> Result<()> + ) -> Result where S: Stream> + Send + Sync + 'static, W: AsyncWrite + Unpin, { let mut stream = ChunkedStream::new(body, data_size, block_size, true); + let mut total: usize = 0; while let Some(result) = stream.next().await { match result { Ok(data) => { @@ -46,6 +47,8 @@ impl Erasure { let mut errs = Vec::new(); for (i, w) in writers.iter_mut().enumerate() { + total += blocks[i].len(); + match w.write_all(blocks[i].as_ref()).await { Ok(_) => errs.push(None), Err(e) => errs.push(Some(e)), @@ -59,7 +62,7 @@ impl Erasure { } } - Ok(()) + Ok(total) // loop { // match rd.next().await { diff --git a/ecstore/src/sets.rs b/ecstore/src/sets.rs index 4096b6eb6..3f9814497 100644 --- a/ecstore/src/sets.rs +++ b/ecstore/src/sets.rs @@ -176,7 +176,7 @@ impl StorageAPI for Sets { let parts_metadata = vec![fi.clone(); disks.len()]; - let (shuffle_disks, shuffle_parts_metadata) = shuffle_disks_and_parts_metadata(&disks, &parts_metadata, &fi); + let (shuffle_disks, mut shuffle_parts_metadata) = shuffle_disks_and_parts_metadata(&disks, &parts_metadata, &fi); let mut writers = Vec::with_capacity(disks.len()); @@ -209,7 +209,7 @@ impl StorageAPI for Sets { let erasure = Erasure::new(fi.erasure.data_blocks, fi.erasure.parity_blocks); - erasure + let w_size = erasure .encode(data.stream, &mut writers, fi.erasure.block_size, data.content_length, write_quorum) .await?; @@ -235,6 +235,11 @@ impl StorageAPI for Sets { // TODO: reduceWriteQuorumErrs // evalDisks + for fi in shuffle_parts_metadata.iter_mut() { + fi.mod_time = OffsetDateTime::now_utc(); + fi.size = w_size; + } + let rename_errs = self .rename_data( &shuffle_disks, @@ -248,6 +253,8 @@ impl StorageAPI for Sets { // TODO: reduceWriteQuorumErrs + debug!("put_object rename_errs:{:?}", rename_errs); + // self.commit_rename_data_dir(&shuffle_disks,&bucket,&object,) Ok(()) From 98df9b91ba7e831231de257c906dc8e5254b80d5 Mon Sep 17 00:00:00 2001 From: weisd Date: Fri, 5 Jul 2024 17:29:31 +0800 Subject: [PATCH 2/2] test:put_object --- ecstore/src/disk.rs | 61 ++++++++++++++++++++++++++++++++++++---- ecstore/src/sets.rs | 2 ++ ecstore/src/store_api.rs | 2 ++ 3 files changed, 60 insertions(+), 5 deletions(-) diff --git a/ecstore/src/disk.rs b/ecstore/src/disk.rs index ab1597b4b..7c0201363 100644 --- a/ecstore/src/disk.rs +++ b/ecstore/src/disk.rs @@ -184,6 +184,43 @@ impl LocalDisk { clean_tmp: true, }) } + + pub async fn rename_all(&self, src_data_path: &PathBuf, dst_data_path: &PathBuf, skip: &PathBuf) -> Result<()> { + if !skip.starts_with(&src_data_path) { + fs::create_dir_all(dst_data_path.parent().unwrap_or(Path::new("/"))).await?; + } + + debug!( + "rename_all from \n {:?} \n to \n {:?} \n skip:{:?}", + &src_data_path, &dst_data_path, &skip + ); + + fs::rename(&src_data_path, &dst_data_path).await?; + + Ok(()) + } + + pub async fn delete_file(&self, base_path: &PathBuf, delete_path: &PathBuf) -> Result<()> { + if is_root_path(base_path) || is_root_path(delete_path) { + return Ok(()); + } + + if !delete_path.starts_with(base_path) || base_path == delete_path { + return Ok(()); + } + + if delete_path.is_dir() { + fs::remove_dir(delete_path).await?; + } else { + fs::remove_file(delete_path).await?; + } + + Ok(()) + } +} + +fn is_root_path(path: &PathBuf) -> bool { + path.components().count() == 1 && path.has_root() } // 过滤 std::io::ErrorKind::NotFound @@ -337,9 +374,9 @@ impl DiskAPI for LocalDisk { } 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)?; if !skip_access_checks(&src_volume) { - let vol_path = self.get_bucket_path(&src_volume)?; - check_volume_exists(&vol_path).await?; + check_volume_exists(&src_volume_path).await?; } let dst_volume_path = self.get_bucket_path(&dst_volume)?; @@ -369,15 +406,15 @@ impl DiskAPI for LocalDisk { } }; - let curreng_data_path = self.get_object_path(&dst_volume, &dst_path); + // let curreng_data_path = self.get_object_path(&dst_volume, &dst_path); let mut meta = FileMeta::new(); let (dst_buf, _) = read_file_exists(&dst_file_path).await?; - let mut skipParent = dst_volume_path.as_path(); + let mut skipParent = dst_volume_path; if !&dst_buf.is_empty() { - skipParent = skipParent.parent().unwrap_or(Path::new("/")); + skipParent = PathBuf::from(&dst_file_path.parent().unwrap_or(Path::new("/"))); } if !dst_buf.is_empty() { @@ -395,6 +432,20 @@ impl DiskAPI for LocalDisk { fs::write(&src_file_path, fm_data).await?; + let no_inline = src_data_path.has_root() && fi.data.is_empty() && fi.size > 0; + if no_inline { + self.rename_all(&src_data_path, &dst_data_path, &skipParent).await?; + } + + self.rename_all(&src_file_path, &dst_file_path, &skipParent).await?; + + if src_volume != RUSTFS_META_MULTIPART_BUCKET { + fs::remove_dir(&src_file_path.parent().unwrap()).await?; + } else { + self.delete_file(&src_volume_path, &PathBuf::from(src_file_path.parent().unwrap())) + .await?; + } + Ok(()) } diff --git a/ecstore/src/sets.rs b/ecstore/src/sets.rs index 3f9814497..4516e64e9 100644 --- a/ecstore/src/sets.rs +++ b/ecstore/src/sets.rs @@ -16,6 +16,8 @@ use crate::{ utils::hash, }; +const DEFAULT_INLINE_BLOCKS: usize = 128 * 1024; + #[derive(Debug)] pub struct Sets { pub id: Uuid, diff --git a/ecstore/src/store_api.rs b/ecstore/src/store_api.rs index aa608be6e..d6acf6241 100644 --- a/ecstore/src/store_api.rs +++ b/ecstore/src/store_api.rs @@ -20,6 +20,7 @@ pub struct FileInfo { pub data_dir: Uuid, pub mod_time: OffsetDateTime, pub size: usize, + pub data: Vec, } impl FileInfo { @@ -38,6 +39,7 @@ impl Default for FileInfo { data_dir: Uuid::nil(), mod_time: OffsetDateTime::UNIX_EPOCH, size: Default::default(), + data: Default::default(), } } }