Redis Transactions

Redis Transactions

A client usually runs a sequence of commands to apply a group of related changes to a data object. However, another client may use similar commands during that window to modify the same object, which can lead to corruption or inconsistency. To solve this, Redis introduces transactions: a transaction groups multiple commands from a client into a single unit. Commands inside a transaction are guaranteed to execute in order and are not interrupted by commands from other clients.

Why is Redis’s transaction implementation much simpler than PostgreSQL’s? Because Redis’s critical processing is single-threaded—only one client’s commands run at a time, and the next client’s commands can run only after the current client’s commands finish. PostgreSQL, by contrast, is a multi-process architecture where many processes run concurrently, making its transaction implementation far more complex. Redis implements only a limited form of transactions: it guarantees that commands run in order and are not disturbed by other clients, but a command that fails inside a transaction cannot be rolled back (whereas PostgreSQL rolls back the whole transaction on failure); execution simply continues to the next command. Therefore the client must handle transaction failures appropriately.

Redis Transaction Implementation

Redis transactions involve five commands: MULTI, EXEC, DISCARD, WATCH, UNWATCH.

Command	Description	Return value
MULTI	Start a transaction	OK
EXEC	Execute the transaction	The return values of all commands in the transaction
DISCARD	Abort the transaction; if WATCH is monitoring some key, cancel all watches	OK
WATCH	Monitor one or more keys; if a watched key is modified by another command before the transaction executes, the transaction is aborted	OK
UNWATCH	Cancel the watch on all keys	OK

MULTI starts a transaction, EXEC executes it, DISCARD aborts it, and WATCH monitors one or more keys—if those keys are modified by another client before the transaction runs, the transaction will not execute.

You can test the transaction feature with the following commands:

127.0.0.1:6379> set foo 1
OK
127.0.0.1:6379> set bar 1
OK
127.0.0.1:6379> multi      # Enter a Redis transaction with MULTI. This command always replies OK
OK
127.0.0.1:6379(TX)> incr foo    # The client may issue several commands. Redis does not run them immediately; it queues them on the server
127.0.0.1:6379(TX)> incr bar
QUEUED
127.0.0.1:6379(TX)> exec   # Once EXEC is called, all queued commands run as a transaction
1) (integer) 2
2) (integer) 2

Calling DISCARD flushes the transaction queue and leaves the transaction:

127.0.0.1:6379> multi
OK
127.0.0.1:6379(TX)> incr foo
QUEUED
127.0.0.1:6379(TX)> incr bar
QUEUED
127.0.0.1:6379(TX)> discard   # DISCARD flushes the transaction queue and leaves the transaction
OK

If an error occurs during execution, Redis continues running the remaining queued commands:

127.0.0.1:6379> multi
OK
127.0.0.1:6379(TX)> set k1 v1
QUEUED
127.0.0.1:6379(TX)> incr k1   # incr only works on numbers; running it on a string errors
QUEUED
127.0.0.1:6379(TX)> set k2 1  # A failed command does not affect later commands
QUEUED
127.0.0.1:6379(TX)> get k2
QUEUED
127.0.0.1:6379(TX)> exec
1) OK
2) (error) ERR value is not an integer or out of range
3) OK
4) "1"

The Redis transaction execution flow is as follows:

1. The client sends MULTI. The server marks the client with the CLIENT_MULTI flag (transaction mode) and replies OK.
2. The client sends commands. The server enqueues each via queueMultiCommand(c) into c->mstate and replies QUEUED.
3. The client sends EXEC. The server runs the queued commands and returns their results.

Source Analysis

After a connection is established, when a client sends a command (data becomes readable), readQueryFromClient is called: it first reads the command string, then parses it according to the RESP protocol, and finally invokes the matching command handler.

readQueryFromClient(connection *conn)
--> connRead(c->conn, c->querybuf+qblen, readlen)  // read from socket into the buffer
    --> conn->type->read(conn, buf, buf_len);
--> processInputBuffer(c);    // parse the redis protocol, store the command in the client's argv array
    --> processInlineBuffer(c)   // handle inline commands and create argument objects
        --> sdsfreesplitres(argv,argc);
    --> processMultibulkBuffer(c) // convert protocol bytes in c->querybuf into argument objects in c->argv
    --> processCommandAndResetClient(c)        // execute the command and return the result
        --> processCommand(c)
            --> call(c,CMD_CALL_FULL);
                --> c->cmd->proc(c);  // run the specific command

