myxce/src/scheduler/testsuite/test_repl_trace_e2e_fixture.c

/*
 * test_repl_trace_e2e_fixture.c - the replays and the REPL runner of the
 * end-to-end decision-trace test (test_repl_trace_e2e.c, which documents
 * the scenarios; scheduler-sim-decision-trace task 6.1).
 *
 * write_extra_replay() generates the extra replay with the technique of
 * test_repl_trace_baseline.c's write_contention_replay(): the header and
 * snapshot through the real sched_record_start() (resources and both
 * limits baked in), then workers, queue, the license value and jobs as
 * wire events. run_repl() runs the built xce_scheduler on a replay with a
 * script on stdin in a fresh work directory and returns the console
 * transcript. Kept apart from the scenarios for the 800-line file cap.
 */
#include <errno.h>
#include <limits.h>
#include <stdatomic.h>
#include <stdint.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/stat.h>
#include <sys/types.h>
#include <sys/wait.h>
#include <unistd.h>

#include <unity/unity.h>

#include "config.h"
#include "controller.capnp.h"
#include "recording.h"
#include "resobj.h"
#include "scheduler.h"
#include "sim.h"
#include "utils.h"
#include "xcelimit.h"
#include "xstring.h"

#include "test_repl_trace_e2e.h"

#if !defined(XCE_SCHEDULER_BIN) || !defined(XCE_BASELINE_DIR)
#error "XCE_SCHEDULER_BIN and XCE_BASELINE_DIR must be defined by meson"
#endif

extern int write_capnp_message(FILE* f, struct capn* c);

#define SCRIPT_NAME  "script.txt"
#define OUT_NAME     "console.txt"
#define ERR_NAME     "stderr.txt"
#define LOG_NAME     "sched.log"

#define QUEUE_NAME  "normal"
#define LICENSE_RES "license"
#define SNAPSHOT_TS 1000ULL
#define GLIMIT_NAME "cluster_slots"
#define HLIMIT_NAME "host_slots"
#define ARRAY_SEQ   ((uint64_t)32768)

/* ---------------------------------------------------------------------- */
/* File helpers                                                            */
/* ---------------------------------------------------------------------- */

char* slurp(const char* path, long* len)
{
    FILE* f = fopen(path, "rb");
    char* buf;
    long  n;

    *len = 0;
    if (f == NULL) {
        return NULL;
    }
    if ((fseek(f, 0, SEEK_END) != 0) || ((n = ftell(f)) < 0)) {
        fclose(f);
        return NULL;
    }
    rewind(f);
    buf = (char*)malloc((size_t)n + 1);
    if (buf == NULL) {
        fclose(f);
        return NULL;
    }
    if ((n > 0) && (fread(buf, 1, (size_t)n, f) != (size_t)n)) {
        free(buf);
        fclose(f);
        return NULL;
    }
    buf[n] = '\0';
    fclose(f);
    *len = n;
    return buf;
}

static void spit(const char* path, const char* buf, long len)
{
    FILE* f = fopen(path, "wb");

    TEST_ASSERT_NOT_NULL_MESSAGE(f, path);
    if (len > 0) {
        TEST_ASSERT_EQUAL_UINT((size_t)len, fwrite(buf, 1, (size_t)len, f));
    }
    TEST_ASSERT_EQUAL_INT(0, fclose(f));
}

static void copy_file(const char* from, const char* to)
{
    long  len;
    char* buf = slurp(from, &len);

    TEST_ASSERT_NOT_NULL_MESSAGE(buf, from);
    spit(to, buf, len);
    free(buf);
}

static void join_path(char* out, size_t outsz, const char* dir,
                      const char* name)
{
    int n = snprintf(out, outsz, "%s/%s", dir, name);

    TEST_ASSERT_TRUE((n > 0) && ((size_t)n < outsz));
}

/* ---------------------------------------------------------------------- */
/* Extra replay writer (test_repl_trace_baseline.c's technique)           */
/* ---------------------------------------------------------------------- */

