feat(scheduler-core): implement session-start signal + decision callback (task 4.4)

Adds xce_core_go()/xce_core_end_session() (new core_session.c):
xce_core_go() is non-blocking, performs begin-session state reset
(select-eval-ctx, resval resets), signals the picker thread via wbuf,
and returns -EINVAL on a reentrant call while a session is already
active (proven a true no-op, not just an early return). Draining
core->decisions and invoking the driver's on_decision callback is the
driver's own responsibility between core_go() and core_end_session()
(confirmed driver-owned, not core-owned, per the Driver <-> Core
protocol). xce_core_set_decision_cb() registers the callback + driver
context as a separate setter, keeping xce_sched_core_init()'s live
sched_init() call site untouched. Adds XCE_DECISION_PREEMPT to
decision.h (jobid/victims[]/reason) alongside the existing DISPATCH/
PEND types. Additive only — sched_go/sched_begin_session/
sched_end_session/sched_allocator/main.c remain untouched.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_019dw5WnN3DXStaEaPYfACU1
This commit is contained in:
Rongsong Shen 2026-08-04 03:08:18 +00:00
parent e6c0a2448a
commit de69b6288d
Signed by: rshen
SSH key fingerprint: SHA256:8Vf2pZ3h6rJa7EhGK+WaAQ9Mta/apZvteW6g9U/WKKk
9 changed files with 900 additions and 26 deletions

View file

