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

feat(scheduler-allocator-runner): runner result assembly (task 3.3)

Browse source 
sched_finalize_job_alloc_() turns a satisfied plan's claims into the job
allocation exactly as parjob's commit tail does (host records from the
claim-time snapshots in claim order, shared records, jobid 0, no
duration) and publishes ctx->alloc only on success. Every out-of-memory
step frees what it built and releases the debits of claims not yet
turned into host allocations; each claim snapshot is freed exactly once.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
This commit is contained in:
Rongsong Shen 2026-09-30 09:27:32 +00:00
parent 0c011f784e
commit 0e68ed991a
Signed by untrusted user: rshen
GPG key ID: 80533536B0EB651C
5 changed files with 666 additions and 49 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -14,6 +14,8 @@
* 4. the shared-resource precheck (Req 2.2-2.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)
* 6. result assembly from the winning plan's claims (Req 4.1-4.4,
* 5.4-5.6)
*
* Returns > 0 when satisfied (ctx->alloc set), 0 when pending, < 0 a
* negative errno. No plugin file or plugin-specific symbol is referenced
@ -30,6 +32,7 @@
#include "sched/types.h"
#include "scheduler.h"
#include "utils.h"
#include "xcealloc.h"
#include "xvec.h"
/* Creates the attempt's pend-reason record unless an earlier primary of
@ -329,14 +332,88 @@ static int sched_alloc_claims_reserve_(sched_alloc_claim_t** claims, int* cap,
return 0;
}
/* Builds the attempt's result from a satisfied plan's claims and
* publishes it on ctx->alloc (Req 4.1-4.4, 5.6): one host allocation per
* claim, in claim order, from the claimed host's name and its request
* snapshot; one shared record per shared-request entry, none when the
* shared request is empty; job id 0, empty reservation, no duration.
*
* Consumes the claims: each snapshot is freed once its host allocation is
* built; on failure the claims not yet converted are rolled back (debit
* released, snapshot freed). Converted claims' debits stay in ctx.
*
* 1, or -ENOMEM with ctx->alloc untouched and every piece built freed
* (Req 5.4, 5.5). */
static int sched_finalize_job_alloc_(xce_scheduler_t* sched,
xce_sched_ctx_t* ctx,
sched_alloc_claim_t* claims, int nclaimed)
{
xce_host_alloc_t** hallocs;
xce_job_alloc_t* local;
int nshared = ctx->alloc_req.shared.count;
int i;
hallocs = (xce_host_alloc_t**)calloc((size_t)nclaimed, sizeof(*hallocs));
if (hallocs == NULL) {
sched_rollback_claims_(ctx, claims, nclaimed);
return -ENOMEM;
}
for (i = 0; i < nclaimed; i++) {
int nres = claims[i].host_req.count;
xce_alloc_t** allocs = sched_make_allocs(
sched, "", nres, (int*)claims[i].host_req.idxs.data,
(float*)claims[i].host_req.vals.data,
/* FIXME: no duration now */ NULL);
if ((allocs != NULL) || (nres <= 0)) {
hallocs[i] = make_host_alloc(claims[i].w->name, nres, allocs);
if (hallocs[i] == NULL) {
free_ResAlloc_list(nres, allocs);
}
}
if (hallocs[i] == NULL) {
sched_rollback_claims_(ctx, claims + i, nclaimed - i);
free_HostAlloc_list(nclaimed, hallocs);
return -ENOMEM;
}
xvec_idval_fini(&(claims[i].host_req));
}
local = make_job_alloc(0, nclaimed, hallocs);
if (local == NULL) {
free_HostAlloc_list(nclaimed, hallocs);
return -ENOMEM;
}
local->allocs_shared = NULL;
local->nres_shared = 0;
if (nshared > 0) {
local->allocs_shared = sched_make_allocs(
sched, "", nshared, (int*)ctx->alloc_req.shared.idxs.data,
(float*)ctx->alloc_req.shared.vals.data, NULL);
if (local->allocs_shared == NULL) {
free_JobAlloc_ptr(&local);
return -ENOMEM;
}
local->nres_shared = nshared;
}
ctx->alloc = local;
return 1;
}
/* 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. */
* Satisfied: the result built from the plan's claims, which the
* assembly consumes (1, or -ENOMEM). No plan satisfied: 0. Claims-buffer
* growth or snapshot failure: that error. No snapshot is freed here. */
static int sched_alloc_search_(const xce_sched_allocator_iface_t* iface,
xce_sched_core_t* core, xce_sched_ctx_t* ctx)
xce_scheduler_t* sched, xce_sched_core_t* core,
xce_sched_ctx_t* ctx)
{
xce_alloc_plan_iter_t iter;
xce_alloc_plan_t* plan;
@ -368,10 +445,7 @@ static int sched_alloc_search_(const xce_sched_allocator_iface_t* iface,
}
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;
rc = sched_finalize_job_alloc_(sched, ctx, claims, n);
}
iface->plan_fini(&iter);
@ -411,5 +485,6 @@ int sched_allocator_run(void* plugin_arg, xce_sched_ctx_t* ctx)
return rc;
}
return sched_alloc_search_(iface, plugin->core, ctx);
return sched_alloc_search_(iface, (xce_scheduler_t*)plugin->sched,
plugin->core, ctx);
}

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

