From de69b6288d4914678f676859ee94270c5d4a652f Mon Sep 17 00:00:00 2001 From: Rongsong Shen Date: Tue, 4 Aug 2026 03:08:18 +0000 Subject: [PATCH] feat(scheduler-core): implement session-start signal + decision callback (task 4.4) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 Claude-Session: https://claude.ai/code/session_019dw5WnN3DXStaEaPYfACU1 --- .kiro/specs/scheduler-core/tasks.md | 3 +- src/scheduler/include/core.h | 117 ++++-- src/scheduler/include/decision.h | 31 +- src/scheduler/meson.build | 1 + src/scheduler/src/core_session.c | 388 ++++++++++++++++++ src/scheduler/src/decision.c | 15 + src/scheduler/testsuite/meson.build | 21 + src/scheduler/testsuite/test_core_session.c | 349 ++++++++++++++++ .../test_picker_pendreason_factory_audit.c | 1 + 9 files changed, 900 insertions(+), 26 deletions(-) create mode 100644 src/scheduler/src/core_session.c create mode 100644 src/scheduler/testsuite/test_core_session.c diff --git a/.kiro/specs/scheduler-core/tasks.md b/.kiro/specs/scheduler-core/tasks.md index 5d42e2f0..1e5a7a58 100644 --- a/.kiro/specs/scheduler-core/tasks.md +++ b/.kiro/specs/scheduler-core/tasks.md @@ -107,12 +107,13 @@ - _Depends: 4.1_ - _Verified: 2026-08-03 — xce_core_event_type_t (9 kinds) + xce_core_event_t in core.h; xce_core_push_events() in new core_events.c (split out of core.c to stay under the 800-line cap — design.md's file-structure sketch not updated yet); rejects -EBUSY mid-session, fires each hook exactly once synchronously (JOBS_PUSH once per job); additive only, not wired into the live path (task 5.1/5.2's job). Round 1 REJECTED: CONFIG_PUSH didn't decompose queues/resources/resvals like sched_init_config() does — fixed round 2 (core_apply_config_resources/resvals/queues + shared core_upsert_queue(), also used by QUEUE_UPDATE), round-2 review APPROVED with live RED→GREEN reproduction. 16/16 scheduler suite passing._ -- [ ] 4.4 Implement the session-start signal and decision callback protocol +- [x] 4.4 Implement the session-start signal and decision callback protocol - Implement the non-blocking session-start call that signals the picker thread and returns immediately, rejecting a call while a session is already active - Define the three decision output shapes (dispatch, pend, preempt) and the decision callback invoked once per decision as the allocator produces it, plus the explicit end-of-session marker - Observable: starting a session while one is already active is rejected with a distinct error; each decision reaches the registered callback in allocator-output order, followed by the end-of-session marker - _Requirements: 5.5, 5.6_ - _Depends: 4.3_ + - _Verified: 2026-08-03 — xce_core_go()/xce_core_end_session() (new core_session.c) implement the non-blocking session-start signal (-EINVAL on reentrant call, proven a true no-op via cbuf_tryget) and end-session cleanup; xce_core_set_decision_cb() registers the driver's on_decision callback (separate setter, not an xce_sched_core_init() param, to avoid touching its live sched_init() call site); XCE_DECISION_PREEMPT added to decision.h matching the DISPATCH/PEND/PREEMPT AC 6 table exactly. Decision-drain loop confirmed driver-owned (not a core function) via 3 independent requirements.md/design.md citations, re-derived independently by review. Additive only, sched_go/sched_begin_session/sched_end_session/sched_allocator/main.c untouched. REJECTED round 1 on a clang-format alignment regression in core.h (mid-struct insertion de-aligned 24 pre-existing lines) — fixed with clang-format -i, whitespace-only, re-verified 17/17 green._ - [ ] 4.5 Implement clock virtualization and session timing - Implement the virtualized clock accessor using the two-anchor-plus-speed formula, backed by atomically-updated fields; no direct wall-clock call anywhere else in the core diff --git a/src/scheduler/include/core.h b/src/scheduler/include/core.h index 96f49194..b73980d9 100644 --- a/src/scheduler/include/core.h +++ b/src/scheduler/include/core.h @@ -31,6 +31,7 @@ #include #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 diff --git a/src/scheduler/include/decision.h b/src/scheduler/include/decision.h index 74fecbb8..c08d725f 100644 --- a/src/scheduler/include/decision.h +++ b/src/scheduler/include/decision.h @@ -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 diff --git a/src/scheduler/meson.build b/src/scheduler/meson.build index 6c185889..e454edfa 100644 --- a/src/scheduler/meson.build +++ b/src/scheduler/meson.build @@ -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', diff --git a/src/scheduler/src/core_session.c b/src/scheduler/src/core_session.c new file mode 100644 index 00000000..99f6184b --- /dev/null +++ b/src/scheduler/src/core_session.c @@ -0,0 +1,388 @@ +#include +#include +#include + +#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; +} diff --git a/src/scheduler/src/decision.c b/src/scheduler/src/decision.c index 786e3af5..2ebd7cfa 100644 --- a/src/scheduler/src/decision.c +++ b/src/scheduler/src/decision.c @@ -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; +} diff --git a/src/scheduler/testsuite/meson.build b/src/scheduler/testsuite/meson.build index fcd25bb4..435f250a 100644 --- a/src/scheduler/testsuite/meson.build +++ b/src/scheduler/testsuite/meson.build @@ -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']) diff --git a/src/scheduler/testsuite/test_core_session.c b/src/scheduler/testsuite/test_core_session.c new file mode 100644 index 00000000..f6f7ec0a --- /dev/null +++ b/src/scheduler/testsuite/test_core_session.c @@ -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 +#include +#include +#include + +#include + +#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(); +} diff --git a/src/scheduler/testsuite/test_picker_pendreason_factory_audit.c b/src/scheduler/testsuite/test_picker_pendreason_factory_audit.c index a09defb5..36122fe4 100644 --- a/src/scheduler/testsuite/test_picker_pendreason_factory_audit.c +++ b/src/scheduler/testsuite/test_picker_pendreason_factory_audit.c @@ -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",