@ -31,6 +31,7 @@
#include <stdint.h>
#include "dataq.h"
#include "decision.h"
#include "mylock.h"
#include "plugin.h"
#include "sched/types.h"
@ -45,6 +46,16 @@
* matter). */
typedef struct xce_sched_core_s xce_sched_core_t;
/*
* xce_core_decision_cb_t - the driver-registered decision callback (task
* 4.4, Req 5 AC 6): `int (*on_decision)(void* driver_ctx, xce_decision_t*
* decision)`. Invoked once per decision, in allocator-output order, as the
* driver drains `core->decisions` between `xce_core_go()` and
* `xce_core_end_session()` -- see xce_core_go()/xce_core_end_session()'s
* own doc comments (core_session.c) for the full protocol.
*/
typedef int (*xce_core_decision_cb_t)(void*, xce_decision_t*);
struct xce_sched_core_s {
mylock_t lock;
int stop;
@ -55,30 +66,37 @@ struct xce_sched_core_s {
* name. NULL until sched_set_default_job_pick_plugin() (schedobj.c)
* applies the "builtin_fcfs" default, or a future config-driven
* assignment sets it explicitly. */
char* job_pick_plugin;
var plugins;
void* seqjob_allocator;
void* parjob_allocator;
void* affinity_allocator;
void* gpu_allocator;
dataq_t* pend_updates[2];
var resmgr;
var ugmgr;
var qgmgr;
var pgmgr;
var hgmgr;
zarray_t sharedInst;
zarray_t sharedInUse;
var jobs;
var queues;
var workers;
var users;
var projects;
pthread_t jobpicker;
pthread_t allocator;
cbuf_t wbuf; /* action to trigger scheduling */
cbuf_t jbuf; /* job picked for allocation */
cbuf_t decisions; /* scheduler decisions */
char* job_pick_plugin;
var plugins;
void* seqjob_allocator;
void* parjob_allocator;
void* affinity_allocator;
void* gpu_allocator;
dataq_t* pend_updates[2];
var resmgr;
var ugmgr;
var qgmgr;
var pgmgr;
var hgmgr;
zarray_t sharedInst;
zarray_t sharedInUse;
var jobs;
var queues;
var workers;
var users;
var projects;
pthread_t jobpicker;
pthread_t allocator;
cbuf_t wbuf; /* action to trigger scheduling */
cbuf_t jbuf; /* job picked for allocation */
cbuf_t decisions; /* scheduler decisions */
/* Req 5 AC 6 (scheduler-core spec, task 4.4): driver-registered
* decision callback + opaque driver context, set via
* xce_core_set_decision_cb() (core_session.c). NULL until a driver
* registers one -- nothing in the still-live sched_go()/main.c path
* reads these fields today. */
void* driver_ctx;
xce_core_decision_cb_t on_decision;
_Atomic int session_active; /* 1 during scheduling session, 0 between */
_Atomic int eval_in_progress; /* 1 while eval_select loop runs */
_Atomic int staging_merge_complete; /* 1 after staging merge completes */
@ -233,4 +251,55 @@ typedef struct {
*/
extern int xce_core_push_events(xce_sched_core_t*, xce_core_event_t*, int);
/*
* xce_core_set_decision_cb - registers the driver's decision callback +
* opaque driver context (task 4.4, Req 5 AC 6). May be called at any time
* (typically once, at driver startup, before the first xce_core_go()) --
* no session-active restriction, since it only stores two pointers and
* neither is read outside the decision-drain window between
* xce_core_go() and xce_core_end_session(). Returns -EINVAL if core is
* NULL.
*
* Deliberately a separate setter rather than an xce_sched_core_init()
* parameter (design.md's eventual `xce_sched_core_init(core, decision_cb,
* driver_ctx, prng_seed)` signature, not yet adopted): the sole existing
* xce_sched_core_init() call site (scheduler.c's sched_init(), still the
* live production path) has no real driver/callback to pass yet, and this
* task must not perturb that call site's behavior. A setter keeps
* xce_sched_core_init() untouched while still letting a future driver (or
* this task's own test) register a callback on an already-initialized
* core.
*/
extern int xce_core_set_decision_cb(xce_sched_core_t*, xce_core_decision_cb_t,
void*);
/*
* xce_core_go - Req 5 AC 5 (scheduler-core spec, task 4.4). Starts a
* scheduling session: NON-BLOCKING -- performs the begin-session state
* reset (select-eval-ctx, shared resval, per-worker resval/host-ctx) then
* signals the picker thread via a single `cbuf_put(&core->wbuf, ...)` and
* returns immediately. Does NOT wait for the picker/allocator to produce
* any decisions -- draining `core->decisions` and invoking the registered
* `on_decision` callback is the driver's own responsibility, in the
* driver's own loop between this call and xce_core_end_session()
* (design.md's Normal-Mode Scheduling Round: "the drain loop in the
* driver thread interleaves cbuf_get/on_decision one at a time").
*
* Returns 0 on success; -EINVAL if core is NULL OR a session is already
* active (core->session_active != 0) -- the AC's required "distinct
* error" for the reentrancy-rejection case.
*/
extern int xce_core_go(xce_sched_core_t*);
/*
* xce_core_end_session - Req 5 AC 5 (scheduler-core spec, task 4.4). Called
* by the driver after it has drained core->decisions down to the
* end-of-session NULL marker (the marker itself terminates the driver's
* drain loop -- it is never passed to on_decision; see core_session.c).
* Performs the end-session cleanup (select-eval-ctx teardown, select-node
* GC mark/sweep) then clears core->session_active to 0, unblocking the
* next xce_core_go() call. Returns -EINVAL if core is NULL.
*/
extern int xce_core_end_session(xce_sched_core_t*);
#endif

View file

@ -9,18 +9,47 @@ typedef enum {
XCE_DECISION_NONE = 0,
XCE_DECISION_DISPATCH,
XCE_DECISION_PEND,
XCE_DECISION_NEW_SESSION
XCE_DECISION_NEW_SESSION,
XCE_DECISION_PREEMPT
} xce_decision_op_t;
/*
* xce_preempt_victim_t / xce_preempt_t - PREEMPT decision payload
* (scheduler-core spec, task 4.4, Req 5 AC 6). Scheduler-owned (not
* xce/types.h): design.md's Boundary Commitments note that the PREEMPT
* wire shape is owned by this spec while victim-*selection* semantics
* belong to the (separate, not-yet-implemented) scheduler-preemption spec
* -- so these two structs carry exactly AC 6's field list ("jobid,
* victims[] (jobid, workerid per victim), reason") and nothing more.
*
* `victims` follows the same "int count; T** items" idiom as
* xce_job_dispatch_t.allocs / xce_preason_set_t.reasons (xce/types.h) --
* an array of pointers, not a flat array, for consistency with every other
* decision-adjacent list in this codebase.
*/
typedef struct {
uint64_t jobid;
char* workerid;
} xce_preempt_victim_t;
typedef struct {
uint64_t jobid;
int nvictims;
xce_preempt_victim_t** victims;
xce_pendreason_t* reason;
} xce_preempt_t;
typedef struct {
xce_decision_op_t op;
union {
xce_job_alloc_t* alloc;
xce_pendreason_t* reason;
xce_preempt_t* preempt;
} u;
} xce_decision_t;
extern xce_decision_t* make_pend_decision(xce_pendreason_t*);
extern xce_decision_t* make_dispatch_decision(xce_job_alloc_t*);
extern xce_decision_t* make_preempt_decision(xce_preempt_t*);
#endif

View file

@ -4,6 +4,7 @@ scheduler_srcs = [
'src/scheduler.c',
'src/core.c',
'src/core_events.c',
'src/core_session.c',
'src/decision.c',
'src/schedobj.c',
'src/plugin.c',

View file

@ -0,0 +1,388 @@
#include <errno.h>
#include <stdatomic.h>
#include <stdlib.h>
#include "core.h"
#include "misc.h"
#include "resreq.h"
#include "utils.h"
/*
* core_session.c - the session-start signal and decision callback protocol
* (scheduler-core spec, task 4.4, Req 5 AC 5 / AC 6). New file (not
* appended to core.c/core_events.c) purely to respect this project's
* 800-line-per-file cap -- core_events.c is already 592 lines (see that
* file's own header comment for the same rationale splitting it out of
* core.c).
*
* Same boundary as core.c/core.h/core_events.c (design.md "Allowed
* Dependencies", Req 5 AC 3): zero controller I/O, network, filesystem, or
* wall-clock dependency -- this file must never include
* client.h/netio.h/xceapi.h/delay.h/idgen.h/xce/cluster.h/scheduler.h, and
* must never call a wall-clock function directly.
*
* Public surface (core.h): xce_core_set_decision_cb(), xce_core_go(),
* xce_core_end_session() -- see core.h's own doc comments for the full
* contract. This file is purely ADDITIVE: it does not modify, and is not
* called by, the still-live sched_go()/sched_begin_session()/
* sched_end_session()/main.c loop (scheduler.c) -- those remain the
* production path until a later task (5.x) wires a driver against this
* new protocol.
*
* Decision-drain ownership (the one genuinely ambiguous design point in
* this task -- see this file's own commit/PR notes for the full
* reasoning): design.md's Normal-Mode Scheduling Round sequence diagram
* and two of its own narrative passages ("the drain loop in the driver
* thread interleaves cbuf_get/on_decision one at a time"; "The core_go()
* call itself is non-blocking ... After all decisions are drained and the
* NULL marker arrives, core_end_session() runs cleanup") are unambiguous
* and mutually consistent: the actual `cbuf_get(&core->decisions, ...)` /
* `on_decision(...)` loop is DRIVER-owned code that runs BETWEEN
* xce_core_go() and xce_core_end_session() -- it is not bundled inside
* either call, and design.md's own SchedulerCore Service Interface lists
* exactly these two functions (plus core_push_events/core_init), no third
* "drain" function. xce_core_go() therefore does only the signal-and-
* return described below; xce_core_end_session() does only cleanup. This
* file does not add a core-owned drain function -- wiring the drain loop
* into a real driver's main loop is task 5.2's job (per this task's own
* scope note); this file's test proves the mechanism using the same
* public cbuf_get()/on_decision() primitives a real driver would use.
*/
/* core_reset_resval_ - zarray_iter_fn_t adapter for reset_resval() (misc.h),
* whose signature already takes only the item, not a context arg. Shared
* by core_reset_shared_resval() and core_reset_workers() below. */
static int core_reset_resval_(void* arg, void* val)
{
(void)arg;
return reset_resval((xce_val_t*)val);
}
/* core_reset_shared_resval - mirrors scheduler.c's sched_reset_shared_resval
* (static, xce_scheduler_t*-typed, only ever touches sched->core.sharedInUse)
* parameterized on xce_sched_core_t* directly -- same "narrow, don't touch
* scheduler.c" rationale as core_events.c's core_build_job/core_build_queue/
* core_build_worker (see that file's header comment). */
static void core_reset_shared_resval(xce_sched_core_t* core)
{
zarray_iter(&(core->sharedInUse), core_reset_resval_, NULL);
}
/* sched_worker_release_host_ctx - declared in scheduler.h (xce_sched_worker_t*
* only; its body never touches xce_scheduler_t's apictx/cluster_cfg fields),
* implemented in resource.c. Redeclared here rather than #include
* "scheduler.h" to keep this file's dependency surface identical to
* core.c/core_events.c's own boundary (no controller/client pull-in) -- same
* "identical repeated extern declaration is valid C11/gnu17" idiom core.h's
* own header comment documents for xce_sched_core_t's typedef. */
extern void sched_worker_release_host_ctx(xce_sched_worker_t*);
/* core_reset_workers - mirrors scheduler.c's sched_reset_workers/
* sched_reset_worker_ (release each available worker's stale per-session
* host_eval_ctx snapshot + reset its in-use resource values), parameterized
* on xce_sched_core_t* directly. */
static void core_reset_workers(xce_sched_core_t* core)
{
iterator_t it;
int rc;
rc = iter_make(&it, core->workers);
if (rc < 0) {
return;
}
for (;;) {
xce_sched_worker_t* w = NULL;
rc = iter_next(&it, (void**)&w);
if ((rc < 0) || (rc == RTTI_ITER_NO_DATA)) {
break;
}
if ((w == NULL) || IS_WORKER_UNAVAIL(w->status)) {
continue;
}
sched_worker_release_host_ctx(w);
zarray_iter(&(w->resInUse), core_reset_resval_, NULL);
}
iter_destroy(&it);
}
/*
* core_begin_session_internal - the pseudocode's `core_begin_session_
* internal(core); // select eval ctx, reset workers, account jobs`
* (requirements.md's Req 1 AC 4 / Req 5 AC 5 pseudocode). Reproduces
* sched_begin_session()'s core-scoped state reset: a fresh select-eval
* memoisation context sized to the current interned-node population, then
* shared/per-worker resource-value reset.
*
* Deliberately excludes two pieces of the real sched_begin_session():
* - plugin_begin_session(sched): fires the OLD per-plugin
* begin_session_fn hook category (plugin.c), unrelated to the NEW
* picker interface this spec is about -- same exclusion rationale
* core_events.c's own header comment documents for JOB_UPDATE's
* plugin_job_update() fan-out ("the OLD per-plugin hook category,
* unrelated to the picker interface this event drives"). Also
* structurally unreachable without depending on xce_scheduler_t*
* (plugin_begin_session(void* arg) casts arg to xce_scheduler_t* and
* reads sched->core.plugins -- the exact "reuse only if genuinely
* core-scoped" test task 4.3's own precedent applies here too, and
* this one fails it).
* - sched_account_jobs(sched) (per-running-job resource charging via
* sched_charge_account()): sched_charge_account() is exported with an
* xce_scheduler_t* signature (scheduler.h) and its real implementation
* (resource.c's sched_update_account/sched_update_res_accounts) is not
* safely narrowable or reproducible without deeper resource-accounting
* knowledge this task does not own -- out of scope, documented here
* rather than silently dropped.
*/
static void core_begin_session_internal(xce_sched_core_t* core)
{
atomic_store_explicit(&(core->staging_merge_complete), 0,
memory_order_release);
select_eval_ctx_fini(core->sel_eval_ctx);
core->sel_eval_ctx = select_eval_ctx_init(sel_get_node_count());
core_reset_shared_resval(core);
core_reset_workers(core);
/* Staging merge done -- core_gc_select_nodes()'s GC guard reads this
* field, same discipline sched_begin_session()/sched_end_session()
* already follow. */
atomic_store_explicit(&(core->staging_merge_complete), 1,
memory_order_release);
}
/* Initial capacity for the GC mark phase's flat live-jobs array; doubles on
* overflow (core_gc_select_nodes() below) -- named per this project's
* "no hardcoded values" convention, matches sched_end_session()'s own
* (literal, unnamed) starting capacity. */
#define GC_LIVE_JOBS_INITIAL_CAP 64
/* core_mark_resreq_select_trees - marks every select-tree root reachable
* from one job's resreq (primary/compound/alter tag dispatch), mirroring
* sched_end_session()'s inline mark-phase body verbatim, factored into its
* own function for readability. */
static void core_mark_resreq_select_trees(xce_resreq_t* rr)
{
if (rr == NULL) {
return;
}
if (rr->tag == XCE_RESREQ_PRIMARY) {
xce_primary_resreq_t* pr = rr->u.primary;
if ((pr != NULL) && (pr->select != NULL)) {
gc_mark_select_tree(pr->select->root);
}
} else if (rr->tag == XCE_RESREQ_COMPOUND) {
xce_compound_resreq_t* cp = rr->u.compound;
int j;
if (cp == NULL) {
return;
}
for (j = 0; j < cp->primaries.count; j++) {
xce_primary_resreq_t* pr = xlist_get(&(cp->primaries), j);
if ((pr != NULL) && (pr->select != NULL)) {
gc_mark_select_tree(pr->select->root);
}
}
} else if (rr->tag == XCE_RESREQ_ALTER) {
xce_alter_resreq_t* alt = rr->u.alter;
int j;
if (alt == NULL) {
return;
}
for (j = 0; j < alt->compounds.count; j++) {
xce_compound_resreq_t* cp = xlist_get(&(alt->compounds), j);
int k;
if (cp == NULL) {
continue;
}
for (k = 0; k < cp->primaries.count; k++) {
xce_primary_resreq_t* pr = xlist_get(&(cp->primaries), k);
if ((pr != NULL) && (pr->select != NULL)) {
gc_mark_select_tree(pr->select->root);
}
}
}
}
}
/*
* core_gc_select_nodes - mirrors sched_end_session()'s GC mark/sweep block
* verbatim (same guard, same flat-array mark-phase collection, same
* gc_clear_marks()/gc_mark_select_tree()/fence/gc_sweep_select_nodes()/
* recompact_node_ids() protocol), parameterized on xce_sched_core_t*
* directly. Every symbol this block touches (gc_clear_marks,
* gc_mark_select_tree, gc_sweep_select_nodes, recompact_node_ids,
* select_eval_ctx_init/fini, sel_get_node_count -- all declared in
* resreq.h) is already a free function over the process-global select-node
* intern table, not coupled to xce_scheduler_t* -- unlike
* plugin_begin_session/sched_account_jobs above, no narrowing judgment
* call is needed here, this is a direct, safe reuse of already-narrow
* primitives.
*/
static void core_gc_select_nodes(xce_sched_core_t* core)
{
iterator_t it;
xce_sched_job_t* job = NULL;
int njobs = 0;
int cap = GC_LIVE_JOBS_INITIAL_CAP;
xce_sched_job_t** live_jobs;
int i;
int rc;
/* Guard: GC requires the evaluation pipeline to be idle (same
* discipline sched_end_session() applies). */
if (atomic_load_explicit(&(core->eval_in_progress), memory_order_acquire)
|| !atomic_load_explicit(&(core->staging_merge_complete),
memory_order_acquire)) {
TRACE("%s: GC skipped -- eval_in_progress or staging merge not "
"complete",
__func__);
return;
}
live_jobs = (xce_sched_job_t**)calloc(cap, sizeof(*live_jobs));
if (live_jobs == NULL) {
/* OOM: cannot safely GC -- skip to avoid sweeping nodes still in
* use. */
atomic_store_explicit(&(core->staging_merge_complete), 1,
memory_order_release);
return;
}
rc = iter_make(&it, core->jobs);
if (rc >= 0) {
for (;;) {
rc = iter_next(&it, (void**)&job);
if ((rc < 0) || (rc == RTTI_ITER_NO_DATA)) {
break;
}
if (njobs >= cap) {
int newcap = cap * 2;
xce_sched_job_t** tmp = (xce_sched_job_t**)realloc(
live_jobs, newcap * sizeof(*live_jobs));
if (tmp == NULL) {
free(live_jobs);
live_jobs = NULL;
break;
}
live_jobs = tmp;
cap = newcap;
}
live_jobs[njobs++] = job;
}
iter_destroy(&it);
}
if (live_jobs == NULL) {
atomic_store_explicit(&(core->staging_merge_complete), 1,
memory_order_release);
return;
}
gc_clear_marks();
for (i = 0; i < njobs; i++) {
core_mark_resreq_select_trees(live_jobs[i]->resreq);
}
/* Fence: mark writes visible before sweep reads them */
atomic_thread_fence(memory_order_seq_cst);
gc_sweep_select_nodes(NULL, 0);
recompact_node_ids();
free(live_jobs);
atomic_store_explicit(&(core->staging_merge_complete), 1,
memory_order_release);
}
/*
* core_end_session_internal - the pseudocode's `core_end_session_internal
* (core); // GC, reset workers`. Reproduces sched_end_session()'s
* core-scoped cleanup: select-eval-ctx teardown, then select-node GC.
*
* Excludes plugin_end_session(sched) for the same reason
* core_begin_session_internal() excludes plugin_begin_session() above (OLD
* per-plugin hook category, xce_scheduler_t*-coupled).
*/
static void core_end_session_internal(xce_sched_core_t* core)
{
select_eval_ctx_fini(core->sel_eval_ctx);
core->sel_eval_ctx = NULL;
core_gc_select_nodes(core);
}
int xce_core_set_decision_cb(xce_sched_core_t* core, xce_core_decision_cb_t cb,
void* driver_ctx)
{
if (core == NULL) {
return -EINVAL;
}
core->on_decision = cb;
core->driver_ctx = driver_ctx;
return 0;
}
int xce_core_go(xce_sched_core_t* core)
{
int go = 1;
if (core == NULL) {
return -EINVAL;
}
/* Req 5 AC 5: reject with a distinct error (-EINVAL) if a session is
* already active -- checked BEFORE touching any state, so a rejected
* call is a true no-op (no second signal placed on wbuf, no
* begin-session reset performed). */
if (atomic_load_explicit(&(core->session_active), memory_order_acquire)
!= 0) {
return -EINVAL;
}
atomic_store_explicit(&(core->session_active), 1, memory_order_release);
core_begin_session_internal(core);
/* NON-BLOCKING: wbuf has capacity 1 and is empty between sessions (the
* picker thread only ever drains it via cbuf_get, one "go" per
* session) -- this cbuf_put() completes immediately and this function
* returns without waiting for the picker/allocator to produce
* anything. */
cbuf_put(&(core->wbuf), &go);
return 0;
}
int xce_core_end_session(xce_sched_core_t* core)
{
if (core == NULL) {
return -EINVAL;
}
core_end_session_internal(core);
atomic_store_explicit(&(core->session_active), 0, memory_order_release);
return 0;
}

View file

@ -31,3 +31,18 @@ xce_decision_t* make_dispatch_decision(xce_job_alloc_t* alloc)
return decision;
}
xce_decision_t* make_preempt_decision(xce_preempt_t* preempt)
{
xce_decision_t* decision;
decision = (xce_decision_t*)calloc(1, sizeof(*decision));
if (decision == NULL) {
return NULL;
}
decision->op = XCE_DECISION_PREEMPT;
decision->u.preempt = preempt;
return decision;
}

View file

@ -241,6 +241,24 @@ sched_core_events_test = executable(
dependencies: [scheduler_test_deps],
install: false)
# Contract test for the session-start signal and decision callback protocol
# (scheduler-core spec, task 4.4, Req 5 AC 5 / AC 6): xce_core_go() signals
# the picker thread non-blockingly and rejects reentrant calls with
# -EINVAL; decisions placed on core->decisions reach the driver-registered
# on_decision callback once each, in order, followed by the end-of-session
# NULL marker terminating the drain loop; xce_core_end_session() clears
# session_active. Same white-box #include pattern as test_core_events.c
# (pulls ../src/core.c, ../src/core_session.c, and ../src/decision.c into
# this TU) -- see the test's own header comment for why this file needs no
# picker-plugin fixture (xce_core_go()/xce_core_end_session() never call
# sched_resolve_job_picker()).
sched_core_session_test = executable(
'sched_core_session_test',
['test_core_session.c'],
include_directories:[incdir],
dependencies: [scheduler_test_deps],
install: false)
# Register as meson tests so `meson test --suite scheduler` runs them all.
# detect_leaks=0 matches the suite convention (fixtures are not torn down).
test('scheduler test', scheduler_test, suite: 'scheduler',
@ -280,3 +298,6 @@ test('scheduler core jbuf capacity pin test',
test('scheduler core events test',
sched_core_events_test, suite: 'scheduler',
env: ['ASAN_OPTIONS=detect_leaks=0'])
test('scheduler core session test',
sched_core_session_test, suite: 'scheduler',
env: ['ASAN_OPTIONS=detect_leaks=0'])

View file

@ -0,0 +1,349 @@
/*
* test_core_session.c - contract test for the session-start signal and
* decision callback protocol (scheduler-core spec, task 4.4, Req 5 AC 5 /
* AC 6).
*
* Observable under test:
* - xce_core_go() on a fresh core (session_active == 0) signals the
* picker thread NON-BLOCKINGLY -- a single item lands on core->wbuf --
* and returns 0 without waiting for anything downstream.
* - xce_core_go() while a session is already active returns -EINVAL (a
* distinct error from success) and is a true no-op: session_active is
* unchanged and no second signal is placed on wbuf.
* - Decisions placed on core->decisions (standing in for what the
* allocator thread would produce in a real session -- the picker/
* allocator pipeline itself is already covered by
* test_picker_fcfs_session.c / test_picker_core_loop_boundary.c, out
* of this task's scope) reach the driver-registered on_decision
* callback exactly once each, in allocator-output order, via the same
* cbuf_get()/on_decision() drain loop a real driver runs between
* xce_core_go() and xce_core_end_session() (design.md's Normal-Mode
* Scheduling Round: "the drain loop in the driver thread interleaves
* cbuf_get/on_decision one at a time"). The end-of-session NULL marker
* terminates that loop and is never itself passed to on_decision
* (requirements.md's Driver <-> Core Protocol table marks the NULL
* marker "(implicit)", consumed by "Picker -> Allocator -> Driver",
* never listed as an on_decision argument).
* - xce_core_end_session() clears session_active back to 0, unblocking
* the next xce_core_go() call.
*
* White-box #include of ../src/core.c, ../src/core_session.c, and
* ../src/decision.c (same convention as test_core_events.c/
* test_core_jbuf_capacity_pin.c) so this test exercises the REAL
* xce_sched_core_init()/xce_core_go()/xce_core_end_session()/
* xce_core_set_decision_cb()/make_*_decision(), not a re-implementation.
* Unlike test_core_events.c, this file does not need core_events.c or a
* working picker plugin resolution -- xce_core_go()/xce_core_end_session()
* never call sched_resolve_job_picker() (verified: grep confirms neither
* core.c nor core_session.c references it) -- so the kill_sched_ family
* and plugin_unregister stubs below exist purely to satisfy the linker for
* xce_sched_core_init()'s kvmgr_params_t.datakiller fields (never actually
* invoked -- fixtures in this file are never torn down, same convention as
* every other white-box test in this directory).
*/
#include <errno.h>
#include <stdint.h>
#include <stdlib.h>
#include <string.h>
#include <unity/unity.h>
#include "preason.h"
/* Pull in the real xce_sched_core_init()/xce_core_go()/
* xce_core_end_session()/xce_core_set_decision_cb()/make_*_decision()
* under test into this TU (white-box #include). */
#include "../src/core.c"
#include "../src/core_session.c"
#include "../src/decision.c"
void kill_sched_job(void* arg) { (void)arg; }
void kill_sched_queue(void* arg) { (void)arg; }
void kill_sched_worker(void* arg) { (void)arg; }
void kill_sched_user(void* arg) { (void)arg; }
void kill_sched_project(void* arg) { (void)arg; }
void kill_sched_resval(xce_val_t* val) { (void)val; }
void plugin_unregister(void* arg) { (void)arg; }
/* sched_worker_release_host_ctx (resource.c) is outside this file's
* white-box TU (only core.c/core_session.c/decision.c are #included) --
* stubbed the same way the kill_sched_ family is above: satisfies the
* linker for core_reset_workers()'s call in core_session.c. None of this
* file's fixtures ever add a worker to core->workers, so this is never
* actually invoked at runtime either. */
void sched_worker_release_host_ctx(xce_sched_worker_t* w) { (void)w; }
void setUp(void) { return; }
void tearDown(void) { return; }
/* ---------------------------------------------------------------------- */
/* Fixture */
/* ---------------------------------------------------------------------- */
static void fixture_init(xce_sched_core_t* core)
{
int rc = xce_sched_core_init(core);
TEST_ASSERT_EQUAL_INT(0, rc);
}
/* Test driver context + on_decision callback: records each decision
* pointer, in call order, so tests can assert both order and identity. */
#define MAX_RECORDED_DECISIONS 8
typedef struct {
int calls;
xce_decision_t* seen[MAX_RECORDED_DECISIONS];
} test_driver_ctx_t;
static int test_on_decision(void* driver_ctx, xce_decision_t* decision)
{
test_driver_ctx_t* ctx = (test_driver_ctx_t*)driver_ctx;
TEST_ASSERT_NOT_NULL(decision);
if (ctx->calls < MAX_RECORDED_DECISIONS) {
ctx->seen[ctx->calls] = decision;
}
ctx->calls++;
return 0;
}
/* drain_and_dispatch - the driver's own drain loop (design.md's Normal-
* Mode Scheduling Round), reproduced here exactly as a real driver would
* write it against the public cbuf_get()/on_decision() primitives: NOT a
* core-owned function (see core_session.c's file header comment for why
* this task deliberately does not add one). Returns the number of
* decisions delivered to on_decision before the NULL end-of-session
* marker was seen. */
static int drain_and_dispatch(xce_sched_core_t* core)
{
int delivered = 0;
for (;;) {
xce_decision_t* decision = NULL;
cbuf_get(&(core->decisions), &decision);
if (decision == NULL) {
break;
}
core->on_decision(core->driver_ctx, decision);
delivered++;
}
return delivered;
}
/* ---------------------------------------------------------------------- */
/* xce_core_go() -- session-start signal (Req 5 AC 5) */
/* ---------------------------------------------------------------------- */
static void test_core_go_signals_picker_and_returns_immediately(void)
{
xce_sched_core_t core;
int rc;
int go = 0;
fixture_init(&core);
rc = xce_core_go(&core);
TEST_ASSERT_EQUAL_INT(0, rc);
TEST_ASSERT_EQUAL_INT(
1, atomic_load_explicit(&(core.session_active), memory_order_acquire));
/* Non-blocking: the signal already landed on wbuf without anything
* having to drain it first -- a blocking cbuf_get() here would prove
* nothing (it would simply wait), so use the non-blocking tryget to
* prove the item is already present. */
rc = cbuf_tryget(&(core.wbuf), &go);
TEST_ASSERT_EQUAL_INT(0, rc);
TEST_ASSERT_EQUAL_INT(1, go);
}
static void test_core_go_rejects_reentrant_call_as_true_noop(void)
{
xce_sched_core_t core;
int rc;
int go = 0;
fixture_init(&core);
rc = xce_core_go(&core);
TEST_ASSERT_EQUAL_INT(0, rc);
/* Drain the first signal so a hypothetical buggy second cbuf_put()
* would not simply hang this test on a full capacity-1 buffer -- it
* would instead land visibly on the tryget below. */
rc = cbuf_tryget(&(core.wbuf), &go);
TEST_ASSERT_EQUAL_INT(0, rc);
rc = xce_core_go(&core);
TEST_ASSERT_EQUAL_INT(-EINVAL, rc);
TEST_ASSERT_EQUAL_INT(
1, atomic_load_explicit(&(core.session_active), memory_order_acquire));
/* No second signal was sent -- wbuf is still empty. */
rc = cbuf_tryget(&(core.wbuf), &go);
TEST_ASSERT_EQUAL_INT(-EAGAIN, rc);
}
static void test_core_go_null_core_returns_einval(void)
{
int rc = xce_core_go(NULL);
TEST_ASSERT_EQUAL_INT(-EINVAL, rc);
}
/* ---------------------------------------------------------------------- */
/* Decision callback protocol (Req 5 AC 6) */
/* ---------------------------------------------------------------------- */
static void
test_decisions_reach_callback_in_order_then_end_of_session_marker(void)
{
xce_sched_core_t core;
test_driver_ctx_t driver_ctx;
xce_job_alloc_t alloc1;
xce_pendreason_t* reason2;
xce_preempt_t preempt3;
xce_decision_t* d1;
xce_decision_t* d2;
xce_decision_t* d3;
xce_decision_t* end_marker = NULL;
int rc;
int delivered;
fixture_init(&core);
memset(&driver_ctx, 0, sizeof(driver_ctx));
rc = xce_core_set_decision_cb(&core, test_on_decision, &driver_ctx);
TEST_ASSERT_EQUAL_INT(0, rc);
rc = xce_core_go(&core);
TEST_ASSERT_EQUAL_INT(0, rc);
/* Stand in for the allocator thread's output: one decision of each
* type, in a deliberately chosen order, pushed directly onto
* core->decisions -- the same cbuf sched_allocator() (scheduler.c)
* already writes to today. */
memset(&alloc1, 0, sizeof(alloc1));
alloc1.jobid = 1;
d1 = make_dispatch_decision(&alloc1);
TEST_ASSERT_NOT_NULL(d1);
reason2 = make_pendreason(2);
TEST_ASSERT_NOT_NULL(reason2);
d2 = make_pend_decision(reason2);
TEST_ASSERT_NOT_NULL(d2);
memset(&preempt3, 0, sizeof(preempt3));
preempt3.jobid = 3;
preempt3.nvictims = 0;
preempt3.victims = NULL;
preempt3.reason = make_pendreason(3);
d3 = make_preempt_decision(&preempt3);
TEST_ASSERT_NOT_NULL(d3);
cbuf_put(&(core.decisions), &d1);
cbuf_put(&(core.decisions), &d2);
cbuf_put(&(core.decisions), &d3);
cbuf_put(&(core.decisions), &end_marker);
delivered = drain_and_dispatch(&core);
TEST_ASSERT_EQUAL_INT(3, delivered);
TEST_ASSERT_EQUAL_INT(3, driver_ctx.calls);
/* Allocator-output order preserved end to end. */
TEST_ASSERT_EQUAL_PTR(d1, driver_ctx.seen[0]);
TEST_ASSERT_EQUAL_INT(XCE_DECISION_DISPATCH, driver_ctx.seen[0]->op);
TEST_ASSERT_EQUAL_UINT64(1, driver_ctx.seen[0]->u.alloc->jobid);
TEST_ASSERT_EQUAL_PTR(d2, driver_ctx.seen[1]);
TEST_ASSERT_EQUAL_INT(XCE_DECISION_PEND, driver_ctx.seen[1]->op);
TEST_ASSERT_EQUAL_UINT64(2, driver_ctx.seen[1]->u.reason->jobid);
TEST_ASSERT_EQUAL_PTR(d3, driver_ctx.seen[2]);
TEST_ASSERT_EQUAL_INT(XCE_DECISION_PREEMPT, driver_ctx.seen[2]->op);
TEST_ASSERT_EQUAL_UINT64(3, driver_ctx.seen[2]->u.preempt->jobid);
rc = xce_core_end_session(&core);
TEST_ASSERT_EQUAL_INT(0, rc);
TEST_ASSERT_EQUAL_INT(
0, atomic_load_explicit(&(core.session_active), memory_order_acquire));
}
static void test_core_set_decision_cb_null_core_returns_einval(void)
{
int rc = xce_core_set_decision_cb(NULL, test_on_decision, NULL);
TEST_ASSERT_EQUAL_INT(-EINVAL, rc);
}
static void test_core_set_decision_cb_registers_cb_and_driver_ctx(void)
{
xce_sched_core_t core;
test_driver_ctx_t driver_ctx;
int rc;
fixture_init(&core);
memset(&driver_ctx, 0, sizeof(driver_ctx));
rc = xce_core_set_decision_cb(&core, test_on_decision, &driver_ctx);
TEST_ASSERT_EQUAL_INT(0, rc);
TEST_ASSERT_EQUAL_PTR((void*)test_on_decision, (void*)core.on_decision);
TEST_ASSERT_EQUAL_PTR(&driver_ctx, core.driver_ctx);
}
/* ---------------------------------------------------------------------- */
/* xce_core_end_session() */
/* ---------------------------------------------------------------------- */
static void test_core_end_session_null_core_returns_einval(void)
{
int rc = xce_core_end_session(NULL);
TEST_ASSERT_EQUAL_INT(-EINVAL, rc);
}
static void test_end_session_unblocks_next_core_go(void)
{
xce_sched_core_t core;
int rc;
int go = 0;
fixture_init(&core);
rc = xce_core_go(&core);
TEST_ASSERT_EQUAL_INT(0, rc);
cbuf_tryget(&(core.wbuf), &go); /* drain the first signal */
rc = xce_core_end_session(&core);
TEST_ASSERT_EQUAL_INT(0, rc);
TEST_ASSERT_EQUAL_INT(
0, atomic_load_explicit(&(core.session_active), memory_order_acquire));
/* A second xce_core_go() now succeeds -- proves end_session actually
* unblocked reentry, not just that the first rejection test worked. */
rc = xce_core_go(&core);
TEST_ASSERT_EQUAL_INT(0, rc);
TEST_ASSERT_EQUAL_INT(
1, atomic_load_explicit(&(core.session_active), memory_order_acquire));
}
int main(void)
{
UNITY_BEGIN();
RUN_TEST(test_core_go_signals_picker_and_returns_immediately);
RUN_TEST(test_core_go_rejects_reentrant_call_as_true_noop);
RUN_TEST(test_core_go_null_core_returns_einval);
RUN_TEST(test_decisions_reach_callback_in_order_then_end_of_session_marker);
RUN_TEST(test_core_set_decision_cb_null_core_returns_einval);
RUN_TEST(test_core_set_decision_cb_registers_cb_and_driver_ctx);
RUN_TEST(test_core_end_session_null_core_returns_einval);
RUN_TEST(test_end_session_unblocks_next_core_go);
return UNITY_END();
}

View file

@ -62,6 +62,7 @@ static const char* const SCANNED_RELATIVE_PATHS[] = {
"include/scheduler.h",
"include/sched/types.h",
"src/core.c",
"src/core_session.c",
"src/decision.c",
"src/handler.c",
"src/main.c",