--- /dev/null
+/*
+ Unix SMB/CIFS implementation.
+
+ main select loop and event handling - kqueue implementation
+
+ Copyright (C) Stefan Metzmacher 2013
+
+ ** NOTE! The following LGPL license applies to the tevent
+ ** library. This does NOT imply that all of Samba is released
+ ** under the LGPL
+
+ This library is free software; you can redistribute it and/or
+ modify it under the terms of the GNU Lesser General Public
+ License as published by the Free Software Foundation; either
+ version 3 of the License, or (at your option) any later version.
+
+ This library is distributed in the hope that it will be useful,
+ but WITHOUT ANY WARRANTY; without even the implied warranty of
+ MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
+ Lesser General Public License for more details.
+
+ You should have received a copy of the GNU Lesser General Public
+ License along with this library; if not, see <http://www.gnu.org/licenses/>.
+*/
+
+#include "replace.h"
+#include "system/filesys.h"
+#include "system/time.h"
+#ifdef HAVE_SYS_EVENT_H
+#include <sys/event.h>
+#endif
+#include "tevent.h"
+#include "tevent_internal.h"
+#include "tevent_util.h"
+
+struct kqueue_event_context {
+ /* a pointer back to the generic event_context */
+ struct tevent_context *ev;
+
+ pid_t pid;
+ int kqueue_fd;
+
+ bool panic_force_replay;
+ bool *panic_state;
+ bool (*panic_fallback)(struct tevent_context *ev, bool replay);
+};
+
+#define KQUEUE_ADDITIONAL_FD_FLAG_HAS_READ (1<<0)
+#define KQUEUE_ADDITIONAL_FD_FLAG_DISABLED_READ (1<<1)
+#define KQUEUE_ADDITIONAL_FD_FLAG_HAS_WRITE (1<<2)
+#define KQUEUE_ADDITIONAL_FD_FLAG_DISABLED_WRITE (1<<3)
+#define KQUEUE_ADDITIONAL_FD_FLAG_REPORT_ERROR (1<<4)
+#define KQUEUE_ADDITIONAL_FD_FLAG_GOT_ERROR (1<<5)
+
+/*
+ called to set the panic fallback function.
+*/
+_PRIVATE_ bool tevent_kqueue_set_panic_fallback(struct tevent_context *ev,
+ bool (*panic_fallback)(struct tevent_context *ev,
+ bool replay))
+{
+ struct kqueue_event_context *kqueue_ev;
+
+ if (ev->additional_data == NULL) {
+ return false;
+ }
+
+ kqueue_ev = talloc_get_type(ev->additional_data,
+ struct kqueue_event_context);
+ if (kqueue_ev == NULL) {
+ return false;
+ }
+ kqueue_ev->panic_fallback = panic_fallback;
+ return true;
+}
+
+/*
+ called when a epoll call fails
+*/
+static void kqueue_panic(struct kqueue_event_context *kqueue_ev,
+ const char *reason, bool replay)
+{
+ struct tevent_context *ev = kqueue_ev->ev;
+ bool (*panic_fallback)(struct tevent_context *ev, bool replay);
+
+ panic_fallback = kqueue_ev->panic_fallback;
+
+ if (kqueue_ev->panic_state != NULL) {
+ *kqueue_ev->panic_state = true;
+ }
+
+ if (kqueue_ev->panic_force_replay) {
+ replay = true;
+ }
+
+ TALLOC_FREE(ev->additional_data);
+
+ if (panic_fallback == NULL) {
+ tevent_debug(ev, TEVENT_DEBUG_FATAL,
+ "%s (%s) replay[%u] - calling abort()\n",
+ reason, strerror(errno), (unsigned)replay);
+ abort();
+ }
+
+ tevent_debug(ev, TEVENT_DEBUG_ERROR,
+ "%s (%s) replay[%u] - calling panic_fallback\n",
+ reason, strerror(errno), (unsigned)replay);
+
+ if (!panic_fallback(ev, replay)) {
+ /* Fallback failed. */
+ tevent_debug(ev, TEVENT_DEBUG_FATAL,
+ "%s (%s) replay[%u] - calling abort()\n",
+ reason, strerror(errno), (unsigned)replay);
+ abort();
+ }
+}
+
+static int kqueue_ctx_destructor(struct kqueue_event_context *kqueue_ev)
+{
+ kqueue_ev->ev->additional_data = NULL;
+
+ if (kqueue_ev->kqueue_fd != -1) {
+ close(kqueue_ev->kqueue_fd);
+ }
+ kqueue_ev->kqueue_fd = -1;
+
+ return 0;
+}
+
+static int kqueue_event_context_init(struct tevent_context *ev)
+{
+ struct kqueue_event_context *kqueue_ev;
+
+ /*
+ * We might be called during tevent_re_initialise()
+ * which means we need to free our old additional_data.
+ */
+ TALLOC_FREE(ev->additional_data);
+
+ kqueue_ev = talloc_zero(ev, struct kqueue_event_context);
+ if (kqueue_ev == NULL) {
+ return -1;
+ }
+ kqueue_ev->ev = ev;
+ kqueue_ev->pid = getpid();
+ kqueue_ev->kqueue_fd = -1;
+
+ talloc_set_destructor(kqueue_ev, kqueue_ctx_destructor);
+
+ kqueue_ev->kqueue_fd = kqueue();
+ if (kqueue_ev->kqueue_fd == -1) {
+ TALLOC_FREE(kqueue_ev);
+ tevent_debug(ev, TEVENT_DEBUG_FATAL,
+ "Failed to create kqueue() handle.\n");
+ return -1;
+ }
+
+ if (!ev_set_close_on_exec(kqueue_ev->kqueue_fd)) {
+ tevent_debug(kqueue_ev->ev, TEVENT_DEBUG_WARNING,
+ "kqueue_fd pid[%d] - failed to set close-on-exec, "
+ "file descriptor may be leaked to children.\n",
+ kqueue_ev->pid);
+ }
+
+ ev->additional_data = kqueue_ev;
+ return 0;
+}
+
+static void kqueue_update_fd_event(struct kqueue_event_context *kqueue_ev,
+ struct tevent_fd *fde);
+
+static void kqueue_check_reopen(struct kqueue_event_context *kqueue_ev)
+{
+ struct tevent_fd *fde;
+ bool *caller_panic_state = kqueue_ev->panic_state;
+ bool panic_triggered = false;
+
+ if (kqueue_ev->pid == getpid()) {
+ return;
+ }
+ kqueue_ev->pid = getpid();
+
+ close(kqueue_ev->kqueue_fd);
+ kqueue_ev->kqueue_fd = kqueue();
+ if (kqueue_ev->kqueue_fd == -1) {
+ kqueue_panic(kqueue_ev, "kqueue() failed", false);
+ return;
+ }
+
+ if (!ev_set_close_on_exec(kqueue_ev->kqueue_fd)) {
+ tevent_debug(kqueue_ev->ev, TEVENT_DEBUG_WARNING,
+ "kqueue_fd pid[%d] - failed to set close-on-exec, "
+ "file descriptor may be leaked to children.\n",
+ kqueue_ev->pid);
+ }
+
+ kqueue_ev->panic_state = &panic_triggered;
+ for (fde=kqueue_ev->ev->fd_events;fde;fde=fde->next) {
+ fde->additional_flags &= KQUEUE_ADDITIONAL_FD_FLAG_GOT_ERROR;
+ kqueue_update_fd_event(kqueue_ev, fde);
+
+ if (panic_triggered) {
+ if (caller_panic_state != NULL) {
+ *caller_panic_state = true;
+ }
+ return;
+ }
+ }
+ kqueue_ev->panic_state = NULL;
+}
+
+static void kqueue_update_fd_event(struct kqueue_event_context *kqueue_ev,
+ struct tevent_fd *fde)
+{
+ bool add_read = false;
+ bool enable_read = false;
+ bool disable_read = false;
+ bool delete_read = false;
+ bool add_write = false;
+ bool enable_write = false;
+ bool disable_write = false;
+ bool delete_write = false;
+ bool got_error = (fde->additional_flags & KQUEUE_ADDITIONAL_FD_FLAG_GOT_ERROR);
+ struct kevent kev;
+ int ret;
+
+ if (fde->additional_flags & KQUEUE_ADDITIONAL_FD_FLAG_DISABLED_WRITE) {
+ if (got_error) {
+ delete_write = true;
+ } else if (fde->flags & TEVENT_FD_WRITE) {
+ enable_write = true;
+ } else if (fde->flags == 0) {
+ delete_write = true;
+ }
+ } else if (fde->additional_flags & KQUEUE_ADDITIONAL_FD_FLAG_HAS_WRITE) {
+ if (got_error) {
+ delete_write = true;
+ } else if (fde->flags == 0) {
+ delete_write = true;
+ } else if (!(fde->flags & TEVENT_FD_WRITE)) {
+ disable_write = true;
+ }
+ } else {
+ if (got_error) {
+ /* nothing */
+ } else if (fde->flags & TEVENT_FD_WRITE) {
+ add_write = true;
+ }
+ }
+
+ if (fde->additional_flags & KQUEUE_ADDITIONAL_FD_FLAG_DISABLED_READ) {
+ if (fde->flags & TEVENT_FD_READ) {
+ enable_read = true;
+ } else if (fde->flags == 0) {
+ delete_read = true;
+ }
+ } else if (fde->additional_flags & KQUEUE_ADDITIONAL_FD_FLAG_HAS_READ) {
+ if (fde->flags == 0) {
+ delete_read = true;
+ } else if (!(fde->flags & TEVENT_FD_READ)) {
+ disable_read = true;
+ }
+ } else {
+ if (fde->flags & TEVENT_FD_READ) {
+ add_read = true;
+ }
+ }
+
+ if (add_write) {
+ EV_SET(&kev, fde->fd, EVFILT_WRITE, EV_ADD,
+ 0 /* fflags */, NULL /* data */,
+ fde);
+ ret = kevent(kqueue_ev->kqueue_fd, &kev, 1, NULL, 0, NULL);
+ if (ret != 0) {
+ kqueue_panic(kqueue_ev, "EVFILT_WRITE EV_ADD failed",
+ false);
+ return;
+ }
+
+ fde->additional_flags |= KQUEUE_ADDITIONAL_FD_FLAG_HAS_WRITE;
+ } else if (enable_write) {
+ EV_SET(&kev, fde->fd, EVFILT_WRITE, EV_ENABLE,
+ 0 /* fflags */, NULL /* data */,
+ fde);
+ ret = kevent(kqueue_ev->kqueue_fd, &kev, 1, NULL, 0, NULL);
+ if (ret != 0) {
+ kqueue_panic(kqueue_ev, "EVFILT_WRITE EV_ENABLE failed",
+ false);
+ return;
+ }
+
+ fde->additional_flags &= ~KQUEUE_ADDITIONAL_FD_FLAG_DISABLED_WRITE;
+ } else if (disable_write) {
+ EV_SET(&kev, fde->fd, EVFILT_WRITE, EV_DISABLE,
+ 0 /* fflags */, NULL /* data */,
+ fde);
+ ret = kevent(kqueue_ev->kqueue_fd, &kev, 1, NULL, 0, NULL);
+ if (ret != 0) {
+ kqueue_panic(kqueue_ev, "EVFILT_WRITE EV_DISABLE failed",
+ false);
+ return;
+ }
+
+ fde->additional_flags |= KQUEUE_ADDITIONAL_FD_FLAG_DISABLED_WRITE;
+ } else if (delete_write) {
+ EV_SET(&kev, fde->fd, EVFILT_WRITE, EV_DELETE,
+ 0 /* fflags */, NULL /* data */,
+ fde);
+ ret = kevent(kqueue_ev->kqueue_fd, &kev, 1, NULL, 0, NULL);
+ if (ret != 0) {
+ kqueue_panic(kqueue_ev, "EVFILT_WRITE EV_DEL failed",
+ false);
+ return;
+ }
+
+ fde->additional_flags &= ~KQUEUE_ADDITIONAL_FD_FLAG_HAS_WRITE;
+ fde->additional_flags &= ~KQUEUE_ADDITIONAL_FD_FLAG_DISABLED_WRITE;
+ }
+
+ if (add_read) {
+ EV_SET(&kev, fde->fd, EVFILT_READ, EV_ADD,
+ 0 /* fflags */, NULL /* data */,
+ fde);
+ ret = kevent(kqueue_ev->kqueue_fd, &kev, 1, NULL, 0, NULL);
+ if (ret != 0) {
+ kqueue_panic(kqueue_ev, "EVFILT_READ EV_ADD failed",
+ false);
+ return;
+ }
+
+ fde->additional_flags |= KQUEUE_ADDITIONAL_FD_FLAG_HAS_READ;
+ fde->additional_flags |= KQUEUE_ADDITIONAL_FD_FLAG_REPORT_ERROR;
+ } else if (enable_read) {
+ EV_SET(&kev, fde->fd, EVFILT_READ, EV_ENABLE,
+ 0 /* fflags */, NULL /* data */,
+ fde);
+ ret = kevent(kqueue_ev->kqueue_fd, &kev, 1, NULL, 0, NULL);
+ if (ret != 0) {
+ kqueue_panic(kqueue_ev, "EVFILT_READ EV_ENABLE failed",
+ false);
+ return;
+ }
+
+ fde->additional_flags &= ~KQUEUE_ADDITIONAL_FD_FLAG_DISABLED_READ;
+ fde->additional_flags |= KQUEUE_ADDITIONAL_FD_FLAG_REPORT_ERROR;
+ } else if (disable_read) {
+ EV_SET(&kev, fde->fd, EVFILT_READ, EV_DISABLE,
+ 0 /* fflags */, NULL /* data */,
+ fde);
+ ret = kevent(kqueue_ev->kqueue_fd, &kev, 1, NULL, 0, NULL);
+ if (ret != 0) {
+ kqueue_panic(kqueue_ev, "EVFILT_READ EV_DISABLE failed",
+ false);
+ return;
+ }
+
+ fde->additional_flags |= KQUEUE_ADDITIONAL_FD_FLAG_DISABLED_READ;
+ fde->additional_flags &= ~KQUEUE_ADDITIONAL_FD_FLAG_REPORT_ERROR;
+ } else if (delete_read) {
+ EV_SET(&kev, fde->fd, EVFILT_READ, EV_DELETE,
+ 0 /* fflags */, NULL /* data */,
+ fde);
+ ret = kevent(kqueue_ev->kqueue_fd, &kev, 1, NULL, 0, NULL);
+ if (ret != 0) {
+ kqueue_panic(kqueue_ev, "EVFILT_READ EV_DEL failed",
+ false);
+ return;
+ }
+
+ fde->additional_flags &= ~KQUEUE_ADDITIONAL_FD_FLAG_HAS_READ;
+ fde->additional_flags &= ~KQUEUE_ADDITIONAL_FD_FLAG_DISABLED_READ;
+ fde->additional_flags &= ~KQUEUE_ADDITIONAL_FD_FLAG_REPORT_ERROR;
+ }
+}
+
+/*
+ destroy an fd_event
+*/
+static int kqueue_event_fd_destructor(struct tevent_fd *fde)
+{
+ struct tevent_context *ev = fde->event_ctx;
+ struct kqueue_event_context *kqueue_ev = NULL;
+ bool panic_triggered = false;
+ int flags = fde->flags;
+
+ if (ev == NULL) {
+ return tevent_common_fd_destructor(fde);
+ }
+
+ kqueue_ev = talloc_get_type_abort(ev->additional_data,
+ struct kqueue_event_context);
+
+ /*
+ * we must remove the event from the list
+ * otherwise a panic fallback handler may
+ * reuse invalid memory
+ */
+ DLIST_REMOVE(ev->fd_events, fde);
+
+ kqueue_ev->panic_state = &panic_triggered;
+ kqueue_check_reopen(kqueue_ev);
+ if (panic_triggered) {
+ return tevent_common_fd_destructor(fde);
+ }
+
+ fde->flags = 0;
+ kqueue_update_fd_event(kqueue_ev, fde);
+ fde->flags = flags;
+ if (panic_triggered) {
+ return tevent_common_fd_destructor(fde);
+ }
+ kqueue_ev->panic_state = NULL;
+
+ return tevent_common_fd_destructor(fde);
+}
+
+static struct tevent_fd *kqueue_event_add_fd(struct tevent_context *ev, TALLOC_CTX *mem_ctx,
+ int fd, uint16_t flags,
+ tevent_fd_handler_t handler,
+ void *private_data,
+ const char *handler_name,
+ const char *location)
+{
+ struct kqueue_event_context *kqueue_ev =
+ talloc_get_type_abort(ev->additional_data,
+ struct kqueue_event_context);
+ struct tevent_fd *fde;
+ bool panic_triggered = false;
+
+ fde = tevent_common_add_fd(ev, mem_ctx, fd, flags,
+ handler, private_data,
+ handler_name, location);
+ if (fde == NULL) {
+ return NULL;
+ }
+ talloc_set_destructor(fde, kqueue_event_fd_destructor);
+
+ kqueue_ev->panic_state = &panic_triggered;
+ kqueue_check_reopen(kqueue_ev);
+ if (panic_triggered) {
+ return fde;
+ }
+ kqueue_ev->panic_state = NULL;
+
+ kqueue_update_fd_event(kqueue_ev, fde);
+
+ return fde;
+}
+
+/*
+ set the fd event flags
+*/
+static void kqueue_event_set_fd_flags(struct tevent_fd *fde, uint16_t flags)
+{
+ struct tevent_context *ev;
+ struct kqueue_event_context *kqueue_ev;
+ bool panic_triggered = false;
+
+ if (fde->flags == flags) return;
+
+ ev = fde->event_ctx;
+ kqueue_ev = talloc_get_type_abort(ev->additional_data,
+ struct kqueue_event_context);
+
+ fde->flags = flags;
+
+ kqueue_ev->panic_state = &panic_triggered;
+ kqueue_check_reopen(kqueue_ev);
+ if (panic_triggered) {
+ return;
+ }
+ kqueue_ev->panic_state = NULL;
+
+ kqueue_update_fd_event(kqueue_ev, fde);
+}
+
+/*
+ event loop handling using kqueue
+*/
+static int kqueue_event_loop(struct kqueue_event_context *kqueue_ev, struct timeval *tvalp)
+{
+ int ret, i;
+#define MAXEVENTS 1
+ struct kevent kevs[MAXEVENTS];
+ struct timespec _ts;
+ const struct timespec *timeout = NULL;
+ int kevent_errno;
+
+ if (kqueue_ev->ev->signal_events &&
+ tevent_common_check_signal(kqueue_ev->ev)) {
+ return 0;
+ }
+
+ if (tvalp) {
+ _ts.tv_sec = tvalp->tv_sec;
+ _ts.tv_nsec = tvalp->tv_usec * 1000;
+ timeout = &_ts;
+ }
+
+ tevent_trace_point_callback(kqueue_ev->ev, TEVENT_TRACE_BEFORE_WAIT);
+ ret = kevent(kqueue_ev->kqueue_fd, NULL, 0, kevs, MAXEVENTS, timeout);
+ kevent_errno = errno;
+ tevent_trace_point_callback(kqueue_ev->ev, TEVENT_TRACE_AFTER_WAIT);
+
+ if (ret == -1 && kevent_errno == EINTR && kqueue_ev->ev->signal_events) {
+ if (tevent_common_check_signal(kqueue_ev->ev)) {
+ return 0;
+ }
+ }
+
+ if (ret == -1 && kevent_errno != EINTR) {
+ kqueue_panic(kqueue_ev, "kevent() failed", true);
+ return -1;
+ }
+
+ if (ret == 0 && tvalp) {
+ /* we don't care about a possible delay here */
+ tevent_common_loop_timer_delay(kqueue_ev->ev);
+ return 0;
+ }
+
+ for (i=0;i<ret;i++) {
+ struct tevent_fd *fde = talloc_get_type(kevs[i].udata,
+ struct tevent_fd);
+ uint16_t flags = 0;
+
+ if (fde == NULL) {
+ kqueue_panic(kqueue_ev, "kevent() gave bad data",
+ true);
+ return -1;
+ }
+
+ if (kevs[i].flags & EV_EOF) {
+ fde->additional_flags |= KQUEUE_ADDITIONAL_FD_FLAG_GOT_ERROR;
+ /*
+ * if we only wait for TEVENT_FD_WRITE, we should not
+ * tell the event handler about it, and remove the
+ * EVFILT_WRITE filter, as we only report errors when
+ * waiting for read events, to match the select()
+ * behavior
+ */
+ if (!(fde->additional_flags & KQUEUE_ADDITIONAL_FD_FLAG_REPORT_ERROR)) {
+ kqueue_update_fd_event(kqueue_ev, fde);
+ continue;
+ }
+ flags |= TEVENT_FD_READ;
+ }
+ if (kevs[i].filter == EVFILT_READ) {
+ flags |= TEVENT_FD_READ;
+ }
+ if (kevs[i].filter == EVFILT_WRITE) {
+ flags |= TEVENT_FD_WRITE;
+ }
+
+ /*
+ * make sure we only pass the flags
+ * the handler is expecting.
+ */
+ flags &= fde->flags;
+ if (flags) {
+ fde->handler(kqueue_ev->ev, fde, flags, fde->private_data);
+ break;
+ }
+ }
+
+ return 0;
+}
+
+/*
+ do a single event loop using the events defined in ev
+*/
+static int kqueue_event_loop_once(struct tevent_context *ev, const char *location)
+{
+ struct kqueue_event_context *kqueue_ev = talloc_get_type(ev->additional_data,
+ struct kqueue_event_context);
+ struct timeval tval;
+ bool panic_triggered = false;
+
+ if (ev->signal_events &&
+ tevent_common_check_signal(ev)) {
+ return 0;
+ }
+
+ if (ev->immediate_events &&
+ tevent_common_loop_immediate(ev)) {
+ return 0;
+ }
+
+ tval = tevent_common_loop_timer_delay(ev);
+ if (tevent_timeval_is_zero(&tval)) {
+ return 0;
+ }
+
+ kqueue_ev->panic_state = &panic_triggered;
+ kqueue_ev->panic_force_replay = true;
+ kqueue_check_reopen(kqueue_ev);
+ if (panic_triggered) {
+ errno = EINVAL;
+ return -1;
+ }
+ kqueue_ev->panic_force_replay = false;
+ kqueue_ev->panic_state = NULL;
+
+ return kqueue_event_loop(kqueue_ev, &tval);
+}
+
+static const struct tevent_ops kqueue_event_ops = {
+ .context_init = kqueue_event_context_init,
+ .add_fd = kqueue_event_add_fd,
+ .set_fd_close_fn = tevent_common_fd_set_close_fn,
+ .get_fd_flags = tevent_common_fd_get_flags,
+ .set_fd_flags = kqueue_event_set_fd_flags,
+ .add_timer = tevent_common_add_timer_v2,
+ .schedule_immediate = tevent_common_schedule_immediate,
+ .add_signal = tevent_common_add_signal,
+ .loop_once = kqueue_event_loop_once,
+ .loop_wait = tevent_common_loop_wait,
+};
+
+_PRIVATE_ bool tevent_kqueue_init(void)
+{
+ return tevent_register_backend("kqueue", &kqueue_event_ops);
+}
git.samba.org / metze / samba / wip.git / commitdiff