fix(driver): patch polling crate for kqueue EVFILT_USER EV_CLEAR bug

Cargo.toml

@@ -84,6 +84,9 @@ slotmap = "1.1.1"
 crossfire = "3.1.5"
 rustix = "1.1.4"
 
+[patch.crates-io]
+polling = { git = "https://github.com/paddor/polling.git", branch = "fix/kqueue-evfilt-user-clear" }
+
 [profile.bench]
 debug = true
 lto = true

compio-driver/src/key.rs

@@ -293,7 +293,7 @@ impl ErasedKey {
     }
 
     /// Whether the op is completed.
-    pub(crate) fn has_result(&self) -> bool {
+    pub fn has_result(&self) -> bool {
         self.borrow().result.is_ready()
     }
 

compio-runtime/src/future/future.rs

@@ -66,7 +66,9 @@ pin_project_lite::pin_project! {
     impl<T: OpCode, E> PinnedDrop for Submit<T, E> {
         fn drop(this: Pin<&mut Self>) {
             let this = this.project();
-            if let Some(State::Submitted { key, .. }) = this.state.take() {
+            if let Some(State::Submitted { key, .. }) = this.state.take()
+                && !key.has_result()
+            {
                 this.runtime.cancel(key);
             }
         }

compio-runtime/src/time.rs

@@ -12,7 +12,7 @@ use std::{
     time::{Duration, Instant},
 };
 
-use futures_util::{FutureExt, select};
+use futures_util::{FutureExt, select_biased};
 
 use crate::Runtime;
 
@@ -81,7 +81,7 @@ impl Error for Elapsed {}
 /// value is returned. Otherwise, an error is returned and the future is
 /// cancelled.
 pub async fn timeout<F: Future>(duration: Duration, future: F) -> Result<F::Output, Elapsed> {
-    select! {
+    select_biased! {
         res = future.fuse() => Ok(res),
         _ = sleep(duration).fuse() => Err(Elapsed(())),
     }

compio-runtime/tests/waker.rs

@@ -0,0 +1,38 @@
+use std::sync::Arc;
+use std::sync::atomic::{AtomicU32, Ordering};
+use std::time::{Duration, Instant};
+
+#[test]
+fn cross_thread_waker_interrupts_poll() {
+    let rt = compio_runtime::Runtime::new().unwrap();
+    rt.block_on(async {
+        let val = Arc::new(AtomicU32::new(0));
+        let val2 = val.clone();
+
+        let handle = compio_runtime::spawn(async move {
+            loop {
+                let v = val2.load(Ordering::Acquire);
+                if v != 0 {
+                    return v;
+                }
+                compio_runtime::time::sleep(Duration::from_millis(1)).await;
+            }
+        });
+
+        std::thread::spawn(move || {
+            std::thread::sleep(Duration::from_millis(10));
+            val.store(42, Ordering::Release);
+        });
+
+        let start = Instant::now();
+        use compio_runtime::ResumeUnwind;
+        let v = handle.await.resume_unwind().unwrap();
+        let elapsed = start.elapsed();
+
+        assert_eq!(v, 42);
+        assert!(
+            elapsed < Duration::from_millis(500),
+            "took {elapsed:?}, expected < 500ms"
+        );
+    });
+}