From 8607348c9b8603d80c74aa4a0078b8a9a05aab11 Mon Sep 17 00:00:00 2001 From: Rongsong Shen Date: Sun, 23 Aug 2026 08:25:35 +0000 Subject: [PATCH] feat(scheduler-dispatch-commit): scheduler transitions job to RUN at dispatch commit (task 2.1) sched_commit_decision()'s DISPATCH branch now sets job->state = JOB_STAT_RUN in the same operation that already commits job->allocs/job->accounts, per Requirement 1.1/1.2/1.3. sched_account_jobs()'s existing JOB_IS_RUN gate is unmodified -- this closes bug-1624 by giving it the input it was missing: a dispatched job now survives a round boundary in resource-in-use totals regardless of how many rounds pass before worker confirmation arrives. Regression test drives the real, unmodified sched_account_jobs() across a simulated round boundary (resInUse reset, no confirmation event) and independently reverting the one-line change reproduces bug-1624 exactly. Co-Authored-By: Claude Sonnet 5 --- .../specs/scheduler-dispatch-commit/tasks.md | 2 +- src/scheduler/src/scheduler.c | 1 + src/scheduler/testsuite/meson.build | 24 ++ .../test_scheduler_dispatch_commit.c | 263 ++++++++++++++++++ 4 files changed, 289 insertions(+), 1 deletion(-) create mode 100644 src/scheduler/testsuite/test_scheduler_dispatch_commit.c diff --git a/.kiro/specs/scheduler-dispatch-commit/tasks.md b/.kiro/specs/scheduler-dispatch-commit/tasks.md index 30d5f244..40f11e3a 100644 --- a/.kiro/specs/scheduler-dispatch-commit/tasks.md +++ b/.kiro/specs/scheduler-dispatch-commit/tasks.md @@ -24,7 +24,7 @@ - [ ] 2. Core: scheduler transitions a job to running state at dispatch-commit time - No dependency on Task 1 — scheduler-side files only, independent of the controller-side foundation work above. -- [ ] 2.1 (P) Scheduler marks a job as running in the same operation that commits its dispatch decision +- [x] 2.1 (P) Scheduler marks a job as running in the same operation that commits its dispatch decision - When the scheduler commits a decision to dispatch a job, the job's state becomes running immediately, in the same operation that already records the job's resource allocation — not waiting for any later confirmation - The scheduler's existing resource-in-use accounting, which only counts jobs already in a running state, requires no changes — this task provides the input it was missing - Observable completion: a job dispatched in one round appears in resource-in-use totals starting that same round, and continues to appear in every subsequent round regardless of how many rounds pass before any confirmation arrives diff --git a/src/scheduler/src/scheduler.c b/src/scheduler/src/scheduler.c index 66d9f252..6f41d05c 100644 --- a/src/scheduler/src/scheduler.c +++ b/src/scheduler/src/scheduler.c @@ -1442,6 +1442,7 @@ int sched_commit_decision(xce_scheduler_t* sched, xce_decision_t* decision) job = sched_get_job(&(sched->core), decision->u.alloc->jobid); job->allocs = decision->u.alloc; job->accounts = make_job_account(job->allocs); + job->state = JOB_STAT_RUN; return sched_commit_job_alloc(&(sched->core), job); } diff --git a/src/scheduler/testsuite/meson.build b/src/scheduler/testsuite/meson.build index a72b00a6..755b2333 100644 --- a/src/scheduler/testsuite/meson.build +++ b/src/scheduler/testsuite/meson.build @@ -1102,6 +1102,27 @@ sched_scheduler_decision_redelivery_test = executable( link_args: ['-Wl,--wrap=xcecli_jobdispatch'], install: false) +# scheduler-dispatch-commit spec, task 2.1 (Req 1.1/1.2/1.3): proves +# sched_commit_decision()'s DISPATCH branch transitions job->state to +# JOB_STAT_RUN in the same operation that already commits job->allocs/ +# job->accounts. Same real-scheduler_srcs link technique as +# sched_scheduler_connect_test above, but no `-Wl,--wrap=` is needed: this +# file never calls sched_init()/sched_fini(), so none of the xcecli_* calls +# those make are ever reached. +scheduler_dispatch_commit_test_srcs = ['test_scheduler_dispatch_commit.c'] +foreach f : scheduler_srcs + if f != 'src/main.c' + scheduler_dispatch_commit_test_srcs += ['../' + f] + endif +endforeach + +sched_scheduler_dispatch_commit_test = executable( + 'sched_scheduler_dispatch_commit_test', + scheduler_dispatch_commit_test_srcs, + include_directories:[incdir], + dependencies: [scheduler_deps, unity_dep], + 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', @@ -1246,3 +1267,6 @@ test('scheduler backpressure test', test('scheduler decision redelivery test', sched_scheduler_decision_redelivery_test, suite: 'scheduler', env: ['ASAN_OPTIONS=detect_leaks=0']) +test('scheduler dispatch commit test', + sched_scheduler_dispatch_commit_test, suite: 'scheduler', + env: ['ASAN_OPTIONS=detect_leaks=0']) diff --git a/src/scheduler/testsuite/test_scheduler_dispatch_commit.c b/src/scheduler/testsuite/test_scheduler_dispatch_commit.c new file mode 100644 index 00000000..d5f7ee5f --- /dev/null +++ b/src/scheduler/testsuite/test_scheduler_dispatch_commit.c @@ -0,0 +1,263 @@ +/* + * test_scheduler_dispatch_commit.c - scheduler-dispatch-commit spec, task + * 2.1 (Req 1.1, 1.2, 1.3): sched_commit_decision()'s DISPATCH branch + * transitions a job to JOB_STAT_RUN in the same operation that already + * commits job->allocs/job->accounts, so sched_account_jobs()'s existing + * JOB_IS_RUN() gate (scheduler.c, unmodified by this task) picks up a + * dispatched job's resource consumption starting the same round it was + * dispatched, and continues to include it regardless of how many rounds + * elapse before worker confirmation arrives (bug-1624). + * + * Links the REAL production scheduler_srcs (same "whole-scheduler_srcs, + * minus main.c" technique test_scheduler_connect.c/ + * test_scheduler_backpressure.c already establish in meson.build for + * reaching scheduler.c's non-static entry points without white-box + * #including a ~1400-line file -- no existing test in this directory does + * that, for exactly the link-surface reason test_resource_accounting.c's + * own header comment documents). Unlike those two files, this one never + * calls sched_init()/sched_fini() (no controller-join, no picker/allocator + * threads), so it needs none of their `-Wl,--wrap=xcecli_*` interposition: + * sched_commit_decision() itself only touches sched->core and the (empty, + * in this fixture) plugin-hook chain, never sched->apictx. + */ +#include +#include +#include + +#include + +#include "preason.h" +#include "scheduler.h" +#include "xcelimit.h" + +#define WORKER_NAME ((char*)"worker1") +#define JOB_ID 4242ULL +#define RES_NAME "slots" +#define WANT_VALUE 4.0f + +static xce_sched_worker_t* add_worker(xce_sched_core_t* core, char* name) +{ + xce_sched_worker_t* w = (xce_sched_worker_t*)calloc(1, sizeof(*w)); + int rc; + + TEST_ASSERT_NOT_NULL(w); + + w->name = strdup(name); + TEST_ASSERT_NOT_NULL(w->name); + + rc = zarray_init(&(w->resvals), 8, NULL); + TEST_ASSERT_EQUAL_INT(0, rc); + rc = zarray_init(&(w->resInUse), 8, NULL); + TEST_ASSERT_EQUAL_INT(0, rc); + + rc = sched_add_worker(core, w); + TEST_ASSERT_TRUE(rc >= 0); + + return w; +} + +static xce_sched_job_t* add_pending_job(xce_sched_core_t* core, uint64_t jobid) +{ + xce_sched_job_t* job = (xce_sched_job_t*)calloc(1, sizeof(*job)); + int rc; + + TEST_ASSERT_NOT_NULL(job); + + job->jobid = jobid; + job->state = JOB_STAT_PEND; + + rc = sched_add_job(core, job); + TEST_ASSERT_TRUE(rc >= 0); + + return job; +} + +/* Builds the wire-shaped xce_job_alloc_t a DISPATCH decision carries -- same + * field shape every other fixture in this directory building one by hand + * uses (e.g. test_sim_resource_contention.c's own make_dispatch_alloc()). */ +static xce_job_alloc_t* make_dispatch_alloc(uint64_t jobid, char* host, + float value) +{ + xce_alloc_t* res = (xce_alloc_t*)calloc(1, sizeof(*res)); + xce_host_alloc_t* h = (xce_host_alloc_t*)calloc(1, sizeof(*h)); + xce_host_alloc_t** hosts = (xce_host_alloc_t**)calloc(1, sizeof(*hosts)); + xce_alloc_t** allocs = (xce_alloc_t**)calloc(1, sizeof(*allocs)); + xce_job_alloc_t* j = (xce_job_alloc_t*)calloc(1, sizeof(*j)); + + TEST_ASSERT_NOT_NULL(res); + TEST_ASSERT_NOT_NULL(h); + TEST_ASSERT_NOT_NULL(hosts); + TEST_ASSERT_NOT_NULL(allocs); + TEST_ASSERT_NOT_NULL(j); + + res->resname = strdup(RES_NAME); + res->reserve = strdup("run"); + res->instid = 1; + res->value = value; + res->from = 0; + res->duration = 0; + + allocs[0] = res; + + h->host = strdup(host); + h->nallocs = 1; + h->allocs = allocs; + + hosts[0] = h; + + j->jobid = jobid; + j->nhosts = 1; + j->allocs = hosts; + j->nres_shared = 0; + j->allocs_shared = NULL; + + return j; +} + +void setUp(void) { return; } +void tearDown(void) { return; } + +/* + * Requirement 1.1: committing a DISPATCH decision transitions the job to + * JOB_STAT_RUN in the SAME operation that commits job->allocs/job->accounts + * -- the one new line this task adds to sched_commit_decision(), alongside + * the pre-existing allocs/accounts assignment it must not disturb. + */ +static void test_dispatch_commit_transitions_job_to_run(void) +{ + xce_scheduler_t sched; + xce_sched_job_t* job; + xce_job_alloc_t* alloc; + xce_decision_t* decision; + int rc; + + memset(&sched, 0, sizeof(sched)); + rc = xce_sched_core_init(&(sched.core)); + TEST_ASSERT_EQUAL_INT(0, rc); + + add_worker(&(sched.core), WORKER_NAME); + job = add_pending_job(&(sched.core), JOB_ID); + TEST_ASSERT_EQUAL_UINT(JOB_STAT_PEND, job->state); + + alloc = make_dispatch_alloc(JOB_ID, WORKER_NAME, WANT_VALUE); + decision = make_dispatch_decision(alloc); + TEST_ASSERT_NOT_NULL(decision); + + rc = sched_commit_decision(&sched, decision); + TEST_ASSERT_TRUE(rc >= 0); + + /* the new behavior this task adds */ + TEST_ASSERT_EQUAL_UINT(JOB_STAT_RUN, job->state); + + /* the pre-existing behavior this task must not disturb */ + TEST_ASSERT_EQUAL_PTR(alloc, job->allocs); + TEST_ASSERT_NOT_NULL(job->accounts); + TEST_ASSERT_EQUAL_INT(1, job->accounts->nhosts); + TEST_ASSERT_EQUAL_STRING(WORKER_NAME, job->accounts->accounts[0]->host); +} + +/* + * Requirement 1.2 / 1.3 (bug-1624's actual regression test): the + * scheduler's existing sched_account_jobs()/JOB_IS_RUN() gate -- unmodified + * by this task -- naturally includes a dispatched job's resource + * consumption starting the SAME round it was dispatched, and continues to + * include it across a simulated round boundary (resInUse unconditionally + * zeroed, exactly as xce_core_begin_session() does every round) with no + * worker confirmation event in between. Before this task, job->state would + * still read JOB_STAT_PEND at this point, so sched_account_jobs()'s + * !JOB_IS_RUN() branch would skip the job and the final assertion below + * would see 0.0 instead of WANT_VALUE. + */ +static void test_dispatch_commit_survives_round_boundary_before_confirmation(void) +{ + xce_scheduler_t sched; + xce_sched_worker_t* w; + xce_sched_job_t* job; + xce_job_alloc_t* alloc; + xce_decision_t* decision; + xce_resource_t* vres = NULL; + int rc; + int residx; + float vals[1]; + float zero[1] = { 0.0f }; + + memset(&sched, 0, sizeof(sched)); + rc = xce_sched_core_init(&(sched.core)); + TEST_ASSERT_EQUAL_INT(0, rc); + + w = add_worker(&(sched.core), WORKER_NAME); + job = add_pending_job(&(sched.core), JOB_ID); + + /* Register RES_NAME up front, the way real resource config-load does + * before any job is ever charged, rather than relying on + * sched_update_res_accounts()'s own lazy first-use registration path -- + * out of scope for what this test is proving. */ + residx = add_vres(&vres, sched.core.resmgr, RES_NAME, 0); + TEST_ASSERT_TRUE(residx >= 0); + + alloc = make_dispatch_alloc(JOB_ID, WORKER_NAME, WANT_VALUE); + decision = make_dispatch_decision(alloc); + TEST_ASSERT_NOT_NULL(decision); + + rc = sched_commit_decision(&sched, decision); + TEST_ASSERT_TRUE(rc >= 0); + TEST_ASSERT_EQUAL_UINT(JOB_STAT_RUN, job->state); + TEST_ASSERT_NOT_NULL(job->accounts); + TEST_ASSERT_EQUAL_INT(residx, job->accounts->accounts[0]->accounts[0]->residx); + + get_resvals(&(w->resInUse), 1, &residx, vals); + TEST_ASSERT_EQUAL_FLOAT(WANT_VALUE, vals[0]); + + /* simulate the next round boundary */ + put_resvals(&(w->resInUse), 1, &residx, zero); + get_resvals(&(w->resInUse), 1, &residx, vals); + TEST_ASSERT_EQUAL_FLOAT(0.0f, vals[0]); + + rc = sched_account_jobs(&(sched.core)); + TEST_ASSERT_EQUAL_INT(0, rc); + + get_resvals(&(w->resInUse), 1, &residx, vals); + TEST_ASSERT_EQUAL_FLOAT(WANT_VALUE, vals[0]); +} + +/* + * Negative-space guard: this task's new line lives strictly inside the + * XCE_DECISION_DISPATCH branch. A PEND decision must not touch job->state + * at all (still JOB_STAT_PEND) -- proves the fix is scoped to DISPATCH, not + * an unconditional job->state = JOB_STAT_RUN hoisted above the branch. + */ +static void test_pend_decision_does_not_touch_job_state(void) +{ + xce_scheduler_t sched; + xce_sched_job_t* job; + xce_pendreason_t* reason; + xce_decision_t* decision; + int rc; + + memset(&sched, 0, sizeof(sched)); + rc = xce_sched_core_init(&(sched.core)); + TEST_ASSERT_EQUAL_INT(0, rc); + + job = add_pending_job(&(sched.core), JOB_ID); + + reason = make_pendreason(JOB_ID); + TEST_ASSERT_NOT_NULL(reason); + decision = make_pend_decision(reason); + TEST_ASSERT_NOT_NULL(decision); + + rc = sched_commit_decision(&sched, decision); + TEST_ASSERT_TRUE(rc >= 0); + + TEST_ASSERT_EQUAL_UINT(JOB_STAT_PEND, job->state); + TEST_ASSERT_NULL(job->allocs); + TEST_ASSERT_NULL(job->accounts); +} + +int main(void) +{ + UNITY_BEGIN(); + RUN_TEST(test_dispatch_commit_transitions_job_to_run); + RUN_TEST(test_dispatch_commit_survives_round_boundary_before_confirmation); + RUN_TEST(test_pend_decision_does_not_touch_job_state); + return UNITY_END(); +}