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

feat(job-array-one-based-index): guard scheduler array expansion (task 3.1)

Browse source 
The scheduler expands an array push only when the first ID carries the
array marking and the size fits in its block (size <= 32768 - idx, no
addition); otherwise it logs an ERROR naming the array identity and
expands nothing. Pre-upgrade arrays starting at index 0 still expand.
The INFO log names the array identity (array_jobid=...). This also
closes a crash: a malformed push with a huge size overflowed calloc.
Expansion tests move to tagged block|1 first IDs.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
This commit is contained in:
Rongsong Shen 2026-10-01 10:24:42 +00:00
parent fe5ddc9854
commit 4f44a68973
Signed by untrusted user: rshen
GPG key ID: 80533536B0EB651C
5 changed files with 147 additions and 31 deletions
Download patch file Download diff file Expand all files Collapse all files

2
.kiro/specs/job-array-one-based-index/tasks.md
View file

@ -50,7 +50,7 @@
- _Requirements: 7.7, 7.10, 7.11_
- [ ] 3. Scheduler and worker
- [ ] 3.1 (P) Guard scheduler array expansion and log the array identity
- [x] 3.1 (P) Guard scheduler array expansion and log the array identity
- Expand an array push only when the first ID carries the array marking and its size fits within the block (checked without addition); otherwise log an ERROR naming the array identity and size and expand nothing
- Log the received array by its array identity
- Move the existing expansion tests (translation, array-expand and wire end-to-end tests) to tagged block-plus-one first IDs

20
src/scheduler/src/driver.c
View file

