From e53c086ec5e441909c3f94e6a6b5f1dfd3262ace Mon Sep 17 00:00:00 2001 From: Rongsong Shen Date: Wed, 22 Jul 2026 09:22:04 +0000 Subject: [PATCH] fix(schema): rename struct Data -> XceData to unshadow capnp builtin blob type controller.capnp declared `struct Data $C.mapname("xce_data_t")` (an OPCode_get query-result tagged union) at file scope, which shadows Cap'n Proto's builtin Data blob type for every later `: Data` field in the same file. This silently miscompiled 7 fields across 6 structs as a struct-pointer type instead of a raw byte blob: UpgradeConfigRequest .sha256/.bytes, ReplConfigReply.sha256/.bytes, ReplConfigPushRequest .sha256/.bytes, ReplEnvelope.payload, SnapshotTransfer's chunk payload, WalCatchupEntry.payload/.batchPayloads, and CtrlHelloRequest .key. Production C code had already routed around it for years via hand-rolled raw capn_ptr manipulation (see request.c's own comments); nothing before now needed the generated codec/accessors to actually work for these fields. Renamed the local struct to XceData (capnp-level name only; the $C.mapname("xce_data_t") C typedef is unchanged) and explicitly repointed getdata/pushmsg at XceData by name, since those two fields correctly want the local tagged union. Updated every derived generated-identifier reference across 19 hand-written files spanning controller, lib, scheduler, and the Go apiserver/xcecli CGo bindings. Fixed 5 call sites in lib/request.c that had been typed against the old workaround alias but now legitimately receive the correct capn_data type (a pure type-name swap -- Data_ptr/XceData_ptr and capn_data share the identical {capn_ptr p} layout). Verified via full rebuild+test: lib (1/1), controller (3/3, incl. the job-array push regression tests), scheduler (15/15), Go apiserver + xcecli (all packages green), and a direct pycapnp round-trip proving UpgradeConfigRequest.sha256/.bytes now encode/decode correctly. Bug: bug-1461 (.wolf/buglog.json). Unblocks admin-config-tool task 3.3. --- src/controller/apiserver.c | 10 +- src/controller/opget.c | 12 +-- src/controller/sched.c | 10 +- src/controller/testsuite/test.c | 8 +- src/controller/testsuite/test_e2e_capnp.c | 26 ++--- .../testsuite/test_job_array_apiserver_push.c | 10 +- .../testsuite/test_job_array_sched_push.c | 10 +- src/frontend/apiserver/backend/clib.go | 8 +- src/frontend/apiserver/backend/handler.go | 4 +- .../apiserver/backend/handler_test.go | 2 +- .../backend/wire_dispatch_support.go | 4 +- src/lib/helper.c | 24 ++--- src/lib/request.c | 100 +++++++++--------- src/lib/testsuite/test_request.c | 10 +- src/lib/xceapi.c | 14 +-- src/scheduler/src/handler.c | 14 +-- src/scheduler/src/schedobj.c | 2 +- src/scheduler/testsuite/test_handlers.c | 2 +- .../testsuite/test_job_array_expand.c | 8 +- .../testsuite/test_job_array_wire_e2e.c | 4 +- src/schema/controller.capnp | 6 +- 21 files changed, 144 insertions(+), 144 deletions(-) diff --git a/src/controller/apiserver.c b/src/controller/apiserver.c index ef200e18..08f9f2a6 100644 --- a/src/controller/apiserver.c +++ b/src/controller/apiserver.c @@ -32,7 +32,7 @@ static int xce_apiserver_push_workers(cluster_t* c, endpoint_t* ep) memset(&data, 0x0, sizeof(data)); xlist_init(&wlist, 8192); - data.table = Data_data_workers; + data.table = XceData_data_workers; data.data.workers = &workers; collect_workers(c, (collect_check_fn_t)check_worker_, &wlist); @@ -77,7 +77,7 @@ static int xce_apiserver_push_jobs(cluster_t* c, endpoint_t* ep) memset(&data, 0x0, sizeof(data)); xlist_init(&jlist, 1024 * 1024); - data.table = Data_data_jobs; + data.table = XceData_data_jobs; data.data.jobs = &jobs; collect_jobs(c, (collect_check_fn_t)check_job_, &jlist); @@ -108,7 +108,7 @@ static int apiserver_push_joblist(cluster_t* c, xlist_t* l) .jobs = (xce_job_t**)l->data }; memset(&data, 0x0, sizeof(data)); - data.table = Data_data_jobs; + data.table = XceData_data_jobs; data.data.jobs = &jobs; rc = xce_push_data_req(c->apiservers[i].ep, c->myhost, &data, @@ -177,7 +177,7 @@ static int apiserver_push_jobarray(cluster_t* c, xlist_t* l) int rc; memset(&data, 0x0, sizeof(data)); - data.table = Data_data_jobArray; + data.table = XceData_data_jobArray; data.data.job_array = &job_array; memset(&request, 0x0, sizeof(request)); @@ -430,7 +430,7 @@ static int apiserver_push_workerlist(cluster_t* c, endpoint_t* ep, xlist_t* l) memset(&data, 0x0, sizeof(data)); - data.table = Data_data_workers; + data.table = XceData_data_workers; data.data.workers = &workers; rc = xce_push_data_req(ep, c->myhost, &data, check_controller, c, 100); diff --git a/src/controller/opget.c b/src/controller/opget.c index 57c8f44e..4194bd51 100644 --- a/src/controller/opget.c +++ b/src/controller/opget.c @@ -199,18 +199,18 @@ int do_get_op(cluster_t* c, endpoint_t* ep, xce_request_t* request) init_reply_by_req(&reply, request); switch (request->body.getmsg->table) { - case Data_data_resources: + case XceData_data_resources: rc = do_reply_reslist(c, ep, request); break; - case Data_data_workers: - case Data_data_schedworkers: + case XceData_data_workers: + case XceData_data_schedworkers: rc = do_reply_workerlist(c, ep, request); break; - case Data_data_schedjobs: - case Data_data_jobs: + case XceData_data_schedjobs: + case XceData_data_jobs: rc = do_reply_joblist(c, ep, request); break; - case Data_data_resvals: + case XceData_data_resvals: rc = do_reply_resvals(c, ep, request); break; default: diff --git a/src/controller/sched.c b/src/controller/sched.c index e2ac9579..f17ebba8 100644 --- a/src/controller/sched.c +++ b/src/controller/sched.c @@ -56,7 +56,7 @@ static int xcesched_push_jobarray(cluster_t* c, xlist_t* l) job_array.submit = first->submit; memset(&data, 0x0, sizeof(data)); - data.table = Data_data_jobArray; + data.table = XceData_data_jobArray; data.data.job_array = &job_array; memset(&request, 0x0, sizeof(request)); @@ -103,7 +103,7 @@ static int xcesched_push_joblist(cluster_t* c, xlist_t* l) memset(&data, 0x0, sizeof(data)); - data.table = Data_data_schedjobs; + data.table = XceData_data_schedjobs; data.data.jobs = &jobs; /* No committed jobs to push → nothing to send (avoids an empty payload). */ @@ -127,7 +127,7 @@ static int xcesched_push_workerlist(cluster_t* c, xlist_t* l) memset(&data, 0x0, sizeof(data)); - data.table = Data_data_schedworkers; + data.table = XceData_data_schedworkers; data.data.workers = &workers; rc = xce_push_data_req(c->sched.ep, c->myhost, &data, check_controller, c, @@ -263,7 +263,7 @@ static int xcesched_push_jobs(cluster_t* c) xlist_init(&jlist, 1024 * 1024); - data.table = Data_data_schedjobs; + data.table = XceData_data_schedjobs; data.data.jobs = &jobs; /* ha-convergence-recovery (3.1, 3.2): scheduler-bound full push gates on the @@ -297,7 +297,7 @@ static int xcesched_push_workers(cluster_t* c) memset(&data, 0x0, sizeof(data)); xlist_init(&wlist, 8192); - data.table = Data_data_schedworkers; + data.table = XceData_data_schedworkers; data.data.workers = &workers; collect_workers(c, (collect_check_fn_t)check_worker_, &wlist); diff --git a/src/controller/testsuite/test.c b/src/controller/testsuite/test.c index 8afb5f34..1a329fcc 100644 --- a/src/controller/testsuite/test.c +++ b/src/controller/testsuite/test.c @@ -3412,12 +3412,12 @@ static int myreq_handler(xce_request_t* request, void* args) dataset_t* datamsg = request->body.datamsg; switch (datamsg->table) { - case Data_data_schedjobs: - case Data_data_jobs: + case XceData_data_schedjobs: + case XceData_data_jobs: sched_on_jobs(datamsg->nrows, (xce_job_t**)datamsg->rows); break; - case Data_data_schedworkers: - case Data_data_workers: + case XceData_data_schedworkers: + case XceData_data_workers: sched_on_workers(datamsg->nrows, (xce_worker_t**)datamsg->rows); break; default: diff --git a/src/controller/testsuite/test_e2e_capnp.c b/src/controller/testsuite/test_e2e_capnp.c index c8f56da4..1d98eadf 100644 --- a/src/controller/testsuite/test_e2e_capnp.c +++ b/src/controller/testsuite/test_e2e_capnp.c @@ -275,7 +275,7 @@ static void test_data_job_array_variant_round_trip(void) xce_data_t data; memset(&data, 0, sizeof(data)); - data.table = Data_data_jobArray; + data.table = XceData_data_jobArray; data.data.job_array = &src; /* Encode -> flat wire buffer -> decode, exercising both the schema @@ -285,8 +285,8 @@ static void test_data_job_array_variant_round_trip(void) capn_init_malloc(&c); struct capn_segment *cs = capn_root(&c).seg; - Data_ptr p; - encode_Data_ptr(cs, &p, &data); + XceData_ptr p; + encode_XceData_ptr(cs, &p, &data); capn_setp(capn_root(&c), 0, p.p); int64_t sz = capn_size(&c); @@ -298,15 +298,15 @@ static void test_data_job_array_variant_round_trip(void) struct capn d; TEST_ASSERT_EQUAL_INT(0, capn_init_mem(&d, buf, (size_t)sz, 0)); - Data_ptr dp; + XceData_ptr dp; dp.p = capn_getp(capn_root(&d), 0, 1); - struct Data wire; - read_Data(&wire, dp); + struct XceData wire; + read_XceData(&wire, dp); xce_data_t dst; memset(&dst, 0, sizeof(dst)); - decode_Data(&dst, &wire); + decode_XceData(&dst, &wire); - TEST_ASSERT_EQUAL_INT(Data_data_jobArray, dst.table); + TEST_ASSERT_EQUAL_INT(XceData_data_jobArray, dst.table); TEST_ASSERT_NOT_NULL(dst.data.job_array); TEST_ASSERT_EQUAL_UINT64(5000ULL, dst.data.job_array->jobid); TEST_ASSERT_EQUAL_UINT64(250ULL, dst.data.job_array->array_size); @@ -320,12 +320,12 @@ static void test_data_job_array_variant_round_trip(void) /* Untouched: the existing schedjobs/jobs/etc. wire layout is unaffected -- * spot-check the enum values these variants relied on before this change * did not shift underneath them. */ - TEST_ASSERT_EQUAL_INT(0, (int)Data_data_schedjobs); - TEST_ASSERT_EQUAL_INT(2, (int)Data_data_jobs); - TEST_ASSERT_EQUAL_INT(5, (int)Data_data_resvals); - TEST_ASSERT_EQUAL_INT(6, (int)Data_data_jobArray); + TEST_ASSERT_EQUAL_INT(0, (int)XceData_data_schedjobs); + TEST_ASSERT_EQUAL_INT(2, (int)XceData_data_jobs); + TEST_ASSERT_EQUAL_INT(5, (int)XceData_data_resvals); + TEST_ASSERT_EQUAL_INT(6, (int)XceData_data_jobArray); - free_Data(&dst); + free_XceData(&dst); capn_free(&d); free(buf); printf("test_data_job_array_variant_round_trip: PASS\n"); diff --git a/src/controller/testsuite/test_job_array_apiserver_push.c b/src/controller/testsuite/test_job_array_apiserver_push.c index 5e827064..c8a0bedc 100644 --- a/src/controller/testsuite/test_job_array_apiserver_push.c +++ b/src/controller/testsuite/test_job_array_apiserver_push.c @@ -201,12 +201,12 @@ static void ap_scan_pushes(endpoint_t* ep, ap_push_obs_t* obs) rp.p = capn_getp(capn_root(&c), 0, 1); read_Request(&rq, rp); if (rq.body_which == Request_body_pushmsg) { - struct Data d; + struct XceData d; - read_Data(&d, rq.body.pushmsg); + read_XceData(&d, rq.body.pushmsg); obs->n_pushes += 1; obs->last_data_which = (int)d.data_which; - if (d.data_which == Data_data_jobArray) { + if (d.data_which == XceData_data_jobArray) { struct JobSubmit js; read_JobSubmit(&js, d.data.jobArray); @@ -255,7 +255,7 @@ test_multi_element_job_new_produces_one_compact_jobarray_push_per_server(void) "job structs and not fewer than one per connected server (Req " "6.2, 6.4)"); TEST_ASSERT_EQUAL_INT_MESSAGE( - Data_data_jobArray, obs.last_data_which, + XceData_data_jobArray, obs.last_data_which, "the push must encode the compact Data.jobArray descriptor, not " "Data.jobs (N individually-encoded job structs)"); TEST_ASSERT_EQUAL_UINT64_MESSAGE( @@ -302,7 +302,7 @@ static void test_single_element_job_new_is_unaffected(void) "an ordinary single-job notification must still produce exactly " "one push packet per connected API server, unchanged"); TEST_ASSERT_EQUAL_INT_MESSAGE( - Data_data_jobs, obs.last_data_which, + XceData_data_jobs, obs.last_data_which, "a single-element notification must go through the EXISTING, " "UNMODIFIED apiserver_push_joblist() path (Data.jobs), never " "the new compact jobArray path (Req 8.1)"); diff --git a/src/controller/testsuite/test_job_array_sched_push.c b/src/controller/testsuite/test_job_array_sched_push.c index 1663e9d9..bb453a99 100644 --- a/src/controller/testsuite/test_job_array_sched_push.c +++ b/src/controller/testsuite/test_job_array_sched_push.c @@ -161,12 +161,12 @@ static void jsp_scan_pushes(endpoint_t* ep, jsp_push_obs_t* obs) rp.p = capn_getp(capn_root(&c), 0, 1); read_Request(&rq, rp); if (rq.body_which == Request_body_pushmsg) { - struct Data d; + struct XceData d; - read_Data(&d, rq.body.pushmsg); + read_XceData(&d, rq.body.pushmsg); obs->n_pushes += 1; obs->last_data_which = (int)d.data_which; - if (d.data_which == Data_data_jobArray) { + if (d.data_which == XceData_data_jobArray) { struct JobSubmit js; read_JobSubmit(&js, d.data.jobArray); @@ -209,7 +209,7 @@ static void test_multi_element_job_new_produces_one_compact_jobarray_push(void) "a 5-element array notification must produce exactly ONE push packet, " "not one per element (Req 6.1, 6.4)"); TEST_ASSERT_EQUAL_INT_MESSAGE( - Data_data_jobArray, obs.last_data_which, + XceData_data_jobArray, obs.last_data_which, "the push must encode the compact Data.jobArray descriptor, not " "Data.schedjobs/Data.jobs (N individually-encoded job structs)"); TEST_ASSERT_EQUAL_UINT64_MESSAGE( @@ -251,7 +251,7 @@ static void test_single_element_job_new_is_unaffected(void) "an ordinary single-job notification must still produce exactly one " "push packet, unchanged"); TEST_ASSERT_EQUAL_INT_MESSAGE( - Data_data_schedjobs, obs.last_data_which, + XceData_data_schedjobs, obs.last_data_which, "a single-element notification must go through the EXISTING, " "UNMODIFIED xcesched_push_joblist() path (Data.schedjobs), never the " "new compact jobArray path (Req 8.1)"); diff --git a/src/frontend/apiserver/backend/clib.go b/src/frontend/apiserver/backend/clib.go index e993c68f..a93a75a0 100644 --- a/src/frontend/apiserver/backend/clib.go +++ b/src/frontend/apiserver/backend/clib.go @@ -229,7 +229,7 @@ func handlePush(req *C.xce_request_t) C.int { // handleDataset processes a dataset push (jobs or workers). func handleDataset(d *C.dataset_t) C.int { switch d.table { - case C.Data_data_jobs: + case C.XceData_data_jobs: n := int(d.nrows) if n == 0 { return 0 @@ -243,7 +243,7 @@ func handleDataset(d *C.dataset_t) C.int { gHandler.OnJobUpdated(info) } - case C.Data_data_workers: + case C.XceData_data_workers: n := int(d.nrows) if n == 0 { return 0 @@ -257,9 +257,9 @@ func handleDataset(d *C.dataset_t) C.int { gHandler.OnHostUpdated(info) } - case C.Data_data_jobArray: + case C.XceData_data_jobArray: // job-array-single-wal-record task 5.4 (Req 6.2, 6.4, 6.5, 8.4): - // sibling receive path to the Data_data_jobs case above, for the + // sibling receive path to the XceData_data_jobs case above, for the // compact array descriptor apiserver_push_jobarray() (apiserver.c, // task 5.2) sends. Unlike the LIST-shaped cases, extract_data() // (request.c, task 5.3b) always delivers exactly one descriptor diff --git a/src/frontend/apiserver/backend/handler.go b/src/frontend/apiserver/backend/handler.go index e78c7b4e..d5465f08 100644 --- a/src/frontend/apiserver/backend/handler.go +++ b/src/frontend/apiserver/backend/handler.go @@ -97,13 +97,13 @@ func (h *DefaultHandler) OnResourceLocationUpdated(res *models.ResourceDistribut // dispatchJobArray delivers a locally-expanded compact array push to the // handler, one call per genuinely-new element (job-array-single-wal-record // task 5.4, Req 6.2, 6.4, 6.5, 8.4). CGO-free (called from handleDataset's -// Data_data_jobArray case in clib.go; the elements themselves are already +// XceData_data_jobArray case in clib.go; the elements themselves are already // decoded by cJobArrayToModels/rawJobArrayToModels before this is called), // mirroring dispatchConfigPush's split for testability without CGO. // // Dedup reuses store.Jobs itself — the SAME jobid-keyed store the ordinary // single-job push path already writes through (handleDataset's -// Data_data_jobs case calls OnJobUpdated per job with no separate dedup +// XceData_data_jobs case calls OnJobUpdated per job with no separate dedup // step, relying on the map being keyed by jobid) — checked here BEFORE // calling the handler so a redelivered push (e.g. after an apiserver // reconnect) can never clobber a jobid whose state has since moved on. diff --git a/src/frontend/apiserver/backend/handler_test.go b/src/frontend/apiserver/backend/handler_test.go index 3a277407..661a89cb 100644 --- a/src/frontend/apiserver/backend/handler_test.go +++ b/src/frontend/apiserver/backend/handler_test.go @@ -365,7 +365,7 @@ func TestDispatchConfigPush_NilClusterIsSkipped(t *testing.T) { // --- job-array-single-wal-record task 5.4: dispatchJobArray --- // -// dispatchJobArray is the sibling of the single-job Data_data_jobs push +// dispatchJobArray is the sibling of the single-job XceData_data_jobs push // path (clib.go handleDataset, which calls gHandler.OnJobUpdated per job // with no separate dedup step because store.Jobs is a map keyed by jobid). // For the array path, dedup happens BEFORE calling the handler so a diff --git a/src/frontend/apiserver/backend/wire_dispatch_support.go b/src/frontend/apiserver/backend/wire_dispatch_support.go index 3aaf183f..686ac501 100644 --- a/src/frontend/apiserver/backend/wire_dispatch_support.go +++ b/src/frontend/apiserver/backend/wire_dispatch_support.go @@ -75,7 +75,7 @@ func wireDispatchBuildPush(baseJobID, arraySize uint64) (unsafe.Pointer, error) jobArray.submit = submit data := (*C.xce_data_t)(C.calloc(1, C.size_t(unsafe.Sizeof(C.xce_data_t{})))) - data.table = C.Data_data_jobArray + data.table = C.XceData_data_jobArray C.wire_test_data_set_job_array(data, jobArray) var request C.xce_request_t @@ -178,7 +178,7 @@ func wireDispatchDeliver(pktHandle unsafe.Pointer) (int, error) { // go_push_handler already freed *ds (see above) -- only the payload // xce_job_submit_t it pointed to still needs freeing: free_Dataset() - // never touches ->rows, and handleDataset()'s Data_data_jobArray case + // never touches ->rows, and handleDataset()'s XceData_data_jobArray case // (clib.go) never nulls d.rows after use (every element is its own // independent Go-native copy via cJobArrayToModels, mirroring the C // scheduler side's sched_handle_dataset() jobArray case). Freeing it now diff --git a/src/lib/helper.c b/src/lib/helper.c index 4491f1d1..bc9013a1 100644 --- a/src/lib/helper.c +++ b/src/lib/helper.c @@ -230,17 +230,17 @@ void* xce_configdata_ptr(xce_configdata_t* d) void* xce_data_ptr(xce_data_t* d) { switch (d->table) { - case Data_data_schedjobs: + case XceData_data_schedjobs: return d->data.jobs; - case Data_data_schedworkers: + case XceData_data_schedworkers: return d->data.workers; - case Data_data_jobs: + case XceData_data_jobs: return d->data.jobs; - case Data_data_workers: + case XceData_data_workers: return d->data.workers; - case Data_data_resources: + case XceData_data_resources: return d->data.resources; - case Data_data_resvals: + case XceData_data_resvals: return d->data.resvals; default: break; @@ -345,17 +345,17 @@ void* xce_reply_data_ptr(xce_reply_t* reply) char* xce_dataset_table(int table) { switch (table) { - case Data_data_schedjobs: + case XceData_data_schedjobs: return "schedjobs"; - case Data_data_jobs: + case XceData_data_jobs: return "jobs"; - case Data_data_workers: + case XceData_data_workers: return "workers"; - case Data_data_schedworkers: + case XceData_data_schedworkers: return "schedworkers"; - case Data_data_resources: + case XceData_data_resources: return "resources"; - case Data_data_resvals: + case XceData_data_resvals: return "resvals"; default: break; diff --git a/src/lib/request.c b/src/lib/request.c index 10cf2bbd..aa85c57f 100644 --- a/src/lib/request.c +++ b/src/lib/request.c @@ -60,7 +60,7 @@ int xce_encode(xce_request_t* req, io_packet_t* pkt) * malloc'd buffer. Returns NULL (*out_len = 0) for absent/empty fields or * allocation failure. Used by xce_decode() to pre-extract raw payload * bytes before capn_free() destroys the segment. */ -static uint8_t *extract_data_bytes(Data_ptr dp, size_t *out_len) +static uint8_t *extract_data_bytes(capn_data dp, size_t *out_len) { *out_len = 0; @@ -100,7 +100,7 @@ int xce_decode(io_packet_t* pkt, xce_request_t* req) /* OPCode_upgradeConfig carries sha256 and bytes as List(UInt8) on the wire. * decode_UpgradeConfigRequest() intentionally leaves raw_sha256/raw_bytes as - * NULL (the project's Data union cannot carry raw bytes via decode_Data_ptr). + * NULL (the project's Data union cannot carry raw bytes via decode_XceData_ptr). * Extract them here using UpgradeConfigRequest_get_* + capn_getv8 BEFORE * capn_free() destroys the capnp segment. */ if (rq.opcode == OPCode_upgradeConfig @@ -112,7 +112,7 @@ int xce_decode(io_packet_t* pkt, xce_request_t* req) /* sha256 */ { - Data_ptr dp = UpgradeConfigRequest_get_sha256(u_ptr); + capn_data dp = UpgradeConfigRequest_get_sha256(u_ptr); capn_list8 sha_list; sha_list.p = dp.p; if (sha_list.p.type != CAPN_NULL) { @@ -130,7 +130,7 @@ int xce_decode(io_packet_t* pkt, xce_request_t* req) /* bytes */ { - Data_ptr dp = UpgradeConfigRequest_get_bytes(u_ptr); + capn_data dp = UpgradeConfigRequest_get_bytes(u_ptr); capn_list8 bytes_list; bytes_list.p = dp.p; if (bytes_list.p.type != CAPN_NULL) { @@ -158,7 +158,7 @@ int xce_decode(io_packet_t* pkt, xce_request_t* req) /* sha256 */ { - Data_ptr dp = ReplConfigPushRequest_get_sha256(r_ptr); + capn_data dp = ReplConfigPushRequest_get_sha256(r_ptr); capn_list8 sha_list; sha_list.p = dp.p; if (sha_list.p.type != CAPN_NULL && capn_len(sha_list) > 0) { int slen = capn_len(sha_list); @@ -172,7 +172,7 @@ int xce_decode(io_packet_t* pkt, xce_request_t* req) /* bytes */ { - Data_ptr dp = ReplConfigPushRequest_get_bytes(r_ptr); + capn_data dp = ReplConfigPushRequest_get_bytes(r_ptr); capn_list8 bytes_list; bytes_list.p = dp.p; if (bytes_list.p.type != CAPN_NULL && capn_len(bytes_list) > 0) { int blen = capn_len(bytes_list); @@ -963,7 +963,7 @@ static int ssl_read_all(SSL *ssl, void *dst, size_t len) * version_tag, sha256 and bytes. * * Bypasses xce_decode (which would route sha256/bytes through - * decode_Data_ptr → xce_data_t and lose the raw byte content). + * decode_XceData_ptr → xce_data_t and lose the raw byte content). * Instead, reads the capnp message directly into capn, walks down * Request → RequestReply → ReplConfigReply, and pulls sha256/bytes * via capn_len + capn_getv8 on the raw capn_list8. @@ -1050,8 +1050,8 @@ int xce_replconfig_reply_ssl_decode(SSL *ssl, *out_up_to_date = (int)reply_struct.upToDate; *out_tag = reply_struct.versionTag; - /* Pull sha256 and bytes via raw capn_list8 access (NOT decode_Data - * which would corrupt via xce_data_t). Data_ptr and capn_list8 + /* Pull sha256 and bytes via raw capn_list8 access (NOT decode_XceData + * which would corrupt via xce_data_t). XceData_ptr and capn_list8 * share {capn_ptr p} layout. */ sha_list.p = reply_struct.sha256.p; bytes_list.p = reply_struct.bytes.p; @@ -1406,16 +1406,16 @@ int xce_push_data_req(endpoint_t* ep, xce_peerinfo_t* from, xce_data_t* data, data1.table = data->table; switch (data->table) { - case Data_data_schedjobs: - case Data_data_jobs: + case XceData_data_schedjobs: + case XceData_data_jobs: nrows = data->data.jobs->n_total; data1.data.jobs = (xce_jobs_data_t*)calloc(1, sizeof(xce_jobs_data_t)); if (data1.data.jobs == NULL) { rc = -ENOMEM; } break; - case Data_data_schedworkers: - case Data_data_workers: + case XceData_data_schedworkers: + case XceData_data_workers: nrows = data->data.workers->n_total; data1.data.workers = (xce_workers_data_t*)calloc(1, sizeof(xce_workers_data_t)); @@ -1423,9 +1423,9 @@ int xce_push_data_req(endpoint_t* ep, xce_peerinfo_t* from, xce_data_t* data, rc = -ENOMEM; } break; - case Data_data_resources: + case XceData_data_resources: break; - case Data_data_resvals: + case XceData_data_resvals: break; default: rc = -EINVAL; @@ -1455,8 +1455,8 @@ int xce_push_data_req(endpoint_t* ep, xce_peerinfo_t* from, xce_data_t* data, } switch (data->table) { - case Data_data_schedjobs: - case Data_data_jobs: + case XceData_data_schedjobs: + case XceData_data_jobs: data1.data.jobs->n_total = data->data.jobs->n_total; data1.data.jobs->pos = pos; data1.data.jobs->count = nc; @@ -1466,8 +1466,8 @@ int xce_push_data_req(endpoint_t* ep, xce_peerinfo_t* from, xce_data_t* data, data1.data.jobs->jobs = &(data->data.jobs->jobs[pos]); } break; - case Data_data_schedworkers: - case Data_data_workers: + case XceData_data_schedworkers: + case XceData_data_workers: data1.data.workers->n_total = data->data.workers->n_total; data1.data.workers->pos = pos; data1.data.workers->count = nc; @@ -1478,9 +1478,9 @@ int xce_push_data_req(endpoint_t* ep, xce_peerinfo_t* from, xce_data_t* data, = &(data->data.workers->workers[pos]); } break; - case Data_data_resources: + case XceData_data_resources: break; - case Data_data_resvals: + case XceData_data_resvals: break; default: break; @@ -1497,12 +1497,12 @@ int xce_push_data_req(endpoint_t* ep, xce_peerinfo_t* from, xce_data_t* data, } while ((pos < nrows) && (checkstatus >= 0)); switch (data->table) { - case Data_data_schedjobs: - case Data_data_jobs: + case XceData_data_schedjobs: + case XceData_data_jobs: FREEUP(data1.data.jobs); break; - case Data_data_schedworkers: - case Data_data_workers: + case XceData_data_schedworkers: + case XceData_data_workers: FREEUP(data1.data.workers); break; default: @@ -1555,24 +1555,24 @@ int xce_reply_get(endpoint_t* ep, xce_peerinfo_t* from, xce_reply_t* reply, } switch (getdata.table) { - case Data_data_schedjobs: - case Data_data_jobs: + case XceData_data_schedjobs: + case XceData_data_jobs: nrows = data->data.jobs->n_total; getdata.data.jobs = (xce_jobs_data_t*)calloc(1, sizeof(xce_jobs_data_t)); break; - case Data_data_schedworkers: - case Data_data_workers: + case XceData_data_schedworkers: + case XceData_data_workers: nrows = data->data.workers->n_total; getdata.data.workers = (xce_workers_data_t*)calloc(1, sizeof(xce_workers_data_t)); break; - case Data_data_resources: + case XceData_data_resources: nrows = data->data.resources->n_total; getdata.data.resources = (xce_resource_data_t*)calloc(1, sizeof(xce_resource_data_t)); break; - case Data_data_resvals: + case XceData_data_resvals: nrows = data->data.resvals->n_total; getdata.data.resvals = (xce_resval_data_t*)calloc(1, sizeof(xce_resval_data_t)); @@ -1607,8 +1607,8 @@ int xce_reply_get(endpoint_t* ep, xce_peerinfo_t* from, xce_reply_t* reply, } switch (getdata.table) { - case Data_data_schedjobs: - case Data_data_jobs: + case XceData_data_schedjobs: + case XceData_data_jobs: getdata.data.jobs->n_total = data->data.jobs->n_total; getdata.data.jobs->pos = pos; getdata.data.jobs->count = nc; @@ -1618,8 +1618,8 @@ int xce_reply_get(endpoint_t* ep, xce_peerinfo_t* from, xce_reply_t* reply, getdata.data.jobs->jobs = &(data->data.jobs->jobs[pos]); } break; - case Data_data_schedworkers: - case Data_data_workers: + case XceData_data_schedworkers: + case XceData_data_workers: getdata.data.workers->n_total = data->data.workers->n_total; getdata.data.workers->pos = pos; getdata.data.workers->count = nc; @@ -1630,7 +1630,7 @@ int xce_reply_get(endpoint_t* ep, xce_peerinfo_t* from, xce_reply_t* reply, = &(data->data.workers->workers[pos]); } break; - case Data_data_resources: + case XceData_data_resources: getdata.data.resources->n_total = data->data.resources->n_total; getdata.data.resources->pos = pos; getdata.data.workers->count = nc; @@ -1641,7 +1641,7 @@ int xce_reply_get(endpoint_t* ep, xce_peerinfo_t* from, xce_reply_t* reply, = &(data->data.resources->resources[pos]); } break; - case Data_data_resvals: + case XceData_data_resvals: getdata.data.resvals->n_total = data->data.resvals->n_total; getdata.data.resvals->pos = pos; getdata.data.resvals->count = nc; @@ -1665,18 +1665,18 @@ int xce_reply_get(endpoint_t* ep, xce_peerinfo_t* from, xce_reply_t* reply, } while ((pos < nrows) && (checkstatus >= 0)); switch (getdata.table) { - case Data_data_schedjobs: - case Data_data_jobs: + case XceData_data_schedjobs: + case XceData_data_jobs: FREEUP(getdata.data.jobs); break; - case Data_data_schedworkers: - case Data_data_workers: + case XceData_data_schedworkers: + case XceData_data_workers: FREEUP(getdata.data.workers); break; - case Data_data_resources: + case XceData_data_resources: FREEUP(getdata.data.resources); break; - case Data_data_resvals: + case XceData_data_resvals: FREEUP(getdata.data.resvals); break; default: @@ -1696,8 +1696,8 @@ int extract_data(dataset_t* dataset, xce_data_t* getdata) TRACE("%s ...", __func__); switch (getdata->table) { - case Data_data_schedjobs: - case Data_data_jobs: + case XceData_data_schedjobs: + case XceData_data_jobs: if (getdata->data.jobs->count <= 0) { return 1; } @@ -1716,8 +1716,8 @@ int extract_data(dataset_t* dataset, xce_data_t* getdata) memset(getdata->data.jobs->jobs, 0x0, count * sizeof(xce_job_t*)); break; - case Data_data_schedworkers: - case Data_data_workers: + case XceData_data_schedworkers: + case XceData_data_workers: if (getdata->data.workers->count <= 0) { return 1; } @@ -1737,7 +1737,7 @@ int extract_data(dataset_t* dataset, xce_data_t* getdata) count * sizeof(xce_worker_t*)); break; - case Data_data_resources: + case XceData_data_resources: if (getdata->data.resources->count <= 0) { return 1; } @@ -1758,7 +1758,7 @@ int extract_data(dataset_t* dataset, xce_data_t* getdata) memset(getdata->data.resources->resources, 0x0, count * sizeof(xce_resource_t*)); break; - case Data_data_resvals: + case XceData_data_resvals: if (getdata->data.resvals->count <= 0) { return 1; } @@ -1778,7 +1778,7 @@ int extract_data(dataset_t* dataset, xce_data_t* getdata) memset(getdata->data.resvals->vals, 0x0, count * sizeof(xce_resval_t*)); break; - case Data_data_jobArray: + case XceData_data_jobArray: /* job-array-single-wal-record task 5.3b: single, unchunked JobSubmit * descriptor (base jobid, array_size, one shared spec) -- unlike the * LIST-shaped cases above, no pagination/accumulation loop is needed: diff --git a/src/lib/testsuite/test_request.c b/src/lib/testsuite/test_request.c index abd974f0..c44d6f77 100644 --- a/src/lib/testsuite/test_request.c +++ b/src/lib/testsuite/test_request.c @@ -177,11 +177,11 @@ static void test_init_reply_by_req_zeroes_reply(void) } /* ── job-array-single-wal-record task 5.3b: extract_data() wire dispatch for - * Data_data_jobArray ───────────────────────────────────────────────────── + * XceData_data_jobArray ───────────────────────────────────────────────────── * extract_data() is the shared inbound-push reassembly step every receiver * (scheduler, API server) passes through inside vtask_entry() (xcectx.c) * before its own handler ever runs. Its switch had no case for - * Data_data_jobArray -- it fell into `default: rc = -EINVAL`, and + * XceData_data_jobArray -- it fell into `default: rc = -EINVAL`, and * vtask_entry()'s `if (callnext)` treats ANY nonzero return (including a * negative errno) as "reassembly complete, proceed" -- so the real payload * was silently freed and the receiver was handed an empty, zeroed @@ -220,7 +220,7 @@ static void test_extract_data_job_array_passes_through_intact(void) job_array.submit = &submit; memset(&data, 0, sizeof(data)); - data.table = Data_data_jobArray; + data.table = XceData_data_jobArray; data.data.job_array = &job_array; memset(&request, 0, sizeof(request)); @@ -236,7 +236,7 @@ static void test_extract_data_job_array_passes_through_intact(void) TEST_ASSERT_EQUAL_INT(0, xce_decode(pkt, &decoded)); /* consumes/frees pkt */ TEST_ASSERT_EQUAL_INT(Request_body_pushmsg, decoded.body_which); TEST_ASSERT_NOT_NULL(decoded.body.pushmsg); - TEST_ASSERT_EQUAL_INT(Data_data_jobArray, decoded.body.pushmsg->table); + TEST_ASSERT_EQUAL_INT(XceData_data_jobArray, decoded.body.pushmsg->table); /* Act: this is exactly what vtask_entry() calls on every inbound push. */ memset(&dataset, 0, sizeof(dataset)); @@ -287,6 +287,6 @@ void test_request(void) RUN_TEST(test_ha_heartbeat_req_roundtrip_fields); RUN_TEST(test_nop_req_remains_bodyless); - printf("\n>> extract_data() Data_data_jobArray wire dispatch tests\n\n"); + printf("\n>> extract_data() XceData_data_jobArray wire dispatch tests\n\n"); RUN_TEST(test_extract_data_job_array_passes_through_intact); } diff --git a/src/lib/xceapi.c b/src/lib/xceapi.c index e9858420..1fcf7940 100644 --- a/src/lib/xceapi.c +++ b/src/lib/xceapi.c @@ -153,7 +153,7 @@ int xce_get(xce_ctx_t* ctx, int table, void*** data) int xce_reslist(xce_ctx_t* ctx, xce_resource_t*** res) { - int table = Data_data_resources; + int table = XceData_data_resources; return xce_get(ctx, table, (void***)res); } @@ -174,11 +174,11 @@ int xce_resval(xce_ctx_t* ctx, char* host, char* res, xce_resval_t*** vals) } }; xce_get_t get = { - .table = Data_data_resvals, + .table = XceData_data_resvals, .condition = &condition }; dataset_t getreply = { - .table = Data_data_resvals, + .table = XceData_data_resvals, .nrows = 0, .rows = NULL }; @@ -244,7 +244,7 @@ int xce_join(xce_ctx_t* ctx, int role, hostattrs_t* attrs) int xce_workerlist(xce_ctx_t* ctx, xce_worker_t*** workers) { - int table = Data_data_workers; + int table = XceData_data_workers; return xce_get(ctx, table, (void***)workers); } @@ -326,9 +326,9 @@ int xce_joblist_ex(xce_ctx_t* ctx, int sched, xce_job_t*** jobs) int table; if (sched) { - table = Data_data_schedjobs; + table = XceData_data_schedjobs; } else { - table = Data_data_jobs; + table = XceData_data_jobs; } return xce_get(ctx, table, (void***)jobs); @@ -714,7 +714,7 @@ int xce_jobinfo(xce_ctx_t* ctx, uint64_t jobid, xce_job_t** job) } }; xce_get_t get = { - .table = Data_data_jobs, + .table = XceData_data_jobs, .condition = &condition }; /* clang-format on */ diff --git a/src/scheduler/src/handler.c b/src/scheduler/src/handler.c index a701b945..7a598da8 100644 --- a/src/scheduler/src/handler.c +++ b/src/scheduler/src/handler.c @@ -41,7 +41,7 @@ static int sched_on_jobs(xce_scheduler_t* sched, int njobs, xce_job_t** jobs) /* job-array-single-wal-record task 5.3 (design.md "Compact Scheduler/ * API-Server Push", File Structure Plan handler.c bullet; Req 6.3, 6.4, 6.5, - * 8.4): sibling case to Data_data_schedjobs (sched_handle_dataset, below) + * 8.4): sibling case to XceData_data_schedjobs (sched_handle_dataset, below) * decoding the compact Data.jobArray descriptor -- task 5.1's * xcesched_push_jobarray() wire shape, one JobSubmit{jobid, arraySize, * submit} carrying the array's base jobid, element count and ONE shared job @@ -247,7 +247,7 @@ static int sched_on_job_array(xce_scheduler_t* sched, free(jobs); } /* On success sched_on_jobs()/the pend-update killer now owns jobs[] and - * every element in it (mirrors the existing Data_data_schedjobs case + * every element in it (mirrors the existing XceData_data_schedjobs case * below, which hands dataset->rows to sched_on_jobs the same way). */ return rc; @@ -325,7 +325,7 @@ static int sched_handle_dataset(xce_scheduler_t* sched, dataset_t* dataset) TRACE("%s table = %s", __func__, xce_dataset_table(dataset->table)); switch (dataset->table) { - case Data_data_schedjobs: + case XceData_data_schedjobs: rc = sched_on_jobs(sched, dataset->nrows, (xce_job_t**)dataset->rows); if (rc >= 0) { /* jobs will be put in pend update request queue */ @@ -334,7 +334,7 @@ static int sched_handle_dataset(xce_scheduler_t* sched, dataset_t* dataset) } break; - case Data_data_schedworkers: + case XceData_data_schedworkers: rc = sched_on_workers(sched, dataset->nrows, (xce_worker_t**)dataset->rows); if (rc >= 0) { @@ -343,7 +343,7 @@ static int sched_handle_dataset(xce_scheduler_t* sched, dataset_t* dataset) dataset->rows = NULL; } break; - case Data_data_resources: + case XceData_data_resources: rc = sched_handle_resources(sched, dataset->nrows, (xce_resource_t**)dataset->rows); if (rc >= 0) { @@ -351,7 +351,7 @@ static int sched_handle_dataset(xce_scheduler_t* sched, dataset_t* dataset) dataset->rows = NULL; } break; - case Data_data_resvals: + case XceData_data_resvals: rc = sched_handle_resvals(sched, dataset->nrows, (xce_resval_t**)dataset->rows); if (rc >= 0) { @@ -359,7 +359,7 @@ static int sched_handle_dataset(xce_scheduler_t* sched, dataset_t* dataset) dataset->rows = NULL; } break; - case Data_data_jobArray: + case XceData_data_jobArray: rc = sched_on_job_array(sched, (xce_job_submit_t*)dataset->rows); /* Unlike the list-shaped cases above, we never alias dataset->rows * into the pend-update queue -- every array element is its own diff --git a/src/scheduler/src/schedobj.c b/src/scheduler/src/schedobj.c index c9d76195..8a610e75 100644 --- a/src/scheduler/src/schedobj.c +++ b/src/scheduler/src/schedobj.c @@ -518,7 +518,7 @@ int sched_update_existing_queue(xce_scheduler_t* sched, xce_queue_t* updated) /* * sched_update_existing_job — apply a notify single-job update to an existing - * job IN PLACE. The job must already exist (created via the Data_data_schedjobs + * job IN PLACE. The job must already exist (created via the XceData_data_schedjobs * dataset path); this function never creates a job. * * - NULL payload → -EINVAL diff --git a/src/scheduler/testsuite/test_handlers.c b/src/scheduler/testsuite/test_handlers.c index 2287bc1b..35c7ae8b 100644 --- a/src/scheduler/testsuite/test_handlers.c +++ b/src/scheduler/testsuite/test_handlers.c @@ -23,7 +23,7 @@ * scheduler-internal externs handler.c references but this file does not * exercise are stubbed below: sched_init_config (config-init request path) * and sched_get_job (job-array-single-wal-record task 5.3's per-element - * dedup check, Data_data_jobArray — not tested here, see + * dedup check, XceData_data_jobArray — not tested here, see * test_job_array_expand.c for the real kvmgr-backed sched_get_job). */ int sched_init_config(xce_scheduler_t* s, xce_conf4sched_t* c) { diff --git a/src/scheduler/testsuite/test_job_array_expand.c b/src/scheduler/testsuite/test_job_array_expand.c index 888fca14..7ca66c51 100644 --- a/src/scheduler/testsuite/test_job_array_expand.c +++ b/src/scheduler/testsuite/test_job_array_expand.c @@ -89,7 +89,7 @@ static void seed_existing_job(xce_scheduler_t* sched, uint64_t jobid) /* Builds a heap-allocated, heap-owned-strings (jobArray, submit) pair -- the * same shape decode_JobSubmit_ptr() produces on the real wire-decode path * (controller.capnp.c), not a stack/string-literal stand-in. Task 7.3 made - * sched_handle_dataset()'s Data_data_jobArray case free_JobSubmit_ptr() its + * sched_handle_dataset()'s XceData_data_jobArray case free_JobSubmit_ptr() its * dataset->rows unconditionally, so any fixture routed through that handler * must be genuinely heap-owned or the free segfaults on stack memory / * string literals. Caller owns the returned pointer until it hands it to @@ -154,7 +154,7 @@ static void test_job_array_push_expands_into_n_tasks(void) ja = init_job_array(5000, 3); memset(&dataset, 0, sizeof(dataset)); - dataset.table = Data_data_jobArray; + dataset.table = XceData_data_jobArray; dataset.rows = (void*)ja; /* sched_handle_dataset() owns and frees dataset.rows on return (task @@ -195,7 +195,7 @@ static void test_job_array_redelivery_does_not_duplicate(void) fixture_init(&sched); memset(&dataset, 0, sizeof(dataset)); - dataset.table = Data_data_jobArray; + dataset.table = XceData_data_jobArray; dataset.rows = (void*)init_job_array(6000, 2); /* first delivery: 2 new tasks enqueued. sched_handle_dataset() owns and @@ -309,7 +309,7 @@ static void test_wire_dispatch_chain_delivers_n_tasks(void) job_array.submit = &submit; memset(&data, 0, sizeof(data)); - data.table = Data_data_jobArray; + data.table = XceData_data_jobArray; data.data.job_array = &job_array; memset(&request, 0, sizeof(request)); diff --git a/src/scheduler/testsuite/test_job_array_wire_e2e.c b/src/scheduler/testsuite/test_job_array_wire_e2e.c index a24c9e7d..1867f2cb 100644 --- a/src/scheduler/testsuite/test_job_array_wire_e2e.c +++ b/src/scheduler/testsuite/test_job_array_wire_e2e.c @@ -49,7 +49,7 @@ * it: via xce_decode()'s output, never a hand-built dataset_t. * * Bug-catching demonstration (this task's "RED phase"): temporarily - * reverting extract_data()'s `case Data_data_jobArray:` (task 5.3b's fix) + * reverting extract_data()'s `case XceData_data_jobArray:` (task 5.3b's fix) * back to falling through to `default: rc = -EINVAL` makes this test's first * assertion (`callnext != 0` while dataset->rows stays NULL) fail loudly, * and the subsequent job-count assertions fail too (zero tasks created) -- @@ -244,7 +244,7 @@ static void build_encoded_push(io_packet_t** out_pkt, uint64_t base_jobid, job_array.submit = submit_storage; memset(&data, 0, sizeof(data)); - data.table = Data_data_jobArray; + data.table = XceData_data_jobArray; data.data.job_array = &job_array; memset(&request, 0, sizeof(request)); diff --git a/src/schema/controller.capnp b/src/schema/controller.capnp index 0e0a39bf..5423523a 100644 --- a/src/schema/controller.capnp +++ b/src/schema/controller.capnp @@ -352,7 +352,7 @@ struct ResourceVals $C.mapname("xce_resval_data_t") { vals @2: List(ResourceVal) $C.mapname("vals") $C.maplistcount("count"); } -struct Data $C.mapname("xce_data_t"){ +struct XceData $C.mapname("xce_data_t"){ data :union $C.mapuniontag("table") { schedjobs @0: SchedJobData $C.mapname("jobs"); schedworkers @1: WorkerData $C.mapname("workers"); @@ -578,7 +578,7 @@ struct RequestReply $C.mapname("xce_reply_t"){ data : union { nodata @3: Void; debugmsg @4: Text; - getdata @5: Data; + getdata @5: XceData; jobid @6: UInt64; clusterconf @7: ClusterConf; replconfigreplymsg @8: ReplConfigReply; @@ -635,7 +635,7 @@ struct Request $C.mapname("xce_request_t"){ hellomsg @5: Text; joinmsg @6: JoinData; configpushmsg @7: ConfigData; - pushmsg @8: Data; + pushmsg @8: XceData; getmsg @9: Get; datamsg @10: Dataset; dispatchmsg @11: JobDispatch;