X-Git-Url: http://git.meshlink.io/?a=blobdiff_plain;f=src%2Fevent.c;h=a7c1f25a3f3b9b05edd94a6ee579b818dd5b3412;hb=HEAD;hp=40dbd18e5e5cce004c3776b7574a02233c0b0636;hpb=f83d168f21b87170b5b02357c378c4b819080369;p=meshlink

diff --git a/src/event.c b/src/event.c
index 40dbd18e..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"
@@ -199,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);
 	}
 }
@@ -209,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);
 	}
 }
@@ -224,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;
 }
@@ -240,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);
 	}
@@ -270,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);
@@ -311,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);
@@ -320,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);
 
@@ -328,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;
 		}
@@ -364,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;
 }
@@ -394,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);
 	}
 }