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 34551cff9c713196d7e690b111067ea40bf0dfcf..f4b02526cebf376c50a2bbb9af62aabf14dfc04c 100644 (file)
--- 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 7018b08b0653265dae0cd82acb8de2fd4ae9792a..25d6cd34f20c99b71abde926f70101c980648297 100644 (file)
--- 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 <state> on <rx>, 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 c4c8fa8003dc560d6a90a47c947de7bc10e68a18..0ea381abab78c3d4129bd0ff125db182da7df7a0 100644 (file)
--- 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 00b3b0c19498f9f3575bb14c6f13d83d9ed2166a..3d88342223210a2a29d8397139fd7af79eed52c7 100644 (file)
--- 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 */