#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 {
SOURCE_SIGNAL,
SOURCE_CHILD,
SOURCE_DEFER,
- SOURCE_QUIT,
+ SOURCE_EXIT,
SOURCE_WATCHDOG
} EventSourceType;
EventSourceType type:4;
int enabled:3;
bool pending:1;
+ bool dispatching:1;
int priority;
unsigned pending_index;
struct {
sd_event_handler_t callback;
unsigned prioq_index;
- } quit;
+ } exit;
};
};
Hashmap *child_sources;
unsigned n_enabled_child_sources;
- Prioq *quit;
+ Prioq *exit;
pid_t original_pid;
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) {
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)
static void event_free(sd_event *e) {
assert(e);
+ assert(e->n_sources == 0);
if (e->default_event_ptr)
*(e->default_event_ptr) = NULL;
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);
assert(s);
if (s->event) {
+ assert(s->event->n_sources > 0);
+
switch (s->type) {
case SOURCE_IO:
/* 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)
if (s->prepare)
prioq_remove(s->event->prepare, s, &s->prepare_index);
+ s->event->n_sources--;
sd_event_unref(s->event);
}
int r;
assert(s);
- assert(s->type != SOURCE_QUIT);
+ assert(s->type != SOURCE_EXIT);
if (s->pending == b)
return 0;
s->type = type;
s->pending_index = s->prepare_index = PRIOQ_IDX_NULL;
+ e->n_sources ++;
+
return s;
}
return 0;
}
-_public_ int sd_event_add_quit(
+_public_ int sd_event_add_exit(
sd_event *e,
sd_event_handler_t callback,
void *userdata,
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;
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;
}
_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);
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);
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;
}
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 {
}
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.");
}
}
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);
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);
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);
}
return -errno;
}
- if (ss != sizeof(x))
+ if (_unlikely_(ss != sizeof(x)))
return -EIO;
if (next)
return -errno;
}
- if (ss != sizeof(si))
+ if (_unlikely_(ss != sizeof(si)))
return -EIO;
read_one = true;
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;
return r;
}
- sd_event_source_ref(s);
+ s->dispatching = true;
switch (s->type) {
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) {
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;
sd_event_ref(e);
e->iteration++;
- e->state = SD_EVENT_QUITTING;
+ e->state = SD_EVENT_EXITING;
r = source_dispatch(p);
}
_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;
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++;
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;
}
goto finish;
}
- r = 0;
+ r = e->exit_code;
finish:
sd_event_unref(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;
}
_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;
int r;
assert_return(e, -EINVAL);
+ assert_return(!event_pid_changed(e), -ECHILD);
if (e->watchdog == !!b)
return e->watchdog;
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;
+}
~ianmdlvl / elogind.git / blobdiff