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

feat(scheduler-sim-decision-trace): hook the host pre-filter, host scan and satisfaction check

Browse source 
Task 3.3: the pre-filter records selection mismatches and internal
errors as host verdicts, a scan stopped by an error marks the position
as a scan error, and unsatisfy_() hands every checked row to the current
record. Iterator and check results are unchanged.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
This commit is contained in:
Rongsong Shen 2026-10-08 11:42:19 +00:00
parent 542c6e602a
commit d4bee4b5d0
Signed by untrusted user: rshen
GPG key ID: 80533536B0EB651C
6 changed files with 834 additions and 3 deletions
Download patch file Download diff file Expand all files Collapse all files

6
src/scheduler/include/sched/trace.h
View file

@ -94,8 +94,10 @@ typedef struct xce_sched_trace_rec_s xce_sched_trace_rec_t;
* effects and argument evaluation included) runs only while `core` (an
* xce_sched_core_t*) has a record under construction in trace_rec, so a
* hook costs one pointer test when nothing is traced, and nothing else.
* Only the allocator thread sets trace_rec, so hooks are placed only on
* that thread. One statement: safe in an unbraced if/else.
* Only the allocator thread sets trace_rec. A hooked function called from
* another thread (the sim thread's reconciliation reaches unsatisfy_)
* sees NULL there, because the allocator clears trace_rec before it hands
* each decision back. One statement: safe in an unbraced if/else.
*/
#define SCHED_TRACE(core, call) \
do { \

31
src/scheduler/src/host_iter.c
View file

@ -44,6 +44,7 @@
#include <stdlib.h>
#include "sched/alloc_iter.h"
#include "sched/trace.h"
#include "sched/types.h"
#include "scheduler.h"
@ -124,6 +125,12 @@ xce_sched_worker_t* sched_host_iter_next(xce_host_iter_t* iter)
rc = iter_next(&(priv->it), (void**)&w);
if ((rc < 0) || (rc == RTTI_ITER_NO_DATA)) {
if (rc < 0) {
/* Decision trace (4.4): the scan stopped on an error, not
* on exhaustion; the position's unfilled end shows it */
SCHED_TRACE(core,
sched_trace_pos_scan_error(core->trace_rec, rc));
}
priv->exhausted = 1;
return NULL;
}
@ -132,10 +139,20 @@ xce_sched_worker_t* sched_host_iter_next(xce_host_iter_t* iter)
continue;
}
/* Decision trace: the pre-filter's own host scope. A rejection
* below ends it with the host's verdict; a host returned to the
* caller leaves it open for the caller's own host_begin, which
* replaces it (alloc_runner.c, task 3.4). */
SCHED_TRACE(core, sched_trace_host_begin(core->trace_rec,
ctx->schedobj, w));
/* 1. Select-expression filter -- existing evaluator, unchanged. */
if (!sched_meet_host_selection(
core, w,
(ctx->primary != NULL) ? ctx->primary->select : NULL)) {
/* Decision trace (5.3): selection mismatch, pre-filter phase */
SCHED_TRACE(core, sched_trace_host_verdict(
core->trace_rec, XCE_TRACE_V_SELECT, 0));
continue;
}
@ -152,6 +169,10 @@ xce_sched_worker_t* sched_host_iter_next(xce_host_iter_t* iter)
* growing its per-host uncommitted-debit vector) -- not a
* resource-availability determination. Skip this host and
* keep scanning rather than aborting the whole search. */
/* Decision trace (5.7): not evaluated, internal error */
SCHED_TRACE(core, sched_trace_host_verdict(
core->trace_rec, XCE_TRACE_V_INTERNAL,
rc));
continue;
}
@ -161,6 +182,16 @@ xce_sched_worker_t* sched_host_iter_next(xce_host_iter_t* iter)
&(ctx->alloc_req.host),
&(ctx->unmeet));
if (rc <= 0) {
/* Decision trace: an error is an internal-error verdict
* (5.7); a rejection takes its verdict from the unmet rows
* the check just recorded in this host scope (5.4, 5.5).
* That verdict carries the per-host phase; the runner never
* reaches it, since it empties alloc_req.host before every
* next() and an empty want is satisfied at once. */
SCHED_TRACE(core, sched_trace_host_verdict(
core->trace_rec,
(rc < 0) ? XCE_TRACE_V_INTERNAL : 0u,
rc));
continue;
}
}

6
src/scheduler/src/resource.c
View file

