diff --git a/redis.conf b/redis.conf index a470c98c..e7a01eec 100644 --- a/redis.conf +++ b/redis.conf @@ -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) diff --git a/src/aof.c b/src/aof.c index 8ce6cd12..5d75c374 100644 --- a/src/aof.c +++ b/src/aof.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; + char cmd[]="*4\r\n$4\r\nZADD\r\n"; - while((de = dictNext(di)) != NULL) { - char cmd[]="*4\r\n$4\r\nZADD\r\n"; - robj *eleobj = dictGetEntryKey(de); - double *score = dictGetEntryVal(de); + 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; - 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; + 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"; diff --git a/src/config.c b/src/config.c index 48f90349..98fdb15d 100644 --- a/src/config.c +++ b/src/config.c @@ -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); } diff --git a/src/debug.c b/src/debug.c index c1fc26cf..080e2b2e 100644 --- a/src/debug.c +++ b/src/debug.c @@ -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; diff --git a/src/object.c b/src/object.c index 6a9b0214..c7c90c54 100644 --- a/src/object.c +++ b/src/object.c @@ -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) { diff --git a/src/rdb.c b/src/rdb.c index f14467d1..2557f5b8 100644 --- a/src/rdb.c +++ b/src/rdb.c @@ -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; + /* Save a sorted set value */ + if (o->encoding == REDIS_ENCODING_ZIPLIST) { + size_t l = ziplistBlobLen((unsigned char*)o->ptr); - 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; + 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 { diff --git a/src/redis.c b/src/redis.c index 9112adb5..b0c3179e 100644 --- a/src/redis.c +++ b/src/redis.c @@ -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; diff --git a/src/redis.h b/src/redis.h index 7e3a8c8f..5506e365 100644 --- a/src/redis.h +++ b/src/redis.h @@ -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); diff --git a/src/sort.c b/src/sort.c index 1cf8932e..ff275c95 100644 --- a/src/sort.c +++ b/src/sort.c @@ -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; diff --git a/src/t_zset.c b/src/t_zset.c index 27522367..7ce60349 100644 --- a/src/t_zset.c +++ b/src/t_zset.c @@ -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 (zobj->encoding == REDIS_ENCODING_ZIPLIST) { + unsigned char *eptr; - 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; - } - } + /* 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; + } + } - /* 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; + /* Remove and re-insert when score changed. */ + if (score != curscore) { + zobj->ptr = zzlDelete(zobj->ptr,eptr); + zobj->ptr = zzlInsert(zobj->ptr,ele,score); - /* New element */ - incrRefCount(ele); /* added to hash */ - znode = zslInsert(zs->zsl,score,ele); - incrRefCount(ele); /* added to skiplist */ + signalModifiedKey(c->db,key); + server.dirty++; + } - /* 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 { - dictEntry *de; - robj *curobj; - double *curscore; - int deleted; + 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); - /* Update score */ - 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]); + 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 if (zobj->encoding == REDIS_ENCODING_SKIPLIST) { + zset *zs = zobj->ptr; + zskiplistNode *znode; + dictEntry *de; + + ele = c->argv[3] = tryObjectEncoding(c->argv[3]); + de = dictFind(zs->dict,ele); + 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); + } + } 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; - 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 (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"); + } + + 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 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; + 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; - /* 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; + /* If reversed, get the last node in range as starting point. */ + if (reverse) + eptr = zzlLastInRange(zl,range); + else + eptr = zzlFirstInRange(zl,range); - /* 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; + /* No "first" element in the specified interval. */ + if (eptr == 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; + } + + /* 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 reversed, get the last node in range as starting point. */ + if (reverse) + ln = zslLastInRange(zsl,range); + else + ln = zslFirstInRange(zsl,range); + + /* 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 { + if (!zslValueLteMax(ln->score,&range)) break; + } + + /* 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; } } 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 (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--) { - if (reverse) - ln = ln->backward; - else - ln = ln->level[0].forward; - } - - 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; - } else { - /* Element should have score >= range.max */ - if (ln->score < range.max) 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); - } - - if (reverse) - ln = ln->backward; - else - ln = ln->level[0].forward; + 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); + 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; + + 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 { - double *score = dictGetEntryVal(de); - - addReplyDouble(c,*score); + 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); - 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; - } + redisAssert(ele->encoding == REDIS_ENCODING_RAW); + if (zobj->encoding == REDIS_ENCODING_ZIPLIST) { + unsigned char *zl = zobj->ptr; + unsigned char *eptr, *sptr; - score = dictGetEntryVal(de); - rank = zslistTypeGetRank(zsl, *score, c->argv[2]); - if (rank) { - if (reverse) { - addReplyLongLong(c, zsl->length - rank); + 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); + } + + if (eptr != NULL) { + if (reverse) + addReplyLongLong(c,llen-rank); + else + addReplyLongLong(c,rank-1); } else { - addReplyLongLong(c, rank-1); + addReply(c,shared.nullbulk); + } + } else if (zobj->encoding == REDIS_ENCODING_SKIPLIST) { + zset *zs = zobj->ptr; + zskiplist *zsl = zs->zsl; + dictEntry *de; + double score; + + 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 { + addReply(c,shared.nullbulk); } } else { - addReply(c,shared.nullbulk); + redisPanic("Unknown sorted set encoding"); } } diff --git a/src/util.c b/src/util.c index e304ff83..599dc690 100644 --- a/src/util.c +++ b/src/util.c @@ -1,6 +1,7 @@ #include "redis.h" #include #include +#include /* 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 diff --git a/src/ziplist.c b/src/ziplist.c index 55bb662b..1c492f25 100644 --- a/src/ziplist.c +++ b/src/ziplist.c @@ -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 diff --git a/tests/unit/type/zset.tcl b/tests/unit/type/zset.tcl index 6b8fc54a..4fa7af1b 100644 --- a/tests/unit/type/zset.tcl +++ b/tests/unit/type/zset.tcl @@ -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] - } - - 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 {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 + 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 } - 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 "Check encoding - $encoding" { + r del ztmp + r zadd ztmp 10 x + assert_encoding $encoding ztmp + } + + 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] + } + + test "ZSET element can't be set to NaN with ZADD - $encoding" { + 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} + } + + 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 } - } - 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 + # 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] } - 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 "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 _ $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)] + # 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] + } + + 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 } - set auxarray($i) $score - r zadd myzset $score $i - # Random update - if {[expr rand()] < .2} { - set j [expr int(rand()*1000)] + } + } + } + + 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 "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 + } + + assert_encoding $encoding zscoretest + for {set i 0} {$i < $elements} {incr i} { + assert_equal [lindex $aux $i] [r zscore zscoretest $i] + } + } + + 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 + } + + r debug reload + assert_encoding $encoding zscoretest + for {set i 0} {$i < $elements} {incr i} { + assert_equal [lindex $aux $i] [r zscore zscoretest $i] + } + } + + 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($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] - } - 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 + 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 } - 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] - } - - 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} - } - - tags {"slow"} { - test {ZRANGEBYSCORE fuzzy test, 100 ranges in 1000 elements sorted set} { + 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 + assert_equal {} $err } - # 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 - } - } - } - - 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 - } {} + assert_equal {} $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} - } - - 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 } }