static int createAllCFSfromConfig()
{
	dictEntry *de = NULL;
	dictIterator *di = NULL;

	if(!g_dataModel.cfsHashTable)
		return -1;

	di = dictGetIterator(g_cfmDict);
	if(!di){
		return -1;
	}

	de = dictNext(di);
	while(de){
		columnFamilyMetadata *cfmd = (columnFamilyMetadata *)
							dictGetEntryKey(de);
		columnFamilyStore *cfs = NULL;

		if(!cfmd || !(cfs = getCFStore(cfmd))){
			dictReleaseIterator(di);
			return -1;
		}

		if(insertDMCFS(cfs) < 0){
			dictReleaseIterator(di);
			freeHeapCFStore(cfs);
			return -1;
		}
		de = dictNext(di);
	}
	dictReleaseIterator(di);
	return 0;
}

void 
callback_recurse(Engine *engine, dict *term_index, dict *query_nodes, void *doc, bool positive) {
	assert(doc);
  if(positive) {
    QueryNode *n = dictFetchValue(query_nodes, &SENTINEL);
    if (n) {
      DEBUG("recursing SENTINEL");
      callback_recurse(engine, term_index, n->children, doc, false);
    }
  }
	if(dictSize(term_index) > dictSize(query_nodes) || !positive){
		dictIterator *iterator = dictGetIterator(query_nodes);
		dictEntry *next;
	
    while ((next = dictNext(iterator))) {
      if (positive == !!dictFetchValue(term_index, dictGetKey(next))) {
		  	QueryNode *node = dictGetVal(next);
  			Callback *cb = node->callback;
        assert(node->term->owns_string);
        DEBUG("perc'd %.*s:%.*s", node->term->field.len, node->term->field.val, node->term->text.len, node->term->text.val);
  			while(cb) {
  			  DEBUG("running callback %d", cb->id);
  				cb->handler(cb, doc);
  				cb = cb->next;
  			}
  			if(node->children) {
  				callback_recurse(engine, term_index, node->children, doc, positive);
  			}
  		}
		}
    dictReleaseIterator(iterator);
	} else {
		dictIterator *iterator = dictGetIterator(term_index);
		dictEntry *next;
		QueryNode *node;

    while ((next = dictNext(iterator))) {
      if(positive == !!(node = dictFetchValue(query_nodes, dictGetKey(next)))) {
  			Callback *cb = node->callback;
  			DEBUG("perc'd %.*s:%.*s", node->term->field.len, node->term->field.val, node->term->text.len, node->term->text.val);
  			while(cb) {
  			  DEBUG("running callback %d", cb->id);
  				cb->handler(cb, doc);
  				cb = cb->next;
  			}
  			if(node->children){
  				callback_recurse(engine, term_index, node->children, doc,positive);
  			}
  		}
		}
		dictReleaseIterator(iterator);
	}
}

/* Helper function to free the context. */
static void __redisAsyncFree(redisAsyncContext *ac) {
    redisContext *c = &(ac->c);
    redisCallback cb;
    dictIterator *it;
    dictEntry *de;

    // 执行所有等待队列中回调函数，将空回复传进去
    /* Execute pending callbacks with NULL reply. */
    while (__redisShiftCallback(&ac->replies,&cb) == REDIS_OK)
        __redisRunCallback(ac,&cb,NULL);

    // 执行所有等待队列中回调函数，将空回复传进去
    // 这里是无效命令的回调函数
    /* Execute callbacks for invalid commands */
    while (__redisShiftCallback(&ac->sub.invalid,&cb) == REDIS_OK)
        __redisRunCallback(ac,&cb,NULL);

    // 执行所有等待队列中回调函数，将空回复传进去
    // 这里是订阅发布的回调函数
    /* Run subscription callbacks callbacks with NULL reply */
    it = dictGetIterator(ac->sub.channels);
    while ((de = dictNext(it)) != NULL)
        __redisRunCallback(ac,dictGetEntryVal(de),NULL);
    dictReleaseIterator(it);
    dictRelease(ac->sub.channels);

    // 执行所有等待队列中回调函数，将空回复传进去
    // 这里是订阅发布的回调函数
    it = dictGetIterator(ac->sub.patterns);
    while ((de = dictNext(it)) != NULL)
        __redisRunCallback(ac,dictGetEntryVal(de),NULL);
    dictReleaseIterator(it);
    dictRelease(ac->sub.patterns);

    // 注销连接上的读写事件
    /* Signal event lib to clean up */
    _EL_CLEANUP(ac);

    // 执行连接关闭回调函数，可做一些清理工作
    /* Execute disconnect callback. When redisAsyncFree() initiated destroying
     * this context, the status will always be REDIS_OK. */
    if (ac->onDisconnect && (c->flags & REDIS_CONNECTED)) {
        if (c->flags & REDIS_FREEING) {
            ac->onDisconnect(ac,REDIS_OK);
        } else {
            ac->onDisconnect(ac,(ac->err == 0) ? REDIS_OK : REDIS_ERR);
        }
    }

    // 释放 redisContext 中的内存
    /* Cleanup self */
    redisFree(c);
}

