From 0c011f784e2f027412f941fe0cfe13fc519adc83 Mon Sep 17 00:00:00 2001 From: Rongsong Shen Date: Wed, 30 Sep 2026 09:12:01 +0000 Subject: [PATCH] feat(scheduler-allocator-runner): runner plan and claim search (task 3.2) The runner tries the plugin's plans in order and claims one distinct host per bucket, with parjob's rollback and claim snapshots, nslots/nhosts per candidate, a growable claims buffer and malformed-plan skipping. It clears alloc_req.host before every candidate so each short host gets its pend reasons (Req 3.6/3.7). Result assembly follows in task 3.3; until then a satisfied plan's claims are released and the attempt pends. Co-Authored-By: Claude Opus 5.5 (1M context) --- .../specs/scheduler-allocator-runner/tasks.md | 2 +- src/scheduler/src/alloc_runner.c | 293 ++++++- src/scheduler/testsuite/meson.build | 15 + .../testsuite/test_alloc_runner_search.c | 784 ++++++++++++++++++ 4 files changed, 1080 insertions(+), 14 deletions(-) create mode 100644 src/scheduler/testsuite/test_alloc_runner_search.c diff --git a/.kiro/specs/scheduler-allocator-runner/tasks.md b/.kiro/specs/scheduler-allocator-runner/tasks.md index 55107f10..336e9935 100644 --- a/.kiro/specs/scheduler-allocator-runner/tasks.md +++ b/.kiro/specs/scheduler-allocator-runner/tasks.md @@ -38,7 +38,7 @@ - _Depends: 1.1, 1.2_ - _Requirements: 1.3, 1.4, 2.1, 2.2, 2.3, 2.4, 5.1, 7.2, 7.3_ -- [ ] 3.2 Build the plan and claim search with rollback, test first +- [x] 3.2 Build the plan and claim search with rollback, test first - Plans are tried in generator order and the first fully satisfied plan wins; each bucket claims the first unclaimed satisfying host in host-iteration order - Shared amounts are counted as uncommitted exactly once across an attempt that tries several plans - The per-host request is empty before every candidate host is pre-filtered (cleared at the start of each bucket and after every rejected candidate), so every selection-matching unclaimed short host gets its host/resource reasons; before each candidate the slot count is the bucket's task count and the host count is 1 diff --git a/src/scheduler/src/alloc_runner.c b/src/scheduler/src/alloc_runner.c index b8c47670..acac3100 100644 --- a/src/scheduler/src/alloc_runner.c +++ b/src/scheduler/src/alloc_runner.c @@ -12,7 +12,8 @@ * no check (Req 1.3, 1.4) * 3. the pend-reason record, created only after both guards (Req 2.1) * 4. the shared-resource precheck (Req 2.2-2.4) - * 5. the plan generator (Req 3.4) + * 5. the plan generator (Req 3.4) and the plan and claim search with + * rollback (Req 3.1-3.8, 5.5, 5.7) * * Returns > 0 when satisfied (ctx->alloc set), 0 when pending, < 0 a * negative errno. No plugin file or plugin-specific symbol is referenced @@ -20,12 +21,16 @@ * iface type (alloc_iter.h). */ #include +#include +#include #include "preason.h" #include "resobj.h" #include "sched/alloc_iter.h" #include "sched/types.h" #include "scheduler.h" +#include "utils.h" +#include "xvec.h" /* Creates the attempt's pend-reason record unless an earlier primary of * the same COMPOUND/ALTER attempt already did. 0 or -ENOMEM (ctx->reason @@ -102,11 +107,283 @@ static int sched_alloc_prepare_shared_(xce_sched_core_t* core, return 1; } +/* One bucket's claim within the plan being tried: the claimed host and a + * deep copy of ctx->alloc_req.host as it stood when the host was accepted. + * The copy is needed because the per-host prepare pass rebuilds + * ctx->alloc_req.host for every later candidate; rollback subtracts + * exactly the copied request. */ +typedef struct { + xce_sched_worker_t* w; + xvec_idval_t host_req; +} sched_alloc_claim_t; + +static int sched_alloc_is_claimed_(const sched_alloc_claim_t* claims, int n, + const xce_sched_worker_t* w) +{ + int i; + + for (i = 0; i < n; i++) { + if (claims[i].w == w) { + return 1; + } + } + + return 0; +} + +/* Empties the per-host request, so the host iterator's pre-filter + * (host_iter.c) applies the select expression only. */ +static void sched_alloc_clear_host_req_(xce_sched_ctx_t* ctx) +{ + xvec_idval_fini(&(ctx->alloc_req.host)); + xvec_idval_init(&(ctx->alloc_req.host), 8); +} + +/* Deep-copies `src` into `dst` (xvec.h has no xvec_idval_t copy). 0, or + * the xvec_dup() error with `dst` zeroed and nothing left allocated. */ +static int sched_snapshot_host_req_(xvec_idval_t* dst, const xvec_idval_t* src) +{ + xvec_t* idxs = NULL; + xvec_t* vals = NULL; + int rc; + + memset(dst, 0, sizeof(*dst)); + + rc = xvec_dup(&idxs, (xvec_t*)&(src->idxs)); + if (rc < 0) { + if (idxs != NULL) { + xvec_fini(idxs); + } + FREEUP(idxs); + return rc; + } + + rc = xvec_dup(&vals, (xvec_t*)&(src->vals)); + if (rc < 0) { + xvec_fini(idxs); + FREEUP(idxs); + if (vals != NULL) { + xvec_fini(vals); + } + FREEUP(vals); + return rc; + } + + /* dst takes over the data buffers; only the wrappers are freed */ + dst->count = src->count; + dst->idxs = *idxs; + dst->vals = *vals; + FREEUP(idxs); + FREEUP(vals); + + return 0; +} + +/* Releases the per-host debit of claims[0..n) and frees their snapshots + * (Req 3.3, 5.5). Touches only ctx->uncomitted. */ +static void sched_rollback_claims_(xce_sched_ctx_t* ctx, + sched_alloc_claim_t* claims, int n) +{ + int i; + + for (i = 0; i < n; i++) { + sched_ctx_uncommit_host(ctx, claims[i].w->idx, &(claims[i].host_req)); + xvec_idval_fini(&(claims[i].host_req)); + } +} + +/* Claims one host for bucket `b` of `plan` into claims[*n]. Candidates + * come in host-iteration order; hosts already claimed by this plan are + * skipped (Req 3.2). Each candidate is checked with the bucket's slot + * count and one host (Req 3.5), and the per-host request is empty before + * every next() call: at the start of the bucket and after every rejected + * candidate (Req 3.6), so every selection-matching unclaimed short host + * reaches the per-host check and gets its reasons (Req 3.7). + * + * 1 claimed (*n advanced), 0 no host, < 0 snapshot failure (the + * candidate's debit released here; earlier claims left to the caller). */ +static int sched_claim_bucket_(xce_sched_core_t* core, xce_sched_ctx_t* ctx, + xce_alloc_plan_t* plan, int b, + sched_alloc_claim_t* claims, int* n) +{ + xce_host_iter_t it; + xce_sched_worker_t* cand; + int rc = 0; + + sched_alloc_clear_host_req_(ctx); + + it.init = sched_host_iter_init; + it.next = sched_host_iter_next; + it.fini = sched_host_iter_fini; + + /* An iterator that cannot start offers no candidate: an + * infrastructure failure, not a resource verdict (alloc_iter.h). */ + if (it.init(&it, plan, core->workers) != 0) { + return 0; + } + + while ((cand = it.next(&it)) != NULL) { + if (sched_alloc_is_claimed_(claims, *n, cand)) { + continue; + } + + ctx->nslots = plan->bucket_slots[b]; + ctx->nhosts = 1; + + if (sched_try_alloc_on_host_(cand, core, ctx) > 0) { + rc = sched_snapshot_host_req_(&(claims[*n].host_req), + &(ctx->alloc_req.host)); + if (rc < 0) { + /* no claim records this debit: undo it with the live + * request that made it */ + sched_ctx_uncommit_host(ctx, cand->idx, &(ctx->alloc_req.host)); + break; + } + + claims[*n].w = cand; + (*n)++; + rc = 1; + break; + } + + /* rejected: its request must not pre-filter the next candidate */ + sched_alloc_clear_host_req_(ctx); + } + + it.fini(&it); + + return rc; +} + +/* Tries one plan: one distinct host per bucket, buckets in order. + * `claims` holds at least plan->nbuckets entries. + * + * 1: every bucket claimed; claims[0..*nclaimed) and their snapshots + * belong to the caller. + * 0: a bucket had no host; every claim of the plan released (Req 3.3). + * < 0: a snapshot failed; the candidate and every earlier claim released + * (Req 5.5). + * On 0 and < 0, *nclaimed is 0 and no snapshot remains. */ +static int sched_try_plan_(xce_sched_core_t* core, xce_sched_ctx_t* ctx, + xce_alloc_plan_t* plan, sched_alloc_claim_t* claims, + int* nclaimed) +{ + int n = 0; + int b; + int rc; + + plan->ctx = ctx; + *nclaimed = 0; + + for (b = 0; b < plan->nbuckets; b++) { + rc = sched_claim_bucket_(core, ctx, plan, b, claims, &n); + if (rc <= 0) { + sched_rollback_claims_(ctx, claims, n); + return rc; + } + } + + *nclaimed = n; + + return 1; +} + +/* A plan the iterator contract does not allow: no bucket, or a bucket of + * fewer than one slot. */ +static int sched_plan_malformed_(const xce_alloc_plan_t* plan) +{ + int i; + + if ((plan->nbuckets < 1) || (plan->bucket_slots == NULL)) { + return 1; + } + + for (i = 0; i < plan->nbuckets; i++) { + if (plan->bucket_slots[i] < 1) { + return 1; + } + } + + return 0; +} + +/* Grows *claims to hold `need` entries (Req 3.8: no fixed limit). 0 or + * -ENOMEM with *claims unchanged. */ +static int sched_alloc_claims_reserve_(sched_alloc_claim_t** claims, int* cap, + int need) +{ + sched_alloc_claim_t* grown; + + if (need <= *cap) { + return 0; + } + + grown = realloc(*claims, (size_t)need * sizeof(*grown)); + if (grown == NULL) { + return -ENOMEM; + } + + *claims = grown; + *cap = need; + + return 0; +} + +/* Tries the generator's plans in order until one is satisfied (Req 3.1). + * The generator is started and is finished here on every path. + * + * This runner builds no allocation result: a satisfied plan's claims + * are released and the attempt reports pending (0). No plan satisfied: 0. + * Claims-buffer growth or snapshot failure: that error. */ +static int sched_alloc_search_(const xce_sched_allocator_iface_t* iface, + xce_sched_core_t* core, xce_sched_ctx_t* ctx) +{ + xce_alloc_plan_iter_t iter; + xce_alloc_plan_t* plan; + sched_alloc_claim_t* claims = NULL; + int cap = 0; + int n = 0; + int rc = 0; + + /* A generator that cannot start is not a resource verdict: the attempt + * pends, with nothing for plan_fini to release (Req 3.4). */ + if (iface->plan_init(&iter, ctx->primary, core) != 0) { + return 0; + } + + while ((plan = iface->plan_next(&iter)) != NULL) { + if (sched_plan_malformed_(plan)) { + continue; + } + + rc = sched_alloc_claims_reserve_(&claims, &cap, plan->nbuckets); + if (rc < 0) { + break; + } + + rc = sched_try_plan_(core, ctx, plan, claims, &n); + if (rc != 0) { + break; + } + } + + if (rc > 0) { + /* no allocation result is built from the claims: they are + * released and the attempt reports pending */ + sched_rollback_claims_(ctx, claims, n); + rc = 0; + } + + iface->plan_fini(&iter); + free(claims); + + return rc; +} + int sched_allocator_run(void* plugin_arg, xce_sched_ctx_t* ctx) { xce_sched_plugin_t* plugin = plugin_arg; const xce_sched_allocator_iface_t* iface; - xce_alloc_plan_iter_t iter; int rc; if ((plugin == NULL) || (ctx == NULL) || (plugin->core == NULL) @@ -134,15 +411,5 @@ int sched_allocator_run(void* plugin_arg, xce_sched_ctx_t* ctx) return rc; } - /* A plan generator that cannot start is not a resource verdict: the - * attempt pends, with nothing for plan_fini to release. Otherwise no - * plan is searched, so the attempt pends once the generator is - * finished. */ - if (iface->plan_init(&iter, ctx->primary, plugin->core) != 0) { - return 0; - } - - iface->plan_fini(&iter); - - return 0; + return sched_alloc_search_(iface, plugin->core, ctx); } diff --git a/src/scheduler/testsuite/meson.build b/src/scheduler/testsuite/meson.build index 38fb3d76..8c076de9 100644 --- a/src/scheduler/testsuite/meson.build +++ b/src/scheduler/testsuite/meson.build @@ -1756,6 +1756,18 @@ sched_alloc_runner_test = executable( link_args: ['-Wl,--wrap=make_pendreason'], install: false) +# scheduler-allocator-runner task 3.2 (Req 2.3, 3.1-3.8, 5.5, 5.7): the +# runner's plan and claim search with rollback, same white-box stack as +# above. xvec_dup is wrapped to fail a claim snapshot; the claims-buffer +# realloc is replaced inside the test file (alloc_runner.c only). +sched_alloc_runner_search_test = executable( + 'sched_alloc_runner_search_test', + ['test_alloc_runner_search.c'], + include_directories:[incdir], + dependencies: [scheduler_test_deps], + link_args: ['-Wl,--wrap=xvec_dup'], + install: false) + # Task 3.2 (scheduler-allocation spec, Req 9.1): the dedicated pre/post- # rewrap host-selection EQUIVALENCE proof for seqjob_allocator.c's # my_alloc() -- fixture states task 3.1's own smoke test does not cover @@ -2031,6 +2043,9 @@ test('scheduler alloc equivalence parjob test', # Leak detection stays on: every fixture tears down its ctx and core. test('scheduler alloc runner test', sched_alloc_runner_test, suite: 'scheduler') +# Leak detection stays on: every fixture tears down its ctx and core. +test('scheduler alloc runner search test', + sched_alloc_runner_search_test, suite: 'scheduler') test('scheduler seqjob rewrap equivalence test', sched_seqjob_rewrap_equivalence_test, suite: 'scheduler', env: ['ASAN_OPTIONS=detect_leaks=0']) diff --git a/src/scheduler/testsuite/test_alloc_runner_search.c b/src/scheduler/testsuite/test_alloc_runner_search.c new file mode 100644 index 00000000..6acd7cdd --- /dev/null +++ b/src/scheduler/testsuite/test_alloc_runner_search.c @@ -0,0 +1,784 @@ +/* + * test_alloc_runner_search.c - scheduler-allocator-runner task 3.2: the + * runner's plan and claim search with rollback (Req 2.3, 3.1-3.8, 5.5, + * 5.7) through sched_allocator_run() with a scripted fake plan generator, + * plus direct white-box calls of sched_try_plan_()/sched_rollback_claims_(). + * The plugin_prepare_allocation() stub mirrors plugin.c plus parjob's hook + * and logs every per-host call (worker, nslots, nhosts, request size at + * entry). Faults: the runner's claims-buffer realloc() is replaced by a + * call-counting test_realloc() (alloc_runner.c only); the claim snapshot's + * xvec_dup() is link-wrapped and armed to fail the Nth call from arming. + */ +#include +#include +#include + +#include + +#include "preason.h" +#include "resobj.h" +#include "resreq.h" +#include "sched/alloc_iter.h" +#include "sched/types.h" +#include "scheduler.h" +#include "utils.h" +#include "xcealloc.h" +#include "xstring.h" +#include "xvec.h" + +/* White-box sources under test; realloc() is replaced in the runner only */ +static void* test_realloc(void* p, size_t n); +#define realloc test_realloc +#include "../src/alloc_runner.c" +#undef realloc +#include "../src/core.c" +#include "../src/host_iter.c" +#include "../src/resource.c" +#include "../src/schedobj.c" + +/* ---------------------------------------------------------------------- */ +/* Link-satisfying stubs (same convention as test_alloc_runner.c) */ +/* ---------------------------------------------------------------------- */ + +int plugin_register(void* s, void* p) { return (void)s, (void)p, 0; } +void plugin_unregister(void* arg) { (void)arg; } + +int plugin_job_update(void* s, xce_sched_job_t* j) +{ + return (void)s, (void)j, 0; +} +int plugin_queue_update(void* s, xce_sched_queue_t* q) +{ + return (void)s, (void)q, 0; +} +int plugin_job_rusage_update(void* s, xce_sched_job_t* j, xce_job_rusages_t* r) +{ + return (void)s, (void)j, (void)r, 0; +} + +/* Never reached: this runner builds no allocation result. */ +xce_alloc_t** sched_make_allocs(xce_scheduler_t* sched, char* reserve, int nres, + int* idxs, float* vals, int* duration) +{ + (void)sched, (void)reserve, (void)nres, (void)idxs, (void)vals; + (void)duration; + errno = ENOSYS; + return NULL; +} + +/* ---------------------------------------------------------------------- */ +/* Fault injection */ +/* ---------------------------------------------------------------------- */ + +static int g_realloc_calls = 0; +static int g_realloc_fail_on = 0; /* 1-based call number; 0 = never */ + +static void* test_realloc(void* p, size_t n) +{ + g_realloc_calls++; + if ((g_realloc_fail_on > 0) && (g_realloc_calls == g_realloc_fail_on)) { + errno = ENOMEM; + return NULL; + } + return realloc(p, n); +} + +/* xvec_dup(): armed with the call number (1-based, counted from arming) + * that fails; disarms once it has failed. */ +static int g_dup_calls = 0; +static int g_dup_fail_on = 0; + +int __real_xvec_dup(xvec_t** dst, xvec_t* src); + +int __wrap_xvec_dup(xvec_t** dst, xvec_t* src) +{ + if (g_dup_fail_on > 0) { + g_dup_calls++; + if (g_dup_calls == g_dup_fail_on) { + g_dup_fail_on = 0; + *dst = NULL; + return -ENOMEM; + } + } + return __real_xvec_dup(dst, src); +} + +static void dup_arm(int nth) { g_dup_calls = 0, g_dup_fail_on = nth; } + +/* ---------------------------------------------------------------------- */ +/* plugin_prepare_allocation() stub with a per-host call log */ +/* ---------------------------------------------------------------------- */ + +#define MAX_LOG 32 +typedef struct { + xce_sched_worker_t* w; + int nslots; + int nhosts; + int host_count_at_entry; +} prep_log_t; + +static prep_log_t g_log[MAX_LOG]; +static int g_nlog = 0; +static int g_slots_idx = -1; +static int g_lic_idx = -1; +static float g_shared_want = 0.0f; /* lic per job; 0 = none */ + +int plugin_prepare_allocation(xce_sched_core_t* sched, xce_sched_ctx_t* ctx) +{ + (void)sched; + if (ctx->worker == NULL) { + xvec_idval_fini(&(ctx->alloc_req.shared)); + xvec_idval_init(&(ctx->alloc_req.shared), 8); + xvec_idval_fini(&(ctx->alloc_req.host)); + xvec_idval_init(&(ctx->alloc_req.host), 8); + if (g_shared_want > 0.0f) { + TEST_ASSERT_TRUE(xvec_idval_store(&(ctx->alloc_req.shared), + g_lic_idx, g_shared_want) + > 0); + } + return 0; + } + + TEST_ASSERT_TRUE(g_nlog < MAX_LOG); + g_log[g_nlog].w = ctx->worker; + g_log[g_nlog].nslots = ctx->nslots; + g_log[g_nlog].nhosts = ctx->nhosts; + g_log[g_nlog].host_count_at_entry = ctx->alloc_req.host.count; + g_nlog++; + + xvec_idval_fini(&(ctx->alloc_req.host)); + xvec_idval_init(&(ctx->alloc_req.host), 8); + TEST_ASSERT_TRUE(xvec_idval_store(&(ctx->alloc_req.host), g_slots_idx, + (float)ctx->nslots) + > 0); + return 0; +} + +/* ---------------------------------------------------------------------- */ +/* Scripted fake plan generator */ +/* ---------------------------------------------------------------------- */ + +#define MAX_PLANS 6 +#define MAX_BUCKETS 6 +static int g_plan_nb[MAX_PLANS]; +static int g_plan_slots[MAX_PLANS][MAX_BUCKETS]; +static int g_nplans = 0; +static int g_init_result = 0; +static int g_init_calls = 0; +static int g_next_calls = 0; +static int g_fini_calls = 0; +static int g_cur_slots[MAX_BUCKETS]; +static xce_alloc_plan_t g_cur_plan; + +static int fake_plan_init(xce_alloc_plan_iter_t* iter, + xce_primary_resreq_t* resreq, xce_sched_core_t* core) +{ + (void)resreq, (void)core; + g_init_calls++; + iter->private = NULL; + return g_init_result; +} + +/* Yields scripted plan g_next_calls-1; NULL once the script is done */ +static xce_alloc_plan_t* fake_plan_next(xce_alloc_plan_iter_t* iter) +{ + int p = g_next_calls++; + + (void)iter; + if (p >= g_nplans) { + return NULL; + } + memset(&g_cur_plan, 0, sizeof(g_cur_plan)); + memcpy(g_cur_slots, g_plan_slots[p], sizeof(g_cur_slots)); + g_cur_plan.nbuckets = g_plan_nb[p]; + g_cur_plan.bucket_slots = g_cur_slots; + return &g_cur_plan; +} + +static void fake_plan_fini(xce_alloc_plan_iter_t* iter) +{ + (void)iter; + g_fini_calls++; +} + +static xce_sched_allocator_iface_t g_iface = { + .accepts = NULL, + .plan_init = fake_plan_init, + .plan_next = fake_plan_next, + .plan_fini = fake_plan_fini, +}; + +/* Appends one plan of `nb` buckets */ +static void script_plan(int nb, int s0, int s1, int s2) +{ + TEST_ASSERT_TRUE(g_nplans < MAX_PLANS); + g_plan_nb[g_nplans] = nb; + g_plan_slots[g_nplans][0] = s0; + g_plan_slots[g_nplans][1] = s1; + g_plan_slots[g_nplans][2] = s2; + g_nplans++; +} + +void setUp(void) +{ + g_realloc_calls = 0; + g_realloc_fail_on = 0; + g_dup_calls = 0; + g_dup_fail_on = 0; + g_nlog = 0; + g_shared_want = 0.0f; + g_nplans = 0; + g_init_result = 0; + g_init_calls = 0; + g_next_calls = 0; + g_fini_calls = 0; + memset(g_log, 0, sizeof(g_log)); + memset(g_plan_slots, 0, sizeof(g_plan_slots)); +} + +void tearDown(void) { } + +/* ---------------------------------------------------------------------- */ +/* Fixture */ +/* ---------------------------------------------------------------------- */ + +typedef struct { + xce_scheduler_t sched; + xce_resource_t* slots; + xce_resource_t* lic; /* shared, 10 in total */ + xce_sched_plugin_t plugin; + xce_primary_resreq_t req; + xce_sched_obj_t obj; + xce_primary_schedobj_t primary_obj; + xce_sched_ctx_t ctx; +} sr_fixture_t; + +static void sr_fixture_init(sr_fixture_t* f) +{ + int i; + int idx[1]; + float total[1] = { 10.0f }; + + memset(f, 0, sizeof(*f)); + TEST_ASSERT_EQUAL_INT(0, xce_sched_core_init(&(f->sched.core))); + /* xce_sched_core_fini() finalizes these; both drivers calloc them */ + for (i = 0; i < 2; i++) { + f->sched.core.pend_updates[i] = (dataq_t*)calloc(1, sizeof(dataq_t)); + TEST_ASSERT_NOT_NULL(f->sched.core.pend_updates[i]); + dataq_init(f->sched.core.pend_updates[i], NULL); + } + + f->slots = makeResource(f->sched.core.resmgr, + &(xce_resource_t) { .name = (char*)"slots" }); + f->lic = makeResource( + f->sched.core.resmgr, + &(xce_resource_t) { .name = (char*)"lic", .flags = XCE_RESF_GLOBAL }); + TEST_ASSERT_NOT_NULL(f->slots); + TEST_ASSERT_NOT_NULL(f->lic); + g_slots_idx = f->slots->idx; + g_lic_idx = f->lic->idx; + idx[0] = f->lic->idx; + put_resvals(&(f->sched.core.sharedInst), 1, idx, total); + + f->plugin.name = (char*)"fake_allocator"; + f->plugin.sched = &(f->sched); + f->plugin.core = &(f->sched.core); + f->plugin.allocator_iface = &g_iface; +} + +/* Host with `slots` free slots, none in use */ +static xce_sched_worker_t* sr_add_host(sr_fixture_t* f, const char* name, + float slots) +{ + xce_sched_worker_t* w; + int idx[1] = { f->slots->idx }; + float total[1] = { slots }; + float inuse[1] = { 0.0f }; + + w = (xce_sched_worker_t*)calloc(1, sizeof(*w)); + TEST_ASSERT_NOT_NULL(w); + w->name = makeString((char*)name); + TEST_ASSERT_NOT_NULL(w->name); + zarray_init(&(w->resvals), 8, (zarray_item_killer_t)kill_sched_resval); + zarray_init(&(w->resInUse), 8, (zarray_item_killer_t)kill_sched_resval); + put_resvals(&(w->resvals), 1, idx, total); + put_resvals(&(w->resInUse), 1, idx, inuse); + TEST_ASSERT_TRUE(sched_add_worker(&(f->sched.core), w) >= 0); + + return w; +} + +/* ctx over a `max_tasks` primary, job-level prepare pass included. Call + * after every sr_add_host(). */ +static void sr_ctx_begin(sr_fixture_t* f, int max_tasks) +{ + f->req.minTasks = 1; + f->req.maxTasks = max_tasks; + f->primary_obj.resreq = &(f->req); + f->obj.tag = XCE_RESREQ_PRIMARY; + f->obj.u.primary = &(f->primary_obj); + TEST_ASSERT_EQUAL_INT( + 0, sched_ctx_init(&(f->ctx), &(f->sched.core), &(f->obj))); + sched_ctx_set_primary(&(f->ctx), &(f->req)); +} + +static void sr_fixture_fini(sr_fixture_t* f) +{ + free_JobAlloc_ptr(&(f->ctx.alloc)); + free_PendReason_ptr(&(f->ctx.reason)); + sched_ctx_fini(&(f->ctx)); + xce_sched_core_fini(&(f->sched.core)); + /* xce_sched_core_init() creates the string table on every call and + * xce_sched_core_fini() leaves it; release it per fixture */ + xstring_table_fini(); +} + +static int sr_run(sr_fixture_t* f) +{ + return sched_allocator_run(&(f->plugin), &(f->ctx)); +} + +static float sr_slots_debit(sr_fixture_t* f, xce_sched_worker_t* w) +{ + xvec_t* v = (xvec_t*)zarray_get(&(f->ctx.uncomitted.hosts), w->idx); + + return (v == NULL) ? 0.0f : ((float*)v->data)[f->slots->idx]; +} + +static float sr_shared_debit(sr_fixture_t* f) +{ + return ((float*)f->ctx.uncomitted.shared->data)[f->lic->idx]; +} + +static void sr_assert_log(int i, xce_sched_worker_t* w, int nslots) +{ + TEST_ASSERT_TRUE(i < g_nlog); + TEST_ASSERT_EQUAL_PTR(w, g_log[i].w); + TEST_ASSERT_EQUAL_INT(nslots, g_log[i].nslots); + TEST_ASSERT_EQUAL_INT(1, g_log[i].nhosts); + TEST_ASSERT_EQUAL_INT(0, g_log[i].host_count_at_entry); +} + +static void sr_assert_slots_reason(sr_fixture_t* f, int i, const char* host) +{ + xce_res_reason_t* rr; + + TEST_ASSERT_TRUE(i < f->ctx.reason->reasons->nres_reasons); + rr = f->ctx.reason->reasons->res_based_reasons[i]; + TEST_ASSERT_NOT_NULL(rr); + TEST_ASSERT_EQUAL_INT(PREASON_NO_RESOURCE, rr->reason); + TEST_ASSERT_EQUAL_STRING(host, rr->host); + TEST_ASSERT_EQUAL_STRING("slots", rr->resource); +} + +/* Direct sched_try_plan_() over a one-plan script entry: the caller owns + * the reason record, as the runner does. */ +static int sr_try_plan(sr_fixture_t* f, xce_alloc_plan_t* plan, + sched_alloc_claim_t* claims, int* nclaimed) +{ + if (f->ctx.reason == NULL) { + f->ctx.reason = make_pendreason(0); + TEST_ASSERT_NOT_NULL(f->ctx.reason); + } + return sched_try_plan_(&(f->sched.core), &(f->ctx), plan, claims, nclaimed); +} + +/* ---------------------------------------------------------------------- */ +/* Plan order and shared counting (Req 3.1, 3.4, 2.3) */ +/* ---------------------------------------------------------------------- */ + +/* [4] fails on 2-slot hosts, [2,2] is satisfied, [1,1,1,1] never offered: + * candidates in host order, each with its bucket's slot count. */ +static void test_plans_in_order_first_satisfied_wins(void) +{ + sr_fixture_t f; + xce_sched_worker_t *a, *b; + + sr_fixture_init(&f); + a = sr_add_host(&f, "sr-a", 2.0f); + b = sr_add_host(&f, "sr-b", 2.0f); + sr_ctx_begin(&f, 4); + script_plan(1, 4, 0, 0); + script_plan(2, 2, 2, 0); + script_plan(4, 1, 1, 1); + + TEST_ASSERT_EQUAL_INT(0, sr_run(&f)); + TEST_ASSERT_EQUAL_INT(2, g_next_calls); + TEST_ASSERT_EQUAL_INT(1, g_init_calls); + TEST_ASSERT_EQUAL_INT(1, g_fini_calls); + TEST_ASSERT_EQUAL_INT(4, g_nlog); + sr_assert_log(0, a, 4); + sr_assert_log(1, b, 4); + sr_assert_log(2, a, 2); + sr_assert_log(3, b, 2); + sr_fixture_fini(&f); +} + +/* Three unsatisfiable plans: the shared amount is counted once. */ +static void test_shared_counted_once_across_three_plans(void) +{ + sr_fixture_t f; + + sr_fixture_init(&f); + sr_add_host(&f, "sr-a", 2.0f); + g_shared_want = 3.0f; + sr_ctx_begin(&f, 4); + script_plan(1, 4, 0, 0); + script_plan(2, 3, 1, 0); + script_plan(2, 2, 2, 0); + + TEST_ASSERT_EQUAL_INT(0, sr_run(&f)); + TEST_ASSERT_EQUAL_INT(4, g_next_calls); + TEST_ASSERT_EQUAL_INT(1, g_fini_calls); + TEST_ASSERT_EQUAL_FLOAT(3.0f, sr_shared_debit(&f)); + sr_fixture_fini(&f); +} + +/* Generator initialisation failure: pending, no plan_next/plan_fini. */ +static void test_plan_init_failure_pends_without_fini(void) +{ + sr_fixture_t f; + + sr_fixture_init(&f); + sr_add_host(&f, "sr-a", 2.0f); + sr_ctx_begin(&f, 2); + script_plan(1, 2, 0, 0); + g_init_result = -ENOMEM; + + TEST_ASSERT_EQUAL_INT(0, sr_run(&f)); + TEST_ASSERT_EQUAL_INT(1, g_init_calls); + TEST_ASSERT_EQUAL_INT(0, g_next_calls); + TEST_ASSERT_EQUAL_INT(0, g_fini_calls); + TEST_ASSERT_EQUAL_INT(0, g_nlog); + sr_fixture_fini(&f); +} + +/* ---------------------------------------------------------------------- */ +/* Claims: distinct hosts, rollback (Req 3.2, 3.3, 3.5, 5.5) */ +/* ---------------------------------------------------------------------- */ + +/* A could take both buckets; bucket 2 skips A (claimed) and takes B. */ +static void test_distinct_hosts_per_bucket(void) +{ + sr_fixture_t f; + xce_sched_worker_t *a, *b; + sched_alloc_claim_t claims[2]; + int slots[2] = { 2, 2 }; + xce_alloc_plan_t plan = { .nbuckets = 2, .bucket_slots = slots }; + int n = -1; + + sr_fixture_init(&f); + a = sr_add_host(&f, "sr-a", 4.0f); + b = sr_add_host(&f, "sr-b", 4.0f); + sr_ctx_begin(&f, 4); + + TEST_ASSERT_EQUAL_INT(1, sr_try_plan(&f, &plan, claims, &n)); + TEST_ASSERT_EQUAL_INT(2, n); + TEST_ASSERT_EQUAL_PTR(&(f.ctx), plan.ctx); + TEST_ASSERT_EQUAL_PTR(a, claims[0].w); + TEST_ASSERT_EQUAL_PTR(b, claims[1].w); + TEST_ASSERT_EQUAL_INT(2, g_nlog); + sr_assert_log(0, a, 2); + sr_assert_log(1, b, 2); + + /* each snapshot is its own copy of the claim's request */ + TEST_ASSERT_EQUAL_INT(1, claims[1].host_req.count); + TEST_ASSERT_TRUE(claims[0].host_req.vals.data + != claims[1].host_req.vals.data); + TEST_ASSERT_TRUE(claims[1].host_req.vals.data + != f.ctx.alloc_req.host.vals.data); + TEST_ASSERT_EQUAL_FLOAT(2.0f, ((float*)claims[1].host_req.vals.data)[0]); + TEST_ASSERT_EQUAL_FLOAT(2.0f, sr_slots_debit(&f, a)); + TEST_ASSERT_EQUAL_FLOAT(2.0f, sr_slots_debit(&f, b)); + + sched_rollback_claims_(&(f.ctx), claims, n); + TEST_ASSERT_EQUAL_FLOAT(0.0f, sr_slots_debit(&f, a)); + TEST_ASSERT_EQUAL_FLOAT(0.0f, sr_slots_debit(&f, b)); + sr_fixture_fini(&f); +} + +/* Bucket 1 claims A, bucket 2 finds no host (A claimed, B short): 0, A's + * debit released, no claim left. */ +static void test_failing_bucket_releases_earlier_claims(void) +{ + sr_fixture_t f; + xce_sched_worker_t *a, *b; + sched_alloc_claim_t claims[2]; + int slots[2] = { 2, 2 }; + xce_alloc_plan_t plan = { .nbuckets = 2, .bucket_slots = slots }; + int n = -1; + + sr_fixture_init(&f); + a = sr_add_host(&f, "sr-a", 2.0f); + b = sr_add_host(&f, "sr-b", 1.0f); + sr_ctx_begin(&f, 4); + + TEST_ASSERT_EQUAL_INT(0, sr_try_plan(&f, &plan, claims, &n)); + TEST_ASSERT_EQUAL_INT(0, n); + TEST_ASSERT_EQUAL_FLOAT(0.0f, sr_slots_debit(&f, a)); + TEST_ASSERT_EQUAL_FLOAT(0.0f, sr_slots_debit(&f, b)); + TEST_ASSERT_EQUAL_INT(2, g_nlog); + sr_assert_log(0, a, 2); + sr_assert_log(1, b, 2); + sr_fixture_fini(&f); +} + +/* Through the runner: [2,2] fails in bucket 2, [2] then wins on A; the + * slot and host counts seen by every candidate check are the bucket's. */ +static void test_rollback_then_later_plan_with_bucket_counts(void) +{ + sr_fixture_t f; + xce_sched_worker_t *a, *b; + + sr_fixture_init(&f); + a = sr_add_host(&f, "sr-a", 2.0f); + b = sr_add_host(&f, "sr-b", 1.0f); + sr_ctx_begin(&f, 2); + f.ctx.nslots = 99; + f.ctx.nhosts = 99; + script_plan(3, 1, 1, 1); + script_plan(1, 2, 0, 0); + + TEST_ASSERT_EQUAL_INT(0, sr_run(&f)); + TEST_ASSERT_EQUAL_INT(2, g_next_calls); + /* [1,1,1]: A, B claimed; bucket 3 has no unclaimed host */ + TEST_ASSERT_EQUAL_INT(3, g_nlog); + sr_assert_log(0, a, 1); + sr_assert_log(1, b, 1); + sr_assert_log(2, a, 2); + TEST_ASSERT_EQUAL_FLOAT(0.0f, sr_slots_debit(&f, a)); + TEST_ASSERT_EQUAL_FLOAT(0.0f, sr_slots_debit(&f, b)); + TEST_ASSERT_EQUAL_INT(1, g_fini_calls); + sr_fixture_fini(&f); +} + +/* ---------------------------------------------------------------------- */ +/* Per-candidate clear and complete host reasons (Req 3.6, 3.7) */ +/* ---------------------------------------------------------------------- */ + +/* One bucket of 2 over three 1-slot hosts: every host is checked with an + * empty request at entry and gets its (host, slots) reason. A clear only + * at the start of the bucket leaves A's rejected want in place, and the + * iterator's pre-filter then skips B and C with no reason. */ +static void test_every_short_host_gets_reason(void) +{ + sr_fixture_t f; + xce_sched_worker_t *a, *b, *c; + + sr_fixture_init(&f); + a = sr_add_host(&f, "sr-a", 1.0f); + b = sr_add_host(&f, "sr-b", 1.0f); + c = sr_add_host(&f, "sr-c", 1.0f); + sr_ctx_begin(&f, 2); + script_plan(1, 2, 0, 0); + + TEST_ASSERT_EQUAL_INT(0, sr_run(&f)); + TEST_ASSERT_EQUAL_INT(3, g_nlog); + sr_assert_log(0, a, 2); + sr_assert_log(1, b, 2); + sr_assert_log(2, c, 2); + TEST_ASSERT_NOT_NULL(f.ctx.reason); + TEST_ASSERT_EQUAL_INT(0, f.ctx.reason->reasons->main_reason); + TEST_ASSERT_EQUAL_INT(3, f.ctx.reason->reasons->nres_reasons); + sr_assert_slots_reason(&f, 0, "sr-a"); + sr_assert_slots_reason(&f, 1, "sr-b"); + sr_assert_slots_reason(&f, 2, "sr-c"); + sr_fixture_fini(&f); +} + +/* ---------------------------------------------------------------------- */ +/* Malformed plans and the claims buffer (Req 3.8, 5.7) */ +/* ---------------------------------------------------------------------- */ + +/* nbuckets 0 and a zero-slot bucket are skipped unchecked; the next valid + * plan is still tried. */ +static void test_malformed_plans_skipped(void) +{ + sr_fixture_t f; + xce_sched_worker_t* a; + + sr_fixture_init(&f); + a = sr_add_host(&f, "sr-a", 2.0f); + sr_ctx_begin(&f, 2); + script_plan(0, 2, 0, 0); + script_plan(2, 2, 0, 0); + script_plan(2, -1, 2, 0); + script_plan(1, 2, 0, 0); + + TEST_ASSERT_EQUAL_INT(0, sr_run(&f)); + TEST_ASSERT_EQUAL_INT(4, g_next_calls); + TEST_ASSERT_EQUAL_INT(1, g_nlog); + sr_assert_log(0, a, 2); + TEST_ASSERT_EQUAL_INT(1, g_fini_calls); + sr_fixture_fini(&f); +} + +/* A one-bucket plan followed by a three-bucket plan: the buffer grows + * (two realloc calls) and the larger plan claims three hosts. */ +static void test_claims_buffer_grows_to_larger_plan(void) +{ + sr_fixture_t f; + xce_sched_worker_t *a, *b, *c; + + sr_fixture_init(&f); + a = sr_add_host(&f, "sr-a", 1.0f); + b = sr_add_host(&f, "sr-b", 1.0f); + c = sr_add_host(&f, "sr-c", 1.0f); + sr_ctx_begin(&f, 3); + script_plan(1, 3, 0, 0); + script_plan(3, 1, 1, 1); + + TEST_ASSERT_EQUAL_INT(0, sr_run(&f)); + TEST_ASSERT_EQUAL_INT(2, g_realloc_calls); + TEST_ASSERT_EQUAL_INT(2, g_next_calls); + TEST_ASSERT_EQUAL_INT(6, g_nlog); + sr_assert_log(3, a, 1); + sr_assert_log(4, b, 1); + sr_assert_log(5, c, 1); + TEST_ASSERT_EQUAL_INT(1, g_fini_calls); + sr_fixture_fini(&f); +} + +/* Growth failure at the first and at the second realloc: -ENOMEM, the + * generator finished once, no host debit outstanding. */ +static void test_claims_buffer_growth_failure(void) +{ + int fail_on; + + for (fail_on = 1; fail_on <= 2; fail_on++) { + sr_fixture_t f; + xce_sched_worker_t *a, *b, *c; + + setUp(); + sr_fixture_init(&f); + a = sr_add_host(&f, "sr-a", 1.0f); + b = sr_add_host(&f, "sr-b", 1.0f); + c = sr_add_host(&f, "sr-c", 1.0f); + sr_ctx_begin(&f, 3); + script_plan(1, 3, 0, 0); + script_plan(3, 1, 1, 1); + g_realloc_fail_on = fail_on; + + TEST_ASSERT_EQUAL_INT(-ENOMEM, sr_run(&f)); + TEST_ASSERT_EQUAL_INT(fail_on, g_realloc_calls); + TEST_ASSERT_EQUAL_INT(1, g_fini_calls); + TEST_ASSERT_EQUAL_INT((fail_on == 1) ? 0 : 3, g_nlog); + TEST_ASSERT_EQUAL_FLOAT(0.0f, sr_slots_debit(&f, a)); + TEST_ASSERT_EQUAL_FLOAT(0.0f, sr_slots_debit(&f, b)); + TEST_ASSERT_EQUAL_FLOAT(0.0f, sr_slots_debit(&f, c)); + sr_fixture_fini(&f); + } +} + +/* ---------------------------------------------------------------------- */ +/* Snapshot failure (Req 5.5, 5.7) */ +/* ---------------------------------------------------------------------- */ + +/* Plan [2,2] on two 2-slot hosts; xvec_dup() fails at call `nth` (two + * calls per claim: 1-2 bucket 1, 3-4 bucket 2): the error is returned, + * the candidate and every earlier claim released, the generator finished + * once, nothing leaked. */ +static void sr_check_snapshot_failure(int nth) +{ + sr_fixture_t f; + xce_sched_worker_t *a, *b; + + setUp(); + sr_fixture_init(&f); + a = sr_add_host(&f, "sr-a", 2.0f); + b = sr_add_host(&f, "sr-b", 2.0f); + sr_ctx_begin(&f, 4); + script_plan(2, 2, 2, 0); + script_plan(1, 4, 0, 0); + dup_arm(nth); + + TEST_ASSERT_EQUAL_INT(-ENOMEM, sr_run(&f)); + TEST_ASSERT_EQUAL_INT(0, g_dup_fail_on); /* the armed call failed */ + TEST_ASSERT_EQUAL_INT(1, g_next_calls); + TEST_ASSERT_EQUAL_INT(1, g_fini_calls); + TEST_ASSERT_EQUAL_INT((nth <= 2) ? 1 : 2, g_nlog); + TEST_ASSERT_EQUAL_FLOAT(0.0f, sr_slots_debit(&f, a)); + TEST_ASSERT_EQUAL_FLOAT(0.0f, sr_slots_debit(&f, b)); + sr_fixture_fini(&f); +} + +static void test_snapshot_failure_releases_candidate_and_claims(void) +{ + int nth; + + for (nth = 1; nth <= 4; nth++) { + sr_check_snapshot_failure(nth); + } +} + +/* Direct sched_try_plan_(): a snapshot failure in bucket 2 returns the + * error with no claim reported. */ +static void test_try_plan_snapshot_failure_reports_no_claim(void) +{ + sr_fixture_t f; + xce_sched_worker_t *a, *b; + sched_alloc_claim_t claims[2]; + int slots[2] = { 2, 2 }; + xce_alloc_plan_t plan = { .nbuckets = 2, .bucket_slots = slots }; + int n = -1; + + sr_fixture_init(&f); + a = sr_add_host(&f, "sr-a", 2.0f); + b = sr_add_host(&f, "sr-b", 2.0f); + sr_ctx_begin(&f, 4); + dup_arm(3); + + TEST_ASSERT_EQUAL_INT(-ENOMEM, sr_try_plan(&f, &plan, claims, &n)); + TEST_ASSERT_EQUAL_INT(0, n); + TEST_ASSERT_EQUAL_FLOAT(0.0f, sr_slots_debit(&f, a)); + TEST_ASSERT_EQUAL_FLOAT(0.0f, sr_slots_debit(&f, b)); + sr_fixture_fini(&f); +} + +/* ---------------------------------------------------------------------- */ +/* Satisfied plan, current runner (Req 3.1; task 3.3 replaces this test) */ +/* ---------------------------------------------------------------------- */ + +/* This runner builds no allocation result: a satisfied plan's claims are + * released and the attempt pends, with nothing on ctx->alloc. */ +static void test_satisfied_plan_releases_claims_and_pends_interim(void) +{ + sr_fixture_t f; + xce_sched_worker_t *a, *b; + + sr_fixture_init(&f); + a = sr_add_host(&f, "sr-a", 2.0f); + b = sr_add_host(&f, "sr-b", 2.0f); + g_shared_want = 3.0f; + sr_ctx_begin(&f, 4); + script_plan(2, 2, 2, 0); + + TEST_ASSERT_EQUAL_INT(0, sr_run(&f)); + TEST_ASSERT_NULL(f.ctx.alloc); + TEST_ASSERT_EQUAL_INT(2, g_nlog); + TEST_ASSERT_EQUAL_FLOAT(0.0f, sr_slots_debit(&f, a)); + TEST_ASSERT_EQUAL_FLOAT(0.0f, sr_slots_debit(&f, b)); + TEST_ASSERT_EQUAL_FLOAT(3.0f, sr_shared_debit(&f)); + TEST_ASSERT_EQUAL_INT(1, g_next_calls); + TEST_ASSERT_EQUAL_INT(1, g_fini_calls); + sr_fixture_fini(&f); +} + +int main(void) +{ + UNITY_BEGIN(); + RUN_TEST(test_plans_in_order_first_satisfied_wins); + RUN_TEST(test_shared_counted_once_across_three_plans); + RUN_TEST(test_plan_init_failure_pends_without_fini); + RUN_TEST(test_distinct_hosts_per_bucket); + RUN_TEST(test_failing_bucket_releases_earlier_claims); + RUN_TEST(test_rollback_then_later_plan_with_bucket_counts); + RUN_TEST(test_every_short_host_gets_reason); + RUN_TEST(test_malformed_plans_skipped); + RUN_TEST(test_claims_buffer_grows_to_larger_plan); + RUN_TEST(test_claims_buffer_growth_failure); + RUN_TEST(test_snapshot_failure_releases_candidate_and_claims); + RUN_TEST(test_try_plan_snapshot_failure_reports_no_claim); + RUN_TEST(test_satisfied_plan_releases_claims_and_pends_interim); + return UNITY_END(); +}