dualopend: make opening teardown failures explicit

ccan/ccan/io/backend.h

@@ -11,6 +11,8 @@ struct fd {
 	/* We could put these in io_plan, but they pack nicely here */
 	bool exclusive[2];
 	size_t backend_info;
+	/* Stable while a poll result still refers to this registration. */
+	struct io_fd_registration *registration;
 };
 
 /* Listeners create connections. */

ccan/ccan/io/io.h

@@ -816,6 +816,11 @@ struct timemono (*io_time_override(struct timemono (*now)(void)))(void);
  */
 int (*io_poll_override(int (*poll)(struct pollfd *fds, nfds_t nfds, int timeout)))(struct pollfd *, nfds_t, int);
 
+/* Protect daemons which replace connections inside readiness callbacks from
+ * delivering an old poll result to a newly registered fd object. */
+void io_poll_protect_stale_fds(void);
+
+
 /**
  * io_have_fd - do we own this file descriptor?
  * @fd: the file descriptor.

ccan/ccan/io/poll.c

@@ -11,13 +11,69 @@
 #include <ccan/time/time.h>
 #include <ccan/timer/timer.h>
 
+struct io_fd_registration {
+	struct fd *fd;
+	size_t refs;
+};
+
+struct ready_fd {
+	/* NULL is the position at which to service an IO_ALWAYS plan. */
+	struct io_fd_registration *registration;
+	short revents;
+};
+
 static size_t num_fds = 0, max_fds = 0, num_waiting = 0, num_always = 0, max_always = 0, num_exclusive = 0;
 static struct pollfd *pollfds = NULL;
 static struct fd **fds = NULL;
+static bool protect_stale_fds = false;
+static struct ready_fd *ready_fds = NULL;
+static size_t ready_fds_capacity = 0;
 static struct io_plan **always = NULL;
 static struct timemono (*nowfn)(void) = time_mono;
 static int (*pollfn)(struct pollfd *fds, nfds_t nfds, int timeout) = poll;
 
+static void cleanup_ready_fds(void)
+{
+	free(ready_fds);
+	ready_fds = NULL;
+	ready_fds_capacity = 0;
+}
+
+static struct io_fd_registration *new_registration(struct fd *fd)
+{
+	struct io_fd_registration *registration = malloc(sizeof(*registration));
+
+	if (!registration)
+		return NULL;
+	registration->fd = fd;
+	registration->refs = 1;
+	return registration;
+}
+
+static void registration_put(struct io_fd_registration *registration)
+{
+	assert(registration->refs != 0);
+	registration->refs--;
+	if (registration->refs == 0)
+		free(registration);
+}
+
+void io_poll_protect_stale_fds(void)
+{
+	if (protect_stale_fds)
+		return;
+	if (atexit(cleanup_ready_fds) != 0)
+		abort();
+
+	/* This can be enabled after callers have already registered fds. */
+	for (size_t i = 0; i < num_fds; i++) {
+		fds[i]->registration = new_registration(fds[i]);
+		if (!fds[i]->registration)
+			abort();
+	}
+	protect_stale_fds = true;
+}
+
 struct timemono (*io_time_override(struct timemono (*now)(void)))(void)
 {
 	struct timemono (*old)(void) = nowfn;
@@ -54,6 +110,12 @@ static bool add_fd(struct fd *fd, short events)
 			return false;
 		max_fds = num;
 	}
+	if (protect_stale_fds) {
+		fd->registration = new_registration(fd);
+		if (!fd->registration)
+			return false;
+	} else
+		fd->registration = NULL;
 
 	pollfds[num_fds].events = events;
 	/* In case it's idle. */
@@ -75,6 +137,7 @@ static bool add_fd(struct fd *fd, short events)
 static void del_fd(struct fd *fd)
 {
 	size_t n = fd->backend_info;
+	struct io_fd_registration *registration = fd->registration;
 
 	assert(n != -1);
 	assert(n < num_fds);
@@ -98,6 +161,12 @@ static void del_fd(struct fd *fd)
 	}
 	num_fds--;
 	fd->backend_info = -1;
+	if (registration) {
+		assert(registration->fd == fd);
+		registration->fd = NULL;
+		fd->registration = NULL;
+		registration_put(registration);
+	}
 
 	if (fd->exclusive[IO_IN])
 		num_exclusive--;
@@ -369,6 +438,26 @@ static void restore_pollfds(void)
 	}
 }
 
