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