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

feat(scheduler-sim-decision-trace): hook the allocator loop

Browse source 
Task 3.1: when the sim publishes trace settings, sched_allocator() counts
job positions, starts a record for wanted jobs with their selection and
cluster-wide limits, pushes no-decision kinds as trace-only decisions,
and attaches the captured outcome to built decisions; every handoff
clears core->trace_rec. Live (trace_ctl NULL) is unchanged.

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

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

@ -124,7 +124,7 @@
- _Depends: 2.3_
- [ ] 3. Allocator-thread hooks at the decision seams
- [ ] 3.1 Hook the allocator loop: position, record lifecycle, selection and no-decision paths (P)
- [x] 3.1 Hook the allocator loop: position, record lifecycle, selection and no-decision paths (P)
- **Position:** when session settings are present, every real job pop increments it.
- **Record start:** a wanted job starts a record with its selection facts; its cluster-wide limits are added after election.
- **No-decision paths:**
@ -360,3 +360,4 @@
- 2.2 OPEN for 6.2: if `limit_row()` runs out of memory, `sched_trace_check_rows` returns early, so that row's LIMIT bit and later rows' SHORTFALL bits are lost. The verdict derived from rows then becomes SELECT and the Req 9.3 counts are off. This needs fault injection to test, so fix it in 6.2: keep setting the verdict bits and continue.
- 2.3: the outcome capture functions are `sched_trace_capture_dispatch`, `sched_trace_capture_pend` and `sched_trace_rec_reconstructed`, declared in trace.h; 3.x and 5.3 call them. Limit charges are classified by the limit flag or a `limit=` prefix and resolved via glimits (shared) or the worker's hlimits (host, worker looked up by host name). An unresolved charge has `limited_res == NULL`, which is the raw-number fallback. Shared pend entries carry host "" and are treated as host-less. For 4.1: limit totals over omitted hosts have a scope without the host, and a NULL host name yields the scope "h=". Follow-up (low): capture's `key_find`/`vget` on resmgr/workers run without the RTTI read lock that `sched_get_worker` takes. That is safe while the sim runs in lock-step; the 2.4 split may wrap them.
- 2.4: the trace sources are now sched_trace.c (record, arena, entries, budget, requirement/plan/check), sched_trace_scan.c (positions, host verdicts/rows, tallies, rollback, commit, search end), sched_trace_capture.c, sched_trace_outcome.c (dispatch, pend, reconstructed) and sched_trace_internal.h (12 shared `sti_` helpers). Every target that compiles one trace .c file needs all four. Later tasks that extend the trace must keep each file under 800 lines and add new files to the core isolation scan.
- 3.1: allocator-loop hooks live in `sched_allocator()` (sched_election.c), with helpers `sched_trace_push_no_decision()` and `sched_trace_attach()`. Selection is captured at job lookup, before election, so NO_SCHEDOBJ still has its selection; cluster-wide limits are captured after election. The position counter (`ctl->position`) is not SCHED_TRACE-gated (it creates the record) and counts only real pops. `sched_sim_isolation_lib` stays without the trace sources (unresolved references in a static archive are harmless); 5.1 decides whether to add them. `core->decisions` holds 5 items, so trace-only decisions take FIFO slots and cannot reorder real ones.

80
src/scheduler/src/sched_election.c
View file

