X-Git-Url: https://git.saurik.com/redis.git/blobdiff_plain/22b9bf15949933b351525ae658dc32e40e5784ab..3ea204e1031a94dafca7f7e4eed2f79ec3bd7fd0:/src/t_zset.c diff --git a/src/t_zset.c b/src/t_zset.c index 01b1d035..25da7871 100644 --- a/src/t_zset.c +++ b/src/t_zset.c @@ -180,27 +180,31 @@ typedef struct { int minex, maxex; /* are min or max exclusive? */ } zrangespec; -static inline int zslValueInMinRange(double value, zrangespec *spec) { +static int zslValueGteMin(double value, zrangespec *spec) { return spec->minex ? (value > spec->min) : (value >= spec->min); } -static inline int zslValueInMaxRange(double value, zrangespec *spec) { +static int zslValueLteMax(double value, zrangespec *spec) { return spec->maxex ? (value < spec->max) : (value <= spec->max); } -static inline int zslValueInRange(double value, zrangespec *spec) { - return zslValueInMinRange(value,spec) && zslValueInMaxRange(value,spec); +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; + /* 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 || !zslValueInMinRange(x->score,range)) + if (x == NULL || !zslValueGteMin(x->score,range)) return 0; x = zsl->header->level[0].forward; - if (x == NULL || !zslValueInMaxRange(x->score,range)) + if (x == NULL || !zslValueLteMax(x->score,range)) return 0; return 1; } @@ -218,7 +222,7 @@ zskiplistNode *zslFirstInRange(zskiplist *zsl, zrangespec range) { for (i = zsl->level-1; i >= 0; i--) { /* Go forward while *OUT* of range. */ while (x->level[i].forward && - !zslValueInMinRange(x->level[i].forward->score,&range)) + !zslValueGteMin(x->level[i].forward->score,&range)) x = x->level[i].forward; } @@ -242,7 +246,7 @@ zskiplistNode *zslLastInRange(zskiplist *zsl, zrangespec range) { for (i = zsl->level-1; i >= 0; i--) { /* Go forward while *IN* range. */ while (x->level[i].forward && - zslValueInMaxRange(x->level[i].forward->score,&range)) + zslValueLteMax(x->level[i].forward->score,&range)) x = x->level[i].forward; } @@ -399,150 +403,514 @@ 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(robj *zobj) { + unsigned char *zl = zobj->ptr; + 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(robj *zobj, zrangespec range) { + unsigned char *zl = zobj->ptr; + 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)) + return eptr; + + /* 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(robj *zobj, zrangespec range) { + unsigned char *zl = zobj->ptr; + 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)) + return eptr; + + /* 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(robj *zobj, robj *ele, double *score) { + unsigned char *zl = zobj->ptr; + 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. */ +int zzlDelete(robj *zobj, unsigned char *eptr) { + unsigned char *zl = zobj->ptr; + 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; +} + +int zzlInsertAt(robj *zobj, robj *ele, double score, unsigned char *eptr) { + unsigned char *zl = zobj->ptr; + unsigned char *sptr; + char scorebuf[128]; + int scorelen; + int 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); + } + + zobj->ptr = zl; + return REDIS_OK; +} + +/* 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 *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. */ + zzlInsertAt(zobj,ele,score,eptr); + break; + } else if (s == score) { + /* Ensure lexicographical ordering for elements. */ + if (zzlCompareElements(eptr,ele->ptr,sdslen(ele->ptr)) > 0) { + zzlInsertAt(zobj,ele,score,eptr); + break; + } + } + + /* Move to next element. */ + eptr = ziplistNext(zl,sptr); + } + + /* Push on tail of list when it was not yet inserted. */ + if (eptr == NULL) + zzlInsertAt(zobj,ele,score,NULL); + + decrRefCount(ele); + return REDIS_OK; +} + +unsigned long zzlDeleteRangeByScore(robj *zobj, zrangespec range) { + unsigned char *zl = zobj->ptr; + unsigned char *eptr, *sptr; + double score; + unsigned long deleted = 0; + + eptr = zzlFirstInRange(zobj,range); + if (eptr == NULL) return deleted; + + + /* 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); + deleted++; + } else { + /* No longer in range. */ + break; + } + } + + return deleted; +} + +/* 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 int num = (end-start)+1; + zobj->ptr = ziplistDeleteRange(zobj->ptr,2*(start-1),2*num); + return num; +} + +/*----------------------------------------------------------------------------- + * Common sorted set API + *----------------------------------------------------------------------------*/ + +int zsLength(robj *zobj) { + int length = -1; + if (zobj->encoding == REDIS_ENCODING_ZIPLIST) { + length = zzlLength(zobj); + } else if (zobj->encoding == REDIS_ENCODING_RAW) { + length = ((zset*)zobj->ptr)->zsl->length; + } else { + redisPanic("Unknown sorted set encoding"); + } + return length; +} /*----------------------------------------------------------------------------- * 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; + + if (getDoubleFromObjectOrReply(c,c->argv[2],&score,NULL) != REDIS_OK) + return; - zsetobj = lookupKeyWrite(c->db,key); - if (zsetobj == NULL) { - zsetobj = createZsetObject(); - dbAdd(c->db,key,zsetobj); + zobj = lookupKeyWrite(c->db,key); + if (zobj == NULL) { + zobj = createZsetZiplistObject(); + dbAdd(c->db,key,zobj); } else { - if (zsetobj->type != REDIS_ZSET) { + if (zobj->type != REDIS_ZSET) { addReply(c,shared.wrongtypeerr); return; } } - zs = zsetobj->ptr; - - /* Since both ZADD and ZINCRBY are implemented here, we need to increment - * the score first by the current score if ZINCRBY is called. */ - if (incr) { - /* Read the old score. If the element was not present starts from 0 */ - dictEntry *de = dictFind(zs->dict,ele); - if (de != NULL) - score += *(double*)dictGetEntryVal(de); - - if (isnan(score)) { - addReplyError(c,"resulting score is not a number (NaN)"); - /* Note that we don't need to check if the zset may be empty and - * should be removed here, as we can only obtain Nan as score if - * there was already an element in the sorted set. */ - return; - } - } - /* We need to remove and re-insert the element when it was already present - * in the dictionary, to update the skiplist. Note that we delay adding a - * pointer to the score because we want to reference the score in the - * skiplist node. */ - if (dictAdd(zs->dict,ele,NULL) == DICT_OK) { - dictEntry *de; + if (zobj->encoding == REDIS_ENCODING_ZIPLIST) { + unsigned char *eptr; + + /* Prefer non-encoded element when dealing with ziplists. */ + ele = c->argv[3]; + if ((eptr = zzlFind(zobj,ele,&curscore)) != NULL) { + if (incr) { + score += curscore; + if (isnan(score)) { + addReplyError(c,nanerr); + /* Don't need to check if the sorted set is empty, because + * we know it has at least one element. */ + return; + } + } - /* New element */ - incrRefCount(ele); /* added to hash */ - znode = zslInsert(zs->zsl,score,ele); - incrRefCount(ele); /* added to skiplist */ + /* Remove and re-insert when score changed. */ + if (score != curscore) { + redisAssert(zzlDelete(zobj,eptr) == REDIS_OK); + redisAssert(zzlInsert(zobj,ele,score) == REDIS_OK); - /* Update the score in the dict entry */ - de = dictFind(zs->dict,ele); - redisAssert(de != NULL); - dictGetEntryVal(de) = &znode->score; - signalModifiedKey(c->db,c->argv[1]); - server.dirty++; - if (incr) - addReplyDouble(c,score); - else - addReply(c,shared.cone); - } else { + signalModifiedKey(c->db,key); + server.dirty++; + } + + if (incr) /* ZINCRBY */ + addReplyDouble(c,score); + else /* ZADD */ + addReply(c,shared.czero); + } else { + redisAssert(zzlInsert(zobj,ele,score) == REDIS_OK); + + 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; - robj *curobj; - double *curscore; - int deleted; - /* Update score */ + ele = c->argv[3] = tryObjectEncoding(c->argv[3]); de = dictFind(zs->dict,ele); - redisAssert(de != NULL); - curobj = dictGetEntryKey(de); - curscore = dictGetEntryVal(de); - - /* When the score is updated, reuse the existing string object to - * prevent extra alloc/dealloc of strings on ZINCRBY. */ - if (score != *curscore) { - deleted = zslDelete(zs->zsl,*curscore,curobj); - redisAssert(deleted != 0); - znode = zslInsert(zs->zsl,score,curobj); - incrRefCount(curobj); - - /* Update the score in the current dict entry */ - dictGetEntryVal(de) = &znode->score; - signalModifiedKey(c->db,c->argv[1]); + if (de != NULL) { + curobj = dictGetEntryKey(de); + curscore = *(double*)dictGetEntryVal(de); + + if (incr) { + score += curscore; + if (isnan(score)) { + addReplyError(c,nanerr); + /* Don't need to check if the sorted set is empty, because + * we know it has at least one element. */ + return; + } + } + + /* Remove and re-insert when score changed. We can safely delete + * the key object from the skiplist, since the dictionary still has + * a reference to it. */ + if (score != curscore) { + redisAssert(zslDelete(zs->zsl,curscore,curobj)); + znode = zslInsert(zs->zsl,score,curobj); + incrRefCount(curobj); /* Re-inserted in skiplist. */ + dictGetEntryVal(de) = &znode->score; /* Update score ptr. */ + + signalModifiedKey(c->db,key); + server.dirty++; + } + + if (incr) /* ZINCRBY */ + addReplyDouble(c,score); + else /* ZADD */ + addReply(c,shared.czero); + } else { + znode = zslInsert(zs->zsl,score,ele); + incrRefCount(ele); /* Inserted in skiplist. */ + redisAssert(dictAdd(zs->dict,ele,&znode->score) == DICT_OK); + incrRefCount(ele); /* Added to dictionary. */ + + signalModifiedKey(c->db,key); server.dirty++; + + if (incr) /* ZINCRBY */ + addReplyDouble(c,score); + else /* ZADD */ + addReply(c,shared.cone); } - if (incr) - addReplyDouble(c,score); - else - addReply(c,shared.czero); + } else { + redisPanic("Unknown sorted set encoding"); } } void zaddCommand(redisClient *c) { - double scoreval; - if (getDoubleFromObjectOrReply(c,c->argv[2],&scoreval,NULL) != REDIS_OK) return; - c->argv[3] = tryObjectEncoding(c->argv[3]); - zaddGenericCommand(c,c->argv[1],c->argv[3],scoreval,0); + zaddGenericCommand(c,0); } void zincrbyCommand(redisClient *c) { - double scoreval; - if (getDoubleFromObjectOrReply(c,c->argv[2],&scoreval,NULL) != REDIS_OK) return; - c->argv[3] = tryObjectEncoding(c->argv[3]); - zaddGenericCommand(c,c->argv[1],c->argv[3],scoreval,1); + zaddGenericCommand(c,1); } void zremCommand(redisClient *c) { - robj *zsetobj; - zset *zs; - dictEntry *de; - double curscore; - int deleted; + robj *key = c->argv[1]; + robj *ele = c->argv[2]; + robj *zobj; - if ((zsetobj = lookupKeyWriteOrReply(c,c->argv[1],shared.czero)) == NULL || - checkType(c,zsetobj,REDIS_ZSET)) return; + if ((zobj = lookupKeyWriteOrReply(c,key,shared.czero)) == NULL || + checkType(c,zobj,REDIS_ZSET)) return; - zs = zsetobj->ptr; - c->argv[2] = tryObjectEncoding(c->argv[2]); - de = dictFind(zs->dict,c->argv[2]); - if (de == NULL) { - addReply(c,shared.czero); - return; + if (zobj->encoding == REDIS_ENCODING_ZIPLIST) { + unsigned char *eptr; + + if ((eptr = zzlFind(zobj,ele,NULL)) != NULL) { + redisAssert(zzlDelete(zobj,eptr) == REDIS_OK); + if (zzlLength(zobj) == 0) dbDelete(c->db,key); + } else { + addReply(c,shared.czero); + return; + } + } else if (zobj->encoding == REDIS_ENCODING_RAW) { + 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) { @@ -550,35 +918,41 @@ void zremrangebyscoreCommand(redisClient *c) { return; } - if ((o = lookupKeyWriteOrReply(c,c->argv[1],shared.czero)) == NULL || - checkType(c,o,REDIS_ZSET)) return; + if ((zobj = lookupKeyWriteOrReply(c,key,shared.czero)) == NULL || + checkType(c,zobj,REDIS_ZSET)) return; + + if (zobj->encoding == REDIS_ENCODING_ZIPLIST) { + deleted = zzlDeleteRangeByScore(zobj,range); + } else if (zobj->encoding == REDIS_ENCODING_RAW) { + zset *zs = zobj->ptr; + deleted = zslDeleteRangeByScore(zs->zsl,range,zs->dict); + if (htNeedsResize(zs->dict)) dictResize(zs->dict); + if (dictSize(zs->dict) == 0) dbDelete(c->db,key); + } else { + redisPanic("Unknown sorted set encoding"); + } - zs = o->ptr; - deleted = zslDeleteRangeByScore(zs->zsl,range,zs->dict); - if (htNeedsResize(zs->dict)) dictResize(zs->dict); - if (dictSize(zs->dict) == 0) dbDelete(c->db,c->argv[1]); - if (deleted) signalModifiedKey(c->db,c->argv[1]); + if (deleted) signalModifiedKey(c->db,key); server.dirty += deleted; addReplyLongLong(c,deleted); } void zremrangebyrankCommand(redisClient *c) { + robj *key = c->argv[1]; + robj *zobj; long start; long end; int llen; - long deleted; - robj *zsetobj; - zset *zs; + unsigned long deleted; if ((getLongFromObjectOrReply(c, c->argv[2], &start, NULL) != REDIS_OK) || (getLongFromObjectOrReply(c, c->argv[3], &end, NULL) != REDIS_OK)) return; - if ((zsetobj = lookupKeyWriteOrReply(c,c->argv[1],shared.czero)) == NULL || - checkType(c,zsetobj,REDIS_ZSET)) return; - zs = zsetobj->ptr; - llen = zs->zsl->length; + if ((zobj = lookupKeyWriteOrReply(c,key,shared.czero)) == NULL || + checkType(c,zobj,REDIS_ZSET)) return; - /* convert negative indexes */ + /* Sanitize indexes. */ + llen = zsLength(zobj); if (start < 0) start = llen+start; if (end < 0) end = llen+end; if (start < 0) start = 0; @@ -591,14 +965,23 @@ 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. */ + deleted = zzlDeleteRangeByRank(zobj,start+1,end+1); + } else if (zobj->encoding == REDIS_ENCODING_RAW) { + 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 { @@ -822,16 +1205,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; @@ -843,13 +1223,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 = zsLength(zobj); if (start < 0) start = llen+start; if (end < 0) end = llen+end; if (start < 0) start = 0; @@ -863,23 +1241,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 : zslGetElementByRank(zsl, llen-start); - } else { - ln = start == 0 ? - zsl->header->level[0].forward : zslGetElementByRank(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_RAW) { + 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"); } } @@ -895,10 +1318,8 @@ 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; @@ -942,61 +1363,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 reversed, get the last node in range as starting point. */ - if (reverse) { - ln = zslLastInRange(zsl,range); - } else { - ln = zslFirstInRange(zsl,range); - } + if ((zobj = lookupKeyReadOrReply(c,key,emptyreply)) == NULL || + checkType(c,zobj,REDIS_ZSET)) return; + + if (zobj->encoding == REDIS_ENCODING_ZIPLIST) { + unsigned char *zl = zobj->ptr; + unsigned char *eptr, *sptr; + unsigned char *vstr; + unsigned int vlen; + long long vlong; + double score; + + /* If reversed, get the last node in range as starting point. */ + if (reverse) + eptr = zzlLastInRange(zobj,range); + else + eptr = zzlFirstInRange(zobj,range); - /* No "first" element in the specified interval. */ - if (ln == NULL) { - addReply(c,emptyreply); - return; - } + /* No "first" element in the specified interval. */ + if (eptr == 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); + /* 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; + } - /* 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; - } + /* 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); + } - while (ln && limit--) { - /* Abort when the node is no longer in range. */ - if (reverse) { - if (!zslValueInMinRange(ln->score,&range)) break; - } else { - if (!zslValueInMaxRange(ln->score,&range)) break; + /* Move to next node */ + if (reverse) + zzlPrev(zl,&eptr,&sptr); + else + zzlNext(zl,&eptr,&sptr); } + } else if (zobj->encoding == REDIS_ENCODING_RAW) { + 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); - /* Do our magic */ - rangelen++; - if (!justcount) { - addReplyBulk(c,ln->obj); - if (withscores) - addReplyDouble(c,ln->score); + /* No "first" element in the specified interval. */ + if (ln == NULL) { + addReply(c,emptyreply); + return; } - /* Move to next node */ - ln = reverse ? ln->backward : ln->level[0].forward; + /* 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 { + 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); } } @@ -1013,66 +1505,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,zzlLength(zobj)); } void zscoreCommand(redisClient *c) { - robj *o; - zset *zs; - dictEntry *de; + robj *key = c->argv[1]; + robj *zobj; + double score; - if ((o = lookupKeyReadOrReply(c,c->argv[1],shared.nullbulk)) == NULL || - checkType(c,o,REDIS_ZSET)) return; + if ((zobj = lookupKeyReadOrReply(c,key,shared.nullbulk)) == NULL || + checkType(c,zobj,REDIS_ZSET)) return; - zs = o->ptr; - c->argv[2] = tryObjectEncoding(c->argv[2]); - de = dictFind(zs->dict,c->argv[2]); - if (!de) { - addReply(c,shared.nullbulk); - } else { - double *score = dictGetEntryVal(de); + if (zobj->encoding == REDIS_ENCODING_ZIPLIST) { + if (zzlFind(zobj,c->argv[2],&score) != NULL) + addReplyDouble(c,score); + else + addReply(c,shared.nullbulk); + } else if (zobj->encoding == REDIS_ENCODING_RAW) { + zset *zs = zobj->ptr; + dictEntry *de; - addReplyDouble(c,*score); + c->argv[2] = tryObjectEncoding(c->argv[2]); + de = dictFind(zs->dict,c->argv[2]); + if (de != NULL) { + score = *(double*)dictGetEntryVal(de); + addReplyDouble(c,score); + } else { + addReply(c,shared.nullbulk); + } + } else { + redisPanic("Unknown sorted set encoding"); } } void zrankGenericCommand(redisClient *c, int reverse) { - robj *o; - zset *zs; - zskiplist *zsl; - dictEntry *de; + robj *key = c->argv[1]; + robj *ele = c->argv[2]; + robj *zobj; + unsigned long llen; unsigned long rank; - double *score; - if ((o = lookupKeyReadOrReply(c,c->argv[1],shared.nullbulk)) == NULL || - checkType(c,o,REDIS_ZSET)) return; + if ((zobj = lookupKeyReadOrReply(c,key,shared.nullbulk)) == NULL || + checkType(c,zobj,REDIS_ZSET)) return; + llen = zsLength(zobj); + + redisAssert(ele->encoding == REDIS_ENCODING_RAW); + if (zobj->encoding == REDIS_ENCODING_ZIPLIST) { + unsigned char *zl = zobj->ptr; + unsigned char *eptr, *sptr; + + eptr = ziplistIndex(zl,0); + redisAssert(eptr != NULL); + sptr = ziplistNext(zl,eptr); + redisAssert(sptr != NULL); + + rank = 1; + while(eptr != NULL) { + if (ziplistCompare(eptr,ele->ptr,sdslen(ele->ptr))) + break; + rank++; + zzlNext(zl,&eptr,&sptr); + } - zs = o->ptr; - zsl = zs->zsl; - c->argv[2] = tryObjectEncoding(c->argv[2]); - de = dictFind(zs->dict,c->argv[2]); - if (!de) { - addReply(c,shared.nullbulk); - return; - } + if (eptr != NULL) { + if (reverse) + addReplyLongLong(c,llen-rank); + else + addReplyLongLong(c,rank-1); + } else { + addReply(c,shared.nullbulk); + } + } else if (zobj->encoding == REDIS_ENCODING_RAW) { + zset *zs = zobj->ptr; + zskiplist *zsl = zs->zsl; + dictEntry *de; + double score; - score = dictGetEntryVal(de); - rank = zslGetRank(zsl, *score, c->argv[2]); - if (rank) { - if (reverse) { - addReplyLongLong(c, zsl->length - rank); + ele = c->argv[2] = tryObjectEncoding(c->argv[2]); + de = dictFind(zs->dict,ele); + if (de != NULL) { + score = *(double*)dictGetEntryVal(de); + rank = zslGetRank(zsl,score,ele); + redisAssert(rank); /* Existing elements always have a rank. */ + if (reverse) + addReplyLongLong(c,llen-rank); + else + addReplyLongLong(c,rank-1); } else { - addReplyLongLong(c, rank-1); + addReply(c,shared.nullbulk); } } else { - addReply(c,shared.nullbulk); + redisPanic("Unknown sorted set encoding"); } }