]>
Commit | Line | Data |
---|---|---|
4365e5b2 | 1 | /* |
2 | * Copyright (c) 2009-2012, Salvatore Sanfilippo <antirez at gmail dot com> | |
3 | * All rights reserved. | |
4 | * | |
5 | * Redistribution and use in source and binary forms, with or without | |
6 | * modification, are permitted provided that the following conditions are met: | |
7 | * | |
8 | * * Redistributions of source code must retain the above copyright notice, | |
9 | * this list of conditions and the following disclaimer. | |
10 | * * Redistributions in binary form must reproduce the above copyright | |
11 | * notice, this list of conditions and the following disclaimer in the | |
12 | * documentation and/or other materials provided with the distribution. | |
13 | * * Neither the name of Redis nor the names of its contributors may be used | |
14 | * to endorse or promote products derived from this software without | |
15 | * specific prior written permission. | |
16 | * | |
17 | * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" | |
18 | * AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE | |
19 | * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE | |
20 | * ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE | |
21 | * LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR | |
22 | * CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF | |
23 | * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS | |
24 | * INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN | |
25 | * CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) | |
26 | * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE | |
27 | * POSSIBILITY OF SUCH DAMAGE. | |
28 | */ | |
29 | ||
e2641e09 | 30 | #include "redis.h" |
31 | ||
32 | /* ================================ MULTI/EXEC ============================== */ | |
33 | ||
34 | /* Client state initialization for MULTI/EXEC */ | |
35 | void initClientMultiState(redisClient *c) { | |
36 | c->mstate.commands = NULL; | |
37 | c->mstate.count = 0; | |
38 | } | |
39 | ||
40 | /* Release all the resources associated with MULTI/EXEC state */ | |
41 | void freeClientMultiState(redisClient *c) { | |
42 | int j; | |
43 | ||
44 | for (j = 0; j < c->mstate.count; j++) { | |
45 | int i; | |
46 | multiCmd *mc = c->mstate.commands+j; | |
47 | ||
48 | for (i = 0; i < mc->argc; i++) | |
49 | decrRefCount(mc->argv[i]); | |
50 | zfree(mc->argv); | |
51 | } | |
52 | zfree(c->mstate.commands); | |
53 | } | |
54 | ||
55 | /* Add a new command into the MULTI commands queue */ | |
09e2d9ee | 56 | void queueMultiCommand(redisClient *c) { |
e2641e09 | 57 | multiCmd *mc; |
58 | int j; | |
59 | ||
60 | c->mstate.commands = zrealloc(c->mstate.commands, | |
61 | sizeof(multiCmd)*(c->mstate.count+1)); | |
62 | mc = c->mstate.commands+c->mstate.count; | |
09e2d9ee | 63 | mc->cmd = c->cmd; |
e2641e09 | 64 | mc->argc = c->argc; |
65 | mc->argv = zmalloc(sizeof(robj*)*c->argc); | |
66 | memcpy(mc->argv,c->argv,sizeof(robj*)*c->argc); | |
67 | for (j = 0; j < c->argc; j++) | |
68 | incrRefCount(mc->argv[j]); | |
69 | c->mstate.count++; | |
70 | } | |
71 | ||
f3fd419f | 72 | void discardTransaction(redisClient *c) { |
73 | freeClientMultiState(c); | |
74 | initClientMultiState(c); | |
75 | c->flags &= ~(REDIS_MULTI|REDIS_DIRTY_CAS);; | |
76 | unwatchAllKeys(c); | |
77 | } | |
78 | ||
e2641e09 | 79 | void multiCommand(redisClient *c) { |
80 | if (c->flags & REDIS_MULTI) { | |
3ab20376 | 81 | addReplyError(c,"MULTI calls can not be nested"); |
e2641e09 | 82 | return; |
83 | } | |
84 | c->flags |= REDIS_MULTI; | |
85 | addReply(c,shared.ok); | |
86 | } | |
87 | ||
88 | void discardCommand(redisClient *c) { | |
89 | if (!(c->flags & REDIS_MULTI)) { | |
3ab20376 | 90 | addReplyError(c,"DISCARD without MULTI"); |
e2641e09 | 91 | return; |
92 | } | |
f3fd419f | 93 | discardTransaction(c); |
e2641e09 | 94 | addReply(c,shared.ok); |
95 | } | |
96 | ||
97 | /* Send a MULTI command to all the slaves and AOF file. Check the execCommand | |
98 | * implememntation for more information. */ | |
99 | void execCommandReplicateMulti(redisClient *c) { | |
e2641e09 | 100 | robj *multistring = createStringObject("MULTI",5); |
101 | ||
e394114d | 102 | if (server.aof_state != REDIS_AOF_OFF) |
1b1f47c9 | 103 | feedAppendOnlyFile(server.multiCommand,c->db->id,&multistring,1); |
e2641e09 | 104 | if (listLength(server.slaves)) |
105 | replicationFeedSlaves(server.slaves,c->db->id,&multistring,1); | |
106 | decrRefCount(multistring); | |
107 | } | |
108 | ||
109 | void execCommand(redisClient *c) { | |
110 | int j; | |
111 | robj **orig_argv; | |
112 | int orig_argc; | |
09e2d9ee | 113 | struct redisCommand *orig_cmd; |
e2641e09 | 114 | |
115 | if (!(c->flags & REDIS_MULTI)) { | |
3ab20376 | 116 | addReplyError(c,"EXEC without MULTI"); |
e2641e09 | 117 | return; |
118 | } | |
119 | ||
120 | /* Check if we need to abort the EXEC if some WATCHed key was touched. | |
121 | * A failed EXEC will return a multi bulk nil object. */ | |
122 | if (c->flags & REDIS_DIRTY_CAS) { | |
123 | freeClientMultiState(c); | |
124 | initClientMultiState(c); | |
125 | c->flags &= ~(REDIS_MULTI|REDIS_DIRTY_CAS); | |
126 | unwatchAllKeys(c); | |
127 | addReply(c,shared.nullmultibulk); | |
a1b1c1ea | 128 | goto handle_monitor; |
e2641e09 | 129 | } |
130 | ||
131 | /* Replicate a MULTI request now that we are sure the block is executed. | |
132 | * This way we'll deliver the MULTI/..../EXEC block as a whole and | |
133 | * both the AOF and the replication link will have the same consistency | |
134 | * and atomicity guarantees. */ | |
135 | execCommandReplicateMulti(c); | |
136 | ||
137 | /* Exec all the queued commands */ | |
138 | unwatchAllKeys(c); /* Unwatch ASAP otherwise we'll waste CPU cycles */ | |
139 | orig_argv = c->argv; | |
140 | orig_argc = c->argc; | |
09e2d9ee | 141 | orig_cmd = c->cmd; |
0537e7bf | 142 | addReplyMultiBulkLen(c,c->mstate.count); |
e2641e09 | 143 | for (j = 0; j < c->mstate.count; j++) { |
144 | c->argc = c->mstate.commands[j].argc; | |
145 | c->argv = c->mstate.commands[j].argv; | |
09e2d9ee | 146 | c->cmd = c->mstate.commands[j].cmd; |
ce8b772b | 147 | call(c,REDIS_CALL_FULL); |
6c682e55 PN |
148 | |
149 | /* Commands may alter argc/argv, restore mstate. */ | |
150 | c->mstate.commands[j].argc = c->argc; | |
151 | c->mstate.commands[j].argv = c->argv; | |
09e2d9ee | 152 | c->mstate.commands[j].cmd = c->cmd; |
e2641e09 | 153 | } |
154 | c->argv = orig_argv; | |
155 | c->argc = orig_argc; | |
09e2d9ee | 156 | c->cmd = orig_cmd; |
e2641e09 | 157 | freeClientMultiState(c); |
158 | initClientMultiState(c); | |
159 | c->flags &= ~(REDIS_MULTI|REDIS_DIRTY_CAS); | |
160 | /* Make sure the EXEC command is always replicated / AOF, since we | |
161 | * always send the MULTI command (we can't know beforehand if the | |
162 | * next operations will contain at least a modification to the DB). */ | |
163 | server.dirty++; | |
a1b1c1ea | 164 | |
165 | handle_monitor: | |
166 | /* Send EXEC to clients waiting data from MONITOR. We do it here | |
167 | * since the natural order of commands execution is actually: | |
168 | * MUTLI, EXEC, ... commands inside transaction ... | |
169 | * Instead EXEC is flagged as REDIS_CMD_SKIP_MONITOR in the command | |
170 | * table, and we do it here with correct ordering. */ | |
171 | if (listLength(server.monitors) && !server.loading) | |
172 | replicationFeedMonitors(c,server.monitors,c->db->id,c->argv,c->argc); | |
e2641e09 | 173 | } |
174 | ||
175 | /* ===================== WATCH (CAS alike for MULTI/EXEC) =================== | |
176 | * | |
177 | * The implementation uses a per-DB hash table mapping keys to list of clients | |
178 | * WATCHing those keys, so that given a key that is going to be modified | |
179 | * we can mark all the associated clients as dirty. | |
180 | * | |
181 | * Also every client contains a list of WATCHed keys so that's possible to | |
182 | * un-watch such keys when the client is freed or when UNWATCH is called. */ | |
183 | ||
184 | /* In the client->watched_keys list we need to use watchedKey structures | |
185 | * as in order to identify a key in Redis we need both the key name and the | |
186 | * DB */ | |
187 | typedef struct watchedKey { | |
188 | robj *key; | |
189 | redisDb *db; | |
190 | } watchedKey; | |
191 | ||
192 | /* Watch for the specified key */ | |
193 | void watchForKey(redisClient *c, robj *key) { | |
194 | list *clients = NULL; | |
195 | listIter li; | |
196 | listNode *ln; | |
197 | watchedKey *wk; | |
198 | ||
199 | /* Check if we are already watching for this key */ | |
200 | listRewind(c->watched_keys,&li); | |
201 | while((ln = listNext(&li))) { | |
202 | wk = listNodeValue(ln); | |
203 | if (wk->db == c->db && equalStringObjects(key,wk->key)) | |
204 | return; /* Key already watched */ | |
205 | } | |
206 | /* This key is not already watched in this DB. Let's add it */ | |
207 | clients = dictFetchValue(c->db->watched_keys,key); | |
208 | if (!clients) { | |
209 | clients = listCreate(); | |
210 | dictAdd(c->db->watched_keys,key,clients); | |
211 | incrRefCount(key); | |
212 | } | |
213 | listAddNodeTail(clients,c); | |
214 | /* Add the new key to the lits of keys watched by this client */ | |
215 | wk = zmalloc(sizeof(*wk)); | |
216 | wk->key = key; | |
217 | wk->db = c->db; | |
218 | incrRefCount(key); | |
219 | listAddNodeTail(c->watched_keys,wk); | |
220 | } | |
221 | ||
222 | /* Unwatch all the keys watched by this client. To clean the EXEC dirty | |
223 | * flag is up to the caller. */ | |
224 | void unwatchAllKeys(redisClient *c) { | |
225 | listIter li; | |
226 | listNode *ln; | |
227 | ||
228 | if (listLength(c->watched_keys) == 0) return; | |
229 | listRewind(c->watched_keys,&li); | |
230 | while((ln = listNext(&li))) { | |
231 | list *clients; | |
232 | watchedKey *wk; | |
233 | ||
234 | /* Lookup the watched key -> clients list and remove the client | |
235 | * from the list */ | |
236 | wk = listNodeValue(ln); | |
237 | clients = dictFetchValue(wk->db->watched_keys, wk->key); | |
eab0e26e | 238 | redisAssertWithInfo(c,NULL,clients != NULL); |
e2641e09 | 239 | listDelNode(clients,listSearchKey(clients,c)); |
240 | /* Kill the entry at all if this was the only client */ | |
241 | if (listLength(clients) == 0) | |
242 | dictDelete(wk->db->watched_keys, wk->key); | |
243 | /* Remove this watched key from the client->watched list */ | |
244 | listDelNode(c->watched_keys,ln); | |
245 | decrRefCount(wk->key); | |
246 | zfree(wk); | |
247 | } | |
248 | } | |
249 | ||
250 | /* "Touch" a key, so that if this key is being WATCHed by some client the | |
251 | * next EXEC will fail. */ | |
252 | void touchWatchedKey(redisDb *db, robj *key) { | |
253 | list *clients; | |
254 | listIter li; | |
255 | listNode *ln; | |
256 | ||
257 | if (dictSize(db->watched_keys) == 0) return; | |
258 | clients = dictFetchValue(db->watched_keys, key); | |
259 | if (!clients) return; | |
260 | ||
261 | /* Mark all the clients watching this key as REDIS_DIRTY_CAS */ | |
262 | /* Check if we are already watching for this key */ | |
263 | listRewind(clients,&li); | |
264 | while((ln = listNext(&li))) { | |
265 | redisClient *c = listNodeValue(ln); | |
266 | ||
267 | c->flags |= REDIS_DIRTY_CAS; | |
268 | } | |
269 | } | |
270 | ||
271 | /* On FLUSHDB or FLUSHALL all the watched keys that are present before the | |
272 | * flush but will be deleted as effect of the flushing operation should | |
273 | * be touched. "dbid" is the DB that's getting the flush. -1 if it is | |
274 | * a FLUSHALL operation (all the DBs flushed). */ | |
275 | void touchWatchedKeysOnFlush(int dbid) { | |
276 | listIter li1, li2; | |
277 | listNode *ln; | |
278 | ||
279 | /* For every client, check all the waited keys */ | |
280 | listRewind(server.clients,&li1); | |
281 | while((ln = listNext(&li1))) { | |
282 | redisClient *c = listNodeValue(ln); | |
283 | listRewind(c->watched_keys,&li2); | |
284 | while((ln = listNext(&li2))) { | |
285 | watchedKey *wk = listNodeValue(ln); | |
286 | ||
287 | /* For every watched key matching the specified DB, if the | |
288 | * key exists, mark the client as dirty, as the key will be | |
289 | * removed. */ | |
290 | if (dbid == -1 || wk->db->id == dbid) { | |
291 | if (dictFind(wk->db->dict, wk->key->ptr) != NULL) | |
292 | c->flags |= REDIS_DIRTY_CAS; | |
293 | } | |
294 | } | |
295 | } | |
296 | } | |
297 | ||
298 | void watchCommand(redisClient *c) { | |
299 | int j; | |
300 | ||
301 | if (c->flags & REDIS_MULTI) { | |
3ab20376 | 302 | addReplyError(c,"WATCH inside MULTI is not allowed"); |
e2641e09 | 303 | return; |
304 | } | |
305 | for (j = 1; j < c->argc; j++) | |
306 | watchForKey(c,c->argv[j]); | |
307 | addReply(c,shared.ok); | |
308 | } | |
309 | ||
310 | void unwatchCommand(redisClient *c) { | |
311 | unwatchAllKeys(c); | |
312 | c->flags &= (~REDIS_DIRTY_CAS); | |
313 | addReply(c,shared.ok); | |
314 | } |