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

fix(worker): keep a job-control signal that reaches a remote host before its runtasks (bug-1955)

Browse source 
The controller sends job control to every host of a multi-host job
(bug-1942) as soon as the job is dispatched, but a remote host's runtasks
request is relayed by the controller (task.c do_runtasks, stateless) only
after the head worker has started and asked for it. A kill/stop that
reached a remote host first found no worker and no pending start, was
dropped ("Failed to find worker for job ID"), and the task started later
and kept running.

pending_signal now also holds "early" entries: handle_jobcontrol() records
a signal for a job with no task here (pending_signal_record_early), and
handle_runtasks() adopts it (pending_signal_start_task), so it is delivered
at RUNNING through the bug-1908 path and the task's exit is reported as
usual. Early entries are bounded (256) and expire after 60 s, pruned on
every early record and task start. A late control for a task that already
finished here (finished-job retention) is still dropped, and a head-host
start does not adopt an early entry, so neither can hit a later run of the
same jobid.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
This commit is contained in:
Rongsong Shen 2026-10-10 11:37:23 +00:00
parent 8faf59618e
commit 81e7187ddc
Signed by untrusted user: rshen
GPG key ID: 80533536B0EB651C
6 changed files with 419 additions and 19 deletions
Download patch file Download diff file Expand all files Collapse all files

43
src/worker/include/pending_signal.h
View file

@ -37,7 +37,48 @@ bool pending_signal_take(uint64_t jobid, int* signo, bool* force);
bool pending_signal_take_ex(uint64_t jobid, int* signo, bool* force,
int* grace);
/* Number of pending starts (for tests and diagnostics). */
/*
* bug-1955: signals that arrive before this worker has any start for the
* job.
*
* The controller sends job control to every host of a multi-host job
* (bug-1942) as soon as the job is dispatched, but a remote host's start --
* its runtasks request -- is relayed by the controller only after the head
* worker has started, so the control can reach a remote host first. Such a
* signal is kept as an "early" entry; the task's start
* (pending_signal_start_task) adopts it, and it is delivered when the task
* reaches RUNNING like any other kept signal. A kill thus still stops the
* task, and its exit is reported through the normal task-finish path.
*
* Early entries are bounded (PENDING_SIGNAL_EARLY_MAX) and expire
* PENDING_SIGNAL_EARLY_TTL_MS after they were recorded, so controls for jobs
* that never start here do not accumulate and cannot hit a much later start
* of the same jobid. Expired entries are pruned by every early record and
* every task start. Times are caller-supplied monotonic milliseconds
* (pending_signal_now_ms() in production, fixed values in tests).
*/
#define PENDING_SIGNAL_EARLY_TTL_MS 60000ULL
#define PENDING_SIGNAL_EARLY_MAX 256
/* Monotonic milliseconds, for the now_ms arguments below. */
uint64_t pending_signal_now_ms(void);
/* Keep signo for jobid, which has no start on this worker yet. If a start
* for jobid is already pending (it won the race), the signal is recorded on
* it as pending_signal_record_ex() would. Returns false only when the early
* table is full. */
bool pending_signal_record_early(uint64_t jobid, int signo, bool force,
int grace, uint64_t now_ms);
/* The worker has accepted a task start (runtasks) for jobid: as
* pending_signal_start(), but an unexpired early signal for jobid is kept
* for the start. Returns true when an early signal was adopted. A head-host
* start (pending_signal_start) does not adopt one: the controller sends a
* head its start before any control for the job. */
bool pending_signal_start_task(uint64_t jobid, uint64_t now_ms);
/* Number of entries, pending starts and early signals (for tests and
* diagnostics). */
int pending_signal_count(void);
#endif /* PENDING_SIGNAL_H */

43
src/worker/src/controller_connector.c
View file

