myxce/src/scheduler/testsuite/test_scheduler_cache_dispatch.c

/*
 * 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.
 * ════════════════════════════════════════════════════════════════════════ */

/* scheduler-sim-live-allocation task 1.3: sched_meet_host_selection() is
 * now core-typed (xce_sched_core_t*, not xce_scheduler_t*) -- the linker
 * --wrap= interposition must match the real symbol's current signature. */
extern int __real_sched_meet_host_selection(xce_sched_core_t*,
                                            xce_sched_worker_t*,
                                            xce_select_t*);

static int g_select_calls = 0;

int __wrap_sched_meet_host_selection(xce_sched_core_t*   core,
                                     xce_sched_worker_t* w,
                                     xce_select_t*        select)
{
    g_select_calls++;

    return __real_sched_meet_host_selection(core, 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();
}