From c1960457f892392172c1caaf47540706c69a8c95 Mon Sep 17 00:00:00 2001 From: Rongsong Shen Date: Mon, 28 Sep 2026 10:14:35 +0000 Subject: [PATCH] feat(libxce-alloc-preason-equality): stamp every allocation's from with the deciding session start (task 3.2) Co-Authored-By: Claude Opus 5.5 (1M context) --- .../libxce-alloc-preason-equality/tasks.md | 4 +- src/scheduler/src/scheduler.c | 57 ++ src/scheduler/testsuite/meson.build | 26 + .../testsuite/test_sched_alloc_from_stamp.c | 569 ++++++++++++++++++ 4 files changed, 654 insertions(+), 2 deletions(-) create mode 100644 src/scheduler/testsuite/test_sched_alloc_from_stamp.c diff --git a/.kiro/specs/libxce-alloc-preason-equality/tasks.md b/.kiro/specs/libxce-alloc-preason-equality/tasks.md index 8b1a3ee3..ce0da2a1 100644 --- a/.kiro/specs/libxce-alloc-preason-equality/tasks.md +++ b/.kiro/specs/libxce-alloc-preason-equality/tasks.md @@ -28,14 +28,14 @@ Test-first throughout (red → green → refactor). Every sub-task leaves the bu - _Boundary: libxce pending-reason module; the allocation OOM test binary (shared test harness, sequenced after 2.1)_ - _Depends: 2.1_ -- [ ] 3. Core: reproducible reservation start time in the scheduler +- [x] 3. Core: reproducible reservation start time in the scheduler - [x] 3.1 (P) Record each session's start from the same clock reading as the session start - Write the test first, in its own test function (not shared with 3.2): with recording enabled, a recorded session's begin time equals the session's start time exactly. - Take the recorded begin time from the session start instead of a second clock reading (design.md, Components: Recording). - Done when the test passes and the existing recording tests still pass. - _Requirements: 4.2, 6.5_ - _Boundary: scheduler normal-mode driver_ -- [ ] 3.2 (P) Stamp every allocation with its deciding session's start time, from the session core +- [x] 3.2 (P) Stamp every allocation with its deciding session's start time, from the session core - Write the tests first, in test functions separate from 3.1's: every allocation of one normal-mode session, including every primary of a merged COMPOUND request, carries the session start in whole seconds; the same holds in simulation mode, where the allocators' scheduler handle is the simulator's session-less wrapper, using a realistic epoch-scale recorded session start so a stamp taken from the wrong core (0) fails. - Stamp the reservation start time on every per-host and shared allocation record where the allocator result is received with the session core, not in the record-building helper (design.md, Components: `from` stamping, Decision 8). - Confirm no allocation path bypasses that point; if one does, stop and raise it. diff --git a/src/scheduler/src/scheduler.c b/src/scheduler/src/scheduler.c index cc993481..581b4409 100644 --- a/src/scheduler/src/scheduler.c +++ b/src/scheduler/src/scheduler.c @@ -182,6 +182,56 @@ static void sched_pend_from_cached_reason_(xce_sched_ctx_t* ctx, int category) add_job_preason(ctx->reason->reasons, category, NULL, NULL); } +/* + * sched_stamp_session_from_ - libxce-alloc-preason-equality Req 4.1/4.4, + * design.md Decision 8: stamp every allocation record the allocator just + * produced (each per-host allocs[i]->allocs[j] and each allocs_shared[k]) + * with the deciding session's start, in whole seconds since the epoch + * (core->session_start is nanoseconds; floor division). make_alloc()'s + * own time(NULL) default is not reproducible under replay. + * + * Done here, with the SESSION core sched_make_primary_alloc_() receives, + * not in sched_make_allocs(): in simulation mode that helper's `sched` is + * sim_alloc.c's session-less wrapper, whose core's session_start is 0. + * Every allocation (PRIMARY, each primary of a COMPOUND before + * merge_job_alloc() folds it, each ALTER alternative) passes through + * sched_make_primary_alloc_(), so all of one session's allocations carry + * the same `from`. + */ +#define SCHED_NSEC_PER_SEC 1000000000ULL + +static void sched_stamp_session_from_(const xce_sched_core_t* core, + xce_job_alloc_t* alloc) +{ + uint64_t from = core->session_start / SCHED_NSEC_PER_SEC; + int i; + int j; + + if (alloc == NULL) { + return; + } + + for (i = 0; (alloc->allocs != NULL) && (i < alloc->nhosts); i++) { + xce_host_alloc_t* ha = alloc->allocs[i]; + + if ((ha == NULL) || (ha->allocs == NULL)) { + continue; + } + for (j = 0; j < ha->nallocs; j++) { + if (ha->allocs[j] != NULL) { + ha->allocs[j]->from = from; + } + } + } + + for (i = 0; (alloc->allocs_shared != NULL) && (i < alloc->nres_shared); + i++) { + if (alloc->allocs_shared[i] != NULL) { + alloc->allocs_shared[i]->from = from; + } + } +} + static int sched_make_primary_alloc_(xce_sched_core_t* core, xce_sched_ctx_t* ctx) { @@ -214,6 +264,13 @@ static int sched_make_primary_alloc_(xce_sched_core_t* core, rc = sched_make_parjob_allocation(core, ctx); } + /* Req 4.1/4.4: the allocator's result is received here with the + * session core -- stamp every record's `from` before a COMPOUND's + * merge_job_alloc() or the dispatch decision sees it. */ + if ((rc > 0) && (ctx->alloc != NULL)) { + sched_stamp_session_from_(core, ctx->alloc); + } + /* Req 7.3: once the chosen allocator has exhausted every candidate * plan (the not-satisfied signal, rc == 0), populate the cache with * this resreq's fingerprint + identity and the exhausted attempt's own diff --git a/src/scheduler/testsuite/meson.build b/src/scheduler/testsuite/meson.build index ade18318..d4c112dd 100644 --- a/src/scheduler/testsuite/meson.build +++ b/src/scheduler/testsuite/meson.build @@ -2307,3 +2307,29 @@ sched_sim_e2e_limit_visibility_test = executable( test('scheduler sim e2e limit visibility test', sched_sim_e2e_limit_visibility_test, suite: 'scheduler', env: ['ASAN_OPTIONS=detect_leaks=0']) + +# libxce-alloc-preason-equality spec, task 3.2 (Req 4.1, 4.4, 6.5): every +# allocation of one session -- normal mode (real sched_init()) and +# simulation mode (sim_alloc.c's session-less wrapper allocators) -- +# carries the session start in whole seconds as `from`, stamped by +# sched_make_primary_alloc_() from the session core. Same real- +# scheduler_srcs link technique as sched_parjob_e2e_multislot_test. +sched_alloc_from_stamp_test_srcs = ['test_sched_alloc_from_stamp.c'] +foreach f : scheduler_srcs + if f != 'src/main.c' + sched_alloc_from_stamp_test_srcs += ['../' + f] + endif +endforeach + +sched_alloc_from_stamp_test = executable( + 'sched_alloc_from_stamp_test', + sched_alloc_from_stamp_test_srcs, + include_directories:[incdir], + dependencies: [scheduler_deps, unity_dep], + link_args: ['-Wl,--wrap=xcecli_start', + '-Wl,--wrap=xcecli_join'], + install: false) + +test('scheduler alloc from stamp test', + sched_alloc_from_stamp_test, suite: 'scheduler', + env: ['ASAN_OPTIONS=detect_leaks=0']) diff --git a/src/scheduler/testsuite/test_sched_alloc_from_stamp.c b/src/scheduler/testsuite/test_sched_alloc_from_stamp.c new file mode 100644 index 00000000..ffc183b0 --- /dev/null +++ b/src/scheduler/testsuite/test_sched_alloc_from_stamp.c @@ -0,0 +1,569 @@ +/* + * test_sched_alloc_from_stamp.c - libxce-alloc-preason-equality spec, task + * 3.2 (Req 4.1, 4.4, 4.5, 6.5, 6.6; design.md "`from` stamping + * (`scheduler.c`)", Decision 8). + * + * Every allocation a scheduling session produces must carry, as its + * reservation start time `from`, the deciding session's start in whole + * seconds (core->session_start is nanoseconds; `from` is + * session_start / 10^9, floor) -- not the wall-clock second make_alloc() + * happened to run in. The stamp is applied where the allocator's result is + * received together with the SESSION core (sched_make_primary_alloc_(), + * scheduler.c), never in sched_make_allocs(): in simulation mode that + * helper's `sched` is sim_alloc.c's session-less wrapper, whose own core + * never begins a session (session_start == 0). + * + * Two tests, one per driver shape, both going through the real, public + * sched_make_allocation() chain and the real seqjob/parjob allocator + * plugins (real scheduler_srcs link, same technique as test_parjob_e2e_ + * multislot.c/test_sim_alloc_limit_fanout.c): + * + * - normal mode: a real sched_init()'d scheduler, real wire ingest, the + * live allocator thread; one session dispatches a plain PRIMARY job + * and a two-primary COMPOUND job (merged by merge_job_alloc()). + * - simulation mode: a bare xce_sched_sim_driver_t bootstrapped with + * sim_bootstrap_seqjob_allocator()/sim_bootstrap_parjob_allocator() + * (so the allocators' plugin->sched is the session-less wrapper); + * sched_make_allocation(&sim.core, ...) called directly. + * + * Both sessions start at a realistic epoch-scale time deliberately far from + * "now", so neither a time(NULL) stamp (the old make_alloc() default) nor a + * stamp taken from the wrong core (0) can pass. + * + * Fixtures are never torn down (same convention as every other real- + * scheduler_srcs test in this directory). + */ +#include +#include +#include +#include +#include +#include + +#include + +#include "client.h" +#include "controller.capnp.h" +#include "core.h" +#include "decision.h" +#include "plugin.h" +#include "resobj.h" +#include "resreq.h" +#include "sched/alloc_cache.h" +#include "sched/types.h" +#include "scheduler.h" +#include "sim.h" +#include "xce/cluster.h" +#include "xstring.h" + +extern int scheduler_req_handler(xce_request_t*, void*); +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 NSEC_PER_SEC 1000000000ULL + +/* 2023-11-14T22:13:20.123456789Z -- epoch-scale, but years away from any + * time(NULL) this test can run at, and not a whole second (so the floor + * division is exercised). */ +#define NORMAL_SESSION_EPOCH_NS ((uint64_t)1700000000123456789ULL) +/* A recorded session start (beginRound) for the simulation test, also + * epoch-scale and far from now. */ +#define SIM_RECORDED_BEGIN_NS ((uint64_t)1690000000987654321ULL) + +#define WORKER_A_NAME "from-stamp-host-a" +#define WORKER_B_NAME "from-stamp-host-b" +#define WORKER_SLOTS 2.0f +#define QUEUE_NAME "from_stamp_queue" + +#define PRIMARY_RESREQ "rusage[mem=1]" +#define COMPOUND_RESREQ "1*{rusage[mem=1]} + 2*{rusage[mem=1]}" + +#define NORMAL_PRIMARY_JOBID ((uint64_t)630201ULL) +#define NORMAL_COMPOUND_JOBID ((uint64_t)630202ULL) +#define SIM_PRIMARY_JOBID ((uint64_t)630211ULL) +#define SIM_COMPOUND_JOBID ((uint64_t)630212ULL) + +int __wrap_xcecli_start(xcecli_ctx_t c) +{ + (void)c; + return 0; +} + +int __wrap_xcecli_join(xcecli_ctx_t c, int role, hostattrs_t* attrs) +{ + (void)c; + (void)role; + (void)attrs; + return 0; +} + +void setUp(void) { return; } +void tearDown(void) { return; } + +/* ---------------------------------------------------------------------- */ +/* Shared helpers */ +/* ---------------------------------------------------------------------- */ + +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); +} + +/* Anchor a core's virtual clock so that it reads `epoch_ns` right now and + * advances at `speed_num` (1: normal mode's real-time pace; 0: the + * simulator's frozen clock between recorded timestamps). Same atomics + * xce_core_set_wallclock_anchor()/sim.c's sim_set_virtual_clock() store. */ +static void anchor_core_clock(xce_sched_core_t* core, uint64_t epoch_ns, + uint64_t speed_num) +{ + struct timespec ts; + uint64_t mono_ns; + + clock_gettime(CLOCK_MONOTONIC, &ts); + mono_ns = ((uint64_t)ts.tv_sec * NSEC_PER_SEC) + (uint64_t)ts.tv_nsec; + + atomic_store_explicit(&(core->virtual_clock_base), epoch_ns, + memory_order_release); + atomic_store_explicit(&(core->real_clock_base), mono_ns, + memory_order_release); + atomic_store_explicit(&(core->speed_num), speed_num, memory_order_release); + atomic_store_explicit(&(core->speed_den), 1, memory_order_release); +} + +/* Assert every per-host and shared allocation record of `alloc` carries + * `want_from`; returns the number of records checked. */ +static int assert_every_alloc_from(const xce_job_alloc_t* alloc, + uint64_t want_from, const char* what) +{ + char msg[256]; + int checked = 0; + int i, j; + + TEST_ASSERT_NOT_NULL(alloc); + + for (i = 0; i < alloc->nhosts; i++) { + const xce_host_alloc_t* ha = alloc->allocs[i]; + + TEST_ASSERT_NOT_NULL(ha); + for (j = 0; j < ha->nallocs; j++) { + const xce_alloc_t* a = ha->allocs[j]; + + if (a == NULL) { + continue; + } + snprintf(msg, sizeof(msg), + "%s: allocs[%d](host=%s).allocs[%d](resname=%s).from " + "must be the session start in whole seconds", + what, i, ha->host != NULL ? ha->host : "?", j, + a->resname != NULL ? a->resname : "?"); + TEST_ASSERT_EQUAL_UINT64_MESSAGE(want_from, a->from, msg); + checked++; + } + } + + for (i = 0; i < alloc->nres_shared; i++) { + const xce_alloc_t* a = alloc->allocs_shared[i]; + + if (a == NULL) { + continue; + } + snprintf(msg, sizeof(msg), + "%s: allocs_shared[%d](resname=%s).from must be the " + "session start in whole seconds", + what, i, a->resname != NULL ? a->resname : "?"); + TEST_ASSERT_EQUAL_UINT64_MESSAGE(want_from, a->from, msg); + checked++; + } + + return checked; +} + +static double alloc_total_slots(const xce_job_alloc_t* alloc) +{ + double total = 0.0; + int i, j; + + for (i = 0; i < alloc->nhosts; i++) { + for (j = 0; j < alloc->allocs[i]->nallocs; j++) { + const xce_alloc_t* a = alloc->allocs[i]->allocs[j]; + + if ((a->resname != NULL) && (strcmp(a->resname, "slots") == 0)) { + total += a->value; + } + } + } + return total; +} + +/* ---------------------------------------------------------------------- */ +/* Normal-mode fixture (real sched_init(), real wire ingest) */ +/* ---------------------------------------------------------------------- */ + +static xce_worker_t* make_wire_worker(const char* name, float slots) +{ + xce_worker_t* w = (xce_worker_t*)calloc(1, sizeof(*w)); + xce_val_t* val = (xce_val_t*)calloc(1, sizeof(*val)); + xce_name_val_t* nv = (xce_name_val_t*)calloc(1, sizeof(*nv)); + xce_name_val_t** nv_ptrs = (xce_name_val_t**)calloc(1, sizeof(*nv_ptrs)); + + TEST_ASSERT_NOT_NULL(w); + TEST_ASSERT_NOT_NULL(val); + TEST_ASSERT_NOT_NULL(nv); + TEST_ASSERT_NOT_NULL(nv_ptrs); + + val->which = Value_u_fvalue; + val->u.fvalue = slots; + + nv->name = strdup("slots"); + nv->val = val; + nv_ptrs[0] = nv; + + w->name = strdup(name); + w->status = 0; + w->nresvals_to_sched = 1; + w->resvals_to_sched = nv_ptrs; + + return w; +} + +static xce_job_t* make_push_job(uint64_t jobid, const char* resreq) +{ + xce_job_t* j = (xce_job_t*)calloc(1, sizeof(*j)); + + TEST_ASSERT_NOT_NULL(j); + + j->jobid = jobid; + j->state = JOB_STAT_PEND; + j->sub_time = 1000; + j->queue_time = 1000; + + j->submit = (job_submit_data_t*)calloc(1, sizeof(*(j->submit))); + TEST_ASSERT_NOT_NULL(j->submit); + + j->submit->queue = (char*)QUEUE_NAME; + j->submit->user = (char*)"from_stamp_user"; + j->submit->project_name = (char*)"from_stamp_proj"; + j->submit->resreq = (char*)resreq; + j->submit->mintasks = 1; + j->submit->maxtasks = 1; + + return j; +} + +static xce_scheduler_t* start_normal_sched(void) +{ + static xce_client_config_t config; + static xce_cluster_config_t cluster_cfg; + xce_scheduler_t* sched + = (xce_scheduler_t*)calloc(1, sizeof(*sched)); + xce_core_event_t ev; + xce_worker_t* worker_ptrs[2]; + xce_queue_t* q; + int rc; + + TEST_ASSERT_NOT_NULL(sched); + + memset(&config, 0, sizeof(config)); + config.nmasters = 1; + config.masters[0] = "127.0.0.1"; + config.master_ports[0] = 19996; + config.from = "sched_alloc_from_stamp_test"; + config.key = "keepsecret$"; + + memset(&cluster_cfg, 0, sizeof(cluster_cfg)); + cluster_cfg.candidates_count = 1; + strncpy(cluster_cfg.candidates[0].hostname, "127.0.0.1", + XCE_MAXHOSTNAME_LEN - 1); + cluster_cfg.candidates[0].port = 19996; + strncpy(cluster_cfg.identity, "sched_alloc_from_stamp_test", + sizeof(cluster_cfg.identity) - 1); + + rc = sched_init(sched, &config, &cluster_cfg, scheduler_req_handler); + TEST_ASSERT_TRUE_MESSAGE(rc >= 0, "fixture: sched_init() must succeed"); + + TEST_ASSERT_NOT_NULL( + register_plain_resource(sched->core.resmgr, "slots")); + xce_core_set_cache_clear_cb(&(sched->core), xce_alloc_cache_clear_cb); + + q = (xce_queue_t*)calloc(1, sizeof(*q)); + TEST_ASSERT_NOT_NULL(q); + q->queue = strdup(QUEUE_NAME); + q->status = QUEUE_STAT_OPEN; + q->priority = 1; + + memset(&ev, 0, sizeof(ev)); + ev.type = XCE_CORE_EVENT_QUEUE_UPDATE; + ev.u.queue_update.queue = q; + TEST_ASSERT_TRUE(xce_core_push_events(&(sched->core), &ev, 1) >= 0); + + worker_ptrs[0] = make_wire_worker(WORKER_A_NAME, WORKER_SLOTS); + worker_ptrs[1] = make_wire_worker(WORKER_B_NAME, WORKER_SLOTS); + memset(&ev, 0, sizeof(ev)); + ev.type = XCE_CORE_EVENT_WORKERS_PUSH; + ev.u.workers_push.nworkers = 2; + ev.u.workers_push.workers = worker_ptrs; + TEST_ASSERT_TRUE(xce_core_push_events(&(sched->core), &ev, 1) >= 0); + + return sched; +} + +static void push_job(xce_scheduler_t* sched, uint64_t jobid, + const char* resreq) +{ + xce_core_event_t ev; + xce_job_t* job_ptrs[1]; + + job_ptrs[0] = make_push_job(jobid, resreq); + memset(&ev, 0, sizeof(ev)); + ev.type = XCE_CORE_EVENT_JOBS_PUSH; + ev.u.jobs_push.njobs = 1; + ev.u.jobs_push.jobs = job_ptrs; + TEST_ASSERT_TRUE(xce_core_push_events(&(sched->core), &ev, 1) >= 0); + TEST_ASSERT_NOT_NULL(sched_get_job(&(sched->core), jobid)); +} + +/* Feed jobid straight into the live allocator thread (the picker stays + * parked: xce_core_signal_go() is never called) and wait, bounded and + * non-blocking, for its decision -- same technique as + * test_parjob_e2e_multislot.c's dispatch_and_wait(). */ +static xce_decision_t* dispatch_and_wait(xce_scheduler_t* sched, + uint64_t jobid) +{ + xce_decision_t* decision = NULL; + int rc; + int waited_ms; + + rc = cbuf_tryput(&(sched->core.jbuf), &jobid); + TEST_ASSERT_TRUE_MESSAGE(rc >= 0, "fixture: jbuf must have room"); + + for (waited_ms = 0; waited_ms < 2000; waited_ms += 5) { + struct timespec ts = { 0, 5 * 1000 * 1000 }; + + if (cbuf_tryget(&(sched->core.decisions), &decision) >= 0) { + return decision; + } + nanosleep(&ts, NULL); + } + + return NULL; +} + +/* + * Req 4.1, 4.4 (normal mode): one session dispatches a plain PRIMARY job + * and a two-primary COMPOUND job; every per-host and shared allocation + * record of both carries session_start / 10^9. + */ +static void test_normal_mode_every_session_alloc_from_is_session_start(void) +{ + xce_scheduler_t* sched = start_normal_sched(); + xce_decision_t* dprimary; + xce_decision_t* dcompound; + uint64_t want_from; + int n; + + push_job(sched, NORMAL_PRIMARY_JOBID, PRIMARY_RESREQ); + push_job(sched, NORMAL_COMPOUND_JOBID, COMPOUND_RESREQ); + + /* Normal mode's clock runs at real-time pace (speed 1/1) from an + * epoch anchor, exactly as xce_core_set_wallclock_anchor() sets it -- + * anchored here at a known past instant rather than "now", so a + * time(NULL) stamp cannot coincide with the session start. */ + anchor_core_clock(&(sched->core), NORMAL_SESSION_EPOCH_NS, 1); + TEST_ASSERT_EQUAL_INT(0, xce_core_begin_session(&(sched->core))); + + TEST_ASSERT_TRUE_MESSAGE( + sched->core.session_start >= NORMAL_SESSION_EPOCH_NS, + "fixture: the session must start at the anchored epoch time"); + want_from = sched->core.session_start / NSEC_PER_SEC; + TEST_ASSERT_NOT_EQUAL_MESSAGE( + (uint64_t)time(NULL), want_from, + "fixture: the session start must not be the current second"); + + dprimary = dispatch_and_wait(sched, NORMAL_PRIMARY_JOBID); + TEST_ASSERT_NOT_NULL(dprimary); + TEST_ASSERT_EQUAL_INT(XCE_DECISION_DISPATCH, dprimary->op); + n = assert_every_alloc_from(dprimary->u.alloc, want_from, + "normal-mode PRIMARY"); + TEST_ASSERT_TRUE_MESSAGE(n > 0, "PRIMARY must carry allocations"); + + dcompound = dispatch_and_wait(sched, NORMAL_COMPOUND_JOBID); + TEST_ASSERT_NOT_NULL(dcompound); + TEST_ASSERT_EQUAL_INT(XCE_DECISION_DISPATCH, dcompound->op); + TEST_ASSERT_EQUAL_FLOAT_MESSAGE( + 3.0f, (float)alloc_total_slots(dcompound->u.alloc), + "fixture: the merged COMPOUND must cover both primaries (1 + 2)"); + n = assert_every_alloc_from(dcompound->u.alloc, want_from, + "normal-mode COMPOUND"); + TEST_ASSERT_TRUE_MESSAGE(n > 0, "COMPOUND must carry allocations"); +} + +/* ---------------------------------------------------------------------- */ +/* Simulation-mode fixture (bare sim core, sim_alloc.c wrapper allocators) */ +/* ---------------------------------------------------------------------- */ + +static void add_sim_worker(xce_sched_core_t* core, const char* name, + double slots) +{ + xce_sched_worker_t* worker + = (xce_sched_worker_t*)calloc(1, sizeof(*worker)); + xce_resource_t* r = NULL; + xce_val_t* val = NULL; + int rc; + + TEST_ASSERT_NOT_NULL(worker); + + worker->name = strdup(name); + worker->status = 0; + worker->idx = -1; + zarray_init(&(worker->resvals), 128, + (zarray_item_killer_t)kill_sched_resval); + zarray_init(&(worker->resInUse), 128, + (zarray_item_killer_t)kill_sched_resval); + + rc = key_find(core->resmgr, (void*)"slots", (void**)&r); + TEST_ASSERT_TRUE(rc >= 0); + TEST_ASSERT_NOT_NULL(r); + + val = (xce_val_t*)calloc(1, sizeof(*val)); + TEST_ASSERT_NOT_NULL(val); + val->which = Value_u_fvalue; + val->u.fvalue = slots; + zarray_put(&(worker->resvals), r->idx, val); + + rc = sched_add_worker(core, worker); + TEST_ASSERT_TRUE(rc >= 0); +} + +static xce_sched_job_t* make_sim_job(xce_scheduler_t* job_wrapper, + uint64_t jobid, const char* resreq) +{ + xce_job_t* j = (xce_job_t*)calloc(1, sizeof(*j)); + job_submit_data_t* submit = (job_submit_data_t*)calloc(1, sizeof(*submit)); + xce_sched_job_t* job; + + TEST_ASSERT_NOT_NULL(j); + TEST_ASSERT_NOT_NULL(submit); + + submit->queue = (char*)QUEUE_NAME; + submit->user = (char*)"from_stamp_user"; + submit->project_name = (char*)"from_stamp_proj"; + submit->resreq = (char*)resreq; + submit->mintasks = 1; + submit->maxtasks = 1; + + j->jobid = jobid; + j->state = JOB_STAT_PEND; + j->submit = submit; + + job = make_sched_job(job_wrapper, j); + TEST_ASSERT_NOT_NULL(job); + TEST_ASSERT_NOT_NULL(job->schedobj); + + return job; +} + +/* + * Req 4.1, 4.4 (simulation mode): the allocators' plugin->sched is + * sim_alloc.c's session-less wrapper (its core's session_start is 0); the + * session core (sim.core) starts at a recorded, epoch-scale beginRound. + * Every allocation must carry that recorded start in whole seconds. + */ +static void test_sim_mode_every_session_alloc_from_is_session_start(void) +{ + xce_sched_sim_driver_t sim; + xce_scheduler_t job_wrapper; /* make_sched_job() setup only */ + xce_sched_queue_t* queue; + xce_sched_job_t* jprimary; + xce_sched_job_t* jcompound; + xce_sched_plugin_t* seqp; + xce_sched_plugin_t* parp; + uint64_t want_from; + int rc; + int n; + + memset(&sim, 0, sizeof(sim)); + rc = xce_sched_core_init(&(sim.core)); + TEST_ASSERT_EQUAL_INT(0, rc); + + TEST_ASSERT_NOT_NULL(register_plain_resource(sim.core.resmgr, "slots")); + + sim_bootstrap_seqjob_allocator(&sim); + sim_bootstrap_parjob_allocator(&sim); + seqp = (xce_sched_plugin_t*)sim.core.seqjob_allocator; + parp = (xce_sched_plugin_t*)sim.core.parjob_allocator; + TEST_ASSERT_NOT_NULL(seqp); + TEST_ASSERT_NOT_NULL(parp); + TEST_ASSERT_NOT_NULL(seqp->sched); + TEST_ASSERT_NOT_NULL(parp->sched); + TEST_ASSERT_TRUE_MESSAGE( + &(((xce_scheduler_t*)seqp->sched)->core) + != &(sim.core), + "fixture: the allocators' scheduler handle must be the simulator's " + "wrapper, not the session core"); + + queue = (xce_sched_queue_t*)calloc(1, sizeof(*queue)); + TEST_ASSERT_NOT_NULL(queue); + queue->name = makeString(QUEUE_NAME); + TEST_ASSERT_TRUE(sched_add_queue(&(sim.core), queue) >= 0); + + add_sim_worker(&(sim.core), WORKER_A_NAME, WORKER_SLOTS); + add_sim_worker(&(sim.core), WORKER_B_NAME, WORKER_SLOTS); + + memset(&job_wrapper, 0, sizeof(job_wrapper)); + TEST_ASSERT_EQUAL_INT(0, xce_sched_core_init(&(job_wrapper.core))); + job_wrapper.core.resmgr = sim.core.resmgr; + job_wrapper.core.workers = sim.core.workers; + job_wrapper.core.queues = sim.core.queues; + + jprimary = make_sim_job(&job_wrapper, SIM_PRIMARY_JOBID, PRIMARY_RESREQ); + TEST_ASSERT_EQUAL_INT(XCE_RESREQ_PRIMARY, jprimary->schedobj->tag); + jprimary->schedobj->u.primary->resreq->minTasks = 1; + jprimary->schedobj->u.primary->resreq->maxTasks = 1; + TEST_ASSERT_TRUE(sched_add_job(&(sim.core), jprimary) >= 0); + + jcompound + = make_sim_job(&job_wrapper, SIM_COMPOUND_JOBID, COMPOUND_RESREQ); + TEST_ASSERT_EQUAL_INT(XCE_RESREQ_COMPOUND, jcompound->schedobj->tag); + TEST_ASSERT_TRUE(sched_add_job(&(sim.core), jcompound) >= 0); + + /* The simulator's clock is frozen at the recorded beginRound for the + * session (speed 0), so the session core's session_start is exactly + * the recorded begin time. */ + anchor_core_clock(&(sim.core), SIM_RECORDED_BEGIN_NS, 0); + TEST_ASSERT_EQUAL_INT(0, xce_core_begin_session(&(sim.core))); + TEST_ASSERT_EQUAL_UINT64(SIM_RECORDED_BEGIN_NS, sim.core.session_start); + TEST_ASSERT_EQUAL_UINT64_MESSAGE( + 0, + ((xce_scheduler_t*)seqp->sched) + ->core.session_start, + "fixture: the wrapper's core never begins a session"); + want_from = SIM_RECORDED_BEGIN_NS / NSEC_PER_SEC; + + rc = sched_make_allocation(&(sim.core), jprimary->schedobj); + TEST_ASSERT_TRUE_MESSAGE(rc > 0, "sim PRIMARY must be allocated"); + n = assert_every_alloc_from(jprimary->schedobj->alloc, want_from, + "sim-mode PRIMARY"); + TEST_ASSERT_TRUE_MESSAGE(n > 0, "PRIMARY must carry allocations"); + + rc = sched_make_allocation(&(sim.core), jcompound->schedobj); + TEST_ASSERT_TRUE_MESSAGE(rc > 0, "sim COMPOUND must be allocated"); + TEST_ASSERT_EQUAL_FLOAT_MESSAGE( + 3.0f, (float)alloc_total_slots(jcompound->schedobj->alloc), + "fixture: the merged COMPOUND must cover both primaries (1 + 2)"); + n = assert_every_alloc_from(jcompound->schedobj->alloc, want_from, + "sim-mode COMPOUND"); + TEST_ASSERT_TRUE_MESSAGE(n > 0, "COMPOUND must carry allocations"); +} + +int main(void) +{ + UNITY_BEGIN(); + RUN_TEST(test_normal_mode_every_session_alloc_from_is_session_start); + RUN_TEST(test_sim_mode_every_session_alloc_from_is_session_start); + return UNITY_END(); +}