/* Return the value associated to the key with a name obtained using
 * the following rules:
 *
 * 1) The first occurrence of '*' in 'pattern' is substituted with 'subst'.
 *
 * 2) If 'pattern' matches the "->" string, everything on the left of
 *    the arrow is treated as the name of a hash field, and the part on the
 *    left as the key name containing a hash. The value of the specified
 *    field is returned.
 *
 * 3) If 'pattern' equals "#", the function simply returns 'subst' itself so
 *    that the SORT command can be used like: SORT key GET # to retrieve
 *    the Set/List elements directly.
 *
 * The returned object will always have its refcount increased by 1
 * when it is non-NULL. */
robj *lookupKeyByPattern(redisDb *db, robj *pattern, robj *subst) {
    char *p, *f, *k;
    sds spat, ssub;
    robj *keyobj, *fieldobj = NULL, *o;
    int prefixlen, sublen, postfixlen, fieldlen;

    /* If the pattern is "#" return the substitution object itself in order
     * to implement the "SORT ... GET #" feature. */
    spat = pattern->ptr;
    if (spat[0] == '#' && spat[1] == '\0') {
        incrRefCount(subst);
        return subst;
    }

    /* The substitution object may be specially encoded. If so we create
     * a decoded object on the fly. Otherwise getDecodedObject will just
     * increment the ref count, that we'll decrement later. */
    subst = getDecodedObject(subst);
    ssub = subst->ptr;

    /* If we can't find '*' in the pattern we return NULL as to GET a
     * fixed key does not make sense. */
    p = strchr(spat,'*');
    if (!p) {
        decrRefCount(subst);
        return NULL;
    }

    /* Find out if we're dealing with a hash dereference. */
    if ((f = strstr(p+1, "->")) != NULL && *(f+2) != '\0') {
        fieldlen = sdslen(spat)-(f-spat)-2;
        fieldobj = createStringObject(f+2,fieldlen);
    } else {
        fieldlen = 0;
    }

    /* Perform the '*' substitution. */
    prefixlen = p-spat;
    sublen = sdslen(ssub);
    postfixlen = sdslen(spat)-(prefixlen+1)-(fieldlen ? fieldlen+2 : 0);
    keyobj = createStringObject(NULL,prefixlen+sublen+postfixlen);
    k = keyobj->ptr;
    memcpy(k,spat,prefixlen);
    memcpy(k+prefixlen,ssub,sublen);
    memcpy(k+prefixlen+sublen,p+1,postfixlen);
    decrRefCount(subst); /* Incremented by decodeObject() */

    /* Lookup substituted key */
    o = lookupKeyRead(db,keyobj);
    if (o == NULL) goto noobj;

    if (fieldobj) {
        if (o->type != REDIS_HASH) goto noobj;

        /* Retrieve value from hash by the field name. This operation
         * already increases the refcount of the returned object. */
        o = hashTypeGetObject(o, fieldobj);
    } else {
        if (o->type != REDIS_STRING) goto noobj;

        /* Every object that this function returns needs to have its refcount
         * increased. sortCommand decreases it again. */
        incrRefCount(o);
    }
    decrRefCount(keyobj);
    if (fieldobj) decrRefCount(fieldobj);
    return o;

noobj:
    decrRefCount(keyobj);
    if (fieldlen) decrRefCount(fieldobj);
    return NULL;
}

void freePubsubPattern(void *p) {
    pubsubPattern *pat = p;

    decrRefCount(pat->pattern);
    zfree(pat);
}

