diff --git a/ccan/ccan/io/backend.h b/ccan/ccan/io/backend.h index 714972d15ea4..5934e154cd2c 100644 --- a/ccan/ccan/io/backend.h +++ b/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. */ diff --git a/ccan/ccan/io/io.h b/ccan/ccan/io/io.h index 5d084828b96a..b4eecb3cfd77 100644 --- a/ccan/ccan/io/io.h +++ b/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. diff --git a/ccan/ccan/io/poll.c b/ccan/ccan/io/poll.c index c4cbaee85678..f49b93b28423 100644 --- a/ccan/ccan/io/poll.c +++ b/ccan/ccan/io/poll.c @@ -11,13 +11,58 @@ #include #include +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 io_plan **always = NULL; static struct timemono (*nowfn)(void) = time_mono; static int (*pollfn)(struct pollfd *fds, nfds_t nfds, int timeout) = poll; +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; + + /* 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 +99,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 +126,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 +150,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,10 +427,35 @@ static void restore_pollfds(void) } } +static void append_ready_fd(struct ready_fd **ready_fds, + size_t *ready_fds_capacity, + 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) { void *ret; + /* Callbacks can enter a nested io_loop, so snapshots cannot be global. */ + struct ready_fd *ready_fds = NULL; + size_t ready_fds_capacity = 0; /* This ensures we don't always service lower fds first */ static int fairness_counter; @@ -431,6 +514,89 @@ 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(&ready_fds, + &ready_fds_capacity, + &num_ready, NULL, 0); + + if (!pollfds[i].revents) + continue; + registration = fds[i]->registration; + assert(registration); + registration->refs++; + append_ready_fd(&ready_fds, + &ready_fds_capacity, + &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,9 +652,9 @@ void *io_loop(struct timers *timers, struct timer **expired) } } } - ret = io_loop_return; io_loop_return = NULL; + free(ready_fds); return ret; } diff --git a/ccan/ccan/io/test/run-49-stale-fd-readiness.c b/ccan/ccan/io/test/run-49-stale-fd-readiness.c new file mode 100644 index 000000000000..5cdfb3f41b82 --- /dev/null +++ b/ccan/ccan/io/test/run-49-stale-fd-readiness.c @@ -0,0 +1,143 @@ +#include "config.h" +#include +#include +/* Include the C files directly to make each readiness delivery observable. */ +#include +#include +#include + +static int replacement_fd; +static unsigned int poll_calls, actor_events, victim_events, + survivor_events, replacement_events; +static char completed, nested_completed; + +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(&nested_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(); + /* A nested loop must not overwrite or release the outer loop's readiness + * snapshot. */ + assert(io_loop(NULL, NULL) == &nested_completed); + 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_ready(struct io_conn *conn, void *unused) +{ + io_break(&completed); + return io_close(conn); +} + +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, victim_ready, 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)); + + 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; +} diff --git a/common/test/Makefile b/common/test/Makefile index 71f9dc17d981..dd165bef0f45 100644 --- a/common/test/Makefile +++ b/common/test/Makefile @@ -1,7 +1,10 @@ +CCAN_IO_TEST_SRC := ccan/ccan/io/test/run-49-stale-fd-readiness.c COMMON_TEST_SRC := $(wildcard common/test/run-*.c) COMMON_TEST_OBJS := $(COMMON_TEST_SRC:.c=.o) COMMON_TEST_PROGRAMS := $(COMMON_TEST_OBJS:.o=) +CCAN_IO_TEST_OBJS := $(CCAN_IO_TEST_SRC:.c=.o) +CCAN_IO_TEST_PROGRAMS := $(CCAN_IO_TEST_OBJS:.o=) COMMON_TEST_COMMON_OBJS := \ common/autodata.o \ @@ -13,8 +16,8 @@ COMMON_TEST_COMMON_OBJS := \ $(COMMON_TEST_PROGRAMS): $(COMMON_TEST_COMMON_OBJS) $(BITCOIN_OBJS) $(COMMON_TEST_OBJS): $(COMMON_HEADERS) $(WIRE_HEADERS) $(COMMON_SRC) common/test/Makefile -ALL_C_SOURCES += $(COMMON_TEST_SRC) -ALL_TEST_PROGRAMS += $(COMMON_TEST_PROGRAMS) +ALL_C_SOURCES += $(COMMON_TEST_SRC) $(CCAN_IO_TEST_SRC) +ALL_TEST_PROGRAMS += $(COMMON_TEST_PROGRAMS) $(CCAN_IO_TEST_PROGRAMS) # Make them all depend on common/ files, for simplicity (they directly #include some) $(COMMON_TEST_OBJS): $(COMMON_SRC) @@ -178,4 +181,4 @@ common/test/run-close_tx: \ wire/fromwire.o \ wire/towire.o -check-units: $(COMMON_TEST_PROGRAMS:%=unittest/%) +check-units: $(COMMON_TEST_PROGRAMS:%=unittest/%) $(CCAN_IO_TEST_PROGRAMS:%=unittest/%) diff --git a/connectd/connectd.c b/connectd/connectd.c index cb95ef9b0cac..9e96a31ec8f9 100644 --- a/connectd/connectd.c +++ b/connectd/connectd.c @@ -2532,6 +2532,7 @@ int main(int argc, char *argv[]) /* Common subdaemon setup code. */ developer = subdaemon_setup(argc, argv); + io_poll_protect_stale_fds(); /* Allocate and set up our simple top-level structure. */ daemon = tal(NULL, struct daemon); diff --git a/lightningd/lightningd.c b/lightningd/lightningd.c index 44ed3fa66425..dfd159efa51d 100644 --- a/lightningd/lightningd.c +++ b/lightningd/lightningd.c @@ -1199,6 +1199,7 @@ int main(int argc, char *argv[]) /*~ What happens in strange locales should stay there. */ setup_locale(); + io_poll_protect_stale_fds(); /*~ This handles --dev-debug-self really early, which we otherwise ignore */ daemon_developer_mode(argv);