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

static robj *lookupKey(redisDb *db, robj *key) {
    dictEntry *de = dictFind(db->dict,key);
    if (de) {
        robj *key = dictGetEntryKey(de);
        robj *val = dictGetEntryVal(de);

        if (server.vm_enabled) {
            if (key->storage == REDIS_VM_MEMORY ||
                key->storage == REDIS_VM_SWAPPING)
            {
                /* If we were swapping the object out, stop it, this key
                 * was requested. */
                if (key->storage == REDIS_VM_SWAPPING)
                    vmCancelThreadedIOJob(key);
                /* Update the access time of the key for the aging algorithm. */
                key->vm.atime = server.unixtime;
            } else {
                int notify = (key->storage == REDIS_VM_LOADING);

                /* Our value was swapped on disk. Bring it at home. */
                redisAssert(val == NULL);
                val = vmLoadObject(key);
                dictGetEntryVal(de) = val;

                /* Clients blocked by the VM subsystem may be waiting for
                 * this key... */
                if (notify) handleClientsBlockedOnSwappedKey(db,key);
            }
        }
        return val;
    } else {
        return NULL;
    }
}

robj *lookupKey(redisDb *db, robj *key) {
    dictEntry *de = dictFind(db->dict,key->ptr);
    if (de) {
        robj *val = dictGetEntryVal(de);

        if (server.vm_enabled) {
            if (val->storage == REDIS_VM_MEMORY ||
                val->storage == REDIS_VM_SWAPPING)
            {
                /* If we were swapping the object out, cancel the operation */
                if (val->storage == REDIS_VM_SWAPPING)
                    vmCancelThreadedIOJob(val);
                /* Update the access time for the aging algorithm. */
                val->lru = server.lruclock;
            } else {
                int notify = (val->storage == REDIS_VM_LOADING);

                /* Our value was swapped on disk. Bring it at home. */
                redisAssert(val->type == REDIS_VMPOINTER);
                val = vmLoadObject(val);
                dictGetEntryVal(de) = val;

                /* Clients blocked by the VM subsystem may be waiting for
                 * this key... */
                if (notify) handleClientsBlockedOnSwappedKey(db,key);
            }
        }
        return val;
    } else {
        return NULL;
    }
}

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

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

/* Remove the 'key' from the list of blocked keys for a given client.
 *
 * The function returns 1 when there are no longer blocking keys after
 * the current one was removed (and the client can be unblocked). */
int dontWaitForSwappedKey(redisClient *c, robj *key) {
    list *l;
    listNode *ln;
    listIter li;
    struct dictEntry *de;

    /* The key object might be destroyed when deleted from the c->io_keys
     * list (and the "key" argument is physically the same object as the
     * object inside the list), so we need to protect it. */
    incrRefCount(key);

    /* Remove the key from the list of keys this client is waiting for. */
    listRewind(c->io_keys,&li);
    while ((ln = listNext(&li)) != NULL) {
        if (equalStringObjects(ln->value,key)) {
            listDelNode(c->io_keys,ln);
            break;
        }
    }
    redisAssert(ln != NULL);

    /* Remove the client form the key => waiting clients map. */
    de = dictFind(c->db->io_keys,key);
    redisAssert(de != NULL);
    l = dictGetEntryVal(de);
    ln = listSearchKey(l,c);
    redisAssert(ln != NULL);
    listDelNode(l,ln);
    if (listLength(l) == 0)
        dictDelete(c->db->io_keys,key);

    decrRefCount(key);
    return listLength(c->io_keys) == 0;
}

/* Unblock a client that's waiting in a blocking operation such as BLPOP */
void unblockClientWaitingData(redisClient *c) {
    dictEntry *de;
    list *l;
    int j;

    redisAssert(c->bpop.keys != NULL);
    /* The client may wait for multiple keys, so unblock it for every key. */
    for (j = 0; j < c->bpop.count; j++) {
        /* Remove this client from the list of clients waiting for this key. */
        de = dictFind(c->db->blocking_keys,c->bpop.keys[j]);
        redisAssert(de != NULL);
        l = dictGetEntryVal(de);
        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,c->bpop.keys[j]);
        decrRefCount(c->bpop.keys[j]);
    }

    /* Cleanup the client structure */
    zfree(c->bpop.keys);
    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);
}

