From: Olivier Houchard Date: Tue, 7 Jul 2026 08:55:57 +0000 (+0200) Subject: MEDIUM: listener: Properly handle unshared fd tables between tgroups X-Git-Url: http://git.ipfire.org/gitweb/?a=commitdiff_plain;h=2e6ff48da257411b5a7c8e542a7a4c94b6841400;p=thirdparty%2Fhaproxy.git MEDIUM: listener: Properly handle unshared fd tables between tgroups With different FD tables for each thread group, a new mechanism is needed to get each thread group to close/open their fds, and to transfer fds from one thread group to another. So we add the concept of rx agents. Each thread group has a tasklet dedicated to that, bound to the first thread of each thread group, and are used to get those operations done. Each receiver contains context bits so that the tasklet will know which action to do when woken up. A new socketpair is also created per group, to be able to send fds to other thread groups. --- diff --git a/include/haproxy/listener.h b/include/haproxy/listener.h index be6058a1c..6013683fc 100644 --- a/include/haproxy/listener.h +++ b/include/haproxy/listener.h @@ -28,12 +28,19 @@ #include #include +#include #include #include +#include struct proxy; struct task; +static inline uint rx_owner_tgid(const struct receiver *rx) +{ + return rx->bind_tgroup ? rx->bind_tgroup : 1; +} + int li_init_per_thr(struct listener *li); /* adjust the listener's state and its proxy's listener counters if needed */ @@ -175,6 +182,10 @@ void default_add_listener(struct protocol *proto, struct listener *listener); */ void default_unbind_listener(struct listener *listener); +extern int tg_agents_enabled; +int rx_agent_init(void); +void rx_agent_close(struct receiver *rx); + /* 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 * must be called under the listener's lock. It will return non-zero on success, diff --git a/include/haproxy/protocol.h b/include/haproxy/protocol.h index 45e96ee51..77c3aac40 100644 --- a/include/haproxy/protocol.h +++ b/include/haproxy/protocol.h @@ -53,6 +53,18 @@ void protocol_setf_all(uint flag); */ int protocol_supports_flag(struct protocol *proto, uint flag); +/* + * initializes the per-thread-group agent state of every receiver, at boot, + * when FD tables are not shared between thread groups. + */ +void protocol_init_rx_agents(void); + +/* + * gives up the calling thread's group's copies of the other groups' receiver + * FDs, once all groups have unshared their FD tables. + */ +void protocol_localize_rx_fds(void); + /* 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 320e04b47..89dd97cef 100644 --- a/include/haproxy/receiver-t.h +++ b/include/haproxy/receiver-t.h @@ -68,6 +68,16 @@ struct shard_info { struct receiver **members; /* all members of the shard (one per thread group) */ }; +/* operations pending on a receiver, run by its owner group's agent */ +#define RX_AGENT_OP_SUSPEND 0x00000001 /* suspend_listener() */ +#define RX_AGENT_OP_RESUME 0x00000002 /* resume_listener() (includes rebind) */ +#define RX_AGENT_OP_ENABLE 0x00000004 /* enable_listener() */ + +struct rx_agent_link { + struct mt_list list; /* position in one group agent's queue */ + struct receiver *rx; /* the receiver this link belongs to */ +}; + /* This describes a receiver with all its characteristics (address, options, etc) */ struct receiver { int fd; /* handle we receive from (fd only for now) */ @@ -79,6 +89,12 @@ struct receiver { uint bind_tgroup; /* thread group ID: 0=global IDs, non-zero=local IDs */ struct rx_settings *settings; /* points to the settings used by this receiver */ struct shard_info *shard_info; /* points to info about the owning shard, NULL if single rx */ + struct { + 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_* */ + int xfer_fd; /* FD copy received for a rebind, -1 if none */ + } agent; /* only used with per-tgroup FD tables */ struct list proto_list; /* list in the protocol header */ #ifdef USE_QUIC enum quic_sock_mode quic_mode; /* QUIC socket allocation strategy */ diff --git a/src/haproxy.c b/src/haproxy.c index aecb97a61..373429007 100644 --- a/src/haproxy.c +++ b/src/haproxy.c @@ -3200,6 +3200,9 @@ void *run_thread_poll_loop(void *data) if (init_left == 0) protocol_enable_all(); + if (tg_agents_enabled && ti->ltid == 0) + protocol_localize_rx_fds(); + #ifdef USE_THREAD pthread_cond_broadcast(&init_cond); pthread_mutex_unlock(&init_mutex); @@ -3936,6 +3939,13 @@ int main(int argc, char **argv) } + if ((global.tune.options & GTUNE_NO_TG_FD_SHARING) && global.nbtgroups > 1) { + if (!rx_agent_init()) { + ha_alert("Failed to set up the per-thread-group agents.\n"); + exit(1); + } + } + /* start threads 2 and above */ setup_extra_threads(&run_thread_poll_loop); diff --git a/src/listener.c b/src/listener.c index fb5123326..034ac1db0 100644 --- a/src/listener.c +++ b/src/listener.c @@ -57,6 +57,8 @@ static struct task *global_listener_queue_task; ullong maxconn_reached = 0; __decl_thread(static HA_RWLOCK_T global_listener_rwlock); +static void rx_agent_schedule_op(struct receiver *rx, uint op); + /* listener status for stats */ const char* li_status_st[LI_STATE_COUNT] = { [LI_STATUS_WAITING] = "WAITING", @@ -324,6 +326,11 @@ void listener_set_state(struct listener *l, enum li_state st) */ void enable_listener(struct listener *listener) { + if (tg_agents_enabled && rx_owner_tgid(&listener->rx) != tgid) { + rx_agent_schedule_op(&listener->rx, RX_AGENT_OP_ENABLE); + return; + } + HA_RWLOCK_WRLOCK(LISTENER_LOCK, &listener->lock); /* If this listener is supposed to be only in the master, close it in @@ -539,6 +546,11 @@ int suspend_listener(struct listener *l, int lpx, int lli) if (!(l->flags & LI_F_FINALIZED) || l->state <= LI_PAUSED) goto end; + if (tg_agents_enabled && rx_owner_tgid(&l->rx) != tgid) { + rx_agent_schedule_op(&l->rx, RX_AGENT_OP_SUSPEND); + goto end; + } + if (l->rx.proto->suspend) { ret = l->rx.proto->suspend(l); /* if the suspend() fails, we don't want to change the @@ -615,6 +627,22 @@ int resume_listener(struct listener *l, int lpx, int lli) if (!(l->flags & LI_F_FINALIZED) || l->state == LI_READY) goto end; + if (tg_agents_enabled && l->state == LI_ASSIGNED && + (l->rx.flags & RX_F_MUST_DUP)) { + if (l->rx.agent.xfer_fd < 0) { + rx_agent_schedule_op(l->rx.shard_info->ref, RX_AGENT_OP_RESUME); + goto end; + } + } + else if (tg_agents_enabled && rx_owner_tgid(&l->rx) != tgid) { + /* all FD operations, including a possible rebind, must act on + * the owner group's kernel and fdtab tables: delegate to its + * agent and report success, failures will be handled there. + */ + rx_agent_schedule_op(&l->rx, RX_AGENT_OP_RESUME); + goto end; + } + if (l->rx.proto->resume) { ret = l->rx.proto->resume(l); if (!ret) @@ -747,6 +775,233 @@ void dequeue_proxy_listeners(struct proxy *px, int lpx) } +/* + * Per-thread-group agents, only set up when FD tables are not shared between + * thread groups. Each thread group runs one agent tasklet on its first thread, + * to which any thread may post the operations that can only be performed from + * inside that group. + */ +struct tg_agent { + struct tasklet *tl; /* runs on the group's first thread */ + struct mt_list ops; /* receivers with pending operations */ + int xfer_sock[2]; /* [0]=any sender, [1]=this group's agent */ +}; + +static struct tg_agent tg_agents[MAX_TGROUPS]; +int tg_agents_enabled = 0; + +enum rx_xfer_op { + RX_XFER_OP_REBIND = 0, +}; + +struct rx_xfer_msg { + struct receiver *rx; + enum rx_xfer_op op; +}; + +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. + */ +static void rx_agent_post(struct receiver *rx, uint grp) +{ + if (MT_LIST_TRY_APPEND(&tg_agents[grp - 1].ops, &rx->agent.link.list)) + tasklet_wakeup(tg_agents[grp - 1].tl); +} + +static void rx_agent_process_one(struct receiver *rx) +{ + int fd = HA_ATOMIC_XCHG(&rx->agent.close_fd, -1); + uint ops; + + if (fd >= 0) + fd_delete(fd); + + ops = HA_ATOMIC_XCHG(&rx->agent.ops, 0); + if (ops) { + struct listener *l = LIST_ELEM(rx, struct listener *, rx); + + if (ops & RX_AGENT_OP_SUSPEND) + suspend_listener(l, 0, 0); + + if (ops & RX_AGENT_OP_RESUME) { + resume_listener(l, 0, 0); + + /* + * if this is the reference of a shard spanning several + * groups, pass a copy of the (re)bound FD to the groups + * of the members that still need one, their own group + * will finish their resume with it. + */ + if ((rx->flags & RX_F_BOUND) && rx->shard_info && + rx->shard_info->ref == rx) + rx_xfer_send_members(rx); + } + + if (ops & RX_AGENT_OP_ENABLE) + enable_listener(l); + } +} + +static struct task *tg_agent_process(struct task *t, void *context, unsigned int state) +{ + uint grp = (uint)(ulong)context; + struct rx_agent_link *lnk; + + while ((lnk = MT_LIST_POP(&tg_agents[grp].ops, struct rx_agent_link *, list))) + rx_agent_process_one(lnk->rx); + + rx_xfer_drain(grp); + return t; +} + +void rx_agent_close(struct receiver *rx) +{ + HA_ATOMIC_STORE(&rx->agent.close_fd, rx->fd); + rx_agent_post(rx, rx_owner_tgid(rx)); +} + +static void rx_agent_schedule_op(struct receiver *rx, uint op) +{ + HA_ATOMIC_OR(&rx->agent.ops, op); + 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; + uint i; + + for (i = 0; i < si->nbgroups; i++) { + struct receiver *rx = si->members[i]; + uint grp; + + if (rx == ref || (rx->flags & RX_F_BOUND)) + continue; + + grp = rx_owner_tgid(rx); + rx_xfer_send_fd(grp, rx, RX_XFER_OP_REBIND, ref->fd); + } +} + +static int rx_xfer_send_fd(uint grp, struct receiver *rx, enum rx_xfer_op op, int fd) +{ + struct rx_xfer_msg xmsg; + struct msghdr msg; + struct iovec iov; + struct cmsghdr *cmsg; + union { + char buf[CMSG_SPACE(sizeof(int))]; + struct cmsghdr align; + } u; + + xmsg.rx = rx; + xmsg.op = op; + + memset(&msg, 0, sizeof(msg)); + memset(&u, 0, sizeof(u)); + iov.iov_base = &xmsg; + iov.iov_len = sizeof(xmsg); + msg.msg_iov = &iov; + msg.msg_iovlen = 1; + msg.msg_control = u.buf; + msg.msg_controllen = sizeof(u.buf); + cmsg = CMSG_FIRSTHDR(&msg); + cmsg->cmsg_level = SOL_SOCKET; + cmsg->cmsg_type = SCM_RIGHTS; + cmsg->cmsg_len = CMSG_LEN(sizeof(int)); + memcpy(CMSG_DATA(cmsg), &fd, sizeof(int)); + + if (sendmsg(tg_agents[grp - 1].xfer_sock[0], &msg, MSG_DONTWAIT) < 0) { + ha_warning("Failed to transfer a listener FD copy to thread group %u (%s).\n", + grp, strerror(errno)); + return 0; + } + tasklet_wakeup(tg_agents[grp - 1].tl); + return 1; +} + +static void rx_xfer_drain(uint grp) +{ + while (1) { + struct rx_xfer_msg xmsg; + struct receiver *rx; + struct listener *l; + struct msghdr msg; + struct iovec iov; + struct cmsghdr *cmsg; + union { + char buf[CMSG_SPACE(sizeof(int))]; + struct cmsghdr align; + } u; + ssize_t ret; + int fd = -1; + + memset(&msg, 0, sizeof(msg)); + iov.iov_base = &xmsg; + iov.iov_len = sizeof(xmsg); + msg.msg_iov = &iov; + msg.msg_iovlen = 1; + msg.msg_control = u.buf; + msg.msg_controllen = sizeof(u.buf); + + ret = recvmsg(tg_agents[grp].xfer_sock[1], &msg, MSG_DONTWAIT | MSG_CMSG_CLOEXEC); + if (ret <= 0) + break; + + cmsg = CMSG_FIRSTHDR(&msg); + if (cmsg && cmsg->cmsg_level == SOL_SOCKET && cmsg->cmsg_type == SCM_RIGHTS) + memcpy(&fd, CMSG_DATA(cmsg), sizeof(int)); + + if (ret != sizeof(xmsg) || fd < 0) { + if (fd >= 0) + close(fd); + continue; + } + rx = xmsg.rx; + + switch (xmsg.op) { + case RX_XFER_OP_REBIND: + rx->agent.xfer_fd = fd; + l = LIST_ELEM(rx, struct listener *, rx); + resume_listener(l, 0, 0); + if (rx->agent.xfer_fd >= 0) { + close(rx->agent.xfer_fd); + rx->agent.xfer_fd = -1; + } + break; + + default: + close(fd); + break; + } + } +} + +int rx_agent_init(void) +{ + uint i; + + for (i = 0; i < global.nbtgroups; i++) { + MT_LIST_INIT(&tg_agents[i].ops); + if (socketpair(AF_UNIX, SOCK_SEQPACKET, 0, tg_agents[i].xfer_sock) < 0) + return 0; + fd_set_nonblock(tg_agents[i].xfer_sock[1]); + tg_agents[i].tl = tasklet_new(); + if (!tg_agents[i].tl) + return 0; + tg_agents[i].tl->process = tg_agent_process; + tg_agents[i].tl->context = (void *)(ulong)i; + tg_agents[i].tl->tid = ha_tgroup_info[i].base; + } + protocol_init_rx_agents(); + tg_agents_enabled = 1; + return 1; +} + /* default function used to unbind a listener. This is for use by standard * protocols working on top of accepted sockets. The receiver's rx_unbind() * will automatically be used after the listener is disabled if the socket is diff --git a/src/protocol.c b/src/protocol.c index fe91656ba..1e743b428 100644 --- a/src/protocol.c +++ b/src/protocol.c @@ -10,6 +10,8 @@ * */ +#include + #include #include @@ -131,6 +133,54 @@ static inline int protocol_may_bind_quic(struct listener *l) } #endif +void protocol_init_rx_agents(void) +{ + struct protocol *proto; + struct receiver *rx; + + HA_SPIN_LOCK(PROTO_LOCK, &proto_lock); + list_for_each_entry(proto, &protocols, list) { + list_for_each_entry(rx, &proto->receivers, proto_list) { + MT_LIST_INIT(&rx->agent.link.list); + rx->agent.link.rx = rx; + rx->agent.close_fd = -1; + rx->agent.xfer_fd = -1; + } + } + HA_SPIN_UNLOCK(PROTO_LOCK, &proto_lock); +} + +/* + * With per-thread-group FD tables, gives up the calling thread's group's + * inherited copies of the receiver FDs it does not own: a receiver's FD is + * only ever used from its owner group, and keeping the other groups' copies + * alive would make every release require all groups' cooperation. + */ +void protocol_localize_rx_fds(void) +{ + struct protocol *proto; + struct receiver *rx; + + HA_SPIN_LOCK(PROTO_LOCK, &proto_lock); + list_for_each_entry(proto, &protocols, list) { + list_for_each_entry(rx, &proto->receivers, proto_list) { + int fd = rx->fd; + + if (fd < 0 || rx_owner_tgid(rx) == tgid) + continue; + + fdtab[fd].owner = NULL; + fdtab[fd].state = 0; + fdtab[fd].thread_mask = 0; + fdtab[fd].update_mask = 0; + fdtab[fd].running_mask = 0; + HA_ATOMIC_STORE(&fdtab[fd].refc_tgid, 0); + close(fd); + } + } + HA_SPIN_UNLOCK(PROTO_LOCK, &proto_lock); +} + /* binds all listeners of all registered protocols. Returns a composition * of ERR_NONE, ERR_RETRYABLE, ERR_FATAL. */ diff --git a/src/proxy.c b/src/proxy.c index 76104a083..78f2aa64e 100644 --- a/src/proxy.c +++ b/src/proxy.c @@ -3976,7 +3976,7 @@ int pause_proxy(struct proxy *p) list_for_each_entry(l, &p->conf.listeners, by_fe) suspend_listener(l, 1, 0); - if (p->li_ready) { + if (p->li_ready && !tg_agents_enabled) { ha_warning("%s %s failed to enter pause mode.\n", proxy_cap_str(p->cap), p->id); send_log(p, LOG_WARNING, "%s %s failed to enter pause mode.\n", proxy_cap_str(p->cap), p->id); HA_RWLOCK_WRUNLOCK(PROXY_LOCK, &p->lock); diff --git a/src/sock.c b/src/sock.c index 8474c1c22..0452805cc 100644 --- a/src/sock.c +++ b/src/sock.c @@ -385,8 +385,12 @@ void sock_unbind(struct receiver *rx) return; rx->flags &= ~RX_F_BOUND; - if (rx->fd != -1) - fd_delete(rx->fd); + if (rx->fd != -1) { + if (tg_agents_enabled) + rx_agent_close(rx); + else + fd_delete(rx->fd); + } rx->fd = -1; } diff --git a/src/sock_inet.c b/src/sock_inet.c index 342f3ac0f..0b065c33e 100644 --- a/src/sock_inet.c +++ b/src/sock_inet.c @@ -25,6 +25,7 @@ #include #include #include +#include #include #include #include @@ -319,15 +320,21 @@ int sock_inet_bind_receiver(struct receiver *rx, char **errmsg) * try hard not to reconfigure the socket since it's shared. */ BUG_ON(!rx->shard_info); - if (!(rx->shard_info->ref->flags & RX_F_BOUND)) { - /* it's assumed that the first one has already reported - * the error, let's not spam with another one, and do - * not set ERR_ALERT. - */ - err |= ERR_RETRYABLE; - goto bind_ret_err; + if (tg_agents_enabled && rx->agent.xfer_fd >= 0) { + rx->fd = rx->agent.xfer_fd; + rx->agent.xfer_fd = -1; + } + else { + if (!(rx->shard_info->ref->flags & RX_F_BOUND)) { + /* it's assumed that the first one has already reported + * the error, let's not spam with another one, and do + * not set ERR_ALERT. + */ + err |= ERR_RETRYABLE; + goto bind_ret_err; + } + rx->fd = rx->shard_info->ref->fd; } - rx->fd = rx->shard_info->ref->fd; } /* if no FD was assigned yet, we'll have to either find a compatible diff --git a/src/sock_unix.c b/src/sock_unix.c index 36d60fab3..76221602f 100644 --- a/src/sock_unix.c +++ b/src/sock_unix.c @@ -229,15 +229,21 @@ int sock_unix_bind_receiver(struct receiver *rx, char **errmsg) * try hard not to reconfigure the socket since it's shared. */ BUG_ON(!rx->shard_info); - if (!(rx->shard_info->ref->flags & RX_F_BOUND)) { - /* it's assumed that the first one has already reported - * the error, let's not spam with another one, and do - * not set ERR_ALERT. - */ - err |= ERR_RETRYABLE; - goto bind_ret_err; + if (tg_agents_enabled && rx->agent.xfer_fd >= 0) { + rx->fd = rx->agent.xfer_fd; + rx->agent.xfer_fd = -1; + } + else { + if (!(rx->shard_info->ref->flags & RX_F_BOUND)) { + /* it's assumed that the first one has already reported + * the error, let's not spam with another one, and do + * not set ERR_ALERT. + */ + err |= ERR_RETRYABLE; + goto bind_ret_err; + } + rx->fd = rx->shard_info->ref->fd; } - rx->fd = rx->shard_info->ref->fd; } /* if no FD was assigned yet, we'll have to either find a compatible