selfwake: use epoll_pwait on Linux instead of eventfd

This saves us a file descriptor in Linux, which provides
epoll_pwait in 2.6.19+ (and ppoll for 2.6.18, the oldest
kernel we support).

12 files changed, 190 insertions, 158 deletions

diff --git a/Makefile.am b/Makefile.am
index 8ec860c..1e2b598 100644
--- a/Makefile.am
+++ b/Makefile.am
@@ -25,6 +25,7 @@ mog_src += cloexec_from.c
 mog_src += close.c
 mog_src += cmogstored.h
 mog_src += compat_accept.h
+mog_src += compat_epoll_pwait.h
 mog_src += compat_memstream.h
 mog_src += compat_sendfile.h
 mog_src += defaults.h
@@ -77,8 +78,7 @@ mog_src += queue_kqueue.c
 mog_src += queue_kqueue.h
 mog_src += queue_loop.c
 mog_src += queue_step.c
-mog_src += selfwake_eventfd.h
-mog_src += selfwake_pipe.h
+mog_src += selfwake.h
 mog_src += selfwake.c
 mog_src += sig.c
 mog_src += svc.c
diff --git a/cmogstored.h b/cmogstored.h
index 2fe0f05..2a31857 100644
--- a/cmogstored.h
+++ b/cmogstored.h
@@ -45,6 +45,7 @@
 #include <argp.h>
 #include <sched.h>
 #include <error.h> /* GNU */
+#include <poll.h>
 #include "bsd/queue_safe.h"
 #include "bsd/simpleq.h"
 
