BSD license added to every C source and header file.

[redis.git] / src / networking.c

diff --git a/src/networking.c b/src/networking.c
index fe745102f1b8288158e6cb224f5d5a36d304e641..c0dd4d0d80b997bb80b3f73f22afcb89db31309a 100644 (file)
--- a/src/networking.c
+++ b/src/networking.c
@@ -1,8 +1,45 @@
+/*
+ * Copyright (c) 2009-2012, Salvatore Sanfilippo <antirez at gmail dot com>
+ * All rights reserved.
+ *
+ * Redistribution and use in source and binary forms, with or without
+ * modification, are permitted provided that the following conditions are met:
+ *
+ *   * Redistributions of source code must retain the above copyright notice,
+ *     this list of conditions and the following disclaimer.
+ *   * Redistributions in binary form must reproduce the above copyright
+ *     notice, this list of conditions and the following disclaimer in the
+ *     documentation and/or other materials provided with the distribution.
+ *   * Neither the name of Redis nor the names of its contributors may be used
+ *     to endorse or promote products derived from this software without
+ *     specific prior written permission.
+ *
+ * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
+ * AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
+ * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
+ * ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE
+ * LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
+ * CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
+ * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
+ * INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
+ * CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
+ * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
+ * POSSIBILITY OF SUCH DAMAGE.
+ */
+

 #include "redis.h"
 #include <sys/uio.h>
 
 static void setProtocolError(redisClient *c, int pos);
 
 #include "redis.h"
 #include <sys/uio.h>
 
 static void setProtocolError(redisClient *c, int pos);
 

