From 1668a43998dae00d4057a319d2bcda924d916e86 Mon Sep 17 00:00:00 2001 From: Rongsong Shen Date: Sat, 10 Oct 2026 08:24:58 +0000 Subject: [PATCH] fix(scheduler): exclude admin-closed hosts from new placements (bug-1939) No scheduler code read XCE_WORKER_CLOSED / XCE_WORKER_CLOSED_BY_ADMIN: sched_host_iter_next() (host_iter.c), the candidate-host source for both the seqjob and the parjob allocator, filtered only on the select expression and resource availability, so jobs kept landing on a host an admin had closed (parjob's skip_bad_worker_ hook is empty and unused). sched_host_iter_next() now skips a host with either closed bit before opening its trace scope, which covers every new placement for serial and parallel jobs on the live and simulation drivers. Accounting of jobs already running on the host is unchanged. No new pend-reason code: a closed host is simply not a candidate. test_host_admin_close.c covers a serial and a parallel job on both fixture shapes. Co-Authored-By: Claude Opus 5.5 (1M context) --- src/scheduler/src/host_iter.c | 13 + src/scheduler/testsuite/meson.build | 21 ++ .../testsuite/test_host_admin_close.c | 280 ++++++++++++++++++ 3 files changed, 314 insertions(+) create mode 100644 src/scheduler/testsuite/test_host_admin_close.c diff --git a/src/scheduler/src/host_iter.c b/src/scheduler/src/host_iter.c index 11a8ca5c..b1f69a9f 100644 --- a/src/scheduler/src/host_iter.c +++ b/src/scheduler/src/host_iter.c @@ -139,6 +139,19 @@ xce_sched_worker_t* sched_host_iter_next(xce_host_iter_t* iter) continue; } + /* bug-1939: a closed host (the controller's host close sets + * XCE_WORKER_CLOSED | XCE_WORKER_CLOSED_BY_ADMIN) takes no new + * jobs. This is the one candidate-host source of both the seqjob + * and the parjob allocator on either driver, so excluding it here + * covers every new placement; jobs already running on the host are + * accounted elsewhere and stay untouched. Skipped before the trace + * host scope opens: a closed host is not a candidate, so it gets + * no verdict and no pend reason. */ + if ((w->status & (XCE_WORKER_CLOSED | XCE_WORKER_CLOSED_BY_ADMIN)) + != 0) { + continue; + } + /* Decision trace: the pre-filter's own host scope. A rejection * below ends it with the host's verdict; a host returned to the * caller leaves it open for the caller's own host_begin, which diff --git a/src/scheduler/testsuite/meson.build b/src/scheduler/testsuite/meson.build index cf7ae00d..7a35f469 100644 --- a/src/scheduler/testsuite/meson.build +++ b/src/scheduler/testsuite/meson.build @@ -3070,3 +3070,24 @@ sched_queue_admin_control_test = executable( test('scheduler queue admin control test', sched_queue_admin_control_test, suite: 'scheduler', env: ['ASAN_OPTIONS=detect_leaks=0']) + +# bug-1939: a host closed by an admin gets no new jobs from either the +# serial or the parallel allocator, on both the normal-mode and sim-mode +# fixture shapes. Same real-scheduler_srcs link technique as above. +sched_host_admin_close_test_srcs = ['test_host_admin_close.c'] +foreach f : scheduler_srcs + if f != 'src/main.c' + sched_host_admin_close_test_srcs += ['../' + f] + endif +endforeach + +sched_host_admin_close_test = executable( + 'sched_host_admin_close_test', + sched_host_admin_close_test_srcs, + include_directories:[incdir], + dependencies: [scheduler_test_deps], + install: false) + +test('scheduler host admin close test', + sched_host_admin_close_test, suite: 'scheduler', + env: ['ASAN_OPTIONS=detect_leaks=0']) diff --git a/src/scheduler/testsuite/test_host_admin_close.c b/src/scheduler/testsuite/test_host_admin_close.c new file mode 100644 index 00000000..469b8318 --- /dev/null +++ b/src/scheduler/testsuite/test_host_admin_close.c @@ -0,0 +1,280 @@ +/* + * test_host_admin_close.c - bug-1939: a host closed by an admin + * (XCE_WORKER_CLOSED | XCE_WORKER_CLOSED_BY_ADMIN, set by the controller's + * host control, controller/worker.c) gets no new jobs, from either the + * serial (seqjob) or the parallel (parjob) allocator, identically on the + * normal-mode and sim-mode drivers. + * + * Fixture: an open host with 1 slot and a closed host with 4 slots; three + * pending jobs in FCFS order: serial A, serial B, then parallel P (exactly + * 2 tasks). Only A fits on the open host. With the closed host excluded, B + * and P must stay pending and the closed host must carry no new usage; + * without the exclusion B and P (or A) land on the closed host. Both + * drivers' fixtures are built like test_sim_contention_limit_driver_ + * parity.c (real scheduler_srcs, real picker/allocator threads), `static` + * fixtures and exactly one RUN_TEST() (bug-1598/1632/1763 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 OPEN_HOST "hclose_open" +#define CLOSED_HOST "hclose_closed" +#define QUEUE_NAME "normal" +#define TEST_RESREQ ((char*)"select[type==any] rusage[mem=1]") + +#define JOBA_ID ((uint64_t)193901ULL) +#define JOBB_ID ((uint64_t)193902ULL) +#define JOBP_ID ((uint64_t)193903ULL) + +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* add_worker(xce_sched_core_t* core, const char* name, + double slots, int status) +{ + 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(name); + w->status = status; + 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); + + rc = sched_add_worker(core, w); + TEST_ASSERT_TRUE(rc >= 0); + + return w; +} + +static void push_job(xce_sched_core_t* core, xce_scheduler_t* job_ctor_wrapper, + xce_sched_plugin_t* picker, uint64_t jobid, int ntasks, + 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_NAME; + submit.user = (char*)"hclose_user"; + submit.project_name = (char*)"default"; + submit.resreq = TEST_RESREQ; + submit.mintasks = ntasks; + submit.maxtasks = ntasks; + + 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 = ntasks; + job->schedobj->u.primary->resreq->maxTasks = ntasks; + + rc = sched_add_job(core, job); + TEST_ASSERT_TRUE(rc >= 0); + picker->picker_iface->on_job_push(picker, job); +} + +static void run_round(xce_sched_core_t* core) +{ + pthread_t picker_tid; + pthread_t allocator_tid; + xce_decision_t* d; + 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); + + do { + d = NULL; + cbuf_get(&(core->decisions), &d); + } while (d != NULL); + + rc = xce_core_end_session(core); + TEST_ASSERT_EQUAL_INT(0, rc); +} + +static xce_sched_worker_t* populate(xce_sched_core_t* core, + xce_scheduler_t* job_wrapper, + xce_sched_plugin_t* picker) +{ + xce_sched_queue_t* q = (xce_sched_queue_t*)calloc(1, sizeof(*q)); + xce_sched_worker_t* closed; + int rc; + + TEST_ASSERT_NOT_NULL(q); + q->name = makeString(QUEUE_NAME); + rc = sched_add_queue(core, q); + TEST_ASSERT_TRUE(rc >= 0); + picker->picker_iface->on_queue_update(picker, q); + + add_worker(core, OPEN_HOST, 1.0, 0); + closed = add_worker(core, CLOSED_HOST, 4.0, + XCE_WORKER_CLOSED | XCE_WORKER_CLOSED_BY_ADMIN); + + push_job(core, job_wrapper, picker, JOBA_ID, 1, 1000ULL); + push_job(core, job_wrapper, picker, JOBB_ID, 1, 1001ULL); + push_job(core, job_wrapper, picker, JOBP_ID, 2, 1002ULL); + + return closed; +} + +static void assert_outcome(xce_sched_core_t* core, xce_sched_worker_t* closed, + const char* mode) +{ + xce_resource_t* r = NULL; + float inuse = 0.0f; + int residx; + int rc; + char msg[160]; + + rc = key_find(core->resmgr, (void*)"slots", (void**)&r); + TEST_ASSERT_TRUE(rc >= 0); + residx = r->idx; + rc = get_resvals(&(closed->resInUse), 1, &residx, &inuse); + if (rc != 1) { + inuse = 0.0f; + } + + snprintf(msg, sizeof(msg), + "%s: the admin-closed host must get no new usage", mode); + TEST_ASSERT_EQUAL_FLOAT_MESSAGE(0.0f, inuse, msg); + + snprintf(msg, sizeof(msg), "%s: serial job A fits the open host", mode); + TEST_ASSERT_EQUAL_UINT_MESSAGE(JOB_STAT_RUN, + sched_get_job(core, JOBA_ID)->state, msg); + snprintf(msg, sizeof(msg), + "%s: serial job B must pend (only the closed host has room)", + mode); + TEST_ASSERT_EQUAL_UINT_MESSAGE(JOB_STAT_PEND, + sched_get_job(core, JOBB_ID)->state, msg); + snprintf(msg, sizeof(msg), + "%s: parallel job P must pend (only the closed host has room)", + mode); + TEST_ASSERT_EQUAL_UINT_MESSAGE(JOB_STAT_PEND, + sched_get_job(core, JOBP_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_admin_closed_host_gets_no_new_jobs(void) +{ + xce_sched_plugin_t* fcfs_normal = NULL; + xce_sched_plugin_t* fcfs_sim = NULL; + xce_sched_worker_t* closed_normal; + xce_sched_worker_t* closed_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); + + closed_normal = populate(&(g_normal.core), &g_normal, fcfs_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; + + closed_sim = populate(&(g_sim.core), &g_sim_job_wrapper, fcfs_sim); + run_round(&(g_sim.core)); + + assert_outcome(&(g_normal.core), closed_normal, "normal mode"); + assert_outcome(&(g_sim.core), closed_sim, "sim mode"); +} + +int main(void) +{ + UNITY_BEGIN(); + RUN_TEST(test_admin_closed_host_gets_no_new_jobs); + return UNITY_END(); +}