feat(job-array-single-wal-record): scheduler-side local expansion of the compact push

Adds sched_on_job_array() (handler.c), a sibling case to
Data_data_schedjobs decoding the new Data.jobArray variant and
expanding it locally into N ordinary schedulable tasks via
base_jobid + i, routed through the existing unmodified per-job
pipeline (sched_on_jobs()). Each element gets its own deep clone
(clone_job_submit_data/clone_job_startinfo) of the shared submit
spec, since per-element teardown (free_Job_ptr) frees it
independently and a shared pointer would double-free.

Dedup reuses sched_get_job() against the same kvtab sched_add_job()
guards, rather than a new mechanism. A confirmation log line fires
once per successful expansion (skipped on a fully-duplicate
redelivery).

Adds a new task (5.3b) discovered during this task's review: the
shared wire dispatcher extract_data() (src/lib/request.c) has no case
for Data_data_jobArray and silently drops every real array push
before any receiver's handler runs -- outside every current task's
boundary, confirmed to affect the future API-server receive path
(task 5.4) identically. design.md's File Structure Plan corrected to
record this. User confirmed sequencing: fix it now, before 5.4.

Task 5.3 of job-array-single-wal-record (Requirements 6.3, 6.4, 6.5, 8.4).
This commit is contained in:
Rongsong Shen 2026-07-15 10:10:14 +00:00
parent e0cec75284
commit d16847185e
6 changed files with 526 additions and 6 deletions

View file

