Changeset 6099 for pjproject

Timestamp:

Nov 6, 2019 6:17:53 AM (4 years ago)

Author:

riza

Message:

Close #2249: Use sorted linked list for timer implementation.

Location:

pjproject/trunk/pjlib

Files:

• include/pj/config.h (modified) (3 diffs)
• include/pj/timer.h (modified) (3 diffs)
• src/pj/timer.c (modified) (20 diffs)
• src/pjlib-test/timer.c (modified) (5 diffs)

pjproject/trunk/pjlib/include/pj/config.h

@@ -544 +544 @@
 
 /**
- * Enable timer heap debugging facility. When this is enabled, application
- * can call pj_timer_heap_dump() to show the contents of the timer heap
+ * Enable timer debugging facility. When this is enabled, application
+ * can call pj_timer_heap_dump() to show the contents of the timer
  * along with the source location where the timer entries were scheduled.
  * See https://trac.pjsip.org/repos/ticket/1527 for more info.
@@ -557 +557 @@
 
 /**
- * If enabled, the timer heap will keep internal copies of the timer entries.
- * This will increase the robustness and stability of the timer heap (against
+ * If enabled, the timer will keep internal copies of the timer entries.
+ * This will increase the robustness and stability of the timer (against
  * accidental modification or premature deallocation of the timer entries) and
  * makes it easier to troubleshoot any timer related issues, with the overhead
@@ -571 +571 @@
  * Default: 1 (enabled)
  */
-#ifndef PJ_TIMER_HEAP_USE_COPY
-#  define PJ_TIMER_HEAP_USE_COPY    1
-#endif
-
+#ifndef PJ_TIMER_USE_COPY
+#  define PJ_TIMER_USE_COPY    1
+#endif
+
+
+/**
+ * If enabled, the timer use sorted linked list instead of binary heap tree
+ * structure. Note that using sorted linked list is intended for debugging
+ * purposes and will hamper performance significantly when scheduling large
+ * number of entries.
+ *
+ * Default: 0 (Use binary heap tree)
+ */
+#ifndef PJ_TIMER_USE_LINKED_LIST
+#  define PJ_TIMER_USE_LINKED_LIST    0
+#endif
 
 /**

pjproject/trunk/pjlib/include/pj/timer.h

@@ -26 +26 @@
 #include <pj/types.h>
 #include <pj/lock.h>
+
+#if PJ_TIMER_USE_LINKED_LIST
+#  include <pj/list.h>
+#endif
 
 PJ_BEGIN_DECL
@@ -89 +93 @@
 typedef struct pj_timer_entry
 {
+#if !PJ_TIMER_USE_COPY && PJ_TIMER_USE_LINKED_LIST
+    /**
+    * Standard list members.
+    */
+    PJ_DECL_LIST_MEMBER(struct pj_timer_entry);
+#endif
+
     /** 
      * User data to be associated with this entry. 
@@ -114 +125 @@
     pj_timer_id_t _timer_id;
 
-#if !PJ_TIMER_HEAP_USE_COPY
+#if !PJ_TIMER_USE_COPY
     /** 
      * The future time when the timer expires, which the value is updated

pjproject/trunk/pjlib/src/pj/timer.c

@@ -51 +51 @@
  * the timer heap will simply remove the destroyed entry (and print log)
  * and resume normally.
- * This setting only works if PJ_TIMER_HEAP_USE_COPY is enabled.
+ * This setting only works if PJ_TIMER_USE_COPY is enabled.
  */
-#define ASSERT_IF_ENTRY_DESTROYED (PJ_TIMER_HEAP_USE_COPY? 0: 0)
+#define ASSERT_IF_ENTRY_DESTROYED (PJ_TIMER_USE_COPY? 0: 0)
 
 
@@ -63 +63 @@
 };
 
