feat(scheduler-allocation): implement parjob's per-bucket allocation loop
Replace parjob's return-0 stub with a real multi-host allocator: for each candidate plan, walk buckets in the given (descending) order, claiming a distinct host per bucket via a fresh host_iter excluding hosts already claimed elsewhere in the same plan attempt. On a bucket failure, reverse every earlier successful claim's provisional debit via sched_ctx_uncommit_host() before advancing to the next plan (Req 4.2, 4.4, 4.5, 4.6, 4.7). Fix parjob_prepare_allocation() to compute each bucket's want from that bucket's own slot count (ctx->nslots) instead of a fixed value, and reset ctx->alloc_req.host at the top of each bucket's scan so host_iter's pre-filter can't inherit a stale want from a differently-sized prior bucket (safe: sched_try_alloc_on_host_() always repopulates it from the real per-bucket want before its own authoritative check runs). Add a maxTasks<=1 guard matching Req 4 AC1's precondition, needed because a snapshot-restored job with unpopulated maxTasks==0 (pre-existing, tracked separately as bug-1767) would otherwise reach parjob's now-real allocator instead of being silently ignored as before. On full success, merges every claimed bucket into one xce_job_alloc_t via make_job_alloc()'s existing runtime-sized host array, satisfying task 4.3's multi-host commit scope as well. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01XYdtrEWWZfKRRGED7raEYg
This commit is contained in:
parent
6ca717fb86
commit
3d5e1c32c0
3 changed files with 1109 additions and 46 deletions
|
|
@ -57,7 +57,7 @@
|
|||
- Write a unit test proving one-plan-per-call generation, most-preferred-first ordering (including an uneven split, e.g. splitting 7 tasks across 3 hosts), and a stable exhaustion signal
|
||||
- Observable: a passing test suite covers plan ordering, uneven-split partitioning, and exhaustion signaling
|
||||
- _Requirements: 2.1, 2.2, 2.3, 2.4, 2.5, 2.6, 4.1_
|
||||
- [ ] 4.2 Implement parjob's per-bucket allocation loop
|
||||
- [x] 4.2 Implement parjob's per-bucket allocation loop
|
||||
- Replace the parallel allocator's stub with a real implementation that, for each candidate plan, claims a distinct host for every bucket using a freshly-initialized host search per bucket, excluding only hosts already claimed elsewhere in the same plan attempt
|
||||
- Fix the parallel allocator's resource-request preparation to compute each bucket's want from that bucket's own slot count rather than a fixed value
|
||||
- On a bucket failure, reverse the provisional debits already made for that plan's earlier successful buckets before advancing to the next candidate plan
|
||||
|
|
@ -115,3 +115,6 @@
|
|||
- (3.1) `src/scheduler/meson.build`'s production `scheduler_srcs` list was missing `host_iter.c` since task 2.1 only wired that file into `testsuite/meson.build` — invisible until task 3.1 became the first *production* caller of `host_iter.c`'s functions (a shared .c file with zero production callers builds/links fine without being in `scheduler_srcs` at all, so task 2.1's own green build/test run couldn't have caught this). Any future task adding a new shared scheduler `.c` file should double check it lands in `scheduler_srcs`, not only `testsuite/meson.build`, even if nothing calls it yet.
|
||||
- (3.2) `host_iter.c`'s pre-filter select check and `sched_try_alloc_on_host_()`'s (`resource.c`) own independent `sched_meet_host_selection` re-check are pre-existing, intentional defense-in-depth: both run the identical check against the same live state, so a regression in either one alone is invisible at the observable-pick level — only a regression in *both simultaneously* changes behavior. This is benign and predates this feature; task 4.2 (parjob) will inherit the same double-check when it reuses `host_iter.c`/`sched_try_alloc_on_host_()`, and should not "fix" the redundancy as part of that task.
|
||||
- (3.2) `sched_collect_workers()`'s deleted (task 3.1) iteration and `host_iter.c`'s `iter_make()/iter_next()` iteration are structurally identical, not coincidentally similar: both called the exact same third-party `iter_make()/iter_next()` (RTTI lib) over the literal same `sched->core.workers` var, and the old `xlist_find()` walk was forward/insertion-order — confirmed via `git show b18c5f78^:src/scheduler/src/scheduler.c` and `xlist.c` (unmodified across this feature). Useful precedent if task 4.1/4.2's parjob equivalence claims ever need the same grounding.
|
||||
- (4.2) `sched_host_iter_next()`'s pre-filter reads `ctx->alloc_req.host` (which `sched_check_host_satisfy()` trivially passes when `count==0`), but nothing primes it per-bucket — a fresh per-bucket host_iter scan must reset `ctx->alloc_req.host` to empty at the top of each bucket's scan (`parjob_allocator.c`'s `parjob_try_plan_()`) or it can inherit a stale want from a differently-sized prior bucket. This is SAFE, not an oversell risk: `sched_try_alloc_on_host_()` always calls `plugin_prepare_allocation()` (which repopulates `.host` from the bucket's real `ctx->nslots`) before its own authoritative `sched_check_host_satisfy()` — the reset only makes the pre-filter a one-time-per-bucket no-op, never wrong. Task 5.x (cache/dispatch wiring) doesn't touch this path, but any future allocator built on `host_iter.c` with a multi-bucket loop should reset `ctx->alloc_req.host` the same way.
|
||||
- (4.2) `parjob_allocator.c`'s `my_alloc()` guards `primary->maxTasks <= 1 → return 0` (matching Req 4 AC1's literal `maxTasks > 1` precondition) — required because a snapshot-restored job with an unpopulated `maxTasks==0` (bug-1767's pre-existing, out-of-scope gap in `snapshot_load.c`) fails seqjob's `maxTasks != 1` guard (task 3.1) and falls through to parjob; without this guard parjob's now-real `my_alloc()` creates a `ctx->reason` where the old inert stub never did, corrupting unrelated snapshot-regression assertions. Causally verified (temporarily disabled, confirmed the exact predicted failure, restored).
|
||||
- (4.2) The multi-host allocation merge (task 4.3's stated scope: "merge every claimed bucket's host allocation into a single job allocation record") is already implemented and verified as part of 4.2 — `make_job_alloc()` takes a plain runtime-sized `xce_host_alloc_t**` array, and `my_alloc()` already builds/passes it with `nhosts == nclaimed == plan->nbuckets` on every full-plan success. Task 4.3 may only need targeted verification of the job-level shared-resource-computed-once property, not new implementation — check before dispatching a full implementer round.
|
||||
|
|
|
|||
|
|
@ -42,13 +42,49 @@ static int skip_bad_worker_(xce_sched_worker_t* w)
|
|||
return 0;
|
||||
}
|
||||
|
||||
/*
|
||||
* parjob_prepare_allocation - task 4.2 (Req 4 AC 5/AC 6): populates
|
||||
* ctx->alloc_req.host/ctx->alloc_req.shared for whichever bucket/host is
|
||||
* currently being tried, replacing the unconditional `return 0` stub.
|
||||
* Structured exactly like seqjob_prepare_allocation() (obj->tag guard,
|
||||
* then the ctx->worker == NULL / != NULL branch), since obj->u.primary is
|
||||
* only a valid read when obj->tag == XCE_RESREQ_PRIMARY (a union of
|
||||
* pointer-typed members reinterprets the same storage for COMPOUND/ALTER
|
||||
* tags) -- the original stub's `if (obj->tag == XCE_RESREQ_PRIMARY) {...}
|
||||
* return 0;` shape never actually enforced this ordering.
|
||||
*
|
||||
* Two passes, exactly as seqjob's does:
|
||||
*
|
||||
* 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/
|
||||
* 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
|
||||
* self-contained "1 task total" literal to "maxTasks tasks total." This
|
||||
* keeps the shared check independent of whichever plan/bucket-shape is
|
||||
* later tried, per Req 4 AC 7 ("computed once per primary resreq...
|
||||
* Plan/bucket iteration... does not change when or how often the
|
||||
* 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
|
||||
* 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
|
||||
* own slot count) instead of the hardcoded `1, 1` seqjob's version
|
||||
* uses, which ignored bucket shape entirely.
|
||||
*/
|
||||
static int parjob_prepare_allocation(void* arg, xce_sched_ctx_t* ctx)
|
||||
{
|
||||
xce_sched_plugin_t* plugin = (xce_sched_plugin_t*)arg;
|
||||
xce_sched_worker_t* worker = NULL;
|
||||
xce_sched_obj_t* obj = NULL;
|
||||
xce_sched_plugin_t* plugin = (xce_sched_plugin_t*)arg;
|
||||
xce_scheduler_t* sched = (xce_scheduler_t*)plugin->sched;
|
||||
xce_sched_obj_t* obj = NULL;
|
||||
xce_resource_t* r = NULL;
|
||||
xce_primary_resreq_t* primary = NULL;
|
||||
int rc = 0;
|
||||
|
||||
if ((plugin == NULL) || (ctx == NULL)) {
|
||||
if ((plugin == NULL) || (ctx == NULL) || (sched == NULL)) {
|
||||
return -EINVAL;
|
||||
}
|
||||
|
||||
|
|
@ -57,21 +93,55 @@ static int parjob_prepare_allocation(void* arg, xce_sched_ctx_t* ctx)
|
|||
return -EINVAL;
|
||||
}
|
||||
|
||||
if (obj->tag == XCE_RESREQ_PRIMARY) {
|
||||
xce_primary_resreq_t* primary = obj->u.primary->resreq;
|
||||
|
||||
if (primary == NULL) {
|
||||
return 0;
|
||||
}
|
||||
|
||||
if (primary->maxTasks == 1) {
|
||||
/* do not care sequence job */
|
||||
|
||||
return 0;
|
||||
}
|
||||
if (obj->tag != XCE_RESREQ_PRIMARY) {
|
||||
/* we just take care of parallel job */
|
||||
return 0;
|
||||
}
|
||||
|
||||
/* we care parallel jobs */
|
||||
primary = obj->u.primary->resreq;
|
||||
if (primary == NULL) {
|
||||
return -EINVAL;
|
||||
}
|
||||
|
||||
if (primary->maxTasks == 1) {
|
||||
/* do not care sequence job -- seqjob_allocator handles this
|
||||
* (Req 4 AC 6's own guard on the seqjob side). */
|
||||
return 0;
|
||||
}
|
||||
|
||||
if (ctx->worker == NULL) {
|
||||
ctx->nhosts = 1;
|
||||
ctx->nslots = primary->maxTasks;
|
||||
|
||||
calc_wanted_global_resvals(&(ctx->alloc_req.shared), sched->core.resmgr,
|
||||
&(primary->rusages), PER_TASK, ctx->nhosts,
|
||||
ctx->nslots);
|
||||
} else {
|
||||
calc_wanted_host_resvals(&(ctx->alloc_req.host), sched->core.resmgr,
|
||||
&(primary->rusages), PER_TASK, ctx->nhosts,
|
||||
ctx->nslots);
|
||||
}
|
||||
|
||||
/* add slots into requested resources */
|
||||
lock(sched->core.resmgr, RTTI_LOCK_READ);
|
||||
rc = key_find(sched->core.resmgr, (void*)"slots", (void**)&r);
|
||||
unlock(sched->core.resmgr);
|
||||
|
||||
if (rc < 0) {
|
||||
return rc;
|
||||
}
|
||||
|
||||
if (rc == 0) {
|
||||
return -ENOENT;
|
||||
}
|
||||
|
||||
if (r == NULL) {
|
||||
return -EINVAL;
|
||||
}
|
||||
|
||||
xvec_idval_store(&(ctx->alloc_req.host), r->idx, (float)ctx->nslots);
|
||||
|
||||
/* as r has been locked with OBJREF_NOP, so no need to unlock it */
|
||||
return 0;
|
||||
}
|
||||
|
||||
|
|
@ -229,11 +299,419 @@ 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)
|
||||
{
|
||||
int i;
|
||||
|
||||
for (i = 0; i < nclaimed; i++) {
|
||||
if (claimed[i].w == w) {
|
||||
return 1;
|
||||
}
|
||||
}
|
||||
|
||||
return 0;
|
||||
}
|
||||
|
||||
/*
|
||||
* 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_scheduler_t* sched, 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 own dead-write into .host) -- 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, sched->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;
|
||||
|
||||
if (sched_try_alloc_on_host_(cand, sched, 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_scheduler_t* sched = (xce_scheduler_t*)plugin->sched;
|
||||
xce_sched_obj_t* schedobj = ctx->schedobj;
|
||||
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 ((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;
|
||||
}
|
||||
|
||||
rc = sched_check_shared_satisfy(sched, 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(sched->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, sched) != 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_(sched, 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;
|
||||
}
|
||||
|
||||
for (i = 0; i < nclaimed; i++) {
|
||||
xce_alloc_t** allocs
|
||||
= sched_make_allocs(sched, "", claimed[i].host_req.count,
|
||||
(int*)claimed[i].host_req.idxs.data,
|
||||
(float*)claimed[i].host_req.vals.data,
|
||||
/* FIXME: no duration now */ NULL);
|
||||
|
||||
hallocs[i] = make_host_alloc(claimed[i].w->name,
|
||||
claimed[i].host_req.count, allocs);
|
||||
|
||||
xvec_idval_fini(&(claimed[i].host_req));
|
||||
}
|
||||
|
||||
ctx->alloc = make_job_alloc(0, nclaimed, hallocs);
|
||||
|
||||
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);
|
||||
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;
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -1,26 +1,33 @@
|
|||
/*
|
||||
* test_parjob_combinatorial.c - task 4.1 (scheduler-allocation spec):
|
||||
* unit tests for parjob_allocator.c's REAL combinatorial
|
||||
* xce_alloc_plan_iter_t implementation (Req 2 AC 1-6, Req 4 AC 1).
|
||||
* test_parjob_combinatorial.c - tasks 4.1 + 4.2 (scheduler-allocation
|
||||
* spec): unit tests for parjob_allocator.c's REAL combinatorial
|
||||
* xce_alloc_plan_iter_t implementation (Req 2 AC 1-6, Req 4 AC 1) AND
|
||||
* (task 4.2, this pass) the real per-bucket allocation loop
|
||||
* (my_alloc()/parjob_try_plan_()/parjob_prepare_allocation(), Req 4 AC
|
||||
* 2/4/5/6/7).
|
||||
*
|
||||
* Task 4.1's boundary is the plan-ITERATOR only -- parjob_allocator.c's
|
||||
* my_alloc()/parjob_prepare_allocation() bodies are untouched stubs,
|
||||
* still returning 0 unconditionally (task 4.2's job). So this file never
|
||||
* calls my_alloc(); it drives parjob_plan_iter_init()/_next()/_fini()
|
||||
* (static functions, hence the white-box #include below) directly,
|
||||
* exactly the way a future my_alloc() will.
|
||||
* Task 4.1's own tests below never call my_alloc() -- they drive
|
||||
* parjob_plan_iter_init()/_next()/_fini() (static, hence the white-box
|
||||
* #include) directly, and needed no fixture beyond a bare
|
||||
* xce_primary_resreq_t, since the pre-4.2 my_alloc()/parjob_prepare_
|
||||
* allocation() stub bodies made no calls into resource.c/schedobj.c/
|
||||
* core.c/host_iter.c at all. Left completely unmodified below.
|
||||
*
|
||||
* White-box #include convention: same as test_seqjob_allocator.c
|
||||
* (task 3.1), but simpler -- unlike seqjob_allocator.c's my_alloc(),
|
||||
* parjob_allocator.c's current my_alloc()/parjob_prepare_allocation()
|
||||
* stub bodies make no calls into resource.c/schedobj.c/core.c/host_iter.c
|
||||
* at all (confirmed by reading the file: both just touch struct fields
|
||||
* and return 0), so only ../src/plugins/parjob_allocator.c itself needs
|
||||
* to be pulled in, plus a link-satisfying stub for plugin_register()/
|
||||
* plugin_unregister() (referenced by register_parjob_allocator(), never
|
||||
* itself invoked by any test below).
|
||||
* Task 4.2 replaces that stub with a real implementation that DOES call
|
||||
* into those files (sched_try_alloc_on_host_(), sched_ctx_uncommit_host(),
|
||||
* sched_host_iter_init/next/fini(), sched_ctx_get_uncomit_shared(),
|
||||
* sched_check_shared_satisfy(), plugin_prepare_allocation()'s fan-out),
|
||||
* so this pass adds the SAME white-box #include combination
|
||||
* test_seqjob_allocator.c (task 3.1)/test_seqjob_rewrap_equivalence.c
|
||||
* (task 3.2) already established (core.c + resource.c + schedobj.c +
|
||||
* host_iter.c), plus the matching link-satisfying stubs (a
|
||||
* plugin_prepare_allocation() that delegates to the real
|
||||
* parjob_prepare_allocation() under test, and a sched_make_allocs()
|
||||
* stand-in that avoids white-box-including the rest of the 1382-line
|
||||
* scheduler.c) -- and a real fixture (registered "slots" resource, real
|
||||
* sched_add_worker() hosts) for the new bucket-loop tests.
|
||||
*
|
||||
* Covers:
|
||||
* Covers (task 4.1, unchanged):
|
||||
* - one-plan-per-call generation (Req 2 AC 2)
|
||||
* - most-preferred-first ordering: ascending host-count k = 1..N
|
||||
* (Req 2 AC 3, Req 4 AC 1)
|
||||
|
|
@ -40,32 +47,62 @@
|
|||
* - Req 2 AC 6: next() calls pthread_testcancel() (or an equivalent
|
||||
* real cancellation point) -- proved behaviorally, same technique
|
||||
* test_host_iter.c already established for host_iter_next().
|
||||
*
|
||||
* Covers (task 4.2, new):
|
||||
* - multi-bucket success through the real my_alloc() end to end (Req
|
||||
* 4 AC 2/AC 3), including per-bucket resource want scaling with that
|
||||
* bucket's own slot count, not a fixed value (Req 4 AC 5/AC 6)
|
||||
* - the descending-vs-ascending worked example from requirements.md's
|
||||
* Composition Algorithm, driven directly through parjob_try_plan_()
|
||||
* with a hand-built plan for each order against the identical host
|
||||
* pool (Req 4 AC 2)
|
||||
* - a mid-plan bucket failure correctly reversing the earlier
|
||||
* successful bucket's provisional debit before the next candidate
|
||||
* plan is tried, proved by that same host being successfully
|
||||
* reclaimed by the later, winning plan (Req 4 AC 2, "Rollback scope")
|
||||
* - Req 4 AC 4: the same not-satisfied signal (0) once every plan is
|
||||
* exhausted
|
||||
*/
|
||||
#include <errno.h>
|
||||
#include <pthread.h>
|
||||
#include <rtti.h>
|
||||
#include <stdlib.h>
|
||||
#include <string.h>
|
||||
#include <unistd.h>
|
||||
|
||||
#include <unity/unity.h>
|
||||
|
||||
#include "resobj.h"
|
||||
#include "resreq.h"
|
||||
#include "sched/alloc_iter.h"
|
||||
#include "scheduler.h"
|
||||
#include "xcealloc.h"
|
||||
|
||||
/* Pull in parjob_allocator.c itself (the file under test) -- the plan
|
||||
* iterator functions (parjob_plan_iter_init/next/fini) are `static`, so
|
||||
* this white-box #include is the only way to reach them directly,
|
||||
* matching test_seqjob_allocator.c's own established convention. */
|
||||
/* Pull in the real xce_sched_core_init()/sched_add_worker() (core.c),
|
||||
* sched_meet_host_selection()/sched_check_host_satisfy()/
|
||||
* sched_try_alloc_on_host_()/sched_ctx_uncommit_host()/put_resvals()
|
||||
* (resource.c), the real sched_ctx_init()/sched_ctx_get_uncomit_*()
|
||||
* (schedobj.c), the shared host_iter.c implementation, and
|
||||
* parjob_allocator.c itself (the file under test) into this TU -- same
|
||||
* combination test_seqjob_allocator.c (task 3.1) established for its own
|
||||
* sibling plugin. The plan-iterator and (new, task 4.2) bucket-loop
|
||||
* functions are all `static`, so this white-box #include is the only way
|
||||
* to reach them directly.
|
||||
*/
|
||||
#include "../src/core.c"
|
||||
#include "../src/host_iter.c"
|
||||
#include "../src/plugins/parjob_allocator.c"
|
||||
#include "../src/resource.c"
|
||||
#include "../src/schedobj.c"
|
||||
|
||||
/* ---------------------------------------------------------------------- */
|
||||
/* Link-satisfying stubs */
|
||||
/* ---------------------------------------------------------------------- */
|
||||
|
||||
/* plugin_register()/plugin_unregister() are referenced by
|
||||
* register_parjob_allocator() (parjob_allocator.c), but no test below
|
||||
* calls register_parjob_allocator() -- the plan iterator is exercised
|
||||
* register_parjob_allocator() (parjob_allocator.c) and core.c
|
||||
* respectively, but no test below calls register_parjob_allocator() --
|
||||
* the plan iterator and my_alloc()/parjob_try_plan_() are exercised
|
||||
* directly. Never actually invoked by anything this file's tests
|
||||
* exercise; provided only so the TU links. */
|
||||
int plugin_register(void* sched, void* plugin)
|
||||
|
|
@ -77,10 +114,187 @@ int plugin_register(void* sched, void* plugin)
|
|||
|
||||
void plugin_unregister(void* arg) { (void)arg; }
|
||||
|
||||
/* Delegates to the real parjob_prepare_allocation() under test, standing
|
||||
* in for plugin_prepare_allocation()'s (plugin.c) real fan-out across
|
||||
* every registered plugin -- exactly one (parjob's own) in this file's
|
||||
* fixtures. Mirrors plugin.c:591's own reset-before-fan-out step, same
|
||||
* rationale as test_seqjob_allocator.c's identical stub. */
|
||||
int plugin_prepare_allocation(void* sched, xce_sched_ctx_t* ctx)
|
||||
{
|
||||
xce_sched_plugin_t fake_plugin;
|
||||
|
||||
if (ctx->worker == NULL) {
|
||||
xvec_idval_fini(&(ctx->alloc_req.shared));
|
||||
xvec_idval_init(&(ctx->alloc_req.shared), 8);
|
||||
} else {
|
||||
xvec_idval_fini(&(ctx->alloc_req.host));
|
||||
xvec_idval_init(&(ctx->alloc_req.host), 8);
|
||||
}
|
||||
|
||||
memset(&fake_plugin, 0, sizeof(fake_plugin));
|
||||
fake_plugin.sched = sched;
|
||||
|
||||
return parjob_prepare_allocation(&fake_plugin, ctx);
|
||||
}
|
||||
|
||||
int plugin_job_update(void* s, xce_sched_job_t* j)
|
||||
{
|
||||
(void)s;
|
||||
(void)j;
|
||||
return 0;
|
||||
}
|
||||
int plugin_queue_update(void* s, xce_sched_queue_t* q)
|
||||
{
|
||||
(void)s;
|
||||
(void)q;
|
||||
return 0;
|
||||
}
|
||||
int plugin_job_rusage_update(void* s, xce_sched_job_t* j, xce_job_rusages_t* r)
|
||||
{
|
||||
(void)s;
|
||||
(void)j;
|
||||
(void)r;
|
||||
return 0;
|
||||
}
|
||||
|
||||
/* Avoids white-box-including the rest of scheduler.c just to build the
|
||||
* leaf xce_alloc_t records my_alloc()'s success path constructs -- same
|
||||
* rationale/shape as test_seqjob_allocator.c's identical stub. This
|
||||
* file's own quantitative per-host assertions go through
|
||||
* assert_uncommitted_host_debit() (real ctx->uncomitted state, untouched
|
||||
* by this stub) rather than these fake xce_alloc_t's fields, exactly like
|
||||
* the seqjob equivalence tests already do. */
|
||||
xce_alloc_t** sched_make_allocs(xce_scheduler_t* sched, char* reserve, int nres,
|
||||
int* idxs, float* vals, int* duration)
|
||||
{
|
||||
xce_alloc_t** allocs;
|
||||
int i;
|
||||
|
||||
(void)sched;
|
||||
(void)reserve;
|
||||
(void)idxs;
|
||||
(void)vals;
|
||||
(void)duration;
|
||||
|
||||
if (nres <= 0) {
|
||||
return NULL;
|
||||
}
|
||||
|
||||
allocs = (xce_alloc_t**)calloc(nres, sizeof(*allocs));
|
||||
if (allocs == NULL) {
|
||||
return NULL;
|
||||
}
|
||||
|
||||
for (i = 0; i < nres; i++) {
|
||||
allocs[i] = (xce_alloc_t*)calloc(1, sizeof(xce_alloc_t));
|
||||
}
|
||||
|
||||
return allocs;
|
||||
}
|
||||
|
||||
/* ---------------------------------------------------------------------- */
|
||||
/* Helpers */
|
||||
/* Fixture helpers */
|
||||
/* ---------------------------------------------------------------------- */
|
||||
|
||||
/* register_slots_resource - the real makeResource() (resobj.c), giving
|
||||
* "slots" a resmgr-assigned idx -- identical to test_seqjob_allocator.c's
|
||||
* own helper; parjob_prepare_allocation()'s key_find(resmgr, "slots",...)
|
||||
* call needs this to be a genuinely registered resource. */
|
||||
static xce_resource_t* register_slots_resource(var resmgr)
|
||||
{
|
||||
xce_resource_t res;
|
||||
|
||||
memset(&res, 0, sizeof(res));
|
||||
res.name = strdup("slots");
|
||||
|
||||
return makeResource(resmgr, &res);
|
||||
}
|
||||
|
||||
/* add_worker_with_slots - a real sched_add_worker() (core.c) worker
|
||||
* carrying `slots_total` free "slots" (idx == the resmgr-assigned index),
|
||||
* identical to test_seqjob_allocator.c's own helper. */
|
||||
static xce_sched_worker_t* add_worker_with_slots(xce_sched_core_t* core,
|
||||
const char* name,
|
||||
int slots_idx,
|
||||
float slots_total)
|
||||
{
|
||||
xce_sched_worker_t* w;
|
||||
int idx[1];
|
||||
float val[1];
|
||||
int rc;
|
||||
|
||||
w = (xce_sched_worker_t*)calloc(1, sizeof(*w));
|
||||
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);
|
||||
|
||||
idx[0] = slots_idx;
|
||||
val[0] = slots_total;
|
||||
put_resvals(&(w->resvals), 1, idx, val);
|
||||
val[0] = 0.0f;
|
||||
put_resvals(&(w->resInUse), 1, idx, val);
|
||||
|
||||
rc = sched_add_worker(core, w);
|
||||
TEST_ASSERT_TRUE(rc >= 0);
|
||||
|
||||
return w;
|
||||
}
|
||||
|
||||
static void init_primary_obj(xce_sched_obj_t* obj,
|
||||
xce_primary_schedobj_t* primary_obj,
|
||||
xce_primary_resreq_t* primary)
|
||||
{
|
||||
memset(primary_obj, 0, sizeof(*primary_obj));
|
||||
primary_obj->resreq = primary;
|
||||
|
||||
memset(obj, 0, sizeof(*obj));
|
||||
obj->tag = XCE_RESREQ_PRIMARY;
|
||||
obj->u.primary = primary_obj;
|
||||
}
|
||||
|
||||
/* drive_my_alloc - the exact production sequence a job's allocation
|
||||
* 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
|
||||
* 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,
|
||||
xce_primary_resreq_t* primary, xce_sched_ctx_t* ctx)
|
||||
{
|
||||
xce_sched_plugin_t plugin;
|
||||
|
||||
TEST_ASSERT_EQUAL_INT(0, sched_ctx_init(ctx, sched, obj));
|
||||
sched_ctx_set_primary(ctx, primary);
|
||||
|
||||
memset(&plugin, 0, sizeof(plugin));
|
||||
plugin.sched = sched;
|
||||
|
||||
return my_alloc(&plugin, ctx);
|
||||
}
|
||||
|
||||
/* assert_uncommitted_host_debit - reads ctx->uncomitted.hosts[w->idx] for
|
||||
* resource residx and asserts it equals `expected`. Identical to
|
||||
* test_seqjob_rewrap_equivalence.c's own helper. */
|
||||
static void assert_uncommitted_host_debit(xce_sched_ctx_t* ctx,
|
||||
xce_sched_worker_t* w, int residx,
|
||||
float expected)
|
||||
{
|
||||
xvec_t* uncomit = NULL;
|
||||
int rc;
|
||||
|
||||
rc = sched_ctx_get_uncomit_host(&uncomit, ctx, w->idx);
|
||||
TEST_ASSERT_EQUAL_INT(1, rc);
|
||||
TEST_ASSERT_NOT_NULL(uncomit);
|
||||
TEST_ASSERT_EQUAL_FLOAT(expected, ((float*)uncomit->data)[residx]);
|
||||
}
|
||||
|
||||
static void bind_plan_iter(xce_alloc_plan_iter_t* iter)
|
||||
{
|
||||
memset(iter, 0, sizeof(*iter));
|
||||
|
|
@ -386,6 +600,369 @@ static void test_next_has_a_real_cancellation_point(void)
|
|||
iter->fini(iter);
|
||||
}
|
||||
|
||||
/* ---------------------------------------------------------------------- */
|
||||
/* Task 4.2 tests: the real per-bucket allocation loop */
|
||||
/* ---------------------------------------------------------------------- */
|
||||
|
||||
/*
|
||||
* Req 4 AC 2/AC 3/AC 5/AC 6: a real multi-bucket plan, driven end to end
|
||||
* through the real my_alloc() (not parjob_try_plan_() directly). Two
|
||||
* hosts are engineered so the outcome is deterministic REGARDLESS of
|
||||
* sched->core.workers' own hash-based iteration order: host-A has exactly
|
||||
* 2 free slots (only it can satisfy a "need 2" bucket -- host-B's 1 free
|
||||
* slot never qualifies) and host-B has exactly 1 free slot. maxTasks=3
|
||||
* forces k=1 ([3]) to fail first (no single host has >=3), so k=2's
|
||||
* [2,1] plan is the one that actually succeeds -- proving the bucket loop
|
||||
* itself works, not just the trivial single-bucket/single-plan case.
|
||||
* Also proves Req 4 AC 5/AC 6: parjob_prepare_allocation() must have
|
||||
* staged a "slots" want of exactly 2 (not a fixed 1, and not maxTasks=3)
|
||||
* for host-A's bucket, and exactly 1 for host-B's -- an unfixed stub or a
|
||||
* fixed-value bug would either allocate nothing or debit the wrong
|
||||
* amount, and assert_uncommitted_host_debit() below would catch either.
|
||||
*/
|
||||
static void test_multi_bucket_plan_succeeds_through_real_my_alloc(void)
|
||||
{
|
||||
xce_scheduler_t sched;
|
||||
xce_resource_t* slots_res;
|
||||
xce_sched_worker_t* host_a;
|
||||
xce_sched_worker_t* host_b;
|
||||
xce_sched_obj_t obj;
|
||||
xce_primary_schedobj_t primary_obj;
|
||||
xce_primary_resreq_t primary;
|
||||
xce_sched_ctx_t ctx;
|
||||
int rc;
|
||||
int i;
|
||||
int seen_a = 0;
|
||||
int seen_b = 0;
|
||||
|
||||
memset(&sched, 0, sizeof(sched));
|
||||
TEST_ASSERT_EQUAL_INT(0, xce_sched_core_init(&(sched.core)));
|
||||
|
||||
slots_res = register_slots_resource(sched.core.resmgr);
|
||||
TEST_ASSERT_NOT_NULL(slots_res);
|
||||
|
||||
host_a
|
||||
= add_worker_with_slots(&(sched.core), "host-a", slots_res->idx, 2.0f);
|
||||
host_b
|
||||
= add_worker_with_slots(&(sched.core), "host-b", slots_res->idx, 1.0f);
|
||||
|
||||
memset(&primary, 0, sizeof(primary));
|
||||
primary.minTasks = 1;
|
||||
primary.maxTasks = 3;
|
||||
primary.select = NULL;
|
||||
|
||||
init_primary_obj(&obj, &primary_obj, &primary);
|
||||
|
||||
rc = drive_my_alloc(&sched, &obj, &primary, &ctx);
|
||||
|
||||
TEST_ASSERT_EQUAL_INT(1, rc);
|
||||
TEST_ASSERT_NOT_NULL(ctx.alloc);
|
||||
TEST_ASSERT_EQUAL_INT(2, ctx.alloc->nhosts);
|
||||
|
||||
for (i = 0; i < ctx.alloc->nhosts; i++) {
|
||||
xce_host_alloc_t* ha = ctx.alloc->allocs[i];
|
||||
|
||||
TEST_ASSERT_NOT_NULL(ha);
|
||||
TEST_ASSERT_EQUAL_INT(1, ha->nallocs); /* "slots" only */
|
||||
if (strcmp(ha->host, "host-a") == 0) {
|
||||
seen_a = 1;
|
||||
} else if (strcmp(ha->host, "host-b") == 0) {
|
||||
seen_b = 1;
|
||||
}
|
||||
}
|
||||
TEST_ASSERT_TRUE(seen_a);
|
||||
TEST_ASSERT_TRUE(seen_b);
|
||||
|
||||
/* host-a's own bucket needed 2 slots, host-b's needed 1 -- the exact
|
||||
* per-bucket want Req 4 AC 5/AC 6 requires, not a fixed 1 (seqjob's
|
||||
* old hardcode) or maxTasks=3 (a different, equally wrong fixed
|
||||
* value). */
|
||||
assert_uncommitted_host_debit(&ctx, host_a, slots_res->idx, 2.0f);
|
||||
assert_uncommitted_host_debit(&ctx, host_b, slots_res->idx, 1.0f);
|
||||
|
||||
sched_ctx_fini(&ctx);
|
||||
}
|
||||
|
||||
/*
|
||||
* Req 4 AC 2: the exact descending-vs-ascending worked example from
|
||||
* requirements.md's Composition Algorithm. Two hosts, capacities 1 and 2;
|
||||
* WHICH physical host holds which capacity is assigned by a real,
|
||||
* independent iter_make()/iter_next() scan (the identical primitive
|
||||
* host_iter.c itself uses over the same sched.core.workers) so the test
|
||||
* does not depend on assuming insertion order matches hash-table
|
||||
* iteration order -- the same technique
|
||||
* test_seqjob_rewrap_equivalence.c's own iteration-order fixture uses.
|
||||
* The 2-free-slot capacity goes to whichever host is visited FIRST,
|
||||
* matching requirements.md's own worked example precisely (a fresh scan
|
||||
* for a small bucket greedily claims whatever it encounters first that
|
||||
* already satisfies it).
|
||||
*
|
||||
* Driven directly through parjob_try_plan_() (not the real plan
|
||||
* iterator, which only ever generates descending order) with a
|
||||
* hand-built xce_alloc_plan_t for each order, against the SAME
|
||||
* unmodified host pool (the descending attempt's claims are rolled back
|
||||
* before the ascending attempt runs) -- this is the seam task 4.2's own
|
||||
* bucket-loop factoring (out of my_alloc()) exists to make testable
|
||||
* without reimplementing the loop in the test.
|
||||
*/
|
||||
static void test_descending_bucket_order_succeeds_where_ascending_fails(void)
|
||||
{
|
||||
xce_scheduler_t sched;
|
||||
xce_resource_t* slots_res;
|
||||
xce_sched_worker_t* host_first;
|
||||
xce_sched_worker_t* host_second;
|
||||
iterator_t it;
|
||||
xce_sched_obj_t obj;
|
||||
xce_primary_schedobj_t primary_obj;
|
||||
xce_primary_resreq_t primary;
|
||||
xce_sched_ctx_t ctx;
|
||||
xce_alloc_plan_t plan;
|
||||
int bucket_slots_desc[2];
|
||||
int bucket_slots_asc[2];
|
||||
parjob_claim_t claimed[2];
|
||||
int nclaimed;
|
||||
int rc;
|
||||
int idx[1];
|
||||
float val[1];
|
||||
|
||||
memset(&sched, 0, sizeof(sched));
|
||||
TEST_ASSERT_EQUAL_INT(0, xce_sched_core_init(&(sched.core)));
|
||||
|
||||
slots_res = register_slots_resource(sched.core.resmgr);
|
||||
TEST_ASSERT_NOT_NULL(slots_res);
|
||||
|
||||
/* Added with a placeholder capacity -- real capacities are assigned
|
||||
* below, once the shared iteration order is known. */
|
||||
add_worker_with_slots(&(sched.core), "host-1", slots_res->idx, 0.0f);
|
||||
add_worker_with_slots(&(sched.core), "host-2", slots_res->idx, 0.0f);
|
||||
|
||||
rc = iter_make(&it, sched.core.workers);
|
||||
TEST_ASSERT_TRUE(rc >= 0);
|
||||
rc = iter_next(&it, (void**)&host_first);
|
||||
TEST_ASSERT_TRUE(rc > 0);
|
||||
rc = iter_next(&it, (void**)&host_second);
|
||||
TEST_ASSERT_TRUE(rc > 0);
|
||||
iter_destroy(&it);
|
||||
|
||||
idx[0] = slots_res->idx;
|
||||
val[0] = 2.0f;
|
||||
put_resvals(&(host_first->resvals), 1, idx, val);
|
||||
val[0] = 1.0f;
|
||||
put_resvals(&(host_second->resvals), 1, idx, val);
|
||||
|
||||
memset(&primary, 0, sizeof(primary));
|
||||
primary.minTasks = 1;
|
||||
primary.maxTasks = 3;
|
||||
primary.select = NULL;
|
||||
|
||||
init_primary_obj(&obj, &primary_obj, &primary);
|
||||
TEST_ASSERT_EQUAL_INT(0, sched_ctx_init(&ctx, &sched, &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,
|
||||
* 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
|
||||
* ascending attempt below deliberately exercises (bucket1 failing
|
||||
* "need 2" against host_second's 1 free slot). */
|
||||
ctx.reason = make_pendreason(0);
|
||||
TEST_ASSERT_NOT_NULL(ctx.reason);
|
||||
|
||||
/* Descending [2,1] -- the plan iterator's own v1 default order --
|
||||
* succeeds: bucket0 ("need 2") can only be satisfied by host_first
|
||||
* (2 free); bucket1 ("need 1") then claims host_second (1 free),
|
||||
* the only host left. */
|
||||
bucket_slots_desc[0] = 2;
|
||||
bucket_slots_desc[1] = 1;
|
||||
memset(&plan, 0, sizeof(plan));
|
||||
plan.nbuckets = 2;
|
||||
plan.bucket_slots = bucket_slots_desc;
|
||||
|
||||
rc = parjob_try_plan_(&sched, &ctx, &plan, claimed, &nclaimed);
|
||||
TEST_ASSERT_EQUAL_INT(1, rc);
|
||||
TEST_ASSERT_EQUAL_INT(2, nclaimed);
|
||||
TEST_ASSERT_TRUE(claimed[0].w == host_first);
|
||||
TEST_ASSERT_TRUE(claimed[1].w == host_second);
|
||||
|
||||
/* Roll back before trying the ascending variant against the exact
|
||||
* same, otherwise-untouched host pool. */
|
||||
parjob_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);
|
||||
|
||||
/* Ascending [1,2] -- the SAME host pool -- fails: bucket0 ("need 1")
|
||||
* greedily claims host_first (2 free, visited first, trivially
|
||||
* satisfies "need 1" even though it is overqualified), leaving only
|
||||
* host_second (1 free) for bucket1 ("need 2"), which cannot satisfy
|
||||
* it. */
|
||||
bucket_slots_asc[0] = 1;
|
||||
bucket_slots_asc[1] = 2;
|
||||
memset(&plan, 0, sizeof(plan));
|
||||
plan.nbuckets = 2;
|
||||
plan.bucket_slots = bucket_slots_asc;
|
||||
|
||||
rc = parjob_try_plan_(&sched, &ctx, &plan, claimed, &nclaimed);
|
||||
TEST_ASSERT_EQUAL_INT(0, rc);
|
||||
TEST_ASSERT_EQUAL_INT(0, nclaimed);
|
||||
|
||||
/* No phantom debit left behind by the failed ascending attempt. */
|
||||
assert_uncommitted_host_debit(&ctx, host_first, slots_res->idx, 0.0f);
|
||||
assert_uncommitted_host_debit(&ctx, host_second, slots_res->idx, 0.0f);
|
||||
|
||||
sched_ctx_fini(&ctx);
|
||||
}
|
||||
|
||||
/*
|
||||
* Req 4 AC 2 ("Rollback scope"): a mid-plan bucket failure must reverse
|
||||
* every earlier bucket's provisional debit in that SAME plan attempt
|
||||
* before the next candidate plan is tried. Proved the strong way -- not
|
||||
* just "the debit reads back as zero" but "the SAME host is successfully
|
||||
* reclaimed by a LATER plan attempt for a full, real allocation," which
|
||||
* could only happen if the earlier phantom debit was genuinely reversed
|
||||
* (sched_check_host_satisfy() would otherwise reject the later attempt's
|
||||
* own fresh want against a host that still looked partially consumed).
|
||||
*
|
||||
* Fixture: h1 (2 free), h2/h3/h4 (1 free each), maxTasks=4.
|
||||
* k=1 ([4]): fails (no single host has >=4).
|
||||
* k=2 ([2,2]): bucket0 ("need 2") can only be satisfied by h1 (the only
|
||||
* host with >=2 free) -- claims h1. bucket1 ("need 2"), excluding h1,
|
||||
* has only h2/h3/h4 left, each with just 1 free -- ALL fail. The
|
||||
* whole k=2 plan fails, and h1's provisional debit from bucket0 must
|
||||
* be rolled back here.
|
||||
* k=3 ([2,1,1]): a fresh, empty claimed[] (independent of k=2's
|
||||
* rolled-back attempt) -- bucket0 ("need 2") again only h1 qualifies;
|
||||
* if k=2's debit on h1 had NOT been reversed, h1 would appear to
|
||||
* already carry a 2.0 phantom debit and sched_check_host_satisfy()
|
||||
* would reject this fresh "need 2" want against it (2 total < 2
|
||||
* phantom + 2 new) -- the whole allocation would fail outright
|
||||
* instead of succeeding. bucket1/bucket2 ("need 1" each) then claim
|
||||
* two of the three remaining 1-free-slot hosts. SUCCEEDS with exactly
|
||||
* 3 claimed hosts.
|
||||
*/
|
||||
static void
|
||||
test_mid_plan_bucket_failure_rolls_back_before_later_plan_succeeds(void)
|
||||
{
|
||||
xce_scheduler_t sched;
|
||||
xce_resource_t* slots_res;
|
||||
xce_sched_worker_t* h1;
|
||||
xce_sched_worker_t* h2;
|
||||
xce_sched_worker_t* h3;
|
||||
xce_sched_worker_t* h4;
|
||||
xce_sched_obj_t obj;
|
||||
xce_primary_schedobj_t primary_obj;
|
||||
xce_primary_resreq_t primary;
|
||||
xce_sched_ctx_t ctx;
|
||||
int rc;
|
||||
int i;
|
||||
int seen_h1 = 0;
|
||||
int nother = 0;
|
||||
xvec_t* u2 = NULL;
|
||||
xvec_t* u3 = NULL;
|
||||
xvec_t* u4 = NULL;
|
||||
float other_sum;
|
||||
|
||||
memset(&sched, 0, sizeof(sched));
|
||||
TEST_ASSERT_EQUAL_INT(0, xce_sched_core_init(&(sched.core)));
|
||||
|
||||
slots_res = register_slots_resource(sched.core.resmgr);
|
||||
TEST_ASSERT_NOT_NULL(slots_res);
|
||||
|
||||
h1 = add_worker_with_slots(&(sched.core), "h1", slots_res->idx, 2.0f);
|
||||
h2 = add_worker_with_slots(&(sched.core), "h2", slots_res->idx, 1.0f);
|
||||
h3 = add_worker_with_slots(&(sched.core), "h3", slots_res->idx, 1.0f);
|
||||
h4 = add_worker_with_slots(&(sched.core), "h4", slots_res->idx, 1.0f);
|
||||
|
||||
memset(&primary, 0, sizeof(primary));
|
||||
primary.minTasks = 1;
|
||||
primary.maxTasks = 4;
|
||||
primary.select = NULL;
|
||||
|
||||
init_primary_obj(&obj, &primary_obj, &primary);
|
||||
|
||||
rc = drive_my_alloc(&sched, &obj, &primary, &ctx);
|
||||
|
||||
TEST_ASSERT_EQUAL_INT(1, rc);
|
||||
TEST_ASSERT_NOT_NULL(ctx.alloc);
|
||||
TEST_ASSERT_EQUAL_INT(3, ctx.alloc->nhosts);
|
||||
|
||||
for (i = 0; i < ctx.alloc->nhosts; i++) {
|
||||
xce_host_alloc_t* ha = ctx.alloc->allocs[i];
|
||||
|
||||
TEST_ASSERT_NOT_NULL(ha);
|
||||
if (strcmp(ha->host, "h1") == 0) {
|
||||
seen_h1++;
|
||||
} else {
|
||||
nother++;
|
||||
}
|
||||
}
|
||||
TEST_ASSERT_EQUAL_INT(1, seen_h1);
|
||||
TEST_ASSERT_EQUAL_INT(2, nother);
|
||||
|
||||
/* THE regression assertion: h1 carries exactly its own bucket's 2.0
|
||||
* want in the final, committed state -- not 4.0 (which a missing
|
||||
* rollback would leave: 2.0 phantom from the failed k=2 attempt plus
|
||||
* 2.0 genuine from k=3's own successful claim). */
|
||||
assert_uncommitted_host_debit(&ctx, h1, slots_res->idx, 2.0f);
|
||||
|
||||
/* Exactly two of h2/h3/h4 were claimed at 1.0 each (the third stayed
|
||||
* at 0.0, host_iter's own internal candidate order decides which) --
|
||||
* proves neither a double-debit nor a leaked phantom debit from the
|
||||
* rolled-back k=2 attempt. */
|
||||
TEST_ASSERT_EQUAL_INT(1, sched_ctx_get_uncomit_host(&u2, &ctx, h2->idx));
|
||||
TEST_ASSERT_EQUAL_INT(1, sched_ctx_get_uncomit_host(&u3, &ctx, h3->idx));
|
||||
TEST_ASSERT_EQUAL_INT(1, sched_ctx_get_uncomit_host(&u4, &ctx, h4->idx));
|
||||
other_sum = ((float*)u2->data)[slots_res->idx]
|
||||
+ ((float*)u3->data)[slots_res->idx]
|
||||
+ ((float*)u4->data)[slots_res->idx];
|
||||
TEST_ASSERT_EQUAL_FLOAT(2.0f, other_sum);
|
||||
|
||||
sched_ctx_fini(&ctx);
|
||||
}
|
||||
|
||||
/*
|
||||
* Req 4 AC 4: when every candidate plan is exhausted without a full
|
||||
* success, my_alloc() returns the same not-satisfied signal (0) the old
|
||||
* stub always returned, unconditionally -- pend-reason handling
|
||||
* downstream is unaffected by this task's change.
|
||||
*/
|
||||
static void test_returns_zero_when_every_plan_is_exhausted(void)
|
||||
{
|
||||
xce_scheduler_t sched;
|
||||
xce_resource_t* slots_res;
|
||||
xce_sched_obj_t obj;
|
||||
xce_primary_schedobj_t primary_obj;
|
||||
xce_primary_resreq_t primary;
|
||||
xce_sched_ctx_t ctx;
|
||||
int rc;
|
||||
|
||||
memset(&sched, 0, sizeof(sched));
|
||||
TEST_ASSERT_EQUAL_INT(0, xce_sched_core_init(&(sched.core)));
|
||||
|
||||
slots_res = register_slots_resource(sched.core.resmgr);
|
||||
TEST_ASSERT_NOT_NULL(slots_res);
|
||||
|
||||
/* A single host with only 1 free slot can never satisfy ANY plan for
|
||||
* a 3-task job (k=1 needs 3 on one host; k=2/k=3 each need a SECOND
|
||||
* distinct host that does not exist). */
|
||||
add_worker_with_slots(&(sched.core), "lonely-host", slots_res->idx, 1.0f);
|
||||
|
||||
memset(&primary, 0, sizeof(primary));
|
||||
primary.minTasks = 1;
|
||||
primary.maxTasks = 3;
|
||||
primary.select = NULL;
|
||||
|
||||
init_primary_obj(&obj, &primary_obj, &primary);
|
||||
|
||||
rc = drive_my_alloc(&sched, &obj, &primary, &ctx);
|
||||
|
||||
TEST_ASSERT_EQUAL_INT(0, rc);
|
||||
TEST_ASSERT_NULL(ctx.alloc);
|
||||
|
||||
sched_ctx_fini(&ctx);
|
||||
}
|
||||
|
||||
void setUp(void) { return; }
|
||||
|
||||
void tearDown(void) { return; }
|
||||
|
|
@ -399,5 +976,10 @@ int main(void)
|
|||
RUN_TEST(test_plan_generation_ignores_scheduler_state);
|
||||
RUN_TEST(test_init_rejects_invalid_input);
|
||||
RUN_TEST(test_next_has_a_real_cancellation_point);
|
||||
RUN_TEST(test_multi_bucket_plan_succeeds_through_real_my_alloc);
|
||||
RUN_TEST(test_descending_bucket_order_succeeds_where_ascending_fails);
|
||||
RUN_TEST(
|
||||
test_mid_plan_bucket_failure_rolls_back_before_later_plan_succeeds);
|
||||
RUN_TEST(test_returns_zero_when_every_plan_is_exhausted);
|
||||
return UNITY_END();
|
||||
}
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue