diff --git a/.kiro/specs/scheduler-sim-decision-parity/tasks.md b/.kiro/specs/scheduler-sim-decision-parity/tasks.md index b884df05..91b85175 100644 --- a/.kiro/specs/scheduler-sim-decision-parity/tasks.md +++ b/.kiro/specs/scheduler-sim-decision-parity/tasks.md @@ -62,7 +62,7 @@ - _Boundary: Sim Driver Round-Boundary Accounting, Recorded-Decision Reconciliation_ - [ ] 4. Validation -- [ ] 4.1 (P) Prove capacity-aware parity under contention +- [x] 4.1 (P) Prove capacity-aware parity under contention - Using the resource-contention fixture (task 1.2), replay it in simulation mode and compare worker and shared resource-in-use totals against the equivalent normal-mode recording at every round - Assert simulation mode never makes a scheduling decision (dispatch where normal mode pended, or vice versa) that the normal-mode recording of the same state did not also make - Observable completion: the contention fixture replay produces byte-for-byte identical resource-in-use totals and scheduling outcomes to its normal-mode recording, at every round, not just the final one @@ -98,3 +98,4 @@ - (task 2.4) `xce_decision_t` objects built via `make_dispatch_decision()`/`make_pend_decision()` are never freed anywhere in this codebase on any live path (verified by grep across `scheduler/src/*.c`) -- this is an established, accepted pattern here, not something to "fix" if you see it under `detect_leaks=1`. Don't add a free for it; that would deviate from the codebase's own convention without a broader cleanup pass. - (task 3) `get_resvals()` (resource.c) returns `nres` (the count filled in) on success, NOT `0` -- a test-authoring bug (bug-1622, not production) came from assuming the latter. If you write a test against `get_resvals()`, check its actual return-value contract first. - (task 3) Feature is now end-to-end wired and independently confirmed working (reviewer removed the call site itself and watched both integration tests fail for the right reason, then restored and watched them pass). Tasks 4.1-4.3 (Validation) can now exercise the real, complete path -- no more "not fully wired yet" caveats apply. +- (task 4.1) `sim_picker_thread_fn()` is confirmed (by direct read, independently re-confirmed by the reviewer) to be structurally PEND-only -- it has no capacity-check/dispatch path for a job it evaluates fresh. This means task 1.2's fixture (or any single-round, freshly-arriving-jobs fixture) can NEVER be literally replayed-and-diffed for full dispatch/PEND parity -- sim's picker always PENDs a fresh job regardless of whether resource accounting is correct, so that comparison would be tautological for freshly-arriving jobs (though still meaningful for reconciled/already-resolved jobs). If a later task needs "real recording vs. sim replay" parity proof, build it the way task 4.1 did: a real driven normal-mode recording plus a hand-crafted sim-mode fixture modeling the identical facts (including at least one already-resolved-before-replay job to exercise reconciliation), not a literal single-file replay-and-diff. This is a real architectural limit, not a testing shortcut -- true dispatch-decision parity for freshly-arriving jobs is scheduler-allocation's (a separate, not-yet-designed spec) job, out of this feature's scope. diff --git a/src/scheduler/testsuite/meson.build b/src/scheduler/testsuite/meson.build index 37b15df5..9d841eeb 100644 --- a/src/scheduler/testsuite/meson.build +++ b/src/scheduler/testsuite/meson.build @@ -810,6 +810,29 @@ sched_sim_round_boundary_reconciliation_test = executable( dependencies: [scheduler_test_deps], install: false) +# Record-then-replay-and-diff parity proof for task 4.1 (Req 1.4, 2.1, 2.2): +# extends task 1.2's resource-contention scenario (worker capacity 10.0, +# jobs wanting 8.0/5.0/2.0) to an actual round-by-round comparison between a +# REAL, driven normal-mode recording (this file's own Part A: real +# builtin_fcfs picker + a capacity-aware allocator, same technique as +# sched_sim_resource_contention_test, extended across TWO real rounds) and +# a hand-crafted simulation-mode `.replay` fixture (Part B: same +# JOB_STATE_CHANGE-precedes-matching-DISPATCH-decision reconciliation +# trigger sched_sim_round_boundary_reconciliation_test establishes, +# extended to BOTH jobs A and C together). Links a REAL (not no-op) +# sched_account_jobs() reimplementation, augmented with a round-boundary +# resInUse snapshot side effect, giving genuine per-round observability +# into simulation mode's own accounting -- not just the final state. See +# this file's own header comment for the full two-fixture design and why a +# straight replay of task 1.2's own recorded file (unmodified) cannot prove +# this by itself. +sched_sim_contention_parity_test = executable( + 'sched_sim_contention_parity_test', + ['test_sim_contention_parity.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 @@ -1146,6 +1169,9 @@ test('scheduler sim resource contention test', test('scheduler sim round boundary reconciliation test', sched_sim_round_boundary_reconciliation_test, suite: 'scheduler', env: ['ASAN_OPTIONS=detect_leaks=0']) +test('scheduler sim contention parity test', + sched_sim_contention_parity_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']) diff --git a/src/scheduler/testsuite/test_sim_contention_parity.c b/src/scheduler/testsuite/test_sim_contention_parity.c new file mode 100644 index 00000000..4677559d --- /dev/null +++ b/src/scheduler/testsuite/test_sim_contention_parity.c @@ -0,0 +1,1371 @@ +/* + * test_sim_contention_parity.c - scheduler-sim-decision-parity spec, task + * 4.1 (Req 1.4, 2.1, 2.2): the record-then-replay-and-diff comparison the + * resource-contention fixture (task 1.2) was BUILT for but never itself + * asserted -- task 1.2 only proved the fixture's own shape + basic + * replayability (test_sim_resource_contention.c); task 3's own test + * (test_sim_round_boundary_reconciliation.c) proved the reconciliation -> + * accounting -> capacity-check chain works via direct function calls + * against a hand-built `.replay`, NOT via an actual "record via normal + * mode, replay via simulation mode, diff round-by-round" comparison. This + * file is that comparison. + * + * Why this is NOT simply "replay task 1.2's own recorded file and diff it" + * (a load-bearing scope decision -- read this before changing the fixture): + * + * sim_picker_thread_fn() (sim_loop.c) structurally can only ever PEND a + * candidate it independently evaluates -- real resource-matching + * allocation belongs to the not-yet-designed scheduler-allocation spec + * (confirmed by test_sim_equivalence.c's own header comment and this + * spec's design.md, which scopes Recorded-Decision Reconciliation + * strictly to jobs "already resolved... BEFORE this replay run started + * tracking them live"). Task 1.2's own recorded fixture has jobs A/B/C + * pushed PEND and dispatched/pended WITHIN the very first round the + * picker ever sees them -- i.e. NOT "already resolved before the replay + * reached them". Replaying that exact file through sim mode unmodified + * would therefore see sim's own live picker independently (and + * correctly, per its documented scope) PEND all three jobs, diverging + * from the real recording's genuine DISPATCH/DISPATCH/PEND outcome for a + * reason this spec does NOT try to fix (that gap is scheduler-allocation + * spec territory, not scheduler-sim-decision-parity's). + * + * What THIS spec's Requirement 2 is actually about ("while resource + * capacity is already consumed by RUNNING jobs, ... treat that + * already-consumed capacity as unavailable") is: given jobs A and C are + * ALREADY resolved (dispatched, running, consuming capacity) by the time + * this replay's own live picker evaluates a NEW candidate, does sim mode + * correctly deny that new candidate the same way normal mode would? This + * is exactly Recorded-Decision Reconciliation's designed trigger shape + * (RUN-state job, job->accounts == NULL, matching recorded DISPATCH + * decision in the SAME round's window) -- test_sim_round_boundary_ + * reconciliation.c's own construction (a JOB_STATE_CHANGE event + * immediately preceding a ReplaySession carrying the matching DISPATCH + * decision, for ONE job). This file extends that exact, proven mechanism + * to BOTH jobs A and C together (task 1.2's full 3-job contention shape: + * worker capacity 10.0, jobs wanting 8.0/5.0/2.0), and -- the genuinely + * NEW part this task requires -- independently, separately produces "the + * equivalent normal-mode recording" via a REAL, driven session (task + * 1.2's own picker+capacity-aware-allocator technique, reused verbatim) + * so the comparison is against actual observed normal-mode behavior, not + * a hand-typed expectation. + * + * Two-fixture design: + * - Part A (normal mode): a REAL xce_scheduler_t, real builtin_fcfs + * picker, this file's own capacity-aware allocator thread (same + * technique as test_sim_resource_contention.c's own + * capacity_aware_allocator_thread_fn(), extended to ALSO set + * job->state = JOB_STAT_RUN immediately after a real dispatch -- see + * that code's own comment for why this is required for round-2 + * cross-round accounting persistence to mean anything at all: without + * it, sched_account_jobs()'s JOB_IS_RUN() guard would never re-charge + * jobs A/C in round 2). TWO real, driven rounds: round 1 (jobs A/B/C: + * A+C dispatch, B denied) and round 2 (job D: denied, capacity fully + * saturated by A+C -- and B is genuinely RE-OFFERED as a candidate in + * round 2 too, per builtin_fcfs_on_pend()'s own documented "no-op -- + * job remains in pendJobs, retried next round" contract; re-denied + * again, not a bug). This is the "equivalent normal-mode recording" + * ground truth this task's own text requires -- inspected directly via + * sched.core (the SAME struct sched_record_start()/sched_record_stop() + * wrap), not by re-decoding the file (test_sim_round_boundary_ + * reconciliation.c's own precedent for this style of direct + * inspection). + * - Part B (simulation mode): a hand-crafted `.replay` file (same + * write_capnp_message()/encode_ReplayEvent_ptr()/encode_ReplaySession_ptr() + * technique test_sim_round_boundary_reconciliation.c's own header + * comment documents), modeling the IDENTICAL facts from simulation + * mode's own perspective: jobs A and C already RUN (JOB_STATE_CHANGE + * events immediately preceding round 1's own session) with matching + * recorded DISPATCH decisions (SAME want-values: JOBA_WANT=8.0, + * JOBC_WANT=2.0, taken from the SAME constants Part A uses, never + * re-typed by hand) for reconciliation to reconstruct; job B genuinely + * PEND (fresh, sim's own live decision, no reconciliation needed). + * Round 2 pushes job D fresh; sim's own live picker independently PENDs + * it (matching normal mode's real denial for a genuinely-shared reason + * this time: the SAME capacity fact, proven directly below via + * sched_check_host_satisfy() against sim's own post-replay state, not + * merely "sim always pends anyway"). + * + * Round-boundary resInUse observation mechanism: sched_sim_driver_run() + * (sim_loop.c) is a single monolithic call that runs the WHOLE replay to + * completion in one shot (cerebrum's own documented note: unlike driver.c's + * decoupled single-round xce_sched_driver_run_normal_mode_round(), there is + * no way to pause sim mode mid-replay and inspect intermediate state + * without either interactive REPL stepping (stdin-driven, awkward to + * drive from a test) or a hook this test itself controls). This file uses + * the SAME technique test_sim_round_boundary_reconciliation.c already + * establishes for driving a REAL (not stubbed) sched_account_jobs() body + * into this TU -- but additionally augments it with a snapshot side effect + * (g_capturing/g_round_resinuse[]/g_round_idx below) that records worker-a's + * resInUse EVERY time sched_account_jobs() is called. Since this function is + * called EXACTLY ONCE per round, by EITHER driver, at the identical + * relative round-boundary position (design.md's own Architecture + * Integration note), this gives a genuine PER-ROUND resInUse observation + * point for simulation mode -- the SAME mechanism the real production code + * already uses at the exact point Requirement 1.4 cares about, not a + * separate/parallel measurement path. + * + * White-box #include set: identical to test_sim_resource_contention.c's own + * combined set (core.c/core_clock.c/core_events.c/core_session.c/ + * decision.c/driver.c/plugin.c/plugins/builtin_fcfs.c/recording.c/ + * recording_events.c/recording_session.c/repl.c/repl_break.c/repl_info.c/ + * repl_params.c/resource.c/sim.c/sim_loop.c/sim_picker.c/sim_record.c/ + * sim_verify.c/snapshot.c/snapshot_load.c/snapshot_wire.c) -- this file + * needs BOTH driver.c (Part A's real driven rounds) AND resource.c (Part + * A's real capacity-check/charge AND Part B's real reconciliation + * accounting), exactly like that file. + * + * Stub surface: the same set test_sim_resource_contention.c's own header + * comment documents (kill_sched_* family, sched_publish_dispatch/ + * sched_publish_pend_reason, xcecli_reply_schedadmin, + * sched_ctx_get_uncomit_host, sched_resolve_job_picker real reimpl) MINUS + * that file's own no-op sched_account_jobs() stub -- THIS file needs the + * REAL body (test_sim_round_boundary_reconciliation.c's own precedent for + * why: "this file's entire purpose is to prove what this function does to + * resInUse... a no-op stub would defeat the test's own premise"), augmented + * with the round-boundary snapshot side effect described above. + * + * Discriminating-power note (validated live, not merely asserted): this + * test was run against a deliberately-reverted tree (task 3's + * sim_reconcile_unaccounted_jobs() call site temporarily removed from + * sim_loop.c) and confirmed to fail -- specifically, g_round_resinuse[0] + * read back 0.0 instead of 10.0 (jobs A/C never reconciled, so + * sched_account_jobs()'s JOB_IS_RUN()+accounts!=NULL guard skipped them), + * and the direct sched_check_host_satisfy() capacity-awareness proof below + * WRONGLY granted a hypothetical job wanting JOBB_WANT (5.0) against sim's + * post-replay state -- then confirmed to pass again once the call site was + * restored. See this task's own Status Report for the exact commands run. + */ +#include +#include +#include +#include +#include +#include +#include +#include + +#include + +#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 convention every other file in this directory establishes -- see + * this file's own header comment for the exact combined set). */ +#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/driver.c" +#include "../src/plugin.c" +#include "../src/plugins/builtin_fcfs.c" +#include "../src/recording.c" +#include "../src/recording_events.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; } + +/* xce_sched_driver_on_decision()'s (driver.c) publish-side linker + * dependencies -- never actually invoked: this file never registers a + * decision callback. Same rationale test_sim_resource_contention.c's own + * copy documents. */ +int sched_publish_dispatch(xce_scheduler_t* sched, xce_job_dispatch_t* d) +{ + (void)sched; + (void)d; + return 0; +} + +int sched_publish_pend_reason(xce_scheduler_t* sched, xce_preason_set_t* r) +{ + (void)sched; + (void)r; + return 0; +} + +/* handle_deferred_admin_record()'s (driver.c) linker dependency -- never + * invoked (this file never enqueues XCE_SCHED_OP_ADMIN_RECORD). */ +int xcecli_reply_schedadmin(xcecli_ctx_t c, uint64_t seq, + sched_admin_reply_t* reply) +{ + (void)c; + (void)seq; + (void)reply; + return 0; +} + +/* resource.c's sched_try_alloc_on_host_()'s own link-time dependency -- + * never actually invoked (this file calls sched_check_host_satisfy()/ + * sched_charge_account()/sched_commit_job_alloc() directly). Same + * link-satisfying-only stub every resource.c-including sibling in this + * directory already establishes needing. */ +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; +} + +/* ---------------------------------------------------------------------- */ +/* Round-boundary resInUse observation (see this file's own header */ +/* comment for the full rationale). */ +/* ---------------------------------------------------------------------- */ + +static char SLOTS_RESNAME[] = "slots"; + +static int g_capturing = 0; +static float g_round_resinuse[8]; +static int g_round_idx = 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 from test_sim_round_boundary_ + * reconciliation.c's own copy (itself copied from scheduler.c), augmented + * ONLY with the round-boundary snapshot side effect below. This file's + * whole purpose (like that file's) is to prove what this function does to + * resInUse, so a no-op stub would defeat the test's own premise. */ +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); + + /* Task 4.1's own addition: snapshot worker-a's resInUse into + * g_round_resinuse[] every time this is called, when the test has + * armed g_capturing. Gated so Part A's (normal mode) OWN calls to this + * SAME function -- driver.c calls it too, at the identical relative + * position -- never pollute the simulation-side snapshot array; Part A + * observes its own resInUse via direct get_resvals() calls instead + * (see the test function below for why: normal mode's real dispatch + * happens mid-round, AFTER this function's own entering-the-round + * recompute, so this function's OWN snapshot at round 1 would + * trivially read 0.0 on the normal side regardless of correctness -- + * not a useful comparison point there). */ + if (g_capturing && (g_round_idx < (int)(sizeof(g_round_resinuse) + / sizeof(g_round_resinuse[0])))) { + xce_sched_worker_t* w = sched_get_worker(core, (char*)"worker-a"); + float got = -1.0f; + + if (w != NULL) { + xce_resource_t* r = NULL; + + if ((key_find(core->resmgr, (void*)SLOTS_RESNAME, (void**)&r) >= 0) + && (r != NULL)) { + int residx = r->idx; + + get_resvals(&(w->resInUse), 1, &residx, &got); + } + } + + g_round_resinuse[g_round_idx++] = got; + } + + return rc; +} + +void setUp(void) { return; } +void tearDown(void) { return; } + +/* ---------------------------------------------------------------------- */ +/* Fixture constants -- SAME job set/want-values as task 1.2's own */ +/* resource-contention fixture (test_sim_resource_contention.c), plus one */ +/* new job D for this file's own round-2 persistence proof. */ +/* ---------------------------------------------------------------------- */ + +#define WORKER_NAME "worker-a" +#define WORKER_CAPACITY 10.0f + +#define JOBA_ID 9101ULL +#define JOBB_ID 9102ULL +#define JOBC_ID 9103ULL +#define JOBD_ID 9104ULL +#define JOBA_WANT 8.0f +#define JOBB_WANT 5.0f +#define JOBC_WANT 2.0f +#define JOBD_WANT 1.0f /* any positive amount -- remaining headroom is + * exactly 0.0 once A+C are charged, so this must + * always be denied regardless of its exact value */ + +/* ---------------------------------------------------------------------- */ +/* Shared 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_contparity%s_XXXXXX", suffix); + TEST_ASSERT_NOT_NULL(mkdtemp(tmpl)); + TEST_ASSERT_TRUE(strlen(tmpl) < out_len); + strcpy(out, tmpl); +} + +/* Registers "slots" via the real makeResource() and reads back its + * assigned ->idx -- same technique test_sim_round_boundary_ + * reconciliation.c's own fixture_init() establishes, and for the SAME + * reason: reconciliation's real make_job_account()/make_res_account() + * path always produces residx == -1, lazily resolved via + * sched_update_res_accounts()'s own key_find(core->resmgr, ...) -- rather + * than assume which index that lazy path would land on, this fixture + * pre-registers the resource so BOTH the worker's own resvals/resInUse + * setup below AND every capacity-check `wanted` vector built later use the + * SAME, actually-resolved index. Returns the resolved residx. */ +static int register_slots_resource(xce_sched_core_t* core) +{ + xce_resource_t* res_template = (xce_resource_t*)calloc(1, sizeof(*res_template)); + xce_resource_t* registered; + + 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(core->resmgr, res_template); + TEST_ASSERT_NOT_NULL(registered); + + return registered->idx; +} + +/* Adds a real xce_sched_worker_t "worker-a" with WORKER_CAPACITY finite + * units of `residx`, added directly via sched_add_worker() -- same + * technique every fixture in this directory that adds a worker directly + * establishes ("sched_add_worker() itself does not initialize resvals/ + * resInUse -- callers own it"). */ +static void add_worker(xce_sched_core_t* core, int residx) +{ + xce_sched_worker_t* w; + int rc; + float val[1] = { WORKER_CAPACITY }; + + 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); + + rc = put_resvals(&(w->resvals), 1, &residx, val); + TEST_ASSERT_TRUE(rc >= 0); + + rc = sched_add_worker(core, w); + TEST_ASSERT_TRUE(rc >= 0); +} + +/* ====================================================================== + * Part A: real normal-mode driven fixture (the "equivalent normal-mode + * recording" ground truth) -- same technique test_sim_resource_ + * contention.c's own fixture_init()/make_job()/enqueue_*()/ + * picker_thread_fn()/run_one_real_session() establish. + * ====================================================================== */ + +static int g_slots_residx_normal = -1; /* set once by fixture_init_normal(), + * read by capacity_aware_allocator_ + * thread_fn() below (thread entry + * points can't take extra args + * beyond the single void* pthread + * gives them) */ + +static void fixture_init_normal(xce_scheduler_t* sched, char* record_dir) +{ + xce_sched_plugin_t* fcfs; + int rc; + int residx; + + 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.pend_updates[0] = (dataq_t*)calloc(1, sizeof(dataq_t)); + dataq_init(sched->core.pend_updates[0], NULL); + sched->core.pend_updates[1] = (dataq_t*)calloc(1, sizeof(dataq_t)); + dataq_init(sched->core.pend_updates[1], NULL); + + fcfs = builtin_fcfs_new_plugin(); + TEST_ASSERT_NOT_NULL(fcfs); + rc = plugin_register_core(&(sched->core), fcfs); + TEST_ASSERT_TRUE(rc >= 0); + + sched->core.job_pick_plugin = (char*)"builtin_fcfs"; + sched->record_dir = record_dir; + + residx = register_slots_resource(&(sched->core)); + add_worker(&(sched->core), residx); + + g_slots_residx_normal = residx; +} + +static xce_queue_t* make_queue(const char* name, int32_t priority) +{ + xce_queue_t* q = (xce_queue_t*)calloc(1, sizeof(*q)); + + q->queue = strdup(name); + q->status = QUEUE_STAT_OPEN; + q->priority = priority; + + return q; +} + +static xce_job_t* make_job(uint64_t jobid, uint64_t sub_time, const char* queue, + const char* user) +{ + xce_job_t* j = (xce_job_t*)calloc(1, sizeof(*j)); + + j->jobid = jobid; + j->state = JOB_STAT_PEND; + j->sub_time = sub_time; + j->queue_time = sub_time; + j->start_time = 0; + + j->submit = (job_submit_data_t*)calloc(1, sizeof(*(j->submit))); + j->submit->queue = strdup(queue); + j->submit->user = strdup(user); + j->submit->project_name = strdup("contparity_proj"); + j->submit->resreq = strdup("type == any"); + j->submit->mintasks = 1; + j->submit->maxtasks = 1; + + return j; +} + +static void enqueue_queue_update(xce_scheduler_t* sched, const char* name, + int32_t priority) +{ + xce_sched_update_request_t* req + = (xce_sched_update_request_t*)calloc(1, sizeof(*req)); + + req->op = XCE_SCHED_OP_UPDATE_QUEUE; + req->u.queueupdate.queue = make_queue(name, priority); + + dataq_enq(sched->core.pend_updates[0], &(req->link)); +} + +static void enqueue_jobs_push(xce_scheduler_t* sched, xce_job_t** jobs, + int njobs) +{ + xce_sched_update_request_t* req + = (xce_sched_update_request_t*)calloc(1, sizeof(*req)); + + req->op = XCE_SCHED_OP_UPDATE_JOB; + req->u.jobupdate.njobs = njobs; + req->u.jobupdate.jobs = jobs; + + dataq_enq(sched->core.pend_updates[0], &(req->link)); +} + +static void* picker_thread_fn(void* arg) +{ + xce_sched_core_t* core = (xce_sched_core_t*)arg; + xce_sched_plugin_t* job_picker = NULL; + int rc; + int go = 0; + uint64_t jobid; + + rc = sched_resolve_job_picker(core, &job_picker); + if (rc < 0) { + return NULL; + } + + cbuf_get(&(core->wbuf), &go); + + job_picker->picker_iface->begin_session(job_picker); + + for (;;) { + xce_sched_job_t* job = job_picker->picker_iface->next_job(job_picker); + + if (job == NULL) { + break; + } + + jobid = job->jobid; + cbuf_put(&(core->jbuf), &jobid); + } + + jobid = 0; + cbuf_put(&(core->jbuf), &jobid); + + job_picker->picker_iface->end_session(job_picker); + + return NULL; +} + +static float lookup_want(uint64_t jobid) +{ + if (jobid == JOBA_ID) { + return JOBA_WANT; + } + if (jobid == JOBB_ID) { + return JOBB_WANT; + } + if (jobid == JOBC_ID) { + return JOBC_WANT; + } + if (jobid == JOBD_ID) { + return JOBD_WANT; + } + return 0.0f; +} + +/* Builds the wire-shaped xce_job_alloc_t a DISPATCH decision carries -- + * shared by BOTH Part A's real dispatch construction below AND Part B's + * hand-crafted decisions further down -- 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; +} + +/* Builds the accounting-charge shape sched_charge_account()/ + * sched_commit_job_alloc() consume -- same field shape + * test_resource_accounting.c's own make_test_account() establishes, + * parametrized on `residx` (this file's own register_slots_resource() + * result) rather than a hardcoded index. */ +static xce_job_account_t* make_charge_account(const char* host, int residx, + float value) +{ + xce_job_account_t* acct = (xce_job_account_t*)calloc(1, sizeof(*acct)); + + TEST_ASSERT_NOT_NULL(acct); + + acct->nres_shared = 0; + acct->accounts_shared = NULL; + + acct->nhosts = 1; + acct->accounts = (xce_host_account_t**)calloc(1, sizeof(*acct->accounts)); + TEST_ASSERT_NOT_NULL(acct->accounts); + acct->accounts[0] + = (xce_host_account_t*)calloc(1, sizeof(xce_host_account_t)); + TEST_ASSERT_NOT_NULL(acct->accounts[0]); + acct->accounts[0]->host = strdup(host); + acct->accounts[0]->naccount = 1; + acct->accounts[0]->accounts + = (xce_res_account_t**)calloc(1, sizeof(*acct->accounts[0]->accounts)); + TEST_ASSERT_NOT_NULL(acct->accounts[0]->accounts); + acct->accounts[0]->accounts[0] + = (xce_res_account_t*)calloc(1, sizeof(xce_res_account_t)); + TEST_ASSERT_NOT_NULL(acct->accounts[0]->accounts[0]); + acct->accounts[0]->accounts[0]->residx = residx; + acct->accounts[0]->accounts[0]->used = value; + acct->accounts[0]->accounts[0]->reserved = 0.0; + + return acct; +} + +/* Capacity-aware stand-in allocator -- same technique test_sim_resource_ + * contention.c's own capacity_aware_allocator_thread_fn() establishes + * (drives the REAL sched_check_host_satisfy()/sched_charge_account()/ + * sched_commit_job_alloc()), with ONE addition this file's own header + * comment documents: after a real dispatch, job->state is set directly to + * JOB_STAT_RUN. This is required for sched_account_jobs()'s own + * JOB_IS_RUN() guard to recognize the job as accounted-for in a LATER + * round -- without it, round 2's re-charge would skip jobs A/C entirely + * (their `state` would still read JOB_STAT_PEND, since dispatching a job + * via sched_commit_job_alloc()/on_dispatch() alone -- exactly like the + * real production sched_commit_decision() path -- never touches + * job->state; only a separate JOB_STATE_CHANGE mutates it). This mirrors + * what a real system eventually confirms via a worker-reported state + * change, done directly here rather than through the full + * sched_job_state_change()/plugin_job_state_change() machinery (out of + * this file's own boundary -- scheduler.c is not white-box included). */ +static void* capacity_aware_allocator_thread_fn(void* arg) +{ + xce_scheduler_t* sched = (xce_scheduler_t*)arg; + xce_sched_core_t* core = &(sched->core); + xce_sched_plugin_t* picker = NULL; + uint64_t jobid; + + if (sched_resolve_job_picker(core, &picker) < 0) { + xce_decision_t* decision = NULL; + + cbuf_get(&(core->jbuf), &jobid); /* drain the lone 0 sentinel */ + cbuf_put(&(core->decisions), &decision); + return NULL; + } + + for (;;) { + xce_decision_t* decision; + xce_sched_job_t* job; + xce_sched_worker_t* w; + xvec_t uncomitted; + xvec_idval_t wanted; + xvec_idval_t unmeet; + float want; + int rc; + + cbuf_get(&(core->jbuf), &jobid); + if (jobid == 0) { + decision = NULL; + cbuf_put(&(core->decisions), &decision); + break; + } + + job = sched_get_job(core, jobid); + w = sched_get_worker(core, (char*)WORKER_NAME); + TEST_ASSERT_NOT_NULL(job); + TEST_ASSERT_NOT_NULL(w); + + want = lookup_want(jobid); + + 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, g_slots_residx_normal, want); + TEST_ASSERT_TRUE(rc >= 0); + + rc = sched_check_host_satisfy(sched, w, &uncomitted, &wanted, &unmeet); + + xvec_fini(&uncomitted); + xvec_idval_fini(&wanted); + xvec_idval_fini(&unmeet); + + if (rc == 1) { + job->accounts = make_charge_account(WORKER_NAME, + g_slots_residx_normal, want); + sched_commit_job_alloc(core, job); + + job->allocs = make_dispatch_alloc(jobid, WORKER_NAME, want); + + /* This file's own addition -- see this function's header + * comment for why this is required. */ + job->state = JOB_STAT_RUN; + + picker->picker_iface->on_dispatch(picker, jobid); + decision = make_dispatch_decision(job->allocs); + } else { + picker->picker_iface->on_pend(picker, jobid); + decision = make_pend_decision(make_pendreason(jobid)); + } + + if (decision != NULL) { + cbuf_put(&(core->decisions), &decision); + } + } + + return NULL; +} + +static void run_one_real_session(xce_scheduler_t* sched) +{ + pthread_t picker_tid; + pthread_t allocator_tid; + int rc; + + rc = pthread_create(&picker_tid, NULL, picker_thread_fn, &(sched->core)); + TEST_ASSERT_EQUAL_INT(0, rc); + + rc = pthread_create(&allocator_tid, NULL, capacity_aware_allocator_thread_fn, + sched); + TEST_ASSERT_EQUAL_INT(0, rc); + + xce_sched_driver_run_normal_mode_round(sched); + + rc = pthread_join(picker_tid, NULL); + TEST_ASSERT_EQUAL_INT(0, rc); + rc = pthread_join(allocator_tid, NULL); + TEST_ASSERT_EQUAL_INT(0, rc); +} + +/* ====================================================================== + * Part B: hand-crafted `.replay` fixture for simulation-mode reconciliation + * -- same technique test_sim_round_boundary_reconciliation.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). + * ====================================================================== */ + +static void fixture_init_seed(xce_scheduler_t* sched, char* record_dir) +{ + int residx; + + memset(sched, 0, sizeof(*sched)); + mylock_init(&(sched->record_lock)); + + TEST_ASSERT_EQUAL_INT(0, xce_sched_core_init(&(sched->core))); + + sched->core.job_pick_plugin = (char*)"builtin_fcfs"; + sched->record_dir = record_dir; + + residx = register_slots_resource(&(sched->core)); + add_worker(&(sched->core), residx); +} + +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); +} + +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); +} + +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); +} + +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); +} + +static replay_decision_t* make_decision_dispatch(uint64_t ts, + xce_job_alloc_t* alloc) +{ + sched_dispatch_decision_t* dispatch + = (sched_dispatch_decision_t*)calloc(1, sizeof(*dispatch)); + replay_decision_t* d = (replay_decision_t*)calloc(1, sizeof(*d)); + + TEST_ASSERT_NOT_NULL(dispatch); + TEST_ASSERT_NOT_NULL(d); + + dispatch->alloc = alloc; + + d->ts = ts; + d->body_which = ReplayDecision_body_dispatch; + d->body.dispatch = dispatch; + + return d; +} + +static replay_decision_t* make_decision_pend(uint64_t ts, uint64_t jobid, + int main_reason) +{ + sched_pend_decision_t* pend + = (sched_pend_decision_t*)calloc(1, sizeof(*pend)); + xce_job_preason_t* reason + = (xce_job_preason_t*)calloc(1, sizeof(*reason)); + replay_decision_t* d = (replay_decision_t*)calloc(1, sizeof(*d)); + + TEST_ASSERT_NOT_NULL(pend); + TEST_ASSERT_NOT_NULL(reason); + TEST_ASSERT_NOT_NULL(d); + + reason->main_reason = main_reason; + pend->jobid = jobid; + pend->reason = reason; + + d->ts = ts; + d->body_which = ReplayDecision_body_pend; + d->body.pend = pend; + + return d; +} + +static void append_replay_session(FILE* f, uint64_t begin_round, + replay_decision_t** decisions, + int n_decisions) +{ + replay_session_t session; + struct capn c; + struct capn_segment* cs; + ReplaySession_ptr sp; + + memset(&session, 0, sizeof(session)); + session.begin_round = begin_round; + session.decisions = decisions; + session.n_decisions = n_decisions; + 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 central regression test (task 4.1's own Observable Completion). */ +/* ---------------------------------------------------------------------- */ + +static void +test_contention_parity_matches_real_normal_mode_recording_round_by_round( + void) +{ + /* ---- Part A: real, driven normal-mode recording, 2 rounds ---- */ + char record_dir_normal[PATH_MAX]; + xce_scheduler_t sched; + char r_path[PATH_MAX]; + xce_job_t** jobs3; + xce_job_t** jobs1; + xce_sched_worker_t* w_normal; + xce_sched_job_t* jobA_normal; + xce_sched_job_t* jobB_normal; + xce_sched_job_t* jobC_normal; + xce_sched_job_t* jobD_normal; + xce_resource_t* r_normal = NULL; + int residx_normal = -1; + float got; + float normal_resinuse_after_round1; + float normal_resinuse_after_round2; + int rc; + + /* ---- Part B: hand-crafted simulation-mode fixture, 2 rounds ---- */ + char record_dir_sim[PATH_MAX]; + xce_scheduler_t sched_seed; + FILE* f; + char sim_r_path[PATH_MAX]; + xce_job_alloc_t* allocA; + xce_job_alloc_t* allocC; + replay_decision_t* round1_decisions[3]; + replay_decision_t* round2_decisions[1]; + sched_sim_args_t sim_args; + xce_sched_sim_driver_t sim; + xce_sched_worker_t* w_sim; + xce_sched_job_t* jobA_sim; + xce_sched_job_t* jobB_sim; + xce_sched_job_t* jobC_sim; + xce_sched_job_t* jobD_sim; + xce_resource_t* r_sim = NULL; + int residx_sim = -1; + + /* ================================================================== + * Part A + * ================================================================== */ + + make_tmp_dir(record_dir_normal, sizeof(record_dir_normal), "_normal"); + fixture_init_normal(&sched, record_dir_normal); + residx_normal = g_slots_residx_normal; + + rc = sched_record_start(&sched, "contention_parity_normal.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); + + enqueue_queue_update(&sched, "normal", 1); + + jobs3 = (xce_job_t**)calloc(3, sizeof(xce_job_t*)); + jobs3[0] = make_job(JOBA_ID, 1000ULL, "normal", "alice"); + jobs3[1] = make_job(JOBB_ID, 1001ULL, "normal", "bob"); + jobs3[2] = make_job(JOBC_ID, 1002ULL, "normal", "carol"); + enqueue_jobs_push(&sched, jobs3, 3); + + run_one_real_session(&sched); /* round 1: A+C real dispatch, B real deny */ + + jobA_normal = sched_get_job(&(sched.core), JOBA_ID); + jobB_normal = sched_get_job(&(sched.core), JOBB_ID); + jobC_normal = sched_get_job(&(sched.core), JOBC_ID); + TEST_ASSERT_NOT_NULL(jobA_normal); + TEST_ASSERT_NOT_NULL(jobB_normal); + TEST_ASSERT_NOT_NULL(jobC_normal); + TEST_ASSERT_NOT_NULL_MESSAGE( + jobA_normal->accounts, + "sanity: normal mode must have really dispatched job A this round"); + TEST_ASSERT_NULL_MESSAGE( + jobB_normal->accounts, + "sanity: normal mode must have really denied job B this round"); + TEST_ASSERT_NOT_NULL_MESSAGE( + jobC_normal->accounts, + "sanity: normal mode must have really dispatched job C this round"); + + w_normal = sched_get_worker(&(sched.core), (char*)WORKER_NAME); + TEST_ASSERT_NOT_NULL(w_normal); + + rc = key_find(sched.core.resmgr, (void*)SLOTS_RESNAME, (void**)&r_normal); + TEST_ASSERT_TRUE(rc >= 0); + TEST_ASSERT_NOT_NULL(r_normal); + TEST_ASSERT_EQUAL_INT(residx_normal, r_normal->idx); + + got = -1.0f; + rc = get_resvals(&(w_normal->resInUse), 1, &residx_normal, &got); + TEST_ASSERT_EQUAL_INT(1, rc); /* get_resvals() returns nres on success */ + normal_resinuse_after_round1 = got; + TEST_ASSERT_EQUAL_FLOAT_MESSAGE( + JOBA_WANT + JOBC_WANT, normal_resinuse_after_round1, + "sanity: worker-a's real resInUse after round 1 must be A+C's " + "combined charge (8.0+2.0=10.0, exactly saturating capacity)"); + + /* Round 2: push job D. B is genuinely re-offered as a candidate too + * (builtin_fcfs_on_pend()'s own documented "no-op -- retried next + * round" contract), so this round's real session evaluates BOTH B and + * D -- both must be denied, capacity is fully saturated by A+C. */ + jobs1 = (xce_job_t**)calloc(1, sizeof(xce_job_t*)); + jobs1[0] = make_job(JOBD_ID, 2000ULL, "normal", "dave"); + enqueue_jobs_push(&sched, jobs1, 1); + + run_one_real_session(&sched); /* round 2 */ + + jobD_normal = sched_get_job(&(sched.core), JOBD_ID); + TEST_ASSERT_NOT_NULL(jobD_normal); + TEST_ASSERT_NULL_MESSAGE( + jobD_normal->accounts, + "sanity: normal mode must genuinely deny job D too -- capacity is " + "fully saturated by A+C, zero headroom remains"); + TEST_ASSERT_NULL_MESSAGE( + jobB_normal->accounts, + "sanity: job B must still be denied in round 2 too (re-offered, " + "re-evaluated, still no capacity)"); + + got = -1.0f; + rc = get_resvals(&(w_normal->resInUse), 1, &residx_normal, &got); + TEST_ASSERT_EQUAL_INT(1, rc); + normal_resinuse_after_round2 = got; + TEST_ASSERT_EQUAL_FLOAT_MESSAGE( + normal_resinuse_after_round1, normal_resinuse_after_round2, + "sanity: normal mode's resInUse must be unchanged in round 2 -- " + "nothing new was charged (D denied)"); + + rc = sched_record_stop(&sched); + TEST_ASSERT_EQUAL_INT(0, rc); + + /* ================================================================== + * Part B: hand-crafted simulation-mode fixture modeling the SAME facts + * ================================================================== */ + + make_tmp_dir(record_dir_sim, sizeof(record_dir_sim), "_sim"); + fixture_init_seed(&sched_seed, record_dir_sim); + + rc = sched_record_start(&sched_seed, "contention_parity_sim_input.replay"); + TEST_ASSERT_EQUAL_INT(0, rc); + TEST_ASSERT_TRUE(strlen(sched_seed.record_file_path) < sizeof(sim_r_path)); + strcpy(sim_r_path, sched_seed.record_file_path); + f = sched_seed.record_file; + + /* Round 1: jobs A/B/C pushed PEND, then A and C flip to RUN via + * JOB_STATE_CHANGE (the signature of "already resolved before this + * replay's own live picker reaches them", design.md's own Recorded- + * Decision Reconciliation trigger condition) -- BEFORE round 1's own + * ReplaySession, which carries the matching recorded DISPATCH + * decisions for A and C (SAME want-values as Part A's real dispatch: + * JOBA_WANT/JOBC_WANT, never re-typed) plus a genuine PEND for B (no + * reconciliation needed -- sim's own live picker independently PENDs + * it, matching normal mode's real denial). */ + 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_jobs_push_event(f, 1500ULL, JOBC_ID, "normal"); + append_job_state_change_event(f, 1800ULL, JOBA_ID, JOB_STAT_PEND, + JOB_STAT_RUN); + append_job_state_change_event(f, 1800ULL, JOBC_ID, JOB_STAT_PEND, + JOB_STAT_RUN); + + allocA = make_dispatch_alloc(JOBA_ID, WORKER_NAME, JOBA_WANT); + allocC = make_dispatch_alloc(JOBC_ID, WORKER_NAME, JOBC_WANT); + + round1_decisions[0] = make_decision_dispatch(2000ULL, allocA); + round1_decisions[1] = make_decision_dispatch(2000ULL, allocC); + round1_decisions[2] = make_decision_pend(2000ULL, JOBB_ID, 0); + append_replay_session(f, 2000ULL, round1_decisions, 3); + + /* Round 2: job D pushed fresh -- no reconciliation involved (D was + * never dispatched anywhere); sim's own live picker independently + * PENDs it. The recorded decision embedded here (a PEND, matching what + * a real system would ALSO record) is not itself consulted by + * reconciliation or asserted on directly -- it exists only so this + * round has a structurally normal, non-empty ReplaySession boundary + * message. */ + append_jobs_push_event(f, 3000ULL, JOBD_ID, "normal"); + round2_decisions[0] = make_decision_pend(4000ULL, JOBD_ID, 0); + append_replay_session(f, 4000ULL, round2_decisions, 1); + + append_end_of_stream_marker(f); + fclose(f); + + memset(&sim_args, 0, sizeof(sim_args)); + sim_args.replay_path = sim_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)); + g_capturing = 1; + g_round_idx = 0; + rc = sched_sim_driver_run(&sim, &sim_args); + g_capturing = 0; + TEST_ASSERT_EQUAL_INT(0, rc); + + TEST_ASSERT_EQUAL_INT_MESSAGE( + 2, g_round_idx, + "sched_account_jobs() must have been called exactly once per " + "replayed round (2 rounds in this fixture) -- if this isn't 2, " + "the round-boundary snapshot mechanism itself is miscounting, " + "independent of the parity assertions below"); + + /* ================================================================== + * Part C: the actual parity assertions + * ================================================================== */ + + /* Requirement 1.4: sim mode's round-by-round resInUse must match what + * the real normal-mode recording observed at the SAME point -- not + * just at the final round. */ + TEST_ASSERT_EQUAL_FLOAT_MESSAGE( + normal_resinuse_after_round1, g_round_resinuse[0], + "round 1: simulation mode's reconciled resInUse must byte-for-byte " + "match what the real normal-mode recording actually observed after " + "its own round 1 (jobs A+C dispatched, consuming 10.0 of 10.0 " + "total capacity)"); + TEST_ASSERT_EQUAL_FLOAT_MESSAGE( + normal_resinuse_after_round2, g_round_resinuse[1], + "round 2: simulation mode's resInUse must still match normal " + "mode's -- the already-consumed capacity must PERSIST correctly " + "into a SECOND round, not just the first"); + + /* Requirement 2.1/2.2 (decision-level): job A/C dispatched with the + * SAME allocation amount, job B/D denied, in BOTH modes. */ + jobA_sim = sched_get_job(&(sim.core), JOBA_ID); + jobB_sim = sched_get_job(&(sim.core), JOBB_ID); + jobC_sim = sched_get_job(&(sim.core), JOBC_ID); + jobD_sim = sched_get_job(&(sim.core), JOBD_ID); + TEST_ASSERT_NOT_NULL(jobA_sim); + TEST_ASSERT_NOT_NULL(jobB_sim); + TEST_ASSERT_NOT_NULL(jobC_sim); + TEST_ASSERT_NOT_NULL(jobD_sim); + + TEST_ASSERT_NOT_NULL_MESSAGE( + jobA_sim->accounts, + "simulation mode must have reconciled job A's dispatch, matching " + "normal mode's real dispatch decision"); + TEST_ASSERT_NULL_MESSAGE( + jobB_sim->accounts, + "simulation mode must show job B denied, matching normal mode's " + "real denial (dispatch where normal pended, or vice versa, must " + "never happen)"); + TEST_ASSERT_NOT_NULL_MESSAGE( + jobC_sim->accounts, + "simulation mode must have reconciled job C's dispatch, matching " + "normal mode's real dispatch decision"); + TEST_ASSERT_NULL_MESSAGE( + jobD_sim->accounts, + "simulation mode must show job D denied too, matching normal " + "mode's real denial in round 2"); + + /* NOTE: the reconciled xce_job_account_t is built by the REAL + * production make_job_account()/make_res_account() (xcealloc.c), which + * stores the dispatched amount into .reserved/.allocated (and leaves + * .used == 0.0 until sched_charge_account() actually charges it into + * the WORKER's resInUse -- .used on the per-job xce_res_account_t + * itself is never separately updated by that charge). This differs + * from Part A's own make_charge_account() helper above (a test-only + * construction that sets .used directly) -- check .reserved here, + * matching make_res_account()'s own field mapping + * (account->reserved = alloc->value). */ + TEST_ASSERT_EQUAL_FLOAT_MESSAGE( + JOBA_WANT, jobA_sim->accounts->accounts[0]->accounts[0]->reserved, + "job A's reconciled charge amount must match the real dispatch's " + "own amount (8.0)"); + TEST_ASSERT_EQUAL_FLOAT_MESSAGE( + JOBC_WANT, jobC_sim->accounts->accounts[0]->accounts[0]->reserved, + "job C's reconciled charge amount must match the real dispatch's " + "own amount (2.0)"); + + /* Requirement 2's capacity-awareness, proven directly against BOTH + * drivers' independently-derived post-round state via the SAME real, + * unmodified sched_check_host_satisfy() -- mirroring + * test_sim_round_boundary_reconciliation.c's own technique, run here + * against BOTH sides, not just sim's. A hypothetical job wanting + * JOBB_WANT (5.0 -- the SAME amount the real recording genuinely + * denied, reused rather than inventing a new number) must be denied + * against BOTH post-round-2 states: if reconciliation were broken + * (job A/C never reconciled, resInUse wrongly reads 0.0), this SAME + * check would wrongly GRANT it on the simulation side -- this is + * this test's own discriminating-power proof (see this file's header + * comment; independently confirmed live by reverting the production + * fix). */ + { + xce_scheduler_t wrapper_sim; + xvec_t uncomitted; + xvec_idval_t wanted; + xvec_idval_t unmeet; + + w_sim = sched_get_worker(&(sim.core), (char*)WORKER_NAME); + TEST_ASSERT_NOT_NULL(w_sim); + + rc = key_find(sim.core.resmgr, (void*)SLOTS_RESNAME, (void**)&r_sim); + TEST_ASSERT_TRUE(rc >= 0); + TEST_ASSERT_NOT_NULL(r_sim); + residx_sim = r_sim->idx; + + memset(&wrapper_sim, 0, sizeof(wrapper_sim)); + wrapper_sim.core = sim.core; + + /* Normal mode's own post-round-2 state. */ + 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_normal, JOBB_WANT); + TEST_ASSERT_TRUE(rc >= 0); + + rc = sched_check_host_satisfy(&sched, w_normal, &uncomitted, &wanted, + &unmeet); + TEST_ASSERT_EQUAL_INT_MESSAGE( + 0, rc, + "normal mode's real post-round-2 state must deny a hypothetical " + "job wanting 5.0 -- zero headroom remains"); + + xvec_fini(&uncomitted); + xvec_idval_fini(&wanted); + xvec_idval_fini(&unmeet); + + /* Simulation mode's own post-replay state -- the SAME real, + * unmodified capacity-check function, against sim's OWN + * independently-derived resInUse. */ + 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_sim, JOBB_WANT); + TEST_ASSERT_TRUE(rc >= 0); + + rc = sched_check_host_satisfy(&wrapper_sim, w_sim, &uncomitted, + &wanted, &unmeet); + TEST_ASSERT_EQUAL_INT_MESSAGE( + 0, rc, + "simulation mode's post-replay state must ALSO deny a " + "hypothetical job wanting 5.0 -- matching normal mode's own " + "denial above, proving Requirement 2's capacity-awareness " + "holds identically in both drivers"); + + xvec_fini(&uncomitted); + xvec_idval_fini(&wanted); + xvec_idval_fini(&unmeet); + } + + printf("test_contention_parity_matches_real_normal_mode_recording_" + "round_by_round: PASS\n"); +} + +int main(void) +{ + UNITY_BEGIN(); + RUN_TEST( + test_contention_parity_matches_real_normal_mode_recording_round_by_round); + return UNITY_END(); +}