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

feat(scheduler-sim-decision-trace): store the latest trace per job and serve info trace

Browse source 
Task 4.2: a new trace replaces a job's earlier one; above 64 MiB the
oldest traces are evicted (never the newest) and marked discarded until
a newer capture; info trace renders a job's or an array's stored traces
in the current units mode, or reports discarded / no trace with the
job's state / unknown job.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
This commit is contained in:
Rongsong Shen 2026-10-08 13:50:02 +00:00
parent fe445e9453
commit 73416e3e91
Signed by untrusted user: rshen
GPG key ID: 80533536B0EB651C
6 changed files with 1130 additions and 1 deletions
Download patch file Download diff file Expand all files Collapse all files

3
.kiro/specs/scheduler-sim-decision-trace/tasks.md
View file

@ -194,7 +194,7 @@
- _Requirements: 2.1, 2.4, 2.5, 3.1, 3.2, 4.3, 5.1, 5.9, 6.1, 6.2, 6.3, 6.4, 7.6, 7.7, 9.2, 9.3, 9.4_
- _Boundary: TraceRenderer_
- [ ] 4.2 Store the latest trace per job and serve info trace, test first
- [x] 4.2 Store the latest trace per job and serve info trace, test first
- **Storing:** a new trace for a job replaces its earlier one.
- **Memory bound:** accounted from each record's size.
- At 64 MiB, the oldest traces are evicted, never the newest.
@ -365,3 +365,4 @@
- 3.3: the pre-filter (`sched_host_iter_next`) opens a host scope (`sched_trace_host_begin`) for every candidate and closes it with a verdict on each rejection: SELECT (pre-filter), INTERNAL(rc) for an uncommitted-fetch or check error, from-rows for a resource rejection (unreachable from the runner). A host the iterator returns is left in an open scope with no entry, so 3.4 must end or replace it for the accepted, already-claimed, snapshot-failure and per-host-error paths, or later rows attach to a stale host. A scan stopped with rc<0 calls `pos_scan_error`; exhaustion does not. `unsatisfy_` hands every row to `sched_trace_check_rows`. The sim thread's reconciliation reaches that hook but sees trace_rec NULL.
- 3.4: runner hooks are in alloc_runner.c. The 1.5 OPEN item is settled: `sched_trace_no_plan` counts `reqs_none_generated` once and sets `scope_no_plan`, after which `search_end` makes no statement and `resreq_begin`/`search_end` reset the flag. So a declined or generator-failed requirement has a NO_PLAN entry and no SEARCH_END line, and 4.1 must render the NO_PLAN line as "no plan generated: <reason>". The 3.3 open scope is closed: every returned candidate gets a fresh host scope and one verdict, and check_begin drops any stale host rows. When the iterator can't start, the position ends as SCAN_ERROR with err 0, so 4.1 must render a scan error with no code. trace.h is at 790 lines; split it before adding to it.
- 4.1: the renderer is `repl_trace_render(rec, units, out1, out2)`, declared in repl_internal.h, in repl_trace_render.c plus repl_trace_render_outcome.c, with a private repl_trace_render.h. It renders once per sink, so the bytes are identical. Amounts use `repl_format_numeric_resource_value` ("8.00 GB" in human mode) and times use `repl_format_ts_human`. Wording where the design is silent is fixed by the golden tests in test_trace_render.c, so 4.2/4.6/6.1 must reuse the renderer, never re-format. Pend-reason arrays may contain NULL entries (capture keeps them); every walk must skip them. The renderer is not yet in sched_sim_isolation_lib or the sim isolation scan list (5.1).
- 4.2: the store API is `repl_trace_store_put`, `repl_trace_store_bytes`, `repl_cmd_info_trace` and `repl_trace_store_fini` (an addition; nothing calls it at shutdown yet, so 4.5/4.6/5.2 must). `xce_repl_trace_state_t` (repl_internal.h) holds only the store fields; 4.3 adds next/cur/session flags to it. The 1.3 "file tag" is dropped: `info trace` is console-only (Req 8.1) and emission writes through the session's `cur` settings. `info trace` is not yet routed from repl_cmd_info (5.2). Open (6.2): an `id_list_add` OOM can silently drop array elements from a listing.

1
src/scheduler/meson.build
View file

