Fixes to c->reply_bytes computation, and debug messages to closely study the...

Fixes to c->reply_bytes computation, and debug messages to closely study the behavior of memory pressure + slaves + maxmemory + blocked slaves.

src/networking.c

@@ -3,6 +3,14 @@
 
 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;
@@ -117,6 +125,7 @@ void _addReplyObjectToList(redisClient *c, robj *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));
 
@@ -124,14 +133,16 @@ void _addReplyObjectToList(redisClient *c, robj *o) {
         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));
+            c->reply_bytes += zmalloc_size_sds(tail->ptr);
         } else {
             incrRefCount(o);
             listAddNodeTail(c->reply,o);
+            c->reply_bytes += zmalloc_size_sds(o->ptr);
         }
     }
-    c->reply_bytes += zmalloc_size(o->ptr);
 }
 
 /* This method takes responsibility over the sds. When it is no longer
@@ -144,9 +155,9 @@ void _addReplySdsToList(redisClient *c, sds s) {
         return;
     }
 
-    c->reply_bytes += zmalloc_size(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));
 
@@ -154,11 +165,14 @@ void _addReplySdsToList(redisClient *c, sds s) {
         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));
+            c->reply_bytes += zmalloc_size_sds(tail->ptr);
             sdsfree(s);
         } else {
             listAddNodeTail(c->reply,createObject(REDIS_STRING,s));
+            c->reply_bytes += zmalloc_size_sds(s);
         }
     }
 }
@@ -172,7 +186,7 @@ void _addReplyStringToList(redisClient *c, char *s, size_t len) {
         robj *o = createStringObject(s,len);
 
         listAddNodeTail(c->reply,o);
-        c->reply_bytes += zmalloc_size(o->ptr);
+        c->reply_bytes += zmalloc_size_sds(o->ptr);
     } else {
         tail = listNodeValue(listLast(c->reply));
 
@@ -180,15 +194,15 @@ void _addReplyStringToList(redisClient *c, char *s, size_t len) {
         if (tail->ptr != NULL &&
             sdslen(tail->ptr)+len <= REDIS_REPLY_CHUNK_BYTES)
         {
-            c->reply_bytes -= zmalloc_size(tail->ptr);
+            c->reply_bytes -= zmalloc_size_sds(tail->ptr);
             tail = dupLastObjectIfNeeded(c->reply);
             tail->ptr = sdscatlen(tail->ptr,s,len);
-            c->reply_bytes += zmalloc_size(tail->ptr);
+            c->reply_bytes += zmalloc_size_sds(tail->ptr);
         } else {
             robj *o = createStringObject(s,len);
 
             listAddNodeTail(c->reply,o);
-            c->reply_bytes += zmalloc_size(o->ptr);
+            c->reply_bytes += zmalloc_size_sds(o->ptr);
         }
     }
 }
@@ -302,7 +316,7 @@ void setDeferredMultiBulkLength(redisClient *c, void *node, long length) {
 
     len = listNodeValue(ln);
     len->ptr = sdscatprintf(sdsempty(),"*%ld\r\n",length);
-    c->reply_bytes += zmalloc_size(len->ptr);
+    c->reply_bytes += zmalloc_size_sds(len->ptr);
     if (ln->next != NULL) {
         next = listNodeValue(ln->next);
 
@@ -605,7 +619,7 @@ void sendReplyToClient(aeEventLoop *el, int fd, void *privdata, int mask) {
         } else {
             o = listNodeValue(listFirst(c->reply));
             objlen = sdslen(o->ptr);
-            objmem = zmalloc_size(o->ptr);
+            objmem = zmalloc_size_sds(o->ptr);
 
             if (objlen == 0) {
                 listDelNode(c->reply,listFirst(c->reply));

src/redis.c

@@ -1546,10 +1546,18 @@ void monitorCommand(redisClient *c) {
 int freeMemoryIfNeeded(void) {
     size_t mem_used, mem_tofree, mem_freed;
     int slaves = listLength(server.slaves);
+    static time_t xt;
+    int debug = 0;
+
+    if (xt != time(NULL)) {
+        debug = 1;
+        xt = time(NULL);
+    }
 
     /* Remove the size of slaves output buffers and AOF buffer from the
      * count of used memory. */
     mem_used = zmalloc_used_memory();
+    if (debug) printf("used_full: %zu\n", mem_used);
     if (slaves) {
         listIter li;
         listNode *ln;
@@ -1564,6 +1572,7 @@ int freeMemoryIfNeeded(void) {
                 mem_used -= obuf_bytes;
         }
     }
+    if (debug) printf("used_nosl: %zu\n", mem_used);
     if (server.aof_state != REDIS_AOF_OFF) {
         mem_used -= sdslen(server.aof_buf);
         mem_used -= sdslen(server.aof_rewrite_buf);
@@ -1578,6 +1587,7 @@ int freeMemoryIfNeeded(void) {
     /* Compute how much memory we need to free. */
     mem_tofree = mem_used - server.maxmemory;
     mem_freed = 0;
+    if (debug) printf("tofree: %zu\n", mem_tofree);
     while (mem_freed < mem_tofree) {
         int j, k, keys_freed = 0;
 
@@ -1679,8 +1689,12 @@ int freeMemoryIfNeeded(void) {
                 if (slaves) flushSlavesOutputBuffers();
             }
         }
-        if (!keys_freed) return REDIS_ERR; /* nothing to free... */
+        if (!keys_freed) {
+            if (debug) printf("-freed: %zu\n\n", mem_freed);
+            return REDIS_ERR; /* nothing to free... */
+        }
     }
+    if (debug) printf("+freed: %zu\n\n", mem_freed);
     return REDIS_OK;
 }