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