Changeset 988 for vendor/current/lib/tevent/testsuite.c

Timestamp:

Nov 24, 2016, 1:14:11 PM (9 years ago)

Author:

Silvan Scherrer

Message:

Samba Server: update vendor to version 4.4.3

File:

• vendor/current/lib/tevent/testsuite.c (modified) (11 diffs)

vendor/current/lib/tevent/testsuite.c

@@ -5 +5 @@
 
    Copyright (C) Stefan Metzmacher 2006-2009
+   Copyright (C) Jeremy Allison    2013
 
      ** NOTE! The following LGPL license applies to the tevent
@@ -25 +26 @@
 
 #include "includes.h"
-#include "lib/events/events.h"
+#include "lib/tevent/tevent.h"
 #include "system/filesys.h"
+#include "system/select.h"
+#include "system/network.h"
 #include "torture/torture.h"
+#include "torture/local/proto.h"
+#ifdef HAVE_PTHREAD
+#include <pthread.h>
+#include <assert.h>
+#endif
 
 static int fde_count;
 
-static void fde_handler(struct tevent_context *ev_ctx, struct tevent_fd *f, 
+static void fde_handler_read(struct tevent_context *ev_ctx, struct tevent_fd *f,
                         uint16_t flags, void *private_data)
 {
@@ -40 +48 @@
 #endif
         kill(getpid(), SIGALRM);
+
         read(fd[0], &c, 1);
+        fde_count++;
+}
+
+static void fde_handler_write(struct tevent_context *ev_ctx, struct tevent_fd *f,
+                        uint16_t flags, void *private_data)
+{
+        int *fd = (int *)private_data;
+        char c = 0;
         write(fd[1], &c, 1);
+}
+
+
+/* This will only fire if the fd's returned from pipe() are bi-directional. */
+static void fde_handler_read_1(struct tevent_context *ev_ctx, struct tevent_fd *f,
+                        uint16_t flags, void *private_data)
+{
+        int *fd = (int *)private_data;
+        char c;
+#ifdef SA_SIGINFO
+        kill(getpid(), SIGUSR1);
+#endif
+        kill(getpid(), SIGALRM);
+
+        read(fd[1], &c, 1);
         fde_count++;
+}
+
+/* This will only fire if the fd's returned from pipe() are bi-directional. */
+static void fde_handler_write_1(struct tevent_context *ev_ctx, struct tevent_fd *f,
+                        uint16_t flags, void *private_data)
+{
+        int *fd = (int *)private_data;
+        char c = 0;
+        write(fd[0], &c, 1);
 }
 
@@ -52 +93 @@
 }
 