+static void append_ready_fd(size_t *num_ready,
+			    struct io_fd_registration *registration,
+			    short revents)
+{
+	if (*num_ready == ready_fds_capacity) {
+		size_t capacity = ready_fds_capacity ? ready_fds_capacity * 2 : 8;
+		struct ready_fd *new_ready_fds;
+
+		new_ready_fds = realloc(ready_fds,
+					  sizeof(*ready_fds) * capacity);
+		if (!new_ready_fds)
+			abort();
+		ready_fds = new_ready_fds;
+		ready_fds_capacity = capacity;
+	}
+	ready_fds[*num_ready].registration = registration;
+	ready_fds[*num_ready].revents = revents;
+	(*num_ready)++;
+}
+
 /* This is the main loop. */
 void *io_loop(struct timers *timers, struct timer **expired)
 {
@@ -431,6 +520,85 @@ void *io_loop(struct timers *timers, struct timer **expired)
 			break;
 		}
 
+		if (protect_stale_fds) {
+			size_t num_polled = num_fds;
+			size_t num_ready = 0;
+
+			if (r == 0) {
+				handle_always();
+				continue;
+			}
+
+			/* Preserve the old fairness order, but retain only ready fds and
+			 * the point where IO_ALWAYS work was interleaved. */
+			fairness_counter++;
+			for (size_t rotation = 0; rotation < num_polled; rotation++) {
+				struct io_fd_registration *registration;
+
+				i = (rotation + fairness_counter) % num_polled;
+				if (i == 0)
+					append_ready_fd(&num_ready, NULL, 0);
+
+				if (!pollfds[i].revents)
+					continue;
+				registration = fds[i]->registration;
+				assert(registration);
+				registration->refs++;
+				append_ready_fd(&num_ready, registration,
+						pollfds[i].revents);
+			}
+			for (size_t n = 0; n < num_polled; n++)
+				pollfds[n].revents = 0;
+
+			for (size_t n = 0; n < num_ready && !io_loop_return; n++) {
+				socklen_t errno_len = sizeof(errno);
+				struct io_fd_registration *registration;
+				struct fd *fd;
+				struct io_conn *c;
+				int events;
+
+				registration = ready_fds[n].registration;
+				if (!registration) {
+					if (handle_always())
+						break;
+					continue;
+				}
+				fd = registration->fd;
+				if (!fd)
+					continue;
+				events = ready_fds[n].revents;
+				c = (void *)fd;
+				if (fd->listener) {
+					struct io_listener *l = (void *)fd;
+					if (events & POLLIN) {
+						accept_conn(l);
+						r--;
+					} else if (events & (POLLHUP|POLLNVAL|POLLERR)) {
+						r--;
+						errno = EBADF;
+						io_close_listener(l);
+					}
+				} else if (events & (POLLIN|POLLOUT)) {
+					r--;
+					io_ready(c, events);
+				} else if (events & (POLLHUP|POLLNVAL|POLLERR)) {
+					r--;
+					if (getsockopt(fd->fd, SOL_SOCKET,
+						       SO_ERROR, &errno,
+						       &errno_len) == -1)
+						errno = EBADF;
+					io_close(c);
+				}
+			}
+			/* Callbacks can retire registrations, so release snapshot
+			 * references only after dispatch has stopped using the buffer. */
+			for (size_t n = 0; n < num_ready; n++) {
+				if (ready_fds[n].registration)
+					registration_put(ready_fds[n].registration);
+			}
+			continue;
+		}
+
 		fairness_counter++;
 		for (size_t rotation = 0; rotation < num_fds && !io_loop_return; rotation++) {
 			socklen_t errno_len = sizeof(errno);
@@ -486,7 +654,6 @@ void *io_loop(struct timers *timers, struct timer **expired)
 			}
 		}
 	}
-
 	ret = io_loop_return;
 	io_loop_return = NULL;
 

ccan/ccan/io/test/run-49-stale-fd-readiness.c

