myxce/src/scheduler/src/schedobj.c

#include <errno.h>
#include <rtti.h>
#include <stdlib.h>
#include <string.h>

#include "controller.capnp.h"
#include "kvmgr.h"
#include "sched/types.h"
#include "scheduler.h"
#include "utils.h"
#include "xce/types.h"
#include "xcealloc.h"
#include "xlist.h"
#include "xvec.h"
#include "xstring.h"

void kill_sched_job(void* arg)
{
    xce_sched_job_t* job = (xce_sched_job_t*)arg;

    if (job == NULL) {
        return;
    }

    /* release the stored rusage report, if any (Req 6 / bug-765 G-5).
     * free_JobRusages_ptr is NULL-safe and clears the field. */
    free_JobRusages_ptr(&job->rusages);
}

request_res_t* make_resource_request(char* res, float value)
{
    request_res_t* req = NULL;

    req = (request_res_t*)calloc(1, sizeof(*req));
    if (req == NULL) {
        return NULL;
    }

    req->res   = makeString(res);
    req->value = value;

    return req;
}

static int make_primary_schedobj(xce_sched_obj_t*      obj,
                                 xce_primary_resreq_t* resreq)
{
    xce_primary_schedobj_t* primary
        = (xce_primary_schedobj_t*)calloc(1, sizeof(*primary));

    obj->tag = XCE_RESREQ_PRIMARY;
    if (primary == NULL) {
        return -ENOMEM;
    }

    primary->resreq = resreq;
    primary->reason = NULL;
    primary->owner  = obj;

    obj->u.primary = primary;

    return 1;
}

static int make_compound_schedobj(xce_sched_obj_t*       obj,
                                  xce_compound_resreq_t* resreq)
{
    xce_compound_schedobj_t* compound
        = (xce_compound_schedobj_t*)calloc(1, sizeof(*compound));

    obj->tag = XCE_RESREQ_COMPOUND;

    if (compound == NULL) {
        return -EINVAL;
    }

    compound->resreq = resreq;
    compound->owner  = obj;

    obj->u.compound = compound;

    return 1;
}

static int make_alter_schedobj(xce_sched_obj_t* obj, xce_alter_resreq_t* resreq)
{
    int i;
    int rc = 0;

    obj->tag = XCE_RESREQ_ALTER;
    xlist_init(&(obj->u.alters), 4);

    for (i = 0; i < resreq->compounds.count; i++) {
        xce_compound_schedobj_t* cobj = NULL;
        xce_compound_resreq_t*   cresreq
            = (xce_compound_resreq_t*)xlist_get(&(resreq->compounds), i);

        if (cresreq == NULL) {
            continue;
        }

        cobj = (xce_compound_schedobj_t*)calloc(1, sizeof(*cobj));
        if (cobj == NULL) {
            return -ENOMEM;
        }

        cobj->resreq = cresreq;
        cobj->owner  = obj;

        xlist_store(&(obj->u.alters), cobj);
    }

    return 1;
}

static int make_sched_obj(xce_sched_obj_t* obj, xce_scheduler_t* sched,
                          xce_job_t* j, xce_resreq_t* resreq)
{
    int rc = 0;

    obj->user    = makeString(j->submit->user);
    obj->queue   = makeString(j->submit->queue);
    obj->project = makeString(j->submit->project_name);

    xlist_init(&(obj->glimits), 8);
    zarray_init(&(obj->hlimits), 32, NULL);

    switch (resreq->tag) {
    case XCE_RESREQ_PRIMARY:
        rc = make_primary_schedobj(obj, resreq->u.primary);
        break;
    case XCE_RESREQ_COMPOUND:
        rc = make_compound_schedobj(obj, resreq->u.compound);
        break;
    case XCE_RESREQ_ALTER:
        rc = make_alter_schedobj(obj, resreq->u.alter);
        break;
    }

    obj->configver = 0;

    return 1;
}

