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") {