@@ -239,6 +240,7 @@ struct mog_file {
 
 #include "queue_epoll.h"
 #include "notify.h"
+#include "selfwake.h"
 
 enum mog_fd_type {
         MOG_FD_TYPE_UNUSED = 0,
@@ -385,6 +387,7 @@ void mog_idleq_push(struct mog_queue *, struct mog_fd *, enum mog_qev);
 struct mog_fd * mog_idleq_wait(struct mog_queue *, int timeout);
 struct mog_fd *
 mog_queue_xchg(struct mog_queue *, struct mog_fd *, enum mog_qev);
+struct mog_fd * mog_idleq_wait_intr(struct mog_queue *q, int timeout);
 
 /* addrinfo.c */
 struct mog_addrinfo {
diff --git a/compat_epoll_pwait.h b/compat_epoll_pwait.h
new file mode 100644
index 0000000..a3f2102
--- /dev/null
+++ b/compat_epoll_pwait.h
@@ -0,0 +1,38 @@
+/*
+ * fake epoll_pwait() implemented using ppoll + epoll_wait.
+ * This is only for Linux 2.6.18 / glibc 2.5 systems (Enterprise distros :P)
+ *
+ * Copyright (C) 2012-2013 Eric Wong <normalperson@yhbt.net>
+ * License: GPLv3 or later (see COPYING for details)
+ */
+
+#if !defined(HAVE_EPOLL_PWAIT) \
+    && defined(HAVE_PPOLL) && defined(HAVE_EPOLL_WAIT)
+static int my_epoll_pwait(int epfd, struct epoll_event *events,
+                          int maxevents, int timeout, const sigset_t *sigmask)
+{
+        struct pollfd pfds = { .fd = epfd, .events = POLLIN };
+        int rc;
+        struct timespec ts;
+        struct timespec *tsp;
+
+        if (timeout < 0) {
+                tsp = NULL;
+        } else {
+                ts.tv_sec = timeout / 1000;
+                ts.tv_nsec = (timeout % 1000) * 1000000;
+                tsp = &ts;
+        }
+
+        /* wait on just the epoll descriptor itself */
+        rc = ppoll(&pfds, 1, tsp, sigmask);
+        if (rc < 0)
+                assert((errno == EINTR || errno == ENOMEM)
+                       && "ppoll usage bug");
+
+        /* no sleep for the actual epoll call */
+        return rc > 0 ? epoll_wait(epfd, events, maxevents, 0) : 0;
+}
+#define epoll_pwait(epfd,events,maxevents,timeout,sigmask) \
+        my_epoll_pwait((epfd),(events),(maxevents),(timeout),(sigmask))
+#endif /* epoll_pwait */
diff --git a/configure.ac b/configure.ac
index ca202da..08cab08 100644
--- a/configure.ac
+++ b/configure.ac
@@ -34,8 +34,7 @@ AC_C_FLEXIBLE_ARRAY_MEMBER
 dnl gnulib *at functions aren't thread-safe, ask for the real thing
 AC_CHECK_FUNCS([openat renameat mkdirat fstatat unlinkat])
 
-AC_CHECK_HEADERS_ONCE([sys/eventfd.h])
-AC_CHECK_FUNCS([epoll_wait eventfd])
+AC_CHECK_FUNCS([epoll_wait epoll_pwait ppoll])
 
 dnl libkqueue should work in the future
 AC_CHECK_FUNCS([kqueue])
diff --git a/notify.c b/notify.c
index 7d7872a..be9221b 100644
--- a/notify.c
+++ b/notify.c
@@ -4,8 +4,8 @@
  */
 #include "cmogstored.h"
 
+static int notes[MOG_NOTIFY_MAX];
 static struct mog_fd *notify_mfd;
-static struct mog_selfwake *notify;
 static time_t usage_file_updated_at;
 static time_t usage_file_interval = 10;
 struct mog_queue *mog_notify_queue;
@@ -22,16 +22,16 @@ void mog_notify_init(void)
         }
 
         assert(mog_notify_queue == NULL && "notify queue already initialized");
-        assert(notify == NULL && "notify already initialized");
         assert(notify_mfd == NULL && "notify_mfd already initialized");
 
         mog_notify_queue = mog_queue_new();
         notify_mfd = mog_selfwake_new();
-        notify = &notify_mfd->as.selfwake;
-        assert(notify->writer && "notify writer not initialized");
-        notify->queue = mog_notify_queue;
-        mog_idleq_add(notify->queue, notify_mfd, MOG_QEV_RD);
-        memset(&notify->as.notes, 0, sizeof(notify->as.notes));
+        if (notify_mfd) {
+                struct mog_selfwake *notify = &notify_mfd->as.selfwake;
+                assert(notify->writer && "notify writer not initialized");
+                notify->queue = mog_notify_queue;
+                mog_idleq_add(notify->queue, notify_mfd, MOG_QEV_RD);
+        }
 }
 
 static void global_mkusage(void)
@@ -42,20 +42,23 @@ static void global_mkusage(void)
 
 static inline bool note_xchg(enum mog_notification note, int from, int to)
 {
-        return __sync_bool_compare_and_swap(&notify->as.notes[note], from, to);
+        return __sync_bool_compare_and_swap(&notes[note], from, to);
 }
 
-/* drain the pipe and process notifications */
-static void note_queue_step(struct mog_fd *mfd)
+static void note_run(void)
 {
-        mog_selfwake_drain(mfd);
-
         if (note_xchg(MOG_NOTIFY_DEVICE_REFRESH, 1, 0))
                 global_mkusage();
 
         if (note_xchg(MOG_NOTIFY_SET_N_THREADS, 1, 0))
                 mog_thrpool_process_queue();
+}
 
+/* drain the pipe and process notifications */
+static void note_queue_step(struct mog_fd *mfd)
+{
+        mog_selfwake_drain(mfd);
+        note_run();
         mog_idleq_push(mfd->as.selfwake.queue, mfd, MOG_QEV_RD);
 }
 
@@ -84,7 +87,7 @@ void mog_notify_wait(bool need_usage_file)
          * epoll_wait() with timeout==0 can avoid some slow paths,
          * so take anything that's already ready before sleeping
          */
-        while ((mfd = mog_idleq_wait(notify->queue, 0)))
+        while ((mfd = mog_idleq_wait(mog_notify_queue, 0)))
                 notify_queue_step(mfd);
 
         if (need_usage_file == false)
@@ -94,11 +97,11 @@ void mog_notify_wait(bool need_usage_file)
         else
                 timeout = 0;
 
-        mog_intr_enable();
-        mfd = mog_idleq_wait(notify->queue, timeout);
-        mog_intr_disable();
+        mfd = mog_idleq_wait_intr(mog_notify_queue, timeout);
         if (mfd)
                 notify_queue_step(mfd);
+        else if (errno == EINTR)
+                note_run();
 }
 
 /* this is async-signal safe */
@@ -108,6 +111,7 @@ void mog_notify(enum mog_notification note)
         case MOG_NOTIFY_DEVICE_REFRESH:
         case MOG_NOTIFY_SET_N_THREADS:
                 note_xchg(note, 0, 1);