void reset_sched_obj(xce_sched_obj_t* obj)
{
    xlist_fini(&(obj->glimits));
    zarray_fini(&(obj->hlimits));

    return;
}

xce_sched_job_t* make_sched_job(void* arg, xce_job_t* j)
{
    xce_sched_job_t* job    = (xce_sched_job_t*)calloc(1, sizeof(*job));
    xce_scheduler_t* sched  = (xce_scheduler_t*)arg;
    int              rc     = 0;
    char*            errmsg = NULL;

    if ((sched == NULL) || (job == NULL)) {
        return NULL;
    }

    job->jobid      = j->jobid;
    job->sub_time   = j->sub_time;
    job->queue_time = j->queue_time;
    job->start_time = j->start_time;
    job->state      = j->state;
    job->user       = sched_get_user(sched, j->submit->user);
    job->queue      = sched_get_queue(sched, j->submit->queue);
    job->project    = sched_get_project(sched, j->submit->project_name);
    job->resreq     = (xce_resreq_t*)calloc(1, sizeof(*(job->resreq)));
    if (job->resreq == NULL) {
        return NULL;
    }

    {
        char* resreq_str = j->submit->resreq;
        if (NULLSTR(resreq_str)) {
            resreq_str = "type == any";
        }
        rc = parse_resreq(job->resreq, resreq_str, sched->resmgr, &errmsg);
    }
    if (rc < 0) {
        ERROR("%s parse_resreq failed, %s", __func__, errmsg);
        FREEUP(errmsg);

        return NULL;
    }
    job->mintasks = j->submit->mintasks;
    job->maxtasks = j->submit->maxtasks;
    job->priority = 0; /* FIXME: support job level priority */

    job->schedobj = (xce_sched_obj_t*)calloc(1, sizeof(*(job->schedobj)));
    if (job->schedobj == NULL) {
        /* FIXME: */
        free(job);

        return NULL;
    }

    make_sched_obj(job->schedobj, sched, j, job->resreq);
    job->allocs   = NULL;
    job->accounts = NULL;
    if (j->allocs != NULL) {
        MOVE_PTR(job->allocs, j->allocs);
        job->accounts = make_job_account(job->allocs);
    }

    return job;
}

void kill_sched_queue(void* arg)
{
    if (arg == NULL) {
        return;
    }
}

xce_sched_queue_t* make_sched_queue(void* arg, xce_queue_t* q)
{
    xce_scheduler_t*   sched = (xce_scheduler_t*)arg;
    xce_sched_queue_t* queue = NULL;

    if ((sched == NULL) || (q == NULL)) {
        return NULL;
    }

    queue = (xce_sched_queue_t*)calloc(1, sizeof(*queue));
    if (queue == NULL) {
        return NULL;
    }

    queue->name            = makeString(q->queue);
    queue->attrs           = q->attrs;
    queue->status          = q->status;
    queue->priority        = q->priority;
    queue->accept_interval = q->accept_interval;
    queue->sched_delay     = q->sched_delay;
    /* Req 2 AC 2/3 (scheduler-core spec, task 1.3): xce_queue_t (`q`, the
     * raw config struct) does not carry project_name/job_pick_plugin yet
     * -- calloc() above already leaves both NULL (queue inherits the
     * global picker; project ownership unset). Explicit here for
     * documentation/testability; wiring these from real config is a later
     * task. */
    queue->project_name    = NULL;
    queue->job_pick_plugin = NULL;

    return queue;
}

void kill_sched_worker(void* arg)
{
    if (arg == NULL) {
        return;
    }
    sched_worker_release_host_ctx((xce_sched_worker_t*)arg);
}

void kill_sched_resval(xce_val_t* val)
{
    if (val == NULL) {
        return;
    }

    free_Value(val);
}

