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

feat(scheduler-sim-decision-trace): capture decision outcomes and reconstructed dispatches

Browse source 
Task 2.3: dispatch rows with limit charges listed separately and
resolved through the job's matched limits (raw-number fallback), pend
reason set deep-copied, 50 hosts in detail with the rest aggregated,
and RECONSTRUCTED records holding the dispatch outcome.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
This commit is contained in:
Rongsong Shen 2026-10-08 09:58:21 +00:00
parent 79c5a0cbda
commit 8efe142bad
Signed by untrusted user: rshen
GPG key ID: 80533536B0EB651C
4 changed files with 1269 additions and 68 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -371,6 +371,68 @@ typedef struct sched_trace_entry_s {
} u;
} sched_trace_entry_t;
/*
* Decision outcome (6.1-6.4, 9.4), filled by TraceCapture and rendered
* outside the line budget. All pointers point into the record's arena.
*/
typedef enum {
SCHED_TRACE_OUTCOME_NONE = 0,
SCHED_TRACE_OUTCOME_DISPATCH,
SCHED_TRACE_OUTCOME_PEND
} sched_trace_outcome_kind_t;
/* One allocated amount: a resource row, or a limit charge (6.3: a charge is
* never a resource row). A charge names its limit and scope instance (a
* host charge has "h=<host>" appended); `limited_res` is the resource the
* limit limits, NULL when no matched limit names the charge, in which case
* the amount is a raw number. In an aggregate the scope has no host part. */
typedef struct {
const char* res; /* allocated resource name (a charge: the vres name;
* NULL in a limit aggregate) */
double value; /* amount, or the total in an aggregate */
const char* limit; /* charges only */
const char* scope; /* charges only */
const char* limited_res; /* charges only; NULL: unknown (raw number) */
} sched_trace_alloc_row_t;
/* One allocated host kept in detail (6.1), in allocation order. */
typedef struct {
const char* host;
int nres;
sched_trace_alloc_row_t* res;
int ncharges;
sched_trace_alloc_row_t* charges;
} sched_trace_host_out_t;
/* Dispatch outcome: the first XCE_TRACE_OUTCOME_HOSTS hosts in detail, the
* job's shared rows and charges (6.2), and the omitted hosts' totals per
* resource (limits excluded) and per limit, in the order first seen (9.4). */
typedef struct {
int nhosts; /* kept in detail */
sched_trace_host_out_t* hosts;
int nshared;
sched_trace_alloc_row_t* shared;
int nshared_charges;
sched_trace_alloc_row_t* shared_charges;
int omitted_hosts;
int nagg_res;
sched_trace_alloc_row_t* agg_res;
int nagg_charges;
sched_trace_alloc_row_t* agg_charges;
} sched_trace_dispatch_out_t;
/* Pend outcome (6.4): a copy of the decision's pend-reason set holding the
* main reason, every entry that names no host and every entry of the first
* XCE_TRACE_OUTCOME_HOSTS distinct hosts, in source order (equal to the
* source under job_preason_eq() when no host was omitted). Entries of the
* hosts after them are counted per pend-reason code (9.4). */
typedef struct {
xce_job_preason_t* reasons; /* NULL: the decision had none */
int omitted_hosts;
int ncodes;
sched_trace_reason_count_t* omitted_codes; /* first-seen order */
} sched_trace_pend_out_t;
/* One arena chunk; the record frees the whole chain at once. */
typedef struct sched_trace_chunk_s {
struct sched_trace_chunk_s* next;
@ -416,6 +478,11 @@ struct xce_sched_trace_rec_s {
sched_trace_counters_t counters;
/* outcome (6.x, 9.4), outside the line budget */
sched_trace_outcome_kind_t outcome;
sched_trace_dispatch_out_t dispatch; /* outcome DISPATCH */
sched_trace_pend_out_t pend; /* outcome PEND */
/* stored entries, in the order made; at most line_bound of them */
sched_trace_entry_t* entries;
sched_trace_entry_t* entries_tail;
@ -677,4 +744,37 @@ extern void sched_trace_check_rows(xce_sched_trace_rec_t* rec,
extern void sched_trace_capture_attempt_reasons(xce_sched_trace_rec_t* rec,
const xce_pendreason_t* reason);
/*
* A dispatch decision's allocation as the record's outcome (6.1-6.3, 9.4).
* Each row is a limit charge when its resource, found by name in
* core->resmgr, has XCE_RESF_LIMIT, or, when the name is not registered,
* when it starts with "limit=". A charge is resolved by vres name through
* the job's matched limits: obj->glimits for shared rows,
* obj->hlimits[w->idx] for a host's rows (w found by host name in
* core->workers); unresolved, its name and scope are parsed from the
* resource name and its limited resource is unknown. NULL obj: nothing
* resolves. A NULL alloc stores an empty dispatch. `core` is only read.
*/
extern void sched_trace_capture_dispatch(xce_sched_trace_rec_t* rec,
xce_sched_core_t* core,
const xce_sched_obj_t* obj,
const xce_job_alloc_t* alloc);
/* A pend decision's pend-reason set as the record's outcome (6.4, 9.4). */
extern void sched_trace_capture_pend(xce_sched_trace_rec_t* rec,
const xce_pendreason_t* reason);
/*
* A RECONSTRUCTED record (2.5) for a dispatch the simulator rebuilt from a
* recorded allocation, built on the sim thread: the header from `ctl` and
* `allocs` as its dispatch outcome, captured as above (`obj` is the job's
* schedobj, NULL when it has none). NULL when ctl is NULL or the record
* cannot be allocated.
*/
extern xce_sched_trace_rec_t*
sched_trace_rec_reconstructed(const xce_sched_trace_ctl_t* ctl,
xce_sched_core_t* core, uint64_t jobid,
const xce_sched_obj_t* obj,
const xce_job_alloc_t* allocs);
#endif

689
src/scheduler/src/sched_trace_capture.c
View file

@ -16,10 +16,12 @@
*
* Task 2.1: selection facts (2.2), cluster-wide limits (2.3) and attempt
* failure reasons (3.5). Task 2.2: the satisfaction check's rows (4.1,
* 4.2, 5.4, 5.5). These read struct fields and call no scheduler function
* (the row capture resolves resources with librtti's vget(), which every
* scheduler target links), so every white-box test that #includes this
* file links without extra sources.
* 4.2, 5.4, 5.5). Task 2.3: the dispatch and pend outcomes (6.1-6.4, 9.4)
* and the reconstructed-dispatch record (2.5). These read struct fields and
* call no scheduler function (resources and workers are looked up with
* librtti's vget()/key_find(), which every scheduler target links), so
* every white-box test that #includes this file links without extra
* sources.
*/
#include <stdlib.h>
#include <string.h>
@ -69,21 +71,34 @@ static bool split_vres_name(const char* vname, const char* known,
return true;
}
/* Arena copy of a vres scope part, "u=a/q=b" -> "u=a,q=b". */
static char* copy_scope(xce_sched_trace_rec_t* rec, const char* scope)
/* Arena copy of a vres scope part, "u=a/q=b" -> "u=a,q=b", with
* ",h=<host>" ("h=<host>" when the scope is empty) appended for a non-NULL
* host: host-limit vres names do not encode the host. */
static char* copy_scope(xce_sched_trace_rec_t* rec, const char* scope,
const char* host)
{
size_t n = strlen(scope);
size_t hn = (host != NULL) ? strlen(host) : 0;
char* out;
char* p;
size_t i;
out = (char*)sched_trace_arena_alloc(rec, n + 1);
out = (char*)sched_trace_arena_alloc(rec, n + hn + 4);
if (out == NULL) {
return NULL;
}
for (i = 0; i < n; i++) {
out[i] = (scope[i] == '/') ? ',' : scope[i];
for (i = 0, p = out; i < n; i++) {
*p++ = (scope[i] == '/') ? ',' : scope[i];
}
out[n] = '\0';
if (host != NULL) {
if (n > 0) {
*p++ = ',';
}
memcpy(p, "h=", 2);
memcpy(p + 2, host, hn);
p += 2 + hn;
}
*p = '\0';
return out;
}
@ -135,7 +150,7 @@ static bool limit_ref(xce_sched_trace_rec_t* rec, const xce_vres_t* vres,
} else {
out->limit = copy_prefix(rec, name, name_len);
}
out->scope = copy_scope(rec, scope);
out->scope = copy_scope(rec, scope, NULL);
return (out->limit != NULL) && (out->scope != NULL);
}
@ -216,35 +231,38 @@ static float row_uncomit_at(const xvec_t* uncomitted, int idx)
return ((const float*)uncomitted->data)[idx];
}
/* The current scope's matched limit list: obj->glimits in the cluster-wide
* check, obj->hlimits[w->idx] in a host scope (sparse: NULL when the
* worker has none). Read directly: zarray_get()/xlist_get() take
* non-const containers. */
static const xlist_t* scope_limits(const xce_sched_trace_rec_t* rec)
/* Worker `idx`'s matched host limits, obj->hlimits[idx] (sparse: NULL
* when the worker has none). Read directly: zarray_get() takes a
* non-const array. */
static const xlist_t* worker_limits(const xce_sched_obj_t* obj, int idx)
{
const zarray_t* hl;
int idx;
const zarray_t* hl = &obj->hlimits;
if (rec->scope_obj == NULL) {
return NULL;
}
if (rec->scope_w == NULL) {
return &rec->scope_obj->glimits;
}
hl = &rec->scope_obj->hlimits;
idx = rec->scope_w->idx;
if ((hl->data == NULL) || (idx < 0) || (idx >= hl->caps)) {
return NULL;
}
return (const xlist_t*)hl->data[idx];
}
/* The matched limit whose vres resource is `r`, or NULL. */
static const xce_vres_t* scope_vres(const xce_sched_trace_rec_t* rec,
const xce_resource_t* r)
/* The current scope's matched limit list: obj->glimits in the cluster-wide
* check, the worker's host limits in a host scope. */
static const xlist_t* scope_limits(const xce_sched_trace_rec_t* rec)
{
const xlist_t* list = scope_limits(rec);
int i;
if (rec->scope_obj == NULL) {
return NULL;
}
if (rec->scope_w == NULL) {
return &rec->scope_obj->glimits;
}
return worker_limits(rec->scope_obj, rec->scope_w->idx);
}
/* The matched limit in `list` whose vres resource is `r` or is named
* `name`, or NULL. Read directly: xlist_get() takes a non-const list. */
static const xce_vres_t* list_vres(const xlist_t* list,
const xce_resource_t* r, const char* name)
{
int i;
if ((list == NULL) || (list->data == NULL)) {
return NULL;
@ -255,15 +273,45 @@ static const xce_vres_t* scope_vres(const xce_sched_trace_rec_t* rec,
if ((v == NULL) || (v->vres == NULL)) {
continue;
}
if ((v->vres == r)
|| ((v->vres->name != NULL) && (r->name != NULL)
&& (strcmp(v->vres->name, r->name) == 0))) {
if (((r != NULL) && (v->vres == r))
|| ((v->vres->name != NULL) && (name != NULL)
&& (strcmp(v->vres->name, name) == 0))) {
return v;
}
}
return NULL;
}
/* A limit resource named `resname`, resolved: the limit name (not
* NUL-terminated), the vres-form scope part and the limited resource. */
typedef struct {
const char* name;
size_t name_len;
const char* scope;
const char* limited_res; /* NULL: no matched limit names it */
} charge_ref_t;
/* Resolves `resname` through its matched limit `v` (NULL: none, so the
* limit name and scope are parsed from the name, and the limited resource
* is unknown). A name without the "limit=" prefix names the limit
* itself. */
static void charge_resolve(const xce_vres_t* v, const char* resname,
charge_ref_t* c)
{
const char* known = ((v != NULL) && (v->limit != NULL)) ? v->limit->name
: NULL;
if (!split_vres_name(resname, known, &c->name, &c->name_len, &c->scope)) {
c->name = (known != NULL) ? known : resname;
c->name_len = (c->name != NULL) ? strlen(c->name) : 0;
c->scope = "";
} else if (known != NULL) {
c->name = known;
c->name_len = strlen(known);
}
c->limited_res = ((v != NULL) && (v->res != NULL)) ? v->res->name : NULL;
}
/*
* Fills a limit row's limit name, scope instance and limited resource for
* the limit vres resource `r`, composing the strings in a scratch buffer
@ -273,49 +321,34 @@ static const xce_vres_t* scope_vres(const xce_sched_trace_rec_t* rec,
static bool limit_row(const xce_sched_trace_rec_t* rec, const xce_resource_t* r,
sched_trace_row_t* row, char** buf)
{
const xce_vres_t* v = scope_vres(rec, r);
const char* known = NULL;
const char* host = NULL;
const char* name;
size_t name_len;
const char* scope;
size_t scope_len;
size_t host_len = 0;
char* p;
size_t k;
charge_ref_t c;
const char* host = NULL;
size_t host_len = 0;
size_t scope_len;
char* p;
size_t k;
if ((v != NULL) && (v->limit != NULL)) {
known = v->limit->name;
}
if (!split_vres_name(r->name, known, &name, &name_len, &scope)) {
/* not a "limit=" name: the resource names the limit itself */
name = (known != NULL) ? known : r->name;
name_len = (name != NULL) ? strlen(name) : 0;
scope = "";
} else if (known != NULL) {
name = known;
name_len = strlen(known);
}
charge_resolve(list_vres(scope_limits(rec), r, r->name), r->name, &c);
if (rec->scope_w != NULL) {
/* host-limit vres names do not encode the host */
host = (rec->scope_w->name != NULL) ? rec->scope_w->name : "";
host_len = strlen(host);
}
scope_len = strlen(scope);
scope_len = strlen(c.scope);
/* "<name>\0<scope>[,]h=<host>\0" */
*buf = (char*)malloc(name_len + scope_len + host_len + 5);
*buf = (char*)malloc(c.name_len + scope_len + host_len + 5);
if (*buf == NULL) {
return false;
}
p = *buf;
memcpy(p, (name != NULL) ? name : "", name_len);
p[name_len] = '\0';
row->limit = p;
p += name_len + 1;
memcpy(p, (c.name != NULL) ? c.name : "", c.name_len);
p[c.name_len] = '\0';
row->limit = p;
p += c.name_len + 1;
row->scope = p;
for (k = 0; k < scope_len; k++) {
*p++ = (scope[k] == '/') ? ',' : scope[k];
*p++ = (c.scope[k] == '/') ? ',' : c.scope[k];
}
if (host != NULL) {
if (scope_len > 0) {
@ -328,10 +361,7 @@ static bool limit_row(const xce_sched_trace_rec_t* rec, const xce_resource_t* r,
}
*p = '\0';
row->limited_res = NULL;
if ((v != NULL) && (v->res != NULL)) {
row->limited_res = v->res->name;
}
row->limited_res = c.limited_res;
return true;
}
@ -508,3 +538,528 @@ void sched_trace_capture_attempt_reasons(xce_sched_trace_rec_t* rec,
free((void*)hosts);
free((void*)res);
}
/* ---- dispatch outcome (6.1-6.3, 9.4) ---------------------------------- */
/* True iff the allocated resource `resname` is a limit charge: its
* registered resource has XCE_RESF_LIMIT, or, when the name is not
* registered in this replay, it has the "limit=" prefix. */
static bool is_limit_charge(xce_sched_core_t* core, const char* resname)
{
xce_resource_t* r = NULL;
if ((core != NULL) && (core->resmgr != NULL)
&& (key_find(core->resmgr, (void*)resname, (void**)&r) >= 0)
&& (r != NULL)) {
return (r->flags & XCE_RESF_LIMIT) != 0;
}
return strncmp(resname, VRES_PREFIX, VRES_PREFIX_LEN) == 0;
}
/* The host limits matched to the job on the worker named `host`, or NULL
* (no schedobj, unknown worker, none matched). */
static const xlist_t* host_limits(xce_sched_core_t* core,
const xce_sched_obj_t* obj, const char* host)
{
xce_sched_worker_t* w = NULL;
if ((obj == NULL) || (core == NULL) || (core->workers == NULL)
|| (host == NULL)
|| (key_find(core->workers, (void*)host, (void**)&w) < 0)
|| (w == NULL)) {
return NULL;
}
return worker_limits(obj, w->idx);
}
/* Stores allocated amount `a` as a limit charge, resolved through `limits`;
* a host charge (`host` non-NULL) gets the host appended to its scope. */
static bool charge_row(xce_sched_trace_rec_t* rec, const xlist_t* limits,
const xce_alloc_t* a, const char* host,
sched_trace_alloc_row_t* out)
{
charge_ref_t c;
charge_resolve(list_vres(limits, NULL, a->resname), a->resname, &c);
out->res = sched_trace_arena_strdup(rec, a->resname);
out->value = a->value;
out->limit = copy_prefix(rec, c.name, c.name_len);
out->scope = copy_scope(rec, c.scope, host);
out->limited_res = sched_trace_arena_strdup(rec, c.limited_res);
return (out->res != NULL) && (out->limit != NULL) && (out->scope != NULL)
&& ((c.limited_res == NULL) || (out->limited_res != NULL));
}
/* One allocation list (a host's, or the shared one) as resource rows and
* limit charges, each in allocation order. False on an arena failure. */
static bool alloc_rows(xce_sched_trace_rec_t* rec, xce_sched_core_t* core,
const xlist_t* limits, int n, xce_alloc_t** allocs,
const char* host, int* nres,
sched_trace_alloc_row_t** res, int* ncharges,
sched_trace_alloc_row_t** charges)
{
size_t sz = sizeof(**res) * (size_t)n;
int i;
if ((n <= 0) || (allocs == NULL)) {
return true;
}
*res = (sched_trace_alloc_row_t*)sched_trace_arena_alloc(rec, sz);
*charges = (sched_trace_alloc_row_t*)sched_trace_arena_alloc(rec, sz);
if ((*res == NULL) || (*charges == NULL)) {
return false;
}
for (i = 0; i < n; i++) {
const xce_alloc_t* a = allocs[i];
sched_trace_alloc_row_t* row;
if ((a == NULL) || (a->resname == NULL)) {
continue;
}
if (is_limit_charge(core, a->resname)) {
if (!charge_row(rec, limits, a, host, &(*charges)[(*ncharges)++])) {
return false;
}
continue;
}
row = &(*res)[(*nres)++];
row->res = sched_trace_arena_strdup(rec, a->resname);
row->value = a->value;
if (row->res == NULL) {
return false;
}
}
return true;
}
/* A total over the omitted hosts, keyed by resource name or, for a charge,
* by limit name and scope. Scratch memory; the strings are the sources'. */
typedef struct {
const char* res; /* resource total; NULL: a limit total */
charge_ref_t c; /* limit total */
double value;
} agg_t;
static void agg_add(agg_t* agg, int* n, const char* res,
const charge_ref_t* c, double value)
{
int i;
for (i = 0; i < *n; i++) {
if ((res != NULL)
? (strcmp(agg[i].res, res) == 0)
: ((agg[i].c.name_len == c->name_len)
&& (strncmp(agg[i].c.name, c->name, c->name_len) == 0)
&& (strcmp(agg[i].c.scope, c->scope) == 0))) {
break;
}
}
if (i == *n) {
memset(&agg[i], 0, sizeof(agg[i]));
agg[i].res = res;
if (c != NULL) {
agg[i].c = *c;
}
(*n)++;
}
agg[i].value += value;
if ((c != NULL) && (agg[i].c.limited_res == NULL)) {
agg[i].c.limited_res = c->limited_res;
}
}
/* Copies `n` totals into the arena as rows (a charge's scope has no
* host part: it totals hosts). */
static bool agg_store(xce_sched_trace_rec_t* rec, const agg_t* agg, int n,
int* nout, sched_trace_alloc_row_t** out)
{
int i;
if (n == 0) {
return true;
}
*out = (sched_trace_alloc_row_t*)sched_trace_arena_alloc(
rec, sizeof(**out) * (size_t)n);
if (*out == NULL) {
return false;
}
for (i = 0; i < n; i++) {
sched_trace_alloc_row_t* row = &(*out)[i];
row->value = agg[i].value;
if (agg[i].res != NULL) {
row->res = sched_trace_arena_strdup(rec, agg[i].res);
if (row->res == NULL) {
return false;
}
continue;
}
row->limit = copy_prefix(rec, agg[i].c.name, agg[i].c.name_len);
row->scope = copy_scope(rec, agg[i].c.scope, NULL);
row->limited_res = sched_trace_arena_strdup(rec, agg[i].c.limited_res);
if ((row->limit == NULL) || (row->scope == NULL)) {
return false;
}
}
*nout = n;
return true;
}
/* Totals of the hosts from index `from` on (9.4): per resource without the
* limit charges, and per limit. False on an allocation failure. */
static bool aggregate_hosts(xce_sched_trace_rec_t* rec, xce_sched_core_t* core,
const xce_sched_obj_t* obj,
const xce_job_alloc_t* alloc, int from)
{
sched_trace_dispatch_out_t* d = &rec->dispatch;
size_t total = 0;
agg_t* res;
agg_t* chg;
int nres = 0;
int nchg = 0;
bool ok;
int i;
int k;
for (i = from; i < alloc->nhosts; i++) {
if ((alloc->allocs[i] != NULL) && (alloc->allocs[i]->allocs != NULL)
&& (alloc->allocs[i]->nallocs > 0)) {
total += (size_t)alloc->allocs[i]->nallocs;
}
}
res = (agg_t*)malloc(sizeof(*res) * (total + 1));
chg = (agg_t*)malloc(sizeof(*chg) * (total + 1));
ok = (res != NULL) && (chg != NULL);
for (i = from; ok && (i < alloc->nhosts); i++) {
const xce_host_alloc_t* h = alloc->allocs[i];
const xlist_t* limits;
if ((h == NULL) || (h->allocs == NULL)) {
continue;
}
limits = host_limits(core, obj, h->host);
for (k = 0; k < h->nallocs; k++) {
const xce_alloc_t* a = h->allocs[k];
charge_ref_t c;
if ((a == NULL) || (a->resname == NULL)) {
continue;
}
if (!is_limit_charge(core, a->resname)) {
agg_add(res, &nres, a->resname, NULL, a->value);
continue;
}
charge_resolve(list_vres(limits, NULL, a->resname), a->resname,
&c);
agg_add(chg, &nchg, NULL, &c, a->value);
}
}
ok = ok && agg_store(rec, res, nres, &d->nagg_res, &d->agg_res)
&& agg_store(rec, chg, nchg, &d->nagg_charges, &d->agg_charges);
free(res);
free(chg);
return ok;
}
void sched_trace_capture_dispatch(xce_sched_trace_rec_t* rec,
xce_sched_core_t* core,
const xce_sched_obj_t* obj,
const xce_job_alloc_t* alloc)
{
sched_trace_dispatch_out_t* d = &rec->dispatch;
int from = -1;
int nh;
int i;
memset(d, 0, sizeof(*d));
rec->outcome = SCHED_TRACE_OUTCOME_DISPATCH;
if (alloc == NULL) {
return;
}
if (!alloc_rows(rec, core, (obj != NULL) ? &obj->glimits : NULL,
alloc->nres_shared, alloc->allocs_shared, NULL,
&d->nshared, &d->shared, &d->nshared_charges,
&d->shared_charges)) {
goto fail;
}
nh = (alloc->allocs != NULL) ? alloc->nhosts : 0;
if (nh > 0) {
int keep = (nh < XCE_TRACE_OUTCOME_HOSTS) ? nh : XCE_TRACE_OUTCOME_HOSTS;
d->hosts = (sched_trace_host_out_t*)sched_trace_arena_alloc(
rec, sizeof(*d->hosts) * (size_t)keep);
if (d->hosts == NULL) {
goto fail;
}
}
for (i = 0; i < nh; i++) {
const xce_host_alloc_t* h = alloc->allocs[i];
sched_trace_host_out_t* ho;
if (h == NULL) {
continue;
}
if (d->nhosts == XCE_TRACE_OUTCOME_HOSTS) {
if (from < 0) {
from = i;
}
d->omitted_hosts++;
continue;
}
ho = &d->hosts[d->nhosts++];
ho->host = sched_trace_arena_strdup(rec, h->host);
if (((h->host != NULL) && (ho->host == NULL))
|| !alloc_rows(rec, core, host_limits(core, obj, h->host),
h->nallocs, h->allocs,
(h->host != NULL) ? h->host : "", &ho->nres,
&ho->res, &ho->ncharges, &ho->charges)) {
goto fail;
}
}
if ((from >= 0) && !aggregate_hosts(rec, core, obj, alloc, from)) {
goto fail;
}
return;
fail:
rec->incomplete = true; /* design "Error Strategy" */
}
xce_sched_trace_rec_t*
sched_trace_rec_reconstructed(const xce_sched_trace_ctl_t* ctl,
xce_sched_core_t* core, uint64_t jobid,
const xce_sched_obj_t* obj,
const xce_job_alloc_t* allocs)
{
xce_sched_trace_rec_t* rec;
rec = sched_trace_rec_new(ctl, jobid, XCE_TRACE_REC_RECONSTRUCTED);
if (rec != NULL) {
sched_trace_capture_dispatch(rec, core, obj, allocs);
}
return rec;
}
/* ---- pend outcome (6.4, 9.4) ------------------------------------------ */
/* A set of host names, open addressing over a power-of-two table. Scratch
* memory; the names are the source's. */
typedef struct {
const char** slot;
size_t mask;
int n;
} host_set_t;
static bool host_set_init(host_set_t* s, int nmax)
{
size_t cap = 16;
while (cap < (size_t)nmax * 2 + 2) {
cap <<= 1;
}
s->slot = (const char**)calloc(cap, sizeof(*s->slot));
s->mask = cap - 1;
s->n = 0;
return s->slot != NULL;
}
/* True iff `h` is in the set; when absent and `insert`, adds it. */
static bool host_set_has(host_set_t* s, const char* h, bool insert)
{
size_t i = 2166136261u; /* FNV-1a */
const unsigned char* p;
for (p = (const unsigned char*)h; *p != '\0'; p++) {
i = (i ^ *p) * 16777619u;
}
for (i &= s->mask; s->slot[i] != NULL; i = (i + 1) & s->mask) {
if (strcmp(s->slot[i], h) == 0) {
return true;
}
}
if (insert) {
s->slot[i] = h;
s->n++;
}
return false;
}
typedef struct {
host_set_t kept; /* the first 50 distinct hosts */
host_set_t omitted; /* the hosts after them */
sched_trace_reason_count_t* codes; /* omitted entries per code */
int ncodes;
} pend_build_t;
/* True iff an entry naming `host` is kept: it names no host, or one of the
* first 50 distinct hosts. Otherwise it is counted under its code. */
static bool pend_keep(pend_build_t* b, const char* host, int code)
{
if ((host == NULL) || (host[0] == '\0')
|| host_set_has(&b->kept, host, false)) {
return true;
}
if (b->kept.n < XCE_TRACE_OUTCOME_HOSTS) {
(void)host_set_has(&b->kept, host, true);
return true;
}
(void)host_set_has(&b->omitted, host, true);
count_code(b->codes, &b->ncodes, code);
return false;
}
/* Arena copies of the kept host entries (NULL entries stay NULL). */
static bool copy_host_reasons(xce_sched_trace_rec_t* rec,
const xce_job_preason_t* r, const bool* keep,
int nh, xce_job_preason_t* c)
{
int i;
for (i = 0; i < nh; i++) {
c->nhost_reasons += keep[i] ? 1 : 0;
}
if (c->nhost_reasons == 0) {
return true;
}
c->host_based_reasons = (xce_host_reason_t**)sched_trace_arena_alloc(
rec, sizeof(*c->host_based_reasons) * (size_t)c->nhost_reasons);
if (c->host_based_reasons == NULL) {
return false;
}
c->nhost_reasons = 0;
for (i = 0; i < nh; i++) {
const xce_host_reason_t* src = r->host_based_reasons[i];
xce_host_reason_t* dst = NULL;
if (!keep[i]) {
continue;
}
if (src != NULL) {
dst = (xce_host_reason_t*)sched_trace_arena_alloc(rec, sizeof(*dst));
if (dst == NULL) {
return false;
}
dst->reason = src->reason;
dst->host = sched_trace_arena_strdup(rec, src->host);
if ((src->host != NULL) && (dst->host == NULL)) {
return false;
}
}
c->host_based_reasons[c->nhost_reasons++] = dst;
}
return true;
}
/* Arena copies of the kept resource entries (NULL entries stay NULL). */
static bool copy_res_reasons(xce_sched_trace_rec_t* rec,
const xce_job_preason_t* r, const bool* keep,
int nr, xce_job_preason_t* c)
{
int i;
for (i = 0; i < nr; i++) {
c->nres_reasons += keep[i] ? 1 : 0;
}
if (c->nres_reasons == 0) {
return true;
}
c->res_based_reasons = (xce_res_reason_t**)sched_trace_arena_alloc(
rec, sizeof(*c->res_based_reasons) * (size_t)c->nres_reasons);
if (c->res_based_reasons == NULL) {
return false;
}
c->nres_reasons = 0;
for (i = 0; i < nr; i++) {
const xce_res_reason_t* src = r->res_based_reasons[i];
xce_res_reason_t* dst = NULL;
if (!keep[i]) {
continue;
}
if (src != NULL) {
dst = (xce_res_reason_t*)sched_trace_arena_alloc(rec, sizeof(*dst));
if (dst == NULL) {
return false;
}
dst->reason = src->reason;
dst->host = sched_trace_arena_strdup(rec, src->host);
dst->resource = sched_trace_arena_strdup(rec, src->resource);
if (((src->host != NULL) && (dst->host == NULL))
|| ((src->resource != NULL) && (dst->resource == NULL))) {
return false;
}
}
c->res_based_reasons[c->nres_reasons++] = dst;
}
return true;
}
void sched_trace_capture_pend(xce_sched_trace_rec_t* rec,
const xce_pendreason_t* reason)
{
sched_trace_pend_out_t* p = &rec->pend;
const xce_job_preason_t* r = (reason != NULL) ? reason->reasons : NULL;
pend_build_t b;
bool* keep;
int nh;
int nr;
int i;
bool ok;
memset(p, 0, sizeof(*p));
rec->outcome = SCHED_TRACE_OUTCOME_PEND;
if (r == NULL) {
return;
}
nh = ((r->host_based_reasons != NULL) && (r->nhost_reasons > 0))
? r->nhost_reasons : 0;
nr = ((r->res_based_reasons != NULL) && (r->nres_reasons > 0))
? r->nres_reasons : 0;
/* which entries are kept, in one pass over the source order */
memset(&b, 0, sizeof(b));
keep = (bool*)malloc(sizeof(*keep) * (size_t)(nh + nr + 1));
b.codes = (sched_trace_reason_count_t*)malloc(sizeof(*b.codes)
* (size_t)(nh + nr + 1));
ok = (keep != NULL) && (b.codes != NULL)
&& host_set_init(&b.kept, XCE_TRACE_OUTCOME_HOSTS)
&& host_set_init(&b.omitted, nh + nr);
for (i = 0; ok && (i < nh); i++) {
const xce_host_reason_t* e = r->host_based_reasons[i];
keep[i] = (e == NULL) || pend_keep(&b, e->host, e->reason);
}
for (i = 0; ok && (i < nr); i++) {
const xce_res_reason_t* e = r->res_based_reasons[i];
keep[nh + i] = (e == NULL) || pend_keep(&b, e->host, e->reason);
}
if (ok) {
p->reasons = (xce_job_preason_t*)sched_trace_arena_alloc(
rec, sizeof(*p->reasons));
ok = (p->reasons != NULL);
}
if (ok) {
p->reasons->main_reason = r->main_reason;
ok = copy_host_reasons(rec, r, keep, nh, p->reasons)
&& copy_res_reasons(rec, r, keep + nh, nr, p->reasons);
}
if (ok && (b.ncodes > 0)) {
p->omitted_codes = (sched_trace_reason_count_t*)sched_trace_arena_memdup(
rec, b.codes, sizeof(*b.codes) * (size_t)b.ncodes);
ok = (p->omitted_codes != NULL);
p->ncodes = ok ? b.ncodes : 0;
}
p->omitted_hosts = b.omitted.n;
if (!ok) {
rec->incomplete = true; /* design "Error Strategy" */
}
free(keep);
free(b.codes);
free((void*)b.kept.slot);
free((void*)b.omitted.slot);
}

545
src/scheduler/testsuite/test_trace_capture.c
View file

@ -27,6 +27,19 @@
* unmet rows, stored after the host's verdict line, and they set its
* SHORTFALL and/or LIMIT verdict.
*
* Task 2.3 covers the decision outcomes (Requirements 6.1-6.4, 9.4) and the
* reconstructed-dispatch record (2.5), over the same fixture plus a workers
* table:
* - a dispatch keeps per-host and shared resource rows apart from the
* limit charges, each charge resolved through the job's matched limits
* (or, unresolved, parsed from its name with an unknown limited
* resource: the raw-number fallback);
* - past 50 hosts, the rest become per-resource and per-limit totals and
* an omitted-host count; a pend keeps the first 50 hosts' entries and
* the host-less ones and counts the rest per pend-reason code;
* - with 50 or fewer hosts the pend copy equals its source under
* job_preason_eq(), the equality libxce defines for pend reasons.
*
* Links src/sched_trace.c and src/sched_trace_capture.c, plus libxce for
* the pend-reason, resmgr and container helpers.
*/
@ -525,6 +538,7 @@ typedef struct {
xce_vres_t vres_lic;
xce_vres_t vres_host;
xce_vres_t vres_hcap;
xce_sched_worker_t w_bare; /* n02: registered, no matched host limits */
} rows_fx_t;
#define ROWS_HOST_IDX 2
@ -593,6 +607,25 @@ static rows_fx_t* rows_fx_new(void)
fx->w.name = "n01";
fx->w.idx = ROWS_HOST_IDX;
/* workers by name, for the dispatch capture's host-limit lookup */
{
kvmgr_params_t wk_params = { .key_type = KVMGR_KEY_STRING,
.nslots = 16,
.delta = 8,
.datakiller = NULL };
fx->core.workers = makeInstance(&xkvmgr_type, &wk_params);
TEST_ASSERT_NOT_NULL(fx->core.workers);
fx->w_bare.name = "n02";
TEST_ASSERT_TRUE(key_add(fx->core.workers, (void*)fx->w.name, &fx->w)
>= 0);
TEST_ASSERT_TRUE(key_add(fx->core.workers, (void*)fx->w_bare.name,
&fx->w_bare)
>= 0);
fx->w.idx = ROWS_HOST_IDX; /* key_add's slot is not the idx */
fx->w_bare.idx = ROWS_HOST_IDX + 1;
}
return fx;
}
@ -1023,6 +1056,510 @@ static void test_check_rows_host_rows_count_against_budget(void)
rows_fx_free(fx);
}
/* ---- decision outcomes (6.1-6.4, 9.4) and reconstructed records (2.5) - */
typedef struct {
const char* res;
double value;
} t_av_t;
static xce_alloc_t* t_alloc(const char* res, double value)
{
xce_alloc_t* a = (xce_alloc_t*)calloc(1, sizeof(*a));
TEST_ASSERT_NOT_NULL(a);
a->resname = strdup(res);
a->reserve = strdup("");
a->value = value;
return a;
}
static xce_alloc_t** t_allocs(int n, const t_av_t* av)
{
xce_alloc_t** v = (xce_alloc_t**)calloc((size_t)n + 1, sizeof(*v));
int i;
TEST_ASSERT_NOT_NULL(v);
for (i = 0; i < n; i++) {
v[i] = t_alloc(av[i].res, av[i].value);
}
return v;
}
static xce_host_alloc_t* t_host(const char* host, int n, const t_av_t* av)
{
xce_host_alloc_t* h = (xce_host_alloc_t*)calloc(1, sizeof(*h));
TEST_ASSERT_NOT_NULL(h);
h->host = strdup(host);
h->nallocs = n;
h->allocs = t_allocs(n, av);
return h;
}
static void t_free_allocs(int n, xce_alloc_t** v)
{
int i;
for (i = 0; (v != NULL) && (i < n); i++) {
if (v[i] != NULL) {
free(v[i]->resname);
free(v[i]->reserve);
free(v[i]);
}
}
free(v);
}
static void t_free_job(xce_job_alloc_t* j)
{
int i;
for (i = 0; i < j->nhosts; i++) {
t_free_allocs(j->allocs[i]->nallocs, j->allocs[i]->allocs);
free(j->allocs[i]->host);
free(j->allocs[i]);
}
free(j->allocs);
t_free_allocs(j->nres_shared, j->allocs_shared);
free(j);
}
static xce_job_alloc_t* t_job(int nhosts)
{
xce_job_alloc_t* j = (xce_job_alloc_t*)calloc(1, sizeof(*j));
TEST_ASSERT_NOT_NULL(j);
j->jobid = 0x77;
j->nhosts = nhosts;
j->allocs = (xce_host_alloc_t**)calloc((size_t)nhosts + 1,
sizeof(*j->allocs));
TEST_ASSERT_NOT_NULL(j->allocs);
return j;
}
static void assert_alloc_row(const sched_trace_alloc_row_t* row,
const char* res, double value)
{
TEST_ASSERT_EQUAL_STRING(res, row->res);
TEST_ASSERT_EQUAL_FLOAT((float)value, (float)row->value);
TEST_ASSERT_NULL(row->limit);
TEST_ASSERT_NULL(row->scope);
TEST_ASSERT_NULL(row->limited_res);
}
static void assert_charge(const sched_trace_alloc_row_t* row, const char* res,
double value, const char* limit, const char* scope,
const char* limited)
{
if (res == NULL) {
TEST_ASSERT_NULL(row->res);
} else {
TEST_ASSERT_EQUAL_STRING(res, row->res);
}
TEST_ASSERT_EQUAL_FLOAT((float)value, (float)row->value);
TEST_ASSERT_EQUAL_STRING(limit, row->limit);
TEST_ASSERT_EQUAL_STRING(scope, row->scope);
if (limited == NULL) {
TEST_ASSERT_NULL(row->limited_res);
} else {
TEST_ASSERT_EQUAL_STRING(limited, row->limited_res);
}
}
/* Per-host rows and shared rows, limit charges listed apart: resolved
* through obj->hlimits[w->idx] (w by host name) and obj->glimits; a
* registered limit flag or, for an unregistered name, the "limit=" prefix
* makes a charge; an unresolved charge keeps its parsed name and scope and
* an unknown limited resource, so it renders as a raw number (6.1-6.3). */
static void test_dispatch_separates_limit_charges(void)
{
rows_fx_t* fx = rows_fx_new();
xce_sched_trace_ctl_t ctl = make_ctl("fcfs");
xce_sched_trace_rec_t* rec;
xce_job_alloc_t* job = t_job(2);
const t_av_t h1[] = { { "mem", 8 }, { "limit=host_slots/u=alice", 4 },
{ "slots", 4 }, { "limit=hcap", 4 } };
const t_av_t h2[] = { { "mem", 2 }, { "limit=host_slots/u=alice", 1 },
{ "gpu", 1 }, { "limit=ghost/u=alice", 1 } };
const t_av_t sh[] = { { "lic", 1 },
{ "limit=lic_cap/u=alice/q=normal", 1 },
{ "limit=site_total/p=proj1", 1 } };
const sched_trace_dispatch_out_t* d;
job->allocs[0] = t_host("n01", 4, h1);
job->allocs[1] = t_host("n02", 4, h2);
job->nres_shared = 3;
job->allocs_shared = t_allocs(3, sh);
rec = sched_trace_rec_new(&ctl, 0x77, XCE_TRACE_REC_DECISION);
TEST_ASSERT_EQUAL_INT(SCHED_TRACE_OUTCOME_NONE, rec->outcome);
sched_trace_capture_dispatch(rec, &fx->core, &fx->obj, job);
TEST_ASSERT_EQUAL_INT(SCHED_TRACE_OUTCOME_DISPATCH, rec->outcome);
TEST_ASSERT_EQUAL_INT(0, rec->nentries); /* outside the line budget */
d = &rec->dispatch;
TEST_ASSERT_EQUAL_INT(2, d->nhosts);
TEST_ASSERT_EQUAL_INT(0, d->omitted_hosts);
TEST_ASSERT_EQUAL_INT(0, d->nagg_res);
TEST_ASSERT_EQUAL_INT(0, d->nagg_charges);
/* n01: resources in allocation order, charges resolved through its
* host limits, the host appended to the scope */
TEST_ASSERT_EQUAL_STRING("n01", d->hosts[0].host);
TEST_ASSERT_EQUAL_INT(2, d->hosts[0].nres);
assert_alloc_row(&d->hosts[0].res[0], "mem", 8);
assert_alloc_row(&d->hosts[0].res[1], "slots", 4);
TEST_ASSERT_EQUAL_INT(2, d->hosts[0].ncharges);
assert_charge(&d->hosts[0].charges[0], "limit=host_slots/u=alice", 4,
"host_slots", "u=alice,h=n01", "slots");
assert_charge(&d->hosts[0].charges[1], "limit=hcap", 4, "hcap", "h=n01",
"slots");
/* n02 has no matched host limits: the charge is unresolved; "gpu" is
* not registered and has no limit prefix: a resource; "limit=ghost..."
* is not registered: a charge by its prefix */
TEST_ASSERT_EQUAL_STRING("n02", d->hosts[1].host);
TEST_ASSERT_EQUAL_INT(2, d->hosts[1].nres);
assert_alloc_row(&d->hosts[1].res[0], "mem", 2);
assert_alloc_row(&d->hosts[1].res[1], "gpu", 1);
TEST_ASSERT_EQUAL_INT(2, d->hosts[1].ncharges);
assert_charge(&d->hosts[1].charges[0], "limit=host_slots/u=alice", 1,
"host_slots", "u=alice,h=n02", NULL);
assert_charge(&d->hosts[1].charges[1], "limit=ghost/u=alice", 1, "ghost",
"u=alice,h=n02", NULL);
/* shared: once for the job, charges resolved through glimits */
TEST_ASSERT_EQUAL_INT(1, d->nshared);
assert_alloc_row(&d->shared[0], "lic", 1);
TEST_ASSERT_EQUAL_INT(2, d->nshared_charges);
assert_charge(&d->shared_charges[0], "limit=lic_cap/u=alice/q=normal", 1,
"lic_cap", "u=alice,q=normal", "lic");
assert_charge(&d->shared_charges[1], "limit=site_total/p=proj1", 1,
"site_total", "p=proj1", NULL);
/* a self-contained copy */
TEST_ASSERT_TRUE(in_arena(rec, d->hosts));
TEST_ASSERT_TRUE(in_arena(rec, d->hosts[0].host));
TEST_ASSERT_TRUE(in_arena(rec, d->hosts[0].res[0].res));
TEST_ASSERT_TRUE(in_arena(rec, d->hosts[0].charges[0].limit));
TEST_ASSERT_TRUE(in_arena(rec, d->hosts[0].charges[0].scope));
TEST_ASSERT_TRUE(in_arena(rec, d->hosts[0].charges[0].limited_res));
TEST_ASSERT_TRUE(in_arena(rec, d->shared_charges[0].limited_res));
t_free_job(job);
rows_fx_free(fx);
TEST_ASSERT_EQUAL_STRING("n01", d->hosts[0].host);
TEST_ASSERT_EQUAL_STRING("u=alice,h=n01", d->hosts[0].charges[0].scope);
TEST_ASSERT_EQUAL_STRING("lic", d->shared_charges[0].limited_res);
TEST_ASSERT_FALSE(rec->incomplete);
sched_trace_rec_free(rec);
}
/* No schedobj (a job this replay never evaluated): every charge is still
* listed as a charge, by its flag, with the name and scope parsed from the
* resource name and the amount left as a raw number (design "TraceCapture",
* the raw-number fallback). */
static void test_dispatch_raw_number_fallback(void)
{
rows_fx_t* fx = rows_fx_new();
xce_sched_trace_ctl_t ctl = make_ctl("fcfs");
xce_sched_trace_rec_t* rec;
xce_job_alloc_t* job = t_job(1);
const t_av_t h1[] = { { "limit=host_slots/u=alice", 3 }, { "mem", 1 } };
const t_av_t sh[] = { { "limit=lic_cap/u=alice/q=normal", 2 } };
const sched_trace_dispatch_out_t* d;
job->allocs[0] = t_host("n01", 2, h1);
job->nres_shared = 1;
job->allocs_shared = t_allocs(1, sh);
rec = sched_trace_rec_new(&ctl, 0x77, XCE_TRACE_REC_DECISION);
sched_trace_capture_dispatch(rec, &fx->core, NULL, job);
d = &rec->dispatch;
TEST_ASSERT_EQUAL_INT(1, d->hosts[0].nres);
assert_alloc_row(&d->hosts[0].res[0], "mem", 1);
TEST_ASSERT_EQUAL_INT(1, d->hosts[0].ncharges);
assert_charge(&d->hosts[0].charges[0], "limit=host_slots/u=alice", 3,
"host_slots", "u=alice,h=n01", NULL);
TEST_ASSERT_EQUAL_INT(0, d->nshared);
TEST_ASSERT_EQUAL_INT(1, d->nshared_charges);
assert_charge(&d->shared_charges[0], "limit=lic_cap/u=alice/q=normal", 2,
"lic_cap", "u=alice,q=normal", NULL);
sched_trace_rec_free(rec);
/* no resmgr entry and no prefix either way: no core at all */
rec = sched_trace_rec_new(&ctl, 0x77, XCE_TRACE_REC_DECISION);
sched_trace_capture_dispatch(rec, NULL, NULL, job);
d = &rec->dispatch;
TEST_ASSERT_EQUAL_INT(1, d->hosts[0].nres);
TEST_ASSERT_EQUAL_INT(1, d->hosts[0].ncharges);
assert_charge(&d->hosts[0].charges[0], "limit=host_slots/u=alice", 3,
"host_slots", "u=alice,h=n01", NULL);
sched_trace_rec_free(rec);
t_free_job(job);
rows_fx_free(fx);
}
/* A reconstructed dispatch (2.5): a RECONSTRUCTED record with the session
* header, no entries and the dispatch outcome, its charges resolved the same
* way. */
static void test_reconstructed_record_holds_dispatch(void)
{
rows_fx_t* fx = rows_fx_new();
xce_sched_trace_ctl_t ctl = make_ctl("fcfs");
xce_sched_trace_rec_t* rec;
xce_job_alloc_t* job = t_job(1);
const t_av_t h1[] = { { "slots", 2 }, { "limit=host_slots/u=alice", 2 } };
job->allocs[0] = t_host("n01", 2, h1);
TEST_ASSERT_NULL(
sched_trace_rec_reconstructed(NULL, &fx->core, 0x99, &fx->obj, job));
rec = sched_trace_rec_reconstructed(&ctl, &fx->core, 0x99, &fx->obj, job);
TEST_ASSERT_NOT_NULL(rec);
TEST_ASSERT_EQUAL_INT(XCE_TRACE_REC_RECONSTRUCTED, rec->kind);
TEST_ASSERT_EQUAL_UINT64(0x99, rec->jobid);
TEST_ASSERT_EQUAL_UINT64(2, rec->session_no);
TEST_ASSERT_EQUAL_INT(0, rec->nentries);
TEST_ASSERT_EQUAL_INT(SCHED_TRACE_OUTCOME_DISPATCH, rec->outcome);
TEST_ASSERT_EQUAL_INT(1, rec->dispatch.nhosts);
assert_alloc_row(&rec->dispatch.hosts[0].res[0], "slots", 2);
assert_charge(&rec->dispatch.hosts[0].charges[0],
"limit=host_slots/u=alice", 2, "host_slots", "u=alice,h=n01",
"slots");
t_free_job(job);
rows_fx_free(fx);
TEST_ASSERT_EQUAL_STRING("n01", rec->dispatch.hosts[0].host);
sched_trace_rec_free(rec);
/* a job with no schedobj: still built, charges unresolved */
fx = rows_fx_new();
job = t_job(1);
job->allocs[0] = t_host("n01", 2, h1);
rec = sched_trace_rec_reconstructed(&ctl, &fx->core, 0x99, NULL, job);
TEST_ASSERT_NOT_NULL(rec);
TEST_ASSERT_NULL(rec->dispatch.hosts[0].charges[0].limited_res);
sched_trace_rec_free(rec);
t_free_job(job);
rows_fx_free(fx);
}
/* More than 50 hosts: the first 50 in detail, the rest as an omitted-host
* count, per-resource totals without the limits, and per-limit totals with
* the host-free scope (9.4). */
static void test_dispatch_aggregates_hosts_past_50(void)
{
rows_fx_t* fx = rows_fx_new();
xce_sched_trace_ctl_t ctl = make_ctl("fcfs");
xce_sched_trace_rec_t* rec;
int nh;
for (nh = XCE_TRACE_OUTCOME_HOSTS + 1; nh <= XCE_TRACE_OUTCOME_HOSTS + 3;
nh += 2) {
xce_job_alloc_t* job = t_job(nh);
const sched_trace_dispatch_out_t* d;
int i;
for (i = 0; i < nh; i++) {
char name[16];
t_av_t av[3] = { { "mem", 2 + i },
{ "limit=host_slots/u=alice", 1 },
{ "slots", 1 } };
snprintf(name, sizeof(name), "h%02d", i);
job->allocs[i] = t_host(name, 3, av);
}
rec = sched_trace_rec_new(&ctl, 0x77, XCE_TRACE_REC_DECISION);
sched_trace_capture_dispatch(rec, &fx->core, &fx->obj, job);
d = &rec->dispatch;
TEST_ASSERT_EQUAL_INT(XCE_TRACE_OUTCOME_HOSTS, d->nhosts);
TEST_ASSERT_EQUAL_STRING("h49", d->hosts[49].host);
TEST_ASSERT_EQUAL_INT(1, d->hosts[49].ncharges);
TEST_ASSERT_EQUAL_INT(nh - XCE_TRACE_OUTCOME_HOSTS, d->omitted_hosts);
TEST_ASSERT_EQUAL_INT(2, d->nagg_res);
if (nh == XCE_TRACE_OUTCOME_HOSTS + 1) {
assert_alloc_row(&d->agg_res[0], "mem", 52); /* h50 only */
} else {
assert_alloc_row(&d->agg_res[0], "mem", 52 + 53 + 54);
}
assert_alloc_row(&d->agg_res[1], "slots", nh - XCE_TRACE_OUTCOME_HOSTS);
TEST_ASSERT_EQUAL_INT(1, d->nagg_charges);
assert_charge(&d->agg_charges[0], NULL, nh - XCE_TRACE_OUTCOME_HOSTS,
"host_slots", "u=alice", NULL);
TEST_ASSERT_TRUE(in_arena(rec, d->agg_res[0].res));
TEST_ASSERT_TRUE(in_arena(rec, d->agg_charges[0].limit));
t_free_job(job);
TEST_ASSERT_EQUAL_STRING("mem", d->agg_res[0].res);
sched_trace_rec_free(rec);
}
/* exactly 50: nothing omitted */
{
xce_job_alloc_t* job = t_job(XCE_TRACE_OUTCOME_HOSTS);
int i;
for (i = 0; i < XCE_TRACE_OUTCOME_HOSTS; i++) {
char name[16];
t_av_t av[1] = { { "mem", 1 } };
snprintf(name, sizeof(name), "h%02d", i);
job->allocs[i] = t_host(name, 1, av);
}
rec = sched_trace_rec_new(&ctl, 0x77, XCE_TRACE_REC_DECISION);
sched_trace_capture_dispatch(rec, &fx->core, &fx->obj, job);
TEST_ASSERT_EQUAL_INT(XCE_TRACE_OUTCOME_HOSTS, rec->dispatch.nhosts);
TEST_ASSERT_EQUAL_INT(0, rec->dispatch.omitted_hosts);
TEST_ASSERT_EQUAL_INT(0, rec->dispatch.nagg_res);
sched_trace_rec_free(rec);
t_free_job(job);
}
rows_fx_free(fx);
}
static void test_dispatch_null_alloc_is_empty(void)
{
xce_sched_trace_ctl_t ctl = make_ctl("fcfs");
xce_sched_trace_rec_t* rec = sched_trace_rec_new(&ctl, 0x77,
XCE_TRACE_REC_DECISION);
sched_trace_capture_dispatch(rec, NULL, NULL, NULL);
TEST_ASSERT_EQUAL_INT(SCHED_TRACE_OUTCOME_DISPATCH, rec->outcome);
TEST_ASSERT_EQUAL_INT(0, rec->dispatch.nhosts);
TEST_ASSERT_EQUAL_INT(0, rec->dispatch.nshared);
TEST_ASSERT_FALSE(rec->incomplete);
sched_trace_rec_free(rec);
}
/* 50 or fewer hosts: the pend copy is the decision's set, equal under the
* scheduler's own pend-reason equality, deep-copied into the arena (6.4). */
static void test_pend_copy_equals_source(void)
{
xce_sched_trace_ctl_t ctl = make_ctl("fcfs");
xce_sched_trace_rec_t* rec = sched_trace_rec_new(&ctl, 0x55,
XCE_TRACE_REC_DECISION);
xce_pendreason_t* r = make_pendreason(0x55);
char* why = NULL;
int i;
add_job_preason(r->reasons, PREASON_BAD_RESREQ, NULL, NULL); /* main */
for (i = 0; i < XCE_TRACE_OUTCOME_HOSTS; i++) {
char name[16];
snprintf(name, sizeof(name), "h%02d", i);
add_job_preason(r->reasons, PREASON_MATCH_FAIL, name, NULL);
if ((i % 7) == 0) {
add_job_preason(r->reasons, PREASON_NO_RESOURCE, name, "mem");
}
}
add_job_preason(r->reasons, PREASON_NO_RESOURCE, NULL, "lic"); /* shared */
sched_trace_capture_pend(rec, r);
TEST_ASSERT_EQUAL_INT(SCHED_TRACE_OUTCOME_PEND, rec->outcome);
TEST_ASSERT_EQUAL_INT(0, rec->nentries);
TEST_ASSERT_NOT_NULL(rec->pend.reasons);
TEST_ASSERT_TRUE_MESSAGE(job_preason_eq(rec->pend.reasons, r->reasons,
&why),
(why != NULL) ? why : "");
TEST_ASSERT_EQUAL_INT(0, rec->pend.omitted_hosts);
TEST_ASSERT_EQUAL_INT(0, rec->pend.ncodes);
/* no aliasing: every level is the record's own */
TEST_ASSERT_TRUE(in_arena(rec, rec->pend.reasons));
TEST_ASSERT_TRUE(in_arena(rec, rec->pend.reasons->host_based_reasons));
TEST_ASSERT_TRUE(in_arena(rec, rec->pend.reasons->host_based_reasons[0]));
TEST_ASSERT_TRUE(
in_arena(rec, rec->pend.reasons->host_based_reasons[0]->host));
TEST_ASSERT_TRUE(in_arena(rec, rec->pend.reasons->res_based_reasons[0]));
TEST_ASSERT_TRUE(
in_arena(rec, rec->pend.reasons->res_based_reasons[0]->resource));
free_pendreason(r);
TEST_ASSERT_EQUAL_INT(PREASON_BAD_RESREQ, rec->pend.reasons->main_reason);
TEST_ASSERT_EQUAL_STRING("h49",
rec->pend.reasons->host_based_reasons[49]->host);
sched_trace_rec_free(rec);
/* main reason only, and an empty set: still equal */
r = make_pendreason(0x55);
rec = sched_trace_rec_new(&ctl, 0x55, XCE_TRACE_REC_DECISION);
sched_trace_capture_pend(rec, r);
TEST_ASSERT_TRUE(job_preason_eq(rec->pend.reasons, r->reasons, NULL));
sched_trace_rec_free(rec);
free_pendreason(r);
}
/* More than 50 distinct hosts: the main reason, host-less entries and
* every entry of the first 50 hosts (also later ones) are kept; the other
* hosts' entries are counted per pend-reason code (9.4). */
static void test_pend_aggregates_hosts_past_50(void)
{
xce_sched_trace_ctl_t ctl = make_ctl("fcfs");
xce_sched_trace_rec_t* rec = sched_trace_rec_new(&ctl, 0x55,
XCE_TRACE_REC_DECISION);
xce_pendreason_t* r = make_pendreason(0x55);
const xce_job_preason_t* c;
int i;
add_job_preason(r->reasons, PREASON_BAD_RESREQ, NULL, NULL);
for (i = 0; i < XCE_TRACE_OUTCOME_HOSTS + 1; i++) {
char name[16];
snprintf(name, sizeof(name), "h%02d", i);
add_job_preason(r->reasons, PREASON_MATCH_FAIL, name, NULL);
}
/* resource entries: a kept host, the omitted host, a host-less one */
add_job_preason(r->reasons, PREASON_NO_RESOURCE, "h03", "mem");
add_job_preason(r->reasons, PREASON_NO_RESOURCE, "h50", "mem");
add_job_preason(r->reasons, PREASON_NO_RESOURCE, NULL, "lic");
sched_trace_capture_pend(rec, r);
c = rec->pend.reasons;
TEST_ASSERT_NOT_NULL(c);
TEST_ASSERT_EQUAL_INT(PREASON_BAD_RESREQ, c->main_reason);
TEST_ASSERT_EQUAL_INT(XCE_TRACE_OUTCOME_HOSTS, c->nhost_reasons);
TEST_ASSERT_EQUAL_STRING("h00", c->host_based_reasons[0]->host);
TEST_ASSERT_EQUAL_STRING("h49", c->host_based_reasons[49]->host);
TEST_ASSERT_EQUAL_INT(2, c->nres_reasons);
TEST_ASSERT_EQUAL_STRING("h03", c->res_based_reasons[0]->host);
TEST_ASSERT_EQUAL_STRING("mem", c->res_based_reasons[0]->resource);
TEST_ASSERT_EQUAL_STRING("", c->res_based_reasons[1]->host); /* shared */
TEST_ASSERT_EQUAL_STRING("lic", c->res_based_reasons[1]->resource);
TEST_ASSERT_EQUAL_INT(PREASON_NO_RESOURCE, c->res_based_reasons[1]->reason);
TEST_ASSERT_EQUAL_INT(1, rec->pend.omitted_hosts);
TEST_ASSERT_EQUAL_INT(2, rec->pend.ncodes);
TEST_ASSERT_EQUAL_INT(PREASON_MATCH_FAIL, rec->pend.omitted_codes[0].code);
TEST_ASSERT_EQUAL_INT(1, rec->pend.omitted_codes[0].count);
TEST_ASSERT_EQUAL_INT(PREASON_NO_RESOURCE, rec->pend.omitted_codes[1].code);
TEST_ASSERT_EQUAL_INT(1, rec->pend.omitted_codes[1].count);
TEST_ASSERT_TRUE(in_arena(rec, rec->pend.omitted_codes));
free_pendreason(r);
TEST_ASSERT_EQUAL_STRING("h49", c->host_based_reasons[49]->host);
sched_trace_rec_free(rec);
}
static void test_pend_null_reason_has_no_set(void)
{
xce_sched_trace_ctl_t ctl = make_ctl("fcfs");
xce_sched_trace_rec_t* rec = sched_trace_rec_new(&ctl, 0x55,
XCE_TRACE_REC_DECISION);
xce_pendreason_t empty;
sched_trace_capture_pend(rec, NULL);
TEST_ASSERT_EQUAL_INT(SCHED_TRACE_OUTCOME_PEND, rec->outcome);
TEST_ASSERT_NULL(rec->pend.reasons);
sched_trace_rec_free(rec);
memset(&empty, 0, sizeof(empty));
rec = sched_trace_rec_new(&ctl, 0x55, XCE_TRACE_REC_DECISION);
sched_trace_capture_pend(rec, &empty);
TEST_ASSERT_NULL(rec->pend.reasons);
TEST_ASSERT_FALSE(rec->incomplete);
sched_trace_rec_free(rec);
}
int main(void)
{
xstring_table_init(1024);
@ -1048,5 +1585,13 @@ int main(void)
RUN_TEST(test_check_rows_host_summary_tallies_limit);
RUN_TEST(test_check_rows_scope_edges);
RUN_TEST(test_check_rows_host_rows_count_against_budget);
RUN_TEST(test_dispatch_separates_limit_charges);
RUN_TEST(test_dispatch_raw_number_fallback);
RUN_TEST(test_reconstructed_record_holds_dispatch);
RUN_TEST(test_dispatch_aggregates_hosts_past_50);
RUN_TEST(test_dispatch_null_alloc_is_empty);
RUN_TEST(test_pend_copy_equals_source);
RUN_TEST(test_pend_aggregates_hosts_past_50);
RUN_TEST(test_pend_null_reason_has_no_set);
return UNITY_END();
}