/* Set a client in blocking mode for the specified key, with the specified
 * timeout */
void blockForKeys(redisClient *c, robj **keys, int numkeys, time_t timeout) {
    dictEntry *de;
    list *l;
    int j;

    c->blocking_keys = zmalloc(sizeof(robj*)*numkeys);
    c->blocking_keys_num = numkeys;
    c->blockingto = timeout;
    for (j = 0; j < numkeys; j++) {
        /* Add the key in the client structure, to map clients -> keys */
        c->blocking_keys[j] = keys[j];
        incrRefCount(keys[j]);

        /* And in the other "side", to map keys -> clients */
        de = dictFind(c->db->blocking_keys,keys[j]);
        if (de == NULL) {
            int retval;

            /* For every key we take a list of clients blocked for it */
            l = listCreate();
            retval = dictAdd(c->db->blocking_keys,keys[j],l);
            incrRefCount(keys[j]);
            redisAssert(retval == DICT_OK);
        } else {
            l = dictGetEntryVal(de);
        }
        listAddNodeTail(l,c);
    }
    /* Mark the client as a blocked client */
    c->flags |= REDIS_BLOCKED;
    server.blpop_blocked_clients++;
}

void zremCommand(redisClient *c) {
    robj *zsetobj;
    zset *zs;
    dictEntry *de;
    double curscore;
    int deleted;

    if ((zsetobj = lookupKeyWriteOrReply(c,c->argv[1],shared.czero)) == NULL ||
        checkType(c,zsetobj,REDIS_ZSET)) return;

    zs = zsetobj->ptr;
    c->argv[2] = tryObjectEncoding(c->argv[2]);
    de = dictFind(zs->dict,c->argv[2]);
    if (de == NULL) {
        addReply(c,shared.czero);
        return;
    }
    /* Delete from the skiplist */
    curscore = *(double*)dictGetEntryVal(de);
    deleted = zslDelete(zs->zsl,curscore,c->argv[2]);
    redisAssert(deleted != 0);

    /* Delete from the hash table */
    dictDelete(zs->dict,c->argv[2]);
    if (htNeedsResize(zs->dict)) dictResize(zs->dict);
    if (dictSize(zs->dict) == 0) dbDelete(c->db,c->argv[1]);
    touchWatchedKey(c->db,c->argv[1]);
    server.dirty++;
    addReply(c,shared.cone);
}

void zrankGenericCommand(redisClient *c, int reverse) {
    robj *o;
    zset *zs;
    zskiplist *zsl;
    dictEntry *de;
    unsigned long rank;
    double *score;

    if ((o = lookupKeyReadOrReply(c,c->argv[1],shared.nullbulk)) == NULL ||
        checkType(c,o,REDIS_ZSET)) return;

    zs = o->ptr;
    zsl = zs->zsl;
    c->argv[2] = tryObjectEncoding(c->argv[2]);
    de = dictFind(zs->dict,c->argv[2]);
    if (!de) {
        addReply(c,shared.nullbulk);
        return;
    }

    score = dictGetEntryVal(de);
    rank = zslGetRank(zsl, *score, c->argv[2]);
    if (rank) {
        if (reverse) {
            addReplyLongLong(c, zsl->length - rank);
        } else {
            addReplyLongLong(c, rank-1);
        }
    } else {
        addReply(c,shared.nullbulk);
    }
}

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

