diff --git a/src/include/netio.h b/src/include/netio.h index fc41db7e..a5490a02 100644 --- a/src/include/netio.h +++ b/src/include/netio.h @@ -170,6 +170,10 @@ typedef struct endpoint_t { typedef struct { endpoint_t* ep; io_packet_t* pkt; + /* ep->reconnect_gen when the message was received (bug-1917): a task of + * a connection that has since been re-established is stale and is + * dropped instead of being delivered to the new session. */ + uint64_t gen; } io_task_data_t; extern int alloc_packet(io_packet_t*); diff --git a/src/include/vtaskmgr.h b/src/include/vtaskmgr.h index 8c081764..d24c7922 100644 --- a/src/include/vtaskmgr.h +++ b/src/include/vtaskmgr.h @@ -109,6 +109,9 @@ extern int vtask_enq(vtaskmgr_t*, vtask_t*); extern vtask_t* vtask_deq_eligible(dataq_t* taskq, int (*is_blocked)(endpoint_t*)); extern int vtask_handle_msg(vtaskmgr_t*, vtask_func_t, void* arg); +/* bug-1917: the endpoint whose VTASK_IO task the calling thread is running, + * or NULL when it is not running one. */ +extern endpoint_t* vtask_current_io_ep(void); /* LivenessAdmission (raft-drive-async task 1.6) — pool-level reservation. * diff --git a/src/lib/testsuite/test_ctx_reconnect_quiesce.c b/src/lib/testsuite/test_ctx_reconnect_quiesce.c index a6edfb62..a4f91a4b 100644 --- a/src/lib/testsuite/test_ctx_reconnect_quiesce.c +++ b/src/lib/testsuite/test_ctx_reconnect_quiesce.c @@ -25,6 +25,7 @@ #include #include #include +#include #include #include @@ -126,7 +127,88 @@ static void test_reconnect_drains_old_connection_tasks(void) close(lfd); } +/* bug-1917 follow-up: a reconnect made from inside a task of the endpoint it + * re-establishes (a request handler that failed over) cannot drain: its own + * task holds a reference, and the old connection's queued tasks are + * serialised behind it. It must not wait out the 2 s bound for nothing, and + * those queued tasks -- which run only after it returns, when the owner is + * re-armed -- must still never reach the new session. */ +static xce_ctx_t* g_own_ctx; +static int g_own_port; +static atomic_int g_own_rc; +static atomic_int g_own_ms; +static atomic_int g_own_done; + +static int reconnect_from_own_task(vtask_t* task) +{ + io_task_data_t* data = &(task->data->u.iodata); + struct timespec t0, t1; + + free_packet(data->pkt); + free(data->pkt); + data->pkt = NULL; + + /* a stale message of the old connection arrives meanwhile: it is queued + * behind this task (same endpoint) */ + endpoint_t* ep = g_own_ctx->ep; + ep->fsm_in.state = FSM_STATE_DONE; + ep->fsm_in.pkt = encode_nop(); + INVOKE_MSG_HANDLER(ep); + + clock_gettime(CLOCK_MONOTONIC, &t0); + atomic_store(&g_own_rc, xce_ctx_reconnect(g_own_ctx, "127.0.0.1", g_own_port)); + clock_gettime(CLOCK_MONOTONIC, &t1); + atomic_store(&g_own_ms, (int)((t1.tv_sec - t0.tv_sec) * 1000 + + (t1.tv_nsec - t0.tv_nsec) / 1000000)); + atomic_store(&g_own_done, 1); + return 0; +} + +static void test_reconnect_from_own_task_does_not_wait_or_leak(void) +{ + xce_ctx_t ctx; + endpoint_t ep; + int lfd = listen_local(&g_own_port); + + atomic_store(&g_handler_calls, 0); + atomic_store(&g_own_done, 0); + TEST_ASSERT_EQUAL_INT(0, xce_ctx_init(&ctx, "t1917b", "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)); + TEST_ASSERT_EQUAL_INT(0, xce_ctx_reconnect(&ctx, "127.0.0.1", g_own_port)); + g_own_ctx = &ctx; + TEST_ASSERT_EQUAL_INT(0, vtaskmgr_start(&ctx.vtaskmgr)); + + /* a request handler of this connection that fails over */ + ep.fsm_in.state = FSM_STATE_DONE; + ep.fsm_in.pkt = encode_nop(); + TEST_ASSERT_TRUE(vtask_handle_msg(&ctx.vtaskmgr, + (vtask_func_t)reconnect_from_own_task, + &ep) >= 0); + + for (int i = 0; i < 50000 && !atomic_load(&g_own_done); i++) { + usleep(100); + } + TEST_ASSERT_TRUE(atomic_load(&g_own_done)); + usleep(300 * 1000); /* let the queued stale task run */ + + TEST_ASSERT_EQUAL_INT(0, atomic_load(&g_own_rc)); + TEST_ASSERT_TRUE_MESSAGE(atomic_load(&g_own_ms) < 500, + "bug-1917: a reconnect from the endpoint's own task must not wait out " + "the quiesce bound"); + TEST_ASSERT_EQUAL_INT_MESSAGE(0, atomic_load(&g_handler_calls), + "bug-1917: an old-connection task queued behind the reconnect must " + "not reach the reconnected session"); + + xce_ctx_fini(&ctx); + endpoint_fini(&ep); + close(lfd); +} + void test_ctx_reconnect_quiesce(void) { RUN_TEST(test_reconnect_drains_old_connection_tasks); + RUN_TEST(test_reconnect_from_own_task_does_not_wait_or_leak); } diff --git a/src/lib/vtaskmgr.c b/src/lib/vtaskmgr.c index 842a700c..30bea2f4 100644 --- a/src/lib/vtaskmgr.c +++ b/src/lib/vtaskmgr.c @@ -73,13 +73,27 @@ void vcpu_idle(vcpu_t* vcpu) return; } +/* bug-1917: the endpoint of the VTASK_IO task this thread is running. */ +static __thread endpoint_t* t_current_io_ep; + +endpoint_t* vtask_current_io_ep(void) +{ + return t_current_io_ep; +} + void run_task(vtask_t* task) { vtaskmgr_t* vtaskmgr; vtask_data_t* data; if (task->entry != NULL) { + endpoint_t* prev = t_current_io_ep; + + if (task->data != NULL && task->data->vtype == VTASK_IO) { + t_current_io_ep = task->data->u.iodata.ep; + } (task->entry)(task); + t_current_io_ep = prev; } data = task->data; @@ -647,6 +661,8 @@ int vtask_handle_msg(vtaskmgr_t* mgr, vtask_func_t entry, void* arg) data->vtype = VTASK_IO; data->u.iodata.ep = ep; data->u.iodata.pkt = ep->fsm_in.pkt; + data->u.iodata.gen = atomic_load_explicit(&ep->reconnect_gen, + memory_order_acquire); ep->fsm_in.pkt = NULL; vtask_init(task, entry, data); diff --git a/src/lib/xcectx.c b/src/lib/xcectx.c index eb3872f1..707d1b41 100644 --- a/src/lib/xcectx.c +++ b/src/lib/xcectx.c @@ -139,6 +139,14 @@ static int vtask_entry(vtask_t* task) mylock_lock(&(ep->lock)); ctx = (xce_ctx_t*)ep->owner; mylock_unlock(&(ep->lock)); + /* bug-1917: a message of a connection that has since been + * re-established (the task was queued behind a reconnect made from + * this endpoint's own task, so the reconnect could not drain it) must + * not reach the new session. */ + if (data->gen != atomic_load_explicit(&ep->reconnect_gen, + memory_order_acquire)) { + ctx = NULL; + } } if (ctx == NULL) { if (data->pkt != NULL) { @@ -396,6 +404,15 @@ static void xce_ctx_quiesce_ep(xce_ctx_t* ctx) ep->owner = NULL; mylock_unlock(&(ep->lock)); + /* Called from a task of this very endpoint (a request handler that + * failed over): its own reference cannot drain until it returns, and the + * old connection's queued tasks are serialised behind it, so waiting is + * pointless. Those queued tasks are dropped by their stale + * reconnect_gen in vtask_entry instead. */ + if (vtask_current_io_ep() == ep) { + return; + } + for (waited = 0; waited < XCE_RECONNECT_QUIESCE_MS; waited++) { if (atomic_load_explicit(&ep->refcnt, memory_order_acquire) <= base) { return;