myxce/src/scheduler/src/sched_trace_scan.c

/*
 * sched_trace_scan.c - decision trace record model, part 2
 * (scheduler-sim-decision-trace spec, design.md "TraceModel").
 *
 * The host scan's entries and counters: positions, host verdicts and the
 * rows held under them, the summary-level per-position tallies (5.9),
 * rollback runs, commit and finalize failures, and the per-requirement
 * search-end statement with its decision-wide counters (9.3). Split from
 * sched_trace.c (task 2.4), which keeps the record, its arena and the entry
 * sequence these functions append to (sched_trace_internal.h).
 *
 * Same rules as sched_trace.c: one building thread, no locking, and no
 * I/O and no logging (test_core_isolation.c scans this file).
 */
#include <stdlib.h>
#include <string.h>

#include "sched/trace.h"
#include "sched_trace_internal.h"

/* Starts the current position's tallies over. */
static void pos_tally_reset(xce_sched_trace_rec_t* rec)
{
    memset(rec->pos_counts, 0, sizeof(rec->pos_counts));
    sti_pos_limits_clear(rec);
}

void sched_trace_pos_begin(xce_sched_trace_rec_t* rec, int pos, int slots)
{
    sched_trace_entry_t* e;

    rec->cur_pos      = pos;
    rec->scan_err     = 0;
    rec->scan_err_set = false;
    pos_tally_reset(rec);

    e = sti_scope_begin(rec, SCHED_TRACE_E_POS_BEGIN);
    if (e == NULL) {
        return;
    }
    e->u.pos_begin.pos   = pos;
    e->u.pos_begin.slots = slots;
}

void sched_trace_pos_scan_error(xce_sched_trace_rec_t* rec, int err)
{
    sti_rollback_run_end(rec);
    rec->scan_err     = err;
    rec->scan_err_set = true;
}

/* Fills a POS_SUMMARY entry from the current position's tallies; only
 * limits some host was rejected for are kept. */
static void pos_summary_fill(xce_sched_trace_rec_t* rec, sched_trace_entry_t* e,
                             xce_sched_trace_pos_end_t how, const char* host,
                             int err)
{
    int n = 0;
    int k;

    e->u.pos_summary.pos  = rec->cur_pos;
    e->u.pos_summary.how  = how;
    e->u.pos_summary.err  = err;
    e->u.pos_summary.host = sched_trace_arena_strdup(rec, host);
    memcpy(e->u.pos_summary.counts, rec->pos_counts,
           sizeof(e->u.pos_summary.counts));

    for (k = 0; k < rec->pos_nlimits; k++) {
        if (rec->pos_limits[k].count > 0) {
            n++;
        }
    }
    if (n == 0) {
        return;
    }
    e->u.pos_summary.limits = (sched_trace_limit_tally_t*)
        sched_trace_arena_alloc(rec, sizeof(sched_trace_limit_tally_t)
                                         * (size_t)n);
    if (e->u.pos_summary.limits == NULL) {
        return;
    }
    n = 0;
    for (k = 0; k < rec->pos_nlimits; k++) {
        const struct sched_trace_tally_build_s* b = &rec->pos_limits[k];
        sched_trace_limit_tally_t*              t;

        if (b->count <= 0) {
            continue;
        }
        t        = &e->u.pos_summary.limits[n++];
        t->limit = sched_trace_arena_strdup(rec, b->limit);
        t->scope = sched_trace_arena_strdup(rec, b->scope);
        t->count = b->count;
    }
    e->u.pos_summary.nlimits = n;
}

void sched_trace_pos_end(xce_sched_trace_rec_t*    rec,
                         xce_sched_trace_pos_end_t how, const char* host)
{
    sched_trace_entry_t* e;
    int                  err = 0;

    if (rec->scan_err_set) {
        /* the host scan stopped: the runner's UNFILLED is a scan error */
        err = rec->scan_err;
        if (how == XCE_TRACE_POS_UNFILLED) {
            how = XCE_TRACE_POS_SCAN_ERROR;
        }
    }
    rec->scan_err     = 0;
    rec->scan_err_set = false;

    if (rec->level == XCE_TRACE_LEVEL_SUMMARY) {
        /* the position's end carries its host tallies (5.9) */
        e = sti_scope_end(rec, SCHED_TRACE_E_POS_SUMMARY);
        if (e != NULL) {
            pos_summary_fill(rec, e, how, host, err);
        }
        pos_tally_reset(rec);
        return;
    }

    e = sti_scope_end(rec, SCHED_TRACE_E_POS_END);
    if (e == NULL) {
        return;
    }
    e->u.pos_end.pos  = rec->cur_pos;
    e->u.pos_end.how  = how;
    e->u.pos_end.err  = err;
    e->u.pos_end.host = sched_trace_arena_strdup(rec, host);
}

