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

feat(scheduler-allocation): wire unsatisfiable-resreq cache into dispatch

Browse source 
Wrap sched_make_primary_alloc_() -- the single choke point all three
primary-resreq shapes (standalone, compound-nested, alter-nested via
compound delegation) already funnel through -- with a cache lookup
before allocation and a cache insert on exhaustion. On a hit, pend the
job immediately via the same make_pendreason()/add_job_preason() path
every exhaustion case already uses, without invoking either allocator
(Req 7.3, 7.4, 7.5).

A nested primary has no real per-primary schedobj, so
sched_primary_cache_obj_() synthesizes one on the stack per dispatch for
cache-key purposes; verified memory-safe since alloc_cache.c's
build_key_() copies the fingerprint/identity by value synchronously
before the stack frame unwinds.

Log bug-1768 (compound/alter-nested primaries never get real slot/rusage
availability checks -- seqjob/parjob's prepare_allocation hooks no-op
whenever ctx->schedobj->tag != XCE_RESREQ_PRIMARY) and bug-1769
(sched_make_alter_allocation() type-confuses xce_compound_schedobj_t*
as xce_sched_obj_t*, a ~112-byte heap over-read, currently untested).
Both are pre-existing and out of this feature's scope; discovered and
confirmed while building this task's dispatch tests.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01XYdtrEWWZfKRRGED7raEYg
This commit is contained in:
Rongsong Shen 2026-09-04 23:27:56 +00:00
parent 9d23060b76
commit 5c022d0b4f
No known key found for this signature in database
GPG key ID: 5BCDAAE0123E9FE2
6 changed files with 1015 additions and 5 deletions
Download patch file Download diff file Expand all files Collapse all files

138
src/scheduler/src/scheduler.c
View file