int luaRedisGenericCommand(lua_State *lua, int raise_error) {
    int j, argc = lua_gettop(lua);
    struct redisCommand *cmd;
    robj **argv;
    redisClient *c = server.lua_client;
    sds reply;

    // 参数必须大于零
    /* Require at least one argument */
    if (argc == 0) {
        luaPushError(lua,
            "Please specify at least one argument for redis.call()");
        return 1;
    }

    // 处理参数
    /* Build the arguments vector */
    argv = zmalloc(sizeof(robj*)*argc);
    for (j = 0; j < argc; j++) {
        if (!lua_isstring(lua,j+1)) break;
        argv[j] = createStringObject((char*)lua_tostring(lua,j+1),
                                     lua_strlen(lua,j+1));
    }

    /* Check if one of the arguments passed by the Lua script
     * is not a string or an integer (lua_isstring() return true for
     * integers as well). */
    if (j != argc) {
        j--;
        while (j >= 0) {
            decrRefCount(argv[j]);
            j--;
        }
        zfree(argv);
        luaPushError(lua,
            "Lua redis() command arguments must be strings or integers");
        return 1;
    }

    // 将参数设置为虚拟客户端的参数
    /* Setup our fake client for command execution */
    c->argv = argv;
    c->argc = argc;

    // 查找命令
    /* Command lookup */
    cmd = lookupCommand(argv[0]->ptr);
    if (!cmd || ((cmd->arity > 0 && cmd->arity != argc) ||
                   (argc < -cmd->arity)))
    {
        if (cmd)
            luaPushError(lua,
                "Wrong number of args calling Redis command From Lua script");
        else
            luaPushError(lua,"Unknown Redis command called from Lua script");
        goto cleanup;
    }

    /* There are commands that are not allowed inside scripts. */
    if (cmd->flags & REDIS_CMD_NOSCRIPT) {
        luaPushError(lua, "This Redis command is not allowed from scripts");
        goto cleanup;
    }

    /* Write commands are forbidden against read-only slaves, or if a
     * command marked as non-deterministic was already called in the context
     * of this script. */
    // 涉及写操作的命令
    if (cmd->flags & REDIS_CMD_WRITE) {
        // 如果曾经 lua 脚本中执行未决命令，譬如 RANDOMKEY，那么写操作是不
        // 允许的
        if (server.lua_random_dirty) {
            luaPushError(lua,
                "Write commands not allowed after non deterministic commands");
            goto cleanup;

        // 此服务器为设定了只读的从机，且未处于加载数据阶段，那么写操作是不
        // 允许的
        } else if (server.masterhost && server.repl_slave_ro &&
                   !server.loading &&
                   !(server.lua_caller->flags & REDIS_MASTER))
        {
            luaPushError(lua, shared.roslaveerr->ptr);
            goto cleanup;

        // 如果设定了当 BGSAVE 失败时不能执行写操作，而且 BGSAVE 失败了，
        // 那么写操作是不允许的
        // server.stop_writes_on_bgsave_err 选项是说默认当 BGSAVE 持久化
        // 失败的时候，禁止接受写操作。当客户端发现写操作失败的时候，可以及时
        // 发现 redis 服务器持久化失败了，因此，这是给客户端发现持久化失败的
        // 机会
        } else if (server.stop_writes_on_bgsave_err &&
                   server.saveparamslen > 0 &&
                   server.lastbgsave_status == REDIS_ERR)
        {
            luaPushError(lua, shared.bgsaveerr->ptr);
            goto cleanup;
        }
    }

//.........这里部分代码省略.........

/// 尝试将STRING类型的obj编码做编码类型的改变或者移除sds剩余空间,以节省内存
robj *tryObjectEncoding(robj *o) {
    long value;
    sds s = o->ptr;
    size_t len;

    /* Make sure this is a string object, the only type we encode
     * in this function. Other types use encoded memory efficient
     * representations but are handled by the commands implementing
     * the type. */
    /// 一定得是STRING类型的哦
    redisAssertWithInfo(NULL,o,o->type == REDIS_STRING);

    /* We try some specialized encoding only for objects that are
     * RAW or EMBSTR encoded, in other words objects that are still
     * in represented by an actually array of chars. */
    /// 一定得是字符串编码的哦
    if (!sdsEncodedObject(o)) return o;

    /* It's not safe to encode shared objects: shared objects can be shared
     * everywhere in the "object space" of Redis and may end in places where
     * they are not handled. We handle them only as values in the keyspace. */
    /// 引用计数大于1,说明在别处引用了,不能改 
    if (o->refcount > 1) return o;

    /* Check if we can represent this string as a long integer.
     * Note that we are sure that a string larger than 21 chars is not
     * representable as a 32 nor 64 bit integer. */
    len = sdslen(s);
    /// 可以编码为long
    if (len <= 21 && string2l(s,len,&value)) {
        /* This object is encodable as a long. Try to use a shared object.
         * Note that we avoid using shared integers when maxmemory is used
         * because every object needs to have a private LRU field for the LRU
         * algorithm to work well. */
        /// 重新设置为共享obj
        if ((server.maxmemory == 0 ||
             (server.maxmemory_policy != REDIS_MAXMEMORY_VOLATILE_LRU &&
              server.maxmemory_policy != REDIS_MAXMEMORY_ALLKEYS_LRU)) &&
            value >= 0 &&
            value < REDIS_SHARED_INTEGERS)
        {
            decrRefCount(o);
            incrRefCount(shared.integers[value]);
            return shared.integers[value];
        } else {
            /// 将字符编码转为INT编码
            if (o->encoding == REDIS_ENCODING_RAW) sdsfree(o->ptr);
            o->encoding = REDIS_ENCODING_INT;
            o->ptr = (void*) value;
            return o;
        }
    }

    /* If the string is small and is still RAW encoded,
     * try the EMBSTR encoding which is more efficient.
     * In this representation the object and the SDS string are allocated
     * in the same chunk of memory to save space and cache misses. */
    /// 可以编码为嵌入式字符串类型
    if (len <= REDIS_ENCODING_EMBSTR_SIZE_LIMIT) {
        robj *emb;

        /// 已经是嵌入式字符串编码了,直接返回
        if (o->encoding == REDIS_ENCODING_EMBSTR) return o;

        emb = createEmbeddedStringObject(s,sdslen(s));
        decrRefCount(o);
        return emb;
    }

    /* We can't encode the object...
     *
     * Do the last try, and at least optimize the SDS string inside
     * the string object to require little space, in case there
     * is more than 10% of free space at the end of the SDS string.
     *
     * We do that only for relatively large strings as this branch
     * is only entered if the length of the string is greater than
     * REDIS_ENCODING_EMBSTR_SIZE_LIMIT. */
    /// 编码类型为RAW且sds剩余字节数大于字符长度的十分之一
    if (o->encoding == REDIS_ENCODING_RAW &&
        sdsavail(s) > len/10)
    {
        /// 将剩余空间移除
        o->ptr = sdsRemoveFreeSpace(o->ptr);
    }

    /* Return the original object. */
    return o;
}

/* The SORT command is the most complex command in Redis. Warning: this code
 * is optimized for speed and a bit less for readability */
void sortCommand(redisClient *c) {
    list *operations;
    unsigned int outputlen = 0;
    int desc = 0, alpha = 0;
    int limit_start = 0, limit_count = -1, start, end;
    int j, dontsort = 0, vectorlen;
    int getop = 0; /* GET operation counter */
    robj *sortval, *sortby = NULL, *storekey = NULL;
    redisSortObject *vector; /* Resulting vector to sort */

    /* Lookup the key to sort. It must be of the right types */
    sortval = lookupKeyRead(c->db,c->argv[1]);
    if (sortval && sortval->type != REDIS_SET && sortval->type != REDIS_LIST &&
        sortval->type != REDIS_ZSET)
    {
        addReply(c,shared.wrongtypeerr);
        return;
    }

    /* Create a list of operations to perform for every sorted element.
     * Operations can be GET/DEL/INCR/DECR */
    operations = listCreate();
    listSetFreeMethod(operations,zfree);
    j = 2;

    /* Now we need to protect sortval incrementing its count, in the future
     * SORT may have options able to overwrite/delete keys during the sorting
     * and the sorted key itself may get destroied */
    if (sortval)
        incrRefCount(sortval);
    else
        sortval = createListObject();

    /* The SORT command has an SQL-alike syntax, parse it */
    while(j < c->argc) {
        int leftargs = c->argc-j-1;
        if (!strcasecmp(c->argv[j]->ptr,"asc")) {
            desc = 0;
        } else if (!strcasecmp(c->argv[j]->ptr,"desc")) {
            desc = 1;
        } else if (!strcasecmp(c->argv[j]->ptr,"alpha")) {
            alpha = 1;
        } else if (!strcasecmp(c->argv[j]->ptr,"limit") && leftargs >= 2) {
            limit_start = atoi(c->argv[j+1]->ptr);
            limit_count = atoi(c->argv[j+2]->ptr);
            j+=2;
        } else if (!strcasecmp(c->argv[j]->ptr,"store") && leftargs >= 1) {
            storekey = c->argv[j+1];
            j++;
        } else if (!strcasecmp(c->argv[j]->ptr,"by") && leftargs >= 1) {
            sortby = c->argv[j+1];
            /* If the BY pattern does not contain '*', i.e. it is constant,
             * we don't need to sort nor to lookup the weight keys. */
            if (strchr(c->argv[j+1]->ptr,'*') == NULL) dontsort = 1;
            j++;
        } else if (!strcasecmp(c->argv[j]->ptr,"get") && leftargs >= 1) {
            listAddNodeTail(operations,createSortOperation(
                REDIS_SORT_GET,c->argv[j+1]));
            getop++;
            j++;
        } else {
            decrRefCount(sortval);
            listRelease(operations);
            addReply(c,shared.syntaxerr);
            return;
        }
        j++;
    }

    /* Destructively convert encoded sorted sets for SORT. */
    if (sortval->type == REDIS_ZSET)
        zsetConvert(sortval, REDIS_ENCODING_SKIPLIST);

    /* Load the sorting vector with all the objects to sort */
    switch(sortval->type) {
    case REDIS_LIST: vectorlen = listTypeLength(sortval); break;
    case REDIS_SET: vectorlen =  setTypeSize(sortval); break;
    case REDIS_ZSET: vectorlen = dictSize(((zset*)sortval->ptr)->dict); break;
    default: vectorlen = 0; redisPanic("Bad SORT type"); /* Avoid GCC warning */
    }
    vector = zmalloc(sizeof(redisSortObject)*vectorlen);
    j = 0;

    if (sortval->type == REDIS_LIST) {
        listTypeIterator *li = listTypeInitIterator(sortval,0,REDIS_TAIL);
        listTypeEntry entry;
        while(listTypeNext(li,&entry)) {
            vector[j].obj = listTypeGet(&entry);
            vector[j].u.score = 0;
            vector[j].u.cmpobj = NULL;
            j++;
        }
        listTypeReleaseIterator(li);
    } else if (sortval->type == REDIS_SET) {
        setTypeIterator *si = setTypeInitIterator(sortval);
        robj *ele;
        while((ele = setTypeNextObject(si)) != NULL) {
            vector[j].obj = ele;
//.........这里部分代码省略.........

//.........这里部分代码省略.........
                    bestdbid = j;
                    break;
                }
            }
        }

        /* volatile-ttl */
        else if (server.maxmemory_policy == MAXMEMORY_VOLATILE_TTL) {
            long bestttl = 0; /* Initialized to avoid warning. */

            /* In this policy we scan a single DB per iteration (visiting
             * a different DB per call), expiring the key with the smallest
             * TTL among the few sampled.
             *
             * Note that this algorithm makes local-DB choices, and should
             * use a pool and code more similr to the one used in the
             * LRU eviction policies in the future. */
            for (i = 0; i < server.dbnum; i++) {
                j = (++next_db) % server.dbnum;
                db = server.db+j;
                dict = db->expires;
                if (dictSize(dict) != 0) {
                    for (k = 0; k < server.maxmemory_samples; k++) {
                        sds thiskey;
                        long thisttl;

                        de = dictGetRandomKey(dict);
                        thiskey = dictGetKey(de);
                        thisttl = (long) dictGetVal(de);

                        /* Keys expiring sooner (smaller unix timestamp) are
                         * better candidates for deletion */
                        if (bestkey == NULL || thisttl < bestttl) {
                            bestkey = thiskey;
                            bestttl = thisttl;
                            bestdbid = j;
                        }
                    }
                }
            }
        }

        /* Finally remove the selected key. */
        if (bestkey) {
            db = server.db+bestdbid;
            robj *keyobj = createStringObject(bestkey,sdslen(bestkey));
            propagateExpire(db,keyobj,server.lazyfree_lazy_eviction);
            /* We compute the amount of memory freed by db*Delete() alone.
             * It is possible that actually the memory needed to propagate
             * the DEL in AOF and replication link is greater than the one
             * we are freeing removing the key, but we can't account for
             * that otherwise we would never exit the loop.
             *
             * AOF and Output buffer memory will be freed eventually so
             * we only care about memory used by the key space. */
            delta = (long long) zmalloc_used_memory();
            latencyStartMonitor(eviction_latency);
            if (server.lazyfree_lazy_eviction)
                dbAsyncDelete(db,keyobj);
            else
                dbSyncDelete(db,keyobj);
            latencyEndMonitor(eviction_latency);
            latencyAddSampleIfNeeded("eviction-del",eviction_latency);
            latencyRemoveNestedEvent(latency,eviction_latency);
            delta -= (long long) zmalloc_used_memory();
            mem_freed += delta;
            server.stat_evictedkeys++;
            notifyKeyspaceEvent(NOTIFY_EVICTED, "evicted",
                keyobj, db->id);
            decrRefCount(keyobj);
            keys_freed++;

            /* When the memory to free starts to be big enough, we may
             * start spending so much time here that is impossible to
             * deliver data to the slaves fast enough, so we force the
             * transmission here inside the loop. */
            if (slaves) flushSlavesOutputBuffers();
        }

        if (!keys_freed) {
            latencyEndMonitor(latency);
            latencyAddSampleIfNeeded("eviction-cycle",latency);
            goto cant_free; /* nothing to free... */
        }
    }
    latencyEndMonitor(latency);
    latencyAddSampleIfNeeded("eviction-cycle",latency);
    return C_OK;

