diff --git a/.kiro/specs/job-array-one-based-index/tasks.md b/.kiro/specs/job-array-one-based-index/tasks.md index ceebe64b..adf98fc9 100644 --- a/.kiro/specs/job-array-one-based-index/tasks.md +++ b/.kiro/specs/job-array-one-based-index/tasks.md @@ -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 diff --git a/src/scheduler/src/driver.c b/src/scheduler/src/driver.c index 737be1bb..6954842c 100644 --- a/src/scheduler/src/driver.c +++ b/src/scheduler/src/driver.c @@ -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); diff --git a/src/scheduler/testsuite/test_driver_translate.c b/src/scheduler/testsuite/test_driver_translate.c index d6ff12c9..29fd8027 100644 --- a/src/scheduler/testsuite/test_driver_translate.c +++ b/src/scheduler/testsuite/test_driver_translate.c @@ -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); diff --git a/src/scheduler/testsuite/test_job_array_expand.c b/src/scheduler/testsuite/test_job_array_expand.c index 3062c7c9..8f4e99d9 100644 --- a/src/scheduler/testsuite/test_job_array_expand.c +++ b/src/scheduler/testsuite/test_job_array_expand.c @@ -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); diff --git a/src/scheduler/testsuite/test_job_array_wire_e2e.c b/src/scheduler/testsuite/test_job_array_wire_e2e.c index 98d76dbb..c6b510eb 100644 --- a/src/scheduler/testsuite/test_job_array_wire_e2e.c +++ b/src/scheduler/testsuite/test_job_array_wire_e2e.c @@ -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);