@ -466,6 +466,21 @@ static int translate_one_pending_update(xce_sched_core_t* core,
base_jobid = job_array->jobid;
array_size = job_array->array_size;
/* SchedGuard (job-array-one-based-index, Req 5.9): expand only a
* tagged first ID whose elements all fit in its 32768-index block.
* The size check subtracts instead of adding, so a UInt64 size
* cannot wrap. Index 0 (pre-upgrade arrays) is accepted. The
* controller never sends anything else; defensive only. */
if (!is_array_tag(base_jobid)
|| (array_size > (32768 - array_idx(base_jobid)))) {
fmt_jobid(jobidString, sizeof(jobidString) - 1,
array_jobid(base_jobid));
ERROR("%s UPDATE_JOB_ARRAY: not a well-formed array, not "
"expanded: array_jobid=%s array_size=%llu",
__func__, jobidString, (unsigned long long)array_size);
return 0;
}
jobs = (xce_job_t**)calloc((size_t)array_size, sizeof(xce_job_t*));
if (jobs == NULL) {
ERROR("%s UPDATE_JOB_ARRAY: out of memory allocating a "
@ -506,8 +521,9 @@ static int translate_one_pending_update(xce_sched_core_t* core,
return 0;
}
fmt_jobid(jobidString, sizeof(jobidString) - 1, base_jobid);
INFO("job array push received and expanded: base_jobid=%s "
fmt_jobid(jobidString, sizeof(jobidString) - 1,
array_jobid(base_jobid));
INFO("job array push received and expanded: array_jobid=%s "
"array_size=%llu new_tasks=%d",
jobidString, (unsigned long long)array_size, n_new);

122
src/scheduler/testsuite/test_driver_translate.c
View file

@ -409,6 +409,13 @@ static xce_job_t* make_wire_job(uint64_t jobid, uint32_t state,
return j;
}
/* job-array-one-based-index (SchedGuard, Req 5.9): a real array's first
* element ID carries TAG_JOB_ARRAY and sits in a 32768-aligned block of the
* sequence; element k of block `blk` is ARR_ID(blk, k). */
#define ARR_ID(blk, k) \
MAKE_JOBID(1, 2, 1000, TAG_JOB_ARRAY, \
(((uint64_t)(blk)) << N_ARRAY_IDX_BITS) | (uint64_t)(k))
/* Compact job-array descriptor, same shape init_job_array() (test_job_
* array_expand.c) builds -- heap-owned strings, matching what a real
* XCE_SCHED_OP_UPDATE_JOB_ARRAY request now carries unexpanded (connection-
@ -720,7 +727,7 @@ static void test_translate_job_rusage_null_rusages_skips(void)
static void test_translate_job_array_expands_into_n_events(void)
{
xce_sched_core_t core;
xce_job_submit_t* ja = make_wire_job_array(9000, 3);
xce_job_submit_t* ja = make_wire_job_array(ARR_ID(1, 1), 3);
xce_sched_update_request_t* req = make_req(XCE_SCHED_OP_UPDATE_JOB_ARRAY);
xce_core_event_t ev;
int rc;
@ -734,9 +741,9 @@ static void test_translate_job_array_expands_into_n_events(void)
TEST_ASSERT_EQUAL_INT(XCE_CORE_EVENT_JOBS_PUSH, ev.type);
TEST_ASSERT_EQUAL_UINT64(42, ev.ts);
TEST_ASSERT_EQUAL_INT(3, ev.u.jobs_push.njobs);
TEST_ASSERT_EQUAL_UINT64(9000, ev.u.jobs_push.jobs[0]->jobid);
TEST_ASSERT_EQUAL_UINT64(9001, ev.u.jobs_push.jobs[1]->jobid);
TEST_ASSERT_EQUAL_UINT64(9002, ev.u.jobs_push.jobs[2]->jobid);
TEST_ASSERT_EQUAL_UINT64(ARR_ID(1, 1), ev.u.jobs_push.jobs[0]->jobid);
TEST_ASSERT_EQUAL_UINT64(ARR_ID(1, 2), ev.u.jobs_push.jobs[1]->jobid);
TEST_ASSERT_EQUAL_UINT64(ARR_ID(1, 3), ev.u.jobs_push.jobs[2]->jobid);
TEST_ASSERT_EQUAL_STRING("normal", ev.u.jobs_push.jobs[0]->submit->queue);
TEST_ASSERT_EQUAL_STRING("normal", ev.u.jobs_push.jobs[2]->submit->queue);
/* every element gets its OWN independent clone, not a shared pointer --
@ -763,7 +770,7 @@ static void test_translate_job_array_expands_into_n_events(void)
static void test_translate_job_array_skips_already_existing_elements(void)
{
xce_sched_core_t core;
xce_job_submit_t* ja = make_wire_job_array(9100, 3);
xce_job_submit_t* ja = make_wire_job_array(ARR_ID(2, 1), 3);
xce_sched_update_request_t* req = make_req(XCE_SCHED_OP_UPDATE_JOB_ARRAY);
xce_core_event_t ev;
xce_sched_job_t existing0;
@ -774,11 +781,11 @@ static void test_translate_job_array_skips_already_existing_elements(void)
req->u.jobarrayupdate.job_array = ja;
memset(&existing0, 0, sizeof(existing0));
existing0.jobid = 9100;
existing0.jobid = ARR_ID(2, 1);
sched_add_job(&core, &existing0);
memset(&existing1, 0, sizeof(existing1));
existing1.jobid = 9101;
existing1.jobid = ARR_ID(2, 2);
sched_add_job(&core, &existing1);
rc = translate_one_pending_update(&core, req, 1, &ev);
@ -786,7 +793,7 @@ static void test_translate_job_array_skips_already_existing_elements(void)
TEST_ASSERT_EQUAL_INT(1, rc);
TEST_ASSERT_EQUAL_INT(XCE_CORE_EVENT_JOBS_PUSH, ev.type);
TEST_ASSERT_EQUAL_INT(1, ev.u.jobs_push.njobs);
TEST_ASSERT_EQUAL_UINT64(9102, ev.u.jobs_push.jobs[0]->jobid);
TEST_ASSERT_EQUAL_UINT64(ARR_ID(2, 3), ev.u.jobs_push.jobs[0]->jobid);
}
/* Every index already exists: a fully-duplicate redelivery -- nothing new
@ -795,7 +802,7 @@ static void test_translate_job_array_skips_already_existing_elements(void)
static void test_translate_job_array_all_duplicates_returns_zero_noop(void)
{
xce_sched_core_t core;
xce_job_submit_t* ja = make_wire_job_array(9200, 2);
xce_job_submit_t* ja = make_wire_job_array(ARR_ID(3, 1), 2);
xce_sched_update_request_t* req = make_req(XCE_SCHED_OP_UPDATE_JOB_ARRAY);
xce_core_event_t ev;
xce_sched_job_t existing0;
@ -806,11 +813,11 @@ static void test_translate_job_array_all_duplicates_returns_zero_noop(void)
req->u.jobarrayupdate.job_array = ja;
memset(&existing0, 0, sizeof(existing0));
existing0.jobid = 9200;
existing0.jobid = ARR_ID(3, 1);
sched_add_job(&core, &existing0);
memset(&existing1, 0, sizeof(existing1));
existing1.jobid = 9201;
existing1.jobid = ARR_ID(3, 2);
sched_add_job(&core, &existing1);
rc = translate_one_pending_update(&core, req, 1, &ev);
@ -820,6 +827,81 @@ static void test_translate_job_array_all_duplicates_returns_zero_noop(void)
TEST_ASSERT_NULL(req->u.jobarrayupdate.expanded_jobs);
}
/* SchedGuard (Req 5.9): pushes that are not a well-formed array are logged
* at ERROR and expand nothing. expect_not_expanded() runs one through
* translate and asserts nothing was produced. */
static void expect_not_expanded(uint64_t first, uint64_t array_size)
{
xce_sched_core_t core;
xce_job_submit_t* ja = make_wire_job_array(first, array_size);
xce_sched_update_request_t* req = make_req(XCE_SCHED_OP_UPDATE_JOB_ARRAY);
xce_core_event_t ev;
int rc;
fixture_init(&core);
req->u.jobarrayupdate.job_array = ja;
rc = translate_one_pending_update(&core, req, 1, &ev);
TEST_ASSERT_EQUAL_INT(0, rc);
TEST_ASSERT_EQUAL_INT(0, req->u.jobarrayupdate.expanded_njobs);
TEST_ASSERT_NULL(req->u.jobarrayupdate.expanded_jobs);
TEST_ASSERT_NULL(sched_get_job(&core, first));
}
static void test_translate_job_array_untagged_first_not_expanded(void)
{
expect_not_expanded(9000, 3);
}
static void test_translate_job_array_overflowing_block_not_expanded(void)
{
/* index 32766 + 3 elements would run past 32767 into the next block */
expect_not_expanded(ARR_ID(5, 32766), 3);
/* index 1 + 32768 elements: one past the block */
expect_not_expanded(ARR_ID(5, 1), 32768);
}
static void test_translate_job_array_max_u64_size_not_expanded(void)
{
/* idx + size would wrap to 1 with an addition-based check */
expect_not_expanded(ARR_ID(6, 2), UINT64_MAX);
}
/* Boundary: the last two indices of the block fit exactly. */
static void test_translate_job_array_block_end_expands(void)
{
xce_sched_core_t core;
xce_job_submit_t* ja = make_wire_job_array(ARR_ID(7, 32766), 2);
xce_sched_update_request_t* req = make_req(XCE_SCHED_OP_UPDATE_JOB_ARRAY);
xce_core_event_t ev;
fixture_init(&core);
req->u.jobarrayupdate.job_array = ja;
TEST_ASSERT_EQUAL_INT(1, translate_one_pending_update(&core, req, 1, &ev));
TEST_ASSERT_EQUAL_INT(2, ev.u.jobs_push.njobs);
TEST_ASSERT_EQUAL_UINT64(ARR_ID(7, 32766), ev.u.jobs_push.jobs[0]->jobid);
TEST_ASSERT_EQUAL_UINT64(ARR_ID(7, 32767), ev.u.jobs_push.jobs[1]->jobid);
}
/* A pre-upgrade array (tagged, numbered from 0) still expands to 0..n-1. */
static void test_translate_job_array_pre_upgrade_index_zero_expands(void)
{
xce_sched_core_t core;
xce_job_submit_t* ja = make_wire_job_array(ARR_ID(8, 0), 3);
xce_sched_update_request_t* req = make_req(XCE_SCHED_OP_UPDATE_JOB_ARRAY);
xce_core_event_t ev;
fixture_init(&core);
req->u.jobarrayupdate.job_array = ja;
TEST_ASSERT_EQUAL_INT(1, translate_one_pending_update(&core, req, 1, &ev));
TEST_ASSERT_EQUAL_INT(3, ev.u.jobs_push.njobs);
TEST_ASSERT_EQUAL_UINT64(ARR_ID(8, 0), ev.u.jobs_push.jobs[0]->jobid);
TEST_ASSERT_EQUAL_UINT64(ARR_ID(8, 2), ev.u.jobs_push.jobs[2]->jobid);
}
/* A never-decoded/malformed descriptor (NULL job_array, the enqueue side
* should never produce this, guarded defensively here anyway, same
* discipline every other case in this function follows) skips cleanly. */
@ -1078,18 +1160,19 @@ static void test_drain_translates_and_pushes_job_array_op(void)
fixture_init(&core);
ja = make_wire_job_array(9300, 3);
req = make_req(XCE_SCHED_OP_UPDATE_JOB_ARRAY);
ja = make_wire_job_array(ARR_ID(4, 1), 3);
req = make_req(XCE_SCHED_OP_UPDATE_JOB_ARRAY);
req->u.jobarrayupdate.job_array = ja;
enqueue(&core, req);
rc = xce_sched_driver_translate_pending_events(&core, NULL);
TEST_ASSERT_EQUAL_INT(1, rc);
TEST_ASSERT_NOT_NULL(sched_get_job(&core, 9300));
TEST_ASSERT_NOT_NULL(sched_get_job(&core, 9301));
TEST_ASSERT_NOT_NULL(sched_get_job(&core, 9302));
TEST_ASSERT_EQUAL_UINT32(JOB_STAT_PEND, sched_get_job(&core, 9300)->state);
TEST_ASSERT_NOT_NULL(sched_get_job(&core, ARR_ID(4, 1)));
TEST_ASSERT_NOT_NULL(sched_get_job(&core, ARR_ID(4, 2)));
TEST_ASSERT_NOT_NULL(sched_get_job(&core, ARR_ID(4, 3)));
TEST_ASSERT_EQUAL_UINT32(JOB_STAT_PEND,
sched_get_job(&core, ARR_ID(4, 1))->state);
}
/* ---------------------------------------------------------------------- */
@ -1174,6 +1257,11 @@ int main(void)
RUN_TEST(test_translate_job_array_skips_already_existing_elements);
RUN_TEST(test_translate_job_array_all_duplicates_returns_zero_noop);
RUN_TEST(test_translate_job_array_null_payload_returns_zero);
RUN_TEST(test_translate_job_array_untagged_first_not_expanded);
RUN_TEST(test_translate_job_array_overflowing_block_not_expanded);
RUN_TEST(test_translate_job_array_max_u64_size_not_expanded);
RUN_TEST(test_translate_job_array_block_end_expands);
RUN_TEST(test_translate_job_array_pre_upgrade_index_zero_expands);
RUN_TEST(test_translate_unknown_op_skips);
RUN_TEST(test_drain_translates_and_pushes_all_ten_op_types);
RUN_TEST(test_drain_translates_and_pushes_job_array_op);

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

@ -37,6 +37,12 @@
* elsewhere is sched_init_config, stubbed below (same as test_handlers.c). */
#include "../src/handler.c"
/* job-array-one-based-index (Req 5.9): array first IDs are tagged
* TAG_JOB_ARRAY, block-aligned, numbered from 1. */
#define ARR_FIRST(blk) \
MAKE_JOBID(1, 2, 1000, TAG_JOB_ARRAY, \
(((uint64_t)(blk)) << N_ARRAY_IDX_BITS) | 1)
int sched_init_config(xce_scheduler_t* s, xce_conf4sched_t* c)
{
(void)s;
@ -179,7 +185,7 @@ static void test_job_array_push_enqueues_raw_descriptor_unexpanded(void)
xce_job_submit_t* popped;
fixture_init(&sched);
ja = init_job_array(5000, 3);
ja = init_job_array(ARR_FIRST(1), 3);
memset(&dataset, 0, sizeof(dataset));
dataset.table = XceData_data_jobArray;
@ -193,7 +199,7 @@ static void test_job_array_push_enqueues_raw_descriptor_unexpanded(void)
popped = pop_job_array_update(&sched);
TEST_ASSERT_EQUAL_PTR(ja, popped); /* same descriptor, not a clone */
TEST_ASSERT_EQUAL_UINT64(5000, popped->jobid);
TEST_ASSERT_EQUAL_UINT64(ARR_FIRST(1), popped->jobid);
TEST_ASSERT_EQUAL_UINT64(3, popped->array_size);
TEST_ASSERT_EQUAL_STRING("normal", popped->submit->queue);
@ -210,7 +216,7 @@ static void test_job_array_null_submit_returns_einval(void)
fixture_init(&sched);
memset(&ja, 0, sizeof(ja));
ja.jobid = 1;
ja.jobid = ARR_FIRST(2);
ja.array_size = 3;
ja.submit = NULL;
@ -280,7 +286,7 @@ static void test_wire_dispatch_chain_delivers_job_array_descriptor(void)
submit.resreq = "type == local";
memset(&job_array, 0, sizeof(job_array));
job_array.jobid = 8000;
job_array.jobid = ARR_FIRST(3);
job_array.array_size = 4;
job_array.submit = &submit;
@ -316,7 +322,7 @@ static void test_wire_dispatch_chain_delivers_job_array_descriptor(void)
* correctly addressed via base_jobid + array_size. */
popped = pop_job_array_update(&sched);
TEST_ASSERT_NOT_NULL(popped);
TEST_ASSERT_EQUAL_UINT64(8000, popped->jobid);
TEST_ASSERT_EQUAL_UINT64(ARR_FIRST(3), popped->jobid);
TEST_ASSERT_EQUAL_UINT64(4, popped->array_size);
TEST_ASSERT_EQUAL_STRING("normal", popped->submit->queue);
free_JobSubmit_ptr(&popped);

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

@ -80,6 +80,12 @@
* self-contained; sched_init_config is its only unresolved extern). */
#include "../src/handler.c"
/* job-array-one-based-index (Req 5.9): array first IDs are tagged
* TAG_JOB_ARRAY, block-aligned, numbered from 1. */
#define ARR_FIRST(blk) \
MAKE_JOBID(1, 2, 1000, TAG_JOB_ARRAY, \
(((uint64_t)(blk)) << N_ARRAY_IDX_BITS) | 1)
int sched_init_config(xce_scheduler_t* s, xce_conf4sched_t* c)
{
(void)s;
@ -292,14 +298,14 @@ static void test_one_push_enqueues_job_array_descriptor(void)
fixture_init(&sched);
build_encoded_push(&pkt, 42000, 6, &submit_storage);
build_encoded_push(&pkt, ARR_FIRST(1), 6, &submit_storage);
TEST_ASSERT_TRUE(
deliver_via_real_dispatch_chain(pkt, &sched, &ds) >= 0);
popped = pop_job_array_update(&sched);
TEST_ASSERT_NOT_NULL(popped);
TEST_ASSERT_EQUAL_UINT64(42000, popped->jobid);
TEST_ASSERT_EQUAL_UINT64(ARR_FIRST(1), popped->jobid);
TEST_ASSERT_EQUAL_UINT64(6, popped->array_size);
TEST_ASSERT_EQUAL_STRING("normal", popped->submit->queue);
free_JobSubmit_ptr(&popped);
@ -335,14 +341,14 @@ static void test_redelivery_enqueues_independent_second_descriptor(void)
/* Build the wire bytes twice (two independent xce_encode() calls with
* identical inputs) so each delivery genuinely goes through its own
* encode and decode -- not one decode fed to the handler twice. */
build_encoded_push(&pkt1, 43000, 4, &submit_storage);
build_encoded_push(&pkt2, 43000, 4, &submit_storage);
build_encoded_push(&pkt1, ARR_FIRST(2), 4, &submit_storage);
build_encoded_push(&pkt2, ARR_FIRST(2), 4, &submit_storage);
/* First delivery. */
TEST_ASSERT_TRUE(deliver_via_real_dispatch_chain(pkt1, &sched, &ds1) >= 0);
popped1 = pop_job_array_update(&sched);
TEST_ASSERT_NOT_NULL(popped1);
TEST_ASSERT_EQUAL_UINT64(43000, popped1->jobid);
TEST_ASSERT_EQUAL_UINT64(ARR_FIRST(2), popped1->jobid);
TEST_ASSERT_EQUAL_UINT64(4, popped1->array_size);
free_JobSubmit_ptr(&popped1);
free_JobSubmit_ptr((xce_job_submit_t**)&ds1->rows);
@ -356,7 +362,7 @@ static void test_redelivery_enqueues_independent_second_descriptor(void)
TEST_ASSERT_TRUE(deliver_via_real_dispatch_chain(pkt2, &sched, &ds2) >= 0);
popped2 = pop_job_array_update(&sched);
TEST_ASSERT_NOT_NULL(popped2);
TEST_ASSERT_EQUAL_UINT64(43000, popped2->jobid);
TEST_ASSERT_EQUAL_UINT64(ARR_FIRST(2), popped2->jobid);
TEST_ASSERT_EQUAL_UINT64(4, popped2->array_size);
free_JobSubmit_ptr(&popped2);