tevent zero...
[metze/samba/wip.git] / lib / tevent / tevent.c
index a91e56893123d90cd2a1f249162b426026557fef..b74948433dd87014236032d7a3d4b3a21062d670 100644 (file)
 */
 #include "replace.h"
 #include "system/filesys.h"
+#ifdef HAVE_PTHREAD
+#include "system/threads.h"
+#endif
 #define TEVENT_DEPRECATED 1
 #include "tevent.h"
 #include "tevent_internal.h"
 #include "tevent_util.h"
+#ifdef HAVE_EVENTFD
+#include <sys/eventfd.h>
+#endif
+
+static void tevent_abort(struct tevent_context *ev, const char *reason);
 
 struct tevent_ops_list {
        struct tevent_ops_list *next, *prev;
@@ -88,7 +96,7 @@ bool tevent_register_backend(const char *name, const struct tevent_ops *ops)
                }
        }
 
-       e = talloc(talloc_autofree_context(), struct tevent_ops_list);
+       e = talloc(NULL, struct tevent_ops_list);
        if (e == NULL) return false;
 
        e->name = name;
@@ -104,8 +112,7 @@ bool tevent_register_backend(const char *name, const struct tevent_ops *ops)
 void tevent_set_default_backend(const char *backend)
 {
        talloc_free(tevent_default_backend);
-       tevent_default_backend = talloc_strdup(talloc_autofree_context(),
-                                              backend);
+       tevent_default_backend = talloc_strdup(NULL, backend);
 }
 
 /*
@@ -113,11 +120,48 @@ void tevent_set_default_backend(const char *backend)
 */
 static void tevent_backend_init(void)
 {
+       static bool done;
+
+       if (done) {
+               return;
+       }
+
+       done = true;
+
        tevent_select_init();
-       tevent_standard_init();
-#ifdef HAVE_EPOLL
+       tevent_poll_init();
+       tevent_poll_mt_init();
+#if defined(HAVE_EPOLL)
        tevent_epoll_init();
+#elif defined(HAVE_SOLARIS_PORTS)
+       tevent_port_init();
+#endif
+#ifdef HAVE_KQUEUE
+       tevent_kqueue_init();
 #endif
+       tevent_standard_init();
+}
+
+_PRIVATE_ const struct tevent_ops *tevent_find_ops_byname(const char *name)
+{
+       struct tevent_ops_list *e;
+
+       tevent_backend_init();
+
+       if (name == NULL) {
+               name = tevent_default_backend;
+       }
+       if (name == NULL) {
+               name = "standard";
+       }
+
+       for (e = tevent_backends; e != NULL; e = e->next) {
+               if (0 == strcmp(e->name, name)) {
+                       return e->ops;
+               }
+       }
+
+       return NULL;
 }
 
 /*
@@ -139,6 +183,93 @@ const char **tevent_backend_list(TALLOC_CTX *mem_ctx)
        return list;
 }
 
+static void tevent_common_wakeup_fini(struct tevent_context *ev);
+
+#ifdef HAVE_PTHREAD
+
+static pthread_mutex_t tevent_contexts_mutex = PTHREAD_MUTEX_INITIALIZER;
+static struct tevent_context *tevent_contexts = NULL;
+static pthread_once_t tevent_atfork_initialized = PTHREAD_ONCE_INIT;
+
+static void tevent_atfork_prepare(void)
+{
+       struct tevent_context *ev;
+       int ret;
+
+       ret = pthread_mutex_lock(&tevent_contexts_mutex);
+       if (ret != 0) {
+               abort();
+       }
+
+       for (ev = tevent_contexts; ev != NULL; ev = ev->next) {
+               ret = pthread_mutex_lock(&ev->scheduled_mutex);
+               if (ret != 0) {
+                       tevent_abort(ev, "pthread_mutex_lock failed");
+               }
+       }
+}
+
+static void tevent_atfork_parent(void)
+{
+       struct tevent_context *ev;
+       int ret;
+
+       for (ev = DLIST_TAIL(tevent_contexts); ev != NULL;
+            ev = DLIST_PREV(ev)) {
+               ret = pthread_mutex_unlock(&ev->scheduled_mutex);
+               if (ret != 0) {
+                       tevent_abort(ev, "pthread_mutex_unlock failed");
+               }
+       }
+
+       ret = pthread_mutex_unlock(&tevent_contexts_mutex);
+       if (ret != 0) {
+               abort();
+       }
+}
+
+static void tevent_atfork_child(void)
+{
+       struct tevent_context *ev;
+       int ret;
+
+       for (ev = DLIST_TAIL(tevent_contexts); ev != NULL;
+            ev = DLIST_PREV(ev)) {
+               struct tevent_threaded_context *tctx;
+
+               for (tctx = ev->threaded_contexts; tctx != NULL;
+                    tctx = tctx->next) {
+                       tctx->event_ctx = NULL;
+               }
+
+               ev->threaded_contexts = NULL;
+
+               ret = pthread_mutex_unlock(&ev->scheduled_mutex);
+               if (ret != 0) {
+                       tevent_abort(ev, "pthread_mutex_unlock failed");
+               }
+       }
+
+       ret = pthread_mutex_unlock(&tevent_contexts_mutex);
+       if (ret != 0) {
+               abort();
+       }
+}
+
+static void tevent_prep_atfork(void)
+{
+       int ret;
+
+       ret = pthread_atfork(tevent_atfork_prepare,
+                            tevent_atfork_parent,
+                            tevent_atfork_child);
+       if (ret != 0) {
+               abort();
+       }
+}
+
+#endif
+
 int tevent_common_context_destructor(struct tevent_context *ev)
 {
        struct tevent_fd *fd, *fn;
@@ -146,19 +277,43 @@ int tevent_common_context_destructor(struct tevent_context *ev)
        struct tevent_immediate *ie, *in;
        struct tevent_signal *se, *sn;
 
-       if (ev->pipe_fde) {
-               talloc_free(ev->pipe_fde);
-               close(ev->pipe_fds[0]);
-               close(ev->pipe_fds[1]);
-               ev->pipe_fde = NULL;
+#ifdef HAVE_PTHREAD
+       int ret;
+
+       ret = pthread_mutex_lock(&tevent_contexts_mutex);
+       if (ret != 0) {
+               abort();
+       }
+
+       DLIST_REMOVE(tevent_contexts, ev);
+
+       ret = pthread_mutex_unlock(&tevent_contexts_mutex);
+       if (ret != 0) {
+               abort();
        }
+#endif
+
+       if (ev->threaded_contexts != NULL) {
+               /*
+                * Threaded contexts are indicators that threads are
+                * about to send us immediates via
+                * tevent_threaded_schedule_immediate. The caller
+                * needs to make sure that the tevent context lives
+                * long enough to receive immediates from all threads.
+                */
+               tevent_abort(ev, "threaded contexts exist");
+       }
+
+       tevent_common_wakeup_fini(ev);
 
        for (fd = ev->fd_events; fd; fd = fn) {
                fn = fd->next;
+               tevent_fd_set_flags(fd, 0);
                fd->event_ctx = NULL;
                DLIST_REMOVE(ev->fd_events, fd);
        }
 
