improve zarray_iter

src/controller/dataex.c

@@ -186,7 +186,7 @@ void collect_confvars(cluster_t* c, collect_check_fn_t fn, int* nconfig,
     xlist_store(&lst, count_);
     xlist_store(&lst, add_confvar_);
 
-    rc = kvmgr_iter(&(c->configmgr), &lst, &args);
+    rc = kvmgr_iter(&(c->configmgr), NULL, &lst, &args);
 
     xlist_fini(&lst);
 

src/controller/jobmgr.c

@@ -148,7 +148,7 @@ int jobmgr_iter(jobmgr_t* mgr, xlist_t* l, void* arg)
     int rc = 0;
 
     mylock_lock(&(mgr->lock));
-    rc = zarray_iter_ex(&(mgr->jobs), l, arg);
+    rc = zarray_iter_ex(&(mgr->jobs), NULL, l, arg);
     mylock_unlock(&(mgr->lock));
 
     return rc;

src/controller/qmgr.c

@@ -149,7 +149,7 @@ int qmgr_iter(qmgr_t* mgr, xlist_t* l, void* arg)
     int rc = 0;
 
     mylock_lock(&(mgr->lock));
-    rc = zarray_iter_ex(&(mgr->queues), l, arg);
+    rc = zarray_iter_ex(&(mgr->queues), NULL, l, arg);
     mylock_unlock(&(mgr->lock));
 
     return rc;

src/controller/resmgr.c

@@ -184,7 +184,7 @@ int resmgr_iter(resmgr_t* mgr, xlist_t* l, void* arg)
     int rc = 0;
 
     mylock_lock(&(mgr->lock));
-    rc = zarray_iter_ex(&(mgr->defs), l, arg);
+    rc = zarray_iter_ex(&(mgr->defs), NULL, l, arg);
     mylock_unlock(&(mgr->lock));
 
     return rc;

src/controller/workermgr.c

@@ -107,14 +107,10 @@ worker_t* workermgr_lock(workermgr_t* mgr, char* name, int op)
         return NULL;
     }
 
-    /* clang-format off */
+    if ((op != OBJREF_NOP) && (op != OBJREF_READ) && (op != OBJREF_UPDATE)) {
-    if ((op != OBJREF_NOP) &&
+        SET_ERROR(EINVAL);
-	(op != OBJREF_READ) &&
+        return NULL;
-	(op != OBJREF_UPDATE)) {
-	SET_ERROR(EINVAL);
-	return NULL;
     }
-    /* clang-format on */
 
     mylock_lock(&(mgr->lock));
     w = workermgr_lock_(mgr, name, op);
@@ -137,14 +133,10 @@ worker_t* workermgr_lock_byid(workermgr_t* mgr, int idx, int op)
         return NULL;
     }
 
