diff --git a/src/scheduler/src/host_iter.c b/src/scheduler/src/host_iter.c index b1f69a9f..db19296f 100644 --- a/src/scheduler/src/host_iter.c +++ b/src/scheduler/src/host_iter.c @@ -152,6 +152,15 @@ xce_sched_worker_t* sched_host_iter_next(xce_host_iter_t* iter) continue; } + /* bug-1963: likewise an unavailable host (heartbeat timeout) or one + * whose job-state reconciliation is still outstanding: neither + * takes new jobs. Their resInUse is frozen meanwhile (bug-1799/ + * 1801), so the resource check below cannot be relied on to reject + * them. */ + if (IS_WORKER_UNAVAIL(w->status) || IS_WORKER_RECONCILING(w->status)) { + 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 10cfaf6f..c805ba15 100644 --- a/src/scheduler/testsuite/meson.build +++ b/src/scheduler/testsuite/meson.build @@ -3092,6 +3092,27 @@ test('scheduler host admin close test', sched_host_admin_close_test, suite: 'scheduler', env: ['ASAN_OPTIONS=detect_leaks=0']) +# bug-1963: an unavailable or reconciling host 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_unavail_reconciling_test_srcs = ['test_host_unavail_reconciling.c'] +foreach f : scheduler_srcs + if f != 'src/main.c' + sched_host_unavail_reconciling_test_srcs += ['../' + f] + endif +endforeach + +sched_host_unavail_reconciling_test = executable( + 'sched_host_unavail_reconciling_test', + sched_host_unavail_reconciling_test_srcs, + include_directories:[incdir], + dependencies: [scheduler_test_deps], + install: false) + +test('scheduler host unavail reconciling test', + sched_host_unavail_reconciling_test, suite: 'scheduler', + env: ['ASAN_OPTIONS=detect_leaks=0']) + # bug-1943: the simulator loads the config's named user/host groups the way # the live scheduler does (snapshot config and CONFIG_PUSH event), so a # user-group-scoped limit matches the same job on both drivers. Same diff --git a/src/scheduler/testsuite/test_host_unavail_reconciling.c b/src/scheduler/testsuite/test_host_unavail_reconciling.c new file mode 100644 index 00000000..c26cf068 --- /dev/null +++ b/src/scheduler/testsuite/test_host_unavail_reconciling.c @@ -0,0 +1,286 @@ +/* + * test_host_unavail_reconciling.c - bug-1963: a host the controller marks + * XCE_WORKER_UNAVAIL (heartbeat timeout, controller/cluster.c) or + * XCE_WORKER_RECONCILING (worker-rejoin-resync, controller/resync.c) gets + * no new jobs, from either the serial (seqjob) or the parallel (parjob) + * allocator, identically on the normal-mode and sim-mode drivers -- + * exactly as an admin-closed host does since bug-1939 + * (test_host_admin_close.c, whose fixture this one mirrors). + * + * Fixture, built once per (driver, status bit): an open host with 1 slot + * and a "bad" host with 4 slots carrying the status bit; 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 bad host excluded, B and + * P must stay pending; without the exclusion they land on the bad host. + * The assertion is on job state, not on the bad host's resInUse: the + * accounting paths deliberately leave an UNAVAIL/RECONCILING host's + * resInUse frozen (bug-1799/1801), so usage alone cannot show a placement. + * `static` fixtures and exactly one RUN_TEST() (bug-1598/1632/1763 + * convention). + */ +#include +#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 "hunavail_open" +#define BAD_HOST "hunavail_bad" +#define QUEUE_NAME "normal" +#define TEST_RESREQ ((char*)"select[type==any] rusage[mem=1]") + +#define JOBA_ID ((uint64_t)196301ULL) +#define JOBB_ID ((uint64_t)196302ULL) +#define JOBP_ID ((uint64_t)196303ULL) + +#define NSCENARIOS 2 + +static const int g_bad_status[NSCENARIOS] = { XCE_WORKER_UNAVAIL, + XCE_WORKER_RECONCILING }; +static const char* g_bad_label[NSCENARIOS] = { "unavailable", "reconciling" }; + +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*)"hunavail_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 void populate(xce_sched_core_t* core, xce_scheduler_t* job_wrapper, + xce_sched_plugin_t* picker, int bad_status) +{ + xce_sched_queue_t* q = (xce_sched_queue_t*)calloc(1, sizeof(*q)); + 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); + add_worker(core, BAD_HOST, 4.0, bad_status); + + 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); +} + +static void assert_outcome(xce_sched_core_t* core, const char* mode, + const char* label) +{ + char msg[160]; + + snprintf(msg, sizeof(msg), "%s, %s host: serial job A fits the open host", + mode, label); + TEST_ASSERT_EQUAL_UINT_MESSAGE(JOB_STAT_RUN, + sched_get_job(core, JOBA_ID)->state, msg); + snprintf(msg, sizeof(msg), + "%s, %s host: serial job B must pend (only the %s host has " + "room)", + mode, label, label); + TEST_ASSERT_EQUAL_UINT_MESSAGE(JOB_STAT_PEND, + sched_get_job(core, JOBB_ID)->state, msg); + snprintf(msg, sizeof(msg), + "%s, %s host: parallel job P must pend (only the %s host has " + "room)", + mode, label, label); + TEST_ASSERT_EQUAL_UINT_MESSAGE(JOB_STAT_PEND, + sched_get_job(core, JOBP_ID)->state, msg); +} + +static xce_scheduler_t g_normal[NSCENARIOS]; +static xce_sched_sim_driver_t g_sim[NSCENARIOS]; +static xce_scheduler_t g_sim_job_wrapper[NSCENARIOS]; + +static void run_normal(int i) +{ + xce_sched_plugin_t* fcfs = NULL; + int rc; + + memset(&g_normal[i], 0, sizeof(g_normal[i])); + rc = xce_sched_core_init(&(g_normal[i].core)); + TEST_ASSERT_EQUAL_INT(0, rc); + register_slots_resource(g_normal[i].core.resmgr); + + fcfs = builtin_fcfs_new_plugin(); + TEST_ASSERT_NOT_NULL(fcfs); + rc = plugin_register(&g_normal[i], fcfs); + TEST_ASSERT_TRUE(rc >= 0); + g_normal[i].core.job_pick_plugin = (char*)"builtin_fcfs"; + rc = register_seqjob_allocator(&g_normal[i]); + TEST_ASSERT_TRUE(rc >= 0); + rc = register_parjob_allocator(&g_normal[i]); + TEST_ASSERT_TRUE(rc >= 0); + + populate(&(g_normal[i].core), &g_normal[i], fcfs, g_bad_status[i]); + run_round(&(g_normal[i].core)); +} + +static void run_sim(int i) +{ + xce_sched_plugin_t* fcfs = NULL; + int rc; + + memset(&g_sim[i], 0, sizeof(g_sim[i])); + rc = xce_sched_core_init(&(g_sim[i].core)); + TEST_ASSERT_EQUAL_INT(0, rc); + register_slots_resource(g_sim[i].core.resmgr); + + sim_bootstrap_picker(&g_sim[i]); + rc = key_find(g_sim[i].core.plugins, "builtin_fcfs", (void**)&fcfs); + TEST_ASSERT_TRUE(rc >= 0); + TEST_ASSERT_NOT_NULL(fcfs); + sim_bootstrap_seqjob_allocator(&g_sim[i]); + sim_bootstrap_parjob_allocator(&g_sim[i]); + + memset(&g_sim_job_wrapper[i], 0, sizeof(g_sim_job_wrapper[i])); + rc = xce_sched_core_init(&(g_sim_job_wrapper[i].core)); + TEST_ASSERT_EQUAL_INT(0, rc); + g_sim_job_wrapper[i].core.resmgr = g_sim[i].core.resmgr; + g_sim_job_wrapper[i].core.workers = g_sim[i].core.workers; + g_sim_job_wrapper[i].core.queues = g_sim[i].core.queues; + + populate(&(g_sim[i].core), &g_sim_job_wrapper[i], fcfs, g_bad_status[i]); + run_round(&(g_sim[i].core)); +} + +static void test_unavail_and_reconciling_hosts_get_no_new_jobs(void) +{ + int i; + + for (i = 0; i < NSCENARIOS; i++) { + run_normal(i); + run_sim(i); + } + + for (i = 0; i < NSCENARIOS; i++) { + assert_outcome(&(g_normal[i].core), "normal mode", g_bad_label[i]); + assert_outcome(&(g_sim[i].core), "sim mode", g_bad_label[i]); + } +} + +int main(void) +{ + UNITY_BEGIN(); + RUN_TEST(test_unavail_and_reconciling_hosts_get_no_new_jobs); + return UNITY_END(); +}