db_redis: re-do appended commands after reconnect

Since hiredis sends out the pipelined command only after
calling redisGetReply(), we need a mechamism to queue up all
appended commands, so we can re-queue them once the connection
is re-established.

This commit introduces a redis command queue used for pipelined
commands to re-do all appended commands in case of a connection drop
(which typically happens in db-mode write-through with very low
traffic, where redis would close the connection due to inactivity).

src/modules/db_redis/redis_connection.c

@@ -24,6 +24,100 @@
 #include "redis_connection.h"
 #include "redis_table.h"
 
+static void print_query(redis_key_t *query) {
+    LM_DBG("Query dump:\n");
+    for (redis_key_t *k = query; k; k = k->next) {
+        LM_DBG("  %s\n", k->key.s);
+    }
+}
+
+static int db_redis_push_query(km_redis_con_t *con, redis_key_t *query) {
+
+    redis_command_t *cmd = NULL;
+    redis_command_t *tmp = NULL;
+    redis_key_t *new_query = NULL;
+
+    if (!query)
+        return 0;
+
+    cmd = (redis_command_t*)pkg_malloc(sizeof(redis_command_t));
+    if (!cmd) {
+        LM_ERR("Failed to allocate memory for redis command\n");
+        goto err;
+    }
+
+    // duplicate query, as original one might be free'd after being
+    // appended
+    while(query) {
+         if (db_redis_key_add_str(&new_query, &query->key) != 0) {
+            LM_ERR("Failed to duplicate query\n");
+            goto err;
+        }
+        query = query->next;
+    }
+
+    cmd->query = new_query;
+    cmd->next = NULL;
+
+    if (!con->command_queue) {
+        con->command_queue = cmd;
+    } else {
+        tmp = con->command_queue;
+        while (tmp->next)
+            tmp = tmp->next;
+        tmp->next = cmd;
+    }
+
+    return 0;
+
+err:
+    if (new_query) {
+        db_redis_key_free(&new_query);
+    }
+    if (cmd) {
+        pkg_free(cmd);
+    }
+    return -1;
+}
+
+static redis_key_t* db_redis_shift_query(km_redis_con_t *con) {
+    redis_command_t *cmd;
+    redis_key_t *query;
+
+    query = NULL;
+    cmd = con->command_queue;
+
+    if (cmd) {
+        query = cmd->query;
+        con->command_queue = cmd->next;
+        pkg_free(cmd);
+    }
+
+    return query;
+}
+
+static redis_key_t* db_redis_pop_query(km_redis_con_t *con) {
+    redis_command_t **current;
+    redis_command_t *prev;
+    redis_key_t *query;
+
+    current = &con->command_queue;
+    if (!*current)
+        return NULL;
+
+    do {
+        query = (*current)->query;
+        prev = *current;
+        *current = (*current)->next;
+    } while (*current);
+
+    prev->next = NULL;
+    pkg_free(*current);
+    *current = NULL;
+
+    return query;
+}
+
 int db_redis_connect(km_redis_con_t *con) {
     struct timeval tv;
     redisReply *reply;
@@ -189,16 +283,6 @@ void db_redis_free_connection(struct pool_con* con) {
     pkg_free(_c);
 }
 
-
-static void print_query(redis_key_t *query) {
-	redis_key_t *k;
-
-    LM_DBG("Query dump:\n");
-    for (k = query; k; k = k->next) {
-        LM_DBG("  %s\n", k->key.s);
-    }
-}
-
 void *db_redis_command_argv(km_redis_con_t *con, redis_key_t *query) {
     char **argv = NULL;
     int argc;
@@ -229,12 +313,17 @@ void *db_redis_command_argv(km_redis_con_t *con, redis_key_t *query) {
     return reply;
 }
 
-int db_redis_append_command_argv(km_redis_con_t *con, redis_key_t *query) {
+int db_redis_append_command_argv(km_redis_con_t *con, redis_key_t *query, int queue) {
     char **argv = NULL;
     int ret, argc;
 
     print_query(query);
 
+    if (queue > 0 && db_redis_push_query(con, query) != 0) {
+        LM_ERR("Failed to queue redis command\n");
+        return -1;
+    }
+
     argc = db_redis_key_list2arr(query, &argv);
     if (argc < 0) {
         LM_ERR("Failed to allocate memory for query array\n");
@@ -243,6 +332,10 @@ int db_redis_append_command_argv(km_redis_con_t *con, redis_key_t *query) {
     LM_DBG("query has %d args\n", argc);
 
     ret = redisAppendCommandArgv(con->con, argc, (const char**)argv, NULL);
+
+    // this should actually never happen, because if all replies
+    // are properly consumed for the previous command, it won't send
+    // out a new query until redisGetReply is called
     if (con->con->err == REDIS_ERR_EOF) {
         if (db_redis_connect(con) != 0) {
             LM_ERR("Failed to reconnect to redis db\n");
@@ -264,22 +357,40 @@ int db_redis_append_command_argv(km_redis_con_t *con, redis_key_t *query) {
 
 int db_redis_get_reply(km_redis_con_t *con, void **reply) {
     int ret;
+    redis_key_t *query;
 
     *reply = NULL;
     ret = redisGetReply(con->con, reply);
     if (con->con->err == REDIS_ERR_EOF) {
+        LM_DBG("redis connection is gone, try reconnect\n");
+        con->append_counter = 0;
         if (db_redis_connect(con) != 0) {
             LM_ERR("Failed to reconnect to redis db\n");
             if (con->con) {
                 redisFree(con->con);
                 con->con = NULL;
             }
-            return ret;
+        }
+        // take commands from oldest to newest and re-do again,
+        // but don't queue them once again in retry-mode
+        while ((query = db_redis_shift_query(con))) {
+            LM_DBG("re-queueing appended command\n");
+            if (db_redis_append_command_argv(con, query, 0) != 0) {
+                LM_ERR("Failed to re-queue redis command");
+                return -1;
+            }
+            db_redis_key_free(&query);
         }
         ret = redisGetReply(con->con, reply);
-    }
-    if (!con->con->err)
+        if (con->con->err != REDIS_ERR_EOF) {
+            con->append_counter--;
+        }
+    } else {
+        LM_DBG("get_reply successful, popping query\n");
+        query = db_redis_pop_query(con);
+        db_redis_key_free(&query);
         con->append_counter--;
+    }
     return ret;
 }
 
@@ -292,6 +403,7 @@ void db_redis_free_reply(redisReply **reply) {
 
 void db_redis_consume_replies(km_redis_con_t *con) {
     redisReply *reply = NULL;
+    redis_key_t *query;
     while (con->append_counter > 0 && !con->con->err) {
         LM_DBG("consuming outstanding reply %u", con->append_counter);
         db_redis_get_reply(con, (void**)&reply);
@@ -300,4 +412,8 @@ void db_redis_consume_replies(km_redis_con_t *con) {
             reply = NULL;
         }
     }
+    while ((query = db_redis_shift_query(con))) {
+        LM_DBG("consuming queued command\n");
+        db_redis_key_free(&query);
+    }
 }

src/modules/db_redis/redis_connection.h

@@ -44,12 +44,20 @@
     } \
 } while(0);
 
+typedef struct redis_key redis_key_t;
+
+typedef struct redis_command {
+    redis_key_t *query;
+    struct redis_command *next;
+} redis_command_t;
+
 typedef struct km_redis_con {
     struct db_id* id;
     unsigned int ref;
     struct pool_con* next;
 
     redisContext *con;
+    redis_command_t *command_queue;
     unsigned int append_counter;
     struct str_hash_table tables;
 } km_redis_con_t;
@@ -65,7 +73,7 @@ void db_redis_free_connection(struct pool_con* con);
 
 int db_redis_connect(km_redis_con_t *con);
 void *db_redis_command_argv(km_redis_con_t *con, redis_key_t *query);
-int db_redis_append_command_argv(km_redis_con_t *con, redis_key_t *query);
+int db_redis_append_command_argv(km_redis_con_t *con, redis_key_t *query, int queue);
 int db_redis_get_reply(km_redis_con_t *con, void **reply);
 void db_redis_consume_replies(km_redis_con_t *con);
 void db_redis_free_reply(redisReply **reply);

src/modules/db_redis/redis_dbase.c

@@ -1088,7 +1088,7 @@ static int db_redis_perform_query(const db1_con_t* _h, km_redis_con_t *con, cons
             LM_ERR("Failed to add key name to list\n");
             goto error;
         }
-        if (db_redis_append_command_argv(con, query_v) != REDIS_OK) {
+        if (db_redis_append_command_argv(con, query_v, 1) != REDIS_OK) {
             LM_ERR("Failed to append redis command\n");
             goto error;
         }
@@ -1127,7 +1127,7 @@ static int db_redis_perform_query(const db1_con_t* _h, km_redis_con_t *con, cons
             }
         }
 
-        if (db_redis_append_command_argv(con, query_v) != REDIS_OK) {
+        if (db_redis_append_command_argv(con, query_v, 1) != REDIS_OK) {
             LM_ERR("Failed to append redis command\n");
             goto error;
         }
@@ -1451,7 +1451,7 @@ static int db_redis_perform_update(const db1_con_t* _h, km_redis_con_t *con, con
             LM_ERR("Failed to add key name to pre-update exists query\n");
             goto error;
         }
-        if (db_redis_append_command_argv(con, query_v) != REDIS_OK) {
+        if (db_redis_append_command_argv(con, query_v, 1) != REDIS_OK) {
             LM_ERR("Failed to append redis command\n");
             goto error;
         }
@@ -1486,7 +1486,7 @@ static int db_redis_perform_update(const db1_con_t* _h, km_redis_con_t *con, con
             }
         }
 
-        if (db_redis_append_command_argv(con, query_v) != REDIS_OK) {
+        if (db_redis_append_command_argv(con, query_v, 1) != REDIS_OK) {
             LM_ERR("Failed to append redis command\n");
             goto error;
         }
@@ -1601,7 +1601,7 @@ static int db_redis_perform_update(const db1_con_t* _h, km_redis_con_t *con, con
             pkg_free(v.s);
         }
         update_queries++;
-        if (db_redis_append_command_argv(con, query_v) != REDIS_OK) {
+        if (db_redis_append_command_argv(con, query_v, 1) != REDIS_OK) {
             LM_ERR("Failed to append redis command\n");
             goto error;
         }

src/modules/db_redis/redis_dbase.h

@@ -77,7 +77,7 @@ const db_key_t* _uk, const db_val_t* _uv, const int _n, const int _un);
  * Just like insert, but replace the row if it exists
  */
 int db_redis_replace(const db1_con_t* handle, const db_key_t* keys, const db_val_t* vals,
-		const int n, const int _un, const int _m);
+        const int n, const int _un, const int _m);
 
 /*
  * Store name of table that will be used by

src/modules/db_redis/redis_table.h

@@ -41,7 +41,7 @@ struct redis_type {
 
 typedef struct redis_table redis_table_t;
 struct redis_table {
-	int version;
+    int version;
     redis_key_t *entry_keys;
     redis_type_t *types;
     struct str_hash_table columns;