cant_free:
    /* We are here if we are not able to reclaim memory. There is only one
     * last thing we can try: check if the lazyfree thread has jobs in queue
     * and wait... */
    while(bioPendingJobsOfType(BIO_LAZY_FREE)) {
        if (((mem_reported - zmalloc_used_memory()) + mem_freed) >= mem_tofree)
            break;
        usleep(1000);
    }
    return C_ERR;
}

/* Try to encode a string object in order to save space */
robj *tryObjectEncoding(robj *o) {
    long value;
    sds s = o->ptr;
    size_t len;

    if (o->encoding != REDIS_ENCODING_RAW)
        return o; /* Already encoded */

    /* It's not safe to encode shared objects: shared objects can be shared
     * everywhere in the "object space" of Redis. Encoded objects can only
     * appear as "values" (and not, for instance, as keys) */
     if (o->refcount > 1) return o;

    /* Currently we try to encode only strings */
    redisAssertWithInfo(NULL,o,o->type == REDIS_STRING);

    /* Check if we can represent this string as a long integer */
    len = sdslen(s);
    if (len > 21 || !string2l(s,len,&value)) {
        /* We can't encode the object...
         *
         * Do the last try, and at least optimize the SDS string inside
         * the string object to require little space, in case there
         * is more than 10% of free space at the end of the SDS string.
         *
         * We do that for larger strings, using the arbitrary value
         * of 32 bytes. This code was backported from the unstable branch
         * where this is performed when the object is too large to be
         * encoded as EMBSTR. */
        if (len > 32 &&
            o->encoding == REDIS_ENCODING_RAW &&
            sdsavail(s) > len/10)
        {
            o->ptr = sdsRemoveFreeSpace(o->ptr);
        }
        /* Return the original object. */
        return o;
    }

    /* Ok, this object can be encoded...
     *
     * Can I use a shared object? Only if the object is inside a given range
     *
     * Note that we also avoid using shared integers when maxmemory is used
     * because every object needs to have a private LRU field for the LRU
     * algorithm to work well. */
    if ((server.maxmemory == 0 ||
         (server.maxmemory_policy != REDIS_MAXMEMORY_VOLATILE_LRU &&
          server.maxmemory_policy != REDIS_MAXMEMORY_ALLKEYS_LRU)) &&
        value >= 0 && value < REDIS_SHARED_INTEGERS)
    {
        decrRefCount(o);
        incrRefCount(shared.integers[value]);
        return shared.integers[value];
    } else {
        o->encoding = REDIS_ENCODING_INT;
        sdsfree(o->ptr);
        o->ptr = (void*) value;
        return o;
    }
}