void sched_trace_host_begin(xce_sched_trace_rec_t*    rec,
                            const xce_sched_obj_t*    obj,
                            const xce_sched_worker_t* w)
{
    sti_rollback_run_end(rec);
    rec->host_seq++;
    rec->scope_obj         = obj;
    rec->scope_w           = w;
    rec->host_row_verdicts = 0;
    sti_host_rows_clear(rec); /* a scope that ended without its verdict */
}

/* Counts one host evaluation: decision-wide per verdict kind, and at
 * summary level per position, with each limit the host hit counted once
 * when its verdict is limit exceeded (5.9, 9.3). */
static void host_count(xce_sched_trace_rec_t* rec, unsigned verdicts)
{
    int k;

    for (k = 0; k < XCE_TRACE_V_NKINDS; k++) {
        if ((verdicts & (1u << k)) != 0) {
            rec->counters.host_evals[k]++;
            rec->pos_counts[k]++;
        }
    }
    if ((verdicts & XCE_TRACE_V_LIMIT) == 0) {
        return;
    }
    for (k = 0; k < rec->pos_nlimits; k++) {
        if (rec->pos_limits[k].last_host == rec->host_seq) {
            rec->pos_limits[k].count++;
        }
    }
}

/* Copies a row's strings into the arena (NULL stays NULL). */
static void row_to_arena(xce_sched_trace_rec_t* rec, sched_trace_row_t* dst,
                         const sched_trace_row_t* src)
{
    *dst             = *src;
    dst->res         = sched_trace_arena_strdup(rec, src->res);
    dst->limit       = sched_trace_arena_strdup(rec, src->limit);
    dst->scope       = sched_trace_arena_strdup(rec, src->scope);
    dst->limited_res = sched_trace_arena_strdup(rec, src->limited_res);
}

/* Stores the current host's held rows after its verdict line, one level
 * below it, then drops them. Each counts against the budget. */
static void host_rows_flush(xce_sched_trace_rec_t* rec)
{
    int k;

    for (k = 0; k < rec->host_nrows; k++) {
        sched_trace_entry_t* e = sti_entry_add(rec, SCHED_TRACE_E_HOST_ROW);

        if (e == NULL) {
            continue;
        }
        e->depth++;
        row_to_arena(rec, &e->u.row, &rec->host_rows[k]);
    }
    sti_host_rows_clear(rec);
}

/* NULL-preserving heap copy for held rows; sets *failed on failure. */
static const char* row_strdup(const char* s, bool* failed)
{
    char* d;

    if (s == NULL) {
        return NULL;
    }
    d = strdup(s);
    if (d == NULL) {
        *failed = true;
    }
    return d;
}

void sched_trace_row(xce_sched_trace_rec_t* rec, const sched_trace_row_t* row)
{
    sched_trace_entry_t* e;
    sched_trace_row_t*   h;
    bool                 failed = false;

    sti_rollback_run_end(rec);
    if (row == NULL) {
        return;
    }

    if (rec->scope_w == NULL) {
        /* the cluster-wide check: stored now, before its check-end line */
        e = sti_entry_add(rec, SCHED_TRACE_E_CHECK_ROW);
        if (e != NULL) {
            row_to_arena(rec, &e->u.row, row);
        }
        return;
    }

    /* a host scope: no host lines at summary level; at full level the row
     * waits for the host's verdict line, which it belongs under */
    if ((rec->level == XCE_TRACE_LEVEL_SUMMARY) || rec->incomplete) {
        return;
    }
    if (rec->host_nrows == rec->host_rows_cap) {
        int   cap = (rec->host_rows_cap > 0) ? rec->host_rows_cap * 2 : 4;
        void* p   = realloc(rec->host_rows, sizeof(*h) * (size_t)cap);

        if (p == NULL) {
            rec->incomplete = true; /* design "Error Strategy" */
            return;
        }
        rec->host_rows     = (sched_trace_row_t*)p;
        rec->host_rows_cap = cap;
    }
    h              = &rec->host_rows[rec->host_nrows];
    *h             = *row;
    h->res         = row_strdup(row->res, &failed);
    h->limit       = row_strdup(row->limit, &failed);
    h->scope       = row_strdup(row->scope, &failed);
    h->limited_res = row_strdup(row->limited_res, &failed);
    if (failed) {
        free((void*)h->res);
        free((void*)h->limit);
        free((void*)h->scope);
        free((void*)h->limited_res);
        rec->incomplete = true;
        return;
    }
    rec->host_nrows++;
}