@@ -0,0 +1,142 @@
+#include "config.h"
+#include <assert.h>
+#include <ccan/io/io.h>
+/* Include the C files directly to make each readiness delivery observable. */
+#include <ccan/io/poll.c>
+#include <ccan/io/io.c>
+#include <unistd.h>
+
+static int replacement_fd;
+static unsigned int poll_calls, actor_events, victim_events,
+		    survivor_events, replacement_events;
+static char completed;
+
+static void check_ready_fds_cleaned(void)
+{
+	assert(ready_fds == NULL);
+	assert(ready_fds_capacity == 0);
+}
+
+static int replacement_event(int fd, struct io_plan_arg *arg)
+{
+	replacement_events++;
+	return 1;
+}
+
+static struct io_plan *replacement_ready(struct io_conn *conn, void *unused)
+{
+	io_break(&completed);
+	return io_close(conn);
+}
+
+static struct io_plan *replacement_init(struct io_conn *conn, void *unused)
+{
+	io_plan_arg(conn, IO_IN);
+	return io_set_plan(conn, IO_IN, replacement_event,
+			   replacement_ready, NULL);
+}
+
+static int actor_event(int fd, struct io_plan_arg *arg)
+{
+	actor_events++;
+	return 1;
+}
+
+static struct io_plan *actor_ready(struct io_conn *conn, void *unused)
+{
+	struct io_plan *closed = io_close(conn);
+
+	/* Closing the current fd compacts the table.  Adding its replacement
+	 * restores the old table length, but must not make it part of the poll
+	 * result currently being dispatched. */
+	if (!io_new_conn(NULL, replacement_fd, replacement_init, NULL))
+		abort();
+	return closed;
+}
+
+static struct io_plan *actor_init(struct io_conn *conn, void *unused)
+{
+	io_plan_arg(conn, IO_IN);
+	return io_set_plan(conn, IO_IN, actor_event, actor_ready, NULL);
+}
+
+static int victim_event(int fd, struct io_plan_arg *arg)
+{
+	victim_events++;
+	return 1;
+}
+
+static struct io_plan *victim_init(struct io_conn *conn, void *unused)
+{
+	io_plan_arg(conn, IO_IN);
+	return io_set_plan(conn, IO_IN, victim_event, io_close_cb, NULL);
+}
+
+static int survivor_event(int fd, struct io_plan_arg *arg)
+{
+	survivor_events++;
+	return 1;
+}
+
+static struct io_plan *survivor_init(struct io_conn *conn, void *unused)
+{
+	io_plan_arg(conn, IO_IN);
+	return io_set_plan(conn, IO_IN, survivor_event, io_close_cb, NULL);
+}
+
+static int fake_poll(struct pollfd *fds, nfds_t nfds, int timeout)
+{
+	poll_calls++;
+	for (size_t i = 0; i < nfds; i++)
+		fds[i].revents = 0;
+
+	if (poll_calls == 1) {
+		/* Fairness rotation handles slot 1 first.  That actor replaces
+		 * itself while slots 0 and 2 still have readiness pending. */
+		assert(nfds == 3);
+		for (size_t i = 0; i < nfds; i++)
+			fds[i].revents = POLLIN;
+		return 3;
+	}
+
+	assert(poll_calls == 2);
+	for (size_t i = 0; i < nfds; i++) {
+		if (fds[i].fd != replacement_fd)
+			continue;
+		fds[i].revents = POLLIN;
+		return 1;
+	}
+	abort();
+}
+
+int main(void)
+{
+	int actor_pipe[2], victim_pipe[2], survivor_pipe[2], replacement_pipe[2];
+
+	assert(pipe(actor_pipe) == 0);
+	assert(pipe(victim_pipe) == 0);
+	assert(pipe(survivor_pipe) == 0);
+	assert(pipe(replacement_pipe) == 0);
+	replacement_fd = replacement_pipe[0];
+
+	assert(io_poll_override(fake_poll) == poll);
+	assert(io_new_conn(NULL, victim_pipe[0], victim_init, NULL));
+	assert(io_new_conn(NULL, actor_pipe[0], actor_init, NULL));
+	assert(io_new_conn(NULL, survivor_pipe[0], survivor_init, NULL));
+
+	/* Registered first so this runs after the backend's LIFO cleanup. */
+	assert(atexit(check_ready_fds_cleaned) == 0);
+	io_poll_protect_stale_fds();
+	assert(io_loop(NULL, NULL) == &completed);
+	assert(poll_calls == 2);
+	assert(actor_events == 1);
+	assert(victim_events == 1);
+	assert(survivor_events == 1);
+	assert(replacement_events == 1);
+
+	close(actor_pipe[1]);
+	close(victim_pipe[1]);
+	close(survivor_pipe[1]);
+	close(replacement_pipe[1]);
+	return 0;
+}