Skip to content

HTTPS clone URL

Subversion checkout URL

You can clone with HTTPS or Subversion.

Download ZIP
Browse files

Implement 'noreply' option for update commands. (Tomash Brechko <toma…

…sh.brechko@gmail.com>)

Commands add, set, replace, append, prepend, cas, delete, incr, decr,
flush_all, verbosity can take last optional parameter, 'noreply',
which instructs the server to not send the reply.

Add benchmark script for noreply parameter.


git-svn-id: http://code.sixapart.com/svn/memcached/trunk/server@708 b0b603af-a30f-0410-a34e-baf09ae79d0b
  • Loading branch information...
commit d9ece78091f1dc15332b16131258fb25d9ba3224 1 parent 57e3367
@kroki kroki authored
View
51 devtools/bench_noreply.pl
@@ -0,0 +1,51 @@
+#! /usr/bin/perl
+#
+use warnings;
+use strict;
+
+use IO::Socket::INET;
+
+use FindBin;
+
+@ARGV == 1 or @ARGV == 2
+ or die "Usage: $FindBin::Script HOST:PORT [COUNT]\n";
+
+# Note that it's better to run the test over the wire, because for
+# localhost the task may become CPU bound.
+my $addr = $ARGV[0];
+my $count = $ARGV[1] || 10_000;
+
+my $sock = IO::Socket::INET->new(PeerAddr => $addr,
+ Timeout => 3);
+die "$!\n" unless $sock;
+
+
+# By running 'noreply' test first we also ensure there are no reply
+# packets left in the network.
+foreach my $noreply (1, 0) {
+ use Time::HiRes qw(gettimeofday tv_interval);
+
+ print "'noreply' is ", $noreply ? "enabled" : "disabled", ":\n";
+ my $param = $noreply ? 'noreply' : '';
+ my $start = [gettimeofday];
+ foreach (1 .. $count) {
+ print $sock "add foo 0 0 1 $param\r\n1\r\n";
+ scalar<$sock> unless $noreply;
+ print $sock "set foo 0 0 1 $param\r\n1\r\n";
+ scalar<$sock> unless $noreply;
+ print $sock "replace foo 0 0 1 $param\r\n1\r\n";
+ scalar<$sock> unless $noreply;
+ print $sock "append foo 0 0 1 $param\r\n1\r\n";
+ scalar<$sock> unless $noreply;
+ print $sock "prepend foo 0 0 1 $param\r\n1\r\n";
+ scalar<$sock> unless $noreply;
+ print $sock "incr foo 1 $param\r\n";
+ scalar<$sock> unless $noreply;
+ print $sock "decr foo 1 $param\r\n";
+ scalar<$sock> unless $noreply;
+ print $sock "delete foo $param\r\n";
+ scalar<$sock> unless $noreply;
+ }
+ my $end = [gettimeofday];
+ printf("update commands: %.2f secs\n\n", tv_interval($start, $end));
+}
View
53 doc/protocol.txt
@@ -126,9 +126,10 @@ Storage commands
First, the client sends a command line which looks like this:
-<command name> <key> <flags> <exptime> <bytes> [<cas unqiue>]\r\n
+<command name> <key> <flags> <exptime> <bytes> [noreply]\r\n
+cas <key> <flags> <exptime> <bytes> <cas unqiue> [noreply]\r\n
-- <command name> is "set", "add", "replace", "append", "prepend", or "cas"
+- <command name> is "set", "add", "replace", "append" or "prepend"
"set" means "store this data".
@@ -174,6 +175,12 @@ First, the client sends a command line which looks like this:
Clients should use the value returned from the "gets" command
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:
<data block>\r\n
@@ -245,7 +252,7 @@ Deletion
The command "delete" allows for explicit deletion of items:
-delete <key> <time>\r\n
+delete <key> [<time>] [noreply]\r\n
- <key> is the key of the item the client wishes the server to delete
@@ -261,6 +268,10 @@ delete <key> <time>\r\n
(which means that the item will be deleted immediately and further
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:
- "DELETED\r\n" to indicate success
@@ -285,17 +296,21 @@ non-existent key exists with value 0; instead, they will fail.
The client sends the command line:
-incr <key> <value>\r\n
+incr <key> <value> [noreply]\r\n
or
-decr <key> <value>\r\n
+decr <key> <value> [noreply]\r\n
- <key> is the key of the item the client wishes to change
- <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:
- "NOT_FOUND\r\n" to indicate the item with this value was not found
@@ -399,16 +414,17 @@ Other commands
--------------
"flush_all" is a command with an optional numeric argument. It always
-succeeds, and the server sends "OK\r\n" in response. Its effect is to
-invalidate all existing items immediately (by default) or after the
-expiration specified. After invalidation none of the items will be returned
-in response to a retrieval command (unless it's stored again under the
-same key *after* flush_all has invalidated the items). flush_all doesn't
-actually free all the memory taken up by existing items; that will
-happen gradually as new items are stored. The most precise definition
-of what flush_all does is the following: it causes all items whose
-update time is earlier than the time at which flush_all was set to be
-executed to be ignored for retrieval purposes.
+succeeds, and the server sends "OK\r\n" in response (unless "noreply"
+is given as the last parameter). Its effect is to invalidate all
+existing items immediately (by default) or after the expiration
+specified. After invalidation none of the items will be returned in
+response to a retrieval command (unless it's stored again under the
+same key *after* flush_all has invalidated the items). flush_all
+doesn't actually free all the memory taken up by existing items; that
+will happen gradually as new items are stored. The most precise
+definition of what flush_all does is the following: it causes all
+items whose update time is earlier than the time at which flush_all
+was set to be executed to be ignored for retrieval purposes.
The intent of flush_all with a delay, was that in a setting where you
have a pool of memcached servers, and you need to flush all content,
@@ -431,9 +447,10 @@ In response, the server sends
"VERSION <version>\r\n", where <version> is the version string for the
server.
-"verbosity" is a command with a numeric argument. It always
-succeeds, and the server sends "OK\r\n" in response. Its effect is to
-set the verbosity level of the logging output.
+"verbosity" is a command with a numeric argument. It always succeeds,
+and the server sends "OK\r\n" in response (unless "noreply" is given
+as the last parameter). Its effect is to set the verbosity level of
+the logging output.
"quit" is a command with no arguments:
View
55 memcached.c
@@ -372,6 +372,8 @@ conn *conn_new(const int sfd, const int init_state, const int event_flags,
c->bucket = -1;
c->gen = 0;
+ c->noreply = false;
+
event_set(&c->event, sfd, event_flags, event_handler, (void *)c);
event_base_set(base, &c->event);
c->ev_flags = event_flags;
@@ -733,6 +735,12 @@ static void out_string(conn *c, const char *str) {
assert(c != NULL);
+ if (c->noreply) {
+ c->noreply = false;
+ conn_set_state(c, conn_read);
+ return;
+ }
+
if (settings.verbose > 1)
fprintf(stderr, ">%d %s\n", c->sfd, str);
@@ -898,7 +906,7 @@ typedef struct token_s {
#define KEY_TOKEN 1
#define KEY_MAX_LENGTH 250
-#define MAX_TOKENS 7
+#define MAX_TOKENS 8
/*
* Tokenize the command string by replacing whitespace with '\0' and update
@@ -968,6 +976,23 @@ static void write_and_free(conn *c, char *buf, int bytes) {
}
}
+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;
+ }
+}
+
inline static void process_stats_detail(conn *c, const char *command) {
assert(c != NULL);
@@ -1347,6 +1372,8 @@ static void process_update_command(conn *c, token_t *tokens, const size_t ntoken
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");
return;
@@ -1418,6 +1445,8 @@ static void process_arithmetic_command(conn *c, token_t *tokens, const size_t nt
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");
return;
@@ -1514,6 +1543,8 @@ static void process_delete_command(conn *c, token_t *tokens, const size_t ntoken
assert(c != NULL);
+ set_noreply_maybe(c, tokens, ntokens);
+
if (settings.managed) {
int bucket = c->bucket;
if (bucket == -1) {
@@ -1535,7 +1566,7 @@ static void process_delete_command(conn *c, token_t *tokens, const size_t ntoken
return;
}
- if(ntokens == 4) {
+ if(ntokens == (c->noreply ? 5 : 4)) {
exptime = strtol(tokens[2].value, NULL, 10);
if(errno == ERANGE) {
@@ -1598,6 +1629,8 @@ static void process_verbosity_command(conn *c, token_t *tokens, const size_t nto
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;
out_string(c, "OK");
@@ -1635,7 +1668,7 @@ static void process_command(conn *c, char *command) {
process_get_command(c, tokens, ntokens, false);
- } else if (ntokens == 6 &&
+ } else if ((ntokens == 6 || ntokens == 7) &&
((strcmp(tokens[COMMAND_TOKEN].value, "add") == 0 && (comm = NREAD_ADD)) ||
(strcmp(tokens[COMMAND_TOKEN].value, "set") == 0 && (comm = NREAD_SET)) ||
(strcmp(tokens[COMMAND_TOKEN].value, "replace") == 0 && (comm = NREAD_REPLACE)) ||
@@ -1644,11 +1677,11 @@ static void process_command(conn *c, char *command) {
process_update_command(c, tokens, ntokens, comm, false);
- } else if (ntokens == 7 && (strcmp(tokens[COMMAND_TOKEN].value, "cas") == 0 && (comm = NREAD_CAS))) {
+ } else if ((ntokens == 7 || ntokens == 8) && (strcmp(tokens[COMMAND_TOKEN].value, "cas") == 0 && (comm = NREAD_CAS))) {
process_update_command(c, tokens, ntokens, comm, true);
- } else if (ntokens == 4 && (strcmp(tokens[COMMAND_TOKEN].value, "incr") == 0)) {
+ } else if ((ntokens == 4 || ntokens == 5) && (strcmp(tokens[COMMAND_TOKEN].value, "incr") == 0)) {
process_arithmetic_command(c, tokens, ntokens, 1);
@@ -1656,11 +1689,11 @@ static void process_command(conn *c, char *command) {
process_get_command(c, tokens, ntokens, true);
- } else if (ntokens == 4 && (strcmp(tokens[COMMAND_TOKEN].value, "decr") == 0)) {
+ } else if ((ntokens == 4 || ntokens == 5) && (strcmp(tokens[COMMAND_TOKEN].value, "decr") == 0)) {
process_arithmetic_command(c, tokens, ntokens, 0);
- } else if (ntokens >= 3 && ntokens <= 4 && (strcmp(tokens[COMMAND_TOKEN].value, "delete") == 0)) {
+ } else if (ntokens >= 3 && ntokens <= 5 && (strcmp(tokens[COMMAND_TOKEN].value, "delete") == 0)) {
process_delete_command(c, tokens, ntokens);
@@ -1729,11 +1762,13 @@ static void process_command(conn *c, char *command) {
process_stat(c, tokens, ntokens);
- } else if (ntokens >= 2 && ntokens <= 3 && (strcmp(tokens[COMMAND_TOKEN].value, "flush_all") == 0)) {
+ } else if (ntokens >= 2 && ntokens <= 4 && (strcmp(tokens[COMMAND_TOKEN].value, "flush_all") == 0)) {
time_t exptime = 0;
set_current_time();
- if(ntokens == 2) {
+ set_noreply_maybe(c, tokens, ntokens);
+
+ if(ntokens == (c->noreply ? 3 : 2)) {
settings.oldest_live = current_time - 1;
item_flush_expired();
out_string(c, "OK");
@@ -1798,7 +1833,7 @@ static void process_command(conn *c, char *command) {
#else
out_string(c, "CLIENT_ERROR Slab reassignment not supported");
#endif
- } else if (ntokens == 3 && (strcmp(tokens[COMMAND_TOKEN].value, "verbosity") == 0)) {
+ } else if ((ntokens == 3 || ntokens == 4) && (strcmp(tokens[COMMAND_TOKEN].value, "verbosity") == 0)) {
process_verbosity_command(c, tokens, ntokens);
} else {
out_string(c, "ERROR");
View
1  memcached.h
@@ -220,6 +220,7 @@ struct conn {
int bucket; /* bucket number for the next command, if running as
a managed instance. -1 (_not_ 0) means invalid. */
int gen; /* generation requested for the bucket */
+ bool noreply; /* True if the reply should not be sent. */
conn *next; /* Used for generating a list of conn structures */
};
View
53 t/noreply.t
@@ -0,0 +1,53 @@
+#!/usr/bin/perl
+
+use strict;
+use Test::More tests => 10;
+use FindBin qw($Bin);
+use lib "$Bin/lib";
+use MemcachedTest;
+
+
+my $server = new_memcached();
+my $sock = $server->sock;
+
+
+# Test that commands can take 'noreply' parameter.
+print $sock "flush_all noreply\r\n";
+print $sock "flush_all 0 noreply\r\n";
+
+print $sock "verbosity 0 noreply\r\n";
+
+print $sock "add noreply:foo 0 0 1 noreply\r\n1\r\n";
+mem_get_is($sock, "noreply:foo", "1");
+
+print $sock "set noreply:foo 0 0 1 noreply\r\n2\r\n";
+mem_get_is($sock, "noreply:foo", "2");
+
+print $sock "replace noreply:foo 0 0 1 noreply\r\n3\r\n";
+mem_get_is($sock, "noreply:foo", "3");
+
+print $sock "append noreply:foo 0 0 1 noreply\r\n4\r\n";
+mem_get_is($sock, "noreply:foo", "34");
+
+print $sock "prepend noreply:foo 0 0 1 noreply\r\n5\r\n";
+my @result = mem_gets($sock, "noreply:foo");
+ok($result[1] eq "534");
+
+print $sock "cas noreply:foo 0 0 1 $result[0] noreply\r\n6\r\n";
+mem_get_is($sock, "noreply:foo", "6");
+
+print $sock "incr noreply:foo 3 noreply\r\n";
+mem_get_is($sock, "noreply:foo", "9");
+
+print $sock "decr noreply:foo 2 noreply\r\n";
+mem_get_is($sock, "noreply:foo", "7");
+
+print $sock "delete noreply:foo noreply\r\n";
+mem_get_is($sock, "noreply:foo");
+
+# Test that delete accepts both <time> and 'noreply'.
+print $sock "add noreply:foo 0 0 1 noreply\r\n1\r\n";
+print $sock "delete noreply:foo 10 noreply\r\n";
+print $sock "add noreply:foo 0 0 1 noreply\r\n1\r\n";
+# undef result means we couldn't add an entry because the key is locked.
+mem_get_is($sock, "noreply:foo");
Please sign in to comment.
Something went wrong with that request. Please try again.