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>
This commit is contained in:
parent
2d7ba66019
commit
c1960457f8
GPG key ID:
80533536B0EB651C
4 changed files with 654 additions and 2 deletions
|
|
@ -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.
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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'])
|
||||
|
|
|
|||
569
src/scheduler/testsuite/test_sched_alloc_from_stamp.c
Normal file
569
src/scheduler/testsuite/test_sched_alloc_from_stamp.c
Normal file
|
|
@ -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 <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();
|
||||
}
|
||||
Loading…
Add table
Add a link
Reference in a new issue