Changeset 722

Timestamp:

02/25/08 07:55:50 (6 months ago)

Author:

dsallings

Message:

Merged through r712

Conflicts:

server/memcached.h

Files:

• branches/binary/api/perl/ChangeLog (modified) (1 diff)
• branches/binary/api/perl/lib/Cache/Memcached.pm (modified) (9 diffs)
• branches/binary/server/ChangeLog (modified) (2 diffs)
• branches/binary/server/devtools/bench_noreply.pl (added)
• branches/binary/server/doc/protocol.txt (modified) (8 diffs)
• branches/binary/server/memcached.c (modified) (15 diffs)
• branches/binary/server/memcached.h (modified) (1 diff)
• branches/binary/server/t/noreply.t (added)

branches/binary/api/perl/ChangeLog

@@ +1 @@
+        * 'noreply' support from Tomash Brechko <tomash.brechko@gmail.com>
+
-        * various test updates from Ronald J Kimball <rkimball@pangeamedia.com>
-

branches/binary/api/perl/lib/Cache/Memcached.pm

@@ -336 +336 @@
-sub _write_and_read {
-    my Cache::Memcached $self = shift;
-
+, $noreply
-    my $res;
-    my ($ret, $offset) = (undef, 0);
@@ -376 +376 @@
-            }
-            if ($res == length($line)) { # all sent
-
+$noreply ? 2 : 
-            } else { # we only succeeded in sending some of it
-                substr($line, 0, $res, ''); # delete the part we sent
@@ -411 +411 @@
-    return 0 unless $sock;
-
+    my @params;
+    my $noreply = not defined wantarray;
+    push @params, "noreply" if $noreply;
+
-    $self->{'stats'}->{"delete"}++;
-    $key = ref $key ? $key->[1] : $key;
-    $time = $time ? " $time" : "";
-
-
+ @params
+, undef, $noreply
-
-    if ($self->{'stat_callback'}) {
@@ -447 +451 @@
-    return 0 unless $sock;
-
+    my @params;
+    my $noreply = not defined wantarray;
+    push @params, "noreply" if $noreply;
+
-    use bytes; # return bytes from length()
-
@@ -478 +486 @@
-
-    local $SIG{'PIPE'} = "IGNORE" unless $FLAG_NOSIGNAL;
-
-
-
+ @params
+
+, undef, $noreply
-
-    if ($self->{'debug'} && $line) {
@@ -515 +523 @@
-    $value = 1 unless defined $value;
-
-
-
+
+
+
+
+
+
-
-    if ($self->{'stat_callback'}) {
@@ -777 +789 @@
-sub flush_all {
-    my Cache::Memcached $self = shift;
+    my ($time) = @_;
+
+    $time = 0 unless defined $time;
-
-    my $success = 1;
-
+    my @params;
+    my $noreply = not defined wantarray;
+    push @params, "noreply" if $noreply;
+
-    my @hosts = @{$self->{'buckets'}};
+
+    # Distribute the delay among the servers.  For instance, if $time
+    # is 30 seconds, and we have 3 servers, they will get 30, 15, 0.
+    my $delay_step = 0;
+    if (@hosts > 1) {
+        $delay_step = $time / (@hosts - 1);
+    }
+
-    foreach my $host (@hosts) {
+        my $delay = int($time);
+        $time -= $delay_step;
+        my $line = "flush_all $delay @params\r\n";
-        my $sock = $self->sock_to_host($host);
-@res = $self->run_command($sock, "flush_all\r\n"
-@res
+$res = _write_and_read($self, $sock, $line, undef, $noreply
+$noreply or $res eq "OK\r\n"
-    }
-
@@ -791 +821 @@
-
-# returns array of lines, or () on failure.
+# FIXME: current implementation is broken.
-sub run_command {
-    my Cache::Memcached $self = shift;
@@ -798 +829 @@
-    my $line = $cmd;
-    while (my $res = _write_and_read($self, $sock, $line)) {
+        # FIXME: _write_and_read() won't handle undefined $line.
-        undef $line;
-        $ret .= $res;
+        # FIXME: end condition is not correct.
-        last if $ret =~ /(?:OK|END|ERROR)\r\n$/;
-    }

branches/binary/server/ChangeLog

@@ -1 @@
-
+
-
-       * IPv6 support, and IPv6 multi-interface support (brian@tangent.org)
@@ -9 +9 @@
-
-       * Use gettimeofday(2) instead of time(2).
-
-2008-02-10
-
-       * Make -k option work (Tomash Brechko)

branches/binary/server/doc/protocol.txt

@@ -127 +127 @@
-First, the client sends a command line which looks like this:
-
-
-
-
+
+
+
+
-
-  "set" means "store this data".  
@@ -175 +176 @@
-  when issuing "cas" updates.
-
+- "noreply" optional parameter instructs the server to not send the
+  reply.  NOTE: if the request line is malformed, the server can't
+  parse "noreply" option reliably.  In this case it may send the error
+  to the client, and not reading it on the client side will break
+  things.  Client should construct only valid requests.
+
-After this line, the client sends the data block:
-
@@ -246 +253 @@
-The command "delete" allows for explicit deletion of items:
-
-<time>
+[<time>] [noreply]
-
-- <key> is the key of the item the client wishes the server to delete
@@ -262 +269 @@
-  storage commands with this key will succeed).
-
+- "noreply" optional parameter instructs the server to not send the
+  reply.  See the note in Storage commands regarding malformed
+  requests.
+
-The response line to this command can be one of:
-
@@ -286 +297 @@
-The client sends the command line:
-
-
+ [noreply]
-
-or
-
-
+ [noreply]
-
-- <key> is the key of the item the client wishes to change
@@ -296 +307 @@
-- <value> is the amount by which the client wants to increase/decrease
-the item. It is a decimal representation of a 64-bit unsigned integer.
+
+- "noreply" optional parameter instructs the server to not send the
+  reply.  See the note in Storage commands regarding malformed
+  requests.
-
-The response will be one of:
@@ -400 +415 @@
-
-"flush_all" is a command with an optional numeric argument. It always
-
-
-
-
-
-
-
-
-
-
+
+
+
+
+
+
+
+
+
+
+
-
-The intent of flush_all with a delay, was that in a setting where you
@@ -432 +448 @@
-server.
-
-
-
-
+
+
+
+
-
-"quit" is a command with no arguments:

branches/binary/server/memcached.c

@@ -410 +410 @@
-    c->gen = 0;
-
+    c->noreply = false;
+
-    event_set(&c->event, sfd, event_flags, event_handler, (void *)c);
-    event_base_set(base, &c->event);
@@ -791 +793 @@
-
-    assert(c != NULL);
+
+    if (c->noreply) {
+        if (settings.verbose > 1)
+            fprintf(stderr, ">%d NOREPLY %s\n", c->sfd, str);
+        c->noreply = false;
+        conn_set_state(c, conn_read);
+        return;
+    }
-
-    if (settings.verbose > 1)
@@ -1481 +1491 @@
-#define KEY_MAX_LENGTH 250
-
-7
+8
-
-/*
@@ -1548 +1558 @@
-    } else {
-        out_string(c, "SERVER_ERROR out of memory");
+    }
+}
+
+static inline void set_noreply_maybe(conn *c, token_t *tokens, size_t ntokens)
+{
+    int noreply_index = ntokens - 2;
+
+    /*
+      NOTE: this function is not the first place where we are going to
+      send the reply.  We could send it instead from process_command()
+      if the request line has wrong number of tokens.  However parsing
+      malformed line for "noreply" option is not reliable anyway, so
+      it can't be helped.
+    */
+    if (tokens[noreply_index].value
+        && strcmp(tokens[noreply_index].value, "noreply") == 0) {
+        c->noreply = true;
-    }
-}
@@ -1930 +1957 @@
-    assert(c != NULL);
-
+    set_noreply_maybe(c, tokens, ntokens);
+
-    if (tokens[KEY_TOKEN].length > KEY_MAX_LENGTH) {
-        out_string(c, "CLIENT_ERROR bad command line format");
@@ -2000 +2029 @@
-
-    assert(c != NULL);
+
+    set_noreply_maybe(c, tokens, ntokens);
-
-    if(tokens[KEY_TOKEN].length > KEY_MAX_LENGTH) {
@@ -2099 +2130 @@
-    assert(c != NULL);
-
+    set_noreply_maybe(c, tokens, ntokens);
+
-    if (settings.managed) {
-        int bucket = c->bucket;
@@ -2120 +2153 @@
-    }
-
-4
+(c->noreply ? 5 : 4)
-        exptime = strtol(tokens[2].value, NULL, 10);
-
@@ -2183 +2216 @@
-    assert(c != NULL);
-
+    set_noreply_maybe(c, tokens, ntokens);
+
-    level = strtoul(tokens[1].value, NULL, 10);
-    settings.verbose = level > MAX_VERBOSITY_LEVEL ? MAX_VERBOSITY_LEVEL : level;
@@ -2220 +2255 @@
-        process_get_command(c, tokens, ntokens, false);
-
-ntokens == 6
+(ntokens == 6 || ntokens == 7)
-               ((strcmp(tokens[COMMAND_TOKEN].value, "add") == 0 && (comm = NREAD_ADD)) ||
-                (strcmp(tokens[COMMAND_TOKEN].value, "set") == 0 && (comm = NREAD_SET)) ||
@@ -2229 +2264 @@
-        process_update_command(c, tokens, ntokens, comm, false);
-
-ntokens == 7
+(ntokens == 7 || ntokens == 8)
-
-        process_update_command(c, tokens, ntokens, comm, true);
-
-ntokens == 4
+(ntokens == 4 || ntokens == 5)
-
-        process_arithmetic_command(c, tokens, ntokens, 1);
@@ -2241 +2276 @@
-        process_get_command(c, tokens, ntokens, true);
-
-ntokens == 4
+(ntokens == 4 || ntokens == 5)
-
-        process_arithmetic_command(c, tokens, ntokens, 0);
-
-4
+5
-
-        process_delete_command(c, tokens, ntokens);
@@ -2314 +2349 @@
-        process_stat(c, tokens, ntokens);
-
-3
+4
-        time_t exptime = 0;
-        set_current_time();
-
-
+
+
+
-            settings.oldest_live = current_time - 1;
-            item_flush_expired();
@@ -2383 +2420 @@
-        out_string(c, "CLIENT_ERROR Slab reassignment not supported");
-#endif
-ntokens == 3
+(ntokens == 3 || ntokens == 4)
-        process_verbosity_command(c, tokens, ntokens);
-    } else {
@@ -2516 +2553 @@
-        if (res == -1) {
-            if (errno == EAGAIN || errno == EWOULDBLOCK) break;
-
+
+
+
-        }
-    }

branches/binary/server/memcached.h

@@ -285 +285 @@
-                         a managed instance. -1 (_not_ 0) means invalid. */
-    int    gen;       /* generation requested for the bucket */
-
+
-    /* Binary protocol stuff */
-    /* This is where the binary header goes */