Issue 61: reqs_per_event handling (-R) is incorrect leading to client…

… lockups

memcached.c

@@ -2136,6 +2136,7 @@ static void server_stats(ADD_STAT add_stats, conn *c) {
     APPEND_STAT("accepting_conns", "%u", stats.accepting_conns);
     APPEND_STAT("listen_disabled_num", "%llu", (unsigned long long)stats.listen_disabled_num);
     APPEND_STAT("threads", "%d", settings.num_threads);
+    APPEND_STAT("conn_yields", "%llu", (unsigned long long)thread_stats.conn_yields);
     STATS_UNLOCK();
 }
 
@@ -3199,10 +3200,7 @@ static void drive_machine(conn *c) {
                 conn_set_state(c, conn_waiting);
                 break;
             case READ_DATA_RECEIVED:
-             /* Only process nreqs at a time to avoid starving other
-                connections */
-                if (--nreqs)
-                    conn_set_state(c, conn_parse_cmd);
+                conn_set_state(c, conn_parse_cmd);
                 break;
             case READ_ERROR:
                 conn_set_state(c, conn_closing);
@@ -3222,7 +3220,31 @@ static void drive_machine(conn *c) {
             break;
 
         case conn_new_cmd:
-            reset_cmd_handler(c);
+            /* Only process nreqs at a time to avoid starving other
+               connections */
+
+            --nreqs;
+            if (nreqs >= 0) {
+                reset_cmd_handler(c);
+            } else {
+                pthread_mutex_lock(&c->thread->stats.mutex);
+                c->thread->stats.conn_yields++;
+                pthread_mutex_unlock(&c->thread->stats.mutex);
+                if (c->rbytes > 0) {
+                    /* We have already read in data into the input buffer,
+                       so libevent will most likely not signal read events
+                       on the socket (unless more data is available. As a
+                       hack we should just put in a request to write data,
+                       because that should be possible ;-)
+                    */
+                    if (!update_event(c, EV_WRITE | EV_PERSIST)) {
+                        if (settings.verbose > 0)
+                            fprintf(stderr, "Couldn't update event\n");
+                        conn_set_state(c, conn_closing);
+                    }
+                }
+                stop = true;
+            }
             break;
 
         case conn_nread:

memcached.h

@@ -210,6 +210,7 @@ struct thread_stats {
     uint64_t          bytes_read;
     uint64_t          bytes_written;
     uint64_t          flush_cmds;
+    uint64_t          conn_yields; /* # of yields for connections (-R option)*/
     struct slab_stats slab_stats[MAX_NUMBER_OF_SLAB_CLASSES];
 };
 

t/issue_61.t

@@ -0,0 +1,20 @@
+#!/usr/bin/perl
+
+use strict;
+use Test::More tests => 7;
+use FindBin qw($Bin);
+use lib "$Bin/lib";
+use MemcachedTest;
+
+my $server = new_memcached("-R 1");
+my $sock = $server->sock;
+
+print $sock "set foobar 0 0 5\r\nBubba\r\nset foobar 0 0 5\r\nBubba\r\nset foobar 0 0 5\r\nBubba\r\nset foobar 0 0 5\r\nBubba\r\nset foobar 0 0 5\r\nBubba\r\nset foobar 0 0 5\r\nBubba\r\n";
+is (scalar <$sock>, "STORED\r\n", "stored foobar");
+is (scalar <$sock>, "STORED\r\n", "stored foobar");
+is (scalar <$sock>, "STORED\r\n", "stored foobar");
+is (scalar <$sock>, "STORED\r\n", "stored foobar");
+is (scalar <$sock>, "STORED\r\n", "stored foobar");
+is (scalar <$sock>, "STORED\r\n", "stored foobar");
+my $stats = mem_stats($sock);
+is ($stats->{"conn_yields"}, "5", "Got a decent number of yields");

t/stats.t

@@ -47,7 +47,7 @@ my $sock = $server->sock;
 my $stats = mem_stats($sock);
 
 # Test number of keys
-is(scalar(keys(%$stats)), 34, "34 stats values");
+is(scalar(keys(%$stats)), 35, "35 stats values");
 
 # Test initial state
 foreach my $key (qw(curr_items total_items bytes cmd_get cmd_set get_hits evictions get_misses

thread.c

@@ -485,6 +485,7 @@ void threadlocal_stats_reset(void) {
         threads[ii].stats.bytes_read = 0;
         threads[ii].stats.bytes_written = 0;
         threads[ii].stats.flush_cmds = 0;
+        threads[ii].stats.conn_yields = 0;
 
         for(sid = 0; sid < MAX_NUMBER_OF_SLAB_CLASSES; sid++) {
             threads[ii].stats.slab_stats[sid].set_cmds = 0;
@@ -512,6 +513,7 @@ void threadlocal_stats_aggregate(struct thread_stats *stats) {
     stats->bytes_written = 0;
     stats->bytes_read = 0;
     stats->flush_cmds = 0;
+    stats->conn_yields = 0;
 
     memset(stats->slab_stats, 0,
            sizeof(struct slab_stats) * MAX_NUMBER_OF_SLAB_CLASSES);
@@ -528,6 +530,7 @@ void threadlocal_stats_aggregate(struct thread_stats *stats) {
         stats->bytes_read += threads[ii].stats.bytes_read;
         stats->bytes_written += threads[ii].stats.bytes_written;
         stats->flush_cmds += threads[ii].stats.flush_cmds;
+        stats->conn_yields += threads[ii].stats.conn_yields;
 
         for (sid = 0; sid < MAX_NUMBER_OF_SLAB_CLASSES; sid++) {
             stats->slab_stats[sid].set_cmds +=