static void
conf_organizations_dump(rmt_conf *cf)
{
    dict *orgs, *org;
    dictIterator *di;
    dictEntry *de;
    sds name;
    int log_level = LOG_VERB;
    
    if(cf == NULL){
        return;
    }

    orgs = cf->organizations;
    if(orgs == NULL){
        log_debug(log_level, "organization is NULL");
        return;
    }
    
    di = dictGetIterator(orgs);

    while((de = dictNext(di)) != NULL){
        name = dictGetKey(de);
        org = dictGetVal(de);

        conf_organization_dump(name, org, log_level);
        log_debug(log_level, "");
    }

    dictReleaseIterator(di);
}

void countersSync(clusterNode *node) {
    dictIterator *it;
    dictEntry *de;

    it = dictGetIterator(server.counters);
    while ((de = dictNext(it)) != NULL) {
        counter *cntr;
        listNode *ln;
        listIter li;
        shard *shrd;

        cntr = dictGetVal(de);

        /* If we have our own shard make sure to send it's prediction
         * to the new node. */
        if (cntr->myshard) {
            clusterSendShardToNode(cntr, node);
        }

        listRewind(cntr->shards,&li);
        while ((ln = listNext(&li)) != NULL) {
            shrd = listNodeValue(ln);

            if (memcmp(shrd->node_name, node->name, CLUSTER_NAMELEN) == 0) {
                shrd->node = node;
            }
        }
    }
    dictReleaseIterator(it);
}

void setTypeReleaseIterator(setTypeIterator* si)
{
    if (si->encoding == REDIS_ENCODING_HT) {
        dictReleaseIterator(si->di);
    }
    zfree(si);
}

void call_expire_delete_event(void *pdb,void *pkeyobj)
{
    redisDb *db=(redisDb *)pdb;
    robj *myobj=(robj *)pkeyobj;
    robj *keyobj = createStringObject(myobj->ptr,sdslen(myobj->ptr));

    struct dictIterator *iter=dictGetIterator(server.bridge_db.triggle_scipts[db->id]);
    dictEntry *trigs;
    do{
        trigs=dictNext(iter);
        if(trigs!=NULL)
        {
            struct bridge_db_triggle_t * tmptrg=dictGetVal(trigs);
            //add func str check for the function only the key satisfy the funcname:XXXX can call the event 
            if(tmptrg->event==DELETE_EXPIRE&&strncmp(keyobj->ptr,dictGetKey(trigs),sdslen(dictGetKey(trigs)))==0){ //找到指定的类型事件
                redisLog(REDIS_NOTICE,"triggle_event:%d,%s",DELETE_EXPIRE,(char *)dictGetKey(trigs));
                triggle_expire_event(db,dictGetKey(trigs),keyobj);
            }
        }
    }while(trigs!=NULL);
    dictReleaseIterator(iter);

    decrRefCount(keyobj);  

}

/* Unsubscribe from all the channels. Return the number of channels the
 * client was subscribed from. 
 *
 * 退订客户端 c 订阅的所有频道。
 *
 * 返回被退订频道的总数。
 */