Because a transaction contains multiple commands, its flow differs from a normal command. A data structure is needed to hold the queued commands.

typedef struct multiCmd {
    robj **argv;       // arguments
    int argc;          // argument count
    struct redisCommand *cmd;      // command pointer
} multiCmd;  // holds a single queued command

Transaction state: each client needs to hold all of its transaction state, so a queue holds the commands—the commands array stores all queued commands and count records how many.

typedef struct multiState {
    // transaction queue, FIFO order
    multiCmd *commands;     /* Array of MULTI commands */
    // number of queued commands
    int count;              /* Total number of MULTI commands */
    int minreplicas;        /* MINREPLICAS for synchronous replication */
    time_t minreplicas_timeout; /* MINREPLICAS timeout as unixtime. */
} multiState;

client is a very important struct; it holds everything about the client, including the currently selected database, the input/output buffers, the command queue, and the transaction state.

typedef struct client {
    redisDb *db;            // currently selected database

    struct redisCommand *cmd, *lastcmd;      // the command executed by the client

    int reqtype;      // request type: inline command or multibulk
    // client flags
    uint64_t flags;         /* Client flags: CLIENT_* macros. */
    // transaction state
    multiState mstate;      /* MULTI/EXEC state */

    // watched keys
    list *watched_keys;     /* Keys WATCHED for MULTI/EXEC CAS */
    // ...
} client;

With the data structures covered, let’s look at processCommand. After receiving a command, transaction commands are enqueued and QUEUED is returned; the transaction-control commands (EXEC, DISCARD, MULTI, WATCH) are not enqueued.

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 &&
        c->cmd->proc != resetCommand)
    {
        // not EXEC/DISCARD/MULTI/WATCH/RESET: enqueue into the transaction
        queueMultiCommand(c);
        addReply(c,shared.queued);  // reply QUEUED
    } else {
        call(c,CMD_CALL_FULL);
        c->woff = server.master_repl_offset;
        if (listLength(server.ready_keys))
            handleClientsBlockedOnKeys();
    }
}

queueMultiCommand appends a command to the transaction queue, i.e. client->mstate.commands.

void queueMultiCommand(client *c) {
    multiCmd *mc;
    int j;

    /* No sense to waste memory if the transaction is already aborted.
     * this is useful in case client sends these in a pipeline, or doesn't
     * bother to read previous responses and didn't notice the multi was already
     * aborted. */
    if (c->flags & CLIENT_DIRTY_EXEC)
        return;

    c->mstate.commands = zrealloc(c->mstate.commands,
            sizeof(multiCmd)*(c->mstate.count+1)); // allocate room for one more multiCmd
    mc = c->mstate.commands+c->mstate.count;
    mc->cmd = c->cmd;
    mc->argc = c->argc;
    mc->argv = zmalloc(sizeof(robj*)*c->argc);
    memcpy(mc->argv,c->argv,sizeof(robj*)*c->argc);
    for (j = 0; j < c->argc; j++)
        incrRefCount(mc->argv[j]);
    c->mstate.count++;   // increment command count
    c->mstate.cmd_flags |= c->cmd->flags;
    c->mstate.cmd_inv_flags |= ~c->cmd->flags;
}

MULTI opens a transaction. In multiCommand, the client is flagged CLIENT_MULTI (transaction mode) and OK is returned.

void multiCommand(client *c) {
    if (c->flags & CLIENT_MULTI) {
        addReplyError(c,"MULTI calls can not be nested");
        return;
    }
    c->flags |= CLIENT_MULTI;  // set the transaction flag

    addReply(c,shared.ok);   // reply OK
}

Executing a transaction is the EXEC command. In execCommand: if the client is in a transaction, the queued commands run in order and their results are returned; otherwise an error is returned. Note that blocking commands are forbidden inside a transaction.

