src/sort.c

#include "redis.h"
#include "pqsort.h" /* Partial qsort for SORT+LIMIT */
#include <math.h> /* isnan() */

redisSortOperation *createSortOperation(int type, robj *pattern) {
    redisSortOperation *so = zmalloc(sizeof(*so));
    so->type = type;
    so->pattern = pattern;
    return so;
}

/* Return the value associated to the key with a name obtained
 * substituting the first occurence of '*' in 'pattern' with 'subst'.
 * 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;
    sds spat, ssub;
    robj keyobj, fieldobj, *o;
    int prefixlen, sublen, postfixlen, fieldlen;
    /* Expoit the internal sds representation to create a sds string allocated on the stack in order to make this function faster */
    struct {
        int len;
        int free;
        char buf[REDIS_SORTKEY_MAX+1];
    } keyname, fieldname;

    /* 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 (sdslen(spat)+sdslen(ssub)-1 > REDIS_SORTKEY_MAX) return NULL;
    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) {
        fieldlen = sdslen(spat)-(f-spat);
        /* this also copies \0 character */
        memcpy(fieldname.buf,f+2,fieldlen-1);
        fieldname.len = fieldlen-2;
    } else {
        fieldlen = 0;
    }

    prefixlen = p-spat;
    sublen = sdslen(ssub);
    postfixlen = sdslen(spat)-(prefixlen+1)-fieldlen;
    memcpy(keyname.buf,spat,prefixlen);
    memcpy(keyname.buf+prefixlen,ssub,sublen);
    memcpy(keyname.buf+prefixlen+sublen,p+1,postfixlen);
    keyname.buf[prefixlen+sublen+postfixlen] = '\0';
    keyname.len = prefixlen+sublen+postfixlen;
    decrRefCount(subst);

    /* Lookup substituted key */
    initStaticStringObject(keyobj,((char*)&keyname)+(sizeof(struct sdshdr)));
    o = lookupKeyRead(db,&keyobj);
    if (o == NULL) return NULL;

    if (fieldlen > 0) {
        if (o->type != REDIS_HASH || fieldname.len < 1) return NULL;

        /* Retrieve value from hash by the field name. This operation
         * already increases the refcount of the returned object. */
        initStaticStringObject(fieldobj,((char*)&fieldname)+(sizeof(struct sdshdr)));
        o = hashTypeGetObject(o, &fieldobj);
    } else {
        if (o->type != REDIS_STRING) return NULL;

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

    return o;
}

/* sortCompare() is used by qsort in sortCommand(). Given that qsort_r with
 * the additional parameter is not standard but a BSD-specific we have to
 * pass sorting parameters via the global 'server' structure */
int sortCompare(const void *s1, const void *s2) {
    const redisSortObject *so1 = s1, *so2 = s2;
    int cmp;

    if (!server.sort_alpha) {
        /* Numeric sorting. Here it's trivial as we precomputed scores */
        if (so1->u.score > so2->u.score) {
            cmp = 1;
        } else if (so1->u.score < so2->u.score) {
            cmp = -1;
        } else {
            /* Objects have the same score, but we don't want the comparison
             * to be undefined, so we compare objects lexicographycally.
             * This way the result of SORT is deterministic. */
            cmp = compareStringObjects(so1->obj,so2->obj);
        }
    } else {
        /* Alphanumeric sorting */
        if (server.sort_bypattern) {
            if (!so1->u.cmpobj || !so2->u.cmpobj) {
                /* At least one compare object is NULL */
                if (so1->u.cmpobj == so2->u.cmpobj)
                    cmp = 0;
                else if (so1->u.cmpobj == NULL)
                    cmp = -1;
                else
                    cmp = 1;
            } else {
                /* We have both the objects, use strcoll */
                cmp = strcoll(so1->u.cmpobj->ptr,so2->u.cmpobj->ptr);
            }
        } else {
            /* Compare elements directly. */
            cmp = compareStringObjects(so1->obj,so2->obj);
        }
    }
    return server.sort_desc ? -cmp : cmp;
}

