diff --git a/.kiro/specs/scheduler-sim-decision-trace/tasks.md b/.kiro/specs/scheduler-sim-decision-trace/tasks.md index b87bf681..03d74ec7 100644 --- a/.kiro/specs/scheduler-sim-decision-trace/tasks.md +++ b/.kiro/specs/scheduler-sim-decision-trace/tasks.md @@ -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: ". 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. diff --git a/src/scheduler/meson.build b/src/scheduler/meson.build index f44b5f2f..ad160970 100644 --- a/src/scheduler/meson.build +++ b/src/scheduler/meson.build @@ -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', diff --git a/src/scheduler/src/repl_internal.h b/src/scheduler/src/repl_internal.h index f273d998..e1aaa38d 100644 --- a/src/scheduler/src/repl_internal.h +++ b/src/scheduler/src/repl_internal.h @@ -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 [ ...]`, 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 diff --git a/src/scheduler/src/repl_trace_store.c b/src/scheduler/src/repl_trace_store.c new file mode 100644 index 00000000..ee8848ba --- /dev/null +++ b/src/scheduler/src/repl_trace_store.c @@ -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 +#include +#include + +#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 " + "|[|-] ...\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; +} diff --git a/src/scheduler/testsuite/meson.build b/src/scheduler/testsuite/meson.build index 0c3c06bf..7ce2bd20 100644 --- a/src/scheduler/testsuite/meson.build +++ b/src/scheduler/testsuite/meson.build @@ -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']) diff --git a/src/scheduler/testsuite/test_trace_store.c b/src/scheduler/testsuite/test_trace_store.c new file mode 100644 index 00000000..4967e08d --- /dev/null +++ b/src/scheduler/testsuite/test_trace_store.c @@ -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 +#include +#include +#include + +#include + +#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 ` 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(); +}