]>
Commit | Line | Data |
---|---|---|
e2641e09 | 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 */ | |
09e2d9ee | 27 | void queueMultiCommand(redisClient *c) { |
e2641e09 | 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; | |
09e2d9ee | 34 | mc->cmd = c->cmd; |
e2641e09 | 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) { | |
3ab20376 | 45 | addReplyError(c,"MULTI calls can not be nested"); |
e2641e09 | 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)) { | |
3ab20376 | 54 | addReplyError(c,"DISCARD without MULTI"); |
e2641e09 | 55 | return; |
56 | } | |
57 | ||
58 | freeClientMultiState(c); | |
59 | initClientMultiState(c); | |
f371e721 | 60 | c->flags &= ~(REDIS_MULTI|REDIS_DIRTY_CAS);; |
e2641e09 | 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) { | |
e2641e09 | 68 | robj *multistring = createStringObject("MULTI",5); |
69 | ||
e2641e09 | 70 | if (server.appendonly) |
1b1f47c9 | 71 | feedAppendOnlyFile(server.multiCommand,c->db->id,&multistring,1); |
e2641e09 | 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; | |
09e2d9ee | 81 | struct redisCommand *orig_cmd; |
e2641e09 | 82 | |
83 | if (!(c->flags & REDIS_MULTI)) { | |
3ab20376 | 84 | addReplyError(c,"EXEC without MULTI"); |
e2641e09 | 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; | |
09e2d9ee | 109 | orig_cmd = c->cmd; |
0537e7bf | 110 | addReplyMultiBulkLen(c,c->mstate.count); |
e2641e09 | 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; | |
09e2d9ee | 114 | c->cmd = c->mstate.commands[j].cmd; |
115 | call(c); | |
6c682e55 PN |
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; | |
09e2d9ee | 120 | c->mstate.commands[j].cmd = c->cmd; |
e2641e09 | 121 | } |
122 | c->argv = orig_argv; | |
123 | c->argc = orig_argc; | |
09e2d9ee | 124 | c->cmd = orig_cmd; |
e2641e09 | 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); | |
eab0e26e | 197 | redisAssertWithInfo(c,NULL,clients != NULL); |
e2641e09 | 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) { | |
3ab20376 | 261 | addReplyError(c,"WATCH inside MULTI is not allowed"); |
e2641e09 | 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 | } |