#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) { 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 = c->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|REDIS_DIRTY_CAS);; 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) { robj *multistring = createStringObject("MULTI",5); if (server.appendonly) feedAppendOnlyFile(server.multiCommand,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; struct redisCommand *orig_cmd; 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; orig_cmd = c->cmd; 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; c->cmd = c->mstate.commands[j].cmd; call(c); /* Commands may alter argc/argv, restore mstate. */ c->mstate.commands[j].argc = c->argc; c->mstate.commands[j].argv = c->argv; c->mstate.commands[j].cmd = c->cmd; } c->argv = orig_argv; c->argc = orig_argc; c->cmd = orig_cmd; 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); redisAssertWithInfo(c,NULL,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); }