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) {