X-Git-Url: https://www.chiark.greenend.org.uk/ucgi/~ianmdlvl/git?p=elogind.git;a=blobdiff_plain;f=src%2Flibsystemd-bus%2Fsd-event.c;h=a1baac52aaaa324a5b372d05583d30293ad18502;hp=462dd413e27c263cd804ffaf8f15fdf794cc9158;hb=ff0031069ffeb0ccda209297b6b39c4786ac770b;hpb=ba276c8153e510a6741dc8a9492f5bed2b375825

diff --git a/src/libsystemd-bus/sd-event.c b/src/libsystemd-bus/sd-event.c
index 462dd413e..a1baac52a 100644
--- a/src/libsystemd-bus/sd-event.c
+++ b/src/libsystemd-bus/sd-event.c
@@ -34,7 +34,7 @@
 
 #include "sd-event.h"
 
-#define EPOLL_QUEUE_MAX 64
+#define EPOLL_QUEUE_MAX 512U
 #define DEFAULT_ACCURACY_USEC (250 * USEC_PER_MSEC)
 
 typedef enum EventSourceType {
@@ -44,7 +44,7 @@ typedef enum EventSourceType {
         SOURCE_SIGNAL,
         SOURCE_CHILD,
         SOURCE_DEFER,
-        SOURCE_QUIT,
+        SOURCE_EXIT,
         SOURCE_WATCHDOG
 } EventSourceType;
 
@@ -58,6 +58,7 @@ struct sd_event_source {
         EventSourceType type:4;
         int enabled:3;
         bool pending:1;
+        bool dispatching:1;
 
         int priority;
         unsigned pending_index;
@@ -96,7 +97,7 @@ struct sd_event_source {
                 struct {
                         sd_event_handler_t callback;
                         unsigned prioq_index;
-                } quit;
+                } exit;
         };
 };
 
