myxce/src/scheduler/testsuite/test_core_session.c

/*
 * test_core_session.c - contract test for the session-start signal and
 * decision callback protocol (scheduler-core spec, task 4.4, Req 5 AC 5 /
 * AC 6).
 *
 * Observable under test:
 *   - xce_core_go() on a fresh core (session_active == 0) signals the
 *     picker thread NON-BLOCKINGLY -- a single item lands on core->wbuf --
 *     and returns 0 without waiting for anything downstream.
 *   - xce_core_go() while a session is already active returns -EINVAL (a
 *     distinct error from success) and is a true no-op: session_active is
 *     unchanged and no second signal is placed on wbuf.
 *   - Decisions placed on core->decisions (standing in for what the
 *     allocator thread would produce in a real session -- the picker/
 *     allocator pipeline itself is already covered by
 *     test_picker_fcfs_session.c / test_picker_core_loop_boundary.c, out
 *     of this task's scope) reach the driver-registered on_decision
 *     callback exactly once each, in allocator-output order, via the same
 *     cbuf_get()/on_decision() drain loop a real driver runs between
 *     xce_core_go() and xce_core_end_session() (design.md's Normal-Mode
 *     Scheduling Round: "the drain loop in the driver thread interleaves
 *     cbuf_get/on_decision one at a time"). The end-of-session NULL marker
 *     terminates that loop and is never itself passed to on_decision
 *     (requirements.md's Driver <-> Core Protocol table marks the NULL
 *     marker "(implicit)", consumed by "Picker -> Allocator -> Driver",
 *     never listed as an on_decision argument).
 *   - xce_core_end_session() clears session_active back to 0, unblocking
 *     the next xce_core_go() call.
 *
 * White-box #include of ../src/core.c, ../src/core_session.c,
 * ../src/core_clock.c, and ../src/decision.c (same convention as
 * test_core_events.c/test_core_jbuf_capacity_pin.c) so this test exercises
 * the REAL xce_sched_core_init()/xce_core_go()/xce_core_end_session()/
 * xce_core_set_decision_cb()/make_*_decision(), not a re-implementation.
 * core_clock.c joined this file's #include list in task 4.5 (scheduler-core
 * spec): xce_core_go()/xce_core_end_session() now call xce_core_now()
 * (Req 5 AC 12 session timing), which core_clock.c is the only TU that
 * defines -- see test_clock.c for that function's own dedicated contract
 * tests; this file does not re-test the clock formula itself.
 * Unlike test_core_events.c, this file does not need core_events.c or a
 * working picker plugin resolution -- xce_core_go()/xce_core_end_session()
 * never call sched_resolve_job_picker() (verified: grep confirms neither
 * core.c nor core_session.c references it) -- so the kill_sched_ family
 * and plugin_unregister stubs below exist purely to satisfy the linker for
 * xce_sched_core_init()'s kvmgr_params_t.datakiller fields (never actually
 * invoked -- fixtures in this file are never torn down, same convention as
 * every other white-box test in this directory).
 */
#include <errno.h>
#include <stdint.h>
#include <stdlib.h>
#include <string.h>

#include <unity/unity.h>

#include "preason.h"

/* Pull in the real xce_sched_core_init()/xce_core_go()/
 * xce_core_end_session()/xce_core_set_decision_cb()/make_*_decision()
 * under test into this TU (white-box #include). */
#include "../src/core.c"
#include "../src/core_clock.c"
#include "../src/core_session.c"
#include "../src/decision.c"

void kill_sched_job(void* arg) { (void)arg; }
void kill_sched_queue(void* arg) { (void)arg; }
void kill_sched_worker(void* arg) { (void)arg; }
void kill_sched_user(void* arg) { (void)arg; }
void kill_sched_project(void* arg) { (void)arg; }
void kill_sched_resval(xce_val_t* val) { (void)val; }
void plugin_unregister(void* arg) { (void)arg; }

/* sched_worker_release_host_ctx (resource.c) is outside this file's
 * white-box TU (only core.c/core_session.c/decision.c are #included) --
 * stubbed the same way the kill_sched_ family is above: satisfies the
 * linker for core_reset_workers()'s call in core_session.c. None of this
 * file's fixtures ever add a worker to core->workers, so this is never
 * actually invoked at runtime either. */
void sched_worker_release_host_ctx(xce_sched_worker_t* w) { (void)w; }

void setUp(void) { return; }
void tearDown(void) { return; }

/* ---------------------------------------------------------------------- */
/* Fixture                                                                 */
/* ---------------------------------------------------------------------- */

