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

feat(scheduler-allocation): rewrap seqjob allocation onto shared iterators

Browse source 
Restructure seqjob's my_alloc() as a trivial single-bucket case over the
shared xce_alloc_plan_iter_t/xce_host_iter_t contracts, replacing its own
inline worker-collection loop. Guard seqjob_prepare_allocation() to
maxTasks == 1, closing the fan-out gap for the parallel allocator's later
fix (Req 9.1, 4.6).

Export sched_try_alloc_on_host_() from resource.c (drop static/va_list,
plain (worker, sched, ctx) signature) so seqjob calls the real per-host
try-alloc leaf instead of a private duplicate; delete the now-dead
sched_try_alloc_on_hosts() variadic wrapper. Also wires host_iter.c into
the production scheduler_srcs list (task 2.1 only added it to testsuite).

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01XYdtrEWWZfKRRGED7raEYg
This commit is contained in:
Rongsong Shen 2026-09-04 18:48:49 +00:00
parent 86c86352d1
commit b18c5f780d
No known key found for this signature in database
GPG key ID: 5BCDAAE0123E9FE2
7 changed files with 602 additions and 37 deletions
Download patch file Download diff file Expand all files Collapse all files

10
src/scheduler/include/scheduler.h
View file

@ -311,7 +311,15 @@ extern int sched_collect_workers(xce_scheduler_t*, xlist_t*,
extern uint64_t sched_get_id(xce_scheduler_t*);
extern xce_alloc_t** sched_make_allocs(xce_scheduler_t*, char*, int, int*,
float*, int*);
extern int sched_try_alloc_on_hosts(xce_worker_group_t*, va_list);
/* scheduler-allocation spec, task 3.1 remediation: exported (was `static`,
* wrapped by a variadic sched_try_alloc_on_hosts() whose sole caller was
* seqjob_allocator.c's pre-rewrap xlist_find() call site -- both now
* deleted) so both seqjob_allocator.c and parjob_allocator.c can call this
* shared Composition-Algorithm try_alloc leaf directly with real
* parameters, instead of each carrying its own private copy. */
extern int sched_try_alloc_on_host_(xce_sched_worker_t*,
xce_scheduler_t*,
xce_sched_ctx_t*);
extern int get_resvals(zarray_t*, int, int*, float*);
extern int put_resvals(zarray_t*, int, int*, float*);
extern request_res_t* make_resource_request(char*, float);

1
src/scheduler/meson.build
View file

@ -29,6 +29,7 @@ scheduler_srcs = [
'src/plugin.c',
'src/resource.c',
'src/alloc_cache.c',
'src/host_iter.c',
'src/plugins/fcfs.c',
'src/plugins/builtin_fcfs.c',
'src/plugins/seqjob_allocator.c',

205
src/scheduler/src/plugins/seqjob_allocator.c
View file

@ -9,11 +9,11 @@
#include "kvmgr.h"
#include "plugin.h"
#include "preason.h"
#include "sched/alloc_iter.h"
#include "scheduler.h"
#include "utils.h"
#include "xce/types.h"
#include "xcealloc.h"
#include "xlist.h"
#include "xvec.h"
static int seqjob_allocator_init(xce_sched_plugin_t* plugin)
@ -32,16 +32,6 @@ static void seqjob_allocator_fini(xce_sched_plugin_t* plugin)
FREEUP(plugin);
}
static int skip_bad_worker_(xce_sched_worker_t* w)
{
if (w == NULL) {
return 1;
}
/* add more check here */
return 0;
}
static int seqjob_prepare_allocation(void* arg, xce_sched_ctx_t* ctx)
{
xce_sched_plugin_t* plugin = (xce_sched_plugin_t*)arg;
@ -71,6 +61,24 @@ static int seqjob_prepare_allocation(void* arg, xce_sched_ctx_t* ctx)
return -EINVAL;
}
if (primary->maxTasks != 1) {
/* Req 4 AC 6: seqjob only ever handles maxTasks == 1 jobs -- this
* is a no-op for seqjob's own observable behavior. It is required,
* not optional cleanup, once parjob_prepare_allocation() stops
* being a no-op: plugin_prepare_allocation() (plugin.c:591)
* unconditionally fans out to every registered plugin's
* prepare_allocation_hook, and xvec_idval_store() (xvec.c:247)
* appends (idx, val) pairs with no dedup by resource idx. Without
* this guard, an unguarded seqjob_prepare_allocation() would ALSO
* fire for the same maxTasks > 1 job and append a second, wrong
* (nslots=1) entry for the same resource idx into ctx->alloc_req,
* corrupting the availability check parjob_allocator's own
* combinatorial allocation depends on. Mirrors the existing
* maxTasks == 1 gate parjob_prepare_allocation() already has (this
* file's own sibling, parjob_allocator.c). */
return 0;
}
if (ctx->worker == NULL) {
ctx->nhosts = 1;
ctx->nslots = 1;
@ -104,15 +112,105 @@ static int seqjob_prepare_allocation(void* arg, xce_sched_ctx_t* ctx)
return 0;
}
/*
* seqjob's trivial xce_alloc_plan_iter_t (Req 2/9.1, "Who Uses This
* Composition" table, requirements.md): seqjob only ever handles
* maxTasks == 1 jobs (seqjob_prepare_allocation()'s maxTasks==1 guard
* above is the other half of that invariant), so there is exactly one
* candidate plan -- a single bucket needing a single slot on a single
* host -- and no dynamic plan-generation state to track beyond "has
* next() already been served once." Same calloc-in-init()/free-in-fini()
* convention host_iter.c (task 2.1) already established for the sibling
* xce_host_iter_t -- no generic reusable "trivial iterator" abstraction,
* just this one allocator's own fixed shape.
*/
typedef struct {
xce_alloc_plan_t plan;
int bucket_slots[1];
int served;
} seqjob_plan_iter_priv_t;
static int seqjob_plan_iter_init(xce_alloc_plan_iter_t* iter,
xce_primary_resreq_t* resreq,
xce_scheduler_t* sched)
{
seqjob_plan_iter_priv_t* priv;
/* Trivial plan: fixed shape (1 host x 1 slot), independent of the
* resreq/scheduler state -- both parameters exist only to satisfy Req
* 2's shared init() signature. */
(void)resreq;
(void)sched;
if (iter == NULL) {
return -EINVAL;
}
priv = (seqjob_plan_iter_priv_t*)calloc(1, sizeof(*priv));
if (priv == NULL) {
return -ENOMEM;
}
priv->bucket_slots[0] = 1;
priv->plan.nbuckets = 1;
priv->plan.bucket_slots = priv->bucket_slots;
/* plan.ctx is set by the caller (my_alloc(), below) on every plan
* returned from next() -- tasks.md's task 2.1 implementation note:
* init()'s own fixed (iter, resreq, sched) signature has no ctx
* parameter to thread it through. */
priv->plan.ctx = NULL;
priv->served = 0;
iter->private = priv;
return 0;
}
static xce_alloc_plan_t* seqjob_plan_iter_next(xce_alloc_plan_iter_t* iter)
{
seqjob_plan_iter_priv_t* priv;
/* Req 2 AC 6: a real cancellation point at least once per invocation
* -- yielding the single fixed plan has no cancellation point of its
* own. */
pthread_testcancel();
if ((iter == NULL) || (iter->private == NULL)) {
return NULL;
}
priv = (seqjob_plan_iter_priv_t*)iter->private;
/* Req 2 AC 4: stable NULL once exhausted. */
if (priv->served) {
return NULL;
}
priv->served = 1;
return &(priv->plan);
}
static void seqjob_plan_iter_fini(xce_alloc_plan_iter_t* iter)
{
if ((iter == NULL) || (iter->private == NULL)) {
return;
}
free(iter->private);
iter->private = NULL;
}
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;
int rc;
xlist_t workers;
xce_pendreason_t* reason = ctx->reason;
xvec_t* uncomit = NULL;
xce_sched_worker_t* w = NULL;
xce_scheduler_t* sched = (xce_scheduler_t*)plugin->sched;
int rc;
xce_pendreason_t* reason = ctx->reason;
xvec_t* uncomit = NULL;
xce_sched_worker_t* w = NULL;
xce_alloc_plan_iter_t plan_iter_storage;
xce_alloc_plan_iter_t* plan_iter = &plan_iter_storage;
xce_alloc_plan_t* plan;
if (reason == NULL) {
reason = make_pendreason(0);
@ -158,16 +256,73 @@ static int my_alloc(xce_sched_plugin_t* plugin, xce_sched_ctx_t* ctx)
return rc;
}
xlist_init(&workers, 128);
/*
* Req 9.1 / Composition Algorithm (requirements.md): plan_iter ->
* bucket loop -> host_iter, degenerate to seqjob's trivial single-
* plan / single-bucket / first-matching-host case. Replaces the old
* sched_collect_workers() + xlist_find(sched_try_alloc_on_hosts, ...)
* inline worker-collection loop -- no allocator-specific worker-
* collection logic remains here, only the shared iterators.
*/
plan_iter->init = seqjob_plan_iter_init;
plan_iter->next = seqjob_plan_iter_next;
plan_iter->fini = seqjob_plan_iter_fini;
rc = sched_collect_workers(sched, &workers,
(zarray_iter_skip_fn_t)skip_bad_worker_, NULL);
if (rc < 0) {
return rc;
if (plan_iter->init(plan_iter, ctx->primary, sched) != 0) {
/* An infrastructure error (e.g. OOM) is not a resource-
* availability determination -- treat it the same as "no
* candidates", matching alloc_iter.h's own error-propagation
* contract for a failing init(). */
return 0;
}
w = xlist_find(&workers, (xlist_find_fn_t)sched_try_alloc_on_hosts, sched,
ctx);
while ((plan = plan_iter->next(plan_iter)) != NULL) {
int b;
w = NULL;
/* Composition Algorithm's bucket loop -- degenerates to exactly
* one iteration since seqjob's plan always has nbuckets == 1. */
for (b = 0; b < plan->nbuckets; b++) {
xce_host_iter_t host_iter_storage;
xce_host_iter_t* host_iter = &host_iter_storage;
/* tasks.md task 2.1 implementation note: thread ctx onto
* every plan next() returns -- init()'s own fixed signature
* has no ctx parameter for this. */
plan->ctx = ctx;
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) {
xce_sched_worker_t* cand;
while ((cand = host_iter->next(host_iter)) != NULL) {
if (sched_try_alloc_on_host_(cand, sched, ctx) > 0) {
w = cand;
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 (w == NULL) {
break; /* this bucket found no satisfying host */
}
}
if (w != NULL) {
break; /* plan satisfied -- seqjob only ever has one plan */
}
}
plan_iter->fini(plan_iter);
if (w != NULL) {
xce_host_alloc_t** hallocs = NULL;

30
src/scheduler/src/resource.c
View file

@ -476,10 +476,27 @@ static int sched_update_res_accounts(xce_sched_core_t* core, zarray_t* array,
return 0;
}
static int sched_try_alloc_on_host_(xce_sched_worker_t* w, va_list ap)
/*
* sched_try_alloc_on_host_ - the "try_alloc(worker, plan)" leaf of the
* Composition Algorithm (requirements.md, scheduler-allocation spec):
* select filter, then per-host prepare (plugin_prepare_allocation() with
* ctx->worker set, which is what actually populates ctx->alloc_req.host
* for THIS candidate -- xce_host_iter_t's own next() never does this; it
* only pre-filters against whatever is already staged), then the real
* resource-satisfaction check, then commit. Shared by every allocator
* plugin's bucket loop (seqjob_allocator.c's single-bucket case,
* parjob_allocator.c's multi-bucket case) -- exported (was `static`, and
* wrapped by the now-deleted variadic sched_try_alloc_on_hosts(), whose
* sole caller was seqjob_allocator.c's own pre-rewrap xlist_find() call
* site) so both plugins can call it directly with real parameters instead
* of each carrying its own private copy of this logic. Returns >0 on
* success (host claimed, ctx->alloc_req.host/ctx->worker hold the
* committed want), 0 if this host does not work (a pend reason is
* recorded), <0 on infrastructure error.
*/
int sched_try_alloc_on_host_(xce_sched_worker_t* w, xce_scheduler_t* sched,
xce_sched_ctx_t* ctx)
{
xce_scheduler_t* sched = va_arg(ap, xce_scheduler_t*);
xce_sched_ctx_t* ctx = va_arg(ap, xce_sched_ctx_t*);
xce_sched_obj_t* schedobj = ctx->schedobj;
xce_primary_resreq_t* primary = ctx->primary;
int rc;
@ -550,13 +567,6 @@ static int sched_try_alloc_on_host_(xce_sched_worker_t* w, va_list ap)
return 1;
}
int sched_try_alloc_on_hosts(xce_worker_group_t* g, va_list ap)
{
/* FIXME: !!! */
xce_sched_worker_t* w = xlist_get(&(g->members), 0);
return sched_try_alloc_on_host_(w, ap);
}
/*
* sched_ctx_uncommit_host - reverses exactly one provisional per-host
* resource debit a prior, successful sched_try_alloc_on_host_() call made

18
src/scheduler/testsuite/meson.build
View file

@ -1527,6 +1527,21 @@ sched_alloc_cache_test = executable(
dependencies: [scheduler_test_deps],
install: false)
# Task 3.1 (scheduler-allocation spec, Req 9.1, 4.6): seqjob_allocator.c's
# rewrap onto the shared plan_iter/host_iter interfaces, plus the new
# maxTasks==1 guard on seqjob_prepare_allocation() (Req 4.6). Same
# white-box #include pattern as sched_host_iter_test above (../src/core.c,
# ../src/resource.c, ../src/schedobj.c, ../src/host_iter.c) PLUS
# ../src/plugins/seqjob_allocator.c itself, the file under test. Smoke-
# level coverage only -- the exhaustive pre/post-rewrap equivalence proof
# is task 3.2's dedicated sched_seqjob_rewrap_equivalence_test.
sched_seqjob_allocator_test = executable(
'sched_seqjob_allocator_test',
['test_seqjob_allocator.c'],
include_directories:[incdir],
dependencies: [scheduler_test_deps],
install: false)
# Register as meson tests so `meson test --suite scheduler` runs them all.
# detect_leaks=0 matches the suite convention (fixtures are not torn down).
test('scheduler test', scheduler_test, suite: 'scheduler',
@ -1718,3 +1733,6 @@ test('scheduler host iter test',
test('scheduler alloc cache test',
sched_alloc_cache_test, suite: 'scheduler',
env: ['ASAN_OPTIONS=detect_leaks=0'])
test('scheduler seqjob allocator test',
sched_seqjob_allocator_test, suite: 'scheduler',
env: ['ASAN_OPTIONS=detect_leaks=0'])

371
src/scheduler/testsuite/test_seqjob_allocator.c Normal file
View file

@ -0,0 +1,371 @@
/*
* test_seqjob_allocator.c - task 3.1 (scheduler-allocation spec): tests
* for seqjob_allocator.c's rewrap onto the shared xce_alloc_plan_iter_t /
* xce_host_iter_t interfaces (Req 9.1) and the new maxTasks==1 guard on
* seqjob_prepare_allocation() (Req 4.6).
*
* This is the task-3.1-level smoke coverage only -- proving the rewrapped
* my_alloc() actually runs end to end through the shared iterators (not
* just that it compiles) and that the guard closes the fan-out gap Req
* 4.6 describes. The exhaustive pre/post-rewrap equivalence proof across
* multiple candidate hosts (including a select-filter failure) is task
* 3.2's dedicated job (test_seqjob_rewrap_equivalence.c) -- not
* duplicated here.
*
* White-box #include convention: same combination test_host_iter.c (task
* 2.1) already established (core.c + resource.c + schedobj.c +
* host_iter.c) for a real worker/resmgr/select/availability fixture, plus
* ../src/plugins/seqjob_allocator.c itself so this file can call its
* `static` my_alloc()/seqjob_prepare_allocation() directly.
*
* seqjob_allocator.c's my_alloc() calls the real, exported
* plugin_prepare_allocation() (plugin.c) to (re-)populate
* ctx->alloc_req.host per candidate host -- plugin.c itself is NOT
* white-box included here (it needs the full plugins kvmgr/registration
* machinery, unrelated to what this file tests), so, like
* test_host_iter.c's own link-satisfying stub, this file provides one --
* but unlike that file's no-op stub, this one delegates straight through
* to the real seqjob_prepare_allocation() under test, standing in for a
* fan-out of exactly the one plugin relevant here.
*/
#include <errno.h>
#include <stdlib.h>
#include <string.h>
#include <unity/unity.h>
#include "resobj.h"
#include "resreq.h"
#include "sched/alloc_iter.h"
#include "scheduler.h"
#include "xcealloc.h"
/* Pull in the real xce_sched_core_init()/sched_add_worker() (core.c),
* sched_meet_host_selection()/sched_check_host_satisfy()/put_resvals()
* (resource.c), the real sched_ctx_init()/sched_ctx_get_uncomit_*()
* (schedobj.c), the shared host_iter.c implementation, and
* seqjob_allocator.c itself (the file under test) into this TU. */
#include "../src/core.c"
#include "../src/resource.c"
#include "../src/schedobj.c"
#include "../src/host_iter.c"
#include "../src/plugins/seqjob_allocator.c"
/* ---------------------------------------------------------------------- */
/* Link-satisfying stubs */
/* ---------------------------------------------------------------------- */
/* plugin_register()/plugin_unregister() are referenced by
* register_seqjob_allocator()/core.c respectively, but this file never
* calls register_seqjob_allocator() (it calls my_alloc()/
* seqjob_prepare_allocation() directly, avoiding the full plugins kvmgr
* this codebase's real plugin_register() needs) -- never actually
* invoked by anything this file's tests exercise. */
int plugin_register(void* sched, void* plugin)
{
(void)sched;
(void)plugin;
return 0;
}
void plugin_unregister(void* arg) { (void)arg; }
/* Unlike test_host_iter.c's no-op stub, this one delegates to the real
* seqjob_prepare_allocation() under test -- standing in for
* plugin_prepare_allocation()'s (plugin.c) real fan-out across every
* registered plugin, here exactly one (seqjob's own), which is all this
* file's fixtures ever register. Mirrors plugin.c:591's own reset-before-
* fan-out step (xvec_idval_fini()+init() on whichever of
* ctx->alloc_req.{shared,host} this call's ctx->worker selects) -- without
* it, ctx->alloc_req.host would accumulate a second "slots" entry across
* the real production call sequence's two prepare_allocation() calls per
* host try (once at ctx->worker==NULL via sched_ctx_set_primary(), again
* per host at ctx->worker!=NULL), since xvec_idval_store() never dedups. */
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 seqjob_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;
}
/* Link-satisfying stub for sched_make_allocs() (scheduler.c) -- avoids
* white-box-including the rest of that 1382-line file (job/queue/picker
* machinery unrelated to seqjob_allocator.c) just to build the leaf
* xce_alloc_t records my_alloc()'s success path constructs. Returns one
* non-NULL, zeroed xce_alloc_t per requested resource index -- sufficient
* for this file's assertions, which check WHICH host got allocated and
* how many resource entries it carries, not the internal xce_alloc_t
* field contents make_alloc() (xcealloc.c, unmodified, real) would fill
* in for a live cluster. */
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;
}
void setUp(void) { return; }
void tearDown(void) { return; }
/* ---------------------------------------------------------------------- */
/* Fixture helpers */
/* ---------------------------------------------------------------------- */
#define WORKER_NAME "seqjob-host"
/* register_slots_resource - the real makeResource() (resobj.c), giving
* "slots" a resmgr-assigned idx -- seqjob_prepare_allocation()'s own
* key_find(sched->core.resmgr, "slots", ...) call needs this to be a
* genuinely registered resource, unlike test_host_iter.c's fixture (which
* stubs plugin_prepare_allocation() entirely and never needs resmgr
* lookups to resolve).
*/
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),
* mirroring test_host_iter.c's own add_worker() 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;
}
/* ---------------------------------------------------------------------- */
/* Tests */
/* ---------------------------------------------------------------------- */
/*
* Req 9.1: a maxTasks==1 job with one available, matching host is still
* allocated correctly through the REWRAPPED my_alloc() -- driven through
* the real shared plan_iter -> host_iter composition (seqjob_plan_iter_*,
* sched_host_iter_*), not a re-implementation of any of it. Proves the
* rewrap actually runs end to end (host selected, resource debited,
* ctx->alloc populated), not just that it compiles.
*/
static void
test_my_alloc_allocates_single_matching_host_through_shared_iterators(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;
xce_sched_plugin_t plugin;
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);
add_worker_with_slots(&(sched.core), WORKER_NAME, slots_res->idx, 4.0f);
memset(&primary, 0, sizeof(primary));
primary.minTasks = 1;
primary.maxTasks = 1;
primary.select = NULL; /* matches every host */
init_primary_obj(&obj, &primary_obj, &primary);
TEST_ASSERT_EQUAL_INT(0, sched_ctx_init(&ctx, &sched, &obj));
/* Real call -- exercises plugin_prepare_allocation()'s ctx->worker ==
* NULL branch exactly as sched_make_primary_allocation() does today,
* populating ctx->alloc_req.shared before my_alloc() ever runs. */
sched_ctx_set_primary(&ctx, &primary);
memset(&plugin, 0, sizeof(plugin));
plugin.sched = &sched;
rc = my_alloc(&plugin, &ctx);
TEST_ASSERT_EQUAL_INT(1, rc);
TEST_ASSERT_NOT_NULL(ctx.alloc);
TEST_ASSERT_EQUAL_INT(1, ctx.alloc->nhosts);
TEST_ASSERT_NOT_NULL(ctx.alloc->allocs);
TEST_ASSERT_NOT_NULL(ctx.alloc->allocs[0]);
TEST_ASSERT_EQUAL_STRING(WORKER_NAME, ctx.alloc->allocs[0]->host);
/* "slots" is the one resource seqjob_prepare_allocation() always
* stages -- proves ctx->alloc_req.host was populated by the real
* per-host prepare cycle the try_alloc leaf runs, not left empty by
* an under-wired iterator hookup. */
TEST_ASSERT_EQUAL_INT(1, ctx.alloc->allocs[0]->nallocs);
sched_ctx_fini(&ctx);
}
/*
* Req 4.6: seqjob_prepare_allocation() must never fire for maxTasks > 1
* jobs. Regression test against the exact fan-out corruption this guard
* prevents: pre-stage a want entry into ctx->alloc_req.host as
* parjob_prepare_allocation() would for its own bucket, then call
* seqjob_prepare_allocation() for the SAME (maxTasks > 1) ctx and confirm
* it returns immediately, WITHOUT xvec_idval_store()'s append-only
* behavior adding a second, wrong entry alongside parjob's.
*/
static void
test_seqjob_prepare_allocation_guards_against_maxtasks_gt_1_fanout(void)
{
xce_scheduler_t sched;
xce_sched_obj_t obj;
xce_primary_schedobj_t primary_obj;
xce_primary_resreq_t primary;
xce_sched_ctx_t ctx;
xce_sched_plugin_t plugin;
int rc;
memset(&sched, 0, sizeof(sched));
TEST_ASSERT_EQUAL_INT(0, xce_sched_core_init(&(sched.core)));
memset(&primary, 0, sizeof(primary));
primary.minTasks = 1;
primary.maxTasks = 4; /* parjob's job, not seqjob's */
init_primary_obj(&obj, &primary_obj, &primary);
TEST_ASSERT_EQUAL_INT(0, sched_ctx_init(&ctx, &sched, &obj));
ctx.primary = &primary;
/* Simulate parjob_prepare_allocation() having already staged this
* bucket's real want (Req 4.6's own scenario) before seqjob's hook is
* fanned out to for the SAME maxTasks > 1 job. */
rc = xvec_idval_store(&(ctx.alloc_req.host), 7 /* arbitrary residx */,
3.0f);
TEST_ASSERT_TRUE(rc >= 0);
memset(&plugin, 0, sizeof(plugin));
plugin.sched = &sched;
ctx.worker = NULL;
rc = seqjob_prepare_allocation(&plugin, &ctx);
TEST_ASSERT_EQUAL_INT(0, rc);
/* THE regression assertion: seqjob's hook must not have appended a
* second, wrong entry via xvec_idval_store()'s append-only, no-dedup
* behavior -- parjob's staged want is exactly as it left it. */
TEST_ASSERT_EQUAL_INT(1, ctx.alloc_req.host.count);
TEST_ASSERT_EQUAL_INT(7, ((int*)ctx.alloc_req.host.idxs.data)[0]);
TEST_ASSERT_EQUAL_FLOAT(3.0f, ((float*)ctx.alloc_req.host.vals.data)[0]);
/* Same guard, the other prepare branch (ctx->worker != NULL, the
* per-host pass a real bucket try would also trigger): still a no-op,
* still untouched. */
ctx.worker = add_worker_with_slots(&(sched.core), "some-host", 9, 4.0f);
rc = seqjob_prepare_allocation(&plugin, &ctx);
TEST_ASSERT_EQUAL_INT(0, rc);
TEST_ASSERT_EQUAL_INT(1, ctx.alloc_req.host.count);
TEST_ASSERT_EQUAL_INT(7, ((int*)ctx.alloc_req.host.idxs.data)[0]);
sched_ctx_fini(&ctx);
}
int main(void)
{
UNITY_BEGIN();
RUN_TEST(
test_my_alloc_allocates_single_matching_host_through_shared_iterators);
RUN_TEST(test_seqjob_prepare_allocation_guards_against_maxtasks_gt_1_fanout);
return UNITY_END();
}