refactor task id

Author: loongs-zhang

core/src/co_pool/mod.rs

@@ -26,10 +26,10 @@ mod state;
 /// Creator for coroutine pool.
 mod creator;
 
-/// `task_name` -> `co_name`
-static RUNNING_TASKS: Lazy<DashMap<&str, &str>> = Lazy::new(DashMap::new);
+/// `task_id` -> `co_id`
+static RUNNING_TASKS: Lazy<DashMap<u64, u64>> = Lazy::new(DashMap::new);
 
-static CANCEL_TASKS: Lazy<DashSet<&str>> = Lazy::new(DashSet::new);
+static CANCEL_TASKS: Lazy<DashSet<u64>> = Lazy::new(DashSet::new);
 
 /// The coroutine pool impls.
 #[repr(C)]
@@ -55,10 +55,10 @@ pub struct CoroutinePool<'p> {
     //阻滞器
     blocker: Arc<CondvarBlocker>,
     //正在等待结果的
-    waits: DashMap<&'p str, Arc<(Mutex<bool>, Condvar)>>,
+    waits: DashMap<u64, Arc<(Mutex<bool>, Condvar)>>,
     //任务执行结果
-    results: DashMap<String, Result<Option<usize>, &'p str>>,
-    no_waits: DashSet<&'p str>,
+    results: DashMap<u64, Result<Option<usize>, &'p str>>,
+    no_waits: DashSet<u64>,
 }
 
 impl Drop for CoroutinePool<'_> {
