Minimize calls to base_notify implementation functions, thereby avoiding needless...

The trick here is that if we already told the base to wake up, and it
hasn't woken up yet, we don't need to tell it to wake up again.  This
should help lots with inherently multithreaded code like IOCP.

diff --git a/event-internal.h b/event-internal.h
index e52c129bc248ca673973a120286bcb4d6faa8235..122195b3ea695b8527d6d260a0cfdeb88a39735d 100644 (file)
--- a/event-internal.h
+++ b/event-internal.h
@@ -262,6 +262,9 @@ struct event_base {
        enum event_base_config_flag flags;
 
        /* Notify main thread to wake up break, etc. */
+       /** True if the base already has a pending notify, and we don't need
+        * to add any more. */
+       int is_notify_pending;
        /** A socketpair used by some th_notify functions to wake up the main
         * thread. */
        int th_notify_fd[2];

diff --git a/event.c b/event.c
index 0c8238d78358326da273463c210d77906b36b93b..902aed545f6a3535453964c8727eb6dcd75d22ab 100644 (file)
--- a/event.c
+++ b/event.c
@@ -1902,8 +1902,12 @@ evthread_notify_base_eventfd(struct event_base *base)
 static int
 evthread_notify_base(struct event_base *base)
 {
+       EVENT_BASE_ASSERT_LOCKED(base);
        if (!base->th_notify_fn)
                return -1;
+       if (base->is_notify_pending)
+               return 0;
+       base->is_notify_pending = 1;
        return base->th_notify_fn(base);
 }
 
@@ -2247,8 +2251,6 @@ event_deferred_cb_schedule(struct deferred_cb_queue *queue,
                cb->queued = 1;
                TAILQ_INSERT_TAIL(&queue->deferred_cb_list, cb, cb_next);
                ++queue->active_count;
-               /* XXXX Can we get away with doing this only when adding
-                * the first active deferred_cb to the queue? */
                if (queue->notify_fn)
                        queue->notify_fn(queue, queue->notify_arg);
        }
@@ -2594,11 +2596,15 @@ evthread_notify_drain_eventfd(evutil_socket_t fd, short what, void *arg)
 {
        ev_uint64_t msg;
        ev_ssize_t r;
+       struct event_base *base = arg;
 
        r = read(fd, (void*) &msg, sizeof(msg));
        if (r<0 && errno != EAGAIN) {
                event_sock_warn(fd, "Error reading from eventfd");
        }
+       EVBASE_ACQUIRE_LOCK(base, th_base_lock);
+       base->is_notify_pending = 0;
+       EVBASE_RELEASE_LOCK(base, th_base_lock);
 }
 #endif
 
@@ -2606,6 +2612,7 @@ static void
 evthread_notify_drain_default(evutil_socket_t fd, short what, void *arg)
 {
        unsigned char buf[1024];
+       struct event_base *base = arg;
 #ifdef WIN32
        while (recv(fd, (char*)buf, sizeof(buf), 0) > 0)
                ;
@@ -2613,6 +2620,10 @@ evthread_notify_drain_default(evutil_socket_t fd, short what, void *arg)
        while (read(fd, (char*)buf, sizeof(buf)) > 0)
                ;
 #endif
+
+       EVBASE_ACQUIRE_LOCK(base, th_base_lock);
+       base->is_notify_pending = 0;
+       EVBASE_RELEASE_LOCK(base, th_base_lock);
 }
 
 int