use librtti for worker/queue/resource

2026-02-25 16:56:42 +08:00 · 2026-02-25 16:56:42 +08:00 · 2ee87d4ee6
commit 2ee87d4ee6
parent 532c68e6e1
23 changed files with 1078 additions and 396 deletions
--- a/src/lib/qobj.c
+++ b/src/lib/qobj.c
@ -3,3 +3,168 @@
 #include <stdarg.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;
+    uint64_t    refcount;
+} queue_private_t;
+
+static size_t queue_datasize(va_list ap)
+{
+    size_t sz = sizeof(xce_queue_t);
+
+    return sz;
+}
+
+static int queue_init(meta_obj_t* obj, va_list ap)
+{
+    xce_queue_t*     dst  = RTTI_OBJ_TO_DATA(xce_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           = makeString(src->queue);
+    dst->attrs           = src->attrs;
+    dst->status          = src->status;
+    dst->priority        = src->priority;
+    dst->accept_interval = src->accept_interval;
+    dst->sched_delay     = src->sched_delay;
+    dst->resreq          = makeString(src->resreq);
+    MOVE_PTR(dst->users, src->users);
+    MOVE_PTR(dst->hosts, src->hosts);
+    dst->nthresholds = src->nthresholds;
+    MOVE_PTR(dst->thresholds, src->thresholds);
+
+    MOVE_PTR(dst->runwindow, src->runwindow);
+    MOVE_PTR(dst->dispatchwindow, src->dispatchwindow);
+    MOVE_PTR(dst->runwindow_config, src->runwindow_config);
+    MOVE_PTR(dst->dispatchwindow_config, src->dispatchwindow_config);
+    dst->pending_ctrl = src->pending_ctrl;
+    dst->pre_exec     = makeString(src->pre_exec);
+    dst->post_exec    = makeString(src->post_exec);
+    dst->autopscaler  = makeString(src->autopscaler);
+
+    my_rwlock_init(&(priv->lock));
+    priv->refcount = 1;
+
+    return 1;
+}
+
+static void queue_destroy(meta_obj_t* obj)
+{
+    xce_queue_t*     q    = RTTI_OBJ_TO_DATA(xce_queue_t, obj);
+    queue_private_t* priv = (queue_private_t*)RTTI_OBJ_TO_PRIV(obj);
+
+    dereference(q->queue);
+    dereference(q->resreq);
+
+    /* FIXME: */
+    if (q->users != NULL) { }
+
+    if (q->hosts != NULL) { }
+
+    if (q->thresholds != NULL) { }
+
+    if (q->runwindow != NULL) { }
+
+    if (q->dispatchwindow != NULL) { }
+
+    if (q->runwindow_config != NULL) { }
+
+    if (q->dispatchwindow_config != NULL) { }
+
+    dereference(q->pre_exec);
+    dereference(q->post_exec);
+    dereference(q->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);
+
+    my_rwlock_wlock(&(priv->lock));
+    priv->refcount++;
+    my_rwlock_unlock(&(priv->lock));
+
+    return 1;
+}
+
+static void queue_dereference(meta_obj_t* obj)
+{
+    queue_private_t* priv  = (queue_private_t*)RTTI_OBJ_TO_PRIV(obj);
+    xce_queue_t*     w     = RTTI_OBJ_TO_DATA(xce_queue_t, obj);
+    uint64_t         count = 0;
+
+    my_rwlock_wlock(&(priv->lock));
+    priv->refcount--;
+    count = priv->refcount;
+    my_rwlock_unlock(&(priv->lock));
+
+    if (count == 0) {
+        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 } }
+};
+
+xce_queue_t* makeQueue(xce_queue_t* w)
+{
+    xce_queue_t* queue = makeInstance(&queue_type, w);
+    return queue;
+}