kqueue: use cross-thread wakeup properly

By setting the proper filter flag depending on the file type, we can
eliminate the awkward NOTE_TRIGGER usage and unnecessary system
calls/locking of the active queue when migrating sockets between
threads.

1 files changed, 18 insertions, 34 deletions

diff --git a/queue_kqueue.c b/queue_kqueue.c
index 1242f51..a90aba4 100644
--- a/queue_kqueue.c
+++ b/queue_kqueue.c
@@ -9,26 +9,14 @@
  */
 #ifdef HAVE_KQUEUE
 #include "activeq.h"
-static const uintptr_t MOG_XFER_IDENT = (uintptr_t)INT_MAX + 1;
 
 struct mog_queue * mog_queue_new(void)
 {
         int kqueue_fd = kqueue();
-        struct kevent event;
-        int rc;
 
         if (kqueue_fd < 0)
                 die("kqueue failed: %s\n", strerror(errno));
 
-        EV_SET(&event, MOG_XFER_IDENT, EVFILT_USER,
-               EV_ADD | EV_CLEAR, 0, 0, NULL);
-        do {
-                rc = kevent(kqueue_fd, &event, 1, NULL, 0, NULL);
-        } while (rc < 0 && errno == EINTR);
-
-        if (rc != 0)
-                die("failed to add MOG_XFER_IDENT event: %s", strerror(errno));
-
         return mog_queue_init(kqueue_fd);
 }
 
@@ -57,8 +45,6 @@ retry:
         rc = kevent(q->queue_fd, NULL, 0, &event, 1, tsp);
         switch (rc) {
         case 1:
-                if (event.ident == MOG_XFER_IDENT)
-                        return NULL;
                 mfd = event.udata;
                 mog_fd_check_out(mfd);
                 return mfd;
@@ -99,25 +85,6 @@ kevent_add_error(struct mog_queue *q, struct mog_fd *mfd)
         }
 }
 
-static void qxfer(struct mog_queue *q, struct mog_fd *mfd)
-{
-        struct kevent event;
-        int rc;
-
-        assert(mfd->queue_state == MOG_QUEUE_STATE_NEW && "bad queue state");
-        mfd->queue_state = MOG_QUEUE_STATE_OLD;
-
-        mog_activeq_insert(q, mfd);
-
-        EV_SET(&event, MOG_XFER_IDENT, EVFILT_USER, 0, NOTE_TRIGGER, 0, NULL);
-        do {
-                rc = kevent(q->queue_fd, &event, 1, NULL, 0, NULL);
-        } while (rc < 0 && errno == EINTR);
-
-        if (rc != 0)
-                kevent_add_error(q, NULL);
-}
-
 static void qpush(struct mog_queue *q, struct mog_fd *mfd, enum mog_qev ev)
 {
         struct kevent event;
@@ -146,7 +113,24 @@ static void qpush(struct mog_queue *q, struct mog_fd *mfd, enum mog_qev ev)
  */
 void mog_idleq_push(struct mog_queue *q, struct mog_fd *mfd, enum mog_qev ev)
 {
-        ev == MOG_QEV_RW ? qxfer(q, mfd) : qpush(q, mfd, ev);
+        if (ev == MOG_QEV_RW) {
+                switch (mfd->fd_type) {
+                case MOG_FD_TYPE_IOSTAT:
+                case MOG_FD_TYPE_NOTIFYRD:
+                        ev = MOG_QEV_RD;
+                        break;
+                case MOG_FD_TYPE_UNUSED:
+                case MOG_FD_TYPE_ACCEPT:
+                case MOG_FD_TYPE_FILE:
+                case MOG_FD_TYPE_QUEUE:
+                case MOG_FD_TYPE_SVC:
+                        assert(0 && "invalid fd_type for mog_idleq_push");
+                default:
+                        ev = MOG_QEV_WR;
+                        break;
+                }
+        }
+        qpush(q, mfd, ev);
 }
 #else /* ! HAVE_KQUEUE */
 typedef int avoid_empty_file;