static void fixture_init(xce_sched_core_t* core)
{
    int rc = xce_sched_core_init(core);
    TEST_ASSERT_EQUAL_INT(0, rc);
}

/* Test driver context + on_decision callback: records each decision
 * pointer, in call order, so tests can assert both order and identity. */
#define MAX_RECORDED_DECISIONS 8

typedef struct {
    int             calls;
    xce_decision_t* seen[MAX_RECORDED_DECISIONS];
} test_driver_ctx_t;

static int test_on_decision(void* driver_ctx, xce_decision_t* decision)
{
    test_driver_ctx_t* ctx = (test_driver_ctx_t*)driver_ctx;

    TEST_ASSERT_NOT_NULL(decision);

    if (ctx->calls < MAX_RECORDED_DECISIONS) {
        ctx->seen[ctx->calls] = decision;
    }
    ctx->calls++;

    return 0;
}

/* drain_and_dispatch - the driver's own drain loop (design.md's Normal-
 * Mode Scheduling Round), reproduced here exactly as a real driver would
 * write it against the public cbuf_get()/on_decision() primitives: NOT a
 * core-owned function (see core_session.c's file header comment for why
 * this task deliberately does not add one). Returns the number of
 * decisions delivered to on_decision before the NULL end-of-session
 * marker was seen. */
static int drain_and_dispatch(xce_sched_core_t* core)
{
    int delivered = 0;

    for (;;) {
        xce_decision_t* decision = NULL;

        cbuf_get(&(core->decisions), &decision);
        if (decision == NULL) {
            break;
        }

        core->on_decision(core->driver_ctx, decision);
        delivered++;
    }

    return delivered;
}

/* ---------------------------------------------------------------------- */
/* xce_core_go() -- session-start signal (Req 5 AC 5)                     */
/* ---------------------------------------------------------------------- */

static void test_core_go_signals_picker_and_returns_immediately(void)
{
    xce_sched_core_t core;
    int              rc;
    int              go = 0;

    fixture_init(&core);

    rc = xce_core_go(&core);

    TEST_ASSERT_EQUAL_INT(0, rc);
    TEST_ASSERT_EQUAL_INT(
        1, atomic_load_explicit(&(core.session_active), memory_order_acquire));

    /* Non-blocking: the signal already landed on wbuf without anything
     * having to drain it first -- a blocking cbuf_get() here would prove
     * nothing (it would simply wait), so use the non-blocking tryget to
     * prove the item is already present. */
    rc = cbuf_tryget(&(core.wbuf), &go);
    TEST_ASSERT_EQUAL_INT(0, rc);
    TEST_ASSERT_EQUAL_INT(1, go);
}

static void test_core_go_rejects_reentrant_call_as_true_noop(void)
{
    xce_sched_core_t core;
    int              rc;
    int              go = 0;

    fixture_init(&core);

    rc = xce_core_go(&core);
    TEST_ASSERT_EQUAL_INT(0, rc);

    /* Drain the first signal so a hypothetical buggy second cbuf_put()
     * would not simply hang this test on a full capacity-1 buffer -- it
     * would instead land visibly on the tryget below. */
    rc = cbuf_tryget(&(core.wbuf), &go);
    TEST_ASSERT_EQUAL_INT(0, rc);

    rc = xce_core_go(&core);
    TEST_ASSERT_EQUAL_INT(-EINVAL, rc);
    TEST_ASSERT_EQUAL_INT(
        1, atomic_load_explicit(&(core.session_active), memory_order_acquire));

    /* No second signal was sent -- wbuf is still empty. */
    rc = cbuf_tryget(&(core.wbuf), &go);
    TEST_ASSERT_EQUAL_INT(-EAGAIN, rc);
}

static void test_core_go_null_core_returns_einval(void)
{
    int rc = xce_core_go(NULL);

    TEST_ASSERT_EQUAL_INT(-EINVAL, rc);
}

/* ---------------------------------------------------------------------- */
/* Decision callback protocol (Req 5 AC 6)                                 */
/* ---------------------------------------------------------------------- */

