Transactions: Use CMD_CLAL_NOQUEUE now that call() handles +QUEUED.

As a side effect this makes things like #5201 fixed automatically.
The problem with handling it in ProcessQueryBuffer() or similar is that
a client that is handled just via the call() interface will misbehave,
and call()-ing MULTI and later some command will see just the other
command executed like if the client was not in a transaction.

For now this has worked well enough but probably it makes more sense if
we can handle that case on call().

WARNING: this comment is not enough and will introduce bugs. It's a WIP.

src/multi.c

@@ -158,7 +158,7 @@ void execCommand(client *c) {
             must_propagate = 1;
         }
 
-        call(c,CMD_CALL_FULL);
+        call(c,CMD_CALL_FULL|CMD_CALL_NOQUEUE);
 
         /* Commands may alter argc/argv, restore mstate. */
         c->mstate.commands[j].argc = c->argc;

src/scripting.c

@@ -560,7 +560,7 @@ int luaRedisGenericCommand(lua_State *lua, int raise_error) {
     }
 
     /* Run the command */
-    int call_flags = CMD_CALL_SLOWLOG | CMD_CALL_STATS;
+    int call_flags = CMD_CALL_SLOWLOG | CMD_CALL_STATS | CMD_CALL_NOQUEUE;
     if (server.lua_replicate_commands) {
         /* Set flags according to redis.set_repl() settings. */
         if (server.lua_repl & PROPAGATE_AOF)

src/server.c

@@ -2410,7 +2410,7 @@ void call(client *c, int flags) {
     int client_old_flags = c->flags;
     struct redisCommand *real_cmd = c->cmd;
 
-    /* Sent the command to clients in MONITOR mode, only if the commands are
+    /* Send the command to clients in MONITOR mode, only if the commands are
      * not generated from reading an AOF. */
     if (listLength(server.monitors) &&
         !server.loading &&
@@ -2419,6 +2419,25 @@ void call(client *c, int flags) {
         replicationFeedMonitors(c,server.monitors,c->db->id,c->argv,c->argc);
     }
 
+    /* If the client is in the context of a transaction, reply with
+     * +QUEUED and just accumulate the command in the client transaction
+     * commands vector.
+     *
+     * Note that CALL_NOQUEUE can override this check and execute the command
+     * straight away even if the client is in "MULTI" state. This is useful
+     * every time we want to actually execute commands even if the client is
+     * in MULTI state, like in scripting.c or in the implementation of EXEC
+     * itself. */
+    if (!(flags & CMD_CALL_NOQUEUE) &&
+        c->flags & CLIENT_MULTI &&
+        c->cmd->proc != execCommand && c->cmd->proc != discardCommand &&
+        c->cmd->proc != multiCommand && c->cmd->proc != watchCommand)
+    {
+        queueMultiCommand(c);
+        addReply(c,shared.queued);
+        return;
+    }
+
     /* Initialization: clear the flags that must be set by the command on
      * demand, and initialize the array for additional commands propagation. */
     c->flags &= ~(CLIENT_FORCE_AOF|CLIENT_FORCE_REPL|CLIENT_PREVENT_PROP);
@@ -2714,18 +2733,9 @@ int processCommand(client *c) {
     }
 
     /* Exec the command */
-    if (c->flags & CLIENT_MULTI &&
-        c->cmd->proc != execCommand && c->cmd->proc != discardCommand &&
-        c->cmd->proc != multiCommand && c->cmd->proc != watchCommand)
-    {
-        queueMultiCommand(c);
-        addReply(c,shared.queued);
-    } else {
-        call(c,CMD_CALL_FULL);
-        c->woff = server.master_repl_offset;
-        if (listLength(server.ready_keys))
-            handleClientsBlockedOnKeys();
-    }
+    call(c,CMD_CALL_FULL);
+    c->woff = server.master_repl_offset;
+    if (listLength(server.ready_keys)) handleClientsBlockedOnKeys();
     return C_OK;
 }
 

src/server.h

@@ -416,6 +416,9 @@ typedef long long mstime_t; /* millisecond time type. */
 #define CMD_CALL_PROPAGATE_REPL (1<<3)
 #define CMD_CALL_PROPAGATE (CMD_CALL_PROPAGATE_AOF|CMD_CALL_PROPAGATE_REPL)
 #define CMD_CALL_FULL (CMD_CALL_SLOWLOG | CMD_CALL_STATS | CMD_CALL_PROPAGATE)
+/* --- call() special semantics that can be requested by the caller
+ * --- but are not included in CALL_FULL. */
+#define CMD_CALL_NOQUEUE (1<<4)
 
 /* Command propagation flags, see propagate() function */
 #define PROPAGATE_NONE 0