X-Git-Url: http://git.meshlink.io/?a=blobdiff_plain;f=src%2Fevent.c;h=a7c1f25a3f3b9b05edd94a6ee579b818dd5b3412;hb=HEAD;hp=739ff35de5a595b76fee4f56ed43ca42b31e9e4c;hpb=4c57e6902219ecca1872e18e34365d8e54a0f407;p=meshlink diff --git a/src/event.c b/src/event.c index 739ff35d..f8ebe8b8 100644 --- a/src/event.c +++ b/src/event.c @@ -21,6 +21,8 @@ #include "dropin.h" #include "event.h" +#include "logger.h" +#include "meshlink.h" #include "net.h" #include "splay_tree.h" #include "utils.h" @@ -64,10 +66,6 @@ static void timespec_clear(struct timespec *a) { a->tv_sec = 0; } -static bool timespec_isset(const struct timespec *a) { - return a->tv_sec; -} - static int io_compare(const io_t *a, const io_t *b) { return a->fd - b->fd; } @@ -137,7 +135,6 @@ void io_del(event_loop_t *loop, io_t *io) { void timeout_add(event_loop_t *loop, timeout_t *timeout, timeout_cb_t cb, void *data, struct timespec *tv) { timeout->cb = cb; timeout->data = data; - timeout->node.data = timeout; timeout_set(loop, timeout, tv); } @@ -145,8 +142,10 @@ void timeout_add(event_loop_t *loop, timeout_t *timeout, timeout_cb_t cb, void * void timeout_set(event_loop_t *loop, timeout_t *timeout, struct timespec *tv) { assert(timeout->cb); - if(timespec_isset(&timeout->tv)) { + if(timeout->node.data) { splay_unlink_node(&loop->timeouts, &timeout->node); + } else { + timeout->node.data = timeout; } if(!loop->now.tv_sec) { @@ -163,7 +162,11 @@ void timeout_set(event_loop_t *loop, timeout_t *timeout, struct timespec *tv) { } static void timeout_disable(event_loop_t *loop, timeout_t *timeout) { - splay_unlink_node(&loop->timeouts, &timeout->node); + if(timeout->node.data) { + splay_unlink_node(&loop->timeouts, &timeout->node); + timeout->node.data = NULL; + } + timespec_clear(&timeout->tv); } @@ -172,7 +175,7 @@ void timeout_del(event_loop_t *loop, timeout_t *timeout) { return; } - if(timespec_isset(&timeout->tv)) { + if(timeout->node.data) { timeout_disable(loop, timeout); } @@ -198,7 +201,9 @@ static void signalio_handler(event_loop_t *loop, void *data, int flags) { }); if(sig) { - sig->set = false; +#ifdef HAVE_STDATOMIC_H + atomic_flag_clear(&sig->set); +#endif sig->cb(loop, sig->data); } } @@ -208,6 +213,10 @@ static void pipe_init(event_loop_t *loop) { assert(result == 0); if(result == 0) { +#ifdef O_NONBLOCK + fcntl(loop->pipefd[0], F_SETFL, O_NONBLOCK); + fcntl(loop->pipefd[1], F_SETFL, O_NONBLOCK); +#endif io_add(loop, &loop->signalio, signalio_handler, NULL, loop->pipefd[0], IO_READ); } } @@ -223,12 +232,15 @@ static void pipe_exit(event_loop_t *loop) { } void signal_trigger(event_loop_t *loop, signal_t *sig) { - if(sig->set) { +#ifdef HAVE_STDATOMIC_H + + if(atomic_flag_test_and_set(&sig->set)) { return; } +#endif + uint8_t signum = sig->signum; - sig->set = true; write(loop->pipefd[1], &signum, 1); return; } @@ -239,9 +251,12 @@ void signal_add(event_loop_t *loop, signal_t *sig, signal_cb_t cb, void *data, u sig->cb = cb; sig->data = data; sig->signum = signum; - sig->set = false; sig->node.data = sig; +#ifdef HAVE_STDATOMIC_H + atomic_flag_clear(&sig->set); +#endif + if(loop->pipefd[0] == -1) { pipe_init(loop); } @@ -269,11 +284,68 @@ void idle_set(event_loop_t *loop, idle_cb_t cb, void *data) { loop->idle_data = data; } -bool event_loop_run(event_loop_t *loop, pthread_mutex_t *mutex) { - assert(mutex); +static void check_bad_fds(event_loop_t *loop, meshlink_handle_t *mesh) { + // Just call all registered callbacks and have them check their fds + + do { + loop->deletion = false; + + for splay_each(io_t, io, &loop->ios) { + if(io->flags & IO_WRITE) { + io->cb(loop, io->data, IO_WRITE); + } + + if(loop->deletion) { + break; + } + + if(io->flags & IO_READ) { + io->cb(loop, io->data, IO_READ); + } + + if(loop->deletion) { + break; + } + } + } while(loop->deletion); + + // Rebuild the fdsets + + fd_set old_readfds; + fd_set old_writefds; + memcpy(&old_readfds, &loop->readfds, sizeof(old_readfds)); + memcpy(&old_writefds, &loop->writefds, sizeof(old_writefds)); + + memset(&loop->readfds, 0, sizeof(loop->readfds)); + memset(&loop->writefds, 0, sizeof(loop->writefds)); + + for splay_each(io_t, io, &loop->ios) { + if(io->flags & IO_READ) { + FD_SET(io->fd, &loop->readfds); + io->cb(loop, io->data, IO_READ); + } + + if(io->flags & IO_WRITE) { + FD_SET(io->fd, &loop->writefds); + io->cb(loop, io->data, IO_WRITE); + } + } + + if(memcmp(&old_readfds, &loop->readfds, sizeof(old_readfds))) { + logger(mesh, MESHLINK_WARNING, "Incorrect readfds fixed"); + } + + if(memcmp(&old_writefds, &loop->writefds, sizeof(old_writefds))) { + logger(mesh, MESHLINK_WARNING, "Incorrect writefds fixed"); + } +} + +bool event_loop_run(event_loop_t *loop, meshlink_handle_t *mesh) { + assert(mesh); fd_set readable; fd_set writable; + int errors = 0; while(loop->running) { clock_gettime(EVENT_CLOCK, &loop->now); @@ -310,7 +382,7 @@ bool event_loop_run(event_loop_t *loop, pthread_mutex_t *mutex) { } // release mesh mutex during select - pthread_mutex_unlock(mutex); + pthread_mutex_unlock(&mesh->mutex); #ifdef HAVE_PSELECT int n = pselect(fds, &readable, &writable, NULL, &ts, NULL); @@ -319,7 +391,9 @@ bool event_loop_run(event_loop_t *loop, pthread_mutex_t *mutex) { int n = select(fds, &readable, &writable, NULL, (struct timeval *)&tv); #endif - pthread_mutex_lock(mutex); + if(pthread_mutex_lock(&mesh->mutex) != 0) { + abort(); + } clock_gettime(EVENT_CLOCK, &loop->now); @@ -327,10 +401,21 @@ bool event_loop_run(event_loop_t *loop, pthread_mutex_t *mutex) { if(sockwouldblock(errno)) { continue; } else { - return false; + errors++; + + if(errors > 10) { + logger(mesh, MESHLINK_ERROR, "Unrecoverable error from select(): %s", strerror(errno)); + return false; + } + + logger(mesh, MESHLINK_WARNING, "Error from select(), checking for bad fds: %s", strerror(errno)); + check_bad_fds(loop, mesh); + continue; } } + errors = 0; + if(!n) { continue; } @@ -363,13 +448,6 @@ bool event_loop_run(event_loop_t *loop, pthread_mutex_t *mutex) { return true; } -void event_flush_output(event_loop_t *loop) { - for splay_each(io_t, io, &loop->ios) - if(FD_ISSET(io->fd, &loop->writefds)) { - io->cb(loop, io->data, IO_WRITE); - } -} - void event_loop_start(event_loop_t *loop) { loop->running = true; } @@ -393,14 +471,14 @@ void event_loop_exit(event_loop_t *loop) { assert(!loop->signals.count); for splay_each(io_t, io, &loop->ios) { - splay_unlink_node(&loop->ios, node); + splay_unlink_node(&loop->ios, splay_node); } for splay_each(timeout_t, timeout, &loop->timeouts) { - splay_unlink_node(&loop->timeouts, node); + splay_unlink_node(&loop->timeouts, splay_node); } for splay_each(signal_t, signal, &loop->signals) { - splay_unlink_node(&loop->signals, node); + splay_unlink_node(&loop->signals, splay_node); } }