sorted_queue.[hc]: sorted_queue_pop and sorted_queue_pop_all

src/sorted_queue.c

@@ -70,17 +70,48 @@ void sorted_queue_push_ex(struct sorted_queue *sq, void *data, bool *notify)
             }
             fc_list_add_after(dlink, current);
             *notify = false;
-
-            if (count >= 10) {
-                logInfo("================== compare count: %d ==========", count);
-            }
         }
     }
 
     PTHREAD_MUTEX_UNLOCK(&sq->lcp.lock);
 }
 
-void sorted_queue_pop_ex(struct sorted_queue *sq, void *less_equal,
+void *sorted_queue_pop_ex(struct sorted_queue *sq,
+        void *less_equal, const bool blocked)
+{
+    void *data;
+    struct fc_list_head *current;
+
+    PTHREAD_MUTEX_LOCK(&sq->lcp.lock);
+    do {
+        if (fc_list_empty(&sq->head)) {
+            if (!blocked) {
+                data = NULL;
+                break;
+            }
+
+            pthread_cond_wait(&sq->lcp.cond,
+                    &sq->lcp.lock);
+            if (fc_list_empty(&sq->head)) {
+                data = NULL;
+                break;
+            }
+        }
+
+        current = sq->head.next;
+        data = FC_SORTED_QUEUE_DATA_PTR(sq, current);
+        if (sq->compare_func(data, less_equal) <= 0) {
+            fc_list_del_init(current);
+        } else {
+            data = NULL;
+        }
+    } while (0);
+    PTHREAD_MUTEX_UNLOCK(&sq->lcp.lock);
+
+    return data;
+}
+
+void sorted_queue_pop_all_ex(struct sorted_queue *sq, void *less_equal,
         struct fc_list_head *head, const bool blocked)
 {
     struct fc_list_head *current;

src/sorted_queue.h

@@ -78,14 +78,23 @@ static inline void sorted_queue_push_silence(
     sorted_queue_push_ex(sq, data, &notify);
 }
 
-void sorted_queue_pop_ex(struct sorted_queue *sq, void *less_equal,
+void *sorted_queue_pop_ex(struct sorted_queue *sq,
+        void *less_equal, const bool blocked);
+
+#define sorted_queue_pop(sq, less_equal) \
+    sorted_queue_pop_ex(sq, less_equal, true)
+
+#define sorted_queue_try_pop(sq, less_equal) \
+    sorted_queue_pop_ex(sq, less_equal, false)
+
+void sorted_queue_pop_all_ex(struct sorted_queue *sq, void *less_equal,
         struct fc_list_head *head, const bool blocked);
 
-#define sorted_queue_pop(sq, less_equal, head) \
-    sorted_queue_pop_ex(sq, less_equal, head, true)
+#define sorted_queue_pop_all(sq, less_equal, head) \
+    sorted_queue_pop_all_ex(sq, less_equal, head, true)
 
-#define sorted_queue_try_pop(sq, less_equal, head) \
-    sorted_queue_pop_ex(sq, less_equal, head, false)
+#define sorted_queue_try_pop_all(sq, less_equal, head) \
+    sorted_queue_pop_all_ex(sq, less_equal, head, false)
 
 static inline bool sorted_queue_empty(struct sorted_queue *sq)
 {

src/tests/test_sorted_queue.c

@@ -72,13 +72,12 @@ static void test1()
     struct fc_list_head head;
 
     set_rand_numbers(1);
-
     for (i=0; i<COUNT; i++) {
         sorted_queue_push_silence(&sq, numbers + i);
     }
 
     less_equal.n = COUNT;
-    sorted_queue_try_pop(&sq, &less_equal, &head);
+    sorted_queue_try_pop_all(&sq, &less_equal, &head);
     assert(sorted_queue_empty(&sq));
 
     i = 0;
@@ -91,7 +90,7 @@ static void test1()
     }
     assert(i == COUNT);
 
-    sorted_queue_try_pop(&sq, &less_equal, &head);
+    sorted_queue_try_pop_all(&sq, &less_equal, &head);
     assert(fc_list_empty(&head));
 }
 
@@ -105,17 +104,16 @@ static void test2()
     struct fc_list_head head;
 
     set_rand_numbers(MULTIPLE);
-
     for (i=0; i<COUNT; i++) {
         sorted_queue_push_silence(&sq, numbers + i);
     }
 
     less_equal.n = 0;
-    sorted_queue_try_pop(&sq, &less_equal, &head);
+    sorted_queue_try_pop_all(&sq, &less_equal, &head);
     assert(fc_list_empty(&head));
 
     less_equal.n = COUNT;
-    sorted_queue_try_pop(&sq, &less_equal, &head);
+    sorted_queue_try_pop_all(&sq, &less_equal, &head);
     assert(!sorted_queue_empty(&sq));
 
     i = 0;
@@ -128,7 +126,7 @@ static void test2()
     }
 
     less_equal.n = 2 * COUNT + 1;
-    sorted_queue_try_pop(&sq, &less_equal, &head);
+    sorted_queue_try_pop_all(&sq, &less_equal, &head);
     assert(sorted_queue_empty(&sq));
     fc_list_for_each_entry (number, &head, dlink) {
         n = i++ * MULTIPLE + 1;
@@ -140,6 +138,30 @@ static void test2()
     assert(i == COUNT);
 }
 
+static void test3()
+{
+    int i;
+    DoubleLinkNumber less_equal;
+    DoubleLinkNumber *number;
+
+    set_rand_numbers(1);
+    for (i=0; i<COUNT; i++) {
+        sorted_queue_push_silence(&sq, numbers + i);
+    }
+
+    less_equal.n = COUNT;
+    for (i=1; i<=COUNT; i++) {
+        number = sorted_queue_try_pop(&sq, &less_equal);
+        assert(number != NULL);
+        if (i != number->n) {
+            fprintf(stderr, "i: %d != value: %d\n", i, number->n);
+            break;
+        }
+    }
+
+    assert(sorted_queue_try_pop(&sq, &less_equal) == NULL);
+}
+
 int main(int argc, char *argv[])
 {
     int result;
@@ -160,9 +182,9 @@ int main(int argc, char *argv[])
 
     test1();
     test2();
+    test3();
 
     end_time = get_current_time_ms();
     printf("pass OK, time used: %"PRId64" ms\n", end_time - start_time);
     return 0;
 }
-