void sched_trace_host_verdict(xce_sched_trace_rec_t* rec, unsigned verdicts,
                              int err)
{
    sched_trace_entry_t* e;
    sched_trace_phase_t  phase = SCHED_TRACE_PHASE_NONE;
    const char*          name
        = (rec->scope_w != NULL) ? rec->scope_w->name : NULL;

    if (verdicts == 0) {
        /* from the rows the per-host check recorded; none means the
         * check's own selection branch rejected the host */
        verdicts = rec->host_row_verdicts & (XCE_TRACE_V_SHORTFALL
                                             | XCE_TRACE_V_LIMIT);
        if (verdicts == 0) {
            verdicts = XCE_TRACE_V_SELECT;
        }
        phase = SCHED_TRACE_PHASE_PERHOST;
    } else if ((verdicts & XCE_TRACE_V_SELECT) != 0) {
        phase = SCHED_TRACE_PHASE_PREFILTER;
    } else if ((verdicts & (XCE_TRACE_V_SHORTFALL | XCE_TRACE_V_LIMIT)) != 0) {
        phase = SCHED_TRACE_PHASE_PERHOST;
    }

    /* one host evaluation, counted once under each kind it matched (9.3) */
    host_count(rec, verdicts);

    /* the host scope ends here */
    rec->scope_obj         = NULL;
    rec->scope_w           = NULL;
    rec->host_row_verdicts = 0;
    rec->host_seq++; /* later hits belong to no host until the next one */
    sti_rollback_run_end(rec);

    if (rec->level == XCE_TRACE_LEVEL_SUMMARY) {
        sti_host_rows_clear(rec);
        return; /* tallied for the position summary, no host line */
    }

    e = sti_entry_add(rec, SCHED_TRACE_E_HOST);
    if (e != NULL) {
        e->u.host.host     = sched_trace_arena_strdup(rec, name);
        e->u.host.verdicts = verdicts;
        e->u.host.phase    = phase;
        e->u.host.err      = err;
    }
    /* the host's unmet rows, under its verdict line (5.4, 5.5) */
    host_rows_flush(rec);
}

void sched_trace_commit(xce_sched_trace_rec_t* rec)
{
    sched_trace_entry_t* e = sti_entry_add(rec, SCHED_TRACE_E_COMMIT);

    if (e == NULL) {
        return;
    }
    e->u.commit.plan_no = rec->cur_plan_no;
}

void sched_trace_finalize_failed(xce_sched_trace_rec_t* rec, int err)
{
    sched_trace_entry_t* e = sti_entry_add(rec, SCHED_TRACE_E_FINALIZE_FAILED);

    if (e == NULL) {
        return;
    }
    e->u.commit.plan_no = rec->cur_plan_no;
    e->u.commit.err     = err;
}


static bool tally_str_eq(const char* a, const char* b)
{
    if ((a == NULL) || (b == NULL)) {
        return a == b;
    }
    return strcmp(a, b) == 0;
}

/* NULL-preserving heap copy for the builder's tallies; sets *failed when a
 * non-NULL string could not be copied. */
static char* tally_strdup(const char* s, bool* failed)
{
    char* d;

    if (s == NULL) {
        return NULL;
    }
    d = strdup(s);
    if (d == NULL) {
        *failed = true;
    }
    return d;
}

