Replace pipe-based notification with EVFILT_USER where possible

Sufficiently recent kqueue implementations have an EVFILT_USER filter
that we can use to wake up an event base from another thread.  When
it's supported, we now use this mechanism rather than our old
(pipe-based) mechanism.  This should save some time and complications
on newer OSX and freebsds.

4 files changed, 133 insertions, 2 deletions

diff --git a/Makefile.am b/Makefile.am
index 838864e2..320294f3 100644
--- a/Makefile.am
+++ b/Makefile.am
@@ -234,6 +234,8 @@ noinst_HEADERS = util-internal.h mm-internal.h ipv6-internal.h \
 	changelist-internal.h iocp-internal.h \
 	ratelim-internal.h \
 	evconfig-private.h \
+	ratelim-internal.h \
+	kqueue-internal.h \
 	WIN32-Code/event2/event-config.h \
 	WIN32-Code/evconfig-private.h \
 	WIN32-Code/tree.h \
diff --git a/event.c b/event.c
index 2fc1c22e..a8867112 100644
--- a/event.c
+++ b/event.c
@@ -72,6 +72,11 @@
 #include "ht-internal.h"
 #include "util-internal.h"
 
+
+#ifdef EVENT__HAVE_WORKING_KQUEUE
+#include "kqueue-internal.h"
+#endif
+
 #ifdef EVENT__HAVE_EVENT_PORTS
 extern const struct eventop evportops;
 #endif
@@ -927,8 +932,11 @@ event_reinit(struct event_base *base)
 		had_signal_added = 1;
 		base->sig.ev_signal_added = 0;
 	}
-	if (base->th_notify_fd[0] != -1) {
+	if (base->th_notify_fn != NULL) {
 		was_notifiable = 1;
+		base->th_notify_fn = NULL;
+	}
+	if (base->th_notify_fd[0] != -1) {
 		event_del(&base->th_notify);
 		EVUTIL_CLOSESOCKET(base->th_notify_fd[0]);
 		if (base->th_notify_fd[1] != -1)
@@ -936,7 +944,6 @@ event_reinit(struct event_base *base)
 		base->th_notify_fd[0] = -1;
 		base->th_notify_fd[1] = -1;
 		event_debug_unassign(&base->th_notify);
-		base->th_notify_fn = NULL;
 	}
 
 	/* Replace the original evsel. */
@@ -3056,6 +3063,16 @@ evthread_make_base_notifiable(struct event_base *base)
 		return 0;
 	}
 
+
+#if defined(EVENT__HAVE_WORKING_KQUEUE)
+	if (base->evsel == &kqops && event_kq_add_notify_event_(base) == 0) {
+		base->th_notify_fn = event_kq_notify_base_;
+		/* No need to add an event here; the backend can wake
+		 * itself up just fine. */
+		return 0;
+	}
+#endif
+
 #ifdef EVENT__HAVE_EVENTFD
 	base->th_notify_fd[0] = evutil_eventfd_(0,
 	    EVUTIL_EFD_CLOEXEC|EVUTIL_EFD_NONBLOCK);
