[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) {
        addReplyError(c,"MULTI calls can not be nested");
        return;
    }
    c->flags |= REDIS_MULTI;
    addReply(c,shared.ok);
}

void discardCommand(redisClient *c) {
    if (!(c->flags & REDIS_MULTI)) {
        addReplyError(c,"DISCARD without MULTI");
        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)) {
        addReplyError(c,"EXEC without MULTI");
        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;
    addReplyMultiBulkLen(c,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) {
        addReplyError(c,"WATCH inside MULTI is not allowed");
        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) {
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);
	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)) {
3ab20376	85	addReplyError(c,"EXEC without MULTI");
e2641e09	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;
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;
	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) {
3ab20376	254	addReplyError(c,"WATCH inside MULTI is not allowed");
e2641e09	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	}