void sched_trace_host_limit_hit(xce_sched_trace_rec_t* rec, const char* limit,
                                const char* scope)
{
    struct sched_trace_tally_build_s* b;
    bool                              failed = false;
    int                               k;

    sti_rollback_run_end(rec);
    if (rec->scope_w == NULL) {
        return; /* not a host scope: the cluster-wide check's rows */
    }
    rec->host_row_verdicts |= XCE_TRACE_V_LIMIT;
    if ((rec->level != XCE_TRACE_LEVEL_SUMMARY) || rec->incomplete) {
        return; /* full level shows the rows themselves (2.2) */
    }

    for (k = 0; k < rec->pos_nlimits; k++) {
        b = &rec->pos_limits[k];
        if (tally_str_eq(b->limit, limit) && tally_str_eq(b->scope, scope)) {
            b->last_host = rec->host_seq; /* once per host, at its verdict */
            return;
        }
    }

    if (rec->pos_nlimits == rec->pos_limits_cap) {
        int   cap = (rec->pos_limits_cap > 0) ? rec->pos_limits_cap * 2 : 4;
        void* p   = realloc(rec->pos_limits, sizeof(*b) * (size_t)cap);

        if (p == NULL) {
            rec->incomplete = true;
            return;
        }
        rec->pos_limits     = (struct sched_trace_tally_build_s*)p;
        rec->pos_limits_cap = cap;
    }
    b            = &rec->pos_limits[rec->pos_nlimits];
    b->limit     = tally_strdup(limit, &failed);
    b->scope     = tally_strdup(scope, &failed);
    b->count     = 0;
    b->last_host = rec->host_seq;
    if (failed) {
        free(b->limit);
        free(b->scope);
        rec->incomplete = true;
        return;
    }
    rec->pos_nlimits++;
}

/* Appends one host to a ROLLBACK entry, growing its array in the arena. */
static void rollback_append(xce_sched_trace_rec_t* rec, sched_trace_entry_t* e,
                            const char* host)
{
    if (e->u.rollback.nhosts == e->u.rollback.cap) {
        int          cap = (e->u.rollback.cap > 0) ? e->u.rollback.cap * 2 : 4;
        const char** h   = (const char**)sched_trace_arena_alloc(
            rec, sizeof(char*) * (size_t)cap);

        if (h == NULL) {
            return;
        }
        if (e->u.rollback.nhosts > 0) {
            memcpy(h, e->u.rollback.hosts,
                   sizeof(char*) * (size_t)e->u.rollback.nhosts);
        }
        e->u.rollback.hosts = h;
        e->u.rollback.cap   = cap;
    }
    e->u.rollback.hosts[e->u.rollback.nhosts++]
        = sched_trace_arena_strdup(rec, host);
}

void sched_trace_rollback_host(xce_sched_trace_rec_t* rec, const char* host)
{
    sched_trace_entry_t* e;

    if (rec->rollback_open) {
        /* the same run of released hosts: the same line */
        if ((rec->rollback_e != NULL) && !rec->incomplete) {
            rollback_append(rec, rec->rollback_e, host);
        }
        return;
    }

    e                  = sti_entry_add(rec, SCHED_TRACE_E_ROLLBACK);
    rec->rollback_open = true; /* also when omitted: one line either way */
    rec->rollback_e    = e;
    if (e != NULL) {
        rollback_append(rec, e, host);
    }
}

void sched_trace_search_end(xce_sched_trace_rec_t* rec, int rc)
{
    sched_trace_entry_t*      e;
    sched_trace_search_stmt_t stmt = SCHED_TRACE_SEARCH_NO_STMT;
    int                       t    = rec->scope_tried;
    int                       d    = rec->scope_discarded;
    int                       x    = rec->scope_not_tried;
    int                       g    = t + d + x;

    if (rc == 0) {
        if (g == 0) {
            stmt = SCHED_TRACE_SEARCH_NONE_GENERATED;
            rec->counters.reqs_none_generated++;
        } else if (d == g) {
            stmt = SCHED_TRACE_SEARCH_ALL_DISCARDED;
            rec->counters.reqs_all_discarded++;
        } else if ((t > 0) && !rec->scope_plan_err) {
            stmt = SCHED_TRACE_SEARCH_NONE_PLACED;
            rec->counters.reqs_none_placed++;
        }
    }

    /* the search is over: a later search in this requirement counts anew */
    rec->scope_tried     = 0;
    rec->scope_discarded = 0;
    rec->scope_not_tried = 0;
    rec->scope_plan_err  = false;

    e = sti_entry_add(rec, SCHED_TRACE_E_SEARCH_END);
    if (e == NULL) {
        return;
    }
    e->u.search_end.generated = g;
    e->u.search_end.tried     = t;
    e->u.search_end.discarded = d;
    e->u.search_end.not_tried = x;
    e->u.search_end.rc        = rc;
    e->u.search_end.stmt      = stmt;
}