]> git.saurik.com Git - redis.git/blobdiff - src/networking.c
Only incremnet stats for key miss/hit when the key is semantically accessed in read...
[redis.git] / src / networking.c
index bc807c4b905a041d6fa7f294374b7fdb0267a5d2..3938f3f6410f759f789b91b488e97e2173284cb8 100644 (file)
@@ -1,6 +1,8 @@
 #include "redis.h"
 #include <sys/uio.h>
 
 #include "redis.h"
 #include <sys/uio.h>
 
+static void setProtocolError(redisClient *c, int pos);
+
 void *dupClientReplyValue(void *o) {
     incrRefCount((robj*)o);
     return o;
 void *dupClientReplyValue(void *o) {
     incrRefCount((robj*)o);
     return o;
@@ -45,6 +47,8 @@ redisClient *createClient(int fd) {
     c->authenticated = 0;
     c->replstate = REDIS_REPL_NONE;
     c->reply = listCreate();
     c->authenticated = 0;
     c->replstate = REDIS_REPL_NONE;
     c->reply = listCreate();
+    c->reply_bytes = 0;
+    c->obuf_soft_limit_reached_time = 0;
     listSetFreeMethod(c->reply,decrRefCount);
     listSetDupMethod(c->reply,dupClientReplyValue);
     c->bpop.keys = NULL;
     listSetFreeMethod(c->reply,decrRefCount);
     listSetDupMethod(c->reply,dupClientReplyValue);
     c->bpop.keys = NULL;
@@ -63,11 +67,23 @@ redisClient *createClient(int fd) {
     return c;
 }
 
     return c;
 }
 
-/* Set the event loop to listen for write events on the client's socket.
- * Typically gets called every time a reply is built. */
-int _installWriteEvent(redisClient *c) {
+/* This function is called every time we are going to transmit new data
+ * to the client. The behavior is the following:
+ *
+ * If the client should receive new data (normal clients will) the function
+ * returns REDIS_OK, and make sure to install the write handler in our event
+ * loop so that when the socket is writable new data gets written.
+ *
+ * If the client should not receive new data, because it is a fake client
+ * or a slave, or because the setup of the write handler failed, the function
+ * returns REDIS_ERR.
+ *
+ * Typically gets called every time a reply is built, before adding more
+ * data to the clients output buffers. If the function returns REDIS_ERR no
+ * data should be appended to the output buffers. */
+int prepareClientToWrite(redisClient *c) {
     if (c->flags & REDIS_LUA_CLIENT) return REDIS_OK;
     if (c->flags & REDIS_LUA_CLIENT) return REDIS_OK;
-    if (c->fd <= 0) return REDIS_ERR;
+    if (c->fd <= 0) return REDIS_ERR; /* Fake client */
     if (c->bufpos == 0 && listLength(c->reply) == 0 &&
         (c->replstate == REDIS_REPL_NONE ||
          c->replstate == REDIS_REPL_ONLINE) &&
     if (c->bufpos == 0 && listLength(c->reply) == 0 &&
         (c->replstate == REDIS_REPL_NONE ||
          c->replstate == REDIS_REPL_ONLINE) &&
@@ -135,6 +151,8 @@ void _addReplyObjectToList(redisClient *c, robj *o) {
             listAddNodeTail(c->reply,o);
         }
     }
             listAddNodeTail(c->reply,o);
         }
     }
+    c->reply_bytes += sdslen(o->ptr);
+    asyncCloseClientOnOutputBufferLimitReached(c);
 }
 
 /* This method takes responsibility over the sds. When it is no longer
 }
 
 /* This method takes responsibility over the sds. When it is no longer
@@ -147,6 +165,7 @@ 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));
     } else {
     if (listLength(c->reply) == 0) {
         listAddNodeTail(c->reply,createObject(REDIS_STRING,s));
     } else {
@@ -163,6 +182,7 @@ void _addReplySdsToList(redisClient *c, sds s) {
             listAddNodeTail(c->reply,createObject(REDIS_STRING,s));
         }
     }
             listAddNodeTail(c->reply,createObject(REDIS_STRING,s));
         }
     }
+    asyncCloseClientOnOutputBufferLimitReached(c);
 }
 
 void _addReplyStringToList(redisClient *c, char *s, size_t len) {
 }
 
 void _addReplyStringToList(redisClient *c, char *s, size_t len) {
@@ -185,6 +205,8 @@ void _addReplyStringToList(redisClient *c, char *s, size_t len) {
             listAddNodeTail(c->reply,createStringObject(s,len));
         }
     }
             listAddNodeTail(c->reply,createStringObject(s,len));
         }
     }
+    c->reply_bytes += len;
+    asyncCloseClientOnOutputBufferLimitReached(c);
 }
 
 /* -----------------------------------------------------------------------------
 }
 
 /* -----------------------------------------------------------------------------
@@ -193,7 +215,7 @@ void _addReplyStringToList(redisClient *c, char *s, size_t len) {
  * -------------------------------------------------------------------------- */
 
 void addReply(redisClient *c, robj *obj) {
  * -------------------------------------------------------------------------- */
 
 void addReply(redisClient *c, robj *obj) {
-    if (_installWriteEvent(c) != REDIS_OK) return;
+    if (prepareClientToWrite(c) != REDIS_OK) return;
 
     /* This is an important place where we can avoid copy-on-write
      * when there is a saving child running, avoiding touching the
 
     /* This is an important place where we can avoid copy-on-write
      * when there is a saving child running, avoiding touching the
@@ -205,19 +227,31 @@ void addReply(redisClient *c, robj *obj) {
     if (obj->encoding == REDIS_ENCODING_RAW) {
         if (_addReplyToBuffer(c,obj->ptr,sdslen(obj->ptr)) != REDIS_OK)
             _addReplyObjectToList(c,obj);
     if (obj->encoding == REDIS_ENCODING_RAW) {
         if (_addReplyToBuffer(c,obj->ptr,sdslen(obj->ptr)) != REDIS_OK)
             _addReplyObjectToList(c,obj);
-    } else {
-        /* FIXME: convert the long into string and use _addReplyToBuffer()
-         * instead of calling getDecodedObject. As this place in the
-         * code is too performance critical. */
+    } else if (obj->encoding == REDIS_ENCODING_INT) {
+        /* Optimization: if there is room in the static buffer for 32 bytes
+         * (more than the max chars a 64 bit integer can take as string) we
+         * avoid decoding the object and go for the lower level approach. */
+        if (listLength(c->reply) == 0 && (sizeof(c->buf) - c->bufpos) >= 32) {
+            char buf[32];
+            int len;
+
+            len = ll2string(buf,sizeof(buf),(long)obj->ptr);
+            if (_addReplyToBuffer(c,buf,len) == REDIS_OK)
+                return;
+            /* else... continue with the normal code path, but should never
+             * happen actually since we verified there is room. */
+        }
         obj = getDecodedObject(obj);
         if (_addReplyToBuffer(c,obj->ptr,sdslen(obj->ptr)) != REDIS_OK)
             _addReplyObjectToList(c,obj);
         decrRefCount(obj);
         obj = getDecodedObject(obj);
         if (_addReplyToBuffer(c,obj->ptr,sdslen(obj->ptr)) != REDIS_OK)
             _addReplyObjectToList(c,obj);
         decrRefCount(obj);
+    } else {
+        redisPanic("Wrong obj->encoding in addReply()");
     }
 }
 
 void addReplySds(redisClient *c, sds s) {
     }
 }
 
 void addReplySds(redisClient *c, sds s) {
-    if (_installWriteEvent(c) != REDIS_OK) {
+    if (prepareClientToWrite(c) != REDIS_OK) {
         /* The caller expects the sds to be free'd. */
         sdsfree(s);
         return;
         /* The caller expects the sds to be free'd. */
         sdsfree(s);
         return;
@@ -231,19 +265,19 @@ void addReplySds(redisClient *c, sds s) {
 }
 
 void addReplyString(redisClient *c, char *s, size_t len) {
 }
 
 void addReplyString(redisClient *c, char *s, size_t len) {
-    if (_installWriteEvent(c) != REDIS_OK) return;
+    if (prepareClientToWrite(c) != REDIS_OK) return;
     if (_addReplyToBuffer(c,s,len) != REDIS_OK)
         _addReplyStringToList(c,s,len);
 }
 
     if (_addReplyToBuffer(c,s,len) != REDIS_OK)
         _addReplyStringToList(c,s,len);
 }
 
-void _addReplyError(redisClient *c, char *s, size_t len) {
+void addReplyErrorLength(redisClient *c, char *s, size_t len) {
     addReplyString(c,"-ERR ",5);
     addReplyString(c,s,len);
     addReplyString(c,"\r\n",2);
 }
 
 void addReplyError(redisClient *c, char *err) {
     addReplyString(c,"-ERR ",5);
     addReplyString(c,s,len);
     addReplyString(c,"\r\n",2);
 }
 
 void addReplyError(redisClient *c, char *err) {
-    _addReplyError(c,err,strlen(err));
+    addReplyErrorLength(c,err,strlen(err));
 }
 
 void addReplyErrorFormat(redisClient *c, const char *fmt, ...) {
 }
 
 void addReplyErrorFormat(redisClient *c, const char *fmt, ...) {
@@ -258,18 +292,18 @@ void addReplyErrorFormat(redisClient *c, const char *fmt, ...) {
     for (j = 0; j < l; j++) {
         if (s[j] == '\r' || s[j] == '\n') s[j] = ' ';
     }
     for (j = 0; j < l; j++) {
         if (s[j] == '\r' || s[j] == '\n') s[j] = ' ';
     }
-    _addReplyError(c,s,sdslen(s));
+    addReplyErrorLength(c,s,sdslen(s));
     sdsfree(s);
 }
 
     sdsfree(s);
 }
 
-void _addReplyStatus(redisClient *c, char *s, size_t len) {
+void addReplyStatusLength(redisClient *c, char *s, size_t len) {
     addReplyString(c,"+",1);
     addReplyString(c,s,len);
     addReplyString(c,"\r\n",2);
 }
 
 void addReplyStatus(redisClient *c, char *status) {
     addReplyString(c,"+",1);
     addReplyString(c,s,len);
     addReplyString(c,"\r\n",2);
 }
 
 void addReplyStatus(redisClient *c, char *status) {
-    _addReplyStatus(c,status,strlen(status));
+    addReplyStatusLength(c,status,strlen(status));
 }
 
 void addReplyStatusFormat(redisClient *c, const char *fmt, ...) {
 }
 
 void addReplyStatusFormat(redisClient *c, const char *fmt, ...) {
@@ -277,7 +311,7 @@ void addReplyStatusFormat(redisClient *c, const char *fmt, ...) {
     va_start(ap,fmt);
     sds s = sdscatvprintf(sdsempty(),fmt,ap);
     va_end(ap);
     va_start(ap,fmt);
     sds s = sdscatvprintf(sdsempty(),fmt,ap);
     va_end(ap);
-    _addReplyStatus(c,s,sdslen(s));
+    addReplyStatusLength(c,s,sdslen(s));
     sdsfree(s);
 }
 
     sdsfree(s);
 }
 
@@ -287,7 +321,7 @@ void *addDeferredMultiBulkLength(redisClient *c) {
     /* Note that we install the write event here even if the object is not
      * ready to be sent, since we are sure that before returning to the
      * event loop setDeferredMultiBulkLength() will be called. */
     /* Note that we install the write event here even if the object is not
      * ready to be sent, since we are sure that before returning to the
      * event loop setDeferredMultiBulkLength() will be called. */
-    if (_installWriteEvent(c) != REDIS_OK) return NULL;
+    if (prepareClientToWrite(c) != REDIS_OK) return NULL;
     listAddNodeTail(c->reply,createObject(REDIS_STRING,NULL));
     return listLast(c->reply);
 }
     listAddNodeTail(c->reply,createObject(REDIS_STRING,NULL));
     return listLast(c->reply);
 }
@@ -302,6 +336,7 @@ 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);
     if (ln->next != NULL) {
         next = listNodeValue(ln->next);
 
     if (ln->next != NULL) {
         next = listNodeValue(ln->next);
 
@@ -311,6 +346,7 @@ void setDeferredMultiBulkLength(redisClient *c, void *node, long length) {
             listDelNode(c->reply,ln->next);
         }
     }
             listDelNode(c->reply,ln->next);
         }
     }
+    asyncCloseClientOnOutputBufferLimitReached(c);
 }
 
 /* Add a duble as a bulk reply */
 }
 
 /* Add a duble as a bulk reply */
