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

fix(scheduler): repeat WORKERS_PUSH with a NULL resval clobbers good values (bug-1688)

Browse source 
info worker all/<name> in the interactive/sim REPL initially printed
full per-resource values correctly, then lost them after a later
WORKERS_PUSH event replayed. core_apply_worker_resvals()'s per-resval
loop only guarded against nv==NULL/nv->name==NULL; a resval entry with
a NULL nv->val (a legitimate "no update this round" marker some
producers send) fell through and overwrote the worker's existing good
value with nothing, instead of being skipped -- same defensive pattern
already used elsewhere in this function, just missing here. A related
count-doubling gap in the same event path is fixed by switching an
iteration bound from `.count` to `.caps`.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01AgYRU3zDTqNGV5Jk8aiKLg
This commit is contained in:
Rongsong Shen 2026-08-27 08:26:20 +00:00
parent 47532f0a76
commit 0ebf3a817f
No known key found for this signature in database
GPG key ID: 5BCDAAE0123E9FE2
4 changed files with 286 additions and 3 deletions
Download patch file Download diff file Expand all files Collapse all files

26
src/scheduler/src/core_events.c
View file

@ -452,7 +452,29 @@ static void core_apply_worker_resvals(xce_sched_core_t* core,
xce_resource_t* r = NULL; xce_resource_t* r = NULL;
xce_val_t* val = NULL; xce_val_t* val = NULL;
if ((nv == NULL) || (nv->name == NULL)) { /* bug-1688: nv->val legitimately decodes to NULL (controller.capnp.c's
* decode_Value_ptr() sets *d = NULL whenever the wire's NameValue.val
* capnp pointer is CAPN_NULL) whenever a WORKERS_PUSH re-announces a
* worker's resource NAME list without fresh values for some/all of
* them (live-traced: a repeat WORKERS_PUSH for an already-registered
* worker can carry 20 resval names with nv->val==NULL for every one).
* Before this fix, the code below unconditionally MOVE_PTR'd this
* NULL into `val` and called zarray_put(&worker->resvals, r->idx,
* NULL) -- silently overwriting a perfectly good previously-known
* resource value with NULL. Because zarray_put_ex() (zarray.c) only
* increments array->count when the slot being written was NULL
* beforehand, the first such NULL-write left resvals.count
* unchanged (old slot was non-NULL, so the "else count++" branch
* never fired) while silently nulling the slot's DATA -- then a
* SECOND repeat push, now finding every slot already NULL, hit the
* "else count++" branch for all of them, producing the count-
* doubling anomaly (20 -> 40) this bug was originally reported
* with. Treating a NULL nv->val as "no update for this resource in
* this push" (skip, preserve whatever value is already there) is
* the same defensive "continue" posture this loop already applies
* to nv==NULL/nv->name==NULL above -- it protects the scheduler's
* own state regardless of why the wire push omitted a value. */
if ((nv == NULL) || (nv->name == NULL) || (nv->val == NULL)) {
continue; continue;
} }
@ -468,7 +490,7 @@ static void core_apply_worker_resvals(xce_sched_core_t* core,
* src/controller/dataex.c's collect_resval() TRACE (the matching * src/controller/dataex.c's collect_resval() TRACE (the matching
* "about to be wire-packaged" point) for the same resource/ * "about to be wire-packaged" point) for the same resource/
* worker/timestamp to isolate the wire encode/decode step. */ * worker/timestamp to isolate the wire encode/decode step. */
if (nv->val != NULL && nv->val->which == Value_u_fvalue) { if (nv->val->which == Value_u_fvalue) {
TRACE("%s worker '%s' resource '%s' received fvalue=%f", TRACE("%s worker '%s' resource '%s' received fvalue=%f",
__func__, w->name ? w->name : "(null)", nv->name, __func__, w->name ? w->name : "(null)", nv->name,
nv->val->u.fvalue); nv->val->u.fvalue);

14
src/scheduler/src/repl_info.c
View file

@ -321,7 +321,19 @@ static void repl_info_worker_one(xce_sched_sim_driver_t* sim,
iter_destroy(&it); iter_destroy(&it);
} }
for (i = 0; i < w->resvals.count; i++) { /* bug-1688: `.count` (number of currently-populated slots) is NOT a
* safe index upper bound for iterating this zarray BY INDEX -- it can
* legitimately be less than the highest occupied index once entries
* are sparse/non-contiguous (e.g. zarray_put_ex()'s count bookkeeping,
* zarray.c, only tracks "how many slots read NULL at write time", not
* "the current occupied-index high-water mark"). `.caps` (allocated
* capacity, monotonic, never shrinks) is the correct bound for
* index-based iteration -- mirrors resource.c:228's
* `int num_res = (int)w->resvals.caps;` for this identical structure.
* A `.caps`-bound loop can hit legitimately-empty slots that a
* `.count`-bound loop never would, so avail_v/rname's existing
* NULL-skip below (unchanged) is what keeps this safe. */
for (i = 0; i < w->resvals.caps; i++) {
xce_val_t* avail_v = (xce_val_t*)zarray_get(&(w->resvals), i); xce_val_t* avail_v = (xce_val_t*)zarray_get(&(w->resvals), i);
xce_val_t* used_v = (xce_val_t*)zarray_get(&(w->resInUse), i); xce_val_t* used_v = (xce_val_t*)zarray_get(&(w->resInUse), i);
const char* rname = find_resource_name_by_idx(res_arr, res_n, i); const char* rname = find_resource_name_by_idx(res_arr, res_n, i);

142
src/scheduler/testsuite/test_core_events.c
View file

@ -644,6 +644,147 @@ static void test_workers_push_twice_updates_existing_worker_in_place(void)
TEST_ASSERT_EQUAL_FLOAT(200.0f, (float)applied->u.fvalue); TEST_ASSERT_EQUAL_FLOAT(200.0f, (float)applied->u.fvalue);
} }
/*
* bug-1688 regression: a WORKERS_PUSH re-announcing an already-registered
* worker's resource NAME list can legitimately carry nv->val == NULL for
* some/all entries (controller.capnp.c's decode_Value_ptr() decodes a wire
* NameValue.val CAPN_NULL pointer to exactly this -- live-traced against a
* real replay: a repeat push for worker 'pc13v' carried 20 resource names
* with nv->val == NULL for every one). Before this fix,
* core_apply_worker_resvals() unconditionally zarray_put()'d this NULL,
* silently clobbering a previously-known-good resource value with NULL.
* Because zarray_put_ex() (zarray.c) only increments array->count when the
* slot being written was NULL beforehand, that first NULL-write left
* resvals.count unchanged (the old slot was non-NULL) while zeroing the
* slot's DATA -- so a THIRD push, now finding every slot already NULL, hit
* the "was NULL, count++" branch for each one, doubling resvals.count (the
* literal 20 -> 40 anomaly this bug was reported with) despite carrying the
* exact same resource names both times. Two resources (not one) are used
* here so the count-doubling assertion is meaningful, not a 1-vs-2
* coincidence.
*/
static void test_workers_push_with_null_val_does_not_clobber_existing_resval(
void)
{
xce_sched_core_t core;
xce_core_event_t event;
xce_resource_t wire_mem;
xce_resource_t wire_r1m;
xce_worker_t wire_worker;
xce_worker_t* worker_ptrs[1];
xce_val_t val_mem;
xce_val_t val_r1m;
xce_name_val_t nv_mem;
xce_name_val_t nv_r1m;
xce_name_val_t* nv_ptrs[2];
xce_name_val_t nv_mem_null;
xce_name_val_t nv_r1m_null;
xce_name_val_t* nv_null_ptrs[2];
xce_sched_worker_t* w;
xce_resource_t* r_mem = NULL;
xce_resource_t* r_r1m = NULL;
xce_val_t* applied;
int rc;
fixture_init(&core, &g_full_plugin);
memset(&wire_mem, 0, sizeof(wire_mem));
wire_mem.name = "mem";
TEST_ASSERT_NOT_NULL(makeResource(core.resmgr, &wire_mem));
memset(&wire_r1m, 0, sizeof(wire_r1m));
wire_r1m.name = "r1m";
TEST_ASSERT_NOT_NULL(makeResource(core.resmgr, &wire_r1m));
TEST_ASSERT_TRUE(key_find(core.resmgr, (void*)"mem", (void**)&r_mem) >= 0);
TEST_ASSERT_NOT_NULL(r_mem);
TEST_ASSERT_TRUE(key_find(core.resmgr, (void*)"r1m", (void**)&r_r1m) >= 0);
TEST_ASSERT_NOT_NULL(r_r1m);
/* First push: worker "hostA" joins with real mem/r1m values. */
memset(&val_mem, 0, sizeof(val_mem));
val_mem.which = Value_u_fvalue;
val_mem.u.fvalue = 100.0;
memset(&nv_mem, 0, sizeof(nv_mem));
nv_mem.name = "mem";
nv_mem.val = &val_mem;
memset(&val_r1m, 0, sizeof(val_r1m));
val_r1m.which = Value_u_fvalue;
val_r1m.u.fvalue = 5.0;
memset(&nv_r1m, 0, sizeof(nv_r1m));
nv_r1m.name = "r1m";
nv_r1m.val = &val_r1m;
nv_ptrs[0] = &nv_mem;
nv_ptrs[1] = &nv_r1m;
memset(&wire_worker, 0, sizeof(wire_worker));
wire_worker.name = "hostA";
wire_worker.status = 0;
wire_worker.nresvals_to_sched = 2;
wire_worker.resvals_to_sched = nv_ptrs;
worker_ptrs[0] = &wire_worker;
memset(&event, 0, sizeof(event));
event.type = XCE_CORE_EVENT_WORKERS_PUSH;
event.u.workers_push.nworkers = 1;
event.u.workers_push.workers = worker_ptrs;
rc = xce_core_push_events(&core, &event, 1);
TEST_ASSERT_EQUAL_INT(0, rc);
w = sched_get_worker(&core, "hostA");
TEST_ASSERT_NOT_NULL(w);
TEST_ASSERT_EQUAL_INT(2, w->resvals.count);
applied = (xce_val_t*)zarray_get(&(w->resvals), r_mem->idx);
TEST_ASSERT_NOT_NULL(applied);
TEST_ASSERT_EQUAL_FLOAT(100.0f, (float)applied->u.fvalue);
applied = (xce_val_t*)zarray_get(&(w->resvals), r_r1m->idx);
TEST_ASSERT_NOT_NULL(applied);
TEST_ASSERT_EQUAL_FLOAT(5.0f, (float)applied->u.fvalue);
/* Second push: SAME worker, SAME resource names, but nv->val == NULL
* for both -- the exact "re-announce names, no fresh values" shape
* live-traced from the real replay. Must NOT clobber the existing
* values, and must NOT change resvals.count. */
memset(&nv_mem_null, 0, sizeof(nv_mem_null));
nv_mem_null.name = "mem";
nv_mem_null.val = NULL;
memset(&nv_r1m_null, 0, sizeof(nv_r1m_null));
nv_r1m_null.name = "r1m";
nv_r1m_null.val = NULL;
nv_null_ptrs[0] = &nv_mem_null;
nv_null_ptrs[1] = &nv_r1m_null;
wire_worker.resvals_to_sched = nv_null_ptrs;
rc = xce_core_push_events(&core, &event, 1);
TEST_ASSERT_EQUAL_INT(0, rc);
w = sched_get_worker(&core, "hostA");
TEST_ASSERT_NOT_NULL(w);
TEST_ASSERT_EQUAL_INT(2, w->resvals.count);
applied = (xce_val_t*)zarray_get(&(w->resvals), r_mem->idx);
TEST_ASSERT_NOT_NULL(applied);
TEST_ASSERT_EQUAL_FLOAT(100.0f, (float)applied->u.fvalue);
applied = (xce_val_t*)zarray_get(&(w->resvals), r_r1m->idx);
TEST_ASSERT_NOT_NULL(applied);
TEST_ASSERT_EQUAL_FLOAT(5.0f, (float)applied->u.fvalue);
/* Third push: SAME shape again (still NULL vals) -- if the bug were
* present, resvals.count would double here (20 -> 40 in the live
* trace; 2 -> 4 here) because the SECOND push (above) would have
* already nulled the slots, making THIS push's writes look like
* fresh/first-time occupancy. Confirms the fix holds across repeats,
* not just once. */
rc = xce_core_push_events(&core, &event, 1);
TEST_ASSERT_EQUAL_INT(0, rc);
w = sched_get_worker(&core, "hostA");
TEST_ASSERT_NOT_NULL(w);
TEST_ASSERT_EQUAL_INT(2, w->resvals.count);
}
/* JOB_RUSAGE's on_job_rusage hook is OPTIONAL (Req 5 AC 4) -- a picker /* JOB_RUSAGE's on_job_rusage hook is OPTIONAL (Req 5 AC 4) -- a picker
* without it must still apply the state mutation and must not error or * without it must still apply the state mutation and must not error or
* crash. */ * crash. */
@ -834,6 +975,7 @@ int main(void)
RUN_TEST(test_jobs_push_fires_on_job_push_once_per_job); RUN_TEST(test_jobs_push_fires_on_job_push_once_per_job);
RUN_TEST(test_non_picker_events_mutate_state_but_never_call_hooks); RUN_TEST(test_non_picker_events_mutate_state_but_never_call_hooks);
RUN_TEST(test_workers_push_twice_updates_existing_worker_in_place); RUN_TEST(test_workers_push_twice_updates_existing_worker_in_place);
RUN_TEST(test_workers_push_with_null_val_does_not_clobber_existing_resval);
RUN_TEST(test_job_rusage_applies_state_when_hook_is_null); RUN_TEST(test_job_rusage_applies_state_when_hook_is_null);
RUN_TEST(test_config_push_decomposes_queues_resources_and_resvals); RUN_TEST(test_config_push_decomposes_queues_resources_and_resvals);
RUN_TEST(test_push_events_rejects_with_ebusy_when_session_active_no_mutation); RUN_TEST(test_push_events_rejects_with_ebusy_when_session_active_no_mutation);

107
src/scheduler/testsuite/test_repl_breakpoints_inspection.c
View file

@ -927,6 +927,112 @@ static void test_info_worker_reports_string_and_boolean_resource_values(void)
"PASS\n"); "PASS\n");
} }
/*
* bug-1688: repl_info_worker_one()'s resource-print loop used
* `for (i = 0; i < w->resvals.count; i++)` -- treating `.count` (number of
* currently-populated slots) as a safe index upper bound for iterating the
* zarray BY INDEX. `.count` can legitimately be less than the highest
* occupied index once entries are sparse/non-contiguous (zarray_put_ex()'s
* count bookkeeping, zarray.c, only tracks "how many slots read NULL at
* write time" -- it is not recomputed from the array's actual contents).
* `.caps` (allocated capacity, monotonic, never shrinks) is this codebase's
* own established convention for this identical structure -- see
* resource.c:228's `int num_res = (int)w->resvals.caps;`. Regression: a
* worker whose resvals zarray has TWO populated slots (idx 0 "mem", idx 1
* "r1m") but whose `.count` field has been driven down to 1 (the exact
* post-corruption shape bug-1688's OTHER root cause -- the NULL-nv->val
* WORKERS_PUSH overwrite, core_events.c -- can leave resvals in, and
* reproducible directly here without needing that whole mechanism) must
* still show BOTH resource lines, not silently truncate to the first one.
*/
static void test_info_worker_shows_all_populated_resources_when_count_is_stale(
void)
{
xce_sched_sim_driver_t sim;
xce_core_event_t events[1];
xce_resource_t wire_mem;
xce_resource_t wire_r1m;
xce_worker_t wire_worker;
xce_worker_t* worker_ptrs[1];
xce_val_t val_mem;
xce_val_t val_r1m;
xce_name_val_t nv_mem;
xce_name_val_t nv_r1m;
xce_name_val_t* nv_ptrs[2];
xce_sched_worker_t* sw;
char* captured;
int rc;
memset(&sim, 0, sizeof(sim));
rc = xce_sched_core_init(&(sim.core));
TEST_ASSERT_EQUAL_INT(0, rc);
memset(&wire_mem, 0, sizeof(wire_mem));
wire_mem.name = "mem";
TEST_ASSERT_NOT_NULL(makeResource(sim.core.resmgr, &wire_mem));
memset(&wire_r1m, 0, sizeof(wire_r1m));
wire_r1m.name = "r1m";
TEST_ASSERT_NOT_NULL(makeResource(sim.core.resmgr, &wire_r1m));
memset(&val_mem, 0, sizeof(val_mem));
val_mem.which = Value_u_fvalue;
val_mem.u.fvalue = 100.0;
memset(&nv_mem, 0, sizeof(nv_mem));
nv_mem.name = "mem";
nv_mem.val = &val_mem;
memset(&val_r1m, 0, sizeof(val_r1m));
val_r1m.which = Value_u_fvalue;
val_r1m.u.fvalue = 5.0;
memset(&nv_r1m, 0, sizeof(nv_r1m));
nv_r1m.name = "r1m";
nv_r1m.val = &val_r1m;
nv_ptrs[0] = &nv_mem;
nv_ptrs[1] = &nv_r1m;
memset(&wire_worker, 0, sizeof(wire_worker));
wire_worker.name = "hostA";
wire_worker.status = 0;
wire_worker.nresvals_to_sched = 2;
wire_worker.resvals_to_sched = nv_ptrs;
worker_ptrs[0] = &wire_worker;
memset(events, 0, sizeof(events));
events[0].type = XCE_CORE_EVENT_WORKERS_PUSH;
events[0].u.workers_push.nworkers = 1;
events[0].u.workers_push.workers = worker_ptrs;
rc = xce_core_push_events(&(sim.core), events, 1);
TEST_ASSERT_EQUAL_INT(0, rc);
sw = sched_get_worker(&(sim.core), "hostA");
TEST_ASSERT_NOT_NULL(sw);
TEST_ASSERT_EQUAL_INT(2, sw->resvals.count);
/* Directly force the sparse-vs-stale-count shape: both idx 0 ("mem")
* and idx 1 ("r1m") stay populated, but .count now under-reports the
* true occupied high-water mark -- exactly what a `.count`-bound loop
* gets wrong and a `.caps`-bound loop (this fix) does not. */
sw->resvals.count = 1;
captured = capture_dispatch(&sim, "info worker hostA");
TEST_ASSERT_NOT_NULL_MESSAGE(
strstr(captured, "resource mem: value=100.00"),
"the first populated slot must still display");
TEST_ASSERT_NOT_NULL_MESSAGE(
strstr(captured, "resource r1m: value=5.00"),
"a populated slot at an index >= the (stale) .count must still "
"display -- iterating by .count instead of .caps (bug-1688) "
"silently truncates output here");
free(captured);
printf("test_info_worker_shows_all_populated_resources_when_count_is_"
"stale: PASS\n");
}
/* /*
* bug-1590: repl_info_resource_one() (the "info resource <name>" * bug-1590: repl_info_resource_one() (the "info resource <name>"
* aggregate-across-workers command) had the same defect one layer up -- * aggregate-across-workers command) had the same defect one layer up --
@ -1880,6 +1986,7 @@ int main(void)
RUN_TEST(test_info_worker_reports_resource_values); RUN_TEST(test_info_worker_reports_resource_values);
RUN_TEST(test_info_worker_reports_consumable_value_and_used); RUN_TEST(test_info_worker_reports_consumable_value_and_used);
RUN_TEST(test_info_worker_reports_string_and_boolean_resource_values); RUN_TEST(test_info_worker_reports_string_and_boolean_resource_values);
RUN_TEST(test_info_worker_shows_all_populated_resources_when_count_is_stale);
RUN_TEST(test_info_resource_reports_string_value); RUN_TEST(test_info_resource_reports_string_value);
RUN_TEST(test_info_resource_reports_consumable_value_and_used); RUN_TEST(test_info_resource_reports_consumable_value_and_used);
RUN_TEST(test_info_resource_all_lists_only_global_shared_resources); RUN_TEST(test_info_resource_all_lists_only_global_shared_resources);