@ -43,9 +43,57 @@
#include "plugin.h"
#include "preason.h"
#include "resreq.h"
#include "sched/trace.h"
#include "scheduler.h"
#include "utils.h"
/*
* Decision trace, allocator loop (scheduler-sim-decision-trace, design.md
* "TraceHooks"). Both helpers are reached only through SCHED_TRACE, i.e.
* while core->trace_rec holds the record of the job being allocated, and
* both hand that record away, so both clear core->trace_rec first: no
* later job can append to a record the drain loop owns.
*/
/* A selection that produces no scheduling decision (Requirement 2.4): the
* record is finished as `kind` and pushed alone as a trace-only decision.
* If that decision cannot be allocated, the record is freed and this trace
* is lost (design.md "Error Handling"). */
static void sched_trace_push_no_decision(xce_sched_core_t* core,
xce_sched_trace_rec_kind_t kind,
int err)
{
xce_sched_trace_rec_t* rec = core->trace_rec;
xce_decision_t* decision = NULL;
core->trace_rec = NULL;
sched_trace_rec_finish(rec, kind, err);
decision = make_trace_decision(rec);
if (decision == NULL) {
sched_trace_rec_free(rec);
return;
}
cbuf_put(&(core->decisions), &decision);
}
/* A built decision: its outcome is captured (6.1-6.4) and the record rides
* on it to the drain loop. */
static void sched_trace_attach(xce_sched_core_t* core, xce_sched_job_t* job,
xce_decision_t* decision)
{
xce_sched_trace_rec_t* rec = core->trace_rec;
core->trace_rec = NULL;
if (decision->op == XCE_DECISION_DISPATCH) {
sched_trace_capture_dispatch(rec, core, job->schedobj,
decision->u.alloc);
} else if (decision->op == XCE_DECISION_PEND) {
sched_trace_capture_pend(rec, decision->u.reason);
}
decision->trace = rec;
}
/*
* sched_job_picker - the scheduler's dedicated picker thread entry point
* (Req 1 AC 3/AC 4, Req 3 AC 5, scheduler-core spec, task 2.4).
@ -197,6 +245,7 @@ void* sched_allocator(void* arg)
cbuf_get(&(core->jbuf), &jobid);
if (jobid == 0) {
/* end of scheduling session */
SCHED_TRACE(core, core->trace_rec = NULL); /* defensive */
decision = NULL;
cbuf_put(&(core->decisions), &decision);
continue;
@ -223,12 +272,28 @@ void* sched_allocator(void* arg)
continue;
}
/* Decision trace prologue (Requirement 2.1): with session settings
* published (simulator REPL only; live leaves trace_ctl NULL, so
* this is its one trace branch, 11.2) every real job pop counts a
* position, and a wanted job starts a record. */
if (core->trace_ctl != NULL) {
core->trace_ctl->position++;
if (sched_trace_ctl_wants(core->trace_ctl, jobid)) {
core->trace_rec = sched_trace_rec_new(core->trace_ctl, jobid,
XCE_TRACE_REC_DECISION);
}
}
job = sched_get_job(core, jobid);
if (job == NULL) {
fmt_jobid(jobidString, sizeof(jobidString), jobid);
TRACE("%s job %s can not be found", __func__, jobidString);
SCHED_TRACE(core, sched_trace_push_no_decision(
core, XCE_TRACE_REC_NOT_FOUND, 0));
continue;
}
SCHED_TRACE(core,
sched_trace_capture_selection(core->trace_rec, core, job));
/* bug-1897: election and allocation both work on job->schedobj; a
* job without one (its resreq failed to parse) cannot be scheduled.
@ -237,6 +302,8 @@ void* sched_allocator(void* arg)
fmt_jobid(jobidString, sizeof(jobidString), jobid);
ERROR("%s job %s has no scheduling object, skipped", __func__,
jobidString);
SCHED_TRACE(core, sched_trace_push_no_decision(
core, XCE_TRACE_REC_NO_SCHEDOBJ, 0));
continue;
}
@ -244,6 +311,8 @@ void* sched_allocator(void* arg)
* is core-typed (mirrors plugin_begin_session(core) in this same
* codebase's driver files) -- pass core directly. */
plugin_job_election(core, job);
SCHED_TRACE(
core, sched_trace_capture_glimits(core->trace_rec, job->schedobj));
/* calculate decision for jobid */
/* scheduler-sim-live-allocation spec, task 2.2: sched_make_
@ -251,6 +320,9 @@ void* sched_allocator(void* arg)
alloc_rc = sched_make_allocation(core, job->schedobj);
if (alloc_rc < 0) {
/* something wrong */
SCHED_TRACE(core,
sched_trace_push_no_decision(
core, XCE_TRACE_REC_INTERNAL_ERROR, alloc_rc));
continue;
} else if (alloc_rc == 0) {
/* job is pend */
@ -267,6 +339,8 @@ void* sched_allocator(void* arg)
if (reason == NULL) {
fmt_jobid(jobidString, sizeof(jobidString), jobid);
WARN("job %s is pending but pend reason is NULL", jobidString);
SCHED_TRACE(core, sched_trace_push_no_decision(
core, XCE_TRACE_REC_PEND_NO_REASON, 0));
} else {
reason->jobid = job->jobid;
decision = make_pend_decision(reason);
@ -284,8 +358,14 @@ void* sched_allocator(void* arg)
}
if (decision != NULL) {
SCHED_TRACE(core, sched_trace_attach(core, job, decision));
sched_commit_decision(core, decision);
cbuf_put(&(core->decisions), &decision);
} else {
/* the decision could not be built (a pend without a reason
* already handed its record away, so the gate skips it) */
SCHED_TRACE(core, sched_trace_push_no_decision(
core, XCE_TRACE_REC_INTERNAL_ERROR, -ENOMEM));
}
}

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

