FIX: wait_for_fd in epoll reap events when timeout is 0

io/epoll.cpp

@@ -299,9 +299,30 @@ 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");
-        ret = thread_usleep(timeout);
+        // if timeout is just simple 0, wait for a tiny little moment
+        // so that events can be collect.
+        if (timeout.expired()) {
+            ret = -1;
+            wait_for_events(
+                0,
+                [current, &ret](void* data) __INLINE__ {
+                    if ((thread*)data == current) {
+                        ret = 0;
+                    } else {
+                        thread_interrupt((thread*)data, EOK);
+                    }
+                },
+                [&]() __INLINE__ { return true; });
+            if (ret < 0) {
+                rm_interest({fd, interest, 0});
+                errno = ETIMEDOUT;
+            }
+            return ret;
+        }
+        ret = thread_usleep(timeout.timeout() ? timeout : Timeout(10));
         ERRNO err;
         if (ret == -1 && err.no == EOK) {
             return 0;  // Event arrived

io/kqueue.cpp

@@ -84,7 +84,8 @@ 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)) {
-            int ret = kevent(_kq, _events, _n, nullptr, 0, nullptr);
+            struct timespec tm{0, 0};
+            int ret = kevent(_kq, _events, _n, nullptr, 0, &tm);
             if (ret < 0) {
                 // debug_breakpoint();
                 LOG_ERRNO_RETURN(0, -1, "failed to submit events with kevent()");
@@ -94,22 +95,8 @@ class KQueue : public MasterEventEngine, public CascadingEventEngine, public Res
         return 0;
     }
 
-    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;
-        enqueue(fd, ev, EV_ADD | EV_ONESHOT, 0, CURRENT);
-        int ret = thread_usleep(timeout);
-        ERRNO err;
-        if (ret == -1 && err.no == EOK) {
-            return 0;  // event arrived
-        }
-
-        errno = (ret == 0) ? ETIMEDOUT : err.no;
-        return -1;
-    }
-
-    ssize_t wait_and_fire_events(uint64_t timeout) override {
+    template<typename EVCB>
+    ssize_t do_wait_and_fire_events(uint64_t timeout, EVCB&& event_callback) {
         ssize_t nev = 0;
         struct timespec tm;
         tm.tv_sec = timeout / 1000 / 1000;
@@ -125,7 +112,7 @@ class KQueue : public MasterEventEngine, public CascadingEventEngine, public Res
         for (int i = 0; i < ret; ++i) {
             if (_events[i].filter == EVFILT_USER) continue;
             auto th = (thread*) _events[i].udata;
-            if (th) thread_interrupt(th, EOK);
+            if (th) event_callback(th);
         }
         if (ret == (int) LEN(_events)) {  // there may be more events
             tm.tv_sec = tm.tv_nsec = 0;
@@ -134,6 +121,39 @@ class KQueue : public MasterEventEngine, public CascadingEventEngine, public Res
         return nev;
     }
 
+    int wait_for_fd(int fd, uint32_t interests, Timeout timeout) override {
+        short ev = (interests == EVENT_READ) ? EVFILT_READ : EVFILT_WRITE;
+        auto current = CURRENT;
+        enqueue(fd, ev, EV_ADD | EV_ONESHOT, 0, current, true);
+        if (timeout.expired()) {
+            int ret = -1;
+            do_wait_and_fire_events(0, [current, &ret](thread* th) {
+                if (th == current)
+                    ret = 0;
+                else
+                    thread_interrupt(th, EOK);
+            });
+            if (ret <0) {
+                enqueue(fd, ev, EV_DELETE, 0, current, true);
+                errno = ETIMEDOUT;
+            }
+            return ret;
+        }
+        int ret = thread_usleep(timeout);
+        ERRNO err;
+        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, EOK); });
+    }
+
     int cancel_wait() override {
         enqueue(_kq, EVFILT_USER, EV_ONESHOT, NOTE_TRIGGER, nullptr, true);
         return 0;

net/pooled_socket.cpp

@@ -121,9 +121,10 @@ class TCPSocketPool : public ForwardSocketClient {
     photon::Timer timer;
 
     // all fd < 0 treated as socket not based on fd
-    // and always alive. Using such socket needs user
+    // and always reuseable. Using such socket needs user
     // to check if connected socket is still usable.
-    bool stream_alive(int fd) {
+    // if there still have unread bytes in strema, it should be closed.
+    bool stream_reusable(int fd) {
         return (fd < 0) || (wait_for_fd_readable(fd, 0) != 0);
     }
 
@@ -196,7 +197,7 @@ class TCPSocketPool : public ForwardSocketClient {
         if (!stream) {
             stream = m_underlay->connect(remote, local);
             if (!stream) return nullptr;
-        } else if (!stream_alive(stream->get_underlay_fd())) {
+        } else if (!stream_reusable(stream->get_underlay_fd())) {
             delete stream;
             goto again;
         }
@@ -231,7 +232,7 @@ class TCPSocketPool : public ForwardSocketClient {
     bool release(const EndPoint& ep, ISocketStream* stream) {
         auto fd = stream->get_underlay_fd();
         ERRNO err;
-        if (!stream_alive(fd)) return false;
+        if (!stream_reusable(fd)) return false;
         auto node = new StreamListNode(ep, stream, fd, TTL_us);
         push_into_pool(node);
         errno = err.no;