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

feat(scheduler-sim-decision-trace): core trace gate and SCHED_TRACE hook macro

Browse source 
Task 1.8: the core gains trace_ctl/trace_rec (NULL unless the sim REPL
sets them) and SCHED_TRACE evaluates its call only while a record is
being built.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
This commit is contained in:
Rongsong Shen 2026-10-08 08:53:40 +00:00
parent 29b6734649
commit a801aff46f
Signed by untrusted user: rshen
GPG key ID: 80533536B0EB651C
6 changed files with 482 additions and 2 deletions
Download patch file Download diff file Expand all files Collapse all files

24
src/scheduler/include/core.h
View file

@ -347,6 +347,30 @@ struct xce_sched_core_s {
* establishes for driver_ctx/on_decision.
*/
uint64_t prng_seed;
/*
* Decision trace gate (scheduler-sim-decision-trace spec, design.md
* "TraceModel"; Requirements 11.1-11.3). Two opaque pointers; the types
* are defined in sched/trace.h, which is not included here.
*
* trace_ctl: the session's trace settings. Only the simulator REPL
* writes it, from the driver thread inside the session window (set
* after the session begins, cleared after the drain loop); the
* allocator thread reads it once per picked job. The jbuf/wbuf
* cbuf handoffs order those accesses, so it is a plain pointer.
*
* trace_rec: the record being built for the job the allocator is
* working on. Only the allocator thread reads or writes it: set in the
* per-job prologue for a traced job, cleared when the record is handed
* away on a decision. SCHED_TRACE (sched/trace.h) runs its hook only
* while it is non-NULL.
*
* Both are NULL from xce_sched_core_init() (memset) and always NULL in
* the live scheduler and in a non-interactive simulation, which never
* publish settings. The core owns neither: xce_sched_core_fini() clears
* both without freeing them.
*/
struct xce_sched_trace_ctl_s* trace_ctl;
struct xce_sched_trace_rec_s* trace_rec;
};
/*

15
src/scheduler/include/sched/trace.h
View file

@ -87,6 +87,21 @@ typedef struct xce_sched_trace_ctl_s {
typedef struct xce_sched_trace_rec_s xce_sched_trace_rec_t;
/*
* Hook gate (Requirement 11.2). `call` (a statement or expression, side
* effects and argument evaluation included) runs only while `core` (an
* xce_sched_core_t*) has a record under construction in trace_rec, so a
* hook costs one pointer test when nothing is traced, and nothing else.
* Only the allocator thread sets trace_rec, so hooks are placed only on
* that thread. One statement: safe in an unbraced if/else.
*/
#define SCHED_TRACE(core, call) \
do { \
if ((core)->trace_rec != NULL) { \
call; \
} \
} while (0)
#define XCE_TRACE_LINE_BOUND_DEFAULT 1000 /* Requirement 9.1 */
#define XCE_TRACE_LINE_BOUND_MIN 50
#define XCE_TRACE_LINE_BOUND_MAX 100000

5
src/scheduler/src/core.c
View file

@ -412,6 +412,11 @@ void xce_sched_core_fini(xce_sched_core_t* core)
mylock_fini(&(core->lock));
/* Trace gate: the REPL owns the settings and a handed-off record
* belongs to its decision, so neither is freed here. */
core->trace_ctl = NULL;
core->trace_rec = NULL;
return;
}

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