-#if PJ_TIMER_HEAP_USE_COPY
+#if PJ_TIMER_USE_COPY
 
 /* Duplicate/copy of the timer entry. */
 typedef struct pj_timer_entry_dup
 {
+#if PJ_TIMER_USE_LINKED_LIST
+    /**
+    * Standard list members.
+    */
+    PJ_DECL_LIST_MEMBER(struct pj_timer_entry_dup);
+#endif
+
     /**
      * The duplicate copy.
@@ -141 +148 @@
      */
     pj_timer_entry_dup **heap;
+
+#if PJ_TIMER_USE_LINKED_LIST
+    /**
+    * If timer heap uses linked list, then this will represent the head of
+    * the list.
+    */
+    pj_timer_entry_dup head_list;
+#endif
 
     /**
@@ -289 +304 @@
 {
     pj_timer_entry_dup *removed_node = ht->heap[slot];
-    
+
     // Return this timer id to the freelist.
     push_freelist( ht, GET_FIELD(removed_node, _timer_id) );
@@ -321 +336 @@
     GET_FIELD(removed_node, _timer_id) = -1;
 
+#if !PJ_TIMER_USE_LINKED_LIST
     // Only try to reheapify if we're not deleting the last entry.
     
@@ -344 +360 @@
         }
     }
-    
+#else
+    pj_list_erase(removed_node);
+#endif
+
     return removed_node;
 }
@@ -352 +371 @@
     // All the containers will double in size from max_size_
     size_t new_size = ht->max_size * 2;
+#if PJ_TIMER_USE_COPY
     pj_timer_entry_dup *new_timer_dups = 0;
+#endif
     pj_timer_id_t *new_timer_ids;
     pj_size_t i;
     pj_timer_entry_dup **new_heap = 0;
+
+#if PJ_TIMER_USE_LINKED_LIST
+    pj_timer_entry_dup *tmp_dup = NULL;
+    pj_timer_entry_dup *new_dup;
+#endif
     
     PJ_LOG(6,(THIS_FILE, "Growing heap size from %d to %d",
@@ -366 +392 @@
         return PJ_ENOMEM;
     
-#if PJ_TIMER_HEAP_USE_COPY
+#if PJ_TIMER_USE_COPY
     // Grow the array of timer copies.
     
@@ -387 +413 @@
     memcpy(new_heap, ht->heap, ht->max_size * sizeof(pj_timer_entry *));
 #endif
+
+#if PJ_TIMER_USE_LINKED_LIST
+    tmp_dup = ht->head_list.next;
+    pj_list_init(&ht->head_list);
+    for (; tmp_dup != &ht->head_list; tmp_dup = tmp_dup->next)
+    {
+        int slot = ht->timer_ids[GET_FIELD(tmp_dup, _timer_id)];
+        new_dup = new_heap[slot];
+        pj_list_push_back(&ht->head_list, new_dup);
+    }
+#endif
+
     ht->heap = new_heap;
     
@@ -417 +455 @@
     pj_timer_entry_dup *timer_copy;
 
+#if PJ_TIMER_USE_LINKED_LIST
+    pj_timer_entry_dup *tmp_node = NULL;
+#endif
+
     if (ht->cur_size + 2 >= ht->max_size) {
         pj_status_t status = grow_heap(ht);
@@ -424 +466 @@
 
     timer_copy = GET_TIMER(ht, new_node);
-#if PJ_TIMER_HEAP_USE_COPY    
+#if PJ_TIMER_USE_COPY
     // Create a duplicate of the timer entry.
     pj_bzero(timer_copy, sizeof(*timer_copy));
@@ -430 +472 @@
     timer_copy->entry = new_node;
 #endif
+
+#if PJ_TIMER_USE_LINKED_LIST
+    pj_list_init(timer_copy);
+#endif
+
     timer_copy->_timer_value = *future_time;
 
-    reheap_up( ht, timer_copy, ht->cur_size, HEAP_PARENT(ht->cur_size));
+#if !PJ_TIMER_USE_LINKED_LIST
+    reheap_up(ht, timer_copy, ht->cur_size, HEAP_PARENT(ht->cur_size));
+#else
+    if (ht->cur_size == 0) {
+        pj_list_push_back(&ht->head_list, timer_copy);
+    } else if (PJ_TIME_VAL_GTE(*future_time,
+                               ht->head_list.prev->_timer_value))
+    {
+        /* Insert the max value to the end of the list. */
+        pj_list_insert_before(&ht->head_list, timer_copy);
+    } else {
+        tmp_node = ht->head_list.next;
+        while (tmp_node->next != &ht->head_list &&
+               PJ_TIME_VAL_GT(*future_time, tmp_node->_timer_value))
+        {
+            tmp_node = tmp_node->next;
+        }
+        if (PJ_TIME_VAL_LT(*future_time, tmp_node->_timer_value)) {
+            pj_list_insert_before(tmp_node, timer_copy);
+        } else {
+            pj_list_insert_after(tmp_node, timer_copy);
+        }
+    }
+    copy_node(ht, new_node->_timer_id-1, timer_copy);
+#endif
     ht->cur_size++;
 
