myxce/src/controller/misc.c

#include <errno.h>
#include <stdlib.h>

#include "cluster.h"
#include "worker.h"
#include "xce/types.h"
#include "xlist.h"

int add_to_xlist(xlist_t* l, void* data) { return xlist_store(l, data); }

int add_worker_to_xlist(xlist_t* l, void* data)
{
    worker_t* w = (worker_t*)data;

    return xlist_store(l, &(w->host));
}

int add_queue_to_xlist(xlist_t* l, void* data)
{
    queue_t* q = (queue_t*)data;

    return xlist_store(l, &(q->queue));
}

int add_job_to_xlist(xlist_t* l, void* data)
{
    job_t* j = (job_t*)data;

    return xlist_store(l, &(j->job));
}

int get_ep_role(endpoint_t* ep)
{
    ep_private_t* priv;

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

    priv = (ep_private_t*)ep->data;
    if (priv == NULL) {
        ERROR("priv data of endpoint %p is NULL", ep);
        return -EINVAL;
    }

    return priv->role;
}

worker_t* get_ep_worker(cluster_t* c, endpoint_t* ep)
{
    int       role, idx;
    worker_t* w  = NULL;
    int       rc = 0;

    role = get_ep_role(ep);
    if (role != HostRole_worker) {
        ERROR("endpoint %p is not a worker", ep);
        return NULL;
    }

    idx = ((ep_private_t*)ep->data)->u.idx;
    if (idx < 0) {
        ERROR("invalid idx (%d) for endpoint %p", idx, ep);
        return NULL;
    }

    lock(c->workermgr, RTTI_LOCK_READ);
    rc = vget(c->workermgr, idx, (void**)&w);
    if (w != NULL) {
        reference(w);
    }
    unlock(c->workermgr);

    if (rc < 0) {
        return NULL;
    }

    if (w == NULL) {
        ERROR("can not find worker with id (%d)", idx);
    }

    return w;
}

int update_host_runusage(xce_job_runusage_t** runusage, char* host, int count,
                         xce_rusage_t** usages)
{
    xce_job_runusage_t* rusages = NULL;
    int                 i, j;

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

    rusages = *runusage;
    j       = -1;

    for (i = 0; i < rusages->num_hosts; i++) {
        if (strcmp(host, rusages->host_runusage[i]->host_name) == 0) {
            j = i;
            break;
        }
    }

    if (j < 0) {
        int                   cnt = rusages->num_hosts + 1;
        xce_host_runusage_t** hru = (xce_host_runusage_t**)realloc(
            rusages->host_runusage, cnt * sizeof(xce_host_runusage_t*));
        if (hru == NULL) {
            return -ENOMEM;
        }

        rusages->host_runusage = hru;
        j                      = cnt - 1;

        hru[j] = (xce_host_runusage_t*)calloc(1, sizeof(xce_host_runusage_t));
        if (hru[j] == NULL) {
            return -ENOMEM;
        }

        rusages->num_hosts++;
        hru[j]->host_name = host;
    }

    for (i = 0; i < count; i++) {
        if (strcmp(usages[i]->resname, "cputime") == 0) {
            rusages->host_runusage[j]->cpu_time = usages[i]->value;
            continue;
        }

        if (strcmp(usages[i]->resname, "mem") == 0) {
            rusages->host_runusage[j]->mem_usage = usages[i]->value;
            continue;
        }
    }

    return 0;
}

int update_pids(job_pidinfo_t** pidinfo, int npids, int* pids)
{
    job_pidinfo_t* info;
    int            i;

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

    info = *pidinfo;

    if (info->pids == NULL) {
        info->pids = (unsigned int*)calloc(npids, sizeof(unsigned int));
    } else if (info->npids < npids) {
        unsigned int* ptr;
        ptr = (unsigned int*)realloc(info->pids, npids * sizeof(unsigned int));
        if (ptr == NULL) {
            return -ENOMEM;
        }

        info->pids = ptr;
    }

    for (i = 0; i < npids; i++) {
        info->pids[i] = pids[i];
    }

    info->npids = npids;

    return 0;
}

int update_pgids(job_pidinfo_t** pidinfo, int npgids, int* pgids)
{
    job_pidinfo_t* info;
    int            i;

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

    info = *pidinfo;

    if (info->pgids == NULL) {
        info->pgids = (unsigned int*)calloc(npgids, sizeof(unsigned int));
    } else if (info->npgids < npgids) {
        unsigned int* ptr;
        ptr = (unsigned int*)realloc(info->pgids,
                                     npgids * sizeof(unsigned int));
        if (ptr == NULL) {
            return -ENOMEM;
        }

        info->pgids = ptr;
    }

    for (i = 0; i < npgids; i++) {
        info->pgids[i] = pgids[i];
    }

    info->npgids = npgids;

    return 0;
}