myxce/src/lib/xceapi.c

#include <errno.h>
#include <time.h>

#include "mylock.h"
#include "utils.h"
#include "xceapi.h"

/* All synchronous API calls wait at most 10 seconds for a reply. */
static void cbuf_deadline_10s(struct timespec* ts)
{
    clock_gettime(CLOCK_REALTIME, ts);
    ts->tv_sec += 10;
}

int xce_hello(xce_ctx_t* ctx, char* msg)
{
    int rc, status;

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

    if (ctx->state != XCE_CTX_RUN) {
        return -EINVAL;
    }

    rc = xce_hello_req(ctx->ep, msg);
    if (rc < 0) {
        return rc;
    }
    struct timespec _dl; cbuf_deadline_10s(&_dl);
    rc = cbuf_timedget(&(ctx->wbuf), &status, &_dl);
    if (rc < 0) {
        return rc;
    }

    return status;
}

int xce_nop(xce_ctx_t* ctx)
{
    int rc;

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

    if (ctx->state != XCE_CTX_RUN) {
        return -EINVAL;
    }

    rc = xce_nop_req(ctx->ep);

    /* no need to wait reply, just as heartbeat */
    return rc;
}

int xce_heartbeat(xce_ctx_t* ctx, int hb_interval)
{
    time_t now;

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

    if (ctx->state != XCE_CTX_RUN) {
        return -EINVAL;
    }

    if (ctx->ep == NULL) {
        return -EINVAL;
    }

    now = time(NULL);
    if (now - ctx->ep->ts_last_out < hb_interval) {
        return 0;
    }

    return xce_nop(ctx);
}

int xce_shutdown(xce_ctx_t* ctx)
{
    int rc;

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

    if (ctx->state != XCE_CTX_RUN) {
        return -EINVAL;
    }

    rc = xce_shutdown_req(ctx->ep);

    /* do not need wait controller back as it will be shutdown */
    return rc;
}

int xce_get_with_condition(xce_ctx_t* ctx, xce_get_t* get, void*** data)
{
    int       rc;
    int       status;
    dataset_t getreply;

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

    if (ctx->state != XCE_CTX_RUN) {
        return -EINVAL;
    }

    if ((get == NULL) || (data == NULL)) {
        return -EINVAL;
    }

    memset(&getreply, 0x0, sizeof(getreply));

    ctx->reply.u.get = &getreply;
    rc               = xce_get_req(ctx->ep, get);
    if (rc < 0) {
        goto done;
    }

    struct timespec _dl; cbuf_deadline_10s(&_dl);
    rc = cbuf_timedget(&(ctx->wbuf), &status, &_dl);
    if (rc < 0) {
        goto done;
    }

    if (status >= 0) {
        *data = getreply.rows;
        rc    = getreply.nrows;
    } else {
        rc = status;
    }

done:
    /* Always clear the destination — including on the xce_get_req/cbuf_get
     * failover error paths — so the single, untagged reply union never retains
     * the address of this now-returning stack `getreply`. A later ctrlstatus
     * reply would otherwise free that dead-stack pointer as a CtrlStatus*
     * (bug-1077, ASan stack-use-after-return). */
    ctx->reply.u.get = NULL;

    return rc;
}

int xce_get(xce_ctx_t* ctx, int table, void*** data)
{
    /* clang-format off */
    xce_match_t condition = {
        .type = MatchCondition_condition_matchall
    };
    xce_get_t get = {
        .table     = table,
        .condition = &condition
    };
    /* clang-format on */

    return xce_get_with_condition(ctx, &get, data);
}

int xce_reslist(xce_ctx_t* ctx, xce_resource_t*** res)
{
    int table = XceData_data_resources;

    return xce_get(ctx, table, (void***)res);
}

int xce_resval(xce_ctx_t* ctx, char* host, char* res, xce_resval_t*** vals)
{
    int rc;
    int status;
    /* clang-format off */
    xce_reskey_t reskey = {
        .resname = res,
        .hostname = host
    };
    xce_match_t condition = {
        .type = MatchCondition_condition_matchreskey,
        .condition = {
            .reskey = &reskey
        }
    };
    xce_get_t get = {
        .table = XceData_data_resvals,
        .condition = &condition
    };
    dataset_t getreply = {
        .table = XceData_data_resvals,
        .nrows = 0,
        .rows = NULL
    };
    /* clang-format on */

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

    if (ctx->state != XCE_CTX_RUN) {
        return -EINVAL;
    }

    ctx->reply.u.get = &getreply;
    rc               = xce_get_req(ctx->ep, &get);
    if (rc < 0) {
        goto done;
    }

    struct timespec _dl; cbuf_deadline_10s(&_dl);
    rc = cbuf_timedget(&(ctx->wbuf), &status, &_dl);
    if (rc < 0) {
        goto done;
    }

    if (status >= 0) {
        *vals = (xce_resval_t**)getreply.rows;
        rc    = getreply.nrows;
    } else {
        rc = status;
    }

done:
    /* Clear on every path (incl. the failover error returns) so the shared
     * reply union never keeps this stack `getreply` address — bug-1077. */
    ctx->reply.u.get = NULL;

    return rc;
}