int pubsubUnsubscribeAllChannels(redisClient *c, int notify) {

    // 频道迭代器
    dictIterator *di = dictGetSafeIterator(c->pubsub_channels);
    dictEntry *de;
    int count = 0;

    // 退订
    while((de = dictNext(di)) != NULL) {
        robj *channel = dictGetKey(de);

        count += pubsubUnsubscribeChannel(c,channel,notify);
    }

    /* We were subscribed to nothing? Still reply to the client. */
    // 如果在执行这个函数时，客户端没有订阅任何频道，
    // 那么向客户端发送回复
    if (notify && count == 0) {
        addReply(c,shared.mbulkhdr[3]);
        addReply(c,shared.unsubscribebulk);
        addReply(c,shared.nullbulk);
        addReplyLongLong(c,dictSize(c->pubsub_channels)+
                       listLength(c->pubsub_patterns));
    }

    dictReleaseIterator(di);

    // 被退订的频道的数量
    return count;
}

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

		redisDb *db = &(c->db)[keyHashSlot(key->ptr, sdslen(key->ptr))];
        /* Remove this client from the list of clients waiting for this key. */
        l = dictFetchValue(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(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;
    }
}

void keysCommand(redisClient *c) {
    dictIterator *di;
    dictEntry *de;
    sds pattern = c->argv[1]->ptr;
    int plen = sdslen(pattern), allkeys;
    unsigned long numkeys = 0;
    void *replylen = addDeferredMultiBulkLength(c);

    di = dictGetSafeIterator(c->db->dict);
    allkeys = (pattern[0] == '*' && pattern[1] == '\0');
    while((de = dictNext(di)) != NULL) {
        sds key = dictGetKey(de);
        robj *keyobj;

        if (allkeys || stringmatchlen(pattern,plen,key,sdslen(key),0)) {
            keyobj = createStringObject(key,sdslen(key));
            if (expireIfNeeded(c->db,keyobj) == 0) {
                addReplyBulk(c,keyobj);
                numkeys++;
            }
            decrRefCount(keyobj);
        }
    }
    dictReleaseIterator(di);
    setDeferredMultiBulkLength(c,replylen,numkeys);
}

//解阻塞一个正在阻塞中的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;
    }
}

static void
conf_organization_dump(sds name, dict *org, int log_level)
{
    dictIterator *di;
    dictEntry *de;
    sds key;
    conf_value *cv;

    if(name == NULL || org == NULL){
        return;
    }

    log_debug(log_level, "[%.*s]", sdslen(name), name);
    
    di = dictGetIterator(org);

    while((de = dictNext(di)) != NULL){
        key = dictGetKey(de);
        cv = dictGetVal(de);

        if(cv->type == CONF_VALUE_STRING){
            log_debug(log_level, "%.*s: %.*s", 
                sdslen(key), key,
                sdslen(cv->value), cv->value);
        }else if(cv->type == CONF_VALUE_ARRAY){
            log_debug(log_level, "%.*s:",sdslen(key), key);
            conf_value_dump(cv, log_level);
        }else{
            NOT_REACHED();
        }
    }

    dictReleaseIterator(di);
}

void delkeysCommand(redisClient *c) {
    dictIterator *di;
    dictEntry *de;
    sds pattern = c->argv[1]->ptr;
    int plen = sdslen(pattern), allkeys;
    unsigned long deleted = 0;

    di = dictGetIterator(c->db->dict);
    allkeys = (pattern[0] == '*' && pattern[1] == '\0');
    while((de = dictNext(di)) != NULL) {
        sds key = dictGetEntryKey(de);
        robj *keyobj;

        if (allkeys || stringmatchlen(pattern,plen,key,sdslen(key),0)) {
            keyobj = createStringObject(key,sdslen(key));
            if (dbDelete(c->db,keyobj)) {
                touchWatchedKey(c->db,keyobj);
                server.dirty++;
                deleted++;
            }
            decrRefCount(keyobj);
        }
    }
    dictReleaseIterator(di);
    addReplyLongLong(c,deleted);
}

/* copy a dictionary of redis objects
   Assumes copied directory uses COW destructors */
