improve ordq

src/include/ordq.h

@@ -2,6 +2,7 @@
 
 #define _ORDQ_H_ 1
 
+#include <stdarg.h>
 #include <stdint.h>
 
 #include "mylock.h"
@@ -13,7 +14,7 @@ typedef struct _ordq_item {
 
 typedef int (*ordq_cmp_fn_t)(void*, void*, void*);
 typedef void (*ordq_data_killer_t)(void*);
-typedef int (*ordq_skip_fn_t)(void*, void*);
+typedef int (*ordq_skip_fn_t)(void*, va_list);
 
 typedef struct {
     mylock_t           lock;
@@ -26,8 +27,8 @@ typedef struct {
 extern int          ordq_init(ordq_t*, ordq_data_killer_t, ordq_cmp_fn_t);
 extern void         ordq_fini(ordq_t* q);
 extern int          ordq_put(ordq_t*, void*);
-extern ordq_item_t* ordq_get(ordq_t*, ordq_skip_fn_t);
-extern ordq_item_t* ordq_peek(ordq_t*, ordq_skip_fn_t);
+extern ordq_item_t* ordq_get(ordq_t*, ordq_skip_fn_t, ...);
+extern ordq_item_t* ordq_peek(ordq_t*, ordq_skip_fn_t, ...);
 extern void         ordq_set_ctx(ordq_t*, void*);
 
 #endif

src/lib/ordq.c

@@ -79,49 +79,83 @@ void ordq_fini(ordq_t* q)
     mylock_fini(&(q->lock));
 }
 
-static ordq_item_t* ordq_get_(ordq_t* q, ordq_skip_fn_t skipfn)
+static ordq_item_t* ordq_get_(ordq_t* q, ordq_skip_fn_t skipfn, va_list ap)
 {
-    ordq_item_t** p = &(q->head);
+    ordq_item_t** p  = NULL;
+    ordq_item_t*  it = NULL;
+
+    p = &(q->head);
 
     while ((*p) != NULL) {
-        if ((skipfn == NULL) || !skipfn(q->ctx, (*p)->data)) {
-            ordq_item_t* it = (*p);
-            *p              = it->next;
-            it->next        = NULL;
-            return it;
+        va_list ap1;
+        int     rc = 0;
+
+        if (skipfn == NULL) {
+            break;
+        }
+
+        va_copy(ap1, ap);
+        rc = skipfn((*p)->data, ap1);
+        va_end(ap1);
+
+        if (!rc) {
+            break;
         }
 
         p = &((*p)->next);
     }
 
-    return NULL;
-}
-
-ordq_item_t* ordq_get(ordq_t* q, ordq_skip_fn_t skipfn)
-{
-    ordq_item_t* it;
-
-    mylock_lock(&(q->lock));
-    it = ordq_get_(q, skipfn);
-    mylock_unlock(&(q->lock));
+    it = *p;
+    if (it != NULL) {
+        *p       = it->next;
+        it->next = NULL;
+    }
 
     return it;
 }
 
-ordq_item_t* ordq_peek(ordq_t* q, ordq_skip_fn_t skipfn)
+ordq_item_t* ordq_get(ordq_t* q, ordq_skip_fn_t skipfn, ...)
+{
+    ordq_item_t* it;
+    va_list      ap;
+
+    va_start(ap, skipfn);
+    mylock_lock(&(q->lock));
+    it = ordq_get_(q, skipfn, ap);
+    mylock_unlock(&(q->lock));
+    va_end(ap);
+
+    return it;
+}
+
+ordq_item_t* ordq_peek(ordq_t* q, ordq_skip_fn_t skipfn, ...)
 {
     ordq_item_t* it = NULL;
+    va_list      ap;
 
+    va_start(ap, skipfn);
     mylock_lock(&(q->lock));
     it = q->head;
     while (it != NULL) {
-        if ((skipfn == NULL) || !skipfn(q->ctx, it->data)) {
+        int     rc = 0;
+        va_list ap1;
+
+        if (skipfn == NULL) {
+            break;
+        }
+
+        va_copy(ap1, ap);
+        rc = skipfn(it->data, ap1);
+        va_end(ap1);
+
+        if (!rc) {
             break;
         }
 
         it = it->next;
     }
     mylock_unlock(&(q->lock));
+    va_end(ap);
 
     return it;
 }

src/lib/testsuite/test_ordq.c

@@ -1,4 +1,5 @@
 #include <math.h>
+#include <stdarg.h>
 #include <stdio.h>
 #include <stdlib.h>
 #include <string.h>
@@ -46,7 +47,7 @@ static void test_ordq_get(void)
     return;
 }
 
-static int test_skip(void* ctx, void* data)
+static int test_skip(void* data, va_list ap)
 {
     int val = (int)(long)data;
 
@@ -124,7 +125,7 @@ static void test_ordq_peek(void)
     return;
 }
 
-static int test_skip2(void* ctx, void* data)
+static int test_skip2(void* data, va_list ap)
 {
     int val = (int)(long)data;
 

src/scheduler/src/plugin_fcfs.c

@@ -105,10 +105,10 @@ static int fcfs_update_config(void* arg, xce_conf4sched_t* config)
     return 0;
 }
 
-static int skip_not_matched_queue(void* arg1, void* arg2)
+static int skip_not_matched_queue(void* arg1, va_list ap)
 {
-    char*         name = (char*)arg1;
-    fcfs_queue_t* q    = (fcfs_queue_t*)arg2;
+    fcfs_queue_t* q    = (fcfs_queue_t*)arg1;
+    char*         name = va_arg(ap, char*);
 
     if ((name == NULL) || (q == NULL)) {
         return 1;
@@ -126,8 +126,8 @@ static int fcfs_update_queue(void* arg, xce_sched_queue_t* q)
 
     TRACE("%s queue %s...", __func__, q->name);
 
-    privdata->queues.ctx = q->name;
-    it = ordq_get(&(privdata->queues), (ordq_skip_fn_t)skip_not_matched_queue);
+    it = ordq_get(&(privdata->queues), (ordq_skip_fn_t)skip_not_matched_queue,
+                  q->name);
     if (it == NULL) {
         queue = (fcfs_queue_t*)calloc(1, sizeof(*queue));
         if (queue == NULL) {
@@ -151,8 +151,7 @@ static fcfs_queue_t* fcfs_get_queue(fcfs_plugin_data_t* data, char* name)
 {
     ordq_item_t* it = NULL;
 
-    data->queues.ctx = name;
-    it = ordq_peek(&(data->queues), (ordq_skip_fn_t)skip_not_matched_queue);
+    it = ordq_peek(&(data->queues), (ordq_skip_fn_t)skip_not_matched_queue, name);
     if (it == NULL) {
         return NULL;
     }
@@ -194,9 +193,9 @@ static int fcfs_update_job(void* arg, xce_sched_job_t* j)
     return 0;
 }
 
-static int skip_no_pend_job_queue(void* arg1, void* arg2)
+static int skip_no_pend_job_queue(void* arg1, va_list ap)
 {
-    fcfs_queue_t* q = (fcfs_queue_t*)arg2;
+    fcfs_queue_t* q = (fcfs_queue_t*)arg1;
 
     if (q == NULL) {
         return 1;
@@ -205,9 +204,9 @@ static int skip_no_pend_job_queue(void* arg1, void* arg2)
     return (ordq_peek(&(q->pendJobs), NULL) == NULL);
 }
 
-static int skip_not_ready_job(void* arg1, void* arg2)
+static int skip_not_ready_job(void* arg1, va_list ap)
 {
-    fcfs_job_t* job = (fcfs_job_t*)arg2;
+    fcfs_job_t* job = (fcfs_job_t*)arg1;
 
     return !sched_job_is_ready(job->job);
 }