/*

* Copyright (c) 2009-2012, Salvatore Sanfilippo <antirez at gmail dot com>

* All rights reserved.

*

* Redistribution and use in source and binary forms, with or without

* modification, are permitted provided that the following conditions are met:

*

* * Redistributions of source code must retain the above copyright notice,

* this list of conditions and the following disclaimer.

* * Redistributions in binary form must reproduce the above copyright

* notice, this list of conditions and the following disclaimer in the

* documentation and/or other materials provided with the distribution.

* * Neither the name of Redis nor the names of its contributors may be used

* to endorse or promote products derived from this software without

* specific prior written permission.

*

* THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"

* AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE

* IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE

* ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE

* LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR

* CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF

* SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS

* INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN

* CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)

* ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE

* POSSIBILITY OF SUCH DAMAGE.

*/

#include "redis.h"

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

void initClientMultiState(redisClient *c) /* 初始化客户端操作 */

void freeClientMultiState(redisClient *c) /* 释放客户端所有与multi/exec相关的资源 */

void queueMultiCommand(redisClient *c) /* 客户端的multi命令队列添加一条新的命令 */

void discardTransaction(redisClient *c) /* 撤销事务操作 */

void flagTransaction(redisClient *c) /* 标记一个事物为DIRTY_EXEC状态，最后这个事物会执行失败，，此方法调用于插入命令的时候 */

void multiCommand(redisClient *c) /* 加入multi命令 */

void discardCommand(redisClient *c) /* 撤销命令 */

void execCommandPropagateMulti(redisClient *c) /* 发送multi命令给所有的从客户端和aof文件 */

void execCommand(redisClient *c) /* 客户单执行Command命令 */

void watchForKey(redisClient *c, robj *key) /* 为客户端添加key监听 */

void unwatchAllKeys(redisClient *c) /* 客户端移除所有的key */

void touchWatchedKey(redisDb *db, robj *key) /* touch key的意思，表示key正在被监听，下一条执行操作将会失败 */

void touchWatchedKeysOnFlush(int dbid) /* 根据key所在的的db，把此db下的watched-key统统touch一遍 */

void watchCommand(redisClient *c) /* watch key 的命令方法，通过client中的参数传值 */

void unwatchCommand(redisClient *c) /* 取消监听key的命令方法 */

/* 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 */

/* 释放客户端所有与multi/exec相关的资源 */

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 */

/* 客户端的multi命令队列添加一条新的命令 */

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 discardTransaction(redisClient *c) {

freeClientMultiState(c);

initClientMultiState(c);

c->flags &= ~(REDIS_MULTI|REDIS_DIRTY_CAS|REDIS_DIRTY_EXEC);

//客户端取消监听所有的key

unwatchAllKeys(c);

}

/* Flag the transacation as DIRTY_EXEC so that EXEC will fail.

* Should be called every time there is an error while queueing a command. */

/* 标记一个事物为DIRTY_EXEC状态，最后这个事物会执行失败，，此方法调用于插入命令的时候 */

void flagTransaction(redisClient *c) {

if (c->flags & REDIS_MULTI)

c->flags |= REDIS_DIRTY_EXEC;

}

/* 加入multi命令 */

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;

}

//撤销事务，回应客户端操作成功

discardTransaction(c);

addReply(c,shared.ok);

}

/* Send a MULTI command to all the slaves and AOF file. Check the execCommand

* implementation for more information. */

/* 发送multi命令给所有的从客户端和aof文件 */

void execCommandPropagateMulti(redisClient *c) {

//构件命令字符串

robj *multistring = createStringObject("MULTI",5);

//服务端发送命令

propagate(server.multiCommand,c->db->id,&multistring,1,

REDIS_PROPAGATE_AOF|REDIS_PROPAGATE_REPL);

decrRefCount(multistring);

}

/* 客户单执行Command命令 */

