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

feat(scheduler-snapshot-job-allocation): populate allocs on snapshot save

Browse source 
Task 2.1: build_wire_job() now deep-copies job->allocs into the wire
job's already-existing allocs field via dup_job_alloc() (task 1.1) when
the live job has an allocation, leaving it unset otherwise -- matching
how every other optional field in this function is handled. Never
aliases job->allocs.

New tests prove: a running job with a multi-host allocation saves with
matching content (host names, resource values) and a pointer distinct
from the live job's own allocation; a job with no allocation saves with
the wire field left NULL, no error.

Requirements: 1.1
This commit is contained in:
Rongsong Shen 2026-08-27 13:44:49 +00:00
parent 22353849f6
commit 6fa2d970f1
No known key found for this signature in database
GPG key ID: 5BCDAAE0123E9FE2
3 changed files with 188 additions and 3 deletions
Download patch file Download diff file Expand all files Collapse all files

2
.kiro/specs/scheduler-snapshot-job-allocation/tasks.md
View file

@ -18,7 +18,7 @@
- _Boundary: snapshot load (scheduling-representation construction)_
- [ ] 2. Core: wire the allocation restoration into the save and load paths
- [ ] 2.1 Populate a job's resource allocation on the save side
- [x] 2.1 Populate a job's resource allocation on the save side
- Write a failing test first (TDD red) asserting that saving a running job with a resource allocation produces a saved representation carrying that allocation.
- Populate the already-existing allocation slot on the saved job representation using the deep-copy primitive from 1.1 whenever the live job has an allocation; leave it absent otherwise, matching how every other optional value here is already handled.
- Observable completion: the new test goes from failing to passing, and saving a job with a multi-host allocation produces a saved job whose allocation content matches the live job's.

22
src/scheduler/src/snapshot_wire.c
View file

@ -5,6 +5,7 @@
#include "core.h"
#include "sched/types.h"
#include "utils.h"
#include "xcealloc.h"
#include "xstring.h"
/*
@ -164,8 +165,25 @@ xce_job_t* build_wire_job(xce_sched_job_t* job)
}
j->submit->resreq = resreq_str;
/* job->allocs/accounts: not captured -- see snapshot.c header "Scope
* decisions". */
/* job->allocs (scheduler-snapshot-job-allocation task 2.1, Req 1.1):
* deep-copied via dup_job_alloc() -- never aliases job->allocs, since
* this wire job may outlive job's own core or be loaded into an
* independently-alive one (Req 1.4). Left at its calloc()-zeroed NULL
* default when the live job has none (never dispatched, or an old-
* format snapshot's own job that predates this feature) -- the same
* optional-field pattern `resreq` above already follows.
*
* job->accounts: not captured here -- it is derived, never itself
* serialized; the load side reconstructs it via make_job_account() from
* the restored allocs (see snapshot.c header "Scope decisions"). */
if (job->allocs != NULL) {
j->allocs = dup_job_alloc(job->allocs);
if (j->allocs == NULL) {
WARN("%s dup_job_alloc failed for jobid %llu; this job's "
"allocation will be missing from the snapshot",
__func__, (unsigned long long)job->jobid);
}
}
return j;
}

167
src/scheduler/testsuite/test_snapshot_save_load.c
View file

@ -62,6 +62,7 @@
#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()/
@ -1175,6 +1176,169 @@ test_snapshot_load_leaves_schedobj_null_when_resreq_fails_to_parse(void)
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);
}
int main(void)
{
UNITY_BEGIN();
@ -1188,5 +1352,8 @@ int main(void)
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);
return UNITY_END();
}