int xce_join(xce_ctx_t* ctx, int role, hostattrs_t* attrs)
{
    int rc, status;

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

    if (ctx->state != XCE_CTX_RUN) {
        return -EINVAL;
    }

    rc = xce_join_req(ctx->ep, role, attrs);
    if (rc < 0) {
        return rc;
    }
    struct timespec _dl; cbuf_deadline_10s(&_dl);
    rc = cbuf_timedget(&(ctx->wbuf), &status, &_dl);
    if (rc < 0) {
        return rc;
    }

    return status;
}

int xce_workerlist(xce_ctx_t* ctx, xce_worker_t*** workers)
{
    int table = XceData_data_workers;

    return xce_get(ctx, table, (void***)workers);
}

int xce_free_workers(int nworkers, xce_worker_t* workers) { return 0; }

int xce_jobdispatch(xce_ctx_t* ctx, xce_job_dispatch_t* info)
{
    int rc, status;

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

    if (ctx->state != XCE_CTX_RUN) {
        return -EINVAL;
    }

    rc = xce_jobdispatch_req(ctx->ep, info);
    if (rc < 0) {
        return rc;
    }

    struct timespec _dl; cbuf_deadline_10s(&_dl);
    rc = cbuf_timedget(&(ctx->wbuf), &status, &_dl);
    if (rc < 0) {
        return rc;
    }

    return status;
}

int xce_notify(xce_ctx_t* ctx, xce_notify_data_t* data)
{
    int rc;

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

    if (ctx->state != XCE_CTX_RUN) {
        return -EINVAL;
    }

    rc = xce_notify_req(ctx->ep, data);

    return rc;
}

int xce_jobsubmit(xce_ctx_t* ctx, xce_job_submit_t* submit, uint64_t* jobid)
{
    int rc, status;

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

    if (ctx->state != XCE_CTX_RUN) {
        return -EINVAL;
    }

    rc = xce_jobsubmit_req(ctx->ep, submit);
    if (rc < 0) {
        return rc;
    }

    struct timespec _dl; cbuf_deadline_10s(&_dl);
    rc = cbuf_timedget(&(ctx->wbuf), &status, &_dl);
    if (rc < 0) {
        return rc;
    }

    if (status >= 0) {
        *jobid = ctx->reply.u.jobid;
    }
    return status;
}

int xce_joblist_ex(xce_ctx_t* ctx, int sched, xce_job_t*** jobs)
{
    int table;

    if (sched) {
        table = XceData_data_schedjobs;
    } else {
        table = XceData_data_jobs;
    }

    return xce_get(ctx, table, (void***)jobs);
}

int xce_sched_joblist(xce_ctx_t* ctx, xce_job_t*** jobs)
{
    return xce_joblist_ex(ctx, 1, jobs);
}

int xce_joblist(xce_ctx_t* ctx, xce_job_t*** jobs)
{
    return xce_joblist_ex(ctx, 0, jobs);
}

int xce_varlist(xce_ctx_t* ctx, xce_config_t*** vars)
{
#if 0
    int            table = XCETables_conf;

    return xce_get(ctx, table, (void***)vars);
#endif
    return 0;
}

int xce_queuelist(xce_ctx_t* ctx, xce_queue_t*** queues)
{
#if 0
    int            table = XCETables_queues;

    return xce_get(ctx, table, (void***)queues);
#else
    return 0;
#endif
}

int xce_free_queues(int nqueues, xce_queue_t* queues)
{
    int i;

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

    for (i = 0; i < nqueues; i++) {
        FREEUP(queues[i].resreq);
    }
    return 0;
}

int xce_group_configlist(xce_ctx_t* ctx, int group_type,
                         xce_group_config_t*** groups)
{
#if 0
    int            table;

    switch (group_type) {
    case 0:
        table = XCETables_ugroups;
        break;
    default:
        table = XCETables_hgroups;
        break;
    }

    return xce_get(ctx, table, (void***)groups);
#else
    return 0;
#endif
}

