myxce/src/lib/xcelimit.c

#include <errno.h>
#include <rtti.h>
#include <stdarg.h>
#include <stdint.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>

#include "group.h"
#include "kvmgr.h"
#include "utils.h"
#include "xce/types.h"
#include "xcelimit.h"

int fmt_consumer(char* buf, int len, xce_consumer_t* c)
{
    struct {
        xce_consumer_type_t type;
        char*               name;
        char*               per_name;
    } consumer_mapping[] = {
        { CONSUMER_USER,    "USER",    "PER_USER"    },
        { CONSUMER_HOST,    "HOST",    "PER_HOST"    },
        { CONSUMER_PROJECT, "PROJECT", "PER_PROJECT" },
        { CONSUMER_QUEUE,   "QUEUE",   "PER_QUEUE"   },
        { -1,               NULL,      NULL          }
    };
    int i;

    if ((buf == NULL) || (c == NULL) || (len <= 0)) {
        return -EINVAL;
    }
    for (i = 0; consumer_mapping[i].type >= 0; i++) {
        if (c->ctype == consumer_mapping[i].type) {
            char* t;
            int   l;
            int   rc;

            if (c->per_consumer) {
                t = consumer_mapping[i].per_name;
            } else {
                t = consumer_mapping[i].name;
            }

            snprintf(buf, len, "%s = ", t);
            l = strlen(buf);

            rc = fmt_group(buf + l, len - l, c->group);
            if (rc < 0) {
                return rc;
            }

            return strlen(buf);
        }
    }

    return -EINVAL;
}

int build_consumer(xce_consumer_t** pconsumer, var mgr,
                   xce_consumer_type_t type, int per_consumer,
                   xce_group_config_t* g_config, char* default_name_prefix)
{
    xce_consumer_t* c = NULL;
    xce_group_t*    g = NULL;

    *pconsumer = NULL;
    c          = (xce_consumer_t*)calloc(1, sizeof(*c));
    if (c == NULL) {
        return -ENOMEM;
    }

    *pconsumer = c;

    c->ctype        = type;
    c->per_consumer = (per_consumer != 0) ? 1 : 0;
    c->group        = NULL;

    if (g_config != NULL) {
        char* default_g_name = NULL;
        char  consumer_group_name[2 * XCE_MAXNAME_LEN + 1];

        if ((g_config->name == NULL) || (strlen(g_config->name) == 0)) {
            char* suffix = NULL;

            switch (type) {
            case CONSUMER_USER:
                suffix = "ug";
                break;
            case CONSUMER_PROJECT:
                suffix = "pg";
                break;
            case CONSUMER_HOST:
                suffix = "hg";
                break;
            case CONSUMER_QUEUE:
                suffix = "qg";
                break;
            default:
                suffix = "unknown";
                break;
            }

            sprintf(consumer_group_name, "%s-%s-group", default_name_prefix,
                    suffix);
            default_g_name = consumer_group_name;
        } else {
            default_g_name = g_config->name;
        }

        g = makeGroup(mgr, g_config, default_g_name);
        if (g == NULL) {
            return -ENOMEM;
        }
    } else {
        g = get_group(mgr, "all");
    }

    c->group = g;

    return 1;
}

void kill_consumer(xce_consumer_t* c)
{
    if (c == NULL) {
        return;
    }

    /* leave c->group managed by group manager */
    FREEUP(c);
}

int fmt_limit(char* buf, int len, xce_limit_t* limit)
{
    int pos = 0;
    int i, rc;

    rc = snprintf(buf + pos, len - pos, "limit: %s\n", limit->name);
    if (rc < 0) {
        return rc;
    }

    pos += rc;

    for (i = 0; i < CONSUMER_MAX; i++) {
        xce_consumer_t* c = limit->consumers[i];

        if (c != NULL) {
            rc = fmt_consumer(buf + pos, len - pos, c);
            if (rc < 0) {
                return rc;
            }

            pos += rc;
            rc = snprintf(buf + pos, len - pos, "\n");
            if (rc < 0) {
                return rc;
            }
            pos += rc;
        }
    }

    rc = snprintf(buf + pos, len - pos, "\n%s = %f%s\n", limit->val.resname,
                  limit->val.val, (limit->val.percent != 0) ? "%" : "");
    if (rc < 0) {
        return rc;
    }

    return strlen(buf);
}