static void bake_resource(var resmgr, const char* name, int flags)
{
    xce_resource_t res;

    memset(&res, 0, sizeof(res));
    res.name    = strdup(name);
    res.valtype = XCE_VAL_NUMERIC;
    res.flags   = flags;
    TEST_ASSERT_NOT_NULL(makeResource(resmgr, &res));
}

static xce_name_val_t* make_nv(const char* name, float v)
{
    xce_name_val_t* nv  = (xce_name_val_t*)calloc(1, sizeof(*nv));
    xce_val_t*      val = (xce_val_t*)calloc(1, sizeof(*val));

    TEST_ASSERT_NOT_NULL(nv);
    TEST_ASSERT_NOT_NULL(val);
    val->which    = Value_u_fvalue;
    val->u.fvalue = v;
    nv->name      = strdup(name);
    nv->val       = val;
    return nv;
}

static xce_worker_t* make_wire_worker(const char* name, float slots,
                                      float mem)
{
    xce_worker_t*    w  = (xce_worker_t*)calloc(1, sizeof(*w));
    xce_name_val_t** nv = (xce_name_val_t**)calloc(2, sizeof(*nv));

    TEST_ASSERT_NOT_NULL(w);
    TEST_ASSERT_NOT_NULL(nv);
    nv[0]                = make_nv("slots", slots);
    nv[1]                = make_nv("mem", mem);
    w->name              = strdup(name);
    w->nresvals_to_sched = 2;
    w->resvals_to_sched  = nv;
    return w;
}

static xce_job_t* make_push_job(uint64_t jobid, uint64_t sub_time,
                                const char* resreq, const char* user,
                                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   = sub_time;
    j->queue_time = sub_time;

    j->submit = (job_submit_data_t*)calloc(1, sizeof(*(j->submit)));
    TEST_ASSERT_NOT_NULL(j->submit);
    j->submit->queue        = (char*)QUEUE_NAME;
    j->submit->user         = (char*)user;
    j->submit->project_name = (char*)"proj1";
    j->submit->resreq       = (char*)resreq;
    j->submit->mintasks     = mintasks;
    j->submit->maxtasks     = maxtasks;
    return j;
}

static void write_event(FILE* f, replay_event_t* src)
{
    struct capn          c;
    struct capn_segment* cs;
    ReplayEvent_ptr      p;

    capn_init_malloc(&c);
    cs = capn_root(&c).seg;
    encode_ReplayEvent_ptr(cs, &p, src);
    capn_setp(capn_root(&c), 0, p.p);
    TEST_ASSERT_EQUAL_INT(0, write_capnp_message(f, &c));
    capn_free(&c);
}

static void append_workers(FILE* f, uint64_t ts, xce_worker_t** w, int n)
{
    xce_workers_data_t wrap;
    replay_event_t     src;

    memset(&wrap, 0, sizeof(wrap));
    wrap.n_total = n;
    wrap.count   = n;
    wrap.workers = w;

    memset(&src, 0, sizeof(src));
    src.ts                = ts;
    src.body_which        = ReplayEvent_body_workersPush;
    src.body.workers_push = &wrap;
    write_event(f, &src);
}

static void append_queue(FILE* f, uint64_t ts)
{
    xce_queue_t    q;
    replay_event_t src;

    memset(&q, 0, sizeof(q));
    q.queue    = (char*)QUEUE_NAME;
    q.status   = QUEUE_STAT_OPEN;
    q.priority = 1;

    memset(&src, 0, sizeof(src));
    src.ts                = ts;
    src.body_which        = ReplayEvent_body_queueUpdate;
    src.body.queue_update = &q;
    write_event(f, &src);
}