-static void count_handler(struct tevent_context *ev_ctx, struct signal_event *te,
+static void count_handler(struct tevent_context *ev_ctx, struct tevent_signal *te,
                           int signum, int count, void *info, void *private_data)
 {
@@ -66 +107 @@
         const char *backend = (const char *)test_data;
         int alarm_count=0, info_count=0;
-        struct tevent_fd *fde;
+        struct tevent_fd *fde_read;
+        struct tevent_fd *fde_read_1;
+        struct tevent_fd *fde_write;
+        struct tevent_fd *fde_write_1;
 #ifdef SA_RESTART
         struct tevent_signal *se1 = NULL;
 #endif
+#ifdef SA_RESETHAND
         struct tevent_signal *se2 = NULL;
+#endif
 #ifdef SA_SIGINFO
         struct tevent_signal *se3 = NULL;
@@ -76 +122 @@
         int finished=0;
         struct timeval t;
-        char c = 0;
-
-        ev_ctx = event_context_init_byname(test, backend);
+        int ret;
+
+        ev_ctx = tevent_context_init_byname(test, backend);
         if (ev_ctx == NULL) {
                 torture_comment(test, "event backend '%s' not supported\n", backend);
@@ -84 +130 @@
         }
 
-        torture_comment(test, "Testing event backend '%s'\n", backend);
+        torture_comment(test, "backend '%s' - %s\n",
+                        backend, __FUNCTION__);
 
         /* reset globals */
@@ -90 +137 @@
 
         /* create a pipe */
-        pipe(fd);
-
-        fde = event_add_fd(ev_ctx, ev_ctx, fd[0], EVENT_FD_READ,
-                           fde_handler, fd);
-        tevent_fd_set_auto_close(fde);
-
-        event_add_timed(ev_ctx, ev_ctx, timeval_current_ofs(2,0), 
-                        finished_handler, &finished);
+        ret = pipe(fd);
+        torture_assert_int_equal(test, ret, 0, "pipe failed");
+
+        fde_read = tevent_add_fd(ev_ctx, ev_ctx, fd[0], TEVENT_FD_READ,
+                            fde_handler_read, fd);
+        fde_write_1 = tevent_add_fd(ev_ctx, ev_ctx, fd[0], TEVENT_FD_WRITE,
+                            fde_handler_write_1, fd);
+
+        fde_write = tevent_add_fd(ev_ctx, ev_ctx, fd[1], TEVENT_FD_WRITE,
+                            fde_handler_write, fd);
+        fde_read_1 = tevent_add_fd(ev_ctx, ev_ctx, fd[1], TEVENT_FD_READ,
+                            fde_handler_read_1, fd);
+
+        tevent_fd_set_auto_close(fde_read);
+        tevent_fd_set_auto_close(fde_write);
+
+        tevent_add_timer(ev_ctx, ev_ctx, timeval_current_ofs(2,0),
+                         finished_handler, &finished);
 
 #ifdef SA_RESTART
-        se1 = event_add_signal(ev_ctx, ev_ctx, SIGALRM, SA_RESTART, count_handler, &alarm_count);
+        se1 = tevent_add_signal(ev_ctx, ev_ctx, SIGALRM, SA_RESTART, count_handler, &alarm_count);
+        torture_assert(test, se1 != NULL, "failed to setup se1");
 #endif
 #ifdef SA_RESETHAND
-        se2 = event_add_signal(ev_ctx, ev_ctx, SIGALRM, SA_RESETHAND, count_handler, &alarm_count);
+        se2 = tevent_add_signal(ev_ctx, ev_ctx, SIGALRM, SA_RESETHAND, count_handler, &alarm_count);
+        torture_assert(test, se2 != NULL, "failed to setup se2");
 #endif
 #ifdef SA_SIGINFO
-        se3 = event_add_signal(ev_ctx, ev_ctx, SIGUSR1, SA_SIGINFO, count_handler, &info_count);
-#endif
-
-        write(fd[1], &c, 1);
+        se3 = tevent_add_signal(ev_ctx, ev_ctx, SIGUSR1, SA_SIGINFO, count_handler, &info_count);
+        torture_assert(test, se3 != NULL, "failed to setup se3");
+#endif
 
         t = timeval_current();
         while (!finished) {
                 errno = 0;
-                if (event_loop_once(ev_ctx) == -1) {
+                if (tevent_loop_once(ev_ctx) == -1) {
                         talloc_free(ev_ctx);
                         torture_fail(test, talloc_asprintf(test, "Failed event loop %s\n", strerror(errno)));
@@ -120 +178 @@
         }
 
-        talloc_free(fde);
-        close(fd[1]);
+        talloc_free(fde_read_1);
+        talloc_free(fde_write_1);
+        talloc_free(fde_read);
+        talloc_free(fde_write);
 
         while (alarm_count < fde_count+1) {
-                if (event_loop_once(ev_ctx) == -1) {
+                if (tevent_loop_once(ev_ctx) == -1) {
                         break;
                 }
@@ -136 +196 @@
 
         torture_assert_int_equal(test, alarm_count, 1+fde_count, "alarm count mismatch");
+
+#ifdef SA_RESETHAND
+        /*
+         * we do not call talloc_free(se2)
+         * because it is already gone,
+         * after triggering the event handler.
+         */
+#endif
 
 #ifdef SA_SIGINFO
@@ -147 +215 @@
 }
 
+struct test_event_fd1_state {
+        struct torture_context *tctx;
+        const char *backend;
+        struct tevent_context *ev;
+        int sock[2];
+        struct tevent_timer *te;
+        struct tevent_fd *fde0;
+        struct tevent_fd *fde1;
+        bool got_write;
+        bool got_read;
+        bool drain;
+        bool drain_done;
+        unsigned loop_count;
+        bool finished;
+        const char *error;
+};
+
+static void test_event_fd1_fde_handler(struct tevent_context *ev_ctx,
+                                       struct tevent_fd *fde,
+                                       uint16_t flags,
+                                       void *private_data)
+{
+        struct test_event_fd1_state *state =
+                (struct test_event_fd1_state *)private_data;
+
+        if (state->drain_done) {
+                state->finished = true;
+                state->error = __location__;
+                return;
+        }
+
+        if (state->drain) {
+                ssize_t ret;
+                uint8_t c = 0;
+
+                if (!(flags & TEVENT_FD_READ)) {
+                        state->finished = true;
+                        state->error = __location__;
+                        return;
+                }
+
+                ret = read(state->sock[0], &c, 1);
+                if (ret == 1) {
+                        return;
+                }
+
+                /*
+                 * end of test...
+                 */
+                tevent_fd_set_flags(fde, 0);
+                state->drain_done = true;
+                return;
+        }
+
+        if (!state->got_write) {
+                uint8_t c = 0;
+
+                if (flags != TEVENT_FD_WRITE) {
+                        state->finished = true;
+                        state->error = __location__;
+                        return;
+                }
+                state->got_write = true;
+
+                /*
+                 * we write to the other socket...
+                 */
+                write(state->sock[1], &c, 1);
+                TEVENT_FD_NOT_WRITEABLE(fde);
+                TEVENT_FD_READABLE(fde);
+                return;
+        }
+
+        if (!state->got_read) {
+                if (flags != TEVENT_FD_READ) {
+                        state->finished = true;
+                        state->error = __location__;
+                        return;
+                }
+                state->got_read = true;
+
+                TEVENT_FD_NOT_READABLE(fde);
+                return;
+        }
+
+        state->finished = true;
+        state->error = __location__;
+        return;
+}
+
+static void test_event_fd1_finished(struct tevent_context *ev_ctx,
+                                    struct tevent_timer *te,
+                                    struct timeval tval,
+                                    void *private_data)
+{
+        struct test_event_fd1_state *state =
+                (struct test_event_fd1_state *)private_data;
+
+        if (state->drain_done) {
+                state->finished = true;
+                return;
+        }
+
+        if (!state->got_write) {
+                state->finished = true;
+                state->error = __location__;
+                return;
+        }
+
+        if (!state->got_read) {
+                state->finished = true;
+                state->error = __location__;
+                return;
+        }
+
+        state->loop_count++;
+        if (state->loop_count > 3) {
+                state->finished = true;
+                state->error = __location__;
+                return;
+        }
+
+        state->got_write = false;
+        state->got_read = false;
+
+        tevent_fd_set_flags(state->fde0, TEVENT_FD_WRITE);
+
+        if (state->loop_count > 2) {
+                state->drain = true;
+                TALLOC_FREE(state->fde1);
+                TEVENT_FD_READABLE(state->fde0);
+        }
+
+        state->te = tevent_add_timer(state->ev, state->ev,
+                                    timeval_current_ofs(0,2000),
+                                    test_event_fd1_finished, state);
+}
+
+static bool test_event_fd1(struct torture_context *tctx,
+                           const void *test_data)
+{
+        struct test_event_fd1_state state;
+
+        ZERO_STRUCT(state);
+        state.tctx = tctx;
+        state.backend = (const char *)test_data;
+
+        state.ev = tevent_context_init_byname(tctx, state.backend);
+        if (state.ev == NULL) {
+                torture_skip(tctx, talloc_asprintf(tctx,
+                             "event backend '%s' not supported\n",
+                             state.backend));
+                return true;
+        }
+
+        tevent_set_debug_stderr(state.ev);
+        torture_comment(tctx, "backend '%s' - %s\n",
+                        state.backend, __FUNCTION__);
+
+        /*
+         * This tests the following:
+         *
+         * It monitors the state of state.sock[0]
+         * with tevent_fd, but we never read/write on state.sock[0]
+         * while state.sock[1] * is only used to write a few bytes.
+         *
+         * We have a loop:
+         *   - we wait only for TEVENT_FD_WRITE on state.sock[0]
+         *   - we write 1 byte to state.sock[1]
+         *   - we wait only for TEVENT_FD_READ on state.sock[0]
+         *   - we disable events on state.sock[0]
+         *   - the timer event restarts the loop
+         * Then we close state.sock[1]
+         * We have a loop:
+         *   - we wait for TEVENT_FD_READ/WRITE on state.sock[0]
+         *   - we try to read 1 byte
+         *   - if the read gets an error of returns 0
+         *     we disable the event handler
+         *   - the timer finishes the test
+         */
+        state.sock[0] = -1;
+        state.sock[1] = -1;
+        socketpair(AF_UNIX, SOCK_STREAM, 0, state.sock);
+
+        state.te = tevent_add_timer(state.ev, state.ev,
+                                    timeval_current_ofs(0,1000),
+                                    test_event_fd1_finished, &state);
+        state.fde0 = tevent_add_fd(state.ev, state.ev,
+                                   state.sock[0], TEVENT_FD_WRITE,
+                                   test_event_fd1_fde_handler, &state);
+        /* state.fde1 is only used to auto close */
+        state.fde1 = tevent_add_fd(state.ev, state.ev,
+                                   state.sock[1], 0,
+                                   test_event_fd1_fde_handler, &state);
+
+        tevent_fd_set_auto_close(state.fde0);
+        tevent_fd_set_auto_close(state.fde1);
+
+        while (!state.finished) {
+                errno = 0;
+                if (tevent_loop_once(state.ev) == -1) {
+                        talloc_free(state.ev);
+                        torture_fail(tctx, talloc_asprintf(tctx,
+                                     "Failed event loop %s\n",
+                                     strerror(errno)));
+                }
+        }
+
+        talloc_free(state.ev);
+
+        torture_assert(tctx, state.error == NULL, talloc_asprintf(tctx,
+                       "%s", state.error));
+
+        return true;
+}
+
+struct test_event_fd2_state {
+        struct torture_context *tctx;
+        const char *backend;
+        struct tevent_context *ev;
+        struct tevent_timer *te;
+        struct test_event_fd2_sock {
+                struct test_event_fd2_state *state;
+                int fd;
+                struct tevent_fd *fde;
+                size_t num_written;
+                size_t num_read;
+                bool got_full;
+        } sock0, sock1;
+        bool finished;
+        const char *error;
+};
+
+static void test_event_fd2_sock_handler(struct tevent_context *ev_ctx,
+                                        struct tevent_fd *fde,
+                                        uint16_t flags,
+                                        void *private_data)
+{
+        struct test_event_fd2_sock *cur_sock =
+                (struct test_event_fd2_sock *)private_data;
+        struct test_event_fd2_state *state = cur_sock->state;
+        struct test_event_fd2_sock *oth_sock = NULL;
+        uint8_t v = 0, c;
+        ssize_t ret;
+
+        if (cur_sock == &state->sock0) {
+                oth_sock = &state->sock1;
+        } else {
+                oth_sock = &state->sock0;
+        }
+
+        if (oth_sock->num_written == 1) {
+                if (flags != (TEVENT_FD_READ | TEVENT_FD_WRITE)) {
+                        state->finished = true;
+                        state->error = __location__;
+                        return;
+                }
+        }
+
+        if (cur_sock->num_read == oth_sock->num_written) {
+                state->finished = true;
+                state->error = __location__;
+                return;
+        }
+
+        if (!(flags & TEVENT_FD_READ)) {
+                state->finished = true;
+                state->error = __location__;
+                return;
+        }
+
+        if (oth_sock->num_read >= PIPE_BUF) {
+                /*
+                 * On Linux we become writable once we've read
+                 * one byte. On Solaris we only become writable
+                 * again once we've read 4096 bytes. PIPE_BUF
+                 * is probably a safe bet to test against.
+                 *
+                 * There should be room to write a byte again
+                 */
+                if (!(flags & TEVENT_FD_WRITE)) {
+                        state->finished = true;
+                        state->error = __location__;
+                        return;
+                }
+        }
+
+        if ((flags & TEVENT_FD_WRITE) && !cur_sock->got_full) {
+                v = (uint8_t)cur_sock->num_written;
+                ret = write(cur_sock->fd, &v, 1);
+                if (ret != 1) {
+                        state->finished = true;
+                        state->error = __location__;
+                        return;
+                }
+                cur_sock->num_written++;
+                if (cur_sock->num_written > 0x80000000) {
+                        state->finished = true;
+                        state->error = __location__;
+                        return;
+                }
+                return;
+        }
+
+        if (!cur_sock->got_full) {
+                cur_sock->got_full = true;
+
+                if (!oth_sock->got_full) {
+                        /*
+                         * cur_sock is full,
+                         * lets wait for oth_sock
+                         * to be filled
+                         */
+                        tevent_fd_set_flags(cur_sock->fde, 0);
+                        return;
+                }
+
+                /*
+                 * oth_sock waited for cur_sock,
+                 * lets restart it
+                 */
+                tevent_fd_set_flags(oth_sock->fde,
+                                    TEVENT_FD_READ|TEVENT_FD_WRITE);
+        }
+
+        ret = read(cur_sock->fd, &v, 1);
+        if (ret != 1) {
+                state->finished = true;
+                state->error = __location__;
+                return;
+        }
+        c = (uint8_t)cur_sock->num_read;
+        if (c != v) {
+                state->finished = true;
+                state->error = __location__;
+                return;
+        }
+        cur_sock->num_read++;
+
+        if (cur_sock->num_read < oth_sock->num_written) {
+                /* there is more to read */
+                return;
+        }
+        /*
+         * we read everything, we need to remove TEVENT_FD_WRITE
+         * to avoid spinning
+         */
+        TEVENT_FD_NOT_WRITEABLE(cur_sock->fde);
+
+        if (oth_sock->num_read == cur_sock->num_written) {
+                /*
+                 * both directions are finished
+                 */
+                state->finished = true;
+        }
+
+        return;
+}
+
+static void test_event_fd2_finished(struct tevent_context *ev_ctx,
+                                    struct tevent_timer *te,
+                                    struct timeval tval,
+                                    void *private_data)
+{
+        struct test_event_fd2_state *state =
+                (struct test_event_fd2_state *)private_data;
+
+        /*
+         * this should never be triggered
+         */
+        state->finished = true;
+        state->error = __location__;
+}
+
+static bool test_event_fd2(struct torture_context *tctx,
+                           const void *test_data)
+{
+        struct test_event_fd2_state state;
+        int sock[2];
+        uint8_t c = 0;
+
+        ZERO_STRUCT(state);
+        state.tctx = tctx;
+        state.backend = (const char *)test_data;
+
+        state.ev = tevent_context_init_byname(tctx, state.backend);
+        if (state.ev == NULL) {
+                torture_skip(tctx, talloc_asprintf(tctx,
+                             "event backend '%s' not supported\n",
+                             state.backend));
+                return true;
+        }
+
+        tevent_set_debug_stderr(state.ev);
+        torture_comment(tctx, "backend '%s' - %s\n",
+                        state.backend, __FUNCTION__);
+
+        /*
+         * This tests the following
+         *
+         * - We write 1 byte to each socket
+         * - We wait for TEVENT_FD_READ/WRITE on both sockets
+         * - When we get TEVENT_FD_WRITE we write 1 byte
+         *   until both socket buffers are full, which
+         *   means both sockets only get TEVENT_FD_READ.
+         * - Then we read 1 byte until we have consumed
+         *   all bytes the other end has written.
+         */
+        sock[0] = -1;
+        sock[1] = -1;
+        socketpair(AF_UNIX, SOCK_STREAM, 0, sock);
+
+        /*
+         * the timer should never expire
+         */
+        state.te = tevent_add_timer(state.ev, state.ev,
+                                    timeval_current_ofs(600, 0),
+                                    test_event_fd2_finished, &state);
+        state.sock0.state = &state;
+        state.sock0.fd = sock[0];
+        state.sock0.fde = tevent_add_fd(state.ev, state.ev,
+                                        state.sock0.fd,
+                                        TEVENT_FD_READ | TEVENT_FD_WRITE,
+                                        test_event_fd2_sock_handler,
+                                        &state.sock0);
+        state.sock1.state = &state;
+        state.sock1.fd = sock[1];
+        state.sock1.fde = tevent_add_fd(state.ev, state.ev,
+                                        state.sock1.fd,
+                                        TEVENT_FD_READ | TEVENT_FD_WRITE,
+                                        test_event_fd2_sock_handler,
+                                        &state.sock1);
+
+        tevent_fd_set_auto_close(state.sock0.fde);
+        tevent_fd_set_auto_close(state.sock1.fde);
+
+        write(state.sock0.fd, &c, 1);
+        state.sock0.num_written++;
+        write(state.sock1.fd, &c, 1);
+        state.sock1.num_written++;
+
+        while (!state.finished) {
+                errno = 0;
+                if (tevent_loop_once(state.ev) == -1) {
+                        talloc_free(state.ev);
+                        torture_fail(tctx, talloc_asprintf(tctx,
+                                     "Failed event loop %s\n",
+                                     strerror(errno)));
+                }
+        }
+
+        talloc_free(state.ev);
+
+        torture_assert(tctx, state.error == NULL, talloc_asprintf(tctx,
+                       "%s", state.error));
+
+        return true;
+}
+
+#ifdef HAVE_PTHREAD
+
+static pthread_mutex_t threaded_mutex = PTHREAD_MUTEX_INITIALIZER;
+static bool do_shutdown = false;
+
+static void test_event_threaded_lock(void)
+{
+        int ret;
+        ret = pthread_mutex_lock(&threaded_mutex);
+        assert(ret == 0);
+}
+
+static void test_event_threaded_unlock(void)
+{
+        int ret;
+        ret = pthread_mutex_unlock(&threaded_mutex);
+        assert(ret == 0);
+}
+
+static void test_event_threaded_trace(enum tevent_trace_point point,
+                                      void *private_data)
+{
+        switch (point) {
+        case TEVENT_TRACE_BEFORE_WAIT:
+                test_event_threaded_unlock();
+                break;
+        case TEVENT_TRACE_AFTER_WAIT:
+                test_event_threaded_lock();
+                break;
+        case TEVENT_TRACE_BEFORE_LOOP_ONCE:
+        case TEVENT_TRACE_AFTER_LOOP_ONCE:
+                break;
+        }
+}
+
+static void test_event_threaded_timer(struct tevent_context *ev,
+                                      struct tevent_timer *te,
+                                      struct timeval current_time,
+                                      void *private_data)
+{
+        return;
+}
+
+static void *test_event_poll_thread(void *private_data)
+{
+        struct tevent_context *ev = (struct tevent_context *)private_data;
+
+        test_event_threaded_lock();
+
+        while (true) {
+                int ret;
+                ret = tevent_loop_once(ev);
+                assert(ret == 0);
+                if (do_shutdown) {
+                        test_event_threaded_unlock();
+                        return NULL;
+                }
+        }
+
+}
+
+static void test_event_threaded_read_handler(struct tevent_context *ev,
+                                             struct tevent_fd *fde,
+                                             uint16_t flags,
+                                             void *private_data)
+{
+        int *pfd = (int *)private_data;
+        char c;
+        ssize_t nread;
+
+        if ((flags & TEVENT_FD_READ) == 0) {
+                return;
+        }
+
+        do {
+                nread = read(*pfd, &c, 1);
+        } while ((nread == -1) && (errno == EINTR));
+
+        assert(nread == 1);
+}
+
+static bool test_event_context_threaded(struct torture_context *test,
+                                        const void *test_data)
+{
+        struct tevent_context *ev;
+        struct tevent_timer *te;
+        struct tevent_fd *fde;
+        pthread_t poll_thread;
+        int fds[2];
+        int ret;
+        char c = 0;
+
+        ev = tevent_context_init_byname(test, "poll_mt");
+        torture_assert(test, ev != NULL, "poll_mt not supported");
+
+        tevent_set_trace_callback(ev, test_event_threaded_trace, NULL);
+
+        te = tevent_add_timer(ev, ev, timeval_current_ofs(5, 0),
+                              test_event_threaded_timer, NULL);
+        torture_assert(test, te != NULL, "Could not add timer");
+
+        ret = pthread_create(&poll_thread, NULL, test_event_poll_thread, ev);
+        torture_assert(test, ret == 0, "Could not create poll thread");
+
+        ret = pipe(fds);
+        torture_assert(test, ret == 0, "Could not create pipe");
+
+        poll(NULL, 0, 100);
+
+        test_event_threaded_lock();
+
+        fde = tevent_add_fd(ev, ev, fds[0], TEVENT_FD_READ,
+                            test_event_threaded_read_handler, &fds[0]);
+        torture_assert(test, fde != NULL, "Could not add fd event");
+
+        test_event_threaded_unlock();
+
+        poll(NULL, 0, 100);
+
+        write(fds[1], &c, 1);
+
+        poll(NULL, 0, 100);
+
+        test_event_threaded_lock();
+        do_shutdown = true;
+        test_event_threaded_unlock();
+
+        write(fds[1], &c, 1);
+
+        ret = pthread_join(poll_thread, NULL);
+        torture_assert(test, ret == 0, "pthread_join failed");
+
+        return true;
+}
+
+#define NUM_TEVENT_THREADS 100
+
+/* Ugly, but needed for torture_comment... */
+static struct torture_context *thread_test_ctx;
+static pthread_t thread_map[NUM_TEVENT_THREADS];
+static unsigned thread_counter;
+
+/* Called in master thread context */
+static void callback_nowait(struct tevent_context *ev,
+                                struct tevent_immediate *im,
+                                void *private_ptr)
+{
+        pthread_t *thread_id_ptr =
+                talloc_get_type_abort(private_ptr, pthread_t);
+        unsigned i;
+
+        for (i = 0; i < NUM_TEVENT_THREADS; i++) {
+                if (pthread_equal(*thread_id_ptr,
+                                thread_map[i])) {
+                        break;
+                }
+        }
+        torture_comment(thread_test_ctx,
+                        "Callback %u from thread %u\n",
+                        thread_counter,
+                        i);
+        thread_counter++;
+}
+
+/* Blast the master tevent_context with a callback, no waiting. */
+static void *thread_fn_nowait(void *private_ptr)
+{
+        struct tevent_thread_proxy *master_tp =
+                talloc_get_type_abort(private_ptr, struct tevent_thread_proxy);
+        struct tevent_immediate *im;
+        pthread_t *thread_id_ptr;
+
+        im = tevent_create_immediate(NULL);
+        if (im == NULL) {
+                return NULL;
+        }
+        thread_id_ptr = talloc(NULL, pthread_t);
+        if (thread_id_ptr == NULL) {
+                return NULL;
+        }
+        *thread_id_ptr = pthread_self();
+
+        tevent_thread_proxy_schedule(master_tp,
+                                &im,
+                                callback_nowait,
+                                &thread_id_ptr);
+        return NULL;
+}
+
+static void timeout_fn(struct tevent_context *ev,
+                        struct tevent_timer *te,
+                        struct timeval tv, void *p)
+{
+        thread_counter = NUM_TEVENT_THREADS * 10;
+}
+
+static bool test_multi_tevent_threaded(struct torture_context *test,
+                                        const void *test_data)
+{
+        unsigned i;
+        struct tevent_context *master_ev;
+        struct tevent_thread_proxy *tp;
+
+        talloc_disable_null_tracking();
+
+        /* Ugly global stuff. */
+        thread_test_ctx = test;
+        thread_counter = 0;
+
+        master_ev = tevent_context_init(NULL);
+        if (master_ev == NULL) {
+                return false;
+        }
+        tevent_set_debug_stderr(master_ev);
+
+        tp = tevent_thread_proxy_create(master_ev);
+        if (tp == NULL) {
+                torture_fail(test,
+                        talloc_asprintf(test,
+                                "tevent_thread_proxy_create failed\n"));
+                talloc_free(master_ev);
+                return false;
+        }
+
+        for (i = 0; i < NUM_TEVENT_THREADS; i++) {
+                int ret = pthread_create(&thread_map[i],
+                                NULL,
+                                thread_fn_nowait,
+                                tp);
+                if (ret != 0) {
+                        torture_fail(test,
+                                talloc_asprintf(test,
+                                        "Failed to create thread %i, %d\n",
+                                        i, ret));
+                        return false;
+                }
+        }
+
+        /* Ensure we don't wait more than 10 seconds. */
+        tevent_add_timer(master_ev,
+                        master_ev,
+                        timeval_current_ofs(10,0),
+                        timeout_fn,
+                        NULL);
+
+        while (thread_counter < NUM_TEVENT_THREADS) {
+                int ret = tevent_loop_once(master_ev);
+                torture_assert(test, ret == 0, "tevent_loop_once failed");
+        }
+
+        torture_assert(test, thread_counter == NUM_TEVENT_THREADS,
+                "thread_counter fail\n");
+
+        talloc_free(master_ev);
+        return true;
+}
+
+struct reply_state {
+        struct tevent_thread_proxy *reply_tp;
+        pthread_t thread_id;
+        int *p_finished;
+};
+
+static void thread_timeout_fn(struct tevent_context *ev,
+                        struct tevent_timer *te,
+                        struct timeval tv, void *p)
+{
+        int *p_finished = (int *)p;
+
+        *p_finished = 2;
+}
+
+/* Called in child-thread context */
+static void thread_callback(struct tevent_context *ev,
+                                struct tevent_immediate *im,
+                                void *private_ptr)
+{
+        struct reply_state *rsp =
+                talloc_get_type_abort(private_ptr, struct reply_state);
+
+        talloc_steal(ev, rsp);
+        *rsp->p_finished = 1;
+}
+
+/* Called in master thread context */
+static void master_callback(struct tevent_context *ev,
+                                struct tevent_immediate *im,
+                                void *private_ptr)
+{
+        struct reply_state *rsp =
+                talloc_get_type_abort(private_ptr, struct reply_state);
+        unsigned i;
+
+        talloc_steal(ev, rsp);
+
+        for (i = 0; i < NUM_TEVENT_THREADS; i++) {
+                if (pthread_equal(rsp->thread_id,
+                                thread_map[i])) {
+                        break;
+                }
+        }
+        torture_comment(thread_test_ctx,
+                        "Callback %u from thread %u\n",
+                        thread_counter,
+                        i);
+        /* Now reply to the thread ! */
+        tevent_thread_proxy_schedule(rsp->reply_tp,
+                                &im,
+                                thread_callback,
+                                &rsp);
+
+        thread_counter++;
+}
+
+static void *thread_fn_1(void *private_ptr)
+{
+        struct tevent_thread_proxy *master_tp =
+                talloc_get_type_abort(private_ptr, struct tevent_thread_proxy);
+        struct tevent_thread_proxy *tp;
+        struct tevent_immediate *im;
+        struct tevent_context *ev;
+        struct reply_state *rsp;
+        int finished = 0;
+        int ret;
+
+        ev = tevent_context_init(NULL);
+        if (ev == NULL) {
+                return NULL;
+        }
+
+        tp = tevent_thread_proxy_create(ev);
+        if (tp == NULL) {
+                talloc_free(ev);
+                return NULL;
+        }
+
+        im = tevent_create_immediate(ev);
+        if (im == NULL) {
+                talloc_free(ev);
+                return NULL;
+        }
+
+        rsp = talloc(ev, struct reply_state);
+        if (rsp == NULL) {
+                talloc_free(ev);
+                return NULL;
+        }
+
+        rsp->thread_id = pthread_self();
+        rsp->reply_tp = tp;
+        rsp->p_finished = &finished;
+
+        /* Introduce a little randomness into the mix.. */
+        usleep(random() % 7000);
+
+        tevent_thread_proxy_schedule(master_tp,
+                                &im,
+                                master_callback,
+                                &rsp);
+
+        /* Ensure we don't wait more than 10 seconds. */
+        tevent_add_timer(ev,
+                        ev,
+                        timeval_current_ofs(10,0),
+                        thread_timeout_fn,
+                        &finished);
+
+        while (finished == 0) {
+                ret = tevent_loop_once(ev);
+                assert(ret == 0);
+        }
+
+        if (finished > 1) {
+                /* Timeout ! */
+                abort();
+        }
+
+        /*
+         * NB. We should talloc_free(ev) here, but if we do
+         * we currently get hit by helgrind Fix #323432
+         * "When calling pthread_cond_destroy or pthread_mutex_destroy
+         * with initializers as argument Helgrind (incorrectly) reports errors."
+         *
+         * http://valgrind.10908.n7.nabble.com/Helgrind-3-9-0-false-positive-
+         * with-pthread-mutex-destroy-td47757.html
+         *
+         * Helgrind doesn't understand that the request/reply
+         * messages provide synchronization between the lock/unlock
+         * in tevent_thread_proxy_schedule(), and the pthread_destroy()
+         * when the struct tevent_thread_proxy object is talloc_free'd.
+         *
+         * As a work-around for now return ev for the parent thread to free.
+         */
+        return ev;
+}
+
+static bool test_multi_tevent_threaded_1(struct torture_context *test,
+                                        const void *test_data)
+{
+        unsigned i;
+        struct tevent_context *master_ev;
+        struct tevent_thread_proxy *master_tp;
+        int ret;
+
+        talloc_disable_null_tracking();
+
+        /* Ugly global stuff. */
+        thread_test_ctx = test;
+        thread_counter = 0;
+
+        master_ev = tevent_context_init(NULL);
+        if (master_ev == NULL) {
+                return false;
+        }
+        tevent_set_debug_stderr(master_ev);
+
+        master_tp = tevent_thread_proxy_create(master_ev);
+        if (master_tp == NULL) {
+                torture_fail(test,
+                        talloc_asprintf(test,
+                                "tevent_thread_proxy_create failed\n"));
+                talloc_free(master_ev);
+                return false;
+        }
+
+        for (i = 0; i < NUM_TEVENT_THREADS; i++) {
+                ret = pthread_create(&thread_map[i],
+                                NULL,
+                                thread_fn_1,
+                                master_tp);
+                if (ret != 0) {
+                        torture_fail(test,
+                                talloc_asprintf(test,
+                                        "Failed to create thread %i, %d\n",
+                                        i, ret));
+                                return false;
+                }
+        }
+
+        while (thread_counter < NUM_TEVENT_THREADS) {
+                ret = tevent_loop_once(master_ev);
+                torture_assert(test, ret == 0, "tevent_loop_once failed");
+        }
+
+        /* Wait for all the threads to finish - join 'em. */
+        for (i = 0; i < NUM_TEVENT_THREADS; i++) {
+                void *retval;
+                ret = pthread_join(thread_map[i], &retval);
+                torture_assert(test, ret == 0, "pthread_join failed");
+                /* Free the child thread event context. */
+                talloc_free(retval);
+        }
+
+        talloc_free(master_ev);
+        return true;
+}
+#endif
+
 struct torture_suite *torture_local_event(TALLOC_CTX *mem_ctx)
 {
         struct torture_suite *suite = torture_suite_create(mem_ctx, "event");
-        const char **list = event_backend_list(suite);
+        const char **list = tevent_backend_list(suite);
         int i;
 
         for (i=0;list && list[i];i++) {
-                torture_suite_add_simple_tcase_const(suite, list[i],
+                struct torture_suite *backend_suite;
+
+                backend_suite = torture_suite_create(mem_ctx, list[i]);
+
+                torture_suite_add_simple_tcase_const(backend_suite,
+                                               "context",
                                                test_event_context,
                                                (const void *)list[i]);
-        }
+                torture_suite_add_simple_tcase_const(backend_suite,
+                                               "fd1",
+                                               test_event_fd1,
+                                               (const void *)list[i]);
+                torture_suite_add_simple_tcase_const(backend_suite,
+                                               "fd2",
+                                               test_event_fd2,
+                                               (const void *)list[i]);
+
+                torture_suite_add_suite(suite, backend_suite);
+        }
+
+#ifdef HAVE_PTHREAD
+        torture_suite_add_simple_tcase_const(suite, "threaded_poll_mt",
+                                             test_event_context_threaded,
+                                             NULL);
+
+        torture_suite_add_simple_tcase_const(suite, "multi_tevent_threaded",
+                                             test_multi_tevent_threaded,
+                                             NULL);
+
+        torture_suite_add_simple_tcase_const(suite, "multi_tevent_threaded_1",
+                                             test_multi_tevent_threaded_1,
+                                             NULL);
+
+#endif
 
         return suite;