@ -31,6 +31,7 @@ scheduler_srcs = [
'src/repl_params.c',
'src/repl_trace_render.c',
'src/repl_trace_render_outcome.c',
'src/repl_trace_store.c',
'src/schedobj.c',
'src/plugin.c',
'src/resource.c',

62
src/scheduler/src/repl_internal.h
View file

@ -2,6 +2,7 @@
#define _REPL_INTERNAL_H_ 1
#include "htab.h" /* the trace store's jobid tables */
#include "repl.h"
#include "utils.h" /* xce_units_mode_t (translate_units_mode) */
@ -154,6 +155,29 @@ typedef enum {
XCE_REPL_UNITS_BOTH
} xce_repl_units_mode_t;
/*
* TraceStore state (repl_trace_store.c, scheduler-sim-decision-trace spec,
* task 4.2, design.md "TraceStore"): the latest trace record per job, kept
* for `info trace`. `by_job` maps a jobid to its stored entry and
* `discarded` maps a jobid evicted under the memory bound to its eviction
* sequence number (9.5, 9.6). Entries form a capture-order list from
* `oldest` to `newest`; `total_bytes` is the sum of their
* sched_trace_rec_bytes(). Zero-initialized means empty: the tables are
* created on the first put. Read and written only on the sim thread.
*/
#define XCE_REPL_TRACE_STORE_BOUND ((size_t)64 * 1024 * 1024) /* 9.5 */
struct repl_trace_entry_s;
typedef struct {
htab_t by_job; /* jobid -> repl_trace_entry_s* */
htab_t discarded; /* jobid -> eviction seq */
bool tables_ready;
struct repl_trace_entry_s* oldest; /* capture-order list head */
struct repl_trace_entry_s* newest; /* ... and tail */
size_t total_bytes;
uint64_t seq; /* last capture/eviction seq */
} xce_repl_trace_state_t;
/*
* xce_sched_repl_state_s - see this header's own "Why a split at all"
* section above for why the full definition lives here now instead of
@ -302,6 +326,9 @@ struct xce_sched_repl_state_s {
* actually printed in a preamble; 0
* (calloc zero) until the first
* preamble prints this session. */
/* TraceStore (task 4.2): stored traces for `info trace`. */
xce_repl_trace_state_t trace;
};
/* repl_split_tokens/XCE_REPL_MAX_TOKENS/XCE_REPL_TOKEN_LEN (repl.c) -
@ -538,4 +565,39 @@ extern void repl_trace_render(const struct xce_sched_trace_rec_s* rec,
xce_repl_units_mode_t units, FILE* out1,
FILE* out2);
/*
* TraceStore (repl_trace_store.c, scheduler-sim-decision-trace spec, task
* 4.2, design.md "TraceStore").
*
* repl_trace_store_put - takes ownership of `rec`. Replaces and frees any
* stored record of the same jobid (7.10) and clears that jobid's discarded
* mark, then evicts the oldest records until the total is within
* XCE_REPL_TRACE_STORE_BOUND, never the one just stored; each evicted
* jobid is marked discarded (9.5). If the store cannot take the record
* (out of memory) the record is freed. NULL `ts` or `rec`: nothing stored.
*
* repl_trace_store_bytes - the sum of sched_trace_rec_bytes() over the
* stored records.
*
* repl_trace_store_fini - frees every stored record and both tables, and
* leaves `ts` empty and reusable.
*
* repl_cmd_info_trace - `info trace <jobsel> [<jobsel> ...]`, tokens as
* split by repl_split_tokens() (tokens[0] "info", tokens[1] "trace"). For
* a job or element: the stored trace at its captured level in the current
* units mode (7.8), else "trace discarded (memory bound)" (9.6), else "no
* trace available" with the job's state, else "unknown job" (7.11). For an
* array (or an element range): each element's trace in index order with
* discarded elements marked, and the elements' states when none has a
* stored trace (7.9). Reads job states with sched_get_job() and an
* iteration of core->jobs only. Always returns false (re-prompt).
*/
struct xce_sched_trace_rec_s;
extern void repl_trace_store_put(xce_repl_trace_state_t* ts,
struct xce_sched_trace_rec_s* rec);
extern size_t repl_trace_store_bytes(const xce_repl_trace_state_t* ts);
extern void repl_trace_store_fini(xce_repl_trace_state_t* ts);
extern bool repl_cmd_info_trace(xce_sched_sim_driver_t* sim,
char tokens[][XCE_REPL_TOKEN_LEN], int ntok);
#endif

477
src/scheduler/src/repl_trace_store.c Normal file
View file

