lei.chen/myxce
1
0
Fork 1
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions
myxce/src/scheduler/testsuite/test_core_events.c
Rongsong Shen 0e78f86377
fix(bug-1738,bug-1739): terminal-state regression + phantom RUN in JOBS_PUSH merge 
bug-1738: a scheduler-side notify could clobber an already-terminal
job state. The controller's own terminal-transition guard
(job_state_is_terminal(), job.c) treats DONE/EXIT as permanent once
set -- apply_job_fail_transition()'s POST_EXEC/Q_POST_EXEC branch only
ever MERGES a JOB_STAT_PERR flag onto whatever DONE/EXIT already
holds, never re-deciding DONE vs EXIT after the fact. The scheduler
had no equivalent guard: HOOK_DETAIL_JOB_FAIL's schedFailJob notify
(sched_job_fail_fun, sched.c) unconditionally requests JOB_STAT_EXIT
with no regard for the job's current state, so a job that finished
successfully (DONE) and then hit an unrelated post-exec script failure
would have the scheduler's view forcibly clobbered back to EXIT.
Fixed with one shared helper, job_state_regresses_terminal(), applied
at both XCE_CORE_EVENT_JOB_STATE_CHANGE and the JOBS_PUSH duplicate-
jobid resync merge (a full resync can just as easily redeliver a stale
snapshot as a genuinely newer one) -- once a job is DONE/EXIT, neither
path can move it to a different terminal state.

bug-1739 (Symptom 1, "phantom RUN"): the bug-1734 fix's own JOBS_PUSH
merge adopted a redelivery's RUN state without transferring the real
allocation data core_build_job() had already attached to the
redelivered clone -- leaving existing->state==RUN with
existing->allocs==NULL, a job the scheduler's own resource/slot
bookkeeping has no record of. Live-reproduced: `info summary` showed
30 elements of an array marked RUN, but `info worker <name>` only
listed 22 real slot assignments for a 22-core worker. Fixed by
transferring job->accounts/job->allocs onto the existing job on a RUN
merge (freeing whatever it previously held), and refusing the merge
entirely (WARN-logged, existing state kept) if the redelivery carries
no allocation data -- treating an unbacked RUN claim as untrustworthy
rather than manufacturing a phantom.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_019QxaSzHRB7pmYFp4cVnwkA
2026-08-31 04:20:46 +00:00

1422 lines
53 KiB
C
Raw Blame History

