From 5b40acf2738c889edb00d4d10a2261b7dcdeed4d Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?=E9=B8=A3=E8=88=9F?= <ray.dr@alibaba-inc.com>
Date: Wed, 11 Sep 2024 16:38:38 +0800
Subject: [PATCH] fix

---
 io/epoll.cpp  |  7 ++++---
 io/kqueue.cpp | 15 ++++++---------
 2 files changed, 10 insertions(+), 12 deletions(-)

diff --git a/io/epoll.cpp b/io/epoll.cpp
index 8b8d4a8c..8a3a3142 100644
--- a/io/epoll.cpp
+++ b/io/epoll.cpp
@@ -299,7 +299,8 @@ ok:     entry.interests |= eint;
             LOG_ERROR_RETURN(EINVAL, -1, "can not wait for multiple interests");
         if (unlikely(interest == 0))
             return rm_interest({fd, EVENT_RWE| ONE_SHOT, 0}); // remove fd from epoll
-        int ret = add_interest({fd, interest | ONE_SHOT, CURRENT});
+        thread* current = CURRENT;
+        int ret = add_interest({fd, interest | ONE_SHOT, current});
         if (ret < 0) LOG_ERROR_RETURN(0, -1, "failed to add event interest");
         // if timeout is just simple 0, wait for a tiny little moment
         // so that events can be collect.
@@ -307,8 +308,8 @@ ok:     entry.interests |= eint;
             ret = -1;
             wait_for_events(
                 0,
-                [&](void* data) __INLINE__ {
-                    if ((thread*)data == CURRENT) {
+                [current, &ret](void* data) __INLINE__ {
+                    if ((thread*)data == current) {
                         ret = 0;
                     } else {
                         thread_interrupt((thread*)data, EOK);
diff --git a/io/kqueue.cpp b/io/kqueue.cpp
index 4a0842dd..c80d659e 100644
--- a/io/kqueue.cpp
+++ b/io/kqueue.cpp
@@ -84,7 +84,7 @@ class KQueue : public MasterEventEngine, public CascadingEventEngine, public Res
         auto entry = &_events[_n++];
         EV_SET(entry, fd, event, action, event_flags, 0, udata);
         if (immediate || _n == LEN(_events)) {
-            struct timespec tm {0, 0};
+            struct timespec tm{0, 0};
             int ret = kevent(_kq, _events, _n, nullptr, 0, &tm);
             if (ret < 0) {
                 // debug_breakpoint();
@@ -111,7 +111,8 @@ class KQueue : public MasterEventEngine, public CascadingEventEngine, public Res
         nev += ret;
         for (int i = 0; i < ret; ++i) {
             if (_events[i].filter == EVFILT_USER) continue;
-            event_callback((thread*)_events[i].udata);
+            auto th = (thread*) _events[i].udata;
+            if (th) event_callback(th);
         }
         if (ret == (int) LEN(_events)) {  // there may be more events
             tm.tv_sec = tm.tv_nsec = 0;
@@ -121,8 +122,6 @@ class KQueue : public MasterEventEngine, public CascadingEventEngine, public Res
     }
 
     int wait_for_fd(int fd, uint32_t interests, Timeout timeout) override {
-        if (unlikely(interests == 0))
-            return 0;
         short ev = (interests == EVENT_READ) ? EVFILT_READ : EVFILT_WRITE;
         auto current = CURRENT;
         enqueue(fd, ev, EV_ADD | EV_ONESHOT, 0, current, true);
@@ -132,7 +131,7 @@ class KQueue : public MasterEventEngine, public CascadingEventEngine, public Res
                 if (th == current)
                     ret = 0;
                 else
-                    thread_interrupt(th);
+                    thread_interrupt(th, EOK);
             });
             if (ret <0) {
                 enqueue(fd, ev, EV_DELETE, 0, current, true);
@@ -145,16 +144,14 @@ class KQueue : public MasterEventEngine, public CascadingEventEngine, public Res
         if (ret == -1 && err.no == EOK) {
             return 0;  // event arrived
         }
-    
+
         errno = (ret == 0) ? ETIMEDOUT : err.no;
         enqueue(fd, ev, EV_DELETE, 0, current, true);
         return -1;
     }
 
     ssize_t wait_and_fire_events(uint64_t timeout) override {
-        return do_wait_and_fire_events(timeout, [](thread* th) {
-            thread_interrupt(th);
-        });
+        return do_wait_and_fire_events(timeout, [](thread *th) { thread_interrupt(th, EOK); });
     }
 
     int cancel_wait() override {