redict/src/t_zset.c
Oran Agra 9ca5e8c547
fix zslGetRank bug in dead-code (#9246)
This fixes an issue with zslGetRank which will happen only if the
skiplist data stracture is added two entries with the same element name,
this can't happen in redis zsets (we use dict), but in theory this is a
bug in the underlaying skiplist code.

Fixes #3081 and #4032

Co-authored-by: minjian.cai <cmjgithub@163.com>
2021-07-22 13:40:00 +03:00

4218 lines
139 KiB
C

/*
* Copyright (c) 2009-2012, Salvatore Sanfilippo <antirez at gmail dot com>
* Copyright (c) 2009-2012, Pieter Noordhuis <pcnoordhuis at gmail dot com>
* All rights reserved.
*
* Redistribution and use in source and binary forms, with or without
* modification, are permitted provided that the following conditions are met:
*
* * Redistributions of source code must retain the above copyright notice,
* this list of conditions and the following disclaimer.
* * Redistributions in binary form must reproduce the above copyright
* notice, this list of conditions and the following disclaimer in the
* documentation and/or other materials provided with the distribution.
* * Neither the name of Redis nor the names of its contributors may be used
* to endorse or promote products derived from this software without
* specific prior written permission.
*
* THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
* AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
* IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
* ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE
* LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
* CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
* SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
* INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
* CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
* ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
* POSSIBILITY OF SUCH DAMAGE.
*/
/*-----------------------------------------------------------------------------
* Sorted set API
*----------------------------------------------------------------------------*/
/* ZSETs are ordered sets using two data structures to hold the same elements
* in order to get O(log(N)) INSERT and REMOVE operations into a sorted
* data structure.
*
* The elements are added to a hash table mapping Redis objects to scores.
* At the same time the elements are added to a skip list mapping scores
* to Redis objects (so objects are sorted by scores in this "view").
*
* Note that the SDS string representing the element is the same in both
* the hash table and skiplist in order to save memory. What we do in order
* to manage the shared SDS string more easily is to free the SDS string
* only in zslFreeNode(). The dictionary has no value free method set.
* So we should always remove an element from the dictionary, and later from
* the skiplist.
*
* 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 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
* from tail to head, useful for ZREVRANGE. */
#include "server.h"
#include <math.h>
/*-----------------------------------------------------------------------------
* Skiplist implementation of the low level API
*----------------------------------------------------------------------------*/
int zslLexValueGteMin(sds value, zlexrangespec *spec);
int zslLexValueLteMax(sds value, zlexrangespec *spec);
/* Create a skiplist node with the specified number of levels.
* The SDS string 'ele' is referenced by the node after the call. */
zskiplistNode *zslCreateNode(int level, double score, sds ele) {
zskiplistNode *zn =
zmalloc(sizeof(*zn)+level*sizeof(struct zskiplistLevel));
zn->score = score;
zn->ele = ele;
return zn;
}
/* Create a new skiplist. */
zskiplist *zslCreate(void) {
int j;
zskiplist *zsl;
zsl = zmalloc(sizeof(*zsl));
zsl->level = 1;
zsl->length = 0;
zsl->header = zslCreateNode(ZSKIPLIST_MAXLEVEL,0,NULL);
for (j = 0; j < ZSKIPLIST_MAXLEVEL; j++) {
zsl->header->level[j].forward = NULL;
zsl->header->level[j].span = 0;
}
zsl->header->backward = NULL;
zsl->tail = NULL;
return zsl;
}
/* Free the specified skiplist node. The referenced SDS string representation
* of the element is freed too, unless node->ele is set to NULL before calling
* this function. */
void zslFreeNode(zskiplistNode *node) {
sdsfree(node->ele);
zfree(node);
}
/* Free a whole skiplist. */
void zslFree(zskiplist *zsl) {
zskiplistNode *node = zsl->header->level[0].forward, *next;
zfree(zsl->header);
while(node) {
next = node->level[0].forward;
zslFreeNode(node);
node = next;
}
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))
level += 1;
return (level<ZSKIPLIST_MAXLEVEL) ? level : ZSKIPLIST_MAXLEVEL;
}
/* Insert a new node in the skiplist. Assumes the element does not already
* exist (up to the caller to enforce that). The skiplist takes ownership
* of the passed SDS string 'ele'. */
zskiplistNode *zslInsert(zskiplist *zsl, double score, sds ele) {
zskiplistNode *update[ZSKIPLIST_MAXLEVEL], *x;
unsigned long rank[ZSKIPLIST_MAXLEVEL];
int i, level;
serverAssert(!isnan(score));
x = zsl->header;
for (i = zsl->level-1; i >= 0; i--) {
/* store rank that is crossed to reach the insert position */
rank[i] = i == (zsl->level-1) ? 0 : rank[i+1];
while (x->level[i].forward &&
(x->level[i].forward->score < score ||
(x->level[i].forward->score == score &&
sdscmp(x->level[i].forward->ele,ele) < 0)))
{
rank[i] += x->level[i].span;
x = x->level[i].forward;
}
update[i] = x;
}
/* we assume the element is not already inside, since we allow duplicated
* scores, reinserting the same element should never happen since the
* caller of zslInsert() should test in the hash table if the element is
* already inside or not. */
level = zslRandomLevel();
if (level > zsl->level) {
for (i = zsl->level; i < level; i++) {
rank[i] = 0;
update[i] = zsl->header;
update[i]->level[i].span = zsl->length;
}
zsl->level = level;
}
x = zslCreateNode(level,score,ele);
for (i = 0; i < level; i++) {
x->level[i].forward = update[i]->level[i].forward;
update[i]->level[i].forward = x;
/* update span covered by update[i] as x is inserted here */
x->level[i].span = update[i]->level[i].span - (rank[0] - rank[i]);
update[i]->level[i].span = (rank[0] - rank[i]) + 1;
}
/* increment span for untouched levels */
for (i = level; i < zsl->level; i++) {
update[i]->level[i].span++;
}
x->backward = (update[0] == zsl->header) ? NULL : update[0];
if (x->level[0].forward)
x->level[0].forward->backward = x;
else
zsl->tail = x;
zsl->length++;
return x;
}
/* Internal function used by zslDelete, zslDeleteRangeByScore and
* zslDeleteRangeByRank. */
void zslDeleteNode(zskiplist *zsl, zskiplistNode *x, zskiplistNode **update) {
int i;
for (i = 0; i < zsl->level; i++) {
if (update[i]->level[i].forward == x) {
update[i]->level[i].span += x->level[i].span - 1;
update[i]->level[i].forward = x->level[i].forward;
} else {
update[i]->level[i].span -= 1;
}
}
if (x->level[0].forward) {
x->level[0].forward->backward = x->backward;
} else {
zsl->tail = x->backward;
}
while(zsl->level > 1 && zsl->header->level[zsl->level-1].forward == NULL)
zsl->level--;
zsl->length--;
}
/* Delete an element with matching score/element from the skiplist.
* The function returns 1 if the node was found and deleted, otherwise
* 0 is returned.
*
* If 'node' is NULL the deleted node is freed by zslFreeNode(), otherwise
* it is not freed (but just unlinked) and *node is set to the node pointer,
* so that it is possible for the caller to reuse the node (including the
* referenced SDS string at node->ele). */
int zslDelete(zskiplist *zsl, double score, sds ele, zskiplistNode **node) {
zskiplistNode *update[ZSKIPLIST_MAXLEVEL], *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->level[i].forward->score == score &&
sdscmp(x->level[i].forward->ele,ele) < 0)))
{
x = x->level[i].forward;
}
update[i] = x;
}
/* We may have multiple elements with the same score, what we need
* is to find the element with both the right score and object. */
x = x->level[0].forward;
if (x && score == x->score && sdscmp(x->ele,ele) == 0) {
zslDeleteNode(zsl, x, update);
if (!node)
zslFreeNode(x);
else
*node = x;
return 1;
}
return 0; /* not found */
}
/* Update the score of an element inside the sorted set skiplist.
* Note that the element must exist and must match 'score'.
* This function does not update the score in the hash table side, the
* caller should take care of it.
*
* Note that this function attempts to just update the node, in case after
* the score update, the node would be exactly at the same position.
* Otherwise the skiplist is modified by removing and re-adding a new
* element, which is more costly.
*
* The function returns the updated element skiplist node pointer. */
zskiplistNode *zslUpdateScore(zskiplist *zsl, double curscore, sds ele, double newscore) {
zskiplistNode *update[ZSKIPLIST_MAXLEVEL], *x;
int i;
/* We need to seek to element to update to start: this is useful anyway,
* we'll have to update or remove it. */
x = zsl->header;
for (i = zsl->level-1; i >= 0; i--) {
while (x->level[i].forward &&
(x->level[i].forward->score < curscore ||
(x->level[i].forward->score == curscore &&
sdscmp(x->level[i].forward->ele,ele) < 0)))
{
x = x->level[i].forward;
}
update[i] = x;
}
/* Jump to our element: note that this function assumes that the
* element with the matching score exists. */
x = x->level[0].forward;
serverAssert(x && curscore == x->score && sdscmp(x->ele,ele) == 0);
/* If the node, after the score update, would be still exactly
* at the same position, we can just update the score without
* actually removing and re-inserting the element in the skiplist. */
if ((x->backward == NULL || x->backward->score < newscore) &&
(x->level[0].forward == NULL || x->level[0].forward->score > newscore))
{
x->score = newscore;
return x;
}
/* No way to reuse the old node: we need to remove and insert a new
* one at a different place. */
zslDeleteNode(zsl, x, update);
zskiplistNode *newnode = zslInsert(zsl,newscore,x->ele);
/* We reused the old node x->ele SDS string, free the node now
* since zslInsert created a new one. */
x->ele = NULL;
zslFreeNode(x);
return newnode;
}
int zslValueGteMin(double value, zrangespec *spec) {
return spec->minex ? (value > spec->min) : (value >= spec->min);
}
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;
serverAssert(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. */
serverAssert(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.
* Both min and max can be inclusive or exclusive (see range->minex and
* range->maxex). When inclusive a score >= min && score <= max is deleted.
* Note that this function takes the reference to the hash table view of the
* sorted set, in order to remove the elements from the hash table too. */
unsigned long zslDeleteRangeByScore(zskiplist *zsl, zrangespec *range, dict *dict) {
zskiplistNode *update[ZSKIPLIST_MAXLEVEL], *x;
unsigned long removed = 0;
int i;
x = zsl->header;
for (i = zsl->level-1; i >= 0; i--) {
while (x->level[i].forward &&
!zslValueGteMin(x->level[i].forward->score, range))
x = x->level[i].forward;
update[i] = x;
}
/* Current node is the last with score < or <= min. */
x = x->level[0].forward;
/* Delete nodes while in range. */
while (x && zslValueLteMax(x->score, range)) {
zskiplistNode *next = x->level[0].forward;
zslDeleteNode(zsl,x,update);
dictDelete(dict,x->ele);
zslFreeNode(x); /* Here is where x->ele is actually released. */
removed++;
x = next;
}
return removed;
}
unsigned long zslDeleteRangeByLex(zskiplist *zsl, zlexrangespec *range, dict *dict) {
zskiplistNode *update[ZSKIPLIST_MAXLEVEL], *x;
unsigned long removed = 0;
int i;
x = zsl->header;
for (i = zsl->level-1; i >= 0; i--) {
while (x->level[i].forward &&
!zslLexValueGteMin(x->level[i].forward->ele,range))
x = x->level[i].forward;
update[i] = x;
}
/* Current node is the last with score < or <= min. */
x = x->level[0].forward;
/* Delete nodes while in range. */
while (x && zslLexValueLteMax(x->ele,range)) {
zskiplistNode *next = x->level[0].forward;
zslDeleteNode(zsl,x,update);
dictDelete(dict,x->ele);
zslFreeNode(x); /* Here is where x->ele is actually released. */
removed++;
x = next;
}
return removed;
}
/* 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 zslDeleteRangeByRank(zskiplist *zsl, unsigned int start, unsigned int end, dict *dict) {
zskiplistNode *update[ZSKIPLIST_MAXLEVEL], *x;
unsigned long traversed = 0, removed = 0;
int i;
x = zsl->header;
for (i = zsl->level-1; i >= 0; i--) {
while (x->level[i].forward && (traversed + x->level[i].span) < start) {
traversed += x->level[i].span;
x = x->level[i].forward;
}
update[i] = x;
}
traversed++;
x = x->level[0].forward;
while (x && traversed <= end) {
zskiplistNode *next = x->level[0].forward;
zslDeleteNode(zsl,x,update);
dictDelete(dict,x->ele);
zslFreeNode(x);
removed++;
traversed++;
x = next;
}
return removed;
}
/* 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 zslGetRank(zskiplist *zsl, double score, sds ele) {
zskiplistNode *x;
unsigned long rank = 0;
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->level[i].forward->score == score &&
sdscmp(x->level[i].forward->ele,ele) <= 0))) {
rank += x->level[i].span;
x = x->level[i].forward;
}
/* x might be equal to zsl->header, so test if obj is non-NULL */
if (x->ele && x->score == score && sdscmp(x->ele,ele) == 0) {
return rank;
}
}
return 0;
}
/* Finds an element by its rank. The rank argument needs to be 1-based. */
zskiplistNode* zslGetElementByRank(zskiplist *zsl, unsigned long rank) {
zskiplistNode *x;
unsigned long traversed = 0;
int i;
x = zsl->header;
for (i = zsl->level-1; i >= 0; i--) {
while (x->level[i].forward && (traversed + x->level[i].span) <= rank)
{
traversed += x->level[i].span;
x = x->level[i].forward;
}
if (traversed == rank) {
return x;
}
}
return NULL;
}
/* Populate the rangespec according to the objects min and max. */
static int zslParseRange(robj *min, robj *max, zrangespec *spec) {
char *eptr;
spec->minex = spec->maxex = 0;
/* Parse the min-max interval. If one of the values is prefixed
* by the "(" character, it's considered "open". For instance
* ZRANGEBYSCORE zset (1.5 (2.5 will match min < x < max
* ZRANGEBYSCORE zset 1.5 2.5 will instead match min <= x <= max */
if (min->encoding == OBJ_ENCODING_INT) {
spec->min = (long)min->ptr;
} else {
if (((char*)min->ptr)[0] == '(') {
spec->min = strtod((char*)min->ptr+1,&eptr);
if (eptr[0] != '\0' || isnan(spec->min)) return C_ERR;
spec->minex = 1;
} else {
spec->min = strtod((char*)min->ptr,&eptr);
if (eptr[0] != '\0' || isnan(spec->min)) return C_ERR;
}
}
if (max->encoding == OBJ_ENCODING_INT) {
spec->max = (long)max->ptr;
} else {
if (((char*)max->ptr)[0] == '(') {
spec->max = strtod((char*)max->ptr+1,&eptr);
if (eptr[0] != '\0' || isnan(spec->max)) return C_ERR;
spec->maxex = 1;
} else {
spec->max = strtod((char*)max->ptr,&eptr);
if (eptr[0] != '\0' || isnan(spec->max)) return C_ERR;
}
}
return C_OK;
}
/* ------------------------ Lexicographic ranges ---------------------------- */
/* Parse max or min argument of ZRANGEBYLEX.
* (foo means foo (open interval)
* [foo means foo (closed interval)
* - means the min string possible
* + means the max string possible
*
* If the string is valid the *dest pointer is set to the redis object
* that will be used for the comparison, and ex will be set to 0 or 1
* respectively if the item is exclusive or inclusive. C_OK will be
* returned.
*
* If the string is not a valid range C_ERR is returned, and the value
* of *dest and *ex is undefined. */
int zslParseLexRangeItem(robj *item, sds *dest, int *ex) {
char *c = item->ptr;
switch(c[0]) {
case '+':
if (c[1] != '\0') return C_ERR;
*ex = 1;
*dest = shared.maxstring;
return C_OK;
case '-':
if (c[1] != '\0') return C_ERR;
*ex = 1;
*dest = shared.minstring;
return C_OK;
case '(':
*ex = 1;
*dest = sdsnewlen(c+1,sdslen(c)-1);
return C_OK;
case '[':
*ex = 0;
*dest = sdsnewlen(c+1,sdslen(c)-1);
return C_OK;
default:
return C_ERR;
}
}
/* Free a lex range structure, must be called only after zslParseLexRange()
* populated the structure with success (C_OK returned). */
void zslFreeLexRange(zlexrangespec *spec) {
if (spec->min != shared.minstring &&
spec->min != shared.maxstring) sdsfree(spec->min);
if (spec->max != shared.minstring &&
spec->max != shared.maxstring) sdsfree(spec->max);
}
/* Populate the lex rangespec according to the objects min and max.
*
* Return C_OK on success. On error C_ERR is returned.
* When OK is returned the structure must be freed with zslFreeLexRange(),
* otherwise no release is needed. */
int zslParseLexRange(robj *min, robj *max, zlexrangespec *spec) {
/* The range can't be valid if objects are integer encoded.
* Every item must start with ( or [. */
if (min->encoding == OBJ_ENCODING_INT ||
max->encoding == OBJ_ENCODING_INT) return C_ERR;
spec->min = spec->max = NULL;
if (zslParseLexRangeItem(min, &spec->min, &spec->minex) == C_ERR ||
zslParseLexRangeItem(max, &spec->max, &spec->maxex) == C_ERR) {
zslFreeLexRange(spec);
return C_ERR;
} else {
return C_OK;
}
}
/* This is just a wrapper to sdscmp() that is able to
* handle shared.minstring and shared.maxstring as the equivalent of
* -inf and +inf for strings */
int sdscmplex(sds a, sds b) {
if (a == b) return 0;
if (a == shared.minstring || b == shared.maxstring) return -1;
if (a == shared.maxstring || b == shared.minstring) return 1;
return sdscmp(a,b);
}
int zslLexValueGteMin(sds value, zlexrangespec *spec) {
return spec->minex ?
(sdscmplex(value,spec->min) > 0) :
(sdscmplex(value,spec->min) >= 0);
}
int zslLexValueLteMax(sds value, zlexrangespec *spec) {
return spec->maxex ?
(sdscmplex(value,spec->max) < 0) :
(sdscmplex(value,spec->max) <= 0);
}
/* Returns if there is a part of the zset is in the lex range. */
int zslIsInLexRange(zskiplist *zsl, zlexrangespec *range) {
zskiplistNode *x;
/* Test for ranges that will always be empty. */
int cmp = sdscmplex(range->min,range->max);
if (cmp > 0 || (cmp == 0 && (range->minex || range->maxex)))
return 0;
x = zsl->tail;
if (x == NULL || !zslLexValueGteMin(x->ele,range))
return 0;
x = zsl->header->level[0].forward;
if (x == NULL || !zslLexValueLteMax(x->ele,range))
return 0;
return 1;
}
/* Find the first node that is contained in the specified lex range.
* Returns NULL when no element is contained in the range. */
zskiplistNode *zslFirstInLexRange(zskiplist *zsl, zlexrangespec *range) {
zskiplistNode *x;
int i;
/* If everything is out of range, return early. */
if (!zslIsInLexRange(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 &&
!zslLexValueGteMin(x->level[i].forward->ele,range))
x = x->level[i].forward;
}
/* This is an inner range, so the next node cannot be NULL. */
x = x->level[0].forward;
serverAssert(x != NULL);
/* Check if score <= max. */
if (!zslLexValueLteMax(x->ele,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 *zslLastInLexRange(zskiplist *zsl, zlexrangespec *range) {
zskiplistNode *x;
int i;
/* If everything is out of range, return early. */
if (!zslIsInLexRange(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 &&
zslLexValueLteMax(x->level[i].forward->ele,range))
x = x->level[i].forward;
}
/* This is an inner range, so this node cannot be NULL. */
serverAssert(x != NULL);
/* Check if score >= min. */
if (!zslLexValueGteMin(x->ele,range)) return NULL;
return x;
}
/*-----------------------------------------------------------------------------
* Ziplist-backed sorted set API
*----------------------------------------------------------------------------*/
double zzlStrtod(unsigned char *vstr, unsigned int vlen) {
char buf[128];
if (vlen > sizeof(buf))
vlen = sizeof(buf);
memcpy(buf,vstr,vlen);
buf[vlen] = '\0';
return strtod(buf,NULL);
}
double zzlGetScore(unsigned char *sptr) {
unsigned char *vstr;
unsigned int vlen;
long long vlong;
double score;
serverAssert(sptr != NULL);
serverAssert(ziplistGet(sptr,&vstr,&vlen,&vlong));
if (vstr) {
score = zzlStrtod(vstr,vlen);
} else {
score = vlong;
}
return score;
}
/* Return a ziplist element as an SDS string. */
sds ziplistGetObject(unsigned char *sptr) {
unsigned char *vstr;
unsigned int vlen;
long long vlong;
serverAssert(sptr != NULL);
serverAssert(ziplistGet(sptr,&vstr,&vlen,&vlong));
if (vstr) {
return sdsnewlen((char*)vstr,vlen);
} else {
return sdsfromlonglong(vlong);
}
}
/* 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;
serverAssert(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;
serverAssert(*eptr != NULL && *sptr != NULL);
_eptr = ziplistNext(zl,*sptr);
if (_eptr != NULL) {
_sptr = ziplistNext(zl,_eptr);
serverAssert(_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 prev entry. */
void zzlPrev(unsigned char *zl, unsigned char **eptr, unsigned char **sptr) {
unsigned char *_eptr, *_sptr;
serverAssert(*eptr != NULL && *sptr != NULL);
_sptr = ziplistPrev(zl,*eptr);
if (_sptr != NULL) {
_eptr = ziplistPrev(zl,_sptr);
serverAssert(_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. */
if (p == NULL) return 0; /* Empty sorted set */
score = zzlGetScore(p);
if (!zslValueGteMin(score,range))
return 0;
p = ziplistIndex(zl,1); /* First score. */
serverAssert(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);
serverAssert(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);
serverAssert(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)
serverAssert((eptr = ziplistPrev(zl,sptr)) != NULL);
else
eptr = NULL;
}
return NULL;
}
int zzlLexValueGteMin(unsigned char *p, zlexrangespec *spec) {
sds value = ziplistGetObject(p);
int res = zslLexValueGteMin(value,spec);
sdsfree(value);
return res;
}
int zzlLexValueLteMax(unsigned char *p, zlexrangespec *spec) {
sds value = ziplistGetObject(p);
int res = zslLexValueLteMax(value,spec);
sdsfree(value);
return res;
}
/* Returns if there is a part of the zset is in range. Should only be used
* internally by zzlFirstInRange and zzlLastInRange. */
int zzlIsInLexRange(unsigned char *zl, zlexrangespec *range) {
unsigned char *p;
/* Test for ranges that will always be empty. */
int cmp = sdscmplex(range->min,range->max);
if (cmp > 0 || (cmp == 0 && (range->minex || range->maxex)))
return 0;
p = ziplistIndex(zl,-2); /* Last element. */
if (p == NULL) return 0;
if (!zzlLexValueGteMin(p,range))
return 0;
p = ziplistIndex(zl,0); /* First element. */
serverAssert(p != NULL);
if (!zzlLexValueLteMax(p,range))
return 0;
return 1;
}
/* Find pointer to the first element contained in the specified lex range.
* Returns NULL when no element is contained in the range. */
unsigned char *zzlFirstInLexRange(unsigned char *zl, zlexrangespec *range) {
unsigned char *eptr = ziplistIndex(zl,0), *sptr;
/* If everything is out of range, return early. */
if (!zzlIsInLexRange(zl,range)) return NULL;
while (eptr != NULL) {
if (zzlLexValueGteMin(eptr,range)) {
/* Check if score <= max. */
if (zzlLexValueLteMax(eptr,range))
return eptr;
return NULL;
}
/* Move to next element. */
sptr = ziplistNext(zl,eptr); /* This element score. Skip it. */
serverAssert(sptr != NULL);
eptr = ziplistNext(zl,sptr); /* Next element. */
}
return NULL;
}
/* Find pointer to the last element contained in the specified lex range.
* Returns NULL when no element is contained in the range. */
unsigned char *zzlLastInLexRange(unsigned char *zl, zlexrangespec *range) {
unsigned char *eptr = ziplistIndex(zl,-2), *sptr;
/* If everything is out of range, return early. */
if (!zzlIsInLexRange(zl,range)) return NULL;
while (eptr != NULL) {
if (zzlLexValueLteMax(eptr,range)) {
/* Check if score >= min. */
if (zzlLexValueGteMin(eptr,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)
serverAssert((eptr = ziplistPrev(zl,sptr)) != NULL);
else
eptr = NULL;
}
return NULL;
}
unsigned char *zzlFind(unsigned char *zl, sds ele, double *score) {
unsigned char *eptr = ziplistIndex(zl,0), *sptr;
while (eptr != NULL) {
sptr = ziplistNext(zl,eptr);
serverAssert(sptr != NULL);
if (ziplistCompare(eptr,(unsigned char*)ele,sdslen(ele))) {
/* Matching element, pull out score. */
if (score != NULL) *score = zzlGetScore(sptr);
return eptr;
}
/* Move to next element. */
eptr = ziplistNext(zl,sptr);
}
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, sds ele, double score) {
unsigned char *sptr;
char scorebuf[128];
int scorelen;
size_t offset;
scorelen = d2string(scorebuf,sizeof(scorebuf),score);
if (eptr == NULL) {
zl = ziplistPush(zl,(unsigned char*)ele,sdslen(ele),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,(unsigned char*)ele,sdslen(ele));
eptr = zl+offset;
/* Insert score after the element. */
serverAssert((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, sds ele, double score) {
unsigned char *eptr = ziplistIndex(zl,0), *sptr;
double s;
while (eptr != NULL) {
sptr = ziplistNext(zl,eptr);
serverAssert(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,(unsigned char*)ele,sdslen(ele)) > 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);
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;
}
unsigned char *zzlDeleteRangeByLex(unsigned char *zl, zlexrangespec *range, unsigned long *deleted) {
unsigned char *eptr, *sptr;
unsigned long num = 0;
if (deleted != NULL) *deleted = 0;
eptr = zzlFirstInLexRange(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) {
if (zzlLexValueLteMax(eptr,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 long zsetLength(const robj *zobj) {
unsigned long length = 0;
if (zobj->encoding == OBJ_ENCODING_ZIPLIST) {
length = zzlLength(zobj->ptr);
} else if (zobj->encoding == OBJ_ENCODING_SKIPLIST) {
length = ((const zset*)zobj->ptr)->zsl->length;
} else {
serverPanic("Unknown sorted set encoding");
}
return length;
}
void zsetConvert(robj *zobj, int encoding) {
zset *zs;
zskiplistNode *node, *next;
sds ele;
double score;
if (zobj->encoding == encoding) return;
if (zobj->encoding == OBJ_ENCODING_ZIPLIST) {
unsigned char *zl = zobj->ptr;
unsigned char *eptr, *sptr;
unsigned char *vstr;
unsigned int vlen;
long long vlong;
if (encoding != OBJ_ENCODING_SKIPLIST)
serverPanic("Unknown target encoding");
zs = zmalloc(sizeof(*zs));
zs->dict = dictCreate(&zsetDictType,NULL);
zs->zsl = zslCreate();
eptr = ziplistIndex(zl,0);
serverAssertWithInfo(NULL,zobj,eptr != NULL);
sptr = ziplistNext(zl,eptr);
serverAssertWithInfo(NULL,zobj,sptr != NULL);
while (eptr != NULL) {
score = zzlGetScore(sptr);
serverAssertWithInfo(NULL,zobj,ziplistGet(eptr,&vstr,&vlen,&vlong));
if (vstr == NULL)
ele = sdsfromlonglong(vlong);
else
ele = sdsnewlen((char*)vstr,vlen);
node = zslInsert(zs->zsl,score,ele);
serverAssert(dictAdd(zs->dict,ele,&node->score) == DICT_OK);
zzlNext(zl,&eptr,&sptr);
}
zfree(zobj->ptr);
zobj->ptr = zs;
zobj->encoding = OBJ_ENCODING_SKIPLIST;
} else if (zobj->encoding == OBJ_ENCODING_SKIPLIST) {
unsigned char *zl = ziplistNew();
if (encoding != OBJ_ENCODING_ZIPLIST)
serverPanic("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) {
zl = zzlInsertAt(zl,NULL,node->ele,node->score);
next = node->level[0].forward;
zslFreeNode(node);
node = next;
}
zfree(zs);
zobj->ptr = zl;
zobj->encoding = OBJ_ENCODING_ZIPLIST;
} else {
serverPanic("Unknown sorted set encoding");
}
}
/* Convert the sorted set object into a ziplist if it is not already a ziplist
* and if the number of elements and the maximum element size is within the
* expected ranges. */
void zsetConvertToZiplistIfNeeded(robj *zobj, size_t maxelelen) {
if (zobj->encoding == OBJ_ENCODING_ZIPLIST) return;
zset *zset = zobj->ptr;
if (zset->zsl->length <= server.zset_max_ziplist_entries &&
maxelelen <= server.zset_max_ziplist_value)
zsetConvert(zobj,OBJ_ENCODING_ZIPLIST);
}
/* Return (by reference) the score of the specified member of the sorted set
* storing it into *score. If the element does not exist C_ERR is returned
* otherwise C_OK is returned and *score is correctly populated.
* If 'zobj' or 'member' is NULL, C_ERR is returned. */
int zsetScore(robj *zobj, sds member, double *score) {
if (!zobj || !member) return C_ERR;
if (zobj->encoding == OBJ_ENCODING_ZIPLIST) {
if (zzlFind(zobj->ptr, member, score) == NULL) return C_ERR;
} else if (zobj->encoding == OBJ_ENCODING_SKIPLIST) {
zset *zs = zobj->ptr;
dictEntry *de = dictFind(zs->dict, member);
if (de == NULL) return C_ERR;
*score = *(double*)dictGetVal(de);
} else {
serverPanic("Unknown sorted set encoding");
}
return C_OK;
}
/* Add a new element or update the score of an existing element in a sorted
* set, regardless of its encoding.
*
* The set of flags change the command behavior.
*
* The input flags are the following:
*
* ZADD_INCR: Increment the current element score by 'score' instead of updating
* the current element score. If the element does not exist, we
* assume 0 as previous score.
* ZADD_NX: Perform the operation only if the element does not exist.
* ZADD_XX: Perform the operation only if the element already exist.
* ZADD_GT: Perform the operation on existing elements only if the new score is
* greater than the current score.
* ZADD_LT: Perform the operation on existing elements only if the new score is
* less than the current score.
*
* When ZADD_INCR is used, the new score of the element is stored in
* '*newscore' if 'newscore' is not NULL.
*
* The returned flags are the following:
*
* ZADD_NAN: The resulting score is not a number.
* ZADD_ADDED: The element was added (not present before the call).
* ZADD_UPDATED: The element score was updated.
* ZADD_NOP: No operation was performed because of NX or XX.
*
* Return value:
*
* The function returns 1 on success, and sets the appropriate flags
* ADDED or UPDATED to signal what happened during the operation (note that
* none could be set if we re-added an element using the same score it used
* to have, or in the case a zero increment is used).
*
* The function returns 0 on error, currently only when the increment
* produces a NAN condition, or when the 'score' value is NAN since the
* start.
*
* The command as a side effect of adding a new element may convert the sorted
* set internal encoding from ziplist to hashtable+skiplist.
*
* Memory management of 'ele':
*
* The function does not take ownership of the 'ele' SDS string, but copies
* it if needed. */
int zsetAdd(robj *zobj, double score, sds ele, int in_flags, int *out_flags, double *newscore) {
/* Turn options into simple to check vars. */
int incr = (in_flags & ZADD_IN_INCR) != 0;
int nx = (in_flags & ZADD_IN_NX) != 0;
int xx = (in_flags & ZADD_IN_XX) != 0;
int gt = (in_flags & ZADD_IN_GT) != 0;
int lt = (in_flags & ZADD_IN_LT) != 0;
*out_flags = 0; /* We'll return our response flags. */
double curscore;
/* NaN as input is an error regardless of all the other parameters. */
if (isnan(score)) {
*out_flags = ZADD_OUT_NAN;
return 0;
}
/* Update the sorted set according to its encoding. */
if (zobj->encoding == OBJ_ENCODING_ZIPLIST) {
unsigned char *eptr;
if ((eptr = zzlFind(zobj->ptr,ele,&curscore)) != NULL) {
/* NX? Return, same element already exists. */
if (nx) {
*out_flags |= ZADD_OUT_NOP;
return 1;
}
/* Prepare the score for the increment if needed. */
if (incr) {
score += curscore;
if (isnan(score)) {
*out_flags |= ZADD_OUT_NAN;
return 0;
}
}
/* GT/LT? Only update if score is greater/less than current. */
if ((lt && score >= curscore) || (gt && score <= curscore)) {
*out_flags |= ZADD_OUT_NOP;
return 1;
}
if (newscore) *newscore = score;
/* Remove and re-insert when score changed. */
if (score != curscore) {
zobj->ptr = zzlDelete(zobj->ptr,eptr);
zobj->ptr = zzlInsert(zobj->ptr,ele,score);
*out_flags |= ZADD_OUT_UPDATED;
}
return 1;
} else if (!xx) {
/* 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 ||
sdslen(ele) > server.zset_max_ziplist_value)
zsetConvert(zobj,OBJ_ENCODING_SKIPLIST);
if (newscore) *newscore = score;
*out_flags |= ZADD_OUT_ADDED;
return 1;
} else {
*out_flags |= ZADD_OUT_NOP;
return 1;
}
} else if (zobj->encoding == OBJ_ENCODING_SKIPLIST) {
zset *zs = zobj->ptr;
zskiplistNode *znode;
dictEntry *de;
de = dictFind(zs->dict,ele);
if (de != NULL) {
/* NX? Return, same element already exists. */
if (nx) {
*out_flags |= ZADD_OUT_NOP;
return 1;
}
curscore = *(double*)dictGetVal(de);
/* Prepare the score for the increment if needed. */
if (incr) {
score += curscore;
if (isnan(score)) {
*out_flags |= ZADD_OUT_NAN;
return 0;
}
}
/* GT/LT? Only update if score is greater/less than current. */
if ((lt && score >= curscore) || (gt && score <= curscore)) {
*out_flags |= ZADD_OUT_NOP;
return 1;
}
if (newscore) *newscore = score;
/* Remove and re-insert when score changes. */
if (score != curscore) {
znode = zslUpdateScore(zs->zsl,curscore,ele,score);
/* Note that we did not removed the original element from
* the hash table representing the sorted set, so we just
* update the score. */
dictGetVal(de) = &znode->score; /* Update score ptr. */
*out_flags |= ZADD_OUT_UPDATED;
}
return 1;
} else if (!xx) {
ele = sdsdup(ele);
znode = zslInsert(zs->zsl,score,ele);
serverAssert(dictAdd(zs->dict,ele,&znode->score) == DICT_OK);
*out_flags |= ZADD_OUT_ADDED;
if (newscore) *newscore = score;
return 1;
} else {
*out_flags |= ZADD_OUT_NOP;
return 1;
}
} else {
serverPanic("Unknown sorted set encoding");
}
return 0; /* Never reached. */
}
/* Deletes the element 'ele' from the sorted set encoded as a skiplist+dict,
* returning 1 if the element existed and was deleted, 0 otherwise (the
* element was not there). It does not resize the dict after deleting the
* element. */
static int zsetRemoveFromSkiplist(zset *zs, sds ele) {
dictEntry *de;
double score;
de = dictUnlink(zs->dict,ele);
if (de != NULL) {
/* Get the score in order to delete from the skiplist later. */
score = *(double*)dictGetVal(de);
/* Delete from the hash table and later from the skiplist.
* Note that the order is important: deleting from the skiplist
* actually releases the SDS string representing the element,
* which is shared between the skiplist and the hash table, so
* we need to delete from the skiplist as the final step. */
dictFreeUnlinkedEntry(zs->dict,de);
/* Delete from skiplist. */
int retval = zslDelete(zs->zsl,score,ele,NULL);
serverAssert(retval);
return 1;
}
return 0;
}
/* Delete the element 'ele' from the sorted set, returning 1 if the element
* existed and was deleted, 0 otherwise (the element was not there). */
int zsetDel(robj *zobj, sds ele) {
if (zobj->encoding == OBJ_ENCODING_ZIPLIST) {
unsigned char *eptr;
if ((eptr = zzlFind(zobj->ptr,ele,NULL)) != NULL) {
zobj->ptr = zzlDelete(zobj->ptr,eptr);
return 1;
}
} else if (zobj->encoding == OBJ_ENCODING_SKIPLIST) {
zset *zs = zobj->ptr;
if (zsetRemoveFromSkiplist(zs, ele)) {
if (htNeedsResize(zs->dict)) dictResize(zs->dict);
return 1;
}
} else {
serverPanic("Unknown sorted set encoding");
}
return 0; /* No such element found. */
}
/* Given a sorted set object returns the 0-based rank of the object or
* -1 if the object does not exist.
*
* For rank we mean the position of the element in the sorted collection
* of elements. So the first element has rank 0, the second rank 1, and so
* forth up to length-1 elements.
*
* If 'reverse' is false, the rank is returned considering as first element
* the one with the lowest score. Otherwise if 'reverse' is non-zero
* the rank is computed considering as element with rank 0 the one with
* the highest score. */
long zsetRank(robj *zobj, sds ele, int reverse) {
unsigned long llen;
unsigned long rank;
llen = zsetLength(zobj);
if (zobj->encoding == OBJ_ENCODING_ZIPLIST) {
unsigned char *zl = zobj->ptr;
unsigned char *eptr, *sptr;
eptr = ziplistIndex(zl,0);
serverAssert(eptr != NULL);
sptr = ziplistNext(zl,eptr);
serverAssert(sptr != NULL);
rank = 1;
while(eptr != NULL) {
if (ziplistCompare(eptr,(unsigned char*)ele,sdslen(ele)))
break;
rank++;
zzlNext(zl,&eptr,&sptr);
}
if (eptr != NULL) {
if (reverse)
return llen-rank;
else
return rank-1;
} else {
return -1;
}
} else if (zobj->encoding == OBJ_ENCODING_SKIPLIST) {
zset *zs = zobj->ptr;
zskiplist *zsl = zs->zsl;
dictEntry *de;
double score;
de = dictFind(zs->dict,ele);
if (de != NULL) {
score = *(double*)dictGetVal(de);
rank = zslGetRank(zsl,score,ele);
/* Existing elements always have a rank. */
serverAssert(rank != 0);
if (reverse)
return llen-rank;
else
return rank-1;
} else {
return -1;
}
} else {
serverPanic("Unknown sorted set encoding");
}
}
/* This is a helper function for the COPY command.
* Duplicate a sorted set object, with the guarantee that the returned object
* has the same encoding as the original one.
*
* The resulting object always has refcount set to 1 */
robj *zsetDup(robj *o) {
robj *zobj;
zset *zs;
zset *new_zs;
serverAssert(o->type == OBJ_ZSET);
/* Create a new sorted set object that have the same encoding as the original object's encoding */
if (o->encoding == OBJ_ENCODING_ZIPLIST) {
unsigned char *zl = o->ptr;
size_t sz = ziplistBlobLen(zl);
unsigned char *new_zl = zmalloc(sz);
memcpy(new_zl, zl, sz);
zobj = createObject(OBJ_ZSET, new_zl);
zobj->encoding = OBJ_ENCODING_ZIPLIST;
} else if (o->encoding == OBJ_ENCODING_SKIPLIST) {
zobj = createZsetObject();
zs = o->ptr;
new_zs = zobj->ptr;
dictExpand(new_zs->dict,dictSize(zs->dict));
zskiplist *zsl = zs->zsl;
zskiplistNode *ln;
sds ele;
long llen = zsetLength(o);
/* We copy the skiplist elements from the greatest to the
* smallest (that's trivial since the elements are already ordered in
* the skiplist): this improves the load process, since the next loaded
* element will always be the smaller, so adding to the skiplist
* will always immediately stop at the head, making the insertion
* O(1) instead of O(log(N)). */
ln = zsl->tail;
while (llen--) {
ele = ln->ele;
sds new_ele = sdsdup(ele);
zskiplistNode *znode = zslInsert(new_zs->zsl,ln->score,new_ele);
dictAdd(new_zs->dict,new_ele,&znode->score);
ln = ln->backward;
}
} else {
serverPanic("Unknown sorted set encoding");
}
return zobj;
}
/* callback for to check the ziplist doesn't have duplicate records */
static int _zsetZiplistValidateIntegrity(unsigned char *p, void *userdata) {
struct {
long count;
dict *fields;
} *data = userdata;
/* Even records are field names, add to dict and check that's not a dup */
if (((data->count) & 1) == 0) {
unsigned char *str;
unsigned int slen;
long long vll;
if (!ziplistGet(p, &str, &slen, &vll))
return 0;
sds field = str? sdsnewlen(str, slen): sdsfromlonglong(vll);
if (dictAdd(data->fields, field, NULL) != DICT_OK) {
/* Duplicate, return an error */
sdsfree(field);
return 0;
}
}
(data->count)++;
return 1;
}
/* Validate the integrity of the data structure.
* when `deep` is 0, only the integrity of the header is validated.
* when `deep` is 1, we scan all the entries one by one. */
int zsetZiplistValidateIntegrity(unsigned char *zl, size_t size, int deep) {
if (!deep)
return ziplistValidateIntegrity(zl, size, 0, NULL, NULL);
/* Keep track of the field names to locate duplicate ones */
struct {
long count;
dict *fields;
} data = {0, dictCreate(&hashDictType, NULL)};
int ret = ziplistValidateIntegrity(zl, size, 1, _zsetZiplistValidateIntegrity, &data);
/* make sure we have an even number of records. */
if (data.count & 1)
ret = 0;
dictRelease(data.fields);
return ret;
}
/* Create a new sds string from the ziplist entry. */
sds zsetSdsFromZiplistEntry(ziplistEntry *e) {
return e->sval ? sdsnewlen(e->sval, e->slen) : sdsfromlonglong(e->lval);
}
/* Reply with bulk string from the ziplist entry. */
void zsetReplyFromZiplistEntry(client *c, ziplistEntry *e) {
if (e->sval)
addReplyBulkCBuffer(c, e->sval, e->slen);
else
addReplyBulkLongLong(c, e->lval);
}
/* Return random element from a non empty zset.
* 'key' and 'val' will be set to hold the element.
* The memory in `key` is not to be freed or modified by the caller.
* 'score' can be NULL in which case it's not extracted. */
void zsetTypeRandomElement(robj *zsetobj, unsigned long zsetsize, ziplistEntry *key, double *score) {
if (zsetobj->encoding == OBJ_ENCODING_SKIPLIST) {
zset *zs = zsetobj->ptr;
dictEntry *de = dictGetFairRandomKey(zs->dict);
sds s = dictGetKey(de);
key->sval = (unsigned char*)s;
key->slen = sdslen(s);
if (score)
*score = *(double*)dictGetVal(de);
} else if (zsetobj->encoding == OBJ_ENCODING_ZIPLIST) {
ziplistEntry val;
ziplistRandomPair(zsetobj->ptr, zsetsize, key, &val);
if (score) {
if (val.sval) {
*score = zzlStrtod(val.sval,val.slen);
} else {
*score = (double)val.lval;
}
}
} else {
serverPanic("Unknown zset encoding");
}
}
/*-----------------------------------------------------------------------------
* Sorted set commands
*----------------------------------------------------------------------------*/
/* This generic command implements both ZADD and ZINCRBY. */
void zaddGenericCommand(client *c, int flags) {
static char *nanerr = "resulting score is not a number (NaN)";
robj *key = c->argv[1];
robj *zobj;
sds ele;
double score = 0, *scores = NULL;
int j, elements, ch = 0;
int scoreidx = 0;
/* The following vars are used in order to track what the command actually
* did during the execution, to reply to the client and to trigger the
* notification of keyspace change. */
int added = 0; /* Number of new elements added. */
int updated = 0; /* Number of elements with updated score. */
int processed = 0; /* Number of elements processed, may remain zero with
options like XX. */
/* Parse options. At the end 'scoreidx' is set to the argument position
* of the score of the first score-element pair. */
scoreidx = 2;
while(scoreidx < c->argc) {
char *opt = c->argv[scoreidx]->ptr;
if (!strcasecmp(opt,"nx")) flags |= ZADD_IN_NX;
else if (!strcasecmp(opt,"xx")) flags |= ZADD_IN_XX;
else if (!strcasecmp(opt,"ch")) ch = 1; /* Return num of elements added or updated. */
else if (!strcasecmp(opt,"incr")) flags |= ZADD_IN_INCR;
else if (!strcasecmp(opt,"gt")) flags |= ZADD_IN_GT;
else if (!strcasecmp(opt,"lt")) flags |= ZADD_IN_LT;
else break;
scoreidx++;
}
/* Turn options into simple to check vars. */
int incr = (flags & ZADD_IN_INCR) != 0;
int nx = (flags & ZADD_IN_NX) != 0;
int xx = (flags & ZADD_IN_XX) != 0;
int gt = (flags & ZADD_IN_GT) != 0;
int lt = (flags & ZADD_IN_LT) != 0;
/* After the options, we expect to have an even number of args, since
* we expect any number of score-element pairs. */
elements = c->argc-scoreidx;
if (elements % 2 || !elements) {
addReplyErrorObject(c,shared.syntaxerr);
return;
}
elements /= 2; /* Now this holds the number of score-element pairs. */
/* Check for incompatible options. */
if (nx && xx) {
addReplyError(c,
"XX and NX options at the same time are not compatible");
return;
}
if ((gt && nx) || (lt && nx) || (gt && lt)) {
addReplyError(c,
"GT, LT, and/or NX options at the same time are not compatible");
return;
}
/* Note that XX is compatible with either GT or LT */
if (incr && elements > 1) {
addReplyError(c,
"INCR option supports a single increment-element pair");
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[scoreidx+j*2],&scores[j],NULL)
!= C_OK) goto cleanup;
}
/* Lookup the key and create the sorted set if does not exist. */
zobj = lookupKeyWrite(c->db,key);
if (checkType(c,zobj,OBJ_ZSET)) goto cleanup;
if (zobj == NULL) {
if (xx) goto reply_to_client; /* No key + XX option: nothing to do. */
if (server.zset_max_ziplist_entries == 0 ||
server.zset_max_ziplist_value < sdslen(c->argv[scoreidx+1]->ptr))
{
zobj = createZsetObject();
} else {
zobj = createZsetZiplistObject();
}
dbAdd(c->db,key,zobj);
}
for (j = 0; j < elements; j++) {
double newscore;
score = scores[j];
int retflags = 0;
ele = c->argv[scoreidx+1+j*2]->ptr;
int retval = zsetAdd(zobj, score, ele, flags, &retflags, &newscore);
if (retval == 0) {
addReplyError(c,nanerr);
goto cleanup;
}
if (retflags & ZADD_OUT_ADDED) added++;
if (retflags & ZADD_OUT_UPDATED) updated++;
if (!(retflags & ZADD_OUT_NOP)) processed++;
score = newscore;
}
server.dirty += (added+updated);
reply_to_client:
if (incr) { /* ZINCRBY or INCR option. */
if (processed)
addReplyDouble(c,score);
else
addReplyNull(c);
} else { /* ZADD. */
addReplyLongLong(c,ch ? added+updated : added);
}
cleanup:
zfree(scores);
if (added || updated) {
signalModifiedKey(c,c->db,key);
notifyKeyspaceEvent(NOTIFY_ZSET,
incr ? "zincr" : "zadd", key, c->db->id);
}
}
void zaddCommand(client *c) {
zaddGenericCommand(c,ZADD_IN_NONE);
}
void zincrbyCommand(client *c) {
zaddGenericCommand(c,ZADD_IN_INCR);
}
void zremCommand(client *c) {
robj *key = c->argv[1];
robj *zobj;
int deleted = 0, keyremoved = 0, j;
if ((zobj = lookupKeyWriteOrReply(c,key,shared.czero)) == NULL ||
checkType(c,zobj,OBJ_ZSET)) return;
for (j = 2; j < c->argc; j++) {
if (zsetDel(zobj,c->argv[j]->ptr)) deleted++;
if (zsetLength(zobj) == 0) {
dbDelete(c->db,key);
keyremoved = 1;
break;
}
}
if (deleted) {
notifyKeyspaceEvent(NOTIFY_ZSET,"zrem",key,c->db->id);
if (keyremoved)
notifyKeyspaceEvent(NOTIFY_GENERIC,"del",key,c->db->id);
signalModifiedKey(c,c->db,key);
server.dirty += deleted;
}
addReplyLongLong(c,deleted);
}
typedef enum {
ZRANGE_AUTO = 0,
ZRANGE_RANK,
ZRANGE_SCORE,
ZRANGE_LEX,
} zrange_type;
/* Implements ZREMRANGEBYRANK, ZREMRANGEBYSCORE, ZREMRANGEBYLEX commands. */
void zremrangeGenericCommand(client *c, zrange_type rangetype) {
robj *key = c->argv[1];
robj *zobj;
int keyremoved = 0;
unsigned long deleted = 0;
zrangespec range;
zlexrangespec lexrange;
long start, end, llen;
char *notify_type = NULL;
/* Step 1: Parse the range. */
if (rangetype == ZRANGE_RANK) {
notify_type = "zremrangebyrank";
if ((getLongFromObjectOrReply(c,c->argv[2],&start,NULL) != C_OK) ||
(getLongFromObjectOrReply(c,c->argv[3],&end,NULL) != C_OK))
return;
} else if (rangetype == ZRANGE_SCORE) {
notify_type = "zremrangebyscore";
if (zslParseRange(c->argv[2],c->argv[3],&range) != C_OK) {
addReplyError(c,"min or max is not a float");
return;
}
} else if (rangetype == ZRANGE_LEX) {
notify_type = "zremrangebylex";
if (zslParseLexRange(c->argv[2],c->argv[3],&lexrange) != C_OK) {
addReplyError(c,"min or max not valid string range item");
return;
}
} else {
serverPanic("unknown rangetype %d", (int)rangetype);
}
/* Step 2: Lookup & range sanity checks if needed. */
if ((zobj = lookupKeyWriteOrReply(c,key,shared.czero)) == NULL ||
checkType(c,zobj,OBJ_ZSET)) goto cleanup;
if (rangetype == ZRANGE_RANK) {
/* Sanitize indexes. */
llen = zsetLength(zobj);
if (start < 0) start = llen+start;
if (end < 0) end = llen+end;
if (start < 0) start = 0;
/* Invariant: start >= 0, so this test will be true when end < 0.
* The range is empty when start > end or start >= length. */
if (start > end || start >= llen) {
addReply(c,shared.czero);
goto cleanup;
}
if (end >= llen) end = llen-1;
}
/* Step 3: Perform the range deletion operation. */
if (zobj->encoding == OBJ_ENCODING_ZIPLIST) {
switch(rangetype) {
case ZRANGE_AUTO:
case ZRANGE_RANK:
zobj->ptr = zzlDeleteRangeByRank(zobj->ptr,start+1,end+1,&deleted);
break;
case ZRANGE_SCORE:
zobj->ptr = zzlDeleteRangeByScore(zobj->ptr,&range,&deleted);
break;
case ZRANGE_LEX:
zobj->ptr = zzlDeleteRangeByLex(zobj->ptr,&lexrange,&deleted);
break;
}
if (zzlLength(zobj->ptr) == 0) {
dbDelete(c->db,key);
keyremoved = 1;
}
} else if (zobj->encoding == OBJ_ENCODING_SKIPLIST) {
zset *zs = zobj->ptr;
switch(rangetype) {
case ZRANGE_AUTO:
case ZRANGE_RANK:
deleted = zslDeleteRangeByRank(zs->zsl,start+1,end+1,zs->dict);
break;
case ZRANGE_SCORE:
deleted = zslDeleteRangeByScore(zs->zsl,&range,zs->dict);
break;
case ZRANGE_LEX:
deleted = zslDeleteRangeByLex(zs->zsl,&lexrange,zs->dict);
break;
}
if (htNeedsResize(zs->dict)) dictResize(zs->dict);
if (dictSize(zs->dict) == 0) {
dbDelete(c->db,key);
keyremoved = 1;
}
} else {
serverPanic("Unknown sorted set encoding");
}
/* Step 4: Notifications and reply. */
if (deleted) {
signalModifiedKey(c,c->db,key);
notifyKeyspaceEvent(NOTIFY_ZSET,notify_type,key,c->db->id);
if (keyremoved)
notifyKeyspaceEvent(NOTIFY_GENERIC,"del",key,c->db->id);
}
server.dirty += deleted;
addReplyLongLong(c,deleted);
cleanup:
if (rangetype == ZRANGE_LEX) zslFreeLexRange(&lexrange);
}
void zremrangebyrankCommand(client *c) {
zremrangeGenericCommand(c,ZRANGE_RANK);
}
void zremrangebyscoreCommand(client *c) {
zremrangeGenericCommand(c,ZRANGE_SCORE);
}
void zremrangebylexCommand(client *c) {
zremrangeGenericCommand(c,ZRANGE_LEX);
}
typedef struct {
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;
/* 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_SDS 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. */
sds 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 == OBJ_SET) {
iterset *it = &op->iter.set;
if (op->encoding == OBJ_ENCODING_INTSET) {
it->is.is = op->subject->ptr;
it->is.ii = 0;
} else if (op->encoding == OBJ_ENCODING_HT) {
it->ht.dict = op->subject->ptr;
it->ht.di = dictGetIterator(op->subject->ptr);
it->ht.de = dictNext(it->ht.di);
} else {
serverPanic("Unknown set encoding");
}
} else if (op->type == OBJ_ZSET) {
/* Sorted sets are traversed in reverse order to optimize for
* the insertion of the elements in a new list as in
* ZDIFF/ZINTER/ZUNION */
iterzset *it = &op->iter.zset;
if (op->encoding == OBJ_ENCODING_ZIPLIST) {
it->zl.zl = op->subject->ptr;
it->zl.eptr = ziplistIndex(it->zl.zl,-2);
if (it->zl.eptr != NULL) {
it->zl.sptr = ziplistNext(it->zl.zl,it->zl.eptr);
serverAssert(it->zl.sptr != NULL);
}
} else if (op->encoding == OBJ_ENCODING_SKIPLIST) {
it->sl.zs = op->subject->ptr;
it->sl.node = it->sl.zs->zsl->tail;
} else {
serverPanic("Unknown sorted set encoding");
}
} else {
serverPanic("Unsupported type");
}
}
void zuiClearIterator(zsetopsrc *op) {
if (op->subject == NULL)
return;
if (op->type == OBJ_SET) {
iterset *it = &op->iter.set;
if (op->encoding == OBJ_ENCODING_INTSET) {
UNUSED(it); /* skip */
} else if (op->encoding == OBJ_ENCODING_HT) {
dictReleaseIterator(it->ht.di);
} else {
serverPanic("Unknown set encoding");
}
} else if (op->type == OBJ_ZSET) {
iterzset *it = &op->iter.zset;
if (op->encoding == OBJ_ENCODING_ZIPLIST) {
UNUSED(it); /* skip */
} else if (op->encoding == OBJ_ENCODING_SKIPLIST) {
UNUSED(it); /* skip */
} else {
serverPanic("Unknown sorted set encoding");
}
} else {
serverPanic("Unsupported type");
}
}
unsigned long zuiLength(zsetopsrc *op) {
if (op->subject == NULL)
return 0;
if (op->type == OBJ_SET) {
if (op->encoding == OBJ_ENCODING_INTSET) {
return intsetLen(op->subject->ptr);
} else if (op->encoding == OBJ_ENCODING_HT) {
dict *ht = op->subject->ptr;
return dictSize(ht);
} else {
serverPanic("Unknown set encoding");
}
} else if (op->type == OBJ_ZSET) {
if (op->encoding == OBJ_ENCODING_ZIPLIST) {
return zzlLength(op->subject->ptr);
} else if (op->encoding == OBJ_ENCODING_SKIPLIST) {
zset *zs = op->subject->ptr;
return zs->zsl->length;
} else {
serverPanic("Unknown sorted set encoding");
}
} else {
serverPanic("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_SDS)
sdsfree(val->ele);
memset(val,0,sizeof(zsetopval));
if (op->type == OBJ_SET) {
iterset *it = &op->iter.set;
if (op->encoding == OBJ_ENCODING_INTSET) {
int64_t ell;
if (!intsetGet(it->is.is,it->is.ii,&ell))
return 0;
val->ell = ell;
val->score = 1.0;
/* Move to next element. */
it->is.ii++;
} else if (op->encoding == OBJ_ENCODING_HT) {
if (it->ht.de == NULL)
return 0;
val->ele = dictGetKey(it->ht.de);
val->score = 1.0;
/* Move to next element. */
it->ht.de = dictNext(it->ht.di);
} else {
serverPanic("Unknown set encoding");
}
} else if (op->type == OBJ_ZSET) {
iterzset *it = &op->iter.zset;
if (op->encoding == OBJ_ENCODING_ZIPLIST) {
/* No need to check both, but better be explicit. */
if (it->zl.eptr == NULL || it->zl.sptr == NULL)
return 0;
serverAssert(ziplistGet(it->zl.eptr,&val->estr,&val->elen,&val->ell));
val->score = zzlGetScore(it->zl.sptr);
/* Move to next element (going backwards, see zuiInitIterator). */
zzlPrev(it->zl.zl,&it->zl.eptr,&it->zl.sptr);
} else if (op->encoding == OBJ_ENCODING_SKIPLIST) {
if (it->sl.node == NULL)
return 0;
val->ele = it->sl.node->ele;
val->score = it->sl.node->score;
/* Move to next element. (going backwards, see zuiInitIterator) */
it->sl.node = it->sl.node->backward;
} else {
serverPanic("Unknown sorted set encoding");
}
} else {
serverPanic("Unsupported type");
}
return 1;
}
int zuiLongLongFromValue(zsetopval *val) {
if (!(val->flags & OPVAL_DIRTY_LL)) {
val->flags |= OPVAL_DIRTY_LL;
if (val->ele != NULL) {
if (string2ll(val->ele,sdslen(val->ele),&val->ell))
val->flags |= OPVAL_VALID_LL;
} 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;
}
sds zuiSdsFromValue(zsetopval *val) {
if (val->ele == NULL) {
if (val->estr != NULL) {
val->ele = sdsnewlen((char*)val->estr,val->elen);
} else {
val->ele = sdsfromlonglong(val->ell);
}
val->flags |= OPVAL_DIRTY_SDS;
}
return val->ele;
}
/* This is different from zuiSdsFromValue since returns a new SDS string
* which is up to the caller to free. */
sds zuiNewSdsFromValue(zsetopval *val) {
if (val->flags & OPVAL_DIRTY_SDS) {
/* We have already one to return! */
sds ele = val->ele;
val->flags &= ~OPVAL_DIRTY_SDS;
val->ele = NULL;
return ele;
} else if (val->ele) {
return sdsdup(val->ele);
} else if (val->estr) {
return sdsnewlen((char*)val->estr,val->elen);
} else {
return sdsfromlonglong(val->ell);
}
}
int zuiBufferFromValue(zsetopval *val) {
if (val->estr == NULL) {
if (val->ele != NULL) {
val->elen = sdslen(val->ele);
val->estr = (unsigned char*)val->ele;
} 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 == OBJ_SET) {
if (op->encoding == OBJ_ENCODING_INTSET) {
if (zuiLongLongFromValue(val) &&
intsetFind(op->subject->ptr,val->ell))
{
*score = 1.0;
return 1;
} else {
return 0;
}
} else if (op->encoding == OBJ_ENCODING_HT) {
dict *ht = op->subject->ptr;
zuiSdsFromValue(val);
if (dictFind(ht,val->ele) != NULL) {
*score = 1.0;
return 1;
} else {
return 0;
}
} else {
serverPanic("Unknown set encoding");
}
} else if (op->type == OBJ_ZSET) {
zuiSdsFromValue(val);
if (op->encoding == OBJ_ENCODING_ZIPLIST) {
if (zzlFind(op->subject->ptr,val->ele,score) != NULL) {
/* Score is already set by zzlFind. */
return 1;
} else {
return 0;
}
} else if (op->encoding == OBJ_ENCODING_SKIPLIST) {
zset *zs = op->subject->ptr;
dictEntry *de;
if ((de = dictFind(zs->dict,val->ele)) != NULL) {
*score = *(double*)dictGetVal(de);
return 1;
} else {
return 0;
}
} else {
serverPanic("Unknown sorted set encoding");
}
} else {
serverPanic("Unsupported type");
}
}
int zuiCompareByCardinality(const void *s1, const void *s2) {
unsigned long first = zuiLength((zsetopsrc*)s1);
unsigned long second = zuiLength((zsetopsrc*)s2);
if (first > second) return 1;
if (first < second) return -1;
return 0;
}
static int zuiCompareByRevCardinality(const void *s1, const void *s2) {
return zuiCompareByCardinality(s1, s2) * -1;
}
#define REDIS_AGGR_SUM 1
#define REDIS_AGGR_MIN 2
#define REDIS_AGGR_MAX 3
#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) {
*target = *target + val;
/* The result of adding two doubles is NaN when one variable
* is +inf and the other is -inf. When these numbers are added,
* we maintain the convention of the result being 0.0. */
if (isnan(*target)) *target = 0.0;
} else if (aggregate == REDIS_AGGR_MIN) {
*target = val < *target ? val : *target;
} else if (aggregate == REDIS_AGGR_MAX) {
*target = val > *target ? val : *target;
} else {
/* safety net */
serverPanic("Unknown ZUNION/INTER aggregate type");
}
}
static int zsetDictGetMaxElementLength(dict *d) {
dictIterator *di;
dictEntry *de;
size_t maxelelen = 0;
di = dictGetIterator(d);
while((de = dictNext(di)) != NULL) {
sds ele = dictGetKey(de);
if (sdslen(ele) > maxelelen) maxelelen = sdslen(ele);
}
dictReleaseIterator(di);
return maxelelen;
}
static void zdiffAlgorithm1(zsetopsrc *src, long setnum, zset *dstzset, size_t *maxelelen) {
/* DIFF Algorithm 1:
*
* We perform the diff by iterating all the elements of the first set,
* and only adding it to the target set if the element does not exist
* into all the other sets.
*
* This way we perform at max N*M operations, where N is the size of
* the first set, and M the number of sets.
*
* There is also a O(K*log(K)) cost for adding the resulting elements
* to the target set, where K is the final size of the target set.
*
* The final complexity of this algorithm is O(N*M + K*log(K)). */
int j;
zsetopval zval;
zskiplistNode *znode;
sds tmp;
/* With algorithm 1 it is better to order the sets to subtract
* by decreasing size, so that we are more likely to find
* duplicated elements ASAP. */
qsort(src+1,setnum-1,sizeof(zsetopsrc),zuiCompareByRevCardinality);
memset(&zval, 0, sizeof(zval));
zuiInitIterator(&src[0]);
while (zuiNext(&src[0],&zval)) {
double value;
int exists = 0;
for (j = 1; j < setnum; j++) {
/* It is not safe to access the zset we are
* iterating, so explicitly check for equal object.
* This check isn't really needed anymore since we already
* check for a duplicate set in the zsetChooseDiffAlgorithm
* function, but we're leaving it for future-proofing. */
if (src[j].subject == src[0].subject ||
zuiFind(&src[j],&zval,&value)) {
exists = 1;
break;
}
}
if (!exists) {
tmp = zuiNewSdsFromValue(&zval);
znode = zslInsert(dstzset->zsl,zval.score,tmp);
dictAdd(dstzset->dict,tmp,&znode->score);
if (sdslen(tmp) > *maxelelen) *maxelelen = sdslen(tmp);
}
}
zuiClearIterator(&src[0]);
}
static void zdiffAlgorithm2(zsetopsrc *src, long setnum, zset *dstzset, size_t *maxelelen) {
/* DIFF Algorithm 2:
*
* Add all the elements of the first set to the auxiliary set.
* Then remove all the elements of all the next sets from it.
*
* This is O(L + (N-K)log(N)) where L is the sum of all the elements in every
* set, N is the size of the first set, and K is the size of the result set.
*
* Note that from the (L-N) dict searches, (N-K) got to the zsetRemoveFromSkiplist
* which costs log(N)
*
* There is also a O(K) cost at the end for finding the largest element
* size, but this doesn't change the algorithm complexity since K < L, and
* O(2L) is the same as O(L). */
int j;
int cardinality = 0;
zsetopval zval;
zskiplistNode *znode;
sds tmp;
for (j = 0; j < setnum; j++) {
if (zuiLength(&src[j]) == 0) continue;
memset(&zval, 0, sizeof(zval));
zuiInitIterator(&src[j]);
while (zuiNext(&src[j],&zval)) {
if (j == 0) {
tmp = zuiNewSdsFromValue(&zval);
znode = zslInsert(dstzset->zsl,zval.score,tmp);
dictAdd(dstzset->dict,tmp,&znode->score);
cardinality++;
} else {
tmp = zuiSdsFromValue(&zval);
if (zsetRemoveFromSkiplist(dstzset, tmp)) {
cardinality--;
}
}
/* Exit if result set is empty as any additional removal
* of elements will have no effect. */
if (cardinality == 0) break;
}
zuiClearIterator(&src[j]);
if (cardinality == 0) break;
}
/* Resize dict if needed after removing multiple elements */
if (htNeedsResize(dstzset->dict)) dictResize(dstzset->dict);
/* Using this algorithm, we can't calculate the max element as we go,
* we have to iterate through all elements to find the max one after. */
*maxelelen = zsetDictGetMaxElementLength(dstzset->dict);
}
static int zsetChooseDiffAlgorithm(zsetopsrc *src, long setnum) {
int j;
/* Select what DIFF algorithm to use.
*
* Algorithm 1 is O(N*M + K*log(K)) where N is the size of the
* first set, M the total number of sets, and K is the size of the
* result set.
*
* Algorithm 2 is O(L + (N-K)log(N)) where L is the total number of elements
* in all the sets, N is the size of the first set, and K is the size of the
* result set.
*
* We compute what is the best bet with the current input here. */
long long algo_one_work = 0;
long long algo_two_work = 0;
for (j = 0; j < setnum; j++) {
/* If any other set is equal to the first set, there is nothing to be
* done, since we would remove all elements anyway. */
if (j > 0 && src[0].subject == src[j].subject) {
return 0;
}
algo_one_work += zuiLength(&src[0]);
algo_two_work += zuiLength(&src[j]);
}
/* Algorithm 1 has better constant times and performs less operations
* if there are elements in common. Give it some advantage. */
algo_one_work /= 2;
return (algo_one_work <= algo_two_work) ? 1 : 2;
}
static void zdiff(zsetopsrc *src, long setnum, zset *dstzset, size_t *maxelelen) {
/* Skip everything if the smallest input is empty. */
if (zuiLength(&src[0]) > 0) {
int diff_algo = zsetChooseDiffAlgorithm(src, setnum);
if (diff_algo == 1) {
zdiffAlgorithm1(src, setnum, dstzset, maxelelen);
} else if (diff_algo == 2) {
zdiffAlgorithm2(src, setnum, dstzset, maxelelen);
} else if (diff_algo != 0) {
serverPanic("Unknown algorithm");
}
}
}
dictType setAccumulatorDictType = {
dictSdsHash, /* hash function */
NULL, /* key dup */
NULL, /* val dup */
dictSdsKeyCompare, /* key compare */
NULL, /* key destructor */
NULL, /* val destructor */
NULL /* allow to expand */
};
/* The zunionInterDiffGenericCommand() function is called in order to implement the
* following commands: ZUNION, ZINTER, ZDIFF, ZUNIONSTORE, ZINTERSTORE, ZDIFFSTORE.
*
* 'numkeysIndex' parameter position of key number. for ZUNION/ZINTER/ZDIFF command,
* this value is 1, for ZUNIONSTORE/ZINTERSTORE/ZDIFFSTORE command, this value is 2.
*
* 'op' SET_OP_INTER, SET_OP_UNION or SET_OP_DIFF.
*/
void zunionInterDiffGenericCommand(client *c, robj *dstkey, int numkeysIndex, int op) {
int i, j;
long setnum;
int aggregate = REDIS_AGGR_SUM;
zsetopsrc *src;
zsetopval zval;
sds tmp;
size_t maxelelen = 0;
robj *dstobj;
zset *dstzset;
zskiplistNode *znode;
int withscores = 0;
/* expect setnum input keys to be given */
if ((getLongFromObjectOrReply(c, c->argv[numkeysIndex], &setnum, NULL) != C_OK))
return;
if (setnum < 1) {
addReplyErrorFormat(c,
"at least 1 input key is needed for %s", c->cmd->name);
return;
}
/* test if the expected number of keys would overflow */
if (setnum > (c->argc-(numkeysIndex+1))) {
addReplyErrorObject(c,shared.syntaxerr);
return;
}
/* read keys to be used for input */
src = zcalloc(sizeof(zsetopsrc) * setnum);
for (i = 0, j = numkeysIndex+1; i < setnum; i++, j++) {
robj *obj = dstkey ?
lookupKeyWrite(c->db,c->argv[j]) :
lookupKeyRead(c->db,c->argv[j]);
if (obj != NULL) {
if (obj->type != OBJ_ZSET && obj->type != OBJ_SET) {
zfree(src);
addReplyErrorObject(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. */
src[i].weight = 1.0;
}
/* parse optional extra arguments */
if (j < c->argc) {
int remaining = c->argc - j;
while (remaining) {
if (op != SET_OP_DIFF &&
remaining >= (setnum + 1) &&
!strcasecmp(c->argv[j]->ptr,"weights"))
{
j++; remaining--;
for (i = 0; i < setnum; i++, j++, remaining--) {
if (getDoubleFromObjectOrReply(c,c->argv[j],&src[i].weight,
"weight value is not a float") != C_OK)
{
zfree(src);
return;
}
}
} else if (op != SET_OP_DIFF &&
remaining >= 2 &&
!strcasecmp(c->argv[j]->ptr,"aggregate"))
{
j++; remaining--;
if (!strcasecmp(c->argv[j]->ptr,"sum")) {
aggregate = REDIS_AGGR_SUM;
} else if (!strcasecmp(c->argv[j]->ptr,"min")) {
aggregate = REDIS_AGGR_MIN;
} else if (!strcasecmp(c->argv[j]->ptr,"max")) {
aggregate = REDIS_AGGR_MAX;
} else {
zfree(src);
addReplyErrorObject(c,shared.syntaxerr);
return;
}
j++; remaining--;
} else if (remaining >= 1 &&
!dstkey &&
!strcasecmp(c->argv[j]->ptr,"withscores"))
{
j++; remaining--;
withscores = 1;
} else {
zfree(src);
addReplyErrorObject(c,shared.syntaxerr);
return;
}
}
}
if (op != SET_OP_DIFF) {
/* sort sets from the smallest to largest, this will improve our
* algorithm's performance */
qsort(src,setnum,sizeof(zsetopsrc),zuiCompareByCardinality);
}
dstobj = createZsetObject();
dstzset = dstobj->ptr;
memset(&zval, 0, sizeof(zval));
if (op == SET_OP_INTER) {
/* 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. */
zuiInitIterator(&src[0]);
while (zuiNext(&src[0],&zval)) {
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. */
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 {
break;
}
}
/* Only continue when present in every input. */
if (j == setnum) {
tmp = zuiNewSdsFromValue(&zval);
znode = zslInsert(dstzset->zsl,score,tmp);
dictAdd(dstzset->dict,tmp,&znode->score);
if (sdslen(tmp) > maxelelen) maxelelen = sdslen(tmp);
}
}
zuiClearIterator(&src[0]);
}
} else if (op == SET_OP_UNION) {
dict *accumulator = dictCreate(&setAccumulatorDictType,NULL);
dictIterator *di;
dictEntry *de, *existing;
double score;
if (setnum) {
/* Our union is at least as large as the largest set.
* Resize the dictionary ASAP to avoid useless rehashing. */
dictExpand(accumulator,zuiLength(&src[setnum-1]));
}
/* Step 1: Create a dictionary of elements -> aggregated-scores
* by iterating one sorted set after the other. */
for (i = 0; i < setnum; i++) {
if (zuiLength(&src[i]) == 0) continue;
zuiInitIterator(&src[i]);
while (zuiNext(&src[i],&zval)) {
/* Initialize value */
score = src[i].weight * zval.score;
if (isnan(score)) score = 0;
/* Search for this element in the accumulating dictionary. */
de = dictAddRaw(accumulator,zuiSdsFromValue(&zval),&existing);
/* If we don't have it, we need to create a new entry. */
if (!existing) {
tmp = zuiNewSdsFromValue(&zval);
/* Remember the longest single element encountered,
* to understand if it's possible to convert to ziplist
* at the end. */
if (sdslen(tmp) > maxelelen) maxelelen = sdslen(tmp);
/* Update the element with its initial score. */
dictSetKey(accumulator, de, tmp);
dictSetDoubleVal(de,score);
} else {
/* Update the score with the score of the new instance
* of the element found in the current sorted set.
*
* Here we access directly the dictEntry double
* value inside the union as it is a big speedup
* compared to using the getDouble/setDouble API. */
zunionInterAggregate(&existing->v.d,score,aggregate);
}
}
zuiClearIterator(&src[i]);
}
/* Step 2: convert the dictionary into the final sorted set. */
di = dictGetIterator(accumulator);
/* We now are aware of the final size of the resulting sorted set,
* let's resize the dictionary embedded inside the sorted set to the
* right size, in order to save rehashing time. */
dictExpand(dstzset->dict,dictSize(accumulator));
while((de = dictNext(di)) != NULL) {
sds ele = dictGetKey(de);
score = dictGetDoubleVal(de);
znode = zslInsert(dstzset->zsl,score,ele);
dictAdd(dstzset->dict,ele,&znode->score);
}
dictReleaseIterator(di);
dictRelease(accumulator);
} else if (op == SET_OP_DIFF) {
zdiff(src, setnum, dstzset, &maxelelen);
} else {
serverPanic("Unknown operator");
}
if (dstkey) {
if (dstzset->zsl->length) {
zsetConvertToZiplistIfNeeded(dstobj, maxelelen);
setKey(c, c->db, dstkey, dstobj);
addReplyLongLong(c, zsetLength(dstobj));
notifyKeyspaceEvent(NOTIFY_ZSET,
(op == SET_OP_UNION) ? "zunionstore" :
(op == SET_OP_INTER ? "zinterstore" : "zdiffstore"),
dstkey, c->db->id);
server.dirty++;
} else {
addReply(c, shared.czero);
if (dbDelete(c->db, dstkey)) {
signalModifiedKey(c, c->db, dstkey);
notifyKeyspaceEvent(NOTIFY_GENERIC, "del", dstkey, c->db->id);
server.dirty++;
}
}
} else {
unsigned long length = dstzset->zsl->length;
zskiplist *zsl = dstzset->zsl;
zskiplistNode *zn = zsl->header->level[0].forward;
/* In case of WITHSCORES, respond with a single array in RESP2, and
* nested arrays in RESP3. We can't use a map response type since the
* client library needs to know to respect the order. */
if (withscores && c->resp == 2)
addReplyArrayLen(c, length*2);
else
addReplyArrayLen(c, length);
while (zn != NULL) {
if (withscores && c->resp > 2) addReplyArrayLen(c,2);
addReplyBulkCBuffer(c,zn->ele,sdslen(zn->ele));
if (withscores) addReplyDouble(c,zn->score);
zn = zn->level[0].forward;
}
}
decrRefCount(dstobj);
zfree(src);
}
void zunionstoreCommand(client *c) {
zunionInterDiffGenericCommand(c, c->argv[1], 2, SET_OP_UNION);
}
void zinterstoreCommand(client *c) {
zunionInterDiffGenericCommand(c, c->argv[1], 2, SET_OP_INTER);
}
void zdiffstoreCommand(client *c) {
zunionInterDiffGenericCommand(c, c->argv[1], 2, SET_OP_DIFF);
}
void zunionCommand(client *c) {
zunionInterDiffGenericCommand(c, NULL, 1, SET_OP_UNION);
}
void zinterCommand(client *c) {
zunionInterDiffGenericCommand(c, NULL, 1, SET_OP_INTER);
}
void zdiffCommand(client *c) {
zunionInterDiffGenericCommand(c, NULL, 1, SET_OP_DIFF);
}
typedef enum {
ZRANGE_DIRECTION_AUTO = 0,
ZRANGE_DIRECTION_FORWARD,
ZRANGE_DIRECTION_REVERSE
} zrange_direction;
typedef enum {
ZRANGE_CONSUMER_TYPE_CLIENT = 0,
ZRANGE_CONSUMER_TYPE_INTERNAL
} zrange_consumer_type;
typedef struct zrange_result_handler zrange_result_handler;
typedef void (*zrangeResultBeginFunction)(zrange_result_handler *c);
typedef void (*zrangeResultFinalizeFunction)(
zrange_result_handler *c, size_t result_count);
typedef void (*zrangeResultEmitCBufferFunction)(
zrange_result_handler *c, const void *p, size_t len, double score);
typedef void (*zrangeResultEmitLongLongFunction)(
zrange_result_handler *c, long long ll, double score);
void zrangeGenericCommand (zrange_result_handler *handler, int argc_start, int store,
zrange_type rangetype, zrange_direction direction);
/* Interface struct for ZRANGE/ZRANGESTORE generic implementation.
* There is one implementation of this interface that sends a RESP reply to clients.
* and one implementation that stores the range result into a zset object. */
struct zrange_result_handler {
zrange_consumer_type type;
client *client;
robj *dstkey;
robj *dstobj;
void *userdata;
int withscores;
int should_emit_array_length;
zrangeResultBeginFunction beginResultEmission;
zrangeResultFinalizeFunction finalizeResultEmission;
zrangeResultEmitCBufferFunction emitResultFromCBuffer;
zrangeResultEmitLongLongFunction emitResultFromLongLong;
};
/* Result handler methods for responding the ZRANGE to clients. */
static void zrangeResultBeginClient(zrange_result_handler *handler) {
handler->userdata = addReplyDeferredLen(handler->client);
}
static void zrangeResultEmitCBufferToClient(zrange_result_handler *handler,
const void *value, size_t value_length_in_bytes, double score)
{
if (handler->should_emit_array_length) {
addReplyArrayLen(handler->client, 2);
}
addReplyBulkCBuffer(handler->client, value, value_length_in_bytes);
if (handler->withscores) {
addReplyDouble(handler->client, score);
}
}
static void zrangeResultEmitLongLongToClient(zrange_result_handler *handler,
long long value, double score)
{
if (handler->should_emit_array_length) {
addReplyArrayLen(handler->client, 2);
}
addReplyBulkLongLong(handler->client, value);
if (handler->withscores) {
addReplyDouble(handler->client, score);
}
}
static void zrangeResultFinalizeClient(zrange_result_handler *handler,
size_t result_count)
{
/* In case of WITHSCORES, respond with a single array in RESP2, and
* nested arrays in RESP3. We can't use a map response type since the
* client library needs to know to respect the order. */
if (handler->withscores && (handler->client->resp == 2)) {
result_count *= 2;
}
setDeferredArrayLen(handler->client, handler->userdata, result_count);
}
/* Result handler methods for storing the ZRANGESTORE to a zset. */
static void zrangeResultBeginStore(zrange_result_handler *handler)
{
handler->dstobj = createZsetZiplistObject();
}
static void zrangeResultEmitCBufferForStore(zrange_result_handler *handler,
const void *value, size_t value_length_in_bytes, double score)
{
double newscore;
int retflags = 0;
sds ele = sdsnewlen(value, value_length_in_bytes);
int retval = zsetAdd(handler->dstobj, score, ele, ZADD_IN_NONE, &retflags, &newscore);
sdsfree(ele);
serverAssert(retval);
}
static void zrangeResultEmitLongLongForStore(zrange_result_handler *handler,
long long value, double score)
{
double newscore;
int retflags = 0;
sds ele = sdsfromlonglong(value);
int retval = zsetAdd(handler->dstobj, score, ele, ZADD_IN_NONE, &retflags, &newscore);
sdsfree(ele);
serverAssert(retval);
}
static void zrangeResultFinalizeStore(zrange_result_handler *handler, size_t result_count)
{
if (result_count) {
setKey(handler->client, handler->client->db, handler->dstkey, handler->dstobj);
addReplyLongLong(handler->client, result_count);
notifyKeyspaceEvent(NOTIFY_ZSET, "zrangestore", handler->dstkey, handler->client->db->id);
server.dirty++;
} else {
addReply(handler->client, shared.czero);
if (dbDelete(handler->client->db, handler->dstkey)) {
signalModifiedKey(handler->client, handler->client->db, handler->dstkey);
notifyKeyspaceEvent(NOTIFY_GENERIC, "del", handler->dstkey, handler->client->db->id);
server.dirty++;
}
}
decrRefCount(handler->dstobj);
}
/* Initialize the consumer interface type with the requested type. */
static void zrangeResultHandlerInit(zrange_result_handler *handler,
client *client, zrange_consumer_type type)
{
memset(handler, 0, sizeof(*handler));
handler->client = client;
switch (type) {
case ZRANGE_CONSUMER_TYPE_CLIENT:
handler->beginResultEmission = zrangeResultBeginClient;
handler->finalizeResultEmission = zrangeResultFinalizeClient;
handler->emitResultFromCBuffer = zrangeResultEmitCBufferToClient;
handler->emitResultFromLongLong = zrangeResultEmitLongLongToClient;
break;
case ZRANGE_CONSUMER_TYPE_INTERNAL:
handler->beginResultEmission = zrangeResultBeginStore;
handler->finalizeResultEmission = zrangeResultFinalizeStore;
handler->emitResultFromCBuffer = zrangeResultEmitCBufferForStore;
handler->emitResultFromLongLong = zrangeResultEmitLongLongForStore;
break;
}
}
static void zrangeResultHandlerScoreEmissionEnable(zrange_result_handler *handler) {
handler->withscores = 1;
handler->should_emit_array_length = (handler->client->resp > 2);
}
static void zrangeResultHandlerDestinationKeySet (zrange_result_handler *handler,
robj *dstkey)
{
handler->dstkey = dstkey;
}
/* This command implements ZRANGE, ZREVRANGE. */
void genericZrangebyrankCommand(zrange_result_handler *handler,
robj *zobj, long start, long end, int withscores, int reverse) {
client *c = handler->client;
long llen;
long rangelen;
size_t result_cardinality;
/* Sanitize indexes. */
llen = zsetLength(zobj);
if (start < 0) start = llen+start;
if (end < 0) end = llen+end;
if (start < 0) start = 0;
handler->beginResultEmission(handler);
/* Invariant: start >= 0, so this test will be true when end < 0.
* The range is empty when start > end or start >= length. */
if (start > end || start >= llen) {
handler->finalizeResultEmission(handler, 0);
return;
}
if (end >= llen) end = llen-1;
rangelen = (end-start)+1;
result_cardinality = rangelen;
if (zobj->encoding == OBJ_ENCODING_ZIPLIST) {
unsigned char *zl = zobj->ptr;
unsigned char *eptr, *sptr;
unsigned char *vstr;
unsigned int vlen;
long long vlong;
double score = 0.0;
if (reverse)
eptr = ziplistIndex(zl,-2-(2*start));
else
eptr = ziplistIndex(zl,2*start);
serverAssertWithInfo(c,zobj,eptr != NULL);
sptr = ziplistNext(zl,eptr);
while (rangelen--) {
serverAssertWithInfo(c,zobj,eptr != NULL && sptr != NULL);
serverAssertWithInfo(c,zobj,ziplistGet(eptr,&vstr,&vlen,&vlong));
if (withscores) /* don't bother to extract the score if it's gonna be ignored. */
score = zzlGetScore(sptr);
if (vstr == NULL) {
handler->emitResultFromLongLong(handler, vlong, score);
} else {
handler->emitResultFromCBuffer(handler, vstr, vlen, score);
}
if (reverse)
zzlPrev(zl,&eptr,&sptr);
else
zzlNext(zl,&eptr,&sptr);
}
} else if (zobj->encoding == OBJ_ENCODING_SKIPLIST) {
zset *zs = zobj->ptr;
zskiplist *zsl = zs->zsl;
zskiplistNode *ln;
/* 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--) {
serverAssertWithInfo(c,zobj,ln != NULL);
sds ele = ln->ele;
handler->emitResultFromCBuffer(handler, ele, sdslen(ele), ln->score);
ln = reverse ? ln->backward : ln->level[0].forward;
}
} else {
serverPanic("Unknown sorted set encoding");
}
handler->finalizeResultEmission(handler, result_cardinality);
}
/* ZRANGESTORE <dst> <src> <min> <max> [BYSCORE | BYLEX] [REV] [LIMIT offset count] */
void zrangestoreCommand (client *c) {
robj *dstkey = c->argv[1];
zrange_result_handler handler;
zrangeResultHandlerInit(&handler, c, ZRANGE_CONSUMER_TYPE_INTERNAL);
zrangeResultHandlerDestinationKeySet(&handler, dstkey);
zrangeGenericCommand(&handler, 2, 1, ZRANGE_AUTO, ZRANGE_DIRECTION_AUTO);
}
/* ZRANGE <key> <min> <max> [BYSCORE | BYLEX] [REV] [WITHSCORES] [LIMIT offset count] */
void zrangeCommand(client *c) {
zrange_result_handler handler;
zrangeResultHandlerInit(&handler, c, ZRANGE_CONSUMER_TYPE_CLIENT);
zrangeGenericCommand(&handler, 1, 0, ZRANGE_AUTO, ZRANGE_DIRECTION_AUTO);
}
/* ZREVRANGE <key> <min> <max> [WITHSCORES] */
void zrevrangeCommand(client *c) {
zrange_result_handler handler;
zrangeResultHandlerInit(&handler, c, ZRANGE_CONSUMER_TYPE_CLIENT);
zrangeGenericCommand(&handler, 1, 0, ZRANGE_RANK, ZRANGE_DIRECTION_REVERSE);
}
/* This command implements ZRANGEBYSCORE, ZREVRANGEBYSCORE. */
void genericZrangebyscoreCommand(zrange_result_handler *handler,
zrangespec *range, robj *zobj, long offset, long limit,
int reverse) {
client *c = handler->client;
unsigned long rangelen = 0;
handler->beginResultEmission(handler);
/* For invalid offset, return directly. */
if (offset > 0 && offset >= (long)zsetLength(zobj)) {
handler->finalizeResultEmission(handler, 0);
return;
}
if (zobj->encoding == OBJ_ENCODING_ZIPLIST) {
unsigned char *zl = zobj->ptr;
unsigned char *eptr, *sptr;
unsigned char *vstr;
unsigned int vlen;
long long vlong;
/* If reversed, get the last node in range as starting point. */
if (reverse) {
eptr = zzlLastInRange(zl,range);
} else {
eptr = zzlFirstInRange(zl,range);
}
/* Get score pointer for the first element. */
if (eptr)
sptr = ziplistNext(zl,eptr);
/* 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--) {
double score = zzlGetScore(sptr);
/* Abort when the node is no longer in range. */
if (reverse) {
if (!zslValueGteMin(score,range)) break;
} else {
if (!zslValueLteMax(score,range)) break;
}
/* We know the element exists, so ziplistGet should always
* succeed */
serverAssertWithInfo(c,zobj,ziplistGet(eptr,&vstr,&vlen,&vlong));
rangelen++;
if (vstr == NULL) {
handler->emitResultFromLongLong(handler, vlong, score);
} else {
handler->emitResultFromCBuffer(handler, vstr, vlen, score);
}
/* Move to next node */
if (reverse) {
zzlPrev(zl,&eptr,&sptr);
} else {
zzlNext(zl,&eptr,&sptr);
}
}
} else if (zobj->encoding == OBJ_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);
}
/* 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--) {
/* Abort when the node is no longer in range. */
if (reverse) {
if (!zslValueGteMin(ln->score,range)) break;
} else {
if (!zslValueLteMax(ln->score,range)) break;
}
rangelen++;
handler->emitResultFromCBuffer(handler, ln->ele, sdslen(ln->ele), ln->score);
/* Move to next node */
if (reverse) {
ln = ln->backward;
} else {
ln = ln->level[0].forward;
}
}
} else {
serverPanic("Unknown sorted set encoding");
}
handler->finalizeResultEmission(handler, rangelen);
}
/* ZRANGEBYSCORE <key> <min> <max> [WITHSCORES] [LIMIT offset count] */
void zrangebyscoreCommand(client *c) {
zrange_result_handler handler;
zrangeResultHandlerInit(&handler, c, ZRANGE_CONSUMER_TYPE_CLIENT);
zrangeGenericCommand(&handler, 1, 0, ZRANGE_SCORE, ZRANGE_DIRECTION_FORWARD);
}
/* ZREVRANGEBYSCORE <key> <min> <max> [WITHSCORES] [LIMIT offset count] */
void zrevrangebyscoreCommand(client *c) {
zrange_result_handler handler;
zrangeResultHandlerInit(&handler, c, ZRANGE_CONSUMER_TYPE_CLIENT);
zrangeGenericCommand(&handler, 1, 0, ZRANGE_SCORE, ZRANGE_DIRECTION_REVERSE);
}
void zcountCommand(client *c) {
robj *key = c->argv[1];
robj *zobj;
zrangespec range;
unsigned long count = 0;
/* Parse the range arguments */
if (zslParseRange(c->argv[2],c->argv[3],&range) != C_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, OBJ_ZSET)) return;
if (zobj->encoding == OBJ_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);
serverAssertWithInfo(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 == OBJ_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->ele);
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->ele);
count -= (zsl->length - rank);
}
}
} else {
serverPanic("Unknown sorted set encoding");
}
addReplyLongLong(c, count);
}
void zlexcountCommand(client *c) {
robj *key = c->argv[1];
robj *zobj;
zlexrangespec range;
unsigned long count = 0;
/* Parse the range arguments */
if (zslParseLexRange(c->argv[2],c->argv[3],&range) != C_OK) {
addReplyError(c,"min or max not valid string range item");
return;
}
/* Lookup the sorted set */
if ((zobj = lookupKeyReadOrReply(c, key, shared.czero)) == NULL ||
checkType(c, zobj, OBJ_ZSET))
{
zslFreeLexRange(&range);
return;
}
if (zobj->encoding == OBJ_ENCODING_ZIPLIST) {
unsigned char *zl = zobj->ptr;
unsigned char *eptr, *sptr;
/* Use the first element in range as the starting point */
eptr = zzlFirstInLexRange(zl,&range);
/* No "first" element */
if (eptr == NULL) {
zslFreeLexRange(&range);
addReply(c, shared.czero);
return;
}
/* First element is in range */
sptr = ziplistNext(zl,eptr);
serverAssertWithInfo(c,zobj,zzlLexValueLteMax(eptr,&range));
/* Iterate over elements in range */
while (eptr) {
/* Abort when the node is no longer in range. */
if (!zzlLexValueLteMax(eptr,&range)) {
break;
} else {
count++;
zzlNext(zl,&eptr,&sptr);
}
}
} else if (zobj->encoding == OBJ_ENCODING_SKIPLIST) {
zset *zs = zobj->ptr;
zskiplist *zsl = zs->zsl;
zskiplistNode *zn;
unsigned long rank;
/* Find first element in range */
zn = zslFirstInLexRange(zsl, &range);
/* Use rank of first element, if any, to determine preliminary count */
if (zn != NULL) {
rank = zslGetRank(zsl, zn->score, zn->ele);
count = (zsl->length - (rank - 1));
/* Find last element in range */
zn = zslLastInLexRange(zsl, &range);
/* Use rank of last element, if any, to determine the actual count */
if (zn != NULL) {
rank = zslGetRank(zsl, zn->score, zn->ele);
count -= (zsl->length - rank);
}
}
} else {
serverPanic("Unknown sorted set encoding");
}
zslFreeLexRange(&range);
addReplyLongLong(c, count);
}
/* This command implements ZRANGEBYLEX, ZREVRANGEBYLEX. */
void genericZrangebylexCommand(zrange_result_handler *handler,
zlexrangespec *range, robj *zobj, int withscores, long offset, long limit,
int reverse)
{
client *c = handler->client;
unsigned long rangelen = 0;
handler->beginResultEmission(handler);
if (zobj->encoding == OBJ_ENCODING_ZIPLIST) {
unsigned char *zl = zobj->ptr;
unsigned char *eptr, *sptr;
unsigned char *vstr;
unsigned int vlen;
long long vlong;
/* If reversed, get the last node in range as starting point. */
if (reverse) {
eptr = zzlLastInLexRange(zl,range);
} else {
eptr = zzlFirstInLexRange(zl,range);
}
/* Get score pointer for the first element. */
if (eptr)
sptr = ziplistNext(zl,eptr);
/* 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--) {
double score = 0;
if (withscores) /* don't bother to extract the score if it's gonna be ignored. */
score = zzlGetScore(sptr);
/* Abort when the node is no longer in range. */
if (reverse) {
if (!zzlLexValueGteMin(eptr,range)) break;
} else {
if (!zzlLexValueLteMax(eptr,range)) break;
}
/* We know the element exists, so ziplistGet should always
* succeed. */
serverAssertWithInfo(c,zobj,ziplistGet(eptr,&vstr,&vlen,&vlong));
rangelen++;
if (vstr == NULL) {
handler->emitResultFromLongLong(handler, vlong, score);
} else {
handler->emitResultFromCBuffer(handler, vstr, vlen, score);
}
/* Move to next node */
if (reverse) {
zzlPrev(zl,&eptr,&sptr);
} else {
zzlNext(zl,&eptr,&sptr);
}
}
} else if (zobj->encoding == OBJ_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 = zslLastInLexRange(zsl,range);
} else {
ln = zslFirstInLexRange(zsl,range);
}
/* 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--) {
/* Abort when the node is no longer in range. */
if (reverse) {
if (!zslLexValueGteMin(ln->ele,range)) break;
} else {
if (!zslLexValueLteMax(ln->ele,range)) break;
}
rangelen++;
handler->emitResultFromCBuffer(handler, ln->ele, sdslen(ln->ele), ln->score);
/* Move to next node */
if (reverse) {
ln = ln->backward;
} else {
ln = ln->level[0].forward;
}
}
} else {
serverPanic("Unknown sorted set encoding");
}
handler->finalizeResultEmission(handler, rangelen);
}
/* ZRANGEBYLEX <key> <min> <max> [LIMIT offset count] */
void zrangebylexCommand(client *c) {
zrange_result_handler handler;
zrangeResultHandlerInit(&handler, c, ZRANGE_CONSUMER_TYPE_CLIENT);
zrangeGenericCommand(&handler, 1, 0, ZRANGE_LEX, ZRANGE_DIRECTION_FORWARD);
}
/* ZREVRANGEBYLEX <key> <min> <max> [LIMIT offset count] */
void zrevrangebylexCommand(client *c) {
zrange_result_handler handler;
zrangeResultHandlerInit(&handler, c, ZRANGE_CONSUMER_TYPE_CLIENT);
zrangeGenericCommand(&handler, 1, 0, ZRANGE_LEX, ZRANGE_DIRECTION_REVERSE);
}
/**
* This function handles ZRANGE and ZRANGESTORE, and also the deprecated
* Z[REV]RANGE[BYPOS|BYLEX] commands.
*
* The simple ZRANGE and ZRANGESTORE can take _AUTO in rangetype and direction,
* other command pass explicit value.
*
* The argc_start points to the src key argument, so following syntax is like:
* <src> <min> <max> [BYSCORE | BYLEX] [REV] [WITHSCORES] [LIMIT offset count]
*/
void zrangeGenericCommand(zrange_result_handler *handler, int argc_start, int store,
zrange_type rangetype, zrange_direction direction)
{
client *c = handler->client;
robj *key = c->argv[argc_start];
robj *zobj;
zrangespec range;
zlexrangespec lexrange;
int minidx = argc_start + 1;
int maxidx = argc_start + 2;
/* Options common to all */
long opt_start = 0;
long opt_end = 0;
int opt_withscores = 0;
long opt_offset = 0;
long opt_limit = -1;
/* Step 1: Skip the <src> <min> <max> args and parse remaining optional arguments. */
for (int j=argc_start + 3; j < c->argc; j++) {
int leftargs = c->argc-j-1;
if (!store && !strcasecmp(c->argv[j]->ptr,"withscores")) {
opt_withscores = 1;
} else if (!strcasecmp(c->argv[j]->ptr,"limit") && leftargs >= 2) {
if ((getLongFromObjectOrReply(c, c->argv[j+1], &opt_offset, NULL) != C_OK) ||
(getLongFromObjectOrReply(c, c->argv[j+2], &opt_limit, NULL) != C_OK))
{
return;
}
j += 2;
} else if (direction == ZRANGE_DIRECTION_AUTO &&
!strcasecmp(c->argv[j]->ptr,"rev"))
{
direction = ZRANGE_DIRECTION_REVERSE;
} else if (rangetype == ZRANGE_AUTO &&
!strcasecmp(c->argv[j]->ptr,"bylex"))
{
rangetype = ZRANGE_LEX;
} else if (rangetype == ZRANGE_AUTO &&
!strcasecmp(c->argv[j]->ptr,"byscore"))
{
rangetype = ZRANGE_SCORE;
} else {
addReplyErrorObject(c,shared.syntaxerr);
return;
}
}
/* Use defaults if not overridden by arguments. */
if (direction == ZRANGE_DIRECTION_AUTO)
direction = ZRANGE_DIRECTION_FORWARD;
if (rangetype == ZRANGE_AUTO)
rangetype = ZRANGE_RANK;
/* Check for conflicting arguments. */
if (opt_limit != -1 && rangetype == ZRANGE_RANK) {
addReplyError(c,"syntax error, LIMIT is only supported in combination with either BYSCORE or BYLEX");
return;
}
if (opt_withscores && rangetype == ZRANGE_LEX) {
addReplyError(c,"syntax error, WITHSCORES not supported in combination with BYLEX");
return;
}
if (direction == ZRANGE_DIRECTION_REVERSE &&
((ZRANGE_SCORE == rangetype) || (ZRANGE_LEX == rangetype)))
{
/* Range is given as [max,min] */
int tmp = maxidx;
maxidx = minidx;
minidx = tmp;
}
/* Step 2: Parse the range. */
switch (rangetype) {
case ZRANGE_AUTO:
case ZRANGE_RANK:
/* Z[REV]RANGE, ZRANGESTORE [REV]RANGE */
if ((getLongFromObjectOrReply(c, c->argv[minidx], &opt_start,NULL) != C_OK) ||
(getLongFromObjectOrReply(c, c->argv[maxidx], &opt_end,NULL) != C_OK))
{
return;
}
break;
case ZRANGE_SCORE:
/* Z[REV]RANGEBYSCORE, ZRANGESTORE [REV]RANGEBYSCORE */
if (zslParseRange(c->argv[minidx], c->argv[maxidx], &range) != C_OK) {
addReplyError(c, "min or max is not a float");
return;
}
break;
case ZRANGE_LEX:
/* Z[REV]RANGEBYLEX, ZRANGESTORE [REV]RANGEBYLEX */
if (zslParseLexRange(c->argv[minidx], c->argv[maxidx], &lexrange) != C_OK) {
addReplyError(c, "min or max not valid string range item");
return;
}
break;
}
if (opt_withscores || store) {
zrangeResultHandlerScoreEmissionEnable(handler);
}
/* Step 3: Lookup the key and get the range. */
zobj = handler->dstkey ?
lookupKeyWrite(c->db,key) :
lookupKeyRead(c->db,key);
if (zobj == NULL) {
if (store) {
handler->beginResultEmission(handler);
handler->finalizeResultEmission(handler, 0);
} else {
addReply(c, shared.emptyarray);
}
goto cleanup;
}
if (checkType(c,zobj,OBJ_ZSET)) goto cleanup;
/* Step 4: Pass this to the command-specific handler. */
switch (rangetype) {
case ZRANGE_AUTO:
case ZRANGE_RANK:
genericZrangebyrankCommand(handler, zobj, opt_start, opt_end,
opt_withscores || store, direction == ZRANGE_DIRECTION_REVERSE);
break;
case ZRANGE_SCORE:
genericZrangebyscoreCommand(handler, &range, zobj, opt_offset,
opt_limit, direction == ZRANGE_DIRECTION_REVERSE);
break;
case ZRANGE_LEX:
genericZrangebylexCommand(handler, &lexrange, zobj, opt_withscores || store,
opt_offset, opt_limit, direction == ZRANGE_DIRECTION_REVERSE);
break;
}
/* Instead of returning here, we'll just fall-through the clean-up. */
cleanup:
if (rangetype == ZRANGE_LEX) {
zslFreeLexRange(&lexrange);
}
}
void zcardCommand(client *c) {
robj *key = c->argv[1];
robj *zobj;
if ((zobj = lookupKeyReadOrReply(c,key,shared.czero)) == NULL ||
checkType(c,zobj,OBJ_ZSET)) return;
addReplyLongLong(c,zsetLength(zobj));
}
void zscoreCommand(client *c) {
robj *key = c->argv[1];
robj *zobj;
double score;
if ((zobj = lookupKeyReadOrReply(c,key,shared.null[c->resp])) == NULL ||
checkType(c,zobj,OBJ_ZSET)) return;
if (zsetScore(zobj,c->argv[2]->ptr,&score) == C_ERR) {
addReplyNull(c);
} else {
addReplyDouble(c,score);
}
}
void zmscoreCommand(client *c) {
robj *key = c->argv[1];
robj *zobj;
double score;
zobj = lookupKeyRead(c->db,key);
if (checkType(c,zobj,OBJ_ZSET)) return;
addReplyArrayLen(c,c->argc - 2);
for (int j = 2; j < c->argc; j++) {
/* Treat a missing set the same way as an empty set */
if (zobj == NULL || zsetScore(zobj,c->argv[j]->ptr,&score) == C_ERR) {
addReplyNull(c);
} else {
addReplyDouble(c,score);
}
}
}
void zrankGenericCommand(client *c, int reverse) {
robj *key = c->argv[1];
robj *ele = c->argv[2];
robj *zobj;
long rank;
if ((zobj = lookupKeyReadOrReply(c,key,shared.null[c->resp])) == NULL ||
checkType(c,zobj,OBJ_ZSET)) return;
serverAssertWithInfo(c,ele,sdsEncodedObject(ele));
rank = zsetRank(zobj,ele->ptr,reverse);
if (rank >= 0) {
addReplyLongLong(c,rank);
} else {
addReplyNull(c);
}
}
void zrankCommand(client *c) {
zrankGenericCommand(c, 0);
}
void zrevrankCommand(client *c) {
zrankGenericCommand(c, 1);
}
void zscanCommand(client *c) {
robj *o;
unsigned long cursor;
if (parseScanCursorOrReply(c,c->argv[2],&cursor) == C_ERR) return;
if ((o = lookupKeyReadOrReply(c,c->argv[1],shared.emptyscan)) == NULL ||
checkType(c,o,OBJ_ZSET)) return;
scanGenericCommand(c,o,cursor);
}
/* This command implements the generic zpop operation, used by:
* ZPOPMIN, ZPOPMAX, BZPOPMIN and BZPOPMAX. This function is also used
* inside blocked.c in the unblocking stage of BZPOPMIN and BZPOPMAX.
*
* If 'emitkey' is true also the key name is emitted, useful for the blocking
* behavior of BZPOP[MIN|MAX], since we can block into multiple keys.
*
* The synchronous version instead does not need to emit the key, but may
* use the 'count' argument to return multiple items if available. */
void genericZpopCommand(client *c, robj **keyv, int keyc, int where, int emitkey, robj *countarg) {
int idx;
robj *key = NULL;
robj *zobj = NULL;
sds ele;
double score;
long count = 1;
/* If a count argument as passed, parse it or return an error. */
if (countarg) {
if (getLongFromObjectOrReply(c,countarg,&count,NULL) != C_OK)
return;
if (count <= 0) {
addReply(c,shared.emptyarray);
return;
}
}
/* Check type and break on the first error, otherwise identify candidate. */
idx = 0;
while (idx < keyc) {
key = keyv[idx++];
zobj = lookupKeyWrite(c->db,key);
if (!zobj) continue;
if (checkType(c,zobj,OBJ_ZSET)) return;
break;
}
/* No candidate for zpopping, return empty. */
if (!zobj) {
addReply(c,shared.emptyarray);
return;
}
void *arraylen_ptr = addReplyDeferredLen(c);
long result_count = 0;
/* We emit the key only for the blocking variant. */
if (emitkey) addReplyBulk(c,key);
/* Respond with a single (flat) array in RESP2 or if countarg is not
* provided (returning a single element). In RESP3, when countarg is
* provided, use nested array. */
int use_nested_array = c->resp > 2 && countarg != NULL;
/* Remove the element. */
do {
if (zobj->encoding == OBJ_ENCODING_ZIPLIST) {
unsigned char *zl = zobj->ptr;
unsigned char *eptr, *sptr;
unsigned char *vstr;
unsigned int vlen;
long long vlong;
/* Get the first or last element in the sorted set. */
eptr = ziplistIndex(zl,where == ZSET_MAX ? -2 : 0);
serverAssertWithInfo(c,zobj,eptr != NULL);
serverAssertWithInfo(c,zobj,ziplistGet(eptr,&vstr,&vlen,&vlong));
if (vstr == NULL)
ele = sdsfromlonglong(vlong);
else
ele = sdsnewlen(vstr,vlen);
/* Get the score. */
sptr = ziplistNext(zl,eptr);
serverAssertWithInfo(c,zobj,sptr != NULL);
score = zzlGetScore(sptr);
} else if (zobj->encoding == OBJ_ENCODING_SKIPLIST) {
zset *zs = zobj->ptr;
zskiplist *zsl = zs->zsl;
zskiplistNode *zln;
/* Get the first or last element in the sorted set. */
zln = (where == ZSET_MAX ? zsl->tail :
zsl->header->level[0].forward);
/* There must be an element in the sorted set. */
serverAssertWithInfo(c,zobj,zln != NULL);
ele = sdsdup(zln->ele);
score = zln->score;
} else {
serverPanic("Unknown sorted set encoding");
}
serverAssertWithInfo(c,zobj,zsetDel(zobj,ele));
server.dirty++;
if (result_count == 0) { /* Do this only for the first iteration. */
char *events[2] = {"zpopmin","zpopmax"};
notifyKeyspaceEvent(NOTIFY_ZSET,events[where],key,c->db->id);
signalModifiedKey(c,c->db,key);
}
if (use_nested_array) {
addReplyArrayLen(c,2);
}
addReplyBulkCBuffer(c,ele,sdslen(ele));
addReplyDouble(c,score);
sdsfree(ele);
++result_count;
/* Remove the key, if indeed needed. */
if (zsetLength(zobj) == 0) {
dbDelete(c->db,key);
notifyKeyspaceEvent(NOTIFY_GENERIC,"del",key,c->db->id);
break;
}
} while(--count);
if (!use_nested_array) {
result_count *= 2;
}
setDeferredArrayLen(c,arraylen_ptr,result_count + (emitkey != 0));
}
/* ZPOPMIN key [<count>] */
void zpopminCommand(client *c) {
if (c->argc > 3) {
addReplyErrorObject(c,shared.syntaxerr);
return;
}
genericZpopCommand(c,&c->argv[1],1,ZSET_MIN,0,
c->argc == 3 ? c->argv[2] : NULL);
}
/* ZMAXPOP key [<count>] */
void zpopmaxCommand(client *c) {
if (c->argc > 3) {
addReplyErrorObject(c,shared.syntaxerr);
return;
}
genericZpopCommand(c,&c->argv[1],1,ZSET_MAX,0,
c->argc == 3 ? c->argv[2] : NULL);
}
/* BZPOPMIN / BZPOPMAX actual implementation. */
void blockingGenericZpopCommand(client *c, int where) {
robj *o;
mstime_t timeout;
int j;
if (getTimeoutFromObjectOrReply(c,c->argv[c->argc-1],&timeout,UNIT_SECONDS)
!= C_OK) return;
for (j = 1; j < c->argc-1; j++) {
o = lookupKeyWrite(c->db,c->argv[j]);
if (checkType(c,o,OBJ_ZSET)) return;
if (o != NULL) {
if (zsetLength(o) != 0) {
/* Non empty zset, this is like a normal ZPOP[MIN|MAX]. */
genericZpopCommand(c,&c->argv[j],1,where,1,NULL);
/* Replicate it as an ZPOP[MIN|MAX] instead of BZPOP[MIN|MAX]. */
rewriteClientCommandVector(c,2,
where == ZSET_MAX ? shared.zpopmax : shared.zpopmin,
c->argv[j]);
return;
}
}
}
/* If we are not allowed to block the client and the zset is empty the only thing
* we can do is treating it as a timeout (even with timeout 0). */
if (c->flags & CLIENT_DENY_BLOCKING) {
addReplyNullArray(c);
return;
}
/* If the keys do not exist we must block */
blockForKeys(c,BLOCKED_ZSET,c->argv + 1,c->argc - 2,timeout,NULL,NULL,NULL);
}
// BZPOPMIN key [key ...] timeout
void bzpopminCommand(client *c) {
blockingGenericZpopCommand(c,ZSET_MIN);
}
// BZPOPMAX key [key ...] timeout
void bzpopmaxCommand(client *c) {
blockingGenericZpopCommand(c,ZSET_MAX);
}
static void zarndmemberReplyWithZiplist(client *c, unsigned int count, ziplistEntry *keys, ziplistEntry *vals) {
for (unsigned long i = 0; i < count; i++) {
if (vals && c->resp > 2)
addReplyArrayLen(c,2);
if (keys[i].sval)
addReplyBulkCBuffer(c, keys[i].sval, keys[i].slen);
else
addReplyBulkLongLong(c, keys[i].lval);
if (vals) {
if (vals[i].sval) {
addReplyDouble(c, zzlStrtod(vals[i].sval,vals[i].slen));
} else
addReplyDouble(c, vals[i].lval);
}
}
}
/* How many times bigger should be the zset compared to the requested size
* for us to not use the "remove elements" strategy? Read later in the
* implementation for more info. */
#define ZRANDMEMBER_SUB_STRATEGY_MUL 3
/* If client is trying to ask for a very large number of random elements,
* queuing may consume an unlimited amount of memory, so we want to limit
* the number of randoms per time. */
#define ZRANDMEMBER_RANDOM_SAMPLE_LIMIT 1000
void zrandmemberWithCountCommand(client *c, long l, int withscores) {
unsigned long count, size;
int uniq = 1;
robj *zsetobj;
if ((zsetobj = lookupKeyReadOrReply(c, c->argv[1], shared.emptyarray))
== NULL || checkType(c, zsetobj, OBJ_ZSET)) return;
size = zsetLength(zsetobj);
if(l >= 0) {
count = (unsigned long) l;
} else {
count = -l;
uniq = 0;
}
/* If count is zero, serve it ASAP to avoid special cases later. */
if (count == 0) {
addReply(c,shared.emptyarray);
return;
}
/* CASE 1: The count was negative, so the extraction method is just:
* "return N random elements" sampling the whole set every time.
* This case is trivial and can be served without auxiliary data
* structures. This case is the only one that also needs to return the
* elements in random order. */
if (!uniq || count == 1) {
if (withscores && c->resp == 2)
addReplyArrayLen(c, count*2);
else
addReplyArrayLen(c, count);
if (zsetobj->encoding == OBJ_ENCODING_SKIPLIST) {
zset *zs = zsetobj->ptr;
while (count--) {
dictEntry *de = dictGetFairRandomKey(zs->dict);
sds key = dictGetKey(de);
if (withscores && c->resp > 2)
addReplyArrayLen(c,2);
addReplyBulkCBuffer(c, key, sdslen(key));
if (withscores)
addReplyDouble(c, *(double*)dictGetVal(de));
}
} else if (zsetobj->encoding == OBJ_ENCODING_ZIPLIST) {
ziplistEntry *keys, *vals = NULL;
unsigned long limit, sample_count;
limit = count > ZRANDMEMBER_RANDOM_SAMPLE_LIMIT ? ZRANDMEMBER_RANDOM_SAMPLE_LIMIT : count;
keys = zmalloc(sizeof(ziplistEntry)*limit);
if (withscores)
vals = zmalloc(sizeof(ziplistEntry)*limit);
while (count) {
sample_count = count > limit ? limit : count;
count -= sample_count;
ziplistRandomPairs(zsetobj->ptr, sample_count, keys, vals);
zarndmemberReplyWithZiplist(c, sample_count, keys, vals);
}
zfree(keys);
zfree(vals);
}
return;
}
zsetopsrc src;
zsetopval zval;
src.subject = zsetobj;
src.type = zsetobj->type;
src.encoding = zsetobj->encoding;
zuiInitIterator(&src);
memset(&zval, 0, sizeof(zval));
/* Initiate reply count, RESP3 responds with nested array, RESP2 with flat one. */
long reply_size = count < size ? count : size;
if (withscores && c->resp == 2)
addReplyArrayLen(c, reply_size*2);
else
addReplyArrayLen(c, reply_size);
/* CASE 2:
* The number of requested elements is greater than the number of
* elements inside the zset: simply return the whole zset. */
if (count >= size) {
while (zuiNext(&src, &zval)) {
if (withscores && c->resp > 2)
addReplyArrayLen(c,2);
addReplyBulkSds(c, zuiNewSdsFromValue(&zval));
if (withscores)
addReplyDouble(c, zval.score);
}
zuiClearIterator(&src);
return;
}
/* CASE 3:
* The number of elements inside the zset is not greater than
* ZRANDMEMBER_SUB_STRATEGY_MUL times the number of requested elements.
* In this case we create a dict from scratch with all the elements, and
* subtract random elements to reach the requested number of elements.
*
* This is done because if the number of requested elements is just
* a bit less than the number of elements in the set, the natural approach
* used into CASE 4 is highly inefficient. */
if (count*ZRANDMEMBER_SUB_STRATEGY_MUL > size) {
dict *d = dictCreate(&sdsReplyDictType, NULL);
dictExpand(d, size);
/* Add all the elements into the temporary dictionary. */
while (zuiNext(&src, &zval)) {
sds key = zuiNewSdsFromValue(&zval);
dictEntry *de = dictAddRaw(d, key, NULL);
serverAssert(de);
if (withscores)
dictSetDoubleVal(de, zval.score);
}
serverAssert(dictSize(d) == size);
/* Remove random elements to reach the right count. */
while (size > count) {
dictEntry *de;
de = dictGetRandomKey(d);
dictUnlink(d,dictGetKey(de));
sdsfree(dictGetKey(de));
dictFreeUnlinkedEntry(d,de);
size--;
}
/* Reply with what's in the dict and release memory */
dictIterator *di;
dictEntry *de;
di = dictGetIterator(d);
while ((de = dictNext(di)) != NULL) {
if (withscores && c->resp > 2)
addReplyArrayLen(c,2);
addReplyBulkSds(c, dictGetKey(de));
if (withscores)
addReplyDouble(c, dictGetDoubleVal(de));
}
dictReleaseIterator(di);
dictRelease(d);
}
/* CASE 4: We have a big zset compared to the requested number of elements.
* In this case we can simply get random elements from the zset and add
* to the temporary set, trying to eventually get enough unique elements
* to reach the specified count. */
else {
if (zsetobj->encoding == OBJ_ENCODING_ZIPLIST) {
/* it is inefficient to repeatedly pick one random element from a
* ziplist. so we use this instead: */
ziplistEntry *keys, *vals = NULL;
keys = zmalloc(sizeof(ziplistEntry)*count);
if (withscores)
vals = zmalloc(sizeof(ziplistEntry)*count);
serverAssert(ziplistRandomPairsUnique(zsetobj->ptr, count, keys, vals) == count);
zarndmemberReplyWithZiplist(c, count, keys, vals);
zfree(keys);
zfree(vals);
zuiClearIterator(&src);
return;
}
/* Hashtable encoding (generic implementation) */
unsigned long added = 0;
dict *d = dictCreate(&hashDictType, NULL);
dictExpand(d, count);
while (added < count) {
ziplistEntry key;
double score;
zsetTypeRandomElement(zsetobj, size, &key, withscores ? &score: NULL);
/* Try to add the object to the dictionary. If it already exists
* free it, otherwise increment the number of objects we have
* in the result dictionary. */
sds skey = zsetSdsFromZiplistEntry(&key);
if (dictAdd(d,skey,NULL) != DICT_OK) {
sdsfree(skey);
continue;
}
added++;
if (withscores && c->resp > 2)
addReplyArrayLen(c,2);
zsetReplyFromZiplistEntry(c, &key);
if (withscores)
addReplyDouble(c, score);
}
/* Release memory */
dictRelease(d);
}
zuiClearIterator(&src);
}
/* ZRANDMEMBER key [<count> [WITHSCORES]] */
void zrandmemberCommand(client *c) {
long l;
int withscores = 0;
robj *zset;
ziplistEntry ele;
if (c->argc >= 3) {
if (getLongFromObjectOrReply(c,c->argv[2],&l,NULL) != C_OK) return;
if (c->argc > 4 || (c->argc == 4 && strcasecmp(c->argv[3]->ptr,"withscores"))) {
addReplyErrorObject(c,shared.syntaxerr);
return;
} else if (c->argc == 4)
withscores = 1;
zrandmemberWithCountCommand(c, l, withscores);
return;
}
/* Handle variant without <count> argument. Reply with simple bulk string */
if ((zset = lookupKeyReadOrReply(c,c->argv[1],shared.null[c->resp]))== NULL ||
checkType(c,zset,OBJ_ZSET)) {
return;
}
zsetTypeRandomElement(zset, zsetLength(zset), &ele,NULL);
zsetReplyFromZiplistEntry(c,&ele);
}