lei.chen/myxce
1
0
Fork 1
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions
myxce/src/scheduler/testsuite/test_snapshot_save_load.c
Rongsong Shen 7481cedb3a
test(scheduler-snapshot-job-allocation): full cross-instance round-trip proof 
Task 3.1: comprehensive integration test proving tasks 1.1/1.2/2.1/2.2
work together end to end -- a RUNNING job (multi-host, multi-resource,
plus a genuine shared-resource-pool component) and a PENDING job saved
together and loaded into a second, independent core. Asserts:

- Both jobs get a valid, non-NULL schedobj in the restored core.
- The running job's allocation/accounting content-match the source,
  including the shared-pool component.
- Worker resInUse and core sharedInUse, computed via the real
  sched_charge_account() chain (white-box included from resource.c,
  following the established test_resource_accounting.c/test.c
  link-only-stub convention), match exactly between source and
  restored cores -- the actual proof of Req 1.2/1.3, not just that
  job->accounts holds the right values.
- Destroying the source core's job list afterward does not corrupt or
  crash the restored core (Req 1.4's independence guarantee exercised
  via real teardown, not just pointer-distinctness checks).

Test-only task; no production code changed. Independently verified via
a live RED/GREEN toggle against task 2.2's load-side restoration block.

Requirements: 1.1, 1.2, 1.3, 1.4, 2.1, 2.3, 2.4
2026-08-27 15:22:04 +00:00

2150 lines
89 KiB
C
Raw Blame History

