myxce/src/controller/dataex.c

#include <errno.h>

#include "cluster.h"
#include "config_repl.h"
#include "request.h"
#include "xce/cluster_def.h"
#include "xce/types.h"
#include "xlist.h"

typedef int (*check_fn_t)(void*);

int check_controller(void* arg)
{
    cluster_t* c = (cluster_t*)arg;

    if (!c->running) {
        return -EINVAL;
    }

    return 1;
}

int collect_resval(var resmgr, zarray_t* array, xlist_t* l)
{
    iterator_t it;
    int        rc;

    lock(resmgr, RTTI_LOCK_READ);
    rc = iter_make(&it, resmgr);
    if (rc < 0) {
        unlock(resmgr);
        return rc;
    }

    for (;;) {
        xce_resource_t* r   = NULL;
        xce_val_t*      src = NULL;
        xce_name_val_t* nv  = NULL;

        rc = iter_next(&it, (void**)&r);
        if ((rc < 0) || (rc == RTTI_ITER_NO_DATA)) {
            break;
        }

        if (r == NULL) {
            continue;
        }

        src = (xce_val_t*)zarray_get(array, r->idx);
        if (src == NULL) {
            continue;
        }

        nv = (xce_name_val_t*)calloc(1, sizeof(*nv));
        if (nv == NULL) {
            rc = -ENOMEM;
            break;
        }

        nv->name = strdup(r->name);
        nv->val  = (xce_val_t*)calloc(1, sizeof(xce_val_t));
        if (nv->val == NULL) {
            FREEUP(nv->name);
            FREEUP(nv);
            rc = -ENOMEM;
            break;
        }

        if (src->which == Value_u_svalue) {
            nv->val->which     = Value_u_svalue;
            nv->val->u.svalue  = strdup(src->u.svalue ? src->u.svalue : "-");
        } else {
            memcpy(nv->val, src, sizeof(xce_val_t));
        }

        /* bug-1568 instrumentation: the actual value read out of `array`
         * (w->host.resvals for a per-worker call, c->sharedinst for a
         * shared-resource call) right here -- the true "about to be
         * wire-packaged" point shared by BOTH push paths
         * (xcesched_push_workers -> collect_resvals_workers -> this
         * function, and sched_worker_change -> this function directly).
         * If this shows a real non-zero fvalue but the scheduler's own
         * core_build_worker TRACE (core_events.c) shows 0 for the same
         * resource, the loss is in the wire encode/decode or the
         * scheduler-side resolution, not here or upstream. */
        if (src->which == Value_u_fvalue) {
            TRACE("%s resource '%s' idx %d -> fvalue=%f", __func__, r->name,
                  r->idx, src->u.fvalue);
        }

        xlist_store(l, nv);
    }

    iter_destroy(&it);
    unlock(resmgr);

    return (rc == RTTI_ITER_NO_DATA || rc == 0) ? l->count : rc;
}

static void lock_objs(xlist_t* l, int op)
{
    int i;

    for (i = 0; i < l->count; i++) {
        var obj = xlist_get(l, i);

        if (obj == NULL) {
            continue;
        }

        lock(obj, op);
    }
}

static void lock_collect_queues(xlist_t* l, int op)
{
    int i;

    for (i = 0; i < l->count; i++) {
        xce_queue_t* q = (xce_queue_t*)xlist_get(l, i);
        queue_t*     q1;

        if (q == NULL) {
            continue;
        }

        q1 = CONTAINER_OF(q, queue_t, queue);
        lock(q1, op);
    }
}

static void lock_collect_workers(xlist_t* l, int op)
{
    int i;

    for (i = 0; i < l->count; i++) {
        xce_worker_t* w = (xce_worker_t*)xlist_get(l, i);
        worker_t*     w1;

        if (w == NULL) {
            continue;
        }

        w1 = CONTAINER_OF(w, worker_t, host);
        lock(w1, op);
    }
}

static void lock_collect_jobs(xlist_t* l, int op)
{
    int i;

    for (i = 0; i < l->count; i++) {
        xce_job_t* j = (xce_job_t*)xlist_get(l, i);
        job_t*     j1;

        if (j == NULL) {
            continue;
        }

        j1 = CONTAINER_OF(j, job_t, job);
        lock(j1, op);
    }
}

static void collect_resvals_workers(var resmgr, xlist_t* l)
{
    int i;

    for (i = 0; i < l->count; i++) {
        xlist_t       lst;
        xce_worker_t* w = xlist_get(l, i);
        void*         p = NULL;

        if (w == NULL) {
            continue;
        }

        xlist_init(&lst, 128);

        collect_resval(resmgr, &(w->resvals), &lst);

        w->nresvals_to_sched = lst.count;
        MOVE_PTR(p, lst.data);
        w->resvals_to_sched = (xce_name_val_t**)p;

        xlist_fini(&lst);
    }
}

void collect_confvars(cluster_t* c, collect_check_fn_t fn, xlist_t* l)
{
    iterator_t it;
    int        rc;

    lock(c->configmgr, RTTI_LOCK_READ);
    rc = iter_make(&it, c->configmgr);
    if (rc < 0) {
        unlock(c->configmgr);

        return;
    }

    for (;;) {
        xce_config_t* config = NULL;

        rc = iter_next(&it, (void**)&config);
        if ((rc < 0) || (rc == RTTI_ITER_NO_DATA)) {
            break;
        }

        if (config == NULL) {
            continue;
        }

        if ((fn == NULL) || (fn(config))) {
            reference(config);
            xlist_store(l, config);
        }
    }

    iter_destroy(&it);
    unlock(c->configmgr);

    lock_objs(l, RTTI_LOCK_READ);

    return;
}

