[PATCH] guarantee order in gw-prioqueue

Alexander Malysh <[email protected]>
Newsgroups gmane.comp.mobile.kannel.devel
Message-ID <[email protected]>
Hi all,

attached patch implements extra sequence in the gw-prioqueue that will
guarantee the order of the equal messages. It should fix #436.

Any objections?

-- 
Thanks,
Alex
prioqueue-order.patch (text/x-diff, 4.9 KB)
=== gwlib/gw-prioqueue.c
==================================================================
--- gwlib/gw-prioqueue.c	(revision 892)
+++ gwlib/gw-prioqueue.c	(local)
@@ -55,11 +55,11 @@
  */ 
 
 /*
- * gw-prioqueue.c - generic priority queue.
+ * gw-prioqueue.c - generic priority queue with guaranteed order.
  *
  * Algorithm ala Robert Sedgewick.
  *
- * Alexander Malysh <olek2002 at hotmail.com>, 2004
+ * Alexander Malysh <amalysh at kannel.org>, 2004, 2008
  */
 
 #include "gw-config.h"
@@ -71,12 +71,18 @@
 #include "gw-prioqueue.h"
 
 
+struct element {
+    void *item;
+    long long seq;
+};
+
 struct gw_prioqueue {
     Mutex *mutex;
-    void **tab;
+    struct element **tab;
     size_t size;
     long len;
     long producers;
+    long long seq;
     pthread_cond_t nonempty;
     int (*cmp)(const void*, const void *);
 };
@@ -97,7 +103,7 @@
 static void make_bigger(gw_prioqueue_t *queue, long items)
 {
     size_t size = queue->size;
-    size_t new_size = sizeof(void*) * (queue->len + items);
+    size_t new_size = sizeof(*queue->tab) * (queue->len + items);
     
     if (size >= new_size)
         return;
@@ -107,6 +113,23 @@
 }
 
 
+static int compare(struct element *a, struct element *b, int(*cmp)(const void*, const void *))
+{
+    int rc;
+
+    rc = cmp(a->item, b->item);
+    if (rc == 0) {
+        /* check sequence to guarantee order */
+        if (a->seq < b->seq)
+            rc = 1;
+        else if (a->seq > b->seq)
+            rc = -1;
+    }
+
+    return rc;
+}
+
+
 /**
  * Heapize up
  * @queue - our prioqueue
@@ -114,8 +137,8 @@
  */
 static void upheap(gw_prioqueue_t *queue, register long index)
 {
-    void *v = queue->tab[index];
-    while (queue->tab[index / 2] != NULL && queue->cmp(queue->tab[index / 2], v) <= 0) {
+    struct element *v = queue->tab[index];
+    while (queue->tab[index / 2]->item != NULL && compare(queue->tab[index / 2], v, queue->cmp) < 0) {
         queue->tab[index] = queue->tab[index / 2];
         index /= 2;
     }
@@ -130,16 +153,16 @@
  */
 static void downheap(gw_prioqueue_t *queue, register long index)
 {
-    void *v = queue->tab[index];
+    struct element *v = queue->tab[index];
     register long j;
     
     while (index <= queue->len / 2) {
         j = 2 * index;
         /* take the biggest child item */
-        if (j < queue->len && queue->cmp(queue->tab[j], queue->tab[j + 1]) < 0)
+        if (j < queue->len && compare(queue->tab[j], queue->tab[j + 1], queue->cmp) < 0)
             j++;
         /* break if our item bigger */
-        if (queue->cmp(v, queue->tab[j]) >= 0)
+        if (compare(v, queue->tab[j], queue->cmp) >= 0)
             break;
         queue->tab[index] = queue->tab[j];
         index = j;
@@ -161,11 +184,14 @@
     ret->tab = NULL;
     ret->size = 0;
     ret->len = 0;
+    ret->seq = 0;
     ret->cmp = cmp;
     
     /* put NULL item at pos 0 that is our stop marker */
     make_bigger(ret, 1);
-    ret->tab[0] = NULL;
+    ret->tab[0] = gw_malloc(sizeof(**ret->tab));
+    ret->tab[0]->item = NULL;
+    ret->tab[0]->seq = ret->seq++;
     ret->len++;
     
     return ret;
@@ -174,13 +200,15 @@
 
 void gw_prioqueue_destroy(gw_prioqueue_t *queue, void(*item_destroy)(void*))
 {
+    long i;
+
     if (queue == NULL)
         return;
     
-    if (item_destroy != NULL) {
-        void *item;
-        while((item = gw_prioqueue_remove(queue)) != NULL)
-            item_destroy(item);
+    for (i = 0; i < queue->len; i++) {
+        if (item_destroy != NULL && queue->tab[i]->item != NULL)
+            item_destroy(queue->tab[i]->item);
+        gw_free(queue->tab[i]);
     }
     mutex_destroy(queue->mutex);
     pthread_cond_destroy(&queue->nonempty);
@@ -211,7 +239,9 @@
     
     queue_lock(queue);
     make_bigger(queue, 1);
-    queue->tab[queue->len] = item;
+    queue->tab[queue->len] = gw_malloc(sizeof(**queue->tab));
+    queue->tab[queue->len]->item = item;
+    queue->tab[queue->len]->seq = queue->seq++;
     upheap(queue, queue->len);
     queue->len++;
     pthread_cond_signal(&queue->nonempty);
@@ -227,7 +257,7 @@
     
     queue_lock(queue);
     for (i = 1; i < queue->len; i++)
-        fn(queue->tab[i], i - 1);
+        fn(queue->tab[i]->item, i - 1);
     queue_unlock(queue);
 }
 
@@ -243,7 +273,8 @@
         queue_unlock(queue);
         return NULL;
     }
-    ret = queue->tab[1];
+    ret = queue->tab[1]->item;
+    gw_free(queue->tab[1]);
     queue->tab[1] = queue->tab[--queue->len];
     downheap(queue, 1);
     queue_unlock(queue);
@@ -260,7 +291,7 @@
     
     queue_lock(queue);
     if (queue->len > 1)
-        ret = queue->tab[1];
+        ret = queue->tab[1]->item;
     else
         ret = NULL;
     queue_unlock(queue);
@@ -282,7 +313,8 @@
         queue->mutex->owner = gwthread_self();
     }
     if (queue->len > 1) {
-        ret = queue->tab[1];
+        ret = queue->tab[1]->item;
+        gw_free(queue->tab[1]);
         queue->tab[1] = queue->tab[--queue->len];
         downheap(queue, 1);
     } else {
lmpx.com only provides a reader for public news (NNTP) servers. It is not affiliated with the servers or forums shown here and is not responsible for the content of articles, which is written by their respective authors.