From f51a19d3d5a6c5f27d5c5842774ecebc50029733 Mon Sep 17 00:00:00 2001 From: Rongsong Shen Date: Sat, 10 Oct 2026 08:22:20 +0000 Subject: [PATCH] fix(scheduler): hold jobs of inactive queues at dispatch (bug-1938) sched_queue_is_ready() in core.c was a `return 1` stub, so the picker (builtin_fcfs and the legacy fcfs) took jobs from every queue and admin queue inactivation never reached dispatch. It now returns false for a queue with QUEUE_STAT_INACT_ADMIN or QUEUE_STAT_INACT_WINDOW set, so an inactive queue's jobs stay pending. QUEUE_STAT_CLOSED is deliberately not a gate (LSF semantics: closing only blocks new submissions; already-pending jobs still dispatch). The status reaches the core through the same queue add/update paths in both drivers, so live and sim behave identically. test_queue_admin_control.c covers open, closed and inactive queues on both fixture shapes. Co-Authored-By: Claude Opus 5.5 (1M context) --- src/scheduler/src/core.c | 22 +- src/scheduler/src/plugins/builtin_fcfs.c | 11 +- src/scheduler/testsuite/meson.build | 22 ++ .../testsuite/test_queue_admin_control.c | 290 ++++++++++++++++++ 4 files changed, 337 insertions(+), 8 deletions(-) create mode 100644 src/scheduler/testsuite/test_queue_admin_control.c diff --git a/src/scheduler/src/core.c b/src/scheduler/src/core.c index 92e150ba..f9b381a3 100644 --- a/src/scheduler/src/core.c +++ b/src/scheduler/src/core.c @@ -645,5 +645,23 @@ xce_sched_project_t* sched_get_project(xce_sched_core_t* core, char* name) /* FIXME this is just stub, we need to add more */ int sched_job_is_ready(xce_sched_job_t* j) { return 1; } -/* FIXME this is just stub, we need to add more */ -int sched_queue_is_ready(xce_sched_queue_t* q) { return 1; } +/* + * sched_queue_is_ready - may the picker take jobs from this queue? (bug-1938) + * + * LSF semantics: an inactive queue (inactivated by an admin, or outside its + * dispatch/run window) dispatches nothing, so its pending jobs stay pending. + * QUEUE_STAT_CLOSED does not gate dispatch: closing a queue only stops new + * submissions (refused by the controller), and jobs already pending in it + * still run. The status is the controller's, copied in by every queue + * add/update path (core_build_queue, core_upsert_queue, make_sched_queue), + * so the live and simulation drivers read the same bits. + */ +int sched_queue_is_ready(xce_sched_queue_t* q) +{ + if (q == NULL) { + return 0; + } + + return (q->status & (QUEUE_STAT_INACT_ADMIN | QUEUE_STAT_INACT_WINDOW)) + == 0; +} diff --git a/src/scheduler/src/plugins/builtin_fcfs.c b/src/scheduler/src/plugins/builtin_fcfs.c index 1ca4a543..76591431 100644 --- a/src/scheduler/src/plugins/builtin_fcfs.c +++ b/src/scheduler/src/plugins/builtin_fcfs.c @@ -827,12 +827,11 @@ int builtin_fcfs_begin_session(void* picker) * and at exit -- both real, safe, per-call cancellation points (mirroring * sched_job_picker()'s own pthread_testcancel() immediately before every * next_job() call, which remains in place too -- belt-and-suspenders, not - * redundant dead code). This does not weaken cancel-safety in practice - * today: sched_job_is_ready()/sched_queue_is_ready() are unconditional - * `return 1` stubs (scheduler.c), so this walk always returns on the very - * first candidate examined -- fcfs_elect_job()'s own per-iteration - * testcancel calls are equally never exercised mid-walk today, for the - * same reason. + * redundant dead code). This does not weaken cancel-safety in practice: + * sched_job_is_ready() is an unconditional `return 1` stub (core.c), and + * sched_queue_is_ready() only skips inactive queues (bug-1938), so a walk + * skips at most every queue once, a bounded, non-blocking scan, before it + * returns on the first job of the first ready queue. */ xce_sched_job_t* builtin_fcfs_next_job(void* picker) { diff --git a/src/scheduler/testsuite/meson.build b/src/scheduler/testsuite/meson.build index 02edbb96..cf7ae00d 100644 --- a/src/scheduler/testsuite/meson.build +++ b/src/scheduler/testsuite/meson.build @@ -3048,3 +3048,25 @@ test('scheduler trace equivalence test', sched_trace_equivalence_test, suite: 'scheduler', env: ['ASAN_OPTIONS=detect_leaks=0'], depends: scheduler) + +# bug-1938: admin queue control reaches dispatch (an inactive queue holds its +# pending jobs, a closed queue still dispatches them) on both the +# normal-mode and sim-mode fixture shapes. Real-scheduler_srcs link +# technique of sched_sim_contention_limit_driver_parity_test. +sched_queue_admin_control_test_srcs = ['test_queue_admin_control.c'] +foreach f : scheduler_srcs + if f != 'src/main.c' + sched_queue_admin_control_test_srcs += ['../' + f] + endif +endforeach + +sched_queue_admin_control_test = executable( + 'sched_queue_admin_control_test', + sched_queue_admin_control_test_srcs, + include_directories:[incdir], + dependencies: [scheduler_test_deps], + install: false) + +test('scheduler queue admin control test', + sched_queue_admin_control_test, suite: 'scheduler', + env: ['ASAN_OPTIONS=detect_leaks=0']) diff --git a/src/scheduler/testsuite/test_queue_admin_control.c b/src/scheduler/testsuite/test_queue_admin_control.c new file mode 100644 index 00000000..582efd06 --- /dev/null +++ b/src/scheduler/testsuite/test_queue_admin_control.c @@ -0,0 +1,290 @@ +/* + * test_queue_admin_control.c - bug-1938: admin queue control must reach + * dispatch, identically on the normal-mode and sim-mode drivers. + * + * LSF semantics, which the controller's queue control (controller/queue.c) + * follows: + * - QUEUE_STAT_INACT_ADMIN ("inactive"): the queue dispatches nothing; + * its pending jobs stay pending. + * - QUEUE_STAT_CLOSED ("closed"): the queue accepts no new submissions, + * but jobs already pending in it are still dispatched. + * + * The fixture has three queues (open, closed, inactive), one pending job in + * each and a worker big enough for all three, so capacity never decides + * the outcome: only the queue state does. Both drivers' fixtures are built + * the way test_sim_contention_limit_driver_parity.c builds them (real + * scheduler_srcs, real picker/allocator threads, the sim side bootstrapped + * through sim_loop.c's own sim_bootstrap_* entry points), `static` fixtures + * and exactly one RUN_TEST() (bug-1598/1632/1763 fixture-safety convention). + */ +#include +#include +#include +#include +#include + +#include + +#include "controller.capnp.h" +#include "decision.h" +#include "plugin.h" +#include "preason.h" +#include "resobj.h" +#include "sched/types.h" +#include "scheduler.h" +#include "sim.h" +#include "xstring.h" + +extern void sim_bootstrap_picker(xce_sched_sim_driver_t* sim); +extern void sim_bootstrap_seqjob_allocator(xce_sched_sim_driver_t* sim); +extern void sim_bootstrap_parjob_allocator(xce_sched_sim_driver_t* sim); + +#define WORKER_NAME "qctl_worker" +#define QUEUE_OPEN_N "qctl_open" +#define QUEUE_CLOSED_N "qctl_closed" +#define QUEUE_INACT_N "qctl_inactive" +#define TEST_RESREQ ((char*)"select[type==any] rusage[mem=1]") + +#define JOB_OPEN_ID ((uint64_t)193801ULL) +#define JOB_CLOSED_ID ((uint64_t)193802ULL) +#define JOB_INACT_ID ((uint64_t)193803ULL) + +void setUp(void) { return; } +void tearDown(void) { return; } + +static void register_slots_resource(var resmgr) +{ + xce_resource_t res; + + memset(&res, 0, sizeof(res)); + res.name = strdup("slots"); + res.valtype = XCE_VAL_NUMERIC; + res.ordtype = XCE_ORD_DECR; + res.flags = XCE_RESF_LIMIT | XCE_RESF_CONSUMABLE | XCE_RESF_RELEASE + | XCE_RESF_DEFAULT_HOST; + + TEST_ASSERT_NOT_NULL(makeResource(resmgr, &res)); +} + +static xce_sched_worker_t* make_worker(xce_sched_core_t* core, double slots) +{ + xce_sched_worker_t* w = (xce_sched_worker_t*)calloc(1, sizeof(*w)); + xce_resource_t* r = NULL; + xce_val_t* val = NULL; + int rc; + + TEST_ASSERT_NOT_NULL(w); + w->name = strdup(WORKER_NAME); + w->status = 0; + w->idx = -1; + + zarray_init(&(w->resvals), 128, (zarray_item_killer_t)kill_sched_resval); + zarray_init(&(w->resInUse), 128, (zarray_item_killer_t)kill_sched_resval); + + rc = key_find(core->resmgr, (void*)"slots", (void**)&r); + TEST_ASSERT_TRUE(rc >= 0); + TEST_ASSERT_NOT_NULL(r); + + val = (xce_val_t*)calloc(1, sizeof(*val)); + TEST_ASSERT_NOT_NULL(val); + val->which = Value_u_fvalue; + val->u.fvalue = slots; + zarray_put(&(w->resvals), r->idx, val); + + return w; +} + +static void add_queue(xce_sched_core_t* core, xce_sched_plugin_t* picker, + const char* name, int status) +{ + xce_sched_queue_t* q = (xce_sched_queue_t*)calloc(1, sizeof(*q)); + int rc; + + TEST_ASSERT_NOT_NULL(q); + q->name = makeString((char*)name); + q->status = status; + rc = sched_add_queue(core, q); + TEST_ASSERT_TRUE(rc >= 0); + picker->picker_iface->on_queue_update(picker, q); +} + +static void push_job(xce_sched_core_t* core, xce_scheduler_t* job_ctor_wrapper, + xce_sched_plugin_t* picker, uint64_t jobid, + const char* queue, uint64_t queue_time) +{ + job_submit_data_t submit; + xce_job_t j; + xce_sched_job_t* job; + int rc; + + memset(&submit, 0, sizeof(submit)); + submit.queue = (char*)queue; + submit.user = (char*)"qctl_user"; + submit.project_name = (char*)"default"; + submit.resreq = TEST_RESREQ; + submit.mintasks = 1; + submit.maxtasks = 1; + + memset(&j, 0, sizeof(j)); + j.jobid = jobid; + j.state = JOB_STAT_PEND; + j.sub_time = queue_time; + j.queue_time = queue_time; + j.submit = &submit; + + job = make_sched_job(job_ctor_wrapper, &j); + TEST_ASSERT_NOT_NULL(job); + TEST_ASSERT_EQUAL_INT(XCE_RESREQ_PRIMARY, job->schedobj->tag); + job->schedobj->u.primary->resreq->minTasks = 1; + job->schedobj->u.primary->resreq->maxTasks = 1; + + rc = sched_add_job(core, job); + TEST_ASSERT_TRUE(rc >= 0); + picker->picker_iface->on_job_push(picker, job); +} + +/* run_round - one full picker/allocator session; returns how many decisions + * the session produced before its end-of-session NULL marker. Draining up + * to the marker (instead of expecting a fixed count) keeps a wrong count a + * plain assertion failure rather than a hang. */ +static int run_round(xce_sched_core_t* core) +{ + pthread_t picker_tid; + pthread_t allocator_tid; + xce_decision_t* d; + int n = 0; + int rc; + + rc = pthread_create(&picker_tid, NULL, sched_job_picker, core); + TEST_ASSERT_EQUAL_INT(0, rc); + rc = pthread_create(&allocator_tid, NULL, sched_allocator, core); + TEST_ASSERT_EQUAL_INT(0, rc); + + rc = xce_core_go(core); + TEST_ASSERT_EQUAL_INT(0, rc); + + for (;;) { + d = NULL; + cbuf_get(&(core->decisions), &d); + if (d == NULL) { + break; + } + n++; + } + + rc = xce_core_end_session(core); + TEST_ASSERT_EQUAL_INT(0, rc); + + return n; +} + +static void populate(xce_sched_core_t* core, xce_scheduler_t* job_wrapper, + xce_sched_plugin_t* picker) +{ + int rc; + + add_queue(core, picker, QUEUE_OPEN_N, QUEUE_STAT_OPEN); + add_queue(core, picker, QUEUE_CLOSED_N, QUEUE_STAT_CLOSED); + add_queue(core, picker, QUEUE_INACT_N, QUEUE_STAT_INACT_ADMIN); + + rc = sched_add_worker(core, make_worker(core, 8.0)); + TEST_ASSERT_TRUE(rc >= 0); + + push_job(core, job_wrapper, picker, JOB_OPEN_ID, QUEUE_OPEN_N, 1000ULL); + push_job(core, job_wrapper, picker, JOB_CLOSED_ID, QUEUE_CLOSED_N, + 1001ULL); + push_job(core, job_wrapper, picker, JOB_INACT_ID, QUEUE_INACT_N, 1002ULL); +} + +static void assert_outcome(xce_sched_core_t* core, int ndecisions, + const char* mode) +{ + char msg[160]; + + snprintf(msg, sizeof(msg), + "%s: only the open and closed queues' jobs may be decided " + "(an inactive queue dispatches nothing)", + mode); + TEST_ASSERT_EQUAL_INT_MESSAGE(2, ndecisions, msg); + + snprintf(msg, sizeof(msg), "%s: open queue job must dispatch", mode); + TEST_ASSERT_EQUAL_UINT_MESSAGE(JOB_STAT_RUN, + sched_get_job(core, JOB_OPEN_ID)->state, + msg); + snprintf(msg, sizeof(msg), + "%s: closed queue job already pending must still dispatch", + mode); + TEST_ASSERT_EQUAL_UINT_MESSAGE(JOB_STAT_RUN, + sched_get_job(core, JOB_CLOSED_ID)->state, + msg); + snprintf(msg, sizeof(msg), "%s: inactive queue job must stay pending", + mode); + TEST_ASSERT_EQUAL_UINT_MESSAGE(JOB_STAT_PEND, + sched_get_job(core, JOB_INACT_ID)->state, + msg); +} + +static xce_scheduler_t g_normal; +static xce_sched_sim_driver_t g_sim; +static xce_scheduler_t g_sim_job_wrapper; + +static void test_inactive_queue_holds_jobs_closed_queue_dispatches(void) +{ + xce_sched_plugin_t* fcfs_normal = NULL; + xce_sched_plugin_t* fcfs_sim = NULL; + int n_normal; + int n_sim; + int rc; + + /* normal-mode-shaped fixture */ + memset(&g_normal, 0, sizeof(g_normal)); + rc = xce_sched_core_init(&(g_normal.core)); + TEST_ASSERT_EQUAL_INT(0, rc); + register_slots_resource(g_normal.core.resmgr); + + fcfs_normal = builtin_fcfs_new_plugin(); + TEST_ASSERT_NOT_NULL(fcfs_normal); + rc = plugin_register(&g_normal, fcfs_normal); + TEST_ASSERT_TRUE(rc >= 0); + g_normal.core.job_pick_plugin = (char*)"builtin_fcfs"; + rc = register_seqjob_allocator(&g_normal); + TEST_ASSERT_TRUE(rc >= 0); + rc = register_parjob_allocator(&g_normal); + TEST_ASSERT_TRUE(rc >= 0); + + populate(&(g_normal.core), &g_normal, fcfs_normal); + n_normal = run_round(&(g_normal.core)); + + /* sim-mode-shaped fixture */ + memset(&g_sim, 0, sizeof(g_sim)); + rc = xce_sched_core_init(&(g_sim.core)); + TEST_ASSERT_EQUAL_INT(0, rc); + register_slots_resource(g_sim.core.resmgr); + + sim_bootstrap_picker(&g_sim); + rc = key_find(g_sim.core.plugins, "builtin_fcfs", (void**)&fcfs_sim); + TEST_ASSERT_TRUE(rc >= 0); + TEST_ASSERT_NOT_NULL(fcfs_sim); + sim_bootstrap_seqjob_allocator(&g_sim); + sim_bootstrap_parjob_allocator(&g_sim); + + memset(&g_sim_job_wrapper, 0, sizeof(g_sim_job_wrapper)); + rc = xce_sched_core_init(&(g_sim_job_wrapper.core)); + TEST_ASSERT_EQUAL_INT(0, rc); + g_sim_job_wrapper.core.resmgr = g_sim.core.resmgr; + g_sim_job_wrapper.core.workers = g_sim.core.workers; + g_sim_job_wrapper.core.queues = g_sim.core.queues; + + populate(&(g_sim.core), &g_sim_job_wrapper, fcfs_sim); + n_sim = run_round(&(g_sim.core)); + + assert_outcome(&(g_normal.core), n_normal, "normal mode"); + assert_outcome(&(g_sim.core), n_sim, "sim mode"); +} + +int main(void) +{ + UNITY_BEGIN(); + RUN_TEST(test_inactive_queue_holds_jobs_closed_queue_dispatches); + return UNITY_END(); +}