Fix PMTU test case.

[meshlink] / src / event.c

diff --git a/src/event.c b/src/event.c
index 8ceab36240224ab8192a80071469ff8160f0cba0..857aaec240c5d5eb216159649b8794c868b95e52 100644 (file)
--- 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"
@@ -34,36 +36,6 @@
 #endif
 #endif
 
-static void timespec_add(const struct timespec *a, const struct timespec *b, struct timespec *r) {
-       r->tv_sec = a->tv_sec + b->tv_sec;
-       r->tv_nsec = a->tv_nsec + b->tv_nsec;
-
-       if(r->tv_nsec > 1000000000) {
-               r->tv_sec++, r->tv_nsec -= 1000000000;
-       }
-}
-
-static void timespec_sub(const struct timespec *a, const struct timespec *b, struct timespec *r) {
-       r->tv_sec = a->tv_sec - b->tv_sec;
-       r->tv_nsec = a->tv_nsec - b->tv_nsec;
-
-       if(r->tv_nsec < 0) {
-               r->tv_sec--, r->tv_nsec += 1000000000;
-       }
-}
-
-static bool timespec_lt(const struct timespec *a, const struct timespec *b) {
-       if(a->tv_sec == b->tv_sec) {
-               return a->tv_nsec < b->tv_nsec;
-       } else {
-               return a->tv_sec < b->tv_sec;
-       }
-}
-
-static void timespec_clear(struct timespec *a) {
-       a->tv_sec = 0;
-}
-
 static int io_compare(const io_t *a, const io_t *b) {
        return a->fd - b->fd;
 }
@@ -199,7 +171,9 @@ static void signalio_handler(event_loop_t *loop, void *data, int flags) {
        });
 
        if(sig) {
+#ifdef HAVE_STDATOMIC_H
                atomic_flag_clear(&sig->set);
+#endif
                sig->cb(loop, sig->data);
        }
 }
@@ -209,6 +183,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);
        }
 }
@@ -224,10 +202,14 @@ static void pipe_exit(event_loop_t *loop) {
 }
 
 void signal_trigger(event_loop_t *loop, signal_t *sig) {
+#ifdef HAVE_STDATOMIC_H
+
        if(atomic_flag_test_and_set(&sig->set)) {
                return;
        }
 
+#endif
+
        uint8_t signum = sig->signum;
        write(loop->pipefd[1], &signum, 1);
        return;
@@ -241,7 +223,9 @@ void signal_add(event_loop_t *loop, signal_t *sig, signal_cb_t cb, void *data, u
        sig->signum = signum;
        sig->node.data = sig;
 
+#ifdef HAVE_STDATOMIC_H
        atomic_flag_clear(&sig->set);
+#endif
 
        if(loop->pipefd[0] == -1) {
                pipe_init(loop);
@@ -270,11 +254,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);
@@ -311,7 +352,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);
@@ -320,7 +361,7 @@ bool event_loop_run(event_loop_t *loop, pthread_mutex_t *mutex) {
                int n = select(fds, &readable, &writable, NULL, (struct timeval *)&tv);
 #endif
 
-               if(pthread_mutex_lock(mutex) != 0) {
+               if(pthread_mutex_lock(&mesh->mutex) != 0) {
                        abort();
                }
 
@@ -330,10 +371,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;
                }