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

feat(scheduler-sim-decision-parity): wire reconciliation into the round boundary (task 3)

Browse source 
This is the integration point that makes the whole feature real
end-to-end. Calls sim_reconcile_unaccounted_jobs() (tasks 2.2-2.4) at
exactly the position design.md's mermaid diagram requires: after
xce_core_begin_session()'s reset, strictly before sched_account_jobs()
(so the same round's re-charge picks up every newly-reconstructed job,
Req 1.3/1.4) and before xce_core_signal_go() unblocks the live picker (so
still-pending jobs' capacity checks this round see correct resInUse, Req
2.1/2.2). decoded_session and record_ctx are the same per-round locals
sched_sim_driver_run() already declares -- no new plumbing.

sim_verify.c needs zero changes: reconciliation consumes/compacts
decoded_session->decisions before the verify drain loop runs each round,
which is what keeps the existing positional comparison correct (Req 5.2)
-- confirmed by extending test_sim_verify.c with a no-false-mismatch test
against a recording containing already-resolved jobs.

New end-to-end test proves the ordering matters for real: a reconciled
job's resInUse is charged before a live sched_check_host_satisfy() call
for a different job in the same round correctly denies it (5.0 wanted vs
2.0 remaining) and grants an exact-fit request (2.0) -- using the real
scheduler.c/resource.c accounting and capacity-check code, not stubs.

Found and fixed a test-authoring bug (get_resvals() returns nres on
success, not 0) during GREEN verification -- logged as bug-1622, a test
bug, not production.

