diff --git a/src/scheduler/include/core.h b/src/scheduler/include/core.h index 79ed411f..a7b8b921 100644 --- a/src/scheduler/include/core.h +++ b/src/scheduler/include/core.h @@ -571,6 +571,19 @@ typedef struct { */ extern int xce_core_push_events(xce_sched_core_t*, xce_core_event_t*, int); +/* + * xce_core_load_groups - registers a config's named user groups + * (config->ugroups) and host groups (config->hgroups) into core->ugmgr/ + * core->hgmgr by name (bug-1784, bug-1943). The one group-loading path of + * both drivers: live's sched_init_config(), the core's CONFIG_PUSH event + * and snapshot restore all call it, and each calls it before anything that + * may reference a group by name (the limit plugin's consumers) is built. + * An existing group is updated in place, so limits built earlier see the + * new membership. Returns 0, -EINVAL for a NULL core, or the first + * groupmgr_load_configs() error. A NULL config is a no-op. + */ +extern int xce_core_load_groups(xce_sched_core_t*, xce_conf4sched_t*); + /* * xce_core_set_decision_cb - registers the driver's decision callback + * opaque driver context (task 4.4, Req 5 AC 6). May be called at any time diff --git a/src/scheduler/src/core.c b/src/scheduler/src/core.c index f9b381a3..323c5a9c 100644 --- a/src/scheduler/src/core.c +++ b/src/scheduler/src/core.c @@ -642,6 +642,28 @@ xce_sched_project_t* sched_get_project(xce_sched_core_t* core, char* name) return make_sched_project(core, name); } +int xce_core_load_groups(xce_sched_core_t* core, xce_conf4sched_t* config) +{ + int rc; + + if (core == NULL) { + return -EINVAL; + } + + if (config == NULL) { + return 0; + } + + rc = groupmgr_load_configs(core->ugmgr, config->n_ugroups, + config->ugroups); + if (rc < 0) { + return rc; + } + + return groupmgr_load_configs(core->hgmgr, config->n_hgroups, + config->hgroups); +} + /* FIXME this is just stub, we need to add more */ int sched_job_is_ready(xce_sched_job_t* j) { return 1; } diff --git a/src/scheduler/src/core_events.c b/src/scheduler/src/core_events.c index c7cb0e21..ab958d89 100644 --- a/src/scheduler/src/core_events.c +++ b/src/scheduler/src/core_events.c @@ -843,6 +843,14 @@ int xce_core_push_events(xce_sched_core_t* core, xce_core_event_t* events, core->config = config; core->configver++; + /* bug-1943: named groups first, as live's sched_init_config() + * does, so the simulator resolves group-scoped limits too */ + rc = xce_core_load_groups(core, config); + if (rc < 0) { + WARN("%s CONFIG_PUSH: loading groups failed rc=%d", __func__, + rc); + } + if (config != NULL) { core_apply_config_resources(core, config); core_apply_config_resvals(core, config); diff --git a/src/scheduler/src/scheduler.c b/src/scheduler/src/scheduler.c index 360737ed..65d3a22b 100644 --- a/src/scheduler/src/scheduler.c +++ b/src/scheduler/src/scheduler.c @@ -1140,9 +1140,9 @@ int sched_init_config(xce_scheduler_t* sched, xce_conf4sched_t* config) * config that references one of these by name (e.g. a limit scoped to * users.members=["ug1"]) resolves it via build_consumer()'s own * makeGroup() call, which can only find an existing named group if it - * was already registered under that name. */ - groupmgr_load_configs(sched->core.ugmgr, config->n_ugroups, config->ugroups); - groupmgr_load_configs(sched->core.hgmgr, config->n_hgroups, config->hgroups); + * was already registered under that name. Shared with the core's + * CONFIG_PUSH event and snapshot restore (bug-1943). */ + xce_core_load_groups(&(sched->core), config); plugin_config_update(sched, config); older = sched->core.config; diff --git a/src/scheduler/src/snapshot.c b/src/scheduler/src/snapshot.c index a128c6bf..33bbbfa2 100644 --- a/src/scheduler/src/snapshot.c +++ b/src/scheduler/src/snapshot.c @@ -287,10 +287,13 @@ * current codebase ever sets either field on a live queue (scaffolding * only, per task 1.3's own design.md note), so this has zero observed * impact today. - * - core->ugmgr/qgmgr/pgmgr/hgmgr (group managers) are NOT captured -- - * they have no field anywhere in sched_snapshot_data_t, and nothing in - * core.c/core_events.c populates them from any event today either (no - * regression, no existing behavior to preserve). + * - core->ugmgr/qgmgr/pgmgr/hgmgr (group managers) are NOT captured as + * such -- they have no field in sched_snapshot_data_t. The named + * user/host groups are derived state: they travel inside `config` + * (ugroups/hgroups) and snapshot_load.c rebuilds ugmgr/hgmgr from it + * via xce_core_load_groups() (bug-1943), the same loader the live + * config apply and the CONFIG_PUSH event use. qgmgr/pgmgr have no + * config source on any path. * * ========================================================================== * Why not reuse core_events.c's core_build_job/core_build_queue/ diff --git a/src/scheduler/src/snapshot_load.c b/src/scheduler/src/snapshot_load.c index 8d46af35..5f7ca80b 100644 --- a/src/scheduler/src/snapshot_load.c +++ b/src/scheduler/src/snapshot_load.c @@ -99,6 +99,14 @@ int xce_sched_core_load_snapshot(xce_sched_core_t* core, core->config = in->config; core->configver++; + /* bug-1943: the snapshot's config carries the named user/host groups; + * register them now, before the simulator seeds the limit plugin from + * this config (sim_bootstrap_limit_plugin()), as live does */ + rc = xce_core_load_groups(core, core->config); + if (rc < 0) { + WARN("%s failed to load the config's groups (rc=%d)", __func__, rc); + } + if ((picker != NULL) && (picker->picker_iface->on_config != NULL)) { picker->picker_iface->on_config(picker, core->config); } diff --git a/src/scheduler/testsuite/meson.build b/src/scheduler/testsuite/meson.build index 7a35f469..a9aa7d4b 100644 --- a/src/scheduler/testsuite/meson.build +++ b/src/scheduler/testsuite/meson.build @@ -3091,3 +3091,25 @@ sched_host_admin_close_test = executable( test('scheduler host admin close test', sched_host_admin_close_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 +# real-scheduler_srcs link technique as above. +sched_sim_group_limit_parity_test_srcs = ['test_sim_group_limit_parity.c'] +foreach f : scheduler_srcs + if f != 'src/main.c' + sched_sim_group_limit_parity_test_srcs += ['../' + f] + endif +endforeach + +sched_sim_group_limit_parity_test = executable( + 'sched_sim_group_limit_parity_test', + sched_sim_group_limit_parity_test_srcs, + include_directories:[incdir], + dependencies: [scheduler_test_deps], + install: false) + +test('scheduler sim group limit parity test', + sched_sim_group_limit_parity_test, suite: 'scheduler', + env: ['ASAN_OPTIONS=detect_leaks=0']) diff --git a/src/scheduler/testsuite/test_sim_group_limit_parity.c b/src/scheduler/testsuite/test_sim_group_limit_parity.c new file mode 100644 index 00000000..a3a8721e --- /dev/null +++ b/src/scheduler/testsuite/test_sim_group_limit_parity.c @@ -0,0 +1,348 @@ +/* + * test_sim_group_limit_parity.c - bug-1943: the simulator must load the + * cluster's named user/host groups exactly as the live scheduler does, so + * a limit scoped to a user group matches the same jobs on both drivers. + * + * Live applies a pushed config through sched_init_config() (scheduler.c), + * which registers config->ugroups/hgroups into core.ugmgr/hgmgr before the + * limit plugin builds its consumers (bug-1784: a limit member "ug1" only + * resolves to the group if "ug1" is already registered). The simulator + * takes its config from the replay snapshot (xce_sched_core_load_ + * snapshot(), snapshot_load.c), seeds the limit plugin from it + * (sim_bootstrap_limit_plugin()), and applies later pushes through the + * core's CONFIG_PUSH event (core_events.c). + * + * Part 1: one config carrying user group ug1 = {user1} and a host-scoped + * limit whose users are ["ug1"], applied the live way and the sim way; a + * job of user1 must match the limit (one usage instance) on both drivers. + * Part 2: the core's CONFIG_PUSH event alone must register ug1 and hg1. + * + * Fixture technique and safety conventions as in + * test_sim_contention_limit_driver_parity.c (real scheduler_srcs, real + * picker/allocator threads, `static` fixtures, exactly one RUN_TEST()). + */ +#include +#include +#include +#include +#include + +#include + +#include "controller.capnp.h" +#include "decision.h" +#include "group.h" +#include "plugin.h" +#include "preason.h" +#include "resobj.h" +#include "sched/types.h" +#include "scheduler.h" +#include "sim.h" +#include "xcelimit.h" +#include "xstring.h" + +extern void sim_bootstrap_picker(xce_sched_sim_driver_t* sim); +extern void sim_bootstrap_limit_plugin(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 "grplim_worker" +#define QUEUE_NAME "normal" +#define TEST_RESREQ ((char*)"select[type==any] rusage[mem=1]") +#define LIMIT_RESNAME "grplim_host_res" +#define LIMIT_NAME "grplim_ug1_limit" +#define JOB_USER "user1" +#define JOB_ID ((uint64_t)194301ULL) + +void setUp(void) { return; } +void tearDown(void) { return; } + +static void register_resources(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)); + + memset(&res, 0, sizeof(res)); + res.name = strdup(LIMIT_RESNAME); + TEST_ASSERT_NOT_NULL(makeResource(resmgr, &res)); +} + +static void add_worker_resval(xce_sched_worker_t* w, xce_sched_core_t* core, + const char* resname, double value) +{ + xce_resource_t* r = NULL; + xce_val_t* val = NULL; + int rc; + + rc = key_find(core->resmgr, (void*)resname, (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 = value; + zarray_put(&(w->resvals), r->idx, val); +} + +static void 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* w = (xce_sched_worker_t*)calloc(1, sizeof(*w)); + job_submit_data_t submit; + xce_job_t j; + xce_sched_job_t* job; + 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); + + TEST_ASSERT_NOT_NULL(w); + w->name = strdup(WORKER_NAME); + 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); + add_worker_resval(w, core, "slots", 4.0); + add_worker_resval(w, core, LIMIT_RESNAME, 1000.0); + rc = sched_add_worker(core, w); + TEST_ASSERT_TRUE(rc >= 0); + + memset(&submit, 0, sizeof(submit)); + submit.queue = (char*)QUEUE_NAME; + submit.user = (char*)JOB_USER; + submit.project_name = (char*)"default"; + submit.resreq = TEST_RESREQ; + submit.mintasks = 1; + submit.maxtasks = 1; + + memset(&j, 0, sizeof(j)); + j.jobid = JOB_ID; + j.state = JOB_STAT_PEND; + j.sub_time = 1000ULL; + j.queue_time = 1000ULL; + j.submit = &submit; + + job = make_sched_job(job_wrapper, &j); + TEST_ASSERT_NOT_NULL(job); + 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); +} + +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 int count_usage_instances(xce_sched_core_t* core) +{ + var handle = limit_get_usage_instances(core); + iterator_t it; + void* item = NULL; + int count = 0; + int rc; + + if (handle == NULL) { + return 0; + } + + rc = iter_make(&it, handle); + if (rc < 0) { + return 0; + } + + for (;;) { + rc = iter_next(&it, &item); + if ((rc < 0) || (rc == RTTI_ITER_NO_DATA)) { + break; + } + count++; + } + + iter_destroy(&it); + + return count; +} + +static xce_scheduler_t g_normal; +static xce_sched_sim_driver_t g_sim; +static xce_scheduler_t g_sim_job_wrapper; +static xce_sched_sim_driver_t g_evsim; + +/* config pieces: `static` so the configs the cores take over outlive the + * test function (no core here is ever torn down or handed a newer config, + * so none of them is freed) */ +static xce_group_config_t g_ug1 = { + .name = "ug1", + .members = { .count = 1, .data = (char*[]) { JOB_USER } } +}; +static xce_group_config_t g_hg1 = { + .name = "hg1", + .members = { .count = 1, .data = (char*[]) { WORKER_NAME } } +}; +static xce_group_config_t* g_ugroups[1] = { &g_ug1 }; +static xce_group_config_t* g_hgroups[1] = { &g_hg1 }; +static xce_group_config_t g_limit_users = { + .members = { .count = 1, .data = (char*[]) { "ug1" } } +}; +static xce_limit_val_t g_limit_val = { .resname = (char*)LIMIT_RESNAME, + .percent = 1, + .val = 50.0 }; +static xce_limit_config_t g_limit_cfg = { .name = (char*)LIMIT_NAME, + .users = &g_limit_users, + .val = &g_limit_val }; +static xce_limit_config_t* g_limits[1] = { &g_limit_cfg }; + +static void init_config(xce_conf4sched_t* config) +{ + memset(config, 0, sizeof(*config)); + config->n_ugroups = 1; + config->ugroups = g_ugroups; + config->n_hgroups = 1; + config->hgroups = g_hgroups; + config->nlimits = 1; + config->limits = g_limits; +} + +static xce_conf4sched_t g_config_normal; +static xce_conf4sched_t g_config_sim; +static xce_conf4sched_t g_config_ev; + +static void test_user_group_limit_matches_in_sim_as_in_live(void) +{ + xce_sched_plugin_t* fcfs_normal = NULL; + xce_sched_plugin_t* fcfs_sim = NULL; + sched_snapshot_data_t snap; + xce_core_event_t ev; + xce_group_t* g; + int n_normal; + int n_sim; + int rc; + + init_config(&g_config_normal); + init_config(&g_config_sim); + init_config(&g_config_ev); + + /* ---- Part 1, live: config applied by sched_init_config() ---- */ + memset(&g_normal, 0, sizeof(g_normal)); + rc = xce_sched_core_init(&(g_normal.core)); + TEST_ASSERT_EQUAL_INT(0, rc); + register_resources(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 = plugin_register(&g_normal, limit_new_plugin()); + TEST_ASSERT_TRUE(rc >= 0); + rc = register_seqjob_allocator(&g_normal); + TEST_ASSERT_TRUE(rc >= 0); + rc = register_parjob_allocator(&g_normal); + TEST_ASSERT_TRUE(rc >= 0); + + rc = sched_init_config(&g_normal, &g_config_normal); + TEST_ASSERT_EQUAL_INT(0, rc); + + populate(&(g_normal.core), &g_normal, fcfs_normal); + run_round(&(g_normal.core)); + + /* ---- Part 1, sim: config from the replay snapshot, then the same + * bootstrap order as sim_driver_run_body() (sim_loop.c) ---- */ + memset(&g_sim, 0, sizeof(g_sim)); + rc = xce_sched_core_init(&(g_sim.core)); + TEST_ASSERT_EQUAL_INT(0, rc); + register_resources(g_sim.core.resmgr); + + memset(&snap, 0, sizeof(snap)); + snap.config = &g_config_sim; + rc = xce_sched_core_load_snapshot(&(g_sim.core), &snap); + TEST_ASSERT_TRUE(rc >= 0); + + 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_limit_plugin(&g_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); + run_round(&(g_sim.core)); + + n_normal = count_usage_instances(&(g_normal.core)); + n_sim = count_usage_instances(&(g_sim.core)); + + TEST_ASSERT_EQUAL_INT_MESSAGE( + 1, n_normal, + "live: user1 is in ug1, so the ug1-scoped limit must match its job"); + TEST_ASSERT_EQUAL_INT_MESSAGE( + n_normal, n_sim, + "sim must match the user-group-scoped limit exactly as live does"); + + /* ---- Part 2: the core CONFIG_PUSH event registers the groups ---- */ + memset(&g_evsim, 0, sizeof(g_evsim)); + rc = xce_sched_core_init(&(g_evsim.core)); + TEST_ASSERT_EQUAL_INT(0, rc); + + memset(&ev, 0, sizeof(ev)); + ev.type = XCE_CORE_EVENT_CONFIG_PUSH; + ev.u.config_push.config = &g_config_ev; + rc = xce_core_push_events(&(g_evsim.core), &ev, 1); + TEST_ASSERT_TRUE(rc >= 0); + + g = get_group(g_evsim.core.ugmgr, (char*)"ug1"); + TEST_ASSERT_NOT_NULL_MESSAGE( + g, "CONFIG_PUSH must register the config's user groups by name"); + TEST_ASSERT_TRUE(is_in_group(g, (char*)JOB_USER) > 0); + g = get_group(g_evsim.core.hgmgr, (char*)"hg1"); + TEST_ASSERT_NOT_NULL_MESSAGE( + g, "CONFIG_PUSH must register the config's host groups by name"); + TEST_ASSERT_TRUE(is_in_group(g, (char*)WORKER_NAME) > 0); +} + +int main(void) +{ + UNITY_BEGIN(); + RUN_TEST(test_user_group_limit_matches_in_sim_as_in_live); + return UNITY_END(); +}