myxce/src/scheduler/testsuite/test_sim_event_loop.c

/*
 * test_sim_event_loop.c - contract test for the simulation-mode event loop
 * and virtual clock advancement (scheduler-core spec, task 11.2, Req 5 AC
 * 14/18).
 *
 * Observable under test (task 11.2's own text):
 *   - replaying a fixture file with events at non-uniform timestamps
 *     processes them in file order with the virtual clock reporting each
 *     event's exact timestamp at the moment it's pushed.
 *
 * Two tests, at two different granularities:
 *
 *   Test 1 (unit-level, white-box): calls sim_process_replay_event_message()
 *   -- the extracted per-message decode+clock-advance+translate step --
 *   directly, sequentially, once per hand-built ReplayEvent message, with
 *   deliberately non-uniform timestamps (including a same-timestamp pair,
 *   Req 5 AC 14's "same-timestamp events processed in file order" clause),
 *   asserting a REAL xce_core_now(&sim.core) read matches EXACTLY after
 *   each call -- this is the precise, sequential, file-order proof: a
 *   picker-hook spy (the only other observation point xce_core_push_events()
 *   offers) cannot see this because every event in one accumulated batch is
 *   pushed through ONE xce_core_push_events() call, after the clock has
 *   already been walked through all of them.
 *
 *   Test 2 (integration-level): drives the full sched_sim_driver_run() --
 *   this task's other Observable, "treat each session-boundary message as
 *   the trigger for one complete scheduling round" -- against a fixture
 *   with TWO ReplaySession boundaries at increasing beginRound timestamps,
 *   asserting the run completes (rc == 0 -- this is the RED/GREEN proof:
 *   before task 11.2, sched_sim_driver_run() returned -ENOSYS here) and
 *   that the virtual clock ends at the SECOND session's beginRound -- which
 *   only happens if both session boundaries were reached, in order.
 *
 * Fixture strategy: same production-code-reuse technique test_sim_startup.c
 * already establishes for the header+snapshot prefix (sched_record_start()/
 * sched_record_stop() against a small real xce_scheduler_t fixture) PLUS
 * hand-constructed ReplayEvent/ReplaySession messages appended afterward,
 * using the SAME encode_ReplayEvent_ptr()/write_capnp_message() technique
 * test_e2e_capnp.c's own test_replay_event_job_state_change_round_trip/
 * test_replay_event_end_session_round_trip already prove works, and
 * low-level new_ReplaySession()/ReplaySession_set_beginRound() field
 * setters for ReplaySession, mirroring sim.c's own
 * write_lone_snapshotentry_message() precedent (test_sim_startup.c) for
 * minimal hand-built messages that leave irrelevant fields at their
 * zero-value defaults.
 *
 * White-box #include of ../src/core.c, ../src/core_clock.c,
 * ../src/core_events.c, ../src/core_session.c, ../src/decision.c (task
 * 11.2's own core-driving calls: xce_core_push_events()/xce_core_go()/
 * xce_core_end_session()/make_pend_decision()) PLUS ../src/plugin.c,
 * ../src/plugins/builtin_fcfs.c (the real plugin_register_core()/
 * builtin_fcfs_new_plugin(), task 11.2's own new core.h-scoped picker
 * registration path -- NOT schedobj.c/plugin_register(), which need a
 * genuine xce_scheduler_t* this test deliberately never builds one of,
 * same Req 5 AC 17 boundary sim.c/sim_loop.c themselves respect) PLUS
 * ../src/recording.c (fixture-building only, same as test_sim_startup.c)
 * PLUS ../src/sim.c/../src/sim_loop.c (the code under test) PLUS
 * ../src/sim_verify.c (task 11.4's sibling file -- sim_loop.c's
 * sched_sim_driver_run() now calls sim_decision_eq()/sim_log_mismatch() as
 * extern functions defined there; this white-box TU must include it too or
 * those calls are undefined references at link time, even though neither
 * test in this file exercises --sim-verify itself -- that is
 * test_sim_verify.c's own job) PLUS
 * ../src/snapshot.c/snapshot_wire.c/snapshot_load.c (task 8.1, needed by
 * both the fixture-building and sim.c's own snapshot load).
 */
#include <errno.h>
#include <limits.h>
#include <stdatomic.h>
#include <stdint.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>

#include <unity/unity.h>

#include "controller.capnp.h"
#include "kvmgr.h"
#include "scheduler.h"
#include "xstring.h"

/* Pull in the real functions under test into this TU (white-box #include,
 * same convention as test_sim_startup.c/test_core_full_cycle.c). */
