walker API for both pqueue and binheap

src/binheap.c

@@ -754,6 +754,62 @@ binheap_decrease_minimum(binheap_t *bh, int decr)
     binomial_tree_node_increase_key(minitem, -decr);
 }
 
+static int
+binomial_tree_walk(binomial_tree_node_t *node, int *count, binheap_walk_callback_t cb, void *priv)
+{
+    int proceed = 0;
+    int remove = 0;
+    (*count)++;
+    int rc = cb(node->bh, node->key, node->klen, node->value, node->vlen, priv);
+    switch(rc) {
+        case -2:
+            proceed = 0;
+            remove = 1;
+            break;
+        case -1:
+            proceed = 1;
+            remove = 1;
+            break;
+        case 0:
+            proceed = 0;
+            remove = 0;
+            break;
+        case 1:
+            proceed = 1;
+            remove = 0;
+            break;
+        default:
+            // TODO - Warning messages? (the callback returned an invalid return code)
+            break;
+    }
+    if (proceed) {
+        int i;
+        for (i = 0; i < node->num_children; i ++) {
+            binomial_tree_node_t *child = node->children[i];
+            proceed = binomial_tree_walk(child, count, cb, priv); 
+            if (!proceed)
+                break;
+        }
+    }
+    if (remove) {
+        binomial_tree_node_destroy(node, 0);
+    }
+    return proceed;
+}
+
+int
+binheap_walk(binheap_t *bh, binheap_walk_callback_t cb, void *priv)
+{
+    int cnt = 0;
+    int i;
+    for (i = 0; i < list_count(bh->trees); i++) {
+        binomial_tree_node_t *curtree = pick_value(bh->trees, i);
+        if (!binomial_tree_walk(curtree, &cnt, cb, priv))
+            break;
+    }
+    return cnt;
+}
+
 
 #define BINHEAP_CMP_KEYS_TYPE(__type, __k1, __k1s, __k2, __k2s) \
 { \

src/binheap.h

@@ -256,6 +256,35 @@ void binheap_decrease_key(binheap_t *bh, void *key, size_t klen, int decr);
  */
 binheap_t *binheap_merge(binheap_t *bh1, binheap_t *bh2);
 
+/**
+ * @brief Callback called for each node when walking the priority queue
+ * @param bh A valid pointer to an initialized binheap_t structure
+ * @param key  The key of the current node
+ * @param klen The size of the key
+ * @param value The value of the current node
+ * @param vlen  The size of the value
+ * @param priv The private pointer passed to binheap_walk()
+ * @return 1 If the walker can go ahead visiting the next node,
+ *         0 if the walker should stop and return
+ *        -1 if the current node should be removed and the walker can go ahead
+ *        -2 if the current node should be removed and the walker should stop
+ */
+typedef int (*binheap_walk_callback_t)(binheap_t *bh, void *key, size_t klen, void *value, size_t vlen, void *priv);
+
+/**
+ * @brief Walk the entire priority queue and call the provided
+ *        callback for each visited node
+ * @note The callback can both stop the walker and/or remove the currently
+ *       visited node using its return value (check binheap_walk_callback_t)
+ * @param bh A valid pointer to an initialized binheap_t structure
+ * @param cb The callback to call for each visited node
+ * @param priv A private pointer which will be passed to the callback at each
+ *             call
+ * @return The number of visited nodes
+ *
+ */
+int binheap_walk(binheap_t *bh, binheap_walk_callback_t cb, void *priv);
+
 /**
  * @brief Return the number of items in the heap
  * @param bh A valid pointer to an initialized binheap_t structure

src/pqueue.c

@@ -1,4 +1,5 @@
 #include <stdlib.h>
+#include <string.h>
 #include <pthread.h>
 
 #include "binheap.h"
@@ -143,6 +144,62 @@ pqueue_pull_lowest(pqueue_t *pq, void **value, size_t *len, uint64_t *prio)
     return rc;
 }
 
+typedef struct {
+    pqueue_t *pq;
+    pqueue_walk_callback_t cb;
+    void *priv;
+} pqueue_walk_helper_arg_t;
+
+static int 
+pqueue_walk_helper(binheap_t *bh, void *key, size_t klen, void *value, size_t vlen, void *priv)
+{
+    pqueue_walk_helper_arg_t *arg = (pqueue_walk_helper_arg_t *)priv;
+    uint64_t *prio = (uint64_t *)key;
+    return arg->cb(arg->pq, *prio, value, vlen, arg->priv);
+}
+
+int
+pqueue_walk(pqueue_t *pq, pqueue_walk_callback_t cb, void *priv)
+{
+    pqueue_walk_helper_arg_t arg = {
+        .pq = pq,
+        .cb = cb,
+        .priv = priv
+    };
+    return binheap_walk(pq->heap, pqueue_walk_helper, &arg);
+}
+
+
+typedef struct {
+    void *value;
+    size_t len;
+    int found;
+} pqueue_remove_helper_arg_t;
+
+static int
+pqueue_remove_helper(pqueue_t *pq, uint64_t prio, void *value, size_t len, void *priv)
+{
+    pqueue_remove_helper_arg_t *arg = (pqueue_remove_helper_arg_t *)priv;
+    if (len == arg->len && *((char *)value) == *((char *)arg->value) &&
+        memcmp(value, arg->value, len) == 0)
+    {
+        return -2;
+    }
+    return 1;
+}
+
+int
+pqueue_remove(pqueue_t *pq, void *value, size_t len)
+{
+    pqueue_remove_helper_arg_t arg = {
+        .value = value,
+        .len = len,
+        .found = 0
+    };
+    pqueue_walk(pq, pqueue_remove_helper, &arg);
+    return arg.found ? 0 : -1;
+}
+
 uint32_t
 pqueue_count(pqueue_t *pq)
 {

src/pqueue.h

@@ -47,6 +47,7 @@ pqueue_t *pqueue_create(pqueue_mode_t mode, uint32_t size, pqueue_free_value_cal
  * @param pq A valid pointer to an initialized pqueue_t structure
  * @param prio The priority to assign to the new value
  * @param value The new value to add to the queue
+ * @param len The size of the value
  * @return 0 if the value has been successfully added to the queue;\n
  *         -1 in case of errors
  * @note If the number of items after the insertion will be bigger than the
@@ -82,6 +83,46 @@ int pqueue_pull_highest(pqueue_t *pq, void **value, size_t *len, uint64_t *prio)
  */
 int pqueue_pull_lowest(pqueue_t *pq, void **value, size_t *len, uint64_t *prio);
 
+/**
+ * @brief Callback called for each node when walking the priority queue
+ * @param pq A valid pointer to an initialized pqueue_t structure
+ * @param prio The priority of the current node
+ * @param value The value stored in the current node
+ * @param len The size of the value
+ * @param priv The private pointer passed to pqueue_walk()
+ * @return 1 If the walker can go ahead visiting the next node,
+ *         0 if the walker should stop and return
+ *        -1 if the current node should be removed and the walker can go ahead
+ *        -2 if the current node should be removed and the walker should stop
+ */
+typedef int (*pqueue_walk_callback_t)(pqueue_t *pq, uint64_t prio, void *value, size_t len, void *priv);
+
+/**
+ * @brief Walk the entire priority queue and call the provided
+ *        callback for each visited node
+ * @note The callback can both stop the walker and/or remove the currently
+ *       visited node using its return value (check pqueue_walk_callback_t)
+ * @param pq A valid pointer to an initialized pqueue_t structure
+ * @param cb The callback to call for each visited node
+ * @param priv A private pointer which will be passed to the callback at each
+ *             call
+ * @return The number of visited nodes
+ *
+ */
+int pqueue_walk(pqueue_t *pq, pqueue_walk_callback_t cb, void *priv);
+
+/**
+ * @brief Remove the first node matching the given value
+ * @note If more than one node have the same value, only the first one will be
+ *       removed from the priority queue
+ * @param pq A valid pointer to an initialized pqueue_t structure
+ * @param value The value to mach
+ * @param len The size of the value
+ * @return 0 if a matching node has been found and successfully removed;\n
+ *        -1 If no matching node was found or an error occurred
+ */
+int pqueue_remove(pqueue_t *pq, void *value, size_t len);
+
 /**
  * @brief Return the number of values stored in the priority queue
  * @param pq A valid pointer to an initialized pqueue_t structure

test/pqueue_test.c

@@ -13,6 +13,14 @@ void free_item(void *item)
     free(item);
 }
 
+static int
+test_walk(pqueue_t *pq, uint64_t prio, void *value, size_t len, void *priv)
+{
+    int *count = (int *)priv;
+    (*count)++;
+    return 1;
+}
+
 int
 main(int argc, char **argv)
 {
@@ -77,6 +85,14 @@ main(int argc, char **argv)
     ut_testing("pqueue_count(pq) == 99");
     ut_validate_int(pqueue_count(pq), 99);
 
+    int cnt = 0;
+
+    ut_testing("pqueue_walk(pq, test_walk, &count)");
+    int visited = pqueue_walk(pq, test_walk, &cnt);
+    // pqueue_walk should report we have visited 99 nodes
+    // and count should have also been summed up to 99
+    ut_validate_int(cnt+visited, 99*2);
+
     ut_testing("pqueue_destroy() and the free_value_callback");
     pqueue_destroy(pq);
     // 100 items need to have been freed so far