+       ev->last_zero_timer = NULL;
        for (te = ev->timer_events; te; te = tn) {
                tn = te->next;
                te->event_ctx = NULL;
@@ -185,6 +340,17 @@ int tevent_common_context_destructor(struct tevent_context *ev)
                tevent_cleanup_pending_signal_handlers(se);
        }
 
+       /* removing nesting hook or we get an abort when nesting is
+        * not allowed. -- SSS
+        * Note that we need to leave the allowed flag at its current
+        * value, otherwise the use in tevent_re_initialise() will
+        * leave the event context with allowed forced to false, which
+        * will break users that expect nesting to be allowed
+        */
+       ev->nesting.level = 0;
+       ev->nesting.hook_fn = NULL;
+       ev->nesting.hook_private = NULL;
+
        return 0;
 }
 
@@ -199,8 +365,9 @@ int tevent_common_context_destructor(struct tevent_context *ev)
 
   NOTE: use tevent_context_init() inside of samba!
 */
-static struct tevent_context *tevent_context_init_ops(TALLOC_CTX *mem_ctx,
-                                                     const struct tevent_ops *ops)
+struct tevent_context *tevent_context_init_ops(TALLOC_CTX *mem_ctx,
+                                              const struct tevent_ops *ops,
+                                              void *additional_data)
 {
        struct tevent_context *ev;
        int ret;
@@ -208,9 +375,40 @@ static struct tevent_context *tevent_context_init_ops(TALLOC_CTX *mem_ctx,
        ev = talloc_zero(mem_ctx, struct tevent_context);
        if (!ev) return NULL;
 
+#ifdef HAVE_PTHREAD
+
+       ret = pthread_once(&tevent_atfork_initialized, tevent_prep_atfork);
+       if (ret != 0) {
+               talloc_free(ev);
+               return NULL;
+       }
+
+       ret = pthread_mutex_init(&ev->scheduled_mutex, NULL);
+       if (ret != 0) {
+               talloc_free(ev);
+               return NULL;
+       }
+
+       ret = pthread_mutex_lock(&tevent_contexts_mutex);
+       if (ret != 0) {
+               pthread_mutex_destroy(&ev->scheduled_mutex);
+               talloc_free(ev);
+               return NULL;
+       }
+
+       DLIST_ADD(tevent_contexts, ev);
+
+       ret = pthread_mutex_unlock(&tevent_contexts_mutex);
+       if (ret != 0) {
+               abort();
+       }
+
+#endif
+
        talloc_set_destructor(ev, tevent_common_context_destructor);
 
        ev->ops = ops;
+       ev->additional_data = additional_data;
 
        ret = ev->ops->context_init(ev);
        if (ret != 0) {
@@ -229,23 +427,14 @@ static struct tevent_context *tevent_context_init_ops(TALLOC_CTX *mem_ctx,
 struct tevent_context *tevent_context_init_byname(TALLOC_CTX *mem_ctx,
                                                  const char *name)
 {
-       struct tevent_ops_list *e;
-
-       tevent_backend_init();
+       const struct tevent_ops *ops;
 
-       if (name == NULL) {
-               name = tevent_default_backend;
-       }
-       if (name == NULL) {
-               name = "standard";
+       ops = tevent_find_ops_byname(name);
+       if (ops == NULL) {
+               return NULL;
        }
 
-       for (e=tevent_backends;e;e=e->next) {
-               if (strcmp(name, e->name) == 0) {
-                       return tevent_context_init_ops(mem_ctx, e->ops);
-               }
-       }
-       return NULL;
+       return tevent_context_init_ops(mem_ctx, ops, NULL);
 }
 
 
@@ -391,7 +580,6 @@ struct tevent_immediate *_tevent_create_immediate(TALLOC_CTX *mem_ctx,
 
 /*
   schedule an immediate event
-  return NULL on failure
 */
 void _tevent_schedule_immediate(struct tevent_immediate *im,
                                struct tevent_context *ev,
@@ -491,7 +679,9 @@ int _tevent_loop_once(struct tevent_context *ev, const char *location)
                }
        }
 
+       tevent_trace_point_callback(ev, TEVENT_TRACE_BEFORE_LOOP_ONCE);
        ret = ev->ops->loop_once(ev, location);
+       tevent_trace_point_callback(ev, TEVENT_TRACE_AFTER_LOOP_ONCE);
 
        if (ev->nesting.level > 0) {
                if (ev->nesting.hook_fn) {
@@ -551,7 +741,9 @@ int _tevent_loop_until(struct tevent_context *ev,
        }
 
        while (!finished(private_data)) {
+               tevent_trace_point_callback(ev, TEVENT_TRACE_BEFORE_LOOP_ONCE);
                ret = ev->ops->loop_once(ev, location);
+               tevent_trace_point_callback(ev, TEVENT_TRACE_AFTER_LOOP_ONCE);
                if (ret != 0) {
                        break;
                }
@@ -578,6 +770,28 @@ done:
        return ret;
 }
 
+bool tevent_common_have_events(struct tevent_context *ev)
+{
+       if (ev->fd_events != NULL) {
+               if (ev->fd_events != ev->wakeup_fde) {
+                       return true;
+               }
+               if (ev->fd_events->next != NULL) {
+                       return true;
+               }
+
+               /*
+                * At this point we just have the wakeup pipe event as
+                * the only fd_event. That one does not count as a
+                * regular event, so look at the other event types.
+                */
+       }
+
+       return ((ev->timer_events != NULL) ||
+               (ev->immediate_events != NULL) ||
+               (ev->signal_events != NULL));
+}
+
 /*
   return on failure or (with 0) if all fd events are removed
 */
@@ -587,10 +801,7 @@ int tevent_common_loop_wait(struct tevent_context *ev,
        /*
         * loop as long as we have events pending
         */
-       while (ev->fd_events ||
-              ev->timer_events ||
-              ev->immediate_events ||
-              ev->signal_events) {
+       while (tevent_common_have_events(ev)) {
                int ret;
                ret = _tevent_loop_once(ev, location);
                if (ret != 0) {
@@ -628,3 +839,102 @@ int tevent_re_initialise(struct tevent_context *ev)
 
        return ev->ops->context_init(ev);
 }
+
+static void wakeup_pipe_handler(struct tevent_context *ev,
+                               struct tevent_fd *fde,
+                               uint16_t flags, void *_private)
+{
+       ssize_t ret;
+
+       do {
+               /*
+                * This is the boilerplate for eventfd, but it works
+                * for pipes too. And as we don't care about the data
+                * we read, we're fine.
+                */
+               uint64_t val;
+               ret = read(fde->fd, &val, sizeof(val));
+       } while (ret == -1 && errno == EINTR);
+}
+
+/*
+ * Initialize the wakeup pipe and pipe fde
+ */
+
+int tevent_common_wakeup_init(struct tevent_context *ev)
+{
+       int ret;
+
+       if (ev->wakeup_fde != NULL) {
+               return 0;
+       }
+
+#ifdef HAVE_EVENTFD
+       ret = eventfd(0, EFD_NONBLOCK);
+       if (ret == -1) {
+               return errno;
+       }
+       ev->wakeup_fd = ret;
+#else
+       {
+               int pipe_fds[2];
+               ret = pipe(pipe_fds);
+               if (ret == -1) {
+                       return errno;
+               }
+               ev->wakeup_fd = pipe_fds[0];
+               ev->wakeup_write_fd = pipe_fds[1];
+
+               ev_set_blocking(ev->wakeup_fd, false);
+               ev_set_blocking(ev->wakeup_write_fd, false);
+       }
+#endif
+
+       ev->wakeup_fde = tevent_add_fd(ev, ev, ev->wakeup_fd,
+                                    TEVENT_FD_READ,
+                                    wakeup_pipe_handler, NULL);
+       if (ev->wakeup_fde == NULL) {
+               close(ev->wakeup_fd);
+#ifndef HAVE_EVENTFD
+               close(ev->wakeup_write_fd);
+#endif
+               return ENOMEM;
+       }
+
+       return 0;
+}
+
+int tevent_common_wakeup(struct tevent_context *ev)
+{
+       ssize_t ret;
+
+       if (ev->wakeup_fde == NULL) {
+               return ENOTCONN;
+       }
+
+       do {
+#ifdef HAVE_EVENTFD
+               uint64_t val = 1;
+               ret = write(ev->wakeup_fd, &val, sizeof(val));
+#else
+               char c = '\0';
+               ret = write(ev->wakeup_write_fd, &c, 1);
+#endif
+       } while ((ret == -1) && (errno == EINTR));
+
+       return 0;
+}
+
+static void tevent_common_wakeup_fini(struct tevent_context *ev)
+{
+       if (ev->wakeup_fde == NULL) {
+               return;
+       }
+
+       TALLOC_FREE(ev->wakeup_fde);
+
+       close(ev->wakeup_fd);
+#ifndef HAVE_EVENTFD
+       close(ev->wakeup_write_fd);
+#endif
+}