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

test(scheduler-core): add core contract test for full push-go-drain-end cycle (task 4.6)

Browse source 
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 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_019dw5WnN3DXStaEaPYfACU1
This commit is contained in:
Rongsong Shen 2026-08-04 04:05:57 +00:00
parent c5432f7786
commit 19d8df3532
Signed by: rshen
SSH key fingerprint: SHA256:8Vf2pZ3h6rJa7EhGK+WaAQ9Mta/apZvteW6g9U/WKKk
3 changed files with 768 additions and 1 deletions
Download patch file Download diff file Expand all files Collapse all files

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

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

743
src/scheduler/testsuite/test_core_full_cycle.c Normal file
View file

@ -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 <errno.h>
#include <pthread.h>
#include <stdatomic.h>
#include <stdint.h>
#include <stdlib.h>
#include <string.h>
#include <unity/unity.h>
#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();
}