src/multi.c

   1 #include "redis.h"
   2
   3 /* ================================ MULTI/EXEC ============================== */
   4
   5 /* Client state initialization for MULTI/EXEC */
   6 void initClientMultiState(redisClient *c) {
   7     c->mstate.commands = NULL;
   8     c->mstate.count = 0;
   9 }
  10
  11 /* Release all the resources associated with MULTI/EXEC state */
  12 void freeClientMultiState(redisClient *c) {
  13     int j;
  14
  15     for (j = 0; j < c->mstate.count; j++) {
  16         int i;
  17         multiCmd *mc = c->mstate.commands+j;
  18
  19         for (i = 0; i < mc->argc; i++)
  20             decrRefCount(mc->argv[i]);
  21         zfree(mc->argv);
  22     }
  23     zfree(c->mstate.commands);
  24 }
  25
  26 /* Add a new command into the MULTI commands queue */
  27 void queueMultiCommand(redisClient *c) {
  28     multiCmd *mc;
  29     int j;
  30
  31     c->mstate.commands = zrealloc(c->mstate.commands,
  32             sizeof(multiCmd)*(c->mstate.count+1));
  33     mc = c->mstate.commands+c->mstate.count;
  34     mc->cmd = c->cmd;
  35     mc->argc = c->argc;
  36     mc->argv = zmalloc(sizeof(robj*)*c->argc);
  37     memcpy(mc->argv,c->argv,sizeof(robj*)*c->argc);
  38     for (j = 0; j < c->argc; j++)
  39         incrRefCount(mc->argv[j]);
  40     c->mstate.count++;
  41 }
  42
  43 void discardTransaction(redisClient *c) {
  44     freeClientMultiState(c);
  45     initClientMultiState(c);
  46     c->flags &= ~(REDIS_MULTI|REDIS_DIRTY_CAS);;
  47     unwatchAllKeys(c);
  48 }
  49
  50 void multiCommand(redisClient *c) {
  51     if (c->flags & REDIS_MULTI) {
  52         addReplyError(c,"MULTI calls can not be nested");
  53         return;
  54     }
  55     c->flags |= REDIS_MULTI;
  56     addReply(c,shared.ok);
  57 }
  58
  59 void discardCommand(redisClient *c) {
  60     if (!(c->flags & REDIS_MULTI)) {
  61         addReplyError(c,"DISCARD without MULTI");
  62         return;
  63     }
  64     discardTransaction(c);
  65     addReply(c,shared.ok);
  66 }
  67
  68 /* Send a MULTI command to all the slaves and AOF file. Check the execCommand
  69  * implememntation for more information. */
  70 void execCommandReplicateMulti(redisClient *c) {
  71     robj *multistring = createStringObject("MULTI",5);
  72
  73     if (server.aof_state != REDIS_AOF_OFF)
  74         feedAppendOnlyFile(server.multiCommand,c->db->id,&multistring,1);
  75     if (listLength(server.slaves))
  76         replicationFeedSlaves(server.slaves,c->db->id,&multistring,1);
  77     decrRefCount(multistring);
  78 }
  79
  80 void execCommand(redisClient *c) {
  81     int j;
  82     robj **orig_argv;
  83     int orig_argc;
  84     struct redisCommand *orig_cmd;
  85
  86     if (!(c->flags & REDIS_MULTI)) {
  87         addReplyError(c,"EXEC without MULTI");
  88         return;
  89     }
  90
  91     /* Check if we need to abort the EXEC if some WATCHed key was touched.
  92      * A failed EXEC will return a multi bulk nil object. */
  93     if (c->flags & REDIS_DIRTY_CAS) {
  94         freeClientMultiState(c);
  95         initClientMultiState(c);
  96         c->flags &= ~(REDIS_MULTI|REDIS_DIRTY_CAS);
  97         unwatchAllKeys(c);
  98         addReply(c,shared.nullmultibulk);
  99         goto handle_monitor;
 100     }
 101
 102     /* Replicate a MULTI request now that we are sure the block is executed.
 103      * This way we'll deliver the MULTI/..../EXEC block as a whole and
 104      * both the AOF and the replication link will have the same consistency
 105      * and atomicity guarantees. */
 106     execCommandReplicateMulti(c);
 107
 108     /* Exec all the queued commands */
 109     unwatchAllKeys(c); /* Unwatch ASAP otherwise we'll waste CPU cycles */
 110     orig_argv = c->argv;
 111     orig_argc = c->argc;
 112     orig_cmd = c->cmd;
 113     addReplyMultiBulkLen(c,c->mstate.count);
 114     for (j = 0; j < c->mstate.count; j++) {
 115         c->argc = c->mstate.commands[j].argc;
 116         c->argv = c->mstate.commands[j].argv;
 117         c->cmd = c->mstate.commands[j].cmd;
 118         call(c,REDIS_CALL_FULL);
 119
 120         /* Commands may alter argc/argv, restore mstate. */
 121         c->mstate.commands[j].argc = c->argc;
 122         c->mstate.commands[j].argv = c->argv;
 123         c->mstate.commands[j].cmd = c->cmd;
 124     }
 125     c->argv = orig_argv;
 126     c->argc = orig_argc;
 127     c->cmd = orig_cmd;
 128     freeClientMultiState(c);
 129     initClientMultiState(c);
 130     c->flags &= ~(REDIS_MULTI|REDIS_DIRTY_CAS);
 131     /* Make sure the EXEC command is always replicated / AOF, since we
 132      * always send the MULTI command (we can't know beforehand if the
 133      * next operations will contain at least a modification to the DB). */
 134     server.dirty++;
 135
 136 handle_monitor:
 137     /* Send EXEC to clients waiting data from MONITOR. We do it here
 138      * since the natural order of commands execution is actually:
 139      * MUTLI, EXEC, ... commands inside transaction ...
 140      * Instead EXEC is flagged as REDIS_CMD_SKIP_MONITOR in the command
 141      * table, and we do it here with correct ordering. */
 142     if (listLength(server.monitors) && !server.loading)
 143         replicationFeedMonitors(c,server.monitors,c->db->id,c->argv,c->argc);
 144 }
 145
 146 /* ===================== WATCH (CAS alike for MULTI/EXEC) ===================
 147  *
 148  * The implementation uses a per-DB hash table mapping keys to list of clients
 149  * WATCHing those keys, so that given a key that is going to be modified
 150  * we can mark all the associated clients as dirty.
 151  *
 152  * Also every client contains a list of WATCHed keys so that's possible to
 153  * un-watch such keys when the client is freed or when UNWATCH is called. */
 154
 155 /* In the client->watched_keys list we need to use watchedKey structures
 156  * as in order to identify a key in Redis we need both the key name and the
 157  * DB */
 158 typedef struct watchedKey {
 159     robj *key;
 160     redisDb *db;
 161 } watchedKey;
 162
 163 /* Watch for the specified key */
 164 void watchForKey(redisClient *c, robj *key) {
 165     list *clients = NULL;
 166     listIter li;
 167     listNode *ln;
 168     watchedKey *wk;
 169
 170     /* Check if we are already watching for this key */
 171     listRewind(c->watched_keys,&li);
 172     while((ln = listNext(&li))) {
 173         wk = listNodeValue(ln);
 174         if (wk->db == c->db && equalStringObjects(key,wk->key))
 175             return; /* Key already watched */
 176     }
 177     /* This key is not already watched in this DB. Let's add it */
 178     clients = dictFetchValue(c->db->watched_keys,key);
 179     if (!clients) {
 180         clients = listCreate();
 181         dictAdd(c->db->watched_keys,key,clients);
 182         incrRefCount(key);
 183     }
 184     listAddNodeTail(clients,c);
 185     /* Add the new key to the lits of keys watched by this client */
 186     wk = zmalloc(sizeof(*wk));
 187     wk->key = key;
 188     wk->db = c->db;
 189     incrRefCount(key);
 190     listAddNodeTail(c->watched_keys,wk);
 191 }
 192
 193 /* Unwatch all the keys watched by this client. To clean the EXEC dirty
 194  * flag is up to the caller. */
 195 void unwatchAllKeys(redisClient *c) {
 196     listIter li;
 197     listNode *ln;
 198
 199     if (listLength(c->watched_keys) == 0) return;
 200     listRewind(c->watched_keys,&li);
 201     while((ln = listNext(&li))) {
 202         list *clients;
 203         watchedKey *wk;
 204
 205         /* Lookup the watched key -> clients list and remove the client
 206          * from the list */
 207         wk = listNodeValue(ln);
 208         clients = dictFetchValue(wk->db->watched_keys, wk->key);
 209         redisAssertWithInfo(c,NULL,clients != NULL);
 210         listDelNode(clients,listSearchKey(clients,c));
 211         /* Kill the entry at all if this was the only client */
 212         if (listLength(clients) == 0)
 213             dictDelete(wk->db->watched_keys, wk->key);
 214         /* Remove this watched key from the client->watched list */
 215         listDelNode(c->watched_keys,ln);
 216         decrRefCount(wk->key);
 217         zfree(wk);
 218     }
 219 }
 220
 221 /* "Touch" a key, so that if this key is being WATCHed by some client the
 222  * next EXEC will fail. */
 223 void touchWatchedKey(redisDb *db, robj *key) {
 224     list *clients;
 225     listIter li;
 226     listNode *ln;
 227
 228     if (dictSize(db->watched_keys) == 0) return;
 229     clients = dictFetchValue(db->watched_keys, key);
 230     if (!clients) return;
 231
 232     /* Mark all the clients watching this key as REDIS_DIRTY_CAS */
 233     /* Check if we are already watching for this key */
 234     listRewind(clients,&li);
 235     while((ln = listNext(&li))) {
 236         redisClient *c = listNodeValue(ln);
 237
 238         c->flags |= REDIS_DIRTY_CAS;
 239     }
 240 }
 241
 242 /* On FLUSHDB or FLUSHALL all the watched keys that are present before the
 243  * flush but will be deleted as effect of the flushing operation should
 244  * be touched. "dbid" is the DB that's getting the flush. -1 if it is
 245  * a FLUSHALL operation (all the DBs flushed). */
 246 void touchWatchedKeysOnFlush(int dbid) {
 247     listIter li1, li2;
 248     listNode *ln;
 249
 250     /* For every client, check all the waited keys */
 251     listRewind(server.clients,&li1);
 252     while((ln = listNext(&li1))) {
 253         redisClient *c = listNodeValue(ln);
 254         listRewind(c->watched_keys,&li2);
 255         while((ln = listNext(&li2))) {
 256             watchedKey *wk = listNodeValue(ln);
 257
 258             /* For every watched key matching the specified DB, if the
 259              * key exists, mark the client as dirty, as the key will be
 260              * removed. */
 261             if (dbid == -1 || wk->db->id == dbid) {
 262                 if (dictFind(wk->db->dict, wk->key->ptr) != NULL)
 263                     c->flags |= REDIS_DIRTY_CAS;
 264             }
 265         }
 266     }
 267 }
 268
 269 void watchCommand(redisClient *c) {
 270     int j;
 271
 272     if (c->flags & REDIS_MULTI) {
 273         addReplyError(c,"WATCH inside MULTI is not allowed");
 274         return;
 275     }
 276     for (j = 1; j < c->argc; j++)
 277         watchForKey(c,c->argv[j]);
 278     addReply(c,shared.ok);
 279 }
 280
 281 void unwatchCommand(redisClient *c) {
 282     unwatchAllKeys(c);
 283     c->flags &= (~REDIS_DIRTY_CAS);
 284     addReply(c,shared.ok);
 285 }