@ -2568,3 +2568,29 @@ sched_snapshot_sparse_resvals_test = executable(
test('scheduler snapshot sparse resvals test',
sched_snapshot_sparse_resvals_test, suite: 'scheduler',
env: ['ASAN_OPTIONS=detect_leaks=0'])
# scheduler-sim-decision-trace task 1.8: the core trace gate and the
# SCHED_TRACE hook macro (design.md "TraceModel", Testing Strategy 3;
# Requirements 11.1-11.3). Real-scheduler_srcs link technique (every source
# its own translation unit), so --wrap=sched_trace_rec_new counts every
# record the allocation path creates.
sched_trace_hooks_gate_test_srcs = ['test_trace_hooks_gate.c']
foreach f : scheduler_srcs
if f != 'src/main.c'
sched_trace_hooks_gate_test_srcs += ['../' + f]
endif
endforeach
sched_trace_hooks_gate_test = executable(
'sched_trace_hooks_gate_test',
sched_trace_hooks_gate_test_srcs,
include_directories:[incdir],
dependencies: [scheduler_deps, unity_dep],
link_args: ['-Wl,--wrap=xcecli_start',
'-Wl,--wrap=xcecli_join',
'-Wl,--wrap=sched_trace_rec_new'],
install: false)
test('scheduler trace hooks gate test',
sched_trace_hooks_gate_test, suite: 'scheduler',
env: ['ASAN_OPTIONS=detect_leaks=0'])

409
src/scheduler/testsuite/test_trace_hooks_gate.c Normal file
View file

@ -0,0 +1,409 @@
/*
* test_trace_hooks_gate.c - the core trace gate and the SCHED_TRACE hook
* macro (scheduler-sim-decision-trace spec, task 1.8; design.md
* "TraceModel", "Testing Strategy" 3; Requirements 11.1, 11.2, 11.3).
*
* - xce_sched_core_init() leaves trace_ctl and trace_rec NULL, and
* xce_sched_core_fini() clears them without freeing what they point to
* (the REPL owns the settings, the decision owns a handed-off record);
* - SCHED_TRACE does not evaluate its call, side effects included, while
* no record is being built, also when settings are published;
* - SCHED_TRACE evaluates its call exactly once while a record is being
* built, and is a single statement (safe in an unbraced if/else);
* - a full allocation through the real live-mode scheduler (sched_init(),
* real picker and allocator threads, real ingest) with tracing never
* enabled creates no record: no sched_trace_rec_new() call is made,
* both core pointers stay NULL and no decision carries a trace.
*
* Record-creation proof: the binary links every scheduler source as its
* own translation unit (the real-scheduler_srcs technique), and
* -Wl,--wrap=sched_trace_rec_new routes every cross-unit call to the
* constructor through __wrap_sched_trace_rec_new() below, which counts it.
* The gate tests call the constructor through the same seam, which shows
* the counter sees real calls. The only intra-unit callers are inside
* sched_trace.c itself, which the allocation path reaches only through a
* hook that has already passed the gate.
*/
#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 "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 "gate-host-a"
#define WORKER_B_NAME "gate-host-b"
#define WORKER_SLOTS 3.0f
#define QUEUE_NAME "gate_queue"
#define JOB_ALT ((uint64_t)611001ULL)
#define JOB_PAR ((uint64_t)611002ULL)
#define JOB_PEND ((uint64_t)611003ULL)
/* ── 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_sched_trace_rec_t*
__real_sched_trace_rec_new(const xce_sched_trace_ctl_t* ctl, uint64_t jobid,
xce_sched_trace_rec_kind_t kind);
/* Written by the allocator thread once hooks exist, read by the test. */
static atomic_int g_rec_new_calls;
xce_sched_trace_rec_t*
__wrap_sched_trace_rec_new(const xce_sched_trace_ctl_t* ctl, uint64_t jobid,
xce_sched_trace_rec_kind_t kind)
{
atomic_fetch_add(&g_rec_new_calls, 1);
return __real_sched_trace_rec_new(ctl, jobid, kind);
}
void setUp(void) { return; }
void tearDown(void) { return; }
/* ── helpers ────────────────────────────────────────────────────────── */
static int g_bumps;
static int bump(int* n)
{
(*n)++;
g_bumps++;
return *n;
}
static xce_sched_trace_ctl_t make_ctl(void)
{
xce_sched_trace_ctl_t ctl;
memset(&ctl, 0, sizeof(ctl));
ctl.all = true;
ctl.level = XCE_TRACE_LEVEL_SUMMARY;
ctl.line_bound = XCE_TRACE_LINE_BOUND_DEFAULT;
ctl.session_no = 1;
ctl.policy_name = "fcfs";
return ctl;
}
/* ── core fields ────────────────────────────────────────────────────── */
static void test_core_init_leaves_trace_pointers_null(void)
{
xce_sched_core_t* core = (xce_sched_core_t*)malloc(sizeof(*core));
TEST_ASSERT_NOT_NULL(core);
memset(core, 0xA5, sizeof(*core));
TEST_ASSERT_EQUAL_INT(0, xce_sched_core_init(core));
TEST_ASSERT_NULL(core->trace_ctl);
TEST_ASSERT_NULL(core->trace_rec);
/* xce_sched_core_fini() finalizes these; both drivers calloc them */
for (int i = 0; i < 2; i++) {
core->pend_updates[i] = (dataq_t*)calloc(1, sizeof(dataq_t));
TEST_ASSERT_NOT_NULL(core->pend_updates[i]);
dataq_init(core->pend_updates[i], NULL);
}
/* fini clears both and frees neither: these point at storage the core
* does not own (a stack ctl, a stack byte), which ASAN would flag as a
* bad free. */
{
xce_sched_trace_ctl_t ctl = make_ctl();
char not_a_record;
core->trace_ctl = &ctl;
core->trace_rec = (xce_sched_trace_rec_t*)&not_a_record;
xce_sched_core_fini(core);
TEST_ASSERT_NULL(core->trace_ctl);
TEST_ASSERT_NULL(core->trace_rec);
}
free(core);
}
/* ── SCHED_TRACE gate ───────────────────────────────────────────────── */
static void test_gate_skips_call_and_side_effects_without_record(void)
{
xce_sched_core_t core;
xce_sched_core_t* cp = &core;
xce_sched_trace_ctl_t ctl = make_ctl();
int n = 0;
memset(&core, 0, sizeof(core));
g_bumps = 0;
SCHED_TRACE(cp, n++);
SCHED_TRACE(cp, bump(&n));
TEST_ASSERT_EQUAL_INT_MESSAGE(0, n, "hook argument evaluated with "
"no record being built");
TEST_ASSERT_EQUAL_INT(0, g_bumps);
/* published settings alone do not open the gate: only a record does */
core.trace_ctl = &ctl;
SCHED_TRACE(cp, n++);
SCHED_TRACE(cp, bump(&n));
TEST_ASSERT_EQUAL_INT_MESSAGE(0, n, "hook argument evaluated with "
"settings published but no record");
TEST_ASSERT_EQUAL_INT(0, g_bumps);
}
static void test_gate_evaluates_call_once_with_record(void)
{
xce_sched_core_t core;
xce_sched_core_t* cp = &core;
xce_sched_trace_ctl_t ctl = make_ctl();
xce_sched_trace_rec_t* rec;
int before = atomic_load(&g_rec_new_calls);
int n = 0;
memset(&core, 0, sizeof(core));
rec = sched_trace_rec_new(&ctl, 42, XCE_TRACE_REC_DECISION);
TEST_ASSERT_NOT_NULL(rec);
TEST_ASSERT_EQUAL_INT_MESSAGE(before + 1, atomic_load(&g_rec_new_calls),
"the counting seam must see a real "
"constructor call");
core.trace_ctl = &ctl;
core.trace_rec = rec;
SCHED_TRACE(cp, n++);
TEST_ASSERT_EQUAL_INT(1, n);
SCHED_TRACE(cp, sched_trace_path_push(core.trace_rec,
XCE_TRACE_PATH_SINGLE, 1, 1));
SCHED_TRACE(cp, sched_trace_path_pop(core.trace_rec));
/* one statement: an unbraced if/else around it still parses as such */
if (n == 1)
SCHED_TRACE(cp, n += 10);
else
n = -1;
TEST_ASSERT_EQUAL_INT(11, n);
core.trace_rec = NULL;
SCHED_TRACE(cp, n++);
TEST_ASSERT_EQUAL_INT(11, n);
sched_trace_rec_free(rec);
}
/* ── full allocation with tracing never enabled ─────────────────────── */
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 = 1000;
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*)"gate_user";
j->submit->project_name = (char*)"gate_proj";
j->submit->resreq = (char*)resreq;
j->submit->mintasks = 1;
j->submit->maxtasks = maxtasks;
return j;
}
/* Real sched_init(): registers the real plugins and starts the real picker
* and allocator threads, as the live scheduler does. Never torn down, the
* same convention as the other real-scheduler_srcs tests. */
static xce_scheduler_t* start_sched(void)
{
static xce_client_config_t config;
static xce_cluster_config_t cluster_cfg;
xce_scheduler_t* sched = (xce_scheduler_t*)calloc(1, sizeof(*sched));
xce_resource_t res;
int rc;
TEST_ASSERT_NOT_NULL(sched);
memset(&config, 0, sizeof(config));
config.nmasters = 1;
config.masters[0] = "127.0.0.1";
config.master_ports[0] = 19997;
config.from = "sched_trace_hooks_gate_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 = 19997;
strncpy(cluster_cfg.identity, "sched_trace_hooks_gate_test",
sizeof(cluster_cfg.identity) - 1);
rc = sched_init(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(sched->core.resmgr, &res));
xce_core_set_cache_clear_cb(&(sched->core), xce_alloc_cache_clear_cb);
return sched;
}
/* Feed jobid to the live allocator thread and wait (bounded poll) for its
* decision, the technique test_parjob_e2e_multislot.c uses. */
static xce_decision_t* allocate_and_wait(xce_scheduler_t* sched,
uint64_t jobid)
{
xce_decision_t* decision = NULL;
int waited_ms;
TEST_ASSERT_TRUE(cbuf_tryput(&(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(&(sched->core.decisions), &decision) >= 0) {
return decision;
}
nanosleep(&ts, NULL);
}
return NULL;
}
static void assert_untraced(xce_scheduler_t* sched, xce_decision_t* d,
int op, uint64_t jobid)
{
TEST_ASSERT_NOT_NULL(d);
TEST_ASSERT_EQUAL_INT_MESSAGE(op, d->op, "unexpected decision op");
if (op == XCE_DECISION_DISPATCH) {
TEST_ASSERT_EQUAL_UINT64(jobid, d->u.alloc->jobid);
} else {
TEST_ASSERT_EQUAL_UINT64(jobid, d->u.reason->jobid);
}
TEST_ASSERT_NULL_MESSAGE(d->trace, "untraced decision carries a trace");
TEST_ASSERT_NULL(sched->core.trace_ctl);
TEST_ASSERT_NULL(sched->core.trace_rec);
}
static void test_full_allocation_without_tracing_creates_no_record(void)
{
xce_scheduler_t* sched = start_sched();
xce_core_event_t ev;
xce_worker_t* workers[2];
xce_job_t* jobs[3];
xce_queue_t* q = (xce_queue_t*)calloc(1, sizeof(*q));
TEST_ASSERT_NULL(sched->core.trace_ctl);
TEST_ASSERT_NULL(sched->core.trace_rec);
TEST_ASSERT_NOT_NULL(q);
q->queue = strdup(QUEUE_NAME);
q->status = QUEUE_STAT_OPEN;
q->priority = 1;
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(&(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(&(sched->core), &ev, 1) >= 0);
/* On 2 x 3 slots, in allocation order: a 4-task parallel job spanning
* both hosts, an ALTER whose 5-task first alternative cannot fit in the
* 2 slots left (falls back to the 1-task second), and a 5-task compound
* that cannot fit in the last slot: the single, alternative, compound,
* plan, position and per-host paths all run. */
jobs[0] = make_push_job(JOB_ALT,
"{5*{rusage[mem=1]}} || {1*{rusage[mem=1]}}", 1);
jobs[1] = make_push_job(JOB_PAR, "rusage[mem=1]", 4);
jobs[2] = make_push_job(JOB_PEND, "5*{rusage[mem=1]}", 1);
memset(&ev, 0, sizeof(ev));
ev.type = XCE_CORE_EVENT_JOBS_PUSH;
ev.u.jobs_push.njobs = 3;
ev.u.jobs_push.jobs = jobs;
TEST_ASSERT_TRUE(xce_core_push_events(&(sched->core), &ev, 1) >= 0);
{
xce_sched_job_t* par = sched_get_job(&(sched->core), JOB_PAR);
TEST_ASSERT_NOT_NULL(par);
TEST_ASSERT_NOT_NULL(par->schedobj);
}
atomic_store(&g_rec_new_calls, 0);
TEST_ASSERT_TRUE(xce_core_begin_session(&(sched->core)) >= 0);
assert_untraced(sched, allocate_and_wait(sched, JOB_PAR),
XCE_DECISION_DISPATCH, JOB_PAR);
assert_untraced(sched, allocate_and_wait(sched, JOB_ALT),
XCE_DECISION_DISPATCH, JOB_ALT);
assert_untraced(sched, allocate_and_wait(sched, JOB_PEND),
XCE_DECISION_PEND, JOB_PEND);
TEST_ASSERT_EQUAL_INT_MESSAGE(0, atomic_load(&g_rec_new_calls),
"a trace record was created with tracing "
"never enabled");
}
int main(void)
{
UNITY_BEGIN();
RUN_TEST(test_core_init_leaves_trace_pointers_null);
RUN_TEST(test_gate_skips_call_and_side_effects_without_record);
RUN_TEST(test_gate_evaluates_call_once_with_record);
/* last: it leaves the real scheduler threads running */
RUN_TEST(test_full_allocation_without_tracing_creates_no_record);
return UNITY_END();
}