static void append_license_value(FILE* f, uint64_t ts, float total)
{
    xce_val_t         val;
    xce_resval_t      rv;
    xce_resval_t*     rvs[1];
    xce_resval_list_t list;
    replay_event_t    src;

    memset(&val, 0, sizeof(val));
    val.which    = Value_u_fvalue;
    val.u.fvalue = total;

    memset(&rv, 0, sizeof(rv));
    rv.name = (char*)LICENSE_RES;
    rv.val  = &val;
    rvs[0]  = &rv;

    memset(&list, 0, sizeof(list));
    list.count = 1;
    list.vals  = rvs;

    memset(&src, 0, sizeof(src));
    src.ts                = ts;
    src.body_which        = ReplayEvent_body_resvalsFull;
    src.body.resvals_full = &list;
    write_event(f, &src);
}

static void append_jobs(FILE* f, uint64_t ts, xce_job_t** jobs, int n)
{
    xce_jobs_data_t wrap;
    replay_event_t  src;

    memset(&wrap, 0, sizeof(wrap));
    wrap.n_total = n;
    wrap.count   = n;
    wrap.jobs    = jobs;

    memset(&src, 0, sizeof(src));
    src.ts             = ts;
    src.body_which     = ReplayEvent_body_jobsPush;
    src.body.jobs_push = &wrap;
    write_event(f, &src);
}

static void append_state_change(FILE* f, uint64_t ts, uint64_t jobid,
                                uint32_t old_state, uint32_t new_state)
{
    job_state_change_event_t change;
    replay_event_t           src;

    memset(&change, 0, sizeof(change));
    change.jobid     = jobid;
    change.old_state = old_state;
    change.new_state = new_state;

    memset(&src, 0, sizeof(src));
    src.ts                    = ts;
    src.body_which            = ReplayEvent_body_jobStateChange;
    src.body.job_state_change = &change;
    write_event(f, &src);
}

static void append_session(FILE* f, uint64_t begin_round)
{
    struct capn          c;
    struct capn_segment* cs;
    ReplaySession_ptr    sp;

    capn_init_malloc(&c);
    cs = capn_root(&c).seg;
    sp = new_ReplaySession(cs);
    ReplaySession_set_beginRound(sp, begin_round);
    capn_setp(capn_root(&c), 0, sp.p);
    TEST_ASSERT_EQUAL_INT(0, write_capnp_message(f, &c));
    capn_free(&c);
}

static void append_end_of_stream(FILE* f)
{
    replay_event_t src;

    memset(&src, 0, sizeof(src));
    src.body_which = ReplayEvent_body_endSession;
    write_event(f, &src);
}

/* Header + snapshot via the real sched_record_start(), resources and both
 * limits baked into the snapshot (the baseline fixture's configuration). */
static FILE* write_header_and_snapshot(char* out_path, size_t out_len,
                                       const char* record_dir)
{
    xce_scheduler_t     sched;
    xce_sched_plugin_t* fcfs;
    xce_limit_val_t     gval = { .resname = (char*)"slots",
                                 .percent = 0,
                                 .val     = 6.0 };
    xce_limit_config_t  gcfg = { .name = (char*)GLIMIT_NAME, .val = &gval };
    xce_limit_val_t     hval = { .resname = (char*)"slots",
                                 .percent = 1,
                                 .val     = 0.75 };
    xce_limit_config_t  hcfg = { .name = (char*)HLIMIT_NAME, .val = &hval };
    xce_limit_config_t* limits[2] = { &gcfg, &hcfg };
    xce_conf4sched_t    config;

    memset(&sched, 0, sizeof(sched));
    mylock_init(&(sched.record_lock));
    TEST_ASSERT_EQUAL_INT(0, xce_sched_core_init(&(sched.core)));

    fcfs = builtin_fcfs_new_plugin();
    TEST_ASSERT_NOT_NULL(fcfs);
    TEST_ASSERT_TRUE(plugin_register_core(&(sched.core), fcfs) >= 0);

    bake_resource(sched.core.resmgr, "slots", XCE_RESF_CONSUMABLE);
    bake_resource(sched.core.resmgr, "mem", XCE_RESF_CONSUMABLE);
    bake_resource(sched.core.resmgr, LICENSE_RES,
                  XCE_RESF_GLOBAL | XCE_RESF_SHARED | XCE_RESF_CONSUMABLE);

    memset(&config, 0, sizeof(config));
    config.nlimits    = 2;
    config.limits     = limits;
    sched.core.config = &config;

    atomic_store(&(sched.core.virtual_clock_base), SNAPSHOT_TS);
    atomic_store(&(sched.core.speed_num), 0ULL);
    atomic_store(&(sched.core.speed_den), 1ULL);

    sched.core.job_pick_plugin = (char*)"builtin_fcfs";
    sched.record_dir           = (char*)record_dir;

    TEST_ASSERT_EQUAL_INT(0, sched_record_start(&sched, EXTRA_REPLAY));
    TEST_ASSERT_TRUE(strlen(sched.record_file_path) < out_len);
    strcpy(out_path, sched.record_file_path);
    return sched.record_file;
}

