myxce/src/controller/dataex.c

#include <errno.h>

#include "cluster.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;
}

static int collect_resval_with_idx_(zarray_t* array, xlist_t* l)
{
    int i;

    for (i = 0; i < array->caps; i++) {
        xce_val_with_idx_t* val;
        xce_val_t*          src = array->data[i];

        if (array->data[i] == NULL) {
            continue;
        }

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

        val->idx = i;
        val->val = (xce_val_t*)calloc(1, sizeof(xce_val_t));
        if (val->val == NULL) {
            return -ENOMEM;
        }

        if (src->which == Value_u_svalue) {
            val->val->which = Value_u_svalue;
            if (src->u.svalue != NULL) {
                val->val->u.svalue = strdup(src->u.svalue);
            } else {
                val->val->u.svalue = strdup("-");
            }
        } else {
            memcpy(val->val, array->data[i], sizeof(xce_val_t));
        }

        xlist_store(l, val);
    }

    return l->count;
}

int collect_resval_with_idx(zarray_t* array, xlist_t* l)
{
    int rc = 0;

    mylock_lock(&(array->lock));

    rc = collect_resval_with_idx_(array, l);

    mylock_unlock(&(array->lock));

    return 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(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_with_idx(&(w->resvals), &lst);

        w->nresvals_to_sched = lst.count;
        MOVE_PTR(p, lst.data);
        w->resvals_to_sched = (xce_val_with_idx_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 ((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 ((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(l);

    return;
}

void collect_shared_resvals(cluster_t* c, collect_check_fn_t fn, xlist_t* l)
{

    collect_resval_with_idx(&(c->sharedinst), l);

    return;
}

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