mirror of
https://codeberg.org/redict/redict.git
synced 2025-01-24 00:59:02 -05:00
3d1391272a
After the transcation starts with a MULIT, the previous behavior was to return an error on problems such as maxmemory limit reached. But still to execute the transaction with the subset of queued commands on EXEC. While it is true that the client was able to check for errors distinguish QUEUED by an error reply, MULTI/EXEC in most client implementations uses pipelining for speed, so all the commands and EXEC are sent without caring about replies. With this change: 1) EXEC fails if at least one command was not queued because of an error. The EXECABORT error is used. 2) A generic error is always reported on EXEC. 3) The client DISCARDs the MULTI state after a failed EXEC, otherwise pipelining multiple transactions would be basically impossible: After a failed EXEC the next transaction would be simply queued as the tail of the previous transaction.
327 lines
11 KiB
C
327 lines
11 KiB
C
/*
|
|
* Copyright (c) 2009-2012, Salvatore Sanfilippo <antirez at gmail dot com>
|
|
* All rights reserved.
|
|
*
|
|
* Redistribution and use in source and binary forms, with or without
|
|
* modification, are permitted provided that the following conditions are met:
|
|
*
|
|
* * Redistributions of source code must retain the above copyright notice,
|
|
* this list of conditions and the following disclaimer.
|
|
* * Redistributions in binary form must reproduce the above copyright
|
|
* notice, this list of conditions and the following disclaimer in the
|
|
* documentation and/or other materials provided with the distribution.
|
|
* * Neither the name of Redis nor the names of its contributors may be used
|
|
* to endorse or promote products derived from this software without
|
|
* specific prior written permission.
|
|
*
|
|
* THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
|
|
* AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
|
|
* IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
|
|
* ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE
|
|
* LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
|
|
* CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
|
|
* SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
|
|
* INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
|
|
* CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
|
|
* ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
|
|
* POSSIBILITY OF SUCH DAMAGE.
|
|
*/
|
|
|
|
#include "redis.h"
|
|
|
|
/* ================================ MULTI/EXEC ============================== */
|
|
|
|
/* Client state initialization for MULTI/EXEC */
|
|
void initClientMultiState(redisClient *c) {
|
|
c->mstate.commands = NULL;
|
|
c->mstate.count = 0;
|
|
}
|
|
|
|
/* Release all the resources associated with MULTI/EXEC state */
|
|
void freeClientMultiState(redisClient *c) {
|
|
int j;
|
|
|
|
for (j = 0; j < c->mstate.count; j++) {
|
|
int i;
|
|
multiCmd *mc = c->mstate.commands+j;
|
|
|
|
for (i = 0; i < mc->argc; i++)
|
|
decrRefCount(mc->argv[i]);
|
|
zfree(mc->argv);
|
|
}
|
|
zfree(c->mstate.commands);
|
|
}
|
|
|
|
/* Add a new command into the MULTI commands queue */
|
|
void queueMultiCommand(redisClient *c) {
|
|
multiCmd *mc;
|
|
int j;
|
|
|
|
c->mstate.commands = zrealloc(c->mstate.commands,
|
|
sizeof(multiCmd)*(c->mstate.count+1));
|
|
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++;
|
|
}
|
|
|
|
void discardTransaction(redisClient *c) {
|
|
freeClientMultiState(c);
|
|
initClientMultiState(c);
|
|
c->flags &= ~(REDIS_MULTI|REDIS_DIRTY_CAS|REDIS_DIRTY_EXEC);;
|
|
unwatchAllKeys(c);
|
|
}
|
|
|
|
/* Flag the transacation as DIRTY_EXEC so that EXEC will fail.
|
|
* Should be called every time there is an error while queueing a command. */
|
|
void flagTransaction(redisClient *c) {
|
|
if (c->flags & REDIS_MULTI)
|
|
c->flags |= REDIS_DIRTY_EXEC;
|
|
}
|
|
|
|
void multiCommand(redisClient *c) {
|
|
if (c->flags & REDIS_MULTI) {
|
|
addReplyError(c,"MULTI calls can not be nested");
|
|
return;
|
|
}
|
|
c->flags |= REDIS_MULTI;
|
|
addReply(c,shared.ok);
|
|
}
|
|
|
|
void discardCommand(redisClient *c) {
|
|
if (!(c->flags & REDIS_MULTI)) {
|
|
addReplyError(c,"DISCARD without MULTI");
|
|
return;
|
|
}
|
|
discardTransaction(c);
|
|
addReply(c,shared.ok);
|
|
}
|
|
|
|
/* Send a MULTI command to all the slaves and AOF file. Check the execCommand
|
|
* implememntation for more information. */
|
|
void execCommandReplicateMulti(redisClient *c) {
|
|
robj *multistring = createStringObject("MULTI",5);
|
|
|
|
if (server.aof_state != REDIS_AOF_OFF)
|
|
feedAppendOnlyFile(server.multiCommand,c->db->id,&multistring,1);
|
|
if (listLength(server.slaves))
|
|
replicationFeedSlaves(server.slaves,c->db->id,&multistring,1);
|
|
decrRefCount(multistring);
|
|
}
|
|
|
|
void execCommand(redisClient *c) {
|
|
int j;
|
|
robj **orig_argv;
|
|
int orig_argc;
|
|
struct redisCommand *orig_cmd;
|
|
|
|
if (!(c->flags & REDIS_MULTI)) {
|
|
addReplyError(c,"EXEC without MULTI");
|
|
return;
|
|
}
|
|
|
|
/* 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 & (REDIS_DIRTY_CAS|REDIS_DIRTY_EXEC)) {
|
|
addReply(c, c->flags & REDIS_DIRTY_EXEC ? shared.execaborterr :
|
|
shared.nullmultibulk);
|
|
freeClientMultiState(c);
|
|
initClientMultiState(c);
|
|
c->flags &= ~(REDIS_MULTI|REDIS_DIRTY_CAS|REDIS_DIRTY_EXEC);
|
|
unwatchAllKeys(c);
|
|
goto handle_monitor;
|
|
}
|
|
|
|
/* Replicate a MULTI request now that we are sure the block is executed.
|
|
* This way we'll deliver the MULTI/..../EXEC block as a whole and
|
|
* both the AOF and the replication link will have the same consistency
|
|
* and atomicity guarantees. */
|
|
execCommandReplicateMulti(c);
|
|
|
|
/* Exec all the queued commands */
|
|
unwatchAllKeys(c); /* Unwatch ASAP otherwise we'll waste CPU cycles */
|
|
orig_argv = c->argv;
|
|
orig_argc = c->argc;
|
|
orig_cmd = c->cmd;
|
|
addReplyMultiBulkLen(c,c->mstate.count);
|
|
for (j = 0; j < c->mstate.count; j++) {
|
|
c->argc = c->mstate.commands[j].argc;
|
|
c->argv = c->mstate.commands[j].argv;
|
|
c->cmd = c->mstate.commands[j].cmd;
|
|
call(c,REDIS_CALL_FULL);
|
|
|
|
/* 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;
|
|
}
|
|
c->argv = orig_argv;
|
|
c->argc = orig_argc;
|
|
c->cmd = orig_cmd;
|
|
freeClientMultiState(c);
|
|
initClientMultiState(c);
|
|
c->flags &= ~(REDIS_MULTI|REDIS_DIRTY_CAS|REDIS_DIRTY_EXEC);
|
|
/* Make sure the EXEC command is always replicated / AOF, since we
|
|
* always send the MULTI command (we can't know beforehand if the
|
|
* next operations will contain at least a modification to the DB). */
|
|
server.dirty++;
|
|
|
|
handle_monitor:
|
|
/* Send EXEC to clients waiting data from MONITOR. We do it here
|
|
* since the natural order of commands execution is actually:
|
|
* MUTLI, EXEC, ... commands inside transaction ...
|
|
* Instead EXEC is flagged as REDIS_CMD_SKIP_MONITOR in the command
|
|
* table, and we do it here with correct ordering. */
|
|
if (listLength(server.monitors) && !server.loading)
|
|
replicationFeedMonitors(c,server.monitors,c->db->id,c->argv,c->argc);
|
|
}
|
|
|
|
/* ===================== WATCH (CAS alike for MULTI/EXEC) ===================
|
|
*
|
|
* The implementation uses a per-DB hash table mapping keys to list of clients
|
|
* WATCHing those keys, so that given a key that is going to be modified
|
|
* we can mark all the associated clients as dirty.
|
|
*
|
|
* Also every client contains a list of WATCHed keys so that's possible to
|
|
* un-watch such keys when the client is freed or when UNWATCH is called. */
|
|
|
|
/* In the client->watched_keys list we need to use watchedKey structures
|
|
* as in order to identify a key in Redis we need both the key name and the
|
|
* DB */
|
|
typedef struct watchedKey {
|
|
robj *key;
|
|
redisDb *db;
|
|
} watchedKey;
|
|
|
|
/* Watch for the specified key */
|
|
void watchForKey(redisClient *c, robj *key) {
|
|
list *clients = NULL;
|
|
listIter li;
|
|
listNode *ln;
|
|
watchedKey *wk;
|
|
|
|
/* Check if we are already watching for this key */
|
|
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 */
|
|
}
|
|
/* This key is not already watched in this DB. Let's add it */
|
|
clients = dictFetchValue(c->db->watched_keys,key);
|
|
if (!clients) {
|
|
clients = listCreate();
|
|
dictAdd(c->db->watched_keys,key,clients);
|
|
incrRefCount(key);
|
|
}
|
|
listAddNodeTail(clients,c);
|
|
/* Add the new key to the lits of keys watched by this client */
|
|
wk = zmalloc(sizeof(*wk));
|
|
wk->key = key;
|
|
wk->db = c->db;
|
|
incrRefCount(key);
|
|
listAddNodeTail(c->watched_keys,wk);
|
|
}
|
|
|
|
/* Unwatch all the keys watched by this client. To clean the EXEC dirty
|
|
* flag is up to the caller. */
|
|
void unwatchAllKeys(redisClient *c) {
|
|
listIter li;
|
|
listNode *ln;
|
|
|
|
if (listLength(c->watched_keys) == 0) return;
|
|
listRewind(c->watched_keys,&li);
|
|
while((ln = listNext(&li))) {
|
|
list *clients;
|
|
watchedKey *wk;
|
|
|
|
/* Lookup the watched key -> clients list and remove the client
|
|
* from the list */
|
|
wk = listNodeValue(ln);
|
|
clients = dictFetchValue(wk->db->watched_keys, wk->key);
|
|
redisAssertWithInfo(c,NULL,clients != NULL);
|
|
listDelNode(clients,listSearchKey(clients,c));
|
|
/* Kill the entry at all if this was the only client */
|
|
if (listLength(clients) == 0)
|
|
dictDelete(wk->db->watched_keys, wk->key);
|
|
/* Remove this watched key from the client->watched list */
|
|
listDelNode(c->watched_keys,ln);
|
|
decrRefCount(wk->key);
|
|
zfree(wk);
|
|
}
|
|
}
|
|
|
|
/* "Touch" a key, so that if this key is being WATCHed by some client the
|
|
* next EXEC will fail. */
|
|
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 REDIS_DIRTY_CAS */
|
|
/* Check if we are already watching for this key */
|
|
listRewind(clients,&li);
|
|
while((ln = listNext(&li))) {
|
|
redisClient *c = listNodeValue(ln);
|
|
|
|
c->flags |= REDIS_DIRTY_CAS;
|
|
}
|
|
}
|
|
|
|
/* On FLUSHDB or FLUSHALL all the watched keys that are present before the
|
|
* flush but will be deleted as effect of the flushing operation should
|
|
* be touched. "dbid" is the DB that's getting the flush. -1 if it is
|
|
* a FLUSHALL operation (all the DBs flushed). */
|
|
void touchWatchedKeysOnFlush(int dbid) {
|
|
listIter li1, li2;
|
|
listNode *ln;
|
|
|
|
/* For every client, check all the waited keys */
|
|
listRewind(server.clients,&li1);
|
|
while((ln = listNext(&li1))) {
|
|
redisClient *c = listNodeValue(ln);
|
|
listRewind(c->watched_keys,&li2);
|
|
while((ln = listNext(&li2))) {
|
|
watchedKey *wk = listNodeValue(ln);
|
|
|
|
/* For every watched key matching the specified DB, if the
|
|
* key exists, mark the client as dirty, as the key will be
|
|
* removed. */
|
|
if (dbid == -1 || wk->db->id == dbid) {
|
|
if (dictFind(wk->db->dict, wk->key->ptr) != NULL)
|
|
c->flags |= REDIS_DIRTY_CAS;
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
void watchCommand(redisClient *c) {
|
|
int j;
|
|
|
|
if (c->flags & REDIS_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);
|
|
}
|
|
|
|
void unwatchCommand(redisClient *c) {
|
|
unwatchAllKeys(c);
|
|
c->flags &= (~REDIS_DIRTY_CAS);
|
|
addReply(c,shared.ok);
|
|
}
|