void execCommand(client *c) {
    int j;
    robj **orig_argv;
    int orig_argc;
    struct redisCommand *orig_cmd;
    int was_master = server.masterhost == NULL;

    // ensure the client is in a transaction
    if (!(c->flags & CLIENT_MULTI)) {
        addReplyError(c,"EXEC without MULTI");
        return;
    }

    /* EXEC with expired watched key is disallowed */
    if (isWatchedKeyExpired(c)) { // check whether a watched key has expired
        c->flags |= (CLIENT_DIRTY_CAS);
    }

    /* Check if we need to abort the EXEC because:
     * 1) Some WATCHed key was touched.
     * 2) There was a previous error while queueing commands.
     * A failed EXEC in the first case returns a multi bulk nil object
     * (technically it is not an error but a special behavior), while
     * in the second an EXECABORT error is returned. */
    if (c->flags & (CLIENT_DIRTY_CAS | CLIENT_DIRTY_EXEC)) {
        if (c->flags & CLIENT_DIRTY_EXEC) {
            addReplyErrorObject(c, shared.execaborterr); // return the error
        } else {
            addReply(c, shared.nullarray[c->resp]); // return a null array
        }

        discardTransaction(c);  // abort the transaction
        return;
    }

    uint64_t old_flags = c->flags;

    /* we do not want to allow blocking commands inside multi */
    c->flags |= CLIENT_DENY_BLOCKING; // prevent queued commands from blocking the server (e.g. BLPOP)

    /* Exec all the queued commands */
    unwatchAllKeys(c); /* cancel the watch on all keys */

    server.in_exec = 1;

    orig_argv = c->argv;
    orig_argc = c->argc;
    orig_cmd = c->cmd;
    addReplyArrayLen(c,c->mstate.count);
    for (j = 0; j < c->mstate.count; j++) {   // iterate the transaction queue
        c->argc = c->mstate.commands[j].argc;
        c->argv = c->mstate.commands[j].argv;
        c->cmd = c->mstate.commands[j].cmd;

        /* ACL permissions are also checked at the time of execution in case
         * they were changed after the commands were queued. */
        int acl_errpos;
        int acl_retval = ACLCheckAllPerm(c,&acl_errpos); // check ACL permissions
        if (acl_retval != ACL_OK) {
            char *reason;
            switch (acl_retval) {
            case ACL_DENIED_CMD:
                reason = "no permission to execute the command or subcommand";
                break;
            case ACL_DENIED_KEY:
                reason = "no permission to touch the specified keys";
                break;
            case ACL_DENIED_CHANNEL:
                reason = "no permission to access one of the channels used "
                         "as arguments";
                break;
            default:
                reason = "no permission";
                break;
            }
            addACLLogEntry(c,acl_retval,acl_errpos,NULL);
            addReplyErrorFormat(c,
                "-NOPERM ACLs rules changed between the moment the "
                "transaction was accumulated and the EXEC call. "
                "This command is no longer allowed for the "
                "following reason: %s", reason);
        } else {
            call(c,server.loading ? CMD_CALL_NONE : CMD_CALL_FULL); // run the queued command
            serverAssert((c->flags & CLIENT_BLOCKED) == 0);
        }

        /* Commands may alter argc/argv, restore mstate. */
        c->mstate.commands[j].argc = c->argc;
        c->mstate.commands[j].argv = c->argv;
        c->mstate.commands[j].cmd = c->cmd;
    }

    // restore old DENY_BLOCKING value
    if (!(old_flags & CLIENT_DENY_BLOCKING))
        c->flags &= ~CLIENT_DENY_BLOCKING; // restore the allow-blocking flag

    c->argv = orig_argv;
    c->argc = orig_argc;
    c->cmd = orig_cmd;
    discardTransaction(c);

    /* Make sure the EXEC command will be propagated as well if MULTI
     * was already propagated. */
    if (server.propagate_in_transaction) {
        int is_master = server.masterhost == NULL;
        server.dirty++;
        /* If inside the MULTI/EXEC block this instance was suddenly
         * switched from master to slave (using the SLAVEOF command), the
         * initial MULTI was propagated into the replication backlog, but the
         * rest was not. We need to make sure to at least terminate the
         * backlog with the final EXEC. */
        if (server.repl_backlog && was_master && !is_master) {
            char *execcmd = "*1\r\n$4\r\nEXEC\r\n";
            feedReplicationBacklog(execcmd,strlen(execcmd));
        }
        afterPropagateExec();
    }

    server.in_exec = 0;
}

