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

feat(scheduler-allocator-runner): runner plan and claim search (task 3.2)

Browse source 
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) <noreply@anthropic.com>
This commit is contained in:
Rongsong Shen 2026-09-30 09:12:01 +00:00
parent d9ef983014
commit 0c011f784e
Signed by untrusted user: rshen
GPG key ID: 80533536B0EB651C
4 changed files with 1080 additions and 14 deletions
Download patch file Download diff file Expand all files Collapse all files

293
src/scheduler/src/alloc_runner.c
View file

@ -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 <errno.h>
#include <stdlib.h>
#include <string.h>
#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);
}

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

@ -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'])

784
src/scheduler/testsuite/test_alloc_runner_search.c Normal file
View file

@ -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 <errno.h>
#include <stdlib.h>
#include <string.h>
#include <unity/unity.h>
#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();
}