@ -0,0 +1,477 @@
/*
* repl_trace_store.c - the simulator REPL's stored decision traces and
* `info trace` (scheduler-sim-decision-trace spec, task 4.2; design.md
* "TraceStore").
*
* The store keeps the latest trace record per job (7.10) so `info trace`
* can show it again later, including for a job that has since finished or
* left the replay (7.8). Two libxce htabs keyed by jobid hold the state:
* - `by_job`: jobid -> repl_trace_entry_t (the record, its byte size and
* its capture sequence number);
* - `discarded`: jobid -> eviction sequence number, for jobs whose record
* was evicted under the memory bound (9.6).
* Entries are also chained in capture order, oldest first. After every put
* the oldest entries are evicted until the summed sched_trace_rec_bytes()
* is within XCE_REPL_TRACE_STORE_BOUND (64 MiB), but the entry just stored
* is never evicted, even when it alone is over the bound (9.5). A jobid is
* in at most one of the two tables: a put clears its discarded mark, and an
* eviction removes it from `by_job` before marking it.
*
* Records are rendered with repl_trace_render() (task 4.1) at the level
* they were captured with and in the units mode current when `info trace`
* is issued (7.7, 7.8); nothing here formats a record itself.
*
* Only the sim thread reads or writes the store. Live scheduler state is
* read only for the no-trace replies: sched_get_job() for one job (as
* `info job` does) and an iteration of core->jobs for an array's elements
* (7.9, 7.11). Nothing is written to the core.
*
* The record header's "file tag" (design.md "Data Models") is not used
* here: `info trace` writes to the console only, and inline emission to a
* trace file (task 4.6) uses the settings the session was captured under.
*/
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include "core.h"
#include "repl.h"
#include "repl_internal.h"
#include "sched/trace.h"
#include "sim.h"
#include "utils.h"
#define STORE_NSLOTS 4099
typedef struct repl_trace_entry_s {
struct repl_trace_entry_s* prev; /* older */
struct repl_trace_entry_s* next; /* newer */
xce_sched_trace_rec_t* rec;
size_t bytes;
uint64_t seq;
} repl_trace_entry_t;
/* ---- the store ------------------------------------------------------------ */
static void* jobid_key(uint64_t jobid) { return (void*)(uintptr_t)jobid; }
static bool tables_ensure(xce_repl_trace_state_t* ts)
{
if (ts->tables_ready) {
return true;
}
if (ht_init(&ts->by_job, STORE_NSLOTS, int_hash_key) < 0) {
return false;
}
if (ht_init(&ts->discarded, STORE_NSLOTS, int_hash_key) < 0) {
ht_fini(&ts->by_job);
memset(&ts->by_job, 0, sizeof(ts->by_job));
return false;
}
ts->tables_ready = true;
return true;
}
static void list_unlink(xce_repl_trace_state_t* ts, repl_trace_entry_t* e)
{
if (e->prev != NULL) {
e->prev->next = e->next;
} else {
ts->oldest = e->next;
}
if (e->next != NULL) {
e->next->prev = e->prev;
} else {
ts->newest = e->prev;
}
e->prev = NULL;
e->next = NULL;
ts->total_bytes -= e->bytes;
}
static void list_append(xce_repl_trace_state_t* ts, repl_trace_entry_t* e)
{
e->prev = ts->newest;
e->next = NULL;
if (ts->newest != NULL) {
ts->newest->next = e;
} else {
ts->oldest = e;
}
ts->newest = e;
ts->total_bytes += e->bytes;
}
static repl_trace_entry_t* stored_entry(const xce_repl_trace_state_t* ts,
uint64_t jobid)
{
hent_t* h;
if ((ts == NULL) || !ts->tables_ready || (jobid == 0)) {
return NULL;
}
h = ht_get((htab_t*)&ts->by_job, jobid_key(jobid));
return (h != NULL) ? (repl_trace_entry_t*)(intptr_t)h->data : NULL;
}
static bool is_discarded(const xce_repl_trace_state_t* ts, uint64_t jobid)
{
if ((ts == NULL) || !ts->tables_ready || (jobid == 0)) {
return false;
}
return ht_get((htab_t*)&ts->discarded, jobid_key(jobid)) != NULL;
}
/* Evicts the oldest entries until the total is within the bound, never the
* newest one (9.5). An evicted jobid is marked discarded (9.6); if the mark
* cannot be allocated, the job later reads as having no trace. */
static void evict_over_bound(xce_repl_trace_state_t* ts)
{
while ((ts->total_bytes > XCE_REPL_TRACE_STORE_BOUND)
&& (ts->oldest != NULL) && (ts->oldest != ts->newest)) {
repl_trace_entry_t* e = ts->oldest;
uint64_t jobid = e->rec->jobid;
hent_t* mark;
list_unlink(ts, e);
ht_del(&ts->by_job, jobid_key(jobid));
mark = ht_add(&ts->discarded, jobid_key(jobid));
if (mark != NULL) {
mark->data = (int64_t)++ts->seq;
}
sched_trace_rec_free(e->rec);
free(e);
}
}
void repl_trace_store_put(xce_repl_trace_state_t* ts,
xce_sched_trace_rec_t* rec)
{
repl_trace_entry_t* e;
if (rec == NULL) {
return;
}
if ((ts == NULL) || (rec->jobid == 0) || !tables_ensure(ts)) {
sched_trace_rec_free(rec);
return;
}
e = stored_entry(ts, rec->jobid);
if (e != NULL) {
/* 7.10: the new trace replaces the stored one */
list_unlink(ts, e);
sched_trace_rec_free(e->rec);
} else {
hent_t* h;
e = (repl_trace_entry_t*)calloc(1, sizeof(*e));
if (e == NULL) {
sched_trace_rec_free(rec);
return;
}
h = ht_add(&ts->by_job, jobid_key(rec->jobid));
if (h == NULL) {
free(e);
sched_trace_rec_free(rec);
return;
}
h->data = (int64_t)(intptr_t)e;
}
/* a newer capture clears the discarded mark */
ht_del(&ts->discarded, jobid_key(rec->jobid));
e->rec = rec;
e->bytes = sched_trace_rec_bytes(rec);
e->seq = ++ts->seq;
list_append(ts, e);
evict_over_bound(ts);
}
size_t repl_trace_store_bytes(const xce_repl_trace_state_t* ts)
{
return (ts != NULL) ? ts->total_bytes : 0;
}
void repl_trace_store_fini(xce_repl_trace_state_t* ts)
{
repl_trace_entry_t* e;
if (ts == NULL) {
return;
}
e = ts->oldest;
while (e != NULL) {
repl_trace_entry_t* next = e->next;
sched_trace_rec_free(e->rec);
free(e);
e = next;
}
if (ts->tables_ready) {
ht_fini(&ts->by_job);
ht_fini(&ts->discarded);
}
memset(ts, 0, sizeof(*ts));
}
/* ---- info trace ----------------------------------------------------------- */
typedef struct {
uint64_t* ids;
int n;
int cap;
} id_list_t;
static bool id_list_add(id_list_t* l, uint64_t id)
{
if (l->n == l->cap) {
int cap = (l->cap == 0) ? 16 : (l->cap * 2);
uint64_t* p = (uint64_t*)realloc(l->ids, sizeof(*p) * (size_t)cap);
if (p == NULL) {
return false;
}
l->ids = p;
l->cap = cap;
}
l->ids[l->n++] = id;
return true;
}
static int id_cmp(const void* a, const void* b)
{
uint64_t x = *(const uint64_t*)a;
uint64_t y = *(const uint64_t*)b;
return (x < y) ? -1 : ((x > y) ? 1 : 0);
}
/* True when `jobid` is an element of the array `base` with an index in
* [lo, hi]. */
static bool in_array(uint64_t jobid, uint64_t base, uint32_t lo, uint32_t hi)
{
uint64_t idx;
if (!is_array_tag(jobid) || (array_jobid(jobid) != base)) {
return false;
}
idx = array_idx(jobid);
return (idx >= lo) && (idx <= hi);
}
/* Adds the jobids of `t` that are elements of `base` in [lo, hi]. */
static void collect_table(const htab_t* t, uint64_t base, uint32_t lo,
uint32_t hi, id_list_t* out)
{
int i;
for (i = 0; i < t->nslots; i++) {
const hent_t* h;
for (h = t->slots[i]; h != NULL; h = h->next) {
uint64_t id = (uint64_t)(uintptr_t)h->key;
if (in_array(id, base, lo, hi)) {
(void)id_list_add(out, id);
}
}
}
}
/* Adds the jobids in core->jobs that are elements of `base` in [lo, hi]. */
static void collect_core(xce_sched_core_t* core, uint64_t base, uint32_t lo,
uint32_t hi, id_list_t* out)
{
iterator_t it;
if ((core->jobs == NULL) || (iter_make(&it, core->jobs) < 0)) {
return;
}
for (;;) {
xce_sched_job_t* job = NULL;
int rc = iter_next(&it, (void**)&job);
if ((rc < 0) || (rc == RTTI_ITER_NO_DATA) || (job == NULL)) {
break;
}
if (in_array(job->jobid, base, lo, hi)) {
(void)id_list_add(out, job->jobid);
}
}
iter_destroy(&it);
}
static void print_discarded(uint64_t jobid)
{
char idbuf[64];
fmt_jobid(idbuf, sizeof(idbuf), jobid);
fprintf(stdout,
"[sim] info trace: job %s: trace discarded (memory bound)\n",
idbuf);
}
/* One job or element (7.8, 7.11, 9.6). */
static void info_trace_one(xce_sched_sim_driver_t* sim,
const xce_repl_trace_state_t* ts, uint64_t jobid)
{
repl_trace_entry_t* e = stored_entry(ts, jobid);
xce_sched_job_t* job;
char idbuf[64];
if (e != NULL) {
repl_trace_render(e->rec, repl_current_units_mode(sim), stdout,
NULL);
return;
}
if (is_discarded(ts, jobid)) {
print_discarded(jobid);
return;
}
fmt_jobid(idbuf, sizeof(idbuf), jobid);
job = sched_get_job(&(sim->core), jobid);
if (job != NULL) {
fprintf(stdout,
"[sim] info trace: job %s: no trace available (state %s)\n",
idbuf, repl_job_state_name(job->state));
} else {
fprintf(stdout, "[sim] info trace: job %s: unknown job\n", idbuf);
}
}
/* The elements of array `base` with an index in [lo, hi] (7.9). Returns
* false, printing nothing, when the store and the job table know no such
* element. */
static bool info_trace_array(xce_sched_sim_driver_t* sim,
const xce_repl_trace_state_t* ts, uint64_t base,
uint32_t lo, uint32_t hi)
{
id_list_t traced = { NULL, 0, 0 };
id_list_t states = { NULL, 0, 0 };
int nstored = 0;
int i;
char idbuf[64];
if ((ts != NULL) && ts->tables_ready) {
collect_table(&ts->by_job, base, lo, hi, &traced);
collect_table(&ts->discarded, base, lo, hi, &traced);
}
qsort(traced.ids, (size_t)traced.n, sizeof(uint64_t), id_cmp);
for (i = 0; i < traced.n; i++) {
repl_trace_entry_t* e = stored_entry(ts, traced.ids[i]);
if (e != NULL) {
repl_trace_render(e->rec, repl_current_units_mode(sim), stdout,
NULL);
nstored++;
} else {
print_discarded(traced.ids[i]);
}
}
if (nstored == 0) {
collect_core(&(sim->core), base, lo, hi, &states);
if ((traced.n == 0) && (states.n == 0)) {
free(traced.ids);
return false;
}
qsort(states.ids, (size_t)states.n, sizeof(uint64_t), id_cmp);
fmt_jobid(idbuf, sizeof(idbuf), base);
fprintf(stdout,
"[sim] info trace: array %s: no element has a stored trace\n",
idbuf);
for (i = 0; i < states.n; i++) {
xce_sched_job_t* job = sched_get_job(&(sim->core), states.ids[i]);
fmt_jobid(idbuf, sizeof(idbuf), states.ids[i]);
fprintf(stdout, " %s %s\n", idbuf,
(job != NULL) ? repl_job_state_name(job->state)
: "UNKNOWN");
}
}
free(traced.ids);
free(states.ids);
return true;
}
static bool all_digits(const char* s)
{
if (*s == '\0') {
return false;
}
for (; *s != '\0'; s++) {
if ((*s < '0') || (*s > '9')) {
return false;
}
}
return true;
}
/* One `info trace` argument: a job, `array[k]`, `array[a-b]`, an array id,
* or (like `info job`) a raw packed jobid. */
static void info_trace_sel(xce_sched_sim_driver_t* sim,
const xce_repl_trace_state_t* ts, const char* arg)
{
jobsel_t sel;
jobsel_err_t err = parse_jobsel(arg, &sel);
char idbuf[64];
if (err != JOBSEL_OK) {
if (all_digits(arg)) {
info_trace_one(sim, ts, strtoull(arg, NULL, 10));
return;
}
fprintf(stdout, "[sim] info trace: invalid job id %s: %s\n", arg,
jobsel_strerror(err));
return;
}
switch (sel.form) {
case JOBSEL_SINGLE:
info_trace_one(sim, ts, jobsel_element_id(&sel, sel.lo));
return;
case JOBSEL_RANGE:
if (!info_trace_array(sim, ts, sel.key, sel.lo, sel.hi)) {
fmt_jobid(idbuf, sizeof(idbuf), sel.key);
fprintf(stdout, "[sim] info trace: array %s: unknown job\n",
idbuf);
}
return;
case JOBSEL_BARE:
break;
}
/* A bare id is a job when the store or the job table knows it, else an
* array base when an element of it is known (7.9), else unknown. */
if ((stored_entry(ts, sel.key) != NULL) || is_discarded(ts, sel.key)
|| (sched_get_job(&(sim->core), sel.key) != NULL)
|| (EXTRACT_ARRAY_IDX(sel.key) != 0) /* unaligned: not an array */
|| !info_trace_array(sim, ts, sel.key, 0, JOBID_FIELD_MAX_INDEX)) {
info_trace_one(sim, ts, sel.key);
}
}
bool repl_cmd_info_trace(xce_sched_sim_driver_t* sim,
char tokens[][XCE_REPL_TOKEN_LEN], int ntok)
{
const xce_repl_trace_state_t* ts;
int i;
if (ntok < 3) {
fprintf(stdout, "[sim] info trace: usage: info trace "
"<jobid>|<array-id>[<k>|<a>-<b>] ...\n");
return false;
}
ts = (sim->repl_state != NULL) ? &(sim->repl_state->trace) : NULL;
for (i = 2; i < ntok; i++) {
info_trace_sel(sim, ts, tokens[i]);
}
return false;
}

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

