From 19d8df3532d77c98762043dd0fe7d41ed7841bae Mon Sep 17 00:00:00 2001 From: Rongsong Shen Date: Tue, 4 Aug 2026 04:05:57 +0000 Subject: [PATCH] test(scheduler-core): add core contract test for full push-go-drain-end cycle (task 4.6) Adds test_core_full_cycle.c: drives xce_core_push_events() -> xce_core_go() -> decision drain -> xce_core_end_session() end-to-end against a real picker pthread (running the real xce_job_picker_iface_t hooks on a test-local stub picker) and a main-thread stand-in allocator, since sched_allocator() is xce_scheduler_t*-coupled and out of pure-core scope. Proves both preconditions (core_go reentrancy, push_events mid-session) reject against a genuinely active session and recover once it ends. Closes out group 4 (Foundation: Scheduler Core Extraction) -- tasks 4.1-4.6 all complete. Co-Authored-By: Claude Sonnet 5 Claude-Session: https://claude.ai/code/session_019dw5WnN3DXStaEaPYfACU1 --- .kiro/specs/scheduler-core/tasks.md | 3 +- src/scheduler/testsuite/meson.build | 23 + .../testsuite/test_core_full_cycle.c | 743 ++++++++++++++++++ 3 files changed, 768 insertions(+), 1 deletion(-) create mode 100644 src/scheduler/testsuite/test_core_full_cycle.c diff --git a/.kiro/specs/scheduler-core/tasks.md b/.kiro/specs/scheduler-core/tasks.md index 9616ec94..9fbe4f20 100644 --- a/.kiro/specs/scheduler-core/tasks.md +++ b/.kiro/specs/scheduler-core/tasks.md @@ -124,11 +124,12 @@ - _Verified: 2026-08-03 — xce_core_now() (new core_clock.c, kept out of core.c so test_core_isolation.c's wall-clock scan still passes) implements the exact two-anchor+speed formula; all 4 modes proven with fixed anchors, no sleep-based flakiness. requirements.md's literal tlog_reg_format_func(fn, core_ptr) call shape doesn't exist in the real vendored tlog API (verified against tlog.h: registration takes only the callback, userptr is per-call-site) — resolved via a module-level active-core pointer read by a new core-scoped format function, not wired into mylog_init()/main.c yet. session_start/session_duration_ns stamped correctly in xce_core_go()/xce_core_end_session() (task 4.4's behavior otherwise untouched). 18/18 scheduler suite passing, clang-format clean, review APPROVED first pass. Follow-up noted: core_clock.c is the 3rd core-adjacent file (after core_events.c, core_session.c) outside test_core_isolation.c's nm-based audit — manually verified clean, not mechanically enforced._ - _Depends: 4.1_ -- [ ] 4.6 Core contract tests +- [x] 4.6 Core contract tests - Test the full push-events → start-session → drain-decisions → end-session cycle against a stub picker (no driver needed), asserting hook firing order and timing - Observable: the contract test suite passes, exercising both the batch-push precondition (rejects mid-session) and the double-session-start rejection - _Requirements: 5.1, 5.2_ - _Depends: 4.4_ + - _Verified: 2026-08-03 — new test_core_full_cycle.c drives push_events→go→drain→end_session against a real picker pthread (running the real xce_job_picker_iface_t hooks on a test-local stub picker) plus a main-thread stand-in allocator (sched_allocator() is xce_scheduler_t*-coupled, out of pure-core scope) calling real on_dispatch/on_pend/make_dispatch_decision/make_pend_decision. Both preconditions (-EINVAL reentrant core_go, -EBUSY mid-session push_events) proven against a genuinely active session, not a hand-set flag, and proven to un-latch after the session ends. Two live RED→GREEN reproductions (neutered each guard in turn), both independently reproduced by review. 19/19 scheduler suite passing, clang-format clean, 30/30 stress runs with zero flakiness, review APPROVED first pass. Zero production files touched — test-only, as scoped._ - [ ] 5. Core: Normal-Mode Driver diff --git a/src/scheduler/testsuite/meson.build b/src/scheduler/testsuite/meson.build index 81825777..3d9dda99 100644 --- a/src/scheduler/testsuite/meson.build +++ b/src/scheduler/testsuite/meson.build @@ -276,6 +276,26 @@ sched_core_clock_test = executable( dependencies: [scheduler_test_deps], install: false) +# Core contract test for the full push-events -> start-session -> +# drain-decisions -> end-session cycle (scheduler-core spec, task 4.6, Req 5 +# AC 1 / AC 2), against a real picker thread + a stand-in allocator (this +# test's own driver thread) driving a minimal stub picker plugin through +# xce_core_push_events()/xce_core_go()/xce_core_end_session() end to end -- +# also proving, mid-cycle against a GENUINELY active session, both the +# reentrant xce_core_go() rejection (Req 5 AC 5) and the xce_core_push_events() +# -EBUSY rejection (Req 5 AC 8) compose correctly with the rest of the +# cycle. Same white-box #include pattern as test_core_events.c/ +# test_core_session.c (pulls ../src/core.c, ../src/core_events.c, +# ../src/core_session.c, ../src/core_clock.c, and ../src/decision.c into +# this TU) -- see the test's own header comment for the allocator-thread- +# vs-direct-jbuf-drain design call. +sched_core_full_cycle_test = executable( + 'sched_core_full_cycle_test', + ['test_core_full_cycle.c'], + include_directories:[incdir], + dependencies: [scheduler_test_deps], + install: false) + # Register as meson tests so `meson test --suite scheduler` runs them all. # detect_leaks=0 matches the suite convention (fixtures are not torn down). test('scheduler test', scheduler_test, suite: 'scheduler', @@ -321,3 +341,6 @@ test('scheduler core session test', test('scheduler core clock test', sched_core_clock_test, suite: 'scheduler', env: ['ASAN_OPTIONS=detect_leaks=0']) +test('scheduler core full cycle test', + sched_core_full_cycle_test, suite: 'scheduler', + env: ['ASAN_OPTIONS=detect_leaks=0']) diff --git a/src/scheduler/testsuite/test_core_full_cycle.c b/src/scheduler/testsuite/test_core_full_cycle.c new file mode 100644 index 00000000..b89da616 --- /dev/null +++ b/src/scheduler/testsuite/test_core_full_cycle.c @@ -0,0 +1,743 @@ +/* + * test_core_full_cycle.c - core contract test for the complete + * push-events -> start-session -> drain-decisions -> end-session cycle + * (scheduler-core spec, task 4.6, Req 5 AC 1 / AC 2, exercised via AC 4/ + * AC 5/AC 6/AC 8's already-implemented machinery). + * + * Every task 4.1-4.5 piece this test drives is REAL production code, called + * exactly the way a real driver would call it: xce_core_push_events() + * (task 4.3), xce_core_go()/xce_core_set_decision_cb()/ + * xce_core_end_session() (task 4.4), and (transitively, via xce_core_go()/ + * xce_core_end_session()) xce_core_now() (task 4.5). The only thing this + * file writes that is NOT already-tested production code is a minimal, but + * genuinely conforming, STUB picker plugin (xce_job_picker_iface_t) -- per + * this task's own text, a stub is deliberately preferred over builtin_fcfs + * here: it proves the CORE'S side of the contract (does xce_sched_core_t + * correctly drive an arbitrary conforming picker through the full cycle?) + * independent of builtin_fcfs's own ordering logic, which already has its + * own dedicated test files (test_picker_fcfs.c / test_picker_fcfs_session.c). + * + * Design call: real picker thread + a "stand-in allocator" in the test's own + * (main) thread, NOT a spawned sched_allocator() thread. + * + * sched_allocator() (scheduler.c) is `static void* + * sched_allocator(void* arg)` taking xce_scheduler_t* -- its body calls + * sched_make_allocation()/plugin_job_election()/sched_commit_decision(), + * all of which are themselves xce_scheduler_t*-typed and reach into + * sched->apictx-adjacent machinery this spec's Phase 2 core extraction + * deliberately left OUT of xce_sched_core_t (core.h's own boundary + * comment: "no apictx/cluster_cfg -- zero controller I/O"). Spinning up a + * real sched_allocator() thread here would mean either (a) constructing a + * full xce_scheduler_t fixture (dragging in scheduler.c/schedobj.c's whole + * driver-only stub surface -- exactly what this task's own scope boundary + * forbids: "without reaching into xce_scheduler_t/scheduler.c's + * driver-only machinery, which is out of a pure-core contract test's + * scope"), or (b) reimplementing real resource-matching allocation logic + * in a test file, which is scheduler-allocation spec territory, not this + * contract test's job. + * + * What core.h's own public API DOES expose directly on xce_sched_core_t + * is exactly the three cbufs the picker/allocator/driver pipeline + * communicates over -- wbuf, jbuf, decisions (core.h's own field + * comments: "action to trigger scheduling" / "job picked for allocation" + * / "scheduler decisions") -- plus sched_resolve_job_picker() and + * sched_get_job(). That is sufficient to reproduce sched_job_picker()'s + * real per-session body verbatim (test_picker_thread_fn() below, + * parameterized on xce_sched_core_t* directly instead of + * xce_scheduler_t*) as a REAL, separate pthread, and to reproduce + * sched_allocator()'s real jbuf-drain / on_dispatch-on_pend / decision- + * push loop structure (minus its resource-matching internals, which this + * task is not testing) directly in this test function -- "a test function + * IS the driver here" per this task's own framing, and per this design + * call, the test function is ALSO the stand-in allocator. + * + * A real, separate picker pthread is not optional scaffolding: jbuf's + * capacity is hardcoded to 1 (Req 5 AC 7, test_core_jbuf_capacity_pin.c) + * -- a purely sequential single-threaded test would deadlock the instant + * the picker side tried to enqueue a second item (or the end-of-session 0 + * sentinel) without a concurrent consumer draining the first. Running the + * picker loop on its own real thread, with the main thread draining jbuf + * concurrently, is not just closer to production reality, it is the ONLY + * way to genuinely exercise more than one job through this pipeline at + * all -- which is exactly the AC 7 invariant this design organically + * proves as a byproduct, on top of this task's own AC 1/AC 2 target. + * + * Thread-safety note: the stub picker's on_dispatch()/on_pend() hooks + * (called from the main/allocator thread) deliberately do NOT touch the + * same candidate-list state next_job()/begin_session() (picker thread) + * read/write -- unlike builtin_fcfs.c's real candidate-removal logic + * (which needs its own mylock_t for exactly this cross-thread hazard, + * see that file's own comments), this stub only needs to prove each hook + * fires with the right jobid at the right pipeline stage for a SINGLE + * session, so on_dispatch()/on_pend() are pure event-log recorders here. + * The event log itself (g_events) is written from both threads but each + * log_event() call claims a unique slot via atomic_fetch_add() before + * writing to it, so no two threads ever write the same memory location. + * Every other piece of cross-thread ordering this test relies on is + * already guaranteed by cbuf_put()/cbuf_get()'s own mutex+condvar pair + * (mylock.c) acting as a release/acquire synchronization point per item, + * plus a final pthread_join() before this test reads anything the picker + * thread touched. + * + * Req 5 AC 1 (mode-agnostic core): nothing in this test passes or asserts a + * mode flag -- xce_sched_core_init()/xce_core_push_events()/xce_core_go()/ + * xce_core_end_session() all take the exact same arguments this test uses, + * with no branch anywhere in the call chain conditioned on a "mode". There + * is only one mode exercised anywhere today (no driver exists yet), so this + * AC is satisfied by construction/absence-of-evidence, not a dedicated + * assertion. + * + * White-box #include of core.c, core_events.c, core_session.c, core_clock.c, + * and decision.c (same convention as test_core_events.c/test_core_session.c) + * so this test drives the REAL production functions under test, not + * reimplementations. Stubs below are the union of what test_core_events.c + * and test_core_session.c each already document needing: the kill_sched_ + * family / plugin_unregister (kvmgr/zarray datakiller addresses, never invoked + * -- fixtures here are never torn down, same convention as every white-box test + * in this directory), sched_worker_release_host_ctx (core_session.c's + * core_reset_workers() linker dependency, never invoked -- no worker is + * ever added to core->workers here), and sched_resolve_job_picker (a real, + * working reimplementation of schedobj.c's function of the same name, + * identical algorithm -- needed both by core_events.c's + * xce_core_push_events() internally, and by this file's own picker-thread/ + * stand-in-allocator code, exactly mirroring how the real production code + * has two independent call sites for the same resolver, scheduler.c's + * sched_job_picker()/sched_allocator()). + */ +#include +#include +#include +#include +#include +#include + +#include + +#include "kvmgr.h" +#include "preason.h" +#include "scheduler.h" +#include "xstring.h" + +/* Pull in the real xce_sched_core_init()/xce_core_push_events()/ + * xce_core_go()/xce_core_end_session()/xce_core_set_decision_cb()/ + * xce_core_now()/make_*_decision() under test into this TU (white-box + * #include, same convention as test_core_events.c/test_core_session.c). */ +#include "../src/core.c" +#include "../src/core_clock.c" +#include "../src/core_events.c" +#include "../src/core_session.c" +#include "../src/decision.c" + +void kill_sched_job(void* arg) { (void)arg; } +void kill_sched_queue(void* arg) { (void)arg; } +void kill_sched_worker(void* arg) { (void)arg; } +void kill_sched_user(void* arg) { (void)arg; } +void kill_sched_project(void* arg) { (void)arg; } +void kill_sched_resval(xce_val_t* val) { (void)val; } +void plugin_unregister(void* arg) { (void)arg; } + +/* core_reset_workers()'s (core_session.c) linker dependency -- see file + * header comment; never invoked (no worker is ever added here). */ +void sched_worker_release_host_ctx(xce_sched_worker_t* w) { (void)w; } + +/* Real reimplementation of schedobj.c's sched_resolve_job_picker -- see + * file header comment for why this file provides its own copy instead of + * #include-ing schedobj.c (identical algorithm to test_core_events.c's own + * copy). */ +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; } + +/* ---------------------------------------------------------------------- */ +/* Event log: records hook firing order/timing across both threads (the */ +/* main/driver+allocator thread and the real picker thread) via a single */ +/* atomically-claimed index per entry -- see file header comment for the */ +/* thread-safety argument. */ +/* ---------------------------------------------------------------------- */ + +typedef enum { + EVT_ON_QUEUE_UPDATE, + EVT_ON_JOB_PUSH, + EVT_BEGIN_SESSION, + EVT_NEXT_JOB, /* jobid == 0 means "returned NULL, session exhausted" */ + EVT_ON_DISPATCH, + EVT_ON_PEND, + EVT_END_SESSION, +} stub_event_tag_t; + +#define MAX_EVENTS 32 + +typedef struct { + stub_event_tag_t tags[MAX_EVENTS]; + uint64_t jobids[MAX_EVENTS]; + _Atomic int n; +} event_log_t; + +static event_log_t g_events; + +static void log_event(stub_event_tag_t tag, uint64_t jobid) +{ + int idx = atomic_fetch_add(&(g_events.n), 1); + + if (idx < MAX_EVENTS) { + g_events.tags[idx] = tag; + g_events.jobids[idx] = jobid; + } +} + +static int find_log_index(stub_event_tag_t tag, uint64_t jobid) +{ + int n = atomic_load(&(g_events.n)); + int cap = (n > MAX_EVENTS) ? MAX_EVENTS : n; + int i; + + for (i = 0; i < cap; i++) { + if ((g_events.tags[i] == tag) && (g_events.jobids[i] == jobid)) { + return i; + } + } + + return -1; +} + +/* ---------------------------------------------------------------------- */ +/* Stub picker: a minimal, genuinely conforming xce_job_picker_iface_t -- */ +/* see file header comment for why a stub (not builtin_fcfs) is used here. */ +/* ---------------------------------------------------------------------- */ + +#define STUB_MAX_CANDIDATES 8 + +typedef struct { + xce_sched_job_t* candidates[STUB_MAX_CANDIDATES]; + int ncandidates; + int cursor; +} stub_picker_state_t; + +static stub_picker_state_t g_stub_state; + +static int stub_on_job_push(void* picker, xce_sched_job_t* job) +{ + (void)picker; + + if (g_stub_state.ncandidates < STUB_MAX_CANDIDATES) { + g_stub_state.candidates[g_stub_state.ncandidates++] = job; + } + log_event(EVT_ON_JOB_PUSH, job->jobid); + + return 0; +} + +static int stub_on_queue_update(void* picker, xce_sched_queue_t* queue) +{ + (void)picker; + (void)queue; + + log_event(EVT_ON_QUEUE_UPDATE, 0); + + return 0; +} + +static int stub_on_job_state_change(void* picker, xce_sched_job_t* job, + uint32_t state) +{ + (void)picker; + (void)job; + (void)state; + + return 0; /* not exercised by this test's event batch */ +} + +static int stub_on_job_update(void* picker, xce_sched_job_t* job) +{ + (void)picker; + (void)job; + + return 0; /* not exercised by this test's event batch */ +} + +/* begin_session()/next_job() run in the real picker thread ONLY -- see + * file header comment for why no lock is needed against + * on_dispatch()/on_pend() below. */ +static int stub_begin_session(void* picker) +{ + (void)picker; + + /* Session-aware requirement (plugin.h): reset the iteration cursor -- + * never carry it over from the previous session. */ + g_stub_state.cursor = 0; + log_event(EVT_BEGIN_SESSION, 0); + + return 0; +} + +static xce_sched_job_t* stub_next_job(void* picker) +{ + (void)picker; + + if (g_stub_state.cursor >= g_stub_state.ncandidates) { + log_event(EVT_NEXT_JOB, 0); + return NULL; + } + + xce_sched_job_t* job = g_stub_state.candidates[g_stub_state.cursor++]; + + log_event(EVT_NEXT_JOB, job->jobid); + + return job; +} + +static int stub_end_session(void* picker) +{ + (void)picker; + + log_event(EVT_END_SESSION, 0); + + return 0; +} + +/* Post-allocation hooks (plugin.h: allocator thread, in production; the + * main thread's stand-in allocator here) -- pure recorders, deliberately + * not touching g_stub_state (see file header comment). */ +static int stub_on_dispatch(void* picker, uint64_t jobid) +{ + (void)picker; + + log_event(EVT_ON_DISPATCH, jobid); + + return 0; +} + +static int stub_on_pend(void* picker, uint64_t jobid) +{ + (void)picker; + + log_event(EVT_ON_PEND, jobid); + + return 0; +} + +static xce_job_picker_iface_t g_stub_picker_iface = { + .init = NULL, + .fini = NULL, + .begin_session = stub_begin_session, + .next_job = stub_next_job, + .end_session = stub_end_session, + .on_dispatch = stub_on_dispatch, + .on_pend = stub_on_pend, + .on_queue_update = stub_on_queue_update, + .on_job_push = stub_on_job_push, + .on_job_state_change = stub_on_job_state_change, + .on_job_update = stub_on_job_update, + .on_config = NULL, /* MAY be NULL (plugin.h) -- not pushed */ + .on_job_rusage = NULL, /* MAY be NULL (plugin.h) -- not pushed */ +}; + +static xce_sched_plugin_t g_stub_plugin + = { .name = "test_stub_picker", .picker_iface = &g_stub_picker_iface }; + +/* ---------------------------------------------------------------------- */ +/* Picker thread: reproduces scheduler.c's sched_job_picker() real */ +/* per-session body verbatim, parameterized on xce_sched_core_t* directly */ +/* instead of xce_scheduler_t* -- single-shot (one session, then returns) */ +/* rather than that function's outer while(1), since this test drives */ +/* exactly one session and joins this thread afterward. */ +/* ---------------------------------------------------------------------- */ + +static void* test_picker_thread_fn(void* arg) +{ + xce_sched_core_t* core = (xce_sched_core_t*)arg; + xce_sched_plugin_t* job_picker = NULL; + int rc; + int go = 0; + uint64_t jobid; + + rc = sched_resolve_job_picker(core, &job_picker); + if (rc < 0) { + return NULL; + } + + cbuf_get(&(core->wbuf), &go); + + job_picker->picker_iface->begin_session(job_picker); + + for (;;) { + xce_sched_job_t* job = job_picker->picker_iface->next_job(job_picker); + + if (job == NULL) { + break; + } + + jobid = job->jobid; + cbuf_put(&(core->jbuf), &jobid); + } + + jobid = 0; + cbuf_put(&(core->jbuf), &jobid); + + job_picker->picker_iface->end_session(job_picker); + + return NULL; +} + +/* ---------------------------------------------------------------------- */ +/* Fixture (same conventions as test_core_events.c/test_core_session.c: */ +/* real xce_sched_core_init(), the stub picker key_add'ed directly into */ +/* core.plugins, fixtures never torn down). */ +/* ---------------------------------------------------------------------- */ + +static void fixture_init(xce_sched_core_t* core, xce_sched_plugin_t* plugin) +{ + int rc; + + rc = xce_sched_core_init(core); + TEST_ASSERT_EQUAL_INT(0, rc); + + lock(core->plugins, RTTI_LOCK_WRITE); + rc = key_add(core->plugins, (void*)plugin->name, (void*)plugin); + unlock(core->plugins); + TEST_ASSERT_GREATER_OR_EQUAL(0, rc); + + core->job_pick_plugin = plugin->name; +} + +static xce_job_t* make_job(uint64_t jobid, uint32_t state, char* queue, + char* user, char* project) +{ + xce_job_t* j = (xce_job_t*)calloc(1, sizeof(*j)); + + j->jobid = jobid; + j->state = state; + j->sub_time = 1000; + j->queue_time = 1000; + j->start_time = 0; + j->submit = (job_submit_data_t*)calloc(1, sizeof(*(j->submit))); + j->submit->queue = queue; + j->submit->user = user; + j->submit->project_name = project; + j->submit->resreq = "type == any"; + j->submit->mintasks = 1; + j->submit->maxtasks = 1; + + return j; +} + +/* Driver-side decision drain -- identical to test_core_session.c's own + * drain_and_dispatch() (not a core-owned function; see that file's header + * comment for why). */ +#define MAX_RECORDED_DECISIONS 8 + +typedef struct { + int calls; + xce_decision_t* seen[MAX_RECORDED_DECISIONS]; +} test_driver_ctx_t; + +static int test_on_decision(void* driver_ctx, xce_decision_t* decision) +{ + test_driver_ctx_t* ctx = (test_driver_ctx_t*)driver_ctx; + + TEST_ASSERT_NOT_NULL(decision); + + if (ctx->calls < MAX_RECORDED_DECISIONS) { + ctx->seen[ctx->calls] = decision; + } + ctx->calls++; + + return 0; +} + +static int drain_and_dispatch(xce_sched_core_t* core) +{ + int delivered = 0; + + for (;;) { + xce_decision_t* decision = NULL; + + cbuf_get(&(core->decisions), &decision); + if (decision == NULL) { + break; + } + + core->on_decision(core->driver_ctx, decision); + delivered++; + } + + return delivered; +} + +/* ---------------------------------------------------------------------- */ +/* The contract test. */ +/* ---------------------------------------------------------------------- */ + +/* + * test_full_cycle_push_go_drain_end_with_precondition_rejections - + * exercises the entire push-events -> start-session -> drain-decisions -> + * end-session cycle end to end, against a real picker thread and a + * stand-in allocator (this function itself), asserting: + * + * 1. xce_core_push_events() (session inactive) genuinely mutates core + * state and fires the stub picker's on_queue_update/on_job_push hooks + * for real, synchronously, before it returns. + * 2. xce_core_go() genuinely starts a session (real picker thread wakes + * up, runs begin_session/next_job/end_session for real). + * 3. Req 5 AC 5: a REENTRANT xce_core_go() call, issued while the session + * started in step 2 is genuinely still active, is rejected with + * -EINVAL as a true no-op. + * 4. Req 5 AC 8: an xce_core_push_events() call issued while that SAME + * genuinely active session is still active is rejected with -EBUSY, + * no event applied. + * 5. The stand-in allocator's on_dispatch()/on_pend() calls and resulting + * make_dispatch_decision()/make_pend_decision() decisions reach the + * driver's on_decision callback, via core->decisions, in the exact + * order the picker produced them. + * 6. xce_core_end_session() genuinely clears session_active. + * 7. Req 5 AC 8 (again): xce_core_push_events() now succeeds, proving the + * precondition toggles both ways across one real cycle, not just + * rejects-then-never-recovers. + * 8. The event log proves the hook firing ORDER this cycle guarantees: + * every push-events-triggered hook precedes begin_session; begin_ + * session precedes the first next_job; next_job(jobid) precedes that + * SAME job's on_dispatch/on_pend (same-item cbuf handoff); the picker + * thread's own calls stay in its own program order (next_job calls, + * then the exhaustion call, then end_session). + */ +static void test_full_cycle_push_go_drain_end_with_precondition_rejections(void) +{ + xce_sched_core_t core; + pthread_t picker_tid; + test_driver_ctx_t driver_ctx; + xce_core_event_t events[2]; + xce_queue_t wire_queue; + xce_job_t* jobs[3]; + xce_job_t* job99; + xce_core_event_t reopen_event; + int rc; + uint64_t jobid; + int delivered; + int idx_queue_update, idx_push10, idx_push11, idx_push12; + int idx_begin, idx_next10, idx_next11, idx_next12, idx_next_exhausted; + int idx_dispatch10, idx_pend11, idx_dispatch12, idx_end; + + memset(&g_stub_state, 0, sizeof(g_stub_state)); + memset(&g_events, 0, sizeof(g_events)); + memset(&driver_ctx, 0, sizeof(driver_ctx)); + + fixture_init(&core, &g_stub_plugin); + + rc = xce_core_set_decision_cb(&core, test_on_decision, &driver_ctx); + TEST_ASSERT_EQUAL_INT(0, rc); + + /* --- Phase 1: push real events while the session is inactive --- */ + memset(&wire_queue, 0, sizeof(wire_queue)); + wire_queue.queue = "normal"; + wire_queue.priority = 1; + wire_queue.status = QUEUE_STAT_OPEN; + + jobs[0] = make_job(10, JOB_STAT_PEND, "normal", "alice", "proj1"); + jobs[1] = make_job(11, JOB_STAT_PEND, "normal", "bob", "proj1"); + jobs[2] = make_job(12, JOB_STAT_PEND, "normal", "carol", "proj1"); + + memset(events, 0, sizeof(events)); + events[0].type = XCE_CORE_EVENT_QUEUE_UPDATE; + events[0].u.queue_update.queue = &wire_queue; + events[1].type = XCE_CORE_EVENT_JOBS_PUSH; + events[1].u.jobs_push.njobs = 3; + events[1].u.jobs_push.jobs = jobs; + + rc = xce_core_push_events(&core, events, 2); + TEST_ASSERT_EQUAL_INT(0, rc); + TEST_ASSERT_EQUAL_INT(3, g_stub_state.ncandidates); + TEST_ASSERT_NOT_NULL(sched_get_job(&core, 10)); + TEST_ASSERT_NOT_NULL(sched_get_job(&core, 11)); + TEST_ASSERT_NOT_NULL(sched_get_job(&core, 12)); + + /* --- Phase 2: spawn the real picker thread, then start the session --- */ + rc = pthread_create(&picker_tid, NULL, test_picker_thread_fn, &core); + TEST_ASSERT_EQUAL_INT(0, rc); + + rc = xce_core_go(&core); + TEST_ASSERT_EQUAL_INT(0, rc); + TEST_ASSERT_EQUAL_INT( + 1, atomic_load_explicit(&(core.session_active), memory_order_acquire)); + + /* Req 5 AC 5: reentrant xce_core_go() while GENUINELY active is + * rejected as a true no-op. */ + rc = xce_core_go(&core); + TEST_ASSERT_EQUAL_INT(-EINVAL, rc); + TEST_ASSERT_EQUAL_INT( + 1, atomic_load_explicit(&(core.session_active), memory_order_acquire)); + + /* Req 5 AC 8: xce_core_push_events() while GENUINELY active is + * rejected with -EBUSY, no event applied. */ + job99 = make_job(99, JOB_STAT_PEND, "normal", "dave", "proj1"); + memset(&reopen_event, 0, sizeof(reopen_event)); + reopen_event.type = XCE_CORE_EVENT_JOBS_PUSH; + reopen_event.u.jobs_push.njobs = 1; + reopen_event.u.jobs_push.jobs = &job99; + + rc = xce_core_push_events(&core, &reopen_event, 1); + TEST_ASSERT_EQUAL_INT(-EBUSY, rc); + TEST_ASSERT_NULL(sched_get_job(&core, 99)); + + /* --- Phase 3: stand-in allocator -- drain jbuf directly (this + * function IS the driver/allocator; see file header comment for why no + * sched_allocator()/xce_scheduler_t dependency is used here) --- */ + for (;;) { + xce_sched_job_t* job; + xce_decision_t* decision = NULL; + + cbuf_get(&(core.jbuf), &jobid); + if (jobid == 0) { + cbuf_put(&(core.decisions), &decision); + break; + } + + job = sched_get_job(&core, jobid); + TEST_ASSERT_NOT_NULL(job); + + if ((jobid % 2) == 0) { + xce_job_alloc_t* alloc + = (xce_job_alloc_t*)calloc(1, sizeof(*alloc)); + alloc->jobid = jobid; + g_stub_picker_iface.on_dispatch(&g_stub_plugin, jobid); + decision = make_dispatch_decision(alloc); + } else { + xce_pendreason_t* reason = make_pendreason(jobid); + g_stub_picker_iface.on_pend(&g_stub_plugin, jobid); + decision = make_pend_decision(reason); + } + + TEST_ASSERT_NOT_NULL(decision); + cbuf_put(&(core.decisions), &decision); + } + + /* Full synchronization point: everything the picker thread touched + * (g_stub_state, its own event-log entries) is now visible here. */ + rc = pthread_join(picker_tid, NULL); + TEST_ASSERT_EQUAL_INT(0, rc); + + /* --- Phase 4: driver drains decisions via the registered callback, + * in allocator-output order --- */ + delivered = drain_and_dispatch(&core); + TEST_ASSERT_EQUAL_INT(3, delivered); + TEST_ASSERT_EQUAL_INT(3, driver_ctx.calls); + + TEST_ASSERT_EQUAL_INT(XCE_DECISION_DISPATCH, driver_ctx.seen[0]->op); + TEST_ASSERT_EQUAL_UINT64(10, driver_ctx.seen[0]->u.alloc->jobid); + + TEST_ASSERT_EQUAL_INT(XCE_DECISION_PEND, driver_ctx.seen[1]->op); + TEST_ASSERT_EQUAL_UINT64(11, driver_ctx.seen[1]->u.reason->jobid); + + TEST_ASSERT_EQUAL_INT(XCE_DECISION_DISPATCH, driver_ctx.seen[2]->op); + TEST_ASSERT_EQUAL_UINT64(12, driver_ctx.seen[2]->u.alloc->jobid); + + /* --- Phase 5: end the session --- */ + rc = xce_core_end_session(&core); + TEST_ASSERT_EQUAL_INT(0, rc); + TEST_ASSERT_EQUAL_INT( + 0, atomic_load_explicit(&(core.session_active), memory_order_acquire)); + + /* Req 5 AC 8: push_events succeeds again now that the session has + * GENUINELY ended -- the precondition toggles both ways across one + * real cycle. */ + rc = xce_core_push_events(&core, &reopen_event, 1); + TEST_ASSERT_EQUAL_INT(0, rc); + TEST_ASSERT_NOT_NULL(sched_get_job(&core, 99)); + + /* --- Phase 6: hook firing order/timing, from the event log --- */ + idx_queue_update = find_log_index(EVT_ON_QUEUE_UPDATE, 0); + idx_push10 = find_log_index(EVT_ON_JOB_PUSH, 10); + idx_push11 = find_log_index(EVT_ON_JOB_PUSH, 11); + idx_push12 = find_log_index(EVT_ON_JOB_PUSH, 12); + idx_begin = find_log_index(EVT_BEGIN_SESSION, 0); + idx_next10 = find_log_index(EVT_NEXT_JOB, 10); + idx_next11 = find_log_index(EVT_NEXT_JOB, 11); + idx_next12 = find_log_index(EVT_NEXT_JOB, 12); + idx_next_exhausted = find_log_index(EVT_NEXT_JOB, 0); + idx_dispatch10 = find_log_index(EVT_ON_DISPATCH, 10); + idx_pend11 = find_log_index(EVT_ON_PEND, 11); + idx_dispatch12 = find_log_index(EVT_ON_DISPATCH, 12); + idx_end = find_log_index(EVT_END_SESSION, 0); + + TEST_ASSERT_TRUE(idx_queue_update >= 0); + TEST_ASSERT_TRUE(idx_push10 >= 0); + TEST_ASSERT_TRUE(idx_push11 >= 0); + TEST_ASSERT_TRUE(idx_push12 >= 0); + TEST_ASSERT_TRUE(idx_begin >= 0); + TEST_ASSERT_TRUE(idx_next10 >= 0); + TEST_ASSERT_TRUE(idx_next11 >= 0); + TEST_ASSERT_TRUE(idx_next12 >= 0); + TEST_ASSERT_TRUE(idx_next_exhausted >= 0); + TEST_ASSERT_TRUE(idx_dispatch10 >= 0); + TEST_ASSERT_TRUE(idx_pend11 >= 0); + TEST_ASSERT_TRUE(idx_dispatch12 >= 0); + TEST_ASSERT_TRUE(idx_end >= 0); + + /* push_events (session inactive) hooks fire in push order, ALL of + * them strictly before begin_session (the wbuf handoff is the + * cross-thread proof: begin_session only runs after xce_core_go()'s + * cbuf_put(&wbuf) is observed by the picker thread's cbuf_get). */ + TEST_ASSERT_TRUE(idx_queue_update < idx_push10); + TEST_ASSERT_TRUE(idx_push10 < idx_push11); + TEST_ASSERT_TRUE(idx_push11 < idx_push12); + TEST_ASSERT_TRUE(idx_push12 < idx_begin); + + /* begin_session precedes the first next_job (picker thread's own + * program order). */ + TEST_ASSERT_TRUE(idx_begin < idx_next10); + + /* Picker thread's own next_job() calls stay in its own program order, + * ending with the NULL-returning exhaustion call, then end_session. */ + TEST_ASSERT_TRUE(idx_next10 < idx_next11); + TEST_ASSERT_TRUE(idx_next11 < idx_next12); + TEST_ASSERT_TRUE(idx_next12 < idx_next_exhausted); + TEST_ASSERT_TRUE(idx_next_exhausted < idx_end); + + /* Per-job same-item cbuf handoff: next_job(jobid) precedes that SAME + * job's post-allocation hook. */ + TEST_ASSERT_TRUE(idx_next10 < idx_dispatch10); + TEST_ASSERT_TRUE(idx_next11 < idx_pend11); + TEST_ASSERT_TRUE(idx_next12 < idx_dispatch12); +} + +int main(void) +{ + UNITY_BEGIN(); + RUN_TEST(test_full_cycle_push_go_drain_end_with_precondition_rejections); + return UNITY_END(); +}