/* * test_core_events.c - contract test for the core event interface and * batch event push (scheduler-core spec, task 4.3, Req 5 AC 4 / AC 8). * * Observable under test: pushing a batch of events with the picker idle * (core->session_active == 0) fires the corresponding state-change hook * exactly once per event, synchronously, before xce_core_push_events() * returns -- and the session-active rejection path (-EBUSY, no event * applied) when a session IS active. * * White-box #include of both ../src/core.c and ../src/core_events.c (same * convention as test_core_isolation.c / test_core_jbuf_capacity_pin.c) so * this test exercises the REAL xce_sched_core_init()/xce_core_push_events(), * not a re-implementation -- xce_core_push_events() lives in core_events.c * (split out of core.c purely to respect the 800-line-per-file cap; see * that file's own header comment), which itself calls straight back into * core.c's real sched_get_job()/sched_add_job()/etc. accessors, all within * this one translation unit. Stubs below are only for scheduler-internal * externs these two files reference (as kvmgr datakiller callback * addresses, or -- for sched_resolve_job_picker -- because the real * implementation lives in schedobj.c, a separate translation unit this * file does not pull in): * * - kill_sched_job/kill_sched_queue/kill_sched_worker/kill_sched_user/ * kill_sched_project/kill_sched_resval/plugin_unregister: pure no-op * stubs, identical to test_core_jbuf_capacity_pin.c's own -- this * file's fixtures are never torn down (never call * xce_sched_core_fini()), same "fixtures aren't torn down" convention * established across this directory, so these killers are never * actually invoked; present only to satisfy the linker for * xce_sched_core_init()'s kvmgr_params_t.datakiller fields. * - sched_resolve_job_picker: a real, working reimplementation of * schedobj.c's function of the same name (identical algorithm -- * key_find core->job_pick_plugin in core->plugins, validate the * picker MUST hooks) -- needed because the true implementation lives * in schedobj.c, and pulling that whole file in would drag in its own * large stub surface (plugin_queue_update, plugin_job_update, etc.) * for functions this file's tests never exercise. This IS real logic, * not a null stub, because the hook-firing contract under test * depends on picker resolution actually working. * * make_job_account/free_job_account/free_JobAlloc_ptr/free_JobRusages_ptr/ * makeResource/parse_resreq/makeString are all real, linked from the * shared libxce library (src/lib), not stubbed -- same as every other * scheduler white-box test in this directory. */ #include #include #include #include #include #include "kvmgr.h" #include "scheduler.h" #include "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"`). * * resource.c added (bug-1740, the reconciliation wiring fix): XCE_CORE_ * EVENT_JOB_RUSAGE now calls sched_reconcile_job_resource_usage() (that * file), the ONLY live call site anywhere in this codebase -- the whole * point of this file's own end-to-end test for it is to exercise the REAL * implementation, not a stub that would silently let the wiring rot again * unnoticed, exactly as it did the first time. */ #include "../src/core.c" #include "../src/core_events.c" #include "../src/resource.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; } /* Link-satisfying stubs for resource.c's sched_try_alloc_on_host_() (its * OWN callees, live in plugin.c/schedobj.c, neither pulled in here) -- same * two stubs test_resource_accounting.c already establishes needing for the * identical reason. Never actually invoked: this file's tests never call * sched_try_alloc_on_hosts()/sched_try_alloc_on_host_(). */ int plugin_prepare_allocation(void* s, xce_sched_ctx_t* c) { (void)s; (void)c; return 0; } int sched_ctx_get_uncomit_host(xvec_t** v, xce_sched_ctx_t* c, int i) { (void)v; (void)c; (void)i; return 0; } /* 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-1812 ("stale resync regression"): symmetric to bug-1739's own * "phantom RUN" guard just above (refusing to FABRICATE an allocation-less * RUN claim), but for the descent direction -- a JOBS_PUSH resync * redelivery must not DESTROY a job's REAL, already-captured allocation by * trusting an unsupported non-RUN, non-terminal claim over it. * Live-confirmed via a real production scheduler.log: on_scheduler_join()'s * full resync (xcesched_push_jobs(), controller/sched.c) fires on every * scheduler reconnect, built from collect_jobs_committed() -- a * committed-data snapshot that can race the scheduler's own very recent, * self-initiated dispatch. The log showed a JOBS_PUSH resync WARN landing * just 16ms after this exact job's own schedStartJob confirmation * (sched_job_state_change -> state 2), repeating in bursts on later * reconnects -- each collision wiping the job's real allocation while its * state stayed RUN, exactly matching the simulator's live-reported symptom * ("info jobs" shows RUN, "info worker"'s assigned-jobs listing -- which * walks job->allocs -- omits it). Asserts a job tracked RUN with a real * allocation, redelivered via resync as PEND with no allocation, KEEPS its * RUN state and real allocation -- the resync's unsupported claim must be * refused, not trusted. */ static void test_jobs_push_duplicate_jobid_run_with_allocation_survives_stale_nonrun_resync( void) { xce_sched_core_t core; xce_core_event_t event; xce_job_t* job1; xce_job_t* stale_resync_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(65, 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); tracked = sched_get_job(&core, 65); TEST_ASSERT_NOT_NULL(tracked); /* Give it a real, captured allocation -- mirrors what a genuine * dispatch (scheduler.c's sched_commit_decision()) would already have * set by the time a stale resync arrives. */ 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(65, 1, make_host_allocs(1, hallocs[0])); tracked->allocs = jalloc; tracked->accounts = make_job_account(tracked->allocs); /* A stale on_scheduler_join() resync redelivery -- the controller's * committed-data snapshot hasn't yet caught up to this job's own very * recent dispatch, so it still reports PEND with no allocation. */ stale_resync_job = make_job(65, 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 = &stale_resync_job; rc = xce_core_push_events(&core, &event, 1); TEST_ASSERT_EQUAL_INT(0, rc); TEST_ASSERT_EQUAL_PTR(tracked, sched_get_job(&core, 65)); /* Must still be RUN with its real allocation intact -- the stale resync * claim must not have clobbered it. */ TEST_ASSERT_EQUAL_UINT32(JOB_STAT_RUN, tracked->state); TEST_ASSERT_EQUAL_PTR(jalloc, tracked->allocs); TEST_ASSERT_NOT_NULL(tracked->accounts); } /* Companion: a resync claiming a genuinely TERMINAL state (DONE/EXIT) for a * RUN-with-real-allocation job must still apply -- this is the legitimate * "job finished while this scheduler was disconnected" case bug-1734's own * fix exists for, and the new stale-resync guard above must not block it. */ static void test_jobs_push_duplicate_jobid_terminal_resync_releases_real_allocation(void) { xce_sched_core_t core; xce_core_event_t event; xce_job_t* job1; xce_job_t* finished_resync_job; xce_sched_job_t* tracked; xce_alloc_t* a_slots; xce_host_alloc_t* h0; xce_host_alloc_t* hallocs[1]; int rc; fixture_init(&core, &g_full_plugin); job1 = make_job(66, 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); tracked = sched_get_job(&core, 66); TEST_ASSERT_NOT_NULL(tracked); 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; tracked->allocs = make_job_alloc(66, 1, make_host_allocs(1, hallocs[0])); tracked->accounts = make_job_account(tracked->allocs); finished_resync_job = make_job(66, 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 = &finished_resync_job; rc = xce_core_push_events(&core, &event, 1); TEST_ASSERT_EQUAL_INT(0, rc); TEST_ASSERT_EQUAL_PTR(tracked, sched_get_job(&core, 66)); TEST_ASSERT_EQUAL_UINT32(JOB_STAT_DONE, tracked->state); TEST_ASSERT_NULL(tracked->allocs); TEST_ASSERT_NULL(tracked->accounts); } /* 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); } /* * bug-1740: XCE_CORE_EVENT_JOB_RUSAGE -- the ONLY live production path for * job rusage, per both the normal-mode driver (driver.c) and the sim/replay * driver (sim_loop.c), which both translate their own rusage events into * this exact core-level event type -- must actually drive resource- * accounting.md's reservation-vs-usage reconciliation. It didn't: the * reconciliation code was wired into schedobj.c's sched_apply_job_rusage() * instead, a function with exactly ONE caller anywhere in this codebase * (scheduler.c's sched_update(), itself confirmed to have NO live caller at * all -- superseded by this file's own driver-event mechanism, per driver.c's * own comments: "Replaces sched_update()'s inline handling"). The entire * reconciliation feature was therefore unreachable in both normal-mode and * sim/replay operation despite passing every unit test that called its * functions directly -- exactly the trap of testing a function in isolation * without a test proving the real event path actually reaches it. Fixed by * calling sched_reconcile_job_resource_usage() directly from THIS handler. * This test drives the full path: register "mem", join a worker, dispatch a * job with a real 4MB "mem" allocation (core_build_job() -> make_job_account() * populates job->accounts exactly as a genuine dispatch would), then push a * JOB_RUSAGE report claiming 7MB actual usage (> the 4MB reservation) and * asserts the OVERAGE (3MB) landed on the account and the pool. */ static void test_job_rusage_reconciles_mem_reservation_against_usage(void) { xce_sched_core_t core; xce_core_event_t event; xce_resource_t wire_mem; xce_worker_t wire_worker; xce_worker_t* worker_ptrs[1]; xce_sched_worker_t* w; xce_resource_t* r_mem = NULL; xce_job_t* job1; xce_alloc_t* a_mem; xce_host_alloc_t* h0; xce_host_alloc_t* hallocs[1]; xce_job_alloc_t* jalloc; xce_sched_job_t* tracked; xce_rusage_t ru; xce_rusage_t* ru_ptrs[1]; xce_job_rusages_t rusages; int rc; int idx[1]; float val[1]; 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)); TEST_ASSERT_TRUE(key_find(core.resmgr, (void*)"mem", (void**)&r_mem) >= 0); TEST_ASSERT_NOT_NULL(r_mem); memset(&wire_worker, 0, sizeof(wire_worker)); wire_worker.name = "hostA"; wire_worker.nresvals_to_sched = 0; 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); /* Dispatch job 77 with a real 4MB "mem" allocation on hostA -- exactly * what a genuine DISPATCH decision's job->allocs would carry. */ a_mem = make_alloc((char*)"mem", NULL, 0, 4.0f, 0); h0 = make_host_alloc((char*)"hostA", 1, make_allocs(1, a_mem)); hallocs[0] = h0; jalloc = make_job_alloc(77, 1, make_host_allocs(1, hallocs[0])); job1 = make_job(77, JOB_STAT_RUN, "normal", "alice", "proj1"); job1->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 = &job1; rc = xce_core_push_events(&core, &event, 1); TEST_ASSERT_EQUAL_INT(0, rc); tracked = sched_get_job(&core, 77); TEST_ASSERT_NOT_NULL(tracked); TEST_ASSERT_NOT_NULL(tracked->accounts); /* bug-1758: JOBS_PUSH itself never charges resInUse -- by design, that * is sched_account_jobs()'s job (scheduler.c), a round-boundary * reconciliation the real normal-mode/sim drivers both call once per * round (driver.c/sim_loop.c), BEFORE anything in that round ever reads * resInUse, re-deriving it from every currently-RUN job's accounts via * sched_charge_account() -- see core.h's own doc comment on * sched_account_jobs(). This test's own JOBS_PUSH call above is not * itself a round boundary, so simulate the one sched_account_jobs() * would perform for this test's single RUN job directly (sched_account_ * jobs() itself lives in scheduler.c, not white-box #included here -- * this file already pulls in resource.c for sched_charge_account()). */ sched_charge_account(&core, tracked->accounts); /* Sanity: the reservation charge lands via that reconciliation, before * any rusage tick. */ idx[0] = r_mem->idx; get_resvals(&(w->resInUse), 1, idx, val); TEST_ASSERT_EQUAL_FLOAT(4.0f, val[0]); /* Now the worker reports job 77's REAL usage: 7MB, exceeding the 4MB * reservation. */ memset(&ru, 0, sizeof(ru)); ru.resname = (char*)"memory_rss"; ru.value = 7.0f; ru_ptrs[0] = &ru; memset(&rusages, 0, sizeof(rusages)); rusages.jobid = 77; rusages.count = 1; rusages.usages = ru_ptrs; memset(&event, 0, sizeof(event)); event.type = XCE_CORE_EVENT_JOB_RUSAGE; event.u.job_rusage.jobid = 77; event.u.job_rusage.rusages = &rusages; rc = xce_core_push_events(&core, &event, 1); TEST_ASSERT_EQUAL_INT(0, rc); /* The overage (3MB) must now be reflected both on the account and in * the pool -- the whole point of this test: this ONLY happens if the * live event handler actually calls the reconciliation code. */ TEST_ASSERT_EQUAL_FLOAT( 3.0f, (float)tracked->accounts->accounts[0]->accounts[0]->used); get_resvals(&(w->resInUse), 1, idx, val); TEST_ASSERT_EQUAL_FLOAT(7.0f, val[0]); } /* 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); } /* * Req 7.1 regression (scheduler-allocation spec, task 1.2, round 2): * "The resreq fingerprint is stored on the scheduling object * (xce_primary_schedobj_t) and computed once on job ingest." Round 1 added * xce_resreq_fingerprint_t (sched/types.h) and rusage_hash (resreq.h/ * resreq.c) but never actually wired population into the real job-ingest * construction site, core_build_primary_schedobj() (this file, right below * the white-box #include) -- obj->u.primary->resreq_fp was left calloc'd * to all-zero forever. This test calls the REAL core_build_primary_schedobj() * (visible here via this file's white-box #include of ../src/core_events.c, * same convention as every other static-function test in this file) with a * REAL xce_primary_resreq_t produced by the REAL parse_resreq_primary() * (src/lib/resreq.c, linked from libxce) -- not a mirror struct like * src/lib/testsuite/test_resreq_fingerprint.c's own fingerprint_from_resreq() * helper had to use (that file cannot reach src/scheduler/include's real * xce_resreq_fingerprint_t/xce_primary_schedobj_t -- see its header * comment) -- so this is the first test anywhere that proves the real * struct is actually populated at the real ingest call site, not just that * the standalone hashing logic works in isolation. * * select_node_id-nonzero case intentionally NOT covered here: producing a * real interned CSE select-root (root->node_id != 0) requires * sel_node_table_init() + sel_optimize_set(1) + UNIT_TEST-gated * sel_optimize_test_reset() teardown -- machinery no existing fixture in * this file sets up, and this file's tests are never torn down by design * (see file header comment), so enabling CSE mode here would leak into * every other test in this binary via the process-global select intern * table. The underlying select_node_id assignment logic itself is already * fully covered by test_resreq_fingerprint.c's * test_resreq_fingerprint_select_node_id_stable_for_identical_select -- * this is a documented gap, not a missed case, per task 1.2's own escape * hatch for exactly this situation. */ static void test_core_build_primary_schedobj_populates_resreq_fp_from_parsed_resreq(void) { xce_sched_obj_t obj; xce_primary_resreq_t resreq; char* errmsg = NULL; int rc; memset(&obj, 0, sizeof(obj)); memset(&resreq, 0, sizeof(resreq)); rc = parse_resreq_primary(&resreq, (char*)"rusage[mem=20:duration=300:decay=1]", NULL, &errmsg); TEST_ASSERT_TRUE(rc >= 0); TEST_ASSERT_NULL(resreq.select); /* Non-degenerate: a real hash over non-empty rusage content -- rules * out a trivial "both sides are 0" false pass below. */ TEST_ASSERT_NOT_EQUAL_UINT64(0, resreq.rusage_hash); core_build_primary_schedobj(&obj, &resreq); TEST_ASSERT_EQUAL_INT(XCE_RESREQ_PRIMARY, obj.tag); TEST_ASSERT_NOT_NULL(obj.u.primary); TEST_ASSERT_EQUAL_PTR(&resreq, obj.u.primary->resreq); /* The fix under test: resreq_fp must mirror the parsed resreq's * min/max tasks, select_node_id (0 -- no select expression here), and * rusage_hash -- not sit at calloc's all-zero default. */ TEST_ASSERT_EQUAL_INT(resreq.minTasks, obj.u.primary->resreq_fp.min_tasks); TEST_ASSERT_EQUAL_INT(resreq.maxTasks, obj.u.primary->resreq_fp.max_tasks); TEST_ASSERT_EQUAL_UINT32(0, obj.u.primary->resreq_fp.select_node_id); TEST_ASSERT_EQUAL_UINT64(resreq.rusage_hash, obj.u.primary->resreq_fp.rusage_hash); kill_primary_resreq(&resreq); FREEUP(errmsg); } /* * resreq-maxtasks-wiring task 2.1 (bug-1770, Req 1.1): a job's real * requested task count (job_submit_data_t.mintasks/maxtasks, already * correctly threaded onto xce_job_t.submit by the time a JOBS_PUSH event * reaches core_build_job()) must reach the standalone primary resreq the * allocator actually dispatches on (job->resreq->u.primary->minTasks/ * maxTasks) -- not parse_resreq_primary()'s always-1 hardcoded default for * every freshly-parsed resreq. Before this fix, core_build_job() never * applied the new apply_job_task_count() helper (src/lib/resreq.c, task * 1), so a job submitted with e.g. -n 4,6 silently ingested with * minTasks==maxTasks==1, permanently unreachable by scheduler-allocation's * parjob_allocator. */ static void test_core_build_job_wires_real_task_count_into_resreq_primary(void) { xce_sched_core_t core; xce_core_event_t event; xce_job_t* job1; xce_sched_job_t* tracked; int rc; fixture_init(&core, &g_full_plugin); job1 = make_job(4160, JOB_STAT_PEND, "normal", "alice", "proj1"); job1->submit->mintasks = 4; job1->submit->maxtasks = 6; 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, 4160); TEST_ASSERT_NOT_NULL(tracked); TEST_ASSERT_NOT_NULL(tracked->resreq); TEST_ASSERT_EQUAL_INT(XCE_RESREQ_PRIMARY, tracked->resreq->tag); TEST_ASSERT_NOT_NULL(tracked->resreq->u.primary); /* The fix under test: the standalone primary's minTasks/maxTasks must * reflect the job's real submitted task count, not parse_resreq_ * primary()'s always-1 default. */ TEST_ASSERT_EQUAL_INT(4, tracked->resreq->u.primary->minTasks); TEST_ASSERT_EQUAL_INT(6, tracked->resreq->u.primary->maxTasks); } /* resreq-maxtasks-wiring task 2.1 (Req 1.2): a job submitted without an * explicit task count (make_job()'s default mintasks==maxtasks==1) must * continue to ingest with minTasks==maxTasks==1 -- zero behavioral change * for the common single-task case. */ static void test_core_build_job_default_task_count_still_one(void) { xce_sched_core_t core; xce_core_event_t event; xce_job_t* job1; xce_sched_job_t* tracked; int rc; fixture_init(&core, &g_full_plugin); job1 = make_job(4161, 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, 4161); TEST_ASSERT_NOT_NULL(tracked); TEST_ASSERT_NOT_NULL(tracked->resreq); TEST_ASSERT_EQUAL_INT(XCE_RESREQ_PRIMARY, tracked->resreq->tag); TEST_ASSERT_NOT_NULL(tracked->resreq->u.primary); TEST_ASSERT_EQUAL_INT(1, tracked->resreq->u.primary->minTasks); TEST_ASSERT_EQUAL_INT(1, tracked->resreq->u.primary->maxTasks); } /* 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_with_allocation_survives_stale_nonrun_resync); RUN_TEST(test_jobs_push_duplicate_jobid_terminal_resync_releases_real_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_job_rusage_reconciles_mem_reservation_against_usage); RUN_TEST(test_config_push_decomposes_queues_resources_and_resvals); RUN_TEST(test_core_build_primary_schedobj_populates_resreq_fp_from_parsed_resreq); RUN_TEST(test_core_build_job_wires_real_task_count_into_resreq_primary); RUN_TEST(test_core_build_job_default_task_count_still_one); 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(); }