/* 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;
    long limit_start = 0, limit_count = -1, start, end;
    int j, dontsort = 0, vectorlen;
    int getop = 0; /* GET operation counter */
    int int_convertion_error = 0;
    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) {
            if ((getLongFromObjectOrReply(c, c->argv[j+1], &limit_start, NULL) != REDIS_OK) ||
                (getLongFromObjectOrReply(c, c->argv[j+2], &limit_count, NULL) != REDIS_OK)) return;
            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++;
    }

    /* If we have STORE we need to force sorting for deterministic output
     * and replication. We use alpha sorting since this is guaranteed to
     * work with any input. */
    if (storekey && dontsort) {
        dontsort = 0;
        alpha = 1;
        sortby = NULL;
    }

    /* 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;
            vector[j].u.score = 0;
            vector[j].u.cmpobj = NULL;
            j++;
        }
        setTypeReleaseIterator(si);
    } else if (sortval->type == REDIS_ZSET) {
        dict *set = ((zset*)sortval->ptr)->dict;
        dictIterator *di;
        dictEntry *setele;
        di = dictGetIterator(set);
        while((setele = dictNext(di)) != NULL) {
            vector[j].obj = dictGetKey(setele);
            vector[j].u.score = 0;
            vector[j].u.cmpobj = NULL;
            j++;
        }
        dictReleaseIterator(di);
    } else {
        redisPanic("Unknown type");
    }
    redisAssertWithInfo(c,sortval,j == vectorlen);

    /* Now it's time to load the right scores in the sorting vector */
    if (dontsort == 0) {
        for (j = 0; j < vectorlen; j++) {
            robj *byval;
            if (sortby) {
                /* lookup value to sort by */
                byval = lookupKeyByPattern(c->db,sortby,vector[j].obj);
                if (!byval) continue;
            } else {
                /* use object itself to sort by */
                byval = vector[j].obj;
            }

            if (alpha) {
                if (sortby) vector[j].u.cmpobj = getDecodedObject(byval);
            } else {
                if (byval->encoding == REDIS_ENCODING_RAW) {
                    char *eptr;

                    vector[j].u.score = strtod(byval->ptr,&eptr);
                    if (eptr[0] != '\0' || errno == ERANGE ||
                        isnan(vector[j].u.score))
                    {
                        int_convertion_error = 1;
                    }
                } else if (byval->encoding == REDIS_ENCODING_INT) {
                    /* Don't need to decode the object if it's
                     * integer-encoded (the only encoding supported) so
                     * far. We can just cast it */
                    vector[j].u.score = (long)byval->ptr;
                } else {
                    redisAssertWithInfo(c,sortval,1 != 1);
                }
            }

            /* when the object was retrieved using lookupKeyByPattern,
             * its refcount needs to be decreased. */
            if (sortby) {
                decrRefCount(byval);
            }
        }
    }

    /* We are ready to sort the vector... perform a bit of sanity check
     * on the LIMIT option too. We'll use a partial version of quicksort. */
    start = (limit_start < 0) ? 0 : limit_start;
    end = (limit_count < 0) ? vectorlen-1 : start+limit_count-1;
    if (start >= vectorlen) {
        start = vectorlen-1;
        end = vectorlen-2;
    }
    if (end >= vectorlen) end = vectorlen-1;

    server.sort_dontsort = dontsort;
    if (dontsort == 0) {
        server.sort_desc = desc;
        server.sort_alpha = alpha;
        server.sort_bypattern = sortby ? 1 : 0;
        if (sortby && (start != 0 || end != vectorlen-1))
            pqsort(vector,vectorlen,sizeof(redisSortObject),sortCompare, start,end);
        else
            qsort(vector,vectorlen,sizeof(redisSortObject),sortCompare);
    }

    /* Send command output to the output buffer, performing the specified
     * GET/DEL/INCR/DECR operations if any. */
    outputlen = getop ? getop*(end-start+1) : end-start+1;
    if (int_convertion_error) {
        addReplyError(c,"One or more scores can't be converted into double");
    } else if (storekey == NULL) {
        /* STORE option not specified, sent the sorting result to client */
        addReplyMultiBulkLen(c,outputlen);
        for (j = start; j <= end; j++) {
            listNode *ln;
            listIter li;

            if (!getop) addReplyBulk(c,vector[j].obj);
            listRewind(operations,&li);
            while((ln = listNext(&li))) {
                redisSortOperation *sop = ln->value;
                robj *val = lookupKeyByPattern(c->db,sop->pattern,
                    vector[j].obj);

                if (sop->type == REDIS_SORT_GET) {
                    if (!val) {
                        addReply(c,shared.nullbulk);
                    } else {
                        addReplyBulk(c,val);
                        decrRefCount(val);
                    }
                } else {
                    /* Always fails */
                    redisAssertWithInfo(c,sortval,sop->type == REDIS_SORT_GET);
                }
            }
        }
    } else {
        robj *sobj = createZiplistObject();

        /* STORE option specified, set the sorting result as a List object */
        for (j = start; j <= end; j++) {
            listNode *ln;
            listIter li;

            if (!getop) {
                listTypePush(sobj,vector[j].obj,REDIS_TAIL);
            } else {
                listRewind(operations,&li);
                while((ln = listNext(&li))) {
                    redisSortOperation *sop = ln->value;
                    robj *val = lookupKeyByPattern(c->db,sop->pattern,
                        vector[j].obj);

                    if (sop->type == REDIS_SORT_GET) {
                        if (!val) val = createStringObject("",0);

                        /* listTypePush does an incrRefCount, so we should take care
                         * care of the incremented refcount caused by either
                         * lookupKeyByPattern or createStringObject("",0) */
                        listTypePush(sobj,val,REDIS_TAIL);
                        decrRefCount(val);
                    } else {
                        /* Always fails */
                        redisAssertWithInfo(c,sortval,sop->type == REDIS_SORT_GET);
                    }
                }
            }
        }
        if (outputlen) {
            setKey(c->db,storekey,sobj);
            server.dirty += outputlen;
        } else if (dbDelete(c->db,storekey)) {
            signalModifiedKey(c->db,storekey);
            server.dirty++;
        }
        decrRefCount(sobj);
        addReplyLongLong(c,outputlen);
    }

    /* Cleanup */
    if (sortval->type == REDIS_LIST || sortval->type == REDIS_SET)
        for (j = 0; j < vectorlen; j++)
            decrRefCount(vector[j].obj);
    decrRefCount(sortval);
    listRelease(operations);
    for (j = 0; j < vectorlen; j++) {
        if (alpha && vector[j].u.cmpobj)
            decrRefCount(vector[j].u.cmpobj);
    }
    zfree(vector);
}
Commit	Line	Data
	1	#include "redis.h"
	2	#include "pqsort.h" /* Partial qsort for SORT+LIMIT */
	3	#include <math.h> /* isnan() */
	4
	5	redisSortOperation createSortOperation(int type, robj pattern) {
	6	redisSortOperation so = zmalloc(sizeof(so));
	7	so->type = type;
	8	so->pattern = pattern;
	9	return so;
	10	}
	11
	12	/* Return the value associated to the key with a name obtained
	13	* substituting the first occurence of '*' in 'pattern' with 'subst'.
	14	* The returned object will always have its refcount increased by 1
	15	* when it is non-NULL. */
	16	robj lookupKeyByPattern(redisDb db, robj pattern, robj subst) {
	17	char p, f;
	18	sds spat, ssub;
	19	robj keyobj, fieldobj, *o;
	20	int prefixlen, sublen, postfixlen, fieldlen;
	21	/* Expoit the internal sds representation to create a sds string allocated on the stack in order to make this function faster */
	22	struct {
	23	int len;
	24	int free;
	25	char buf[REDIS_SORTKEY_MAX+1];
	26	} keyname, fieldname;
	27
	28	/* If the pattern is "#" return the substitution object itself in order
	29	* to implement the "SORT ... GET #" feature. */
	30	spat = pattern->ptr;
	31	if (spat[0] == '#' && spat[1] == '\0') {
	32	incrRefCount(subst);
	33	return subst;
	34	}
	35
	36	/* The substitution object may be specially encoded. If so we create
	37	* a decoded object on the fly. Otherwise getDecodedObject will just
	38	* increment the ref count, that we'll decrement later. */
	39	subst = getDecodedObject(subst);
	40
	41	ssub = subst->ptr;
	42	if (sdslen(spat)+sdslen(ssub)-1 > REDIS_SORTKEY_MAX) return NULL;
	43	p = strchr(spat,'*');
	44	if (!p) {
	45	decrRefCount(subst);
	46	return NULL;
	47	}
	48
	49	/* Find out if we're dealing with a hash dereference. */
	50	if ((f = strstr(p+1, "->")) != NULL) {
	51	fieldlen = sdslen(spat)-(f-spat);
	52	/* this also copies \0 character */
	53	memcpy(fieldname.buf,f+2,fieldlen-1);
	54	fieldname.len = fieldlen-2;
	55	} else {
	56	fieldlen = 0;
	57	}
	58
	59	prefixlen = p-spat;
	60	sublen = sdslen(ssub);
	61	postfixlen = sdslen(spat)-(prefixlen+1)-fieldlen;
	62	memcpy(keyname.buf,spat,prefixlen);
	63	memcpy(keyname.buf+prefixlen,ssub,sublen);
	64	memcpy(keyname.buf+prefixlen+sublen,p+1,postfixlen);
	65	keyname.buf[prefixlen+sublen+postfixlen] = '\0';
	66	keyname.len = prefixlen+sublen+postfixlen;
	67	decrRefCount(subst);
	68
	69	/* Lookup substituted key */
	70	initStaticStringObject(keyobj,((char*)&keyname)+(sizeof(struct sdshdr)));
	71	o = lookupKeyRead(db,&keyobj);
	72	if (o == NULL) return NULL;
	73
	74	if (fieldlen > 0) {
	75	if (o->type != REDIS_HASH \|\| fieldname.len < 1) return NULL;
	76
	77	/* Retrieve value from hash by the field name. This operation
	78	* already increases the refcount of the returned object. */
	79	initStaticStringObject(fieldobj,((char*)&fieldname)+(sizeof(struct sdshdr)));
	80	o = hashTypeGetObject(o, &fieldobj);
	81	} else {
	82	if (o->type != REDIS_STRING) return NULL;
	83
	84	/* Every object that this function returns needs to have its refcount
	85	* increased. sortCommand decreases it again. */
	86	incrRefCount(o);
	87	}
	88
	89	return o;
	90	}
	91
	92	/* sortCompare() is used by qsort in sortCommand(). Given that qsort_r with
	93	* the additional parameter is not standard but a BSD-specific we have to
	94	* pass sorting parameters via the global 'server' structure */
	95	int sortCompare(const void s1, const void s2) {
	96	const redisSortObject so1 = s1, so2 = s2;
	97	int cmp;
	98
	99	if (!server.sort_alpha) {
	100	/* Numeric sorting. Here it's trivial as we precomputed scores */
	101	if (so1->u.score > so2->u.score) {
	102	cmp = 1;
	103	} else if (so1->u.score < so2->u.score) {
	104	cmp = -1;
	105	} else {
	106	/* Objects have the same score, but we don't want the comparison
	107	* to be undefined, so we compare objects lexicographycally.
	108	* This way the result of SORT is deterministic. */
	109	cmp = compareStringObjects(so1->obj,so2->obj);
	110	}
	111	} else {
	112	/* Alphanumeric sorting */
	113	if (server.sort_bypattern) {
	114	if (!so1->u.cmpobj \|\| !so2->u.cmpobj) {
	115	/* At least one compare object is NULL */
	116	if (so1->u.cmpobj == so2->u.cmpobj)
	117	cmp = 0;
	118	else if (so1->u.cmpobj == NULL)
	119	cmp = -1;
	120	else
	121	cmp = 1;
	122	} else {
	123	/* We have both the objects, use strcoll */
	124	cmp = strcoll(so1->u.cmpobj->ptr,so2->u.cmpobj->ptr);
	125	}
	126	} else {
	127	/* Compare elements directly. */
	128	cmp = compareStringObjects(so1->obj,so2->obj);
	129	}
	130	}
	131	return server.sort_desc ? -cmp : cmp;
	132	}
	133
	134	/* The SORT command is the most complex command in Redis. Warning: this code
	135	* is optimized for speed and a bit less for readability */
	136	void sortCommand(redisClient *c) {
	137	list *operations;
	138	unsigned int outputlen = 0;
	139	int desc = 0, alpha = 0;
	140	long limit_start = 0, limit_count = -1, start, end;
	141	int j, dontsort = 0, vectorlen;
	142	int getop = 0; /* GET operation counter */
	143	int int_convertion_error = 0;
	144	robj sortval, sortby = NULL, *storekey = NULL;
	145	redisSortObject vector; / Resulting vector to sort */
	146
	147	/* Lookup the key to sort. It must be of the right types */
	148	sortval = lookupKeyRead(c->db,c->argv[1]);
	149	if (sortval && sortval->type != REDIS_SET && sortval->type != REDIS_LIST &&
	150	sortval->type != REDIS_ZSET)
	151	{
	152	addReply(c,shared.wrongtypeerr);
	153	return;
	154	}
	155
	156	/* Create a list of operations to perform for every sorted element.
	157	* Operations can be GET/DEL/INCR/DECR */
	158	operations = listCreate();
	159	listSetFreeMethod(operations,zfree);
	160	j = 2;
	161
	162	/* Now we need to protect sortval incrementing its count, in the future
	163	* SORT may have options able to overwrite/delete keys during the sorting
	164	* and the sorted key itself may get destroied */
	165	if (sortval)
	166	incrRefCount(sortval);
	167	else
	168	sortval = createListObject();
	169
	170	/* The SORT command has an SQL-alike syntax, parse it */
	171	while(j < c->argc) {
	172	int leftargs = c->argc-j-1;
	173	if (!strcasecmp(c->argv[j]->ptr,"asc")) {
	174	desc = 0;
	175	} else if (!strcasecmp(c->argv[j]->ptr,"desc")) {
	176	desc = 1;
	177	} else if (!strcasecmp(c->argv[j]->ptr,"alpha")) {
	178	alpha = 1;
	179	} else if (!strcasecmp(c->argv[j]->ptr,"limit") && leftargs >= 2) {
	180	if ((getLongFromObjectOrReply(c, c->argv[j+1], &limit_start, NULL) != REDIS_OK) \|\|
	181	(getLongFromObjectOrReply(c, c->argv[j+2], &limit_count, NULL) != REDIS_OK)) return;
	182	j+=2;
	183	} else if (!strcasecmp(c->argv[j]->ptr,"store") && leftargs >= 1) {
	184	storekey = c->argv[j+1];
	185	j++;
	186	} else if (!strcasecmp(c->argv[j]->ptr,"by") && leftargs >= 1) {
	187	sortby = c->argv[j+1];
	188	/* If the BY pattern does not contain '*', i.e. it is constant,
	189	* we don't need to sort nor to lookup the weight keys. */
	190	if (strchr(c->argv[j+1]->ptr,'*') == NULL) dontsort = 1;
	191	j++;
	192	} else if (!strcasecmp(c->argv[j]->ptr,"get") && leftargs >= 1) {
	193	listAddNodeTail(operations,createSortOperation(
	194	REDIS_SORT_GET,c->argv[j+1]));
	195	getop++;
	196	j++;
	197	} else {
	198	decrRefCount(sortval);
	199	listRelease(operations);
	200	addReply(c,shared.syntaxerr);
	201	return;
	202	}
	203	j++;
	204	}
	205
	206	/* If we have STORE we need to force sorting for deterministic output
	207	* and replication. We use alpha sorting since this is guaranteed to
	208	* work with any input. */
	209	if (storekey && dontsort) {
	210	dontsort = 0;
	211	alpha = 1;
	212	sortby = NULL;
	213	}
	214
	215	/* Destructively convert encoded sorted sets for SORT. */
	216	if (sortval->type == REDIS_ZSET)
	217	zsetConvert(sortval, REDIS_ENCODING_SKIPLIST);
	218
	219	/* Load the sorting vector with all the objects to sort */
	220	switch(sortval->type) {
	221	case REDIS_LIST: vectorlen = listTypeLength(sortval); break;
	222	case REDIS_SET: vectorlen = setTypeSize(sortval); break;
	223	case REDIS_ZSET: vectorlen = dictSize(((zset*)sortval->ptr)->dict); break;
	224	default: vectorlen = 0; redisPanic("Bad SORT type"); /* Avoid GCC warning */
	225	}
	226	vector = zmalloc(sizeof(redisSortObject)*vectorlen);
	227	j = 0;
	228
	229	if (sortval->type == REDIS_LIST) {
	230	listTypeIterator *li = listTypeInitIterator(sortval,0,REDIS_TAIL);
	231	listTypeEntry entry;
	232	while(listTypeNext(li,&entry)) {
	233	vector[j].obj = listTypeGet(&entry);
	234	vector[j].u.score = 0;
	235	vector[j].u.cmpobj = NULL;
	236	j++;
	237	}
	238	listTypeReleaseIterator(li);
	239	} else if (sortval->type == REDIS_SET) {
	240	setTypeIterator *si = setTypeInitIterator(sortval);
	241	robj *ele;
	242	while((ele = setTypeNextObject(si)) != NULL) {
	243	vector[j].obj = ele;
	244	vector[j].u.score = 0;
	245	vector[j].u.cmpobj = NULL;
	246	j++;
	247	}
	248	setTypeReleaseIterator(si);
	249	} else if (sortval->type == REDIS_ZSET) {
	250	dict set = ((zset)sortval->ptr)->dict;
	251	dictIterator *di;
	252	dictEntry *setele;
	253	di = dictGetIterator(set);
	254	while((setele = dictNext(di)) != NULL) {
	255	vector[j].obj = dictGetKey(setele);
	256	vector[j].u.score = 0;
	257	vector[j].u.cmpobj = NULL;
	258	j++;
	259	}
	260	dictReleaseIterator(di);
	261	} else {
	262	redisPanic("Unknown type");
	263	}
	264	redisAssertWithInfo(c,sortval,j == vectorlen);
	265
	266	/* Now it's time to load the right scores in the sorting vector */
	267	if (dontsort == 0) {
	268	for (j = 0; j < vectorlen; j++) {
	269	robj *byval;
	270	if (sortby) {
	271	/* lookup value to sort by */
	272	byval = lookupKeyByPattern(c->db,sortby,vector[j].obj);
	273	if (!byval) continue;
	274	} else {
	275	/* use object itself to sort by */
	276	byval = vector[j].obj;
	277	}
	278
	279	if (alpha) {
	280	if (sortby) vector[j].u.cmpobj = getDecodedObject(byval);
	281	} else {
	282	if (byval->encoding == REDIS_ENCODING_RAW) {
	283	char *eptr;
	284
	285	vector[j].u.score = strtod(byval->ptr,&eptr);
	286	if (eptr[0] != '\0' \|\| errno == ERANGE \|\|
	287	isnan(vector[j].u.score))
	288	{
	289	int_convertion_error = 1;
	290	}
	291	} else if (byval->encoding == REDIS_ENCODING_INT) {
	292	/* Don't need to decode the object if it's
	293	* integer-encoded (the only encoding supported) so
	294	* far. We can just cast it */
	295	vector[j].u.score = (long)byval->ptr;
	296	} else {
	297	redisAssertWithInfo(c,sortval,1 != 1);
	298	}
	299	}
	300
	301	/* when the object was retrieved using lookupKeyByPattern,
	302	* its refcount needs to be decreased. */
	303	if (sortby) {
	304	decrRefCount(byval);
	305	}
	306	}
	307	}
	308
	309	/* We are ready to sort the vector... perform a bit of sanity check
	310	* on the LIMIT option too. We'll use a partial version of quicksort. */
	311	start = (limit_start < 0) ? 0 : limit_start;
	312	end = (limit_count < 0) ? vectorlen-1 : start+limit_count-1;
	313	if (start >= vectorlen) {
	314	start = vectorlen-1;
	315	end = vectorlen-2;
	316	}
	317	if (end >= vectorlen) end = vectorlen-1;
	318
	319	server.sort_dontsort = dontsort;
	320	if (dontsort == 0) {
	321	server.sort_desc = desc;
	322	server.sort_alpha = alpha;
	323	server.sort_bypattern = sortby ? 1 : 0;
	324	if (sortby && (start != 0 \|\| end != vectorlen-1))
	325	pqsort(vector,vectorlen,sizeof(redisSortObject),sortCompare, start,end);
	326	else
	327	qsort(vector,vectorlen,sizeof(redisSortObject),sortCompare);
	328	}
	329
	330	/* Send command output to the output buffer, performing the specified
	331	* GET/DEL/INCR/DECR operations if any. */
	332	outputlen = getop ? getop*(end-start+1) : end-start+1;
	333	if (int_convertion_error) {
	334	addReplyError(c,"One or more scores can't be converted into double");
	335	} else if (storekey == NULL) {
	336	/* STORE option not specified, sent the sorting result to client */
	337	addReplyMultiBulkLen(c,outputlen);
	338	for (j = start; j <= end; j++) {
	339	listNode *ln;
	340	listIter li;
	341
	342	if (!getop) addReplyBulk(c,vector[j].obj);
	343	listRewind(operations,&li);
	344	while((ln = listNext(&li))) {
	345	redisSortOperation *sop = ln->value;
	346	robj *val = lookupKeyByPattern(c->db,sop->pattern,
	347	vector[j].obj);
	348
	349	if (sop->type == REDIS_SORT_GET) {
	350	if (!val) {
	351	addReply(c,shared.nullbulk);
	352	} else {
	353	addReplyBulk(c,val);
	354	decrRefCount(val);
	355	}
	356	} else {
	357	/* Always fails */
	358	redisAssertWithInfo(c,sortval,sop->type == REDIS_SORT_GET);
	359	}
	360	}
	361	}
	362	} else {
	363	robj *sobj = createZiplistObject();
	364
	365	/* STORE option specified, set the sorting result as a List object */
	366	for (j = start; j <= end; j++) {
	367	listNode *ln;
	368	listIter li;
	369
	370	if (!getop) {
	371	listTypePush(sobj,vector[j].obj,REDIS_TAIL);
	372	} else {
	373	listRewind(operations,&li);
	374	while((ln = listNext(&li))) {
	375	redisSortOperation *sop = ln->value;
	376	robj *val = lookupKeyByPattern(c->db,sop->pattern,
	377	vector[j].obj);
	378
	379	if (sop->type == REDIS_SORT_GET) {
	380	if (!val) val = createStringObject("",0);
	381
	382	/* listTypePush does an incrRefCount, so we should take care
	383	* care of the incremented refcount caused by either
	384	* lookupKeyByPattern or createStringObject("",0) */
	385	listTypePush(sobj,val,REDIS_TAIL);
	386	decrRefCount(val);
	387	} else {
	388	/* Always fails */
	389	redisAssertWithInfo(c,sortval,sop->type == REDIS_SORT_GET);
	390	}
	391	}
	392	}
	393	}
	394	if (outputlen) {
	395	setKey(c->db,storekey,sobj);
	396	server.dirty += outputlen;
	397	} else if (dbDelete(c->db,storekey)) {
	398	signalModifiedKey(c->db,storekey);
	399	server.dirty++;
	400	}
	401	decrRefCount(sobj);
	402	addReplyLongLong(c,outputlen);
	403	}
	404
	405	/* Cleanup */
	406	if (sortval->type == REDIS_LIST \|\| sortval->type == REDIS_SET)
	407	for (j = 0; j < vectorlen; j++)
	408	decrRefCount(vector[j].obj);
	409	decrRefCount(sortval);
	410	listRelease(operations);
	411	for (j = 0; j < vectorlen; j++) {
	412	if (alpha && vector[j].u.cmpobj)
	413	decrRefCount(vector[j].u.cmpobj);
	414	}
	415	zfree(vector);
	416	}
	417
	418