From bbb3972ae60a965db3b36d4811ddccb989def952 Mon Sep 17 00:00:00 2001 From: Rongsong Shen Date: Tue, 4 Aug 2026 13:11:26 +0000 Subject: [PATCH] feat(scheduler-core): define decision payload types and snapshot schema (task 7.3) Adds DispatchDecision (wraps JobAlloc), PendDecision (wraps JobPendReason), PreemptVictim, PreemptDecision to controller.capnp. Replaces task 7.1's empty SnapshotData placeholder wholesale with the real 11-field struct (config/jobs/queues/workers/resources/resvals/ users/projects/prngSeed/internalState/internalVersion), in both the capnp schema and its typemap.h C mirror. Extends ReplayDecision's ts-only placeholder with a purely additive body union (dispatch@1/ pend@2/preempt@3). Deviates from design.md's literal SchedUser/SchedProject mapnames (xce_sched_user_t/xce_sched_project_t): those names are already real C structs in sched/types.h, which shares a translation unit with typemap.h in core_events.c/schedobj.c -- reusing them would be a genuine duplicate-typedef compile error, not a style choice. Used fresh mapnames (sched_snapshot_user_t/sched_snapshot_project_t) instead, field lists unchanged, documented in both files. Found and fixed a real bug mid-implementation: a doc comment's literal */ substring prematurely closed a C block comment, corrupting sched_snapshot_data_t and cascading into ~20 unrelated-looking compile errors (bug-1493). Five deep round-trip tests added to test_e2e_capnp.c. Full project build and controller_unit_test (1082/1082) verified clean, confirmed independently three times. Co-Authored-By: Claude Sonnet 5 Claude-Session: https://claude.ai/code/session_019dw5WnN3DXStaEaPYfACU1 --- .kiro/specs/scheduler-core/tasks.md | 3 +- src/controller/testsuite/test_e2e_capnp.c | 562 ++++++++++++++++++++++ src/include/typemap.h | 88 +++- src/schema/controller.capnp | 135 +++++- 4 files changed, 755 insertions(+), 33 deletions(-) diff --git a/.kiro/specs/scheduler-core/tasks.md b/.kiro/specs/scheduler-core/tasks.md index ea064eb2..26c4ab05 100644 --- a/.kiro/specs/scheduler-core/tasks.md +++ b/.kiro/specs/scheduler-core/tasks.md @@ -173,12 +173,13 @@ - _Boundary: ReplayFormat — not (P): 7.2/7.3/7.4 all edit the same schema file (controller.capnp); run sequentially to avoid concurrent edits to one file, even though each adds a logically disjoint set of structs_ - _Verified: 2026-08-04 — JobStateChangeEvent/JobUpdateEvent added to controller.capnp matching design.md's spec verbatim; JobUpdateEvent.job correctly reuses the full 12-field Job/xce_job_t record (not the narrower SchedJob sharing the same mapname — verified by field list). ReplayEvent.body's union extended with jobStateChange@10/jobUpdate@11, the only two valid next ordinals per task 7.1's baseline (capnp requires no gaps) — confirmed via a from-scratch capnp regeneration. Existing @0-@9 fields byte-identical. Deep round-trip tests (non-default values throughout, including 3-level nested Job substructs) prove the Observable; JOB_F_* bitmask values deliberately left undefined (later task's job). 1077/1077 controller_unit_test, full project build clean, review APPROVED first pass._ -- [ ] 7.3 Define decision payload types and the snapshot schema +- [x] 7.3 Define decision payload types and the snapshot schema - Add the dispatch, pend, and preempt decision payload message types, reusing existing allocation and pend-reason wire messages where they already carry the needed fields - Add the full snapshot message type covering configuration, jobs, queues, workers, resources, resource values, users, and projects, reusing existing wire messages for each domain object list; represent the two scheduler-internal-only caches and session/PRNG state as a versioned opaque payload with no external structure - Observable: encoding one of each decision type and a populated snapshot and decoding them back yields field-for-field equality; a snapshot encoded with one internal-state version and decoded expecting a different version is rejected rather than misinterpreted - _Requirements: 5.5, 5.20, 5.21, 6.5_ - _Depends: 7.1_ + - _Verified: 2026-08-04 — DispatchDecision/PendDecision/PreemptVictim/PreemptDecision added, reusing JobAlloc/JobPendReason per spec. SnapshotData's task-7.1 empty placeholder replaced wholesale with the real 11 fields (config/jobs/queues/workers/resources/resvals/users/projects/prngSeed/internalState/internalVersion), both in the capnp schema and its typemap.h C mirror. ReplayDecision's ts-only placeholder gained a purely additive body union (dispatch@1/pend@2/preempt@3). One deviation from design.md's literal text: SchedUser/SchedProject use fresh mapnames (sched_snapshot_user_t/sched_snapshot_project_t) instead of design.md's literal xce_sched_user_t/xce_sched_project_t, since those names are already real C structs in sched/types.h that share a translation unit with typemap.h in core_events.c/schedobj.c — reusing them would be a genuine duplicate-typedef compile error, not a style choice. Field lists (name-only; name+priority) verified unchanged, deviation rigorously re-derived and confirmed necessary by independent review, documented in both files. Found and fixed a real bug mid-implementation (a doc comment's literal */ substring prematurely closing a C block comment) logged as bug-1493. ReplayEvent's union (tasks 7.1/7.2) and every other existing struct byte-identical. Five deep round-trip tests (all 11 SnapshotData fields populated, version-mismatch demonstrated). 1082/1082 controller_unit_test, full clean project build, confirmed independently three times (implementer, orchestrator, reviewer, all matching exactly), review APPROVED first pass._ - _Boundary: ReplayFormat — not (P), same file as 7.2/7.4 (see 7.2)_ - [ ] 7.4 Add the scheduler admin opcode schema diff --git a/src/controller/testsuite/test_e2e_capnp.c b/src/controller/testsuite/test_e2e_capnp.c index 82dac9bd..2b05641f 100644 --- a/src/controller/testsuite/test_e2e_capnp.c +++ b/src/controller/testsuite/test_e2e_capnp.c @@ -787,6 +787,563 @@ static void test_job_update_event_round_trip(void) printf("test_job_update_event_round_trip: PASS\n"); } +/* ------------------------------------------------------------------ */ +/* Test 11: DispatchDecision round-trip (scheduler-core task 7.3, Req 5 */ +/* AC 5, AC 20) -- reuses JobAlloc verbatim (jobid lives at */ +/* JobAlloc.jobid, no redundant top-level jobid field, per design */ +/* review). Populates a real multi-level allocation tree (one host, one */ +/* resource) via the JobAlloc/HostAlloc/ResAlloc codec chain already */ +/* proven in Test 10. */ +/* ------------------------------------------------------------------ */ + +static void test_dispatch_decision_round_trip(void) +{ + xce_alloc_t res_alloc; + memset(&res_alloc, 0, sizeof(res_alloc)); + res_alloc.resname = "gpu"; + res_alloc.reserve = "resv2"; + res_alloc.instid = 7ULL; + res_alloc.value = 2.0; + res_alloc.from = 100ULL; + res_alloc.duration = 1800LL; + xce_alloc_t *res_allocs[1] = { &res_alloc }; + + xce_host_alloc_t host_alloc; + memset(&host_alloc, 0, sizeof(host_alloc)); + host_alloc.host = "hostB"; + host_alloc.nallocs = 1; + host_alloc.allocs = res_allocs; + xce_host_alloc_t *host_allocs[1] = { &host_alloc }; + + xce_job_alloc_t job_alloc; + memset(&job_alloc, 0, sizeof(job_alloc)); + job_alloc.jobid = 9001ULL; + job_alloc.nhosts = 1; + job_alloc.allocs = host_allocs; + job_alloc.nres_shared = 0; + job_alloc.allocs_shared = NULL; + + sched_dispatch_decision_t src; + memset(&src, 0, sizeof(src)); + src.alloc = &job_alloc; + + struct capn c; + capn_init_malloc(&c); + struct capn_segment *cs = capn_root(&c).seg; + + DispatchDecision_ptr p; + encode_DispatchDecision_ptr(cs, &p, &src); + capn_setp(capn_root(&c), 0, p.p); + + int64_t sz = capn_size(&c); + TEST_ASSERT_TRUE(sz > 0); + uint8_t *buf = (uint8_t *)malloc((size_t)sz); + TEST_ASSERT_NOT_NULL(buf); + capn_write_mem(&c, buf, (size_t)sz, 0); + capn_free(&c); + + struct capn d; + TEST_ASSERT_EQUAL_INT(0, capn_init_mem(&d, buf, (size_t)sz, 0)); + DispatchDecision_ptr dp; + dp.p = capn_getp(capn_root(&d), 0, 1); + struct DispatchDecision wire; + read_DispatchDecision(&wire, dp); + sched_dispatch_decision_t dst; + memset(&dst, 0, sizeof(dst)); + decode_DispatchDecision(&dst, &wire); + + TEST_ASSERT_NOT_NULL(dst.alloc); + TEST_ASSERT_EQUAL_UINT64(9001ULL, dst.alloc->jobid); + TEST_ASSERT_EQUAL_INT(1, dst.alloc->nhosts); + TEST_ASSERT_NOT_NULL(dst.alloc->allocs); + TEST_ASSERT_EQUAL_STRING("hostB", dst.alloc->allocs[0]->host); + TEST_ASSERT_EQUAL_INT(1, dst.alloc->allocs[0]->nallocs); + TEST_ASSERT_NOT_NULL(dst.alloc->allocs[0]->allocs); + TEST_ASSERT_EQUAL_STRING("gpu", dst.alloc->allocs[0]->allocs[0]->resname); + TEST_ASSERT_EQUAL_STRING("resv2", dst.alloc->allocs[0]->allocs[0]->reserve); + TEST_ASSERT_EQUAL_UINT64(7ULL, dst.alloc->allocs[0]->allocs[0]->instid); + TEST_ASSERT_EQUAL_FLOAT(2.0f, (float)dst.alloc->allocs[0]->allocs[0]->value); + TEST_ASSERT_EQUAL_UINT64(100ULL, dst.alloc->allocs[0]->allocs[0]->from); + TEST_ASSERT_EQUAL_INT64(1800LL, dst.alloc->allocs[0]->allocs[0]->duration); + TEST_ASSERT_EQUAL_INT(0, dst.alloc->nres_shared); + TEST_ASSERT_NULL(dst.alloc->allocs_shared); + + free_DispatchDecision(&dst); + capn_free(&d); + free(buf); + printf("test_dispatch_decision_round_trip: PASS\n"); +} + +/* ------------------------------------------------------------------ */ +/* Test 12: PendDecision round-trip (scheduler-core task 7.3, Req 5 */ +/* AC 5, AC 20) -- reuses JobPendReason (xce_job_preason_t) verbatim. */ +/* ------------------------------------------------------------------ */ + +static void test_pend_decision_round_trip(void) +{ + xce_job_preason_t reason; + memset(&reason, 0, sizeof(reason)); + reason.main_reason = 3; + reason.nhost_reasons = 0; + reason.host_based_reasons = NULL; + reason.nres_reasons = 0; + reason.res_based_reasons = NULL; + + sched_pend_decision_t src; + memset(&src, 0, sizeof(src)); + src.jobid = 4242ULL; + src.reason = &reason; + + struct capn c; + capn_init_malloc(&c); + struct capn_segment *cs = capn_root(&c).seg; + + PendDecision_ptr p; + encode_PendDecision_ptr(cs, &p, &src); + capn_setp(capn_root(&c), 0, p.p); + + int64_t sz = capn_size(&c); + TEST_ASSERT_TRUE(sz > 0); + uint8_t *buf = (uint8_t *)malloc((size_t)sz); + TEST_ASSERT_NOT_NULL(buf); + capn_write_mem(&c, buf, (size_t)sz, 0); + capn_free(&c); + + struct capn d; + TEST_ASSERT_EQUAL_INT(0, capn_init_mem(&d, buf, (size_t)sz, 0)); + PendDecision_ptr dp; + dp.p = capn_getp(capn_root(&d), 0, 1); + struct PendDecision wire; + read_PendDecision(&wire, dp); + sched_pend_decision_t dst; + memset(&dst, 0, sizeof(dst)); + decode_PendDecision(&dst, &wire); + + TEST_ASSERT_EQUAL_UINT64(4242ULL, dst.jobid); + TEST_ASSERT_NOT_NULL(dst.reason); + TEST_ASSERT_EQUAL_INT(3, dst.reason->main_reason); + TEST_ASSERT_EQUAL_INT(0, dst.reason->nhost_reasons); + TEST_ASSERT_NULL(dst.reason->host_based_reasons); + TEST_ASSERT_EQUAL_INT(0, dst.reason->nres_reasons); + TEST_ASSERT_NULL(dst.reason->res_based_reasons); + + free_PendDecision(&dst); + capn_free(&d); + free(buf); + printf("test_pend_decision_round_trip: PASS\n"); +} + +/* ------------------------------------------------------------------ */ +/* Test 13: PreemptDecision round-trip (scheduler-core task 7.3, Req 5 */ +/* AC 5, AC 20) -- PreemptVictim is fresh (no pre-existing wire */ +/* counterpart) and deliberately distinct from decision.h's */ +/* core-internal xce_preempt_victim_t/xce_preempt_t (task 4.4's */ +/* in-process live-decision representation). Populates a 2-entry victim */ +/* list to exercise List(PreemptVictim) with real data. */ +/* ------------------------------------------------------------------ */ + +static void test_preempt_decision_round_trip(void) +{ + preempt_victim_t v1; + memset(&v1, 0, sizeof(v1)); + v1.jobid = 111ULL; + v1.workerid = "hostX"; + + preempt_victim_t v2; + memset(&v2, 0, sizeof(v2)); + v2.jobid = 222ULL; + v2.workerid = "hostY"; + + preempt_victim_t *victims[2] = { &v1, &v2 }; + + xce_job_preason_t reason; + memset(&reason, 0, sizeof(reason)); + reason.main_reason = 9; + + sched_preempt_decision_t src; + memset(&src, 0, sizeof(src)); + src.jobid = 5555ULL; + src.victims = victims; + src.n_victims = 2; + src.reason = &reason; + + struct capn c; + capn_init_malloc(&c); + struct capn_segment *cs = capn_root(&c).seg; + + PreemptDecision_ptr p; + encode_PreemptDecision_ptr(cs, &p, &src); + capn_setp(capn_root(&c), 0, p.p); + + int64_t sz = capn_size(&c); + TEST_ASSERT_TRUE(sz > 0); + uint8_t *buf = (uint8_t *)malloc((size_t)sz); + TEST_ASSERT_NOT_NULL(buf); + capn_write_mem(&c, buf, (size_t)sz, 0); + capn_free(&c); + + struct capn d; + TEST_ASSERT_EQUAL_INT(0, capn_init_mem(&d, buf, (size_t)sz, 0)); + PreemptDecision_ptr dp; + dp.p = capn_getp(capn_root(&d), 0, 1); + struct PreemptDecision wire; + read_PreemptDecision(&wire, dp); + sched_preempt_decision_t dst; + memset(&dst, 0, sizeof(dst)); + decode_PreemptDecision(&dst, &wire); + + TEST_ASSERT_EQUAL_UINT64(5555ULL, dst.jobid); + TEST_ASSERT_EQUAL_INT(2, dst.n_victims); + TEST_ASSERT_NOT_NULL(dst.victims); + TEST_ASSERT_EQUAL_UINT64(111ULL, dst.victims[0]->jobid); + TEST_ASSERT_EQUAL_STRING("hostX", dst.victims[0]->workerid); + TEST_ASSERT_EQUAL_UINT64(222ULL, dst.victims[1]->jobid); + TEST_ASSERT_EQUAL_STRING("hostY", dst.victims[1]->workerid); + TEST_ASSERT_NOT_NULL(dst.reason); + TEST_ASSERT_EQUAL_INT(9, dst.reason->main_reason); + + free_PreemptDecision(&dst); + capn_free(&d); + free(buf); + printf("test_preempt_decision_round_trip: PASS\n"); +} + +/* ------------------------------------------------------------------ */ +/* Test 14: ReplayDecision's `body` union round-trips through the */ +/* dispatch variant (scheduler-core task 7.3, Req 5 AC 20) -- proves */ +/* the purely-additive union bolted onto task 7.1's ts@0-only stub */ +/* actually wires up end to end (mirrors the JobControl/JoinData */ +/* `:union` group precedent: the group name consumes no ordinal). */ +/* ------------------------------------------------------------------ */ + +static void test_replay_decision_dispatch_union_round_trip(void) +{ + xce_job_alloc_t job_alloc; + memset(&job_alloc, 0, sizeof(job_alloc)); + job_alloc.jobid = 7777ULL; + job_alloc.nhosts = 0; + job_alloc.allocs = NULL; + job_alloc.nres_shared = 0; + job_alloc.allocs_shared = NULL; + + sched_dispatch_decision_t dispatch; + memset(&dispatch, 0, sizeof(dispatch)); + dispatch.alloc = &job_alloc; + + replay_decision_t src; + memset(&src, 0, sizeof(src)); + src.ts = 55555ULL; + src.body_which = ReplayDecision_body_dispatch; + src.body.dispatch = &dispatch; + + struct capn c; + capn_init_malloc(&c); + struct capn_segment *cs = capn_root(&c).seg; + + ReplayDecision_ptr p; + encode_ReplayDecision_ptr(cs, &p, &src); + capn_setp(capn_root(&c), 0, p.p); + + int64_t sz = capn_size(&c); + TEST_ASSERT_TRUE(sz > 0); + uint8_t *buf = (uint8_t *)malloc((size_t)sz); + TEST_ASSERT_NOT_NULL(buf); + capn_write_mem(&c, buf, (size_t)sz, 0); + capn_free(&c); + + struct capn d; + TEST_ASSERT_EQUAL_INT(0, capn_init_mem(&d, buf, (size_t)sz, 0)); + ReplayDecision_ptr dp; + dp.p = capn_getp(capn_root(&d), 0, 1); + struct ReplayDecision wire; + read_ReplayDecision(&wire, dp); + replay_decision_t dst; + memset(&dst, 0, sizeof(dst)); + decode_ReplayDecision(&dst, &wire); + + TEST_ASSERT_EQUAL_UINT64(55555ULL, dst.ts); + TEST_ASSERT_EQUAL_INT((int)ReplayDecision_body_dispatch, (int)dst.body_which); + TEST_ASSERT_NOT_NULL(dst.body.dispatch); + TEST_ASSERT_NOT_NULL(dst.body.dispatch->alloc); + TEST_ASSERT_EQUAL_UINT64(7777ULL, dst.body.dispatch->alloc->jobid); + + free_ReplayDecision(&dst); + capn_free(&d); + free(buf); + printf("test_replay_decision_dispatch_union_round_trip: PASS\n"); +} + +/* ------------------------------------------------------------------ */ +/* Test 15: SnapshotData round-trip (scheduler-core task 7.3, Req 5 */ +/* AC 21, Req 6 AC 5) -- the full snapshot message, replacing task */ +/* 7.1's zero-field placeholder wholesale. Populates all 11 fields with */ +/* non-trivial data (including >=1 entry in each of the top-level */ +/* List(...) fields: queues/users/projects) and round-trips them */ +/* field-for-field through encode_SnapshotData_ptr/decode_SnapshotData */ +/* -- EXCEPT internalState (a Data field), whose generated codec is a */ +/* documented no-op (same limitation as every other Data field in this */ +/* schema, e.g. ReplayHeader.configHash in Test 7) -- verified instead */ +/* via the low-level SnapshotData_get/set_internalState accessor on the */ +/* SAME struct pointer the codec bridge already populated, same hybrid */ +/* technique used nowhere else in this file but safe because encode_/ */ +/* set_ operate on independent field slots of one already-allocated */ +/* capnp struct. */ +/* */ +/* Also demonstrates this task's internalVersion Observable: "a */ +/* snapshot encoded with one internal-state version and decoded */ +/* expecting a different version is rejected rather than */ +/* misinterpreted." internalVersion is a schema-side format TAG only -- */ +/* the real accept/reject decision is production logic for task 8.1+ to */ +/* write; here we prove the tag survives the wire faithfully and that a */ +/* simple equality check (the only sound way to interpret a version */ +/* tag) distinguishes a matching reader from a mismatched one. The */ +/* comparison below is TEST-ONLY logic illustrating the intended usage, */ +/* not production code. */ +/* ------------------------------------------------------------------ */ + +static void test_snapshot_data_round_trip(void) +{ + /* One queue, reused both as SnapshotData's own top-level `queues` + * list and inside `config.queues` (Conf4Sched's own queue list) -- + * proves both call sites of the pre-existing QueueConf codec. */ + xce_queue_t queue; + memset(&queue, 0, sizeof(queue)); + queue.queue = "batch"; + queue.attrs = 1u; + queue.status = 0u; + queue.priority = 5; + queue.accept_interval = 10; + queue.sched_delay = 2; + queue.resreq = "select=1"; + xce_queue_t *queues[1] = { &queue }; + + xce_conf4sched_t config; + memset(&config, 0, sizeof(config)); + config.clustername = "xce-snap-cluster"; + config.n_queues = 1; + config.queues = queues; + + xce_job_t sjob; + memset(&sjob, 0, sizeof(sjob)); + sjob.jobid = 8001ULL; + sjob.sub_time = 1000; + sjob.queue_time = 2000; + sjob.start_time = 3000; + sjob.finish_time = 4000; + sjob.state = 5u; + sjob.run_time = 120; + xce_job_t *sjobs[1] = { &sjob }; + + xce_jobs_data_t jobs_data; + memset(&jobs_data, 0, sizeof(jobs_data)); + jobs_data.n_total = 1u; + jobs_data.pos = 0u; + jobs_data.count = 1; + jobs_data.jobs = sjobs; + + xce_worker_t worker; + memset(&worker, 0, sizeof(worker)); + worker.name = "workerA"; + worker.status = XCE_WORKER_UNAVAIL; + xce_worker_t *workers_arr[1] = { &worker }; + + xce_workers_data_t workers_data; + memset(&workers_data, 0, sizeof(workers_data)); + workers_data.n_total = 1u; + workers_data.pos = 0u; + workers_data.count = 1; + workers_data.workers = workers_arr; + + xce_resource_t resource; + memset(&resource, 0, sizeof(resource)); + resource.name = "cpu-res"; + resource.desc = "CPU resource"; + resource.idx = 0; + resource.valtype = 1; + resource.ordtype = XCE_ORD_INCR; + resource.flags = 0; + resource.interval = 60; + resource.last_report = 999; + xce_resource_t *resources_arr[1] = { &resource }; + + xce_resource_data_t resources_data; + memset(&resources_data, 0, sizeof(resources_data)); + resources_data.n_total = 1u; + resources_data.pos = 0u; + resources_data.count = 1; + resources_data.resources = resources_arr; + + xce_val_t val; + memset(&val, 0, sizeof(val)); + val.which = 1; /* uvalue32 */ + val.u.uvalue32 = 4096u; + + xce_resval_t resval; + memset(&resval, 0, sizeof(resval)); + resval.name = "mem"; + resval.host = "hostA"; + resval.val = &val; + xce_resval_t *resvals_arr[1] = { &resval }; + + xce_resval_list_t resval_list; + memset(&resval_list, 0, sizeof(resval_list)); + resval_list.count = 1; + resval_list.vals = resvals_arr; + + sched_snapshot_user_t user; + memset(&user, 0, sizeof(user)); + user.name = "alice"; + sched_snapshot_user_t *users_arr[1] = { &user }; + + sched_snapshot_project_t project; + memset(&project, 0, sizeof(project)); + project.name = "proj1"; + project.priority = 5; + sched_snapshot_project_t *projects_arr[1] = { &project }; + + uint8_t internal_bytes[8] = { 0xDE, 0xAD, 0xBE, 0xEF, 0x01, 0x02, 0x03, 0x04 }; + + sched_snapshot_data_t src; + memset(&src, 0, sizeof(src)); + src.config = &config; + src.jobs = &jobs_data; + src.queues = queues; + src.n_queues = 1; + src.workers = &workers_data; + src.resources = &resources_data; + src.resvals = &resval_list; + src.users = users_arr; + src.n_users = 1; + src.projects = projects_arr; + src.n_projects = 1; + src.prng_seed = 123456789ULL; + src.internal_state = internal_bytes; + src.internal_state_len = sizeof(internal_bytes); + src.internal_version = 2u; + + struct capn c; + capn_init_malloc(&c); + struct capn_segment *cs = capn_root(&c).seg; + + SnapshotData_ptr p; + encode_SnapshotData_ptr(cs, &p, &src); + + /* internalState (Data field): codec no-op, set via low-level accessor + * on the same already-allocated struct pointer (see comment above). */ + capn_list8 state_list = capn_new_list8(cs, (int)sizeof(internal_bytes)); + capn_setv8(state_list, 0, internal_bytes, sizeof(internal_bytes)); + capn_data state_data; + state_data.p = state_list.p; + SnapshotData_set_internalState(p, state_data); + + capn_setp(capn_root(&c), 0, p.p); + int64_t sz = capn_size(&c); + TEST_ASSERT_TRUE(sz > 0); + uint8_t *buf = (uint8_t *)malloc((size_t)sz); + TEST_ASSERT_NOT_NULL(buf); + capn_write_mem(&c, buf, (size_t)sz, 0); + capn_free(&c); + + struct capn d; + TEST_ASSERT_EQUAL_INT(0, capn_init_mem(&d, buf, (size_t)sz, 0)); + SnapshotData_ptr dp; + dp.p = capn_getp(capn_root(&d), 0, 1); + struct SnapshotData wire; + read_SnapshotData(&wire, dp); + sched_snapshot_data_t dst; + memset(&dst, 0, sizeof(dst)); + decode_SnapshotData(&dst, &wire); + + /* config (Conf4Sched) */ + TEST_ASSERT_NOT_NULL(dst.config); + TEST_ASSERT_EQUAL_STRING("xce-snap-cluster", dst.config->clustername); + TEST_ASSERT_EQUAL_INT(1, dst.config->n_queues); + TEST_ASSERT_NOT_NULL(dst.config->queues); + TEST_ASSERT_EQUAL_STRING("batch", dst.config->queues[0]->queue); + TEST_ASSERT_EQUAL_UINT32(1u, dst.config->queues[0]->attrs); + TEST_ASSERT_EQUAL_INT32(5, dst.config->queues[0]->priority); + TEST_ASSERT_EQUAL_STRING("select=1", dst.config->queues[0]->resreq); + + /* jobs (SchedJobData) */ + TEST_ASSERT_NOT_NULL(dst.jobs); + TEST_ASSERT_EQUAL_UINT32(1u, dst.jobs->n_total); + TEST_ASSERT_EQUAL_INT(1, dst.jobs->count); + TEST_ASSERT_NOT_NULL(dst.jobs->jobs); + TEST_ASSERT_EQUAL_UINT64(8001ULL, dst.jobs->jobs[0]->jobid); + TEST_ASSERT_EQUAL_UINT32(5u, dst.jobs->jobs[0]->state); + TEST_ASSERT_EQUAL_INT(120, dst.jobs->jobs[0]->run_time); + + /* queues (top-level List(QueueConf)) */ + TEST_ASSERT_EQUAL_INT(1, dst.n_queues); + TEST_ASSERT_NOT_NULL(dst.queues); + TEST_ASSERT_EQUAL_STRING("batch", dst.queues[0]->queue); + TEST_ASSERT_EQUAL_INT32(5, dst.queues[0]->priority); + + /* workers (WorkerData) */ + TEST_ASSERT_NOT_NULL(dst.workers); + TEST_ASSERT_EQUAL_UINT32(1u, dst.workers->n_total); + TEST_ASSERT_NOT_NULL(dst.workers->workers); + TEST_ASSERT_EQUAL_STRING("workerA", dst.workers->workers[0]->name); + TEST_ASSERT_EQUAL_UINT32((uint32_t)XCE_WORKER_UNAVAIL, dst.workers->workers[0]->status); + + /* resources (ResourceData) */ + TEST_ASSERT_NOT_NULL(dst.resources); + TEST_ASSERT_EQUAL_UINT32(1u, dst.resources->n_total); + TEST_ASSERT_NOT_NULL(dst.resources->resources); + TEST_ASSERT_EQUAL_STRING("cpu-res", dst.resources->resources[0]->name); + TEST_ASSERT_EQUAL_STRING("CPU resource", dst.resources->resources[0]->desc); + + /* resvals (ResValList) */ + TEST_ASSERT_NOT_NULL(dst.resvals); + TEST_ASSERT_EQUAL_INT(1, dst.resvals->count); + TEST_ASSERT_NOT_NULL(dst.resvals->vals); + TEST_ASSERT_EQUAL_STRING("mem", dst.resvals->vals[0]->name); + TEST_ASSERT_EQUAL_STRING("hostA", dst.resvals->vals[0]->host); + TEST_ASSERT_NOT_NULL(dst.resvals->vals[0]->val); + TEST_ASSERT_EQUAL_UINT32(4096u, dst.resvals->vals[0]->val->u.uvalue32); + + /* users / projects (top-level List(SchedUser)/List(SchedProject)) */ + TEST_ASSERT_EQUAL_INT(1, dst.n_users); + TEST_ASSERT_NOT_NULL(dst.users); + TEST_ASSERT_EQUAL_STRING("alice", dst.users[0]->name); + + TEST_ASSERT_EQUAL_INT(1, dst.n_projects); + TEST_ASSERT_NOT_NULL(dst.projects); + TEST_ASSERT_EQUAL_STRING("proj1", dst.projects[0]->name); + TEST_ASSERT_EQUAL_INT32(5, dst.projects[0]->priority); + + /* prngSeed / internalVersion (plain scalars, codec-bridge tested) */ + TEST_ASSERT_EQUAL_UINT64(123456789ULL, dst.prng_seed); + TEST_ASSERT_EQUAL_UINT32(2u, dst.internal_version); + + /* internalState -- low-level accessor read-back (Data field, codec + * no-op; see comment above). */ + capn_data out_state = SnapshotData_get_internalState(dp); + TEST_ASSERT_EQUAL_INT((int)sizeof(internal_bytes), + capn_len((capn_list8){ out_state.p })); + uint8_t out_state_bytes[sizeof(internal_bytes)]; + TEST_ASSERT_EQUAL_INT((int)sizeof(internal_bytes), + capn_getv8((capn_list8){ out_state.p }, 0, + out_state_bytes, sizeof(internal_bytes))); + TEST_ASSERT_EQUAL_MEMORY(internal_bytes, out_state_bytes, sizeof(internal_bytes)); + + /* Observable: version-mismatch rejection. A reader that expects the + * SAME internal_version accepts; a reader that expects a DIFFERENT + * version must reject rather than misinterpret the opaque bytes. This + * is the only sound reject/accept rule for an opaque versioned blob -- + * demonstrated here with test-only comparison logic, not production + * code (task 8.1+ writes the real load_snapshot gate). */ + const uint32_t matching_reader_expects = 2u; + const uint32_t mismatched_reader_expects = 3u; + TEST_ASSERT_TRUE(dst.internal_version == matching_reader_expects); + TEST_ASSERT_FALSE(dst.internal_version == mismatched_reader_expects); + + free_SnapshotData(&dst); + capn_free(&d); + free(buf); + printf("test_snapshot_data_round_trip: PASS\n"); +} + /* ------------------------------------------------------------------ */ /* Suite entry point */ /* ------------------------------------------------------------------ */ @@ -806,4 +1363,9 @@ void test_e2e_capnp(void) RUN_TEST(test_session_summary_round_trip); RUN_TEST(test_job_state_change_event_round_trip); RUN_TEST(test_job_update_event_round_trip); + RUN_TEST(test_dispatch_decision_round_trip); + RUN_TEST(test_pend_decision_round_trip); + RUN_TEST(test_preempt_decision_round_trip); + RUN_TEST(test_replay_decision_dispatch_union_round_trip); + RUN_TEST(test_snapshot_data_round_trip); } diff --git a/src/include/typemap.h b/src/include/typemap.h index d858afe1..fd5be3b1 100644 --- a/src/include/typemap.h +++ b/src/include/typemap.h @@ -573,15 +573,52 @@ typedef struct { char *sched_policy; } replay_header_t; -/* Placeholder -- task 7.3 replaces this wholesale (not merely appends to) - * with the real 11-field snapshot struct (config/jobs/queues/workers/ - * resources/resvals/users/projects/prngSeed/internalState/internalVersion, - * design.md's Wire Protocol > ReplayFormat section). The single reserved - * byte below has no capnp counterpart -- SnapshotData has zero capnp fields - * in this task, so the generated encode_SnapshotData/decode_SnapshotData - * bodies are no-ops regardless of what this C struct contains. */ +/* task 7.3 (Req 5 AC 21): SchedUser/SchedProject mirror controller.capnp's + * SchedUser/SchedProject structs. Field lists match xce_sched_user_t/ + * xce_sched_project_t (src/scheduler/include/sched/types.h) exactly -- name + * only, name+priority respectively -- but these are FRESH, wire-only C + * types under different names, NOT aliases of the scheduler-internal ones. + * See controller.capnp's comment at `struct SchedUser` for the full + * rationale: sched/types.h is scheduler-private and already `#include`s + * this file, so a duplicate `xce_sched_user_t`/`xce_sched_project_t` + * typedef here would break every scheduler .c file that includes both + * headers (confirmed: core_events.c, schedobj.c already do). */ typedef struct { - uint8_t _reserved; + char *name; +} sched_snapshot_user_t; + +typedef struct { + char *name; + int32_t priority; +} sched_snapshot_project_t; + +/* task 7.3 (Req 5 AC 21): the full snapshot struct, replacing the task 7.1 + * placeholder wholesale. Every domain-object pointer reuses a pre-existing + * native type (already used natively throughout the scheduler/controller) + * except users/projects (sched_snapshot_user_t/sched_snapshot_project_t + * above) and the two scheduler-internal-only opaque fields. internal_state + * follows the same raw uint8_t-pointer + size_t pattern as every other Data + * field in this file (e.g. replay_header_t.config_hash above) -- the generated + * encode_SnapshotData/decode_SnapshotData codec is a documented no-op for + * it; real extraction is task 9+ hand-written code. internal_version is a + * format tag for internal_state so an incompatible future internal layout + * is rejected rather than misinterpreted (this task's Observable). */ +typedef struct { + xce_conf4sched_t *config; + xce_jobs_data_t *jobs; + xce_queue_t **queues; + int n_queues; + xce_workers_data_t *workers; + xce_resource_data_t *resources; + xce_resval_list_t *resvals; + sched_snapshot_user_t **users; + int n_users; + sched_snapshot_project_t **projects; + int n_projects; + uint64_t prng_seed; + uint8_t *internal_state; + size_t internal_state_len; + uint32_t internal_version; } sched_snapshot_data_t; typedef struct { @@ -636,10 +673,41 @@ typedef struct { uint32_t n_pended; } session_summary_t; -/* Placeholder -- task 7.3 adds the `body` union (dispatch/pend/preempt) - * here once DispatchDecision/PendDecision/PreemptDecision exist. */ +/* task 7.3 (Req 5 AC 5, AC 20): the three decision payload types -- + * dispatch/pend reuse existing native types (xce_job_alloc_t, + * xce_job_preason_t); preempt_victim_t/sched_preempt_decision_t are fresh, + * wire-only types, deliberately distinct from decision.h's core-internal + * xce_preempt_victim_t/xce_preempt_t (task 4.4's in-process live-decision + * representation) -- see controller.capnp's matching comment. */ +typedef struct { + xce_job_alloc_t *alloc; +} sched_dispatch_decision_t; + +typedef struct { + uint64_t jobid; + xce_job_preason_t *reason; +} sched_pend_decision_t; + +typedef struct { + uint64_t jobid; + char *workerid; +} preempt_victim_t; + +typedef struct { + uint64_t jobid; + preempt_victim_t **victims; + int n_victims; + xce_job_preason_t *reason; +} sched_preempt_decision_t; + typedef struct { uint64_t ts; + int body_which; + union { + sched_dispatch_decision_t *dispatch; + sched_pend_decision_t *pend; + sched_preempt_decision_t *preempt; + } body; } replay_decision_t; typedef struct { diff --git a/src/schema/controller.capnp b/src/schema/controller.capnp index 273b7156..7cb889b3 100644 --- a/src/schema/controller.capnp +++ b/src/schema/controller.capnp @@ -999,17 +999,26 @@ struct CtrlHelloReply $C.mapname("ctrl_hello_reply_t") { # - task 7.2: adds the new JobStateChangeEvent + JobUpdateEvent structs, # then adds two new union members (jobStateChange, jobUpdate) to # ReplayEvent's `body` union at new ordinals (10, 11) -- purely additive. -# - task 7.3: adds SnapshotData's real 11 fields (Req 5 AC 21) starting at -# ordinal @0 -- replacing this task's zero-field placeholder wholesale, -# both in this schema and in typemap.h's sched_snapshot_data_t (which -# has no capnp-visible fields yet, so nothing there is "renumbered" -# either). task 7.3 also adds DispatchDecision/PendDecision/ -# PreemptVictim/PreemptDecision, then adds a `body :union { dispatch @1: -# DispatchDecision; pend @2: PendDecision; preempt @3: PreemptDecision; -# }` to ReplayDecision -- purely additive new ordinals on the ts@0 -# struct this task creates (see the JobControl/JoinData `:union` group -# precedent above: a union group name never consumes its own ordinal, -# only its member variants do, so this is safe to bolt on later). +# - task 7.3 (DONE): added SnapshotData's real 11 fields (Req 5 AC 21) +# starting at ordinal @0, replacing task 7.1's zero-field placeholder +# wholesale (both here and in typemap.h's sched_snapshot_data_t). Added +# DispatchDecision/PendDecision/PreemptVictim/PreemptDecision, then +# added `body :union { dispatch @1: DispatchDecision; pend @2: +# PendDecision; preempt @3: PreemptDecision; }` to ReplayDecision -- +# purely additive new ordinals on the ts@0 struct task 7.1 created (see +# the JobControl/JoinData `:union` group precedent above: a union group +# name never consumes its own ordinal, only its member variants do). +# DEVIATION from design.md's literal text: SnapshotData also needed two +# new user/project wire types. design.md's literal spec names the +# mapname target `xce_sched_user_t`/`xce_sched_project_t` (the +# scheduler-core-internal C types in src/scheduler/include/sched/ +# types.h) -- but that header is scheduler-private, unreachable from +# typemap.h, and already `#include`s typemap.h itself, so reusing that +# exact name here would be a guaranteed duplicate-typedef compile error +# across src/scheduler/. Used fresh mapnames `sched_snapshot_user_t`/ +# `sched_snapshot_project_t` instead (see `struct SchedUser` comment +# below for the full rationale) -- field lists still match design.md's +# spec exactly, only the C type name differs. struct ReplayCluster $C.mapname("replay_cluster_t") { name @0: Text; @@ -1026,14 +1035,61 @@ struct ReplayHeader $C.mapname("replay_header_t") { schedPolicy @5: Text $C.mapname("sched_policy"); # active picker plugin name (e.g. "builtin_fcfs") } -# Placeholder -- task 7.3 replaces this with the full snapshot schema -# (Req 5 AC 21): config, jobs, queues, workers, resources, resvals, users, -# projects, prngSeed, internalState, internalVersion (see design.md's Wire -# Protocol > ReplayFormat section for the already-committed 11-field -# layout). Deliberately left with ZERO capnp fields here (instead of a -# placeholder scalar) so task 7.3 is free to start numbering its real -# fields at ordinal @0 exactly as design.md specifies, with no renumbering. +# task 7.3 (Req 5 AC 21): the two scheduler-only user/project wire types +# SnapshotData embeds. design.md's literal spec reuses the scheduler-core +# core-internal C type names xce_sched_user_t/xce_sched_project_t +# (sched/types.h) verbatim as the mapname target -- but those types are +# scheduler-PRIVATE (sched/types.h is under src/scheduler/include/, out of +# this schema-only task's edit scope, and is NOT reachable from typemap.h, +# which every OTHER binary linking controller.capnp -- controller, apiserver +# CGO, xceadmin, worker -- also compiles against). sched/types.h itself +# `#include`s typemap.h (confirmed: scheduler/src/core_events.c and +# schedobj.c already include both headers in the same translation unit), so +# adding a second, textually-distinct `xce_sched_user_t`/`xce_sched_project_t` +# typedef to typemap.h would be a guaranteed duplicate-typedef compile error +# across most of src/scheduler/ the moment both headers are included +# together -- not hypothetical, a certainty given the current include graph. +# Field lists match design.md's literal spec exactly (verified against +# sched/types.h: xce_sched_user_t has name only; xce_sched_project_t has +# name+priority) -- only the C TARGET TYPE NAME differs, to stay reachable +# from typemap.h without touching src/scheduler/. This mirrors the +# PreemptVictim/PreemptDecision precedent below: a fresh, wire-only mapname +# distinct from a similarly-shaped core-internal type of a different name. +struct SchedUser $C.mapname("sched_snapshot_user_t") { + name @0: Text; +} + +struct SchedProject $C.mapname("sched_snapshot_project_t") { + name @0: Text; + priority @1: Int32; +} + +# task 7.3 (Req 5 AC 21): the full snapshot message type, replacing task +# 7.1's zero-field placeholder wholesale. Every domain-object field reuses +# a pre-existing wire struct (Conf4Sched/SchedJobData/QueueConf/WorkerData/ +# ResourceData/ResValList, all already used by ReplayEvent above) except +# SchedUser/SchedProject (fresh, see comment above) and the two +# scheduler-internal-only opaque fields. internalState is a versioned +# opaque blob (shared-resource tracking + unsatisfiable-resreq cache + +# session/iterator state) with no capnp-visible structure -- the core owns +# its byte layout; internalVersion is a format tag so a future incompatible +# internal layout is rejected rather than misinterpreted (this task's +# Observable). Like every other Data field in this schema (see +# ReplayHeader.configHash above), internalState's generated encode_/decode_ +# codec is a documented no-op -- real extraction is hand-written +# low-level-accessor code, task 9+'s job, same pattern. struct SnapshotData $C.mapname("sched_snapshot_data_t") { + config @0: Conf4Sched; + jobs @1: SchedJobData; + queues @2: List(QueueConf) $C.maplistcount("n_queues"); + workers @3: WorkerData; + resources @4: ResourceData; + resvals @5: ResValList; + users @6: List(SchedUser) $C.maplistcount("n_users"); + projects @7: List(SchedProject) $C.maplistcount("n_projects"); + prngSeed @8: UInt64 $C.mapname("prng_seed"); + internalState @9: Data $C.mapname("internal_state"); + internalVersion @10: UInt32 $C.mapname("internal_version"); } struct SnapshotEntry $C.mapname("replay_snapshot_entry_t") { @@ -1091,12 +1147,47 @@ struct SessionSummary $C.mapname("session_summary_t") { nPended @3: UInt32 $C.mapname("n_pended"); } -# Placeholder -- task 7.3 adds `body :union { dispatch @1: DispatchDecision; -# pend @2: PendDecision; preempt @3: PreemptDecision; }` here once those -# payload types exist (purely additive new ordinals 1-3; ts@0 untouched -- -# see the JobControl/JoinData precedent noted above). +# task 7.3 (Req 5 AC 5, AC 20): the three decision payload types. dispatch +# and pend reuse pre-existing wire structs verbatim (JobAlloc, JobPendReason +# -- design review already confirmed JobAlloc.jobid is DispatchDecision's +# sole job-identity carrier, no redundant jobid field). PreemptVictim is +# fresh (jobid+workerid has no pre-existing wire counterpart). PreemptVictim/ +# PreemptDecision are DELIBERATELY distinct wire-only types from decision.h's +# core-internal xce_preempt_victim_t/xce_preempt_t (task 4.4's in-process +# live-decision representation) -- the wire format and the core's live +# in-process decision representation are independent, never-aliased things +# that happen to model similar data. +struct DispatchDecision $C.mapname("sched_dispatch_decision_t") { + alloc @0: JobAlloc; +} + +struct PendDecision $C.mapname("sched_pend_decision_t") { + jobid @0: UInt64; + reason @1: JobPendReason; +} + +struct PreemptVictim $C.mapname("preempt_victim_t") { + jobid @0: UInt64; + workerid @1: Text; +} + +struct PreemptDecision $C.mapname("sched_preempt_decision_t") { + jobid @0: UInt64; + victims @1: List(PreemptVictim) $C.maplistcount("n_victims"); + reason @2: JobPendReason; +} + +# task 7.3 (Req 5 AC 20): purely additive `body` union on top of task 7.1's +# ts@0-only stub -- new ordinals 1-3, ts@0 untouched (see the JobControl/ +# JoinData precedent noted above: a union group name never consumes its own +# ordinal). struct ReplayDecision $C.mapname("replay_decision_t") { ts @0: UInt64; + body :union $C.mapuniontag("body_which") { + dispatch @1: DispatchDecision; + pend @2: PendDecision; + preempt @3: PreemptDecision; + } } struct ReplaySession $C.mapname("replay_session_t") {