static void
test_decisions_reach_callback_in_order_then_end_of_session_marker(void)
{
    xce_sched_core_t  core;
    test_driver_ctx_t driver_ctx;
    xce_job_alloc_t   alloc1;
    xce_pendreason_t* reason2;
    xce_preempt_t     preempt3;
    xce_decision_t*   d1;
    xce_decision_t*   d2;
    xce_decision_t*   d3;
    xce_decision_t*   end_marker = NULL;
    int               rc;
    int               delivered;

    fixture_init(&core);
    memset(&driver_ctx, 0, sizeof(driver_ctx));

    rc = xce_core_set_decision_cb(&core, test_on_decision, &driver_ctx);
    TEST_ASSERT_EQUAL_INT(0, rc);

    rc = xce_core_go(&core);
    TEST_ASSERT_EQUAL_INT(0, rc);

    /* Stand in for the allocator thread's output: one decision of each
     * type, in a deliberately chosen order, pushed directly onto
     * core->decisions -- the same cbuf sched_allocator() (scheduler.c)
     * already writes to today. */
    memset(&alloc1, 0, sizeof(alloc1));
    alloc1.jobid = 1;
    d1           = make_dispatch_decision(&alloc1);
    TEST_ASSERT_NOT_NULL(d1);

    reason2 = make_pendreason(2);
    TEST_ASSERT_NOT_NULL(reason2);
    d2 = make_pend_decision(reason2);
    TEST_ASSERT_NOT_NULL(d2);

    memset(&preempt3, 0, sizeof(preempt3));
    preempt3.jobid    = 3;
    preempt3.nvictims = 0;
    preempt3.victims  = NULL;
    preempt3.reason   = make_pendreason(3);
    d3                = make_preempt_decision(&preempt3);
    TEST_ASSERT_NOT_NULL(d3);

    cbuf_put(&(core.decisions), &d1);
    cbuf_put(&(core.decisions), &d2);
    cbuf_put(&(core.decisions), &d3);
    cbuf_put(&(core.decisions), &end_marker);

    delivered = drain_and_dispatch(&core);

    TEST_ASSERT_EQUAL_INT(3, delivered);
    TEST_ASSERT_EQUAL_INT(3, driver_ctx.calls);

    /* Allocator-output order preserved end to end. */
    TEST_ASSERT_EQUAL_PTR(d1, driver_ctx.seen[0]);
    TEST_ASSERT_EQUAL_INT(XCE_DECISION_DISPATCH, driver_ctx.seen[0]->op);
    TEST_ASSERT_EQUAL_UINT64(1, driver_ctx.seen[0]->u.alloc->jobid);

    TEST_ASSERT_EQUAL_PTR(d2, driver_ctx.seen[1]);
    TEST_ASSERT_EQUAL_INT(XCE_DECISION_PEND, driver_ctx.seen[1]->op);
    TEST_ASSERT_EQUAL_UINT64(2, driver_ctx.seen[1]->u.reason->jobid);

    TEST_ASSERT_EQUAL_PTR(d3, driver_ctx.seen[2]);
    TEST_ASSERT_EQUAL_INT(XCE_DECISION_PREEMPT, driver_ctx.seen[2]->op);
    TEST_ASSERT_EQUAL_UINT64(3, driver_ctx.seen[2]->u.preempt->jobid);

    rc = xce_core_end_session(&core);
    TEST_ASSERT_EQUAL_INT(0, rc);
    TEST_ASSERT_EQUAL_INT(
        0, atomic_load_explicit(&(core.session_active), memory_order_acquire));
}

static void test_core_set_decision_cb_null_core_returns_einval(void)
{
    int rc = xce_core_set_decision_cb(NULL, test_on_decision, NULL);

    TEST_ASSERT_EQUAL_INT(-EINVAL, rc);
}

static void test_core_set_decision_cb_registers_cb_and_driver_ctx(void)
{
    xce_sched_core_t  core;
    test_driver_ctx_t driver_ctx;
    int               rc;

    fixture_init(&core);
    memset(&driver_ctx, 0, sizeof(driver_ctx));

    rc = xce_core_set_decision_cb(&core, test_on_decision, &driver_ctx);

    TEST_ASSERT_EQUAL_INT(0, rc);
    TEST_ASSERT_EQUAL_PTR((void*)test_on_decision, (void*)core.on_decision);
    TEST_ASSERT_EQUAL_PTR(&driver_ctx, core.driver_ctx);
}

/* ---------------------------------------------------------------------- */
/* xce_core_end_session()                                                  */
/* ---------------------------------------------------------------------- */

static void test_core_end_session_null_core_returns_einval(void)
{
    int rc = xce_core_end_session(NULL);

    TEST_ASSERT_EQUAL_INT(-EINVAL, rc);
}

