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

feat(scheduler-allocator-runner): parjob allocates through the framework runner (task 4.1)

Browse source 
parjob registers sched_allocator_run with parjob_alloc_iface (accepts
maxTasks > 1, bug-1767, plus its plan iterator); its private my_alloc()
and claim helpers are gone. Characterization now pins the intended
changes: a reason for every short host (2/6/5) and the bug-1893 shared
reason. No other test expectation changed.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
This commit is contained in:
Rongsong Shen 2026-09-30 09:40:47 +00:00
parent 0e68ed991a
commit 239bcf95cd
Signed by untrusted user: rshen
GPG key ID: 80533536B0EB651C
4 changed files with 116 additions and 563 deletions
Download patch file Download diff file Expand all files Collapse all files

3
.kiro/specs/scheduler-allocator-runner/tasks.md
View file

@ -57,7 +57,7 @@
- _Requirements: 4.1, 4.2, 4.3, 4.4, 5.4, 5.5, 5.6, 5.7, 7.2_
- [ ] 4. Move both allocators onto the framework routine
- [ ] 4.1 Switch the parallel-job allocator
- [x] 4.1 Switch the parallel-job allocator
- First rewrite the parjob characterization tests' `_today` host-reason assertions to the complete per-host reasons of 3.7 (every selection-matching unclaimed short host of each bucket search) and see them fail against the old code
- Also add the parjob bug-1893 fixture (two shared resources, one short: exactly one host-less reason, naming the short one; today none is recorded) to the parjob characterization tests and see it fail against the old code
- Give parjob its allocator contract with a task-count acceptance guard (more than one task), register the framework routine as its allocation entry point, and remove its private allocation routine and claim helpers
@ -103,3 +103,4 @@
- Req 2.4 error-path test scope (3.1 review): `make_pendreason` -ENOMEM is tested (`--wrap=make_pendreason`). The uncommitted-read, shared-check and uncommitted-add error paths are returned correctly but not tested: the read cannot fail past the input guard (`sched_ctx_get_uncomit_shared` only rejects NULL args), and the other two fail only on a libxce-internal allocation failure (would need malloc-level injection).
- The runner ignores `add_job_preason`'s return value, as the old allocators did (a failed add drops that reason).
- Incremental rebuilds in `src/build.alloc` can miss a file restored after a temporary RED edit if its mtime ends up older than the object; `touch` restored files before the GREEN run (seen in 3.1).
- Stale references for 5.1 (from 4.1 review): test_parjob_combinatorial.c lines ~6, 56, 106, 668, 806 and helper `drive_my_alloc`; scheduler.h:415; sim_alloc.c:48; test_host_iter.c:665; test_parjob_e2e_multislot.c:138 (plus seqjob ones after 4.2).

484
src/scheduler/src/plugins/parjob_allocator.c
View file