int xce_free_groups(int ngroups, xce_group_config_t* groups)
{
    /* TODO */
    return 0;
}

int xce_clusterconf(xce_ctx_t* ctx, xce_clusterconf_t** conf)
{
    int rc, status;

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

    if (ctx->state != XCE_CTX_RUN) {
        return -EINVAL;
    }

    rc = xce_clusterconf_req(ctx->ep);
    if (rc < 0) {
        return rc;
    }

    struct timespec _dl; cbuf_deadline_10s(&_dl);
    rc = cbuf_timedget(&(ctx->wbuf), &status, &_dl);
    if (rc < 0) {
        return rc;
    }

    if (status < 0) {
        return status;
    }

    MOVE_PTR((*conf), ctx->reply.u.clusterconf);
    return 0;
}

int xce_ctrlstatus(xce_ctx_t* ctx, xce_ctrlstatus_t** status)
{
    int rc, st;

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

    if (ctx->state != XCE_CTX_RUN) {
        return -EINVAL;
    }

    rc = xce_ctrlstatus_req(ctx->ep);
    if (rc < 0) {
        return rc;
    }

    rc = cbuf_get(&(ctx->wbuf), &st);
    if (rc < 0) {
        return rc;
    }

    if (st < 0) {
        return st;
    }

    MOVE_PTR((*status), ctx->reply.u.ctrlstatus);
    return 0;
}

int xce_update_resvals(xce_ctx_t* ctx, int count, xce_resval_t** vals)
{
    int rc;
    int status;

    /* clang-format off */
    xce_resval_list_t valist = {
        .count = count,
        .vals  = vals
    };
    xce_update_data_t data = {
        .table = Update_data_resvals,
        .data  = {
             .resvals = &valist
	}
    };
    /* clang-format on */

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

    if (ctx->state != XCE_CTX_RUN) {
        return -EINVAL;
    }

    rc = xce_update_req(ctx->ep, &data);
    if (rc < 0) {
        return rc;
    }

    struct timespec _dl; cbuf_deadline_10s(&_dl);
    rc = cbuf_timedget(&(ctx->wbuf), &status, &_dl);
    if (rc < 0) {
        return rc;
    }

    return status;
}

int xce_update_jusage(xce_ctx_t* ctx, uint64_t jobid, int npids, int* pids,
                      int ngids, int* pgids, int nusage, xce_rusage_t** usages)
{
    int rc;
    int status;

    /* clang-format off */
    xce_job_rusages_t rusages = {
        .jobid  = jobid,
        .npids  = npids,
        .pids   = pids,
        .pgids  = pgids,
        .count  = nusage,
        .usages = usages
    };
    xce_update_data_t data = {
        .table = Update_data_jobrusages,
        .data  = {
             .jobrusages = &rusages }
    };
    /* clang-format on */

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

    if (ctx->state != XCE_CTX_RUN) {
        return -EINVAL;
    }

    rc = xce_update_req(ctx->ep, &data);
    if (rc < 0) {
        return rc;
    }

    struct timespec _dl; cbuf_deadline_10s(&_dl);
    rc = cbuf_timedget(&(ctx->wbuf), &status, &_dl);
    if (rc < 0) {
        return rc;
    }

    return status;
}

int xce_jobctrl(xce_ctx_t* ctx, xce_jobctrl_t* ctrl)
{
    int rc;
    int status;

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

    if (ctx->state != XCE_CTX_RUN) {
        return -EINVAL;
    }

    rc = xce_jobctrl_req(ctx->ep, ctrl);
    if (rc < 0) {
        return rc;
    }

    struct timespec _dl; cbuf_deadline_10s(&_dl);
    rc = cbuf_timedget(&(ctx->wbuf), &status, &_dl);
    if (rc < 0) {
        return rc;
    }

    return status;
}

int xce_queuectrl(xce_ctx_t* ctx, xce_queuectrl_t* ctrl)
{
    int rc;
    int status;

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

    if (ctx->state != XCE_CTX_RUN) {
        return -EINVAL;
    }

    rc = xce_queuectrl_req(ctx->ep, ctrl);
    if (rc < 0) {
        return rc;
    }

    struct timespec _dl; cbuf_deadline_10s(&_dl);
    rc = cbuf_timedget(&(ctx->wbuf), &status, &_dl);
    if (rc < 0) {
        return rc;
    }

    return status;
}

