From 57f0370ab802afacbad1a7614f7b4e5f929d0d63 Mon Sep 17 00:00:00 2001 From: junxiang Mu <1948535941@qq.com> Date: Thu, 26 Dec 2024 16:25:13 +0800 Subject: [PATCH 1/6] tmp1 Signed-off-by: junxiang Mu <1948535941@qq.com> --- ecstore/src/heal/background_heal_ops.rs | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/ecstore/src/heal/background_heal_ops.rs b/ecstore/src/heal/background_heal_ops.rs index 7cd54ee34..74821f9f6 100644 --- a/ecstore/src/heal/background_heal_ops.rs +++ b/ecstore/src/heal/background_heal_ops.rs @@ -1,3 +1,4 @@ +use futures::future::join_all; use madmin::heal_commands::HealResultItem; use std::{cmp::Ordering, env, path::PathBuf, sync::Arc, time::Duration}; use tokio::{ @@ -118,9 +119,10 @@ async fn monitor_local_disks_and_heal() { } } + let mut futures = Vec::new(); for disk in heal_disks.into_ref().iter() { let disk_clone = disk.clone(); - spawn(async move { + futures.push(async move { GLOBAL_BackgroundHealState .set_disk_healing_status(disk_clone.clone(), true) .await; @@ -133,6 +135,7 @@ async fn monitor_local_disks_and_heal() { GLOBAL_BackgroundHealState.pop_heal_local_disks(&[disk_clone]).await; }); } + let _ = join_all(futures).await; interval.reset(); } } From 874679e65dd29051bf8cc47c866c97e20f084ff2 Mon Sep 17 00:00:00 2001 From: junxiang Mu <1948535941@qq.com> Date: Thu, 26 Dec 2024 20:56:21 +0800 Subject: [PATCH 2/6] tmp 2 Signed-off-by: junxiang Mu <1948535941@qq.com> --- ecstore/src/cache_value/metacache_set.rs | 1 + ecstore/src/heal/background_heal_ops.rs | 10 ++++++++++ ecstore/src/set_disk.rs | 6 ++++++ 3 files changed, 17 insertions(+) diff --git a/ecstore/src/cache_value/metacache_set.rs b/ecstore/src/cache_value/metacache_set.rs index ad35d8126..098724bd8 100644 --- a/ecstore/src/cache_value/metacache_set.rs +++ b/ecstore/src/cache_value/metacache_set.rs @@ -57,6 +57,7 @@ impl Clone for ListPathRawOptions { } pub async fn list_path_raw(mut rx: B_Receiver, opts: ListPathRawOptions) -> Result<()> { + info!("list_path_raw"); if opts.disks.is_empty() { info!("list_path_raw 0 drives provided"); return Err(Error::from_string("list_path_raw: 0 drives provided")); diff --git a/ecstore/src/heal/background_heal_ops.rs b/ecstore/src/heal/background_heal_ops.rs index 74821f9f6..27ee28417 100644 --- a/ecstore/src/heal/background_heal_ops.rs +++ b/ecstore/src/heal/background_heal_ops.rs @@ -123,19 +123,26 @@ async fn monitor_local_disks_and_heal() { for disk in heal_disks.into_ref().iter() { let disk_clone = disk.clone(); futures.push(async move { + info!("1"); GLOBAL_BackgroundHealState .set_disk_healing_status(disk_clone.clone(), true) .await; + info!("2"); if heal_fresh_disk(&disk_clone).await.is_err() { + info!("heal_fresh_disk is err"); GLOBAL_BackgroundHealState .set_disk_healing_status(disk_clone.clone(), false) .await; return; } + info!("3"); GLOBAL_BackgroundHealState.pop_heal_local_disks(&[disk_clone]).await; + info!("4"); }); } + info!("monitor_local_disks_and_heal wait"); let _ = join_all(futures).await; + info!("monitor_local_disks_and_heal end"); interval.reset(); } } @@ -226,11 +233,14 @@ async fn heal_fresh_disk(endpoint: &Endpoint) -> Result<()> { let tracker = Arc::new(RwLock::new(tracker)); let qb = tracker.read().await.queue_buckets.clone(); + info!("0.4"); store.pools[pool_idx].disk_set[set_idx] .clone() .heal_erasure_set(&qb, tracker.clone()) .await?; + info!("4"); let mut tracker_w = tracker.write().await; + info!("5"); if tracker_w.items_failed > 0 && tracker_w.retry_attempts < 4 { tracker_w.retry_attempts += 1; tracker_w.reset_healing().await; diff --git a/ecstore/src/set_disk.rs b/ecstore/src/set_disk.rs index 73c646249..04617f736 100644 --- a/ecstore/src/set_disk.rs +++ b/ecstore/src/set_disk.rs @@ -1348,6 +1348,7 @@ impl SetDisks { Err(e) => { warn!("connect_endpoint err {:?}", &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; } return; @@ -3478,14 +3479,19 @@ impl SetDisks { if let Some(err) = ret_err.as_ref() { return Err(err.clone()); } + info!("step 6"); if !tracker.read().await.queue_buckets.is_empty() { return Err(Error::from_string(format!( "not all buckets were healed: {:?}", tracker.read().await.queue_buckets ))); } + info!("step 7"); + drop(result_tx); let _ = task.await; + info!("step 8"); defer.await; + info!("step 9"); Ok(()) } } From dc3154b883a359864264e69c6fb4073edfc166f3 Mon Sep 17 00:00:00 2001 From: junxiang Mu <1948535941@qq.com> Date: Sun, 29 Dec 2024 18:19:46 +0800 Subject: [PATCH 3/6] tmp 3 Signed-off-by: junxiang Mu <1948535941@qq.com> --- ecstore/src/cache_value/metacache_set.rs | 78 +++++++++++------------- 1 file changed, 34 insertions(+), 44 deletions(-) diff --git a/ecstore/src/cache_value/metacache_set.rs b/ecstore/src/cache_value/metacache_set.rs index 098724bd8..a2535ce2b 100644 --- a/ecstore/src/cache_value/metacache_set.rs +++ b/ecstore/src/cache_value/metacache_set.rs @@ -8,10 +8,7 @@ use crate::{ }; use futures::future::join_all; use std::{future::Future, pin::Pin, sync::Arc}; -use tokio::{ - spawn, - sync::{broadcast::Receiver as B_Receiver, RwLock}, -}; +use tokio::{spawn, sync::broadcast::Receiver as B_Receiver}; use tracing::{error, info}; type AgreedFn = Box Pin + Send>> + Send + 'static>; @@ -65,7 +62,7 @@ pub async fn list_path_raw(mut rx: B_Receiver, opts: ListPathRawOptions) - let mut jobs: Vec>> = Vec::new(); let mut readers = Vec::with_capacity(opts.disks.len()); - let fds = Arc::new(RwLock::new(opts.fallback_disks.clone())); + let fds = Arc::new(opts.fallback_disks.clone()); for disk in opts.disks.iter() { let opdisk = disk.clone(); @@ -101,47 +98,40 @@ pub async fn list_path_raw(mut rx: B_Receiver, opts: ListPathRawOptions) - } while need_fallback { - let f_disk = loop { - let mut fds_w = fds_clone.write().await; - if fds_w.is_empty() { - break None; - } - - if let Some(fd) = fds_w.remove(0) { - if fd.is_online().await { - break Some(fd); - } - } - }; - - if let Some(disk) = f_disk { - match disk - .as_ref() - .walk_dir( - WalkDirOptions { - bucket: opts_clone.bucket.clone(), - base_dir: opts_clone.path.clone(), - recursive: opts_clone.recursice, - report_notfound: opts_clone.report_not_found, - filter_prefix: opts_clone.filter_prefix.clone(), - forward_to: opts_clone.forward_to.clone(), - limit: opts_clone.per_disk_limit, - ..Default::default() - }, - &mut wr, - ) - .await - { - Ok(_r) => { - need_fallback = false; - } - Err(err) => { - error!("walk dir2 err {:?}", &err); + let disk = match fds_clone.iter().find(|d| d.is_some()) { + Some(d) => { + if let Some(disk) = d.clone() { + disk + } else { break; } } - } else { - break; + None => break, + }; + match disk + .as_ref() + .walk_dir( + WalkDirOptions { + bucket: opts_clone.bucket.clone(), + base_dir: opts_clone.path.clone(), + recursive: opts_clone.recursice, + report_notfound: opts_clone.report_not_found, + filter_prefix: opts_clone.filter_prefix.clone(), + forward_to: opts_clone.forward_to.clone(), + limit: opts_clone.per_disk_limit, + ..Default::default() + }, + &mut wr, + ) + .await + { + Ok(_r) => { + need_fallback = false; + } + Err(err) => { + error!("walk dir2 err {:?}", &err); + break; + } } } @@ -295,7 +285,7 @@ pub async fn list_path_raw(mut rx: B_Receiver, opts: ListPathRawOptions) - if let Some(partial_fn) = opts.partial.as_ref() { partial_fn(MetaCacheEntries(top_entries), &errs).await; } - break; + // break; } Ok(()) }); From 2f5f352f9a450dacfd8e12eb17ea9b4f296b7cfb Mon Sep 17 00:00:00 2001 From: junxiang Mu <1948535941@qq.com> Date: Sun, 29 Dec 2024 20:25:23 +0800 Subject: [PATCH 4/6] tmp 4 Signed-off-by: junxiang Mu <1948535941@qq.com> --- Cargo.lock | 1 + common/workers/Cargo.toml | 3 ++- common/workers/src/workers.rs | 24 ++++++++++++++++-------- ecstore/src/cache_value/metacache_set.rs | 12 +++++++----- 4 files changed, 26 insertions(+), 14 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 0916ae0fe..b90be2160 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3561,6 +3561,7 @@ version = "0.0.1" dependencies = [ "common", "tokio", + "tracing", ] [[package]] diff --git a/common/workers/Cargo.toml b/common/workers/Cargo.toml index fa9186886..f5b6f25fc 100644 --- a/common/workers/Cargo.toml +++ b/common/workers/Cargo.toml @@ -11,4 +11,5 @@ workspace = true [dependencies] common.workspace = true -tokio.workspace = true \ No newline at end of file +tokio.workspace = true +tracing.workspace = true diff --git a/common/workers/src/workers.rs b/common/workers/src/workers.rs index 85be5efbc..e33ebd757 100644 --- a/common/workers/src/workers.rs +++ b/common/workers/src/workers.rs @@ -1,5 +1,6 @@ use std::sync::Arc; use tokio::sync::{Mutex, Notify}; +use tracing::info; pub struct Workers { available: Mutex, // 可用的工作槽 @@ -23,18 +24,23 @@ impl Workers { // 让一个作业获得执行的机会 pub async fn take(&self) { - let mut available = self.available.lock().await; - while *available == 0 { - // 等待直到有可用槽 - self.notify.notified().await; - available = self.available.lock().await; + loop { + let mut available = self.available.lock().await; + info!("worker take, {}", *available); + if *available == 0 { + drop(available); + self.notify.notified().await; + } else { + *available -= 1; + break; + } } - *available -= 1; // 减少可用槽 } // 让一个作业释放其机会 pub async fn give(&self) { let mut available = self.available.lock().await; + info!("worker give, {}", *available); *available += 1; // 增加可用槽 self.notify.notify_one(); // 通知一个等待的任务 } @@ -68,15 +74,17 @@ mod tests { async fn test_workers() { let workers = Arc::new(Workers::new(5).unwrap()); - for _ in 0..4 { + for _ in 0..5 { let workers = workers.clone(); tokio::spawn(async move { workers.take().await; sleep(Duration::from_secs(3)).await; - workers.give().await; }); } + for _ in 0..5 { + workers.give().await; + } // Sleep: wait for spawn task started sleep(Duration::from_secs(1)).await; workers.wait().await; diff --git a/ecstore/src/cache_value/metacache_set.rs b/ecstore/src/cache_value/metacache_set.rs index a2535ce2b..1f5535651 100644 --- a/ecstore/src/cache_value/metacache_set.rs +++ b/ecstore/src/cache_value/metacache_set.rs @@ -164,6 +164,7 @@ pub async fn list_path_raw(mut rx: B_Receiver, opts: ListPathRawOptions) - let entry = match r.peek().await { Ok(res) => { if let Some(entry) = res { + info!("read entry disk: {}, name: {}", i, entry.name); entry } else { // eof @@ -195,7 +196,7 @@ pub async fn list_path_raw(mut rx: B_Receiver, opts: ListPathRawOptions) - // If no current, add it. if current.name.is_empty() { - top_entries.insert(i, Some(entry.clone())); + top_entries[i] = Some(entry.clone()); current = entry; agree += 1; @@ -203,14 +204,14 @@ pub async fn list_path_raw(mut rx: B_Receiver, opts: ListPathRawOptions) - } // If exact match, we agree. if let Ok((_, true)) = current.matches(&entry, true) { - top_entries.insert(i, Some(entry)); + top_entries[i] = Some(entry); agree += 1; continue; } // If only the name matches we didn't agree, but add it for resolution. if entry.name == current.name { - top_entries.insert(i, Some(entry)); + top_entries[i] = Some(entry); continue; } @@ -220,9 +221,9 @@ pub async fn list_path_raw(mut rx: B_Receiver, opts: ListPathRawOptions) - } // We got a new, better current. // Clear existing entries. - top_entries.clear(); + top_entries = vec![None; top_entries.len()]; agree += 1; - top_entries.insert(i, Some(entry.clone())); + top_entries[i] = Some(entry.clone()); current = entry; } @@ -282,6 +283,7 @@ pub async fn list_path_raw(mut rx: B_Receiver, opts: ListPathRawOptions) - } } + info!("read entry should heal: {}", current.name); if let Some(partial_fn) = opts.partial.as_ref() { partial_fn(MetaCacheEntries(top_entries), &errs).await; } From a55b45c0ee8912e7d622a7b9e907eabccf75a40a Mon Sep 17 00:00:00 2001 From: junxiang Mu <1948535941@qq.com> Date: Sun, 29 Dec 2024 20:27:05 +0800 Subject: [PATCH 5/6] tmp 5 Signed-off-by: junxiang Mu <1948535941@qq.com> --- common/workers/src/workers.rs | 1 + 1 file changed, 1 insertion(+) diff --git a/common/workers/src/workers.rs b/common/workers/src/workers.rs index e33ebd757..bd12d9285 100644 --- a/common/workers/src/workers.rs +++ b/common/workers/src/workers.rs @@ -57,6 +57,7 @@ impl Workers { // 等待直到所有槽都被释放 self.notify.notified().await; } + info!("worker wait end"); } pub async fn available(&self) -> usize { From f52882ede6c488b0d021fb388a3902d0a40b3093 Mon Sep 17 00:00:00 2001 From: junxiang Mu <1948535941@qq.com> Date: Sun, 29 Dec 2024 20:34:39 +0800 Subject: [PATCH 6/6] remove debug info Signed-off-by: junxiang Mu <1948535941@qq.com> --- ecstore/src/heal/background_heal_ops.rs | 9 --------- ecstore/src/set_disk.rs | 4 ---- 2 files changed, 13 deletions(-) diff --git a/ecstore/src/heal/background_heal_ops.rs b/ecstore/src/heal/background_heal_ops.rs index 27ee28417..5ce1430ac 100644 --- a/ecstore/src/heal/background_heal_ops.rs +++ b/ecstore/src/heal/background_heal_ops.rs @@ -123,11 +123,9 @@ async fn monitor_local_disks_and_heal() { for disk in heal_disks.into_ref().iter() { let disk_clone = disk.clone(); futures.push(async move { - info!("1"); GLOBAL_BackgroundHealState .set_disk_healing_status(disk_clone.clone(), true) .await; - info!("2"); if heal_fresh_disk(&disk_clone).await.is_err() { info!("heal_fresh_disk is err"); GLOBAL_BackgroundHealState @@ -135,14 +133,10 @@ async fn monitor_local_disks_and_heal() { .await; return; } - info!("3"); GLOBAL_BackgroundHealState.pop_heal_local_disks(&[disk_clone]).await; - info!("4"); }); } - info!("monitor_local_disks_and_heal wait"); let _ = join_all(futures).await; - info!("monitor_local_disks_and_heal end"); interval.reset(); } } @@ -233,14 +227,11 @@ async fn heal_fresh_disk(endpoint: &Endpoint) -> Result<()> { let tracker = Arc::new(RwLock::new(tracker)); let qb = tracker.read().await.queue_buckets.clone(); - info!("0.4"); store.pools[pool_idx].disk_set[set_idx] .clone() .heal_erasure_set(&qb, tracker.clone()) .await?; - info!("4"); let mut tracker_w = tracker.write().await; - info!("5"); if tracker_w.items_failed > 0 && tracker_w.retry_attempts < 4 { tracker_w.retry_attempts += 1; tracker_w.reset_healing().await; diff --git a/ecstore/src/set_disk.rs b/ecstore/src/set_disk.rs index 04617f736..fab2458f2 100644 --- a/ecstore/src/set_disk.rs +++ b/ecstore/src/set_disk.rs @@ -3479,19 +3479,15 @@ impl SetDisks { if let Some(err) = ret_err.as_ref() { return Err(err.clone()); } - info!("step 6"); if !tracker.read().await.queue_buckets.is_empty() { return Err(Error::from_string(format!( "not all buckets were healed: {:?}", tracker.read().await.queue_buckets ))); } - info!("step 7"); drop(result_tx); let _ = task.await; - info!("step 8"); defer.await; - info!("step 9"); Ok(()) } }