@ -1768,6 +1768,20 @@ sched_alloc_runner_search_test = executable(
link_args: ['-Wl,--wrap=xvec_dup'],
install: false)
# scheduler-allocator-runner task 3.3 (Req 4.1-4.4, 5.4-5.7, 7.2): the
# runner's result assembly and its out-of-memory paths, same white-box
# stack as above. make_host_alloc/make_job_alloc are wrapped to fail on
# call N; the host-array calloc is replaced inside the test file
# (alloc_runner.c only).
sched_alloc_runner_result_test = executable(
'sched_alloc_runner_result_test',
['test_alloc_runner_result.c'],
include_directories:[incdir],
dependencies: [scheduler_test_deps],
link_args: ['-Wl,--wrap=make_host_alloc',
'-Wl,--wrap=make_job_alloc'],
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
@ -2046,6 +2060,9 @@ test('scheduler alloc runner test',
# 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')
# Leak detection stays on: every fixture tears down its ctx and core.
test('scheduler alloc runner result test',
sched_alloc_runner_result_test, suite: 'scheduler')
test('scheduler seqjob rewrap equivalence test',
sched_seqjob_rewrap_equivalence_test, suite: 'scheduler',
env: ['ASAN_OPTIONS=detect_leaks=0'])

544
src/scheduler/testsuite/test_alloc_runner_result.c Normal file
View file

@ -0,0 +1,544 @@
/*
* test_alloc_runner_result.c - scheduler-allocator-runner task 3.3: the
* runner's result assembly (Req 4.1-4.4) and its out-of-memory paths
* (Req 5.4-5.7, 7.2), through sched_allocator_run() with a scripted fake
* plan generator over real workers.
*
* Injectable steps, in the order the runner reaches them:
* HOST_ARRAY calloc() of the host-allocation array: a call-counting
* test_calloc() replacing calloc in alloc_runner.c only
* RECORDS sched_make_allocs() (stubbed here): host 0's records,
* host 1's records, then the shared records
* HOST_ALLOC make_host_alloc() (--wrap): host 0, then host 1
* JOB_ALLOC make_job_alloc() (--wrap)
* Each is armed with the 1-based call number that fails; 0 = never.
*/
#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; calloc() is replaced in the runner only */
static void* test_calloc(size_t n, size_t sz);
#define calloc test_calloc
#include "../src/alloc_runner.c"
#undef calloc
#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;
}
/* ---------------------------------------------------------------------- */
/* Fault injection */
/* ---------------------------------------------------------------------- */
typedef struct {
int calls;
int fail_on; /* 1-based call number; 0 = never */
} inj_t;
enum {
HOST_ARRAY,
RECORDS,
HOST_ALLOC,
JOB_ALLOC,
NSTEPS
};
static inj_t g_inj[NSTEPS];
static int inj_fails(int step)
{
g_inj[step].calls++;
return (g_inj[step].fail_on > 0)
&& (g_inj[step].calls == g_inj[step].fail_on);
}
static void* test_calloc(size_t n, size_t sz)
{
if (inj_fails(HOST_ARRAY)) {
errno = ENOMEM;
return NULL;
}
return calloc(n, sz);
}
/* Stub for scheduler.c's sched_make_allocs(): real make_alloc() records
* (resmgr name, requested value, the given reservation, duration -1 for a
* NULL duration array); NULL for nres <= 0 as the real one. */
xce_alloc_t** sched_make_allocs(xce_scheduler_t* sched, char* reserve, int nres,
int* idxs, float* vals, int* duration)
{
xce_alloc_t** allocs;
int i;
if (inj_fails(RECORDS)) {
errno = ENOMEM;
return NULL;
}
if ((sched == NULL) || (nres <= 0) || (idxs == NULL) || (vals == NULL)) {
errno = EINVAL;
return NULL;
}
allocs = (xce_alloc_t**)calloc(nres, sizeof(*allocs));
TEST_ASSERT_NOT_NULL(allocs);
for (i = 0; i < nres; i++) {
xce_resource_t* r = NULL;
vget(sched->core.resmgr, idxs[i], (void**)&r);
TEST_ASSERT_NOT_NULL(r);
allocs[i] = make_alloc(r->name, reserve, 0, vals[i],
(duration != NULL) ? duration[i] : -1);
TEST_ASSERT_NOT_NULL(allocs[i]);
}
return allocs;
}
xce_host_alloc_t* __real_make_host_alloc(char*, int, xce_alloc_t**);
xce_job_alloc_t* __real_make_job_alloc(uint64_t, int, xce_host_alloc_t**);
xce_host_alloc_t* __wrap_make_host_alloc(char* host, int nalloc,
xce_alloc_t** allocs)
{
if (inj_fails(HOST_ALLOC)) {
errno = ENOMEM;
return NULL;
}
return __real_make_host_alloc(host, nalloc, allocs);
}
xce_job_alloc_t* __wrap_make_job_alloc(uint64_t jobid, int nhost,
xce_host_alloc_t** hallocs)
{
if (inj_fails(JOB_ALLOC)) {
errno = ENOMEM;
return NULL;
}
return __real_make_job_alloc(jobid, nhost, hallocs);
}
/* ---------------------------------------------------------------------- */
/* plugin_prepare_allocation() stub */
/* ---------------------------------------------------------------------- */
static int g_slots_idx = -1;
static int g_mem_idx = -1;
static int g_lic_idx = -1;
static int g_lic2_idx = -1;
static int g_shared_on = 0; /* job wants lic 2 and lic2 5 */
static int g_host_none = 0; /* per-host request left empty */
/* Job-level pass: the shared wants. Per-host pass: mem 10 per slot, then
* the bucket's slot count. */
int plugin_prepare_allocation(xce_sched_core_t* sched, xce_sched_ctx_t* ctx)
{
(void)sched;
xvec_idval_fini(&(ctx->alloc_req.host));
xvec_idval_init(&(ctx->alloc_req.host), 8);
if (ctx->worker == NULL) {
xvec_idval_fini(&(ctx->alloc_req.shared));
xvec_idval_init(&(ctx->alloc_req.shared), 8);
if (g_shared_on) {
TEST_ASSERT_TRUE(
xvec_idval_store(&(ctx->alloc_req.shared), g_lic_idx, 2.0f)
> 0);
TEST_ASSERT_TRUE(
xvec_idval_store(&(ctx->alloc_req.shared), g_lic2_idx, 5.0f)
> 0);
}
return 0;
}
if (g_host_none) {
return 0;
}
TEST_ASSERT_TRUE(xvec_idval_store(&(ctx->alloc_req.host), g_mem_idx,
10.0f * (float)ctx->nslots)
> 0);
TEST_ASSERT_TRUE(xvec_idval_store(&(ctx->alloc_req.host), g_slots_idx,
(float)ctx->nslots)
> 0);
return 0;
}
/* ---------------------------------------------------------------------- */
/* Scripted fake plan generator: plan [3,1], or none */
/* ---------------------------------------------------------------------- */
static int g_no_plan = 0;
static int g_next_calls = 0;
static int g_fini_calls = 0;
static int g_slots[2] = { 3, 1 };
static xce_alloc_plan_t g_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;
iter->private = NULL;
return 0;
}
static xce_alloc_plan_t* fake_plan_next(xce_alloc_plan_iter_t* iter)
{
(void)iter;
if (g_no_plan || (g_next_calls++ > 0)) {
return NULL;
}
memset(&g_plan, 0, sizeof(g_plan));
g_plan.nbuckets = 2;
g_plan.bucket_slots = g_slots;
return &g_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,
};
void setUp(void)
{
memset(g_inj, 0, sizeof(g_inj));
g_shared_on = 0;
g_host_none = 0;
g_no_plan = 0;
g_next_calls = 0;
g_fini_calls = 0;
}
void tearDown(void) { }
/* ---------------------------------------------------------------------- */
/* Fixture: hosts ra-a and ra-b (4 slots, 100 mem), lic/lic2 10 each */
/* ---------------------------------------------------------------------- */
typedef struct {
xce_scheduler_t sched;
xce_resource_t * slots, *mem, *lic, *lic2;
xce_sched_worker_t * a, *b;
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;
} rr_fixture_t;
static xce_resource_t* rr_res(rr_fixture_t* f, const char* name, int flags)
{
xce_resource_t* r = makeResource(
f->sched.core.resmgr,
&(xce_resource_t) { .name = (char*)name, .flags = flags });
TEST_ASSERT_NOT_NULL(r);
return r;
}
static xce_sched_worker_t* rr_add_host(rr_fixture_t* f, const char* name)
{
xce_sched_worker_t* w;
int idx[2] = { f->slots->idx, f->mem->idx };
float total[2] = { 4.0f, 100.0f };
float inuse[2] = { 0.0f, 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), 2, idx, total);
put_resvals(&(w->resInUse), 2, idx, inuse);
TEST_ASSERT_TRUE(sched_add_worker(&(f->sched.core), w) >= 0);
return w;
}
static void rr_fixture_init(rr_fixture_t* f)
{
int i;
int idx[2];
float total[2] = { 10.0f, 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 = rr_res(f, "slots", 0);
f->mem = rr_res(f, "mem", 0);
f->lic = rr_res(f, "lic", XCE_RESF_GLOBAL);
f->lic2 = rr_res(f, "lic2", XCE_RESF_GLOBAL);
g_slots_idx = f->slots->idx;
g_mem_idx = f->mem->idx;
g_lic_idx = f->lic->idx;
g_lic2_idx = f->lic2->idx;
idx[0] = f->lic->idx;
idx[1] = f->lic2->idx;
put_resvals(&(f->sched.core.sharedInst), 2, idx, total);
f->a = rr_add_host(f, "ra-a");
f->b = rr_add_host(f, "ra-b");
f->plugin.name = (char*)"fake_allocator";
f->plugin.sched = &(f->sched);
f->plugin.core = &(f->sched.core);
f->plugin.allocator_iface = &g_iface;
f->req.minTasks = 1;
f->req.maxTasks = 4;
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 rr_fixture_fini(rr_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));
xstring_table_fini();
}
static int rr_run(rr_fixture_t* f)
{
return sched_allocator_run(&(f->plugin), &(f->ctx));
}
static float rr_debit(rr_fixture_t* f, xce_sched_worker_t* w, xce_resource_t* r)
{
xvec_t* v = (xvec_t*)zarray_get(&(f->ctx.uncomitted.hosts), w->idx);
return (v == NULL) ? 0.0f : ((float*)v->data)[r->idx];
}
/* Host debit kept (the claim's request) or released (0) */
static void rr_assert_debit(rr_fixture_t* f, xce_sched_worker_t* w, float slots)
{
TEST_ASSERT_EQUAL_FLOAT(10.0f * slots, rr_debit(f, w, f->mem));
TEST_ASSERT_EQUAL_FLOAT(slots, rr_debit(f, w, f->slots));
}
static void rr_assert_shared_debit(rr_fixture_t* f)
{
float* v = (float*)f->ctx.uncomitted.shared->data;
TEST_ASSERT_EQUAL_FLOAT(2.0f, v[f->lic->idx]);
TEST_ASSERT_EQUAL_FLOAT(5.0f, v[f->lic2->idx]);
}
static void rr_assert_record(xce_alloc_t* a, const char* resname, float value)
{
TEST_ASSERT_NOT_NULL(a);
TEST_ASSERT_EQUAL_STRING(resname, a->resname);
TEST_ASSERT_EQUAL_STRING("", a->reserve);
TEST_ASSERT_EQUAL_FLOAT(value, a->value);
TEST_ASSERT_EQUAL_INT(-1, a->duration);
}
static void rr_assert_host(xce_host_alloc_t* ha, const char* host, float slots)
{
TEST_ASSERT_NOT_NULL(ha);
TEST_ASSERT_EQUAL_STRING(host, ha->host);
TEST_ASSERT_EQUAL_INT(2, ha->nallocs);
TEST_ASSERT_NOT_NULL(ha->allocs);
rr_assert_record(ha->allocs[0], "mem", 10.0f * slots);
rr_assert_record(ha->allocs[1], "slots", slots);
}
/* ---------------------------------------------------------------------- */
/* Success (Req 4.1-4.4, 5.6) */
/* ---------------------------------------------------------------------- */
/* Plan [3,1]: bucket 1 on ra-a, bucket 2 on ra-b; two shared records. */
static void test_success_builds_hosts_and_shared_records(void)
{
rr_fixture_t f;
xce_job_alloc_t* ja;
g_shared_on = 1;
rr_fixture_init(&f);
TEST_ASSERT_EQUAL_INT(1, rr_run(&f));
ja = f.ctx.alloc;
TEST_ASSERT_NOT_NULL(ja);
TEST_ASSERT_TRUE(ja->jobid == 0);
TEST_ASSERT_EQUAL_INT(2, ja->nhosts);
rr_assert_host(ja->allocs[0], "ra-a", 3.0f);
rr_assert_host(ja->allocs[1], "ra-b", 1.0f);
TEST_ASSERT_EQUAL_INT(2, ja->nres_shared);
TEST_ASSERT_NOT_NULL(ja->allocs_shared);
rr_assert_record(ja->allocs_shared[0], "lic", 2.0f);
rr_assert_record(ja->allocs_shared[1], "lic2", 5.0f);
TEST_ASSERT_EQUAL_INT(3, g_inj[RECORDS].calls);
TEST_ASSERT_EQUAL_INT(1, g_inj[HOST_ARRAY].calls);
/* debits stay with the attempt context */
rr_assert_debit(&f, f.a, 3.0f);
rr_assert_debit(&f, f.b, 1.0f);
rr_assert_shared_debit(&f);
TEST_ASSERT_EQUAL_INT(1, g_fini_calls);
rr_fixture_fini(&f);
}
/* No shared request: no shared record, no shared records call. */
static void test_success_without_shared_request(void)
{
rr_fixture_t f;
rr_fixture_init(&f);
TEST_ASSERT_EQUAL_INT(1, rr_run(&f));
TEST_ASSERT_NOT_NULL(f.ctx.alloc);
TEST_ASSERT_EQUAL_INT(2, f.ctx.alloc->nhosts);
TEST_ASSERT_NULL(f.ctx.alloc->allocs_shared);
TEST_ASSERT_EQUAL_INT(0, f.ctx.alloc->nres_shared);
TEST_ASSERT_EQUAL_INT(2, g_inj[RECORDS].calls);
rr_fixture_fini(&f);
}
/* An empty per-host request builds a host allocation with no record. */
static void test_success_with_empty_host_request(void)
{
rr_fixture_t f;
g_host_none = 1;
rr_fixture_init(&f);
TEST_ASSERT_EQUAL_INT(1, rr_run(&f));
TEST_ASSERT_NOT_NULL(f.ctx.alloc);
TEST_ASSERT_EQUAL_INT(2, f.ctx.alloc->nhosts);
TEST_ASSERT_EQUAL_STRING("ra-a", f.ctx.alloc->allocs[0]->host);
TEST_ASSERT_EQUAL_INT(0, f.ctx.alloc->allocs[0]->nallocs);
TEST_ASSERT_NULL(f.ctx.alloc->allocs[0]->allocs);
TEST_ASSERT_EQUAL_STRING("ra-b", f.ctx.alloc->allocs[1]->host);
rr_fixture_fini(&f);
}
/* No plan satisfied: pending, nothing on the attempt. */
static void test_pending_leaves_no_alloc(void)
{
rr_fixture_t f;
g_no_plan = 1;
rr_fixture_init(&f);
TEST_ASSERT_EQUAL_INT(0, rr_run(&f));
TEST_ASSERT_NULL(f.ctx.alloc);
TEST_ASSERT_EQUAL_INT(0, g_inj[HOST_ARRAY].calls);
TEST_ASSERT_EQUAL_INT(1, g_fini_calls);
rr_fixture_fini(&f);
}
/* ---------------------------------------------------------------------- */
/* Out of memory (Req 5.4-5.7, 7.2) */
/* ---------------------------------------------------------------------- */
/* Fails `step` at call `nth` (shared request on): -ENOMEM, no allocation
* on the attempt, generator finished once; ra-a / ra-b keep their debit
* only if their claim became a host allocation (or the failure came after
* all of them); the shared amounts stay counted. Leaks and double frees
* are ASAN's to report. */
static void rr_check_oom(int step, int nth, int keep_a, int keep_b)
{
rr_fixture_t f;
setUp();
g_shared_on = 1;
rr_fixture_init(&f);
g_inj[step].fail_on = nth;
TEST_ASSERT_EQUAL_INT(-ENOMEM, rr_run(&f));
TEST_ASSERT_EQUAL_INT(nth, g_inj[step].calls);
TEST_ASSERT_NULL(f.ctx.alloc);
TEST_ASSERT_EQUAL_INT(1, g_fini_calls);
rr_assert_debit(&f, f.a, keep_a ? 3.0f : 0.0f);
rr_assert_debit(&f, f.b, keep_b ? 1.0f : 0.0f);
rr_assert_shared_debit(&f);
rr_fixture_fini(&f);
}
static void test_oom_host_array(void) { rr_check_oom(HOST_ARRAY, 1, 0, 0); }
static void test_oom_host0_records(void) { rr_check_oom(RECORDS, 1, 0, 0); }
static void test_oom_host1_records(void) { rr_check_oom(RECORDS, 2, 1, 0); }
static void test_oom_host0_host_alloc(void)
{
rr_check_oom(HOST_ALLOC, 1, 0, 0);
}
static void test_oom_host1_host_alloc(void)
{
rr_check_oom(HOST_ALLOC, 2, 1, 0);
}
static void test_oom_job_alloc(void) { rr_check_oom(JOB_ALLOC, 1, 1, 1); }
static void test_oom_shared_records(void) { rr_check_oom(RECORDS, 3, 1, 1); }
int main(void)
{
UNITY_BEGIN();
RUN_TEST(test_success_builds_hosts_and_shared_records);
RUN_TEST(test_success_without_shared_request);
RUN_TEST(test_success_with_empty_host_request);
RUN_TEST(test_pending_leaves_no_alloc);
RUN_TEST(test_oom_host_array);
RUN_TEST(test_oom_host0_records);
RUN_TEST(test_oom_host1_records);
RUN_TEST(test_oom_host0_host_alloc);
RUN_TEST(test_oom_host1_host_alloc);
RUN_TEST(test_oom_job_alloc);
RUN_TEST(test_oom_shared_records);
return UNITY_END();
}

