myxce/src/controller/opget.c

#include "config.h"

#include <errno.h>
#include <string.h>
#include <unistd.h>

#include "cluster.h"
#include "controller.capnp.h"
#include "utils.h"
#include "xce/cluster_def.h"

static int do_reply_reslist(cluster_t* c, endpoint_t* ep,
                            xce_request_t* request)
{
    xce_reply_t         reply;
    xce_resource_data_t resdata;
    xce_data_t          getdata = { .table = request->body.getmsg->table,
                                    .data  = { .resources = &resdata } };
    int                 rc      = 0;

    init_reply_by_req(&reply, request);
    collect_resources(c, NULL, &(resdata.n_total), &(resdata.resources));

    reply.status       = 0;
    reply.data_which   = RequestReply_data_getdata;
    reply.data.getdata = &getdata;

    rc = xce_reply_get(ep, c->myhost, &reply, NULL, NULL, -1);

    deref_objs(resdata.n_total, (void**)resdata.resources);
    FREEUP(resdata.resources);

    return rc;
}

static int do_reply_resvals(cluster_t* c, endpoint_t* ep,
                            xce_request_t* request)
{
    xce_reply_t       reply;
    xce_resval_data_t resvaldata;
    xce_get_t*        getmsg  = request->body.getmsg;
    xce_data_t        getdata = { .data = { .resvals = &resvaldata } };
    int               rc      = 0;
    xce_val_t         val;
    xce_resval_t      resval;
    xce_resval_t*     vals[] = { &resval };

    init_reply_by_req(&reply, request);

    resvaldata.n_total = 0;
    resvaldata.vals    = NULL;

    if (getmsg != NULL) {
        xce_match_t* condition = getmsg->condition;

        if (condition != NULL) {
            getdata.table = getmsg->table;
            rc            = get_resval(c, condition->condition.reskey->resname,
                                       condition->condition.reskey->hostname, &val);
            if (rc >= 0) {
                resval.val         = &val;
                resval.name        = condition->condition.reskey->resname;
                resval.host        = condition->condition.reskey->hostname;
                resvaldata.n_total = 1;
                resvaldata.vals    = vals;
            }
        } else {
            rc = -EINVAL;
        }
    } else {
        rc = -EINVAL;
    }

    reply.status       = rc;
    reply.data_which   = RequestReply_data_getdata;
    reply.data.getdata = &getdata;

    rc = xce_reply_get(ep, c->myhost, &reply, NULL, NULL, -1);

    return rc;
}

int check_worker_(xce_worker_t* w)
{
    if (w == NULL) {
        return 0;
    }

    return 1;
}

static int do_reply_workerlist(cluster_t* c, endpoint_t* ep,
                               xce_request_t* request)
{
    xce_reply_t        reply;
    xce_workers_data_t workerdata;
    /* clang-format off */
    xce_data_t         getdata = {
                .table = request->body.getmsg->table,
                .data  = {
                     .workers = &workerdata }
    };
    /* clang-format on */
    int rc = 0;
    int i;

    init_reply_by_req(&reply, request);
    collect_workers(c, (collect_check_fn_t)check_worker_, &(workerdata.n_total),
                    &(workerdata.workers));

    reply.status       = 0;
    reply.data_which   = RequestReply_data_getdata;
    reply.data.getdata = &getdata;

    rc = xce_reply_get(ep, c->myhost, &reply, check_controller, c, 100);

    for (i = 0; i < workerdata.n_total; i++) {
        xce_worker_t* w = workerdata.workers[i];

        free_resval_with_idx_list(&(w->resvals_to_sched), w->nresvals_to_sched);
    }

    deref_objs(workerdata.n_total, (void**)workerdata.workers);
    FREEUP(workerdata.workers);

    return rc;
}

static int check_job_(xce_job_t* j)
{
    if (j == NULL) {
        return 0;
    }

    return 1;
}

static int do_reply_joblist(cluster_t* c, endpoint_t* ep,
                            xce_request_t* request)
{
    xce_reply_t     reply;
    xce_jobs_data_t jobdata;
    xce_data_t      getdata = { .table = request->body.getmsg->table,
                                .data  = { .jobs = &jobdata } };
    int             rc      = 0;
    xce_match_t*    match   = request->body.getmsg->condition;

    init_reply_by_req(&reply, request);

    if (match->type == MatchCondition_condition_matchjid) {
        job_t* j;

        jobdata.n_total = 0;
        jobdata.jobs    = NULL;

        j = jobmgr_lockjob(&(c->jobmgr), match->condition.jobid, OBJREF_READ);
        if (j != NULL) {
            jobdata.jobs = (xce_job_t**)calloc(1, sizeof(xce_job_t*));
            if (jobdata.jobs != NULL) {
                jobdata.jobs[0] = &(j->job);
                jobdata.n_total = 1;
            }
        }
    } else {
        collect_jobs(c, (collect_check_fn_t)check_job_, &(jobdata.n_total),
                     &(jobdata.jobs));
    }

    reply.status       = 0;
    reply.data_which   = RequestReply_data_getdata;
    reply.data.getdata = &getdata;

    rc = xce_reply_get(ep, c->myhost, &reply, check_controller, c, 100);

    deref_objs(jobdata.n_total, (void**)jobdata.jobs);
    FREEUP(jobdata.jobs);

    return rc;
}

int do_get_op(cluster_t* c, endpoint_t* ep, xce_request_t* request)
{
    int         rc = 0;
    xce_reply_t reply;

    TRACE("%s table %d", __func__, request->body.getmsg->table);

    init_reply_by_req(&reply, request);

    switch (request->body.getmsg->table) {
    case Data_data_resources:
        rc = do_reply_reslist(c, ep, request);
        break;
    case Data_data_workers:
    case Data_data_schedworkers:
        rc = do_reply_workerlist(c, ep, request);
        break;
    case Data_data_schedjobs:
    case Data_data_jobs:
        rc = do_reply_joblist(c, ep, request);
        break;
    case Data_data_resvals:
        rc = do_reply_resvals(c, ep, request);
        break;
    default:
        rc = -ENOENT;

        reply.status     = rc;
        reply.data_which = RequestReply_data_nodata;

        xce_reply(ep, c->myhost, &reply);
        break;
    }

    return rc;
}