refactor co id

Author: loongs-zhang

core/src/co_pool/mod.rs

@@ -188,7 +188,7 @@ impl<'p> CoroutinePool<'p> {
 
     /// Returns `true` if the task queue is empty.
     pub fn is_empty(&self) -> bool {
-        self.size() == 0
+        self.task_queue.is_empty()
     }
 
     /// Returns the number of tasks owned by this pool.
@@ -210,7 +210,14 @@ impl<'p> CoroutinePool<'p> {
     }
 
     fn do_stop(&mut self, dur: Duration) -> std::io::Result<()> {
-        _ = self.try_timed_schedule_task(dur)?;
+        let timeout_time = get_timeout_time(dur);
+        loop {
+            _ = self.try_timeout_schedule_task(timeout_time)?;
+            if self.get_running_size() == 0 || timeout_time.saturating_sub(now()) == 0 {
+                break;
+            }
+            std::thread::sleep(Duration::from_millis(1));
+        }
         assert_eq!(PoolState::Stopping, self.stopped()?);
         self.do_clean();
         Ok(())

core/src/coroutine/korosensei.rs

@@ -11,6 +11,7 @@ use std::cell::{Cell, RefCell, UnsafeCell};
 use std::collections::VecDeque;
 use std::ffi::c_longlong;
 use std::fmt::Debug;
+use std::hash::{DefaultHasher, Hash, Hasher};
 use std::io::Error;
 
 cfg_if::cfg_if! {
@@ -25,6 +26,7 @@ cfg_if::cfg_if! {
 /// Use `corosensei` as the low-level coroutine.
 #[repr(C)]
 pub struct Coroutine<'c, Param, Yield, Return> {
+    pub(crate) id: u64,
     pub(crate) name: String,
     inner: corosensei::Coroutine<Param, Yield, Result<Return, &'static str>, DefaultStack>,
     pub(crate) state: Cell<CoroutineState<Yield, Return>>,
@@ -427,8 +429,12 @@ where
                 co_name
             )
         });
+        let mut hasher = DefaultHasher::new();
+        name.hash(&mut hasher);
+        let id = hasher.finish();
         #[allow(unused_mut)]
         let mut co = Coroutine {
+            id,
             name,
             inner,
             stack_infos,

core/src/coroutine/mod.rs

@@ -66,6 +66,12 @@ impl<'c, Param, Yield, Return> Coroutine<'c, Param, Yield, Return> {
         &self.name
     }
 
+    /// Get the id of this coroutine.
+    #[allow(clippy::cast_possible_truncation)]
+    pub fn id(&self) -> usize {
+        self.id as usize
+    }
+
     /// Returns the current state of this `StateCoroutine`.
     pub fn state(&self) -> CoroutineState<Yield, Return>
     where

core/src/net/event_loop.rs

@@ -7,7 +7,9 @@ use crate::{error, impl_current_for, impl_display_by_debug, info};
 use dashmap::DashSet;
 use once_cell::sync::Lazy;
 use rand::RngExt;
-use std::ffi::{c_char, c_int, c_void, CStr, CString};
+use std::collections::hash_map::DefaultHasher;
+use std::ffi::c_int;
+use std::hash::{Hash, Hasher};
 use std::io::{Error, ErrorKind};
 use std::marker::PhantomData;
 use std::ops::{Deref, DerefMut};
@@ -143,15 +145,11 @@ impl<'e> EventLoop<'e> {
 
     #[allow(trivial_numeric_casts, clippy::cast_possible_truncation)]
     fn token(syscall: SyscallName) -> usize {
+        // Coroutine path: consistent hash of (coroutine_name).
         if let Some(co) = SchedulableCoroutine::current() {
-            let boxed: &'static mut CString = Box::leak(Box::from(
-                CString::new(co.name()).expect("build name failed!"),
-            ));
-            let cstr: &'static CStr = boxed.as_c_str();
-            let token = cstr.as_ptr().cast::<c_void>() as usize;
-            assert!(COROUTINE_TOKENS.insert(token));
-            return token;
+            return co.id();
         }
+        // Thread path: consistent hash of (thread_id, syscall_name).
         unsafe {
             cfg_if::cfg_if! {
                 if #[cfg(windows)] {
@@ -160,8 +158,10 @@ impl<'e> EventLoop<'e> {
                     let thread_id = libc::pthread_self();
                 }
             }
-            let syscall_mask = <SyscallName as Into<&str>>::into(syscall).as_ptr() as usize;
-            let token = thread_id as usize ^ syscall_mask;
+            let mut hasher = DefaultHasher::new();
+            (thread_id as usize).hash(&mut hasher);
+            std::mem::discriminant(&syscall).hash(&mut hasher);
+            let token = hasher.finish() as usize;
             if SyscallName::nio() != syscall {
                 eprintln!("generate token:{token} for {syscall}");
             }
@@ -350,9 +350,7 @@ impl<'e> EventLoop<'e> {
         if COROUTINE_TOKENS.remove(&token).is_none() {
             return;
         }
-        if let Ok(co_name) = CStr::from_ptr((token as *const c_void).cast::<c_char>()).to_str() {
-            self.try_resume(co_name);
-        }
+        self.try_resume(token);
     }
 
     pub(super) fn start(self) -> std::io::Result<Arc<Self>>

core/src/net/selector/mod.rs

@@ -86,7 +86,7 @@ pub(crate) trait Selector<I: Interest, E: Event, S: EventIterator<E>> {
             //同时对读写事件感兴趣
             let interests = I::read_and_write(token);
             self.reregister(fd, token, interests)
-                .or(self.register(fd, token, interests))
+                .or_else(|_| self.register(fd, token, interests))
         } else {
             self.register(fd, token, I::read(token))
         }?;
@@ -105,7 +105,7 @@ pub(crate) trait Selector<I: Interest, E: Event, S: EventIterator<E>> {
             //同时对读写事件感兴趣
             let interests = I::read_and_write(token);
             self.reregister(fd, token, interests)
-                .or(self.register(fd, token, interests))
+                .or_else(|_| self.register(fd, token, interests))
         } else {
             self.register(fd, token, I::write(token))
         }?;

core/src/scheduler.rs

@@ -55,26 +55,26 @@ impl Ord for SuspendItem<'_> {
 
 #[repr(C)]
 #[derive(Debug)]
-struct SyscallSuspendItem<'s> {
+struct SyscallSuspendItem {
     timestamp: u64,
-    co_name: &'s str,
+    co_id: usize,
 }
 
-impl PartialEq<Self> for SyscallSuspendItem<'_> {
+impl PartialEq<Self> for SyscallSuspendItem {
     fn eq(&self, other: &Self) -> bool {
         self.timestamp.eq(&other.timestamp)
     }
 }
 
-impl Eq for SyscallSuspendItem<'_> {}
+impl Eq for SyscallSuspendItem {}
 
-impl PartialOrd<Self> for SyscallSuspendItem<'_> {
+impl PartialOrd<Self> for SyscallSuspendItem {
     fn partial_cmp(&self, other: &Self) -> Option<std::cmp::Ordering> {
         Some(self.cmp(other))
     }
 }
 
-impl Ord for SyscallSuspendItem<'_> {
+impl Ord for SyscallSuspendItem {
     fn cmp(&self, other: &Self) -> std::cmp::Ordering {
         // BinaryHeap defaults to a large top heap, but we need a small top heap
         other.timestamp.cmp(&self.timestamp)
@@ -98,8 +98,8 @@ pub struct Scheduler<'s> {
     #[doc = include_str!("../docs/en/ordered-work-steal.md")]
     ready: OrderedLocalQueue<'s, SchedulableCoroutine<'s>>,
     suspend: BinaryHeap<SuspendItem<'s>>,
-    syscall: DashMap<&'s str, SchedulableCoroutine<'s>>,
-    syscall_suspend: BinaryHeap<SyscallSuspendItem<'s>>,
+    syscall: DashMap<usize, SchedulableCoroutine<'s>>,
+    syscall_suspend: BinaryHeap<SyscallSuspendItem>,
 }
 
 impl Default for Scheduler<'_> {
@@ -220,8 +220,8 @@ impl<'s> Scheduler<'s> {
     ///
     /// # Errors
     /// if change to ready fails.
-    pub fn try_resume(&self, co_name: &'s str) {
-        if let Some((_, co)) = self.syscall.remove(&co_name) {
+    pub fn try_resume(&self, co_id: usize) {
+        if let Some((_, co)) = self.syscall.remove(&co_id) {
             match co.state() {
                 CoroutineState::Syscall(val, syscall, SyscallState::Suspend(_)) => {
                     co.syscall(val, syscall, SyscallState::Callback)
@@ -313,10 +313,11 @@ impl<'s> Scheduler<'s> {
                     CoroutineState::Syscall((), _, state) => {
                         //挂起协程到系统调用表
                         //如果已包含，说明当前系统调用还有上层父系统调用，因此直接忽略插入结果
-                        _ = self.syscall.insert(co_name, coroutine);
+                        let co_id = coroutine.id();
+                        _ = self.syscall.insert(co_id, coroutine);
                         if let SyscallState::Suspend(timestamp) = state {
                             self.syscall_suspend
-                                .push(SyscallSuspendItem { timestamp, co_name });
+                                .push(SyscallSuspendItem { timestamp, co_id });
                         }
                     }
                     CoroutineState::Suspend((), timestamp) => {
@@ -373,7 +374,7 @@ impl<'s> Scheduler<'s> {
                 break;
             }
             if let Some(item) = self.syscall_suspend.pop() {
-                if let Some((_, co)) = self.syscall.remove(item.co_name) {
+                if let Some((_, co)) = self.syscall.remove(&item.co_id) {
                     match co.state() {
                         CoroutineState::Syscall(val, syscall, SyscallState::Suspend(_)) => {
                             co.syscall(val, syscall, SyscallState::Timeout)?;
@@ -420,7 +421,7 @@ mod tests {
         for timestamp in (0..10).rev() {
             heap.push(SyscallSuspendItem {
                 timestamp,
-                co_name: "test",
+                co_id: 1,
             });
         }
         for timestamp in 0..10 {