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

feat(scheduler-sim-live-allocation): wire simulation mode onto shared picker/allocator (task 4.1)

Browse source 
Replaces sim_picker_thread_fn() (the always-PEND placeholder) with the
real sched_job_picker()/sched_allocator() threads (task 3.1), spawned
against &(sim->core) in sched_sim_driver_run(). Simulation mode now
genuinely uses the jbuf capacity-1 rendezvous for the first time
(confirmed still exactly one producer/one consumer). Adds
sched_election.c to sched_sim_isolation_lib.

Fixes bug-1829 (blocking, only reachable once sim_allocator() actually
called sched_make_allocation() for the first time): core->seqjob_
allocator/parjob_allocator were never bootstrapped for simulation mode
at all. Fixed via new sim_alloc.c (sim_bootstrap_seqjob_allocator()/
sim_bootstrap_parjob_allocator()), mirroring the sim_picker.c/
sim_limit.c precedent.

That fix's own workaround -- a throwaway xce_scheduler_t wrapper
satisfying seqjob/parjob's init_fn -- has a documented, narrow side
effect: prepare_allocation_hook fan-out during a live seqjob/parjob
dispatch attempt reaches the wrapper's own separate plugin registry,
not sim->core.plugins, so a plugin registered only against the real
sim core (the limit plugin) isn't reached for host-scoped matching
during a live dispatch yet. Documented in design.md's new "Known Seam
Gap" section; task 4.3 added to tasks.md to close it before task 5.2's
acceptance test can pass for the host-scoped case.