static void test_end_session_unblocks_next_core_go(void)
{
    xce_sched_core_t core;
    int              rc;
    int              go = 0;

    fixture_init(&core);

    rc = xce_core_go(&core);
    TEST_ASSERT_EQUAL_INT(0, rc);
    cbuf_tryget(&(core.wbuf), &go); /* drain the first signal */

    rc = xce_core_end_session(&core);
    TEST_ASSERT_EQUAL_INT(0, rc);
    TEST_ASSERT_EQUAL_INT(
        0, atomic_load_explicit(&(core.session_active), memory_order_acquire));

    /* A second xce_core_go() now succeeds -- proves end_session actually
     * unblocked reentry, not just that the first rejection test worked. */
    rc = xce_core_go(&core);
    TEST_ASSERT_EQUAL_INT(0, rc);
    TEST_ASSERT_EQUAL_INT(
        1, atomic_load_explicit(&(core.session_active), memory_order_acquire));
}

/* ---------------------------------------------------------------------- */
/* xce_core_begin_session() / core_reset_workers() -- per-session worker  */
/* reset skip for excluded hosts (worker-rejoin-resync spec, task 5,      */
/* Req 5.1/5.2). core_reset_workers() (core_session.c) is the live,       */
/* actually-called-every-round mirror of scheduler.c's sched_reset_       */
/* worker_() -- xce_sched_driver_run_normal_mode_round() (driver.c) and   */
/* sched_sim_driver_run() (sim_loop.c) both call xce_core_begin_session() */
/* directly, not the legacy sched_begin_session()/sched_reset_workers()   */
/* path, so this is where the RECONCILING exclusion must actually take   */
/* effect to be observable on either real driver.                        */
/*                                                                        */
/* Observable signal: core_reset_workers() skips (does not zero) a       */
/* worker's resInUse entries for a host it excludes, and resets them to  */
/* 0 for a host it doesn't -- sched_worker_release_host_ctx() itself is  */
/* stubbed to a no-op above (this TU's existing convention), so resInUse */
/* is the one real, unstubbed effect this file can observe directly.     */
/* ---------------------------------------------------------------------- */

/* fixture_worker - builds a minimal xce_sched_worker_t with one resInUse
 * entry pre-set to a known non-zero value (42.0), registered into
 * core->workers under `name` via the same low-level key_add() primitive
 * test_core_events.c's fixture_init() already uses directly for
 * core->plugins (core->workers is populated by core_build_worker()/
 * sched_add_worker() in production, both in core_events.c, a TU this file
 * deliberately does not pull in -- see this file's own header comment). */
static xce_sched_worker_t* fixture_worker(xce_sched_core_t* core,
                                          const char* name, uint32_t status)
{
    xce_sched_worker_t* w = (xce_sched_worker_t*)calloc(1, sizeof(*w));
    xce_val_t*          v = (xce_val_t*)calloc(1, sizeof(*v));
    int                 rc;

    TEST_ASSERT_NOT_NULL(w);
    TEST_ASSERT_NOT_NULL(v);

    w->name   = strdup(name);
    w->status = status;
    zarray_init(&(w->resInUse), 4, NULL);
    zarray_init(&(w->resvals), 4, NULL);

    v->which    = Value_u_fvalue;
    v->u.fvalue = 42.0f;
    zarray_put(&(w->resInUse), 0, v);

    lock(core->workers, RTTI_LOCK_WRITE);
    rc = key_add(core->workers, (void*)w->name, (void*)w);
    unlock(core->workers);
    TEST_ASSERT_GREATER_OR_EQUAL(0, rc);

    return w;
}

static float worker_resinuse_0(xce_sched_worker_t* w)
{
    xce_val_t* v = (xce_val_t*)zarray_get(&(w->resInUse), 0);

    TEST_ASSERT_NOT_NULL(v);
    return v->u.fvalue;
}

static void test_begin_session_resets_available_host_resinuse(void)
{
    xce_sched_core_t    core;
    xce_sched_worker_t* w;
    int                 rc;

    fixture_init(&core);
    w = fixture_worker(&core, "avail-host", 0);

    rc = xce_core_begin_session(&core);
    TEST_ASSERT_EQUAL_INT(0, rc);

    TEST_ASSERT_EQUAL_FLOAT_MESSAGE(
        0.0f, worker_resinuse_0(w),
        "baseline: an ordinary available host's resInUse IS reset every "
        "session -- proves the fixture/harness itself is sound before the "
        "exclusion-path assertions below rely on it");
}

static void test_begin_session_skips_unavailable_host_resinuse(void)
{
    xce_sched_core_t    core;
    xce_sched_worker_t* w;
    int                 rc;

    fixture_init(&core);
    w = fixture_worker(&core, "unavail-host", XCE_WORKER_UNAVAIL);

    rc = xce_core_begin_session(&core);
    TEST_ASSERT_EQUAL_INT(0, rc);

    TEST_ASSERT_EQUAL_FLOAT_MESSAGE(
        42.0f, worker_resinuse_0(w),
        "baseline: the pre-existing IS_WORKER_UNAVAIL exclusion must keep "
        "behaving exactly as before -- unchanged by this task's addition");
}

