myxce/src/lib/wobj.c

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

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

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

static size_t worker_datasize(va_list ap)
{
    size_t sz = sizeof(worker_t);

    return sz;
}

static void kill_worker(worker_t* w)
{
    dereference(w->host.name);

    if (w->host.attrs != NULL) {
        dereference(w->host.attrs->hosttype);
        dereference(w->host.attrs->hostmodel);
        FREEUP(w->host.attrs);
    }

    if (w->host.dispatch_window_config != NULL) {
        free_timewindow_config(w->host.dispatch_window_config);
        w->host.dispatch_window_config = NULL;
    }

    if (w->host.thresholds != NULL) {
        free_resource_thresholds(w->host.thresholds, w->host.nthresholds);
        w->host.thresholds  = NULL;
        w->host.nthresholds = 0;
    }

    zarray_fini(&(w->host.resvals));

    return;
}

static int copy_worker(worker_t* dst, xce_worker_t* src)
{
    if ((src == NULL) || (src->attrs == NULL)) {
        return -EINVAL;
    }

    dst->host.name  = makeString(src->name);
    dst->host.attrs = (hostattrs_t*)calloc(1, sizeof(hostattrs_t));
    if (dst->host.attrs == NULL) {
        return -ENOMEM;
    }

    memcpy(dst->host.attrs, src->attrs, sizeof(*(dst->host.attrs)));

    dst->host.attrs->hosttype  = makeString(src->attrs->hosttype);
    dst->host.attrs->hostmodel = makeString(src->attrs->hostmodel);
    dst->host.status           = src->status;

    /* FIXME: */
    zarray_init(&(dst->host.resvals), 64, NULL);

    dst->host.dead_since   = src->dead_since;
    dst->host.pending_ctrl = src->pending_ctrl;
    MOVE_PTR(dst->host.dispatch_window_config, src->dispatch_window_config);
    dst->host.nthresholds = src->nthresholds;
    MOVE_PTR(dst->host.thresholds, src->thresholds);
    dst->host.nresvals_to_sched = 0;
    dst->host.resvals_to_sched  = NULL;

    return 1;
}

static int override_worker(worker_t* dst, xce_worker_t* w)
{
#define XCE_WORKER_ADMIN_MASK \
    (XCE_WORKER_LOCKEDU | XCE_WORKER_CLOSED | XCE_WORKER_HIDDEN | XCE_WORKER_CLOSED_BY_ADMIN)
    uint32_t admin_status = 0;
    int      rc           = 0;

    admin_status = dst->host.status & XCE_WORKER_ADMIN_MASK;

    kill_worker(dst);

    rc = copy_worker(dst, w);
    if (rc >= 0) {
        dst->host.status |= admin_status;
        dst->ep = NULL;
    }

    return rc;
#undef XCE_WORKER_ADMIN_MASK
}

static int worker_init(meta_obj_t* obj, va_list ap)
{
    worker_t*         dst  = RTTI_OBJ_TO_DATA(worker_t, obj);
    worker_private_t* priv = (worker_private_t*)RTTI_OBJ_TO_PRIV(obj);
    xce_worker_t*     src  = va_arg(ap, xce_worker_t*);
    int               rc   = 0;

    rc = copy_worker(dst, src);
    if (rc < 0) {
        return rc;
    }

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

    return 1;
}

static void worker_destroy(meta_obj_t* obj)
{
    worker_t*         w    = RTTI_OBJ_TO_DATA(worker_t, obj);
    worker_private_t* priv = (worker_private_t*)RTTI_OBJ_TO_PRIV(obj);

    kill_worker(w);

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

    return;
}

static int worker_reference(meta_obj_t* obj)
{
    worker_private_t* priv = (worker_private_t*)RTTI_OBJ_TO_PRIV(obj);

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

    return 1;
}

static void worker_dereference(meta_obj_t* obj)
{
    worker_private_t* priv  = (worker_private_t*)RTTI_OBJ_TO_PRIV(obj);
    worker_t*         w     = RTTI_OBJ_TO_DATA(worker_t, obj);
    uint64_t          count = 0;

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

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

    return;
}

static int worker_lock(meta_obj_t* obj, uint64_t option)
{
    worker_private_t* priv = (worker_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 worker_unlock(meta_obj_t* obj)
{
    worker_private_t* priv = (worker_private_t*)RTTI_OBJ_TO_PRIV(obj);

    my_rwlock_unlock(&(priv->lock));

    return;
}

static typeinfo_t worker_type = {
    .name        = "xce worker",
    .privateSize = sizeof(worker_private_t),
    .dataSize    = (datasize_fn_t)worker_datasize,
    .init        = (obj_init_fn_t)worker_init,
    .destroy     = NULL,
    .interfaces
    = { .reference
        = &(intf_reference_t) { .reference = (reference_fn_t)worker_reference,
                                .dereference
                                = (dereference_fn_t)worker_dereference },
       .lock = &(intf_lock_t) { .lock   = (lock_fn_t)worker_lock,
                                 .unlock = (unlock_fn_t)worker_unlock } }
};

static worker_t* makeWorker_(var workermgr, xce_worker_t* w)
{
    worker_t* worker = NULL;
    worker_t* w1     = NULL;
    int       rc     = 0;

    rc = key_find(workermgr, (void*)w->name, (void**)&w1);

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

        return NULL;
    }

    if (rc > 0) {
        rc = override_worker(w1, w);
        if (rc < 0) {
            return NULL;
        }

        worker = w1;
    } else {
        worker = makeInstance(&worker_type, w);
        if (worker != NULL) {
            rc = key_add(workermgr, (void*)worker->host.name, (void*)worker);

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

                errno = -rc;
                return NULL;
            }
        }
    }

    return worker;
}

worker_t* makeWorker(var workermgr, xce_worker_t* w)
{
    worker_t* worker = NULL;

    lock(workermgr, RTTI_LOCK_WRITE);
    worker = makeWorker_(workermgr, w);
    unlock(workermgr);

    return worker;
}