From d9831b0bc8dc805c4a18eb0d5d80ec6031fe04df Mon Sep 17 00:00:00 2001 From: Rongsong Shen Date: Fri, 4 Sep 2026 16:46:03 +0000 Subject: [PATCH] feat(scheduler-allocation): implement the shared host-candidate iterator Adds host_iter.c: the one shared xce_host_iter_t implementation, lazily filtering sched->core.workers by select expression (sched_meet_host_selection, unchanged) and resource availability (sched_check_host_satisfy, unchanged), with a real cancellation point per candidate visited. xce_alloc_plan_t gains a ctx field (xce_sched_ctx_t*) so the iterator can reach the scheduler/select expression/staged resource-want -- additive to task 1.1's approved interface, NULL-guarded at init(). Test proves the core design property this spec exists to establish: a host too small for one search remains a valid candidate for a later, smaller search via a fresh iterator instance over the same worker set. Task 2.1 (scheduler-allocation spec). Requirements 3.1-3.6. Co-Authored-By: Claude Sonnet 5 Claude-Session: https://claude.ai/code/session_01XYdtrEWWZfKRRGED7raEYg --- .kiro/specs/scheduler-allocation/tasks.md | 3 +- src/scheduler/include/sched/alloc_iter.h | 45 +- src/scheduler/src/host_iter.c | 181 +++++++ src/scheduler/testsuite/meson.build | 21 + src/scheduler/testsuite/test_host_iter.c | 624 ++++++++++++++++++++++ 5 files changed, 872 insertions(+), 2 deletions(-) create mode 100644 src/scheduler/src/host_iter.c create mode 100644 src/scheduler/testsuite/test_host_iter.c diff --git a/.kiro/specs/scheduler-allocation/tasks.md b/.kiro/specs/scheduler-allocation/tasks.md index 52ea48c9..a2539e3c 100644 --- a/.kiro/specs/scheduler-allocation/tasks.md +++ b/.kiro/specs/scheduler-allocation/tasks.md @@ -23,7 +23,7 @@ - _Requirements: 4.2_ - [ ] 2. Core: shared allocation infrastructure -- [ ] 2.1 (P) Implement and test the lazy host-candidate iterator +- [x] 2.1 (P) Implement and test the lazy host-candidate iterator - Implement the shared host-candidate iterator against the existing select-expression evaluator and resource-availability check, with no allocator-specific logic - Ensure each call advances a real cancellation point so a long-running scan can be safely interrupted - Write a unit test proving a host too small for one candidate search remains a valid candidate for a later, smaller search using a freshly-initialized iterator over the same underlying worker set @@ -106,6 +106,7 @@ ## Implementation Notes +- (2.1) `xce_alloc_plan_t` gained a `ctx` field (`xce_sched_ctx_t*`) so `host_iter_init()` can reach `ctx->sched`/`ctx->primary->select`/`ctx->alloc_req.host` — `host_iter_init()` rejects a plan with `ctx == NULL` with `-EINVAL` rather than crashing. **Tasks 3.1 and 4.1 must set `plan->ctx = ctx` on every plan their plan-iterator returns from `next()`** (not just once) before constructing a `host_iter` from it — a plan-iterator that reuses one internal buffer across calls/jobs must refresh this field each time, since a stale-but-non-NULL pointer from a torn-down `ctx` is not caught by the NULL guard. - (1.4) When a new scheduler test needs a real function from `schedobj.c`/`core.c` (e.g. `sched_ctx_get_uncomit_host()`), white-box `#include` the real source file rather than privately reimplementing its logic in the test — established precedent: `test_picker_config_fields.c`'s `#include "../src/core.c"` + `#include "../src/schedobj.c"` pairing. A private reimplementation may pass today but silently drifts from the real function with nothing to catch it later. - (1.1) `xce_host_iter_t.next()` returns `xce_sched_worker_t*` (`sched/types.h`, carries `resvals`/`resInUse`/`host_eval_ctx`) — never `xce_worker_t*` (`xce/types.h`, the wire/config type `make_sched_worker()` converts FROM). requirements.md/design.md briefly had this wrong; fixed in commit `d1a3e158`. Task 2.1 (`host_iter.c`'s real implementation) and any later task touching worker candidates must use `xce_sched_worker_t*` from the start. - (1.2) `resreq_fp`'s ingest-time population lives in `core_build_primary_schedobj()` (`src/scheduler/src/core_events.c`), right after `primary->resreq = resreq;` — confirmed wired (min_tasks/max_tasks/select_node_id/rusage_hash all set from the parsed `xce_primary_resreq_t`) as of task 1.2 round 2, regression-covered by `test_core_build_primary_schedobj_populates_resreq_fp_from_parsed_resreq` in `src/scheduler/testsuite/test_core_events.c`. Round 1 added the type/field/hash machinery but never wired this call site, leaving `resreq_fp` permanently zeroed — do not rediscover this gap in a later task. diff --git a/src/scheduler/include/sched/alloc_iter.h b/src/scheduler/include/sched/alloc_iter.h index 70103e35..25d04ba0 100644 --- a/src/scheduler/include/sched/alloc_iter.h +++ b/src/scheduler/include/sched/alloc_iter.h @@ -49,7 +49,24 @@ typedef struct { * availability -- consumed by xce_host_iter_t (Req 3), not exposed as * a separate field here. This interface does not mandate a * representation for it; each concrete plan-iterator implementation - * carries whatever it needs to reconstruct the filter. */ + * carries whatever it needs to reconstruct the filter. + * + * Added by task 2.1 (host_iter.c, the real implementation): `ctx` is + * exactly that carrier. xce_host_iter_t.init()'s own three declared + * parameters (iter, plan, workers) cannot, on their own, reach the + * select expression or the currently-staged resource want -- `workers` + * is a bare RTTI var with no back-pointer to the owning + * xce_scheduler_t (confirmed against sched_collect_workers()'s own + * iter_make(&it, sched->core.workers) body, scheduler.c), and this + * struct's other two fields (nbuckets/bucket_slots) carry only the + * partition shape. `ctx` (already carrying ->sched, ->primary->select, + * and ->alloc_req.host -- the resource want staged by whichever + * allocator's prepare_allocation_hook ran) supplies everything + * sched_meet_host_selection()/sched_check_host_satisfy() need, without + * widening init()'s own parameter list. Not owned by the plan; the + * caller (allocate_primary(), a later task) retains ownership for the + * duration of one job's allocation attempt. */ + xce_sched_ctx_t* ctx; } xce_alloc_plan_t; typedef struct xce_alloc_plan_iter_s xce_alloc_plan_iter_t; @@ -137,4 +154,30 @@ struct xce_host_iter_s { void* private; }; +/* + * sched_host_iter_init/next/fini - the one shared xce_host_iter_t + * implementation (host_iter.c, task 2.1, Req 3). Callers wire these onto a + * stack-allocated, caller-owned xce_host_iter_t before calling init(), + * following the exact same manual-vtable-assignment convention task 1.1's + * own reference stubs established (see testsuite/test_alloc_plan_iter.c): + * + * xce_host_iter_t host_iter_storage; + * xce_host_iter_t* host_iter = &host_iter_storage; + * + * 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) { ... } + * + * plan->ctx (see xce_alloc_plan_t above) must be non-NULL: it supplies the + * xce_scheduler_t/select-expression/resource-want state this + * implementation needs but init()'s own (iter, plan, workers) parameters + * cannot reach on their own. + */ +extern int sched_host_iter_init(xce_host_iter_t* iter, xce_alloc_plan_t* plan, + var workers); +extern xce_sched_worker_t* sched_host_iter_next(xce_host_iter_t* iter); +extern void sched_host_iter_fini(xce_host_iter_t* iter); + #endif diff --git a/src/scheduler/src/host_iter.c b/src/scheduler/src/host_iter.c new file mode 100644 index 00000000..be4e3bd2 --- /dev/null +++ b/src/scheduler/src/host_iter.c @@ -0,0 +1,181 @@ +/* + * host_iter.c - the one shared xce_host_iter_t implementation + * (scheduler-allocation spec, task 2.1, Req 3 AC 1-6). + * + * Filters candidate hosts (xce_sched_worker_t*) drawn lazily from a + * scheduler's live worker collection (the same RTTI `var` object + * sched_collect_workers() iterates via iter_make()/iter_next() -- + * scheduler.c's own sched->core.workers, a kvmgr keyed by worker name) + * against exactly two things, reusing existing, unmodified machinery: + * + * 1. the select-expression evaluator (sched_meet_host_selection(), + * resource.c) + * 2. the resource-availability check (sched_check_host_satisfy(), + * resource.c), against whatever resource want is currently staged + * in the allocation context (ctx->alloc_req.host) -- populated by + * whichever allocator's prepare_allocation_hook ran, not by this + * file. + * + * No allocator-specific logic lives here: no bucket/plan bookkeeping, no + * "already claimed" exclusion (that is the caller's is_claimed(), per + * requirements.md's Composition Algorithm), no commit/debit. This file + * only ever answers "does this one candidate host currently satisfy the + * select expression and have enough available slots" -- repeatedly, once + * per next() call, for exactly one plan bucket's search. A fresh instance + * (new init() call) is required per bucket -- see alloc_iter.h's own doc + * comment on xce_host_iter_s for why a shared, forward-only cursor across + * buckets would be incorrect. + * + * Interface note (see alloc_iter.h's xce_alloc_plan_t.ctx field, added by + * this task): init()'s own three declared parameters (iter, plan, + * workers) cannot, on their own, reach the select expression or the + * currently-staged resource want -- `workers` is a bare RTTI var with no + * back-pointer to the owning xce_scheduler_t (confirmed against + * sched_collect_workers()'s own iter_make(&it, sched->core.workers) body, + * scheduler.c), and xce_alloc_plan_t's other two fields (nbuckets/ + * bucket_slots) carry only the partition shape. plan->ctx (already + * carrying ->sched, ->primary->select, and ->alloc_req.host) supplies + * everything sched_meet_host_selection()/sched_check_host_satisfy() need, + * without widening init()'s own parameter list -- see this task's status + * report CONCERNS for the full rationale. + */ +#include +#include +#include + +#include "sched/alloc_iter.h" +#include "sched/types.h" +#include "scheduler.h" + +typedef struct { + xce_alloc_plan_t* plan; /* not owned; must outlive this instance */ + iterator_t it; /* RTTI iterator over the live worker var */ + int exhausted; +} host_iter_priv_t; + +int sched_host_iter_init(xce_host_iter_t* iter, xce_alloc_plan_t* plan, + var workers) +{ + host_iter_priv_t* priv; + int rc; + + if ((iter == NULL) || (plan == NULL) || (plan->ctx == NULL)) { + return -EINVAL; + } + + priv = (host_iter_priv_t*)calloc(1, sizeof(*priv)); + if (priv == NULL) { + return -ENOMEM; + } + + rc = iter_make(&(priv->it), workers); + if (rc < 0) { + free(priv); + return rc; + } + + priv->plan = plan; + priv->exhausted = 0; + + iter->private = priv; + + return 0; +} + +xce_sched_worker_t* sched_host_iter_next(xce_host_iter_t* iter) +{ + host_iter_priv_t* priv; + xce_sched_ctx_t* ctx; + xce_scheduler_t* sched; + + /* Req 3 AC 6: a real cancellation point at least once per invocation + * -- scanning/filtering candidate hosts is fast, non-blocking pure + * computation with no cancellation point of its own. */ + pthread_testcancel(); + + if ((iter == NULL) || (iter->private == NULL)) { + return NULL; + } + + priv = (host_iter_priv_t*)iter->private; + + /* Req 3 AC 3: stable NULL for the rest of this instance's lifetime + * once exhausted. */ + if (priv->exhausted) { + return NULL; + } + + ctx = priv->plan->ctx; + sched = (xce_scheduler_t*)ctx->sched; + + for (;;) { + xce_sched_worker_t* w = NULL; + int rc; + + /* Called again on every candidate visited (not just once at + * function entry): a single next() call can internally skip an + * arbitrary number of non-matching hosts before returning a + * match or exhausting -- exactly the "long-running scan" this + * requirement guards against being uninterruptible. */ + pthread_testcancel(); + + rc = iter_next(&(priv->it), (void**)&w); + if ((rc < 0) || (rc == RTTI_ITER_NO_DATA)) { + priv->exhausted = 1; + return NULL; + } + + if (w == NULL) { + continue; + } + + /* 1. Select-expression filter -- existing evaluator, unchanged. */ + if (!sched_meet_host_selection( + sched, w, + (ctx->primary != NULL) ? ctx->primary->select : NULL)) { + continue; + } + + /* 2. Resource-availability filter, against the resource want + * already staged in ctx->alloc_req.host by whichever allocator's + * prepare_allocation_hook ran -- this file never populates that + * vector itself (no allocator-specific logic). */ + { + xvec_t* uncomit = NULL; + + rc = sched_ctx_get_uncomit_host(&uncomit, ctx, w->idx); + if (rc < 0) { + /* Infrastructure error for this one candidate (e.g. OOM + * growing its per-host uncommitted-debit vector) -- not a + * resource-availability determination. Skip this host and + * keep scanning rather than aborting the whole search. */ + continue; + } + + rc = sched_check_host_satisfy(sched, w, uncomit, + &(ctx->alloc_req.host), + &(ctx->unmeet)); + if (rc <= 0) { + continue; + } + } + + return w; + } +} + +void sched_host_iter_fini(xce_host_iter_t* iter) +{ + host_iter_priv_t* priv; + + if ((iter == NULL) || (iter->private == NULL)) { + return; + } + + priv = (host_iter_priv_t*)iter->private; + + iter_destroy(&(priv->it)); + free(priv); + + iter->private = NULL; +} diff --git a/src/scheduler/testsuite/meson.build b/src/scheduler/testsuite/meson.build index f086f0d2..f883c407 100644 --- a/src/scheduler/testsuite/meson.build +++ b/src/scheduler/testsuite/meson.build @@ -1492,6 +1492,24 @@ sched_alloc_plan_iter_test = executable( dependencies: [scheduler_test_deps], install: false) +# Contract + behavioral test for the one shared xce_host_iter_t +# implementation (scheduler-allocation spec, task 2.1, Req 3 AC 1-6). Unlike +# sched_alloc_plan_iter_test's own trivial no-op stubs, this drives the REAL +# ../src/host_iter.c against real fixture state -- select-expression +# filtering, availability filtering, exhaustion, the fresh-per-bucket +# adversarial case (a host too small for one search remains a valid +# candidate for a later, smaller search over a freshly-initialized +# instance), and a behavioral proof of the required pthread_testcancel() +# cancellation point. Same white-box #include pattern as +# test_resource_accounting.c (../src/core.c, ../src/resource.c, +# ../src/schedobj.c) PLUS the new ../src/host_iter.c under test. +sched_host_iter_test = executable( + 'sched_host_iter_test', + ['test_host_iter.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', @@ -1677,3 +1695,6 @@ test('scheduler builtin fcfs cursor uaf test', test('scheduler alloc plan iter test', sched_alloc_plan_iter_test, suite: 'scheduler', env: ['ASAN_OPTIONS=detect_leaks=0']) +test('scheduler host iter test', + sched_host_iter_test, suite: 'scheduler', + env: ['ASAN_OPTIONS=detect_leaks=0']) diff --git a/src/scheduler/testsuite/test_host_iter.c b/src/scheduler/testsuite/test_host_iter.c new file mode 100644 index 00000000..f59ed3b8 --- /dev/null +++ b/src/scheduler/testsuite/test_host_iter.c @@ -0,0 +1,624 @@ +/* + * test_host_iter.c - contract + behavioral test for the one shared + * xce_host_iter_t implementation (scheduler-allocation spec, task 2.1, + * Req 3 AC 1-6). + * + * Unlike test_alloc_plan_iter.c (task 1.1), which proved the CONTRACT + * SHAPE compiles and is usable via trivial no-op stub implementations, + * this file drives the REAL host_iter.c against real fixture state: + * real xce_sched_core_t workers (../src/core.c), the real select- + * expression evaluator + sched_check_host_satisfy() (../src/resource.c), + * and the real sched_ctx_get_uncomit_host() accessor (../src/schedobj.c) + * -- the exact same white-box #include combination + * test_resource_accounting.c already established for this fixture shape + * (core.c + resource.c + schedobj.c), plus the new ../src/host_iter.c + * under test. + * + * Covers: + * - select-expression filtering (a host failing the select expression + * is never returned) -- Req 3 AC 2. + * - availability filtering (a host without enough resources is never + * returned) -- Req 3 AC 2. + * - exhaustion, stable across repeated calls -- Req 3 AC 3. + * - the fresh-per-bucket adversarial case (Req 3 AC 4, Req 4 AC 2): a + * host too small for one candidate search remains a valid candidate + * for a later, smaller search using a freshly-initialized host_iter + * instance over the SAME underlying worker set -- proving this + * implementation carries no persistent "consumed"/"rejected" state + * across instances. This is the single most important test in this + * file (design.md's own words for this exact case). + * - Req 3 AC 6: next() calls pthread_testcancel() (or an equivalent + * real cancellation point) -- proved behaviorally by cancelling a + * thread parked in a tight next()-calling loop. + * + * plan->ctx (the small, documented addition to xce_alloc_plan_t this + * task made -- see alloc_iter.h) is how each test wires a scheduler + + * select expression + resource want into the plan host_iter consumes; + * see host_iter.c's own file header comment for why init()'s existing + * three parameters cannot reach that state on their own. + */ +#include +#include +#include +#include +#include + +#include + +#include "resreq.h" +#include "sched/alloc_iter.h" +#include "scheduler.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 evaluator + availability check under test), the real + * sched_ctx_init()/sched_ctx_get_uncomit_host() (schedobj.c), and the + * new host_iter.c implementation under test into this TU -- same + * white-box convention test_resource_accounting.c already established + * for this exact core.c+resource.c+schedobj.c combination. */ +#include "../src/core.c" +#include "../src/resource.c" +#include "../src/schedobj.c" +#include "../src/host_iter.c" + +/* Link-satisfying stubs -- identical set to test_resource_accounting.c's + * own (see that file's header comment for the full rationale). None of + * these are ever actually invoked by this file's tests: this file never + * calls sched_ctx_set_primary() (it sets ctx.primary directly, a plain + * field assignment, to avoid the plugin_prepare_allocation() side effect + * entirely) nor sched_try_alloc_on_hosts()/sched_update_existing_*(). */ +void plugin_unregister(void* arg) { (void)arg; } + +int plugin_prepare_allocation(void* s, xce_sched_ctx_t* c) +{ + (void)s; + (void)c; + return 0; +} + +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; +} + +void setUp(void) { return; } +void tearDown(void) { return; } + +/* ---------------------------------------------------------------------- */ +/* Fixture helpers */ +/* ---------------------------------------------------------------------- */ + +#define SLOTS_RESIDX 0 +#define RACK_RESIDX 1 + +/* + * add_worker - a real sched_add_worker() (core.c) worker with a "slots" + * resource (total = slots_total, in-use = slots_inuse) and a "rack" + * resource (numeric), both set via the real put_resvals() (resource.c). + */ +static void add_worker(xce_sched_core_t* core, const char* name, + float slots_total, float slots_inuse, float rack) +{ + 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_RESIDX; + val[0] = slots_total; + put_resvals(&(w->resvals), 1, idx, val); + val[0] = slots_inuse; + put_resvals(&(w->resInUse), 1, idx, val); + + idx[0] = RACK_RESIDX; + val[0] = rack; + put_resvals(&(w->resvals), 1, idx, val); + + rc = sched_add_worker(core, w); + TEST_ASSERT_TRUE(rc >= 0); +} + +/* make_rack_eq_select - builds `rack == literal`, a real, evaluator- + * exercised select expression (SEL_EQ over a numeric SEL_RESNAME leaf and + * a SEL_NUM literal) -- no parser needed, eval_select_simple() (used via + * sched_meet_host_selection()'s no-sel_eval_ctx fallback path, since this + * fixture's xce_scheduler_t never sets sched.core.sel_eval_ctx) walks + * this hand-built tree directly. */ +static void make_rack_eq_select(xce_select_node_t* resname, + xce_select_node_t* num, xce_select_node_t* eq, + xce_select_t* select, double literal) +{ + memset(resname, 0, sizeof(*resname)); + resname->type = SEL_RESNAME; + resname->u.resname.res_idx = RACK_RESIDX; + resname->u.resname.res_type = RES_TYPE_NUMERIC; + + memset(num, 0, sizeof(*num)); + num->type = SEL_NUM; + num->u.numval = literal; + + memset(eq, 0, sizeof(*eq)); + eq->type = SEL_EQ; + eq->u.binop.left = resname; + eq->u.binop.right = num; + + select->root = eq; +} + +/* init_ctx - a real sched_ctx_init() (schedobj.c), with ctx->primary set + * by plain field assignment (not sched_ctx_set_primary(), whose + * plugin_prepare_allocation() side effect this file's tests don't need + * or want -- see the stub's own comment above). */ +static void init_ctx(xce_sched_ctx_t* ctx, xce_scheduler_t* sched, + xce_sched_obj_t* obj, xce_primary_resreq_t* primary) +{ + int rc = sched_ctx_init(ctx, sched, obj); + TEST_ASSERT_EQUAL_INT(0, rc); + ctx->primary = primary; +} + +static void init_primary_obj(xce_sched_obj_t* obj) +{ + memset(obj, 0, sizeof(*obj)); + obj->tag = XCE_RESREQ_PRIMARY; +} + +/* set_wanted_slots - stages a "slots" resource want into ctx->alloc_req.host + * -- the vector whichever allocator's prepare_allocation_hook would + * normally populate; this file populates it directly since no allocator + * plugin is wired in this fixture (host_iter.c never populates it + * itself, by design). */ +static void set_wanted_slots(xce_sched_ctx_t* ctx, float slots_wanted) +{ + int rc = xvec_idval_store(&(ctx->alloc_req.host), SLOTS_RESIDX, + slots_wanted); + TEST_ASSERT_TRUE(rc >= 0); +} + +static void bind_host_iter(xce_host_iter_t* iter) +{ + memset(iter, 0, sizeof(*iter)); + iter->init = sched_host_iter_init; + iter->next = sched_host_iter_next; + iter->fini = sched_host_iter_fini; +} + +#define MAX_COLLECT 16 + +static int collect_all(xce_host_iter_t* iter, char names[][64]) +{ + int n = 0; + xce_sched_worker_t* w; + + while ((n < MAX_COLLECT) && ((w = iter->next(iter)) != NULL)) { + strncpy(names[n], w->name, 63); + names[n][63] = '\0'; + n++; + } + + return n; +} + +static int contains(char names[][64], int n, const char* target) +{ + int i; + + for (i = 0; i < n; i++) { + if (strcmp(names[i], target) == 0) { + return 1; + } + } + + return 0; +} + +/* ---------------------------------------------------------------------- */ +/* Tests */ +/* ---------------------------------------------------------------------- */ + +/* + * Req 3 AC 2 (select half): a host failing the select expression is never + * returned, regardless of having plenty of available slots. + */ +static void test_select_expression_filtering_excludes_non_matching_host(void) +{ + xce_scheduler_t sched; + xce_sched_ctx_t ctx; + xce_sched_obj_t obj; + xce_primary_resreq_t primary; + xce_select_node_t resname, num, eq; + xce_select_t select; + xce_host_iter_t host_iter_storage; + xce_host_iter_t* host_iter = &host_iter_storage; + int bucket_slots[1] = { 1 }; + xce_alloc_plan_t plan; + char names[MAX_COLLECT][64]; + int n; + + memset(&sched, 0, sizeof(sched)); + TEST_ASSERT_EQUAL_INT(0, xce_sched_core_init(&(sched.core))); + + add_worker(&(sched.core), "rack-match", 4.0f, 0.0f, 1.0); + add_worker(&(sched.core), "rack-nomatch", 4.0f, 0.0f, 2.0); + + make_rack_eq_select(&resname, &num, &eq, &select, 1.0); + + memset(&primary, 0, sizeof(primary)); + primary.minTasks = 1; + primary.maxTasks = 1; + primary.select = &select; + + init_primary_obj(&obj); + init_ctx(&ctx, &sched, &obj, &primary); + /* No slots requirement staged -- isolates select filtering from + * availability filtering for this test. */ + + memset(&plan, 0, sizeof(plan)); + plan.nbuckets = 1; + plan.bucket_slots = bucket_slots; + plan.ctx = &ctx; + + bind_host_iter(host_iter); + TEST_ASSERT_EQUAL_INT( + 0, host_iter->init(host_iter, &plan, sched.core.workers)); + + n = collect_all(host_iter, names); + + TEST_ASSERT_TRUE(contains(names, n, "rack-match")); + TEST_ASSERT_FALSE(contains(names, n, "rack-nomatch")); + + host_iter->fini(host_iter); + TEST_ASSERT_NULL(host_iter->private); +} + +/* + * Req 3 AC 2 (availability half): a host without enough available slots + * is never returned, even when it passes the select expression cleanly + * (no select set at all, here). + */ +static void test_availability_filtering_excludes_too_small_host(void) +{ + xce_scheduler_t sched; + xce_sched_ctx_t ctx; + xce_sched_obj_t obj; + xce_primary_resreq_t primary; + xce_host_iter_t host_iter_storage; + xce_host_iter_t* host_iter = &host_iter_storage; + int bucket_slots[1] = { 3 }; + xce_alloc_plan_t plan; + char names[MAX_COLLECT][64]; + int n; + + memset(&sched, 0, sizeof(sched)); + TEST_ASSERT_EQUAL_INT(0, xce_sched_core_init(&(sched.core))); + + add_worker(&(sched.core), "big", 4.0f, 0.0f, 0.0); + add_worker(&(sched.core), "small", 1.0f, 0.0f, 0.0); + + memset(&primary, 0, sizeof(primary)); + primary.minTasks = 1; + primary.maxTasks = 1; + primary.select = NULL; /* no select filter -- isolates availability */ + + init_primary_obj(&obj); + init_ctx(&ctx, &sched, &obj, &primary); + set_wanted_slots(&ctx, 3.0f); /* "small" (1 free) cannot satisfy this */ + + memset(&plan, 0, sizeof(plan)); + plan.nbuckets = 1; + plan.bucket_slots = bucket_slots; + plan.ctx = &ctx; + + bind_host_iter(host_iter); + TEST_ASSERT_EQUAL_INT( + 0, host_iter->init(host_iter, &plan, sched.core.workers)); + + n = collect_all(host_iter, names); + + TEST_ASSERT_TRUE(contains(names, n, "big")); + TEST_ASSERT_FALSE(contains(names, n, "small")); + + host_iter->fini(host_iter); +} + +/* + * Req 3 AC 3: next() returns NULL once exhausted, and this signal is + * stable across repeated calls on the same instance. + */ +static void test_exhaustion_is_stable_across_repeated_calls(void) +{ + xce_scheduler_t sched; + xce_sched_ctx_t ctx; + xce_sched_obj_t obj; + xce_primary_resreq_t primary; + xce_host_iter_t host_iter_storage; + xce_host_iter_t* host_iter = &host_iter_storage; + int bucket_slots[1] = { 1 }; + xce_alloc_plan_t plan; + xce_sched_worker_t* w; + + memset(&sched, 0, sizeof(sched)); + TEST_ASSERT_EQUAL_INT(0, xce_sched_core_init(&(sched.core))); + + add_worker(&(sched.core), "only", 4.0f, 0.0f, 0.0); + + memset(&primary, 0, sizeof(primary)); + primary.minTasks = 1; + primary.maxTasks = 1; + primary.select = NULL; + + init_primary_obj(&obj); + init_ctx(&ctx, &sched, &obj, &primary); + set_wanted_slots(&ctx, 1.0f); + + memset(&plan, 0, sizeof(plan)); + plan.nbuckets = 1; + plan.bucket_slots = bucket_slots; + plan.ctx = &ctx; + + bind_host_iter(host_iter); + TEST_ASSERT_EQUAL_INT( + 0, host_iter->init(host_iter, &plan, sched.core.workers)); + + w = host_iter->next(host_iter); + TEST_ASSERT_NOT_NULL(w); + TEST_ASSERT_EQUAL_STRING("only", w->name); + + TEST_ASSERT_NULL(host_iter->next(host_iter)); + TEST_ASSERT_NULL(host_iter->next(host_iter)); + TEST_ASSERT_NULL(host_iter->next(host_iter)); + + host_iter->fini(host_iter); +} + +/* + * THE critical adversarial case (Req 3 AC 4, Req 4 AC 2; design.md calls + * this "the single most important test in this spec"): a host too small + * for one search remains a valid candidate for a LATER, smaller search + * using a freshly-initialized xce_host_iter_t instance over the SAME + * underlying worker set -- proving this implementation carries no + * persistent "consumed"/"rejected" state across instances (which is + * exactly what would happen if a plan's buckets wrongly shared one + * cursor instead of each getting a fresh host_iter, per requirements.md's + * Composition Algorithm). + */ +static void +test_host_too_small_for_larger_search_remains_candidate_for_smaller_search( + void) +{ + xce_scheduler_t sched; + char names_large[MAX_COLLECT][64]; + char names_small[MAX_COLLECT][64]; + int n_large, n_small; + + memset(&sched, 0, sizeof(sched)); + TEST_ASSERT_EQUAL_INT(0, xce_sched_core_init(&(sched.core))); + + /* "big" has 4 free slots, "small" has only 1 -- the same underlying + * worker set is reused, unmodified, across both searches below. */ + add_worker(&(sched.core), "big", 4.0f, 0.0f, 0.0); + add_worker(&(sched.core), "small", 1.0f, 0.0f, 0.0); + + /* --- Search 1: a LARGE want (3 slots) -- "small" must be rejected --- */ + { + xce_sched_ctx_t ctx1; + xce_sched_obj_t obj1; + xce_primary_resreq_t primary1; + xce_host_iter_t host_iter_storage; + xce_host_iter_t* host_iter = &host_iter_storage; + int bucket_slots[1] = { 3 }; + xce_alloc_plan_t plan; + + memset(&primary1, 0, sizeof(primary1)); + primary1.minTasks = 1; + primary1.maxTasks = 1; + primary1.select = NULL; + + init_primary_obj(&obj1); + init_ctx(&ctx1, &sched, &obj1, &primary1); + set_wanted_slots(&ctx1, 3.0f); + + memset(&plan, 0, sizeof(plan)); + plan.nbuckets = 1; + plan.bucket_slots = bucket_slots; + plan.ctx = &ctx1; + + bind_host_iter(host_iter); + TEST_ASSERT_EQUAL_INT( + 0, host_iter->init(host_iter, &plan, sched.core.workers)); + + n_large = collect_all(host_iter, names_large); + + host_iter->fini(host_iter); + sched_ctx_fini(&ctx1); + } + + TEST_ASSERT_TRUE(contains(names_large, n_large, "big")); + TEST_ASSERT_FALSE(contains(names_large, n_large, "small")); + + /* --- Search 2: a fresh xce_host_iter_t instance, a SMALLER want (1 + * slot), over the SAME sched.core.workers -- "small" must now be + * found. If host_iter carried any persistent "rejected by an earlier + * search" state, this would wrongly fail. --- */ + { + xce_sched_ctx_t ctx2; + xce_sched_obj_t obj2; + xce_primary_resreq_t primary2; + xce_host_iter_t host_iter_storage; + xce_host_iter_t* host_iter = &host_iter_storage; + int bucket_slots[1] = { 1 }; + xce_alloc_plan_t plan; + + memset(&primary2, 0, sizeof(primary2)); + primary2.minTasks = 1; + primary2.maxTasks = 1; + primary2.select = NULL; + + init_primary_obj(&obj2); + init_ctx(&ctx2, &sched, &obj2, &primary2); + set_wanted_slots(&ctx2, 1.0f); + + memset(&plan, 0, sizeof(plan)); + plan.nbuckets = 1; + plan.bucket_slots = bucket_slots; + plan.ctx = &ctx2; + + bind_host_iter(host_iter); + TEST_ASSERT_EQUAL_INT( + 0, host_iter->init(host_iter, &plan, sched.core.workers)); + + n_small = collect_all(host_iter, names_small); + + host_iter->fini(host_iter); + sched_ctx_fini(&ctx2); + } + + TEST_ASSERT_TRUE(contains(names_small, n_small, "small")); + TEST_ASSERT_TRUE(contains(names_small, n_small, "big")); +} + +/* + * Req 3 AC 6: next() calls pthread_testcancel() (or an equivalent real + * cancellation point) at least once per invocation. Proved behaviorally: + * a worker thread parked in a tight loop calling next() repeatedly is + * reliably torn down by pthread_cancel() from the main thread. If + * host_iter_next() had no real cancellation point, this tight loop would + * never yield to the pending cancellation and pthread_join() below would + * hang (timing out the whole test binary) rather than return. + */ +typedef struct { + xce_host_iter_t* host_iter; +} cancel_thread_arg_t; + +static void* cancel_spin_thread(void* argp) +{ + cancel_thread_arg_t* arg = (cancel_thread_arg_t*)argp; + + for (;;) { + (void)arg->host_iter->next(arg->host_iter); + } + + return NULL; /* unreachable */ +} + +static void test_next_has_a_real_cancellation_point(void) +{ + xce_scheduler_t sched; + xce_sched_ctx_t ctx; + xce_sched_obj_t obj; + xce_primary_resreq_t primary; + xce_host_iter_t host_iter_storage; + xce_host_iter_t* host_iter = &host_iter_storage; + int bucket_slots[1] = { 1 }; + xce_alloc_plan_t plan; + cancel_thread_arg_t arg; + pthread_t tid; + int rc; + + memset(&sched, 0, sizeof(sched)); + TEST_ASSERT_EQUAL_INT(0, xce_sched_core_init(&(sched.core))); + + add_worker(&(sched.core), "only", 4.0f, 0.0f, 0.0); + + memset(&primary, 0, sizeof(primary)); + primary.minTasks = 1; + primary.maxTasks = 1; + primary.select = NULL; + + init_primary_obj(&obj); + init_ctx(&ctx, &sched, &obj, &primary); + set_wanted_slots(&ctx, 1.0f); + + memset(&plan, 0, sizeof(plan)); + plan.nbuckets = 1; + plan.bucket_slots = bucket_slots; + plan.ctx = &ctx; + + bind_host_iter(host_iter); + TEST_ASSERT_EQUAL_INT( + 0, host_iter->init(host_iter, &plan, sched.core.workers)); + + arg.host_iter = host_iter; + + rc = pthread_create(&tid, NULL, cancel_spin_thread, &arg); + TEST_ASSERT_EQUAL_INT(0, rc); + + /* Let the spin thread actually get running before cancelling it. */ + usleep(20000); + + rc = pthread_cancel(tid); + TEST_ASSERT_EQUAL_INT(0, rc); + + rc = pthread_join(tid, NULL); + TEST_ASSERT_EQUAL_INT(0, rc); + + /* Safe post-cancellation cleanup: pthread_testcancel() only ever + * fires between complete logical steps (function entry, or right + * before the next iter_next() call), never mid-mutation, so the + * iterator's private state is always well-formed here regardless of + * exactly where cancellation landed. */ + host_iter->fini(host_iter); + sched_ctx_fini(&ctx); +} + +/* + * init() defensively rejects a plan with no ctx -- this implementation's + * only way of reaching the scheduler/select/resource-want state it needs + * (see alloc_iter.h's xce_alloc_plan_t.ctx doc comment). + */ +static void test_init_rejects_missing_ctx(void) +{ + xce_host_iter_t host_iter_storage; + xce_host_iter_t* host_iter = &host_iter_storage; + int bucket_slots[1] = { 1 }; + xce_alloc_plan_t plan; + + memset(&plan, 0, sizeof(plan)); + plan.nbuckets = 1; + plan.bucket_slots = bucket_slots; + plan.ctx = NULL; + + bind_host_iter(host_iter); + + TEST_ASSERT_EQUAL_INT(-EINVAL, host_iter->init(host_iter, &plan, NULL)); + TEST_ASSERT_EQUAL_INT(-EINVAL, host_iter->init(host_iter, NULL, NULL)); +} + +int main(void) +{ + UNITY_BEGIN(); + RUN_TEST(test_select_expression_filtering_excludes_non_matching_host); + RUN_TEST(test_availability_filtering_excludes_too_small_host); + RUN_TEST(test_exhaustion_is_stable_across_repeated_calls); + RUN_TEST( + test_host_too_small_for_larger_search_remains_candidate_for_smaller_search); + RUN_TEST(test_next_has_a_real_cancellation_point); + RUN_TEST(test_init_rejects_missing_ctx); + return UNITY_END(); +}