void kill_limit(xce_limit_t* limit)
{
    int i;

    if (limit == NULL) {
        return;
    }

    FREEUP(limit->name);

    for (i = 0; i < CONSUMER_MAX; i++) {
        xce_consumer_t* c = limit->consumers[i];

        if (c != NULL) {
            kill_consumer(c);
        }
    }

    FREEUP(limit->val.resname);
    FREEUP(limit);
}

int build_limit(xce_limit_t** plimit, var ugmgr, var qgmgr, var hgmgr,
                var pgmgr, xce_limit_config_t* config)
{
    xce_limit_t* limit = NULL;
    int          rc    = 0;

    *plimit = NULL;
    limit   = (xce_limit_t*)calloc(1, sizeof(*limit));
    if (limit == NULL) {
        return -ENOMEM;
    }

    *plimit = limit;

    limit->name = strdup(config->name);

    rc = build_consumer(&(limit->consumers[CONSUMER_USER]), ugmgr,
                        CONSUMER_USER, config->per_user, config->users,
                        config->name);
    if (rc < 0) {
        return rc;
    }

    rc = build_consumer(&(limit->consumers[CONSUMER_HOST]), hgmgr,
                        CONSUMER_HOST, config->per_host, config->hosts,
                        config->name);
    if (rc < 0) {
        return rc;
    }

    rc = build_consumer(&(limit->consumers[CONSUMER_QUEUE]), qgmgr,
                        CONSUMER_QUEUE, config->per_queue, config->queues,
                        config->name);
    if (rc < 0) {
        return rc;
    }

    rc = build_consumer(&(limit->consumers[CONSUMER_PROJECT]), pgmgr,
                        CONSUMER_PROJECT, config->per_project, config->projects,
                        config->name);
    if (rc < 0) {
        return rc;
    }

    limit->val.resname = strdup(config->val->resname);
    limit->val.percent = config->val->percent;
    limit->val.val     = config->val->val;

    return 1;
}

int match_limit(xce_limit_t* limit, int global, ...)
{
    va_list         ap;
    char*           user    = NULL;
    char*           queue   = NULL;
    char*           project = NULL;
    char*           host    = NULL;
    xce_consumer_t* c       = NULL;

    va_start(ap, global);
    user    = va_arg(ap, char*);
    queue   = va_arg(ap, char*);
    project = va_arg(ap, char*);

    if ((limit == NULL) || (user == NULL) || (queue == NULL)
        || (project == NULL)) {
        return -EINVAL;
    }

    if (!global) {
        host = va_arg(ap, char*);
        if (host == NULL) {
            return -EINVAL;
        }
    }

    c = limit->consumers[CONSUMER_USER];
    if (c != NULL) {
        if (is_in_group(c->group, user) <= 0) {
            return 0;
        }
    }

    c = limit->consumers[CONSUMER_QUEUE];
    if (c != NULL) {
        if (is_in_group(c->group, queue) <= 0) {
            return 0;
        }
    }

    c = limit->consumers[CONSUMER_PROJECT];
    if (c != NULL) {
        if (is_in_group(c->group, project) <= 0) {
            return 0;
        }
    }

    if (!global) {
        c = limit->consumers[CONSUMER_HOST];
        if (c != NULL) {
            if (is_in_group(c->group, host) <= 0) {
                return 0;
            }
        }
    }

    return 1;
}

int build_vres_name(char** vname, xce_limit_t* limit, char* user, char* queue,
                    char* project)
{
    size_t          bufsize = 0;
    FILE*           fp      = NULL;
    xce_consumer_t* c       = NULL;
    int             rc      = 0;

    *vname = NULL;
    if ((vname == NULL) || (limit == NULL) || (user == NULL) || (queue == NULL)
        || (project == NULL)) {
        return -EINVAL;
    }

    fp = open_memstream(vname, &bufsize);
    if (fp == NULL) {
        return -errno;
    }

    fprintf(fp, "limit=%s", limit->name);

    c = limit->consumers[CONSUMER_USER];

    if ((c != NULL) && (c->per_consumer == 1)) {
        fprintf(fp, "/u=%s", user);
    }

    c = limit->consumers[CONSUMER_QUEUE];
    if ((c != NULL) && (c->per_consumer == 1)) {
        fprintf(fp, "/q=%s", queue);
    }

    c = limit->consumers[CONSUMER_PROJECT];
    if ((c != NULL) && (c->per_consumer == 1)) {
        fprintf(fp, "/p=%s", project);
    }

    fflush(fp);
    
    fclose(fp);
    
    if ((*vname) != NULL) {
        return 1;
    }

    return -ENOMEM;
}