#include "../src/core.c"
#include "../src/core_clock.c"
#include "../src/core_events.c"
#include "../src/core_session.c"
#include "../src/decision.c"
#include "../src/plugin.c"
#include "../src/plugins/builtin_fcfs.c"
#include "../src/recording.c"
#include "../src/sim.c"
#include "../src/sim_loop.c"
#include "../src/sim_verify.c"
#include "../src/snapshot.c"
#include "../src/snapshot_load.c"
#include "../src/snapshot_wire.c"

void kill_sched_job(void* arg) { (void)arg; }
void kill_sched_queue(void* arg) { (void)arg; }
void kill_sched_worker(void* arg) { (void)arg; }
void kill_sched_user(void* arg) { (void)arg; }
void kill_sched_project(void* arg) { (void)arg; }
void kill_sched_resval(xce_val_t* val) { (void)val; }

/* core_reset_workers()'s (core_session.c) linker dependency -- see
 * test_core_full_cycle.c's own file header comment for the same stub;
 * never invoked (no worker is ever added here). */
void sched_worker_release_host_ctx(xce_sched_worker_t* w) { (void)w; }

/* plugin_prepare_allocation()'s (plugin.c, the "limit" plugin category)
 * linker dependency -- never invoked (this file never drives real
 * allocation, Req 5 AC 17 / this file's own header comment). */
int sched_fill_limit_requests(xce_scheduler_t* sched, xvec_idval_t* vec,
                              xlist_t* limits)
{
    (void)sched;
    (void)vec;
    (void)limits;
    return 0;
}

/* Real reimplementation of schedobj.c's sched_resolve_job_picker -- same
 * rationale/algorithm as test_sim_startup.c's/test_core_full_cycle.c's own
 * copies (schedobj.c is not linked: it would need plugin_register(), which
 * requires a genuine xce_scheduler_t* neither this file's fixtures nor
 * sim_loop.c's own core-scoped registration path have, Req 5 AC 17). */
int sched_resolve_job_picker(xce_sched_core_t* core, xce_sched_plugin_t** out)
{
    xce_sched_plugin_t* plugin = NULL;
    int                 rc     = 0;

    if ((core == NULL) || (out == NULL)) {
        return -EINVAL;
    }

    *out = NULL;

    if (core->job_pick_plugin == NULL) {
        return -EINVAL;
    }

    lock(core->plugins, RTTI_LOCK_READ);
    rc = key_find(core->plugins, core->job_pick_plugin, (void**)&plugin);
    unlock(core->plugins);

    if (rc <= 0) {
        return (rc < 0) ? rc : -ENOENT;
    }

    if ((plugin == NULL) || (plugin->picker_iface == NULL)
        || (plugin->picker_iface->next_job == NULL)
        || (plugin->picker_iface->begin_session == NULL)
        || (plugin->picker_iface->end_session == NULL)
        || (plugin->picker_iface->on_dispatch == NULL)
        || (plugin->picker_iface->on_pend == NULL)) {
        return -EINVAL;
    }

    *out = plugin;

    return 0;
}

void setUp(void) { return; }
void tearDown(void) { return; }

/* ---------------------------------------------------------------------- */
/* Fixture helpers                                                         */
/* ---------------------------------------------------------------------- */

static void make_tmp_dir(char* out, size_t out_len, const char* suffix)
{
    char tmpl[PATH_MAX];

    snprintf(tmpl, sizeof(tmpl), "/tmp/sched_simloop%s_XXXXXX", suffix);
    TEST_ASSERT_NOT_NULL(mkdtemp(tmpl));
    TEST_ASSERT_TRUE(strlen(tmpl) < out_len);
    strcpy(out, tmpl);
}

static int stub_picker_begin_session(void* picker)
{
    (void)picker;
    return 0;
}

static xce_sched_job_t* stub_picker_next_job(void* picker)
{
    (void)picker;
    return NULL;
}

static int stub_picker_end_session(void* picker)
{
    (void)picker;
    return 0;
}

static int stub_picker_jobid_noop(void* picker, uint64_t jobid)
{
    (void)picker;
    (void)jobid;
    return 0;
}

static int stub_picker_queue_update_noop(void* picker, xce_sched_queue_t* q)
{
    (void)picker;
    (void)q;
    return 0;
}

static int stub_picker_job_noop(void* picker, xce_sched_job_t* job)
{
    (void)picker;
    (void)job;
    return 0;
}

static int stub_picker_job_state_change_noop(void* picker, xce_sched_job_t* job,
                                             uint32_t state)
{
    (void)picker;
    (void)job;
    (void)state;
    return 0;
}