-    /* clang-format off */
+    if ((op != OBJREF_NOP) && (op != OBJREF_READ) && (op != OBJREF_UPDATE)) {
-    if ((op != OBJREF_NOP) &&
+        SET_ERROR(EINVAL);
-	(op != OBJREF_READ) &&
+        return NULL;
-	(op != OBJREF_UPDATE)) {
-	SET_ERROR(EINVAL);
-	return NULL;
     }
-    /* clang-format on */
 
     mylock_lock(&(mgr->lock));
     w = (worker_t*)zarray_get(&(mgr->workers), idx);
@@ -191,7 +183,7 @@ int workermgr_iter(workermgr_t* mgr, xlist_t* l, void* arg)
     int rc = 0;
 
     mylock_lock(&(mgr->lock));
-    rc = zarray_iter_ex(&(mgr->workers), l, arg);
+    rc = zarray_iter_ex(&(mgr->workers), NULL, l, arg);
     mylock_unlock(&(mgr->lock));
 
     return rc;

src/include/kvmgr.h

@@ -16,9 +16,9 @@ typedef struct {
 extern int  kvmgr_init(kvmgr_t*, hash_key_fn_t, int, int, zarray_item_killer_t);
 extern void kvmgr_fini(kvmgr_t*);
 extern void*   kvmgr_lock(kvmgr_t*, void*, int);
+extern void*   kvmgr_lock_byid(kvmgr_t*, int, int);
 extern int32_t kvmgr_add(kvmgr_t*, void*, void*);
 extern void    kvmgr_del(kvmgr_t*, void*);
-extern int     kvmgr_publish(kvmgr_t*, xlist_action_fn_t, void*);
+extern int     kvmgr_iter(kvmgr_t*, zarray_iter_skip_fn_t, xlist_t*, void*);
-extern int     kvmgr_iter(kvmgr_t*, xlist_t*, void*);
 
 #endif

src/include/zarray.h

@@ -7,6 +7,7 @@
 
 typedef void (*zarray_item_killer_t)(void*);
 typedef int (*zarray_iter_fn_t)(void*, void*);
+typedef int (*zarray_iter_skip_fn_t)(void*);
 
 typedef struct {
     mylock_t             lock;
@@ -21,7 +22,7 @@ typedef struct {
 extern int   zarray_init(zarray_t*, int, zarray_item_killer_t);
 extern int   zarray_store(zarray_t*, void*);
 extern int   zarray_clear(zarray_t*, int);
-extern int   zarray_iter_ex(zarray_t*, xlist_t*, void*);
+extern int   zarray_iter_ex(zarray_t*, zarray_iter_skip_fn_t, xlist_t*, void*);
 extern int   zarray_iter(zarray_t*, zarray_iter_fn_t, void*);
 extern int   zarray_del(zarray_t*, int);
 extern int   zarray_put(zarray_t*, int, void*);

src/lib/kvmgr.c

@@ -44,6 +44,9 @@ static void* kvmgr_lock_(kvmgr_t* mgr, void* key, int op)
     }
 
     ptr = (void*)zarray_get(&(mgr->data), idx);
+    if (ptr == NULL) {
+        return NULL;
+    }
 
     return obj_ref(ptr, op);
 }
@@ -57,14 +60,10 @@ void* kvmgr_lock(kvmgr_t* mgr, void* key, int op)
         return NULL;
     }
 
-    /* clang-format off */
+    if ((op != OBJREF_NOP) && (op != OBJREF_READ) && (op != OBJREF_UPDATE)) {
-    if ((op != OBJREF_NOP) &&
+        SET_ERROR(EINVAL);
-	(op != OBJREF_READ) &&
+        return NULL;
-	(op != OBJREF_UPDATE)) {
-	SET_ERROR(EINVAL);
-	return NULL;
     }
-    /* clang-format on */
 
     mylock_lock(&(mgr->lock));
     ptr = kvmgr_lock_(mgr, key, op);
@@ -73,6 +72,36 @@ void* kvmgr_lock(kvmgr_t* mgr, void* key, int op)
     return ptr;
 }
 
+void* kvmgr_lock_byid(kvmgr_t* mgr, int idx, int op)
+{
+    void* ptr = NULL;
+
+    if (mgr == NULL) {
+        SET_ERROR(-EINVAL);
+        return NULL;
+    }
+
+    if ((idx < 0) || (idx >= mgr->data.caps)) {
+        SET_ERROR(-EINVAL);
+        return NULL;
+    }
+
+    if ((op != OBJREF_NOP) && (op != OBJREF_READ) && (op != OBJREF_UPDATE)) {
+        SET_ERROR(-EINVAL);
+        return NULL;
+    }
+
+    mylock_lock(&(mgr->lock));
+    ptr = (void*)zarray_get(&(mgr->data), idx);
+    mylock_unlock(&(mgr->lock));
+
+    if (ptr == NULL) {
+        return NULL;
+    }
+
+    return obj_ref(ptr, op);
+}
+
 static int32_t kvmgr_add_(kvmgr_t* mgr, void* key, void* p)
 {
     hent_t* e = NULL;
@@ -143,48 +172,13 @@ void kvmgr_del(kvmgr_t* mgr, void* key)
     mylock_unlock(&(mgr->lock));
 }
 
-static int add_config_(xlist_t* l, void* data)
+int kvmgr_iter(kvmgr_t* mgr, zarray_iter_skip_fn_t skipfn, xlist_t* l,
-{
+               void* arg)
-    xce_config_t* config;
-
-    if (data == NULL) {
-        return -EINVAL;
-    }
-
-    config = (xce_config_t*)obj_ref(data, OBJREF_READ);
-    return xlist_store(l, config);
-}
-
-int kvmgr_publish(kvmgr_t* mgr, xlist_action_fn_t fn, void* arg)
-{
-    xlist_t l;
-    int     rc = 0;
-    int     i;
-
-    xlist_init(&l, 128);
-
-    mylock_lock(&(mgr->lock));
-    zarray_iter(&(mgr->data), (zarray_iter_fn_t)add_config_, &l);
-    mylock_unlock(&(mgr->lock));
-
-    rc = (*fn)(&l, arg);
-
-    for (i = 0; i < l.count; i++) {
-        xce_config_t* config = l.data[i];
-
-        DEREF_OBJ(config);
-    }
-
-    xlist_fini(&l);
-    return rc;
-}
-
-int kvmgr_iter(kvmgr_t* mgr, xlist_t* l, void* arg)
 {
     int rc = 0;
 
     mylock_lock(&(mgr->lock));
-    rc = zarray_iter_ex(&(mgr->data), l, arg);
+    rc = zarray_iter_ex(&(mgr->data), skipfn, l, arg);
     mylock_unlock(&(mgr->lock));
 
     return rc;

src/lib/zarray.c

@@ -136,7 +136,8 @@ void zarray_fini(zarray_t* array)
     }
 }
 
