feat(job-array-single-wal-record): build the compact scheduler push
Adds xcesched_push_jobarray() (sched.c), encoding one compact (base jobid, array_size, one spec) Data.jobArray descriptor -- reusing the JobSubmit capnp struct already added to the Data union in task 1.2 -- instead of N individually-encoded job structs. sched_job_new_fun() branches on l->count > 1 to call it; the existing single-job path (l->count == 1 -> xcesched_push_joblist()) is byte-for-byte unchanged. Sends via direct xce_request()/OPCode_push/Request_body_pushmsg rather than the existing xce_push_data_req() helper, which hard-rejects any payload outside jobs/workers/resources/resvals and cannot carry a single non-list descriptor. design.md corrected to record this transport deviation for task 5.2 to follow. New standalone sched_push_unit_test binary (registered in meson.build): sched.c/dataex.c/egress.c were never linked into controller_unit_test, and two existing test files there ship local fakes of on_scheduler_join()/notify()/collect_confvars() that would collide with the real definitions at link time. Task 5.1 of job-array-single-wal-record (Requirements 6.1, 6.3, 6.4).
This commit is contained in:
parent
1888498401
commit
5e7a4aa392
5 changed files with 385 additions and 5 deletions
282
src/controller/testsuite/test_job_array_sched_push.c
Normal file
282
src/controller/testsuite/test_job_array_sched_push.c
Normal file
|
|
@ -0,0 +1,282 @@
|
|||
/* test_job_array_sched_push.c — job-array-single-wal-record task 5.1
|
||||
*
|
||||
* Unit tests for the Compact Scheduler/API-Server Push component's
|
||||
* controller-side scheduler push (design.md "Compact Scheduler/API-Server
|
||||
* Push"; Req 6.1, 6.3, 6.4):
|
||||
*
|
||||
* - A synthetic multi-element HOOK_DETAIL_JOB_NEW fire (the exact shape
|
||||
* apply_committed_entry()'s WAL_JOB_ARRAY_SUBMIT case produces, task 3.2 —
|
||||
* one AFTER_EVENT fire, one xlist_t carrying every constructed element)
|
||||
* produces EXACTLY ONE queued push packet, encoding the compact
|
||||
* Data.jobArray descriptor (base jobid + array size + one shared spec) —
|
||||
* never N individually-encoded Job/SchedJob structs.
|
||||
* - A synthetic single-element fire (ordinary single-job submission) is
|
||||
* byte-for-byte unaffected: still the existing Data.schedjobs push via
|
||||
* xcesched_push_joblist(), unchanged (Req 8.1).
|
||||
*
|
||||
* Fixture: sched_job_new_fun()/xcesched_push_joblist()/xcesched_push_jobarray()
|
||||
* are static to sched.c, so this file drives them through the same public
|
||||
* entry points production code uses: on_scheduler_join() installs the hook
|
||||
* (its `hook_installed` static is once-per-process, so ALL tests in this
|
||||
* file share ONE cluster_t/hooksys wired exactly once — see jsp_setup) and
|
||||
* xce_run_hooks() fires it exactly as apply_committed_entry() does. The
|
||||
* outbound push itself is observed the same way test_commit_barrier_rollout.c
|
||||
* observes queued replies (cb5_scan_replies): a zeroed, fd=-1 stack
|
||||
* endpoint_t queues packets on ep->outq without touching a real socket
|
||||
* (endpoint_update_epoll no-ops, zeroed pthread mutexes are valid); this
|
||||
* test drains and raw-capnp-decodes those packets instead.
|
||||
*
|
||||
* Job list elements are plain job_t structs with only the two fields either
|
||||
* push function reads (.job.jobid, .job.submit) plus submit_lsn = 0 (makes
|
||||
* job_submit_committed() unconditionally true, matching a freshly-committed
|
||||
* array's elements without needing a real persist_t) — NOT routed through
|
||||
* create_job()/jobj.c, whose N-way construction is already covered by
|
||||
* test_job_array_apply.c.
|
||||
*/
|
||||
|
||||
#include <capnp_c.h>
|
||||
#include <errno.h>
|
||||
#include <stdint.h>
|
||||
#include <stdlib.h>
|
||||
#include <string.h>
|
||||
|
||||
#include <unity/unity.h>
|
||||
|
||||
#include "cluster.h"
|
||||
#include "controller.capnp.h"
|
||||
#include "hook.h"
|
||||
#include "netio.h"
|
||||
#include "request.h"
|
||||
#include "xce/cluster_def.h"
|
||||
#include "xce/types.h"
|
||||
#include "xlist.h"
|
||||
|
||||
extern int on_scheduler_join(cluster_t*);
|
||||
|
||||
/* This file builds a STANDALONE test binary (meson.build's sched_push_unit_
|
||||
* test) linking the real sched.c + its real dataex.c/egress.c dependencies —
|
||||
* see meson.build's comment for why this can't be folded into the shared
|
||||
* controller_unit_test binary. opget.c's check_worker_() is sched.c's only
|
||||
* remaining unresolved reference (from xcesched_push_workers(), reached only
|
||||
* through on_scheduler_join()'s push-on-join branch, which this file's setup
|
||||
* deliberately never triggers at runtime — c->sched.ep is NULL during that
|
||||
* call — but the symbol must still resolve at link time). Stubbed locally
|
||||
* rather than linking all of opget.c, mirroring opget.c's own trivial body. */
|
||||
int check_worker_(xce_worker_t* w) { return (w != NULL) ? 1 : 0; }
|
||||
|
||||
/* ------------------------------------------------------------------ */
|
||||
/* Fixture */
|
||||
/* ------------------------------------------------------------------ */
|
||||
|
||||
static cluster_t g_jsp_c;
|
||||
static endpoint_t g_jsp_ep;
|
||||
static int g_jsp_hook_wired;
|
||||
|
||||
static void jsp_drain_ep(endpoint_t* ep)
|
||||
{
|
||||
data_item_t* it;
|
||||
|
||||
while ((it = endpoint_deq(ep)) != NULL) {
|
||||
io_packet_t* pkt = ITPTR_TO_PACKET(it);
|
||||
|
||||
free_packet(pkt);
|
||||
free(pkt);
|
||||
}
|
||||
}
|
||||
|
||||
static void jsp_setup(void)
|
||||
{
|
||||
if (!g_jsp_hook_wired) {
|
||||
memset(&g_jsp_c, 0, sizeof(g_jsp_c));
|
||||
TEST_ASSERT_EQUAL_INT(
|
||||
0, xce_hook_init(&g_jsp_c.hooksys, XCE_CONTROLLER_HOOK_MAX));
|
||||
/* sched.ep == NULL here so on_scheduler_join's push-on-join branch
|
||||
* (xcesched_push_configs/workers/jobs) is a no-op — this call's only
|
||||
* effect is installing sched_job_hook_fun (and siblings) into
|
||||
* g_jsp_c.hooksys, once, process-wide (hook_installed is a static
|
||||
* inside on_scheduler_join). */
|
||||
g_jsp_c.sched.ep = NULL;
|
||||
TEST_ASSERT_EQUAL_INT(0, on_scheduler_join(&g_jsp_c));
|
||||
g_jsp_hook_wired = 1;
|
||||
}
|
||||
|
||||
memset(&g_jsp_ep, 0, sizeof(g_jsp_ep));
|
||||
g_jsp_ep.fd = -1;
|
||||
|
||||
g_jsp_c.sched.ep = &g_jsp_ep;
|
||||
g_jsp_c.running = 1;
|
||||
g_jsp_c.myhost = &g_jsp_c.myhost_info;
|
||||
g_jsp_c.myhost_info.name = (char*)"ctrl-jsp";
|
||||
g_jsp_c.myhost_info.key = (char*)"";
|
||||
}
|
||||
|
||||
static void jsp_teardown(void)
|
||||
{
|
||||
jsp_drain_ep(&g_jsp_ep);
|
||||
g_jsp_c.sched.ep = NULL;
|
||||
}
|
||||
|
||||
/* ------------------------------------------------------------------ */
|
||||
/* Helpers */
|
||||
/* ------------------------------------------------------------------ */
|
||||
|
||||
static job_submit_data_t g_jsp_submit_data;
|
||||
|
||||
static void jsp_make_job(job_t* out, uint64_t jobid)
|
||||
{
|
||||
memset(out, 0, sizeof(*out));
|
||||
out->job.jobid = jobid;
|
||||
out->job.submit = &g_jsp_submit_data;
|
||||
out->submit_lsn
|
||||
= 0; /* job_submit_committed(): submit_lsn==0 => always exportable */
|
||||
}
|
||||
|
||||
static int jsp_fire_job_new(xlist_t* l)
|
||||
{
|
||||
return xce_run_hooks(&g_jsp_c.hooksys, XCE_CONTROLLER_HOOK_JOB,
|
||||
HOOK_AFTER_EVENT, HOOK_DETAIL_JOB_NEW, &g_jsp_c, l);
|
||||
}
|
||||
|
||||
typedef struct {
|
||||
int n_pushes;
|
||||
int last_data_which;
|
||||
uint64_t last_jobid;
|
||||
uint64_t last_array_size;
|
||||
} jsp_push_obs_t;
|
||||
|
||||
/* Raw-capnp-parse every queued push packet on ep->outq (mirrors
|
||||
* test_commit_barrier_rollout.c's cb5_scan_replies). Drains the queue. */
|
||||
static void jsp_scan_pushes(endpoint_t* ep, jsp_push_obs_t* obs)
|
||||
{
|
||||
data_item_t* it;
|
||||
|
||||
memset(obs, 0, sizeof(*obs));
|
||||
while ((it = endpoint_deq(ep)) != NULL) {
|
||||
io_packet_t* pkt = ITPTR_TO_PACKET(it);
|
||||
struct capn c;
|
||||
struct Request rq;
|
||||
Request_ptr rp;
|
||||
|
||||
capn_init_mem(&c, (const uint8_t*)pkt->data, pkt->sz, 0);
|
||||
rp.p = capn_getp(capn_root(&c), 0, 1);
|
||||
read_Request(&rq, rp);
|
||||
if (rq.body_which == Request_body_pushmsg) {
|
||||
struct Data d;
|
||||
|
||||
read_Data(&d, rq.body.pushmsg);
|
||||
obs->n_pushes += 1;
|
||||
obs->last_data_which = (int)d.data_which;
|
||||
if (d.data_which == Data_data_jobArray) {
|
||||
struct JobSubmit js;
|
||||
|
||||
read_JobSubmit(&js, d.data.jobArray);
|
||||
obs->last_jobid = js.jobid;
|
||||
obs->last_array_size = js.arraySize;
|
||||
}
|
||||
}
|
||||
capn_free(&c);
|
||||
free_packet(pkt);
|
||||
free(pkt);
|
||||
}
|
||||
}
|
||||
|
||||
/* ------------------------------------------------------------------ */
|
||||
/* Test 1: multi-element fire -> exactly one compact Data.jobArray push */
|
||||
/* ------------------------------------------------------------------ */
|
||||
|
||||
static void test_multi_element_job_new_produces_one_compact_jobarray_push(void)
|
||||
{
|
||||
job_t jobs[5];
|
||||
void* storage[5];
|
||||
xlist_t l = { .count = 5, .caps = 5, .delta = 5, .data = storage };
|
||||
jsp_push_obs_t obs;
|
||||
int rc;
|
||||
|
||||
jsp_setup();
|
||||
|
||||
for (int i = 0; i < 5; i++) {
|
||||
jsp_make_job(&jobs[i], 5000 + (uint64_t)i);
|
||||
storage[i] = &jobs[i].job;
|
||||
}
|
||||
|
||||
rc = jsp_fire_job_new(&l);
|
||||
TEST_ASSERT_EQUAL_INT(0, rc);
|
||||
|
||||
jsp_scan_pushes(&g_jsp_ep, &obs);
|
||||
|
||||
TEST_ASSERT_EQUAL_INT_MESSAGE(
|
||||
1, obs.n_pushes,
|
||||
"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,
|
||||
"the push must encode the compact Data.jobArray descriptor, not "
|
||||
"Data.schedjobs/Data.jobs (N individually-encoded job structs)");
|
||||
TEST_ASSERT_EQUAL_UINT64_MESSAGE(
|
||||
5000, obs.last_jobid,
|
||||
"base jobid must be the array's first (lowest) jobid");
|
||||
TEST_ASSERT_EQUAL_UINT64_MESSAGE(
|
||||
5, obs.last_array_size,
|
||||
"arraySize must be the notification's element count");
|
||||
|
||||
jsp_teardown();
|
||||
printf("test_multi_element_job_new_produces_one_compact_jobarray_push: "
|
||||
"PASS\n");
|
||||
}
|
||||
|
||||
/* ------------------------------------------------------------------ */
|
||||
/* Test 2: single-element fire -> unmodified per-job Data.schedjobs push */
|
||||
/* ------------------------------------------------------------------ */
|
||||
|
||||
static void test_single_element_job_new_is_unaffected(void)
|
||||
{
|
||||
job_t job;
|
||||
void* storage[1];
|
||||
xlist_t l = { .count = 1, .caps = 1, .delta = 1, .data = storage };
|
||||
jsp_push_obs_t obs;
|
||||
int rc;
|
||||
|
||||
jsp_setup();
|
||||
|
||||
jsp_make_job(&job, 9001);
|
||||
storage[0] = &job.job;
|
||||
|
||||
rc = jsp_fire_job_new(&l);
|
||||
TEST_ASSERT_EQUAL_INT(0, rc);
|
||||
|
||||
jsp_scan_pushes(&g_jsp_ep, &obs);
|
||||
|
||||
TEST_ASSERT_EQUAL_INT_MESSAGE(
|
||||
1, obs.n_pushes,
|
||||
"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,
|
||||
"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)");
|
||||
|
||||
jsp_teardown();
|
||||
printf("test_single_element_job_new_is_unaffected: PASS\n");
|
||||
}
|
||||
|
||||
/* ------------------------------------------------------------------ */
|
||||
/* Standalone Unity main (this binary is its own test executable — see */
|
||||
/* meson.build's sched_push_unit_test) */
|
||||
/* ------------------------------------------------------------------ */
|
||||
|
||||
void setUp(void) { return; }
|
||||
void tearDown(void) { return; }
|
||||
|
||||
int main(void)
|
||||
{
|
||||
printf("\n>> job-array-single-wal-record task 5.1: compact scheduler "
|
||||
"push\n\n");
|
||||
|
||||
UNITY_BEGIN();
|
||||
|
||||
RUN_TEST(test_multi_element_job_new_produces_one_compact_jobarray_push);
|
||||
RUN_TEST(test_single_element_job_new_is_unaffected);
|
||||
|
||||
return UNITY_END();
|
||||
}
|
||||
Loading…
Add table
Add a link
Reference in a new issue