xce_sched_worker_t* make_sched_worker(void* arg, xce_worker_t* w)
{
    xce_scheduler_t*    sched  = (xce_scheduler_t*)arg;
    xce_sched_worker_t* worker = NULL;
    int                 i;

    if ((sched == NULL) || (w == NULL)) {
        return NULL;
    }

    worker = (xce_sched_worker_t*)calloc(1, sizeof(*worker));
    if (worker == NULL) {
        return NULL;
    }

    worker->name   = makeString(w->name);
    worker->status = w->status;
    worker->idx    = -1;

    zarray_init(&(worker->resvals), 128,
                (zarray_item_killer_t)kill_sched_resval);
    zarray_init(&(worker->resInUse), 128,
                (zarray_item_killer_t)kill_sched_resval);

    /* fill resource values */
    for (i = 0; i < w->nresvals_to_sched; i++) {
        xce_name_val_t* nv  = w->resvals_to_sched[i];
        xce_resource_t* r   = NULL;
        xce_val_t*      val = NULL;

        if (nv == NULL || nv->name == NULL) {
            continue;
        }

        if (key_find(sched->resmgr, (void*)nv->name, (void**)&r) < 0
            || r == NULL) {
            WARN("%s unknown resource '%s' in worker resvals, skipping",
                 __func__, nv->name);
            continue;
        }

        MOVE_PTR(val, nv->val);
        zarray_put(&(worker->resvals), r->idx, val);
    }

    return worker;
}

void kill_sched_user(void* arg)
{
    xce_sched_user_t* user = (xce_sched_user_t*)arg;

    if (arg == NULL) {
        return;
    }

    dereference(user->name);
    FREEUP(user);
}

xce_sched_user_t* make_sched_user(void* arg, char* name)
{
    xce_scheduler_t*  sched = (xce_scheduler_t*)arg;
    xce_sched_user_t* user  = NULL;

    user = (xce_sched_user_t*)calloc(1, sizeof(*user));
    if (user == NULL) {
        return NULL;
    }

    user->name = makeString(name);

    sched_add_user(sched, user);

    return user;
}

void kill_sched_project(void* arg)
{
    xce_sched_project_t* project = (xce_sched_project_t*)arg;

    if (arg == NULL) {
        return;
    }

    dereference(project->name);
    FREEUP(project);
}

xce_sched_project_t* make_sched_project(void* arg, char* name)
{
    xce_scheduler_t*     sched   = (xce_scheduler_t*)arg;
    xce_sched_project_t* project = NULL;

    project = (xce_sched_project_t*)calloc(1, sizeof(*project));
    if (project == NULL) {
        return NULL;
    }

    if ((name == NULL) || (strlen(name) == 0)) {
        project->name = makeString("default");
    } else {
        project->name = makeString(name);
    }

    /* Req 2 AC (scheduler-core spec, task 1.3): scaffolding only -- no
     * quota/hierarchy inputs exist yet, so priority defaults to 0
     * (calloc() above already zeroes it; explicit here for
     * documentation/testability). */
    project->priority = 0;

    sched_add_project(sched, project);

    return project;
}

/*
 * sched_set_default_job_pick_plugin - Req 2 AC 1 (scheduler-core spec,
 * task 1.3). Applies the "builtin_fcfs" default to sched->job_pick_plugin
 * when it is unset (NULL). Idempotent: a name already present (whether
 * previously defaulted or set by a future config-driven path) is left
 * untouched, so this is safe to call defensively.
 *
 * This function only assigns the *name* -- it never touches the plugin
 * registry (sched->plugins) and never calls plugin_register()/key_find().
 * Resolving the configured name against a registered picker plugin, and
 * failing scheduler startup loudly when the name is not registered
 * (Req 2 AC 5), is task 2.4's job. Kept as a separate, standalone function
 * (rather than inlined into sched_init()) so the default-value behavior is
 * unit-testable without the network/plugin-registry dependencies
 * sched_init() pulls in.
 */
void sched_set_default_job_pick_plugin(xce_scheduler_t* sched)
{
    if (sched == NULL) {
        return;
    }

    if (sched->job_pick_plugin == NULL) {
        sched->job_pick_plugin = makeString("builtin_fcfs");
    }
}