47/48 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 10:00:06 +00:00
parent 75edb7a329
commit 1264409478
No known key found for this signature in database
GPG key ID: 5BCDAAE0123E9FE2
5 changed files with 947 additions and 11 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -497,11 +497,10 @@ static void* sim_picker_thread_fn(void* arg)
* (Requirement 1.1, 1.2, 3.2), and (task 2.4) feeding the reconstructed
* decision through the exact same decision-reporting channel a
* freshly-drained decision uses -- see the end of the FOUND branch below
* (Requirement 3.1, 3.2). 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.
* (Requirement 3.1, 3.2). Task 3 wires this into sched_sim_driver_run()'s
* round-boundary sequence, before sched_account_jobs() -- see that call
* site's own comment below. Still directly unit-testable standalone too
* (test_sim_reconcile_unaccounted.c), per design.md's own Testing Strategy.
*
* Trigger condition: mirrors sched_account_jobs()'s own `job->accounts ==
* NULL` guard (scheduler.c) -- reused here as both the trigger AND (now that
@ -923,11 +922,31 @@ int sched_sim_driver_run(xce_sched_sim_driver_t* sim,
continue;
}
/* Recorded-Decision Reconciliation (sim_reconcile_unaccounted_
* jobs(), tasks 2.2-2.4 -- implemented above but not yet wired
* in here; task 3 wires this call site) belongs here, before
* sched_account_jobs() below -- see design.md's "Recorded-
* Decision Reconciliation" component. */
/* Task 3 (scheduler-sim-decision-parity): wires Recorded-
* Decision Reconciliation (sim_reconcile_unaccounted_jobs(),
* tasks 2.2-2.4) into the round boundary, strictly between the
* session-start reset above and the re-accounting call below --
* see design.md's "Recorded-Decision Reconciliation" component
* and its "Integration with --sim-verify" Implementation Note.
* Ordering matters for two reasons: (1) any job reconciled THIS
* round must already have job->accounts set before
* sched_account_jobs() runs, so the same round's re-charge
* picks up every newly-reconstructed job (Requirement 1.3,
* 1.4); (2) reconciliation must finish before
* xce_core_signal_go() unblocks the live picker/allocator, so
* still-pending jobs' capacity checks this round see correct
* resInUse/sharedInUse (Requirement 2.1, 2.2). `decoded_session`
* and `record_ctx` are the same per-round locals already
* declared above in this function's ReplaySession branch --
* design.md's Implementation Notes on the reconciliation
* function's own (task 2.4-added) third parameter confirms
* record_ctx has no other home to come from. Consuming and
* compacting decoded_session->decisions here, before the
* verify drain loop below ever runs, is what keeps that loop's
* existing positional comparison correct with zero changes to
* sim_verify.c (Requirement 5.2). */
sim_reconcile_unaccounted_jobs(sim, decoded_session, record_ctx);
sched_account_jobs(&(sim->core));
xce_core_signal_go(&(sim->core));

21
src/scheduler/testsuite/meson.build
View file

@ -792,6 +792,24 @@ sched_sim_resource_contention_test = executable(
dependencies: [scheduler_test_deps],
install: false)
# End-to-end integration proof for task 3 (Req 1.3/1.4, 2.1/2.2, 5.2): wires
# Recorded-Decision Reconciliation (tasks 2.2-2.4) into sim_loop.c's real
# round-boundary sequence, before sched_account_jobs()/xce_core_signal_go().
# Unlike every other sim_loop.c-including file in this directory, this one
# links a REAL (not no-op) sched_account_jobs() reimplementation (copied
# verbatim from scheduler.c) plus resource.c, so it can prove the actual
# resInUse/capacity-check outcome of the wired call site against the real,
# unmodified sched_sim_driver_run() -- not just that reconciliation runs in
# isolation (test_sim_reconcile_unaccounted.c) or that sched_account_jobs()
# is called at the right point (test_sim_session_ordering.c, via a spy).
# See this file's own header comment for the full scenario and rationale.
sched_sim_round_boundary_reconciliation_test = executable(
'sched_sim_round_boundary_reconciliation_test',
['test_sim_round_boundary_reconciliation.c'],
include_directories:[incdir],
dependencies: [scheduler_test_deps],
install: false)
# Contract test for the interactive simulation REPL's execution-control slice
# (scheduler-core spec, task 12.1, Req 5 AC 22): starting simulation with
# --interactive pauses before any event is processed, and a `step N` command
@ -1125,6 +1143,9 @@ test('scheduler sim reconcile unaccounted test',
test('scheduler sim resource contention test',
sched_sim_resource_contention_test, suite: 'scheduler',
env: ['ASAN_OPTIONS=detect_leaks=0'])
test('scheduler sim round boundary reconciliation test',
sched_sim_round_boundary_reconciliation_test, suite: 'scheduler',
env: ['ASAN_OPTIONS=detect_leaks=0'])
test('scheduler repl execution control test',
sched_repl_execution_control_test, suite: 'scheduler',
env: ['ASAN_OPTIONS=detect_leaks=0'])

739
src/scheduler/testsuite/test_sim_round_boundary_reconciliation.c Normal file
View file

@ -0,0 +1,739 @@
/*
* test_sim_round_boundary_reconciliation.c - end-to-end integration proof
* for scheduler-sim-decision-parity task 3 (Req 1.3, 1.4, 2.1, 2.2, 5.2):
* Recorded-Decision Reconciliation (sim_reconcile_unaccounted_jobs(),
* tasks 2.2-2.4) wired into sim_loop.c's round-boundary sequence, strictly
* before sched_account_jobs() and xce_core_signal_go().
*
* Why this file exists, distinct from the tests that already cover the
* pieces: test_sim_reconcile_unaccounted.c (tasks 2.2-2.4) already proves
* sim_reconcile_unaccounted_jobs() itself is correct in isolation, calling
* it directly against hand-built fixtures. test_sim_session_ordering.c
* (task 2.1) already proves sched_account_jobs() is called once per round
* inside the session-active window, via a spy. test_sim_resource_
* contention.c (task 1.2) proves a resource-contention `.replay` fixture's
* shape, with sched_account_jobs() stubbed to a no-op (it deliberately does
* not link scheduler.c/resource.c). NONE of these exercises the actual
* WIRED call site inside the real, unmodified sched_sim_driver_run() with
* the REAL accounting/capacity-check code linked in -- which is exactly
* what task 3's own Observable Completion requires: "a job whose
* allocation was reconciled this round is reflected in resource-in-use
* totals before the live picker/allocator evaluates capacity for any
* other job in that same round." This file drives sched_sim_driver_run()
* for real, against a hand-crafted `.replay` file (same technique
* test_sim_verify.c's own header comment documents: sched_record_start()
* for a real header+snapshot prefix, then hand-constructed ReplayEvent/
* ReplaySession messages appended via the real encode_ReplayEvent_ptr()/
* encode_ReplaySession_ptr()/write_capnp_message() codec), then inspects
* the REAL post-replay xce_sched_core_t/xce_sched_worker_t state directly.
*
* Scenario: worker "worker-a" has 10.0 total units of one resource
* ("slots"). Job A (9301) is pushed PEND, then flipped to RUN by a
* JOB_STATE_CHANGE event in the SAME round's event batch -- the signature
* of a job already resolved before this replay's own live picker ever
* reaches it (design.md's Recorded-Decision Reconciliation trigger
* condition: RUN-state, job->accounts == NULL). This round's recorded
* ReplaySession carries a matching DISPATCH decision for job A (8.0
* units) plus a genuine PEND decision for job B (9302, still PEND this
* round -- the real picker independently re-pends it fresh, needing no
* reconciliation). After replay: (a) worker-a's resInUse must show job
* A's 8.0-unit charge (Req 1.3/1.4) -- proving reconciliation's real
* make_job_account()/sched_charge_account() round trip, INCLUDING the
* lazy resname->residx resolution sched_update_res_accounts() performs
* (make_res_account(), xcealloc.c, always produces residx == -1); (b)
* sched_check_host_satisfy() (resource.c, unmodified) against this SAME
* post-round worker state must now correctly DENY a hypothetical job
* wanting more than the 2.0-unit remaining headroom, and GRANT one wanting
* exactly 2.0 -- proving Requirement 2's capacity-awareness is a genuine
* property of the now-correct resInUse, not merely "resInUse is some
* nonzero number."
*
* Because reconciliation's real make_job_account()/make_res_account()
* path is exercised (unlike test_sim_resource_contention.c's own stand-in
* allocator, which builds xce_job_account_t by hand with residx already
* resolved, deliberately bypassing this), this fixture pre-registers
* "slots" in the core's resmgr via the real makeResource() and reads back
* its assigned ->idx, rather than assuming which index the lazy
* add_vres() auto-registration path (resource.c) would land on -- see
* fixture_init()'s own comment below.
*
* White-box #include set: the same combined core/sim/recording/repl/
* snapshot stack test_sim_resource_contention.c's own header comment
* documents, PLUS a REAL (not no-op) sched_account_jobs() reimplementation
* below -- copied verbatim from scheduler.c:1258 (body unchanged, this
* spec's own task 1.1 already relocated it to take xce_sched_core_t*
* directly), following the SAME "real reimplementation of a scheduler.c/
* schedobj.c function, to avoid pulling in the whole file's heavier
* dependencies" convention this whole test-file family already uses for
* sched_resolve_job_picker(). This is the one thing that MUST differ from
* every other sibling file that #includes sim_loop.c: this file's whole
* purpose is to prove what sched_account_jobs() actually does to
* resInUse, so stubbing it as a no-op (as every other sim_loop.c-including
* file in this directory does) would defeat the test.
*/
#include <errno.h>
#include <limits.h>
#include <stdint.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>
#include <unity/unity.h>
#include "controller.capnp.h"
#include "kvmgr.h"
#include "preason.h"
#include "resobj.h"
#include "scheduler.h"
#include "xstring.h"
/* Pull in the real functions under test into this TU (white-box #include,
* same combined set test_sim_resource_contention.c's own header comment
* documents -- driver.c/recording_events.c are deliberately NOT included
* here: unlike that file, this one never drives a live normal-mode round
* (no dataq/pend_updates machinery needed) -- it builds its `.replay` file
* by hand, the same technique test_sim_verify.c's own header comment
* documents, so it needs only recording.c/recording_session.c for the
* header+snapshot prefix). */
#include "../src/core.c"
#include "../src/core_clock.c"
#include "../src/core_events.c"
#include "../src/core_session.c"
#include "../src/decision.c"
#include "../src/plugin.c"
#include "../src/plugins/builtin_fcfs.c"
#include "../src/recording.c"
#include "../src/recording_session.c"
#include "../src/repl.c"
#include "../src/repl_break.c"
#include "../src/repl_info.c"
#include "../src/repl_params.c"
#include "../src/resource.c"
#include "../src/sim.c"
#include "../src/sim_loop.c"
#include "../src/sim_picker.c" /* bug-1567: sched_sim_driver_run() calls sim_bootstrap_picker() (sim_loop.c-cap sibling) as an extern -- undefined reference at link time otherwise */
#include "../src/sim_record.c"
#include "../src/sim_verify.c"
#include "../src/snapshot.c"
#include "../src/snapshot_load.c"
#include "../src/snapshot_wire.c"
void kill_sched_job(void* arg) { (void)arg; }
void kill_sched_queue(void* arg) { (void)arg; }
void kill_sched_worker(void* arg) { (void)arg; }
void kill_sched_user(void* arg) { (void)arg; }
void kill_sched_project(void* arg) { (void)arg; }
void kill_sched_resval(xce_val_t* val) { (void)val; }
/* resource.c's sched_try_alloc_on_host_()'s own link-time dependency --
* never actually invoked (this file calls sched_check_host_satisfy()
* directly, never sched_try_alloc_on_hosts()/sched_try_alloc_on_host_()
* itself). Same link-satisfying-only stub test_sim_resource_contention.c's
* own resource.c inclusion already establishes needing.
*
* sched_worker_release_host_ctx/sched_fill_limit_requests are deliberately
* NOT stubbed here (resource.c defines both for real, and this file's
* combined include set already provides plugin_prepare_allocation via
* plugin.c) -- same rationale test_sim_resource_contention.c's own header
* comment documents. */
int sched_ctx_get_uncomit_host(xvec_t** v, xce_sched_ctx_t* c, int i)
{
(void)v;
(void)c;
(void)i;
return 0;
}
/* Real reimplementation of schedobj.c's sched_resolve_job_picker -- same
* copy every white-box test in this directory that does not link
* schedobj.c already establishes. */
int sched_resolve_job_picker(xce_sched_core_t* core, xce_sched_plugin_t** out)
{
xce_sched_plugin_t* plugin = NULL;
int rc = 0;
if ((core == NULL) || (out == NULL)) {
return -EINVAL;
}
*out = NULL;
if (core->job_pick_plugin == NULL) {
return -EINVAL;
}
lock(core->plugins, RTTI_LOCK_READ);
rc = key_find(core->plugins, core->job_pick_plugin, (void**)&plugin);
unlock(core->plugins);
if (rc <= 0) {
return (rc < 0) ? rc : -ENOENT;
}
if ((plugin == NULL) || (plugin->picker_iface == NULL)
|| (plugin->picker_iface->next_job == NULL)
|| (plugin->picker_iface->begin_session == NULL)
|| (plugin->picker_iface->end_session == NULL)
|| (plugin->picker_iface->on_dispatch == NULL)
|| (plugin->picker_iface->on_pend == NULL)) {
return -EINVAL;
}
*out = plugin;
return 0;
}
/* Real reimplementation of scheduler.c's sched_account_jobs() (scheduler.c:
* 1258, scheduler-sim-decision-parity task 1.1's relocated xce_sched_core_t*
* signature) -- body copied verbatim, unlike every OTHER sibling test file
* in this directory that #includes sim_loop.c (all of which stub this as a
* no-op, since none of them need real accounting). This file's entire
* purpose is to prove what this function does to resInUse once task 3
* wires reconciliation ahead of it, so a no-op stub would defeat the
* test's own premise. Calls the REAL sched_charge_account() (resource.c,
* genuinely linked into this TU), not a stand-in. */
int sched_account_jobs(xce_sched_core_t* core)
{
iterator_t it;
int rc = 0;
rc = iter_make(&it, core->jobs);
if (rc < 0) {
return rc;
}
for (;;) {
xce_sched_job_t* job = NULL;
rc = iter_next(&it, (void**)&job);
if ((rc < 0) || (rc == RTTI_ITER_NO_DATA)) {
break;
}
if (!JOB_IS_RUN(job->state)) {
continue;
}
if (job->accounts != NULL) {
sched_charge_account(core, job->accounts);
}
}
if (rc == RTTI_ITER_NO_DATA) {
rc = 0;
}
iter_destroy(&it);
return rc;
}
void setUp(void) { return; }
void tearDown(void) { return; }
/* ---------------------------------------------------------------------- */
/* Fixture constants */
/* ---------------------------------------------------------------------- */
#define WORKER_NAME "worker-a"
#define WORKER_CAPACITY 10.0f
#define JOBA_ID 9301ULL /* already resolved -- reconciled this round */
#define JOBB_ID 9302ULL /* genuinely still PEND this round */
#define JOBA_WANT 8.0f
/* ---------------------------------------------------------------------- */
/* Fixture helpers */
/* ---------------------------------------------------------------------- */
static void make_tmp_dir(char* out, size_t out_len, const char* suffix)
{
char tmpl[PATH_MAX];
snprintf(tmpl, sizeof(tmpl), "/tmp/sched_simreconcile%s_XXXXXX", suffix);
TEST_ASSERT_NOT_NULL(mkdtemp(tmpl));
TEST_ASSERT_TRUE(strlen(tmpl) < out_len);
strcpy(out, tmpl);
}
/* Builds a real xce_scheduler_t fixture to drive sched_record_start() (same
* technique test_sim_verify.c's own build_fixture_scheduler() establishes --
* no dataq/pend_updates needed since this file never drives a live
* normal-mode round, only hand-appends ReplayEvent/ReplaySession messages)
* PLUS a real worker "worker-a" with 10.0 finite units of one resource,
* added directly via sched_add_worker() so the worker (with its resvals)
* is captured into sched_record_start()'s own initial snapshot -- the SAME
* "worker added directly, captured by the initial snapshot" technique
* test_sim_resource_contention.c's own dual-use replay test already proves
* works.
*
* Unlike test_sim_resource_contention.c's own fixture (which bypasses
* resmgr name resolution entirely, setting residx directly on hand-built
* xce_job_account_t structures), THIS file's reconciliation path goes
* through the REAL make_job_account()/make_res_account() (xcealloc.c),
* which always produces residx == -1 -- sched_update_res_accounts()
* (resource.c) then resolves it lazily via key_find(core->resmgr,
* resname, ...), auto-registering the resource via add_vres() if not
* already present. Rather than assume which index that lazy path would
* land on, this fixture pre-registers "slots" via the real makeResource()
* and reads back its assigned ->idx, then uses THAT same index for the
* worker's resvals/resInUse setup and for every capacity-check `wanted`
* vector built later in the test -- eliminating any dependency on
* add_vres()'s own index-assignment behavior.
*/
static int fixture_init(xce_scheduler_t* sched, char* record_dir)
{
xce_sched_worker_t* w;
xce_resource_t* res_template;
xce_resource_t* registered;
int rc;
int residx;
float val[1];
memset(sched, 0, sizeof(*sched));
mylock_init(&(sched->record_lock));
rc = xce_sched_core_init(&(sched->core));
TEST_ASSERT_EQUAL_INT(0, rc);
sched->core.job_pick_plugin = (char*)"builtin_fcfs";
sched->record_dir = record_dir;
res_template = (xce_resource_t*)calloc(1, sizeof(*res_template));
TEST_ASSERT_NOT_NULL(res_template);
res_template->name = strdup("slots");
res_template->valtype = XCE_VAL_NUMERIC;
res_template->ordtype = XCE_ORD_DECR;
res_template->flags = XCE_RESF_LIMIT | XCE_RESF_CONSUMABLE
| XCE_RESF_RELEASE | XCE_RESF_DEFAULT_HOST;
registered = makeResource(sched->core.resmgr, res_template);
TEST_ASSERT_NOT_NULL(registered);
residx = registered->idx;
w = (xce_sched_worker_t*)calloc(1, sizeof(*w));
TEST_ASSERT_NOT_NULL(w);
w->name = strdup(WORKER_NAME);
TEST_ASSERT_NOT_NULL(w->name);
rc = zarray_init(&(w->resvals), 8, NULL);
TEST_ASSERT_EQUAL_INT(0, rc);
rc = zarray_init(&(w->resInUse), 8, NULL);
TEST_ASSERT_EQUAL_INT(0, rc);
val[0] = WORKER_CAPACITY;
rc = put_resvals(&(w->resvals), 1, &residx, val);
TEST_ASSERT_TRUE(rc >= 0);
rc = sched_add_worker(&(sched->core), w);
TEST_ASSERT_TRUE(rc >= 0);
return residx;
}
/* Appends one ReplayEvent{queueUpdate} message -- same technique
* test_sim_verify.c's own append_queue_update_event() establishes. */
static void append_queue_update_event(FILE* f, uint64_t ts,
const char* queue_name)
{
xce_queue_t queue;
replay_event_t src;
struct capn c;
struct capn_segment* cs;
ReplayEvent_ptr p;
memset(&queue, 0, sizeof(queue));
queue.queue = (char*)queue_name;
queue.status = QUEUE_STAT_OPEN;
queue.priority = 1;
memset(&src, 0, sizeof(src));
src.ts = ts;
src.body_which = ReplayEvent_body_queueUpdate;
src.body.queue_update = &queue;
capn_init_malloc(&c);
cs = capn_root(&c).seg;
encode_ReplayEvent_ptr(cs, &p, &src);
capn_setp(capn_root(&c), 0, p.p);
TEST_ASSERT_EQUAL_INT(0, write_capnp_message(f, &c));
capn_free(&c);
}
/* Appends one ReplayEvent{jobsPush} message with exactly one JOB_STAT_PEND
* job -- same technique test_sim_verify.c's own append_jobs_push_event()
* establishes. */
static void append_jobs_push_event(FILE* f, uint64_t ts, uint64_t jobid,
const char* queue_name)
{
job_submit_data_t submit;
xce_job_t job;
xce_job_t* jobs[1];
xce_jobs_data_t jobs_data;
replay_event_t src;
struct capn c;
struct capn_segment* cs;
ReplayEvent_ptr p;
memset(&submit, 0, sizeof(submit));
submit.queue = (char*)queue_name;
submit.user = (char*)"alice";
submit.mintasks = 1;
submit.maxtasks = 1;
memset(&job, 0, sizeof(job));
job.jobid = jobid;
job.state = JOB_STAT_PEND;
job.submit = &submit;
jobs[0] = &job;
memset(&jobs_data, 0, sizeof(jobs_data));
jobs_data.n_total = 1;
jobs_data.count = 1;
jobs_data.jobs = jobs;
memset(&src, 0, sizeof(src));
src.ts = ts;
src.body_which = ReplayEvent_body_jobsPush;
src.body.jobs_push = &jobs_data;
capn_init_malloc(&c);
cs = capn_root(&c).seg;
encode_ReplayEvent_ptr(cs, &p, &src);
capn_setp(capn_root(&c), 0, p.p);
TEST_ASSERT_EQUAL_INT(0, write_capnp_message(f, &c));
capn_free(&c);
}
/* Appends one ReplayEvent{jobStateChange} message -- the mechanism
* (core_events.c's XCE_CORE_EVENT_JOB_STATE_CHANGE handling) that flips a
* job to RUN state during replay independently of any decision, leaving
* job->accounts/job->allocs untouched (still NULL) -- exactly the trigger
* shape Recorded-Decision Reconciliation looks for. Same technique
* test_sim_event_loop.c's own append_jobstatechange_event() establishes. */
static void append_job_state_change_event(FILE* f, uint64_t ts,
uint64_t jobid, uint32_t old_state,
uint32_t new_state)
{
job_state_change_event_t jsc;
replay_event_t src;
struct capn c;
struct capn_segment* cs;
ReplayEvent_ptr p;
memset(&jsc, 0, sizeof(jsc));
jsc.jobid = jobid;
jsc.old_state = old_state;
jsc.new_state = new_state;
jsc.exit_code = 0;
memset(&src, 0, sizeof(src));
src.ts = ts;
src.body_which = ReplayEvent_body_jobStateChange;
src.body.job_state_change = &jsc;
capn_init_malloc(&c);
cs = capn_root(&c).seg;
encode_ReplayEvent_ptr(cs, &p, &src);
capn_setp(capn_root(&c), 0, p.p);
TEST_ASSERT_EQUAL_INT(0, write_capnp_message(f, &c));
capn_free(&c);
}
/* Appends the .replay stream's whole-stream termination marker
* (ReplayEvent{endSession}) -- same technique test_sim_verify.c's own
* append_end_of_stream_marker() establishes. */
static void append_end_of_stream_marker(FILE* f)
{
replay_event_t src;
struct capn c;
struct capn_segment* cs;
ReplayEvent_ptr p;
memset(&src, 0, sizeof(src));
src.body_which = ReplayEvent_body_endSession;
capn_init_malloc(&c);
cs = capn_root(&c).seg;
encode_ReplayEvent_ptr(cs, &p, &src);
capn_setp(capn_root(&c), 0, p.p);
TEST_ASSERT_EQUAL_INT(0, write_capnp_message(f, &c));
capn_free(&c);
}
/* Builds the wire-shaped xce_job_alloc_t a DISPATCH decision carries --
* same field shape test_sim_verify.c's own make_test_alloc()/
* make_test_host_alloc()/make_test_job_alloc() establish. */
static xce_job_alloc_t* make_dispatch_alloc(uint64_t jobid, const char* host,
float value)
{
xce_alloc_t* res = (xce_alloc_t*)calloc(1, sizeof(*res));
xce_host_alloc_t* h = (xce_host_alloc_t*)calloc(1, sizeof(*h));
xce_host_alloc_t** hosts = (xce_host_alloc_t**)calloc(1, sizeof(*hosts));
xce_alloc_t** allocs = (xce_alloc_t**)calloc(1, sizeof(*allocs));
xce_job_alloc_t* j = (xce_job_alloc_t*)calloc(1, sizeof(*j));
TEST_ASSERT_NOT_NULL(res);
TEST_ASSERT_NOT_NULL(h);
TEST_ASSERT_NOT_NULL(hosts);
TEST_ASSERT_NOT_NULL(allocs);
TEST_ASSERT_NOT_NULL(j);
res->resname = strdup("slots");
res->reserve = strdup("run");
res->instid = 1;
res->value = value;
res->from = 0;
res->duration = 0;
allocs[0] = res;
h->host = strdup(host);
h->nallocs = 1;
h->allocs = allocs;
hosts[0] = h;
j->jobid = jobid;
j->nhosts = 1;
j->allocs = hosts;
j->nres_shared = 0;
j->allocs_shared = NULL;
return j;
}
/* Appends one ReplaySession carrying a DISPATCH decision (job A) and a
* PEND decision (job B) -- same encode_ReplaySession_ptr() technique
* test_sim_verify.c's own append_replay_session_with_pend() establishes. */
static void append_replay_session_with_dispatch_and_pend(
FILE* f, uint64_t begin_round, xce_job_alloc_t* dispatch_alloc,
uint64_t pend_jobid, int pend_main_reason)
{
sched_dispatch_decision_t dispatch;
xce_job_preason_t reason;
sched_pend_decision_t pend;
replay_decision_t d_dispatch;
replay_decision_t d_pend;
replay_decision_t* decisions[2];
replay_session_t session;
struct capn c;
struct capn_segment* cs;
ReplaySession_ptr sp;
memset(&dispatch, 0, sizeof(dispatch));
dispatch.alloc = dispatch_alloc;
memset(&reason, 0, sizeof(reason));
reason.main_reason = pend_main_reason;
memset(&pend, 0, sizeof(pend));
pend.jobid = pend_jobid;
pend.reason = &reason;
memset(&d_dispatch, 0, sizeof(d_dispatch));
d_dispatch.ts = begin_round;
d_dispatch.body_which = ReplayDecision_body_dispatch;
d_dispatch.body.dispatch = &dispatch;
memset(&d_pend, 0, sizeof(d_pend));
d_pend.ts = begin_round;
d_pend.body_which = ReplayDecision_body_pend;
d_pend.body.pend = &pend;
decisions[0] = &d_dispatch;
decisions[1] = &d_pend;
memset(&session, 0, sizeof(session));
session.begin_round = begin_round;
session.decisions = decisions;
session.n_decisions = 2;
session.summary = NULL;
capn_init_malloc(&c);
cs = capn_root(&c).seg;
encode_ReplaySession_ptr(cs, &sp, &session);
capn_setp(capn_root(&c), 0, sp.p);
TEST_ASSERT_EQUAL_INT(0, write_capnp_message(f, &c));
capn_free(&c);
}
/* ---------------------------------------------------------------------- */
/* The end-to-end proof (task 3's own Observable Completion). */
/* ---------------------------------------------------------------------- */
/*
* test_reconciled_job_charges_resinuse_before_live_capacity_check - builds
* a `.replay` fixture where job A is already RUN with no live decision this
* round (reconciliation's trigger shape) and job B is genuinely still
* PEND, replays it through the REAL, unmodified sched_sim_driver_run(),
* then proves:
*
* (a) worker-a's resInUse reflects job A's 8.0-unit charge immediately
* after this SAME round (Req 1.3, 1.4) -- reconciliation ran, built
* job->accounts via the real make_job_account(), and
* sched_account_jobs() (called strictly after reconciliation, per
* task 3's own ordering) charged it into resInUse before this
* function returns.
*
* (b) sched_check_host_satisfy() (resource.c, completely unmodified)
* against this SAME post-round worker state now correctly DENIES a
* hypothetical job wanting more than the 2.0-unit remaining
* headroom, and GRANTS one wanting exactly 2.0 -- proving
* Requirement 2's capacity-awareness is a genuine, working property
* of the now-correct resInUse (not merely that resInUse holds some
* nonzero number).
*
* Before task 3's wiring, sim_reconcile_unaccounted_jobs() was never
* called, so job A's RUN state was never reconciled: job->accounts stays
* NULL, sched_account_jobs() skips it (its own job->accounts != NULL
* guard), and resInUse for worker-a stays at 0.0 for the whole run --
* this is bug-1613/1611's own documented behavior, and exactly what
* this test's assertion (a) would catch as a regression if the wiring
* were ever removed.
*/
static void
test_reconciled_job_charges_resinuse_before_live_capacity_check(void)
{
char record_dir[PATH_MAX];
xce_scheduler_t sched;
xce_scheduler_t wrapper;
FILE* f;
char r_path[PATH_MAX];
int residx;
xce_job_alloc_t* alloc;
sched_sim_args_t sim_args;
xce_sched_sim_driver_t sim;
xce_sched_worker_t* w;
xce_sched_job_t* jobA;
xce_sched_job_t* jobB;
int rc;
float got[1];
xvec_t uncomitted;
xvec_idval_t wanted;
xvec_idval_t unmeet;
make_tmp_dir(record_dir, sizeof(record_dir), "_e2e");
residx = fixture_init(&sched, record_dir);
rc = sched_record_start(&sched, "reconciliation.replay");
TEST_ASSERT_EQUAL_INT(0, rc);
TEST_ASSERT_TRUE(strlen(sched.record_file_path) < sizeof(r_path));
strcpy(r_path, sched.record_file_path);
f = sched.record_file;
append_queue_update_event(f, 1000ULL, "normal");
append_jobs_push_event(f, 1500ULL, JOBA_ID, "normal");
append_jobs_push_event(f, 1500ULL, JOBB_ID, "normal");
append_job_state_change_event(f, 1800ULL, JOBA_ID, JOB_STAT_PEND,
JOB_STAT_RUN);
alloc = make_dispatch_alloc(JOBA_ID, WORKER_NAME, JOBA_WANT);
append_replay_session_with_dispatch_and_pend(f, 2000ULL, alloc, JOBB_ID,
0);
append_end_of_stream_marker(f);
fclose(f);
memset(&sim_args, 0, sizeof(sim_args));
sim_args.replay_path = r_path;
sim_args.sim_start_ts = SCHED_SIM_START_TS_LATEST;
sim_args.sim_speed = 0; /* frozen -- deterministic, no wall-clock skew */
memset(&sim, 0, sizeof(sim));
rc = sched_sim_driver_run(&sim, &sim_args);
TEST_ASSERT_EQUAL_INT(0, rc);
/* Sanity: job A really is RUN with a reconstructed allocation, job B
* really is still PEND with none -- the fixture's own shape proof,
* not merely trusting the input constants. */
jobA = sched_get_job(&(sim.core), JOBA_ID);
TEST_ASSERT_NOT_NULL(jobA);
TEST_ASSERT_TRUE(JOB_IS_RUN(jobA->state));
TEST_ASSERT_NOT_NULL_MESSAGE(
jobA->accounts,
"reconciliation must have reconstructed job A's accounts from the "
"recorded DISPATCH decision");
jobB = sched_get_job(&(sim.core), JOBB_ID);
TEST_ASSERT_NOT_NULL(jobB);
TEST_ASSERT_FALSE(JOB_IS_RUN(jobB->state));
TEST_ASSERT_NULL(jobB->accounts);
/* (a) Req 1.3/1.4: worker-a's resInUse reflects job A's charge this
* SAME round. */
w = sched_get_worker(&(sim.core), WORKER_NAME);
TEST_ASSERT_NOT_NULL(w);
got[0] = -1.0f;
rc = get_resvals(&(w->resInUse), 1, &residx, got);
TEST_ASSERT_EQUAL_INT(1, rc); /* get_resvals() returns nres on success */
TEST_ASSERT_EQUAL_FLOAT_MESSAGE(
JOBA_WANT, got[0],
"worker-a's resInUse must reflect job A's reconciled 8.0-unit "
"charge before this function returns");
/* (b) Req 2.1/2.2: the REAL, unmodified sched_check_host_satisfy()
* against this SAME post-round state must now correctly deny a job
* wanting more than the 2.0-unit remaining headroom (10.0 total -
* 8.0 charged), and grant one wanting exactly the remainder. sched
* is only used via sched->core.resmgr inside resource.c's own
* unsatisfy_() TRACE line (design.md's own confirmed-by-grep note),
* so a throwaway xce_scheduler_t wrapping the SAME sim.core is a
* valid, safe caller-owned argument here -- not a special case. */
memset(&wrapper, 0, sizeof(wrapper));
wrapper.core = sim.core;
rc = xvec_init(&uncomitted, 1, sizeof(float));
TEST_ASSERT_TRUE(rc >= 0);
rc = xvec_idval_init(&wanted, 1);
TEST_ASSERT_EQUAL_INT(0, rc);
rc = xvec_idval_store(&wanted, residx, 5.0f); /* > 2.0 remaining */
TEST_ASSERT_TRUE(rc >= 0);
rc = sched_check_host_satisfy(&wrapper, w, &uncomitted, &wanted, &unmeet);
TEST_ASSERT_EQUAL_INT_MESSAGE(
0, rc,
"a job wanting more than the 2.0-unit remaining headroom must be "
"denied -- capacity awareness must hold against the reconciled "
"resInUse, not just report a number");
xvec_fini(&uncomitted);
xvec_idval_fini(&wanted);
xvec_idval_fini(&unmeet);
rc = xvec_init(&uncomitted, 1, sizeof(float));
TEST_ASSERT_TRUE(rc >= 0);
rc = xvec_idval_init(&wanted, 1);
TEST_ASSERT_EQUAL_INT(0, rc);
rc = xvec_idval_store(&wanted, residx, 2.0f); /* == 2.0 remaining, exact fit */
TEST_ASSERT_TRUE(rc >= 0);
rc = sched_check_host_satisfy(&wrapper, w, &uncomitted, &wanted, &unmeet);
TEST_ASSERT_EQUAL_INT_MESSAGE(
1, rc,
"a job wanting exactly the 2.0-unit remaining headroom must be "
"granted -- proving the denial above is genuine capacity "
"arithmetic, not a fixture that always denies");
xvec_fini(&uncomitted);
xvec_idval_fini(&wanted);
xvec_idval_fini(&unmeet);
printf("test_reconciled_job_charges_resinuse_before_live_capacity_"
"check: PASS\n");
}
int main(void)
{
UNITY_BEGIN();
RUN_TEST(test_reconciled_job_charges_resinuse_before_live_capacity_check);
return UNITY_END();
}

155
src/scheduler/testsuite/test_sim_verify.c
View file

@ -748,6 +748,89 @@ static void append_replay_session_with_pend(FILE* f, uint64_t begin_round,
capn_free(&c);
}
/* Appends one ReplayEvent{jobStateChange} message -- scheduler-sim-
* decision-parity spec, task 3 (Req 1.3/1.4, 2.1/2.2, 5.2): this is the
* mechanism (core_events.c's XCE_CORE_EVENT_JOB_STATE_CHANGE handling)
* that flips a job to RUN state during replay independently of any
* decision -- it does NOT set job->accounts/job->allocs (core_events.c
* only clears them when leaving RUN state), which is exactly what leaves
* a RUN-state job with job->accounts == NULL for Recorded-Decision
* Reconciliation (sim_reconcile_unaccounted_jobs(), sim_loop.c) to pick
* up. Same encode/write technique append_queue_update_event() above
* establishes. */
static void append_job_state_change_event(FILE* f, uint64_t ts,
uint64_t jobid, uint32_t old_state,
uint32_t new_state)
{
job_state_change_event_t jsc;
replay_event_t src;
struct capn c;
struct capn_segment* cs;
ReplayEvent_ptr p;
memset(&jsc, 0, sizeof(jsc));
jsc.jobid = jobid;
jsc.old_state = old_state;
jsc.new_state = new_state;
jsc.exit_code = 0;
memset(&src, 0, sizeof(src));
src.ts = ts;
src.body_which = ReplayEvent_body_jobStateChange;
src.body.job_state_change = &jsc;
capn_init_malloc(&c);
cs = capn_root(&c).seg;
encode_ReplayEvent_ptr(cs, &p, &src);
capn_setp(capn_root(&c), 0, p.p);
TEST_ASSERT_EQUAL_INT(0, write_capnp_message(f, &c));
capn_free(&c);
}
/* Appends one ReplaySession carrying TWO recorded decisions -- a DISPATCH
* for `dispatch_alloc`'s jobid and a PEND for `pend_jobid` -- scheduler-
* sim-decision-parity spec, task 3 (Req 5.2). Reuses
* make_replay_dispatch_decision()/make_replay_pend_decision() (this file's
* own unit-level fixture helpers, above) rather than hand-rolling the
* decision shape a second time. Mirrors append_replay_session_with_pend()'s
* own encode_ReplaySession_ptr() technique, just with two decisions
* instead of one. */
static void append_replay_session_with_dispatch_and_pend(
FILE* f, uint64_t begin_round, xce_job_alloc_t* dispatch_alloc,
uint64_t pend_jobid, int pend_main_reason)
{
replay_decision_t d_dispatch;
replay_decision_t d_pend;
replay_decision_t* decisions[2];
replay_session_t session;
struct capn c;
struct capn_segment* cs;
ReplaySession_ptr sp;
d_dispatch = make_replay_dispatch_decision(dispatch_alloc);
d_dispatch.ts = begin_round;
d_pend = make_replay_pend_decision(pend_jobid, pend_main_reason);
d_pend.ts = begin_round;
decisions[0] = &d_dispatch;
decisions[1] = &d_pend;
memset(&session, 0, sizeof(session));
session.begin_round = begin_round;
session.decisions = decisions;
session.n_decisions = 2;
session.summary = NULL;
capn_init_malloc(&c);
cs = capn_root(&c).seg;
encode_ReplaySession_ptr(cs, &sp, &session);
capn_setp(capn_root(&c), 0, sp.p);
TEST_ASSERT_EQUAL_INT(0, write_capnp_message(f, &c));
capn_free(&c);
}
/* Runs sched_sim_driver_run() against `path` with --sim-verify set to
* `verify`, capturing everything written to stdout during the call (the
* same dup()/dup2() technique capture_stdout_around() above uses, inlined
@ -922,6 +1005,77 @@ static void test_verify_disabled_reports_no_mismatches_even_when_altered(void)
"PASS\n");
}
/*
* test_verify_with_already_resolved_job_reports_no_mismatch -- scheduler-
* sim-decision-parity spec, task 3 (Req 5.2): --sim-verify against a
* recording containing an already-resolved job (job A: pushed PEND, then
* flipped to RUN by a JOB_STATE_CHANGE event in the SAME round's event
* batch, with a matching recorded DISPATCH decision this round -- the
* Recorded-Decision Reconciliation trigger shape) must report ZERO
* mismatches, alongside a genuinely-live-pended job (job B) whose own
* recorded PEND decision the real picker independently reproduces.
*
* Before task 3's round-boundary wiring, sim_reconcile_unaccounted_jobs()
* was never called, so job A's RUN state was never reconciled: the live
* picker never independently re-decides an already-RUN job (it only ever
* offers PEND-state jobs), so the live-drained decision stream has only
* job B's PEND -- one entry -- while decoded_session->decisions still
* carries BOTH the DISPATCH(A) and PEND(B) entries (neither ever
* consumed). The positional verify walk then finds job A's recorded
* DISPATCH left over with no corresponding live decision once the live
* stream ends, and reports it as a mismatch (expected=DISPATCH(A),
* actual=NULL) -- a false positive entirely attributable to this
* feature's own gap, not a real behavioral divergence. After task 3's
* fix, reconciliation consumes and removes job A's DISPATCH entry from
* decoded_session BEFORE this verify loop ever runs, so only job B's PEND
* remains to compare against the live stream's own single PEND -- zero
* mismatches.
*/
static void test_verify_with_already_resolved_job_reports_no_mismatch(void)
{
char path[PATH_MAX];
FILE* f;
int rc;
char* captured;
xce_alloc_t* res[1];
xce_host_alloc_t* host[1];
xce_job_alloc_t* alloc;
f = write_header_and_snapshot(path, sizeof(path), "_reconciled");
TEST_ASSERT_NOT_NULL(f);
append_queue_update_event(f, 1000ULL, "normal");
/* Job A (501): already resolved before this round's live picker ever
* considers it -- pushed PEND, then flipped to RUN in the SAME event
* batch, so it never reaches next_job() this round. Job B (502):
* genuinely still PEND -- the real picker independently re-pends it. */
append_jobs_push_event(f, 1500ULL, 501ULL, "normal");
append_jobs_push_event(f, 1500ULL, 502ULL, "normal");
append_job_state_change_event(f, 1800ULL, 501ULL, JOB_STAT_PEND,
JOB_STAT_RUN);
res[0] = make_test_alloc("slots", "run", 1, 4.0, 0, 0);
host[0] = make_test_host_alloc("hostA", res, 1);
alloc = make_test_job_alloc(501ULL, host, 1);
append_replay_session_with_dispatch_and_pend(f, 2000ULL, alloc, 502ULL,
0);
append_end_of_stream_marker(f);
fclose(f);
captured = run_sim_and_capture_stdout(path, true, &rc);
TEST_ASSERT_EQUAL_INT(0, rc);
TEST_ASSERT_NULL_MESSAGE(
strstr(captured, "MISMATCH"),
"reconciling an already-resolved job's recorded DISPATCH decision "
"must not produce a false --sim-verify mismatch (Requirement 5.2)");
free(captured);
printf("test_verify_with_already_resolved_job_reports_no_mismatch: "
"PASS\n");
}
int main(void)
{
UNITY_BEGIN();
@ -940,6 +1094,7 @@ int main(void)
RUN_TEST(test_verify_against_self_reports_zero_mismatches);
RUN_TEST(test_verify_against_altered_copy_reports_exact_mismatch);
RUN_TEST(test_verify_disabled_reports_no_mismatches_even_when_altered);
RUN_TEST(test_verify_with_already_resolved_job_reports_no_mismatch);
return UNITY_END();
}