@ -167,16 +167,146 @@ static int sched_make_parjob_allocation(xce_scheduler_t* sched,
return plugin->job_alloc_fn(plugin, ctx);
}
/*
* sched_primary_cache_obj_ - Req 7.1/7.3/7.4: resolve the xce_sched_obj_t*
* cache_lookup()/cache_insert() (alloc_cache.c, task 2.2) require for the
* PRIMARY resreq sched_make_primary_alloc_() is about to dispatch --
* whether that primary is the job's own top-level request, or one nested
* inside a COMPOUND or an ALTER's compound.
*
* Standalone top-level primary: ctx->schedobj IS the real, ingest-time
* PRIMARY-tagged xce_sched_obj_t (core_build_primary_schedobj(),
* core_events.c, task 1.2) -- its obj->u.primary->resreq_fp was already
* computed once at ingest, so it is reused as-is (the entire point of Req
* 7.1's "compute once, reuse on every lookup").
*
* Nested primary (compound- or alter-nested): ctx->schedobj is the
* enclosing COMPOUND-tagged xce_sched_obj_t (sched_make_compound_
* allocation() calls sched_ctx_init() once with it, then sched_ctx_
* set_primary() once per nested xce_primary_resreq_t in its primaries
* list) -- no xce_primary_schedobj_t/resreq_fp was ever built for the
* individual nested resreq; only the top-level PRIMARY case gets one.
* cache_lookup()/cache_insert()'s own precondition (obj->tag ==
* XCE_RESREQ_PRIMARY, obj->u.primary != NULL) cannot be satisfied by
* ctx->schedobj itself here, so a stack-local synthetic wrapper is built
* instead: resreq_fp is computed fresh, field-for-field identical to
* core_build_primary_schedobj()'s own formula, from ctx->primary directly
* -- minTasks/maxTasks/select/rusage_hash are populated uniformly for
* EVERY primary in a resreq's parse tree by parse_resreq_primary()
* (resreq.c), not only top-level ones, so this is always valid, never
* stale. Identity (user/queue/project) is borrowed from ctx->schedobj:
* those three fields live outside the tag union on xce_sched_obj_t, so
* they are valid regardless of whether ctx->schedobj's own tag is
* PRIMARY, COMPOUND, or ALTER.
*
* *scratch_primary/*scratch_obj are caller-owned stack storage, written
* here and valid only for the caller's own stack frame -- cache_lookup()/
* cache_insert() only ever dereference the returned pointer during the
* call itself (cache_key_assign_() copies the key by value on insert), so
* no lifetime issue results from returning a pointer into the caller's
* own locals.
*/
static xce_sched_obj_t*
sched_primary_cache_obj_(xce_sched_ctx_t* ctx,
xce_primary_schedobj_t* scratch_primary,
xce_sched_obj_t* scratch_obj)
{
xce_sched_obj_t* schedobj = ctx->schedobj;
if ((schedobj != NULL) && (schedobj->tag == XCE_RESREQ_PRIMARY)
&& (schedobj->u.primary != NULL)
&& (schedobj->u.primary->resreq == ctx->primary)) {
return schedobj;
}
memset(scratch_primary, 0, sizeof(*scratch_primary));
scratch_primary->resreq = ctx->primary;
scratch_primary->owner = schedobj;
scratch_primary->resreq_fp.min_tasks = ctx->primary->minTasks;
scratch_primary->resreq_fp.max_tasks = ctx->primary->maxTasks;
scratch_primary->resreq_fp.select_node_id
= (ctx->primary->select != NULL && ctx->primary->select->root != NULL)
? ctx->primary->select->root->node_id
: 0;
scratch_primary->resreq_fp.rusage_hash = ctx->primary->rusage_hash;
memset(scratch_obj, 0, sizeof(*scratch_obj));
scratch_obj->tag = XCE_RESREQ_PRIMARY;
if (schedobj != NULL) {
scratch_obj->user = schedobj->user;
scratch_obj->queue = schedobj->queue;
scratch_obj->project = schedobj->project;
}
scratch_obj->u.primary = scratch_primary;
return scratch_obj;
}
/*
* sched_pend_from_cached_reason_ - Req 7.4/7.5: on a cache hit, attach the
* STORED reason category to ctx->reason using the identical mechanism
* every allocator-exhaustion path already uses to create/populate a pend
* reason (make_pendreason() + add_job_preason(), never a direct calloc()/
* field write -- see preason.h/preason.c and e.g. resource.c's
* sched_try_alloc_on_host_() or seqjob_allocator.c's/parjob_allocator.c's
* my_alloc(), all of which follow this exact "if (ctx->reason == NULL) {
* ctx->reason = make_pendreason(0); }" pattern). add_job_preason(reasons,
* code, NULL, NULL) is that same function's own "set main reason" branch
* (preason.c) -- the identical call shape design.md's cache_insert()
* precondition ties reason_category to (xce_job_preason_t.main_reason).
*/
static void sched_pend_from_cached_reason_(xce_sched_ctx_t* ctx, int category)
{
if (ctx->reason == NULL) {
ctx->reason = make_pendreason(0);
}
if (ctx->reason == NULL) {
return;
}
add_job_preason(ctx->reason->reasons, category, NULL, NULL);
}
static int sched_make_primary_alloc_(xce_scheduler_t* sched,
xce_sched_ctx_t* ctx)
{
xce_primary_resreq_t* primary = ctx->primary;
xce_primary_resreq_t* primary = ctx->primary;
xce_primary_schedobj_t scratch_primary;
xce_sched_obj_t scratch_obj;
xce_sched_obj_t* cache_obj;
int reason_category = 0;
int rc;
if (primary->maxTasks == 1) {
return sched_make_seqjob_allocation(sched, ctx);
cache_obj = sched_primary_cache_obj_(ctx, &scratch_primary, &scratch_obj);
/* Req 7.4: a cache hit pends the job immediately, without invoking
* either allocator for this attempt. */
if (cache_lookup(sched, cache_obj, &reason_category) == 1) {
sched_pend_from_cached_reason_(ctx, reason_category);
return 0;
}
return sched_make_parjob_allocation(sched, ctx);
if (primary->maxTasks == 1) {
rc = sched_make_seqjob_allocation(sched, ctx);
} else {
rc = sched_make_parjob_allocation(sched, ctx);
}
/* 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
* final pend-reason category, before the caller pends the job. A
* negative rc is an infrastructure error, not an unsatisfiable
* determination, so it is never cached; a positive rc is success. */
if (rc == 0) {
int category
= (ctx->reason != NULL) ? ctx->reason->reasons->main_reason : 0;
cache_insert(sched, cache_obj, category);
}
return rc;
}
static int sched_make_primary_allocation(xce_scheduler_t* sched,

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

@ -1439,6 +1439,58 @@ sched_snapshot_load_allocator_regression_test = executable(
'-Wl,--wrap=xcecli_join'],
install: false)
# scheduler-allocation spec, task 5.1 (Req 7.3, 7.4, 7.5): wiring the
# unsatisfiable-resreq cache around sched_make_primary_alloc_() (`static`
# to scheduler.c). Same real-scheduler_srcs link technique as
# sched_snapshot_load_allocator_regression_test above (see this test
# file's own header comment for the full rationale). In addition to the
# usual xcecli_start/xcecli_join doubles, this file also wraps
# sched_meet_host_selection (resource.c, exported) to count real calls --
# the decisive "did the real allocator run" signal a cache HIT and a
# genuine exhausted attempt cannot otherwise be told apart by (both pend,
# and Req 7.5 requires them to report the SAME reason category).
scheduler_cache_dispatch_test_srcs = ['test_scheduler_cache_dispatch.c']
foreach f : scheduler_srcs
if f != 'src/main.c'
scheduler_cache_dispatch_test_srcs += ['../' + f]
endif
endforeach
sched_cache_dispatch_test = executable(
'sched_cache_dispatch_test',
scheduler_cache_dispatch_test_srcs,
include_directories:[incdir],
dependencies: [scheduler_deps, unity_dep],
link_args: ['-Wl,--wrap=xcecli_start',
'-Wl,--wrap=xcecli_join',
'-Wl,--wrap=sched_meet_host_selection'],
install: false)
# Companion to sched_cache_dispatch_test above: the primary-nested-inside-
# a-COMPOUND half of the same task 5.1 proof, kept in its own binary/
# process -- calling sched_init() twice in one process is not safe here
# (plugins/fcfs.c's legacy job_elect_fn path carries process-global state;
# a real run with both tests in one binary crashed with a genuine SEGV in
# fcfs_del_job() the first time this was tried -- see this test file's own
# header comment).
scheduler_cache_dispatch_compound_test_srcs = [
'test_scheduler_cache_dispatch_compound.c']
foreach f : scheduler_srcs
if f != 'src/main.c'
scheduler_cache_dispatch_compound_test_srcs += ['../' + f]
endif
endforeach
sched_cache_dispatch_compound_test = executable(
'sched_cache_dispatch_compound_test',
scheduler_cache_dispatch_compound_test_srcs,
include_directories:[incdir],
dependencies: [scheduler_deps, unity_dep],
link_args: ['-Wl,--wrap=xcecli_start',
'-Wl,--wrap=xcecli_join',
'-Wl,--wrap=sched_meet_host_selection'],
install: false)
# White-box regression test for bug-1685 (double-free/heap-corruption
# SIGABRT in limit_update_config() on a duplicate limit-name config
# update). It #includes ../src/plugins/limit.c directly (same white-box
@ -1780,3 +1832,9 @@ test('scheduler seqjob rewrap equivalence test',
test('scheduler parjob combinatorial test',
sched_parjob_combinatorial_test, suite: 'scheduler',
env: ['ASAN_OPTIONS=detect_leaks=0'])
test('scheduler cache dispatch test',
sched_cache_dispatch_test, suite: 'scheduler',
env: ['ASAN_OPTIONS=detect_leaks=0'])
test('scheduler cache dispatch compound test',
sched_cache_dispatch_compound_test, suite: 'scheduler',
env: ['ASAN_OPTIONS=detect_leaks=0'])

401
src/scheduler/testsuite/test_scheduler_cache_dispatch.c Normal file
View file

@ -0,0 +1,401 @@
/*
* test_scheduler_cache_dispatch.c - scheduler-allocation spec, task 5.1
* (Req 7.3, 7.4, 7.5): wiring the unsatisfiable-resreq cache (alloc_cache.c,
* task 2.2) around the allocator dispatch choke point, sched_make_primary_
* alloc_() (scheduler.c, `static`) -- for a standalone top-level PRIMARY
* job. See test_scheduler_cache_dispatch_compound.c for the identical proof
* against a primary nested inside a COMPOUND (a separate binary: sched_init()
* is not safe to call twice in one process -- plugins/fcfs.c's legacy
* job_elect_fn path carries process-global state that a second sched_init()
* corrupts, confirmed by a real SEGV in fcfs_del_job() when both tests were
* first written into one RUN_TEST binary).
*
* ── Link technique ──────────────────────────────────────────────────────
* sched_make_primary_alloc_() (and the sched_make_primary_allocation()/
* sched_make_compound_allocation() callers that reach it) are `static` to
* scheduler.c. This directory's own established precedent (test_scheduler_
* connect.c, test_scheduler_backpressure.c, test_scheduler_dispatch_
* commit.c, test_snapshot_load_allocator_regression.c -- see each file's
* own header comment) already rules out white-box #include-ing scheduler.c
* itself: its dependency surface (controller-join, capnp decision
* publishing, HA candidate handling, the full plugins/ tree) is too large
* for a single-TU combination. This file follows the same lower-risk
* precedent instead: link the REAL, whole scheduler_srcs list (meson.build)
* and call sched_init()'s real, public entry point, with `-Wl,--wrap=` for
* xcecli_start()/xcecli_join() (trivial always-succeed doubles, identical
* rationale to every file above) so sched_init() succeeds with no live
* controller and the real picker/allocator threads actually start.
*
* ── Proving "without invoking either allocator" ─────────────────────────
* A cache HIT and a genuine exhausted allocation attempt both end up
* pending the job -- on a cache hit, Req 7.5 even requires the SAME pend
* reason category, so the pend reason's own contents cannot distinguish
* the two paths (in fact, main_reason is 0 in both cases today: no
* existing allocator code path ever sets it to anything else -- see
* scheduler.c's sched_pend_from_cached_reason_() comment). This file
* instead wraps sched_meet_host_selection() (resource.c, exported): it is
* the one leaf BOTH host_iter.c's pre-filter and sched_try_alloc_on_host_()
* 's post-check call, for every real candidate-host examination either
* allocator's plan/host-iterator loop makes. A cache hit skips
* sched_make_seqjob_allocation()/sched_make_parjob_allocation() entirely,
* so it can never reach this function -- with exactly ONE worker
* registered (below), its call count sampled per job dispatch is direct,
* decisive proof of whether the real allocator ran for that job.
*
* ── Forcing a deterministic, real exhaustion ─────────────────────────────
* The one registered worker carries 0.0 free "slots" (register_slots_
* resource()/add_worker_with_slots(), same real makeResource()/
* sched_add_worker() helpers test_seqjob_allocator.c already established).
* seqjob_prepare_allocation()/parjob_prepare_allocation() always stage a
* "slots" want of exactly ctx->nslots per task (test_seqjob_allocator.c's
* own header comment: "'slots' is the one resource seqjob_prepare_
* allocation() always stages") -- independent of the resreq's own rusage
* content -- so a 0-slot worker is unconditionally unsatisfiable for any
* job, regardless of resreq shape.
*
* The resreq string is a bare "rusage[mem=1]" clause, deliberately WITHOUT
* a select[...] clause: parse_select() (resreq.c) resolves every select
* field name against the live resmgr (xce_select.peg's resolve_resname()),
* so "select[type==any]" would fail to parse ("unknown resource 'type'")
* against this fixture's minimal resmgr (only "slots" is registered) --
* confirmed by a real RED-phase run (job1's primary came back with
* maxTasks=0/minTasks=0, all-zero, because xce_resreq.peg's top-level
* primary_resreq action does not check parse_resreq_primary()'s return
* value before wrapping whatever partially-populated struct it left
* behind). parse_rusage() (unlike parse_select()) takes no resmgr
* parameter at all -- resource names in a rusage clause are resolved later,
* at allocation time, not at parse time -- so a bare rusage clause sidesteps
* this fixture's minimal resmgr entirely while still landing on a bare
* XCE_RESREQ_PRIMARY tag through the FULL parse_resreq() grammar (confirmed
* against lib/testsuite/test_core_events.c's own core_build_primary_
* schedobj() unit test, which parses a rusage-only string the same way).
*
* Two jobs pushed through the REAL job-ingest path (xce_core_push_events(),
* XCE_CORE_EVENT_JOBS_PUSH -- core_build_job()/core_build_primary_
* schedobj(), task 1.2's real resreq_fp population), identical resreq
* string and identity (user/queue/project -- xstring-interned by
* makeString(), so identical content really does mean identical pointers,
* the property xce_cache_key_t's own comparator relies on). Job 1 (cache
* miss) must run the real allocator (sched_meet_host_selection() calls > 0)
* and end up pended; job 2 (cache hit) must be pended WITHOUT any new
* sched_meet_host_selection() call, with the identical pend-reason
* category.
*/
#include <errno.h>
#include <stdint.h>
#include <stdlib.h>
#include <string.h>
#include <time.h>
#include <unity/unity.h>
#include "client.h"
#include "core.h"
#include "decision.h"
#include "preason.h"
#include "resobj.h"
#include "resreq.h"
#include "scheduler.h"
#include "xce/cluster.h"
#include "xstring.h"
extern int scheduler_req_handler(xce_request_t*, void*);
#define WORKER_NAME "cache-dispatch-host"
#define RESREQ_STR ((char*)"rusage[mem=1]")
#define JOB1_ID ((uint64_t)530001ULL)
#define JOB2_ID ((uint64_t)530002ULL)
/* ════════════════════════════════════════════════════════════════════════
* __wrap_xcecli_start / __wrap_xcecli_join -- trivial always-succeed
* doubles, identical rationale/shape to every other real-scheduler_srcs
* test in this directory: this file's real sched_init() call must succeed
* with no live controller so the real picker/allocator threads actually
* start.
* ════════════════════════════════════════════════════════════════════════ */
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;
}
/* ════════════════════════════════════════════════════════════════════════
* __wrap_sched_meet_host_selection -- counts real calls, then delegates to
* the real implementation (__real_sched_meet_host_selection, provided by
* the linker's -Wl,--wrap= interposition over resource.c's real, exported
* sched_meet_host_selection()). See file header for why this call count is
* the decisive "did the real allocator run" signal.
* ════════════════════════════════════════════════════════════════════════ */
extern int __real_sched_meet_host_selection(xce_scheduler_t*,
xce_sched_worker_t*,
xce_select_t*);
static int g_select_calls = 0;
int __wrap_sched_meet_host_selection(xce_scheduler_t* sched,
xce_sched_worker_t* w,
xce_select_t* select)
{
g_select_calls++;
return __real_sched_meet_host_selection(sched, w, select);
}
void setUp(void) { return; }
void tearDown(void) { return; }
/* ---------------------------------------------------------------------- */
/* Fixture helpers */
/* ---------------------------------------------------------------------- */
/* register_slots_resource / add_worker_with_slots -- identical helpers to
* test_seqjob_allocator.c's own (same file header comment there explains
* why: seqjob_prepare_allocation()'s real key_find(sched->core.resmgr,
* "slots", ...) call needs "slots" to be a genuinely registered
* resource). */
static xce_resource_t* register_slots_resource(var resmgr)
{
xce_resource_t res;
memset(&res, 0, sizeof(res));
res.name = strdup("slots");
return makeResource(resmgr, &res);
}
static xce_sched_worker_t* add_worker_with_slots(xce_sched_core_t* core,
const char* name,
int slots_idx,
float slots_total)
{
xce_sched_worker_t* w;
int idx[1];
float val[1];
int rc;
w = (xce_sched_worker_t*)calloc(1, sizeof(*w));
TEST_ASSERT_NOT_NULL(w);
w->name = strdup(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);
idx[0] = slots_idx;
val[0] = slots_total;
put_resvals(&(w->resvals), 1, idx, val);
val[0] = 0.0f;
put_resvals(&(w->resInUse), 1, idx, val);
rc = sched_add_worker(core, w);
TEST_ASSERT_TRUE(rc >= 0);
return w;
}
static xce_scheduler_t* start_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_resource_t* slots_res;
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] = 19999;
config.from = "sched_cache_dispatch_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 = 19999;
strncpy(cluster_cfg.identity, "sched_cache_dispatch_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 (xcecli_start()/"
"xcecli_join() wrapped to succeed with no I/O) so the "
"real picker/allocator threads actually start");
slots_res = register_slots_resource(sched->core.resmgr);
TEST_ASSERT_NOT_NULL(slots_res);
/* Exactly one worker, zero free "slots" -- unconditionally
* unsatisfiable for any job (see file header). */
add_worker_with_slots(&(sched->core), WORKER_NAME, slots_res->idx, 0.0f);
return sched;
}
/* dispatch_and_wait - bypass the dormant picker thread (sched_go() is
* never called here, so sched_job_picker() stays parked on
* cbuf_get(&core.wbuf) and never touches jbuf itself) and feed jobid
* directly into the live allocator thread's input queue, then wait
* (bounded, non-blocking poll -- never a real blocking call) for the
* resulting decision. Same technique test_snapshot_load_allocator_
* regression.c/test_scheduler_backpressure.c already establish. */
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;
}
/* make_push_job - a heap-allocated xce_job_t/job_submit_data_t pair ready
* for a real XCE_CORE_EVENT_JOBS_PUSH ingest (core_build_job()). Never
* torn down -- same "fixtures here are never torn down" convention every
* other white-box/real-scheduler_srcs test in this directory already
* follows. */
static xce_job_t* make_push_job(uint64_t jobid, char* resreq_str, char* user,
char* queue, char* project)
{
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 = queue;
j->submit->user = user;
j->submit->project_name = project;
j->submit->resreq = resreq_str;
j->submit->mintasks = 1;
j->submit->maxtasks = 1;
return j;
}
/* ---------------------------------------------------------------------- */
/* Test */
/* ---------------------------------------------------------------------- */
/*
* Req 7.3/7.4/7.5, standalone top-level PRIMARY: job 2 shares an identical
* fingerprint (same resreq string, parsed via the real ingest path) and
* identity (same user/queue/project) with job 1, already proven
* unsatisfiable. Job 2 must be pended immediately, with the SAME pend
* reason category, WITHOUT the real allocator running (zero new
* sched_meet_host_selection() calls).
*/
static void test_standalone_primary_cache_hit_skips_allocator(void)
{
xce_scheduler_t* sched = start_sched();
xce_job_t* j1;
xce_job_t* j2;
xce_job_t* jobs[2];
xce_core_event_t event;
xce_decision_t* decision1;
xce_decision_t* decision2;
int calls_before;
int calls_job1;
int calls_job2;
int rc;
j1 = make_push_job(JOB1_ID, RESREQ_STR, "alice", "normal", "proj1");
j2 = make_push_job(JOB2_ID, RESREQ_STR, "alice", "normal", "proj1");
jobs[0] = j1;
jobs[1] = j2;
memset(&event, 0, sizeof(event));
event.type = XCE_CORE_EVENT_JOBS_PUSH;
event.u.jobs_push.njobs = 2;
event.u.jobs_push.jobs = jobs;
rc = xce_core_push_events(&(sched->core), &event, 1);
TEST_ASSERT_TRUE_MESSAGE(rc >= 0,
"fixture: JOBS_PUSH ingest must succeed");
TEST_ASSERT_NOT_NULL(sched_get_job(&(sched->core), JOB1_ID)->schedobj);
TEST_ASSERT_EQUAL_INT(
XCE_RESREQ_PRIMARY,
sched_get_job(&(sched->core), JOB1_ID)->schedobj->tag);
TEST_ASSERT_EQUAL_INT(
1, sched_get_job(&(sched->core), JOB1_ID)
->schedobj->u.primary->resreq->maxTasks);
/* Job 1: cache miss -- the real allocator must run. */
calls_before = g_select_calls;
decision1 = dispatch_and_wait(sched, JOB1_ID);
calls_job1 = g_select_calls - calls_before;
TEST_ASSERT_NOT_NULL_MESSAGE(
decision1, "job 1 (cache miss) must produce a decision");
TEST_ASSERT_EQUAL_INT(XCE_DECISION_PEND, decision1->op);
TEST_ASSERT_NOT_NULL(decision1->u.reason);
TEST_ASSERT_TRUE_MESSAGE(
calls_job1 > 0,
"job 1 (cache miss) must genuinely run the allocator's host "
"selection check at least once");
/* Job 2: identical fingerprint + identity -- cache HIT. Must pend
* immediately, with zero additional sched_meet_host_selection()
* calls (neither allocator ever runs), and the same reason
* category job 1's exhaustion stored. */
calls_before = g_select_calls;
decision2 = dispatch_and_wait(sched, JOB2_ID);
calls_job2 = g_select_calls - calls_before;
TEST_ASSERT_NOT_NULL_MESSAGE(
decision2, "job 2 (cache hit) must still produce a PEND decision");
TEST_ASSERT_EQUAL_INT(XCE_DECISION_PEND, decision2->op);
TEST_ASSERT_NOT_NULL(decision2->u.reason);
TEST_ASSERT_EQUAL_INT_MESSAGE(
0, calls_job2,
"job 2 (cache hit) must pend WITHOUT invoking either allocator -- "
"zero new sched_meet_host_selection() calls");
TEST_ASSERT_EQUAL_INT_MESSAGE(
decision1->u.reason->reasons->main_reason,
decision2->u.reason->reasons->main_reason,
"a cache hit must attribute the SAME pend reason category as the "
"original unsatisfiable determination (Req 7.5)");
}
int main(void)
{
UNITY_BEGIN();
RUN_TEST(test_standalone_primary_cache_hit_skips_allocator);
return UNITY_END();
}

