From aafca410e0f277640951476b4f7bda11d96d6b67 Mon Sep 17 00:00:00 2001 From: Andy Postnikov Date: Wed, 16 Sep 2026 01:23:27 +0200 Subject: [PATCH] port: bound the queued messages that hold a descriptor port->messages is unbounded. Before a queued message owned the descriptors it names, a peer that stopped reading cost the sender memory. Now each queued message holds up to two descriptors of this process open for as long as it waits, so the same peer applies RLIMIT_NOFILE pressure as well. Bound that, and only that. A message with no descriptor still costs only memory, which is not what changed, and bounding the queue as a whole would change behaviour on every ordinary reply and every body fragment. So NXT_PORT_MAX_FD_MSGS caps the descriptor-carrying entries of one port, and the queue itself stays unbounded. A send that would go past the bound is refused with NXT_ERROR rather than dropped. That answer already means "nothing was consumed, the message is still yours" -- it is what nxt_port_msg_chk_insert() answers when the copy cannot be allocated -- so every caller that sends a descriptor already closes it on a failed send. Co-Authored-By: Claude Opus 5 (1M context) --- src/nxt_port.h | 34 +++ src/nxt_port_socket.c | 125 +++++++++- src/test/nxt_port_queued_fd_test.c | 356 +++++++++++++++++++++++++++++ 3 files changed, 510 insertions(+), 5 deletions(-) diff --git a/src/nxt_port.h b/src/nxt_port.h index b2db9bb39..e7f8d1359 100644 --- a/src/nxt_port.h +++ b/src/nxt_port.h @@ -282,6 +282,30 @@ nxt_port_recv_msg_close_fds(nxt_port_recv_msg_t *msg) } +/* + * How many queued messages on one port may carry a file descriptor. + * + * port->messages itself stays unbounded. A message with no descriptor costs + * only memory, which was true before a queued message took ownership of what + * it names, and bounding it would change behaviour on paths that never hold a + * descriptor at all -- every ordinary reply and every request body fragment. + * + * A message that does carry one is different: it holds up to two descriptors + * of this process open for as long as it waits, so a peer that stops reading + * turns into RLIMIT_NOFILE pressure on the sender rather than memory pressure + * alone. That is the cost this bounds. + * + * The traffic being bounded is control-plane and one message per event: a new + * port, a process start, a listening socket, a certificate, a script, a shared + * memory segment. A port with 128 of them outstanding is not a busy port, it + * is a peer that has stopped reading, so the bound is far above any legitimate + * burst. At two descriptors an entry it caps one stalled port at 256 open + * descriptors, which leaves room for several of them under the 1024 soft + * RLIMIT_NOFILE that is still the common default. + */ +#define NXT_PORT_MAX_FD_MSGS 128 + + typedef struct nxt_app_s nxt_app_t; struct nxt_port_s { @@ -300,6 +324,16 @@ struct nxt_port_s { nxt_queue_t messages; /* of nxt_port_send_msg_t */ nxt_thread_mutex_t write_mutex; + /* + * How many entries of ->messages still carry a file descriptor. + * + * A queued message owns the descriptors it names, so each such entry + * holds up to two of this process's descriptors open for as long as it + * waits. The count exists to bound that; it is maintained under + * ->write_mutex and is described with the bound in src/nxt_port_socket.c. + */ + uint32_t fd_messages; + /* Maximum size of message part. */ uint32_t max_size; /* Maximum interleave of message parts. */ diff --git a/src/nxt_port_socket.c b/src/nxt_port_socket.c index 0c99a9453..c7439977a 100644 --- a/src/nxt_port_socket.c +++ b/src/nxt_port_socket.c @@ -13,7 +13,6 @@ #define NXT_PORT_MAX_ENQUEUE_BUF_SIZE \ (int) (NXT_PORT_QUEUE_MSG_SIZE - sizeof(nxt_port_msg_t)) - static nxt_bool_t nxt_port_can_enqueue_buf(nxt_buf_t *b); static uint8_t nxt_port_enqueue_buf(nxt_task_t *task, nxt_port_msg_t *pm, void *qbuf, nxt_buf_t *b); @@ -25,6 +24,11 @@ static nxt_int_t nxt_port_write_msgs(nxt_task_t *task, void *obj, void *data, nxt_bool_t *send_failed); static void nxt_port_write_handler(nxt_task_t *task, void *obj, void *data); static nxt_port_send_msg_t *nxt_port_msg_first(nxt_port_t *port); +nxt_inline nxt_bool_t nxt_port_msg_has_fd(const nxt_port_send_msg_t *msg); +nxt_inline void nxt_port_msg_fd_uncount_locked(nxt_port_t *port, + nxt_port_send_msg_t *msg); +static void nxt_port_msg_fd_uncount(nxt_port_t *port, + nxt_port_send_msg_t *msg); nxt_inline void nxt_port_msg_close_fd(nxt_port_send_msg_t *msg); nxt_inline void nxt_port_close_fds(nxt_fd_t *fd); static nxt_buf_t *nxt_port_buf_completion(nxt_task_t *task, @@ -471,7 +475,11 @@ static nxt_int_t nxt_port_msg_chk_insert(nxt_task_t *task, nxt_port_t *port, nxt_port_send_msg_t *msg) { - nxt_int_t res; + nxt_int_t res; + nxt_bool_t has_fd, over_bound; + + has_fd = nxt_port_msg_has_fd(msg); + over_bound = 0; nxt_thread_mutex_lock(&port->write_mutex); @@ -480,11 +488,30 @@ nxt_port_msg_chk_insert(nxt_task_t *task, nxt_port_t *port, { res = NXT_DECLINED; + } else if (nxt_slow_path(has_fd + && port->fd_messages >= NXT_PORT_MAX_FD_MSGS)) + { + /* + * Refuse rather than drop. NXT_ERROR from here says nothing of the + * message was consumed and it is still the caller's -- the same + * answer this function already gives when nxt_port_msg_alloc() + * fails, and the contract in src/nxt_port.h -- so every caller that + * sends a descriptor already has cleanup for it. + */ + + over_bound = 1; + res = NXT_ERROR; + } else { msg = nxt_port_msg_alloc(msg); if (nxt_fast_path(msg != NULL)) { nxt_queue_insert_tail(&port->messages, &msg->link); + + if (has_fd) { + port->fd_messages++; + } + nxt_port_use(task, port, 1); res = NXT_OK; @@ -495,6 +522,13 @@ nxt_port_msg_chk_insert(nxt_task_t *task, nxt_port_t *port, nxt_thread_mutex_unlock(&port->write_mutex); + if (nxt_slow_path(over_bound)) { + nxt_alert(task, "port{%d,%d} %d: %d queued messages already hold a " + "descriptor, refusing to queue another", + (int) port->pid, (int) port->id, port->socket.fd, + NXT_PORT_MAX_FD_MSGS); + } + return res; } @@ -540,7 +574,8 @@ nxt_port_msg_alloc(const nxt_port_send_msg_t *m) * * The dup holds the descriptor until the message goes out, so a peer * that stops reading holds this side's descriptor table open in - * proportion to what it was sent; port->messages has no bound. + * proportion to what it was sent. That is what NXT_PORT_MAX_FD_MSGS + * bounds; the callers of this function check it before they get here. */ if (!msg->close_fd && nxt_slow_path(nxt_port_msg_dup_fds(msg) != NXT_OK)) { @@ -879,6 +914,8 @@ nxt_port_write_msgs(nxt_task_t *task, void *obj, void *data, goto fail; } + nxt_port_msg_fd_uncount(port, msg); + nxt_port_msg_close_fd(msg); msg->buf = nxt_port_buf_completion(task, wq, msg->buf, plain_size, @@ -1119,6 +1156,8 @@ nxt_port_socket_cancel(nxt_task_t *task, nxt_port_t *port, nxt_uint_t type, break; } + nxt_port_msg_fd_uncount_locked(port, msg); + nxt_queue_remove(&msg->link); msg->link.next = NULL; @@ -1192,6 +1231,51 @@ nxt_port_msg_first(nxt_port_t *port) } +nxt_inline nxt_bool_t +nxt_port_msg_has_fd(const nxt_port_send_msg_t *msg) +{ + return msg->fd[0] != -1 || msg->fd[1] != -1; +} + + +/* + * Stop counting a queued message against NXT_PORT_MAX_FD_MSGS. + * + * The count follows one invariant: an entry of port->messages is counted + * exactly while it carries a descriptor. So this runs at the two moments + * that end it -- the message leaves the queue, or it is still queued but has + * just had its descriptors sent -- and must run before whatever clears + * msg->fd[], since that is what it reads. msg->link.next is the queue's own + * "is it in the list" marker, which keeps the inline stack copy in + * nxt_port_socket_write2() out of the count. + * + * The _locked form is for the callers that already hold port->write_mutex. + */ + +nxt_inline void +nxt_port_msg_fd_uncount_locked(nxt_port_t *port, nxt_port_send_msg_t *msg) +{ + if (msg->link.next != NULL && nxt_port_msg_has_fd(msg)) { + port->fd_messages--; + } +} + + +static void +nxt_port_msg_fd_uncount(nxt_port_t *port, nxt_port_send_msg_t *msg) +{ + if (msg->link.next == NULL || !nxt_port_msg_has_fd(msg)) { + return; + } + + nxt_thread_mutex_lock(&port->write_mutex); + + port->fd_messages--; + + nxt_thread_mutex_unlock(&port->write_mutex); +} + + nxt_inline void nxt_port_msg_close_fd(nxt_port_send_msg_t *msg) { @@ -1282,21 +1366,50 @@ nxt_port_buf_completion(nxt_task_t *task, nxt_work_queue_t *wq, nxt_buf_t *b, } +/* + * The bound applies here as well: this is the other way a descriptor-carrying + * message enters port->messages, when an inline first fragment hits EAGAIN and + * has to be held for a later attempt. Answering NULL is what an allocation + * failure already answers, and nxt_port_write_msgs() turns both into the same + * "nothing was consumed" NXT_ERROR for a first fragment. + */ + static nxt_port_send_msg_t * nxt_port_msg_insert_tail(nxt_port_t *port, nxt_port_send_msg_t *msg) { + nxt_bool_t has_fd; + + has_fd = nxt_port_msg_has_fd(msg); + + nxt_thread_mutex_lock(&port->write_mutex); + + if (nxt_slow_path(has_fd && port->fd_messages >= NXT_PORT_MAX_FD_MSGS)) { + nxt_thread_mutex_unlock(&port->write_mutex); + + nxt_thread_log_alert("port{%d,%d}: %d queued messages already hold a " + "descriptor, refusing to queue another", + (int) port->pid, (int) port->id, + NXT_PORT_MAX_FD_MSGS); + + return NULL; + } + if (msg->allocated == 0) { msg = nxt_port_msg_alloc(msg); if (nxt_slow_path(msg == NULL)) { + nxt_thread_mutex_unlock(&port->write_mutex); + return NULL; } } - nxt_thread_mutex_lock(&port->write_mutex); - nxt_queue_insert_tail(&port->messages, &msg->link); + if (has_fd) { + port->fd_messages++; + } + nxt_thread_mutex_unlock(&port->write_mutex); return msg; @@ -2148,6 +2261,8 @@ nxt_port_error_handler(nxt_task_t *task, void *obj, void *data) nxt_queue_each(msg, &port->messages, nxt_port_send_msg_t, link) { + nxt_port_msg_fd_uncount_locked(port, msg); + nxt_port_msg_close_fd(msg); for (b = msg->buf; b != NULL; b = next) { diff --git a/src/test/nxt_port_queued_fd_test.c b/src/test/nxt_port_queued_fd_test.c index a99d63154..0ff4b4aa5 100644 --- a/src/test/nxt_port_queued_fd_test.c +++ b/src/test/nxt_port_queued_fd_test.c @@ -21,6 +21,14 @@ * reopen something else at the same number, then let the write handler * send. The peer must receive the descriptor that was queued, not the * one that took its number. + * + * A third leg covers the bound that ownership made necessary + * (NXT_PORT_MAX_FD_MSGS, freeunitorg/freeunit#394). A peer that stops + * reading holds two of this process's descriptors per queued message, so + * the number of descriptor-carrying entries is capped. The leg fills a + * port to the cap, checks that the next descriptor-carrying send is + * refused with nothing consumed, that a send with no descriptor is still + * accepted, and that nothing is leaked either way. */ #include @@ -31,10 +39,12 @@ #include #include #include +#include static nxt_int_t nxt_port_queued_fd_test_owned(nxt_thread_t *thr); static nxt_int_t nxt_port_queued_fd_test_reused(nxt_thread_t *thr); +static nxt_int_t nxt_port_queued_fd_test_bounded(nxt_thread_t *thr); static nxt_port_t *nxt_port_queued_fd_test_port(nxt_task_t *task, nxt_event_engine_t *engine); static void nxt_port_queued_fd_test_stub(nxt_event_engine_t *engine, @@ -45,6 +55,8 @@ static void nxt_port_queued_fd_test_drain_wq(nxt_work_queue_t *wq); static nxt_bool_t nxt_port_queued_fd_test_same_file(nxt_fd_t a, const struct stat *b); static nxt_fd_t nxt_port_queued_fd_test_recv_fd(nxt_fd_t sock); +static nxt_uint_t nxt_port_queued_fd_test_open_fds(void); +static nxt_uint_t nxt_port_queued_fd_test_queued(nxt_port_t *port); static nxt_uint_t nxt_port_queued_fd_test_completions; @@ -64,6 +76,10 @@ nxt_port_queued_fd_test(nxt_thread_t *thr) return NXT_ERROR; } + if (nxt_port_queued_fd_test_bounded(thr) != NXT_OK) { + return NXT_ERROR; + } + nxt_thread_time_update(thr); nxt_log_error(NXT_LOG_NOTICE, thr->log, "port queued fd test passed"); @@ -501,6 +517,346 @@ nxt_port_queued_fd_test_reused(nxt_thread_t *thr) } +/* + * The bound on descriptor-carrying entries. + * + * The port is not writable, so every send queues, and each queued copy dups + * what it was given: the open descriptor count rises by one per message and + * is the measurement the leg makes. Nothing here reaches inside the port to + * read a counter -- the bound is observed the way a caller observes it, from + * the answer to a send and from what the process still holds open. + */ + +static nxt_int_t +nxt_port_queued_fd_test_bounded(nxt_thread_t *thr) +{ + nxt_mp_t *mp; + nxt_uint_t i, base, filled; + nxt_fd_t fd, got, pair[2]; + nxt_buf_t *buf; + nxt_int_t ret; + struct stat st; + nxt_task_t *task; + nxt_port_t *port; + nxt_event_engine_t engine, *saved_engine; + nxt_event_interface_t stub; + + task = thr->task; + task->thread = thr; + + ret = NXT_ERROR; + fd = -1; + got = -1; + pair[0] = -1; + pair[1] = -1; + + nxt_memzero(&engine, sizeof(engine)); + nxt_memzero(&stub, sizeof(stub)); + + nxt_work_queue_cache_create(&engine.work_queue_cache, 1024); + engine.fast_work_queue.cache = &engine.work_queue_cache; + nxt_work_queue_name(&engine.fast_work_queue, "fast"); + + stub.enable_write = nxt_port_queued_fd_test_stub; + stub.block_write = nxt_port_queued_fd_test_stub; + engine.event = stub; + + saved_engine = thr->engine; + thr->engine = &engine; + + mp = nxt_mp_create(1024, 128, 256, 32); + if (nxt_slow_path(mp == NULL)) { + thr->engine = saved_engine; + return NXT_ERROR; + } + + port = nxt_port_queued_fd_test_port(task, &engine); + if (nxt_slow_path(port == NULL)) { + nxt_mp_destroy(mp); + thr->engine = saved_engine; + return NXT_ERROR; + } + + if (nxt_slow_path(socketpair(AF_UNIX, SOCK_DGRAM, 0, pair) != 0)) { + nxt_log_error(NXT_LOG_NOTICE, thr->log, + "port queued fd test: socketpair failed"); + goto done; + } + + if (nxt_slow_path(fcntl(pair[1], F_SETFL, O_NONBLOCK) == -1)) { + goto done; + } + + port->pair[0] = pair[0]; + port->pair[1] = pair[1]; + + nxt_port_write_enable(task, port); + + port->socket.write_ready = 0; + port->socket.write = NXT_EVENT_INACTIVE; + + fd = open("/dev/null", O_RDONLY); + if (fd == -1 || fstat(fd, &st) != 0) { + nxt_log_error(NXT_LOG_NOTICE, thr->log, + "port queued fd test: failed to open /dev/null"); + goto done; + } + + nxt_port_queued_fd_test_completions = 0; + + base = nxt_port_queued_fd_test_open_fds(); + + /* Fill the queue to the bound. Every one of these must be taken. */ + + for (i = 0; i < NXT_PORT_MAX_FD_MSGS; i++) { + if (nxt_port_socket_write2(task, port, NXT_PORT_MSG_NEW_PORT, fd, -1, + i, 0, NULL) + != NXT_OK) + { + nxt_log_error(NXT_LOG_NOTICE, thr->log, + "port queued fd test: message %ui of %d was " + "refused below the bound", i, NXT_PORT_MAX_FD_MSGS); + goto done; + } + } + + filled = nxt_port_queued_fd_test_queued(port); + + if (filled != NXT_PORT_MAX_FD_MSGS) { + nxt_log_error(NXT_LOG_NOTICE, thr->log, + "port queued fd test: %ui messages queued, expected %d", + filled, NXT_PORT_MAX_FD_MSGS); + goto done; + } + + if (nxt_port_queued_fd_test_open_fds() != base + NXT_PORT_MAX_FD_MSGS) { + nxt_log_error(NXT_LOG_NOTICE, thr->log, + "port queued fd test: %ui descriptors open after the " + "fill, expected %ui", + nxt_port_queued_fd_test_open_fds(), + base + NXT_PORT_MAX_FD_MSGS); + goto done; + } + + /* + * One more, with a payload. It must be refused, and refused whole: the + * descriptor is still the caller's and open, the buffer is still the + * caller's and uncompleted, and the queue is as it was. + */ + + buf = nxt_buf_mem_alloc(mp, 1, 0); + if (nxt_slow_path(buf == NULL)) { + goto done; + } + + buf->completion_handler = nxt_port_queued_fd_test_completion; + + if (nxt_port_socket_write2(task, port, NXT_PORT_MSG_NEW_PORT, fd, -1, + NXT_PORT_MAX_FD_MSGS, 0, buf) + != NXT_ERROR) + { + nxt_log_error(NXT_LOG_NOTICE, thr->log, + "port queued fd test: a message past the bound of %d " + "was accepted", NXT_PORT_MAX_FD_MSGS); + goto done; + } + + nxt_port_queued_fd_test_drain_wq(&engine.fast_work_queue); + + if (nxt_port_queued_fd_test_completions != 0) { + nxt_log_error(NXT_LOG_NOTICE, thr->log, + "port queued fd test: the refused message completed " + "%ui buffers, expected 0", + nxt_port_queued_fd_test_completions); + goto done; + } + + if (!nxt_test_fd_is_open(fd)) { + nxt_log_error(NXT_LOG_NOTICE, thr->log, + "port queued fd test: the refused message closed the " + "caller's descriptor"); + fd = -1; + goto done; + } + + if (nxt_port_queued_fd_test_queued(port) != NXT_PORT_MAX_FD_MSGS + || nxt_port_queued_fd_test_open_fds() != base + NXT_PORT_MAX_FD_MSGS) + { + nxt_log_error(NXT_LOG_NOTICE, thr->log, + "port queued fd test: the refused message changed the " + "queue"); + goto done; + } + + /* + * A message with no descriptor is not what the bound is about, and is + * still taken on the same full port. + */ + + if (nxt_port_socket_write2(task, port, NXT_PORT_MSG_DATA, -1, -1, + NXT_PORT_MAX_FD_MSGS + 1, 0, buf) + != NXT_OK) + { + nxt_log_error(NXT_LOG_NOTICE, thr->log, + "port queued fd test: a message with no descriptor was " + "refused on a port at the bound"); + goto done; + } + + if (nxt_port_queued_fd_test_queued(port) != NXT_PORT_MAX_FD_MSGS + 1) { + nxt_log_error(NXT_LOG_NOTICE, thr->log, + "port queued fd test: the message with no descriptor " + "was not queued"); + goto done; + } + + /* + * Draining is what releases the bound. The peer gets what was queued + * below it, and the duplicates go with the messages: the descriptor + * count comes back to where it started. + */ + + port->socket.write_ready = 1; + + port->socket.write_handler(task, &port->socket, NULL); + + nxt_port_queued_fd_test_drain_wq(&engine.fast_work_queue); + + got = nxt_port_queued_fd_test_recv_fd(pair[0]); + + if (got == -1 || !nxt_port_queued_fd_test_same_file(got, &st)) { + nxt_log_error(NXT_LOG_NOTICE, thr->log, + "port queued fd test: the peer did not receive the " + "first message queued below the bound"); + goto done; + } + + nxt_fd_close(got); + got = -1; + + /* Whatever the socket would not take is dropped the ordinary way. */ + + if (!nxt_queue_is_empty(&port->messages)) { + nxt_port_test_run_error_handler(task, port); + nxt_port_queued_fd_test_drain_wq(&engine.fast_work_queue); + } + + while ((got = nxt_port_queued_fd_test_recv_fd(pair[0])) != -1) { + nxt_fd_close(got); + } + + if (nxt_port_queued_fd_test_open_fds() != base) { + nxt_log_error(NXT_LOG_NOTICE, thr->log, + "port queued fd test: %ui descriptors open after the " + "queue drained, expected the %ui it started with", + nxt_port_queued_fd_test_open_fds(), base); + goto done; + } + + /* And the port takes descriptor-carrying messages again. */ + + if (nxt_port_socket_write2(task, port, NXT_PORT_MSG_NEW_PORT, fd, -1, + 0, 0, NULL) + != NXT_OK) + { + nxt_log_error(NXT_LOG_NOTICE, thr->log, + "port queued fd test: the drained port still refuses a " + "descriptor"); + goto done; + } + + ret = NXT_OK; + +done: + + if (fd != -1 && nxt_test_fd_is_open(fd)) { + nxt_fd_close(fd); + } + + if (got != -1) { + nxt_fd_close(got); + } + + if (!nxt_queue_is_empty(&port->messages)) { + nxt_port_test_run_error_handler(task, port); + nxt_port_queued_fd_test_drain_wq(&engine.fast_work_queue); + } + + port->pair[0] = -1; + port->pair[1] = -1; + port->socket.fd = -1; + + if (pair[0] != -1) { + nxt_fd_close(pair[0]); + } + + if (pair[1] != -1) { + nxt_fd_close(pair[1]); + } + + nxt_port_use(task, port, -1); + nxt_mp_destroy(mp); + + nxt_work_queue_cache_destroy(&engine.work_queue_cache); + thr->engine = saved_engine; + + return ret; +} + + +/* + * How many descriptors this process holds open. Counted by probing numbers + * rather than by reading /proc, which is not there on every platform the C + * suite builds on. + */ + +static nxt_uint_t +nxt_port_queued_fd_test_open_fds(void) +{ + nxt_fd_t fd; + nxt_uint_t n, limit; + struct rlimit rlmt; + + limit = 4096; + + if (getrlimit(RLIMIT_NOFILE, &rlmt) == 0 + && rlmt.rlim_cur != RLIM_INFINITY + && rlmt.rlim_cur < limit) + { + limit = rlmt.rlim_cur; + } + + n = 0; + + for (fd = 0; fd < (nxt_fd_t) limit; fd++) { + if (nxt_test_fd_is_open(fd)) { + n++; + } + } + + return n; +} + + +static nxt_uint_t +nxt_port_queued_fd_test_queued(nxt_port_t *port) +{ + nxt_uint_t n; + nxt_queue_link_t *lnk; + + n = 0; + + for (lnk = nxt_queue_first(&port->messages); + lnk != nxt_queue_tail(&port->messages); + lnk = nxt_queue_next(lnk)) + { + n++; + } + + return n; +} + + static void nxt_port_queued_fd_test_completion(nxt_task_t *task, void *obj, void *data) {