Commit 71c37605 authored by antirez's avatar antirez
Browse files

Transactions: move the QUEUED logic inside call(). (Work in progress)

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.
parent 3e78344d
...@@ -2410,7 +2410,7 @@ void call(client *c, int flags) { ...@@ -2410,7 +2410,7 @@ void call(client *c, int flags) {
int client_old_flags = c->flags; int client_old_flags = c->flags;
struct redisCommand *real_cmd = c->cmd; 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. */ * not generated from reading an AOF. */
if (listLength(server.monitors) && if (listLength(server.monitors) &&
!server.loading && !server.loading &&
...@@ -2419,6 +2419,18 @@ void call(client *c, int flags) { ...@@ -2419,6 +2419,18 @@ void call(client *c, int flags) {
replicationFeedMonitors(c,server.monitors,c->db->id,c->argv,c->argc); 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. */
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);
return;
}
/* Initialization: clear the flags that must be set by the command on /* Initialization: clear the flags that must be set by the command on
* demand, and initialize the array for additional commands propagation. */ * demand, and initialize the array for additional commands propagation. */
c->flags &= ~(CLIENT_FORCE_AOF|CLIENT_FORCE_REPL|CLIENT_PREVENT_PROP); c->flags &= ~(CLIENT_FORCE_AOF|CLIENT_FORCE_REPL|CLIENT_PREVENT_PROP);
...@@ -2714,18 +2726,9 @@ int processCommand(client *c) { ...@@ -2714,18 +2726,9 @@ int processCommand(client *c) {
} }
/* Exec the command */ /* Exec the command */
if (c->flags & CLIENT_MULTI && call(c,CMD_CALL_FULL);
c->cmd->proc != execCommand && c->cmd->proc != discardCommand && c->woff = server.master_repl_offset;
c->cmd->proc != multiCommand && c->cmd->proc != watchCommand) if (listLength(server.ready_keys)) handleClientsBlockedOnKeys();
{
queueMultiCommand(c);
addReply(c,shared.queued);
} else {
call(c,CMD_CALL_FULL);
c->woff = server.master_repl_offset;
if (listLength(server.ready_keys))
handleClientsBlockedOnKeys();
}
return C_OK; return C_OK;
} }
......
Markdown is supported
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment