test:FileMeta

This commit is contained in:
weisd
2024-07-05 11:48:34 +08:00
parent 735c08caf4
commit 91c3fc2fe1
3 changed files with 52 additions and 13 deletions
+38 -9
View File
@@ -341,32 +341,61 @@ impl DiskAPI for LocalDisk {
let vol_path = self.get_bucket_path(&src_volume)?; let vol_path = self.get_bucket_path(&src_volume)?;
check_volume_exists(&vol_path).await?; check_volume_exists(&vol_path).await?;
} }
let dst_volume_path = self.get_bucket_path(&dst_volume)?;
if !skip_access_checks(&dst_volume) { if !skip_access_checks(&dst_volume) {
let vol_path = self.get_bucket_path(&dst_volume)?; check_volume_exists(&dst_volume_path).await?;
check_volume_exists(&vol_path).await?;
} }
let src_file_path = self.get_object_path(&src_volume, format!("{}/{}", &src_path, STORAGE_FORMAT_FILE).as_str())?; 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 dst_file_path = self.get_object_path(&dst_volume, format!("{}/{}", &dst_path, STORAGE_FORMAT_FILE).as_str())?;
// let mut data_dir = String::new(); let (src_data_path, dst_data_path) = {
// if !fi.is_remote() { let mut data_dir = String::new();
// data_dir = utils::path::retain_slash(&fi.data_dir); 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 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 (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() { if !dst_buf.is_empty() {
meta = match FileMeta::unmarshal(dst_buf) {
Ok(m) => m,
Err(e) => FileMeta::new(),
}
// xl.load // xl.load
// meta.from(dst_buf); // 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<()> { async fn make_volumes(&self, volumes: Vec<&str>) -> Result<()> {
+5 -2
View File
@@ -32,12 +32,13 @@ impl Erasure {
block_size: usize, block_size: usize,
data_size: usize, data_size: usize,
write_quorum: usize, write_quorum: usize,
) -> Result<()> ) -> Result<usize>
where where
S: Stream<Item = Result<Bytes, StdError>> + Send + Sync + 'static, S: Stream<Item = Result<Bytes, StdError>> + Send + Sync + 'static,
W: AsyncWrite + Unpin, W: AsyncWrite + Unpin,
{ {
let mut stream = ChunkedStream::new(body, data_size, block_size, true); let mut stream = ChunkedStream::new(body, data_size, block_size, true);
let mut total: usize = 0;
while let Some(result) = stream.next().await { while let Some(result) = stream.next().await {
match result { match result {
Ok(data) => { Ok(data) => {
@@ -46,6 +47,8 @@ impl Erasure {
let mut errs = Vec::new(); let mut errs = Vec::new();
for (i, w) in writers.iter_mut().enumerate() { for (i, w) in writers.iter_mut().enumerate() {
total += blocks[i].len();
match w.write_all(blocks[i].as_ref()).await { match w.write_all(blocks[i].as_ref()).await {
Ok(_) => errs.push(None), Ok(_) => errs.push(None),
Err(e) => errs.push(Some(e)), Err(e) => errs.push(Some(e)),
@@ -59,7 +62,7 @@ impl Erasure {
} }
} }
Ok(()) Ok(total)
// loop { // loop {
// match rd.next().await { // match rd.next().await {
+9 -2
View File
@@ -176,7 +176,7 @@ impl StorageAPI for Sets {
let parts_metadata = vec![fi.clone(); disks.len()]; 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()); 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); 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) .encode(data.stream, &mut writers, fi.erasure.block_size, data.content_length, write_quorum)
.await?; .await?;
@@ -235,6 +235,11 @@ impl StorageAPI for Sets {
// TODO: reduceWriteQuorumErrs // TODO: reduceWriteQuorumErrs
// evalDisks // evalDisks
for fi in shuffle_parts_metadata.iter_mut() {
fi.mod_time = OffsetDateTime::now_utc();
fi.size = w_size;
}
let rename_errs = self let rename_errs = self
.rename_data( .rename_data(
&shuffle_disks, &shuffle_disks,
@@ -248,6 +253,8 @@ impl StorageAPI for Sets {
// TODO: reduceWriteQuorumErrs // TODO: reduceWriteQuorumErrs
debug!("put_object rename_errs:{:?}", rename_errs);
// self.commit_rename_data_dir(&shuffle_disks,&bucket,&object,) // self.commit_rename_data_dir(&shuffle_disks,&bucket,&object,)
Ok(()) Ok(())