@@ -324,7 +360,7 @@ void addReplyDouble(redisClient *c, double d) {
 
 /* Add a long long as integer reply or bulk len / multi bulk count.
  * Basically this is used to output <prefix><long long><crlf>. */
 
 /* Add a long long as integer reply or bulk len / multi bulk count.
  * Basically this is used to output <prefix><long long><crlf>. */
-void _addReplyLongLong(redisClient *c, long long ll, char prefix) {
+void addReplyLongLongWithPrefix(redisClient *c, long long ll, char prefix) {
     char buf[128];
     int len;
     buf[0] = prefix;
     char buf[128];
     int len;
     buf[0] = prefix;
@@ -340,11 +376,11 @@ void addReplyLongLong(redisClient *c, long long ll) {
     else if (ll == 1)
         addReply(c,shared.cone);
     else
     else if (ll == 1)
         addReply(c,shared.cone);
     else
-        _addReplyLongLong(c,ll,':');
+        addReplyLongLongWithPrefix(c,ll,':');
 }
 
 void addReplyMultiBulkLen(redisClient *c, long length) {
 }
 
 void addReplyMultiBulkLen(redisClient *c, long length) {
-    _addReplyLongLong(c,length,'*');
+    addReplyLongLongWithPrefix(c,length,'*');
 }
 
 /* Create the length prefix of a bulk reply, example: $2234 */
 }
 
 /* Create the length prefix of a bulk reply, example: $2234 */
@@ -366,7 +402,7 @@ void addReplyBulkLen(redisClient *c, robj *obj) {
             len++;
         }
     }
             len++;
         }
     }