@ -1391,8 +1391,14 @@ static int handle_runtasks(xce_request_t* request)
task_ctx->clientidx = runmsg->clientidx;
/* bug-1955: like a head-host start (bug-1908), from here until the
* task runs a signal for it is kept instead of being dropped */
pending_signal_start(runmsg->jobid);
* task runs a signal for it is kept instead of being dropped; a
* signal that reached this host before this request is kept too */
if (pending_signal_start_task(runmsg->jobid, pending_signal_now_ms())) {
XCECD_LOG(XCE_LOG_DEBUG,
"Job %s: a signal arrived before its tasks; it will be "
"delivered when they run",
jobidStr);
}
if (submit_fork_task(cc_ctx->thread_pool, task_ctx) != 0) {
XCECD_LOG(XCE_LOG_ERROR, "Failed to add task to thread pool");
@ -1610,7 +1616,38 @@ static int handle_jobcontrol(xce_request_t* request)
return 0;
}
} else {
XCECD_LOG(XCE_LOG_ERROR, "Failed to find worker for job ID: %s",
xcecd_finished_job_t finished;
/* a late control for a task that already finished here: drop
* it, so it cannot hit a later run of the job on this host */
if (xcecd_finished_job_lookup(d->jobid, resync_wallclock_now_ms(),
&finished)) {
XCECD_LOG(XCE_LOG_ERROR,
"Failed to find worker for job ID: %s (it has "
"finished on this host)",
jobidStr);
FREEUP(task_ctx);
return -1;
}
/* bug-1955: the controller sends control to every host of a
* multi-host job (bug-1942), but this host's runtasks request
* is relayed later, after the head has started: keep the
* signal until the task's start arrives (bounded, expiring) */
if (pending_signal_record_early(
d->jobid, d->u.siginfo->signo, d->u.siginfo->force != 0,
d->u.siginfo->grace, pending_signal_now_ms())) {
XCECD_LOG(XCE_LOG_DEBUG,
"Job %s has no task on this host yet; signal %d "
"is kept for its start",
jobidStr, d->u.siginfo->signo);
FREEUP(task_ctx);
return 0;
}
XCECD_LOG(XCE_LOG_ERROR,
"Failed to find worker for job ID: %s (too many "
"signals for jobs not started here)",
jobidStr);
FREEUP(task_ctx);
return -1;

125
src/worker/src/pending_signal.c
View file

@ -2,14 +2,17 @@
#include <pthread.h>
#include <stdlib.h>
#include <time.h>
/* bug-1908: see pending_signal.h. A short singly linked list -- it holds
* only jobs whose start is queued on this worker. */
typedef struct pending_signal_entry {
uint64_t jobid;
int signo; /* 0: no signal recorded */
int signo; /* 0: no signal recorded */
bool force;
int grace; /* bug-1957 */
int grace; /* bug-1957 */
bool early; /* bug-1955: no start yet */
uint64_t early_ms; /* bug-1955: when recorded */
struct pending_signal_entry* next;
} pending_signal_entry_t;
@ -26,6 +29,57 @@ static pending_signal_entry_t* find_locked(uint64_t jobid)
return e;
}
uint64_t pending_signal_now_ms(void)
{
struct timespec ts;
if (clock_gettime(CLOCK_MONOTONIC, &ts) != 0) {
return 0;
}
return (uint64_t)ts.tv_sec * 1000ULL + (uint64_t)ts.tv_nsec / 1000000ULL;
}
static bool early_expired(const pending_signal_entry_t* e, uint64_t now_ms)
{
return e->early && now_ms > e->early_ms
&& now_ms - e->early_ms > PENDING_SIGNAL_EARLY_TTL_MS;
}
/* bug-1955: drop expired early entries; returns how many early entries
* remain */
static int prune_early_locked(uint64_t now_ms)
{
pending_signal_entry_t** pp = &g_pending;
int early = 0;
while (*pp != NULL) {
pending_signal_entry_t* e = *pp;
if (early_expired(e, now_ms)) {
*pp = e->next;
free(e);
continue;
}
if (e->early) {
early++;
}
pp = &e->next;
}
return early;
}
static pending_signal_entry_t* add_locked(uint64_t jobid)
{
pending_signal_entry_t* e = (pending_signal_entry_t*)calloc(1, sizeof(*e));
if (e != NULL) {
e->jobid = jobid;
e->next = g_pending;
g_pending = e;
}
return e;
}
void pending_signal_start(uint64_t jobid)
{
pending_signal_entry_t* e;
@ -33,22 +87,72 @@ void pending_signal_start(uint64_t jobid)
pthread_mutex_lock(&g_lock);
e = find_locked(jobid);
if (e == NULL) {
e = (pending_signal_entry_t*)calloc(1, sizeof(*e));
if (e != NULL) {
e->jobid = jobid;
e->next = g_pending;
g_pending = e;
}
e = add_locked(jobid);
}
if (e != NULL) {
/* a new start of the same job does not inherit an old signal */
/* a new start of the same job does not inherit an old signal
* (nor, on a head host, an early one -- bug-1955) */
e->signo = 0;
e->force = false;
e->grace = 0;
e->early = false;
}
pthread_mutex_unlock(&g_lock);
}
bool pending_signal_start_task(uint64_t jobid, uint64_t now_ms)
{
pending_signal_entry_t* e;
bool adopted = false;
pthread_mutex_lock(&g_lock);
(void)prune_early_locked(now_ms);
e = find_locked(jobid);
if (e != NULL && e->early) {
/* bug-1955: the control came before this start; keep it */
e->early = false;
adopted = (e->signo != 0);
} else {
if (e == NULL) {
e = add_locked(jobid);
}
if (e != NULL) {
e->signo = 0;
e->force = false;
e->grace = 0;
}
}
pthread_mutex_unlock(&g_lock);
return adopted;
}
bool pending_signal_record_early(uint64_t jobid, int signo, bool force,
int grace, uint64_t now_ms)
{
pending_signal_entry_t* e;
int early;
pthread_mutex_lock(&g_lock);
early = prune_early_locked(now_ms);
e = find_locked(jobid);
if (e == NULL && early < PENDING_SIGNAL_EARLY_MAX) {
e = add_locked(jobid);
if (e != NULL) {
e->early = true;
e->early_ms = now_ms;
}
}
if (e != NULL) {
e->signo = signo;
e->force = force;
e->grace = grace;
}
pthread_mutex_unlock(&g_lock);
return (e != NULL);
}
bool pending_signal_record(uint64_t jobid, int signo, bool force)
{
return pending_signal_record_ex(jobid, signo, force, 0);
@ -75,8 +179,7 @@ bool pending_signal_take(uint64_t jobid, int* signo, bool* force)
return pending_signal_take_ex(jobid, signo, force, NULL);
}
bool pending_signal_take_ex(uint64_t jobid, int* signo, bool* force,
int* grace)
bool pending_signal_take_ex(uint64_t jobid, int* signo, bool* force, int* grace)
{
pending_signal_entry_t** pp;
bool found = false;

113
src/worker/testsuite/test_controller_connector.c
View file

@ -187,6 +187,24 @@ void xcecd_finished_job_retention_set_connection_healthy(bool healthy)
(void)healthy;
}
/* bug-1955: finished-job retention lookup; reports g_finished_jobid (when
* non-zero) as a task that already finished on this host. */
static uint64_t g_finished_jobid;
bool xcecd_finished_job_lookup(uint64_t jobid, uint64_t now_ms,
xcecd_finished_job_t* out)
{
(void)now_ms;
if (g_finished_jobid == 0 || jobid != g_finished_jobid) {
return false;
}
if (out != NULL) {
memset(out, 0, sizeof(*out));
out->jobid = jobid;
}
return true;
}
/* bug-1862 worker-list lock (xcecd_worker_utils.c is not linked here) */
void xcecd_worker_list_lock(void) {}
void xcecd_worker_list_unlock(void) {}
@ -1576,8 +1594,10 @@ static void test_jobcontrol_signal_for_queued_start_is_kept(void)
TEST_ASSERT_EQUAL_INT(15, signo);
}
/* A kill for a job this worker knows nothing about still fails. */
static void test_jobcontrol_signal_for_unknown_job_fails(void)
/* A kill for a job this worker knows nothing about used to fail; since
* bug-1955 it is kept, bounded and expiring, for a task start that may
* still come (a remote host of a multi-host job). */
static void test_jobcontrol_signal_for_unknown_job_is_kept_early(void)
{
controller_connector_ctx_t fake_cc = { 0 };
void* worker_list_head = NULL;
@ -1588,12 +1608,17 @@ static void test_jobcontrol_signal_for_unknown_job_fails(void)
xce_signal_t sig = { .signo = 15, .force = 0 };
xce_jobctrl_t ctrl = { .action = JobControl_u_siginfo, .jobid = 4343 };
xce_request_t request = { 0 };
int signo = 0;
(void)pending_signal_take(4242, NULL, NULL); /* independent of above */
ctrl.u.siginfo = &sig;
request.body.jobctrlmsg = &ctrl;
TEST_ASSERT_EQUAL_INT(-1, handle_jobcontrol(&request));
g_finished_jobid = 0;
TEST_ASSERT_EQUAL_INT(0, handle_jobcontrol(&request));
TEST_ASSERT_EQUAL_INT(1, pending_signal_count());
TEST_ASSERT_TRUE(pending_signal_take(4343, &signo, NULL));
TEST_ASSERT_EQUAL_INT(15, signo);
TEST_ASSERT_EQUAL_INT(0, pending_signal_count());
}
@ -1640,6 +1665,84 @@ static void test_jobcontrol_signal_for_queued_remote_task_is_kept(void)
xce_client_cfg.from = old_from;
}
/* bug-1955 (rest): the controller sends job control to every host of a
* multi-host job (bug-1942) directly, but a remote host's runtasks request
* comes later, relayed by the controller from the head worker once the head
* has started. A kill that reaches the remote host before its runtasks
* request must be kept and take effect when that request arrives. */
static void test_jobcontrol_signal_before_remote_runtasks_is_applied(void)
{
controller_connector_ctx_t fake_cc = { 0 };
void* worker_list_head = NULL;
fake_cc.running = true;
fake_cc.worker_list = &worker_list_head;
cc_ctx = &fake_cc;
char host[] = "rt-host";
char cmd[] = "sleep 60";
char user[] = "nobody";
xce_taskhost_t th = { .host = host, .ntasks = 1, .execuser = user,
.cmdline = cmd };
xce_taskhost_t* ths[1] = { &th };
xce_taskrun_t run = { .jobid = 4646, .nhosts = 1, .clientidx = 0,
.taskhosts = ths };
xce_request_t runreq = { 0 };
xce_signal_t sig = { .signo = 9, .force = 1 };
xce_jobctrl_t ctrl = { .action = JobControl_u_siginfo, .jobid = 4646 };
xce_request_t sigreq = { 0 };
char* old_from = xce_client_cfg.from;
int signo = 0;
bool force = false;
g_finished_jobid = 0;
/* the kill comes first: this host has no task for the job yet */
ctrl.u.siginfo = &sig;
sigreq.body.jobctrlmsg = &ctrl;
TEST_ASSERT_EQUAL_INT_MESSAGE(
0, handle_jobcontrol(&sigreq),
"a kill arriving before runtasks must not be dropped");
/* then the runtasks request */
xce_client_cfg.from = host;
runreq.body.runtaskmsg = &run;
TEST_ASSERT_EQUAL_INT(0, handle_runtasks(&runreq));
/* the task carries the kill to RUNNING (bug-1908 delivery path) */
TEST_ASSERT_TRUE_MESSAGE(pending_signal_take(4646, &signo, &force),
"the early kill must apply to the task");
TEST_ASSERT_EQUAL_INT(9, signo);
TEST_ASSERT_TRUE(force);
TEST_ASSERT_EQUAL_INT(0, pending_signal_count());
xce_client_cfg.from = old_from;
}
/* bug-1955: a late control for a task that already finished on this host
* (finished-job retention, worker-rejoin-resync) is not kept as an early
* signal -- it would otherwise hit a rerun of the job on this host. */
static void test_jobcontrol_signal_for_finished_task_is_not_kept(void)
{
controller_connector_ctx_t fake_cc = { 0 };
void* worker_list_head = NULL;
fake_cc.running = true;
fake_cc.worker_list = &worker_list_head;
cc_ctx = &fake_cc;
xce_signal_t sig = { .signo = 15, .force = 0 };
xce_jobctrl_t ctrl = { .action = JobControl_u_siginfo, .jobid = 4747 };
xce_request_t request = { 0 };
ctrl.u.siginfo = &sig;
request.body.jobctrlmsg = &ctrl;
g_finished_jobid = 4747;
TEST_ASSERT_EQUAL_INT(-1, handle_jobcontrol(&request));
TEST_ASSERT_EQUAL_INT_MESSAGE(0, pending_signal_count(),
"a control for a finished task is dropped");
g_finished_jobid = 0;
}
int main(void)
{
xce_init_xcecd_log("/tmp/test_controller_connector.log", 1024 * 1024, 2);
@ -1648,8 +1751,10 @@ int main(void)
RUN_TEST(test_build_cluster_config_no_peers_synthesizes_single_candidate);
RUN_TEST(test_jobcontrol_signal_for_queued_start_is_kept);
RUN_TEST(test_jobcontrol_signal_for_unknown_job_fails);
RUN_TEST(test_jobcontrol_signal_for_unknown_job_is_kept_early);
RUN_TEST(test_jobcontrol_signal_for_queued_remote_task_is_kept);
RUN_TEST(test_jobcontrol_signal_before_remote_runtasks_is_applied);
RUN_TEST(test_jobcontrol_signal_for_finished_task_is_not_kept);
RUN_TEST(
test_build_cluster_config_single_entry_peers_matches_synthesized_shape);
RUN_TEST(test_build_cluster_config_multi_candidate_passthrough);

104
src/worker/testsuite/test_pending_signal.c
View file

@ -75,6 +75,105 @@ static void test_restart_of_same_job_resets_signal(void)
TEST_ASSERT_FALSE(pending_signal_take(7, NULL, NULL));
}
/* bug-1955: a signal that arrives before this host has any start for the
* job (a remote host of a multi-host job, ahead of its runtasks request) is
* kept as an early entry and applied when the task's start arrives. */
static void test_early_signal_is_adopted_by_task_start(void)
{
int signo = 0;
bool force = false;
int grace = 0;
TEST_ASSERT_TRUE(pending_signal_record_early(8, 19, false, 5, 1000));
TEST_ASSERT_EQUAL_INT(1, pending_signal_count());
TEST_ASSERT_TRUE(pending_signal_start_task(8, 2000));
TEST_ASSERT_EQUAL_INT(1, pending_signal_count());
TEST_ASSERT_TRUE(pending_signal_take_ex(8, &signo, &force, &grace));
TEST_ASSERT_EQUAL_INT(19, signo);
TEST_ASSERT_FALSE(force);
TEST_ASSERT_EQUAL_INT(5, grace);
TEST_ASSERT_EQUAL_INT(0, pending_signal_count());
}
/* bug-1955: an early signal for a job whose start is already pending just
* records on that start (the start won the race). */
static void test_early_signal_on_pending_start_is_recorded(void)
{
int signo = 0;
pending_signal_start_task(9, 1000);
TEST_ASSERT_TRUE(pending_signal_record_early(9, 9, true, 0, 1000));
TEST_ASSERT_EQUAL_INT(1, pending_signal_count());
TEST_ASSERT_TRUE(pending_signal_take(9, &signo, NULL));
TEST_ASSERT_EQUAL_INT(9, signo);
}
/* bug-1955: an early entry whose start never comes (the control went to a
* host the job never ran on) expires, and an expired one is not applied to
* a later start of the same jobid. */
static void test_stale_early_signal_expires(void)
{
uint64_t t0 = 5000;
TEST_ASSERT_TRUE(pending_signal_record_early(10, 9, true, 0, t0));
TEST_ASSERT_TRUE(pending_signal_record_early(1, 15, false, 0, t0));
/* any later early record or task start prunes what has expired */
TEST_ASSERT_TRUE(pending_signal_record_early(
2, 15, false, 0, t0 + PENDING_SIGNAL_EARLY_TTL_MS + 1));
TEST_ASSERT_EQUAL_INT(1, pending_signal_count());
/* the start of the expired job does not inherit its kill */
TEST_ASSERT_FALSE(
pending_signal_start_task(10, t0 + PENDING_SIGNAL_EARLY_TTL_MS + 2));
TEST_ASSERT_FALSE(pending_signal_take(10, NULL, NULL));
TEST_ASSERT_EQUAL_INT(1, pending_signal_count()); /* job 2 */
/* the unexpired one is still applied */
TEST_ASSERT_TRUE(
pending_signal_start_task(2, t0 + PENDING_SIGNAL_EARLY_TTL_MS + 3));
TEST_ASSERT_TRUE(pending_signal_take(2, NULL, NULL));
TEST_ASSERT_EQUAL_INT(0, pending_signal_count());
}
/* bug-1955: early entries are bounded, so controls for unrelated jobids
* cannot grow the table without limit. */
static void test_early_signals_are_bounded(void)
{
uint64_t base = 1000000;
int i;
for (i = 0; i < PENDING_SIGNAL_EARLY_MAX; i++) {
TEST_ASSERT_TRUE(
pending_signal_record_early(base + (uint64_t)i, 15, false, 0, 1));
}
TEST_ASSERT_FALSE(pending_signal_record_early(
base + PENDING_SIGNAL_EARLY_MAX, 15, false, 0, 1));
TEST_ASSERT_EQUAL_INT(PENDING_SIGNAL_EARLY_MAX, pending_signal_count());
/* starts accepted by this worker are not limited by the early bound */
pending_signal_start(3);
TEST_ASSERT_EQUAL_INT(PENDING_SIGNAL_EARLY_MAX + 1,
pending_signal_count());
for (i = 0; i < PENDING_SIGNAL_EARLY_MAX; i++) {
(void)pending_signal_take(base + (uint64_t)i, NULL, NULL);
}
(void)pending_signal_take(3, NULL, NULL);
TEST_ASSERT_EQUAL_INT(0, pending_signal_count());
}
/* bug-1955: a head-host start (startjob) does not inherit an early entry;
* the controller sends a head its start before any control for the job,
* so an early entry for it can only be stale. */
static void test_head_start_does_not_inherit_early_signal(void)
{
TEST_ASSERT_TRUE(pending_signal_record_early(6, 9, true, 0, 1000));
pending_signal_start(6);
TEST_ASSERT_EQUAL_INT(1, pending_signal_count());
TEST_ASSERT_FALSE(pending_signal_take(6, NULL, NULL));
}
int main(void)
{
UNITY_BEGIN();
@ -84,5 +183,10 @@ int main(void)
RUN_TEST(test_later_signal_replaces_earlier);
RUN_TEST(test_jobs_are_independent);
RUN_TEST(test_restart_of_same_job_resets_signal);
RUN_TEST(test_early_signal_is_adopted_by_task_start);
RUN_TEST(test_early_signal_on_pending_start_is_recorded);
RUN_TEST(test_stale_early_signal_expires);
RUN_TEST(test_early_signals_are_bounded);
RUN_TEST(test_head_start_does_not_inherit_early_signal);
return UNITY_END();
}

10
src/worker/testsuite/test_worker_launch_type_dispatch.c
View file

@ -264,6 +264,16 @@ void xcecd_finished_job_retention_set_connection_healthy(bool healthy)
(void)healthy;
}
/* bug-1955: handle_jobcontrol() checks finished-job retention */
bool xcecd_finished_job_lookup(uint64_t jobid, uint64_t now_ms,
xcecd_finished_job_t* out)
{
(void)jobid;
(void)now_ms;
(void)out;
return false;
}
/* bug-1862 worker-list lock (xcecd_worker_utils.c is not linked here) */
void xcecd_worker_list_lock(void) {}
void xcecd_worker_list_unlock(void) {}