diff --git a/.kiro/specs/scheduler-core/tasks.md b/.kiro/specs/scheduler-core/tasks.md index 21a5f0fa..dcead18e 100644 --- a/.kiro/specs/scheduler-core/tasks.md +++ b/.kiro/specs/scheduler-core/tasks.md @@ -315,12 +315,13 @@ - _Depends: 8.1, 7.5_ - _Verified: 2026-08-06 — new `sim.h`/`sim.c` implement AC 13 steps 1-4 (`sched_sim_driver_start()`): open+validate `ReplayHeader`, scan for `SnapshotEntry` messages tracking the most recent with `ts <= --sim-start-ts`, call `xce_sched_core_load_snapshot()`, set virtual clock anchors from `core_clock.c`'s real field names. Step 5 (event loop) deliberately deferred to task 11.2 via an `-ENOSYS` placeholder in `sched_sim_driver_run()`. All 7 CLI flags (`--sim`, `--sim-start-ts`, `--sim-speed`, `--sim-verify`, `--sim-alt-events`, `--sim-output`, `--interactive`/`-i`) wired into `main.c`'s existing task-5.2 mode-selection branch, replacing its `-ENOSYS` stub with real startup logic. Two real bugs found+fixed during TDD, both independently reproduced by review: a `SnapshotEntry_get_ts()` use-after-free (read through a `capn_ptr` after `capn_free()` — reviewer reverted the fix and reproduced a genuine ASan heap-UAF, then restored it) and a `pend_updates[0]/[1]` NULL-deref in `xce_sched_core_fini()` for any bare-core caller, worked around by allocating+`dataq_init()`ing them right after `xce_sched_core_init()`, matching `scheduler.c`'s existing `sched_init()` pattern. Deliberately does not register a picker plugin (no `xce_scheduler_t*` available under AC 17 isolation) — relies on `load_snapshot()`'s own documented graceful degradation; a working picker becomes load-bearing only once 11.2 drives real sessions. Filenames (`sim.h`/`sim.c`, not design.md's literal `sim_driver.h`/`sim_driver.c`) and a concrete `sched_sim_args_t*` (not design.md's literal `const void*`) are documented, justified deviations preserving the mandated struct/function contract. 27/27 scheduler suite (26 baseline + 6 new `sched_sim_startup_test` sub-tests, individually ASan-clean), full project build clean, RED phase independently re-reproduced by review (moved `sim.c` aside, confirmed the same missing-file build failure). Review APPROVED with one non-blocking coverage gap noted (multi-snapshot `ts`-selection logic — correct by code trace and by the reviewer's own bug-reversion experiment, but not yet covered by a dedicated 2+-snapshot fixture test) and two FYI-only naming/signature deviations, both already justified in `sim.h`'s header comment. Follow-up for a later task: `--sim-start-ts`/`--sim-speed` parse via `strtoull()`, which silently wraps a leading `-` into a huge unsigned value instead of a clear parse error._ -- [ ] 11.2 Implement the simulation event loop and virtual clock advancement +- [x] 11.2 Implement the simulation event loop and virtual clock advancement - Advance the virtual clock by jumping directly to each queued event's timestamp — no fixed tick, strictly monotonic, same-timestamp events processed in file order - Treat each session-boundary message as the trigger for one complete scheduling round: push accumulated events, start the session, drain decisions, end the session - Observable: replaying a fixture file with events at non-uniform timestamps processes them in file order with the virtual clock reporting each event's exact timestamp at the moment it's pushed - _Requirements: 5.14, 5.18_ - _Depends: 11.1_ + - _Verified: 2026-08-06 — new `sim_loop.c` (562 lines) implements `sched_sim_driver_run()` (replacing task 11.1's `-ENOSYS` placeholder): decodes `.replay` messages in file order, jumps the virtual clock to each `ReplayEvent`'s exact timestamp as it's queued, and on each `ReplaySession` boundary runs a real `core_push_events→core_go→drain decisions→core_end_session` cycle before resetting the batch; mid-stream `SnapshotEntry` messages correctly skipped per AC 20. Discovered task 11.1 left no picker registered and no thread backing the core's cbufs, which would deadlock the decision-drain loop unconditionally the moment a real event loop existed — necessarily in-scope, not a convenience shortcut (independently confirmed by review). Fixed via new `plugin_register_core()` (`plugin.c`/`plugin.h`, registers a picker against a bare `xce_sched_core_t*`, no `xce_scheduler_t*` needed, precedent-matched to task 4.6's `test_core_full_cycle.c`) + `builtin_fcfs_new_plugin()` factory extracted from the existing `register_builtin_fcfs()` (`plugins/builtin_fcfs.c`) + a persistent picker-driving thread (`sim_picker_thread_fn`). Review independently verified the extraction is genuinely additive and behavior-preserving on the LIVE normal-mode path (32/32 whole-project tests including controller/libxce, not just the scheduler suite) — the only finding was a non-blocking doc-accuracy nit (`plugin_register_core()`'s doc comment overclaims parity with the legacy `job_elect_fn` adapter step, which it correctly omits since its only caller never uses that path). Since real resource-matching allocation is unreachable without a genuine `xce_scheduler_t*` (a pre-existing gap belonging to the not-yet-designed scheduler-allocation spec), `sim_picker_thread_fn()` conservatively PENDs every candidate rather than fabricating a DISPATCH — judged by review as honestly-scoped and sufficient for this task's narrow ordering/clock-timing Observable (AC 18's pseudocode drains "decisions" generically, not specifically dispatches), but flagged as a real gap once `--sim`'s decisions need to be meaningful (task 11.4/scheduler-allocation's problem, not this task's). RED phase independently re-reproduced by review (reverted to `-ENOSYS`, confirmed real test failure, restored). 28/28 scheduler suite, 32/32 whole-project, clang-format clean. Non-blocking follow-up noted: the PEND-candidate loop itself has no direct test yet (both new tests exercise the "no resolvable picker" branch only) — worth covering when task 11.4 needs that path exercised for real._ - [ ] 11.3 Enforce simulation isolation - The simulation driver's state holds no live-controller client context, no cluster configuration, and no capnp socket of any kind — confirmed by the struct definition itself, not runtime checks alone diff --git a/src/scheduler/include/plugin.h b/src/scheduler/include/plugin.h index 4f21f27a..f9042da7 100644 --- a/src/scheduler/include/plugin.h +++ b/src/scheduler/include/plugin.h @@ -123,6 +123,43 @@ extern int register_parjob_allocator(void*); * init() below only ever holds/passes a pointer to it. */ typedef struct xce_sched_core_s xce_sched_core_t; +/* + * builtin_fcfs_new_plugin - factory half of register_builtin_fcfs()'s + * template-build step (plugins/builtin_fcfs.c), extracted so a caller with + * no xce_scheduler_t* (e.g. the simulation driver, via + * plugin_register_core() below) can obtain a freshly calloc'd, fully + * populated `xce_sched_plugin_t` for builtin_fcfs -- same picker_iface/ + * init_fn/fini_fn template register_builtin_fcfs() itself uses, just not + * yet registered anywhere. Returns NULL on allocation failure. Caller owns + * the returned plugin (frees it only on a subsequent registration failure + * -- once registered, the plugin registry owns it, same lifetime rule + * every other plugin category in this codebase already follows). + */ +extern xce_sched_plugin_t* builtin_fcfs_new_plugin(void); + +/* + * plugin_register_core - Req 5 AC 17 (scheduler-core spec, task 11.2): the + * "core.h-scoped plugin registration path" sim.h's own doc comment on + * sched_sim_driver_start() anticipated but did not build (task 11.1's + * documented "Known scope boundary"). plugin_register() above requires a + * genuine `xce_scheduler_t*` (it dereferences `sched->core.plugins` and + * stores `plugin->sched = sched` for plugin_unregister()'s later use) -- + * unavailable to a driver like the simulation driver that deliberately has + * no such type. This is the SAME registration mechanism minus that one + * dependency: run the plugin's init_fn (same "driver thread, at plugin + * register" timing plugin.h's own hook-category doc comment documents), + * then key_add() it into `core->plugins` directly -- identical steps + * plugin_register() itself performs, in the same order, just addressed at + * xce_sched_core_t* instead of xce_scheduler_t*. `plugin->sched` is left + * NULL (no xce_scheduler_t* exists to store) -- safe, since + * plugin_unregister() is a no-op on a NULL `plugin->sched` and nothing + * calls it for a plugin registered this way (no sim-driver teardown path + * exists yet). Returns the assigned plugin_id (>= 0) on success, or a + * negative errno: -EINVAL for a NULL core/plugin, whatever init_fn + * returned if it failed, or key_add()'s own negative return. + */ +extern int plugin_register_core(xce_sched_core_t*, xce_sched_plugin_t*); + /* * Hook categories and the thread each category runs in (Req 1 AC 3): * diff --git a/src/scheduler/include/sim.h b/src/scheduler/include/sim.h index c06c81fb..315d20c0 100644 --- a/src/scheduler/include/sim.h +++ b/src/scheduler/include/sim.h @@ -205,22 +205,19 @@ extern int sched_sim_driver_start(xce_sched_sim_driver_t* sim, * void* args)`, adapted to take the concrete sched_sim_args_t* this header * defines rather than an opaque void* -- see sched_sim_args_t's own doc * comment for why). This is what main.c's mode-selection branch calls when - * `--sim` is present (design.md's own Integration note). + * `--sim` is present (design.md's own Integration note). Defined in + * sim_loop.c (scheduler-core spec, task 11.2), not sim.c: calls + * sched_sim_driver_start() above (task 11.1) then runs Req 5 AC 13 step + * 5's event loop (Req 5 AC 14/18) -- decode each `.replay` message in file + * order, jump the virtual clock to each ReplayEvent's own timestamp as it + * is queued (AC 14), and on each ReplaySession boundary run one complete + * push-events -> start-session -> drain-decisions -> end-session round. * - * Calls sched_sim_driver_start() above (task 11.1's own work, this task's - * Observable) then -- since task 11.2's event loop (Req 5 AC 13 step 5 / - * AC 14/18) does not exist yet -- fails loudly with -ENOSYS rather than - * silently returning success having processed zero events. This is the - * same "honest, temporary placeholder" precedent main.c's own pre-task-11.1 - * `--sim` stub already established (task 5.2): startup is now genuinely - * real, working behavior; only the event loop remains a documented gap for - * task 11.2 to fill in, without requiring another main.c edit when it does - * (task 11.2 only needs to extend THIS function's body, after the - * sched_sim_driver_start() call, with the real event loop). - * - * Returns 0 (once task 11.2 lands) or a negative errno: whatever - * sched_sim_driver_start() returned on a startup failure, or -ENOSYS if - * startup succeeded but the event loop is not yet implemented. + * Returns 0 once the replay stream is exhausted (its ReplayEvent{endSession} + * marker is reached, Req 5 AC 20), or a negative errno: whatever + * sched_sim_driver_start() returned on a startup failure, or a pthread + * error standing up the picker-driving thread the event loop needs (see + * sim_loop.c's own file header comment for why one exists). */ extern int sched_sim_driver_run(xce_sched_sim_driver_t* sim, const sched_sim_args_t* args); diff --git a/src/scheduler/meson.build b/src/scheduler/meson.build index f1c3591e..b61cdb40 100644 --- a/src/scheduler/meson.build +++ b/src/scheduler/meson.build @@ -15,6 +15,7 @@ scheduler_srcs = [ 'src/recording_events.c', 'src/recording_session.c', 'src/sim.c', + 'src/sim_loop.c', 'src/schedobj.c', 'src/plugin.c', 'src/resource.c', diff --git a/src/scheduler/src/plugin.c b/src/scheduler/src/plugin.c index a5bc97ea..a27b44d9 100644 --- a/src/scheduler/src/plugin.c +++ b/src/scheduler/src/plugin.c @@ -167,6 +167,35 @@ int plugin_register(void* arg1, void* arg2) return rc; } +/* See plugin.h's declaration comment for the full contract -- the same + * init_fn + key_add(core->plugins) steps plugin_register() performs above, + * minus the xce_scheduler_t*-only plugin->sched bookkeeping. */ +int plugin_register_core(xce_sched_core_t* core, xce_sched_plugin_t* plugin) +{ + int rc = 0; + + if ((core == NULL) || (plugin == NULL)) { + return -EINVAL; + } + + if (plugin->init_fn != NULL) { + rc = plugin->init_fn(plugin); + if (rc < 0) { + return rc; + } + } + + lock(core->plugins, RTTI_LOCK_WRITE); + rc = key_add(core->plugins, (void*)plugin->name, (void*)plugin); + unlock(core->plugins); + + if (rc >= 0) { + plugin->plugin_id = rc; + } + + return rc; +} + void plugin_unregister(void* arg) { xce_sched_plugin_t* plugin = (xce_sched_plugin_t*)arg; diff --git a/src/scheduler/src/plugins/builtin_fcfs.c b/src/scheduler/src/plugins/builtin_fcfs.c index 8f1c3bb8..5596f37c 100644 --- a/src/scheduler/src/plugins/builtin_fcfs.c +++ b/src/scheduler/src/plugins/builtin_fcfs.c @@ -982,7 +982,18 @@ static xce_job_picker_iface_t builtin_fcfs_picker_iface = { * != NULL))`) never fires for builtin_fcfs -- it registers as-is, no * deprecation warning. */ -int register_builtin_fcfs(void* sched) +/* + * builtin_fcfs_new_plugin - factory half of the template-build step below, + * extracted (scheduler-core spec, task 11.2) so a caller with no + * xce_scheduler_t* -- the simulation driver, via plugin_register_core() + * (plugin.c) -- can build the exact same real builtin_fcfs plugin instance + * register_builtin_fcfs() itself registers, without going through + * plugin_register()'s xce_scheduler_t*-only bookkeeping. See plugin.h's + * own declaration comment for the full contract. Behavior-preserving + * extraction: register_builtin_fcfs() below is unchanged in what it builds + * and registers, just delegates the build step here. + */ +xce_sched_plugin_t* builtin_fcfs_new_plugin(void) { static xce_sched_plugin_t builtin_fcfs_plugin_template = { .name = "builtin_fcfs", @@ -995,10 +1006,21 @@ int register_builtin_fcfs(void* sched) = (xce_sched_plugin_t*)calloc(1, sizeof(*plugin)); if (plugin == NULL) { - return -ENOMEM; + return NULL; } memcpy(plugin, &builtin_fcfs_plugin_template, sizeof(*plugin)); + return plugin; +} + +int register_builtin_fcfs(void* sched) +{ + xce_sched_plugin_t* plugin = builtin_fcfs_new_plugin(); + + if (plugin == NULL) { + return -ENOMEM; + } + return plugin_register(sched, plugin); } diff --git a/src/scheduler/src/sim.c b/src/scheduler/src/sim.c index 2c2c489a..829ec90d 100644 --- a/src/scheduler/src/sim.c +++ b/src/scheduler/src/sim.c @@ -13,9 +13,12 @@ * sim.c - simulation-mode driver startup (scheduler-core spec, task 11.1, * Req 5 AC 13/17/20). See sim.h for the full per-function contract. This * file implements Req 5 AC 13's five-step startup sequence, steps 1-4 only - * (step 5, the event loop, is task 11.2's job -- sched_sim_driver_run() - * below fails loudly with -ENOSYS after a successful startup, rather than - * silently no-op'ing). + * -- step 5, the event loop, is sim_loop.c's job (task 11.2), which also + * hosts sched_sim_driver_run() itself (design.md's literal SimDriver entry + * point: sched_sim_driver_start() below, then the real event loop). + * msg_root_has_shape() below is shared with sim_loop.c (not static) since + * both files need the identical shape-check technique this comment + * documents. * * Message-type discrimination (Req 5 AC 20): a `.replay` stream has no * self-describing type tag for any message -- position alone pins down the @@ -48,8 +51,8 @@ * allocates; safe to call speculatively before deciding whether to fully * decode a message. */ -static int msg_root_has_shape(struct capn* c, capn_ptr* out, size_t want_words, - size_t want_ptrs) +int msg_root_has_shape(struct capn* c, capn_ptr* out, size_t want_words, + size_t want_ptrs) { capn_ptr p = capn_getp(capn_root(c), 0, 1); @@ -426,25 +429,10 @@ int sched_sim_driver_start(xce_sched_sim_driver_t* sim, return 0; } -int sched_sim_driver_run(xce_sched_sim_driver_t* sim, - const sched_sim_args_t* args) -{ - int rc = sched_sim_driver_start(sim, args); - - if (rc < 0) { - return rc; - } - - /* Task 11.2 (scheduler-core spec, Req 5 AC 14/18) owns the event loop -- - * not yet implemented. Startup (task 11.1, this file's own work) is - * genuinely complete and correct up to this point: sim->core holds the - * fully restored, ready-to-run state and virtual clock. Fail loudly - * rather than silently returning success having processed zero events - * -- same "honest, temporary placeholder" precedent main.c's own - * pre-task-11.1 --sim stub already established (task 5.2). */ - ERROR("%s simulation startup complete but the event loop is not yet " - "implemented (scheduler-core spec task 11.2)", - __func__); - - return -ENOSYS; -} +/* + * sched_sim_driver_run() -- design.md's literal SimDriver entry point -- + * moved to sim_loop.c (scheduler-core spec, task 11.2): its body is now the + * real event loop (Req 5 AC 14/18), no longer a thin start()-then-stub + * wrapper, and sim.c stays scoped to Req 5 AC 13's startup sequence (this + * file's own header comment). Its declaration is unchanged in sim.h. + */ diff --git a/src/scheduler/src/sim_loop.c b/src/scheduler/src/sim_loop.c new file mode 100644 index 00000000..731e8a79 --- /dev/null +++ b/src/scheduler/src/sim_loop.c @@ -0,0 +1,562 @@ +#include +#include +#include +#include +#include +#include + +#include "controller.capnp.h" +#include "plugin.h" +#include "preason.h" +#include "sim.h" +#include "utils.h" +#include "xstring.h" + +/* + * sim_loop.c - simulation-mode driver event loop (scheduler-core spec, task + * 11.2, Req 5 AC 14/18). Completes Req 5 AC 13's five-step startup + * sequence step 5 (sim.c/task 11.1 implements steps 1-4) and hosts + * sched_sim_driver_run() itself -- the design.md-literal SimDriver entry + * point sim.h declares. + * + * Scope boundary vs. sibling tasks 11.4/11.5 (both depend on this task): + * task 11.4 owns comparing each drained decision against the recorded + * ReplaySession.decisions when --sim-verify is set; task 11.5 owns writing + * decisions to a --sim-output file and loading --sim-alt-events. Neither + * branch is built here -- sim_print_decision() below is this task's own, + * narrower Req 5 AC 19 default ("if --sim-output is not specified, + * decisions are printed to stdout only"). + * + * Picker-driving thread (sim_picker_thread_fn, below): Req 5 AC 18's + * pseudocode calls `core_go()` then blocks draining `core->decisions` until + * the end-of-session NULL marker -- exactly like the normal-mode driver's + * own protocol (core_session.c's doc comments on xce_core_go()/ + * xce_core_end_session()), which only works because sched_job_picker()/ + * sched_allocator() (scheduler.c) are two real, persistent threads started + * once at sched_init() and living for the scheduler's whole lifetime. The + * simulation driver (xce_sched_sim_driver_t, sim.h) deliberately has no + * xce_scheduler_t* (Req 5 AC 17) and therefore cannot use either of those + * xce_scheduler_t*-typed thread entry points -- without SOME equivalent + * thread, the very first ReplaySession boundary in any fixture would call + * xce_core_go() and then block forever on cbuf_get(&core->decisions), since + * nothing would ever produce the closing NULL marker. sim_picker_thread_fn() + * below is that equivalent: core-scoped (xce_sched_core_t*, not + * xce_scheduler_t*), spawned once here, living for the simulation driver's + * whole run -- see that function's own doc comment for why it also stands + * in for the allocator role (no separate thread/jbuf hand-off), and for the + * documented, out-of-this-task's-scope reason every candidate job it sees + * is conservatively PENDed rather than DISPATCHed with a fabricated + * allocation. + */ + +/* Shared with sim.c (task 11.1) -- see that file's own file header comment + * for the full message-shape discrimination technique. Declared locally + * here rather than in sim.h (which must stay capnp-type-free -- main.c and + * other non-capnp-aware callers include it), the same "extern declared + * where consumed" convention recording_events.c/recording_session.c + * already establish for write_capnp_message() (recording.c). */ +extern int msg_root_has_shape(struct capn* c, capn_ptr* out, size_t want_words, + size_t want_ptrs); + +/* + * sim_advance_virtual_clock - Req 5 AC 14 (task 11.2): re-anchors + * sim->core's virtual clock so xce_core_now(&sim->core) reports exactly + * new_virtual_ts at the instant this call returns (elapsed=0 against a + * freshly read real_clock_base), then evolves from there per whatever + * speed_num/speed_den sched_sim_driver_start() already set (task 4.5's + * two-anchor + speed formula, core_clock.c) -- speed settings are left + * untouched here, only the two clock anchors jump. Same driver-level + * clock_gettime() legitimacy as sim.c's own sim_set_virtual_clock() (task + * 11.1, Step 4): core.c's "zero wall-clock calls" isolation boundary + * (test_core_isolation.c) is scoped to core.c/core.h's own source text, not + * to driver-level code like this one that legitimately re-seeds the core's + * clock anchors as the replay stream advances (Req 5 AC 14: "jumping + * directly to the timestamp of the next queued event"). + */ +static void sim_advance_virtual_clock(xce_sched_sim_driver_t* sim, + uint64_t new_virtual_ts) +{ + struct timespec ts; + uint64_t real_now_ns; + + clock_gettime(CLOCK_MONOTONIC, &ts); + real_now_ns = ((uint64_t)ts.tv_sec * 1000000000ULL) + (uint64_t)ts.tv_nsec; + + atomic_store_explicit(&(sim->core.virtual_clock_base), new_virtual_ts, + memory_order_release); + atomic_store_explicit(&(sim->core.real_clock_base), real_now_ns, + memory_order_release); +} + +/* + * replay_event_to_core_event - translates a decoded `replay_event_t` (the + * .replay wire event, task 7.2's C mirror of ReplayEvent) into the core's + * own xce_core_event_t (core.h, task 4.3) -- the replay-path analog of + * driver.c's translate_one_pending_update(), which performs the identical + * job for the LIVE controller-push path (xce_sched_update_request_t -> + * xce_core_event_t). The two source shapes differ (wire-decoded + * replay_event_t vs. the driver's own xce_sched_update_request_t wrapper) + * so the field-by-field mapping is not literally shared code, but every + * mapping below was verified against the SAME xce_core_event_t union + * core_events.c actually consumes -- not assumed, exactly as driver.c's own + * file header comment documents for its table. + * + * Returns 1 and populates *ev on a translatable body_which; returns 0 + * (nothing pushed) for a payload the batch cannot use: a malformed/empty + * wrapper (defensively guarded, mirroring translate_one_pending_update()'s + * own NULL guards) or an unrecognized body_which. ReplayEvent_body_endSession + * is never passed here -- the caller (sim_process_replay_event_message(), + * below) intercepts it as the whole-stream termination marker before + * translation is ever attempted (Req 5 AC 20). + */ +static int replay_event_to_core_event(const replay_event_t* src, + xce_core_event_t* ev) +{ + memset(ev, 0, sizeof(*ev)); + ev->ts = src->ts; + + switch (src->body_which) { + case ReplayEvent_body_configPush: + if (src->body.config_push == NULL) { + WARN("%s configPush: NULL config payload, skipping", __func__); + return 0; + } + ev->type = XCE_CORE_EVENT_CONFIG_PUSH; + ev->u.config_push.config = src->body.config_push; + return 1; + + case ReplayEvent_body_jobsPush: + if (src->body.jobs_push == NULL) { + WARN("%s jobsPush: NULL jobs payload, skipping", __func__); + return 0; + } + ev->type = XCE_CORE_EVENT_JOBS_PUSH; + ev->u.jobs_push.njobs = src->body.jobs_push->count; + ev->u.jobs_push.jobs = src->body.jobs_push->jobs; + return 1; + + case ReplayEvent_body_workersPush: + if (src->body.workers_push == NULL) { + WARN("%s workersPush: NULL workers payload, skipping", __func__); + return 0; + } + ev->type = XCE_CORE_EVENT_WORKERS_PUSH; + ev->u.workers_push.nworkers = src->body.workers_push->count; + ev->u.workers_push.workers = src->body.workers_push->workers; + return 1; + + case ReplayEvent_body_resourcesPush: + if (src->body.resources_push == NULL) { + WARN("%s resourcesPush: NULL resources payload, skipping", + __func__); + return 0; + } + ev->type = XCE_CORE_EVENT_RESOURCES_PUSH; + ev->u.resources_push.n = src->body.resources_push->count; + ev->u.resources_push.resources = src->body.resources_push->resources; + return 1; + + case ReplayEvent_body_resvalsFull: + if (src->body.resvals_full == NULL) { + WARN("%s resvalsFull: NULL resvals payload, skipping", __func__); + return 0; + } + ev->type = XCE_CORE_EVENT_RESVALS_FULL; + ev->u.resvals_full.n = src->body.resvals_full->count; + ev->u.resvals_full.resvals = src->body.resvals_full->vals; + return 1; + + case ReplayEvent_body_resvalsDelta: + ev->type = XCE_CORE_EVENT_RESVALS_DELTA; + ev->u.resvals_delta.deltas = src->body.resvals_delta; + return 1; + + case ReplayEvent_body_queueUpdate: + if (src->body.queue_update == NULL) { + WARN("%s queueUpdate: NULL queue payload, skipping", __func__); + return 0; + } + ev->type = XCE_CORE_EVENT_QUEUE_UPDATE; + ev->u.queue_update.queue = src->body.queue_update; + return 1; + + case ReplayEvent_body_jobRusage: + if (src->body.job_rusage == NULL) { + WARN("%s jobRusage: NULL rusages payload, skipping", __func__); + return 0; + } + ev->type = XCE_CORE_EVENT_JOB_RUSAGE; + ev->u.job_rusage.jobid = src->body.job_rusage->jobid; + ev->u.job_rusage.rusages = src->body.job_rusage; + return 1; + + case ReplayEvent_body_jobStateChange: + if (src->body.job_state_change == NULL) { + WARN("%s jobStateChange: NULL payload, skipping", __func__); + return 0; + } + ev->type = XCE_CORE_EVENT_JOB_STATE_CHANGE; + ev->u.job_state_change.jobid = src->body.job_state_change->jobid; + ev->u.job_state_change.old_state + = src->body.job_state_change->old_state; + ev->u.job_state_change.new_state + = src->body.job_state_change->new_state; + ev->u.job_state_change.exit_code + = src->body.job_state_change->exit_code; + return 1; + + case ReplayEvent_body_jobUpdate: + if (src->body.job_update == NULL) { + WARN("%s jobUpdate: NULL payload, skipping", __func__); + return 0; + } + ev->type = XCE_CORE_EVENT_JOB_UPDATE; + ev->u.job_update.jobid = src->body.job_update->jobid; + ev->u.job_update.changed_fields = src->body.job_update->changed_fields; + ev->u.job_update.job = src->body.job_update->job; + return 1; + + default: + WARN("%s unrecognized ReplayEvent body_which %d, skipping", __func__, + src->body_which); + return 0; + } +} + +/* + * sim_process_replay_event_message - decodes one ReplayEvent-shaped capnp + * message already positioned via `c`/`root` (msg_root_has_shape() already + * matched), advances sim's virtual clock to that event's own timestamp + * (Req 5 AC 14 -- "the virtual clock reporting each event's exact + * timestamp at the moment it's pushed"), and translates it into *out_ev + * for the caller's accumulating batch. Split out of sched_sim_driver_run() + * as its own, directly unit-testable function -- same "small, focused, + * independently testable piece of a larger driver loop" precedent sim.c's + * own sim_open_and_validate_header()/sim_locate_and_load_snapshot() split + * already establishes for task 11.1. + * + * Frees `c`'s capn arena before returning (matches every other per-message + * decode in this module, sim.c's own SnapshotEntry/ReplayHeader handling) + * -- decode_ReplayEvent() deep-copies every payload out of the arena into + * plain heap structures first, so this is always safe. + * + * Returns 1 and populates *out_ev if the event was decoded and is a + * translatable, non-terminal event; returns 0 if it decoded but was not + * translatable (see replay_event_to_core_event()); returns -1 if this + * message IS the whole-stream termination marker (ReplayEvent_body_endSession, + * Req 5 AC 20) -- the caller must stop reading, not accumulate anything. + * + * Ownership: the decoded replay_event_t's payload is never freed here (nor + * by the caller) -- it is aliased directly into *out_ev's pointers, which + * must stay valid until xce_core_push_events() consumes the accumulated + * batch at the next ReplaySession boundary. Deliberately not tracked for + * later release: unlike driver.c's release_translated_payload() (which + * exists because the LIVE scheduler runs indefinitely and must not leak + * per live round), a `--sim` run processes one bounded fixture file and + * exits -- this mirrors sim.c's own already-established "accepted leak, + * detect_leaks=0 by design" precedent for the snapshot-load path + * (sim_locate_and_load_snapshot()'s doc comment). + */ +static int sim_process_replay_event_message(xce_sched_sim_driver_t* sim, + struct capn* c, capn_ptr root, + xce_core_event_t* out_ev) +{ + ReplayEvent_ptr ep; + struct ReplayEvent wire; + replay_event_t decoded; + + ep.p = root; + read_ReplayEvent(&wire, ep); + memset(&decoded, 0, sizeof(decoded)); + decode_ReplayEvent(&decoded, &wire); + capn_free(c); + + if (decoded.body_which == ReplayEvent_body_endSession) { + free_ReplayEvent(&decoded); /* Void body -- no payload, safe/no-op */ + return -1; + } + + sim_advance_virtual_clock(sim, decoded.ts); + + return replay_event_to_core_event(&decoded, out_ev); +} + +/* + * sim_register_default_picker - Req 5 AC 17's "core.h-scoped plugin + * registration path" (task 11.1's own documented "Known scope boundary", + * quoted on sim.h's sched_sim_driver_start() doc comment) -- this task + * (11.2) is where a resolvable picker becomes load-bearing: without one, + * sim_picker_thread_fn()'s sched_resolve_job_picker() call always fails and + * every scheduling round closes with zero decisions (see that function's + * own graceful-degradation handling) -- harmless for a fixture with no + * pending jobs, but never actually exercises picker candidate selection for + * one that has any. + * + * Defaults sim->core.job_pick_plugin to "builtin_fcfs" when the + * ReplayHeader carried no schedPolicy (mirrors schedobj.c's + * sched_set_default_job_pick_plugin(), normal mode's own default), then + * registers the real builtin_fcfs plugin -- via builtin_fcfs_new_plugin() + + * plugin_register_core() (plugin.h), NOT plugin_register() (no + * xce_scheduler_t* exists here, Req 5 AC 17) -- ONLY when the configured + * policy is genuinely "builtin_fcfs". A replay file recorded under any + * OTHER picker name is left unresolvable here: this task builds no generic + * "load an arbitrary named plugin for simulation" mechanism (a materially + * bigger, separate problem) -- sched_resolve_job_picker() then legitimately + * fails for it, and the picker thread degrades gracefully, the same + * documented behavior xce_sched_core_load_snapshot() already establishes + * for an unresolvable picker (sim.h's own doc comment). + */ +static void sim_register_default_picker(xce_sched_sim_driver_t* sim) +{ + xce_sched_plugin_t* plugin; + + if (sim->core.job_pick_plugin == NULL) { + sim->core.job_pick_plugin = makeString("builtin_fcfs"); + } + + if (strcmp(sim->core.job_pick_plugin, "builtin_fcfs") != 0) { + return; + } + + plugin = builtin_fcfs_new_plugin(); + if (plugin == NULL) { + ERROR("%s out of memory building the builtin_fcfs plugin instance", + __func__); + return; + } + + if (plugin_register_core(&(sim->core), plugin) < 0) { + ERROR("%s failed to register builtin_fcfs against the simulation " + "core", + __func__); + free(plugin); + } +} + +/* + * sim_print_decision - Req 5 AC 19's default: "If --sim-output is not + * specified, decisions are printed to stdout only." This task (11.2) owns + * only that unconditional default path; the --sim-output file-write branch + * (sim->record_file) and the --sim-verify comparison branch (sim->verify) + * are sibling tasks 11.5/11.4's own jobs, deliberately not built here (see + * this file's own header comment). + */ +static void sim_print_decision(const xce_decision_t* d) +{ + switch (d->op) { + case XCE_DECISION_DISPATCH: + printf("[sim] DISPATCH jobid=%llu\n", + (unsigned long long)d->u.alloc->jobid); + break; + case XCE_DECISION_PEND: + printf("[sim] PEND jobid=%llu\n", + (unsigned long long)d->u.reason->jobid); + break; + case XCE_DECISION_PREEMPT: + printf("[sim] PREEMPT jobid=%llu\n", + (unsigned long long)d->u.preempt->jobid); + break; + default: + break; + } +} + +/* + * sim_picker_thread_fn - the simulation driver's picker-driving thread + * (Req 5 AC 18's pseudocode: `core_go()` signals it, the driver's own + * decision-drain loop blocks on `cbuf_get(&core->decisions)` until it + * produces the end-of-session NULL marker). Mirrors scheduler.c's + * sched_job_picker() outer wbuf-wait/begin_session/next_job/end_session + * loop, parameterized on xce_sched_core_t* directly instead of + * xce_scheduler_t* -- same core-scoped-thread precedent + * test_core_full_cycle.c's test_picker_thread_fn() already establishes for + * testing, now load-bearing PRODUCTION code (see this file's own header + * comment for why one is unavoidable here). + * + * Known scope boundary (task 11.2): real resource-matching allocation + * (sched_make_allocation(), scheduler.c) requires a live xce_scheduler_t* + * this driver deliberately does not have (Req 5 AC 17) and has no + * pure-core equivalent yet -- that extraction belongs to the separate, + * not-yet-designed scheduler-allocation spec (.kiro/specs/scheduler- + * allocation), not this task. Until it exists, this thread -- acting as + * BOTH picker and allocator in one thread (no separate jbuf hand-off is + * needed: nothing else ever drains jbuf here) -- conservatively PENDs every + * candidate next_job() returns, via the real on_pend()/make_pend_decision()/ + * make_pendreason() production calls, rather than fabricate a DISPATCH + * decision with no real worker/resource assignment behind it. + * + * Runs for the simulation driver's entire lifetime (one wbuf-wait/session + * cycle per xce_core_go(), forever) -- never joined or cancelled: no + * sim-driver teardown/fini function exists yet (task 11.1 did not build + * one either), the same gap sched_job_picker()/sched_allocator() do not + * have in normal mode only because sched_fini() already exists there. + */ +static void* sim_picker_thread_fn(void* arg) +{ + xce_sched_core_t* core = (xce_sched_core_t*)arg; + + for (;;) { + int go = 0; + xce_sched_plugin_t* picker = NULL; + xce_decision_t* decision = NULL; + + cbuf_get(&(core->wbuf), &go); + + if (sched_resolve_job_picker(core, &picker) < 0) { + /* No resolvable picker (sim_register_default_picker()'s own + * doc comment) -- close the session immediately with zero + * decisions rather than leave the driver's drain loop blocked + * forever. */ + cbuf_put(&(core->decisions), &decision); + continue; + } + + picker->picker_iface->begin_session(picker); + + for (;;) { + xce_sched_job_t* job = picker->picker_iface->next_job(picker); + + if (job == NULL) { + break; + } + + picker->picker_iface->on_pend(picker, job->jobid); + decision = make_pend_decision(make_pendreason(job->jobid)); + cbuf_put(&(core->decisions), &decision); + } + + decision = NULL; + cbuf_put(&(core->decisions), &decision); + + picker->picker_iface->end_session(picker); + } + + return NULL; +} + +int sched_sim_driver_run(xce_sched_sim_driver_t* sim, + const sched_sim_args_t* args) +{ + pthread_t picker_tid; + xce_core_event_t* buf_events = NULL; + int buf_cap = 0; + int buf_n = 0; + int rc; + + rc = sched_sim_driver_start(sim, args); + if (rc < 0) { + return rc; + } + + sim_register_default_picker(sim); + + rc = pthread_create(&picker_tid, NULL, sim_picker_thread_fn, &(sim->core)); + if (rc != 0) { + ERROR("%s failed to start the simulation picker thread: %s", __func__, + strerror(rc)); + return -rc; + } + + for (;;) { + struct capn c; + capn_ptr root; + + if (capn_init_fp(&c, sim->replay_file, 0) != 0) { + break; /* end of file (or malformed trailing data, same + * "nonzero means no more messages" convention sim.c's + * own snapshot scan already follows) */ + } + + if (msg_root_has_shape(&c, &root, ReplayEvent_word_count, + ReplayEvent_pointer_count)) { + xce_core_event_t ev; + int translated + = sim_process_replay_event_message(sim, &c, root, &ev); + + if (translated < 0) { + break; /* Req 5 AC 20 end-of-stream marker */ + } + + if (translated > 0) { + if (buf_n == buf_cap) { + int newcap = buf_cap ? (buf_cap * 2) : 16; + xce_core_event_t* tmp = (xce_core_event_t*)realloc( + buf_events, (size_t)newcap * sizeof(*tmp)); + + if (tmp == NULL) { + ERROR("%s out of memory growing the accumulated " + "event batch (buf_n=%d)", + __func__, buf_n); + free(buf_events); + return -ENOMEM; + } + + buf_events = tmp; + buf_cap = newcap; + } + + buf_events[buf_n++] = ev; + } + + continue; + } + + if (msg_root_has_shape(&c, &root, ReplaySession_word_count, + ReplaySession_pointer_count)) { + ReplaySession_ptr sp; + uint64_t begin_round; + xce_decision_t* d; + int push_rc; + + sp.p = root; + begin_round = ReplaySession_get_beginRound(sp); + capn_free(&c); + + /* Req 5 AC 18 pseudocode: push the accumulated batch first, + * THEN (mirroring requirements.md's own "beginRound = + * xce_core_now(core) on core_go()" prose) jump the clock to + * this session's own begin_round right before starting it. */ + push_rc = xce_core_push_events(&(sim->core), buf_events, buf_n); + if (push_rc < 0) { + ERROR("%s xce_core_push_events failed rc=%d (buf_n=%d)", + __func__, push_rc, buf_n); + } + buf_n = 0; + + sim_advance_virtual_clock(sim, begin_round); + + if (xce_core_go(&(sim->core)) < 0) { + ERROR("%s xce_core_go failed -- skipping this round's " + "decision drain", + __func__); + continue; + } + + while ((cbuf_get(&(sim->core.decisions), &d) == 0) && (d != NULL)) { + sim_print_decision(d); + } + + xce_core_end_session(&(sim->core)); + continue; + } + + if (msg_root_has_shape(&c, &root, SnapshotEntry_word_count, + SnapshotEntry_pointer_count)) { + /* Mid-replay periodic checkpoints -- skipped (task 11.1's own + * doc comment on sched_sim_driver_start(), and this task's own + * text): only the INITIAL SnapshotEntry (already consumed at + * startup) matters. */ + capn_free(&c); + continue; + } + + ERROR("%s unrecognized message shape in replay stream -- stopping", + __func__); + capn_free(&c); + break; + } + + free(buf_events); + + return 0; +} diff --git a/src/scheduler/testsuite/meson.build b/src/scheduler/testsuite/meson.build index 8583fd7d..1ef213f4 100644 --- a/src/scheduler/testsuite/meson.build +++ b/src/scheduler/testsuite/meson.build @@ -479,6 +479,28 @@ sched_sim_startup_test = executable( dependencies: [scheduler_test_deps], install: false) +# Contract test for the simulation-mode event loop and virtual clock +# advancement (scheduler-core spec, task 11.2, Req 5 AC 14/18): the virtual +# clock jumps to each ReplayEvent's own timestamp, in exact file order +# (including a same-timestamp pair), at the moment it is queued; each +# ReplaySession boundary triggers one complete push-events -> +# start-session -> drain-decisions -> end-session round, in file order +# across multiple rounds. Same white-box #include pattern as +# sched_sim_startup_test (core.c, core_clock.c, recording.c, sim.c, +# snapshot.c/snapshot_wire.c/snapshot_load.c) PLUS core_events.c, +# core_session.c, decision.c (the real session/decision machinery this +# task's event loop drives) PLUS plugin.c, plugins/builtin_fcfs.c (task +# 11.2's own new core.h-scoped picker registration path, +# plugin_register_core()/builtin_fcfs_new_plugin() -- deliberately NOT +# schedobj.c, which needs a genuine xce_scheduler_t* this driver has none +# of, Req 5 AC 17) PLUS the new ../src/sim_loop.c under test. +sched_sim_event_loop_test = executable( + 'sched_sim_event_loop_test', + ['test_sim_event_loop.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', @@ -551,3 +573,6 @@ test('scheduler admin request test', test('scheduler sim startup test', sched_sim_startup_test, suite: 'scheduler', env: ['ASAN_OPTIONS=detect_leaks=0']) +test('scheduler sim event loop test', + sched_sim_event_loop_test, suite: 'scheduler', + env: ['ASAN_OPTIONS=detect_leaks=0']) diff --git a/src/scheduler/testsuite/test_sim_event_loop.c b/src/scheduler/testsuite/test_sim_event_loop.c new file mode 100644 index 00000000..b6f8c17f --- /dev/null +++ b/src/scheduler/testsuite/test_sim_event_loop.c @@ -0,0 +1,533 @@ +/* + * test_sim_event_loop.c - contract test for the simulation-mode event loop + * and virtual clock advancement (scheduler-core spec, task 11.2, Req 5 AC + * 14/18). + * + * Observable under test (task 11.2's own text): + * - replaying a fixture file with events at non-uniform timestamps + * processes them in file order with the virtual clock reporting each + * event's exact timestamp at the moment it's pushed. + * + * Two tests, at two different granularities: + * + * Test 1 (unit-level, white-box): calls sim_process_replay_event_message() + * -- the extracted per-message decode+clock-advance+translate step -- + * directly, sequentially, once per hand-built ReplayEvent message, with + * deliberately non-uniform timestamps (including a same-timestamp pair, + * Req 5 AC 14's "same-timestamp events processed in file order" clause), + * asserting a REAL xce_core_now(&sim.core) read matches EXACTLY after + * each call -- this is the precise, sequential, file-order proof: a + * picker-hook spy (the only other observation point xce_core_push_events() + * offers) cannot see this because every event in one accumulated batch is + * pushed through ONE xce_core_push_events() call, after the clock has + * already been walked through all of them. + * + * Test 2 (integration-level): drives the full sched_sim_driver_run() -- + * this task's other Observable, "treat each session-boundary message as + * the trigger for one complete scheduling round" -- against a fixture + * with TWO ReplaySession boundaries at increasing beginRound timestamps, + * asserting the run completes (rc == 0 -- this is the RED/GREEN proof: + * before task 11.2, sched_sim_driver_run() returned -ENOSYS here) and + * that the virtual clock ends at the SECOND session's beginRound -- which + * only happens if both session boundaries were reached, in order. + * + * Fixture strategy: same production-code-reuse technique test_sim_startup.c + * already establishes for the header+snapshot prefix (sched_record_start()/ + * sched_record_stop() against a small real xce_scheduler_t fixture) PLUS + * hand-constructed ReplayEvent/ReplaySession messages appended afterward, + * using the SAME encode_ReplayEvent_ptr()/write_capnp_message() technique + * test_e2e_capnp.c's own test_replay_event_job_state_change_round_trip/ + * test_replay_event_end_session_round_trip already prove works, and + * low-level new_ReplaySession()/ReplaySession_set_beginRound() field + * setters for ReplaySession, mirroring sim.c's own + * write_lone_snapshotentry_message() precedent (test_sim_startup.c) for + * minimal hand-built messages that leave irrelevant fields at their + * zero-value defaults. + * + * White-box #include of ../src/core.c, ../src/core_clock.c, + * ../src/core_events.c, ../src/core_session.c, ../src/decision.c (task + * 11.2's own core-driving calls: xce_core_push_events()/xce_core_go()/ + * xce_core_end_session()/make_pend_decision()) PLUS ../src/plugin.c, + * ../src/plugins/builtin_fcfs.c (the real plugin_register_core()/ + * builtin_fcfs_new_plugin(), task 11.2's own new core.h-scoped picker + * registration path -- NOT schedobj.c/plugin_register(), which need a + * genuine xce_scheduler_t* this test deliberately never builds one of, + * same Req 5 AC 17 boundary sim.c/sim_loop.c themselves respect) PLUS + * ../src/recording.c (fixture-building only, same as test_sim_startup.c) + * PLUS ../src/sim.c/../src/sim_loop.c (the code under test) PLUS + * ../src/snapshot.c/snapshot_wire.c/snapshot_load.c (task 8.1, needed by + * both the fixture-building and sim.c's own snapshot load). + */ +#include +#include +#include +#include +#include +#include +#include + +#include + +#include "controller.capnp.h" +#include "kvmgr.h" +#include "scheduler.h" +#include "xstring.h" + +/* Pull in the real functions under test into this TU (white-box #include, + * same convention as test_sim_startup.c/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/plugin.c" +#include "../src/plugins/builtin_fcfs.c" +#include "../src/recording.c" +#include "../src/sim.c" +#include "../src/sim_loop.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; } + +/* core_reset_workers()'s (core_session.c) linker dependency -- see + * test_core_full_cycle.c's own file header comment for the same stub; + * never invoked (no worker is ever added here). */ +void sched_worker_release_host_ctx(xce_sched_worker_t* w) { (void)w; } + +/* plugin_prepare_allocation()'s (plugin.c, the "limit" plugin category) + * linker dependency -- never invoked (this file never drives real + * allocation, Req 5 AC 17 / this file's own header comment). */ +int sched_fill_limit_requests(xce_scheduler_t* sched, xvec_idval_t* vec, + xlist_t* limits) +{ + (void)sched; + (void)vec; + (void)limits; + return 0; +} + +/* Real reimplementation of schedobj.c's sched_resolve_job_picker -- same + * rationale/algorithm as test_sim_startup.c's/test_core_full_cycle.c's own + * copies (schedobj.c is not linked: it would need plugin_register(), which + * requires a genuine xce_scheduler_t* neither this file's fixtures nor + * sim_loop.c's own core-scoped registration path have, Req 5 AC 17). */ +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; } + +/* ---------------------------------------------------------------------- */ +/* Fixture helpers */ +/* ---------------------------------------------------------------------- */ + +static void make_tmp_dir(char* out, size_t out_len, const char* suffix) +{ + char tmpl[PATH_MAX]; + + snprintf(tmpl, sizeof(tmpl), "/tmp/sched_simloop%s_XXXXXX", suffix); + TEST_ASSERT_NOT_NULL(mkdtemp(tmpl)); + TEST_ASSERT_TRUE(strlen(tmpl) < out_len); + strcpy(out, tmpl); +} + +static int stub_picker_begin_session(void* picker) +{ + (void)picker; + return 0; +} + +static xce_sched_job_t* stub_picker_next_job(void* picker) +{ + (void)picker; + return NULL; +} + +static int stub_picker_end_session(void* picker) +{ + (void)picker; + return 0; +} + +static int stub_picker_jobid_noop(void* picker, uint64_t jobid) +{ + (void)picker; + (void)jobid; + return 0; +} + +static int stub_picker_queue_update_noop(void* picker, xce_sched_queue_t* q) +{ + (void)picker; + (void)q; + return 0; +} + +static int stub_picker_job_noop(void* picker, xce_sched_job_t* job) +{ + (void)picker; + (void)job; + return 0; +} + +static int stub_picker_job_state_change_noop(void* picker, xce_sched_job_t* job, + uint32_t state) +{ + (void)picker; + (void)job; + (void)state; + return 0; +} + +static xce_job_picker_iface_t g_stub_picker_iface + = { .init = NULL, + .fini = NULL, + .begin_session = stub_picker_begin_session, + .next_job = stub_picker_next_job, + .end_session = stub_picker_end_session, + .on_dispatch = stub_picker_jobid_noop, + .on_pend = stub_picker_jobid_noop, + .on_queue_update = stub_picker_queue_update_noop, + .on_job_push = stub_picker_job_noop, + .on_job_state_change = stub_picker_job_state_change_noop, + .on_job_update = stub_picker_job_noop, + .on_config = NULL, + .on_job_rusage = NULL }; + +static xce_sched_plugin_t g_stub_picker_plugin + = { .name = "test_sim_event_loop_picker", + .picker_iface = &g_stub_picker_iface }; + +/* Builds a small, real xce_scheduler_t fixture purely to drive + * sched_record_start()/sched_record_stop() and produce a REAL, valid + * `.replay` header+snapshot prefix this test's OWN code under test then + * consumes (same pattern test_sim_startup.c's build_fixture_scheduler() + * establishes). Deliberately registers a picker OTHER than "builtin_fcfs" + * (name never matches) -- neither test in this file pushes a real + * candidate job, so which picker resolves at replay time never matters, + * and this keeps the fixture's own recorded schedPolicy from accidentally + * satisfying sim_register_default_picker()'s exact-name check. */ +static void build_fixture_scheduler(xce_scheduler_t* sched, char* record_dir) +{ + int rc; + + memset(sched, 0, sizeof(*sched)); + mylock_init(&(sched->record_lock)); + + rc = xce_sched_core_init(&(sched->core)); + TEST_ASSERT_EQUAL_INT(0, rc); + + lock(sched->core.plugins, RTTI_LOCK_WRITE); + rc = key_add(sched->core.plugins, (void*)g_stub_picker_plugin.name, + (void*)&g_stub_picker_plugin); + TEST_ASSERT_GREATER_OR_EQUAL(0, rc); + unlock(sched->core.plugins); + + sched->core.job_pick_plugin = g_stub_picker_plugin.name; + sched->record_dir = record_dir; +} + +/* Produces a valid `.replay` header+snapshot prefix at `out_path` via the + * real sched_record_start() -- identical technique test_sim_startup.c's own + * write_valid_fixture() establishes for the header+snapshot bytes + * themselves -- but deliberately does NOT call sched_record_stop(): that + * function unconditionally appends its own ReplayEvent{endSession} marker + * (recording.c, Req 5 AC 20) immediately after the snapshot, which would + * terminate sched_sim_driver_run()'s read loop the instant it reached that + * marker -- before ever seeing anything this test's own callers append + * afterward. Returns the STILL-OPEN FILE* (sched_record_start()'s own + * `sched.record_file`, positioned right after the snapshot) so the caller + * can append its own hand-built events/sessions directly onto the same + * stream, ending with its own append_end_of_stream_marker() call, then + * fclose() it directly -- bypassing sched_record_stop()'s real + * event/session-recording API entirely, which this test never exercises. */ +static FILE* write_header_and_snapshot(char* out_path, size_t out_path_len, + const char* suffix) +{ + char record_dir[PATH_MAX]; + xce_scheduler_t sched; + int rc; + + make_tmp_dir(record_dir, sizeof(record_dir), suffix); + build_fixture_scheduler(&sched, record_dir); + + rc = sched_record_start(&sched, "fixture.replay"); + TEST_ASSERT_EQUAL_INT(0, rc); + TEST_ASSERT_TRUE(strlen(sched.record_file_path) < out_path_len); + strcpy(out_path, sched.record_file_path); + + return sched.record_file; +} + +/* Appends one ReplayEvent{jobStateChange} message to the already-open + * stream `f` -- same encode_ReplayEvent_ptr()/write_capnp_message() + * technique test_e2e_capnp.c's own + * test_replay_event_job_state_change_round_trip already proves works. + * jobStateChange is used throughout this file purely as a cheap, + * already-proven-safe payload shape (a plain value struct, no further + * nested pointers) -- core_events.c's own JOB_STATE_CHANGE handling + * WARN+skips an unknown jobid rather than fail (verified in core_events.c), + * so neither test here needs a real job to exist in the loaded snapshot. + */ +static void append_jobstatechange_event(FILE* f, uint64_t ts, uint64_t jobid) +{ + job_state_change_event_t change; + replay_event_t src; + struct capn c; + struct capn_segment* cs; + ReplayEvent_ptr p; + + memset(&change, 0, sizeof(change)); + change.jobid = jobid; + change.old_state = 1u; + change.new_state = 2u; + change.exit_code = 0; + + memset(&src, 0, sizeof(src)); + src.ts = ts; + src.body_which = ReplayEvent_body_jobStateChange; + src.body.job_state_change = &change; + + capn_init_malloc(&c); + cs = capn_root(&c).seg; + encode_ReplayEvent_ptr(cs, &p, &src); + capn_setp(capn_root(&c), 0, p.p); + + TEST_ASSERT_EQUAL_INT(0, write_capnp_message(f, &c)); + capn_free(&c); +} + +/* Appends the .replay stream's whole-stream termination marker + * (ReplayEvent{endSession}, Req 5 AC 20) -- same technique + * test_e2e_capnp.c's own test_replay_event_end_session_round_trip already + * proves works. */ +static void append_end_of_stream_marker(FILE* f) +{ + replay_event_t src; + struct capn c; + struct capn_segment* cs; + ReplayEvent_ptr p; + + memset(&src, 0, sizeof(src)); + src.body_which = ReplayEvent_body_endSession; + + capn_init_malloc(&c); + cs = capn_root(&c).seg; + encode_ReplayEvent_ptr(cs, &p, &src); + capn_setp(capn_root(&c), 0, p.p); + + TEST_ASSERT_EQUAL_INT(0, write_capnp_message(f, &c)); + capn_free(&c); +} + +/* Appends a minimal ReplaySession message with only beginRound set -- + * decisions/summary are left at new_ReplaySession()'s zero-value defaults + * (empty list / capn_null), mirroring sim.c's own + * write_lone_snapshotentry_message() precedent (test_sim_startup.c) for + * hand-built messages whose irrelevant fields nothing under test reads. + * sched_sim_driver_run() only ever reads beginRound off a ReplaySession + * (via the lightweight ReplaySession_get_beginRound() accessor, task 11.2's + * own scope: the decisions/summary fields are task 11.4/11.5 territory). */ +static void append_replay_session(FILE* f, uint64_t begin_round) +{ + struct capn c; + struct capn_segment* cs; + ReplaySession_ptr sp; + + capn_init_malloc(&c); + cs = capn_root(&c).seg; + sp = new_ReplaySession(cs); + ReplaySession_set_beginRound(sp, begin_round); + capn_setp(capn_root(&c), 0, sp.p); + + TEST_ASSERT_EQUAL_INT(0, write_capnp_message(f, &c)); + capn_free(&c); +} + +/* ---------------------------------------------------------------------- */ +/* Test 1: sim_process_replay_event_message() advances the virtual clock */ +/* to each event's own timestamp, in exact file order, including a */ +/* same-timestamp pair (Req 5 AC 14). */ +/* ---------------------------------------------------------------------- */ + +static void +test_process_replay_event_advances_virtual_clock_to_exact_event_ts_in_file_order( + void) +{ + static const uint64_t ts_values[5] + = { 1000ULL, 1050ULL, 1050ULL, 500000ULL, 9999999999ULL }; + char path[PATH_MAX]; + xce_sched_sim_driver_t sim; + FILE* f; + int i; + int rc; + + /* A minimal sim driver: only sim->core is exercised by + * sim_process_replay_event_message() (via sim_advance_virtual_clock()), + * so xce_sched_core_init() plus a frozen (speed=0) clock is a + * sufficient, self-contained fixture -- no replay file, snapshot, or + * full sched_sim_driver_start() needed for this unit-level test. */ + memset(&sim, 0, sizeof(sim)); + rc = xce_sched_core_init(&(sim.core)); + TEST_ASSERT_EQUAL_INT(0, rc); + + /* speed_num=0 (discrete/frozen -- AC 3's own documented mode table, + * core_clock.c) so xce_core_now() reports EXACTLY virtual_clock_base + * regardless of real wall-clock time elapsed between + * sim_advance_virtual_clock()'s clock_gettime() call and this test's + * own assertion immediately after -- deterministic, never flaky. */ + atomic_store_explicit(&(sim.core.speed_num), 0, memory_order_release); + atomic_store_explicit(&(sim.core.speed_den), 1, memory_order_release); + + make_tmp_dir(path, sizeof(path), "_clockjump"); + strcat(path, "/events.replay"); + + f = fopen(path, "w"); + TEST_ASSERT_NOT_NULL(f); + for (i = 0; i < 5; i++) { + append_jobstatechange_event(f, ts_values[i], 42ULL); + } + fclose(f); + + f = fopen(path, "r"); + TEST_ASSERT_NOT_NULL(f); + + for (i = 0; i < 5; i++) { + struct capn c; + capn_ptr root; + xce_core_event_t ev; + int translated; + + TEST_ASSERT_EQUAL_INT(0, capn_init_fp(&c, f, 0)); + TEST_ASSERT_TRUE(msg_root_has_shape(&c, &root, ReplayEvent_word_count, + ReplayEvent_pointer_count)); + + translated = sim_process_replay_event_message(&sim, &c, root, &ev); + + TEST_ASSERT_EQUAL_INT(1, translated); + TEST_ASSERT_EQUAL_INT(XCE_CORE_EVENT_JOB_STATE_CHANGE, ev.type); + TEST_ASSERT_EQUAL_UINT64(ts_values[i], ev.ts); + + /* The Observable itself: a REAL clock read, immediately after this + * one event was processed, matches that event's exact timestamp -- + * not a trivially-true check, and not observable any other way + * (see this file's own header comment for why a picker-hook spy + * cannot see per-event granularity). */ + TEST_ASSERT_EQUAL_UINT64(ts_values[i], xce_core_now(&(sim.core))); + } + + fclose(f); + + printf("test_process_replay_event_advances_virtual_clock_to_exact_event_" + "ts_in_file_order: PASS\n"); +} + +/* ---------------------------------------------------------------------- */ +/* Test 2: sched_sim_driver_run() treats each ReplaySession boundary as */ +/* the trigger for one complete scheduling round, in file order, across */ +/* multiple rounds -- this task's other Observable, and the RED->GREEN */ +/* proof (sched_sim_driver_run() returned -ENOSYS here before task 11.2). */ +/* ---------------------------------------------------------------------- */ + +static void +test_run_processes_multiple_sessions_in_order_and_no_longer_returns_enosys(void) +{ + char path[PATH_MAX]; + FILE* f; + sched_sim_args_t args; + xce_sched_sim_driver_t sim; + int rc; + + f = write_header_and_snapshot(path, sizeof(path), "_run"); + TEST_ASSERT_NOT_NULL(f); + + /* Non-uniform timestamps throughout, two ReplaySession boundaries at + * strictly increasing beginRound values -- the final virtual-clock + * assertion below only holds if BOTH boundaries were reached, in + * order (a driver that stalled after the first, or processed them out + * of order, would leave the clock at the wrong value). */ + append_jobstatechange_event(f, 2000ULL, 999ULL); /* unknown jobid -- + * core_events.c + * WARN+skips it, + * verified safe */ + append_jobstatechange_event(f, 2500ULL, 999ULL); + append_replay_session(f, 3000ULL); /* session boundary #1 */ + append_jobstatechange_event(f, 4000ULL, 999ULL); + append_replay_session(f, 5000ULL); /* session boundary #2 */ + append_end_of_stream_marker(f); + + fclose(f); + + memset(&args, 0, sizeof(args)); + args.replay_path = path; + args.sim_start_ts = SCHED_SIM_START_TS_LATEST; + args.sim_speed = 0; /* frozen -- deterministic final-clock assertion */ + + rc = sched_sim_driver_run(&sim, &args); + + /* The RED->GREEN proof: before task 11.2, this returned -ENOSYS + * (test_sim_startup.c's own former test_run_starts_then_fails_enosys_ + * pending_event_loop, removed now that this Observable supersedes it). */ + TEST_ASSERT_EQUAL_INT(0, rc); + + /* Both ReplaySession boundaries were reached, in file order: the clock + * ends at the SECOND (later) beginRound, not the first, and not the + * last ReplayEvent's own ts (2000/2500/4000 all precede 5000). */ + TEST_ASSERT_EQUAL_UINT64(5000ULL, xce_core_now(&(sim.core))); + + printf("test_run_processes_multiple_sessions_in_order_and_no_longer_" + "returns_enosys: PASS\n"); +} + +int main(void) +{ + UNITY_BEGIN(); + + RUN_TEST( + test_process_replay_event_advances_virtual_clock_to_exact_event_ts_in_file_order); + RUN_TEST( + test_run_processes_multiple_sessions_in_order_and_no_longer_returns_enosys); + + return UNITY_END(); +} diff --git a/src/scheduler/testsuite/test_sim_startup.c b/src/scheduler/testsuite/test_sim_startup.c index efa3b206..cd58a9e7 100644 --- a/src/scheduler/testsuite/test_sim_startup.c +++ b/src/scheduler/testsuite/test_sim_startup.c @@ -482,36 +482,14 @@ static void test_null_arguments_return_einval(void) printf("test_null_arguments_return_einval: PASS\n"); } -/* ---------------------------------------------------------------------- */ -/* Test 6: sched_sim_driver_run() -- design.md's literal entry point -- */ -/* calls start() then fails loudly with -ENOSYS (task 11.2's event loop */ -/* does not exist yet), never silently returning success. */ -/* ---------------------------------------------------------------------- */ - -static void test_run_starts_then_fails_enosys_pending_event_loop(void) -{ - char path[PATH_MAX]; - uint64_t snapshot_ts = 0; - sched_sim_args_t args; - xce_sched_sim_driver_t sim; - int rc; - - write_valid_fixture(path, sizeof(path), &snapshot_ts, "_run"); - - memset(&args, 0, sizeof(args)); - args.replay_path = path; - args.sim_start_ts = SCHED_SIM_START_TS_LATEST; - args.sim_speed = 0; - - rc = sched_sim_driver_run(&sim, &args); - TEST_ASSERT_EQUAL_INT(-ENOSYS, rc); - - /* Startup itself still genuinely succeeded -- state is real and - * inspectable, matching sched_sim_driver_start()'s own contract. */ - TEST_ASSERT_EQUAL_UINT64(snapshot_ts, xce_core_now(&(sim.core))); - - printf("test_run_starts_then_fails_enosys_pending_event_loop: PASS\n"); -} +/* Test 6 (sched_sim_driver_run() -- design.md's literal entry point -- + * calling start() then failing loudly with -ENOSYS pending the event loop) + * was removed here: the event loop now exists (scheduler-core spec, task + * 11.2), and sched_sim_driver_run() itself moved to sim_loop.c (a much + * heavier white-box dependency set this file's own header comment + * deliberately stays without -- "this task stops at 'ready-to-run'"). Its + * successor lives in test_sim_event_loop.c, alongside every other + * sched_sim_driver_run()-level test. */ int main(void) { @@ -522,7 +500,6 @@ int main(void) RUN_TEST(test_first_message_not_replayheader_fails_cleanly); RUN_TEST(test_missing_replay_file_fails_cleanly); RUN_TEST(test_null_arguments_return_einval); - RUN_TEST(test_run_starts_then_fails_enosys_pending_event_loop); return UNITY_END(); }