void closeTimedoutClients(void) {
    redisClient *c;
    listNode *ln;
    time_t now = time(NULL);
    listIter li;

    listRewind(server.clients,&li);
    while ((ln = listNext(&li)) != NULL) {
        c = listNodeValue(ln);
        if (server.maxidletime &&
            !(c->flags & REDIS_SLAVE) &&    /* no timeout for slaves */
            !(c->flags & REDIS_MASTER) &&   /* no timeout for masters */
            dictSize(c->pubsub_channels) == 0 && /* no timeout for pubsub */
            listLength(c->pubsub_patterns) == 0 &&
            (now - c->lastinteraction > server.maxidletime))
        {
            redisLog(REDIS_VERBOSE,"Closing idle client");
            freeClient(c);
        } else if (c->flags & REDIS_BLOCKED) {
            if (c->blockingto != 0 && c->blockingto < now) {
                addReply(c,shared.nullmultibulk);
                unblockClientWaitingData(c);
            }
        }
    }
}

/* Unblock a client that's waiting in a blocking operation such as BLPOP.
 * You should never call this function directly, but unblockClient() instead. */
void unblockClientWaitingData(redisClient *c) {
    dictEntry *de;
    dictIterator *di;
    list *l;

    redisAssertWithInfo(c,NULL,dictSize(c->bpop.keys) != 0);
    di = dictGetIterator(c->bpop.keys);
    /* The client may wait for multiple keys, so unblock it for every key. */
    while((de = dictNext(di)) != NULL) {
        robj *key = dictGetKey(de);

        /* Remove this client from the list of clients waiting for this key. */
        l = dictFetchValue(c->db->blocking_keys,key);
        redisAssertWithInfo(c,key,l != NULL);
        listDelNode(l,listSearchKey(l,c));
        /* If the list is empty we need to remove it to avoid wasting memory */
        if (listLength(l) == 0)
            dictDelete(c->db->blocking_keys,key);
    }
    dictReleaseIterator(di);

    /* Cleanup the client structure */
    dictEmpty(c->bpop.keys,NULL);
    if (c->bpop.target) {
        decrRefCount(c->bpop.target);
        c->bpop.target = NULL;
    }
}

//解阻塞一个正在阻塞中的client
void unblockClientWaitingData(client *c) {
    dictEntry *de;
    dictIterator *di;
    list *l;

    serverAssertWithInfo(c,NULL,dictSize(c->bpop.keys) != 0);
    //创建一个字典的迭代器，指向的是造成client阻塞的键所组成的字典
    di = dictGetIterator(c->bpop.keys);
    /* The client may wait for multiple keys, so unblock it for every key. */
    //因为client可能被多个key所阻塞，所以要遍历所有的键
    while((de = dictNext(di)) != NULL) {
        robj *key = dictGetKey(de); //获得key对象

        /* Remove this client from the list of clients waiting for this key. */
        //根据key找到对应的列表类型值，值保存着被阻塞的client，从中找c->db->blocking_keys中寻找
        l = dictFetchValue(c->db->blocking_keys,key);
        serverAssertWithInfo(c,key,l != NULL);
        // 将阻塞的client从列表中移除
        listDelNode(l,listSearchKey(l,c));
        /* If the list is empty we need to remove it to avoid wasting memory */
        //如果当前列表为空了，则从c->db->blocking_keys中将key删除
        if (listLength(l) == 0)
            dictDelete(c->db->blocking_keys,key);
    }
    dictReleaseIterator(di);    //释放迭代器

    /* Cleanup the client structure */
    //清空bpop.keys的所有节点
    dictEmpty(c->bpop.keys,NULL);
    //如果保存有新添加的元素，则应该释放
    if (c->bpop.target) {
        decrRefCount(c->bpop.target);
        c->bpop.target = NULL;
    }
}

void flushdbCommand(redisClient *c) {
    server.dirty += dictSize(c->db->dict);
    signalFlushedDb(c->db->id);
    dictEmpty(c->db->dict,NULL);
    dictEmpty(c->db->expires,NULL);
    addReply(c,shared.ok);
}

/* Subscribe a client to a channel. Returns 1 if the operation succeeded, or
 * 0 if the client was already subscribed to that channel. */
