[redis.git] / src / multi.c

#include "redis.h"

/* ================================ MULTI/EXEC ============================== */

/* Client state initialization for MULTI/EXEC */
void initClientMultiState(redisClient *c) {
    c->mstate.commands = NULL;
    c->mstate.count = 0;
}

/* Release all the resources associated with MULTI/EXEC state */
void freeClientMultiState(redisClient *c) {
    int j;

    for (j = 0; j < c->mstate.count; j++) {
        int i;
        multiCmd *mc = c->mstate.commands+j;

        for (i = 0; i < mc->argc; i++)
            decrRefCount(mc->argv[i]);
        zfree(mc->argv);
    }
    zfree(c->mstate.commands);
}

/* Add a new command into the MULTI commands queue */
void queueMultiCommand(redisClient *c, struct redisCommand *cmd) {
    multiCmd *mc;
    int j;

    c->mstate.commands = zrealloc(c->mstate.commands,
            sizeof(multiCmd)*(c->mstate.count+1));
    mc = c->mstate.commands+c->mstate.count;
    mc->cmd = cmd;
    mc->argc = c->argc;
    mc->argv = zmalloc(sizeof(robj*)*c->argc);
    memcpy(mc->argv,c->argv,sizeof(robj*)*c->argc);
    for (j = 0; j < c->argc; j++)
        incrRefCount(mc->argv[j]);
    c->mstate.count++;
}

void multiCommand(redisClient *c) {
    if (c->flags & REDIS_MULTI) {
        addReplySds(c,sdsnew("-ERR MULTI calls can not be nested\r\n"));
        return;
    }
    c->flags |= REDIS_MULTI;
    addReply(c,shared.ok);
}

void discardCommand(redisClient *c) {
    if (!(c->flags & REDIS_MULTI)) {
        addReplySds(c,sdsnew("-ERR DISCARD without MULTI\r\n"));
        return;
    }

    freeClientMultiState(c);
    initClientMultiState(c);
    c->flags &= (~REDIS_MULTI);
    unwatchAllKeys(c);
    addReply(c,shared.ok);
}

/* Send a MULTI command to all the slaves and AOF file. Check the execCommand
 * implememntation for more information. */
void execCommandReplicateMulti(redisClient *c) {
    struct redisCommand *cmd;
    robj *multistring = createStringObject("MULTI",5);

    cmd = lookupCommand("multi");
    if (server.appendonly)
        feedAppendOnlyFile(cmd,c->db->id,&multistring,1);
    if (listLength(server.slaves))
        replicationFeedSlaves(server.slaves,c->db->id,&multistring,1);
    decrRefCount(multistring);
}

void execCommand(redisClient *c) {
    int j;
    robj **orig_argv;
    int orig_argc;

    if (!(c->flags & REDIS_MULTI)) {
        addReplySds(c,sdsnew("-ERR EXEC without MULTI\r\n"));
        return;
    }

    /* Check if we need to abort the EXEC if some WATCHed key was touched.
     * A failed EXEC will return a multi bulk nil object. */
    if (c->flags & REDIS_DIRTY_CAS) {
        freeClientMultiState(c);
        initClientMultiState(c);
        c->flags &= ~(REDIS_MULTI|REDIS_DIRTY_CAS);
        unwatchAllKeys(c);
        addReply(c,shared.nullmultibulk);
        return;
    }

    /* Replicate a MULTI request now that we are sure the block is executed.
     * This way we'll deliver the MULTI/..../EXEC block as a whole and
     * both the AOF and the replication link will have the same consistency
     * and atomicity guarantees. */
    execCommandReplicateMulti(c);

    /* Exec all the queued commands */
    unwatchAllKeys(c); /* Unwatch ASAP otherwise we'll waste CPU cycles */
    orig_argv = c->argv;
    orig_argc = c->argc;
    addReplySds(c,sdscatprintf(sdsempty(),"*%d\r\n",c->mstate.count));
    for (j = 0; j < c->mstate.count; j++) {
        c->argc = c->mstate.commands[j].argc;
        c->argv = c->mstate.commands[j].argv;
        call(c,c->mstate.commands[j].cmd);
    }
    c->argv = orig_argv;
    c->argc = orig_argc;
    freeClientMultiState(c);
    initClientMultiState(c);
    c->flags &= ~(REDIS_MULTI|REDIS_DIRTY_CAS);
    /* Make sure the EXEC command is always replicated / AOF, since we
     * always send the MULTI command (we can't know beforehand if the
     * next operations will contain at least a modification to the DB). */
    server.dirty++;
}