374
src/scheduler/testsuite/test_scheduler_cache_dispatch_compound.c Normal file
View file

@ -0,0 +1,374 @@
/*
* test_scheduler_cache_dispatch_compound.c - scheduler-allocation spec,
* task 5.1 (Req 7.3, 7.4, 7.5): the primary-nested-inside-a-COMPOUND half
* of the unsatisfiable-resreq cache dispatch-wiring proof. See
* test_scheduler_cache_dispatch.c for the standalone top-level PRIMARY
* half and the shared rationale (link technique, why sched_meet_host_
* selection() call-count is the "did the real allocator run" signal, why
* the resreq is a bare rusage-only clause). Kept in its OWN binary/process
* rather than a second RUN_TEST alongside that file's test: calling
* sched_init() twice in one process is not safe here -- plugins/fcfs.c's
* legacy job_elect_fn path carries process-global state, and a real run
* with both tests in one binary crashed with a genuine SEGV in
* fcfs_del_job() (called from the SECOND sched_init()'s allocator thread)
* the first time this was tried.
*
* xce_compound_schedobj_t's primaries are bare xce_primary_resreq_t
* entries with no per-primary schedobj/resreq_fp of their own (only
* core_build_primary_schedobj() -- the top-level PRIMARY case -- ever
* populates one), so sched_make_primary_alloc_()'s cache-key resolution
* must synthesize a fingerprint fresh from the nested resreq's own fields
* for this shape to work at all -- exactly the case this file targets.
* Two hand-built COMPOUND jobs (xce_compound_schedobj_t wrapping one real,
* parse_resreq_primary()-parsed nested primary each), added via the same
* sched_add_job() + direct jbuf feed technique test_snapshot_load_
* allocator_regression.c/test_scheduler_backpressure.c already establish
* for bypassing the dormant picker thread. Same assertions as the
* standalone-primary test.
*
* ── Why SELECT-based rejection here, unlike the standalone test's
* rusage/"slots"-starvation approach ────────────────────────────────
* A pre-existing, out-of-boundary gap (unrelated to this task, discovered
* during this test's own RED-phase debugging): seqjob_prepare_allocation()/
* parjob_prepare_allocation() (plugins/seqjob_allocator.c/parjob_
* allocator.c) both read the primary resreq via `ctx->schedobj->u.primary
* ->resreq` and bail out immediately (a silent no-op return) when
* `ctx->schedobj->tag != XCE_RESREQ_PRIMARY` -- i.e. whenever ctx->schedobj
* is the enclosing COMPOUND, exactly this test's own shape. This means
* ctx->alloc_req.host/.shared are NEVER staged for a compound-nested
* primary dispatched through either allocator, so a resource-starved
* worker (the standalone test's own "slots"-only mechanism) is trivially,
* wrongly judged satisfiable for a compound-nested primary regardless of
* real availability -- confirmed live (job A was DISPATCHED, not pended,
* against a worker it should have been unable to satisfy, crashing later
* in fcfs_del_job() on this fixture's deliberately-NULL job->queue).
* SELECT-based rejection sidesteps this cleanly: both host_iter.c's
* pre-filter and sched_try_alloc_on_host_()'s (resource.c) own re-check
* read the select expression off `ctx->primary` directly (never through
* ctx->schedobj), so a select clause the fixture's one worker fails to
* match is enforced correctly for a compound-nested primary too, with no
* dependency on the broken prepare_allocation staging path. Tracked
* separately (not fixed here -- out of this task's `scheduler.c dispatch`
* boundary, and files this task must not modify without genuine
* necessity).
*/
#include <errno.h>
#include <stdint.h>
#include <stdlib.h>
#include <string.h>
#include <time.h>
#include <unity/unity.h>
#include "client.h"
#include "core.h"
#include "decision.h"
#include "preason.h"
#include "resobj.h"
#include "resreq.h"
#include "scheduler.h"
#include "xce/cluster.h"
#include "xstring.h"
extern int scheduler_req_handler(xce_request_t*, void*);
#define WORKER_NAME "cache-dispatch-compound-host"
#define RESREQ_STR ((char*)"select[slots >= 1]")
#define JOBA_ID ((uint64_t)530011ULL)
#define JOBB_ID ((uint64_t)530012ULL)
/* ════════════════════════════════════════════════════════════════════════
* __wrap_xcecli_start / __wrap_xcecli_join -- see test_scheduler_cache_
* dispatch.c's identical doubles for rationale.
* ════════════════════════════════════════════════════════════════════════ */
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;
}
/* ════════════════════════════════════════════════════════════════════════
* __wrap_sched_meet_host_selection -- see test_scheduler_cache_dispatch.c's
* identical wrap for rationale.
* ════════════════════════════════════════════════════════════════════════ */
extern int __real_sched_meet_host_selection(xce_scheduler_t*,
xce_sched_worker_t*,
xce_select_t*);
static int g_select_calls = 0;
int __wrap_sched_meet_host_selection(xce_scheduler_t* sched,
xce_sched_worker_t* w,
xce_select_t* select)
{
g_select_calls++;
return __real_sched_meet_host_selection(sched, w, select);
}
void setUp(void) { return; }
void tearDown(void) { return; }
/* ---------------------------------------------------------------------- */
/* Fixture helpers -- identical to test_scheduler_cache_dispatch.c's own */
/* ---------------------------------------------------------------------- */
static xce_resource_t* register_slots_resource(var resmgr)
{
xce_resource_t res;
memset(&res, 0, sizeof(res));
res.name = strdup("slots");
res.valtype = XCE_VAL_NUMERIC;
return makeResource(resmgr, &res);
}
static xce_sched_worker_t* add_worker_with_slots(xce_sched_core_t* core,
const char* name,
int slots_idx,
float slots_total)
{
xce_sched_worker_t* w;
int idx[1];
float val[1];
int rc;
w = (xce_sched_worker_t*)calloc(1, sizeof(*w));
TEST_ASSERT_NOT_NULL(w);
w->name = strdup(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);
idx[0] = slots_idx;
val[0] = slots_total;
put_resvals(&(w->resvals), 1, idx, val);
val[0] = 0.0f;
put_resvals(&(w->resInUse), 1, idx, val);
rc = sched_add_worker(core, w);
TEST_ASSERT_TRUE(rc >= 0);
return w;
}
static xce_scheduler_t* start_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_resource_t* slots_res;
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] = 19998;
config.from = "sched_cache_dispatch_compound_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 = 19998;
strncpy(cluster_cfg.identity, "sched_cache_dispatch_compound_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 (xcecli_start()/"
"xcecli_join() wrapped to succeed with no I/O) so the "
"real picker/allocator threads actually start");
slots_res = register_slots_resource(sched->core.resmgr);
TEST_ASSERT_NOT_NULL(slots_res);
/* Exactly one worker, zero free "slots" -- unconditionally
* unsatisfiable for any job. */
add_worker_with_slots(&(sched->core), WORKER_NAME, slots_res->idx, 0.0f);
return sched;
}
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;
}
/* build_compound_job - a hand-built xce_sched_job_t whose schedobj is
* XCE_RESREQ_COMPOUND-tagged, wrapping exactly one real, parse_resreq_
* primary()-parsed nested primary. Mirrors core_build_compound_schedobj()
* 's own shape (compound->resreq->primaries, an xlist_t of bare
* xce_primary_resreq_t* entries with no per-primary schedobj of their
* own) without going through the full parse_resreq() grammar for the
* COMPOUND wrapper itself -- this file only needs a well-formed COMPOUND
* shape to drive sched_make_compound_allocation(), not to prove the
* parser produces one from any particular string. */
static xce_sched_job_t*
build_compound_job(uint64_t jobid, xce_primary_resreq_t* primary, char* user,
char* queue, char* project)
{
xce_sched_job_t* job = (xce_sched_job_t*)calloc(1, sizeof(*job));
xce_sched_obj_t* schedobj
= (xce_sched_obj_t*)calloc(1, sizeof(*schedobj));
xce_compound_schedobj_t* compound
= (xce_compound_schedobj_t*)calloc(1, sizeof(*compound));
xce_compound_resreq_t* resreq
= (xce_compound_resreq_t*)calloc(1, sizeof(*resreq));
TEST_ASSERT_NOT_NULL(job);
TEST_ASSERT_NOT_NULL(schedobj);
TEST_ASSERT_NOT_NULL(compound);
TEST_ASSERT_NOT_NULL(resreq);
xlist_init(&(resreq->primaries), 4);
xlist_store(&(resreq->primaries), primary);
compound->owner = schedobj;
compound->resreq = resreq;
compound->reason = NULL;
schedobj->tag = XCE_RESREQ_COMPOUND;
schedobj->u.compound = compound;
schedobj->user = makeString(user);
schedobj->queue = makeString(queue);
schedobj->project = makeString(project);
job->jobid = jobid;
job->state = JOB_STAT_PEND;
job->schedobj = schedobj;
return job;
}
/* ---------------------------------------------------------------------- */
/* Test */
/* ---------------------------------------------------------------------- */
/*
* Req 7.3/7.4/7.5, primary nested inside a COMPOUND: identical wiring
* proof as the standalone-primary test, but for a primary reached via
* sched_make_compound_allocation() rather than the job's own top-level
* tag -- exercising sched_make_primary_alloc_()'s synthetic-cache-key
* path (no per-primary schedobj/resreq_fp exists for a compound-nested
* primary; see file header).
*/
static void test_compound_nested_primary_cache_hit_skips_allocator(void)
{
xce_scheduler_t* sched = start_sched();
xce_primary_resreq_t* primaryA;
xce_primary_resreq_t* primaryB;
xce_sched_job_t* jobA;
xce_sched_job_t* jobB;
xce_decision_t* decisionA;
xce_decision_t* decisionB;
char* errmsg = NULL;
int calls_before;
int calls_jobA;
int calls_jobB;
int rc;
primaryA = (xce_primary_resreq_t*)calloc(1, sizeof(*primaryA));
primaryB = (xce_primary_resreq_t*)calloc(1, sizeof(*primaryB));
TEST_ASSERT_NOT_NULL(primaryA);
TEST_ASSERT_NOT_NULL(primaryB);
rc = parse_resreq_primary(primaryA, RESREQ_STR, sched->core.resmgr,
&errmsg);
TEST_ASSERT_TRUE_MESSAGE(rc >= 0, "fixture: primaryA must parse");
rc = parse_resreq_primary(primaryB, RESREQ_STR, sched->core.resmgr,
&errmsg);
TEST_ASSERT_TRUE_MESSAGE(rc >= 0, "fixture: primaryB must parse");
TEST_ASSERT_EQUAL_INT(1, primaryA->maxTasks);
TEST_ASSERT_EQUAL_INT(1, primaryB->maxTasks);
jobA = build_compound_job(JOBA_ID, primaryA, "bob", "normal", "proj1");
jobB = build_compound_job(JOBB_ID, primaryB, "bob", "normal", "proj1");
rc = sched_add_job(&(sched->core), jobA);
TEST_ASSERT_TRUE(rc >= 0);
rc = sched_add_job(&(sched->core), jobB);
TEST_ASSERT_TRUE(rc >= 0);
/* Job A: cache miss -- the real allocator must run for its one
* nested primary. */
calls_before = g_select_calls;
decisionA = dispatch_and_wait(sched, JOBA_ID);
calls_jobA = g_select_calls - calls_before;
TEST_ASSERT_NOT_NULL_MESSAGE(
decisionA, "job A (cache miss) must produce a decision");
TEST_ASSERT_EQUAL_INT(XCE_DECISION_PEND, decisionA->op);
TEST_ASSERT_NOT_NULL(decisionA->u.reason);
TEST_ASSERT_TRUE_MESSAGE(
calls_jobA > 0,
"job A (cache miss) must genuinely run the allocator's host "
"selection check at least once for its nested primary");
/* Job B: a different job, but its one nested primary shares an
* identical fingerprint (same resreq string) and identity (same
* user/queue/project) with job A's -- cache HIT. */
calls_before = g_select_calls;
decisionB = dispatch_and_wait(sched, JOBB_ID);
calls_jobB = g_select_calls - calls_before;
TEST_ASSERT_NOT_NULL_MESSAGE(
decisionB, "job B (cache hit) must still produce a PEND decision");
TEST_ASSERT_EQUAL_INT(XCE_DECISION_PEND, decisionB->op);
TEST_ASSERT_NOT_NULL(decisionB->u.reason);
TEST_ASSERT_EQUAL_INT_MESSAGE(
0, calls_jobB,
"job B's nested primary (cache hit) must pend WITHOUT invoking "
"either allocator -- zero new sched_meet_host_selection() calls");
TEST_ASSERT_EQUAL_INT_MESSAGE(
decisionA->u.reason->reasons->main_reason,
decisionB->u.reason->reasons->main_reason,
"a cache hit must attribute the SAME pend reason category as the "
"original unsatisfiable determination (Req 7.5), even across a "
"compound-nested primary");
FREEUP(errmsg);
}
int main(void)
{
UNITY_BEGIN();
RUN_TEST(test_compound_nested_primary_cache_hit_skips_allocator);
return UNITY_END();
}