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