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));
}

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);
    return UNITY_END();
}