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

fix(schema): rename struct Data -> XceData to unshadow capnp builtin blob type

Browse source 
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.
This commit is contained in:
Rongsong Shen 2026-07-22 09:22:04 +00:00
parent 7ed3a3941f
commit e53c086ec5
21 changed files with 144 additions and 144 deletions
Download patch file Download diff file Expand all files Collapse all files

10
src/controller/apiserver.c
View file

@ -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);

12
src/controller/opget.c
View file

@ -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:

10
src/controller/sched.c
View file

@ -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);

8
src/controller/testsuite/test.c
View file

@ -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:

26
src/controller/testsuite/test_e2e_capnp.c
View file

@ -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");

10
src/controller/testsuite/test_job_array_apiserver_push.c
View file

@ -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)");

10
src/controller/testsuite/test_job_array_sched_push.c
View file

@ -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)");

8
src/frontend/apiserver/backend/clib.go
View file

@ -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

4
src/frontend/apiserver/backend/handler.go
View file

@ -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.

2
src/frontend/apiserver/backend/handler_test.go
View file

@ -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

4
src/frontend/apiserver/backend/wire_dispatch_support.go
View file

@ -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

24
src/lib/helper.c
View file

@ -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;

100
src/lib/request.c
View file

@ -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:

10
src/lib/testsuite/test_request.c
View file

@ -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);
}

14
src/lib/xceapi.c
View file

@ -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 */

14
src/scheduler/src/handler.c
View file

@ -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

2
src/scheduler/src/schedobj.c
View file

@ -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

2
src/scheduler/testsuite/test_handlers.c
View file

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

8
src/scheduler/testsuite/test_job_array_expand.c
View file

@ -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));

4
src/scheduler/testsuite/test_job_array_wire_e2e.c
View file

@ -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));

6
src/schema/controller.capnp
View file

@ -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;