static void host_uncommit_killer(void* obj)
{
    xvec_t* v = (xvec_t*)obj;

    if (obj == NULL) {
        return;
    }

    xvec_fini(v);
    FREEUP(v);
}

int sched_ctx_init(xce_sched_ctx_t* ctx, void* arg, xce_sched_obj_t* obj)
{
    int rc = 0;

    if ((ctx == NULL) || (obj == NULL)) {
        return -EINVAL;
    }

    memset(ctx, 0x0, sizeof(*ctx));

    switch (obj->tag) {
    case XCE_RESREQ_PRIMARY:
        break;
    case XCE_RESREQ_COMPOUND:
        break;
    default:
        ERROR("%s bad sched obj type %d", __func__, obj->tag);
        return -EINVAL;
    }

    ctx->schedobj = obj;
    ctx->sched    = arg;

    ctx->uncomitted.shared = (xvec_t*)calloc(1, sizeof(xvec_t));
    if (ctx->uncomitted.shared == NULL) {
        return -ENOMEM;
    }

    rc = xvec_init(ctx->uncomitted.shared, 1024, sizeof(float));
    if (rc < 0) {
        return rc;
    }

    xvec_idval_init(&(ctx->unmeet), 8);
    xvec_idval_init(&(ctx->alloc_req.shared), 8);
    xvec_idval_init(&(ctx->alloc_req.host), 8);

    zarray_init(&(ctx->uncomitted.hosts), 32,
                (zarray_item_killer_t)host_uncommit_killer);

    return 0;
}

void sched_ctx_fini(xce_sched_ctx_t* ctx)
{
    if (ctx == NULL) {
        return;
    }

    if (ctx->uncomitted.shared != NULL) {
        xvec_fini(ctx->uncomitted.shared);
        FREEUP(ctx->uncomitted.shared);
    }

    zarray_fini(&(ctx->uncomitted.hosts));

    xvec_idval_fini(&(ctx->alloc_req.shared));
    xvec_idval_fini(&(ctx->alloc_req.host));
    xvec_idval_fini(&(ctx->unmeet));
}

void sched_ctx_set_primary(xce_sched_ctx_t* ctx, xce_primary_resreq_t* primary)
{
    ctx->primary = primary;

    plugin_prepare_allocation((xce_scheduler_t*)ctx->sched, ctx);

    return;
}

int sched_ctx_get_uncomit_shared(xvec_t** shared, xce_sched_ctx_t* ctx)
{
    if ((shared == NULL) || (ctx == NULL)) {
        return -EINVAL;
    }

    *shared = ctx->uncomitted.shared;

    return 1;
}

int sched_ctx_get_uncomit_host(xvec_t** host, xce_sched_ctx_t* ctx, int idx)
{
    xvec_t* v;

    if ((host == NULL) || (ctx == NULL) || (idx < 0)) {
        return -EINVAL;
    }

    *host = NULL;
    v     = zarray_get(&(ctx->uncomitted.hosts), idx);
    if (v == NULL) {
        int rc = 0;

        v = (xvec_t*)calloc(1, sizeof(*v));
        if (v == NULL) {
            return -ENOMEM;
        }

        rc = xvec_init(v, 1024, sizeof(float));
        if (rc < 0) {
            xvec_fini(v);
            free(v);

            return rc;
        }

        zarray_put(&(ctx->uncomitted.hosts), idx, v);
    }

    *host = v;
    return 1;
}

int sched_update_existing_queue(xce_scheduler_t* sched, xce_queue_t* updated)
{
    xce_sched_queue_t* q;

    if (updated == NULL) {
        return -EINVAL;
    }

    q = sched_get_queue(sched, updated->queue);
    if (q == NULL) {
        return -ENOENT;
    }

    q->status          = updated->status;
    q->priority        = updated->priority;
    q->accept_interval = updated->accept_interval;
    q->sched_delay     = updated->sched_delay;

    if (updated->resreq != NULL) {
        free(q->resreq);
        q->resreq = makeString(updated->resreq);
    }

    plugin_queue_update(sched, q);

    return 0;
}

