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 <noreply@anthropic.com>
This commit is contained in:
parent
0649050ae8
commit
8607348c9b
4 changed files with 289 additions and 1 deletions
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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);
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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'])
|
||||
|
|
|
|||
263
src/scheduler/testsuite/test_scheduler_dispatch_commit.c
Normal file
263
src/scheduler/testsuite/test_scheduler_dispatch_commit.c
Normal file
|
|
@ -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 <errno.h>
|
||||
#include <stdlib.h>
|
||||
#include <string.h>
|
||||
|
||||
#include <unity/unity.h>
|
||||
|
||||
#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();
|
||||
}
|
||||
Loading…
Add table
Add a link
Reference in a new issue