Replicate EEXIST epoll_ctl behavior in kqueue

* In the epoll implementation, you get an EEXIST if you try to register the same event for the same fd more than once for a particular epoll instance

* Otherwise kevent will just override the previous event registration, and if multiple threads listen on the same fd only the last one to register will ever finish, the others are stuck

* This approach will lead to native threads getting created, similar to the epoll implementation. This is not ideal, but it fixes certain test cases for now, like test/socket/test_tcp.rb#test_accept_multithread

1 files changed, 19 insertions, 0 deletions

diff --git a/thread_pthread_mn.c b/thread_pthread_mn.c
index 137b0b2efa..a9c8dae74d 100644
--- a/thread_pthread_mn.c
+++ b/thread_pthread_mn.c
@@ -616,6 +616,21 @@ kqueue_unregister_waiting(int fd, enum thread_sched_waiting_flag flags)
     }
 }
 
+static bool
+kqueue_already_registered(int fd)
+{
+    rb_thread_t *wth, *found_wth = NULL;
+    ccan_list_for_each(&timer_th.waiting, wth, sched.waiting_reason.node) {
+        // Similar to EEXIST in epoll_ctl, but more strict because it checks fd rather than flags
+        //   for simplicity
+        if (wth->sched.waiting_reason.flags && wth->sched.waiting_reason.data.fd == fd) {
+            found_wth = wth;
+            break;
+        }
+    }
+    return found_wth != NULL;
+}
+
 #endif // HAVE_SYS_EVENT_H
 
 // return false if the fd is not waitable or not need to wait.
@@ -645,6 +660,10 @@ timer_thread_register_waiting(rb_thread_t *th, int fd, enum thread_sched_waiting
 #if HAVE_SYS_EVENT_H
     struct kevent ke[2];
     int num_events = 0;
+
+    if (kqueue_already_registered(fd)) {
+        return false;
+    }
 #else
     uint32_t epoll_events = 0;
 #endif