/* Returns ID(s) of documents added */
void jsondocsetbyjsonCommand(redisClient *c) {
    /* args 0-N: ["jsondocsetbyjson", id-field-name, json] */
    jsonSyncClients(c);

    sds field_name = c->argv[1]->ptr;
    struct jsonObj *root = yajl_decode(c->argv[2]->ptr, NULL);

    if (!root) {
        addReply(c, g.err_parse);
        return;
    }

    robj *key;
    if (root->type == JSON_TYPE_LIST) {
        robj *keys[root->content.obj.elements];
        D("Procesing list!\n");
        /* First, process all documents for names to make sure they are valid
         * documents.  We don't want to add half the documents then reach a
         * failure scenario. */
        for (int i = 0; i < root->content.obj.elements; i++) {
            struct jsonObj *o = root->content.obj.fields[i];
            key = jsonObjFindId(field_name, o);
            keys[i] = key;
            if (!key) {
                /* Free any allocated keys so far */
                for (int j = 0; i < j; j++)
                    decrRefCount(keys[j]);

                jsonObjFree(root);
                addReplyErrorFormat(
                    c,
                    "field '%s' not found or unusable as key for document %d",
                    field_name, i);
                return;
            }
        }
        /* Now actually add all the documents */
        /* Note how a multi-set gets a multibulk reply while
         * a regular one-document set gets just one bulk result. */
        addReplyMultiBulkLen(c, root->content.obj.elements);
        for (int i = 0; i < root->content.obj.elements; i++) {
            struct jsonObj *o = root->content.obj.fields[i];
            jsonObjAddToDB(keys[i], o);
            addReplyBulkCBuffer(c, keys[i]->ptr, sdslen(keys[i]->ptr));
            decrRefCount(keys[i]);
        }
    } else if (root->type == JSON_TYPE_MAP) {
        key = jsonObjFindId(field_name, root);
        if (key) {
            jsonObjAddToDB(key, root);
            addReplyBulkCBuffer(c, key->ptr, sdslen(key->ptr));
            decrRefCount(key);
        } else {
            addReplyErrorFormat(
                c, "field '%s' not found or unusable as key for document",
                field_name);
        }
    } else {
        addReplyError(c, "JSON isn't map or array of maps.");
    }

    jsonObjFree(root);
}

