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

feat(scheduler-sim-decision-trace): wire tracing into the simulation driver

Browse source 
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) <noreply@anthropic.com>
This commit is contained in:
Rongsong Shen 2026-10-08 15:35:20 +00:00
parent 6bac16812c
commit a1cb9cba4b
Signed by untrusted user: rshen
GPG key ID: 80533536B0EB651C
6 changed files with 743 additions and 7 deletions
Download patch file Download diff file Expand all files Collapse all files

5
.kiro/specs/scheduler-sim-decision-trace/tasks.md
View file

@ -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.

16
src/scheduler/include/repl.h
View file

@ -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)
{

29
src/scheduler/src/repl_trace_emit.c
View file

@ -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);
}

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

@ -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. */

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

@ -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'])

647
src/scheduler/testsuite/test_sim_trace_driver.c Normal file
View file

@ -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 <errno.h>
#include <limits.h>
#include <stdint.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/wait.h>
#include <unistd.h>
#include <unity/unity.h>
#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();
}