+                mog_selfwake_interrupt();
                 break;
         case MOG_NOTIFY_SIGNAL: break;
         default: assert(0 && "bad note passed");
diff --git a/notify.h b/notify.h
index 1888fda..f6a6f37 100644
--- a/notify.h
+++ b/notify.h
@@ -9,23 +9,4 @@ enum mog_notification {
         MOG_NOTIFY_MAX
 };
 
-struct mog_selfwake {
-        struct mog_queue *queue;
-        struct mog_fd *writer;
-        union {
-                int notes[MOG_NOTIFY_MAX];
-        } as;
-};
-
-/* only for pipe */
-struct mog_selfpipe {
-        struct mog_fd *reader; /* points to mog_selfwake */
-};
-
 extern struct mog_queue *mog_notify_queue;
-
-/* selfwake*.[ch] */
-struct mog_fd * mog_selfwake_new(void);
-void mog_selfwake_trigger(struct mog_fd *);
-void mog_selfwake_drain(struct mog_fd *);
-void mog_selfwake_wait(struct mog_fd *);
diff --git a/queue_epoll.c b/queue_epoll.c
index ab289c7..a680180 100644
--- a/queue_epoll.c
+++ b/queue_epoll.c
@@ -9,6 +9,7 @@
  * migrating clients between threads
  */
 #if defined(HAVE_EPOLL_WAIT) && ! MOG_LIBKQUEUE
+#include "compat_epoll_pwait.h"
 #include <sys/utsname.h>
 
 /*
@@ -100,6 +101,25 @@ struct mog_queue * mog_queue_new(void)
         return mog_queue_init(epoll_fd);
 }
 
+static struct mog_fd * epoll_event_check(int rc, struct epoll_event *event)
+{
+        struct mog_fd *mfd;
+
+        switch (rc) {
+        case 1:
+                mfd = event->data.ptr;
+                mog_fd_check_out(mfd);
+                return mfd;
+        case 0:
+                return NULL;
+        }
+
+        if (errno != EINTR)
+                /* rc could be > 1 if the kernel is broken :P */
+                die_errno("epoll_wait() failed with (%d)", rc);
+        return NULL;
+}
+
 /*
  * grabs one active event off the event queue
  * epoll_wait() has "wake-one" behavior (like accept())
@@ -109,7 +129,6 @@ struct mog_fd * mog_idleq_wait(struct mog_queue *q, int timeout)
 {
         int rc;
         struct epoll_event event;
-        struct mog_fd *mfd;
         bool cancellable = timeout != 0;
 
         if (cancellable)
@@ -120,20 +139,18 @@ struct mog_fd * mog_idleq_wait(struct mog_queue *q, int timeout)
 
         if (cancellable)
                 mog_cancel_disable();
+        return epoll_event_check(rc, &event);
+}
 
-        switch (rc) {
-        case 1:
-                mfd = event.data.ptr;
-                mog_fd_check_out(mfd);
-                return mfd;
-        case 0:
-                return NULL;
-        }
+struct mog_fd * mog_idleq_wait_intr(struct mog_queue *q, int timeout)
+{
+        int rc;
+        struct epoll_event event;
+        sigset_t set;
 
-        if (errno != EINTR)
-                /* rc could be > 1 if the kernel is broken :P */
-                die_errno("epoll_wait() failed with (%d)", rc);
-        return NULL;
+        CHECK(int, 0, sigemptyset(&set));
+        rc = epoll_pwait(q->queue_fd, &event, 1, timeout, &set);
+        return epoll_event_check(rc, &event);
 }
 
 MOG_NOINLINE static void
diff --git a/queue_kqueue.c b/queue_kqueue.c
index 92d5bac..6d2da43 100644
--- a/queue_kqueue.c
+++ b/queue_kqueue.c
@@ -79,6 +79,17 @@ struct mog_fd * mog_idleq_wait(struct mog_queue *q, int timeout)
         return NULL;
 }
 