/* Unblock a client that's waiting in a blocking operation such as BLPOP */
void unblockClientWaitingData(redisClient *c) {
    dictEntry *de;
    list *l;
    int j;

    redisAssert(c->blocking_keys != NULL);
    /* The client may wait for multiple keys, so unblock it for every key. */
    for (j = 0; j < c->blocking_keys_num; j++) {
        /* Remove this client from the list of clients waiting for this key. */
        de = dictFind(c->db->blocking_keys,c->blocking_keys[j]);
        redisAssert(de != NULL);
        l = dictGetEntryVal(de);
        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,c->blocking_keys[j]);
        decrRefCount(c->blocking_keys[j]);
    }
    /* Cleanup the client structure */
    zfree(c->blocking_keys);
    c->blocking_keys = NULL;
    c->flags &= (~REDIS_BLOCKED);
    server.blpop_blocked_clients--;
    /* We want to process data if there is some command waiting
     * in the input buffer. Note that this is safe even if
     * unblockClientWaitingData() gets called from freeClient() because
     * freeClient() will be smart enough to call this function
     * *after* c->querybuf was set to NULL. */
    if (c->querybuf && sdslen(c->querybuf) > 0) processInputBuffer(c);
}

void processCleanup(void) {
    dictIterator *di = dictGetIterator(server.process);
    dictEntry *de;
    while ((de = dictNext(di)) != NULL) {
        struct ProcessInfo *p = dictGetEntryVal(de);
        if (p->time_stamp != g_time) deleteProcess(p->pid);
    }
    dictReleaseIterator(di);
}

void movekeysCommand(redisClient *c) {
    redisDb *src, *dst;
    int srcid;

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

    /* Obtain source and target DB pointers */
    src = c->db;
    srcid = c->db->id;
    if (selectDb(c,atoi(c->argv[2]->ptr)) == REDIS_ERR) {
        addReply(c,shared.outofrangeerr);
        return;
    }
    dst = c->db;
    selectDb(c,srcid); /* Back to the source DB */

    /* If the user is moving using as target the same
     * DB as the source DB it is probably an error. */
    if (src == dst) {
        addReply(c,shared.sameobjecterr);
        return;
    }

    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 (expireIfNeeded(c->db,keyobj) == 0) {
                robj *val = dictGetEntryVal(de);

                /* Try to add the element to the target DB */
                if (dbAdd(dst,keyobj,val) != REDIS_ERR) {
                    incrRefCount(val);

                    /* OK! key moved, free the entry in the source DB */
                    dbDelete(src,keyobj);
                    server.dirty++;
                    numkeys++;
                }
            }
            decrRefCount(keyobj);
        }
    }
    dictReleaseIterator(di);
    addReplyLongLong(c,numkeys);
}

/* Return the expire time of the specified key, or -1 if no expire
 * is associated with this key (i.e. the key is non volatile) */
time_t getExpire(redisDb *db, robj *key) {
    dictEntry *de;

    /* No expire? return ASAP */
    if (dictSize(db->expires) == 0 ||
       (de = dictFind(db->expires,key->ptr)) == NULL) return -1;

    /* The entry was found in the expire dict, this means it should also
     * be present in the main dict (safety check). */
    redisAssert(dictFind(db->dict,key->ptr) != NULL);
    return (time_t) dictGetEntryVal(de);
}

/* This should be called from any function PUSHing into lists.
 * 'c' is the "pushing client", 'key' is the key it is pushing data against,
 * 'ele' is the element pushed.
 *
 * If the function returns 0 there was no client waiting for a list push
 * against this key.
 *
 * If the function returns 1 there was a client waiting for a list push
 * against this key, the element was passed to this client thus it's not
 * needed to actually add it to the list and the caller should return asap. */
