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

test(scheduler-allocation): end-to-end multi-host parallel job proof

Browse source 
Drive a real multi-slot parallel job through the actual production
dispatch path -- real sched_init(), real queue/worker/job ingest via
xce_core_push_events(), a real scheduling session (exercising task 5.2's
cache-clear wiring), the real sched_allocator() thread, cache lookup,
parjob_allocator's real my_alloc(), host_iter, and commit -- and confirm
the resulting job allocation genuinely spans 2 hosts with correct
per-host slot counts (Req 4.1, 4.2, 4.3).

Discovered and logged bug-1770: parse_resreq_primary() unconditionally
hardcodes primary->minTasks/maxTasks to 1 regardless of what a job
actually requests, and nothing in the real ingest pipeline ever copies
the job's own submitted task count (correctly threaded from the
apiserver's num_processors/maxnum_processors all the way to
xce_sched_job_t.mintasks/maxtasks) onto the nested xce_primary_resreq_t
fields the allocator dispatch actually reads. This predates the feature
by ~7 months and is out of this spec's scope to fix, but it means
parjob_allocator has zero live production path until a dedicated
follow-up lands. The test compensates with one clearly-labeled patch of
the real, already-ingested resreq's task-count fields immediately after
real parsing; everything else in the pipeline is unmodified production
code.