+struct mog_fd * mog_idleq_wait_intr(struct mog_queue *q, int timeout)
+{
+        struct mog_fd *mfd;
+
+        /* this is racy, using a self-pipe covers the race */
+        mog_intr_enable();
+        mfd = mog_idleq_wait(q, timeout);
+        mog_intr_disable();
+        return mfd;
+}
+
 MOG_NOINLINE static void
 kevent_add_error(struct mog_queue *q, struct mog_fd *mfd)
 {
diff --git a/selfwake.c b/selfwake.c
index d0fbf44..77c7ae7 100644
--- a/selfwake.c
+++ b/selfwake.c
@@ -3,8 +3,44 @@
  * License: GPLv3 or later (see COPYING for details)
  */
 #include "cmogstored.h"
-#include "selfwake_eventfd.h"
-#include "selfwake_pipe.h"
+
+#if MOG_SELFPIPE
+struct mog_fd * mog_selfwake_new(void)
+{
+        struct mog_fd *reader, *writer;
+        struct mog_selfwake *selfwake;
+        int self_pipe[2];
+
+        if (pipe2(self_pipe, O_NONBLOCK | O_CLOEXEC) < 0)
+                die_errno("failed to init self-pipe");
+
+        reader = mog_fd_init(self_pipe[0], MOG_FD_TYPE_SELFWAKE);
+        selfwake = &reader->as.selfwake;
+
+        writer = mog_fd_init(self_pipe[1], MOG_FD_TYPE_SELFPIPE);
+        writer->as.selfpipe.reader = reader;
+        selfwake->writer = writer;
+
+        return reader;
+}
+
+static ssize_t selfwake_drain(struct mog_fd *mfd)
+{
+        char buf[64];
+        return read(mfd->fd, buf, sizeof(buf));
+}
+
+/* this allows interrupts */
+void mog_selfwake_wait(struct mog_fd *mfd)
+{
+        /* poll() on a pipe does not work on some *BSDs, so just block */
+        mog_set_nonblocking(mfd->fd, false);
+        mog_intr_enable();
+        (void)selfwake_drain(mfd);
+        mog_intr_disable();
+        mog_set_nonblocking(mfd->fd, true);
+        mog_selfwake_drain(mfd);
+}
 
 /* this is async-signal safe (except in the case of bugs) */
 void mog_selfwake_trigger(struct mog_fd *mfd)
@@ -12,7 +48,7 @@ void mog_selfwake_trigger(struct mog_fd *mfd)
         ssize_t w;
 
 retry:
-        w = selfwake_write(mfd);
+        w = write(mfd->as.selfwake.writer->fd, "", 1);
         if (w >= 0) return;
 
         switch (errno) {
@@ -38,3 +74,4 @@ void mog_selfwake_drain(struct mog_fd *mfd)
         assert(r < 0 && (errno == EAGAIN || errno == EINTR)
                && "selfwake read failed");
 }
+#endif /* MOG_SELFPIPE */
diff --git a/selfwake.h b/selfwake.h
new file mode 100644
index 0000000..6806b94
--- /dev/null
+++ b/selfwake.h
@@ -0,0 +1,45 @@
+/*
+ * Copyright (C) 2012-2013, Eric Wong <normalperson@yhbt.net>
+ * License: GPLv3 or later (see COPYING for details)
+ */
+#if (defined(HAVE_EPOLL_WAIT) && defined(HAVE_PPOLL)) \
+    || defined(HAVE_EPOLL_PWAIT)
+# define MOG_SELFPIPE 0
+#else
+# define MOG_SELFPIPE 1
+#endif
+
+struct mog_selfwake {
+        struct mog_queue *queue;
+        struct mog_fd *writer;
+};
+
+/* only for pipe */
+struct mog_selfpipe {
+        struct mog_fd *reader; /* points to mog_selfwake */
+};
+
+#if MOG_SELFPIPE
+struct mog_fd * mog_selfwake_new(void);
+void mog_selfwake_trigger(struct mog_fd *);
+void mog_selfwake_drain(struct mog_fd *);
+static inline void mog_selfwake_interrupt(void) {}
+void mog_selfwake_wait(struct mog_fd *);
+#else /* use Linux-only facilities like epoll_pwait or ppoll */
+static inline void mog_selfwake_wait(struct mog_fd *mfd)
+{
+        sigset_t set;
+
+        CHECK(int, 0, sigemptyset(&set));
+        if (ppoll(NULL, 0, NULL, &set) != 0)
+                assert((errno == EINTR || errno == ENOMEM)
+                       && "ppoll usage bug");
+}
+static inline struct mog_fd * mog_selfwake_new(void) { return NULL; }
+static inline void mog_selfwake_trigger(struct mog_fd *mfd) {}
+static inline void mog_selfwake_drain(struct mog_fd *mfd) {}
+static inline void mog_selfwake_interrupt(void)
+{
+        CHECK(int, 0, kill(getpid(), SIGURG));
+}
+#endif /* Linux-only stuff */
diff --git a/selfwake_eventfd.h b/selfwake_eventfd.h
deleted file mode 100644
index 836d037..0000000
--- a/selfwake_eventfd.h
+++ /dev/null
@@ -1,54 +0,0 @@
-/*
- * Copyright (C) 2012-2013, Eric Wong <normalperson@yhbt.net>
- * License: GPLv3 or later (see COPYING for details)
- */
-#if defined(HAVE_SYS_EVENTFD_H) && defined(HAVE_EVENTFD)
-#include <sys/eventfd.h>
-#if defined(EFD_CLOEXEC) && defined(EFD_NONBLOCK)
-#include <sys/poll.h>
-
-static ssize_t selfwake_write(struct mog_fd *mfd)
-{
-        static const uint64_t buf = 1;
-
-        return write(mfd->fd, &buf, sizeof(buf));
-}
-
-struct mog_fd * mog_selfwake_new(void)
-{
-        struct mog_fd *mfd;
-        int efd = eventfd(0, EFD_CLOEXEC | EFD_NONBLOCK);
-
-        if (efd < 0)
-                die_errno("failed to create eventfd");
-
-        mfd = mog_fd_init(efd, MOG_FD_TYPE_SELFWAKE);
-        mfd->as.selfwake.writer = mfd;
-
-        return mfd;
-}
-
-static ssize_t selfwake_drain(struct mog_fd *mfd)
-{
-        uint64_t buf;
-
-        return read(mfd->fd, &buf, sizeof(buf));
-}
-
-/* this allows interrupts */
-void mog_selfwake_wait(struct mog_fd *mfd)
-{
-        sigset_t set;
-        struct pollfd pfd = { .fd = mfd->fd, .events = POLLIN };
-        int r;
-
-        CHECK(int, 0, sigemptyset(&set));
-
-        r = ppoll(&pfd, 1, NULL, &set);
-        if ((r < 0) && (errno != ENOMEM))
-                assert(errno == EINTR && "BUG in using ppoll");
-
-        mog_selfwake_drain(mfd);
-}
-#endif /* EFD_CLOEXEC+EFD_NONBLOCK */
-#endif /* HAVE_SYS_EVENTFD_H  */
diff --git a/selfwake_pipe.h b/selfwake_pipe.h
deleted file mode 100644
index f61d755..0000000
--- a/selfwake_pipe.h
+++ /dev/null
@@ -1,49 +0,0 @@
-/*
- * Copyright (C) 2012-2013, Eric Wong <normalperson@yhbt.net>
- * License: GPLv3 or later (see COPYING for details)
- */
-#ifndef HAVE_EVENTFD
-static ssize_t selfwake_write(struct mog_fd *mfd)
-{
-        return write(mfd->as.selfwake.writer->fd, "", 1);
-}
-
-struct mog_fd * mog_selfwake_new(void)
-{
-        struct mog_fd *reader, *writer;
-        struct mog_selfwake *selfwake;
-        int self_pipe[2];
-
-        if (pipe2(self_pipe, O_NONBLOCK | O_CLOEXEC) < 0)
-                die_errno("failed to init self-pipe");
-
-        reader = mog_fd_init(self_pipe[0], MOG_FD_TYPE_SELFWAKE);
-        selfwake = &reader->as.selfwake;
-
-        writer = mog_fd_init(self_pipe[1], MOG_FD_TYPE_SELFPIPE);
-        writer->as.selfpipe.reader = reader;
-        selfwake->writer = writer;
-
-        return reader;
-}
-
-static ssize_t selfwake_drain(struct mog_fd *mfd)
-{
-        char buf[64];
-        return read(mfd->fd, buf, sizeof(buf));
-}
-
-/* this allows interrupts */
-void mog_selfwake_wait(struct mog_fd *mfd)
-{
-        ssize_t r;
-
-        /* poll() on a pipe does not work on some *BSDs, so just block */
-        mog_set_nonblocking(mfd->fd, false);
-        mog_intr_enable();
-        r = selfwake_drain(mfd);
-        mog_intr_disable();
-        mog_set_nonblocking(mfd->fd, true);
-        mog_selfwake_drain(mfd);
-}
-#endif /* !HAVE_EVENTFD */