@@ -547 +618 @@
         return PJ_ENOMEM;
 
-#if PJ_TIMER_HEAP_USE_COPY
+#if PJ_TIMER_USE_COPY
     // Create the timer entry copies array.
     ht->timer_dups = (pj_timer_entry_dup*)
@@ -567 +638 @@
         ht->timer_ids[i] = -((pj_timer_id_t) (i + 1));
 
+#if PJ_TIMER_USE_LINKED_LIST
+    pj_list_init(&ht->head_list);
+#endif
+
     *p_heap = ht;
     return PJ_SUCCESS;
@@ -610 +685 @@
     entry->user_data = user_data;
     entry->cb = cb;
-#if !PJ_TIMER_HEAP_USE_COPY
+#if !PJ_TIMER_USE_COPY
     entry->_grp_lock = NULL;
 #endif
@@ -772 +847 @@
 {
     pj_time_val now;
+    pj_time_val min_time_node = {0,0};
     unsigned count;
+    pj_timer_id_t slot = 0;
 
     PJ_ASSERT_RETURN(ht, 0);
@@ -786 +863 @@
     pj_gettickcount(&now);
 
+    if (ht->cur_size) {
+#if PJ_TIMER_USE_LINKED_LIST
+        slot = ht->timer_ids[GET_FIELD(ht->head_list.next, _timer_id)];
+#endif
+        min_time_node = ht->heap[slot]->_timer_value;
+    }
+
     while ( ht->cur_size && 
-            PJ_TIME_VAL_LTE(ht->heap[0]->_timer_value, now) &&
+            PJ_TIME_VAL_LTE(min_time_node, now) &&
             count < ht->max_entries_per_poll ) 
     {
-        pj_timer_entry_dup *node = remove_node(ht, 0);
+        pj_timer_entry_dup *node = remove_node(ht, slot);
         pj_timer_entry *entry = GET_ENTRY(node);
         /* Avoid re-use of this timer until the callback is done. */
@@ -835 +919 @@
         /* Now, the timer is really free for re-use. */
         ///push_freelist(ht, node_timer_id);
+
+        if (ht->cur_size) {
+#if PJ_TIMER_USE_LINKED_LIST
+            slot = ht->timer_ids[GET_FIELD(ht->head_list.next, _timer_id)];
+#endif
+            min_time_node = ht->heap[slot]->_timer_value;
+        }
     }
     if (ht->cur_size && next_delay) {
@@ -880 +971 @@
 
     if (ht->cur_size) {
+#if PJ_TIMER_USE_LINKED_LIST
+        pj_timer_entry_dup *tmp_dup;
+#else
         unsigned i;
+#endif
         pj_time_val now;
 
@@ -889 +984 @@
         pj_gettickcount(&now);
 
-        for (i=0; i<(unsigned)ht->cur_size; ++i) {
+#if !PJ_TIMER_USE_LINKED_LIST
+        for (i=0; i<(unsigned)ht->cur_size; ++i)
+        {
             pj_timer_entry_dup *e = ht->heap[i];
+#else
+        for (tmp_dup = ht->head_list.next; tmp_dup != &ht->head_list;
+             tmp_dup = tmp_dup->next)
+        {
+            pj_timer_entry_dup *e = tmp_dup;
+#endif
+
             pj_time_val delta;
 

pjproject/trunk/pjlib/src/pjlib-test/timer.c

@@ -208 +208 @@
  */
 #define RANDOMIZED_TEST 1
-#define SIMULATE_CRASH  PJ_TIMER_HEAP_USE_COPY
+#define SIMULATE_CRASH  PJ_TIMER_USE_COPY
 
 #if RANDOMIZED_TEST
@@ -231 +231 @@
 #define ST_ENTRY_GROUP_LOCK_COUNT   1
 
+#define BT_ENTRY_COUNT 100000
+#define BT_ENTRY_SHOW_START 100
+#define BT_ENTRY_SHOW_MULT 10
+#define BT_REPEAT_RANDOM_TEST 4
+#define BT_REPEAT_INC_TEST 4
 
 struct thread_param
@@ -269 +274 @@
 static void dummy_callback(pj_timer_heap_t *ht, pj_timer_entry *e)
 {
+    PJ_UNUSED_ARG(ht);
     PJ_LOG(4,("test", "dummy callback called %p %p", e, e->user_data));
 }
@@ -747 +753 @@
 }
 
+static int get_random_delay()
+{
+    return pj_rand() % BT_ENTRY_COUNT;
+}
+
+static int get_next_delay(int delay)
+{
+    return ++delay;
+}
+
+typedef enum BENCH_TEST_TYPE {
+    RANDOM_SCH = 0,
+    RANDOM_CAN = 1,
+    INCREMENT_SCH = 2,
+    INCREMENT_CAN = 3
+} BENCH_TEST_TYPE;
+
+static char *get_test_name(BENCH_TEST_TYPE test_type) {
+    switch (test_type) {
+    case RANDOM_SCH:
+    case INCREMENT_SCH:
+        return "schedule";
+    case RANDOM_CAN:
+    case INCREMENT_CAN:
+        return "cancel";
+    }
+    return "undefined";
+}
+
+static void *get_format_num(unsigned n, char *out)
+{
+    int c;
+    char buf[64];
+    char *p;
+
+    pj_ansi_snprintf(buf, 64, "%d", n);
+    c = 2 - pj_ansi_strlen(buf) % 3;
+    for (p = buf; *p != 0; ++p) {
+       *out++ = *p;
+       if (c == 1) {
+           *out++ = ',';
+       }
+       c = (c + 1) % 3;
+    }
+    *--out = 0;
+    return out;
+}
+
+static void print_bench(BENCH_TEST_TYPE test_type, pj_timestamp time_freq,
+                        pj_timestamp time_start, int start_idx, int end_idx)
+{
+    char start_idx_str[64];
+    char end_idx_str[64];
+    char num_req_str[64];
+    unsigned num_req;
+    pj_timestamp t2;
+
+    pj_get_timestamp(&t2);
+    pj_sub_timestamp(&t2, &time_start);
+
+    num_req = (unsigned)(time_freq.u64 * (end_idx-start_idx) / t2.u64);
+    if (test_type == RANDOM_CAN || test_type == INCREMENT_CAN) {
+        start_idx = BT_ENTRY_COUNT - start_idx;
+        end_idx = BT_ENTRY_COUNT - end_idx;
+    }
+    get_format_num(start_idx, start_idx_str);
+    get_format_num(end_idx, end_idx_str);
+    get_format_num(num_req, num_req_str);
+
+    PJ_LOG(3, (THIS_FILE, "    Entries %s-%s: %s %s ent/sec",
+               start_idx_str, end_idx_str, get_test_name(test_type),
+               num_req_str));
+}
+
+static int bench_test(pj_timer_heap_t *timer,
+                      pj_timer_entry *entries,
+                      pj_timestamp freq,
+                      BENCH_TEST_TYPE test_type)
+{
+    pj_timestamp t1, t2;
+    unsigned mult = BT_ENTRY_SHOW_START;
+    int i, j;
+
+    pj_get_timestamp(&t1);
+    t2.u64 = 0;
+    /*Schedule random entry.*/
+    for (i=0, j=0; j < BT_ENTRY_COUNT; ++j) {
+        pj_time_val delay = { 0 };
+        pj_status_t status;
+
+        switch (test_type) {
+        case RANDOM_SCH:
+            delay.msec = get_random_delay();
+            break;
+        case INCREMENT_SCH:
+            delay.msec = get_next_delay(delay.msec);
+            break;
+        }
+
+        if (test_type == RANDOM_SCH || test_type == INCREMENT_SCH) {
+            pj_timer_entry_init(&entries[j], 0, NULL, &dummy_callback);
+
+            status = pj_timer_heap_schedule(timer, &entries[j], &delay);
+            if (status != PJ_SUCCESS) {
+                app_perror("...error: unable to schedule timer entry", status);
+                return -50;
+            }
+        } else if (test_type == RANDOM_CAN || test_type == INCREMENT_CAN) {
+            unsigned num_ent = pj_timer_heap_cancel(timer, &entries[j]);
+            if (num_ent == 0) {
+                PJ_LOG(3, ("test", "...error: unable to cancel timer entry"));
+                return -60;
+            }
+        } else {
+            return -70;
+        }
+
+        if (j && (j % mult) == 0) {
+            print_bench(test_type, freq, t1, i, j);
+
+            i = j+1;
+            pj_get_timestamp(&t1);
+            mult *= BT_ENTRY_SHOW_MULT;
+        }
+    }
+    if (j > 0 && ((j-1) % mult != 0)) {
+        print_bench(test_type, freq, t1, i, j);
+    }
+    return 0;
+}
+
+static int timer_bench_test(void)
+{
+    pj_pool_t *pool = NULL;
+    pj_timer_heap_t *timer = NULL;
+    pj_status_t status;
+    int err=0;
+    pj_timer_entry *entries = NULL;
+    pj_timestamp freq;
+    int i;
+
+    PJ_LOG(3,("test", "...Benchmark test"));
+
+    status = pj_get_timestamp_freq(&freq);
+    if (status != PJ_SUCCESS) {
+        PJ_LOG(3,("test", "...error: unable to get timestamp freq"));
+        err = -10;
+        goto on_return;
+    }
+
+    pool = pj_pool_create( mem, NULL, 128, 128, NULL);
+    if (!pool) {
+        PJ_LOG(3,("test", "...error: unable to create pool"));
+        err = -20;
+        goto on_return;
+    }
+
+    /* Create timer heap.*/
+    status = pj_timer_heap_create(pool, BT_ENTRY_COUNT/64, &timer);
+    if (status != PJ_SUCCESS) {
+        app_perror("...error: unable to create timer heap", status);
+        err = -30;
+        goto on_return;
+    }
+
+    /* Create and schedule timer entries */
+    entries = (pj_timer_entry*)pj_pool_calloc(pool, BT_ENTRY_COUNT,
+                                              sizeof(*entries));
+    if (!entries) {
+        err = -40;
+        goto on_return;
+    }
+
+    PJ_LOG(3,("test", "....random scheduling/cancelling test.."));
+    for (i = 0; i < BT_REPEAT_RANDOM_TEST; ++i) {
+        PJ_LOG(3,("test", "    test %d of %d..", i+1, BT_REPEAT_RANDOM_TEST));
+        err = bench_test(timer, entries, freq, RANDOM_SCH);
+        if (err < 0)
+            goto on_return;
+
+        err = bench_test(timer, entries, freq, RANDOM_CAN);
+        if (err < 0)
+            goto on_return;
+    }
+
+    PJ_LOG(3,("test", "....increment scheduling/cancelling test.."));
+    for (i = 0; i < BT_REPEAT_INC_TEST; ++i) {
+        PJ_LOG(3,("test", "    test %d of %d..", i+1, BT_REPEAT_INC_TEST));
+        err = bench_test(timer, entries, freq, INCREMENT_SCH);
+        if (err < 0)
+            goto on_return;
+
+        err = bench_test(timer, entries, freq, INCREMENT_CAN);
+        if (err < 0)
+            goto on_return;
+    }
+ on_return:
+    PJ_LOG(3,("test", "...Cleaning up resources"));
+    if (pool)
+        pj_pool_safe_release(&pool);
+    return err;
+}
+
 int timer_test()
 {
@@ -756 +965 @@
 
     rc = timer_stress_test();
+    if (rc != 0)
+        return rc;
+
+    rc = timer_bench_test();
     if (rc != 0)
         return rc;