/*
* test_snapshot_save_load.c - the cross-core behavioral-equivalence proof
* for xce_sched_core_save_snapshot()/xce_sched_core_load_snapshot()
* (scheduler-core spec, task 8.1, Req 5 AC 21 / Req 6 AC 5).
*
* Observable under test (task 8.1's own text): "save immediately followed
* by load into a freshly-initialized core produces a core whose observable
* scheduling behavior (next session's decisions, given the same events) is
* identical to the pre-save core." This is a genuinely CROSS-CORE proof,
* not a field-level round-trip check alone: two fully independent
* xce_sched_core_t instances (core_a, the original; core_b, freshly
* xce_sched_core_init()'d and populated purely via
* xce_sched_core_load_snapshot()) are each driven through the SAME
* push-events -> go -> drain-decisions -> end-session cycle
* test_core_full_cycle.c (task 4.6) already established as the correct,
* real way to drive a session -- a real picker pthread plus a stand-in
* allocator (this file's own run_identical_session(), the test function's
* own thread) reproducing sched_job_picker()/sched_allocator()'s real
* per-session bodies, NOT a hand-rolled shortcut. The two cores' resulting
* decision sequences (op + jobid, in order) are asserted byte-for-byte
* identical. Field-level assertions (queues/resources/resvals/workers/
* users/projects/prng_seed/jobs restored onto core_b) are ALSO included --
* cheaper to debug from, per this task's own text, but not a substitute
* for the cross-core proof above.
*
* White-box #include of core.c, core_clock.c, core_events.c, core_session.c,
* decision.c (same convention/file set as test_core_full_cycle.c) PLUS the
* three files under test: snapshot.c, snapshot_wire.c, snapshot_load.c
* (task 8.1's own three-file split -- see snapshot.c's file header
* comment). Stubs below are the exact same set test_core_full_cycle.c
* already establishes needing for this file combination (kill_sched_*
* family, plugin_unregister, sched_worker_release_host_ctx,
* sched_resolve_job_picker)
* -- UNLIKE that file, sched_worker_release_host_ctx is genuinely invoked
* here (core_reset_workers(), core_session.c) since this test adds real
* workers to both cores; it remains a safe no-op regardless.
* makeResource()/parse_resreq()/format_resreq()/key_find()/zarray_*() are
* all REAL, already-linked libxce/librtti symbols (src/lib/resobj.c,
* src/lib/resreq.c) -- no stub needed for any of them.
*
* Two independent stub picker plugin instances (g_stub_plugin_a/
* g_stub_plugin_b), each with its OWN private stub_picker_state_t reached
* via xce_sched_plugin_t's existing `data` field (the same mechanism
* builtin_fcfs.c's real fcfs_plugin_data_t uses) -- this is what lets
* core_a and core_b each build and walk their OWN independent candidate
* list without cross-contaminating a shared global, while still reusing
* the exact same core-parameterized picker-thread function
* (test_picker_thread_fn(), copied from test_core_full_cycle.c verbatim)
* for both: sched_resolve_job_picker(core, ...) naturally resolves to
* whichever plugin instance is registered in THAT core's own
* core->plugins kvmgr.
*
* scheduler-snapshot-job-allocation task 3.1's own test (below, the file's
* newest addition) ALSO white-box #includes ../src/resource.c -- the real
* sched_charge_account()/sched_update_account()/sched_update_res_accounts()
* chain sched_account_jobs() (scheduler.c, NOT included here -- too heavy a
* dependency surface for this otherwise-lean core-only TU) calls downstream.
* Task 3.1's test provides its own real, verbatim-copied reimplementation of
* sched_account_jobs() itself, the same "real reimplementation" technique
* this file already uses for sched_resolve_job_picker() above and
* test_sim_contention_parity.c already establishes for this exact function
* (see that file's own header comment: "a no-op stub would defeat the
* test's own premise"). Pulling in resource.c supplies sched_charge_account()
* for real but ALSO supplies a REAL sched_worker_release_host_ctx()
* (superseding this file's own former no-op stub -- removed below, since a
* duplicate definition would be a link error); it also needs two new
* link-only stubs (plugin_prepare_allocation, sched_ctx_get_uncomit_host)
* for resource.c's own never-invoked sched_try_alloc_on_host_() dependency
* chain -- the same two-stub pair test_resource_accounting.c (and test.c)
* already establish needing for the identical reason. (Note:
* test_sim_contention_parity.c only needs one of the two,
* sched_ctx_get_uncomit_host -- it gets plugin_prepare_allocation for real
* via its own #include "../src/plugin.c", so it is not the matching
* precedent for this pair together.)
*/
#include <errno.h>
#include <pthread.h>
#include <stdatomic.h>
#include <stdint.h>
#include <stdlib.h>
#include <string.h>
#include <unity/unity.h>
#include "kvmgr.h"
#include "preason.h"
#include "scheduler.h"
#include "xcealloc.h"
#include "xstring.h"
/* Pull in the real xce_sched_core_init()/xce_core_push_events()/
* xce_core_go()/xce_core_end_session()/xce_core_set_decision_cb()/
* xce_sched_core_save_snapshot()/xce_sched_core_load_snapshot() under test
* into this TU (white-box #include, same convention as
* test_core_full_cycle.c). */
#include "../src/core.c"
#include "../src/core_clock.c"
#include "../src/core_events.c"
#include "../src/core_session.c"
#include "../src/decision.c"
#include "../src/resource.c"
#include "../src/snapshot.c"
#include "../src/snapshot_load.c"
#include "../src/snapshot_wire.c"
void kill_sched_job(void* arg) { (void)arg; }
void kill_sched_queue(void* arg) { (void)arg; }
void kill_sched_worker(void* arg) { (void)arg; }
void kill_sched_user(void* arg) { (void)arg; }
void kill_sched_project(void* arg) { (void)arg; }
void kill_sched_resval(xce_val_t* val) { (void)val; }
void plugin_unregister(void* arg) { (void)arg; }
/* core_reset_workers()'s (core_session.c) linker dependency -- REAL now
* (resource.c, #include'd above for task 3.1's own test), superseding what
* used to be this file's own no-op stub of the identical name (a duplicate
* definition would be a link error now that resource.c supplies the real
* one). Still behaviorally transparent to every pre-existing test in this
* file: every worker's host_eval_ctx starts NULL (calloc()'d), and
* free_host_eval_ctx(NULL) is a documented no-op (resource.c), so the real
* body performs exactly the same "no-op" every earlier test already relied
* on -- it is only genuinely exercised by task 3.1's own test below.
*
* resource.c's sched_try_alloc_on_host_() (never invoked by this file --
* no test here drives live host allocation) still needs two of its own
* callees resolved at link time: plugin_prepare_allocation (plugin.c, not
* included here) and sched_ctx_get_uncomit_host (schedobj.c, not included
* here). Same two link-only, never-invoked stubs test_resource_accounting.c
* (and test.c) already establish needing for the identical reason --
* test_sim_contention_parity.c is not this pair's precedent, since it only
* needs sched_ctx_get_uncomit_host itself (it gets plugin_prepare_allocation
* for real via its own #include "../src/plugin.c"). */
int plugin_prepare_allocation(void* arg, xce_sched_ctx_t* ctx)
{
(void)arg;
(void)ctx;
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;
}
/* Real reimplementation of schedobj.c's sched_resolve_job_picker -- see
* file header comment (same rationale/algorithm as
* test_core_full_cycle.c's own copy). Parameterized purely on `core`, so it
* naturally resolves core_a's own registered plugin for core_a and
* core_b's own registered plugin for core_b. */
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; }
/* ---------------------------------------------------------------------- */
/* Stub picker: per-instance private state via xce_sched_plugin_t->data */
/* (see file header comment) -- each of g_stub_state_a/g_stub_state_b */
/* independently accumulates ITS OWN core's candidate list. */
/* ---------------------------------------------------------------------- */
#define STUB_MAX_CANDIDATES 8
typedef struct {
xce_sched_job_t* candidates[STUB_MAX_CANDIDATES];
int ncandidates;
int cursor;
} stub_picker_state_t;
static stub_picker_state_t g_stub_state_a;
static stub_picker_state_t g_stub_state_b;
static int stub_on_job_push(void* picker, xce_sched_job_t* job)
{
stub_picker_state_t* st
= (stub_picker_state_t*)((xce_sched_plugin_t*)picker)->data;
if (st->ncandidates < STUB_MAX_CANDIDATES) {
st->candidates[st->ncandidates++] = job;
}
return 0;
}
static int stub_on_queue_update(void* picker, xce_sched_queue_t* queue)
{
(void)picker;
(void)queue;
return 0;
}
static int stub_on_job_state_change(void* picker, xce_sched_job_t* job,
uint32_t state)
{
(void)picker;
(void)job;
(void)state;
return 0; /* not exercised by this test's event batch */
}
static int stub_on_job_update(void* picker, xce_sched_job_t* job)
{
(void)picker;
(void)job;
return 0; /* not exercised by this test's event batch */
}
/* begin_session()/next_job() run in the real picker thread ONLY -- same
* "no lock needed against on_dispatch()/on_pend()" reasoning
* test_core_full_cycle.c's own header comment gives (this stub's
* on_dispatch/on_pend below are pure no-ops, touching nothing). */
static int stub_begin_session(void* picker)
{
stub_picker_state_t* st
= (stub_picker_state_t*)((xce_sched_plugin_t*)picker)->data;
st->cursor = 0; /* session-aware requirement (plugin.h): never carry the
* iteration cursor over from a previous session. */
return 0;
}
static xce_sched_job_t* stub_next_job(void* picker)
{
stub_picker_state_t* st
= (stub_picker_state_t*)((xce_sched_plugin_t*)picker)->data;
if (st->cursor >= st->ncandidates) {
return NULL;
}
return st->candidates[st->cursor++];
}
static int stub_end_session(void* picker)
{
(void)picker;
return 0;
}
/* Post-allocation hooks (plugin.h: allocator thread, in production; the
* main thread's stand-in allocator here, run_identical_session() below) --
* pure no-ops: this test's decision-sequence comparison is driven entirely
* off driver_ctx->seen[] (the real xce_decision_t stream), not off these
* hooks. */
static int stub_on_dispatch(void* picker, uint64_t jobid)
{
(void)picker;
(void)jobid;
return 0;
}
static int stub_on_pend(void* picker, uint64_t jobid)
{
(void)picker;
(void)jobid;
return 0;
}
static xce_job_picker_iface_t g_stub_picker_iface = {
.init = NULL,
.fini = NULL,
.begin_session = stub_begin_session,
.next_job = stub_next_job,
.end_session = stub_end_session,
.on_dispatch = stub_on_dispatch,
.on_pend = stub_on_pend,
.on_queue_update = stub_on_queue_update,
.on_job_push = stub_on_job_push,
.on_job_state_change = stub_on_job_state_change,
.on_job_update = stub_on_job_update,
.on_config = NULL, /* MAY be NULL (plugin.h) -- not pushed */
.on_job_rusage = NULL, /* MAY be NULL (plugin.h) -- not pushed */
};
/* Same picker-plugin NAME for both -- safe: each lives in its own core's
* OWN core->plugins kvmgr (see file header comment), so there is no name
* collision. */
static xce_sched_plugin_t g_stub_plugin_a
= { .name = "test_stub_picker",
.picker_iface = &g_stub_picker_iface,
.data = &g_stub_state_a };
static xce_sched_plugin_t g_stub_plugin_b
= { .name = "test_stub_picker",
.picker_iface = &g_stub_picker_iface,
.data = &g_stub_state_b };
/* ---------------------------------------------------------------------- */
/* Picker thread: reproduces scheduler.c's sched_job_picker() real */
/* per-session body verbatim (copied from test_core_full_cycle.c's own */
/* test_picker_thread_fn -- identical, core-parameterized, single-shot). */
/* ---------------------------------------------------------------------- */
static void* test_picker_thread_fn(void* arg)
{
xce_sched_core_t* core = (xce_sched_core_t*)arg;
xce_sched_plugin_t* job_picker = NULL;
int rc;
int go = 0;
uint64_t jobid;
rc = sched_resolve_job_picker(core, &job_picker);
if (rc < 0) {
return NULL;
}
cbuf_get(&(core->wbuf), &go);
job_picker->picker_iface->begin_session(job_picker);
for (;;) {
xce_sched_job_t* job = job_picker->picker_iface->next_job(job_picker);
if (job == NULL) {
break;
}
jobid = job->jobid;
cbuf_put(&(core->jbuf), &jobid);
}
jobid = 0;
cbuf_put(&(core->jbuf), &jobid);
job_picker->picker_iface->end_session(job_picker);
return NULL;
}
/* ---------------------------------------------------------------------- */
/* Fixture helpers */
/* ---------------------------------------------------------------------- */
static void fixture_init(xce_sched_core_t* core, xce_sched_plugin_t* plugin)
{
int rc;
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;
}
/* make_job_with_resreq - same fixture as make_job() above, but with a
* caller-supplied resreq string, so task 1.2's schedobj tests can build
* PRIMARY/COMPOUND/ALTER-tagged jobs (and a deliberately-unparseable one)
* without duplicating make_job()'s own field-population logic. */
static xce_job_t* make_job_with_resreq(uint64_t jobid, uint32_t state,
char* queue, char* user, char* project,
char* resreq)
{
xce_job_t* j = make_job(jobid, state, queue, user, project);
j->submit->resreq = resreq;
return j;
}
#define MAX_RECORDED_DECISIONS 8
typedef struct {
int calls;
xce_decision_t* seen[MAX_RECORDED_DECISIONS];
} test_driver_ctx_t;
static int test_on_decision(void* driver_ctx, xce_decision_t* decision)
{
test_driver_ctx_t* ctx = (test_driver_ctx_t*)driver_ctx;
TEST_ASSERT_NOT_NULL(decision);
if (ctx->calls < MAX_RECORDED_DECISIONS) {
ctx->seen[ctx->calls] = decision;
}
ctx->calls++;
return 0;
}
static int drain_and_dispatch(xce_sched_core_t* core)
{
int delivered = 0;
for (;;) {
xce_decision_t* decision = NULL;
cbuf_get(&(core->decisions), &decision);
if (decision == NULL) {
break;
}
core->on_decision(core->driver_ctx, decision);
delivered++;
}
return delivered;
}
/* run_identical_session - drives ONE core through a full session: a real
* picker thread + a stand-in allocator (this function itself, mirroring
* test_core_full_cycle.c's own Phase 3/4/5) that decides DISPATCH (even
* jobid) vs PEND (odd jobid) -- a deterministic, resource-matching-free
* rule, same scope boundary test_core_full_cycle.c's own header comment
* documents for why real sched_allocator()/resource.c internals are out of
* a pure core-contract test's reach. This is what makes "identical events
* in, identical decisions out" a meaningful, reproducible comparison
* between core_a and core_b: given the SAME candidate jobids in the SAME
* order (which the picker rebuild on load is what this whole task proves),
* this stand-in allocator's own logic is a pure function of jobid, so it
* necessarily produces the SAME op+jobid sequence for both cores. */
static void run_identical_session(xce_sched_core_t* core)
{
pthread_t picker_tid;
int rc;
uint64_t jobid;
rc = pthread_create(&picker_tid, NULL, test_picker_thread_fn, core);
TEST_ASSERT_EQUAL_INT(0, rc);
rc = xce_core_go(core);
TEST_ASSERT_EQUAL_INT(0, rc);
for (;;) {
xce_sched_job_t* job;
xce_decision_t* decision = NULL;
cbuf_get(&(core->jbuf), &jobid);
if (jobid == 0) {
cbuf_put(&(core->decisions), &decision);
break;
}
job = sched_get_job(core, jobid);
TEST_ASSERT_NOT_NULL(job);
if ((jobid % 2) == 0) {
xce_job_alloc_t* alloc
= (xce_job_alloc_t*)calloc(1, sizeof(*alloc));
alloc->jobid = jobid;
decision = make_dispatch_decision(alloc);
} else {
xce_pendreason_t* reason = make_pendreason(jobid);
decision = make_pend_decision(reason);
}
TEST_ASSERT_NOT_NULL(decision);
cbuf_put(&(core->decisions), &decision);
}
rc = pthread_join(picker_tid, NULL);
TEST_ASSERT_EQUAL_INT(0, rc);
(void)drain_and_dispatch(core);
rc = xce_core_end_session(core);
TEST_ASSERT_EQUAL_INT(0, rc);
}
/* ---------------------------------------------------------------------- */
/* Guard tests (Part 3's own "-EBUSY/-EINVAL, directly and cheaply, unit- */
/* level" request) -- no session, no picker plugin needed: both guards */
/* fire before either function touches state or resolves a picker (core.h */
/* 's own doc comment). */
/* ---------------------------------------------------------------------- */
static void
test_snapshot_guards_reject_null_args_and_busy_session_without_mutating_state(
void)
{
xce_sched_core_t core;
sched_snapshot_data_t snap;
sched_snapshot_data_t empty_in;
int rc;
rc = xce_sched_core_init(&core);
TEST_ASSERT_EQUAL_INT(0, rc);
memset(&empty_in, 0, sizeof(empty_in));
/* -EINVAL: NULL core, or NULL out/in, for both directions. */
rc = xce_sched_core_save_snapshot(NULL, &snap);
TEST_ASSERT_EQUAL_INT(-EINVAL, rc);
rc = xce_sched_core_save_snapshot(&core, NULL);
TEST_ASSERT_EQUAL_INT(-EINVAL, rc);
rc = xce_sched_core_load_snapshot(NULL, &empty_in);
TEST_ASSERT_EQUAL_INT(-EINVAL, rc);
rc = xce_sched_core_load_snapshot(&core, NULL);
TEST_ASSERT_EQUAL_INT(-EINVAL, rc);
/* -EBUSY: session_active != 0, checked BEFORE any mutation -- a
* rejected call is a true no-op (core.h's own doc comment for both
* functions, "ZERO state touched"). Directly flip the atomic (same
* precondition-rejection technique test_core_full_cycle.c's own Phase 2
* uses for xce_core_go()/xce_core_push_events()) rather than driving a
* real session, since only the guard itself is under test here. */
atomic_store(&(core.session_active), 1);
memset(&snap, 0xAA, sizeof(snap));
rc = xce_sched_core_save_snapshot(&core, &snap);
TEST_ASSERT_EQUAL_INT(-EBUSY, rc);
/* sentinel bytes untouched -- save_snapshot() never reached its own
* memset(out, 0, ...), proving the guard runs before any mutation. */
TEST_ASSERT_EQUAL_UINT8(0xAA, ((uint8_t*)&snap)[0]);
rc = xce_sched_core_load_snapshot(&core, &empty_in);
TEST_ASSERT_EQUAL_INT(-EBUSY, rc);
/* no queue was created -- proves ZERO state was touched, not just that
* the return code was right. */
TEST_ASSERT_NULL(sched_get_queue(&core, "normal"));
atomic_store(&(core.session_active), 0);
}
/* ---------------------------------------------------------------------- */
/* Guard tests against a GENUINE active session (task 8.2, Req 5 AC 21) -- */
/* the guard test above proves the check fires against a hand-set atomic; */
/* this one drives a real session the same way */
/* test_full_cycle_push_go_drain_end_with_precondition_rejections() */
/* (test_core_full_cycle.c) proves xce_core_go()/xce_core_push_events()'s */
/* own -EBUSY/-EINVAL guards reject reentrant calls while GENUINELY */
/* active -- a real picker pthread plus a real xce_core_go() call, not a */
/* synthetic session_active flip. */
/* ---------------------------------------------------------------------- */
static void
test_snapshot_save_and_load_reject_during_genuinely_active_session(void)
{
xce_sched_core_t core;
pthread_t picker_tid;
xce_core_event_t events[2];
xce_queue_t wire_queue;
xce_job_t* job;
sched_snapshot_data_t snap;
sched_snapshot_data_t reject_in;
xce_queue_t reject_wire_queue;
xce_queue_t* reject_queues[1];
xce_sched_queue_t* q_after;
xce_pendreason_t* reason;
xce_decision_t* decision;
test_driver_ctx_t driver_ctx;
int ncandidates_before;
int rc;
uint64_t jobid;
memset(&driver_ctx, 0, sizeof(driver_ctx));
memset(&g_stub_state_a, 0, sizeof(g_stub_state_a));
fixture_init(&core, &g_stub_plugin_a);
rc = xce_core_set_decision_cb(&core, test_on_decision, &driver_ctx);
TEST_ASSERT_EQUAL_INT(0, rc);
/* ===================================================================
* Phase 1: real state while the session is inactive -- one open queue,
* one pending job (same shape as test_core_full_cycle.c's own Phase 1).
* =================================================================== */
memset(&wire_queue, 0, sizeof(wire_queue));
wire_queue.queue = "normal";
wire_queue.priority = 1;
wire_queue.status = QUEUE_STAT_OPEN;
job = make_job(301, JOB_STAT_PEND, "normal", "alice", "proj1");
memset(events, 0, sizeof(events));
events[0].type = XCE_CORE_EVENT_QUEUE_UPDATE;
events[0].u.queue_update.queue = &wire_queue;
events[1].type = XCE_CORE_EVENT_JOBS_PUSH;
events[1].u.jobs_push.njobs = 1;
events[1].u.jobs_push.jobs = &job;
rc = xce_core_push_events(&core, events, 2);
TEST_ASSERT_EQUAL_INT(0, rc);
TEST_ASSERT_EQUAL_INT(1, g_stub_state_a.ncandidates);
ncandidates_before = g_stub_state_a.ncandidates;
/* ===================================================================
* Phase 2: spawn the real picker thread, then start a GENUINE session
* via xce_core_go(). While it is active, both snapshot functions must
* be rejected with -EBUSY (distinct from the -EINVAL NULL-arg case
* task 8.1's own guard test above already covers), and must leave
* state genuinely untouched -- not just "returned an error code".
* =================================================================== */
rc = pthread_create(&picker_tid, NULL, test_picker_thread_fn, &core);
TEST_ASSERT_EQUAL_INT(0, rc);
rc = xce_core_go(&core);
TEST_ASSERT_EQUAL_INT(0, rc);
TEST_ASSERT_EQUAL_INT(
1, atomic_load_explicit(&(core.session_active), memory_order_acquire));
/* save: sentinel bytes untouched -- proves the guard runs before
* save_snapshot()'s own memset(out, 0, ...), against a real session. */
memset(&snap, 0xAA, sizeof(snap));
rc = xce_sched_core_save_snapshot(&core, &snap);
TEST_ASSERT_EQUAL_INT(-EBUSY, rc);
TEST_ASSERT_EQUAL_UINT8(0xAA, ((uint8_t*)&snap)[0]);
/* load: a distinguishable payload (a queue name never pushed to this
* core) proves the load never ran -- if the guard were bypassed, this
* queue would be added and the picker would gain a 2nd candidate. */
memset(&reject_wire_queue, 0, sizeof(reject_wire_queue));
reject_wire_queue.queue = "should-not-appear";
reject_wire_queue.priority = 99;
reject_wire_queue.status = QUEUE_STAT_OPEN;
reject_queues[0] = &reject_wire_queue;
memset(&reject_in, 0, sizeof(reject_in));
reject_in.queues = reject_queues;
reject_in.n_queues = 1;
rc = xce_sched_core_load_snapshot(&core, &reject_in);
TEST_ASSERT_EQUAL_INT(-EBUSY, rc);
TEST_ASSERT_NULL(sched_get_queue(&core, "should-not-appear"));
q_after = sched_get_queue(&core, "normal");
TEST_ASSERT_NOT_NULL(q_after);
TEST_ASSERT_EQUAL_INT(1, q_after->priority);
TEST_ASSERT_EQUAL_INT(ncandidates_before, g_stub_state_a.ncandidates);
/* both rejections happened without the session ever having ended. */
TEST_ASSERT_EQUAL_INT(
1, atomic_load_explicit(&(core.session_active), memory_order_acquire));
/* ===================================================================
* Phase 3: drain the still-running real session to completion -- same
* drain protocol run_identical_session() above uses (decision content
* is not under test here, only that the session ends cleanly).
* =================================================================== */
for (;;) {
decision = NULL;
cbuf_get(&(core.jbuf), &jobid);
if (jobid == 0) {
cbuf_put(&(core.decisions), &decision);
break;
}
reason = make_pendreason(jobid);
decision = make_pend_decision(reason);
TEST_ASSERT_NOT_NULL(decision);
cbuf_put(&(core.decisions), &decision);
}
rc = pthread_join(picker_tid, NULL);
TEST_ASSERT_EQUAL_INT(0, rc);
(void)drain_and_dispatch(&core);
rc = xce_core_end_session(&core);
TEST_ASSERT_EQUAL_INT(0, rc);
TEST_ASSERT_EQUAL_INT(
0, atomic_load_explicit(&(core.session_active), memory_order_acquire));
}
/* ---------------------------------------------------------------------- */
/* The Observable, end to end. */
/* ---------------------------------------------------------------------- */
/*
* test_snapshot_save_then_load_produces_behaviorally_equivalent_core - see
* file header comment for the full design.
*/
static void
test_snapshot_save_then_load_produces_behaviorally_equivalent_core(void)
{
xce_sched_core_t core_a;
xce_sched_core_t core_b;
sched_snapshot_data_t snap;
xce_core_event_t events[3];
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_normal;
xce_queue_t wire_queue_batch;
xce_queue_t* queues[2];
xce_worker_t wire_worker_a;
xce_worker_t wire_worker_b;
xce_worker_t* workers[2];
xce_val_t worker_a_cpu_val;
xce_val_t worker_b_cpu_val;
xce_name_val_t worker_a_resval;
xce_name_val_t worker_b_resval;
xce_name_val_t* worker_a_resvals[1];
xce_name_val_t* worker_b_resvals[1];
xce_job_t* jobs[3];
xce_sched_project_t* proj1_a;
xce_resource_t* found_resource_a;
xce_resource_t* found_resource_b;
xce_val_t* shared_val_b;
xce_sched_queue_t* q_normal_b;
xce_sched_queue_t* q_batch_b;
xce_sched_worker_t* worker_a_b;
xce_sched_worker_t* worker_b_b;
xce_val_t* worker_a_resval_b;
xce_val_t* worker_b_resval_b;
xce_sched_project_t* proj1_b;
xce_sched_project_t* proj2_b;
xce_sched_job_t* job201_b;
xce_sched_job_t* job202_b;
xce_sched_job_t* job203_b;
test_driver_ctx_t driver_ctx_a;
test_driver_ctx_t driver_ctx_b;
int rc;
int i;
memset(&driver_ctx_a, 0, sizeof(driver_ctx_a));
memset(&driver_ctx_b, 0, sizeof(driver_ctx_b));
memset(&g_stub_state_a, 0, sizeof(g_stub_state_a));
memset(&g_stub_state_b, 0, sizeof(g_stub_state_b));
/* ===================================================================
* Phase 1: build core_a with REAL state -- jobs, queues, workers,
* resources, resvals, a registered picker plugin.
* =================================================================== */
fixture_init(&core_a, &g_stub_plugin_a);
core_a.prng_seed = 0xDEADBEEFULL; /* direct field poke -- prng_seed has
* no wire event of its own (task
* 8.1's own scope note, core.h);
* seeded here to a non-zero value so
* the round-trip assertion below is
* meaningful, not a trivial 0==0. */
memset(&wire_resource, 0, sizeof(wire_resource));
wire_resource.name = "cpu";
resources[0] = &wire_resource;
memset(&resval_value, 0, sizeof(resval_value));
/* .which MUST be XCE_VAL_NUMERIC, not left at the memset-zero default
* -- XCE_VAL_STRING is literally 0 (xce/types.h), and dup_val()
* (snapshot_wire.c, genuinely exercised by save_snapshot() below)
* branches on .which to decide whether .u.svalue is a string pointer
* to copy: leaving this at 0 makes it misinterpret .u.ivalue32's raw
* integer bits as a char* and crash inside makeString(). */
resval_value.which = XCE_VAL_NUMERIC;
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_normal, 0, sizeof(wire_queue_normal));
wire_queue_normal.queue = "normal";
wire_queue_normal.priority = 3;
wire_queue_normal.status = QUEUE_STAT_OPEN;
memset(&wire_queue_batch, 0, sizeof(wire_queue_batch));
wire_queue_batch.queue = "batch";
wire_queue_batch.priority = 1;
wire_queue_batch.status = QUEUE_STAT_OPEN;
queues[0] = &wire_queue_normal;
queues[1] = &wire_queue_batch;
memset(&config, 0, sizeof(config));
config.n_resources = 1;
config.resources = resources;
config.nresvals = 1;
config.resvals = resvals;
config.n_queues = 2;
config.queues = queues;
memset(&worker_a_cpu_val, 0, sizeof(worker_a_cpu_val));
worker_a_cpu_val.which = XCE_VAL_NUMERIC; /* see resval_value's own
* comment above */
worker_a_cpu_val.u.ivalue32 = 4;
memset(&worker_a_resval, 0, sizeof(worker_a_resval));
worker_a_resval.name = "cpu";
worker_a_resval.val = &worker_a_cpu_val;
worker_a_resvals[0] = &worker_a_resval;
memset(&worker_b_cpu_val, 0, sizeof(worker_b_cpu_val));
worker_b_cpu_val.which = XCE_VAL_NUMERIC;
worker_b_cpu_val.u.ivalue32 = 6;
memset(&worker_b_resval, 0, sizeof(worker_b_resval));
worker_b_resval.name = "cpu";
worker_b_resval.val = &worker_b_cpu_val;
worker_b_resvals[0] = &worker_b_resval;
memset(&wire_worker_a, 0, sizeof(wire_worker_a));
wire_worker_a.name = "hostA";
wire_worker_a.status = 0; /* available: IS_WORKER_UNAVAIL
* checks a specific unset bit */
wire_worker_a.nresvals_to_sched = 1;
wire_worker_a.resvals_to_sched = worker_a_resvals;
memset(&wire_worker_b, 0, sizeof(wire_worker_b));
wire_worker_b.name = "hostB";
wire_worker_b.status = 0;
wire_worker_b.nresvals_to_sched = 1;
wire_worker_b.resvals_to_sched = worker_b_resvals;
workers[0] = &wire_worker_a;
workers[1] = &wire_worker_b;
jobs[0] = make_job(201, JOB_STAT_PEND, "normal", "alice", "proj1");
jobs[1] = make_job(202, JOB_STAT_PEND, "normal", "bob", "proj1");
jobs[2] = make_job(203, JOB_STAT_PEND, "batch", "carol", "proj2");
memset(events, 0, sizeof(events));
/* CONFIG_PUSH first: registers "cpu" into core_a->resmgr, which
* WORKERS_PUSH (next) needs already present to resolve each worker's
* resvals_to_sched by name. */
events[0].type = XCE_CORE_EVENT_CONFIG_PUSH;
events[0].u.config_push.config = &config;
events[1].type = XCE_CORE_EVENT_WORKERS_PUSH;
events[1].u.workers_push.nworkers = 2;
events[1].u.workers_push.workers = workers;
events[2].type = XCE_CORE_EVENT_JOBS_PUSH;
events[2].u.jobs_push.njobs = 3;
events[2].u.jobs_push.jobs = jobs;
rc = xce_core_push_events(&core_a, events, 3);
TEST_ASSERT_EQUAL_INT(0, rc);
TEST_ASSERT_EQUAL_INT(3, g_stub_state_a.ncandidates);
/* project priority: no wire event of its own -- direct field poke on
* the auto-created project object (get-or-create side effect of
* JOBS_PUSH), same "genuine, but directly-constructed" state
* test_core_events.c's own fixtures already use for config/queue/
* resource wire payloads. proj2 deliberately left at its default (0)
* -- proves the round-trip is field-accurate, not just "non-zero
* survives". */
proj1_a = sched_get_project(&core_a, "proj1");
TEST_ASSERT_NOT_NULL(proj1_a);
proj1_a->priority = 5;
/* ===================================================================
* Phase 2: save core_a's snapshot; field-level sanity checks on the
* snapshot struct itself (cheap, direct evidence of the field-by-field
* mapping snapshot.c's own header comment documents).
* =================================================================== */
memset(&snap, 0, sizeof(snap));
rc = xce_sched_core_save_snapshot(&core_a, &snap);
TEST_ASSERT_EQUAL_INT(0, rc);
TEST_ASSERT_NOT_NULL(snap.jobs);
TEST_ASSERT_EQUAL_INT(3, snap.jobs->count);
TEST_ASSERT_EQUAL_INT(2, snap.n_queues);
TEST_ASSERT_NOT_NULL(snap.workers);
TEST_ASSERT_EQUAL_INT(2, snap.workers->count);
TEST_ASSERT_NOT_NULL(snap.resources);
TEST_ASSERT_EQUAL_INT(1, snap.resources->count);
TEST_ASSERT_NOT_NULL(snap.resvals);
TEST_ASSERT_EQUAL_INT(1, snap.resvals->count);
TEST_ASSERT_EQUAL_INT(3, snap.n_users);
TEST_ASSERT_EQUAL_INT(2, snap.n_projects);
TEST_ASSERT_EQUAL_UINT64(0xDEADBEEFULL, snap.prng_seed);
/* ===================================================================
* Phase 3: build core_b -- a completely separate, freshly-initialized
* instance -- and load the snapshot into it.
* =================================================================== */
fixture_init(&core_b, &g_stub_plugin_b);
rc = xce_sched_core_load_snapshot(&core_b, &snap);
TEST_ASSERT_EQUAL_INT(0, rc);
/* picker rebuild fired on_job_push for every restored job, exactly as
* the original JOBS_PUSH did for core_a above. */
TEST_ASSERT_EQUAL_INT(3, g_stub_state_b.ncandidates);
/* ===================================================================
* Phase 4: field-level assertions -- core_b's restored state matches
* core_a's original values exactly.
* =================================================================== */
q_normal_b = sched_get_queue(&core_b, "normal");
q_batch_b = sched_get_queue(&core_b, "batch");
TEST_ASSERT_NOT_NULL(q_normal_b);
TEST_ASSERT_NOT_NULL(q_batch_b);
TEST_ASSERT_EQUAL_INT(3, q_normal_b->priority);
TEST_ASSERT_EQUAL_INT(1, q_batch_b->priority);
rc = key_find(core_a.resmgr, (void*)"cpu", (void**)&found_resource_a);
TEST_ASSERT_GREATER_THAN_INT(0, rc);
rc = key_find(core_b.resmgr, (void*)"cpu", (void**)&found_resource_b);
TEST_ASSERT_GREATER_THAN_INT(0, rc);
TEST_ASSERT_NOT_NULL(found_resource_a);
TEST_ASSERT_NOT_NULL(found_resource_b);
shared_val_b
= (xce_val_t*)zarray_get(&(core_b.sharedInst), found_resource_b->idx);
TEST_ASSERT_NOT_NULL(shared_val_b);
TEST_ASSERT_EQUAL_INT(8, shared_val_b->u.ivalue32);
worker_a_b = sched_get_worker(&core_b, "hostA");
worker_b_b = sched_get_worker(&core_b, "hostB");
TEST_ASSERT_NOT_NULL(worker_a_b);
TEST_ASSERT_NOT_NULL(worker_b_b);
worker_a_resval_b
= (xce_val_t*)zarray_get(&(worker_a_b->resvals), found_resource_b->idx);
worker_b_resval_b
= (xce_val_t*)zarray_get(&(worker_b_b->resvals), found_resource_b->idx);
TEST_ASSERT_NOT_NULL(worker_a_resval_b);
TEST_ASSERT_NOT_NULL(worker_b_resval_b);
TEST_ASSERT_EQUAL_INT(4, worker_a_resval_b->u.ivalue32);
TEST_ASSERT_EQUAL_INT(6, worker_b_resval_b->u.ivalue32);
TEST_ASSERT_NOT_NULL(sched_get_user(&core_b, "alice"));
TEST_ASSERT_NOT_NULL(sched_get_user(&core_b, "bob"));
TEST_ASSERT_NOT_NULL(sched_get_user(&core_b, "carol"));
proj1_b = sched_get_project(&core_b, "proj1");
proj2_b = sched_get_project(&core_b, "proj2");
TEST_ASSERT_NOT_NULL(proj1_b);
TEST_ASSERT_NOT_NULL(proj2_b);
TEST_ASSERT_EQUAL_INT(5, proj1_b->priority);
TEST_ASSERT_EQUAL_INT(0, proj2_b->priority);
TEST_ASSERT_EQUAL_UINT64(0xDEADBEEFULL, core_b.prng_seed);
job201_b = sched_get_job(&core_b, 201);
job202_b = sched_get_job(&core_b, 202);
job203_b = sched_get_job(&core_b, 203);
TEST_ASSERT_NOT_NULL(job201_b);
TEST_ASSERT_NOT_NULL(job202_b);
TEST_ASSERT_NOT_NULL(job203_b);
TEST_ASSERT_EQUAL_UINT32(JOB_STAT_PEND, job201_b->state);
TEST_ASSERT_EQUAL_STRING("normal", job201_b->queue->name);
TEST_ASSERT_EQUAL_STRING("alice", job201_b->user->name);
TEST_ASSERT_EQUAL_STRING("proj1", job201_b->project->name);
TEST_ASSERT_EQUAL_STRING("batch", job203_b->queue->name);
TEST_ASSERT_EQUAL_STRING("carol", job203_b->user->name);
TEST_ASSERT_EQUAL_STRING("proj2", job203_b->project->name);
/* ===================================================================
* Phase 5: THE cross-core behavioral-equivalence proof. Drive core_a
* and core_b through an IDENTICAL session (same picker-thread protocol,
* same deterministic stand-in-allocator rule) and compare the
* resulting decision sequences.
* =================================================================== */
rc = xce_core_set_decision_cb(&core_a, test_on_decision, &driver_ctx_a);
TEST_ASSERT_EQUAL_INT(0, rc);
rc = xce_core_set_decision_cb(&core_b, test_on_decision, &driver_ctx_b);
TEST_ASSERT_EQUAL_INT(0, rc);
run_identical_session(&core_a);
run_identical_session(&core_b);
TEST_ASSERT_EQUAL_INT(3, driver_ctx_a.calls);
TEST_ASSERT_EQUAL_INT(driver_ctx_a.calls, driver_ctx_b.calls);
for (i = 0; i < driver_ctx_a.calls; i++) {
xce_decision_t* da = driver_ctx_a.seen[i];
xce_decision_t* db = driver_ctx_b.seen[i];
TEST_ASSERT_NOT_NULL(da);
TEST_ASSERT_NOT_NULL(db);
TEST_ASSERT_EQUAL_INT(da->op, db->op);
if (da->op == XCE_DECISION_DISPATCH) {
TEST_ASSERT_EQUAL_UINT64(da->u.alloc->jobid, db->u.alloc->jobid);
} else {
TEST_ASSERT_EQUAL_INT(XCE_DECISION_PEND, da->op);
TEST_ASSERT_EQUAL_UINT64(da->u.reason->jobid, db->u.reason->jobid);
}
}
/* Spot-check the actual expected sequence (jobid parity -- see
* run_identical_session()'s own comment): 201 odd -> PEND, 202 even ->
* DISPATCH, 203 odd -> PEND, in push/candidate order. Proven identical
* for BOTH cores by the loop above; this just pins the concrete values
* so a regression here fails loudly with a readable expected/actual
* pair instead of only an A==B equality. */
TEST_ASSERT_EQUAL_INT(XCE_DECISION_PEND, driver_ctx_a.seen[0]->op);
TEST_ASSERT_EQUAL_UINT64(201, driver_ctx_a.seen[0]->u.reason->jobid);
TEST_ASSERT_EQUAL_INT(XCE_DECISION_DISPATCH, driver_ctx_a.seen[1]->op);
TEST_ASSERT_EQUAL_UINT64(202, driver_ctx_a.seen[1]->u.alloc->jobid);
TEST_ASSERT_EQUAL_INT(XCE_DECISION_PEND, driver_ctx_a.seen[2]->op);
TEST_ASSERT_EQUAL_UINT64(203, driver_ctx_a.seen[2]->u.reason->jobid);
}
/* ---------------------------------------------------------------------- */
/* scheduler-snapshot-job-allocation task 1.2: every job restored from a */
/* snapshot must get a non-NULL, correctly-shaped job->schedobj -- the */
/* same bug-1569 failure shape (core_events.c's own header comment), */
/* still open on THIS load path. Covers a PRIMARY, a COMPOUND, and an */
/* ALTER resource request (design.md "Load-Side Job Reconstruction"). */
/* ---------------------------------------------------------------------- */
static void
test_snapshot_load_builds_schedobj_for_primary_compound_and_alter_resreq(void)
{
xce_sched_core_t core;
sched_snapshot_data_t in;
xce_jobs_data_t jobs_data;
xce_queue_t wire_queue_normal;
xce_queue_t wire_queue_batch;
xce_queue_t* queues[2];
xce_job_t* job_primary;
xce_job_t* job_compound;
xce_job_t* job_alter;
xce_job_t* jobs[3];
xce_sched_job_t* restored_primary;
xce_sched_job_t* restored_compound;
xce_sched_job_t* restored_alter;
int rc;
memset(&g_stub_state_a, 0, sizeof(g_stub_state_a));
fixture_init(&core, &g_stub_plugin_a);
/* sched_get_queue() is NOT get-or-create (unlike sched_get_user()/
* sched_get_project()) -- both queues this test's jobs reference must
* be present in the snapshot input, same as the cross-core test above. */
memset(&wire_queue_normal, 0, sizeof(wire_queue_normal));
wire_queue_normal.queue = "normal";
wire_queue_normal.priority = 1;
wire_queue_normal.status = QUEUE_STAT_OPEN;
memset(&wire_queue_batch, 0, sizeof(wire_queue_batch));
wire_queue_batch.queue = "batch";
wire_queue_batch.priority = 1;
wire_queue_batch.status = QUEUE_STAT_OPEN;
queues[0] = &wire_queue_normal;
queues[1] = &wire_queue_batch;
job_primary = make_job_with_resreq(401, JOB_STAT_PEND, "normal", "alice",
"proj1", "type == any");
/* COMPOUND: '+' between two '{...}' terms (test_resreq.c's own
* test_resreq2() fixture -- proven, real COMPOUND-tagged input). */
job_compound
= make_job_with_resreq(402, JOB_STAT_PEND, "normal", "bob", "proj1",
"2*{rusage[mem=100:duration=20:decay=1]} + "
"4*{rusage[mem=80:duration=30:decay=1]}");
/* ALTER: '||' between two '{compound}' terms (test_resreq.c's own
* test_resreq3() fixture -- proven, real ALTER-tagged input). */
job_alter
= make_job_with_resreq(403, JOB_STAT_PEND, "batch", "carol", "proj2",
"{2*{rusage[mem=100:duration=20:decay=1]} + "
"4*{rusage[mem=80:duration=30:decay=1]}} || "
"{3*{rusage[mem=150:duration=15:decay=1]} + "
"6*{rusage[mem=120:duration=20:decay=1]}}");
jobs[0] = job_primary;
jobs[1] = job_compound;
jobs[2] = job_alter;
memset(&jobs_data, 0, sizeof(jobs_data));
jobs_data.count = 3;
jobs_data.jobs = jobs;
memset(&in, 0, sizeof(in));
in.jobs = &jobs_data;
in.queues = queues;
in.n_queues = 2;
rc = xce_sched_core_load_snapshot(&core, &in);
TEST_ASSERT_EQUAL_INT(0, rc);
restored_primary = sched_get_job(&core, 401);
restored_compound = sched_get_job(&core, 402);
restored_alter = sched_get_job(&core, 403);
TEST_ASSERT_NOT_NULL(restored_primary);
TEST_ASSERT_NOT_NULL(restored_compound);
TEST_ASSERT_NOT_NULL(restored_alter);
/* (a) PRIMARY */
TEST_ASSERT_NOT_NULL(restored_primary->resreq);
TEST_ASSERT_NOT_NULL(restored_primary->schedobj);
TEST_ASSERT_EQUAL_INT(XCE_RESREQ_PRIMARY, restored_primary->schedobj->tag);
TEST_ASSERT_NOT_NULL(restored_primary->schedobj->u.primary);
/* (b) COMPOUND */
TEST_ASSERT_NOT_NULL(restored_compound->resreq);
TEST_ASSERT_NOT_NULL(restored_compound->schedobj);
TEST_ASSERT_EQUAL_INT(XCE_RESREQ_COMPOUND,
restored_compound->schedobj->tag);
TEST_ASSERT_NOT_NULL(restored_compound->schedobj->u.compound);
/* (c) ALTER */
TEST_ASSERT_NOT_NULL(restored_alter->resreq);
TEST_ASSERT_NOT_NULL(restored_alter->schedobj);
TEST_ASSERT_EQUAL_INT(XCE_RESREQ_ALTER, restored_alter->schedobj->tag);
TEST_ASSERT_TRUE(restored_alter->schedobj->u.alters.count > 0);
/* (d) the restored schedobj's user/queue/project string values match
* the job's own (restored via sched_get_user/queue/project), for all
* three -- proving snapshot_build_schedobj() reads the same wire
* fields core_build_schedobj() would have. */
TEST_ASSERT_EQUAL_STRING(restored_primary->user->name,
restored_primary->schedobj->user);
TEST_ASSERT_EQUAL_STRING(restored_primary->queue->name,
restored_primary->schedobj->queue);
TEST_ASSERT_EQUAL_STRING(restored_primary->project->name,
restored_primary->schedobj->project);
TEST_ASSERT_EQUAL_STRING(restored_compound->user->name,
restored_compound->schedobj->user);
TEST_ASSERT_EQUAL_STRING(restored_compound->queue->name,
restored_compound->schedobj->queue);
TEST_ASSERT_EQUAL_STRING(restored_compound->project->name,
restored_compound->schedobj->project);
TEST_ASSERT_EQUAL_STRING(restored_alter->user->name,
restored_alter->schedobj->user);
TEST_ASSERT_EQUAL_STRING(restored_alter->queue->name,
restored_alter->schedobj->queue);
TEST_ASSERT_EQUAL_STRING(restored_alter->project->name,
restored_alter->schedobj->project);
}
/* Regression-safety test (NOT a new feature): a job whose resreq fails to
* parse on load is a pre-existing, out-of-boundary gap (design.md "Known
* limitation") that task 1.2 must NOT try to fix -- it must only make sure
* schedobj construction is skipped rather than crashing on a NULL resreq.
* Proves job->schedobj stays NULL, identical to today's behavior. */
static void
test_snapshot_load_leaves_schedobj_null_when_resreq_fails_to_parse(void)
{
xce_sched_core_t core;
sched_snapshot_data_t in;
xce_jobs_data_t jobs_data;
xce_job_t* bad_job;
xce_job_t* jobs[1];
xce_sched_job_t* restored;
int rc;
memset(&g_stub_state_a, 0, sizeof(g_stub_state_a));
fixture_init(&core, &g_stub_plugin_a);
/* An unbalanced/unterminated compound-resreq brace: the top-level PEG
* grammar (xce_resreq.peg) requires a full input match through EOF; a
* lone unterminated '{' cannot complete as alter/compound/primary and
* leaves trailing unconsumed input, so parse_resreq() genuinely fails
* (unlike plain non-brace garbage, which the grammar's permissive
* primary_resreq rule accepts as a literal select expression and only
* fails later, silently, inside parse_resreq_primary() without
* propagating an error -- confirmed empirically while writing this
* test). */
bad_job = make_job_with_resreq(404, JOB_STAT_PEND, "normal", "alice",
"proj1", "{");
jobs[0] = bad_job;
memset(&jobs_data, 0, sizeof(jobs_data));
jobs_data.count = 1;
jobs_data.jobs = jobs;
memset(&in, 0, sizeof(in));
in.jobs = &jobs_data;
rc = xce_sched_core_load_snapshot(&core, &in);
TEST_ASSERT_EQUAL_INT(0, rc);
restored = sched_get_job(&core, 404);
TEST_ASSERT_NOT_NULL(restored);
/* pre-existing, unrelated gap: resreq parse failure leaves job->resreq
* NULL (FREEUP'd in xce_sched_core_load_snapshot's own job loop) --
* unchanged by this feature. */
TEST_ASSERT_NULL(restored->resreq);
/* THIS feature's guard: schedobj construction is skipped, not crashed
* into, when resreq is NULL. */
TEST_ASSERT_NULL(restored->schedobj);
}
/* ---------------------------------------------------------------------- */
/* scheduler-snapshot-job-allocation task 2.1: build_wire_job() (save side)
* must populate the wire job's `allocs` field via dup_job_alloc() whenever
* the live job has a resource allocation (Req 1.1), and must leave it at
* its calloc()-zeroed NULL default when the live job has none -- the same
* optional-field pattern every other field in build_wire_job() already
* follows (e.g. resreq, only formatted/set when job->resreq != NULL).
* Drives the real save path (xce_sched_core_save_snapshot() ->
* build_wire_job()) against a genuine core rather than calling
* build_wire_job() directly, so this also proves the wiring reaches the
* wire job actually written into the snapshot's jobs array.
* ---------------------------------------------------------------------- */
static void
test_snapshot_save_populates_allocs_for_running_job_with_multi_host_allocation(
void)
{
xce_sched_core_t core;
sched_snapshot_data_t snap;
xce_core_event_t events[2];
xce_queue_t wire_queue;
xce_job_t* wire_job;
xce_sched_job_t* job;
xce_alloc_t* a0;
xce_alloc_t* a1;
xce_host_alloc_t* h0;
xce_host_alloc_t* h1;
xce_host_alloc_t** hallocs;
xce_job_alloc_t* jalloc;
xce_job_t* saved_job;
int rc;
memset(&g_stub_state_a, 0, sizeof(g_stub_state_a));
fixture_init(&core, &g_stub_plugin_a);
memset(&wire_queue, 0, sizeof(wire_queue));
wire_queue.queue = "normal";
wire_queue.priority = 1;
wire_queue.status = QUEUE_STAT_OPEN;
wire_job = make_job(501, JOB_STAT_PEND, "normal", "alice", "proj1");
memset(events, 0, sizeof(events));
events[0].type = XCE_CORE_EVENT_QUEUE_UPDATE;
events[0].u.queue_update.queue = &wire_queue;
events[1].type = XCE_CORE_EVENT_JOBS_PUSH;
events[1].u.jobs_push.njobs = 1;
events[1].u.jobs_push.jobs = &wire_job;
rc = xce_core_push_events(&core, events, 2);
TEST_ASSERT_EQUAL_INT(0, rc);
job = sched_get_job(&core, 501);
TEST_ASSERT_NOT_NULL(job);
/* Directly promote the job to RUNNING with a real, multi-host
* allocation -- a direct field poke, same "genuine, but directly-
* constructed" convention the cross-core test above already uses for
* project priority (there is no wire event to dispatch a job outside a
* real scheduling session in this test harness). */
a0 = make_alloc((char*)"slots", NULL, 0, 2.0f, 0);
a1 = make_alloc((char*)"slots", NULL, 0, 3.0f, 0);
TEST_ASSERT_NOT_NULL(a0);
TEST_ASSERT_NOT_NULL(a1);
h0 = make_host_alloc((char*)"hostA", 1, make_allocs(1, a0));
h1 = make_host_alloc((char*)"hostB", 1, make_allocs(1, a1));
TEST_ASSERT_NOT_NULL(h0);
TEST_ASSERT_NOT_NULL(h1);
hallocs = make_host_allocs(2, h0, h1);
TEST_ASSERT_NOT_NULL(hallocs);
jalloc = make_job_alloc(501, 2, hallocs);
TEST_ASSERT_NOT_NULL(jalloc);
job->state = JOB_STAT_RUN;
job->allocs = jalloc;
memset(&snap, 0, sizeof(snap));
rc = xce_sched_core_save_snapshot(&core, &snap);
TEST_ASSERT_EQUAL_INT(0, rc);
TEST_ASSERT_NOT_NULL(snap.jobs);
TEST_ASSERT_EQUAL_INT(1, snap.jobs->count);
saved_job = snap.jobs->jobs[0];
TEST_ASSERT_NOT_NULL(saved_job);
TEST_ASSERT_EQUAL_UINT64(501, saved_job->jobid);
TEST_ASSERT_EQUAL_UINT32(JOB_STAT_RUN, saved_job->state);
TEST_ASSERT_NOT_NULL(saved_job->allocs);
/* independent copy -- never aliases the live job's own allocation
* (Req 1.4's "no aliasing across independently-alive cores"; this is
* the save-side half of that guarantee). */
TEST_ASSERT_TRUE(saved_job->allocs != job->allocs);
TEST_ASSERT_EQUAL_UINT64(501, saved_job->allocs->jobid);
TEST_ASSERT_EQUAL_INT(2, saved_job->allocs->nhosts);
TEST_ASSERT_NOT_NULL(saved_job->allocs->allocs);
TEST_ASSERT_NOT_NULL(saved_job->allocs->allocs[0]);
TEST_ASSERT_NOT_NULL(saved_job->allocs->allocs[1]);
TEST_ASSERT_EQUAL_STRING("hostA", saved_job->allocs->allocs[0]->host);
TEST_ASSERT_EQUAL_STRING("hostB", saved_job->allocs->allocs[1]->host);
TEST_ASSERT_EQUAL_INT(1, saved_job->allocs->allocs[0]->nallocs);
TEST_ASSERT_EQUAL_INT(1, saved_job->allocs->allocs[1]->nallocs);
TEST_ASSERT_EQUAL_STRING("slots",
saved_job->allocs->allocs[0]->allocs[0]->resname);
TEST_ASSERT_EQUAL_FLOAT(
2.0f, (float)saved_job->allocs->allocs[0]->allocs[0]->value);
TEST_ASSERT_EQUAL_STRING("slots",
saved_job->allocs->allocs[1]->allocs[0]->resname);
TEST_ASSERT_EQUAL_FLOAT(
3.0f, (float)saved_job->allocs->allocs[1]->allocs[0]->value);
}
/* Companion test (Req 3.5's "absent allocation is the normal case, not an
* error" -- save-side half): a PENDING job with no allocation at all must
* still save cleanly, with the wire job's allocs left at its calloc()-
* zeroed NULL default -- no crash, no WARN. */
static void
test_snapshot_save_leaves_allocs_null_for_job_without_allocation(void)
{
xce_sched_core_t core;
sched_snapshot_data_t snap;
xce_core_event_t events[2];
xce_queue_t wire_queue;
xce_job_t* wire_job;
xce_job_t* saved_job;
int rc;
memset(&g_stub_state_a, 0, sizeof(g_stub_state_a));
fixture_init(&core, &g_stub_plugin_a);
memset(&wire_queue, 0, sizeof(wire_queue));
wire_queue.queue = "normal";
wire_queue.priority = 1;
wire_queue.status = QUEUE_STAT_OPEN;
wire_job = make_job(502, JOB_STAT_PEND, "normal", "alice", "proj1");
memset(events, 0, sizeof(events));
events[0].type = XCE_CORE_EVENT_QUEUE_UPDATE;
events[0].u.queue_update.queue = &wire_queue;
events[1].type = XCE_CORE_EVENT_JOBS_PUSH;
events[1].u.jobs_push.njobs = 1;
events[1].u.jobs_push.jobs = &wire_job;
rc = xce_core_push_events(&core, events, 2);
TEST_ASSERT_EQUAL_INT(0, rc);
memset(&snap, 0, sizeof(snap));
rc = xce_sched_core_save_snapshot(&core, &snap);
TEST_ASSERT_EQUAL_INT(0, rc);
TEST_ASSERT_NOT_NULL(snap.jobs);
TEST_ASSERT_EQUAL_INT(1, snap.jobs->count);
saved_job = snap.jobs->jobs[0];
TEST_ASSERT_NOT_NULL(saved_job);
TEST_ASSERT_EQUAL_UINT64(502, saved_job->jobid);
TEST_ASSERT_NULL(saved_job->allocs);
}
/* ---------------------------------------------------------------------- */
/* scheduler-snapshot-job-allocation task 2.2: xce_sched_core_load_snapshot()
* (load side) must restore job->allocs via a genuine deep copy
* (dup_job_alloc()) whenever the wire job carries allocation data (Req 1.1-
* 1.4), and rebuild job->accounts from that restored allocation via the
* existing make_job_account() constructor -- so worker/shared resource "used"
* accounting is correct immediately on load, with no further replay
* activity required (Req 1.2, 1.3). Drives the REAL, full cross-core round
* trip -- save core_a's real state, load into a completely independent
* core_b -- the same "genuinely cross-core, not a field-level shortcut"
* convention this file's header comment establishes for the top-level
* save/load equivalence test above, since Req 1.4's independence guarantee
* can only be genuinely proven across two separately-alive cores.
* ---------------------------------------------------------------------- */
static void
test_snapshot_load_restores_allocs_and_accounts_for_running_job_across_cores(
void)
{
xce_sched_core_t core_a;
xce_sched_core_t core_b;
sched_snapshot_data_t snap;
xce_core_event_t events[2];
xce_queue_t wire_queue;
xce_job_t* wire_job;
xce_sched_job_t* job_a;
xce_alloc_t* a0;
xce_alloc_t* a1;
xce_alloc_t* a2;
xce_host_alloc_t* h0;
xce_host_alloc_t* h1;
xce_host_alloc_t** hallocs;
xce_job_alloc_t* jalloc;
xce_job_t* saved_job;
xce_sched_job_t* restored;
int rc;
memset(&g_stub_state_a, 0, sizeof(g_stub_state_a));
memset(&g_stub_state_b, 0, sizeof(g_stub_state_b));
/* ===================================================================
* Phase 1: build core_a with a RUNNING job carrying a multi-host,
* multi-resource allocation -- hostA gets TWO resources (slots, mem),
* hostB gets ONE (slots), so both "multi-host" and "multi-resource"
* are genuinely exercised, not just one alloc per host.
* =================================================================== */
fixture_init(&core_a, &g_stub_plugin_a);
memset(&wire_queue, 0, sizeof(wire_queue));
wire_queue.queue = "normal";
wire_queue.priority = 1;
wire_queue.status = QUEUE_STAT_OPEN;
wire_job = make_job(601, JOB_STAT_PEND, "normal", "alice", "proj1");
memset(events, 0, sizeof(events));
events[0].type = XCE_CORE_EVENT_QUEUE_UPDATE;
events[0].u.queue_update.queue = &wire_queue;
events[1].type = XCE_CORE_EVENT_JOBS_PUSH;
events[1].u.jobs_push.njobs = 1;
events[1].u.jobs_push.jobs = &wire_job;
rc = xce_core_push_events(&core_a, events, 2);
TEST_ASSERT_EQUAL_INT(0, rc);
job_a = sched_get_job(&core_a, 601);
TEST_ASSERT_NOT_NULL(job_a);
a0 = make_alloc((char*)"slots", NULL, 0, 2.0f, 0);
a1 = make_alloc((char*)"mem", NULL, 0, 100.0f, 0);
a2 = make_alloc((char*)"slots", NULL, 0, 3.0f, 0);
TEST_ASSERT_NOT_NULL(a0);
TEST_ASSERT_NOT_NULL(a1);
TEST_ASSERT_NOT_NULL(a2);
h0 = make_host_alloc((char*)"hostA", 2, make_allocs(2, a0, a1));
h1 = make_host_alloc((char*)"hostB", 1, make_allocs(1, a2));
TEST_ASSERT_NOT_NULL(h0);
TEST_ASSERT_NOT_NULL(h1);
hallocs = make_host_allocs(2, h0, h1);
TEST_ASSERT_NOT_NULL(hallocs);
jalloc = make_job_alloc(601, 2, hallocs);
TEST_ASSERT_NOT_NULL(jalloc);
job_a->state = JOB_STAT_RUN;
job_a->allocs = jalloc;
/* ===================================================================
* Phase 2: save core_a's snapshot.
* =================================================================== */
memset(&snap, 0, sizeof(snap));
rc = xce_sched_core_save_snapshot(&core_a, &snap);
TEST_ASSERT_EQUAL_INT(0, rc);
TEST_ASSERT_NOT_NULL(snap.jobs);
TEST_ASSERT_EQUAL_INT(1, snap.jobs->count);
saved_job = snap.jobs->jobs[0];
TEST_ASSERT_NOT_NULL(saved_job);
TEST_ASSERT_NOT_NULL(saved_job->allocs);
/* ===================================================================
* Phase 3: load into core_b -- a completely separate, independently-
* alive instance.
* =================================================================== */
fixture_init(&core_b, &g_stub_plugin_b);
rc = xce_sched_core_load_snapshot(&core_b, &snap);
TEST_ASSERT_EQUAL_INT(0, rc);
restored = sched_get_job(&core_b, 601);
TEST_ASSERT_NOT_NULL(restored);
TEST_ASSERT_EQUAL_UINT32(JOB_STAT_RUN, restored->state);
/* ===================================================================
* Phase 4: job->allocs content matches the original, host-for-host,
* resource-for-resource.
* =================================================================== */
TEST_ASSERT_NOT_NULL(restored->allocs);
TEST_ASSERT_EQUAL_UINT64(601, restored->allocs->jobid);
TEST_ASSERT_EQUAL_INT(2, restored->allocs->nhosts);
TEST_ASSERT_NOT_NULL(restored->allocs->allocs[0]);
TEST_ASSERT_NOT_NULL(restored->allocs->allocs[1]);
TEST_ASSERT_EQUAL_STRING("hostA", restored->allocs->allocs[0]->host);
TEST_ASSERT_EQUAL_STRING("hostB", restored->allocs->allocs[1]->host);
TEST_ASSERT_EQUAL_INT(2, restored->allocs->allocs[0]->nallocs);
TEST_ASSERT_EQUAL_INT(1, restored->allocs->allocs[1]->nallocs);
TEST_ASSERT_EQUAL_STRING("slots",
restored->allocs->allocs[0]->allocs[0]->resname);
TEST_ASSERT_EQUAL_FLOAT(
2.0f, (float)restored->allocs->allocs[0]->allocs[0]->value);
TEST_ASSERT_EQUAL_STRING("mem",
restored->allocs->allocs[0]->allocs[1]->resname);
TEST_ASSERT_EQUAL_FLOAT(
100.0f, (float)restored->allocs->allocs[0]->allocs[1]->value);
TEST_ASSERT_EQUAL_STRING("slots",
restored->allocs->allocs[1]->allocs[0]->resname);
TEST_ASSERT_EQUAL_FLOAT(
3.0f, (float)restored->allocs->allocs[1]->allocs[0]->value);
/* ===================================================================
* Phase 5: job->accounts correctly reconstructed from the restored
* allocation via make_job_account() -- nhosts matches, and per-resource
* allocated/reserved values reflect the allocation (mirrors
* make_res_account()'s own reserved=allocated=alloc->value, used=0.0
* convention -- see xcealloc.c).
* =================================================================== */
TEST_ASSERT_NOT_NULL(restored->accounts);
TEST_ASSERT_EQUAL_INT(2, restored->accounts->nhosts);
TEST_ASSERT_NOT_NULL(restored->accounts->accounts[0]);
TEST_ASSERT_NOT_NULL(restored->accounts->accounts[1]);
TEST_ASSERT_EQUAL_INT(2, restored->accounts->accounts[0]->naccount);
TEST_ASSERT_EQUAL_INT(1, restored->accounts->accounts[1]->naccount);
TEST_ASSERT_EQUAL_FLOAT(
2.0f, (float)restored->accounts->accounts[0]->accounts[0]->allocated);
TEST_ASSERT_EQUAL_FLOAT(
2.0f, (float)restored->accounts->accounts[0]->accounts[0]->reserved);
TEST_ASSERT_EQUAL_FLOAT(
0.0f, (float)restored->accounts->accounts[0]->accounts[0]->used);
TEST_ASSERT_EQUAL_FLOAT(
100.0f, (float)restored->accounts->accounts[0]->accounts[1]->allocated);
TEST_ASSERT_EQUAL_FLOAT(
3.0f, (float)restored->accounts->accounts[1]->accounts[0]->allocated);
/* ===================================================================
* Phase 6: Req 1.4's independence guarantee, proven end-to-end across
* the full save->load round trip -- THREE genuinely distinct pointers:
* core_b's restored copy, core_a's own live copy, and the wire job's
* own (already-independent, task 2.1) copy. Freeing/mutating any one
* must never be able to corrupt either of the others.
* =================================================================== */
TEST_ASSERT_TRUE(restored->allocs != job_a->allocs);
TEST_ASSERT_TRUE(restored->allocs != saved_job->allocs);
TEST_ASSERT_TRUE(job_a->allocs != saved_job->allocs);
}
/* Companion test (Req 3.5's "absent allocation is the normal case, not an
* error" -- load-side half): a wire job with no allocation data at all
* (old snapshot file, or a job that was pending/never dispatched at save
* time) must load cleanly with job->allocs/job->accounts left at their
* calloc()-zeroed NULL default -- no crash, no WARN. */
static void
test_snapshot_load_leaves_allocs_and_accounts_null_for_job_without_allocation(
void)
{
xce_sched_core_t core;
sched_snapshot_data_t in;
xce_jobs_data_t jobs_data;
xce_queue_t wire_queue;
xce_queue_t* queues[1];
xce_job_t* wire_job;
xce_job_t* jobs[1];
xce_sched_job_t* restored;
int rc;
memset(&g_stub_state_a, 0, sizeof(g_stub_state_a));
fixture_init(&core, &g_stub_plugin_a);
memset(&wire_queue, 0, sizeof(wire_queue));
wire_queue.queue = "normal";
wire_queue.priority = 1;
wire_queue.status = QUEUE_STAT_OPEN;
queues[0] = &wire_queue;
/* make_job() never sets ->allocs -- calloc()-zeroed NULL default,
* exactly the "old snapshot file" / "never dispatched" case Req 3.5
* covers. */
wire_job = make_job(602, JOB_STAT_PEND, "normal", "alice", "proj1");
jobs[0] = wire_job;
memset(&jobs_data, 0, sizeof(jobs_data));
jobs_data.count = 1;
jobs_data.jobs = jobs;
memset(&in, 0, sizeof(in));
in.jobs = &jobs_data;
in.queues = queues;
in.n_queues = 1;
rc = xce_sched_core_load_snapshot(&core, &in);
TEST_ASSERT_EQUAL_INT(0, rc);
restored = sched_get_job(&core, 602);
TEST_ASSERT_NOT_NULL(restored);
TEST_ASSERT_NULL(restored->allocs);
TEST_ASSERT_NULL(restored->accounts);
}
/* ---------------------------------------------------------------------- */
/* scheduler-snapshot-job-allocation task 3.1: the full cross-instance */
/* round-trip proof -- everything task 2.2's own single-running-job test */
/* (above) and task 1.2's own no-allocation schedobj tests (above) stop */
/* short of, exercised together in ONE save/load round trip: */
/* (1) a RUNNING job (multi-host, multi-resource, PLUS a genuine */
/* shared-resource-pool component) saved TOGETHER with a PENDING job */
/* (no allocation at all) in the SAME source core, loaded together */
/* into a second, independent core; */
/* (2) job->schedobj non-NULL for BOTH the running and the pending job; */
/* (3) the restored allocation proven, via the REAL */
/* sched_account_jobs()/sched_charge_account()/sched_update_account()*/
/* chain (resource.c, white-box #include'd above), to feed the */
/* restored core's worker resInUse AND shared-pool sharedInUse */
/* totals to EXACTLY what the identical real chain produces on the */
/* SOURCE core before saving -- not merely that job->accounts' own */
/* fields look right (Req 1.2, 1.3's actual claim: "included in */
/* in-use totals immediately" / "reported identically to normal */
/* dispatch"); */
/* (4) a REAL teardown of one core's job list (destroy(), the same RTTI */
/* primitive xce_sched_core_fini() itself uses for its own kvmgrs --*/
/* xce_sched_core_fini() itself is NOT callable against a bare- */
/* xce_sched_core_init()'d fixture: core->pend_updates[0]/[1] are */
/* only dataq_init()'d by scheduler.c's sched_init(), never by */
/* xce_sched_core_init() itself, so dataq_fini() on them crashes -- */
/* see test_scheduler_decision_redelivery.c's own header comment for */
/* this exact, already-documented hazard) leaves the OTHER, */
/* independently-alive core fully intact and usable afterward. */
/* */
/* sched_account_jobs() itself (scheduler.c) is not white-box #include'd */
/* into this lean core-only TU (too heavy a dependency surface); this test */
/* instead provides its own real, verbatim reimplementation of its exact */
/* body -- the same "real reimplementation" technique this file already */
/* uses for sched_resolve_job_picker() above, and the precedent */
/* test_sim_contention_parity.c already establishes for this identical */
/* function ("a no-op stub would defeat the test's own premise"). */
/* ---------------------------------------------------------------------- */
/* Real reimplementation of scheduler.c's sched_account_jobs() -- verbatim
* body copy (see block comment above for why this file does not white-box
* #include scheduler.c itself). Calls the REAL sched_charge_account()
* (resource.c, #include'd above), so this genuinely exercises the same
* accounting chain live dispatch and sim_reconcile_unaccounted_jobs() both
* rely on downstream. */
static int test_sched_account_jobs(xce_sched_core_t* core)
{
iterator_t it;
int rc = 0;
rc = iter_make(&it, core->jobs);
if (rc < 0) {
return rc;
}
for (;;) {
xce_sched_job_t* job = NULL;
rc = iter_next(&it, (void**)&job);
if ((rc < 0) || (rc == RTTI_ITER_NO_DATA)) {
break;
}
if (!JOB_IS_RUN(job->state)) {
continue;
}
if (job->accounts != NULL) {
sched_charge_account(core, job->accounts);
}
}
if (rc == RTTI_ITER_NO_DATA) {
rc = 0;
}
iter_destroy(&it);
return rc;
}
static void
test_snapshot_round_trip_mixed_jobs_shared_pool_resinuse_and_teardown_independence(
void)
{
xce_sched_core_t core_a;
xce_sched_core_t core_b;
sched_snapshot_data_t snap;
xce_core_event_t events[3];
xce_conf4sched_t config;
xce_resource_t res_slots;
xce_resource_t res_mem;
xce_resource_t res_license;
xce_resource_t* resources[3];
xce_queue_t wire_queue;
xce_queue_t* queues[1];
xce_worker_t wire_worker_a;
xce_worker_t wire_worker_b;
xce_worker_t* workers[2];
xce_val_t worker_a_slots_val;
xce_val_t worker_a_mem_val;
xce_val_t worker_b_slots_val;
xce_name_val_t worker_a_slots_nv;
xce_name_val_t worker_a_mem_nv;
xce_name_val_t worker_b_slots_nv;
xce_name_val_t* worker_a_resvals[2];
xce_name_val_t* worker_b_resvals[1];
xce_job_t* job_running;
xce_job_t* job_pending;
xce_job_t* jobs[2];
xce_sched_job_t* job_a_running;
xce_alloc_t* a_slots_hostA;
xce_alloc_t* a_mem_hostA;
xce_alloc_t* a_slots_hostB;
xce_alloc_t* a_license_shared;
xce_host_alloc_t* h0;
xce_host_alloc_t* h1;
xce_host_alloc_t** hallocs;
xce_job_alloc_t* jalloc;
xce_resource_t* slots_res_a;
xce_resource_t* slots_res_b;
xce_resource_t* mem_res_a;
xce_resource_t* mem_res_b;
xce_resource_t* license_res_a;
xce_resource_t* license_res_b;
xce_sched_worker_t* worker_a_a;
xce_sched_worker_t* worker_b_a;
xce_sched_worker_t* worker_a_b;
xce_sched_worker_t* worker_b_b;
xce_sched_job_t* restored_running;
xce_sched_job_t* restored_pending;
float source_hostA_slots = -1.0f;
float source_hostA_mem = -1.0f;
float source_hostB_slots = -1.0f;
float source_shared_lic = -1.0f;
float restored_hostA_slots = -1.0f;
float restored_hostA_mem = -1.0f;
float restored_hostB_slots = -1.0f;
float restored_shared_lic = -1.0f;
int rc;
memset(&g_stub_state_a, 0, sizeof(g_stub_state_a));
memset(&g_stub_state_b, 0, sizeof(g_stub_state_b));
/* ===================================================================
* Phase 1: build core_a -- register resources/queue/workers, then push
* ONE running job (destined for a multi-host, multi-resource,
* shared-pool allocation) TOGETHER with ONE pending job (no allocation
* at all), in the SAME JOBS_PUSH batch, into the SAME core -- the
* "mix in the same round trip" task 2.2's own single-job test never
* exercised.
* =================================================================== */
fixture_init(&core_a, &g_stub_plugin_a);
memset(&res_slots, 0, sizeof(res_slots));
res_slots.name = "slots";
memset(&res_mem, 0, sizeof(res_mem));
res_mem.name = "mem";
memset(&res_license, 0, sizeof(res_license));
res_license.name = "license";
resources[0] = &res_slots;
resources[1] = &res_mem;
resources[2] = &res_license;
memset(&wire_queue, 0, sizeof(wire_queue));
wire_queue.queue = "normal";
wire_queue.priority = 1;
wire_queue.status = QUEUE_STAT_OPEN;
queues[0] = &wire_queue;
memset(&config, 0, sizeof(config));
config.n_resources = 3;
config.resources = resources;
config.n_queues = 1;
config.queues = queues;
/* .which MUST be XCE_VAL_NUMERIC (bug-1497) -- dup_val()/get_resvals()
* both branch on it, and XCE_VAL_STRING is literally 0 (xce/types.h). */
memset(&worker_a_slots_val, 0, sizeof(worker_a_slots_val));
worker_a_slots_val.which = XCE_VAL_NUMERIC;
worker_a_slots_val.u.ivalue32 = 16;
memset(&worker_a_slots_nv, 0, sizeof(worker_a_slots_nv));
worker_a_slots_nv.name = "slots";
worker_a_slots_nv.val = &worker_a_slots_val;
memset(&worker_a_mem_val, 0, sizeof(worker_a_mem_val));
worker_a_mem_val.which = XCE_VAL_NUMERIC;
worker_a_mem_val.u.ivalue32 = 1000;
memset(&worker_a_mem_nv, 0, sizeof(worker_a_mem_nv));
worker_a_mem_nv.name = "mem";
worker_a_mem_nv.val = &worker_a_mem_val;
worker_a_resvals[0] = &worker_a_slots_nv;
worker_a_resvals[1] = &worker_a_mem_nv;
memset(&worker_b_slots_val, 0, sizeof(worker_b_slots_val));
worker_b_slots_val.which = XCE_VAL_NUMERIC;
worker_b_slots_val.u.ivalue32 = 16;
memset(&worker_b_slots_nv, 0, sizeof(worker_b_slots_nv));
worker_b_slots_nv.name = "slots";
worker_b_slots_nv.val = &worker_b_slots_val;
worker_b_resvals[0] = &worker_b_slots_nv;
memset(&wire_worker_a, 0, sizeof(wire_worker_a));
wire_worker_a.name = "hostA";
wire_worker_a.status = 0; /* available */
wire_worker_a.nresvals_to_sched = 2;
wire_worker_a.resvals_to_sched = worker_a_resvals;
memset(&wire_worker_b, 0, sizeof(wire_worker_b));
wire_worker_b.name = "hostB";
wire_worker_b.status = 0;
wire_worker_b.nresvals_to_sched = 1;
wire_worker_b.resvals_to_sched = worker_b_resvals;
workers[0] = &wire_worker_a;
workers[1] = &wire_worker_b;
job_running = make_job(701, JOB_STAT_PEND, "normal", "alice", "proj1");
job_pending = make_job(702, JOB_STAT_PEND, "normal", "bob", "proj1");
jobs[0] = job_running;
jobs[1] = job_pending;
memset(events, 0, sizeof(events));
/* CONFIG_PUSH first (registers slots/mem/license into core_a->resmgr,
* which WORKERS_PUSH needs already present to resolve resvals_to_sched
* by name), then WORKERS_PUSH, then JOBS_PUSH -- same ordering the
* cross-core equivalence test above already establishes. */
events[0].type = XCE_CORE_EVENT_CONFIG_PUSH;
events[0].u.config_push.config = &config;
events[1].type = XCE_CORE_EVENT_WORKERS_PUSH;
events[1].u.workers_push.nworkers = 2;
events[1].u.workers_push.workers = workers;
events[2].type = XCE_CORE_EVENT_JOBS_PUSH;
events[2].u.jobs_push.njobs = 2;
events[2].u.jobs_push.jobs = jobs;
rc = xce_core_push_events(&core_a, events, 3);
TEST_ASSERT_EQUAL_INT(0, rc);
TEST_ASSERT_EQUAL_INT(2, g_stub_state_a.ncandidates);
/* ===================================================================
* Phase 2: promote job 701 to RUNNING with a real multi-host (hostA:
* slots+mem, hostB: slots), multi-resource allocation PLUS a genuine
* shared-resource-pool component (nres_shared/allocs_shared, "license")
* -- the direct-field-poke convention task 2.2's own test above already
* uses (no wire event exists to dispatch a job outside a real
* scheduling session in this test harness). Job 702 is left PENDING
* with no allocation at all.
* =================================================================== */
job_a_running = sched_get_job(&core_a, 701);
TEST_ASSERT_NOT_NULL(job_a_running);
a_slots_hostA = make_alloc((char*)"slots", NULL, 0, 2.0f, 0);
a_mem_hostA = make_alloc((char*)"mem", NULL, 0, 100.0f, 0);
a_slots_hostB = make_alloc((char*)"slots", NULL, 0, 3.0f, 0);
TEST_ASSERT_NOT_NULL(a_slots_hostA);
TEST_ASSERT_NOT_NULL(a_mem_hostA);
TEST_ASSERT_NOT_NULL(a_slots_hostB);
h0 = make_host_alloc((char*)"hostA", 2,
make_allocs(2, a_slots_hostA, a_mem_hostA));
h1 = make_host_alloc((char*)"hostB", 1, make_allocs(1, a_slots_hostB));
TEST_ASSERT_NOT_NULL(h0);
TEST_ASSERT_NOT_NULL(h1);
hallocs = make_host_allocs(2, h0, h1);
TEST_ASSERT_NOT_NULL(hallocs);
jalloc = make_job_alloc(701, 2, hallocs);
TEST_ASSERT_NOT_NULL(jalloc);
a_license_shared = make_alloc((char*)"license", NULL, 0, 1.0f, 0);
TEST_ASSERT_NOT_NULL(a_license_shared);
jalloc->nres_shared = 1;
jalloc->allocs_shared = make_allocs(1, a_license_shared);
TEST_ASSERT_NOT_NULL(jalloc->allocs_shared);
job_a_running->state = JOB_STAT_RUN;
job_a_running->allocs = jalloc;
/* make_job_account() is the SAME constructor the real dispatch path
* uses -- calling it directly here, on the SOURCE core, puts core_a in
* exactly the accounted state a normal live dispatch would have left
* it in at the moment of saving, which is what makes the resInUse
* comparison in Phase 3/8 below a meaningful "reported identically to
* normal dispatch" proof (Req 1.3), not a trivial self-comparison. */
job_a_running->accounts = make_job_account(job_a_running->allocs);
TEST_ASSERT_NOT_NULL(job_a_running->accounts);
/* ===================================================================
* Phase 3: the actual proof point (Req 1.2/1.3) -- call the REAL
* sched_account_jobs() chain (test_sched_account_jobs() above, which
* calls the REAL sched_charge_account(), resource.c) on core_a BEFORE
* saving, capturing what "resInUse/sharedInUse, reported the normal
* way" looks like for the source instance.
* =================================================================== */
rc = test_sched_account_jobs(&core_a);
TEST_ASSERT_EQUAL_INT(0, rc);
worker_a_a = sched_get_worker(&core_a, "hostA");
worker_b_a = sched_get_worker(&core_a, "hostB");
TEST_ASSERT_NOT_NULL(worker_a_a);
TEST_ASSERT_NOT_NULL(worker_b_a);
rc = key_find(core_a.resmgr, (void*)"slots", (void**)&slots_res_a);
TEST_ASSERT_GREATER_THAN_INT(0, rc);
rc = key_find(core_a.resmgr, (void*)"mem", (void**)&mem_res_a);
TEST_ASSERT_GREATER_THAN_INT(0, rc);
rc = key_find(core_a.resmgr, (void*)"license", (void**)&license_res_a);
TEST_ASSERT_GREATER_THAN_INT(0, rc);
TEST_ASSERT_NOT_NULL(slots_res_a);
TEST_ASSERT_NOT_NULL(mem_res_a);
TEST_ASSERT_NOT_NULL(license_res_a);
get_resvals(&(worker_a_a->resInUse), 1, &(slots_res_a->idx),
&source_hostA_slots);
get_resvals(&(worker_a_a->resInUse), 1, &(mem_res_a->idx),
&source_hostA_mem);
get_resvals(&(worker_b_a->resInUse), 1, &(slots_res_a->idx),
&source_hostB_slots);
get_resvals(&(core_a.sharedInUse), 1, &(license_res_a->idx),
&source_shared_lic);
TEST_ASSERT_EQUAL_FLOAT(2.0f, source_hostA_slots);
TEST_ASSERT_EQUAL_FLOAT(100.0f, source_hostA_mem);
TEST_ASSERT_EQUAL_FLOAT(3.0f, source_hostB_slots);
TEST_ASSERT_EQUAL_FLOAT(1.0f, source_shared_lic);
/* ===================================================================
* Phase 4: save core_a's snapshot -- both jobs, mixed running/pending,
* saved together in the same batch.
* =================================================================== */
memset(&snap, 0, sizeof(snap));
rc = xce_sched_core_save_snapshot(&core_a, &snap);
TEST_ASSERT_EQUAL_INT(0, rc);
TEST_ASSERT_NOT_NULL(snap.jobs);
TEST_ASSERT_EQUAL_INT(2, snap.jobs->count);
/* ===================================================================
* Phase 5: load into core_b -- a completely separate, independently-
* alive instance.
* =================================================================== */
fixture_init(&core_b, &g_stub_plugin_b);
rc = xce_sched_core_load_snapshot(&core_b, &snap);
TEST_ASSERT_EQUAL_INT(0, rc);
TEST_ASSERT_EQUAL_INT(2, g_stub_state_b.ncandidates);
restored_running = sched_get_job(&core_b, 701);
restored_pending = sched_get_job(&core_b, 702);
TEST_ASSERT_NOT_NULL(restored_running);
TEST_ASSERT_NOT_NULL(restored_pending);
TEST_ASSERT_EQUAL_UINT32(JOB_STAT_RUN, restored_running->state);
TEST_ASSERT_EQUAL_UINT32(JOB_STAT_PEND, restored_pending->state);
/* ===================================================================
* Phase 6: EVERY restored job -- running AND pending -- gets a valid,
* non-NULL schedobj (Req 2.1-2.4) -- the "every job" claim task 1.2's
* own tests never exercised alongside a real allocation in the same
* core.
* =================================================================== */
TEST_ASSERT_NOT_NULL(restored_running->schedobj);
TEST_ASSERT_EQUAL_INT(XCE_RESREQ_PRIMARY, restored_running->schedobj->tag);
TEST_ASSERT_NOT_NULL(restored_running->schedobj->u.primary);
TEST_ASSERT_NOT_NULL(restored_pending->schedobj);
TEST_ASSERT_EQUAL_INT(XCE_RESREQ_PRIMARY, restored_pending->schedobj->tag);
TEST_ASSERT_NOT_NULL(restored_pending->schedobj->u.primary);
/* ===================================================================
* Phase 7: allocation/accounting content matches, host-for-host,
* resource-for-resource, INCLUDING the shared-pool component. Job 702
* (pending) restored with allocs/accounts left NULL -- the normal,
* non-error case (Req 3.5).
* =================================================================== */
TEST_ASSERT_NOT_NULL(restored_running->allocs);
TEST_ASSERT_EQUAL_UINT64(701, restored_running->allocs->jobid);
TEST_ASSERT_EQUAL_INT(2, restored_running->allocs->nhosts);
TEST_ASSERT_EQUAL_STRING("hostA", restored_running->allocs->allocs[0]->host);
TEST_ASSERT_EQUAL_STRING("hostB", restored_running->allocs->allocs[1]->host);
TEST_ASSERT_EQUAL_INT(1, restored_running->allocs->nres_shared);
TEST_ASSERT_NOT_NULL(restored_running->allocs->allocs_shared);
TEST_ASSERT_EQUAL_STRING("license",
restored_running->allocs->allocs_shared[0]->resname);
TEST_ASSERT_EQUAL_FLOAT(
1.0f, (float)restored_running->allocs->allocs_shared[0]->value);
TEST_ASSERT_TRUE(restored_running->allocs->allocs_shared[0]
!= a_license_shared);
TEST_ASSERT_NOT_NULL(restored_running->accounts);
TEST_ASSERT_EQUAL_INT(2, restored_running->accounts->nhosts);
TEST_ASSERT_EQUAL_INT(1, restored_running->accounts->nres_shared);
TEST_ASSERT_NOT_NULL(restored_running->accounts->accounts_shared);
TEST_ASSERT_EQUAL_FLOAT(
1.0f, (float)restored_running->accounts->accounts_shared[0]->allocated);
TEST_ASSERT_NULL(restored_pending->allocs);
TEST_ASSERT_NULL(restored_pending->accounts);
/* Req 1.4 independence -- three genuinely distinct pointers throughout
* the chain (source live job, saved wire job, restored job), same
* technique task 2.2's own test above establishes. */
TEST_ASSERT_TRUE(restored_running->allocs != job_a_running->allocs);
TEST_ASSERT_TRUE(restored_running->accounts != job_a_running->accounts);
TEST_ASSERT_TRUE(snap.jobs->jobs[0]->allocs != job_a_running->allocs);
TEST_ASSERT_TRUE(restored_running->allocs != snap.jobs->jobs[0]->allocs);
/* ===================================================================
* Phase 8: the actual proof (Req 1.2/1.3) -- call the SAME real
* accounting chain on core_b (the RESTORED instance) and confirm its
* resInUse/sharedInUse totals EXACTLY match what the identical chain
* produced on core_a (Phase 3) before saving -- proving the restored
* allocation is "included in in-use totals immediately" and "reported
* identically to ... normal dispatch" (Req 1.2, 1.3), not merely that
* job->accounts' own fields happen to look right.
* =================================================================== */
rc = test_sched_account_jobs(&core_b);
TEST_ASSERT_EQUAL_INT(0, rc);
worker_a_b = sched_get_worker(&core_b, "hostA");
worker_b_b = sched_get_worker(&core_b, "hostB");
TEST_ASSERT_NOT_NULL(worker_a_b);
TEST_ASSERT_NOT_NULL(worker_b_b);
rc = key_find(core_b.resmgr, (void*)"slots", (void**)&slots_res_b);
TEST_ASSERT_GREATER_THAN_INT(0, rc);
rc = key_find(core_b.resmgr, (void*)"mem", (void**)&mem_res_b);
TEST_ASSERT_GREATER_THAN_INT(0, rc);
rc = key_find(core_b.resmgr, (void*)"license", (void**)&license_res_b);
TEST_ASSERT_GREATER_THAN_INT(0, rc);
TEST_ASSERT_NOT_NULL(slots_res_b);
TEST_ASSERT_NOT_NULL(mem_res_b);
TEST_ASSERT_NOT_NULL(license_res_b);
get_resvals(&(worker_a_b->resInUse), 1, &(slots_res_b->idx),
&restored_hostA_slots);
get_resvals(&(worker_a_b->resInUse), 1, &(mem_res_b->idx),
&restored_hostA_mem);
get_resvals(&(worker_b_b->resInUse), 1, &(slots_res_b->idx),
&restored_hostB_slots);
get_resvals(&(core_b.sharedInUse), 1, &(license_res_b->idx),
&restored_shared_lic);
TEST_ASSERT_EQUAL_FLOAT(source_hostA_slots, restored_hostA_slots);
TEST_ASSERT_EQUAL_FLOAT(source_hostA_mem, restored_hostA_mem);
TEST_ASSERT_EQUAL_FLOAT(source_hostB_slots, restored_hostB_slots);
TEST_ASSERT_EQUAL_FLOAT(source_shared_lic, restored_shared_lic);
/* spot-pin the concrete expected values too, not just A==B equality. */
TEST_ASSERT_EQUAL_FLOAT(2.0f, restored_hostA_slots);
TEST_ASSERT_EQUAL_FLOAT(100.0f, restored_hostA_mem);
TEST_ASSERT_EQUAL_FLOAT(3.0f, restored_hostB_slots);
TEST_ASSERT_EQUAL_FLOAT(1.0f, restored_shared_lic);
/* ===================================================================
* Phase 9: teardown safety (Req 1.4's "later activity or shutdown in
* either instance does not corrupt or crash the other") -- a REAL
* cleanup path on core_a's job list: destroy(), the same generic RTTI
* teardown primitive xce_sched_core_fini() itself uses for
* resmgr/ugmgr/qgmgr/pgmgr/hgmgr (core.c), applied here directly to
* core_a.jobs. (xce_sched_core_fini() itself is deliberately NOT called
* here -- core_a was built via bare xce_sched_core_init(), and
* core->pend_updates[0]/[1] are only dataq_init()'d by scheduler.c's
* sched_init(), never by xce_sched_core_init() itself, so
* xce_sched_core_fini()'s own dataq_fini() calls would dereference
* still-NULL pointers and crash -- test_scheduler_decision_redelivery.c
* already documents this exact hazard for the identical minimal-fixture
* technique this file also uses.) core_b must remain fully intact and
* usable afterward -- not merely non-crashing: every value re-read
* below must still match Phase 8's, proving core_b's lifetime is
* genuinely independent of core_a's, not merely pointer-distinct.
* =================================================================== */
destroy(core_a.jobs);
TEST_ASSERT_NOT_NULL(sched_get_job(&core_b, 701));
TEST_ASSERT_NOT_NULL(sched_get_job(&core_b, 702));
TEST_ASSERT_EQUAL_UINT32(JOB_STAT_RUN, sched_get_job(&core_b, 701)->state);
TEST_ASSERT_EQUAL_UINT32(JOB_STAT_PEND,
sched_get_job(&core_b, 702)->state);
TEST_ASSERT_NOT_NULL(sched_get_job(&core_b, 701)->schedobj);
TEST_ASSERT_NOT_NULL(sched_get_job(&core_b, 702)->schedobj);
worker_a_b = sched_get_worker(&core_b, "hostA");
worker_b_b = sched_get_worker(&core_b, "hostB");
TEST_ASSERT_NOT_NULL(worker_a_b);
TEST_ASSERT_NOT_NULL(worker_b_b);
get_resvals(&(worker_a_b->resInUse), 1, &(slots_res_b->idx),
&restored_hostA_slots);
get_resvals(&(worker_a_b->resInUse), 1, &(mem_res_b->idx),
&restored_hostA_mem);
get_resvals(&(worker_b_b->resInUse), 1, &(slots_res_b->idx),
&restored_hostB_slots);
get_resvals(&(core_b.sharedInUse), 1, &(license_res_b->idx),
&restored_shared_lic);
TEST_ASSERT_EQUAL_FLOAT(2.0f, restored_hostA_slots);
TEST_ASSERT_EQUAL_FLOAT(100.0f, restored_hostA_mem);
TEST_ASSERT_EQUAL_FLOAT(3.0f, restored_hostB_slots);
TEST_ASSERT_EQUAL_FLOAT(1.0f, restored_shared_lic);
}
int main(void)
{
UNITY_BEGIN();
RUN_TEST(
test_snapshot_guards_reject_null_args_and_busy_session_without_mutating_state);
RUN_TEST(
test_snapshot_save_and_load_reject_during_genuinely_active_session);
RUN_TEST(
test_snapshot_save_then_load_produces_behaviorally_equivalent_core);
RUN_TEST(
test_snapshot_load_builds_schedobj_for_primary_compound_and_alter_resreq);
RUN_TEST(
test_snapshot_load_leaves_schedobj_null_when_resreq_fails_to_parse);
RUN_TEST(
test_snapshot_save_populates_allocs_for_running_job_with_multi_host_allocation);
RUN_TEST(test_snapshot_save_leaves_allocs_null_for_job_without_allocation);
RUN_TEST(
test_snapshot_load_restores_allocs_and_accounts_for_running_job_across_cores);
RUN_TEST(
test_snapshot_load_leaves_allocs_and_accounts_null_for_job_without_allocation);
RUN_TEST(
test_snapshot_round_trip_mixed_jobs_shared_pool_resinuse_and_teardown_independence);
return UNITY_END();
}
Reference in a new issue View git blame Copy permalink