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

feat(scheduler-sim-decision-parity): reconstruct job allocation with safe ownership (task 2.3)

Browse source 
Extends sim_reconcile_unaccounted_jobs()'s FOUND branch: takes ownership of
a matched recorded DISPATCH decision's allocation (job->allocs = entry's
alloc, not a copy -- mirrors scheduler.c's sched_commit_decision()
precedent exactly), builds job->accounts via make_job_account(), then
safely consumes the recorded entry -- nulling the source alloc pointer
BEFORE freeing the entry (preventing a double-free through job->allocs),
then compacting decoded_session->decisions via shift-down (memmove, never
swap-with-last, which would reorder the remaining entries and break the
positional --sim-verify comparison a later task depends on).

Fixed a real correctness bug found while implementing: the "not found"
check (i == n_decisions) goes stale once n_decisions can be decremented
mid-loop by compaction -- replaced with an explicit found flag.

Primary correctness bar: reconcile-then-release-through-normal-job-cleanup
is ASAN-clean. Verified with ASAN_OPTIONS=detect_leaks=1 (beyond this
suite's registered detect_leaks=0): zero double-free/UAF/overflow. Found
and logged (not fixed, out of boundary) a real pre-existing leak in
src/lib/misc.c's free_host_account() -- equally present on the live
dispatch path, bug-1621.

46/47 scheduler tests pass; the 1 failure is pre-existing bug-1560
(unrelated).

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
This commit is contained in:
Rongsong Shen 2026-08-22 08:36:03 +00:00
parent c38207458f
commit eccba84112
No known key found for this signature in database
GPG key ID: 5BCDAAE0123E9FE2
3 changed files with 306 additions and 26 deletions
Download patch file Download diff file Expand all files Collapse all files

118
src/scheduler/src/sim_loop.c
View file

@ -11,6 +11,9 @@
#include "repl.h"
#include "sim.h"
#include "utils.h"
#include "xcealloc.h" /* task 2.3: make_job_account() -- Recorded-Decision
* Reconciliation's FOUND branch, sim_reconcile_
* unaccounted_jobs() below */
/*
* sim_loop.c - simulation-mode driver event loop (scheduler-core spec, task
@ -488,27 +491,25 @@ static void* sim_picker_thread_fn(void* arg)
* sim_reconcile_unaccounted_jobs - Recorded-Decision Reconciliation
* (scheduler-sim-decision-parity spec, design.md's "Recorded-Decision
* Reconciliation" component; tasks 2.2-2.4). THIS function, as it stands
* right now, implements only task 2.2's slice: the once-per-round trigger,
* the jobid-based lookup, and the "not found" path (Requirement 4.1, 4.2).
* Tasks 2.3 (allocation reconstruction, ownership transfer, array
* compaction) and 2.4 (feeding the reconstructed decision through the
* decision-reporting channel) build on top of the FOUND branch below,
* deliberately left as a no-op marker for now -- see the comment inside
* that branch. Not yet called from anywhere (task 3 wires this into
* right now, implements tasks 2.2 AND 2.3: the once-per-round trigger, the
* jobid-based lookup, the "not found" path (Requirement 4.1, 4.2), and (task
* 2.3) the FOUND path's allocation reconstruction, ownership transfer, and
* recorded-array compaction (Requirement 1.1, 1.2, 3.2). Task 2.4 (feeding
* the reconstructed decision through the decision-reporting channel) is
* deliberately NOT built here yet -- see the comment at the end of the FOUND
* branch below. Not yet called from anywhere (task 3 wires this into
* sched_sim_driver_run()'s round-boundary sequence, before
* sched_account_jobs() -- see that call site's own comment above) --
* standalone and directly unit-testable for now, per design.md's own
* Testing Strategy ("Recorded-Decision Reconciliation leaves a job
* unaccounted, without fabricating data or erroring, when no matching
* recorded decision exists").
* Testing Strategy.
*
* Trigger condition: mirrors sched_account_jobs()'s own `job->accounts ==
* NULL` guard (scheduler.c) -- reused here as both the trigger AND (once
* tasks 2.3/2.4 populate job->accounts) the double-application guard, per
* design.md's Implementation Notes. A RUN-state job in `sim->core.jobs`
* with `job->accounts == NULL` is the signature of a job already resolved
* earlier in the original recording, before this replay run's own live
* decision-making ever considered it.
* NULL` guard (scheduler.c) -- reused here as both the trigger AND (now that
* the FOUND branch below populates job->accounts) the double-application
* guard, per design.md's Implementation Notes. A RUN-state job in
* `sim->core.jobs` with `job->accounts == NULL` is the signature of a job
* already resolved earlier in the original recording, before this replay
* run's own live decision-making ever considered it.
*
* For each such job, searches `decoded_session->decisions` for a matching
* DISPATCH entry by jobid -- a job can only reach RUN-state via a recorded
@ -517,6 +518,19 @@ static void* sim_picker_thread_fn(void* arg)
* Implementation Notes for why PEND/PREEMPT are structurally impossible to
* legitimately match here).
*
* If a match IS found (task 2.3): `job->allocs` takes ownership of the
* matched entry's `xce_job_alloc_t*` (a pointer TAKEOVER, not a deep copy --
* the same pattern scheduler.c's sched_commit_decision() already uses for a
* freshly-drained DISPATCH decision), `job->accounts` is built from it via
* make_job_account(), the source field is nulled out BEFORE the entry is
* freed (avoids a dangling job->allocs / later double-free -- see the
* FOUND branch's own comment for the full ownership-hazard reasoning), and
* the now-fully-consumed entry is removed from `decoded_session->decisions`
* by shift-down (memmove), not swap-with-last, keeping the remaining
* entries in their original relative order. `job->accounts != NULL`
* afterward permanently prevents this job from being reconsidered here on a
* later round.
*
* If no match is found: `job->accounts`/`job->allocs` are left exactly as
* they were (still NULL) -- not an error, not fabricated. Nothing here
* needs to separately record "usage unavailable": leaving `job->accounts ==
@ -549,6 +563,7 @@ static void sim_reconcile_unaccounted_jobs(xce_sched_sim_driver_t* sim,
for (;;) {
xce_sched_job_t* job = NULL;
int i;
int found = 0;
rc = iter_next(&it, (void**)&job);
if ((rc < 0) || (rc == RTTI_ITER_NO_DATA)) {
@ -562,6 +577,7 @@ static void sim_reconcile_unaccounted_jobs(xce_sched_sim_driver_t* sim,
for (i = 0; i < decoded_session->n_decisions; i++) {
replay_decision_t* entry = decoded_session->decisions[i];
xce_job_alloc_t* alloc;
if ((entry == NULL)
|| (entry->body_which != ReplayDecision_body_dispatch)
@ -571,18 +587,70 @@ static void sim_reconcile_unaccounted_jobs(xce_sched_sim_driver_t* sim,
continue;
}
/* FOUND -- tasks 2.3/2.4 (NOT this task) reconstruct
* job->allocs/job->accounts from `entry` here (move-then-null-
* then-free ownership transfer, then array compaction) and feed
* the result through the decision-reporting channel. Task 2.2
* does nothing further for a found match: the job simply
* remains job->accounts == NULL for now, which
* sched_account_jobs() already tolerates safely (same skip path
* as the "not found" case below). */
/* FOUND -- task 2.3: reconstruct job->allocs/job->accounts by
* TAKING OVER `entry`'s allocation, the same ownership-transfer
* pattern scheduler.c's sched_commit_decision() already uses
* for a freshly-drained DISPATCH decision (job->allocs =
* decision->u.alloc; job->accounts =
* make_job_account(job->allocs);) -- not a duplicate copy.
*
* Ownership hazard (design.md's Implementation Notes): `alloc`
* is now aliased by BOTH job->allocs and `entry`, which is
* about to be freed below. Null out the source field
* IMMEDIATELY, before free_ReplayDecision(entry) runs --
* free_ReplayDecision() -> free_DispatchDecision() ->
* free_JobAlloc_ptr(&entry->body.dispatch->alloc) would
* otherwise free this exact memory out from under job->allocs
* (dangling pointer now, double-free later when the job's own
* allocation is released through its normal end-of-life path).
* Every `_ptr`-style free helper in this codebase (including
* free_JobAlloc_ptr) already no-ops on NULL, so nulling here
* makes freeing the rest of the entry safe automatically. */
alloc = entry->body.dispatch->alloc;
job->allocs = alloc;
entry->body.dispatch->alloc = NULL;
job->accounts = make_job_account(job->allocs);
found = 1;
/* Memory ownership (design.md's Implementation Notes): this
* entry is now fully consumed -- its `alloc` payload was taken
* over above (now NULL here), so freeing it is safe. Simply
* decrementing n_decisions WITHOUT this free would leak the
* rest of the entry's payload every round a reconciliation
* happens, since the round-end free_ReplaySession_ptr() ->
* free_ReplayDecision_list() walk only ever reaches the first
* n_decisions entries. */
free_ReplayDecision(entry);
free(entry);
/* Compact by SHIFT-DOWN (memmove), NOT swap-with-last:
* swap-with-last would relocate the last entry into this slot
* and change its position relative to the other remaining
* entries, corrupting the relative ordering the (not yet
* wired -- task 3) --sim-verify positional walk over this same
* decoded_session depends on. This round's decoded_session is
* decoded fresh every round, so this compaction has no
* cross-round effect. */
memmove(&(decoded_session->decisions[i]),
&(decoded_session->decisions[i + 1]),
(size_t)(decoded_session->n_decisions - i - 1)
* sizeof(*(decoded_session->decisions)));
decoded_session->n_decisions--;
/* Task 2.4 (NOT this task) feeds this reconstructed decision
* through the same decision-reporting channel a freshly-drained
* decision uses. This task's job ends here: job->allocs/
* job->accounts are now correctly reconstructed and safe to
* release later, and job->accounts != NULL now permanently
* prevents this job from being reconsidered by this mechanism
* again (Requirement 3.3's double-application guard, reused as
* this task's own "never reconsidered on a later round"
* guarantee). */
break;
}
if (i == decoded_session->n_decisions) {
if (!found) {
char jobidStr[64];
/* Requirement 4.1/4.2: no matching recorded decision this round

209
src/scheduler/testsuite/test_sim_reconcile_unaccounted.c
View file

@ -266,6 +266,42 @@ static replay_session_t* make_decoded_session(uint64_t begin_round,
return s;
}
/* task 2.3 fixture helper: a real, one-host/one-resource xce_job_alloc_t --
* same field shape sim_verify_job_alloc_eq() already validates (design.md's
* Allowed Dependencies), built via the real xcealloc.c production helpers
* (make_alloc/make_allocs/make_host_alloc/make_host_allocs/make_job_alloc),
* not a hand-rolled struct literal -- so a reconciled job's `job->accounts`
* shape can be checked against real, non-trivial resource values. */
static xce_job_alloc_t* make_test_job_alloc_with_host(uint64_t jobid,
const char* host,
const char* resname,
float value)
{
xce_alloc_t* a = make_alloc((char*)resname, NULL, 0, value, 0);
xce_alloc_t** allocs = make_allocs(1, a);
xce_host_alloc_t* h = make_host_alloc((char*)host, 1, allocs);
xce_host_alloc_t** hallocs = make_host_allocs(1, h);
return make_job_alloc(jobid, 1, hallocs);
}
/* task 2.3 fixture helper: wraps an already-built xce_job_alloc_t (as
* opposed to make_dispatch_entry() above, which builds its own empty
* (nhosts=0) alloc) into a DISPATCH-shaped replay_decision_t. */
static replay_decision_t* make_dispatch_entry_from_alloc(xce_job_alloc_t* alloc)
{
sched_dispatch_decision_t* dispatch
= (sched_dispatch_decision_t*)calloc(1, sizeof(*dispatch));
replay_decision_t* rd = (replay_decision_t*)calloc(1, sizeof(*rd));
dispatch->alloc = alloc;
rd->body_which = ReplayDecision_body_dispatch;
rd->body.dispatch = dispatch;
return rd;
}
/* ---------------------------------------------------------------------- */
/* Test 1 (the task's own primary Observable): a RUN-state job with no */
/* known allocation and no matching recorded decision in this round's */
@ -446,6 +482,175 @@ static void test_reconcile_skips_non_run_state_job(void)
printf("test_reconcile_skips_non_run_state_job: PASS\n");
}
/* ---------------------------------------------------------------------- */
/* Test 7 (task 2.3): a matching recorded DISPATCH entry is reconstructed */
/* into job->allocs (ownership TAKEOVER -- same pointer, not a copy) and */
/* job->accounts (built from it), and the consumed entry is removed from */
/* decoded_session (Requirement 1.1, 1.2, 3.2). */
/* ---------------------------------------------------------------------- */
static void
test_reconcile_reconstructs_allocs_and_accounts_from_matching_dispatch(void)
{
xce_sched_sim_driver_t sim;
xce_sched_job_t* job;
xce_job_alloc_t* jalloc;
replay_session_t* decoded;
fixture_init(&sim);
job = add_job(&(sim.core), 7010ULL, JOB_STAT_RUN, NULL);
TEST_ASSERT_NULL(job->accounts);
TEST_ASSERT_NULL(job->allocs);
jalloc = make_test_job_alloc_with_host(7010ULL, "worker1", "slots", 2.0f);
decoded
= make_decoded_session(1000ULL, make_dispatch_entry_from_alloc(jalloc));
sim_reconcile_unaccounted_jobs(&sim, decoded);
TEST_ASSERT_EQUAL_PTR_MESSAGE(
jalloc, job->allocs,
"job->allocs must take OWNERSHIP of the recorded entry's alloc "
"(identical pointer), not a duplicate copy -- same pattern "
"scheduler.c's sched_commit_decision() uses");
TEST_ASSERT_NOT_NULL_MESSAGE(
job->accounts,
"job->accounts must be built (make_job_account()) from the "
"reconstructed job->allocs");
TEST_ASSERT_EQUAL_INT(1, job->accounts->nhosts);
TEST_ASSERT_EQUAL_STRING("worker1", job->accounts->accounts[0]->host);
TEST_ASSERT_EQUAL_FLOAT(2.0f,
job->accounts->accounts[0]->accounts[0]->reserved);
TEST_ASSERT_EQUAL_INT_MESSAGE(
0, decoded->n_decisions,
"the now-fully-consumed recorded entry must be removed from "
"decoded_session->decisions");
printf("test_reconcile_reconstructs_allocs_and_accounts_from_matching_"
"dispatch: PASS\n");
}
/* ---------------------------------------------------------------------- */
/* Test 8 (task 2.3): compaction is SHIFT-DOWN, not swap-with-last -- the */
/* remaining recorded entries must keep their ORIGINAL RELATIVE ORDER, per */
/* design.md's Implementation Notes (a later --sim-verify positional walk */
/* depends on this). A 4-entry fixture with the match in the middle */
/* distinguishes the two strategies: swap-with-last would relocate the */
/* last entry (C) into the removed slot, producing [A, C, B] instead of the */
/* correct [A, B, C]. */
/* ---------------------------------------------------------------------- */
static void test_reconcile_compacts_via_shift_down_preserving_order(void)
{
xce_sched_sim_driver_t sim;
xce_sched_job_t* job;
replay_session_t* decoded;
fixture_init(&sim);
job = add_job(&(sim.core), 7011ULL, JOB_STAT_RUN, NULL);
decoded = (replay_session_t*)calloc(1, sizeof(*decoded));
decoded->begin_round = 1000ULL;
decoded->summary = NULL;
decoded->decisions
= (replay_decision_t**)calloc(4, sizeof(*(decoded->decisions)));
decoded->decisions[0] = make_pend_entry(9001ULL); /* A */
decoded->decisions[1] = make_dispatch_entry(7011ULL); /* matched */
decoded->decisions[2] = make_pend_entry(9002ULL); /* B */
decoded->decisions[3] = make_pend_entry(9003ULL); /* C */
decoded->n_decisions = 4;
sim_reconcile_unaccounted_jobs(&sim, decoded);
TEST_ASSERT_NOT_NULL(job->accounts);
TEST_ASSERT_EQUAL_INT_MESSAGE(
3, decoded->n_decisions,
"exactly one entry (the matched DISPATCH) must be consumed");
TEST_ASSERT_EQUAL_UINT64_MESSAGE(9001ULL,
decoded->decisions[0]->body.pend->jobid,
"entry A must stay first");
TEST_ASSERT_EQUAL_UINT64_MESSAGE(
9002ULL, decoded->decisions[1]->body.pend->jobid,
"entry B must be second (shift-down) -- swap-with-last would put C "
"here instead");
TEST_ASSERT_EQUAL_UINT64_MESSAGE(9003ULL,
decoded->decisions[2]->body.pend->jobid,
"entry C must be third, untouched");
printf("test_reconcile_compacts_via_shift_down_preserving_order: "
"PASS\n");
}
/* ---------------------------------------------------------------------- */
/* Test 9 (task 2.3, the task's own PRIMARY correctness bar): reconciling a */
/* job's allocation from a recorded DISPATCH entry, then releasing that */
/* job's allocation through its NORMAL end-of-life path -- */
/* XCE_CORE_EVENT_JOB_STATE_CHANGE (core_events.c) -- is clean: no */
/* double-free, no use-after-free. This directly exercises the */
/* move-then-null-then-free ownership handling design.md's Implementation */
/* Notes require: `entry->body.dispatch->alloc` must be nulled out BEFORE */
/* free_ReplayDecision(entry) frees the rest of the entry, or that free */
/* would free the exact memory job->allocs now owns, and this exact */
/* end-of-life release would then double-free it. Meaningful evidence for */
/* this only under the project's ASAN build (see meson.build/tasks.md), */
/* not as a plain functional assertion. */
/* ---------------------------------------------------------------------- */
static void test_reconcile_then_release_via_job_state_change_is_asan_clean(void)
{
xce_sched_sim_driver_t sim;
xce_sched_job_t* job;
replay_session_t* decoded;
xce_core_event_t ev;
int push_rc;
fixture_init(&sim);
job = add_job(&(sim.core), 7012ULL, JOB_STAT_RUN, NULL);
decoded = make_decoded_session(
1000ULL, make_dispatch_entry_from_alloc(make_test_job_alloc_with_host(
7012ULL, "worker1", "slots", 4.0f)));
sim_reconcile_unaccounted_jobs(&sim, decoded);
TEST_ASSERT_NOT_NULL(job->accounts);
TEST_ASSERT_NOT_NULL(job->allocs);
TEST_ASSERT_EQUAL_INT(0, decoded->n_decisions);
memset(&ev, 0, sizeof(ev));
ev.type = XCE_CORE_EVENT_JOB_STATE_CHANGE;
ev.u.job_state_change.jobid = 7012ULL;
ev.u.job_state_change.old_state = JOB_STAT_RUN;
ev.u.job_state_change.new_state = JOB_STAT_DONE;
push_rc = xce_core_push_events(&(sim.core), &ev, 1);
TEST_ASSERT_EQUAL_INT(0, push_rc);
TEST_ASSERT_NULL_MESSAGE(
job->accounts, "released cleanly via the normal end-of-life path");
TEST_ASSERT_NULL_MESSAGE(
job->allocs, "released cleanly via the normal end-of-life path");
/* Also exercise decoded_session's own round-end cleanup path
* (free_ReplaySession_ptr(), the same call sched_sim_driver_run() makes
* every round) now that n_decisions correctly reflects only the
* remaining, still-owned entries (here: none) -- ASAN-clean under a
* leak-detecting run (this suite's registered meson test uses
* detect_leaks=0 per its own established fixture-teardown convention;
* this call is exercised here regardless, and was additionally verified
* with ASAN_OPTIONS=detect_leaks=1 against the standalone binary as
* part of this task's own validation). */
free_ReplaySession_ptr(&decoded);
printf("test_reconcile_then_release_via_job_state_change_is_asan_clean: "
"PASS\n");
}
/* ---------------------------------------------------------------------- */
/* Test 6: NULL `sim`/`decoded_session` is a safe no-op (design.md's */
/* Service Interface preconditions -- "never load-bearing" convention). */
@ -478,6 +683,10 @@ int main(void)
RUN_TEST(test_reconcile_skips_job_that_already_has_accounts);
RUN_TEST(test_reconcile_ignores_non_dispatch_entry_with_matching_jobid);
RUN_TEST(test_reconcile_skips_non_run_state_job);
RUN_TEST(
test_reconcile_reconstructs_allocs_and_accounts_from_matching_dispatch);
RUN_TEST(test_reconcile_compacts_via_shift_down_preserving_order);
RUN_TEST(test_reconcile_then_release_via_job_state_change_is_asan_clean);
RUN_TEST(test_reconcile_null_args_are_a_safe_noop);
return UNITY_END();
}