static xce_job_picker_iface_t g_stub_picker_iface
    = { .init                = NULL,
        .fini                = NULL,
        .begin_session       = stub_picker_begin_session,
        .next_job            = stub_picker_next_job,
        .end_session         = stub_picker_end_session,
        .on_dispatch         = stub_picker_jobid_noop,
        .on_pend             = stub_picker_jobid_noop,
        .on_queue_update     = stub_picker_queue_update_noop,
        .on_job_push         = stub_picker_job_noop,
        .on_job_state_change = stub_picker_job_state_change_noop,
        .on_job_update       = stub_picker_job_noop,
        .on_config           = NULL,
        .on_job_rusage       = NULL };

static xce_sched_plugin_t g_stub_picker_plugin
    = { .name         = "test_sim_event_loop_picker",
        .picker_iface = &g_stub_picker_iface };

/* Builds a small, real xce_scheduler_t fixture purely to drive
 * sched_record_start()/sched_record_stop() and produce a REAL, valid
 * `.replay` header+snapshot prefix this test's OWN code under test then
 * consumes (same pattern test_sim_startup.c's build_fixture_scheduler()
 * establishes). Deliberately registers a picker OTHER than "builtin_fcfs"
 * (name never matches) -- neither test in this file pushes a real
 * candidate job, so which picker resolves at replay time never matters,
 * and this keeps the fixture's own recorded schedPolicy from accidentally
 * satisfying sim_register_default_picker()'s exact-name check. */
static void build_fixture_scheduler(xce_scheduler_t* sched, char* record_dir)
{
    int rc;

    memset(sched, 0, sizeof(*sched));
    mylock_init(&(sched->record_lock));

    rc = xce_sched_core_init(&(sched->core));
    TEST_ASSERT_EQUAL_INT(0, rc);

    lock(sched->core.plugins, RTTI_LOCK_WRITE);
    rc = key_add(sched->core.plugins, (void*)g_stub_picker_plugin.name,
                 (void*)&g_stub_picker_plugin);
    TEST_ASSERT_GREATER_OR_EQUAL(0, rc);
    unlock(sched->core.plugins);

    sched->core.job_pick_plugin = g_stub_picker_plugin.name;
    sched->record_dir           = record_dir;
}

/* Produces a valid `.replay` header+snapshot prefix at `out_path` via the
 * real sched_record_start() -- identical technique test_sim_startup.c's own
 * write_valid_fixture() establishes for the header+snapshot bytes
 * themselves -- but deliberately does NOT call sched_record_stop(): that
 * function unconditionally appends its own ReplayEvent{endSession} marker
 * (recording.c, Req 5 AC 20) immediately after the snapshot, which would
 * terminate sched_sim_driver_run()'s read loop the instant it reached that
 * marker -- before ever seeing anything this test's own callers append
 * afterward. Returns the STILL-OPEN FILE* (sched_record_start()'s own
 * `sched.record_file`, positioned right after the snapshot) so the caller
 * can append its own hand-built events/sessions directly onto the same
 * stream, ending with its own append_end_of_stream_marker() call, then
 * fclose() it directly -- bypassing sched_record_stop()'s real
 * event/session-recording API entirely, which this test never exercises. */
static FILE* write_header_and_snapshot(char* out_path, size_t out_path_len,
                                       const char* suffix)
{
    char            record_dir[PATH_MAX];
    xce_scheduler_t sched;
    int             rc;

    make_tmp_dir(record_dir, sizeof(record_dir), suffix);
    build_fixture_scheduler(&sched, record_dir);

    rc = sched_record_start(&sched, "fixture.replay");
    TEST_ASSERT_EQUAL_INT(0, rc);
    TEST_ASSERT_TRUE(strlen(sched.record_file_path) < out_path_len);
    strcpy(out_path, sched.record_file_path);

    return sched.record_file;
}

/* Appends one ReplayEvent{jobStateChange} message to the already-open
 * stream `f` -- same encode_ReplayEvent_ptr()/write_capnp_message()
 * technique test_e2e_capnp.c's own
 * test_replay_event_job_state_change_round_trip already proves works.
 * jobStateChange is used throughout this file purely as a cheap,
 * already-proven-safe payload shape (a plain value struct, no further
 * nested pointers) -- core_events.c's own JOB_STATE_CHANGE handling
 * WARN+skips an unknown jobid rather than fail (verified in core_events.c),
 * so neither test here needs a real job to exist in the loaded snapshot.
 */
