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

feat(scheduler-core): wire decision publishing and mode-selection into main.c (task 5.2)

Browse source 
Rewires main.c's live scheduling loop onto the new core-driven
architecture: xce_sched_driver_translate_pending_events() (task 5.1)
-> xce_core_begin_session() -> plugin_begin_session()+sched_account_jobs()
-> xce_core_signal_go() -> decision-drain via the registered
xce_core_decision_cb_t -> plugin_end_session() -> xce_core_end_session(),
all in the new exported xce_sched_driver_run_normal_mode_round()
(driver.c), replacing the old inline sched_update()+sched_begin_session()
+sched_go()+switch+sched_end_session() sequence.

xce_core_go() (task 4.4) is split into xce_core_begin_session()/
xce_core_signal_go() -- a pure decomposition (xce_core_go() itself is
now defined in terms of both, task 4.4's approved test suite passes
unmodified) -- specifically so the driver can interleave
plugin_begin_session()/sched_account_jobs() in the exact safety-critical
window task 4.4 deliberately excluded from pure-core scope: after the
resval reset, strictly before the picker thread is signaled. This is
empirically proven by test_driver_session_ordering.c via real
cross-thread happens-before assertions, not just documented.

xce_sched_driver_on_decision() is the new registered callback:
DISPATCH/PEND forward to the existing sched_publish_dispatch()/
sched_publish_pend_reason(); PREEMPT fails loudly (-ENOSYS, logged)
since no xcecli_preempt() client API exists and nothing in the live
allocator produces one yet (scheduler-preemption's territory).

Adds a --sim CLI flag (parse-only, loud not-yet-implemented failure,
since SimDriver is Phase 3).

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 08:05:57 +00:00
parent dceafc6b12
commit 062cf40d29
Signed by: rshen
SSH key fingerprint: SHA256:8Vf2pZ3h6rJa7EhGK+WaAQ9Mta/apZvteW6g9U/WKKk
12 changed files with 1537 additions and 53 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -335,6 +335,56 @@ extern int xce_core_push_events(xce_sched_core_t*, xce_core_event_t*, int);
extern int xce_core_set_decision_cb(xce_sched_core_t*, xce_core_decision_cb_t,
void*);
/*
* xce_core_begin_session / xce_core_signal_go - Req 5 AC 5 (task 4.4) split
* into two exported halves (task 5.2, Req 1 AC 7 / Req 5 AC 9) so a driver
* can interleave ITS OWN begin-session bookkeeping strictly between the
* state reset and the picker signal -- specifically
* plugin_begin_session()/sched_account_jobs() (the OLD per-plugin hook
* category and per-running-job resource-in-use accounting; both real,
* load-bearing production behavior, both xce_scheduler_t*-coupled and
* therefore outside this pure-core module's reach, same exclusion
* rationale core_session.c's own file header comment documents for why
* core_begin_session_internal() never called them itself).
*
* This split exists to preserve a safety property main.c's pre-task-5.2
* `sched_begin_session()` + `sched_go()` sequence already guaranteed: the
* picker thread must never observe a session whose resource-in-use
* accounting has been reset to zero (by the resval resets below) but not
* yet re-summed from currently running jobs (sched_account_jobs()) -- that
* would let the picker/allocator threads over-allocate resources already
* consumed by running jobs. A driver MUST call xce_core_signal_go() only
* AFTER any such bookkeeping following a successful xce_core_begin_session()
* completes; see driver.c's xce_sched_driver_run_normal_mode_round() for the
* real call site and testsuite/test_driver_session_ordering.c for the
* safety-property proof.
*
* xce_core_begin_session() - performs the begin-session state reset
* (select-eval-ctx, shared resval, per-worker resval/host-ctx). Does NOT
* signal the picker thread. 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. On success,
* core->session_active is already 1 and core->session_start already
* stamped -- identical to xce_core_go()'s own post-condition at the
* equivalent point.
*
* xce_core_signal_go() - NON-BLOCKING: 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"). Only valid to call after a
* successful xce_core_begin_session() for the same session -- calling it
* without one first (or twice for one session) is a caller bug this
* function does not itself guard against (xce_core_go() below is the safe
* composition for callers with no bookkeeping to interleave). Returns
* -EINVAL if core is NULL.
*/
extern int xce_core_begin_session(xce_sched_core_t*);
extern int xce_core_signal_go(xce_sched_core_t*);
/*
* xce_core_go - Req 5 AC 5 (scheduler-core spec, task 4.4). Starts a
* scheduling session: NON-BLOCKING -- performs the begin-session state
@ -347,6 +397,13 @@ extern int xce_core_set_decision_cb(xce_sched_core_t*, xce_core_decision_cb_t,
* (design.md's Normal-Mode Scheduling Round: "the drain loop in the
* driver thread interleaves cbuf_get/on_decision one at a time").
*
* As of task 5.2, implemented as `xce_core_begin_session()` immediately
* followed by `xce_core_signal_go()` -- net behavior, return codes, and
* the existing test_core_session.c contract are all unchanged; this is a
* pure decomposition, kept as the simple, safe entry point for any caller
* (e.g. a future simulation driver) with no begin-session bookkeeping of
* its own to interleave.
*
* 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.

78
src/scheduler/include/driver.h
View file

@ -52,4 +52,82 @@
*/
extern int xce_sched_driver_translate_pending_events(xce_sched_core_t* core);
/*
* xce_sched_driver_on_decision - the normal-mode driver's decision-publish
* callback (scheduler-core spec, task 5.2, Req 5 AC 6 / AC 9's publishing
* half: "The driver shall publish core-emitted decisions back to the
* controller via the existing xcecli_ctx_t API"). Matches
* xce_core_decision_cb_t's signature (core.h) exactly -- registered via
* `xce_core_set_decision_cb(&sched->core, xce_sched_driver_on_decision,
* sched)` in main.c; `driver_ctx` is always the live xce_scheduler_t*, cast
* back inside driver.c so this function can reach
* sched_publish_dispatch()/sched_publish_pend_reason() (scheduler.h), both
* of which need the FULL xce_scheduler_t* (apictx) -- not just
* xce_sched_core_t*, which is why this declaration (unlike
* xce_sched_driver_translate_pending_events above) takes an opaque void*
* rather than pulling scheduler.h into this header (this file's own
* boundary comment: driver.h only depends on core.h).
*
* Replaces main.c's former inline `switch (decision->op)` (Req 5 AC 6:
* "This replaces the current internal decisions cbuf poll loop in
* main.c"). Called once per decision, in allocator-output order, by the
* driver's own drain loop between xce_core_signal_go() and
* xce_core_end_session() -- publishing immediately as each decision is
* drained, never batched (Req 5 AC 9).
*
* XCE_DECISION_PREEMPT: no live allocator path produces this decision type
* yet (preemption belongs to the sibling scheduler-preemption spec, out of
* this spec's boundary) and no `xcecli_preempt()` client API function
* exists anywhere in client.h to publish it through even if one did. This
* case fails loudly (logged, -ENOSYS) rather than silently no-op'ing or
* calling an undefined symbol -- see driver.c for the full reasoning.
*
* Returns 0 on successful publish (or a deliberate no-op case); a negative
* errno otherwise: -EINVAL if driver_ctx or decision is NULL; -ENOSYS for
* XCE_DECISION_PREEMPT (see above); or sched_publish_dispatch()'s/
* sched_publish_pend_reason()'s own negative return, passed straight
* through.
*/
extern int xce_sched_driver_on_decision(void* driver_ctx,
xce_decision_t* decision);
/*
* xce_sched_driver_run_normal_mode_round - one normal-mode scheduling round
* (scheduler-core spec, task 5.2, Req 1 AC 7 / Req 5 AC 6 / AC 8 / AC 9).
* Originally implemented as a `static` helper directly inside main.c's
* loop body; relocated here (same file, same naming convention as
* xce_sched_driver_translate_pending_events()/xce_sched_driver_on_decision()
* above) so it is a real, linkable, independently testable driver entry
* point rather than a main.c-private static function no test binary could
* reach -- see testsuite/test_driver_session_ordering.c, which white-box
* #includes driver.c to exercise this exact function.
*
* Takes an opaque `void*` (always the live `xce_scheduler_t*`, cast back
* inside driver.c) for the SAME reason xce_sched_driver_on_decision() above
* does: this function needs the FULL xce_scheduler_t* to reach
* plugin_begin_session()/sched_account_jobs()/plugin_end_session()/
* sched_publish_dispatch()/sched_publish_pend_reason() (all
* xce_scheduler_t*-coupled, scheduler.h), and this file's own boundary
* comment restricts driver.h to depending on core.h only.
*
* One round: xce_sched_driver_translate_pending_events() (task 5.1) drains
* buffered controller events; xce_core_begin_session() performs the
* begin-session state reset; plugin_begin_session()/sched_account_jobs()
* run (in that window specifically -- see core.h's doc comment on
* xce_core_begin_session()/xce_core_signal_go() for the full safety
* argument this ordering preserves); xce_core_signal_go() wakes the picker
* thread; the driver drains core->decisions, invoking the registered
* on_decision callback once per decision until the end-of-session NULL
* marker; plugin_end_session() and xce_core_end_session() close out the
* round.
*
* No return value (matches the original main.c helper's signature/
* behavior exactly -- a pure relocation, not a redesign): failures along
* the way (e.g. xce_core_begin_session() rejecting a reentrant call) are
* logged (ERROR) and the round is abandoned early: no picker signal, no
* decision drain, no plugin_end_session()/xce_core_end_session() for that
* malformed round -- the next call to this function tries again.
*/
extern void xce_sched_driver_run_normal_mode_round(void* driver_ctx);
#endif

11
src/scheduler/include/scheduler.h
View file

@ -121,6 +121,17 @@ extern int sched_update(xce_scheduler_t*);
extern void sched_go(xce_scheduler_t*);
extern void sched_begin_session(xce_scheduler_t*);
extern void sched_end_session(xce_scheduler_t*);
/* sched_account_jobs - per-running-job resource-in-use accounting
* (re-charges every RUN-state job's resources via sched_charge_account()
* after sched_begin_session()'s resval reset zeroes them). Exported as of
* scheduler-core spec task 5.2 so the new normal-mode driver loop
* (driver.c's xce_sched_driver_run_normal_mode_round()) can call it
* directly at the exact
* point xce_core_begin_session()'s caller must call it: after the resval
* reset, strictly before xce_core_signal_go() unblocks the picker thread.
* Was `static`; body unchanged, still also called from
* sched_begin_session() below (unchanged legacy call site). */
extern int sched_account_jobs(xce_scheduler_t*);
/* sched_get_job/sched_add_job/sched_get_queue/sched_add_queue/
* sched_get_worker/sched_add_worker/sched_get_user/sched_add_user/
* sched_get_project/sched_add_project/sched_job_is_ready/

41
src/scheduler/src/core_session.c
View file

@ -343,18 +343,27 @@ int xce_core_set_decision_cb(xce_sched_core_t* core, xce_core_decision_cb_t cb,
return 0;
}
int xce_core_go(xce_sched_core_t* core)
/*
* xce_core_begin_session / xce_core_signal_go - task 5.2 split of the
* former monolithic xce_core_go() body (see core.h's doc comment on these
* two functions for the full safety rationale: a driver interleaving its
* own begin-session bookkeeping -- plugin_begin_session()/
* sched_account_jobs() -- between these two calls is what preserves the
* "picker never observes zeroed-but-not-yet-resummed resource accounting"
* property). xce_core_go() below is now defined purely in terms of these
* two, so its own behavior/return codes/tests are unaffected.
*/
int xce_core_begin_session(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). */
* call is a true no-op (no begin-session reset performed, and -- for
* the xce_core_go() composition below -- no signal placed on wbuf
* either). */
if (atomic_load_explicit(&(core->session_active), memory_order_acquire)
!= 0) {
return -EINVAL;
@ -369,6 +378,17 @@ int xce_core_go(xce_sched_core_t* core)
core_begin_session_internal(core);
return 0;
}
int xce_core_signal_go(xce_sched_core_t* core)
{
int go = 1;
if (core == NULL) {
return -EINVAL;
}
/* 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
@ -379,6 +399,17 @@ int xce_core_go(xce_sched_core_t* core)
return 0;
}
int xce_core_go(xce_sched_core_t* core)
{
int rc = xce_core_begin_session(core);
if (rc < 0) {
return rc;
}
return xce_core_signal_go(core);
}
int xce_core_end_session(xce_sched_core_t* core)
{
if (core == NULL) {

174
src/scheduler/src/driver.c
View file

@ -384,3 +384,177 @@ int xce_sched_driver_translate_pending_events(xce_sched_core_t* core)
return (push_rc < 0) ? push_rc : n;
}
/*
* xce_sched_driver_on_decision - see driver.h's declaration comment for the
* full contract (scheduler-core spec, task 5.2, Req 5 AC 6 / AC 9's
* publishing half).
*
* DISPATCH/PEND: forwards to sched_publish_dispatch()/
* sched_publish_pend_reason() (scheduler.c) with the SAME single-item
* njobs=1/nreasons=1 wrapping main.c's former inline switch used --
* xce_job_dispatch_t/xce_preason_set_t only ever carry one decision's
* worth of payload per call here, unchanged.
*
* PREEMPT: task 4.4 added XCE_DECISION_PREEMPT as a decision TYPE
* (decision.h) and its own test suite (test_core_session.c) already
* exercises constructing/draining one -- but nothing in the live allocator
* path (sched_allocator()/sched_make_allocation(), scheduler.c) actually
* PRODUCES one today: preemption/can_preempt() logic belongs to the
* sibling scheduler-preemption spec, explicitly out of this spec's
* boundary (design.md "Out of Boundary"). Verified via grep of the whole
* lib/ tree: `xcecli_preempt()` -- the client API function a real
* implementation would need to call -- does not exist anywhere in
* client.h. Fabricating one here would be scope creep into src/lib, a
* bigger change than "wire decision publishing" should require, and
* arguably scheduler-preemption's own job to define the wire contract for.
* This case is therefore currently unreachable in production (dead code by
* construction), but must not be UNSAFE dead code -- it fails LOUDLY
* (ERROR log naming exactly what is missing) and returns -ENOSYS, never a
* silent no-op and never a call through an undefined symbol.
*/
int xce_sched_driver_on_decision(void* driver_ctx, xce_decision_t* decision)
{
xce_scheduler_t* sched = (xce_scheduler_t*)driver_ctx;
int rc = 0;
if ((sched == NULL) || (decision == NULL)) {
return -EINVAL;
}
switch (decision->op) {
case XCE_DECISION_DISPATCH: {
xce_job_alloc_t* allocs[1] = { decision->u.alloc };
xce_job_dispatch_t jdispatch = { .njobs = 1, .allocs = allocs };
rc = sched_publish_dispatch(sched, &jdispatch);
break;
}
case XCE_DECISION_PEND: {
xce_pendreason_t* preasons[1] = { decision->u.reason };
xce_preason_set_t reasons = { .nreasons = 1, .reasons = preasons };
rc = sched_publish_pend_reason(sched, &reasons);
break;
}
case XCE_DECISION_PREEMPT:
ERROR("%s PREEMPT decision publishing not yet implemented "
"(scheduler-preemption spec territory) -- jobid=%llu "
"dropped, no xcecli_preempt() client API exists to publish "
"it",
__func__,
(unsigned long long)((decision->u.preempt != NULL)
? decision->u.preempt->jobid
: 0));
rc = -ENOSYS;
break;
default:
/* XCE_DECISION_NONE / XCE_DECISION_NEW_SESSION / any future op:
* no publishing action, same as main.c's old inline switch's own
* "default: break" -- not an error. */
break;
}
return rc;
}
/*
* xce_sched_driver_run_normal_mode_round - see driver.h's declaration
* comment for the full contract (scheduler-core spec, task 5.2, Req 1 AC 7
* / Req 5 AC 6 / AC 8 / AC 9). Relocated here (from a main.c-private
* `static run_normal_mode_round()` helper) purely so it is independently
* testable -- see driver.h's declaration comment for the full rationale.
* Body unchanged from the original main.c helper.
*
* CRITICAL ordering (see core.h's doc comment on xce_core_begin_session()/
* xce_core_signal_go() for the full safety argument, and
* testsuite/test_driver_session_ordering.c for the proof): today's
* sched_begin_session() also runs plugin_begin_session() (the OLD
* per-plugin begin_session_fn hook category -- real, load-bearing
* per-session bookkeeping for EVERY registered plugin, e.g. the legacy
* "xce_sched_fcfs" plugin's fcfs_end_session(), still registered alongside
* builtin_fcfs) and sched_account_jobs() (re-charges every RUNNING job's
* resources after the resval reset zeroes them) BEFORE sched_go() signals
* the picker thread. xce_core_begin_session()/xce_core_end_session()
* (core_session.c, task 4.4) deliberately exclude both -- they need the
* full xce_scheduler_t*, not just xce_sched_core_t* -- so this function
* calls them itself, in the same slot: strictly AFTER
* xce_core_begin_session()'s resval reset, strictly BEFORE
* xce_core_signal_go() unblocks the picker. Neither plugin_begin_session()
* nor sched_account_jobs() has any data dependency on WHERE within that
* window it runs (verified: plugin_begin_session() only touches
* plugin-private state via begin_session_fn(plugin), never
* core->sel_eval_ctx/resvals; sched_account_jobs() only needs the resval
* reset to have already happened) -- so this ordering, while not
* byte-identical to the old sched_begin_session()'s internal interleaving
* (select-eval-ctx reset -> plugin_begin_session -> resval resets ->
* sched_account_jobs), is a proven-safe reordering, not a silent drop.
*
* plugin_end_session() (mirrors plugin_begin_session() above, e.g.
* resetting fcfs's own `new_session` flag) is likewise called every round,
* after the decision drain completes -- its exact position relative to
* xce_core_end_session()'s select-eval-ctx teardown/GC is not
* safety-critical either (disjoint state, single-threaded driver loop --
* the picker/allocator threads are idle between sessions, and nothing else
* calls xce_core_begin_session() concurrently), so it runs immediately
* before xce_core_end_session(), mirroring "plugin bookkeeping is part of
* ending the session."
*/
void xce_sched_driver_run_normal_mode_round(void* driver_ctx)
{
xce_scheduler_t* sched = (xce_scheduler_t*)driver_ctx;
int rc;
if (sched == NULL) {
return;
}
/* Req 5 AC 8 (task 5.1): drain + translate controller events
* accumulated since the last round. Replaces sched_update()'s inline
* dataq-swap handling. Only valid between sessions (driver.h's own doc
* comment) -- called here, before xce_core_begin_session(), exactly
* where sched_update() ran relative to sched_begin_session() before. */
xce_sched_driver_translate_pending_events(&(sched->core));
rc = xce_core_begin_session(&(sched->core));
if (rc < 0) {
ERROR("%s xce_core_begin_session failed rc=%d", __func__, rc);
return;
}
/* See this function's header comment: MUST run after the resval reset
* xce_core_begin_session() just performed, MUST complete before
* xce_core_signal_go() below. */
plugin_begin_session(sched);
sched_account_jobs(sched);
xce_core_signal_go(&(sched->core));
for (;;) {
xce_decision_t* decision = NULL;
cbuf_get(&(sched->core.decisions), &decision);
TRACE("%s decision = %p", __func__, decision);
if (decision == NULL) {
/* end-of-session marker -- start new session */
break;
}
if (sched->core.on_decision != NULL) {
rc = sched->core.on_decision(sched->core.driver_ctx, decision);
TRACE("%s rc %d", __func__, rc);
} else {
WARN("%s no decision callback registered, dropping decision "
"op=%d",
__func__, (int)decision->op);
}
}
plugin_end_session(sched);
xce_core_end_session(&(sched->core));
}

86
src/scheduler/src/main.c
View file

@ -10,6 +10,7 @@
#include "xce/cluster.h"
#include "xce/types.h"
#include "driver.h"
#include "scheduler.h"
#include "resreq.h"
@ -27,13 +28,22 @@ const char* argp_program_bug_address = BUG_EMAIL;
struct arguments {
char* logfile;
int verbose;
char* peers; /* --peers value (optional) */
char* peers; /* --peers value (optional) */
char* sim_file; /* --sim value (optional; Phase 3, not yet implemented) */
};
static struct argp_option options[] = {
{ "logfile", 'l', "FILE", 0, "The full path of log file" },
{ "verbose", 'v', 0, 0, "Produce verbose log" },
{ "peers", 'P', "LIST", 0, "HA candidate list: [name=]host:port,..." },
{ "logfile", 'l', "FILE", 0, "The full path of log file" },
{ "verbose", 'v', 0, 0, "Produce verbose log" },
{ "peers", 'P', "LIST", 0, "HA candidate list: [name=]host:port,..." },
/* Req 5 AC 9 "Mode selection" / "CLI flags" (scheduler-core spec, task
* 5.2): --sim FILE selects the simulation driver. Parsing only, as of
* this task -- SimDriver itself is Phase 3 (design.md's File Structure
* Plan marks sim_driver.c/sim_driver.h "NEW (Phase 3)"); see main()'s
* mode-selection branch below for the loud not-yet-implemented failure. */
{ "sim", 's', "FILE", 0,
"Replay FILE for simulation mode "
"(Phase 3 -- not yet implemented)" },
{ 0 }
};
@ -51,6 +61,9 @@ static error_t parse_opt(int key, char* arg, struct argp_state* state)
case 'P':
arguments->peers = arg;
break;
case 's':
arguments->sim_file = arg;
break;
default:
return ARGP_ERR_UNKNOWN;
}
@ -119,6 +132,20 @@ int main(int argc, char* argv[])
mylog_init(logfile, loglevel, logsize, logcount);
/* Req 5 AC 9 "Mode selection" (scheduler-core spec, task 5.2): --sim
* selects the simulation driver (SimDriver). Not yet implemented as of
* this task (Phase 3 -- design.md's File Structure Plan explicitly
* marks sim_driver.c/sim_driver.h "NEW (Phase 3)"; sched_sim_driver_run()
* does not exist). Fail LOUDLY and immediately rather than silently
* no-op'ing or crashing -- an honest, temporary placeholder for Phase 3
* to fill in. Checked before any network/config setup below: there is
* no reason to pay for a controller connection attempt on a path that
* is about to fail. */
if (arguments.sim_file != NULL) {
fprintf(stderr, "xce_scheduler: --sim mode not yet implemented\n");
return 1;
}
/* P2.5-6: populate and validate xce_cluster_config_t when --peers given */
if (arguments.peers) {
if (xce_parse_peers(arguments.peers, &cluster_cfg) != 0) {
@ -152,53 +179,20 @@ int main(int argc, char* argv[])
sched_init(&sched, &config, cluster_cfg_ptr, scheduler_req_handler);
/* Req 5 AC 6 (scheduler-core spec, task 5.2): register the driver's
* decision callback -- replaces main.c's former inline
* `switch (decision->op)`. Registered once, before the first
* scheduling round, alongside the other one-time sched_init()
* follow-up steps below. */
xce_core_set_decision_cb(&(sched.core), xce_sched_driver_on_decision,
&sched);
signal(SIGINT, sig_handler);
INFO("xce scheduler is running");
while (1) {
sched_update(&sched);
sched_begin_session(&sched);
sched_go(&sched);
while (1) {
xce_decision_t* decision = NULL;
int rc = 0;
xce_pendreason_t* preasons[1];
xce_preason_set_t reasons = { .nreasons = 1, .reasons = preasons };
xce_job_alloc_t* allocs[1];
xce_job_dispatch_t jdispatch = { .njobs = 1, .allocs = allocs };
cbuf_get(&(sched.core.decisions), &decision);
TRACE("%s decision = %p", __func__, decision);
if (decision == NULL) {
/* start new session */
break;
}
switch (decision->op) {
case XCE_DECISION_NONE:
break;
case XCE_DECISION_DISPATCH:
/* tell controller to dispatch job */
allocs[0] = decision->u.alloc;
rc = sched_publish_dispatch(&sched, &jdispatch);
break;
case XCE_DECISION_PEND:
/* set pend reason for jobs */
preasons[0] = decision->u.reason;
rc = sched_publish_pend_reason(&sched, &reasons);
break;
default:
break;
}
TRACE("%s rc %d", __func__, rc);
}
sched_end_session(&sched);
xce_sched_driver_run_normal_mode_round(&sched);
if (sched.core.stop) {
break;

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

@ -1018,7 +1018,10 @@ static int sched_reset_workers(xce_scheduler_t* sched)
return rc;
}
static int sched_account_jobs(xce_scheduler_t* sched)
/* No longer `static` as of scheduler-core spec task 5.2 -- see the
* declaration comment in scheduler.h for why the new normal-mode driver
* loop (main.c) needs to call this directly. Body unchanged. */
int sched_account_jobs(xce_scheduler_t* sched)
{
iterator_t it;
int rc = 0;

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

@ -315,6 +315,61 @@ sched_driver_translate_test = executable(
dependencies: [scheduler_test_deps],
install: false)
# Contract test for the normal-mode driver's decision-publish callback
# (scheduler-core spec, task 5.2, Req 5 AC 6 / AC 9's publishing half):
# xce_sched_driver_on_decision() forwards DISPATCH/PEND decisions to
# sched_publish_dispatch()/sched_publish_pend_reason() exactly once each,
# with the same single-item njobs=1/nreasons=1 wrapping; PREEMPT fails
# loudly (-ENOSYS, no publish) rather than fabricating a nonexistent
# xcecli_preempt() call; NULL driver_ctx/decision each return -EINVAL, no
# publish. Same white-box #include pattern as test_driver_translate.c
# (pulls ../src/core.c, ../src/core_clock.c, ../src/core_events.c, and the
# new ../src/driver.c into this TU) PLUS ../src/decision.c for the real
# make_dispatch_decision()/make_pend_decision()/make_preempt_decision()
# constructors this file's fixtures use. sched_publish_dispatch()/
# sched_publish_pend_reason() are SPY test doubles (this file never links
# scheduler.c), same rationale as test_driver_translate.c's own stubs.
sched_driver_decision_cb_test = executable(
'sched_driver_decision_cb_test',
['test_driver_decision_cb.c'],
include_directories:[incdir],
dependencies: [scheduler_test_deps],
install: false)
# Safety-property proof for the begin/end-session ordering seam task 5.2
# introduced (Req 1 AC 7 / Req 5 AC 9): xce_sched_driver_run_normal_mode_round()
# (driver.c) must call plugin_begin_session()/sched_account_jobs() strictly
# AFTER xce_core_begin_session()'s resval reset and strictly BEFORE
# xce_core_signal_go() wakes the picker thread -- otherwise the picker could
# observe a session whose resource-in-use accounting was reset to zero but
# not yet re-summed from running jobs. Drives the REAL
# xce_sched_driver_run_normal_mode_round() (white-box #include of
# ../src/driver.c, same TU as test_driver_translate.c/
# test_driver_decision_cb.c, PLUS ../src/core_session.c for the real
# xce_core_begin_session()/xce_core_signal_go()/xce_core_end_session()
# under test) against a real picker pthread (same design as
# test_core_full_cycle.c's test_picker_thread_fn -- see that file's own
# header comment for the full "why a real thread, not a sequential drain"
# rationale: jbuf's hardcoded capacity-1 rendezvous requires one). Uses the
# REAL plugin_begin_session()/plugin_end_session() (../src/plugin.c) driven
# through a genuinely test-registered xce_sched_plugin_t whose
# begin_session_fn/end_session_fn set/check an ordering flag -- per this
# task's own suggested technique, this is a stronger proof than spying on
# plugin_begin_session() itself. sched_account_jobs() (scheduler.c,
# xce_scheduler_t*-coupled, pulls in resource.c's sched_charge_account())
# is instead a SPY test double recording an order index -- this test's
# target is the DRIVER's ordering discipline, not sched_account_jobs()'s
# own resource-accounting correctness (out of this task's scope, same
# "spy the xce_scheduler_t*-coupled collaborator" rationale
# test_driver_decision_cb.c/test_driver_translate.c already established for
# sched_publish_dispatch()/sched_publish_pend_reason()).
sched_driver_session_ordering_test = executable(
'sched_driver_session_ordering_test',
['test_driver_session_ordering.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',
@ -366,3 +421,9 @@ test('scheduler core full cycle test',
test('scheduler driver translate test',
sched_driver_translate_test, suite: 'scheduler',
env: ['ASAN_OPTIONS=detect_leaks=0'])
test('scheduler driver decision cb test',
sched_driver_decision_cb_test, suite: 'scheduler',
env: ['ASAN_OPTIONS=detect_leaks=0'])
test('scheduler driver session ordering test',
sched_driver_session_ordering_test, suite: 'scheduler',
env: ['ASAN_OPTIONS=detect_leaks=0'])

304
src/scheduler/testsuite/test_driver_decision_cb.c Normal file
View file

@ -0,0 +1,304 @@
/*
* test_driver_decision_cb.c - contract test for the normal-mode driver's
* decision-publish callback (scheduler-core spec, task 5.2, Req 5 AC 6 /
* AC 9's publishing half).
*
* Observable under test: xce_sched_driver_on_decision() -- registered via
* xce_core_set_decision_cb() in main.c and matching xce_core_decision_cb_t's
* signature exactly -- forwards each decision to the right publish call:
*
* - XCE_DECISION_DISPATCH -> sched_publish_dispatch() called exactly
* once, with njobs=1 and allocs[0] == the SAME xce_job_alloc_t* the
* decision carried (not a copy).
* - XCE_DECISION_PEND -> sched_publish_pend_reason() called exactly
* once, with nreasons=1 and reasons[0] == the SAME xce_pendreason_t*
* the decision carried.
* - XCE_DECISION_PREEMPT -> NEITHER publish function is called (no live
* allocator path produces this decision type today, and no
* xcecli_preempt() client API exists to publish it through even if one
* did -- see driver.c's own header comment on this function for the
* full reasoning); returns -ENOSYS, never crashes, never silently
* drops without a trace (verified via the ERROR log call count below).
* - NULL driver_ctx / NULL decision -> -EINVAL, no publish calls.
*
* White-box #include of ../src/core.c, ../src/core_clock.c,
* ../src/core_events.c, ../src/driver.c, and ../src/decision.c -- same
* convention (and same file set) as test_driver_translate.c (task 5.1),
* since driver.c's xce_sched_driver_translate_pending_events() -- unused by
* this file's tests but part of the same TU -- still references
* xce_core_push_events()/xce_core_now()/sched_get_job() at link time.
* decision.c is added (over test_driver_translate.c's set) for the real
* make_dispatch_decision()/make_pend_decision()/make_preempt_decision()
* constructors this file's fixtures use to build realistic xce_decision_t
* values, same convention as test_core_session.c.
*
* sched_publish_dispatch()/sched_publish_pend_reason() below are SPY test
* doubles (not the real scheduler.c/xcecli_ctx_t-backed versions -- this
* file never links scheduler.c, same rationale as test_driver_translate.c's
* own stubs) that record every call's arguments so this file can assert on
* the driver's call CONTRACT (Req 5 AC 9: "publish core-emitted decisions
* back to the controller via the existing xcecli_ctx_t API") without a real
* or mocked network connection.
*/
#include <errno.h>
#include <stdint.h>
#include <stdlib.h>
#include <string.h>
#include <unity/unity.h>
#include "controller.capnp.h"
#include "kvmgr.h"
#include "preason.h"
#include "scheduler.h"
#include "xstring.h"
/* Pull in the real xce_sched_driver_on_decision() (and the rest of
* driver.c's TU -- see file header comment) under test. */
#include "../src/core.c"
#include "../src/core_clock.c"
#include "../src/core_events.c"
#include "../src/decision.c"
#include "../src/driver.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; }
/* Link-satisfying stubs: as of scheduler-core spec task 5.2,
* xce_sched_driver_run_normal_mode_round() (driver.c, real, #included
* above, alongside xce_sched_driver_on_decision() this file DOES exercise)
* references xce_core_begin_session()/xce_core_signal_go()/
* xce_core_end_session() (core_session.c) and plugin_begin_session()/
* plugin_end_session() (plugin.c) / sched_account_jobs() (scheduler.c).
* This file never exercises xce_sched_driver_run_normal_mode_round() itself
* (out of THIS file's scope -- see test_driver_session_ordering.c) and
* never links core_session.c/plugin.c/scheduler.c. Never actually invoked
* at runtime by any test in this file. */
int xce_core_begin_session(xce_sched_core_t* core)
{
(void)core;
return 0;
}
int xce_core_signal_go(xce_sched_core_t* core)
{
(void)core;
return 0;
}
int xce_core_end_session(xce_sched_core_t* core)
{
(void)core;
return 0;
}
int plugin_begin_session(void* arg)
{
(void)arg;
return 0;
}
int plugin_end_session(void* arg)
{
(void)arg;
return 0;
}
int sched_account_jobs(xce_scheduler_t* sched)
{
(void)sched;
return 0;
}
/* Real reimplementation of schedobj.c's sched_resolve_job_picker -- same
* rationale as test_driver_translate.c's own copy (this file never links
* schedobj.c/scheduler.c). Never actually exercised (no test here pushes
* core events), present only to satisfy the linker for core_events.c. */
int sched_resolve_job_picker(xce_sched_core_t* core, xce_sched_plugin_t** out)
{
(void)core;
(void)out;
return -EINVAL;
}
/* ---------------------------------------------------------------------- */
/* Spy test doubles for scheduler.c's sched_publish_dispatch()/ */
/* sched_publish_pend_reason() -- see file header comment. */
/* ---------------------------------------------------------------------- */
static int g_dispatch_calls;
static xce_job_dispatch_t g_last_dispatch;
static xce_job_alloc_t* g_last_dispatch_alloc;
static xce_scheduler_t* g_last_dispatch_sched;
static int g_pend_calls;
static xce_preason_set_t g_last_pend;
static xce_pendreason_t* g_last_pend_reason;
static xce_scheduler_t* g_last_pend_sched;
static void reset_spies(void)
{
g_dispatch_calls = 0;
g_last_dispatch_alloc = NULL;
g_last_dispatch_sched = NULL;
memset(&g_last_dispatch, 0, sizeof(g_last_dispatch));
g_pend_calls = 0;
g_last_pend_reason = NULL;
g_last_pend_sched = NULL;
memset(&g_last_pend, 0, sizeof(g_last_pend));
}
int sched_publish_dispatch(xce_scheduler_t* sched, xce_job_dispatch_t* d)
{
g_dispatch_calls++;
g_last_dispatch_sched = sched;
g_last_dispatch = *d;
g_last_dispatch_alloc = (d->njobs > 0) ? d->allocs[0] : NULL;
return 0;
}
int sched_publish_pend_reason(xce_scheduler_t* sched, xce_preason_set_t* r)
{
g_pend_calls++;
g_last_pend_sched = sched;
g_last_pend = *r;
g_last_pend_reason = (r->nreasons > 0) ? r->reasons[0] : NULL;
return 0;
}
void setUp(void) { reset_spies(); }
void tearDown(void) { return; }
/* ---------------------------------------------------------------------- */
/* XCE_DECISION_DISPATCH */
/* ---------------------------------------------------------------------- */
static void test_dispatch_decision_publishes_via_sched_publish_dispatch(void)
{
xce_scheduler_t sched;
xce_job_alloc_t* alloc = (xce_job_alloc_t*)calloc(1, sizeof(*alloc));
xce_decision_t* d;
int rc;
memset(&sched, 0, sizeof(sched));
alloc->jobid = 42;
d = make_dispatch_decision(alloc);
TEST_ASSERT_NOT_NULL(d);
rc = xce_sched_driver_on_decision(&sched, d);
TEST_ASSERT_EQUAL_INT(0, rc);
TEST_ASSERT_EQUAL_INT(1, g_dispatch_calls);
TEST_ASSERT_EQUAL_INT(0, g_pend_calls);
TEST_ASSERT_EQUAL_PTR(&sched, g_last_dispatch_sched);
TEST_ASSERT_EQUAL_INT(1, g_last_dispatch.njobs);
TEST_ASSERT_EQUAL_PTR(alloc, g_last_dispatch_alloc);
TEST_ASSERT_EQUAL_UINT64(42, g_last_dispatch_alloc->jobid);
}
/* ---------------------------------------------------------------------- */
/* XCE_DECISION_PEND */
/* ---------------------------------------------------------------------- */
static void test_pend_decision_publishes_via_sched_publish_pend_reason(void)
{
xce_scheduler_t sched;
xce_pendreason_t* reason = make_pendreason(7);
xce_decision_t* d;
int rc;
memset(&sched, 0, sizeof(sched));
TEST_ASSERT_NOT_NULL(reason);
d = make_pend_decision(reason);
TEST_ASSERT_NOT_NULL(d);
rc = xce_sched_driver_on_decision(&sched, d);
TEST_ASSERT_EQUAL_INT(0, rc);
TEST_ASSERT_EQUAL_INT(1, g_pend_calls);
TEST_ASSERT_EQUAL_INT(0, g_dispatch_calls);
TEST_ASSERT_EQUAL_PTR(&sched, g_last_pend_sched);
TEST_ASSERT_EQUAL_INT(1, g_last_pend.nreasons);
TEST_ASSERT_EQUAL_PTR(reason, g_last_pend_reason);
TEST_ASSERT_EQUAL_UINT64(7, g_last_pend_reason->jobid);
}
/* ---------------------------------------------------------------------- */
/* XCE_DECISION_PREEMPT -- documented gap, must fail loudly, not crash */
/* or fabricate a call to a nonexistent client API. */
/* ---------------------------------------------------------------------- */
static void test_preempt_decision_fails_loudly_publishes_nothing(void)
{
xce_scheduler_t sched;
xce_preempt_t preempt;
xce_decision_t* d;
int rc;
memset(&sched, 0, sizeof(sched));
memset(&preempt, 0, sizeof(preempt));
preempt.jobid = 99;
preempt.nvictims = 0;
preempt.victims = NULL;
preempt.reason = make_pendreason(99);
d = make_preempt_decision(&preempt);
TEST_ASSERT_NOT_NULL(d);
rc = xce_sched_driver_on_decision(&sched, d);
TEST_ASSERT_EQUAL_INT(-ENOSYS, rc);
TEST_ASSERT_EQUAL_INT(0, g_dispatch_calls);
TEST_ASSERT_EQUAL_INT(0, g_pend_calls);
}
/* ---------------------------------------------------------------------- */
/* NULL argument guards */
/* ---------------------------------------------------------------------- */
static void test_null_driver_ctx_returns_einval_no_publish(void)
{
xce_job_alloc_t* alloc = (xce_job_alloc_t*)calloc(1, sizeof(*alloc));
xce_decision_t* d = make_dispatch_decision(alloc);
int rc;
TEST_ASSERT_NOT_NULL(d);
rc = xce_sched_driver_on_decision(NULL, d);
TEST_ASSERT_EQUAL_INT(-EINVAL, rc);
TEST_ASSERT_EQUAL_INT(0, g_dispatch_calls);
}
static void test_null_decision_returns_einval_no_publish(void)
{
xce_scheduler_t sched;
int rc;
memset(&sched, 0, sizeof(sched));
rc = xce_sched_driver_on_decision(&sched, NULL);
TEST_ASSERT_EQUAL_INT(-EINVAL, rc);
TEST_ASSERT_EQUAL_INT(0, g_dispatch_calls);
TEST_ASSERT_EQUAL_INT(0, g_pend_calls);
}
int main(void)
{
UNITY_BEGIN();
RUN_TEST(test_dispatch_decision_publishes_via_sched_publish_dispatch);
RUN_TEST(test_pend_decision_publishes_via_sched_publish_pend_reason);
RUN_TEST(test_preempt_decision_fails_loudly_publishes_nothing);
RUN_TEST(test_null_driver_ctx_returns_einval_no_publish);
RUN_TEST(test_null_decision_returns_einval_no_publish);
return UNITY_END();
}

700
src/scheduler/testsuite/test_driver_session_ordering.c Normal file
View file

@ -0,0 +1,700 @@
/*
* test_driver_session_ordering.c - the safety-property proof referenced by
* name in core.h's xce_core_begin_session()/xce_core_signal_go() doc
* comment, driver.h's xce_sched_driver_run_normal_mode_round() doc comment,
* and driver.c's own header comment (scheduler-core spec, task 5.2, Req 1
* AC 7 / Req 5 AC 9).
*
* The property under test: a normal-mode scheduling round MUST call
* plugin_begin_session()/sched_account_jobs() strictly AFTER
* xce_core_begin_session()'s resval reset and strictly BEFORE
* xce_core_signal_go() wakes the picker thread -- otherwise the picker
* could observe a session whose resource-in-use accounting was reset to
* zero but not yet re-summed from currently running jobs, letting the
* picker/allocator over-allocate resources already consumed by running
* jobs. plugin_end_session() must likewise actually run, once per
* completed round.
*
* Function under test: the REAL xce_sched_driver_run_normal_mode_round()
* (driver.c) -- NOT a hand-copied reimplementation of its call sequence.
* That function started life as a `static` helper inside main.c's loop
* body, unreachable from any test binary; it was relocated (same file
* content, task 5.2's own scope) into driver.c and exported via driver.h
* specifically so this safety property could be proven against the real
* production code path instead of a copy that could silently drift from
* it over time -- see driver.h's declaration comment for the full
* rationale. This file white-box #includes ../src/driver.c (same
* convention as test_driver_translate.c/test_driver_decision_cb.c) to pull
* that real function into this TU.
*
* Design: real picker thread + real stand-in allocator thread, NOT a
* sequential drain in the test's own thread -- same reasoning
* test_core_full_cycle.c's own header comment gives for its identical
* design choice, and the same reason applies with EXTRA force here:
* xce_sched_driver_run_normal_mode_round() is now a single BLOCKING call
* (it does not return control to its caller between xce_core_signal_go()
* and the decision-drain loop the way test_core_full_cycle.c's own test
* function -- which plays driver AND allocator itself -- can), so the
* calling (main/test) thread cannot ALSO act as the stand-in allocator the
* way test_core_full_cycle.c's test function does. This test therefore
* spawns two collaborator threads:
*
* - picker_thread_fn(): reproduces sched_job_picker()'s real per-session
* body verbatim (identical to test_core_full_cycle.c's
* test_picker_thread_fn), single-shot for this test's one session.
* This fixture pushes no jobs, so the picker's own next_job() returns
* NULL immediately -- but the wbuf rendezvous (cbuf_get(&core->wbuf))
* this thread blocks on before doing ANYTHING else is exactly the
* cross-thread synchronization point that makes the ordering
* assertions below sound: whatever the driver did in its own thread
* BEFORE xce_core_signal_go()'s cbuf_put(&core->wbuf) is guaranteed
* visible-before whatever this thread does AFTER its cbuf_get(&wbuf)
* returns (the SAME reasoning test_core_full_cycle.c's own header
* comment documents for its analogous claim).
*
* - stand_in_allocator_thread_fn(): drains the picker's jbuf
* end-of-session sentinel (jobid == 0 -- the only jbuf item this
* fixture ever produces, since no jobs are pushed) and pushes the
* matching NULL end-of-session marker onto core->decisions, exactly
* as sched_allocator() (scheduler.c) does for a session that elects no
* jobs. This is what unblocks xce_sched_driver_run_normal_mode_round()'s
* own decision-drain loop so the round can complete and reach
* plugin_end_session()/xce_core_end_session().
*
* plugin_begin_session()/plugin_end_session() are the REAL production
* functions (white-box #include of ../src/plugin.c), driven through a
* genuinely test-registered xce_sched_plugin_t (g_observer_plugin below)
* whose begin_session_fn/end_session_fn set an atomic counter + log an
* event -- per this task's own suggested technique, this is a stronger
* proof than spying on plugin_begin_session() itself: it proves BOTH that
* xce_sched_driver_run_normal_mode_round() calls the real
* plugin_begin_session()/plugin_end_session() symbols AND that their real
* iterate-registered-plugins logic fires a genuinely registered plugin's
* hooks when driven this way. plugin.c is not otherwise exercised by any
* existing test file in this directory (codegraph's own blast-radius
* report flags plugin_begin_session/plugin_end_session/sched_account_jobs
* as having "no covering tests found" before this file), so this is
* incidental first coverage for plugin_begin_session()/plugin_end_session()
* as a byproduct of this task's own target.
*
* sched_account_jobs() (scheduler.c) is instead a SPY test double (this
* file never links scheduler.c) -- same rationale
* test_driver_decision_cb.c/test_driver_translate.c already established
* for sched_publish_dispatch()/sched_publish_pend_reason(): it is
* xce_scheduler_t*-coupled and pulls in resource.c's sched_charge_account()
* machinery, whose OWN resource-accounting correctness is out of THIS
* task's scope (this test's target is the DRIVER's ordering discipline --
* does it call sched_account_jobs() at all, and in the right window --
* not whether sched_account_jobs() itself charges resources correctly,
* which is scheduler-allocation spec territory with its own coverage).
*
* White-box #include of ../src/core.c, ../src/core_clock.c,
* ../src/core_events.c, ../src/core_session.c, ../src/decision.c,
* ../src/driver.c (same file set as test_core_full_cycle.c, needed for the
* real xce_sched_core_init()/xce_core_begin_session()/xce_core_signal_go()/
* xce_core_end_session()/xce_core_set_decision_cb() under test, transitively
* required by driver.c) PLUS ../src/plugin.c (new to this file, for the
* real plugin_begin_session()/plugin_end_session() -- see above).
*
* Stubs below: the usual kill_sched_* / sched_worker_release_host_ctx /
* sched_resolve_job_picker set (same rationale as test_core_full_cycle.c's
* own identical stubs -- this fixture never tears down, never adds a
* worker, and needs its own real reimplementation of schedobj.c's
* sched_resolve_job_picker since schedobj.c itself is not linked) PLUS
* sched_publish_dispatch()/sched_publish_pend_reason() link-satisfying
* no-op stubs (driver.c's xce_sched_driver_on_decision references them;
* never actually invoked here since this fixture produces no DISPATCH/PEND
* decisions) PLUS sched_fill_limit_requests() (resource.c) -- plugin.c's
* plugin_prepare_allocation() references it; never actually invoked here
* either (this test never calls plugin_prepare_allocation()), a pure
* link-satisfying no-op like the others. Unlike every other file in this
* directory, plugin_unregister() is NOT stubbed here -- it is the REAL
* plugin.c symbol (needed as core.c's plugins-kvmgr datakiller), never
* actually invoked at runtime since this fixture, like every other file's,
* is never torn down.
*
* CONCERN surfaced during this test's construction (documented here, not
* silently worked around): plugin_end_session() (plugin.c, pre-existing,
* untouched by this task's diff) has a latent bug -- its per-plugin guard
* checks `plugin->begin_session_fn != NULL` but then calls
* `plugin->end_session_fn(plugin)`. A plugin registering ONLY
* end_session_fn (no begin_session_fn) would silently never have its
* end_session_fn invoked; a plugin registering ONLY begin_session_fn (no
* end_session_fn) would crash via a NULL function-pointer call the first
* time a round ends. g_observer_plugin below deliberately sets BOTH
* fields, so this test's own result is unaffected by the bug -- but the
* bug is real, pre-existing, and out of this task's scope to fix (it is
* not part of task 5.2's diff and not something
* xce_sched_driver_run_normal_mode_round() can compensate for from the
* outside); flagged here for the reviewer rather than silently working around
* it.
*/
#include <errno.h>
#include <pthread.h>
#include <stdatomic.h>
#include <stdint.h>
#include <stdlib.h>
#include <string.h>
#include <unity/unity.h>
#include "controller.capnp.h"
#include "kvmgr.h"
#include "preason.h"
#include "scheduler.h"
#include "xstring.h"
/* Pull in the real xce_sched_core_init()/xce_core_begin_session()/
* xce_core_signal_go()/xce_core_end_session()/xce_core_set_decision_cb()/
* xce_sched_driver_run_normal_mode_round()/plugin_begin_session()/
* plugin_end_session() under test into this TU (white-box #include, same
* convention as test_core_full_cycle.c/test_driver_translate.c/
* test_driver_decision_cb.c). */
#include "../src/core.c"
#include "../src/core_clock.c"
#include "../src/core_events.c"
#include "../src/core_session.c"
#include "../src/decision.c"
#include "../src/driver.c"
#include "../src/plugin.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; }
/* core_reset_workers()'s (core_session.c) linker dependency -- see file
* header comment; never invoked (no worker is ever added here). */
void sched_worker_release_host_ctx(xce_sched_worker_t* w) { (void)w; }
/* Link-satisfying stub for scheduler.c's sched_publish_dispatch()/
* sched_publish_pend_reason() -- see file header comment. Never actually
* invoked at runtime: this fixture pushes no jobs, so the driver's
* decision-drain loop never observes a non-NULL decision, so
* xce_sched_driver_on_decision() (driver.c) is never called here. */
int sched_publish_dispatch(xce_scheduler_t* sched, xce_job_dispatch_t* d)
{
(void)sched;
(void)d;
return 0;
}
int sched_publish_pend_reason(xce_scheduler_t* sched, xce_preason_set_t* r)
{
(void)sched;
(void)r;
return 0;
}
/* Link-satisfying stub for resource.c's sched_fill_limit_requests() --
* plugin.c's plugin_prepare_allocation() references it; never actually
* invoked here (this test never calls plugin_prepare_allocation()). See
* file header comment. */
int sched_fill_limit_requests(xce_scheduler_t* sched, xvec_idval_t* vec,
xlist_t* limits)
{
(void)sched;
(void)vec;
(void)limits;
return 0;
}
/* Real reimplementation of schedobj.c's sched_resolve_job_picker -- see
* file header comment (same rationale/algorithm as test_core_full_cycle.c's
* own copy). */
int sched_resolve_job_picker(xce_sched_core_t* core, xce_sched_plugin_t** out)
{
xce_sched_plugin_t* plugin = NULL;
int rc = 0;
if ((core == NULL) || (out == NULL)) {
return -EINVAL;
}
*out = NULL;
if (core->job_pick_plugin == NULL) {
return -EINVAL;
}
lock(core->plugins, RTTI_LOCK_READ);
rc = key_find(core->plugins, core->job_pick_plugin, (void**)&plugin);
unlock(core->plugins);
if (rc <= 0) {
return (rc < 0) ? rc : -ENOENT;
}
if ((plugin == NULL) || (plugin->picker_iface == NULL)
|| (plugin->picker_iface->next_job == NULL)
|| (plugin->picker_iface->begin_session == NULL)
|| (plugin->picker_iface->end_session == NULL)
|| (plugin->picker_iface->on_dispatch == NULL)
|| (plugin->picker_iface->on_pend == NULL)) {
return -EINVAL;
}
*out = plugin;
return 0;
}
/* sched_account_jobs() itself is defined further below (SPY test double
* section) -- its prototype already comes from scheduler.h, included
* above. */
void setUp(void) { return; }
void tearDown(void) { return; }
/* ---------------------------------------------------------------------- */
/* Event log: records hook/thread firing order across all three threads */
/* involved (main/driver thread, picker thread, stand-in allocator thread) */
/* via a single atomically-claimed index per entry -- same pattern/ */
/* thread-safety argument as test_core_full_cycle.c's own g_events. */
/* ---------------------------------------------------------------------- */
typedef enum {
EVT_OBSERVER_BEGIN_SESSION,
EVT_ACCOUNT_JOBS,
EVT_PICKER_BEGIN_SESSION,
EVT_PICKER_NEXT_JOB_NULL,
EVT_PICKER_END_SESSION,
EVT_ALLOCATOR_DRAINED_JBUF_SENTINEL,
EVT_ALLOCATOR_PUSHED_END_DECISION,
EVT_OBSERVER_END_SESSION,
} ordering_event_tag_t;
#define MAX_EVENTS 32
typedef struct {
ordering_event_tag_t tags[MAX_EVENTS];
_Atomic int n;
} event_log_t;
static event_log_t g_events;
static void log_event(ordering_event_tag_t tag)
{
int idx = atomic_fetch_add(&(g_events.n), 1);
if (idx < MAX_EVENTS) {
g_events.tags[idx] = tag;
}
}
static int find_log_index(ordering_event_tag_t tag)
{
int n = atomic_load(&(g_events.n));
int cap = (n > MAX_EVENTS) ? MAX_EVENTS : n;
int i;
for (i = 0; i < cap; i++) {
if (g_events.tags[i] == tag) {
return i;
}
}
return -1;
}
/* ---------------------------------------------------------------------- */
/* sched_account_jobs() SPY -- see file header comment. */
/* ---------------------------------------------------------------------- */
static _Atomic int g_account_jobs_calls;
int sched_account_jobs(xce_scheduler_t* sched)
{
(void)sched;
atomic_fetch_add(&g_account_jobs_calls, 1);
log_event(EVT_ACCOUNT_JOBS);
return 0;
}
/* ---------------------------------------------------------------------- */
/* Observer plugin: OLD begin_session_fn/end_session_fn hook category */
/* (plugin.h) -- genuinely registered into core->plugins and driven by */
/* the REAL plugin_begin_session()/plugin_end_session() (plugin.c). See */
/* file header comment for why both fields are set (sidesteps the */
/* documented plugin_end_session() guard-field bug). */
/* ---------------------------------------------------------------------- */
static _Atomic int g_observer_begin_calls;
static _Atomic int g_observer_end_calls;
static int observer_begin_session_fn(void* arg)
{
(void)arg;
atomic_fetch_add(&g_observer_begin_calls, 1);
log_event(EVT_OBSERVER_BEGIN_SESSION);
return 0;
}
static int observer_end_session_fn(void* arg)
{
(void)arg;
atomic_fetch_add(&g_observer_end_calls, 1);
log_event(EVT_OBSERVER_END_SESSION);
return 0;
}
static xce_sched_plugin_t g_observer_plugin
= { .name = "test_ordering_observer",
.begin_session_fn = observer_begin_session_fn,
.end_session_fn = observer_end_session_fn };
/* ---------------------------------------------------------------------- */
/* Stub picker plugin: NEW picker_iface category (plugin.h) -- resolved by */
/* sched_resolve_job_picker() and driven by a real picker pthread below. */
/* This fixture never pushes any jobs, so next_job() always returns NULL */
/* immediately -- the picker's SOLE purpose here is to prove the wbuf */
/* rendezvous timing (see file header comment), not to exercise job flow. */
/* ---------------------------------------------------------------------- */
static int stub_picker_begin_session(void* picker)
{
(void)picker;
log_event(EVT_PICKER_BEGIN_SESSION);
return 0;
}
static xce_sched_job_t* stub_picker_next_job(void* picker)
{
(void)picker;
log_event(EVT_PICKER_NEXT_JOB_NULL);
return NULL;
}
static int stub_picker_end_session(void* picker)
{
(void)picker;
log_event(EVT_PICKER_END_SESSION);
return 0;
}
static int stub_picker_jobid_noop(void* picker, uint64_t jobid)
{
(void)picker;
(void)jobid;
return 0;
}
static int stub_picker_queue_update_noop(void* picker, xce_sched_queue_t* q)
{
(void)picker;
(void)q;
return 0;
}
static int stub_picker_job_noop(void* picker, xce_sched_job_t* job)
{
(void)picker;
(void)job;
return 0;
}
static int stub_picker_job_state_change_noop(void* picker, xce_sched_job_t* job,
uint32_t state)
{
(void)picker;
(void)job;
(void)state;
return 0;
}
static xce_job_picker_iface_t g_stub_picker_iface
= { .init = NULL,
.fini = NULL,
.begin_session = stub_picker_begin_session,
.next_job = stub_picker_next_job,
.end_session = stub_picker_end_session,
.on_dispatch = stub_picker_jobid_noop,
.on_pend = stub_picker_jobid_noop,
.on_queue_update = stub_picker_queue_update_noop,
.on_job_push = stub_picker_job_noop,
.on_job_state_change = stub_picker_job_state_change_noop,
.on_job_update = stub_picker_job_noop,
.on_config = NULL,
.on_job_rusage = NULL };
static xce_sched_plugin_t g_stub_picker_plugin
= { .name = "test_stub_picker", .picker_iface = &g_stub_picker_iface };
/* ---------------------------------------------------------------------- */
/* Picker thread: reproduces sched_job_picker()'s real per-session body */
/* verbatim (identical to test_core_full_cycle.c's own test_picker_thread_fn),
*/
/* single-shot -- see file header comment. */
/* ---------------------------------------------------------------------- */
static void* picker_thread_fn(void* arg)
{
xce_sched_core_t* core = (xce_sched_core_t*)arg;
xce_sched_plugin_t* job_picker = NULL;
int rc;
int go = 0;
uint64_t jobid;
rc = sched_resolve_job_picker(core, &job_picker);
if (rc < 0) {
return NULL;
}
cbuf_get(&(core->wbuf), &go);
job_picker->picker_iface->begin_session(job_picker);
for (;;) {
xce_sched_job_t* job = job_picker->picker_iface->next_job(job_picker);
if (job == NULL) {
break;
}
jobid = job->jobid;
cbuf_put(&(core->jbuf), &jobid);
}
jobid = 0;
cbuf_put(&(core->jbuf), &jobid);
job_picker->picker_iface->end_session(job_picker);
return NULL;
}
/* ---------------------------------------------------------------------- */
/* Stand-in allocator thread -- see file header comment for why this must */
/* be a real, separate thread here (unlike test_core_full_cycle.c, whose */
/* test function itself can play this role sequentially). */
/* ---------------------------------------------------------------------- */
static void* stand_in_allocator_thread_fn(void* arg)
{
xce_sched_core_t* core = (xce_sched_core_t*)arg;
uint64_t jobid;
xce_decision_t* decision = NULL;
cbuf_get(&(core->jbuf), &jobid);
(void)jobid; /* always 0 here -- this fixture pushes no jobs */
log_event(EVT_ALLOCATOR_DRAINED_JBUF_SENTINEL);
cbuf_put(&(core->decisions), &decision);
log_event(EVT_ALLOCATOR_PUSHED_END_DECISION);
return NULL;
}
/* ---------------------------------------------------------------------- */
/* Driver-side decision callback spy -- must never actually fire in this */
/* test (this fixture produces no DISPATCH/PEND decisions), asserted below.*/
/* ---------------------------------------------------------------------- */
typedef struct {
int calls;
} 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;
(void)decision;
ctx->calls++;
return 0;
}
/* ---------------------------------------------------------------------- */
/* The contract test. */
/* ---------------------------------------------------------------------- */
/*
* test_round_runs_plugin_begin_and_account_jobs_before_picker_signaled -
* drives the REAL xce_sched_driver_run_normal_mode_round() through one
* complete round against a real picker thread + a real stand-in allocator
* thread, and proves (see file header comment for the full design):
*
* 1. plugin_begin_session() (observer) and sched_account_jobs() (spy)
* both actually fire, exactly once each.
* 2. Both complete strictly BEFORE the picker thread's begin_session
* hook runs -- i.e. strictly before xce_core_signal_go()'s
* cbuf_put(&core->wbuf) is observed by the picker thread's
* cbuf_get(&wbuf). This is the central safety property (Req 1 AC 7 /
* Req 5 AC 9): "called in the right window", not just "eventually
* called".
* 3. plugin_end_session() (observer) fires exactly once, after the
* round's decision-drain loop genuinely receives the end-of-session
* marker (proven via the stand-in allocator's own cbuf_put ->
* cbuf_get cross-thread edge).
* 4. xce_core_end_session() genuinely clears session_active.
* 5. No decision ever reaches the registered on_decision callback (this
* fixture pushes no jobs -- a sanity check that the ordering proof
* above isn't accidentally relying on unrelated activity).
*/
static void
test_round_runs_plugin_begin_and_account_jobs_before_picker_signaled(void)
{
xce_scheduler_t sched;
pthread_t picker_tid;
pthread_t allocator_tid;
test_driver_ctx_t driver_ctx;
int rc;
int idx_observer_begin;
int idx_account_jobs;
int idx_picker_begin;
int idx_picker_next_null;
int idx_picker_end;
int idx_allocator_sentinel;
int idx_allocator_pushed_end;
int idx_observer_end;
memset(&g_events, 0, sizeof(g_events));
memset(&sched, 0, sizeof(sched));
memset(&driver_ctx, 0, sizeof(driver_ctx));
atomic_store(&g_account_jobs_calls, 0);
atomic_store(&g_observer_begin_calls, 0);
atomic_store(&g_observer_end_calls, 0);
rc = xce_sched_core_init(&(sched.core));
TEST_ASSERT_EQUAL_INT(0, rc);
/* Register both plugins directly into core->plugins (bypassing
* plugin_register() -- same convention test_core_full_cycle.c's
* fixture_init() uses: this test does not need plugin_register()'s
* own init_fn/legacy-wrapper logic, only the registered-plugin
* iteration plugin_begin_session()/plugin_end_session()/
* sched_resolve_job_picker() perform). Must run AFTER
* xce_sched_core_init() (which memsets the whole core, including
* job_pick_plugin, back to zero). */
lock(sched.core.plugins, RTTI_LOCK_WRITE);
rc = key_add(sched.core.plugins, (void*)g_observer_plugin.name,
(void*)&g_observer_plugin);
TEST_ASSERT_GREATER_OR_EQUAL(0, rc);
rc = key_add(sched.core.plugins, (void*)g_stub_picker_plugin.name,
(void*)&g_stub_picker_plugin);
TEST_ASSERT_GREATER_OR_EQUAL(0, rc);
unlock(sched.core.plugins);
sched.core.job_pick_plugin = g_stub_picker_plugin.name;
rc = xce_core_set_decision_cb(&(sched.core), test_on_decision, &driver_ctx);
TEST_ASSERT_EQUAL_INT(0, rc);
rc = pthread_create(&picker_tid, NULL, picker_thread_fn, &(sched.core));
TEST_ASSERT_EQUAL_INT(0, rc);
rc = pthread_create(&allocator_tid, NULL, stand_in_allocator_thread_fn,
&(sched.core));
TEST_ASSERT_EQUAL_INT(0, rc);
/* The function under test: the REAL production
* xce_sched_driver_run_normal_mode_round(), driven exactly the way
* main.c's while(1) loop drives it. Blocks until the round completes
* (the stand-in allocator thread above pushes the end-of-session NULL
* decision, unblocking this function's own decision-drain loop). */
xce_sched_driver_run_normal_mode_round(&sched);
rc = pthread_join(picker_tid, NULL);
TEST_ASSERT_EQUAL_INT(0, rc);
rc = pthread_join(allocator_tid, NULL);
TEST_ASSERT_EQUAL_INT(0, rc);
/* --- plugin_begin_session()/sched_account_jobs()/plugin_end_session()
* genuinely ran, exactly once each --- */
TEST_ASSERT_EQUAL_INT(1, atomic_load(&g_observer_begin_calls));
TEST_ASSERT_EQUAL_INT(1, atomic_load(&g_account_jobs_calls));
TEST_ASSERT_EQUAL_INT(1, atomic_load(&g_observer_end_calls));
/* --- No decision ever reached the driver callback --- */
TEST_ASSERT_EQUAL_INT(0, driver_ctx.calls);
/* --- xce_core_end_session() genuinely cleared session_active --- */
TEST_ASSERT_EQUAL_INT(0, atomic_load_explicit(&(sched.core.session_active),
memory_order_acquire));
/* --- Ordering, from the event log --- */
idx_observer_begin = find_log_index(EVT_OBSERVER_BEGIN_SESSION);
idx_account_jobs = find_log_index(EVT_ACCOUNT_JOBS);
idx_picker_begin = find_log_index(EVT_PICKER_BEGIN_SESSION);
idx_picker_next_null = find_log_index(EVT_PICKER_NEXT_JOB_NULL);
idx_picker_end = find_log_index(EVT_PICKER_END_SESSION);
idx_allocator_sentinel
= find_log_index(EVT_ALLOCATOR_DRAINED_JBUF_SENTINEL);
idx_allocator_pushed_end
= find_log_index(EVT_ALLOCATOR_PUSHED_END_DECISION);
idx_observer_end = find_log_index(EVT_OBSERVER_END_SESSION);
TEST_ASSERT_TRUE(idx_observer_begin >= 0);
TEST_ASSERT_TRUE(idx_account_jobs >= 0);
TEST_ASSERT_TRUE(idx_picker_begin >= 0);
TEST_ASSERT_TRUE(idx_picker_next_null >= 0);
TEST_ASSERT_TRUE(idx_picker_end >= 0);
TEST_ASSERT_TRUE(idx_allocator_sentinel >= 0);
TEST_ASSERT_TRUE(idx_allocator_pushed_end >= 0);
TEST_ASSERT_TRUE(idx_observer_end >= 0);
/* THE central safety-property assertion (Req 1 AC 7 / Req 5 AC 9): both
* plugin_begin_session() and sched_account_jobs() complete strictly
* BEFORE the picker thread's begin_session runs -- proven via the wbuf
* cross-thread handoff (see file header comment). Not "eventually
* called" -- "called in the right window". */
TEST_ASSERT_TRUE(idx_observer_begin < idx_picker_begin);
TEST_ASSERT_TRUE(idx_account_jobs < idx_picker_begin);
/* Documents the actual call order xce_sched_driver_run_normal_mode_round()
* uses today (plugin_begin_session() then sched_account_jobs()) -- per
* driver.c's own header comment this relative order is NOT itself
* safety-critical (neither has a data dependency on the other), but it
* is what genuinely happens, so it is asserted here too. */
TEST_ASSERT_TRUE(idx_observer_begin < idx_account_jobs);
/* plugin_end_session() runs strictly AFTER the round's decision-drain
* loop genuinely observed the end-of-session marker -- proven via the
* allocator thread's cbuf_put(&core->decisions)/main thread's
* cbuf_get(&core->decisions) cross-thread edge, the same style of proof
* as the wbuf handoff above. */
TEST_ASSERT_TRUE(idx_allocator_pushed_end < idx_observer_end);
}
/*
* test_null_driver_ctx_is_safe_noop - xce_sched_driver_run_normal_mode_round()
* guards `driver_ctx == NULL` before touching anything (driver.c's own
* `if (sched == NULL) { return; }`) -- must not crash, must not call any
* of the observer/spy hooks above.
*/
static void test_null_driver_ctx_is_safe_noop(void)
{
memset(&g_events, 0, sizeof(g_events));
atomic_store(&g_account_jobs_calls, 0);
atomic_store(&g_observer_begin_calls, 0);
atomic_store(&g_observer_end_calls, 0);
xce_sched_driver_run_normal_mode_round(NULL);
TEST_ASSERT_EQUAL_INT(0, atomic_load(&g_observer_begin_calls));
TEST_ASSERT_EQUAL_INT(0, atomic_load(&g_account_jobs_calls));
TEST_ASSERT_EQUAL_INT(0, atomic_load(&g_observer_end_calls));
}
int main(void)
{
UNITY_BEGIN();
RUN_TEST(
test_round_runs_plugin_begin_and_account_jobs_before_picker_signaled);
RUN_TEST(test_null_driver_ctx_is_safe_noop);
return UNITY_END();
}

69
src/scheduler/testsuite/test_driver_translate.c
View file

@ -67,6 +67,75 @@ 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; }
/* Link-satisfying stubs for scheduler.c's sched_publish_dispatch()/
* sched_publish_pend_reason() -- as of scheduler-core spec task 5.2,
* driver.c's (real, #included above) xce_sched_driver_on_decision() calls
* both, but this file never exercises that function (it is out of THIS
* file's scope -- see test_driver_decision_cb.c) and never links
* scheduler.c (same rationale as the sched_resolve_job_picker
* reimplementation below: this file's own fixtures never need a real
* xcecli_ctx_t/apictx). Never actually invoked at runtime by any test in
* this file. */
int sched_publish_dispatch(xce_scheduler_t* sched, xce_job_dispatch_t* d)
{
(void)sched;
(void)d;
return 0;
}
int sched_publish_pend_reason(xce_scheduler_t* sched, xce_preason_set_t* r)
{
(void)sched;
(void)r;
return 0;
}
/* Link-satisfying stubs: as of scheduler-core spec task 5.2,
* xce_sched_driver_run_normal_mode_round() (driver.c, real, #included
* above, alongside xce_sched_driver_translate_pending_events() this file
* DOES exercise) references xce_core_begin_session()/xce_core_signal_go()/
* xce_core_end_session() (core_session.c) and plugin_begin_session()/
* plugin_end_session() (plugin.c) / sched_account_jobs() (scheduler.c).
* This file never exercises xce_sched_driver_run_normal_mode_round() itself
* (out of THIS file's scope -- see test_driver_session_ordering.c) and
* never links core_session.c/plugin.c/scheduler.c. Never actually invoked
* at runtime by any test in this file. */
int xce_core_begin_session(xce_sched_core_t* core)
{
(void)core;
return 0;
}
int xce_core_signal_go(xce_sched_core_t* core)
{
(void)core;
return 0;
}
int xce_core_end_session(xce_sched_core_t* core)
{
(void)core;
return 0;
}
int plugin_begin_session(void* arg)
{
(void)arg;
return 0;
}
int plugin_end_session(void* arg)
{
(void)arg;
return 0;
}
int sched_account_jobs(xce_scheduler_t* sched)
{
(void)sched;
return 0;
}
/* Real reimplementation of schedobj.c's sched_resolve_job_picker -- see
* file header comment (same rationale as test_core_events.c's own copy). */
int sched_resolve_job_picker(xce_sched_core_t* core, xce_sched_plugin_t** out)