myxce/src/lib/qobj.c

#include <errno.h>
#include <rtti.h>
#include <stdarg.h>
#include <stdatomic.h>
#include <stdlib.h>
#include <string.h>

#include "mylock.h"
#include "qobj.h"
#include "utils.h"
#include "xstring.h"

typedef struct {
    my_rwlock_t  lock;
    atomic_ulong refcount;
} queue_private_t;

static size_t queue_datasize(va_list ap)
{
    size_t sz = sizeof(queue_t);

    return sz;
}

static int queue_init(meta_obj_t* obj, va_list ap)
{
    queue_t*         dst  = RTTI_OBJ_TO_DATA(queue_t, obj);
    queue_private_t* priv = (queue_private_t*)RTTI_OBJ_TO_PRIV(obj);
    xce_queue_t*     src  = va_arg(ap, xce_queue_t*);

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

    dst->queue.queue           = makeString(src->queue);
    dst->queue.attrs           = src->attrs;
    dst->queue.status          = src->status;
    dst->queue.priority        = src->priority;
    dst->queue.accept_interval = src->accept_interval;
    dst->queue.sched_delay     = src->sched_delay;
    dst->queue.resreq          = makeString(src->resreq);
    MOVE_PTR(dst->queue.users, src->users);
    MOVE_PTR(dst->queue.hosts, src->hosts);
    dst->queue.nthresholds = src->nthresholds;
    MOVE_PTR(dst->queue.thresholds, src->thresholds);
    src->nthresholds = 0;

    MOVE_PTR(dst->queue.runwindow, src->runwindow);
    MOVE_PTR(dst->queue.dispatchwindow, src->dispatchwindow);
    MOVE_PTR(dst->queue.runwindow_config, src->runwindow_config);
    MOVE_PTR(dst->queue.dispatchwindow_config, src->dispatchwindow_config);
    dst->queue.pending_ctrl = src->pending_ctrl;
    dst->queue.pre_exec     = makeString(src->pre_exec);
    dst->queue.post_exec    = makeString(src->post_exec);
    dst->queue.autopscaler  = makeString(src->autopscaler);

    my_rwlock_init(&(priv->lock));
    priv->refcount = 1;

    return 1;
}

static void queue_destroy(meta_obj_t* obj)
{
    queue_t*         q    = RTTI_OBJ_TO_DATA(queue_t, obj);
    queue_private_t* priv = (queue_private_t*)RTTI_OBJ_TO_PRIV(obj);

    dereference(q->queue.queue);
    dereference(q->queue.resreq);

    if (q->queue.users != NULL) {
        free_GroupConfig_ptr(&(q->queue.users));
    }

    if (q->queue.hosts != NULL) {
        free_GroupConfig_ptr(&(q->queue.hosts));
    }

    if (q->queue.thresholds != NULL) {
        free_Threshold_list(q->queue.nthresholds, q->queue.thresholds);
    }

    if (q->queue.runwindow != NULL) {
        /* FIXME: */
    }

    if (q->queue.dispatchwindow != NULL) {
        /* FIXME: */
    }

    if (q->queue.runwindow_config != NULL) {
        free_TimeWindow_ptr(&(q->queue.runwindow_config));
    }

    if (q->queue.dispatchwindow_config != NULL) {
        free_TimeWindow_ptr(&(q->queue.dispatchwindow_config));
    }

    dereference(q->queue.pre_exec);
    dereference(q->queue.post_exec);
    dereference(q->queue.autopscaler);

    my_rwlock_fini(&(priv->lock));
    memset(priv, 0x0, sizeof(*priv));

    return;
}

static int queue_reference(meta_obj_t* obj)
{
    queue_private_t* priv = (queue_private_t*)RTTI_OBJ_TO_PRIV(obj);

    atomic_fetch_add(&(priv->refcount), 1);

    return 1;
}

static void queue_dereference(meta_obj_t* obj)
{
    queue_private_t* priv  = (queue_private_t*)RTTI_OBJ_TO_PRIV(obj);
    queue_t*         w     = RTTI_OBJ_TO_DATA(queue_t, obj);
    uint64_t         count = 0;

    count = atomic_fetch_sub(&(priv->refcount), 1);

    if (count == 1) {
        destroy_(w, (obj_destroy_fn_t)queue_destroy);
    }

    return;
}

static int queue_lock(meta_obj_t* obj, uint64_t option)
{
    queue_private_t* priv = (queue_private_t*)RTTI_OBJ_TO_PRIV(obj);
    int              rc   = -EINVAL;

    switch (option) {
    case RTTI_LOCK_READ:
        rc = my_rwlock_rlock(&(priv->lock));
        break;
    case RTTI_LOCK_WRITE:
        rc = my_rwlock_wlock(&(priv->lock));
        break;
    default:
        break;
    }

    return rc;
}

static void queue_unlock(meta_obj_t* obj)
{
    queue_private_t* priv = (queue_private_t*)RTTI_OBJ_TO_PRIV(obj);

    my_rwlock_unlock(&(priv->lock));

    return;
}

static typeinfo_t queue_type = {
    .name        = "xce queue",
    .privateSize = sizeof(queue_private_t),
    .dataSize    = (datasize_fn_t)queue_datasize,
    .init        = (obj_init_fn_t)queue_init,
    .destroy     = NULL,
    .interfaces
    = { .reference
        = &(intf_reference_t) { .reference = (reference_fn_t)queue_reference,
                                .dereference
                                = (dereference_fn_t)queue_dereference },
       .lock = &(intf_lock_t) { .lock   = (lock_fn_t)queue_lock,
                                 .unlock = (unlock_fn_t)queue_unlock } }
};

static queue_t* makeQueue_(var qmgr, xce_queue_t* q)
{
    queue_t* queue = NULL;
    int      rc    = 0;

    rc = key_find(qmgr, (void*)q->queue, (void**)&queue);

    if ((rc < 0) && (rc != -ENOENT)) {
        errno = -rc;
        return NULL;
    }

    if (rc > 0) {
        errno = EEXIST;
        return NULL;
    }

    queue = makeInstance(&queue_type, q);
    if (queue != NULL) {
        rc = key_add(qmgr, (void*)q->queue, (void*)queue);

        if (rc < 0) {
            destroy(queue);

            errno = -rc;
            return NULL;
        }
    }

    return queue;
}

queue_t* makeQueue(var qmgr, xce_queue_t* q)
{
    queue_t* queue = NULL;
    int      rc;

    lock(qmgr, RTTI_LOCK_WRITE);
    queue = makeQueue_(qmgr, q);
    rc    = errno;
    unlock(qmgr);
    errno = rc;

    return queue;
}