Try to recover from select() returning an error.

[meshlink] / src / event.c

diff --git a/src/event.c b/src/event.c
index 739ff35de5a595b76fee4f56ed43ca42b31e9e4c..82133eb471826ce1d9921d0f4a3fbac83df9387a 100644 (file)
--- a/src/event.c
+++ b/src/event.c
@@ -64,10 +64,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 +133,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 +140,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 +160,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 +173,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 +199,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 +211,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 +230,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 +249,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 +282,56 @@ void idle_set(event_loop_t *loop, idle_cb_t cb, void *data) {
        loop->idle_data = data;
 }
 
+static void check_bad_fds(event_loop_t *loop) {
+       // 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
+
+       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);
+               }
+
+       }
+}
+
 bool event_loop_run(event_loop_t *loop, pthread_mutex_t *mutex) {
        assert(mutex);
 
        fd_set readable;
        fd_set writable;
+       int errors = 0;
 
        while(loop->running) {
                clock_gettime(EVENT_CLOCK, &loop->now);
@@ -319,7 +377,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(mutex) != 0) {
+                       abort();
+               }
 
                clock_gettime(EVENT_CLOCK, &loop->now);
 
@@ -327,10 +387,19 @@ bool event_loop_run(event_loop_t *loop, pthread_mutex_t *mutex) {
                        if(sockwouldblock(errno)) {
                                continue;
                        } else {
-                               return false;
+                               errors++;
+
+                               if(errors > 10) {
+                                       return false;
+                               }
+
+                               check_bad_fds(loop);
+                               continue;
                        }
                }
 
+               errors = 0;
+
                if(!n) {
                        continue;
                }
@@ -363,13 +432,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 +455,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);
        }
 }