@ -2627,3 +2627,30 @@ sched_trace_hooks_gate_test = executable(
test('scheduler trace hooks gate test',
sched_trace_hooks_gate_test, suite: 'scheduler',
env: ['ASAN_OPTIONS=detect_leaks=0'])
# scheduler-sim-decision-trace task 3.1: the allocator-loop trace hooks
# (design.md "TraceHooks", allocator-loop rows; Requirements 2.1-2.4, 6.1,
# 6.4, 11.2) on the real allocator thread. Real-scheduler_srcs link
# technique, so the decision-constructor wraps see the allocator's calls
# (decision-build failures).
sched_trace_hooks_alloc_loop_test_srcs = ['test_trace_hooks_alloc_loop.c']
foreach f : scheduler_srcs
if f != 'src/main.c'
sched_trace_hooks_alloc_loop_test_srcs += ['../' + f]
endif
endforeach
sched_trace_hooks_alloc_loop_test = executable(
'sched_trace_hooks_alloc_loop_test',
sched_trace_hooks_alloc_loop_test_srcs,
include_directories:[incdir],
dependencies: [scheduler_deps, unity_dep],
link_args: ['-Wl,--wrap=xcecli_start',
'-Wl,--wrap=xcecli_join',
'-Wl,--wrap=make_pend_decision',
'-Wl,--wrap=make_dispatch_decision'],
install: false)
test('scheduler trace hooks alloc loop test',
sched_trace_hooks_alloc_loop_test, suite: 'scheduler',
env: ['ASAN_OPTIONS=detect_leaks=0'])

508
src/scheduler/testsuite/test_trace_hooks_alloc_loop.c Normal file
View file

@ -0,0 +1,508 @@
/*
* test_trace_hooks_alloc_loop.c - the allocator-loop trace hooks
* (scheduler-sim-decision-trace spec, task 3.1; design.md "TraceHooks",
* allocator-loop rows; Requirements 2.1-2.4, 6.1, 6.4, 11.2).
*
* Runs the real allocator thread (sched_init(): real plugins, real picker
* and allocator threads, real ingest) with session settings published on
* the core, feeding jobids straight into core->jbuf and reading what the
* allocator pushes onto core->decisions:
*
* - position: every real job pop increments ctl->position, traced or
* not, and the record carries the position of its own pop (2.1);
* - a job the settings do not want gets no record;
* - a dispatch and a pend decision carry their record, with the
* selection facts, the cluster-wide limits and the outcome (2.2, 2.3,
* 6.1, 6.4);
* - every no-decision path pushes a trace-only decision finished as its
* kind (2.4): job not found (no selection), no scheduling object
* (selection without limits), allocation error, pend without a reason,
* and a decision-build failure for both a pend and a dispatch;
* - core->trace_rec is NULL after each job, and after the session-end
* marker.
*
* Fault seams: a stub sequential-job allocator plugin (an allocation error
* and a pend without a reason), a job whose schedobj is detached for one
* pop, a jobid nobody ingested, and link-level wraps of make_pend_decision/
* make_dispatch_decision for the decision-build failures. Every scheduler
* source is linked as its own translation unit, so the wraps see the
* allocator's calls.
*/
#include <errno.h>
#include <stdatomic.h>
#include <stdint.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 "resobj.h"
#include "sched/alloc_cache.h"
#include "sched/trace.h"
#include "scheduler.h"
#include "xce/cluster.h"
extern int scheduler_req_handler(xce_request_t*, void*);
#define WORKER_A_NAME "loop-host-a"
#define WORKER_B_NAME "loop-host-b"
#define WORKER_SLOTS 3.0f
#define QUEUE_NAME "loop_queue"
#define QUEUE_PRIO 7
#define JOB_DISPATCH ((uint64_t)631001ULL)
#define JOB_PEND ((uint64_t)631002ULL)
#define JOB_UNWANTED ((uint64_t)631003ULL)
#define JOB_NO_SCHEDOBJ ((uint64_t)631004ULL)
#define JOB_ALLOC_ERR ((uint64_t)631005ULL)
#define JOB_PEND_NOREASON ((uint64_t)631006ULL)
#define JOB_PEND_NOBUILD ((uint64_t)631007ULL)
#define JOB_DISP_NOBUILD ((uint64_t)631008ULL)
#define JOB_NOT_FOUND ((uint64_t)631999ULL) /* never ingested */
/* ── link seams ─────────────────────────────────────────────────────── */
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;
}
extern xce_decision_t* __real_make_pend_decision(xce_pendreason_t*);
extern xce_decision_t* __real_make_dispatch_decision(xce_job_alloc_t*);
/* Armed by the test thread before it feeds the job, read by the allocator
* thread; the jbuf/decisions cbufs order the two. */
static atomic_int g_fail_pend_build;
static atomic_int g_fail_dispatch_build;
xce_decision_t* __wrap_make_pend_decision(xce_pendreason_t* reason)
{
if (atomic_exchange(&g_fail_pend_build, 0) != 0) {
return NULL;
}
return __real_make_pend_decision(reason);
}
xce_decision_t* __wrap_make_dispatch_decision(xce_job_alloc_t* alloc)
{
if (atomic_exchange(&g_fail_dispatch_build, 0) != 0) {
return NULL;
}
return __real_make_dispatch_decision(alloc);
}
/* Stub sequential-job allocator: returns g_stub_rc and sets no pend
* reason, so 0 is a pend without a reason and < 0 an allocation error. */
static int g_stub_rc;
static int stub_job_alloc(void* plugin, xce_sched_ctx_t* ctx)
{
(void)plugin;
(void)ctx;
return g_stub_rc;
}
void setUp(void) { return; }
void tearDown(void) { return; }
/* ── fixture ────────────────────────────────────────────────────────── */
static xce_scheduler_t* g_sched;
static xce_sched_trace_ctl_t g_ctl;
static xce_sched_trace_sel_t g_sel[8];
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,
uint32_t maxtasks)
{
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*)"loop_user";
j->submit->project_name = (char*)"loop_proj";
j->submit->resreq = (char*)resreq;
j->submit->mintasks = 1;
j->submit->maxtasks = maxtasks;
return j;
}
/* Real sched_init() with two 3-slot workers, one queue and every test job
* ingested, then a session begun. Never torn down (the real threads keep
* running), the convention of the other real-scheduler_srcs 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[2];
xce_job_t* jobs[7];
xce_queue_t* q = (xce_queue_t*)calloc(1, sizeof(*q));
xce_resource_t res;
int rc;
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] = 19996;
config.from = "sched_trace_hooks_alloc_loop_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 = 19996;
strncpy(cluster_cfg.identity, "sched_trace_hooks_alloc_loop_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);
/* distinct requirements, so no job hits another's unsatisfiable-cache
* entry */
jobs[0] = make_push_job(JOB_DISPATCH, "rusage[mem=1]", 1);
jobs[1] = make_push_job(JOB_PEND, "7*{rusage[mem=2]}", 1);
jobs[2] = make_push_job(JOB_UNWANTED, "rusage[mem=3]", 1);
jobs[3] = make_push_job(JOB_NO_SCHEDOBJ, "rusage[mem=4]", 1);
jobs[4] = make_push_job(JOB_ALLOC_ERR, "rusage[mem=5]", 1);
jobs[5] = make_push_job(JOB_PEND_NOREASON, "rusage[mem=6]", 1);
jobs[6] = make_push_job(JOB_PEND_NOBUILD, "7*{rusage[mem=7]}", 1);
memset(&ev, 0, sizeof(ev));
ev.type = XCE_CORE_EVENT_JOBS_PUSH;
ev.u.jobs_push.njobs = 7;
ev.u.jobs_push.jobs = jobs;
TEST_ASSERT_TRUE(xce_core_push_events(&(g_sched->core), &ev, 1) >= 0);
{
xce_job_t* disp = make_push_job(JOB_DISP_NOBUILD, "rusage[mem=8]", 1);
memset(&ev, 0, sizeof(ev));
ev.type = XCE_CORE_EVENT_JOBS_PUSH;
ev.u.jobs_push.njobs = 1;
ev.u.jobs_push.jobs = &disp;
TEST_ASSERT_TRUE(xce_core_push_events(&(g_sched->core), &ev, 1) >= 0);
}
TEST_ASSERT_TRUE(xce_core_begin_session(&(g_sched->core)) >= 0);
}
/* Publishes settings selecting `n` plain jobids at full level, as the REPL's
* session-begin hook does (position reset to 0). */
static void publish_ctl(const uint64_t* ids, int n)
{
int i;
memset(&g_ctl, 0, sizeof(g_ctl));
memset(g_sel, 0, sizeof(g_sel));
for (i = 0; i < n; i++) {
g_sel[i].key = ids[i];
g_sel[i].whole_array = false;
}
g_ctl.nsel = n;
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 = 3;
g_ctl.policy_name = "builtin_fcfs";
g_ctl.position = 0;
g_sched->core.trace_ctl = &g_ctl;
}
/* Feeds jobid to the allocator thread and waits (bounded poll) for the next
* entry it pushes onto core->decisions. *got is false on timeout. */
static xce_decision_t* feed(uint64_t jobid, bool* got)
{
xce_decision_t* decision = NULL;
int waited_ms;
*got = false;
TEST_ASSERT_TRUE(cbuf_tryput(&(g_sched->core.jbuf), &jobid) >= 0);
for (waited_ms = 0; waited_ms < 2000; waited_ms += 5) {
struct timespec ts = { 0, 5 * 1000 * 1000 };
if (cbuf_tryget(&(g_sched->core.decisions), &decision) >= 0) {
*got = true;
return decision;
}
nanosleep(&ts, NULL);
}
return NULL;
}
static xce_decision_t* feed_expect(uint64_t jobid)
{
bool got;
xce_decision_t* d = feed(jobid, &got);
TEST_ASSERT_TRUE_MESSAGE(got, "the allocator pushed nothing for the job");
TEST_ASSERT_NOT_NULL_MESSAGE(d, "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 d;
}
/* A trace-only decision for jobid whose record was finished as `kind`. */
static xce_sched_trace_rec_t* expect_trace_only(uint64_t jobid,
xce_sched_trace_rec_kind_t kind,
int err, int position)
{
xce_decision_t* d = feed_expect(jobid);
xce_sched_trace_rec_t* rec = d->trace;
TEST_ASSERT_EQUAL_INT_MESSAGE(XCE_DECISION_TRACE, d->op,
"a no-decision path must push a trace-only "
"decision");
TEST_ASSERT_NOT_NULL(rec);
TEST_ASSERT_EQUAL_UINT64(jobid, rec->jobid);
TEST_ASSERT_EQUAL_INT(kind, rec->kind);
TEST_ASSERT_EQUAL_INT(err, rec->err);
TEST_ASSERT_EQUAL_INT(position, rec->position);
TEST_ASSERT_EQUAL_UINT64(3, rec->session_no);
TEST_ASSERT_EQUAL_INT(SCHED_TRACE_OUTCOME_NONE, rec->outcome);
free(d);
return rec;
}
static void assert_selection(const xce_sched_trace_rec_t* rec)
{
TEST_ASSERT_TRUE_MESSAGE(rec->selection_set, "selection not captured");
TEST_ASSERT_NOT_NULL(rec->policy_name);
TEST_ASSERT_EQUAL_STRING("builtin_fcfs", rec->policy_name);
TEST_ASSERT_NOT_NULL(rec->queue_name);
TEST_ASSERT_EQUAL_STRING(QUEUE_NAME, rec->queue_name);
TEST_ASSERT_EQUAL_INT(QUEUE_PRIO, rec->queue_priority);
TEST_ASSERT_EQUAL_INT64(1234, (int64_t)rec->queue_time);
}
static void stub_seqjob(xce_sched_plugin_t* stub, int rc)
{
memset(stub, 0, sizeof(*stub));
stub->name = (char*)"loop_stub_seqjob";
stub->job_alloc_fn = stub_job_alloc;
g_stub_rc = rc;
}
/* ── tests ──────────────────────────────────────────────────────────── */
static void test_position_and_outcomes(void)
{
static const uint64_t wanted[] = { JOB_DISPATCH, JOB_PEND };
xce_decision_t* d;
xce_sched_trace_rec_t* rec;
xce_sched_job_t* job;
publish_ctl(wanted, 2);
/* an unwanted job: counted, untraced */
d = feed_expect(JOB_UNWANTED);
TEST_ASSERT_EQUAL_INT(XCE_DECISION_DISPATCH, d->op);
TEST_ASSERT_NULL_MESSAGE(d->trace, "an unwanted job carries a trace");
TEST_ASSERT_EQUAL_INT_MESSAGE(1, g_ctl.position,
"every real job pop counts a position");
free(d);
/* dispatch: selection, limits and the dispatch outcome */
d = feed_expect(JOB_DISPATCH);
TEST_ASSERT_EQUAL_INT(XCE_DECISION_DISPATCH, d->op);
rec = d->trace;
TEST_ASSERT_NOT_NULL_MESSAGE(rec, "a traced dispatch carries no record");
TEST_ASSERT_EQUAL_INT(2, g_ctl.position);
TEST_ASSERT_EQUAL_INT(2, rec->position);
TEST_ASSERT_EQUAL_UINT64(JOB_DISPATCH, rec->jobid);
TEST_ASSERT_EQUAL_INT(XCE_TRACE_REC_DECISION, rec->kind);
assert_selection(rec);
TEST_ASSERT_TRUE_MESSAGE(rec->glimits_set, "limits not captured");
TEST_ASSERT_EQUAL_INT(SCHED_TRACE_OUTCOME_DISPATCH, rec->outcome);
TEST_ASSERT_EQUAL_INT(1, rec->dispatch.nhosts);
job = sched_get_job(&(g_sched->core), JOB_DISPATCH);
TEST_ASSERT_NOT_NULL(job);
TEST_ASSERT_TRUE(job->allocs == d->u.alloc);
sched_trace_rec_free(rec);
free(d);
/* pend: the pend-reason set as the outcome */
d = feed_expect(JOB_PEND);
TEST_ASSERT_EQUAL_INT(XCE_DECISION_PEND, d->op);
rec = d->trace;
TEST_ASSERT_NOT_NULL_MESSAGE(rec, "a traced pend carries no record");
TEST_ASSERT_EQUAL_INT(3, rec->position);
TEST_ASSERT_EQUAL_INT(XCE_TRACE_REC_DECISION, rec->kind);
assert_selection(rec);
TEST_ASSERT_TRUE(rec->glimits_set);
TEST_ASSERT_EQUAL_INT(SCHED_TRACE_OUTCOME_PEND, rec->outcome);
TEST_ASSERT_NOT_NULL(rec->pend.reasons);
TEST_ASSERT_TRUE(
job_preason_eq(d->u.reason->reasons, rec->pend.reasons, NULL));
sched_trace_rec_free(rec);
free(d);
TEST_ASSERT_EQUAL_INT(3, g_ctl.position);
}
static void test_no_decision_kinds(void)
{
static const uint64_t wanted[]
= { JOB_NOT_FOUND, JOB_NO_SCHEDOBJ, JOB_ALLOC_ERR,
JOB_PEND_NOREASON, JOB_PEND_NOBUILD, JOB_DISP_NOBUILD };
xce_sched_core_t* core = &(g_sched->core);
xce_sched_trace_rec_t* rec;
xce_sched_job_t* job;
xce_sched_obj_t* obj;
xce_sched_plugin_t* real_seq = core->seqjob_allocator;
static xce_sched_plugin_t stub;
publish_ctl(wanted, 6);
/* not found: header only, no selection, no limits */
rec = expect_trace_only(JOB_NOT_FOUND, XCE_TRACE_REC_NOT_FOUND, 0, 1);
TEST_ASSERT_FALSE(rec->selection_set);
TEST_ASSERT_FALSE(rec->glimits_set);
sched_trace_rec_free(rec);
/* no scheduling object: selection without limits */
job = sched_get_job(core, JOB_NO_SCHEDOBJ);
TEST_ASSERT_NOT_NULL(job);
obj = job->schedobj;
job->schedobj = NULL;
rec = expect_trace_only(JOB_NO_SCHEDOBJ, XCE_TRACE_REC_NO_SCHEDOBJ, 0, 2);
job->schedobj = obj;
assert_selection(rec);
TEST_ASSERT_FALSE(rec->glimits_set);
sched_trace_rec_free(rec);
/* allocation error: the allocator's error code */
stub_seqjob(&stub, -EIO);
core->seqjob_allocator = &stub;
rec = expect_trace_only(JOB_ALLOC_ERR, XCE_TRACE_REC_INTERNAL_ERROR, -EIO,
3);
assert_selection(rec);
TEST_ASSERT_TRUE(rec->glimits_set);
sched_trace_rec_free(rec);
/* pend without a reason */
stub_seqjob(&stub, 0);
rec = expect_trace_only(JOB_PEND_NOREASON, XCE_TRACE_REC_PEND_NO_REASON, 0,
4);
core->seqjob_allocator = real_seq;
assert_selection(rec);
sched_trace_rec_free(rec);
/* decision-build failure: pend, then dispatch */
atomic_store(&g_fail_pend_build, 1);
rec = expect_trace_only(JOB_PEND_NOBUILD, XCE_TRACE_REC_INTERNAL_ERROR,
-ENOMEM, 5);
TEST_ASSERT_EQUAL_INT(0, atomic_load(&g_fail_pend_build));
sched_trace_rec_free(rec);
atomic_store(&g_fail_dispatch_build, 1);
rec = expect_trace_only(JOB_DISP_NOBUILD, XCE_TRACE_REC_INTERNAL_ERROR,
-ENOMEM, 6);
TEST_ASSERT_EQUAL_INT(0, atomic_load(&g_fail_dispatch_build));
sched_trace_rec_free(rec);
TEST_ASSERT_EQUAL_INT(6, g_ctl.position);
}
static void test_session_end_marker_clears_record_pointer(void)
{
xce_sched_core_t* core = &(g_sched->core);
bool got;
xce_decision_t* d;
char stale;
/* a record pointer left behind (defensive): the marker clears it, and
* the marker is not a job pop */
publish_ctl(NULL, 0);
core->trace_rec = (xce_sched_trace_rec_t*)&stale;
d = feed(0, &got);
TEST_ASSERT_TRUE(got);
TEST_ASSERT_NULL_MESSAGE(d, "the session-end marker must stay NULL");
TEST_ASSERT_NULL_MESSAGE(core->trace_rec,
"the session-end marker must clear trace_rec");
TEST_ASSERT_EQUAL_INT(0, g_ctl.position);
core->trace_ctl = NULL;
}
int main(void)
{
UNITY_BEGIN();
start_sched();
RUN_TEST(test_position_and_outcomes);
RUN_TEST(test_no_decision_kinds);
RUN_TEST(test_session_end_marker_clears_record_pointer);
return UNITY_END();
}