TT#170203 use linked list to manage call IDs

Using an array in this context (to return a list of call IDs) is mostly
pointless as it wastes memory and incurs the additional overhead of
having to initialise the array and an extra layer of copying strings
around. This also eliminates the auxiliary type `med_callid_t` and
ultimately allows us to dynamically append to the list of call IDs
without having to reallocate the array.

Use g_strdup for string allocation as it guarantees a non-NULL return
value.

Change-Id: Iae6c97f80c216352ab36de89d361f09ee355b6c8

mediator.c

@@ -178,10 +178,10 @@ static uint64_t mediator_calc_runtime(struct timeval *tv_start, struct timeval *
 /**********************************************************************/
 int main(int argc, char **argv)
 {
-    med_callid_t *mysql_callids;
-    med_callid_t *redis_callids;
+    GQueue mysql_callids = G_QUEUE_INIT;
+    GQueue redis_callids = G_QUEUE_INIT;
     med_entry_t *mysql_records, *redis_records;
-    uint64_t mysql_id_count, redis_id_count, mysql_rec_count, redis_rec_count, i;
+    uint64_t mysql_rec_count, redis_rec_count;
     uint64_t cdr_count, last_count;
     int maprefresh;
     struct medmysql_batches *batches;
@@ -191,6 +191,7 @@ int main(int argc, char **argv)
     struct timeval tv_start, tv_stop;
     uint64_t runtime;
 #endif    
+    gboolean success;
 
     openlog(MEDIATOR_SYSLOG_NAME, LOG_PID|LOG_NDELAY, LOG_DAEMON);
     atexit(mediator_exit);
@@ -303,22 +304,24 @@ int main(int argc, char **argv)
             }
         }
 
-        mysql_id_count = redis_id_count = mysql_rec_count = redis_rec_count = cdr_count = 0;
+        g_queue_clear_full(&mysql_callids, g_free);
+        g_queue_clear_full(&redis_callids, g_free);
+        mysql_rec_count = redis_rec_count = cdr_count = 0;
         last_count = mediator_count;
 
