diff --git a/core/src/co_pool/mod.rs b/core/src/co_pool/mod.rs index 5a784c82..c63edcc6 100644 --- a/core/src/co_pool/mod.rs +++ b/core/src/co_pool/mod.rs @@ -7,7 +7,7 @@ use crate::common::{get_timeout_time, now, CondvarBlocker}; use crate::coroutine::suspender::Suspender; use crate::scheduler::{SchedulableCoroutine, Scheduler}; use crate::{error, impl_current_for, impl_display_by_debug, impl_for_named, trace}; -use dashmap::DashMap; +use dashmap::{DashMap, DashSet}; use std::cell::Cell; use std::ffi::c_longlong; use std::io::{Error, ErrorKind}; @@ -52,6 +52,7 @@ pub struct CoroutinePool<'p> { waits: DashMap<&'p str, Arc<(Mutex, Condvar)>>, //任务执行结果 results: DashMap, &'p str>>, + no_waits: DashSet<&'p str>, } impl Drop for CoroutinePool<'_> { @@ -136,6 +137,7 @@ impl<'p> CoroutinePool<'p> { blocker: Arc::default(), results: DashMap::new(), waits: DashMap::default(), + no_waits: DashSet::default(), } } @@ -240,6 +242,14 @@ impl<'p> CoroutinePool<'p> { self.results.remove(task_name).map(|(_, r)| r) } + /// clean the task result data. + pub fn clean_task_result(&self, task_name: &str) { + if self.try_take_task_result(task_name).is_some() { + return; + } + _ = self.no_waits.insert(Box::leak(Box::from(task_name))); + } + /// Use the given `task_name` to obtain task results, and if no results are found, /// block the current thread for `wait_time`. /// @@ -375,6 +385,11 @@ impl<'p> CoroutinePool<'p> { fn try_run(&self) -> Option<()> { self.task_queue.pop().map(|task| { let (task_name, result) = task.run(); + let n = task_name.clone().leak(); + if self.no_waits.contains(n) { + _ = self.no_waits.remove(n); + return; + } assert!( self.results.insert(task_name.clone(), result).is_none(), "The previous result was not retrieved in a timely manner" diff --git a/core/src/net/join.rs b/core/src/net/join.rs index 2323e65f..0966f15f 100644 --- a/core/src/net/join.rs +++ b/core/src/net/join.rs @@ -12,8 +12,7 @@ pub struct JoinHandle(&'static Arc>, *const c_char); impl Drop for JoinHandle { fn drop(&mut self) { if let Ok(name) = self.get_name() { - // clean data - _ = self.0.try_take_task_result(name); + self.0.clean_task_result(name); } } }