From 80eead4d4ab6983362bf9bd59883e223ac2a7630 Mon Sep 17 00:00:00 2001 From: niftynei Date: Mon, 24 Aug 2026 19:05:58 -0500 Subject: [PATCH 1/2] ccan/io: retain registrations during poll dispatch A callback can close a connection while readiness from the current poll call is still being dispatched. Deleting that connection compacts the fd table, and registering a replacement can restore its previous length with different occupants. A snapshot tied to mutable table positions can consequently skip an original event or deliver stale readiness to the replacement. Give each fd a stable registration object and retain that registration while a poll result refers to it. Deletion clears the registration's fd pointer, so pending readiness for a retired connection is ignored, while a replacement receives a distinct registration and cannot inherit the old event. Snapshot only entries with nonzero revents and preserve the IO_ALWAYS marker in the existing fairness order. Reuse a growable snapshot buffer between loop iterations and dispatch directly through the retained registration, avoiding both a fresh allocation on every poll and the repeated linear lookup required by the earlier generation-based approach. Register one process-exit cleanup for that reusable buffer. Retaining it for the daemon lifetime preserves allocation reuse, while releasing it after event dispatch can no longer be active avoids a reachable allocation in Valgrind and LeakSanitizer runs. Changelog-Fixed: io: replacing a file descriptor during poll dispatch no longer delivers stale readiness events to the replacement connection. --- ccan/ccan/io/backend.h | 2 + ccan/ccan/io/io.h | 5 ++ ccan/ccan/io/poll.c | 168 +++++++++++++++++++++++++++++++++++++++- connectd/connectd.c | 1 + lightningd/lightningd.c | 1 + 5 files changed, 176 insertions(+), 1 deletion(-) 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/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); From e80ded25b348d12af1714416fdc74bc9e74a916b Mon Sep 17 00:00:00 2001 From: niftynei Date: Mon, 24 Aug 2026 17:12:02 -0500 Subject: [PATCH 2/2] ccan/io: test fd replacement during poll dispatch Closing a connection while dispatching poll results compacts the fd table. If the callback immediately registers a replacement, the table can return to its original length with a different connection occupying a slot represented in the current poll result. Dispatching readiness by mutable table position can then deliver an old event to the replacement or skip another ready connection. Use a fake poll implementation to make three original connections ready. The connection selected first closes itself and installs a replacement while the other readiness events remain pending. Verify that every original event is delivered exactly once, that the replacement does not inherit stale readiness, and that it is handled normally by the following poll call. Register an exit assertion before enabling protection so atexit LIFO ordering checks that the backend releases its reusable snapshot buffer at process exit. This keeps leak-sensitive unit runs from accepting a permanently reachable allocation. Add the regression to check-units so this CCAN behavior is covered by the project's normal unit-test suite. Changelog-None --- ccan/ccan/io/test/run-49-stale-fd-readiness.c | 143 ++++++++++++++++++ common/test/Makefile | 9 +- 2 files changed, 149 insertions(+), 3 deletions(-) create mode 100644 ccan/ccan/io/test/run-49-stale-fd-readiness.c 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/%)