-    _addReplyLongLong(c,len,'$');
+    addReplyLongLongWithPrefix(c,len,'$');
 }
 
 /* Add a Redis Object as a bulk reply */
 }
 
 /* Add a Redis Object as a bulk reply */
@@ -378,7 +414,7 @@ void addReplyBulk(redisClient *c, robj *obj) {
 
 /* Add a C buffer as bulk reply */
 void addReplyBulkCBuffer(redisClient *c, void *p, size_t len) {
 
 /* Add a C buffer as bulk reply */
 void addReplyBulkCBuffer(redisClient *c, void *p, size_t len) {
-    _addReplyLongLong(c,len,'$');
+    addReplyLongLongWithPrefix(c,len,'$');
     addReplyString(c,p,len);
     addReply(c,shared.crlf);
 }
     addReplyString(c,p,len);
     addReply(c,shared.crlf);
 }
@@ -401,6 +437,17 @@ void addReplyBulkLongLong(redisClient *c, long long ll) {
     addReplyBulkCBuffer(c,buf,len);
 }
 
     addReplyBulkCBuffer(c,buf,len);
 }
 
+/* Copy 'src' client output buffers into 'dst' client output buffers.
+ * The function takes care of freeing the old output buffers of the
+ * destination client. */
+void copyClientOutputBuffer(redisClient *dst, redisClient *src) {
+    listRelease(dst->reply);
+    dst->reply = listDup(src->reply);
+    memcpy(dst->buf,src->buf,src->bufpos);
+    dst->bufpos = src->bufpos;
+    dst->reply_bytes = src->reply_bytes;
+}
+
 static void acceptCommonHandler(int fd) {
     redisClient *c;
     if ((c = createClient(fd)) == NULL) {
 static void acceptCommonHandler(int fd) {
     redisClient *c;
     if ((c = createClient(fd)) == NULL) {
@@ -469,6 +516,9 @@ static void freeClientArgv(redisClient *c) {
 void freeClient(redisClient *c) {
     listNode *ln;
 
 void freeClient(redisClient *c) {
     listNode *ln;
 
+    /* If this is marked as current client unset it */
+    if (server.current_client == c) server.current_client = NULL;
+
     /* Note that if the client we are freeing is blocked into a blocking
      * call, we have to set querybuf to NULL *before* to call
      * unblockClientWaitingData() to avoid processInputBuffer() will get
     /* Note that if the client we are freeing is blocked into a blocking
      * call, we have to set querybuf to NULL *before* to call
      * unblockClientWaitingData() to avoid processInputBuffer() will get
@@ -519,7 +569,7 @@ void freeClient(redisClient *c) {
     /* Case 2: we lost the connection with the master. */
     if (c->flags & REDIS_MASTER) {
         server.master = NULL;
     /* Case 2: we lost the connection with the master. */
     if (c->flags & REDIS_MASTER) {
         server.master = NULL;
-        server.replstate = REDIS_REPL_CONNECT;
+        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
         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
@@ -537,12 +587,42 @@ void freeClient(redisClient *c) {
             }
         }
     }
             }
         }
     }
+
+    /* If this client was scheduled for async freeing we need to remove it
+     * from the queue. */
+    if (c->flags & REDIS_CLOSE_ASAP) {
+        ln = listSearchKey(server.clients_to_close,c);
+        redisAssert(ln != NULL);
+        listDelNode(server.clients_to_close,ln);
+    }
+
     /* Release memory */
     zfree(c->argv);
     freeClientMultiState(c);
     zfree(c);
 }
 
     /* Release memory */
     zfree(c->argv);
     freeClientMultiState(c);
     zfree(c);
 }
 
+/* Schedule a client to free it at a safe time in the serverCron() function.
+ * This function is useful when we need to terminate a client but we are in
+ * a context where calling freeClient() is not possible, because the client
+ * should be valid for the continuation of the flow of the program. */
+void freeClientAsync(redisClient *c) {
+    if (c->flags & REDIS_CLOSE_ASAP) return;
+    c->flags |= REDIS_CLOSE_ASAP;
+    listAddNodeTail(server.clients_to_close,c);
+}
+
+void freeClientsInAsyncFreeQueue(void) {
+    while (listLength(server.clients_to_close)) {
+        listNode *ln = listFirst(server.clients_to_close);
+        redisClient *c = listNodeValue(ln);
+
+        c->flags &= ~REDIS_CLOSE_ASAP;
+        freeClient(c);
+        listDelNode(server.clients_to_close,ln);
+    }
+}
+
 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;
@@ -591,6 +671,7 @@ 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;
             }
         }
         /* Note that we avoid to send more thank REDIS_MAX_WRITE_PER_EVENT
             }
         }
         /* Note that we avoid to send more thank REDIS_MAX_WRITE_PER_EVENT
@@ -665,8 +746,13 @@ int processInlineBuffer(redisClient *c) {
     size_t querylen;
 
     /* Nothing to do without a \r\n */
     size_t querylen;
 
     /* Nothing to do without a \r\n */