int pubsubSubscribeChannel(redisClient *c, robj *channel) {
    struct dictEntry *de;
    list *clients = NULL;
    int retval = 0;

    /* Add the channel to the client -> channels hash table */
    if (dictAdd(c->pubsub_channels,channel,NULL) == DICT_OK) {
        retval = 1;
        incrRefCount(channel);
        /* Add the client to the channel -> list of clients hash table */
        de = dictFind(server.pubsub_channels,channel);
        if (de == NULL) {
            clients = listCreate();
            dictAdd(server.pubsub_channels,channel,clients);
            incrRefCount(channel);
        } else {
            clients = dictGetEntryVal(de);
        }
        listAddNodeTail(clients,c);
    }
    /* Notify the client */
    addReply(c,shared.mbulk3);
    addReply(c,shared.subscribebulk);
    addReplyBulk(c,channel);
    addReplyLongLong(c,dictSize(c->pubsub_channels)+listLength(c->pubsub_patterns));
    return retval;
}

/* Unsubscribe a client from a channel. Returns 1 if the operation succeeded, or
 * 0 if the client was not subscribed to the specified channel. */
int pubsubUnsubscribePattern(redisClient *c, robj *pattern, int notify) {
    listNode *ln;
    pubsubPattern pat;
    int retval = 0;

    incrRefCount(pattern); /* Protect the object. May be the same we remove */
    if ((ln = listSearchKey(c->pubsub_patterns,pattern)) != NULL) {
        retval = 1;
        listDelNode(c->pubsub_patterns,ln);
        pat.client = c;
        pat.pattern = pattern;
        ln = listSearchKey(server.pubsub_patterns,&pat);
        listDelNode(server.pubsub_patterns,ln);
    }
    /* Notify the client */
    if (notify) {
        addReply(c,shared.mbulk3);
        addReply(c,shared.punsubscribebulk);
        addReplyBulk(c,pattern);
        addReplyLongLong(c,dictSize(c->pubsub_channels)+
                       listLength(c->pubsub_patterns));
    }
    decrRefCount(pattern);
    return retval;
}

/* Check if we need to resend our prediction of the counter to some nodes. */
void counterMaybeResend(counter *cntr) {
    clusterNode *node;
    int i;

    /* No acks to send? */
    if (cntr->want_acks == NULL || dictSize(cntr->want_acks) == 0) {
        return;
    }

    /* Too soon? */
    if (cntr->updated > mstime()-5000) {
        return;
    }

    /* Only check reachable nodes that have a valid link. */
    for (i = 0; i < server.cluster->reachable_nodes_count; i++) {
        node = server.cluster->reachable_nodes[i];

        if (node->link == NULL) continue;

        if (dictFind(cntr->want_acks, node->name) != NULL) {
            clusterSendShardToNode(cntr, node);
        }
    }
}

/*
 * “碰触”（touch）给定 key ，如果这个 key 正在被监视的话，
 * 让监视它的客户端在执行 EXEC 命令时失败。
 *
 * T = O(N)
 */
void touchWatchedKey(redisDb *db, robj *key)
{
    list *clients;
    listIter li;
    listNode *ln;

    // 如果数据库中没有任何 key 被监视，那么直接返回
    if (dictSize(db->watched_keys) == 0) return;

    // 取出数据库中所有监视给定 key 的客户端
    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 */
    // 打开所有监视这个 key 的客户端的 REDIS_DIRTY_CAS 状态
    // O(N)
    listRewind(clients,&li);
    while((ln = listNext(&li)))
    {
        redisClient *c = listNodeValue(ln);

        c->flags |= REDIS_DIRTY_CAS;
    }
}

int htNeedsResize(dict *dict) {
    long long size, used;

    size = dictSlots(dict);
    used = dictSize(dict);
    return (size && used && size > DICT_HT_INITIAL_SIZE &&
            (used*100/size < HASHTABLE_MIN_FILL));
}

int roDBDictSize(int id) {
    if (server.isBackgroundSaving != 0) {
        if (server.cowSaveDbExt[id].dictArray != NULL) {
            return server.cowSaveDbExt[id].dictArray->numele;
        }
    }
    return dictSize(server.db[id].dict);
}

void flushdbCommand(client *c) {
    server.dirty += dictSize(c->db->dict);
    signalFlushedDb(c->db->id);
    dictEmpty(c->db->dict,NULL);
    dictEmpty(c->db->expires,NULL);
    if (server.cluster_enabled) slotToKeyFlush();
    addReply(c,shared.ok);
}

void flushdbCommand(redisClient *c) {
    server.dirty += dictSize(c->db->dict);
    signalFlushedDb(c->db->id);
    dictEmpty(c->db->dict);
    dictEmpty(c->db->expires);
    if (server.ds_enabled) dsFlushDb(c->db->id);
    addReply(c,shared.ok);
}

/* scriptNameCommand() has compound sub-arguments, so it looks slightly more
 * convoluted than it actually is.  Just read each if/else branch as
 * if it were an individual command. */
