From dceafc6b12ddcc1321935d58f8398d2963b07f5a Mon Sep 17 00:00:00 2001 From: Rongsong Shen Date: Tue, 4 Aug 2026 04:39:17 +0000 Subject: [PATCH] feat(scheduler-core): build normal-mode driver event translation path (task 5.1) Adds driver.h/driver.c: xce_sched_driver_translate_pending_events() drains core->pend_updates via the existing dataq-swap pattern (same mechanism sched_update() uses), translates all 10 xce_sched_update_ request_t op types into the matching xce_core_event_t, and pushes the batch through xce_core_push_events(). Scoped to xce_sched_core_t* alone -- pend_updates already lives on core since task 4.1, and this task is translation-only (no decision publishing, no mode-selection, no recording -- all later work). handler.c (the enqueue side) is untouched. First task in Group 5 (Core: Normal-Mode Driver). Co-Authored-By: Claude Sonnet 5 Claude-Session: https://claude.ai/code/session_019dw5WnN3DXStaEaPYfACU1 --- .kiro/specs/scheduler-core/tasks.md | 3 +- src/scheduler/include/driver.h | 55 ++ src/scheduler/meson.build | 1 + src/scheduler/src/driver.c | 386 +++++++++ src/scheduler/testsuite/meson.build | 22 + .../testsuite/test_driver_translate.c | 751 ++++++++++++++++++ 6 files changed, 1217 insertions(+), 1 deletion(-) create mode 100644 src/scheduler/include/driver.h create mode 100644 src/scheduler/src/driver.c create mode 100644 src/scheduler/testsuite/test_driver_translate.c diff --git a/.kiro/specs/scheduler-core/tasks.md b/.kiro/specs/scheduler-core/tasks.md index 9fbe4f20..6cae0c16 100644 --- a/.kiro/specs/scheduler-core/tasks.md +++ b/.kiro/specs/scheduler-core/tasks.md @@ -133,10 +133,11 @@ - [ ] 5. Core: Normal-Mode Driver -- [ ] 5.1 Build the normal-mode driver's event translation path +- [x] 5.1 Build the normal-mode driver's event translation path - Translate controller capnp push/notify messages into core events, buffering them in a pending-event queue between scheduling rounds - Observable: a sequence of simulated controller push/notify messages, once drained, produces a core event batch whose contents match the input messages one-to-one - _Requirements: 5.8, 5.9_ + - _Verified: 2026-08-03 — new driver.h/driver.c: xce_sched_driver_translate_pending_events(xce_sched_core_t*) drains pend_updates via the existing dataq-swap pattern, translates all 10 op types to xce_core_event_t (field-verified against core_events.c's actual consumption, not assumed), stamps via xce_core_now(), pushes through the real xce_core_push_events(). Deliberately scoped to xce_sched_core_t* alone (not the full design.md xce_sched_driver_t) since pend_updates already lives on core since task 4.1 and this task is translation-only — task 5.2 can wrap trivially with &driver->core. handler.c untouched. Two flagged risk items (RESVALS zarray dangling-pointer shape, zero-filled changed_fields/exit_code for fields the legacy payload can't supply) both independently verified as pre-existing/honestly-inert, not new bugs. 20/20 scheduler suite passing, real-ASan clean, clang-format clean, review APPROVED first pass._ - _Depends: 4.4_ - [ ] 5.2 Wire decision publishing, the mode-selection entry point, and retire the hardcoded decision switch diff --git a/src/scheduler/include/driver.h b/src/scheduler/include/driver.h new file mode 100644 index 00000000..107b66f2 --- /dev/null +++ b/src/scheduler/include/driver.h @@ -0,0 +1,55 @@ +#if !defined(_DRIVER_H_) + +#define _DRIVER_H_ 1 + +#include "core.h" + +/* + * driver.h - normal-mode driver, controller event translation (scheduler- + * core spec, task 5.1, Req 5 AC 8 / AC 9). NEW file per design.md's File + * Structure Plan (`driver.h` bullet: "NEW (Phase 2): xce_sched_driver_t + * (normal-mode)"). This task deliberately does NOT define the full + * `xce_sched_driver_t` struct design.md sketches (embedding `xcecli_ctx_t`, + * `cluster_cfg`, recording-control fields) -- that struct's fields + * (decision publishing, `record_dir`/`record_file`, etc.) belong to task + * 5.2 and Phase 3's recording work, none of which this task implements or + * needs. See driver.c's own header comment for the full scoping rationale: + * this task's translation logic only needs an `xce_sched_core_t*` (which + * already carries `pend_updates[2]` as of task 4.1), not a driver-level + * struct wrapping one. Task 5.2 extends this header with the full + * `xce_sched_driver_t` and `sched_driver_run()` when it wires this function + * into a live driver loop -- nothing here needs to change for that. + */ + +/* + * xce_sched_driver_translate_pending_events - Req 5 AC 8 ("Driver event + * queue" -- the driver buffers controller events between scheduling + * rounds) and Req 5 AC 9's translation half ("receive capnp push/notify + * messages from the controller, translate each to the corresponding core + * event type, and invoke the core's event push interface" -- decision + * publishing, AC 9's other half, is task 5.2's job, not this function's). + * + * Drains `core->pend_updates` -- the SAME existing dataq-swap-and-dequeue + * mechanism scheduler.c's `sched_update()` already uses, populated by the + * unchanged `handler.c` capnp push/notify enqueue path -- translating every + * dequeued `xce_sched_update_request_t` into the matching `xce_core_event_t` + * (see driver.c's op-type -> event-type table), accumulating the whole + * batch, and pushing it through `xce_core_push_events()` exactly once. + * + * Only valid to call between scheduling sessions (the same precondition + * `xce_core_push_events()` itself documents, core.h) -- this function does + * not drive a scheduling session and does not decide WHEN it is safe to + * call; that session-boundary orchestration (BEGIN_ROUND) is a later task's + * responsibility (see this file's header comment). + * + * Returns the number of events translated and pushed (>= 0, may be 0 if + * nothing was pending) on success. Returns a negative errno on failure: + * -EINVAL if `core` is NULL or its `pend_updates` queues are not + * initialized; -ENOMEM on an allocation failure while growing the + * translation batch (rare, defensive path); or `xce_core_push_events()`'s + * own negative return (e.g. -EBUSY if a session was unexpectedly active) + * passed straight through. + */ +extern int xce_sched_driver_translate_pending_events(xce_sched_core_t* core); + +#endif diff --git a/src/scheduler/meson.build b/src/scheduler/meson.build index 12fc6aae..0f83cdb9 100644 --- a/src/scheduler/meson.build +++ b/src/scheduler/meson.build @@ -6,6 +6,7 @@ scheduler_srcs = [ 'src/core_events.c', 'src/core_session.c', 'src/core_clock.c', + 'src/driver.c', 'src/decision.c', 'src/schedobj.c', 'src/plugin.c', diff --git a/src/scheduler/src/driver.c b/src/scheduler/src/driver.c new file mode 100644 index 00000000..8ff33a32 --- /dev/null +++ b/src/scheduler/src/driver.c @@ -0,0 +1,386 @@ +#include +#include +#include + +#include "controller.capnp.h" +#include "core.h" +#include "driver.h" +#include "scheduler.h" +#include "utils.h" + +/* + * driver.c - normal-mode driver, controller event translation (scheduler- + * core spec, task 5.1, Req 5 AC 8 / AC 9's translation half). + * + * Op-type -> event-type table (requirements.md Req 5 AC 9 / AC 4; every + * field mapping below was verified field-by-field against the CURRENT + * `xce_sched_update_request_t` (scheduler.h) and `xce_core_event_t` (core.h) + * union layouts, and against how `xce_core_push_events()` (core_events.c) + * actually consumes each payload -- not assumed): + * + * XCE_SCHED_OP_UPDATE_CONFIG -> XCE_CORE_EVENT_CONFIG_PUSH + * XCE_SCHED_OP_UPDATE_JOB -> XCE_CORE_EVENT_JOBS_PUSH + * XCE_SCHED_OP_UPDATE_JOB_FIELDS -> XCE_CORE_EVENT_JOB_UPDATE + * XCE_SCHED_OP_UPDATE_WORKER -> XCE_CORE_EVENT_WORKERS_PUSH + * XCE_SCHED_OP_JSTATE_CHANGE -> XCE_CORE_EVENT_JOB_STATE_CHANGE + * XCE_SCHED_OP_UPDATE_QUEUE -> XCE_CORE_EVENT_QUEUE_UPDATE + * XCE_SCHED_OP_UPDATE_RESOURCES -> XCE_CORE_EVENT_RESOURCES_PUSH + * XCE_SCHED_OP_UPDATE_RESVALS_FULL -> XCE_CORE_EVENT_RESVALS_FULL + * XCE_SCHED_OP_UPDATE_RESVALS -> XCE_CORE_EVENT_RESVALS_DELTA + * XCE_SCHED_OP_JOB_RUSAGE -> XCE_CORE_EVENT_JOB_RUSAGE + * + * Two documented judgment calls on fields the legacy + * `xce_sched_update_request_t` payload simply does not carry (handler.c is + * unchanged by this task -- these gaps are not introduced by translation, + * they already exist between what the controller sends today and the + * richer `xce_core_event_t` field set task 4.3 defined ahead of the wire + * format that will eventually populate it): + * + * - `job_update.changed_fields` (bitmask): no JOB_F_* constants are + * defined anywhere in this codebase yet (requirements.md's JOB_F_ + * PRIORITY/QUEUE/RESREQ names are illustrative only) and the legacy + * XCE_SCHED_OP_UPDATE_JOB_FIELDS payload carries no per-field diff. + * Set to 0 ("unspecified"), never a fabricated bitmask. + * - `job_state_change.exit_code`: the legacy XCE_SCHED_OP_JSTATE_CHANGE + * payload carries only (jobid, state) -- no exit code. Set to 0. + * + * `job_state_change.old_state` IS derivable (unlike the two above) by + * looking up the job's CURRENT state via `sched_get_job()` at translation + * time, before this batch is applied -- so it is looked up for real, not + * defaulted. Known limitation: if a single drained batch contains two + * JSTATE_CHANGE events for the SAME jobid, both report the same pre-batch + * old_state (this function does not track intra-batch shadow state). Req 5 + * AC 4 documents old_state as "(for context)" -- diagnostic only, not read + * by any core mutation or picker hook (core_events.c's JOB_STATE_CHANGE + * case only reads new_state) -- so this is a proportionate, documented + * simplification, not a correctness gap in scheduling behavior. + * + * Ownership/cleanup: every dequeued `xce_sched_update_request_t`'s wire + * payload is released after translation via release_translated_payload() + * below, mirroring scheduler.c's sched_pend_update_killer() -- SAME + * per-op-type free calls -- with exactly two deltas, both verified by + * reading xce_core_push_events()'s real mutation code (core_events.c), not + * assumed: + * + * - CONFIG_PUSH: sched_pend_update_killer() always frees the config + * (sched_update()'s own CONFIG case never actually applies it -- "should + * never get here"). xce_core_push_events() DOES apply it for real + * (`core->config = config`), permanently transferring ownership to the + * core. This function frees it only when the batch was NOT applied + * (`xce_core_push_events()` returned < 0) -- otherwise the core now + * owns it and freeing it would be a use-after-free. + * - JOB_RUSAGE: core_events.c's JOB_RUSAGE case sets `job->rusages = + * rusages` (ownership transfer) ONLY if the jobid is found -- an + * unknown jobid just WARNs and never touches the pointer. Since + * `xce_core_push_events()` returns one aggregate rc for the whole + * batch (no per-event outcome), this function re-reads + * `sched_get_job(core, jobid)->rusages` after the push and compares it + * against the pointer it sent, freeing only when it was NOT adopted. + * + * Every other op type's wire wrapper is never adopted by + * xce_core_push_events() (it always shallow-copies scalar fields or + * deep-copies strings via makeString()/parse_resreq() -- see core_events.c's + * core_build_job/core_build_worker/core_build_queue/core_upsert_queue), + * exactly mirroring sched_update()'s own inline handling of the same op + * types (scheduler.c) -- so their cleanup is unconditional, identical to + * sched_pend_update_killer(), regardless of whether the push succeeded. + * + * This function does NOT drive a scheduling session, does NOT publish + * decisions to a controller, and is not wired into any live loop -- + * unwired-until-a-driver-exists, same additive-only pattern as tasks + * 4.1-4.6 (main.c's mode-selection branch and the full xce_sched_driver_t + * are task 5.2's job). + */ + +/* + * release_translated_payload - see this file's header comment for the full + * ownership reasoning. `applied` is true iff xce_core_push_events() ran and + * returned >= 0 for the batch this request's translated event belonged to + * (false for records that failed translation before ever reaching a batch, + * or that were part of a batch never successfully pushed). + */ +static void release_translated_payload(xce_sched_core_t* core, + xce_sched_update_request_t* req, + int applied) +{ + int i; + + switch (req->op) { + case XCE_SCHED_OP_UPDATE_CONFIG: + if (!applied) { + free_Conf4Sched_ptr(&(req->u.config)); + } + /* else: core->config now owns it permanently -- do not free. */ + break; + + case XCE_SCHED_OP_UPDATE_JOB: + for (i = 0; i < req->u.jobupdate.njobs; i++) { + free_Job_ptr(&(req->u.jobupdate.jobs[i])); + } + free(req->u.jobupdate.jobs); + break; + + case XCE_SCHED_OP_UPDATE_JOB_FIELDS: + free_Job_ptr(&(req->u.jobfieldsupdate.job)); /* NULL-safe */ + break; + + case XCE_SCHED_OP_UPDATE_WORKER: + for (i = 0; i < req->u.workerupdate.nworkers; i++) { + free_Worker_ptr(&(req->u.workerupdate.workers[i])); + } + free(req->u.workerupdate.workers); + break; + + case XCE_SCHED_OP_JSTATE_CHANGE: + break; /* no payload pointers */ + + case XCE_SCHED_OP_UPDATE_QUEUE: + free_QueueConf_ptr(&(req->u.queueupdate.queue)); + break; + + case XCE_SCHED_OP_UPDATE_RESOURCES: + for (i = 0; i < req->u.resourceupdate.nresources; i++) { + free_Resource_ptr(&(req->u.resourceupdate.resources[i])); + } + free(req->u.resourceupdate.resources); + break; + + case XCE_SCHED_OP_UPDATE_RESVALS_FULL: + for (i = 0; i < req->u.resvalfullupdate.nvals; i++) { + free_ResourceVal_ptr(&(req->u.resvalfullupdate.vals[i])); + } + free(req->u.resvalfullupdate.vals); + break; + + case XCE_SCHED_OP_UPDATE_RESVALS: + free_ResValList_ptr(&(req->u.resvalupdate.resvals)); /* NULL-safe */ + break; + + case XCE_SCHED_OP_JOB_RUSAGE: { + xce_job_rusages_t* rusages = req->u.job_rusage.rusages; + xce_sched_job_t* job; + + if (rusages == NULL) { + break; + } + + job = applied ? sched_get_job(core, rusages->jobid) : NULL; + if ((job != NULL) && (job->rusages == rusages)) { + break; /* adopted by the job object -- do not free */ + } + + free_JobRusages_ptr(&(req->u.job_rusage.rusages)); /* NULL-safe */ + break; + } + + default: + break; + } +} + +/* + * translate_one_pending_update - translates one dequeued + * xce_sched_update_request_t into *ev per this file's op-type table. + * Returns 1 and populates *ev on success; returns 0 (logged, not fatal to + * the rest of the drain -- same discipline xce_core_push_events()'s own + * per-event loop follows) when the record cannot be translated: a + * malformed payload (NULL job/rusages the enqueue side should never have + * produced, guarded defensively here anyway) or an unrecognized op. + */ +static int translate_one_pending_update(xce_sched_core_t* core, + xce_sched_update_request_t* req, + uint64_t ts, xce_core_event_t* ev) +{ + memset(ev, 0, sizeof(*ev)); + ev->ts = ts; + + switch (req->op) { + case XCE_SCHED_OP_UPDATE_CONFIG: + ev->type = XCE_CORE_EVENT_CONFIG_PUSH; + ev->u.config_push.config = req->u.config; + return 1; + + case XCE_SCHED_OP_UPDATE_JOB: + ev->type = XCE_CORE_EVENT_JOBS_PUSH; + ev->u.jobs_push.njobs = req->u.jobupdate.njobs; + ev->u.jobs_push.jobs = req->u.jobupdate.jobs; + return 1; + + case XCE_SCHED_OP_UPDATE_JOB_FIELDS: + if (req->u.jobfieldsupdate.job == NULL) { + WARN("%s UPDATE_JOB_FIELDS: NULL job payload, skipping", __func__); + return 0; + } + ev->type = XCE_CORE_EVENT_JOB_UPDATE; + ev->u.job_update.jobid = req->u.jobfieldsupdate.job->jobid; + ev->u.job_update.job = req->u.jobfieldsupdate.job; + /* changed_fields: see file header comment -- documented 0, not a + * fabricated bitmask. */ + ev->u.job_update.changed_fields = 0; + return 1; + + case XCE_SCHED_OP_UPDATE_WORKER: + ev->type = XCE_CORE_EVENT_WORKERS_PUSH; + ev->u.workers_push.nworkers = req->u.workerupdate.nworkers; + ev->u.workers_push.workers = req->u.workerupdate.workers; + return 1; + + case XCE_SCHED_OP_JSTATE_CHANGE: { + xce_sched_job_t* job + = sched_get_job(core, req->u.job_state_change.jobid); + + ev->type = XCE_CORE_EVENT_JOB_STATE_CHANGE; + ev->u.job_state_change.jobid = req->u.job_state_change.jobid; + /* old_state read from core BEFORE this batch is applied -- see file + * header comment for the same-jobid-twice-in-one-batch limitation. */ + ev->u.job_state_change.old_state = (job != NULL) ? job->state : 0; + ev->u.job_state_change.new_state = req->u.job_state_change.state; + /* exit_code: no source of truth on the legacy payload -- see file + * header comment. */ + ev->u.job_state_change.exit_code = 0; + return 1; + } + + case XCE_SCHED_OP_UPDATE_QUEUE: + ev->type = XCE_CORE_EVENT_QUEUE_UPDATE; + ev->u.queue_update.queue = req->u.queueupdate.queue; + return 1; + + case XCE_SCHED_OP_UPDATE_RESOURCES: + ev->type = XCE_CORE_EVENT_RESOURCES_PUSH; + ev->u.resources_push.n = req->u.resourceupdate.nresources; + ev->u.resources_push.resources = req->u.resourceupdate.resources; + return 1; + + case XCE_SCHED_OP_UPDATE_RESVALS_FULL: + ev->type = XCE_CORE_EVENT_RESVALS_FULL; + ev->u.resvals_full.n = req->u.resvalfullupdate.nvals; + ev->u.resvals_full.resvals = req->u.resvalfullupdate.vals; + return 1; + + case XCE_SCHED_OP_UPDATE_RESVALS: + ev->type = XCE_CORE_EVENT_RESVALS_DELTA; + ev->u.resvals_delta.deltas = req->u.resvalupdate.resvals; + return 1; + + case XCE_SCHED_OP_JOB_RUSAGE: + if (req->u.job_rusage.rusages == NULL) { + WARN("%s JOB_RUSAGE: NULL rusages payload, skipping", __func__); + return 0; + } + ev->type = XCE_CORE_EVENT_JOB_RUSAGE; + ev->u.job_rusage.jobid = req->u.job_rusage.rusages->jobid; + ev->u.job_rusage.rusages = req->u.job_rusage.rusages; + return 1; + + default: + WARN("%s unrecognized pending-update op %d, skipping", __func__, + (int)req->op); + return 0; + } +} + +int xce_sched_driver_translate_pending_events(xce_sched_core_t* core) +{ + dataq_t* q; + data_item_t* it; + xce_core_event_t* events = NULL; + xce_sched_update_request_t** reqs = NULL; + int cap = 0; + int n = 0; + uint64_t now; + int push_rc; + int i; + + if ((core == NULL) || (core->pend_updates[0] == NULL) + || (core->pend_updates[1] == NULL)) { + return -EINVAL; + } + + /* Same dataq-swap pattern scheduler.c's sched_update() already uses + * (Req 5 AC 8): swap the two pend_updates queues under core->lock so + * the handler-thread enqueue path (dataq_enq into pend_updates[0], + * unchanged by this task) never contends with the drain below -- + * events keep accumulating into the now-swapped-in pend_updates[0] + * while this function drains what was collected up to this call. */ + mylock_lock(&(core->lock)); + q = core->pend_updates[0]; + core->pend_updates[0] = core->pend_updates[1]; + core->pend_updates[1] = q; + mylock_unlock(&(core->lock)); + + /* One timestamp for the whole drained batch -- see file header comment + * (all events in one drain are applied atomically as a single + * xce_core_push_events() call, so one "this drain happened at time T" + * stamp is more coherent than N near-identical per-record reads). */ + now = xce_core_now(core); + + while ((it = dataq_deq(q, NULL)) != NULL) { + xce_sched_update_request_t* req = ITPTR_TO_UPDATE_REQ(it); + xce_core_event_t ev; + + if (!translate_one_pending_update(core, req, now, &ev)) { + release_translated_payload(core, req, 0); + free(req); + continue; + } + + if (n == cap) { + int newcap = cap ? (cap * 2) : 16; + xce_core_event_t* nevents; + xce_sched_update_request_t** nreqs; + + nevents = (xce_core_event_t*)realloc(events, (size_t)newcap + * sizeof(*events)); + if (nevents != NULL) { + events = nevents; + } + + nreqs = (xce_sched_update_request_t**)realloc( + reqs, (size_t)newcap * sizeof(*reqs)); + if (nreqs != NULL) { + reqs = nreqs; + } + + if ((nevents == NULL) || (nreqs == NULL)) { + ERROR("%s out of memory growing translation batch (n=%d)", + __func__, n); + + release_translated_payload(core, req, 0); + free(req); + + for (i = 0; i < n; i++) { + release_translated_payload(core, reqs[i], 0); + free(reqs[i]); + } + + free(events); + free(reqs); + + return -ENOMEM; + } + + cap = newcap; + } + + events[n] = ev; + reqs[n] = req; + n++; + } + + if (n == 0) { + return 0; + } + + push_rc = xce_core_push_events(core, events, n); + + for (i = 0; i < n; i++) { + release_translated_payload(core, reqs[i], (push_rc >= 0)); + free(reqs[i]); + } + + free(events); + free(reqs); + + return (push_rc < 0) ? push_rc : n; +} diff --git a/src/scheduler/testsuite/meson.build b/src/scheduler/testsuite/meson.build index 3d9dda99..0abf12e3 100644 --- a/src/scheduler/testsuite/meson.build +++ b/src/scheduler/testsuite/meson.build @@ -296,6 +296,25 @@ sched_core_full_cycle_test = executable( dependencies: [scheduler_test_deps], install: false) +# Contract test for the normal-mode driver's controller event translation +# path (scheduler-core spec, task 5.1, Req 5 AC 8 / AC 9's translation +# half): xce_sched_driver_translate_pending_events() drains core->pend_updates +# (the same dataq-swap mechanism handler.c already populates and +# scheduler.c's sched_update() already drains), translates each dequeued +# xce_sched_update_request_t into the matching xce_core_event_t for all 10 +# op types, and pushes the batch through a genuine xce_core_push_events() +# call. Same white-box #include pattern as test_core_events.c (pulls the +# real ../src/core.c and ../src/core_events.c into this TU) PLUS the new +# ../src/driver.c under test -- see that test's own header comment for why +# it provides its own real sched_resolve_job_picker() rather than +# #include-ing schedobj.c. +sched_driver_translate_test = executable( + 'sched_driver_translate_test', + ['test_driver_translate.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', @@ -344,3 +363,6 @@ test('scheduler core clock test', test('scheduler core full cycle test', sched_core_full_cycle_test, suite: 'scheduler', env: ['ASAN_OPTIONS=detect_leaks=0']) +test('scheduler driver translate test', + sched_driver_translate_test, suite: 'scheduler', + env: ['ASAN_OPTIONS=detect_leaks=0']) diff --git a/src/scheduler/testsuite/test_driver_translate.c b/src/scheduler/testsuite/test_driver_translate.c new file mode 100644 index 00000000..52fa88c5 --- /dev/null +++ b/src/scheduler/testsuite/test_driver_translate.c @@ -0,0 +1,751 @@ +/* + * test_driver_translate.c - contract test for the normal-mode driver's + * controller event translation path (scheduler-core spec, task 5.1, Req 5 + * AC 8 / AC 9's translation half). + * + * Observable under test: a sequence of synthetic xce_sched_update_request_t + * records enqueued into a real dataq (the same pend_updates mechanism + * handler.c already populates from capnp push/notify, unchanged by this + * task), once drained by xce_sched_driver_translate_pending_events(), + * produces an xce_core_event_t batch whose contents match the input + * records one-to-one (same count, same types, same order, same field + * values) -- and xce_core_push_events() genuinely gets called with that + * batch, proven via its REAL, observable side effects (core state + * mutation), not a mocked stand-in. + * + * White-box #include of ../src/core.c, ../src/core_clock.c (needed for + * xce_core_now(), which driver.c calls to stamp each translated batch), + * ../src/core_events.c, and ../src/driver.c (same convention as + * test_core_events.c) so this test + * exercises the REAL xce_sched_core_init()/xce_core_push_events()/ + * xce_sched_driver_translate_pending_events(), and can additionally call + * driver.c's static translate_one_pending_update() directly for tight, + * per-op-type field assertions on the translation step in isolation from + * the batch/push machinery around it. + * + * No picker plugin is registered in this file's fixture (unlike + * test_core_events.c's g_full_plugin): core->job_pick_plugin stays NULL, + * so sched_resolve_job_picker() (this file's own real reimplementation, + * copied from test_core_events.c's identical rationale -- schedobj.c is not + * pulled in) returns -EINVAL, xce_core_push_events() WARNs and proceeds + * with picker=NULL. Every case in core_events.c already NULL-guards its + * picker hook call, so this is safe -- and this file's Observable is core + * STATE mutation (proving the real push happened), not hook-firing, which + * task 4.3's own test suite already covers exhaustively. + * + * Stubs below (kill_sched_job/kill_sched_queue/kill_sched_worker/ + * kill_sched_user/kill_sched_project/kill_sched_resval/plugin_unregister) + * are pure no-ops, identical to test_core_events.c's own -- fixtures in + * this file are never torn down (this directory's established convention), + * so these killers are never actually invoked; present only to satisfy the + * linker for xce_sched_core_init()'s kvmgr_params_t.datakiller fields. + */ +#include +#include +#include +#include + +#include + +#include "controller.capnp.h" +#include "kvmgr.h" +#include "scheduler.h" +#include "xstring.h" + +/* Pull in the real xce_sched_core_init()/xce_core_push_events()/ + * xce_sched_driver_translate_pending_events() under test into this TU. */ +#include "../src/core.c" +#include "../src/core_clock.c" +#include "../src/core_events.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; } + +/* 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) +{ + 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; } + +/* ---------------------------------------------------------------------- */ +/* Fixture: real xce_sched_core_init() + manually dataq_init'd */ +/* pend_updates (core.c's own header comment: dataq_init for pend_updates */ +/* stays driver/scheduler.c-owned, not xce_sched_core_init()'s job) -- */ +/* same pattern test_handlers.c's fixture_init() uses. Never torn down. */ +/* ---------------------------------------------------------------------- */ + +static void fixture_init(xce_sched_core_t* core) +{ + int rc; + + rc = xce_sched_core_init(core); + TEST_ASSERT_EQUAL_INT(0, rc); + + core->pend_updates[0] = (dataq_t*)calloc(1, sizeof(dataq_t)); + dataq_init(core->pend_updates[0], NULL); + core->pend_updates[1] = (dataq_t*)calloc(1, sizeof(dataq_t)); + dataq_init(core->pend_updates[1], NULL); +} + +static void enqueue(xce_sched_core_t* core, xce_sched_update_request_t* req) +{ + dataq_enq(core->pend_updates[0], &(req->link)); +} + +static xce_sched_update_request_t* make_req(xce_sched_op_t op) +{ + xce_sched_update_request_t* req + = (xce_sched_update_request_t*)calloc(1, sizeof(*req)); + + req->op = op; + return req; +} + +/* Every string field here is strdup()'d, and the wrapper structs below are + * all individually calloc()'d (never a stack local or a shared array) -- + * matching the real shape release_translated_payload()'s free_Job_ptr() etc. + * (capnp-generated, controller.capnp.c) expect to free: every free_X_ptr() + * in this codebase does `free(field); ...; free(*d);` unconditionally on + * every non-NULL string/wrapper it owns. A string literal or stack address + * here would crash the moment the real cleanup path (this task's own new + * code, exercised for real by xce_sched_driver_translate_pending_events()) + * tries to free it -- this is what a genuinely capnp-decoded xce_job_t + * looks like in production, not a test shortcut. */ +static xce_job_t* make_wire_job(uint64_t jobid, uint32_t state, + const char* queue, const char* user, + const 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->submit = (job_submit_data_t*)calloc(1, sizeof(*(j->submit))); + j->submit->queue = strdup(queue); + j->submit->user = strdup(user); + j->submit->project_name = strdup(project); + j->submit->resreq = strdup("type == any"); + j->submit->mintasks = 1; + j->submit->maxtasks = 1; + + return j; +} + +/* ---------------------------------------------------------------------- */ +/* Per-op unit tests of translate_one_pending_update() (white-box static */ +/* call) -- one per op type, asserting event type + every field matches. */ +/* ---------------------------------------------------------------------- */ + +static void test_translate_config_push(void) +{ + xce_sched_core_t core; + xce_conf4sched_t config; + xce_sched_update_request_t* req = make_req(XCE_SCHED_OP_UPDATE_CONFIG); + xce_core_event_t ev; + int rc; + + fixture_init(&core); + memset(&config, 0, sizeof(config)); + req->u.config = &config; + + rc = translate_one_pending_update(&core, req, 12345, &ev); + + TEST_ASSERT_EQUAL_INT(1, rc); + TEST_ASSERT_EQUAL_INT(XCE_CORE_EVENT_CONFIG_PUSH, ev.type); + TEST_ASSERT_EQUAL_UINT64(12345, ev.ts); + TEST_ASSERT_EQUAL_PTR(&config, ev.u.config_push.config); +} + +static void test_translate_jobs_push(void) +{ + xce_sched_core_t core; + xce_job_t* job1 = make_wire_job(1, JOB_STAT_PEND, "normal", "a", "p"); + xce_job_t* jobs[1]; + xce_sched_update_request_t* req = make_req(XCE_SCHED_OP_UPDATE_JOB); + xce_core_event_t ev; + int rc; + + fixture_init(&core); + jobs[0] = job1; + req->u.jobupdate.njobs = 1; + req->u.jobupdate.jobs = jobs; + + rc = translate_one_pending_update(&core, req, 99, &ev); + + TEST_ASSERT_EQUAL_INT(1, rc); + TEST_ASSERT_EQUAL_INT(XCE_CORE_EVENT_JOBS_PUSH, ev.type); + TEST_ASSERT_EQUAL_UINT64(99, ev.ts); + TEST_ASSERT_EQUAL_INT(1, ev.u.jobs_push.njobs); + TEST_ASSERT_EQUAL_PTR(jobs, ev.u.jobs_push.jobs); + TEST_ASSERT_EQUAL_PTR(job1, ev.u.jobs_push.jobs[0]); +} + +static void test_translate_job_fields_update(void) +{ + xce_sched_core_t core; + xce_job_t* job = make_wire_job(7, JOB_STAT_RUN, "normal", "a", "p"); + xce_sched_update_request_t* req = make_req(XCE_SCHED_OP_UPDATE_JOB_FIELDS); + xce_core_event_t ev; + int rc; + + fixture_init(&core); + req->u.jobfieldsupdate.job = job; + + rc = translate_one_pending_update(&core, req, 5, &ev); + + TEST_ASSERT_EQUAL_INT(1, rc); + TEST_ASSERT_EQUAL_INT(XCE_CORE_EVENT_JOB_UPDATE, ev.type); + TEST_ASSERT_EQUAL_UINT64(7, ev.u.job_update.jobid); + TEST_ASSERT_EQUAL_PTR(job, ev.u.job_update.job); + TEST_ASSERT_EQUAL_UINT32(0, ev.u.job_update.changed_fields); +} + +static void test_translate_job_fields_update_null_job_skips(void) +{ + xce_sched_core_t core; + xce_sched_update_request_t* req = make_req(XCE_SCHED_OP_UPDATE_JOB_FIELDS); + xce_core_event_t ev; + int rc; + + fixture_init(&core); + req->u.jobfieldsupdate.job = NULL; + + rc = translate_one_pending_update(&core, req, 5, &ev); + + TEST_ASSERT_EQUAL_INT(0, rc); +} + +static void test_translate_workers_push(void) +{ + xce_sched_core_t core; + xce_worker_t wire_worker; + xce_worker_t* workers[1]; + xce_sched_update_request_t* req = make_req(XCE_SCHED_OP_UPDATE_WORKER); + xce_core_event_t ev; + int rc; + + fixture_init(&core); + memset(&wire_worker, 0, sizeof(wire_worker)); + wire_worker.name = "hostA"; + workers[0] = &wire_worker; + req->u.workerupdate.nworkers = 1; + req->u.workerupdate.workers = workers; + + rc = translate_one_pending_update(&core, req, 3, &ev); + + TEST_ASSERT_EQUAL_INT(1, rc); + TEST_ASSERT_EQUAL_INT(XCE_CORE_EVENT_WORKERS_PUSH, ev.type); + TEST_ASSERT_EQUAL_INT(1, ev.u.workers_push.nworkers); + TEST_ASSERT_EQUAL_PTR(workers, ev.u.workers_push.workers); +} + +static void test_translate_jstate_change_looks_up_old_state(void) +{ + xce_sched_core_t core; + xce_sched_job_t existing_job; + xce_sched_update_request_t* req = make_req(XCE_SCHED_OP_JSTATE_CHANGE); + xce_core_event_t ev; + int rc; + + fixture_init(&core); + memset(&existing_job, 0, sizeof(existing_job)); + existing_job.jobid = 42; + existing_job.state = JOB_STAT_PEND; + sched_add_job(&core, &existing_job); + + req->u.job_state_change.jobid = 42; + req->u.job_state_change.state = JOB_STAT_RUN; + + rc = translate_one_pending_update(&core, req, 8, &ev); + + TEST_ASSERT_EQUAL_INT(1, rc); + TEST_ASSERT_EQUAL_INT(XCE_CORE_EVENT_JOB_STATE_CHANGE, ev.type); + TEST_ASSERT_EQUAL_UINT64(42, ev.u.job_state_change.jobid); + TEST_ASSERT_EQUAL_UINT32(JOB_STAT_PEND, ev.u.job_state_change.old_state); + TEST_ASSERT_EQUAL_UINT32(JOB_STAT_RUN, ev.u.job_state_change.new_state); + TEST_ASSERT_EQUAL_INT32(0, ev.u.job_state_change.exit_code); +} + +static void test_translate_jstate_change_unknown_job_old_state_zero(void) +{ + xce_sched_core_t core; + xce_sched_update_request_t* req = make_req(XCE_SCHED_OP_JSTATE_CHANGE); + xce_core_event_t ev; + int rc; + + fixture_init(&core); + req->u.job_state_change.jobid = 999; + req->u.job_state_change.state = JOB_STAT_EXIT; + + rc = translate_one_pending_update(&core, req, 8, &ev); + + TEST_ASSERT_EQUAL_INT(1, rc); + TEST_ASSERT_EQUAL_UINT32(0, ev.u.job_state_change.old_state); + TEST_ASSERT_EQUAL_UINT32(JOB_STAT_EXIT, ev.u.job_state_change.new_state); +} + +static void test_translate_queue_update(void) +{ + xce_sched_core_t core; + xce_queue_t wire_queue; + xce_sched_update_request_t* req = make_req(XCE_SCHED_OP_UPDATE_QUEUE); + xce_core_event_t ev; + int rc; + + fixture_init(&core); + memset(&wire_queue, 0, sizeof(wire_queue)); + wire_queue.queue = "batch"; + req->u.queueupdate.queue = &wire_queue; + + rc = translate_one_pending_update(&core, req, 1, &ev); + + TEST_ASSERT_EQUAL_INT(1, rc); + TEST_ASSERT_EQUAL_INT(XCE_CORE_EVENT_QUEUE_UPDATE, ev.type); + TEST_ASSERT_EQUAL_PTR(&wire_queue, ev.u.queue_update.queue); +} + +static void test_translate_resources_push(void) +{ + xce_sched_core_t core; + xce_resource_t wire_resource; + xce_resource_t* resources[1]; + xce_sched_update_request_t* req = make_req(XCE_SCHED_OP_UPDATE_RESOURCES); + xce_core_event_t ev; + int rc; + + fixture_init(&core); + memset(&wire_resource, 0, sizeof(wire_resource)); + wire_resource.name = "cpu"; + resources[0] = &wire_resource; + req->u.resourceupdate.nresources = 1; + req->u.resourceupdate.resources = resources; + + rc = translate_one_pending_update(&core, req, 2, &ev); + + TEST_ASSERT_EQUAL_INT(1, rc); + TEST_ASSERT_EQUAL_INT(XCE_CORE_EVENT_RESOURCES_PUSH, ev.type); + TEST_ASSERT_EQUAL_INT(1, ev.u.resources_push.n); + TEST_ASSERT_EQUAL_PTR(resources, ev.u.resources_push.resources); +} + +static void test_translate_resvals_full(void) +{ + xce_sched_core_t core; + xce_val_t val; + xce_resval_t wire_resval; + xce_resval_t* vals[1]; + xce_sched_update_request_t* req + = make_req(XCE_SCHED_OP_UPDATE_RESVALS_FULL); + xce_core_event_t ev; + int rc; + + fixture_init(&core); + memset(&val, 0, sizeof(val)); + memset(&wire_resval, 0, sizeof(wire_resval)); + wire_resval.name = "cpu"; + wire_resval.val = &val; + vals[0] = &wire_resval; + req->u.resvalfullupdate.nvals = 1; + req->u.resvalfullupdate.vals = vals; + + rc = translate_one_pending_update(&core, req, 6, &ev); + + TEST_ASSERT_EQUAL_INT(1, rc); + TEST_ASSERT_EQUAL_INT(XCE_CORE_EVENT_RESVALS_FULL, ev.type); + TEST_ASSERT_EQUAL_INT(1, ev.u.resvals_full.n); + TEST_ASSERT_EQUAL_PTR(vals, ev.u.resvals_full.resvals); +} + +static void test_translate_resvals_delta(void) +{ + xce_sched_core_t core; + xce_resval_list_t deltas; + xce_sched_update_request_t* req = make_req(XCE_SCHED_OP_UPDATE_RESVALS); + xce_core_event_t ev; + int rc; + + fixture_init(&core); + memset(&deltas, 0, sizeof(deltas)); + req->u.resvalupdate.resvals = &deltas; + + rc = translate_one_pending_update(&core, req, 4, &ev); + + TEST_ASSERT_EQUAL_INT(1, rc); + TEST_ASSERT_EQUAL_INT(XCE_CORE_EVENT_RESVALS_DELTA, ev.type); + TEST_ASSERT_EQUAL_PTR(&deltas, ev.u.resvals_delta.deltas); +} + +static void test_translate_job_rusage(void) +{ + xce_sched_core_t core; + xce_job_rusages_t rusages; + xce_sched_update_request_t* req = make_req(XCE_SCHED_OP_JOB_RUSAGE); + xce_core_event_t ev; + int rc; + + fixture_init(&core); + memset(&rusages, 0, sizeof(rusages)); + rusages.jobid = 55; + req->u.job_rusage.rusages = &rusages; + + rc = translate_one_pending_update(&core, req, 9, &ev); + + TEST_ASSERT_EQUAL_INT(1, rc); + TEST_ASSERT_EQUAL_INT(XCE_CORE_EVENT_JOB_RUSAGE, ev.type); + TEST_ASSERT_EQUAL_UINT64(55, ev.u.job_rusage.jobid); + TEST_ASSERT_EQUAL_PTR(&rusages, ev.u.job_rusage.rusages); +} + +static void test_translate_job_rusage_null_rusages_skips(void) +{ + xce_sched_core_t core; + xce_sched_update_request_t* req = make_req(XCE_SCHED_OP_JOB_RUSAGE); + xce_core_event_t ev; + int rc; + + fixture_init(&core); + req->u.job_rusage.rusages = NULL; + + rc = translate_one_pending_update(&core, req, 9, &ev); + + TEST_ASSERT_EQUAL_INT(0, rc); +} + +static void test_translate_unknown_op_skips(void) +{ + xce_sched_core_t core; + xce_sched_update_request_t* req = make_req((xce_sched_op_t)999); + xce_core_event_t ev; + int rc; + + fixture_init(&core); + + rc = translate_one_pending_update(&core, req, 9, &ev); + + TEST_ASSERT_EQUAL_INT(0, rc); +} + +/* ---------------------------------------------------------------------- */ +/* End-to-end: enqueue all 10 op types into a real dataq, drain via the */ +/* public entry point, assert real core-state mutation for every one -- */ +/* proves xce_core_push_events() genuinely ran with the translated batch. */ +/* ---------------------------------------------------------------------- */ + +static void test_drain_translates_and_pushes_all_ten_op_types(void) +{ + xce_sched_core_t core; + xce_conf4sched_t* config; + xce_job_t** jobs_batch; + xce_job_t* job1; + xce_job_t* job_fields_payload; + xce_worker_t* wire_worker; + xce_worker_t** workers; + xce_queue_t* wire_queue; + xce_resource_t* wire_resource; + xce_resource_t** resources; + xce_val_t* full_val; + xce_resval_t* full_resval; + xce_resval_t** full_vals; + xce_val_t* delta_val; + xce_resval_t* delta_resval; + xce_resval_t** delta_vals; + xce_resval_list_t* deltas; + xce_job_rusages_t rusages; + int rc; + + fixture_init(&core); + + /* Every payload below is individually calloc()'d (wrapper structs, + * arrays, and strings via strdup()) -- see make_wire_job()'s own + * comment for why: release_translated_payload() (driver.c) genuinely + * calls the real capnp-generated free_X_ptr() on each of these after + * translation, exactly mirroring scheduler.c's sched_pend_update_killer() + * in production. A stack local or string literal here would crash the + * moment that real cleanup path runs -- this is what genuinely + * capnp-decoded wire payloads look like, not a test shortcut. */ + + /* 1. UPDATE_JOB -> JOBS_PUSH: creates job 42 (needed by later ops). */ + job1 = make_wire_job(42, JOB_STAT_PEND, "normal", "alice", "proj1"); + jobs_batch = (xce_job_t**)calloc(1, sizeof(*jobs_batch)); + jobs_batch[0] = job1; + xce_sched_update_request_t* r1 = make_req(XCE_SCHED_OP_UPDATE_JOB); + r1->u.jobupdate.njobs = 1; + r1->u.jobupdate.jobs = jobs_batch; + enqueue(&core, r1); + + /* 2. UPDATE_WORKER -> WORKERS_PUSH */ + wire_worker = (xce_worker_t*)calloc(1, sizeof(*wire_worker)); + wire_worker->name = strdup("hostA"); + workers = (xce_worker_t**)calloc(1, sizeof(*workers)); + workers[0] = wire_worker; + xce_sched_update_request_t* r2 = make_req(XCE_SCHED_OP_UPDATE_WORKER); + r2->u.workerupdate.nworkers = 1; + r2->u.workerupdate.workers = workers; + enqueue(&core, r2); + + /* 3. UPDATE_RESOURCES -> RESOURCES_PUSH: registers "cpu" (needed by 4). + * makeResource() (resobj.c) deep-copies name/desc via makeString() -- + * verified by reading resobj_init()'s source -- so freeing + * wire_resource afterward never dangles the resmgr's own copy. */ + wire_resource = (xce_resource_t*)calloc(1, sizeof(*wire_resource)); + wire_resource->name = strdup("cpu"); + resources = (xce_resource_t**)calloc(1, sizeof(*resources)); + resources[0] = wire_resource; + xce_sched_update_request_t* r3 = make_req(XCE_SCHED_OP_UPDATE_RESOURCES); + r3->u.resourceupdate.nresources = 1; + r3->u.resourceupdate.resources = resources; + enqueue(&core, r3); + + /* 4. UPDATE_RESVALS_FULL -> RESVALS_FULL. NOTE: core_events.c's + * RESVALS_FULL case does `zarray_put(&core->sharedInst, res->idx, + * rv->val)` -- a SHALLOW store of the SAME `val` pointer (verified by + * reading zarray_put_ex()'s source, src/lib/zarray.c) -- and this + * function's cleanup then frees `full_resval` (and, via + * free_ResourceVal->free_Value_ptr, `full_val` too), exactly mirroring + * scheduler.c's existing sched_pend_update_killer()'s identical + * RESVALS_FULL cleanup. This is a PRE-EXISTING production ownership + * characteristic (the same hazard already exists in sched_update()'s + * own live path today, untouched by this task) -- so after the drain + * below, core->sharedInst's stored pointer for "cpu" is only safe to + * check for non-NULL (the slot was populated), never to dereference. */ + full_val = (xce_val_t*)calloc(1, sizeof(*full_val)); + full_val->u.ivalue32 = 16; + full_resval = (xce_resval_t*)calloc(1, sizeof(*full_resval)); + full_resval->name = strdup("cpu"); + full_resval->val = full_val; + full_vals = (xce_resval_t**)calloc(1, sizeof(*full_vals)); + full_vals[0] = full_resval; + xce_sched_update_request_t* r4 = make_req(XCE_SCHED_OP_UPDATE_RESVALS_FULL); + r4->u.resvalfullupdate.nvals = 1; + r4->u.resvalfullupdate.vals = full_vals; + enqueue(&core, r4); + + /* 5. UPDATE_RESVALS -> RESVALS_DELTA: attributes to worker "hostA". + * Same shallow-store-then-free hazard as 4 above (per-worker resvals + * zarray instead of core->sharedInst) -- same "non-NULL only" caveat. */ + delta_val = (xce_val_t*)calloc(1, sizeof(*delta_val)); + delta_val->u.ivalue32 = 4; + delta_resval = (xce_resval_t*)calloc(1, sizeof(*delta_resval)); + delta_resval->name = strdup("cpu"); + delta_resval->host = strdup("hostA"); + delta_resval->val = delta_val; + delta_vals = (xce_resval_t**)calloc(1, sizeof(*delta_vals)); + delta_vals[0] = delta_resval; + deltas = (xce_resval_list_t*)calloc(1, sizeof(*deltas)); + deltas->count = 1; + deltas->vals = delta_vals; + xce_sched_update_request_t* r5 = make_req(XCE_SCHED_OP_UPDATE_RESVALS); + r5->u.resvalupdate.resvals = deltas; + enqueue(&core, r5); + + /* 6. UPDATE_QUEUE -> QUEUE_UPDATE. core_build_queue() deep-copies the + * queue name via makeString() and priority is a plain int copy, so + * freeing wire_queue afterward never dangles the core's own queue + * object -- safe to assert on post-drain. */ + wire_queue = (xce_queue_t*)calloc(1, sizeof(*wire_queue)); + wire_queue->queue = strdup("batch"); + wire_queue->priority = 9; + wire_queue->status = QUEUE_STAT_OPEN; + xce_sched_update_request_t* r6 = make_req(XCE_SCHED_OP_UPDATE_QUEUE); + r6->u.queueupdate.queue = wire_queue; + enqueue(&core, r6); + + /* 7. UPDATE_JOB_FIELDS -> JOB_UPDATE: mutates job 42's resreq. */ + job_fields_payload + = make_wire_job(42, JOB_STAT_PEND, "normal", "alice", "proj1"); + xce_sched_update_request_t* r7 = make_req(XCE_SCHED_OP_UPDATE_JOB_FIELDS); + r7->u.jobfieldsupdate.job = job_fields_payload; + enqueue(&core, r7); + + /* 8. JSTATE_CHANGE -> JOB_STATE_CHANGE: job 42 PEND -> RUN. */ + xce_sched_update_request_t* r8 = make_req(XCE_SCHED_OP_JSTATE_CHANGE); + r8->u.job_state_change.jobid = 42; + r8->u.job_state_change.state = JOB_STAT_RUN; + enqueue(&core, r8); + + /* 9. JOB_RUSAGE -> JOB_RUSAGE: job 42 (already pushed by #1, in the same + * batch, in FIFO order) adopts `rusages` by pointer + * (core_events.c: `job->rusages = rusages`) -- release_translated_payload() + * detects the adoption via job->rusages == rusages and does NOT free it, + * so a stack local is safe here (unlike every payload above). */ + memset(&rusages, 0, sizeof(rusages)); + rusages.jobid = 42; + xce_sched_update_request_t* r9 = make_req(XCE_SCHED_OP_JOB_RUSAGE); + r9->u.job_rusage.rusages = &rusages; + enqueue(&core, r9); + + /* 10. UPDATE_CONFIG -> CONFIG_PUSH (dead in the live enqueue path today + * -- handler.c never creates this op -- but the translation must still + * be real and correct per this task's scope). core->config adopts this + * pointer permanently on a successful push -- never freed -- so it must + * outlive this function's return; heap-allocated for that reason. */ + config = (xce_conf4sched_t*)calloc(1, sizeof(*config)); + xce_sched_update_request_t* r10 = make_req(XCE_SCHED_OP_UPDATE_CONFIG); + r10->u.config = config; + enqueue(&core, r10); + + rc = xce_sched_driver_translate_pending_events(&core); + + TEST_ASSERT_EQUAL_INT(10, rc); + + /* Real, observable core-state mutation proves xce_core_push_events() + * genuinely ran with the translated batch -- not a mocked call. */ + TEST_ASSERT_NOT_NULL(sched_get_job(&core, 42)); + TEST_ASSERT_EQUAL_UINT32(JOB_STAT_RUN, sched_get_job(&core, 42)->state); + TEST_ASSERT_EQUAL_PTR(&rusages, sched_get_job(&core, 42)->rusages); + + TEST_ASSERT_NOT_NULL(sched_get_worker(&core, "hostA")); + + { + xce_resource_t* found = NULL; + int frc = key_find(core.resmgr, (void*)"cpu", (void**)&found); + TEST_ASSERT_GREATER_THAN_INT(0, frc); + TEST_ASSERT_NOT_NULL(found); + + /* Slot-populated check only -- see comment on payload #4/#5 above + * for why the stored pointers are not safe to dereference here. */ + TEST_ASSERT_NOT_NULL(zarray_get(&(core.sharedInst), found->idx)); + + { + xce_sched_worker_t* w = sched_get_worker(&core, "hostA"); + TEST_ASSERT_NOT_NULL(zarray_get(&(w->resvals), found->idx)); + } + } + + { + xce_sched_queue_t* q = sched_get_queue(&core, "batch"); + TEST_ASSERT_NOT_NULL(q); + TEST_ASSERT_EQUAL_INT(9, q->priority); + } + + TEST_ASSERT_EQUAL_PTR(config, core.config); +} + +/* ---------------------------------------------------------------------- */ +/* Buffering discipline (Req 5 AC 8): events accumulate between drains; */ +/* an empty drain is a valid no-op returning 0, not an error. */ +/* ---------------------------------------------------------------------- */ + +static void test_accumulates_between_drains_and_empty_drain_is_noop(void) +{ + xce_sched_core_t core; + int rc; + + fixture_init(&core); + + /* Nothing pending yet. */ + rc = xce_sched_driver_translate_pending_events(&core); + TEST_ASSERT_EQUAL_INT(0, rc); + + /* One record enqueued -- drained on the next call. Heap-allocated (see + * make_wire_job()'s comment above): the real drain call below genuinely + * frees this payload via free_Worker_ptr(). */ + { + xce_worker_t* wire_worker + = (xce_worker_t*)calloc(1, sizeof(*wire_worker)); + xce_worker_t** workers = (xce_worker_t**)calloc(1, sizeof(*workers)); + xce_sched_update_request_t* req = make_req(XCE_SCHED_OP_UPDATE_WORKER); + + wire_worker->name = strdup("hostB"); + workers[0] = wire_worker; + req->u.workerupdate.nworkers = 1; + req->u.workerupdate.workers = workers; + enqueue(&core, req); + + rc = xce_sched_driver_translate_pending_events(&core); + TEST_ASSERT_EQUAL_INT(1, rc); + TEST_ASSERT_NOT_NULL(sched_get_worker(&core, "hostB")); + } + + /* Nothing new pending -- must not re-process anything. */ + rc = xce_sched_driver_translate_pending_events(&core); + TEST_ASSERT_EQUAL_INT(0, rc); +} + +/* ---------------------------------------------------------------------- */ +/* Bad-argument guard clauses. */ +/* ---------------------------------------------------------------------- */ + +static void test_null_core_returns_einval(void) +{ + int rc = xce_sched_driver_translate_pending_events(NULL); + TEST_ASSERT_EQUAL_INT(-EINVAL, rc); +} + +static void test_uninitialized_pend_updates_returns_einval(void) +{ + xce_sched_core_t core; + int rc; + + memset(&core, 0, sizeof(core)); + rc = xce_sched_driver_translate_pending_events(&core); + + TEST_ASSERT_EQUAL_INT(-EINVAL, rc); +} + +int main(void) +{ + UNITY_BEGIN(); + RUN_TEST(test_translate_config_push); + RUN_TEST(test_translate_jobs_push); + RUN_TEST(test_translate_job_fields_update); + RUN_TEST(test_translate_job_fields_update_null_job_skips); + RUN_TEST(test_translate_workers_push); + RUN_TEST(test_translate_jstate_change_looks_up_old_state); + RUN_TEST(test_translate_jstate_change_unknown_job_old_state_zero); + RUN_TEST(test_translate_queue_update); + RUN_TEST(test_translate_resources_push); + RUN_TEST(test_translate_resvals_full); + RUN_TEST(test_translate_resvals_delta); + RUN_TEST(test_translate_job_rusage); + RUN_TEST(test_translate_job_rusage_null_rusages_skips); + RUN_TEST(test_translate_unknown_op_skips); + RUN_TEST(test_drain_translates_and_pushes_all_ten_op_types); + RUN_TEST(test_accumulates_between_drains_and_empty_drain_is_noop); + RUN_TEST(test_null_core_returns_einval); + RUN_TEST(test_uninitialized_pend_updates_returns_einval); + return UNITY_END(); +}