From 431a6d709e26f0ff929b455cac6fa8fafb7f5d4c Mon Sep 17 00:00:00 2001 From: Rongsong Shen Date: Sun, 4 Oct 2026 23:59:29 +0000 Subject: [PATCH] fix(lib): quiesce the old endpoint before xce_ctx_reconnect re-inits it (bug-1917) Work dispatched for the old connection (the io_worker's epoll batch, queued VTASK_IO reply tasks) could still run while the reconnect tore down and re-armed ctx->ep; a queued reply of the old connection then reached the new session. The reconnect now unmonitors the old fd, clears ep->owner (stale tasks are dropped) and waits for the endpoint's references to drain, bounded at 2 s so a caller that is itself one of those tasks cannot deadlock. vtask_entry reads ep->owner under ep->lock. Co-Authored-By: Claude Opus 5.5 (1M context) --- src/lib/meson.build | 1 + src/lib/testsuite/test.c | 2 + .../testsuite/test_ctx_reconnect_quiesce.c | 132 ++++++++++++++++++ src/lib/xcectx.c | 51 +++++++ 4 files changed, 186 insertions(+) create mode 100644 src/lib/testsuite/test_ctx_reconnect_quiesce.c diff --git a/src/lib/meson.build b/src/lib/meson.build index 4cf7d8cc..9a2ac048 100644 --- a/src/lib/meson.build +++ b/src/lib/meson.build @@ -230,6 +230,7 @@ libxce_test_src = ['testsuite/test.c', 'testsuite/test_vtask_admission.c', 'testsuite/test_vtask_enq_race.c', 'testsuite/test_vtask_pkt_handoff.c', + 'testsuite/test_ctx_reconnect_quiesce.c', 'testsuite/test_vtask_order.c', 'testsuite/test_vtaskmgr_teardown.c', 'testsuite/test_xcealloc.c', diff --git a/src/lib/testsuite/test.c b/src/lib/testsuite/test.c index c4fcbe3a..53f7e7f0 100644 --- a/src/lib/testsuite/test.c +++ b/src/lib/testsuite/test.c @@ -59,6 +59,7 @@ extern void test_ctrlstatus(void); extern void test_vtask_admission(void); extern void test_vtask_enq_race(void); extern void test_vtask_pkt_handoff(void); +extern void test_ctx_reconnect_quiesce(void); extern void test_vtask_order(void); extern void test_vtaskmgr_teardown(void); extern void test_xcealloc(void); @@ -190,6 +191,7 @@ int main(int argc, char* argv[]) TEST(test_vtask_admission); TEST(test_vtask_enq_race); TEST(test_vtask_pkt_handoff); + TEST(test_ctx_reconnect_quiesce); TEST(test_vtask_order); TEST(test_vtaskmgr_teardown); TEST(test_xcealloc); diff --git a/src/lib/testsuite/test_ctx_reconnect_quiesce.c b/src/lib/testsuite/test_ctx_reconnect_quiesce.c new file mode 100644 index 00000000..a6edfb62 --- /dev/null +++ b/src/lib/testsuite/test_ctx_reconnect_quiesce.c @@ -0,0 +1,132 @@ +/* + * test_ctx_reconnect_quiesce.c — bug-1917 regression: xce_ctx_reconnect must + * quiesce the old endpoint before re-initialising it. + * + * xce_ctx_reconnect re-inits ctx->ep (an endpoint embedded in the client + * context) on the caller's thread. Work already dispatched for the OLD + * connection — the io_worker's in-flight batch, and VTASK_IO reply tasks still + * queued on the context's vtask pool — holds endpoint references and reads + * ep->owner. Before the fix nothing waited for that work: a reply task of the + * old connection that ran after the reconnect found ep->owner re-armed to the + * context and was delivered to the NEW session (and the io_worker could still + * be inside SSL_read / handle_output on the fd and SSL being freed). + * + * The test queues an inbound packet for the old connection exactly as + * handle_input does (INVOKE_MSG_HANDLER -> handle_msg -> vtask_handle_msg) + * while the vtask pool is stopped, starts the pool 100 ms into a second + * reconnect, and asserts the stale task never reaches the context's request + * handler. + */ +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include + +#include "netio.h" +#include "request.h" +#include "vtaskmgr.h" +#include "xcectx.h" + +static atomic_int g_handler_calls; + +static int counting_handler(xce_request_t* req, void* args) +{ + (void)req; + (void)args; + atomic_fetch_add(&g_handler_calls, 1); + return 0; +} + +static int listen_local(int* port_out) +{ + struct sockaddr_in sa; + socklen_t len = sizeof(sa); + int fd = socket(AF_INET, SOCK_STREAM, 0); + + TEST_ASSERT_TRUE(fd >= 0); + memset(&sa, 0, sizeof(sa)); + sa.sin_family = AF_INET; + sa.sin_addr.s_addr = htonl(INADDR_LOOPBACK); + sa.sin_port = 0; + TEST_ASSERT_EQUAL_INT(0, bind(fd, (struct sockaddr*)&sa, sizeof(sa))); + TEST_ASSERT_EQUAL_INT(0, listen(fd, 8)); + TEST_ASSERT_EQUAL_INT(0, getsockname(fd, (struct sockaddr*)&sa, &len)); + *port_out = ntohs(sa.sin_port); + return fd; +} + +static io_packet_t* encode_nop(void) +{ + xce_request_t request; + io_packet_t* pkt = (io_packet_t*)calloc(1, sizeof(*pkt)); + + TEST_ASSERT_NOT_NULL(pkt); + memset(&request, 0, sizeof(request)); + request.opcode = OPCode_nop; + TEST_ASSERT_EQUAL_INT(1, xce_encode(&request, pkt)); + return pkt; +} + +static void* start_pool_later(void* arg) +{ + usleep(100 * 1000); + vtaskmgr_start((vtaskmgr_t*)arg); + return NULL; +} + +static void test_reconnect_drains_old_connection_tasks(void) +{ + xce_ctx_t ctx; + endpoint_t ep; + pthread_t th; + int port; + int lfd = listen_local(&port); + + atomic_store(&g_handler_calls, 0); + TEST_ASSERT_EQUAL_INT(0, xce_ctx_init(&ctx, "t1917", "key")); + memset(&ep, 0, sizeof(ep)); + TEST_ASSERT_TRUE(endpoint_init(&ep, AF_INET, SOCK_STREAM, 0, 0) >= 0); + TEST_ASSERT_EQUAL_INT(0, xce_ctx_set_endpoint(&ctx, &ep)); + TEST_ASSERT_EQUAL_INT(0, xce_ctx_set_handler(&ctx, counting_handler, NULL)); + + /* First connection: arms owner + handle_msg on the endpoint. */ + TEST_ASSERT_EQUAL_INT(0, xce_ctx_reconnect(&ctx, "127.0.0.1", port)); + + /* A message of the OLD connection is dispatched (io_worker path) while + * the pool is busy (here: not started), so its task is still queued. */ + ep.fsm_in.state = FSM_STATE_DONE; + ep.fsm_in.pkt = encode_nop(); + INVOKE_MSG_HANDLER(&ep); + TEST_ASSERT_NULL(ep.fsm_in.pkt); + TEST_ASSERT_EQUAL_INT(1, atomic_load(&ep.refcnt)); + + /* Failover reconnect; the pool gets to run 100 ms into it. */ + TEST_ASSERT_EQUAL_INT(0, pthread_create(&th, NULL, start_pool_later, + &ctx.vtaskmgr)); + TEST_ASSERT_EQUAL_INT(0, xce_ctx_reconnect(&ctx, "127.0.0.1", port)); + pthread_join(th, NULL); + + usleep(300 * 1000); /* let a still-queued stale task run (pre-fix) */ + TEST_ASSERT_EQUAL_INT_MESSAGE( + 0, atomic_load(&g_handler_calls), + "bug-1917: a task of the old connection must not be delivered to the " + "reconnected session"); + TEST_ASSERT_EQUAL_INT(0, atomic_load(&ep.refcnt)); + TEST_ASSERT_EQUAL_PTR(&ctx, ep.owner); + + xce_ctx_fini(&ctx); + endpoint_fini(&ep); + close(lfd); +} + +void test_ctx_reconnect_quiesce(void) +{ + RUN_TEST(test_reconnect_drains_old_connection_tasks); +} diff --git a/src/lib/xcectx.c b/src/lib/xcectx.c index 2dc4d49c..eb3872f1 100644 --- a/src/lib/xcectx.c +++ b/src/lib/xcectx.c @@ -1,6 +1,8 @@ #include +#include #include #include +#include #include "config.h" /* DEFAULT_CONNECT_TIMEOUT_MS */ #include "dataq.h" @@ -132,7 +134,11 @@ static int vtask_entry(vtask_t* task) * unowned ep is reachable: failover's xce_ctx_reconnect -> endpoint_fini * clears ep->owner while a reply's task is still queued. */ if (ep != NULL) { + /* Under ep->lock: xce_ctx_reconnect clears/re-arms the owner while a + * task of the old connection may still run (bug-1917). */ + mylock_lock(&(ep->lock)); ctx = (xce_ctx_t*)ep->owner; + mylock_unlock(&(ep->lock)); } if (ctx == NULL) { if (data->pkt != NULL) { @@ -360,6 +366,47 @@ int xce_ctx_start(xce_ctx_t* ctx) * (crash at netio.c endpoint_connect); without re-arming the handlers + monitor * the reconnected endpoint would never route replies to ctx->wbuf. */ +/* Upper bound on waiting for the old connection's in-flight work. The wait + * normally ends within one vtask run; the bound only matters if the caller is + * itself a task of this endpoint (its own reference never drains), where the + * reconnect proceeds as it did before bug-1917 rather than deadlocking. */ +#define XCE_RECONNECT_QUIESCE_MS 2000 + +/* xce_ctx_quiesce_ep (bug-1917): stop new work on ctx->ep and wait out the + * work already dispatched for the old connection before the caller + * re-initialises the endpoint. Unmonitor the old fd so the io_worker gets no + * further events for it; clear the owner so reply tasks of the old connection + * still queued on the vtask pool are dropped (vtask_entry frees their packet) + * instead of reaching the reconnected session; then wait until the endpoint's + * reference count is back at its base value -- the io_worker holds a reference + * for the whole epoll batch it is dispatching, and every queued task holds + * one -- so nothing still touches the old fd, SSL or packets when + * endpoint_fini runs. */ +static void xce_ctx_quiesce_ep(xce_ctx_t* ctx) +{ + endpoint_t* ep = ctx->ep; + int base = ep->heap_alloc ? 1 : 0; + int waited; + + if (ep->fd >= 0) { + (void)iomgr_unmonitor(&(ctx->iomgr), ep); + } + + mylock_lock(&(ep->lock)); + ep->owner = NULL; + mylock_unlock(&(ep->lock)); + + for (waited = 0; waited < XCE_RECONNECT_QUIESCE_MS; waited++) { + if (atomic_load_explicit(&ep->refcnt, memory_order_acquire) <= base) { + return; + } + usleep(1000); + } + WARN("xce_ctx_reconnect: old connection still has %d in-flight " + "reference(s) after %d ms; reconnecting anyway", + atomic_load(&ep->refcnt) - base, XCE_RECONNECT_QUIESCE_MS); +} + int xce_ctx_reconnect(xce_ctx_t* ctx, char* host, int port) { int rc; @@ -368,6 +415,10 @@ int xce_ctx_reconnect(xce_ctx_t* ctx, char* host, int port) return -EINVAL; } + /* bug-1917: nothing may still be dispatching the old connection when the + * endpoint struct is torn down and re-initialised below. */ + xce_ctx_quiesce_ep(ctx); + /* Tear down the dead connection (closes fd, frees SSL, clears handlers and * resets the endpoint struct — including ep->iomgr). */ endpoint_fini(ctx->ep);