CLIENT REPLY command implemented: ON, OFF and SKIP modes.

Sometimes it can be useful for clients to completely disable replies
from the Redis server. For example when the client sends fire and forget
commands or performs a mass loading of data, or in caching contexts
where new data is streamed constantly. In such contexts to use server
time and bandwidth in order to send back replies to clients, which are
going to be ignored, is a shame.

Multiple mechanisms are possible to implement such a feature. For
example it could be a feature of MULTI/EXEC, or a command prefix
such as "NOREPLY SADD myset foo", or a different mechanism that allows
to switch on/off requests using the CLIENT command.

The MULTI/EXEC approach has the problem that transactions are not
strictly part of the no-reply semantics, and if we want to insert a lot
of data in a bulk way, creating a huge MULTI/EXEC transaction in the
server memory is bad.

The prefix is the best in this specific use case since it does not allow
desynchronizations, and is pretty clear semantically. However Redis
internals and client libraries are not prepared to handle this
currently.

So the implementation uses the CLIENT command, providing a new REPLY
subcommand with three options:

    CLIENT REPLY OFF disables the replies, and does not reply itself.
    CLIENT REPLY ON re-enables the replies, replying +OK.
    CLIENT REPLY SKIP only discards the reply of the next command, and
                      like OFF does not reply anything itself.

The reason to add the SKIP command is that it allows to have an easy
way to send conceptually "single" commands that don't need a reply
as the sum of two pipelined commands:

    CLIENT REPLY SKIP
    SET key value

Note that CLIENT REPLY ON replies with +OK so it should be used when
sending multiple commands that don't need a reply. However since it
replies with +OK the client can check that the connection is still
active and all the previous commands were received.

This is currently just into Redis "unstable" so the proposal can be
modified or abandoned based on users inputs.

src/networking.c

@@ -160,6 +160,9 @@ int prepareClientToWrite(client *c) {
      * handler since there is no socket at all. */
     if (c->flags & CLIENT_LUA) return C_OK;
 
+    /* CLIENT REPLY OFF / SKIP handling: don't send replies. */
+    if (c->flags & (CLIENT_REPLY_OFF|CLIENT_REPLY_SKIP)) return C_ERR;
+
     /* Masters don't receive replies, unless CLIENT_MASTER_FORCE_REPLY flag
      * is set. */
     if ((c->flags & CLIENT_MASTER) &&
@@ -945,10 +948,20 @@ void resetClient(client *c) {
     c->reqtype = 0;
     c->multibulklen = 0;
     c->bulklen = -1;
+
     /* We clear the ASKING flag as well if we are not inside a MULTI, and
      * if what we just executed is not the ASKING command itself. */
     if (!(c->flags & CLIENT_MULTI) && prevcmd != askingCommand)
-        c->flags &= (~CLIENT_ASKING);
+        c->flags &= ~CLIENT_ASKING;
+
+    /* Remove the CLIENT_REPLY_SKIP flag if any so that the reply
+     * to the next command will be sent, but set the flag if the command
+     * we just processed was "CLIENT REPLY SKIP". */
+    c->flags &= ~CLIENT_REPLY_SKIP;
+    if (c->flags & CLIENT_REPLY_SKIP_NEXT) {
+        c->flags |= CLIENT_REPLY_SKIP;
+        c->flags &= ~CLIENT_REPLY_SKIP_NEXT;
+    }
 }
 
 int processInlineBuffer(client *c) {
@@ -1396,6 +1409,20 @@ void clientCommand(client *c) {
         sds o = getAllClientsInfoString();
         addReplyBulkCBuffer(c,o,sdslen(o));
         sdsfree(o);
+    } else if (!strcasecmp(c->argv[1]->ptr,"reply") && c->argc == 3) {
+        /* CLIENT REPLY ON|OFF|SKIP */
+        if (!strcasecmp(c->argv[2]->ptr,"on")) {
+            c->flags &= ~(CLIENT_REPLY_SKIP|CLIENT_REPLY_OFF);
+            addReply(c,shared.ok);
+        } else if (!strcasecmp(c->argv[2]->ptr,"off")) {
+            c->flags |= CLIENT_REPLY_OFF;
+        } else if (!strcasecmp(c->argv[2]->ptr,"skip")) {
+            if (!(c->flags & CLIENT_REPLY_OFF))
+                c->flags |= CLIENT_REPLY_SKIP_NEXT;
+        } else {
+            addReply(c,shared.syntaxerr);
+            return;
+        }
     } else if (!strcasecmp(c->argv[1]->ptr,"kill")) {
         /* CLIENT KILL <ip:port>
          * CLIENT KILL <option> [value] ... <option> [value] */

src/server.h

@@ -245,6 +245,9 @@ typedef long long mstime_t; /* millisecond time type. */
 #define CLIENT_PREVENT_PROP (1<<19)  /* Don't propagate to AOF / Slaves. */
 #define CLIENT_PENDING_WRITE (1<<20) /* Client has output to send but a write
                                         handler is yet not installed. */
+#define CLIENT_REPLY_OFF (1<<21)   /* Don't send replies to client. */
+#define CLIENT_REPLY_SKIP_NEXT (1<<22)  /* Set CLIENT_REPLY_SKIP for next cmd */
+#define CLIENT_REPLY_SKIP (1<<23)  /* Don't send just this reply. */
 
 /* Client block type (btype field in client structure)
  * if CLIENT_BLOCKED flag is set. */