void scriptNameCommand(redisClient *c) {
    char *req = c->argv[1]->ptr;
    sds script_name = c->argv[2]->ptr;

    if (c->argc == 4 && !strcasecmp(req, "set")) {
        sds target_sha = c->argv[3]->ptr;

        if (sdslen(target_sha) != 40 ||
            dictFind(server.lua_scripts,target_sha) == NULL) {
            addReply(c, g.err.nosha);
            return;
        }

        /* If name doesn't exist, dictReplace == dictAdd */
        dictReplace(g.names, script_name, target_sha);

        addReplyBulkCBuffer(c, script_name, sdslen(script_name));
    } else if (c->argc == 3 && !strcasecmp(req, "get")) {
        sds found;
        if ((found = dictFetchValue(g.names, script_name))) {
            addReplyBulkCBuffer(c, found, sdslen(found));
        } else {
            addReply(c, g.err.noname);
        }
    } else if (c->argc == 2 && !strcasecmp(req, "getall")) {
        dictIterator *di;
        dictEntry *de;

        unsigned long sz = dictSize(g.names);

        if (!sz) {
            addReply(c, shared.emptymultibulk);
            return;
        }

        /* Multiply by 2 because the size of the dict is the number of keys.
         * We are returning keys *and* values, so length is dictSize * 2 */
        addReplyMultiBulkLen(c, sz * 2);

        di = dictGetIterator(g.names);
        while ((de = dictNext(di))) {
            addReplyBulkCString(c, dictGetKey(de));
            addReplyBulkCString(c, dictGetVal(de));
        }
        dictReleaseIterator(di);
    } else if (c->argc == 3 && !strcasecmp(req, "del")) {
        sds deleted;

        if ((deleted = dictFetchValue(g.names, script_name))) {
            dictDelete(g.names, script_name);
            addReplyBulkCBuffer(c, deleted, sdslen(deleted));
        } else {
            addReply(c, g.err.noname);
        }
    } else {
        addReplyError(c, "Unknown scriptName subcommand or arg count");
    }
}

void rdb_save_triggles(rio *rdb)
{

    //save event 
    //db_num int int int int 
    //db 
    //scripts_num
    //key event lua_scripts 
    //key event lua_scripts 
    //.......
    dictIterator *di = NULL;
    dictEntry *de;   
    int i=0;
    for(i=0;i<server.dbnum;i++){
        int eventid=server.bridge_db.bridge_event[i]; 
        rioWrite(rdb,&eventid,4);
    }
    for(i=0;i<server.dbnum;i++)
    {
        dict *d = server.bridge_db.triggle_scipts[i];
        int mysize=dictSize(d);
        rioWrite(rdb,&mysize,4);
        if (dictSize(d) == 0) continue;
        di = dictGetSafeIterator(d);
        if (!di) {
            return ;
        }
        /* Iterate this DB writing every entry */
        while((de = dictNext(di)) != NULL) {
            sds keystr = dictGetKey(de);
            robj key;
            initStaticStringObject(key,keystr);
            if (rdbSaveStringObject(rdb,&key) == -1) return;

            struct bridge_db_triggle_t * tmptrg=dictGetVal(de);
            int event_id=tmptrg->event; 
            rioWrite(rdb,&event_id,4);
            int db_id=tmptrg->dbid; 
            rioWrite(rdb,&db_id,4);
            if (rdbSaveObjectType(rdb,tmptrg->lua_scripts) == -1) return ;
            if (rdbSaveObject(rdb,tmptrg->lua_scripts) == -1) return ; 
        }
    }
    if (di) dictReleaseIterator(di);
}

unsigned long setTypeSize(robj *subject) {
    if (subject->encoding == REDIS_ENCODING_HT) {
        return dictSize((dict*)subject->ptr);
    } else if (subject->encoding == REDIS_ENCODING_INTSET) {
        return intsetLen((intset*)subject->ptr);
    } else {
        redisPanic("Unknown set encoding");
    }
}

long long emptyDb(memoryDb *db, void (callback)(void*)) {
	int j;
	long long removed = 0;

	removed += dictSize(db.dict);
	dictEmpty(db.dict, callback);
	dictEmpty(db.expires, callback);
	return removed;
}

/* Subscribe a client to a channel. Returns 1 if the operation succeeded, or
 * 0 if the client was already subscribed to that channel. 
 *
 * 设置客户端 c 订阅频道 channel 。
 *
 * 订阅成功返回 1 ，如果客户端已经订阅了该频道，那么返回 0 。
 */
