Do not notify the main thread more than needed.

Basically, we suppress the notification when an event is added or deleted
and:
  - The event has no fd, or there is no change in whether we are
    reading/writing on the event's fd.
  - The event has no timeout, or adding the event did not make the earliest
    timeout become earlier.

This should be a big efficiency win in applications with multiple threads and
lots of timeouts.

svn:r1439

diff --git a/event.c b/event.c
index 0884ffccfe1ba0175475b781f4b5373cd097c856..5fb943aecbde04b52412db07c7fc61e71dae2f25 100644 (file)
--- a/event.c
+++ b/event.c
@@ -1221,6 +1221,7 @@ event_add_internal(struct event *ev, const struct timeval *tv)
 {
        struct event_base *base = ev->ev_base;
        int res = 0;
+       int notify = 0;
 
        event_debug((
                 "event_add: event: %p, %s%s%scall %p",
@@ -1250,6 +1251,11 @@ event_add_internal(struct event *ev, const struct timeval *tv)
                        res = evmap_signal_add(base, ev->ev_fd, ev);
                if (res != -1)
                        event_queue_insert(base, ev, EVLIST_INSERTED);
+               if (res == 1) {
+                       /* evmap says we need to notify the main thread. */
+                       notify = 1;
+                       res = 0;
+               }
        }
 
        /*
@@ -1270,8 +1276,11 @@ event_add_internal(struct event *ev, const struct timeval *tv)
                 * we already reserved memory above for the case where we
                 * are not replacing an exisiting timeout.
                 */
-               if (ev->ev_flags & EVLIST_TIMEOUT)
+               if (ev->ev_flags & EVLIST_TIMEOUT) {
+                       if (min_heap_elt_is_top(ev))
+                               notify = 1;
                        event_queue_remove(base, ev, EVLIST_TIMEOUT);
+               }
 
                /* Check if it is active due to a timeout.  Rescheduling
                 * this timeout before the callback can be executed
@@ -1299,10 +1308,16 @@ event_add_internal(struct event *ev, const struct timeval *tv)
                         (int)tv->tv_sec, ev->ev_callback));
 
                event_queue_insert(base, ev, EVLIST_TIMEOUT);
+               if (min_heap_elt_is_top(ev)) {
+                       /* The earliest timeout is now earlier than it was
+                        * before: we will need to tell the main thread to
+                        * wake up earlier than it would otherwise. */
+                       notify = 1;
+               }
        }
 
        /* if we are not in the right thread, we need to wake up the loop */
-       if (res != -1 && !EVBASE_IN_THREAD(base))
+       if (res != -1 && notify && !EVBASE_IN_THREAD(base))
                evthread_notify_base(base);
 
        return (res);
@@ -1327,7 +1342,7 @@ static inline int
 event_del_internal(struct event *ev)
 {
        struct event_base *base;
-       int res = 0;
+       int res = 0, notify = 0;
        int need_cur_lock;
 
        event_debug(("event_del: %p, callback %p",
@@ -1357,8 +1372,16 @@ event_del_internal(struct event *ev)
                }
        }
 
-       if (ev->ev_flags & EVLIST_TIMEOUT)
+       if (ev->ev_flags & EVLIST_TIMEOUT) {
+               /* NOTE: We never need to notify the main thread because of a
+                * deleted timeout event: all that could happen if we don't is
+                * that the dispatch loop might wake up too early.  But the
+                * point of notifying the main thread _is_ to wake up the
+                * dispatch loop early anyway, so we wouldn't gain anything by
+                * doing it.
+                */
                event_queue_remove(base, ev, EVLIST_TIMEOUT);
+       }
 
        if (ev->ev_flags & EVLIST_ACTIVE)
                event_queue_remove(base, ev, EVLIST_ACTIVE);
@@ -1369,10 +1392,15 @@ event_del_internal(struct event *ev)
                        res = evmap_io_del(base, ev->ev_fd, ev);
                else
                        res = evmap_signal_del(base, ev->ev_fd, ev);
+               if (res == 1) {
+                       /* evmap says we need to notify the main thread. */
+                       notify = 1;
+                       res = 0;
+               }
        }
 
        /* if we are not in the right thread, we need to wake up the loop */
-       if (res != -1 && !EVBASE_IN_THREAD(base))
+       if (res != -1 && notify && !EVBASE_IN_THREAD(base))
                evthread_notify_base(base);
 
        if (need_cur_lock)

diff --git a/evmap.c b/evmap.c
index eb714b4b5417f9aeb15ec5c56b75e69c2584fdf1..0b858245a1c3a59eefff6bb3edba63cfefcd9cfa 100644 (file)
--- a/evmap.c
+++ b/evmap.c
@@ -254,13 +254,15 @@ evmap_io_init(struct evmap_io *entry)
 }
 
 
