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(); +}