src/multi.c

   1 /*
   2  * Copyright (c) 2009-2012, Salvatore Sanfilippo <antirez at gmail dot com>
   3  * All rights reserved.
   4  *
   5  * Redistribution and use in source and binary forms, with or without
   6  * modification, are permitted provided that the following conditions are met:
   7  *
   8  *   * Redistributions of source code must retain the above copyright notice,
   9  *     this list of conditions and the following disclaimer.
  10  *   * Redistributions in binary form must reproduce the above copyright
  11  *     notice, this list of conditions and the following disclaimer in the
  12  *     documentation and/or other materials provided with the distribution.
  13  *   * Neither the name of Redis nor the names of its contributors may be used
  14  *     to endorse or promote products derived from this software without
  15  *     specific prior written permission.
  16  *
  17  * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
  18  * AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
  19  * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
  20  * ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE
  21  * LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
  22  * CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
  23  * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
  24  * INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
  25  * CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
  26  * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
  27  * POSSIBILITY OF SUCH DAMAGE.
  28  */
  29
  30 #include "redis.h"
  31
  32 /* ================================ MULTI/EXEC ============================== */
  33
  34 /* Client state initialization for MULTI/EXEC */
  35 void initClientMultiState(redisClient *c) {
  36     c->mstate.commands = NULL;
  37     c->mstate.count = 0;
  38 }
  39
  40 /* Release all the resources associated with MULTI/EXEC state */
  41 void freeClientMultiState(redisClient *c) {
  42     int j;
  43
  44     for (j = 0; j < c->mstate.count; j++) {
  45         int i;
  46         multiCmd *mc = c->mstate.commands+j;
  47
  48         for (i = 0; i < mc->argc; i++)
  49             decrRefCount(mc->argv[i]);
  50         zfree(mc->argv);
  51     }
  52     zfree(c->mstate.commands);
  53 }
  54
  55 /* Add a new command into the MULTI commands queue */
  56 void queueMultiCommand(redisClient *c) {
  57     multiCmd *mc;
  58     int j;
  59
  60     c->mstate.commands = zrealloc(c->mstate.commands,
  61             sizeof(multiCmd)*(c->mstate.count+1));
  62     mc = c->mstate.commands+c->mstate.count;
  63     mc->cmd = c->cmd;
  64     mc->argc = c->argc;
  65     mc->argv = zmalloc(sizeof(robj*)*c->argc);
  66     memcpy(mc->argv,c->argv,sizeof(robj*)*c->argc);
  67     for (j = 0; j < c->argc; j++)
  68         incrRefCount(mc->argv[j]);
  69     c->mstate.count++;
  70 }
  71
  72 void discardTransaction(redisClient *c) {
  73     freeClientMultiState(c);
  74     initClientMultiState(c);
  75     c->flags &= ~(REDIS_MULTI|REDIS_DIRTY_CAS);;
  76     unwatchAllKeys(c);
  77 }
  78
  79 void multiCommand(redisClient *c) {
  80     if (c->flags & REDIS_MULTI) {
  81         addReplyError(c,"MULTI calls can not be nested");
  82         return;
  83     }
  84     c->flags |= REDIS_MULTI;
  85     addReply(c,shared.ok);
  86 }
  87
  88 void discardCommand(redisClient *c) {
  89     if (!(c->flags & REDIS_MULTI)) {
  90         addReplyError(c,"DISCARD without MULTI");
  91         return;
  92     }
  93     discardTransaction(c);
  94     addReply(c,shared.ok);
  95 }
  96
  97 /* Send a MULTI command to all the slaves and AOF file. Check the execCommand
  98  * implememntation for more information. */
  99 void execCommandReplicateMulti(redisClient *c) {
 100     robj *multistring = createStringObject("MULTI",5);
 101
 102     if (server.aof_state != REDIS_AOF_OFF)
 103         feedAppendOnlyFile(server.multiCommand,c->db->id,&multistring,1);
 104     if (listLength(server.slaves))
 105         replicationFeedSlaves(server.slaves,c->db->id,&multistring,1);
 106     decrRefCount(multistring);
 107 }
 108
 109 void execCommand(redisClient *c) {
 110     int j;
 111     robj **orig_argv;
 112     int orig_argc;
 113     struct redisCommand *orig_cmd;
 114
 115     if (!(c->flags & REDIS_MULTI)) {
 116         addReplyError(c,"EXEC without MULTI");
 117         return;
 118     }
 119
 120     /* Check if we need to abort the EXEC if some WATCHed key was touched.
 121      * A failed EXEC will return a multi bulk nil object. */
 122     if (c->flags & REDIS_DIRTY_CAS) {
 123         freeClientMultiState(c);
 124         initClientMultiState(c);
 125         c->flags &= ~(REDIS_MULTI|REDIS_DIRTY_CAS);
 126         unwatchAllKeys(c);
 127         addReply(c,shared.nullmultibulk);
 128         goto handle_monitor;
 129     }
 130
 131     /* Replicate a MULTI request now that we are sure the block is executed.
 132      * This way we'll deliver the MULTI/..../EXEC block as a whole and
 133      * both the AOF and the replication link will have the same consistency
 134      * and atomicity guarantees. */
 135     execCommandReplicateMulti(c);
 136
 137     /* Exec all the queued commands */
 138     unwatchAllKeys(c); /* Unwatch ASAP otherwise we'll waste CPU cycles */
 139     orig_argv = c->argv;
 140     orig_argc = c->argc;
 141     orig_cmd = c->cmd;
 142     addReplyMultiBulkLen(c,c->mstate.count);
 143     for (j = 0; j < c->mstate.count; j++) {
 144         c->argc = c->mstate.commands[j].argc;
 145         c->argv = c->mstate.commands[j].argv;
 146         c->cmd = c->mstate.commands[j].cmd;
 147         call(c,REDIS_CALL_FULL);
 148
 149         /* Commands may alter argc/argv, restore mstate. */
 150         c->mstate.commands[j].argc = c->argc;
 151         c->mstate.commands[j].argv = c->argv;
 152         c->mstate.commands[j].cmd = c->cmd;
 153     }
 154     c->argv = orig_argv;
 155     c->argc = orig_argc;
 156     c->cmd = orig_cmd;
 157     freeClientMultiState(c);
 158     initClientMultiState(c);
 159     c->flags &= ~(REDIS_MULTI|REDIS_DIRTY_CAS);
 160     /* Make sure the EXEC command is always replicated / AOF, since we
 161      * always send the MULTI command (we can't know beforehand if the
 162      * next operations will contain at least a modification to the DB). */
 163     server.dirty++;
 164
 165 handle_monitor:
 166     /* Send EXEC to clients waiting data from MONITOR. We do it here
 167      * since the natural order of commands execution is actually:
 168      * MUTLI, EXEC, ... commands inside transaction ...
 169      * Instead EXEC is flagged as REDIS_CMD_SKIP_MONITOR in the command
 170      * table, and we do it here with correct ordering. */
 171     if (listLength(server.monitors) && !server.loading)
 172         replicationFeedMonitors(c,server.monitors,c->db->id,c->argv,c->argc);
 173 }
 174
 175 /* ===================== WATCH (CAS alike for MULTI/EXEC) ===================
 176  *
 177  * The implementation uses a per-DB hash table mapping keys to list of clients
 178  * WATCHing those keys, so that given a key that is going to be modified
 179  * we can mark all the associated clients as dirty.
 180  *
 181  * Also every client contains a list of WATCHed keys so that's possible to
 182  * un-watch such keys when the client is freed or when UNWATCH is called. */
 183
 184 /* In the client->watched_keys list we need to use watchedKey structures
 185  * as in order to identify a key in Redis we need both the key name and the
 186  * DB */
 187 typedef struct watchedKey {
 188     robj *key;
 189     redisDb *db;
 190 } watchedKey;
 191
 192 /* Watch for the specified key */
 193 void watchForKey(redisClient *c, robj *key) {
 194     list *clients = NULL;
 195     listIter li;
 196     listNode *ln;
 197     watchedKey *wk;
 198
 199     /* Check if we are already watching for this key */
 200     listRewind(c->watched_keys,&li);
 201     while((ln = listNext(&li))) {
 202         wk = listNodeValue(ln);
 203         if (wk->db == c->db && equalStringObjects(key,wk->key))
 204             return; /* Key already watched */
 205     }
 206     /* This key is not already watched in this DB. Let's add it */
 207     clients = dictFetchValue(c->db->watched_keys,key);
 208     if (!clients) {
 209         clients = listCreate();
 210         dictAdd(c->db->watched_keys,key,clients);
 211         incrRefCount(key);
 212     }
 213     listAddNodeTail(clients,c);
 214     /* Add the new key to the lits of keys watched by this client */
 215     wk = zmalloc(sizeof(*wk));
 216     wk->key = key;
 217     wk->db = c->db;
 218     incrRefCount(key);
 219     listAddNodeTail(c->watched_keys,wk);
 220 }
 221
 222 /* Unwatch all the keys watched by this client. To clean the EXEC dirty
 223  * flag is up to the caller. */
 224 void unwatchAllKeys(redisClient *c) {
 225     listIter li;
 226     listNode *ln;
 227
 228     if (listLength(c->watched_keys) == 0) return;
 229     listRewind(c->watched_keys,&li);
 230     while((ln = listNext(&li))) {
 231         list *clients;
 232         watchedKey *wk;
 233
 234         /* Lookup the watched key -> clients list and remove the client
 235          * from the list */
 236         wk = listNodeValue(ln);
 237         clients = dictFetchValue(wk->db->watched_keys, wk->key);
 238         redisAssertWithInfo(c,NULL,clients != NULL);
 239         listDelNode(clients,listSearchKey(clients,c));
 240         /* Kill the entry at all if this was the only client */
 241         if (listLength(clients) == 0)
 242             dictDelete(wk->db->watched_keys, wk->key);
 243         /* Remove this watched key from the client->watched list */
 244         listDelNode(c->watched_keys,ln);
 245         decrRefCount(wk->key);
 246         zfree(wk);
 247     }
 248 }
 249
 250 /* "Touch" a key, so that if this key is being WATCHed by some client the
 251  * next EXEC will fail. */
 252 void touchWatchedKey(redisDb *db, robj *key) {
 253     list *clients;
 254     listIter li;
 255     listNode *ln;
 256
 257     if (dictSize(db->watched_keys) == 0) return;
 258     clients = dictFetchValue(db->watched_keys, key);
 259     if (!clients) return;
 260
 261     /* Mark all the clients watching this key as REDIS_DIRTY_CAS */
 262     /* Check if we are already watching for this key */
 263     listRewind(clients,&li);
 264     while((ln = listNext(&li))) {
 265         redisClient *c = listNodeValue(ln);
 266
 267         c->flags |= REDIS_DIRTY_CAS;
 268     }
 269 }
 270
 271 /* On FLUSHDB or FLUSHALL all the watched keys that are present before the
 272  * flush but will be deleted as effect of the flushing operation should
 273  * be touched. "dbid" is the DB that's getting the flush. -1 if it is
 274  * a FLUSHALL operation (all the DBs flushed). */
 275 void touchWatchedKeysOnFlush(int dbid) {
 276     listIter li1, li2;
 277     listNode *ln;
 278
 279     /* For every client, check all the waited keys */
 280     listRewind(server.clients,&li1);
 281     while((ln = listNext(&li1))) {
 282         redisClient *c = listNodeValue(ln);
 283         listRewind(c->watched_keys,&li2);
 284         while((ln = listNext(&li2))) {
 285             watchedKey *wk = listNodeValue(ln);
 286
 287             /* For every watched key matching the specified DB, if the
 288              * key exists, mark the client as dirty, as the key will be
 289              * removed. */
 290             if (dbid == -1 || wk->db->id == dbid) {
 291                 if (dictFind(wk->db->dict, wk->key->ptr) != NULL)
 292                     c->flags |= REDIS_DIRTY_CAS;
 293             }
 294         }
 295     }
 296 }
 297
 298 void watchCommand(redisClient *c) {
 299     int j;
 300
 301     if (c->flags & REDIS_MULTI) {
 302         addReplyError(c,"WATCH inside MULTI is not allowed");
 303         return;
 304     }
 305     for (j = 1; j < c->argc; j++)
 306         watchForKey(c,c->argv[j]);
 307     addReply(c,shared.ok);
 308 }
 309
 310 void unwatchCommand(redisClient *c) {
 311     unwatchAllKeys(c);
 312     c->flags &= (~REDIS_DIRTY_CAS);
 313     addReply(c,shared.ok);
 314 }