static uint64_t array_elem(uint32_t k)
{
    return MAKE_JOBID(0, 0, 0, TAG_JOB_ARRAY, ARRAY_SEQ | (uint64_t)k);
}

/*
 * write_extra_replay - w01 (4 slots, 16384 mem) and w02 (4 slots, 32768
 * mem), the baseline's license and limits. Rounds:
 *   2000  201 compound {license} + {2 x mem} (satisfied, 3 slots);
 *         202 compound whose first member (4 tasks) breaks the cluster-wide
 *         limit (6 - 3 < 4), so its second member is not tried;
 *         array elements [1] and [2] (1 slot each, dispatched);
 *         203 selects mem>20000 (w01 mismatches), pended by the
 *         cluster-wide limit (6 of 6 in use)
 *   3000  after 201 and both elements finish: 202 and 203 are tried again
 *   4000  after 202 and 203 finish: 204 and 205 (identical requests, more
 *         mem than any host has; bug-1932) pend in every session -- 204
 *         by a full search, 205 by the unsatisfiable-request cache
 */
void write_extra_replay(const char* dest)
{
    char          record_dir[PATH_MAX];
    char          path[PATH_MAX];
    FILE*         f;
    xce_worker_t* workers[2];
    xce_job_t*    jobs[7];

    snprintf(record_dir, sizeof(record_dir), "/tmp/sched_tre2e_gen_XXXXXX");
    TEST_ASSERT_NOT_NULL(mkdtemp(record_dir));

    f = write_header_and_snapshot(path, sizeof(path), record_dir);

    workers[0] = make_wire_worker("w01", 4.0f, 16384.0f);
    workers[1] = make_wire_worker("w02", 4.0f, 32768.0f);
    append_workers(f, 1001ULL, workers, 2);
    append_queue(f, 1002ULL);
    append_license_value(f, 1002ULL, 2.0f);

    jobs[0] = make_push_job(201, 1101,
                            "1*{rusage[license=1]} + "
                            "2*{rusage[mem=1024]}",
                            "alice", 1, 1);
    jobs[1] = make_push_job(202, 1102,
                            "4*{rusage[mem=512]} + "
                            "1*{rusage[mem=512]}",
                            "bob", 1, 1);
    jobs[2] = make_push_job(array_elem(1), 1103, "rusage[mem=256]", "bob",
                            1, 1);
    jobs[3] = make_push_job(array_elem(2), 1103, "rusage[mem=256]", "bob",
                            1, 1);
    jobs[4] = make_push_job(203, 1104, "select[mem>20000] rusage[mem=512]",
                            "alice", 1, 1);
    jobs[5] = make_push_job(204, 1105, "rusage[mem=999999]", "carol", 1, 1);
    jobs[6] = make_push_job(205, 1106, "rusage[mem=999999]", "carol", 1, 1);
    append_jobs(f, 1003ULL, jobs, 7);

    append_session(f, 2000ULL);
    append_state_change(f, 2500ULL, 201, JOB_STAT_RUN, JOB_STAT_DONE);
    append_state_change(f, 2600ULL, array_elem(1), JOB_STAT_RUN,
                        JOB_STAT_DONE);
    append_state_change(f, 2700ULL, array_elem(2), JOB_STAT_RUN,
                        JOB_STAT_DONE);
    append_session(f, 3000ULL);
    append_state_change(f, 3500ULL, 202, JOB_STAT_RUN, JOB_STAT_DONE);
    append_state_change(f, 3600ULL, 203, JOB_STAT_RUN, JOB_STAT_DONE);
    append_session(f, 4000ULL);

    append_end_of_stream(f);
    TEST_ASSERT_EQUAL_INT(0, fclose(f));

    copy_file(path, dest);
    unlink(path);
    rmdir(record_dir);
}