@ -1,6 +1,8 @@
#include <errno.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <time.h>
#include <unistd.h>
#include "controller.capnp.h"
@ -37,6 +39,220 @@ static int sched_on_jobs(xce_scheduler_t* sched, int njobs, xce_job_t** jobs)
return 0;
}
/* 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)
* 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
* spec -- and expanding it locally into N ordinary schedulable tasks via the
* same base_jobid + i arithmetic the controller's own restore_job_array()
* (src/controller/job.c) uses to derive each element's identity.
*
* Each expanded element gets its OWN independent deep copy of the shared
* submit spec (clone_job_submit_data/clone_job_startinfo below), never a
* shared pointer: the existing per-job pipeline this feeds into
* (sched_on_jobs -> XCE_SCHED_OP_UPDATE_JOB -> sched_pend_update_killer)
* calls free_Job_ptr() on every element independently, which would
* double-free a submit pointer shared across more than one xce_job_t. */
static job_startinfo_t* clone_job_startinfo(const job_startinfo_t* src)
{
job_startinfo_t* d;
if (src == NULL) {
return NULL;
}
d = (job_startinfo_t*)calloc(1, sizeof(*d));
if (d == NULL) {
return NULL;
}
#define CLONE_STR_FIELD(f) \
do { \
if (src->f != NULL) { \
d->f = strdup(src->f); \
if (d->f == NULL) { \
goto fail; \
} \
} \
} while (0)
CLONE_STR_FIELD(exechome);
CLONE_STR_FIELD(execcwd);
CLONE_STR_FIELD(execuser);
CLONE_STR_FIELD(cmdline);
CLONE_STR_FIELD(exdata);
CLONE_STR_FIELD(launch_type);
CLONE_STR_FIELD(q_pre_exec);
CLONE_STR_FIELD(q_post_exec);
CLONE_STR_FIELD(job_pre_exec);
CLONE_STR_FIELD(job_post_exec);
#undef CLONE_STR_FIELD
return d;
fail:
free_JobStartInfo_ptr(&d);
return NULL;
}
static job_submit_data_t* clone_job_submit_data(const job_submit_data_t* src)
{
job_submit_data_t* d;
if (src == NULL) {
return NULL;
}
d = (job_submit_data_t*)calloc(1, sizeof(*d));
if (d == NULL) {
return NULL;
}
d->mintasks = src->mintasks;
d->maxtasks = src->maxtasks;
#define CLONE_STR_FIELD(f) \
do { \
if (src->f != NULL) { \
d->f = strdup(src->f); \
if (d->f == NULL) { \
goto fail; \
} \
} \
} while (0)
CLONE_STR_FIELD(queue);
CLONE_STR_FIELD(user);
CLONE_STR_FIELD(job_name);
CLONE_STR_FIELD(project_name);
CLONE_STR_FIELD(user_group);
CLONE_STR_FIELD(job_description);
CLONE_STR_FIELD(resreq);
#undef CLONE_STR_FIELD
if (src->start_info != NULL) {
d->start_info = clone_job_startinfo(src->start_info);
if (d->start_info == NULL) {
goto fail;
}
}
return d;
fail:
free_JobSubmitData_ptr(&d);
return NULL;
}
/* One array element, freshly-submitted: JOB_STAT_PEND / receive-time
* sub_time, mirroring the state a genuinely-new single-job push arrives
* with. The compact wire descriptor (Req 1.1/6.1) carries only (jobid,
* array_size, submit) by design -- no per-element timestamps -- so there is
* no "real" sub_time to recover here. */
static xce_job_t* make_array_element(const xce_job_submit_t* job_array,
uint64_t jobid)
{
xce_job_t* j = (xce_job_t*)calloc(1, sizeof(*j));
if (j == NULL) {
return NULL;
}
j->jobid = jobid;
j->state = JOB_STAT_PEND;
j->sub_time = (uint64_t)time(NULL);
j->submit = clone_job_submit_data(job_array->submit);
if (j->submit == NULL) {
free(j);
return NULL;
}
return j;
}
static int sched_on_job_array(xce_scheduler_t* sched,
xce_job_submit_t* job_array)
{
xce_job_t** jobs;
uint64_t base_jobid;
uint64_t array_size;
uint64_t i;
int n_new = 0;
int rc;
char jobidString[128];
if ((sched == NULL) || (job_array == NULL) || (job_array->submit == NULL)
|| (job_array->array_size == 0)) {
return -EINVAL;
}
base_jobid = job_array->jobid;
array_size = job_array->array_size;
jobs = (xce_job_t**)calloc((size_t)array_size, sizeof(xce_job_t*));
if (jobs == NULL) {
return -ENOMEM;
}
for (i = 0; i < array_size; i++) {
uint64_t jobid = base_jobid + i;
/* Req 6.5/8.4: deduplicate a redelivered push by element identity --
* reuse the SAME jobid-keyed exists-check (sched_get_job against
* sched->jobs) the ordinary single-job path relies on downstream
* (sched_add_job -> key_add -> -EEXIST), applied here per-element so
* an already-materialized element is never re-queued. Not a new
* dedup mechanism. */
if (sched_get_job(sched, jobid) != NULL) {
continue;
}
jobs[n_new] = make_array_element(job_array, jobid);
if (jobs[n_new] == NULL) {
int j;
for (j = 0; j < n_new; j++) {
free_Job_ptr(&jobs[j]);
}
free(jobs);
return -ENOMEM;
}
n_new++;
}
if (n_new == 0) {
/* Every index already exists: a fully-duplicate redelivery. Nothing
* new to schedule, and nothing to confirm -- Req 8.4 ties the
* confirmation log to a successful *expansion*; a redelivery that
* expands zero new elements is a no-op, not a fresh expansion
* (task 5.3 Observable note). */
free(jobs);
return 0;
}
fmt_jobid(jobidString, sizeof(jobidString) - 1, base_jobid);
INFO("job array push received and expanded: base_jobid=%s "
"array_size=%llu new_tasks=%d",
jobidString, (unsigned long long)array_size, n_new);
rc = sched_on_jobs(sched, n_new, jobs);
if (rc < 0) {
for (i = 0; i < (uint64_t)n_new; i++) {
free_Job_ptr(&jobs[i]);
}
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
* below, which hands dataset->rows to sched_on_jobs the same way). */
return rc;
}
static int sched_on_workers(xce_scheduler_t* sched, int nworkers,
xce_worker_t** workers)
{
@ -143,6 +359,15 @@ static int sched_handle_dataset(xce_scheduler_t* sched, dataset_t* dataset)
dataset->rows = NULL;
}
break;
case Data_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
* independent clone (see make_array_element/clone_job_submit_data
* above) -- so the original (jobid, arraySize, submit) descriptor
* stays exactly as owned/freed by whatever handed it to us. Do not
* null it out here. */
break;
}
return rc;

