proxy: add request and buffer memory limits

Adds:
mcp.active_req_limit(count)
mcp.buffer_memory_limit(kilobytes)

Divides by the number of worker threads and creates a per-worker-thread
limit for the number of concurrent proxy requests, and how many bytes
used specifically for value bytes. This does not represent total memory
usage but will be close.

Buffer memory for inbound set requests is not accounted for until after
the object has been read from the socket; to be improved in a future
update. This should be fine unless clients send just the SET request and
then hang without sending further data.

Limits should be live-adjustable via configuration reloads.

1 files changed, 176 insertions, 1 deletions

diff --git a/t/proxyconfig.t b/t/proxyconfig.t
index 8c9332c..73b0169 100644
--- a/t/proxyconfig.t
+++ b/t/proxyconfig.t
@@ -60,7 +60,7 @@ sub wait_reload {
 }
 
 my @mocksrvs = ();
-diag "making mock servers";
+#diag "making mock servers";
 for my $port (11511, 11512, 11513) {
     my $srv = mock_server($port);
     ok(defined $srv, "mock server created");
@@ -267,7 +267,182 @@ is(<$watcher>, "OK\r\n", "watcher enabled");
 
 }
 
+###
+# diag "starting proxy again from scratch"
+###
+
+# TODO: probably time to abstract the entire "start the server with mocked
+# listeners" routine.
+$watcher = undef;
+write_modefile('return "reqlimit"');
+$p_srv->stop;
+while (1) {
+    if ($p_srv->is_running) {
+        sleep 1;
+    } else {
+        ok(!$p_srv->is_running, "stopped proxy");
+        last;
+    }
+}
+
+@mocksrvs = ();
+# re-create the mock servers so we get clean connects, the previous
+# backends could be reconnecting still.
+for my $port (11511, 11512, 11513) {
+    my $srv = mock_server($port);
+    ok(defined $srv, "mock server created");
+    push(@mocksrvs, $srv);
+}
+
+# Start up a clean server.
+# Since limits are per worker thread, cut the worker threads down to 1 to ease
+# testing.
+$p_srv = new_memcached('-o proxy_config=./t/proxyconfig.lua -l 127.0.0.1 -t 1', 11510);
+$ps = $p_srv->sock;
+$ps->autoflush(1);
+
+{
+    for my $msrv ($mocksrvs[0], $mocksrvs[1], $mocksrvs[2]) {
+        my $be = $msrv->accept();
+        $be->autoflush(1);
+        ok(defined $be, "mock backend created");
+        push(@mbe, $be);
+    }
+
+    for my $be (@mbe) {
+        like(<$be>, qr/version/, "received version command");
+        print $be "VERSION 1.0.0-mock\r\n";
+    }
+
+    my $stats = mem_stats($ps, 'proxy');
+    isnt($stats->{active_req_limit}, 0, "active request limit is set");
+
+    # active request limit is 4, pipeline 6 requests and ensure the last two
+    # get junked
+    my $cmd = '';
+    for ('a', 'b', 'c', 'd', 'e', 'f') {
+        $cmd .= "mg /test/$_\r\n";
+    }
+    print $ps $cmd;
+
+    # Lua config only sends commands to the first backend for this test.
+    my $be = $mbe[0];
+    for (1 .. 4) {
+        like(<$be>, qr/^mg \/test\/\w\r\n$/, "backend received mg");
+        print $be "EN\r\n";
+    }
+    my $s = IO::Select->new();
+    $s->add($be);
+    my @readable = $s->can_read(0.25);
+    is(scalar @readable, 0, "no more pending reads on backend");
+
+    for (1 .. 4) {
+        is(scalar <$ps>, "EN\r\n", "received miss from backend");
+    }
+
+    is(scalar <$ps>, "SERVER_ERROR active request limit reached\r\n", "got error back");
+    is(scalar <$ps>, "SERVER_ERROR active request limit reached\r\n", "got two limit errors");
+
+    # Test turning the limit back off.
+    write_modefile('return "noreqlimit"');
+    $watcher = $p_srv->new_sock;
+    print $watcher "watch proxyevents\n";
+    is(<$watcher>, "OK\r\n", "watcher enabled");
+    $p_srv->reload();
+    wait_reload($watcher);
+
+    $stats = mem_stats($ps, 'proxy');
+    is($stats->{active_req_limit}, 0, "active request limit unset");
+
+    $cmd = '';
+    for ('a', 'b', 'c', 'd', 'e', 'f') {
+        $cmd .= "mg /test/$_\r\n";
+    }
+    print $ps $cmd;
+    for (1 .. 6) {
+        like(<$be>, qr/^mg \/test\/\w\r\n$/, "backend received mg");
+        print $be "EN\r\n";
+    }
+    for (1 .. 6) {
+        is(scalar <$ps>, "EN\r\n", "received miss from backend");
+    }
+}
+
+{
+    # Test the buffer memory limiter.
+    # - limit per worker will be 1/t global limit
+    write_modefile('return "buflimit"');
+    $p_srv->reload();
+    wait_reload($watcher);
+    # Get a secondary client to trample limit.
+    my $sps = $p_srv->new_sock;
+
+    my $stats = mem_stats($ps, 'proxy');
+    isnt($stats->{buffer_memory_limit}, 0, "buf mem limit is set");
+
+    # - test SET commands with values, but nothing being read on backend
+    my $data = 'x' x 30000;
+    my $cmd = "ms foo 30000 T30\r\n" . $data . "\r\n";
+    print $ps $cmd;
+
+    my $be = $mbe[0];
+    my $s = IO::Select->new;
+    $s->add($be);
+    # Wait until the backend has the request queued, then send the second one.
+    my @readable = $s->can_read(1);
+    print $sps $cmd;
+
+    my $res;
+    is(scalar <$be>, "ms foo 30000 T30\r\n", "received first ms");
+    $res = scalar <$be>;
+    print $be "HD\r\n";
+
+    # The second request should have been caught by the memory limiter
+    is(scalar <$sps>, "SERVER_ERROR out of memory\r\n", "got server error");
+    # FIXME: The original response cannot succeed because we cannot allocate
+    # enough memory to read the response from the backend.
+    # This is conveniently testing both paths right here but I would prefer
+    # something better.
+    # TODO: need to see if it's possible to surface an OOM from the backend
+    # handler, but that requires more rewiring.
+    is(scalar <$ps>, "SERVER_ERROR backend failure\r\n", "first request succeeded");
+
+    # Backend gets killed from a read OOM, so we need to re-establish.
+    $mbe[0] = $mocksrvs[0]->accept();
+    $be = $mbe[0];
+    $be->autoflush(1);
+    like(<$be>, qr/version/, "received version command");
+    print $be "VERSION 1.0.0-mock\r\n";
+    like(<$watcher>, qr/error=outofmemory/, "OOM log line");
+
+    # Memory limits won't drop until the garbage collectors run, which
+    # requires a bit more push, so instead we raise the limits here so we can
+    # retry from the pre-existing connections to test swallow mode.
+    write_modefile('return "buflimit2"');
+    $p_srv->reload();
+    wait_reload($watcher);
+
+    # Test sending another request down both pipes to ensure they still work.
+    $cmd = "ms foo 2 T30\r\nhi\r\n";
+    print $ps $cmd;
+    is(scalar <$be>, "ms foo 2 T30\r\n", "client works after oom");
+    is(scalar <$be>, "hi\r\n", "client works after oom");
+    print $be "HD\r\n";
+    is(scalar <$ps>, "HD\r\n", "client received resp after oom");
+    print $sps $cmd;
+    is(scalar <$be>, "ms foo 2 T30\r\n", "client works after oom");
+    is(scalar <$be>, "hi\r\n", "client works after oom");
+    print $be "HD\r\n";
+    is(scalar <$sps>, "HD\r\n", "client received resp after oom");
+
+    # - test GET commands but don't read back, large backend values
+    # - test disabling the limiter
+    # extended testing:
+    # - create errors while holding the buffers?
+}
+
 # TODO:
+# check reqlimit/bwlimit counters
 # remove backends
 # do dead sockets close?
 # adding user stats