int pubsubSubscribeChannel(redisClient *c, robj *channel) {
    dictEntry *de;
    list *clients = NULL;
    int retval = 0;

    /* Add the channel to the client -> channels hash table */
    // 将 channels 填接到 c->pubsub_channels 的集合中（值为 NULL 的字典视为集合）
    if (dictAdd(c->pubsub_channels,channel,NULL) == DICT_OK) {
        retval = 1;
        incrRefCount(channel);

        // 关联示意图
        // {
        //  频道名        订阅频道的客户端
        //  'channel-a' : [c1, c2, c3],
        //  'channel-b' : [c5, c2, c1],
        //  'channel-c' : [c10, c2, c1]
        // }
        /* Add the client to the channel -> list of clients hash table */
        // 从 pubsub_channels 字典中取出保存着所有订阅了 channel 的客户端的链表
        // 如果 channel 不存在于字典，那么添加进去
        de = dictFind(server.pubsub_channels,channel);
        if (de == NULL) {
            clients = listCreate();
            dictAdd(server.pubsub_channels,channel,clients);
            incrRefCount(channel);
        } else {
            clients = dictGetVal(de);
        }

        // before:
        // 'channel' : [c1, c2]
        // after:
        // 'channel' : [c1, c2, c3]
        // 将客户端添加到链表的末尾
        listAddNodeTail(clients,c);
    }

    /* Notify the client */
    // 回复客户端。
    // 示例：
    // redis 127.0.0.1:6379> SUBSCRIBE xxx
    // Reading messages... (press Ctrl-C to quit)
    // 1) "subscribe"
    // 2) "xxx"
    // 3) (integer) 1
    addReply(c,shared.mbulkhdr[3]);
    // "subscribe\n" 字符串
    addReply(c,shared.subscribebulk);
    // 被订阅的客户端
    addReplyBulk(c,channel);
    // 客户端订阅的频道和模式总数
    addReplyLongLong(c,dictSize(c->pubsub_channels)+listLength(c->pubsub_patterns));

    return retval;
}

/* Delete a key, value, and associated expiration entry if any, from the DB */
int dbDelete(redisDb *db, robj *key) {
    /* Deleting an entry from the expires dict will not free the sds of
     * the key, because it is shared with the main dictionary. */
    if (dictSize(db->expires) > 0) dictDelete(db->expires,key->ptr);
    if (dictDelete(db->dict,key->ptr) == DICT_OK) {
        return 1;
    } else {
        return 0;
    }
}

/* COMMAND <subcommand> <args> */
void commandCommand(client *c) {
    dictIterator *di;
    dictEntry *de;

    if (c->argc == 1) {
        addReplyMultiBulkLen(c, dictSize(server.commands));
        di = dictGetIterator(server.commands);
        while ((de = dictNext(di)) != NULL) {
            addReplyCommand(c, dictGetVal(de));
        }
        dictReleaseIterator(di);
    } else if (!strcasecmp(c->argv[1]->ptr, "info")) {
        int i;
        addReplyMultiBulkLen(c, c->argc-2);
        for (i = 2; i < c->argc; i++) {
            addReplyCommand(c, dictFetchValue(server.commands, c->argv[i]->ptr));
        }
    } else if (!strcasecmp(c->argv[1]->ptr, "count") && c->argc == 2) {
        addReplyLongLong(c, dictSize(server.commands));
    } else if (!strcasecmp(c->argv[1]->ptr,"getkeys") && c->argc >= 3) {
        struct redisCommand *cmd = lookupCommand(c->argv[2]->ptr);
        int *keys, numkeys, j;

        if (!cmd) {
            addReplyErrorFormat(c,"Invalid command specified");
            return;
        } else if ((cmd->arity > 0 && cmd->arity != c->argc-2) ||
                   ((c->argc-2) < -cmd->arity))
        {
            addReplyError(c,"Invalid number of arguments specified for command");
            return;
        }

        keys = getKeysFromCommand(cmd,c->argv+2,c->argc-2,&numkeys);
        addReplyMultiBulkLen(c,numkeys);
        for (j = 0; j < numkeys; j++) addReplyBulk(c,c->argv[keys[j]+2]);
        getKeysFreeResult(keys);
    } else {
        addReplyError(c, "Unknown subcommand or wrong number of arguments.");
        return;
    }
}

long long emptyDb() {
    int j;
    long long removed = 0;

    for (j = 0; j < server.dbnum; j++) {
        removed += dictSize(server.db[j].dict);
        dictEmpty(server.db[j].dict);
        dictEmpty(server.db[j].expires);
    }
    return removed;
}