Also logged bug-1771 (xce_sched_fcfs's fcfs_del_job() dereferences
job->queue->name with no NULL guard) discovered while building this
task's fixture; worked around by registering the job's queue before
push, mirroring the apiserver's own pre-submission queue validation.

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-05 01:56:54 +00:00
parent e8c2087d5e
commit c009f108b7
No known key found for this signature in database
GPG key ID: 5BCDAAE0123E9FE2
4 changed files with 694 additions and 1 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -1896,3 +1896,35 @@ test('scheduler cache dispatch test',
test('scheduler cache dispatch compound test',
sched_cache_dispatch_compound_test, suite: 'scheduler',
env: ['ASAN_OPTIONS=detect_leaks=0'])
# scheduler-allocation spec, task 5.3 (Req 4.1, 4.2, 4.3): the capstone
# end-to-end integration test for this whole feature -- drives a real
# maxTasks>1 parallel job through the full production dispatch path
# (sched_init()'s real, live sched_allocator() thread -> the real, `static`
# sched_make_allocation()/sched_make_primary_alloc_() choke point -> the
# real parjob_allocator.c my_alloc() -> the real host_iter.c host iterator
# -> a real multi-host xce_job_alloc_t commit), asserting the resulting
# job->allocs/decision->u.alloc spans exactly 2 hosts with the correct
# per-host slot counts. Same real-scheduler_srcs link technique as
# sched_snapshot_load_allocator_regression_test/sched_cache_dispatch_test
# above (see this test file's own header comment for the full rationale).
scheduler_parjob_e2e_multislot_test_srcs = [
'test_parjob_e2e_multislot.c']
foreach f : scheduler_srcs
if f != 'src/main.c'
scheduler_parjob_e2e_multislot_test_srcs += ['../' + f]
endif
endforeach
sched_parjob_e2e_multislot_test = executable(
'sched_parjob_e2e_multislot_test',
scheduler_parjob_e2e_multislot_test_srcs,
include_directories:[incdir],
dependencies: [scheduler_deps, unity_dep],
link_args: ['-Wl,--wrap=xcecli_start',
'-Wl,--wrap=xcecli_join'],
install: false)
test('scheduler parjob e2e multislot test',
sched_parjob_e2e_multislot_test, suite: 'scheduler',
env: ['ASAN_OPTIONS=detect_leaks=0'])

615
src/scheduler/testsuite/test_parjob_e2e_multislot.c Normal file
View file

@ -0,0 +1,615 @@
/*
* test_parjob_e2e_multislot.c - scheduler-allocation spec, task 5.3 (Req
* 4.1, 4.2, 4.3): the capstone END-TO-END integration proof for this whole
* feature. Every other test file in this spec (test_parjob_combinatorial.c,
* test_host_iter.c, test_alloc_cache.c, test_seqjob_rewrap_equivalence.c,
* test_scheduler_cache_dispatch*.c) is white-box: it either #includes the
* unit under test directly, or hand-builds an `xce_sched_ctx_t`/worker set
* and calls a plugin's `my_alloc()`/`prepare_allocation_hook` directly. This
* file does neither. It links the REAL, whole `scheduler_srcs` object list
* (meson.build's `scheduler_deps`) and drives a real job from submission to
* a committed allocation through the SAME functions production jobs use:
*
* sched_init() real production entry point (scheduler.c) --
* starts the REAL sched_job_picker()/
* sched_allocator() threads (not test doubles)
* xce_core_push_events() real ingest (core_events.c) -- WORKERS_PUSH
* (-> core_upsert_worker()/make_sched_worker())
* and JOBS_PUSH (-> core_build_job()/
* core_build_primary_schedobj(), task 1.2's
* real resreq_fp population)
* xce_core_begin_session() real per-session reset (core_session.c,
* task 5.2's cache-clear wiring)
* sched_allocator() real allocator thread (scheduler.c) calling
* the real, `static` sched_make_allocation()
* -> sched_make_primary_allocation() ->
* sched_make_primary_alloc_() [task 5.1's
* cache-wrapped choke point] ->
* sched_make_parjob_allocation() -> the REAL
* parjob_allocator.c my_alloc() -> the REAL
* host_iter.c host iterator -> the REAL
* resource.c sched_try_alloc_on_host_() ->
* a REAL multi-host xce_job_alloc_t commit
* (task 4.3's merge)
*
* ── Link technique ──────────────────────────────────────────────────────
* Identical, established precedent to test_snapshot_load_allocator_
* regression.c / test_scheduler_cache_dispatch.c (both already reviewed and
* accepted for this exact "reach the real, `static` sched_make_allocation()
* dispatch chain without white-box #include-ing all of scheduler.c" problem
* -- see either file's own header comment for why whole-file #include was
* ruled out for scheduler.c specifically): link the REAL, whole
* `scheduler_srcs` list (meson.build) and call sched_init()'s real, public
* entry point, with `-Wl,--wrap=xcecli_start`/`xcecli_join` (trivial
* always-succeed doubles) so sched_init() succeeds with no live controller
* and the real picker/allocator threads actually start.
*
* ── Bypassing the picker thread, not the allocator ──────────────────────
* Like every real-scheduler_srcs test in this directory, this file never
* calls xce_core_signal_go()/sched_go() -- the real sched_job_picker()
* thread (started for real by sched_init() above) stays parked on
* cbuf_get(&core.wbuf) and never touches jbuf. The job id is fed directly
* into sched.core.jbuf (cbuf_tryput()), the exact technique test_snapshot_
* load_allocator_regression.c/test_scheduler_cache_dispatch.c already
* establish for reaching the real, live sched_allocator() thread without
* needing a real controller connection or a real FCFS queue-priority
* dance. This is a deliberate scope choice: task 5.3's Requirements (4.1,
* 4.2, 4.3) are about the ALLOCATOR dispatch chain (cache -> parjob ->
* host_iter -> commit), not job ELECTION policy (scheduler-core's own,
* separately-tested concern) -- so "the same dispatch path production jobs
* use" is satisfied at the allocator choke point sched_make_primary_alloc_()
* onward, exactly where task 5.1's own accepted integration test
* (test_scheduler_cache_dispatch.c) already drew this same line. What THIS
* file adds beyond that precedent, per its own task brief: workers are
* registered via the REAL WORKERS_PUSH ingest event (core_upsert_worker()/
* make_sched_worker()), not sched_add_worker() poked directly, and the job
* is a genuine maxTasks>1 parallel job forcing a genuine multi-host split
* (task 5.1's own test only ever used maxTasks==1, since it was proving
* cache semantics, not parjob's combinatorial allocation).
* xce_core_begin_session() is still called before feeding jbuf -- not
* because the picker needs it (it doesn't; jbuf is fed directly), but
* because it is the literal call driver.c's xce_sched_driver_run_normal_
* mode_round() (main.c's own per-round production entry point) makes
* immediately before xce_core_signal_go(), and it exercises task 5.2's own
* cache-clear wiring (`_Depends: 5.1, 5.2, 4.3` on this task) as part of the
* same real call sequence production makes every round.
*
* ── bug-1770 (discovered by this file, NOT fixed here) ──────────────────
* Confirmed by direct code read (resreq.c's parse_resreq_primary(),
* core_events.c's core_build_job()/core_build_primary_schedobj(),
* schedobj.c's make_sched_job()) and by a genuine RED-phase run of this
* very file: the REAL job-ingest path can NEVER produce a primary resreq
* with maxTasks > 1. parse_resreq_primary() unconditionally hardcodes
* `primary->minTasks = 1; primary->maxTasks = 1;` regardless of the
* resreq string's own content, and neither core_build_job() nor
* make_sched_job() ever copies the job's own submitted task count
* (job_submit_data_t.mintasks/maxtasks, correctly propagated onto the
* TOP-LEVEL xce_sched_job_t.mintasks/maxtasks by both functions) onto the
* NESTED xce_primary_resreq_t.minTasks/maxTasks that sched_make_primary_
* alloc_()'s maxTasks>1 dispatch check (scheduler.c:290) and parjob_
* allocator.c's my_alloc() actually read. Net effect: parjob_allocator is
* UNREACHABLE from any real production job submission today, regardless of
* what task count a real job requests -- every job's primary resreq that
* parses successfully has maxTasks pinned to 1 by parse_resreq_primary(),
* so sched_make_primary_alloc_() always routes it to seqjob_allocator
* instead (scheduler.c:290-294). This is a genuine, real, pre-existing gap
* this end-to-end test is what surfaced it -- every prior unit test in this
* feature (test_parjob_combinatorial.c etc.) hand-constructs its
* xce_primary_resreq_t fixture directly, setting maxTasks by plain struct
* assignment, so none of them exercise the real ingest path closely enough
* to notice maxTasks never actually arrives that way.
*
* Deliberately NOT fixed as part of this task: wiring job_submit_data_t's
* task count into xce_primary_resreq_t requires a real design decision
* (parse_resreq_primary()'s signature would need the job's task count
* threaded in from BOTH of its two call sites, plus a decision about how a
* COMPOUND resreq's several nested primaries should each receive a single
* job-level task count) well outside this spec's Boundary Context (the
* iterator/allocator rework, not resreq-parsing/job-submission plumbing) --
* exactly the same class of pre-existing, out-of-scope gap tasks 5.1/5.2
* already found and deferred in this same feature (bug-1766, bug-1768,
* bug-1769). Logged as bug-1770 (.wolf/buglog.json) for dedicated
* follow-up.
*
* This file's fixture compensates with ONE minimal, explicitly-labeled
* patch of the real, already-ingested xce_primary_resreq_t's task-count
* fields (see PATCH_MAXTASKS_FOR_BUG_1770 below) immediately after the real
* JOBS_PUSH ingest completes and before the round is driven -- everything
* else (cache lookup, parjob dispatch, plan iteration, host claiming,
* multi-host commit, decision production) runs through the unmodified real
* dispatch path. Removing that patch reproduces the RED-phase failure this
* file's own task-5.3 implementation captured (see the implementer's status
* report for the exact command/output): the job would route to seqjob_
* allocator instead of parjob_allocator, and decision->u.alloc->nhosts
* would be 1, not 2.
*
* ── Fixture: forcing a genuine multi-host split ─────────────────────────
* Two real workers, each with exactly 2.0 free "slots" (WORKER_SLOTS),
* registered via a real WORKERS_PUSH event. One job with maxTasks == 4
* (JOB_MAXTASKS). parjob_prepare_allocation() (parjob_allocator.c) always
* stages a "slots" want of exactly the current bucket's own slot count
* (ctx->nslots) per host try, independent of the resreq's own rusage
* content -- so a bare "rusage[mem=1]" resreq (RESREQ_STR, identical
* convention to test_scheduler_cache_dispatch.c/test_seqjob_allocator.c:
* "mem" is never registered in this fixture's minimal resmgr, so it is
* silently skipped by calc_wanted_host_resvals()'s own `key_find()`-miss
* continue -- see resreq.c -- leaving ONLY the auto-injected "slots" want)
* is enough to exercise the real per-bucket resource check. With maxTasks=4
* and 2 hosts of 2.0 slots each:
* - plan k=1 (1 host x 4 slots): NEITHER worker has 4.0 free slots ->
* bucket 0 fails immediately -> plan 1 rolled back (nothing claimed).
* - plan k=2 (2 hosts x [2,2], the balanced split of 4 into 2 buckets):
* bucket 0 (need 2) claims whichever host host_iter offers first
* (2.0 >= 2.0 satisfies); bucket 1 (need 2) claims the other, excluded
* hosts' free capacity. Plan succeeds -> committed as a single
* xce_job_alloc_t with nhosts == 2, one xce_host_alloc_t per worker,
* each holding a "slots" xce_alloc_t of value 2.0.
* No single-host plan can ever satisfy this job; only the genuine
* 2-host combinatorial split can. This is the same "no host alone
* suffices, but two together do" shape task 4.2's own worked example
* establishes, now proven through the REAL production dispatch path
* instead of a direct parjob_try_plan_() call.
*/
#include <errno.h>
#include <stdint.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 "preason.h"
#include "resobj.h"
#include "resreq.h"
#include "sched/alloc_cache.h"
#include "scheduler.h"
#include "xce/cluster.h"
#include "xstring.h"
extern int scheduler_req_handler(xce_request_t*, void*);
#define WORKER_A_NAME "e2e-parjob-host-a"
#define WORKER_B_NAME "e2e-parjob-host-b"
#define WORKER_SLOTS 2.0f
#define RESREQ_STR ((char*)"rusage[mem=1]")
#define JOB_ID ((uint64_t)540101ULL)
#define JOB_MAXTASKS 4
/* ════════════════════════════════════════════════════════════════════════
* __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;
}
void setUp(void) { return; }
void tearDown(void) { return; }
/* ---------------------------------------------------------------------- */
/* Fixture helpers */
/* ---------------------------------------------------------------------- */
/* register_slots_resource -- identical helper to test_seqjob_allocator.c's/
* test_scheduler_cache_dispatch.c's own: seqjob_prepare_allocation()'s AND
* parjob_prepare_allocation()'s real key_find(sched->core.resmgr, "slots",
* ...) call needs "slots" to be a genuinely registered resource. Registered
* directly via makeResource(), the same startup-time-config convention
* every real-scheduler_srcs test in this directory follows (resources are
* static cluster config, loaded once before any job/worker event flows in
* -- see sched_init_config()/sched_init_resources(), never reached by these
* tests since none of them push a real CONFIG_PUSH event either). */
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);
}
/* make_wire_worker -- a real xce_worker_t wire payload carrying one
* "slots" resval, ready for a real XCE_CORE_EVENT_WORKERS_PUSH ingest
* (core_upsert_worker()/make_sched_worker(), core_events.c/schedobj.c).
* Same field shapes test_core_events.c's own
* test_workers_push_twice_updates_existing_worker_in_place() establishes.
* Never torn down -- same "fixtures here are never torn down" convention
* every other real-scheduler_srcs test in this directory follows. */
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;
}
/* make_wire_queue -- a real xce_queue_t wire payload, ready for a real
* XCE_CORE_EVENT_QUEUE_UPDATE ingest (core_upsert_queue(), core_events.c).
* Required: sched_get_queue() (core.c) is get-ONLY (returns NULL for an
* unregistered name, never auto-creates), and this feature's legacy
* "xce_sched_fcfs" plugin (still registered alongside builtin_fcfs by
* sched_register_plugins(), scheduler.c) unconditionally dereferences
* job->queue->name in its own commit_hook (fcfs.c's fcfs_del_job(), called
* from plugin_commit_decision()'s real per-plugin fan-out for EVERY
* DISPATCH decision, regardless of which plugin is the active picker) --
* see bug-1771 (.wolf/buglog.json) for the live-confirmed SEGV this file's
* own RED-phase run hit before this queue push was added, and why a real
* queue registration (not a NULL-safety patch to fcfs.c) is the correct
* fix for THIS fixture specifically. */
static xce_queue_t* make_wire_queue(const char* name, int32_t priority)
{
xce_queue_t* q = (xce_queue_t*)calloc(1, sizeof(*q));
TEST_ASSERT_NOT_NULL(q);
q->queue = strdup(name);
q->status = QUEUE_STAT_OPEN;
q->priority = priority;
return q;
}
/* 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()). Same field
* shapes test_scheduler_cache_dispatch.c's/test_sim_e2e_original_
* motivation.c's own make_push_job()/make_job() establish. Never torn
* down, same convention. */
static xce_job_t* make_push_job(uint64_t jobid, char* resreq_str, char* user,
char* queue, char* project, uint32_t mintasks,
uint32_t maxtasks)
{
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 = mintasks;
j->submit->maxtasks = maxtasks;
return j;
}
/* start_sched -- real sched_init() (registers the real builtin_fcfs/
* seqjob_allocator/parjob_allocator plugins, sched_register_plugins(), and
* starts the real sched_job_picker()/sched_allocator() threads), plus the
* real per-session cache-clear hook registration main.c itself performs
* (xce_core_set_cache_clear_cb(), Req 7.2) -- so xce_core_begin_session()
* below exercises the identical real call sequence production's
* xce_sched_driver_run_normal_mode_round() makes, not a partial stand-in. */
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] = 19997;
config.from = "sched_parjob_e2e_multislot_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 = 19997;
strncpy(cluster_cfg.identity, "sched_parjob_e2e_multislot_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);
xce_core_set_cache_clear_cb(&(sched->core), xce_alloc_cache_clear_cb);
return sched;
}
/* dispatch_and_wait -- bypass the dormant picker thread (xce_core_signal_
* go()/sched_go() is never called, 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_cache_dispatch.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;
}
/* ---------------------------------------------------------------------- */
/* Test */
/* ---------------------------------------------------------------------- */
/*
* Req 4.1/4.2/4.3: a real maxTasks=4 parallel job, submitted through the
* real ingest path against two real 2.0-slot workers (also ingested via
* the real WORKERS_PUSH path), reaches a committed, 2-host allocation
* through the unmodified real dispatch chain -- confirmed via the exact
* same job->allocs/decision->u.alloc production storage a real dispatch
* populates, not a test-only shortcut.
*/
static void
test_multihost_parallel_job_allocation_spans_two_hosts_via_real_dispatch(void)
{
xce_scheduler_t* sched = start_sched();
xce_queue_t* queue;
xce_core_event_t queue_event;
xce_worker_t* worker_ptrs[2];
xce_core_event_t worker_event;
xce_job_t* job;
xce_job_t* job_ptrs[1];
xce_core_event_t job_event;
xce_sched_job_t* sjob;
xce_primary_resreq_t* primary;
xce_decision_t* decision;
int rc;
int saw_a = 0;
int saw_b = 0;
double total_slots = 0.0;
/* ── Real queue ingest (QUEUE_UPDATE) ────────────────────────────── */
queue = make_wire_queue("e2e_queue", 1);
memset(&queue_event, 0, sizeof(queue_event));
queue_event.type = XCE_CORE_EVENT_QUEUE_UPDATE;
queue_event.u.queue_update.queue = queue;
rc = xce_core_push_events(&(sched->core), &queue_event, 1);
TEST_ASSERT_TRUE_MESSAGE(rc >= 0,
"fixture: real QUEUE_UPDATE ingest must succeed");
TEST_ASSERT_NOT_NULL_MESSAGE(
sched_get_queue(&(sched->core), "e2e_queue"),
"queue must be registered in the REAL scheduler queue set via the "
"real ingest path before any job targets it -- see bug-1771");
/* ── Real worker ingest (WORKERS_PUSH) ───────────────────────────── */
worker_ptrs[0] = make_wire_worker(WORKER_A_NAME, WORKER_SLOTS);
worker_ptrs[1] = make_wire_worker(WORKER_B_NAME, WORKER_SLOTS);
memset(&worker_event, 0, sizeof(worker_event));
worker_event.type = XCE_CORE_EVENT_WORKERS_PUSH;
worker_event.u.workers_push.nworkers = 2;
worker_event.u.workers_push.workers = worker_ptrs;
rc = xce_core_push_events(&(sched->core), &worker_event, 1);
TEST_ASSERT_TRUE_MESSAGE(rc >= 0,
"fixture: real WORKERS_PUSH ingest must succeed");
TEST_ASSERT_NOT_NULL_MESSAGE(
sched_get_worker(&(sched->core), WORKER_A_NAME),
"worker A must be registered in the REAL scheduler worker set via "
"the real ingest path");
TEST_ASSERT_NOT_NULL_MESSAGE(
sched_get_worker(&(sched->core), WORKER_B_NAME),
"worker B must be registered in the REAL scheduler worker set via "
"the real ingest path");
/* ── Real job ingest (JOBS_PUSH) ─────────────────────────────────── */
job = make_push_job(JOB_ID, RESREQ_STR, "e2e_user", "e2e_queue",
"e2e_proj", 1, JOB_MAXTASKS);
job_ptrs[0] = job;
memset(&job_event, 0, sizeof(job_event));
job_event.type = XCE_CORE_EVENT_JOBS_PUSH;
job_event.u.jobs_push.njobs = 1;
job_event.u.jobs_push.jobs = job_ptrs;
rc = xce_core_push_events(&(sched->core), &job_event, 1);
TEST_ASSERT_TRUE_MESSAGE(rc >= 0, "fixture: real JOBS_PUSH ingest must "
"succeed");
sjob = sched_get_job(&(sched->core), JOB_ID);
TEST_ASSERT_NOT_NULL(sjob);
TEST_ASSERT_NOT_NULL_MESSAGE(
sjob->schedobj, "job->schedobj must be built by the real ingest "
"path (core_build_job())");
TEST_ASSERT_EQUAL_INT(XCE_RESREQ_PRIMARY, sjob->schedobj->tag);
/* job->maxtasks (the TOP-LEVEL xce_sched_job_t field) correctly
* carries the real submitted task count through ingest... */
TEST_ASSERT_EQUAL_UINT32(JOB_MAXTASKS, sjob->maxtasks);
/* ...but the NESTED xce_primary_resreq_t.maxTasks the allocator
* dispatch actually reads does not -- bug-1770 (see file header). This
* assertion is the direct, standing proof of that gap: it will keep
* passing (documenting the bug) until bug-1770 is fixed for real. */
primary = sjob->schedobj->u.primary->resreq;
TEST_ASSERT_NOT_NULL(primary);
TEST_ASSERT_EQUAL_INT_MESSAGE(
1, primary->maxTasks,
"bug-1770: the real ingest path currently hardcodes primary->"
"maxTasks=1 regardless of the job's own submitted maxtasks -- if "
"this assertion ever fails because maxTasks now correctly arrives "
"as JOB_MAXTASKS, bug-1770 has been fixed for real and the "
"PATCH_MAXTASKS_FOR_BUG_1770 workaround below should be deleted");
/* PATCH_MAXTASKS_FOR_BUG_1770: compensate for the confirmed,
* out-of-boundary gap above by patching the REAL, already-ingested
* primary resreq's task-count fields directly (see file header for
* the full rationale for not fixing this in production code as part
* of this task). Everything downstream of this point -- cache lookup,
* parjob dispatch, plan iteration, host claiming, multi-host commit,
* decision production -- runs through the unmodified real dispatch
* path. resreq_fp is kept consistent too (same fields
* core_build_primary_schedobj() itself derives from primary->min/
* maxTasks at ingest time), though this single-dispatch test never
* exercises a second cache lookup that would depend on it. */
primary->minTasks = 1;
primary->maxTasks = JOB_MAXTASKS;
sjob->schedobj->u.primary->resreq_fp.min_tasks = 1;
sjob->schedobj->u.primary->resreq_fp.max_tasks = JOB_MAXTASKS;
/* ── Real per-session begin (task 5.2's cache-clear wiring) ──────── */
rc = xce_core_begin_session(&(sched->core));
TEST_ASSERT_TRUE_MESSAGE(rc >= 0, "fixture: xce_core_begin_session() "
"must succeed");
/* ── Drive the job through the REAL, live sched_allocator() thread ─ */
decision = dispatch_and_wait(sched, JOB_ID);
TEST_ASSERT_NOT_NULL_MESSAGE(
decision, "the real allocator must produce a decision for the "
"multi-host parallel job within 2s");
TEST_ASSERT_EQUAL_INT_MESSAGE(
XCE_DECISION_DISPATCH, decision->op,
"the job must be DISPATCHed, not PENDed -- the two 2.0-slot "
"workers together satisfy the balanced [2,2] 2-host plan");
/* ── The resulting REAL job allocation record ────────────────────── */
TEST_ASSERT_NOT_NULL(decision->u.alloc);
TEST_ASSERT_TRUE_MESSAGE(
sjob->allocs == decision->u.alloc,
"decision->u.alloc must be the SAME object as job->allocs -- "
"sched_allocator()'s real MOVE_PTR(job->allocs, "
"job->schedobj->alloc), not a copy -- proving this is the real "
"production-stored allocation record, not a test-only artifact");
TEST_ASSERT_EQUAL_UINT64(JOB_ID, decision->u.alloc->jobid);
TEST_ASSERT_EQUAL_INT_MESSAGE(
2, decision->u.alloc->nhosts,
"the committed allocation must span exactly 2 hosts -- the "
"balanced [2,2] plan for maxTasks=4 -- since neither single "
"2.0-slot worker alone can satisfy a 1-host x 4-slot plan");
/* Per-host slot counts: each claimed host must be one of the two real
* workers, its "slots" allocation must be exactly 2.0 (the balanced
* bucket size, not a fixed/wrong value), and the two hosts' totals
* must sum to the job's full maxTasks. */
{
int i;
for (i = 0; i < decision->u.alloc->nhosts; i++) {
xce_host_alloc_t* h = decision->u.alloc->allocs[i];
int j;
int found_slots = 0;
TEST_ASSERT_NOT_NULL(h);
TEST_ASSERT_NOT_NULL(h->host);
if (strcmp(h->host, WORKER_A_NAME) == 0) {
saw_a = 1;
} else if (strcmp(h->host, WORKER_B_NAME) == 0) {
saw_b = 1;
} else {
TEST_FAIL_MESSAGE(
"claimed host is neither of the two real ingested "
"workers");
}
for (j = 0; j < h->nallocs; j++) {
xce_alloc_t* a = h->allocs[j];
TEST_ASSERT_NOT_NULL(a);
if ((a->resname != NULL) && (strcmp(a->resname, "slots") == 0)) {
found_slots = 1;
TEST_ASSERT_EQUAL_FLOAT_MESSAGE(
2.0f, (float)a->value,
"each claimed host's own slots allocation must be "
"exactly this bucket's balanced share (2 of 4), "
"not a fixed 1-slot value or the job's full "
"maxTasks");
total_slots += a->value;
}
}
TEST_ASSERT_TRUE_MESSAGE(
found_slots, "each claimed host must carry a real 'slots' "
"allocation entry");
}
}
TEST_ASSERT_TRUE_MESSAGE(saw_a, "worker A must be one of the two "
"claimed hosts");
TEST_ASSERT_TRUE_MESSAGE(saw_b, "worker B must be one of the two "
"claimed hosts");
TEST_ASSERT_EQUAL_FLOAT_MESSAGE(
(float)JOB_MAXTASKS, (float)total_slots,
"the two hosts' per-host slot counts must sum to the job's full "
"maxTasks");
}
int main(void)
{
UNITY_BEGIN();
RUN_TEST(
test_multihost_parallel_job_allocation_spans_two_hosts_via_real_dispatch);
return UNITY_END();
}