From: Olivier Houchard Date: Wed, 8 Jul 2026 09:20:40 +0000 (+0200) Subject: MEDIUM: cli: Transfer sockets with unshared file descriptor tables X-Git-Url: http://git.ipfire.org/gitweb/?a=commitdiff_plain;h=ff8840f28cf5cbe1811f70355f97cbab408b6652;p=thirdparty%2Fhaproxy.git MEDIUM: cli: Transfer sockets with unshared file descriptor tables Teach the _getsocks command how to use the new rx agent facilities to obtain the file descriptors from other thread groups when each one has its own file descriptor table. --- diff --git a/include/haproxy/listener.h b/include/haproxy/listener.h index 6013683fc..a872b6850 100644 --- a/include/haproxy/listener.h +++ b/include/haproxy/listener.h @@ -185,6 +185,8 @@ void default_unbind_listener(struct listener *listener); extern int tg_agents_enabled; int rx_agent_init(void); void rx_agent_close(struct receiver *rx); +void rx_agent_getsocks_request(struct receiver *rx, uint dest_grp); +void rx_xfer_drain(uint grp); /* default function called to suspend a listener: it simply passes the call to * the underlying receiver. This is find for most socket-based protocols. This diff --git a/include/haproxy/protocol.h b/include/haproxy/protocol.h index 77c3aac40..95cfde0cf 100644 --- a/include/haproxy/protocol.h +++ b/include/haproxy/protocol.h @@ -65,6 +65,11 @@ void protocol_init_rx_agents(void); */ void protocol_localize_rx_fds(void); +/* + * Collects fd of other thread groups to be sent when _getsocks is used. + */ +int protocol_getsocks_foreign_fds(struct receiver ***orxs, int **ofds); + /* binds all listeners of all registered protocols. Returns a composition * of ERR_NONE, ERR_RETRYABLE, ERR_FATAL, ERR_ABORT. */ diff --git a/include/haproxy/receiver-t.h b/include/haproxy/receiver-t.h index 89dd97cef..34551cff9 100644 --- a/include/haproxy/receiver-t.h +++ b/include/haproxy/receiver-t.h @@ -93,7 +93,9 @@ struct receiver { struct rx_agent_link link; /* position in the owner group agent's queue */ int close_fd; /* FD to release, -1 if none */ uint ops; /* pending RX_AGENT_OP_* */ + uint getsocks_grp; /* group requesting an FD copy for _getsocks */ int xfer_fd; /* FD copy received for a rebind, -1 if none */ + int getsocks_fd; /* FD copy received for _getsocks, -1 if none */ } agent; /* only used with per-tgroup FD tables */ struct list proto_list; /* list in the protocol header */ #ifdef USE_QUIC diff --git a/src/cli.c b/src/cli.c index 8f3c74727..6fb3f4a56 100644 --- a/src/cli.c +++ b/src/cli.c @@ -2444,6 +2444,80 @@ static int bind_parse_severity_output(char **args, int cur_arg, struct proxy *px } } +/* + * Send one FD, with its relevant informations. + */ +static int _getsocks_send_one(int sock, int send_fd, const struct receiver *rx, + struct msghdr *msghdr, unsigned char *tmpbuf, + int *tmpfd, int *nb_queued, int *curoff) +{ + const char *ns_name, *if_name; + unsigned char ns_nlen, if_nlen; + + ns_name = if_name = ""; + ns_nlen = if_nlen = 0; + + /* for now we can only retrieve namespaces and interfaces from + * pure listeners. + */ + if (rx && rx->iocb == sock_accept_iocb) { + if (rx->settings->interface) { + if_name = rx->settings->interface; + if_nlen = strlen(if_name); + } + +#ifdef USE_NS + if (rx->settings->netns) { + ns_name = rx->settings->netns->node.key; + ns_nlen = rx->settings->netns->name_len; + } +#endif + } + + /* put the FD into the CMSG_DATA */ + tmpfd[(*nb_queued)++] = send_fd; + + /* first block is */ + tmpbuf[(*curoff)++] = ns_nlen; + if (ns_nlen) + memcpy(tmpbuf + *curoff, ns_name, ns_nlen); + *curoff += ns_nlen; + + /* second block is */ + tmpbuf[(*curoff)++] = if_nlen; + if (if_nlen) + memcpy(tmpbuf + *curoff, if_name, if_nlen); + *curoff += if_nlen; + + /* we used to send the listener options here before 2.3 */ + memset(tmpbuf + *curoff, 0, sizeof(int)); + *curoff += sizeof(int); + + /* there's a limit to how many FDs may be sent at once */ + if (*nb_queued == MAX_SEND_FD) { + int ack, ret; + + msghdr->msg_iov->iov_len = *curoff; + if (sendmsg(sock, msghdr, 0) != *curoff) { + ha_warning("Failed to transfer sockets\n"); + return 0; + } + + /* Wait for an ack */ + do { + ret = recv(sock, &ack, sizeof(ack), 0); + } while (ret == -1 && errno == EINTR); + + if (ret <= 0) { + ha_warning("Unexpected error while transferring sockets\n"); + return 0; + } + *curoff = 0; + *nb_queued = 0; + } + return 1; +} + /* Send all the bound sockets, always returns 1 */ static int _getsocks(char **args, char *payload, struct appctx *appctx, void *private) { @@ -2457,8 +2531,10 @@ static int _getsocks(char **args, char *payload, struct appctx *appctx, void *pr struct msghdr msghdr; struct iovec iov; struct timeval tv = { .tv_sec = 1, .tv_usec = 0 }; - const char *ns_name, *if_name; - unsigned char ns_nlen, if_nlen; + struct receiver **foreign_rxs = NULL; + int *foreign_fds = NULL; + int nb_foreign = 0; + int foreign_idx; int nb_queued; int cur_fd = 0; int *tmpfd; @@ -2496,6 +2572,17 @@ static int _getsocks(char **args, char *payload, struct appctx *appctx, void *pr if (!(strm_li(s)->bind_conf->level & ACCESS_FD_LISTENERS)) goto out; memset(&msghdr, 0, sizeof(msghdr)); + + /* + * Collect file descriptors from other thread groups, if they have + * a separate file descriptor table. + */ + nb_foreign = protocol_getsocks_foreign_fds(&foreign_rxs, &foreign_fds); + if (nb_foreign < 0) { + ha_warning("Failed to allocate memory to transfer other groups' sockets\n"); + goto out; + } + /* * First, calculates the total number of FD, so that we can let * the caller know how much it should expect. @@ -2503,6 +2590,8 @@ static int _getsocks(char **args, char *payload, struct appctx *appctx, void *pr for (cur_fd = 0;cur_fd < global.maxsock; cur_fd++) tot_fd_nb += !!(fdtab[cur_fd].state & FD_EXPORTED); + tot_fd_nb += nb_foreign; + if (tot_fd_nb == 0) { if (already_sent) ha_warning("_getsocks: attempt to get sockets but they were already sent and closed in this process!\n"); @@ -2548,73 +2637,27 @@ static int _getsocks(char **args, char *payload, struct appctx *appctx, void *pr nb_queued = 0; iov.iov_base = tmpbuf; for (cur_fd = 0; cur_fd < global.maxsock; cur_fd++) { + const struct receiver *rx = NULL; + if (!(fdtab[cur_fd].state & FD_EXPORTED)) continue; /* this FD is now shared between processes */ HA_ATOMIC_OR(&fdtab[cur_fd].state, FD_CLONED); - ns_name = if_name = ""; - ns_nlen = if_nlen = 0; + if (fdtab[cur_fd].iocb == sock_accept_iocb) + rx = &((const struct listener *)fdtab[cur_fd].owner)->rx; - /* for now we can only retrieve namespaces and interfaces from - * pure listeners. - */ - if (fdtab[cur_fd].iocb == sock_accept_iocb) { - const struct listener *l = fdtab[cur_fd].owner; - - if (l->rx.settings->interface) { - if_name = l->rx.settings->interface; - if_nlen = strlen(if_name); - } - -#ifdef USE_NS - if (l->rx.settings->netns) { - ns_name = l->rx.settings->netns->node.key; - ns_nlen = l->rx.settings->netns->name_len; - } -#endif - } - - /* put the FD into the CMSG_DATA */ - tmpfd[nb_queued++] = cur_fd; - - /* first block is */ - tmpbuf[curoff++] = ns_nlen; - if (ns_nlen) - memcpy(tmpbuf + curoff, ns_name, ns_nlen); - curoff += ns_nlen; - - /* second block is */ - tmpbuf[curoff++] = if_nlen; - if (if_nlen) - memcpy(tmpbuf + curoff, if_name, if_nlen); - curoff += if_nlen; - - /* we used to send the listener options here before 2.3 */ - memset(tmpbuf + curoff, 0, sizeof(int)); - curoff += sizeof(int); - - /* there's a limit to how many FDs may be sent at once */ - if (nb_queued == MAX_SEND_FD) { - iov.iov_len = curoff; - if (sendmsg(fd, &msghdr, 0) != curoff) { - ha_warning("Failed to transfer sockets\n"); - goto out; - } - - /* Wait for an ack */ - do { - ret = recv(fd, &tot_fd_nb, sizeof(tot_fd_nb), 0); - } while (ret == -1 && errno == EINTR); + if (!_getsocks_send_one(fd, cur_fd, rx, &msghdr, tmpbuf, + tmpfd, &nb_queued, &curoff)) + goto out; + } - if (ret <= 0) { - ha_warning("Unexpected error while transferring sockets\n"); - goto out; - } - curoff = 0; - nb_queued = 0; - } + for (foreign_idx = 0; foreign_idx < nb_foreign; foreign_idx++) { + if (!_getsocks_send_one(fd, foreign_fds[foreign_idx], + foreign_rxs[foreign_idx], &msghdr, + tmpbuf, tmpfd, &nb_queued, &curoff)) + goto out; } already_sent = 1; @@ -2641,6 +2684,10 @@ static int _getsocks(char **args, char *payload, struct appctx *appctx, void *pr } out: + for (foreign_idx = 0; foreign_idx < nb_foreign; foreign_idx++) + close(foreign_fds[foreign_idx]); + free(foreign_rxs); + free(foreign_fds); if (fd >= 0 && old_fcntl >= 0 && fcntl(fd, F_SETFL, old_fcntl) == -1) ha_warning("Cannot make the unix socket non-blocking\n"); applet_set_eoi(appctx); diff --git a/src/listener.c b/src/listener.c index 034ac1db0..29a938090 100644 --- a/src/listener.c +++ b/src/listener.c @@ -792,6 +792,7 @@ int tg_agents_enabled = 0; enum rx_xfer_op { RX_XFER_OP_REBIND = 0, + RX_XFER_OP_GETSOCKS, }; struct rx_xfer_msg { @@ -801,7 +802,6 @@ struct rx_xfer_msg { static void rx_xfer_send_members(struct receiver *ref); static int rx_xfer_send_fd(uint grp, struct receiver *rx, enum rx_xfer_op op, int fd); -static void rx_xfer_drain(uint grp); /* Queues receiver to 1-based group 's agent and wakes it. The * operation fields must be set before calling this, see struct tg_agent. @@ -815,6 +815,7 @@ static void rx_agent_post(struct receiver *rx, uint grp) static void rx_agent_process_one(struct receiver *rx) { int fd = HA_ATOMIC_XCHG(&rx->agent.close_fd, -1); + uint dest; uint ops; if (fd >= 0) @@ -844,6 +845,12 @@ static void rx_agent_process_one(struct receiver *rx) if (ops & RX_AGENT_OP_ENABLE) enable_listener(l); } + + dest = HA_ATOMIC_XCHG(&rx->agent.getsocks_grp, 0); + if (dest && rx->fd >= 0) { + HA_ATOMIC_OR(&fdtab[rx->fd].state, FD_CLONED); + rx_xfer_send_fd(dest, rx, RX_XFER_OP_GETSOCKS, rx->fd); + } } static struct task *tg_agent_process(struct task *t, void *context, unsigned int state) @@ -870,6 +877,12 @@ static void rx_agent_schedule_op(struct receiver *rx, uint op) rx_agent_post(rx, rx_owner_tgid(rx)); } +void rx_agent_getsocks_request(struct receiver *rx, uint dest_grp) +{ + HA_ATOMIC_STORE(&rx->agent.getsocks_grp, dest_grp); + rx_agent_post(rx, rx_owner_tgid(rx)); +} + static void rx_xfer_send_members(struct receiver *ref) { const struct shard_info *si = ref->shard_info; @@ -924,7 +937,7 @@ static int rx_xfer_send_fd(uint grp, struct receiver *rx, enum rx_xfer_op op, in return 1; } -static void rx_xfer_drain(uint grp) +void rx_xfer_drain(uint grp) { while (1) { struct rx_xfer_msg xmsg; @@ -974,6 +987,10 @@ static void rx_xfer_drain(uint grp) } break; + case RX_XFER_OP_GETSOCKS: + HA_ATOMIC_STORE(&rx->agent.getsocks_fd, fd); + break; + default: close(fd); break; diff --git a/src/protocol.c b/src/protocol.c index 1e743b428..a4a0332d5 100644 --- a/src/protocol.c +++ b/src/protocol.c @@ -145,6 +145,7 @@ void protocol_init_rx_agents(void) rx->agent.link.rx = rx; rx->agent.close_fd = -1; rx->agent.xfer_fd = -1; + rx->agent.getsocks_fd = -1; } } HA_SPIN_UNLOCK(PROTO_LOCK, &proto_lock); @@ -181,6 +182,116 @@ void protocol_localize_rx_fds(void) HA_SPIN_UNLOCK(PROTO_LOCK, &proto_lock); } +/* + * Collect file descriptor from other thread groups, if they have their + * own file descriptors table. + */ +int protocol_getsocks_foreign_fds(struct receiver ***orxs, int **ofds) +{ + struct protocol *proto; + struct receiver *rx; + struct receiver **rxs; + int *fds; + int nb, filled, done, i; + + *orxs = NULL; + *ofds = NULL; + + if (!(global.tune.options & GTUNE_NO_TG_FD_SHARING) || global.nbtgroups < 2) + return 0; + + /* first pass: count the candidates */ + nb = 0; + HA_SPIN_LOCK(PROTO_LOCK, &proto_lock); + list_for_each_entry(proto, &protocols, list) { + list_for_each_entry(rx, &proto->receivers, proto_list) { + if (rx->fd < 0 || rx_owner_tgid(rx) == tgid) + continue; + if (!(ha_tgroup_ctx[rx_owner_tgid(rx) - 1].fdtab[rx->fd].state & FD_EXPORTED)) + continue; + nb++; + } + } + HA_SPIN_UNLOCK(PROTO_LOCK, &proto_lock); + + if (!nb) + return 0; + + rxs = calloc(nb, sizeof(*rxs)); + fds = calloc(nb, sizeof(*fds)); + if (!rxs || !fds) { + free(rxs); + free(fds); + return -1; + } + + /* second pass: arm the requests. A copy left over by a previous + * timed-out collection is closed on the way. + */ + filled = 0; + HA_SPIN_LOCK(PROTO_LOCK, &proto_lock); + list_for_each_entry(proto, &protocols, list) { + list_for_each_entry(rx, &proto->receivers, proto_list) { + int old; + + if (filled >= nb) + break; + if (rx->fd < 0 || rx_owner_tgid(rx) == tgid) + continue; + if (!(ha_tgroup_ctx[rx_owner_tgid(rx) - 1].fdtab[rx->fd].state & FD_EXPORTED)) + continue; + + old = rx->agent.getsocks_fd; + if (old >= 0) + close(old); + HA_ATOMIC_STORE(&rx->agent.getsocks_fd, -1); + rxs[filled++] = rx; + rx_agent_getsocks_request(rx, tgid); + } + } + HA_SPIN_UNLOCK(PROTO_LOCK, &proto_lock); + nb = filled; + + for (i = 0; i < 50; i++) { + rx_xfer_drain(tgid - 1); + done = 1; + for (filled = 0; filled < nb; filled++) { + if (HA_ATOMIC_LOAD(&rxs[filled]->agent.getsocks_fd) < 0) + done = 0; + } + if (done) + break; + /* XXX: we're waiting for other thread groups to send their + * fds, let's sleep for a bit. + */ + usleep(10000); + } + + done = 0; + for (i = 0; i < nb; i++) { + int f = HA_ATOMIC_LOAD(&rxs[i]->agent.getsocks_fd); + + if (f < 0) { + ha_warning("_getsocks: could not get a copy of a listener FD from thread group %u, the new process will have to bind it itself.\n", + rx_owner_tgid(rxs[i])); + continue; + } + HA_ATOMIC_STORE(&rxs[i]->agent.getsocks_fd, -1); + rxs[done] = rxs[i]; + fds[done] = f; + done++; + } + + if (!done) { + free(rxs); + free(fds); + return 0; + } + *orxs = rxs; + *ofds = fds; + return done; +} + /* binds all listeners of all registered protocols. Returns a composition * of ERR_NONE, ERR_RETRYABLE, ERR_FATAL. */