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

fix(scheduler): ALTER allocation type confusion (bug-1769); NULL queue in legacy fcfs hooks (bug-1771)

Browse source 
bug-1769: every builder stores xce_compound_schedobj_t* (24 bytes) in
obj->u.alters, but sched_make_alter_allocation() read them as
xce_sched_obj_t* (176 bytes) and ran the compound allocation on them -- a
heap-buffer-overflow read of ->u.compound and write of ->alloc for any real
`{..} || {..}` resreq. The compound allocation now takes its resreq
explicitly (sched_make_compound_alloc_), and each ALTER alternative is
tried on behalf of the ALTER obj itself; a failed alternative's partial
allocation is dropped before the next (rollback itself is bug-1766).
sched_ctx_init() rejected ALTER objs ("bad sched obj type"), so ALTER jobs
could never allocate at all -- it now accepts them like COMPOUND (all
downstream checks only distinguish PRIMARY).

bug-1771: plugin_commit_decision() fans every DISPATCH out to the legacy
"xce_sched_fcfs" plugin even when it never tracked the job; fcfs_del_job()
(and fcfs's job-add hook) dereferenced job->queue->name with no NULL
check, SEGV'ing the allocator for a job whose queue was never registered.

Tests (test_parjob_e2e_multislot.c, real ingest + live allocator thread):
test_alter_resreq_job_dispatches_via_real_allocator (ASAN
heap-buffer-overflow before, DISPATCH after) and
test_dispatch_of_job_with_unregistered_queue_does_not_crash (SEGV at
fcfs.c:318 before). Scheduler suite 87/87.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
This commit is contained in:
Rongsong Shen 2026-09-25 12:59:31 +00:00
parent 4ba0f32766
commit 1ecd348f03
Signed by untrusted user: rshen
GPG key ID: 80533536B0EB651C
4 changed files with 162 additions and 15 deletions
Download patch file Download diff file Expand all files Collapse all files

11
src/scheduler/src/plugins/fcfs.c
View file

@ -180,6 +180,12 @@ static int fcfs_update_job(xce_sched_plugin_t* plugin, xce_sched_job_t* j)
return 0;
}
/* bug-1771: an unregistered queue (j->queue NULL) is the same "no such
* queue" case as the lookup miss below */
if (j->queue == NULL) {
return -EINVAL;
}
q = fcfs_get_queue(privdata, j->queue->name);
if (q == NULL) {
return -EINVAL;
@ -311,7 +317,10 @@ static int fcfs_del_job(xce_sched_plugin_t* plugin, uint64_t jobid)
}
job = sched_get_job(&(((xce_scheduler_t*)plugin->sched)->core), jobid);
if (job == NULL) {
/* bug-1771: every DISPATCH is fanned out to this plugin even when it
* never tracked the job, and a job whose queue was never registered
* has job->queue == NULL -- nothing of ours to delete */
if ((job == NULL) || (job->queue == NULL)) {
return 0;
}

5
src/scheduler/src/schedobj.c
View file

@ -472,6 +472,11 @@ int sched_ctx_init(xce_sched_ctx_t* ctx, void* arg, xce_sched_obj_t* obj)
break;
case XCE_RESREQ_COMPOUND:
break;
case XCE_RESREQ_ALTER:
/* bug-1769: an ALTER obj allocates one compound alternative at a
* time on its own behalf (sched_make_alter_allocation) -- same as
* COMPOUND here; downstream only distinguishes PRIMARY */
break;
default:
ERROR("%s bad sched obj type %d", __func__, obj->tag);
return -EINVAL;

43
src/scheduler/src/scheduler.c
View file

@ -273,13 +273,16 @@ static int sched_make_primary_allocation(xce_sched_core_t* core,
return rc;
}
static int sched_make_compound_allocation(xce_sched_core_t* core,
xce_sched_obj_t* obj)
/* Allocate one compound resreq on behalf of obj (a COMPOUND schedobj, or
* an ALTER schedobj trying one of its alternatives): the allocation and
* pend reason land on obj itself. */
static int sched_make_compound_alloc_(xce_sched_core_t* core,
xce_sched_obj_t* obj,
xce_compound_resreq_t* resreq)
{
int i;
xce_compound_resreq_t* resreq = obj->u.compound->resreq;
xce_sched_ctx_t ctx;
int rc = 0;
int i;
xce_sched_ctx_t ctx;
int rc = 0;
sched_ctx_init(&ctx, core, obj);
@ -307,28 +310,40 @@ static int sched_make_compound_allocation(xce_sched_core_t* core,
return rc;
}
static int sched_make_compound_allocation(xce_sched_core_t* core,
xce_sched_obj_t* obj)
{
return sched_make_compound_alloc_(core, obj, obj->u.compound->resreq);
}
static int sched_make_alter_allocation(xce_sched_core_t* core,
xce_sched_obj_t* obj)
{
int i;
for (i = 0; i < obj->u.alters.count; i++) {
xce_sched_obj_t* obj1
= (xce_sched_obj_t*)xlist_get(&(obj->u.alters), i);
/* bug-1769: u.alters holds xce_compound_schedobj_t* (every builder:
* core_events.c, snapshot_load.c, schedobj.c), not xce_sched_obj_t*
* -- reading them as the latter overran the 24-byte allocation.
* Each alternative is tried on behalf of the ALTER obj itself. */
xce_compound_schedobj_t* alt
= (xce_compound_schedobj_t*)xlist_get(&(obj->u.alters), i);
int rc;
if (obj1 == NULL) {
if ((alt == NULL) || (alt->resreq == NULL)) {
continue;
}
rc = sched_make_compound_allocation(core, obj1);
rc = sched_make_compound_alloc_(core, obj, alt->resreq);
if (rc > 0) {
if (obj1->alloc != NULL) {
MOVE_PTR(obj->alloc, obj1->alloc);
}
return rc;
}
/* a failed alternative may leave a partial allocation (no
* rollback yet, bug-1766) -- drop it before trying the next one */
if (obj->alloc != NULL) {
free_JobAlloc_ptr(&(obj->alloc));
}
}
/* FIXME: merge pend reason from all children */

118
src/scheduler/testsuite/test_parjob_e2e_multislot.c
View file

@ -585,10 +585,128 @@ test_multihost_parallel_job_allocation_spans_two_hosts_via_real_dispatch(void)
"maxTasks");
}
/*
* bug-1771: plugin_commit_decision() fans every DISPATCH out to every
* registered plugin's commit_hook, including the legacy "xce_sched_fcfs"
* plugin that never tracked the job. Its fcfs_del_job() dereferenced
* job->queue->name unconditionally, so a job whose queue was never
* registered (job->queue == NULL) SEGV'd the allocator thread. Same real
* dispatch chain as above, minus the QUEUE_UPDATE.
*/
static void test_dispatch_of_job_with_unregistered_queue_does_not_crash(void)
{
xce_scheduler_t* sched = start_sched();
xce_worker_t* worker_ptrs[2];
xce_core_event_t worker_event;
xce_job_t* job_ptrs[1];
xce_core_event_t job_event;
xce_sched_job_t* sjob;
xce_decision_t* decision;
worker_ptrs[0] = make_wire_worker(WORKER_A_NAME, WORKER_SLOTS);
worker_ptrs[1] = make_wire_worker(WORKER_B_NAME, WORKER_SLOTS);
memset(&worker_event, 0, sizeof(worker_event));
worker_event.type = XCE_CORE_EVENT_WORKERS_PUSH;
worker_event.u.workers_push.nworkers = 2;
worker_event.u.workers_push.workers = worker_ptrs;
TEST_ASSERT_TRUE(xce_core_push_events(&(sched->core), &worker_event, 1)
>= 0);
job_ptrs[0] = make_push_job(JOB_ID + 1, RESREQ_STR, "e2e_user",
"never_registered_queue", "e2e_proj", 1,
JOB_MAXTASKS);
memset(&job_event, 0, sizeof(job_event));
job_event.type = XCE_CORE_EVENT_JOBS_PUSH;
job_event.u.jobs_push.njobs = 1;
job_event.u.jobs_push.jobs = job_ptrs;
TEST_ASSERT_TRUE(xce_core_push_events(&(sched->core), &job_event, 1)
>= 0);
sjob = sched_get_job(&(sched->core), JOB_ID + 1);
TEST_ASSERT_NOT_NULL(sjob);
TEST_ASSERT_NULL_MESSAGE(sjob->queue,
"fixture: the job's queue must be unregistered (job->queue NULL)");
TEST_ASSERT_TRUE(xce_core_begin_session(&(sched->core)) >= 0);
decision = dispatch_and_wait(sched, JOB_ID + 1);
TEST_ASSERT_NOT_NULL_MESSAGE(decision,
"the allocator must still produce a decision (it used to SEGV in "
"the legacy fcfs plugin's commit hook)");
TEST_ASSERT_EQUAL_INT(XCE_DECISION_DISPATCH, decision->op);
}
/*
* bug-1769: core_build_alter_schedobj() stores xce_compound_schedobj_t*
* (24 bytes) in obj->u.alters, but sched_make_alter_allocation() read them
* back as xce_sched_obj_t* (176 bytes) and ran the compound allocation on
* that -- a heap-buffer-overflow read of ->u.compound and write of ->alloc.
* No test drove a real ALTER job before: a real `{..} || {..}` resreq,
* ingested and dispatched through the live allocator thread.
*/
static void test_alter_resreq_job_dispatches_via_real_allocator(void)
{
xce_scheduler_t* sched = start_sched();
xce_queue_t* queue;
xce_core_event_t queue_event;
xce_worker_t* worker_ptrs[2];
xce_core_event_t worker_event;
xce_job_t* job_ptrs[1];
xce_core_event_t job_event;
xce_sched_job_t* sjob;
xce_decision_t* decision;
const uint64_t jobid = JOB_ID + 2;
queue = make_wire_queue("alter_queue", 1);
memset(&queue_event, 0, sizeof(queue_event));
queue_event.type = XCE_CORE_EVENT_QUEUE_UPDATE;
queue_event.u.queue_update.queue = queue;
TEST_ASSERT_TRUE(xce_core_push_events(&(sched->core), &queue_event, 1)
>= 0);
worker_ptrs[0] = make_wire_worker(WORKER_A_NAME, WORKER_SLOTS);
worker_ptrs[1] = make_wire_worker(WORKER_B_NAME, WORKER_SLOTS);
memset(&worker_event, 0, sizeof(worker_event));
worker_event.type = XCE_CORE_EVENT_WORKERS_PUSH;
worker_event.u.workers_push.nworkers = 2;
worker_event.u.workers_push.workers = worker_ptrs;
TEST_ASSERT_TRUE(xce_core_push_events(&(sched->core), &worker_event, 1)
>= 0);
job_ptrs[0] = make_push_job(
jobid, (char*)"{1*{rusage[mem=1]}} || {1*{rusage[mem=2]}}",
"e2e_user", "alter_queue", "e2e_proj", 1, 1);
memset(&job_event, 0, sizeof(job_event));
job_event.type = XCE_CORE_EVENT_JOBS_PUSH;
job_event.u.jobs_push.njobs = 1;
job_event.u.jobs_push.jobs = job_ptrs;
TEST_ASSERT_TRUE(xce_core_push_events(&(sched->core), &job_event, 1)
>= 0);
sjob = sched_get_job(&(sched->core), jobid);
TEST_ASSERT_NOT_NULL(sjob);
TEST_ASSERT_NOT_NULL(sjob->schedobj);
TEST_ASSERT_EQUAL_INT_MESSAGE(XCE_RESREQ_ALTER, sjob->schedobj->tag,
"fixture: the resreq must parse as ALTER");
TEST_ASSERT_EQUAL_INT(2, sjob->schedobj->u.alters.count);
TEST_ASSERT_TRUE(xce_core_begin_session(&(sched->core)) >= 0);
decision = dispatch_and_wait(sched, jobid);
TEST_ASSERT_NOT_NULL_MESSAGE(decision,
"the allocator must produce a decision for an ALTER job");
TEST_ASSERT_EQUAL_INT_MESSAGE(XCE_DECISION_DISPATCH, decision->op,
"the first alternative fits on either 2-slot worker");
TEST_ASSERT_NOT_NULL(decision->u.alloc);
TEST_ASSERT_EQUAL_UINT64(jobid, decision->u.alloc->jobid);
}
int main(void)
{
UNITY_BEGIN();
RUN_TEST(
test_multihost_parallel_job_allocation_spans_two_hosts_via_real_dispatch);
RUN_TEST(test_dispatch_of_job_with_unregistered_queue_does_not_crash);
RUN_TEST(test_alter_resreq_job_dispatches_via_real_allocator);
return UNITY_END();
}