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

fix(controller): job_ctrl lock/reference and dispatch retry requeue (bug-448, bug-1626, bug-1631)

Browse source 
bug-448: job_ctrl() took a reference on the job but only one error path
released it; every successful call leaked one (refcount 1 -> 3 after two
calls). It is now released on every exit path, including the key_find
rc<0 one.

bug-1626: job_ctrl()'s not-started branch changes the job (mark_job_update
/ mark_job_fail -> apply_job_*_transition, which require RTTI_LOCK_WRITE)
while holding only RTTI_LOCK_READ, so it could run concurrently with
readers. It now takes the write lock. (dispatch_a_job_ already did.)

bug-1631: dispatch_a_job_()'s four retryable-failure branches never reset
state/allocs, correct for a first dispatch (still PEND) but not for a job
that had already been dispatched once: it stayed RUN holding its old
allocation. Each branch now calls record_job_reconcile_pending_fact() --
a no-op without a WAL fact for a PEND job; for a RUN job it returns it to
PEND, releases allocs and appends a durable WAL_JOB_UPDATE -- and notifies
after unlocking when it applied, the same pattern as the resync reconcile
path (cfe66dde). Not reachable until something re-dispatches RUN jobs
(e.g. preemption), fixed now so it is correct when that lands.

- job_get_refcount() (jobj.c) read accessor for the leak test.
- test_apply_pump.c: test_job_ctrl_releases_its_job_reference and
  test_job_ctrl_mutates_not_started_job_under_write_lock (both RED before).
- test_dispatch_fail_retry.c now asserts PEND + allocs released after a
  post-success retryable failure (RED before).

Controller suite 1236/0; all 111 meson suites pass.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
This commit is contained in:
Rongsong Shen 2026-09-27 03:05:20 +00:00
parent d43ae58edc
commit 0c1bac4426
Signed by untrusted user: rshen
GPG key ID: 80533536B0EB651C
5 changed files with 163 additions and 4 deletions
Download patch file Download diff file Expand all files Collapse all files

63
src/controller/job.c
View file