@ -7,6 +7,7 @@
#include "plugin.h"
#include "preason.h"
#include "resreq.h"
#include "sched/trace.h"
#include "sched/types.h"
#include "scheduler.h"
#include "utils.h"
@ -151,6 +152,11 @@ static int unsatisfy_(xce_sched_core_t* core, int nres, int* idxs, float* total,
int i, j;
int count;
/* Decision trace (4.1, 4.2, 5.4, 5.5): every row to the record; the
* capture keeps them by the open scope (cluster-wide check or host) */
SCHED_TRACE(core, sched_trace_check_rows(core->trace_rec, core, nres, idxs,
total, inuse, uncomitted, wanted));
count = 0;
for (i = 0; i < nres; i++) {
float uncomit = uncomit_at_(uncomitted, idxs[i]);

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

@ -2680,3 +2680,19 @@ sched_trace_hooks_resreq_test = executable(
test('scheduler trace hooks resreq test',
sched_trace_hooks_resreq_test, suite: 'scheduler',
env: ['ASAN_OPTIONS=detect_leaks=0'])
# scheduler-sim-decision-trace task 3.3: the host pre-filter, host scan and
# satisfaction-check trace hooks (design.md "TraceHooks": Pre-filter, Scan
# stop, Shortfall rows; Requirements 4.1, 4.2, 4.4, 5.2-5.5, 5.7) on the
# real host_iter.c and resource.c. Same white-box #include set as
# sched_host_iter_test, with host_iter.c's fault seams renamed by macros.
sched_trace_host_hooks_test = executable(
'sched_trace_host_hooks_test',
['test_trace_host_hooks.c'],
include_directories:[incdir],
dependencies: [scheduler_test_deps],
install: false)
test('scheduler trace host hooks test',
sched_trace_host_hooks_test, suite: 'scheduler',
env: ['ASAN_OPTIONS=detect_leaks=0'])

775
src/scheduler/testsuite/test_trace_host_hooks.c Normal file
View file

@ -0,0 +1,775 @@
/*
* test_trace_host_hooks.c - decision-trace hooks in the host pre-filter,
* the host scan and the satisfaction check (scheduler-sim-decision-trace
* task 3.3; design.md "TraceHooks": Pre-filter, Scan stop, Shortfall rows;
* Requirements 4.1, 4.2, 4.4, 5.2-5.5, 5.7).
*
* Drives the real host_iter.c and resource.c over the same white-box
* fixture as test_host_iter.c (core.c + resource.c + schedobj.c +
* host_iter.c, all #included), with a trace record set on the core:
* - the pre-filter's selection mismatch is a SELECT verdict in the
* pre-filter phase, and its internal errors (the per-host uncommitted
* fetch, the availability check) are INTERNAL verdicts with the error;
* - a pre-filter resource rejection carries the unmet rows it found;
* - a scan that stops on an error marks the position, so its unfilled
* end is a scan error; plain exhaustion does not;
* - the satisfaction check hands its rows to the record in both scopes:
* the cluster-wide check keeps every row, a host scope its unmet rows;
* outside any scope nothing is kept.
* Every scenario is also run untraced, and the iterator's hosts and the
* check's return value and unmet vector must be identical (Requirement 10).
*
* Faults are injected by renaming three calls inside host_iter.c only
* (macros around its #include): the per-host uncommitted fetch, the
* availability check and the RTTI iterator step.
*/
#include <errno.h>
#include <pthread.h>
#include <stdlib.h>
#include <string.h>
#include <unity/unity.h>
#include "resreq.h"
#include "sched/alloc_iter.h"
#include "sched/trace.h"
#include "scheduler.h"
#include "../src/core.c"
#include "../src/resource.c"
#include "../src/schedobj.c"
/* ---- fault injection inside host_iter.c ------------------------------- */
static int g_uncomit_calls;
static int g_uncomit_fail_at; /* 1-based call to fail; 0: never */
static int g_check_calls;
static int g_check_fail_at;
static int g_iter_calls;
static int g_iter_fail_at;
static int test_get_uncomit_host(xvec_t** host, xce_sched_ctx_t* ctx, int idx)
{
if (++g_uncomit_calls == g_uncomit_fail_at) {
return -ENOMEM;
}
return sched_ctx_get_uncomit_host(host, ctx, idx);
}
static int test_check_host_satisfy(xce_sched_core_t* core,
xce_sched_worker_t* w, xvec_t* uncomitted,
xvec_idval_t* wanted, xvec_idval_t* unmeet)
{
if (++g_check_calls == g_check_fail_at) {
return -ENOMEM;
}
return sched_check_host_satisfy(core, w, uncomitted, wanted, unmeet);
}
static int test_iter_next(iterator_t* it, void** item)
{
if (++g_iter_calls == g_iter_fail_at) {
return -EIO;
}
return iter_next(it, item);
}
#define sched_ctx_get_uncomit_host test_get_uncomit_host
#define sched_check_host_satisfy test_check_host_satisfy
#define iter_next test_iter_next
#include "../src/host_iter.c"
#undef sched_ctx_get_uncomit_host
#undef sched_check_host_satisfy
#undef iter_next
#include "../src/sched_trace.c"
#include "../src/sched_trace_capture.c"
#include "../src/sched_trace_outcome.c"
#include "../src/sched_trace_scan.c"
/* Link-satisfying stubs, as in test_host_iter.c: none is reached here. */
void plugin_unregister(void* arg) { (void)arg; }
int plugin_prepare_allocation(xce_sched_core_t* s, xce_sched_ctx_t* c)
{
(void)s;
(void)c;
return 0;
}
int plugin_job_update(void* s, xce_sched_job_t* j)
{
(void)s;
(void)j;
return 0;
}
int plugin_queue_update(void* s, xce_sched_queue_t* q)
{
(void)s;
(void)q;
return 0;
}
int plugin_job_rusage_update(void* s, xce_sched_job_t* j, xce_job_rusages_t* r)
{
(void)s;
(void)j;
(void)r;
return 0;
}
static void reset_faults(void)
{
g_uncomit_calls = 0;
g_uncomit_fail_at = 0;
g_check_calls = 0;
g_check_fail_at = 0;
g_iter_calls = 0;
g_iter_fail_at = 0;
}
void setUp(void) { reset_faults(); }
void tearDown(void) { return; }
/* ---- fixture ------------------------------------------------------------ */
#define SLOTS_RESIDX 0
#define RACK_RESIDX 1
#define MAX_COLLECT 16
typedef struct {
xce_scheduler_t sched;
xce_sched_ctx_t ctx;
xce_sched_obj_t obj;
xce_primary_resreq_t primary;
xce_select_node_t resname, num, eq;
xce_select_t select;
int bucket_slots[1];
xce_alloc_plan_t plan;
xce_sched_trace_ctl_t ctl;
xce_sched_trace_rec_t* rec; /* NULL: untraced */
} fx_t;
static void add_worker(xce_sched_core_t* core, const char* name,
float slots_total, float slots_inuse, float rack)
{
xce_sched_worker_t* w;
int idx[1];
float val[1];
w = (xce_sched_worker_t*)calloc(1, sizeof(*w));
TEST_ASSERT_NOT_NULL(w);
w->name = strdup(name);
TEST_ASSERT_NOT_NULL(w->name);
TEST_ASSERT_EQUAL_INT(0, zarray_init(&(w->resvals), 8, NULL));
TEST_ASSERT_EQUAL_INT(0, zarray_init(&(w->resInUse), 8, NULL));
idx[0] = SLOTS_RESIDX;
val[0] = slots_total;
put_resvals(&(w->resvals), 1, idx, val);
val[0] = slots_inuse;
put_resvals(&(w->resInUse), 1, idx, val);
idx[0] = RACK_RESIDX;
val[0] = rack;
put_resvals(&(w->resvals), 1, idx, val);
TEST_ASSERT_TRUE(sched_add_worker(core, w) >= 0);
}
/* A fixture with no workers, no select and no staged want; traced when
* `traced`, with a full-level record set on the core. */
static fx_t* fx_new(bool traced)
{
fx_t* fx = (fx_t*)calloc(1, sizeof(*fx));
TEST_ASSERT_NOT_NULL(fx);
TEST_ASSERT_EQUAL_INT(0, xce_sched_core_init(&(fx->sched.core)));
fx->primary.minTasks = 1;
fx->primary.maxTasks = 1;
fx->obj.tag = XCE_RESREQ_PRIMARY;
TEST_ASSERT_EQUAL_INT(
0, sched_ctx_init(&(fx->ctx), &(fx->sched.core), &(fx->obj)));
fx->ctx.primary = &(fx->primary);
fx->bucket_slots[0] = 1;
fx->plan.nbuckets = 1;
fx->plan.bucket_slots = fx->bucket_slots;
fx->plan.ctx = &(fx->ctx);
fx->ctl.all = true;
fx->ctl.level = XCE_TRACE_LEVEL_FULL;
fx->ctl.line_bound = XCE_TRACE_LINE_BOUND_DEFAULT;
fx->ctl.position = 1;
if (traced) {
fx->rec = sched_trace_rec_new(&(fx->ctl), 7, XCE_TRACE_REC_DECISION);
TEST_ASSERT_NOT_NULL(fx->rec);
fx->sched.core.trace_rec = fx->rec;
}
return fx;
}
static void fx_free(fx_t* fx)
{
fx->sched.core.trace_rec = NULL;
sched_trace_rec_free(fx->rec);
free(fx);
}
/* select `rack == literal` (as test_host_iter.c) */
static void fx_select_rack(fx_t* fx, double literal)
{
fx->resname.type = SEL_RESNAME;
fx->resname.u.resname.res_idx = RACK_RESIDX;
fx->resname.u.resname.res_type = RES_TYPE_NUMERIC;
fx->num.type = SEL_NUM;
fx->num.u.numval = literal;
fx->eq.type = SEL_EQ;
fx->eq.u.binop.left = &(fx->resname);
fx->eq.u.binop.right = &(fx->num);
fx->select.root = &(fx->eq);
fx->primary.select = &(fx->select);
}
static void fx_want(fx_t* fx, int idx, float val)
{
TEST_ASSERT_TRUE(xvec_idval_store(&(fx->ctx.alloc_req.host), idx, val)
>= 0);
}
/* One whole scan of a fresh iterator; the hosts returned, in order. */
static int fx_scan(fx_t* fx, char names[][64])
{
xce_host_iter_t it;
xce_sched_worker_t* w;
int n = 0;
memset(&it, 0, sizeof(it));
it.init = sched_host_iter_init;
it.next = sched_host_iter_next;
it.fini = sched_host_iter_fini;
TEST_ASSERT_EQUAL_INT(0, it.init(&it, &(fx->plan), fx->sched.core.workers));
while ((n < MAX_COLLECT) && ((w = it.next(&it)) != NULL)) {
strncpy(names[n], w->name, 63);
names[n][63] = '\0';
n++;
}
it.fini(&it);
return n;
}
static void assert_same_hosts(char a[][64], int na, char b[][64], int nb)
{
int i;
TEST_ASSERT_EQUAL_INT(na, nb);
for (i = 0; i < na; i++) {
TEST_ASSERT_EQUAL_STRING(a[i], b[i]);
}
}
static int count_entries(const xce_sched_trace_rec_t* rec, int type)
{
const sched_trace_entry_t* e;
int n = 0;
for (e = rec->entries; e != NULL; e = e->next) {
if ((type < 0) || ((int)e->type == type)) {
n++;
}
}
return n;
}
/* The HOST entry naming `host`, or NULL. */
static sched_trace_entry_t* host_entry(xce_sched_trace_rec_t* rec,
const char* host)
{
sched_trace_entry_t* e;
for (e = rec->entries; e != NULL; e = e->next) {
if ((e->type == SCHED_TRACE_E_HOST) && (e->u.host.host != NULL)
&& (strcmp(e->u.host.host, host) == 0)) {
return e;
}
}
return NULL;
}
static sched_trace_entry_t* first_entry(xce_sched_trace_rec_t* rec, int type)
{
sched_trace_entry_t* e;
for (e = rec->entries; e != NULL; e = e->next) {
if ((int)e->type == type) {
return e;
}
}
return NULL;
}
static void assert_row(const sched_trace_entry_t* e, int type, int idx,
float total, float inuse, float reserved,
float requested, bool met)
{
TEST_ASSERT_NOT_NULL(e);
TEST_ASSERT_EQUAL_INT(type, e->type);
TEST_ASSERT_EQUAL_INT(idx, e->u.row.idx);
TEST_ASSERT_EQUAL_FLOAT(total, e->u.row.total);
TEST_ASSERT_EQUAL_FLOAT(inuse, e->u.row.inuse);
TEST_ASSERT_EQUAL_FLOAT(reserved, e->u.row.reserved);
TEST_ASSERT_EQUAL_FLOAT(requested, e->u.row.requested);
TEST_ASSERT_EQUAL(met, e->u.row.met);
}
/* verdict bit -> host_evals index */
static int vk(unsigned bit)
{
int k = 0;
while ((bit >> k) != 1u) {
k++;
}
return k;
}
/* ---- pre-filter verdicts ------------------------------------------------ */
/* 5.2, 5.3: a host failing the select expression gets a selection-mismatch
* verdict from the pre-filter; the host it returns gets none here (the
* caller's own per-host check gives it one, task 3.4). */
static void test_prefilter_selection_mismatch_is_a_select_verdict(void)
{
char plain[MAX_COLLECT][64], traced[MAX_COLLECT][64];
int np, nt;
fx_t* fx;
sched_trace_entry_t* e;
fx = fx_new(false);
add_worker(&(fx->sched.core), "rack-match", 4.0f, 0.0f, 1.0f);
add_worker(&(fx->sched.core), "rack-nomatch", 4.0f, 0.0f, 2.0f);
fx_select_rack(fx, 1.0);
np = fx_scan(fx, plain);
fx_free(fx);
fx = fx_new(true);
add_worker(&(fx->sched.core), "rack-match", 4.0f, 0.0f, 1.0f);
add_worker(&(fx->sched.core), "rack-nomatch", 4.0f, 0.0f, 2.0f);
fx_select_rack(fx, 1.0);
nt = fx_scan(fx, traced);
assert_same_hosts(plain, np, traced, nt);
TEST_ASSERT_EQUAL_INT(1, nt);
TEST_ASSERT_EQUAL_STRING("rack-match", traced[0]);
e = host_entry(fx->rec, "rack-nomatch");
TEST_ASSERT_NOT_NULL(e);
TEST_ASSERT_EQUAL_UINT(XCE_TRACE_V_SELECT, e->u.host.verdicts);
TEST_ASSERT_EQUAL_INT(SCHED_TRACE_PHASE_PREFILTER, e->u.host.phase);
TEST_ASSERT_EQUAL_INT(0, e->u.host.err);
TEST_ASSERT_NULL(host_entry(fx->rec, "rack-match"));
TEST_ASSERT_EQUAL_INT(1, count_entries(fx->rec, -1));
TEST_ASSERT_EQUAL_INT(1,
fx->rec->counters.host_evals[vk(XCE_TRACE_V_SELECT)]);
fx_free(fx);
}
/* 5.7: the per-host uncommitted fetch failing in the pre-filter makes the
* host "not evaluated", with the error; the scan goes on. */
static void test_prefilter_uncomit_error_is_an_internal_verdict(void)
{
char plain[MAX_COLLECT][64], traced[MAX_COLLECT][64];
int np, nt;
fx_t* fx;
sched_trace_entry_t* e;
const char* skipped;
fx = fx_new(false);
add_worker(&(fx->sched.core), "a", 4.0f, 0.0f, 0.0f);
add_worker(&(fx->sched.core), "b", 4.0f, 0.0f, 0.0f);
g_uncomit_fail_at = 1;
np = fx_scan(fx, plain);
fx_free(fx);
reset_faults();
fx = fx_new(true);
add_worker(&(fx->sched.core), "a", 4.0f, 0.0f, 0.0f);
add_worker(&(fx->sched.core), "b", 4.0f, 0.0f, 0.0f);
g_uncomit_fail_at = 1;
nt = fx_scan(fx, traced);
assert_same_hosts(plain, np, traced, nt);
TEST_ASSERT_EQUAL_INT(1, nt);
skipped = (strcmp(traced[0], "a") == 0) ? "b" : "a";
e = host_entry(fx->rec, skipped);
TEST_ASSERT_NOT_NULL(e);
TEST_ASSERT_EQUAL_UINT(XCE_TRACE_V_INTERNAL, e->u.host.verdicts);
TEST_ASSERT_EQUAL_INT(SCHED_TRACE_PHASE_NONE, e->u.host.phase);
TEST_ASSERT_EQUAL_INT(-ENOMEM, e->u.host.err);
TEST_ASSERT_NULL(host_entry(fx->rec, traced[0]));
TEST_ASSERT_EQUAL_INT(
1, fx->rec->counters.host_evals[vk(XCE_TRACE_V_INTERNAL)]);
fx_free(fx);
}
/* 5.7: the pre-filter's availability check failing is likewise an
* internal-error verdict, not a shortfall. */
static void test_prefilter_check_error_is_an_internal_verdict(void)
{
char plain[MAX_COLLECT][64], traced[MAX_COLLECT][64];
int np, nt;
fx_t* fx;
sched_trace_entry_t* e;
const char* skipped;
fx = fx_new(false);
add_worker(&(fx->sched.core), "a", 4.0f, 0.0f, 0.0f);
add_worker(&(fx->sched.core), "b", 4.0f, 0.0f, 0.0f);
fx_want(fx, SLOTS_RESIDX, 1.0f);
g_check_fail_at = 1;
np = fx_scan(fx, plain);
fx_free(fx);
reset_faults();
fx = fx_new(true);
add_worker(&(fx->sched.core), "a", 4.0f, 0.0f, 0.0f);
add_worker(&(fx->sched.core), "b", 4.0f, 0.0f, 0.0f);
fx_want(fx, SLOTS_RESIDX, 1.0f);
g_check_fail_at = 1;
nt = fx_scan(fx, traced);
assert_same_hosts(plain, np, traced, nt);
TEST_ASSERT_EQUAL_INT(1, nt);
skipped = (strcmp(traced[0], "a") == 0) ? "b" : "a";
e = host_entry(fx->rec, skipped);
TEST_ASSERT_NOT_NULL(e);
TEST_ASSERT_EQUAL_UINT(XCE_TRACE_V_INTERNAL, e->u.host.verdicts);
TEST_ASSERT_EQUAL_INT(-ENOMEM, e->u.host.err);
TEST_ASSERT_EQUAL_INT(0, count_entries(fx->rec, SCHED_TRACE_E_HOST_ROW));
fx_free(fx);
}
/* 5.4: a host the pre-filter's availability check rejects is recorded with
* the unmet rows that check found, under its verdict line. */
static void test_prefilter_resource_rejection_carries_its_rows(void)
{
char plain[MAX_COLLECT][64], traced[MAX_COLLECT][64];
int np, nt;
fx_t* fx;
sched_trace_entry_t* e;
fx = fx_new(false);
add_worker(&(fx->sched.core), "big", 4.0f, 0.0f, 0.0f);
add_worker(&(fx->sched.core), "small", 4.0f, 3.0f, 0.0f);
fx_want(fx, SLOTS_RESIDX, 2.0f);
np = fx_scan(fx, plain);
fx_free(fx);
fx = fx_new(true);
add_worker(&(fx->sched.core), "big", 4.0f, 0.0f, 0.0f);
add_worker(&(fx->sched.core), "small", 4.0f, 3.0f, 0.0f);
fx_want(fx, SLOTS_RESIDX, 2.0f);
nt = fx_scan(fx, traced);
assert_same_hosts(plain, np, traced, nt);
TEST_ASSERT_EQUAL_INT(1, nt);
TEST_ASSERT_EQUAL_STRING("big", traced[0]);
e = host_entry(fx->rec, "small");
TEST_ASSERT_NOT_NULL(e);
TEST_ASSERT_EQUAL_UINT(XCE_TRACE_V_SHORTFALL, e->u.host.verdicts);
assert_row(e->next, SCHED_TRACE_E_HOST_ROW, SLOTS_RESIDX, 4.0f, 3.0f, 0.0f,
2.0f, false);
TEST_ASSERT_EQUAL_INT(e->depth + 1, e->next->depth);
TEST_ASSERT_EQUAL_INT(1, count_entries(fx->rec, SCHED_TRACE_E_HOST_ROW));
TEST_ASSERT_NULL(host_entry(fx->rec, "big"));
fx_free(fx);
}
/* ---- scan stop ---------------------------------------------------------- */
/* 4.4: a scan that stops on an iterator error marks the position, so the
* position's unfilled end becomes a scan error with that error. */
static void test_scan_error_marks_the_position(void)
{
char plain[MAX_COLLECT][64], traced[MAX_COLLECT][64];
int np, nt;
fx_t* fx;
sched_trace_entry_t* e;
fx = fx_new(false);
add_worker(&(fx->sched.core), "a", 4.0f, 0.0f, 0.0f);
add_worker(&(fx->sched.core), "b", 4.0f, 0.0f, 0.0f);
g_iter_fail_at = 2;
np = fx_scan(fx, plain);
fx_free(fx);
reset_faults();
fx = fx_new(true);
add_worker(&(fx->sched.core), "a", 4.0f, 0.0f, 0.0f);
add_worker(&(fx->sched.core), "b", 4.0f, 0.0f, 0.0f);
g_iter_fail_at = 2;
sched_trace_pos_begin(fx->rec, 1, 1);
nt = fx_scan(fx, traced);
assert_same_hosts(plain, np, traced, nt);
TEST_ASSERT_EQUAL_INT(1, nt);
TEST_ASSERT_TRUE(fx->rec->scan_err_set);
TEST_ASSERT_EQUAL_INT(-EIO, fx->rec->scan_err);
sched_trace_pos_end(fx->rec, XCE_TRACE_POS_UNFILLED, NULL);
e = first_entry(fx->rec, SCHED_TRACE_E_POS_END);
TEST_ASSERT_NOT_NULL(e);
TEST_ASSERT_EQUAL_INT(XCE_TRACE_POS_SCAN_ERROR, e->u.pos_end.how);
TEST_ASSERT_EQUAL_INT(-EIO, e->u.pos_end.err);
fx_free(fx);
}
/* 4.4: an error on the very first step leaves no host and still marks the
* position. */
static void test_scan_error_before_any_host(void)
{
char plain[MAX_COLLECT][64], traced[MAX_COLLECT][64];
int np, nt;
fx_t* fx;
fx = fx_new(false);
add_worker(&(fx->sched.core), "a", 4.0f, 0.0f, 0.0f);
g_iter_fail_at = 1;
np = fx_scan(fx, plain);
fx_free(fx);
reset_faults();
fx = fx_new(true);
add_worker(&(fx->sched.core), "a", 4.0f, 0.0f, 0.0f);
g_iter_fail_at = 1;
nt = fx_scan(fx, traced);
assert_same_hosts(plain, np, traced, nt);
TEST_ASSERT_EQUAL_INT(0, nt);
TEST_ASSERT_TRUE(fx->rec->scan_err_set);
TEST_ASSERT_EQUAL_INT(-EIO, fx->rec->scan_err);
TEST_ASSERT_EQUAL_INT(0, count_entries(fx->rec, -1));
fx_free(fx);
}
/* A scan that simply runs out of hosts is not a scan error. */
static void test_exhausted_scan_is_not_a_scan_error(void)
{
char traced[MAX_COLLECT][64];
fx_t* fx;
sched_trace_entry_t* e;
fx = fx_new(true);
add_worker(&(fx->sched.core), "a", 4.0f, 0.0f, 0.0f);
sched_trace_pos_begin(fx->rec, 1, 1);
TEST_ASSERT_EQUAL_INT(1, fx_scan(fx, traced));
TEST_ASSERT_FALSE(fx->rec->scan_err_set);
sched_trace_pos_end(fx->rec, XCE_TRACE_POS_UNFILLED, NULL);
e = first_entry(fx->rec, SCHED_TRACE_E_POS_END);
TEST_ASSERT_NOT_NULL(e);
TEST_ASSERT_EQUAL_INT(XCE_TRACE_POS_UNFILLED, e->u.pos_end.how);
fx_free(fx);
}
/* ---- satisfaction check rows -------------------------------------------- */
static void make_uncomit(xvec_t* v, float* buf, int n, int idx, float val)
{
memset(buf, 0, sizeof(float) * (size_t)n);
buf[idx] = val;
v->count = n;
v->elsize = sizeof(float);
v->data = buf;
}
static void assert_same_unmeet(int rc_a, const xvec_idval_t* a, int rc_b,
const xvec_idval_t* b)
{
int i;
TEST_ASSERT_EQUAL_INT(rc_a, rc_b);
TEST_ASSERT_EQUAL_INT(a->count, b->count);
for (i = 0; i < a->count; i++) {
TEST_ASSERT_EQUAL_INT(((int*)a->idxs.data)[i], ((int*)b->idxs.data)[i]);
TEST_ASSERT_EQUAL_FLOAT(((float*)a->vals.data)[i],
((float*)b->vals.data)[i]);
}
}
/* Shared state: resource 0 has 10, 4 in use; resource 1 has 2, 2 in use. */
static void shared_state(xce_sched_core_t* core)
{
int idx[2] = { 0, 1 };
float tot[2] = { 10.0f, 2.0f };
float use[2] = { 4.0f, 2.0f };
put_resvals(&(core->sharedInst), 2, idx, tot);
put_resvals(&(core->sharedInUse), 2, idx, use);
}
/* Runs the cluster-wide check for {0: 3, 1: 1} with 1 of resource 0
* reserved; inside an open check scope when `scoped`. */
static int shared_check(fx_t* fx, bool scoped, xvec_idval_t* unmeet)
{
xvec_idval_t want;
xvec_t uncomit;
float buf[4];
int rc;
shared_state(&(fx->sched.core));
make_uncomit(&uncomit, buf, 4, 0, 1.0f);
TEST_ASSERT_EQUAL_INT(0, xvec_idval_init(&want, 2));
TEST_ASSERT_TRUE(xvec_idval_store(&want, 0, 3.0f) >= 0);
TEST_ASSERT_TRUE(xvec_idval_store(&want, 1, 1.0f) >= 0);
memset(unmeet, 0, sizeof(*unmeet));
if (scoped && (fx->rec != NULL)) {
sched_trace_check_begin(fx->rec, &(fx->obj));
}
rc = sched_check_shared_satisfy(&(fx->sched.core), &uncomit, &want, unmeet);
if (scoped && (fx->rec != NULL)) {
sched_trace_check_end(fx->rec, rc, rc > 0);
}
xvec_idval_fini(&want);
return rc;
}
/* 4.1: in the cluster-wide check every row is kept, met or not, with its
* total, usage, reserved and requested amounts, before the check's end. */
static void test_check_rows_cluster_wide_scope(void)
{
xvec_idval_t um_plain, um_traced;
int rc_plain, rc_traced;
fx_t* fx;
sched_trace_entry_t* e;
fx = fx_new(false);
rc_plain = shared_check(fx, true, &um_plain);
fx_free(fx);
fx = fx_new(true);
rc_traced = shared_check(fx, true, &um_traced);
assert_same_unmeet(rc_plain, &um_plain, rc_traced, &um_traced);
TEST_ASSERT_EQUAL_INT(0, rc_traced);
TEST_ASSERT_EQUAL_INT(3, count_entries(fx->rec, -1));
e = fx->rec->entries;
assert_row(e, SCHED_TRACE_E_CHECK_ROW, 0, 10.0f, 4.0f, 1.0f, 3.0f, true);
assert_row(e->next, SCHED_TRACE_E_CHECK_ROW, 1, 2.0f, 2.0f, 0.0f, 1.0f,
false);
TEST_ASSERT_EQUAL_INT(SCHED_TRACE_E_CHECK_END, e->next->next->type);
xvec_idval_fini(&um_plain);
xvec_idval_fini(&um_traced);
fx_free(fx);
}
/* Rows of a check run outside any scope belong to nothing and are not
* kept; the check's result is unchanged. */
static void test_check_rows_outside_a_scope_are_dropped(void)
{
xvec_idval_t um_plain, um_traced;
int rc_plain, rc_traced;
fx_t* fx;
fx = fx_new(false);
rc_plain = shared_check(fx, false, &um_plain);
fx_free(fx);
fx = fx_new(true);
rc_traced = shared_check(fx, false, &um_traced);
assert_same_unmeet(rc_plain, &um_plain, rc_traced, &um_traced);
TEST_ASSERT_EQUAL_INT(0, count_entries(fx->rec, -1));
xvec_idval_fini(&um_plain);
xvec_idval_fini(&um_traced);
fx_free(fx);
}
/* Runs the per-host check of worker "h" (4 slots, 1 in use, rack 0) for
* {slots: 2, rack: 1} inside a host scope ended by a from-rows verdict,
* as the runner's per-host check is (task 3.4). */
static int host_check(fx_t* fx, xvec_idval_t* unmeet)
{
xce_sched_worker_t* w;
xvec_idval_t want;
xvec_t uncomit;
float buf[4];
int rc;
add_worker(&(fx->sched.core), "h", 4.0f, 1.0f, 0.0f);
w = sched_get_worker(&(fx->sched.core), "h");
TEST_ASSERT_NOT_NULL(w);
make_uncomit(&uncomit, buf, 4, SLOTS_RESIDX, 1.0f);
TEST_ASSERT_EQUAL_INT(0, xvec_idval_init(&want, 2));
TEST_ASSERT_TRUE(xvec_idval_store(&want, SLOTS_RESIDX, 2.0f) >= 0);
TEST_ASSERT_TRUE(xvec_idval_store(&want, RACK_RESIDX, 1.0f) >= 0);
memset(unmeet, 0, sizeof(*unmeet));
if (fx->rec != NULL) {
sched_trace_host_begin(fx->rec, &(fx->obj), w);
}
rc = sched_check_host_satisfy(&(fx->sched.core), w, &uncomit, &want,
unmeet);
if (fx->rec != NULL) {
sched_trace_host_verdict(fx->rec, (rc < 0) ? XCE_TRACE_V_INTERNAL : 0u,
rc);
}
xvec_idval_fini(&want);
return rc;
}
/* 5.4: in a host scope only the unmet rows are kept, under the host's
* verdict line, and they make the verdict a resource shortfall. */
static void test_check_rows_host_scope(void)
{
xvec_idval_t um_plain, um_traced;
int rc_plain, rc_traced;
fx_t* fx;
sched_trace_entry_t* e;
fx = fx_new(false);
rc_plain = host_check(fx, &um_plain);
fx_free(fx);
fx = fx_new(true);
rc_traced = host_check(fx, &um_traced);
assert_same_unmeet(rc_plain, &um_plain, rc_traced, &um_traced);
TEST_ASSERT_EQUAL_INT(0, rc_traced);
TEST_ASSERT_EQUAL_INT(2, count_entries(fx->rec, -1));
e = host_entry(fx->rec, "h");
TEST_ASSERT_NOT_NULL(e);
TEST_ASSERT_EQUAL_UINT(XCE_TRACE_V_SHORTFALL, e->u.host.verdicts);
TEST_ASSERT_EQUAL_INT(SCHED_TRACE_PHASE_PERHOST, e->u.host.phase);
/* slots: 1 in use + 1 reserved + 2 wanted <= 4, met and not kept */
assert_row(e->next, SCHED_TRACE_E_HOST_ROW, RACK_RESIDX, 0.0f, 0.0f, 0.0f,
1.0f, false);
TEST_ASSERT_EQUAL_INT(e->depth + 1, e->next->depth);
xvec_idval_fini(&um_plain);
xvec_idval_fini(&um_traced);
fx_free(fx);
}
int main(void)
{
UNITY_BEGIN();
RUN_TEST(test_prefilter_selection_mismatch_is_a_select_verdict);
RUN_TEST(test_prefilter_uncomit_error_is_an_internal_verdict);
RUN_TEST(test_prefilter_check_error_is_an_internal_verdict);
RUN_TEST(test_prefilter_resource_rejection_carries_its_rows);
RUN_TEST(test_scan_error_marks_the_position);
RUN_TEST(test_scan_error_before_any_host);
RUN_TEST(test_exhausted_scan_is_not_a_scan_error);
RUN_TEST(test_check_rows_cluster_wide_scope);
RUN_TEST(test_check_rows_outside_a_scope_are_dropped);
RUN_TEST(test_check_rows_host_scope);
return UNITY_END();
}