Added a define to set the size threshold to enable the multi bulk parsing big objects...

[redis.git] / src / networking.c

diff --git a/src/networking.c b/src/networking.c
index 3979ab6225c4ca24ca64f557db46e03a0edc6f9f..4ecf78e80e251f7d2991a8c2a9f52357b6496995 100644 (file)
--- a/src/networking.c
+++ b/src/networking.c
@@ -412,7 +412,7 @@ static void acceptCommonHandler(int fd) {
      * connection. Note that we create the client instead to check before
      * for this condition, since now the socket is already set in nonblocking
      * mode and we can send an error for free using the Kernel I/O */
      * connection. Note that we create the client instead to check before
      * for this condition, since now the socket is already set in nonblocking
      * mode and we can send an error for free using the Kernel I/O */

-    if (server.maxclients && listLength(server.clients) > server.maxclients) {
+    if (listLength(server.clients) > server.maxclients) {

         char *err = "-ERR max number of clients reached\r\n";
 
         /* That's a best effort error message, don't check write errors */
         char *err = "-ERR max number of clients reached\r\n";
 
         /* That's a best effort error message, don't check write errors */


@@ -625,6 +625,8 @@ void resetClient(redisClient *c) {
     c->reqtype = 0;
     c->multibulklen = 0;
     c->bulklen = -1;
     c->reqtype = 0;
     c->multibulklen = 0;
     c->bulklen = -1;

+    /* We clear the ASKING flag as well if we are not inside a MULTI. */
+    if (!(c->flags & REDIS_MULTI)) c->flags &= (~REDIS_ASKING);

 }
 
 void closeTimedoutClients(void) {
 }
 
 void closeTimedoutClients(void) {


@@ -703,7 +705,7 @@ int processMultibulkBuffer(redisClient *c) {
 
     if (c->multibulklen == 0) {
         /* The client should have been reset */
 
     if (c->multibulklen == 0) {
         /* The client should have been reset */

-        redisAssert(c->argc == 0);
+        redisAssertWithInfo(c,NULL,c->argc == 0);

 
         /* Multi bulk length cannot be read without a \r\n */
         newline = strchr(c->querybuf,'\r');
 
         /* Multi bulk length cannot be read without a \r\n */
         newline = strchr(c->querybuf,'\r');


@@ -716,7 +718,7 @@ int processMultibulkBuffer(redisClient *c) {
 
         /* We know for sure there is a whole line since newline != NULL,
          * so go ahead and find out the multi bulk length. */
 
         /* We know for sure there is a whole line since newline != NULL,
          * so go ahead and find out the multi bulk length. */

-        redisAssert(c->querybuf[0] == '*');
+        redisAssertWithInfo(c,NULL,c->querybuf[0] == '*');

         ok = string2ll(c->querybuf+1,newline-(c->querybuf+1),&ll);
         if (!ok || ll > 1024*1024) {
             addReplyError(c,"Protocol error: invalid multibulk length");
         ok = string2ll(c->querybuf+1,newline-(c->querybuf+1),&ll);
         if (!ok || ll > 1024*1024) {
             addReplyError(c,"Protocol error: invalid multibulk length");


@@ -737,7 +739,7 @@ int processMultibulkBuffer(redisClient *c) {
         c->argv = zmalloc(sizeof(robj*)*c->multibulklen);
     }
 
         c->argv = zmalloc(sizeof(robj*)*c->multibulklen);
     }
 

-    redisAssert(c->multibulklen > 0);
+    redisAssertWithInfo(c,NULL,c->multibulklen > 0);

     while(c->multibulklen) {
         /* Read bulk length if unknown */
         if (c->bulklen == -1) {
     while(c->multibulklen) {
         /* Read bulk length if unknown */
         if (c->bulklen == -1) {


@@ -765,6 +767,20 @@ int processMultibulkBuffer(redisClient *c) {
             }
 
             pos += newline-(c->querybuf+pos)+2;
             }
 
             pos += newline-(c->querybuf+pos)+2;

+#ifdef REDIS_MBULK_BIG_ARG
+            if (ll >= REDIS_MBULK_BIG_ARG) {
+                /* If we are going to read a large object from network
+                 * try to make it likely that it will start at c->querybuf
+                 * boundary so that we can optimized object creation
+                 * avoiding a large copy of data. */
+                c->querybuf = sdsrange(c->querybuf,pos,-1);
+                pos = 0;
+            }
+            /* Hint the sds library about the amount of bytes this string is
+             * going to contain. */
+            if (ll >= REDIS_MBULK_BIG_ARG)
+                c->querybuf = sdsMakeRoomFor(c->querybuf,ll+2);
+#endif

             c->bulklen = ll;
         }
 
             c->bulklen = ll;
         }
 


@@ -773,15 +789,38 @@ int processMultibulkBuffer(redisClient *c) {
             /* Not enough data (+2 == trailing \r\n) */
             break;
         } else {
             /* Not enough data (+2 == trailing \r\n) */
             break;
         } else {

-            c->argv[c->argc++] = createStringObject(c->querybuf+pos,c->bulklen);
-            pos += c->bulklen+2;
+            /* Optimization: if the buffer contanins JUST our bulk element
+             * instead of creating a new object by *copying* the sds we
+             * just use the current sds string. */
+#ifdef REDIS_MBULK_BIG_ARG
+            if (pos == 0 &&
+                c->bulklen >= REDIS_MBULK_BIG_ARG &&
+                (signed) sdslen(c->querybuf) == c->bulklen+2)
+            {
+                // printf("HERE (arg %d)\n",c->argc);
+                c->argv[c->argc++] = createObject(REDIS_STRING,c->querybuf);
+                sdsIncrLen(c->querybuf,-2); /* remove CRLF */
+                c->querybuf = sdsempty();
+                /* Assume that if we saw a fat argument we'll see another one
+                 * likely... */
+                c->querybuf = sdsMakeRoomFor(c->querybuf,c->bulklen+2);
+                pos = 0;
+            } else {
+#endif
+                // printf("NOT HERE (arg %d) (pos %d)\n",c->argc, pos);
+                c->argv[c->argc++] =
+                    createStringObject(c->querybuf+pos,c->bulklen);
+                pos += c->bulklen+2;
+#ifdef REDIS_MBULK_BIG_ARG
+            }
+#endif

             c->bulklen = -1;
             c->multibulklen--;
         }
     }
 
     /* Trim to pos */
             c->bulklen = -1;
             c->multibulklen--;
         }
     }
 
     /* Trim to pos */

-    c->querybuf = sdsrange(c->querybuf,pos,-1);
+    if (pos) c->querybuf = sdsrange(c->querybuf,pos,-1);

 
     /* We're done when c->multibulk == 0 */
     if (c->multibulklen == 0) {
 
     /* We're done when c->multibulk == 0 */
     if (c->multibulklen == 0) {


@@ -831,12 +870,31 @@ void processInputBuffer(redisClient *c) {
 
 void readQueryFromClient(aeEventLoop *el, int fd, void *privdata, int mask) {
     redisClient *c = (redisClient*) privdata;
 
 void readQueryFromClient(aeEventLoop *el, int fd, void *privdata, int mask) {
     redisClient *c = (redisClient*) privdata;

-    char buf[REDIS_IOBUF_LEN];
-    int nread;
+    int nread, readlen;
+    size_t qblen;

     REDIS_NOTUSED(el);
     REDIS_NOTUSED(mask);
 
     REDIS_NOTUSED(el);
     REDIS_NOTUSED(mask);
 

-    nread = read(fd, buf, REDIS_IOBUF_LEN);
+    readlen = REDIS_IOBUF_LEN;
+#if REDIS_MBULK_BIG_ARG
+    /* If this is a multi bulk request, and we are processing a bulk reply
+     * that is large enough, try to maximize the probabilty that the query
+     * buffer contains excatly the SDS string representing the object, even
+     * at the risk of requring more read(2) calls. This way the function
+     * processMultiBulkBuffer() can avoid copying buffers to create the
+     * Redis Object representing the argument. */
+    if (c->reqtype == REDIS_REQ_MULTIBULK && c->multibulklen && c->bulklen != -1
+        && c->bulklen >= REDIS_MBULK_BIG_ARG)
+    {
+        int remaining = (unsigned)(c->bulklen+2)-sdslen(c->querybuf);
+
+        if (remaining < readlen) readlen = remaining;
+    }
+#endif
+
+    qblen = sdslen(c->querybuf);
+    c->querybuf = sdsMakeRoomFor(c->querybuf, readlen);
+    nread = read(fd, c->querybuf+qblen, readlen);

     if (nread == -1) {
         if (errno == EAGAIN) {
             nread = 0;
     if (nread == -1) {
         if (errno == EAGAIN) {
             nread = 0;


@@ -851,7 +909,7 @@ void readQueryFromClient(aeEventLoop *el, int fd, void *privdata, int mask) {
         return;
     }
     if (nread) {
         return;
     }
     if (nread) {

-        c->querybuf = sdscatlen(c->querybuf,buf,nread);
+        sdsIncrLen(c->querybuf,nread);

         c->lastinteraction = time(NULL);
     } else {
         return;
         c->lastinteraction = time(NULL);
     } else {
         return;


@@ -970,7 +1028,7 @@ void rewriteClientCommandVector(redisClient *c, int argc, ...) {
     c->argv = argv;
     c->argc = argc;
     c->cmd = lookupCommand(c->argv[0]->ptr);
     c->argv = argv;
     c->argc = argc;
     c->cmd = lookupCommand(c->argv[0]->ptr);

-    redisAssert(c->cmd != NULL);
+    redisAssertWithInfo(c,NULL,c->cmd != NULL);

     va_end(ap);
 }
 
     va_end(ap);
 }
 


@@ -979,7 +1037,7 @@ void rewriteClientCommandVector(redisClient *c, int argc, ...) {
 void rewriteClientCommandArgument(redisClient *c, int i, robj *newval) {
     robj *oldval;
    
 void rewriteClientCommandArgument(redisClient *c, int i, robj *newval) {
     robj *oldval;
    

-    redisAssert(i < c->argc);
+    redisAssertWithInfo(c,NULL,i < c->argc);

     oldval = c->argv[i];
     c->argv[i] = newval;
     incrRefCount(newval);
     oldval = c->argv[i];
     c->argv[i] = newval;
     incrRefCount(newval);


@@ -988,6 +1046,6 @@ void rewriteClientCommandArgument(redisClient *c, int i, robj *newval) {
     /* If this is the command name make sure to fix c->cmd. */
     if (i == 0) {
         c->cmd = lookupCommand(c->argv[0]->ptr);
     /* If this is the command name make sure to fix c->cmd. */
     if (i == 0) {
         c->cmd = lookupCommand(c->argv[0]->ptr);

-        redisAssert(c->cmd != NULL);
+        redisAssertWithInfo(c,NULL,c->cmd != NULL);

     }
 }
     }
 }