enhance ordq

src/include/ordq.h

@@ -16,6 +16,7 @@ typedef struct _ordq_item {
 
 typedef int (*ordq_cmp_fn_t)(void*, ordkey_t, ordkey_t);
 typedef void (*ordq_data_killer_t)(void*);
+typedef int (*ordq_skip_fn_t)(void*, ordq_item_t*);
 
 typedef struct {
     mylock_t           lock;
@@ -30,8 +31,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*, ordkey_t, void*);
-extern ordq_item_t* ordq_get(ordq_t*);
-extern ordq_item_t* ordq_peek(ordq_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/delay.c

@@ -19,7 +19,7 @@ static void* delay_worker(void* args)
         pthread_testcancel();
         sleep(mgr->precision);
 
-        it = ordq_peek(&(mgr->tasks));
+        it = ordq_peek(&(mgr->tasks), NULL);
         if (it == NULL) {
             continue;
         }
@@ -29,7 +29,7 @@ static void* delay_worker(void* args)
             continue;
         }
 
-        it = ordq_get(&(mgr->tasks));
+        it = ordq_get(&(mgr->tasks), NULL);
         t  = (delay_task_t*)it->data;
 
         mgr->runner_fn(t->entry, t->args);

src/lib/ordq.c

@@ -80,28 +80,48 @@ void ordq_fini(ordq_t* q)
     mylock_fini(&(q->lock));
 }
 
-ordq_item_t* ordq_get(ordq_t* q)
+static ordq_item_t* ordq_get_(ordq_t* q, ordq_skip_fn_t skipfn)
+{
+    ordq_item_t** p = &(q->head);
+
+    while ((*p) != NULL) {
+        if ((skipfn == NULL) || !skipfn(q->ctx, (*p))) {
+            ordq_item_t* it = (*p);
+            *p              = it->next;
+            it->next        = NULL;
+            return it;
+        }
+
+        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 = q->head;
-    if (it != NULL) {
-        q->head = it->next;
-    }
-
+    it = ordq_get_(q, skipfn);
     mylock_unlock(&(q->lock));
 
     return it;
 }
 
-ordq_item_t* ordq_peek(ordq_t* q)
+ordq_item_t* ordq_peek(ordq_t* q, ordq_skip_fn_t skipfn)
 {
     ordq_item_t* it = NULL;
 
     mylock_lock(&(q->lock));
     it = q->head;
+    while (it != NULL) {
+        if ((skipfn == NULL) || !skipfn(q->ctx, it)) {
+            break;
+        }
+
+        it = it->next;
+    }
     mylock_unlock(&(q->lock));
 
     return it;

src/lib/testsuite/test_ordq.c

@@ -35,7 +35,7 @@ static void test_ordq_get(void)
     }
 
     for (i = 0; i < 8; i++) {
-        it = ordq_get(&q);
+        it = ordq_get(&q, NULL);
 
         TEST_ASSERT_TRUE((it != NULL) && (it->key == mk_ordkey(val[i])));
         free(it);
@@ -46,6 +46,55 @@ static void test_ordq_get(void)
     return;
 }
 
+static int test_skip(void *ctx, ordq_item_t *data) {
+    int val = (int)(long)data->key;
+
+    return (val % 2 != 0);
+}
+
+static void test_ordq_get_skip(void)
+{
+    int          rc;
+    ordq_t       q;
+    ordq_item_t* it;
+    int          keys[8] = { 9, 8, 1, 4, 7, 2, 6, 2 };
+    int          val[8]  = { 1, 2, 2, 4, 6, 7, 8, 9 };
+    int          i, j, nc;
+
+    rc = ordq_init(&q, NULL, (ordq_cmp_fn_t)long_cmp);
+    TEST_ASSERT_TRUE(rc >= 0);
+
+    for (i = 0; i < 8; i++) {
+        ordq_put(&q, mk_ordkey(keys[i]), (void*)NULL);
+    }
+
+    j = 0; nc = 0;
+    for (i = 0; i < 5; i++) {
+        it = ordq_get(&q, test_skip);
+
+        if ( it == NULL) {
+            break;
+        }
+        
+        nc ++;
+        
+        while((j < 8) && (val[j] % 2 != 0)) {
+            j ++;
+        }
+        
+        TEST_ASSERT_TRUE(it->key == mk_ordkey(val[j]));
+        free(it);
+
+        j ++;
+    }
+
+    TEST_ASSERT_TRUE(nc == 5);
+    
+    ordq_fini(&q);
+
+    return;
+}
+
 static void test_ordq_peek(void)
 {
     int          rc;
@@ -63,7 +112,7 @@ static void test_ordq_peek(void)
     }
 
     for (i = 0; i < 8; i++) {
-        it = ordq_peek(&q); /* ordq_peek() will not remove item from q */
+        it = ordq_peek(&q, NULL); /* ordq_peek() will not remove item from q */
 
         TEST_ASSERT_TRUE((it != NULL) && (it->key == mk_ordkey(val[0])));
     }
@@ -73,10 +122,45 @@ static void test_ordq_peek(void)
     return;
 }
 
+static int test_skip2(void *ctx, ordq_item_t *data) {
+    int val = (int)(long)data->key;
+
+    return (val % 3 != 0);
+}
+
+static void test_ordq_peek_skip(void)
+{
+    int          rc;
+    ordq_t       q;
+    ordq_item_t* it;
+    int          keys[8] = { 9, 8, 1, 4, 7, 2, 6, 2 };
+    int          val[8]  = { 1, 2, 2, 4, 6, 7, 8, 9 };
+    int          i;
+
+    rc = ordq_init(&q, NULL, (ordq_cmp_fn_t)long_cmp);
+    TEST_ASSERT_TRUE(rc >= 0);
+
+    for (i = 0; i < 8; i++) {
+        ordq_put(&q, mk_ordkey(keys[i]), (void*)NULL);
+    }
+
+    for (i = 0; i < 8; i++) {
+        it = ordq_peek(&q, test_skip2); /* ordq_peek() will not remove item from q */
+
+        TEST_ASSERT_TRUE((it != NULL) && (it->key == mk_ordkey(val[4])));
+    }
+
+    ordq_fini(&q);
+
+    return;
+}
+
 void test_ordq(void)
 {
     printf("\n>> ordq test\n\n");
 
     RUN_TEST(test_ordq_get);
     RUN_TEST(test_ordq_peek);
+    RUN_TEST(test_ordq_get_skip);
+    RUN_TEST(test_ordq_peek_skip);
 }