dict *copyonwrite_dictobj(dict *curdict, bkgdDbExt *extDict) {
    dict *newdict;
    dictIterator * di;
    dictEntry *de;

    /* checks if copy needed. else return curdict */
    if (server.isBackgroundSaving == 0 || server.cowDictCopied == NULL) {
        return curdict;
    }

    /* create copy */
    newdict = dictCreate(curdict->type, curdict->privdata);
    if (newdict != NULL) {
        /* copy all entries without refcounting or copying values */
        /* can't just memcpy the whole dictionary because entries are allocated */
        di = dictGetSafeIterator(curdict);
        while((de = dictNext(di)) != NULL) {
            dictAdd(newdict, de->key, de->val);
        }
        dictReleaseIterator(di);

        if (extDict != NULL) {
            /* fix types to not delete while saving */
            extDict->savedType = newdict->type;
            newdict->type = extDict->cowType;
            curdict->type = extDict->readonlyType;
        }
    }

    return newdict;
}

void hashTypeReleaseIterator(hashTypeIterator *hi) {
    if (hi->encoding == REDIS_ENCODING_HT) {
        dictReleaseIterator(hi->di);
    }

    zfree(hi);
}

void roZDictReleaseIterator(roZDictIter *iter) {
    server.cowCurIters.curObjZDictIter = NULL;
    if (iter->di != NULL) {
        dictReleaseIterator(iter->di);
    }
    zfree(iter);
}

/*	取消客户端订阅的所有频道，最后返回退订的频道数量	 */
int pubsubUnsubscribeAllChannels(redisClient *c, int notify) {
	// 获取迭代器
    dictIterator *di = dictGetSafeIterator(c->pubsub_channels);
    dictEntry *de;
    int count = 0;

    // 遍历c->pubsub_channels字典，逐一退订所订阅的频道
    while((de = dictNext(di)) != NULL) {
        robj *channel = dictGetKey(de);

        // 统计退订的频道数量
        count += pubsubUnsubscribeChannel(c,channel,notify);
    }
    /* We were subscribed to nothing? Still reply to the client. */
    // 如果count == 0，说明客户端没有订阅任何频道，回复客户端
    if (notify && count == 0) {
        addReply(c,shared.mbulkhdr[3]);
        addReply(c,shared.unsubscribebulk);
        addReply(c,shared.nullbulk);
        addReplyLongLong(c,dictSize(c->pubsub_channels)+
                       listLength(c->pubsub_patterns));
    }
    dictReleaseIterator(di);
    //	最后返回退订的频道数量
    return count;
}

void dbPrintSets(FILE *f)
{
    dictIterator *iter = NULL;
    dictEntry *entry = NULL;

    if (NULL == f)
        return;

    lockRead(sets);

    if (0 == dictSize(sets))
    {
        unlockRead(sets);
        fprintf(f, "No sets in db.\r\n");
        return;
    }

    if (NULL == (iter = dictGetIterator(sets)))
    {
        unlockRead(sets);
        return;
    }

    while (NULL != (entry = dictNext(iter)))
    {
        fprintf(f, "%s\r\n", dictGetEntryKey(entry));
        lockRead(dictGetEntryVal(entry));
        setPrint((set *) dictGetEntryVal(entry), f, 1);
        unlockRead(dictGetEntryVal(entry));
        fprintf(f, "\r\n");
    }

    dictReleaseIterator(iter);
    unlockRead(sets);
}

int dbFlushAll(void)
{
    dictIterator *iter = NULL;
    dictEntry *entry = NULL;

    lockWrite(sets);

    if (NULL == (iter = dictGetIterator(sets)))
    {
        unlockWrite(sets);
        return -1;
    }

    while (NULL != (entry = dictNext(iter)))
    {
        set *s = (set *) dictGetEntryVal(entry);
        lockWrite(s);
        if (!s->registered)
            setDestroy(s);
        unregisterSyncObject(s);
    }

    dictReleaseIterator(iter);
    dictEmpty(sets);
    unlockWrite(sets);
    return 0;
}

/* Unblock a client that's waiting in a blocking operation such as BLPOP */
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;
    }
    c->flags &= ~REDIS_BLOCKED;
    c->flags |= REDIS_UNBLOCKED;
    server.bpop_blocked_clients--;
    listAddNodeTail(server.unblocked_clients,c);
}