lei.chen/myxce
1
0
Fork 1
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

test(scheduler-sim-decision-parity): add resource-contention replay fixture (task 1.2)

Browse source 
Adds a dual-use .replay fixture (3 jobs competing for insufficient worker
capacity) that later tasks need to prove capacity-aware parity under
contention: a real recorded normal-mode session (built by driving the
genuine sched_check_host_satisfy()/sched_commit_job_alloc() capacity/
accounting chain, not mocked) that is also directly replayable through the
real simulation driver. Two tests: fixture shape (job B genuinely denied
while A+C hold the full 10.0-unit capacity, verified against the decoded
recorded decision stream) and sim-driver replayability.

RED-phase proven: raising WORKER_CAPACITY to eliminate contention makes the
shape assertion fail as expected before reverting to the genuine-contention
value. 44/45 scheduler tests pass; the 1 failure is pre-existing bug-1560
(unrelated).

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
This commit is contained in:
Rongsong Shen 2026-08-22 06:37:04 +00:00
parent 3106c82ed0
commit f11f0df3bd
No known key found for this signature in database
GPG key ID: 5BCDAAE0123E9FE2
3 changed files with 913 additions and 1 deletions
Download patch file Download diff file Expand all files Collapse all files

5
.kiro/specs/scheduler-sim-decision-parity/tasks.md
View file

@ -9,7 +9,7 @@
- _Requirements: 1.1, 1.2, 1.3, 1.4, 2.1, 2.2, 5.3_
- _Boundary: Core-Level Resource Accounting_
- [ ] 1.2 (P) Build a resource-contention replay fixture for testing
- [x] 1.2 (P) Build a resource-contention replay fixture for testing
- Create a recorded-session test fixture with multiple jobs competing for capacity insufficient to run all of them at once (existing fixtures only ever have enough capacity for every job, which cannot exercise or prove capacity-check correctness)
- Fixture must be usable both as a normal-mode recording source and as a simulation-mode replay input, so the same scenario can be run through both drivers and compared
- Observable completion: a scheduling round exists in the fixture where at least one job is denied capacity that a currently-running job is consuming
@ -86,3 +86,6 @@
- _Requirements: 1.1, 1.2, 1.3, 3.1, 3.2, 3.3_
- _Depends: 4.1, 4.2_
- _Boundary: Scheduler test suite_
## Implementation Notes
- (task 1.2) Whole-scenario white-box test files in `src/scheduler/testsuite/` (fixture construction + real driver/allocator plumbing) routinely exceed the project's 800-line guidance (`test_sim_verify.c` is 934 lines, uncorrected; the new `test_sim_resource_contention.c` is 883) — the steering's "don't use as precedent" caveat names only non-test files, so this is an accepted pattern for this class of test, not a defect to fix per task. Tasks 4.1-4.3 will likely add similarly-sized files; don't flag file size alone as blocking for them.

26
src/scheduler/testsuite/meson.build
View file

@ -726,6 +726,29 @@ sched_sim_equivalence_test = executable(
dependencies: [scheduler_test_deps],
install: false)
# Resource-contention replay fixture (scheduler-sim-decision-parity spec,
# task 1.2, Req 2.1/2.2): existing fixtures in this directory never have
# scarce capacity, so none can exercise or prove capacity-check correctness.
# This file records ONE real normal-mode session (real builtin_fcfs picker +
# a capacity-aware stand-in allocator that drives the REAL
# sched_check_host_satisfy()/sched_charge_account()/sched_commit_job_alloc()
# (resource.c) against a real, finite-capacity worker) where three jobs
# compete for 10.0 units of one resource -- two dispatch, one is genuinely
# denied capacity the other two, now running, are consuming. First file in
# this directory to combine resource.c with the full driver/sim/recording
# stack (test_resource_accounting.c, task 1.1, deliberately pairs resource.c
# with only core.c -- see that file's own header comment for why). The
# resulting `.replay` file is dual-use: it IS a genuine normal-mode
# recording, and this file's own second test proves it is also directly
# replayable via the real, unmodified simulation driver -- see this file's
# own header comment for the full rationale.
sched_sim_resource_contention_test = executable(
'sched_sim_resource_contention_test',
['test_sim_resource_contention.c'],
include_directories:[incdir],
dependencies: [scheduler_test_deps],
install: false)
# Contract test for the interactive simulation REPL's execution-control slice
# (scheduler-core spec, task 12.1, Req 5 AC 22): starting simulation with
# --interactive pauses before any event is processed, and a `step N` command
@ -1050,6 +1073,9 @@ test('scheduler sim record output test',
test('scheduler sim equivalence test',
sched_sim_equivalence_test, suite: 'scheduler',
env: ['ASAN_OPTIONS=detect_leaks=0'])
test('scheduler sim resource contention test',
sched_sim_resource_contention_test, suite: 'scheduler',
env: ['ASAN_OPTIONS=detect_leaks=0'])
test('scheduler repl execution control test',
sched_repl_execution_control_test, suite: 'scheduler',
env: ['ASAN_OPTIONS=detect_leaks=0'])

883
src/scheduler/testsuite/test_sim_resource_contention.c Normal file
View file

@ -0,0 +1,883 @@
/*
* 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 <errno.h>
#include <limits.h>
#include <pthread.h>
#include <stdint.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>
#include <unity/unity.h>
#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_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_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; }
/* 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;
}
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;
}
/* 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)
{
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 = 0;
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);
rc = sched_check_host_satisfy(sched, 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);
job->allocs = make_dispatch_alloc(jobid, WORKER_NAME, want);
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();
}