From e6c0a2448a60b6d3c028686407b5c48ea6e604d2 Mon Sep 17 00:00:00 2001 From: Rongsong Shen Date: Mon, 3 Aug 2026 16:05:33 +0000 Subject: [PATCH] feat(scheduler-core): implement core event batch push interface (task 4.3) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Defines xce_core_event_type_t/xce_core_event_t (9 event kinds: config, job, worker, resource, resval full/delta, job state change, job update, queue update, job rusage) in core.h. Implements xce_core_push_events() in a new core_events.c (split out of core.c to stay under the 800-line cap): rejects -EBUSY mid-session, otherwise applies each event's core- state mutation and fires the corresponding picker hook synchronously, exactly once per event, before returning. CONFIG_PUSH decomposes the pushed config's embedded queues/resources/resvals via shared helpers (also reused by QUEUE_UPDATE's upsert path), mirroring the existing sched_init_config()/sched_init_resources()/sched_init_sharedInst()/ sched_init_queues(). Narrows sched_resolve_job_picker() to take xce_sched_core_t* (pure type change, zero behavior change). Purely additive — not wired into the live scheduling path yet (task 5.x). Co-Authored-By: Claude Sonnet 5 Claude-Session: https://claude.ai/code/session_019dw5WnN3DXStaEaPYfACU1 --- .kiro/specs/scheduler-core/tasks.md | 4 +- src/scheduler/include/core.h | 119 +++ src/scheduler/include/scheduler.h | 20 +- src/scheduler/meson.build | 1 + src/scheduler/src/core_events.c | 716 +++++++++++++++++++ src/scheduler/src/schedobj.c | 39 +- src/scheduler/src/scheduler.c | 6 +- src/scheduler/testsuite/meson.build | 19 + src/scheduler/testsuite/test_core_events.c | 714 ++++++++++++++++++ src/scheduler/testsuite/test_picker_wiring.c | 18 +- 10 files changed, 1613 insertions(+), 43 deletions(-) create mode 100644 src/scheduler/src/core_events.c create mode 100644 src/scheduler/testsuite/test_core_events.c diff --git a/.kiro/specs/scheduler-core/tasks.md b/.kiro/specs/scheduler-core/tasks.md index 1fa2d290..5d42e2f0 100644 --- a/.kiro/specs/scheduler-core/tasks.md +++ b/.kiro/specs/scheduler-core/tasks.md @@ -99,12 +99,13 @@ - _Verified: 2026-08-03 — jbuf capacity hardcoded to literal 1 (core.c), regression test test_core_jbuf_capacity_pin.c; kvmgr audit found jobs/queues/users/projects already fully order-stable (xzarray ascending-slot walk, no process-random seeding), no fix needed, findings recorded durably as a comment in core.c above xce_sched_core_init(); 15/15 scheduler suite passing; two-round review (round 1 REJECTED on missing durable record only, round 2 APPROVED after fix); core->workers flagged as an open follow-up (see Implementation Notes), not this task's scope_ - _Depends: 4.1_ -- [ ] 4.3 Implement the core event interface and batch event push +- [x] 4.3 Implement the core event interface and batch event push - Define the typed core event representation (timestamp plus type-specific payload) covering configuration push, job push, worker push, resource push, resource-value full/delta push, job state change, job update, queue update, and job resource-usage events - Implement batch event push: state-change picker hooks fire synchronously for each event in the batch, valid only when no scheduling session is active - Observable: pushing a batch of events with the picker idle fires the corresponding state-change hook exactly once per event, synchronously, before the push call returns - _Requirements: 5.4, 5.8_ - _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 - Implement the non-blocking session-start call that signals the picker thread and returns immediately, rejecting a call while a session is already active @@ -363,3 +364,4 @@ - Task 2.4 closed the above race with `fcfs_plugin_data_t.lock` (mylock_t), taken by exactly 7 hooks (`on_queue_update`, `on_job_push`, `on_job_state_change`, `on_job_update`, `begin_session`, `next_job`, `on_dispatch`) — `on_pend`/`end_session`/`init`/`fini` correctly excluded (they never touch `priv->queues`/`cursor`/`job_cursor`), a verified refinement of this note's original "all five hooks" estimate, not a gap. `sched_job_picker()` now branches `job_elect_fn != NULL` (legacy plugins like `xce_sched_fcfs`, delegates unchanged to their own self-contained loop) vs NULL (native `xce_job_picker_iface_t` pickers, runs a new generic per-session loop) — this split exists because legacy `job_elect_fn` implementations (e.g. `fcfs_elect_job`) already contain their own internal `cbuf_get(&wbuf)`, so a single uniform outer loop would double-consume the wbuf signal for legacy pickers. `sched_allocator()` now calls `picker_iface->on_dispatch()`/`on_pend()` after each allocation attempt (Req 1 AC 4), which was previously called for no picker at all — required for builtin_fcfs to actually remove dispatched jobs from its private state once live. - Two follow-ups surfaced by task 2.4's review, not yet fixed (outside its file boundary): (1) `main.c` discards `sched_init()`'s return value, so a bad `job_pick_plugin` config fails `sched_init()` correctly but the process doesn't yet halt on it — needs a fast follow-up before relying on "fails startup" in production. (2) `pthread_testcancel()` in `builtin_fcfs_next_job()` is only at the loop entry/exit, not scattered through the queue/job walk (the walk holds `fcfs_plugin_data_t.lock` throughout, and this codebase has no `pthread_cleanup_push` precedent to safely unlock on mid-walk cancellation) — behaviorally inert today since `sched_queue_is_ready`/`sched_job_is_ready` are `return 1` stubs, but becomes a real cancellation-latency gap once Phase 2 implements real filtering. - Task 4.2's Req 5 AC 7 kvmgr iteration-order audit (scope: jobs/queues/users/projects, per design.md's bounded-scope note) found no unstable site — recorded durably as a comment in `core.c` above `xce_sched_core_init()`. One open gap surfaced but explicitly out of this task's scope: `core->workers` is iterated order-sensitively by `sched_collect_workers()` → `seqjob_allocator.c`'s first-fit `sched_try_alloc_on_hosts()`, and design.md itself says the full AC 7 guarantee must close "before Phase 3 simulation work depends on it" — needs a follow-up audit/fix (or an explicit documented exemption) before Phase 3 (tasks 7+) starts relying on scheduling determinism across worker selection. +- Task 4.3 split its implementation into a NEW file `core_events.c` (not `core.c`, which design.md's File Structure Plan literally names) because implementing `xce_core_push_events()` directly in `core.c` would have pushed it to 977 lines, over this project's 800-line hard cap. `core.h` (the public contract) matches design.md's intent; only the `.c` file location differs. Worth reflecting in design.md next time it's touched. Also: `test_core_isolation.c`'s nm(1)-based zero-controller-I/O boundary audit (Req 5 AC 3) covers only `core.c`, not `core_events.c` — manually verified clean (no netio/xceapi/client/delay/idgen usage) but not mechanically enforced; extending that test to cover `core_events.c` would be a reasonable low-risk follow-up before later tasks add more logic to this file. diff --git a/src/scheduler/include/core.h b/src/scheduler/include/core.h index 3369be67..96f49194 100644 --- a/src/scheduler/include/core.h +++ b/src/scheduler/include/core.h @@ -32,6 +32,7 @@ #include "dataq.h" #include "mylock.h" +#include "plugin.h" #include "sched/types.h" #include "xce/types.h" #include "zarray.h" @@ -114,4 +115,122 @@ extern int sched_add_project(xce_sched_core_t*, xce_sched_project_t*); extern int sched_job_is_ready(xce_sched_job_t*); extern int sched_queue_is_ready(xce_sched_queue_t*); +/* + * sched_resolve_job_picker - Req 1 AC 6 / Req 2 AC 5 (scheduler-core spec, + * task 2.4). Resolves core->job_pick_plugin against the plugin registry + * (core->plugins, the same plugin_register()/key_find() mechanism every + * other plugin category uses) and validates the result actually implements + * the picker MUST hooks (xce_job_picker_iface_t: next_job, begin_session, + * end_session, on_dispatch, on_pend). Returns 0 and sets *out on success; a + * negative errno on failure: + * -EINVAL bad arguments, or core->job_pick_plugin is unset (NULL), or + * the name IS registered but is not a usable picker + * -ENOENT the configured name is not registered at all + * Implemented in schedobj.c. Declared here (task 4.3, scheduler-core spec) + * -- moved from scheduler.h's xce_scheduler_t* signature to this + * xce_sched_core_t* signature so xce_core_push_events() (core.c, below) can + * call it directly without core.c depending on scheduler.h (forbidden -- + * see this header's own boundary comment). Every existing call site already + * held an xce_scheduler_t* and now passes &sched->core instead; no + * resolution logic is duplicated. + */ +extern int sched_resolve_job_picker(xce_sched_core_t*, xce_sched_plugin_t**); + +/* + * xce_core_event_type_t / xce_core_event_t - the core's typed input-event + * representation (scheduler-core spec, task 4.3, Req 5 AC 4). Exactly the 9 + * event kinds task 4.3 scopes in: BEGIN_ROUND and END_SESSION belong to + * task 4.4's session-start signal, deliberately not added here. + * + * Each event carries a timestamp (`ts`, nanoseconds since epoch) and a + * type-specific payload (Req 5 AC 4's field tables). Six of the nine kinds + * additionally drive a picker state-change hook (xce_core_push_events()'s + * own doc comment, core.c, maps each type to its hook one-for-one); + * WORKERS_PUSH/RESOURCES_PUSH/RESVALS_FULL/RESVALS_DELTA are state-update + * only -- "picking is about job ordering, not resource availability" + * (plugin.h's own Req 1 AC 2 comment). + */ +typedef enum { + XCE_CORE_EVENT_CONFIG_PUSH = 1, + XCE_CORE_EVENT_JOBS_PUSH, + XCE_CORE_EVENT_WORKERS_PUSH, + XCE_CORE_EVENT_RESOURCES_PUSH, + XCE_CORE_EVENT_RESVALS_FULL, + XCE_CORE_EVENT_RESVALS_DELTA, + XCE_CORE_EVENT_JOB_STATE_CHANGE, + XCE_CORE_EVENT_JOB_UPDATE, + XCE_CORE_EVENT_QUEUE_UPDATE, + XCE_CORE_EVENT_JOB_RUSAGE, +} xce_core_event_type_t; + +typedef struct { + xce_core_event_type_t type; + uint64_t ts; /* nanoseconds since epoch */ + union { + struct { + xce_conf4sched_t* config; + } config_push; + struct { + int njobs; + xce_job_t** jobs; + } jobs_push; + struct { + int nworkers; + xce_worker_t** workers; + } workers_push; + struct { + int n; + xce_resource_t** resources; + } resources_push; + struct { + int n; + xce_resval_t** resvals; + } resvals_full; + struct { + xce_resval_list_t* deltas; + } resvals_delta; + struct { + uint64_t jobid; + uint32_t old_state; + uint32_t new_state; + int32_t exit_code; + } job_state_change; + struct { + uint64_t jobid; + uint32_t changed_fields; + xce_job_t* job; + } job_update; + struct { + xce_queue_t* queue; + } queue_update; + struct { + uint64_t jobid; + xce_job_rusages_t* rusages; + } job_rusage; + } u; +} xce_core_event_t; + +/* + * xce_core_push_events - Req 5 AC 4 / AC 8 (scheduler-core spec, task 4.3; + * design.md "Driver -> Core Protocol", "Core Interface Contract"). Batch- + * pushes `nevents` events, applying each one's core-state mutation in + * order and -- for the six event kinds that have one -- synchronously + * calling the resolved picker's corresponding state-change hook exactly + * once per event, in the calling (driver) thread, before this call + * returns. See core.c for the full per-event-type contract and the + * type-to-hook mapping. + * + * Only valid between scheduling sessions (core->session_active == 0): + * this is what makes calling picker hooks synchronously, from the driver + * thread, safe -- the picker thread is blocked on cbuf_get(&wbuf) and the + * allocator thread on cbuf_get(&jbuf), so no session is concurrently + * reading the state this call mutates. Returns -EBUSY (no event applied) + * if a session is active; -EINVAL for bad arguments (core NULL, or events + * NULL with nevents > 0). Per-event failures (e.g. JOB_UPDATE naming an + * unknown jobid) are logged and do not abort the rest of the batch -- + * same discipline scheduler.c's existing sched_update() dataq loop + * already follows. + */ +extern int xce_core_push_events(xce_sched_core_t*, xce_core_event_t*, int); + #endif diff --git a/src/scheduler/include/scheduler.h b/src/scheduler/include/scheduler.h index 0842f019..077e06ef 100644 --- a/src/scheduler/include/scheduler.h +++ b/src/scheduler/include/scheduler.h @@ -112,21 +112,11 @@ extern int sched_init_config(xce_scheduler_t* sched, xce_conf4sched_t*); * default to sched->job_pick_plugin when unset. See the field comment on * xce_scheduler_t above -- this does NOT resolve/register the name. */ extern void sched_set_default_job_pick_plugin(xce_scheduler_t*); -/* Req 1 AC 6 / Req 2 AC 5 (scheduler-core spec, task 2.4): resolves - * sched->job_pick_plugin against the plugin registry (the same - * plugin_register()/key_find() mechanism every other plugin category - * uses) and validates the result actually implements the picker MUST - * hooks (xce_job_picker_iface_t: next_job, begin_session, end_session, - * on_dispatch, on_pend). Returns 0 and sets *out on success; a negative - * errno on failure: - * -EINVAL bad arguments, or sched->job_pick_plugin is unset (NULL), - * or the name IS registered but is not a usable picker - * -ENOENT the configured name is not registered at all - * Called from sched_init() (before the picker thread is created, so an - * unregistered/non-picker name fails scheduler startup synchronously) and - * from sched_job_picker() (the picker thread, at thread start) -- one - * implementation, two call sites, no duplicated lookup logic. */ -extern int sched_resolve_job_picker(xce_scheduler_t*, xce_sched_plugin_t**); +/* sched_resolve_job_picker relocated to core.h (scheduler-core spec, task + * 4.3) -- it now operates on xce_sched_core_t*, not xce_scheduler_t*, so + * xce_core_push_events() (core.c) can call it without core.c depending on + * this header. Call sites in this file (sched_job_picker(), sched_allocator(), + * sched_init()) now pass &sched->core. See core.h for the full contract. */ extern int sched_update(xce_scheduler_t*); extern void sched_go(xce_scheduler_t*); extern void sched_begin_session(xce_scheduler_t*); diff --git a/src/scheduler/meson.build b/src/scheduler/meson.build index a0633fe6..6c185889 100644 --- a/src/scheduler/meson.build +++ b/src/scheduler/meson.build @@ -3,6 +3,7 @@ scheduler_srcs = [ 'src/handler.c', 'src/scheduler.c', 'src/core.c', + 'src/core_events.c', 'src/decision.c', 'src/schedobj.c', 'src/plugin.c', diff --git a/src/scheduler/src/core_events.c b/src/scheduler/src/core_events.c new file mode 100644 index 00000000..88cf83ff --- /dev/null +++ b/src/scheduler/src/core_events.c @@ -0,0 +1,716 @@ +#include +#include +#include + +#include "controller.capnp.h" +#include "core.h" +#include "misc.h" +#include "resobj.h" +#include "sched/types.h" +#include "utils.h" +#include "xcealloc.h" +#include "xstring.h" + +/* + * core_events.c - the core event interface and batch event push + * (scheduler-core spec, task 4.3, Req 5 AC 4 / AC 8). Split out of core.c + * (rather than appended there, as design.md's File Structure Plan + * originally sketched) purely to respect this project's 800-line-per-file + * hard cap -- core.c was already 407 lines before this task, and this + * feature's full state-mutation + hook-dispatch contract does not fit + * under that cap in the same file. Same boundary as core.c/core.h + * (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, and + * must never call a wall-clock function directly. (Not yet covered by + * test_core_isolation.c's automated nm(1) audit, which today only inspects + * core.c's own compiled object -- extending that audit to this file is a + * reasonable follow-up, out of this task's scope.) + * + * See core.h for xce_core_event_type_t / xce_core_event_t / + * xce_core_push_events()'s own doc comments (the public contract). + */ + +/* + * core_build_job/core_build_queue/core_build_worker - construct core-owned + * scheduling objects from the wire (xce_job_t/xce_queue_t/xce_worker_t) + * payloads carried by JOBS_PUSH/QUEUE_UPDATE/WORKERS_PUSH events. + * + * schedobj.c's make_sched_job()/make_sched_queue()/make_sched_worker() + * already implement this exact field-copy algorithm, but their `void* arg` + * is cast internally to xce_scheduler_t* -- even though their bodies only + * ever touch arg->core.* (never arg->apictx/arg->cluster_cfg), calling them + * with a bare xce_sched_core_t* that is not actually embedded inside a real + * xce_scheduler_t (as this file's own tests build) would rely on treating + * a foreign object as if it had xce_scheduler_t's layout -- undefined + * behavior this file will not introduce, on top of which this file must + * never depend on scheduler.h (this file's own boundary comment) to even + * name xce_scheduler_t in the first place. The three helpers below are the + * same algorithm, parameterized on xce_sched_core_t* directly -- reusing + * sched_get_user()/sched_get_queue()/sched_get_project()/parse_resreq() + * verbatim, not reinventing them. + * + * core_build_job() intentionally does not construct job->schedobj (the + * resreq-tag-dispatch allocation-context object make_sched_job() builds via + * its own make_sched_obj() helper): no picker hook reads job->schedobj + * (verified against every builtin_fcfs.c hook body), and schedobj/ + * allocation-context construction belongs to the allocation path + * (scheduler-allocation spec), out of this task's event-interface scope. + */ +static xce_sched_job_t* core_build_job(xce_sched_core_t* core, xce_job_t* j) +{ + xce_sched_job_t* job; + char* errmsg = NULL; + int rc; + + if ((core == NULL) || (j == NULL) || (j->submit == NULL)) { + return NULL; + } + + job = (xce_sched_job_t*)calloc(1, sizeof(*job)); + if (job == NULL) { + return NULL; + } + + job->jobid = j->jobid; + job->sub_time = j->sub_time; + job->queue_time = j->queue_time; + job->start_time = j->start_time; + job->state = j->state; + job->user = sched_get_user(core, j->submit->user); + job->queue = sched_get_queue(core, j->submit->queue); + job->project = sched_get_project(core, j->submit->project_name); + + job->resreq = (xce_resreq_t*)calloc(1, sizeof(*(job->resreq))); + if (job->resreq == NULL) { + FREEUP(job); + return NULL; + } + + { + char* resreq_str = j->submit->resreq; + + if (NULLSTR(resreq_str)) { + resreq_str = "type == any"; + } + rc = parse_resreq(job->resreq, resreq_str, core->resmgr, &errmsg); + } + + if (rc < 0) { + ERROR("%s parse_resreq failed for jobid %llu: %s", __func__, + (unsigned long long)j->jobid, errmsg ? errmsg : "(null)"); + FREEUP(errmsg); + FREEUP(job->resreq); + FREEUP(job); + return NULL; + } + + job->mintasks = j->submit->mintasks; + job->maxtasks = j->submit->maxtasks; + job->priority = 0; /* FIXME: support job level priority (mirrors + * make_sched_job's own FIXME, schedobj.c) */ + + if (j->allocs != NULL) { + MOVE_PTR(job->allocs, j->allocs); + job->accounts = make_job_account(job->allocs); + } + + return job; +} + +static xce_sched_queue_t* core_build_queue(xce_queue_t* q) +{ + xce_sched_queue_t* queue; + + if (q == NULL) { + return NULL; + } + + queue = (xce_sched_queue_t*)calloc(1, sizeof(*queue)); + if (queue == NULL) { + return NULL; + } + + queue->name = makeString(q->queue); + queue->attrs = q->attrs; + queue->status = q->status; + queue->priority = q->priority; + queue->accept_interval = q->accept_interval; + queue->sched_delay = q->sched_delay; + /* Same scope note as make_sched_queue's own (schedobj.c): project_name/ + * job_pick_plugin have no wire-config source yet, and (matching + * existing behavior) resreq is not populated here either -- wiring + * these is a later task, not a regression introduced by this one. */ + queue->project_name = NULL; + queue->job_pick_plugin = NULL; + + return queue; +} + +/* + * core_upsert_queue - shared add-or-update logic for one xce_queue_t wire + * payload, used by both the QUEUE_UPDATE event and CONFIG_PUSH's embedded + * config->queues[] array (Req 5 AC 4: CONFIG_PUSH's queues are documented + * to behave the same as individual queue updates, just batched). Looks up + * `updated->queue` in core->queues; if found, updates the existing + * xce_sched_queue_t in place (mirrors schedobj.c's + * sched_update_existing_queue()); otherwise builds+adds a new one via + * core_build_queue()/sched_add_queue() (mirrors scheduler.c's + * sched_init_queues()). Fires on_queue_update(picker, q) once if a picker + * is resolved, matching the QUEUE_UPDATE row's hook contract exactly. + * No-ops (WARN-logged) on a malformed payload or a failed build. + */ +static void core_upsert_queue(xce_sched_core_t* core, xce_sched_plugin_t* picker, + xce_queue_t* updated) +{ + xce_sched_queue_t* q; + + if ((updated == NULL) || (updated->queue == NULL)) { + WARN("%s NULL queue payload", __func__); + return; + } + + q = sched_get_queue(core, updated->queue); + if (q == NULL) { + q = core_build_queue(updated); + if (q == NULL) { + WARN("%s failed to build new queue %s", __func__, + updated->queue); + return; + } + sched_add_queue(core, q); + } else { + q->status = updated->status; + q->priority = updated->priority; + q->accept_interval = updated->accept_interval; + q->sched_delay = updated->sched_delay; + + if (updated->resreq != NULL) { + FREEUP(q->resreq); + q->resreq = makeString(updated->resreq); + } + } + + if (picker != NULL) { + picker->picker_iface->on_queue_update(picker, q); + } +} + +/* + * core_apply_config_resources/core_apply_config_resvals/ + * core_apply_config_queues - CONFIG_PUSH's decomposition of a pushed + * xce_conf4sched_t's embedded queues[]/resources[]/resvals[] arrays into + * core state (Req 5 AC 4's CONFIG_PUSH row: "All queues, resource + * thresholds, ..." is part of the payload, and decomposing it into + * core->resmgr/core->sharedInst/core->queues is the core's job, not the + * driver's, since this IS the core-side entry point). These three mirror + * scheduler.c's sched_init_resources()/sched_init_sharedInst()/ + * sched_init_queues() exactly (same algorithm, parameterized on + * xce_sched_core_t* directly -- same boundary rationale as + * core_build_job/core_build_queue/core_build_worker above), which is what + * sched_init_config() calls today for this exact same xce_conf4sched_t + * payload on the still-live scheduler.c path. core_apply_config_queues() + * reuses core_upsert_queue() rather than duplicating queue add-or-update + * logic a third time. + */ +static void core_apply_config_resources(xce_sched_core_t* core, + xce_conf4sched_t* config) +{ + int i; + + for (i = 0; i < config->n_resources; i++) { + xce_resource_t* r = config->resources[i]; + xce_resource_t* r1; + + if (r == NULL) { + continue; + } + + r1 = makeResource(core->resmgr, r); + if (r1 == NULL) { + ERROR("%s add resource %s failed", __func__, + r->name ? r->name : "(null)"); + } + } +} + +static void core_apply_config_resvals(xce_sched_core_t* core, + xce_conf4sched_t* config) +{ + int i; + + for (i = 0; i < config->nresvals; i++) { + xce_name_val_t* nv = config->resvals[i]; + xce_resource_t* r = NULL; + xce_val_t* val = NULL; + + if ((nv == NULL) || (nv->name == NULL)) { + continue; + } + + if ((key_find(core->resmgr, (void*)nv->name, (void**)&r) < 0) + || (r == NULL)) { + WARN("%s unknown resource '%s' in resvals, skipping", __func__, + nv->name); + continue; + } + + MOVE_PTR(val, nv->val); + zarray_put(&(core->sharedInst), r->idx, val); + } +} + +static void core_apply_config_queues(xce_sched_core_t* core, + xce_sched_plugin_t* picker, + xce_conf4sched_t* config) +{ + int i; + + for (i = 0; i < config->n_queues; i++) { + if (config->queues[i] == NULL) { + continue; + } + + core_upsert_queue(core, picker, config->queues[i]); + } +} + +static xce_sched_worker_t* core_build_worker(xce_sched_core_t* core, + xce_worker_t* w) +{ + xce_sched_worker_t* worker; + int i; + + if ((core == NULL) || (w == NULL)) { + return NULL; + } + + worker = (xce_sched_worker_t*)calloc(1, sizeof(*worker)); + if (worker == NULL) { + return NULL; + } + + worker->name = makeString(w->name); + worker->status = w->status; + worker->idx = -1; + + zarray_init(&(worker->resvals), 128, + (zarray_item_killer_t)kill_sched_resval); + zarray_init(&(worker->resInUse), 128, + (zarray_item_killer_t)kill_sched_resval); + + for (i = 0; i < w->nresvals_to_sched; i++) { + xce_name_val_t* nv = w->resvals_to_sched[i]; + xce_resource_t* r = NULL; + xce_val_t* val = NULL; + + if ((nv == NULL) || (nv->name == NULL)) { + continue; + } + + if ((key_find(core->resmgr, (void*)nv->name, (void**)&r) < 0) + || (r == NULL)) { + WARN("%s unknown resource '%s' in worker resvals, skipping", + __func__, nv->name); + continue; + } + + MOVE_PTR(val, nv->val); + zarray_put(&(worker->resvals), r->idx, val); + } + + return worker; +} + +/* + * xce_core_push_events - see core.h's doc comment for the full contract + * (Req 5 AC 4 / AC 8). Per event type: + * + * CONFIG_PUSH core->config = event's config; configver++; then + * decomposes the pushed config's embedded queues[]/ + * resources[]/resvals[] arrays into core->queues/ + * core->resmgr/core->sharedInst via + * core_apply_config_resources()/ + * core_apply_config_resvals()/ + * core_apply_config_queues() (mirrors scheduler.c's + * sched_init_resources()/sched_init_sharedInst()/ + * sched_init_queues(), the same decomposition + * sched_init_config() already performs for this exact + * payload) -- firing on_queue_update(picker, queue) + * once per embedded queue (core_upsert_queue(), shared + * with QUEUE_UPDATE below), then + * on_config(picker, config) if the hook is non-NULL. + * JOBS_PUSH core_build_job() + sched_add_job() per job, then + * on_job_push(picker, job) once PER JOB in the batch. + * WORKERS_PUSH core_build_worker() + sched_add_worker() per worker. + * No picker hook (Req 1 AC 2: worker updates are not + * forwarded to the picker). + * RESOURCES_PUSH makeResource()/key_find()-update-in-place per + * resource (mirrors scheduler.c's sched_update() + * XCE_SCHED_OP_UPDATE_RESOURCES case exactly). No + * picker hook. + * RESVALS_FULL key_find() + zarray_put(&core->sharedInst, ...) per + * value (mirrors XCE_SCHED_OP_UPDATE_RESVALS_FULL). No + * picker hook. + * RESVALS_DELTA sched_get_worker() + key_find() + + * zarray_put(&worker->resvals, ...) per delta (mirrors + * XCE_SCHED_OP_UPDATE_RESVALS, the existing + * per-worker resval-update case). No picker hook. + * JOB_STATE_CHANGE sched_get_job() + free stale accounts/allocs if the + * new state left RUN + state assignment (mirrors + * scheduler.c's sched_job_state_change()), then + * on_job_state_change(picker, job, new_state). + * JOB_UPDATE sched_get_job() + queue/resreq field-copy (mirrors + * schedobj.c's sched_update_existing_job(), minus its + * plugin_job_update() fan-out call -- that is the OLD + * per-plugin hook category, unrelated to the picker + * interface this event drives), then + * on_job_update(picker, job). + * QUEUE_UPDATE core_upsert_queue(): sched_get_queue(); update in + * place if found, else core_build_queue() + + * sched_add_queue() (mirrors schedobj.c's + * sched_update_existing_queue() for the "modified" + * half, add-or-update per Req 5 AC 4's "added or + * modified"), then on_queue_update(picker, queue). + * JOB_RUSAGE sched_get_job() + rusages ownership transfer (mirrors + * schedobj.c's sched_apply_job_rusage()), then + * on_job_rusage(picker, job, rusages) IF the hook is + * non-NULL -- Req 5 AC 4: this hook is OPTIONAL, a + * NULL hook is skipped, never treated as an error. + * + * A single picker resolution (sched_resolve_job_picker()) is shared across + * the whole batch. If resolution fails, state mutation still proceeds for + * every event (adding a job to core->jobs, say, does not depend on a + * picker existing) but every hook call is skipped for this call, with a + * WARN -- sched_init() already gates on this exact resolution succeeding + * before either scheduling thread is created, so this should not occur in + * practice; handled defensively, same "should never occur" posture + * scheduler.c's sched_allocator() already documents for its own call to + * this same resolver. + * + * Per-event failures (unknown jobid, malformed payload, ...) are logged + * (WARN/ERROR) and do not abort the rest of the batch -- same discipline + * scheduler.c's existing sched_update() dataq loop already follows. + */ +int xce_core_push_events(xce_sched_core_t* core, xce_core_event_t* events, + int nevents) +{ + xce_sched_plugin_t* picker = NULL; + int rc; + int i; + + if ((core == NULL) || (nevents < 0) + || ((events == NULL) && (nevents > 0))) { + return -EINVAL; + } + + if (atomic_load(&(core->session_active)) != 0) { + return -EBUSY; + } + + rc = sched_resolve_job_picker(core, &picker); + if (rc < 0) { + WARN("%s failed to resolve job picker plugin (rc=%d); " + "state-change hooks will be skipped for this batch", + __func__, rc); + picker = NULL; + } + + for (i = 0; i < nevents; i++) { + xce_core_event_t* ev = &(events[i]); + + switch (ev->type) { + case XCE_CORE_EVENT_CONFIG_PUSH: { + xce_conf4sched_t* config = ev->u.config_push.config; + + core->config = config; + core->configver++; + + if (config != NULL) { + core_apply_config_resources(core, config); + core_apply_config_resvals(core, config); + core_apply_config_queues(core, picker, config); + } + + if ((picker != NULL) && (picker->picker_iface->on_config != NULL)) { + picker->picker_iface->on_config(picker, config); + } + break; + } + + case XCE_CORE_EVENT_JOBS_PUSH: { + int j; + + for (j = 0; j < ev->u.jobs_push.njobs; j++) { + xce_sched_job_t* job + = core_build_job(core, ev->u.jobs_push.jobs[j]); + + if (job == NULL) { + WARN("%s JOBS_PUSH: failed to build job (batch index %d)", + __func__, j); + continue; + } + + sched_add_job(core, job); + + if (picker != NULL) { + picker->picker_iface->on_job_push(picker, job); + } + } + break; + } + + case XCE_CORE_EVENT_WORKERS_PUSH: { + int w; + + for (w = 0; w < ev->u.workers_push.nworkers; w++) { + xce_sched_worker_t* worker + = core_build_worker(core, ev->u.workers_push.workers[w]); + + if (worker == NULL) { + WARN("%s WORKERS_PUSH: failed to build worker (batch " + "index %d)", + __func__, w); + continue; + } + + sched_add_worker(core, worker); + } + break; + } + + case XCE_CORE_EVENT_RESOURCES_PUSH: { + int r; + + for (r = 0; r < ev->u.resources_push.n; r++) { + xce_resource_t* res = ev->u.resources_push.resources[r]; + xce_resource_t* existing; + + if (res == NULL) { + continue; + } + + existing = makeResource(core->resmgr, res); + if (existing == NULL) { + /* EEXIST: update-in-place on the existing resource + * (add-or-update), mirrors sched_update()'s own + * XCE_SCHED_OP_UPDATE_RESOURCES case. */ + xce_resource_t* found = NULL; + + if ((key_find(core->resmgr, res->name, (void**)&found) + == 0) + && (found != NULL)) { + found->valtype = res->valtype; + found->ordtype = res->ordtype; + found->flags = res->flags; + found->interval = res->interval; + } else { + WARN("%s RESOURCES_PUSH: makeResource failed for " + "%s and lookup not found", + __func__, res->name); + } + } + } + break; + } + + case XCE_CORE_EVENT_RESVALS_FULL: { + int r; + + for (r = 0; r < ev->u.resvals_full.n; r++) { + xce_resval_t* rv = ev->u.resvals_full.resvals[r]; + xce_resource_t* res = NULL; + + if (rv == NULL) { + continue; + } + + if ((key_find(core->resmgr, rv->name, (void**)&res) < 0) + || (res == NULL)) { + WARN("%s RESVALS_FULL: unknown resource %s, skipping", + __func__, rv->name); + continue; + } + + zarray_put(&(core->sharedInst), res->idx, rv->val); + } + break; + } + + case XCE_CORE_EVENT_RESVALS_DELTA: { + xce_resval_list_t* deltas = ev->u.resvals_delta.deltas; + int r; + + if (deltas == NULL) { + break; + } + + for (r = 0; r < deltas->count; r++) { + xce_resval_t* rv = deltas->vals[r]; + xce_sched_worker_t* w = NULL; + xce_resource_t* res = NULL; + + if (rv == NULL) { + continue; + } + + w = sched_get_worker(core, rv->host); + if (w == NULL) { + WARN("%s RESVALS_DELTA: unknown worker %s, skipping", + __func__, rv->host); + continue; + } + + if ((key_find(core->resmgr, rv->name, (void**)&res) < 0) + || (res == NULL)) { + WARN("%s RESVALS_DELTA: unknown resource %s, skipping", + __func__, rv->name); + continue; + } + + zarray_put(&(w->resvals), res->idx, rv->val); + } + break; + } + + case XCE_CORE_EVENT_JOB_STATE_CHANGE: { + uint64_t jobid = ev->u.job_state_change.jobid; + uint32_t new_state = ev->u.job_state_change.new_state; + xce_sched_job_t* job = sched_get_job(core, jobid); + + if (job == NULL) { + WARN("%s JOB_STATE_CHANGE: unknown jobid %llu", __func__, + (unsigned long long)jobid); + break; + } + + if (!JOB_IS_RUN(new_state)) { + if (job->accounts != NULL) { + free_job_account(job->accounts); + job->accounts = NULL; + } + if (job->allocs != NULL) { + free_JobAlloc_ptr(&(job->allocs)); + } + } + + job->state = new_state; + + if (picker != NULL) { + picker->picker_iface->on_job_state_change(picker, job, + new_state); + } + break; + } + + case XCE_CORE_EVENT_JOB_UPDATE: { + xce_job_t* updated = ev->u.job_update.job; + xce_sched_job_t* job; + + if (updated == NULL) { + WARN("%s JOB_UPDATE: NULL job payload", __func__); + break; + } + + job = sched_get_job(core, updated->jobid); + if (job == NULL) { + WARN("%s JOB_UPDATE: unknown jobid %llu", __func__, + (unsigned long long)updated->jobid); + break; + } + + job->state = updated->state; + + if (updated->submit != NULL) { + if (updated->submit->queue != NULL) { + xce_sched_queue_t* q + = sched_get_queue(core, updated->submit->queue); + + if (q != NULL) { + job->queue = q; + job->priority = q->priority; + } + } + + if (updated->submit->resreq != NULL) { + xce_resreq_t* newreq + = (xce_resreq_t*)calloc(1, sizeof(*newreq)); + char* errmsg = NULL; + int prc; + + if (newreq == NULL) { + break; + } + + prc = parse_resreq(newreq, updated->submit->resreq, + core->resmgr, &errmsg); + if (prc < 0) { + ERROR("%s JOB_UPDATE parse_resreq failed: %s", + __func__, errmsg ? errmsg : "(null)"); + FREEUP(errmsg); + FREEUP(newreq); + break; + } + + FREEUP(job->resreq); + job->resreq = newreq; + } + } + + if (picker != NULL) { + picker->picker_iface->on_job_update(picker, job); + } + break; + } + + case XCE_CORE_EVENT_QUEUE_UPDATE: { + xce_queue_t* updated = ev->u.queue_update.queue; + + if ((updated == NULL) || (updated->queue == NULL)) { + WARN("%s QUEUE_UPDATE: NULL queue payload", __func__); + break; + } + + core_upsert_queue(core, picker, updated); + break; + } + + case XCE_CORE_EVENT_JOB_RUSAGE: { + xce_job_rusages_t* rusages = ev->u.job_rusage.rusages; + xce_sched_job_t* job; + + if (rusages == NULL) { + WARN("%s JOB_RUSAGE: NULL rusages payload", __func__); + break; + } + + job = sched_get_job(core, rusages->jobid); + if (job == NULL) { + WARN("%s JOB_RUSAGE: unknown jobid %llu", __func__, + (unsigned long long)rusages->jobid); + break; + } + + /* job takes ownership of rusages, same documented contract as + * sched_apply_job_rusage() (schedobj.c). */ + if (job->rusages != rusages) { + free_JobRusages_ptr(&(job->rusages)); + } + job->rusages = rusages; + + if ((picker != NULL) + && (picker->picker_iface->on_job_rusage != NULL)) { + picker->picker_iface->on_job_rusage(picker, job, rusages); + } + break; + } + + default: + WARN("%s unknown event type %d at batch index %d, skipping", + __func__, (int)ev->type, i); + break; + } + } + + return 0; +} diff --git a/src/scheduler/src/schedobj.c b/src/scheduler/src/schedobj.c index 3a459760..d581f7ca 100644 --- a/src/scheduler/src/schedobj.c +++ b/src/scheduler/src/schedobj.c @@ -367,37 +367,46 @@ void sched_set_default_job_pick_plugin(xce_scheduler_t* sched) /* * sched_resolve_job_picker - Req 1 AC 6 / Req 2 AC 5 (scheduler-core spec, - * task 2.4). See the declaration comment in scheduler.h for the full - * contract. + * task 2.4). See the declaration comment in core.h for the full contract. * - * Kept here (rather than in scheduler.c, where both its callers -- + * Kept here (rather than in scheduler.c, where its scheduler.c call sites -- * sched_init() and sched_job_picker() -- live) for the same testability * reason sched_set_default_job_pick_plugin() was: this function only needs - * sched->core.job_pick_plugin and sched->core.plugins (the plugin registry - * kvmgr), not the network/xcecli dependencies sched_init() otherwise pulls - * in, so it can be exercised directly by a white-box test of this - * translation unit (mirroring test_picker_config_fields.c's own #include - * "schedobj.c" convention) without linking scheduler.c. + * core->job_pick_plugin and core->plugins (the plugin registry kvmgr), not + * the network/xcecli dependencies sched_init() otherwise pulls in, so it + * can be exercised directly by a white-box test of this translation unit + * (mirroring test_picker_config_fields.c's own #include "schedobj.c" + * convention) without linking scheduler.c. + * + * Task 4.3 (scheduler-core spec): signature changed from xce_scheduler_t* + * to xce_sched_core_t* -- the function's own body only ever touched + * sched->core.job_pick_plugin/sched->core.plugins, never any other + * xce_scheduler_t field, so this is a pure narrowing, not a behavior + * change. Every call site already held an xce_scheduler_t* and now passes + * &sched->core instead (scheduler.c's sched_job_picker()/sched_allocator()/ + * sched_init(), and this file's own callers in the white-box test suite). + * Declared in core.h now (moved from scheduler.h) so xce_core_push_events() + * (core.c) -- which must never depend on scheduler.h -- can call it + * directly. */ -int sched_resolve_job_picker(xce_scheduler_t* sched, xce_sched_plugin_t** out) +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 ((sched == NULL) || (out == NULL)) { + if ((core == NULL) || (out == NULL)) { return -EINVAL; } *out = NULL; - if (sched->core.job_pick_plugin == NULL) { + if (core->job_pick_plugin == NULL) { return -EINVAL; } - lock(sched->core.plugins, RTTI_LOCK_READ); - rc = key_find(sched->core.plugins, sched->core.job_pick_plugin, - (void**)&plugin); - unlock(sched->core.plugins); + 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; diff --git a/src/scheduler/src/scheduler.c b/src/scheduler/src/scheduler.c index 85fdcf91..701097bf 100644 --- a/src/scheduler/src/scheduler.c +++ b/src/scheduler/src/scheduler.c @@ -79,7 +79,7 @@ static void* sched_job_picker(void* arg) xce_sched_plugin_t* job_picker = NULL; int rc = 0; - rc = sched_resolve_job_picker(sched, &job_picker); + rc = sched_resolve_job_picker(&(sched->core), &job_picker); if (rc < 0) { ERROR("%s failed to resolve job picker plugin '%s' (rc=%d)", __func__, sched->core.job_pick_plugin ? sched->core.job_pick_plugin @@ -309,7 +309,7 @@ static void* sched_allocator(void* arg) * exact resolution succeeding (before either thread is started), so * failure here should not occur in practice; handled the same * fail-return-NULL way sched_job_picker() handles it, for symmetry. */ - rc = sched_resolve_job_picker(sched, &job_picker); + rc = sched_resolve_job_picker(&(sched->core), &job_picker); if (rc < 0) { ERROR("%s failed to resolve job picker plugin '%s' (rc=%d)", __func__, sched->core.job_pick_plugin ? sched->core.job_pick_plugin @@ -532,7 +532,7 @@ int sched_init(xce_scheduler_t* sched, xce_client_config_t* config, * fast, without first paying for a controller connection attempt. */ sched_set_default_job_pick_plugin(sched); - rc = sched_resolve_job_picker(sched, &resolved_picker); + rc = sched_resolve_job_picker(&(sched->core), &resolved_picker); if (rc < 0) { ERROR("%s configured picker plugin '%s' failed to resolve as a " "valid registered picker (rc=%d)", diff --git a/src/scheduler/testsuite/meson.build b/src/scheduler/testsuite/meson.build index 4109c642..fcd25bb4 100644 --- a/src/scheduler/testsuite/meson.build +++ b/src/scheduler/testsuite/meson.build @@ -225,6 +225,22 @@ sched_core_jbuf_capacity_pin_test = executable( + '"'], install: false) +# Contract test for the core event interface and batch event push +# (scheduler-core spec, task 4.3, Req 5 AC 4 / AC 8): xce_core_push_events() +# fires the resolved picker's corresponding state-change hook exactly once +# per event, synchronously, before it returns, and rejects with -EBUSY (no +# event applied) when a session is active. Same white-box #include pattern +# as test_core_jbuf_capacity_pin.c/test_core_isolation.c (pulls the real +# ../src/core.c into this TU) -- see the test's own header comment for why +# it provides its own real sched_resolve_job_picker() rather than +# #include-ing schedobj.c. +sched_core_events_test = executable( + 'sched_core_events_test', + ['test_core_events.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', @@ -261,3 +277,6 @@ test('scheduler core isolation test', test('scheduler core jbuf capacity pin test', sched_core_jbuf_capacity_pin_test, suite: 'scheduler', env: ['ASAN_OPTIONS=detect_leaks=0']) +test('scheduler core events test', + sched_core_events_test, suite: 'scheduler', + env: ['ASAN_OPTIONS=detect_leaks=0']) diff --git a/src/scheduler/testsuite/test_core_events.c b/src/scheduler/testsuite/test_core_events.c new file mode 100644 index 00000000..fac03f8b --- /dev/null +++ b/src/scheduler/testsuite/test_core_events.c @@ -0,0 +1,714 @@ +/* + * test_core_events.c - contract test for the core event interface and + * batch event push (scheduler-core spec, task 4.3, Req 5 AC 4 / AC 8). + * + * Observable under test: pushing a batch of events with the picker idle + * (core->session_active == 0) fires the corresponding state-change hook + * exactly once per event, synchronously, before xce_core_push_events() + * returns -- and the session-active rejection path (-EBUSY, no event + * applied) when a session IS active. + * + * White-box #include of both ../src/core.c and ../src/core_events.c (same + * convention as test_core_isolation.c / test_core_jbuf_capacity_pin.c) so + * this test exercises the REAL xce_sched_core_init()/xce_core_push_events(), + * not a re-implementation -- xce_core_push_events() lives in core_events.c + * (split out of core.c purely to respect the 800-line-per-file cap; see + * that file's own header comment), which itself calls straight back into + * core.c's real sched_get_job()/sched_add_job()/etc. accessors, all within + * this one translation unit. Stubs below are only for scheduler-internal + * externs these two files reference (as kvmgr datakiller callback + * addresses, or -- for sched_resolve_job_picker -- because the real + * implementation lives in schedobj.c, a separate translation unit this + * file does not pull in): + * + * - kill_sched_job/kill_sched_queue/kill_sched_worker/kill_sched_user/ + * kill_sched_project/kill_sched_resval/plugin_unregister: pure no-op + * stubs, identical to test_core_jbuf_capacity_pin.c's own -- this + * file's fixtures are never torn down (never call + * xce_sched_core_fini()), same "fixtures aren't torn down" convention + * established across this directory, so these killers are never + * actually invoked; present only to satisfy the linker for + * xce_sched_core_init()'s kvmgr_params_t.datakiller fields. + * - sched_resolve_job_picker: a real, working reimplementation of + * schedobj.c's function of the same name (identical algorithm -- + * key_find core->job_pick_plugin in core->plugins, validate the + * picker MUST hooks) -- needed because the true implementation lives + * in schedobj.c, and pulling that whole file in would drag in its own + * large stub surface (plugin_queue_update, plugin_job_update, etc.) + * for functions this file's tests never exercise. This IS real logic, + * not a null stub, because the hook-firing contract under test + * depends on picker resolution actually working. + * + * make_job_account/free_job_account/free_JobAlloc_ptr/free_JobRusages_ptr/ + * makeResource/parse_resreq/makeString are all real, linked from the + * shared libxce library (src/lib), not stubbed -- same as every other + * scheduler white-box test in this directory. + */ +#include +#include +#include +#include + +#include + +#include "kvmgr.h" +#include "scheduler.h" +#include "xstring.h" + +/* Pull in the real xce_sched_core_init()/xce_core_push_events() under test + * into this TU (white-box #include, same convention as + * test_core_jbuf_capacity_pin.c's `#include "../src/core.c"`). */ +#include "../src/core.c" +#include "../src/core_events.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; } + +/* Real reimplementation of schedobj.c's sched_resolve_job_picker -- see + * file header comment for why this file provides its own copy instead of + * #include-ing schedobj.c. */ +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; +} + +void setUp(void) { return; } +void tearDown(void) { return; } + +/* ---------------------------------------------------------------------- */ +/* Test picker: records call counts + last-seen arguments per hook, so */ +/* tests can assert "exactly once, with the right argument" per event. */ +/* ---------------------------------------------------------------------- */ + +typedef struct { + int config_calls; + int job_push_calls; + int job_state_change_calls; + int job_update_calls; + int queue_update_calls; + int job_rusage_calls; + xce_conf4sched_t* last_config; + xce_sched_job_t* last_job_push; + xce_sched_job_t* last_job_state_change_job; + uint32_t last_job_state_change_state; + xce_sched_job_t* last_job_update; + xce_sched_queue_t* last_queue_update; + xce_sched_job_t* last_job_rusage_job; + xce_job_rusages_t* last_job_rusage_rusages; +} test_picker_counts_t; + +static test_picker_counts_t g_counts; + +static int test_begin_session(void* picker) +{ + (void)picker; + return 0; +} +static xce_sched_job_t* test_next_job(void* picker) +{ + (void)picker; + return NULL; +} +static int test_end_session(void* picker) +{ + (void)picker; + return 0; +} +static int test_on_dispatch(void* picker, uint64_t jobid) +{ + (void)picker; + (void)jobid; + return 0; +} +static int test_on_pend(void* picker, uint64_t jobid) +{ + (void)picker; + (void)jobid; + return 0; +} + +static int test_on_config(void* picker, xce_conf4sched_t* config) +{ + (void)picker; + g_counts.config_calls++; + g_counts.last_config = config; + return 0; +} + +static int test_on_job_push(void* picker, xce_sched_job_t* job) +{ + (void)picker; + g_counts.job_push_calls++; + g_counts.last_job_push = job; + return 0; +} + +static int test_on_job_state_change(void* picker, xce_sched_job_t* job, + uint32_t state) +{ + (void)picker; + g_counts.job_state_change_calls++; + g_counts.last_job_state_change_job = job; + g_counts.last_job_state_change_state = state; + return 0; +} + +static int test_on_job_update(void* picker, xce_sched_job_t* job) +{ + (void)picker; + g_counts.job_update_calls++; + g_counts.last_job_update = job; + return 0; +} + +static int test_on_queue_update(void* picker, xce_sched_queue_t* queue) +{ + (void)picker; + g_counts.queue_update_calls++; + g_counts.last_queue_update = queue; + return 0; +} + +static int test_on_job_rusage(void* picker, xce_sched_job_t* job, + xce_job_rusages_t* rusages) +{ + (void)picker; + g_counts.job_rusage_calls++; + g_counts.last_job_rusage_job = job; + g_counts.last_job_rusage_rusages = rusages; + return 0; +} + +/* Full picker: every hook wired, including the two optional ones + * (on_config/on_job_rusage). */ +static xce_job_picker_iface_t g_full_picker_iface = { + .init = NULL, + .fini = NULL, + .begin_session = test_begin_session, + .next_job = test_next_job, + .end_session = test_end_session, + .on_dispatch = test_on_dispatch, + .on_pend = test_on_pend, + .on_queue_update = test_on_queue_update, + .on_job_push = test_on_job_push, + .on_job_state_change = test_on_job_state_change, + .on_job_update = test_on_job_update, + .on_config = test_on_config, + .on_job_rusage = test_on_job_rusage, +}; + +/* Picker with on_job_rusage left NULL (Req 5 AC 4: "optional -- NULL if + * not registered") -- proves JOB_RUSAGE applies state without erroring + * when the hook is absent. */ +static xce_job_picker_iface_t g_no_rusage_picker_iface = { + .init = NULL, + .fini = NULL, + .begin_session = test_begin_session, + .next_job = test_next_job, + .end_session = test_end_session, + .on_dispatch = test_on_dispatch, + .on_pend = test_on_pend, + .on_queue_update = test_on_queue_update, + .on_job_push = test_on_job_push, + .on_job_state_change = test_on_job_state_change, + .on_job_update = test_on_job_update, + .on_config = test_on_config, + .on_job_rusage = NULL, +}; + +static xce_sched_plugin_t g_full_plugin + = { .name = "test_full_picker", .picker_iface = &g_full_picker_iface }; +static xce_sched_plugin_t g_no_rusage_plugin + = { .name = "test_no_rusage_picker", + .picker_iface = &g_no_rusage_picker_iface }; + +/* ---------------------------------------------------------------------- */ +/* Fixture: real xce_sched_core_init(), a test picker registered directly */ +/* into core.plugins (key_add -- the same mechanism plugin_register() */ +/* itself uses, schedobj.c not included so its wrapper is not available), */ +/* and core.job_pick_plugin pointed at it. Never torn down (this */ +/* directory's established convention -- see test_core_jbuf_capacity_pin.c */ +/* header comment). */ +/* ---------------------------------------------------------------------- */ + +static void fixture_init(xce_sched_core_t* core, xce_sched_plugin_t* plugin) +{ + int rc; + + memset(&g_counts, 0, sizeof(g_counts)); + + rc = xce_sched_core_init(core); + TEST_ASSERT_EQUAL_INT(0, rc); + + lock(core->plugins, RTTI_LOCK_WRITE); + rc = key_add(core->plugins, (void*)plugin->name, (void*)plugin); + unlock(core->plugins); + TEST_ASSERT_GREATER_OR_EQUAL(0, rc); + + core->job_pick_plugin = plugin->name; +} + +static xce_job_t* make_job(uint64_t jobid, uint32_t state, char* queue, + char* user, char* project) +{ + xce_job_t* j = (xce_job_t*)calloc(1, sizeof(*j)); + + j->jobid = jobid; + j->state = state; + j->sub_time = 1000; + j->queue_time = 1000; + j->start_time = 0; + j->submit = (job_submit_data_t*)calloc(1, sizeof(*(j->submit))); + j->submit->queue = queue; + j->submit->user = user; + j->submit->project_name = project; + j->submit->resreq = "type == any"; + j->submit->mintasks = 1; + j->submit->maxtasks = 1; + + return j; +} + +/* ---------------------------------------------------------------------- */ +/* Tests */ +/* ---------------------------------------------------------------------- */ + +/* One event of each of the 6 hook-having kinds in a single batch -- + * asserts each corresponding hook fired EXACTLY once, synchronously + * (already true by the time push_events returns, in this single-threaded + * test), with the right argument. */ +static void +test_push_events_fires_each_hook_once_per_event_with_picker_idle(void) +{ + xce_sched_core_t core; + xce_core_event_t events[6]; + xce_conf4sched_t config; + xce_job_t* job1; + xce_queue_t wire_queue; + xce_job_t job_update_payload; + xce_job_rusages_t rusages; + int rc; + + fixture_init(&core, &g_full_plugin); + + memset(&config, 0, sizeof(config)); + job1 = make_job(42, JOB_STAT_PEND, "normal", "alice", "proj1"); + + memset(&wire_queue, 0, sizeof(wire_queue)); + wire_queue.queue = "normal"; + wire_queue.priority = 5; + wire_queue.status = QUEUE_STAT_OPEN; + + memset(events, 0, sizeof(events)); + + events[0].type = XCE_CORE_EVENT_CONFIG_PUSH; + events[0].u.config_push.config = &config; + + events[1].type = XCE_CORE_EVENT_QUEUE_UPDATE; + events[1].u.queue_update.queue = &wire_queue; + + events[2].type = XCE_CORE_EVENT_JOBS_PUSH; + events[2].u.jobs_push.njobs = 1; + events[2].u.jobs_push.jobs = &job1; + + events[3].type = XCE_CORE_EVENT_JOB_STATE_CHANGE; + events[3].u.job_state_change.jobid = 42; + events[3].u.job_state_change.old_state = JOB_STAT_PEND; + events[3].u.job_state_change.new_state = JOB_STAT_RUN; + events[3].u.job_state_change.exit_code = 0; + + memset(&job_update_payload, 0, sizeof(job_update_payload)); + job_update_payload.jobid = 42; + job_update_payload.state = JOB_STAT_RUN; + + events[4].type = XCE_CORE_EVENT_JOB_UPDATE; + events[4].u.job_update.jobid = 42; + events[4].u.job_update.changed_fields = 0; + events[4].u.job_update.job = &job_update_payload; + + memset(&rusages, 0, sizeof(rusages)); + rusages.jobid = 42; + + events[5].type = XCE_CORE_EVENT_JOB_RUSAGE; + events[5].u.job_rusage.jobid = 42; + events[5].u.job_rusage.rusages = &rusages; + + rc = xce_core_push_events(&core, events, 6); + + TEST_ASSERT_EQUAL_INT(0, rc); + + TEST_ASSERT_EQUAL_INT(1, g_counts.config_calls); + TEST_ASSERT_EQUAL_PTR(&config, g_counts.last_config); + + TEST_ASSERT_EQUAL_INT(1, g_counts.queue_update_calls); + TEST_ASSERT_NOT_NULL(g_counts.last_queue_update); + TEST_ASSERT_EQUAL_STRING("normal", g_counts.last_queue_update->name); + + TEST_ASSERT_EQUAL_INT(1, g_counts.job_push_calls); + TEST_ASSERT_NOT_NULL(g_counts.last_job_push); + TEST_ASSERT_EQUAL_UINT64(42, g_counts.last_job_push->jobid); + + TEST_ASSERT_EQUAL_INT(1, g_counts.job_state_change_calls); + TEST_ASSERT_NOT_NULL(g_counts.last_job_state_change_job); + TEST_ASSERT_EQUAL_UINT64(42, g_counts.last_job_state_change_job->jobid); + TEST_ASSERT_EQUAL_UINT32(JOB_STAT_RUN, g_counts.last_job_state_change_state); + /* state mutation actually applied on the core-owned job object */ + TEST_ASSERT_EQUAL_UINT32(JOB_STAT_RUN, sched_get_job(&core, 42)->state); + + TEST_ASSERT_EQUAL_INT(1, g_counts.job_update_calls); + TEST_ASSERT_NOT_NULL(g_counts.last_job_update); + TEST_ASSERT_EQUAL_UINT64(42, g_counts.last_job_update->jobid); + + TEST_ASSERT_EQUAL_INT(1, g_counts.job_rusage_calls); + TEST_ASSERT_NOT_NULL(g_counts.last_job_rusage_job); + TEST_ASSERT_EQUAL_PTR(&rusages, g_counts.last_job_rusage_rusages); + TEST_ASSERT_EQUAL_PTR(&rusages, sched_get_job(&core, 42)->rusages); +} + +/* JOBS_PUSH with njobs > 1: on_job_push must fire once PER JOB, not once + * per event -- Req 5 AC 4's "on_job_push(picker, job) for each job in the + * batch". */ +static void test_jobs_push_fires_on_job_push_once_per_job(void) +{ + xce_sched_core_t core; + xce_core_event_t event; + xce_job_t* jobs[3]; + int rc; + + fixture_init(&core, &g_full_plugin); + + jobs[0] = make_job(1, JOB_STAT_PEND, "normal", "alice", "proj1"); + jobs[1] = make_job(2, JOB_STAT_PEND, "normal", "bob", "proj1"); + jobs[2] = make_job(3, JOB_STAT_PEND, "normal", "carol", "proj1"); + + memset(&event, 0, sizeof(event)); + event.type = XCE_CORE_EVENT_JOBS_PUSH; + event.u.jobs_push.njobs = 3; + event.u.jobs_push.jobs = jobs; + + rc = xce_core_push_events(&core, &event, 1); + + TEST_ASSERT_EQUAL_INT(0, rc); + TEST_ASSERT_EQUAL_INT(3, g_counts.job_push_calls); + TEST_ASSERT_NOT_NULL(sched_get_job(&core, 1)); + TEST_ASSERT_NOT_NULL(sched_get_job(&core, 2)); + TEST_ASSERT_NOT_NULL(sched_get_job(&core, 3)); +} + +/* WORKERS_PUSH/RESOURCES_PUSH/RESVALS_FULL/RESVALS_DELTA are state-update + * only -- Req 1 AC 2: "Hooks not defined on the picker interface (worker + * updates, resource updates, resval updates) are not forwarded to the + * picker." Every hook counter must stay at 0, while the underlying state + * IS genuinely mutated (not a no-op placeholder). */ +static void test_non_picker_events_mutate_state_but_never_call_hooks(void) +{ + xce_sched_core_t core; + xce_core_event_t events[4]; + xce_worker_t wire_worker; + xce_worker_t* worker_ptrs[1]; + xce_resource_t wire_resource; + xce_resource_t* resource_ptrs[1]; + xce_resval_t full_resval; + xce_resval_t* full_vals[1]; + xce_resval_t delta_resval; + xce_resval_t* delta_vals[1]; + xce_resval_list_t deltas; + xce_val_t full_val; + xce_val_t delta_val; + int rc; + + fixture_init(&core, &g_full_plugin); + + memset(&wire_worker, 0, sizeof(wire_worker)); + wire_worker.name = "hostA"; + wire_worker.status = 0; + wire_worker.nresvals_to_sched = 0; + worker_ptrs[0] = &wire_worker; + + memset(&wire_resource, 0, sizeof(wire_resource)); + wire_resource.name = "cpu"; + wire_resource.idx = 0; + resource_ptrs[0] = &wire_resource; + + memset(&full_val, 0, sizeof(full_val)); + memset(&full_resval, 0, sizeof(full_resval)); + full_resval.name = "cpu"; + full_resval.val = &full_val; + full_vals[0] = &full_resval; + + memset(&delta_val, 0, sizeof(delta_val)); + memset(&delta_resval, 0, sizeof(delta_resval)); + delta_resval.name = "cpu"; + delta_resval.host = "hostA"; + delta_resval.val = &delta_val; + delta_vals[0] = &delta_resval; + deltas.count = 1; + deltas.vals = delta_vals; + + memset(events, 0, sizeof(events)); + + events[0].type = XCE_CORE_EVENT_WORKERS_PUSH; + events[0].u.workers_push.nworkers = 1; + events[0].u.workers_push.workers = worker_ptrs; + + events[1].type = XCE_CORE_EVENT_RESOURCES_PUSH; + events[1].u.resources_push.n = 1; + events[1].u.resources_push.resources = resource_ptrs; + + events[2].type = XCE_CORE_EVENT_RESVALS_FULL; + events[2].u.resvals_full.n = 1; + events[2].u.resvals_full.resvals = full_vals; + + events[3].type = XCE_CORE_EVENT_RESVALS_DELTA; + events[3].u.resvals_delta.deltas = &deltas; + + /* worker must exist before RESVALS_DELTA can attribute a value to it */ + rc = xce_core_push_events(&core, events, 1); /* WORKERS_PUSH only, first */ + TEST_ASSERT_EQUAL_INT(0, rc); + rc = xce_core_push_events(&core, &events[1], 3); /* RESOURCES/RESVALS_FULL/RESVALS_DELTA */ + TEST_ASSERT_EQUAL_INT(0, rc); + + TEST_ASSERT_EQUAL_INT(0, g_counts.config_calls); + TEST_ASSERT_EQUAL_INT(0, g_counts.job_push_calls); + TEST_ASSERT_EQUAL_INT(0, g_counts.job_state_change_calls); + TEST_ASSERT_EQUAL_INT(0, g_counts.job_update_calls); + TEST_ASSERT_EQUAL_INT(0, g_counts.queue_update_calls); + TEST_ASSERT_EQUAL_INT(0, g_counts.job_rusage_calls); + + TEST_ASSERT_NOT_NULL(sched_get_worker(&core, "hostA")); +} + +/* JOB_RUSAGE's on_job_rusage hook is OPTIONAL (Req 5 AC 4) -- a picker + * without it must still apply the state mutation and must not error or + * crash. */ +static void test_job_rusage_applies_state_when_hook_is_null(void) +{ + xce_sched_core_t core; + xce_core_event_t events[2]; + xce_job_t* job1; + xce_job_rusages_t rusages; + int rc; + + fixture_init(&core, &g_no_rusage_plugin); + + job1 = make_job(7, JOB_STAT_RUN, "normal", "alice", "proj1"); + + memset(&rusages, 0, sizeof(rusages)); + rusages.jobid = 7; + + memset(events, 0, sizeof(events)); + events[0].type = XCE_CORE_EVENT_JOBS_PUSH; + events[0].u.jobs_push.njobs = 1; + events[0].u.jobs_push.jobs = &job1; + + events[1].type = XCE_CORE_EVENT_JOB_RUSAGE; + events[1].u.job_rusage.jobid = 7; + events[1].u.job_rusage.rusages = &rusages; + + rc = xce_core_push_events(&core, events, 2); + + TEST_ASSERT_EQUAL_INT(0, rc); + TEST_ASSERT_EQUAL_INT(0, g_counts.job_rusage_calls); + TEST_ASSERT_EQUAL_PTR(&rusages, sched_get_job(&core, 7)->rusages); +} + +/* CONFIG_PUSH must decompose the pushed xce_conf4sched_t's embedded + * queues[]/resources[]/resvals[] arrays into core state -- not just store + * the config pointer and bump configver (Req 5 AC 4's CONFIG_PUSH row: + * "All queues, resource thresholds, ... " is part of the payload, and + * decomposing it is the core's job, mirroring scheduler.c's + * sched_init_resources()/sched_init_sharedInst()/sched_init_queues(), all + * three of which sched_init_config() calls today for this exact same + * xce_conf4sched_t payload). Also asserts on_queue_update fires once per + * embedded queue (design intent: a config push's queues behave the same as + * individual queue updates, just batched). */ +static void test_config_push_decomposes_queues_resources_and_resvals(void) +{ + xce_sched_core_t core; + xce_core_event_t event; + xce_conf4sched_t config; + xce_resource_t wire_resource; + xce_resource_t* resources[1]; + xce_val_t resval_value; + xce_name_val_t wire_resval; + xce_name_val_t* resvals[1]; + xce_queue_t wire_queue; + xce_queue_t* queues[1]; + xce_resource_t* found_resource = NULL; + xce_sched_queue_t* found_queue; + xce_val_t* shared_val; + int rc; + + fixture_init(&core, &g_full_plugin); + + memset(&wire_resource, 0, sizeof(wire_resource)); + wire_resource.name = "cpu"; + resources[0] = &wire_resource; + + memset(&resval_value, 0, sizeof(resval_value)); + resval_value.u.ivalue32 = 8; + + memset(&wire_resval, 0, sizeof(wire_resval)); + wire_resval.name = "cpu"; + wire_resval.val = &resval_value; + resvals[0] = &wire_resval; + + memset(&wire_queue, 0, sizeof(wire_queue)); + wire_queue.queue = "batch"; + wire_queue.priority = 7; + wire_queue.status = QUEUE_STAT_OPEN; + queues[0] = &wire_queue; + + memset(&config, 0, sizeof(config)); + config.n_resources = 1; + config.resources = resources; + config.nresvals = 1; + config.resvals = resvals; + config.n_queues = 1; + config.queues = queues; + + memset(&event, 0, sizeof(event)); + event.type = XCE_CORE_EVENT_CONFIG_PUSH; + event.u.config_push.config = &config; + + rc = xce_core_push_events(&core, &event, 1); + + TEST_ASSERT_EQUAL_INT(0, rc); + TEST_ASSERT_EQUAL_INT(1, g_counts.config_calls); + + /* resource decomposed into core->resmgr */ + rc = key_find(core.resmgr, (void*)"cpu", (void**)&found_resource); + TEST_ASSERT_GREATER_THAN_INT(0, rc); + TEST_ASSERT_NOT_NULL(found_resource); + + /* resval decomposed into core->sharedInst at the resource's idx */ + shared_val = (xce_val_t*)zarray_get(&(core.sharedInst), found_resource->idx); + TEST_ASSERT_NOT_NULL(shared_val); + TEST_ASSERT_EQUAL_INT(8, shared_val->u.ivalue32); + + /* queue decomposed into core->queues, and on_queue_update fired once */ + found_queue = sched_get_queue(&core, "batch"); + TEST_ASSERT_NOT_NULL(found_queue); + TEST_ASSERT_EQUAL_INT(7, found_queue->priority); + TEST_ASSERT_EQUAL_INT(1, g_counts.queue_update_calls); + TEST_ASSERT_EQUAL_PTR(found_queue, g_counts.last_queue_update); +} + +/* Session-active rejection (Req 5 AC 8 / design.md precondition): + * core_push_events() is only valid between sessions. */ +static void +test_push_events_rejects_with_ebusy_when_session_active_no_mutation(void) +{ + xce_sched_core_t core; + xce_core_event_t event; + xce_job_t* job1; + int rc; + + fixture_init(&core, &g_full_plugin); + + atomic_store(&(core.session_active), 1); + + job1 = make_job(99, JOB_STAT_PEND, "normal", "alice", "proj1"); + + memset(&event, 0, sizeof(event)); + event.type = XCE_CORE_EVENT_JOBS_PUSH; + event.u.jobs_push.njobs = 1; + event.u.jobs_push.jobs = &job1; + + rc = xce_core_push_events(&core, &event, 1); + + TEST_ASSERT_EQUAL_INT(-EBUSY, rc); + TEST_ASSERT_EQUAL_INT(0, g_counts.job_push_calls); + TEST_ASSERT_NULL(sched_get_job(&core, 99)); +} + +/* Bad-argument guard clauses. */ +static void test_push_events_null_core_returns_einval(void) +{ + xce_core_event_t event; + int rc; + + memset(&event, 0, sizeof(event)); + event.type = XCE_CORE_EVENT_CONFIG_PUSH; + + rc = xce_core_push_events(NULL, &event, 1); + + TEST_ASSERT_EQUAL_INT(-EINVAL, rc); +} + +static void test_push_events_null_events_with_positive_nevents_returns_einval(void) +{ + xce_sched_core_t core; + int rc; + + fixture_init(&core, &g_full_plugin); + + rc = xce_core_push_events(&core, NULL, 1); + + TEST_ASSERT_EQUAL_INT(-EINVAL, rc); +} + +/* Zero-length batch is a valid no-op (not an error). */ +static void test_push_events_zero_nevents_is_a_valid_noop(void) +{ + xce_sched_core_t core; + int rc; + + fixture_init(&core, &g_full_plugin); + + rc = xce_core_push_events(&core, NULL, 0); + + TEST_ASSERT_EQUAL_INT(0, rc); +} + +int main(void) +{ + UNITY_BEGIN(); + RUN_TEST(test_push_events_fires_each_hook_once_per_event_with_picker_idle); + RUN_TEST(test_jobs_push_fires_on_job_push_once_per_job); + RUN_TEST(test_non_picker_events_mutate_state_but_never_call_hooks); + RUN_TEST(test_job_rusage_applies_state_when_hook_is_null); + RUN_TEST(test_config_push_decomposes_queues_resources_and_resvals); + RUN_TEST(test_push_events_rejects_with_ebusy_when_session_active_no_mutation); + RUN_TEST(test_push_events_null_core_returns_einval); + RUN_TEST(test_push_events_null_events_with_positive_nevents_returns_einval); + RUN_TEST(test_push_events_zero_nevents_is_a_valid_noop); + return UNITY_END(); +} diff --git a/src/scheduler/testsuite/test_picker_wiring.c b/src/scheduler/testsuite/test_picker_wiring.c index d1d10f25..86fb0796 100644 --- a/src/scheduler/testsuite/test_picker_wiring.c +++ b/src/scheduler/testsuite/test_picker_wiring.c @@ -297,7 +297,7 @@ static void test_unset_job_pick_plugin_resolves_to_builtin_fcfs_by_default(void) sched_set_default_job_pick_plugin(&sched); TEST_ASSERT_EQUAL_STRING("builtin_fcfs", sched.core.job_pick_plugin); - rc = sched_resolve_job_picker(&sched, &resolved); + rc = sched_resolve_job_picker(&sched.core, &resolved); TEST_ASSERT_EQUAL_INT(0, rc); TEST_ASSERT_NOT_NULL(resolved); @@ -311,7 +311,7 @@ static void test_unset_job_pick_plugin_resolves_to_builtin_fcfs_by_default(void) /* tasks.md Observable, half 2: "starting it with an unregistered picker * name fails startup with a clear error instead of silently falling * back." -ENOENT specifically (distinct from -EINVAL below) -- - * sched_resolve_job_picker's documented contract (scheduler.h). */ + * sched_resolve_job_picker's documented contract (core.h). */ static void test_unregistered_configured_name_fails_with_enoent_not_silent_fallback(void) { @@ -324,7 +324,7 @@ test_unregistered_configured_name_fails_with_enoent_not_silent_fallback(void) sched.core.job_pick_plugin = "totally_unregistered_picker_name"; - rc = sched_resolve_job_picker(&sched, &resolved); + rc = sched_resolve_job_picker(&sched.core, &resolved); TEST_ASSERT_EQUAL_INT(-ENOENT, rc); /* Not a silent fallback to builtin_fcfs (still registered) or to @@ -358,7 +358,7 @@ static void test_registered_but_not_a_picker_fails_with_einval_not_enoent(void) sched.core.job_pick_plugin = "not_a_picker"; - rc = sched_resolve_job_picker(&sched, &resolved); + rc = sched_resolve_job_picker(&sched.core, &resolved); TEST_ASSERT_EQUAL_INT(-EINVAL, rc); TEST_ASSERT_NULL(resolved); @@ -377,7 +377,7 @@ test_unset_job_pick_plugin_without_defaulting_fails_with_einval(void) fixture_init(&sched); TEST_ASSERT_NULL(sched.core.job_pick_plugin); - rc = sched_resolve_job_picker(&sched, &resolved); + rc = sched_resolve_job_picker(&sched.core, &resolved); TEST_ASSERT_EQUAL_INT(-EINVAL, rc); TEST_ASSERT_NULL(resolved); @@ -411,7 +411,7 @@ test_resolved_legacy_plugin_has_job_elect_fn_set_selecting_legacy_branch(void) sched.core.job_pick_plugin = "test_legacy_picker_for_wiring"; - rc = sched_resolve_job_picker(&sched, &resolved); + rc = sched_resolve_job_picker(&sched.core, &resolved); TEST_ASSERT_EQUAL_INT(0, rc); TEST_ASSERT_NOT_NULL(resolved); @@ -437,7 +437,7 @@ test_resolved_builtin_fcfs_has_job_elect_fn_null_selecting_native_branch(void) sched.core.job_pick_plugin = "builtin_fcfs"; /* explicit, not defaulted */ - rc = sched_resolve_job_picker(&sched, &resolved); + rc = sched_resolve_job_picker(&sched.core, &resolved); TEST_ASSERT_EQUAL_INT(0, rc); TEST_ASSERT_NOT_NULL(resolved); @@ -467,12 +467,12 @@ test_legacy_and_native_pickers_coexist_and_resolve_independently(void) TEST_ASSERT_GREATER_OR_EQUAL(0, register_builtin_fcfs(&sched)); sched.core.job_pick_plugin = "xce_sched_fcfs_stub"; - rc = sched_resolve_job_picker(&sched, &resolved); + rc = sched_resolve_job_picker(&sched.core, &resolved); TEST_ASSERT_EQUAL_INT(0, rc); TEST_ASSERT_NOT_NULL(resolved->job_elect_fn); sched.core.job_pick_plugin = "builtin_fcfs"; - rc = sched_resolve_job_picker(&sched, &resolved); + rc = sched_resolve_job_picker(&sched.core, &resolved); TEST_ASSERT_EQUAL_INT(0, rc); TEST_ASSERT_NULL(resolved->job_elect_fn); }