-    if (newline == NULL)
+    if (newline == NULL) {
+        if (sdslen(c->querybuf) > REDIS_INLINE_MAX_SIZE) {
+            addReplyError(c,"Protocol error: too big inline request");
+            setProtocolError(c,0);
+        }
         return REDIS_ERR;
         return REDIS_ERR;
+    }
 
     /* Split the input buffer up to the \r\n */
     querylen = newline-(c->querybuf);
 
     /* Split the input buffer up to the \r\n */
     querylen = newline-(c->querybuf);
@@ -716,8 +802,13 @@ int processMultibulkBuffer(redisClient *c) {
 
         /* 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');
-        if (newline == NULL)
+        if (newline == NULL) {
+            if (sdslen(c->querybuf) > REDIS_INLINE_MAX_SIZE) {
+                addReplyError(c,"Protocol error: too big mbulk count string");
+                setProtocolError(c,0);
+            }
             return REDIS_ERR;
             return REDIS_ERR;
+        }
 
         /* Buffer should also contain \n */
         if (newline-(c->querybuf) > ((signed)sdslen(c->querybuf)-2))
 
         /* Buffer should also contain \n */
         if (newline-(c->querybuf) > ((signed)sdslen(c->querybuf)-2))
@@ -751,8 +842,13 @@ int processMultibulkBuffer(redisClient *c) {
         /* Read bulk length if unknown */
         if (c->bulklen == -1) {
             newline = strchr(c->querybuf+pos,'\r');
         /* Read bulk length if unknown */
         if (c->bulklen == -1) {
             newline = strchr(c->querybuf+pos,'\r');
-            if (newline == NULL)
+            if (newline == NULL) {
+                if (sdslen(c->querybuf) > REDIS_INLINE_MAX_SIZE) {
+                    addReplyError(c,"Protocol error: too big bulk count string");
+                    setProtocolError(c,0);
+                }
                 break;
                 break;
+            }
 
             /* Buffer should also contain \n */
             if (newline-(c->querybuf) > ((signed)sdslen(c->querybuf)-2))
 
             /* Buffer should also contain \n */
             if (newline-(c->querybuf) > ((signed)sdslen(c->querybuf)-2))
@@ -821,9 +917,9 @@ int processMultibulkBuffer(redisClient *c) {
     if (pos) c->querybuf = sdsrange(c->querybuf,pos,-1);
 
     /* We're done when c->multibulk == 0 */
     if (pos) c->querybuf = sdsrange(c->querybuf,pos,-1);
 
     /* We're done when c->multibulk == 0 */
-    if (c->multibulklen == 0) {
-        return REDIS_OK;
-    }
+    if (c->multibulklen == 0) return REDIS_OK;
+
+    /* Still not read to process the command */
     return REDIS_ERR;
 }
 
     return REDIS_ERR;
 }
 
@@ -873,6 +969,7 @@ void readQueryFromClient(aeEventLoop *el, int fd, void *privdata, int mask) {
     REDIS_NOTUSED(el);
     REDIS_NOTUSED(mask);
 
     REDIS_NOTUSED(el);
     REDIS_NOTUSED(mask);
 
+    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
     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
@@ -908,6 +1005,7 @@ void readQueryFromClient(aeEventLoop *el, int fd, void *privdata, int mask) {
         sdsIncrLen(c->querybuf,nread);
         c->lastinteraction = time(NULL);
     } else {
         sdsIncrLen(c->querybuf,nread);
         c->lastinteraction = time(NULL);
     } else {
+        server.current_client = NULL;
         return;
     }
     if (sdslen(c->querybuf) > server.client_max_querybuf_len) {
         return;
     }
     if (sdslen(c->querybuf) > server.client_max_querybuf_len) {
@@ -921,6 +1019,7 @@ void readQueryFromClient(aeEventLoop *el, int fd, void *privdata, int mask) {
         return;
     }
     processInputBuffer(c);
         return;
     }
     processInputBuffer(c);
+    server.current_client = NULL;
 }
 
 void getClientsMaxBuffers(unsigned long *longest_output_list,
 }
 
 void getClientsMaxBuffers(unsigned long *longest_output_list,
@@ -966,6 +1065,7 @@ sds getClientInfoString(redisClient *client) {
     if (client->flags & REDIS_DIRTY_CAS) *p++ = 'd';
     if (client->flags & REDIS_CLOSE_AFTER_REPLY) *p++ = 'c';
     if (client->flags & REDIS_UNBLOCKED) *p++ = 'u';
     if (client->flags & REDIS_DIRTY_CAS) *p++ = 'd';
     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 (p == flags) *p++ = 'N';
     *p++ = '\0';
 
     if (p == flags) *p++ = 'N';
     *p++ = '\0';
 
@@ -975,7 +1075,7 @@ 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 events=%s cmd=%s",
+        "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),
         flags,
         ip,port,client->fd,
         (long)(now - client->lastinteraction),
         flags,
@@ -985,6 +1085,7 @@ sds getClientInfoString(redisClient *client) {
         (unsigned long) sdslen(client->querybuf),
         (unsigned long) client->bufpos,
         (unsigned long) listLength(client->reply),
         (unsigned long) sdslen(client->querybuf),
         (unsigned long) client->bufpos,
         (unsigned long) listLength(client->reply),
+        getClientOutputBufferMemoryUsage(client),
         events,
         client->lastcmd ? client->lastcmd->name : "NULL");
 }
         events,
         client->lastcmd ? client->lastcmd->name : "NULL");
 }
@@ -1089,3 +1190,111 @@ void rewriteClientCommandArgument(redisClient *c, int i, robj *newval) {
         redisAssertWithInfo(c,NULL,c->cmd != NULL);
     }
 }
         redisAssertWithInfo(c,NULL,c->cmd != NULL);
     }
 }