/*
 * sched_update_existing_job — apply a notify single-job update to an existing
 * job IN PLACE. The job must already exist (created via the XceData_data_schedjobs
 * dataset path); this function never creates a job.
 *
 *   - NULL payload          → -EINVAL
 *   - unknown jobid         → -ENOENT (no job created)
 *   - otherwise             → update mutable fields (state, priority, queue,
 *                             resource request) and notify plugins; return 0
 *
 * Runtime state (schedobj, allocs, accounts) is preserved — the live
 * xce_sched_job_t object is mutated, not replaced. Mirrors the field-copy +
 * resreq-replace pattern of sched_update_existing_queue().
 */
int sched_update_existing_job(xce_scheduler_t* sched, xce_job_t* updated)
{
    xce_sched_job_t* job;

    if (updated == NULL) {
        return -EINVAL;
    }

    job = sched_get_job(sched, updated->jobid);
    if (job == NULL) {
        return -ENOENT;
    }

    /* state — always carried on xce_job_t */
    job->state = updated->state;

    /* mutable submit-derived fields */
    if (updated->submit != NULL) {
        /* queue — re-resolve the queue pointer by name. xce_job_t carries no
         * per-job priority, so job priority follows the (possibly new) queue,
         * consistent with the queue-derived priority model. */
        if (updated->submit->queue != NULL) {
            xce_sched_queue_t* q =
                sched_get_queue(sched, updated->submit->queue);
            if (q != NULL) {
                job->queue    = q;
                job->priority = q->priority;
            }
        }

        /* resource request — re-parse and swap; free the old parsed container.
         * Mirrors sched_update_existing_queue's free-old/replace pattern. The
         * inner select tree is intern/GC-owned, consistent with make_sched_job. */
        if (updated->submit->resreq != NULL) {
            xce_resreq_t* newreq =
                (xce_resreq_t*)calloc(1, sizeof(*newreq));
            if (newreq == NULL) {
                return -ENOMEM;
            }

            char* errmsg = NULL;
            int   rc = parse_resreq(newreq, updated->submit->resreq,
                                    sched->resmgr, &errmsg);
            if (rc < 0) {
                ERROR("%s parse_resreq failed: %s", __func__,
                      errmsg ? errmsg : "(null)");
                FREEUP(errmsg);
                free(newreq);
                return -EINVAL;
            }

            free(job->resreq);
            job->resreq = newreq;
        }
    }

    plugin_job_update(sched, job);

    return 0;
}

/*
 * sched_apply_job_rusage — store a controller rusage report ON the matching
 * job (Req 6, UB-6.1/6.2/6.3; bug-765 G-5). On success the job TAKES OWNERSHIP
 * of `rusages` (the caller must not free it); the previous report, if any, is
 * released. Returns:
 *   -EINVAL  rusages is NULL
 *   -ENOENT  no job with rusages->jobid (no state modified, ownership retained
 *            by the caller so it can free the payload)
 *        0  stored on the job and plugins notified
 */
int sched_apply_job_rusage(xce_scheduler_t* sched, xce_job_rusages_t* rusages)
{
    xce_sched_job_t* job;

    if (rusages == NULL) {
        return -EINVAL;
    }

    job = sched_get_job(sched, rusages->jobid);
    if (job == NULL) {
        return -ENOENT;
    }

    /* replace any prior report; the job now owns this one. free_JobRusages_ptr
     * is NULL-safe and clears the field before we reassign. */
    if (job->rusages != rusages) {
        free_JobRusages_ptr(&job->rusages);
    }
    job->rusages = rusages;

    plugin_job_rusage_update(sched, job, rusages);

    return 0;
}