-        mysql_callids = medmysql_fetch_callids(&mysql_id_count);
-        if(!mysql_callids && mysql_id_count) {
+        success = medmysql_fetch_callids(&mysql_callids);
+        if(!success) {
             L_ERROR("Failed to fetch callids from MySQL\n");
             break;
         }
 
-        redis_callids = medredis_fetch_callids(&redis_id_count);
-        if (!redis_callids && redis_id_count) {
+        success = medredis_fetch_callids(&redis_callids);
+        if (!success) {
             L_ERROR("Failed to fetch callids from Redis\n");
             break;
         }
 
-        if (!mysql_id_count && !redis_id_count) {
+        if (!mysql_callids.length && !redis_callids.length) {
             L_DEBUG("No callids found, going idle\n");
             goto idle;
         }
@@ -331,17 +334,18 @@ int main(int argc, char **argv)
 
         //////////////// mysql handling //////////////////
 
-        L_DEBUG("Processing %"PRIu64" mysql accounting record group(s).", mysql_id_count);
-        for(i = 0; i < mysql_id_count && !mediator_shutdown; ++i)
+        L_DEBUG("Processing %u mysql accounting record group(s).", mysql_callids.length);
+        for(GList *l = mysql_callids.head; l && !mediator_shutdown; l = l->next)
         {
+            char *mysql_callid = l->data;
 #ifdef WITH_TIME_CALC
             gettimeofday(&tv_start, NULL);
 #endif
 
-            if(medmysql_fetch_records(&(mysql_callids[i]), &mysql_records, &mysql_rec_count, 1) != 0)
+            if(medmysql_fetch_records(mysql_callid, &mysql_records, &mysql_rec_count, 1) != 0)
                 goto out;
 
-            if(medredis_fetch_records(&(mysql_callids[i]), &redis_records, &redis_rec_count) == 0
+            if(medredis_fetch_records(mysql_callid, &redis_records, &redis_rec_count) == 0
                     && redis_rec_count)
             {
                 med_entry_t *mysql_new_records;
@@ -365,7 +369,7 @@ int main(int argc, char **argv)
 
             if (!are_records_complete && !do_intermediate)
             {
-                L_DEBUG("Found incomplete call with cid '%s', skipping...\n", mysql_callids[i].value);
+                L_DEBUG("Found incomplete call with cid '%s', skipping...\n", mysql_callid);
                 free(mysql_records);
                 continue;
             }
@@ -384,27 +388,25 @@ int main(int argc, char **argv)
             gettimeofday(&tv_stop, NULL);
             runtime = mediator_calc_runtime(&tv_start, &tv_stop);
             L_INFO("Runtime for mysql record group was %"PRIu64" us.", runtime);
-            L_INFO("CDR creation rate for mysql record group was %f CDR/sec", (double)mysql_id_count/runtime * 1000000);
+            L_INFO("CDR creation rate for mysql record group was %f CDR/sec", (double)mysql_callids.length/runtime * 1000000);
 #endif
         }
-        if (mysql_id_count) {
-            free(mysql_callids);
-        }
 
 
         //////////////// redis handling //////////////////
 
-        L_DEBUG("Processing %"PRIu64" redis accounting record group(s).", redis_id_count);
-        for(i = 0; i < redis_id_count && !mediator_shutdown; ++i)
+        L_DEBUG("Processing %u redis accounting record group(s).", redis_callids.length);
+        for(GList *l = redis_callids.head; l && !mediator_shutdown; l = l->next)
         {
+            char *redis_callid = l->data;
 #ifdef WITH_TIME_CALC
             gettimeofday(&tv_start, NULL);
 #endif
 
-            if(medredis_fetch_records(&(redis_callids[i]), &redis_records, &redis_rec_count) != 0)
+            if(medredis_fetch_records(redis_callid, &redis_records, &redis_rec_count) != 0)
                 goto out;
 
-            if(medmysql_fetch_records(&(redis_callids[i]), &mysql_records, &mysql_rec_count, 0) == 0
+            if(medmysql_fetch_records(redis_callid, &mysql_records, &mysql_rec_count, 0) == 0
                     && mysql_rec_count)
             {
                 med_entry_t *redis_new_records;
@@ -428,12 +430,12 @@ int main(int argc, char **argv)
 
             if (!are_records_complete && !do_intermediate)
             {
-                L_DEBUG("Found incomplete call with cid '%s', skipping...\n", redis_callids[i].value);
+                L_DEBUG("Found incomplete call with cid '%s', skipping...\n", redis_callid);
                 free(redis_records);
                 continue;
             }
 
-            L_DEBUG("process cdr with cid '%s' and %"PRIu64" records\n", redis_callids[i].value, redis_rec_count);
+            L_DEBUG("process cdr with cid '%s' and %"PRIu64" records\n", redis_callid, redis_rec_count);
 
             if (redis_rec_count) {
                 if(cdr_process_records(redis_records, redis_rec_count, &cdr_count, batches, do_intermediate) != 0) {
@@ -449,12 +451,9 @@ int main(int argc, char **argv)
             gettimeofday(&tv_stop, NULL);
             runtime = mediator_calc_runtime(&tv_start, &tv_stop);
             L_INFO("Runtime for redis record group was %"PRIu64" us.", runtime);
-            L_INFO("CDR creation rate for redis record group was %f CDR/sec", (double)redis_id_count/runtime * 1000000);
+            L_INFO("CDR creation rate for redis record group was %f CDR/sec", (double)redis_callids.length/runtime * 1000000);
 #endif
         }
-        if (redis_id_count) {
-            free(redis_callids);
-        }
 
         //////////////// end //////////////////
 
@@ -463,8 +462,8 @@ int main(int argc, char **argv)
 
         gettimeofday(&loop_tv_stop, NULL);
         loop_runtime = mediator_calc_runtime(&loop_tv_start, &loop_tv_stop);
-        L_INFO("Runtime for loop processing %"PRIu64" callids and generating %"PRIu64" CDRs was %"PRIu64" us.",
-            mysql_id_count + redis_id_count, mediator_count - last_count, loop_runtime);
+        L_INFO("Runtime for loop processing %u callids and generating %"PRIu64" CDRs was %"PRIu64" us.",
+            mysql_callids.length + redis_callids.length, mediator_count - last_count, loop_runtime);
         L_INFO("Total CDR creation rate %f CDR/sec", (double)(mediator_count - last_count)/loop_runtime * 1000000);
 
 idle:
@@ -484,6 +483,8 @@ out:
     L_INFO("Shutting down.");
     sd_notify(0, "STOPPING=1\n");
 
+    g_queue_clear_full(&mysql_callids, g_free);
+    g_queue_clear_full(&redis_callids, g_free);
     mediator_destroy_maps();
     mediator_destroy_caches();
     medmysql_cleanup();

mediator.h

@@ -91,10 +91,6 @@ typedef struct {
     uint8_t timed_out;
 } med_entry_t;
 
-typedef struct {
-    char value[256];
-} med_callid_t;
-
 typedef struct {
     char *str_value;
     time_t created;

medmysql.c

@@ -627,16 +627,11 @@ out:
 }
 
 /**********************************************************************/
-med_callid_t *medmysql_fetch_callids(uint64_t *count)
+gboolean medmysql_fetch_callids(GQueue *output)
 {
     MYSQL_RES *res;
     MYSQL_ROW row;
     /* char query[1024] = ""; */
-    size_t callid_size;
-    uint64_t i = 0;
-    med_callid_t *callids = NULL;
-
-    *count = (uint64_t) -1; /* non-zero count and return of NULL == error */
 
     /* g_strlcpy(query, MED_CALLID_QUERY, sizeof(query)); */
 
@@ -646,68 +641,51 @@ med_callid_t *medmysql_fetch_callids(uint64_t *count)
     {
         L_CRITICAL("Error getting acc callids: %s",
                 mysql_error(med_handler->m));
-        return NULL;
+        return FALSE;
     }
 
     res = mysql_store_result(med_handler->m);
-    *count = mysql_num_rows(res);
-    if(*count == 0)
-    {
-        goto out;
-    }
-
-    callid_size = sizeof(med_callid_t) * (*count);
-    callids = malloc(callid_size);
-    if(callids == NULL)
-    {
-        L_CRITICAL("Error allocating callid memory: %s", strerror(errno));
-        free(callids);
-        callids = NULL;
-        goto out;
-    }
-
-    memset(callids, '\0', callid_size);
 
     while((row = mysql_fetch_row(res)) != NULL)
     {
-        med_callid_t *c = &callids[i++];
         if(row[0] == NULL)
         {
-            g_strlcpy(c->value, "0", sizeof(c->value));
+            g_queue_push_tail(output, g_strdup("0"));
         } else {
-            g_strlcpy(c->value, row[0], sizeof(c->value));
+            g_queue_push_tail(output, g_strdup(row[0]));
         }
 
         /*L_DEBUG("callid[%"PRIu64"]='%s'", i, c->value);*/
 
         if (check_shutdown()) {
-            free(callids);
-            return NULL;
+            mysql_free_result(res);
+            g_queue_clear_full(output, free);
+            return FALSE;
         }
     }
 
-out:
     mysql_free_result(res);
-    return callids;
+    return TRUE;
 }
 
 /**********************************************************************/
-int medmysql_fetch_records(med_callid_t *callid,
+int medmysql_fetch_records(char *callid,
         med_entry_t **entries, uint64_t *count, int warn_empty)
 {
     MYSQL_RES *res;
     MYSQL_ROW row;
-    char query[strlen(MED_FETCH_QUERY) + sizeof(callid->value) * 7 + 1];
+    size_t callid_len = strlen(callid);
+    char query[strlen(MED_FETCH_QUERY) + callid_len * 7 + 1];
     size_t entry_size;
     uint64_t i = 0;
     int ret = 0;
     int len;
 
-    char esc_callid[sizeof(((med_callid_t*)0)->value)*2+1];
+    char esc_callid[callid_len*2+1];
 
     *count = 0;
 
-    mysql_real_escape_string(med_handler->m, esc_callid, callid->value, strlen(callid->value));
+    mysql_real_escape_string(med_handler->m, esc_callid, callid, callid_len);
 
     len = snprintf(query, sizeof(query), MED_FETCH_QUERY,
         esc_callid,
@@ -722,7 +700,7 @@ int medmysql_fetch_records(med_callid_t *callid,
     if(medmysql_query_wrapper(med_handler, query, len) != 0)
     {
         L_CRITICAL("Error getting acc records for callid '%s': %s",
-                callid->value, mysql_error(med_handler->m));
+                callid, mysql_error(med_handler->m));
         return -1;
     }
 
@@ -732,7 +710,7 @@ int medmysql_fetch_records(med_callid_t *callid,
     {
         if (warn_empty)
             L_CRITICAL("No records found for callid '%s'!",
-                    callid->value);
+                    callid);
         ret = -1;
         goto out;
     }

medmysql.h

@@ -52,8 +52,8 @@ struct medmysql_call_stat_info_t {
 
 int medmysql_init(void);
 void medmysql_cleanup(void);
-med_callid_t *medmysql_fetch_callids(uint64_t *count);
-int medmysql_fetch_records(med_callid_t *callid, med_entry_t **entries, uint64_t *count, int warn_empty);
+gboolean medmysql_fetch_callids(GQueue *output);
+int medmysql_fetch_records(char *callid, med_entry_t **entries, uint64_t *count, int warn_empty);
 int medmysql_trash_entries(const char *callid, struct medmysql_batches *);
 int medmysql_backup_entries(const char *callid, struct medmysql_batches *);
 int medmysql_delete_entries(const char *callid, struct medmysql_batches *);

medredis.c

@@ -39,11 +39,6 @@ typedef struct {
     unsigned int append_counter;
 } medredis_con_t;
 
-typedef struct {
-    med_callid_t *entries;
-    uint64_t index;
-} medredis_cidlist_t;
-
 static medredis_con_t *con = NULL;
 static char medredis_srem_key_lua[41]; // sha-1 hex string
 
@@ -482,29 +477,16 @@ void medredis_cleanup() {
 }
 
 /**********************************************************************/
-static void medredis_add_cid(void *key, void *val, void *data) {
-    medredis_cidlist_t *list = (medredis_cidlist_t*)data;
-    (void)val;
-    L_DEBUG("Adding cid %s at index %"PRIu64"\n", (char*)key, list->index);
-    g_strlcpy(list->entries[list->index].value, (char*)key, sizeof(list->entries[list->index].value));
-    list->index++;
-}
-
-/**********************************************************************/
-med_callid_t *medredis_fetch_callids(uint64_t *count) {
+gboolean medredis_fetch_callids(GQueue *output) {
     unsigned int cursor = 0;
     size_t i = 0;
     redisReply *reply = NULL;
     char *cmd = "SSCAN acc:meth::INVITE %u COUNT 1000";
-    med_callid_t *entries = NULL;
     GHashTable *cid_table;
-    medredis_cidlist_t cid_list;
     char buffer[256];
     char *tmp;
 
-    cid_table = g_hash_table_new_full(g_str_hash, g_str_equal, free, NULL);
-
-    *count = 0;
+    cid_table = g_hash_table_new(g_str_hash, g_str_equal);
 
     do {
         snprintf(buffer, sizeof(buffer), cmd, cursor);
@@ -557,7 +539,7 @@ med_callid_t *medredis_fetch_callids(uint64_t *count) {
 
 
             // strip leading "acc:entry::" and trailing ":<time_hires>:<branch_id>"
-            cid = strdup(entry->str + strlen("acc:entry::"));
+            cid = g_strdup(entry->str + strlen("acc:entry::"));
             tmp = strrchr(cid, ':');
             if (tmp) {
                 *tmp = '\0';
@@ -574,34 +556,24 @@ med_callid_t *medredis_fetch_callids(uint64_t *count) {
             }
 
             if (g_hash_table_insert(cid_table, cid, cid)) {
-                (*count)++;
+                g_queue_push_tail(output, cid);
+            } else {
+                free(cid);
             }
 
         }
         medredis_free_reply(&reply);
-    } while (cursor && (*count) < 200000);
+    } while (cursor && output->length < 200000);
 
-    *count = g_hash_table_size(cid_table);
-    if (*count) {
-        entries = (med_callid_t*)malloc(*count * sizeof(*entries));
-        if (!entries) {
-            L_ERROR("Failed to allocate memory for callid list\n");
-            goto err;
-        }
-        cid_list.entries = entries;
-        cid_list.index = 0;
-        g_hash_table_foreach(cid_table, medredis_add_cid, &cid_list);
-    }
     g_hash_table_destroy(cid_table);
 
-    return entries;
+    return TRUE;
 
 err:
     if (reply)
         freeReplyObject(reply);
-    *count = (uint64_t) -1;
     g_hash_table_destroy(cid_table);
-    return NULL;
+    return FALSE;
 }    
 
 /**********************************************************************/
@@ -639,7 +611,7 @@ static void medredis_free_keys_list(gpointer data) {
 }
 
 /**********************************************************************/
-int medredis_fetch_records(med_callid_t *callid,
+int medredis_fetch_records(char *callid,
         med_entry_t **entries, uint64_t *count) {
 
 
@@ -673,24 +645,24 @@ int medredis_fetch_records(med_callid_t *callid,
 
 
     memset(cids, 0, sizeof(cids));
-    cids[0] = strdup(callid->value);
+    cids[0] = strdup(callid);
     if (!cids[0]) {
         L_ERROR("Failed to allocate memory for callid\n");
         goto err;
     }
-    snprintf(buffer, sizeof(buffer), "%s%s", callid->value, PBXSUFFIX);
+    snprintf(buffer, sizeof(buffer), "%s%s", callid, PBXSUFFIX);
     cids[1] = strdup(buffer);
     if (!cids[1]) {
         L_ERROR("Failed to allocate memory for callid with pbxsuffix\n");
         goto err;
     }
-    snprintf(buffer, sizeof(buffer), "%s%s", callid->value, XFERSUFFIX);
+    snprintf(buffer, sizeof(buffer), "%s%s", callid, XFERSUFFIX);
     cids[2] = strdup(buffer);
     if (!cids[2]) {
         L_ERROR("Failed to allocate memory for callid with xfersuffix\n");
         goto err;
     }
-    snprintf(buffer, sizeof(buffer), "%s%s%s", callid->value, PBXSUFFIX, XFERSUFFIX);
+    snprintf(buffer, sizeof(buffer), "%s%s%s", callid, PBXSUFFIX, XFERSUFFIX);
     cids[3] = strdup(buffer);
     if (!cids[3]) {
         L_ERROR("Failed to allocate memory for callid with pbxsuffix and xfersuffix\n");
@@ -707,7 +679,7 @@ int medredis_fetch_records(med_callid_t *callid,
         snprintf(buffer, sizeof(buffer), "acc:cid::%s", cid);
         cid_set_argv[1] = buffer;
         if (medredis_append_command_argv(cid_set_argc, cid_set_argv, 1) != 0) {
-            L_ERROR("Failed to append redis command to fetch entries for cid '%s'\n", callid->value);
+            L_ERROR("Failed to append redis command to fetch entries for cid '%s'\n", callid);
             goto err;
         }
         free(cid);
@@ -716,7 +688,7 @@ int medredis_fetch_records(med_callid_t *callid,
 
     for (i = 0; i < 4; ++i) {
         if (medredis_get_reply(&reply) != 0) {
-            L_ERROR("Failed to get redis reply for command to fetch entries for cid '%s'\n", callid->value);
+            L_ERROR("Failed to get redis reply for command to fetch entries for cid '%s'\n", callid);
             goto err;    
         }
         medredis_check_reply("smembers for cid", reply, err);
@@ -729,14 +701,14 @@ int medredis_fetch_records(med_callid_t *callid,
         }
         if (!reply->elements) {
             medredis_free_reply(&reply);
-            L_DEBUG("No matching entries for cid '%s' at suffix idx %lu\n", callid->value, i);
+            L_DEBUG("No matching entries for cid '%s' at suffix idx %lu\n", callid, i);
             continue;
         }
 
         for (size_t j = 0; j < reply->elements; ++j) {
             char *key = strdup(reply->element[j]->str);
             if (!key) {
-                L_ERROR("Failed to allocate memory for redis key (cid '%s')\n", callid->value);
+                L_ERROR("Failed to allocate memory for redis key (cid '%s')\n", callid);
                 goto err;
             }
             L_DEBUG("Putting key '%s' to keys list\n", key);
@@ -755,7 +727,7 @@ int medredis_fetch_records(med_callid_t *callid,
         
         L_DEBUG("Fetching next reply record, query key was '%s'\n", key);
         if (medredis_get_reply(&reply) != 0) {
-            L_ERROR("Failed to get reply from redis (cid '%s')\n", callid->value);
+            L_ERROR("Failed to get reply from redis (cid '%s')\n", callid);
             goto err;
         }
         if (!reply)
@@ -763,9 +735,9 @@ int medredis_fetch_records(med_callid_t *callid,
         medredis_check_reply("get reply", reply, err);
         medredis_dump_reply(reply);
 
-        e = medredis_reply_to_entry(reply, callid->value, key);
+        e = medredis_reply_to_entry(reply, callid, key);
         if (!e) {
-            L_WARNING("Failed to convert redis reply to entry (cid '%s')\n", callid->value);
+            L_WARNING("Failed to convert redis reply to entry (cid '%s')\n", callid);
             medredis_free_reply(&reply);
             continue;
         }
@@ -779,7 +751,7 @@ int medredis_fetch_records(med_callid_t *callid,
 
     *entries = (med_entry_t*)malloc(*count * sizeof(med_entry_t));
     if (!*entries) {
-        L_ERROR("Failed to allocate memory for entries (cid '%s')\n", callid->value);
+        L_ERROR("Failed to allocate memory for entries (cid '%s')\n", callid);
         goto err;
     }
     i = 0;

medredis.h

@@ -8,8 +8,8 @@
 
 int medredis_init(void);
 void medredis_cleanup(void);
-med_callid_t *medredis_fetch_callids(uint64_t *count);
-int medredis_fetch_records(med_callid_t *callid, med_entry_t **entries, uint64_t *count);
+gboolean medredis_fetch_callids(GQueue *output);
+int medredis_fetch_records(char *callid, med_entry_t **entries, uint64_t *count);
 int medredis_trash_entries(med_entry_t *records, uint64_t count);
 int medredis_backup_entries(med_entry_t *records, uint64_t count);