From 3a6b503f07d42341860482bec407498cae580e9c Mon Sep 17 00:00:00 2001 From: Rongsong Shen Date: Tue, 29 Sep 2026 04:41:14 +0000 Subject: [PATCH] feat(scheduler-sim-discrete-clock): pin the simulation clock so it moves only at recorded timestamps (task 2.1) Co-Authored-By: Claude Opus 5.5 (1M context) --- .../scheduler-sim-discrete-clock/tasks.md | 3 +- src/scheduler/src/sim.c | 33 +- src/scheduler/src/sim_loop.c | 12 +- src/scheduler/src/sim_record.c | 8 +- .../testsuite/test_sim_discrete_clock.c | 828 ++++++++++++++++++ 5 files changed, 865 insertions(+), 19 deletions(-) diff --git a/.kiro/specs/scheduler-sim-discrete-clock/tasks.md b/.kiro/specs/scheduler-sim-discrete-clock/tasks.md index 2c6cc23d..77d812c5 100644 --- a/.kiro/specs/scheduler-sim-discrete-clock/tasks.md +++ b/.kiro/specs/scheduler-sim-discrete-clock/tasks.md @@ -12,7 +12,7 @@ Test-first throughout (red → green → refactor). Every sub-task leaves the bu - _Boundary: CliCompat, scheduler test build_ - [ ] 2. Core: pin the simulator clock, remove the REPL speed parameter, retire the CLI option -- [ ] 2.1 (P) Pin the simulation-mode clock so it moves only at recorded timestamps +- [x] 2.1 (P) Pin the simulation-mode clock so it moves only at recorded timestamps - Write the tests first in the new test binary, each passing a non-zero speed through the still-present simulator argument so they fail on the current code: after start the clock equals the chosen snapshot's timestamp and stays equal after a real sleep, with the core holding the frozen speed setting; with a stand-in allocator that sleeps and a recorded session start that is not a whole second, the session start seen inside the session equals the recorded one exactly, and simulation output has that session start, a zero session duration and every decision timestamp equal to it, with at least one decision per session; a session start earlier than the preceding event, and events earlier than the start snapshot, set the clock to exactly those values and the replay completes; a mid-replay snapshot followed by the end marker leaves the clock at the last event's timestamp after the run and after a sleep, with the start timestamp pinned so the real skip branch runs; the same recording replayed twice gives identical logical timestamps and decisions in both outputs, ignoring only the wall-clock header field (design.md, Testing Strategy: Unit (SimClock), Integration (replay loop)). - Make simulator startup always select the frozen clock setting and drop the speed parameter from the start-clock function. The start-clock function keeps writing the shadow copy, now always the frozen setting, because `info config` and the startup test still read it; the simulator's speed argument is no longer read. Both are removed in 3.1. - Update the speed-drift and never-runs-backwards comments in the simulator driver, replay loop and simulation-output sources; the simulator header's comments are left to 3.1 (design.md, Modified Files). @@ -68,3 +68,4 @@ Test-first throughout (red → green → refactor). Every sub-task leaves the bu ## Implementation Notes - Build dir: `bash src/tools/xce-build.sh scheduler asan xce.build.discrete` builds into `src/xce.build.discrete`; iterate with `meson test -C src/xce.build.discrete --suite scheduler` and grep its testlog for `:FAIL`. +- 2.1 added `args->sim_speed = DISCRETE_CLOCK_RED_SPEED` (and that define) to `test_sim_discrete_clock.c` to prove RED; 3.1 removes both. Its fixture writer leaves mkdtemp dirs under `/tmp` (as `test_sim_record_output.c` does). diff --git a/src/scheduler/src/sim.c b/src/scheduler/src/sim.c index 3c8d05d4..5daec7e1 100644 --- a/src/scheduler/src/sim.c +++ b/src/scheduler/src/sim.c @@ -224,12 +224,14 @@ static int sim_locate_and_load_snapshot(xce_sched_sim_driver_t* sim, decode_SnapshotEntry(&decoded, &entry_wire); capn_free(&c); - /* SnapshotEntry timestamps are monotonically non-decreasing - * across a well-formed `.replay` stream (both normal-mode - * recording and simulation --sim-output derive ts from - * xce_core_now(), which never runs backwards) -- so any - * qualifying candidate found later in the scan is always at - * least as recent as the current best. Freeing the OLD best + /* The LAST qualifying SnapshotEntry in file order wins + * (Req 1.1 of scheduler-sim-discrete-clock). Timestamps are + * normally non-decreasing across the stream, but not + * guaranteed: in simulation mode xce_core_now() is set exactly + * to each recorded timestamp, which may move it backwards (Req + * 1.3), so a --sim-output snapshot can carry an earlier ts than + * the one before it. Either way, a later candidate replaces + * the current best. Freeing the OLD best * here is always safe: xce_sched_core_load_snapshot() is only * ever called once, below, on whichever candidate survives to * the end of this loop -- an intermediate candidate that gets @@ -305,9 +307,19 @@ static int sim_locate_and_load_snapshot(xce_sched_sim_driver_t* sim, * boundary (test_core_isolation.c) -- that boundary is scoped to core.c * itself, not to driver-level code legitimately seeding the core's clock * anchors, exactly as AC 13's own pseudocode specifies. + * + * scheduler-sim-discrete-clock (Req 1.1, 1.5): simulation mode always runs + * the core clock frozen -- speed_num = 0, speed_den = 1 -- so + * xce_core_now() returns exactly snapshot_ts, and afterwards exactly the + * last value sim_advance_virtual_clock() (sim_loop.c) set, however much + * real time passes. This is the only place simulation mode sets the speed; + * the --sim-speed value (sched_sim_args_t.sim_speed) is no longer read. + * The sim_speed_num/sim_speed_den shadow copy is still written (always + * 0/1) for `info config` and test_sim_startup.c; the shadow copy and + * sched_sim_args_t.sim_speed are removed in task 3.1. */ static void sim_set_virtual_clock(xce_sched_sim_driver_t* sim, - uint64_t snapshot_ts, uint64_t sim_speed) + uint64_t snapshot_ts) { struct timespec ts; uint64_t real_clock_base_ns; @@ -320,11 +332,10 @@ static void sim_set_virtual_clock(xce_sched_sim_driver_t* sim, memory_order_release); atomic_store_explicit(&(sim->core.real_clock_base), real_clock_base_ns, memory_order_release); - atomic_store_explicit(&(sim->core.speed_num), sim_speed, - memory_order_release); + atomic_store_explicit(&(sim->core.speed_num), 0, memory_order_release); atomic_store_explicit(&(sim->core.speed_den), 1, memory_order_release); - sim->sim_speed_num = sim_speed; + sim->sim_speed_num = 0; sim->sim_speed_den = 1; } @@ -591,7 +602,7 @@ int sched_sim_driver_start(xce_sched_sim_driver_t* sim, return rc; } - sim_set_virtual_clock(sim, snapshot_ts, args->sim_speed); + sim_set_virtual_clock(sim, snapshot_ts); /* Orientation for an operator about to `step`/`continue` through this * file (or watch a batch --sim run go by): how many sessions are there diff --git a/src/scheduler/src/sim_loop.c b/src/scheduler/src/sim_loop.c index 749643e8..74c4c657 100644 --- a/src/scheduler/src/sim_loop.c +++ b/src/scheduler/src/sim_loop.c @@ -174,10 +174,14 @@ extern int sched_record_session_finalize(FILE* f, /* * sim_advance_virtual_clock - Req 5 AC 14 (task 11.2): re-anchors * sim->core's virtual clock so xce_core_now(&sim->core) reports exactly - * new_virtual_ts at the instant this call returns (elapsed=0 against a - * freshly read real_clock_base), then evolves from there per whatever - * speed_num/speed_den sched_sim_driver_start() already set (task 4.5's - * two-anchor + speed formula, core_clock.c) -- speed settings are left + * new_virtual_ts. scheduler-sim-discrete-clock (Req 1.2, 1.3, 1.5): the + * set is exact and unconditional -- a new_virtual_ts earlier than the + * current clock (e.g. a recorder-stamped event batch older than the start + * snapshot, or out-of-order --sim-alt-events) is set as is, with no + * clamping, warning or error -- and, because sched_sim_driver_start() + * always pins speed_num/speed_den to 0/1 (sim.c's + * sim_set_virtual_clock()), the clock then stays at exactly that value, + * with no real-time drift, until the next call. Speed settings are left * untouched here, only the two clock anchors jump. Same driver-level * clock_gettime() legitimacy as sim.c's own sim_set_virtual_clock() (task * 11.1, Step 4): core.c's "zero wall-clock calls" isolation boundary diff --git a/src/scheduler/src/sim_record.c b/src/scheduler/src/sim_record.c index 8cdc0479..92912df8 100644 --- a/src/scheduler/src/sim_record.c +++ b/src/scheduler/src/sim_record.c @@ -84,9 +84,11 @@ extern int write_snapshot_entry(FILE* f, xce_sched_core_t* core, uint64_t ts); * 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. + * deliberately NOT re-derived via xce_core_now() here: this message + * records the loaded snapshot's own timestamp, which must not depend on a + * clock read. (In simulation mode the clock moves only at recorded + * timestamps -- scheduler-sim-discrete-clock Req 1 -- and no recorded + * event has been read yet, so the two values are equal today.) * * configHash is left unset, same documented no-op limitation * write_replay_header() itself already carries for this field (no utility diff --git a/src/scheduler/testsuite/test_sim_discrete_clock.c b/src/scheduler/testsuite/test_sim_discrete_clock.c index 7e55dbcd..e2a5115a 100644 --- a/src/scheduler/testsuite/test_sim_discrete_clock.c +++ b/src/scheduler/testsuite/test_sim_discrete_clock.c @@ -9,6 +9,22 @@ * non-numeric, negative, empty), and logs nothing when the option was not * given (NULL). * + * Task 2.1 (Requirements 1.1-1.8, 4.2; design.md, Components: SimClock, + * Testing Strategy: Unit (SimClock) and Integration (replay loop)): in + * simulation mode the logical clock moves only at the start snapshot, at + * each recorded event and at each recorded session, is set exactly (never + * clamped, even backwards), and never drifts with real time. Every clock + * test passes a non-zero speed through the still-present + * sched_sim_args_t.sim_speed (DISCRETE_CLOCK_RED_SPEED) to prove the + * simulator no longer reads it; task 3.1 removes that field and these + * assignments while keeping the pinned-clock assertions. + * + * White-box #include of the same simulator TU set test_sim_record_output.c + * uses, with its link-satisfying stubs and stand-in picker; the stand-in + * allocator here additionally records core->session_start on every + * cbuf_get() return (including the jobid==0 terminator) and can sleep + * before deciding, so real time visibly passes inside a session. + * * This binary deliberately never calls mylog_init(): without it, tlog's * early-print path writes WARN lines to stdout, which the test-local * capture_stdout_around() below captures. @@ -18,6 +34,10 @@ * later tasks add to this file. */ +#include +#include +#include +#include #include #include #include @@ -28,11 +48,292 @@ #include #include "cli_compat.h" +#include "controller.capnp.h" +#include "kvmgr.h" +#include "scheduler.h" +#include "xstring.h" + +/* White-box #include of the simulator under test (same set and order as + * test_sim_record_output.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/repl.c" +#include "../src/repl_fmt.c" +#include "../src/repl_rollup.c" +#include "../src/repl_break.c" +#include "../src/repl_info.c" +#include "../src/repl_params.c" +#include "../src/sim.c" +#include "../src/sim_loop.c" +#include "../src/sim_picker.c" +#include "../src/sim_limit.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" #if !defined(XCE_SCHEDULER_BIN) #error "XCE_SCHEDULER_BIN must be defined by testsuite/meson.build" #endif +/* Link-satisfying stubs and the stand-in picker below are copied from + * test_sim_record_output.c, whose comments give the rationale for each. */ +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; } + +var limit_get_configured_limits(xce_sched_core_t* core) +{ + (void)core; + return NULL; +} + +var limit_get_usage_instances(xce_sched_core_t* core) +{ + (void)core; + return NULL; +} + +xce_sched_plugin_t* limit_new_plugin(void) +{ + return NULL; +} + +int limit_seed_config(xce_sched_plugin_t* plugin, xce_sched_core_t* core, + xce_conf4sched_t* config) +{ + (void)plugin; + (void)core; + (void)config; + return 0; +} + +int sched_reconcile_job_resource_usage(xce_sched_core_t* c, + xce_job_account_t* a, + xce_job_rusages_t* r) +{ + (void)c; (void)a; (void)r; return 0; +} + +void sched_worker_release_host_ctx(xce_sched_worker_t* w) { (void)w; } + +int sched_fill_limit_requests(xce_sched_alloc_req_t* alloc_req, + xlist_t* limits, xlist_t* rusages, int nhosts, + int nslots) +{ + (void)alloc_req; + (void)limits; + (void)rusages; + (void)nhosts; + (void)nslots; + return 0; +} + +int sched_account_jobs(xce_sched_core_t* core) +{ + (void)core; + return 0; +} + +int sched_check_host_satisfy(xce_sched_core_t* core, xce_sched_worker_t* w, + xvec_t* uncomitted, xvec_idval_t* wanted, + xvec_idval_t* unmeet) +{ + (void)core; + (void)w; + (void)uncomitted; + (void)wanted; + (void)unmeet; + return 1; +} + +int sched_check_shared_satisfy(xce_sched_core_t* core, xvec_t* uncomitted, + xvec_idval_t* wanted, xvec_idval_t* unmeet) +{ + (void)core; + (void)uncomitted; + (void)wanted; + (void)unmeet; + return 1; +} + +int sched_charge_account(xce_sched_core_t* core, xce_job_account_t* acct) +{ + (void)core; + (void)acct; + return 0; +} + +int sched_uncharge_account(xce_sched_core_t* core, xce_job_account_t* acct) +{ + (void)core; + (void)acct; + 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* sched_job_picker(void* arg) +{ + xce_sched_core_t* core = (xce_sched_core_t*)arg; + xce_sched_plugin_t* picker = NULL; + + for (;;) { + int go = 0; + uint64_t jobid = 0; + + cbuf_get(&(core->wbuf), &go); + + if (sched_resolve_job_picker(core, &picker) < 0) { + jobid = 0; + cbuf_put(&(core->jbuf), &jobid); + continue; + } + + picker->picker_iface->begin_session(picker); + + for (;;) { + xce_sched_job_t* job = picker->picker_iface->next_job(picker); + + if (job == NULL) { + break; + } + + jobid = job->jobid; + cbuf_put(&(core->jbuf), &jobid); + } + + jobid = 0; + cbuf_put(&(core->jbuf), &jobid); + + picker->picker_iface->end_session(picker); + } + + return NULL; +} + +/* Stand-in allocator (design.md, Testing Strategy: Integration): same + * PEND-only contract as test_sim_record_output.c's, plus two observation + * hooks for the core under observation (g_obs_core): every cbuf_get() + * return -- including the jobid==0 end-of-session terminator, so a session + * with no pending job is still observed -- records core->session_start + * and the jobid, and a real job decision is preceded by a real sleep + * (g_alloc_sleep_us) so wall-clock time visibly passes inside a session. + * Other cores (threads left parked by earlier runs, never joined) are + * served but not recorded. The main thread reads the records only after + * sched_sim_driver_run() returns, which it cannot do before this thread's + * last cbuf_put(&core->decisions) -- the cbuf's mutex orders the writes. */ +#define MAX_OBS 64 + +static xce_sched_core_t* _Atomic g_obs_core; +static _Atomic unsigned g_alloc_sleep_us; +static int g_obs_n; +static uint64_t g_obs_session_start[MAX_OBS]; +static uint64_t g_obs_jobid[MAX_OBS]; + +void* sched_allocator(void* arg) +{ + xce_sched_core_t* core = (xce_sched_core_t*)arg; + xce_sched_plugin_t* picker = NULL; + + for (;;) { + uint64_t jobid = 0; + xce_decision_t* decision = NULL; + xce_sched_job_t* job; + bool observed; + + cbuf_get(&(core->jbuf), &jobid); + + observed = (atomic_load(&g_obs_core) == core); + if (observed && (g_obs_n < MAX_OBS)) { + g_obs_session_start[g_obs_n] = core->session_start; + g_obs_jobid[g_obs_n] = jobid; + g_obs_n++; + } + + if (jobid == 0) { + decision = NULL; + cbuf_put(&(core->decisions), &decision); + continue; + } + + job = sched_get_job(core, jobid); + if (job == NULL) { + continue; + } + + if (observed && (atomic_load(&g_alloc_sleep_us) > 0)) { + usleep(atomic_load(&g_alloc_sleep_us)); + } + + (void)sched_resolve_job_picker(core, &picker); + if (picker != NULL) { + picker->picker_iface->on_pend(picker, job->jobid); + } + + decision = make_pend_decision(make_pendreason(job->jobid)); + cbuf_put(&(core->decisions), &decision); + } + + return NULL; +} + +void sim_bootstrap_seqjob_allocator(xce_sched_sim_driver_t* sim) +{ + (void)sim; +} + +void sim_bootstrap_parjob_allocator(xce_sched_sim_driver_t* sim) +{ + (void)sim; +} + void setUp(void) { return; } void tearDown(void) { return; } @@ -149,6 +450,527 @@ static void test_no_warning_when_option_absent(void) free(out); } +/* ---------------------------------------------------------------------- */ +/* SimClock fixtures (Requirements 1.1-1.8, 4.2). Fixture writers copied */ +/* from test_sim_record_output.c; the fixture core's clock is frozen at a */ +/* chosen value first, so the recorded start snapshot's ts is exact. */ +/* ---------------------------------------------------------------------- */ + +/* Task 2.1 RED: a non-zero speed through the still-present simulator + * argument. The simulator must ignore it; removed with the field in 3.1. */ +#define DISCRETE_CLOCK_RED_SPEED 7 + +#define SNAP_TS 1000000000000ULL /* start snapshot, 1000 s */ +#define JOB_A 4101ULL +#define JOB_B 4102ULL +#define SLEEP_US 12000U /* stand-in allocator sleep per decision */ +#define WAIT_US 25000U /* real time allowed to pass between reads */ + +static FILE* write_header_and_snapshot(char* out_path, size_t out_path_len, + uint64_t snap_ts) +{ + char tmpl[] = "/tmp/sched_discrete_clock_XXXXXX"; + xce_scheduler_t sched; + + TEST_ASSERT_NOT_NULL(mkdtemp(tmpl)); + memset(&sched, 0, sizeof(sched)); + mylock_init(&(sched.record_lock)); + TEST_ASSERT_EQUAL_INT(0, xce_sched_core_init(&(sched.core))); + sched.core.job_pick_plugin = (char*)"builtin_fcfs"; + sched.record_dir = tmpl; + atomic_store(&(sched.core.virtual_clock_base), snap_ts); + atomic_store(&(sched.core.speed_num), 0); + atomic_store(&(sched.core.speed_den), 1); + + TEST_ASSERT_EQUAL_INT(0, sched_record_start(&sched, "fixture.replay")); + TEST_ASSERT_EQUAL_UINT64(snap_ts, sched.record_start_ts); + TEST_ASSERT_TRUE(strlen(sched.record_file_path) < out_path_len); + strcpy(out_path, sched.record_file_path); + return sched.record_file; +} + +static void write_event_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_queue_update_event(FILE* f, uint64_t ts, + const char* queue_name) +{ + xce_queue_t queue; + replay_event_t src; + + 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; + write_event_msg(f, &src); +} + +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; + + 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; + write_event_msg(f, &src); +} + +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); +} + +/* endSession marker with ts 0, as test_sim_record_output.c writes it. */ +static void append_end_of_stream_marker(FILE* f) +{ + replay_event_t src; + + memset(&src, 0, sizeof(src)); + src.body_which = ReplayEvent_body_endSession; + write_event_msg(f, &src); +} + +/* Hand-built mid-replay SnapshotEntry (test_sim_startup.c's technique). */ +static void append_snapshot_entry(FILE* f, uint64_t ts) +{ + struct capn c; + struct capn_segment* cs; + SnapshotData_ptr sdata; + SnapshotEntry_ptr snap; + + capn_init_malloc(&c); + cs = capn_root(&c).seg; + sdata = new_SnapshotData(cs); + snap = new_SnapshotEntry(cs); + SnapshotEntry_set_ts(snap, ts); + SnapshotEntry_set_data(snap, sdata); + capn_setp(capn_root(&c), 0, snap.p); + TEST_ASSERT_EQUAL_INT(0, write_capnp_message(f, &c)); + capn_free(&c); +} + +static void make_tmp_file(char* out, size_t out_len, const char* tag) +{ + int fd; + + snprintf(out, out_len, "/tmp/xce_discrete_clock_%s_XXXXXX", tag); + fd = mkstemp(out); + TEST_ASSERT_TRUE(fd >= 0); + close(fd); +} + +static void sim_args_init(sched_sim_args_t* args, const char* replay_path) +{ + memset(args, 0, sizeof(*args)); + args->replay_path = replay_path; + args->sim_start_ts = SCHED_SIM_START_TS_LATEST; + args->sim_speed = DISCRETE_CLOCK_RED_SPEED; +} + +/* Runs a batch replay with `sim` as the observed core. */ +static int run_observed(xce_sched_sim_driver_t* sim, + const sched_sim_args_t* args, unsigned sleep_us) +{ + g_obs_n = 0; + atomic_store(&g_alloc_sleep_us, sleep_us); + atomic_store(&g_obs_core, &(sim->core)); + return sched_sim_driver_run(sim, args); +} + +/* Asserts the stand-in allocator's records split (at each jobid==0 + * terminator) into exactly n sessions, the k-th seeing session_start == + * begin[k]. */ +static void assert_observed_session_starts(const uint64_t* begin, int n) +{ + int k = 0; + int i; + + for (i = 0; i < g_obs_n; i++) { + TEST_ASSERT_TRUE_MESSAGE(k < n, "more sessions observed than " + "recorded"); + TEST_ASSERT_EQUAL_UINT64_MESSAGE(begin[k], g_obs_session_start[i], + "session_start must equal the " + "recorded beginRound exactly"); + if (g_obs_jobid[i] == 0) { + k++; + } + } + TEST_ASSERT_EQUAL_INT_MESSAGE(n, k, "every recorded session must be " + "observed"); +} + +/* Fixture shared by the output and determinism tests: two sessions whose + * beginRound is not a whole second, each with a pending job before it. */ +#define OUT_B1 (SNAP_TS + 1234567891ULL) +#define OUT_B2 (OUT_B1 + 2000000333ULL) + +static void write_two_session_fixture(char* path, size_t path_len) +{ + FILE* f = write_header_and_snapshot(path, path_len, SNAP_TS); + + append_queue_update_event(f, SNAP_TS + 100ULL, "normal"); + append_jobs_push_event(f, SNAP_TS + 200ULL, JOB_A, "normal"); + append_replay_session(f, OUT_B1); + append_jobs_push_event(f, OUT_B1 + 1000ULL, JOB_B, "normal"); + append_replay_session(f, OUT_B2); + append_end_of_stream_marker(f); + fclose(f); +} + +/* ---------------------------------------------------------------------- */ +/* SimClock unit tests (Requirements 1.1, 1.3, 1.5) */ +/* ---------------------------------------------------------------------- */ + +/* 1.1, 1.5: after start the clock is the start snapshot's ts, and stays + * there however much real time passes, with the core at speed 0/1. */ +static void test_start_pins_clock_to_snapshot_ts(void) +{ + static xce_sched_sim_driver_t sim; + char path[PATH_MAX]; + sched_sim_args_t args; + FILE* f; + + f = write_header_and_snapshot(path, sizeof(path), SNAP_TS); + append_end_of_stream_marker(f); + fclose(f); + + sim_args_init(&args, path); + TEST_ASSERT_EQUAL_INT(0, sched_sim_driver_start(&sim, &args)); + + TEST_ASSERT_EQUAL_UINT64(SNAP_TS, xce_core_now(&(sim.core))); + usleep(WAIT_US); + TEST_ASSERT_EQUAL_UINT64_MESSAGE(SNAP_TS, xce_core_now(&(sim.core)), + "clock must not drift with real time"); + TEST_ASSERT_EQUAL_UINT64(0ULL, atomic_load(&(sim.core.speed_num))); + TEST_ASSERT_EQUAL_UINT64(1ULL, atomic_load(&(sim.core.speed_den))); +} + +/* 1.3 (session case): a session whose beginRound is earlier than the + * preceding event's ts starts at exactly that beginRound; the replay + * completes. */ +static void test_session_earlier_than_preceding_event_sets_exact(void) +{ + static xce_sched_sim_driver_t sim; + char path[PATH_MAX]; + sched_sim_args_t args; + FILE* f; + const uint64_t b1 = SNAP_TS + 1500000001ULL; + const uint64_t e = b1 + 900000000ULL; + const uint64_t b2 = b1 + 500000007ULL; /* < e */ + const uint64_t begin[2] = { b1, b2 }; + + f = write_header_and_snapshot(path, sizeof(path), SNAP_TS); + append_queue_update_event(f, SNAP_TS + 100ULL, "normal"); + append_jobs_push_event(f, SNAP_TS + 200ULL, JOB_A, "normal"); + append_replay_session(f, b1); + append_queue_update_event(f, e, "padq"); /* inert padding event */ + append_replay_session(f, b2); + append_end_of_stream_marker(f); + fclose(f); + + sim_args_init(&args, path); + TEST_ASSERT_EQUAL_INT(0, run_observed(&sim, &args, SLEEP_US)); + assert_observed_session_starts(begin, 2); + TEST_ASSERT_EQUAL_UINT64(b2, xce_core_now(&(sim.core))); +} + +/* 1.3 (event case, as the normal-mode recorder produces): events whose ts + * is earlier than the start snapshot set the clock to exactly that ts -- + * observed through the static sim_process_replay_event_message() -- and a + * full replay of the same file completes. */ +static void test_events_earlier_than_start_snapshot_set_exact(void) +{ + static xce_sched_sim_driver_t start_sim; + static xce_sched_sim_driver_t run_sim; + char path[PATH_MAX]; + sched_sim_args_t args; + FILE* f; + const uint64_t ev_ts[2] = { SNAP_TS - 5000ULL, + SNAP_TS - 3000ULL }; + const uint64_t begin[1] = { SNAP_TS + 1000000017ULL }; + int i; + + f = write_header_and_snapshot(path, sizeof(path), SNAP_TS); + append_queue_update_event(f, ev_ts[0], "normal"); + append_jobs_push_event(f, ev_ts[1], JOB_A, "normal"); + append_replay_session(f, begin[0]); + append_end_of_stream_marker(f); + fclose(f); + + sim_args_init(&args, path); + TEST_ASSERT_EQUAL_INT(0, sched_sim_driver_start(&start_sim, &args)); + TEST_ASSERT_EQUAL_UINT64(SNAP_TS, xce_core_now(&(start_sim.core))); + + for (i = 0; i < 2; i++) { + struct capn c; + capn_ptr root; + xce_core_event_t ev; + int record_rc = 0; + + TEST_ASSERT_EQUAL_INT(0, capn_init_fp(&c, start_sim.replay_file, 0)); + TEST_ASSERT_TRUE(msg_root_has_shape(&c, &root, ReplayEvent_word_count, + ReplayEvent_pointer_count)); + TEST_ASSERT_TRUE(sim_process_replay_event_message(&start_sim, &c, root, + &ev, &record_rc) + >= 0); + TEST_ASSERT_EQUAL_INT(0, record_rc); + usleep(WAIT_US / 5); + TEST_ASSERT_EQUAL_UINT64_MESSAGE(ev_ts[i], + xce_core_now(&(start_sim.core)), + "an event earlier than the clock " + "must set it exactly, unclamped"); + } + + TEST_ASSERT_EQUAL_INT(0, run_observed(&run_sim, &args, 0)); + assert_observed_session_starts(begin, 1); +} + +/* 1.5: a mid-replay SnapshotEntry (ts S) followed by the endSession marker + * (ts 0) leaves the clock at the last event's ts T, after the run and + * after a real sleep. sim_start_ts is pinned to the initial snapshot so S + * is not chosen as the start snapshot and the real skip branch runs. */ +static void test_mid_replay_snapshot_and_end_leave_clock_at_last_event(void) +{ + static xce_sched_sim_driver_t sim; + char path[PATH_MAX]; + sched_sim_args_t args; + FILE* f; + const uint64_t t = SNAP_TS + 700000011ULL; + const uint64_t s = SNAP_TS + 900000000ULL; + + f = write_header_and_snapshot(path, sizeof(path), SNAP_TS); + append_queue_update_event(f, t, "padq"); + append_snapshot_entry(f, s); + append_end_of_stream_marker(f); + fclose(f); + + sim_args_init(&args, path); + args.sim_start_ts = SNAP_TS; + TEST_ASSERT_EQUAL_INT(0, run_observed(&sim, &args, 0)); + TEST_ASSERT_EQUAL_UINT64(SNAP_TS, sim.snapshot_ts); + TEST_ASSERT_EQUAL_UINT64(t, xce_core_now(&(sim.core))); + usleep(WAIT_US); + TEST_ASSERT_EQUAL_UINT64_MESSAGE(t, xce_core_now(&(sim.core)), + "clock must stay at the last event " + "after the end of the recording"); +} + +/* ---------------------------------------------------------------------- */ +/* Replay-loop integration tests (Requirements 1.2, 1.4, 1.6, 1.7, 1.8) */ +/* ---------------------------------------------------------------------- */ + +typedef struct { + uint64_t begin_round; + uint64_t duration_ns; + int n_decisions; + uint64_t decision_ts[8]; +} out_session_t; + +/* Decodes every ReplaySession in a simulation output file. */ +static int read_output_sessions(const char* path, out_session_t* out, int max) +{ + FILE* f = fopen(path, "r"); + struct capn c; + int n = 0; + + TEST_ASSERT_NOT_NULL(f); + while (capn_init_fp(&c, f, 0) == 0) { + capn_ptr root; + ReplaySession_ptr sp; + replay_session_t* dst = NULL; + int i; + + if (!msg_root_has_shape(&c, &root, ReplaySession_word_count, + ReplaySession_pointer_count)) { + capn_free(&c); + continue; + } + TEST_ASSERT_TRUE(n < max); + sp.p = root; + decode_ReplaySession_ptr(&dst, sp); + capn_free(&c); + TEST_ASSERT_NOT_NULL(dst); + TEST_ASSERT_NOT_NULL(dst->summary); + TEST_ASSERT_TRUE(dst->n_decisions <= 8); + out[n].begin_round = dst->begin_round; + out[n].duration_ns = dst->summary->duration_ns; + out[n].n_decisions = dst->n_decisions; + for (i = 0; i < dst->n_decisions; i++) { + out[n].decision_ts[i] = dst->decisions[i]->ts; + } + free_ReplaySession_ptr(&dst); + n++; + } + fclose(f); + return n; +} + +/* 1.2, 1.4, 1.7, 1.8: with an allocator that sleeps before deciding and + * beginRounds that are not whole seconds, session_start seen inside each + * session equals the recorded beginRound exactly, and the output has that + * beginRound, durationNs 0, and every decision ts equal to it. */ +static void test_session_start_and_output_times_equal_begin_round(void) +{ + static xce_sched_sim_driver_t sim; + char path[PATH_MAX]; + char out_path[PATH_MAX]; + sched_sim_args_t args; + const uint64_t begin[2] = { OUT_B1, OUT_B2 }; + out_session_t sessions[4]; + int n; + int k; + int i; + + write_two_session_fixture(path, sizeof(path)); + make_tmp_file(out_path, sizeof(out_path), "out"); + + sim_args_init(&args, path); + args.output_path = out_path; + TEST_ASSERT_EQUAL_INT(0, run_observed(&sim, &args, SLEEP_US)); + assert_observed_session_starts(begin, 2); + + n = read_output_sessions(out_path, sessions, 4); + TEST_ASSERT_EQUAL_INT(2, n); + for (k = 0; k < n; k++) { + TEST_ASSERT_EQUAL_UINT64(begin[k], sessions[k].begin_round); + TEST_ASSERT_EQUAL_UINT64_MESSAGE(0ULL, sessions[k].duration_ns, + "no logical time passes inside " + "a replayed session"); + TEST_ASSERT_TRUE_MESSAGE(sessions[k].n_decisions >= 1, + "each session must decide at least once"); + for (i = 0; i < sessions[k].n_decisions; i++) { + TEST_ASSERT_EQUAL_UINT64(begin[k], sessions[k].decision_ts[i]); + } + } + unlink(out_path); +} + +/* Reads a whole file; *hdr_end is the offset just past its first message + * (the ReplayHeader), whose fields other than recordingStartWall are + * returned in *hdr. */ +static char* slurp_output(const char* path, long* len, long* hdr_end, + replay_header_t* hdr) +{ + FILE* f = fopen(path, "r"); + struct capn c; + ReplayHeader_ptr p; + struct ReplayHeader wire; + char* buf; + + TEST_ASSERT_NOT_NULL(f); + 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(hdr, 0, sizeof(*hdr)); + decode_ReplayHeader(hdr, &wire); + capn_free(&c); + *hdr_end = ftell(f); + + TEST_ASSERT_EQUAL_INT(0, fseek(f, 0, SEEK_END)); + *len = ftell(f); + buf = (char*)malloc((size_t)*len); + TEST_ASSERT_NOT_NULL(buf); + rewind(f); + TEST_ASSERT_EQUAL_UINT((size_t)*len, fread(buf, 1, (size_t)*len, f)); + fclose(f); + return buf; +} + +/* 1.6: the same recording replayed twice gives identical output (every + * logical timestamp and every decision), except recordingStartWall. */ +static void test_two_replays_produce_identical_output(void) +{ + static xce_sched_sim_driver_t sim1; + static xce_sched_sim_driver_t sim2; + char path[PATH_MAX]; + char out[2][PATH_MAX]; + sched_sim_args_t args; + replay_header_t hdr[2]; + long len[2]; + long hdr_end[2]; + char* buf[2]; + int r; + + write_two_session_fixture(path, sizeof(path)); + for (r = 0; r < 2; r++) { + make_tmp_file(out[r], sizeof(out[r]), "det"); + sim_args_init(&args, path); + args.output_path = out[r]; + TEST_ASSERT_EQUAL_INT(0, run_observed((r == 0) ? &sim1 : &sim2, &args, + SLEEP_US)); + buf[r] = slurp_output(out[r], &len[r], &hdr_end[r], &hdr[r]); + } + + TEST_ASSERT_EQUAL_UINT64(hdr[0].recording_start_ts, + hdr[1].recording_start_ts); + TEST_ASSERT_EQUAL_UINT64(SNAP_TS, hdr[0].recording_start_ts); + TEST_ASSERT_EQUAL_STRING(hdr[0].recorder_version, hdr[1].recorder_version); + TEST_ASSERT_EQUAL_STRING(hdr[0].sched_policy, hdr[1].sched_policy); + TEST_ASSERT_EQUAL_PTR(NULL, hdr[0].cluster); + TEST_ASSERT_EQUAL_PTR(NULL, hdr[1].cluster); + + TEST_ASSERT_EQUAL_INT64(len[0] - hdr_end[0], len[1] - hdr_end[1]); + TEST_ASSERT_EQUAL_MEMORY_MESSAGE(buf[0] + hdr_end[0], buf[1] + hdr_end[1], + (size_t)(len[0] - hdr_end[0]), + "everything after the header must be " + "byte-identical across replays"); + + for (r = 0; r < 2; r++) { + free_ReplayHeader(&hdr[r]); + free(buf[r]); + unlink(out[r]); + } +} + int main(void) { UNITY_BEGIN(); @@ -157,5 +979,11 @@ int main(void) RUN_TEST(test_warn_negative_value); RUN_TEST(test_warn_empty_value); RUN_TEST(test_no_warning_when_option_absent); + RUN_TEST(test_start_pins_clock_to_snapshot_ts); + RUN_TEST(test_session_earlier_than_preceding_event_sets_exact); + RUN_TEST(test_events_earlier_than_start_snapshot_set_exact); + RUN_TEST(test_mid_replay_snapshot_and_end_leave_clock_at_last_event); + RUN_TEST(test_session_start_and_output_times_equal_begin_round); + RUN_TEST(test_two_replays_produce_identical_output); return UNITY_END(); }