From 70d56a5df6e55b6609980c7690efe7a423086dde Mon Sep 17 00:00:00 2001 From: Rongsong Shen Date: Tue, 6 Oct 2026 14:42:28 +0000 Subject: [PATCH] fix(harness): xce_test fake worker defers handler notifications and re-joins after a reconnect (bug-1921) Worker-context handlers (startjob, runtasks, jobcontrol) run as tasks of the worker's own connection, so their synchronous notify RPCs waited for a reply queued behind themselves, timed out (-110) and failed the client over; the session never joined again, so the controller rejected every later worker request ("is not a worker") and job usage stayed zero. Handlers now queue the notification (new worker_notify.{c,h} FIFO) and a one-shot delay-manager task sends it; housekeeping retries anything left. A recorded worker join is repeated with the saved attributes once the connection's reconnect_gen changes. Co-Authored-By: Claude Opus 5.5 (1M context) --- src/harness/include/worker_notify.h | 68 +++++++ src/harness/meson.build | 3 + src/harness/test.c | 214 ++++++++++++++++----- src/harness/testsuite/harness_test.c | 2 + src/harness/testsuite/test_worker_notify.c | 146 ++++++++++++++ src/harness/worker_notify.c | 108 +++++++++++ 6 files changed, 494 insertions(+), 47 deletions(-) create mode 100644 src/harness/include/worker_notify.h create mode 100644 src/harness/testsuite/test_worker_notify.c create mode 100644 src/harness/worker_notify.c diff --git a/src/harness/include/worker_notify.h b/src/harness/include/worker_notify.h new file mode 100644 index 00000000..ab15a4d1 --- /dev/null +++ b/src/harness/include/worker_notify.h @@ -0,0 +1,68 @@ +#ifndef _XCE_HARNESS_WORKER_NOTIFY_H +#define _XCE_HARNESS_WORKER_NOTIFY_H +/* + * worker_notify -- deferred worker notifications for xce_test's fake worker + * (bug-1921). + * + * xce_test's worker-context request handlers (startjob, runtasks, + * jobcontrol) run as tasks of the worker's own connection. A notify RPC + * waits for its reply (notifyAcked, bug-1741), and that reply is dispatched + * as another task of the same connection -- queued behind the handler that + * waits for it -- so the call can only time out (-110) and fail the client + * over to a new connection. Handlers therefore queue the notification with + * wn_defer() and return; the worker's delay-manager thread sends the queue + * with wn_drain(). + * + * wn_need_rejoin() is the companion check for a reconnect: the controller + * knows only the joined connection as a worker, so after + * xce_ctx_reconnect() (endpoint reconnect_gen bumped) a joined worker joins + * again before its next worker request. + */ +#include +#include +#include + +typedef enum { + WN_JOB_START = 1, /* jobid, pid, pgid */ + WN_JOB_FAIL, /* jobid, code */ + WN_TASK_START, /* jobid, clientidx, taskidx, pid, pgid */ + WN_TASK_FAIL, /* jobid, clientidx, taskidx, code */ + WN_JOB_CHKPT, /* jobid, info */ +} wn_kind_t; + +typedef struct wn_item { + wn_kind_t kind; + uint64_t jobid; + int clientidx; + int taskidx; + int pid; + int pgid; + int code; + char* info; /* owned copy inside the queue */ + struct wn_item* next; +} wn_item_t; + +typedef struct { + pthread_mutex_t lock; + wn_item_t* head; + wn_item_t* tail; + int count; +} wn_queue_t; + +typedef int (*wn_send_fn)(const wn_item_t* item, void* arg); + +void wn_init(wn_queue_t* q); +/* Frees every notification still queued. */ +void wn_fini(wn_queue_t* q); +/* Queue a copy of *proto (info is duplicated). Never sends. 0 or -ENOMEM. */ +int wn_defer(wn_queue_t* q, const wn_item_t* proto); +/* Send queued notifications in order. Stops at the first send that fails, + * keeping that one and the rest queued for the next drain. Returns the + * number sent. */ +int wn_drain(wn_queue_t* q, wn_send_fn send, void* arg); +int wn_pending(wn_queue_t* q); +/* True when a worker that joined at connection generation joined_gen must + * join again because the connection is now at cur_gen. */ +bool wn_need_rejoin(bool joined, uint64_t joined_gen, uint64_t cur_gen); + +#endif diff --git a/src/harness/meson.build b/src/harness/meson.build index de8f37cf..b8fd595e 100644 --- a/src/harness/meson.build +++ b/src/harness/meson.build @@ -20,6 +20,7 @@ xce_test_src = [ 'resval_typeresolve.c', 'joblist_display.c', 'join_worker_args.c', + 'worker_notify.c', 'rcfile.c', 'flagspec.c', 'interruptcall.c', @@ -91,6 +92,8 @@ harness_unit_test_src = [ 'listfmt_json.c', 'testsuite/test_xcetest_process_entry.c', 'testsuite/test_test_cli.c', + 'testsuite/test_worker_notify.c', + 'worker_notify.c', ] harness_unit_test_deps = [ diff --git a/src/harness/test.c b/src/harness/test.c index ae54b9f0..1bfc8702 100644 --- a/src/harness/test.c +++ b/src/harness/test.c @@ -79,6 +79,7 @@ #include "resval_typeresolve.h" /* resval_resolve_type (resource-value-verification task 2.1) */ #include "xce/cluster.h" /* xce_parse_peers, xce_cluster_config_t */ #include "xce/types.h" +#include "worker_notify.h" /* deferred worker notifications (bug-1921) */ #include "xcetest_interrupt.h" /* xcetest_install_sigint, xcetest_interrupted (xce-test-cli-ux task 1.3) */ typedef int (*cmd_func_t)(int, char*[]); @@ -217,6 +218,9 @@ extern int cmd_snapdiff(int, char*[]); extern int cmd_configcompare(int, char*[]); extern int cmd_source(int, char*[]); static int worker_house_keeping(void*); +static void worker_record_join(const hostattrs_t* attrs); /* bug-1921 */ +static wn_queue_t g_worker_notify; /* bug-1921 */ +static int g_worker_notify_inited; /* bug-1921 */ /* Session verdict / loop state (xce-test-e2e-primitives task 1.1). * @@ -1914,6 +1918,11 @@ int cmd_worker(int argc, char* argv[]) active_ctx = &workerctx; + if (!g_worker_notify_inited) { + wn_init(&g_worker_notify); + g_worker_notify_inited = 1; + } + delaymgr_init(&delaymgr, 2000 /* precision_ms */, NULL); delaymgr_start(&delaymgr); delay_task(&delaymgr, 5, -1, (delay_task_entry_fn_t)worker_house_keeping, @@ -5377,6 +5386,7 @@ int cmd_join_worker(int argc, char* argv[]) return rc; } + worker_record_join(&attrs); printf("host join ok\n"); return 0; @@ -6397,73 +6407,175 @@ static int sched_on_workers(int nworkers, xce_worker_t** workers) return 0; } +/* bug-1921: worker-context handlers (startjob, runtasks, jobcontrol) run as + * tasks of the worker's own connection, so a notify RPC made there waits for + * a reply queued behind itself and times out, failing the client over to a + * connection the controller does not know as a worker. They queue the + * notification instead (worker_notify.h); the worker's delay-manager thread + * sends it, re-joining first if the connection was re-established. */ +static bool g_worker_joined; +static uint64_t g_worker_joined_gen; +static hostattrs_t g_worker_join_attrs; + +static uint64_t worker_conn_gen(command_ctx_t* ctx) +{ + endpoint_t* ep = xcecli_get_my_endpoint(ctx->ctx); + + return (ep == NULL) ? 0 + : atomic_load_explicit(&ep->reconnect_gen, + memory_order_acquire); +} + +static void worker_record_join(const hostattrs_t* attrs) +{ + if (active_ctx != &workerctx) { + return; + } + free(g_worker_join_attrs.hosttype); + free(g_worker_join_attrs.hostmodel); + g_worker_join_attrs = *attrs; + g_worker_join_attrs.hosttype = attrs->hosttype ? strdup(attrs->hosttype) + : NULL; + g_worker_join_attrs.hostmodel = attrs->hostmodel ? strdup(attrs->hostmodel) + : NULL; + g_worker_joined_gen = worker_conn_gen(&workerctx); + g_worker_joined = true; +} + +/* Re-join after xce_ctx_reconnect: the controller only treats the joined + * connection as a worker. Runs on the delay-manager thread. */ +static void worker_ensure_joined(command_ctx_t* ctx) +{ + uint64_t gen = worker_conn_gen(ctx); + int rc; + + if (!wn_need_rejoin(g_worker_joined, g_worker_joined_gen, gen)) { + return; + } + rc = xcecli_join(ctx->ctx, HostRole_worker, &g_worker_join_attrs); + if (rc < 0) { + cmdutil_rpc_error("xce_join (re-join after reconnect)", rc); + return; + } + g_worker_joined_gen = gen; + printf("host re-join ok (connection re-established)\n"); +} + +static int worker_notify_send(const wn_item_t* it, void* arg) +{ + command_ctx_t* ctx = (command_ctx_t*)arg; + + switch (it->kind) { + case WN_JOB_START: { + job_pidinfo_t pidinfo = { + .jobid = it->jobid, + .npids = 1, + .pids = (unsigned int[]) { (unsigned int)it->pid }, + .npgids = 0, + .pgids = (unsigned int[]) { (unsigned int)it->pgid }, + .exdata = NULL, + }; + return xcecli_notify_jobstart(ctx->ctx, &pidinfo); + } + case WN_JOB_FAIL: { + job_failinfo_t failinfo = { + .jobid = it->jobid, + .stage = JOB_EXEC_STAGE_MAIN, + .code = it->code, + .detail = "no detail information", + }; + return xcecli_notify_jobfail(ctx->ctx, &failinfo); + } + case WN_TASK_START: { + xce_task_pidinfo_t pidinfo = { + .jobid = it->jobid, + .host = get_setting("myname"), + .clientidx = it->clientidx, + .taskidx = it->taskidx, + .npids = 1, + .pids = (unsigned int[]) { (unsigned int)it->pid }, + .npgids = 1, + .pgids = (unsigned int[]) { (unsigned int)it->pgid }, + }; + return xcecli_notify_taskstart(ctx->ctx, &pidinfo); + } + case WN_TASK_FAIL: { + xce_task_failinfo_t failinfo = { + .jobid = it->jobid, + .host = get_setting("myname"), + .clientidx = it->clientidx, + .taskidx = it->taskidx, + .code = it->code, + .detail = NULL, + }; + return xcecli_notify_taskfail(ctx->ctx, &failinfo); + } + case WN_JOB_CHKPT: + return xcecli_notify_jobchkpt(ctx->ctx, it->jobid, it->info); + } + return 0; +} + +static void worker_notify_drain(command_ctx_t* ctx) +{ + xce_use(ctx); + worker_ensure_joined(ctx); + (void)wn_drain(&g_worker_notify, worker_notify_send, ctx); +} + +static int worker_notify_flush(void* arg) +{ + worker_notify_drain((command_ctx_t*)arg); + return 0; +} + +/* Queue a worker notification and have the delay-manager thread send it on + * its next tick; a send that fails is retried by worker_house_keeping. */ +static void worker_notify_defer(const wn_item_t* it) +{ + if (wn_defer(&g_worker_notify, it) < 0) { + fprintf(stderr, "worker notify: out of memory, notification lost\n"); + return; + } + (void)delay_task_ms(&delaymgr, 0, 1, + (delay_task_entry_fn_t)worker_notify_flush, &workerctx); +} + static void notify_job_fail(uint64_t jobid, int code) { - /* clang-format off */ - job_failinfo_t failinfo = { - .jobid = jobid, - .stage = JOB_EXEC_STAGE_MAIN, - .code = code, - .detail = "no detail information" - }; - /*clang-format on*/ - - xcecli_notify_jobfail(active_ctx->ctx, &failinfo); - return; + wn_item_t it = { .kind = WN_JOB_FAIL, .jobid = jobid, .code = code }; + + worker_notify_defer(&it); } static void notify_task_fail(uint64_t jobid, int clientidx, int taskidx, int code) { - xce_task_failinfo_t failinfo = { - .jobid = jobid, - .host = get_setting("myname"), - .clientidx = clientidx, - .taskidx = taskidx, - .code = code, - .detail = NULL - }; + wn_item_t it = { .kind = WN_TASK_FAIL, .jobid = jobid, + .clientidx = clientidx, .taskidx = taskidx, + .code = code }; - xcecli_notify_taskfail(active_ctx->ctx, &failinfo); + worker_notify_defer(&it); } static void notify_jobstart(uint64_t jobid, int pid, int pgid) { - /* clang-format off */ - job_pidinfo_t pidinfo = { - .jobid = jobid, - .npids = 1, - .pids = (unsigned int []){(unsigned int)pid}, - .npgids = 0, - .pgids = (unsigned int []){(unsigned int)pgid}, - .exdata = NULL - }; - /* clang-format on */ + wn_item_t it = { .kind = WN_JOB_START, .jobid = jobid, .pid = pid, + .pgid = pgid }; - xcecli_notify_jobstart(active_ctx->ctx, &pidinfo); - - return; + worker_notify_defer(&it); } static void notify_task_start(uint64_t jobid, int clientidx, int taskidx, int pid, int pgid) { - xce_task_pidinfo_t pidinfo - = { .jobid = jobid, - .host = get_setting("myname"), - .clientidx = clientidx, - .taskidx = taskidx, - .npids = 1, - .pids = (unsigned int[]) { (unsigned int)pid }, - .npgids = 1, - .pgids = (unsigned int[]) { (unsigned int)pgid }, - .exdata = NULL }; + wn_item_t it = { .kind = WN_TASK_START, .jobid = jobid, + .clientidx = clientidx, .taskidx = taskidx, + .pid = pid, .pgid = pgid }; - xcecli_notify_taskstart(active_ctx->ctx, &pidinfo); - - return; + worker_notify_defer(&it); } static int run_task_remote(uint64_t jobid, job_startinfo_t* startinfo, @@ -6714,7 +6826,11 @@ static int job_chkpnt(uint64_t jobid, char* method, int kill) "{\nhost_select=\"criu\"; \nrestart_command=\"cd %s; criu restore " "-vvv\"}\n", buf); - xcecli_notify_jobchkpt(active_ctx->ctx, jobid, info); + { + wn_item_t it = { .kind = WN_JOB_CHKPT, .jobid = jobid, .info = info }; + + worker_notify_defer(&it); /* bug-1921: handler context */ + } if (kill) { job_kill(jobid, SIGTERM); @@ -7002,6 +7118,10 @@ static int worker_house_keeping(void* arg) xce_use(ctx); + /* bug-1921: re-join after a reconnect, and retry notifications a + * handler queued that could not be sent yet. */ + worker_notify_drain(ctx); + check_run_jobs(); check_run_tasks(); diff --git a/src/harness/testsuite/harness_test.c b/src/harness/testsuite/harness_test.c index 7e1ff970..46208d6c 100644 --- a/src/harness/testsuite/harness_test.c +++ b/src/harness/testsuite/harness_test.c @@ -18,6 +18,7 @@ extern void test_confirm(void); extern void test_listfmt_json(void); extern void test_xcetest_process_entry(void); extern void test_test_cli(void); +extern void test_worker_notify(void); /* test_test_cli.c's leak-safety net (see its own comment): reaps and * closes any clitest session left open by a test function that aborted @@ -114,6 +115,7 @@ int main(int argc, char* argv[]) TEST(test_listfmt_json); TEST(test_xcetest_process_entry); TEST(test_test_cli); + TEST(test_worker_notify); return UNITY_END(); } diff --git a/src/harness/testsuite/test_worker_notify.c b/src/harness/testsuite/test_worker_notify.c new file mode 100644 index 00000000..fee1f6bd --- /dev/null +++ b/src/harness/testsuite/test_worker_notify.c @@ -0,0 +1,146 @@ +/* + * test_worker_notify.c -- bug-1921: xce_test's fake worker must not make a + * synchronous notify RPC from a handler running as a task of its own + * connection (the reply is queued behind the handler, so the call times out + * and the client fails over). Handlers queue worker notifications instead; + * the worker's delay-manager thread drains the queue. After a reconnect the + * worker re-joins before sending worker requests. + */ +#include +#include +#include +#include + +#include + +#include "worker_notify.h" + +#define MAX_SENT 8 +static wn_item_t g_sent[MAX_SENT]; +static char g_sent_info[MAX_SENT][32]; +static int g_nsent; +static int g_fail_at; /* 1-based send call that fails; 0 = never */ +static int g_calls; + +static int record_send(const wn_item_t* it, void* arg) +{ + (void)arg; + g_calls++; + if (g_fail_at != 0 && g_calls == g_fail_at) + return -110; + g_sent[g_nsent] = *it; + g_sent_info[g_nsent][0] = '\0'; + if (it->info != NULL) + snprintf(g_sent_info[g_nsent], sizeof(g_sent_info[0]), "%s", it->info); + g_nsent++; + return 0; +} + +static void reset_sends(int fail_at) +{ + memset(g_sent, 0, sizeof(g_sent)); + g_nsent = 0; + g_calls = 0; + g_fail_at = fail_at; +} + +static void test_defer_does_not_send(void) +{ + wn_queue_t q; + wn_init(&q); + reset_sends(0); + + wn_item_t start = { .kind = WN_JOB_START, .jobid = 7, .pid = 100, .pgid = 100 }; + TEST_ASSERT_EQUAL_INT(0, wn_defer(&q, &start)); + TEST_ASSERT_EQUAL_INT_MESSAGE(0, g_calls, + "bug-1921: queuing a notification must not send it"); + TEST_ASSERT_EQUAL_INT(1, wn_pending(&q)); + + wn_fini(&q); +} + +static void test_drain_sends_fifo_with_copied_payload(void) +{ + wn_queue_t q; + wn_init(&q); + reset_sends(0); + + char info[16]; + strcpy(info, "ckpt-a"); + wn_item_t a = { .kind = WN_JOB_START, .jobid = 1, .pid = 11, .pgid = 11 }; + wn_item_t b = { .kind = WN_TASK_FAIL, .jobid = 2, .clientidx = 3, + .taskidx = 4, .code = -5 }; + wn_item_t c = { .kind = WN_JOB_CHKPT, .jobid = 3, .info = info }; + TEST_ASSERT_EQUAL_INT(0, wn_defer(&q, &a)); + TEST_ASSERT_EQUAL_INT(0, wn_defer(&q, &b)); + TEST_ASSERT_EQUAL_INT(0, wn_defer(&q, &c)); + strcpy(info, "CLOBBERED"); /* the queue keeps its own copy */ + + TEST_ASSERT_EQUAL_INT(3, wn_drain(&q, record_send, NULL)); + TEST_ASSERT_EQUAL_INT(3, g_nsent); + TEST_ASSERT_EQUAL_INT(WN_JOB_START, g_sent[0].kind); + TEST_ASSERT_EQUAL_UINT64(1, g_sent[0].jobid); + TEST_ASSERT_EQUAL_INT(11, g_sent[0].pid); + TEST_ASSERT_EQUAL_INT(WN_TASK_FAIL, g_sent[1].kind); + TEST_ASSERT_EQUAL_INT(3, g_sent[1].clientidx); + TEST_ASSERT_EQUAL_INT(4, g_sent[1].taskidx); + TEST_ASSERT_EQUAL_INT(-5, g_sent[1].code); + TEST_ASSERT_EQUAL_INT(WN_JOB_CHKPT, g_sent[2].kind); + TEST_ASSERT_EQUAL_STRING("ckpt-a", g_sent_info[2]); + TEST_ASSERT_EQUAL_INT(0, wn_pending(&q)); + + wn_fini(&q); +} + +/* A failed send (e.g. not connected) keeps that notification and every later + * one queued, in order, for the next drain: a job start is never lost. */ +static void test_drain_stops_at_failure_and_resumes(void) +{ + wn_queue_t q; + wn_init(&q); + reset_sends(2); + + for (int i = 1; i <= 3; i++) { + wn_item_t it = { .kind = WN_JOB_START, .jobid = (uint64_t)i }; + TEST_ASSERT_EQUAL_INT(0, wn_defer(&q, &it)); + } + TEST_ASSERT_EQUAL_INT(1, wn_drain(&q, record_send, NULL)); + TEST_ASSERT_EQUAL_INT(2, wn_pending(&q)); + + reset_sends(0); + TEST_ASSERT_EQUAL_INT(2, wn_drain(&q, record_send, NULL)); + TEST_ASSERT_EQUAL_UINT64(2, g_sent[0].jobid); + TEST_ASSERT_EQUAL_UINT64(3, g_sent[1].jobid); + TEST_ASSERT_EQUAL_INT(0, wn_pending(&q)); + + wn_fini(&q); +} + +static void test_fini_frees_undrained(void) +{ + wn_queue_t q; + wn_init(&q); + wn_item_t c = { .kind = WN_JOB_CHKPT, .jobid = 9, .info = "x" }; + TEST_ASSERT_EQUAL_INT(0, wn_defer(&q, &c)); + wn_fini(&q); /* LeakSanitizer checks the copy is released */ + TEST_ASSERT_EQUAL_INT(0, wn_pending(&q)); +} + +/* A worker that joined must join again once its connection was re-established + * (the controller only knows the joined connection as a worker). */ +static void test_need_rejoin(void) +{ + TEST_ASSERT_FALSE(wn_need_rejoin(false, 0, 0)); + TEST_ASSERT_FALSE(wn_need_rejoin(false, 0, 3)); + TEST_ASSERT_FALSE(wn_need_rejoin(true, 2, 2)); + TEST_ASSERT_TRUE(wn_need_rejoin(true, 2, 3)); +} + +void test_worker_notify(void) +{ + RUN_TEST(test_defer_does_not_send); + RUN_TEST(test_drain_sends_fifo_with_copied_payload); + RUN_TEST(test_drain_stops_at_failure_and_resumes); + RUN_TEST(test_fini_frees_undrained); + RUN_TEST(test_need_rejoin); +} diff --git a/src/harness/worker_notify.c b/src/harness/worker_notify.c new file mode 100644 index 00000000..841cbe4b --- /dev/null +++ b/src/harness/worker_notify.c @@ -0,0 +1,108 @@ +/* + * worker_notify.c -- deferred worker notifications for xce_test's fake + * worker (bug-1921). See include/worker_notify.h. + */ +#include +#include +#include + +#include "worker_notify.h" + +void wn_init(wn_queue_t* q) +{ + pthread_mutex_init(&q->lock, NULL); + q->head = NULL; + q->tail = NULL; + q->count = 0; +} + +static void wn_free_item(wn_item_t* it) +{ + free(it->info); + free(it); +} + +void wn_fini(wn_queue_t* q) +{ + pthread_mutex_lock(&q->lock); + wn_item_t* it = q->head; + while (it != NULL) { + wn_item_t* next = it->next; + wn_free_item(it); + it = next; + } + q->head = NULL; + q->tail = NULL; + q->count = 0; + pthread_mutex_unlock(&q->lock); + pthread_mutex_destroy(&q->lock); +} + +int wn_defer(wn_queue_t* q, const wn_item_t* proto) +{ + wn_item_t* it = malloc(sizeof(*it)); + if (it == NULL) + return -ENOMEM; + *it = *proto; + it->next = NULL; + it->info = NULL; + if (proto->info != NULL) { + it->info = strdup(proto->info); + if (it->info == NULL) { + free(it); + return -ENOMEM; + } + } + + pthread_mutex_lock(&q->lock); + if (q->tail != NULL) + q->tail->next = it; + else + q->head = it; + q->tail = it; + q->count++; + pthread_mutex_unlock(&q->lock); + return 0; +} + +int wn_drain(wn_queue_t* q, wn_send_fn send, void* arg) +{ + int sent = 0; + + for (;;) { + /* Peek under the lock, send without it (a send blocks on the RPC), + * then pop: only the drain thread removes items, so the head is + * still this item; wn_defer only ever appends. */ + pthread_mutex_lock(&q->lock); + wn_item_t* it = q->head; + pthread_mutex_unlock(&q->lock); + if (it == NULL) + break; + + if (send(it, arg) < 0) + break; /* keep it (and the rest) for the next drain */ + + pthread_mutex_lock(&q->lock); + q->head = it->next; + if (q->head == NULL) + q->tail = NULL; + q->count--; + pthread_mutex_unlock(&q->lock); + wn_free_item(it); + sent++; + } + return sent; +} + +int wn_pending(wn_queue_t* q) +{ + pthread_mutex_lock(&q->lock); + int n = q->count; + pthread_mutex_unlock(&q->lock); + return n; +} + +bool wn_need_rejoin(bool joined, uint64_t joined_gen, uint64_t cur_gen) +{ + return joined && cur_gen != joined_gen; +}