int handleClientsWaitingListPush(redisClient *c, robj *key, robj *ele) {
    struct dictEntry *de;
    redisClient *receiver;
    int numclients;
    list *clients;
    listNode *ln;
    robj *dstkey, *dstobj;

    de = dictFind(c->db->blocking_keys,key);
    if (de == NULL) return 0;
    clients = dictGetEntryVal(de);
    numclients = listLength(clients);

    /* Try to handle the push as long as there are clients waiting for a push.
     * Note that "numclients" is used because the list of clients waiting for a
     * push on "key" is deleted by unblockClient() when empty.
     *
     * This loop will have more than 1 iteration when there is a BRPOPLPUSH
     * that cannot push the target list because it does not contain a list. If
     * this happens, it simply tries the next client waiting for a push. */
    while (numclients--) {
        ln = listFirst(clients);
        redisAssert(ln != NULL);
        receiver = ln->value;
        dstkey = receiver->bpop.target;

        /* Protect receiver->bpop.target, that will be freed by
         * the next unblockClientWaitingData() call. */
        if (dstkey) incrRefCount(dstkey);
        /* This should remove the first element of the "clients" list. */
        unblockClientWaitingData(receiver);

        if (dstkey == NULL) {
            /* BRPOP/BLPOP */
            addReplyMultiBulkLen(receiver,2);
            addReplyBulk(receiver,key);
            addReplyBulk(receiver,ele);
            return 1; /* Serve just the first client as in B[RL]POP semantics */
        } else {
            /* BRPOPLPUSH, note that receiver->db is always equal to c->db. */
            dstobj = lookupKeyWrite(receiver->db,dstkey);
            if (!(dstobj && checkType(receiver,dstobj,REDIS_LIST))) {
                rpoplpushHandlePush(c,receiver,dstkey,dstobj,ele);
                decrRefCount(dstkey);
                return 1;
            }
            decrRefCount(dstkey);
        }
    }

    return 0;
}

void  destroyTaskMap(dict* p) 
{
    if (p) {
        dictIterator* di = dictGetIterator(p);
        dictEntry* de;
        while ((de = dictNext(di)) != NULL) {
            ugTaskType* ptask = dictGetEntryVal(de); 
            luaworkUnrefFunction(server.ls, ptask->handle, NULL);            
        }
        dictReleaseIterator(di);
        dictRelease(p);
    }
}

robj *lookupKey(redisDb *db, robj *key) {
    dictEntry *de = dictFind(db->dict,key->ptr);
    if (de) {
        robj *val = dictGetEntryVal(de);

        /* Update the access time for the aging algorithm.
         * Don't do it if we have a saving child, as this will trigger
         * a copy on write madness. */
        if (server.bgsavechildpid == -1 && server.bgrewritechildpid == -1)
            val->lru = server.lruclock;

        if (server.vm_enabled) {
            if (val->storage == REDIS_VM_MEMORY ||
                val->storage == REDIS_VM_SWAPPING)
            {
                /* If we were swapping the object out, cancel the operation */
                if (val->storage == REDIS_VM_SWAPPING)
                    vmCancelThreadedIOJob(val);
            } else {
                int notify = (val->storage == REDIS_VM_LOADING);

                /* Our value was swapped on disk. Bring it at home. */
                redisAssert(val->type == REDIS_VMPOINTER);
                val = vmLoadObject(val);
                dictGetEntryVal(de) = val;

                /* Clients blocked by the VM subsystem may be waiting for
                 * this key... */
                if (notify) handleClientsBlockedOnSwappedKey(db,key);
            }
        }
        server.stat_keyspace_hits++;
        return val;
    } else {
        server.stat_keyspace_misses++;
        return NULL;
    }
}

robj *lookupKeyWithVersion(redisDb *db, robj *key, uint16_t* version) {
    dictEntry *de = dictFind(db->dict, key->ptr);
    if(de) {
        robj *val = dictGetEntryVal(de);
        sds key_tmp = dictGetEntryKey(de);

        *version = sdsversion(key_tmp);

        val->lru = shared.lruclock;
        db->stat_keyspace_hits++;
        return val;
    } else {
        db->stat_keyspace_misses++;
        return NULL;
    }
}

robj *lookupKey(redisDb *db, robj *key) {
    dictEntry *de = dictFind(db->dict,key->ptr);
    if (de) {
        robj *val = dictGetEntryVal(de);

        /* Update the access time for the aging algorithm.
         * Don't do it if we have a saving child, as this will trigger
         * a copy on write madness. */
        if (server.bgsavechildpid == -1 && server.bgrewritechildpid == -1)
            val->lru = server.lruclock;

        return val;
    } else {
        return NULL;
    }
}