@ -1395,6 +1395,7 @@ static int dispatch_a_job_(cluster_t* c, endpoint_t* ep, xce_job_alloc_t* alloc)
xce_dispatch_fail_class_t dfclass;
job_failinfo_t failinfo = { .jobid = alloc->jobid };
bool resolved_retryable = false;
bool requeued = false;
uint32_t failcount;
lock(j, RTTI_LOCK_WRITE);
@ -1406,6 +1407,11 @@ static int dispatch_a_job_(cluster_t* c, endpoint_t* ep, xce_job_alloc_t* alloc)
failcount = ++j->dispatch_fail_count;
if (failcount <= XCE_DISPATCH_FAIL_RETRY_BOUND) {
resolved_retryable = true;
/* bug-1631: a job that was already dispatched once must go
* back to PEND without its old allocation; a no-op (no WAL
* fact) for the usual first-dispatch case, still PEND. */
requeued = (record_job_reconcile_pending_fact(c, j)
== (int)JOB_TRANSITION_APPLIED);
} else {
failinfo.code = rc;
failinfo.detail = "dispatch failed: worker lookup failed "
@ -1438,8 +1444,15 @@ static int dispatch_a_job_(cluster_t* c, endpoint_t* ep, xce_job_alloc_t* alloc)
unlock(j);
if (requeued)
persist_notify_job_updated_already_persisted(c, j);
dereference(j);
if (resolved_retryable) {
notify_dispatch(c, ep, alloc->jobid, rc);
}
@ -1452,6 +1465,7 @@ static int dispatch_a_job_(cluster_t* c, endpoint_t* ep, xce_job_alloc_t* alloc)
xce_dispatch_fail_class_t dfclass;
job_failinfo_t failinfo = { .jobid = alloc->jobid };
bool resolved_retryable = false;
bool requeued = false;
uint32_t failcount;
/* scheduler-dispatch-commit task 3.3: capture errno FIRST, before
@ -1473,6 +1487,11 @@ static int dispatch_a_job_(cluster_t* c, endpoint_t* ep, xce_job_alloc_t* alloc)
failcount = ++j->dispatch_fail_count;
if (failcount <= XCE_DISPATCH_FAIL_RETRY_BOUND) {
resolved_retryable = true;
/* bug-1631: a job that was already dispatched once must go
* back to PEND without its old allocation; a no-op (no WAL
* fact) for the usual first-dispatch case, still PEND. */
requeued = (record_job_reconcile_pending_fact(c, j)
== (int)JOB_TRANSITION_APPLIED);
} else {
failinfo.code = dfrc;
failinfo.detail = "dispatch failed: worker not found "
@ -1500,8 +1519,15 @@ static int dispatch_a_job_(cluster_t* c, endpoint_t* ep, xce_job_alloc_t* alloc)
unlock(j);
if (requeued)
persist_notify_job_updated_already_persisted(c, j);
dereference(j);
if (resolved_retryable) {
notify_dispatch(c, ep, alloc->jobid, dfrc);
}
@ -1574,6 +1600,7 @@ static int dispatch_a_job_(cluster_t* c, endpoint_t* ep, xce_job_alloc_t* alloc)
xce_dispatch_fail_class_t dfclass;
job_failinfo_t failinfo = { .jobid = alloc->jobid };
bool resolved_retryable = false;
bool requeued = false;
uint32_t failcount;
ERROR("worker %s is not in correct state, status %x ep %p",
@ -1602,6 +1629,11 @@ static int dispatch_a_job_(cluster_t* c, endpoint_t* ep, xce_job_alloc_t* alloc)
failcount = ++j->dispatch_fail_count;
if (failcount <= XCE_DISPATCH_FAIL_RETRY_BOUND) {
resolved_retryable = true;
/* bug-1631: a job that was already dispatched once must go
* back to PEND without its old allocation; a no-op (no WAL
* fact) for the usual first-dispatch case, still PEND. */
requeued = (record_job_reconcile_pending_fact(c, j)
== (int)JOB_TRANSITION_APPLIED);
} else {
failinfo.code = dfrc;
failinfo.detail = "dispatch failed: worker not ready "
@ -1630,8 +1662,15 @@ static int dispatch_a_job_(cluster_t* c, endpoint_t* ep, xce_job_alloc_t* alloc)
unlock(j);
if (requeued)
persist_notify_job_updated_already_persisted(c, j);
dereference(j);
if (resolved_retryable) {
notify_dispatch(c, ep, alloc->jobid, dfrc);
}
@ -1810,6 +1849,7 @@ static int dispatch_a_job_(cluster_t* c, endpoint_t* ep, xce_job_alloc_t* alloc)
xce_dispatch_fail_class_t dfclass;
job_failinfo_t failinfo = { .jobid = alloc->jobid };
bool resolved_retryable = false;
bool requeued = false;
uint32_t failcount;
SET_ERROR(-rc);
@ -1830,6 +1870,11 @@ static int dispatch_a_job_(cluster_t* c, endpoint_t* ep, xce_job_alloc_t* alloc)
failcount = ++j->dispatch_fail_count;
if (failcount <= XCE_DISPATCH_FAIL_RETRY_BOUND) {
resolved_retryable = true;
/* bug-1631: a job that was already dispatched once must go
* back to PEND without its old allocation; a no-op (no WAL
* fact) for the usual first-dispatch case, still PEND. */
requeued = (record_job_reconcile_pending_fact(c, j)
== (int)JOB_TRANSITION_APPLIED);
} else {
failinfo.code = rc;
failinfo.detail = "dispatch failed: xce_startjob_req "
@ -1863,8 +1908,15 @@ static int dispatch_a_job_(cluster_t* c, endpoint_t* ep, xce_job_alloc_t* alloc)
unlock(j);
if (requeued)
persist_notify_job_updated_already_persisted(c, j);
dereference(j);
if (resolved_retryable) {
notify_dispatch(c, ep, alloc->jobid, rc);
}
@ -1901,6 +1953,8 @@ int job_ctrl(cluster_t* c, xce_jobctrl_t* ctrinfo)
unlock(c->jobmgr);
if (rc < 0) {
if (j != NULL)
dereference(j);
return rc;
}
@ -1918,8 +1972,12 @@ int job_ctrl(cluster_t* c, xce_jobctrl_t* ctrinfo)
* -ENOENT, and every other branch sets rc itself. */
rc = 0;
/* FIXME: eliminate held multiple locks */
lock(j, RTTI_LOCK_READ);
/* FIXME: eliminate held multiple locks
*
* bug-1626: WRITE, not READ -- the not-started branch below changes the
* job (mark_job_update / mark_job_fail -> apply_job_*_transition, which
* require the caller to hold the job under RTTI_LOCK_WRITE). */
lock(j, RTTI_LOCK_WRITE);
if (j->job.allocs != NULL) {
xce_host_alloc_t** allocs = j->job.allocs->allocs;
worker_t* w = NULL;
@ -2024,6 +2082,7 @@ int job_ctrl(cluster_t* c, xce_jobctrl_t* ctrinfo)
}
unlock(j);
dereference(j); /* bug-448: the reference taken above, on every path */
return rc;
}

82
src/controller/testsuite/test_apply_pump.c
View file

@ -968,6 +968,86 @@ static void test_apply_pump_guard_does_not_break_job_ctrl_live_append(void)
"PASS\n");
}
/* ── bug-448 / bug-1626: job_ctrl's own reference and lock discipline ── */
/* bug-448: job_ctrl takes a reference on the job and must drop it on every
* exit path; the normal (not-started) path never did. */
static void test_job_ctrl_releases_its_job_reference(void)
{
tap_setup();
job_t* j = tap_seed_job(907, JOB_STAT_PSUSP);
TEST_ASSERT_NOT_NULL(j);
unsigned long before = job_get_refcount(j);
xce_signal_t sig = { .signo = SIGCONT, .force = 0 };
xce_jobctrl_t ctrl = { .action = JobControl_u_siginfo, .jobid = 907 };
ctrl.u.siginfo = &sig;
int rc = job_ctrl(&g_tap_c, &ctrl);
sig.signo = SIGKILL;
int rc2 = job_ctrl(&g_tap_c, &ctrl);
unsigned long after = job_get_refcount(j);
tap_teardown();
TEST_ASSERT_EQUAL_INT(0, rc);
TEST_ASSERT_EQUAL_INT(0, rc2);
TEST_ASSERT_EQUAL_UINT_MESSAGE(before, after,
"bug-448: job_ctrl must release the reference it takes on the job");
}
typedef struct {
job_t* j;
atomic_int done;
int rc;
} tap_jobctrl_arg_t;
static void* tap_jobctrl_thread(void* arg)
{
tap_jobctrl_arg_t* a = (tap_jobctrl_arg_t*)arg;
xce_signal_t sig = { .signo = SIGSTOP, .force = 0 };
xce_jobctrl_t ctrl = { .action = JobControl_u_siginfo,
.jobid = a->j->job.jobid };
ctrl.u.siginfo = &sig;
a->rc = job_ctrl(&g_tap_c, &ctrl);
atomic_store(&a->done, 1);
return NULL;
}
/* bug-1626: the not-started branch mutates the job (mark_job_update /
* mark_job_fail), so job_ctrl must hold the job's WRITE lock -- it may not
* proceed while another thread holds a read lock on the job. */
static void test_job_ctrl_mutates_not_started_job_under_write_lock(void)
{
tap_setup();
job_t* j = tap_seed_job(908, JOB_STAT_PEND);
TEST_ASSERT_NOT_NULL(j);
tap_jobctrl_arg_t arg = { .j = j, .rc = -1 };
atomic_store(&arg.done, 0);
lock(j, RTTI_LOCK_READ);
pthread_t th;
TEST_ASSERT_EQUAL_INT(0, pthread_create(&th, NULL, tap_jobctrl_thread,
&arg));
usleep(200 * 1000);
int done_while_reader_held = atomic_load(&arg.done);
unlock(j);
pthread_join(th, NULL);
uint32_t state = j->job.state;
tap_teardown();
TEST_ASSERT_EQUAL_INT_MESSAGE(0, done_while_reader_held,
"bug-1626: job_ctrl changed a job while another thread held its "
"read lock -- it must take RTTI_LOCK_WRITE");
TEST_ASSERT_EQUAL_INT(0, arg.rc);
TEST_ASSERT_TRUE(state & JOB_STAT_PSUSP);
}
/* ── 4. Merge-staleness regression: FAIL must not clobber a racing FINISH ─
*
* Reproduces the round-1 corruption bug found by review: a FINISH fact
@ -1769,6 +1849,8 @@ void test_apply_pump(void)
RUN_TEST(test_apply_pump_run_engages_live_fence_on_failed_lsn);
RUN_TEST(test_apply_pump_run_self_clears_fence_on_successful_retry);
RUN_TEST(test_apply_pump_guard_does_not_break_job_ctrl_live_append);
RUN_TEST(test_job_ctrl_releases_its_job_reference);
RUN_TEST(test_job_ctrl_mutates_not_started_job_under_write_lock);
RUN_TEST(test_apply_pump_fail_racing_finish_merges_against_current_state);
RUN_TEST(test_apply_pump_late_pending_apply_registration_self_heals);
RUN_TEST(test_register_pending_apply_race_against_concurrent_apply_self_heals);

12
src/controller/testsuite/test_dispatch_fail_retry.c
View file

@ -1058,8 +1058,16 @@ static void test_rpc_failure_bound_resets_after_intervening_success(void)
"Req 4.3: the counter must count purely from the post-success "
"reset, not from the job's total pre-success failure history");
/* NOT asserted here: job.state == JOB_STAT_PEND / job.allocs ==
* NULL after this retryable failure. A kiro-debug investigation
/* bug-1631 (fixed): a retryable failure of a job that had already
* been dispatched returns it to PEND and releases its allocation,
* durably (record_job_reconcile_pending_fact). */
TEST_ASSERT_EQUAL_UINT32_MESSAGE(JOB_STAT_PEND, j->job.state,
"bug-1631: a retryable dispatch failure must return the job to "
"PEND even after a prior successful dispatch");
TEST_ASSERT_NULL_MESSAGE(j->job.allocs,
"bug-1631: the job's prior allocation must be released");
/* History: originally NOT asserted. A kiro-debug investigation
* (HIGH confidence; bug-1631) found this exact sub-scenario -- a
* retryable dispatch failure occurring while the job's
* controller-side state is already RUN, from step 2's real

1
src/include/jobj.h
View file

@ -9,6 +9,7 @@
extern job_submit_data_t* makeJobsub(job_submit_data_t*);
extern job_startinfo_t* makeJobstartinfo(job_startinfo_t*);
extern job_t* makeJob(var, xce_job_t*);
extern unsigned long job_get_refcount(job_t*);
extern int job_obj_init(int, int);
extern void job_obj_fini();

9
src/lib/jobj.c
View file

@ -708,6 +708,15 @@ static void job_dereference(meta_obj_t* obj)
return;
}
/* Current reference count of `j` (diagnostics / leak tests, bug-448). */
unsigned long job_get_refcount(job_t* j)
{
meta_obj_t* obj = RTTI_DATA_TO_OBJ(j);
job_private_t* priv = (job_private_t*)RTTI_OBJ_TO_PRIV(obj);
return (unsigned long)atomic_load(&(priv->refcount));
}
static int job_lock(meta_obj_t* obj, uint64_t option)
{
job_private_t* priv = (job_private_t*)RTTI_OBJ_TO_PRIV(obj);