+/* To evaluate the output buffer size of a client we need to get size of
+ * allocated objects, however we can't used zmalloc_size() directly on sds
+ * strings because of the trick they use to work (the header is before the
+ * returned pointer), so we use this helper function. */
+size_t zmalloc_size_sds(sds s) {
+    return zmalloc_size(s-sizeof(struct sdshdr));
+}
+

 void *dupClientReplyValue(void *o) {
     incrRefCount((robj*)o);
     return o;
 void *dupClientReplyValue(void *o) {
     incrRefCount((robj*)o);
     return o;


@@ -14,7 +51,6 @@ int listMatchObjects(void *a, void *b) {
 
 redisClient *createClient(int fd) {
     redisClient *c = zmalloc(sizeof(redisClient));
 
 redisClient *createClient(int fd) {
     redisClient *c = zmalloc(sizeof(redisClient));

-    c->bufpos = 0;

 
     /* passing -1 as fd it is possible to create a non connected client.
      * This is useful since all the Redis commands needs to be executed
 
     /* passing -1 as fd it is possible to create a non connected client.
      * This is useful since all the Redis commands needs to be executed


@@ -34,7 +70,9 @@ redisClient *createClient(int fd) {
 
     selectDb(c,0);
     c->fd = fd;
 
     selectDb(c,0);
     c->fd = fd;

+    c->bufpos = 0;

     c->querybuf = sdsempty();
     c->querybuf = sdsempty();

+    c->querybuf_peak = 0;

     c->reqtype = 0;
     c->argc = 0;
     c->argv = NULL;
     c->reqtype = 0;
     c->argc = 0;
     c->argv = NULL;


@@ -43,9 +81,10 @@ redisClient *createClient(int fd) {
     c->bulklen = -1;
     c->sentlen = 0;
     c->flags = 0;
     c->bulklen = -1;
     c->sentlen = 0;
     c->flags = 0;

-    c->lastinteraction = time(NULL);
+    c->ctime = c->lastinteraction = server.unixtime;

     c->authenticated = 0;
     c->replstate = REDIS_REPL_NONE;
     c->authenticated = 0;
     c->replstate = REDIS_REPL_NONE;

+    c->slave_listening_port = 0;

     c->reply = listCreate();
     c->reply_bytes = 0;
     c->obuf_soft_limit_reached_time = 0;
     c->reply = listCreate();
     c->reply_bytes = 0;
     c->obuf_soft_limit_reached_time = 0;


@@ -137,6 +176,7 @@ void _addReplyObjectToList(redisClient *c, robj *o) {
     if (listLength(c->reply) == 0) {
         incrRefCount(o);
         listAddNodeTail(c->reply,o);
     if (listLength(c->reply) == 0) {
         incrRefCount(o);
         listAddNodeTail(c->reply,o);

+        c->reply_bytes += zmalloc_size_sds(o->ptr);

     } else {
         tail = listNodeValue(listLast(c->reply));
 
     } else {
         tail = listNodeValue(listLast(c->reply));
 


@@ -144,14 +184,16 @@ void _addReplyObjectToList(redisClient *c, robj *o) {
         if (tail->ptr != NULL &&
             sdslen(tail->ptr)+sdslen(o->ptr) <= REDIS_REPLY_CHUNK_BYTES)
         {
         if (tail->ptr != NULL &&
             sdslen(tail->ptr)+sdslen(o->ptr) <= REDIS_REPLY_CHUNK_BYTES)
         {

+            c->reply_bytes -= zmalloc_size_sds(tail->ptr);

             tail = dupLastObjectIfNeeded(c->reply);
             tail->ptr = sdscatlen(tail->ptr,o->ptr,sdslen(o->ptr));
             tail = dupLastObjectIfNeeded(c->reply);
             tail->ptr = sdscatlen(tail->ptr,o->ptr,sdslen(o->ptr));

+            c->reply_bytes += zmalloc_size_sds(tail->ptr);

         } else {
             incrRefCount(o);
             listAddNodeTail(c->reply,o);
         } else {
             incrRefCount(o);
             listAddNodeTail(c->reply,o);

+            c->reply_bytes += zmalloc_size_sds(o->ptr);

         }
     }
         }
     }

-    c->reply_bytes += sdslen(o->ptr);

     asyncCloseClientOnOutputBufferLimitReached(c);
 }
 
     asyncCloseClientOnOutputBufferLimitReached(c);
 }
 


@@ -165,9 +207,9 @@ void _addReplySdsToList(redisClient *c, sds s) {
         return;
     }
 
         return;
     }
 

-    c->reply_bytes += sdslen(s);

     if (listLength(c->reply) == 0) {
         listAddNodeTail(c->reply,createObject(REDIS_STRING,s));
     if (listLength(c->reply) == 0) {
         listAddNodeTail(c->reply,createObject(REDIS_STRING,s));

+        c->reply_bytes += zmalloc_size_sds(s);

     } else {
         tail = listNodeValue(listLast(c->reply));
 
     } else {
         tail = listNodeValue(listLast(c->reply));
 


@@ -175,11 +217,14 @@ void _addReplySdsToList(redisClient *c, sds s) {
         if (tail->ptr != NULL &&
             sdslen(tail->ptr)+sdslen(s) <= REDIS_REPLY_CHUNK_BYTES)
         {
         if (tail->ptr != NULL &&
             sdslen(tail->ptr)+sdslen(s) <= REDIS_REPLY_CHUNK_BYTES)
         {

+            c->reply_bytes -= zmalloc_size_sds(tail->ptr);

             tail = dupLastObjectIfNeeded(c->reply);
             tail->ptr = sdscatlen(tail->ptr,s,sdslen(s));
             tail = dupLastObjectIfNeeded(c->reply);
             tail->ptr = sdscatlen(tail->ptr,s,sdslen(s));

+            c->reply_bytes += zmalloc_size_sds(tail->ptr);

             sdsfree(s);
         } else {
             listAddNodeTail(c->reply,createObject(REDIS_STRING,s));
             sdsfree(s);
         } else {
             listAddNodeTail(c->reply,createObject(REDIS_STRING,s));

+            c->reply_bytes += zmalloc_size_sds(s);

         }
     }
     asyncCloseClientOnOutputBufferLimitReached(c);
         }
     }
     asyncCloseClientOnOutputBufferLimitReached(c);


@@ -191,7 +236,10 @@ void _addReplyStringToList(redisClient *c, char *s, size_t len) {
     if (c->flags & REDIS_CLOSE_AFTER_REPLY) return;
 
     if (listLength(c->reply) == 0) {
     if (c->flags & REDIS_CLOSE_AFTER_REPLY) return;
 
     if (listLength(c->reply) == 0) {

-        listAddNodeTail(c->reply,createStringObject(s,len));
+        robj *o = createStringObject(s,len);
+
+        listAddNodeTail(c->reply,o);
+        c->reply_bytes += zmalloc_size_sds(o->ptr);

     } else {
         tail = listNodeValue(listLast(c->reply));
 
     } else {
         tail = listNodeValue(listLast(c->reply));
 


@@ -199,13 +247,17 @@ void _addReplyStringToList(redisClient *c, char *s, size_t len) {
         if (tail->ptr != NULL &&
             sdslen(tail->ptr)+len <= REDIS_REPLY_CHUNK_BYTES)
         {
         if (tail->ptr != NULL &&
             sdslen(tail->ptr)+len <= REDIS_REPLY_CHUNK_BYTES)
         {

+            c->reply_bytes -= zmalloc_size_sds(tail->ptr);

             tail = dupLastObjectIfNeeded(c->reply);
             tail->ptr = sdscatlen(tail->ptr,s,len);
             tail = dupLastObjectIfNeeded(c->reply);
             tail->ptr = sdscatlen(tail->ptr,s,len);

+            c->reply_bytes += zmalloc_size_sds(tail->ptr);

         } else {
         } else {

-            listAddNodeTail(c->reply,createStringObject(s,len));
+            robj *o = createStringObject(s,len);
+
+            listAddNodeTail(c->reply,o);
+            c->reply_bytes += zmalloc_size_sds(o->ptr);

         }
     }
         }
     }

-    c->reply_bytes += len;

     asyncCloseClientOnOutputBufferLimitReached(c);
 }
 
     asyncCloseClientOnOutputBufferLimitReached(c);
 }
 


@@ -336,13 +388,16 @@ void setDeferredMultiBulkLength(redisClient *c, void *node, long length) {
 
     len = listNodeValue(ln);
     len->ptr = sdscatprintf(sdsempty(),"*%ld\r\n",length);
 
     len = listNodeValue(ln);
     len->ptr = sdscatprintf(sdsempty(),"*%ld\r\n",length);

-    c->reply_bytes += sdslen(len->ptr);
+    c->reply_bytes += zmalloc_size_sds(len->ptr);

     if (ln->next != NULL) {
         next = listNodeValue(ln->next);
 
         /* Only glue when the next node is non-NULL (an sds in this case) */
         if (next->ptr != NULL) {
     if (ln->next != NULL) {
         next = listNodeValue(ln->next);
 
         /* Only glue when the next node is non-NULL (an sds in this case) */
         if (next->ptr != NULL) {

+            c->reply_bytes -= zmalloc_size_sds(len->ptr);
+            c->reply_bytes -= zmalloc_size_sds(next->ptr);

             len->ptr = sdscatlen(len->ptr,next->ptr,sdslen(next->ptr));
             len->ptr = sdscatlen(len->ptr,next->ptr,sdslen(next->ptr));

+            c->reply_bytes += zmalloc_size_sds(len->ptr);

             listDelNode(c->reply,ln->next);
         }
     }
             listDelNode(c->reply,ln->next);
         }
     }


@@ -363,6 +418,18 @@ void addReplyDouble(redisClient *c, double d) {
 void addReplyLongLongWithPrefix(redisClient *c, long long ll, char prefix) {
     char buf[128];
     int len;
 void addReplyLongLongWithPrefix(redisClient *c, long long ll, char prefix) {
     char buf[128];
     int len;

+
+    /* Things like $3\r\n or *2\r\n are emitted very often by the protocol
+     * so we have a few shared objects to use if the integer is small
+     * like it is most of the times. */
+    if (prefix == '*' && ll < REDIS_SHARED_BULKHDR_LEN) {
+        addReply(c,shared.mbulkhdr[ll]);
+        return;
+    } else if (prefix == '$' && ll < REDIS_SHARED_BULKHDR_LEN) {
+        addReply(c,shared.bulkhdr[ll]);
+        return;
+    }
+

     buf[0] = prefix;
     len = ll2string(buf+1,sizeof(buf)-1,ll);
     buf[len+1] = '\r';
     buf[0] = prefix;
     len = ll2string(buf+1,sizeof(buf)-1,ll);
     buf[len+1] = '\r';


@@ -448,11 +515,11 @@ void copyClientOutputBuffer(redisClient *dst, redisClient *src) {
     dst->reply_bytes = src->reply_bytes;
 }
 
     dst->reply_bytes = src->reply_bytes;
 }
 

-static void acceptCommonHandler(int fd) {
+static void acceptCommonHandler(int fd, int flags) {

     redisClient *c;
     if ((c = createClient(fd)) == NULL) {
     redisClient *c;
     if ((c = createClient(fd)) == NULL) {

-        redisLog(REDIS_WARNING,"Error allocating resoures for the client");
-        close(fd); /* May be already closed, just ingore errors */
+        redisLog(REDIS_WARNING,"Error allocating resources for the client");
+        close(fd); /* May be already closed, just ignore errors */

         return;
     }
     /* If maxclient directive is set and this is one client more... close the
         return;
     }
     /* If maxclient directive is set and this is one client more... close the


@@ -471,6 +538,7 @@ static void acceptCommonHandler(int fd) {
         return;
     }
     server.stat_numconnections++;
         return;
     }
     server.stat_numconnections++;

+    c->flags |= flags;

 }
 
 void acceptTcpHandler(aeEventLoop *el, int fd, void *privdata, int mask) {
 }
 
 void acceptTcpHandler(aeEventLoop *el, int fd, void *privdata, int mask) {


@@ -486,7 +554,7 @@ void acceptTcpHandler(aeEventLoop *el, int fd, void *privdata, int mask) {
         return;
     }
     redisLog(REDIS_VERBOSE,"Accepted %s:%d", cip, cport);
         return;
     }
     redisLog(REDIS_VERBOSE,"Accepted %s:%d", cip, cport);

-    acceptCommonHandler(cfd);
+    acceptCommonHandler(cfd,0);

 }
 
 void acceptUnixHandler(aeEventLoop *el, int fd, void *privdata, int mask) {
 }
 
 void acceptUnixHandler(aeEventLoop *el, int fd, void *privdata, int mask) {


@@ -501,7 +569,7 @@ void acceptUnixHandler(aeEventLoop *el, int fd, void *privdata, int mask) {
         return;
     }
     redisLog(REDIS_VERBOSE,"Accepted connection to %s", server.unixsocket);
         return;
     }
     redisLog(REDIS_VERBOSE,"Accepted connection to %s", server.unixsocket);

-    acceptCommonHandler(cfd);
+    acceptCommonHandler(cfd,REDIS_UNIX_SOCKET);

 }
 
 
 }
 
 


@@ -513,6 +581,16 @@ static void freeClientArgv(redisClient *c) {
     c->cmd = NULL;
 }
 
     c->cmd = NULL;
 }
 

+/* Close all the slaves connections. This is useful in chained replication
+ * when we resync with our own master and want to force all our slaves to
+ * resync with us as well. */
+void disconnectSlaves(void) {
+    while (listLength(server.slaves)) {
+        listNode *ln = listFirst(server.slaves);
+        freeClient((redisClient*)ln->value);
+    }
+}
+

 void freeClient(redisClient *c) {
     listNode *ln;
 
 void freeClient(redisClient *c) {
     listNode *ln;
 


@@ -570,22 +648,13 @@ void freeClient(redisClient *c) {
     if (c->flags & REDIS_MASTER) {
         server.master = NULL;
         server.repl_state = REDIS_REPL_CONNECT;
     if (c->flags & REDIS_MASTER) {
         server.master = NULL;
         server.repl_state = REDIS_REPL_CONNECT;

-        server.repl_down_since = time(NULL);
-        /* Since we lost the connection with the master, we should also
-         * close the connection with all our slaves if we have any, so
-         * when we'll resync with the master the other slaves will sync again
-         * with us as well. Note that also when the slave is not connected
-         * to the master it will keep refusing connections by other slaves.
+        server.repl_down_since = server.unixtime;
+        /* We lost connection with our master, force our slaves to resync
+         * with us as well to load the new data set.

          *
          *

-         * We do this only if server.masterhost != NULL. If it is NULL this
-         * means the user called SLAVEOF NO ONE and we are freeing our
-         * link with the master, so no need to close link with slaves. */
-        if (server.masterhost != NULL) {
-            while (listLength(server.slaves)) {
-                ln = listFirst(server.slaves);
-                freeClient((redisClient*)ln->value);
-            }
-        }
+         * If server.masterhost is NULL the user called SLAVEOF NO ONE so
+         * slave resync is not needed. */
+        if (server.masterhost != NULL) disconnectSlaves();

     }
 
     /* If this client was scheduled for async freeing we need to remove it
     }
 
     /* If this client was scheduled for async freeing we need to remove it


@@ -626,6 +695,7 @@ void freeClientsInAsyncFreeQueue(void) {
 void sendReplyToClient(aeEventLoop *el, int fd, void *privdata, int mask) {
     redisClient *c = privdata;
     int nwritten = 0, totwritten = 0, objlen;
 void sendReplyToClient(aeEventLoop *el, int fd, void *privdata, int mask) {
     redisClient *c = privdata;
     int nwritten = 0, totwritten = 0, objlen;

+    size_t objmem;

     robj *o;
     REDIS_NOTUSED(el);
     REDIS_NOTUSED(mask);
     robj *o;
     REDIS_NOTUSED(el);
     REDIS_NOTUSED(mask);


@@ -651,6 +721,7 @@ void sendReplyToClient(aeEventLoop *el, int fd, void *privdata, int mask) {
         } else {
             o = listNodeValue(listFirst(c->reply));
             objlen = sdslen(o->ptr);
         } else {
             o = listNodeValue(listFirst(c->reply));
             objlen = sdslen(o->ptr);

+            objmem = zmalloc_size_sds(o->ptr);

 
             if (objlen == 0) {
                 listDelNode(c->reply,listFirst(c->reply));
 
             if (objlen == 0) {
                 listDelNode(c->reply,listFirst(c->reply));


@@ -671,15 +742,20 @@ void sendReplyToClient(aeEventLoop *el, int fd, void *privdata, int mask) {
             if (c->sentlen == objlen) {
                 listDelNode(c->reply,listFirst(c->reply));
                 c->sentlen = 0;
             if (c->sentlen == objlen) {
                 listDelNode(c->reply,listFirst(c->reply));
                 c->sentlen = 0;

-                c->reply_bytes -= objlen;
+                c->reply_bytes -= objmem;

             }
         }
             }
         }

-        /* Note that we avoid to send more thank REDIS_MAX_WRITE_PER_EVENT
+        /* Note that we avoid to send more than REDIS_MAX_WRITE_PER_EVENT

          * bytes, in a single threaded server it's a good idea to serve
          * other clients as well, even if a very large request comes from
          * super fast link that is always able to accept data (in real world
          * bytes, in a single threaded server it's a good idea to serve
          * other clients as well, even if a very large request comes from
          * super fast link that is always able to accept data (in real world

-         * scenario think about 'KEYS *' against the loopback interfae) */
-        if (totwritten > REDIS_MAX_WRITE_PER_EVENT) break;
+         * scenario think about 'KEYS *' against the loopback interface).
+         *
+         * However if we are over the maxmemory limit we ignore that and
+         * just deliver as much data as it is possible to deliver. */
+        if (totwritten > REDIS_MAX_WRITE_PER_EVENT &&
+            (server.maxmemory == 0 ||
+             zmalloc_used_memory() < server.maxmemory)) break;

     }
     if (nwritten == -1) {
         if (errno == EAGAIN) {
     }
     if (nwritten == -1) {
         if (errno == EAGAIN) {


@@ -691,7 +767,7 @@ void sendReplyToClient(aeEventLoop *el, int fd, void *privdata, int mask) {
             return;
         }
     }
             return;
         }
     }

-    if (totwritten > 0) c->lastinteraction = time(NULL);
+    if (totwritten > 0) c->lastinteraction = server.unixtime;

     if (c->bufpos == 0 && listLength(c->reply) == 0) {
         c->sentlen = 0;
         aeDeleteFileEvent(server.el,c->fd,AE_WRITABLE);
     if (c->bufpos == 0 && listLength(c->reply) == 0) {
         c->sentlen = 0;
         aeDeleteFileEvent(server.el,c->fd,AE_WRITABLE);


@@ -711,34 +787,6 @@ void resetClient(redisClient *c) {
     if (!(c->flags & REDIS_MULTI)) c->flags &= (~REDIS_ASKING);
 }
 
     if (!(c->flags & REDIS_MULTI)) c->flags &= (~REDIS_ASKING);
 }
 

-void closeTimedoutClients(void) {
-    redisClient *c;
-    listNode *ln;
-    time_t now = time(NULL);
-    listIter li;
-
-    listRewind(server.clients,&li);
-    while ((ln = listNext(&li)) != NULL) {
-        c = listNodeValue(ln);
-        if (server.maxidletime &&
-            !(c->flags & REDIS_SLAVE) &&    /* no timeout for slaves */
-            !(c->flags & REDIS_MASTER) &&   /* no timeout for masters */
-            !(c->flags & REDIS_BLOCKED) &&  /* no timeout for BLPOP */
-            dictSize(c->pubsub_channels) == 0 && /* no timeout for pubsub */
-            listLength(c->pubsub_patterns) == 0 &&
-            (now - c->lastinteraction > server.maxidletime))
-        {
-            redisLog(REDIS_VERBOSE,"Closing idle client");
-            freeClient(c);
-        } else if (c->flags & REDIS_BLOCKED) {
-            if (c->bpop.timeout != 0 && c->bpop.timeout < now) {
-                addReply(c,shared.nullmultibulk);
-                unblockClientWaitingData(c);
-            }
-        }
-    }
-}
-

 int processInlineBuffer(redisClient *c) {
     char *newline = strstr(c->querybuf,"\r\n");
     int argc, j;
 int processInlineBuffer(redisClient *c) {
     char *newline = strstr(c->querybuf,"\r\n");
     int argc, j;


@@ -972,9 +1020,9 @@ void readQueryFromClient(aeEventLoop *el, int fd, void *privdata, int mask) {
     server.current_client = c;
     readlen = REDIS_IOBUF_LEN;
     /* If this is a multi bulk request, and we are processing a bulk reply
     server.current_client = c;
     readlen = REDIS_IOBUF_LEN;
     /* 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
+     * that is large enough, try to maximize the probability that the query
+     * buffer contains exactly the SDS string representing the object, even
+     * at the risk of requiring 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
      * processMultiBulkBuffer() can avoid copying buffers to create the
      * Redis Object representing the argument. */
     if (c->reqtype == REDIS_REQ_MULTIBULK && c->multibulklen && c->bulklen != -1


@@ -986,6 +1034,7 @@ void readQueryFromClient(aeEventLoop *el, int fd, void *privdata, int mask) {
     }
 
     qblen = sdslen(c->querybuf);
     }
 
     qblen = sdslen(c->querybuf);

+    if (c->querybuf_peak < qblen) c->querybuf_peak = qblen;

     c->querybuf = sdsMakeRoomFor(c->querybuf, readlen);
     nread = read(fd, c->querybuf+qblen, readlen);
     if (nread == -1) {
     c->querybuf = sdsMakeRoomFor(c->querybuf, readlen);
     nread = read(fd, c->querybuf+qblen, readlen);
     if (nread == -1) {


@@ -1003,7 +1052,7 @@ void readQueryFromClient(aeEventLoop *el, int fd, void *privdata, int mask) {
     }
     if (nread) {
         sdsIncrLen(c->querybuf,nread);
     }
     if (nread) {
         sdsIncrLen(c->querybuf,nread);

-        c->lastinteraction = time(NULL);
+        c->lastinteraction = server.unixtime;

     } else {
         server.current_client = NULL;
         return;
     } else {
         server.current_client = NULL;
         return;


@@ -1043,15 +1092,11 @@ void getClientsMaxBuffers(unsigned long *longest_output_list,
 /* Turn a Redis client into an sds string representing its state. */
 sds getClientInfoString(redisClient *client) {
     char ip[32], flags[16], events[3], *p;
 /* Turn a Redis client into an sds string representing its state. */
 sds getClientInfoString(redisClient *client) {
     char ip[32], flags[16], events[3], *p;

-    int port;
-    time_t now = time(NULL);
+    int port = 0; /* initialized to zero for the unix socket case. */

     int emask;
 
     int emask;
 

-    if (anetPeerToString(client->fd,ip,&port) == -1) {
-        ip[0] = '?';
-        ip[1] = '\0';
-        port = 0;
-    }
+    if (!(client->flags & REDIS_UNIX_SOCKET))
+        anetPeerToString(client->fd,ip,&port);

     p = flags;
     if (client->flags & REDIS_SLAVE) {
         if (client->flags & REDIS_MONITOR)
     p = flags;
     if (client->flags & REDIS_SLAVE) {
         if (client->flags & REDIS_MONITOR)


@@ -1066,6 +1111,7 @@ sds getClientInfoString(redisClient *client) {
     if (client->flags & REDIS_CLOSE_AFTER_REPLY) *p++ = 'c';
     if (client->flags & REDIS_UNBLOCKED) *p++ = 'u';
     if (client->flags & REDIS_CLOSE_ASAP) *p++ = 'A';
     if (client->flags & REDIS_CLOSE_AFTER_REPLY) *p++ = 'c';
     if (client->flags & REDIS_UNBLOCKED) *p++ = 'u';
     if (client->flags & REDIS_CLOSE_ASAP) *p++ = 'A';

+    if (client->flags & REDIS_UNIX_SOCKET) *p++ = 'U';

     if (p == flags) *p++ = 'N';
     *p++ = '\0';
 
     if (p == flags) *p++ = 'N';
     *p++ = '\0';
 


@@ -1075,14 +1121,18 @@ sds getClientInfoString(redisClient *client) {
     if (emask & AE_WRITABLE) *p++ = 'w';
     *p = '\0';
     return sdscatprintf(sdsempty(),
     if (emask & AE_WRITABLE) *p++ = 'w';
     *p = '\0';
     return sdscatprintf(sdsempty(),

-        "addr=%s:%d fd=%d idle=%ld flags=%s db=%d sub=%d psub=%d qbuf=%lu obl=%lu oll=%lu omem=%lu events=%s cmd=%s",
-        ip,port,client->fd,
-        (long)(now - client->lastinteraction),
+        "addr=%s:%d fd=%d age=%ld idle=%ld flags=%s db=%d sub=%d psub=%d multi=%d qbuf=%lu qbuf-free=%lu obl=%lu oll=%lu omem=%lu events=%s cmd=%s",
+        (client->flags & REDIS_UNIX_SOCKET) ? server.unixsocket : ip,
+        port,client->fd,
+        (long)(server.unixtime - client->ctime),
+        (long)(server.unixtime - client->lastinteraction),

         flags,
         client->db->id,
         (int) dictSize(client->pubsub_channels),
         (int) listLength(client->pubsub_patterns),
         flags,
         client->db->id,
         (int) dictSize(client->pubsub_channels),
         (int) listLength(client->pubsub_patterns),

+        (client->flags & REDIS_MULTI) ? client->mstate.count : -1,

         (unsigned long) sdslen(client->querybuf),
         (unsigned long) sdslen(client->querybuf),

+        (unsigned long) sdsavail(client->querybuf),

         (unsigned long) client->bufpos,
         (unsigned long) listLength(client->reply),
         getClientOutputBufferMemoryUsage(client),
         (unsigned long) client->bufpos,
         (unsigned long) listLength(client->reply),
         getClientOutputBufferMemoryUsage(client),


@@ -1205,12 +1255,12 @@ void rewriteClientCommandArgument(redisClient *c, int i, robj *newval) {
  * the caller wishes. The main usage of this function currently is
  * enforcing the client output length limits. */
 unsigned long getClientOutputBufferMemoryUsage(redisClient *c) {
  * the caller wishes. The main usage of this function currently is
  * enforcing the client output length limits. */
 unsigned long getClientOutputBufferMemoryUsage(redisClient *c) {

-    unsigned long list_item_size = sizeof(listNode);
+    unsigned long list_item_size = sizeof(listNode)+sizeof(robj);

 
     return c->reply_bytes + (list_item_size*listLength(c->reply));
 }
 
 
     return c->reply_bytes + (list_item_size*listLength(c->reply));
 }
 

-/* Get the class of a client, used in order to envorce limits to different
+/* Get the class of a client, used in order to enforce limits to different

  * classes of clients.
  *
  * The function will return one of the following:
  * classes of clients.
  *
  * The function will return one of the following:


@@ -1228,7 +1278,7 @@ int getClientLimitClass(redisClient *c) {
 int getClientLimitClassByName(char *name) {
     if (!strcasecmp(name,"normal")) return REDIS_CLIENT_LIMIT_CLASS_NORMAL;
     else if (!strcasecmp(name,"slave")) return REDIS_CLIENT_LIMIT_CLASS_SLAVE;
 int getClientLimitClassByName(char *name) {
     if (!strcasecmp(name,"normal")) return REDIS_CLIENT_LIMIT_CLASS_NORMAL;
     else if (!strcasecmp(name,"slave")) return REDIS_CLIENT_LIMIT_CLASS_SLAVE;

-    else if (!strcasecmp(name,"pubsub")) return REDIS_CLIENT_LIMIT_CLASS_SLAVE;
+    else if (!strcasecmp(name,"pubsub")) return REDIS_CLIENT_LIMIT_CLASS_PUBSUB;

     else return -1;
 }
 
     else return -1;
 }
 


@@ -1289,7 +1339,8 @@ int checkClientOutputBufferLimits(redisClient *c) {
  * called from contexts where the client can't be freed safely, i.e. from the
  * lower level functions pushing data inside the client output buffers. */
 void asyncCloseClientOnOutputBufferLimitReached(redisClient *c) {
  * called from contexts where the client can't be freed safely, i.e. from the
  * lower level functions pushing data inside the client output buffers. */
 void asyncCloseClientOnOutputBufferLimitReached(redisClient *c) {

-    if (c->flags & REDIS_CLOSE_ASAP) return;
+    redisAssert(c->reply_bytes < ULONG_MAX-(1024*64));
+    if (c->reply_bytes == 0 || c->flags & REDIS_CLOSE_ASAP) return;

     if (checkClientOutputBufferLimits(c)) {
         sds client = getClientInfoString(c);
 
     if (checkClientOutputBufferLimits(c)) {
         sds client = getClientInfoString(c);
 


@@ -1298,3 +1349,24 @@ void asyncCloseClientOnOutputBufferLimitReached(redisClient *c) {
         sdsfree(client);
     }
 }
         sdsfree(client);
     }
 }

+
+/* Helper function used by freeMemoryIfNeeded() in order to flush slaves
+ * output buffers without returning control to the event loop. */
+void flushSlavesOutputBuffers(void) {
+    listIter li;
+    listNode *ln;
+
+    listRewind(server.slaves,&li);
+    while((ln = listNext(&li))) {
+        redisClient *slave = listNodeValue(ln);
+        int events;
+
+        events = aeGetFileEvents(server.el,slave->fd);
+        if (events & AE_WRITABLE &&
+            slave->replstate == REDIS_REPL_ONLINE &&
+            listLength(slave->reply))
+        {
+            sendReplyToClient(server.el,slave->fd,slave,0);
+        }
+    }
+}