Remove nested depth limitation. (#797)

* Remove nested depth limitation.

This commit removes the nested multi-bulk depth limitation of 7.
We do this by switching to pointer to pointer indirection and
growing the stack in chunks when needed.

See: #794, #421

read.c

@@ -46,6 +46,9 @@
 #include "sds.h"
 #include "win32.h"
 
+/* Initial size of our nested reply stack and how much we grow it when needd */
+#define REDIS_READER_STACK_SIZE 9
+
 static void __redisReaderSetError(redisReader *r, int type, const char *str) {
     size_t len;
 
@@ -243,8 +246,8 @@ static void moveToNextTask(redisReader *r) {
             return;
         }
 
-        cur = &(r->rstack[r->ridx]);
+        cur = r->task[r->ridx];
-        prv = &(r->rstack[r->ridx-1]);
+        prv = r->task[r->ridx-1];
         assert(prv->type == REDIS_REPLY_ARRAY ||
                prv->type == REDIS_REPLY_MAP ||
                prv->type == REDIS_REPLY_SET);
@@ -262,7 +265,7 @@ static void moveToNextTask(redisReader *r) {
 }
 
 static int processLineItem(redisReader *r) {
-    redisReadTask *cur = &(r->rstack[r->ridx]);
+    redisReadTask *cur = r->task[r->ridx];
     void *obj;
     char *p;
     int len;
@@ -344,7 +347,7 @@ static int processLineItem(redisReader *r) {
 }
 
 static int processBulkItem(redisReader *r) {
-    redisReadTask *cur = &(r->rstack[r->ridx]);
+    redisReadTask *cur = r->task[r->ridx];
     void *obj = NULL;
     char *p, *s;
     long long len;
@@ -407,19 +410,43 @@ static int processBulkItem(redisReader *r) {
     return REDIS_ERR;
 }
 
+static int redisReaderGrow(redisReader *r) {
+    redisReadTask **aux;
+    int newlen;
+
+    /* Grow our stack size */
+    newlen = r->tasks + REDIS_READER_STACK_SIZE;
+    aux = realloc(r->task, sizeof(*r->task) * newlen);
+    if (aux == NULL)
+        goto oom;
+
+    r->task = aux;
+
+    /* Allocate new tasks */
+    for (; r->tasks < newlen; r->tasks++) {
+        r->task[r->tasks] = calloc(1, sizeof(**r->task));
+        if (r->task[r->tasks] == NULL)
+            goto oom;
+    }
+
+    return REDIS_OK;
+oom:
+    __redisReaderSetErrorOOM(r);
+    return REDIS_ERR;
+}
+
 /* Process the array, map and set types. */
 static int processAggregateItem(redisReader *r) {
-    redisReadTask *cur = &(r->rstack[r->ridx]);
+    redisReadTask *cur = r->task[r->ridx];
     void *obj;
     char *p;
     long long elements;
     int root = 0, len;
 
     /* Set error for nested multi bulks with depth > 7 */
-    if (r->ridx == 8) {
+    if (r->ridx == r->tasks - 1) {
-        __redisReaderSetError(r,REDIS_ERR_PROTOCOL,
+        if (redisReaderGrow(r) == REDIS_ERR)
-            "No support for nested multi bulk replies with depth > 7");
+            return REDIS_ERR;
-        return REDIS_ERR;
     }
 
     if ((p = readLine(r,&len)) != NULL) {
@@ -467,12 +494,12 @@ static int processAggregateItem(redisReader *r) {
                 cur->elements = elements;
                 cur->obj = obj;
                 r->ridx++;
-                r->rstack[r->ridx].type = -1;
+                r->task[r->ridx]->type = -1;
-                r->rstack[r->ridx].elements = -1;
+                r->task[r->ridx]->elements = -1;
-                r->rstack[r->ridx].idx = 0;
+                r->task[r->ridx]->idx = 0;
-                r->rstack[r->ridx].obj = NULL;
+                r->task[r->ridx]->obj = NULL;
-                r->rstack[r->ridx].parent = cur;
+                r->task[r->ridx]->parent = cur;
-                r->rstack[r->ridx].privdata = r->privdata;
+                r->task[r->ridx]->privdata = r->privdata;
             } else {
                 moveToNextTask(r);
             }
@@ -487,7 +514,7 @@ static int processAggregateItem(redisReader *r) {
 }
 
 static int processItem(redisReader *r) {
-    redisReadTask *cur = &(r->rstack[r->ridx]);
+    redisReadTask *cur = r->task[r->ridx];
     char *p;
 
     /* check if we need to read type */
@@ -562,23 +589,46 @@ redisReader *redisReaderCreateWithFunctions(redisReplyObjectFunctions *fn) {
     if (r == NULL)
         return NULL;
 
-    r->fn = fn;
     r->buf = sdsempty();
-    r->maxbuf = REDIS_READER_MAX_BUF;
+    if (r->buf == NULL)
-    if (r->buf == NULL) {
+        goto oom;
-        free(r);
+
-        return NULL;
+    r->task = calloc(REDIS_READER_STACK_SIZE, sizeof(*r->task));
+    if (r->task == NULL)
+        goto oom;
+
+    for (; r->tasks < REDIS_READER_STACK_SIZE; r->tasks++) {
+        r->task[r->tasks] = calloc(1, sizeof(**r->task));
+        if (r->task[r->tasks] == NULL)
+            goto oom;
     }
 
+    r->fn = fn;
+    r->maxbuf = REDIS_READER_MAX_BUF;
+
     r->ridx = -1;
     return r;
+
+oom:
+    redisReaderFree(r);
+    return NULL;
 }
 
 void redisReaderFree(redisReader *r) {
     if (r == NULL)
         return;
+
     if (r->reply != NULL && r->fn && r->fn->freeObject)
         r->fn->freeObject(r->reply);
+
+    /* We know r->task[i] is allocatd if i < r->tasks */
+    for (int i = 0; i < r->tasks; i++) {
+        free(r->task[i]);
+    }
+
+    if (r->task)
+        free(r->task);
+
     sdsfree(r->buf);
     free(r);
 }
@@ -630,12 +680,12 @@ int redisReaderGetReply(redisReader *r, void **reply) {
 
     /* Set first item to process when the stack is empty. */
     if (r->ridx == -1) {
-        r->rstack[0].type = -1;
+        r->task[0]->type = -1;
-        r->rstack[0].elements = -1;
+        r->task[0]->elements = -1;
-        r->rstack[0].idx = -1;
+        r->task[0]->idx = -1;
-        r->rstack[0].obj = NULL;
+        r->task[0]->obj = NULL;
-        r->rstack[0].parent = NULL;
+        r->task[0]->parent = NULL;
-        r->rstack[0].privdata = r->privdata;
+        r->task[0]->privdata = r->privdata;
         r->ridx = 0;
     }
 

read.h

@@ -97,7 +97,9 @@ typedef struct redisReader {
     size_t len; /* Buffer length */
     size_t maxbuf; /* Max length of unused buffer */
 
-    redisReadTask rstack[9];
+    redisReadTask **task;
+    int tasks;
+
     int ridx; /* Index of current read task */
     void *reply; /* Temporary reply pointer */
 

test.c

@@ -316,7 +316,7 @@ static void test_append_formatted_commands(struct config config) {
 
 static void test_reply_reader(void) {
     redisReader *reader;
-    void *reply;
+    void *reply, *root;
     int ret;
     int i;
 
@@ -340,16 +340,26 @@ static void test_reply_reader(void) {
               strcasecmp(reader->errstr,"Protocol error, got \"@\" as reply type byte") == 0);
     redisReaderFree(reader);
 
-    test("Set error on nested multi bulks with depth > 7: ");
     reader = redisReaderCreate();
-
+    test("Can handle arbitrarily nested multi-bulks: ");
-    for (i = 0; i < 9; i++) {
+    for (i = 0; i < 128; i++) {
-        redisReaderFeed(reader,(char*)"*1\r\n",4);
+        redisReaderFeed(reader,(char*)"*1\r\n", 4);
     }
+    redisReaderFeed(reader,(char*)"$6\r\nLOLWUT\r\n",12);
+    ret = redisReaderGetReply(reader,&reply);
+    root = reply; /* Keep track of the root reply */
+    test_cond(ret == REDIS_OK &&
+        ((redisReply*)reply)->type == REDIS_REPLY_ARRAY &&
+        ((redisReply*)reply)->elements == 1);
 
-    ret = redisReaderGetReply(reader,NULL);
+    test("Can parse arbitrarily nested multi-bulks correctly: ");
-    test_cond(ret == REDIS_ERR &&
+    while(i--) {
-              strncasecmp(reader->errstr,"No support for",14) == 0);
+        assert(reply != NULL && ((redisReply*)reply)->type == REDIS_REPLY_ARRAY);
+        reply = ((redisReply*)reply)->element[0];
+    }
+    test_cond(((redisReply*)reply)->type == REDIS_REPLY_STRING &&
+        !memcmp(((redisReply*)reply)->str, "LOLWUT", 6));
+    freeReplyObject(root);
     redisReaderFree(reader);
 
     test("Correctly parses LLONG_MAX: ");