void execCommand(redisClient *c) {

int j;

robj **orig_argv;

int orig_argc;

struct redisCommand *orig_cmd;

//判断是执行aaof的还是slave

int must_propagate = 0; /* Need to propagate MULTI/EXEC to AOF / slaves? */

if (!(c->flags & REDIS_MULTI)) {

addReplyError(c,"EXEC without MULTI");

return;

}

/* Check if we need to abort the EXEC because:

* 1) Some WATCHed key was touched.

* 2) There was a previous error while queueing commands.

* A failed EXEC in the first case returns a multi bulk nil object

* (technically it is not an error but a special behavior), while

* in the second an EXECABORT error is returned. */

/* 在执行exec命令之前检测是否已经有异常情况发生，比如在queuing commands 时已经出错了 */

if (c->flags & (REDIS_DIRTY_CAS|REDIS_DIRTY_EXEC)) {

addReply(c, c->flags & REDIS_DIRTY_EXEC ? shared.execaborterr :

shared.nullmultibulk);

//取消事务

discardTransaction(c);

//goto语句处理后续操作

goto handle_monitor;

}

/* Exec all the queued commands */

//取消key监听，否则会消耗大量的CPU

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;

/* Propagate a MULTI request once we encounter the first write op.

* 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. */

if (!must_propagate && !(c->cmd->flags & REDIS_CMD_READONLY)) {

//下面的方法就是执行命令的方法

execCommandPropagateMulti(c);

must_propagate = 1;

}

call(c,REDIS_CALL_FULL);

/* 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;

discardTransaction(c);

/* Make sure the EXEC command will be propagated as well if MULTI

* was already propagated. */

if (must_propagate) server.dirty++;

handle_monitor:

/* Send EXEC to clients waiting data from MONITOR. We do it here

* since the natural order of commands execution is actually:

* MUTLI, EXEC, ... commands inside transaction ...

* Instead EXEC is flagged as REDIS_CMD_SKIP_MONITOR in the command

* table, and we do it here with correct ordering. */

/* 当Command异常时，使其跳过MONITOR操作 */

if (listLength(server.monitors) && !server.loading)

replicationFeedMonitors(c,server.monitors,c->db->id,c->argv,c->argc);

}

/* ===================== 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 */

/* 在事务处理中，存在2种mapping映射，key-->client lists ,表示所有列表中的Client都在监听这个key

，当这个key的value发生改变了，可以标记这些Client为DIRTY状态，需要更新了，同时在Client内部也会维护

一个key of list,表示一个客户端所监视的所有key，当Client发生free操作等，就要把key里面维护的Client列表

做更新*/

/* 定义了watchedKey结构体 */

typedef struct watchedKey {

robj *key;

redisDb *db;

} watchedKey;

/* Watch for the specified key */

/* 为客户端添加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))) {

//通过变量c->watched_keys，判断是否已经存在watched_key，存在，就直接返回

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();

//往字典中的db->watched_keys里添加key

dictAdd(c->db->watched_keys,key,clients);

incrRefCount(key);

}

listAddNodeTail(clients,c);

/* Add the new key to the list of keys watched by this client */

wk = zmalloc(sizeof(*wk));

wk->key = key;

wk->db = c->db;

incrRefCount(key);

//在客户端的watched_keys里面添加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. */

/* 客户端移除所有的key */

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. */

/* touch key的意思，表示key正在被监听，下一条执行操作将会失败 */

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);

//遍历该key拥有的Client，把flag标记为DIRTY_CAS状态

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). */

/* 根据key所在的的db，把此db下的watched-key统统touch一遍 */

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;

}

}

}

}

/* watch key 的命令方法，通过client中的参数传值 */

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);

}

/* 取消监听key的命令方法 */

void unwatchCommand(redisClient *c) {

unwatchAllKeys(c);

c->flags &= (~REDIS_DIRTY_CAS);

addReply(c,shared.ok);

}