void jsonfieldincrbyGeneric(redisClient *c, bool use_incrbyfloat) {
    /* args 0-N: ["jsonfieldincrby", json field components, incrby number] */
    jsonSyncClients(c);

    if (!validateKeyFormatAndReply(c, c->argv[1]->ptr))
        return;

    sds key = genFieldAccessor(c, 2);
    sds found;
    int type;
    int decode_as = findKeyForHash(key, &found, &type);
    sdsfree(key);

    if (!found) {
        addReplyError(c, "JSON document not found");
        return;
    } else if (decode_as == DECODE_INDIVIDUAL) {
        /* 'field' is the second to last argv[] element */
        struct jsonObj *f =
            hgetToJsonObj(found, decode_as, c->argv[c->argc - 2]->ptr);
        switch (f->type) {
        case JSON_TYPE_MAP:
        case JSON_TYPE_LIST:
        case JSON_TYPE_TRUE:
        case JSON_TYPE_FALSE:
        case JSON_TYPE_NULL:
            addReplyError(c, "Requested field not usable for incrby. Can't "
                             "increment non-number types.");
            break;
        case JSON_TYPE_NUMBER:
        /* found->content.number += incrby;
        break; */
        case JSON_TYPE_STRING:
        case JSON_TYPE_NUMBER_AS_STRING:
            /* strtoi -> += incrby -> store again */
            addReplyError(c,
                          "Not currently supported on mixed-type containers.");
            break;
        }
        jsonObjFree(f);
        return;
    }

    /* Target args: 0-3: [_, HASH, FIELD, INCRBY] */
    /* The hincrby* commands don't check argc, so we don't care if
     * we have extra arguments after INCRBY.  They'll get cleaned up
     * when the client exits. */
    decrRefCount(c->argv[1]);
    c->argv[1] = dbstrTake(found);

    /* If argc == 4, then the second and third arguments are already okay.
     * If argc  > 4, we move the last two arguments to positions 3-4 */
    if (c->argc > 4) {
        decrRefCount(c->argv[2]);          /* goodbye, original argv[2] */
        c->argv[2] = c->argv[c->argc - 2]; /* field is 2nd to last argument */
        c->argv[c->argc - 2] = NULL;

        /* If argc == 5, then argv[3] is the one we moved to argv[2] above.
         * We can't release it because it just moved storage locations */
        if (c->argc > 5)
            decrRefCount(c->argv[3]);      /* goodbye, original argv[3] */
        c->argv[3] = c->argv[c->argc - 1]; /* incrby value is last argument */
        c->argv[c->argc - 1] = NULL;
        c->argc -= c->argc > 5 ? 2 : 1; /* if > 5, we removed two. else, we
                                         * removed one and moved the other. */
    }

    if (use_incrbyfloat)
        hincrbyfloatCommand(c);
    else
        hincrbyCommand(c);
}

/* *
 * do migrate mutli key-value(s) for {slotsmgrt/slotsmgrtone}with tag commands
 * return value:
 *    -1 - error happens
 *   >=0 - # of success migration
 * */
