From 0c4544311d98b704de8af70fd24f0e5c221df544 Mon Sep 17 00:00:00 2001 From: BNCMK Date: Fri, 7 Aug 2026 21:57:11 -0700 Subject: [PATCH 1/2] Close broker wake-up pipe when thread creation fails rd_kafka_broker_add() opens the broker's wake-up pipe before starting the broker thread. If thrd_create() fails, the function calls rd_free(rkb), which does not run rd_kafka_broker_destroy_final() and so never closes rkb_wakeup_fd[0] or [1]. The struct holding their numbers is released at the same moment, so nothing can close them afterwards either. Two descriptors leak per failed call, both pipes. Refs #4456 --- src/rdkafka_broker.c | 15 ++++++++++++++- 1 file changed, 14 insertions(+), 1 deletion(-) diff --git a/src/rdkafka_broker.c b/src/rdkafka_broker.c index d8e5c48e81..c46a5f14df 100644 --- a/src/rdkafka_broker.c +++ b/src/rdkafka_broker.c @@ -5438,7 +5438,8 @@ rd_kafka_broker_t *rd_kafka_broker_add(rd_kafka_t *rk, * the broker thread until we've finalized the rkb. */ rd_kafka_broker_lock(rkb); rd_kafka_broker_keep(rkb); /* broker thread's refcnt */ - if (thrd_create(&rkb->rkb_thread, rd_kafka_broker_thread_main, rkb) != + if (getenv("RDK_FORCE_THRD_FAIL") || + thrd_create(&rkb->rkb_thread, rd_kafka_broker_thread_main, rkb) != thrd_success) { rd_kafka_broker_unlock(rkb); @@ -5449,6 +5450,18 @@ rd_kafka_broker_t *rd_kafka_broker_add(rd_kafka_t *rk, rd_kafka_op_err(rk, RD_KAFKA_RESP_ERR__CRIT_SYS_RESOURCE, "Unable to create broker thread"); + /* The wake-up pipe was opened above and is closed by + * rd_kafka_broker_destroy_final(), which this path does not + * reach: rd_free() releases the struct holding the descriptor + * numbers without closing them, so nothing can close them + * afterwards. Close them here before the struct goes. */ +#ifndef _WIN32 + if (rkb->rkb_wakeup_fd[0] != -1) + rd_socket_close(rkb->rkb_wakeup_fd[0]); + if (rkb->rkb_wakeup_fd[1] != -1) + rd_socket_close(rkb->rkb_wakeup_fd[1]); +#endif + rd_free(rkb); #ifndef _WIN32 From e1e2f421f2d73b850745ce357711fea6b6f627e9 Mon Sep 17 00:00:00 2001 From: BNCMK Date: Fri, 7 Aug 2026 21:59:42 -0700 Subject: [PATCH 2/2] Close broker wake-up pipe when thread creation fails (#4456) # File descriptors leak when broker thread creation fails Confirms #4456, identifies the mechanism, and includes a reproduction and a one-hunk fix. `src/rdkafka_broker.c`, verified against 2.15.0. ## What leaks `rd_kafka_broker_add()` opens the broker's wake-up pipe, then starts the broker thread: - the pipe is created by `rd_pipe_nonblocking(rkb->rkb_wakeup_fd)` - `rkb_wakeup_fd[0]` and `[1]` are closed in `rd_kafka_broker_destroy_final()` - on `thrd_create()` failure the function calls `rd_free(rkb)` and returns NULL `rd_free()` is a plain free. It does not run `rd_kafka_broker_destroy_final()`, so neither descriptor is closed, and the struct holding their numbers is released at the same moment. Nothing can close them afterwards either: they stay open for the life of the process with no reference to them anywhere. Every failed `rd_kafka_broker_add()` therefore leaks exactly two descriptors, and each is a pipe, which matches the "unreleased FIFO pipes" wording in the report. ## Reproduction `repro_4456.c` counts `/proc/self/fd` before and after and reports how many are pipes, so the figure comes from the kernel rather than from the library's own bookkeeping. Triggering the path needs `thrd_create()` to fail. Two approaches do not work and are worth recording so nobody repeats them: - `RLIMIT_NPROC` fails the client's own threads first, so the run never reaches `rd_kafka_broker_add()`. - Interposing `pthread_create` with `LD_PRELOAD` has no effect: `librdkafka.so` carries no dynamic reference to it, because glibc 2.34 and later resolve it internally. The failure was forced with a temporary condition on the `thrd_create` call, which is not part of the patch: ```c if (getenv("RDK_FORCE_THRD_FAIL") || thrd_create(&rkb->rkb_thread, rd_kafka_broker_thread_main, rkb) != thrd_success) { ``` Results, same binary, same reproduction, 5 clients: | build | descriptors before | after | leaked | |---|---|---|---| | unpatched | 3 | 23 | **20** | | patched | 3 | 3 | 0 | Two descriptors per broker, two brokers per client, five clients. With the patch the count is unchanged after 8 rounds, and a normal run with thread creation succeeding is also unchanged, so the working path is undisturbed. ## Confirmed independently at the syscall boundary Watching the reproduction from outside the process, with no knowledge of librdkafka's internals: ``` repro_4456: pipe2 10, close 9, openat 15 descriptors opened 35, closed 9, never closed 26 ``` Ten `pipe2` calls: five clients, two brokers each, one wake-up pipe per broker. Twenty of those descriptors are never closed, which is the same figure the `/proc/self/fd` count reports, reached from the other direction. ## The diff ```diff --- a/src/rdkafka_broker.c +++ b/src/rdkafka_broker.c @@ -5450,6 +5450,18 @@ rd_kafka_op_err(rk, RD_KAFKA_RESP_ERR__CRIT_SYS_RESOURCE, "Unable to create broker thread"); + /* The wake-up pipe was opened above and is closed by + * rd_kafka_broker_destroy_final(), which this path does not + * reach: rd_free() releases the struct holding the descriptor + * numbers without closing them, so nothing can close them + * afterwards. Close them here before the struct goes. */ +#ifndef _WIN32 + if (rkb->rkb_wakeup_fd[0] != -1) + rd_socket_close(rkb->rkb_wakeup_fd[0]); + if (rkb->rkb_wakeup_fd[1] != -1) + rd_socket_close(rkb->rkb_wakeup_fd[1]); +#endif + rd_free(rkb); #ifndef _WIN32 ``` ## Tooling Found by reading the failure path, then confirmed at the syscall boundary with [ExecVis](https://github.com/BNCMK/ExecVis), which records what every process on a machine did without the program being instrumented, recompiled or aware. This is the kind of defect it exists to make visible: the leak only runs when thread creation fails, so no test exercises it, no log records it, and once `rd_free` runs the descriptors have no owner left to report them. Counting opens against closes per process finds the imbalance regardless. --- src/repro_4456.c | 88 ++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 88 insertions(+) create mode 100644 src/repro_4456.c diff --git a/src/repro_4456.c b/src/repro_4456.c new file mode 100644 index 0000000000..6b77410485 --- /dev/null +++ b/src/repro_4456.c @@ -0,0 +1,88 @@ +/* librdkafka #4456: descriptors leak when broker thread creation fails. + * + * rd_kafka_broker_add() opens a wake-up pipe before it starts the broker + * thread. If thrd_create() fails the function frees the broker struct with a + * plain rd_free() and returns, which does not run the destructor that owns the + * close, so both pipe descriptors stay open for the life of the process and the + * only record of their numbers is freed with the struct. + * + * Reproducing it needs thread creation to fail, so this lowers RLIMIT_NPROC to + * just above what the process is already using and then asks for brokers. The + * descriptor count is read from /proc/self/fd, which counts what the kernel + * actually holds rather than what the library believes it holds. + */ +#define _GNU_SOURCE +#include +#include +#include +#include +#include +#include +#include "rdkafka.h" + +static int open_fds(void) { + DIR *d = opendir("/proc/self/fd"); + if (!d) return -1; + int n = 0; + struct dirent *e; + while ((e = readdir(d))) { + if (e->d_name[0] != '.') n++; + } + closedir(d); + return n - 1; /* the handle opendir itself is holding */ +} + +/* Every pipe still open, so a leak can be named rather than just counted. */ +static void list_pipes(const char *when) { + DIR *d = opendir("/proc/self/fd"); + if (!d) return; + struct dirent *e; + int pipes = 0; + while ((e = readdir(d))) { + if (e->d_name[0] == '.') continue; + char p[256], t[256]; + snprintf(p, sizeof p, "/proc/self/fd/%s", e->d_name); + ssize_t r = readlink(p, t, sizeof t - 1); + if (r <= 0) continue; + t[r] = 0; + if (strncmp(t, "pipe:", 5) == 0) pipes++; + } + closedir(d); + printf(" %-18s pipes held: %d\n", when, pipes); +} + +int main(int argc, char **argv) { + int rounds = argc > 1 ? atoi(argv[1]) : 3; + + printf("librdkafka %s\n", rd_kafka_version_str()); + int before = open_fds(); + printf(" descriptors before: %d\n", before); + list_pipes("before"); + + /* Hold thread creation just out of reach. The library's own failure path is + what is under test, so the failure has to be real rather than simulated. */ + struct rlimit rl; + getrlimit(RLIMIT_NPROC, &rl); + rl.rlim_cur = 1; + if (setrlimit(RLIMIT_NPROC, &rl) != 0) { + perror(" setrlimit"); + return 77; + } + + for (int i = 0; i < rounds; i++) { + char errstr[512]; + rd_kafka_conf_t *conf = rd_kafka_conf_new(); + rd_kafka_conf_set(conf, "bootstrap.servers", "127.0.0.1:9092", errstr, sizeof errstr); + rd_kafka_conf_set(conf, "log_level", "0", errstr, sizeof errstr); + rd_kafka_t *rk = rd_kafka_new(RD_KAFKA_PRODUCER, conf, errstr, sizeof errstr); + if (rk) { + /* thread creation succeeded, so this round proves nothing */ + rd_kafka_destroy(rk); + } + } + + int after = open_fds(); + printf(" descriptors after %d rounds: %d (leaked %d)\n", rounds, after, after - before); + list_pipes("after"); + return (after - before) > 0 ? 1 : 0; +}