fix leak in merged resp/read buffers

The list grows toward next, not prev. Also wasn't zeroing out next ptr.

Also didn't unmark free for first resp object. (thanks Prudhviraj!)

Adds a couple counters so users can tell if something is wrong as well.
response_obj_count is solely for response objects in-flight, they
should not be held when idle (except for one bundle per worker thread).

4 files changed, 28 insertions, 6 deletions

diff --git a/doc/protocol.txt b/doc/protocol.txt
index 36fce44..737937c 100644
--- a/doc/protocol.txt
+++ b/doc/protocol.txt
@@ -1203,6 +1203,9 @@ integers separated by a colon (treat this as a floating point number).
 | connection_structures | 32u     | Number of connection structures allocated |
 |                       |         | by the server                             |
 | response_obj_oom      | 64u     | Connections closed by lack of memory      |
+| response_obj_count    | 64u     | Total response objects in use             |
+| response_obj_bytes    | 64u     | Total bytes used for resp. objects. is a  |
+|                       |         | subset of bytes from read_buf_bytes.      |
 | read_buf_count        | 64u     | Total read/resp buffers allocated         |
 | read_buf_bytes        | 64u     | Total read/resp buffer bytes allocated    |
 | read_buf_bytes_free   | 64u     | Total read/resp buffer bytes cached       |
diff --git a/memcached.c b/memcached.c
index bb56a5b..095bca1 100644
--- a/memcached.c
+++ b/memcached.c
@@ -1060,12 +1060,13 @@ static mc_resp* resp_allocate(conn *c) {
             b->refcount++;
             resp->free = false;
             if (b->refcount == MAX_RESP_PER_BUNDLE) {
-                assert(b->next == NULL);
+                assert(b->prev == NULL);
                 // We only allocate off the head. Assign new head.
-                th->open_bundle = b->prev;
+                th->open_bundle = b->next;
                 // Remove ourselves from the list.
-                if (b->prev) {
-                    b->prev->next = 0;
+                if (b->next) {
+                    b->next->prev = 0;
+                    b->next = 0;
                 }
             }
         }
@@ -1075,6 +1076,9 @@ static mc_resp* resp_allocate(conn *c) {
         assert(th->open_bundle == NULL);
         b = do_cache_alloc(th->rbuf_cache);
         if (b) {
+            THR_STATS_LOCK(c);
+            c->thread->stats.response_obj_bytes += READ_BUFFER_SIZE;
+            THR_STATS_UNLOCK(c);
             b->next_check = 1;
             b->refcount = 1;
             for (int i = 0; i < MAX_RESP_PER_BUNDLE; i++) {
@@ -1085,6 +1089,7 @@ static mc_resp* resp_allocate(conn *c) {
             b->prev = 0;
             th->open_bundle = b;
             resp = &b->r[0];
+            resp->free = false;
         } else {
             return NULL;
         }
@@ -1119,6 +1124,9 @@ static void resp_free(conn *c, mc_resp *resp) {
 
             // Now completely done with this buffer.
             do_cache_free(th->rbuf_cache, b);
+            THR_STATS_LOCK(c);
+            c->thread->stats.response_obj_bytes -= READ_BUFFER_SIZE;
+            THR_STATS_UNLOCK(c);
         }
     } else {
         mc_resp_bundle **head = &th->open_bundle;
@@ -1144,6 +1152,10 @@ static bool resp_start(conn *c) {
         THR_STATS_UNLOCK(c);
         return false;
     }
+    // handling the stats counters here to simplify testing
+    THR_STATS_LOCK(c);
+    c->thread->stats.response_obj_count++;
+    THR_STATS_UNLOCK(c);
     // Skip zeroing the bundle pointer at the start.
     // TODO: this line is here temporarily to make the code easy to disable.
     // when it's more mature, move the memset into resp_allocate() and have it
@@ -1190,6 +1202,9 @@ static mc_resp* resp_finish(conn *c, mc_resp *resp) {
         c->resp = NULL;
     }
     resp_free(c, resp);
+    THR_STATS_LOCK(c);
+    c->thread->stats.response_obj_count--;
+    THR_STATS_UNLOCK(c);
     return next;
 }
 
@@ -3235,6 +3250,8 @@ static void server_stats(ADD_STAT add_stats, conn *c) {
     }
     APPEND_STAT("connection_structures", "%u", stats_state.conn_structs);
     APPEND_STAT("response_obj_oom", "%llu", (unsigned long long)thread_stats.response_obj_oom);
+    APPEND_STAT("response_obj_count", "%llu", (unsigned long long)thread_stats.response_obj_count);
+    APPEND_STAT("response_obj_bytes", "%llu", (unsigned long long)thread_stats.response_obj_bytes);
     APPEND_STAT("read_buf_count", "%llu", (unsigned long long)thread_stats.read_buf_count);
     APPEND_STAT("read_buf_bytes", "%llu", (unsigned long long)thread_stats.read_buf_bytes);
     APPEND_STAT("read_buf_bytes_free", "%llu", (unsigned long long)thread_stats.read_buf_bytes_free);
diff --git a/memcached.h b/memcached.h
index 2cafed9..ec1e0e5 100644
--- a/memcached.h
+++ b/memcached.h
@@ -302,6 +302,8 @@ struct slab_stats {
     X(auth_errors) \
     X(idle_kicks) /* idle connections killed */ \
     X(response_obj_oom) \
+    X(response_obj_count) \
+    X(response_obj_bytes) \
     X(read_buf_oom)
 
 #ifdef EXTSTORE
diff --git a/t/stats.t b/t/stats.t
index ad51e5f..a84dc05 100755
--- a/t/stats.t
+++ b/t/stats.t
@@ -28,9 +28,9 @@ if (MemcachedTest::enabled_tls_testing()) {
     # when TLS is enabled, stats contains additional keys:
     #   - ssl_handshake_errors
     #   - time_since_server_cert_refresh
-    is(scalar(keys(%$stats)), 78, "expected count of stats values");
+    is(scalar(keys(%$stats)), 80, "expected count of stats values");
 } else {
-    is(scalar(keys(%$stats)), 76, "expected count of stats values");
+    is(scalar(keys(%$stats)), 78, "expected count of stats values");
 }
 
 # Test initial state