From 1b78fa9638d39a541f2d644630bf5c059c7b4efe Mon Sep 17 00:00:00 2001 From: Rongsong Shen Date: Thu, 8 Oct 2026 17:11:52 +0000 Subject: [PATCH] test(scheduler-sim-decision-trace): bounds under large traces; update the scheduler guide Task 6.4: a 1000-host, 64-task parallel job traced at full level stores exactly the line bound with exact omitted counts and counters; 10,000 traced decisions keep the trace store within 64 MiB without evicting the newest. The scheduler guide lists the REPL trace files and test rules and carries measured file sizes. Full all-components ASAN build passes. Co-Authored-By: Claude Opus 5.5 (1M context) --- .../scheduler-sim-decision-trace/tasks.md | 5 +- src/scheduler/CLAUDE.md | 10 +- src/scheduler/testsuite/meson.build | 28 + src/scheduler/testsuite/test_trace_bounds.c | 478 ++++++++++++++++++ 4 files changed, 518 insertions(+), 3 deletions(-) create mode 100644 src/scheduler/testsuite/test_trace_bounds.c diff --git a/.kiro/specs/scheduler-sim-decision-trace/tasks.md b/.kiro/specs/scheduler-sim-decision-trace/tasks.md index 74196958..42259856 100644 --- a/.kiro/specs/scheduler-sim-decision-trace/tasks.md +++ b/.kiro/specs/scheduler-sim-decision-trace/tasks.md @@ -301,7 +301,7 @@ - a non-interactive simulation never publishes settings. - _Requirements: 1.6, 2.5, 7.1, 7.2, 7.3, 7.5, 10.1, 11.3_ -- [ ] 6. Validation +- [x] 6. Validation - [x] 6.1 End-to-end trace scenarios over the contention fixture - Drive the real simulator REPL with tracing on for selected jobs and for all jobs, at both levels. - Assert that every section of Requirements 2 to 6 appears with the expected content: @@ -337,7 +337,7 @@ - Done when the equivalence test and the full scheduler suite pass under ASAN. - _Requirements: 10.1, 10.2, 10.3, 10.4, 10.5, 11.4_ -- [ ] 6.4 Bounds under large traces and the final full build +- [x] 6.4 Bounds under large traces and the final full build - Trace a 1000-host parallel job with up to 64 tasks, at full level with the default bound. The stored entries must stay within the bound, with exact counters. - Run 10,000 traced decisions. Stored-trace memory must never exceed 64 MiB. - Correct the stale file sizes in the scheduler component guide and list the new files. @@ -376,3 +376,4 @@ - 6.1: the e2e test (test_repl_trace_e2e.c, its _fixture.c and .h) drives the built xce_scheduler: step/full/all with the file copy, continue/breakpoint/summary/info trace, and selected jobs over a run-time extra fixture with compounds, an array and a selection mismatch. `XCE_TRACE_E2E_DUMP=1` prints the transcript. Findings: a compound dispatch shows one entry per member on a shared host, which is faithful (merge_job_alloc keys on instid); `break DECISION jobid=` matches only raw decimal ids, a pre-existing repl_break.c bug logged separately. - 6.2: test_trace_error_paths.c injects seven faults into the real allocator thread. Hooked sources are #included for the macro renames; the rest are linked so the wraps see their callers. `XCE_TRACE_ERR_DUMP=1` prints the rendered traces. The 2.2 OPEN item is fixed: a limit_row OOM now marks the record incomplete, sets LIMIT via `sched_trace_host_limit_hit(rec, NULL, NULL)` and continues. The 4.2 OPEN item is fixed: an array listing OOM prints "[sim] info trace: array : cannot list its elements: ". The capture and store tests now #include their sources. - 6.3: test_trace_equivalence.c (with _cmp.c and .h) runs four variants (off, all/summary, all/full+file, subset) in step and continue mode over the contention and 6.1 extra replays. It compares decoded decisions with full pend reasons, recording and snapshot bytes, verify output, and per-step worker/resource/limit dumps; `XCE_TRACE_EQUIV_DUMP=1` prints the transcripts. No difference was found. job->accounts is compared only indirectly, through snapshot job->allocs and the used totals. +- 6.4: test_trace_bounds.c (real allocator, 1000 workers, maxTasks=64) stores exactly 1000 entries and counts 187,930 omitted, with counters equal to the unbounded run. Over 10,000 traced puts the store peaks at 67,108,800 bytes and never evicts the newest. `XCE_TRACE_BOUNDS_DUMP=1` prints the numbers; the meson timeout is 300 s. The full all-components ASAN build passed: 148 C tests, xceadmin 373, Go ok. The large pend outcome showed ~63 duplicate NO_RESOURCE reasons per host, a pre-existing libxce/scheduler behaviour logged separately. diff --git a/src/scheduler/CLAUDE.md b/src/scheduler/CLAUDE.md index e905f136..2ae1809d 100644 --- a/src/scheduler/CLAUDE.md +++ b/src/scheduler/CLAUDE.md @@ -25,6 +25,14 @@ C-based job scheduling engine with a plugin architecture. Determines job-to-node None of them may use stdio, file I/O or logging, and none belongs in `sched_core_isolation_lib`; `test_core_isolation.c` scans all five. Every white-box test that `#include`s a hooked source (resource, host_iter, alloc_runner, scheduler, sched_election, driver) must also `#include` the four trace `.c` files, exactly once. + The REPL side (sim thread, stdio allowed, declared in `repl_internal.h`; `repl.h` holds the interactive-only gates `sim_loop.c` calls): + - `repl_trace.c`: the `trace` command grammar, `trace status`, and the refcounted `next`/`cur` session settings with their trace file. + - `repl_trace_emit.c`: session begin/end hooks and per-decision emission to the console and trace file, then into the store. + - `repl_trace_store.c`: latest record per job, oldest-first eviction at 64 MiB (never the newest), discarded marks, `info trace`. + - `repl_trace_render.c`, `repl_trace_render_outcome.c`, `repl_trace_render.h` (private): the one renderer; every output path reuses it. + + Every test that `#include`s `repl.c` or `sim_loop.c` (rather than linking them from `scheduler_srcs`) must also `#include` the five REPL trace `.c` files and the four core trace `.c` files, exactly once; `sched_sim_isolation_lib` carries all nine, and `test_sim_isolation.c` scans the REPL trace files and `repl_trace_render.h`. A new REPL trace file goes into all three. Bounds under large traces are covered by `test_trace_bounds.c`. + Links against `libxce` from `src/lib/`. Full design/requirements: `.kiro/specs/scheduler-core/`. ## Conventions @@ -32,7 +40,7 @@ Links against `libxce` from `src/lib/`. Full design/requirements: `.kiro/specs/s - Format with `clang-format` - New scheduling policies → add as a plugin under `src/plugins/`, do not modify the core engine - Shared types defined in `src/include/xce/` (libxce) -- Files stay under the repo's 800-line cap — several core/sim/repl/snapshot/recording modules are deliberately split across multiple `*_events.c`/`*_session.c`/`*_wire.c`-style files for this reason; keep new logic in the file matching its concern rather than growing one file past the cap. Not yet true project-wide: `scheduler.c` (1382 lines) and `plugins/builtin_fcfs.c` (1026 lines) are pre-existing exceptions over the cap with no split or exemption note — do not use them as precedent for new files, and prefer splitting them opportunistically rather than growing them further. +- Files stay under the repo's 800-line cap — several core/sim/repl/snapshot/recording modules are deliberately split across multiple `*_events.c`/`*_session.c`/`*_wire.c`-style files for this reason; keep new logic in the file matching its concern rather than growing one file past the cap. Not yet true project-wide: `scheduler.c` (2060 lines), `sim_loop.c` (1844), `repl_info.c` (1779), `core_events.c` (1488), `driver.c` (1117), `resource.c` (1106), `plugins/builtin_fcfs.c` (1104), `repl_break.c` (980) and `include/core.h` (861) (measured 2026-10-08) are pre-existing exceptions over the cap with no split or exemption note — do not use them as precedent for new files, and prefer splitting them opportunistically rather than growing them further. ## See Also - Shared library: `src/lib/CLAUDE.md` diff --git a/src/scheduler/testsuite/meson.build b/src/scheduler/testsuite/meson.build index f75815ae..d2bb1dd3 100644 --- a/src/scheduler/testsuite/meson.build +++ b/src/scheduler/testsuite/meson.build @@ -2974,6 +2974,34 @@ test('scheduler trace error paths test', sched_trace_error_paths_test, suite: 'scheduler', env: ['ASAN_OPTIONS=detect_leaks=0']) +# scheduler-sim-decision-trace task 6.4 (Requirements 9.1, 9.3, 9.5; +# design.md "Testing Strategy -> Performance"): a 1000-host, maxTasks = 64 +# parallel job traced at full level by the real allocator stays within the +# default line bound with exact counters, and 10,000 traced decisions keep +# the trace store within 64 MiB. Every scheduler source but main.c is +# linked, except repl_trace_store.c, which the test #includes to read the +# store's entry and discarded-mark lookups. +sched_trace_bounds_test_srcs = ['test_trace_bounds.c'] +foreach f : scheduler_srcs + if f != 'src/main.c' and f != 'src/repl_trace_store.c' + sched_trace_bounds_test_srcs += ['../' + f] + endif +endforeach + +sched_trace_bounds_test = executable( + 'sched_trace_bounds_test', + sched_trace_bounds_test_srcs, + include_directories:[incdir], + dependencies: [scheduler_deps, unity_dep], + link_args: ['-Wl,--wrap=xcecli_start', + '-Wl,--wrap=xcecli_join'], + install: false) + +test('scheduler trace bounds test', + sched_trace_bounds_test, suite: 'scheduler', + timeout: 300, + env: ['ASAN_OPTIONS=detect_leaks=0']) + # scheduler-sim-decision-trace task 6.3 (Requirements 10.1-10.5, 11.4; # design.md "Testing Strategy" -> "Equivalence Tests (Requirement 10)"): # tracing never changes scheduling. Runs the built xce_scheduler diff --git a/src/scheduler/testsuite/test_trace_bounds.c b/src/scheduler/testsuite/test_trace_bounds.c new file mode 100644 index 00000000..15fb6122 --- /dev/null +++ b/src/scheduler/testsuite/test_trace_bounds.c @@ -0,0 +1,478 @@ +/* + * test_trace_bounds.c - decision-trace size and memory bounds under large + * traces (scheduler-sim-decision-trace spec, task 6.4; design.md "Testing + * Strategy -> Performance", "TraceModel" (line bound, counters), + * "TraceStore" (64 MiB bound); Requirements 9.1, 9.3, 9.5). + * + * Large trace (9.1, 9.3): the real scheduler (sched_init(): real plugins, + * real allocator thread, real ingest) with 1000 workers, of which every + * 16th has one free slot, and one parallel job with maxTasks = 64. parjob + * tries 64 plans; each scans the 1000 hosts, so the full-level trace has + * tens of thousands of budgeted lines. The job pends (63 one-slot hosts + * for 64 tasks), which changes no scheduler state, so it is fed twice: + * once with an effectively unbounded line bound (INT_MAX) and once with the + * default bound (1000). The bounded record must store exactly the bound, + * count every other line as omitted (stored + omitted == the unbounded + * record's lines) and keep every decision-wide counter identical to the + * unbounded run. Its rendering shows the truncation marker and the totals. + * + * Store bound (9.5): 10,000 traced decisions put into the REPL trace + * store: first the real bounded record from the large trace, then records + * built through the trace model's hooks (the calls the allocator hooks + * make), a quarter of them filled up to the default line bound. After + * every put the store's bytes stay within 64 MiB and the record just put + * is stored; the oldest are discarded first. + * + * Every scheduler source except main.c is linked as its own translation + * unit, except repl_trace_store.c, which is #included so the test can read + * its stored-entry and discarded-mark lookups. + */ +#define _GNU_SOURCE +#include +#include +#include +#include +#include +#include + +#include + +#include "client.h" +#include "controller.capnp.h" +#include "core.h" +#include "decision.h" +#include "repl_internal.h" +#include "resobj.h" +#include "sched/alloc_iter.h" +#include "sched/trace.h" +#include "scheduler.h" +#include "xce/cluster.h" + +#include "../src/repl_trace_store.c" + +extern int scheduler_req_handler(xce_request_t*, void*); + +#define NWORKERS 1000 +#define FREE_EVERY 16 /* every 16th worker has one free slot: 63 of 1000 */ +#define MAX_TASKS 64 +#define QUEUE_NAME "bounds_queue" +#define JOB_LARGE ((uint64_t)664001ULL) +#define NDECISIONS 10000 +#define STORE_JOB0 ((uint64_t)665000ULL) + +int __wrap_xcecli_start(xcecli_ctx_t c) +{ + (void)c; + return 0; +} + +int __wrap_xcecli_join(xcecli_ctx_t c, int role, hostattrs_t* attrs) +{ + (void)c; + (void)role; + (void)attrs; + return 0; +} + +void setUp(void) { } + +void tearDown(void) { } + +/* ---- fixture ------------------------------------------------------------ */ + +static xce_scheduler_t* g_sched; +static xce_sched_trace_ctl_t g_ctl; +static xce_sched_trace_sel_t g_sel; +static xce_sched_trace_rec_t* g_bounded; /* the large trace at the default + * bound, kept for the store run */ + +static xce_worker_t* make_wire_worker(const char* name, float slots) +{ + xce_worker_t* w = (xce_worker_t*)calloc(1, sizeof(*w)); + xce_val_t* val = (xce_val_t*)calloc(1, sizeof(*val)); + xce_name_val_t* nv = (xce_name_val_t*)calloc(1, sizeof(*nv)); + xce_name_val_t** nv_ptrs = (xce_name_val_t**)calloc(1, sizeof(*nv_ptrs)); + + TEST_ASSERT_NOT_NULL(w); + TEST_ASSERT_NOT_NULL(val); + TEST_ASSERT_NOT_NULL(nv); + TEST_ASSERT_NOT_NULL(nv_ptrs); + + val->which = Value_u_fvalue; + val->u.fvalue = slots; + nv->name = strdup("slots"); + nv->val = val; + nv_ptrs[0] = nv; + + w->name = strdup(name); + w->status = 0; + w->nresvals_to_sched = 1; + w->resvals_to_sched = nv_ptrs; + return w; +} + +/* Real sched_init() with the 1000 workers, one queue and the parallel job + * ingested, a session begun and full-level settings selecting the job + * published. The unsatisfiable cache is not registered (as in simulation), + * so feeding the pending job again recomputes it. Never torn down (the + * real threads keep running), the convention of the other real-scheduler + * tests. */ +static void start_sched(void) +{ + static xce_client_config_t config; + static xce_cluster_config_t cluster_cfg; + xce_core_event_t ev; + xce_worker_t** workers; + xce_job_t* jobs[1]; + xce_job_t* j; + xce_queue_t* q = (xce_queue_t*)calloc(1, sizeof(*q)); + xce_resource_t res; + char name[32]; + int rc; + int i; + + g_sched = (xce_scheduler_t*)calloc(1, sizeof(*g_sched)); + TEST_ASSERT_NOT_NULL(g_sched); + + memset(&config, 0, sizeof(config)); + config.nmasters = 1; + config.masters[0] = "127.0.0.1"; + config.master_ports[0] = 19994; + config.from = "sched_trace_bounds_test"; + config.key = "keepsecret$"; + + memset(&cluster_cfg, 0, sizeof(cluster_cfg)); + cluster_cfg.candidates_count = 1; + strncpy(cluster_cfg.candidates[0].hostname, "127.0.0.1", + XCE_MAXHOSTNAME_LEN - 1); + cluster_cfg.candidates[0].port = 19994; + strncpy(cluster_cfg.identity, "sched_trace_bounds_test", + sizeof(cluster_cfg.identity) - 1); + + rc = sched_init(g_sched, &config, &cluster_cfg, scheduler_req_handler); + TEST_ASSERT_TRUE_MESSAGE(rc >= 0, "fixture: sched_init() must succeed"); + + memset(&res, 0, sizeof(res)); + res.name = strdup("slots"); + TEST_ASSERT_NOT_NULL(makeResource(g_sched->core.resmgr, &res)); + + TEST_ASSERT_NOT_NULL(q); + q->queue = strdup(QUEUE_NAME); + q->status = QUEUE_STAT_OPEN; + q->priority = 5; + memset(&ev, 0, sizeof(ev)); + ev.type = XCE_CORE_EVENT_QUEUE_UPDATE; + ev.u.queue_update.queue = q; + TEST_ASSERT_TRUE(xce_core_push_events(&(g_sched->core), &ev, 1) >= 0); + + workers = (xce_worker_t**)calloc(NWORKERS, sizeof(*workers)); + TEST_ASSERT_NOT_NULL(workers); + for (i = 0; i < NWORKERS; i++) { + snprintf(name, sizeof(name), "bh%04d", i); + workers[i] + = make_wire_worker(name, ((i % FREE_EVERY) == 0) ? 1.0f : 0.0f); + } + memset(&ev, 0, sizeof(ev)); + ev.type = XCE_CORE_EVENT_WORKERS_PUSH; + ev.u.workers_push.nworkers = NWORKERS; + ev.u.workers_push.workers = workers; + TEST_ASSERT_TRUE(xce_core_push_events(&(g_sched->core), &ev, 1) >= 0); + + j = (xce_job_t*)calloc(1, sizeof(*j)); + TEST_ASSERT_NOT_NULL(j); + j->jobid = JOB_LARGE; + j->state = JOB_STAT_PEND; + j->sub_time = 1000; + j->queue_time = 1234; + j->submit = (job_submit_data_t*)calloc(1, sizeof(*(j->submit))); + TEST_ASSERT_NOT_NULL(j->submit); + j->submit->queue = (char*)QUEUE_NAME; + j->submit->user = (char*)"bounds_user"; + j->submit->project_name = (char*)"bounds_proj"; + j->submit->resreq = (char*)"rusage[mem=1]"; + j->submit->mintasks = 1; + j->submit->maxtasks = MAX_TASKS; + jobs[0] = j; + memset(&ev, 0, sizeof(ev)); + ev.type = XCE_CORE_EVENT_JOBS_PUSH; + ev.u.jobs_push.njobs = 1; + ev.u.jobs_push.jobs = jobs; + TEST_ASSERT_TRUE(xce_core_push_events(&(g_sched->core), &ev, 1) >= 0); + + TEST_ASSERT_TRUE(xce_core_begin_session(&(g_sched->core)) >= 0); + + memset(&g_ctl, 0, sizeof(g_ctl)); + g_sel.key = JOB_LARGE; + g_sel.whole_array = false; + g_ctl.nsel = 1; + g_ctl.sel = &g_sel; + g_ctl.level = XCE_TRACE_LEVEL_FULL; + g_ctl.line_bound = XCE_TRACE_LINE_BOUND_DEFAULT; + g_ctl.session_no = 1; + g_ctl.policy_name = "builtin_fcfs"; + g_sched->core.trace_ctl = &g_ctl; +} + +/* Feeds the large job to the allocator thread with line bound `bound` and + * returns the record its decision carries (a pend). Frees the decision. */ +static xce_sched_trace_rec_t* feed_large(int bound) +{ + uint64_t jobid = JOB_LARGE; + xce_decision_t* decision = NULL; + xce_sched_trace_rec_t* rec; + int waited_ms; + + g_ctl.line_bound = bound; /* the allocator thread is idle on jbuf */ + TEST_ASSERT_TRUE(cbuf_tryput(&(g_sched->core.jbuf), &jobid) >= 0); + for (waited_ms = 0; waited_ms < 120000; waited_ms += 5) { + struct timespec ts = { 0, 5 * 1000 * 1000 }; + + if (cbuf_tryget(&(g_sched->core.decisions), &decision) >= 0) { + break; + } + nanosleep(&ts, NULL); + } + TEST_ASSERT_NOT_NULL_MESSAGE(decision, "the allocator pushed nothing"); + TEST_ASSERT_EQUAL_INT_MESSAGE(XCE_DECISION_PEND, decision->op, + "the large job must pend"); + rec = decision->trace; + decision->trace = NULL; + free(decision); + TEST_ASSERT_NOT_NULL_MESSAGE(rec, "the decision carries no trace"); + TEST_ASSERT_EQUAL_UINT64(JOB_LARGE, rec->jobid); + TEST_ASSERT_FALSE_MESSAGE(rec->incomplete, "the record is incomplete"); + return rec; +} + +static int count_entries(const xce_sched_trace_rec_t* rec) +{ + const sched_trace_entry_t* e; + int n = 0; + + for (e = rec->entries; e != NULL; e = e->next) { + n++; + } + return n; +} + +static char* render(const xce_sched_trace_rec_t* rec) +{ + char* buf = NULL; + size_t len = 0; + FILE* f = open_memstream(&buf, &len); + + TEST_ASSERT_NOT_NULL(f); + repl_trace_render(rec, XCE_REPL_UNITS_RAW, f, NULL); + fclose(f); + return buf; +} + +static int count_lines(const char* text) +{ + int n = 0; + + for (; *text != '\0'; text++) { + n += (*text == '\n'); + } + return n; +} + +/* ---- large trace (9.1, 9.3) -------------------------------------------- */ + +static void test_large_parallel_job_trace_stays_within_default_bound(void) +{ + xce_sched_trace_rec_t* full = feed_large(INT_MAX); + xce_sched_trace_rec_t* rec = feed_large(XCE_TRACE_LINE_BOUND_DEFAULT); + const sched_trace_counters_t* fc = &full->counters; + const sched_trace_counters_t* bc = &rec->counters; + char marker[128]; + char* full_text; + char* text; + int total_lines; + int evals = 0; + int k; + + /* The unbounded run stores every line: the generated line count. */ + TEST_ASSERT_EQUAL_INT(0, fc->omitted_lines); + TEST_ASSERT_EQUAL_INT(full->nentries, count_entries(full)); + total_lines = full->nentries; + TEST_ASSERT_TRUE_MESSAGE(total_lines > 10 * XCE_TRACE_LINE_BOUND_DEFAULT, + "the large trace must be far over the bound"); + + /* 9.1: the stored entries stay within the default bound. */ + TEST_ASSERT_EQUAL_INT(XCE_TRACE_LINE_BOUND_DEFAULT, rec->line_bound); + TEST_ASSERT_EQUAL_INT(XCE_TRACE_LINE_BOUND_DEFAULT, rec->nentries); + TEST_ASSERT_EQUAL_INT(rec->nentries, count_entries(rec)); + + /* 9.2/9.3: every line past the bound is counted as omitted, exactly. */ + TEST_ASSERT_EQUAL_INT(total_lines - rec->nentries, bc->omitted_lines); + + /* 9.3: the decision-wide counters are exact under truncation. */ + TEST_ASSERT_EQUAL_INT(fc->plans_tried, bc->plans_tried); + TEST_ASSERT_EQUAL_INT(fc->plans_discarded, bc->plans_discarded); + TEST_ASSERT_EQUAL_INT(fc->reqs_none_generated, bc->reqs_none_generated); + TEST_ASSERT_EQUAL_INT(fc->reqs_all_discarded, bc->reqs_all_discarded); + TEST_ASSERT_EQUAL_INT(fc->reqs_none_placed, bc->reqs_none_placed); + TEST_ASSERT_EQUAL_INT_ARRAY(fc->host_evals, bc->host_evals, + XCE_TRACE_V_NKINDS); + + /* What parjob did: one plan per host count 1..64, none placed, every + * plan's scan over the 1000 hosts; plan 64 accepts the 63 free hosts + * and finds the 64th position unfilled. */ + TEST_ASSERT_EQUAL_INT(MAX_TASKS, bc->plans_tried); + TEST_ASSERT_EQUAL_INT(0, bc->plans_discarded); + TEST_ASSERT_EQUAL_INT(1, bc->reqs_none_placed); + TEST_ASSERT_EQUAL_INT(NWORKERS / FREE_EVERY + 1, bc->host_evals[0]); + for (k = 0; k < XCE_TRACE_V_NKINDS; k++) { + evals += bc->host_evals[k]; + } + TEST_ASSERT_TRUE_MESSAGE(evals >= (MAX_TASKS - 1) * NWORKERS, + "every plan must have evaluated the hosts"); + + /* The rendering shows the marker and the totals; the bounded text is a + * fraction of the unbounded one. */ + text = render(rec); + full_text = render(full); + if (getenv("XCE_TRACE_BOUNDS_DUMP") != NULL) { + fprintf(stderr, "%s", text); + } + snprintf(marker, sizeof(marker), + "\n ... %d trace lines omitted (line bound %d)\n", + bc->omitted_lines, XCE_TRACE_LINE_BOUND_DEFAULT); + TEST_ASSERT_NOT_NULL_MESSAGE(strstr(text, marker), "truncation marker"); + TEST_ASSERT_NOT_NULL(strstr(text, "\n totals: plans tried 64, plans " + "discarded 0\n")); + TEST_ASSERT_NOT_NULL(strstr(text, "\n totals: host evaluations: ")); + TEST_ASSERT_NULL(strstr(full_text, "trace lines omitted")); + TEST_ASSERT_TRUE(count_lines(text) < 2 * XCE_TRACE_LINE_BOUND_DEFAULT); + TEST_ASSERT_TRUE(count_lines(full_text) >= total_lines); + free(text); + free(full_text); + + sched_trace_rec_free(full); + g_bounded = rec; +} + +/* ---- store bound (9.5) -------------------------------------------------- */ + +static xce_sched_worker_t g_hosts[1200]; +static char g_host_names[1200][16]; + +/* A pending parallel job's trace built through the model hooks, in the + * order the allocator's hooks make them: requirement, one plan, one + * position, `nhosts` hosts each rejected with one unmet slots row, the + * position and plan ends and the search end. At the default bound, more + * than ~500 hosts truncate it. */ +static xce_sched_trace_rec_t* build_rec(uint64_t jobid, int nhosts) +{ + xce_sched_trace_ctl_t ctl; + xce_sched_trace_rec_t* rec; + xce_alloc_plan_t plan; + int slots[2] = { 32, 32 }; + sched_trace_row_t row; + int i; + + memset(&ctl, 0, sizeof(ctl)); + ctl.all = true; + ctl.level = XCE_TRACE_LEVEL_FULL; + ctl.line_bound = XCE_TRACE_LINE_BOUND_DEFAULT; + ctl.session_no = 1; + rec = sched_trace_rec_new(&ctl, jobid, XCE_TRACE_REC_DECISION); + TEST_ASSERT_NOT_NULL(rec); + + memset(&plan, 0, sizeof(plan)); + plan.nbuckets = 2; + plan.bucket_slots = slots; + memset(&row, 0, sizeof(row)); + row.res = "slots"; + row.total = 1.0f; + row.inuse = 1.0f; + row.requested = 32.0f; + row.met = false; + + sched_trace_path_push(rec, XCE_TRACE_PATH_SINGLE, 1, 1); + sched_trace_resreq_begin(rec, NULL, 0); + sched_trace_plan_begin(rec, &plan); + sched_trace_pos_begin(rec, 1, 32); + for (i = 0; i < nhosts; i++) { + sched_trace_host_begin(rec, NULL, &g_hosts[i]); + sched_trace_row(rec, &row); + sched_trace_host_verdict(rec, XCE_TRACE_V_SHORTFALL, 0); + } + sched_trace_pos_end(rec, XCE_TRACE_POS_UNFILLED, NULL); + sched_trace_plan_end(rec, 0, 1); + sched_trace_search_end(rec, 0); + sched_trace_resreq_end(rec, 0, 0); + sched_trace_path_pop(rec); + TEST_ASSERT_FALSE(rec->incomplete); + TEST_ASSERT_TRUE(rec->nentries <= rec->line_bound); + return rec; +} + +static void test_store_stays_within_64mib_over_10000_decisions(void) +{ + xce_repl_trace_state_t ts; + size_t max_bytes = 0; + int i; + + TEST_ASSERT_NOT_NULL_MESSAGE(g_bounded, "the large trace test ran first"); + for (i = 0; i < 1200; i++) { + snprintf(g_host_names[i], sizeof(g_host_names[i]), "sh%04d", i); + g_hosts[i].name = g_host_names[i]; + } + + memset(&ts, 0, sizeof(ts)); + for (i = 0; i < NDECISIONS; i++) { + xce_sched_trace_rec_t* rec; + uint64_t jobid; + size_t bytes; + + if (i == 0) { + rec = g_bounded; /* the real large trace, first in */ + g_bounded = NULL; + } else { + /* a quarter filled to the line bound, the rest smaller */ + rec = build_rec(STORE_JOB0 + (uint64_t)i, + ((i % 4) == 0) ? 1200 : 20 + (i % 200)); + } + jobid = rec->jobid; + repl_trace_store_put(&ts, rec); + + bytes = repl_trace_store_bytes(&ts); + if (bytes > max_bytes) { + max_bytes = bytes; + } + TEST_ASSERT_TRUE_MESSAGE(bytes <= XCE_REPL_TRACE_STORE_BOUND, + "stored traces over 64 MiB"); + /* the trace just captured is never discarded */ + TEST_ASSERT_NOT_NULL(stored_entry(&ts, jobid)); + TEST_ASSERT_TRUE(ts.newest->rec == rec); + TEST_ASSERT_FALSE(is_discarded(&ts, jobid)); + } + + /* The bound was reached and the oldest went first: the real large + * trace (first in) is discarded, the newest decisions are stored. */ + TEST_ASSERT_TRUE_MESSAGE(max_bytes > XCE_REPL_TRACE_STORE_BOUND / 2, + "the run must reach the memory bound"); + TEST_ASSERT_TRUE(is_discarded(&ts, JOB_LARGE)); + TEST_ASSERT_NULL(stored_entry(&ts, JOB_LARGE)); + TEST_ASSERT_TRUE(is_discarded(&ts, STORE_JOB0 + 1)); + TEST_ASSERT_NOT_NULL( + stored_entry(&ts, STORE_JOB0 + (uint64_t)(NDECISIONS - 2))); + if (getenv("XCE_TRACE_BOUNDS_DUMP") != NULL) { + fprintf(stderr, "store: peak %zu bytes, final %zu bytes\n", max_bytes, + repl_trace_store_bytes(&ts)); + } + + repl_trace_store_fini(&ts); + TEST_ASSERT_EQUAL_size_t(0, repl_trace_store_bytes(&ts)); +} + +int main(void) +{ + UNITY_BEGIN(); + start_sched(); + RUN_TEST(test_large_parallel_job_trace_stays_within_default_bound); + RUN_TEST(test_store_stays_within_64mib_over_10000_decisions); + return UNITY_END(); +}