Commit 7c0e1b53 authored by antirez's avatar antirez
Browse files

Merge remote branch 'pietern/unstable-zset' into unstable

parents d0cfb2be 02e60065
......@@ -340,6 +340,12 @@ list-max-ziplist-value 64
# set in order to use this special memory saving encoding.
set-max-intset-entries 512
# Similarly to hashes and lists, sorted sets are also specially encoded in
# order to save a lot of space. This encoding is only used when the length and
# elements of a sorted set are below the following limits:
zset-max-ziplist-entries 128
zset-max-ziplist-value 64
# Active rehashing uses 1 millisecond every 100 milliseconds of CPU time in
# order to help rehashing the main Redis hash table (the one mapping top-level
# keys to values). The hash table implementation redis uses (see dict.c)
......
......@@ -429,21 +429,55 @@ int rewriteAppendOnlyFile(char *filename) {
}
} else if (o->type == REDIS_ZSET) {
/* Emit the ZADDs needed to rebuild the sorted set */
zset *zs = o->ptr;
dictIterator *di = dictGetIterator(zs->dict);
dictEntry *de;
while((de = dictNext(di)) != NULL) {
char cmd[]="*4\r\n$4\r\nZADD\r\n";
robj *eleobj = dictGetEntryKey(de);
double *score = dictGetEntryVal(de);
if (fwrite(cmd,sizeof(cmd)-1,1,fp) == 0) goto werr;
if (fwriteBulkObject(fp,&key) == 0) goto werr;
if (fwriteBulkDouble(fp,*score) == 0) goto werr;
if (fwriteBulkObject(fp,eleobj) == 0) goto werr;
char cmd[]="*4\r\n$4\r\nZADD\r\n";
if (o->encoding == REDIS_ENCODING_ZIPLIST) {
unsigned char *zl = o->ptr;
unsigned char *eptr, *sptr;
unsigned char *vstr;
unsigned int vlen;
long long vll;
double score;
eptr = ziplistIndex(zl,0);
redisAssert(eptr != NULL);
sptr = ziplistNext(zl,eptr);
redisAssert(sptr != NULL);
while (eptr != NULL) {
redisAssert(ziplistGet(eptr,&vstr,&vlen,&vll));
score = zzlGetScore(sptr);
if (fwrite(cmd,sizeof(cmd)-1,1,fp) == 0) goto werr;
if (fwriteBulkObject(fp,&key) == 0) goto werr;
if (fwriteBulkDouble(fp,score) == 0) goto werr;
if (vstr != NULL) {
if (fwriteBulkString(fp,(char*)vstr,vlen) == 0)
goto werr;
} else {
if (fwriteBulkLongLong(fp,vll) == 0)
goto werr;
}
zzlNext(zl,&eptr,&sptr);
}
} else if (o->encoding == REDIS_ENCODING_SKIPLIST) {
zset *zs = o->ptr;
dictIterator *di = dictGetIterator(zs->dict);
dictEntry *de;
while((de = dictNext(di)) != NULL) {
robj *eleobj = dictGetEntryKey(de);
double *score = dictGetEntryVal(de);
if (fwrite(cmd,sizeof(cmd)-1,1,fp) == 0) goto werr;
if (fwriteBulkObject(fp,&key) == 0) goto werr;
if (fwriteBulkDouble(fp,*score) == 0) goto werr;
if (fwriteBulkObject(fp,eleobj) == 0) goto werr;
}
dictReleaseIterator(di);
} else {
redisPanic("Unknown sorted set encoding");
}
dictReleaseIterator(di);
} else if (o->type == REDIS_HASH) {
char cmd[]="*4\r\n$4\r\nHSET\r\n";
......
......@@ -261,6 +261,10 @@ void loadServerConfig(char *filename) {
server.list_max_ziplist_value = memtoll(argv[1], NULL);
} else if (!strcasecmp(argv[0],"set-max-intset-entries") && argc == 2) {
server.set_max_intset_entries = memtoll(argv[1], NULL);
} else if (!strcasecmp(argv[0],"zset-max-ziplist-entries") && argc == 2) {
server.zset_max_ziplist_entries = memtoll(argv[1], NULL);
} else if (!strcasecmp(argv[0],"zset-max-ziplist-value") && argc == 2) {
server.zset_max_ziplist_value = memtoll(argv[1], NULL);
} else if (!strcasecmp(argv[0],"rename-command") && argc == 3) {
struct redisCommand *cmd = lookupCommand(argv[1]);
int retval;
......@@ -450,6 +454,12 @@ void configSetCommand(redisClient *c) {
} else if (!strcasecmp(c->argv[2]->ptr,"set-max-intset-entries")) {
if (getLongLongFromObject(o,&ll) == REDIS_ERR || ll < 0) goto badfmt;
server.set_max_intset_entries = ll;
} else if (!strcasecmp(c->argv[2]->ptr,"zset-max-ziplist-entries")) {
if (getLongLongFromObject(o,&ll) == REDIS_ERR || ll < 0) goto badfmt;
server.zset_max_ziplist_entries = ll;
} else if (!strcasecmp(c->argv[2]->ptr,"zset-max-ziplist-value")) {
if (getLongLongFromObject(o,&ll) == REDIS_ERR || ll < 0) goto badfmt;
server.zset_max_ziplist_value = ll;
} else {
addReplyErrorFormat(c,"Unsupported CONFIG parameter: %s",
(char*)c->argv[2]->ptr);
......@@ -601,6 +611,16 @@ void configGetCommand(redisClient *c) {
addReplyBulkLongLong(c,server.set_max_intset_entries);
matches++;
}
if (stringmatch(pattern,"zset-max-ziplist-entries",0)) {
addReplyBulkCString(c,"zset-max-ziplist-entries");
addReplyBulkLongLong(c,server.zset_max_ziplist_entries);
matches++;
}
if (stringmatch(pattern,"zset-max-ziplist-value",0)) {
addReplyBulkCString(c,"zset-max-ziplist-value");
addReplyBulkLongLong(c,server.zset_max_ziplist_value);
matches++;
}
setDeferredMultiBulkLength(c,replylen,matches*2);
}
......
......@@ -127,22 +127,57 @@ void computeDatasetDigest(unsigned char *final) {
}
setTypeReleaseIterator(si);
} else if (o->type == REDIS_ZSET) {
zset *zs = o->ptr;
dictIterator *di = dictGetIterator(zs->dict);
dictEntry *de;
unsigned char eledigest[20];
while((de = dictNext(di)) != NULL) {
robj *eleobj = dictGetEntryKey(de);
double *score = dictGetEntryVal(de);
unsigned char eledigest[20];
if (o->encoding == REDIS_ENCODING_ZIPLIST) {
unsigned char *zl = o->ptr;
unsigned char *eptr, *sptr;
unsigned char *vstr;
unsigned int vlen;
long long vll;
double score;
snprintf(buf,sizeof(buf),"%.17g",*score);
memset(eledigest,0,20);
mixObjectDigest(eledigest,eleobj);
mixDigest(eledigest,buf,strlen(buf));
xorDigest(digest,eledigest,20);
eptr = ziplistIndex(zl,0);
redisAssert(eptr != NULL);
sptr = ziplistNext(zl,eptr);
redisAssert(sptr != NULL);
while (eptr != NULL) {
redisAssert(ziplistGet(eptr,&vstr,&vlen,&vll));
score = zzlGetScore(sptr);
memset(eledigest,0,20);
if (vstr != NULL) {
mixDigest(eledigest,vstr,vlen);
} else {
ll2string(buf,sizeof(buf),vll);
mixDigest(eledigest,buf,strlen(buf));
}
snprintf(buf,sizeof(buf),"%.17g",score);
mixDigest(eledigest,buf,strlen(buf));
xorDigest(digest,eledigest,20);
zzlNext(zl,&eptr,&sptr);
}
} else if (o->encoding == REDIS_ENCODING_SKIPLIST) {
zset *zs = o->ptr;
dictIterator *di = dictGetIterator(zs->dict);
dictEntry *de;
while((de = dictNext(di)) != NULL) {
robj *eleobj = dictGetEntryKey(de);
double *score = dictGetEntryVal(de);
snprintf(buf,sizeof(buf),"%.17g",*score);
memset(eledigest,0,20);
mixObjectDigest(eledigest,eleobj);
mixDigest(eledigest,buf,strlen(buf));
xorDigest(digest,eledigest,20);
}
dictReleaseIterator(di);
} else {
redisPanic("Unknown sorted set encoding");
}
dictReleaseIterator(di);
} else if (o->type == REDIS_HASH) {
hashTypeIterator *hi;
robj *obj;
......
......@@ -102,6 +102,13 @@ robj *createZsetObject(void) {
return o;
}
robj *createZsetZiplistObject(void) {
unsigned char *zl = ziplistNew();
robj *o = createObject(REDIS_ZSET,zl);
o->encoding = REDIS_ENCODING_ZIPLIST;
return o;
}
void freeStringObject(robj *o) {
if (o->encoding == REDIS_ENCODING_RAW) {
sdsfree(o->ptr);
......@@ -135,11 +142,20 @@ void freeSetObject(robj *o) {
}
void freeZsetObject(robj *o) {
zset *zs = o->ptr;
dictRelease(zs->dict);
zslFree(zs->zsl);
zfree(zs);
zset *zs;
switch (o->encoding) {
case REDIS_ENCODING_SKIPLIST:
zs = o->ptr;
dictRelease(zs->dict);
zslFree(zs->zsl);
zfree(zs);
break;
case REDIS_ENCODING_ZIPLIST:
zfree(o->ptr);
break;
default:
redisPanic("Unknown sorted set encoding");
}
}
void freeHashObject(robj *o) {
......
......@@ -302,24 +302,33 @@ int rdbSaveObject(FILE *fp, robj *o) {
redisPanic("Unknown set encoding");
}
} else if (o->type == REDIS_ZSET) {
/* Save a set value */
zset *zs = o->ptr;
dictIterator *di = dictGetIterator(zs->dict);
dictEntry *de;
if ((n = rdbSaveLen(fp,dictSize(zs->dict))) == -1) return -1;
nwritten += n;
while((de = dictNext(di)) != NULL) {
robj *eleobj = dictGetEntryKey(de);
double *score = dictGetEntryVal(de);
/* Save a sorted set value */
if (o->encoding == REDIS_ENCODING_ZIPLIST) {
size_t l = ziplistBlobLen((unsigned char*)o->ptr);
if ((n = rdbSaveStringObject(fp,eleobj)) == -1) return -1;
if ((n = rdbSaveRawString(fp,o->ptr,l)) == -1) return -1;
nwritten += n;
if ((n = rdbSaveDoubleValue(fp,*score)) == -1) return -1;
} else if (o->encoding == REDIS_ENCODING_SKIPLIST) {
zset *zs = o->ptr;
dictIterator *di = dictGetIterator(zs->dict);
dictEntry *de;
if ((n = rdbSaveLen(fp,dictSize(zs->dict))) == -1) return -1;
nwritten += n;
while((de = dictNext(di)) != NULL) {
robj *eleobj = dictGetEntryKey(de);
double *score = dictGetEntryVal(de);
if ((n = rdbSaveStringObject(fp,eleobj)) == -1) return -1;
nwritten += n;
if ((n = rdbSaveDoubleValue(fp,*score)) == -1) return -1;
nwritten += n;
}
dictReleaseIterator(di);
} else {
redisPanic("Unknown sorted set encoding");
}
dictReleaseIterator(di);
} else if (o->type == REDIS_HASH) {
/* Save a hash value */
if (o->encoding == REDIS_ENCODING_ZIPMAP) {
......@@ -386,6 +395,8 @@ int rdbSaveKeyValuePair(FILE *fp, robj *key, robj *val,
vtype = REDIS_LIST_ZIPLIST;
else if (vtype == REDIS_SET && val->encoding == REDIS_ENCODING_INTSET)
vtype = REDIS_SET_INTSET;
else if (vtype == REDIS_ZSET && val->encoding == REDIS_ENCODING_ZIPLIST)
vtype = REDIS_ZSET_ZIPLIST;
/* Save type, key, value */
if (rdbSaveType(fp,vtype) == -1) return -1;
if (rdbSaveStringObject(fp,key) == -1) return -1;
......@@ -745,11 +756,13 @@ robj *rdbLoadObject(int type, FILE *fp) {
} else if (type == REDIS_ZSET) {
/* Read list/set value */
size_t zsetlen;
size_t maxelelen = 0;
zset *zs;
if ((zsetlen = rdbLoadLen(fp,NULL)) == REDIS_RDB_LENERR) return NULL;
o = createZsetObject();
zs = o->ptr;
/* Load every single element of the list/set */
while(zsetlen--) {
robj *ele;
......@@ -759,10 +772,21 @@ robj *rdbLoadObject(int type, FILE *fp) {
if ((ele = rdbLoadEncodedStringObject(fp)) == NULL) return NULL;
ele = tryObjectEncoding(ele);
if (rdbLoadDoubleValue(fp,&score) == -1) return NULL;
/* Don't care about integer-encoded strings. */
if (ele->encoding == REDIS_ENCODING_RAW &&
sdslen(ele->ptr) > maxelelen)
maxelelen = sdslen(ele->ptr);
znode = zslInsert(zs->zsl,score,ele);
dictAdd(zs->dict,ele,&znode->score);
incrRefCount(ele); /* added to skiplist */
}
/* Convert *after* loading, since sorted sets are not stored ordered. */
if (zsetLength(o) <= server.zset_max_ziplist_entries &&
maxelelen <= server.zset_max_ziplist_value)
zsetConvert(o,REDIS_ENCODING_ZIPLIST);
} else if (type == REDIS_HASH) {
size_t hashlen;
......@@ -811,7 +835,8 @@ robj *rdbLoadObject(int type, FILE *fp) {
}
} else if (type == REDIS_HASH_ZIPMAP ||
type == REDIS_LIST_ZIPLIST ||
type == REDIS_SET_INTSET)
type == REDIS_SET_INTSET ||
type == REDIS_ZSET_ZIPLIST)
{
robj *aux = rdbLoadStringObject(fp);
......@@ -846,8 +871,14 @@ robj *rdbLoadObject(int type, FILE *fp) {
if (intsetLen(o->ptr) > server.set_max_intset_entries)
setTypeConvert(o,REDIS_ENCODING_HT);
break;
case REDIS_ZSET_ZIPLIST:
o->type = REDIS_ZSET;
o->encoding = REDIS_ENCODING_ZIPLIST;
if (zsetLength(o) > server.zset_max_ziplist_entries)
zsetConvert(o,REDIS_ENCODING_SKIPLIST);
break;
default:
redisPanic("Unknown enoding");
redisPanic("Unknown encoding");
break;
}
} else {
......
......@@ -846,6 +846,8 @@ void initServerConfig() {
server.list_max_ziplist_entries = REDIS_LIST_MAX_ZIPLIST_ENTRIES;
server.list_max_ziplist_value = REDIS_LIST_MAX_ZIPLIST_VALUE;
server.set_max_intset_entries = REDIS_SET_MAX_INTSET_ENTRIES;
server.zset_max_ziplist_entries = REDIS_ZSET_MAX_ZIPLIST_ENTRIES;
server.zset_max_ziplist_value = REDIS_ZSET_MAX_ZIPLIST_VALUE;
server.shutdown_asap = 0;
server.cache_flush_delay = 0;
server.cluster_enabled = 0;
......
......@@ -71,10 +71,12 @@
#define REDIS_ZSET 3
#define REDIS_HASH 4
#define REDIS_VMPOINTER 8
/* Object types only used for persistence in .rdb files */
#define REDIS_HASH_ZIPMAP 9
#define REDIS_LIST_ZIPLIST 10
#define REDIS_SET_INTSET 11
#define REDIS_ZSET_ZIPLIST 12
/* Objects encoding. Some kind of objects like Strings and Hashes can be
* internally represented in multiple ways. The 'encoding' field of the object
......@@ -198,6 +200,8 @@
#define REDIS_LIST_MAX_ZIPLIST_ENTRIES 512
#define REDIS_LIST_MAX_ZIPLIST_VALUE 64
#define REDIS_SET_MAX_INTSET_ENTRIES 512
#define REDIS_ZSET_MAX_ZIPLIST_ENTRIES 128
#define REDIS_ZSET_MAX_ZIPLIST_VALUE 64
/* Sets operations codes */
#define REDIS_OP_UNION 0
......@@ -589,6 +593,8 @@ struct redisServer {
size_t list_max_ziplist_entries;
size_t list_max_ziplist_value;
size_t set_max_intset_entries;
size_t zset_max_ziplist_entries;
size_t zset_max_ziplist_value;
time_t unixtime; /* Unix time sampled every second. */
/* Virtual memory I/O threads stuff */
/* An I/O thread process an element taken from the io_jobs queue and
......@@ -853,6 +859,7 @@ robj *createSetObject(void);
robj *createIntsetObject(void);
robj *createHashObject(void);
robj *createZsetObject(void);
robj *createZsetZiplistObject(void);
int getLongFromObjectOrReply(redisClient *c, robj *o, long *target, const char *msg);
int checkType(redisClient *c, robj *o, int type);
int getLongLongFromObjectOrReply(redisClient *c, robj *o, long long *target, const char *msg);
......@@ -916,6 +923,12 @@ void backgroundRewriteDoneHandler(int exitcode, int bysignal);
zskiplist *zslCreate(void);
void zslFree(zskiplist *zsl);
zskiplistNode *zslInsert(zskiplist *zsl, double score, robj *obj);
unsigned char *zzlInsert(unsigned char *zl, robj *ele, double score);
double zzlGetScore(unsigned char *sptr);
void zzlNext(unsigned char *zl, unsigned char **eptr, unsigned char **sptr);
void zzlPrev(unsigned char *zl, unsigned char **eptr, unsigned char **sptr);
unsigned int zsetLength(robj *zobj);
void zsetConvert(robj *zobj, int encoding);
/* Core functions */
void freeMemoryIfNeeded(void);
......@@ -1009,6 +1022,8 @@ int stringmatchlen(const char *pattern, int patternLen,
int stringmatch(const char *pattern, const char *string, int nocase);
long long memtoll(const char *p, int *err);
int ll2string(char *s, size_t len, long long value);
int string2ll(char *s, size_t len, long long *value);
int d2string(char *s, size_t len, double value);
int isStringRepresentableAsLong(sds s, long *longval);
int isStringRepresentableAsLongLong(sds s, long long *longval);
int isObjectRepresentableAsLongLong(robj *o, long long *llongval);
......
......@@ -199,6 +199,9 @@ void sortCommand(redisClient *c) {
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;
......
......@@ -180,6 +180,82 @@ typedef struct {
int minex, maxex; /* are min or max exclusive? */
} zrangespec;
static int zslValueGteMin(double value, zrangespec *spec) {
return spec->minex ? (value > spec->min) : (value >= spec->min);
}
static int zslValueLteMax(double value, zrangespec *spec) {
return spec->maxex ? (value < spec->max) : (value <= spec->max);
}
/* Returns if there is a part of the zset is in range. */
int zslIsInRange(zskiplist *zsl, zrangespec *range) {
zskiplistNode *x;
/* Test for ranges that will always be empty. */
if (range->min > range->max ||
(range->min == range->max && (range->minex || range->maxex)))
return 0;
x = zsl->tail;
if (x == NULL || !zslValueGteMin(x->score,range))
return 0;
x = zsl->header->level[0].forward;
if (x == NULL || !zslValueLteMax(x->score,range))
return 0;
return 1;
}
/* Find the first node that is contained in the specified range.
* Returns NULL when no element is contained in the range. */
zskiplistNode *zslFirstInRange(zskiplist *zsl, zrangespec range) {
zskiplistNode *x;
int i;
/* If everything is out of range, return early. */
if (!zslIsInRange(zsl,&range)) return NULL;
x = zsl->header;
for (i = zsl->level-1; i >= 0; i--) {
/* Go forward while *OUT* of range. */
while (x->level[i].forward &&
!zslValueGteMin(x->level[i].forward->score,&range))
x = x->level[i].forward;
}
/* This is an inner range, so the next node cannot be NULL. */
x = x->level[0].forward;
redisAssert(x != NULL);
/* Check if score <= max. */
if (!zslValueLteMax(x->score,&range)) return NULL;
return x;
}
/* Find the last node that is contained in the specified range.
* Returns NULL when no element is contained in the range. */
zskiplistNode *zslLastInRange(zskiplist *zsl, zrangespec range) {
zskiplistNode *x;
int i;
/* If everything is out of range, return early. */
if (!zslIsInRange(zsl,&range)) return NULL;
x = zsl->header;
for (i = zsl->level-1; i >= 0; i--) {
/* Go forward while *IN* range. */
while (x->level[i].forward &&
zslValueLteMax(x->level[i].forward->score,&range))
x = x->level[i].forward;
}
/* This is an inner range, so this node cannot be NULL. */
redisAssert(x != NULL);
/* Check if score >= min. */
if (!zslValueGteMin(x->score,&range)) return NULL;
return x;
}
/* Delete all the elements with score between min and max from the skiplist.
* Min and mx are inclusive, so a score >= min || score <= max is deleted.
* Note that this function takes the reference to the hash table view of the
......@@ -243,27 +319,11 @@ unsigned long zslDeleteRangeByRank(zskiplist *zsl, unsigned int start, unsigned
return removed;
}
/* Find the first node having a score equal or greater than the specified one.
* Returns NULL if there is no match. */
zskiplistNode *zslFirstWithScore(zskiplist *zsl, double score) {
zskiplistNode *x;
int i;
x = zsl->header;
for (i = zsl->level-1; i >= 0; i--) {
while (x->level[i].forward && x->level[i].forward->score < score)
x = x->level[i].forward;
}
/* We may have multiple elements with the same score, what we need
* is to find the element with both the right score and object. */
return x->level[0].forward;
}
/* Find the rank for an element by both score and key.
* Returns 0 when the element cannot be found, rank otherwise.
* Note that the rank is 1-based due to the span of zsl->header to the
* first element. */
unsigned long zslistTypeGetRank(zskiplist *zsl, double score, robj *o) {
unsigned long zslGetRank(zskiplist *zsl, double score, robj *o) {
zskiplistNode *x;
unsigned long rank = 0;
int i;
......@@ -287,7 +347,7 @@ unsigned long zslistTypeGetRank(zskiplist *zsl, double score, robj *o) {
}
/* Finds an element by its rank. The rank argument needs to be 1-based. */
zskiplistNode* zslistTypeGetElementByRank(zskiplist *zsl, unsigned long rank) {
zskiplistNode* zslGetElementByRank(zskiplist *zsl, unsigned long rank) {
zskiplistNode *x;
unsigned long traversed = 0;
int i;
......@@ -343,150 +403,603 @@ static int zslParseRange(robj *min, robj *max, zrangespec *spec) {
return REDIS_OK;
}
/*-----------------------------------------------------------------------------
* Ziplist-backed sorted set API
*----------------------------------------------------------------------------*/
double zzlGetScore(unsigned char *sptr) {
unsigned char *vstr;
unsigned int vlen;
long long vlong;
char buf[128];
double score;
redisAssert(sptr != NULL);
redisAssert(ziplistGet(sptr,&vstr,&vlen,&vlong));
if (vstr) {
memcpy(buf,vstr,vlen);
buf[vlen] = '\0';
score = strtod(buf,NULL);
} else {
score = vlong;
}
return score;
}
/* Compare element in sorted set with given element. */
int zzlCompareElements(unsigned char *eptr, unsigned char *cstr, unsigned int clen) {
unsigned char *vstr;
unsigned int vlen;
long long vlong;
unsigned char vbuf[32];
int minlen, cmp;
redisAssert(ziplistGet(eptr,&vstr,&vlen,&vlong));
if (vstr == NULL) {
/* Store string representation of long long in buf. */
vlen = ll2string((char*)vbuf,sizeof(vbuf),vlong);
vstr = vbuf;
}
minlen = (vlen < clen) ? vlen : clen;
cmp = memcmp(vstr,cstr,minlen);
if (cmp == 0) return vlen-clen;
return cmp;
}
unsigned int zzlLength(unsigned char *zl) {
return ziplistLen(zl)/2;
}
/* Move to next entry based on the values in eptr and sptr. Both are set to
* NULL when there is no next entry. */
void zzlNext(unsigned char *zl, unsigned char **eptr, unsigned char **sptr) {
unsigned char *_eptr, *_sptr;
redisAssert(*eptr != NULL && *sptr != NULL);
_eptr = ziplistNext(zl,*sptr);
if (_eptr != NULL) {
_sptr = ziplistNext(zl,_eptr);
redisAssert(_sptr != NULL);
} else {
/* No next entry. */
_sptr = NULL;
}
*eptr = _eptr;
*sptr = _sptr;
}
/* Move to the previous entry based on the values in eptr and sptr. Both are
* set to NULL when there is no next entry. */
void zzlPrev(unsigned char *zl, unsigned char **eptr, unsigned char **sptr) {
unsigned char *_eptr, *_sptr;
redisAssert(*eptr != NULL && *sptr != NULL);
_sptr = ziplistPrev(zl,*eptr);
if (_sptr != NULL) {
_eptr = ziplistPrev(zl,_sptr);
redisAssert(_eptr != NULL);
} else {
/* No previous entry. */
_eptr = NULL;
}
*eptr = _eptr;
*sptr = _sptr;
}
/* Returns if there is a part of the zset is in range. Should only be used
* internally by zzlFirstInRange and zzlLastInRange. */
int zzlIsInRange(unsigned char *zl, zrangespec *range) {
unsigned char *p;
double score;
/* Test for ranges that will always be empty. */
if (range->min > range->max ||
(range->min == range->max && (range->minex || range->maxex)))
return 0;
p = ziplistIndex(zl,-1); /* Last score. */
redisAssert(p != NULL);
score = zzlGetScore(p);
if (!zslValueGteMin(score,range))
return 0;
p = ziplistIndex(zl,1); /* First score. */
redisAssert(p != NULL);
score = zzlGetScore(p);
if (!zslValueLteMax(score,range))
return 0;
return 1;
}
/* Find pointer to the first element contained in the specified range.
* Returns NULL when no element is contained in the range. */
unsigned char *zzlFirstInRange(unsigned char *zl, zrangespec range) {
unsigned char *eptr = ziplistIndex(zl,0), *sptr;
double score;
/* If everything is out of range, return early. */
if (!zzlIsInRange(zl,&range)) return NULL;
while (eptr != NULL) {
sptr = ziplistNext(zl,eptr);
redisAssert(sptr != NULL);
score = zzlGetScore(sptr);
if (zslValueGteMin(score,&range)) {
/* Check if score <= max. */
if (zslValueLteMax(score,&range))
return eptr;
return NULL;
}
/* Move to next element. */
eptr = ziplistNext(zl,sptr);
}
return NULL;
}
/* Find pointer to the last element contained in the specified range.
* Returns NULL when no element is contained in the range. */
unsigned char *zzlLastInRange(unsigned char *zl, zrangespec range) {
unsigned char *eptr = ziplistIndex(zl,-2), *sptr;
double score;
/* If everything is out of range, return early. */
if (!zzlIsInRange(zl,&range)) return NULL;
while (eptr != NULL) {
sptr = ziplistNext(zl,eptr);
redisAssert(sptr != NULL);
score = zzlGetScore(sptr);
if (zslValueLteMax(score,&range)) {
/* Check if score >= min. */
if (zslValueGteMin(score,&range))
return eptr;
return NULL;
}
/* Move to previous element by moving to the score of previous element.
* When this returns NULL, we know there also is no element. */
sptr = ziplistPrev(zl,eptr);
if (sptr != NULL)
redisAssert((eptr = ziplistPrev(zl,sptr)) != NULL);
else
eptr = NULL;
}
return NULL;
}
unsigned char *zzlFind(unsigned char *zl, robj *ele, double *score) {
unsigned char *eptr = ziplistIndex(zl,0), *sptr;
ele = getDecodedObject(ele);
while (eptr != NULL) {
sptr = ziplistNext(zl,eptr);
redisAssert(sptr != NULL);
if (ziplistCompare(eptr,ele->ptr,sdslen(ele->ptr))) {
/* Matching element, pull out score. */
if (score != NULL) *score = zzlGetScore(sptr);
decrRefCount(ele);
return eptr;
}
/* Move to next element. */
eptr = ziplistNext(zl,sptr);
}
decrRefCount(ele);
return NULL;
}
/* Delete (element,score) pair from ziplist. Use local copy of eptr because we
* don't want to modify the one given as argument. */
unsigned char *zzlDelete(unsigned char *zl, unsigned char *eptr) {
unsigned char *p = eptr;
/* TODO: add function to ziplist API to delete N elements from offset. */
zl = ziplistDelete(zl,&p);
zl = ziplistDelete(zl,&p);
return zl;
}
unsigned char *zzlInsertAt(unsigned char *zl, unsigned char *eptr, robj *ele, double score) {
unsigned char *sptr;
char scorebuf[128];
int scorelen;
size_t offset;
redisAssert(ele->encoding == REDIS_ENCODING_RAW);
scorelen = d2string(scorebuf,sizeof(scorebuf),score);
if (eptr == NULL) {
zl = ziplistPush(zl,ele->ptr,sdslen(ele->ptr),ZIPLIST_TAIL);
zl = ziplistPush(zl,(unsigned char*)scorebuf,scorelen,ZIPLIST_TAIL);
} else {
/* Keep offset relative to zl, as it might be re-allocated. */
offset = eptr-zl;
zl = ziplistInsert(zl,eptr,ele->ptr,sdslen(ele->ptr));
eptr = zl+offset;
/* Insert score after the element. */
redisAssert((sptr = ziplistNext(zl,eptr)) != NULL);
zl = ziplistInsert(zl,sptr,(unsigned char*)scorebuf,scorelen);
}
return zl;
}
/* Insert (element,score) pair in ziplist. This function assumes the element is
* not yet present in the list. */
unsigned char *zzlInsert(unsigned char *zl, robj *ele, double score) {
unsigned char *eptr = ziplistIndex(zl,0), *sptr;
double s;
ele = getDecodedObject(ele);
while (eptr != NULL) {
sptr = ziplistNext(zl,eptr);
redisAssert(sptr != NULL);
s = zzlGetScore(sptr);
if (s > score) {
/* First element with score larger than score for element to be
* inserted. This means we should take its spot in the list to
* maintain ordering. */
zl = zzlInsertAt(zl,eptr,ele,score);
break;
} else if (s == score) {
/* Ensure lexicographical ordering for elements. */
if (zzlCompareElements(eptr,ele->ptr,sdslen(ele->ptr)) > 0) {
zl = zzlInsertAt(zl,eptr,ele,score);
break;
}
}
/* Move to next element. */
eptr = ziplistNext(zl,sptr);
}
/* Push on tail of list when it was not yet inserted. */
if (eptr == NULL)
zl = zzlInsertAt(zl,NULL,ele,score);
decrRefCount(ele);
return zl;
}
unsigned char *zzlDeleteRangeByScore(unsigned char *zl, zrangespec range, unsigned long *deleted) {
unsigned char *eptr, *sptr;
double score;
unsigned long num = 0;
if (deleted != NULL) *deleted = 0;
eptr = zzlFirstInRange(zl,range);
if (eptr == NULL) return zl;
/* When the tail of the ziplist is deleted, eptr will point to the sentinel
* byte and ziplistNext will return NULL. */
while ((sptr = ziplistNext(zl,eptr)) != NULL) {
score = zzlGetScore(sptr);
if (zslValueLteMax(score,&range)) {
/* Delete both the element and the score. */
zl = ziplistDelete(zl,&eptr);
zl = ziplistDelete(zl,&eptr);
num++;
} else {
/* No longer in range. */
break;
}
}
if (deleted != NULL) *deleted = num;
return zl;
}
/* Delete all the elements with rank between start and end from the skiplist.
* Start and end are inclusive. Note that start and end need to be 1-based */
unsigned char *zzlDeleteRangeByRank(unsigned char *zl, unsigned int start, unsigned int end, unsigned long *deleted) {
unsigned int num = (end-start)+1;
if (deleted) *deleted = num;
zl = ziplistDeleteRange(zl,2*(start-1),2*num);
return zl;
}
/*-----------------------------------------------------------------------------
* Common sorted set API
*----------------------------------------------------------------------------*/
unsigned int zsetLength(robj *zobj) {
int length = -1;
if (zobj->encoding == REDIS_ENCODING_ZIPLIST) {
length = zzlLength(zobj->ptr);
} else if (zobj->encoding == REDIS_ENCODING_SKIPLIST) {
length = ((zset*)zobj->ptr)->zsl->length;
} else {
redisPanic("Unknown sorted set encoding");
}
return length;
}
void zsetConvert(robj *zobj, int encoding) {
zset *zs;
zskiplistNode *node, *next;
robj *ele;
double score;
if (zobj->encoding == encoding) return;
if (zobj->encoding == REDIS_ENCODING_ZIPLIST) {
unsigned char *zl = zobj->ptr;
unsigned char *eptr, *sptr;
unsigned char *vstr;
unsigned int vlen;
long long vlong;
if (encoding != REDIS_ENCODING_SKIPLIST)
redisPanic("Unknown target encoding");
zs = zmalloc(sizeof(*zs));
zs->dict = dictCreate(&zsetDictType,NULL);
zs->zsl = zslCreate();
eptr = ziplistIndex(zl,0);
redisAssert(eptr != NULL);
sptr = ziplistNext(zl,eptr);
redisAssert(sptr != NULL);
while (eptr != NULL) {
score = zzlGetScore(sptr);
redisAssert(ziplistGet(eptr,&vstr,&vlen,&vlong));
if (vstr == NULL)
ele = createStringObjectFromLongLong(vlong);
else
ele = createStringObject((char*)vstr,vlen);
/* Has incremented refcount since it was just created. */
node = zslInsert(zs->zsl,score,ele);
redisAssert(dictAdd(zs->dict,ele,&node->score) == DICT_OK);
incrRefCount(ele); /* Added to dictionary. */
zzlNext(zl,&eptr,&sptr);
}
zfree(zobj->ptr);
zobj->ptr = zs;
zobj->encoding = REDIS_ENCODING_SKIPLIST;
} else if (zobj->encoding == REDIS_ENCODING_SKIPLIST) {
unsigned char *zl = ziplistNew();
if (encoding != REDIS_ENCODING_ZIPLIST)
redisPanic("Unknown target encoding");
/* Approach similar to zslFree(), since we want to free the skiplist at
* the same time as creating the ziplist. */
zs = zobj->ptr;
dictRelease(zs->dict);
node = zs->zsl->header->level[0].forward;
zfree(zs->zsl->header);
zfree(zs->zsl);
while (node) {
ele = getDecodedObject(node->obj);
zl = zzlInsertAt(zl,NULL,ele,node->score);
decrRefCount(ele);
next = node->level[0].forward;
zslFreeNode(node);
node = next;
}
zfree(zs);
zobj->ptr = zl;
zobj->encoding = REDIS_ENCODING_ZIPLIST;
} else {
redisPanic("Unknown sorted set encoding");
}
}
/*-----------------------------------------------------------------------------
* Sorted set commands
*----------------------------------------------------------------------------*/
/* This generic command implements both ZADD and ZINCRBY. */
void zaddGenericCommand(redisClient *c, robj *key, robj *ele, double score, int incr) {
robj *zsetobj;
zset *zs;
zskiplistNode *znode;
void zaddGenericCommand(redisClient *c, int incr) {
static char *nanerr = "resulting score is not a number (NaN)";
robj *key = c->argv[1];
robj *ele;
robj *zobj;
robj *curobj;
double score, curscore = 0.0;
zsetobj = lookupKeyWrite(c->db,key);
if (zsetobj == NULL) {
zsetobj = createZsetObject();
dbAdd(c->db,key,zsetobj);
if (getDoubleFromObjectOrReply(c,c->argv[2],&score,NULL) != REDIS_OK)
return;
zobj = lookupKeyWrite(c->db,key);
if (zobj == NULL) {
if (server.zset_max_ziplist_entries == 0 ||
server.zset_max_ziplist_value < sdslen(c->argv[3]->ptr))
{
zobj = createZsetObject();
} else {
zobj = createZsetZiplistObject();
}
dbAdd(c->db,key,zobj);
} else {
if (zsetobj->type != REDIS_ZSET) {
if (zobj->type != REDIS_ZSET) {
addReply(c,shared.wrongtypeerr);
return;
}
}
zs = zsetobj->ptr;
/* Since both ZADD and ZINCRBY are implemented here, we need to increment
* the score first by the current score if ZINCRBY is called. */
if (incr) {
/* Read the old score. If the element was not present starts from 0 */
dictEntry *de = dictFind(zs->dict,ele);
if (de != NULL)
score += *(double*)dictGetEntryVal(de);
if (isnan(score)) {
addReplyError(c,"resulting score is not a number (NaN)");
/* Note that we don't need to check if the zset may be empty and
* should be removed here, as we can only obtain Nan as score if
* there was already an element in the sorted set. */
return;
}
}
/* We need to remove and re-insert the element when it was already present
* in the dictionary, to update the skiplist. Note that we delay adding a
* pointer to the score because we want to reference the score in the
* skiplist node. */
if (dictAdd(zs->dict,ele,NULL) == DICT_OK) {
dictEntry *de;
if (zobj->encoding == REDIS_ENCODING_ZIPLIST) {
unsigned char *eptr;
/* Prefer non-encoded element when dealing with ziplists. */
ele = c->argv[3];
if ((eptr = zzlFind(zobj->ptr,ele,&curscore)) != NULL) {
if (incr) {
score += curscore;
if (isnan(score)) {
addReplyError(c,nanerr);
/* Don't need to check if the sorted set is empty, because
* we know it has at least one element. */
return;
}
}
/* New element */
incrRefCount(ele); /* added to hash */
znode = zslInsert(zs->zsl,score,ele);
incrRefCount(ele); /* added to skiplist */
/* Remove and re-insert when score changed. */
if (score != curscore) {
zobj->ptr = zzlDelete(zobj->ptr,eptr);
zobj->ptr = zzlInsert(zobj->ptr,ele,score);
/* Update the score in the dict entry */
de = dictFind(zs->dict,ele);
redisAssert(de != NULL);
dictGetEntryVal(de) = &znode->score;
signalModifiedKey(c->db,c->argv[1]);
server.dirty++;
if (incr)
addReplyDouble(c,score);
else
addReply(c,shared.cone);
} else {
signalModifiedKey(c->db,key);
server.dirty++;
}
if (incr) /* ZINCRBY */
addReplyDouble(c,score);
else /* ZADD */
addReply(c,shared.czero);
} else {
/* Optimize: check if the element is too large or the list becomes
* too long *before* executing zzlInsert. */
zobj->ptr = zzlInsert(zobj->ptr,ele,score);
if (zzlLength(zobj->ptr) > server.zset_max_ziplist_entries)
zsetConvert(zobj,REDIS_ENCODING_SKIPLIST);
if (sdslen(ele->ptr) > server.zset_max_ziplist_value)
zsetConvert(zobj,REDIS_ENCODING_SKIPLIST);
signalModifiedKey(c->db,key);
server.dirty++;
if (incr) /* ZINCRBY */
addReplyDouble(c,score);
else /* ZADD */
addReply(c,shared.cone);
}
} else if (zobj->encoding == REDIS_ENCODING_SKIPLIST) {
zset *zs = zobj->ptr;
zskiplistNode *znode;
dictEntry *de;
robj *curobj;
double *curscore;
int deleted;
/* Update score */
ele = c->argv[3] = tryObjectEncoding(c->argv[3]);
de = dictFind(zs->dict,ele);
redisAssert(de != NULL);
curobj = dictGetEntryKey(de);
curscore = dictGetEntryVal(de);
/* When the score is updated, reuse the existing string object to
* prevent extra alloc/dealloc of strings on ZINCRBY. */
if (score != *curscore) {
deleted = zslDelete(zs->zsl,*curscore,curobj);
redisAssert(deleted != 0);
znode = zslInsert(zs->zsl,score,curobj);
incrRefCount(curobj);
/* Update the score in the current dict entry */
dictGetEntryVal(de) = &znode->score;
signalModifiedKey(c->db,c->argv[1]);
if (de != NULL) {
curobj = dictGetEntryKey(de);
curscore = *(double*)dictGetEntryVal(de);
if (incr) {
score += curscore;
if (isnan(score)) {
addReplyError(c,nanerr);
/* Don't need to check if the sorted set is empty, because
* we know it has at least one element. */
return;
}
}
/* Remove and re-insert when score changed. We can safely delete
* the key object from the skiplist, since the dictionary still has
* a reference to it. */
if (score != curscore) {
redisAssert(zslDelete(zs->zsl,curscore,curobj));
znode = zslInsert(zs->zsl,score,curobj);
incrRefCount(curobj); /* Re-inserted in skiplist. */
dictGetEntryVal(de) = &znode->score; /* Update score ptr. */
signalModifiedKey(c->db,key);
server.dirty++;
}
if (incr) /* ZINCRBY */
addReplyDouble(c,score);
else /* ZADD */
addReply(c,shared.czero);
} else {
znode = zslInsert(zs->zsl,score,ele);
incrRefCount(ele); /* Inserted in skiplist. */
redisAssert(dictAdd(zs->dict,ele,&znode->score) == DICT_OK);
incrRefCount(ele); /* Added to dictionary. */
signalModifiedKey(c->db,key);
server.dirty++;
if (incr) /* ZINCRBY */
addReplyDouble(c,score);
else /* ZADD */
addReply(c,shared.cone);
}
if (incr)
addReplyDouble(c,score);
else
addReply(c,shared.czero);
} else {
redisPanic("Unknown sorted set encoding");
}
}
void zaddCommand(redisClient *c) {
double scoreval;
if (getDoubleFromObjectOrReply(c,c->argv[2],&scoreval,NULL) != REDIS_OK) return;
c->argv[3] = tryObjectEncoding(c->argv[3]);
zaddGenericCommand(c,c->argv[1],c->argv[3],scoreval,0);
zaddGenericCommand(c,0);
}
void zincrbyCommand(redisClient *c) {
double scoreval;
if (getDoubleFromObjectOrReply(c,c->argv[2],&scoreval,NULL) != REDIS_OK) return;
c->argv[3] = tryObjectEncoding(c->argv[3]);
zaddGenericCommand(c,c->argv[1],c->argv[3],scoreval,1);
zaddGenericCommand(c,1);
}
void zremCommand(redisClient *c) {
robj *zsetobj;
zset *zs;
dictEntry *de;
double curscore;
int deleted;
robj *key = c->argv[1];
robj *ele = c->argv[2];
robj *zobj;
if ((zsetobj = lookupKeyWriteOrReply(c,c->argv[1],shared.czero)) == NULL ||
checkType(c,zsetobj,REDIS_ZSET)) return;
if ((zobj = lookupKeyWriteOrReply(c,key,shared.czero)) == NULL ||
checkType(c,zobj,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;
if (zobj->encoding == REDIS_ENCODING_ZIPLIST) {
unsigned char *eptr;
if ((eptr = zzlFind(zobj->ptr,ele,NULL)) != NULL) {
zobj->ptr = zzlDelete(zobj->ptr,eptr);
if (zzlLength(zobj->ptr) == 0) dbDelete(c->db,key);
} else {
addReply(c,shared.czero);
return;
}
} else if (zobj->encoding == REDIS_ENCODING_SKIPLIST) {
zset *zs = zobj->ptr;
dictEntry *de;
double score;
de = dictFind(zs->dict,ele);
if (de != NULL) {
/* Delete from the skiplist */
score = *(double*)dictGetEntryVal(de);
redisAssert(zslDelete(zs->zsl,score,ele));
/* Delete from the hash table */
dictDelete(zs->dict,ele);
if (htNeedsResize(zs->dict)) dictResize(zs->dict);
if (dictSize(zs->dict) == 0) dbDelete(c->db,key);
} else {
addReply(c,shared.czero);
return;
}
} else {
redisPanic("Unknown sorted set encoding");
}
/* 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]);
signalModifiedKey(c->db,c->argv[1]);
signalModifiedKey(c->db,key);
server.dirty++;
addReply(c,shared.cone);
}
void zremrangebyscoreCommand(redisClient *c) {
robj *key = c->argv[1];
robj *zobj;
zrangespec range;
long deleted;
robj *o;
zset *zs;
unsigned long deleted;
/* Parse the range arguments. */
if (zslParseRange(c->argv[2],c->argv[3],&range) != REDIS_OK) {
......@@ -494,35 +1007,42 @@ void zremrangebyscoreCommand(redisClient *c) {
return;
}
if ((o = lookupKeyWriteOrReply(c,c->argv[1],shared.czero)) == NULL ||
checkType(c,o,REDIS_ZSET)) return;
if ((zobj = lookupKeyWriteOrReply(c,key,shared.czero)) == NULL ||
checkType(c,zobj,REDIS_ZSET)) return;
if (zobj->encoding == REDIS_ENCODING_ZIPLIST) {
zobj->ptr = zzlDeleteRangeByScore(zobj->ptr,range,&deleted);
if (zzlLength(zobj->ptr) == 0) dbDelete(c->db,key);
} else if (zobj->encoding == REDIS_ENCODING_SKIPLIST) {
zset *zs = zobj->ptr;
deleted = zslDeleteRangeByScore(zs->zsl,range,zs->dict);
if (htNeedsResize(zs->dict)) dictResize(zs->dict);
if (dictSize(zs->dict) == 0) dbDelete(c->db,key);
} else {
redisPanic("Unknown sorted set encoding");
}
zs = o->ptr;
deleted = zslDeleteRangeByScore(zs->zsl,range,zs->dict);
if (htNeedsResize(zs->dict)) dictResize(zs->dict);
if (dictSize(zs->dict) == 0) dbDelete(c->db,c->argv[1]);
if (deleted) signalModifiedKey(c->db,c->argv[1]);
if (deleted) signalModifiedKey(c->db,key);
server.dirty += deleted;
addReplyLongLong(c,deleted);
}
void zremrangebyrankCommand(redisClient *c) {
robj *key = c->argv[1];
robj *zobj;
long start;
long end;
int llen;
long deleted;
robj *zsetobj;
zset *zs;
unsigned long deleted;
if ((getLongFromObjectOrReply(c, c->argv[2], &start, NULL) != REDIS_OK) ||
(getLongFromObjectOrReply(c, c->argv[3], &end, NULL) != REDIS_OK)) return;
if ((zsetobj = lookupKeyWriteOrReply(c,c->argv[1],shared.czero)) == NULL ||
checkType(c,zsetobj,REDIS_ZSET)) return;
zs = zsetobj->ptr;
llen = zs->zsl->length;
if ((zobj = lookupKeyWriteOrReply(c,key,shared.czero)) == NULL ||
checkType(c,zobj,REDIS_ZSET)) return;
/* convert negative indexes */
/* Sanitize indexes. */
llen = zsetLength(zobj);
if (start < 0) start = llen+start;
if (end < 0) end = llen+end;
if (start < 0) start = 0;
......@@ -535,27 +1055,348 @@ void zremrangebyrankCommand(redisClient *c) {
}
if (end >= llen) end = llen-1;
/* increment start and end because zsl*Rank functions
* use 1-based rank */
deleted = zslDeleteRangeByRank(zs->zsl,start+1,end+1,zs->dict);
if (htNeedsResize(zs->dict)) dictResize(zs->dict);
if (dictSize(zs->dict) == 0) dbDelete(c->db,c->argv[1]);
if (deleted) signalModifiedKey(c->db,c->argv[1]);
if (zobj->encoding == REDIS_ENCODING_ZIPLIST) {
/* Correct for 1-based rank. */
zobj->ptr = zzlDeleteRangeByRank(zobj->ptr,start+1,end+1,&deleted);
if (zzlLength(zobj->ptr) == 0) dbDelete(c->db,key);
} else if (zobj->encoding == REDIS_ENCODING_SKIPLIST) {
zset *zs = zobj->ptr;
/* Correct for 1-based rank. */
deleted = zslDeleteRangeByRank(zs->zsl,start+1,end+1,zs->dict);
if (htNeedsResize(zs->dict)) dictResize(zs->dict);
if (dictSize(zs->dict) == 0) dbDelete(c->db,key);
} else {
redisPanic("Unknown sorted set encoding");
}
if (deleted) signalModifiedKey(c->db,key);
server.dirty += deleted;
addReplyLongLong(c, deleted);
addReplyLongLong(c,deleted);
}
typedef struct {
dict *dict;
robj *subject;
int type; /* Set, sorted set */
int encoding;
double weight;
union {
/* Set iterators. */
union _iterset {
struct {
intset *is;
int ii;
} is;
struct {
dict *dict;
dictIterator *di;
dictEntry *de;
} ht;
} set;
/* Sorted set iterators. */
union _iterzset {
struct {
unsigned char *zl;
unsigned char *eptr, *sptr;
} zl;
struct {
zset *zs;
zskiplistNode *node;
} sl;
} zset;
} iter;
} zsetopsrc;
int qsortCompareZsetopsrcByCardinality(const void *s1, const void *s2) {
zsetopsrc *d1 = (void*) s1, *d2 = (void*) s2;
unsigned long size1, size2;
size1 = d1->dict ? dictSize(d1->dict) : 0;
size2 = d2->dict ? dictSize(d2->dict) : 0;
return size1 - size2;
/* Use dirty flags for pointers that need to be cleaned up in the next
* iteration over the zsetopval. The dirty flag for the long long value is
* special, since long long values don't need cleanup. Instead, it means that
* we already checked that "ell" holds a long long, or tried to convert another
* representation into a long long value. When this was successful,
* OPVAL_VALID_LL is set as well. */
#define OPVAL_DIRTY_ROBJ 1
#define OPVAL_DIRTY_LL 2
#define OPVAL_VALID_LL 4
/* Store value retrieved from the iterator. */
typedef struct {
int flags;
unsigned char _buf[32]; /* Private buffer. */
robj *ele;
unsigned char *estr;
unsigned int elen;
long long ell;
double score;
} zsetopval;
typedef union _iterset iterset;
typedef union _iterzset iterzset;
void zuiInitIterator(zsetopsrc *op) {
if (op->subject == NULL)
return;
if (op->type == REDIS_SET) {
iterset *it = &op->iter.set;
if (op->encoding == REDIS_ENCODING_INTSET) {
it->is.is = op->subject->ptr;
it->is.ii = 0;
} else if (op->encoding == REDIS_ENCODING_HT) {
it->ht.dict = op->subject->ptr;
it->ht.di = dictGetIterator(op->subject->ptr);
it->ht.de = dictNext(it->ht.di);
} else {
redisPanic("Unknown set encoding");
}
} else if (op->type == REDIS_ZSET) {
iterzset *it = &op->iter.zset;
if (op->encoding == REDIS_ENCODING_ZIPLIST) {
it->zl.zl = op->subject->ptr;
it->zl.eptr = ziplistIndex(it->zl.zl,0);
if (it->zl.eptr != NULL) {
it->zl.sptr = ziplistNext(it->zl.zl,it->zl.eptr);
redisAssert(it->zl.sptr != NULL);
}
} else if (op->encoding == REDIS_ENCODING_SKIPLIST) {
it->sl.zs = op->subject->ptr;
it->sl.node = it->sl.zs->zsl->header->level[0].forward;
} else {
redisPanic("Unknown sorted set encoding");
}
} else {
redisPanic("Unsupported type");
}
}
void zuiClearIterator(zsetopsrc *op) {
if (op->subject == NULL)
return;
if (op->type == REDIS_SET) {
iterset *it = &op->iter.set;
if (op->encoding == REDIS_ENCODING_INTSET) {
REDIS_NOTUSED(it); /* skip */
} else if (op->encoding == REDIS_ENCODING_HT) {
dictReleaseIterator(it->ht.di);
} else {
redisPanic("Unknown set encoding");
}
} else if (op->type == REDIS_ZSET) {
iterzset *it = &op->iter.zset;
if (op->encoding == REDIS_ENCODING_ZIPLIST) {
REDIS_NOTUSED(it); /* skip */
} else if (op->encoding == REDIS_ENCODING_SKIPLIST) {
REDIS_NOTUSED(it); /* skip */
} else {
redisPanic("Unknown sorted set encoding");
}
} else {
redisPanic("Unsupported type");
}
}
int zuiLength(zsetopsrc *op) {
if (op->subject == NULL)
return 0;
if (op->type == REDIS_SET) {
iterset *it = &op->iter.set;
if (op->encoding == REDIS_ENCODING_INTSET) {
return intsetLen(it->is.is);
} else if (op->encoding == REDIS_ENCODING_HT) {
return dictSize(it->ht.dict);
} else {
redisPanic("Unknown set encoding");
}
} else if (op->type == REDIS_ZSET) {
iterzset *it = &op->iter.zset;
if (op->encoding == REDIS_ENCODING_ZIPLIST) {
return zzlLength(it->zl.zl);
} else if (op->encoding == REDIS_ENCODING_SKIPLIST) {
return it->sl.zs->zsl->length;
} else {
redisPanic("Unknown sorted set encoding");
}
} else {
redisPanic("Unsupported type");
}
}
/* Check if the current value is valid. If so, store it in the passed structure
* and move to the next element. If not valid, this means we have reached the
* end of the structure and can abort. */
int zuiNext(zsetopsrc *op, zsetopval *val) {
if (op->subject == NULL)
return 0;
if (val->flags & OPVAL_DIRTY_ROBJ)
decrRefCount(val->ele);
bzero(val,sizeof(zsetopval));
if (op->type == REDIS_SET) {
iterset *it = &op->iter.set;
if (op->encoding == REDIS_ENCODING_INTSET) {
if (!intsetGet(it->is.is,it->is.ii,&val->ell))
return 0;
val->score = 1.0;
/* Move to next element. */
it->is.ii++;
} else if (op->encoding == REDIS_ENCODING_HT) {
if (it->ht.de == NULL)
return 0;
val->ele = dictGetEntryKey(it->ht.de);
val->score = 1.0;
/* Move to next element. */
it->ht.de = dictNext(it->ht.di);
} else {
redisPanic("Unknown set encoding");
}
} else if (op->type == REDIS_ZSET) {
iterzset *it = &op->iter.zset;
if (op->encoding == REDIS_ENCODING_ZIPLIST) {
/* No need to check both, but better be explicit. */
if (it->zl.eptr == NULL || it->zl.sptr == NULL)
return 0;
redisAssert(ziplistGet(it->zl.eptr,&val->estr,&val->elen,&val->ell));
val->score = zzlGetScore(it->zl.sptr);
/* Move to next element. */
zzlNext(it->zl.zl,&it->zl.eptr,&it->zl.sptr);
} else if (op->encoding == REDIS_ENCODING_SKIPLIST) {
if (it->sl.node == NULL)
return 0;
val->ele = it->sl.node->obj;
val->score = it->sl.node->score;
/* Move to next element. */
it->sl.node = it->sl.node->level[0].forward;
} else {
redisPanic("Unknown sorted set encoding");
}
} else {
redisPanic("Unsupported type");
}
return 1;
}
int zuiLongLongFromValue(zsetopval *val) {
if (!(val->flags & OPVAL_DIRTY_LL)) {
val->flags |= OPVAL_DIRTY_LL;
if (val->ele != NULL) {
if (val->ele->encoding == REDIS_ENCODING_INT) {
val->ell = (long)val->ele->ptr;
val->flags |= OPVAL_VALID_LL;
} else if (val->ele->encoding == REDIS_ENCODING_RAW) {
if (string2ll(val->ele->ptr,sdslen(val->ele->ptr),&val->ell))
val->flags |= OPVAL_VALID_LL;
} else {
redisPanic("Unsupported element encoding");
}
} else if (val->estr != NULL) {
if (string2ll((char*)val->estr,val->elen,&val->ell))
val->flags |= OPVAL_VALID_LL;
} else {
/* The long long was already set, flag as valid. */
val->flags |= OPVAL_VALID_LL;
}
}
return val->flags & OPVAL_VALID_LL;
}
robj *zuiObjectFromValue(zsetopval *val) {
if (val->ele == NULL) {
if (val->estr != NULL) {
val->ele = createStringObject((char*)val->estr,val->elen);
} else {
val->ele = createStringObjectFromLongLong(val->ell);
}
val->flags |= OPVAL_DIRTY_ROBJ;
}
return val->ele;
}
int zuiBufferFromValue(zsetopval *val) {
if (val->estr == NULL) {
if (val->ele != NULL) {
if (val->ele->encoding == REDIS_ENCODING_INT) {
val->elen = ll2string((char*)val->_buf,sizeof(val->_buf),(long)val->ele->ptr);
val->estr = val->_buf;
} else if (val->ele->encoding == REDIS_ENCODING_RAW) {
val->elen = sdslen(val->ele->ptr);
val->estr = val->ele->ptr;
} else {
redisPanic("Unsupported element encoding");
}
} else {
val->elen = ll2string((char*)val->_buf,sizeof(val->_buf),val->ell);
val->estr = val->_buf;
}
}
return 1;
}
/* Find value pointed to by val in the source pointer to by op. When found,
* return 1 and store its score in target. Return 0 otherwise. */
int zuiFind(zsetopsrc *op, zsetopval *val, double *score) {
if (op->subject == NULL)
return 0;
if (op->type == REDIS_SET) {
iterset *it = &op->iter.set;
if (op->encoding == REDIS_ENCODING_INTSET) {
if (zuiLongLongFromValue(val) && intsetFind(it->is.is,val->ell)) {
*score = 1.0;
return 1;
} else {
return 0;
}
} else if (op->encoding == REDIS_ENCODING_HT) {
zuiObjectFromValue(val);
if (dictFind(it->ht.dict,val->ele) != NULL) {
*score = 1.0;
return 1;
} else {
return 0;
}
} else {
redisPanic("Unknown set encoding");
}
} else if (op->type == REDIS_ZSET) {
iterzset *it = &op->iter.zset;
zuiObjectFromValue(val);
if (op->encoding == REDIS_ENCODING_ZIPLIST) {
if (zzlFind(it->zl.zl,val->ele,score) != NULL) {
/* Score is already set by zzlFind. */
return 1;
} else {
return 0;
}
} else if (op->encoding == REDIS_ENCODING_SKIPLIST) {
dictEntry *de;
if ((de = dictFind(it->sl.zs->dict,val->ele)) != NULL) {
*score = *(double*)dictGetEntryVal(de);
return 1;
} else {
return 0;
}
} else {
redisPanic("Unknown sorted set encoding");
}
} else {
redisPanic("Unsupported type");
}
}
int zuiCompareByCardinality(const void *s1, const void *s2) {
return zuiLength((zsetopsrc*)s1) - zuiLength((zsetopsrc*)s2);
}
#define REDIS_AGGR_SUM 1
......@@ -584,11 +1425,12 @@ void zunionInterGenericCommand(redisClient *c, robj *dstkey, int op) {
int i, j, setnum;
int aggregate = REDIS_AGGR_SUM;
zsetopsrc *src;
zsetopval zval;
robj *tmp;
unsigned int maxelelen = 0;
robj *dstobj;
zset *dstzset;
zskiplistNode *znode;
dictIterator *di;
dictEntry *de;
int touched = 0;
/* expect setnum input keys to be given */
......@@ -606,24 +1448,24 @@ void zunionInterGenericCommand(redisClient *c, robj *dstkey, int op) {
}
/* read keys to be used for input */
src = zmalloc(sizeof(zsetopsrc) * setnum);
src = zcalloc(sizeof(zsetopsrc) * setnum);
for (i = 0, j = 3; i < setnum; i++, j++) {
robj *obj = lookupKeyWrite(c->db,c->argv[j]);
if (!obj) {
src[i].dict = NULL;
} else {
if (obj->type == REDIS_ZSET) {
src[i].dict = ((zset*)obj->ptr)->dict;
} else if (obj->type == REDIS_SET) {
src[i].dict = (obj->ptr);
} else {
if (obj != NULL) {
if (obj->type != REDIS_ZSET && obj->type != REDIS_SET) {
zfree(src);
addReply(c,shared.wrongtypeerr);
return;
}
src[i].subject = obj;
src[i].type = obj->type;
src[i].encoding = obj->encoding;
} else {
src[i].subject = NULL;
}
/* default all weights to 1 */
/* Default all weights to 1. */
src[i].weight = 1.0;
}
......@@ -664,95 +1506,109 @@ void zunionInterGenericCommand(redisClient *c, robj *dstkey, int op) {
}
}
for (i = 0; i < setnum; i++)
zuiInitIterator(&src[i]);
/* sort sets from the smallest to largest, this will improve our
* algorithm's performance */
qsort(src,setnum,sizeof(zsetopsrc),qsortCompareZsetopsrcByCardinality);
qsort(src,setnum,sizeof(zsetopsrc),zuiCompareByCardinality);
dstobj = createZsetObject();
dstzset = dstobj->ptr;
memset(&zval, 0, sizeof(zval));
if (op == REDIS_OP_INTER) {
/* skip going over all entries if the smallest zset is NULL or empty */
if (src[0].dict && dictSize(src[0].dict) > 0) {
/* precondition: as src[0].dict is non-empty and the zsets are ordered
* from small to large, all src[i > 0].dict are non-empty too */
di = dictGetIterator(src[0].dict);
while((de = dictNext(di)) != NULL) {
/* Skip everything if the smallest input is empty. */
if (zuiLength(&src[0]) > 0) {
/* Precondition: as src[0] is non-empty and the inputs are ordered
* by size, all src[i > 0] are non-empty too. */
while (zuiNext(&src[0],&zval)) {
double score, value;
score = src[0].weight * zunionInterDictValue(de);
score = src[0].weight * zval.score;
for (j = 1; j < setnum; j++) {
dictEntry *other = dictFind(src[j].dict,dictGetEntryKey(de));
if (other) {
value = src[j].weight * zunionInterDictValue(other);
if (zuiFind(&src[j],&zval,&value)) {
value *= src[j].weight;
zunionInterAggregate(&score,value,aggregate);
} else {
break;
}
}
/* Only continue when present in every source dict. */
/* Only continue when present in every input. */
if (j == setnum) {
robj *o = dictGetEntryKey(de);
znode = zslInsert(dstzset->zsl,score,o);
incrRefCount(o); /* added to skiplist */
dictAdd(dstzset->dict,o,&znode->score);
incrRefCount(o); /* added to dictionary */
tmp = zuiObjectFromValue(&zval);
znode = zslInsert(dstzset->zsl,score,tmp);
incrRefCount(tmp); /* added to skiplist */
dictAdd(dstzset->dict,tmp,&znode->score);
incrRefCount(tmp); /* added to dictionary */
if (tmp->encoding == REDIS_ENCODING_RAW)
if (sdslen(tmp->ptr) > maxelelen)
maxelelen = sdslen(tmp->ptr);
}
}
dictReleaseIterator(di);
}
} else if (op == REDIS_OP_UNION) {
for (i = 0; i < setnum; i++) {
if (!src[i].dict) continue;
if (zuiLength(&src[0]) == 0)
continue;
di = dictGetIterator(src[i].dict);
while((de = dictNext(di)) != NULL) {
while (zuiNext(&src[i],&zval)) {
double score, value;
/* skip key when already processed */
if (dictFind(dstzset->dict,dictGetEntryKey(de)) != NULL)
/* Skip key when already processed */
if (dictFind(dstzset->dict,zuiObjectFromValue(&zval)) != NULL)
continue;
/* initialize score */
score = src[i].weight * zunionInterDictValue(de);
/* Initialize score */
score = src[i].weight * zval.score;
/* because the zsets are sorted by size, its only possible
* for sets at larger indices to hold this entry */
/* Because the inputs are sorted by size, it's only possible
* for sets at larger indices to hold this element. */
for (j = (i+1); j < setnum; j++) {
dictEntry *other = dictFind(src[j].dict,dictGetEntryKey(de));
if (other) {
value = src[j].weight * zunionInterDictValue(other);
if (zuiFind(&src[j],&zval,&value)) {
value *= src[j].weight;
zunionInterAggregate(&score,value,aggregate);
}
}
robj *o = dictGetEntryKey(de);
znode = zslInsert(dstzset->zsl,score,o);
incrRefCount(o); /* added to skiplist */
dictAdd(dstzset->dict,o,&znode->score);
incrRefCount(o); /* added to dictionary */
tmp = zuiObjectFromValue(&zval);
znode = zslInsert(dstzset->zsl,score,tmp);
incrRefCount(zval.ele); /* added to skiplist */
dictAdd(dstzset->dict,tmp,&znode->score);
incrRefCount(zval.ele); /* added to dictionary */
if (tmp->encoding == REDIS_ENCODING_RAW)
if (sdslen(tmp->ptr) > maxelelen)
maxelelen = sdslen(tmp->ptr);
}
dictReleaseIterator(di);
}
} else {
/* unknown operator */
redisAssert(op == REDIS_OP_INTER || op == REDIS_OP_UNION);
redisPanic("Unknown operator");
}
for (i = 0; i < setnum; i++)
zuiClearIterator(&src[i]);
if (dbDelete(c->db,dstkey)) {
signalModifiedKey(c->db,dstkey);
touched = 1;
server.dirty++;
}
if (dstzset->zsl->length) {
/* Convert to ziplist when in limits. */
if (dstzset->zsl->length <= server.zset_max_ziplist_entries &&
maxelelen <= server.zset_max_ziplist_value)
zsetConvert(dstobj,REDIS_ENCODING_ZIPLIST);
dbAdd(c->db,dstkey,dstobj);
addReplyLongLong(c, dstzset->zsl->length);
addReplyLongLong(c,zsetLength(dstobj));
if (!touched) signalModifiedKey(c->db,dstkey);
server.dirty++;
} else {
decrRefCount(dstobj);
addReply(c, shared.czero);
addReply(c,shared.czero);
}
zfree(src);
}
......@@ -766,16 +1622,13 @@ void zinterstoreCommand(redisClient *c) {
}
void zrangeGenericCommand(redisClient *c, int reverse) {
robj *o;
robj *key = c->argv[1];
robj *zobj;
int withscores = 0;
long start;
long end;
int withscores = 0;
int llen;
int rangelen, j;
zset *zsetobj;
zskiplist *zsl;
zskiplistNode *ln;
robj *ele;
int rangelen;
if ((getLongFromObjectOrReply(c, c->argv[2], &start, NULL) != REDIS_OK) ||
(getLongFromObjectOrReply(c, c->argv[3], &end, NULL) != REDIS_OK)) return;
......@@ -787,13 +1640,11 @@ void zrangeGenericCommand(redisClient *c, int reverse) {
return;
}
if ((o = lookupKeyReadOrReply(c,c->argv[1],shared.emptymultibulk)) == NULL
|| checkType(c,o,REDIS_ZSET)) return;
zsetobj = o->ptr;
zsl = zsetobj->zsl;
llen = zsl->length;
if ((zobj = lookupKeyReadOrReply(c,key,shared.emptymultibulk)) == NULL
|| checkType(c,zobj,REDIS_ZSET)) return;
/* convert negative indexes */
/* Sanitize indexes. */
llen = zsetLength(zobj);
if (start < 0) start = llen+start;
if (end < 0) end = llen+end;
if (start < 0) start = 0;
......@@ -807,23 +1658,68 @@ void zrangeGenericCommand(redisClient *c, int reverse) {
if (end >= llen) end = llen-1;
rangelen = (end-start)+1;
/* check if starting point is trivial, before searching
* the element in log(N) time */
if (reverse) {
ln = start == 0 ? zsl->tail : zslistTypeGetElementByRank(zsl, llen-start);
} else {
ln = start == 0 ?
zsl->header->level[0].forward : zslistTypeGetElementByRank(zsl, start+1);
}
/* Return the result in form of a multi-bulk reply */
addReplyMultiBulkLen(c,withscores ? (rangelen*2) : rangelen);
for (j = 0; j < rangelen; j++) {
ele = ln->obj;
addReplyBulk(c,ele);
if (withscores)
addReplyDouble(c,ln->score);
ln = reverse ? ln->backward : ln->level[0].forward;
addReplyMultiBulkLen(c, withscores ? (rangelen*2) : rangelen);
if (zobj->encoding == REDIS_ENCODING_ZIPLIST) {
unsigned char *zl = zobj->ptr;
unsigned char *eptr, *sptr;
unsigned char *vstr;
unsigned int vlen;
long long vlong;
if (reverse)
eptr = ziplistIndex(zl,-2-(2*start));
else
eptr = ziplistIndex(zl,2*start);
redisAssert(eptr != NULL);
sptr = ziplistNext(zl,eptr);
while (rangelen--) {
redisAssert(eptr != NULL && sptr != NULL);
redisAssert(ziplistGet(eptr,&vstr,&vlen,&vlong));
if (vstr == NULL)
addReplyBulkLongLong(c,vlong);
else
addReplyBulkCBuffer(c,vstr,vlen);
if (withscores)
addReplyDouble(c,zzlGetScore(sptr));
if (reverse)
zzlPrev(zl,&eptr,&sptr);
else
zzlNext(zl,&eptr,&sptr);
}
} else if (zobj->encoding == REDIS_ENCODING_SKIPLIST) {
zset *zs = zobj->ptr;
zskiplist *zsl = zs->zsl;
zskiplistNode *ln;
robj *ele;
/* Check if starting point is trivial, before doing log(N) lookup. */
if (reverse) {
ln = zsl->tail;
if (start > 0)
ln = zslGetElementByRank(zsl,llen-start);
} else {
ln = zsl->header->level[0].forward;
if (start > 0)
ln = zslGetElementByRank(zsl,start+1);
}
while(rangelen--) {
redisAssert(ln != NULL);
ele = ln->obj;
addReplyBulk(c,ele);
if (withscores)
addReplyDouble(c,ln->score);
ln = reverse ? ln->backward : ln->level[0].forward;
}
} else {
redisPanic("Unknown sorted set encoding");
}
}
......@@ -839,17 +1735,24 @@ void zrevrangeCommand(redisClient *c) {
* If "justcount", only the number of elements in the range is returned. */
void genericZrangebyscoreCommand(redisClient *c, int reverse, int justcount) {
zrangespec range;
robj *o, *emptyreply;
zset *zsetobj;
zskiplist *zsl;
zskiplistNode *ln;
robj *key = c->argv[1];
robj *emptyreply, *zobj;
int offset = 0, limit = -1;
int withscores = 0;
unsigned long rangelen = 0;
void *replylen = NULL;
int minidx, maxidx;
/* Parse the range arguments. */
if (zslParseRange(c->argv[2],c->argv[3],&range) != REDIS_OK) {
if (reverse) {
/* Range is given as [max,min] */
maxidx = 2; minidx = 3;
} else {
/* Range is given as [min,max] */
minidx = 2; maxidx = 3;
}
if (zslParseRange(c->argv[minidx],c->argv[maxidx],&range) != REDIS_OK) {
addReplyError(c,"min or max is not a double");
return;
}
......@@ -877,101 +1780,132 @@ void genericZrangebyscoreCommand(redisClient *c, int reverse, int justcount) {
/* Ok, lookup the key and get the range */
emptyreply = justcount ? shared.czero : shared.emptymultibulk;
if ((o = lookupKeyReadOrReply(c,c->argv[1],emptyreply)) == NULL ||
checkType(c,o,REDIS_ZSET)) return;
zsetobj = o->ptr;
zsl = zsetobj->zsl;
if ((zobj = lookupKeyReadOrReply(c,key,emptyreply)) == NULL ||
checkType(c,zobj,REDIS_ZSET)) return;
if (zobj->encoding == REDIS_ENCODING_ZIPLIST) {
unsigned char *zl = zobj->ptr;
unsigned char *eptr, *sptr;
unsigned char *vstr;
unsigned int vlen;
long long vlong;
double score;
/* If reversed, get the last node in range as starting point. */
if (reverse)
eptr = zzlLastInRange(zl,range);
else
eptr = zzlFirstInRange(zl,range);
/* If reversed, assume the elements are sorted from high to low score. */
ln = zslFirstWithScore(zsl,range.min);
if (reverse) {
/* If range.min is out of range, ln will be NULL and we need to use
* the tail of the skiplist as first node of the range. */
if (ln == NULL) ln = zsl->tail;
/* zslFirstWithScore returns the first element with where with
* score >= range.min, so backtrack to make sure the element we use
* here has score <= range.min. */
while (ln && ln->score > range.min) ln = ln->backward;
/* Move to the right element according to the range spec. */
if (range.minex) {
/* Find last element with score < range.min */
while (ln && ln->score == range.min) ln = ln->backward;
} else {
/* Find last element with score <= range.min */
while (ln && ln->level[0].forward &&
ln->level[0].forward->score == range.min)
ln = ln->level[0].forward;
}
} else {
if (range.minex) {
/* Find first element with score > range.min */
while (ln && ln->score == range.min) ln = ln->level[0].forward;
/* No "first" element in the specified interval. */
if (eptr == NULL) {
addReply(c,emptyreply);
return;
}
}
/* No "first" element in the specified interval. */
if (ln == NULL) {
addReply(c,emptyreply);
return;
}
/* Get score pointer for the first element. */
redisAssert(eptr != NULL);
sptr = ziplistNext(zl,eptr);
/* We don't know in advance how many matching elements there are in the
* list, so we push this object that will represent the multi-bulk
* length in the output buffer, and will "fix" it later */
if (!justcount)
replylen = addDeferredMultiBulkLength(c);
/* If there is an offset, just traverse the number of elements without
* checking the score because that is done in the next loop. */
while (eptr && offset--)
if (reverse)
zzlPrev(zl,&eptr,&sptr);
else
zzlNext(zl,&eptr,&sptr);
while (eptr && limit--) {
score = zzlGetScore(sptr);
/* Abort when the node is no longer in range. */
if (reverse) {
if (!zslValueGteMin(score,&range)) break;
} else {
if (!zslValueLteMax(score,&range)) break;
}
/* We don't know in advance how many matching elements there
* are in the list, so we push this object that will represent
* the multi-bulk length in the output buffer, and will "fix"
* it later */
if (!justcount)
replylen = addDeferredMultiBulkLength(c);
/* Do our magic */
rangelen++;
if (!justcount) {
redisAssert(ziplistGet(eptr,&vstr,&vlen,&vlong));
if (vstr == NULL)
addReplyBulkLongLong(c,vlong);
else
addReplyBulkCBuffer(c,vstr,vlen);
if (withscores)
addReplyDouble(c,score);
}
/* Move to next node */
if (reverse)
zzlPrev(zl,&eptr,&sptr);
else
zzlNext(zl,&eptr,&sptr);
}
} else if (zobj->encoding == REDIS_ENCODING_SKIPLIST) {
zset *zs = zobj->ptr;
zskiplist *zsl = zs->zsl;
zskiplistNode *ln;
/* If there is an offset, just traverse the number of elements without
* checking the score because that is done in the next loop. */
while(ln && offset--) {
/* If reversed, get the last node in range as starting point. */
if (reverse)
ln = ln->backward;
ln = zslLastInRange(zsl,range);
else
ln = ln->level[0].forward;
}
ln = zslFirstInRange(zsl,range);
while (ln && limit--) {
/* Check if this this element is in range. */
if (reverse) {
if (range.maxex) {
/* Element should have score > range.max */
if (ln->score <= range.max) break;
/* No "first" element in the specified interval. */
if (ln == NULL) {
addReply(c,emptyreply);
return;
}
/* We don't know in advance how many matching elements there are in the
* list, so we push this object that will represent the multi-bulk
* length in the output buffer, and will "fix" it later */
if (!justcount)
replylen = addDeferredMultiBulkLength(c);
/* If there is an offset, just traverse the number of elements without
* checking the score because that is done in the next loop. */
while (ln && offset--)
ln = reverse ? ln->backward : ln->level[0].forward;
while (ln && limit--) {
/* Abort when the node is no longer in range. */
if (reverse) {
if (!zslValueGteMin(ln->score,&range)) break;
} else {
/* Element should have score >= range.max */
if (ln->score < range.max) break;
if (!zslValueLteMax(ln->score,&range)) break;
}
} else {
if (range.maxex) {
/* Element should have score < range.max */
if (ln->score >= range.max) break;
} else {
/* Element should have score <= range.max */
if (ln->score > range.max) break;
/* Do our magic */
rangelen++;
if (!justcount) {
addReplyBulk(c,ln->obj);
if (withscores)
addReplyDouble(c,ln->score);
}
}
/* Do our magic */
rangelen++;
if (!justcount) {
addReplyBulk(c,ln->obj);
if (withscores)
addReplyDouble(c,ln->score);
/* Move to next node */
ln = reverse ? ln->backward : ln->level[0].forward;
}
if (reverse)
ln = ln->backward;
else
ln = ln->level[0].forward;
} else {
redisPanic("Unknown sorted set encoding");
}
if (justcount) {
addReplyLongLong(c,(long)rangelen);
} else {
setDeferredMultiBulkLength(c,replylen,
withscores ? (rangelen*2) : rangelen);
if (withscores) rangelen *= 2;
setDeferredMultiBulkLength(c,replylen,rangelen);
}
}
......@@ -988,66 +1922,103 @@ void zcountCommand(redisClient *c) {
}
void zcardCommand(redisClient *c) {
robj *o;
zset *zs;
robj *key = c->argv[1];
robj *zobj;
if ((o = lookupKeyReadOrReply(c,c->argv[1],shared.czero)) == NULL ||
checkType(c,o,REDIS_ZSET)) return;
if ((zobj = lookupKeyReadOrReply(c,key,shared.czero)) == NULL ||
checkType(c,zobj,REDIS_ZSET)) return;
zs = o->ptr;
addReplyLongLong(c,zs->zsl->length);
addReplyLongLong(c,zsetLength(zobj));
}
void zscoreCommand(redisClient *c) {
robj *o;
zset *zs;
dictEntry *de;
robj *key = c->argv[1];
robj *zobj;
double score;
if ((o = lookupKeyReadOrReply(c,c->argv[1],shared.nullbulk)) == NULL ||
checkType(c,o,REDIS_ZSET)) return;
if ((zobj = lookupKeyReadOrReply(c,key,shared.nullbulk)) == NULL ||
checkType(c,zobj,REDIS_ZSET)) return;
zs = o->ptr;
c->argv[2] = tryObjectEncoding(c->argv[2]);
de = dictFind(zs->dict,c->argv[2]);
if (!de) {
addReply(c,shared.nullbulk);
} else {
double *score = dictGetEntryVal(de);
if (zobj->encoding == REDIS_ENCODING_ZIPLIST) {
if (zzlFind(zobj->ptr,c->argv[2],&score) != NULL)
addReplyDouble(c,score);
else
addReply(c,shared.nullbulk);
} else if (zobj->encoding == REDIS_ENCODING_SKIPLIST) {
zset *zs = zobj->ptr;
dictEntry *de;
addReplyDouble(c,*score);
c->argv[2] = tryObjectEncoding(c->argv[2]);
de = dictFind(zs->dict,c->argv[2]);
if (de != NULL) {
score = *(double*)dictGetEntryVal(de);
addReplyDouble(c,score);
} else {
addReply(c,shared.nullbulk);
}
} else {
redisPanic("Unknown sorted set encoding");
}
}
void zrankGenericCommand(redisClient *c, int reverse) {
robj *o;
zset *zs;
zskiplist *zsl;
dictEntry *de;
robj *key = c->argv[1];
robj *ele = c->argv[2];
robj *zobj;
unsigned long llen;
unsigned long rank;
double *score;
if ((o = lookupKeyReadOrReply(c,c->argv[1],shared.nullbulk)) == NULL ||
checkType(c,o,REDIS_ZSET)) return;
if ((zobj = lookupKeyReadOrReply(c,key,shared.nullbulk)) == NULL ||
checkType(c,zobj,REDIS_ZSET)) return;
llen = zsetLength(zobj);
redisAssert(ele->encoding == REDIS_ENCODING_RAW);
if (zobj->encoding == REDIS_ENCODING_ZIPLIST) {
unsigned char *zl = zobj->ptr;
unsigned char *eptr, *sptr;
eptr = ziplistIndex(zl,0);
redisAssert(eptr != NULL);
sptr = ziplistNext(zl,eptr);
redisAssert(sptr != NULL);
rank = 1;
while(eptr != NULL) {
if (ziplistCompare(eptr,ele->ptr,sdslen(ele->ptr)))
break;
rank++;
zzlNext(zl,&eptr,&sptr);
}
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;
}
if (eptr != NULL) {
if (reverse)
addReplyLongLong(c,llen-rank);
else
addReplyLongLong(c,rank-1);
} else {
addReply(c,shared.nullbulk);
}
} else if (zobj->encoding == REDIS_ENCODING_SKIPLIST) {
zset *zs = zobj->ptr;
zskiplist *zsl = zs->zsl;
dictEntry *de;
double score;
score = dictGetEntryVal(de);
rank = zslistTypeGetRank(zsl, *score, c->argv[2]);
if (rank) {
if (reverse) {
addReplyLongLong(c, zsl->length - rank);
ele = c->argv[2] = tryObjectEncoding(c->argv[2]);
de = dictFind(zs->dict,ele);
if (de != NULL) {
score = *(double*)dictGetEntryVal(de);
rank = zslGetRank(zsl,score,ele);
redisAssert(rank); /* Existing elements always have a rank. */
if (reverse)
addReplyLongLong(c,llen-rank);
else
addReplyLongLong(c,rank-1);
} else {
addReplyLongLong(c, rank-1);
addReply(c,shared.nullbulk);
}
} else {
addReply(c,shared.nullbulk);
redisPanic("Unknown sorted set encoding");
}
}
......
#include "redis.h"
#include <ctype.h>
#include <limits.h>
#include <math.h>
/* Glob-style pattern matching. */
int stringmatchlen(const char *pattern, int patternLen,
......@@ -200,6 +201,102 @@ int ll2string(char *s, size_t len, long long value) {
return l;
}
/* Convert a string into a long long. Returns 1 if the string could be parsed
* into a (non-overflowing) long long, 0 otherwise. The value will be set to
* the parsed value when appropriate. */
int string2ll(char *s, size_t slen, long long *value) {
char *p = s;
size_t plen = 0;
int negative = 0;
unsigned long long v;
if (plen == slen)
return 0;
if (p[0] == '-') {
negative = 1;
p++; plen++;
/* Abort on only a negative sign. */
if (plen == slen)
return 0;
}
/* First digit should be 1-9. */
if (p[0] >= '1' && p[0] <= '9') {
v = p[0]-'0';
p++; plen++;
} else {
return 0;
}
while (plen < slen && p[0] >= '0' && p[0] <= '9') {
if (v > (ULLONG_MAX / 10)) /* Overflow. */
return 0;
v *= 10;
if (v > (ULLONG_MAX - (p[0]-'0'))) /* Overflow. */
return 0;
v += p[0]-'0';
p++; plen++;
}
/* Return if not all bytes were used. */
if (plen < slen)
return 0;
if (negative) {
if (v > (-(unsigned long long)LLONG_MIN)) /* Overflow. */
return 0;
if (value != NULL) *value = -v;
} else {
if (v > LLONG_MAX) /* Overflow. */
return 0;
if (value != NULL) *value = v;
}
return 1;
}
/* Convert a double to a string representation. Returns the number of bytes
* required. The representation should always be parsable by stdtod(3). */
int d2string(char *buf, size_t len, double value) {
if (isnan(value)) {
len = snprintf(buf,len,"nan");
} else if (isinf(value)) {
if (value < 0)
len = snprintf(buf,len,"-inf");
else
len = snprintf(buf,len,"inf");
} else if (value == 0) {
/* See: http://en.wikipedia.org/wiki/Signed_zero, "Comparisons". */
if (1.0/value < 0)
len = snprintf(buf,len,"-0");
else
len = snprintf(buf,len,"0");
} else {
#if (DBL_MANT_DIG >= 52) && (LLONG_MAX == 0x7fffffffffffffffLL)
/* Check if the float is in a safe range to be casted into a
* long long. We are assuming that long long is 64 bit here.
* Also we are assuming that there are no implementations around where
* double has precision < 52 bit.
*
* Under this assumptions we test if a double is inside an interval
* where casting to long long is safe. Then using two castings we
* make sure the decimal part is zero. If all this is true we use
* integer printing function that is much faster. */
double min = -4503599627370495; /* (2^52)-1 */
double max = 4503599627370496; /* -(2^52) */
if (val > min && val < max && value == ((double)((long long)value)))
len = ll2string(buf,len,(long long)value);
else
#endif
len = snprintf(buf,len,"%.17g",value);
}
return len;
}
/* Check if the sds string 's' can be represented by a long long
* (that is, is a number that fits into long without any other space or
* character before or after the digits, so that converting this number
......
......@@ -385,8 +385,8 @@ static unsigned char *ziplistResize(unsigned char *zl, unsigned int len) {
* The pointer "p" points to the first entry that does NOT need to be
* updated, i.e. consecutive fields MAY need an update. */
static unsigned char *__ziplistCascadeUpdate(unsigned char *zl, unsigned char *p) {
unsigned int curlen = ZIPLIST_BYTES(zl), rawlen, rawlensize;
unsigned int offset, noffset, extra;
size_t curlen = ZIPLIST_BYTES(zl), rawlen, rawlensize;
size_t offset, noffset, extra;
unsigned char *np;
zlentry cur, next;
......@@ -441,7 +441,8 @@ static unsigned char *__ziplistCascadeUpdate(unsigned char *zl, unsigned char *p
/* Delete "num" entries, starting at "p". Returns pointer to the ziplist. */
static unsigned char *__ziplistDelete(unsigned char *zl, unsigned char *p, unsigned int num) {
unsigned int i, totlen, deleted = 0;
int offset, nextdiff = 0;
size_t offset;
int nextdiff = 0;
zlentry first, tail;
first = zipEntry(p);
......@@ -493,8 +494,9 @@ static unsigned char *__ziplistDelete(unsigned char *zl, unsigned char *p, unsig
/* Insert item at "p". */
static unsigned char *__ziplistInsert(unsigned char *zl, unsigned char *p, unsigned char *s, unsigned int slen) {
unsigned int curlen = ZIPLIST_BYTES(zl), reqlen, prevlen = 0;
unsigned int offset, nextdiff = 0;
size_t curlen = ZIPLIST_BYTES(zl), reqlen, prevlen = 0;
size_t offset;
int nextdiff = 0;
unsigned char encoding = 0;
long long value;
zlentry entry, tail;
......@@ -678,7 +680,7 @@ unsigned char *ziplistInsert(unsigned char *zl, unsigned char *p, unsigned char
* Also update *p in place, to be able to iterate over the
* ziplist, while deleting entries. */
unsigned char *ziplistDelete(unsigned char *zl, unsigned char **p) {
unsigned int offset = *p-zl;
size_t offset = *p-zl;
zl = __ziplistDelete(zl,*p,1);
/* Store pointer to current element in p, because ziplistDelete will
......
......@@ -6,266 +6,562 @@ start_server {tags {"zset"}} {
}
}
test {ZSET basic ZADD and score update} {
r zadd ztmp 10 x
r zadd ztmp 20 y
r zadd ztmp 30 z
set aux1 [r zrange ztmp 0 -1]
r zadd ztmp 1 y
set aux2 [r zrange ztmp 0 -1]
list $aux1 $aux2
} {{x y z} {y x z}}
test {ZCARD basics} {
r zcard ztmp
} {3}
test {ZCARD non existing key} {
r zcard ztmp-blabla
} {0}
test "ZRANGE basics" {
r del ztmp
r zadd ztmp 1 a
r zadd ztmp 2 b
r zadd ztmp 3 c
r zadd ztmp 4 d
assert_equal {a b c d} [r zrange ztmp 0 -1]
assert_equal {a b c} [r zrange ztmp 0 -2]
assert_equal {b c d} [r zrange ztmp 1 -1]
assert_equal {b c} [r zrange ztmp 1 -2]
assert_equal {c d} [r zrange ztmp -2 -1]
assert_equal {c} [r zrange ztmp -2 -2]
# out of range start index
assert_equal {a b c} [r zrange ztmp -5 2]
assert_equal {a b} [r zrange ztmp -5 1]
assert_equal {} [r zrange ztmp 5 -1]
assert_equal {} [r zrange ztmp 5 -2]
# out of range end index
assert_equal {a b c d} [r zrange ztmp 0 5]
assert_equal {b c d} [r zrange ztmp 1 5]
assert_equal {} [r zrange ztmp 0 -5]
assert_equal {} [r zrange ztmp 1 -5]
# withscores
assert_equal {a 1 b 2 c 3 d 4} [r zrange ztmp 0 -1 withscores]
}
proc basics {encoding} {
if {$encoding == "ziplist"} {
r config set zset-max-ziplist-entries 128
r config set zset-max-ziplist-value 64
} elseif {$encoding == "skiplist"} {
r config set zset-max-ziplist-entries 0
r config set zset-max-ziplist-value 0
} else {
puts "Unknown sorted set encoding"
exit
}
test "ZREVRANGE basics" {
r del ztmp
r zadd ztmp 1 a
r zadd ztmp 2 b
r zadd ztmp 3 c
r zadd ztmp 4 d
assert_equal {d c b a} [r zrevrange ztmp 0 -1]
assert_equal {d c b} [r zrevrange ztmp 0 -2]
assert_equal {c b a} [r zrevrange ztmp 1 -1]
assert_equal {c b} [r zrevrange ztmp 1 -2]
assert_equal {b a} [r zrevrange ztmp -2 -1]
assert_equal {b} [r zrevrange ztmp -2 -2]
# out of range start index
assert_equal {d c b} [r zrevrange ztmp -5 2]
assert_equal {d c} [r zrevrange ztmp -5 1]
assert_equal {} [r zrevrange ztmp 5 -1]
assert_equal {} [r zrevrange ztmp 5 -2]
# out of range end index
assert_equal {d c b a} [r zrevrange ztmp 0 5]
assert_equal {c b a} [r zrevrange ztmp 1 5]
assert_equal {} [r zrevrange ztmp 0 -5]
assert_equal {} [r zrevrange ztmp 1 -5]
# withscores
assert_equal {d 4 c 3 b 2 a 1} [r zrevrange ztmp 0 -1 withscores]
}
test "Check encoding - $encoding" {
r del ztmp
r zadd ztmp 10 x
assert_encoding $encoding ztmp
}
test {ZRANK basics} {
r zadd zranktmp 10 x
r zadd zranktmp 20 y
r zadd zranktmp 30 z
list [r zrank zranktmp x] [r zrank zranktmp y] [r zrank zranktmp z]
} {0 1 2}
test {ZREVRANK basics} {
list [r zrevrank zranktmp x] [r zrevrank zranktmp y] [r zrevrank zranktmp z]
} {2 1 0}
test {ZRANK - after deletion} {
r zrem zranktmp y
list [r zrank zranktmp x] [r zrank zranktmp z]
} {0 1}
test {ZSCORE} {
set aux {}
set err {}
for {set i 0} {$i < 1000} {incr i} {
set score [expr rand()]
lappend aux $score
r zadd zscoretest $score $i
}
for {set i 0} {$i < 1000} {incr i} {
if {[r zscore zscoretest $i] != [lindex $aux $i]} {
set err "Expected score was [lindex $aux $i] but got [r zscore zscoretest $i] for element $i"
break
}
test "ZSET basic ZADD and score update - $encoding" {
r del ztmp
r zadd ztmp 10 x
r zadd ztmp 20 y
r zadd ztmp 30 z
assert_equal {x y z} [r zrange ztmp 0 -1]
r zadd ztmp 1 y
assert_equal {y x z} [r zrange ztmp 0 -1]
}
set _ $err
} {}
test {ZSCORE after a DEBUG RELOAD} {
set aux {}
set err {}
r del zscoretest
for {set i 0} {$i < 1000} {incr i} {
set score [expr rand()]
lappend aux $score
r zadd zscoretest $score $i
}
r debug reload
for {set i 0} {$i < 1000} {incr i} {
if {[r zscore zscoretest $i] != [lindex $aux $i]} {
set err "Expected score was [lindex $aux $i] but got [r zscore zscoretest $i] for element $i"
break
}
test "ZSET element can't be set to NaN with ZADD - $encoding" {
assert_error "*not a double*" {r zadd myzset nan abc}
}
set _ $err
} {}
test {ZSETs stress tester - sorting is working well?} {
set delta 0
for {set test 0} {$test < 2} {incr test} {
unset -nocomplain auxarray
array set auxarray {}
set auxlist {}
r del myzset
for {set i 0} {$i < 1000} {incr i} {
if {$test == 0} {
set score [expr rand()]
} else {
set score [expr int(rand()*10)]
}
set auxarray($i) $score
r zadd myzset $score $i
# Random update
if {[expr rand()] < .2} {
set j [expr int(rand()*1000)]
if {$test == 0} {
set score [expr rand()]
} else {
set score [expr int(rand()*10)]
}
set auxarray($j) $score
r zadd myzset $score $j
}
test "ZSET element can't be set to NaN with ZINCRBY" {
assert_error "*not a double*" {r zadd myzset nan abc}
}
test "ZINCRBY calls leading to NaN result in error" {
r zincrby myzset +inf abc
assert_error "*NaN*" {r zincrby myzset -inf abc}
}
test "ZCARD basics - $encoding" {
assert_equal 3 [r zcard ztmp]
assert_equal 0 [r zcard zdoesntexist]
}
test "ZREM removes key after last element is removed" {
r del ztmp
r zadd ztmp 10 x
r zadd ztmp 20 y
assert_equal 1 [r exists ztmp]
assert_equal 0 [r zrem ztmp z]
assert_equal 1 [r zrem ztmp y]
assert_equal 1 [r zrem ztmp x]
assert_equal 0 [r exists ztmp]
}
test "ZRANGE basics - $encoding" {
r del ztmp
r zadd ztmp 1 a
r zadd ztmp 2 b
r zadd ztmp 3 c
r zadd ztmp 4 d
assert_equal {a b c d} [r zrange ztmp 0 -1]
assert_equal {a b c} [r zrange ztmp 0 -2]
assert_equal {b c d} [r zrange ztmp 1 -1]
assert_equal {b c} [r zrange ztmp 1 -2]
assert_equal {c d} [r zrange ztmp -2 -1]
assert_equal {c} [r zrange ztmp -2 -2]
# out of range start index
assert_equal {a b c} [r zrange ztmp -5 2]
assert_equal {a b} [r zrange ztmp -5 1]
assert_equal {} [r zrange ztmp 5 -1]
assert_equal {} [r zrange ztmp 5 -2]
# out of range end index
assert_equal {a b c d} [r zrange ztmp 0 5]
assert_equal {b c d} [r zrange ztmp 1 5]
assert_equal {} [r zrange ztmp 0 -5]
assert_equal {} [r zrange ztmp 1 -5]
# withscores
assert_equal {a 1 b 2 c 3 d 4} [r zrange ztmp 0 -1 withscores]
}
test "ZREVRANGE basics - $encoding" {
r del ztmp
r zadd ztmp 1 a
r zadd ztmp 2 b
r zadd ztmp 3 c
r zadd ztmp 4 d
assert_equal {d c b a} [r zrevrange ztmp 0 -1]
assert_equal {d c b} [r zrevrange ztmp 0 -2]
assert_equal {c b a} [r zrevrange ztmp 1 -1]
assert_equal {c b} [r zrevrange ztmp 1 -2]
assert_equal {b a} [r zrevrange ztmp -2 -1]
assert_equal {b} [r zrevrange ztmp -2 -2]
# out of range start index
assert_equal {d c b} [r zrevrange ztmp -5 2]
assert_equal {d c} [r zrevrange ztmp -5 1]
assert_equal {} [r zrevrange ztmp 5 -1]
assert_equal {} [r zrevrange ztmp 5 -2]
# out of range end index
assert_equal {d c b a} [r zrevrange ztmp 0 5]
assert_equal {c b a} [r zrevrange ztmp 1 5]
assert_equal {} [r zrevrange ztmp 0 -5]
assert_equal {} [r zrevrange ztmp 1 -5]
# withscores
assert_equal {d 4 c 3 b 2 a 1} [r zrevrange ztmp 0 -1 withscores]
}
test "ZRANK/ZREVRANK basics - $encoding" {
r del zranktmp
r zadd zranktmp 10 x
r zadd zranktmp 20 y
r zadd zranktmp 30 z
assert_equal 0 [r zrank zranktmp x]
assert_equal 1 [r zrank zranktmp y]
assert_equal 2 [r zrank zranktmp z]
assert_equal "" [r zrank zranktmp foo]
assert_equal 2 [r zrevrank zranktmp x]
assert_equal 1 [r zrevrank zranktmp y]
assert_equal 0 [r zrevrank zranktmp z]
assert_equal "" [r zrevrank zranktmp foo]
}
test "ZRANK - after deletion - $encoding" {
r zrem zranktmp y
assert_equal 0 [r zrank zranktmp x]
assert_equal 1 [r zrank zranktmp z]
}
test "ZINCRBY - can create a new sorted set - $encoding" {
r del zset
r zincrby zset 1 foo
assert_equal {foo} [r zrange zset 0 -1]
assert_equal 1 [r zscore zset foo]
}
test "ZINCRBY - increment and decrement - $encoding" {
r zincrby zset 2 foo
r zincrby zset 1 bar
assert_equal {bar foo} [r zrange zset 0 -1]
r zincrby zset 10 bar
r zincrby zset -5 foo
r zincrby zset -5 bar
assert_equal {foo bar} [r zrange zset 0 -1]
assert_equal -2 [r zscore zset foo]
assert_equal 6 [r zscore zset bar]
}
proc create_default_zset {} {
create_zset zset {-inf a 1 b 2 c 3 d 4 e 5 f +inf g}
}
test "ZRANGEBYSCORE/ZREVRANGEBYSCORE/ZCOUNT basics" {
create_default_zset
# inclusive range
assert_equal {a b c} [r zrangebyscore zset -inf 2]
assert_equal {b c d} [r zrangebyscore zset 0 3]
assert_equal {d e f} [r zrangebyscore zset 3 6]
assert_equal {e f g} [r zrangebyscore zset 4 +inf]
assert_equal {c b a} [r zrevrangebyscore zset 2 -inf]
assert_equal {d c b} [r zrevrangebyscore zset 3 0]
assert_equal {f e d} [r zrevrangebyscore zset 6 3]
assert_equal {g f e} [r zrevrangebyscore zset +inf 4]
assert_equal 3 [r zcount zset 0 3]
# exclusive range
assert_equal {b} [r zrangebyscore zset (-inf (2]
assert_equal {b c} [r zrangebyscore zset (0 (3]
assert_equal {e f} [r zrangebyscore zset (3 (6]
assert_equal {f} [r zrangebyscore zset (4 (+inf]
assert_equal {b} [r zrevrangebyscore zset (2 (-inf]
assert_equal {c b} [r zrevrangebyscore zset (3 (0]
assert_equal {f e} [r zrevrangebyscore zset (6 (3]
assert_equal {f} [r zrevrangebyscore zset (+inf (4]
assert_equal 2 [r zcount zset (0 (3]
# test empty ranges
r zrem zset a
r zrem zset g
# inclusive
assert_equal {} [r zrangebyscore zset 4 2]
assert_equal {} [r zrangebyscore zset 6 +inf]
assert_equal {} [r zrangebyscore zset -inf -6]
assert_equal {} [r zrevrangebyscore zset +inf 6]
assert_equal {} [r zrevrangebyscore zset -6 -inf]
# exclusive
assert_equal {} [r zrangebyscore zset (4 (2]
assert_equal {} [r zrangebyscore zset 2 (2]
assert_equal {} [r zrangebyscore zset (2 2]
assert_equal {} [r zrangebyscore zset (6 (+inf]
assert_equal {} [r zrangebyscore zset (-inf (-6]
assert_equal {} [r zrevrangebyscore zset (+inf (6]
assert_equal {} [r zrevrangebyscore zset (-6 (-inf]
# empty inner range
assert_equal {} [r zrangebyscore zset 2.4 2.6]
assert_equal {} [r zrangebyscore zset (2.4 2.6]
assert_equal {} [r zrangebyscore zset 2.4 (2.6]
assert_equal {} [r zrangebyscore zset (2.4 (2.6]
}
test "ZRANGEBYSCORE with WITHSCORES" {
create_default_zset
assert_equal {b 1 c 2 d 3} [r zrangebyscore zset 0 3 withscores]
assert_equal {d 3 c 2 b 1} [r zrevrangebyscore zset 3 0 withscores]
}
test "ZRANGEBYSCORE with LIMIT" {
create_default_zset
assert_equal {b c} [r zrangebyscore zset 0 10 LIMIT 0 2]
assert_equal {d e f} [r zrangebyscore zset 0 10 LIMIT 2 3]
assert_equal {d e f} [r zrangebyscore zset 0 10 LIMIT 2 10]
assert_equal {} [r zrangebyscore zset 0 10 LIMIT 20 10]
assert_equal {f e} [r zrevrangebyscore zset 10 0 LIMIT 0 2]
assert_equal {d c b} [r zrevrangebyscore zset 10 0 LIMIT 2 3]
assert_equal {d c b} [r zrevrangebyscore zset 10 0 LIMIT 2 10]
assert_equal {} [r zrevrangebyscore zset 10 0 LIMIT 20 10]
}
test "ZRANGEBYSCORE with LIMIT and WITHSCORES" {
create_default_zset
assert_equal {e 4 f 5} [r zrangebyscore zset 2 5 LIMIT 2 3 WITHSCORES]
assert_equal {d 3 c 2} [r zrevrangebyscore zset 5 2 LIMIT 2 3 WITHSCORES]
}
test "ZRANGEBYSCORE with non-value min or max" {
assert_error "*not a double*" {r zrangebyscore fooz str 1}
assert_error "*not a double*" {r zrangebyscore fooz 1 str}
assert_error "*not a double*" {r zrangebyscore fooz 1 NaN}
}
test "ZREMRANGEBYSCORE basics" {
proc remrangebyscore {min max} {
create_zset zset {1 a 2 b 3 c 4 d 5 e}
assert_equal 1 [r exists zset]
r zremrangebyscore zset $min $max
}
foreach {item score} [array get auxarray] {
lappend auxlist [list $score $item]
# inner range
assert_equal 3 [remrangebyscore 2 4]
assert_equal {a e} [r zrange zset 0 -1]
# start underflow
assert_equal 1 [remrangebyscore -10 1]
assert_equal {b c d e} [r zrange zset 0 -1]
# end overflow
assert_equal 1 [remrangebyscore 5 10]
assert_equal {a b c d} [r zrange zset 0 -1]
# switch min and max
assert_equal 0 [remrangebyscore 4 2]
assert_equal {a b c d e} [r zrange zset 0 -1]
# -inf to mid
assert_equal 3 [remrangebyscore -inf 3]
assert_equal {d e} [r zrange zset 0 -1]
# mid to +inf
assert_equal 3 [remrangebyscore 3 +inf]
assert_equal {a b} [r zrange zset 0 -1]
# -inf to +inf
assert_equal 5 [remrangebyscore -inf +inf]
assert_equal {} [r zrange zset 0 -1]
# exclusive min
assert_equal 4 [remrangebyscore (1 5]
assert_equal {a} [r zrange zset 0 -1]
assert_equal 3 [remrangebyscore (2 5]
assert_equal {a b} [r zrange zset 0 -1]
# exclusive max
assert_equal 4 [remrangebyscore 1 (5]
assert_equal {e} [r zrange zset 0 -1]
assert_equal 3 [remrangebyscore 1 (4]
assert_equal {d e} [r zrange zset 0 -1]
# exclusive min and max
assert_equal 3 [remrangebyscore (1 (5]
assert_equal {a e} [r zrange zset 0 -1]
# destroy when empty
assert_equal 5 [remrangebyscore 1 5]
assert_equal 0 [r exists zset]
}
test "ZREMRANGEBYSCORE with non-value min or max" {
assert_error "*not a double*" {r zremrangebyscore fooz str 1}
assert_error "*not a double*" {r zremrangebyscore fooz 1 str}
assert_error "*not a double*" {r zremrangebyscore fooz 1 NaN}
}
test "ZREMRANGEBYRANK basics" {
proc remrangebyrank {min max} {
create_zset zset {1 a 2 b 3 c 4 d 5 e}
assert_equal 1 [r exists zset]
r zremrangebyrank zset $min $max
}
set sorted [lsort -command zlistAlikeSort $auxlist]
set auxlist {}
foreach x $sorted {
lappend auxlist [lindex $x 1]
# inner range
assert_equal 3 [remrangebyrank 1 3]
assert_equal {a e} [r zrange zset 0 -1]
# start underflow
assert_equal 1 [remrangebyrank -10 0]
assert_equal {b c d e} [r zrange zset 0 -1]
# start overflow
assert_equal 0 [remrangebyrank 10 -1]
assert_equal {a b c d e} [r zrange zset 0 -1]
# end underflow
assert_equal 0 [remrangebyrank 0 -10]
assert_equal {a b c d e} [r zrange zset 0 -1]
# end overflow
assert_equal 5 [remrangebyrank 0 10]
assert_equal {} [r zrange zset 0 -1]
# destroy when empty
assert_equal 5 [remrangebyrank 0 4]
assert_equal 0 [r exists zset]
}
test "ZUNIONSTORE against non-existing key doesn't set destination - $encoding" {
r del zseta
assert_equal 0 [r zunionstore dst_key 1 zseta]
assert_equal 0 [r exists dst_key]
}
test "ZUNIONSTORE basics - $encoding" {
r del zseta zsetb zsetc
r zadd zseta 1 a
r zadd zseta 2 b
r zadd zseta 3 c
r zadd zsetb 1 b
r zadd zsetb 2 c
r zadd zsetb 3 d
assert_equal 4 [r zunionstore zsetc 2 zseta zsetb]
assert_equal {a 1 b 3 d 3 c 5} [r zrange zsetc 0 -1 withscores]
}
test "ZUNIONSTORE with weights - $encoding" {
assert_equal 4 [r zunionstore zsetc 2 zseta zsetb weights 2 3]
assert_equal {a 2 b 7 d 9 c 12} [r zrange zsetc 0 -1 withscores]
}
test "ZUNIONSTORE with a regular set and weights - $encoding" {
r del seta
r sadd seta a
r sadd seta b
r sadd seta c
assert_equal 4 [r zunionstore zsetc 2 seta zsetb weights 2 3]
assert_equal {a 2 b 5 c 8 d 9} [r zrange zsetc 0 -1 withscores]
}
test "ZUNIONSTORE with AGGREGATE MIN - $encoding" {
assert_equal 4 [r zunionstore zsetc 2 zseta zsetb aggregate min]
assert_equal {a 1 b 1 c 2 d 3} [r zrange zsetc 0 -1 withscores]
}
test "ZUNIONSTORE with AGGREGATE MAX - $encoding" {
assert_equal 4 [r zunionstore zsetc 2 zseta zsetb aggregate max]
assert_equal {a 1 b 2 c 3 d 3} [r zrange zsetc 0 -1 withscores]
}
test "ZINTERSTORE basics - $encoding" {
assert_equal 2 [r zinterstore zsetc 2 zseta zsetb]
assert_equal {b 3 c 5} [r zrange zsetc 0 -1 withscores]
}
test "ZINTERSTORE with weights - $encoding" {
assert_equal 2 [r zinterstore zsetc 2 zseta zsetb weights 2 3]
assert_equal {b 7 c 12} [r zrange zsetc 0 -1 withscores]
}
test "ZINTERSTORE with a regular set and weights - $encoding" {
r del seta
r sadd seta a
r sadd seta b
r sadd seta c
assert_equal 2 [r zinterstore zsetc 2 seta zsetb weights 2 3]
assert_equal {b 5 c 8} [r zrange zsetc 0 -1 withscores]
}
test "ZINTERSTORE with AGGREGATE MIN - $encoding" {
assert_equal 2 [r zinterstore zsetc 2 zseta zsetb aggregate min]
assert_equal {b 1 c 2} [r zrange zsetc 0 -1 withscores]
}
test "ZINTERSTORE with AGGREGATE MAX - $encoding" {
assert_equal 2 [r zinterstore zsetc 2 zseta zsetb aggregate max]
assert_equal {b 2 c 3} [r zrange zsetc 0 -1 withscores]
}
foreach cmd {ZUNIONSTORE ZINTERSTORE} {
test "$cmd with +inf/-inf scores - $encoding" {
r del zsetinf1 zsetinf2
r zadd zsetinf1 +inf key
r zadd zsetinf2 +inf key
r $cmd zsetinf3 2 zsetinf1 zsetinf2
assert_equal inf [r zscore zsetinf3 key]
r zadd zsetinf1 -inf key
r zadd zsetinf2 +inf key
r $cmd zsetinf3 2 zsetinf1 zsetinf2
assert_equal 0 [r zscore zsetinf3 key]
r zadd zsetinf1 +inf key
r zadd zsetinf2 -inf key
r $cmd zsetinf3 2 zsetinf1 zsetinf2
assert_equal 0 [r zscore zsetinf3 key]
r zadd zsetinf1 -inf key
r zadd zsetinf2 -inf key
r $cmd zsetinf3 2 zsetinf1 zsetinf2
assert_equal -inf [r zscore zsetinf3 key]
}
set fromredis [r zrange myzset 0 -1]
set delta 0
for {set i 0} {$i < [llength $fromredis]} {incr i} {
if {[lindex $fromredis $i] != [lindex $auxlist $i]} {
incr delta
test "$cmd with NaN weights $encoding" {
r del zsetinf1 zsetinf2
r zadd zsetinf1 1.0 key
r zadd zsetinf2 1.0 key
assert_error "*weight value is not a double*" {
r $cmd zsetinf3 2 zsetinf1 zsetinf2 weights nan nan
}
}
}
format $delta
} {0}
test {ZINCRBY - can create a new sorted set} {
r del zset
r zincrby zset 1 foo
list [r zrange zset 0 -1] [r zscore zset foo]
} {foo 1}
test {ZINCRBY - increment and decrement} {
r zincrby zset 2 foo
r zincrby zset 1 bar
set v1 [r zrange zset 0 -1]
r zincrby zset 10 bar
r zincrby zset -5 foo
r zincrby zset -5 bar
set v2 [r zrange zset 0 -1]
list $v1 $v2 [r zscore zset foo] [r zscore zset bar]
} {{bar foo} {foo bar} -2 6}
proc create_default_zset {} {
create_zset zset {-inf a 1 b 2 c 3 d 4 e 5 f +inf g}
}
test "ZRANGEBYSCORE/ZREVRANGEBYSCORE/ZCOUNT basics" {
create_default_zset
# inclusive range
assert_equal {a b c} [r zrangebyscore zset -inf 2]
assert_equal {b c d} [r zrangebyscore zset 0 3]
assert_equal {d e f} [r zrangebyscore zset 3 6]
assert_equal {e f g} [r zrangebyscore zset 4 +inf]
assert_equal {c b a} [r zrevrangebyscore zset 2 -inf]
assert_equal {d c b} [r zrevrangebyscore zset 3 0]
assert_equal {f e d} [r zrevrangebyscore zset 6 3]
assert_equal {g f e} [r zrevrangebyscore zset +inf 4]
assert_equal 3 [r zcount zset 0 3]
# exclusive range
assert_equal {b} [r zrangebyscore zset (-inf (2]
assert_equal {b c} [r zrangebyscore zset (0 (3]
assert_equal {e f} [r zrangebyscore zset (3 (6]
assert_equal {f} [r zrangebyscore zset (4 (+inf]
assert_equal {b} [r zrevrangebyscore zset (2 (-inf]
assert_equal {c b} [r zrevrangebyscore zset (3 (0]
assert_equal {f e} [r zrevrangebyscore zset (6 (3]
assert_equal {f} [r zrevrangebyscore zset (+inf (4]
assert_equal 2 [r zcount zset (0 (3]
}
basics ziplist
basics skiplist
proc stressers {encoding} {
if {$encoding == "ziplist"} {
# Little extra to allow proper fuzzing in the sorting stresser
r config set zset-max-ziplist-entries 256
r config set zset-max-ziplist-value 64
set elements 128
} elseif {$encoding == "skiplist"} {
r config set zset-max-ziplist-entries 0
r config set zset-max-ziplist-value 0
set elements 1000
} else {
puts "Unknown sorted set encoding"
exit
}
test "ZRANGEBYSCORE with WITHSCORES" {
create_default_zset
assert_equal {b 1 c 2 d 3} [r zrangebyscore zset 0 3 withscores]
assert_equal {d 3 c 2 b 1} [r zrevrangebyscore zset 3 0 withscores]
}
test "ZSCORE - $encoding" {
r del zscoretest
set aux {}
for {set i 0} {$i < $elements} {incr i} {
set score [expr rand()]
lappend aux $score
r zadd zscoretest $score $i
}
test "ZRANGEBYSCORE with LIMIT" {
create_default_zset
assert_equal {b c} [r zrangebyscore zset 0 10 LIMIT 0 2]
assert_equal {d e f} [r zrangebyscore zset 0 10 LIMIT 2 3]
assert_equal {d e f} [r zrangebyscore zset 0 10 LIMIT 2 10]
assert_equal {} [r zrangebyscore zset 0 10 LIMIT 20 10]
assert_equal {f e} [r zrevrangebyscore zset 10 0 LIMIT 0 2]
assert_equal {d c b} [r zrevrangebyscore zset 10 0 LIMIT 2 3]
assert_equal {d c b} [r zrevrangebyscore zset 10 0 LIMIT 2 10]
assert_equal {} [r zrevrangebyscore zset 10 0 LIMIT 20 10]
}
assert_encoding $encoding zscoretest
for {set i 0} {$i < $elements} {incr i} {
assert_equal [lindex $aux $i] [r zscore zscoretest $i]
}
}
test "ZRANGEBYSCORE with LIMIT and WITHSCORES" {
create_default_zset
assert_equal {e 4 f 5} [r zrangebyscore zset 2 5 LIMIT 2 3 WITHSCORES]
assert_equal {d 3 c 2} [r zrevrangebyscore zset 5 2 LIMIT 2 3 WITHSCORES]
}
test "ZSCORE after a DEBUG RELOAD - $encoding" {
r del zscoretest
set aux {}
for {set i 0} {$i < $elements} {incr i} {
set score [expr rand()]
lappend aux $score
r zadd zscoretest $score $i
}
test "ZRANGEBYSCORE with non-value min or max" {
assert_error "*not a double*" {r zrangebyscore fooz str 1}
assert_error "*not a double*" {r zrangebyscore fooz 1 str}
assert_error "*not a double*" {r zrangebyscore fooz 1 NaN}
}
r debug reload
assert_encoding $encoding zscoretest
for {set i 0} {$i < $elements} {incr i} {
assert_equal [lindex $aux $i] [r zscore zscoretest $i]
}
}
tags {"slow"} {
test {ZRANGEBYSCORE fuzzy test, 100 ranges in 1000 elements sorted set} {
test "ZSET sorting stresser - $encoding" {
set delta 0
for {set test 0} {$test < 2} {incr test} {
unset -nocomplain auxarray
array set auxarray {}
set auxlist {}
r del myzset
for {set i 0} {$i < $elements} {incr i} {
if {$test == 0} {
set score [expr rand()]
} else {
set score [expr int(rand()*10)]
}
set auxarray($i) $score
r zadd myzset $score $i
# Random update
if {[expr rand()] < .2} {
set j [expr int(rand()*1000)]
if {$test == 0} {
set score [expr rand()]
} else {
set score [expr int(rand()*10)]
}
set auxarray($j) $score
r zadd myzset $score $j
}
}
foreach {item score} [array get auxarray] {
lappend auxlist [list $score $item]
}
set sorted [lsort -command zlistAlikeSort $auxlist]
set auxlist {}
foreach x $sorted {
lappend auxlist [lindex $x 1]
}
assert_encoding $encoding myzset
set fromredis [r zrange myzset 0 -1]
set delta 0
for {set i 0} {$i < [llength $fromredis]} {incr i} {
if {[lindex $fromredis $i] != [lindex $auxlist $i]} {
incr delta
}
}
}
assert_equal 0 $delta
}
test "ZRANGEBYSCORE fuzzy test, 100 ranges in $elements element sorted set - $encoding" {
set err {}
r del zset
for {set i 0} {$i < 1000} {incr i} {
for {set i 0} {$i < $elements} {incr i} {
r zadd zset [expr rand()] $i
}
assert_encoding $encoding zset
for {set i 0} {$i < 100} {incr i} {
set min [expr rand()]
set max [expr rand()]
......@@ -337,198 +633,17 @@ start_server {tags {"zset"}} {
}
}
}
set _ $err
} {}
}
test "ZREMRANGEBYSCORE basics" {
proc remrangebyscore {min max} {
create_zset zset {1 a 2 b 3 c 4 d 5 e}
r zremrangebyscore zset $min $max
}
# inner range
assert_equal 3 [remrangebyscore 2 4]
assert_equal {a e} [r zrange zset 0 -1]
# start underflow
assert_equal 1 [remrangebyscore -10 1]
assert_equal {b c d e} [r zrange zset 0 -1]
# end overflow
assert_equal 1 [remrangebyscore 5 10]
assert_equal {a b c d} [r zrange zset 0 -1]
# switch min and max
assert_equal 0 [remrangebyscore 4 2]
assert_equal {a b c d e} [r zrange zset 0 -1]
# -inf to mid
assert_equal 3 [remrangebyscore -inf 3]
assert_equal {d e} [r zrange zset 0 -1]
# mid to +inf
assert_equal 3 [remrangebyscore 3 +inf]
assert_equal {a b} [r zrange zset 0 -1]
# -inf to +inf
assert_equal 5 [remrangebyscore -inf +inf]
assert_equal {} [r zrange zset 0 -1]
# exclusive min
assert_equal 4 [remrangebyscore (1 5]
assert_equal {a} [r zrange zset 0 -1]
assert_equal 3 [remrangebyscore (2 5]
assert_equal {a b} [r zrange zset 0 -1]
# exclusive max
assert_equal 4 [remrangebyscore 1 (5]
assert_equal {e} [r zrange zset 0 -1]
assert_equal 3 [remrangebyscore 1 (4]
assert_equal {d e} [r zrange zset 0 -1]
# exclusive min and max
assert_equal 3 [remrangebyscore (1 (5]
assert_equal {a e} [r zrange zset 0 -1]
}
test "ZREMRANGEBYSCORE with non-value min or max" {
assert_error "*not a double*" {r zremrangebyscore fooz str 1}
assert_error "*not a double*" {r zremrangebyscore fooz 1 str}
assert_error "*not a double*" {r zremrangebyscore fooz 1 NaN}
}
test "ZREMRANGEBYRANK basics" {
proc remrangebyrank {min max} {
create_zset zset {1 a 2 b 3 c 4 d 5 e}
r zremrangebyrank zset $min $max
}
# inner range
assert_equal 3 [remrangebyrank 1 3]
assert_equal {a e} [r zrange zset 0 -1]
# start underflow
assert_equal 1 [remrangebyrank -10 0]
assert_equal {b c d e} [r zrange zset 0 -1]
# start overflow
assert_equal 0 [remrangebyrank 10 -1]
assert_equal {a b c d e} [r zrange zset 0 -1]
# end underflow
assert_equal 0 [remrangebyrank 0 -10]
assert_equal {a b c d e} [r zrange zset 0 -1]
# end overflow
assert_equal 5 [remrangebyrank 0 10]
assert_equal {} [r zrange zset 0 -1]
}
test {ZUNIONSTORE against non-existing key doesn't set destination} {
r del zseta
list [r zunionstore dst_key 1 zseta] [r exists dst_key]
} {0 0}
test {ZUNIONSTORE basics} {
r del zseta zsetb zsetc
r zadd zseta 1 a
r zadd zseta 2 b
r zadd zseta 3 c
r zadd zsetb 1 b
r zadd zsetb 2 c
r zadd zsetb 3 d
list [r zunionstore zsetc 2 zseta zsetb] [r zrange zsetc 0 -1 withscores]
} {4 {a 1 b 3 d 3 c 5}}
test {ZUNIONSTORE with weights} {
list [r zunionstore zsetc 2 zseta zsetb weights 2 3] [r zrange zsetc 0 -1 withscores]
} {4 {a 2 b 7 d 9 c 12}}
test {ZUNIONSTORE with a regular set and weights} {
r del seta
r sadd seta a
r sadd seta b
r sadd seta c
list [r zunionstore zsetc 2 seta zsetb weights 2 3] [r zrange zsetc 0 -1 withscores]
} {4 {a 2 b 5 c 8 d 9}}
test {ZUNIONSTORE with AGGREGATE MIN} {
list [r zunionstore zsetc 2 zseta zsetb aggregate min] [r zrange zsetc 0 -1 withscores]
} {4 {a 1 b 1 c 2 d 3}}
test {ZUNIONSTORE with AGGREGATE MAX} {
list [r zunionstore zsetc 2 zseta zsetb aggregate max] [r zrange zsetc 0 -1 withscores]
} {4 {a 1 b 2 c 3 d 3}}
test {ZINTERSTORE basics} {
list [r zinterstore zsetc 2 zseta zsetb] [r zrange zsetc 0 -1 withscores]
} {2 {b 3 c 5}}
test {ZINTERSTORE with weights} {
list [r zinterstore zsetc 2 zseta zsetb weights 2 3] [r zrange zsetc 0 -1 withscores]
} {2 {b 7 c 12}}
test {ZINTERSTORE with a regular set and weights} {
r del seta
r sadd seta a
r sadd seta b
r sadd seta c
list [r zinterstore zsetc 2 seta zsetb weights 2 3] [r zrange zsetc 0 -1 withscores]
} {2 {b 5 c 8}}
test {ZINTERSTORE with AGGREGATE MIN} {
list [r zinterstore zsetc 2 zseta zsetb aggregate min] [r zrange zsetc 0 -1 withscores]
} {2 {b 1 c 2}}
test {ZINTERSTORE with AGGREGATE MAX} {
list [r zinterstore zsetc 2 zseta zsetb aggregate max] [r zrange zsetc 0 -1 withscores]
} {2 {b 2 c 3}}
foreach cmd {ZUNIONSTORE ZINTERSTORE} {
test "$cmd with +inf/-inf scores" {
r del zsetinf1 zsetinf2
r zadd zsetinf1 +inf key
r zadd zsetinf2 +inf key
r $cmd zsetinf3 2 zsetinf1 zsetinf2
assert_equal inf [r zscore zsetinf3 key]
r zadd zsetinf1 -inf key
r zadd zsetinf2 +inf key
r $cmd zsetinf3 2 zsetinf1 zsetinf2
assert_equal 0 [r zscore zsetinf3 key]
r zadd zsetinf1 +inf key
r zadd zsetinf2 -inf key
r $cmd zsetinf3 2 zsetinf1 zsetinf2
assert_equal 0 [r zscore zsetinf3 key]
r zadd zsetinf1 -inf key
r zadd zsetinf2 -inf key
r $cmd zsetinf3 2 zsetinf1 zsetinf2
assert_equal -inf [r zscore zsetinf3 key]
}
test "$cmd with NaN weights" {
r del zsetinf1 zsetinf2
r zadd zsetinf1 1.0 key
r zadd zsetinf2 1.0 key
assert_error "*weight value is not a double*" {
r $cmd zsetinf3 2 zsetinf1 zsetinf2 weights nan nan
}
assert_equal {} $err
}
}
tags {"slow"} {
test {ZSETs skiplist implementation backlink consistency test} {
test "ZSETs skiplist implementation backlink consistency test - $encoding" {
set diff 0
set elements 10000
for {set j 0} {$j < $elements} {incr j} {
r zadd myzset [expr rand()] "Element-$j"
r zrem myzset "Element-[expr int(rand()*$elements)]"
}
assert_encoding $encoding myzset
set l1 [r zrange myzset 0 -1]
set l2 [r zrevrange myzset 0 -1]
for {set j 0} {$j < [llength $l1]} {incr j} {
......@@ -536,20 +651,22 @@ start_server {tags {"zset"}} {
incr diff
}
}
format $diff
} {0}
assert_equal 0 $diff
}
test {ZSETs ZRANK augmented skip list stress testing} {
test "ZSETs ZRANK augmented skip list stress testing - $encoding" {
set err {}
r del myzset
for {set k 0} {$k < 10000} {incr k} {
set i [expr {$k%1000}]
for {set k 0} {$k < 2000} {incr k} {
set i [expr {$k % $elements}]
if {[expr rand()] < .2} {
r zrem myzset $i
} else {
set score [expr rand()]
r zadd myzset $score $i
assert_encoding $encoding myzset
}
set card [r zcard myzset]
if {$card > 0} {
set index [randomInt $card]
......@@ -561,20 +678,12 @@ start_server {tags {"zset"}} {
}
}
}
set _ $err
} {}
}
test {ZSET element can't be set to NaN with ZADD} {
assert_error "*not a double*" {r zadd myzset nan abc}
}
test {ZSET element can't be set to NaN with ZINCRBY} {
assert_error "*not a double*" {r zadd myzset nan abc}
assert_equal {} $err
}
}
test {ZINCRBY calls leading to NaN result in error} {
r zincrby myzset +inf abc
assert_error "*NaN*" {r zincrby myzset -inf abc}
tags {"slow"} {
stressers ziplist
stressers skiplist
}
}
Markdown is supported
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment