lei.chen/myxce
1
0
Fork 1
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions
myxce/src/scheduler/testsuite/test_trace_hooks_alloc_loop.c
Rongsong Shen ed7b889bb7
feat(scheduler-sim-decision-trace): hook the allocator loop 
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>
2026-10-08 11:10:28 +00:00

508 lines
18 KiB
C
Raw Blame History

/*
* 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();
}
Reference in a new issue View git blame Copy permalink