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|REDIS_DIRTY_CAS);; 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 redisAssertWithInfo(c,NULL,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 } 274