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;
    xlist_t             rlist;

    init_reply_by_req(&reply, request);
    xlist_init(&rlist, 512);

    collect_resources(c, NULL, &rlist);
    resdata.n_total   = rlist.count;
    resdata.resources = (xce_resource_t**)rlist.data;

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

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

    xlist_fini(&rlist);

    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;
    xlist_t wlist;

    init_reply_by_req(&reply, request);
    xlist_init(&wlist, 1024);

    collect_workers(c, (collect_check_fn_t)check_worker_, &wlist);
    workerdata.n_total = wlist.count;
    workerdata.workers = (xce_worker_t**)wlist.data;

    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_name_val_list(&(w->resvals_to_sched), &(w->nresvals_to_sched));
    }

    unlock_collect_workers(&wlist);

    xlist_fini(&wlist);

    return rc;
}

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;
    xlist_t         jlist;

    init_reply_by_req(&reply, request);

    xlist_init(&jlist, 1024 * 1024);

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

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

        lock(c->jobmgr, RTTI_LOCK_READ);
        rc = key_find(c->jobmgr, (void*)match->condition.jobid, (void**)&j);
        if (j != NULL) {
            reference(j);
        }
        unlock(c->jobmgr);

        if (rc > 0) {
            lock(j, RTTI_LOCK_READ);
            xlist_store(&jlist, &(j->job));
        } else {
            dereference(j);
        }
    } else {
        collect_jobs(c, NULL, &jlist);
    }

    jobdata.n_total = jlist.count;
    jobdata.jobs    = (xce_job_t**)jlist.data;

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

    unlock_collect_jobs(&jlist);

    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 XceData_data_resources:
        rc = do_reply_reslist(c, ep, request);
        break;
    case XceData_data_workers:
    case XceData_data_schedworkers:
        rc = do_reply_workerlist(c, ep, request);
        break;
    case XceData_data_schedjobs:
    case XceData_data_jobs:
        rc = do_reply_joblist(c, ep, request);
        break;
    case XceData_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;
}