@ -2735,3 +2735,24 @@ sched_trace_render_test = executable(
test('scheduler trace render test',
sched_trace_render_test, suite: 'scheduler',
env: ['ASAN_OPTIONS=detect_leaks=0'])
# scheduler-sim-decision-trace task 4.2: the trace store and `info trace`
# (design.md "TraceStore"; Requirements 1.7, 7.8-7.11, 9.5, 9.6).
# Replacement, eviction at 64 MiB that never drops the newest record, the
# three no-trace replies and arrays mixing stored and discarded elements.
# Links the record model, the renderer, the shared REPL formatting module
# and the store; the job table is a fixture kvmgr with sched_get_job() and
# repl_job_state_name() defined by the test.
sched_trace_store_test = executable(
'sched_trace_store_test',
['test_trace_store.c', '../src/sched_trace.c',
'../src/sched_trace_scan.c', '../src/repl_trace_render.c',
'../src/repl_trace_render_outcome.c', '../src/repl_fmt.c',
'../src/repl_trace_store.c'],
include_directories:[incdir],
dependencies: [scheduler_test_deps],
install: false)
test('scheduler trace store test',
sched_trace_store_test, suite: 'scheduler',
env: ['ASAN_OPTIONS=detect_leaks=0'])

567
src/scheduler/testsuite/test_trace_store.c Normal file
View file

@ -0,0 +1,567 @@
/*
* test_trace_store.c - the REPL trace store and `info trace`
* (scheduler-sim-decision-trace spec, task 4.2; design.md "TraceStore").
*
* Links the record model (src/sched_trace.c, src/sched_trace_scan.c), the
* renderer (src/repl_trace_render*.c), the shared REPL formatting module
* (src/repl_fmt.c) and the store (src/repl_trace_store.c). The job table is
* a fixture: a real KVMGR_KEY_INT kvmgr in sim.core.jobs, with
* sched_get_job() and repl_job_state_name() defined here in place of
* core.c and repl_info.c.
*
* Covers:
* - a new trace replaces the stored one (7.10);
* - eviction at 64 MiB drops the oldest first and never the newest, even
* when the newest alone is over the bound (9.5);
* - a newer capture clears a discarded mark;
* - the three replies without a stored trace: discarded (9.6), no trace
* with the job's state, unknown job (7.11);
* - `info trace` renders at the captured level in the units mode current
* when it is issued (7.8);
* - an array mixing stored and discarded elements, in index order, and
* an array with no stored trace listing its elements' states (7.9).
*/
#define _GNU_SOURCE
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>
#include <unity/unity.h>
#include "kvmgr.h"
#include "repl_internal.h"
#include "sched/trace.h"
#include "sim.h"
#include "utils.h"
#define MIB ((size_t)1024 * 1024)
#define PLAIN_JOB MAKE_JOBID(1, 2, 1000, 0x0, (uint64_t)42)
#define PLAIN_JOB2 MAKE_JOBID(1, 2, 1000, 0x0, (uint64_t)43)
#define UNKNOWN_JOB MAKE_JOBID(1, 2, 1000, 0x0, (uint64_t)99)
#define ARRAY_SEQ ((uint64_t)32768 * 5)
#define ARRAY_BASE MAKE_JOBID(1, 2, 1000, 0x0, ARRAY_SEQ)
#define ELEM(k) \
MAKE_JOBID(1, 2, 1000, TAG_JOB_ARRAY, ARRAY_SEQ | (uint64_t)(k))
#define ARRAY2_SEQ ((uint64_t)32768 * 6)
#define ARRAY2_BASE MAKE_JOBID(1, 2, 1000, 0x0, ARRAY2_SEQ)
#define ELEM2(k) \
MAKE_JOBID(1, 2, 1000, TAG_JOB_ARRAY, ARRAY2_SEQ | (uint64_t)(k))
static xce_sched_sim_driver_t g_sim;
static xce_sched_repl_state_t g_rs;
/* ---- fixture job table -------------------------------------------------- */
xce_sched_job_t* sched_get_job(xce_sched_core_t* core, uint64_t jobid)
{
xce_sched_job_t* job = NULL;
if (key_find(core->jobs, (void*)jobid, (void**)&job) <= 0) {
return NULL;
}
return job;
}
const char* repl_job_state_name(uint32_t state)
{
if (JOB_IS_RUN(state)) {
return "RUN";
}
if (JOB_IS_EXIT(state)) {
return "EXIT";
}
if (JOB_IS_DONE(state)) {
return "DONE";
}
if (JOB_IS_PEND(state)) {
return "PEND";
}
return "UNKNOWN";
}
static void add_job(uint64_t jobid, uint32_t state)
{
xce_sched_job_t* j = (xce_sched_job_t*)calloc(1, sizeof(*j));
TEST_ASSERT_NOT_NULL(j);
j->jobid = jobid;
j->state = state;
TEST_ASSERT_TRUE(key_add(g_sim.core.jobs, (void*)jobid, j) >= 0);
}
void setUp(void)
{
static bool made = false;
if (!made) {
kvmgr_params_t p = { .key_type = KVMGR_KEY_INT,
.nslots = 59,
.delta = 16,
.datakiller = NULL };
memset(&g_sim, 0, sizeof(g_sim));
g_sim.core.jobs = makeInstance(&xkvmgr_type, &p);
TEST_ASSERT_NOT_NULL(g_sim.core.jobs);
add_job(PLAIN_JOB, JOB_STAT_PEND);
add_job(PLAIN_JOB2, JOB_STAT_RUN);
add_job(ELEM(1), JOB_STAT_PEND);
add_job(ELEM(2), JOB_STAT_RUN);
add_job(ELEM(3), JOB_STAT_PEND);
add_job(ELEM(4), JOB_STAT_DONE);
add_job(ELEM2(2), JOB_STAT_RUN);
add_job(ELEM2(1), JOB_STAT_PEND);
made = true;
}
memset(&g_rs, 0, sizeof(g_rs));
g_rs.units_mode = XCE_REPL_UNITS_BOTH;
g_sim.repl_state = &g_rs;
}
void tearDown(void)
{
repl_trace_store_fini(&g_rs.trace);
g_sim.repl_state = NULL;
}
/* ---- records ------------------------------------------------------------ */
/* A reconstructed dispatch of 8 GiB mem on n01, plus `pad` bytes of arena so
* the record's size can be set for the memory bound. `session` tells two
* records of one job apart in the rendering. */
static xce_sched_trace_rec_t* new_rec(uint64_t jobid, uint64_t session,
xce_sched_trace_level_t level,
size_t pad)
{
xce_sched_trace_ctl_t ctl;
xce_sched_trace_rec_t* rec;
sched_trace_host_out_t* h;
sched_trace_alloc_row_t row;
memset(&ctl, 0, sizeof(ctl));
ctl.all = true;
ctl.level = level;
ctl.line_bound = 1000;
ctl.session_no = session;
ctl.position = 1;
rec = sched_trace_rec_new(&ctl, jobid, XCE_TRACE_REC_RECONSTRUCTED);
TEST_ASSERT_NOT_NULL(rec);
memset(&row, 0, sizeof(row));
row.res = sched_trace_arena_strdup(rec, "mem");
row.value = 8.0 * 1073741824.0;
rec->outcome = SCHED_TRACE_OUTCOME_DISPATCH;
rec->dispatch.nhosts = 1;
rec->dispatch.hosts = (sched_trace_host_out_t*)sched_trace_arena_alloc(
rec, sizeof(sched_trace_host_out_t));
h = &rec->dispatch.hosts[0];
h->host = sched_trace_arena_strdup(rec, "n01");
h->nres = 1;
h->res = (sched_trace_alloc_row_t*)sched_trace_arena_memdup(
rec, &row, sizeof(row));
if (pad > 0) {
TEST_ASSERT_NOT_NULL(sched_trace_arena_alloc(rec, pad));
}
TEST_ASSERT_FALSE(rec->incomplete);
return rec;
}
static char* render(const xce_sched_trace_rec_t* rec,
xce_repl_units_mode_t units)
{
char* buf = NULL;
size_t len = 0;
FILE* f = open_memstream(&buf, &len);
TEST_ASSERT_NOT_NULL(f);
repl_trace_render(rec, units, f, NULL);
fclose(f);
return buf;
}
static char* fmt_id(uint64_t jobid)
{
static char bufs[8][64];
static int next = 0;
char* b = bufs[next++ % 8];
fmt_jobid(b, 64, jobid);
return b;
}
/* ---- running `info trace` ---------------------------------------------- */
/* Whitespace split into REPL tokens (repl.c's repl_split_tokens is not
* linked here). */
static int split(const char* line, char tokens[][XCE_REPL_TOKEN_LEN])
{
int n = 0;
const char* p = line;
while ((*p != '\0') && (n < XCE_REPL_MAX_TOKENS)) {
size_t k = 0;
while ((*p == ' ') || (*p == '\t')) {
p++;
}
if (*p == '\0') {
break;
}
while ((*p != '\0') && (*p != ' ') && (*p != '\t')) {
if (k < (XCE_REPL_TOKEN_LEN - 1)) {
tokens[n][k++] = *p;
}
p++;
}
tokens[n++][k] = '\0';
}
return n;
}
/* Runs `info trace <args>` with stdout redirected to a temp file and
* returns what it printed (caller frees). */
static char* info_trace(const char* args)
{
char line[512];
char tokens[XCE_REPL_MAX_TOKENS][XCE_REPL_TOKEN_LEN];
int ntok;
char tmp[] = "/tmp/xce_trace_store_stdout_XXXXXX";
int fd = mkstemp(tmp);
int saved;
off_t len;
char* buf;
TEST_ASSERT_TRUE(fd >= 0);
snprintf(line, sizeof(line), "info trace %s", args);
ntok = split(line, tokens);
fflush(stdout);
saved = dup(STDOUT_FILENO);
TEST_ASSERT_TRUE(saved >= 0);
TEST_ASSERT_TRUE(dup2(fd, STDOUT_FILENO) >= 0);
TEST_ASSERT_FALSE(repl_cmd_info_trace(&g_sim, tokens, ntok));
fflush(stdout);
dup2(saved, STDOUT_FILENO);
close(saved);
len = lseek(fd, 0, SEEK_CUR);
TEST_ASSERT_TRUE(len >= 0);
buf = (char*)calloc(1, (size_t)len + 1);
TEST_ASSERT_NOT_NULL(buf);
TEST_ASSERT_TRUE(lseek(fd, 0, SEEK_SET) == 0);
if (len > 0) {
TEST_ASSERT_TRUE(read(fd, buf, (size_t)len) == len);
}
close(fd);
unlink(tmp);
return buf;
}
static void assert_text(const char* want, const char* got)
{
if (strcmp(want, got) != 0) {
fprintf(stderr, "---- want ----\n%s---- got ----\n%s----\n", want,
got);
}
TEST_ASSERT_EQUAL_STRING(want, got);
}
static void assert_info(const char* args, const char* want)
{
char* got = info_trace(args);
assert_text(want, got);
free(got);
}
static char* discarded_line(uint64_t jobid)
{
static char b[4][160];
static int next = 0;
char* p = b[next++ % 4];
snprintf(p, 160, "[sim] info trace: job %s: trace discarded (memory "
"bound)\n",
fmt_id(jobid));
return p;
}
/* ---- tests ---------------------------------------------------------------- */
/* 7.10: a new trace for a job replaces the earlier one; the store holds
* only the new record's bytes. */
static void test_put_replaces_earlier_trace(void)
{
xce_sched_trace_rec_t* r1 = new_rec(PLAIN_JOB, 1, XCE_TRACE_LEVEL_FULL, 0);
xce_sched_trace_rec_t* r2 = new_rec(PLAIN_JOB, 2, XCE_TRACE_LEVEL_FULL, 0);
size_t b1 = sched_trace_rec_bytes(r1);
size_t b2 = sched_trace_rec_bytes(r2);
char* want;
repl_trace_store_put(&g_rs.trace, r1);
TEST_ASSERT_EQUAL_size_t(b1, repl_trace_store_bytes(&g_rs.trace));
want = render(r2, XCE_REPL_UNITS_BOTH);
repl_trace_store_put(&g_rs.trace, r2); /* frees r1 */
TEST_ASSERT_EQUAL_size_t(b2, repl_trace_store_bytes(&g_rs.trace));
assert_info(fmt_id(PLAIN_JOB), want);
TEST_ASSERT_NOT_NULL(strstr(want, "session 2"));
free(want);
}
/* 7.8: rendered at the captured level, in the units mode current when
* `info trace` is issued. */
static void test_info_uses_current_units_and_captured_level(void)
{
xce_sched_trace_rec_t* r
= new_rec(PLAIN_JOB, 3, XCE_TRACE_LEVEL_SUMMARY, 0);
char* raw = render(r, XCE_REPL_UNITS_RAW);
char* human = render(r, XCE_REPL_UNITS_HUMAN);
TEST_ASSERT_TRUE(strcmp(raw, human) != 0);
TEST_ASSERT_NOT_NULL(strstr(raw, "level summary"));
repl_trace_store_put(&g_rs.trace, r);
g_rs.units_mode = XCE_REPL_UNITS_RAW;
assert_info(fmt_id(PLAIN_JOB), raw);
g_rs.units_mode = XCE_REPL_UNITS_HUMAN;
assert_info(fmt_id(PLAIN_JOB), human);
free(raw);
free(human);
}
/* 9.5: at 64 MiB the oldest records go first; each evicted job is marked
* discarded (9.6). */
static void test_eviction_drops_oldest_first(void)
{
xce_sched_trace_rec_t* a
= new_rec(PLAIN_JOB, 1, XCE_TRACE_LEVEL_FULL, 24 * MIB);
xce_sched_trace_rec_t* b
= new_rec(PLAIN_JOB2, 1, XCE_TRACE_LEVEL_FULL, 24 * MIB);
xce_sched_trace_rec_t* c
= new_rec(ELEM(1), 1, XCE_TRACE_LEVEL_FULL, 24 * MIB);
size_t bc = sched_trace_rec_bytes(b) + sched_trace_rec_bytes(c);
char* want_c;
repl_trace_store_put(&g_rs.trace, a);
repl_trace_store_put(&g_rs.trace, b);
want_c = render(c, XCE_REPL_UNITS_BOTH);
repl_trace_store_put(&g_rs.trace, c);
TEST_ASSERT_EQUAL_size_t(bc, repl_trace_store_bytes(&g_rs.trace));
TEST_ASSERT_TRUE(repl_trace_store_bytes(&g_rs.trace)
<= XCE_REPL_TRACE_STORE_BOUND);
assert_info(fmt_id(PLAIN_JOB), discarded_line(PLAIN_JOB));
assert_info(fmt_id(ELEM(1)), want_c);
free(want_c);
}
/* 9.5: the newest record stays even when it alone is over the bound. */
static void test_eviction_never_drops_newest(void)
{
xce_sched_trace_rec_t* a
= new_rec(PLAIN_JOB, 1, XCE_TRACE_LEVEL_FULL, 8 * MIB);
xce_sched_trace_rec_t* big
= new_rec(PLAIN_JOB2, 1, XCE_TRACE_LEVEL_FULL, 70 * MIB);
size_t bbig = sched_trace_rec_bytes(big);
char* want;
TEST_ASSERT_TRUE(bbig > XCE_REPL_TRACE_STORE_BOUND);
repl_trace_store_put(&g_rs.trace, a);
want = render(big, XCE_REPL_UNITS_BOTH);
repl_trace_store_put(&g_rs.trace, big);
TEST_ASSERT_EQUAL_size_t(bbig, repl_trace_store_bytes(&g_rs.trace));
assert_info(fmt_id(PLAIN_JOB2), want);
assert_info(fmt_id(PLAIN_JOB), discarded_line(PLAIN_JOB));
free(want);
}
/* A newer capture of a discarded job clears its mark. */
static void test_newer_capture_clears_discarded(void)
{
xce_sched_trace_rec_t* a
= new_rec(PLAIN_JOB, 1, XCE_TRACE_LEVEL_FULL, 40 * MIB);
xce_sched_trace_rec_t* b
= new_rec(PLAIN_JOB2, 1, XCE_TRACE_LEVEL_FULL, 40 * MIB);
xce_sched_trace_rec_t* a2 = new_rec(PLAIN_JOB, 5, XCE_TRACE_LEVEL_FULL, 0);
char* want;
repl_trace_store_put(&g_rs.trace, a);
repl_trace_store_put(&g_rs.trace, b); /* evicts a */
assert_info(fmt_id(PLAIN_JOB), discarded_line(PLAIN_JOB));
want = render(a2, XCE_REPL_UNITS_BOTH);
repl_trace_store_put(&g_rs.trace, a2);
assert_info(fmt_id(PLAIN_JOB), want);
free(want);
}
/* 7.11: no stored trace and none discarded: the job's state, or unknown. */
static void test_no_trace_replies(void)
{
char want[256];
snprintf(want, sizeof(want),
"[sim] info trace: job %s: no trace available (state PEND)\n",
fmt_id(PLAIN_JOB));
assert_info(fmt_id(PLAIN_JOB), want);
snprintf(want, sizeof(want),
"[sim] info trace: job %s: no trace available (state RUN)\n",
fmt_id(ELEM(2)));
assert_info(fmt_id(ELEM(2)), want);
snprintf(want, sizeof(want), "[sim] info trace: job %s: unknown job\n",
fmt_id(UNKNOWN_JOB));
assert_info(fmt_id(UNKNOWN_JOB), want);
}
/* 7.8: a trace stays available after its job left the job table. */
static void test_trace_of_removed_job(void)
{
xce_sched_trace_rec_t* r = new_rec(UNKNOWN_JOB, 4, XCE_TRACE_LEVEL_FULL, 0);
char* want = render(r, XCE_REPL_UNITS_BOTH);
repl_trace_store_put(&g_rs.trace, r);
assert_info(fmt_id(UNKNOWN_JOB), want);
free(want);
}
/* 7.9: an array mixing stored and discarded elements, in index order;
* elements with neither are not listed. */
static void test_array_mixed_stored_and_discarded(void)
{
xce_sched_trace_rec_t* e2
= new_rec(ELEM(2), 1, XCE_TRACE_LEVEL_FULL, 40 * MIB);
xce_sched_trace_rec_t* e3 = new_rec(ELEM(3), 1, XCE_TRACE_LEVEL_FULL, 0);
xce_sched_trace_rec_t* e1
= new_rec(ELEM(1), 1, XCE_TRACE_LEVEL_SUMMARY, 30 * MIB);
char* r1;
char* r3;
char* want = NULL;
size_t n = 0;
FILE* f;
r3 = render(e3, XCE_REPL_UNITS_BOTH);
r1 = render(e1, XCE_REPL_UNITS_BOTH);
repl_trace_store_put(&g_rs.trace, e2);
repl_trace_store_put(&g_rs.trace, e3);
repl_trace_store_put(&g_rs.trace, e1); /* evicts e2 */
f = open_memstream(&want, &n);
fputs(r1, f);
fputs(discarded_line(ELEM(2)), f);
fputs(r3, f);
fclose(f);
assert_info(fmt_id(ARRAY_BASE), want);
free(want);
/* an element range selects within the array */
want = NULL;
f = open_memstream(&want, &n);
fputs(discarded_line(ELEM(2)), f);
fputs(r3, f);
fclose(f);
{
char sel[96];
snprintf(sel, sizeof(sel), "%s[2-4]", fmt_id(ARRAY_BASE));
assert_info(sel, want);
}
free(want);
/* an element by `array[k]` */
{
char sel[96];
snprintf(sel, sizeof(sel), "%s[1]", fmt_id(ARRAY_BASE));
assert_info(sel, r1);
}
free(r1);
free(r3);
}
/* 7.9: no element has a stored trace: say so and list element states in
* index order. */
static void test_array_without_traces_lists_states(void)
{
char want[512];
snprintf(want, sizeof(want),
"[sim] info trace: array %s: no element has a stored trace\n"
" %s PEND\n"
" %s RUN\n",
fmt_id(ARRAY2_BASE), fmt_id(ELEM2(1)), fmt_id(ELEM2(2)));
assert_info(fmt_id(ARRAY2_BASE), want);
}
/* 7.9 with 9.6: only discarded elements: they are marked, then the states. */
static void test_array_only_discarded(void)
{
xce_sched_trace_rec_t* e1
= new_rec(ELEM2(1), 1, XCE_TRACE_LEVEL_FULL, 40 * MIB);
xce_sched_trace_rec_t* other
= new_rec(PLAIN_JOB, 1, XCE_TRACE_LEVEL_FULL, 40 * MIB);
char want[768];
repl_trace_store_put(&g_rs.trace, e1);
repl_trace_store_put(&g_rs.trace, other); /* evicts e1 */
snprintf(want, sizeof(want),
"%s"
"[sim] info trace: array %s: no element has a stored trace\n"
" %s PEND\n"
" %s RUN\n",
discarded_line(ELEM2(1)), fmt_id(ARRAY2_BASE), fmt_id(ELEM2(1)),
fmt_id(ELEM2(2)));
assert_info(fmt_id(ARRAY2_BASE), want);
}
/* Usage and a malformed id leave the store untouched. */
static void test_usage_and_invalid_id(void)
{
char* got = info_trace("");
TEST_ASSERT_NOT_NULL(strstr(got, "usage: info trace"));
free(got);
got = info_trace("1-2-x-4");
TEST_ASSERT_NOT_NULL(strstr(got, "[sim] info trace: invalid job id"));
free(got);
}
/* A NULL record or store is ignored; fini empties and allows reuse. */
static void test_null_and_fini(void)
{
repl_trace_store_put(NULL, NULL);
repl_trace_store_put(&g_rs.trace, NULL);
TEST_ASSERT_EQUAL_size_t(0, repl_trace_store_bytes(&g_rs.trace));
repl_trace_store_put(&g_rs.trace,
new_rec(PLAIN_JOB, 1, XCE_TRACE_LEVEL_FULL, 0));
repl_trace_store_fini(&g_rs.trace);
TEST_ASSERT_EQUAL_size_t(0, repl_trace_store_bytes(&g_rs.trace));
repl_trace_store_put(&g_rs.trace,
new_rec(PLAIN_JOB, 1, XCE_TRACE_LEVEL_FULL, 0));
TEST_ASSERT_TRUE(repl_trace_store_bytes(&g_rs.trace) > 0);
}
int main(void)
{
UNITY_BEGIN();
RUN_TEST(test_put_replaces_earlier_trace);
RUN_TEST(test_info_uses_current_units_and_captured_level);
RUN_TEST(test_eviction_drops_oldest_first);
RUN_TEST(test_eviction_never_drops_newest);
RUN_TEST(test_newer_capture_clears_discarded);
RUN_TEST(test_no_trace_replies);
RUN_TEST(test_trace_of_removed_job);
RUN_TEST(test_array_mixed_stored_and_discarded);
RUN_TEST(test_array_without_traces_lists_states);
RUN_TEST(test_array_only_discarded);
RUN_TEST(test_usage_and_invalid_id);
RUN_TEST(test_null_and_fini);
return UNITY_END();
}