Fix/sockpool wait_for_fd 0.8

io/kqueue.cpp

@@ -72,21 +72,14 @@ class KQueue : public MasterEventEngine, public CascadingEventEngine, public Res
             close(_kq);
     }
 
-    __attribute__((noinline))
-    static void debug_breakpoint() {
-    }
-
     int enqueue(int fd, short event, uint16_t action, uint32_t event_flags, void* udata, bool immediate = false) {
-        // if (fd == _kq) debug_breakpoint();
-        // immediate = true;
-        // LOG_DEBUG(VALUE(_kq), VALUE(fd), VALUE(event), VALUE(action), VALUE(event_flags), VALUE(udata), VALUE(immediate));
         assert(_n < LEN(_events));
         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()");
             }
             _n = 0;
@@ -95,16 +88,21 @@ 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;
+        if (unlikely(interests == 0)) {
+            errno = ENOSYS;
+            return -1;
+        }
         short ev = (interests == EVENT_READ) ? EVFILT_READ : EVFILT_WRITE;
-        enqueue(fd, ev, EV_ADD | EV_ONESHOT, 0, CURRENT);
-        int ret = thread_usleep(timeout);
+        auto current = CURRENT;
+        int ret = enqueue(fd, ev, EV_ADD | EV_ONESHOT, 0, current);
+        if (ret < 0) return ret;
+        ret = thread_usleep(timeout);
         ERRNO err;
         if (ret == -1 && err.no == EOK) {
             return 0;  // event arrived
         }
 
+        enqueue(fd, ev, EV_DELETE, 0, current, true);
         errno = (ret == 0) ? ETIMEDOUT : err.no;
         return -1;
     }
@@ -117,26 +115,30 @@ class KQueue : public MasterEventEngine, public CascadingEventEngine, public Res
 
     again:
         int ret = kevent(_kq, _events, _n, _events, LEN(_events), &tm);
-        if (ret < 0)
-            LOG_ERRNO_RETURN(0, -1, "failed to call kevent()");
+        if (ret < 0) LOG_ERRNO_RETURN(0, -1, "failed to call kevent()");
 
         _n = 0;
         nev += ret;
         for (int i = 0; i < ret; ++i) {
             if (_events[i].filter == EVFILT_USER) continue;
-            auto th = (thread*) _events[i].udata;
+            auto th = (thread*)_events[i].udata;
             if (th) thread_interrupt(th, EOK);
         }
-        if (ret == (int) LEN(_events)) {  // there may be more events
+        if (ret == (int)LEN(_events)) {  // there may be more events
             tm.tv_sec = tm.tv_nsec = 0;
             goto again;
         }
         return nev;
     }
 
     int cancel_wait() override {
-        enqueue(_kq, EVFILT_USER, EV_ONESHOT, NOTE_TRIGGER, nullptr, true);
-        return 0;
+        // cannot call `enqueue` directly since it will be called from another vCPU.
+        // directly use kqueue to submit event, which is safe.
+        // as same as `enqueue(_kq, EVFILT_USER, EV_ONESHOT, NOTE_TRIGGER, nullptr, true)`
+        struct kevent entry;
+        EV_SET(&entry, _kq, EVFILT_USER, EV_ONESHOT, NOTE_TRIGGER, 0, nullptr);
+        struct timespec tm{0, 0};
+        return kevent(_kq, _events, _n, nullptr, 0, &tm);
     }
 
     // This vector is used to filter invalid add/rm_interest requests which may affect kevent's

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;