diff --git a/kqueue-internal.h b/kqueue-internal.h
new file mode 100644
index 00000000..02c5a360
--- /dev/null
+++ b/kqueue-internal.h
@@ -0,0 +1,39 @@
+/*
+ * Copyright (c) 2012 Niels Provos and Nick Mathewson
+ *
+ * Redistribution and use in source and binary forms, with or without
+ * modification, are permitted provided that the following conditions
+ * are met:
+ * 1. Redistributions of source code must retain the above copyright
+ *    notice, this list of conditions and the following disclaimer.
+ * 2. Redistributions in binary form must reproduce the above copyright
+ *    notice, this list of conditions and the following disclaimer in the
+ *    documentation and/or other materials provided with the distribution.
+ * 3. The name of the author may not be used to endorse or promote products
+ *    derived from this software without specific prior written permission.
+ *
+ * THIS SOFTWARE IS PROVIDED BY THE AUTHOR ``AS IS'' AND ANY EXPRESS OR
+ * IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES
+ * OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED.
+ * IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR ANY DIRECT, INDIRECT,
+ * INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT
+ * NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
+ * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
+ * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+ * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF
+ * THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ */
+#ifndef KQUEUE_INTERNAL_H_INCLUDED_
+#define KQUEUE_INTERNAL_H_INCLUDED_
+
+/** Notification function, used to tell an event base to wake up from another
+ * thread.  Only works when event_kq_add_notify_event_() has previously been
+ * called successfully on that base. */
+int event_kq_notify_base_(struct event_base *base);
+
+/** Prepare a kqueue-using event base to receive notifications via an internal
+ * EVFILT_USER event.  Return 0 on sucess, -1 on failure.
+ */
+int event_kq_add_notify_event_(struct event_base *base);
+
+#endif
diff --git a/kqueue.c b/kqueue.c
index 18a7c64f..39334674 100644
--- a/kqueue.c
+++ b/kqueue.c
@@ -65,6 +65,8 @@
 #include "evthread-internal.h"
 #include "changelist-internal.h"
 
+#include "kqueue-internal.h"
+
 #define NEVENT		64
 
 struct kqop {
@@ -74,6 +76,7 @@ struct kqop {
 	struct kevent *events;
 	int events_size;
 	int kq;
+	int notify_event_added;
 	pid_t pid;
 };
 
@@ -369,6 +372,10 @@ kq_dispatch(struct event_base *base, struct timeval *tv)
 			which |= EV_WRITE;
 		} else if (events[i].filter == EVFILT_SIGNAL) {
 			which |= EV_SIGNAL;
+#ifdef EVFILT_USER
+		} else if (events[i].filter == EVFILT_USER) {
+			base->is_notify_pending = 0;
+#endif
 		}
 
 		if (!which)
@@ -473,4 +480,70 @@ kq_sig_del(struct event_base *base, int nsignal, short old, short events, void *
 	return (0);
 }
 
+
+/* OSX 10.6 and FreeBSD 8.1 add support for EVFILT_USER, which we can use
+ * to wake up the event loop from another thread. */
+
+/* Magic number we use for our filter ID. */
+#define NOTIFY_IDENT 42
+
+int
+event_kq_add_notify_event_(struct event_base *base)
+{
+	struct kqop *kqop = base->evbase;
+#if defined(EVFILT_USER) && defined(NOTE_TRIGGER)
+	struct kevent kev;
+	struct timespec timeout = { 0, 0 };
+#endif
+
+	if (kqop->notify_event_added)
+		return 0;
+
+#if defined(EVFILT_USER) && defined(NOTE_TRIGGER)
+	memset(&kev, 0, sizeof(kev));
+	kev.ident = NOTIFY_IDENT;
+	kev.filter = EVFILT_USER;
+	kev.flags = EV_ADD | EV_CLEAR;
+
+	if (kevent(kqop->kq, &kev, 1, NULL, 0, &timeout) == -1) {
+		event_warn("kevent: adding EVFILT_USER event");
+		return -1;
+	}
+
+	kqop->notify_event_added = 1;
+
+	return 0;
+#else
+	return -1;
+#endif
+}
+
+int
+event_kq_notify_base_(struct event_base *base)
+{
+	struct kqop *kqop = base->evbase;
+#if defined(EVFILT_USER) && defined(NOTE_TRIGGER)
+	struct kevent kev;
+	struct timespec timeout = { 0, 0 };
+#endif
+	if (! kqop->notify_event_added)
+		return -1;
+
+#if defined(EVFILT_USER) && defined(NOTE_TRIGGER)
+	memset(&kev, 0, sizeof(kev));
+	kev.ident = NOTIFY_IDENT;
+	kev.filter = EVFILT_USER;
+	kev.fflags = NOTE_TRIGGER;
+
+	if (kevent(kqop->kq, &kev, 1, NULL, 0, &timeout) == -1) {
+		event_warn("kevent: triggering EVFILT_USER event");
+		return -1;
+	}
+
+	return 0;
+#else
+	return -1;
+#endif
+}
+
 #endif /* EVENT__HAVE_KQUEUE */