Back to home page

Redis cross reference

 
 

    


0001 /*
0002  * Copyright (c) 2009-2012, Salvatore Sanfilippo <antirez at gmail dot com>
0003  * All rights reserved.
0004  *
0005  * Redistribution and use in source and binary forms, with or without
0006  * modification, are permitted provided that the following conditions are met:
0007  *
0008  *   * Redistributions of source code must retain the above copyright notice,
0009  *     this list of conditions and the following disclaimer.
0010  *   * Redistributions in binary form must reproduce the above copyright
0011  *     notice, this list of conditions and the following disclaimer in the
0012  *     documentation and/or other materials provided with the distribution.
0013  *   * Neither the name of Redis nor the names of its contributors may be used
0014  *     to endorse or promote products derived from this software without
0015  *     specific prior written permission.
0016  *
0017  * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
0018  * AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
0019  * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
0020  * ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE
0021  * LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
0022  * CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
0023  * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
0024  * INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
0025  * CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
0026  * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
0027  * POSSIBILITY OF SUCH DAMAGE.
0028  */
0029 
0030 #include "redis.h"
0031 
0032 /* ================================ MULTI/EXEC ============================== */
0033 
0034 /* Client state initialization for MULTI/EXEC */
0035 void initClientMultiState(redisClient *c) {
0036     c->mstate.commands = NULL;
0037     c->mstate.count = 0;
0038 }
0039 
0040 /* Release all the resources associated with MULTI/EXEC state */
0041 void freeClientMultiState(redisClient *c) {
0042     int j;
0043 
0044     for (j = 0; j < c->mstate.count; j++) {
0045         int i;
0046         multiCmd *mc = c->mstate.commands+j;
0047 
0048         for (i = 0; i < mc->argc; i++)
0049             decrRefCount(mc->argv[i]);
0050         zfree(mc->argv);
0051     }
0052     zfree(c->mstate.commands);
0053 }
0054 
0055 /* Add a new command into the MULTI commands queue */
0056 void queueMultiCommand(redisClient *c) {
0057     multiCmd *mc;
0058     int j;
0059 
0060     c->mstate.commands = zrealloc(c->mstate.commands,
0061             sizeof(multiCmd)*(c->mstate.count+1));
0062     mc = c->mstate.commands+c->mstate.count;
0063     mc->cmd = c->cmd;
0064     mc->argc = c->argc;
0065     mc->argv = zmalloc(sizeof(robj*)*c->argc);
0066     memcpy(mc->argv,c->argv,sizeof(robj*)*c->argc);
0067     for (j = 0; j < c->argc; j++)
0068         incrRefCount(mc->argv[j]);
0069     c->mstate.count++;
0070 }
0071 
0072 void discardTransaction(redisClient *c) {
0073     freeClientMultiState(c);
0074     initClientMultiState(c);
0075     c->flags &= ~(REDIS_MULTI|REDIS_DIRTY_CAS|REDIS_DIRTY_EXEC);;
0076     unwatchAllKeys(c);
0077 }
0078 
0079 /* Flag the transacation as DIRTY_EXEC so that EXEC will fail.
0080  * Should be called every time there is an error while queueing a command. */
0081 void flagTransaction(redisClient *c) {
0082     if (c->flags & REDIS_MULTI)
0083         c->flags |= REDIS_DIRTY_EXEC;
0084 }
0085 
0086 void multiCommand(redisClient *c) {
0087     if (c->flags & REDIS_MULTI) {
0088         addReplyError(c,"MULTI calls can not be nested");
0089         return;
0090     }
0091     c->flags |= REDIS_MULTI;
0092     addReply(c,shared.ok);
0093 }
0094 
0095 void discardCommand(redisClient *c) {
0096     if (!(c->flags & REDIS_MULTI)) {
0097         addReplyError(c,"DISCARD without MULTI");
0098         return;
0099     }
0100     discardTransaction(c);
0101     addReply(c,shared.ok);
0102 }
0103 
0104 /* Send a MULTI command to all the slaves and AOF file. Check the execCommand
0105  * implementation for more information. */
0106 void execCommandPropagateMulti(redisClient *c) {
0107     robj *multistring = createStringObject("MULTI",5);
0108 
0109     propagate(server.multiCommand,c->db->id,&multistring,1,
0110               REDIS_PROPAGATE_AOF|REDIS_PROPAGATE_REPL);
0111     decrRefCount(multistring);
0112 }
0113 
0114 void execCommand(redisClient *c) {
0115     int j;
0116     robj **orig_argv;
0117     int orig_argc;
0118     struct redisCommand *orig_cmd;
0119     int must_propagate = 0; /* Need to propagate MULTI/EXEC to AOF / slaves? */
0120 
0121     if (!(c->flags & REDIS_MULTI)) {
0122         addReplyError(c,"EXEC without MULTI");
0123         return;
0124     }
0125 
0126     /* Check if we need to abort the EXEC because:
0127      * 1) Some WATCHed key was touched.
0128      * 2) There was a previous error while queueing commands.
0129      * A failed EXEC in the first case returns a multi bulk nil object
0130      * (technically it is not an error but a special behavior), while
0131      * in the second an EXECABORT error is returned. */
0132     if (c->flags & (REDIS_DIRTY_CAS|REDIS_DIRTY_EXEC)) {
0133         addReply(c, c->flags & REDIS_DIRTY_EXEC ? shared.execaborterr :
0134                                                   shared.nullmultibulk);
0135         discardTransaction(c);
0136         goto handle_monitor;
0137     }
0138 
0139     /* Exec all the queued commands */
0140     unwatchAllKeys(c); /* Unwatch ASAP otherwise we'll waste CPU cycles */
0141     orig_argv = c->argv;
0142     orig_argc = c->argc;
0143     orig_cmd = c->cmd;
0144     addReplyMultiBulkLen(c,c->mstate.count);
0145     for (j = 0; j < c->mstate.count; j++) {
0146         c->argc = c->mstate.commands[j].argc;
0147         c->argv = c->mstate.commands[j].argv;
0148         c->cmd = c->mstate.commands[j].cmd;
0149 
0150         /* Propagate a MULTI request once we encounter the first write op.
0151          * This way we'll deliver the MULTI/..../EXEC block as a whole and
0152          * both the AOF and the replication link will have the same consistency
0153          * and atomicity guarantees. */
0154         if (!must_propagate && !(c->cmd->flags & REDIS_CMD_READONLY)) {
0155             execCommandPropagateMulti(c);
0156             must_propagate = 1;
0157         }
0158 
0159         call(c,REDIS_CALL_FULL);
0160 
0161         /* Commands may alter argc/argv, restore mstate. */
0162         c->mstate.commands[j].argc = c->argc;
0163         c->mstate.commands[j].argv = c->argv;
0164         c->mstate.commands[j].cmd = c->cmd;
0165     }
0166     c->argv = orig_argv;
0167     c->argc = orig_argc;
0168     c->cmd = orig_cmd;
0169     discardTransaction(c);
0170     /* Make sure the EXEC command will be propagated as well if MULTI
0171      * was already propagated. */
0172     if (must_propagate) server.dirty++;
0173 
0174 handle_monitor:
0175     /* Send EXEC to clients waiting data from MONITOR. We do it here
0176      * since the natural order of commands execution is actually:
0177      * MUTLI, EXEC, ... commands inside transaction ...
0178      * Instead EXEC is flagged as REDIS_CMD_SKIP_MONITOR in the command
0179      * table, and we do it here with correct ordering. */
0180     if (listLength(server.monitors) && !server.loading)
0181         replicationFeedMonitors(c,server.monitors,c->db->id,c->argv,c->argc);
0182 }
0183 
0184 /* ===================== WATCH (CAS alike for MULTI/EXEC) ===================
0185  *
0186  * The implementation uses a per-DB hash table mapping keys to list of clients
0187  * WATCHing those keys, so that given a key that is going to be modified
0188  * we can mark all the associated clients as dirty.
0189  *
0190  * Also every client contains a list of WATCHed keys so that's possible to
0191  * un-watch such keys when the client is freed or when UNWATCH is called. */
0192 
0193 /* In the client->watched_keys list we need to use watchedKey structures
0194  * as in order to identify a key in Redis we need both the key name and the
0195  * DB */
0196 typedef struct watchedKey {
0197     robj *key;
0198     redisDb *db;
0199 } watchedKey;
0200 
0201 /* Watch for the specified key */
0202 void watchForKey(redisClient *c, robj *key) {
0203     list *clients = NULL;
0204     listIter li;
0205     listNode *ln;
0206     watchedKey *wk;
0207 
0208     /* Check if we are already watching for this key */
0209     listRewind(c->watched_keys,&li);
0210     while((ln = listNext(&li))) {
0211         wk = listNodeValue(ln);
0212         if (wk->db == c->db && equalStringObjects(key,wk->key))
0213             return; /* Key already watched */
0214     }
0215     /* This key is not already watched in this DB. Let's add it */
0216     clients = dictFetchValue(c->db->watched_keys,key);
0217     if (!clients) { 
0218         clients = listCreate();
0219         dictAdd(c->db->watched_keys,key,clients);
0220         incrRefCount(key);
0221     }
0222     listAddNodeTail(clients,c);
0223     /* Add the new key to the list of keys watched by this client */
0224     wk = zmalloc(sizeof(*wk));
0225     wk->key = key;
0226     wk->db = c->db;
0227     incrRefCount(key);
0228     listAddNodeTail(c->watched_keys,wk);
0229 }
0230 
0231 /* Unwatch all the keys watched by this client. To clean the EXEC dirty
0232  * flag is up to the caller. */
0233 void unwatchAllKeys(redisClient *c) {
0234     listIter li;
0235     listNode *ln;
0236 
0237     if (listLength(c->watched_keys) == 0) return;
0238     listRewind(c->watched_keys,&li);
0239     while((ln = listNext(&li))) {
0240         list *clients;
0241         watchedKey *wk;
0242 
0243         /* Lookup the watched key -> clients list and remove the client
0244          * from the list */
0245         wk = listNodeValue(ln);
0246         clients = dictFetchValue(wk->db->watched_keys, wk->key);
0247         redisAssertWithInfo(c,NULL,clients != NULL);
0248         listDelNode(clients,listSearchKey(clients,c));
0249         /* Kill the entry at all if this was the only client */
0250         if (listLength(clients) == 0)
0251             dictDelete(wk->db->watched_keys, wk->key);
0252         /* Remove this watched key from the client->watched list */
0253         listDelNode(c->watched_keys,ln);
0254         decrRefCount(wk->key);
0255         zfree(wk);
0256     }
0257 }
0258 
0259 /* "Touch" a key, so that if this key is being WATCHed by some client the
0260  * next EXEC will fail. */
0261 void touchWatchedKey(redisDb *db, robj *key) {
0262     list *clients;
0263     listIter li;
0264     listNode *ln;
0265 
0266     if (dictSize(db->watched_keys) == 0) return;
0267     clients = dictFetchValue(db->watched_keys, key);
0268     if (!clients) return;
0269 
0270     /* Mark all the clients watching this key as REDIS_DIRTY_CAS */
0271     /* Check if we are already watching for this key */
0272     listRewind(clients,&li);
0273     while((ln = listNext(&li))) {
0274         redisClient *c = listNodeValue(ln);
0275 
0276         c->flags |= REDIS_DIRTY_CAS;
0277     }
0278 }
0279 
0280 /* On FLUSHDB or FLUSHALL all the watched keys that are present before the
0281  * flush but will be deleted as effect of the flushing operation should
0282  * be touched. "dbid" is the DB that's getting the flush. -1 if it is
0283  * a FLUSHALL operation (all the DBs flushed). */
0284 void touchWatchedKeysOnFlush(int dbid) {
0285     listIter li1, li2;
0286     listNode *ln;
0287 
0288     /* For every client, check all the waited keys */
0289     listRewind(server.clients,&li1);
0290     while((ln = listNext(&li1))) {
0291         redisClient *c = listNodeValue(ln);
0292         listRewind(c->watched_keys,&li2);
0293         while((ln = listNext(&li2))) {
0294             watchedKey *wk = listNodeValue(ln);
0295 
0296             /* For every watched key matching the specified DB, if the
0297              * key exists, mark the client as dirty, as the key will be
0298              * removed. */
0299             if (dbid == -1 || wk->db->id == dbid) {
0300                 if (dictFind(wk->db->dict, wk->key->ptr) != NULL)
0301                     c->flags |= REDIS_DIRTY_CAS;
0302             }
0303         }
0304     }
0305 }
0306 
0307 void watchCommand(redisClient *c) {
0308     int j;
0309 
0310     if (c->flags & REDIS_MULTI) {
0311         addReplyError(c,"WATCH inside MULTI is not allowed");
0312         return;
0313     }
0314     for (j = 1; j < c->argc; j++)
0315         watchForKey(c,c->argv[j]);
0316     addReply(c,shared.ok);
0317 }
0318 
0319 void unwatchCommand(redisClient *c) {
0320     unwatchAllKeys(c);
0321     c->flags &= (~REDIS_DIRTY_CAS);
0322     addReply(c,shared.ok);
0323 }