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

test(scheduler-sim-decision-trace): error-path trace scenarios with the real allocator

Browse source 
Task 6.2: injects plan-claims growth, snapshot, finalize, host-scan,
per-host uncommitted lookup, allocation-function and unknown-jobid
faults and asserts each traced outcome. Also fixes two deferred OOM
paths: a failed limit row keeps the host's verdict bits, and an array
listing that cannot be built is reported instead of silently truncated.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
This commit is contained in:
Rongsong Shen 2026-10-08 16:28:17 +00:00
parent aa401627a8
commit 7ac988a7f5
Signed by untrusted user: rshen
GPG key ID: 80533536B0EB651C
7 changed files with 896 additions and 20 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -316,7 +316,7 @@
- Done when the end-to-end test passes under ASAN.
- _Requirements: 1.1, 1.2, 1.6, 2.1, 2.2, 2.3, 3.1, 3.4, 4.1, 4.2, 4.3, 4.5, 5.1, 5.3, 5.4, 5.5, 5.6, 5.8, 5.9, 6.1, 6.2, 6.4, 7.1, 7.3, 7.4, 7.5, 7.8, 7.9, 8.1_
- [ ] 6.2 Error-path trace scenarios with the real allocator
- [x] 6.2 Error-path trace scenarios with the real allocator
- Include the scheduler sources directly, so source-level and link-level fault injection apply.
- Inject each fault and assert its traced outcome:
- a plan-claims growth failure, through the heap-allocator macro;
@ -374,3 +374,4 @@
- 5.2: repl.h now has the interactive-only gates `xce_repl_trace_session_begin/end(sim)` and `xce_repl_check_trace/xce_repl_check_trace_only(sim, d)`; the two check gates also no-op when d->trace is NULL. 5.3 calls these from sim_loop.c. `trace` is routed with the raw line, `info trace` is in usage and help, `set output` checks `repl_trace_file_conflicts` first, and quit/EOF call `repl_trace_shutdown`. The continue-mode rollup prints a traced round on its own via `rs->trace.round_traced`. The baseline transcript changed only in the help and info-usage lines; regenerating also rewrites the .replay fixtures' wall-clock fields, so restore those. Pre-existing: repl_count_jobs (repl.c) never destroys its iterator, leaking 16 B per continue round; the rollup tests run with detect_leaks=0.
- 5.3: sim_loop.c calls the repl.h gates. Session begin runs after a successful begin-session and before reconcile/signal_go; session end runs after the drain and verify tail, before plugin_end_session. The drain loop handles TRACE first (skipping verify, record, echo and breakpoint), then echo gate, `check_trace`, `check_decision`, `rec_free`, free. The reconstructed block uses the new gate `xce_repl_trace_reconstructed` (`repl_trace_on_reconstructed` in repl_trace_emit.c). Settles the 1.7 OPEN item (both free sites). A reviewer's trace-on vs trace-off run of the contention fixture gave byte-identical --sim-output, identical verify and identical decisions.
- 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 <id>: cannot list its elements: <cause>". The capture and store tests now #include their sources.

44
src/scheduler/src/repl_trace_store.c
View file

