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

feat(scheduler-core): build normal-mode driver event translation path (task 5.1)

Browse source 
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 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_019dw5WnN3DXStaEaPYfACU1
This commit is contained in:
Rongsong Shen 2026-08-04 04:39:17 +00:00
parent 19d8df3532
commit dceafc6b12
Signed by: rshen
SSH key fingerprint: SHA256:8Vf2pZ3h6rJa7EhGK+WaAQ9Mta/apZvteW6g9U/WKKk
6 changed files with 1217 additions and 1 deletions
Download patch file Download diff file Expand all files Collapse all files

3
.kiro/specs/scheduler-core/tasks.md
View file

@ -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

55
src/scheduler/include/driver.h Normal file
View file

@ -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

1
src/scheduler/meson.build
View file

@ -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',

386
src/scheduler/src/driver.c Normal file
View file

@ -0,0 +1,386 @@
#include <errno.h>
#include <stdlib.h>
#include <string.h>
#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;
}

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

@ -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'])

751
src/scheduler/testsuite/test_driver_translate.c Normal file
View file

@ -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 <errno.h>
#include <stdint.h>
#include <stdlib.h>
#include <string.h>
#include <unity/unity.h>
#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();
}