/* * test_sim_resource_contention.c - resource-contention replay fixture * (scheduler-sim-decision-parity spec, task 1.2, Req 2.1/2.2). * * Why this fixture exists: every existing `.replay` fixture in this * directory (test_sim_equivalence.c/test_repl_equivalence.c/ * test_sim_verify.c/etc.) uses jobs with a trivially-satisfiable resreq * ("type == any") against a core that never has a worker with limited * capacity -- capacity is never actually scarce, so no existing fixture can * exercise or prove capacity-check correctness (design.md's own File * Structure Plan entry for this task: "required because existing fixtures * evidently don't exercise this path"). This file builds ONE recorded * normal-mode session where THREE jobs compete for a worker with real, * insufficient total capacity, so at least one job is genuinely denied * capacity a currently-running job is holding. * * Fixture-construction technique: the SAME dual-phase pattern * test_sim_equivalence.c already establishes (real normal-mode recording via * sched_record_start()/sched_record_stop() driving the REAL builtin_fcfs * picker + a "stand-in allocator" thread standing in for scheduler.c's own * sched_allocator() -- that file's own header comment explains why a * stand-in is used instead of the real allocator: the real one pulls in * resreq-DSL compilation/plugin machinery this whole test-file family * deliberately avoids). This file's stand-in allocator differs from * test_sim_equivalence.c's own (which unconditionally PENDs every * candidate): it drives the REAL, unmodified resource.c capacity-check and * accounting primitives directly -- sched_check_host_satisfy() (genuine * total-vs-inuse-vs-wanted arithmetic against a real xce_sched_worker_t) and * sched_charge_account()/sched_commit_job_alloc() (task 1.1's relocated * accounting functions, now xce_sched_core_t*-based) -- against a real * worker added via sched_add_worker() with a real, finite capacity (10.0 * units of a resource, residx set directly the same way * test_resource_accounting.c's own make_test_account() does, bypassing * resmgr name resolution -- orthogonal to what capacity contention proves). * This means every DISPATCH/PEND decision this fixture records is a * genuine, real-arithmetic outcome of real production capacity-check code, * not fabricated -- only the resreq-DSL-to-`wanted`-vector compilation step * is replaced by a fixed per-job lookup table (JOB A/B/C's "wants" below), * exactly the same scope-reduction test_sim_equivalence.c's own header * comment documents for its own simplification. * * The resulting `.replay` file R is dual-use, per this task's own * requirement text ("usable both as a normal-mode recording source and as a * simulation-mode replay input"): it IS a genuine normal-mode recording (produced by * sched_record_start()/sched_record_stop() around a real * xce_sched_driver_run_normal_mode_round() call), and this file's own * test_contention_fixture_is_replayable_via_simulation_driver() below * additionally proves it is directly consumable by the REAL, unmodified * sched_sim_driver_run() as replay input. A later task (4.1, out of this * task's boundary) reuses this same two-phase shape to assert resInUse * parity between a normal-mode recording and its simulation-mode replay; * this task's own job is only to build the fixture and prove its own shape * (multiple jobs, insufficient combined capacity, at least one round where * one job is denied capacity a running job holds) -- not to run that later * comparison. * * Scenario: worker "worker-a" has 10.0 total units of one resource. Job A * (submitted first, sub_time=1000) wants 8.0 -- satisfied, dispatched, and * charged into worker-a's resInUse. Job B (sub_time=1001) wants 5.0 -- 8.0 * (A's charge, now-running) + 5.0 (B's want) = 13.0 > 10.0 total, so B is * genuinely DENIED capacity that job A, now running, is consuming (this * task's own Observable Completion, proven by decoding the recorded * decision stream below, not merely asserted from this file's own input * constants). Job C (sub_time=1002) wants 2.0 -- exactly the remaining * 10.0-8.0=2.0 headroom, so it also dispatches, proving the fixture is a * genuine multi-job contention scenario (8.0+5.0+2.0=15.0 > 10.0 combined, * while the two jobs that DO dispatch, A+C, exactly saturate the 10.0 * total) rather than a degenerate single-denied-job case. * * White-box #include set: identical to test_sim_equivalence.c's own * combined normal-mode-driver + full-simulation-stack set, PLUS * ../src/resource.c (this file's own new dependency: the real * sched_check_host_satisfy()/sched_charge_account()/sched_commit_job_alloc() * capacity-and-accounting primitives under real exercise here). No existing * test file in this directory white-box #includes resource.c together with * the full sim/driver stack -- test_resource_accounting.c (task 1.1) only * pairs resource.c with core.c, deliberately staying below the * driver/picker/recording layer (see that file's own header comment for * why); this file is the first to need both together, since a genuine * recorded fixture needs the real recording pipeline AND real capacity * arithmetic in the same run. * * Stub surface: the union of test_sim_equivalence.c's own stubs (kill_sched_* * family, sched_account_jobs -- scheduler.c itself is still not white-box * included, so this thin ~20-line wrapper stays a no-op exactly as it does * there; harmless here since this fixture's single round never depends on * round-boundary re-accounting of a PRIOR round's charges, only its own * mid-round sequential charging), sched_publish_dispatch/ * sched_publish_pend_reason/xcecli_reply_schedadmin (driver.c link-time * dependencies, never invoked), plus ONE new stub this file's addition of * resource.c requires: sched_ctx_get_uncomit_host() (resource.c's * sched_try_alloc_on_host_()'s own link-time dependency -- that function is * compiled as part of resource.c but never called by this file, which only * calls the public sched_check_host_satisfy()/sched_charge_account()/ * sched_commit_job_alloc() entry points directly). Two stubs * test_resource_accounting.c's own resource.c-only inclusion needed * (plugin_prepare_allocation, sched_fill_limit_requests) are deliberately * NOT redefined here: this file's combined include set already provides * REAL definitions for both (plugin_prepare_allocation from plugin.c, * already part of test_sim_equivalence.c's own set; sched_fill_limit_requests * from resource.c itself, this file's own new addition) -- redefining * either here would be a duplicate-symbol link error, not a harmless * no-op. sched_worker_release_host_ctx is likewise NOT stubbed here (every * other file in this directory stubs it because none of them white-box * #include resource.c, which defines it for real) -- this file gets the * real one via resource.c and must not shadow it. */ #include #include #include #include #include #include #include #include #include #include "controller.capnp.h" #include "kvmgr.h" #include "preason.h" #include "scheduler.h" #include "xstring.h" /* Pull in the real functions under test into this TU (white-box #include, * same convention every other file in this directory establishes -- see * this file's own header comment for the exact combined set and why it is * novel). */ #include "../src/core.c" #include "../src/core_clock.c" #include "../src/core_events.c" #include "../src/core_session.c" #include "../src/decision.c" #include "../src/driver.c" #include "../src/plugin.c" #include "../src/plugins/builtin_fcfs.c" #include "../src/recording.c" #include "../src/recording_events.c" #include "../src/recording_session.c" #include "../src/repl.c" #include "../src/repl_fmt.c" #include "../src/repl_rollup.c" #include "../src/repl_break.c" #include "../src/repl_info.c" #include "../src/repl_params.c" #include "../src/resource.c" #include "../src/sim.c" #include "../src/sim_loop.c" #include "../src/sim_picker.c" /* bug-1567: sched_sim_driver_run() now calls sim_bootstrap_picker() (sim_loop.c-cap sibling) as an extern -- undefined reference at link time otherwise */ #include "../src/sim_limit.c" /* bug-1825: sched_sim_driver_run() now also calls sim_bootstrap_limit_plugin() (sim_loop.c-cap sibling, sim_picker.c's own bug-1567 precedent) as an extern -- undefined reference at link time otherwise */ #include "../src/sim_record.c" #include "../src/sim_verify.c" #include "../src/snapshot.c" #include "../src/snapshot_load.c" #include "../src/snapshot_wire.c" void kill_sched_job(void* arg) { (void)arg; } void kill_sched_queue(void* arg) { (void)arg; } void kill_sched_worker(void* arg) { (void)arg; } void kill_sched_user(void* arg) { (void)arg; } void kill_sched_project(void* arg) { (void)arg; } void kill_sched_resval(xce_val_t* val) { (void)val; } /* repl_info.c (scheduler-sim-diagnostics spec, task 2.1) now calls * limit_get_configured_limits() (plugins/limit.c, not part of this TU) to * resolve `info limit`'s name/all argument -- link-satisfying stub only, * same rationale as kill_sched_job()/kill_sched_resval() above; this * file's own fixtures never register the limit plugin, so NULL * (limit_get_configured_limits()'s own documented "plugin absent" return) * is also the behaviorally correct answer here, not just a link * placeholder. repl_info_limit() itself is not yet wired into * repl_cmd_info() (task 4), so no test in this file can reach it anyway. */ var limit_get_configured_limits(xce_sched_core_t* core) { (void)core; return NULL; } /* scheduler-sim-diagnostics spec, task 3.1: repl_info.c now also calls * limit_get_usage_instances() (plugins/limit.c, not part of this TU) to * read a limit's live usage instances -- identical link-satisfying stub, * same rationale as limit_get_configured_limits() immediately above; this * file's own fixtures never register the limit plugin, so NULL * (limit_get_usage_instances()'s own documented "plugin absent" return) is * also the behaviorally correct answer here, not just a link placeholder. */ var limit_get_usage_instances(xce_sched_core_t* core) { (void)core; return NULL; } /* scheduler-sim-diagnostics spec, task 6 (bug-1825): sim_loop.c's * sched_sim_driver_run() now also calls sim_bootstrap_limit_plugin() * (sim_limit.c, pulled into this TU via the #include above), which in * turn calls limit_new_plugin() (plugins/limit.c, not part of this TU) -- * identical link-satisfying stub rationale as limit_get_configured_ * limits()/limit_get_usage_instances() immediately above. Returning NULL * here is also the behaviorally correct answer, not just a link * placeholder: sim_bootstrap_limit_plugin() already handles a NULL * limit_new_plugin() return gracefully (logs an ERROR, registers nothing, * returns) -- the exact same "out of memory" path sim_bootstrap_picker() * (sim_picker.c) takes for a NULL builtin_fcfs_new_plugin() -- so this * file's fixtures simply end up with no limit plugin registered, same * observable state as before this task. */ xce_sched_plugin_t* limit_new_plugin(void) { return NULL; } /* limit_seed_config() (plugins/limit.c, not part of this TU) -- bug-1825 * remediation round 2's new honestly-typed config-seed entry point. * sim_bootstrap_limit_plugin() (sim_limit.c) only calls it after * plugin_register_core() has already succeeded against a real, non-NULL * limit_new_plugin() plugin -- unreachable in this file since the * limit_new_plugin() stub immediately above always returns NULL, but the * symbol must still resolve at link time. Same link-satisfying-stub * rationale as limit_new_plugin() immediately above. */ int limit_seed_config(xce_sched_plugin_t* plugin, xce_sched_core_t* core, xce_conf4sched_t* config) { (void)plugin; (void)core; (void)config; return 0; } /* xce_sched_driver_run_normal_mode_round()'s (driver.c) linker dependency -- * a no-op stand-in, same rationale test_sim_equivalence.c's own copy * documents (this file drives its own resource accounting directly via * this file's own capacity_aware_allocator_thread_fn(), calling the real * sched_check_host_satisfy()/sched_charge_account() -- scheduler.c's own * sched_account_jobs() thin wrapper is not white-box #included here). */ int sched_account_jobs(xce_sched_core_t* core) { (void)core; return 0; } /* xce_sched_driver_on_decision()'s (driver.c) publish-side linker * dependencies -- never actually invoked: this test never registers a * decision callback, so every decision is simply recorded and dropped. * Same rationale test_sim_equivalence.c's own copy documents. */ int sched_publish_dispatch(xce_scheduler_t* sched, xce_job_dispatch_t* d) { (void)sched; (void)d; return 0; } int sched_publish_pend_reason(xce_scheduler_t* sched, xce_preason_set_t* r) { (void)sched; (void)r; return 0; } /* handle_deferred_admin_record()'s (driver.c) linker dependency -- never * invoked (this test never enqueues XCE_SCHED_OP_ADMIN_RECORD). */ int xcecli_reply_schedadmin(xcecli_ctx_t c, uint64_t seq, sched_admin_reply_t* reply) { (void)c; (void)seq; (void)reply; return 0; } /* resource.c's sched_try_alloc_on_host_()'s own link-time dependency -- * never actually invoked (this file calls sched_check_host_satisfy()/ * sched_charge_account()/sched_commit_job_alloc() directly, never * sched_try_alloc_on_hosts()/sched_try_alloc_on_host_() itself). Same * link-satisfying-only stub test_resource_accounting.c's own resource.c * inclusion already establishes needing. */ int sched_ctx_get_uncomit_host(xvec_t** v, xce_sched_ctx_t* c, int i) { (void)v; (void)c; (void)i; return 0; } /* Real reimplementation of schedobj.c's sched_resolve_job_picker -- same * copy every white-box test in this directory that does not link * schedobj.c already establishes. */ int sched_resolve_job_picker(xce_sched_core_t* core, xce_sched_plugin_t** out) { xce_sched_plugin_t* plugin = NULL; int rc = 0; if ((core == NULL) || (out == NULL)) { return -EINVAL; } *out = NULL; if (core->job_pick_plugin == NULL) { return -EINVAL; } lock(core->plugins, RTTI_LOCK_READ); rc = key_find(core->plugins, core->job_pick_plugin, (void**)&plugin); unlock(core->plugins); if (rc <= 0) { return (rc < 0) ? rc : -ENOENT; } if ((plugin == NULL) || (plugin->picker_iface == NULL) || (plugin->picker_iface->next_job == NULL) || (plugin->picker_iface->begin_session == NULL) || (plugin->picker_iface->end_session == NULL) || (plugin->picker_iface->on_dispatch == NULL) || (plugin->picker_iface->on_pend == NULL)) { return -EINVAL; } *out = plugin; return 0; } /* scheduler-sim-live-allocation spec, task 4.1: sim_loop.c's real * sched_sim_driver_run() (white-box #include'd into this TU above) now * spawns sched_job_picker()/sched_allocator() (sched_election.c) directly * instead of its own, now-removed sim_picker_thread_fn() -- an undefined * reference at link time in this white-box TU otherwise, since * sched_election.c's own sched_allocator() calls the REAL * sched_make_allocation()/sched_commit_decision() (scheduler.c), neither * part of this file's own hand-picked white-box #include set (this file's * own header comment lists that set). This file's own test(s) do not * exercise genuine resource-matching allocation, so this link-satisfying * pair faithfully reproduces the exact PEND-only contract the REMOVED * sim_picker_thread_fn() itself used to provide -- split across the SAME * two-thread jbuf hand-off the real production pair now uses (one * producer, one consumer, matching the project's jbuf capacity-1 * rendezvous invariant) instead of the old single fused thread -- so this * file's own existing assertions are unaffected by this task's production * change. Genuine resource-based dispatch through the real, unmodified * sched_job_picker()/sched_allocator() pair is proved separately by * test_sim_live_allocation.c (task 4.1's own dedicated fixture, which * links the real scheduler_srcs instead of white-boxing individual files, * precisely so it can reach the real allocation chain this stub * deliberately does not). */ void* sched_job_picker(void* arg) { xce_sched_core_t* core = (xce_sched_core_t*)arg; xce_sched_plugin_t* picker = NULL; for (;;) { int go = 0; uint64_t jobid = 0; cbuf_get(&(core->wbuf), &go); if (sched_resolve_job_picker(core, &picker) < 0) { jobid = 0; cbuf_put(&(core->jbuf), &jobid); continue; } picker->picker_iface->begin_session(picker); for (;;) { xce_sched_job_t* job = picker->picker_iface->next_job(picker); if (job == NULL) { break; } jobid = job->jobid; cbuf_put(&(core->jbuf), &jobid); } jobid = 0; cbuf_put(&(core->jbuf), &jobid); picker->picker_iface->end_session(picker); } return NULL; } void* sched_allocator(void* arg) { xce_sched_core_t* core = (xce_sched_core_t*)arg; xce_sched_plugin_t* picker = NULL; for (;;) { uint64_t jobid = 0; xce_decision_t* decision = NULL; xce_sched_job_t* job; cbuf_get(&(core->jbuf), &jobid); if (jobid == 0) { decision = NULL; cbuf_put(&(core->decisions), &decision); continue; } job = sched_get_job(core, jobid); if (job == NULL) { continue; } (void)sched_resolve_job_picker(core, &picker); if (picker != NULL) { picker->picker_iface->on_pend(picker, job->jobid); } decision = make_pend_decision(make_pendreason(job->jobid)); cbuf_put(&(core->decisions), &decision); } return NULL; } /* scheduler-sim-live-allocation spec, task 4.1 (bug-1829 fix): sim_loop.c's * real sched_sim_driver_run() (white-box #include'd into this TU above) now * also calls sim_bootstrap_seqjob_allocator()/sim_bootstrap_parjob_ * allocator() (sim_alloc.c) -- undefined references at link time in this * white-box TU otherwise, since neither symbol is part of this file's own * hand-picked white-box #include set. This file's own test(s) exercise the * link-satisfying sched_job_picker()/sched_allocator() stub pair above * (which never reaches real primary allocation), so a true no-op is the * correct, behavior-preserving stand-in here -- the real bootstrap logic is * exercised separately by test_sim_live_allocation.c (task 4.1's own * dedicated fixture, which links the real scheduler_srcs). */ void sim_bootstrap_seqjob_allocator(xce_sched_sim_driver_t* sim) { (void)sim; } void sim_bootstrap_parjob_allocator(xce_sched_sim_driver_t* sim) { (void)sim; } void setUp(void) { return; } void tearDown(void) { return; } /* ---------------------------------------------------------------------- */ /* Fixture constants */ /* ---------------------------------------------------------------------- */ #define WORKER_NAME "worker-a" #define SLOTS_RESIDX 0 #define WORKER_CAPACITY 10.0f #define JOBA_ID 9101ULL #define JOBB_ID 9102ULL #define JOBC_ID 9103ULL #define JOBA_WANT 8.0f #define JOBB_WANT 5.0f #define JOBC_WANT 2.0f /* ---------------------------------------------------------------------- */ /* Fixture helpers */ /* ---------------------------------------------------------------------- */ static void make_tmp_dir(char* out, size_t out_len, const char* suffix) { char tmpl[PATH_MAX]; snprintf(tmpl, sizeof(tmpl), "/tmp/sched_simcontention%s_XXXXXX", suffix); TEST_ASSERT_NOT_NULL(mkdtemp(tmpl)); TEST_ASSERT_TRUE(strlen(tmpl) < out_len); strcpy(out, tmpl); } /* Real xce_sched_core_init() + real builtin_fcfs registration (same * technique test_sim_equivalence.c's own fixture_init() establishes) PLUS a * real xce_sched_worker_t added directly via sched_add_worker() with a * finite total capacity -- this file's own contention source. Adding the * worker directly (not through a WORKERS_PUSH event) matches * test_resource_accounting.c's own documented convention: "sched_add_worker() * itself does not [initialize resvals/resInUse] -- callers own it, same as * every other white-box fixture in this directory that adds a worker * directly." */ static void fixture_init(xce_scheduler_t* sched, char* record_dir) { xce_sched_plugin_t* fcfs; xce_sched_worker_t* w; int rc; int idx[1] = { SLOTS_RESIDX }; float val[1] = { WORKER_CAPACITY }; memset(sched, 0, sizeof(*sched)); mylock_init(&(sched->record_lock)); rc = xce_sched_core_init(&(sched->core)); TEST_ASSERT_EQUAL_INT(0, rc); sched->core.pend_updates[0] = (dataq_t*)calloc(1, sizeof(dataq_t)); dataq_init(sched->core.pend_updates[0], NULL); sched->core.pend_updates[1] = (dataq_t*)calloc(1, sizeof(dataq_t)); dataq_init(sched->core.pend_updates[1], NULL); fcfs = builtin_fcfs_new_plugin(); TEST_ASSERT_NOT_NULL(fcfs); rc = plugin_register_core(&(sched->core), fcfs); TEST_ASSERT_TRUE(rc >= 0); sched->core.job_pick_plugin = (char*)"builtin_fcfs"; sched->record_dir = record_dir; w = (xce_sched_worker_t*)calloc(1, sizeof(*w)); TEST_ASSERT_NOT_NULL(w); w->name = strdup(WORKER_NAME); TEST_ASSERT_NOT_NULL(w->name); rc = zarray_init(&(w->resvals), 8, NULL); TEST_ASSERT_EQUAL_INT(0, rc); rc = zarray_init(&(w->resInUse), 8, NULL); TEST_ASSERT_EQUAL_INT(0, rc); rc = put_resvals(&(w->resvals), 1, idx, val); TEST_ASSERT_TRUE(rc >= 0); rc = sched_add_worker(&(sched->core), w); TEST_ASSERT_TRUE(rc >= 0); } /* Builds one heap-owned, deep-freeable xce_queue_t -- same technique * test_sim_equivalence.c's own make_queue() establishes. */ static xce_queue_t* make_queue(const char* name, int32_t priority) { xce_queue_t* q = (xce_queue_t*)calloc(1, sizeof(*q)); q->queue = strdup(name); q->status = QUEUE_STAT_OPEN; q->priority = priority; return q; } /* Builds one heap-owned, deep-freeable xce_job_t -- same technique * test_sim_equivalence.c's own make_job() establishes. */ static xce_job_t* make_job(uint64_t jobid, uint64_t sub_time, const char* queue, const char* user) { xce_job_t* j = (xce_job_t*)calloc(1, sizeof(*j)); j->jobid = jobid; j->state = JOB_STAT_PEND; j->sub_time = sub_time; j->queue_time = sub_time; j->start_time = 0; j->submit = (job_submit_data_t*)calloc(1, sizeof(*(j->submit))); j->submit->queue = strdup(queue); j->submit->user = strdup(user); j->submit->project_name = strdup("contention_proj"); j->submit->resreq = strdup("type == any"); j->submit->mintasks = 1; j->submit->maxtasks = 1; return j; } static void enqueue_queue_update(xce_scheduler_t* sched, const char* name, int32_t priority) { xce_sched_update_request_t* req = (xce_sched_update_request_t*)calloc(1, sizeof(*req)); req->op = XCE_SCHED_OP_UPDATE_QUEUE; req->u.queueupdate.queue = make_queue(name, priority); dataq_enq(sched->core.pend_updates[0], &(req->link)); } static void enqueue_jobs_push(xce_scheduler_t* sched, xce_job_t** jobs, int njobs) { xce_sched_update_request_t* req = (xce_sched_update_request_t*)calloc(1, sizeof(*req)); req->op = XCE_SCHED_OP_UPDATE_JOB; req->u.jobupdate.njobs = njobs; req->u.jobupdate.jobs = jobs; dataq_enq(sched->core.pend_updates[0], &(req->link)); } /* ---------------------------------------------------------------------- */ /* Picker thread -- identical to test_sim_equivalence.c's own */ /* picker_thread_fn(): resolves and drives the REAL builtin_fcfs picker. */ /* ---------------------------------------------------------------------- */ static void* picker_thread_fn(void* arg) { xce_sched_core_t* core = (xce_sched_core_t*)arg; xce_sched_plugin_t* job_picker = NULL; int rc; int go = 0; uint64_t jobid; rc = sched_resolve_job_picker(core, &job_picker); if (rc < 0) { return NULL; } cbuf_get(&(core->wbuf), &go); job_picker->picker_iface->begin_session(job_picker); for (;;) { xce_sched_job_t* job = job_picker->picker_iface->next_job(job_picker); if (job == NULL) { break; } jobid = job->jobid; cbuf_put(&(core->jbuf), &jobid); } jobid = 0; cbuf_put(&(core->jbuf), &jobid); job_picker->picker_iface->end_session(job_picker); return NULL; } /* ---------------------------------------------------------------------- */ /* Capacity-aware stand-in allocator -- unlike test_sim_equivalence.c's own */ /* stand_in_allocator_thread_fn() (which unconditionally PENDs), this one */ /* drives the REAL sched_check_host_satisfy()/sched_charge_account()/ */ /* sched_commit_job_alloc() (resource.c) against the real worker capacity */ /* fixture_init() set up -- see this file's own header comment for why a */ /* per-job "want" lookup table stands in for full resreq-DSL compilation. */ /* ---------------------------------------------------------------------- */ static float lookup_want(uint64_t jobid) { if (jobid == JOBA_ID) { return JOBA_WANT; } if (jobid == JOBB_ID) { return JOBB_WANT; } if (jobid == JOBC_ID) { return JOBC_WANT; } return 0.0f; } /* SESSION_FROM - the `from` an allocation decided in a session that began at * `begin_ns` (epoch nanoseconds) carries: the session start in whole * seconds, floor division (libxce-alloc-preason-equality Decision 8; * scheduler.c's sched_make_primary_alloc_() stamps it, and in simulation * mode the session starts at the recorded beginRound). This file's * capacity-aware stand-in allocator stamps its DISPATCH allocations the * same way, so the recording it produces carries the value a live replay * of that round reproduces. */ #define SESSION_FROM(begin_ns) ((uint64_t)(begin_ns) / 1000000000ULL) /* Builds the wire-shaped xce_job_alloc_t a DISPATCH decision carries -- same * field shape test_sim_verify.c's own make_test_alloc()/make_test_host_alloc()/ * make_test_job_alloc() establish. */ static xce_job_alloc_t* make_dispatch_alloc(uint64_t jobid, const char* host, float value, uint64_t from) { xce_alloc_t* res = (xce_alloc_t*)calloc(1, sizeof(*res)); xce_host_alloc_t* h = (xce_host_alloc_t*)calloc(1, sizeof(*h)); xce_host_alloc_t** hosts = (xce_host_alloc_t**)calloc(1, sizeof(*hosts)); xce_alloc_t** allocs = (xce_alloc_t**)calloc(1, sizeof(*allocs)); xce_job_alloc_t* j = (xce_job_alloc_t*)calloc(1, sizeof(*j)); TEST_ASSERT_NOT_NULL(res); TEST_ASSERT_NOT_NULL(h); TEST_ASSERT_NOT_NULL(hosts); TEST_ASSERT_NOT_NULL(allocs); TEST_ASSERT_NOT_NULL(j); res->resname = strdup("slots"); res->reserve = strdup("run"); res->instid = 1; res->value = value; res->from = from; res->duration = 0; allocs[0] = res; h->host = strdup(host); h->nallocs = 1; h->allocs = allocs; hosts[0] = h; j->jobid = jobid; j->nhosts = 1; j->allocs = hosts; j->nres_shared = 0; j->allocs_shared = NULL; return j; } /* Builds the accounting-charge shape sched_charge_account()/ * sched_commit_job_alloc() consume -- same field shape * test_resource_accounting.c's own make_test_account() establishes, * simplified to one host charge (no shared-resource charge needed here). */ static xce_job_account_t* make_charge_account(const char* host, float value) { xce_job_account_t* acct = (xce_job_account_t*)calloc(1, sizeof(*acct)); TEST_ASSERT_NOT_NULL(acct); acct->nres_shared = 0; acct->accounts_shared = NULL; acct->nhosts = 1; acct->accounts = (xce_host_account_t**)calloc(1, sizeof(*acct->accounts)); TEST_ASSERT_NOT_NULL(acct->accounts); acct->accounts[0] = (xce_host_account_t*)calloc(1, sizeof(xce_host_account_t)); TEST_ASSERT_NOT_NULL(acct->accounts[0]); acct->accounts[0]->host = strdup(host); acct->accounts[0]->naccount = 1; acct->accounts[0]->accounts = (xce_res_account_t**)calloc(1, sizeof(*acct->accounts[0]->accounts)); TEST_ASSERT_NOT_NULL(acct->accounts[0]->accounts); acct->accounts[0]->accounts[0] = (xce_res_account_t*)calloc(1, sizeof(xce_res_account_t)); TEST_ASSERT_NOT_NULL(acct->accounts[0]->accounts[0]); acct->accounts[0]->accounts[0]->residx = SLOTS_RESIDX; acct->accounts[0]->accounts[0]->used = value; acct->accounts[0]->accounts[0]->reserved = 0.0; return acct; } static void* capacity_aware_allocator_thread_fn(void* arg) { xce_scheduler_t* sched = (xce_scheduler_t*)arg; xce_sched_core_t* core = &(sched->core); xce_sched_plugin_t* picker = NULL; uint64_t jobid; if (sched_resolve_job_picker(core, &picker) < 0) { xce_decision_t* decision = NULL; cbuf_get(&(core->jbuf), &jobid); /* drain the lone 0 sentinel */ cbuf_put(&(core->decisions), &decision); return NULL; } for (;;) { xce_decision_t* decision; xce_sched_job_t* job; xce_sched_worker_t* w; xvec_t uncomitted; xvec_idval_t wanted; xvec_idval_t unmeet; float want; int rc; cbuf_get(&(core->jbuf), &jobid); if (jobid == 0) { decision = NULL; cbuf_put(&(core->decisions), &decision); break; } job = sched_get_job(core, jobid); w = sched_get_worker(core, (char*)WORKER_NAME); TEST_ASSERT_NOT_NULL(job); TEST_ASSERT_NOT_NULL(w); want = lookup_want(jobid); /* uncomitted -- no in-flight, not-yet-committed reservations across * this fixture's strictly-sequential, single-threaded evaluation of * one candidate at a time (each job's charge is fully committed via * sched_commit_job_alloc() before the next candidate is checked), so * a fresh, zero-initialized vector is correct here, matching * sched_check_host_satisfy()'s own documented caller-owned * contract. */ rc = xvec_init(&uncomitted, 1, sizeof(float)); TEST_ASSERT_TRUE(rc >= 0); rc = xvec_idval_init(&wanted, 1); TEST_ASSERT_EQUAL_INT(0, rc); rc = xvec_idval_store(&wanted, SLOTS_RESIDX, want); TEST_ASSERT_TRUE(rc >= 0); /* scheduler-sim-live-allocation task 1.3: sched_check_host_ * satisfy() is now core-typed -- pass the already-derived `core` * local. */ rc = sched_check_host_satisfy(core, w, &uncomitted, &wanted, &unmeet); xvec_fini(&uncomitted); xvec_idval_fini(&wanted); xvec_idval_fini(&unmeet); if (rc == 1) { /* Satisfied -- charge the real accounting state (resource.c, * task 1.1's relocated sched_charge_account()/ * sched_commit_job_alloc()) so the NEXT candidate in this same * round sees updated resInUse, then report DISPATCH via the * real make_dispatch_decision() constructor -- mirrors * scheduler.c's real sched_allocator()'s own * on_dispatch/commit/decision sequence. */ job->accounts = make_charge_account(WORKER_NAME, want); sched_commit_job_alloc(core, job); /* Stamp `from` exactly as sched_make_primary_alloc_() * (scheduler.c) stamps a real allocation: the deciding * session's start in whole seconds. */ job->allocs = make_dispatch_alloc( jobid, WORKER_NAME, want, SESSION_FROM(core->session_start)); picker->picker_iface->on_dispatch(picker, jobid); decision = make_dispatch_decision(job->allocs); } else { /* Denied -- this job is genuinely blocked by capacity a * currently-running job (already charged above, earlier this * same round) is consuming. */ picker->picker_iface->on_pend(picker, jobid); decision = make_pend_decision(make_pendreason(jobid)); } if (decision != NULL) { cbuf_put(&(core->decisions), &decision); } } return NULL; } /* Runs one complete normal-mode round (real picker thread + this file's own * capacity-aware allocator thread) and blocks until it completes -- same * design as test_sim_equivalence.c's own run_one_real_session(). */ static void run_one_real_session(xce_scheduler_t* sched) { pthread_t picker_tid; pthread_t allocator_tid; int rc; rc = pthread_create(&picker_tid, NULL, picker_thread_fn, &(sched->core)); TEST_ASSERT_EQUAL_INT(0, rc); rc = pthread_create(&allocator_tid, NULL, capacity_aware_allocator_thread_fn, sched); TEST_ASSERT_EQUAL_INT(0, rc); xce_sched_driver_run_normal_mode_round(sched); rc = pthread_join(picker_tid, NULL); TEST_ASSERT_EQUAL_INT(0, rc); rc = pthread_join(allocator_tid, NULL); TEST_ASSERT_EQUAL_INT(0, rc); } /* ---------------------------------------------------------------------- */ /* Decoding */ /* ---------------------------------------------------------------------- */ /* This fixture pushes exactly 2 events in one batch (1 QUEUE_UPDATE + 1 * UPDATE_JOB carrying all 3 jobs) before the round runs -- both * sched_record_start()'s own ReplayHeader+initial-SnapshotEntry pair and * those 2 events come before the ReplaySession message, same convention * test_sim_equivalence.c's own N_MSGS_BEFORE_SESSION documents (that file * has 3 events instead of this file's 2, since it pushes 2 queues). */ #define N_MSGS_BEFORE_SESSION 4 /* ReplayHeader, SnapshotEntry, 2 events */ /* Opens `path`, skips exactly `n_skip` capnp messages structurally, then * decodes the NEXT message as a ReplaySession -- same technique * test_sim_equivalence.c's own decode_session_after_skip() establishes. */ static replay_session_t* decode_session_after_skip(const char* path, int n_skip) { FILE* f; struct capn c; ReplaySession_ptr p; replay_session_t* dst = NULL; int i; f = fopen(path, "r"); TEST_ASSERT_NOT_NULL(f); for (i = 0; i < n_skip; i++) { TEST_ASSERT_EQUAL_INT(0, capn_init_fp(&c, f, 0)); capn_free(&c); } TEST_ASSERT_EQUAL_INT(0, capn_init_fp(&c, f, 0)); p.p = capn_getp(capn_root(&c), 0, 1); decode_ReplaySession_ptr(&dst, p); capn_free(&c); fclose(f); return dst; } /* ---------------------------------------------------------------------- */ /* The fixture's own shape proof (task 1.2's own Observable Completion). */ /* ---------------------------------------------------------------------- */ /* * test_contention_fixture_denies_capacity_to_a_job_a_running_job_holds - * builds the fixture (real recorded normal-mode session, 3 jobs, 1 * capacity-limited worker), then decodes the recorded ReplaySession and * proves -- from the ACTUAL decoded decision stream, not merely this file's * own input constants -- that job A and job C dispatched (consuming the * worker's full 10.0-unit capacity between them) while job B was denied, * genuinely blocked by capacity job A (running) holds. */ static void test_contention_fixture_denies_capacity_to_a_job_a_running_job_holds(void) { char record_dir[PATH_MAX]; xce_scheduler_t sched; char r_path[PATH_MAX]; xce_job_t** jobs; replay_session_t* r_session = NULL; int rc; int i; int n_dispatch = 0; int n_pend = 0; int pend_jobid_ok = 0; float dispatched_total = 0.0f; make_tmp_dir(record_dir, sizeof(record_dir), "_r"); fixture_init(&sched, record_dir); rc = sched_record_start(&sched, "contention.replay"); TEST_ASSERT_EQUAL_INT(0, rc); TEST_ASSERT_TRUE(strlen(sched.record_file_path) < sizeof(r_path)); strcpy(r_path, sched.record_file_path); enqueue_queue_update(&sched, "normal", 1); jobs = (xce_job_t**)calloc(3, sizeof(xce_job_t*)); jobs[0] = make_job(JOBA_ID, 1000ULL, "normal", "alice"); jobs[1] = make_job(JOBB_ID, 1001ULL, "normal", "bob"); jobs[2] = make_job(JOBC_ID, 1002ULL, "normal", "carol"); enqueue_jobs_push(&sched, jobs, 3); run_one_real_session(&sched); /* Sanity check on the recorded batch shape (1 queue update + 1 batched * jobs-push = 2 events) -- pins N_MSGS_BEFORE_SESSION's own assumption. */ TEST_ASSERT_EQUAL_UINT64(2ULL, sched.record_event_count); rc = sched_record_stop(&sched); TEST_ASSERT_EQUAL_INT(0, rc); r_session = decode_session_after_skip(r_path, N_MSGS_BEFORE_SESSION); TEST_ASSERT_NOT_NULL(r_session); /* Non-tautology guard: a fixture with fewer than 3 recorded decisions, * or one that is all-DISPATCH/all-PEND, would not prove multi-job * contention. */ TEST_ASSERT_EQUAL_INT(3, r_session->n_decisions); for (i = 0; i < r_session->n_decisions; i++) { replay_decision_t* d = r_session->decisions[i]; if (d->body_which == ReplayDecision_body_dispatch) { uint64_t jobid; float value; TEST_ASSERT_NOT_NULL(d->body.dispatch); TEST_ASSERT_NOT_NULL(d->body.dispatch->alloc); jobid = d->body.dispatch->alloc->jobid; TEST_ASSERT_TRUE((jobid == JOBA_ID) || (jobid == JOBC_ID)); TEST_ASSERT_EQUAL_INT(1, d->body.dispatch->alloc->nhosts); TEST_ASSERT_EQUAL_STRING( WORKER_NAME, d->body.dispatch->alloc->allocs[0]->host); value = d->body.dispatch->alloc->allocs[0]->allocs[0]->value; if (jobid == JOBA_ID) { TEST_ASSERT_EQUAL_FLOAT(JOBA_WANT, value); } else { TEST_ASSERT_EQUAL_FLOAT(JOBC_WANT, value); } dispatched_total += value; n_dispatch++; } else if (d->body_which == ReplayDecision_body_pend) { TEST_ASSERT_NOT_NULL(d->body.pend); if (d->body.pend->jobid == JOBB_ID) { pend_jobid_ok = 1; } n_pend++; } else { TEST_FAIL_MESSAGE("unexpected decision kind in contention " "fixture"); } } /* This task's own Observable Completion: exactly one job (B) is denied * capacity, exactly two (A, C) dispatch. */ TEST_ASSERT_EQUAL_INT(2, n_dispatch); TEST_ASSERT_EQUAL_INT(1, n_pend); TEST_ASSERT_TRUE_MESSAGE(pend_jobid_ok, "the denied decision must genuinely be job B, " "not merely any PEND"); /* The combined-capacity proof: all three jobs' wants exceed total * capacity (insufficient for all of them at once), while the two jobs * that DID dispatch fit exactly within it -- i.e. job B was denied * capacity that jobs A and C, both running, are genuinely consuming. */ TEST_ASSERT_TRUE((JOBA_WANT + JOBB_WANT + JOBC_WANT) > WORKER_CAPACITY); TEST_ASSERT_TRUE(dispatched_total <= WORKER_CAPACITY); TEST_ASSERT_TRUE((dispatched_total + JOBB_WANT) > WORKER_CAPACITY); free_ReplaySession_ptr(&r_session); printf("test_contention_fixture_denies_capacity_to_a_job_a_running_job_" "holds: PASS\n"); } /* * test_contention_fixture_is_replayable_via_simulation_driver - this task's * own "dual-use" requirement: the SAME `.replay` file the normal-mode * recording above produced is directly consumable by the REAL, unmodified * sched_sim_driver_run() as simulation-mode replay input, with no * conversion step. (Whether simulation mode's own resInUse/decision output * correctly reflects this fixture's contention is the later comparison * task 4.1 proves -- out of this task's own boundary; this test only proves * the file is a valid, replayable simulation-mode input.) */ static void test_contention_fixture_is_replayable_via_simulation_driver(void) { char record_dir[PATH_MAX]; xce_scheduler_t sched; char r_path[PATH_MAX]; xce_job_t** jobs; sched_sim_args_t sim_args; xce_sched_sim_driver_t sim; int rc; make_tmp_dir(record_dir, sizeof(record_dir), "_dualuse"); fixture_init(&sched, record_dir); rc = sched_record_start(&sched, "contention_dualuse.replay"); TEST_ASSERT_EQUAL_INT(0, rc); TEST_ASSERT_TRUE(strlen(sched.record_file_path) < sizeof(r_path)); strcpy(r_path, sched.record_file_path); enqueue_queue_update(&sched, "normal", 1); jobs = (xce_job_t**)calloc(3, sizeof(xce_job_t*)); jobs[0] = make_job(JOBA_ID, 1000ULL, "normal", "alice"); jobs[1] = make_job(JOBB_ID, 1001ULL, "normal", "bob"); jobs[2] = make_job(JOBC_ID, 1002ULL, "normal", "carol"); enqueue_jobs_push(&sched, jobs, 3); run_one_real_session(&sched); rc = sched_record_stop(&sched); TEST_ASSERT_EQUAL_INT(0, rc); memset(&sim_args, 0, sizeof(sim_args)); sim_args.replay_path = r_path; sim_args.sim_start_ts = SCHED_SIM_START_TS_LATEST; sim_args.sim_speed = 0; /* frozen -- deterministic, no wall-clock skew */ memset(&sim, 0, sizeof(sim)); rc = sched_sim_driver_run(&sim, &sim_args); TEST_ASSERT_EQUAL_INT_MESSAGE( 0, rc, "the same file the normal-mode recording produced must be directly " "replayable by the real, unmodified simulation driver"); printf("test_contention_fixture_is_replayable_via_simulation_driver: " "PASS\n"); } int main(void) { UNITY_BEGIN(); RUN_TEST(test_contention_fixture_denies_capacity_to_a_job_a_running_job_holds); RUN_TEST(test_contention_fixture_is_replayable_via_simulation_driver); return UNITY_END(); }