From: Olivier Houchard Date: Tue, 1 Sep 2026 16:45:53 +0000 (+0200) Subject: BUG/MEDIUM: listener: Don't drop suspend/resume deferred requests X-Git-Tag: v3.5-dev7~43 X-Git-Url: https://git.kaiwu.me/?a=commitdiff_plain;h=bc9524eb84665463db972c54232f03f425eba049;p=haproxy.git BUG/MEDIUM: listener: Don't drop suspend/resume deferred requests When using per-thread group file descriptors, we have to send messages to other thread groups to tell them to suspend or resume listeners. Add a current status for each thread group, instead of looking for the global listener state, to know the current state for each thread group, otherwise a rapid succession of suspend/resume could lead to an inconsistant state. --- diff --git a/include/haproxy/receiver-t.h b/include/haproxy/receiver-t.h index 34551cff9..f4b02526c 100644 --- a/include/haproxy/receiver-t.h +++ b/include/haproxy/receiver-t.h @@ -68,10 +68,13 @@ 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() */ +/* one-shot operations pending on a receiver, run by its owner group's agent */ +#define RX_AGENT_OP_ENABLE 0x00000001 /* enable_listener() */ + +/* last listening state requested on a receiver, applied by its owner group */ +#define RX_AGENT_ST_NONE 0 /* nothing requested */ +#define RX_AGENT_ST_PAUSED 1 /* suspend_listener() */ +#define RX_AGENT_ST_READY 2 /* resume_listener() (includes rebind) */ struct rx_agent_link { struct mt_list list; /* position in one group agent's queue */ @@ -92,7 +95,8 @@ struct receiver { 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_* */ + uint ops; /* pending one-shot RX_AGENT_OP_* */ + uint want_state; /* last requested RX_AGENT_ST_* */ 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 */ diff --git a/src/listener.c b/src/listener.c index 7018b08b0..25d6cd34f 100644 --- a/src/listener.c +++ b/src/listener.c @@ -59,6 +59,8 @@ ullong maxconn_reached = 0; __decl_thread(static HA_RWLOCK_T global_listener_rwlock); static void rx_agent_schedule_op(struct receiver *rx, uint op); +static void rx_agent_want_state(struct receiver *rx, uint state); +static void rx_agent_process_one(struct receiver *rx); /* listener status for stats */ const char* li_status_st[LI_STATE_COUNT] = { @@ -544,14 +546,17 @@ int suspend_listener(struct listener *l, int lpx, int lli) if (!lli) HA_RWLOCK_WRLOCK(LISTENER_LOCK, &l->lock); - if (!(l->flags & LI_F_FINALIZED) || l->state <= LI_PAUSED) + if (!(l->flags & LI_F_FINALIZED)) goto end; if (tg_agents_enabled && rx_owner_tgid(&l->rx) != tgid) { - rx_agent_schedule_op(&l->rx, RX_AGENT_OP_SUSPEND); + rx_agent_want_state(&l->rx, RX_AGENT_ST_PAUSED); goto end; } + if (l->state <= LI_PAUSED) + goto end; + if (l->rx.proto->suspend) { ret = l->rx.proto->suspend(l); /* if the suspend() fails, we don't want to change the @@ -625,24 +630,25 @@ int resume_listener(struct listener *l, int lpx, int lli) if (MT_LIST_INLIST(&l->wait_queue)) goto end; - if (!(l->flags & LI_F_FINALIZED) || l->state == LI_READY) + if (!(l->flags & LI_F_FINALIZED)) + goto end; + + if (tg_agents_enabled && rx_owner_tgid(&l->rx) != tgid && + !((l->rx.flags & RX_F_MUST_DUP) && l->state == LI_ASSIGNED)) { + rx_agent_want_state(&l->rx, RX_AGENT_ST_READY); + goto end; + } + + if (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); + rx_agent_want_state(l->rx.shard_info->ref, RX_AGENT_ST_READY); 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); @@ -818,18 +824,20 @@ static void rx_agent_process_one(struct receiver *rx) int fd = HA_ATOMIC_XCHG(&rx->agent.close_fd, -1); uint dest; uint ops; + uint state; if (fd >= 0) fd_delete(fd); ops = HA_ATOMIC_XCHG(&rx->agent.ops, 0); - if (ops) { + state = HA_ATOMIC_XCHG(&rx->agent.want_state, RX_AGENT_ST_NONE); + if (ops || state) { struct listener *l = LIST_ELEM(rx, struct listener *, rx); - if (ops & RX_AGENT_OP_SUSPEND) + if (state == RX_AGENT_ST_PAUSED) suspend_listener(l, 0, 0); - if (ops & RX_AGENT_OP_RESUME) { + if (state == RX_AGENT_ST_READY) { resume_listener(l, 0, 0); /* @@ -878,6 +886,13 @@ static void rx_agent_schedule_op(struct receiver *rx, uint op) rx_agent_post(rx, rx_owner_tgid(rx)); } +/* requests listening state on , superseding any pending one */ +static void rx_agent_want_state(struct receiver *rx, uint state) +{ + HA_ATOMIC_STORE(&rx->agent.want_state, state); + 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); diff --git a/src/protocol.c b/src/protocol.c index c4c8fa800..0ea381aba 100644 --- a/src/protocol.c +++ b/src/protocol.c @@ -143,6 +143,7 @@ void protocol_init_rx_agents(void) list_for_each_entry(rx, &proto->receivers, proto_list) { MT_LIST_INIT(&rx->agent.link.list); rx->agent.link.rx = rx; + rx->agent.want_state = RX_AGENT_ST_NONE; rx->agent.close_fd = -1; rx->agent.xfer_fd = -1; rx->agent.getsocks_fd = -1; diff --git a/src/proxy.c b/src/proxy.c index 00b3b0c19..3d8834222 100644 --- a/src/proxy.c +++ b/src/proxy.c @@ -3997,7 +3997,8 @@ int pause_proxy(struct proxy *p) HA_RWLOCK_WRLOCK(PROXY_LOCK, &p->lock); - if (!(p->cap & PR_CAP_FE) || (p->flags & (PR_FL_DISABLED|PR_FL_STOPPED)) || !p->li_ready) + if (!(p->cap & PR_CAP_FE) || (p->flags & (PR_FL_DISABLED|PR_FL_STOPPED)) || + (!p->li_ready && !tg_agents_enabled)) goto end; list_for_each_entry(l, &p->conf.listeners, by_fe) @@ -4054,7 +4055,8 @@ int resume_proxy(struct proxy *p) HA_RWLOCK_WRLOCK(PROXY_LOCK, &p->lock); - if ((p->flags & (PR_FL_DISABLED|PR_FL_STOPPED)) || !p->li_paused) + if ((p->flags & (PR_FL_DISABLED|PR_FL_STOPPED)) || + (!p->li_paused && !tg_agents_enabled)) goto end; fail = 0; @@ -5280,7 +5282,7 @@ static int cli_parse_disable_frontend(char **args, char *payload, struct appctx if (px->flags & (PR_FL_DISABLED|PR_FL_STOPPED)) return cli_msg(appctx, LOG_NOTICE, "Frontend was previously shut down, cannot disable.\n"); - if (!px->li_ready) + if (!px->li_ready && !tg_agents_enabled) return cli_msg(appctx, LOG_NOTICE, "All sockets are already disabled.\n"); /* pause_proxy will take PROXY_LOCK */ @@ -5311,7 +5313,7 @@ static int cli_parse_enable_frontend(char **args, char *payload, struct appctx * if (px->flags & (PR_FL_DISABLED|PR_FL_STOPPED)) return cli_err(appctx, "Frontend was previously shut down, cannot enable.\n"); - if (px->li_ready == px->li_all) + if (px->li_ready == px->li_all && !tg_agents_enabled) return cli_msg(appctx, LOG_NOTICE, "All sockets are already enabled.\n"); /* resume_proxy will take PROXY_LOCK */