@@ -226,11 +226,11 @@ impl<'p> CoroutinePool<'p> {
     fn do_clean(&mut self) {
         // clean up remaining wait tasks
         for r in &self.waits {
-            let task_name = *r.key();
+            let task_id = *r.key();
             _ = self
                 .results
-                .insert(task_name.to_string(), Err("The coroutine pool has stopped"));
-            self.notify(task_name);
+                .insert(task_id, Err("The coroutine pool has stopped"));
+            self.notify(task_id);
         }
     }
 
@@ -244,16 +244,22 @@ impl<'p> CoroutinePool<'p> {
         func: impl FnOnce(Option<usize>) -> Option<usize> + 'p,
         param: Option<usize>,
         priority: Option<c_longlong>,
-    ) -> std::io::Result<String> {
+    ) -> std::io::Result<u64> {
         match self.state() {
             PoolState::Running => {}
             PoolState::Stopping | PoolState::Stopped => {
                 return Err(Error::other("The coroutine pool is stopping or stopped !"))
             }
         }
-        let name = name.unwrap_or(format!("{}@{}", self.name(), uuid::Uuid::new_v4()));
-        self.submit_raw_task(Task::new(name.clone(), func, param, priority));
-        Ok(name)
+        let task = Task::new(
+            name.unwrap_or(format!("{}@{}", self.name(), uuid::Uuid::new_v4())),
+            func,
+            param,
+            priority,
+        );
+        let task_id = task.id();
+        self.submit_raw_task(task);
+        Ok(task_id)
     }
 
     /// Submit new task to this pool.
@@ -266,17 +272,17 @@ impl<'p> CoroutinePool<'p> {
     }
 
     /// Attempt to obtain task results with the given `task_name`.
-    pub fn try_take_task_result(&self, task_name: &str) -> Option<Result<Option<usize>, &'p str>> {
-        self.results.remove(task_name).map(|(_, r)| r)
+    pub fn try_take_task_result(&self, task_id: u64) -> Option<Result<Option<usize>, &'p str>> {
+        self.results.remove(&task_id).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() {
+    pub fn clean_task_result(&self, task_id: u64) {
+        if self.try_take_task_result(task_id).is_some() {
             return;
         }
-        _ = self.no_waits.insert(Box::leak(Box::from(task_name)));
-        _ = CANCEL_TASKS.remove(task_name);
+        _ = self.no_waits.insert(task_id);
+        _ = CANCEL_TASKS.remove(&task_id);
     }
 
     /// Use the given `task_name` to obtain task results, and if no results are found,
@@ -286,31 +292,30 @@ impl<'p> CoroutinePool<'p> {
     /// if timeout
     pub fn wait_task_result(
         &self,
-        task_name: &str,
+        task_id: u64,
         wait_time: Duration,
     ) -> std::io::Result<Result<Option<usize>, &str>> {
-        let key = Box::leak(Box::from(task_name));
-        if let Some(r) = self.try_take_task_result(key) {
-            self.notify(key);
+        if let Some(r) = self.try_take_task_result(task_id) {
+            self.notify(task_id);
             return Ok(r);
         }
         if SchedulableCoroutine::current().is_some() {
             let timeout_time = get_timeout_time(wait_time);
             loop {
                 _ = self.try_run();
-                if let Some(r) = self.try_take_task_result(key) {
+                if let Some(r) = self.try_take_task_result(task_id) {
                     return Ok(r);
                 }
                 if timeout_time.saturating_sub(now()) == 0 {
                     return Err(Error::new(ErrorKind::TimedOut, "wait timeout"));
                 }
             }
         }
-        let arc = if let Some(arc) = self.waits.get(key) {
+        let arc = if let Some(arc) = self.waits.get(&task_id) {
             arc.clone()
         } else {
             let arc = Arc::new((Mutex::new(true), Condvar::new()));
-            assert!(self.waits.insert(key, arc.clone()).is_none());
+            assert!(self.waits.insert(task_id, arc.clone()).is_none());
             arc
         };
         let (lock, cvar) = &*arc;
@@ -322,8 +327,8 @@ impl<'p> CoroutinePool<'p> {
             )
             .map_err(|e| Error::other(format!("{e}")))?,
         );
-        if let Some(r) = self.try_take_task_result(key) {
-            self.notify(key);
+        if let Some(r) = self.try_take_task_result(task_id) {
+            self.notify(task_id);
             return Ok(r);
         }
         Err(Error::new(ErrorKind::TimedOut, "wait timeout"))
@@ -409,32 +414,31 @@ impl<'p> CoroutinePool<'p> {
 
     fn try_run(&self) -> Option<()> {
         self.task_queue.pop().map(|task| {
-            let tname = task.get_name().to_string().leak();
-            if CANCEL_TASKS.contains(tname) {
-                _ = CANCEL_TASKS.remove(tname);
-                warn!("Cancel task:{} successfully !", tname);
+            let task_id = task.id();
+            if CANCEL_TASKS.contains(&task_id) {
+                _ = CANCEL_TASKS.remove(&task_id);
+                warn!("Cancel task:{} successfully !", task_id);
                 return;
             }
             if let Some(co) = SchedulableCoroutine::current() {
-                _ = RUNNING_TASKS.insert(tname, co.name());
+                _ = RUNNING_TASKS.insert(task_id, co.id);
             }
-            let (task_name, result) = task.run();
-            _ = RUNNING_TASKS.remove(tname);
-            let n = task_name.clone().leak();
-            if self.no_waits.contains(n) {
-                _ = self.no_waits.remove(n);
+            let (_, result) = task.run();
+            _ = RUNNING_TASKS.remove(&task_id);
+            if self.no_waits.contains(&task_id) {
+                _ = self.no_waits.remove(&task_id);
                 return;
             }
             assert!(
-                self.results.insert(task_name.clone(), result).is_none(),
+                self.results.insert(task_id, result).is_none(),
                 "The previous result was not retrieved in a timely manner"
             );
-            self.notify(&task_name);
+            self.notify(task_id);
         })
     }
 
-    fn notify(&self, task_name: &str) {
-        if let Some((_, arc)) = self.waits.remove(task_name) {
+    fn notify(&self, task_id: u64) {
+        if let Some((_, arc)) = self.waits.remove(&task_id) {
             let (lock, cvar) = &*arc;
             let mut pending = lock.lock().expect("notify task failed");
             *pending = false;
@@ -443,9 +447,9 @@ impl<'p> CoroutinePool<'p> {
     }
 
     /// Try to cancel a task.
-    pub fn try_cancel_task(task_name: &str) {
+    pub fn try_cancel_task(task_id: u64) {
         // 检查正在运行的任务是否是要取消的任务
-        if let Some(info) = RUNNING_TASKS.get(task_name) {
+        if let Some(info) = RUNNING_TASKS.get(&task_id) {
             let co_name = *info;
             // todo windows support
             #[allow(unused_variables)]
@@ -470,13 +474,13 @@ impl<'p> CoroutinePool<'p> {
                 Scheduler::try_cancel_coroutine(co_name);
                 warn!(
                     "Attempt to cancel task:{} running on coroutine:{}, cancelling...",
-                    task_name, co_name
+                    task_id, co_name
                 );
             }
         } else {
             // 添加到待取消队列
-            _ = CANCEL_TASKS.insert(Box::leak(Box::from(task_name)));
-            warn!("Attempt to cancel task:{}, cancelling...", task_name);
+            _ = CANCEL_TASKS.insert(task_id);
+            warn!("Attempt to cancel task:{}, cancelling...", task_id);
         }
     }
 

core/src/co_pool/task.rs

@@ -1,6 +1,7 @@
 use crate::catch;
 use crate::common::ordered_work_steal::Ordered;
 use std::ffi::c_longlong;
+use std::hash::{DefaultHasher, Hash, Hasher};
 
 /// 做C兼容时会用到
 pub type UserTaskFunc = extern "C" fn(usize) -> usize;
@@ -10,6 +11,7 @@ pub type UserTaskFunc = extern "C" fn(usize) -> usize;
 #[derive(educe::Educe)]
 #[educe(Debug)]
 pub struct Task<'t> {
+    id: u64,
     name: String,
     #[educe(Debug(ignore))]
     func: Box<dyn FnOnce(Option<usize>) -> Option<usize> + 't>,
@@ -25,7 +27,11 @@ impl<'t> Task<'t> {
         param: Option<usize>,
         priority: Option<c_longlong>,
     ) -> Self {
+        let mut hasher = DefaultHasher::new();
+        name.hash(&mut hasher);
+        let id = hasher.finish();
         Task {
+            id,
             name,
             func: Box::new(func),
             param,
@@ -35,10 +41,16 @@ impl<'t> Task<'t> {
 
     /// get the task name.
     #[must_use]
-    pub fn get_name(&self) -> &str {
+    pub fn name(&self) -> &str {
         &self.name
     }
 
+    /// get the task name.
+    #[must_use]
+    pub fn id(&self) -> u64 {
+        self.id
+    }
+
     /// execute the task
     ///
     /// # Errors

core/src/coroutine/mod.rs

@@ -67,6 +67,7 @@ impl<'c, Param, Yield, Return> Coroutine<'c, Param, Yield, Return> {
     }
 
     /// Get the id of this coroutine.
+    /// todo use u64
     #[allow(clippy::cast_possible_truncation)]
     pub fn id(&self) -> usize {
         self.id as usize

core/src/net/event_loop.rs

@@ -139,8 +139,8 @@ impl<'e> EventLoop<'e> {
     }
 
     /// Try to cancel a task from `CoroutinePool`.
-    pub(super) fn try_cancel_task(name: &str) {
-        CoroutinePool::try_cancel_task(name);
+    pub(super) fn try_cancel_task(task_id: u64) {
+        CoroutinePool::try_cancel_task(task_id);
     }
 
     #[allow(trivial_numeric_casts, clippy::cast_possible_truncation)]
@@ -350,7 +350,7 @@ impl<'e> EventLoop<'e> {
         if COROUTINE_TOKENS.remove(&token).is_none() {
             return;
         }
-        self.try_resume(token);
+        self.try_resume(token as u64);
     }
 
     pub(super) fn start(self) -> std::io::Result<Arc<Self>>

core/src/net/join.rs

@@ -1,45 +1,41 @@
 use crate::net::event_loop::EventLoop;
-use std::ffi::{c_char, CStr, CString};
 use std::io::{Error, ErrorKind};
 use std::sync::Arc;
 use std::time::Duration;
 
 #[allow(missing_docs)]
 #[repr(C)]
 #[derive(Debug)]
-pub struct JoinHandle(&'static Arc<EventLoop<'static>>, *const c_char);
+pub struct JoinHandle(&'static Arc<EventLoop<'static>>, u64);
 
 impl Drop for JoinHandle {
     fn drop(&mut self) {
-        if let Ok(name) = self.get_name() {
-            self.0.clean_task_result(name);
+        if let Ok(task_id) = self.id() {
+            self.0.clean_task_result(task_id);
         }
     }
 }
 
 impl JoinHandle {
     /// create `JoinHandle` instance.
     pub(crate) fn err(pool: &'static Arc<EventLoop<'static>>) -> Self {
-        Self::new(pool, "")
+        Self::new(pool, 0)
     }
 
     /// create `JoinHandle` instance.
-    pub(crate) fn new(pool: &'static Arc<EventLoop<'static>>, name: &str) -> Self {
-        let boxed: &'static mut CString = Box::leak(Box::from(
-            CString::new(name).expect("init JoinHandle failed!"),
-        ));
-        let cstr: &'static CStr = boxed.as_c_str();
-        JoinHandle(pool, cstr.as_ptr())
+    pub(crate) fn new(pool: &'static Arc<EventLoop<'static>>, task_id: u64) -> Self {
+        JoinHandle(pool, task_id)
     }
 
-    /// get the task name.
+    /// get the task id.
     ///
     /// # Errors
-    /// if the task name is invalid.
-    pub fn get_name(&self) -> std::io::Result<&str> {
-        unsafe { CStr::from_ptr(self.1) }
-            .to_str()
-            .map_err(|_| Error::new(ErrorKind::InvalidInput, "Invalid task name"))
+    /// if the task id is invalid.
+    pub fn id(&self) -> std::io::Result<u64> {
+        if 0 == self.1 {
+            return Err(Error::new(ErrorKind::InvalidInput, "Invalid task id"));
+        }
+        Ok(self.1)
     }
 
     /// join with `Duration`.
@@ -66,12 +62,9 @@ impl JoinHandle {
         &self,
         timeout_time: u64,
     ) -> std::io::Result<Result<Option<usize>, &str>> {
-        let name = self.get_name()?;
-        if name.is_empty() {
-            return Err(Error::new(ErrorKind::InvalidInput, "Invalid task name"));
-        }
+        let task_id = self.id()?;
         self.0.wait_task_result(
-            name,
+            task_id,
             Duration::from_nanos(timeout_time.saturating_sub(crate::common::now())),
         )
     }

core/src/net/mod.rs

@@ -150,13 +150,13 @@ impl EventLoops {
             .submit_task(name, func, param, priority)
             .map_or_else(
                 |_| JoinHandle::err(event_loop),
-                |n| JoinHandle::new(event_loop, n.as_str()),
+                |task_id| JoinHandle::new(event_loop, task_id),
             )
     }
 
     /// Try to cancel a task from event-loop.
-    pub fn try_cancel_task(name: &str) {
-        EventLoop::try_cancel_task(name);
+    pub fn try_cancel_task(task_id: u64) {
+        EventLoop::try_cancel_task(task_id);
     }
 
     /// Submit a new coroutine to event-loop.