From a1cb9cba4bbe269d2be08d02ae6c656f96c5297b Mon Sep 17 00:00:00 2001 From: Rongsong Shen Date: Thu, 8 Oct 2026 15:35:20 +0000 Subject: [PATCH] feat(scheduler-sim-decision-trace): wire tracing into the simulation driver Task 5.3: session hooks around each sim session; trace-only decisions are handled first and skip verification, recording, echo and breakpoints; traced decisions are emitted after the echo gate and before the breakpoint check; reconstructed dispatches get reconstructed records; unclaimed traces are freed at both decision free sites. Co-Authored-By: Claude Opus 5.5 (1M context) --- .../scheduler-sim-decision-trace/tasks.md | 5 +- src/scheduler/include/repl.h | 16 + src/scheduler/src/repl_trace_emit.c | 29 +- src/scheduler/src/sim_loop.c | 23 +- src/scheduler/testsuite/meson.build | 30 + .../testsuite/test_sim_trace_driver.c | 647 ++++++++++++++++++ 6 files changed, 743 insertions(+), 7 deletions(-) create mode 100644 src/scheduler/testsuite/test_sim_trace_driver.c diff --git a/.kiro/specs/scheduler-sim-decision-trace/tasks.md b/.kiro/specs/scheduler-sim-decision-trace/tasks.md index 210cb18b..c2e96230 100644 --- a/.kiro/specs/scheduler-sim-decision-trace/tasks.md +++ b/.kiro/specs/scheduler-sim-decision-trace/tasks.md @@ -268,7 +268,7 @@ - the file copy byte-equal to the console output. - _Requirements: 1.6, 7.1, 7.2, 7.3, 8.1, 8.5_ -- [ ] 5. Wire the trace into the REPL and the simulation driver +- [x] 5. Wire the trace into the REPL and the simulation driver - [x] 5.1 Make every white-box test that includes REPL or simulation sources compile the trace sources, and register the trace REPL sources with the simulation isolation checks - Add the trace control, store and renderer sources to the simulation isolation library and to the simulation isolation test's scan list. - Find the tests with the design's full include-pattern search. To the 26 tests that include the REPL or simulation-loop sources, add the trace REPL sources. @@ -286,7 +286,7 @@ - Done when REPL tests pass under ASAN for help sections and alignment, command routing, the `set output` rejection and the traced-round rollup, and the updated baseline test passes. - _Requirements: 1.11, 7.3, 7.4, 8.4, 8.6, 10.4, 10.5_ -- [ ] 5.3 Wire tracing into the simulation driver's session and decision flow +- [x] 5.3 Wire tracing into the simulation driver's session and decision flow - **Session hooks:** session begin runs after a successful begin-session and before reconciliation. Session end runs after the decision stream ends. - **Drain loop:** - trace-only decisions are handled first, with no verification, recording, echo or breakpoint check; @@ -372,3 +372,4 @@ - 4.6: the session hooks and emitter are in repl_trace_emit.c: `repl_trace_session_begin/end(sim)` and `repl_trace_on_decision/on_trace_only(sim, d)`, declared in repl_internal.h. `sim_print_decision` is now exported from sim.h. Session begin stamps `session_no = rs->session_count`, which sim_loop.c increments in `xce_repl_check_begin_round` before `xce_core_begin_session`. 5.3 must call session_begin after a successful begin-session and before signal_go/reconciliation, and session_end after the decision stream ends. Emission moves the record to the store (d->trace=NULL); write errors are reported once per session. policy_name is not stamped; capture falls back to the core picker. repl_trace_emit.c is not yet in sched_sim_isolation_lib or its scan (5.1). - 5.1: all 26 tests that include repl.c/sim_loop.c now `#include` the five REPL trace sources (repl_trace_render.c, repl_trace_render_outcome.c, repl_trace_store.c, repl_trace.c, repl_trace_emit.c) and the four core trace sources, exactly once. sched_sim_isolation_lib carries all nine trace sources. test_sim_isolation.c's `test_sim_trace_repl_sources_are_isolated` scans the REPL trace files plus repl_trace_render.h; stdio is allowed there, controller symbols are not. A new REPL trace file must be added to the 26 tests, the lib and the scan. test_sim_startup.c includes only sim.c, so if sim.c starts calling trace code it needs the sources too. - 5.2: repl.h now has the interactive-only gates `xce_repl_trace_session_begin/end(sim)` and `xce_repl_check_trace/xce_repl_check_trace_only(sim, d)`; the two check gates also no-op when d->trace is NULL. 5.3 calls these from sim_loop.c. `trace` is routed with the raw line, `info trace` is in usage and help, `set output` checks `repl_trace_file_conflicts` first, and quit/EOF call `repl_trace_shutdown`. The continue-mode rollup prints a traced round on its own via `rs->trace.round_traced`. The baseline transcript changed only in the help and info-usage lines; regenerating also rewrites the .replay fixtures' wall-clock fields, so restore those. Pre-existing: repl_count_jobs (repl.c) never destroys its iterator, leaking 16 B per continue round; the rollup tests run with detect_leaks=0. +- 5.3: sim_loop.c calls the repl.h gates. Session begin runs after a successful begin-session and before reconcile/signal_go; session end runs after the drain and verify tail, before plugin_end_session. The drain loop handles TRACE first (skipping verify, record, echo and breakpoint), then echo gate, `check_trace`, `check_decision`, `rec_free`, free. The reconstructed block uses the new gate `xce_repl_trace_reconstructed` (`repl_trace_on_reconstructed` in repl_trace_emit.c). Settles the 1.7 OPEN item (both free sites). A reviewer's trace-on vs trace-off run of the contention fixture gave byte-identical --sim-output, identical verify and identical decisions. diff --git a/src/scheduler/include/repl.h b/src/scheduler/include/repl.h index 64018b49..651b7913 100644 --- a/src/scheduler/include/repl.h +++ b/src/scheduler/include/repl.h @@ -170,6 +170,10 @@ static inline void xce_repl_check_replay_complete(xce_sched_sim_driver_t* sim) * echo gate and before the breakpoint check; takes d->trace. * - xce_repl_check_trace_only: an XCE_DECISION_TRACE decision (a traced * selection with no decision), handled before anything else. + * - xce_repl_trace_reconstructed: a dispatch the simulator rebuilt from + * a recorded allocation (task 5.3, 2.5); attaches a reconstructed + * record to `d` when the session's settings want the job, right after + * the decision is built, so it is then handled like a live one. */ extern void repl_trace_session_begin(xce_sched_sim_driver_t* sim); extern void repl_trace_session_end(xce_sched_sim_driver_t* sim); @@ -177,6 +181,9 @@ extern void repl_trace_on_decision(xce_sched_sim_driver_t* sim, xce_decision_t* d); extern void repl_trace_on_trace_only(xce_sched_sim_driver_t* sim, xce_decision_t* d); +extern void repl_trace_on_reconstructed(xce_sched_sim_driver_t* sim, + xce_decision_t* d, + const xce_sched_job_t* job); static inline void xce_repl_trace_session_begin(xce_sched_sim_driver_t* sim) { @@ -200,6 +207,15 @@ static inline void xce_repl_check_trace(xce_sched_sim_driver_t* sim, } } +static inline void xce_repl_trace_reconstructed(xce_sched_sim_driver_t* sim, + xce_decision_t* d, + const xce_sched_job_t* job) +{ + if (sim->interactive) { + repl_trace_on_reconstructed(sim, d, job); + } +} + static inline void xce_repl_check_trace_only(xce_sched_sim_driver_t* sim, xce_decision_t* d) { diff --git a/src/scheduler/src/repl_trace_emit.c b/src/scheduler/src/repl_trace_emit.c index ba5f000e..02dbbda2 100644 --- a/src/scheduler/src/repl_trace_emit.c +++ b/src/scheduler/src/repl_trace_emit.c @@ -1,9 +1,9 @@ /* * repl_trace_emit.c - the trace session hooks and inline trace emission - * (scheduler-sim-decision-trace spec, task 4.6; design.md "SimIntegration", - * "ReplTraceControl", "System Flows" -> "Session lifecycle and capture" and - * "Drain-loop emission order", "Error Handling"; Requirements 1.6, 7.1, - * 7.2, 7.3, 8.1, 8.5). + * (scheduler-sim-decision-trace spec, tasks 4.6 and 5.3; design.md + * "SimIntegration", "ReplTraceControl", "System Flows" -> "Session + * lifecycle and capture" and "Drain-loop emission order", "Error + * Handling"; Requirements 1.6, 2.5, 7.1, 7.2, 7.3, 8.1, 8.5). * * Session begin makes the `next` settings current for the session and * publishes their control block to the scheduling core; session end @@ -153,3 +153,24 @@ void repl_trace_on_trace_only(xce_sched_sim_driver_t* sim, xce_decision_t* d) { emit(sim, d, false); } + +/* A dispatch the simulator rebuilt from a recorded allocation (2.5): when + * the session's settings want the job, `d` carries a reconstructed record + * and goes through the same emission as a live decision. Runs on the sim + * thread before the allocator is released (signal_go), so these traces + * precede the round's live decisions (7.3). */ +void repl_trace_on_reconstructed(xce_sched_sim_driver_t* sim, + xce_decision_t* d, const xce_sched_job_t* job) +{ + const xce_sched_trace_ctl_t* ctl; + + if ((sim == NULL) || (d == NULL) || (job == NULL) || (d->trace != NULL)) { + return; + } + ctl = sim->core.trace_ctl; + if ((ctl == NULL) || !sched_trace_ctl_wants(ctl, job->jobid)) { + return; + } + d->trace = sched_trace_rec_reconstructed(ctl, &(sim->core), job->jobid, + job->schedobj, job->allocs); +} diff --git a/src/scheduler/src/sim_loop.c b/src/scheduler/src/sim_loop.c index 7a604e45..c4b4ac1e 100644 --- a/src/scheduler/src/sim_loop.c +++ b/src/scheduler/src/sim_loop.c @@ -17,6 +17,7 @@ * note") for why this one narrow exception to * repl.h's usual opaque-`repl_state`-pointer * convention is scoped to this file alone. */ +#include "sched/trace.h" /* decision trace: sched_trace_rec_free() */ #include "sim.h" #include "utils.h" #include "xcealloc.h" /* task 2.3: make_job_account() -- Recorded-Decision @@ -1314,6 +1315,7 @@ static void sim_reconcile_unaccounted_jobs(xce_sched_sim_driver_t* sim, xce_decision_t* d = make_dispatch_decision(job->allocs); if (d != NULL) { + xce_repl_trace_reconstructed(sim, d, job); /* 2.5 */ if (record_ctx != NULL) { sched_record_session_add_decision( record_ctx, xce_core_now(&(sim->core)), d); @@ -1327,7 +1329,9 @@ static void sim_reconcile_unaccounted_jobs(xce_sched_sim_driver_t* sim, * function's own doc comment for why this second call * site matters (bug-1610). */ sim_print_decision_gate(sim, d); + xce_repl_check_trace(sim, d); /* 7.5: before the break */ xce_repl_check_decision(sim, d); + sched_trace_rec_free(d->trace); /* unclaimed only */ free(d); /* bug-1884: its alloc belongs to the job */ } } @@ -1633,6 +1637,9 @@ static int sim_driver_run_body(xce_sched_sim_driver_t* sim, free(record_ctx); continue; } + /* Decision trace (1.6): the session's settings, before + * reconciliation reports any reconstructed dispatch */ + xce_repl_trace_session_begin(sim); /* Task 3 (scheduler-sim-decision-parity): wires Recorded- * Decision Reconciliation (sim_reconcile_unaccounted_jobs(), @@ -1681,6 +1688,14 @@ static int sim_driver_run_body(xce_sched_sim_driver_t* sim, * stream ends) is reported the same way just below the loop. */ di = 0; while ((cbuf_get(&(sim->core.decisions), &d) == 0) && (d != NULL)) { + /* a traced selection with no decision (2.4): not verified, + * recorded, echoed or breakpoint-checked */ + if (d->op == XCE_DECISION_TRACE) { + xce_repl_check_trace_only(sim, d); + sched_trace_rec_free(d->trace); + free(d); + continue; + } if (sim->verify && (decoded_session != NULL)) { const replay_decision_t* expected = (di < decoded_session->n_decisions) @@ -1707,9 +1722,12 @@ static int sim_driver_run_body(xce_sched_sim_driver_t* sim, record_ctx, xce_core_now(&(sim->core)), d); } sim_print_decision_gate(sim, d); + xce_repl_check_trace(sim, d); /* 7.5: before the break */ xce_repl_check_decision(sim, d); /* bug-1884: the decision struct is this loop's; its alloc - * or pend reason belongs to the job's schedobj */ + * or pend reason belongs to the job's schedobj. An + * unclaimed trace record is freed with it. */ + sched_trace_rec_free(d->trace); free(d); } @@ -1726,6 +1744,9 @@ static int sim_driver_run_body(xce_sched_sim_driver_t* sim, free_ReplaySession_ptr(&decoded_session); + /* Decision trace: the decision stream has ended (1.6) */ + xce_repl_trace_session_end(sim); + /* bug-1681: mirror driver.c's plugin_end_session(&(sched->core)) * call site -- runs immediately before xce_core_end_session(), * same as normal mode. */ diff --git a/src/scheduler/testsuite/meson.build b/src/scheduler/testsuite/meson.build index 6e424a94..a5bc10e2 100644 --- a/src/scheduler/testsuite/meson.build +++ b/src/scheduler/testsuite/meson.build @@ -2868,3 +2868,33 @@ sched_repl_trace_wiring_test = executable( test('scheduler repl trace wiring test', sched_repl_trace_wiring_test, suite: 'scheduler', env: ['ASAN_OPTIONS=detect_leaks=0']) + +# scheduler-sim-decision-trace task 5.3 (Requirements 1.6, 2.5, 7.1-7.3, +# 7.5, 10.1, 11.3): the trace wired into the simulation driver's session and +# decision flow, driven through the real picker and allocator threads (the +# real-scheduler_srcs link technique of sched_sim_live_allocation_test). The +# wraps watch verification, recording and the breakpoint check for +# trace-only decisions, the session hook in a non-interactive run, and make +# an allocation fail on demand. +sched_sim_trace_driver_test_srcs = ['test_sim_trace_driver.c'] +foreach f : scheduler_srcs + if f != 'src/main.c' + sched_sim_trace_driver_test_srcs += ['../' + f] + endif +endforeach + +sched_sim_trace_driver_test = executable( + 'sched_sim_trace_driver_test', + sched_sim_trace_driver_test_srcs, + include_directories:[incdir], + dependencies: [scheduler_test_deps], + link_args: ['-Wl,--wrap=sim_decision_eq', + '-Wl,--wrap=sched_record_session_add_decision', + '-Wl,--wrap=xce_repl_on_decision', + '-Wl,--wrap=repl_trace_session_begin', + '-Wl,--wrap=sched_make_allocation'], + install: false) + +test('scheduler sim trace driver test', + sched_sim_trace_driver_test, suite: 'scheduler', + env: ['ASAN_OPTIONS=detect_leaks=0']) diff --git a/src/scheduler/testsuite/test_sim_trace_driver.c b/src/scheduler/testsuite/test_sim_trace_driver.c new file mode 100644 index 00000000..3fbe01ea --- /dev/null +++ b/src/scheduler/testsuite/test_sim_trace_driver.c @@ -0,0 +1,647 @@ +/* + * test_sim_trace_driver.c - the decision trace wired into the simulation + * driver's session and decision flow (scheduler-sim-decision-trace spec, + * task 5.3; design.md "SimIntegration", "System Flows" -> "Session + * lifecycle and capture" and "Drain-loop emission order", + * "DecisionTransport", "TraceCapture"; Requirements 1.6, 2.5, 7.1, 7.2, + * 7.3, 7.5, 10.1, 11.3). + * + * Links the real scheduler_srcs (minus main.c), so sched_sim_driver_run() + * runs the real picker and allocator threads and the real allocator hooks, + * like test_sim_live_allocation.c. Each scenario runs the driver in a + * forked child with a scripted stdin and a captured stdout, as in + * test_repl_breakpoints_inspection.c, so the never-joined driver threads + * and the REPL's quit exit stay out of the test process. + * + * Link-time wraps (testsuite/meson.build) watch the seams the trace must + * not reach. A violation ends the child with a distinct exit code: + * - sim_decision_eq (verify), sched_record_session_add_decision + * (recording) and xce_repl_on_decision (breakpoint check) must never see + * an XCE_DECISION_TRACE decision (exit 77); + * - in a non-interactive run, repl_trace_session_begin must never run and + * the allocator must never see published settings (exit 78, 79) (11.3); + * - sched_make_allocation fails with -EIO while g_fail_alloc is set, which + * makes the real allocator push a trace-only decision (2.4). + */ +#define _GNU_SOURCE +#include +#include +#include +#include +#include +#include +#include +#include + +#include + +#include "controller.capnp.h" +#include "recording.h" +#include "repl.h" +#include "resobj.h" +#include "scheduler.h" +#include "sim.h" +#include "utils.h" +#include "xcealloc.h" +#include "xstring.h" + +extern int write_capnp_message(FILE* f, struct capn* c); +extern int msg_root_has_shape(struct capn* c, capn_ptr* out, size_t want_words, + size_t want_ptrs); + +#define WORKER_NAME "trace_drv_worker" +#define QUEUE_NAME "normal" +#define TEST_RESREQ ((char*)"select[type==any] rusage[mem=1]") +#define JOB_A MAKE_JOBID(1, 2, 1000, 0x0, (uint64_t)41) +#define JOB_B MAKE_JOBID(1, 2, 1000, 0x0, (uint64_t)42) +#define BEGIN_ROUND ((uint64_t)2000ULL) + +void setUp(void) { return; } +void tearDown(void) { return; } + +/* ---- wraps --------------------------------------------------------------- */ + +static bool g_fail_alloc; /* sched_make_allocation returns -EIO */ +static bool g_forbid_session_hook; /* non-interactive run (11.3) */ + +static void trace_reached(const char* seam) +{ + fprintf(stderr, "trace-only decision reached %s\n", seam); + _exit(77); +} + +extern int __real_sim_decision_eq(const replay_decision_t* expected, + const xce_decision_t* actual, char** why); +int __wrap_sim_decision_eq(const replay_decision_t* expected, + const xce_decision_t* actual, char** why) +{ + if ((actual != NULL) && (actual->op == XCE_DECISION_TRACE)) { + trace_reached("verification"); + } + return __real_sim_decision_eq(expected, actual, why); +} + +extern void __real_sched_record_session_add_decision( + sched_record_session_ctx_t* ctx, uint64_t ts, xce_decision_t* d); +void __wrap_sched_record_session_add_decision(sched_record_session_ctx_t* ctx, + uint64_t ts, xce_decision_t* d) +{ + if ((d != NULL) && (d->op == XCE_DECISION_TRACE)) { + trace_reached("recording"); + } + __real_sched_record_session_add_decision(ctx, ts, d); +} + +extern void __real_xce_repl_on_decision(xce_sched_sim_driver_t* sim, + xce_decision_t* d); +void __wrap_xce_repl_on_decision(xce_sched_sim_driver_t* sim, + xce_decision_t* d) +{ + if ((d != NULL) && (d->op == XCE_DECISION_TRACE)) { + trace_reached("the breakpoint check"); + } + __real_xce_repl_on_decision(sim, d); +} + +extern void __real_repl_trace_session_begin(xce_sched_sim_driver_t* sim); +void __wrap_repl_trace_session_begin(xce_sched_sim_driver_t* sim) +{ + if (g_forbid_session_hook) { + fprintf(stderr, "non-interactive run reached the session hook\n"); + _exit(78); + } + __real_repl_trace_session_begin(sim); +} + +extern int __real_sched_make_allocation(xce_sched_core_t* core, + xce_sched_obj_t* obj); +int __wrap_sched_make_allocation(xce_sched_core_t* core, xce_sched_obj_t* obj) +{ + if (g_forbid_session_hook + && ((core->trace_ctl != NULL) || (core->trace_rec != NULL))) { + fprintf(stderr, "non-interactive run published trace settings\n"); + _exit(79); + } + if (g_fail_alloc) { + return -EIO; + } + return __real_sched_make_allocation(core, obj); +} + +/* ---- replay fixture (same encoders as test_sim_live_allocation.c) -------- */ + +static void make_tmp_dir(char* out, size_t out_len, const char* suffix) +{ + char tmpl[PATH_MAX]; + + snprintf(tmpl, sizeof(tmpl), "/tmp/sched_tracedrv%s_XXXXXX", suffix); + TEST_ASSERT_NOT_NULL(mkdtemp(tmpl)); + TEST_ASSERT_TRUE(strlen(tmpl) < out_len); + strcpy(out, tmpl); +} + +static void write_msg(FILE* f, replay_event_t* src) +{ + struct capn c; + struct capn_segment* cs; + ReplayEvent_ptr p; + + 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_cluster(FILE* f, uint64_t ts) +{ + xce_resource_t* slots = (xce_resource_t*)calloc(1, sizeof(*slots)); + xce_resource_t* res[1]; + xce_resource_data_t rwrap; + xce_worker_t* w = (xce_worker_t*)calloc(1, sizeof(*w)); + xce_val_t* val = (xce_val_t*)calloc(1, sizeof(*val)); + xce_name_val_t* nv = (xce_name_val_t*)calloc(1, sizeof(*nv)); + xce_name_val_t** nvs = (xce_name_val_t**)calloc(1, sizeof(*nvs)); + xce_worker_t* wp[1]; + xce_workers_data_t wwrap; + xce_queue_t* q = (xce_queue_t*)calloc(1, sizeof(*q)); + replay_event_t src; + + TEST_ASSERT_TRUE(slots && w && val && nv && nvs && q); + slots->name = strdup("slots"); + res[0] = slots; + memset(&rwrap, 0, sizeof(rwrap)); + rwrap.n_total = 1; + rwrap.count = 1; + rwrap.resources = res; + memset(&src, 0, sizeof(src)); + src.ts = ts; + src.body_which = ReplayEvent_body_resourcesPush; + src.body.resources_push = &rwrap; + write_msg(f, &src); + + val->which = Value_u_fvalue; + val->u.fvalue = 4.0f; + nv->name = strdup("slots"); + nv->val = val; + nvs[0] = nv; + w->name = strdup(WORKER_NAME); + w->nresvals_to_sched = 1; + w->resvals_to_sched = nvs; + wp[0] = w; + memset(&wwrap, 0, sizeof(wwrap)); + wwrap.n_total = 1; + wwrap.count = 1; + wwrap.workers = wp; + memset(&src, 0, sizeof(src)); + src.ts = ts + 1; + src.body_which = ReplayEvent_body_workersPush; + src.body.workers_push = &wwrap; + write_msg(f, &src); + + q->queue = strdup(QUEUE_NAME); + q->status = QUEUE_STAT_OPEN; + q->priority = 1; + memset(&src, 0, sizeof(src)); + src.ts = ts + 2; + src.body_which = ReplayEvent_body_queueUpdate; + src.body.queue_update = q; + write_msg(f, &src); +} + +/* One PEND job, one task of TEST_RESREQ, queued at `queue_time`. */ +static void append_job(FILE* f, uint64_t ts, uint64_t jobid, time_t queue_time) +{ + xce_job_t* j = (xce_job_t*)calloc(1, sizeof(*j)); + xce_job_t* jp[1]; + xce_jobs_data_t wrap; + replay_event_t src; + + TEST_ASSERT_NOT_NULL(j); + j->jobid = jobid; + j->state = JOB_STAT_PEND; + j->sub_time = queue_time; + j->queue_time = queue_time; + j->submit = (job_submit_data_t*)calloc(1, sizeof(*(j->submit))); + TEST_ASSERT_NOT_NULL(j->submit); + j->submit->queue = (char*)QUEUE_NAME; + j->submit->user = (char*)"alice"; + j->submit->project_name = (char*)"proj1"; + j->submit->resreq = TEST_RESREQ; + j->submit->mintasks = 1; + j->submit->maxtasks = 1; + jp[0] = j; + + memset(&wrap, 0, sizeof(wrap)); + wrap.n_total = 1; + wrap.count = 1; + wrap.jobs = jp; + memset(&src, 0, sizeof(src)); + src.ts = ts; + src.body_which = ReplayEvent_body_jobsPush; + src.body.jobs_push = &wrap; + write_msg(f, &src); +} + +static void append_to_run(FILE* f, uint64_t ts, uint64_t jobid) +{ + job_state_change_event_t jsc; + replay_event_t src; + + memset(&jsc, 0, sizeof(jsc)); + jsc.jobid = jobid; + jsc.old_state = JOB_STAT_PEND; + jsc.new_state = JOB_STAT_RUN; + memset(&src, 0, sizeof(src)); + src.ts = ts; + src.body_which = ReplayEvent_body_jobStateChange; + src.body.job_state_change = &jsc; + write_msg(f, &src); +} + +/* A session whose recorded decisions are a DISPATCH of `dispatch_job` on + * the worker (0 for none). */ +static void append_session(FILE* f, uint64_t dispatch_job) +{ + sched_dispatch_decision_t dispatch; + replay_decision_t rd; + replay_decision_t* decisions[1]; + replay_session_t session; + struct capn c; + struct capn_segment* cs; + ReplaySession_ptr sp; + + memset(&session, 0, sizeof(session)); + session.begin_round = BEGIN_ROUND; + if (dispatch_job != 0) { + xce_alloc_t* a = make_alloc((char*)"slots", NULL, 1, 1.0, -1); + + a->from = BEGIN_ROUND / 1000000000ULL; + memset(&dispatch, 0, sizeof(dispatch)); + dispatch.alloc = make_job_alloc( + dispatch_job, 1, + make_host_allocs(1, make_host_alloc((char*)WORKER_NAME, 1, + make_allocs(1, a)))); + memset(&rd, 0, sizeof(rd)); + rd.body_which = ReplayDecision_body_dispatch; + rd.body.dispatch = &dispatch; + rd.ts = BEGIN_ROUND; + decisions[0] = &rd; + session.decisions = decisions; + session.n_decisions = 1; + } + + capn_init_malloc(&c); + cs = capn_root(&c).seg; + encode_ReplaySession_ptr(cs, &sp, &session); + capn_setp(capn_root(&c), 0, sp.p); + TEST_ASSERT_EQUAL_INT(0, write_capnp_message(f, &c)); + capn_free(&c); +} + +static void append_end(FILE* f) +{ + replay_event_t src; + + memset(&src, 0, sizeof(src)); + src.body_which = ReplayEvent_body_endSession; + write_msg(f, &src); +} + +/* Header and snapshot through the real sched_record_start(), with the real + * builtin_fcfs picker registered (test_sim_live_allocation.c). */ +static FILE* start_replay(char* out_path, size_t out_len) +{ + char record_dir[PATH_MAX]; + xce_scheduler_t sched; + xce_sched_plugin_t* fcfs; + + make_tmp_dir(record_dir, sizeof(record_dir), "_r"); + memset(&sched, 0, sizeof(sched)); + mylock_init(&(sched.record_lock)); + TEST_ASSERT_EQUAL_INT(0, xce_sched_core_init(&(sched.core))); + fcfs = builtin_fcfs_new_plugin(); + TEST_ASSERT_NOT_NULL(fcfs); + TEST_ASSERT_TRUE(plugin_register_core(&(sched.core), fcfs) >= 0); + sched.core.job_pick_plugin = (char*)"builtin_fcfs"; + sched.record_dir = record_dir; + TEST_ASSERT_EQUAL_INT(0, sched_record_start(&sched, "trace.replay")); + TEST_ASSERT_TRUE(strlen(sched.record_file_path) < out_len); + strcpy(out_path, sched.record_file_path); + append_cluster(sched.record_file, 1001ULL); + return sched.record_file; +} + +/* ---- running the driver in a child --------------------------------------- */ + +typedef struct { + char dir[256]; + char replay[PATH_MAX]; + char script[PATH_MAX]; + char out[PATH_MAX]; /* captured stdout */ + char record[PATH_MAX]; /* --sim-output */ + const char* commands; /* NULL: non-interactive */ + bool verify; + bool fail_alloc; +} run_t; + +static char* slurp(const char* path) +{ + FILE* f = fopen(path, "r"); + long n; + char* buf; + + TEST_ASSERT_NOT_NULL(f); + TEST_ASSERT_EQUAL_INT(0, fseek(f, 0, SEEK_END)); + n = ftell(f); + TEST_ASSERT_TRUE(n >= 0); + rewind(f); + buf = (char*)calloc(1, (size_t)n + 1); + TEST_ASSERT_NOT_NULL(buf); + TEST_ASSERT_EQUAL_size_t((size_t)n, fread(buf, 1, (size_t)n, f)); + fclose(f); + return buf; +} + +/* Runs the driver in a child and returns its stdout; asserts exit 0. */ +static char* run(run_t* r) +{ + pid_t pid; + int status; + + make_tmp_dir(r->dir, sizeof(r->dir), "_run"); + snprintf(r->script, sizeof(r->script), "%s/script.txt", r->dir); + snprintf(r->out, sizeof(r->out), "%s/stdout.txt", r->dir); + snprintf(r->record, sizeof(r->record), "%s/out.replay", r->dir); + { + FILE* s = fopen(r->script, "w"); + + TEST_ASSERT_NOT_NULL(s); + fputs((r->commands != NULL) ? r->commands : "", s); + fclose(s); + } + + fflush(stdout); + pid = fork(); + TEST_ASSERT_TRUE(pid >= 0); + if (pid == 0) { + static xce_sched_sim_driver_t sim; /* threads are never joined */ + sched_sim_args_t args; + + if ((freopen(r->script, "r", stdin) == NULL) + || (freopen(r->out, "w", stdout) == NULL)) { + _exit(120); + } + g_fail_alloc = r->fail_alloc; + g_forbid_session_hook = (r->commands == NULL); + memset(&args, 0, sizeof(args)); + args.replay_path = r->replay; + args.sim_start_ts = SCHED_SIM_START_TS_LATEST; + args.verify = r->verify; + args.output_path = r->record; + args.interactive = (r->commands != NULL); + if (sched_sim_driver_run(&sim, &args) != 0) { + _exit(121); + } + fflush(stdout); + _exit(0); /* non-interactive runs end here; `quit` exits first */ + } + TEST_ASSERT_EQUAL_INT(pid, waitpid(pid, &status, 0)); + TEST_ASSERT_TRUE_MESSAGE(WIFEXITED(status), "the driver child crashed"); + if (WEXITSTATUS(status) != 0) { + char* out = slurp(r->out); + + fprintf(stderr, "child exit %d; stdout:\n%s\n", WEXITSTATUS(status), + out); + free(out); + } + TEST_ASSERT_EQUAL_INT_MESSAGE(0, WEXITSTATUS(status), + "77: a trace-only decision reached verify, " + "recording or the breakpoint check; 78/79: " + "a non-interactive run published trace " + "settings"); + return slurp(r->out); +} + +/* Decisions recorded per session in `path` (--sim-output). */ +static int recorded_decisions(const char* path, int* nsessions) +{ + FILE* f = fopen(path, "r"); + int total = 0; + + TEST_ASSERT_NOT_NULL(f); + *nsessions = 0; + for (;;) { + struct capn c; + capn_ptr root; + + if (capn_init_fp(&c, f, 0) != 0) { + break; + } + if (msg_root_has_shape(&c, &root, ReplaySession_word_count, + ReplaySession_pointer_count)) { + ReplaySession_ptr sp; + replay_session_t* s = NULL; + + sp.p = root; + decode_ReplaySession_ptr(&s, sp); + TEST_ASSERT_NOT_NULL(s); + total += s->n_decisions; + (*nsessions)++; + free_ReplaySession_ptr(&s); + } + capn_free(&c); + } + fclose(f); + return total; +} + +static const char* find_after(const char* hay, const char* from, + const char* needle) +{ + return strstr((from != NULL) ? from : hay, needle); +} + +static void job_marker(char* out, size_t n, const char* fmt, uint64_t jobid) +{ + char id[64]; + + TEST_ASSERT_TRUE(fmt_jobid(id, sizeof(id), jobid) >= 0); + snprintf(out, n, fmt, id); +} + +/* ---- tests ---------------------------------------------------------------- */ + +/* 2.4, design "Drain-loop emission order": a traced selection whose + * allocation ends in an internal error is a trace-only decision. It is + * printed and stored, and never verified, recorded, echoed or checked + * against a breakpoint (the wraps exit 77 otherwise). The session's + * recorded decisions stay empty and --sim-verify reports no mismatch. */ +static void test_trace_only_decision_skips_verify_record_and_breakpoints(void) +{ + run_t r; + FILE* f; + char* out; + char cmd[256]; + char hdr[96]; + int nsessions; + + memset(&r, 0, sizeof(r)); + f = start_replay(r.replay, sizeof(r.replay)); + append_job(f, 1500ULL, JOB_A, 1000); + append_session(f, 0); + append_end(f); + fclose(f); + + snprintf(cmd, sizeof(cmd), + "trace on all\nbreak DECISION\ncontinue\ninfo trace %llu\nquit\n", + (unsigned long long)JOB_A); + r.commands = cmd; + r.verify = true; + r.fail_alloc = true; + out = run(&r); + + job_marker(hdr, sizeof(hdr), "[trace] job %s session", JOB_A); + TEST_ASSERT_NOT_NULL_MESSAGE(strstr(out, hdr), + "the no-decision trace must be printed"); + TEST_ASSERT_NOT_NULL(strstr(out, "no decision: allocation attempt ended " + "in an internal error")); + TEST_ASSERT_NULL_MESSAGE(strstr(out, "hit breakpoint"), + "a trace-only decision never hits a breakpoint"); + TEST_ASSERT_NULL_MESSAGE(strstr(out, "MISMATCH"), + "a trace-only decision is never verified"); + TEST_ASSERT_NULL(strstr(out, "jobid=")); /* no echo line */ + /* stored: `info trace` prints it a second time */ + TEST_ASSERT_NOT_NULL_MESSAGE(strstr(strstr(out, hdr) + 1, hdr), + "the trace-only record must be stored"); + + TEST_ASSERT_EQUAL_INT(0, recorded_decisions(r.record, &nsessions)); + TEST_ASSERT_EQUAL_INT(1, nsessions); + free(out); +} + +/* 7.1, 7.5: a decision breakpoint stops after that decision's echo line + * and trace, with the trace stored; the round's later decision has been + * neither printed nor stored when the breakpoint stops. */ +static void test_decision_breakpoint_stops_after_its_trace(void) +{ + run_t r; + FILE* f; + char* out; + char cmd[256]; + char echo_a[96], hdr_a[96], hdr_b[96], none_b[128]; + const char *p_echo, *p_hdr, *p_bp, *p_info, *p_none; + + memset(&r, 0, sizeof(r)); + f = start_replay(r.replay, sizeof(r.replay)); + append_job(f, 1500ULL, JOB_A, 1000); + append_job(f, 1501ULL, JOB_B, 1001); + append_session(f, 0); + append_end(f); + fclose(f); + + snprintf(cmd, sizeof(cmd), + "trace on all\nbreak DECISION type=DISPATCH\nstep\n" + "info trace %llu\ninfo trace %llu\nquit\n", + (unsigned long long)JOB_A, (unsigned long long)JOB_B); + r.commands = cmd; + out = run(&r); + + job_marker(echo_a, sizeof(echo_a), "[sim] DISPATCH jobid=%s", JOB_A); + job_marker(hdr_a, sizeof(hdr_a), "[trace] job %s session", JOB_A); + job_marker(hdr_b, sizeof(hdr_b), "[trace] job %s session", JOB_B); + job_marker(none_b, sizeof(none_b), + "[sim] info trace: job %s: no trace available", JOB_B); + + p_echo = strstr(out, echo_a); + p_hdr = find_after(out, p_echo, hdr_a); + p_bp = find_after(out, p_hdr, "hit breakpoint 1 @ decision: DISPATCH"); + p_info = find_after(out, p_bp, hdr_a); + p_none = find_after(out, p_info, none_b); + TEST_ASSERT_NOT_NULL_MESSAGE(p_echo, "the first decision is echoed"); + TEST_ASSERT_NOT_NULL_MESSAGE(p_hdr, "its trace follows its echo line"); + TEST_ASSERT_NOT_NULL_MESSAGE(p_bp, "the breakpoint stops after it"); + TEST_ASSERT_NOT_NULL_MESSAGE(p_info, "`info trace` shows it at the stop"); + TEST_ASSERT_NOT_NULL_MESSAGE(p_none, "no later trace is stored yet"); + TEST_ASSERT_NULL_MESSAGE(strstr(out, hdr_b), + "the later decision's trace is never printed " + "before the breakpoint"); + free(out); +} + +/* 2.5, 7.3: a dispatch rebuilt from the recording is traced as a + * reconstructed record, echoed and printed before the round's live + * decisions. */ +static void test_reconstructed_trace_precedes_live_decisions(void) +{ + run_t r; + FILE* f; + char* out; + char echo_a[96], hdr_a[96], hdr_b[96]; + const char *p_echo, *p_hdr, *p_rec, *p_live; + + memset(&r, 0, sizeof(r)); + f = start_replay(r.replay, sizeof(r.replay)); + append_job(f, 1500ULL, JOB_A, 1000); + append_to_run(f, 1600ULL, JOB_A); + append_job(f, 1700ULL, JOB_B, 1001); + append_session(f, JOB_A); + append_end(f); + fclose(f); + + r.commands = "trace on all\nstep\nquit\n"; + out = run(&r); + + job_marker(echo_a, sizeof(echo_a), "[sim] DISPATCH jobid=%s", JOB_A); + job_marker(hdr_a, sizeof(hdr_a), "[trace] job %s session", JOB_A); + job_marker(hdr_b, sizeof(hdr_b), "[trace] job %s session", JOB_B); + + p_echo = strstr(out, echo_a); + p_hdr = find_after(out, p_echo, hdr_a); + p_rec = find_after(out, p_hdr, "dispatch reconstructed from the " + "recorded allocation"); + p_live = strstr(out, hdr_b); + TEST_ASSERT_NOT_NULL_MESSAGE(p_echo, "the reconstructed dispatch is " + "echoed"); + TEST_ASSERT_NOT_NULL_MESSAGE(p_hdr, "and traced after its echo line"); + TEST_ASSERT_NOT_NULL_MESSAGE(p_rec, "as a reconstructed record"); + TEST_ASSERT_NOT_NULL_MESSAGE(p_live, "the live decision is traced"); + TEST_ASSERT_TRUE_MESSAGE(p_rec < p_live, "the reconstructed trace comes " + "before the live decisions"); + free(out); +} + +/* 11.3: a simulation without the REPL never runs the session hook and the + * allocator never sees settings (wraps exit 78/79), while the jobs are + * still decided as usual. */ +static void test_non_interactive_run_never_publishes_settings(void) +{ + run_t r; + FILE* f; + char* out; + int nsessions; + + memset(&r, 0, sizeof(r)); + f = start_replay(r.replay, sizeof(r.replay)); + append_job(f, 1500ULL, JOB_A, 1000); + append_job(f, 1501ULL, JOB_B, 1001); + append_session(f, 0); + append_end(f); + fclose(f); + + out = run(&r); /* commands NULL: non-interactive */ + TEST_ASSERT_NULL(strstr(out, "[trace]")); + TEST_ASSERT_EQUAL_INT(2, recorded_decisions(r.record, &nsessions)); + free(out); +} + +int main(void) +{ + UNITY_BEGIN(); + RUN_TEST(test_trace_only_decision_skips_verify_record_and_breakpoints); + RUN_TEST(test_decision_breakpoint_stops_after_its_trace); + RUN_TEST(test_reconstructed_trace_precedes_live_decisions); + RUN_TEST(test_non_interactive_run_never_publishes_settings); + return UNITY_END(); +}