59
src/scheduler/testsuite/test_alloc_runner_search.c
View file

@ -56,14 +56,25 @@ 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. */
/* Records for a satisfied plan's result (asserted in
* test_alloc_runner_result.c): resmgr name and requested value. */
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;
xce_alloc_t** allocs = (xce_alloc_t**)calloc(nres, sizeof(*allocs));
int i;
(void)duration;
errno = ENOSYS;
return NULL;
TEST_ASSERT_NOT_NULL(allocs);
for (i = 0; i < nres; i++) {
xce_resource_t* r = NULL;
vget(sched->core.resmgr, idxs[i], (void**)&r);
TEST_ASSERT_NOT_NULL(r);
allocs[i] = make_alloc(r->name, reserve, 0, vals[i], -1);
TEST_ASSERT_NOT_NULL(allocs[i]);
}
return allocs;
}
/* ---------------------------------------------------------------------- */
@ -402,7 +413,7 @@ static void test_plans_in_order_first_satisfied_wins(void)
script_plan(2, 2, 2, 0);
script_plan(4, 1, 1, 1);
TEST_ASSERT_EQUAL_INT(0, sr_run(&f));
TEST_ASSERT_EQUAL_INT(1, 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);
@ -539,14 +550,14 @@ static void test_rollback_then_later_plan_with_bucket_counts(void)
script_plan(3, 1, 1, 1);
script_plan(1, 2, 0, 0);
TEST_ASSERT_EQUAL_INT(0, sr_run(&f));
TEST_ASSERT_EQUAL_INT(1, 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(2.0f, sr_slots_debit(&f, a)); /* [2] won */
TEST_ASSERT_EQUAL_FLOAT(0.0f, sr_slots_debit(&f, b));
TEST_ASSERT_EQUAL_INT(1, g_fini_calls);
sr_fixture_fini(&f);
@ -605,7 +616,7 @@ static void test_malformed_plans_skipped(void)
script_plan(2, -1, 2, 0);
script_plan(1, 2, 0, 0);
TEST_ASSERT_EQUAL_INT(0, sr_run(&f));
TEST_ASSERT_EQUAL_INT(1, sr_run(&f));
TEST_ASSERT_EQUAL_INT(4, g_next_calls);
TEST_ASSERT_EQUAL_INT(1, g_nlog);
sr_assert_log(0, a, 2);
@ -628,7 +639,7 @@ static void test_claims_buffer_grows_to_larger_plan(void)
script_plan(1, 3, 0, 0);
script_plan(3, 1, 1, 1);
TEST_ASSERT_EQUAL_INT(0, sr_run(&f));
TEST_ASSERT_EQUAL_INT(1, 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);
@ -735,35 +746,6 @@ static void test_try_plan_snapshot_failure_reports_no_claim(void)
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();
@ -779,6 +761,5 @@ int main(void)
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();
}