gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-08-24 13:16:28 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix:#159, fix:#160

Browse Source
This commit is contained in:
weisd
2024-12-11 16:05:29 +08:00
parent 87b985ad36
commit f6e3575a50
4 changed files with 166 additions and 161 deletions
Download Patch File Download Diff File Expand all files Collapse all files
ecstore/src/erasure.rs
+14 -7
View File
@@ -533,14 +533,21 @@ impl ShardReader {
let mut ress = Vec::with_capacity(reader_length); let mut ress = Vec::with_capacity(reader_length);
for disk in self.readers.iter_mut() { for disk in self.readers.iter_mut() {
if disk.is_none() { // if disk.is_none() {
ress.push(None); // ress.push(None);
errors.push(Some(Error::new(DiskError::DiskNotFound))); // errors.push(Some(Error::new(DiskError::DiskNotFound)));
continue; // continue;
} // }
let disk: &mut BitrotReader = disk.as_mut().unwrap(); // let disk: &mut BitrotReader = disk.as_mut().unwrap();
futures.push(disk.read_at(self.offset, read_length)); let offset = self.offset;
futures.push(async move {
if let Some(disk) = disk {
disk.read_at(offset, read_length).await
} else {
Err(Error::new(DiskError::DiskNotFound))
}
});
} }
let results = join_all(futures).await; let results = join_all(futures).await;
ecstore/src/set_disk.rs
+143 -145
View File
@@ -197,25 +197,24 @@ impl SetDisks {
let mut errs = Vec::with_capacity(disks.len()); let mut errs = Vec::with_capacity(disks.len());
for (i, disk) in disks.iter().enumerate() { for (i, disk) in disks.iter().enumerate() {
if disk.is_none() {
// ress.push(None);
errs.push(Some(Error::new(DiskError::DiskNotFound)));
continue;
}
let disk = disk.as_ref().unwrap();
let mut file_info = file_infos[i].clone(); let mut file_info = file_infos[i].clone();
if file_info.erasure.index == 0 { futures.push(async move {
file_info.erasure.index = i + 1; if file_info.erasure.index == 0 {
} file_info.erasure.index = i + 1;
}
if !file_info.is_valid() { if !file_info.is_valid() {
// ress.push(None); return Err(Error::new(DiskError::FileCorrupt));
errs.push(Some(Error::new(DiskError::FileCorrupt))); }
continue;
}
futures.push(disk.rename_data(src_bucket, src_object, file_info, dst_bucket, dst_object)) if let Some(disk) = disk {
disk.rename_data(src_bucket, src_object, file_info, dst_bucket, dst_object)
.await
} else {
Err(Error::new(DiskError::DiskNotFound))
}
})
} }
let mut disk_versions = vec![None; disks.len()]; let mut disk_versions = vec![None; disks.len()];
@@ -329,20 +328,22 @@ impl SetDisks {
let mut errs = Vec::with_capacity(disks.len()); let mut errs = Vec::with_capacity(disks.len());
for disk in disks.iter() { for disk in disks.iter() {
if disk.is_none() { let file_path = file_path.clone();
errs.push(Some(Error::new(DiskError::DiskNotFound))); futures.push(async move {
continue; if let Some(disk) = disk {
} disk.delete(
bucket,
let disk = disk.as_ref().unwrap(); &file_path,
futures.push(disk.delete( DeleteOptions {
bucket, recursive: true,
&file_path, ..Default::default()
DeleteOptions { },
recursive: true, )
..Default::default() .await
}, } else {
)); Err(Error::new(DiskError::DiskNotFound))
}
});
} }
let results = join_all(futures).await; let results = join_all(futures).await;
@@ -370,12 +371,13 @@ impl SetDisks {
let mut errs = Vec::with_capacity(disks.len()); let mut errs = Vec::with_capacity(disks.len());
for disk in disks.iter() { for disk in disks.iter() {
if disk.is_none() { futures.push(async move {
errs.push(Some(Error::new(DiskError::DiskNotFound))); if let Some(disk) = disk {
continue; disk.delete_paths(RUSTFS_META_MULTIPART_BUCKET, paths).await
} } else {
let disk = disk.as_ref().unwrap(); Err(Error::new(DiskError::DiskNotFound))
futures.push(disk.delete_paths(RUSTFS_META_MULTIPART_BUCKET, paths)) }
})
} }
let results = join_all(futures).await; let results = join_all(futures).await;
@@ -404,12 +406,14 @@ impl SetDisks {
let mut errs = Vec::with_capacity(disks.len()); let mut errs = Vec::with_capacity(disks.len());
for disk in disks.iter() { for disk in disks.iter() {
if disk.is_none() { let meta = meta.clone();
errs.push(Some(Error::new(DiskError::DiskNotFound))); futures.push(async move {
continue; if let Some(disk) = disk {
} disk.rename_part(src_bucket, src_object, dst_bucket, dst_object, meta).await
let disk = disk.as_ref().unwrap(); } else {
futures.push(disk.rename_part(src_bucket, src_object, dst_bucket, dst_object, meta.clone())) Err(Error::new(DiskError::DiskNotFound))
}
})
} }
let results = join_all(futures).await; let results = join_all(futures).await;
@@ -490,15 +494,15 @@ impl SetDisks {
let mut errors = Vec::with_capacity(disks.len()); let mut errors = Vec::with_capacity(disks.len());
for (i, disk) in disks.iter().enumerate() { for (i, disk) in disks.iter().enumerate() {
if disk.is_none() {
errors.push(Some(Error::new(DiskError::DiskNotFound)));
continue;
}
let disk = disk.as_ref().unwrap();
let mut file_info = files[i].clone(); let mut file_info = files[i].clone();
file_info.erasure.index = i + 1; file_info.erasure.index = i + 1;
futures.push(disk.write_metadata(org_bucket, bucket, prefix, file_info)); futures.push(async move {
if let Some(disk) = disk {
disk.write_metadata(org_bucket, bucket, prefix, file_info).await
} else {
Err(Error::new(DiskError::DiskNotFound))
}
});
} }
let results = join_all(futures).await; let results = join_all(futures).await;
@@ -964,25 +968,22 @@ impl SetDisks {
let mut errors = Vec::with_capacity(disks.len()); let mut errors = Vec::with_capacity(disks.len());
for disk in disks.iter() { for disk in disks.iter() {
if disk.is_none() {
ress.push(FileInfo::default());
errors.push(Some(Error::new(DiskError::DiskNotFound)));
continue;
}
let disk = disk.as_ref().unwrap();
let opts = ReadOptions { read_data, healing }; let opts = ReadOptions { read_data, healing };
futures.push(async move { futures.push(async move {
if version_id.is_empty() { if let Some(disk) = disk {
match disk.read_xl(bucket, object, read_data).await { if version_id.is_empty() {
Ok(info) => { match disk.read_xl(bucket, object, read_data).await {
let fi = file_info_from_raw(info, bucket, object, read_data).await?; Ok(info) => {
Ok(fi) let fi = file_info_from_raw(info, bucket, object, read_data).await?;
Ok(fi)
}
Err(err) => Err(err),
} }
Err(err) => Err(err), } else {
disk.read_version(org_bucket, bucket, object, version_id, &opts).await
} }
} else { } else {
disk.read_version(org_bucket, bucket, object, version_id, &opts).await Err(Error::new(DiskError::DiskNotFound))
} }
}) })
} }
@@ -1026,14 +1027,13 @@ impl SetDisks {
let mut errors = Vec::with_capacity(disks.len()); let mut errors = Vec::with_capacity(disks.len());
for disk in disks.iter() { for disk in disks.iter() {
if disk.is_none() { futures.push(async move {
ress.push(None); if let Some(disk) = disk {
errors.push(Some(Error::new(DiskError::DiskNotFound))); disk.read_xl(bucket, object, read_data).await
continue; } else {
} Err(Error::new(DiskError::DiskNotFound))
}
let disk = disk.as_ref().unwrap(); });
futures.push(disk.read_xl(bucket, object, read_data));
} }
let results = join_all(futures).await; let results = join_all(futures).await;
@@ -1152,15 +1152,14 @@ impl SetDisks {
let mut errors = Vec::with_capacity(disks.len()); let mut errors = Vec::with_capacity(disks.len());
for disk in disks.iter() { for disk in disks.iter() {
if disk.is_none() {
ress.push(None);
errors.push(Some(Error::new(DiskError::DiskNotFound)));
continue;
}
let disk = disk.as_ref().unwrap();
let req = req.clone(); let req = req.clone();
futures.push(disk.read_multiple(req)); futures.push(async move {
if let Some(disk) = disk {
disk.read_multiple(req).await
} else {
Err(Error::new(DiskError::DiskNotFound))
}
});
} }
let results = join_all(futures).await; let results = join_all(futures).await;
@@ -1328,17 +1327,14 @@ impl SetDisks {
let mut ress = Vec::new(); let mut ress = Vec::new();
for disk in disks.iter() { for disk in disks.iter() {
if disk.is_none() {
ress.push(None);
errs.push(Some(Error::new(DiskError::DiskNotFound)));
continue;
}
let disk = disk.as_ref().unwrap();
let opts = opts.clone(); let opts = opts.clone();
// let mut wr = &mut wr; futures.push(async move {
futures.push(disk.walk_dir(opts, Writer::NotUse)); if let Some(disk) = disk {
// tokio::spawn(async move { disk.walk_dir(opts, wr).await }); disk.walk_dir(opts, Writer::NotUse).await
} else {
Err(Error::new(DiskError::DiskNotFound))
}
});
} }
let results = join_all(futures).await; let results = join_all(futures).await;
@@ -1378,20 +1374,18 @@ impl SetDisks {
let mut errors = Vec::with_capacity(disks.len()); let mut errors = Vec::with_capacity(disks.len());
for disk in disks.iter() { for disk in disks.iter() {
if disk.is_none() {
errors.push(Some(Error::new(DiskError::DiskNotFound)));
continue;
}
let disk = disk.as_ref().unwrap();
let file_path = file_path.clone(); let file_path = file_path.clone();
let meta_file_path = format!("{}.meta", file_path); let meta_file_path = format!("{}.meta", file_path);
futures.push(async move { futures.push(async move {
disk.delete(RUSTFS_META_MULTIPART_BUCKET, &file_path, DeleteOptions::default()) if let Some(disk) = disk {
.await?; disk.delete(RUSTFS_META_MULTIPART_BUCKET, &file_path, DeleteOptions::default())
disk.delete(RUSTFS_META_MULTIPART_BUCKET, &meta_file_path, DeleteOptions::default()) .await?;
.await disk.delete(RUSTFS_META_MULTIPART_BUCKET, &meta_file_path, DeleteOptions::default())
.await
} else {
Err(Error::new(DiskError::DiskNotFound))
}
}); });
} }
@@ -1422,13 +1416,15 @@ impl SetDisks {
let mut errors = Vec::with_capacity(disks.len()); let mut errors = Vec::with_capacity(disks.len());
for disk in disks.iter() { for disk in disks.iter() {
if disk.is_none() { let file_path = file_path.clone();
errors.push(Some(Error::new(DiskError::DiskNotFound))); futures.push(async move {
continue; if let Some(disk) = disk {
} disk.delete(RUSTFS_META_MULTIPART_BUCKET, &file_path, DeleteOptions::default())
.await
let disk = disk.as_ref().unwrap(); } else {
futures.push(disk.delete(RUSTFS_META_MULTIPART_BUCKET, &file_path, DeleteOptions::default())); Err(Error::new(DiskError::DiskNotFound))
}
});
} }
let results = join_all(futures).await; let results = join_all(futures).await;
@@ -1456,20 +1452,21 @@ impl SetDisks {
let mut errors = Vec::with_capacity(disks.len()); let mut errors = Vec::with_capacity(disks.len());
for disk in disks.iter() { for disk in disks.iter() {
if disk.is_none() { futures.push(async move {
errors.push(Some(Error::new(DiskError::DiskNotFound))); if let Some(disk) = disk {
continue; disk.delete(
} bucket,
prefix,
let disk = disk.as_ref().unwrap(); DeleteOptions {
futures.push(disk.delete( recursive: true,
bucket, ..Default::default()
prefix, },
DeleteOptions { )
recursive: true, .await
..Default::default() } else {
}, Err(Error::new(DiskError::DiskNotFound))
)); }
});
} }
let results = join_all(futures).await; let results = join_all(futures).await;
@@ -1608,7 +1605,7 @@ impl SetDisks {
// TODO: 优化并发 可用数量中断 // TODO: 优化并发 可用数量中断
let (parts_metadata, errs) = Self::read_all_fileinfo(&disks, "", bucket, object, vid.as_str(), read_data, false).await; let (parts_metadata, errs) = Self::read_all_fileinfo(&disks, "", bucket, object, vid.as_str(), read_data, false).await;
// warn!("get_object_fileinfo parts_metadata {:?}", &parts_metadata); // warn!("get_object_fileinfo parts_metadata {:?}", &parts_metadata);
// warn!("get_object_fileinfo {}/{} errs {:?}", bucket, object, &errs); warn!("get_object_fileinfo {}/{} errs {:?}", bucket, object, &errs);
let _min_disks = self.set_drive_count - self.default_parity_count; let _min_disks = self.set_drive_count - self.default_parity_count;
@@ -1629,9 +1626,9 @@ impl SetDisks {
let fi = Self::pick_valid_fileinfo(&parts_metadata, mot_time, etag, read_quorum as usize)?; let fi = Self::pick_valid_fileinfo(&parts_metadata, mot_time, etag, read_quorum as usize)?;
// debug!("get_object_fileinfo pick fi {:?}", &fi); // debug!("get_object_fileinfo pick fi {:?}", &fi);
let online_disks: Vec<Option<DiskStore>> = op_online_disks.iter().filter(|v| v.is_some()).cloned().collect(); // let online_disks: Vec<Option<DiskStore>> = op_online_disks.iter().filter(|v| v.is_some()).cloned().collect();
Ok((fi, parts_metadata, online_disks)) Ok((fi, parts_metadata, op_online_disks))
} }
#[allow(clippy::too_many_arguments)] #[allow(clippy::too_many_arguments)]
@@ -1766,12 +1763,14 @@ impl SetDisks {
let mut errs = Vec::with_capacity(disks.len()); let mut errs = Vec::with_capacity(disks.len());
for disk in disks.iter() { for disk in disks.iter() {
if disk.is_none() { let fi = fi.clone();
errs.push(Some(Error::new(DiskError::DiskNotFound))); futures.push(async move {
continue; if let Some(disk) = disk {
} disk.update_metadata(bucket, object, fi, opts).await
let disk = disk.as_ref().unwrap(); } else {
futures.push(disk.update_metadata(bucket, object, fi.clone(), opts)) Err(Error::new(DiskError::DiskNotFound))
}
})
} }
let results = join_all(futures).await; let results = join_all(futures).await;
@@ -3725,12 +3724,14 @@ impl StorageAPI for SetDisks {
// let mut errors = Vec::with_capacity(disks.len()); // let mut errors = Vec::with_capacity(disks.len());
for disk in disks.iter() { for disk in disks.iter() {
if disk.is_none() { let vers = vers.clone();
continue; futures.push(async move {
} if let Some(disk) = disk {
disk.delete_versions(bucket, vers, DeleteOptions::default()).await
let disk = disk.as_ref().unwrap(); } else {
futures.push(disk.delete_versions(bucket, vers.clone(), DeleteOptions::default())); Err(Error::new(DiskError::DiskNotFound))
}
});
} }
let results = join_all(futures).await; let results = join_all(futures).await;
@@ -3963,19 +3964,16 @@ impl StorageAPI for SetDisks {
let erasure = Erasure::new(fi.erasure.data_blocks, fi.erasure.parity_blocks, fi.erasure.block_size); let erasure = Erasure::new(fi.erasure.data_blocks, fi.erasure.parity_blocks, fi.erasure.block_size);
for disk in disks.iter() { for disk in disks.iter() {
if disk.is_none() { if let Some(disk) = disk {
// let writer = disk.append_file(RUSTFS_META_TMP_BUCKET, &tmp_part_path).await?;
let filewriter = disk
.create_file("", RUSTFS_META_TMP_BUCKET, &tmp_part_path, data.content_length)
.await?;
let writer = new_bitrot_filewriter(filewriter, DEFAULT_BITROT_ALGO, erasure.shard_size(erasure.block_size));
writers.push(Some(writer));
} else {
writers.push(None); writers.push(None);
continue;
} }
let disk = disk.as_ref().unwrap().clone();
// let writer = disk.append_file(RUSTFS_META_TMP_BUCKET, &tmp_part_path).await?;
let filewriter = disk
.create_file("", RUSTFS_META_TMP_BUCKET, &tmp_part_path, data.content_length)
.await?;
let writer = new_bitrot_filewriter(filewriter, DEFAULT_BITROT_ALGO, erasure.shard_size(erasure.block_size));
writers.push(Some(writer));
} }
let mut erasure = Erasure::new(fi.erasure.data_blocks, fi.erasure.parity_blocks, fi.erasure.block_size); let mut erasure = Erasure::new(fi.erasure.data_blocks, fi.erasure.parity_blocks, fi.erasure.block_size);
ecstore/src/store_init.rs
+7 -7
View File
@@ -191,13 +191,13 @@ pub async fn load_format_erasure_all(disks: &[Option<DiskStore>], heal: bool) ->
let mut errors = Vec::with_capacity(disks.len()); let mut errors = Vec::with_capacity(disks.len());
for disk in disks.iter() { for disk in disks.iter() {
if disk.is_none() { futures.push(async move {
datas.push(None); if let Some(disk) = disk {
errors.push(Some(Error::new(DiskError::DiskNotFound))); load_format_erasure(disk, heal).await
} } else {
Err(Error::new(DiskError::DiskNotFound))
let disk = disk.as_ref().unwrap(); }
futures.push(load_format_erasure(disk, heal)); });
} }
let results = join_all(futures).await; let results = join_all(futures).await;
scripts/run.sh
+2 -2
View File
@@ -18,8 +18,8 @@ fi
export RUSTFS_STORAGE_CLASS_INLINE_BLOCK="512 KB" export RUSTFS_STORAGE_CLASS_INLINE_BLOCK="512 KB"
# DATA_DIR_ARG="./target/volume/test{0...4}" DATA_DIR_ARG="./target/volume/test{0...4}"
DATA_DIR_ARG="./target/volume/test" # DATA_DIR_ARG="./target/volume/test"
if [ -n "$1" ]; then if [ -n "$1" ]; then
DATA_DIR_ARG="$1" DATA_DIR_ARG="$1"