X-Git-Url: https://git.saurik.com/redis.git/blobdiff_plain/255eebe22167e00f74e359bc71718225d6bd70c8..62d774e5ba6056be39012ecedb88c3fec10304fb:/src/t_zset.c diff --git a/src/t_zset.c b/src/t_zset.c index a2882702..7ffed366 100644 --- a/src/t_zset.c +++ b/src/t_zset.c @@ -174,12 +174,6 @@ int zslDelete(zskiplist *zsl, double score, robj *obj) { return 0; /* not found */ } -/* Struct to hold a inclusive/exclusive range spec. */ -typedef struct { - double min, max; - 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); } @@ -188,10 +182,6 @@ static int zslValueLteMax(double value, zrangespec *spec) { return spec->maxex ? (value < spec->max) : (value <= spec->max); } -static int zslValueInRange(double value, zrangespec *spec) { - return zslValueGteMin(value,spec) && zslValueLteMax(value,spec); -} - /* Returns if there is a part of the zset is in range. */ int zslIsInRange(zskiplist *zsl, zrangespec *range) { zskiplistNode *x; @@ -226,10 +216,12 @@ zskiplistNode *zslFirstInRange(zskiplist *zsl, zrangespec range) { x = x->level[i].forward; } - /* The tail is in range, so the previous block should always return a - * node that is non-NULL and the last one to be out of range. */ + /* This is an inner range, so the next node cannot be NULL. */ x = x->level[0].forward; - redisAssert(x != NULL && zslValueInRange(x->score,&range)); + redisAssert(x != NULL); + + /* Check if score <= max. */ + if (!zslValueLteMax(x->score,&range)) return NULL; return x; } @@ -250,9 +242,11 @@ zskiplistNode *zslLastInRange(zskiplist *zsl, zrangespec range) { x = x->level[i].forward; } - /* The header is in range, so the previous block should always return a - * node that is non-NULL and in range. */ - redisAssert(x != NULL && zslValueInRange(x->score,&range)); + /* 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; } @@ -519,8 +513,7 @@ int zzlIsInRange(unsigned char *zl, zrangespec *range) { /* Find pointer to the first element contained in the specified range. * Returns NULL when no element is contained in the range. */ -unsigned char *zzlFirstInRange(robj *zobj, zrangespec range) { - unsigned char *zl = zobj->ptr; +unsigned char *zzlFirstInRange(unsigned char *zl, zrangespec range) { unsigned char *eptr = ziplistIndex(zl,0), *sptr; double score; @@ -532,8 +525,12 @@ unsigned char *zzlFirstInRange(robj *zobj, zrangespec range) { redisAssert(sptr != NULL); score = zzlGetScore(sptr); - if (zslValueGteMin(score,&range)) - return eptr; + if (zslValueGteMin(score,&range)) { + /* Check if score <= max. */ + if (zslValueLteMax(score,&range)) + return eptr; + return NULL; + } /* Move to next element. */ eptr = ziplistNext(zl,sptr); @@ -544,8 +541,7 @@ unsigned char *zzlFirstInRange(robj *zobj, zrangespec range) { /* Find pointer to the last element contained in the specified range. * Returns NULL when no element is contained in the range. */ -unsigned char *zzlLastInRange(robj *zobj, zrangespec range) { - unsigned char *zl = zobj->ptr; +unsigned char *zzlLastInRange(unsigned char *zl, zrangespec range) { unsigned char *eptr = ziplistIndex(zl,-2), *sptr; double score; @@ -557,8 +553,12 @@ unsigned char *zzlLastInRange(robj *zobj, zrangespec range) { redisAssert(sptr != NULL); score = zzlGetScore(sptr); - if (zslValueLteMax(score,&range)) - return eptr; + 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. */ @@ -572,8 +572,7 @@ unsigned char *zzlLastInRange(robj *zobj, zrangespec range) { return NULL; } -unsigned char *zzlFind(robj *zobj, robj *ele, double *score) { - unsigned char *zl = zobj->ptr; +unsigned char *zzlFind(unsigned char *zl, robj *ele, double *score) { unsigned char *eptr = ziplistIndex(zl,0), *sptr; ele = getDecodedObject(ele); @@ -598,23 +597,20 @@ unsigned char *zzlFind(robj *zobj, robj *ele, double *score) { /* Delete (element,score) pair from ziplist. Use local copy of eptr because we * don't want to modify the one given as argument. */ -int zzlDelete(robj *zobj, unsigned char *eptr) { - unsigned char *zl = zobj->ptr; +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); - zobj->ptr = zl; - return REDIS_OK; + return zl; } -int zzlInsertAt(robj *zobj, robj *ele, double score, unsigned char *eptr) { - unsigned char *zl = zobj->ptr; +unsigned char *zzlInsertAt(unsigned char *zl, unsigned char *eptr, robj *ele, double score) { unsigned char *sptr; char scorebuf[128]; int scorelen; - int offset; + size_t offset; redisAssert(ele->encoding == REDIS_ENCODING_RAW); scorelen = d2string(scorebuf,sizeof(scorebuf),score); @@ -632,14 +628,12 @@ int zzlInsertAt(robj *zobj, robj *ele, double score, unsigned char *eptr) { zl = ziplistInsert(zl,sptr,(unsigned char*)scorebuf,scorelen); } - zobj->ptr = zl; - return REDIS_OK; + return zl; } /* Insert (element,score) pair in ziplist. This function assumes the element is * not yet present in the list. */ -int zzlInsert(robj *zobj, robj *ele, double score) { - unsigned char *zl = zobj->ptr; +unsigned char *zzlInsert(unsigned char *zl, robj *ele, double score) { unsigned char *eptr = ziplistIndex(zl,0), *sptr; double s; @@ -653,12 +647,12 @@ int zzlInsert(robj *zobj, robj *ele, double 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. */ - zzlInsertAt(zobj,ele,score,eptr); + 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) { - zzlInsertAt(zobj,ele,score,eptr); + zl = zzlInsertAt(zl,eptr,ele,score); break; } } @@ -669,21 +663,21 @@ int zzlInsert(robj *zobj, robj *ele, double score) { /* Push on tail of list when it was not yet inserted. */ if (eptr == NULL) - zzlInsertAt(zobj,ele,score,NULL); + zl = zzlInsertAt(zl,NULL,ele,score); decrRefCount(ele); - return REDIS_OK; + return zl; } -unsigned long zzlDeleteRangeByScore(robj *zobj, zrangespec range) { - unsigned char *zl = zobj->ptr; +unsigned char *zzlDeleteRangeByScore(unsigned char *zl, zrangespec range, unsigned long *deleted) { unsigned char *eptr, *sptr; double score; - unsigned long deleted = 0; + unsigned long num = 0; - eptr = zzlFirstInRange(zobj,range); - if (eptr == NULL) return deleted; + 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. */ @@ -693,33 +687,35 @@ unsigned long zzlDeleteRangeByScore(robj *zobj, zrangespec range) { /* Delete both the element and the score. */ zl = ziplistDelete(zl,&eptr); zl = ziplistDelete(zl,&eptr); - deleted++; + num++; } else { /* No longer in range. */ break; } } - return deleted; + 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 long zzlDeleteRangeByRank(robj *zobj, unsigned int start, unsigned int end) { +unsigned char *zzlDeleteRangeByRank(unsigned char *zl, unsigned int start, unsigned int end, unsigned long *deleted) { unsigned int num = (end-start)+1; - zobj->ptr = ziplistDeleteRange(zobj->ptr,2*(start-1),2*num); - return num; + if (deleted) *deleted = num; + zl = ziplistDeleteRange(zl,2*(start-1),2*num); + return zl; } /*----------------------------------------------------------------------------- * Common sorted set API *----------------------------------------------------------------------------*/ -int zsLength(robj *zobj) { +unsigned int zsetLength(robj *zobj) { int length = -1; if (zobj->encoding == REDIS_ENCODING_ZIPLIST) { length = zzlLength(zobj->ptr); - } else if (zobj->encoding == REDIS_ENCODING_RAW) { + } else if (zobj->encoding == REDIS_ENCODING_SKIPLIST) { length = ((zset*)zobj->ptr)->zsl->length; } else { redisPanic("Unknown sorted set encoding"); @@ -727,7 +723,7 @@ int zsLength(robj *zobj) { return length; } -void zsConvert(robj *zobj, int encoding) { +void zsetConvert(robj *zobj, int encoding) { zset *zs; zskiplistNode *node, *next; robj *ele; @@ -741,7 +737,7 @@ void zsConvert(robj *zobj, int encoding) { unsigned int vlen; long long vlong; - if (encoding != REDIS_ENCODING_RAW) + if (encoding != REDIS_ENCODING_SKIPLIST) redisPanic("Unknown target encoding"); zs = zmalloc(sizeof(*zs)); @@ -770,8 +766,8 @@ void zsConvert(robj *zobj, int encoding) { zfree(zobj->ptr); zobj->ptr = zs; - zobj->encoding = REDIS_ENCODING_RAW; - } else if (zobj->encoding == REDIS_ENCODING_RAW) { + zobj->encoding = REDIS_ENCODING_SKIPLIST; + } else if (zobj->encoding == REDIS_ENCODING_SKIPLIST) { unsigned char *zl = ziplistNew(); if (encoding != REDIS_ENCODING_ZIPLIST) @@ -785,13 +781,9 @@ void zsConvert(robj *zobj, int encoding) { zfree(zs->zsl->header); zfree(zs->zsl); - /* Immediately store pointer to ziplist in object because it will - * change because of reallocations when pushing to the ziplist. */ - zobj->ptr = zl; - while (node) { ele = getDecodedObject(node->obj); - redisAssert(zzlInsertAt(zobj,ele,node->score,NULL) == REDIS_OK); + zl = zzlInsertAt(zl,NULL,ele,node->score); decrRefCount(ele); next = node->level[0].forward; @@ -800,6 +792,7 @@ void zsConvert(robj *zobj, int encoding) { } zfree(zs); + zobj->ptr = zl; zobj->encoding = REDIS_ENCODING_ZIPLIST; } else { redisPanic("Unknown sorted set encoding"); @@ -817,11 +810,29 @@ void zaddGenericCommand(redisClient *c, int incr) { robj *ele; robj *zobj; robj *curobj; - double score, curscore = 0.0; + double score = 0, *scores, curscore = 0.0; + int j, elements = (c->argc-2)/2; + int added = 0; - if (getDoubleFromObjectOrReply(c,c->argv[2],&score,NULL) != REDIS_OK) + if (c->argc % 2) { + addReply(c,shared.syntaxerr); return; + } + + /* Start parsing all the scores, we need to emit any syntax error + * before executing additions to the sorted set, as the command should + * either execute fully or nothing at all. */ + scores = zmalloc(sizeof(double)*elements); + for (j = 0; j < elements; j++) { + if (getDoubleFromObjectOrReply(c,c->argv[2+j*2],&scores[j],NULL) + != REDIS_OK) + { + zfree(scores); + return; + } + } + /* Lookup the key and create the sorted set if does not exist. */ zobj = lookupKeyWrite(c->db,key); if (zobj == NULL) { if (server.zset_max_ziplist_entries == 0 || @@ -835,111 +846,105 @@ void zaddGenericCommand(redisClient *c, int incr) { } else { if (zobj->type != REDIS_ZSET) { addReply(c,shared.wrongtypeerr); + zfree(scores); return; } } - if (zobj->encoding == REDIS_ENCODING_ZIPLIST) { - unsigned char *eptr; + for (j = 0; j < elements; j++) { + score = scores[j]; - /* Prefer non-encoded element when dealing with ziplists. */ - ele = c->argv[3]; - if ((eptr = zzlFind(zobj,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; + if (zobj->encoding == REDIS_ENCODING_ZIPLIST) { + unsigned char *eptr; + + /* Prefer non-encoded element when dealing with ziplists. */ + ele = c->argv[3+j*2]; + 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. */ + zfree(scores); + return; + } } - } - /* Remove and re-insert when score changed. */ - if (score != curscore) { - redisAssert(zzlDelete(zobj,eptr) == REDIS_OK); - redisAssert(zzlInsert(zobj,ele,score) == REDIS_OK); + /* Remove and re-insert when score changed. */ + if (score != curscore) { + zobj->ptr = zzlDelete(zobj->ptr,eptr); + zobj->ptr = zzlInsert(zobj->ptr,ele,score); + + signalModifiedKey(c->db,key); + server.dirty++; + } + } else { + /* Optimize: check if the element is too large or the list + * becomes too long *before* executing zzlInsert. */ + zobj->ptr = zzlInsert(zobj->ptr,ele,score); + if (zzlLength(zobj->ptr) > server.zset_max_ziplist_entries) + zsetConvert(zobj,REDIS_ENCODING_SKIPLIST); + if (sdslen(ele->ptr) > server.zset_max_ziplist_value) + zsetConvert(zobj,REDIS_ENCODING_SKIPLIST); signalModifiedKey(c->db,key); server.dirty++; + if (!incr) added++; } + } else if (zobj->encoding == REDIS_ENCODING_SKIPLIST) { + zset *zs = zobj->ptr; + zskiplistNode *znode; + dictEntry *de; - 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. */ - redisAssert(zzlInsert(zobj,ele,score) == REDIS_OK); - if (zzlLength(zobj->ptr) > server.zset_max_ziplist_entries) - zsConvert(zobj,REDIS_ENCODING_RAW); - if (sdslen(ele->ptr) > server.zset_max_ziplist_value) - zsConvert(zobj,REDIS_ENCODING_RAW); - - signalModifiedKey(c->db,key); - server.dirty++; - - if (incr) /* ZINCRBY */ - addReplyDouble(c,score); - else /* ZADD */ - addReply(c,shared.cone); - } - } else if (zobj->encoding == REDIS_ENCODING_RAW) { - 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; + 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); + + 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. */ + zfree(scores); + 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. */ + /* 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++; + } + } 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) added++; } - - 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); + redisPanic("Unknown sorted set encoding"); } - } else { - redisPanic("Unknown sorted set encoding"); } + zfree(scores); + if (incr) /* ZINCRBY */ + addReplyDouble(c,score); + else /* ZADD */ + addReplyLongLong(c,added); } void zaddCommand(redisClient *c) { @@ -952,8 +957,8 @@ void zincrbyCommand(redisClient *c) { void zremCommand(redisClient *c) { robj *key = c->argv[1]; - robj *ele = c->argv[2]; robj *zobj; + int deleted = 0, j; if ((zobj = lookupKeyWriteOrReply(c,key,shared.czero)) == NULL || checkType(c,zobj,REDIS_ZSET)) return; @@ -961,39 +966,48 @@ void zremCommand(redisClient *c) { if (zobj->encoding == REDIS_ENCODING_ZIPLIST) { unsigned char *eptr; - if ((eptr = zzlFind(zobj,ele,NULL)) != NULL) { - redisAssert(zzlDelete(zobj,eptr) == REDIS_OK); - if (zzlLength(zobj->ptr) == 0) dbDelete(c->db,key); - } else { - addReply(c,shared.czero); - return; + for (j = 2; j < c->argc; j++) { + if ((eptr = zzlFind(zobj->ptr,c->argv[j],NULL)) != NULL) { + deleted++; + zobj->ptr = zzlDelete(zobj->ptr,eptr); + if (zzlLength(zobj->ptr) == 0) { + dbDelete(c->db,key); + break; + } + } } - } else if (zobj->encoding == REDIS_ENCODING_RAW) { + } 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; + for (j = 2; j < c->argc; j++) { + de = dictFind(zs->dict,c->argv[j]); + if (de != NULL) { + deleted++; + + /* Delete from the skiplist */ + score = *(double*)dictGetEntryVal(de); + redisAssert(zslDelete(zs->zsl,score,c->argv[j])); + + /* Delete from the hash table */ + dictDelete(zs->dict,c->argv[j]); + if (htNeedsResize(zs->dict)) dictResize(zs->dict); + if (dictSize(zs->dict) == 0) { + dbDelete(c->db,key); + break; + } + } } } else { redisPanic("Unknown sorted set encoding"); } - signalModifiedKey(c->db,key); - server.dirty++; - addReply(c,shared.cone); + if (deleted) { + signalModifiedKey(c->db,key); + server.dirty += deleted; + } + addReplyLongLong(c,deleted); } void zremrangebyscoreCommand(redisClient *c) { @@ -1012,8 +1026,9 @@ void zremrangebyscoreCommand(redisClient *c) { checkType(c,zobj,REDIS_ZSET)) return; if (zobj->encoding == REDIS_ENCODING_ZIPLIST) { - deleted = zzlDeleteRangeByScore(zobj,range); - } else if (zobj->encoding == REDIS_ENCODING_RAW) { + 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); @@ -1042,7 +1057,7 @@ void zremrangebyrankCommand(redisClient *c) { checkType(c,zobj,REDIS_ZSET)) return; /* Sanitize indexes. */ - llen = zsLength(zobj); + llen = zsetLength(zobj); if (start < 0) start = llen+start; if (end < 0) end = llen+end; if (start < 0) start = 0; @@ -1057,8 +1072,9 @@ void zremrangebyrankCommand(redisClient *c) { if (zobj->encoding == REDIS_ENCODING_ZIPLIST) { /* Correct for 1-based rank. */ - deleted = zzlDeleteRangeByRank(zobj,start+1,end+1); - } else if (zobj->encoding == REDIS_ENCODING_RAW) { + 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. */ @@ -1158,7 +1174,7 @@ void zuiInitIterator(zsetopsrc *op) { it->zl.sptr = ziplistNext(it->zl.zl,it->zl.eptr); redisAssert(it->zl.sptr != NULL); } - } else if (op->encoding == REDIS_ENCODING_RAW) { + } 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 { @@ -1186,7 +1202,7 @@ void zuiClearIterator(zsetopsrc *op) { iterzset *it = &op->iter.zset; if (op->encoding == REDIS_ENCODING_ZIPLIST) { REDIS_NOTUSED(it); /* skip */ - } else if (op->encoding == REDIS_ENCODING_RAW) { + } else if (op->encoding == REDIS_ENCODING_SKIPLIST) { REDIS_NOTUSED(it); /* skip */ } else { redisPanic("Unknown sorted set encoding"); @@ -1213,7 +1229,7 @@ int zuiLength(zsetopsrc *op) { iterzset *it = &op->iter.zset; if (op->encoding == REDIS_ENCODING_ZIPLIST) { return zzlLength(it->zl.zl); - } else if (op->encoding == REDIS_ENCODING_RAW) { + } else if (op->encoding == REDIS_ENCODING_SKIPLIST) { return it->sl.zs->zsl->length; } else { redisPanic("Unknown sorted set encoding"); @@ -1238,7 +1254,7 @@ int zuiNext(zsetopsrc *op, zsetopval *val) { 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)) + if (!intsetGet(it->is.is,it->is.ii,(int64_t*)&val->ell)) return 0; val->score = 1.0; @@ -1266,7 +1282,7 @@ int zuiNext(zsetopsrc *op, zsetopval *val) { /* Move to next element. */ zzlNext(it->zl.zl,&it->zl.eptr,&it->zl.sptr); - } else if (op->encoding == REDIS_ENCODING_RAW) { + } else if (op->encoding == REDIS_ENCODING_SKIPLIST) { if (it->sl.node == NULL) return 0; val->ele = it->sl.node->obj; @@ -1372,13 +1388,13 @@ int zuiFind(zsetopsrc *op, zsetopval *val, double *score) { zuiObjectFromValue(val); if (op->encoding == REDIS_ENCODING_ZIPLIST) { - if (zzlFind(op->subject,val->ele,score) != NULL) { + 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_RAW) { + } else if (op->encoding == REDIS_ENCODING_SKIPLIST) { dictEntry *de; if ((de = dictFind(it->sl.zs->dict,val->ele)) != NULL) { *score = *(double*)dictGetEntryVal(de); @@ -1514,6 +1530,7 @@ void zunionInterGenericCommand(redisClient *c, robj *dstkey, int op) { dstobj = createZsetObject(); dstzset = dstobj->ptr; + memset(&zval, 0, sizeof(zval)); if (op == REDIS_OP_INTER) { /* Skip everything if the smallest input is empty. */ @@ -1525,7 +1542,12 @@ void zunionInterGenericCommand(redisClient *c, robj *dstkey, int op) { score = src[0].weight * zval.score; for (j = 1; j < setnum; j++) { - if (zuiFind(&src[j],&zval,&value)) { + /* It is not safe to access the zset we are + * iterating, so explicitly check for equal object. */ + if (src[j].subject == src[0].subject) { + value = zval.score*src[j].weight; + zunionInterAggregate(&score,value,aggregate); + } else if (zuiFind(&src[j],&zval,&value)) { value *= src[j].weight; zunionInterAggregate(&score,value,aggregate); } else { @@ -1549,7 +1571,7 @@ void zunionInterGenericCommand(redisClient *c, robj *dstkey, int op) { } } else if (op == REDIS_OP_UNION) { for (i = 0; i < setnum; i++) { - if (zuiLength(&src[0]) == 0) + if (zuiLength(&src[i]) == 0) continue; while (zuiNext(&src[i],&zval)) { @@ -1565,7 +1587,12 @@ void zunionInterGenericCommand(redisClient *c, robj *dstkey, int op) { /* 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++) { - if (zuiFind(&src[j],&zval,&value)) { + /* It is not safe to access the zset we are + * iterating, so explicitly check for equal object. */ + if(src[j].subject == src[i].subject) { + value = zval.score*src[j].weight; + zunionInterAggregate(&score,value,aggregate); + } else if (zuiFind(&src[j],&zval,&value)) { value *= src[j].weight; zunionInterAggregate(&score,value,aggregate); } @@ -1598,10 +1625,10 @@ void zunionInterGenericCommand(redisClient *c, robj *dstkey, int op) { /* Convert to ziplist when in limits. */ if (dstzset->zsl->length <= server.zset_max_ziplist_entries && maxelelen <= server.zset_max_ziplist_value) - zsConvert(dstobj,REDIS_ENCODING_ZIPLIST); + zsetConvert(dstobj,REDIS_ENCODING_ZIPLIST); dbAdd(c->db,dstkey,dstobj); - addReplyLongLong(c,zsLength(dstobj)); + addReplyLongLong(c,zsetLength(dstobj)); if (!touched) signalModifiedKey(c->db,dstkey); server.dirty++; } else { @@ -1642,7 +1669,7 @@ void zrangeGenericCommand(redisClient *c, int reverse) { || checkType(c,zobj,REDIS_ZSET)) return; /* Sanitize indexes. */ - llen = zsLength(zobj); + llen = zsetLength(zobj); if (start < 0) start = llen+start; if (end < 0) end = llen+end; if (start < 0) start = 0; @@ -1691,7 +1718,7 @@ void zrangeGenericCommand(redisClient *c, int reverse) { zzlNext(zl,&eptr,&sptr); } - } else if (zobj->encoding == REDIS_ENCODING_RAW) { + } else if (zobj->encoding == REDIS_ENCODING_SKIPLIST) { zset *zs = zobj->ptr; zskiplist *zsl = zs->zsl; zskiplistNode *ln; @@ -1791,9 +1818,9 @@ void genericZrangebyscoreCommand(redisClient *c, int reverse, int justcount) { /* If reversed, get the last node in range as starting point. */ if (reverse) - eptr = zzlLastInRange(zobj,range); + eptr = zzlLastInRange(zl,range); else - eptr = zzlFirstInRange(zobj,range); + eptr = zzlFirstInRange(zl,range); /* No "first" element in the specified interval. */ if (eptr == NULL) { @@ -1848,7 +1875,7 @@ void genericZrangebyscoreCommand(redisClient *c, int reverse, int justcount) { else zzlNext(zl,&eptr,&sptr); } - } else if (zobj->encoding == REDIS_ENCODING_RAW) { + } else if (zobj->encoding == REDIS_ENCODING_SKIPLIST) { zset *zs = zobj->ptr; zskiplist *zsl = zs->zsl; zskiplistNode *ln; @@ -1916,7 +1943,80 @@ void zrevrangebyscoreCommand(redisClient *c) { } 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 double"); + 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); + redisAssert(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) { @@ -1926,7 +2026,7 @@ void zcardCommand(redisClient *c) { if ((zobj = lookupKeyReadOrReply(c,key,shared.czero)) == NULL || checkType(c,zobj,REDIS_ZSET)) return; - addReplyLongLong(c,zsLength(zobj)); + addReplyLongLong(c,zsetLength(zobj)); } void zscoreCommand(redisClient *c) { @@ -1938,11 +2038,11 @@ void zscoreCommand(redisClient *c) { checkType(c,zobj,REDIS_ZSET)) return; if (zobj->encoding == REDIS_ENCODING_ZIPLIST) { - if (zzlFind(zobj,c->argv[2],&score) != NULL) + if (zzlFind(zobj->ptr,c->argv[2],&score) != NULL) addReplyDouble(c,score); else addReply(c,shared.nullbulk); - } else if (zobj->encoding == REDIS_ENCODING_RAW) { + } else if (zobj->encoding == REDIS_ENCODING_SKIPLIST) { zset *zs = zobj->ptr; dictEntry *de; @@ -1968,7 +2068,7 @@ void zrankGenericCommand(redisClient *c, int reverse) { if ((zobj = lookupKeyReadOrReply(c,key,shared.nullbulk)) == NULL || checkType(c,zobj,REDIS_ZSET)) return; - llen = zsLength(zobj); + llen = zsetLength(zobj); redisAssert(ele->encoding == REDIS_ENCODING_RAW); if (zobj->encoding == REDIS_ENCODING_ZIPLIST) { @@ -1996,7 +2096,7 @@ void zrankGenericCommand(redisClient *c, int reverse) { } else { addReply(c,shared.nullbulk); } - } else if (zobj->encoding == REDIS_ENCODING_RAW) { + } else if (zobj->encoding == REDIS_ENCODING_SKIPLIST) { zset *zs = zobj->ptr; zskiplist *zsl = zs->zsl; dictEntry *de;