static void append_jobstatechange_event(FILE* f, uint64_t ts, uint64_t jobid)
{
    job_state_change_event_t change;
    replay_event_t           src;
    struct capn              c;
    struct capn_segment*     cs;
    ReplayEvent_ptr          p;

    memset(&change, 0, sizeof(change));
    change.jobid     = jobid;
    change.old_state = 1u;
    change.new_state = 2u;
    change.exit_code = 0;

    memset(&src, 0, sizeof(src));
    src.ts                    = ts;
    src.body_which            = ReplayEvent_body_jobStateChange;
    src.body.job_state_change = &change;

    capn_init_malloc(&c);
    cs = capn_root(&c).seg;
    encode_ReplayEvent_ptr(cs, &p, &src);
    capn_setp(capn_root(&c), 0, p.p);

    TEST_ASSERT_EQUAL_INT(0, write_capnp_message(f, &c));
    capn_free(&c);
}

/* Appends the .replay stream's whole-stream termination marker
 * (ReplayEvent{endSession}, Req 5 AC 20) -- same technique
 * test_e2e_capnp.c's own test_replay_event_end_session_round_trip already
 * proves works. */
static void append_end_of_stream_marker(FILE* f)
{
    replay_event_t       src;
    struct capn          c;
    struct capn_segment* cs;
    ReplayEvent_ptr      p;

    memset(&src, 0, sizeof(src));
    src.body_which = ReplayEvent_body_endSession;

    capn_init_malloc(&c);
    cs = capn_root(&c).seg;
    encode_ReplayEvent_ptr(cs, &p, &src);
    capn_setp(capn_root(&c), 0, p.p);

    TEST_ASSERT_EQUAL_INT(0, write_capnp_message(f, &c));
    capn_free(&c);
}

/* Appends a minimal ReplaySession message with only beginRound set --
 * decisions/summary are left at new_ReplaySession()'s zero-value defaults
 * (empty list / capn_null), mirroring sim.c's own
 * write_lone_snapshotentry_message() precedent (test_sim_startup.c) for
 * hand-built messages whose irrelevant fields nothing under test reads.
 * sched_sim_driver_run() only ever reads beginRound off a ReplaySession
 * (via the lightweight ReplaySession_get_beginRound() accessor, task 11.2's
 * own scope: the decisions/summary fields are task 11.4/11.5 territory). */
static void append_replay_session(FILE* f, uint64_t begin_round)
{
    struct capn          c;
    struct capn_segment* cs;
    ReplaySession_ptr    sp;

    capn_init_malloc(&c);
    cs = capn_root(&c).seg;
    sp = new_ReplaySession(cs);
    ReplaySession_set_beginRound(sp, begin_round);
    capn_setp(capn_root(&c), 0, sp.p);

    TEST_ASSERT_EQUAL_INT(0, write_capnp_message(f, &c));
    capn_free(&c);
}

/* ---------------------------------------------------------------------- */
/* Test 1: sim_process_replay_event_message() advances the virtual clock   */
/* to each event's own timestamp, in exact file order, including a        */
/* same-timestamp pair (Req 5 AC 14).                                      */
/* ---------------------------------------------------------------------- */

