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

feat(scheduler-allocator-runner): parjob allocates through the framework runner (task 4.1)

Browse source 
parjob registers sched_allocator_run with parjob_alloc_iface (accepts
maxTasks > 1, bug-1767, plus its plan iterator); its private my_alloc()
and claim helpers are gone. Characterization now pins the intended
changes: a reason for every short host (2/6/5) and the bug-1893 shared
reason. No other test expectation changed.

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

151
src/scheduler/testsuite/test_alloc_equivalence_parjob.c
View file

@ -1,21 +1,8 @@
/*
* test_alloc_equivalence_parjob.c - scheduler-allocator-runner task 2.2:
* characterization of the parallel-job allocator (Req 4.5, 5.1, 5.2, 5.4,
* 5.8, 7.2) through the registered plugin->job_alloc_fn, never the static
* routine by name; must still hold once task 4.1 switches parjob to the
* framework runner. Stubs and OOM injection are those of
* test_alloc_equivalence_seqjob.c: sched_make_allocs() failing on call N
* (claimed hosts' records in claim order, then shared records) and
* --wrap'ed make_host_alloc/make_job_alloc failing on call N; the
* host-array allocation is not injected (task 3.3).
*
* Values the switch changes on purpose sit in `_today` tests: bug-1893
* (Req 2.2) and host reasons (Req 3.6/3.7). Today the host request is
* reset once per bucket, so later short candidates are pre-filtered
* (host_iter.c) against the request the first one left and get no reason;
* the runner clears it before every candidate. Stable tests assert only
* reason entries both rules give. bug-1892 layouts are kept out.
*/
/* test_alloc_equivalence_parjob.c - scheduler-allocator-runner 2.2/4.1:
* parjob through plugin->job_alloc_fn only (Req 2.2, 3.6, 3.7, 4.5, 5.1,
* 5.2, 5.4, 5.8, 7.2). OOM injection as test_alloc_equivalence_seqjob.c:
* sched_make_allocs() (hosts in claim order, then shared) and --wrap'ed
* make_host_alloc/make_job_alloc fail on call N. No bug-1892 layouts. */
#include <errno.h>
#include <stdlib.h>
#include <string.h>
@ -30,21 +17,19 @@
#include "xcealloc.h"
#include "xstring.h"
/* White-box sources under test. Task 4.1 adds the framework runner
* ("../src/alloc_runner.c") as one more line of this list. */
/* White-box sources under test. */
#include "../src/core.c"
#include "../src/resource.c"
#include "../src/schedobj.c"
#include "../src/host_iter.c"
#include "../src/alloc_runner.c"
#include "../src/plugins/parjob_allocator.c"
/* Link-satisfying stubs (same convention as test_seqjob_allocator.c). */
int plugin_register(void* s, void* p) { return (void)s, (void)p, 0; }
void plugin_unregister(void* arg) { (void)arg; }
/* plugin.c's fan-out with parjob's hook only: reset the request vectors as
* plugin.c does (job-level pass: .shared and .host, bug-1883; per-host
* pass: .host), then call the hook. */
/* plugin.c's fan-out, parjob's hook only; resets as plugin.c (bug-1883). */
int plugin_prepare_allocation(xce_sched_core_t* sched, xce_sched_ctx_t* ctx)
{
xce_sched_plugin_t fake_plugin;
@ -87,9 +72,8 @@ static struct {
static void oom_injection_reset(void) { memset(&g, 0, sizeof(g)); }
/* Stub for scheduler.c's sched_make_allocs(): real make_alloc() records
* (resmgr name, requested value, empty reservation, duration -1 for a NULL
* duration array, instid 0); NULL for nres <= 0 as the real one. */
/* scheduler.c's sched_make_allocs(): real make_alloc() records (duration -1
* for a NULL duration array, instid 0); 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)
@ -173,6 +157,7 @@ typedef struct {
xce_resource_t* mem;
xce_resource_t* rack;
xce_resource_t* lic; /* shared (XCE_RESF_GLOBAL) */
xce_resource_t* tok; /* shared, pool 10, not requested by eq_run */
xce_sched_plugin_t* plugin;
xce_sched_obj_t obj;
xce_primary_schedobj_t primary_obj;
@ -190,11 +175,11 @@ typedef struct {
xce_select_node_t sel_res, sel_num, sel_eq;
} eq_primary_t;
/* Shared pool "licenses" has `lic_total` capacity, none in use. */
/* Shared pools "licenses" (`lic_total`) and "tokens" (10), none in use. */
static void eq_fixture_init(eq_fixture_t* f, float lic_total)
{
int idx[1];
float val[1];
int idx[2];
float val[2];
memset(f, 0, sizeof(*f));
TEST_ASSERT_EQUAL_INT(0, xce_sched_core_init(&(f->sched.core)));
@ -213,15 +198,19 @@ static void eq_fixture_init(eq_fixture_t* f, float lic_total)
f->mem = EQ_RES("mem", 0);
f->rack = EQ_RES("rack", 0);
f->lic = EQ_RES("licenses", XCE_RESF_GLOBAL);
f->tok = EQ_RES("tokens", XCE_RESF_GLOBAL);
#undef EQ_RES
TEST_ASSERT_NOT_NULL(f->slots);
TEST_ASSERT_NOT_NULL(f->mem);
TEST_ASSERT_NOT_NULL(f->rack);
TEST_ASSERT_NOT_NULL(f->lic);
TEST_ASSERT_NOT_NULL(f->tok);
idx[0] = f->lic->idx;
val[0] = lic_total;
put_resvals(&(f->sched.core.sharedInst), 1, idx, val);
idx[1] = f->tok->idx;
val[1] = 10.0f;
put_resvals(&(f->sched.core.sharedInst), 2, idx, val);
f->plugin = parjob_allocator_new_plugin();
TEST_ASSERT_NOT_NULL(f->plugin);
@ -286,9 +275,8 @@ static void eq_primary_init(eq_primary_t* p, int max_tasks, float mem,
}
}
/* ctx over a PRIMARY obj for `p`, job-level prepare pass included
* (sched_ctx_set_primary() -> plugin_prepare_allocation(), ctx->worker
* NULL), exactly as sched_make_primary_allocation() drives it. */
/* ctx over a PRIMARY obj for `p`, job-level prepare pass included, exactly
* as sched_make_primary_allocation() drives it. */
static void eq_ctx_begin_primary(eq_fixture_t* f, eq_primary_t* p)
{
f->primary_obj.resreq = &(p->req);
@ -358,8 +346,7 @@ static void eq_assert_record(xce_alloc_t* a, const char* resname, float value)
TEST_ASSERT_EQUAL_INT(-1, a->duration);
}
/* One host allocation carrying mem then slots (parjob's per-host prepare
* computes the rusage wants first, then stores the bucket's slot count). */
/* One host allocation carrying mem then the bucket's slot count. */
static void eq_assert_host(xce_host_alloc_t* ha, const char* host, float mem,
float slots)
{
@ -399,9 +386,8 @@ static void eq_assert_calls(int make_allocs, int host_alloc, int job_alloc)
TEST_ASSERT_EQUAL_INT(job_alloc, g.job_alloc_calls);
}
/* A fixture: `pool` shared licenses; hosts A, B, C (rack 1) with the given
* slots and mem; a job of `tasks` tasks, `task_mem` mem per task and `lic`
* licenses per job. */
/* `pool` licenses; hosts A, B, C (rack 1) with the given slots and mem; a
* job of `tasks` tasks, `task_mem` mem per task, `lic` licenses per job. */
typedef struct {
float pool;
int nhosts;
@ -441,8 +427,7 @@ static int eq_run(eq_fixture_t* f, eq_primary_t* p, const eq_spec_t* s,
return eq_alloc(f);
}
/* Main and host reasons none; at least `n` resource reasons, the first `n`
* of which both rules give (count asserted only in `_today` tests). */
/* Main and host reasons none; at least `n` resource reasons. */
static void eq_assert_reason_prefix(eq_fixture_t* f, int n)
{
TEST_ASSERT_NOT_NULL(f->ctx.reason);
@ -451,7 +436,7 @@ static void eq_assert_reason_prefix(eq_fixture_t* f, int n)
TEST_ASSERT_TRUE(f->ctx.reason->reasons->nres_reasons >= n);
}
/* EQ_TWO_BUCKETS: plan [4] checks A first (slots short) under both rules. */
/* EQ_TWO_BUCKETS: plan [4] checks A first (slots short). */
static void eq_assert_plan1_reasons(eq_fixture_t* f)
{
eq_assert_reason_prefix(f, 1);
@ -508,18 +493,24 @@ static void test_parjob_pending_on_short_shared_resource(void)
eq_finish(&f, &p);
}
/* 2, resource reasons, TODAY's behaviour: none. The reason loop skips
* every unmet entry because vget() returns 0 on success and the loop
* tests `rc <= 0` (bug-1893). Requirement 2.2 changes this in task 4.1
* (one host-less NO_RESOURCE reason naming "licenses"). */
static void test_parjob_pending_on_short_shared_no_reason_today(void)
/* 2, resource reasons (Req 2.2, bug-1893): two shared resources requested,
* only "licenses" short: exactly one host-less reason, naming it. */
static void test_parjob_pending_on_short_shared_names_short_one(void)
{
eq_fixture_t f;
eq_primary_t p;
xce_sched_worker_t* w[3];
rusage_value_t tok = { .res = (char*)"tokens", .value = 1.0f };
TEST_ASSERT_EQUAL_INT(0, eq_run(&f, &p, &EQ_SHORT_SHARED, w));
eq_assert_reason_counts(&f, 0, 0, 0);
eq_fixture_init(&f, EQ_SHORT_SHARED.pool);
eq_add_host(&f, HOST_A, 4.0f, 8.0f, 1.0f);
eq_primary_init(&p, 2, 1.0f, 3.0f, &f, 0);
tok.per_flag = PER_JOB;
xlist_store(&(p.req.rusages), &tok);
eq_ctx_begin_primary(&f, &p);
TEST_ASSERT_EQUAL_INT(0, eq_alloc(&f));
eq_assert_reason_counts(&f, 0, 0, 1);
eq_assert_res_reason(&f, 0, "", "licenses"); /* host-less: "" */
eq_finish(&f, &p);
}
@ -535,7 +526,7 @@ static void test_parjob_pending_on_hosts(void)
TEST_ASSERT_EQUAL_INT(0, eq_run(&f, &p, &EQ_SHORT_HOSTS, w));
TEST_ASSERT_NULL(f.ctx.alloc);
/* plan [2]'s first candidate A, under both rules */
/* plan [2]'s first candidate A */
eq_assert_reason_prefix(&f, 2);
eq_assert_res_reason(&f, 0, HOST_A, "mem");
eq_assert_res_reason(&f, 1, HOST_A, "slots");
@ -571,8 +562,7 @@ static void test_parjob_mid_plan_rollback_then_later_plan_wins(void)
eq_assert_debit(&f, w[2], 1.0f, 1.0f);
TEST_ASSERT_EQUAL_FLOAT(3.0f, eq_shared_uncomit(&f));
/* A then B, slots: plan [4]'s first two checks (new rule), or plan
* [4]'s A and plan [2,2] bucket 2's B (today) */
/* A then B, slots: plan [4]'s first two checks */
eq_assert_reason_prefix(&f, 2);
eq_assert_res_reason(&f, 0, HOST_A, "slots");
eq_assert_res_reason(&f, 1, HOST_B, "slots");
@ -581,28 +571,40 @@ static void test_parjob_mid_plan_rollback_then_later_plan_wins(void)
eq_finish(&f, &p);
}
/* 1, 3, 4: host reasons, TODAY's behaviour (stale pre-filter): only each
* bucket search's first short candidate gets reasons. Req 3.6/3.7 change
* this in task 4.1 to every selection-matching unclaimed short host:
* 1 -> (A,slots) (B,slots); 3 -> A mem, A slots, B mem, B slots, A mem,
* B mem; 4 -> A, B, C (plan [4]) then B, C (plan [2,2] bucket 2), slots. */
static void test_parjob_host_reasons_first_candidate_only_today(void)
static void eq_assert_reasons(eq_fixture_t* f, int n, const char** pairs)
{
int i;
eq_assert_reason_counts(f, 0, 0, n);
for (i = 0; i < n; i++) {
eq_assert_res_reason(f, i, pairs[2 * i], pairs[(2 * i) + 1]);
}
}
/* 1, 3, 4: host reasons (Req 3.6/3.7): exactly one per short resource of
* every selection-matching unclaimed short host of each bucket search. */
static void test_parjob_host_reasons_every_short_host(void)
{
static const char* two[] = { HOST_A, "slots", HOST_B, "slots" };
static const char* hosts[]
= { HOST_A, "mem", HOST_A, "slots", HOST_B, "mem",
HOST_B, "slots", HOST_A, "mem", HOST_B, "mem" };
static const char* back[] = { HOST_A, "slots", HOST_B, "slots", HOST_C,
"slots", HOST_B, "slots", HOST_C, "slots" };
eq_fixture_t f;
eq_primary_t p;
xce_sched_worker_t* w[3];
TEST_ASSERT_EQUAL_INT(1, eq_run(&f, &p, &EQ_TWO_BUCKETS, w));
eq_assert_reason_counts(&f, 0, 0, 1);
eq_assert_reasons(&f, 2, two);
eq_finish(&f, &p);
TEST_ASSERT_EQUAL_INT(0, eq_run(&f, &p, &EQ_SHORT_HOSTS, w));
eq_assert_reason_counts(&f, 0, 0, 3);
eq_assert_res_reason(&f, 2, HOST_A, "mem");
eq_assert_reasons(&f, 6, hosts);
eq_finish(&f, &p);
TEST_ASSERT_EQUAL_INT(1, eq_run(&f, &p, &EQ_ROLLBACK, w));
eq_assert_reason_counts(&f, 0, 0, 2);
eq_assert_reasons(&f, 5, back);
eq_finish(&f, &p);
}
@ -664,14 +666,12 @@ static void test_parjob_missing_input_einval_no_reason(void)
}
}
/* 7. Out of memory while building the result of EQ_TWO_BUCKETS (claims A
* then B). Host stage: the claims not yet turned into a host allocation
* are released, converted ones keep their debit; job allocation and shared
* records: every debit stays (design Error table; today's parjob too).
* `a_kept` / `b_kept`: whether A's / B's debit (2 mem, 2 slots) remains.
* Cases: records of host 0 / host 1 (sched_make_allocs() call 1 / 2),
* make_host_alloc() for host 0 / host 1, make_job_alloc(), shared records
* (sched_make_allocs() call 3, after both hosts). */
/* 7. OOM building EQ_TWO_BUCKETS' result (claims A then B). Host stage:
* unconverted claims released, converted ones keep their debit; job alloc
* and shared records: every debit stays (design Error table). `a_kept` /
* `b_kept`: whether A's / B's debit (2 mem, 2 slots) remains. Steps: host
* 0/1 records (sched_make_allocs() 1/2), make_host_alloc() 1/2,
* make_job_alloc(), shared records (sched_make_allocs() 3). */
static void eq_check_oom(int make_allocs_fail_on, int host_alloc_fail_on,
int job_alloc_fail_on, int a_kept, int b_kept,
int make_allocs, int host_alloc, int job_alloc)
@ -712,10 +712,9 @@ EQ_OOM_TEST(test_parjob_oom_job_alloc_debits_stay,
EQ_OOM_TEST(test_parjob_oom_shared_records_debits_stay,
3, 0, 0, 1, 1, 3, 2, 1)
/* 8. COMPOUND: two primaries allocated in turn on one ctx, as
* sched_make_compound_alloc_() does; rack 1 / rack 2 selects keep them on
* distinct hosts (bug-1892 not exercised). Each is sized from its own
* request (Req 5.8): 2 x 1 mem + 3 licenses on A, 3 x 2 mem on B. */
/* 8. COMPOUND: two primaries in turn on one ctx (sched_make_compound_alloc_);
* rack 1 / rack 2 selects keep them apart. Each sized from its own request
* (Req 5.8): 2 x 1 mem + 3 licenses on A, 3 x 2 mem on B. */
static void test_parjob_compound_second_primary_sized_from_own_request(void)
{
eq_fixture_t f;
@ -782,10 +781,10 @@ int main(void)
UNITY_BEGIN();
RUN_TEST(test_parjob_success_two_buckets_with_shared);
RUN_TEST(test_parjob_pending_on_short_shared_resource);
RUN_TEST(test_parjob_pending_on_short_shared_no_reason_today);
RUN_TEST(test_parjob_pending_on_short_shared_names_short_one);
RUN_TEST(test_parjob_pending_on_hosts);
RUN_TEST(test_parjob_mid_plan_rollback_then_later_plan_wins);
RUN_TEST(test_parjob_host_reasons_first_candidate_only_today);
RUN_TEST(test_parjob_host_reasons_every_short_host);
RUN_TEST(test_parjob_task_count_zero_and_one_pend_without_reason);
RUN_TEST(test_parjob_missing_input_einval_no_reason);
RUN_TEST(test_parjob_oom_records_host0_releases_all);

41
src/scheduler/testsuite/test_parjob_combinatorial.c
View file

@ -90,6 +90,7 @@
* to reach them directly.
*/
#include "../src/core.c"
#include "../src/alloc_runner.c"
#include "../src/host_iter.c"
#include "../src/plugins/parjob_allocator.c"
#include "../src/resource.c"
@ -313,7 +314,8 @@ static void init_primary_obj(xce_sched_obj_t* obj,
* attempt runs: sched_ctx_init() -> sched_ctx_set_primary() (the real
* call that stages ctx->alloc_req.shared via plugin_prepare_allocation()'s
* ctx->worker==NULL branch, exactly as sched_make_primary_allocation()
* does today) -> my_alloc(). Returns my_alloc()'s return code; ctx is left
* does today) -> parjob's registered entry point, the framework routine
* sched_allocator_run(). Returns its return code; ctx is left
* initialized (caller must sched_ctx_fini()). Identical to
* test_seqjob_rewrap_equivalence.c's own helper. */
static int drive_my_alloc(xce_scheduler_t* sched, xce_sched_obj_t* obj,
@ -328,13 +330,12 @@ static int drive_my_alloc(xce_scheduler_t* sched, xce_sched_obj_t* obj,
memset(&plugin, 0, sizeof(plugin));
plugin.sched = sched;
/* scheduler-sim-live-allocation spec, task 4.3: my_alloc() now reads
* plugin->core (honestly typed, task 1.2) for every core-safe call,
* alongside the still-needed plugin->sched (sched_make_allocs(),
* scheduler.c, unchanged). */
plugin.core = &(sched->core);
/* the runner reads plugin->core, plugin->sched (sched_make_allocs())
* and parjob's allocator contract */
plugin.core = &(sched->core);
plugin.allocator_iface = &parjob_alloc_iface;
return my_alloc(&plugin, ctx);
return sched_allocator_run(&plugin, ctx);
}
/* assert_uncommitted_host_debit - reads ctx->uncomitted.hosts[w->idx] for
@ -824,7 +825,7 @@ static void test_descending_bucket_order_succeeds_where_ascending_fails(void)
xce_alloc_plan_t plan;
int bucket_slots_desc[2];
int bucket_slots_asc[2];
parjob_claim_t claimed[2];
sched_alloc_claim_t claimed[2];
int nclaimed;
int rc;
int idx[1];
@ -866,9 +867,9 @@ static void test_descending_bucket_order_succeeds_where_ascending_fails(void)
TEST_ASSERT_EQUAL_INT(0, sched_ctx_init(&ctx, &(sched.core), &obj));
sched_ctx_set_primary(&ctx, &primary);
/* my_alloc() always lazily creates ctx->reason before ever reaching
* parjob_try_plan_() (see my_alloc()'s own top-of-function guard);
* this test calls parjob_try_plan_() directly, bypassing that step,
/* sched_allocator_run() always creates ctx->reason before ever
* reaching sched_try_plan_() (alloc_runner.c); this test calls
* sched_try_plan_() directly, bypassing that step,
* so it must set one up itself -- sched_try_alloc_on_host_()
* (resource.c) unconditionally records a pend reason into
* ctx->reason->reasons on a resource-unmet candidate, which the
@ -887,11 +888,9 @@ static void test_descending_bucket_order_succeeds_where_ascending_fails(void)
plan.nbuckets = 2;
plan.bucket_slots = bucket_slots_desc;
/* scheduler-sim-live-allocation spec, task 4.3: parjob_try_plan_() is
* now honestly typed xce_sched_core_t* (mirrors task 2.1's identical
* ctx->sched ripple immediately above) -- pass &(sched.core), not
* &sched. */
rc = parjob_try_plan_(&(sched.core), &ctx, &plan, claimed, &nclaimed);
/* sched_try_plan_() (alloc_runner.c) takes the core: pass
* &(sched.core), not &sched. */
rc = sched_try_plan_(&(sched.core), &ctx, &plan, claimed, &nclaimed);
TEST_ASSERT_EQUAL_INT(1, rc);
TEST_ASSERT_EQUAL_INT(2, nclaimed);
TEST_ASSERT_TRUE(claimed[0].w == host_first);
@ -899,7 +898,7 @@ static void test_descending_bucket_order_succeeds_where_ascending_fails(void)
/* Roll back before trying the ascending variant against the exact
* same, otherwise-untouched host pool. */
parjob_rollback_claims_(&ctx, claimed, nclaimed);
sched_rollback_claims_(&ctx, claimed, nclaimed);
assert_uncommitted_host_debit(&ctx, host_first, slots_res->idx, 0.0f);
assert_uncommitted_host_debit(&ctx, host_second, slots_res->idx, 0.0f);
@ -914,11 +913,9 @@ static void test_descending_bucket_order_succeeds_where_ascending_fails(void)
plan.nbuckets = 2;
plan.bucket_slots = bucket_slots_asc;
/* scheduler-sim-live-allocation spec, task 4.3: parjob_try_plan_() is
* now honestly typed xce_sched_core_t* (mirrors task 2.1's identical
* ctx->sched ripple immediately above) -- pass &(sched.core), not
* &sched. */
rc = parjob_try_plan_(&(sched.core), &ctx, &plan, claimed, &nclaimed);
/* sched_try_plan_() (alloc_runner.c) takes the core: pass
* &(sched.core), not &sched. */
rc = sched_try_plan_(&(sched.core), &ctx, &plan, claimed, &nclaimed);
TEST_ASSERT_EQUAL_INT(0, rc);
TEST_ASSERT_EQUAL_INT(0, nclaimed);