static int
slotsmgrttag_command(redisClient *c, sds host, sds port, int timeout, robj *key) {
    int taglen;
    void *tag = slots_tag(key->ptr, &taglen);
    if (tag == NULL) {
        return slotsmgrtone_command(c, host, port, timeout, key);
    }

    int fd = slotsmgrt_get_socket(c, host, port, timeout);
    if (fd == -1) {
        return -1;
    }

    list *l = listCreate();
    listSetFreeMethod(l, decrRefCountVoid);
    do {
        uint32_t crc;
        int slot = slots_num(key->ptr, &crc);
        dict *d = c->db->hash_slots[slot];
        long long cursor = 0;
        void *args[] = {l, tag, &taglen, (void *)(long)crc};
        do {
            cursor = dictScan(d, cursor, slotsScanSdsKeyTagCallback, args);
        } while (cursor != 0);
    } while (0);

    int max = listLength(l);
    if (max == 0) {
        listRelease(l);
        return 0;
    }

    robj **keys = zmalloc(sizeof(robj *) * max);
    robj **vals = zmalloc(sizeof(robj *) * max);

    int n = 0;
    for (int i = 0; i < max; i ++) {
        listNode *head = listFirst(l);
        robj *key = listNodeValue(head);
        robj *val = lookupKeyWrite(c->db, key);
        if (val != NULL) {
            keys[n] = key;
            vals[n] = val;
            n ++;
            incrRefCount(key);
        }
        listDelNode(l, head);
    }

    int ret = 0;
    if (n != 0) {
        if (slotsmgrt(c, host, port, fd, c->db->id, timeout, keys, vals, n) != 0) {
            slotsmgrt_close_socket(host, port);
            ret = -1;
        } else {
            slotsremove(c, keys, n, 1);
            ret = n;
        }
    }

    listRelease(l);
    for (int i = 0; i < n; i ++) {
        decrRefCount(keys[i]);
    }
    zfree(keys);
    zfree(vals);
    return ret;
}

/* Check the specified RDB file. Return 0 if the RDB looks sane, otherwise
 * 1 is returned. */