static void
test_process_replay_event_advances_virtual_clock_to_exact_event_ts_in_file_order(
    void)
{
    static const uint64_t ts_values[5]
        = { 1000ULL, 1050ULL, 1050ULL, 500000ULL, 9999999999ULL };
    char                   path[PATH_MAX];
    xce_sched_sim_driver_t sim;
    FILE*                  f;
    int                    i;
    int                    rc;

    /* A minimal sim driver: only sim->core is exercised by
     * sim_process_replay_event_message() (via sim_advance_virtual_clock()),
     * so xce_sched_core_init() plus a frozen (speed=0) clock is a
     * sufficient, self-contained fixture -- no replay file, snapshot, or
     * full sched_sim_driver_start() needed for this unit-level test. */
    memset(&sim, 0, sizeof(sim));
    rc = xce_sched_core_init(&(sim.core));
    TEST_ASSERT_EQUAL_INT(0, rc);

    /* speed_num=0 (discrete/frozen -- AC 3's own documented mode table,
     * core_clock.c) so xce_core_now() reports EXACTLY virtual_clock_base
     * regardless of real wall-clock time elapsed between
     * sim_advance_virtual_clock()'s clock_gettime() call and this test's
     * own assertion immediately after -- deterministic, never flaky. */
    atomic_store_explicit(&(sim.core.speed_num), 0, memory_order_release);
    atomic_store_explicit(&(sim.core.speed_den), 1, memory_order_release);

    make_tmp_dir(path, sizeof(path), "_clockjump");
    strcat(path, "/events.replay");

    f = fopen(path, "w");
    TEST_ASSERT_NOT_NULL(f);
    for (i = 0; i < 5; i++) {
        append_jobstatechange_event(f, ts_values[i], 42ULL);
    }
    fclose(f);

    f = fopen(path, "r");
    TEST_ASSERT_NOT_NULL(f);

    for (i = 0; i < 5; i++) {
        struct capn      c;
        capn_ptr         root;
        xce_core_event_t ev;
        int              translated;

        TEST_ASSERT_EQUAL_INT(0, capn_init_fp(&c, f, 0));
        TEST_ASSERT_TRUE(msg_root_has_shape(&c, &root, ReplayEvent_word_count,
                                            ReplayEvent_pointer_count));

        translated = sim_process_replay_event_message(&sim, &c, root, &ev);

        TEST_ASSERT_EQUAL_INT(1, translated);
        TEST_ASSERT_EQUAL_INT(XCE_CORE_EVENT_JOB_STATE_CHANGE, ev.type);
        TEST_ASSERT_EQUAL_UINT64(ts_values[i], ev.ts);

        /* The Observable itself: a REAL clock read, immediately after this
         * one event was processed, matches that event's exact timestamp --
         * not a trivially-true check, and not observable any other way
         * (see this file's own header comment for why a picker-hook spy
         * cannot see per-event granularity). */
        TEST_ASSERT_EQUAL_UINT64(ts_values[i], xce_core_now(&(sim.core)));
    }

    fclose(f);

    printf("test_process_replay_event_advances_virtual_clock_to_exact_event_"
           "ts_in_file_order: PASS\n");
}

/* ---------------------------------------------------------------------- */
/* Test 2: sched_sim_driver_run() treats each ReplaySession boundary as    */
/* the trigger for one complete scheduling round, in file order, across    */
/* multiple rounds -- this task's other Observable, and the RED->GREEN     */
/* proof (sched_sim_driver_run() returned -ENOSYS here before task 11.2).  */
/* ---------------------------------------------------------------------- */

static void
test_run_processes_multiple_sessions_in_order_and_no_longer_returns_enosys(void)
{
    char                   path[PATH_MAX];
    FILE*                  f;
    sched_sim_args_t       args;
    xce_sched_sim_driver_t sim;
    int                    rc;

    f = write_header_and_snapshot(path, sizeof(path), "_run");
    TEST_ASSERT_NOT_NULL(f);

    /* Non-uniform timestamps throughout, two ReplaySession boundaries at
     * strictly increasing beginRound values -- the final virtual-clock
     * assertion below only holds if BOTH boundaries were reached, in
     * order (a driver that stalled after the first, or processed them out
     * of order, would leave the clock at the wrong value). */
    append_jobstatechange_event(f, 2000ULL, 999ULL); /* unknown jobid --
                                                      * core_events.c
                                                      * WARN+skips it,
                                                      * verified safe */
    append_jobstatechange_event(f, 2500ULL, 999ULL);
    append_replay_session(f, 3000ULL);               /* session boundary #1 */
    append_jobstatechange_event(f, 4000ULL, 999ULL);
    append_replay_session(f, 5000ULL);               /* session boundary #2 */
    append_end_of_stream_marker(f);

    fclose(f);

    memset(&args, 0, sizeof(args));
    args.replay_path  = path;
    args.sim_start_ts = SCHED_SIM_START_TS_LATEST;
    args.sim_speed    = 0; /* frozen -- deterministic final-clock assertion */

    rc = sched_sim_driver_run(&sim, &args);

    /* The RED->GREEN proof: before task 11.2, this returned -ENOSYS
     * (test_sim_startup.c's own former test_run_starts_then_fails_enosys_
     * pending_event_loop, removed now that this Observable supersedes it). */
    TEST_ASSERT_EQUAL_INT(0, rc);

    /* Both ReplaySession boundaries were reached, in file order: the clock
     * ends at the SECOND (later) beginRound, not the first, and not the
     * last ReplayEvent's own ts (2000/2500/4000 all precede 5000). */
    TEST_ASSERT_EQUAL_UINT64(5000ULL, xce_core_now(&(sim.core)));

    printf("test_run_processes_multiple_sessions_in_order_and_no_longer_"
           "returns_enosys: PASS\n");
}

int main(void)
{
    UNITY_BEGIN();

    RUN_TEST(
        test_process_replay_event_advances_virtual_clock_to_exact_event_ts_in_file_order);
    RUN_TEST(
        test_run_processes_multiple_sessions_in_order_and_no_longer_returns_enosys);

    return UNITY_END();
}