int is_host_based_limit(var resmgr, xce_limit_t* limit)
{
    int             rc = 0;
    xce_consumer_t* c  = NULL;
    xce_resource_t* r  = NULL;

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

    lock(resmgr, RTTI_LOCK_READ);
    rc = key_find(resmgr, (void*)limit->val.resname, (void**)&r);
    unlock(resmgr);

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

    if ((r->flags & (XCE_RESF_GLOBAL | XCE_RESF_SHARED)) == 0) {
        if (limit->val.percent != 0) {
            return 1;
        }
    }

    c = limit->consumers[CONSUMER_HOST];
    if (c == NULL) {
        return 0;
    }

    return c->per_consumer;
}

int add_vres(xce_resource_t** pvres, var resmgr, char* name, int global)
{
    xce_resource_t* vres = NULL;
    int             rc;

    vres   = (xce_resource_t*)calloc(1, sizeof(*vres));
    *pvres = vres;

    if (vres == NULL) {
        return -ENOMEM;
    }

    vres->name    = strdup(name);
    vres->idx     = -1;
    vres->desc    = NULL;
    vres->valtype = XCE_VAL_NUMERIC;
    vres->ordtype = XCE_ORD_DECR;
    vres->flags   = XCE_RESF_LIMIT | XCE_RESF_CONSUMABLE | XCE_RESF_RELEASE;
    if (global) {
        vres->flags |= (XCE_RESF_SHARED | XCE_RESF_GLOBAL);
    } else {
        vres->flags |= XCE_RESF_DEFAULT_HOST;
    }

    lock(resmgr, RTTI_LOCK_WRITE);
    rc = key_add(resmgr, (void*)vres->name, (void*)vres);
    unlock(resmgr);

    if (rc < 0) {
        return rc;
    }

    vres->idx = rc;
    return rc;
}

int instantiate_limit(xce_vres_t** pvres, var resmgr, xce_limit_t* limit,
                      char* user, char* queue, char* project)
{
    xce_vres_t* vres     = NULL;
    int         rc       = 0;
    char*       vresname = NULL;
    int         shared   = 1;

    if ((pvres == NULL) || (resmgr == NULL) || (limit == NULL) || (user == NULL)
        || (queue == NULL) || (project == NULL)) {
        return -EINVAL;
    }

    vres   = (xce_vres_t*)calloc(1, sizeof(*vres));
    *pvres = vres;

    if (vres == NULL) {
        return -ENOMEM;
    }

    vres->limit = limit;

    if (is_host_based_limit(resmgr, limit)) {
        shared = 0;
    }

    rc = build_vres_name(&(vresname), limit, user, queue, project);
    if (rc < 0) {
        return rc;
    }

    lock(resmgr, RTTI_LOCK_READ);
    rc = key_find(resmgr, (void*)vresname, (void**)&(vres->vres));
    unlock(resmgr);

    if (rc < 0) {
        return rc;
    }

    if (rc == 0) {
        rc = add_vres(&(vres->vres), resmgr, vresname, shared);
        if (rc < 0) {
            FREEUP(vresname);
            return rc;
        }
    }

    FREEUP(vresname);

    lock(resmgr, RTTI_LOCK_READ);
    rc = key_find(resmgr, (void*)limit->val.resname, (void**)&(vres->res));
    unlock(resmgr);

    if (rc < 0) {
        return rc;
    }

    if (rc == 0) {
        return -ENOENT;
    }

    return rc;
}

void kill_vres(xce_vres_t* vres)
{
    if (vres == NULL) {
        return;
    }
}