X-Git-Url: https://git.saurik.com/redis.git/blobdiff_plain/f9c6f39b2b0039cf29df6597d41c124048c825cd..9aba884b348432a3ef802723a5f6692b353bbaa8:/src/t_zset.c?ds=sidebyside diff --git a/src/t_zset.c b/src/t_zset.c index 6a56c3b4..d482d4c2 100644 --- a/src/t_zset.c +++ b/src/t_zset.c @@ -17,7 +17,7 @@ /* This skiplist implementation is almost a C translation of the original * algorithm described by William Pugh in "Skip Lists: A Probabilistic * Alternative to Balanced Trees", modified in three ways: - * a) this implementation allows for repeated values. + * a) this implementation allows for repeated scores. * b) the comparison is not just by key (our 'score') but by satellite data. * c) there is a back pointer, so it's a doubly linked list with the back * pointers being only at "level 1". This allows to traverse the list @@ -64,6 +64,10 @@ void zslFree(zskiplist *zsl) { zfree(zsl); } +/* Returns a random level for the new skiplist node we are going to create. + * The return value of this function is between 1 and ZSKIPLIST_MAXLEVEL + * (both inclusive), with a powerlaw-alike distribution where higher + * levels are less likely to be returned. */ int zslRandomLevel(void) { int level = 1; while ((random()&0xFFFF) < (ZSKIPLIST_P * 0xFFFF)) @@ -76,6 +80,7 @@ zskiplistNode *zslInsert(zskiplist *zsl, double score, robj *obj) { unsigned int rank[ZSKIPLIST_MAXLEVEL]; int i, level; + redisAssert(!isnan(score)); x = zsl->header; for (i = zsl->level-1; i >= 0; i--) { /* store rank that is crossed to reach the insert position */ @@ -497,7 +502,7 @@ int zzlIsInRange(unsigned char *zl, zrangespec *range) { return 0; p = ziplistIndex(zl,-1); /* Last score. */ - redisAssert(p != NULL); + if (p == NULL) return 0; /* Empty sorted set */ score = zzlGetScore(p); if (!zslValueGteMin(score,range)) return 0; @@ -578,7 +583,7 @@ unsigned char *zzlFind(unsigned char *zl, robj *ele, double *score) { ele = getDecodedObject(ele); while (eptr != NULL) { sptr = ziplistNext(zl,eptr); - redisAssert(sptr != NULL); + redisAssertWithInfo(NULL,ele,sptr != NULL); if (ziplistCompare(eptr,ele->ptr,sdslen(ele->ptr))) { /* Matching element, pull out score. */ @@ -612,7 +617,7 @@ unsigned char *zzlInsertAt(unsigned char *zl, unsigned char *eptr, robj *ele, do int scorelen; size_t offset; - redisAssert(ele->encoding == REDIS_ENCODING_RAW); + redisAssertWithInfo(NULL,ele,ele->encoding == REDIS_ENCODING_RAW); scorelen = d2string(scorebuf,sizeof(scorebuf),score); if (eptr == NULL) { zl = ziplistPush(zl,ele->ptr,sdslen(ele->ptr),ZIPLIST_TAIL); @@ -624,7 +629,7 @@ unsigned char *zzlInsertAt(unsigned char *zl, unsigned char *eptr, robj *ele, do eptr = zl+offset; /* Insert score after the element. */ - redisAssert((sptr = ziplistNext(zl,eptr)) != NULL); + redisAssertWithInfo(NULL,ele,(sptr = ziplistNext(zl,eptr)) != NULL); zl = ziplistInsert(zl,sptr,(unsigned char*)scorebuf,scorelen); } @@ -640,7 +645,7 @@ unsigned char *zzlInsert(unsigned char *zl, robj *ele, double score) { ele = getDecodedObject(ele); while (eptr != NULL) { sptr = ziplistNext(zl,eptr); - redisAssert(sptr != NULL); + redisAssertWithInfo(NULL,ele,sptr != NULL); s = zzlGetScore(sptr); if (s > score) { @@ -745,13 +750,13 @@ void zsetConvert(robj *zobj, int encoding) { zs->zsl = zslCreate(); eptr = ziplistIndex(zl,0); - redisAssert(eptr != NULL); + redisAssertWithInfo(NULL,zobj,eptr != NULL); sptr = ziplistNext(zl,eptr); - redisAssert(sptr != NULL); + redisAssertWithInfo(NULL,zobj,sptr != NULL); while (eptr != NULL) { score = zzlGetScore(sptr); - redisAssert(ziplistGet(eptr,&vstr,&vlen,&vlong)); + redisAssertWithInfo(NULL,zobj,ziplistGet(eptr,&vstr,&vlen,&vlong)); if (vstr == NULL) ele = createStringObjectFromLongLong(vlong); else @@ -759,7 +764,7 @@ void zsetConvert(robj *zobj, int encoding) { /* Has incremented refcount since it was just created. */ node = zslInsert(zs->zsl,score,ele); - redisAssert(dictAdd(zs->dict,ele,&node->score) == DICT_OK); + redisAssertWithInfo(NULL,zobj,dictAdd(zs->dict,ele,&node->score) == DICT_OK); incrRefCount(ele); /* Added to dictionary. */ zzlNext(zl,&eptr,&sptr); } @@ -900,8 +905,8 @@ void zaddGenericCommand(redisClient *c, int incr) { ele = c->argv[3+j*2] = tryObjectEncoding(c->argv[3+j*2]); de = dictFind(zs->dict,ele); if (de != NULL) { - curobj = dictGetEntryKey(de); - curscore = *(double*)dictGetEntryVal(de); + curobj = dictGetKey(de); + curscore = *(double*)dictGetVal(de); if (incr) { score += curscore; @@ -918,10 +923,10 @@ void zaddGenericCommand(redisClient *c, int incr) { * 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)); + redisAssertWithInfo(c,curobj,zslDelete(zs->zsl,curscore,curobj)); znode = zslInsert(zs->zsl,score,curobj); incrRefCount(curobj); /* Re-inserted in skiplist. */ - dictGetEntryVal(de) = &znode->score; /* Update score ptr. */ + dictGetVal(de) = &znode->score; /* Update score ptr. */ signalModifiedKey(c->db,key); server.dirty++; @@ -929,7 +934,7 @@ void zaddGenericCommand(redisClient *c, int incr) { } else { znode = zslInsert(zs->zsl,score,ele); incrRefCount(ele); /* Inserted in skiplist. */ - redisAssert(dictAdd(zs->dict,ele,&znode->score) == DICT_OK); + redisAssertWithInfo(c,NULL,dictAdd(zs->dict,ele,&znode->score) == DICT_OK); incrRefCount(ele); /* Added to dictionary. */ signalModifiedKey(c->db,key); @@ -987,8 +992,8 @@ void zremCommand(redisClient *c) { deleted++; /* Delete from the skiplist */ - score = *(double*)dictGetEntryVal(de); - redisAssert(zslDelete(zs->zsl,score,c->argv[j])); + score = *(double*)dictGetVal(de); + redisAssertWithInfo(c,c->argv[j],zslDelete(zs->zsl,score,c->argv[j])); /* Delete from the hash table */ dictDelete(zs->dict,c->argv[j]); @@ -1018,7 +1023,7 @@ void zremrangebyscoreCommand(redisClient *c) { /* Parse the range arguments. */ if (zslParseRange(c->argv[2],c->argv[3],&range) != REDIS_OK) { - addReplyError(c,"min or max is not a double"); + addReplyError(c,"min or max is not a float"); return; } @@ -1249,7 +1254,7 @@ int zuiNext(zsetopsrc *op, zsetopval *val) { if (val->flags & OPVAL_DIRTY_ROBJ) decrRefCount(val->ele); - bzero(val,sizeof(zsetopval)); + memset(val,0,sizeof(zsetopval)); if (op->type == REDIS_SET) { iterset *it = &op->iter.set; @@ -1263,7 +1268,7 @@ int zuiNext(zsetopsrc *op, zsetopval *val) { } else if (op->encoding == REDIS_ENCODING_HT) { if (it->ht.de == NULL) return 0; - val->ele = dictGetEntryKey(it->ht.de); + val->ele = dictGetKey(it->ht.de); val->score = 1.0; /* Move to next element. */ @@ -1397,7 +1402,7 @@ int zuiFind(zsetopsrc *op, zsetopval *val, double *score) { } else if (op->encoding == REDIS_ENCODING_SKIPLIST) { dictEntry *de; if ((de = dictFind(it->sl.zs->dict,val->ele)) != NULL) { - *score = *(double*)dictGetEntryVal(de); + *score = *(double*)dictGetVal(de); return 1; } else { return 0; @@ -1417,7 +1422,7 @@ int zuiCompareByCardinality(const void *s1, const void *s2) { #define REDIS_AGGR_SUM 1 #define REDIS_AGGR_MIN 2 #define REDIS_AGGR_MAX 3 -#define zunionInterDictValue(_e) (dictGetEntryVal(_e) == NULL ? 1.0 : *(double*)dictGetEntryVal(_e)) +#define zunionInterDictValue(_e) (dictGetVal(_e) == NULL ? 1.0 : *(double*)dictGetVal(_e)) inline static void zunionInterAggregate(double *target, double val, int aggregate) { if (aggregate == REDIS_AGGR_SUM) { @@ -1437,7 +1442,8 @@ inline static void zunionInterAggregate(double *target, double val, int aggregat } void zunionInterGenericCommand(redisClient *c, robj *dstkey, int op) { - int i, j, setnum; + int i, j; + long setnum; int aggregate = REDIS_AGGR_SUM; zsetopsrc *src; zsetopval zval; @@ -1449,7 +1455,9 @@ void zunionInterGenericCommand(redisClient *c, robj *dstkey, int op) { int touched = 0; /* expect setnum input keys to be given */ - setnum = atoi(c->argv[2]->ptr); + if ((getLongFromObjectOrReply(c, c->argv[2], &setnum, NULL) != REDIS_OK)) + return; + if (setnum < 1) { addReplyError(c, "at least 1 input key is needed for ZUNIONSTORE/ZINTERSTORE"); @@ -1493,7 +1501,7 @@ void zunionInterGenericCommand(redisClient *c, robj *dstkey, int op) { j++; remaining--; for (i = 0; i < setnum; i++, j++, remaining--) { if (getDoubleFromObjectOrReply(c,c->argv[j],&src[i].weight, - "weight value is not a double") != REDIS_OK) + "weight value is not a float") != REDIS_OK) { zfree(src); return; @@ -1541,6 +1549,8 @@ void zunionInterGenericCommand(redisClient *c, robj *dstkey, int op) { double score, value; score = src[0].weight * zval.score; + if (isnan(score)) score = 0; + for (j = 1; j < setnum; j++) { /* It is not safe to access the zset we are * iterating, so explicitly check for equal object. */ @@ -1583,6 +1593,7 @@ void zunionInterGenericCommand(redisClient *c, robj *dstkey, int op) { /* Initialize score */ score = src[i].weight * zval.score; + if (isnan(score)) score = 0; /* Because the inputs are sorted by size, it's only possible * for sets at larger indices to hold this element. */ @@ -1698,12 +1709,12 @@ void zrangeGenericCommand(redisClient *c, int reverse) { else eptr = ziplistIndex(zl,2*start); - redisAssert(eptr != NULL); + redisAssertWithInfo(c,zobj,eptr != NULL); sptr = ziplistNext(zl,eptr); while (rangelen--) { - redisAssert(eptr != NULL && sptr != NULL); - redisAssert(ziplistGet(eptr,&vstr,&vlen,&vlong)); + redisAssertWithInfo(c,zobj,eptr != NULL && sptr != NULL); + redisAssertWithInfo(c,zobj,ziplistGet(eptr,&vstr,&vlen,&vlong)); if (vstr == NULL) addReplyBulkLongLong(c,vlong); else @@ -1736,7 +1747,7 @@ void zrangeGenericCommand(redisClient *c, int reverse) { } while(rangelen--) { - redisAssert(ln != NULL); + redisAssertWithInfo(c,zobj,ln != NULL); ele = ln->obj; addReplyBulk(c,ele); if (withscores) @@ -1756,13 +1767,12 @@ void zrevrangeCommand(redisClient *c) { zrangeGenericCommand(c,1); } -/* This command implements ZRANGEBYSCORE, ZREVRANGEBYSCORE and ZCOUNT. - * If "justcount", only the number of elements in the range is returned. */ -void genericZrangebyscoreCommand(redisClient *c, int reverse, int justcount) { +/* This command implements ZRANGEBYSCORE, ZREVRANGEBYSCORE. */ +void genericZrangebyscoreCommand(redisClient *c, int reverse) { zrangespec range; robj *key = c->argv[1]; - robj *emptyreply, *zobj; - int offset = 0, limit = -1; + robj *zobj; + long offset = 0, limit = -1; int withscores = 0; unsigned long rangelen = 0; void *replylen = NULL; @@ -1778,7 +1788,7 @@ void genericZrangebyscoreCommand(redisClient *c, int reverse, int justcount) { } if (zslParseRange(c->argv[minidx],c->argv[maxidx],&range) != REDIS_OK) { - addReplyError(c,"min or max is not a double"); + addReplyError(c,"min or max is not a float"); return; } @@ -1793,8 +1803,8 @@ void genericZrangebyscoreCommand(redisClient *c, int reverse, int justcount) { pos++; remaining--; withscores = 1; } else if (remaining >= 3 && !strcasecmp(c->argv[pos]->ptr,"limit")) { - offset = atoi(c->argv[pos+1]->ptr); - limit = atoi(c->argv[pos+2]->ptr); + if ((getLongFromObjectOrReply(c, c->argv[pos+1], &offset, NULL) != REDIS_OK) || + (getLongFromObjectOrReply(c, c->argv[pos+2], &limit, NULL) != REDIS_OK)) return; pos += 3; remaining -= 3; } else { addReply(c,shared.syntaxerr); @@ -1804,8 +1814,7 @@ void genericZrangebyscoreCommand(redisClient *c, int reverse, int justcount) { } /* Ok, lookup the key and get the range */ - emptyreply = justcount ? shared.czero : shared.emptymultibulk; - if ((zobj = lookupKeyReadOrReply(c,key,emptyreply)) == NULL || + if ((zobj = lookupKeyReadOrReply(c,key,shared.emptymultibulk)) == NULL || checkType(c,zobj,REDIS_ZSET)) return; if (zobj->encoding == REDIS_ENCODING_ZIPLIST) { @@ -1817,34 +1826,36 @@ void genericZrangebyscoreCommand(redisClient *c, int reverse, int justcount) { double score; /* If reversed, get the last node in range as starting point. */ - if (reverse) + if (reverse) { eptr = zzlLastInRange(zl,range); - else + } else { eptr = zzlFirstInRange(zl,range); + } /* No "first" element in the specified interval. */ if (eptr == NULL) { - addReply(c,emptyreply); + addReply(c, shared.emptymultibulk); return; } /* Get score pointer for the first element. */ - redisAssert(eptr != NULL); + redisAssertWithInfo(c,zobj,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); + 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) + while (eptr && offset--) { + if (reverse) { zzlPrev(zl,&eptr,&sptr); - else + } else { zzlNext(zl,&eptr,&sptr); + } + } while (eptr && limit--) { score = zzlGetScore(sptr); @@ -1856,24 +1867,26 @@ void genericZrangebyscoreCommand(redisClient *c, int reverse, int justcount) { if (!zslValueLteMax(score,&range)) break; } - /* Do our magic */ + /* We know the element exists, so ziplistGet should always succeed */ + redisAssertWithInfo(c,zobj,ziplistGet(eptr,&vstr,&vlen,&vlong)); + 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); + if (vstr == NULL) { + addReplyBulkLongLong(c,vlong); + } else { + addReplyBulkCBuffer(c,vstr,vlen); + } + + if (withscores) { + addReplyDouble(c,score); } /* Move to next node */ - if (reverse) + if (reverse) { zzlPrev(zl,&eptr,&sptr); - else + } else { zzlNext(zl,&eptr,&sptr); + } } } else if (zobj->encoding == REDIS_ENCODING_SKIPLIST) { zset *zs = zobj->ptr; @@ -1881,27 +1894,32 @@ void genericZrangebyscoreCommand(redisClient *c, int reverse, int justcount) { zskiplistNode *ln; /* If reversed, get the last node in range as starting point. */ - if (reverse) + if (reverse) { ln = zslLastInRange(zsl,range); - else + } else { ln = zslFirstInRange(zsl,range); + } /* No "first" element in the specified interval. */ if (ln == NULL) { - addReply(c,emptyreply); + addReply(c, shared.emptymultibulk); 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); + 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 && offset--) { + if (reverse) { + ln = ln->backward; + } else { + ln = ln->level[0].forward; + } + } while (ln && limit--) { /* Abort when the node is no longer in range. */ @@ -1911,39 +1929,114 @@ void genericZrangebyscoreCommand(redisClient *c, int reverse, int justcount) { if (!zslValueLteMax(ln->score,&range)) break; } - /* Do our magic */ rangelen++; - if (!justcount) { - addReplyBulk(c,ln->obj); - if (withscores) - addReplyDouble(c,ln->score); + addReplyBulk(c,ln->obj); + + if (withscores) { + addReplyDouble(c,ln->score); } /* Move to next node */ - ln = reverse ? ln->backward : ln->level[0].forward; + if (reverse) { + ln = ln->backward; + } else { + ln = ln->level[0].forward; + } } } else { redisPanic("Unknown sorted set encoding"); } - if (justcount) { - addReplyLongLong(c,(long)rangelen); - } else { - if (withscores) rangelen *= 2; - setDeferredMultiBulkLength(c,replylen,rangelen); + if (withscores) { + rangelen *= 2; } + + setDeferredMultiBulkLength(c, replylen, rangelen); } void zrangebyscoreCommand(redisClient *c) { - genericZrangebyscoreCommand(c,0,0); + genericZrangebyscoreCommand(c,0); } void zrevrangebyscoreCommand(redisClient *c) { - genericZrangebyscoreCommand(c,1,0); + genericZrangebyscoreCommand(c,1); } void zcountCommand(redisClient *c) { - genericZrangebyscoreCommand(c,0,1); + robj *key = c->argv[1]; + robj *zobj; + zrangespec range; + int count = 0; + + /* Parse the range arguments */ + if (zslParseRange(c->argv[2],c->argv[3],&range) != REDIS_OK) { + addReplyError(c,"min or max is not a float"); + return; + } + + /* Lookup the sorted set */ + if ((zobj = lookupKeyReadOrReply(c, key, shared.czero)) == NULL || + checkType(c, zobj, REDIS_ZSET)) return; + + if (zobj->encoding == REDIS_ENCODING_ZIPLIST) { + unsigned char *zl = zobj->ptr; + unsigned char *eptr, *sptr; + double score; + + /* Use the first element in range as the starting point */ + eptr = zzlFirstInRange(zl,range); + + /* No "first" element */ + if (eptr == NULL) { + addReply(c, shared.czero); + return; + } + + /* First element is in range */ + sptr = ziplistNext(zl,eptr); + score = zzlGetScore(sptr); + redisAssertWithInfo(c,zobj,zslValueLteMax(score,&range)); + + /* Iterate over elements in range */ + while (eptr) { + score = zzlGetScore(sptr); + + /* Abort when the node is no longer in range. */ + if (!zslValueLteMax(score,&range)) { + break; + } else { + count++; + zzlNext(zl,&eptr,&sptr); + } + } + } else if (zobj->encoding == REDIS_ENCODING_SKIPLIST) { + zset *zs = zobj->ptr; + zskiplist *zsl = zs->zsl; + zskiplistNode *zn; + unsigned long rank; + + /* Find first element in range */ + zn = zslFirstInRange(zsl, range); + + /* Use rank of first element, if any, to determine preliminary count */ + if (zn != NULL) { + rank = zslGetRank(zsl, zn->score, zn->obj); + count = (zsl->length - (rank - 1)); + + /* Find last element in range */ + zn = zslLastInRange(zsl, range); + + /* Use rank of last element, if any, to determine the actual count */ + if (zn != NULL) { + rank = zslGetRank(zsl, zn->score, zn->obj); + count -= (zsl->length - rank); + } + } + } else { + redisPanic("Unknown sorted set encoding"); + } + + addReplyLongLong(c, count); } void zcardCommand(redisClient *c) { @@ -1976,7 +2069,7 @@ void zscoreCommand(redisClient *c) { c->argv[2] = tryObjectEncoding(c->argv[2]); de = dictFind(zs->dict,c->argv[2]); if (de != NULL) { - score = *(double*)dictGetEntryVal(de); + score = *(double*)dictGetVal(de); addReplyDouble(c,score); } else { addReply(c,shared.nullbulk); @@ -1997,15 +2090,15 @@ void zrankGenericCommand(redisClient *c, int reverse) { checkType(c,zobj,REDIS_ZSET)) return; llen = zsetLength(zobj); - redisAssert(ele->encoding == REDIS_ENCODING_RAW); + redisAssertWithInfo(c,ele,ele->encoding == REDIS_ENCODING_RAW); if (zobj->encoding == REDIS_ENCODING_ZIPLIST) { unsigned char *zl = zobj->ptr; unsigned char *eptr, *sptr; eptr = ziplistIndex(zl,0); - redisAssert(eptr != NULL); + redisAssertWithInfo(c,zobj,eptr != NULL); sptr = ziplistNext(zl,eptr); - redisAssert(sptr != NULL); + redisAssertWithInfo(c,zobj,sptr != NULL); rank = 1; while(eptr != NULL) { @@ -2032,9 +2125,9 @@ void zrankGenericCommand(redisClient *c, int reverse) { ele = c->argv[2] = tryObjectEncoding(c->argv[2]); de = dictFind(zs->dict,ele); if (de != NULL) { - score = *(double*)dictGetEntryVal(de); + score = *(double*)dictGetVal(de); rank = zslGetRank(zsl,score,ele); - redisAssert(rank); /* Existing elements always have a rank. */ + redisAssertWithInfo(c,ele,rank); /* Existing elements always have a rank. */ if (reverse) addReplyLongLong(c,llen-rank); else