Refactor reply buildup for speed on large multi bulk replies

src/networking.c

@@ -1,5 +1,4 @@
 #include "redis.h"
-
 #include <sys/uio.h>
 
 void *dupClientReplyValue(void *o) {
@@ -12,7 +11,16 @@ int listMatchObjects(void *a, void *b) {
 }
 
 redisClient *createClient(int fd) {
-    redisClient *c = zmalloc(sizeof(*c));
+    redisClient *c;
+
+    /* Make sure to allocate a multiple of the page size to prevent wasting
+     * memory. A page size of 4096 is assumed here. We need to compensate
+     * for the zmalloc overhead of sizeof(size_t) bytes. */
+    size_t size = 8192-sizeof(size_t);
+    redisAssert(size > sizeof(redisClient));
+    c = zmalloc(size);
+    c->buflen = size-sizeof(redisClient);
+    c->bufpos = 0;
 
     anetNonBlock(NULL,fd);
     anetTcpNoDelay(NULL,fd);
@@ -53,70 +61,118 @@ redisClient *createClient(int fd) {
     return c;
 }
 
-void addReply(redisClient *c, robj *obj) {
+int _ensureFileEvent(redisClient *c) {
-    if (listLength(c->reply) == 0 &&
+    if (c->bufpos == 0 && listLength(c->reply) == 0 &&
         (c->replstate == REDIS_REPL_NONE ||
          c->replstate == REDIS_REPL_ONLINE) &&
         aeCreateFileEvent(server.el, c->fd, AE_WRITABLE,
-        sendReplyToClient, c) == AE_ERR) return;
+        sendReplyToClient, c) == AE_ERR) return REDIS_ERR;
+    return REDIS_OK;
+}
 
-    if (server.vm_enabled && obj->storage != REDIS_VM_MEMORY) {
+void _addReplyObjectToList(redisClient *c, robj *obj) {
-        obj = dupStringObject(obj);
+    redisAssert(obj->type == REDIS_STRING &&
-        obj->refcount = 0; /* getDecodedObject() will increment the refcount */
+                obj->encoding == REDIS_ENCODING_RAW);
+    listAddNodeTail(c->reply,obj);
+}
+
+void _ensureBufferInReplyList(redisClient *c) {
+    sds buffer = sdsnewlen(NULL,REDIS_REPLY_CHUNK_SIZE);
+    sdsupdatelen(buffer); /* sdsnewlen expects non-empty string */
+    listAddNodeTail(c->reply,createObject(REDIS_REPLY_NODE,buffer));
+}
+
+void _addReplyStringToBuffer(redisClient *c, char *s, size_t len) {
+    size_t available = 0;
+    redisAssert(len < REDIS_REPLY_CHUNK_THRESHOLD);
+    if (listLength(c->reply) > 0) {
+        robj *o = listNodeValue(listLast(c->reply));
+
+        /* Make sure to append to a reply node with enough bytes available. */
+        if (o->type == REDIS_REPLY_NODE) available = sdsavail(o->ptr);
+        if (o->type != REDIS_REPLY_NODE || len > available) {
+            _ensureBufferInReplyList(c);
+            _addReplyStringToBuffer(c,s,len);
+        } else {
+            o->ptr = sdscatlen(o->ptr,s,len);
+        }
+    } else {
+        available = c->buflen-c->bufpos;
+        if (len > available) {
+            _ensureBufferInReplyList(c);
+            _addReplyStringToBuffer(c,s,len);
+        } else {
+            memcpy(c->buf+c->bufpos,s,len);
+            c->bufpos += len;
+        }
+    }
+}
+
+void addReply(redisClient *c, robj *obj) {
+    if (_ensureFileEvent(c) != REDIS_OK) return;
+    if (server.vm_enabled && obj->storage != REDIS_VM_MEMORY) {
+        /* Returns a new object with refcount 1 */
+        obj = dupStringObject(obj);
+    } else {
+        /* This increments the refcount. */
+        obj = getDecodedObject(obj);
+    }
+
+    if (sdslen(obj->ptr) < REDIS_REPLY_CHUNK_THRESHOLD) {
+        _addReplyStringToBuffer(c,obj->ptr,sdslen(obj->ptr));
+        decrRefCount(obj);
+    } else {
+        _addReplyObjectToList(c,obj);
     }
-    listAddNodeTail(c->reply,getDecodedObject(obj));
 }
 
 void addReplySds(redisClient *c, sds s) {
-    robj *o = createObject(REDIS_STRING,s);
+    if (_ensureFileEvent(c) != REDIS_OK) return;
-    addReply(c,o);
+    if (sdslen(s) < REDIS_REPLY_CHUNK_THRESHOLD) {
-    decrRefCount(o);
+        _addReplyStringToBuffer(c,s,sdslen(s));
+        sdsfree(s);
+    } else {
+        _addReplyObjectToList(c,createObject(REDIS_STRING,s));
+    }
+}
+
+void addReplyString(redisClient *c, char *s, size_t len) {
+    if (_ensureFileEvent(c) != REDIS_OK) return;
+    if (len < REDIS_REPLY_CHUNK_THRESHOLD) {
+        _addReplyStringToBuffer(c,s,len);
+    } else {
+        _addReplyObjectToList(c,createStringObject(s,len));
+    }
 }
 
 void addReplyDouble(redisClient *c, double d) {
-    char buf[128];
+    char dbuf[128], sbuf[128];
-
+    int dlen, slen;
-    snprintf(buf,sizeof(buf),"%.17g",d);
+    dlen = snprintf(dbuf,sizeof(dbuf),"%.17g",d);
-    addReplySds(c,sdscatprintf(sdsempty(),"$%lu\r\n%s\r\n",
+    slen = snprintf(sbuf,sizeof(sbuf),"$%d\r\n%s\r\n",dlen,dbuf);
-        (unsigned long) strlen(buf),buf));
+    addReplyString(c,sbuf,slen);
 }
 
-void addReplyLongLong(redisClient *c, long long ll) {
+void _addReplyLongLong(redisClient *c, long long ll, char prefix) {
     char buf[128];
-    size_t len;
+    int len;
-
+    buf[0] = prefix;
-    if (ll == 0) {
-        addReply(c,shared.czero);
-        return;
-    } else if (ll == 1) {
-        addReply(c,shared.cone);
-        return;
-    }
-    buf[0] = ':';
     len = ll2string(buf+1,sizeof(buf)-1,ll);
     buf[len+1] = '\r';
     buf[len+2] = '\n';
-    addReplySds(c,sdsnewlen(buf,len+3));
+    addReplyString(c,buf,len+3);
+}
+
+void addReplyLongLong(redisClient *c, long long ll) {
+    _addReplyLongLong(c,ll,':');
 }
 
 void addReplyUlong(redisClient *c, unsigned long ul) {
-    char buf[128];
+    _addReplyLongLong(c,(long long)ul,':');
-    size_t len;
-
-    if (ul == 0) {
-        addReply(c,shared.czero);
-        return;
-    } else if (ul == 1) {
-        addReply(c,shared.cone);
-        return;
-    }
-    len = snprintf(buf,sizeof(buf),":%lu\r\n",ul);
-    addReplySds(c,sdsnewlen(buf,len));
 }
 
 void addReplyBulkLen(redisClient *c, robj *obj) {
-    size_t len, intlen;
+    size_t len;
-    char buf[128];
 
     if (obj->encoding == REDIS_ENCODING_RAW) {
         len = sdslen(obj->ptr);
@@ -133,11 +189,7 @@ void addReplyBulkLen(redisClient *c, robj *obj) {
             len++;
         }
     }
-    buf[0] = '$';
+    _addReplyLongLong(c,len,'$');
-    intlen = ll2string(buf+1,sizeof(buf)-1,(long long)len);
-    buf[intlen+1] = '\r';
-    buf[intlen+2] = '\n';
-    addReplySds(c,sdsnewlen(buf,intlen+3));
 }
 
 void addReplyBulk(redisClient *c, robj *obj) {
@@ -287,34 +339,6 @@ void freeClient(redisClient *c) {
     zfree(c);
 }
 
-#define GLUEREPLY_UP_TO (1024)
-static void glueReplyBuffersIfNeeded(redisClient *c) {
-    int copylen = 0;
-    char buf[GLUEREPLY_UP_TO];
-    listNode *ln;
-    listIter li;
-    robj *o;
-
-    listRewind(c->reply,&li);
-    while((ln = listNext(&li))) {
-        int objlen;
-
-        o = ln->value;
-        objlen = sdslen(o->ptr);
-        if (copylen + objlen <= GLUEREPLY_UP_TO) {
-            memcpy(buf+copylen,o->ptr,objlen);
-            copylen += objlen;
-            listDelNode(c->reply,ln);
-        } else {
-            if (copylen == 0) return;
-            break;
-        }
-    }
-    /* Now the output buffer is empty, add the new single element */
-    o = createObject(REDIS_STRING,sdsnewlen(buf,copylen));
-    listAddNodeHead(c->reply,o);
-}
-
 void sendReplyToClient(aeEventLoop *el, int fd, void *privdata, int mask) {
     redisClient *c = privdata;
     int nwritten = 0, totwritten = 0, objlen;
@@ -331,31 +355,48 @@ void sendReplyToClient(aeEventLoop *el, int fd, void *privdata, int mask) {
         return;
     }
 
-    while(listLength(c->reply)) {
+    while(c->bufpos > 0 || listLength(c->reply)) {
-        if (server.glueoutputbuf && listLength(c->reply) > 1)
+        if (c->bufpos > 0) {
-            glueReplyBuffersIfNeeded(c);
+            if (c->flags & REDIS_MASTER) {
+                /* Don't reply to a master */
+                nwritten = c->bufpos - c->sentlen;
+            } else {
+                nwritten = write(fd,c->buf+c->sentlen,c->bufpos-c->sentlen);
+                if (nwritten <= 0) break;
+            }
+            c->sentlen += nwritten;
+            totwritten += nwritten;
 
-        o = listNodeValue(listFirst(c->reply));
+            /* If the buffer was sent, set bufpos to zero to continue with
-        objlen = sdslen(o->ptr);
+             * the remainder of the reply. */
-
+            if (c->sentlen == c->bufpos) {
-        if (objlen == 0) {
+                c->bufpos = 0;
-            listDelNode(c->reply,listFirst(c->reply));
+                c->sentlen = 0;
-            continue;
+            }
-        }
-
-        if (c->flags & REDIS_MASTER) {
-            /* Don't reply to a master */
-            nwritten = objlen - c->sentlen;
         } else {
-            nwritten = write(fd, ((char*)o->ptr)+c->sentlen, objlen - c->sentlen);
+            o = listNodeValue(listFirst(c->reply));
-            if (nwritten <= 0) break;
+            objlen = sdslen(o->ptr);
-        }
+
-        c->sentlen += nwritten;
+            if (objlen == 0) {
-        totwritten += nwritten;
+                listDelNode(c->reply,listFirst(c->reply));
-        /* If we fully sent the object on head go to the next one */
+                continue;
-        if (c->sentlen == objlen) {
+            }
-            listDelNode(c->reply,listFirst(c->reply));
+
-            c->sentlen = 0;
+            if (c->flags & REDIS_MASTER) {
+                /* Don't reply to a master */
+                nwritten = objlen - c->sentlen;
+            } else {
+                nwritten = write(fd, ((char*)o->ptr)+c->sentlen,objlen-c->sentlen);
+                if (nwritten <= 0) break;
+            }
+            c->sentlen += nwritten;
+            totwritten += nwritten;
+
+            /* If we fully sent the object on head go to the next one */
+            if (c->sentlen == objlen) {
+                listDelNode(c->reply,listFirst(c->reply));
+                c->sentlen = 0;
+            }
         }
         /* Note that we avoid to send more thank REDIS_MAX_WRITE_PER_EVENT
          * bytes, in a single threaded server it's a good idea to serve

src/object.c

@@ -196,6 +196,7 @@ void decrRefCount(void *obj) {
         case REDIS_SET: freeSetObject(o); break;
         case REDIS_ZSET: freeZsetObject(o); break;
         case REDIS_HASH: freeHashObject(o); break;
+        case REDIS_REPLY_NODE: freeStringObject(o); break;
         default: redisPanic("Unknown object type"); break;
         }
         o->ptr = NULL; /* defensive programming. We'll see NULL in traces. */

src/redis.h

@@ -48,6 +48,15 @@
 #define REDIS_REQUEST_MAX_SIZE (1024*1024*256) /* max bytes in inline command */
 #define REDIS_SHARED_INTEGERS 10000
 
+/* Size of a reply chunk, configured to exactly allocate 4k bytes */
+#define REDIS_REPLY_CHUNK_BYTES (4*1024)
+#define REDIS_REPLY_CHUNK_SIZE (REDIS_REPLY_CHUNK_BYTES-sizeof(struct sdshdr)-1-sizeof(size_t))
+/* It doesn't make sense to memcpy objects to a chunk when the net result is
+ * not being able to glue other objects. We want to make sure it can be glued
+ * to at least a bulk length or \r\n, so set the threshold to be a couple
+ * of bytes less than the size of the buffer. */
+#define REDIS_REPLY_CHUNK_THRESHOLD (REDIS_REPLY_CHUNK_SIZE-16)
+
 /* If more then REDIS_WRITEV_THRESHOLD write packets are pending use writev */
 #define REDIS_WRITEV_THRESHOLD      3
 /* Max number of iovecs used for each writev call */
@@ -72,6 +81,7 @@
 #define REDIS_SET 2
 #define REDIS_ZSET 3
 #define REDIS_HASH 4
+#define REDIS_REPLY_NODE 5
 #define REDIS_VMPOINTER 8
 
 /* Objects encoding. Some kind of objects like Strings and Hashes can be
@@ -309,6 +319,11 @@ typedef struct redisClient {
     list *watched_keys;     /* Keys WATCHED for MULTI/EXEC CAS */
     dict *pubsub_channels;  /* channels a client is interested in (SUBSCRIBE) */
     list *pubsub_patterns;  /* patterns a client is interested in (SUBSCRIBE) */
+
+    /* Response buffer */
+    int bufpos;
+    int buflen;
+    char buf[];
 } redisClient;
 
 struct saveparam {