-int zarray_iter_ex(zarray_t* array, xlist_t* l, void* arg)
+int zarray_iter_ex(zarray_t* array, zarray_iter_skip_fn_t skipfn, xlist_t* l,
+                   void* arg)
 {
     int i, j;
 
@@ -154,6 +155,14 @@ int zarray_iter_ex(zarray_t* array, xlist_t* l, void* arg)
         }
 
         for (i = 0; i < array->caps; i++) {
+            if (skipfn != NULL) {
+                if (skipfn(array->data[i])) {
+                    continue;
+                }
+            } else if (array->data[i] == NULL) {
+                continue;
+            }
+
             (*f)(arg, array->data[i]);
         }
     }
@@ -174,7 +183,7 @@ int zarray_iter(zarray_t* array, zarray_iter_fn_t f, void* arg)
     xlist_init(&l, 2);
     xlist_store(&l, f);
 
-    rc = zarray_iter_ex(array, &l, arg);
+    rc = zarray_iter_ex(array, NULL, &l, arg);
 
     xlist_fini(&l);
     return rc;

src/scheduler/include/scheduler.h

@@ -5,8 +5,8 @@
 #include <pthread.h>
 
 #include "client.h"
-#include "xce/types.h"
 #include "mylock.h"
+#include "xce/types.h"
 #include "xlist.h"
 
 typedef enum {

src/scheduler/src/main.c

@@ -95,8 +95,8 @@ int main(int argc, char* argv[])
         sched_update(&sched);
         sched_go(&sched);
 
-        while(1) {
+        while (1) {
-            xce_decision_t *decision = NULL;
+            xce_decision_t* decision = NULL;
 
             cbuf_get(&(sched.decisions), &decision);
 
@@ -105,7 +105,7 @@ int main(int argc, char* argv[])
                 break;
             }
 
-            switch(decision->op) {
+            switch (decision->op) {
             case XCE_DECISION_NONE:
                 break;
             case XCE_DECISION_DISPATCH:

src/scheduler/src/scheduler.c

@@ -42,7 +42,8 @@ static void* sched_job_picker(void* arg)
     return NULL;
 }
 
-static xce_decision_t *calc_decision(xce_scheduler_t *sched, uint64_t jobid) {
+static xce_decision_t* calc_decision(xce_scheduler_t* sched, uint64_t jobid)
+{
     return NULL;
 }
 
@@ -66,7 +67,7 @@ static void* sched_allocator(void* arg)
 
         /* calculate decision for jobid */
         decision = calc_decision(sched, jobid);
-        
+
         cbuf_put(&(sched->decisions), &decision);
     }