/* ---------------------------------------------------------------------- */
/* Running a scripted REPL session                                         */
/* ---------------------------------------------------------------------- */


/* Runs the built scheduler on `replay` (BASE_REPLAY or EXTRA_REPLAY) in a
 * fresh work directory with `script` on stdin and returns the console
 * transcript. Paths are relative to the work directory. */
void run_repl(run_t* r, const char* replay, const char* script)
{
    char  path[PATH_MAX];
    pid_t pid;
    int   status;

    memset(r, 0, sizeof(*r));
    snprintf(r->dir, sizeof(r->dir), "/tmp/sched_tre2e_XXXXXX");
    TEST_ASSERT_NOT_NULL(mkdtemp(r->dir));

    join_path(path, sizeof(path), r->dir, replay);
    if (strcmp(replay, BASE_REPLAY) == 0) {
        char src[PATH_MAX];

        join_path(src, sizeof(src), XCE_BASELINE_DIR, BASE_REPLAY);
        copy_file(src, path);
    } else {
        write_extra_replay(path);
    }
    join_path(path, sizeof(path), r->dir, SCRIPT_NAME);
    spit(path, script, (long)strlen(script));

    fflush(stdout);
    fflush(stderr);
    pid = fork();
    TEST_ASSERT_TRUE_MESSAGE(pid >= 0, "fork must succeed");
    if (pid == 0) {
        if ((chdir(r->dir) != 0)
            || (freopen(SCRIPT_NAME, "r", stdin) == NULL)
            || (freopen(OUT_NAME, "w", stdout) == NULL)
            || (freopen(ERR_NAME, "w", stderr) == NULL)) {
            _exit(120);
        }
        setenv("TERM", "dumb", 1);
        execl(XCE_SCHEDULER_BIN, XCE_SCHEDULER_BIN, "--sim", replay,
              "--interactive", "--logfile", LOG_NAME, (char*)NULL);
        _exit(121);
    }
    TEST_ASSERT_EQUAL_INT(pid, waitpid(pid, &status, 0));
    TEST_ASSERT_TRUE_MESSAGE(WIFEXITED(status),
                             "the scheduler must exit normally");
    TEST_ASSERT_EQUAL_INT_MESSAGE(0, WEXITSTATUS(status),
                                  "the session must end with quit's exit(0)");

    join_path(path, sizeof(path), r->dir, OUT_NAME);
    r->out = slurp(path, &(r->out_len));
    TEST_ASSERT_NOT_NULL(r->out);
    if (getenv("XCE_TRACE_E2E_DUMP") != NULL) {
        fprintf(stderr, "==== %s (%s)\n%s", replay, r->dir, r->out);
    }
}

void run_done(run_t* r)
{
    static const char* const names[] = { BASE_REPLAY, EXTRA_REPLAY,
                                         SCRIPT_NAME, OUT_NAME, ERR_NAME,
                                         LOG_NAME, TRACE_FILE };
    char   path[PATH_MAX];
    size_t i;

    free(r->out);
    r->out = NULL;
    if (getenv("XCE_TRACE_E2E_DUMP") != NULL) {
        return;
    }
    for (i = 0; i < sizeof(names) / sizeof(names[0]); i++) {
        int n = snprintf(path, sizeof(path), "%s/%s", r->dir, names[i]);

        if ((n > 0) && ((size_t)n < sizeof(path))) {
            unlink(path);
        }
    }
    rmdir(r->dir);
}