fix(scheduler): stop re-dispatching running jobs (bug-1498) + resval double-free (bug-1497)
bug-1498 (re-dispatch): sched_job_is_ready() was a stub returning 1, so any job lingering in the fcfs pendJobs queue was always elected. A dispatched job was never marked non-PEND (the controller defers its schedStartJob notify to the not-yet-implemented apply pump, controller/job.c), so fcfs_update_job re-enqueued it on every job update and the election thread re-dispatched the running job — the worker got conflicting signals and the job was SIGTERM'd. - sched_job_is_ready: elect only PEND, non-suspended jobs - sched_commit_decision (dispatch): mark the job JOB_STAT_RUN on dispatch so it leaves the PEND set and is no longer re-electable - fcfs_update_job: dedup via ordq_peek before enqueue bug-1497 (double-free): sched_update's resval paths transferred ownership of rv->val to sharedInst/w->resvals but left the request pointer set, so sched_pend_update_killer double-freed it -> use-after-free in build_host_eval_ctx. Clear rv->val = NULL on transfer. Tests: new scheduler/testsuite/test_redispatch.c (8 cases): sched_job_is_ready PEND gating + dispatch-commit marks the job RUN. Full scheduler meson suite green. Live-verified: a `sleep 60` job is elected exactly once, runs the full duration and exits 0 (`completed successfully`); before the fix it was re-dispatched at ~11s and killed (SIGTERM 143). Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
lei.chen
parent
07ab05671d
commit
ee7dee0ed4
5 changed files with 381 additions and 2 deletions
|
|
@ -192,6 +192,15 @@ static int fcfs_update_job(xce_sched_plugin_t* plugin, xce_sched_job_t* j)
|
|||
|
||||
job->job = j;
|
||||
|
||||
/* bug-1498: never enqueue the same job twice. fcfs_commit/fcfs_del_job
|
||||
* removes a job on dispatch, but a job update can re-arrive before the
|
||||
* commit runs (or a dispatch-fail reset returns the job to PEND) — a
|
||||
* duplicate queue entry would be elected twice in one pass. */
|
||||
if (ordq_peek(&(q->pendJobs), (ordq_match_fn_t)job_match, j->jobid) != NULL) {
|
||||
free(job);
|
||||
return 0;
|
||||
}
|
||||
|
||||
ordq_put(&(q->pendJobs), job);
|
||||
|
||||
return 0;
|
||||
|
|
|
|||
|
|
@ -826,6 +826,12 @@ int sched_update(xce_scheduler_t* sched)
|
|||
continue;
|
||||
}
|
||||
zarray_put(&sched->sharedInst, r->idx, rv->val);
|
||||
/* Ownership of rv->val transfers to sharedInst (its killer,
|
||||
* kill_sched_resval, frees it). Clear the request's pointer so
|
||||
* sched_pend_update_killer's free_ResValList_ptr does NOT free
|
||||
* it again (bug-1497: double-free left the zarray entry
|
||||
* dangling -> heap-use-after-free in build_host_eval_ctx). */
|
||||
rv->val = NULL;
|
||||
}
|
||||
rc = 0;
|
||||
break;
|
||||
|
|
@ -846,6 +852,11 @@ int sched_update(xce_scheduler_t* sched)
|
|||
continue;
|
||||
}
|
||||
zarray_put(&w->resvals, r->idx, rv->val);
|
||||
/* Ownership of rv->val transfers to w->resvals (killer
|
||||
* kill_sched_resval frees it). Clear the request pointer so
|
||||
* sched_pend_update_killer does not double-free it
|
||||
* (bug-1497: use-after-free in build_host_eval_ctx). */
|
||||
rv->val = NULL;
|
||||
}
|
||||
rc = 0;
|
||||
break;
|
||||
|
|
@ -1263,8 +1274,27 @@ xce_sched_project_t* sched_get_project(xce_scheduler_t* sched, char* name)
|
|||
return make_sched_project(sched, name);
|
||||
}
|
||||
|
||||
/* FIXME this is just stub, we need to add more */
|
||||
int sched_job_is_ready(xce_sched_job_t* j) { return 1; }
|
||||
/*
|
||||
* sched_job_is_ready — a job is electable only while it is PENDING and not
|
||||
* suspended. Once dispatched (RUN), finished (EXIT/DONE), or suspended
|
||||
* (PSUSP/USUSP/SSUSP) it must never be re-elected. bug-1498: the previous
|
||||
* stub returned 1 unconditionally, so a running job that lingered in the fcfs
|
||||
* pendJobs queue (re-added on every job update because the scheduler was never
|
||||
* told the job had left PEND) was re-dispatched to the worker and killed.
|
||||
*/
|
||||
int sched_job_is_ready(xce_sched_job_t* j)
|
||||
{
|
||||
if (j == NULL) {
|
||||
return 0;
|
||||
}
|
||||
if ((j->state & JOB_STAT_PEND) == 0) {
|
||||
return 0; /* already dispatched/running or terminal */
|
||||
}
|
||||
if (j->state & (JOB_STAT_PSUSP | JOB_STAT_USUSP | JOB_STAT_SSUSP)) {
|
||||
return 0; /* suspended — not electable */
|
||||
}
|
||||
return 1;
|
||||
}
|
||||
|
||||
/* FIXME this is just stub, we need to add more */
|
||||
int sched_queue_is_ready(xce_sched_queue_t* q) { return 1; }
|
||||
|
|
@ -1284,6 +1314,16 @@ int sched_commit_decision(xce_scheduler_t* sched, xce_decision_t* decision)
|
|||
job->allocs = decision->u.alloc;
|
||||
job->accounts = make_job_account(job->allocs);
|
||||
|
||||
/* bug-1498: mark the job RUN the moment it is dispatched so it leaves
|
||||
* the PEND set — fcfs_update_job re-enqueues only PEND jobs and
|
||||
* sched_job_is_ready elects only PEND jobs. Without this the scheduler
|
||||
* is never told the job stopped pending (the controller's
|
||||
* record_job_start_fact defers the AFTER_EVENT schedStartJob notify to
|
||||
* the not-yet-implemented apply pump, controller/job.c) and re-dispatches
|
||||
* the running job. A later controller state update corrects it: PEND on
|
||||
* dispatch-fail (re-electable), EXIT/DONE on finish. */
|
||||
job->state = JOB_STAT_RUN;
|
||||
|
||||
return sched_commit_job_alloc(sched, job);
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -56,12 +56,24 @@ sched_job_array_wire_e2e_test = executable('sched_job_array_wire_e2e_test',
|
|||
dependencies: [scheduler_test_deps],
|
||||
install: false)
|
||||
|
||||
# White-box regression test for bug-1498 (scheduler re-dispatches a job it
|
||||
# already dispatched): sched_job_is_ready PEND-gating + dispatch-commit marks
|
||||
# the job RUN. It #includes ../src/scheduler.c directly, so it is a standalone
|
||||
# TU with its own main() — built as a separate executable.
|
||||
sched_redispatch_test = executable('sched_redispatch_test',
|
||||
['test_redispatch.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',
|
||||
env: ['ASAN_OPTIONS=detect_leaks=0'])
|
||||
test('scheduler job-update test', sched_job_update_test, suite: 'scheduler',
|
||||
env: ['ASAN_OPTIONS=detect_leaks=0'])
|
||||
test('scheduler redispatch test', sched_redispatch_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',
|
||||
|
|
|
|||
242
src/scheduler/testsuite/test_redispatch.c
Normal file
242
src/scheduler/testsuite/test_redispatch.c
Normal file
|
|
@ -0,0 +1,242 @@
|
|||
/*
|
||||
* test_redispatch.c - white-box regression tests for bug-1498: the scheduler
|
||||
* re-dispatched a job it had already dispatched (running job killed by a
|
||||
* duplicate dispatch).
|
||||
*
|
||||
* Root cause: sched_job_is_ready() was a stub returning 1 unconditionally, so
|
||||
* a job lingering in the fcfs pendJobs queue was always elected; and a
|
||||
* dispatched job was never marked non-PEND, so fcfs_update_job re-enqueued it
|
||||
* on every job update (the controller defers its schedStartJob notify).
|
||||
*
|
||||
* Strategy: pull the REAL scheduler.c into this TU (same white-box pattern as
|
||||
* test_job_update.c pulling schedobj.c), stub the scheduler-internal externs
|
||||
* defined in other .c files not linked here, and drive:
|
||||
* - sched_job_is_ready: PEND electable; RUN/EXIT/DONE/suspended/NULL not.
|
||||
* - sched_commit_decision(XCE_DECISION_DISPATCH): marks the job RUN so it
|
||||
* leaves the PEND set and is no longer electable.
|
||||
*/
|
||||
#define UNIT_TEST 1
|
||||
#include <errno.h>
|
||||
#include <stdio.h>
|
||||
#include <stdlib.h>
|
||||
#include <string.h>
|
||||
#include <stdatomic.h>
|
||||
|
||||
#include <unity/unity.h>
|
||||
|
||||
#include "kvmgr.h"
|
||||
#include "scheduler.h"
|
||||
|
||||
/* Pull the real implementation under test. */
|
||||
#include "../src/scheduler.c"
|
||||
|
||||
/* ── Stubs for scheduler.c externs defined in other .c files ─────────────── */
|
||||
|
||||
/* plugin.c — plugin dispatch table & lifecycle (not linked here). */
|
||||
int plugin_job_update(void* s, xce_sched_job_t* j) { (void)s; (void)j; return 0; }
|
||||
int plugin_queue_update(void* s, xce_sched_queue_t* q) { (void)s; (void)q; return 0; }
|
||||
int plugin_worker_update(void* s, xce_sched_worker_t* w) { (void)s; (void)w; return 0; }
|
||||
int plugin_job_state_change(void* s, xce_sched_job_t* j, uint32_t st)
|
||||
{
|
||||
(void)s; (void)j; (void)st; return 0;
|
||||
}
|
||||
int plugin_job_rusage_update(void* s, xce_sched_job_t* j, xce_job_rusages_t* r)
|
||||
{
|
||||
(void)s; (void)j; (void)r; return 0;
|
||||
}
|
||||
int plugin_commit_decision(void* s, xce_decision_t* d) { (void)s; (void)d; return 0; }
|
||||
int plugin_begin_session(void* s) { (void)s; return 0; }
|
||||
int plugin_end_session(void* s) { (void)s; return 0; }
|
||||
int plugin_config_update(void* s, xce_conf4sched_t* c) { (void)s; (void)c; return 0; }
|
||||
int plugin_init(void* s) { (void)s; return 0; }
|
||||
void plugin_unregister(void* s) { (void)s; }
|
||||
int plugin_job_election(void* s, xce_sched_job_t* j) { (void)s; (void)j; return 0; }
|
||||
|
||||
/* schedobj.c / resource.c / client helpers — stubbed for these tests. */
|
||||
void kill_sched_job(void* arg) { (void)arg; }
|
||||
void kill_sched_queue(void* arg) { (void)arg; }
|
||||
void kill_sched_worker(void* arg) { (void)arg; }
|
||||
void kill_sched_user(void* arg) { (void)arg; }
|
||||
void kill_sched_project(void* arg) { (void)arg; }
|
||||
int sched_commit_job_alloc(xce_scheduler_t* s, xce_sched_job_t* j)
|
||||
{
|
||||
(void)s; (void)j; return 0;
|
||||
}
|
||||
int sched_charge_account(xce_scheduler_t* s, xce_job_account_t* a)
|
||||
{
|
||||
(void)s; (void)a; return 0;
|
||||
}
|
||||
int sched_uncharge_account(xce_scheduler_t* s, xce_job_account_t* a)
|
||||
{
|
||||
(void)s; (void)a; return 0;
|
||||
}
|
||||
int sched_apply_job_rusage(xce_scheduler_t* s, xce_job_rusages_t* r)
|
||||
{
|
||||
(void)s; (void)r; return -ENOENT;
|
||||
}
|
||||
int sched_update_existing_job(xce_scheduler_t* s, xce_job_t* j)
|
||||
{
|
||||
(void)s; (void)j; return -ENOENT;
|
||||
}
|
||||
int sched_update_existing_queue(xce_scheduler_t* s, xce_queue_t* q)
|
||||
{
|
||||
(void)s; (void)q; return -ENOENT;
|
||||
}
|
||||
|
||||
/* schedobj.c / decision.c / plugins — additional externs scheduler.c
|
||||
* references (defined in .c files not linked here). */
|
||||
int sched_ctx_init(xce_sched_ctx_t* ctx, void* arg, xce_sched_obj_t* obj)
|
||||
{
|
||||
(void)ctx; (void)arg; (void)obj; return 0;
|
||||
}
|
||||
void sched_ctx_fini(xce_sched_ctx_t* ctx) { (void)ctx; }
|
||||
void sched_ctx_set_primary(xce_sched_ctx_t* ctx, xce_primary_resreq_t* p)
|
||||
{
|
||||
(void)ctx; (void)p;
|
||||
}
|
||||
xce_decision_t* make_pend_decision(xce_pendreason_t* r) { (void)r; return NULL; }
|
||||
xce_decision_t* make_dispatch_decision(xce_job_alloc_t* a) { (void)a; return NULL; }
|
||||
int register_fcfs(void* s) { (void)s; return 0; }
|
||||
int register_limit(void* s) { (void)s; return 0; }
|
||||
int register_seqjob_allocator(void* s) { (void)s; return 0; }
|
||||
int register_parjob_allocator(void* s) { (void)s; return 0; }
|
||||
void kill_sched_resval(xce_val_t* v) { (void)v; }
|
||||
xce_sched_job_t* make_sched_job(void* a, xce_job_t* j) { (void)a; (void)j; return NULL; }
|
||||
xce_sched_queue_t* make_sched_queue(void* a, xce_queue_t* q) { (void)a; (void)q; return NULL; }
|
||||
xce_sched_worker_t* make_sched_worker(void* a, xce_worker_t* w) { (void)a; (void)w; return NULL; }
|
||||
xce_sched_user_t* make_sched_user(void* a, char* n) { (void)a; (void)n; return NULL; }
|
||||
xce_sched_project_t* make_sched_project(void* a, char* n) { (void)a; (void)n; return NULL; }
|
||||
void sched_worker_release_host_ctx(xce_sched_worker_t* w) { (void)w; }
|
||||
|
||||
void setUp(void) { return; }
|
||||
void tearDown(void) { return; }
|
||||
|
||||
/* Note: sched_get_job / sched_add_job / sched_get_queue / ... are defined by
|
||||
* scheduler.c itself (real kvmgr-backed) and come in via the #include — no
|
||||
* stubs needed. They operate on sched->jobs etc., which fixture_init creates. */
|
||||
|
||||
/* ── Fixture ─────────────────────────────────────────────────────────────── */
|
||||
|
||||
static void fixture_init(xce_scheduler_t* sched)
|
||||
{
|
||||
kvmgr_params_t jobs_params = { .key_type = KVMGR_KEY_INT,
|
||||
.nslots = 59,
|
||||
.delta = 16,
|
||||
.datakiller = kill_sched_job };
|
||||
|
||||
memset(sched, 0, sizeof(*sched));
|
||||
resreq_init(1);
|
||||
xstring_table_init(1024 * 16);
|
||||
|
||||
sched->jobs = makeInstance(&xkvmgr_type, &jobs_params);
|
||||
}
|
||||
|
||||
static xce_sched_job_t* seed_job(xce_scheduler_t* sched, uint64_t jobid,
|
||||
uint32_t state)
|
||||
{
|
||||
xce_sched_job_t* j = (xce_sched_job_t*)calloc(1, sizeof(*j));
|
||||
TEST_ASSERT_NOT_NULL(j);
|
||||
j->jobid = jobid;
|
||||
j->state = state;
|
||||
sched_add_job(sched, j);
|
||||
return j;
|
||||
}
|
||||
|
||||
/* ── sched_job_is_ready ──────────────────────────────────────────────────── */
|
||||
|
||||
static void test_ready_null_not_ready(void)
|
||||
{
|
||||
TEST_ASSERT_EQUAL_INT(0, sched_job_is_ready(NULL));
|
||||
}
|
||||
|
||||
static void test_ready_pend_is_ready(void)
|
||||
{
|
||||
xce_sched_job_t j = { .state = JOB_STAT_PEND };
|
||||
TEST_ASSERT_EQUAL_INT(1, sched_job_is_ready(&j));
|
||||
}
|
||||
|
||||
static void test_ready_run_not_ready(void)
|
||||
{
|
||||
xce_sched_job_t j = { .state = JOB_STAT_RUN };
|
||||
TEST_ASSERT_EQUAL_INT(0, sched_job_is_ready(&j));
|
||||
}
|
||||
|
||||
static void test_ready_exit_not_ready(void)
|
||||
{
|
||||
xce_sched_job_t j = { .state = JOB_STAT_EXIT };
|
||||
TEST_ASSERT_EQUAL_INT(0, sched_job_is_ready(&j));
|
||||
}
|
||||
|
||||
static void test_ready_done_not_ready(void)
|
||||
{
|
||||
xce_sched_job_t j = { .state = JOB_STAT_DONE };
|
||||
TEST_ASSERT_EQUAL_INT(0, sched_job_is_ready(&j));
|
||||
}
|
||||
|
||||
static void test_ready_suspended_not_ready(void)
|
||||
{
|
||||
xce_sched_job_t j = { .state = JOB_STAT_PEND | JOB_STAT_USUSP };
|
||||
TEST_ASSERT_EQUAL_INT(0, sched_job_is_ready(&j));
|
||||
}
|
||||
|
||||
/* ── sched_commit_decision dispatch marks the job RUN ────────────────────── */
|
||||
|
||||
static void test_dispatch_commit_marks_job_run(void)
|
||||
{
|
||||
xce_scheduler_t sched;
|
||||
xce_sched_job_t* job;
|
||||
xce_job_alloc_t alloc;
|
||||
xce_decision_t decision;
|
||||
|
||||
fixture_init(&sched);
|
||||
|
||||
job = seed_job(&sched, 42, JOB_STAT_PEND);
|
||||
|
||||
memset(&alloc, 0, sizeof(alloc));
|
||||
alloc.jobid = 42;
|
||||
|
||||
memset(&decision, 0, sizeof(decision));
|
||||
decision.op = XCE_DECISION_DISPATCH;
|
||||
decision.u.alloc = &alloc;
|
||||
|
||||
TEST_ASSERT_EQUAL_INT(0, sched_commit_decision(&sched, &decision));
|
||||
|
||||
/* The dispatched job leaves PEND: state is RUN and it is not electable. */
|
||||
TEST_ASSERT_EQUAL_UINT(JOB_STAT_RUN, job->state);
|
||||
TEST_ASSERT_EQUAL_INT(0, sched_job_is_ready(job));
|
||||
}
|
||||
|
||||
static void test_dispatch_commit_null_alloc_unchanged(void)
|
||||
{
|
||||
xce_scheduler_t sched;
|
||||
xce_sched_job_t* job;
|
||||
xce_decision_t decision;
|
||||
|
||||
fixture_init(&sched);
|
||||
job = seed_job(&sched, 43, JOB_STAT_PEND);
|
||||
|
||||
memset(&decision, 0, sizeof(decision));
|
||||
decision.op = XCE_DECISION_PEND;
|
||||
decision.u.reason = NULL;
|
||||
|
||||
TEST_ASSERT_EQUAL_INT(0, sched_commit_decision(&sched, &decision));
|
||||
/* Non-dispatch decision must not touch the job's state. */
|
||||
TEST_ASSERT_EQUAL_UINT(JOB_STAT_PEND, job->state);
|
||||
}
|
||||
|
||||
/* ── Runner ──────────────────────────────────────────────────────────────── */
|
||||
|
||||
int main(int argc, char* argv[])
|
||||
{
|
||||
(void)argc; (void)argv;
|
||||
UNITY_BEGIN();
|
||||
RUN_TEST(test_ready_null_not_ready, 0);
|
||||
RUN_TEST(test_ready_pend_is_ready, 0);
|
||||
RUN_TEST(test_ready_run_not_ready, 0);
|
||||
RUN_TEST(test_ready_exit_not_ready, 0);
|
||||
RUN_TEST(test_ready_done_not_ready, 0);
|
||||
RUN_TEST(test_ready_suspended_not_ready, 0);
|
||||
RUN_TEST(test_dispatch_commit_marks_job_run, 0);
|
||||
RUN_TEST(test_dispatch_commit_null_alloc_unchanged, 0);
|
||||
return UNITY_END();
|
||||
}
|
||||
Loading…
Add table
Add a link
Reference in a new issue