/* * test_sim_contention_limit_driver_parity.c - scheduler-sim-live-allocation * spec, task 5.1 (Req 3.1, 3.2, boundary: none new -- pure validation of * the Core-Safe Allocation Chain / Limit Plugin Core-Safe Matching / Unified * Picker/Allocator Threads components tasks 1.1-4.3 already built). * * Sibling of test_sim_contention_resource_driver_parity.c (read that file's * own header comment first for the full rationale on why this is a NEW * file rather than an extension of test_sim_contention_parity.c, and for * the shared fixture technique both files use). This file adds the one * ingredient that one deliberately scopes out: Requirement 3.2's own text * explicitly calls out "capacity constrained by a matching limit" as part * of what a contention fixture must cover, and `scheduler-driver- * equivalence.md`'s testing discipline names limit-constrained capacity by * name. This file is that coverage. * * ── What "constrained by a matching limit" means in THIS codebase today * (read before assuming this test denies a job via the limit) ───────── * Reading plugins/limit.c in full before writing this fixture: a limit * match (match_job_global_limit_()/match_job_host_limit_()) computes and * records a CAP value (update_limit_res_value(), percent-of-underlying- * resource or flat) into core->sharedInst (global) or worker->resvals * (host-scoped) under a synthetic vres, and stores it in the limit * plugin's own instances registry (limit_get_usage_instances(), the * `info limit` accessor) -- but the code path that would actually ADD a * corresponding "wanted" entry for that vres into a job's own capacity * check (build_limit_vres_requests(), plugins/limit.c) is `#if 0`'d out * dead code with zero live callers, and get_vres_value() (its own would-be * per-job charge amount) is `static` and, once that dead code is excluded, * has zero live callers either. In other words: in this codebase's CURRENT * implementation, a configured limit's cap is tracked and displayed * (Requirement 2's entire scope -- `info limit` visibility) but never * itself gates admission -- that is a genuinely separate, not-yet-built * capability, not a gap this validation-only task is scoped to fix (doing * so would be undisclosed production-code scope expansion, forbidden by * this task's own Critical Constraints). * * This fixture therefore proves the honest, in-scope claim: under the SAME * genuine resource-capacity contention as this file's sibling (raw "slots" * capacity is what actually admits/denies each job), a host-scoped limit * that ACTIVELY MATCHES every contending job (via the real, core-safe * plugin_prepare_allocation()->limit_prepare_allocation()->match_job_host_ * limit_() chain, task 2.3/4.3) produces the IDENTICAL admission outcome * AND the IDENTICAL limit-usage-instance state (same vres registered, same * count) on both drivers -- i.e. limit-matching correctness under * contention is driver-equivalent, exactly as Requirement 2.1/2.2 (matched * the same way normal mode would) and Requirement 3.1 (identical * accounting state, limit usage explicitly included in that list) require, * even though the limit itself does not additionally constrain WHICH jobs * admit in this codebase's current implementation. If a future spec wires * build_limit_vres_requests() in for real, this fixture's own "A and B * admit, C is denied by raw capacity" outcome is unaffected either way, * since the limit's own generous cap (percent=1, val=50.0 against a * WORKER_HOSTRES of 1000.0 -- the exact values test_sim_alloc_limit_ * fanout.c's own proven-safe fixture already uses) is far larger than * anything this fixture's jobs could ever request. * * ── Link style / fixture-safety ─────────────────────────────────────────── * Identical to test_sim_contention_resource_driver_parity.c: real, * unmodified scheduler_srcs (minus src/main.c); two `static` (not stack- * local) fixtures, exactly one RUN_TEST(), per this directory's own 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 "xcelimit.h" #include "xstring.h" /* Consumed the same "extern declared where consumed" idiom sim_loop.c * itself and its own sibling real-scheduler_srcs tests already use (no * shared header declares these). */ 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 "contdp_limit_worker" #define QUEUE_NAME "normal" #define TEST_RESREQ ((char*)"select[type==any] rusage[mem=1]") #define WORKER_CAPACITY 2.0f /* same insufficient-for-all-3 shape as the * sibling resource-only fixture */ #define LIMIT_RESNAME "contdp_host_res" #define LIMIT_NAME "contdp_host_limit" #define WORKER_HOSTRES 1000.0f /* generous -- the limit itself must not be * the thing that denies any job here, see * this file's own header comment */ #define JOBA_ID ((uint64_t)510201ULL) #define JOBB_ID ((uint64_t)510202ULL) #define JOBC_ID ((uint64_t)510203ULL) void setUp(void) { return; } void tearDown(void) { return; } /* ---------------------------------------------------------------------- */ /* Fixture helpers -- identical techniques to this file's sibling */ /* (test_sim_contention_resource_driver_parity.c) plus test_sim_alloc_ */ /* limit_fanout.c's own limit-config shape, duplicated per this directory's */ /* established "duplicate, don't share" testsuite convention. */ /* ---------------------------------------------------------------------- */ 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)); } /* register_plain_resource - same shape test_sim_alloc_limit_fanout.c's own * identically-named helper establishes: a bare resource with no LIMIT/ * CONSUMABLE/host flags of its own -- is_host_based_limit() (xcelimit.c) * resolves LIMIT_NAME as host-based purely from limit->val.percent!=0, not * from any flag on this resource (confirmed by reading is_host_based_ * limit()'s body before writing this fixture). */ static xce_resource_t* register_plain_resource(var resmgr, const char* name) { xce_resource_t res; memset(&res, 0, sizeof(res)); res.name = strdup(name); return makeResource(resmgr, &res); } static void add_worker_resval(xce_sched_worker_t* worker, 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(&(worker->resvals), r->idx, val); } static xce_sched_worker_t* make_worker(xce_sched_core_t* core, const char* name, double slots_capacity, double hostres) { xce_sched_worker_t* w = (xce_sched_worker_t*)calloc(1, sizeof(*w)); TEST_ASSERT_NOT_NULL(w); w->name = strdup(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); add_worker_resval(w, core, "slots", slots_capacity); add_worker_resval(w, core, LIMIT_RESNAME, hostres); return w; } static xce_sched_job_t* push_contention_job(xce_sched_core_t* core, xce_scheduler_t* job_ctor_wrapper, xce_sched_plugin_t* picker, uint64_t jobid, 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*)"contdp_limit_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); return job; } static void run_contention_round(xce_sched_core_t* core, int njobs) { pthread_t picker_tid; pthread_t allocator_tid; xce_decision_t* d; xce_decision_t* end_marker = (xce_decision_t*)0x1; int rc; int i; 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 (i = 0; i < njobs; i++) { d = NULL; cbuf_get(&(core->decisions), &d); TEST_ASSERT_NOT_NULL_MESSAGE( d, "the real thread pair must carry every contending job all " "the way to a committed decision"); } end_marker = (xce_decision_t*)0x1; cbuf_get(&(core->decisions), &end_marker); TEST_ASSERT_NULL_MESSAGE(end_marker, "decisions-cbuf entry after all real jobs must " "be the end-of-session NULL marker"); rc = xce_core_end_session(core); TEST_ASSERT_EQUAL_INT(0, rc); } /* count_usage_instances - iterates limit_get_usage_instances(core) * (plugins/limit.c, unchanged scheduler-sim-diagnostics accessor) and * returns how many usage-instance entries exist -- 0 if the accessor * itself returns NULL (no limit plugin registered / no instance yet). */ static int count_usage_instances(xce_sched_core_t* core) { var handle = limit_get_usage_instances(core); iterator_t it; void* item = NULL; int rc; int count = 0; 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; } /* ---------------------------------------------------------------------- */ /* The two fixtures under test -- `static`, see the sibling file's own */ /* header comment ("bug-1598/1632/1763 fixture-safety requirement"). */ /* ---------------------------------------------------------------------- */ static xce_scheduler_t g_normal; static xce_sched_sim_driver_t g_sim; static xce_scheduler_t g_sim_job_wrapper; /* ---------------------------------------------------------------------- */ /* The test */ /* ---------------------------------------------------------------------- */ static void test_limit_constrained_contention_matches_between_normal_and_sim_drivers( void) { xce_limit_val_t limit_val = { .resname = (char*)LIMIT_RESNAME, .percent = 1, .val = 50.0 }; xce_limit_config_t limit_cfg = { .name = (char*)LIMIT_NAME, .val = &limit_val }; xce_limit_config_t* limits[1] = { &limit_cfg }; xce_conf4sched_t config; xce_sched_plugin_t* fcfs_normal = NULL; xce_sched_plugin_t* fcfs_sim = NULL; xce_sched_plugin_t* limit_normal = NULL; xce_sched_plugin_t* limit_sim = NULL; xce_sched_queue_t* queue_normal; xce_sched_queue_t* queue_sim; xce_sched_job_t* jobA_n; xce_sched_job_t* jobB_n; xce_sched_job_t* jobC_n; xce_sched_job_t* jobA_s; xce_sched_job_t* jobB_s; xce_sched_job_t* jobC_s; xce_sched_worker_t* w_normal; xce_sched_worker_t* w_sim; xce_resource_t* r_normal = NULL; xce_resource_t* r_sim = NULL; int residx_normal; int residx_sim; float resinuse_normal = -1.0f; float resinuse_sim = -1.0f; int instances_normal; int instances_sim; int rc; memset(&config, 0, sizeof(config)); config.nlimits = 1; config.limits = limits; /* ================================================================ * 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); TEST_ASSERT_NOT_NULL( register_plain_resource(g_normal.core.resmgr, LIMIT_RESNAME)); 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"; /* Real production limit registration: register_limit()'s own * sequence (plugin_register(sched, limit_new_plugin())), inlined so * this fixture retains the plugin pointer to seed its config directly * via limit_seed_config() -- the same driver-agnostic, core-safe entry * point (task 2.3/bug-1825) sim mode's own sim_bootstrap_limit_ * plugin() uses below, applied here to normal mode's own registration * path instead of routing through a live CONFIG_PUSH event (out of * this fixture's own boundary). */ limit_normal = limit_new_plugin(); TEST_ASSERT_NOT_NULL(limit_normal); rc = plugin_register(&g_normal, limit_normal); TEST_ASSERT_TRUE(rc >= 0); rc = limit_seed_config(limit_normal, &(g_normal.core), &config); 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); queue_normal = (xce_sched_queue_t*)calloc(1, sizeof(*queue_normal)); TEST_ASSERT_NOT_NULL(queue_normal); queue_normal->name = makeString(QUEUE_NAME); rc = sched_add_queue(&(g_normal.core), queue_normal); TEST_ASSERT_TRUE(rc >= 0); fcfs_normal->picker_iface->on_queue_update(fcfs_normal, queue_normal); w_normal = make_worker(&(g_normal.core), WORKER_NAME, WORKER_CAPACITY, WORKER_HOSTRES); rc = sched_add_worker(&(g_normal.core), w_normal); TEST_ASSERT_TRUE(rc >= 0); push_contention_job(&(g_normal.core), &g_normal, fcfs_normal, JOBA_ID, 1000ULL); push_contention_job(&(g_normal.core), &g_normal, fcfs_normal, JOBB_ID, 1001ULL); push_contention_job(&(g_normal.core), &g_normal, fcfs_normal, JOBC_ID, 1002ULL); run_contention_round(&(g_normal.core), 3); /* ================================================================ * 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); TEST_ASSERT_NOT_NULL( register_plain_resource(g_sim.core.resmgr, LIMIT_RESNAME)); 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); g_sim.core.config = &config; sim_bootstrap_limit_plugin(&g_sim); rc = key_find(g_sim.core.plugins, "xce_sched_limit", (void**)&limit_sim); TEST_ASSERT_TRUE(rc >= 0); TEST_ASSERT_NOT_NULL(limit_sim); TEST_ASSERT_EQUAL_PTR_MESSAGE( &(g_sim.core), limit_sim->core, "sanity: the limit plugin must be registered directly against " "sim.core -- the same registry the seqjob/parjob allocator " "plugin's own prepare_allocation_hook fan-out must reach for a " "host-scoped match to ever be found (task 4.3, bug-1829's own " "follow-up fix)"); sim_bootstrap_seqjob_allocator(&g_sim); sim_bootstrap_parjob_allocator(&g_sim); TEST_ASSERT_NOT_NULL_MESSAGE( g_sim.core.seqjob_allocator, "sim_bootstrap_seqjob_allocator() must populate core->seqjob_" "allocator (task 4.1)"); queue_sim = (xce_sched_queue_t*)calloc(1, sizeof(*queue_sim)); TEST_ASSERT_NOT_NULL(queue_sim); queue_sim->name = makeString(QUEUE_NAME); rc = sched_add_queue(&(g_sim.core), queue_sim); TEST_ASSERT_TRUE(rc >= 0); fcfs_sim->picker_iface->on_queue_update(fcfs_sim, queue_sim); w_sim = make_worker(&(g_sim.core), WORKER_NAME, WORKER_CAPACITY, WORKER_HOSTRES); rc = sched_add_worker(&(g_sim.core), w_sim); TEST_ASSERT_TRUE(rc >= 0); 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; push_contention_job(&(g_sim.core), &g_sim_job_wrapper, fcfs_sim, JOBA_ID, 1000ULL); push_contention_job(&(g_sim.core), &g_sim_job_wrapper, fcfs_sim, JOBB_ID, 1001ULL); push_contention_job(&(g_sim.core), &g_sim_job_wrapper, fcfs_sim, JOBC_ID, 1002ULL); run_contention_round(&(g_sim.core), 3); /* ================================================================ * Assertions * ================================================================ */ jobA_n = sched_get_job(&(g_normal.core), JOBA_ID); jobB_n = sched_get_job(&(g_normal.core), JOBB_ID); jobC_n = sched_get_job(&(g_normal.core), JOBC_ID); jobA_s = sched_get_job(&(g_sim.core), JOBA_ID); jobB_s = sched_get_job(&(g_sim.core), JOBB_ID); jobC_s = sched_get_job(&(g_sim.core), JOBC_ID); TEST_ASSERT_NOT_NULL(jobA_n); TEST_ASSERT_NOT_NULL(jobB_n); TEST_ASSERT_NOT_NULL(jobC_n); TEST_ASSERT_NOT_NULL(jobA_s); TEST_ASSERT_NOT_NULL(jobB_s); TEST_ASSERT_NOT_NULL(jobC_s); /* Same raw-capacity-driven outcome as the sibling fixture -- the * matching limit (generous cap, see file header) must not itself * change which jobs admit. */ TEST_ASSERT_EQUAL_UINT_MESSAGE( JOB_STAT_RUN, jobA_n->state, "normal mode: job A must dispatch (limit must not spuriously deny " "a job it has no admission-gating effect on in this codebase)"); TEST_ASSERT_EQUAL_UINT_MESSAGE(JOB_STAT_RUN, jobB_n->state, "normal mode: job B must dispatch"); TEST_ASSERT_EQUAL_UINT_MESSAGE( JOB_STAT_PEND, jobC_n->state, "normal mode: job C must still be denied by genuine raw-capacity " "contention (2.0 slots, 3x1.0 demand)"); TEST_ASSERT_EQUAL_UINT_MESSAGE(JOB_STAT_RUN, jobA_s->state, "sim mode: job A must dispatch"); TEST_ASSERT_EQUAL_UINT_MESSAGE(JOB_STAT_RUN, jobB_s->state, "sim mode: job B must dispatch"); TEST_ASSERT_EQUAL_UINT_MESSAGE( JOB_STAT_PEND, jobC_s->state, "sim mode: job C must still be denied by genuine raw-capacity " "contention"); TEST_ASSERT_EQUAL_UINT_MESSAGE( jobA_n->state, jobA_s->state, "job A: normal and sim mode must agree under limit-matched " "contention"); TEST_ASSERT_EQUAL_UINT_MESSAGE( jobB_n->state, jobB_s->state, "job B: normal and sim mode must agree under limit-matched " "contention"); TEST_ASSERT_EQUAL_UINT_MESSAGE( jobC_n->state, jobC_s->state, "job C: normal and sim mode must agree under limit-matched " "contention"); /* Final resource-accounting-state parity for the RAW resource, same * as the sibling fixture. */ rc = key_find(g_normal.core.resmgr, (void*)"slots", (void**)&r_normal); TEST_ASSERT_TRUE(rc >= 0); residx_normal = r_normal->idx; rc = key_find(g_sim.core.resmgr, (void*)"slots", (void**)&r_sim); TEST_ASSERT_TRUE(rc >= 0); residx_sim = r_sim->idx; rc = get_resvals(&(w_normal->resInUse), 1, &residx_normal, &resinuse_normal); TEST_ASSERT_EQUAL_INT(1, rc); rc = get_resvals(&(w_sim->resInUse), 1, &residx_sim, &resinuse_sim); TEST_ASSERT_EQUAL_INT(1, rc); TEST_ASSERT_EQUAL_FLOAT_MESSAGE( WORKER_CAPACITY, resinuse_normal, "normal mode: worker resInUse must reach exactly full saturation"); TEST_ASSERT_EQUAL_FLOAT_MESSAGE( WORKER_CAPACITY, resinuse_sim, "sim mode: worker resInUse must reach exactly full saturation"); TEST_ASSERT_EQUAL_FLOAT_MESSAGE( resinuse_normal, resinuse_sim, "driver parity: worker resInUse must match byte-for-byte under " "limit-matched contention too"); /* Limit-usage-instance parity (Requirement 2.1/2.2/3.1's own explicit * inclusion of limit usage in the accounting state that must match). * Every one of the 3 jobs' allocation attempts reaches plugin_prepare_ * allocation() (resource.c's sched_try_alloc_on_host_() calls it * BEFORE sched_check_host_satisfy(), confirmed by reading that * function's body before writing this fixture -- i.e. a job that is * ultimately DENIED by raw capacity still genuinely matches the limit * on its way there), and none of this fixture's consumer scoping * (user/queue/project all left unconfigured) differentiates jobs from * one another, so exactly ONE shared vres/usage-instance is expected * on each side, not three. */ instances_normal = count_usage_instances(&(g_normal.core)); instances_sim = count_usage_instances(&(g_sim.core)); TEST_ASSERT_EQUAL_INT_MESSAGE( 1, instances_normal, "normal mode: the host-scoped limit must genuinely match and " "register exactly one usage instance under this contention"); TEST_ASSERT_EQUAL_INT_MESSAGE( 1, instances_sim, "sim mode: the host-scoped limit must genuinely match and " "register exactly one usage instance under this contention -- " "before task 2.3/4.3's fixes this stayed 0 (plugin->sched NULL / " "wrong plugin registry), which is exactly what this assertion " "would have caught"); TEST_ASSERT_EQUAL_INT_MESSAGE( instances_normal, instances_sim, "driver parity: limit-usage-instance count must match between " "normal and sim mode under identical limit-matched contention"); printf("test_limit_constrained_contention_matches_between_normal_and_" "sim_drivers: PASS\n"); } int main(void) { UNITY_BEGIN(); RUN_TEST( test_limit_constrained_contention_matches_between_normal_and_sim_drivers); return UNITY_END(); }