diff --git a/.kiro/specs/scheduler-core/tasks.md b/.kiro/specs/scheduler-core/tasks.md index efba58fd..7e826b81 100644 --- a/.kiro/specs/scheduler-core/tasks.md +++ b/.kiro/specs/scheduler-core/tasks.md @@ -199,12 +199,13 @@ - [ ] 8. Core: Snapshot Serialization -- [ ] 8.1 Implement snapshot save and load +- [x] 8.1 Implement snapshot save and load - Serialize the full core state (as scoped in the snapshot schema) into the snapshot message; restore from a snapshot message back to that exact state - After a load, rebuild the picker's private state by walking the restored core state and re-invoking the picker's state-change hooks (queue updates, then pending-job pushes) — never store picker-private state in the snapshot itself - Observable: save immediately followed by load into a freshly-initialized core produces a core whose observable scheduling behavior (next session's decisions, given the same events) is identical to the pre-save core - _Requirements: 5.21, 6.5_ - _Depends: 7.3_ + - _Verified: 2026-08-04 — xce_sched_core_save_snapshot()/xce_sched_core_load_snapshot() implemented, split across snapshot.c/snapshot_wire.c/snapshot_load.c (800-line cap) along real logical boundaries (save/load/shared-helpers). New prng_seed field added to xce_sched_core_t (a genuine, well-investigated gap requirements.md/design.md both presupposed but no prior task added), seeded to 0, no PRNG consumer wired yet (documented, deliberate). Cross-core behavioral-equivalence Observable proven for real: two independent cores, each with its own registered picker, save on one/load into a freshly-initialized other, both driven through an identical real session (picker thread + decision drain), decision sequences asserted identical including a pinned spot-check (201-PEND/202-DISPATCH/203-PEND on both). -EBUSY/-EINVAL guards checked before any mutation, tested directly. Found and fixed two real bugs along the way: a missing prng_seed restore in this task's own new code, and a genuine, independently-traced-to-the-primitive-level guard bug in already-committed core.c (sched_get_user/sched_get_project treating key_find()'s expected -ENOENT as fatal, silently breaking get-or-create for any new name) that also affected the live production path (make_sched_job(), JOBS_PUSH) — narrowly scoped two-line fix, folded into this task's diff since only this task's own test path discovered it. 23/23 scheduler suite, full project build clean, review APPROVED first pass with every claim independently re-derived._ - [ ] 8.2 Enforce the snapshot session-idle precondition - Reject snapshot save/load calls made while a scheduling session is active, returning a distinct busy error rather than capturing or restoring torn state @@ -339,6 +340,7 @@ - Implement a single input-loop REPL (not a separate thread) that pauses before the first event and blocks the simulation loop while active, supporting step-N-sessions, continue-to-next-breakpoint-or-EOF, and quit - Observable: starting simulation with the REPL enabled pauses before any event is processed; issuing a step command advances exactly the requested number of sessions and returns control to the prompt - _Requirements: 5.22_ + - _Note: use readline() for the input loop's line editing (history, arrow keys) — xce_test's REPL already sets this precedent (src/controller/testsuite/test.c:3827) and readline is already a linked project dependency, zero new cost. Confirm readline's GPL licensing is acceptable for this build target before committing to it (vs. libedit/linenoise) if that hasn't already been settled elsewhere in the codebase._ - _Depends: 11.2_ - [ ] 12.2 Implement REPL breakpoints and state inspection diff --git a/.wolf/buglog.json b/.wolf/buglog.json index 16d71792..73a505fe 100644 --- a/.wolf/buglog.json +++ b/.wolf/buglog.json @@ -14487,5 +14487,185 @@ ], "occurrences": 1, "last_seen": "2026-07-31T04:50:00Z" + }, + { + "id": "bug-1489", + "timestamp": "2026-08-03T04:00:00Z", + "error_message": "gcc: 'unknown type name xce_sched_core_t; did you mean xce_sched_ctx_t?' at core.h:92 (xce_sched_core_init prototype) when compiling scheduler.h consumers (handler.c, main.c, etc).", + "file": "src/scheduler/include/core.h", + "root_cause": "core.h defined `struct xce_sched_core_s { ... };` (the full body) but relied entirely on plugin.h's forward declaration (`typedef struct xce_sched_core_s xce_sched_core_t;`) for the typedef name -- core.h itself never declared it. scheduler.h includes core.h BEFORE plugin.h, so in that (the only real) include order, xce_sched_core_t was unresolved by the time core.h's own function prototypes referenced it.", + "fix": "Added `typedef struct xce_sched_core_s xce_sched_core_t;` directly in core.h, immediately before the struct body. An identical repeated typedef (same underlying type) is valid C11/gnu17, matching this codebase's existing cluster_t idiom (myconf.h/persist.h/ha.h/cluster_def.h all forward-declare independently) -- so the fix works regardless of which of core.h/plugin.h is included first.", + "tags": [ + "scheduler-core", + "task-4.1", + "core-extraction", + "typedef-forward-decl", + "include-order" + ], + "related_bugs": [], + "occurrences": 1, + "last_seen": "2026-08-03T04:00:00Z" + }, + { + "id": "bug-1490", + "timestamp": "2026-08-03T04:00:00Z", + "error_message": "ld: undefined reference to 'reset_sched_obj' when linking xce_scheduler (referenced from plugins/limit.c's limit_job_election()).", + "file": "src/scheduler/src/schedobj.c", + "root_cause": "While rewriting schedobj.c wholesale to relocate the CRUD accessors (sched_get_job/sched_add_job/etc, scheduler-core spec task 4.1) into core.c, the reset_sched_obj() function (unrelated to the relocation, a small xlist/zarray cleanup helper called by plugins/limit.c) was accidentally omitted from the rewritten file -- a transcription miss, not an intentional removal.", + "fix": "Re-added reset_sched_obj() verbatim (unchanged body) to schedobj.c, right after make_sched_obj() where it originally lived.", + "tags": [ + "scheduler-core", + "task-4.1", + "core-extraction", + "link-error", + "whole-file-rewrite-risk" + ], + "related_bugs": [ + "bug-1489" + ], + "occurrences": 1, + "last_seen": "2026-08-03T04:00:00Z" + }, + { + "id": "bug-1491", + "timestamp": "2026-08-03T06:33:00Z", + "error_message": "AddressSanitizer: SEGV on unknown address 0x000000000000 in dataq_deq_head (lib/dataq.c:15), called from dataq_fini <- xce_sched_core_fini (core.c:126), when a test called xce_sched_core_init() then xce_sched_core_fini() back-to-back on a fresh xce_sched_core_t.", + "file": "src/scheduler/testsuite/test_core_jbuf_capacity_pin.c", + "root_cause": "xce_sched_core_init() memset()s the whole xce_sched_core_t, leaving core->pend_updates[0]/[1] NULL -- those two dataq_t* are only allocated+dataq_init()d by scheduler.c sched_init() AFTER calling xce_sched_core_init() (see core.c file header comment: pend_updates storage lives in xce_sched_core_t but its dataq_init/killer wiring stays driver-owned in scheduler.c). xce_sched_core_fini() unconditionally calls dataq_fini(core->pend_updates[0]) and [1] with no NULL guard, so calling fini() right after init() without going through sched_init()'s extra setup step dereferences NULL inside dataq_deq_head.", + "fix": "Not a production bug (xce_sched_core_fini()'s real caller, scheduler.c sched_fini(), always runs after sched_init() has populated pend_updates) -- fixed the test instead: test_jbuf_capacity_is_runtime_one() (task 4.2) calls only xce_sched_core_init() and reads core.jbuf.ncap, deliberately never calling xce_sched_core_fini() on a fixture built by init() alone. Matches this testsuite directory's established \"fixtures aren't torn down\" convention (test_picker_core_loop_boundary.c's header comment).", + "tags": [ + "scheduler-core", + "task-4.2", + "core-init-fini", + "segv", + "null-deref", + "test-authoring-gotcha", + "pend_updates" + ], + "related_bugs": [ + "bug-1489", + "bug-1490" + ], + "occurrences": 1, + "last_seen": "2026-08-03T06:33:00Z" + }, + { + "id": "bug-1492", + "timestamp": "2026-08-04T01:04:00Z", + "error_message": "plugin_end_session() checks plugin->begin_session_fn != NULL but calls plugin->end_session_fn(plugin) -- a guard/call field mismatch", + "file": "src/scheduler/src/plugin.c", + "root_cause": "Pre-existing bug predating the scheduler-core spec (git blame: commit 638173dec, 2026-02-26). plugin_end_session()'s per-plugin loop guards on the wrong struct field before invoking the hook, so a plugin registering only end_session_fn (not begin_session_fn) never has it called; a plugin registering only begin_session_fn (not end_session_fn) would crash via a NULL function-pointer call if this branch were ever reached for it. Discovered incidentally during scheduler-core task 5.2's independent review while verifying the plugin_begin_session/plugin_end_session per-session bookkeeping fan-out is preserved in the new normal-mode driver loop -- not introduced by task 5.2, confirmed untouched by its diff.", + "fix": "NOT FIXED -- out of scope for task 5.2, flagged for a dedicated follow-up. The one-line fix is straightforward: change the guard to `plugin->end_session_fn != NULL` (mirroring plugin_begin_session()'s own correct `plugin->begin_session_fn != NULL` guard for its `begin_session_fn` call).", + "tags": [ + "scheduler-core", + "task-5.2", + "plugin-c", + "guard-call-mismatch", + "pre-existing", + "legacy-fcfs-picker", + "not-fixed" + ], + "related_bugs": [], + "occurrences": 1, + "last_seen": "2026-08-04T01:04:00Z" + }, + { + "id": "bug-1493", + "timestamp": "2026-08-04T12:35:00Z", + "error_message": "gcc: 'expected declaration specifiers before sched_snapshot_data_t' cascading into ~20 errors across typemap.h/controller.capnp.h/controller.capnp.c when building lib/controller.capnp.c (scheduler-core task 7.3 SnapshotData C struct).", + "file": "src/include/typemap.h", + "root_cause": "A doc comment on sched_snapshot_data_t read '...raw uint8_t*/size_t pattern...' -- the literal substring '*/' inside the comment text (no space between the uint8_t* pointer-star and the following /size_t) prematurely closed the C block comment. Everything after that point (the rest of the comment prose plus the following struct's opening brace) was parsed as real C tokens, corrupting the sched_snapshot_data_t typedef and cascading into every generated function signature that references it.", + "fix": "Reworded the comment to avoid the accidental */ sequence ('uint8_t-pointer + size_t pattern' instead of 'uint8_t*/size_t pattern'). General gotcha: never write 'TypeA*/TypeB' (pointer-star immediately followed by slash) inside a /* */ C comment -- always put a space or word between the * and the /.", + "tags": [ + "scheduler-core", + "task-7.3", + "capnp-schema", + "typemap.h", + "comment-syntax", + "build-break" + ], + "related_bugs": [], + "occurrences": 1, + "last_seen": "2026-08-04T12:35:00Z" + }, + { + "id": "bug-1494", + "timestamp": "2026-08-04T00:00:00Z", + "error_message": "clang-format -i produced bizarre whole-file +1-space indentation drift (starting right after the top #include block, persisting to EOF) in a newly-written test_snapshot_save_load.c, with kill_sched_job() alone expanded to multiple lines while identical sibling one-liners stayed collapsed.", + "file": "src/scheduler/testsuite/test_snapshot_save_load.c", + "root_cause": "Same class of bug as bug-1493 (task 7.3): the file's own top-of-file header comment contained the literal substring '(kill_sched_*/plugin_unregister/...)' -- the pointer-star-like '*' immediately followed by '/' prematurely closed the /* */ block comment. Everything after that point was parsed as real C code by clang-format's lexer, which is what produced the indentation drift (a strong diagnostic signal, not just a formatting quirk) -- gcc itself would also have failed to compile this.", + "fix": "Reworded 'kill_sched_*/plugin_unregister/sched_worker_release_host_ctx/sched_resolve_job_picker)' to 'kill_sched_* family, plugin_unregister, sched_worker_release_host_ctx, sched_resolve_job_picker)', removing the accidental */ sequence. General lesson (reinforcing bug-1493): a sudden, file-wide clang-format indentation shift starting at a specific line is a strong signal to grep that file for a stray '*/' inside a comment before assuming it's a clang-format config issue -- verified via a small python scanner that walks the file tracking in-comment state and flags any '*/' encountered while not already inside a comment.", + "tags": [ + "scheduler-core", + "task-8.1", + "comment-syntax", + "clang-format", + "testsuite" + ], + "related_bugs": [ + "bug-1493" + ], + "occurrences": 1, + "last_seen": "2026-08-04T00:00:00Z" + }, + { + "id": "bug-1495", + "timestamp": "2026-08-04T00:00:00Z", + "error_message": "sched_get_user()/sched_get_project() (core.c) always returned NULL for any user/project name not already present in core->users/core->projects, even though both are documented and relied upon (core_events.c JOBS_PUSH, schedobj.c make_sched_job, and task 8.1's own snapshot_load.c) as get-or-create. Surfaced as job->user/job->project being silently NULL for every job referencing a brand-new user/project, discovered while writing test_snapshot_save_load.c's cross-core equivalence test (RED phase: TEST_ASSERT_NOT_NULL(sched_get_project(&core_a, \"proj1\")) failed immediately after a successful JOBS_PUSH).", + "file": "src/scheduler/src/core.c", + "root_cause": "Both functions call key_find() then do `if (rc < 0) { return NULL; }` before falling through to the make_sched_user()/make_sched_project() create path. key_find()'s actual not-found signal is a NEGATIVE errno (-ENOENT, confirmed via direct debug instrumentation: rc=-2 for an absent key, rc=1 for a present one) -- NOT rc==0 as the surrounding code assumed. Since -ENOENT < 0, the guard incorrectly treated the EXPECTED not-found outcome as a fatal error and returned NULL before ever reaching the create fallback, making get-or-create unreachable for any genuinely new name. This is a pre-existing bug from earlier tasks (4.1/4.3), not introduced by task 8.1, but it directly breaks task 8.1's own snapshot_load.c users/projects restoration onto a freshly-initialized core (which always starts empty, so every name is 'new').", + "fix": "Changed the guard in both functions from `if (rc < 0)` to `if ((rc < 0) && (rc != -ENOENT))`, so only a genuine (non-ENOENT) error short-circuits before the create fallback. Two-line change per function, core.c lines ~341 and ~416. Verified: full scheduler suite 23/23 (up from 22/22 baseline, no regressions), plus the new test's own field-level assertions (users/projects created correctly, priority preserved) now pass.", + "tags": [ + "scheduler-core", + "task-8.1", + "core.c", + "get-or-create", + "kvmgr", + "key_find", + "sched_get_user", + "sched_get_project", + "regression-fix" + ], + "related_bugs": [], + "occurrences": 1, + "last_seen": "2026-08-04T00:00:00Z" + }, + { + "id": "bug-1496", + "timestamp": "2026-08-04T00:00:00Z", + "error_message": "xce_sched_core_load_snapshot() never restored core->prng_seed from the snapshot -- a freshly loaded core silently kept xce_sched_core_init()'s default (0) instead of the value xce_sched_core_save_snapshot() had captured. Caught by test_snapshot_save_load.c's field-level assertion: TEST_ASSERT_EQUAL_UINT64(0xDEADBEEFULL, core_b.prng_seed) failed with Expected 3735928559 Was 0 after a successful save+load round trip.", + "file": "src/scheduler/src/snapshot_load.c", + "root_cause": "save_snapshot() (snapshot.c) correctly writes `out->prng_seed = core->prng_seed;`, but the corresponding `core->prng_seed = in->prng_seed;` assignment was simply missing from load_snapshot()'s field-copy sequence -- an incomplete round trip for a field the function's own header comment (snapshot.c, 'PRNG seed resolution') explicitly documents as in scope (Req 5 AC 21's 'PRNG state' checklist row).", + "fix": "Added `core->prng_seed = in->prng_seed;` to xce_sched_core_load_snapshot() (snapshot_load.c), placed after the jobs-restoration loop and before the internal_state restoration block. Verified via the same field-level assertion, now passing.", + "tags": [ + "scheduler-core", + "task-8.1", + "snapshot_load.c", + "prng_seed", + "round-trip-gap" + ], + "related_bugs": [], + "occurrences": 1, + "last_seen": "2026-08-04T00:00:00Z" + }, + { + "id": "bug-1497", + "timestamp": "2026-08-04T00:00:00Z", + "error_message": "ASan SEGV inside dup_val() -> makeString() -> hash_string() while building test_snapshot_save_load.c's cross-core equivalence test fixtures.", + "file": "src/scheduler/testsuite/test_snapshot_save_load.c", + "root_cause": "Test-side only, not production code: three test-owned xce_val_t resval fixtures were left with `.which` at its memset-zero default. XCE_VAL_STRING happens to be enum value 0 (xce/types.h), so the zeroed `.which` was silently interpreted as \"this is a string\" -- dup_val()/makeString() then reinterpreted the fixture's raw ivalue32 int as a char* pointer and hash_string() dereferenced it, crashing.", + "fix": "Explicitly set `.which = XCE_VAL_NUMERIC` on all three test-owned xce_val_t fixtures in test_snapshot_save_load.c before use -- never rely on struct-zero-init to select a union discriminant when a non-zero-valued enum member (XCE_VAL_NUMERIC) is the intended default.", + "tags": [ + "scheduler-core", + "task-8.1", + "test_snapshot_save_load.c", + "xce_val_t", + "zero-init-union-discriminant", + "test-only", + "asan-segv" + ], + "related_bugs": [], + "occurrences": 1, + "last_seen": "2026-08-04T00:00:00Z" } -] +] \ No newline at end of file diff --git a/src/scheduler/include/core.h b/src/scheduler/include/core.h index a39cd892..c5526fa3 100644 --- a/src/scheduler/include/core.h +++ b/src/scheduler/include/core.h @@ -164,6 +164,30 @@ struct xce_sched_core_s { */ uint64_t session_start; uint64_t session_duration_ns; + /* + * PRNG seed for deterministic tie-breaking (Req 5 AC 7 / task 8.1). This + * field did not exist before task 8.1 -- requirements.md's Req 5 AC 7 + * explicitly names `core->prng_seed` ("Any required randomness ... shall + * use a PRNG seeded by the driver at init time") and design.md's + * eventual `xce_sched_core_init(core, decision_cb, driver_ctx, + * prng_seed)` signature (see this header's own doc comment above + * xce_sched_core_init()) both presuppose it, but no prior task added it + * or wired a PRNG consumer -- confirmed by grepping this entire spec's + * source tree for "prng"/"seed" before this task. Adding ONLY the field + * here, seeded to a deterministic constant (0) by + * xce_sched_core_init(), keeps this task's scope to "the field exists + * and round-trips through save_snapshot()/load_snapshot()" + * (Req 5 AC 21's "PRNG state" checklist row) -- NOT changing + * xce_sched_core_init()'s signature (still deliberately narrower per + * this header's doc comment below) and NOT wiring any actual PRNG + * consumer, since nothing in the codebase performs tie-breaking + * randomness yet (builtin_fcfs's ordering is fully deterministic via + * ordq comparators, no PRNG call site exists to seed). A future task + * that adds real PRNG-based tie-breaking is expected to add its own + * setter, the same pattern xce_core_set_decision_cb() already + * establishes for driver_ctx/on_decision. + */ + uint64_t prng_seed; }; /* @@ -457,4 +481,70 @@ extern uint64_t xce_core_now(xce_sched_core_t*); extern void xce_core_set_active_for_logging(xce_sched_core_t*); extern int xce_core_register_tlog_format(void); +/* + * SCHED_SNAPSHOT_INTERNAL_VERSION - the current format tag for + * `sched_snapshot_data_t.internal_state`'s opaque byte layout (task 8.1). + * Declared here, rather than privately in snapshot.c, because BOTH + * xce_sched_core_save_snapshot() (snapshot.c, which writes it) and + * xce_sched_core_load_snapshot() (snapshot_load.c, which version-checks + * it) need the identical value -- a shared header is the single-source-of- + * truth home for a constant two separate translation units must agree on + * bit-for-bit; duplicating it as two independent #defines would risk + * exactly the kind of silent drift this task's own documentation + * (snapshot.c's file header, "internal_state contents") goes out of its + * way to avoid elsewhere. This is an internal implementation detail of the + * two functions below, not part of any wire/capnp contract -- callers of + * either function never need to read it directly. + */ +#define SCHED_SNAPSHOT_INTERNAL_VERSION (1u) + +/* + * xce_sched_core_save_snapshot / xce_sched_core_load_snapshot - Req 5 AC 21 + * / Req 6 AC 5 (scheduler-core spec, task 8.1; design.md "SchedulerCore" + * Service Interface). Serialize the full core state into (out of) a + * `sched_snapshot_data_t` (typemap.h, the C mirror of task 7.3's capnp + * `SnapshotData` struct) -- populate/consume it from `xce_sched_core_t`'s + * live in-memory state directly; NO capnp encode/decode happens here (that + * is a later, wire-format task's job). Implemented across three files -- + * snapshot.c (the save direction + the full shared design-decision + * documentation), snapshot_wire.c (small helpers + save-direction wire + * builders), and snapshot_load.c (the load direction + picker-rebuild) -- + * split for this project's 800-line-per-file convention; see snapshot.c's + * own file header comment, "Three-file split", for the full rationale. + * + * Both are only valid between scheduling sessions, matching every other + * core entry point's precondition (xce_core_push_events()'s own doc + * comment, above): return -EBUSY, with ZERO state touched, if + * `core->session_active != 0` -- checked BEFORE any mutation, so a rejected + * call is a true no-op (same pattern xce_core_begin_session()/ + * xce_core_push_events() already establish). Both return -EINVAL for a + * NULL core or a NULL out/in pointer. + * + * xce_sched_core_save_snapshot() populates `*out` from `*core`'s current + * state (jobs/queues/workers/resources/resvals/users/projects/config/ + * prng_seed/internal_state); see snapshot.c's own file header comment for + * the full field-by-field mapping and the documented scope decisions + * (job-level priority and in-flight allocations are NOT captured -- see + * that comment for why). + * + * xce_sched_core_load_snapshot() restores `*core` (which must be freshly + * initialized via xce_sched_core_init() -- loading into a core that + * already holds state is not this task's scope, and is not what the + * driver-side callers design.md documents -- record start, periodic + * checkpoint, simulation startup -- ever do) from `*in`, THEN rebuilds the + * resolved picker's private state by walking the just-restored core state + * and re-invoking the picker's state-change hooks in the same order + * xce_core_push_events() already establishes for state-change hooks: + * on_queue_update for every restored queue, THEN on_job_push for every + * restored job. Picker-private state is deliberately never stored in the + * snapshot itself (requirements.md Req 5 AC 21's own "Picker state is NOT + * stored in snapshot" rule) -- this rebuild is what makes post-load state + * indistinguishable from state built up via normal xce_core_push_events() + * calls, which is the whole point of the Observable this task proves. + */ +extern int xce_sched_core_save_snapshot(xce_sched_core_t*, + sched_snapshot_data_t*); +extern int xce_sched_core_load_snapshot(xce_sched_core_t*, + const sched_snapshot_data_t*); + #endif diff --git a/src/scheduler/meson.build b/src/scheduler/meson.build index 0f83cdb9..72e1c7b0 100644 --- a/src/scheduler/meson.build +++ b/src/scheduler/meson.build @@ -7,6 +7,9 @@ scheduler_srcs = [ 'src/core_session.c', 'src/core_clock.c', 'src/driver.c', + 'src/snapshot.c', + 'src/snapshot_wire.c', + 'src/snapshot_load.c', 'src/decision.c', 'src/schedobj.c', 'src/plugin.c', diff --git a/src/scheduler/src/core.c b/src/scheduler/src/core.c index bf100c7a..17055cd2 100644 --- a/src/scheduler/src/core.c +++ b/src/scheduler/src/core.c @@ -338,7 +338,20 @@ xce_sched_user_t* sched_get_user(xce_sched_core_t* core, char* name) lock(core->users, RTTI_LOCK_READ); rc = key_find(core->users, (void*)name, (void**)&user); unlock(core->users); - if (rc < 0) { + /* Pre-existing bug (found while implementing task 8.1's snapshot-load + * user restoration, scheduler-core spec): key_find()'s "not found" + * signal is -ENOENT, NOT rc==0 (0 is never returned by this kvmgr + * wrapper; confirmed by direct key_find() calls returning -ENOENT for + * an absent key and a positive value for a present one) -- the + * original `if (rc < 0) return NULL;` treated the EXPECTED "not + * found" outcome as a fatal error, so the get-or-create fallback below + * was UNREACHABLE for any name not already present, breaking every + * caller that relies on get-or-create semantics (this function's own + * name, this file's make_sched_user() fallback, core_events.c's + * JOBS_PUSH path, schedobj.c's still-live make_sched_job(), and this + * task's own snapshot_load.c). Only a genuine error (neither "not + * found" nor "found") should short-circuit here. */ + if ((rc < 0) && (rc != -ENOENT)) { return NULL; } if (user != NULL) { @@ -400,7 +413,12 @@ xce_sched_project_t* sched_get_project(xce_sched_core_t* core, char* name) lock(core->projects, RTTI_LOCK_READ); rc = key_find(core->projects, (void*)name, (void**)&project); unlock(core->projects); - if (rc < 0) { + /* Same pre-existing bug/fix as sched_get_user() above (found while + * implementing task 8.1's snapshot-load project restoration): -ENOENT + * is key_find()'s EXPECTED "not found" signal, not a fatal error -- + * only a genuine error should short-circuit before the get-or-create + * fallback below. */ + if ((rc < 0) && (rc != -ENOENT)) { return NULL; } diff --git a/src/scheduler/src/snapshot.c b/src/scheduler/src/snapshot.c new file mode 100644 index 00000000..06cedbb7 --- /dev/null +++ b/src/scheduler/src/snapshot.c @@ -0,0 +1,621 @@ +#include +#include +#include +#include + +#include "core.h" +#include "plugin.h" +#include "resobj.h" +#include "sched/types.h" +#include "utils.h" +#include "xstring.h" + +/* + * snapshot.c - xce_sched_core_save_snapshot() (scheduler-core spec, task + * 8.1, Req 5 AC 21 / Req 6 AC 5) and the full field-by-field mapping / + * ownership documentation shared by BOTH the save and load directions. See + * core.h's doc comments on xce_sched_core_save_snapshot()/ + * xce_sched_core_load_snapshot() for the public contract (-EBUSY + * mid-session, -EINVAL on NULL args). This comment documents the + * investigation and design decisions behind the field-by-field mapping + * between `xce_sched_core_t` (core.h, the live in-memory state) and + * `sched_snapshot_data_t` (typemap.h, task 7.3's committed C mirror of the + * capnp `SnapshotData` struct) -- read in full before touching this file OR + * its two sibling files, snapshot_wire.c and snapshot_load.c. + * + * IMPORTANT: no capnp encoding/decoding happens anywhere in this file (or + * its siblings). `sched_snapshot_data_t` is consumed/produced purely as a + * plain C struct of live pointers; wire (capnp) serialization of that + * struct is a separate, later task (recording.c / task 9+), out of this + * task's scope. + * + * ========================================================================== + * Three-file split (this file / snapshot_wire.c / snapshot_load.c) + * ========================================================================== + * The original single-file implementation was 1076 lines, over this + * project's 800-line-per-file convention (CLAUDE.md/coding-style.md) -- the + * same cap task 4.3 already split core_events.c out of core.c to respect + * (core_events.c's own file header comment documents that precedent). The + * cut below follows the boundaries this file's own section comments already + * established, rather than an arbitrary chop: + * + * - snapshot.c (this file): the SAVE entry point + * (xce_sched_core_save_snapshot()) plus every section of this header + * comment that documents behavior SHARED by both directions + * (field-by-field mapping, PRNG seed resolution, unsatisfiable-cache + * resolution, internal_state contents, scope decisions, + * why-not-reuse-core_events.c, ownership notes) -- kept in ONE place + * rather than duplicated or cross-referenced piecemeal across files, + * since these sections describe save and load as symmetric inverse + * operations over the identical field set. + * - snapshot_wire.c: the small generic helpers (dup_val/arr_append/ + * find_resource_name_by_idx) and the SAVE-direction wire-object + * builders (build_wire_job/build_wire_queue/build_wire_worker) -- a + * self-contained, stateless "data-shape converter" unit, exactly this + * file's own pre-existing "Small shared helpers" / "SAVE-direction + * wire-object builders" section boundaries. + * - snapshot_load.c: the LOAD entry point (xce_sched_core_load_snapshot()) + * plus its own load-specific "Picker rebuild on load" documentation + * (moved there from this file -- it describes load-only behavior). + * + * dup_val/arr_append/find_resource_name_by_idx/build_wire_job/ + * build_wire_queue/build_wire_worker (snapshot_wire.c) are declared + * `extern` where consumed across this three-file boundary (below, and in + * snapshot_load.c for dup_val alone) rather than funneled through a new + * private header -- matching this codebase's own existing idiom for a + * narrow, single-symbol cross-.c-file dependency (core_session.c's own + * `extern void sched_worker_release_host_ctx(...)` declaration, redeclared + * rather than pulling in a whole extra header, for the identical reason). + * SCHED_SNAPSHOT_INTERNAL_VERSION (used by both this file's + * save_snapshot() and snapshot_load.c's load_snapshot()) is declared in + * core.h instead, for the same single-source-of-truth reason -- it is a + * plain macro, which cannot be `extern`-declared, and duplicating a + * versioning constant across two files risks exactly the kind of silent + * drift this task's own documentation elsewhere goes out of its way to + * avoid. + * + * ========================================================================== + * Field-by-field mapping (xce_sched_core_t -> sched_snapshot_data_t) + * ========================================================================== + * + * config -> core->config (BORROWED reference, not deep-copied -- see + * "Ownership notes" below). On load, assigned directly to + * core->config (core->configver is incremented, mirroring + * xce_core_push_events()'s CONFIG_PUSH case) WITHOUT going + * through xce_core_push_events()'s CONFIG_PUSH event path. + * This is deliberate: CONFIG_PUSH also unconditionally + * decomposes config's OWN embedded queues[]/resources[]/ + * resvals[] arrays (core_apply_config_queues/resources/ + * resvals, core_events.c) -- which may be stale relative to + * the LIVE queue/resource/resval state by the time a + * snapshot is taken (queues mutate in place via + * QUEUE_UPDATE events after the config that first created + * them was pushed; core->config->queues[] is never + * re-synced). Restoring via the dedicated top-level + * queues/resources/resvals fields below (built from LIVE + * core state, not from whatever config last carried) is + * therefore the authoritative path; going through + * CONFIG_PUSH as well would either double-apply (harmless + * but wasteful and fires on_queue_update twice per queue) + * or, if config's arrays are actually stale, incorrectly + * resurrect old values before the authoritative pass + * overwrites them. `config` in the snapshot is therefore + * purely the opaque "whatever xce_conf4sched_t is/was + * currently assigned" (matching how CONFIG_PUSH already + * treats it -- ownership transfer by pointer, never deep- + * copied anywhere in this codebase today). + * jobs -> core->jobs (kvmgr, iterated), each xce_sched_job_t + * converted to a freshly allocated wire xce_job_t via + * build_wire_job() (the inverse of core_events.c's + * core_build_job(), reimplemented here rather than reused + * -- see "Why not reuse core_events.c" below). + * queues -> core->queues (kvmgr, iterated), each xce_sched_queue_t + * converted via build_wire_queue(). This is the CURRENT, + * live queue state (post any QUEUE_UPDATE deltas) -- + * authoritative, see the `config` note above. + * workers -> core->workers (kvmgr, iterated), each xce_sched_worker_t + * converted via build_wire_worker(), which also flattens + * worker->resvals (a zarray_t indexed by resource idx) back + * into the wire xce_worker_t's own resvals_to_sched[]/ + * nresvals_to_sched fields -- the exact inverse of + * core_build_worker()'s own resvals_to_sched -> resvals + * zarray construction (core_events.c). worker->resInUse is + * NOT captured -- see "Scope decisions" below. + * resources -> core->resmgr (kvmgr, iterated). Resource objects ARE + * already the wire type (xce_resource_t*) natively -- + * makeResource() (resobj.c) stores the wire struct directly + * in resmgr, so no conversion step exists; entries are + * BORROWED references into out->resources (safe: resmgr's + * own xce_resource_t objects are stable, definition-only + * data never mutated or reset by a scheduling session, and + * makeResource() itself always deep-copies on the + * CONSUMING side via makeInstance(), so load_snapshot() + * handing these same pointers to makeResource() on a + * second core produces genuinely independent objects + * there). + * resvals -> core->sharedInst (zarray_t, indexed by resource idx) -- + * the SHARED/global resource-value pool, i.e. exactly what + * RESVALS_FULL/CONFIG_PUSH already populate this exact wire + * type (ResValList/xce_resval_list_t) from in the existing, + * live code path (core_apply_config_resvals, + * core_events.c). Each entry's `host` field is left NULL, + * matching that existing convention (host is only ever + * non-NULL for RESVALS_DELTA's per-worker routing). Every + * xce_val_t value is DEEP-COPIED (dup_val()) -- see + * "Ownership notes" below for why aliasing here would be + * unsafe. Per-worker resource values are NOT part of this + * field; they travel inside each worker's own + * resvals_to_sched (see `workers` above) -- this mirrors + * the existing codebase's own two distinct restoration + * paths (RESVALS_FULL -> sharedInst vs. WORKERS_PUSH's + * embedded resvals_to_sched -> per-worker resvals) rather + * than inventing a third. + * users -> core->users (kvmgr, iterated) -> sched_snapshot_user_t + * (name only, matching xce_sched_user_t's current shape). + * projects -> core->projects (kvmgr, iterated) -> + * sched_snapshot_project_t (name + priority, matching + * xce_sched_project_t's current shape). + * prngSeed -> core->prng_seed (NEW field, added to xce_sched_core_t by + * this task -- see "PRNG seed resolution" below). + * internalState / internalVersion -> see "internal_state contents" below. + * + * ========================================================================== + * PRNG seed resolution (Req 5 AC 7 / AC 21) + * ========================================================================== + * requirements.md Req 5 AC 7 explicitly names `core->prng_seed` ("Any + * required randomness ... shall use a PRNG seeded by the driver at init + * time") and design.md's eventual `xce_sched_core_init(core, decision_cb, + * driver_ctx, prng_seed)` signature both presuppose this field, but neither + * it nor any PRNG consumer existed anywhere in the tree before this task + * (confirmed: `grep -rn "prng\|seed"` across src/scheduler turned up only + * the two requirements.md/design.md doc-comment mentions and one unrelated + * djb2-hash-seed comment in core.c). This is a genuine, real gap -- resolved + * here by adding `uint64_t prng_seed;` to `xce_sched_core_t` (core.h), + * seeded to a deterministic constant (0) by xce_sched_core_init(). Scope is + * deliberately narrow: the field now exists and round-trips through + * save/load (this task's actual job per AC 21's "PRNG state" checklist + * row); xce_sched_core_init()'s signature is deliberately NOT changed to + * take a seed parameter (core.h's own doc comment already documents that + * signature as staying narrower until a later task), and no PRNG consumer + * is wired anywhere -- builtin_fcfs's ordering is fully deterministic via + * ordq comparators today, with no tie-breaking randomness call site to + * seed. A future task adding real PRNG-based tie-breaking should add its + * own setter, the same pattern xce_core_set_decision_cb() already + * establishes for driver_ctx/on_decision. + * + * ========================================================================== + * Unsatisfiable-resreq cache resolution (Req 5 AC 21) + * ========================================================================== + * AC 21's checklist names an "Unsatisfiable cache | Resreqs that could not + * be satisfied in previous rounds (for early skip)". Grepping the ENTIRE + * tree (src/scheduler, src/lib, src/include) for "unsatisfiable"/ + * "resreq_cache"/"cache.*resreq" found ZERO matches -- this cache does not + * exist anywhere in the codebase today; it is aspirational (an optimization + * hinted at in the requirements, never built). There is therefore nothing + * for save_snapshot() to capture. This is documented here rather than + * silently ignored: `internal_state` (below) does NOT include any + * unsatisfiable-cache bytes because there is no such cache to include. + * SHOULD a future task add one to xce_sched_core_t, it can be folded into + * internal_state behind a bumped internal_version (the whole reason that + * field exists). + * + * ========================================================================== + * internal_state contents + * ========================================================================== + * design.md's own capnp-schema comment (ReplayFormat section) describes + * internal_state as covering "shared-resource tracking + unsatisfiable- + * resreq cache + session/iterator state, core-opaque". Given the resolution + * above (unsatisfiable cache does not exist) and that "shared-resource + * tracking" (the DEFINING pool, core->sharedInst) is already faithfully + * captured by the top-level `resvals` field (see above), the only genuinely + * internal-only state left is: + * - core->sharedInUse / each worker's resInUse (in-use, as opposed to + * available, resource tracking): deliberately NOT included. Both are + * fully session-transient -- xce_core_begin_session()'s own doc comment + * (core.h) documents that it resets "shared resval" at the start of + * EVERY session, and the driver's sched_account_jobs() unconditionally + * re-sums them from currently RUNNING jobs' accounts immediately after. + * Since this task also does not restore job->allocs/accounts (see + * "Scope decisions" below), restoring stale in-use numbers with no + * accounts to justify them would be actively inconsistent, and they + * would be wiped by the very next session anyway -- the Observable this + * task proves is specifically about the NEXT session's decisions, which + * never observes pre-session in-use state. + * - core->session_start / core->session_duration_ns: pure diagnostics + * (core.h's own field comment: "nothing in the scheduling/decision path + * reads them"), but cheap and harmless to round-trip, so they ARE + * included, as a fixed 16-byte {session_start, session_duration_ns} + * blob, versioned via internal_version (SCHED_SNAPSHOT_INTERNAL_VERSION, + * core.h) so an incompatible future layout is rejected (WARN + skipped) + * rather than misinterpreted, matching AC 21's own stated purpose for + * this field. + * + * ========================================================================== + * Scope decisions (documented, not silent) + * ========================================================================== + * - job->priority is NOT captured. xce_job_t (the wire type) has no + * priority field at all -- the SAME pre-existing gap core_build_job() + * already documents with its own "FIXME: support job level priority" + * comment (core_events.c). This task does not introduce the gap, and + * does not attempt to work around it via a side channel (e.g. stuffing + * it into internal_state) because job-level priority is not reliably + * distinguishable from "never set" using only wire-representable job + * state today (see git history / design discussion in this file's + * accompanying task report for the full reasoning). Restored jobs get + * priority 0, matching what a fresh JOBS_PUSH already produces. + * - job->allocs / job->accounts (in-flight dispatched-allocation results + * for RUNNING jobs) are NOT captured. AC 21's "Jobs" checklist row + * ("current state, priority, queue assignment, queue_time") does not + * list allocations; xce_job_alloc_t is a nontrivial nested structure + * (per-host, per-resource) with no existing deep-copy helper anywhere + * in this codebase; and aliasing it across two independently-alive + * cores (the Observable's own test scenario keeps both the pre-save and + * post-load core alive simultaneously) would create a genuine double- + * free hazard the instant either core's normal JOB_STATE_CHANGE path + * frees it. A RUNNING job therefore survives the round trip with + * correct state/queue/user/project/resreq, but "forgets" what it is + * currently dispatched on. Flagged as a concern in this task's status + * report, not silently worked around. + * - xce_sched_queue_t's project_name/job_pick_plugin have no field on the + * wire xce_queue_t type at all (confirmed: xce/types.h's xce_queue_t + * has no such fields) -- the SAME pre-existing gap core_build_queue() + * already documents ("no wire-config source yet"). Nothing in the + * current codebase ever sets either field on a live queue (scaffolding + * only, per task 1.3's own design.md note), so this has zero observed + * impact today. + * - core->ugmgr/qgmgr/pgmgr/hgmgr (group managers) are NOT captured -- + * they have no field anywhere in sched_snapshot_data_t, and nothing in + * core.c/core_events.c populates them from any event today either (no + * regression, no existing behavior to preserve). + * + * ========================================================================== + * Why not reuse core_events.c's core_build_job/core_build_queue/ + * core_build_worker + * ========================================================================== + * Those three helpers are `static` to core_events.c -- invisible outside + * that translation unit, and this file is compiled as its own separate + * object in the real build (scheduler_srcs, meson.build), so they cannot be + * called from here directly. Reusing them would mean exporting them (a + * larger, unrequested change to core_events.c's surface) or funnelling + * every restored object through xce_core_push_events() itself -- which + * reintroduces the config/queue double-application problem discussed above + * for queues, and offers no benefit for jobs/workers either (this file + * needs its own construction step regardless, to build the wire objects in + * the SAVE direction, for which no exported helper exists anywhere). The + * field-copy algorithm used by snapshot_load.c's xce_sched_core_load_snapshot() + * (build a fresh xce_sched_job_t/queue_t/worker_t, call the exported + * sched_add_*()) is therefore a second, small, independent implementation + * of the same algorithm core_build_job/core_build_queue/core_build_worker + * already establish -- deliberately kept field-for-field identical to them + * so the two never silently drift. + * + * ========================================================================== + * Picker rebuild on load + * ========================================================================== + * See snapshot_load.c's own file header comment (this file's sibling, + * housing xce_sched_core_load_snapshot()) for the full reasoning -- purely + * load-direction behavior, moved there in the three-file split above. + * + * ========================================================================== + * Ownership notes + * ========================================================================== + * - `config` and each `resources[]` entry are BORROWED references (not + * deep-copied) from the source core into the snapshot -- safe because + * neither is ever mutated or freed by anything in this codebase today + * (config is read-only after being pushed; resmgr's resource objects + * are a stable, session-independent registry). A caller that frees a + * snapshot's contents independently of the core it was taken from would + * need to account for this -- out of this task's scope (no such + * free-snapshot function exists yet; nothing in this spec's committed + * tasks calls for one). + * - Every xce_val_t (resource VALUE, as opposed to resource DEFINITION) + * that crosses the save/load boundary is DEEP-COPIED via dup_val() + * (snapshot_wire.c), because the zarrays holding them (core->sharedInst, + * each worker's resvals) ARE reset/cleared by begin-session machinery + * (xce_core_begin_session()'s "shared resval" reset) -- aliasing here + * would let one core's session-reset free a value the other core still + * references, a real double-free/use-after-free hazard triggered by + * exactly the "drive both cores through an identical session" scenario + * this task's own Observable test performs. + * - jobs/queues/workers/users/projects wire objects (built fresh by + * build_wire_job()/build_wire_queue()/build_wire_worker(), snapshot_wire.c, + * and the inline user/project loops below) are always independently + * heap-allocated, never aliased with the source core's own + * xce_sched_*_t objects. + */ + +/* dup_val/arr_append/find_resource_name_by_idx/build_wire_job/ + * build_wire_queue/build_wire_worker -- implemented in snapshot_wire.c (see + * this file's own header comment, "Three-file split"), redeclared here + * rather than pulling in a private header for six symbols this file alone + * consumes -- same idiom as core_session.c's own + * `extern void sched_worker_release_host_ctx(...)`. */ +extern xce_val_t* dup_val(const xce_val_t*); +extern int arr_append(void***, int*, int*, void*); +extern const char* find_resource_name_by_idx(xce_resource_t**, int, int); +extern xce_job_t* build_wire_job(xce_sched_job_t*); +extern xce_queue_t* build_wire_queue(xce_sched_queue_t*); +extern xce_worker_t* build_wire_worker(xce_sched_worker_t*, xce_resource_t**, + int); + +/* ------------------------------------------------------------------------ + * xce_sched_core_save_snapshot + * ---------------------------------------------------------------------- */ + +int xce_sched_core_save_snapshot(xce_sched_core_t* core, + sched_snapshot_data_t* out) +{ + iterator_t it; + int rc; + int i; + + xce_job_t** jobs_arr = NULL; + int jobs_cap = 0; + int jobs_n = 0; + xce_queue_t** queues_arr = NULL; + int queues_cap = 0; + int queues_n = 0; + xce_worker_t** workers_arr = NULL; + int workers_cap = 0; + int workers_n = 0; + xce_resource_t** res_arr = NULL; + int res_cap = 0; + int res_n = 0; + xce_resval_t** resval_arr = NULL; + int resval_cap = 0; + int resval_n = 0; + sched_snapshot_user_t** users_arr = NULL; + int users_cap = 0; + int users_n = 0; + sched_snapshot_project_t** proj_arr = NULL; + int proj_cap = 0; + int proj_n = 0; + + if ((core == NULL) || (out == NULL)) { + return -EINVAL; + } + + if (atomic_load(&(core->session_active)) != 0) { + return -EBUSY; + } + + memset(out, 0, sizeof(*out)); + + /* config: borrowed reference -- see file header "Ownership notes". */ + out->config = core->config; + + /* resources -- already the wire type natively, borrowed references. */ + rc = iter_make(&it, core->resmgr); + if (rc >= 0) { + for (;;) { + xce_resource_t* r = NULL; + + rc = iter_next(&it, (void**)&r); + if ((rc < 0) || (rc == RTTI_ITER_NO_DATA)) { + break; + } + + if (arr_append((void***)&res_arr, &res_cap, &res_n, r) < 0) { + break; + } + } + iter_destroy(&it); + } + + /* resvals (shared pool only -- see file header "Field-by-field + * mapping", resvals row). */ + for (i = 0; i < core->sharedInst.count; i++) { + xce_val_t* val = (xce_val_t*)zarray_get(&(core->sharedInst), i); + const char* rname; + xce_resval_t* rv; + + if (val == NULL) { + continue; + } + + rname = find_resource_name_by_idx(res_arr, res_n, i); + if (rname == NULL) { + continue; + } + + rv = (xce_resval_t*)calloc(1, sizeof(*rv)); + if (rv == NULL) { + continue; + } + rv->name = makeString((char*)rname); + rv->host = NULL; + rv->val = dup_val(val); + + if (arr_append((void***)&resval_arr, &resval_cap, &resval_n, rv) < 0) { + FREEUP(rv->name); + FREEUP(rv->val); + FREEUP(rv); + break; + } + } + + /* queues */ + rc = iter_make(&it, core->queues); + if (rc >= 0) { + for (;;) { + xce_sched_queue_t* q = NULL; + xce_queue_t* wq; + + rc = iter_next(&it, (void**)&q); + if ((rc < 0) || (rc == RTTI_ITER_NO_DATA)) { + break; + } + + wq = build_wire_queue(q); + if (wq == NULL) { + continue; + } + + if (arr_append((void***)&queues_arr, &queues_cap, &queues_n, wq) + < 0) { + break; + } + } + iter_destroy(&it); + } + + /* workers */ + rc = iter_make(&it, core->workers); + if (rc >= 0) { + for (;;) { + xce_sched_worker_t* w = NULL; + xce_worker_t* ww; + + rc = iter_next(&it, (void**)&w); + if ((rc < 0) || (rc == RTTI_ITER_NO_DATA)) { + break; + } + + ww = build_wire_worker(w, res_arr, res_n); + if (ww == NULL) { + continue; + } + + if (arr_append((void***)&workers_arr, &workers_cap, &workers_n, ww) + < 0) { + break; + } + } + iter_destroy(&it); + } + + /* jobs */ + rc = iter_make(&it, core->jobs); + if (rc >= 0) { + for (;;) { + xce_sched_job_t* job = NULL; + xce_job_t* wj; + + rc = iter_next(&it, (void**)&job); + if ((rc < 0) || (rc == RTTI_ITER_NO_DATA)) { + break; + } + + wj = build_wire_job(job); + if (wj == NULL) { + continue; + } + + if (arr_append((void***)&jobs_arr, &jobs_cap, &jobs_n, wj) < 0) { + break; + } + } + iter_destroy(&it); + } + + /* users */ + rc = iter_make(&it, core->users); + if (rc >= 0) { + for (;;) { + xce_sched_user_t* u = NULL; + sched_snapshot_user_t* su; + + rc = iter_next(&it, (void**)&u); + if ((rc < 0) || (rc == RTTI_ITER_NO_DATA)) { + break; + } + + su = (sched_snapshot_user_t*)calloc(1, sizeof(*su)); + if (su == NULL) { + continue; + } + su->name = (u->name != NULL) ? makeString(u->name) : NULL; + + if (arr_append((void***)&users_arr, &users_cap, &users_n, su) < 0) { + FREEUP(su->name); + FREEUP(su); + break; + } + } + iter_destroy(&it); + } + + /* projects */ + rc = iter_make(&it, core->projects); + if (rc >= 0) { + for (;;) { + xce_sched_project_t* p = NULL; + sched_snapshot_project_t* sp; + + rc = iter_next(&it, (void**)&p); + if ((rc < 0) || (rc == RTTI_ITER_NO_DATA)) { + break; + } + + sp = (sched_snapshot_project_t*)calloc(1, sizeof(*sp)); + if (sp == NULL) { + continue; + } + sp->name = (p->name != NULL) ? makeString(p->name) : NULL; + sp->priority = p->priority; + + if (arr_append((void***)&proj_arr, &proj_cap, &proj_n, sp) < 0) { + FREEUP(sp->name); + FREEUP(sp); + break; + } + } + iter_destroy(&it); + } + + out->jobs = (xce_jobs_data_t*)calloc(1, sizeof(*(out->jobs))); + if (out->jobs != NULL) { + out->jobs->n_total = jobs_n; + out->jobs->pos = 0; + out->jobs->count = jobs_n; + out->jobs->jobs = jobs_arr; + } + + out->queues = queues_arr; + out->n_queues = queues_n; + + out->workers = (xce_workers_data_t*)calloc(1, sizeof(*(out->workers))); + if (out->workers != NULL) { + out->workers->n_total = workers_n; + out->workers->pos = 0; + out->workers->count = workers_n; + out->workers->workers = workers_arr; + } + + out->resources = (xce_resource_data_t*)calloc(1, sizeof(*(out->resources))); + if (out->resources != NULL) { + out->resources->n_total = res_n; + out->resources->pos = 0; + out->resources->count = res_n; + out->resources->resources = res_arr; + } + + out->resvals = (xce_resval_list_t*)calloc(1, sizeof(*(out->resvals))); + if (out->resvals != NULL) { + out->resvals->count = resval_n; + out->resvals->vals = resval_arr; + } + + out->users = users_arr; + out->n_users = users_n; + + out->projects = proj_arr; + out->n_projects = proj_n; + + out->prng_seed = core->prng_seed; + + /* internal_state -- see file header "internal_state contents". */ + { + uint64_t* blob = (uint64_t*)calloc(2, sizeof(uint64_t)); + + if (blob != NULL) { + blob[0] = core->session_start; + blob[1] = core->session_duration_ns; + out->internal_state = (uint8_t*)blob; + out->internal_state_len = 2 * sizeof(uint64_t); + } + } + out->internal_version = SCHED_SNAPSHOT_INTERNAL_VERSION; + + return 0; +} diff --git a/src/scheduler/src/snapshot_load.c b/src/scheduler/src/snapshot_load.c new file mode 100644 index 00000000..28dd3c49 --- /dev/null +++ b/src/scheduler/src/snapshot_load.c @@ -0,0 +1,335 @@ +#include +#include +#include +#include + +#include "core.h" +#include "plugin.h" +#include "resobj.h" +#include "sched/types.h" +#include "utils.h" +#include "xstring.h" + +/* + * snapshot_load.c - xce_sched_core_load_snapshot() (scheduler-core spec, + * task 8.1, Req 5 AC 21 / Req 6 AC 5), factored out of snapshot.c's + * original single-file implementation -- see snapshot.c's own file header + * comment, "Three-file split", for the full rationale (this project's + * 800-line-per-file convention, CLAUDE.md/coding-style.md) and for EVERY + * section of design documentation shared by both the save and load + * directions (field-by-field mapping, PRNG seed resolution, + * unsatisfiable-cache resolution, internal_state contents, scope + * decisions, why-not-reuse-core_events.c, ownership notes). Read that + * comment in full before touching this file. + * + * ========================================================================== + * Picker rebuild on load (Req 5 AC 21's own "Picker state is NOT stored in + * snapshot" rule; this task's own text) + * ========================================================================== + * After queues and jobs are restored (raw state only, via sched_add_queue()/ + * sched_add_job() -- no hook firing yet), xce_sched_core_load_snapshot() + * walks them AGAIN, in the exact same order xce_core_push_events() + * (core_events.c) already establishes for state-change hooks + * (core_events.c's own doc comment: QUEUE_UPDATE fires on_queue_update once + * per queue; JOBS_PUSH fires on_job_push once per job, unconditionally -- + * the hook itself, not the call site, decides whether a non-pending job + * becomes a candidate): on_queue_update for every restored queue, THEN + * on_job_push for every restored job. This is not a new, ad-hoc mechanism + * -- it is literally the same xce_job_picker_iface_t hook calls + * core_events.c already makes, called directly here instead of indirectly + * through xce_core_push_events() (which this file cannot use for + * queues/jobs without the config/queue double-application problem + * snapshot.c's header documents for the `config` field). Picker resolution + * (sched_resolve_job_picker()) is attempted once; on failure, hooks are + * skipped for the whole load with a WARN, mirroring xce_core_push_events()'s + * own defensive posture for the same failure mode. + */ + +/* dup_val - implemented in snapshot_wire.c (see snapshot.c's header + * comment, "Three-file split"), redeclared here rather than pulling in a + * private header for the one symbol this file consumes -- same idiom as + * core_session.c's own `extern void sched_worker_release_host_ctx(...)`. */ +extern xce_val_t* dup_val(const xce_val_t*); + +/* ------------------------------------------------------------------------ + * xce_sched_core_load_snapshot + * ---------------------------------------------------------------------- */ + +int xce_sched_core_load_snapshot(xce_sched_core_t* core, + const sched_snapshot_data_t* in) +{ + xce_sched_plugin_t* picker = NULL; + int rc; + int i; + + if ((core == NULL) || (in == NULL)) { + return -EINVAL; + } + + if (atomic_load(&(core->session_active)) != 0) { + return -EBUSY; + } + + rc = sched_resolve_job_picker(core, &picker); + if (rc < 0) { + WARN("%s failed to resolve job picker plugin (rc=%d); picker " + "rebuild will be skipped for this load", + __func__, rc); + picker = NULL; + } + + /* config -- borrowed reference, matches save_snapshot()'s own choice; + * see snapshot.c header "Field-by-field mapping", config row, for why + * this is a direct assignment rather than a CONFIG_PUSH event. */ + core->config = in->config; + core->configver++; + + if ((picker != NULL) && (picker->picker_iface->on_config != NULL)) { + picker->picker_iface->on_config(picker, core->config); + } + + /* resources */ + if (in->resources != NULL) { + for (i = 0; i < in->resources->count; i++) { + xce_resource_t* r = in->resources->resources[i]; + + if (r == NULL) { + continue; + } + + if (makeResource(core->resmgr, r) == NULL) { + WARN("%s failed to restore resource '%s' (errno=%d)", __func__, + r->name ? r->name : "(null)", errno); + } + } + } + + /* resvals (shared pool) -- host==NULL entries only, see snapshot.c + * header "Field-by-field mapping", resvals row. */ + if (in->resvals != NULL) { + for (i = 0; i < in->resvals->count; i++) { + xce_resval_t* rv = in->resvals->vals[i]; + xce_resource_t* r = NULL; + + if ((rv == NULL) || (rv->name == NULL) || (rv->host != NULL)) { + continue; + } + + if ((key_find(core->resmgr, rv->name, (void**)&r) < 0) + || (r == NULL)) { + WARN("%s resval for unknown resource '%s', skipping", __func__, + rv->name); + continue; + } + + zarray_put(&(core->sharedInst), r->idx, dup_val(rv->val)); + } + } + + /* queues -- authoritative live queue state, restored directly (not via + * CONFIG_PUSH) -- see snapshot.c header "Field-by-field mapping", + * config row. Picker rebuild (on_queue_update) fires here, per-queue, + * matching "queue updates ... first" ordering. */ + if (in->queues != NULL) { + for (i = 0; i < in->n_queues; i++) { + xce_queue_t* wq = in->queues[i]; + xce_sched_queue_t* q; + + if ((wq == NULL) || (wq->queue == NULL)) { + continue; + } + + q = (xce_sched_queue_t*)calloc(1, sizeof(*q)); + if (q == NULL) { + continue; + } + + q->name = makeString(wq->queue); + q->attrs = wq->attrs; + q->status = wq->status; + q->priority = wq->priority; + q->accept_interval = wq->accept_interval; + q->sched_delay = wq->sched_delay; + q->resreq = (wq->resreq != NULL) ? makeString(wq->resreq) : NULL; + q->project_name = NULL; + q->job_pick_plugin = NULL; + + sched_add_queue(core, q); + + if (picker != NULL) { + picker->picker_iface->on_queue_update(picker, q); + } + } + } + + /* workers -- no picker hook (Req 1 AC 2). */ + if (in->workers != NULL) { + for (i = 0; i < in->workers->count; i++) { + xce_worker_t* ww = in->workers->workers[i]; + xce_sched_worker_t* w; + int k; + + if ((ww == NULL) || (ww->name == NULL)) { + continue; + } + + w = (xce_sched_worker_t*)calloc(1, sizeof(*w)); + if (w == NULL) { + continue; + } + + w->name = makeString(ww->name); + w->status = ww->status; + w->idx = -1; + + zarray_init(&(w->resvals), 128, + (zarray_item_killer_t)kill_sched_resval); + zarray_init(&(w->resInUse), 128, + (zarray_item_killer_t)kill_sched_resval); + + for (k = 0; k < ww->nresvals_to_sched; k++) { + xce_name_val_t* nv = ww->resvals_to_sched[k]; + xce_resource_t* r = NULL; + + if ((nv == NULL) || (nv->name == NULL)) { + continue; + } + + if ((key_find(core->resmgr, nv->name, (void**)&r) < 0) + || (r == NULL)) { + WARN("%s worker '%s' resval for unknown resource '%s', " + "skipping", + __func__, w->name, nv->name); + continue; + } + + zarray_put(&(w->resvals), r->idx, dup_val(nv->val)); + } + + sched_add_worker(core, w); + } + } + + /* users -- restored before jobs so sched_get_user() below always finds + * an already-registered user (get-or-create suffices: no other field + * to lose). */ + if (in->users != NULL) { + for (i = 0; i < in->n_users; i++) { + sched_snapshot_user_t* su = in->users[i]; + + if ((su == NULL) || (su->name == NULL)) { + continue; + } + + sched_get_user(core, su->name); + } + } + + /* projects -- restored before jobs, same reasoning as users; priority + * is explicitly re-applied since get-or-create always defaults a + * freshly created project's priority to 0. */ + if (in->projects != NULL) { + for (i = 0; i < in->n_projects; i++) { + sched_snapshot_project_t* sp = in->projects[i]; + xce_sched_project_t* p; + + if ((sp == NULL) || (sp->name == NULL)) { + continue; + } + + p = sched_get_project(core, sp->name); + if (p != NULL) { + p->priority = sp->priority; + } + } + } + + /* jobs -- restored last (queues/users/projects it references must + * already exist). Picker rebuild (on_job_push) fires here, per job, + * matching "... then pending-job pushes" ordering -- unconditionally, + * same as JOBS_PUSH (core_events.c): the hook itself decides whether a + * non-pending job becomes a candidate. */ + if (in->jobs != NULL) { + for (i = 0; i < in->jobs->count; i++) { + xce_job_t* wj = in->jobs->jobs[i]; + xce_sched_job_t* job; + + if ((wj == NULL) || (wj->submit == NULL)) { + continue; + } + + job = (xce_sched_job_t*)calloc(1, sizeof(*job)); + if (job == NULL) { + continue; + } + + job->jobid = wj->jobid; + job->sub_time = wj->sub_time; + job->queue_time = wj->queue_time; + job->start_time = wj->start_time; + job->state = wj->state; + job->user = sched_get_user(core, wj->submit->user); + job->queue = sched_get_queue(core, wj->submit->queue); + job->project = sched_get_project(core, wj->submit->project_name); + job->mintasks = wj->submit->mintasks; + job->maxtasks = wj->submit->maxtasks; + /* priority: no wire source -- see snapshot.c header "Scope + * decisions"; matches core_build_job()'s own default. */ + job->priority = 0; + + job->resreq = (xce_resreq_t*)calloc(1, sizeof(*(job->resreq))); + if (job->resreq != NULL) { + char* errmsg = NULL; + char* resreq_str = NULLSTR(wj->submit->resreq) + ? "type == any" + : wj->submit->resreq; + int prc = parse_resreq(job->resreq, resreq_str, core->resmgr, + &errmsg); + + if (prc < 0) { + ERROR("%s parse_resreq failed for jobid %llu: %s", __func__, + (unsigned long long)wj->jobid, + errmsg ? errmsg : "(null)"); + FREEUP(errmsg); + FREEUP(job->resreq); + } + } + + sched_add_job(core, job); + + if (picker != NULL) { + picker->picker_iface->on_job_push(picker, job); + } + } + } + + /* prng_seed -- see snapshot.c header "PRNG seed resolution". Genuine + * gap found while writing this task's own cross-core equivalence test + * (test_snapshot_save_load.c): save_snapshot() already writes + * out->prng_seed (snapshot.c), but this restore was missing entirely + * -- a freshly loaded core silently kept its xce_sched_core_init() + * default (0) instead of the saved value. */ + core->prng_seed = in->prng_seed; + + /* internal_state -- see snapshot.c header "internal_state contents". + * Version-gated: an unrecognized version or undersized blob is WARNed + * and skipped (session diagnostics only -- non-fatal to the load, but + * never misinterpreted as if it were the current layout). */ + if ((in->internal_state != NULL) && (in->internal_state_len > 0)) { + if ((in->internal_version == SCHED_SNAPSHOT_INTERNAL_VERSION) + && (in->internal_state_len >= 2 * sizeof(uint64_t))) { + const uint64_t* blob = (const uint64_t*)in->internal_state; + + core->session_start = blob[0]; + core->session_duration_ns = blob[1]; + } else { + WARN("%s internal_state version mismatch (got %u, expected %u) " + "or undersized (%zu bytes) -- skipping, session " + "diagnostics left at defaults", + __func__, in->internal_version, + SCHED_SNAPSHOT_INTERNAL_VERSION, in->internal_state_len); + } + } + + return 0; +} diff --git a/src/scheduler/src/snapshot_wire.c b/src/scheduler/src/snapshot_wire.c new file mode 100644 index 00000000..e89408b5 --- /dev/null +++ b/src/scheduler/src/snapshot_wire.c @@ -0,0 +1,258 @@ +#include +#include +#include + +#include "core.h" +#include "sched/types.h" +#include "utils.h" +#include "xstring.h" + +/* + * snapshot_wire.c - the small generic helpers and SAVE-direction + * wire-object builders factored out of snapshot.c's original single-file + * implementation (scheduler-core spec, task 8.1, Req 5 AC 21 / Req 6 AC 5) + * -- see snapshot.c's own file header comment, "Three-file split", for the + * full rationale (this project's 800-line-per-file convention, + * CLAUDE.md/coding-style.md). Read snapshot.c's header comment in full + * before touching this file -- in particular "Field-by-field mapping", + * "Scope decisions", and "Why not reuse core_events.c's core_build_job/ + * core_build_queue/core_build_worker", all of which describe the design + * decisions behind the builders below. + * + * Every symbol in this file is consumed from snapshot.c (all six) and, for + * dup_val() alone, also from snapshot_load.c -- each consumer redeclares + * the symbols it needs via a plain `extern` rather than a shared private + * header, matching this codebase's existing idiom for a narrow cross-.c- + * file dependency (core_session.c's own `extern void + * sched_worker_release_host_ctx(...)`). + */ + +/* ------------------------------------------------------------------------ + * Small shared helpers + * ---------------------------------------------------------------------- */ + +/* dup_val - deep-copies an xce_val_t (see snapshot.c's "Ownership notes" + * for why this must never be a plain pointer alias). NULL-safe (returns + * NULL for a NULL src). Non-static: also consumed by snapshot_load.c. */ +xce_val_t* dup_val(const xce_val_t* src) +{ + xce_val_t* v; + + if (src == NULL) { + return NULL; + } + + v = (xce_val_t*)calloc(1, sizeof(*v)); + if (v == NULL) { + return NULL; + } + + v->which = src->which; + if (src->which == XCE_VAL_STRING) { + v->u.svalue + = (src->u.svalue != NULL) ? makeString(src->u.svalue) : NULL; + } else { + v->u = src->u; + } + + return v; +} + +/* arr_append - generic growing pointer-array append (doubling capacity), + * same growth discipline driver.c's translate_pending_events() already + * establishes for its own events[]/reqs[] arrays, generalized here since + * this file (and snapshot.c's save_snapshot()) repeats the pattern across + * seven distinct collections (jobs/queues/workers/resources/resvals/users/ + * projects). Only ever used to STORE and later READ BACK opaque pointer + * values through the caller's own correctly-typed variable -- never + * dereferenced as void** anywhere else -- so the void*** aliasing is safe + * in practice (well-established C idiom). Returns 0 on success, -ENOMEM on + * allocation failure (the array/cap/n triple is left at its last-good + * state, so already-appended entries are not lost). Non-static: also + * consumed directly by snapshot.c's save_snapshot(). */ +int arr_append(void*** arr, int* cap, int* n, void* item) +{ + if (*n >= *cap) { + int newcap = (*cap != 0) ? (*cap * 2) : 8; + void** narr = (void**)realloc(*arr, (size_t)newcap * sizeof(void*)); + + if (narr == NULL) { + return -ENOMEM; + } + + *arr = narr; + *cap = newcap; + } + + (*arr)[(*n)++] = item; + + return 0; +} + +/* find_resource_name_by_idx - reverse-lookup a resource's name given its + * idx, by linear scan of an already-built resources[] array. Used to label + * sharedInst/worker->resvals zarray entries (indexed by idx) with the + * resource NAME the wire xce_resval_t/xce_name_val_t types require. + * Snapshot save is not a hot path; resource counts are small. Non-static: + * also consumed directly by snapshot.c's save_snapshot(). */ +const char* find_resource_name_by_idx(xce_resource_t** resources, + int nresources, int idx) +{ + int i; + + for (i = 0; i < nresources; i++) { + if ((resources[i] != NULL) && (resources[i]->idx == idx)) { + return resources[i]->name; + } + } + + return NULL; +} + +/* ------------------------------------------------------------------------ + * SAVE-direction wire-object builders (inverse of core_events.c's + * core_build_job/core_build_queue/core_build_worker -- see snapshot.c's + * header comment, "Why not reuse core_events.c") + * ---------------------------------------------------------------------- */ + +xce_job_t* build_wire_job(xce_sched_job_t* job) +{ + xce_job_t* j; + char* resreq_str = NULL; + + if (job == NULL) { + return NULL; + } + + j = (xce_job_t*)calloc(1, sizeof(*j)); + if (j == NULL) { + return NULL; + } + + j->jobid = job->jobid; + j->sub_time = job->sub_time; + j->queue_time = job->queue_time; + j->start_time = job->start_time; + j->state = job->state; + + j->submit = (job_submit_data_t*)calloc(1, sizeof(*(j->submit))); + if (j->submit == NULL) { + FREEUP(j); + return NULL; + } + + j->submit->queue + = (job->queue != NULL) ? makeString(job->queue->name) : NULL; + j->submit->user = (job->user != NULL) ? makeString(job->user->name) : NULL; + j->submit->project_name + = (job->project != NULL) ? makeString(job->project->name) : NULL; + j->submit->mintasks = job->mintasks; + j->submit->maxtasks = job->maxtasks; + + /* job->priority: no wire slot -- see snapshot.c header "Scope + * decisions". */ + + if (job->resreq != NULL) { + int rc = format_resreq(&resreq_str, job->resreq); + + if (rc < 0) { + WARN("%s format_resreq failed for jobid %llu (rc=%d); resreq " + "will be empty for this restored job", + __func__, (unsigned long long)job->jobid, rc); + resreq_str = NULL; + } + } + j->submit->resreq = resreq_str; + + /* job->allocs/accounts: not captured -- see snapshot.c header "Scope + * decisions". */ + + return j; +} + +xce_queue_t* build_wire_queue(xce_sched_queue_t* q) +{ + xce_queue_t* wq; + + if (q == NULL) { + return NULL; + } + + wq = (xce_queue_t*)calloc(1, sizeof(*wq)); + if (wq == NULL) { + return NULL; + } + + wq->queue = (q->name != NULL) ? makeString(q->name) : NULL; + wq->attrs = q->attrs; + wq->status = q->status; + wq->priority = q->priority; + wq->accept_interval = q->accept_interval; + wq->sched_delay = q->sched_delay; + wq->resreq = (q->resreq != NULL) ? makeString(q->resreq) : NULL; + + /* project_name/job_pick_plugin: no wire slot -- see snapshot.c header + * "Scope decisions". */ + + return wq; +} + +xce_worker_t* build_wire_worker(xce_sched_worker_t* w, + xce_resource_t** resources, int nresources) +{ + xce_worker_t* ww; + xce_name_val_t** nvs = NULL; + int cap = 0; + int n = 0; + int i; + + if (w == NULL) { + return NULL; + } + + ww = (xce_worker_t*)calloc(1, sizeof(*ww)); + if (ww == NULL) { + return NULL; + } + + ww->name = (w->name != NULL) ? makeString(w->name) : NULL; + ww->status = w->status; + + for (i = 0; i < w->resvals.count; i++) { + xce_val_t* val = (xce_val_t*)zarray_get(&(w->resvals), i); + const char* rname; + xce_name_val_t* nv; + + if (val == NULL) { + continue; + } + + rname = find_resource_name_by_idx(resources, nresources, i); + if (rname == NULL) { + continue; + } + + nv = (xce_name_val_t*)calloc(1, sizeof(*nv)); + if (nv == NULL) { + continue; + } + + nv->name = makeString((char*)rname); + nv->val = dup_val(val); + + if (arr_append((void***)&nvs, &cap, &n, nv) < 0) { + FREEUP(nv->name); + FREEUP(nv->val); + FREEUP(nv); + break; + } + } + + ww->resvals_to_sched = nvs; + ww->nresvals_to_sched = n; + + /* worker->resInUse: not captured -- see snapshot.c header + * "internal_state contents" (session-transient, always recomputed). */ + + return ww; +} diff --git a/src/scheduler/testsuite/meson.build b/src/scheduler/testsuite/meson.build index 30882ffa..669d741c 100644 --- a/src/scheduler/testsuite/meson.build +++ b/src/scheduler/testsuite/meson.build @@ -370,6 +370,29 @@ sched_driver_session_ordering_test = executable( dependencies: [scheduler_test_deps], install: false) +# Cross-core behavioral-equivalence proof for +# xce_sched_core_save_snapshot()/xce_sched_core_load_snapshot() +# (scheduler-core spec, task 8.1, Req 5 AC 21 / Req 6 AC 5): save immediately +# followed by load into a freshly-initialized, completely separate core +# produces a core whose observable scheduling behavior (next session's +# decisions, given the same events) is identical to the pre-save core, plus +# direct unit-level proof of both functions' -EBUSY (mid-session)/-EINVAL +# (NULL args) guards. Same white-box #include pattern as +# test_core_full_cycle.c (../src/core.c, ../src/core_clock.c, +# ../src/core_events.c, ../src/core_session.c, ../src/decision.c) PLUS the +# three files task 8.1's own 800-line-cap split produced: +# ../src/snapshot.c, ../src/snapshot_wire.c, ../src/snapshot_load.c. See +# that test's own header comment for the two-independent-picker-instance +# design (xce_sched_plugin_t->data-scoped state) that lets this test drive +# two fully independent xce_sched_core_t instances through the identical +# session protocol without cross-contaminating a shared global. +sched_snapshot_save_load_test = executable( + 'sched_snapshot_save_load_test', + ['test_snapshot_save_load.c'], + include_directories:[incdir], + dependencies: [scheduler_test_deps], + install: false) + # Register as meson tests so `meson test --suite scheduler` runs them all. # detect_leaks=0 matches the suite convention (fixtures are not torn down). test('scheduler test', scheduler_test, suite: 'scheduler', @@ -427,3 +450,6 @@ test('scheduler driver decision cb test', test('scheduler driver session ordering test', sched_driver_session_ordering_test, suite: 'scheduler', env: ['ASAN_OPTIONS=detect_leaks=0']) +test('scheduler snapshot save load test', + sched_snapshot_save_load_test, suite: 'scheduler', + env: ['ASAN_OPTIONS=detect_leaks=0']) diff --git a/src/scheduler/testsuite/test_snapshot_save_load.c b/src/scheduler/testsuite/test_snapshot_save_load.c new file mode 100644 index 00000000..cdab661b --- /dev/null +++ b/src/scheduler/testsuite/test_snapshot_save_load.c @@ -0,0 +1,846 @@ +/* + * test_snapshot_save_load.c - the cross-core behavioral-equivalence proof + * for xce_sched_core_save_snapshot()/xce_sched_core_load_snapshot() + * (scheduler-core spec, task 8.1, Req 5 AC 21 / Req 6 AC 5). + * + * Observable under test (task 8.1's own text): "save immediately followed + * by load into a freshly-initialized core produces a core whose observable + * scheduling behavior (next session's decisions, given the same events) is + * identical to the pre-save core." This is a genuinely CROSS-CORE proof, + * not a field-level round-trip check alone: two fully independent + * xce_sched_core_t instances (core_a, the original; core_b, freshly + * xce_sched_core_init()'d and populated purely via + * xce_sched_core_load_snapshot()) are each driven through the SAME + * push-events -> go -> drain-decisions -> end-session cycle + * test_core_full_cycle.c (task 4.6) already established as the correct, + * real way to drive a session -- a real picker pthread plus a stand-in + * allocator (this file's own run_identical_session(), the test function's + * own thread) reproducing sched_job_picker()/sched_allocator()'s real + * per-session bodies, NOT a hand-rolled shortcut. The two cores' resulting + * decision sequences (op + jobid, in order) are asserted byte-for-byte + * identical. Field-level assertions (queues/resources/resvals/workers/ + * users/projects/prng_seed/jobs restored onto core_b) are ALSO included -- + * cheaper to debug from, per this task's own text, but not a substitute + * for the cross-core proof above. + * + * White-box #include of core.c, core_clock.c, core_events.c, core_session.c, + * decision.c (same convention/file set as test_core_full_cycle.c) PLUS the + * three files under test: snapshot.c, snapshot_wire.c, snapshot_load.c + * (task 8.1's own three-file split -- see snapshot.c's file header + * comment). Stubs below are the exact same set test_core_full_cycle.c + * already establishes needing for this file combination (kill_sched_* + * family, plugin_unregister, sched_worker_release_host_ctx, + * sched_resolve_job_picker) + * -- UNLIKE that file, sched_worker_release_host_ctx is genuinely invoked + * here (core_reset_workers(), core_session.c) since this test adds real + * workers to both cores; it remains a safe no-op regardless. + * makeResource()/parse_resreq()/format_resreq()/key_find()/zarray_*() are + * all REAL, already-linked libxce/librtti symbols (src/lib/resobj.c, + * src/lib/resreq.c) -- no stub needed for any of them. + * + * Two independent stub picker plugin instances (g_stub_plugin_a/ + * g_stub_plugin_b), each with its OWN private stub_picker_state_t reached + * via xce_sched_plugin_t's existing `data` field (the same mechanism + * builtin_fcfs.c's real fcfs_plugin_data_t uses) -- this is what lets + * core_a and core_b each build and walk their OWN independent candidate + * list without cross-contaminating a shared global, while still reusing + * the exact same core-parameterized picker-thread function + * (test_picker_thread_fn(), copied from test_core_full_cycle.c verbatim) + * for both: sched_resolve_job_picker(core, ...) naturally resolves to + * whichever plugin instance is registered in THAT core's own + * core->plugins kvmgr. + */ +#include +#include +#include +#include +#include +#include + +#include + +#include "kvmgr.h" +#include "preason.h" +#include "scheduler.h" +#include "xstring.h" + +/* Pull in the real xce_sched_core_init()/xce_core_push_events()/ + * xce_core_go()/xce_core_end_session()/xce_core_set_decision_cb()/ + * xce_sched_core_save_snapshot()/xce_sched_core_load_snapshot() under test + * into this TU (white-box #include, same convention as + * test_core_full_cycle.c). */ +#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/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; } +void plugin_unregister(void* arg) { (void)arg; } + +/* core_reset_workers()'s (core_session.c) linker dependency. UNLIKE + * test_core_full_cycle.c's own identical stub, this one IS actually + * invoked here (both core_a and core_b get real workers below) -- still a + * safe no-op, see file header comment. */ +void sched_worker_release_host_ctx(xce_sched_worker_t* w) { (void)w; } + +/* Real reimplementation of schedobj.c's sched_resolve_job_picker -- see + * file header comment (same rationale/algorithm as + * test_core_full_cycle.c's own copy). Parameterized purely on `core`, so it + * naturally resolves core_a's own registered plugin for core_a and + * core_b's own registered plugin for core_b. */ +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; +} + +void setUp(void) { return; } +void tearDown(void) { return; } + +/* ---------------------------------------------------------------------- */ +/* Stub picker: per-instance private state via xce_sched_plugin_t->data */ +/* (see file header comment) -- each of g_stub_state_a/g_stub_state_b */ +/* independently accumulates ITS OWN core's candidate list. */ +/* ---------------------------------------------------------------------- */ + +#define STUB_MAX_CANDIDATES 8 + +typedef struct { + xce_sched_job_t* candidates[STUB_MAX_CANDIDATES]; + int ncandidates; + int cursor; +} stub_picker_state_t; + +static stub_picker_state_t g_stub_state_a; +static stub_picker_state_t g_stub_state_b; + +static int stub_on_job_push(void* picker, xce_sched_job_t* job) +{ + stub_picker_state_t* st + = (stub_picker_state_t*)((xce_sched_plugin_t*)picker)->data; + + if (st->ncandidates < STUB_MAX_CANDIDATES) { + st->candidates[st->ncandidates++] = job; + } + + return 0; +} + +static int stub_on_queue_update(void* picker, xce_sched_queue_t* queue) +{ + (void)picker; + (void)queue; + + return 0; +} + +static int stub_on_job_state_change(void* picker, xce_sched_job_t* job, + uint32_t state) +{ + (void)picker; + (void)job; + (void)state; + + return 0; /* not exercised by this test's event batch */ +} + +static int stub_on_job_update(void* picker, xce_sched_job_t* job) +{ + (void)picker; + (void)job; + + return 0; /* not exercised by this test's event batch */ +} + +/* begin_session()/next_job() run in the real picker thread ONLY -- same + * "no lock needed against on_dispatch()/on_pend()" reasoning + * test_core_full_cycle.c's own header comment gives (this stub's + * on_dispatch/on_pend below are pure no-ops, touching nothing). */ +static int stub_begin_session(void* picker) +{ + stub_picker_state_t* st + = (stub_picker_state_t*)((xce_sched_plugin_t*)picker)->data; + + st->cursor = 0; /* session-aware requirement (plugin.h): never carry the + * iteration cursor over from a previous session. */ + + return 0; +} + +static xce_sched_job_t* stub_next_job(void* picker) +{ + stub_picker_state_t* st + = (stub_picker_state_t*)((xce_sched_plugin_t*)picker)->data; + + if (st->cursor >= st->ncandidates) { + return NULL; + } + + return st->candidates[st->cursor++]; +} + +static int stub_end_session(void* picker) +{ + (void)picker; + + return 0; +} + +/* Post-allocation hooks (plugin.h: allocator thread, in production; the + * main thread's stand-in allocator here, run_identical_session() below) -- + * pure no-ops: this test's decision-sequence comparison is driven entirely + * off driver_ctx->seen[] (the real xce_decision_t stream), not off these + * hooks. */ +static int stub_on_dispatch(void* picker, uint64_t jobid) +{ + (void)picker; + (void)jobid; + + return 0; +} + +static int stub_on_pend(void* picker, uint64_t jobid) +{ + (void)picker; + (void)jobid; + + return 0; +} + +static xce_job_picker_iface_t g_stub_picker_iface = { + .init = NULL, + .fini = NULL, + .begin_session = stub_begin_session, + .next_job = stub_next_job, + .end_session = stub_end_session, + .on_dispatch = stub_on_dispatch, + .on_pend = stub_on_pend, + .on_queue_update = stub_on_queue_update, + .on_job_push = stub_on_job_push, + .on_job_state_change = stub_on_job_state_change, + .on_job_update = stub_on_job_update, + .on_config = NULL, /* MAY be NULL (plugin.h) -- not pushed */ + .on_job_rusage = NULL, /* MAY be NULL (plugin.h) -- not pushed */ +}; + +/* Same picker-plugin NAME for both -- safe: each lives in its own core's + * OWN core->plugins kvmgr (see file header comment), so there is no name + * collision. */ +static xce_sched_plugin_t g_stub_plugin_a + = { .name = "test_stub_picker", + .picker_iface = &g_stub_picker_iface, + .data = &g_stub_state_a }; +static xce_sched_plugin_t g_stub_plugin_b + = { .name = "test_stub_picker", + .picker_iface = &g_stub_picker_iface, + .data = &g_stub_state_b }; + +/* ---------------------------------------------------------------------- */ +/* Picker thread: reproduces scheduler.c's sched_job_picker() real */ +/* per-session body verbatim (copied from test_core_full_cycle.c's own */ +/* test_picker_thread_fn -- identical, core-parameterized, single-shot). */ +/* ---------------------------------------------------------------------- */ + +static void* test_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; +} + +/* ---------------------------------------------------------------------- */ +/* Fixture helpers */ +/* ---------------------------------------------------------------------- */ + +static void fixture_init(xce_sched_core_t* core, xce_sched_plugin_t* plugin) +{ + int rc; + + rc = xce_sched_core_init(core); + TEST_ASSERT_EQUAL_INT(0, rc); + + lock(core->plugins, RTTI_LOCK_WRITE); + rc = key_add(core->plugins, (void*)plugin->name, (void*)plugin); + unlock(core->plugins); + TEST_ASSERT_GREATER_OR_EQUAL(0, rc); + + core->job_pick_plugin = plugin->name; +} + +static xce_job_t* make_job(uint64_t jobid, uint32_t state, char* queue, + char* user, char* project) +{ + xce_job_t* j = (xce_job_t*)calloc(1, sizeof(*j)); + + j->jobid = jobid; + j->state = state; + j->sub_time = 1000; + j->queue_time = 1000; + j->start_time = 0; + j->submit = (job_submit_data_t*)calloc(1, sizeof(*(j->submit))); + j->submit->queue = queue; + j->submit->user = user; + j->submit->project_name = project; + j->submit->resreq = "type == any"; + j->submit->mintasks = 1; + j->submit->maxtasks = 1; + + return j; +} + +#define MAX_RECORDED_DECISIONS 8 + +typedef struct { + int calls; + xce_decision_t* seen[MAX_RECORDED_DECISIONS]; +} test_driver_ctx_t; + +static int test_on_decision(void* driver_ctx, xce_decision_t* decision) +{ + test_driver_ctx_t* ctx = (test_driver_ctx_t*)driver_ctx; + + TEST_ASSERT_NOT_NULL(decision); + + if (ctx->calls < MAX_RECORDED_DECISIONS) { + ctx->seen[ctx->calls] = decision; + } + ctx->calls++; + + return 0; +} + +static int drain_and_dispatch(xce_sched_core_t* core) +{ + int delivered = 0; + + for (;;) { + xce_decision_t* decision = NULL; + + cbuf_get(&(core->decisions), &decision); + if (decision == NULL) { + break; + } + + core->on_decision(core->driver_ctx, decision); + delivered++; + } + + return delivered; +} + +/* run_identical_session - drives ONE core through a full session: a real + * picker thread + a stand-in allocator (this function itself, mirroring + * test_core_full_cycle.c's own Phase 3/4/5) that decides DISPATCH (even + * jobid) vs PEND (odd jobid) -- a deterministic, resource-matching-free + * rule, same scope boundary test_core_full_cycle.c's own header comment + * documents for why real sched_allocator()/resource.c internals are out of + * a pure core-contract test's reach. This is what makes "identical events + * in, identical decisions out" a meaningful, reproducible comparison + * between core_a and core_b: given the SAME candidate jobids in the SAME + * order (which the picker rebuild on load is what this whole task proves), + * this stand-in allocator's own logic is a pure function of jobid, so it + * necessarily produces the SAME op+jobid sequence for both cores. */ +static void run_identical_session(xce_sched_core_t* core) +{ + pthread_t picker_tid; + int rc; + uint64_t jobid; + + rc = pthread_create(&picker_tid, NULL, test_picker_thread_fn, core); + TEST_ASSERT_EQUAL_INT(0, rc); + + rc = xce_core_go(core); + TEST_ASSERT_EQUAL_INT(0, rc); + + for (;;) { + xce_sched_job_t* job; + xce_decision_t* decision = NULL; + + cbuf_get(&(core->jbuf), &jobid); + if (jobid == 0) { + cbuf_put(&(core->decisions), &decision); + break; + } + + job = sched_get_job(core, jobid); + TEST_ASSERT_NOT_NULL(job); + + if ((jobid % 2) == 0) { + xce_job_alloc_t* alloc + = (xce_job_alloc_t*)calloc(1, sizeof(*alloc)); + alloc->jobid = jobid; + decision = make_dispatch_decision(alloc); + } else { + xce_pendreason_t* reason = make_pendreason(jobid); + decision = make_pend_decision(reason); + } + + TEST_ASSERT_NOT_NULL(decision); + cbuf_put(&(core->decisions), &decision); + } + + rc = pthread_join(picker_tid, NULL); + TEST_ASSERT_EQUAL_INT(0, rc); + + (void)drain_and_dispatch(core); + + rc = xce_core_end_session(core); + TEST_ASSERT_EQUAL_INT(0, rc); +} + +/* ---------------------------------------------------------------------- */ +/* Guard tests (Part 3's own "-EBUSY/-EINVAL, directly and cheaply, unit- */ +/* level" request) -- no session, no picker plugin needed: both guards */ +/* fire before either function touches state or resolves a picker (core.h */ +/* 's own doc comment). */ +/* ---------------------------------------------------------------------- */ + +static void +test_snapshot_guards_reject_null_args_and_busy_session_without_mutating_state( + void) +{ + xce_sched_core_t core; + sched_snapshot_data_t snap; + sched_snapshot_data_t empty_in; + int rc; + + rc = xce_sched_core_init(&core); + TEST_ASSERT_EQUAL_INT(0, rc); + + memset(&empty_in, 0, sizeof(empty_in)); + + /* -EINVAL: NULL core, or NULL out/in, for both directions. */ + rc = xce_sched_core_save_snapshot(NULL, &snap); + TEST_ASSERT_EQUAL_INT(-EINVAL, rc); + + rc = xce_sched_core_save_snapshot(&core, NULL); + TEST_ASSERT_EQUAL_INT(-EINVAL, rc); + + rc = xce_sched_core_load_snapshot(NULL, &empty_in); + TEST_ASSERT_EQUAL_INT(-EINVAL, rc); + + rc = xce_sched_core_load_snapshot(&core, NULL); + TEST_ASSERT_EQUAL_INT(-EINVAL, rc); + + /* -EBUSY: session_active != 0, checked BEFORE any mutation -- a + * rejected call is a true no-op (core.h's own doc comment for both + * functions, "ZERO state touched"). Directly flip the atomic (same + * precondition-rejection technique test_core_full_cycle.c's own Phase 2 + * uses for xce_core_go()/xce_core_push_events()) rather than driving a + * real session, since only the guard itself is under test here. */ + atomic_store(&(core.session_active), 1); + + memset(&snap, 0xAA, sizeof(snap)); + rc = xce_sched_core_save_snapshot(&core, &snap); + TEST_ASSERT_EQUAL_INT(-EBUSY, rc); + /* sentinel bytes untouched -- save_snapshot() never reached its own + * memset(out, 0, ...), proving the guard runs before any mutation. */ + TEST_ASSERT_EQUAL_UINT8(0xAA, ((uint8_t*)&snap)[0]); + + rc = xce_sched_core_load_snapshot(&core, &empty_in); + TEST_ASSERT_EQUAL_INT(-EBUSY, rc); + /* no queue was created -- proves ZERO state was touched, not just that + * the return code was right. */ + TEST_ASSERT_NULL(sched_get_queue(&core, "normal")); + + atomic_store(&(core.session_active), 0); +} + +/* ---------------------------------------------------------------------- */ +/* The Observable, end to end. */ +/* ---------------------------------------------------------------------- */ + +/* + * test_snapshot_save_then_load_produces_behaviorally_equivalent_core - see + * file header comment for the full design. + */ +static void +test_snapshot_save_then_load_produces_behaviorally_equivalent_core(void) +{ + xce_sched_core_t core_a; + xce_sched_core_t core_b; + sched_snapshot_data_t snap; + xce_core_event_t events[3]; + xce_conf4sched_t config; + xce_resource_t wire_resource; + xce_resource_t* resources[1]; + xce_val_t resval_value; + xce_name_val_t wire_resval; + xce_name_val_t* resvals[1]; + xce_queue_t wire_queue_normal; + xce_queue_t wire_queue_batch; + xce_queue_t* queues[2]; + xce_worker_t wire_worker_a; + xce_worker_t wire_worker_b; + xce_worker_t* workers[2]; + xce_val_t worker_a_cpu_val; + xce_val_t worker_b_cpu_val; + xce_name_val_t worker_a_resval; + xce_name_val_t worker_b_resval; + xce_name_val_t* worker_a_resvals[1]; + xce_name_val_t* worker_b_resvals[1]; + xce_job_t* jobs[3]; + xce_sched_project_t* proj1_a; + xce_resource_t* found_resource_a; + xce_resource_t* found_resource_b; + xce_val_t* shared_val_b; + xce_sched_queue_t* q_normal_b; + xce_sched_queue_t* q_batch_b; + xce_sched_worker_t* worker_a_b; + xce_sched_worker_t* worker_b_b; + xce_val_t* worker_a_resval_b; + xce_val_t* worker_b_resval_b; + xce_sched_project_t* proj1_b; + xce_sched_project_t* proj2_b; + xce_sched_job_t* job201_b; + xce_sched_job_t* job202_b; + xce_sched_job_t* job203_b; + test_driver_ctx_t driver_ctx_a; + test_driver_ctx_t driver_ctx_b; + int rc; + int i; + + memset(&driver_ctx_a, 0, sizeof(driver_ctx_a)); + memset(&driver_ctx_b, 0, sizeof(driver_ctx_b)); + memset(&g_stub_state_a, 0, sizeof(g_stub_state_a)); + memset(&g_stub_state_b, 0, sizeof(g_stub_state_b)); + + /* =================================================================== + * Phase 1: build core_a with REAL state -- jobs, queues, workers, + * resources, resvals, a registered picker plugin. + * =================================================================== */ + fixture_init(&core_a, &g_stub_plugin_a); + core_a.prng_seed = 0xDEADBEEFULL; /* direct field poke -- prng_seed has + * no wire event of its own (task + * 8.1's own scope note, core.h); + * seeded here to a non-zero value so + * the round-trip assertion below is + * meaningful, not a trivial 0==0. */ + + memset(&wire_resource, 0, sizeof(wire_resource)); + wire_resource.name = "cpu"; + resources[0] = &wire_resource; + + memset(&resval_value, 0, sizeof(resval_value)); + /* .which MUST be XCE_VAL_NUMERIC, not left at the memset-zero default + * -- XCE_VAL_STRING is literally 0 (xce/types.h), and dup_val() + * (snapshot_wire.c, genuinely exercised by save_snapshot() below) + * branches on .which to decide whether .u.svalue is a string pointer + * to copy: leaving this at 0 makes it misinterpret .u.ivalue32's raw + * integer bits as a char* and crash inside makeString(). */ + resval_value.which = XCE_VAL_NUMERIC; + resval_value.u.ivalue32 = 8; + memset(&wire_resval, 0, sizeof(wire_resval)); + wire_resval.name = "cpu"; + wire_resval.val = &resval_value; + resvals[0] = &wire_resval; + + memset(&wire_queue_normal, 0, sizeof(wire_queue_normal)); + wire_queue_normal.queue = "normal"; + wire_queue_normal.priority = 3; + wire_queue_normal.status = QUEUE_STAT_OPEN; + + memset(&wire_queue_batch, 0, sizeof(wire_queue_batch)); + wire_queue_batch.queue = "batch"; + wire_queue_batch.priority = 1; + wire_queue_batch.status = QUEUE_STAT_OPEN; + + queues[0] = &wire_queue_normal; + queues[1] = &wire_queue_batch; + + memset(&config, 0, sizeof(config)); + config.n_resources = 1; + config.resources = resources; + config.nresvals = 1; + config.resvals = resvals; + config.n_queues = 2; + config.queues = queues; + + memset(&worker_a_cpu_val, 0, sizeof(worker_a_cpu_val)); + worker_a_cpu_val.which = XCE_VAL_NUMERIC; /* see resval_value's own + * comment above */ + worker_a_cpu_val.u.ivalue32 = 4; + memset(&worker_a_resval, 0, sizeof(worker_a_resval)); + worker_a_resval.name = "cpu"; + worker_a_resval.val = &worker_a_cpu_val; + worker_a_resvals[0] = &worker_a_resval; + + memset(&worker_b_cpu_val, 0, sizeof(worker_b_cpu_val)); + worker_b_cpu_val.which = XCE_VAL_NUMERIC; + worker_b_cpu_val.u.ivalue32 = 6; + memset(&worker_b_resval, 0, sizeof(worker_b_resval)); + worker_b_resval.name = "cpu"; + worker_b_resval.val = &worker_b_cpu_val; + worker_b_resvals[0] = &worker_b_resval; + + memset(&wire_worker_a, 0, sizeof(wire_worker_a)); + wire_worker_a.name = "hostA"; + wire_worker_a.status = 0; /* available: IS_WORKER_UNAVAIL + * checks a specific unset bit */ + wire_worker_a.nresvals_to_sched = 1; + wire_worker_a.resvals_to_sched = worker_a_resvals; + + memset(&wire_worker_b, 0, sizeof(wire_worker_b)); + wire_worker_b.name = "hostB"; + wire_worker_b.status = 0; + wire_worker_b.nresvals_to_sched = 1; + wire_worker_b.resvals_to_sched = worker_b_resvals; + + workers[0] = &wire_worker_a; + workers[1] = &wire_worker_b; + + jobs[0] = make_job(201, JOB_STAT_PEND, "normal", "alice", "proj1"); + jobs[1] = make_job(202, JOB_STAT_PEND, "normal", "bob", "proj1"); + jobs[2] = make_job(203, JOB_STAT_PEND, "batch", "carol", "proj2"); + + memset(events, 0, sizeof(events)); + /* CONFIG_PUSH first: registers "cpu" into core_a->resmgr, which + * WORKERS_PUSH (next) needs already present to resolve each worker's + * resvals_to_sched by name. */ + events[0].type = XCE_CORE_EVENT_CONFIG_PUSH; + events[0].u.config_push.config = &config; + events[1].type = XCE_CORE_EVENT_WORKERS_PUSH; + events[1].u.workers_push.nworkers = 2; + events[1].u.workers_push.workers = workers; + events[2].type = XCE_CORE_EVENT_JOBS_PUSH; + events[2].u.jobs_push.njobs = 3; + events[2].u.jobs_push.jobs = jobs; + + rc = xce_core_push_events(&core_a, events, 3); + TEST_ASSERT_EQUAL_INT(0, rc); + TEST_ASSERT_EQUAL_INT(3, g_stub_state_a.ncandidates); + + /* project priority: no wire event of its own -- direct field poke on + * the auto-created project object (get-or-create side effect of + * JOBS_PUSH), same "genuine, but directly-constructed" state + * test_core_events.c's own fixtures already use for config/queue/ + * resource wire payloads. proj2 deliberately left at its default (0) + * -- proves the round-trip is field-accurate, not just "non-zero + * survives". */ + proj1_a = sched_get_project(&core_a, "proj1"); + TEST_ASSERT_NOT_NULL(proj1_a); + proj1_a->priority = 5; + + /* =================================================================== + * Phase 2: save core_a's snapshot; field-level sanity checks on the + * snapshot struct itself (cheap, direct evidence of the field-by-field + * mapping snapshot.c's own header comment documents). + * =================================================================== */ + memset(&snap, 0, sizeof(snap)); + rc = xce_sched_core_save_snapshot(&core_a, &snap); + TEST_ASSERT_EQUAL_INT(0, rc); + + TEST_ASSERT_NOT_NULL(snap.jobs); + TEST_ASSERT_EQUAL_INT(3, snap.jobs->count); + TEST_ASSERT_EQUAL_INT(2, snap.n_queues); + TEST_ASSERT_NOT_NULL(snap.workers); + TEST_ASSERT_EQUAL_INT(2, snap.workers->count); + TEST_ASSERT_NOT_NULL(snap.resources); + TEST_ASSERT_EQUAL_INT(1, snap.resources->count); + TEST_ASSERT_NOT_NULL(snap.resvals); + TEST_ASSERT_EQUAL_INT(1, snap.resvals->count); + TEST_ASSERT_EQUAL_INT(3, snap.n_users); + TEST_ASSERT_EQUAL_INT(2, snap.n_projects); + TEST_ASSERT_EQUAL_UINT64(0xDEADBEEFULL, snap.prng_seed); + + /* =================================================================== + * Phase 3: build core_b -- a completely separate, freshly-initialized + * instance -- and load the snapshot into it. + * =================================================================== */ + fixture_init(&core_b, &g_stub_plugin_b); + + rc = xce_sched_core_load_snapshot(&core_b, &snap); + TEST_ASSERT_EQUAL_INT(0, rc); + + /* picker rebuild fired on_job_push for every restored job, exactly as + * the original JOBS_PUSH did for core_a above. */ + TEST_ASSERT_EQUAL_INT(3, g_stub_state_b.ncandidates); + + /* =================================================================== + * Phase 4: field-level assertions -- core_b's restored state matches + * core_a's original values exactly. + * =================================================================== */ + q_normal_b = sched_get_queue(&core_b, "normal"); + q_batch_b = sched_get_queue(&core_b, "batch"); + TEST_ASSERT_NOT_NULL(q_normal_b); + TEST_ASSERT_NOT_NULL(q_batch_b); + TEST_ASSERT_EQUAL_INT(3, q_normal_b->priority); + TEST_ASSERT_EQUAL_INT(1, q_batch_b->priority); + + rc = key_find(core_a.resmgr, (void*)"cpu", (void**)&found_resource_a); + TEST_ASSERT_GREATER_THAN_INT(0, rc); + rc = key_find(core_b.resmgr, (void*)"cpu", (void**)&found_resource_b); + TEST_ASSERT_GREATER_THAN_INT(0, rc); + TEST_ASSERT_NOT_NULL(found_resource_a); + TEST_ASSERT_NOT_NULL(found_resource_b); + + shared_val_b + = (xce_val_t*)zarray_get(&(core_b.sharedInst), found_resource_b->idx); + TEST_ASSERT_NOT_NULL(shared_val_b); + TEST_ASSERT_EQUAL_INT(8, shared_val_b->u.ivalue32); + + worker_a_b = sched_get_worker(&core_b, "hostA"); + worker_b_b = sched_get_worker(&core_b, "hostB"); + TEST_ASSERT_NOT_NULL(worker_a_b); + TEST_ASSERT_NOT_NULL(worker_b_b); + worker_a_resval_b + = (xce_val_t*)zarray_get(&(worker_a_b->resvals), found_resource_b->idx); + worker_b_resval_b + = (xce_val_t*)zarray_get(&(worker_b_b->resvals), found_resource_b->idx); + TEST_ASSERT_NOT_NULL(worker_a_resval_b); + TEST_ASSERT_NOT_NULL(worker_b_resval_b); + TEST_ASSERT_EQUAL_INT(4, worker_a_resval_b->u.ivalue32); + TEST_ASSERT_EQUAL_INT(6, worker_b_resval_b->u.ivalue32); + + TEST_ASSERT_NOT_NULL(sched_get_user(&core_b, "alice")); + TEST_ASSERT_NOT_NULL(sched_get_user(&core_b, "bob")); + TEST_ASSERT_NOT_NULL(sched_get_user(&core_b, "carol")); + + proj1_b = sched_get_project(&core_b, "proj1"); + proj2_b = sched_get_project(&core_b, "proj2"); + TEST_ASSERT_NOT_NULL(proj1_b); + TEST_ASSERT_NOT_NULL(proj2_b); + TEST_ASSERT_EQUAL_INT(5, proj1_b->priority); + TEST_ASSERT_EQUAL_INT(0, proj2_b->priority); + + TEST_ASSERT_EQUAL_UINT64(0xDEADBEEFULL, core_b.prng_seed); + + job201_b = sched_get_job(&core_b, 201); + job202_b = sched_get_job(&core_b, 202); + job203_b = sched_get_job(&core_b, 203); + TEST_ASSERT_NOT_NULL(job201_b); + TEST_ASSERT_NOT_NULL(job202_b); + TEST_ASSERT_NOT_NULL(job203_b); + TEST_ASSERT_EQUAL_UINT32(JOB_STAT_PEND, job201_b->state); + TEST_ASSERT_EQUAL_STRING("normal", job201_b->queue->name); + TEST_ASSERT_EQUAL_STRING("alice", job201_b->user->name); + TEST_ASSERT_EQUAL_STRING("proj1", job201_b->project->name); + TEST_ASSERT_EQUAL_STRING("batch", job203_b->queue->name); + TEST_ASSERT_EQUAL_STRING("carol", job203_b->user->name); + TEST_ASSERT_EQUAL_STRING("proj2", job203_b->project->name); + + /* =================================================================== + * Phase 5: THE cross-core behavioral-equivalence proof. Drive core_a + * and core_b through an IDENTICAL session (same picker-thread protocol, + * same deterministic stand-in-allocator rule) and compare the + * resulting decision sequences. + * =================================================================== */ + rc = xce_core_set_decision_cb(&core_a, test_on_decision, &driver_ctx_a); + TEST_ASSERT_EQUAL_INT(0, rc); + rc = xce_core_set_decision_cb(&core_b, test_on_decision, &driver_ctx_b); + TEST_ASSERT_EQUAL_INT(0, rc); + + run_identical_session(&core_a); + run_identical_session(&core_b); + + TEST_ASSERT_EQUAL_INT(3, driver_ctx_a.calls); + TEST_ASSERT_EQUAL_INT(driver_ctx_a.calls, driver_ctx_b.calls); + + for (i = 0; i < driver_ctx_a.calls; i++) { + xce_decision_t* da = driver_ctx_a.seen[i]; + xce_decision_t* db = driver_ctx_b.seen[i]; + + TEST_ASSERT_NOT_NULL(da); + TEST_ASSERT_NOT_NULL(db); + TEST_ASSERT_EQUAL_INT(da->op, db->op); + + if (da->op == XCE_DECISION_DISPATCH) { + TEST_ASSERT_EQUAL_UINT64(da->u.alloc->jobid, db->u.alloc->jobid); + } else { + TEST_ASSERT_EQUAL_INT(XCE_DECISION_PEND, da->op); + TEST_ASSERT_EQUAL_UINT64(da->u.reason->jobid, db->u.reason->jobid); + } + } + + /* Spot-check the actual expected sequence (jobid parity -- see + * run_identical_session()'s own comment): 201 odd -> PEND, 202 even -> + * DISPATCH, 203 odd -> PEND, in push/candidate order. Proven identical + * for BOTH cores by the loop above; this just pins the concrete values + * so a regression here fails loudly with a readable expected/actual + * pair instead of only an A==B equality. */ + TEST_ASSERT_EQUAL_INT(XCE_DECISION_PEND, driver_ctx_a.seen[0]->op); + TEST_ASSERT_EQUAL_UINT64(201, driver_ctx_a.seen[0]->u.reason->jobid); + TEST_ASSERT_EQUAL_INT(XCE_DECISION_DISPATCH, driver_ctx_a.seen[1]->op); + TEST_ASSERT_EQUAL_UINT64(202, driver_ctx_a.seen[1]->u.alloc->jobid); + TEST_ASSERT_EQUAL_INT(XCE_DECISION_PEND, driver_ctx_a.seen[2]->op); + TEST_ASSERT_EQUAL_UINT64(203, driver_ctx_a.seen[2]->u.reason->jobid); +} + +int main(void) +{ + UNITY_BEGIN(); + RUN_TEST( + test_snapshot_guards_reject_null_args_and_busy_session_without_mutating_state); + RUN_TEST( + test_snapshot_save_then_load_produces_behaviorally_equivalent_core); + return UNITY_END(); +}