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

feat(scheduler-core): implement what-if branching and simulation output recording (task 11.5)

Browse source 
Add sim_record.c (scheduler_t-free .replay writer: ReplayHeader + loaded
SnapshotEntry once, then per-event pass-through and per-session
ReplaySession finalization) and wire --sim-alt-events event-source
selection into sim_loop.c's drain loop, coexisting with task 11.2's
stdout default and task 11.4's --sim-verify. Extract scheduler_t-free
halves out of recording_session.c's begin/end so sim-side code can reuse
the decision-encoding path without gaining an xce_scheduler_t* dependency,
preserving Req 5 AC 17 isolation. Per Req 5 AC 16/19.
This commit is contained in:
Rongsong Shen 2026-08-07 02:42:50 +00:00
parent a6083d803e
commit 564408e967
Signed by: rshen
SSH key fingerprint: SHA256:8Vf2pZ3h6rJa7EhGK+WaAQ9Mta/apZvteW6g9U/WKKk
14 changed files with 1750 additions and 91 deletions
Download patch file Download diff file Expand all files Collapse all files

38
src/scheduler/include/recording.h
View file

@ -273,6 +273,22 @@ typedef struct sched_record_session_ctx_s sched_record_session_ctx_t;
extern sched_record_session_ctx_t*
sched_record_session_begin(xce_scheduler_t* sched, uint64_t begin_round_ts);
/*
* sched_record_session_ctx_new - scheduler_t-free half of
* sched_record_session_begin() above (scheduler-core spec, task 11.5, Req 5
* AC 19): allocates and stamps a fresh sched_record_session_ctx_t with
* `begin_round_ts`, with no `sched->recording`/`sched->record_file` gate at
* all. sched_record_session_begin() is a thin wrapper around this function
* (gate check, then this call) -- driver.c keeps calling
* sched_record_session_begin() unchanged; this entry point exists so a
* caller with NO xce_scheduler_t* (the simulation driver, sim_record.c,
* Req 5 AC 17) can still build the identical accumulator, gating on
* whatever it has instead (sim->record_file != NULL). Returns NULL only on
* allocation failure.
*/
extern sched_record_session_ctx_t*
sched_record_session_ctx_new(uint64_t begin_round_ts);
/*
* sched_record_session_add_decision - appends one drained decision to
* `ctx`'s in-progress ReplaySession.decisions list, tagged with `ts`. A
@ -303,6 +319,28 @@ extern int sched_record_session_end(xce_scheduler_t* sched,
sched_record_session_ctx_t* ctx,
uint64_t end_ts);
/*
* sched_record_session_finalize - scheduler_t-free half of
* sched_record_session_end() above (scheduler-core spec, task 11.5, Req 5
* AC 19): builds the ReplaySession wire message from `ctx`'s accumulated
* decisions and `end_ts`, writes it to `f` directly, then frees `ctx` and
* everything it owns. sched_record_session_end() is a thin wrapper around
* this function (its own gate check plus the same
* sched_record_abort_on_error() call on failure); this entry point exists
* so a caller with no xce_scheduler_t* (the simulation driver,
* sim_record.c, Req 5 AC 17) can finalize+write its own accumulated
* session against its own FILE* (sim->record_file) without that gate.
* Returns 0 on success, a negative errno from write_capnp_message() on
* failure -- same contract sched_record_session_end() itself documents.
* `ctx` must be non-NULL (unlike sched_record_session_end(), this function
* has no NULL-ctx no-op case -- the caller already knows it has a live ctx
* by construction, since it only ever calls this after a successful
* sched_record_session_ctx_new()).
*/
extern int sched_record_session_finalize(FILE* f,
sched_record_session_ctx_t* ctx,
uint64_t end_ts);
/*
* sched_record_status - Req 5 AC 10 (scheduler-core spec, task 9.4,
* "Reply field semantics by action" -- the `recordStatus` column). A pure

14
src/scheduler/include/sim.h
View file

@ -52,6 +52,19 @@
* it; this task only stores it)
* bool interactive; -- --interactive/-i (task 12 acts on
* it; this task only stores it)
*
* `snapshot_ts` below is ONE field beyond design.md's literal 8-field
* struct sketch (scheduler-core spec, task 11.5, Req 5 AC 19) -- added, not
* merely implied, because writing a `--sim-output` file's initial
* SnapshotEntry needs the EXACT timestamp Step 2's snapshot scan
* (sim_locate_and_load_snapshot(), sim.c) chose, and re-deriving it via
* xce_core_now(&sim->core) from inside sim_loop.c's event loop is NOT
* equivalent for any --sim-speed other than frozen (0): the virtual clock
* keeps advancing with real wall-clock time between
* sched_sim_driver_start() returning and sim_loop.c's first opportunity to
* write the output header/snapshot, so a re-derived value would drift from
* "the same SnapshotEntry that was loaded" (AC 19's own wording -- same ts,
* not merely a same-shaped one taken moments later).
*/
typedef struct {
xce_sched_core_t core;
@ -62,6 +75,7 @@ typedef struct {
uint64_t sim_speed_den;
bool verify;
bool interactive;
uint64_t snapshot_ts;
} xce_sched_sim_driver_t;
/*

14
src/scheduler/meson.build
View file

@ -17,6 +17,7 @@ scheduler_srcs = [
'src/sim.c',
'src/sim_loop.c',
'src/sim_verify.c',
'src/sim_record.c',
'src/schedobj.c',
'src/plugin.c',
'src/resource.c',
@ -64,17 +65,18 @@ sched_core_isolation_lib = static_library(
# Task 11.3 (scheduler-core spec, Req 5 AC 17): isolated compile target for
# the simulation driver's own files, consumed by
# testsuite/test_sim_isolation.c to prove at build time that sim.c/sim_loop.c
# reference zero controller-facing symbols (client/netio/xceapi/delay/idgen).
# Both files are compiled together into ONE archive -- task 11.2 added
# sim_loop.c as a sibling file to task 11.1's sim.c, and covering only one of
# them would repeat core.c's own siblings' (core_events.c/core_session.c/
# testsuite/test_sim_isolation.c to prove at build time that
# sim.c/sim_loop.c/sim_record.c reference zero controller-facing symbols
# (client/netio/xceapi/delay/idgen). All files are compiled together into ONE
# archive -- task 11.2 added sim_loop.c as a sibling file to task 11.1's
# sim.c, and task 11.5 adds sim_record.c as a third; covering only a subset
# of them would repeat core.c's own siblings' (core_events.c/core_session.c/
# core_clock.c) never-mechanically-covered gap for the sim driver. Same
# "compiling to an object is enough, no real linking needed" rationale as
# sched_core_isolation_lib above.
sched_sim_isolation_lib = static_library(
'sched_sim_isolation',
['src/sim.c', 'src/sim_loop.c'],
['src/sim.c', 'src/sim_loop.c', 'src/sim_record.c'],
include_directories: incdir,
dependencies: scheduler_deps,
install: false)

106
src/scheduler/src/recording_session.c
View file

@ -42,16 +42,26 @@ struct sched_record_session_ctx_s {
uint32_t n_pended;
};
sched_record_session_ctx_t* sched_record_session_begin(xce_scheduler_t* sched,
uint64_t begin_round_ts)
/*
* sched_record_session_ctx_new - allocates and stamps a fresh
* sched_record_session_ctx_t with `begin_round_ts`, with NO scheduler_t
* gate at all (scheduler-core spec, task 11.5, Req 5 AC 19). Factored out
* of sched_record_session_begin() below so sim_record.c's own sim-side
* ReplaySession accumulator (the --sim-output writer) can reuse the
* identical allocation shape without needing an xce_scheduler_t* to pass
* sched_record_session_begin()'s own sched->recording/sched->record_file
* gate -- a gate that has no meaning for the simulation driver, which has
* no xce_scheduler_t* at all (Req 5 AC 17). sched_record_session_begin()
* itself is now a thin wrapper: gate check, then this function -- net
* behavior unchanged for every existing (normal-mode) caller. Returns NULL
* only on allocation failure, same as before this split.
*/
sched_record_session_ctx_t*
sched_record_session_ctx_new(uint64_t begin_round_ts)
{
sched_record_session_ctx_t* ctx;
sched_record_session_ctx_t* ctx
= (sched_record_session_ctx_t*)calloc(1, sizeof(*ctx));
if ((sched == NULL) || !sched->recording || (sched->record_file == NULL)) {
return NULL;
}
ctx = (sched_record_session_ctx_t*)calloc(1, sizeof(*ctx));
if (ctx == NULL) {
return NULL;
}
@ -61,6 +71,16 @@ sched_record_session_ctx_t* sched_record_session_begin(xce_scheduler_t* sched,
return ctx;
}
sched_record_session_ctx_t* sched_record_session_begin(xce_scheduler_t* sched,
uint64_t begin_round_ts)
{
if ((sched == NULL) || !sched->recording || (sched->record_file == NULL)) {
return NULL;
}
return sched_record_session_ctx_new(begin_round_ts);
}
/*
* free_record_decision - frees one heap-allocated replay_decision_t built
* by sched_record_session_add_decision() below, including the small
@ -184,8 +204,28 @@ void sched_record_session_add_decision(sched_record_session_ctx_t* ctx,
ctx->decisions[ctx->n++] = rd;
}
int sched_record_session_end(xce_scheduler_t* sched,
sched_record_session_ctx_t* ctx, uint64_t end_ts)
/*
* sched_record_session_finalize - builds the ReplaySession wire message
* from `ctx`'s accumulated decisions and `end_ts`, writes it to `f`, then
* frees `ctx` and everything it owns (scheduler-core spec, task 11.5, Req 5
* AC 19). Factored out of sched_record_session_end() below so
* sim_record.c's own sim-side accumulator (the --sim-output writer) can
* reuse the identical encode+write+free shape against ITS OWN FILE*
* (sim->record_file) without needing an xce_scheduler_t* to pass
* sched_record_session_end()'s own gate or reach sched->record_file --
* `ctx` tracks nothing about "which scheduler this belongs to" (this
* file's own sched_record_session_ctx_s struct, above, has no such field),
* so nothing here loses fidelity by dropping the scheduler_t parameter.
* sched_record_session_end() itself is now a thin wrapper: its own gate
* check (with the same inline free-without-write path as before this
* split, unchanged), then this function plus the same
* sched_record_abort_on_error() call on failure -- net behavior unchanged
* for every existing (normal-mode) caller. Returns 0 on success, a
* negative errno from write_capnp_message() on failure -- same contract
* sched_record_session_end() itself already documents (recording.h).
*/
int sched_record_session_finalize(FILE* f, sched_record_session_ctx_t* ctx,
uint64_t end_ts)
{
struct capn c;
struct capn_segment* cs;
@ -195,19 +235,6 @@ int sched_record_session_end(xce_scheduler_t* sched,
int rc;
int i;
if (ctx == NULL) {
return 0;
}
if ((sched == NULL) || !sched->recording || (sched->record_file == NULL)) {
for (i = 0; i < ctx->n; i++) {
free_record_decision(ctx->decisions[i]);
}
free(ctx->decisions);
free(ctx);
return 0;
}
memset(&summary, 0, sizeof(summary));
summary.duration_ns
= (end_ts >= ctx->begin_round) ? (end_ts - ctx->begin_round) : 0;
@ -227,13 +254,9 @@ int sched_record_session_end(xce_scheduler_t* sched,
encode_ReplaySession_ptr(cs, &ptr, &wire);
capn_setp(capn_root(&c), 0, ptr.p);
rc = write_capnp_message(sched->record_file, &c);
rc = write_capnp_message(f, &c);
capn_free(&c);
if (rc < 0) {
sched_record_abort_on_error(sched, rc);
}
for (i = 0; i < ctx->n; i++) {
free_record_decision(ctx->decisions[i]);
}
@ -242,3 +265,30 @@ int sched_record_session_end(xce_scheduler_t* sched,
return rc;
}
int sched_record_session_end(xce_scheduler_t* sched,
sched_record_session_ctx_t* ctx, uint64_t end_ts)
{
int rc;
int i;
if (ctx == NULL) {
return 0;
}
if ((sched == NULL) || !sched->recording || (sched->record_file == NULL)) {
for (i = 0; i < ctx->n; i++) {
free_record_decision(ctx->decisions[i]);
}
free(ctx->decisions);
free(ctx);
return 0;
}
rc = sched_record_session_finalize(sched->record_file, ctx, end_ts);
if (rc < 0) {
sched_record_abort_on_error(sched, rc);
}
return rc;
}

6
src/scheduler/src/sim.c
View file

@ -393,6 +393,12 @@ int sched_sim_driver_start(xce_sched_sim_driver_t* sim,
sim_set_virtual_clock(sim, snapshot_ts, args->sim_speed);
/* Stashed verbatim (task 11.5, Req 5 AC 19) -- see sim.h's own doc
* comment on this field for why sim_loop.c's later --sim-output header/
* snapshot write must use THIS captured value, not a fresh
* xce_core_now(&sim->core) read taken after this function returns. */
sim->snapshot_ts = snapshot_ts;
sim->verify = args->verify;
sim->interactive = args->interactive;

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

@ -19,16 +19,20 @@
* (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 task 11.5 (depends on this task): task 11.5
* owns writing decisions to a --sim-output file and loading
* --sim-alt-events. That branch is not 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"). Task 11.4 (this
* revision) owns comparing each drained decision against the recorded
* Task 11.5 (Req 5 AC 16/19, this revision) adds the --sim-alt-events event
* source switch (`event_src` in sched_sim_driver_run(), chosen once before
* the loop) and the --sim-output writer: header+snapshot (once, right
* after startup -- sim_record.c's sim_record_write_header_and_snapshot()),
* per-event pass-through (sim_record_write_event(), same file), and
* per-round ReplaySession accumulation (recording_session.c's
* scheduler_t-free sched_record_session_ctx_new()/_add_decision()/
* _finalize(), externed directly -- no wrapper needed). sim_print_decision()
* remains this file's own unconditional Req 5 AC 19 default; it, --sim-output
* recording, and --sim-verify comparison are all independently gated per
* drained decision (AC 19 does not make them mutually exclusive). Task 11.4
* owns comparing each drained decision against the recorded
* ReplaySession.decisions when --sim-verify is set -- see the ReplaySession
* branch in sched_sim_driver_run() below, and sim_verify.c (sibling file,
* same task) for the sim_decision_eq()/sim_log_mismatch() comparison
* logic itself.
* branch below and sim_verify.c (sibling file) for that comparison logic.
*
* Picker-driving thread (sim_picker_thread_fn, below): Req 5 AC 18's
* pseudocode calls `core_go()` then blocks draining `core->decisions` until
@ -75,6 +79,34 @@ extern void sim_log_mismatch(uint64_t begin_round, int decision_index,
const replay_decision_t* expected,
const xce_decision_t* actual);
/*
* Shared with sim_record.c (task 11.5, Req 5 AC 19) -- same "extern
* declared where consumed" idiom as above. recording_session.c's
* scheduler_t-free session-accumulator entry points are externed directly
* (no wrapper in sim_record.c needed -- no extra wire-encoding logic beyond
* what they already provide). sched_record_session_ctx_t is the SAME
* opaque type recording.h forward-declares for driver.c (real struct body
* in recording_session.c: begin_round/decisions/n/cap/n_dispatched/
* n_pended, no scheduler_t field) -- redeclared here rather than via
* #include "recording.h" (which transitively includes scheduler.h,
* forbidden in this file family, Req 5 AC 17), purely as an opaque pointer
* target never dereferenced directly here.
*/
extern int sim_record_write_header_and_snapshot(FILE* f, xce_sched_core_t* core,
uint64_t snapshot_ts);
extern int sim_record_write_event(FILE* f, replay_event_t* decoded);
typedef struct sched_record_session_ctx_s sched_record_session_ctx_t;
extern sched_record_session_ctx_t*
sched_record_session_ctx_new(uint64_t begin_round_ts);
extern void sched_record_session_add_decision(sched_record_session_ctx_t* ctx,
uint64_t ts,
xce_decision_t* decision);
extern int sched_record_session_finalize(FILE* f,
sched_record_session_ctx_t* ctx,
uint64_t end_ts);
/*
* 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
@ -273,21 +305,46 @@ static int replay_event_to_core_event(const replay_event_t* src,
* 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).
*
* `*out_record_rc` (task 11.5, Req 5 AC 19) is 0 unless the --sim-output
* pass-through write fails, in which case it holds the write's own
* negative errno and this function returns immediately (the primary return
* value is then meaningless -- the caller must check `*out_record_rc`
* FIRST). A separate out-parameter, not folded into the 1/0/-1 return
* contract: an orthogonal failure axis, and a negative errno can
* numerically equal -1 (e.g. -EPERM), colliding with the end-of-stream
* sentinel. Every decoded ReplayEvent is passed through when
* `sim->record_file != NULL` -- including the endSession terminator itself
* (the write runs BEFORE the endSession check below) -- so the output
* stream ends up self-terminated exactly like a normal-mode recording.
*/
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)
xce_core_event_t* out_ev,
int* out_record_rc)
{
ReplayEvent_ptr ep;
struct ReplayEvent wire;
replay_event_t decoded;
*out_record_rc = 0;
ep.p = root;
read_ReplayEvent(&wire, ep);
memset(&decoded, 0, sizeof(decoded));
decode_ReplayEvent(&decoded, &wire);
capn_free(c);
if (sim->record_file != NULL) {
int record_rc = sim_record_write_event(sim->record_file, &decoded);
if (record_rc < 0) {
*out_record_rc = record_rc;
free_ReplayEvent(&decoded);
return 0;
}
}
if (decoded.body_which == ReplayEvent_body_endSession) {
free_ReplayEvent(&decoded); /* Void body -- no payload, safe/no-op */
return -1;
@ -467,6 +524,24 @@ int sched_sim_driver_run(xce_sched_sim_driver_t* sim,
sim_register_default_picker(sim);
/* Req 5 AC 19 (task 11.5): the first two --sim-output messages, written
* exactly once, before the event loop below runs -- see sim_record.c's
* sim_record_write_header_and_snapshot() for the full rationale. A
* write failure aborts the whole run (this file's own header comment
* states the fail-loud --sim-output policy). */
if (sim->record_file != NULL) {
rc = sim_record_write_header_and_snapshot(
sim->record_file, &(sim->core), sim->snapshot_ts);
if (rc < 0) {
ERROR("%s failed to write --sim-output header/snapshot: rc=%d "
"-- aborting simulation run",
__func__, rc);
fclose(sim->record_file);
sim->record_file = NULL;
return rc;
}
}
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__,
@ -474,11 +549,21 @@ int sched_sim_driver_run(xce_sched_sim_driver_t* sim,
return -rc;
}
/* Req 5 AC 16 (task 11.5, "What-if branching"): once --sim-alt-events
* is set, the ONGOING event/session stream comes from
* sim->alt_events_file instead of sim->replay_file -- the initial
* snapshot (already loaded above, from replay_file, by
* sched_sim_driver_start()) is untouched by this switch. Chosen once,
* outside the loop: both files are opened once at startup and never
* swapped mid-run. */
FILE* event_src = (sim->alt_events_file != NULL) ? sim->alt_events_file
: sim->replay_file;
for (;;) {
struct capn c;
capn_ptr root;
if (capn_init_fp(&c, sim->replay_file, 0) != 0) {
if (capn_init_fp(&c, event_src, 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) */
@ -487,8 +572,21 @@ int sched_sim_driver_run(xce_sched_sim_driver_t* sim,
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);
int record_rc = 0;
int translated = sim_process_replay_event_message(sim, &c, root,
&ev, &record_rc);
if (record_rc < 0) {
ERROR("%s failed to write --sim-output pass-through event: "
"rc=%d -- aborting simulation run",
__func__, record_rc);
if (sim->record_file != NULL) {
fclose(sim->record_file);
sim->record_file = NULL;
}
free(buf_events);
return record_rc;
}
if (translated < 0) {
break; /* Req 5 AC 20 end-of-stream marker */
@ -520,12 +618,13 @@ int sched_sim_driver_run(xce_sched_sim_driver_t* sim,
if (msg_root_has_shape(&c, &root, ReplaySession_word_count,
ReplaySession_pointer_count)) {
ReplaySession_ptr sp;
uint64_t begin_round;
replay_session_t* decoded_session = NULL;
xce_decision_t* d;
int push_rc;
int di;
ReplaySession_ptr sp;
uint64_t begin_round;
replay_session_t* decoded_session = NULL;
sched_record_session_ctx_t* record_ctx = NULL;
xce_decision_t* d;
int push_rc;
int di;
sp.p = root;
begin_round = ReplaySession_get_beginRound(sp);
@ -566,11 +665,26 @@ int sched_sim_driver_run(xce_sched_sim_driver_t* sim,
sim_advance_virtual_clock(sim, begin_round);
/* Req 5 AC 19 (task 11.5): mirrors driver.c's
* xce_sched_driver_run_normal_mode_round() placement -- "On
* core_go(): begin building a ReplaySession message with
* beginRound = xce_core_now(core)" -- gated on
* sim->record_file != NULL, using xce_core_now() (== begin_round
* here, nothing else advances the clock in between). */
if (sim->record_file != NULL) {
record_ctx
= sched_record_session_ctx_new(xce_core_now(&(sim->core)));
}
if (xce_core_go(&(sim->core)) < 0) {
ERROR("%s xce_core_go failed -- skipping this round's "
"decision drain",
__func__);
free_ReplaySession_ptr(&decoded_session);
/* record_ctx (if any) has nothing accumulated yet -- a
* direct free() is safe: no ReplaySession is written for a
* round that never ran. */
free(record_ctx);
continue;
}
@ -596,6 +710,15 @@ int sched_sim_driver_run(xce_sched_sim_driver_t* sim,
}
di++;
}
/* Req 5 AC 19 (task 11.5): "append a ReplayDecision to the
* current ReplaySession.decisions list" -- mirrors
* driver.c's identically-placed call site; `d`'s payload
* pointers are still guaranteed live here (nothing frees
* `d` anywhere in this loop). */
if (record_ctx != NULL) {
sched_record_session_add_decision(
record_ctx, xce_core_now(&(sim->core)), d);
}
sim_print_decision(d);
}
@ -612,6 +735,30 @@ int sched_sim_driver_run(xce_sched_sim_driver_t* sim,
free_ReplaySession_ptr(&decoded_session);
xce_core_end_session(&(sim->core));
/* Req 5 AC 19: "On core_end_session(): finalize the
* ReplaySession ... write the complete message" -- mirrors
* driver.c's identically-placed sched_record_session_end()
* call (after end_session, end_ts = post-end xce_core_now()).
* A write failure is fatal to the whole run (this file's own
* header comment); sched_record_session_finalize() has already
* freed `record_ctx` unconditionally by the time it returns. */
if (record_ctx != NULL) {
int record_rc = sched_record_session_finalize(
sim->record_file, record_ctx, xce_core_now(&(sim->core)));
if (record_rc < 0) {
ERROR("%s failed to write --sim-output ReplaySession at "
"begin_round=%llu: rc=%d -- aborting simulation "
"run",
__func__, (unsigned long long)begin_round, record_rc);
fclose(sim->record_file);
sim->record_file = NULL;
free(buf_events);
return record_rc;
}
}
continue;
}
@ -631,6 +778,18 @@ int sched_sim_driver_run(xce_sched_sim_driver_t* sim,
break;
}
/* Req 5 AC 19 (task 11.5): flush+close the --sim-output file on a clean
* run completion so its contents are durably on disk before this
* function returns -- no sim-driver "stop"/fini function exists to do
* this otherwise (same documented gap sim_picker_thread_fn()'s own
* comment notes for the picker thread), but a lone FILE* close needs
* no broader teardown function to justify doing it here directly. */
if (sim->record_file != NULL) {
fflush(sim->record_file);
fclose(sim->record_file);
sim->record_file = NULL;
}
free(buf_events);
return 0;

192
src/scheduler/src/sim_record.c Normal file
View file

@ -0,0 +1,192 @@
#include <stdio.h>
#include <string.h>
#include <time.h>
#include "controller.capnp.h"
#include "core.h"
/*
* sim_record.c - scheduler-core spec, task 11.5 (Req 5 AC 19: "Simulation
* record file"). Sibling file to sim.c (task 11.1)/sim_loop.c (task
* 11.2)/sim_verify.c (task 11.4) -- same 800-line-cap precedent those
* files' own header comments already establish for this file family
* (CLAUDE.md/coding-style.md's 800-line convention; sim_verify.c's own
* header comment cites the same core.c/core_events.c split task 4.3
* originally set). Owns the --sim-output `.replay` writer: sim_loop.c's
* sched_sim_driver_run() calls into this file's two entry points at
* exactly the message-stream positions Req 5 AC 19's own symmetry table
* specifies -- ReplayHeader (sim-run metadata) -> SnapshotEntry (the
* loaded snapshot, via recording.c's own write_snapshot_entry(), reused
* unchanged) -> ReplayEvent (pass-through, this file) -> ReplaySession
* (per round, accumulated via recording_session.c's own
* scheduler_t-free entry points, called directly from sim_loop.c -- no
* wire-encoding logic needed here beyond what those already provide).
*
* Boundary (Req 5 AC 17): like every file in this family, this file has no
* xce_scheduler_t* / scheduler.h dependency of any kind. In particular it
* deliberately does NOT #include "recording.h": that header itself
* #includes "scheduler.h" (which in turn #includes "client.h" and
* "xce/cluster.h" -- exactly the dependency chain task 11.3's isolation
* audit (test_sim_isolation.c) exists to catch, even though that test's own
* FORBIDDEN_INCLUDES/FORBIDDEN_SYMBOLS lists were written scanning
* sim.h/sim.c/sim_loop.c specifically -- this task extends that same
* regression guard to cover this file too, see test_sim_isolation.c's own
* updated header comment). Every recording.c helper this file reuses is
* instead redeclared via a local `extern`, the SAME "extern declared where
* consumed" idiom sim.c/sim_loop.c/sim_verify.c already use for
* msg_root_has_shape()/sim_decision_eq()/sim_log_mismatch(), and
* recording_events.c/recording_session.c themselves already use for
* write_capnp_message()/sched_record_abort_on_error() (see recording.c's
* own file header comment for that precedent). This file reuses, never
* re-derives, the wire-encoding logic this task's own brief flags:
* write_capnp_message()/write_snapshot_entry() (recording.c, unchanged).
*
* Error handling (this task's own "your judgment, document the choice"
* instruction): both entry points below return whatever the underlying
* write_capnp_message()/write_snapshot_entry() call returned -- 0 on
* success, a negative errno on failure -- and write nothing silently.
* sim_loop.c's own caller-side policy (documented there) is to treat any
* such failure as fatal to the whole simulation run, not merely to disable
* further recording and keep scheduling the way normal-mode's
* sched_record_abort_on_error() does: simulation is a batch/offline tool
* whose --sim-output file is very often the entire reason `--sim` was
* invoked at all (Req 5 AC 19's own stated use cases -- diff-based
* debugging, tuning feedback loops, what-if comparison -- all require a
* COMPLETE output file to exist afterward), so silently degrading to
* "decisions printed to stdout only" would leave behind a truncated,
* silently-incomplete file that still looks structurally valid to a
* downstream `xceadmin scheduler dump` but is missing everything after the
* failure point -- worse than a loud, immediate, whole-run abort.
*/
extern int write_capnp_message(FILE* f, struct capn* c);
extern int write_snapshot_entry(FILE* f, xce_sched_core_t* core, uint64_t ts);
/*
* sim_record_write_header_and_snapshot - Req 5 AC 19's first two output
* messages: a ReplayHeader carrying SIM-RUN metadata (recorderVersion
* "1.0" -- the same wire-format-version convention recording.c's own
* write_replay_header() uses; NOT copied from the input replay file, per
* AC 19's own text: "carrying scheduler version + config hash from the sim
* run, NOT copied from the replay file"), sched_policy from
* `core->job_pick_plugin`, and `cluster = NULL` -- simulation has no
* cluster context by construction (Req 5 AC 17); write_replay_header()'s
* own doc comment (recording.c) confirms hdr.cluster == NULL is an
* already-supported "single-host mode" encode path, not a workaround --
* followed immediately by a SnapshotEntry built from `core`'s CURRENT
* state (write_snapshot_entry(), reused as-is) stamped with `snapshot_ts`.
*
* Called exactly once per run, by sched_sim_driver_run() (sim_loop.c)
* right after startup succeeds and before the event loop begins -- at that
* point `core`'s state is still exactly the state
* xce_sched_core_load_snapshot() produced (no event has been pushed yet),
* so this genuinely IS "the same SnapshotEntry that was loaded" (Req 5 AC
* 19), not merely a same-shaped one taken later. `snapshot_ts` is the
* caller's own already-captured value from that load
* (xce_sched_sim_driver_t.snapshot_ts, sim.h's task 11.5 field) --
* deliberately NOT re-derived via xce_core_now() here; see sim.h's own doc
* comment on that field for why a re-derived value could drift under a
* non-frozen --sim-speed.
*
* configHash is left unset, same documented no-op limitation
* write_replay_header() itself already carries for this field (no utility
* anywhere in this codebase computes one, recording.c's own comment).
*
* Returns 0 on success; a negative errno from whichever write failed first
* (header or snapshot) -- see this file's own header comment for the
* caller-side "treat as fatal to the whole run" policy.
*/
int sim_record_write_header_and_snapshot(FILE* f, xce_sched_core_t* core,
uint64_t snapshot_ts)
{
struct capn c;
struct capn_segment* cs;
replay_header_t hdr;
char wall[32];
time_t now_wall;
struct tm tmv;
ReplayHeader_ptr ptr;
int rc;
memset(&hdr, 0, sizeof(hdr));
now_wall = time(NULL);
gmtime_r(&now_wall, &tmv);
strftime(wall, sizeof(wall), "%Y-%m-%dT%H:%M:%SZ", &tmv);
hdr.recorder_version = "1.0";
hdr.recording_start_ts = snapshot_ts;
hdr.recording_start_wall = wall;
hdr.sched_policy = core->job_pick_plugin; /* may be NULL */
hdr.cluster = NULL; /* no cluster context in simulation
* mode, Req 5 AC 17 -- see this
* function's own doc comment. */
capn_init_malloc(&c);
cs = capn_root(&c).seg;
encode_ReplayHeader_ptr(cs, &ptr, &hdr);
capn_setp(capn_root(&c), 0, ptr.p);
rc = write_capnp_message(f, &c);
capn_free(&c);
if (rc < 0) {
return rc;
}
return write_snapshot_entry(f, core, snapshot_ts);
}
/*
* sim_record_write_event - Req 5 AC 19's per-event pass-through ("passes
* through events from replay file as ReplayEvent messages", the symmetry
* table's own words): re-encodes the SAME already-decoded `replay_event_t`
* sim_loop.c's sim_process_replay_event_message() just decoded from the
* active input event source (sim->replay_file, or under --sim-alt-events,
* sim->alt_events_file -- Req 5 AC 16) and writes it to `f` verbatim, same
* ts and same body -- including the whole-stream ReplayEvent{endSession}
* terminator itself (Req 5 AC 20), so the output stream ends up
* self-terminated exactly like a normal-mode recording does.
*
* Deliberately does NOT route through recording_events.c's
* write_one_replay_event(): that function builds a replay_event_t FROM a
* live xce_core_event_t -- the correct shape for normal-mode recording,
* which only ever has the xce_core_event_t form available (its input
* source is live controller RPC calls, never a decoded .replay message).
* Simulation mode already HAS the decoded replay_event_t itself (the
* .replay stream IS its event source) -- re-encoding it directly is a
* strictly simpler, equally-correct "pass-through", and avoids re-deriving
* write_one_replay_event()'s own xce_core_event_t -> replay_event_t field
* mapping for a translation this file never needs to perform in the first
* place.
*
* Because this operates on whichever decoded message sim_loop.c just read
* -- from --sim-alt-events when set, per Req 5 AC 16 -- the output stream
* naturally reflects the ALTERNATE events under a what-if run, not the
* base file's original ones, with no special-casing needed here: Req 5 AC
* 19's own "What-if comparison" use case (compare a --sim-alt-events run's
* output against the base run's output) falls out of this function's
* ordinary behavior.
*
* Returns whatever write_capnp_message() returned: 0 on success, a
* negative errno on failure.
*/
int sim_record_write_event(FILE* f, replay_event_t* decoded)
{
struct capn c;
struct capn_segment* cs;
ReplayEvent_ptr ptr;
int rc;
capn_init_malloc(&c);
cs = capn_root(&c).seg;
encode_ReplayEvent_ptr(cs, &ptr, decoded);
capn_setp(capn_root(&c), 0, ptr.p);
rc = write_capnp_message(f, &c);
capn_free(&c);
return rc;
}

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

@ -505,16 +505,18 @@ sched_sim_event_loop_test = executable(
# 17, "Simulation isolation"): "a build-time or structural check confirms no
# controller-facing symbol is reachable from the simulation driver's compiled
# object", plus the AC's own "confirmed by the struct definition itself, not
# runtime checks alone" clause. Same source-text-scan + nm(1) link-time-proof
# genre as sched_core_isolation_test above, PLUS a third, narrower structural
# layer scanning ONLY xce_sched_sim_driver_t's own struct body text for
# forbidden field types -- see the test's own header comment for why a
# comment-stripping pass is required first (sim.h's own boundary doc comment
# legitimately names xcecli_ctx_t/xce_cluster_config_t in prose). link_with
# establishes the build-order dependency on sched_sim_isolation_lib (parent
# meson.build -- sim.c AND sim_loop.c compiled together, isolated from the
# rest of the scheduler); XCE_SIM_*_PATH bake in absolute paths the same way
# XCE_CORE_*_PATH does above.
# runtime checks alone" clause. Extended by task 11.5 to also cover the new
# sim_record.c (--sim-output writer) sibling file. Same source-text-scan +
# nm(1) link-time-proof genre as sched_core_isolation_test above, PLUS a
# third, narrower structural layer scanning ONLY xce_sched_sim_driver_t's own
# struct body text for forbidden field types -- see the test's own header
# comment for why a comment-stripping pass is required first (sim.h's own
# boundary doc comment legitimately names xcecli_ctx_t/xce_cluster_config_t
# in prose). link_with establishes the build-order dependency on
# sched_sim_isolation_lib (parent meson.build -- sim.c + sim_loop.c +
# sim_record.c compiled together, isolated from the rest of the scheduler);
# XCE_SIM_*_PATH bake in absolute paths the same way XCE_CORE_*_PATH does
# above.
sched_sim_isolation_test = executable(
'sched_sim_isolation_test',
['test_sim_isolation.c'],
@ -530,6 +532,9 @@ sched_sim_isolation_test = executable(
'-DXCE_SIM_LOOP_C_PATH="'
+ (meson.current_source_dir() / '..' / 'src' / 'sim_loop.c')
+ '"',
'-DXCE_SIM_RECORD_C_PATH="'
+ (meson.current_source_dir() / '..' / 'src' / 'sim_record.c')
+ '"',
'-DXCE_SIM_LIB_PATH="'
+ sched_sim_isolation_lib.full_path()
+ '"'],
@ -566,6 +571,45 @@ sched_sim_verify_test = executable(
dependencies: [scheduler_test_deps],
install: false)
# Contract test for what-if event-source branching (scheduler-core spec,
# task 11.5, Req 5 AC 16): running with --sim-alt-events produces decisions
# driven by the alternate events, not the base file's original events. A
# baseline run (no --sim-alt-events) proves the base file's own events drive
# scheduling by default; a --sim-alt-events run against the SAME base file's
# snapshot proves the alternate file's events drive it instead, and that the
# base file's own events are never read once --sim-alt-events is set. Same
# white-box #include pattern as sched_sim_verify_test (core.c, core_clock.c,
# core_events.c, core_session.c, decision.c, plugin.c, plugins/builtin_fcfs.c,
# recording.c, sim.c, sim_loop.c, sim_verify.c, snapshot.c/snapshot_wire.c/
# snapshot_load.c) -- sim_verify.c is required at link time even though this
# file never sets --sim-verify (sim_loop.c externs its two entry points
# unconditionally, same reason sched_sim_event_loop_test needs it).
sched_sim_alt_events_test = executable(
'sched_sim_alt_events_test',
['test_sim_alt_events.c'],
include_directories:[incdir],
dependencies: [scheduler_test_deps],
install: false)
# Contract test for the --sim-output `.replay` writer (scheduler-core spec,
# task 11.5, Req 5 AC 19): the output file, when requested, is decodable by
# the same reader used for normal-mode recordings (test_record_lifecycle.c's
# own capn_getp(capn_root(&c), 0, 1) + read_X()/decode_X() technique, reused
# verbatim here), with full ReplayHeader -> SnapshotEntry -> ReplayEvent* ->
# ReplaySession -> ReplayEvent{endSession} stream symmetry (Req 5 AC 19's own
# symmetry table). A second test proves --sim-output composes with
# --sim-alt-events: the output reflects the alternate run's own events and
# decision, not the base file's. Same white-box #include pattern as
# sched_sim_verify_test PLUS ../src/recording_session.c (the scheduler_t-free
# session-accumulator entry points sim_loop.c now calls) PLUS the new
# ../src/sim_record.c (the code under test).
sched_sim_record_output_test = executable(
'sched_sim_record_output_test',
['test_sim_record_output.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',
@ -647,3 +691,9 @@ test('scheduler sim isolation test',
test('scheduler sim verify test',
sched_sim_verify_test, suite: 'scheduler',
env: ['ASAN_OPTIONS=detect_leaks=0'])
test('scheduler sim alt events test',
sched_sim_alt_events_test, suite: 'scheduler',
env: ['ASAN_OPTIONS=detect_leaks=0'])
test('scheduler sim record output test',
sched_sim_record_output_test, suite: 'scheduler',
env: ['ASAN_OPTIONS=detect_leaks=0'])

485
src/scheduler/testsuite/test_sim_alt_events.c Normal file
View file

@ -0,0 +1,485 @@
/*
* test_sim_alt_events.c - contract test for what-if event-source branching
* (scheduler-core spec, task 11.5, Req 5 AC 16).
*
* Observable under test (task 11.5's own text, task text's Observable (a)):
* - running with an alternate-events file produces decisions driven by
* the alternate events, not the base file's original events.
*
* Two tests:
* - a baseline run against the base .replay file ALONE (no
* --sim-alt-events) confirms its own event stream drives a PEND
* decision for the BASE job, and NOT the alt job.
* - a what-if run against the SAME base file's snapshot, but with
* --sim-alt-events pointed at a separate events-only stream, confirms
* the opposite: a PEND decision for the ALT job, and NOT the base job.
* Together these prove the switch is real (not merely "some decision
* appears"): the DRIVING SOURCE genuinely changed.
*
* Fixture strategy: the SAME real QUEUE_UPDATE+JOBS_PUSH-driving-the-real-
* builtin_fcfs-picker technique test_sim_verify.c's own integration-level
* fixture helpers establish (build_fixture_scheduler() records
* schedPolicy="builtin_fcfs" so sim_register_default_picker() resolves the
* real plugin) -- since core.c's own sched_job_is_ready()/
* sched_queue_is_ready() are unconditional `return 1` stubs (task 2.4's own
* documented, still-live scope note), any queue/job pair immediately
* becomes a next_job() candidate, so each run deterministically produces
* exactly one live PEND decision via make_pendreason() for whichever jobid
* that run's own event stream pushed. Decisions are observed by
* redirecting STDOUT_FILENO around sched_sim_driver_run(), the same
* dup()/dup2() capture technique test_sim_verify.c's own
* run_sim_and_capture_stdout() establishes (sim_print_decision()'s
* "[sim] PEND jobid=%llu" line, sim_loop.c).
*
* The alt-events file itself is a BARE sequence of ReplayEvent/ReplaySession
* messages (no ReplayHeader, no SnapshotEntry) -- Req 5 AC 16's own text:
* "the driver loads the snapshot from the base file, then replays events
* from the alt-events file (same .replay capnp stream format)". Once
* --sim-alt-events is set, sched_sim_driver_run()'s main loop reads
* EXCLUSIVELY from it from message 1 onward -- the same shape-based
* message dispatch (msg_root_has_shape(), never position-based beyond the
* base file's own first two messages) that already governs every message
* after a SnapshotEntry in a full `.replay` stream, so an alt-events file
* that is simply "the tail of a .replay stream" needs no header/snapshot
* prefix of its own.
*
* White-box #include of the same TU set test_sim_verify.c uses (see that
* file's own header comment) -- core.c/core_clock.c/core_events.c/
* core_session.c/decision.c/plugin.c/plugins/builtin_fcfs.c/recording.c/
* recording_session.c/sim.c/sim_loop.c/sim_record.c/sim_verify.c/
* snapshot.c/snapshot_load.c/snapshot_wire.c. sim_verify.c AND
* recording_session.c/sim_record.c are all required at link time
* (sched_sim_driver_run() externs sim_decision_eq()/sim_log_mismatch() and
* sched_record_session_ctx_new()/_finalize()/sim_record_write_header_and_
* snapshot()/sim_record_write_event() unconditionally) even though this
* file never sets --sim-verify or --sim-output, same reason
* test_sim_event_loop.c's own header comment documents.
*/
#include <errno.h>
#include <limits.h>
#include <stdatomic.h>
#include <stdbool.h>
#include <stdint.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>
#include <unity/unity.h>
#include "controller.capnp.h"
#include "kvmgr.h"
#include "scheduler.h"
#include "xstring.h"
/* Pull in the real functions under test into this TU (white-box #include,
* same convention as test_sim_verify.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/recording_session.c"
#include "../src/sim.c"
#include "../src/sim_loop.c"
#include "../src/sim_record.c"
#include "../src/sim_verify.c"
#include "../src/snapshot.c"
#include "../src/snapshot_load.c"
#include "../src/snapshot_wire.c"
void kill_sched_job(void* arg) { (void)arg; }
void kill_sched_queue(void* arg) { (void)arg; }
void kill_sched_worker(void* arg) { (void)arg; }
void kill_sched_user(void* arg) { (void)arg; }
void kill_sched_project(void* arg) { (void)arg; }
void kill_sched_resval(xce_val_t* val) { (void)val; }
void sched_worker_release_host_ctx(xce_sched_worker_t* w) { (void)w; }
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
* copy test_sim_verify.c/test_sim_event_loop.c/test_sim_startup.c each
* carry (schedobj.c is not linked here: 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 -- same techniques test_sim_verify.c already */
/* establishes for its own integration-level tests. */
/* ---------------------------------------------------------------------- */
static void make_tmp_dir(char* out, size_t out_len, const char* suffix)
{
char tmpl[PATH_MAX];
snprintf(tmpl, sizeof(tmpl), "/tmp/sched_altevents%s_XXXXXX", suffix);
TEST_ASSERT_NOT_NULL(mkdtemp(tmpl));
TEST_ASSERT_TRUE(strlen(tmpl) < out_len);
strcpy(out, tmpl);
}
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);
sched->core.job_pick_plugin = (char*)"builtin_fcfs";
sched->record_dir = record_dir;
}
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;
}
static void append_queue_update_event(FILE* f, uint64_t ts,
const char* queue_name)
{
xce_queue_t queue;
replay_event_t src;
struct capn c;
struct capn_segment* cs;
ReplayEvent_ptr p;
memset(&queue, 0, sizeof(queue));
queue.queue = (char*)queue_name;
queue.status = QUEUE_STAT_OPEN;
queue.priority = 1;
memset(&src, 0, sizeof(src));
src.ts = ts;
src.body_which = ReplayEvent_body_queueUpdate;
src.body.queue_update = &queue;
capn_init_malloc(&c);
cs = capn_root(&c).seg;
encode_ReplayEvent_ptr(cs, &p, &src);
capn_setp(capn_root(&c), 0, p.p);
TEST_ASSERT_EQUAL_INT(0, write_capnp_message(f, &c));
capn_free(&c);
}
static void append_jobs_push_event(FILE* f, uint64_t ts, uint64_t jobid,
const char* queue_name)
{
job_submit_data_t submit;
xce_job_t job;
xce_job_t* jobs[1];
xce_jobs_data_t jobs_data;
replay_event_t src;
struct capn c;
struct capn_segment* cs;
ReplayEvent_ptr p;
memset(&submit, 0, sizeof(submit));
submit.queue = (char*)queue_name;
submit.user = (char*)"alice";
submit.mintasks = 1;
submit.maxtasks = 1;
memset(&job, 0, sizeof(job));
job.jobid = jobid;
job.state = JOB_STAT_PEND;
job.submit = &submit;
jobs[0] = &job;
memset(&jobs_data, 0, sizeof(jobs_data));
jobs_data.n_total = 1;
jobs_data.count = 1;
jobs_data.jobs = jobs;
memset(&src, 0, sizeof(src));
src.ts = ts;
src.body_which = ReplayEvent_body_jobsPush;
src.body.jobs_push = &jobs_data;
capn_init_malloc(&c);
cs = capn_root(&c).seg;
encode_ReplayEvent_ptr(cs, &p, &src);
capn_setp(capn_root(&c), 0, p.p);
TEST_ASSERT_EQUAL_INT(0, write_capnp_message(f, &c));
capn_free(&c);
}
static void append_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);
}
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);
}
/* Runs sched_sim_driver_run() with optional --sim-alt-events, capturing
* everything written to stdout during the call -- same dup()/dup2()
* technique test_sim_verify.c's own run_sim_and_capture_stdout()
* establishes, extended with an `alt_path` parameter (NULL to omit
* --sim-alt-events entirely). */
static char* run_sim_and_capture_stdout(const char* base_path,
const char* alt_path, int* out_rc)
{
sched_sim_args_t args;
xce_sched_sim_driver_t sim;
char tmp[] = "/tmp/xce_altevents_run_XXXXXX";
int fd = mkstemp(tmp);
int saved_stdout;
off_t len;
char* buf;
TEST_ASSERT_TRUE_MESSAGE(fd >= 0, "mkstemp for stdout capture must "
"succeed");
fflush(stdout);
saved_stdout = dup(STDOUT_FILENO);
TEST_ASSERT_TRUE_MESSAGE(saved_stdout >= 0, "dup(STDOUT_FILENO) must "
"succeed");
TEST_ASSERT_TRUE_MESSAGE(dup2(fd, STDOUT_FILENO) >= 0,
"dup2 redirect of STDOUT_FILENO must succeed");
memset(&args, 0, sizeof(args));
args.replay_path = base_path;
args.sim_start_ts = SCHED_SIM_START_TS_LATEST;
args.sim_speed = 0; /* frozen -- deterministic, no wall-clock wait */
args.alt_events_path = alt_path;
*out_rc = sched_sim_driver_run(&sim, &args);
fflush(stdout);
dup2(saved_stdout, STDOUT_FILENO);
close(saved_stdout);
len = lseek(fd, 0, SEEK_CUR);
TEST_ASSERT_TRUE(len >= 0);
buf = (char*)calloc(1, (size_t)len + 1);
TEST_ASSERT_NOT_NULL(buf);
TEST_ASSERT_TRUE(lseek(fd, 0, SEEK_SET) == 0);
if (len > 0) {
ssize_t nread = read(fd, buf, (size_t)len);
TEST_ASSERT_TRUE(nread == len);
}
close(fd);
unlink(tmp);
return buf;
}
#define BASE_JOBID 777ULL
#define ALT_JOBID 888ULL
/* ---------------------------------------------------------------------- */
/* Test 1: baseline -- no --sim-alt-events, the base file's OWN events */
/* drive scheduling. */
/* ---------------------------------------------------------------------- */
static void test_baseline_run_uses_base_file_events(void)
{
char base_jobid_marker[32];
char alt_jobid_marker[32];
char path[PATH_MAX];
FILE* f;
int rc;
char* captured;
snprintf(base_jobid_marker, sizeof(base_jobid_marker), "jobid=%llu",
(unsigned long long)BASE_JOBID);
snprintf(alt_jobid_marker, sizeof(alt_jobid_marker), "jobid=%llu",
(unsigned long long)ALT_JOBID);
f = write_header_and_snapshot(path, sizeof(path), "_baseline");
TEST_ASSERT_NOT_NULL(f);
append_queue_update_event(f, 1000ULL, "normal");
append_jobs_push_event(f, 1500ULL, BASE_JOBID, "normal");
append_replay_session(f, 2000ULL);
append_end_of_stream_marker(f);
fclose(f);
captured = run_sim_and_capture_stdout(path, NULL, &rc);
TEST_ASSERT_EQUAL_INT(0, rc);
TEST_ASSERT_NOT_NULL_MESSAGE(
strstr(captured, base_jobid_marker),
"baseline run (no --sim-alt-events) must produce a decision for "
"the base file's own jobid");
TEST_ASSERT_NULL_MESSAGE(
strstr(captured, alt_jobid_marker),
"baseline run must never produce a decision for a jobid it never "
"saw");
free(captured);
printf("test_baseline_run_uses_base_file_events: PASS\n");
}
/* ---------------------------------------------------------------------- */
/* Test 2: what-if -- --sim-alt-events set, the ALTERNATE file's events */
/* drive scheduling instead (Req 5 AC 16's own Observable). */
/* ---------------------------------------------------------------------- */
static void test_alt_events_run_uses_alt_file_events_not_base(void)
{
char base_jobid_marker[32];
char alt_jobid_marker[32];
char base_path[PATH_MAX];
char alt_path[PATH_MAX];
FILE* base_f;
FILE* alt_f;
int rc;
char* captured;
snprintf(base_jobid_marker, sizeof(base_jobid_marker), "jobid=%llu",
(unsigned long long)BASE_JOBID);
snprintf(alt_jobid_marker, sizeof(alt_jobid_marker), "jobid=%llu",
(unsigned long long)ALT_JOBID);
/* Base file: header + empty snapshot, then its OWN (never-to-be-read,
* once --sim-alt-events is set) events for BASE_JOBID -- present to
* prove the switch is real: if alt-events were silently ignored, this
* run would fall back to seeing BASE_JOBID here too. */
base_f = write_header_and_snapshot(base_path, sizeof(base_path),
"_whatif_base");
TEST_ASSERT_NOT_NULL(base_f);
append_queue_update_event(base_f, 1000ULL, "normal");
append_jobs_push_event(base_f, 1500ULL, BASE_JOBID, "normal");
append_replay_session(base_f, 2000ULL);
append_end_of_stream_marker(base_f);
fclose(base_f);
/* Alt-events file: a BARE ReplayEvent/ReplaySession stream (no
* ReplayHeader, no SnapshotEntry -- this file's own header comment) for
* a DIFFERENT job, ALT_JOBID, on a queue the alt stream creates for
* itself (the loaded snapshot has zero queues). */
TEST_ASSERT_TRUE(snprintf(alt_path, sizeof(alt_path), "%s.alt", base_path)
< (int)sizeof(alt_path));
alt_f = fopen(alt_path, "w");
TEST_ASSERT_NOT_NULL(alt_f);
append_queue_update_event(alt_f, 3000ULL, "altqueue");
append_jobs_push_event(alt_f, 3500ULL, ALT_JOBID, "altqueue");
append_replay_session(alt_f, 4000ULL);
append_end_of_stream_marker(alt_f);
fclose(alt_f);
captured = run_sim_and_capture_stdout(base_path, alt_path, &rc);
TEST_ASSERT_EQUAL_INT(0, rc);
TEST_ASSERT_NOT_NULL_MESSAGE(
strstr(captured, alt_jobid_marker),
"--sim-alt-events run must produce a decision for the ALTERNATE "
"file's own jobid");
TEST_ASSERT_NULL_MESSAGE(
strstr(captured, base_jobid_marker),
"--sim-alt-events run must NOT produce a decision for the base "
"file's original jobid -- the base file's own events must never "
"be read once --sim-alt-events is set");
free(captured);
printf("test_alt_events_run_uses_alt_file_events_not_base: PASS\n");
}
int main(void)
{
UNITY_BEGIN();
RUN_TEST(test_baseline_run_uses_base_file_events);
RUN_TEST(test_alt_events_run_uses_alt_file_events_not_base);
return UNITY_END();
}

19
src/scheduler/testsuite/test_sim_event_loop.c
View file

@ -60,7 +60,13 @@
* extern functions defined there; this white-box TU must include it too or
* those calls are undefined references at link time, even though neither
* test in this file exercises --sim-verify itself -- that is
* test_sim_verify.c's own job) PLUS
* test_sim_verify.c's own job) PLUS ../src/recording_session.c/
* ../src/sim_record.c (task 11.5's sibling files -- same reasoning:
* sched_sim_driver_run() now unconditionally references
* sched_record_session_ctx_new()/_finalize()/sim_record_write_header_and_
* snapshot()/sim_record_write_event() as extern functions, undefined
* references at link time otherwise, even though this file never sets
* --sim-output either) 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).
*/
@ -89,8 +95,10 @@
#include "../src/plugin.c"
#include "../src/plugins/builtin_fcfs.c"
#include "../src/recording.c"
#include "../src/recording_session.c"
#include "../src/sim.c"
#include "../src/sim_loop.c"
#include "../src/sim_record.c"
#include "../src/sim_verify.c"
#include "../src/snapshot.c"
#include "../src/snapshot_load.c"
@ -444,13 +452,20 @@ test_process_replay_event_advances_virtual_clock_to_exact_event_ts_in_file_order
struct capn c;
capn_ptr root;
xce_core_event_t ev;
int record_rc = 0;
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);
/* record_rc: task 11.5's new --sim-output pass-through out-param
* (sim_loop.c) -- always 0 here, sim.record_file is NULL for this
* whole fixture (no --sim-output involved in this task's own
* Observable). */
translated
= sim_process_replay_event_message(&sim, &c, root, &ev, &record_rc);
TEST_ASSERT_EQUAL_INT(0, record_rc);
TEST_ASSERT_EQUAL_INT(1, translated);
TEST_ASSERT_EQUAL_INT(XCE_CORE_EVENT_JOB_STATE_CHANGE, ev.type);

75
src/scheduler/testsuite/test_sim_isolation.c
View file

@ -4,10 +4,15 @@
* or structural check confirms no controller-facing symbol is reachable
* from the simulation driver's compiled object" -- plus the AC's own
* "confirmed by the struct definition itself, not runtime checks alone"
* clause. Scope: sim.h + sim.c + sim_loop.c together (task 11.1 + task
* 11.2's sibling file) -- NOT a re-audit of core.c (task 4.1's own
* clause. Scope: sim.h + sim.c + sim_loop.c + sim_record.c together (task
* 11.1 + task 11.2's sibling file + task 11.5's sibling file, the
* --sim-output writer) -- NOT a re-audit of core.c (task 4.1's own
* territory, test_core_isolation.c) and NOT plugin.c/builtin_fcfs.c (shared
* normal-mode infra, out of this task's boundary).
* normal-mode infra, out of this task's boundary). sim_verify.c (task 11.4)
* is likewise out of scope here -- it is never compiled into
* sched_sim_isolation_lib below and carries no isolation-sensitive logic of
* its own (pure struct comparison, no I/O); extending this guard to it was
* not part of task 11.4's own text and is left as-is.
*
* Three layers, extending test_core_isolation.c's two-layer pattern (same
* genre: source-text scans of the real, current on-disk files, not a frozen
@ -74,9 +79,13 @@
#error \
"XCE_SIM_LOOP_C_PATH must be defined by meson.build to sim_loop.c's absolute path"
#endif
#ifndef XCE_SIM_RECORD_C_PATH
#error \
"XCE_SIM_RECORD_C_PATH must be defined by meson.build to sim_record.c's absolute path"
#endif
#ifndef XCE_SIM_LIB_PATH
#error \
"XCE_SIM_LIB_PATH must be defined by meson.build to the absolute path of sim.c+sim_loop.c's isolated static_library() archive"
"XCE_SIM_LIB_PATH must be defined by meson.build to the absolute path of sim.c+sim_loop.c+sim_record.c's isolated static_library() archive"
#endif
/* Forbidden headers (design.md "Allowed Dependencies", same list
@ -346,81 +355,100 @@ void tearDown(void) { return; }
* failure mode this whole test exists to avoid). */
static void test_sim_source_paths_are_readable(void)
{
char* h = read_whole_file(XCE_SIM_H_PATH);
char* c = read_whole_file(XCE_SIM_C_PATH);
char* loop = read_whole_file(XCE_SIM_LOOP_C_PATH);
char* h = read_whole_file(XCE_SIM_H_PATH);
char* c = read_whole_file(XCE_SIM_C_PATH);
char* loop = read_whole_file(XCE_SIM_LOOP_C_PATH);
char* record = read_whole_file(XCE_SIM_RECORD_C_PATH);
TEST_ASSERT_NOT_NULL_MESSAGE(h, XCE_SIM_H_PATH);
TEST_ASSERT_NOT_NULL_MESSAGE(c, XCE_SIM_C_PATH);
TEST_ASSERT_NOT_NULL_MESSAGE(loop, XCE_SIM_LOOP_C_PATH);
TEST_ASSERT_NOT_NULL_MESSAGE(record, XCE_SIM_RECORD_C_PATH);
free(h);
free(c);
free(loop);
free(record);
}
static void test_sim_never_includes_forbidden_headers(void)
{
char* h_raw = read_whole_file(XCE_SIM_H_PATH);
char* c_raw = read_whole_file(XCE_SIM_C_PATH);
char* loop_raw = read_whole_file(XCE_SIM_LOOP_C_PATH);
char* h_raw = read_whole_file(XCE_SIM_H_PATH);
char* c_raw = read_whole_file(XCE_SIM_C_PATH);
char* loop_raw = read_whole_file(XCE_SIM_LOOP_C_PATH);
char* record_raw = read_whole_file(XCE_SIM_RECORD_C_PATH);
char* h;
char* c;
char* loop;
char* record;
TEST_ASSERT_NOT_NULL(h_raw);
TEST_ASSERT_NOT_NULL(c_raw);
TEST_ASSERT_NOT_NULL(loop_raw);
TEST_ASSERT_NOT_NULL(record_raw);
h = strip_c_comments(h_raw);
c = strip_c_comments(c_raw);
loop = strip_c_comments(loop_raw);
h = strip_c_comments(h_raw);
c = strip_c_comments(c_raw);
loop = strip_c_comments(loop_raw);
record = strip_c_comments(record_raw);
TEST_ASSERT_NOT_NULL(h);
TEST_ASSERT_NOT_NULL(c);
TEST_ASSERT_NOT_NULL(loop);
TEST_ASSERT_NOT_NULL(record);
assert_no_forbidden_includes(h, "sim.h");
assert_no_forbidden_includes(c, "sim.c");
assert_no_forbidden_includes(loop, "sim_loop.c");
assert_no_forbidden_includes(record, "sim_record.c");
free(h_raw);
free(c_raw);
free(loop_raw);
free(record_raw);
free(h);
free(c);
free(loop);
free(record);
}
static void test_sim_never_calls_forbidden_symbols(void)
{
char* h_raw = read_whole_file(XCE_SIM_H_PATH);
char* c_raw = read_whole_file(XCE_SIM_C_PATH);
char* loop_raw = read_whole_file(XCE_SIM_LOOP_C_PATH);
char* h_raw = read_whole_file(XCE_SIM_H_PATH);
char* c_raw = read_whole_file(XCE_SIM_C_PATH);
char* loop_raw = read_whole_file(XCE_SIM_LOOP_C_PATH);
char* record_raw = read_whole_file(XCE_SIM_RECORD_C_PATH);
char* h;
char* c;
char* loop;
char* record;
TEST_ASSERT_NOT_NULL(h_raw);
TEST_ASSERT_NOT_NULL(c_raw);
TEST_ASSERT_NOT_NULL(loop_raw);
TEST_ASSERT_NOT_NULL(record_raw);
h = strip_c_comments(h_raw);
c = strip_c_comments(c_raw);
loop = strip_c_comments(loop_raw);
h = strip_c_comments(h_raw);
c = strip_c_comments(c_raw);
loop = strip_c_comments(loop_raw);
record = strip_c_comments(record_raw);
TEST_ASSERT_NOT_NULL(h);
TEST_ASSERT_NOT_NULL(c);
TEST_ASSERT_NOT_NULL(loop);
TEST_ASSERT_NOT_NULL(record);
assert_no_forbidden_symbols(h, "sim.h");
assert_no_forbidden_symbols(c, "sim.c");
assert_no_forbidden_symbols(loop, "sim_loop.c");
assert_no_forbidden_symbols(record, "sim_record.c");
free(h_raw);
free(c_raw);
free(loop_raw);
free(record_raw);
free(h);
free(c);
free(loop);
free(record);
}
/*
@ -464,8 +492,9 @@ static void test_sim_driver_struct_has_no_forbidden_fields(void)
/*
* test_sim_object_has_no_forbidden_undefined_symbols - the literal
* build-time/link-time proof: sim.c + sim_loop.c, compiled together in
* isolation into sched_sim_isolation_lib (meson.build), must not reference
* build-time/link-time proof: sim.c + sim_loop.c + sim_record.c, compiled
* together in isolation into sched_sim_isolation_lib (meson.build), must
* not reference
* (as an undefined external symbol in the compiled archive) anything
* matching the forbidden prefixes above. Independent of what any header
* transitively declares -- reflects only what this driver's compiled code
@ -494,8 +523,8 @@ static void test_sim_object_has_no_forbidden_undefined_symbols(void)
/* posix format for an undefined symbol: "<name> U" (no
* value/size columns). Skip anything that doesn't parse as
* exactly that -- e.g. a blank line, or an archive member header
* some nm versions print (this archive has two member objects,
* sim.c.o and sim_loop.c.o). */
* some nm versions print (this archive has three member objects,
* sim.c.o, sim_loop.c.o, and sim_record.c.o). */
if (sscanf(line, "%511s %c", symbol, &type) != 2) {
continue;
}

613
src/scheduler/testsuite/test_sim_record_output.c Normal file
View file

@ -0,0 +1,613 @@
/*
* test_sim_record_output.c - contract test for the --sim-output `.replay`
* writer (scheduler-core spec, task 11.5, Req 5 AC 19).
*
* Observable under test (task 11.5's own text, task text's Observable (b)):
* - the output file, when requested, is decodable by the same reader
* used for normal-mode recordings.
*
* "The same reader" is not a separate module anywhere in this codebase --
* both normal-mode recording's own contract test
* (test_record_lifecycle.c's test_full_cycle_start_event_session_stop_
* produces_expected_stream) and simulation's own production code
* (sim.c/sim_loop.c's msg_root_has_shape() + read_X()/decode_X()) decode a
* `.replay` stream the identical way: capn_init_fp() one message at a time,
* capn_getp(capn_root(&c), 0, 1) to reach the payload pointer (this
* codebase's own universal "real payload lives at pointer slot 0 of the
* implicit root struct" convention, sim.c's own file header comment), then
* the generated read_X()/decode_X() pair from controller.capnp.h. This
* file decodes its own --sim-output file with EXACTLY that same sequence
* (test_record_lifecycle.c's own literal technique, reused verbatim, not
* reinvented) -- proving the output stream is genuinely readable by the
* one true `.replay` reader this codebase has, not merely "non-empty".
*
* Two tests:
* - decodability + full structural symmetry: ReplayHeader (sim-run
* metadata, no cluster) -> SnapshotEntry (the loaded snapshot, same ts
* the driver itself used) -> ReplayEvent (pass-through, x2) ->
* ReplaySession (the live decision the real builtin_fcfs picker
* produced) -> ReplayEvent{endSession} -> EOF. Matches Req 5 AC 19's
* own symmetry table row-for-row.
* - composition with --sim-alt-events (Req 5 AC 19's own "What-if
* comparison" use case): the SAME run, but with --sim-alt-events set,
* produces an output file whose pass-through events and recorded
* decision reflect the ALTERNATE jobid, not the base file's original
* one -- proving the two task-11.5 features compose, not merely
* coexist independently.
*
* Fixture strategy: same real QUEUE_UPDATE+JOBS_PUSH-driving-the-real-
* builtin_fcfs-picker technique test_sim_verify.c/test_sim_alt_events.c
* already establish.
*
* White-box #include of the same TU set test_sim_verify.c uses (see that
* file's own header comment) PLUS ../src/recording_session.c (the
* scheduler_t-free session-accumulator entry points sim_loop.c now calls)
* PLUS the new ../src/sim_record.c (the code under test).
*/
#include <errno.h>
#include <limits.h>
#include <stdatomic.h>
#include <stdbool.h>
#include <stdint.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>
#include <unity/unity.h>
#include "controller.capnp.h"
#include "kvmgr.h"
#include "scheduler.h"
#include "xstring.h"
/* Pull in the real functions under test into this TU (white-box #include,
* same convention as test_sim_verify.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/recording_session.c"
#include "../src/sim.c"
#include "../src/sim_loop.c"
#include "../src/sim_record.c"
#include "../src/sim_verify.c"
#include "../src/snapshot.c"
#include "../src/snapshot_load.c"
#include "../src/snapshot_wire.c"
void kill_sched_job(void* arg) { (void)arg; }
void kill_sched_queue(void* arg) { (void)arg; }
void kill_sched_worker(void* arg) { (void)arg; }
void kill_sched_user(void* arg) { (void)arg; }
void kill_sched_project(void* arg) { (void)arg; }
void kill_sched_resval(xce_val_t* val) { (void)val; }
void sched_worker_release_host_ctx(xce_sched_worker_t* w) { (void)w; }
int sched_fill_limit_requests(xce_scheduler_t* sched, xvec_idval_t* vec,
xlist_t* limits)
{
(void)sched;
(void)vec;
(void)limits;
return 0;
}
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 -- same techniques test_sim_verify.c/ */
/* test_sim_alt_events.c already establish. */
/* ---------------------------------------------------------------------- */
static void make_tmp_dir(char* out, size_t out_len, const char* suffix)
{
char tmpl[PATH_MAX];
snprintf(tmpl, sizeof(tmpl), "/tmp/sched_recordout%s_XXXXXX", suffix);
TEST_ASSERT_NOT_NULL(mkdtemp(tmpl));
TEST_ASSERT_TRUE(strlen(tmpl) < out_len);
strcpy(out, tmpl);
}
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);
sched->core.job_pick_plugin = (char*)"builtin_fcfs";
sched->record_dir = record_dir;
}
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;
}
static void append_queue_update_event(FILE* f, uint64_t ts,
const char* queue_name)
{
xce_queue_t queue;
replay_event_t src;
struct capn c;
struct capn_segment* cs;
ReplayEvent_ptr p;
memset(&queue, 0, sizeof(queue));
queue.queue = (char*)queue_name;
queue.status = QUEUE_STAT_OPEN;
queue.priority = 1;
memset(&src, 0, sizeof(src));
src.ts = ts;
src.body_which = ReplayEvent_body_queueUpdate;
src.body.queue_update = &queue;
capn_init_malloc(&c);
cs = capn_root(&c).seg;
encode_ReplayEvent_ptr(cs, &p, &src);
capn_setp(capn_root(&c), 0, p.p);
TEST_ASSERT_EQUAL_INT(0, write_capnp_message(f, &c));
capn_free(&c);
}
static void append_jobs_push_event(FILE* f, uint64_t ts, uint64_t jobid,
const char* queue_name)
{
job_submit_data_t submit;
xce_job_t job;
xce_job_t* jobs[1];
xce_jobs_data_t jobs_data;
replay_event_t src;
struct capn c;
struct capn_segment* cs;
ReplayEvent_ptr p;
memset(&submit, 0, sizeof(submit));
submit.queue = (char*)queue_name;
submit.user = (char*)"alice";
submit.mintasks = 1;
submit.maxtasks = 1;
memset(&job, 0, sizeof(job));
job.jobid = jobid;
job.state = JOB_STAT_PEND;
job.submit = &submit;
jobs[0] = &job;
memset(&jobs_data, 0, sizeof(jobs_data));
jobs_data.n_total = 1;
jobs_data.count = 1;
jobs_data.jobs = jobs;
memset(&src, 0, sizeof(src));
src.ts = ts;
src.body_which = ReplayEvent_body_jobsPush;
src.body.jobs_push = &jobs_data;
capn_init_malloc(&c);
cs = capn_root(&c).seg;
encode_ReplayEvent_ptr(cs, &p, &src);
capn_setp(capn_root(&c), 0, p.p);
TEST_ASSERT_EQUAL_INT(0, write_capnp_message(f, &c));
capn_free(&c);
}
static void append_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);
}
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);
}
#define BASE_JOBID 777ULL
#define ALT_JOBID 888ULL
/* ---------------------------------------------------------------------- */
/* Test 1: --sim-output produces a file decodable by the same reader used */
/* for normal-mode recordings, with full ReplayHeader -> SnapshotEntry -> */
/* ReplayEvent* -> ReplaySession -> ReplayEvent{endSession} symmetry. */
/* ---------------------------------------------------------------------- */
static void test_output_file_is_decodable_with_full_stream_symmetry(void)
{
char base_path[PATH_MAX];
char out_path[PATH_MAX];
FILE* base_f;
int tmp_fd;
sched_sim_args_t args;
xce_sched_sim_driver_t sim;
int rc;
FILE* f;
struct capn c;
base_f = write_header_and_snapshot(base_path, sizeof(base_path),
"_decode_base");
TEST_ASSERT_NOT_NULL(base_f);
append_queue_update_event(base_f, 1000ULL, "normal");
append_jobs_push_event(base_f, 1500ULL, BASE_JOBID, "normal");
append_replay_session(base_f, 2000ULL);
append_end_of_stream_marker(base_f);
fclose(base_f);
snprintf(out_path, sizeof(out_path), "/tmp/xce_simoutput_decode_XXXXXX");
tmp_fd = mkstemp(out_path);
TEST_ASSERT_TRUE(tmp_fd >= 0);
close(tmp_fd);
memset(&args, 0, sizeof(args));
args.replay_path = base_path;
args.sim_start_ts = SCHED_SIM_START_TS_LATEST;
args.sim_speed = 0; /* frozen -- deterministic snapshot_ts */
args.output_path = out_path;
rc = sched_sim_driver_run(&sim, &args);
TEST_ASSERT_EQUAL_INT(0, rc);
f = fopen(out_path, "r");
TEST_ASSERT_NOT_NULL(f);
/* msg 1: ReplayHeader -- sim-run metadata, no cluster, sched_policy
* "builtin_fcfs" (this fixture's own recorded policy). Same
* read_ReplayHeader()/decode_ReplayHeader() technique
* test_record_lifecycle.c's own decode helpers use. */
{
ReplayHeader_ptr p;
struct ReplayHeader wire;
replay_header_t dst;
TEST_ASSERT_EQUAL_INT(0, capn_init_fp(&c, f, 0));
p.p = capn_getp(capn_root(&c), 0, 1);
read_ReplayHeader(&wire, p);
memset(&dst, 0, sizeof(dst));
decode_ReplayHeader(&dst, &wire);
TEST_ASSERT_EQUAL_STRING("1.0", dst.recorder_version);
TEST_ASSERT_NOT_NULL(dst.sched_policy);
TEST_ASSERT_EQUAL_STRING("builtin_fcfs", dst.sched_policy);
TEST_ASSERT_NULL_MESSAGE(
dst.cluster,
"simulation's own ReplayHeader must carry no cluster context "
"(Req 5 AC 17)");
free_ReplayHeader(&dst);
capn_free(&c);
}
/* msg 2: SnapshotEntry -- the loaded snapshot's own ts, verbatim
* (sim.snapshot_ts, stamped by sched_sim_driver_start()). */
{
SnapshotEntry_ptr p;
struct SnapshotEntry wire;
replay_snapshot_entry_t dst;
TEST_ASSERT_EQUAL_INT(0, capn_init_fp(&c, f, 0));
p.p = capn_getp(capn_root(&c), 0, 1);
read_SnapshotEntry(&wire, p);
memset(&dst, 0, sizeof(dst));
decode_SnapshotEntry(&dst, &wire);
TEST_ASSERT_EQUAL_UINT64(sim.snapshot_ts, dst.ts);
free_SnapshotEntry(&dst);
capn_free(&c);
}
/* msg 3: ReplayEvent{queueUpdate} pass-through. */
{
ReplayEvent_ptr p;
struct ReplayEvent wire;
replay_event_t dst;
TEST_ASSERT_EQUAL_INT(0, capn_init_fp(&c, f, 0));
p.p = capn_getp(capn_root(&c), 0, 1);
read_ReplayEvent(&wire, p);
memset(&dst, 0, sizeof(dst));
decode_ReplayEvent(&dst, &wire);
TEST_ASSERT_EQUAL_INT((int)ReplayEvent_body_queueUpdate,
(int)dst.body_which);
TEST_ASSERT_EQUAL_UINT64(1000ULL, dst.ts);
TEST_ASSERT_NOT_NULL(dst.body.queue_update);
TEST_ASSERT_EQUAL_STRING("normal", dst.body.queue_update->queue);
free_ReplayEvent(&dst);
capn_free(&c);
}
/* msg 4: ReplayEvent{jobsPush} pass-through -- BASE_JOBID (this run
* never set --sim-alt-events). */
{
ReplayEvent_ptr p;
struct ReplayEvent wire;
replay_event_t dst;
TEST_ASSERT_EQUAL_INT(0, capn_init_fp(&c, f, 0));
p.p = capn_getp(capn_root(&c), 0, 1);
read_ReplayEvent(&wire, p);
memset(&dst, 0, sizeof(dst));
decode_ReplayEvent(&dst, &wire);
TEST_ASSERT_EQUAL_INT((int)ReplayEvent_body_jobsPush,
(int)dst.body_which);
TEST_ASSERT_EQUAL_UINT64(1500ULL, dst.ts);
TEST_ASSERT_NOT_NULL(dst.body.jobs_push);
TEST_ASSERT_EQUAL_INT(1, dst.body.jobs_push->count);
TEST_ASSERT_EQUAL_UINT64(BASE_JOBID,
dst.body.jobs_push->jobs[0]->jobid);
free_ReplayEvent(&dst);
capn_free(&c);
}
/* msg 5: ReplaySession -- the LIVE decision the real builtin_fcfs
* picker produced (a PEND for BASE_JOBID -- core.c's own
* sched_job_is_ready()/sched_queue_is_ready() unconditional `return 1`
* stubs make it an immediate candidate). */
{
ReplaySession_ptr p;
replay_session_t* dst = NULL;
TEST_ASSERT_EQUAL_INT(0, capn_init_fp(&c, f, 0));
p.p = capn_getp(capn_root(&c), 0, 1);
decode_ReplaySession_ptr(&dst, p);
TEST_ASSERT_NOT_NULL(dst);
TEST_ASSERT_EQUAL_UINT64(2000ULL, dst->begin_round);
TEST_ASSERT_EQUAL_INT(1, dst->n_decisions);
TEST_ASSERT_EQUAL_INT((int)ReplayDecision_body_pend,
(int)dst->decisions[0]->body_which);
TEST_ASSERT_NOT_NULL(dst->decisions[0]->body.pend);
TEST_ASSERT_EQUAL_UINT64(BASE_JOBID,
dst->decisions[0]->body.pend->jobid);
free_ReplaySession_ptr(&dst);
capn_free(&c);
}
/* msg 6: the whole-stream termination marker, ReplayEvent{endSession}
* -- pass-through of the base file's OWN terminator (Req 5 AC 20), so
* the output stream is self-terminated exactly like a normal-mode
* recording. */
{
ReplayEvent_ptr p;
struct ReplayEvent wire;
replay_event_t dst;
TEST_ASSERT_EQUAL_INT(0, capn_init_fp(&c, f, 0));
p.p = capn_getp(capn_root(&c), 0, 1);
read_ReplayEvent(&wire, p);
memset(&dst, 0, sizeof(dst));
decode_ReplayEvent(&dst, &wire);
TEST_ASSERT_EQUAL_INT((int)ReplayEvent_body_endSession,
(int)dst.body_which);
free_ReplayEvent(&dst);
capn_free(&c);
}
/* Exactly 6 messages -- nothing more to read. */
TEST_ASSERT_NOT_EQUAL(0, capn_init_fp(&c, f, 0));
fclose(f);
unlink(out_path);
printf("test_output_file_is_decodable_with_full_stream_symmetry: "
"PASS\n");
}
/* ---------------------------------------------------------------------- */
/* Test 2: --sim-output composes with --sim-alt-events (Req 5 AC 19's own */
/* "What-if comparison" use case) -- the output reflects the ALTERNATE */
/* events, not the base file's original ones. */
/* ---------------------------------------------------------------------- */
static void test_output_file_reflects_alt_events_not_base_when_both_set(void)
{
char base_path[PATH_MAX];
char alt_path[PATH_MAX];
char out_path[PATH_MAX];
FILE* base_f;
FILE* alt_f;
int tmp_fd;
sched_sim_args_t args;
xce_sched_sim_driver_t sim;
int rc;
FILE* f;
struct capn c;
base_f = write_header_and_snapshot(base_path, sizeof(base_path),
"_decode_whatif_base");
TEST_ASSERT_NOT_NULL(base_f);
append_queue_update_event(base_f, 1000ULL, "normal");
append_jobs_push_event(base_f, 1500ULL, BASE_JOBID, "normal");
append_replay_session(base_f, 2000ULL);
append_end_of_stream_marker(base_f);
fclose(base_f);
TEST_ASSERT_TRUE(snprintf(alt_path, sizeof(alt_path), "%s.alt", base_path)
< (int)sizeof(alt_path));
alt_f = fopen(alt_path, "w");
TEST_ASSERT_NOT_NULL(alt_f);
append_queue_update_event(alt_f, 3000ULL, "altqueue");
append_jobs_push_event(alt_f, 3500ULL, ALT_JOBID, "altqueue");
append_replay_session(alt_f, 4000ULL);
append_end_of_stream_marker(alt_f);
fclose(alt_f);
snprintf(out_path, sizeof(out_path), "/tmp/xce_simoutput_whatif_XXXXXX");
tmp_fd = mkstemp(out_path);
TEST_ASSERT_TRUE(tmp_fd >= 0);
close(tmp_fd);
memset(&args, 0, sizeof(args));
args.replay_path = base_path;
args.sim_start_ts = SCHED_SIM_START_TS_LATEST;
args.sim_speed = 0;
args.alt_events_path = alt_path;
args.output_path = out_path;
rc = sched_sim_driver_run(&sim, &args);
TEST_ASSERT_EQUAL_INT(0, rc);
f = fopen(out_path, "r");
TEST_ASSERT_NOT_NULL(f);
/* msg 1: ReplayHeader; msg 2: SnapshotEntry -- skipped here, already
* proven decodable by the previous test; this test's own focus is
* WHICH jobid the pass-through events and the recorded decision carry. */
TEST_ASSERT_EQUAL_INT(0, capn_init_fp(&c, f, 0));
capn_free(&c);
TEST_ASSERT_EQUAL_INT(0, capn_init_fp(&c, f, 0));
capn_free(&c);
/* msg 3: ReplayEvent{queueUpdate} -- the ALT stream's own "altqueue",
* not the base stream's "normal". */
{
ReplayEvent_ptr p;
struct ReplayEvent wire;
replay_event_t dst;
TEST_ASSERT_EQUAL_INT(0, capn_init_fp(&c, f, 0));
p.p = capn_getp(capn_root(&c), 0, 1);
read_ReplayEvent(&wire, p);
memset(&dst, 0, sizeof(dst));
decode_ReplayEvent(&dst, &wire);
TEST_ASSERT_EQUAL_INT((int)ReplayEvent_body_queueUpdate,
(int)dst.body_which);
TEST_ASSERT_EQUAL_UINT64(3000ULL, dst.ts);
TEST_ASSERT_EQUAL_STRING("altqueue", dst.body.queue_update->queue);
free_ReplayEvent(&dst);
capn_free(&c);
}
/* msg 4: ReplayEvent{jobsPush} -- ALT_JOBID, not BASE_JOBID. */
{
ReplayEvent_ptr p;
struct ReplayEvent wire;
replay_event_t dst;
TEST_ASSERT_EQUAL_INT(0, capn_init_fp(&c, f, 0));
p.p = capn_getp(capn_root(&c), 0, 1);
read_ReplayEvent(&wire, p);
memset(&dst, 0, sizeof(dst));
decode_ReplayEvent(&dst, &wire);
TEST_ASSERT_EQUAL_INT((int)ReplayEvent_body_jobsPush,
(int)dst.body_which);
TEST_ASSERT_EQUAL_UINT64(ALT_JOBID, dst.body.jobs_push->jobs[0]->jobid);
free_ReplayEvent(&dst);
capn_free(&c);
}
/* msg 5: ReplaySession -- the live decision reflects ALT_JOBID. */
{
ReplaySession_ptr p;
replay_session_t* dst = NULL;
TEST_ASSERT_EQUAL_INT(0, capn_init_fp(&c, f, 0));
p.p = capn_getp(capn_root(&c), 0, 1);
decode_ReplaySession_ptr(&dst, p);
TEST_ASSERT_NOT_NULL(dst);
TEST_ASSERT_EQUAL_UINT64(4000ULL, dst->begin_round);
TEST_ASSERT_EQUAL_INT(1, dst->n_decisions);
TEST_ASSERT_EQUAL_UINT64(ALT_JOBID,
dst->decisions[0]->body.pend->jobid);
free_ReplaySession_ptr(&dst);
capn_free(&c);
}
fclose(f);
unlink(out_path);
printf("test_output_file_reflects_alt_events_not_base_when_both_set: "
"PASS\n");
}
int main(void)
{
UNITY_BEGIN();
RUN_TEST(test_output_file_is_decodable_with_full_stream_symmetry);
RUN_TEST(test_output_file_reflects_alt_events_not_base_when_both_set);
return UNITY_END();
}

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

@ -58,8 +58,11 @@
* test_pend_decision_round_trip() already prove works.
*
* White-box #include of the same TU set test_sim_event_loop.c uses (see
* that file's own header comment) PLUS ../src/sim_verify.c (the code under
* test).
* that file's own header comment, including task 11.5's
* recording_session.c/sim_record.c -- required at link time by
* sched_sim_driver_run()'s unconditional externs regardless of whether
* --sim-output is ever set in a given test) PLUS ../src/sim_verify.c (the
* code under test).
*/
#include <errno.h>
#include <limits.h>
@ -89,8 +92,10 @@
#include "../src/plugin.c"
#include "../src/plugins/builtin_fcfs.c"
#include "../src/recording.c"
#include "../src/recording_session.c"
#include "../src/sim.c"
#include "../src/sim_loop.c"
#include "../src/sim_record.c"
#include "../src/sim_verify.c"
#include "../src/snapshot.c"
#include "../src/snapshot_load.c"