feat(scheduler-core): implement scheduler-side admin request handler (task 9.4)

Implements the scheduler's first request-reply opcode (OPCode_schedAdmin).
sched_handle_admin_request() (handler.c) decodes SchedAdminRequest and
dispatches by action: recordStatus answers synchronously (pure read);
recordStart/recordStop are deferred onto the existing pend_updates queue
(XCE_SCHED_OP_ADMIN_RECORD) since sched_record_start()'s snapshot call
requires the session_active==0 precondition only guaranteed on the driver
thread, not the async request-handling thread. driver.c's
xce_sched_driver_translate_pending_events() performs the deferred action
and sends the reply via a new xcecli_reply_schedadmin() (client.c/client.h),
correlated by seq copied before the original request is freed.

Adversarial review found a genuine cross-thread race: sched_record_status()
read driver-owned recording state (including a pointer the driver thread
frees) from the vtask thread with no lock -- the codebase's first such
race, missed by ASan (doesn't detect races). Fixed with a dedicated
record_lock scoped strictly to in-memory field access, never spanning file
I/O; sched_record_status() now strdup()s under the lock instead of
returning a borrowed live pointer. Verified with a real two-thread
lock-contention test (independently confirmed non-vacuous by removing the
lock and observing the test fail) plus a deterministic use-after-free
reproduction.

26/26 scheduler suite, clang-format clean.
This commit is contained in:
Rongsong Shen 2026-08-05 14:11:43 +00:00
parent 7494542a60
commit e819a50600
Signed by: rshen
SSH key fingerprint: SHA256:8Vf2pZ3h6rJa7EhGK+WaAQ9Mta/apZvteW6g9U/WKKk
19 changed files with 1761 additions and 115 deletions

View file

@ -29,7 +29,20 @@
int sched_init_config(xce_scheduler_t* s, xce_conf4sched_t* c)
{
(void)s; (void)c; return 0;
(void)s;
(void)c;
return 0;
}
/* sched_record_status (recording.c, scheduler-core spec, task 9.4) --
* handler.c's OPCode_schedAdmin/recordStatus path references it; this file
* never exercises OPCode_schedAdmin, so a link-satisfying stub is enough,
* same rationale as sched_init_config above. */
int sched_record_status(xce_scheduler_t* sched, sched_admin_reply_t* reply)
{
(void)sched;
(void)reply;
return -EINVAL;
}
/* REAL kvmgr-backed job add/get -- same rtti path production uses (mirrors
@ -84,7 +97,7 @@ static void fixture_init(xce_scheduler_t* sched)
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;
job->jobid = jobid;
sched_add_job(&(sched->core), job);
}
@ -99,9 +112,8 @@ static void seed_existing_job(xce_scheduler_t* sched, uint64_t jobid)
static xce_job_submit_t* init_job_array(uint64_t base_jobid,
uint64_t array_size)
{
job_submit_data_t* submit
= (job_submit_data_t*)calloc(1, sizeof(*submit));
xce_job_submit_t* ja = (xce_job_submit_t*)calloc(1, sizeof(*ja));
job_submit_data_t* submit = (job_submit_data_t*)calloc(1, sizeof(*submit));
xce_job_submit_t* ja = (xce_job_submit_t*)calloc(1, sizeof(*ja));
submit->queue = strdup("normal");
submit->user = strdup("tester");
@ -145,12 +157,12 @@ static int pop_job_update(xce_scheduler_t* sched, xce_job_t*** jobs_out)
static void test_job_array_push_expands_into_n_tasks(void)
{
xce_scheduler_t sched;
xce_job_submit_t* ja;
dataset_t dataset;
xce_job_t** jobs = NULL;
int njobs;
int i;
xce_scheduler_t sched;
xce_job_submit_t* ja;
dataset_t dataset;
xce_job_t** jobs = NULL;
int njobs;
int i;
fixture_init(&sched);
ja = init_job_array(5000, 3);
@ -188,11 +200,11 @@ static void test_job_array_push_expands_into_n_tasks(void)
static void test_job_array_redelivery_does_not_duplicate(void)
{
xce_scheduler_t sched;
dataset_t dataset;
xce_job_t** jobs = NULL;
int njobs;
int i;
xce_scheduler_t sched;
dataset_t dataset;
xce_job_t** jobs = NULL;
int njobs;
int i;
fixture_init(&sched);
@ -289,9 +301,9 @@ static void test_wire_dispatch_chain_delivers_n_tasks(void)
xce_data_t data;
xce_request_t request;
xce_request_t decoded;
io_packet_t *pkt;
io_packet_t* pkt;
dataset_t dataset;
xce_job_t **jobs = NULL;
xce_job_t** jobs = NULL;
int njobs;
int i;
@ -319,12 +331,13 @@ static void test_wire_dispatch_chain_delivers_n_tasks(void)
request.body_which = Request_body_pushmsg;
request.body.pushmsg = &data;
pkt = (io_packet_t *)calloc(1, sizeof(*pkt));
pkt = (io_packet_t*)calloc(1, sizeof(*pkt));
TEST_ASSERT_NOT_NULL(pkt);
TEST_ASSERT_EQUAL_INT(1, xce_encode(&request, pkt));
memset(&decoded, 0, sizeof(decoded));
TEST_ASSERT_EQUAL_INT(0, xce_decode(pkt, &decoded)); /* consumes/frees pkt */
TEST_ASSERT_EQUAL_INT(0,
xce_decode(pkt, &decoded)); /* consumes/frees pkt */
/* Act 1: the REAL transport dispatcher (task 5.3b's fix) -- exactly what
* vtask_entry() (xcectx.c) calls on every inbound push. */
@ -364,7 +377,8 @@ static void test_wire_dispatch_chain_delivers_n_tasks(void)
int main(int argc, char* argv[])
{
(void)argc; (void)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);