View file

@ -30,6 +30,18 @@ sched_handlers_test = executable('sched_handlers_test',
dependencies: [scheduler_test_deps],
install: false)
# White-box test for the compact job-array push receive/expand path
# (job-array-single-wal-record task 5.3). It also #includes ../src/handler.c
# directly, but additionally needs a REAL sched->jobs kvmgr (sched_get_job/
# sched_add_job) to exercise the per-element dedup check for real — kept as
# its own binary rather than folded into sched_handlers_test so that file's
# "self-contained, only sched_init_config stubbed" story stays accurate.
sched_job_array_test = executable('sched_job_array_test',
['test_job_array_expand.c'],
include_directories:[incdir],
dependencies: [scheduler_test_deps],
install: false)
# Register as meson tests so `meson test --suite scheduler` runs them all.
# detect_leaks=0 matches the suite convention (fixtures are not torn down).
test('scheduler test', scheduler_test, suite: 'scheduler',
@ -38,3 +50,5 @@ test('scheduler job-update test', sched_job_update_test, suite: 'scheduler',
env: ['ASAN_OPTIONS=detect_leaks=0'])
test('scheduler handlers test', sched_handlers_test, suite: 'scheduler',
env: ['ASAN_OPTIONS=detect_leaks=0'])
test('scheduler job-array expand test', sched_job_array_test, suite: 'scheduler',
env: ['ASAN_OPTIONS=detect_leaks=0'])

View file

@ -19,16 +19,24 @@
/* Pull the real handlers under test into this TU. handler.c is self-contained
* for the enqueue path: enq_pend_update and sched_job_state_change are both
* static within it; the dispatchers call lib helpers (linked). The only
* scheduler-internal extern handler.c references (in the config-init request
* path, not exercised by these enqueue tests) is sched_init_config — stubbed. */
#include "../src/handler.c"
* static within it; the dispatchers call lib helpers (linked). The
* 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
* 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)
{
(void)s; (void)c; return 0;
}
xce_sched_job_t* sched_get_job(xce_scheduler_t* sched, uint64_t jobid)
{
(void)sched; (void)jobid; return NULL;
}
#include "../src/handler.c"
void setUp(void) { return; }
void tearDown(void)
{

View file

@ -0,0 +1,261 @@
/*
* test_job_array_expand.c - white-box tests for the scheduler's compact
* job-array push receive path (job-array-single-wal-record task 5.3, Req
* 6.3, 6.4, 6.5, 8.4).
*
* Strategy: pull the real (static) sched_handle_dataset / sched_on_job_array
* into this TU via #include "../src/handler.c" (same white-box pattern as
* test_handlers.c), driven against a REAL sched->jobs kvmgr so the
* per-element dedup check (sched_get_job) is exercised for real rather than
* faked. sched_add_job/sched_get_job below are the SAME kvmgr key_add/
* key_find path production uses (mirrors test_job_update.c's precedent).
*/
#define UNIT_TEST 1
#include <errno.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unity/unity.h>
#include "kvmgr.h"
#include "scheduler.h"
/* Pull the real handler-under-test into this TU. handler.c is otherwise
* self-contained for the dataset dispatch path (enq_pend_update is static
* within it); the only scheduler-internal extern it references that lives
* elsewhere is sched_init_config, stubbed below (same as test_handlers.c). */
#include "../src/handler.c"
int sched_init_config(xce_scheduler_t* s, xce_conf4sched_t* c)
{
(void)s; (void)c; return 0;
}
/* REAL kvmgr-backed job add/get -- same rtti path production uses (mirrors
* test_job_update.c's precedent for exercising sched_get_job "for real"). */
int sched_add_job(xce_scheduler_t* sched, xce_sched_job_t* j)
{
int rc;
lock(sched->jobs, RTTI_LOCK_WRITE);
rc = key_add(sched->jobs, (void*)j->jobid, j);
unlock(sched->jobs);
return rc;
}
xce_sched_job_t* sched_get_job(xce_scheduler_t* sched, uint64_t jobid)
{
int rc;
xce_sched_job_t* job = NULL;
lock(sched->jobs, RTTI_LOCK_READ);
rc = key_find(sched->jobs, (void*)jobid, (void**)&job);
unlock(sched->jobs);
if (rc <= 0) {
return NULL;
}
return job;
}
void setUp(void) { return; }
void tearDown(void)
{
printf("\n-----------------------------------------------------\n");
return;
}
static void fixture_init(xce_scheduler_t* sched)
{
kvmgr_params_t jobs_params = { .key_type = KVMGR_KEY_INT,
.nslots = 59,
.delta = 16,
.datakiller = NULL };
memset(sched, 0, sizeof(*sched));
mylock_init(&(sched->lock));
sched->pend_updates[0] = (dataq_t*)calloc(1, sizeof(dataq_t));
dataq_init(sched->pend_updates[0], NULL);
sched->pend_updates[1] = (dataq_t*)calloc(1, sizeof(dataq_t));
dataq_init(sched->pend_updates[1], NULL);
sched->jobs = makeInstance(&xkvmgr_type, &jobs_params);
}
static void seed_existing_job(xce_scheduler_t* sched, uint64_t jobid)
{
xce_sched_job_t* job = (xce_sched_job_t*)calloc(1, sizeof(*job));
job->jobid = jobid;
sched_add_job(sched, job);
}
/* Fills a caller-owned (jobArray, submit) pair with a small, valid, sharable
* job spec -- the same (jobid, arraySize, submit) shape task 5.1's
* xcesched_push_jobarray() sends. */
static void init_job_array(xce_job_submit_t* ja, job_submit_data_t* submit,
uint64_t base_jobid, uint64_t array_size)
{
memset(submit, 0, sizeof(*submit));
submit->queue = "normal";
submit->user = "tester";
submit->mintasks = 1;
submit->maxtasks = 1;
submit->resreq = "type == local";
memset(ja, 0, sizeof(*ja));
ja->jobid = base_jobid;
ja->array_size = array_size;
ja->submit = submit;
}
/* Pop one pending update request, asserting it is an UPDATE_JOB op; hands
* back njobs/jobs (caller now owns them -- free via free_Job_ptr + free() to
* mirror sched_pend_update_killer). Returns -1 if the queue is empty. */
static int pop_job_update(xce_scheduler_t* sched, xce_job_t*** jobs_out)
{
data_item_t* it = dataq_deq(sched->pend_updates[0], NULL);
xce_sched_update_request_t* req;
int njobs;
if (it == NULL) {
return -1;
}
req = ITPTR_TO_UPDATE_REQ(it);
TEST_ASSERT_EQUAL_INT(XCE_SCHED_OP_UPDATE_JOB, req->op);
njobs = req->u.jobupdate.njobs;
if (jobs_out != NULL) {
*jobs_out = req->u.jobupdate.jobs;
}
free(req);
return njobs;
}
/* ── Req 6.3/6.4: a compact Data.jobArray push expands locally into N
* ordinary schedulable tasks via base_jobid + i arithmetic ── */
static void test_job_array_push_expands_into_n_tasks(void)
{
xce_scheduler_t sched;
job_submit_data_t submit;
xce_job_submit_t ja;
dataset_t dataset;
xce_job_t** jobs = NULL;
int njobs;
int i;
fixture_init(&sched);
init_job_array(&ja, &submit, 5000, 3);
memset(&dataset, 0, sizeof(dataset));
dataset.table = Data_data_jobArray;
dataset.rows = (void*)&ja;
TEST_ASSERT_TRUE(sched_handle_dataset(&sched, &dataset) >= 0);
njobs = pop_job_update(&sched, &jobs);
TEST_ASSERT_EQUAL_INT(3, njobs);
TEST_ASSERT_EQUAL_UINT64(5000, jobs[0]->jobid);
TEST_ASSERT_EQUAL_UINT64(5001, jobs[1]->jobid);
TEST_ASSERT_EQUAL_UINT64(5002, jobs[2]->jobid);
TEST_ASSERT_EQUAL_STRING("normal", jobs[0]->submit->queue);
TEST_ASSERT_EQUAL_STRING("normal", jobs[2]->submit->queue);
/* every element gets its OWN independent clone, not a shared pointer --
* required so each is independently free_Job_ptr()-able downstream
* (sched_pend_update_killer) without a double-free. */
TEST_ASSERT_TRUE(jobs[0]->submit != jobs[1]->submit);
TEST_ASSERT_TRUE(jobs[1]->submit != jobs[2]->submit);
for (i = 0; i < njobs; i++) {
free_Job_ptr(&jobs[i]);
}
free(jobs);
}
/* ── Req 6.5/8.4: a redelivered push (e.g. after a scheduler reconnect)
* must not create duplicate tasks -- reuse the jobid-keyed exists-check
* (sched_get_job) the ordinary single-job path relies on ── */
static void test_job_array_redelivery_does_not_duplicate(void)
{
xce_scheduler_t sched;
job_submit_data_t submit;
xce_job_submit_t ja;
dataset_t dataset;
xce_job_t** jobs = NULL;
int njobs;
int i;
fixture_init(&sched);
init_job_array(&ja, &submit, 6000, 2);
memset(&dataset, 0, sizeof(dataset));
dataset.table = Data_data_jobArray;
dataset.rows = (void*)&ja;
/* first delivery: 2 new tasks enqueued */
TEST_ASSERT_TRUE(sched_handle_dataset(&sched, &dataset) >= 0);
njobs = pop_job_update(&sched, &jobs);
TEST_ASSERT_EQUAL_INT(2, njobs);
for (i = 0; i < njobs; i++) {
/* simulate the async consumer (scheduler.c's sched_update_jobs ->
* sched_add_job, not linked into this white-box TU) materializing
* the element into sched->jobs, so the redelivery's dedup check
* sees them as already existing -- exactly what a real reconnect
* scenario looks like from the handler's point of view. */
seed_existing_job(&sched, jobs[i]->jobid);
free_Job_ptr(&jobs[i]);
}
free(jobs);
/* redelivery of the IDENTICAL push: every index already exists */
TEST_ASSERT_EQUAL_INT(0, sched_handle_dataset(&sched, &dataset));
TEST_ASSERT_EQUAL_INT(-1, pop_job_update(&sched, NULL)); /* nothing new */
}
/* ── Guard-clause coverage on the decode step itself ── */
static void test_job_array_null_submit_returns_einval(void)
{
xce_scheduler_t sched;
xce_job_submit_t ja;
fixture_init(&sched);
memset(&ja, 0, sizeof(ja));
ja.jobid = 1;
ja.array_size = 3;
ja.submit = NULL;
TEST_ASSERT_EQUAL_INT(-EINVAL, sched_on_job_array(&sched, &ja));
}
static void test_job_array_zero_size_returns_einval(void)
{
xce_scheduler_t sched;
job_submit_data_t submit;
xce_job_submit_t ja;
fixture_init(&sched);
init_job_array(&ja, &submit, 42, 0);
TEST_ASSERT_EQUAL_INT(-EINVAL, sched_on_job_array(&sched, &ja));
}
static void test_job_array_null_descriptor_returns_einval(void)
{
xce_scheduler_t sched;
fixture_init(&sched);
TEST_ASSERT_EQUAL_INT(-EINVAL, sched_on_job_array(&sched, NULL));
}
int main(int argc, char* argv[])
{
(void)argc; (void)argv;
UNITY_BEGIN();
RUN_TEST(test_job_array_push_expands_into_n_tasks, 0);
RUN_TEST(test_job_array_redelivery_does_not_duplicate, 0);
RUN_TEST(test_job_array_null_submit_returns_einval, 0);
RUN_TEST(test_job_array_zero_size_returns_einval, 0);
RUN_TEST(test_job_array_null_descriptor_returns_einval, 0);
return UNITY_END();
}