TDD: new end-to-end test drives a real .replay with an empty decisions
array (ruling out reconciliation) through the real thread pair,
proving a genuine DISPATCH from live resource availability. 25
pre-existing white-box sim/repl test files given link-satisfying
stubs reproducing the OLD always-PEND contract unchanged -- their own
assertions are untouched (task 5.3's job to reconcile deeper).

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01RLSNJhNoULvVcUhXEFafQC
This commit is contained in:
Rongsong Shen 2026-09-19 04:50:40 +00:00
parent 4195331b47
commit 08df64930b
No known key found for this signature in database
GPG key ID: 5BCDAAE0123E9FE2
33 changed files with 3809 additions and 92 deletions
Download patch file Download diff file Expand all files Collapse all files

524
src/scheduler/testsuite/test_sim_live_allocation.c Normal file
View file

@ -0,0 +1,524 @@
/*
* test_sim_live_allocation.c - scheduler-sim-live-allocation spec, task 4.1
* (Req 1.1, 1.2, 3.3, 4.2, 4.3, 5.1, boundary: Unified Picker/Allocator
* Threads, Sim Driver).
*
* Task 4.1 replaces simulation mode's own always-PEND placeholder thread
* (sim_picker_thread_fn(), now removed from sim_loop.c) with the SAME
* sched_job_picker()/sched_allocator() thread pair (sched_election.c, task
* 3.1) normal mode's own sched_init() spawns -- sched_sim_driver_run()
* (sim_loop.c) now makes two pthread_create() calls against &(sim->core)
* instead of one. This file is the Observable this task's own text
* requires: "a simulation-mode session, driven from a real replay with no
* changes to the replay format, produces a genuine dispatch or pend
* decision for a job based on its actual resource availability, instead of
* unconditionally pending it."
*
* ── Link style: real scheduler_srcs, not a white-box #include ────────────
* Unlike most sim-family tests in this directory (which #include individual
* .c files directly to reach `static` functions or to avoid pulling in
* normal-mode-only machinery), this file links the REAL, unmodified
* `scheduler_srcs` object list (minus src/main.c) -- the exact technique
* test_sched_election_sim_fixture.c's own meson.build entry already
* establishes (see testsuite/meson.build, "same real-scheduler_srcs link
* technique"). This is deliberate: sched_allocator() (sched_election.c)
* calls the REAL sched_make_allocation()/sched_commit_decision()
* (scheduler.c), which in turn reach the REAL seqjob_allocator.c plugin --
* none of that is `static` or reachable via a small, hand-picked white-box
* #include set the way sim_loop.c's own internals are. Linking the real
* scheduler_srcs gets the entire real allocation chain "for free", with no
* stubs standing in for genuine resource-matching logic -- the one thing
* this test exists to prove is genuinely exercised.
*
* ── Why the fixture cannot simply replay a normal-mode recording ─────────
* sim_reconcile_unaccounted_jobs() (sim_loop.c) triggers, unconditionally,
* on ANY job in core->jobs with `accounts == NULL` every round -- including
* a job freshly pushed via a JOBS_PUSH event earlier in the SAME round's
* file position, not merely one "already resolved before this replay run
* started tracking it live" (see test_sim_contention_parity.c's own header
* comment for the identical concern in a sibling test). If the round's own
* ReplaySession carried a matching recorded DISPATCH decision for this
* job, reconciliation -- which runs strictly BEFORE xce_core_signal_go()
* unblocks the live picker each round -- would claim the job first, and
* this test would silently end up proving Recorded-Decision Reconciliation
* instead of the NEW live picker/allocator path. This fixture therefore
* hand-builds a `.replay` stream whose lone ReplaySession carries an EMPTY
* decisions list (append_replay_session() below, same "leave decisions at
* new_ReplaySession()'s zero-value default" technique test_sim_event_loop.c
* already establishes) -- guaranteeing reconciliation finds nothing to
* reconstruct and the job reaches the live sched_job_picker()/
* sched_allocator() pair genuinely.
*
* ── Fixture shape ──────────────────────────────────────────────────────
* A minimal header+snapshot (via a real, throwaway xce_scheduler_t +
* sched_record_start(), same technique test_sim_event_loop.c's/
* test_sim_resource_contention.c's own write_header_and_snapshot()/
* fixture_init() establish, discarded immediately after producing the
* bytes -- never confused with the actual xce_sched_sim_driver_t under
* test below) records job_pick_plugin="builtin_fcfs" (threaded onto
* ReplayHeader.schedPolicy, then back onto sim->core.job_pick_plugin by
* sched_sim_driver_start(), sim.h's own doc comment) with an otherwise
* empty core (no queues/workers/jobs/resources baked into the snapshot
* itself -- avoids any snapshot-save/restore field-completeness risk for
* data this test does not need captured that way). Everything the job
* needs is instead pushed as real, wire-encoded ReplayEvent messages
* immediately after the snapshot -- resourcesPush ("slots"), workersPush
* (one worker with real spare "slots" capacity, register_slots_resource()/
* make_wire_worker()'s own proven shape from test_parjob_e2e_multislot.c),
* queueUpdate, then jobsPush (one job, mintasks=maxtasks=1, resreq
* "select[type==any] rusage[mem=1]" -- test_sched_election_sim_fixture.c's
* own proven-to-parse-as-bare-PRIMARY resreq text) -- exercising the real,
* already-proven-correct JOBS_PUSH ingest path
* (xce_core_push_events()->core_build_job(), which resreq-maxtasks-wiring
* already confirmed correctly threads submit.mintasks/maxtasks into
* primary->resreq->minTasks/maxTasks for this exact event type -- no
* mintasks/maxtasks workaround needed here, unlike a job built directly via
* make_sched_job(), task 3.1's own documented gap).
*
* ── bug-1598/1632/1763 fixture-safety requirement (task 4.1's own text) ──
* sched_sim_driver_run() spawns TWO real, never-joined pthreads (the
* picker, now genuinely sched_job_picker(), AND the allocator,
* sched_allocator()) that read `&(sim->core)` for the rest of the process's
* life -- doubling the surface area bug-1598's own recurring "stack-local
* sim driver instance outlives a never-joined thread" bug class already
* documented three times over (bug-1598, bug-1632, bug-1763, the last still
* NOT FIXED and explicitly out of scope elsewhere). g_sim below is declared
* `static` (file/process lifetime, zero-initialized by the C runtime before
* main() even runs), NOT stack-local inside the test function -- so even
* though this binary's single RUN_TEST() (matching test_sched_election_
* sim_fixture.c's own "exactly one RUN_TEST(), process exits immediately
* after" convention, belt-and-suspenders here) returns and the process
* exits shortly after, the two spawned threads never dereference a stack
* frame that could have already been reclaimed -- positively confirmed,
* not assumed safe by analogy, per this task's own explicit TDD
* requirement. sched_sim_driver_start() (sim.c) unconditionally
* memset()s the whole struct as its own first action (same reasoning
* bug-1598's own fix documents for why `static` is safe here: this test
* binary calls sched_sim_driver_run() exactly once, so no second call's
* memset can ever stomp a still-live zombie thread's state from a first
* call), so this file does not separately memset g_sim itself.
*/
#include <errno.h>
#include <limits.h>
#include <stdint.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>
#include <unity/unity.h>
#include "controller.capnp.h"
#include "recording.h"
#include "resobj.h"
#include "scheduler.h"
#include "sim.h"
#include "xstring.h"
/* write_capnp_message (recording.c) has no public header declaration --
* every existing white-box test reaches it via a direct #include of
* recording.c into the same TU. This file instead LINKS the real
* scheduler_srcs (see this file's own header comment), so it is declared
* locally instead, matching this codebase's established "extern declared
* where consumed" idiom (sim_loop.c's own local externs for msg_root_
* has_shape()/sim_decision_eq()/sched_job_picker()/sched_allocator() follow
* the identical pattern for the identical reason). */
extern int write_capnp_message(FILE* f, struct capn* c);
#define WORKER_NAME "sim_live_alloc_worker"
#define QUEUE_NAME "normal"
#define TEST_RESREQ ((char*)"select[type==any] rusage[mem=1]")
#define TEST_JOBID ((uint64_t)940101ULL)
#define WORKER_SLOTS 4.0f
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_simlivealloc%s_XXXXXX", suffix);
TEST_ASSERT_NOT_NULL(mkdtemp(tmpl));
TEST_ASSERT_TRUE(strlen(tmpl) < out_len);
strcpy(out, tmpl);
}
/* make_wire_worker - identical proven shape to
* test_parjob_e2e_multislot.c's own make_wire_worker(): a real xce_worker_t
* wire payload carrying one "slots" resval, ready for a real
* XCE_CORE_EVENT_WORKERS_PUSH ingest. */
static xce_worker_t* make_wire_worker(const char* name, float slots)
{
xce_worker_t* w = (xce_worker_t*)calloc(1, sizeof(*w));
xce_val_t* val = (xce_val_t*)calloc(1, sizeof(*val));
xce_name_val_t* nv = (xce_name_val_t*)calloc(1, sizeof(*nv));
xce_name_val_t** nv_ptrs = (xce_name_val_t**)calloc(1, sizeof(*nv_ptrs));
TEST_ASSERT_NOT_NULL(w);
TEST_ASSERT_NOT_NULL(val);
TEST_ASSERT_NOT_NULL(nv);
TEST_ASSERT_NOT_NULL(nv_ptrs);
val->which = Value_u_fvalue;
val->u.fvalue = slots;
nv->name = strdup("slots");
nv->val = val;
nv_ptrs[0] = nv;
w->name = strdup(name);
w->status = 0;
w->nresvals_to_sched = 1;
w->resvals_to_sched = nv_ptrs;
return w;
}
static xce_queue_t* make_wire_queue(const char* name, int32_t priority)
{
xce_queue_t* q = (xce_queue_t*)calloc(1, sizeof(*q));
TEST_ASSERT_NOT_NULL(q);
q->queue = strdup(name);
q->status = QUEUE_STAT_OPEN;
q->priority = priority;
return q;
}
/* make_push_job - identical proven shape to test_parjob_e2e_multislot.c's/
* test_maxtasks_wiring_sim_driver_parity.c's own make_push_job(): a real
* xce_job_t/job_submit_data_t pair ready for a real
* XCE_CORE_EVENT_JOBS_PUSH ingest (core_build_job()), which
* resreq-maxtasks-wiring already proved correctly threads
* submit.mintasks/maxtasks into the parsed primary resreq for this exact
* event type. */
static xce_job_t* make_push_job(uint64_t jobid, char* resreq_str, char* user,
char* queue, char* project, uint32_t mintasks,
uint32_t maxtasks)
{
xce_job_t* j = (xce_job_t*)calloc(1, sizeof(*j));
TEST_ASSERT_NOT_NULL(j);
j->jobid = jobid;
j->state = JOB_STAT_PEND;
j->sub_time = 1000;
j->queue_time = 1000;
j->submit = (job_submit_data_t*)calloc(1, sizeof(*(j->submit)));
TEST_ASSERT_NOT_NULL(j->submit);
j->submit->queue = queue;
j->submit->user = user;
j->submit->project_name = project;
j->submit->resreq = resreq_str;
j->submit->mintasks = mintasks;
j->submit->maxtasks = maxtasks;
return j;
}
/* append_resourcespush_event/append_workerspush_event/
* append_queueupdate_event/append_jobspush_event/append_replay_session/
* append_end_of_stream_marker - same encode_<Type>_ptr()/
* write_capnp_message() technique test_maxtasks_wiring_sim_driver_parity.c's
* own append_jobspush_event()/test_sim_event_loop.c's own
* append_jobstatechange_event()/append_replay_session()/
* append_end_of_stream_marker() already prove works, extended here to the
* two sibling push event types (resourcesPush/workersPush) this fixture
* also needs. */
static void append_resourcespush_event(FILE* f, uint64_t ts,
xce_resource_t** resources, int n)
{
xce_resource_data_t wrap;
replay_event_t src;
struct capn c;
struct capn_segment* cs;
ReplayEvent_ptr p;
memset(&wrap, 0, sizeof(wrap));
wrap.n_total = n;
wrap.count = n;
wrap.resources = resources;
memset(&src, 0, sizeof(src));
src.ts = ts;
src.body_which = ReplayEvent_body_resourcesPush;
src.body.resources_push = &wrap;
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_workerspush_event(FILE* f, uint64_t ts,
xce_worker_t** workers, int n)
{
xce_workers_data_t wrap;
replay_event_t src;
struct capn c;
struct capn_segment* cs;
ReplayEvent_ptr p;
memset(&wrap, 0, sizeof(wrap));
wrap.n_total = n;
wrap.count = n;
wrap.workers = workers;
memset(&src, 0, sizeof(src));
src.ts = ts;
src.body_which = ReplayEvent_body_workersPush;
src.body.workers_push = &wrap;
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_queueupdate_event(FILE* f, uint64_t ts, xce_queue_t* q)
{
replay_event_t src;
struct capn c;
struct capn_segment* cs;
ReplayEvent_ptr p;
memset(&src, 0, sizeof(src));
src.ts = ts;
src.body_which = ReplayEvent_body_queueUpdate;
src.body.queue_update = q;
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_jobspush_event(FILE* f, uint64_t ts, xce_job_t** jobs,
int njobs)
{
xce_jobs_data_t wrap;
replay_event_t src;
struct capn c;
struct capn_segment* cs;
ReplayEvent_ptr p;
memset(&wrap, 0, sizeof(wrap));
wrap.n_total = njobs;
wrap.count = njobs;
wrap.jobs = jobs;
memset(&src, 0, sizeof(src));
src.ts = ts;
src.body_which = ReplayEvent_body_jobsPush;
src.body.jobs_push = &wrap;
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);
}
/* append_replay_session - a minimal ReplaySession with only beginRound set
* -- decisions/summary deliberately left at new_ReplaySession()'s
* zero-value defaults (empty decisions list). See this file's own header
* comment ("Why the fixture cannot simply replay a normal-mode recording")
* for why an EMPTY decisions list is load-bearing here, not incidental. */
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);
}
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);
}
/* write_header_and_snapshot - produces a valid ReplayHeader+SnapshotEntry
* prefix at a fresh temp path via the real sched_record_start(), same
* technique test_sim_event_loop.c's own write_header_and_snapshot()/
* test_sim_resource_contention.c's own fixture_init()+sched_record_start()
* pair establish -- but deliberately does NOT call sched_record_stop() (see
* test_sim_event_loop.c's own identical rationale: that call would append
* its own endSession marker immediately, before this fixture's own events
* ever get appended). The throwaway xce_scheduler_t built here exists
* SOLELY to produce these bytes -- it is discarded the moment this function
* returns, never confused with g_sim (the actual xce_sched_sim_driver_t
* under test) below. Registers the REAL builtin_fcfs plugin (not a stub)
* so the recorded job_pick_plugin="builtin_fcfs" genuinely matches what
* sim_bootstrap_picker() (bug-1567, sim_picker.c) looks for -- the same
* "real builtin_fcfs, real worker, no stub" precedent test_sim_resource_
* contention.c's own fixture_init() already establishes. */
static FILE* write_header_and_snapshot(char* out_path, size_t out_path_len)
{
char record_dir[PATH_MAX];
xce_scheduler_t sched;
xce_sched_plugin_t* fcfs;
int rc;
make_tmp_dir(record_dir, sizeof(record_dir), "_r");
memset(&sched, 0, sizeof(sched));
mylock_init(&(sched.record_lock));
rc = xce_sched_core_init(&(sched.core));
TEST_ASSERT_EQUAL_INT(0, rc);
fcfs = builtin_fcfs_new_plugin();
TEST_ASSERT_NOT_NULL(fcfs);
rc = plugin_register_core(&(sched.core), fcfs);
TEST_ASSERT_TRUE(rc >= 0);
sched.core.job_pick_plugin = (char*)"builtin_fcfs";
sched.record_dir = record_dir;
rc = sched_record_start(&sched, "live_alloc.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;
}
/* ---------------------------------------------------------------------- */
/* The sim driver under test -- static storage, see this file's own header */
/* comment ("bug-1598/1632/1763 fixture-safety requirement") for why. */
/* ---------------------------------------------------------------------- */
static xce_sched_sim_driver_t g_sim;
/* ---------------------------------------------------------------------- */
/* The test */
/* ---------------------------------------------------------------------- */
/*
* Observable (task 4.1's own text): "a simulation-mode session, driven from
* a real replay with no changes to the replay format, produces a genuine
* dispatch or pend decision for a job based on its actual resource
* availability, instead of unconditionally pending it." This fixture gives
* the job real, sufficient resource availability (a worker with 4.0 free
* "slots" against a 1-task job's auto-injected 1-slot-per-task want, per
* test_sched_election_sim_fixture.c's own documented convention) --
* asserting the job reaches JOB_STAT_RUN with a real, non-NULL allocation
* is the strongest possible proof this task's Observable calls for: the
* removed sim_picker_thread_fn() placeholder could structurally NEVER
* produce this outcome, only PEND, regardless of resource availability.
*/
static void
test_sim_driver_run_dispatches_a_job_with_available_resources(void)
{
char path[PATH_MAX];
FILE* f;
xce_resource_t* slots_res;
xce_resource_t* res_ptrs[1];
xce_worker_t* worker_ptrs[1];
xce_queue_t* queue;
xce_job_t* job_ptrs[1];
sched_sim_args_t args;
xce_sched_job_t* sjob;
int rc;
f = write_header_and_snapshot(path, sizeof(path));
slots_res = (xce_resource_t*)calloc(1, sizeof(*slots_res));
TEST_ASSERT_NOT_NULL(slots_res);
slots_res->name = strdup("slots");
res_ptrs[0] = slots_res;
append_resourcespush_event(f, 1001ULL, res_ptrs, 1);
worker_ptrs[0] = make_wire_worker(WORKER_NAME, WORKER_SLOTS);
append_workerspush_event(f, 1002ULL, worker_ptrs, 1);
queue = make_wire_queue(QUEUE_NAME, 1);
append_queueupdate_event(f, 1003ULL, queue);
job_ptrs[0] = make_push_job(TEST_JOBID, TEST_RESREQ, (char*)"alice",
(char*)QUEUE_NAME, (char*)"proj1", 1, 1);
append_jobspush_event(f, 1004ULL, job_ptrs, 1);
/* Empty decisions -- see this file's own header comment for why this
* is load-bearing, not incidental. */
append_replay_session(f, 2000ULL);
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;
rc = sched_sim_driver_run(&g_sim, &args);
TEST_ASSERT_EQUAL_INT_MESSAGE(
0, rc, "sched_sim_driver_run() must complete this fixture's single "
"round and reach its own end-of-stream marker");
sjob = sched_get_job(&(g_sim.core), TEST_JOBID);
TEST_ASSERT_NOT_NULL_MESSAGE(
sjob, "the job pushed via the real JOBS_PUSH ingest path must be "
"present in the simulation driver's own core state");
TEST_ASSERT_EQUAL_UINT_MESSAGE(
JOB_STAT_RUN, sjob->state,
"the shared sched_job_picker()/sched_allocator() pair (spawned "
"directly against simulation mode's own core state by "
"sched_sim_driver_run(), scheduler-sim-live-allocation task 4.1) "
"must genuinely DISPATCH a job whose resource request is satisfied "
"by a real, replayed worker -- not unconditionally PEND it the way "
"the removed sim_picker_thread_fn() placeholder always did");
TEST_ASSERT_NOT_NULL_MESSAGE(
sjob->allocs, "a genuine DISPATCH must leave a real, non-NULL "
"allocation behind (sched_commit_decision()'s "
"DISPATCH branch, scheduler.c) -- not merely a "
"state flip with no resource assignment behind it");
}
int main(void)
{
UNITY_BEGIN();
RUN_TEST(test_sim_driver_run_dispatches_a_job_with_available_resources);
return UNITY_END();
}