+
+/* This function returns the number of bytes that Redis is virtually
+ * using to store the reply still not read by the client.
+ * It is "virtual" since the reply output list may contain objects that
+ * are shared and are not really using additional memory.
+ *
+ * The function returns the total sum of the length of all the objects
+ * stored in the output list, plus the memory used to allocate every
+ * list node. The static reply buffer is not taken into account since it
+ * is allocated anyway.
+ *
+ * Note: this function is very fast so can be called as many time as
+ * 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);
+
+    return c->reply_bytes + (list_item_size*listLength(c->reply));
+}
+
+/* Get the class of a client, used in order to envorce limits to different
+ * classes of clients.
+ *
+ * The function will return one of the following:
+ * REDIS_CLIENT_LIMIT_CLASS_NORMAL -> Normal client
+ * REDIS_CLIENT_LIMIT_CLASS_SLAVE  -> Slave or client executing MONITOR command
+ * REDIS_CLIENT_LIMIT_CLASS_PUBSUB -> Client subscribed to Pub/Sub channels
+ */
+int getClientLimitClass(redisClient *c) {
+    if (c->flags & REDIS_SLAVE) return REDIS_CLIENT_LIMIT_CLASS_SLAVE;
+    if (dictSize(c->pubsub_channels) || listLength(c->pubsub_patterns))
+        return REDIS_CLIENT_LIMIT_CLASS_PUBSUB;
+    return REDIS_CLIENT_LIMIT_CLASS_NORMAL;
+}
+
+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_PUBSUB;
+    else return -1;
+}
+
+char *getClientLimitClassName(int class) {
+    switch(class) {
+    case REDIS_CLIENT_LIMIT_CLASS_NORMAL:   return "normal";
+    case REDIS_CLIENT_LIMIT_CLASS_SLAVE:    return "slave";
+    case REDIS_CLIENT_LIMIT_CLASS_PUBSUB:   return "pubsub";
+    default:                                return NULL;
+    }
+}
+
+/* The function checks if the client reached output buffer soft or hard
+ * limit, and also update the state needed to check the soft limit as
+ * a side effect.
+ *
+ * Return value: non-zero if the client reached the soft or the hard limit.
+ *               Otherwise zero is returned. */
+int checkClientOutputBufferLimits(redisClient *c) {
+    int soft = 0, hard = 0, class;
+    unsigned long used_mem = getClientOutputBufferMemoryUsage(c);
+
+    class = getClientLimitClass(c);
+    if (server.client_obuf_limits[class].hard_limit_bytes &&
+        used_mem >= server.client_obuf_limits[class].hard_limit_bytes)
+        hard = 1;
+    if (server.client_obuf_limits[class].soft_limit_bytes &&
+        used_mem >= server.client_obuf_limits[class].soft_limit_bytes)
+        soft = 1;
+
+    /* We need to check if the soft limit is reached continuously for the
+     * specified amount of seconds. */
+    if (soft) {
+        if (c->obuf_soft_limit_reached_time == 0) {
+            c->obuf_soft_limit_reached_time = server.unixtime;
+            soft = 0; /* First time we see the soft limit reached */
+        } else {
+            time_t elapsed = server.unixtime - c->obuf_soft_limit_reached_time;
+
+            if (elapsed <=
+                server.client_obuf_limits[class].soft_limit_seconds) {
+                soft = 0; /* The client still did not reached the max number of
+                             seconds for the soft limit to be considered
+                             reached. */
+            }
+        }
+    } else {
+        c->obuf_soft_limit_reached_time = 0;
+    }
+    return soft || hard;
+}
+
+/* Asynchronously close a client if soft or hard limit is reached on the
+ * output buffer size. The caller can check if the client will be closed
+ * checking if the client REDIS_CLOSE_ASAP flag is set.
+ *
+ * Note: we need to close the client asynchronously because this function is
+ * 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;
+    if (checkClientOutputBufferLimits(c)) {
+        sds client = getClientInfoString(c);
+
+        freeClientAsync(c);
+        redisLog(REDIS_WARNING,"Client %s scheduled to be closed ASAP for overcoming of output buffer limits.", client);
+        sdsfree(client);
+    }
+}