@ -62,7 +62,7 @@ static int skip_bad_worker_(xce_sched_worker_t* w)
*
* ctx->worker == NULL (the job-level shared/global-resource pass, Req 4
* AC 7): fires exactly ONCE, from sched_ctx_set_primary() (scheduler.c),
* before my_alloc()'s plan/bucket loop ever runs -- ctx->nhosts/
* before sched_allocator_run()'s plan/bucket loop ever runs -- ctx->nhosts/
* ctx->nslots have not been set by that loop yet at this point, so this
* branch computes its own job-wide scale (1 host, the job's full
* maxTasks) rather than reading them, generalizing seqjob's own
@ -73,7 +73,7 @@ static int skip_bad_worker_(xce_sched_worker_t* w)
* job-level shared-resource check runs").
*
* ctx->worker != NULL (the per-host pass, Req 4 AC 6): fires once per
* candidate host tried, from my_alloc()'s bucket loop via
* candidate host tried, from sched_allocator_run()'s bucket loop via
* sched_try_alloc_on_host_() -> plugin_prepare_allocation(). THE fix
* this task makes: ctx->nslots/ctx->nhosts are read from ctx (set by
* that loop, right before this host was tried, to the CURRENT bucket's
@ -188,9 +188,9 @@ static int parjob_prepare_allocation(void* arg, xce_sched_ctx_t* ctx)
* signature has no ctx parameter to thread it through. Setting
* plan->ctx = ctx on every plan returned from next(), before
* constructing a host_iter from it, is the CALLER's responsibility
* (task 4.2's my_alloc(), not this iterator) -- see alloc_iter.h's
* xce_alloc_plan_t.ctx doc comment and tasks.md's task 2.1 implementation
* note.
* (the framework routine in alloc_runner.c, not this iterator) -- see
* alloc_iter.h's xce_alloc_plan_t.ctx doc comment and tasks.md's task 2.1
* implementation note.
*/
typedef struct {
xce_alloc_plan_t plan;
@ -303,467 +303,22 @@ static void parjob_plan_iter_fini(xce_alloc_plan_iter_t* iter)
iter->private = NULL;
}
/*
* parjob_claim_t - one bucket's successful host claim within a single plan
* attempt (task 4.2, Req 4 AC 2/AC 3). `host_req` is a deep-copy snapshot
* of ctx->alloc_req.host taken at the exact moment sched_try_alloc_on_
* host_() accepted `w` for this bucket -- required because ctx->alloc_req
* is a single scratch buffer plugin_prepare_allocation() resets and
* repopulates for every subsequent host tried (requirements.md's
* "Rollback scope"), so it cannot be read lazily after the bucket loop
* moves on. Used both for rollback (sched_ctx_uncommit_host() needs the
* exact committed xvec_idval_t back) and for the final multi-host merge
* (each claimed bucket's own per-host resource want).
*/
typedef struct {
xce_sched_worker_t* w;
xvec_idval_t host_req;
} parjob_claim_t;
static int parjob_is_claimed_(parjob_claim_t* claimed, int nclaimed,
xce_sched_worker_t* w)
/* bug-1767 / Req 5.2: parjob solves only real multi-task requests. A task
* count of 0 or 1 reaching it (e.g. a snapshot-restored primary with a
* defaulted maxTasks) pends with no reason record and no resource check. */
static int parjob_accepts(const xce_primary_resreq_t* primary)
{
int i;
for (i = 0; i < nclaimed; i++) {
if (claimed[i].w == w) {
return 1;
}
}
return 0;
return primary->maxTasks > 1;
}
/*
* parjob_snapshot_host_req_ - deep-copies `src` (ctx->alloc_req.host at
* the moment a bucket's host was accepted) into `dst`, using xvec_dup()
* on each of the idxs/vals xvec_t members since xvec.h has no dedicated
* xvec_idval_t-level dup helper. `dst` is safe to release later with the
* ordinary xvec_idval_fini() (xvec.h) -- it owns its own copies of the
* idxs/vals data buffers, fully independent of `src`. Returns 0 on
* success, negative errno (from the underlying xvec_dup() calls) on OOM,
* leaving `dst` zeroed (nothing partially allocated needing release).
*/
static int parjob_snapshot_host_req_(xvec_idval_t* dst, const xvec_idval_t* src)
{
xvec_t* idxs_dup = NULL;
xvec_t* vals_dup = NULL;
int rc;
memset(dst, 0, sizeof(*dst));
rc = xvec_dup(&idxs_dup, (xvec_t*)&(src->idxs));
if (rc < 0) {
FREEUP(idxs_dup);
return rc;
}
rc = xvec_dup(&vals_dup, (xvec_t*)&(src->vals));
if (rc < 0) {
xvec_fini(idxs_dup);
FREEUP(idxs_dup);
FREEUP(vals_dup);
return rc;
}
dst->count = src->count;
dst->idxs = *idxs_dup;
dst->vals = *vals_dup;
/* Only the wrapper xvec_t structs allocated by xvec_dup() are freed
* here -- their .data buffers were just copied by value into dst's
* own .idxs/.vals fields above, so dst now owns them. */
FREEUP(idxs_dup);
FREEUP(vals_dup);
return 0;
}
/*
* parjob_rollback_claims_ - requirements.md's "Rollback scope": reverses
* the provisional per-host debit sched_try_alloc_on_host_() made for
* every one of this plan attempt's already-claimed buckets (via
* resource.c's sched_ctx_uncommit_host(), task 1.4's reversal primitive),
* and releases each claim's snapshot. Scoped entirely to ctx->uncomitted
* (per-job-allocation-attempt scratch state) -- never touches the real,
* session-wide resInUse/sharedInUse books, matching sched_ctx_uncommit_
* host()'s own documented scope.
*/
static void parjob_rollback_claims_(xce_sched_ctx_t* ctx,
parjob_claim_t* claimed, int nclaimed)
{
int i;
for (i = 0; i < nclaimed; i++) {
sched_ctx_uncommit_host(ctx, claimed[i].w->idx, &(claimed[i].host_req));
xvec_idval_fini(&(claimed[i].host_req));
}
}
/*
* parjob_try_plan_ - attempts to satisfy every bucket of ONE candidate
* plan, per requirements.md's Composition Algorithm: for each bucket, a
* FRESH xce_host_iter_t instance (never one shared/reused across buckets,
* Req 3 AC 3/4, Req 4 AC 2) scans the full worker set, skipping only
* hosts already claimed by an EARLIER bucket of this SAME plan attempt
* (is_claimed(); a host merely offered-and-rejected by an earlier bucket
* for being too small remains eligible here). Buckets are tried in
* exactly plan->bucket_slots[]'s own order -- never sorted or reversed --
* which is what makes the plan iterator's descending-size convention a
* real correctness property rather than a cosmetic default (see
* requirements.md's worked example: trying the largest bucket first
* against the full host pool avoids a smaller bucket wastefully consuming
* a host only the larger bucket could have used).
*
* Before each candidate host is checked, ctx->nslots (and ctx->nhosts) is
* set to the CURRENT bucket's own slot count (Req 4 AC 6) so that
* parjob_prepare_allocation() computes that bucket's real per-host want,
* not a fixed value.
*
* On full success: returns 1, `*nclaimed_out` set, and `claimed[]`
* (caller-owned, must be sized >= plan->nbuckets) holds one entry per
* bucket in bucket order -- caller owns releasing each claimed[i].host_req
* (xvec_idval_fini()) once done with it, whether after merging into a
* committed allocation or an explicit rollback.
*
* On a bucket failure (its fresh host_iter exhausted with no viable,
* unclaimed candidate): rolls back every earlier bucket's provisional
* debit in THIS plan attempt itself (so the caller never has to), sets
* `*nclaimed_out = 0`, and returns 0 -- the caller advances to the next
* candidate plan.
*
* Returns a negative errno only on a genuine infrastructure error (e.g.
* OOM snapshotting a claim) -- not a resource-availability determination.
* Already-claimed buckets (including the one that just failed to
* snapshot) are rolled back before returning.
*/
static int parjob_try_plan_(xce_sched_core_t* core, xce_sched_ctx_t* ctx,
xce_alloc_plan_t* plan, parjob_claim_t* claimed,
int* nclaimed_out)
{
int nclaimed = 0;
int b;
plan->ctx = ctx;
for (b = 0; b < plan->nbuckets; b++) {
xce_host_iter_t host_iter_storage;
xce_host_iter_t* host_iter = &host_iter_storage;
xce_sched_worker_t* cand;
int bucket_ok = 0;
/*
* Reset ctx->alloc_req.host to empty BEFORE this bucket's fresh
* host_iter scan begins. sched_host_iter_next() (host_iter.c)
* pre-filters every candidate against WHATEVER this vector
* currently holds and never populates it itself (by design --
* see host_iter.c's own file header). Left un-reset, it would
* still carry a DIFFERENT bucket's want (or, for the very first
* bucket ever tried, the job-level ctx->worker==NULL prepare
* pass's whole-primary slots:maxTasks want, which that pass
* rebuilds per primary since bug-1883) -- harmless for seqjob
* (nslots is always 1, so every bucket's want is identical by
* construction) but actively wrong for parjob, where bucket size
* genuinely varies call to call: a stale want here can wrongly
* REJECT a host that would satisfy THIS bucket's real, different
* want, before sched_try_alloc_on_host_()'s own authoritative
* per-candidate recompute (below, with ctx->nslots freshly set)
* ever gets a chance to run on it. An empty want vector makes
* sched_check_host_satisfy()'s pre-filter a deliberate no-op
* (its own documented `wanted->count == 0` trivial-pass
* shortcut) for this bucket's scan -- safe, since host_iter.c's
* pre-filter is already documented, pre-existing, benign
* defense-in-depth layered on top of the real check, never the
* sole gate (tasks.md's implementation note (3.2)).
*/
xvec_idval_fini(&(ctx->alloc_req.host));
xvec_idval_init(&(ctx->alloc_req.host), 8);
host_iter->init = sched_host_iter_init;
host_iter->next = sched_host_iter_next;
host_iter->fini = sched_host_iter_fini;
if (host_iter->init(host_iter, plan, core->workers) == 0) {
while ((cand = host_iter->next(host_iter)) != NULL) {
if (parjob_is_claimed_(claimed, nclaimed, cand)) {
continue;
}
ctx->nslots = plan->bucket_slots[b];
ctx->nhosts = 1;
/* scheduler-sim-live-allocation spec, task 4.3:
* sched_try_alloc_on_host_() takes core directly (already
* core-typed since task 2.1) -- this is the exact call
* whose own internal plugin_prepare_allocation() fan-out
* (resource.c) must reach the SAME core->plugins registry
* every other plugin (e.g. the limit plugin) is registered
* against, so it reads the caller's honest `core`
* parameter (my_alloc()'s own plugin->core), not a
* throwaway wrapper derived from plugin->sched. */
if (sched_try_alloc_on_host_(cand, core, ctx) > 0) {
int rc = parjob_snapshot_host_req_(
&(claimed[nclaimed].host_req), &(ctx->alloc_req.host));
if (rc < 0) {
/* This host's own debit was never recorded into
* claimed[] (the snapshot that would let rollback
* find it failed), so undo it directly here with
* the still-fresh ctx->alloc_req.host, then roll
* back every earlier bucket in this plan too. */
sched_ctx_uncommit_host(ctx, cand->idx,
&(ctx->alloc_req.host));
parjob_rollback_claims_(ctx, claimed, nclaimed);
host_iter->fini(host_iter);
*nclaimed_out = 0;
return rc;
}
claimed[nclaimed].w = cand;
nclaimed++;
bucket_ok = 1;
break;
}
}
host_iter->fini(host_iter);
}
/* host_iter->init() failing is treated the same as "no
* candidates" for this bucket -- an infrastructure error, not a
* resource-availability determination (alloc_iter.h). */
if (!bucket_ok) {
parjob_rollback_claims_(ctx, claimed, nclaimed);
*nclaimed_out = 0;
return 0;
}
}
*nclaimed_out = nclaimed;
return 1;
}
static int my_alloc(xce_sched_plugin_t* plugin, xce_sched_ctx_t* ctx)
{
xce_sched_core_t* core = plugin->core;
/* Kept alongside `core` above ONLY for the sched_make_allocs() calls
* below (scheduler.c, out of this task's boundary) -- see
* seqjob_allocator.c's identical my_alloc() for the full rationale. */
xce_scheduler_t* sched = (xce_scheduler_t*)plugin->sched;
xce_primary_resreq_t* primary = ctx->primary;
xce_pendreason_t* reason = ctx->reason;
xvec_t* uncomit = NULL;
xce_alloc_plan_iter_t plan_iter_storage;
xce_alloc_plan_iter_t* plan_iter = &plan_iter_storage;
xce_alloc_plan_t* plan;
int rc;
if ((core == NULL) || (sched == NULL) || (primary == NULL)) {
return -EINVAL;
}
/*
* Req 4 AC 1's own stated precondition: my_alloc() is reached only
* for maxTasks > 1 (sched_make_primary_alloc_()'s dispatch routes
* maxTasks == 1 to seqjob instead). A maxTasks <= 1 value reaching
* this function directly is out-of-contract input, not a real
* combinatorial-allocation attempt -- discovered via
* test_snapshot_load_allocator_regression.c: a snapshot-restored
* job's independently-reconstructed primary (snapshot_load.c) can
* carry an unpopulated, defaulted maxTasks of 0 (bug-1767's broader,
* pre-existing "recurring duplication" pattern in that reconstruction
* path -- out of this spec's own scope per requirements.md's Req 7
* note). Bailing out here, before ctx->reason is ever created, keeps
* this exact edge case's observable outcome identical to the stub
* this task replaces (a bare not-satisfied signal, no reason
* fabricated for an input this allocator was never actually asked to
* solve) -- for any GENUINE maxTasks > 1 job, this check is a no-op.
*/
if (primary->maxTasks <= 1) {
return 0;
}
if (reason == NULL) {
reason = make_pendreason(0);
ctx->reason = reason;
if (reason == NULL) {
return -ENOMEM;
}
}
/*
* Req 4 AC 7: shared-resource satisfaction is computed and checked
* once per primary resreq, independent of which plan/bucket-shape is
* later tried -- identical mechanism to seqjob_allocator's own
* my_alloc(). ctx->alloc_req.shared was already populated once by
* parjob_prepare_allocation()'s ctx->worker == NULL branch, fired by
* sched_ctx_set_primary() (scheduler.c) before my_alloc() was ever
* called; this function never re-triggers that pass.
*/
rc = sched_ctx_get_uncomit_shared(&uncomit, ctx);
if (rc < 0) {
return rc;
}
/* scheduler-sim-live-allocation task 4.3: sched_check_shared_satisfy()
* takes core directly (already core-typed since task 1.3) -- read
* from the plugin's own honest `core` field, not derived from
* plugin->sched. */
/* bug-1884: the check (re)initializes unmeet itself */
xvec_idval_fini(&(ctx->unmeet));
rc = sched_check_shared_satisfy(core, uncomit,
&(ctx->alloc_req.shared), &(ctx->unmeet));
if (rc < 0) {
return rc;
}
if (rc == 0) {
int i;
for (i = 0; i < ctx->alloc_req.shared.count; i++) {
xce_resource_t* r = NULL;
rc = vget(core->resmgr, ((int*)ctx->unmeet.idxs.data)[i],
(void**)&r);
if (rc <= 0) {
continue;
}
add_job_preason(reason->reasons, PREASON_NO_RESOURCE, NULL,
r->name);
}
return 0;
}
/* add uncomitted shared */
rc = xvec_idval(uncomit, &(ctx->alloc_req.shared), XVEC_ADD_FLOAT);
if (rc < 0) {
return rc;
}
/*
* Req 2/4: combinatorial plan_iter -> (bucket -> fresh host_iter ->
* sched_try_alloc_on_host_()) per the Composition Algorithm
* (requirements.md). Replaces the unconditional `return 0` stub.
*/
plan_iter->init = parjob_plan_iter_init;
plan_iter->next = parjob_plan_iter_next;
plan_iter->fini = parjob_plan_iter_fini;
if (plan_iter->init(plan_iter, primary, core) != 0) {
/* infra error (e.g. OOM) -- treated the same as "no candidates,"
* not a resource-availability determination (alloc_iter.h). */
return 0;
}
while ((plan = plan_iter->next(plan_iter)) != NULL) {
parjob_claim_t claimed[primary->maxTasks];
int nclaimed = 0;
int plan_rc;
plan_rc = parjob_try_plan_(core, ctx, plan, claimed, &nclaimed);
if (plan_rc < 0) {
/* genuine infrastructure error -- parjob_try_plan_() already
* rolled back whatever it partially claimed before returning
* this; not a resource-availability determination. */
plan_iter->fini(plan_iter);
return plan_rc;
}
if (plan_rc > 0) {
/*
* Req 4 AC 3: every bucket satisfied -- commit as a single
* xce_job_alloc_t merging one xce_host_alloc_t per claimed
* bucket/host (nhosts == plan->nbuckets). design.md's "Final
* allocation result": hallocs is built directly with
* calloc()+a loop (runtime-sized), since make_host_allocs()'s
* variadic signature only fits a compile-time-fixed argument
* list; make_job_alloc() itself already takes a runtime nhost
* + xce_host_alloc_t** array, unchanged.
*/
xce_host_alloc_t** hallocs = NULL;
int i;
hallocs = (xce_host_alloc_t**)calloc((size_t)nclaimed,
sizeof(*hallocs));
if (hallocs == NULL) {
parjob_rollback_claims_(ctx, claimed, nclaimed);
plan_iter->fini(plan_iter);
return -ENOMEM;
}
/* bug-1882: any constructor failing (OOM) is -ENOMEM with
* ctx->alloc left NULL (the caller moves ctx->alloc onto the
* job whatever the rc): free the host records built so far,
* roll back the claims not yet turned into one. */
for (i = 0; i < nclaimed; i++) {
int nres = claimed[i].host_req.count;
xce_alloc_t** allocs
= sched_make_allocs(sched, "", nres,
(int*)claimed[i].host_req.idxs.data,
(float*)claimed[i].host_req.vals.data,
/* FIXME: no duration now */ NULL);
if ((allocs != NULL) || (nres <= 0)) {
hallocs[i]
= make_host_alloc(claimed[i].w->name, nres, allocs);
if (hallocs[i] == NULL) {
free_ResAlloc_list(nres, allocs);
}
}
if (hallocs[i] == NULL) {
parjob_rollback_claims_(ctx, claimed + i, nclaimed - i);
free_HostAlloc_list(nclaimed, hallocs);
plan_iter->fini(plan_iter);
return -ENOMEM;
}
xvec_idval_fini(&(claimed[i].host_req));
}
ctx->alloc = make_job_alloc(0, nclaimed, hallocs);
if (ctx->alloc == NULL) {
free_HostAlloc_list(nclaimed, hallocs);
plan_iter->fini(plan_iter);
return -ENOMEM;
}
if (ctx->alloc_req.shared.count > 0) {
ctx->alloc->allocs_shared
= sched_make_allocs(sched, "", ctx->alloc_req.shared.count,
ctx->alloc_req.shared.idxs.data,
ctx->alloc_req.shared.vals.data,
/* FIXME: no duration now */ NULL);
if (ctx->alloc->allocs_shared == NULL) {
free_JobAlloc_ptr(&(ctx->alloc));
plan_iter->fini(plan_iter);
return -ENOMEM;
}
ctx->alloc->nres_shared = ctx->alloc_req.shared.count;
} else {
ctx->alloc->allocs_shared = NULL;
ctx->alloc->nres_shared = 0;
}
plan_iter->fini(plan_iter);
return 1;
}
/* plan_rc == 0: this plan's buckets could not all be satisfied --
* parjob_try_plan_() already rolled back this plan attempt's
* provisional debits (requirements.md's "Rollback scope") before
* returning; advance to the next candidate plan. */
}
plan_iter->fini(plan_iter);
/* Req 4 AC 4: same not-satisfied signal (0) as today's stub, once
* every candidate plan is exhausted -- pend-reason handling in
* sched_make_primary_alloc_() is unaffected. */
return 0;
}
/* parjob's policy for the framework routine sched_allocator_run()
* (alloc_runner.c): the acceptance guard and the combinatorial plans. */
static const xce_sched_allocator_iface_t parjob_alloc_iface = {
.accepts = parjob_accepts,
.plan_init = parjob_plan_iter_init,
.plan_next = parjob_plan_iter_next,
.plan_fini = parjob_plan_iter_fini
};
/*
* parjob_allocator_new_plugin - scheduler-sim-live-allocation spec, task
@ -786,7 +341,8 @@ xce_sched_plugin_t* parjob_allocator_new_plugin(void)
.begin_session_fn = NULL,
.end_session_fn = NULL,
.job_elect_fn = NULL,
.job_alloc_fn = (plugin_job_allocate_fn_t)my_alloc,
.job_alloc_fn = (plugin_job_allocate_fn_t)sched_allocator_run,
.allocator_iface = &parjob_alloc_iface,
.data = NULL
};

151
src/scheduler/testsuite/test_alloc_equivalence_parjob.c
View file

@ -1,21 +1,8 @@
/*
* test_alloc_equivalence_parjob.c - scheduler-allocator-runner task 2.2:
* characterization of the parallel-job allocator (Req 4.5, 5.1, 5.2, 5.4,
* 5.8, 7.2) through the registered plugin->job_alloc_fn, never the static
* routine by name; must still hold once task 4.1 switches parjob to the
* framework runner. Stubs and OOM injection are those of
* test_alloc_equivalence_seqjob.c: sched_make_allocs() failing on call N
* (claimed hosts' records in claim order, then shared records) and
* --wrap'ed make_host_alloc/make_job_alloc failing on call N; the
* host-array allocation is not injected (task 3.3).
*
* Values the switch changes on purpose sit in `_today` tests: bug-1893
* (Req 2.2) and host reasons (Req 3.6/3.7). Today the host request is
* reset once per bucket, so later short candidates are pre-filtered
* (host_iter.c) against the request the first one left and get no reason;
* the runner clears it before every candidate. Stable tests assert only
* reason entries both rules give. bug-1892 layouts are kept out.
*/
/* test_alloc_equivalence_parjob.c - scheduler-allocator-runner 2.2/4.1:
* parjob through plugin->job_alloc_fn only (Req 2.2, 3.6, 3.7, 4.5, 5.1,
* 5.2, 5.4, 5.8, 7.2). OOM injection as test_alloc_equivalence_seqjob.c:
* sched_make_allocs() (hosts in claim order, then shared) and --wrap'ed
* make_host_alloc/make_job_alloc fail on call N. No bug-1892 layouts. */
#include <errno.h>
#include <stdlib.h>
#include <string.h>
@ -30,21 +17,19 @@
#include "xcealloc.h"
#include "xstring.h"
/* White-box sources under test. Task 4.1 adds the framework runner
* ("../src/alloc_runner.c") as one more line of this list. */
/* White-box sources under test. */
#include "../src/core.c"
#include "../src/resource.c"
#include "../src/schedobj.c"
#include "../src/host_iter.c"
#include "../src/alloc_runner.c"
#include "../src/plugins/parjob_allocator.c"
/* Link-satisfying stubs (same convention as test_seqjob_allocator.c). */
int plugin_register(void* s, void* p) { return (void)s, (void)p, 0; }
void plugin_unregister(void* arg) { (void)arg; }
/* plugin.c's fan-out with parjob's hook only: reset the request vectors as
* plugin.c does (job-level pass: .shared and .host, bug-1883; per-host
* pass: .host), then call the hook. */
/* plugin.c's fan-out, parjob's hook only; resets as plugin.c (bug-1883). */
int plugin_prepare_allocation(xce_sched_core_t* sched, xce_sched_ctx_t* ctx)
{
xce_sched_plugin_t fake_plugin;
@ -87,9 +72,8 @@ static struct {
static void oom_injection_reset(void) { memset(&g, 0, sizeof(g)); }
/* Stub for scheduler.c's sched_make_allocs(): real make_alloc() records
* (resmgr name, requested value, empty reservation, duration -1 for a NULL
* duration array, instid 0); NULL for nres <= 0 as the real one. */
/* scheduler.c's sched_make_allocs(): real make_alloc() records (duration -1
* for a NULL duration array, instid 0); NULL for nres <= 0 as the real one. */
xce_alloc_t** sched_make_allocs(xce_scheduler_t* sched, char* reserve,
int nres, int* idxs, float* vals,
int* duration)
@ -173,6 +157,7 @@ typedef struct {
xce_resource_t* mem;
xce_resource_t* rack;
xce_resource_t* lic; /* shared (XCE_RESF_GLOBAL) */
xce_resource_t* tok; /* shared, pool 10, not requested by eq_run */
xce_sched_plugin_t* plugin;
xce_sched_obj_t obj;
xce_primary_schedobj_t primary_obj;
@ -190,11 +175,11 @@ typedef struct {
xce_select_node_t sel_res, sel_num, sel_eq;
} eq_primary_t;
/* Shared pool "licenses" has `lic_total` capacity, none in use. */
/* Shared pools "licenses" (`lic_total`) and "tokens" (10), none in use. */
static void eq_fixture_init(eq_fixture_t* f, float lic_total)
{
int idx[1];
float val[1];
int idx[2];
float val[2];
memset(f, 0, sizeof(*f));
TEST_ASSERT_EQUAL_INT(0, xce_sched_core_init(&(f->sched.core)));
@ -213,15 +198,19 @@ static void eq_fixture_init(eq_fixture_t* f, float lic_total)
f->mem = EQ_RES("mem", 0);
f->rack = EQ_RES("rack", 0);
f->lic = EQ_RES("licenses", XCE_RESF_GLOBAL);
f->tok = EQ_RES("tokens", XCE_RESF_GLOBAL);
#undef EQ_RES
TEST_ASSERT_NOT_NULL(f->slots);
TEST_ASSERT_NOT_NULL(f->mem);
TEST_ASSERT_NOT_NULL(f->rack);
TEST_ASSERT_NOT_NULL(f->lic);
TEST_ASSERT_NOT_NULL(f->tok);
idx[0] = f->lic->idx;
val[0] = lic_total;
put_resvals(&(f->sched.core.sharedInst), 1, idx, val);
idx[1] = f->tok->idx;
val[1] = 10.0f;
put_resvals(&(f->sched.core.sharedInst), 2, idx, val);
f->plugin = parjob_allocator_new_plugin();
TEST_ASSERT_NOT_NULL(f->plugin);
@ -286,9 +275,8 @@ static void eq_primary_init(eq_primary_t* p, int max_tasks, float mem,
}
}
/* ctx over a PRIMARY obj for `p`, job-level prepare pass included
* (sched_ctx_set_primary() -> plugin_prepare_allocation(), ctx->worker
* NULL), exactly as sched_make_primary_allocation() drives it. */
/* ctx over a PRIMARY obj for `p`, job-level prepare pass included, exactly
* as sched_make_primary_allocation() drives it. */
static void eq_ctx_begin_primary(eq_fixture_t* f, eq_primary_t* p)
{
f->primary_obj.resreq = &(p->req);
@ -358,8 +346,7 @@ static void eq_assert_record(xce_alloc_t* a, const char* resname, float value)
TEST_ASSERT_EQUAL_INT(-1, a->duration);
}
/* One host allocation carrying mem then slots (parjob's per-host prepare
* computes the rusage wants first, then stores the bucket's slot count). */
/* One host allocation carrying mem then the bucket's slot count. */
static void eq_assert_host(xce_host_alloc_t* ha, const char* host, float mem,
float slots)
{
@ -399,9 +386,8 @@ static void eq_assert_calls(int make_allocs, int host_alloc, int job_alloc)
TEST_ASSERT_EQUAL_INT(job_alloc, g.job_alloc_calls);
}
/* A fixture: `pool` shared licenses; hosts A, B, C (rack 1) with the given
* slots and mem; a job of `tasks` tasks, `task_mem` mem per task and `lic`
* licenses per job. */
/* `pool` licenses; hosts A, B, C (rack 1) with the given slots and mem; a
* job of `tasks` tasks, `task_mem` mem per task, `lic` licenses per job. */
typedef struct {
float pool;
int nhosts;
@ -441,8 +427,7 @@ static int eq_run(eq_fixture_t* f, eq_primary_t* p, const eq_spec_t* s,
return eq_alloc(f);
}
/* Main and host reasons none; at least `n` resource reasons, the first `n`
* of which both rules give (count asserted only in `_today` tests). */
/* Main and host reasons none; at least `n` resource reasons. */
static void eq_assert_reason_prefix(eq_fixture_t* f, int n)
{
TEST_ASSERT_NOT_NULL(f->ctx.reason);
@ -451,7 +436,7 @@ static void eq_assert_reason_prefix(eq_fixture_t* f, int n)
TEST_ASSERT_TRUE(f->ctx.reason->reasons->nres_reasons >= n);
}
/* EQ_TWO_BUCKETS: plan [4] checks A first (slots short) under both rules. */
/* EQ_TWO_BUCKETS: plan [4] checks A first (slots short). */
static void eq_assert_plan1_reasons(eq_fixture_t* f)
{
eq_assert_reason_prefix(f, 1);
@ -508,18 +493,24 @@ static void test_parjob_pending_on_short_shared_resource(void)
eq_finish(&f, &p);
}
/* 2, resource reasons, TODAY's behaviour: none. The reason loop skips
* every unmet entry because vget() returns 0 on success and the loop
* tests `rc <= 0` (bug-1893). Requirement 2.2 changes this in task 4.1
* (one host-less NO_RESOURCE reason naming "licenses"). */
static void test_parjob_pending_on_short_shared_no_reason_today(void)
/* 2, resource reasons (Req 2.2, bug-1893): two shared resources requested,
* only "licenses" short: exactly one host-less reason, naming it. */
static void test_parjob_pending_on_short_shared_names_short_one(void)
{
eq_fixture_t f;
eq_primary_t p;
xce_sched_worker_t* w[3];
rusage_value_t tok = { .res = (char*)"tokens", .value = 1.0f };
TEST_ASSERT_EQUAL_INT(0, eq_run(&f, &p, &EQ_SHORT_SHARED, w));
eq_assert_reason_counts(&f, 0, 0, 0);
eq_fixture_init(&f, EQ_SHORT_SHARED.pool);
eq_add_host(&f, HOST_A, 4.0f, 8.0f, 1.0f);
eq_primary_init(&p, 2, 1.0f, 3.0f, &f, 0);
tok.per_flag = PER_JOB;
xlist_store(&(p.req.rusages), &tok);
eq_ctx_begin_primary(&f, &p);
TEST_ASSERT_EQUAL_INT(0, eq_alloc(&f));
eq_assert_reason_counts(&f, 0, 0, 1);
eq_assert_res_reason(&f, 0, "", "licenses"); /* host-less: "" */
eq_finish(&f, &p);
}
@ -535,7 +526,7 @@ static void test_parjob_pending_on_hosts(void)
TEST_ASSERT_EQUAL_INT(0, eq_run(&f, &p, &EQ_SHORT_HOSTS, w));
TEST_ASSERT_NULL(f.ctx.alloc);
/* plan [2]'s first candidate A, under both rules */
/* plan [2]'s first candidate A */
eq_assert_reason_prefix(&f, 2);
eq_assert_res_reason(&f, 0, HOST_A, "mem");
eq_assert_res_reason(&f, 1, HOST_A, "slots");
@ -571,8 +562,7 @@ static void test_parjob_mid_plan_rollback_then_later_plan_wins(void)
eq_assert_debit(&f, w[2], 1.0f, 1.0f);
TEST_ASSERT_EQUAL_FLOAT(3.0f, eq_shared_uncomit(&f));
/* A then B, slots: plan [4]'s first two checks (new rule), or plan
* [4]'s A and plan [2,2] bucket 2's B (today) */
/* A then B, slots: plan [4]'s first two checks */
eq_assert_reason_prefix(&f, 2);
eq_assert_res_reason(&f, 0, HOST_A, "slots");
eq_assert_res_reason(&f, 1, HOST_B, "slots");
@ -581,28 +571,40 @@ static void test_parjob_mid_plan_rollback_then_later_plan_wins(void)
eq_finish(&f, &p);
}
/* 1, 3, 4: host reasons, TODAY's behaviour (stale pre-filter): only each
* bucket search's first short candidate gets reasons. Req 3.6/3.7 change
* this in task 4.1 to every selection-matching unclaimed short host:
* 1 -> (A,slots) (B,slots); 3 -> A mem, A slots, B mem, B slots, A mem,
* B mem; 4 -> A, B, C (plan [4]) then B, C (plan [2,2] bucket 2), slots. */
static void test_parjob_host_reasons_first_candidate_only_today(void)
static void eq_assert_reasons(eq_fixture_t* f, int n, const char** pairs)
{
int i;
eq_assert_reason_counts(f, 0, 0, n);
for (i = 0; i < n; i++) {
eq_assert_res_reason(f, i, pairs[2 * i], pairs[(2 * i) + 1]);
}
}
/* 1, 3, 4: host reasons (Req 3.6/3.7): exactly one per short resource of
* every selection-matching unclaimed short host of each bucket search. */
static void test_parjob_host_reasons_every_short_host(void)
{
static const char* two[] = { HOST_A, "slots", HOST_B, "slots" };
static const char* hosts[]
= { HOST_A, "mem", HOST_A, "slots", HOST_B, "mem",
HOST_B, "slots", HOST_A, "mem", HOST_B, "mem" };
static const char* back[] = { HOST_A, "slots", HOST_B, "slots", HOST_C,
"slots", HOST_B, "slots", HOST_C, "slots" };
eq_fixture_t f;
eq_primary_t p;
xce_sched_worker_t* w[3];
TEST_ASSERT_EQUAL_INT(1, eq_run(&f, &p, &EQ_TWO_BUCKETS, w));
eq_assert_reason_counts(&f, 0, 0, 1);
eq_assert_reasons(&f, 2, two);
eq_finish(&f, &p);
TEST_ASSERT_EQUAL_INT(0, eq_run(&f, &p, &EQ_SHORT_HOSTS, w));
eq_assert_reason_counts(&f, 0, 0, 3);
eq_assert_res_reason(&f, 2, HOST_A, "mem");
eq_assert_reasons(&f, 6, hosts);
eq_finish(&f, &p);
TEST_ASSERT_EQUAL_INT(1, eq_run(&f, &p, &EQ_ROLLBACK, w));
eq_assert_reason_counts(&f, 0, 0, 2);
eq_assert_reasons(&f, 5, back);
eq_finish(&f, &p);
}
@ -664,14 +666,12 @@ static void test_parjob_missing_input_einval_no_reason(void)
}
}
/* 7. Out of memory while building the result of EQ_TWO_BUCKETS (claims A
* then B). Host stage: the claims not yet turned into a host allocation
* are released, converted ones keep their debit; job allocation and shared
* records: every debit stays (design Error table; today's parjob too).
* `a_kept` / `b_kept`: whether A's / B's debit (2 mem, 2 slots) remains.
* Cases: records of host 0 / host 1 (sched_make_allocs() call 1 / 2),
* make_host_alloc() for host 0 / host 1, make_job_alloc(), shared records
* (sched_make_allocs() call 3, after both hosts). */
/* 7. OOM building EQ_TWO_BUCKETS' result (claims A then B). Host stage:
* unconverted claims released, converted ones keep their debit; job alloc
* and shared records: every debit stays (design Error table). `a_kept` /
* `b_kept`: whether A's / B's debit (2 mem, 2 slots) remains. Steps: host
* 0/1 records (sched_make_allocs() 1/2), make_host_alloc() 1/2,
* make_job_alloc(), shared records (sched_make_allocs() 3). */
static void eq_check_oom(int make_allocs_fail_on, int host_alloc_fail_on,
int job_alloc_fail_on, int a_kept, int b_kept,
int make_allocs, int host_alloc, int job_alloc)
@ -712,10 +712,9 @@ EQ_OOM_TEST(test_parjob_oom_job_alloc_debits_stay,
EQ_OOM_TEST(test_parjob_oom_shared_records_debits_stay,
3, 0, 0, 1, 1, 3, 2, 1)
/* 8. COMPOUND: two primaries allocated in turn on one ctx, as
* sched_make_compound_alloc_() does; rack 1 / rack 2 selects keep them on
* distinct hosts (bug-1892 not exercised). Each is sized from its own
* request (Req 5.8): 2 x 1 mem + 3 licenses on A, 3 x 2 mem on B. */
/* 8. COMPOUND: two primaries in turn on one ctx (sched_make_compound_alloc_);
* rack 1 / rack 2 selects keep them apart. Each sized from its own request
* (Req 5.8): 2 x 1 mem + 3 licenses on A, 3 x 2 mem on B. */
static void test_parjob_compound_second_primary_sized_from_own_request(void)
{
eq_fixture_t f;
@ -782,10 +781,10 @@ int main(void)
UNITY_BEGIN();
RUN_TEST(test_parjob_success_two_buckets_with_shared);
RUN_TEST(test_parjob_pending_on_short_shared_resource);
RUN_TEST(test_parjob_pending_on_short_shared_no_reason_today);
RUN_TEST(test_parjob_pending_on_short_shared_names_short_one);
RUN_TEST(test_parjob_pending_on_hosts);
RUN_TEST(test_parjob_mid_plan_rollback_then_later_plan_wins);
RUN_TEST(test_parjob_host_reasons_first_candidate_only_today);
RUN_TEST(test_parjob_host_reasons_every_short_host);
RUN_TEST(test_parjob_task_count_zero_and_one_pend_without_reason);
RUN_TEST(test_parjob_missing_input_einval_no_reason);
RUN_TEST(test_parjob_oom_records_host0_releases_all);

41
src/scheduler/testsuite/test_parjob_combinatorial.c
View file

@ -90,6 +90,7 @@
* to reach them directly.
*/
#include "../src/core.c"
#include "../src/alloc_runner.c"
#include "../src/host_iter.c"
#include "../src/plugins/parjob_allocator.c"
#include "../src/resource.c"
@ -313,7 +314,8 @@ static void init_primary_obj(xce_sched_obj_t* obj,
* attempt runs: sched_ctx_init() -> sched_ctx_set_primary() (the real
* call that stages ctx->alloc_req.shared via plugin_prepare_allocation()'s
* ctx->worker==NULL branch, exactly as sched_make_primary_allocation()
* does today) -> my_alloc(). Returns my_alloc()'s return code; ctx is left
* does today) -> parjob's registered entry point, the framework routine
* sched_allocator_run(). Returns its return code; ctx is left
* initialized (caller must sched_ctx_fini()). Identical to
* test_seqjob_rewrap_equivalence.c's own helper. */
static int drive_my_alloc(xce_scheduler_t* sched, xce_sched_obj_t* obj,
@ -328,13 +330,12 @@ static int drive_my_alloc(xce_scheduler_t* sched, xce_sched_obj_t* obj,
memset(&plugin, 0, sizeof(plugin));
plugin.sched = sched;
/* scheduler-sim-live-allocation spec, task 4.3: my_alloc() now reads
* plugin->core (honestly typed, task 1.2) for every core-safe call,
* alongside the still-needed plugin->sched (sched_make_allocs(),
* scheduler.c, unchanged). */
plugin.core = &(sched->core);
/* the runner reads plugin->core, plugin->sched (sched_make_allocs())
* and parjob's allocator contract */
plugin.core = &(sched->core);
plugin.allocator_iface = &parjob_alloc_iface;
return my_alloc(&plugin, ctx);
return sched_allocator_run(&plugin, ctx);
}
/* assert_uncommitted_host_debit - reads ctx->uncomitted.hosts[w->idx] for
@ -824,7 +825,7 @@ static void test_descending_bucket_order_succeeds_where_ascending_fails(void)
xce_alloc_plan_t plan;
int bucket_slots_desc[2];
int bucket_slots_asc[2];
parjob_claim_t claimed[2];
sched_alloc_claim_t claimed[2];
int nclaimed;
int rc;
int idx[1];
@ -866,9 +867,9 @@ static void test_descending_bucket_order_succeeds_where_ascending_fails(void)
TEST_ASSERT_EQUAL_INT(0, sched_ctx_init(&ctx, &(sched.core), &obj));
sched_ctx_set_primary(&ctx, &primary);
/* my_alloc() always lazily creates ctx->reason before ever reaching
* parjob_try_plan_() (see my_alloc()'s own top-of-function guard);
* this test calls parjob_try_plan_() directly, bypassing that step,
/* sched_allocator_run() always creates ctx->reason before ever
* reaching sched_try_plan_() (alloc_runner.c); this test calls
* sched_try_plan_() directly, bypassing that step,
* so it must set one up itself -- sched_try_alloc_on_host_()
* (resource.c) unconditionally records a pend reason into
* ctx->reason->reasons on a resource-unmet candidate, which the
@ -887,11 +888,9 @@ static void test_descending_bucket_order_succeeds_where_ascending_fails(void)
plan.nbuckets = 2;
plan.bucket_slots = bucket_slots_desc;
/* scheduler-sim-live-allocation spec, task 4.3: parjob_try_plan_() is
* now honestly typed xce_sched_core_t* (mirrors task 2.1's identical
* ctx->sched ripple immediately above) -- pass &(sched.core), not
* &sched. */
rc = parjob_try_plan_(&(sched.core), &ctx, &plan, claimed, &nclaimed);
/* sched_try_plan_() (alloc_runner.c) takes the core: pass
* &(sched.core), not &sched. */
rc = sched_try_plan_(&(sched.core), &ctx, &plan, claimed, &nclaimed);
TEST_ASSERT_EQUAL_INT(1, rc);
TEST_ASSERT_EQUAL_INT(2, nclaimed);
TEST_ASSERT_TRUE(claimed[0].w == host_first);
@ -899,7 +898,7 @@ static void test_descending_bucket_order_succeeds_where_ascending_fails(void)
/* Roll back before trying the ascending variant against the exact
* same, otherwise-untouched host pool. */
parjob_rollback_claims_(&ctx, claimed, nclaimed);
sched_rollback_claims_(&ctx, claimed, nclaimed);
assert_uncommitted_host_debit(&ctx, host_first, slots_res->idx, 0.0f);
assert_uncommitted_host_debit(&ctx, host_second, slots_res->idx, 0.0f);
@ -914,11 +913,9 @@ static void test_descending_bucket_order_succeeds_where_ascending_fails(void)
plan.nbuckets = 2;
plan.bucket_slots = bucket_slots_asc;
/* scheduler-sim-live-allocation spec, task 4.3: parjob_try_plan_() is
* now honestly typed xce_sched_core_t* (mirrors task 2.1's identical
* ctx->sched ripple immediately above) -- pass &(sched.core), not
* &sched. */
rc = parjob_try_plan_(&(sched.core), &ctx, &plan, claimed, &nclaimed);
/* sched_try_plan_() (alloc_runner.c) takes the core: pass
* &(sched.core), not &sched. */
rc = sched_try_plan_(&(sched.core), &ctx, &plan, claimed, &nclaimed);
TEST_ASSERT_EQUAL_INT(0, rc);
TEST_ASSERT_EQUAL_INT(0, nclaimed);