@ -30,6 +30,7 @@
* 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 <errno.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
@ -219,10 +220,13 @@ void repl_trace_store_fini(xce_repl_trace_state_t* ts)
/* ---- info trace ----------------------------------------------------------- */
/* `failed` is set once an add could not grow the list; the list is then
* incomplete and must be reported, never printed short (7.9). */
typedef struct {
uint64_t* ids;
int n;
int cap;
bool failed;
} id_list_t;
static bool id_list_add(id_list_t* l, uint64_t id)
@ -232,6 +236,7 @@ static bool id_list_add(id_list_t* l, uint64_t id)
uint64_t* p = (uint64_t*)realloc(l->ids, sizeof(*p) * (size_t)cap);
if (p == NULL) {
l->failed = true;
return false;
}
l->ids = p;
@ -274,8 +279,8 @@ static void collect_table(const htab_t* t, uint64_t base, uint32_t lo,
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);
if (in_array(id, base, lo, hi) && !id_list_add(out, id)) {
return;
}
}
}
@ -297,13 +302,25 @@ static void collect_core(xce_sched_core_t* core, uint64_t base, uint32_t lo,
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);
if (in_array(job->jobid, base, lo, hi)
&& !id_list_add(out, job->jobid)) {
break;
}
}
iter_destroy(&it);
}
/* An element list that could not be built in full (out of memory). */
static void print_list_failed(uint64_t base)
{
char idbuf[64];
fmt_jobid(idbuf, sizeof(idbuf), base);
fprintf(stdout,
"[sim] info trace: array %s: cannot list its elements: %s\n", idbuf,
strerror(ENOMEM));
}
static void print_discarded(uint64_t jobid)
{
char idbuf[64];
@ -350,15 +367,22 @@ 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 };
id_list_t traced = { NULL, 0, 0, false };
id_list_t states = { NULL, 0, 0, false };
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);
if (!traced.failed) {
collect_table(&ts->discarded, base, lo, hi, &traced);
}
}
if (traced.failed) {
print_list_failed(base);
free(traced.ids);
return true;
}
qsort(traced.ids, (size_t)traced.n, sizeof(uint64_t), id_cmp);
@ -376,6 +400,12 @@ static bool info_trace_array(xce_sched_sim_driver_t* sim,
if (nstored == 0) {
collect_core(&(sim->core), base, lo, hi, &states);
if (states.failed) {
print_list_failed(base);
free(traced.ids);
free(states.ids);
return true;
}
if ((traced.n == 0) && (states.n == 0)) {
free(traced.ids);
return false;

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

@ -406,8 +406,14 @@ void sched_trace_check_rows(xce_sched_trace_rec_t* rec, xce_sched_core_t* core,
if ((r != NULL) && ((r->flags & XCE_RESF_LIMIT) != 0)) {
if (!limit_row(rec, r, &row, &buf)) {
rec->incomplete = true; /* design "Error Strategy" */
return;
/* design "Error Strategy": the record stops appending, but
* the host's verdict and the counts (9.3) stay exact, so
* the limit still counts and the later rows still run */
rec->incomplete = true;
if (host_scope) {
sched_trace_host_limit_hit(rec, NULL, NULL);
}
continue;
}
/* a limit rejection, never a shortfall (4.2, 5.5) */
if (host_scope) {

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

@ -2617,12 +2617,12 @@ test('scheduler decision trace test',
# "TraceCapture") over a fixture core and job -- selection facts,
# cluster-wide limits and attempt failure reasons. Links the record model
# and the capture sources (split in task 2.4); libxce supplies the
# pend-reason helpers.
# pend-reason helpers. sched_trace_capture.c is #included by the test, so
# its malloc() can be renamed for the limit-row out-of-memory case (6.2).
sched_trace_capture_test = executable(
'sched_trace_capture_test',
['test_trace_capture.c', '../src/sched_trace.c',
'../src/sched_trace_scan.c', '../src/sched_trace_capture.c',
'../src/sched_trace_outcome.c'],
'../src/sched_trace_scan.c', '../src/sched_trace_outcome.c'],
include_directories:[incdir],
dependencies: [scheduler_test_deps],
install: false)
@ -2794,13 +2794,14 @@ test('scheduler trace render test',
# 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.
# repl_job_state_name() defined by the test. The store is #included by the
# test, so its realloc() can be renamed for the listing out-of-memory case
# (6.2).
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'],
'../src/repl_trace_render_outcome.c', '../src/repl_fmt.c'],
include_directories:[incdir],
dependencies: [scheduler_test_deps],
install: false)
@ -2935,3 +2936,40 @@ sched_sim_trace_driver_test = executable(
test('scheduler sim trace driver test',
sched_sim_trace_driver_test, suite: 'scheduler',
env: ['ASAN_OPTIONS=detect_leaks=0'])
# scheduler-sim-decision-trace task 6.2 (Requirements 2.4, 3.2, 4.4, 4.5,
# 5.7): error-path traces with the real allocator. The hooked sources are
# #included by the test so source-level renames apply (realloc in
# alloc_runner.c; iter_next and the per-host uncommitted lookup in
# host_iter.c); every other scheduler source is its own translation unit,
# and xvec_dup (claim snapshot) and make_host_alloc/make_job_alloc
# (finalize) are link-wrapped to fail on demand.
sched_trace_error_paths_included = ['src/main.c', 'src/sched_election.c',
'src/scheduler.c', 'src/alloc_runner.c',
'src/host_iter.c', 'src/resource.c',
'src/sched_trace.c',
'src/sched_trace_scan.c',
'src/sched_trace_capture.c',
'src/sched_trace_outcome.c']
sched_trace_error_paths_test_srcs = ['test_trace_error_paths.c']
foreach f : scheduler_srcs
if not sched_trace_error_paths_included.contains(f)
sched_trace_error_paths_test_srcs += ['../' + f]
endif
endforeach
sched_trace_error_paths_test = executable(
'sched_trace_error_paths_test',
sched_trace_error_paths_test_srcs,
include_directories:[incdir],
dependencies: [scheduler_deps, unity_dep],
link_args: ['-Wl,--wrap=xcecli_start',
'-Wl,--wrap=xcecli_join',
'-Wl,--wrap=xvec_dup',
'-Wl,--wrap=make_host_alloc',
'-Wl,--wrap=make_job_alloc'],
install: false)
test('scheduler trace error paths test',
sched_trace_error_paths_test, suite: 'scheduler',
env: ['ASAN_OPTIONS=detect_leaks=0'])

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

@ -40,9 +40,15 @@
* - with 50 or fewer hosts the pend copy equals its source under
* job_preason_eq(), the equality libxce defines for pend reasons.
*
* Links the trace model (src/sched_trace.c, sched_trace_scan.c) and the
* capture sources (src/sched_trace_capture.c, sched_trace_outcome.c), plus
* libxce for the pend-reason, resmgr and container helpers.
* Task 6.2 adds the out-of-memory path of a limit row: the record is
* marked incomplete, yet the failed limit row still sets the host's LIMIT
* verdict and the later rows their SHORTFALL verdict, so the derived
* verdict and the decision-wide counts (9.3) stay exact.
*
* Links the trace model (src/sched_trace.c, sched_trace_scan.c) and
* src/sched_trace_outcome.c, and #includes src/sched_trace_capture.c with
* malloc() renamed for the out-of-memory seam, plus libxce for the
* pend-reason, resmgr and container helpers.
*/
#include <stdio.h>
#include <stdlib.h>
@ -59,6 +65,25 @@
#include "xstring.h"
#include "zarray.h"
/* Out-of-memory seam (task 6.2): the capture source is #included rather
* than linked, so this rename reaches its scratch-buffer allocations and
* nothing else. Armed with the 1-based call number that fails. */
static int g_malloc_calls;
static int g_malloc_fail_on;
static void* test_malloc(size_t n);
#define malloc test_malloc
#include "../src/sched_trace_capture.c"
#undef malloc
static void* test_malloc(size_t n)
{
if ((g_malloc_fail_on > 0) && (++g_malloc_calls == g_malloc_fail_on)) {
g_malloc_fail_on = 0;
return NULL;
}
return malloc(n);
}
void setUp(void) { return; }
void tearDown(void) { return; }
@ -970,6 +995,61 @@ static void test_check_rows_host_summary_tallies_limit(void)
rows_fx_free(fx);
}
/* Task 6.2 (2.2 OPEN): a limit row whose scratch buffer cannot be
* allocated marks the record incomplete, but its LIMIT verdict and the
* later rows' SHORTFALL verdict are still set, so the host's derived
* verdict is LIMIT + SHORTFALL (not a selection mismatch) and the
* decision-wide counts stay exact (9.3). */
static void test_check_rows_host_limit_row_oom_keeps_verdicts(void)
{
rows_fx_t* fx = rows_fx_new();
xce_sched_trace_ctl_t ctl = make_ctl("fcfs");
xce_sched_trace_rec_t* rec;
int idxs[2] = { fx->lim_host->idx, fx->mem->idx };
float total[2] = { 10, 64 };
float inuse[2] = { 9, 60 };
float wanted[2] = { 2, 8 };
rec = sched_trace_rec_new(&ctl, 0x77, XCE_TRACE_REC_DECISION);
TEST_ASSERT_NOT_NULL(rec);
sched_trace_host_begin(rec, &fx->obj, &fx->w);
g_malloc_calls = 0;
g_malloc_fail_on = 1; /* the limit row's scratch buffer */
sched_trace_check_rows(rec, &fx->core, 2, idxs, total, inuse, NULL,
wanted);
TEST_ASSERT_EQUAL_INT_MESSAGE(0, g_malloc_fail_on,
"the limit row never allocated");
TEST_ASSERT_TRUE(rec->incomplete);
TEST_ASSERT_EQUAL_UINT(XCE_TRACE_V_SHORTFALL | XCE_TRACE_V_LIMIT,
rec->host_row_verdicts);
sched_trace_host_verdict(rec, 0, 0);
TEST_ASSERT_EQUAL_INT(0, rec->counters.host_evals[1]); /* SELECT */
TEST_ASSERT_EQUAL_INT(1, rec->counters.host_evals[2]); /* SHORTFALL */
TEST_ASSERT_EQUAL_INT(1, rec->counters.host_evals[3]); /* LIMIT */
sched_trace_rec_free(rec);
/* summary level: the same bits reach the position's tallies */
ctl.level = XCE_TRACE_LEVEL_SUMMARY;
rec = sched_trace_rec_new(&ctl, 0x77, XCE_TRACE_REC_DECISION);
TEST_ASSERT_NOT_NULL(rec);
sched_trace_pos_begin(rec, 1, 1);
sched_trace_host_begin(rec, &fx->obj, &fx->w);
g_malloc_calls = 0;
g_malloc_fail_on = 1;
sched_trace_check_rows(rec, &fx->core, 2, idxs, total, inuse, NULL,
wanted);
TEST_ASSERT_TRUE(rec->incomplete);
TEST_ASSERT_EQUAL_UINT(XCE_TRACE_V_SHORTFALL | XCE_TRACE_V_LIMIT,
rec->host_row_verdicts);
sched_trace_host_verdict(rec, 0, 0);
TEST_ASSERT_EQUAL_INT(0, rec->counters.host_evals[1]);
TEST_ASSERT_EQUAL_INT(1, rec->counters.host_evals[2]);
TEST_ASSERT_EQUAL_INT(1, rec->counters.host_evals[3]);
sched_trace_rec_free(rec);
rows_fx_free(fx);
}
/* Rows outside any check scope are not attributed to anything; a resource
* index the resmgr does not know is kept as an unnamed resource row. */
static void test_check_rows_scope_edges(void)
@ -1586,6 +1666,7 @@ int main(void)
RUN_TEST(test_check_rows_host_summary_tallies_limit);
RUN_TEST(test_check_rows_scope_edges);
RUN_TEST(test_check_rows_host_rows_count_against_budget);
RUN_TEST(test_check_rows_host_limit_row_oom_keeps_verdicts);
RUN_TEST(test_dispatch_separates_limit_charges);
RUN_TEST(test_dispatch_raw_number_fallback);
RUN_TEST(test_reconstructed_record_holds_dispatch);

662
src/scheduler/testsuite/test_trace_error_paths.c Normal file
View file

@ -0,0 +1,662 @@
/*
* test_trace_error_paths.c - error-path decision traces with the real
* allocator (scheduler-sim-decision-trace spec, task 6.2; design.md
* "Testing Strategy -> Integration Tests" item 3, "Error Handling",
* "TraceRenderer", "TraceHooks"; Requirements 2.4, 3.2, 4.4, 4.5, 5.7).
*
* Runs the real scheduler (sched_init(): real plugins, real picker and
* allocator threads, real ingest) with full-level session settings
* published on the core, feeds jobids straight into core->jbuf and renders
* the record each decision (or trace-only decision) carries with the REPL
* renderer, asserting the lines each injected fault must produce:
*
* - plan-claims growth failure (4.4): realloc() renamed inside
* alloc_runner.c fails, so plan 1 is not tried and the attempt ends in
* an internal error;
* - snapshot failure (4.4): the link-wrapped xvec_dup() fails after the
* host was accepted, so the position is not filled because of an
* internal error after accepting that host;
* - finalize failures (4.5): the link-wrapped make_host_alloc() and
* make_job_alloc() fail, so the filled plan ended in an internal error
* building the allocation, not committed;
* - a failing host scan (4.4): iter_next() renamed inside host_iter.c
* fails after one host was considered, so that host is shown and the
* position is not filled because of the scan error;
* - a failing per-host uncommitted lookup (5.7): the pre-filter's
* sched_ctx_get_uncomit_host(), renamed inside host_iter.c, fails for
* the first host, which is not evaluated; the next host is accepted;
* - an allocation function returning an error (2.4): a stub sequential
* allocator returns -EIO;
* - an unknown jobid put directly on jbuf (2.4).
* Every attempt that ends in an internal error shows its requirement as
* ended in an internal error (3.2) and ends with the no-decision line.
*
* The hooked sources (sched_election.c, scheduler.c, alloc_runner.c,
* host_iter.c, resource.c and the four trace sources) are #included here,
* so the source-level renames apply; the other scheduler sources are
* linked as their own translation units (meson.build), so the link-level
* wraps see every call.
*/
#define _GNU_SOURCE
#include <errno.h>
#include <stdatomic.h>
#include <stdint.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <time.h>
#include <unity/unity.h>
#include "client.h"
#include "controller.capnp.h"
#include "core.h"
#include "decision.h"
#include "plugin.h"
#include "preason.h"
#include "repl_internal.h"
#include "resobj.h"
#include "sched/alloc_cache.h"
#include "sched/trace.h"
#include "scheduler.h"
#include "xce/cluster.h"
#include "xvec.h"
/* ---- fault seams renamed inside the sources under test ----------------- */
static void* test_realloc(void* p, size_t n);
#define realloc test_realloc
#include "../src/alloc_runner.c"
#undef realloc
static int test_iter_next(iterator_t* it, void** item);
static int test_get_uncomit_host(xvec_t** host, xce_sched_ctx_t* ctx, int idx);
#define iter_next test_iter_next
#define sched_ctx_get_uncomit_host test_get_uncomit_host
#include "../src/host_iter.c"
#undef iter_next
#undef sched_ctx_get_uncomit_host
#include "../src/resource.c"
#include "../src/sched_election.c"
#include "../src/sched_trace.c"
#include "../src/sched_trace_capture.c"
#include "../src/sched_trace_outcome.c"
#include "../src/sched_trace_scan.c"
#include "../src/scheduler.c"
extern int scheduler_req_handler(xce_request_t*, void*);
#define WORKER_A_NAME "err-host-a"
#define WORKER_B_NAME "err-host-b"
#define WORKER_SLOTS 3.0f
#define QUEUE_NAME "err_queue"
#define QUEUE_PRIO 5
#define JOB_RESERVE ((uint64_t)662001ULL)
#define JOB_SNAPSHOT ((uint64_t)662002ULL)
#define JOB_FIN_HOST ((uint64_t)662003ULL)
#define JOB_FIN_JOB ((uint64_t)662004ULL)
#define JOB_SCAN ((uint64_t)662005ULL)
#define JOB_UNCOMIT ((uint64_t)662006ULL)
#define JOB_ALLOC_ERR ((uint64_t)662007ULL)
#define JOB_NOT_FOUND ((uint64_t)662999ULL) /* never ingested */
#define NJOBS 7
/* ---- faults -------------------------------------------------------------
* Each is armed by the test thread before it feeds the job and fires on
* the allocator thread; the jbuf/decisions cbufs order the two. A fault
* fires once, on the armed call number counted from arming, and disarms. */
static atomic_int g_realloc_fail_on;
static atomic_int g_realloc_calls;
static atomic_int g_iter_fail_on;
static atomic_int g_iter_calls;
static atomic_int g_uncomit_fail_on;
static atomic_int g_uncomit_calls;
static atomic_int g_dup_fail_on;
static atomic_int g_dup_calls;
static atomic_int g_halloc_fail_on;
static atomic_int g_halloc_calls;
static atomic_int g_jalloc_fail_on;
static atomic_int g_jalloc_calls;
/* true when the armed fault `fail_on` fires on this call */
static bool fault_fires(atomic_int* fail_on, atomic_int* calls)
{
if (atomic_load(fail_on) <= 0) {
return false;
}
if (atomic_fetch_add(calls, 1) + 1 == atomic_load(fail_on)) {
atomic_store(fail_on, 0);
return true;
}
return false;
}
static void arm(atomic_int* fail_on, atomic_int* calls, int n)
{
atomic_store(calls, 0);
atomic_store(fail_on, n);
}
static void* test_realloc(void* p, size_t n)
{
if (fault_fires(&g_realloc_fail_on, &g_realloc_calls)) {
errno = ENOMEM;
return NULL;
}
return realloc(p, n);
}
static int test_iter_next(iterator_t* it, void** item)
{
if (fault_fires(&g_iter_fail_on, &g_iter_calls)) {
return -EIO;
}
return iter_next(it, item);
}
static int test_get_uncomit_host(xvec_t** host, xce_sched_ctx_t* ctx, int idx)
{
if (fault_fires(&g_uncomit_fail_on, &g_uncomit_calls)) {
return -ENOMEM;
}
return sched_ctx_get_uncomit_host(host, ctx, idx);
}
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;
}
int __real_xvec_dup(xvec_t** dst, xvec_t* src);
int __wrap_xvec_dup(xvec_t** dst, xvec_t* src)
{
if (fault_fires(&g_dup_fail_on, &g_dup_calls)) {
*dst = NULL;
return -ENOMEM;
}
return __real_xvec_dup(dst, src);
}
xce_host_alloc_t* __real_make_host_alloc(char* host, int nres,
xce_alloc_t** allocs);
xce_host_alloc_t* __wrap_make_host_alloc(char* host, int nres,
xce_alloc_t** allocs)
{
if (fault_fires(&g_halloc_fail_on, &g_halloc_calls)) {
errno = ENOMEM;
return NULL;
}
return __real_make_host_alloc(host, nres, allocs);
}
xce_job_alloc_t* __real_make_job_alloc(uint64_t jobid, int nhost,
xce_host_alloc_t** hallocs);
xce_job_alloc_t* __wrap_make_job_alloc(uint64_t jobid, int nhost,
xce_host_alloc_t** hallocs)
{
if (fault_fires(&g_jalloc_fail_on, &g_jalloc_calls)) {
errno = ENOMEM;
return NULL;
}
return __real_make_job_alloc(jobid, nhost, hallocs);
}
/* Stub sequential-job allocator: an allocation function that fails */
static int stub_job_alloc(void* plugin, xce_sched_ctx_t* ctx)
{
(void)plugin;
(void)ctx;
return -EIO;
}
void setUp(void)
{
arm(&g_realloc_fail_on, &g_realloc_calls, 0);
arm(&g_iter_fail_on, &g_iter_calls, 0);
arm(&g_uncomit_fail_on, &g_uncomit_calls, 0);
arm(&g_dup_fail_on, &g_dup_calls, 0);
arm(&g_halloc_fail_on, &g_halloc_calls, 0);
arm(&g_jalloc_fail_on, &g_jalloc_calls, 0);
}
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[NJOBS + 1];
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;
}
static xce_job_t* make_push_job(uint64_t jobid, const char* resreq)
{
xce_job_t* j = (xce_job_t*)calloc(1, sizeof(*j));
TEST_ASSERT_NOT_NULL(j);
j->jobid = jobid;
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*)"err_user";
j->submit->project_name = (char*)"err_proj";
j->submit->resreq = (char*)resreq;
j->submit->mintasks = 1;
j->submit->maxtasks = 1;
return j;
}
/* Real sched_init() with two 3-slot workers, one queue and every test job
* ingested (distinct requirements, so no job hits another's
* unsatisfiable-cache entry), then a session begun and full-level settings
* published selecting every job. 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;
static const uint64_t ids[NJOBS + 1]
= { JOB_RESERVE, JOB_SNAPSHOT, JOB_FIN_HOST, JOB_FIN_JOB,
JOB_SCAN, JOB_UNCOMIT, JOB_ALLOC_ERR, JOB_NOT_FOUND };
xce_core_event_t ev;
xce_worker_t* workers[2];
xce_job_t* jobs[NJOBS];
xce_queue_t* q = (xce_queue_t*)calloc(1, sizeof(*q));
xce_resource_t res;
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] = 19995;
config.from = "sched_trace_error_paths_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 = 19995;
strncpy(cluster_cfg.identity, "sched_trace_error_paths_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));
xce_core_set_cache_clear_cb(&(g_sched->core), xce_alloc_cache_clear_cb);
TEST_ASSERT_NOT_NULL(q);
q->queue = strdup(QUEUE_NAME);
q->status = QUEUE_STAT_OPEN;
q->priority = QUEUE_PRIO;
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[0] = make_wire_worker(WORKER_A_NAME, WORKER_SLOTS);
workers[1] = make_wire_worker(WORKER_B_NAME, WORKER_SLOTS);
memset(&ev, 0, sizeof(ev));
ev.type = XCE_CORE_EVENT_WORKERS_PUSH;
ev.u.workers_push.nworkers = 2;
ev.u.workers_push.workers = workers;
TEST_ASSERT_TRUE(xce_core_push_events(&(g_sched->core), &ev, 1) >= 0);
jobs[0] = make_push_job(JOB_RESERVE, "rusage[mem=1]");
jobs[1] = make_push_job(JOB_SNAPSHOT, "rusage[mem=2]");
jobs[2] = make_push_job(JOB_FIN_HOST, "rusage[mem=3]");
jobs[3] = make_push_job(JOB_FIN_JOB, "rusage[mem=4]");
jobs[4] = make_push_job(JOB_SCAN, "rusage[mem=5]");
jobs[5] = make_push_job(JOB_UNCOMIT, "rusage[mem=6]");
jobs[6] = make_push_job(JOB_ALLOC_ERR, "rusage[mem=7]");
memset(&ev, 0, sizeof(ev));
ev.type = XCE_CORE_EVENT_JOBS_PUSH;
ev.u.jobs_push.njobs = NJOBS;
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));
memset(g_sel, 0, sizeof(g_sel));
for (i = 0; i < NJOBS + 1; i++) {
g_sel[i].key = ids[i];
}
g_ctl.nsel = NJOBS + 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 = 4;
g_ctl.policy_name = "builtin_fcfs";
g_ctl.position = 0;
g_sched->core.trace_ctl = &g_ctl;
}
/* Feeds jobid to the allocator thread and returns the next entry it pushes
* onto core->decisions (bounded poll). */
static xce_decision_t* feed(uint64_t jobid)
{
xce_decision_t* decision = NULL;
int waited_ms;
TEST_ASSERT_TRUE(cbuf_tryput(&(g_sched->core.jbuf), &jobid) >= 0);
for (waited_ms = 0; waited_ms < 5000; waited_ms += 5) {
struct timespec ts = { 0, 5 * 1000 * 1000 };
if (cbuf_tryget(&(g_sched->core.decisions), &decision) >= 0) {
TEST_ASSERT_NOT_NULL_MESSAGE(decision, "expected a decision, got "
"the end-of-session marker");
TEST_ASSERT_NULL_MESSAGE(g_sched->core.trace_rec,
"trace_rec not cleared after the job");
return decision;
}
nanosleep(&ts, NULL);
}
TEST_FAIL_MESSAGE("the allocator pushed nothing for the job");
return NULL;
}
/* The rendered trace of the record `jobid`'s decision carries, which must
* be of decision op `op`. Frees the decision and the record. */
static char* feed_render(uint64_t jobid, int op)
{
xce_decision_t* d = feed(jobid);
xce_sched_trace_rec_t* rec = d->trace;
char* buf = NULL;
size_t len = 0;
FILE* f;
TEST_ASSERT_NOT_NULL_MESSAGE(rec, "the decision carries no trace");
TEST_ASSERT_EQUAL_UINT64(jobid, rec->jobid);
f = open_memstream(&buf, &len);
TEST_ASSERT_NOT_NULL(f);
repl_trace_render(rec, XCE_REPL_UNITS_RAW, f, NULL);
fclose(f);
if (getenv("XCE_TRACE_ERR_DUMP") != NULL) {
fprintf(stderr, "---- job %llu ----\n%s", (unsigned long long)jobid,
buf);
}
TEST_ASSERT_EQUAL_INT_MESSAGE(op, d->op, "unexpected decision op");
sched_trace_rec_free(rec);
free(d);
return buf;
}
/* `text` holds `line` as one whole line */
static void assert_line(const char* text, const char* line)
{
size_t n = strlen(line);
const char* p = text;
while ((p = strstr(p, line)) != NULL) {
if (((p == text) || (p[-1] == '\n')) && (p[n] == '\n')) {
return;
}
p++;
}
fprintf(stderr, "---- missing line ----\n%s\n---- in ----\n%s----\n", line,
text);
TEST_FAIL_MESSAGE("expected trace line missing");
}
/* `text` holds no line containing `frag` */
static void assert_no_frag(const char* text, const char* frag)
{
if (strstr(text, frag) != NULL) {
fprintf(stderr, "---- unexpected ----\n%s\n---- in ----\n%s----\n",
frag, text);
TEST_FAIL_MESSAGE("unexpected trace text");
}
}
/* ---- scenarios ---------------------------------------------------------- */
#define ENOMEM_TXT "(err -12: Cannot allocate memory)"
#define EIO_TXT "(err -5: Input/output error)"
/* An attempt ended by an allocator internal error surfaces as alloc_rc < 0
* for a single requirement: the requirement ended in an internal error
* (3.2) and the record ends with the no-decision line (2.4). */
static void assert_internal_error_end(const char* t, const char* err)
{
char line[256];
snprintf(line, sizeof(line),
" single requirement ended in an internal error %s", err);
assert_line(t, line);
snprintf(line, sizeof(line),
" no decision: allocation attempt ended in an internal error %s",
err);
assert_line(t, line);
assert_no_frag(t, "outcome:");
}
/* 4.4: the claims buffer cannot grow for plan 1, so the plan is not tried
* and no further plans are tried. */
static void test_plan_claims_growth_failure(void)
{
char* t;
arm(&g_realloc_fail_on, &g_realloc_calls, 1);
t = feed_render(JOB_RESERVE, XCE_DECISION_TRACE);
TEST_ASSERT_EQUAL_INT_MESSAGE(0, atomic_load(&g_realloc_fail_on),
"the claims-buffer growth never ran");
assert_line(t, " cluster-wide check passed");
assert_line(t, " plan 1 hosts 1 tasks per position [1] not tried: "
"internal error " ENOMEM_TXT "; no further plans tried");
assert_line(t, " plans: 1 generated, 0 tried, 0 discarded, 1 not tried");
assert_no_frag(t, "position 1 (");
assert_internal_error_end(t, ENOMEM_TXT);
free(t);
}
/* 4.4: the accepted host's claim snapshot fails, so the position is not
* filled because of an internal error after accepting that host. */
static void test_snapshot_failure(void)
{
char* t;
arm(&g_dup_fail_on, &g_dup_calls, 1);
t = feed_render(JOB_SNAPSHOT, XCE_DECISION_TRACE);
TEST_ASSERT_EQUAL_INT_MESSAGE(0, atomic_load(&g_dup_fail_on),
"the claim snapshot never ran");
assert_line(t, " position 1 (1 task)");
assert_line(t, " host " WORKER_A_NAME " accepted");
assert_line(t, " position 1 not filled: internal error after "
"accepting " WORKER_A_NAME);
assert_line(t, " plan 1 ended in an internal error " ENOMEM_TXT
"; no further plans tried");
assert_line(t, " plans: 1 generated, 1 tried, 0 discarded, 0 not tried");
assert_no_frag(t, "committed");
assert_internal_error_end(t, ENOMEM_TXT);
free(t);
}
/* 4.5: every position filled, but building the allocation fails: the plan
* ended in an internal error, never shown as committed. */
static void assert_finalize_failed(const char* t)
{
assert_line(t, " position 1 filled by " WORKER_A_NAME);
assert_line(t, " plan 1 succeeded");
assert_line(t, " plan 1 ended in an internal error building the "
"allocation " ENOMEM_TXT
"; not committed, no further plans tried");
assert_line(t, " plans: 1 generated, 1 tried, 0 discarded, 0 not tried");
assert_no_frag(t, "plan 1 committed");
assert_internal_error_end(t, ENOMEM_TXT);
}
static void test_finalize_host_alloc_failure(void)
{
char* t;
arm(&g_halloc_fail_on, &g_halloc_calls, 1);
t = feed_render(JOB_FIN_HOST, XCE_DECISION_TRACE);
TEST_ASSERT_EQUAL_INT_MESSAGE(0, atomic_load(&g_halloc_fail_on),
"the host allocation was never built");
assert_finalize_failed(t);
free(t);
}
static void test_finalize_job_alloc_failure(void)
{
char* t;
arm(&g_jalloc_fail_on, &g_jalloc_calls, 1);
t = feed_render(JOB_FIN_JOB, XCE_DECISION_TRACE);
TEST_ASSERT_EQUAL_INT_MESSAGE(0, atomic_load(&g_jalloc_fail_on),
"the job allocation was never built");
assert_finalize_failed(t);
free(t);
}
/* 4.4 with 5.7: the scan stops on an error after one host was considered
* (and not evaluated, its uncommitted lookup failing); that host is shown,
* the position is not filled because of the scan error, and the attempt
* goes on to its search end and a pend: a scan error does not end it. */
static void test_host_scan_failure(void)
{
char* t;
arm(&g_uncomit_fail_on, &g_uncomit_calls, 1);
arm(&g_iter_fail_on, &g_iter_calls, 2);
t = feed_render(JOB_SCAN, XCE_DECISION_PEND);
TEST_ASSERT_EQUAL_INT_MESSAGE(0, atomic_load(&g_iter_fail_on),
"the host scan never failed");
assert_line(t, " host " WORKER_A_NAME " not evaluated: internal "
"error " ENOMEM_TXT);
assert_line(t, " position 1 not filled: internal error in the host "
"scan " EIO_TXT);
assert_line(t, " plan 1 failed: position 1 could not be filled");
assert_line(t, " plans: 1 generated, 1 tried, 0 discarded, 0 not "
"tried; no plan could be placed");
assert_line(t, " single requirement not satisfied");
assert_line(t, " outcome: PEND");
assert_no_frag(t, WORKER_B_NAME);
free(t);
}
/* 5.7: the first host's per-host uncommitted lookup fails: it is not
* evaluated, and the next host is accepted and committed. */
static void test_uncommitted_lookup_failure(void)
{
char* t;
arm(&g_uncomit_fail_on, &g_uncomit_calls, 1);
t = feed_render(JOB_UNCOMIT, XCE_DECISION_DISPATCH);
TEST_ASSERT_EQUAL_INT_MESSAGE(0, atomic_load(&g_uncomit_fail_on),
"the uncommitted lookup never ran");
assert_line(t, " host " WORKER_A_NAME " not evaluated: internal "
"error " ENOMEM_TXT);
assert_line(t, " host " WORKER_B_NAME " accepted");
assert_line(t, " position 1 filled by " WORKER_B_NAME);
assert_line(t, " plan 1 committed; no further plans tried");
assert_line(t, " single requirement satisfied");
assert_line(t, " outcome: DISPATCH");
assert_line(t, " host " WORKER_B_NAME ": slots=1");
free(t);
}
/* 2.4: an allocation function returning an error ends the attempt before
* any plan; the no-decision line carries its code. */
static void test_allocation_function_error(void)
{
xce_sched_core_t* core = &(g_sched->core);
xce_sched_plugin_t* real_seq = core->seqjob_allocator;
static xce_sched_plugin_t stub;
char* t;
memset(&stub, 0, sizeof(stub));
stub.name = (char*)"err_stub_seqjob";
stub.job_alloc_fn = stub_job_alloc;
core->seqjob_allocator = &stub;
t = feed_render(JOB_ALLOC_ERR, XCE_DECISION_TRACE);
core->seqjob_allocator = real_seq;
assert_line(t, " selection: policy builtin_fcfs queue " QUEUE_NAME
" (priority 5) job priority 0 queued "
"1970-01-01T00:20:34Z");
assert_line(t, " requirement single tasks 1..1 requests mem=7.000000");
assert_no_frag(t, "plan");
assert_internal_error_end(t, EIO_TXT);
free(t);
}
/* 2.4: an unknown jobid put directly on jbuf: session, position and jobid,
* ordering keys and limits unavailable. Every earlier job was a position. */
static void test_unknown_jobid_on_jbuf(void)
{
char* t = feed_render(JOB_NOT_FOUND, XCE_DECISION_TRACE);
assert_line(t, "[trace] job 0-0-0-662999 session 4 position 8 level "
"full");
assert_line(t, " selection: ordering keys and limits unavailable (job "
"not found by the allocator)");
assert_line(t, " no decision: job not found by the allocator");
assert_no_frag(t, "requirement");
assert_no_frag(t, "cluster-wide");
free(t);
}
int main(void)
{
UNITY_BEGIN();
start_sched();
RUN_TEST(test_plan_claims_growth_failure);
RUN_TEST(test_snapshot_failure);
RUN_TEST(test_finalize_host_alloc_failure);
RUN_TEST(test_finalize_job_alloc_failure);
RUN_TEST(test_host_scan_failure);
RUN_TEST(test_uncommitted_lookup_failure);
RUN_TEST(test_allocation_function_error);
RUN_TEST(test_unknown_jobid_on_jbuf);
return UNITY_END();
}

60
src/scheduler/testsuite/test_trace_store.c
View file

@ -19,9 +19,15 @@
* - `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).
* an array with no stored trace listing its elements' states (7.9);
* - task 6.2: when the element list cannot grow, the array listing
* reports the failure instead of silently dropping elements.
*
* The store is #included (not linked) so its realloc() can be renamed for
* that out-of-memory case.
*/
#define _GNU_SOURCE
#include <errno.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
@ -35,6 +41,25 @@
#include "sim.h"
#include "utils.h"
/* Out-of-memory seam (task 6.2): the store is #included rather than linked,
* so this rename reaches its element-list growth and nothing else. Armed
* with the 1-based call number that fails. */
static int g_realloc_calls;
static int g_realloc_fail_on;
static void* test_realloc(void* p, size_t n);
#define realloc test_realloc
#include "../src/repl_trace_store.c"
#undef realloc
static void* test_realloc(void* p, size_t n)
{
if ((g_realloc_fail_on > 0) && (++g_realloc_calls == g_realloc_fail_on)) {
g_realloc_fail_on = 0;
return NULL;
}
return realloc(p, n);
}
#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)
@ -521,6 +546,38 @@ static void test_array_only_discarded(void)
assert_info(fmt_id(ARRAY2_BASE), want);
}
/* Task 6.2 (4.2 OPEN): an element list that cannot grow is reported, never
* printed short: both the stored-trace listing and the element-state
* listing say the array could not be listed and print no element. */
static void test_array_listing_oom_reports_error(void)
{
xce_sched_trace_rec_t* e1 = new_rec(ELEM(1), 1, XCE_TRACE_LEVEL_FULL, 0);
xce_sched_trace_rec_t* e3 = new_rec(ELEM(3), 1, XCE_TRACE_LEVEL_FULL, 0);
char want[256];
repl_trace_store_put(&g_rs.trace, e1);
repl_trace_store_put(&g_rs.trace, e3);
/* the stored-trace listing */
snprintf(want, sizeof(want),
"[sim] info trace: array %s: cannot list its elements: %s\n",
fmt_id(ARRAY_BASE), strerror(ENOMEM));
g_realloc_calls = 0;
g_realloc_fail_on = 1;
assert_info(fmt_id(ARRAY_BASE), want);
TEST_ASSERT_EQUAL_INT_MESSAGE(0, g_realloc_fail_on,
"the element list never grew");
/* the element-state listing (no element of ARRAY2 has a trace) */
snprintf(want, sizeof(want),
"[sim] info trace: array %s: cannot list its elements: %s\n",
fmt_id(ARRAY2_BASE), strerror(ENOMEM));
g_realloc_calls = 0;
g_realloc_fail_on = 1;
assert_info(fmt_id(ARRAY2_BASE), want);
TEST_ASSERT_EQUAL_INT(0, g_realloc_fail_on);
}
/* Usage and a malformed id leave the store untouched. */
static void test_usage_and_invalid_id(void)
{
@ -561,6 +618,7 @@ int main(void)
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_array_listing_oom_reports_error);
RUN_TEST(test_usage_and_invalid_id);
RUN_TEST(test_null_and_fini);
return UNITY_END();