/*
* test_core_events.c - contract test for the core event interface and
* batch event push (scheduler-core spec, task 4.3, Req 5 AC 4 / AC 8).
*
* Observable under test: pushing a batch of events with the picker idle
* (core->session_active == 0) fires the corresponding state-change hook
* exactly once per event, synchronously, before xce_core_push_events()
* returns -- and the session-active rejection path (-EBUSY, no event
* applied) when a session IS active.
*
* White-box #include of both ../src/core.c and ../src/core_events.c (same
* convention as test_core_isolation.c / test_core_jbuf_capacity_pin.c) so
* this test exercises the REAL xce_sched_core_init()/xce_core_push_events(),
* not a re-implementation -- xce_core_push_events() lives in core_events.c
* (split out of core.c purely to respect the 800-line-per-file cap; see
* that file's own header comment), which itself calls straight back into
* core.c's real sched_get_job()/sched_add_job()/etc. accessors, all within
* this one translation unit. Stubs below are only for scheduler-internal
* externs these two files reference (as kvmgr datakiller callback
* addresses, or -- for sched_resolve_job_picker -- because the real
* implementation lives in schedobj.c, a separate translation unit this
* file does not pull in):
*
* - kill_sched_job/kill_sched_queue/kill_sched_worker/kill_sched_user/
* kill_sched_project/kill_sched_resval/plugin_unregister: pure no-op
* stubs, identical to test_core_jbuf_capacity_pin.c's own -- this
* file's fixtures are never torn down (never call
* xce_sched_core_fini()), same "fixtures aren't torn down" convention
* established across this directory, so these killers are never
* actually invoked; present only to satisfy the linker for
* xce_sched_core_init()'s kvmgr_params_t.datakiller fields.
* - sched_resolve_job_picker: a real, working reimplementation of
* schedobj.c's function of the same name (identical algorithm --
* key_find core->job_pick_plugin in core->plugins, validate the
* picker MUST hooks) -- needed because the true implementation lives
* in schedobj.c, and pulling that whole file in would drag in its own
* large stub surface (plugin_queue_update, plugin_job_update, etc.)
* for functions this file's tests never exercise. This IS real logic,
* not a null stub, because the hook-firing contract under test
* depends on picker resolution actually working.
*
* make_job_account/free_job_account/free_JobAlloc_ptr/free_JobRusages_ptr/
* makeResource/parse_resreq/makeString are all real, linked from the
* shared libxce library (src/lib), not stubbed -- same as every other
* scheduler white-box test in this directory.
*/
#include <errno.h>
#include <stdint.h>
#include <stdlib.h>
#include <string.h>
#include <unity/unity.h>
#include "kvmgr.h"
#include "scheduler.h"
#include "xcealloc.h"
#include "xstring.h"
/* Pull in the real xce_sched_core_init()/xce_core_push_events() under test
* into this TU (white-box #include, same convention as
* test_core_jbuf_capacity_pin.c's `#include "../src/core.c"`). */
#include "../src/core.c"
#include "../src/core_events.c"
void kill_sched_job(void* arg) { (void)arg; }
void kill_sched_queue(void* arg) { (void)arg; }
void kill_sched_worker(void* arg) { (void)arg; }
void kill_sched_user(void* arg) { (void)arg; }
void kill_sched_project(void* arg) { (void)arg; }
void kill_sched_resval(xce_val_t* val) { (void)val; }
/* dup_val - real reimplementation of snapshot_wire.c's function of the same
* name (verbatim logic), needed because core_events.c's RESVALS_FULL/
* RESVALS_DELTA handlers now call it (bug-1585 fix) and this file's
* white-box #include of ../src/core_events.c does not pull in
* snapshot_wire.c (same "reimplement the one small function instead of the
* whole file" rationale this file's own sched_resolve_job_picker copy
* already documents). This file's own kill_sched_resval is a no-op stub, so
* the deep copy's contents are never inspected here -- only link-time
* presence matters for this file's own tests. */
xce_val_t* dup_val(const xce_val_t* src)
{
xce_val_t* v;
if (src == NULL) {
return NULL;
}
v = (xce_val_t*)calloc(1, sizeof(*v));
if (v == NULL) {
return NULL;
}
v->which = src->which;
if (src->which == XCE_VAL_STRING) {
v->u.svalue
= (src->u.svalue != NULL) ? makeString(src->u.svalue) : NULL;
} else {
v->u = src->u;
}
return v;
}
void plugin_unregister(void* arg) { (void)arg; }
/* Real reimplementation of schedobj.c's sched_resolve_job_picker -- see
* file header comment for why this file provides its own copy instead of
* #include-ing schedobj.c. */
int sched_resolve_job_picker(xce_sched_core_t* core, xce_sched_plugin_t** out)
{
xce_sched_plugin_t* plugin = NULL;
int rc = 0;
if ((core == NULL) || (out == NULL)) {
return -EINVAL;
}
*out = NULL;
if (core->job_pick_plugin == NULL) {
return -EINVAL;
}
lock(core->plugins, RTTI_LOCK_READ);
rc = key_find(core->plugins, core->job_pick_plugin, (void**)&plugin);
unlock(core->plugins);
if (rc <= 0) {
return (rc < 0) ? rc : -ENOENT;
}
if ((plugin == NULL) || (plugin->picker_iface == NULL)
|| (plugin->picker_iface->next_job == NULL)
|| (plugin->picker_iface->begin_session == NULL)
|| (plugin->picker_iface->end_session == NULL)
|| (plugin->picker_iface->on_dispatch == NULL)
|| (plugin->picker_iface->on_pend == NULL)) {
return -EINVAL;
}
*out = plugin;
return 0;
}
void setUp(void) { return; }
void tearDown(void) { return; }
/* ---------------------------------------------------------------------- */
/* Test picker: records call counts + last-seen arguments per hook, so */
/* tests can assert "exactly once, with the right argument" per event. */
/* ---------------------------------------------------------------------- */
typedef struct {
int config_calls;
int job_push_calls;
int job_state_change_calls;
int job_update_calls;
int queue_update_calls;
int job_rusage_calls;
xce_conf4sched_t* last_config;
xce_sched_job_t* last_job_push;
xce_sched_job_t* last_job_state_change_job;
uint32_t last_job_state_change_state;
xce_sched_job_t* last_job_update;
xce_sched_queue_t* last_queue_update;
xce_sched_job_t* last_job_rusage_job;
xce_job_rusages_t* last_job_rusage_rusages;
} test_picker_counts_t;
static test_picker_counts_t g_counts;
static int test_begin_session(void* picker)
{
(void)picker;
return 0;
}
static xce_sched_job_t* test_next_job(void* picker)
{
(void)picker;
return NULL;
}
static int test_end_session(void* picker)
{
(void)picker;
return 0;
}
static int test_on_dispatch(void* picker, uint64_t jobid)
{
(void)picker;
(void)jobid;
return 0;
}
static int test_on_pend(void* picker, uint64_t jobid)
{
(void)picker;
(void)jobid;
return 0;
}
static int test_on_config(void* picker, xce_conf4sched_t* config)
{
(void)picker;
g_counts.config_calls++;
g_counts.last_config = config;
return 0;
}
static int test_on_job_push(void* picker, xce_sched_job_t* job)
{
(void)picker;
g_counts.job_push_calls++;
g_counts.last_job_push = job;
return 0;
}
static int test_on_job_state_change(void* picker, xce_sched_job_t* job,
uint32_t state)
{
(void)picker;
g_counts.job_state_change_calls++;
g_counts.last_job_state_change_job = job;
g_counts.last_job_state_change_state = state;
return 0;
}
static int test_on_job_update(void* picker, xce_sched_job_t* job)
{
(void)picker;
g_counts.job_update_calls++;
g_counts.last_job_update = job;
return 0;
}
static int test_on_queue_update(void* picker, xce_sched_queue_t* queue)
{
(void)picker;
g_counts.queue_update_calls++;
g_counts.last_queue_update = queue;
return 0;
}
static int test_on_job_rusage(void* picker, xce_sched_job_t* job,
xce_job_rusages_t* rusages)
{
(void)picker;
g_counts.job_rusage_calls++;
g_counts.last_job_rusage_job = job;
g_counts.last_job_rusage_rusages = rusages;
return 0;
}
/* Full picker: every hook wired, including the two optional ones
* (on_config/on_job_rusage). */
static xce_job_picker_iface_t g_full_picker_iface = {
.init = NULL,
.fini = NULL,
.begin_session = test_begin_session,
.next_job = test_next_job,
.end_session = test_end_session,
.on_dispatch = test_on_dispatch,
.on_pend = test_on_pend,
.on_queue_update = test_on_queue_update,
.on_job_push = test_on_job_push,
.on_job_state_change = test_on_job_state_change,
.on_job_update = test_on_job_update,
.on_config = test_on_config,
.on_job_rusage = test_on_job_rusage,
};
/* Picker with on_job_rusage left NULL (Req 5 AC 4: "optional -- NULL if
* not registered") -- proves JOB_RUSAGE applies state without erroring
* when the hook is absent. */
static xce_job_picker_iface_t g_no_rusage_picker_iface = {
.init = NULL,
.fini = NULL,
.begin_session = test_begin_session,
.next_job = test_next_job,
.end_session = test_end_session,
.on_dispatch = test_on_dispatch,
.on_pend = test_on_pend,
.on_queue_update = test_on_queue_update,
.on_job_push = test_on_job_push,
.on_job_state_change = test_on_job_state_change,
.on_job_update = test_on_job_update,
.on_config = test_on_config,
.on_job_rusage = NULL,
};
static xce_sched_plugin_t g_full_plugin
= { .name = "test_full_picker", .picker_iface = &g_full_picker_iface };
static xce_sched_plugin_t g_no_rusage_plugin
= { .name = "test_no_rusage_picker",
.picker_iface = &g_no_rusage_picker_iface };
/* ---------------------------------------------------------------------- */
/* Fixture: real xce_sched_core_init(), a test picker registered directly */
/* into core.plugins (key_add -- the same mechanism plugin_register() */
/* itself uses, schedobj.c not included so its wrapper is not available), */
/* and core.job_pick_plugin pointed at it. Never torn down (this */
/* directory's established convention -- see test_core_jbuf_capacity_pin.c */
/* header comment). */
/* ---------------------------------------------------------------------- */
static void fixture_init(xce_sched_core_t* core, xce_sched_plugin_t* plugin)
{
int rc;
memset(&g_counts, 0, sizeof(g_counts));
rc = xce_sched_core_init(core);
TEST_ASSERT_EQUAL_INT(0, rc);
lock(core->plugins, RTTI_LOCK_WRITE);
rc = key_add(core->plugins, (void*)plugin->name, (void*)plugin);
unlock(core->plugins);
TEST_ASSERT_GREATER_OR_EQUAL(0, rc);
core->job_pick_plugin = plugin->name;
}
static xce_job_t* make_job(uint64_t jobid, uint32_t state, char* queue,
char* user, char* project)
{
xce_job_t* j = (xce_job_t*)calloc(1, sizeof(*j));
j->jobid = jobid;
j->state = state;
j->sub_time = 1000;
j->queue_time = 1000;
j->start_time = 0;
j->submit = (job_submit_data_t*)calloc(1, sizeof(*(j->submit)));
j->submit->queue = queue;
j->submit->user = user;
j->submit->project_name = project;
j->submit->resreq = "type == any";
j->submit->mintasks = 1;
j->submit->maxtasks = 1;
return j;
}
/* ---------------------------------------------------------------------- */
/* Tests */
/* ---------------------------------------------------------------------- */
/* One event of each of the 6 hook-having kinds in a single batch --
* asserts each corresponding hook fired EXACTLY once, synchronously
* (already true by the time push_events returns, in this single-threaded
* test), with the right argument. */
static void
test_push_events_fires_each_hook_once_per_event_with_picker_idle(void)
{
xce_sched_core_t core;
xce_core_event_t events[6];
xce_conf4sched_t config;
xce_job_t* job1;
xce_queue_t wire_queue;
xce_job_t job_update_payload;
xce_job_rusages_t rusages;
int rc;
fixture_init(&core, &g_full_plugin);
memset(&config, 0, sizeof(config));
job1 = make_job(42, JOB_STAT_PEND, "normal", "alice", "proj1");
memset(&wire_queue, 0, sizeof(wire_queue));
wire_queue.queue = "normal";
wire_queue.priority = 5;
wire_queue.status = QUEUE_STAT_OPEN;
memset(events, 0, sizeof(events));
events[0].type = XCE_CORE_EVENT_CONFIG_PUSH;
events[0].u.config_push.config = &config;
events[1].type = XCE_CORE_EVENT_QUEUE_UPDATE;
events[1].u.queue_update.queue = &wire_queue;
events[2].type = XCE_CORE_EVENT_JOBS_PUSH;
events[2].u.jobs_push.njobs = 1;
events[2].u.jobs_push.jobs = &job1;
events[3].type = XCE_CORE_EVENT_JOB_STATE_CHANGE;
events[3].u.job_state_change.jobid = 42;
events[3].u.job_state_change.old_state = JOB_STAT_PEND;
events[3].u.job_state_change.new_state = JOB_STAT_RUN;
events[3].u.job_state_change.exit_code = 0;
memset(&job_update_payload, 0, sizeof(job_update_payload));
job_update_payload.jobid = 42;
job_update_payload.state = JOB_STAT_RUN;
events[4].type = XCE_CORE_EVENT_JOB_UPDATE;
events[4].u.job_update.jobid = 42;
events[4].u.job_update.changed_fields = 0;
events[4].u.job_update.job = &job_update_payload;
memset(&rusages, 0, sizeof(rusages));
rusages.jobid = 42;
events[5].type = XCE_CORE_EVENT_JOB_RUSAGE;
events[5].u.job_rusage.jobid = 42;
events[5].u.job_rusage.rusages = &rusages;
rc = xce_core_push_events(&core, events, 6);
TEST_ASSERT_EQUAL_INT(0, rc);
TEST_ASSERT_EQUAL_INT(1, g_counts.config_calls);
TEST_ASSERT_EQUAL_PTR(&config, g_counts.last_config);
TEST_ASSERT_EQUAL_INT(1, g_counts.queue_update_calls);
TEST_ASSERT_NOT_NULL(g_counts.last_queue_update);
TEST_ASSERT_EQUAL_STRING("normal", g_counts.last_queue_update->name);
TEST_ASSERT_EQUAL_INT(1, g_counts.job_push_calls);
TEST_ASSERT_NOT_NULL(g_counts.last_job_push);
TEST_ASSERT_EQUAL_UINT64(42, g_counts.last_job_push->jobid);
TEST_ASSERT_EQUAL_INT(1, g_counts.job_state_change_calls);
TEST_ASSERT_NOT_NULL(g_counts.last_job_state_change_job);
TEST_ASSERT_EQUAL_UINT64(42, g_counts.last_job_state_change_job->jobid);
TEST_ASSERT_EQUAL_UINT32(JOB_STAT_RUN, g_counts.last_job_state_change_state);
/* state mutation actually applied on the core-owned job object */
TEST_ASSERT_EQUAL_UINT32(JOB_STAT_RUN, sched_get_job(&core, 42)->state);
TEST_ASSERT_EQUAL_INT(1, g_counts.job_update_calls);
TEST_ASSERT_NOT_NULL(g_counts.last_job_update);
TEST_ASSERT_EQUAL_UINT64(42, g_counts.last_job_update->jobid);
TEST_ASSERT_EQUAL_INT(1, g_counts.job_rusage_calls);
TEST_ASSERT_NOT_NULL(g_counts.last_job_rusage_job);
TEST_ASSERT_EQUAL_PTR(&rusages, g_counts.last_job_rusage_rusages);
TEST_ASSERT_EQUAL_PTR(&rusages, sched_get_job(&core, 42)->rusages);
}
/* JOBS_PUSH with njobs > 1: on_job_push must fire once PER JOB, not once
* per event -- Req 5 AC 4's "on_job_push(picker, job) for each job in the
* batch". */
static void test_jobs_push_fires_on_job_push_once_per_job(void)
{
xce_sched_core_t core;
xce_core_event_t event;
xce_job_t* jobs[3];
int rc;
fixture_init(&core, &g_full_plugin);
jobs[0] = make_job(1, JOB_STAT_PEND, "normal", "alice", "proj1");
jobs[1] = make_job(2, JOB_STAT_PEND, "normal", "bob", "proj1");
jobs[2] = make_job(3, JOB_STAT_PEND, "normal", "carol", "proj1");
memset(&event, 0, sizeof(event));
event.type = XCE_CORE_EVENT_JOBS_PUSH;
event.u.jobs_push.njobs = 3;
event.u.jobs_push.jobs = jobs;
rc = xce_core_push_events(&core, &event, 1);
TEST_ASSERT_EQUAL_INT(0, rc);
TEST_ASSERT_EQUAL_INT(3, g_counts.job_push_calls);
TEST_ASSERT_NOT_NULL(sched_get_job(&core, 1));
TEST_ASSERT_NOT_NULL(sched_get_job(&core, 2));
TEST_ASSERT_NOT_NULL(sched_get_job(&core, 3));
}
/* bug-1714 follow-up (Bug A): xce_core_push_events()'s JOBS_PUSH case never
* checked sched_add_job()'s return value, so a redelivered duplicate jobid
* (sched_add_job -> key_add -> -EEXIST) was still handed to
* picker->picker_iface->on_job_push() even though it was never actually
* registered in core->jobs -- an orphaned xce_sched_job_t the picker now
* tracks but no core->jobs-keyed lookup (sched_get_job() included) can ever
* find again. Asserts: (1) baseline single push fires on_job_push exactly
* once and registers the job; (2) a SECOND JOBS_PUSH batch carrying the SAME
* jobid must NOT fire on_job_push again (spy count stays at the baseline,
* last-seen job pointer unchanged) and sched_get_job() for that jobid must
* still return the ORIGINAL job object, not the orphaned duplicate. */
static void
test_jobs_push_duplicate_jobid_does_not_call_on_job_push_or_leak(void)
{
xce_sched_core_t core;
xce_core_event_t event;
xce_job_t* job1;
xce_job_t* dup_job;
xce_sched_job_t* original;
int rc;
fixture_init(&core, &g_full_plugin);
job1 = make_job(55, JOB_STAT_PEND, "normal", "alice", "proj1");
memset(&event, 0, sizeof(event));
event.type = XCE_CORE_EVENT_JOBS_PUSH;
event.u.jobs_push.njobs = 1;
event.u.jobs_push.jobs = &job1;
rc = xce_core_push_events(&core, &event, 1);
TEST_ASSERT_EQUAL_INT(0, rc);
TEST_ASSERT_EQUAL_INT(1, g_counts.job_push_calls);
original = sched_get_job(&core, 55);
TEST_ASSERT_NOT_NULL(original);
TEST_ASSERT_EQUAL_PTR(original, g_counts.last_job_push);
/* Redelivered duplicate: a second, independently-built xce_job_t with
* the SAME jobid, in its own JOBS_PUSH batch -- mirrors what
* sched_on_job_array()'s TOCTOU race window (Bug B, handler.c) can let
* through to this layer. */
dup_job = make_job(55, JOB_STAT_PEND, "normal", "alice", "proj1");
memset(&event, 0, sizeof(event));
event.type = XCE_CORE_EVENT_JOBS_PUSH;
event.u.jobs_push.njobs = 1;
event.u.jobs_push.jobs = &dup_job;
rc = xce_core_push_events(&core, &event, 1);
TEST_ASSERT_EQUAL_INT(0, rc);
/* on_job_push must NOT fire a second time for the duplicate. */
TEST_ASSERT_EQUAL_INT(1, g_counts.job_push_calls);
TEST_ASSERT_EQUAL_PTR(original, g_counts.last_job_push);
/* core->jobs must still resolve to the ORIGINAL job -- not an orphaned
* duplicate that a naive re-lookup might otherwise stumble into. */
TEST_ASSERT_EQUAL_PTR(original, sched_get_job(&core, 55));
}
/* bug-1734: a redelivered JOBS_PUSH for an ALREADY-TRACKED jobid is not
* always a harmless no-op duplicate the way bug-1714's Bug A test above
* assumed. Live reproduction (50-element array job, real 3-node HA cluster +
* worker): the scheduler's own outbound xcecli connection to the controller
* repeatedly failed over and reconnected under load; every reconnect made
* the controller's on_scheduler_join() (sched.c) re-run a FULL resync --
* xcesched_push_jobs() -- which re-sends every currently-committed job
* (including ones already terminal on the controller side, e.g. DONE) via
* the SAME XceData_data_schedjobs wire path this JOBS_PUSH case handles.
* Unlike the array-push redelivery bug-1714 covers (both copies start
* identically PEND, so discarding the second is provably harmless), a
* resync redelivery can carry a job whose state has moved on since this
* scheduler last saw it (PEND/RUN -> DONE). The current code frees ANY
* duplicate unconditionally and never updates the existing tracked job, so
* that newer state is silently and PERMANENTLY lost -- confirmed live via
* controller ctrl-1.log's `c->sched.ep is NULL, notify SKIPPED` diagnostic
* (bug-1731) firing for the same jobids during the reconnect gap: the
* resync was that job's ONLY remaining chance to reach the scheduler, and
* this discard-without-merge path threw it away. Asserts a duplicate
* JOBS_PUSH carrying a DIFFERENT state for an already-tracked jobid updates
* the EXISTING xce_sched_job_t's state in place (mirroring how
* XCE_CORE_EVENT_JOB_STATE_CHANGE/JOB_UPDATE already treat a state change)
* rather than leaving it frozen at the stale value. */
static void
test_jobs_push_duplicate_jobid_with_newer_state_updates_existing_job(void)
{
xce_sched_core_t core;
xce_core_event_t event;
xce_job_t* job1;
xce_job_t* resynced_job;
xce_sched_job_t* tracked;
int rc;
fixture_init(&core, &g_full_plugin);
job1 = make_job(56, JOB_STAT_PEND, "normal", "alice", "proj1");
memset(&event, 0, sizeof(event));
event.type = XCE_CORE_EVENT_JOBS_PUSH;
event.u.jobs_push.njobs = 1;
event.u.jobs_push.jobs = &job1;
rc = xce_core_push_events(&core, &event, 1);
TEST_ASSERT_EQUAL_INT(0, rc);
tracked = sched_get_job(&core, 56);
TEST_ASSERT_NOT_NULL(tracked);
TEST_ASSERT_EQUAL_UINT32(JOB_STAT_PEND, tracked->state);
/* A full-resync redelivery (on_scheduler_join()'s xcesched_push_jobs())
* of the SAME jobid, now carrying its real/current terminal state -- the
* job finished on the controller while this scheduler's push channel
* was reconnecting. */
resynced_job = make_job(56, JOB_STAT_DONE, "normal", "alice", "proj1");
memset(&event, 0, sizeof(event));
event.type = XCE_CORE_EVENT_JOBS_PUSH;
event.u.jobs_push.njobs = 1;
event.u.jobs_push.jobs = &resynced_job;
rc = xce_core_push_events(&core, &event, 1);
TEST_ASSERT_EQUAL_INT(0, rc);
/* Must still resolve to the SAME xce_sched_job_t (no orphaned second
* object, no key_add() churn) ... */
TEST_ASSERT_EQUAL_PTR(tracked, sched_get_job(&core, 56));
/* ... but its state must now reflect the resync's newer information,
* not stay frozen at the stale PEND it had before the redelivery. */
TEST_ASSERT_EQUAL_UINT32(JOB_STAT_DONE, tracked->state);
}
/* bug-1739 (Symptom 1, "phantom RUN"): bug-1734's own JOBS_PUSH duplicate-
* jobid merge (test above) adopted a redelivery's RUN state without
* transferring any real allocation, leaving a tracked job with state==RUN
* but allocs==NULL -- a job the scheduler's own resource/slot bookkeeping
* has no record of at all (live-reported: `info worker` listed fewer
* "assigned job:" entries than `info jobs` showed RUN, exceeding the
* worker's real core count). core_build_job() already adopts a wire Job's
* `allocs` field into the freshly-built xce_sched_job_t (job->accounts =
* make_job_account(job->allocs)) exactly like a brand-new RUN job would --
* this asserts a RUN-state resync redelivery that DOES carry real
* allocation data transfers it into the EXISTING tracked job instead of
* discarding it. */
static void
test_jobs_push_duplicate_jobid_run_state_transfers_allocation(void)
{
xce_sched_core_t core;
xce_core_event_t event;
xce_job_t* job1;
xce_job_t* resynced_job;
xce_sched_job_t* tracked;
xce_alloc_t* a_slots;
xce_host_alloc_t* h0;
xce_host_alloc_t* hallocs[1];
xce_job_alloc_t* jalloc;
int rc;
fixture_init(&core, &g_full_plugin);
job1 = make_job(57, JOB_STAT_PEND, "normal", "alice", "proj1");
memset(&event, 0, sizeof(event));
event.type = XCE_CORE_EVENT_JOBS_PUSH;
event.u.jobs_push.njobs = 1;
event.u.jobs_push.jobs = &job1;
rc = xce_core_push_events(&core, &event, 1);
TEST_ASSERT_EQUAL_INT(0, rc);
tracked = sched_get_job(&core, 57);
TEST_ASSERT_NOT_NULL(tracked);
TEST_ASSERT_EQUAL_UINT32(JOB_STAT_PEND, tracked->state);
TEST_ASSERT_NULL(tracked->allocs);
/* A resync redelivery carrying RUN AND the controller's real dispatch
* allocation for this job -- exactly what on_scheduler_join()'s
* xcesched_push_jobs() sends for a genuinely running job. */
a_slots = make_alloc((char*)"slots", NULL, 0, 1.0f, 0);
h0 = make_host_alloc((char*)"hostA", 1, make_allocs(1, a_slots));
hallocs[0] = h0;
jalloc = make_job_alloc(57, 1, make_host_allocs(1, hallocs[0]));
resynced_job = make_job(57, JOB_STAT_RUN, "normal", "alice",
"proj1");
resynced_job->allocs = jalloc;
memset(&event, 0, sizeof(event));
event.type = XCE_CORE_EVENT_JOBS_PUSH;
event.u.jobs_push.njobs = 1;
event.u.jobs_push.jobs = &resynced_job;
rc = xce_core_push_events(&core, &event, 1);
TEST_ASSERT_EQUAL_INT(0, rc);
TEST_ASSERT_EQUAL_PTR(tracked, sched_get_job(&core, 57));
TEST_ASSERT_EQUAL_UINT32(JOB_STAT_RUN, tracked->state);
/* The real allocation must have been transferred, not discarded --
* this is the actual fix: a RUN job must never end up with allocs ==
* NULL via this path. */
TEST_ASSERT_NOT_NULL(tracked->allocs);
TEST_ASSERT_NOT_NULL(tracked->accounts);
}
/* bug-1739 companion: if a RUN-state resync redelivery does NOT carry any
* allocation data (no `allocs` on the wire), merging it in anyway would
* manufacture exactly the same phantom-RUN phenotype from a different
* cause. Asserts the merge is refused (existing job keeps its prior state)
* rather than trusting an unallocated RUN claim. */
static void
test_jobs_push_duplicate_jobid_run_state_without_allocation_is_refused(void)
{
xce_sched_core_t core;
xce_core_event_t event;
xce_job_t* job1;
xce_job_t* resynced_job;
xce_sched_job_t* tracked;
int rc;
fixture_init(&core, &g_full_plugin);
job1 = make_job(58, JOB_STAT_PEND, "normal", "alice", "proj1");
memset(&event, 0, sizeof(event));
event.type = XCE_CORE_EVENT_JOBS_PUSH;
event.u.jobs_push.njobs = 1;
event.u.jobs_push.jobs = &job1;
rc = xce_core_push_events(&core, &event, 1);
TEST_ASSERT_EQUAL_INT(0, rc);
tracked = sched_get_job(&core, 58);
TEST_ASSERT_NOT_NULL(tracked);
/* RUN claim with NO allocation data attached. */
resynced_job = make_job(58, JOB_STAT_RUN, "normal", "alice", "proj1");
memset(&event, 0, sizeof(event));
event.type = XCE_CORE_EVENT_JOBS_PUSH;
event.u.jobs_push.njobs = 1;
event.u.jobs_push.jobs = &resynced_job;
rc = xce_core_push_events(&core, &event, 1);
TEST_ASSERT_EQUAL_INT(0, rc);
TEST_ASSERT_EQUAL_PTR(tracked, sched_get_job(&core, 58));
/* Refused -- stays at its prior (safe) state, never becomes a phantom
* RUN with no allocation. */
TEST_ASSERT_EQUAL_UINT32(JOB_STAT_PEND, tracked->state);
TEST_ASSERT_NULL(tracked->allocs);
}
/* bug-1738: the JOBS_PUSH duplicate-jobid merge must never regress an
* already-terminal job to a DIFFERENT terminal state either -- a resync can
* just as easily carry a stale snapshot as a genuinely newer one. Pushes
* jobid to DONE (as the bug-1734 test above already proves works), then a
* SECOND redelivery claiming EXIT must be ignored. */
static void
test_jobs_push_duplicate_jobid_does_not_regress_terminal_state(void)
{
xce_sched_core_t core;
xce_core_event_t event;
xce_job_t* job1;
xce_job_t* done_job;
xce_job_t* stale_exit_job;
xce_sched_job_t* tracked;
int rc;
fixture_init(&core, &g_full_plugin);
job1 = make_job(59, JOB_STAT_PEND, "normal", "alice", "proj1");
memset(&event, 0, sizeof(event));
event.type = XCE_CORE_EVENT_JOBS_PUSH;
event.u.jobs_push.njobs = 1;
event.u.jobs_push.jobs = &job1;
rc = xce_core_push_events(&core, &event, 1);
TEST_ASSERT_EQUAL_INT(0, rc);
done_job = make_job(59, JOB_STAT_DONE, "normal", "alice", "proj1");
memset(&event, 0, sizeof(event));
event.type = XCE_CORE_EVENT_JOBS_PUSH;
event.u.jobs_push.njobs = 1;
event.u.jobs_push.jobs = &done_job;
rc = xce_core_push_events(&core, &event, 1);
TEST_ASSERT_EQUAL_INT(0, rc);
tracked = sched_get_job(&core, 59);
TEST_ASSERT_NOT_NULL(tracked);
TEST_ASSERT_EQUAL_UINT32(JOB_STAT_DONE, tracked->state);
/* A stale/out-of-order redelivery claiming EXIT for a job already
* correctly recorded as DONE. */
stale_exit_job = make_job(59, JOB_STAT_EXIT, "normal", "alice", "proj1");
memset(&event, 0, sizeof(event));
event.type = XCE_CORE_EVENT_JOBS_PUSH;
event.u.jobs_push.njobs = 1;
event.u.jobs_push.jobs = &stale_exit_job;
rc = xce_core_push_events(&core, &event, 1);
TEST_ASSERT_EQUAL_INT(0, rc);
TEST_ASSERT_EQUAL_PTR(tracked, sched_get_job(&core, 59));
/* Must still be DONE -- the stale EXIT must not have clobbered it. */
TEST_ASSERT_EQUAL_UINT32(JOB_STAT_DONE, tracked->state);
}
/* bug-1738 (single-job notify path): XCE_CORE_EVENT_JOB_STATE_CHANGE must
* apply the SAME "terminal is permanent" guard as the JOBS_PUSH merge above
* -- this is the path a live schedFailJob notify actually reaches
* (sched_job_fail_fun, sched.c, unconditionally requests JOB_STAT_EXIT with
* no regard for the job's current state; the controller's own
* apply_job_fail_transition() only merges JOB_STAT_PERR onto an
* already-DONE/EXIT job in the equivalent real scenario, never re-decides
* DONE vs EXIT). Pushes a job to RUN then DONE via JOB_STATE_CHANGE, then a
* second JOB_STATE_CHANGE claiming EXIT must be ignored. */
static void test_job_state_change_does_not_regress_terminal_state(void)
{
xce_sched_core_t core;
xce_core_event_t events[3];
xce_job_t* job1;
xce_sched_job_t* job;
int rc;
fixture_init(&core, &g_full_plugin);
job1 = make_job(60, JOB_STAT_PEND, "normal", "alice", "proj1");
memset(events, 0, sizeof(events));
events[0].type = XCE_CORE_EVENT_JOBS_PUSH;
events[0].u.jobs_push.njobs = 1;
events[0].u.jobs_push.jobs = &job1;
events[1].type = XCE_CORE_EVENT_JOB_STATE_CHANGE;
events[1].u.job_state_change.jobid = 60;
events[1].u.job_state_change.new_state = JOB_STAT_DONE;
rc = xce_core_push_events(&core, events, 2);
TEST_ASSERT_EQUAL_INT(0, rc);
job = sched_get_job(&core, 60);
TEST_ASSERT_NOT_NULL(job);
TEST_ASSERT_EQUAL_UINT32(JOB_STAT_DONE, job->state);
/* A schedFailJob-shaped notify claiming EXIT for the SAME job, arriving
* after it already correctly reached DONE. */
events[2].type = XCE_CORE_EVENT_JOB_STATE_CHANGE;
events[2].u.job_state_change.jobid = 60;
events[2].u.job_state_change.new_state = JOB_STAT_EXIT;
rc = xce_core_push_events(&core, &events[2], 1);
TEST_ASSERT_EQUAL_INT(0, rc);
/* Must still be DONE -- the stale/narrower EXIT notify must not have
* clobbered it. */
TEST_ASSERT_EQUAL_UINT32(JOB_STAT_DONE, job->state);
}
/* JOB_UPDATE (core_events.c's live wire-shape mirror of JOB_STATE_CHANGE --
* the notify-driven single-job path a real controller's HOOK_DETAIL_JOB_UPDATE
* fire reaches, per handler.c's own XCE_SCHED_OP_UPDATE_JOB_FIELDS comment)
* must release a job's resource allocation/accounting the same way
* JOB_STATE_CHANGE's case already does (this file's own
* test_push_events_fires_each_hook_once_per_event_with_picker_idle only
* proves the hook fires -- it never inspects accounts/allocs afterward).
* Without this, a job that transitions to a terminal state via JOB_UPDATE
* (the only currently-live path -- JOB_STATE_CHANGE's own controller-side
* trigger, HOOK_DETAIL_JOB_EXIT/START/FAIL at HOOK_AFTER_EVENT, is never
* actually fired anywhere in the controller) leaks its account/alloc
* objects for the rest of the process's life. */
static void test_job_update_to_terminal_state_frees_accounts_and_allocs(void)
{
xce_sched_core_t core;
xce_core_event_t event;
xce_job_t* job1;
xce_job_t updated;
xce_alloc_t* a_slots;
xce_host_alloc_t* h0;
xce_host_alloc_t* hallocs[1];
xce_job_alloc_t* jalloc;
xce_sched_job_t* job;
int rc;
fixture_init(&core, &g_full_plugin);
job1 = make_job(7, JOB_STAT_RUN, "normal", "alice", "proj1");
memset(&event, 0, sizeof(event));
event.type = XCE_CORE_EVENT_JOBS_PUSH;
event.u.jobs_push.njobs = 1;
event.u.jobs_push.jobs = &job1;
rc = xce_core_push_events(&core, &event, 1);
TEST_ASSERT_EQUAL_INT(0, rc);
job = sched_get_job(&core, 7);
TEST_ASSERT_NOT_NULL(job);
a_slots = make_alloc((char*)"slots", NULL, 0, 4.0f, 0);
h0 = make_host_alloc((char*)"hostA", 1, make_allocs(1, a_slots));
hallocs[0] = h0;
jalloc = make_job_alloc(7, 1, make_host_allocs(1, hallocs[0]));
job->allocs = jalloc;
job->accounts = make_job_account(job->allocs);
TEST_ASSERT_NOT_NULL(job->accounts);
memset(&updated, 0, sizeof(updated));
updated.jobid = 7;
updated.state = JOB_STAT_DONE;
memset(&event, 0, sizeof(event));
event.type = XCE_CORE_EVENT_JOB_UPDATE;
event.u.job_update.jobid = 7;
event.u.job_update.job = &updated;
rc = xce_core_push_events(&core, &event, 1);
TEST_ASSERT_EQUAL_INT(0, rc);
TEST_ASSERT_EQUAL_UINT32(JOB_STAT_DONE, job->state);
TEST_ASSERT_NULL(job->accounts);
TEST_ASSERT_NULL(job->allocs);
}
/* WORKERS_PUSH/RESOURCES_PUSH/RESVALS_FULL/RESVALS_DELTA are state-update
* only -- Req 1 AC 2: "Hooks not defined on the picker interface (worker
* updates, resource updates, resval updates) are not forwarded to the
* picker." Every hook counter must stay at 0, while the underlying state
* IS genuinely mutated (not a no-op placeholder). */
static void test_non_picker_events_mutate_state_but_never_call_hooks(void)
{
xce_sched_core_t core;
xce_core_event_t events[4];
xce_worker_t wire_worker;
xce_worker_t* worker_ptrs[1];
xce_resource_t wire_resource;
xce_resource_t* resource_ptrs[1];
xce_resval_t full_resval;
xce_resval_t* full_vals[1];
xce_resval_t delta_resval;
xce_resval_t* delta_vals[1];
xce_resval_list_t deltas;
xce_val_t full_val;
xce_val_t delta_val;
int rc;
fixture_init(&core, &g_full_plugin);
memset(&wire_worker, 0, sizeof(wire_worker));
wire_worker.name = "hostA";
wire_worker.status = 0;
wire_worker.nresvals_to_sched = 0;
worker_ptrs[0] = &wire_worker;
memset(&wire_resource, 0, sizeof(wire_resource));
wire_resource.name = "cpu";
wire_resource.idx = 0;
resource_ptrs[0] = &wire_resource;
memset(&full_val, 0, sizeof(full_val));
memset(&full_resval, 0, sizeof(full_resval));
full_resval.name = "cpu";
full_resval.val = &full_val;
full_vals[0] = &full_resval;
memset(&delta_val, 0, sizeof(delta_val));
memset(&delta_resval, 0, sizeof(delta_resval));
delta_resval.name = "cpu";
delta_resval.host = "hostA";
delta_resval.val = &delta_val;
delta_vals[0] = &delta_resval;
deltas.count = 1;
deltas.vals = delta_vals;
memset(events, 0, sizeof(events));
events[0].type = XCE_CORE_EVENT_WORKERS_PUSH;
events[0].u.workers_push.nworkers = 1;
events[0].u.workers_push.workers = worker_ptrs;
events[1].type = XCE_CORE_EVENT_RESOURCES_PUSH;
events[1].u.resources_push.n = 1;
events[1].u.resources_push.resources = resource_ptrs;
events[2].type = XCE_CORE_EVENT_RESVALS_FULL;
events[2].u.resvals_full.n = 1;
events[2].u.resvals_full.resvals = full_vals;
events[3].type = XCE_CORE_EVENT_RESVALS_DELTA;
events[3].u.resvals_delta.deltas = &deltas;
/* worker must exist before RESVALS_DELTA can attribute a value to it */
rc = xce_core_push_events(&core, events, 1); /* WORKERS_PUSH only, first */
TEST_ASSERT_EQUAL_INT(0, rc);
rc = xce_core_push_events(&core, &events[1], 3); /* RESOURCES/RESVALS_FULL/RESVALS_DELTA */
TEST_ASSERT_EQUAL_INT(0, rc);
TEST_ASSERT_EQUAL_INT(0, g_counts.config_calls);
TEST_ASSERT_EQUAL_INT(0, g_counts.job_push_calls);
TEST_ASSERT_EQUAL_INT(0, g_counts.job_state_change_calls);
TEST_ASSERT_EQUAL_INT(0, g_counts.job_update_calls);
TEST_ASSERT_EQUAL_INT(0, g_counts.queue_update_calls);
TEST_ASSERT_EQUAL_INT(0, g_counts.job_rusage_calls);
TEST_ASSERT_NOT_NULL(sched_get_worker(&core, "hostA"));
}
/*
* bug-1568 regression: a worker connecting triggers multiple WORKERS_PUSH
* events in quick succession (live-confirmed against a real deploy -- an
* early push before the controller finishes writing a worker's real
* capacity values, carrying stale/zero data, followed shortly after by a
* push carrying the real values). Before this fix, WORKERS_PUSH always
* built a brand-new xce_sched_worker_t and called sched_add_worker() (a
* plain key_add(), add-only) unconditionally -- so the SECOND push for an
* already-registered worker name silently failed -EEXIST and its resource
* values were discarded, leaving core->workers[name] stuck at whatever the
* FIRST push happened to carry, forever. Two WORKERS_PUSH events for the
* SAME worker name, with DIFFERENT resource values, must both be applied
* -- the worker's resvals end up at the SECOND push's values, not the
* first, and no duplicate/orphaned worker object is created.
*/
static void test_workers_push_twice_updates_existing_worker_in_place(void)
{
xce_sched_core_t core;
xce_core_event_t event;
xce_resource_t wire_resource;
xce_worker_t wire_worker;
xce_worker_t* worker_ptrs[1];
xce_val_t val1;
xce_name_val_t nv1;
xce_name_val_t* nv1_ptrs[1];
xce_val_t val2;
xce_name_val_t nv2;
xce_name_val_t* nv2_ptrs[1];
xce_sched_worker_t* w;
xce_resource_t* r = NULL;
xce_val_t* applied;
int rc;
fixture_init(&core, &g_full_plugin);
memset(&wire_resource, 0, sizeof(wire_resource));
wire_resource.name = "mem";
TEST_ASSERT_NOT_NULL(makeResource(core.resmgr, &wire_resource));
TEST_ASSERT_TRUE(key_find(core.resmgr, (void*)"mem", (void**)&r) >= 0);
TEST_ASSERT_NOT_NULL(r);
/* First push: worker "hostA" joins with mem=100.0. */
memset(&val1, 0, sizeof(val1));
val1.which = Value_u_fvalue;
val1.u.fvalue = 100.0;
memset(&nv1, 0, sizeof(nv1));
nv1.name = "mem";
nv1.val = &val1;
nv1_ptrs[0] = &nv1;
memset(&wire_worker, 0, sizeof(wire_worker));
wire_worker.name = "hostA";
wire_worker.status = 0;
wire_worker.nresvals_to_sched = 1;
wire_worker.resvals_to_sched = nv1_ptrs;
worker_ptrs[0] = &wire_worker;
memset(&event, 0, sizeof(event));
event.type = XCE_CORE_EVENT_WORKERS_PUSH;
event.u.workers_push.nworkers = 1;
event.u.workers_push.workers = worker_ptrs;
rc = xce_core_push_events(&core, &event, 1);
TEST_ASSERT_EQUAL_INT(0, rc);
w = sched_get_worker(&core, "hostA");
TEST_ASSERT_NOT_NULL(w);
applied = (xce_val_t*)zarray_get(&(w->resvals), r->idx);
TEST_ASSERT_NOT_NULL(applied);
TEST_ASSERT_EQUAL_FLOAT(100.0f, (float)applied->u.fvalue);
/* Second push: SAME worker name, mem now really 200.0 (the "arrived
* late with the real value" push bug-1568 traced live). */
memset(&val2, 0, sizeof(val2));
val2.which = Value_u_fvalue;
val2.u.fvalue = 200.0;
memset(&nv2, 0, sizeof(nv2));
nv2.name = "mem";
nv2.val = &val2;
nv2_ptrs[0] = &nv2;
wire_worker.resvals_to_sched = nv2_ptrs;
rc = xce_core_push_events(&core, &event, 1);
TEST_ASSERT_EQUAL_INT(0, rc);
/* Still exactly one "hostA" worker -- updated in place, not a second,
* silently-discarded object leaving the first one stale. */
w = sched_get_worker(&core, "hostA");
TEST_ASSERT_NOT_NULL(w);
applied = (xce_val_t*)zarray_get(&(w->resvals), r->idx);
TEST_ASSERT_NOT_NULL(applied);
TEST_ASSERT_EQUAL_FLOAT(200.0f, (float)applied->u.fvalue);
}
/*
* bug-1688 regression: a WORKERS_PUSH re-announcing an already-registered
* worker's resource NAME list can legitimately carry nv->val == NULL for
* some/all entries (controller.capnp.c's decode_Value_ptr() decodes a wire
* NameValue.val CAPN_NULL pointer to exactly this -- live-traced against a
* real replay: a repeat push for worker 'pc13v' carried 20 resource names
* with nv->val == NULL for every one). Before this fix,
* core_apply_worker_resvals() unconditionally zarray_put()'d this NULL,
* silently clobbering a previously-known-good resource value with NULL.
* Because zarray_put_ex() (zarray.c) only increments array->count when the
* slot being written was NULL beforehand, that first NULL-write left
* resvals.count unchanged (the old slot was non-NULL) while zeroing the
* slot's DATA -- so a THIRD push, now finding every slot already NULL, hit
* the "was NULL, count++" branch for each one, doubling resvals.count (the
* literal 20 -> 40 anomaly this bug was reported with) despite carrying the
* exact same resource names both times. Two resources (not one) are used
* here so the count-doubling assertion is meaningful, not a 1-vs-2
* coincidence.
*/
static void test_workers_push_with_null_val_does_not_clobber_existing_resval(
void)
{
xce_sched_core_t core;
xce_core_event_t event;
xce_resource_t wire_mem;
xce_resource_t wire_r1m;
xce_worker_t wire_worker;
xce_worker_t* worker_ptrs[1];
xce_val_t val_mem;
xce_val_t val_r1m;
xce_name_val_t nv_mem;
xce_name_val_t nv_r1m;
xce_name_val_t* nv_ptrs[2];
xce_name_val_t nv_mem_null;
xce_name_val_t nv_r1m_null;
xce_name_val_t* nv_null_ptrs[2];
xce_sched_worker_t* w;
xce_resource_t* r_mem = NULL;
xce_resource_t* r_r1m = NULL;
xce_val_t* applied;
int rc;
fixture_init(&core, &g_full_plugin);
memset(&wire_mem, 0, sizeof(wire_mem));
wire_mem.name = "mem";
TEST_ASSERT_NOT_NULL(makeResource(core.resmgr, &wire_mem));
memset(&wire_r1m, 0, sizeof(wire_r1m));
wire_r1m.name = "r1m";
TEST_ASSERT_NOT_NULL(makeResource(core.resmgr, &wire_r1m));
TEST_ASSERT_TRUE(key_find(core.resmgr, (void*)"mem", (void**)&r_mem) >= 0);
TEST_ASSERT_NOT_NULL(r_mem);
TEST_ASSERT_TRUE(key_find(core.resmgr, (void*)"r1m", (void**)&r_r1m) >= 0);
TEST_ASSERT_NOT_NULL(r_r1m);
/* First push: worker "hostA" joins with real mem/r1m values. */
memset(&val_mem, 0, sizeof(val_mem));
val_mem.which = Value_u_fvalue;
val_mem.u.fvalue = 100.0;
memset(&nv_mem, 0, sizeof(nv_mem));
nv_mem.name = "mem";
nv_mem.val = &val_mem;
memset(&val_r1m, 0, sizeof(val_r1m));
val_r1m.which = Value_u_fvalue;
val_r1m.u.fvalue = 5.0;
memset(&nv_r1m, 0, sizeof(nv_r1m));
nv_r1m.name = "r1m";
nv_r1m.val = &val_r1m;
nv_ptrs[0] = &nv_mem;
nv_ptrs[1] = &nv_r1m;
memset(&wire_worker, 0, sizeof(wire_worker));
wire_worker.name = "hostA";
wire_worker.status = 0;
wire_worker.nresvals_to_sched = 2;
wire_worker.resvals_to_sched = nv_ptrs;
worker_ptrs[0] = &wire_worker;
memset(&event, 0, sizeof(event));
event.type = XCE_CORE_EVENT_WORKERS_PUSH;
event.u.workers_push.nworkers = 1;
event.u.workers_push.workers = worker_ptrs;
rc = xce_core_push_events(&core, &event, 1);
TEST_ASSERT_EQUAL_INT(0, rc);
w = sched_get_worker(&core, "hostA");
TEST_ASSERT_NOT_NULL(w);
TEST_ASSERT_EQUAL_INT(2, w->resvals.count);
applied = (xce_val_t*)zarray_get(&(w->resvals), r_mem->idx);
TEST_ASSERT_NOT_NULL(applied);
TEST_ASSERT_EQUAL_FLOAT(100.0f, (float)applied->u.fvalue);
applied = (xce_val_t*)zarray_get(&(w->resvals), r_r1m->idx);
TEST_ASSERT_NOT_NULL(applied);
TEST_ASSERT_EQUAL_FLOAT(5.0f, (float)applied->u.fvalue);
/* Second push: SAME worker, SAME resource names, but nv->val == NULL
* for both -- the exact "re-announce names, no fresh values" shape
* live-traced from the real replay. Must NOT clobber the existing
* values, and must NOT change resvals.count. */
memset(&nv_mem_null, 0, sizeof(nv_mem_null));
nv_mem_null.name = "mem";
nv_mem_null.val = NULL;
memset(&nv_r1m_null, 0, sizeof(nv_r1m_null));
nv_r1m_null.name = "r1m";
nv_r1m_null.val = NULL;
nv_null_ptrs[0] = &nv_mem_null;
nv_null_ptrs[1] = &nv_r1m_null;
wire_worker.resvals_to_sched = nv_null_ptrs;
rc = xce_core_push_events(&core, &event, 1);
TEST_ASSERT_EQUAL_INT(0, rc);
w = sched_get_worker(&core, "hostA");
TEST_ASSERT_NOT_NULL(w);
TEST_ASSERT_EQUAL_INT(2, w->resvals.count);
applied = (xce_val_t*)zarray_get(&(w->resvals), r_mem->idx);
TEST_ASSERT_NOT_NULL(applied);
TEST_ASSERT_EQUAL_FLOAT(100.0f, (float)applied->u.fvalue);
applied = (xce_val_t*)zarray_get(&(w->resvals), r_r1m->idx);
TEST_ASSERT_NOT_NULL(applied);
TEST_ASSERT_EQUAL_FLOAT(5.0f, (float)applied->u.fvalue);
/* Third push: SAME shape again (still NULL vals) -- if the bug were
* present, resvals.count would double here (20 -> 40 in the live
* trace; 2 -> 4 here) because the SECOND push (above) would have
* already nulled the slots, making THIS push's writes look like
* fresh/first-time occupancy. Confirms the fix holds across repeats,
* not just once. */
rc = xce_core_push_events(&core, &event, 1);
TEST_ASSERT_EQUAL_INT(0, rc);
w = sched_get_worker(&core, "hostA");
TEST_ASSERT_NOT_NULL(w);
TEST_ASSERT_EQUAL_INT(2, w->resvals.count);
}
/* JOB_RUSAGE's on_job_rusage hook is OPTIONAL (Req 5 AC 4) -- a picker
* without it must still apply the state mutation and must not error or
* crash. */
static void test_job_rusage_applies_state_when_hook_is_null(void)
{
xce_sched_core_t core;
xce_core_event_t events[2];
xce_job_t* job1;
xce_job_rusages_t rusages;
int rc;
fixture_init(&core, &g_no_rusage_plugin);
job1 = make_job(7, JOB_STAT_RUN, "normal", "alice", "proj1");
memset(&rusages, 0, sizeof(rusages));
rusages.jobid = 7;
memset(events, 0, sizeof(events));
events[0].type = XCE_CORE_EVENT_JOBS_PUSH;
events[0].u.jobs_push.njobs = 1;
events[0].u.jobs_push.jobs = &job1;
events[1].type = XCE_CORE_EVENT_JOB_RUSAGE;
events[1].u.job_rusage.jobid = 7;
events[1].u.job_rusage.rusages = &rusages;
rc = xce_core_push_events(&core, events, 2);
TEST_ASSERT_EQUAL_INT(0, rc);
TEST_ASSERT_EQUAL_INT(0, g_counts.job_rusage_calls);
TEST_ASSERT_EQUAL_PTR(&rusages, sched_get_job(&core, 7)->rusages);
}
/* CONFIG_PUSH must decompose the pushed xce_conf4sched_t's embedded
* queues[]/resources[]/resvals[] arrays into core state -- not just store
* the config pointer and bump configver (Req 5 AC 4's CONFIG_PUSH row:
* "All queues, resource thresholds, ... " is part of the payload, and
* decomposing it is the core's job, mirroring scheduler.c's
* sched_init_resources()/sched_init_sharedInst()/sched_init_queues(), all
* three of which sched_init_config() calls today for this exact same
* xce_conf4sched_t payload). Also asserts on_queue_update fires once per
* embedded queue (design intent: a config push's queues behave the same as
* individual queue updates, just batched). */
static void test_config_push_decomposes_queues_resources_and_resvals(void)
{
xce_sched_core_t core;
xce_core_event_t event;
xce_conf4sched_t config;
xce_resource_t wire_resource;
xce_resource_t* resources[1];
xce_val_t resval_value;
xce_name_val_t wire_resval;
xce_name_val_t* resvals[1];
xce_queue_t wire_queue;
xce_queue_t* queues[1];
xce_resource_t* found_resource = NULL;
xce_sched_queue_t* found_queue;
xce_val_t* shared_val;
int rc;
fixture_init(&core, &g_full_plugin);
memset(&wire_resource, 0, sizeof(wire_resource));
wire_resource.name = "cpu";
resources[0] = &wire_resource;
memset(&resval_value, 0, sizeof(resval_value));
resval_value.u.ivalue32 = 8;
memset(&wire_resval, 0, sizeof(wire_resval));
wire_resval.name = "cpu";
wire_resval.val = &resval_value;
resvals[0] = &wire_resval;
memset(&wire_queue, 0, sizeof(wire_queue));
wire_queue.queue = "batch";
wire_queue.priority = 7;
wire_queue.status = QUEUE_STAT_OPEN;
queues[0] = &wire_queue;
memset(&config, 0, sizeof(config));
config.n_resources = 1;
config.resources = resources;
config.nresvals = 1;
config.resvals = resvals;
config.n_queues = 1;
config.queues = queues;
memset(&event, 0, sizeof(event));
event.type = XCE_CORE_EVENT_CONFIG_PUSH;
event.u.config_push.config = &config;
rc = xce_core_push_events(&core, &event, 1);
TEST_ASSERT_EQUAL_INT(0, rc);
TEST_ASSERT_EQUAL_INT(1, g_counts.config_calls);
/* resource decomposed into core->resmgr */
rc = key_find(core.resmgr, (void*)"cpu", (void**)&found_resource);
TEST_ASSERT_GREATER_THAN_INT(0, rc);
TEST_ASSERT_NOT_NULL(found_resource);
/* resval decomposed into core->sharedInst at the resource's idx */
shared_val = (xce_val_t*)zarray_get(&(core.sharedInst), found_resource->idx);
TEST_ASSERT_NOT_NULL(shared_val);
TEST_ASSERT_EQUAL_INT(8, shared_val->u.ivalue32);
/* queue decomposed into core->queues, and on_queue_update fired once */
found_queue = sched_get_queue(&core, "batch");
TEST_ASSERT_NOT_NULL(found_queue);
TEST_ASSERT_EQUAL_INT(7, found_queue->priority);
TEST_ASSERT_EQUAL_INT(1, g_counts.queue_update_calls);
TEST_ASSERT_EQUAL_PTR(found_queue, g_counts.last_queue_update);
}
/* Session-active rejection (Req 5 AC 8 / design.md precondition):
* core_push_events() is only valid between sessions. */
static void
test_push_events_rejects_with_ebusy_when_session_active_no_mutation(void)
{
xce_sched_core_t core;
xce_core_event_t event;
xce_job_t* job1;
int rc;
fixture_init(&core, &g_full_plugin);
atomic_store(&(core.session_active), 1);
job1 = make_job(99, JOB_STAT_PEND, "normal", "alice", "proj1");
memset(&event, 0, sizeof(event));
event.type = XCE_CORE_EVENT_JOBS_PUSH;
event.u.jobs_push.njobs = 1;
event.u.jobs_push.jobs = &job1;
rc = xce_core_push_events(&core, &event, 1);
TEST_ASSERT_EQUAL_INT(-EBUSY, rc);
TEST_ASSERT_EQUAL_INT(0, g_counts.job_push_calls);
TEST_ASSERT_NULL(sched_get_job(&core, 99));
}
/* Bad-argument guard clauses. */
static void test_push_events_null_core_returns_einval(void)
{
xce_core_event_t event;
int rc;
memset(&event, 0, sizeof(event));
event.type = XCE_CORE_EVENT_CONFIG_PUSH;
rc = xce_core_push_events(NULL, &event, 1);
TEST_ASSERT_EQUAL_INT(-EINVAL, rc);
}
static void test_push_events_null_events_with_positive_nevents_returns_einval(void)
{
xce_sched_core_t core;
int rc;
fixture_init(&core, &g_full_plugin);
rc = xce_core_push_events(&core, NULL, 1);
TEST_ASSERT_EQUAL_INT(-EINVAL, rc);
}
/* Zero-length batch is a valid no-op (not an error). */
static void test_push_events_zero_nevents_is_a_valid_noop(void)
{
xce_sched_core_t core;
int rc;
fixture_init(&core, &g_full_plugin);
rc = xce_core_push_events(&core, NULL, 0);
TEST_ASSERT_EQUAL_INT(0, rc);
}
int main(void)
{
UNITY_BEGIN();
RUN_TEST(test_push_events_fires_each_hook_once_per_event_with_picker_idle);
RUN_TEST(test_jobs_push_fires_on_job_push_once_per_job);
RUN_TEST(test_jobs_push_duplicate_jobid_does_not_call_on_job_push_or_leak);
RUN_TEST(test_jobs_push_duplicate_jobid_with_newer_state_updates_existing_job);
RUN_TEST(test_jobs_push_duplicate_jobid_run_state_transfers_allocation);
RUN_TEST(test_jobs_push_duplicate_jobid_run_state_without_allocation_is_refused);
RUN_TEST(test_jobs_push_duplicate_jobid_does_not_regress_terminal_state);
RUN_TEST(test_job_state_change_does_not_regress_terminal_state);
RUN_TEST(test_job_update_to_terminal_state_frees_accounts_and_allocs);
RUN_TEST(test_non_picker_events_mutate_state_but_never_call_hooks);
RUN_TEST(test_workers_push_twice_updates_existing_worker_in_place);
RUN_TEST(test_workers_push_with_null_val_does_not_clobber_existing_resval);
RUN_TEST(test_job_rusage_applies_state_when_hook_is_null);
RUN_TEST(test_config_push_decomposes_queues_resources_and_resvals);
RUN_TEST(test_push_events_rejects_with_ebusy_when_session_active_no_mutation);
RUN_TEST(test_push_events_null_core_returns_einval);
RUN_TEST(test_push_events_null_events_with_positive_nevents_returns_einval);
RUN_TEST(test_push_events_zero_nevents_is_a_valid_noop);
return UNITY_END();
}
Reference in a new issue View git blame Copy permalink