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(); +}