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