@@ -132,7 +133,7 @@ struct sd_event {
         Hashmap *child_sources;
         unsigned n_enabled_child_sources;
 
-        Prioq *quit;
+        Prioq *exit;
 
         pid_t original_pid;
 
@@ -140,14 +141,18 @@ struct sd_event {
         dual_timestamp timestamp;
         int state;
 
-        bool quit_requested:1;
+        bool exit_requested:1;
         bool need_process_child:1;
         bool watchdog:1;
 
+        int exit_code;
+
         pid_t tid;
         sd_event **default_event_ptr;
 
         usec_t watchdog_last, watchdog_period;
+
+        unsigned n_sources;
 };
 
 static int pending_prioq_compare(const void *a, const void *b) {
@@ -284,11 +289,11 @@ static int latest_time_prioq_compare(const void *a, const void *b) {
         return 0;
 }
 
-static int quit_prioq_compare(const void *a, const void *b) {
+static int exit_prioq_compare(const void *a, const void *b) {
         const sd_event_source *x = a, *y = b;
 
-        assert(x->type == SOURCE_QUIT);
-        assert(y->type == SOURCE_QUIT);
+        assert(x->type == SOURCE_EXIT);
+        assert(y->type == SOURCE_EXIT);
 
         /* Enabled ones first */
         if (x->enabled != SD_EVENT_OFF && y->enabled == SD_EVENT_OFF)
@@ -313,6 +318,7 @@ static int quit_prioq_compare(const void *a, const void *b) {
 
 static void event_free(sd_event *e) {
         assert(e);
+        assert(e->n_sources == 0);
 
         if (e->default_event_ptr)
                 *(e->default_event_ptr) = NULL;
@@ -338,7 +344,7 @@ static void event_free(sd_event *e) {
         prioq_free(e->monotonic_latest);
         prioq_free(e->realtime_earliest);
         prioq_free(e->realtime_latest);
-        prioq_free(e->quit);
+        prioq_free(e->exit);
 
         free(e->signal_sources);
 
@@ -467,6 +473,8 @@ static void source_free(sd_event_source *s) {
         assert(s);
 
         if (s->event) {
+                assert(s->event->n_sources > 0);
+
                 switch (s->type) {
 
                 case SOURCE_IO:
@@ -515,9 +523,12 @@ static void source_free(sd_event_source *s) {
                         /* nothing */
                         break;
 
-                case SOURCE_QUIT:
-                        prioq_remove(s->event->quit, s, &s->quit.prioq_index);
+                case SOURCE_EXIT:
+                        prioq_remove(s->event->exit, s, &s->exit.prioq_index);
                         break;
+
+                case SOURCE_WATCHDOG:
+                        assert_not_reached("Wut? I shouldn't exist.");
                 }
 
                 if (s->pending)
@@ -526,6 +537,7 @@ static void source_free(sd_event_source *s) {
                 if (s->prepare)
                         prioq_remove(s->event->prepare, s, &s->prepare_index);
 
+                s->event->n_sources--;
                 sd_event_unref(s->event);
         }
 
@@ -536,7 +548,7 @@ static int source_set_pending(sd_event_source *s, bool b) {
         int r;
 
         assert(s);
-        assert(s->type != SOURCE_QUIT);
+        assert(s->type != SOURCE_EXIT);
 
         if (s->pending == b)
                 return 0;
@@ -579,6 +591,8 @@ static sd_event_source *source_new(sd_event *e, EventSourceType type) {
         s->type = type;
         s->pending_index = s->prepare_index = PRIOQ_IDX_NULL;
 
+        e->n_sources ++;
+
         return s;
 }
 
@@ -934,7 +948,7 @@ _public_ int sd_event_add_defer(
         return 0;
 }
 
-_public_ int sd_event_add_quit(
+_public_ int sd_event_add_exit(
                 sd_event *e,
                 sd_event_handler_t callback,
                 void *userdata,
@@ -949,22 +963,22 @@ _public_ int sd_event_add_quit(
         assert_return(e->state != SD_EVENT_FINISHED, -ESTALE);
         assert_return(!event_pid_changed(e), -ECHILD);
 
-        if (!e->quit) {
-                e->quit = prioq_new(quit_prioq_compare);
-                if (!e->quit)
+        if (!e->exit) {
+                e->exit = prioq_new(exit_prioq_compare);
+                if (!e->exit)
                         return -ENOMEM;
         }
 
-        s = source_new(e, SOURCE_QUIT);
+        s = source_new(e, SOURCE_EXIT);
         if (!s)
                 return -ENOMEM;
 
-        s->quit.callback = callback;
+        s->exit.callback = callback;
         s->userdata = userdata;
-        s->quit.prioq_index = PRIOQ_IDX_NULL;
+        s->exit.prioq_index = PRIOQ_IDX_NULL;
         s->enabled = SD_EVENT_ONESHOT;
 
-        r = prioq_put(s->event->quit, s, &s->quit.prioq_index);
+        r = prioq_put(s->event->exit, s, &s->exit.prioq_index);
         if (r < 0) {
                 source_free(s);
                 return r;
@@ -991,8 +1005,21 @@ _public_ sd_event_source* sd_event_source_unref(sd_event_source *s) {
         assert(s->n_ref >= 1);
         s->n_ref--;
 
-        if (s->n_ref <= 0)
-                source_free(s);
+        if (s->n_ref <= 0) {
+                /* Here's a special hack: when we are called from a
+                 * dispatch handler we won't free the event source
+                 * immediately, but we will detach the fd from the
+                 * epoll. This way it is safe for the caller to unref
+                 * the event source and immediately close the fd, but
+                 * we still retain a valid event source object after
+                 * the callback. */
+
+                if (s->dispatching) {
+                        if (s->type == SOURCE_IO)
+                                source_io_unregister(s);
+                } else
+                        source_free(s);
+        }
 
         return NULL;
 }
@@ -1005,7 +1032,7 @@ _public_ sd_event *sd_event_source_get_event(sd_event_source *s) {
 
 _public_ int sd_event_source_get_pending(sd_event_source *s) {
         assert_return(s, -EINVAL);
-        assert_return(s->type != SOURCE_QUIT, -EDOM);
+        assert_return(s->type != SOURCE_EXIT, -EDOM);
         assert_return(s->event->state != SD_EVENT_FINISHED, -ESTALE);
         assert_return(!event_pid_changed(s->event), -ECHILD);
 
@@ -1020,6 +1047,42 @@ _public_ int sd_event_source_get_io_fd(sd_event_source *s) {
         return s->io.fd;
 }
 
+_public_ int sd_event_source_set_io_fd(sd_event_source *s, int fd) {
+        int r;
+
+        assert_return(s, -EINVAL);
+        assert_return(fd >= 0, -EINVAL);
+        assert_return(s->type == SOURCE_IO, -EDOM);
+        assert_return(!event_pid_changed(s->event), -ECHILD);
+
+        if (s->io.fd == fd)
+                return 0;
+
+        if (s->enabled == SD_EVENT_OFF) {
+                s->io.fd = fd;
+                s->io.registered = false;
+        } else {
+                int saved_fd;
+
+                saved_fd = s->io.fd;
+                assert(s->io.registered);
+
+                s->io.fd = fd;
+                s->io.registered = false;
+
+                r = source_io_register(s, s->enabled, s->io.events);
+                if (r < 0) {
+                        s->io.fd = saved_fd;
+                        s->io.registered = true;
+                        return r;
+                }
+
+                epoll_ctl(s->event->epoll_fd, EPOLL_CTL_DEL, saved_fd, NULL);
+        }
+
+        return 0;
+}
+
 _public_ int sd_event_source_get_io_events(sd_event_source *s, uint32_t* events) {
         assert_return(s, -EINVAL);
         assert_return(events, -EINVAL);
@@ -1096,8 +1159,8 @@ _public_ int sd_event_source_set_priority(sd_event_source *s, int priority) {
         if (s->prepare)
                 prioq_reshuffle(s->event->prepare, s, &s->prepare_index);
 
-        if (s->type == SOURCE_QUIT)
-                prioq_reshuffle(s->event->quit, s, &s->quit.prioq_index);
+        if (s->type == SOURCE_EXIT)
+                prioq_reshuffle(s->event->exit, s, &s->exit.prioq_index);
 
         return 0;
 }
@@ -1168,14 +1231,17 @@ _public_ int sd_event_source_set_enabled(sd_event_source *s, int m) {
 
                         break;
 
-                case SOURCE_QUIT:
+                case SOURCE_EXIT:
                         s->enabled = m;
-                        prioq_reshuffle(s->event->quit, s, &s->quit.prioq_index);
+                        prioq_reshuffle(s->event->exit, s, &s->exit.prioq_index);
                         break;
 
                 case SOURCE_DEFER:
                         s->enabled = m;
                         break;
+
+                case SOURCE_WATCHDOG:
+                        assert_not_reached("Wut? I shouldn't exist.");
                 }
 
         } else {
@@ -1223,14 +1289,17 @@ _public_ int sd_event_source_set_enabled(sd_event_source *s, int m) {
                         }
                         break;
 
-                case SOURCE_QUIT:
+                case SOURCE_EXIT:
                         s->enabled = m;
-                        prioq_reshuffle(s->event->quit, s, &s->quit.prioq_index);
+                        prioq_reshuffle(s->event->exit, s, &s->exit.prioq_index);
                         break;
 
                 case SOURCE_DEFER:
                         s->enabled = m;
                         break;
+
+                case SOURCE_WATCHDOG:
+                        assert_not_reached("Wut? I shouldn't exist.");
                 }
         }
 
@@ -1321,7 +1390,7 @@ _public_ int sd_event_source_set_prepare(sd_event_source *s, sd_event_handler_t
         int r;
 
         assert_return(s, -EINVAL);
-        assert_return(s->type != SOURCE_QUIT, -EDOM);
+        assert_return(s->type != SOURCE_EXIT, -EDOM);
         assert_return(s->event->state != SD_EVENT_FINISHED, -ESTALE);
         assert_return(!event_pid_changed(s->event), -ECHILD);
 
@@ -1355,6 +1424,17 @@ _public_ void* sd_event_source_get_userdata(sd_event_source *s) {
         return s->userdata;
 }
 
+_public_ void *sd_event_source_set_userdata(sd_event_source *s, void *userdata) {
+        void *ret;
+
+        assert_return(s, NULL);
+
+        ret = s->userdata;
+        s->userdata = userdata;
+
+        return ret;
+}
+
 static usec_t sleep_between(sd_event *e, usec_t a, usec_t b) {
         usec_t c;
         assert(e);
@@ -1489,12 +1569,21 @@ static int event_arm_timer(
         return 0;
 }
 
-static int process_io(sd_event *e, sd_event_source *s, uint32_t events) {
+static int process_io(sd_event *e, sd_event_source *s, uint32_t revents) {
         assert(e);
         assert(s);
         assert(s->type == SOURCE_IO);
 
-        s->io.revents = events;
+        /* If the event source was already pending, we just OR in the
+         * new revents, otherwise we reset the value. The ORing is
+         * necessary to handle EPOLLONESHOT events properly where
+         * readability might happen independently of writability, and
+         * we need to keep track of both */
+
+        if (s->pending)
+                s->io.revents |= revents;
+        else
+                s->io.revents = revents;
 
         return source_set_pending(s, true);
 }
@@ -1516,7 +1605,7 @@ static int flush_timer(sd_event *e, int fd, uint32_t events, usec_t *next) {
                 return -errno;
         }
 
-        if (ss != sizeof(x))
+        if (_unlikely_(ss != sizeof(x)))
                 return -EIO;
 
         if (next)
@@ -1644,7 +1733,7 @@ static int process_signal(sd_event *e, uint32_t events) {
                         return -errno;
                 }
 
-                if (ss != sizeof(si))
+                if (_unlikely_(ss != sizeof(si)))
                         return -EIO;
 
                 read_one = true;
@@ -1673,9 +1762,9 @@ static int source_dispatch(sd_event_source *s) {
         int r = 0;
 
         assert(s);
-        assert(s->pending || s->type == SOURCE_QUIT);
+        assert(s->pending || s->type == SOURCE_EXIT);
 
-        if (s->type != SOURCE_DEFER && s->type != SOURCE_QUIT) {
+        if (s->type != SOURCE_DEFER && s->type != SOURCE_EXIT) {
                 r = source_set_pending(s, false);
                 if (r < 0)
                         return r;
@@ -1687,7 +1776,7 @@ static int source_dispatch(sd_event_source *s) {
                         return r;
         }
 
-        sd_event_source_ref(s);
+        s->dispatching = true;
 
         switch (s->type) {
 
@@ -1727,14 +1816,25 @@ static int source_dispatch(sd_event_source *s) {
                 r = s->defer.callback(s, s->userdata);
                 break;
 
-        case SOURCE_QUIT:
-                r = s->quit.callback(s, s->userdata);
+        case SOURCE_EXIT:
+                r = s->exit.callback(s, s->userdata);
                 break;
+
+        case SOURCE_WATCHDOG:
+                assert_not_reached("Wut? I shouldn't exist.");
         }
 
-        sd_event_source_unref(s);
+        s->dispatching = false;
 
-        return r;
+        if (r < 0)
+                log_debug("Event source %p returned error, disabling: %s", s, strerror(-r));
+
+        if (s->n_ref == 0)
+                source_free(s);
+        else if (r < 0)
+                sd_event_source_set_enabled(s, SD_EVENT_OFF);
+
+        return 1;
 }
 
 static int event_prepare(sd_event *e) {
@@ -1755,22 +1855,30 @@ static int event_prepare(sd_event *e) {
                         return r;
 
                 assert(s->prepare);
+
+                s->dispatching = true;
                 r = s->prepare(s, s->userdata);
+                s->dispatching = false;
+
                 if (r < 0)
-                        return r;
+                        log_debug("Prepare callback of event source %p returned error, disabling: %s", s, strerror(-r));
 
+                if (s->n_ref == 0)
+                        source_free(s);
+                else if (r < 0)
+                        sd_event_source_set_enabled(s, SD_EVENT_OFF);
         }
 
         return 0;
 }
 
-static int dispatch_quit(sd_event *e) {
+static int dispatch_exit(sd_event *e) {
         sd_event_source *p;
         int r;
 
         assert(e);
 
-        p = prioq_peek(e->quit);
+        p = prioq_peek(e->exit);
         if (!p || p->enabled == SD_EVENT_OFF) {
                 e->state = SD_EVENT_FINISHED;
                 return 0;
@@ -1778,7 +1886,7 @@ static int dispatch_quit(sd_event *e) {
 
         sd_event_ref(e);
         e->iteration++;
-        e->state = SD_EVENT_QUITTING;
+        e->state = SD_EVENT_EXITING;
 
         r = source_dispatch(p);
 
@@ -1841,7 +1949,8 @@ static int process_watchdog(sd_event *e) {
 }
 
 _public_ int sd_event_run(sd_event *e, uint64_t timeout) {
-        struct epoll_event ev_queue[EPOLL_QUEUE_MAX];
+        struct epoll_event *ev_queue;
+        unsigned ev_queue_max;
         sd_event_source *p;
         int r, i, m;
 
@@ -1850,8 +1959,8 @@ _public_ int sd_event_run(sd_event *e, uint64_t timeout) {
         assert_return(e->state != SD_EVENT_FINISHED, -ESTALE);
         assert_return(e->state == SD_EVENT_PASSIVE, -EBUSY);
 
-        if (e->quit_requested)
-                return dispatch_quit(e);
+        if (e->exit_requested)
+                return dispatch_exit(e);
 
         sd_event_ref(e);
         e->iteration++;
@@ -1871,11 +1980,13 @@ _public_ int sd_event_run(sd_event *e, uint64_t timeout) {
 
         if (event_next_pending(e) || e->need_process_child)
                 timeout = 0;
+        ev_queue_max = CLAMP(e->n_sources, 1U, EPOLL_QUEUE_MAX);
+        ev_queue = newa(struct epoll_event, ev_queue_max);
 
-        m = epoll_wait(e->epoll_fd, ev_queue, EPOLL_QUEUE_MAX,
+        m = epoll_wait(e->epoll_fd, ev_queue, ev_queue_max,
                        timeout == (uint64_t) -1 ? -1 : (int) ((timeout + USEC_PER_MSEC - 1) / USEC_PER_MSEC));
         if (m < 0) {
-                r = errno == EAGAIN || errno == EINTR ? 0 : -errno;
+                r = errno == EAGAIN || errno == EINTR ? 1 : -errno;
                 goto finish;
         }
 
@@ -1946,7 +2057,7 @@ _public_ int sd_event_loop(sd_event *e) {
                         goto finish;
         }
 
-        r = 0;
+        r = e->exit_code;
 
 finish:
         sd_event_unref(e);
@@ -1960,19 +2071,26 @@ _public_ int sd_event_get_state(sd_event *e) {
         return e->state;
 }
 
-_public_ int sd_event_get_quit(sd_event *e) {
+_public_ int sd_event_get_exit_code(sd_event *e, int *code) {
         assert_return(e, -EINVAL);
+        assert_return(code, -EINVAL);
         assert_return(!event_pid_changed(e), -ECHILD);
 
-        return e->quit_requested;
+        if (!e->exit_requested)
+                return -ENODATA;
+
+        *code = e->exit_code;
+        return 0;
 }
 
-_public_ int sd_event_request_quit(sd_event *e) {
+_public_ int sd_event_exit(sd_event *e, int code) {
         assert_return(e, -EINVAL);
         assert_return(e->state != SD_EVENT_FINISHED, -ESTALE);
         assert_return(!event_pid_changed(e), -ECHILD);
 
-        e->quit_requested = true;
+        e->exit_requested = true;
+        e->exit_code = code;
+
         return 0;
 }
 
@@ -1998,7 +2116,7 @@ _public_ int sd_event_get_now_monotonic(sd_event *e, uint64_t *usec) {
 
 _public_ int sd_event_default(sd_event **ret) {
 
-        static __thread sd_event *default_event = NULL;
+        static thread_local sd_event *default_event = NULL;
         sd_event *e;
         int r;
 
@@ -2039,6 +2157,7 @@ _public_ int sd_event_set_watchdog(sd_event *e, int b) {
         int r;
 
         assert_return(e, -EINVAL);
+        assert_return(!event_pid_changed(e), -ECHILD);
 
         if (e->watchdog == !!b)
                 return e->watchdog;
@@ -2094,3 +2213,10 @@ fail:
         e->watchdog_fd = -1;
         return r;
 }
+
+_public_ int sd_event_get_watchdog(sd_event *e) {
+        assert_return(e, -EINVAL);
+        assert_return(!event_pid_changed(e), -ECHILD);
+
+        return e->watchdog;
+}