int redis_check_rdb(char *rdbfilename) {
    uint64_t dbid;
    int type, rdbver;
    char buf[1024];
    long long expiretime, now = mstime();
    FILE *fp;
    static rio rdb; /* Pointed by global struct riostate. */

    if ((fp = fopen(rdbfilename,"r")) == NULL) return 1;

    rioInitWithFile(&rdb,fp);
    rdbstate.rio = &rdb;
    rdb.update_cksum = rdbLoadProgressCallback;
    if (rioRead(&rdb,buf,9) == 0) goto eoferr;
    buf[9] = '\0';
    if (memcmp(buf,"REDIS",5) != 0) {
        rdbCheckError("Wrong signature trying to load DB from file");
        return 1;
    }
    rdbver = atoi(buf+5);
    if (rdbver < 1 || rdbver > RDB_VERSION) {
        rdbCheckError("Can't handle RDB format version %d",rdbver);
        return 1;
    }

    startLoading(fp);
    while(1) {
        robj *key, *val;
        expiretime = -1;

        /* Read type. */
        rdbstate.doing = RDB_CHECK_DOING_READ_TYPE;
        if ((type = rdbLoadType(&rdb)) == -1) goto eoferr;

        /* Handle special types. */
        if (type == RDB_OPCODE_EXPIRETIME) {
            rdbstate.doing = RDB_CHECK_DOING_READ_EXPIRE;
            /* EXPIRETIME: load an expire associated with the next key
             * to load. Note that after loading an expire we need to
             * load the actual type, and continue. */
            if ((expiretime = rdbLoadTime(&rdb)) == -1) goto eoferr;
            /* We read the time so we need to read the object type again. */
            rdbstate.doing = RDB_CHECK_DOING_READ_TYPE;
            if ((type = rdbLoadType(&rdb)) == -1) goto eoferr;
            /* the EXPIRETIME opcode specifies time in seconds, so convert
             * into milliseconds. */
            expiretime *= 1000;
        } else if (type == RDB_OPCODE_EXPIRETIME_MS) {
            /* EXPIRETIME_MS: milliseconds precision expire times introduced
             * with RDB v3. Like EXPIRETIME but no with more precision. */
            rdbstate.doing = RDB_CHECK_DOING_READ_EXPIRE;
            if ((expiretime = rdbLoadMillisecondTime(&rdb)) == -1) goto eoferr;
            /* We read the time so we need to read the object type again. */
            rdbstate.doing = RDB_CHECK_DOING_READ_TYPE;
            if ((type = rdbLoadType(&rdb)) == -1) goto eoferr;
        } else if (type == RDB_OPCODE_EOF) {
            /* EOF: End of file, exit the main loop. */
            break;
        } else if (type == RDB_OPCODE_SELECTDB) {
            /* SELECTDB: Select the specified database. */
            rdbstate.doing = RDB_CHECK_DOING_READ_LEN;
            if ((dbid = rdbLoadLen(&rdb,NULL)) == RDB_LENERR)
                goto eoferr;
            rdbCheckInfo("Selecting DB ID %d", dbid);
            continue; /* Read type again. */
        } else if (type == RDB_OPCODE_RESIZEDB) {
            /* RESIZEDB: Hint about the size of the keys in the currently
             * selected data base, in order to avoid useless rehashing. */
            uint64_t db_size, expires_size;
            rdbstate.doing = RDB_CHECK_DOING_READ_LEN;
            if ((db_size = rdbLoadLen(&rdb,NULL)) == RDB_LENERR)
                goto eoferr;
            if ((expires_size = rdbLoadLen(&rdb,NULL)) == RDB_LENERR)
                goto eoferr;
            continue; /* Read type again. */
        } else if (type == RDB_OPCODE_AUX) {
            /* AUX: generic string-string fields. Use to add state to RDB
             * which is backward compatible. Implementations of RDB loading
             * are requierd to skip AUX fields they don't understand.
             *
             * An AUX field is composed of two strings: key and value. */
            robj *auxkey, *auxval;
            rdbstate.doing = RDB_CHECK_DOING_READ_AUX;
            if ((auxkey = rdbLoadStringObject(&rdb)) == NULL) goto eoferr;
            if ((auxval = rdbLoadStringObject(&rdb)) == NULL) goto eoferr;

            rdbCheckInfo("AUX FIELD %s = '%s'",
                (char*)auxkey->ptr, (char*)auxval->ptr);
            decrRefCount(auxkey);
            decrRefCount(auxval);
            continue; /* Read type again. */
        } else {
            if (!rdbIsObjectType(type)) {
                rdbCheckError("Invalid object type: %d", type);
                return 1;
            }
            rdbstate.key_type = type;
        }
//.........这里部分代码省略.........

/* *
 * slotsrestore key ttl val [key ttl val ...]
 * */
void
slotsrestoreCommand(redisClient *c) {
    if (c->argc < 4 || (c->argc - 1) % 3 != 0) {
        addReplyErrorFormat(c, "wrong number of arguments for 'slotsrestore' command");
        return;
    }
    int n = (c->argc - 1) / 3;

    long long *ttls = zmalloc(sizeof(long long) * n);
    robj **vals = zmalloc(sizeof(robj *) * n);
    for (int i = 0; i < n; i ++) {
        vals[i] = NULL;
    }

    for (int i = 0; i < n; i ++) {
        robj *key = c->argv[i * 3 + 1];
        robj *ttl = c->argv[i * 3 + 2];
        robj *val = c->argv[i * 3 + 3];
        if (lookupKeyWrite(c->db, key) != NULL) {
            redisLog(REDIS_WARNING, "slotsrestore: slot = %d, key = '%s' already exists",
                    slots_num(key->ptr, NULL), (char *)key->ptr);
        }
        if (getLongLongFromObjectOrReply(c, ttl, &ttls[i], NULL) != REDIS_OK) {
            goto cleanup;
        } else if (ttls[i] < 0) {
            addReplyError(c, "invalid ttl value, must be >= 0");
            goto cleanup;
        }
        rio payload;
        int type;
        if (verifyDumpPayload(val->ptr, sdslen(val->ptr)) != REDIS_OK) {
            addReplyError(c, "dump payload version or checksum are wrong");
            goto cleanup;
        }
        rioInitWithBuffer(&payload, val->ptr);
        if (((type = rdbLoadObjectType(&payload)) == -1) ||
                ((vals[i] = rdbLoadObject(type, &payload)) == NULL)) {
            addReplyError(c, "bad data format");
            goto cleanup;
        }
    }

    for (int i = 0; i < n; i ++) {
        robj *key = c->argv[i * 3 + 1];
        long long ttl = ttls[i];
        robj *val = vals[i];
        dbDelete(c->db, key);
        dbAdd(c->db, key, val);
        incrRefCount(val);
        if (ttl) {
            setExpire(c->db, key, mstime() + ttl);
        }
        signalModifiedKey(c->db, key);
        server.dirty ++;
    }
    addReply(c, shared.ok);

cleanup:
    for (int i = 0; i < n; i ++) {
        if (vals[i] != NULL) {
            decrRefCount(vals[i]);
        }
    }
    zfree(vals);
    zfree(ttls);
}

//.........这里部分代码省略.........
                    }
                }
            }
        }

        /* volatile-random and allkeys-random policy */
        else if (server.maxmemory_policy == MAXMEMORY_ALLKEYS_RANDOM ||
                 server.maxmemory_policy == MAXMEMORY_VOLATILE_RANDOM)
        {
            /* When evicting a random key, we try to evict a key for
             * each DB, so we use the static 'next_db' variable to
             * incrementally visit all DBs. */
            for (i = 0; i < server.dbnum; i++) {
                j = (++next_db) % server.dbnum;
                db = server.db+j;
                dict = (server.maxmemory_policy == MAXMEMORY_ALLKEYS_RANDOM) ?
                        db->dict : db->expires;
                if (dictSize(dict) != 0) {
                    de = dictGetRandomKey(dict);
                    bestkey = dictGetKey(de);
                    bestdbid = j;
                    break;
                }
            }
        }

        /* Finally remove the selected key. */
        if (bestkey) {
            db = server.db+bestdbid;
            robj *keyobj = createStringObject(bestkey,sdslen(bestkey));
            propagateExpire(db,keyobj,server.lazyfree_lazy_eviction);
            /* We compute the amount of memory freed by db*Delete() alone.
             * It is possible that actually the memory needed to propagate
             * the DEL in AOF and replication link is greater than the one
             * we are freeing removing the key, but we can't account for
             * that otherwise we would never exit the loop.
             *
             * AOF and Output buffer memory will be freed eventually so
             * we only care about memory used by the key space. */
            delta = (long long) zmalloc_used_memory();
            latencyStartMonitor(eviction_latency);
            if (server.lazyfree_lazy_eviction)
                dbAsyncDelete(db,keyobj);
            else
                dbSyncDelete(db,keyobj);
            latencyEndMonitor(eviction_latency);
            latencyAddSampleIfNeeded("eviction-del",eviction_latency);
            latencyRemoveNestedEvent(latency,eviction_latency);
            delta -= (long long) zmalloc_used_memory();
            mem_freed += delta;
            server.stat_evictedkeys++;
            notifyKeyspaceEvent(NOTIFY_EVICTED, "evicted",
                keyobj, db->id);
            decrRefCount(keyobj);
            keys_freed++;

            /* When the memory to free starts to be big enough, we may
             * start spending so much time here that is impossible to
             * deliver data to the slaves fast enough, so we force the
             * transmission here inside the loop. */
            if (slaves) flushSlavesOutputBuffers();

            /* Normally our stop condition is the ability to release
             * a fixed, pre-computed amount of memory. However when we
             * are deleting objects in another thread, it's better to
             * check, from time to time, if we already reached our target
             * memory, since the "mem_freed" amount is computed only
             * across the dbAsyncDelete() call, while the thread can
             * release the memory all the time. */
            if (server.lazyfree_lazy_eviction && !(keys_freed % 16)) {
                overhead = freeMemoryGetNotCountedMemory();
                mem_used = zmalloc_used_memory();
                mem_used = (mem_used > overhead) ? mem_used-overhead : 0;
                if (mem_used <= server.maxmemory) {
                    mem_freed = mem_tofree;
                }
            }
        }

        if (!keys_freed) {
            latencyEndMonitor(latency);
            latencyAddSampleIfNeeded("eviction-cycle",latency);
            goto cant_free; /* nothing to free... */
        }
    }
    latencyEndMonitor(latency);
    latencyAddSampleIfNeeded("eviction-cycle",latency);
    return C_OK;

