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