void collect_queues(cluster_t* c, collect_check_fn_t fn, xlist_t* l)
{
    iterator_t it;
    int        rc;

    lock(c->qmgr, RTTI_LOCK_READ);
    rc = iter_make(&it, c->qmgr);
    if (rc < 0) {
        unlock(c->qmgr);

        return;
    }

    for (;;) {
        queue_t* q = NULL;

        rc = iter_next(&it, (void**)&q);
        if ((rc < 0) || (rc == RTTI_ITER_NO_DATA)) {
            break;
        }

        if (q == NULL) {
            continue;
        }
        if (q->queue.queue != NULL
                && strcmp(q->queue.queue, XCE_LOST_AND_FOUND_NAME) == 0) {
            continue;
        }
        if ((fn == NULL) || fn(q)) {
            reference(q);
            xlist_store(l, &(q->queue));
        }
    }

    iter_destroy(&it);
    unlock(c->qmgr);

    lock_collect_queues(l, RTTI_LOCK_READ);
}

void collect_resources(cluster_t* c, collect_check_fn_t fn, xlist_t* l)
{
    iterator_t it;
    int        rc;

    lock(c->resmgr, RTTI_LOCK_READ);
    rc = iter_make(&it, c->resmgr);
    if (rc < 0) {
        unlock(c->resmgr);

        return;
    }

    for (;;) {
        xce_resource_t* r = NULL;

        rc = iter_next(&it, (void**)&r);
        if ((rc < 0) || (rc == RTTI_ITER_NO_DATA)) {
            break;
        }

        if (r == NULL) {
            continue;
        }

        if ((fn == NULL) || (fn(r))) {
            reference(r);
            xlist_store(l, r);
        }
    }

    iter_destroy(&it);
    unlock(c->resmgr);

    lock_objs(l, RTTI_LOCK_READ);

    return;
}

void collect_jobs(cluster_t* c, collect_check_fn_t fn, xlist_t* l)
{
    iterator_t it;
    int        rc = 0;

    lock(c->jobmgr, RTTI_LOCK_READ);
    rc = iter_make(&it, c->jobmgr);
    if (rc < 0) {
        unlock(c->jobmgr);

        return;
    }

    for (;;) {
        job_t* j = NULL;

        rc = iter_next(&it, (void**)&j);
        if ((rc < 0) || (rc == RTTI_ITER_NO_DATA)) {
            break;
        }

        if (j == NULL) {
            continue;
        }

        if ((fn == NULL) || (fn(j))) {
            reference(j);
            xlist_store(l, &(j->job));
        }
    }

    iter_destroy(&it);
    unlock(c->jobmgr);

    lock_collect_jobs(l, RTTI_LOCK_READ);

    return;
}

void collect_workers(cluster_t* c, collect_check_fn_t fn, xlist_t* l)
{
    iterator_t it;
    int        rc = 0;

    lock(c->workermgr, RTTI_LOCK_READ);
    rc = iter_make(&it, c->workermgr);
    if (rc < 0) {
        unlock(c->workermgr);

        return;
    }

    for (;;) {
        worker_t* w = NULL;

        rc = iter_next(&it, (void**)&w);
        if ((rc < 0) || (rc == RTTI_ITER_NO_DATA)) {
            break;
        }

        if (w == NULL) {
            continue;
        }

        if (w->host.name != NULL
                && strcmp(w->host.name, XCE_LOST_AND_FOUND_NAME) == 0) {
            continue;
        }

        if ((fn == NULL) || (fn(w))) {
            reference(w);
            xlist_store(l, &(w->host));
        }
    }

    iter_destroy(&it);
    unlock(c->workermgr);

    lock_collect_workers(l, RTTI_LOCK_WRITE);

    collect_resvals_workers(c->resmgr, l);

    return;
}

void collect_shared_resvals(cluster_t* c, collect_check_fn_t fn, xlist_t* l)
{
    collect_resval(c->resmgr, &(c->sharedinst), l);
}

int notify(cluster_t* c, endpoint_t* ep, xce_notify_data_t* data)
{
    return xce_notify_req(ep, data);
}

void unlock_collect_queues(xlist_t* l)
{
    int i;

    for (i = 0; i < l->count; i++) {
        xce_queue_t* q = (xce_queue_t*)xlist_get(l, i);
        queue_t*     q1;

        if (q == NULL) {
            continue;
        }

        q1 = CONTAINER_OF(q, queue_t, queue);
        unlock(q1);

        dereference(q1);
    }
}

void unlock_collect_workers(xlist_t* l)
{
    int i;

    for (i = 0; i < l->count; i++) {
        xce_worker_t* w = (xce_worker_t*)xlist_get(l, i);
        worker_t*     w1;

        if (w == NULL) {
            continue;
        }

        free_name_val_list(&(w->resvals_to_sched), &(w->nresvals_to_sched));

        w1 = CONTAINER_OF(w, worker_t, host);
        unlock(w1);

        dereference(w1);
    }
}

void unlock_collect_jobs(xlist_t* l)
{
    int i;

    for (i = 0; i < l->count; i++) {
        xce_job_t* j = (xce_job_t*)xlist_get(l, i);
        job_t*     j1;

        if (j == NULL) {
            continue;
        }

        j1 = CONTAINER_OF(j, job_t, job);
        unlock(j1);

        dereference(j1);
    }
}