int xce_hostctrl(xce_ctx_t* ctx, xce_hostctrl_t* ctrl)
{
    int rc;
    int status;

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

    if (ctx->state != XCE_CTX_RUN) {
        return -EINVAL;
    }

    rc = xce_hostctrl_req(ctx->ep, ctrl);
    if (rc < 0) {
        return rc;
    }

    struct timespec _dl; cbuf_deadline_10s(&_dl);
    rc = cbuf_timedget(&(ctx->wbuf), &status, &_dl);
    if (rc < 0) {
        return rc;
    }

    return status;
}

int xce_upgradeconfig(xce_ctx_t* ctx, const uint8_t *sha256,
                      size_t sha256_len, const uint8_t *bytes,
                      size_t bytes_len)
{
    int rc;
    int status;

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

    if (ctx->state != XCE_CTX_RUN) {
        return -EINVAL;
    }

    rc = xce_upgradeconfig_req_raw(ctx->ep, sha256, sha256_len, bytes,
                                   bytes_len);
    if (rc < 0) {
        return rc;
    }

    struct timespec _dl; cbuf_deadline_10s(&_dl);
    rc = cbuf_timedget(&(ctx->wbuf), &status, &_dl);
    if (rc < 0) {
        return rc;
    }

    return status;
}

int xce_runtask(xce_ctx_t* ctx, xce_taskrun_t* runinfo)
{
    int rc;
    int status;

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

    if (ctx->state != XCE_CTX_RUN) {
        return -EINVAL;
    }

    rc = xce_runtask_req(ctx->ep, runinfo);
    if (rc < 0) {
        return rc;
    }

    struct timespec _dl; cbuf_deadline_10s(&_dl);
    rc = cbuf_timedget(&(ctx->wbuf), &status, &_dl);
    if (rc < 0) {
        return rc;
    }

    return status;
}

int xce_set_pend_reason(xce_ctx_t* ctx, xce_preason_set_t* reasons)
{
    int rc;
    int status;

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

    if (ctx->state != XCE_CTX_RUN) {
        return -EINVAL;
    }

    rc = xce_set_pend_reason_req(ctx->ep, reasons);
    if (rc < 0) {
        return rc;
    }

    struct timespec _dl; cbuf_deadline_10s(&_dl);
    rc = cbuf_timedget(&(ctx->wbuf), &status, &_dl);
    if (rc < 0) {
        return rc;
    }

    return status;
}

int xce_clear_reason(xce_ctx_t* ctx, xce_clear_reason_t* reasons)
{
    int rc;
    int status;

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

    if (ctx->state != XCE_CTX_RUN) {
        return -EINVAL;
    }

    rc = xce_clear_reason_req(ctx->ep, reasons);
    if (rc < 0) {
        return rc;
    }

    struct timespec _dl; cbuf_deadline_10s(&_dl);
    rc = cbuf_timedget(&(ctx->wbuf), &status, &_dl);
    if (rc < 0) {
        return rc;
    }

    return status;
}

int xce_jobinfo(xce_ctx_t* ctx, uint64_t jobid, xce_job_t** job)
{
    /* clang-format off */
    xce_match_t condition = {
        .type = MatchCondition_condition_matchjid,
        .condition = {
            .jobid = jobid
        }
    };
    xce_get_t get = {
        .table = XceData_data_jobs,
        .condition = &condition
    };
    /* clang-format on */
    int         rc;
    xce_job_t** jobs;

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

    rc = xce_get_with_condition(ctx, &get, (void***)&jobs);

    if (rc <= 0) {
        return rc;
    }

    *job = jobs[0];
    FREEUP(jobs);

    return 1;
}

int xce_fetchfile(xce_ctx_t* ctx, xce_fetchfile_t* msg, xce_fetchfile_reply_t** reply)
{
    int rc;
    int status;

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

    if (ctx->state != XCE_CTX_RUN) {
        return -EINVAL;
    }

    ctx->reply.u.fetchfilereply = NULL;

    rc = xce_fetchfile_req(ctx->ep, msg);
    if (rc < 0) {
        return rc;
    }

    struct timespec _dl; cbuf_deadline_10s(&_dl);
    rc = cbuf_timedget(&(ctx->wbuf), &status, &_dl);
    if (rc < 0) {
        return rc;
    }

    if (status < 0) {
        return status;
    }

    MOVE_PTR(*reply, ctx->reply.u.fetchfilereply);
    return 0;
}