gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-08-25 05:26:50 +00:00
Code Issues Packages Projects Releases Wiki Activity

tmp 2

Browse Source 
Signed-off-by: junxiang Mu <1948535941@qq.com>
This commit is contained in:
junxiang Mu
2024-12-26 20:56:21 +08:00
parent 57f0370ab8
commit 874679e65d
3 changed files with 17 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
ecstore/src/cache_value/metacache_set.rs
+1
View File
@@ -57,6 +57,7 @@ impl Clone for ListPathRawOptions {
} }
pub async fn list_path_raw(mut rx: B_Receiver<bool>, opts: ListPathRawOptions) -> Result<()> { pub async fn list_path_raw(mut rx: B_Receiver<bool>, opts: ListPathRawOptions) -> Result<()> {
info!("list_path_raw");
if opts.disks.is_empty() { if opts.disks.is_empty() {
info!("list_path_raw 0 drives provided"); info!("list_path_raw 0 drives provided");
return Err(Error::from_string("list_path_raw: 0 drives provided")); return Err(Error::from_string("list_path_raw: 0 drives provided"));
ecstore/src/heal/background_heal_ops.rs
+10
View File
@@ -123,19 +123,26 @@ async fn monitor_local_disks_and_heal() {
for disk in heal_disks.into_ref().iter() { for disk in heal_disks.into_ref().iter() {
let disk_clone = disk.clone(); let disk_clone = disk.clone();
futures.push(async move { futures.push(async move {
info!("1");
GLOBAL_BackgroundHealState GLOBAL_BackgroundHealState
.set_disk_healing_status(disk_clone.clone(), true) .set_disk_healing_status(disk_clone.clone(), true)
.await; .await;
info!("2");
if heal_fresh_disk(&disk_clone).await.is_err() { if heal_fresh_disk(&disk_clone).await.is_err() {
info!("heal_fresh_disk is err");
GLOBAL_BackgroundHealState GLOBAL_BackgroundHealState
.set_disk_healing_status(disk_clone.clone(), false) .set_disk_healing_status(disk_clone.clone(), false)
.await; .await;
return; return;
} }
info!("3");
GLOBAL_BackgroundHealState.pop_heal_local_disks(&[disk_clone]).await; GLOBAL_BackgroundHealState.pop_heal_local_disks(&[disk_clone]).await;
info!("4");
}); });
} }
info!("monitor_local_disks_and_heal wait");
let _ = join_all(futures).await; let _ = join_all(futures).await;
info!("monitor_local_disks_and_heal end");
interval.reset(); interval.reset();
} }
} }
@@ -226,11 +233,14 @@ async fn heal_fresh_disk(endpoint: &Endpoint) -> Result<()> {
let tracker = Arc::new(RwLock::new(tracker)); let tracker = Arc::new(RwLock::new(tracker));
let qb = tracker.read().await.queue_buckets.clone(); let qb = tracker.read().await.queue_buckets.clone();
info!("0.4");
store.pools[pool_idx].disk_set[set_idx] store.pools[pool_idx].disk_set[set_idx]
.clone() .clone()
.heal_erasure_set(&qb, tracker.clone()) .heal_erasure_set(&qb, tracker.clone())
.await?; .await?;
info!("4");
let mut tracker_w = tracker.write().await; let mut tracker_w = tracker.write().await;
info!("5");
if tracker_w.items_failed > 0 && tracker_w.retry_attempts < 4 { if tracker_w.items_failed > 0 && tracker_w.retry_attempts < 4 {
tracker_w.retry_attempts += 1; tracker_w.retry_attempts += 1;
tracker_w.reset_healing().await; tracker_w.reset_healing().await;
ecstore/src/set_disk.rs
+6
View File
@@ -1348,6 +1348,7 @@ impl SetDisks {
Err(e) => { Err(e) => {
warn!("connect_endpoint err {:?}", &e); warn!("connect_endpoint err {:?}", &e);
if ep.is_local && DiskError::UnformattedDisk.is(&e) { if ep.is_local && DiskError::UnformattedDisk.is(&e) {
info!("unformatteddisk will push_heal_local_disks, {:?}", ep);
GLOBAL_BackgroundHealState.push_heal_local_disks(&[ep.clone()]).await; GLOBAL_BackgroundHealState.push_heal_local_disks(&[ep.clone()]).await;
} }
return; return;
@@ -3478,14 +3479,19 @@ impl SetDisks {
if let Some(err) = ret_err.as_ref() { if let Some(err) = ret_err.as_ref() {
return Err(err.clone()); return Err(err.clone());
} }
info!("step 6");
if !tracker.read().await.queue_buckets.is_empty() { if !tracker.read().await.queue_buckets.is_empty() {
return Err(Error::from_string(format!( return Err(Error::from_string(format!(
"not all buckets were healed: {:?}", "not all buckets were healed: {:?}",
tracker.read().await.queue_buckets tracker.read().await.queue_buckets
))); )));
} }
info!("step 7");
drop(result_tx);
let _ = task.await; let _ = task.await;
info!("step 8");
defer.await; defer.await;
info!("step 9");
Ok(()) Ok(())
} }
} }