For DISCARD, discardTransaction clears the transaction state and OK is returned.

void discardCommand(client *c) {
    if (!(c->flags & CLIENT_MULTI)) {
        addReplyError(c,"DISCARD without MULTI");
        return;
    }
    discardTransaction(c);
    addReply(c,shared.ok);
}

void discardTransaction(client *c) {
    freeClientMultiState(c);
    initClientMultiState(c);
    c->flags &= ~(CLIENT_MULTI|CLIENT_DIRTY_CAS|CLIENT_DIRTY_EXEC);
    unwatchAllKeys(c);  // cancel all watch keys
}

The WATCH Mechanism

Redis transactions support optimistic locking via WATCH. Each client keeps a watched_keys list recording which keys it is watching (the keys it cares about).

typedef struct client {
    list *watched_keys;     /* Keys WATCHED for MULTI/EXEC CAS */
    // ...
} client;

On the other hand, when a watched key is modified, Redis must know which clients to notify. This is done with the watched_keys dictionary on the database: the key is the watched key, and the value is the list of clients watching it.

typedef struct redisDb {
    dict *watched_keys;    /* Keys WATCHED for MULTI/EXEC CAS */
    // ...
} redisDb;

When a client runs WATCH, the key is added to the watched_keys list via watchCommand -> watchForKey.

void watchCommand(client *c) {
    int j;

    if (c->flags & CLIENT_MULTI) {
        addReplyError(c,"WATCH inside MULTI is not allowed");
        return;
    }
    for (j = 1; j < c->argc; j++)
        watchForKey(c,c->argv[j]);
    addReply(c,shared.ok);
}

/* Watch for the specified key */
void watchForKey(client *c, robj *key) {
    list *clients = NULL;
    listIter li;
    listNode *ln;
    watchedKey *wk;

    // already watched: return
    listRewind(c->watched_keys,&li);
    while((ln = listNext(&li))) {
        wk = listNodeValue(ln);
        if (wk->db == c->db && equalStringObjects(key,wk->key))
            return; /* Key already watched */
    }

    // not yet watched:
    // 1. map the key to this client in c->db->watched_keys
    // 2. add it to the client's watched_keys list
    clients = dictFetchValue(c->db->watched_keys,key);
    if (!clients) {
        clients = listCreate();
        dictAdd(c->db->watched_keys,key,clients); // key: watched key, value: list of watching clients
        incrRefCount(key);
    }
    listAddNodeTail(clients,c);
    /* Add the new key to the list of keys watched by this client */
    wk = zmalloc(sizeof(*wk));
    wk->key = key;
    wk->db = c->db;
    incrRefCount(key);
    listAddNodeTail(c->watched_keys,wk);
}

When a key is modified, signalModifiedKey is called, which in turn calls touchWatchedKey. It marks every client watching that key with the CLIENT_DIRTY_CAS flag, so EXEC detects the conflict and aborts the transaction.

void touchWatchedKey(redisDb *db, robj *key) {
    list *clients;
    listIter li;
    listNode *ln;

    if (dictSize(db->watched_keys) == 0) return;
    clients = dictFetchValue(db->watched_keys, key);
    if (!clients) return;

    /* Mark all the clients watching this key as CLIENT_DIRTY_CAS */
    listRewind(clients,&li);
    while((ln = listNext(&li))) {
        client *c = listNodeValue(ln);

        c->flags |= CLIENT_DIRTY_CAS;
    }
}

Looking back at execCommand, when the CLIENT_DIRTY_CAS flag is detected, the transaction is aborted (see the check inside execCommand above).

That concludes the analysis of Redis’s transaction fundamentals.

Summary

Redis’s transaction implementation is relatively simple yet effective. It does not support rollback the way PostgreSQL does, so you must handle transaction failures on the client side. Supporting rollback in Redis would be much more complex: it would require undo logs, which must be persisted and would hurt performance, and it would greatly complicate Redis’s implementation. Moreover, does Redis really need that feature?

References: Redis Transactions Redis Source Analysis