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..bd12d9285 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(); // 通知一个等待的任务 } @@ -51,6 +57,7 @@ impl Workers { // 等待直到所有槽都被释放 self.notify.notified().await; } + info!("worker wait end"); } pub async fn available(&self) -> usize { @@ -68,15 +75,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 ad35d8126..1f5535651 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>; @@ -57,6 +54,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")); @@ -64,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(); @@ -100,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; + } } } @@ -173,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 @@ -204,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; @@ -212,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; } @@ -229,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; } @@ -291,10 +283,11 @@ 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; } - break; + // break; } Ok(()) }); diff --git a/ecstore/src/heal/background_heal_ops.rs b/ecstore/src/heal/background_heal_ops.rs index 91156bee8..0082bc3f3 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,13 +119,15 @@ 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; 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; @@ -133,6 +136,7 @@ async fn monitor_local_disks_and_heal() { GLOBAL_BackgroundHealState.pop_heal_local_disks(&[disk_clone]).await; }); } + let _ = join_all(futures).await; interval.reset(); } } diff --git a/ecstore/src/set_disk.rs b/ecstore/src/set_disk.rs index c9088a7a6..0ebb73281 100644 --- a/ecstore/src/set_disk.rs +++ b/ecstore/src/set_disk.rs @@ -1349,6 +1349,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; @@ -3485,6 +3486,7 @@ impl SetDisks { tracker.read().await.queue_buckets ))); } + drop(result_tx); let _ = task.await; defer.await; Ok(())