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

fix(scheduler): record events before the core applies them (bug-1852)

Browse source 
The recorder encoded each controller event batch after
xce_core_push_events() had applied it, and applying moves payload fields
out of the events: a WORKERS_PUSH's resvals (core_apply_worker_resvals
MOVE_PTR), a JOBS_PUSH's allocs, a JOB_RUSAGE's rusages, and a
CONFIG_PUSH's shared resvals (sched_apply_pending_config recorded after
sched_init_config). Recordings silently lost exactly those fields -- a
worker's resval names were written with NULL values, so --sim-verify
replayed "no slots" and turned 3 live DISPATCHes into PENDs.

driver.c now records the batch before the push; the periodic in-stream
SnapshotEntry moves out of sched_record_write_events() into new
sched_record_checkpoint_after_apply(), called after the push so a
checkpoint still reflects the events written ahead of it. The config
push is likewise recorded before sched_init_config().

Regression test: test_recorded_workers_push_keeps_resval_values (RED:
recorded value NULL; GREEN after). Periodic-checkpoint tests updated to
the write-then-checkpoint contract. meson test 109/109; 25 live soak
recordings replay with 0 mismatches.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
This commit is contained in:
Rongsong Shen 2026-09-24 08:41:55 +00:00
parent ca4db13a79
commit 9c02184fb4
No known key found for this signature in database
GPG key ID: 5BCDAAE0123E9FE2
10 changed files with 258 additions and 33 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -662,6 +662,135 @@ test_full_cycle_start_event_session_stop_produces_expected_stream(void)
" PASS\n");
}
/* ---------------------------------------------------------------------- */
/* bug-1852: a recorded event must carry its payload as RECEIVED, not what */
/* is left of it after the core applied it. core_apply_worker_resvals() */
/* MOVE_PTRs each resval out of the WORKERS_PUSH payload; recording after */
/* xce_core_push_events() wrote the worker's resval names with NULL values, */
/* so a replay never learned the worker's capacity (live soak: a recording */
/* opened before the worker's first value push replayed as "no slots" -- */
/* 3 live DISPATCHes became sim PENDs). */
/* ---------------------------------------------------------------------- */
static void skip_header_and_initial_snapshot(FILE* f); /* defined below */
static void test_recorded_workers_push_keeps_resval_values(void)
{
char record_dir[PATH_MAX];
xce_scheduler_t sched;
char path_copy[PATH_MAX];
xce_sched_update_request_t* rr;
xce_sched_update_request_t* rw;
xce_resource_t* wire_res;
xce_resource_t** res_ptrs;
xce_worker_t* wire_w;
xce_worker_t** w_ptrs;
xce_name_val_t* nv;
int rc;
int found = 0;
FILE* f;
struct capn c;
make_tmp_dir(record_dir, sizeof(record_dir), "_1852");
fixture_init(&sched, record_dir);
rc = sched_record_start(&sched, "workers.replay");
TEST_ASSERT_EQUAL_INT(0, rc);
strcpy(path_copy, sched.record_file_path);
/* Heap payloads throughout: release_translated_payload() frees them
* like genuinely wire-decoded ones (test_resval_ownership.c's own
* warning about stack/literal payloads applies verbatim). */
wire_res = (xce_resource_t*)calloc(1, sizeof(*wire_res));
wire_res->name = strdup("slots");
res_ptrs = (xce_resource_t**)calloc(1, sizeof(*res_ptrs));
res_ptrs[0] = wire_res;
nv = (xce_name_val_t*)calloc(1, sizeof(*nv));
nv->name = strdup("slots");
nv->val = (xce_val_t*)calloc(1, sizeof(*nv->val));
nv->val->which = Value_u_fvalue;
nv->val->u.fvalue = 22.0;
wire_w = (xce_worker_t*)calloc(1, sizeof(*wire_w));
wire_w->name = strdup("hostA");
wire_w->nresvals_to_sched = 1;
wire_w->resvals_to_sched = (xce_name_val_t**)calloc(1, sizeof(nv));
wire_w->resvals_to_sched[0] = nv;
w_ptrs = (xce_worker_t**)calloc(1, sizeof(*w_ptrs));
w_ptrs[0] = wire_w;
/* Resource definition first, so the core resolves "slots" and actually
* adopts (moves) the worker's value -- the path that emptied the
* recorded payload. */
rr = (xce_sched_update_request_t*)calloc(1, sizeof(*rr));
rr->op = XCE_SCHED_OP_UPDATE_RESOURCES;
rr->u.resourceupdate.nresources = 1;
rr->u.resourceupdate.resources = res_ptrs;
dataq_enq(sched.core.pend_updates[0], &(rr->link));
rw = (xce_sched_update_request_t*)calloc(1, sizeof(*rw));
rw->op = XCE_SCHED_OP_UPDATE_WORKER;
rw->u.workerupdate.nworkers = 1;
rw->u.workerupdate.workers = w_ptrs;
dataq_enq(sched.core.pend_updates[0], &(rw->link));
rc = xce_sched_driver_translate_pending_events(&(sched.core), &sched);
TEST_ASSERT_EQUAL_INT(2, rc);
/* Precondition: the live core did adopt the value. */
{
xce_sched_worker_t* w = sched_get_worker(&(sched.core), "hostA");
xce_resource_t* r = NULL;
TEST_ASSERT_NOT_NULL(w);
TEST_ASSERT_TRUE(key_find(sched.core.resmgr, "slots", (void**)&r) > 0);
TEST_ASSERT_NOT_NULL(zarray_get(&(w->resvals), r->idx));
}
rc = sched_record_stop(&sched);
TEST_ASSERT_EQUAL_INT(0, rc);
f = fopen(path_copy, "r");
TEST_ASSERT_NOT_NULL(f);
skip_header_and_initial_snapshot(f);
while (capn_init_fp(&c, f, 0) == 0) {
ReplayEvent_ptr p;
struct ReplayEvent wire;
replay_event_t dst;
p.p = capn_getp(capn_root(&c), 0, 1);
read_ReplayEvent(&wire, p);
memset(&dst, 0, sizeof(dst));
decode_ReplayEvent(&dst, &wire);
if (dst.body_which == ReplayEvent_body_workersPush) {
xce_worker_t* rw0;
xce_name_val_t* rnv;
found = 1;
TEST_ASSERT_NOT_NULL(dst.body.workers_push);
TEST_ASSERT_EQUAL_INT(1, dst.body.workers_push->count);
rw0 = dst.body.workers_push->workers[0];
TEST_ASSERT_EQUAL_STRING("hostA", rw0->name);
TEST_ASSERT_EQUAL_INT(1, rw0->nresvals_to_sched);
rnv = rw0->resvals_to_sched[0];
TEST_ASSERT_EQUAL_STRING("slots", rnv->name);
TEST_ASSERT_NOT_NULL_MESSAGE(rnv->val,
"the recorded WORKERS_PUSH must keep the resval's value, "
"not the NULL the core's MOVE_PTR left behind (bug-1852)");
TEST_ASSERT_EQUAL_FLOAT(22.0f, (float)rnv->val->u.fvalue);
}
free_ReplayEvent(&dst);
capn_free(&c);
}
fclose(f);
TEST_ASSERT_TRUE_MESSAGE(found, "a WORKERS_PUSH event must be recorded");
printf("test_recorded_workers_push_keeps_resval_values: PASS\n");
}
/* ---------------------------------------------------------------------- */
/* Test 3: when recording is inactive, the driver-loop hooks touch no */
/* file/record state -- zero file I/O (Req 5 AC 10, "Performance"). */
@ -959,7 +1088,10 @@ static void submit_one_jobstate_event(xce_scheduler_t* sched, uint64_t jobid)
ev.type = XCE_CORE_EVENT_JOB_STATE_CHANGE;
ev.u.job_state_change.jobid = jobid;
/* Same two steps driver.c takes per batch (bug-1852): record the
* events, (apply them,) then the periodic checkpoint. */
TEST_ASSERT_EQUAL_INT(0, sched_record_write_events(sched, &ev, 1));
TEST_ASSERT_EQUAL_INT(0, sched_record_checkpoint_after_apply(sched));
}
/* Reads exactly one capnp message off `f` and decodes it as a ReplayEvent,
@ -1171,6 +1303,7 @@ test_periodic_checkpoint_single_batch_crossing_two_boundaries_checkpoints_once(
events[i].u.job_state_change.jobid = (uint64_t)(300 + i);
}
TEST_ASSERT_EQUAL_INT(0, sched_record_write_events(&sched, events, 6));
TEST_ASSERT_EQUAL_INT(0, sched_record_checkpoint_after_apply(&sched));
TEST_ASSERT_EQUAL_UINT64(6ULL, sched.record_event_count);
TEST_ASSERT_TRUE(sched.recording);
@ -1226,9 +1359,11 @@ test_periodic_checkpoint_write_failure_disables_recording_cleanly(void)
ev.type = XCE_CORE_EVENT_JOB_STATE_CHANGE;
ev.u.job_state_change.jobid = 400ULL;
/* The event itself writes fine; the periodic checkpoint that fires
* immediately after it (interval=1) is what fails and aborts. */
TEST_ASSERT_TRUE(sched_record_write_events(&sched, &ev, 1) < 0);
/* The event itself writes fine; the periodic checkpoint that follows
* it (interval=1, bug-1852's separate after-apply step) is what fails
* and aborts. */
TEST_ASSERT_EQUAL_INT(0, sched_record_write_events(&sched, &ev, 1));
TEST_ASSERT_TRUE(sched_record_checkpoint_after_apply(&sched) < 0);
TEST_ASSERT_FALSE(sched.recording);
TEST_ASSERT_NULL(sched.record_file);
@ -1406,6 +1541,7 @@ int main(void)
RUN_TEST(test_start_writes_decodable_header_then_snapshot_entry);
RUN_TEST(test_full_cycle_start_event_session_stop_produces_expected_stream);
RUN_TEST(test_recorded_workers_push_keeps_resval_values);
RUN_TEST(test_recording_inactive_hooks_are_zero_cost_noops);
RUN_TEST(test_double_start_fails_cleanly);
RUN_TEST(test_stop_when_not_recording_fails_cleanly);