static void test_begin_session_skips_reconciling_host_resinuse(void)
{
    xce_sched_core_t    core;
    xce_sched_worker_t* w;
    int                 rc;

    fixture_init(&core);
    w = fixture_worker(&core, "reconciling-host", XCE_WORKER_RECONCILING);

    rc = xce_core_begin_session(&core);
    TEST_ASSERT_EQUAL_INT(0, rc);

    TEST_ASSERT_EQUAL_FLOAT_MESSAGE(
        42.0f, worker_resinuse_0(w),
        "Req 5.1: a RECONCILING host must be excluded from the same "
        "per-session reset an UNAVAIL host already is -- its stale "
        "resInUse must NOT be zeroed while reconciliation is outstanding");
}

static void
test_begin_session_skips_host_with_both_unavail_and_reconciling(void)
{
    /* Truth-table completeness: a host that is simultaneously UNAVAIL and
     * RECONCILING must still be excluded (the two conditions are ORed,
     * never ANDed) -- guards against exactly the class of orthogonal-
     * condition mistake flagged for this feature's task 3.2. */
    xce_sched_core_t    core;
    xce_sched_worker_t* w;
    int                 rc;

    fixture_init(&core);
    w = fixture_worker(&core, "both-host",
                       XCE_WORKER_UNAVAIL | XCE_WORKER_RECONCILING);

    rc = xce_core_begin_session(&core);
    TEST_ASSERT_EQUAL_INT(0, rc);

    TEST_ASSERT_EQUAL_FLOAT(42.0f, worker_resinuse_0(w));
}

static void test_begin_session_reincludes_host_once_reconciling_clears(void)
{
    /* Req 5.2: once XCE_WORKER_RECONCILING clears, the SAME filter must
     * stop excluding the host on its very next evaluation -- no separate
     * or new mechanism, just the live status-bit read falling through to
     * the normal reset path. */
    xce_sched_core_t    core;
    xce_sched_worker_t* w;
    int                 rc;
    xce_val_t*          v;

    fixture_init(&core);
    w = fixture_worker(&core, "clears-host", XCE_WORKER_RECONCILING);

    rc = xce_core_begin_session(&core);
    TEST_ASSERT_EQUAL_INT(0, rc);
    TEST_ASSERT_EQUAL_FLOAT(42.0f, worker_resinuse_0(w));

    rc = xce_core_end_session(&core);
    TEST_ASSERT_EQUAL_INT(0, rc);

    /* Simulate a fresh non-zero in-use value still standing at the next
     * session boundary (as it would if a job were still genuinely running
     * there), then clear the reconciling bit -- the only state change
     * Req 5.2 requires. */
    v = (xce_val_t*)zarray_get(&(w->resInUse), 0);
    TEST_ASSERT_NOT_NULL(v);
    v->u.fvalue = 42.0f;
    w->status   = 0;

    rc = xce_core_begin_session(&core);
    TEST_ASSERT_EQUAL_INT(0, rc);

    TEST_ASSERT_EQUAL_FLOAT_MESSAGE(
        0.0f, worker_resinuse_0(w),
        "Req 5.2: clearing XCE_WORKER_RECONCILING alone must be enough to "
        "make the host eligible again on the very next session -- no "
        "separate re-enable action");
}

int main(void)
{
    UNITY_BEGIN();
    RUN_TEST(test_core_go_signals_picker_and_returns_immediately);
    RUN_TEST(test_core_go_rejects_reentrant_call_as_true_noop);
    RUN_TEST(test_core_go_null_core_returns_einval);
    RUN_TEST(test_decisions_reach_callback_in_order_then_end_of_session_marker);
    RUN_TEST(test_core_set_decision_cb_null_core_returns_einval);
    RUN_TEST(test_core_set_decision_cb_registers_cb_and_driver_ctx);
    RUN_TEST(test_core_end_session_null_core_returns_einval);
    RUN_TEST(test_end_session_unblocks_next_core_go);
    RUN_TEST(test_begin_session_resets_available_host_resinuse);
    RUN_TEST(test_begin_session_skips_unavailable_host_resinuse);
    RUN_TEST(test_begin_session_skips_reconciling_host_resinuse);
    RUN_TEST(test_begin_session_skips_host_with_both_unavail_and_reconciling);
    RUN_TEST(test_begin_session_reincludes_host_once_reconciling_clears);
    return UNITY_END();
}