+/* return -1 on error, 0 on success if nothing changed in the event backend,
+ * and 1 on success if something did. */
 int
 evmap_io_add(struct event_base *base, int fd, struct event *ev)
 {
        const struct eventop *evsel = base->evsel;
        struct event_io_map *io = &base->io;
        struct evmap_io *ctx = NULL;
-       int nread, nwrite;
+       int nread, nwrite, retval = 0;
        short res = 0, old = 0;
 
        assert(fd == ev->ev_fd); /*XXX(nickm) always true? */
@@ -304,22 +306,25 @@ evmap_io_add(struct event_base *base, int fd, struct event *ev)
                if (evsel->add(base, ev->ev_fd,
                                           old, (ev->ev_events & EV_ET) | res, extra) == -1)
                        return (-1);
+               retval = 1;
        }
 
        ctx->nread = nread;
        ctx->nwrite = nwrite;
        TAILQ_INSERT_TAIL(&ctx->events, ev, ev_io_next);
 
-       return (0);
+       return (retval);
 }
 
+/* return -1 on error, 0 on success if nothing changed in the event backend,
+ * and 1 on success if something did. */
 int
 evmap_io_del(struct event_base *base, int fd, struct event *ev)
 {
        const struct eventop *evsel = base->evsel;
        struct event_io_map *io = &base->io;
        struct evmap_io *ctx;
-       int nread, nwrite;
+       int nread, nwrite, retval = 0;
        short res = 0, old = 0;
 
        if (fd < 0)
@@ -359,13 +364,14 @@ evmap_io_del(struct event_base *base, int fd, struct event *ev)
                void *extra = ((char*)ctx) + sizeof(struct evmap_io);
                if (evsel->del(base, ev->ev_fd, old, res, extra) == -1)
                        return (-1);
+               retval = 1;
        }
 
        ctx->nread = nread;
        ctx->nwrite = nwrite;
        TAILQ_REMOVE(&ctx->events, ev, ev_io_next);
 
-       return (0);
+       return (retval);
 }
 
 void
@@ -417,7 +423,7 @@ evmap_signal_add(struct event_base *base, int sig, struct event *ev)
 
        TAILQ_INSERT_TAIL(&ctx->events, ev, ev_signal_next);
 
-       return (0);
+       return (1);
 }
 
 int
@@ -439,7 +445,7 @@ evmap_signal_del(struct event_base *base, int sig, struct event *ev)
 
        TAILQ_REMOVE(&ctx->events, ev, ev_signal_next);
 
-       return (0);
+       return (1);
 }
 
 void

diff --git a/minheap-internal.h b/minheap-internal.h
index 56c87ecd3ac8f3edc303493733d21ea05141a5ac..8b638ecc42702c32f9b94e7e13db4bb978c83049 100644 (file)
--- a/minheap-internal.h
+++ b/minheap-internal.h
@@ -43,6 +43,7 @@ typedef struct min_heap
 static inline void           min_heap_ctor(min_heap_t* s);
 static inline void           min_heap_dtor(min_heap_t* s);
 static inline void           min_heap_elem_init(struct event* e);
+static inline int            min_heap_elt_is_top(const struct event *e);
 static inline int            min_heap_elem_greater(struct event *a, struct event *b);
 static inline int            min_heap_empty(min_heap_t* s);
 static inline unsigned       min_heap_size(min_heap_t* s);
@@ -86,6 +87,11 @@ struct event* min_heap_pop(min_heap_t* s)
     return 0;
 }
 
+int min_heap_elt_is_top(const struct event *e)
+{
+       return e->min_heap_idx == 0;
+}
+
 int min_heap_erase(min_heap_t* s, struct event* e)
 {
     if(((unsigned int)-1) != e->min_heap_idx)

diff --git a/test/regress.c b/test/regress.c
index 71e3776c9dd429dcf96169f53f5a49f1438e7ec0..dce50d6f397da99762ed4bc462c257fb247a47de 100644 (file)
--- a/test/regress.c
+++ b/test/regress.c
@@ -487,20 +487,16 @@ test_combined(void)
        event_set(&w1.ev, pair[0], EV_WRITE, combined_write_cb, &w1);
        event_set(&r2.ev, pair[1], EV_READ, combined_read_cb, &r2);
        event_set(&w2.ev, pair[1], EV_WRITE, combined_write_cb, &w2);
-       if (event_add(&r1.ev, NULL) == -1)
-               exit(1);
-       if (event_add(&w1.ev, NULL))
-               exit(1);
-       if (event_add(&r2.ev, NULL))
-               exit(1);
-       if (event_add(&w2.ev, NULL))
-               exit(1);
-
+       tt_assert(event_add(&r1.ev, NULL) != -1);
+       tt_assert(!event_add(&w1.ev, NULL));
+       tt_assert(!event_add(&r2.ev, NULL));
+       tt_assert(!event_add(&w2.ev, NULL));
        event_dispatch();
 
        if (r1.nread == 8192 && r2.nread == 4096)
                test_ok = 1;
 
+end:
        cleanup_test();
 }