lei.chen/myxce
1
0
Fork 1
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions
myxce/src/scheduler/testsuite/test_sched_alloc_from_stamp.c
Rongsong Shen c1960457f8
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) <noreply@anthropic.com>
2026-09-28 10:14:35 +00:00

569 lines
20 KiB
C
Raw Blame History

/*
* 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 <errno.h>
#include <stdint.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <time.h>
#include <unity/unity.h>
#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();
}
Reference in a new issue View git blame Copy permalink