/* ===================== WATCH (CAS alike for MULTI/EXEC) ===================
 *
 * The implementation uses a per-DB hash table mapping keys to list of clients
 * WATCHing those keys, so that given a key that is going to be modified
 * we can mark all the associated clients as dirty.
 *
 * Also every client contains a list of WATCHed keys so that's possible to
 * un-watch such keys when the client is freed or when UNWATCH is called. */

/* In the client->watched_keys list we need to use watchedKey structures
 * as in order to identify a key in Redis we need both the key name and the
 * DB */
typedef struct watchedKey {
    robj *key;
    redisDb *db;
} watchedKey;

/* Watch for the specified key */
void watchForKey(redisClient *c, robj *key) {
    list *clients = NULL;
    listIter li;
    listNode *ln;
    watchedKey *wk;

    /* Check if we are already watching for this key */
    listRewind(c->watched_keys,&li);
    while((ln = listNext(&li))) {
        wk = listNodeValue(ln);
        if (wk->db == c->db && equalStringObjects(key,wk->key))
            return; /* Key already watched */
    }
    /* This key is not already watched in this DB. Let's add it */
    clients = dictFetchValue(c->db->watched_keys,key);
    if (!clients) { 
        clients = listCreate();
        dictAdd(c->db->watched_keys,key,clients);
        incrRefCount(key);
    }
    listAddNodeTail(clients,c);
    /* Add the new key to the lits of keys watched by this client */
    wk = zmalloc(sizeof(*wk));
    wk->key = key;
    wk->db = c->db;
    incrRefCount(key);
    listAddNodeTail(c->watched_keys,wk);
}

/* Unwatch all the keys watched by this client. To clean the EXEC dirty
 * flag is up to the caller. */
void unwatchAllKeys(redisClient *c) {
    listIter li;
    listNode *ln;

    if (listLength(c->watched_keys) == 0) return;
    listRewind(c->watched_keys,&li);
    while((ln = listNext(&li))) {
        list *clients;
        watchedKey *wk;

        /* Lookup the watched key -> clients list and remove the client
         * from the list */
        wk = listNodeValue(ln);
        clients = dictFetchValue(wk->db->watched_keys, wk->key);
        redisAssert(clients != NULL);
        listDelNode(clients,listSearchKey(clients,c));
        /* Kill the entry at all if this was the only client */
        if (listLength(clients) == 0)
            dictDelete(wk->db->watched_keys, wk->key);
        /* Remove this watched key from the client->watched list */
        listDelNode(c->watched_keys,ln);
        decrRefCount(wk->key);
        zfree(wk);
    }
}

/* "Touch" a key, so that if this key is being WATCHed by some client the
 * next EXEC will fail. */
void touchWatchedKey(redisDb *db, robj *key) {
    list *clients;
    listIter li;
    listNode *ln;

    if (dictSize(db->watched_keys) == 0) return;
    clients = dictFetchValue(db->watched_keys, key);
    if (!clients) return;

    /* Mark all the clients watching this key as REDIS_DIRTY_CAS */
    /* Check if we are already watching for this key */
    listRewind(clients,&li);
    while((ln = listNext(&li))) {
        redisClient *c = listNodeValue(ln);

        c->flags |= REDIS_DIRTY_CAS;
    }
}

/* On FLUSHDB or FLUSHALL all the watched keys that are present before the
 * flush but will be deleted as effect of the flushing operation should
 * be touched. "dbid" is the DB that's getting the flush. -1 if it is
 * a FLUSHALL operation (all the DBs flushed). */
void touchWatchedKeysOnFlush(int dbid) {
    listIter li1, li2;
    listNode *ln;

    /* For every client, check all the waited keys */
    listRewind(server.clients,&li1);
    while((ln = listNext(&li1))) {
        redisClient *c = listNodeValue(ln);
        listRewind(c->watched_keys,&li2);
        while((ln = listNext(&li2))) {
            watchedKey *wk = listNodeValue(ln);

            /* For every watched key matching the specified DB, if the
             * key exists, mark the client as dirty, as the key will be
             * removed. */
            if (dbid == -1 || wk->db->id == dbid) {
                if (dictFind(wk->db->dict, wk->key->ptr) != NULL)
                    c->flags |= REDIS_DIRTY_CAS;
            }
        }
    }
}

void watchCommand(redisClient *c) {
    int j;

    if (c->flags & REDIS_MULTI) {
        addReplySds(c,sdsnew("-ERR WATCH inside MULTI is not allowed\r\n"));
        return;
    }
    for (j = 1; j < c->argc; j++)
        watchForKey(c,c->argv[j]);
    addReply(c,shared.ok);
}

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