cant_free:
    /* We are here if we are not able to reclaim memory. There is only one
     * last thing we can try: check if the lazyfree thread has jobs in queue
     * and wait... */
    while(bioPendingJobsOfType(BIO_LAZY_FREE)) {
        if (((mem_reported - zmalloc_used_memory()) + mem_freed) >= mem_tofree)
            break;
        usleep(1000);
    }
    return C_ERR;
}

void replicationFeedSlaves(list *slaves, int dictid, robj **argv, int argc) {
    listNode *ln;
    listIter li;
    int outc = 0, j;
    robj **outv;
    /* We need 1+(ARGS*3) objects since commands are using the new protocol
     * and we one 1 object for the first "*<count>\r\n" multibulk count, then
     * for every additional object we have "$<count>\r\n" + object + "\r\n". */
    robj *static_outv[REDIS_STATIC_ARGS*3+1];
    robj *lenobj;

    if (argc <= REDIS_STATIC_ARGS) {
        outv = static_outv;
    } else {
        outv = zmalloc(sizeof(robj*)*(argc*3+1));
    }

    lenobj = createObject(REDIS_STRING,
            sdscatprintf(sdsempty(), "*%d\r\n", argc));
    lenobj->refcount = 0;
    outv[outc++] = lenobj;
    for (j = 0; j < argc; j++) {
        lenobj = createObject(REDIS_STRING,
            sdscatprintf(sdsempty(),"$%lu\r\n",
                (unsigned long) stringObjectLen(argv[j])));
        lenobj->refcount = 0;
        outv[outc++] = lenobj;
        outv[outc++] = argv[j];
        outv[outc++] = shared.crlf;
    }

    /* Increment all the refcounts at start and decrement at end in order to
     * be sure to free objects if there is no slave in a replication state
     * able to be feed with commands */
    for (j = 0; j < outc; j++) incrRefCount(outv[j]);
    listRewind(slaves,&li);
    while((ln = listNext(&li))) {
        redisClient *slave = ln->value;

        /* Don't feed slaves that are still waiting for BGSAVE to start */
        if (slave->replstate == REDIS_REPL_WAIT_BGSAVE_START) continue;

        /* Feed all the other slaves, MONITORs and so on */
        if (slave->slaveseldb != dictid) {
            robj *selectcmd;

            switch(dictid) {
            case 0: selectcmd = shared.select0; break;
            case 1: selectcmd = shared.select1; break;
            case 2: selectcmd = shared.select2; break;
            case 3: selectcmd = shared.select3; break;
            case 4: selectcmd = shared.select4; break;
            case 5: selectcmd = shared.select5; break;
            case 6: selectcmd = shared.select6; break;
            case 7: selectcmd = shared.select7; break;
            case 8: selectcmd = shared.select8; break;
            case 9: selectcmd = shared.select9; break;
            default:
                selectcmd = createObject(REDIS_STRING,
                    sdscatprintf(sdsempty(),"select %d\r\n",dictid));
                selectcmd->refcount = 0;
                break;
            }
            addReply(slave,selectcmd);
            slave->slaveseldb = dictid;
        }
        for (j = 0; j < outc; j++) addReply(slave,outv[j]);
    }
    for (j = 0; j < outc; j++) decrRefCount(outv[j]);
    if (outv != static_outv) zfree(outv);
}

/* If you reload the module *without* freeing things you allocate in load(),
 * then you *will* introduce memory leaks. */
void cleanup(void *privdata) {
    /* dictRelease will free every key, every value, then the dict itself. */
    dictRelease(g.names);
    decrRefCount(g.err.nosha);
    decrRefCount(g.err.noname);
}