lei.chen/myxce
1
0
Fork 1
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

fix(harness): xce_test fake worker defers handler notifications and re-joins after a reconnect (bug-1921)

Browse source 
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) <noreply@anthropic.com>
This commit is contained in:
Rongsong Shen 2026-10-06 14:42:28 +00:00
parent 12f5eb2a76
commit 70d56a5df6
Signed by untrusted user: rshen
GPG key ID: 80533536B0EB651C
6 changed files with 494 additions and 47 deletions
Download patch file Download diff file Expand all files Collapse all files

68
src/harness/include/worker_notify.h Normal file
View file

@ -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 <pthread.h>
#include <stdbool.h>
#include <stdint.h>
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

3
src/harness/meson.build
View file

@ -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 = [

214
src/harness/test.c
View file

@ -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();

2
src/harness/testsuite/harness_test.c
View file

@ -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();
}

146
src/harness/testsuite/test_worker_notify.c Normal file
View file

@ -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 <stdbool.h>
#include <stdint.h>
#include <stdlib.h>
#include <string.h>
#include <unity/unity.h>
#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);
}

108
src/harness/worker_notify.c Normal file
View file

@ -0,0 +1,108 @@
/*
* worker_notify.c -- deferred worker notifications for xce_test's fake
* worker (bug-1921). See include/worker_notify.h.
*/
#include <errno.h>
#include <stdlib.h>
#include <string.h>
#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;
}