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

feat(job-array-single-wal-record): build the compact API-server push

Browse source 
Adds apiserver_push_jobarray() (apiserver.c), mirroring task 5.1's
xcesched_push_jobarray() exactly: same (base jobid, array_size, one
spec) Data.jobArray derivation, same direct xce_request()/OPCode_push
transport. apiserver_job_new_fun() branches on l->count > 1; the
existing single-job path (apiserver_push_joblist()) is byte-for-byte
unchanged.

Additionally, and unlike the scheduler (single target), replicates
apiserver_push_joblist()'s existing per-connected-server fan-out loop
(N_APISERVER_MAX iteration, same lock discipline) -- verified against
2 simulated connected servers, each receiving exactly one push.

New standalone apiserver_push_unit_test binary (same link-collision
rationale as task 5.1's sched_push_unit_test), correctly omitting
egress.c since apiserver_push_joblist() has no WAL-commit filter.

Task 5.2 of job-array-single-wal-record (Requirements 6.2, 6.4).
This commit is contained in:
Rongsong Shen 2026-07-15 08:30:27 +00:00
parent 5e7a4aa392
commit e0cec75284
4 changed files with 449 additions and 2 deletions
Download patch file Download diff file Expand all files Collapse all files

85
src/controller/apiserver.c
View file

@ -124,6 +124,79 @@ static int apiserver_push_joblist(cluster_t* c, xlist_t* l)
return 0;
}
/* job-array-single-wal-record task 5.2 (design.md "Compact Scheduler/
* API-Server Push", File Structure Plan apiserver.c bullet; Req 6.2, 6.4):
* encodes ONE compact (base jobid, array size, one shared spec) Data.jobArray
* push per connected API server -- reusing the JobSubmit capnp struct
* (already added to the Data union in task 1.2, controller.capnp), never N
* individually-encoded Job structs. (base_jobid, array_size) are derived
* from the already-constructed xlist_t the HOOK_DETAIL_JOB_NEW fire carries
* (apply_committed_entry's WAL_JOB_ARRAY_SUBMIT case, task 3.2): the first
* element's jobid is the array base and l->count is the array size; every
* element shares the same submit spec by construction (restore_job_array's
* own construction discipline, job.c), so any one element's submit pointer
* is the one shared spec to encode.
*
* Sent via xce_request() directly (OPCode_push / Request_body_pushmsg)
* rather than xce_push_data_req(), mirroring sched.c's xcesched_push_
* jobarray() (task 5.1) for the identical reason: that helper's dispatch
* switch has explicit cases only for jobs/workers/resources/resvals and
* hard-rejects (-EINVAL) anything else via its default case -- it cannot
* carry a single non-list Data.jobArray descriptor. This mirrors xce_push_
* req()'s (request.c, static) exact shape for a single non-list xce_data_t
* payload.
*
* Fan-out mirrors apiserver_push_joblist()'s existing per-connected-server
* loop exactly (same lock discipline, same name!=NULL && ep!=NULL guard) --
* one push sent per connected server, not one push shared/broadcast. */
static int apiserver_push_jobarray(cluster_t* c, xlist_t* l)
{
xce_job_t* first;
xce_job_submit_t job_array;
int i;
if ((c == NULL) || (l == NULL) || (l->count <= 0)) {
return -EINVAL;
}
first = (xce_job_t*)xlist_get(l, 0);
if ((first == NULL) || (first->submit == NULL)) {
return -EINVAL;
}
memset(&job_array, 0x0, sizeof(job_array));
job_array.jobid = first->jobid;
job_array.array_size = (uint64_t)l->count;
job_array.submit = first->submit;
for (i = 0; i < N_APISERVER_MAX; i++) {
mylock_lock(&(c->apiservers[i].lock));
if ((c->apiservers[i].name != NULL) && c->apiservers[i].ep != NULL) {
xce_data_t data;
xce_request_t request;
int rc;
memset(&data, 0x0, sizeof(data));
data.table = Data_data_jobArray;
data.data.job_array = &job_array;
memset(&request, 0x0, sizeof(request));
request.opcode = OPCode_push;
request.body_which = Request_body_pushmsg;
request.body.pushmsg = &data;
rc = xce_request(c->apiservers[i].ep, &request);
if (rc < 0) {
WARN("%s push job array to api server %s has problem, rc %d",
__func__, c->apiservers[i].name, rc);
}
}
mylock_unlock(&(c->apiservers[i].lock));
}
return 0;
}
static int apiserver_job_new_fun(int type, int situation, void* data,
va_list ap)
{
@ -135,7 +208,17 @@ static int apiserver_job_new_fun(int type, int situation, void* data,
case HOOK_AFTER_EVENT:
rc = -EINVAL;
if (l != NULL) {
rc = apiserver_push_joblist(c, l);
/* job-array-single-wal-record task 5.2 (Req 6.2, 6.4):
* apply_committed_entry()'s WAL_JOB_ARRAY_SUBMIT case is the
* ONLY way this hook is ever fired with more than one element
* (Req 3.4 -- one HOOK_DETAIL_JOB_NEW fire per committed array
* record, carrying every constructed element). Route that case
* through the compact descriptor push instead of the per-job
* push. The ordinary single-job branch (l->count == 1) below is
* the exact, unmodified call this function always made
* (Req 8.1). */
rc = (l->count > 1) ? apiserver_push_jobarray(c, l)
: apiserver_push_joblist(c, l);
}
break;
default: