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 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01XYdtrEWWZfKRRGED7raEYg
This commit is contained in:
parent
d0024eb7e8
commit
d9831b0bc8
5 changed files with 872 additions and 2 deletions
|
|
@ -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.
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
181
src/scheduler/src/host_iter.c
Normal file
181
src/scheduler/src/host_iter.c
Normal file
|
|
@ -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 <errno.h>
|
||||
#include <pthread.h>
|
||||
#include <stdlib.h>
|
||||
|
||||
#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;
|
||||
}
|
||||
|
|
@ -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'])
|
||||
|
|
|
|||
624
src/scheduler/testsuite/test_host_iter.c
Normal file
624
src/scheduler/testsuite/test_host_iter.c
Normal file
|
|
@ -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 <errno.h>
|
||||
#include <pthread.h>
|
||||
#include <stdlib.h>
|
||||
#include <string.h>
|
||||
#include <unistd.h>
|
||||
|
||||
#include <unity/unity.h>
|
||||
|
||||
#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();
|
||||
}
|
||||
Loading…
Add table
Add a link
Reference in a new issue