]> git.saurik.com Git - redis.git/blame - src/t_zset.c
more AOF server struct fields renamed.
[redis.git] / src / t_zset.c
CommitLineData
e2641e09 1#include "redis.h"
2
3#include <math.h>
4
5/*-----------------------------------------------------------------------------
6 * Sorted set API
7 *----------------------------------------------------------------------------*/
8
9/* ZSETs are ordered sets using two data structures to hold the same elements
10 * in order to get O(log(N)) INSERT and REMOVE operations into a sorted
11 * data structure.
12 *
13 * The elements are added to an hash table mapping Redis objects to scores.
14 * At the same time the elements are added to a skip list mapping scores
15 * to Redis objects (so objects are sorted by scores in this "view"). */
16
17/* This skiplist implementation is almost a C translation of the original
18 * algorithm described by William Pugh in "Skip Lists: A Probabilistic
19 * Alternative to Balanced Trees", modified in three ways:
20 * a) this implementation allows for repeated values.
21 * b) the comparison is not just by key (our 'score') but by satellite data.
22 * c) there is a back pointer, so it's a doubly linked list with the back
23 * pointers being only at "level 1". This allows to traverse the list
24 * from tail to head, useful for ZREVRANGE. */
25
26zskiplistNode *zslCreateNode(int level, double score, robj *obj) {
2159782b 27 zskiplistNode *zn = zmalloc(sizeof(*zn)+level*sizeof(struct zskiplistLevel));
e2641e09 28 zn->score = score;
29 zn->obj = obj;
30 return zn;
31}
32
33zskiplist *zslCreate(void) {
34 int j;
35 zskiplist *zsl;
36
37 zsl = zmalloc(sizeof(*zsl));
38 zsl->level = 1;
39 zsl->length = 0;
40 zsl->header = zslCreateNode(ZSKIPLIST_MAXLEVEL,0,NULL);
41 for (j = 0; j < ZSKIPLIST_MAXLEVEL; j++) {
2159782b
PN
42 zsl->header->level[j].forward = NULL;
43 zsl->header->level[j].span = 0;
e2641e09 44 }
45 zsl->header->backward = NULL;
46 zsl->tail = NULL;
47 return zsl;
48}
49
50void zslFreeNode(zskiplistNode *node) {
51 decrRefCount(node->obj);
e2641e09 52 zfree(node);
53}
54
55void zslFree(zskiplist *zsl) {
2159782b 56 zskiplistNode *node = zsl->header->level[0].forward, *next;
e2641e09 57
e2641e09 58 zfree(zsl->header);
59 while(node) {
2159782b 60 next = node->level[0].forward;
e2641e09 61 zslFreeNode(node);
62 node = next;
63 }
64 zfree(zsl);
65}
66
67int zslRandomLevel(void) {
68 int level = 1;
69 while ((random()&0xFFFF) < (ZSKIPLIST_P * 0xFFFF))
70 level += 1;
71 return (level<ZSKIPLIST_MAXLEVEL) ? level : ZSKIPLIST_MAXLEVEL;
72}
73
69ef89f2 74zskiplistNode *zslInsert(zskiplist *zsl, double score, robj *obj) {
e2641e09 75 zskiplistNode *update[ZSKIPLIST_MAXLEVEL], *x;
76 unsigned int rank[ZSKIPLIST_MAXLEVEL];
77 int i, level;
78
a244a13b 79 redisAssert(!isnan(score));
e2641e09 80 x = zsl->header;
81 for (i = zsl->level-1; i >= 0; i--) {
82 /* store rank that is crossed to reach the insert position */
83 rank[i] = i == (zsl->level-1) ? 0 : rank[i+1];
2159782b
PN
84 while (x->level[i].forward &&
85 (x->level[i].forward->score < score ||
86 (x->level[i].forward->score == score &&
87 compareStringObjects(x->level[i].forward->obj,obj) < 0))) {
88 rank[i] += x->level[i].span;
89 x = x->level[i].forward;
e2641e09 90 }
91 update[i] = x;
92 }
93 /* we assume the key is not already inside, since we allow duplicated
94 * scores, and the re-insertion of score and redis object should never
95 * happpen since the caller of zslInsert() should test in the hash table
96 * if the element is already inside or not. */
97 level = zslRandomLevel();
98 if (level > zsl->level) {
99 for (i = zsl->level; i < level; i++) {
100 rank[i] = 0;
101 update[i] = zsl->header;
2159782b 102 update[i]->level[i].span = zsl->length;
e2641e09 103 }
104 zsl->level = level;
105 }
106 x = zslCreateNode(level,score,obj);
107 for (i = 0; i < level; i++) {
2159782b
PN
108 x->level[i].forward = update[i]->level[i].forward;
109 update[i]->level[i].forward = x;
e2641e09 110
111 /* update span covered by update[i] as x is inserted here */
2159782b
PN
112 x->level[i].span = update[i]->level[i].span - (rank[0] - rank[i]);
113 update[i]->level[i].span = (rank[0] - rank[i]) + 1;
e2641e09 114 }
115
116 /* increment span for untouched levels */
117 for (i = level; i < zsl->level; i++) {
2159782b 118 update[i]->level[i].span++;
e2641e09 119 }
120
121 x->backward = (update[0] == zsl->header) ? NULL : update[0];
2159782b
PN
122 if (x->level[0].forward)
123 x->level[0].forward->backward = x;
e2641e09 124 else
125 zsl->tail = x;
126 zsl->length++;
69ef89f2 127 return x;
e2641e09 128}
129
130/* Internal function used by zslDelete, zslDeleteByScore and zslDeleteByRank */
131void zslDeleteNode(zskiplist *zsl, zskiplistNode *x, zskiplistNode **update) {
132 int i;
133 for (i = 0; i < zsl->level; i++) {
2159782b
PN
134 if (update[i]->level[i].forward == x) {
135 update[i]->level[i].span += x->level[i].span - 1;
136 update[i]->level[i].forward = x->level[i].forward;
e2641e09 137 } else {
2159782b 138 update[i]->level[i].span -= 1;
e2641e09 139 }
140 }
2159782b
PN
141 if (x->level[0].forward) {
142 x->level[0].forward->backward = x->backward;
e2641e09 143 } else {
144 zsl->tail = x->backward;
145 }
2159782b 146 while(zsl->level > 1 && zsl->header->level[zsl->level-1].forward == NULL)
e2641e09 147 zsl->level--;
148 zsl->length--;
149}
150
151/* Delete an element with matching score/object from the skiplist. */
152int zslDelete(zskiplist *zsl, double score, robj *obj) {
153 zskiplistNode *update[ZSKIPLIST_MAXLEVEL], *x;
154 int i;
155
156 x = zsl->header;
157 for (i = zsl->level-1; i >= 0; i--) {
2159782b
PN
158 while (x->level[i].forward &&
159 (x->level[i].forward->score < score ||
160 (x->level[i].forward->score == score &&
161 compareStringObjects(x->level[i].forward->obj,obj) < 0)))
162 x = x->level[i].forward;
e2641e09 163 update[i] = x;
164 }
165 /* We may have multiple elements with the same score, what we need
166 * is to find the element with both the right score and object. */
2159782b 167 x = x->level[0].forward;
e2641e09 168 if (x && score == x->score && equalStringObjects(x->obj,obj)) {
169 zslDeleteNode(zsl, x, update);
170 zslFreeNode(x);
171 return 1;
172 } else {
173 return 0; /* not found */
174 }
175 return 0; /* not found */
176}
177
45290ad9 178static int zslValueGteMin(double value, zrangespec *spec) {
22b9bf15
PN
179 return spec->minex ? (value > spec->min) : (value >= spec->min);
180}
181
45290ad9 182static int zslValueLteMax(double value, zrangespec *spec) {
22b9bf15
PN
183 return spec->maxex ? (value < spec->max) : (value <= spec->max);
184}
185
22b9bf15
PN
186/* Returns if there is a part of the zset is in range. */
187int zslIsInRange(zskiplist *zsl, zrangespec *range) {
188 zskiplistNode *x;
189
8e1b3277
PN
190 /* Test for ranges that will always be empty. */
191 if (range->min > range->max ||
192 (range->min == range->max && (range->minex || range->maxex)))
193 return 0;
22b9bf15 194 x = zsl->tail;
45290ad9 195 if (x == NULL || !zslValueGteMin(x->score,range))
22b9bf15
PN
196 return 0;
197 x = zsl->header->level[0].forward;
45290ad9 198 if (x == NULL || !zslValueLteMax(x->score,range))
22b9bf15
PN
199 return 0;
200 return 1;
201}
202
203/* Find the first node that is contained in the specified range.
204 * Returns NULL when no element is contained in the range. */
205zskiplistNode *zslFirstInRange(zskiplist *zsl, zrangespec range) {
206 zskiplistNode *x;
207 int i;
208
209 /* If everything is out of range, return early. */
210 if (!zslIsInRange(zsl,&range)) return NULL;
211
212 x = zsl->header;
213 for (i = zsl->level-1; i >= 0; i--) {
214 /* Go forward while *OUT* of range. */
215 while (x->level[i].forward &&
45290ad9 216 !zslValueGteMin(x->level[i].forward->score,&range))
22b9bf15
PN
217 x = x->level[i].forward;
218 }
219
e53ca04b 220 /* This is an inner range, so the next node cannot be NULL. */
22b9bf15 221 x = x->level[0].forward;
e53ca04b
PN
222 redisAssert(x != NULL);
223
224 /* Check if score <= max. */
225 if (!zslValueLteMax(x->score,&range)) return NULL;
22b9bf15
PN
226 return x;
227}
228
229/* Find the last node that is contained in the specified range.
230 * Returns NULL when no element is contained in the range. */
231zskiplistNode *zslLastInRange(zskiplist *zsl, zrangespec range) {
232 zskiplistNode *x;
233 int i;
234
235 /* If everything is out of range, return early. */
236 if (!zslIsInRange(zsl,&range)) return NULL;
237
238 x = zsl->header;
239 for (i = zsl->level-1; i >= 0; i--) {
240 /* Go forward while *IN* range. */
241 while (x->level[i].forward &&
45290ad9 242 zslValueLteMax(x->level[i].forward->score,&range))
22b9bf15
PN
243 x = x->level[i].forward;
244 }
245
e53ca04b
PN
246 /* This is an inner range, so this node cannot be NULL. */
247 redisAssert(x != NULL);
248
249 /* Check if score >= min. */
250 if (!zslValueGteMin(x->score,&range)) return NULL;
22b9bf15
PN
251 return x;
252}
253
e2641e09 254/* Delete all the elements with score between min and max from the skiplist.
255 * Min and mx are inclusive, so a score >= min || score <= max is deleted.
256 * Note that this function takes the reference to the hash table view of the
257 * sorted set, in order to remove the elements from the hash table too. */
91504b6c 258unsigned long zslDeleteRangeByScore(zskiplist *zsl, zrangespec range, dict *dict) {
e2641e09 259 zskiplistNode *update[ZSKIPLIST_MAXLEVEL], *x;
260 unsigned long removed = 0;
261 int i;
262
263 x = zsl->header;
264 for (i = zsl->level-1; i >= 0; i--) {
91504b6c
PN
265 while (x->level[i].forward && (range.minex ?
266 x->level[i].forward->score <= range.min :
267 x->level[i].forward->score < range.min))
268 x = x->level[i].forward;
e2641e09 269 update[i] = x;
270 }
91504b6c
PN
271
272 /* Current node is the last with score < or <= min. */
2159782b 273 x = x->level[0].forward;
91504b6c
PN
274
275 /* Delete nodes while in range. */
276 while (x && (range.maxex ? x->score < range.max : x->score <= range.max)) {
2159782b 277 zskiplistNode *next = x->level[0].forward;
69ef89f2 278 zslDeleteNode(zsl,x,update);
e2641e09 279 dictDelete(dict,x->obj);
280 zslFreeNode(x);
281 removed++;
282 x = next;
283 }
91504b6c 284 return removed;
e2641e09 285}
286
287/* Delete all the elements with rank between start and end from the skiplist.
288 * Start and end are inclusive. Note that start and end need to be 1-based */
289unsigned long zslDeleteRangeByRank(zskiplist *zsl, unsigned int start, unsigned int end, dict *dict) {
290 zskiplistNode *update[ZSKIPLIST_MAXLEVEL], *x;
291 unsigned long traversed = 0, removed = 0;
292 int i;
293
294 x = zsl->header;
295 for (i = zsl->level-1; i >= 0; i--) {
2159782b
PN
296 while (x->level[i].forward && (traversed + x->level[i].span) < start) {
297 traversed += x->level[i].span;
298 x = x->level[i].forward;
e2641e09 299 }
300 update[i] = x;
301 }
302
303 traversed++;
2159782b 304 x = x->level[0].forward;
e2641e09 305 while (x && traversed <= end) {
2159782b 306 zskiplistNode *next = x->level[0].forward;
69ef89f2 307 zslDeleteNode(zsl,x,update);
e2641e09 308 dictDelete(dict,x->obj);
309 zslFreeNode(x);
310 removed++;
311 traversed++;
312 x = next;
313 }
314 return removed;
315}
316
e2641e09 317/* Find the rank for an element by both score and key.
318 * Returns 0 when the element cannot be found, rank otherwise.
319 * Note that the rank is 1-based due to the span of zsl->header to the
320 * first element. */
a3004773 321unsigned long zslGetRank(zskiplist *zsl, double score, robj *o) {
e2641e09 322 zskiplistNode *x;
323 unsigned long rank = 0;
324 int i;
325
326 x = zsl->header;
327 for (i = zsl->level-1; i >= 0; i--) {
2159782b
PN
328 while (x->level[i].forward &&
329 (x->level[i].forward->score < score ||
330 (x->level[i].forward->score == score &&
331 compareStringObjects(x->level[i].forward->obj,o) <= 0))) {
332 rank += x->level[i].span;
333 x = x->level[i].forward;
e2641e09 334 }
335
336 /* x might be equal to zsl->header, so test if obj is non-NULL */
337 if (x->obj && equalStringObjects(x->obj,o)) {
338 return rank;
339 }
340 }
341 return 0;
342}
343
344/* Finds an element by its rank. The rank argument needs to be 1-based. */
a3004773 345zskiplistNode* zslGetElementByRank(zskiplist *zsl, unsigned long rank) {
e2641e09 346 zskiplistNode *x;
347 unsigned long traversed = 0;
348 int i;
349
350 x = zsl->header;
351 for (i = zsl->level-1; i >= 0; i--) {
2159782b 352 while (x->level[i].forward && (traversed + x->level[i].span) <= rank)
e2641e09 353 {
2159782b
PN
354 traversed += x->level[i].span;
355 x = x->level[i].forward;
e2641e09 356 }
357 if (traversed == rank) {
358 return x;
359 }
360 }
361 return NULL;
362}
363
25bb8a44 364/* Populate the rangespec according to the objects min and max. */
7236fdb2
PN
365static int zslParseRange(robj *min, robj *max, zrangespec *spec) {
366 char *eptr;
25bb8a44
PN
367 spec->minex = spec->maxex = 0;
368
369 /* Parse the min-max interval. If one of the values is prefixed
370 * by the "(" character, it's considered "open". For instance
371 * ZRANGEBYSCORE zset (1.5 (2.5 will match min < x < max
372 * ZRANGEBYSCORE zset 1.5 2.5 will instead match min <= x <= max */
373 if (min->encoding == REDIS_ENCODING_INT) {
374 spec->min = (long)min->ptr;
375 } else {
376 if (((char*)min->ptr)[0] == '(') {
7236fdb2
PN
377 spec->min = strtod((char*)min->ptr+1,&eptr);
378 if (eptr[0] != '\0' || isnan(spec->min)) return REDIS_ERR;
25bb8a44
PN
379 spec->minex = 1;
380 } else {
7236fdb2
PN
381 spec->min = strtod((char*)min->ptr,&eptr);
382 if (eptr[0] != '\0' || isnan(spec->min)) return REDIS_ERR;
25bb8a44
PN
383 }
384 }
385 if (max->encoding == REDIS_ENCODING_INT) {
386 spec->max = (long)max->ptr;
387 } else {
388 if (((char*)max->ptr)[0] == '(') {
7236fdb2
PN
389 spec->max = strtod((char*)max->ptr+1,&eptr);
390 if (eptr[0] != '\0' || isnan(spec->max)) return REDIS_ERR;
25bb8a44
PN
391 spec->maxex = 1;
392 } else {
7236fdb2
PN
393 spec->max = strtod((char*)max->ptr,&eptr);
394 if (eptr[0] != '\0' || isnan(spec->max)) return REDIS_ERR;
25bb8a44
PN
395 }
396 }
397
398 return REDIS_OK;
399}
400
21c5b508
PN
401/*-----------------------------------------------------------------------------
402 * Ziplist-backed sorted set API
403 *----------------------------------------------------------------------------*/
404
405double zzlGetScore(unsigned char *sptr) {
406 unsigned char *vstr;
407 unsigned int vlen;
408 long long vlong;
409 char buf[128];
410 double score;
411
412 redisAssert(sptr != NULL);
413 redisAssert(ziplistGet(sptr,&vstr,&vlen,&vlong));
414
415 if (vstr) {
416 memcpy(buf,vstr,vlen);
417 buf[vlen] = '\0';
418 score = strtod(buf,NULL);
419 } else {
420 score = vlong;
421 }
422
423 return score;
424}
425
426/* Compare element in sorted set with given element. */
427int zzlCompareElements(unsigned char *eptr, unsigned char *cstr, unsigned int clen) {
428 unsigned char *vstr;
429 unsigned int vlen;
430 long long vlong;
431 unsigned char vbuf[32];
432 int minlen, cmp;
433
434 redisAssert(ziplistGet(eptr,&vstr,&vlen,&vlong));
435 if (vstr == NULL) {
436 /* Store string representation of long long in buf. */
437 vlen = ll2string((char*)vbuf,sizeof(vbuf),vlong);
438 vstr = vbuf;
439 }
440
441 minlen = (vlen < clen) ? vlen : clen;
442 cmp = memcmp(vstr,cstr,minlen);
443 if (cmp == 0) return vlen-clen;
444 return cmp;
445}
446
bbfe232f 447unsigned int zzlLength(unsigned char *zl) {
0b10e104
PN
448 return ziplistLen(zl)/2;
449}
450
4c5f0966
PN
451/* Move to next entry based on the values in eptr and sptr. Both are set to
452 * NULL when there is no next entry. */
453void zzlNext(unsigned char *zl, unsigned char **eptr, unsigned char **sptr) {
454 unsigned char *_eptr, *_sptr;
455 redisAssert(*eptr != NULL && *sptr != NULL);
456
457 _eptr = ziplistNext(zl,*sptr);
458 if (_eptr != NULL) {
459 _sptr = ziplistNext(zl,_eptr);
460 redisAssert(_sptr != NULL);
461 } else {
462 /* No next entry. */
463 _sptr = NULL;
464 }
465
466 *eptr = _eptr;
467 *sptr = _sptr;
468}
469
470/* Move to the previous entry based on the values in eptr and sptr. Both are
471 * set to NULL when there is no next entry. */
472void zzlPrev(unsigned char *zl, unsigned char **eptr, unsigned char **sptr) {
473 unsigned char *_eptr, *_sptr;
474 redisAssert(*eptr != NULL && *sptr != NULL);
475
476 _sptr = ziplistPrev(zl,*eptr);
477 if (_sptr != NULL) {
478 _eptr = ziplistPrev(zl,_sptr);
479 redisAssert(_eptr != NULL);
480 } else {
481 /* No previous entry. */
482 _eptr = NULL;
483 }
484
485 *eptr = _eptr;
486 *sptr = _sptr;
487}
488
4a14dbba
PN
489/* Returns if there is a part of the zset is in range. Should only be used
490 * internally by zzlFirstInRange and zzlLastInRange. */
491int zzlIsInRange(unsigned char *zl, zrangespec *range) {
492 unsigned char *p;
493 double score;
494
495 /* Test for ranges that will always be empty. */
496 if (range->min > range->max ||
497 (range->min == range->max && (range->minex || range->maxex)))
498 return 0;
499
500 p = ziplistIndex(zl,-1); /* Last score. */
501 redisAssert(p != NULL);
502 score = zzlGetScore(p);
503 if (!zslValueGteMin(score,range))
504 return 0;
505
506 p = ziplistIndex(zl,1); /* First score. */
507 redisAssert(p != NULL);
508 score = zzlGetScore(p);
509 if (!zslValueLteMax(score,range))
510 return 0;
511
512 return 1;
513}
514
515/* Find pointer to the first element contained in the specified range.
516 * Returns NULL when no element is contained in the range. */
8588bfa3 517unsigned char *zzlFirstInRange(unsigned char *zl, zrangespec range) {
4a14dbba
PN
518 unsigned char *eptr = ziplistIndex(zl,0), *sptr;
519 double score;
520
521 /* If everything is out of range, return early. */
522 if (!zzlIsInRange(zl,&range)) return NULL;
523
524 while (eptr != NULL) {
525 sptr = ziplistNext(zl,eptr);
526 redisAssert(sptr != NULL);
527
528 score = zzlGetScore(sptr);
e53ca04b
PN
529 if (zslValueGteMin(score,&range)) {
530 /* Check if score <= max. */
531 if (zslValueLteMax(score,&range))
532 return eptr;
533 return NULL;
534 }
4a14dbba
PN
535
536 /* Move to next element. */
537 eptr = ziplistNext(zl,sptr);
538 }
539
540 return NULL;
541}
542
543/* Find pointer to the last element contained in the specified range.
544 * Returns NULL when no element is contained in the range. */
8588bfa3 545unsigned char *zzlLastInRange(unsigned char *zl, zrangespec range) {
4a14dbba
PN
546 unsigned char *eptr = ziplistIndex(zl,-2), *sptr;
547 double score;
548
549 /* If everything is out of range, return early. */
550 if (!zzlIsInRange(zl,&range)) return NULL;
551
552 while (eptr != NULL) {
553 sptr = ziplistNext(zl,eptr);
554 redisAssert(sptr != NULL);
555
556 score = zzlGetScore(sptr);
e53ca04b
PN
557 if (zslValueLteMax(score,&range)) {
558 /* Check if score >= min. */
559 if (zslValueGteMin(score,&range))
560 return eptr;
561 return NULL;
562 }
4a14dbba
PN
563
564 /* Move to previous element by moving to the score of previous element.
565 * When this returns NULL, we know there also is no element. */
566 sptr = ziplistPrev(zl,eptr);
567 if (sptr != NULL)
568 redisAssert((eptr = ziplistPrev(zl,sptr)) != NULL);
569 else
570 eptr = NULL;
571 }
572
573 return NULL;
574}
575
8588bfa3 576unsigned char *zzlFind(unsigned char *zl, robj *ele, double *score) {
21c5b508
PN
577 unsigned char *eptr = ziplistIndex(zl,0), *sptr;
578
579 ele = getDecodedObject(ele);
580 while (eptr != NULL) {
581 sptr = ziplistNext(zl,eptr);
eab0e26e 582 redisAssertWithInfo(NULL,ele,sptr != NULL);
21c5b508
PN
583
584 if (ziplistCompare(eptr,ele->ptr,sdslen(ele->ptr))) {
585 /* Matching element, pull out score. */
0b10e104 586 if (score != NULL) *score = zzlGetScore(sptr);
21c5b508
PN
587 decrRefCount(ele);
588 return eptr;
589 }
590
591 /* Move to next element. */
592 eptr = ziplistNext(zl,sptr);
593 }
594
595 decrRefCount(ele);
596 return NULL;
597}
598
599/* Delete (element,score) pair from ziplist. Use local copy of eptr because we
600 * don't want to modify the one given as argument. */
8588bfa3 601unsigned char *zzlDelete(unsigned char *zl, unsigned char *eptr) {
21c5b508
PN
602 unsigned char *p = eptr;
603
604 /* TODO: add function to ziplist API to delete N elements from offset. */
605 zl = ziplistDelete(zl,&p);
606 zl = ziplistDelete(zl,&p);
8588bfa3 607 return zl;
21c5b508
PN
608}
609
8588bfa3 610unsigned char *zzlInsertAt(unsigned char *zl, unsigned char *eptr, robj *ele, double score) {
21c5b508
PN
611 unsigned char *sptr;
612 char scorebuf[128];
613 int scorelen;
69298a05 614 size_t offset;
21c5b508 615
eab0e26e 616 redisAssertWithInfo(NULL,ele,ele->encoding == REDIS_ENCODING_RAW);
21c5b508
PN
617 scorelen = d2string(scorebuf,sizeof(scorebuf),score);
618 if (eptr == NULL) {
619 zl = ziplistPush(zl,ele->ptr,sdslen(ele->ptr),ZIPLIST_TAIL);
620 zl = ziplistPush(zl,(unsigned char*)scorebuf,scorelen,ZIPLIST_TAIL);
621 } else {
622 /* Keep offset relative to zl, as it might be re-allocated. */
623 offset = eptr-zl;
624 zl = ziplistInsert(zl,eptr,ele->ptr,sdslen(ele->ptr));
625 eptr = zl+offset;
626
627 /* Insert score after the element. */
eab0e26e 628 redisAssertWithInfo(NULL,ele,(sptr = ziplistNext(zl,eptr)) != NULL);
21c5b508
PN
629 zl = ziplistInsert(zl,sptr,(unsigned char*)scorebuf,scorelen);
630 }
631
8588bfa3 632 return zl;
21c5b508
PN
633}
634
635/* Insert (element,score) pair in ziplist. This function assumes the element is
636 * not yet present in the list. */
8588bfa3 637unsigned char *zzlInsert(unsigned char *zl, robj *ele, double score) {
21c5b508
PN
638 unsigned char *eptr = ziplistIndex(zl,0), *sptr;
639 double s;
21c5b508
PN
640
641 ele = getDecodedObject(ele);
642 while (eptr != NULL) {
643 sptr = ziplistNext(zl,eptr);
eab0e26e 644 redisAssertWithInfo(NULL,ele,sptr != NULL);
21c5b508
PN
645 s = zzlGetScore(sptr);
646
647 if (s > score) {
648 /* First element with score larger than score for element to be
649 * inserted. This means we should take its spot in the list to
650 * maintain ordering. */
8588bfa3 651 zl = zzlInsertAt(zl,eptr,ele,score);
21c5b508 652 break;
8218db3d
PN
653 } else if (s == score) {
654 /* Ensure lexicographical ordering for elements. */
d1c920c5 655 if (zzlCompareElements(eptr,ele->ptr,sdslen(ele->ptr)) > 0) {
8588bfa3 656 zl = zzlInsertAt(zl,eptr,ele,score);
8218db3d
PN
657 break;
658 }
21c5b508
PN
659 }
660
661 /* Move to next element. */
662 eptr = ziplistNext(zl,sptr);
663 }
664
665 /* Push on tail of list when it was not yet inserted. */
8218db3d 666 if (eptr == NULL)
8588bfa3 667 zl = zzlInsertAt(zl,NULL,ele,score);
21c5b508
PN
668
669 decrRefCount(ele);
8588bfa3 670 return zl;
21c5b508 671}
25bb8a44 672
8588bfa3 673unsigned char *zzlDeleteRangeByScore(unsigned char *zl, zrangespec range, unsigned long *deleted) {
4a14dbba
PN
674 unsigned char *eptr, *sptr;
675 double score;
8588bfa3 676 unsigned long num = 0;
4a14dbba 677
8588bfa3 678 if (deleted != NULL) *deleted = 0;
4a14dbba 679
8588bfa3
PN
680 eptr = zzlFirstInRange(zl,range);
681 if (eptr == NULL) return zl;
4a14dbba
PN
682
683 /* When the tail of the ziplist is deleted, eptr will point to the sentinel
684 * byte and ziplistNext will return NULL. */
685 while ((sptr = ziplistNext(zl,eptr)) != NULL) {
686 score = zzlGetScore(sptr);
687 if (zslValueLteMax(score,&range)) {
688 /* Delete both the element and the score. */
689 zl = ziplistDelete(zl,&eptr);
690 zl = ziplistDelete(zl,&eptr);
8588bfa3 691 num++;
4a14dbba
PN
692 } else {
693 /* No longer in range. */
694 break;
695 }
696 }
697
8588bfa3
PN
698 if (deleted != NULL) *deleted = num;
699 return zl;
4a14dbba
PN
700}
701
63b7b7fb
PN
702/* Delete all the elements with rank between start and end from the skiplist.
703 * Start and end are inclusive. Note that start and end need to be 1-based */
8588bfa3 704unsigned char *zzlDeleteRangeByRank(unsigned char *zl, unsigned int start, unsigned int end, unsigned long *deleted) {
63b7b7fb 705 unsigned int num = (end-start)+1;
8588bfa3
PN
706 if (deleted) *deleted = num;
707 zl = ziplistDeleteRange(zl,2*(start-1),2*num);
708 return zl;
63b7b7fb
PN
709}
710
5d1b4fb6
PN
711/*-----------------------------------------------------------------------------
712 * Common sorted set API
713 *----------------------------------------------------------------------------*/
714
df26a0ae 715unsigned int zsetLength(robj *zobj) {
5d1b4fb6
PN
716 int length = -1;
717 if (zobj->encoding == REDIS_ENCODING_ZIPLIST) {
bbfe232f 718 length = zzlLength(zobj->ptr);
100ed062 719 } else if (zobj->encoding == REDIS_ENCODING_SKIPLIST) {
5d1b4fb6
PN
720 length = ((zset*)zobj->ptr)->zsl->length;
721 } else {
722 redisPanic("Unknown sorted set encoding");
723 }
724 return length;
725}
726
df26a0ae 727void zsetConvert(robj *zobj, int encoding) {
a669d5e9
PN
728 zset *zs;
729 zskiplistNode *node, *next;
730 robj *ele;
731 double score;
732
733 if (zobj->encoding == encoding) return;
734 if (zobj->encoding == REDIS_ENCODING_ZIPLIST) {
735 unsigned char *zl = zobj->ptr;
736 unsigned char *eptr, *sptr;
737 unsigned char *vstr;
738 unsigned int vlen;
739 long long vlong;
740
100ed062 741 if (encoding != REDIS_ENCODING_SKIPLIST)
a669d5e9
PN
742 redisPanic("Unknown target encoding");
743
744 zs = zmalloc(sizeof(*zs));
745 zs->dict = dictCreate(&zsetDictType,NULL);
746 zs->zsl = zslCreate();
747
748 eptr = ziplistIndex(zl,0);
eab0e26e 749 redisAssertWithInfo(NULL,zobj,eptr != NULL);
a669d5e9 750 sptr = ziplistNext(zl,eptr);
eab0e26e 751 redisAssertWithInfo(NULL,zobj,sptr != NULL);
a669d5e9
PN
752
753 while (eptr != NULL) {
754 score = zzlGetScore(sptr);
eab0e26e 755 redisAssertWithInfo(NULL,zobj,ziplistGet(eptr,&vstr,&vlen,&vlong));
a669d5e9
PN
756 if (vstr == NULL)
757 ele = createStringObjectFromLongLong(vlong);
758 else
759 ele = createStringObject((char*)vstr,vlen);
760
761 /* Has incremented refcount since it was just created. */
762 node = zslInsert(zs->zsl,score,ele);
eab0e26e 763 redisAssertWithInfo(NULL,zobj,dictAdd(zs->dict,ele,&node->score) == DICT_OK);
a669d5e9
PN
764 incrRefCount(ele); /* Added to dictionary. */
765 zzlNext(zl,&eptr,&sptr);
766 }
767
768 zfree(zobj->ptr);
769 zobj->ptr = zs;
100ed062
PN
770 zobj->encoding = REDIS_ENCODING_SKIPLIST;
771 } else if (zobj->encoding == REDIS_ENCODING_SKIPLIST) {
a669d5e9
PN
772 unsigned char *zl = ziplistNew();
773
774 if (encoding != REDIS_ENCODING_ZIPLIST)
775 redisPanic("Unknown target encoding");
776
777 /* Approach similar to zslFree(), since we want to free the skiplist at
778 * the same time as creating the ziplist. */
779 zs = zobj->ptr;
780 dictRelease(zs->dict);
781 node = zs->zsl->header->level[0].forward;
782 zfree(zs->zsl->header);
783 zfree(zs->zsl);
784
a669d5e9
PN
785 while (node) {
786 ele = getDecodedObject(node->obj);
8588bfa3 787 zl = zzlInsertAt(zl,NULL,ele,node->score);
a669d5e9
PN
788 decrRefCount(ele);
789
790 next = node->level[0].forward;
791 zslFreeNode(node);
792 node = next;
793 }
794
795 zfree(zs);
8588bfa3 796 zobj->ptr = zl;
a669d5e9
PN
797 zobj->encoding = REDIS_ENCODING_ZIPLIST;
798 } else {
799 redisPanic("Unknown sorted set encoding");
800 }
801}
802
e2641e09 803/*-----------------------------------------------------------------------------
804 * Sorted set commands
805 *----------------------------------------------------------------------------*/
806
69ef89f2 807/* This generic command implements both ZADD and ZINCRBY. */
3ca7532a 808void zaddGenericCommand(redisClient *c, int incr) {
21c5b508 809 static char *nanerr = "resulting score is not a number (NaN)";
3ca7532a
PN
810 robj *key = c->argv[1];
811 robj *ele;
21c5b508
PN
812 robj *zobj;
813 robj *curobj;
ef231a7c 814 double score = 0, *scores, curscore = 0.0;
815 int j, elements = (c->argc-2)/2;
816 int added = 0;
3ca7532a 817
ef231a7c 818 if (c->argc % 2) {
819 addReply(c,shared.syntaxerr);
3ca7532a 820 return;
ef231a7c 821 }
822
823 /* Start parsing all the scores, we need to emit any syntax error
824 * before executing additions to the sorted set, as the command should
825 * either execute fully or nothing at all. */
826 scores = zmalloc(sizeof(double)*elements);
827 for (j = 0; j < elements; j++) {
828 if (getDoubleFromObjectOrReply(c,c->argv[2+j*2],&scores[j],NULL)
829 != REDIS_OK)
830 {
831 zfree(scores);
832 return;
833 }
834 }
21c5b508 835
ef231a7c 836 /* Lookup the key and create the sorted set if does not exist. */
21c5b508
PN
837 zobj = lookupKeyWrite(c->db,key);
838 if (zobj == NULL) {
a669d5e9
PN
839 if (server.zset_max_ziplist_entries == 0 ||
840 server.zset_max_ziplist_value < sdslen(c->argv[3]->ptr))
841 {
842 zobj = createZsetObject();
843 } else {
844 zobj = createZsetZiplistObject();
845 }
21c5b508 846 dbAdd(c->db,key,zobj);
e2641e09 847 } else {
21c5b508 848 if (zobj->type != REDIS_ZSET) {
e2641e09 849 addReply(c,shared.wrongtypeerr);
ef231a7c 850 zfree(scores);
e2641e09 851 return;
852 }
853 }
e2641e09 854
ef231a7c 855 for (j = 0; j < elements; j++) {
856 score = scores[j];
857
858 if (zobj->encoding == REDIS_ENCODING_ZIPLIST) {
859 unsigned char *eptr;
860
861 /* Prefer non-encoded element when dealing with ziplists. */
862 ele = c->argv[3+j*2];
863 if ((eptr = zzlFind(zobj->ptr,ele,&curscore)) != NULL) {
864 if (incr) {
865 score += curscore;
866 if (isnan(score)) {
867 addReplyError(c,nanerr);
868 /* Don't need to check if the sorted set is empty
869 * because we know it has at least one element. */
870 zfree(scores);
871 return;
872 }
21c5b508 873 }
21c5b508 874
ef231a7c 875 /* Remove and re-insert when score changed. */
876 if (score != curscore) {
877 zobj->ptr = zzlDelete(zobj->ptr,eptr);
878 zobj->ptr = zzlInsert(zobj->ptr,ele,score);
879
880 signalModifiedKey(c->db,key);
881 server.dirty++;
882 }
883 } else {
884 /* Optimize: check if the element is too large or the list
885 * becomes too long *before* executing zzlInsert. */
8588bfa3 886 zobj->ptr = zzlInsert(zobj->ptr,ele,score);
ef231a7c 887 if (zzlLength(zobj->ptr) > server.zset_max_ziplist_entries)
888 zsetConvert(zobj,REDIS_ENCODING_SKIPLIST);
889 if (sdslen(ele->ptr) > server.zset_max_ziplist_value)
890 zsetConvert(zobj,REDIS_ENCODING_SKIPLIST);
21c5b508
PN
891
892 signalModifiedKey(c->db,key);
893 server.dirty++;
ef231a7c 894 if (!incr) added++;
21c5b508 895 }
ef231a7c 896 } else if (zobj->encoding == REDIS_ENCODING_SKIPLIST) {
897 zset *zs = zobj->ptr;
898 zskiplistNode *znode;
899 dictEntry *de;
900
901 ele = c->argv[3+j*2] = tryObjectEncoding(c->argv[3+j*2]);
902 de = dictFind(zs->dict,ele);
903 if (de != NULL) {
c0ba9ebe 904 curobj = dictGetKey(de);
905 curscore = *(double*)dictGetVal(de);
ef231a7c 906
907 if (incr) {
908 score += curscore;
909 if (isnan(score)) {
910 addReplyError(c,nanerr);
911 /* Don't need to check if the sorted set is empty
912 * because we know it has at least one element. */
913 zfree(scores);
914 return;
915 }
21c5b508 916 }
21c5b508 917
ef231a7c 918 /* Remove and re-insert when score changed. We can safely
919 * delete the key object from the skiplist, since the
920 * dictionary still has a reference to it. */
921 if (score != curscore) {
eab0e26e 922 redisAssertWithInfo(c,curobj,zslDelete(zs->zsl,curscore,curobj));
ef231a7c 923 znode = zslInsert(zs->zsl,score,curobj);
924 incrRefCount(curobj); /* Re-inserted in skiplist. */
c0ba9ebe 925 dictGetVal(de) = &znode->score; /* Update score ptr. */
ef231a7c 926
927 signalModifiedKey(c->db,key);
928 server.dirty++;
929 }
930 } else {
931 znode = zslInsert(zs->zsl,score,ele);
932 incrRefCount(ele); /* Inserted in skiplist. */
f013f400 933 redisAssertWithInfo(c,NULL,dictAdd(zs->dict,ele,&znode->score) == DICT_OK);
ef231a7c 934 incrRefCount(ele); /* Added to dictionary. */
21c5b508
PN
935
936 signalModifiedKey(c->db,key);
937 server.dirty++;
ef231a7c 938 if (!incr) added++;
21c5b508 939 }
21c5b508 940 } else {
ef231a7c 941 redisPanic("Unknown sorted set encoding");
e2641e09 942 }
e2641e09 943 }
ef231a7c 944 zfree(scores);
945 if (incr) /* ZINCRBY */
946 addReplyDouble(c,score);
947 else /* ZADD */
948 addReplyLongLong(c,added);
e2641e09 949}
950
951void zaddCommand(redisClient *c) {
3ca7532a 952 zaddGenericCommand(c,0);
e2641e09 953}
954
955void zincrbyCommand(redisClient *c) {
3ca7532a 956 zaddGenericCommand(c,1);
e2641e09 957}
958
959void zremCommand(redisClient *c) {
0b10e104 960 robj *key = c->argv[1];
0b10e104 961 robj *zobj;
3f7b2b1f 962 int deleted = 0, j;
0b10e104
PN
963
964 if ((zobj = lookupKeyWriteOrReply(c,key,shared.czero)) == NULL ||
965 checkType(c,zobj,REDIS_ZSET)) return;
966
967 if (zobj->encoding == REDIS_ENCODING_ZIPLIST) {
968 unsigned char *eptr;
969
3f7b2b1f 970 for (j = 2; j < c->argc; j++) {
971 if ((eptr = zzlFind(zobj->ptr,c->argv[j],NULL)) != NULL) {
972 deleted++;
973 zobj->ptr = zzlDelete(zobj->ptr,eptr);
974 if (zzlLength(zobj->ptr) == 0) {
975 dbDelete(c->db,key);
976 break;
977 }
978 }
0b10e104 979 }
100ed062 980 } else if (zobj->encoding == REDIS_ENCODING_SKIPLIST) {
0b10e104
PN
981 zset *zs = zobj->ptr;
982 dictEntry *de;
983 double score;
984
3f7b2b1f 985 for (j = 2; j < c->argc; j++) {
986 de = dictFind(zs->dict,c->argv[j]);
987 if (de != NULL) {
988 deleted++;
989
990 /* Delete from the skiplist */
c0ba9ebe 991 score = *(double*)dictGetVal(de);
eab0e26e 992 redisAssertWithInfo(c,c->argv[j],zslDelete(zs->zsl,score,c->argv[j]));
3f7b2b1f 993
994 /* Delete from the hash table */
995 dictDelete(zs->dict,c->argv[j]);
996 if (htNeedsResize(zs->dict)) dictResize(zs->dict);
997 if (dictSize(zs->dict) == 0) {
998 dbDelete(c->db,key);
999 break;
1000 }
1001 }
0b10e104
PN
1002 }
1003 } else {
1004 redisPanic("Unknown sorted set encoding");
e2641e09 1005 }
e2641e09 1006
3f7b2b1f 1007 if (deleted) {
1008 signalModifiedKey(c->db,key);
1009 server.dirty += deleted;
1010 }
1011 addReplyLongLong(c,deleted);
e2641e09 1012}
1013
1014void zremrangebyscoreCommand(redisClient *c) {
4a14dbba
PN
1015 robj *key = c->argv[1];
1016 robj *zobj;
91504b6c 1017 zrangespec range;
4a14dbba 1018 unsigned long deleted;
e2641e09 1019
91504b6c 1020 /* Parse the range arguments. */
7236fdb2 1021 if (zslParseRange(c->argv[2],c->argv[3],&range) != REDIS_OK) {
d93f9a86 1022 addReplyError(c,"min or max is not a float");
7236fdb2
PN
1023 return;
1024 }
e2641e09 1025
4a14dbba
PN
1026 if ((zobj = lookupKeyWriteOrReply(c,key,shared.czero)) == NULL ||
1027 checkType(c,zobj,REDIS_ZSET)) return;
1028
1029 if (zobj->encoding == REDIS_ENCODING_ZIPLIST) {
8588bfa3 1030 zobj->ptr = zzlDeleteRangeByScore(zobj->ptr,range,&deleted);
48991620 1031 if (zzlLength(zobj->ptr) == 0) dbDelete(c->db,key);
100ed062 1032 } else if (zobj->encoding == REDIS_ENCODING_SKIPLIST) {
4a14dbba
PN
1033 zset *zs = zobj->ptr;
1034 deleted = zslDeleteRangeByScore(zs->zsl,range,zs->dict);
1035 if (htNeedsResize(zs->dict)) dictResize(zs->dict);
1036 if (dictSize(zs->dict) == 0) dbDelete(c->db,key);
1037 } else {
1038 redisPanic("Unknown sorted set encoding");
1039 }
e2641e09 1040
4a14dbba 1041 if (deleted) signalModifiedKey(c->db,key);
e2641e09 1042 server.dirty += deleted;
1043 addReplyLongLong(c,deleted);
1044}
1045
1046void zremrangebyrankCommand(redisClient *c) {
63b7b7fb
PN
1047 robj *key = c->argv[1];
1048 robj *zobj;
e2641e09 1049 long start;
1050 long end;
1051 int llen;
63b7b7fb 1052 unsigned long deleted;
e2641e09 1053
1054 if ((getLongFromObjectOrReply(c, c->argv[2], &start, NULL) != REDIS_OK) ||
1055 (getLongFromObjectOrReply(c, c->argv[3], &end, NULL) != REDIS_OK)) return;
1056
63b7b7fb
PN
1057 if ((zobj = lookupKeyWriteOrReply(c,key,shared.czero)) == NULL ||
1058 checkType(c,zobj,REDIS_ZSET)) return;
e2641e09 1059
63b7b7fb 1060 /* Sanitize indexes. */
df26a0ae 1061 llen = zsetLength(zobj);
e2641e09 1062 if (start < 0) start = llen+start;
1063 if (end < 0) end = llen+end;
1064 if (start < 0) start = 0;
e2641e09 1065
d0a4e24e
PN
1066 /* Invariant: start >= 0, so this test will be true when end < 0.
1067 * The range is empty when start > end or start >= length. */
e2641e09 1068 if (start > end || start >= llen) {
1069 addReply(c,shared.czero);
1070 return;
1071 }
1072 if (end >= llen) end = llen-1;
1073
63b7b7fb
PN
1074 if (zobj->encoding == REDIS_ENCODING_ZIPLIST) {
1075 /* Correct for 1-based rank. */
8588bfa3 1076 zobj->ptr = zzlDeleteRangeByRank(zobj->ptr,start+1,end+1,&deleted);
48991620 1077 if (zzlLength(zobj->ptr) == 0) dbDelete(c->db,key);
100ed062 1078 } else if (zobj->encoding == REDIS_ENCODING_SKIPLIST) {
63b7b7fb
PN
1079 zset *zs = zobj->ptr;
1080
1081 /* Correct for 1-based rank. */
1082 deleted = zslDeleteRangeByRank(zs->zsl,start+1,end+1,zs->dict);
1083 if (htNeedsResize(zs->dict)) dictResize(zs->dict);
1084 if (dictSize(zs->dict) == 0) dbDelete(c->db,key);
1085 } else {
1086 redisPanic("Unknown sorted set encoding");
1087 }
1088
1089 if (deleted) signalModifiedKey(c->db,key);
e2641e09 1090 server.dirty += deleted;
63b7b7fb 1091 addReplyLongLong(c,deleted);
e2641e09 1092}
1093
1094typedef struct {
56ce42fa
PN
1095 robj *subject;
1096 int type; /* Set, sorted set */
1097 int encoding;
e2641e09 1098 double weight;
56ce42fa
PN
1099
1100 union {
1101 /* Set iterators. */
1102 union _iterset {
1103 struct {
1104 intset *is;
1105 int ii;
1106 } is;
1107 struct {
1108 dict *dict;
1109 dictIterator *di;
1110 dictEntry *de;
1111 } ht;
1112 } set;
1113
1114 /* Sorted set iterators. */
1115 union _iterzset {
1116 struct {
1117 unsigned char *zl;
1118 unsigned char *eptr, *sptr;
1119 } zl;
1120 struct {
1121 zset *zs;
1122 zskiplistNode *node;
1123 } sl;
1124 } zset;
1125 } iter;
e2641e09 1126} zsetopsrc;
1127
56ce42fa
PN
1128
1129/* Use dirty flags for pointers that need to be cleaned up in the next
1130 * iteration over the zsetopval. The dirty flag for the long long value is
1131 * special, since long long values don't need cleanup. Instead, it means that
1132 * we already checked that "ell" holds a long long, or tried to convert another
1133 * representation into a long long value. When this was successful,
1134 * OPVAL_VALID_LL is set as well. */
1135#define OPVAL_DIRTY_ROBJ 1
1136#define OPVAL_DIRTY_LL 2
1137#define OPVAL_VALID_LL 4
1138
1139/* Store value retrieved from the iterator. */
1140typedef struct {
1141 int flags;
1142 unsigned char _buf[32]; /* Private buffer. */
1143 robj *ele;
1144 unsigned char *estr;
1145 unsigned int elen;
1146 long long ell;
1147 double score;
1148} zsetopval;
1149
1150typedef union _iterset iterset;
1151typedef union _iterzset iterzset;
1152
1153void zuiInitIterator(zsetopsrc *op) {
1154 if (op->subject == NULL)
1155 return;
1156
1157 if (op->type == REDIS_SET) {
1158 iterset *it = &op->iter.set;
1159 if (op->encoding == REDIS_ENCODING_INTSET) {
1160 it->is.is = op->subject->ptr;
1161 it->is.ii = 0;
1162 } else if (op->encoding == REDIS_ENCODING_HT) {
1163 it->ht.dict = op->subject->ptr;
1164 it->ht.di = dictGetIterator(op->subject->ptr);
1165 it->ht.de = dictNext(it->ht.di);
1166 } else {
1167 redisPanic("Unknown set encoding");
1168 }
1169 } else if (op->type == REDIS_ZSET) {
1170 iterzset *it = &op->iter.zset;
1171 if (op->encoding == REDIS_ENCODING_ZIPLIST) {
1172 it->zl.zl = op->subject->ptr;
1173 it->zl.eptr = ziplistIndex(it->zl.zl,0);
1174 if (it->zl.eptr != NULL) {
1175 it->zl.sptr = ziplistNext(it->zl.zl,it->zl.eptr);
1176 redisAssert(it->zl.sptr != NULL);
1177 }
100ed062 1178 } else if (op->encoding == REDIS_ENCODING_SKIPLIST) {
56ce42fa
PN
1179 it->sl.zs = op->subject->ptr;
1180 it->sl.node = it->sl.zs->zsl->header->level[0].forward;
1181 } else {
1182 redisPanic("Unknown sorted set encoding");
1183 }
1184 } else {
1185 redisPanic("Unsupported type");
1186 }
1187}
1188
1189void zuiClearIterator(zsetopsrc *op) {
1190 if (op->subject == NULL)
1191 return;
1192
1193 if (op->type == REDIS_SET) {
1194 iterset *it = &op->iter.set;
1195 if (op->encoding == REDIS_ENCODING_INTSET) {
1196 REDIS_NOTUSED(it); /* skip */
1197 } else if (op->encoding == REDIS_ENCODING_HT) {
1198 dictReleaseIterator(it->ht.di);
1199 } else {
1200 redisPanic("Unknown set encoding");
1201 }
1202 } else if (op->type == REDIS_ZSET) {
1203 iterzset *it = &op->iter.zset;
1204 if (op->encoding == REDIS_ENCODING_ZIPLIST) {
1205 REDIS_NOTUSED(it); /* skip */
100ed062 1206 } else if (op->encoding == REDIS_ENCODING_SKIPLIST) {
56ce42fa
PN
1207 REDIS_NOTUSED(it); /* skip */
1208 } else {
1209 redisPanic("Unknown sorted set encoding");
1210 }
1211 } else {
1212 redisPanic("Unsupported type");
1213 }
1214}
1215
1216int zuiLength(zsetopsrc *op) {
1217 if (op->subject == NULL)
1218 return 0;
1219
1220 if (op->type == REDIS_SET) {
1221 iterset *it = &op->iter.set;
1222 if (op->encoding == REDIS_ENCODING_INTSET) {
1223 return intsetLen(it->is.is);
1224 } else if (op->encoding == REDIS_ENCODING_HT) {
1225 return dictSize(it->ht.dict);
1226 } else {
1227 redisPanic("Unknown set encoding");
1228 }
1229 } else if (op->type == REDIS_ZSET) {
1230 iterzset *it = &op->iter.zset;
1231 if (op->encoding == REDIS_ENCODING_ZIPLIST) {
1232 return zzlLength(it->zl.zl);
100ed062 1233 } else if (op->encoding == REDIS_ENCODING_SKIPLIST) {
56ce42fa
PN
1234 return it->sl.zs->zsl->length;
1235 } else {
1236 redisPanic("Unknown sorted set encoding");
1237 }
1238 } else {
1239 redisPanic("Unsupported type");
1240 }
1241}
1242
1243/* Check if the current value is valid. If so, store it in the passed structure
1244 * and move to the next element. If not valid, this means we have reached the
1245 * end of the structure and can abort. */
1246int zuiNext(zsetopsrc *op, zsetopval *val) {
1247 if (op->subject == NULL)
1248 return 0;
1249
1250 if (val->flags & OPVAL_DIRTY_ROBJ)
1251 decrRefCount(val->ele);
1252
1253 bzero(val,sizeof(zsetopval));
1254
1255 if (op->type == REDIS_SET) {
1256 iterset *it = &op->iter.set;
1257 if (op->encoding == REDIS_ENCODING_INTSET) {
a5dce407 1258 if (!intsetGet(it->is.is,it->is.ii,(int64_t*)&val->ell))
56ce42fa
PN
1259 return 0;
1260 val->score = 1.0;
1261
1262 /* Move to next element. */
1263 it->is.ii++;
1264 } else if (op->encoding == REDIS_ENCODING_HT) {
1265 if (it->ht.de == NULL)
1266 return 0;
c0ba9ebe 1267 val->ele = dictGetKey(it->ht.de);
56ce42fa
PN
1268 val->score = 1.0;
1269
1270 /* Move to next element. */
1271 it->ht.de = dictNext(it->ht.di);
1272 } else {
1273 redisPanic("Unknown set encoding");
1274 }
1275 } else if (op->type == REDIS_ZSET) {
1276 iterzset *it = &op->iter.zset;
1277 if (op->encoding == REDIS_ENCODING_ZIPLIST) {
1278 /* No need to check both, but better be explicit. */
1279 if (it->zl.eptr == NULL || it->zl.sptr == NULL)
1280 return 0;
1281 redisAssert(ziplistGet(it->zl.eptr,&val->estr,&val->elen,&val->ell));
1282 val->score = zzlGetScore(it->zl.sptr);
1283
1284 /* Move to next element. */
1285 zzlNext(it->zl.zl,&it->zl.eptr,&it->zl.sptr);
100ed062 1286 } else if (op->encoding == REDIS_ENCODING_SKIPLIST) {
56ce42fa
PN
1287 if (it->sl.node == NULL)
1288 return 0;
1289 val->ele = it->sl.node->obj;
1290 val->score = it->sl.node->score;
1291
1292 /* Move to next element. */
1293 it->sl.node = it->sl.node->level[0].forward;
1294 } else {
1295 redisPanic("Unknown sorted set encoding");
1296 }
1297 } else {
1298 redisPanic("Unsupported type");
1299 }
1300 return 1;
1301}
1302
1303int zuiLongLongFromValue(zsetopval *val) {
1304 if (!(val->flags & OPVAL_DIRTY_LL)) {
1305 val->flags |= OPVAL_DIRTY_LL;
1306
1307 if (val->ele != NULL) {
1308 if (val->ele->encoding == REDIS_ENCODING_INT) {
1309 val->ell = (long)val->ele->ptr;
1310 val->flags |= OPVAL_VALID_LL;
1311 } else if (val->ele->encoding == REDIS_ENCODING_RAW) {
1312 if (string2ll(val->ele->ptr,sdslen(val->ele->ptr),&val->ell))
1313 val->flags |= OPVAL_VALID_LL;
1314 } else {
1315 redisPanic("Unsupported element encoding");
1316 }
1317 } else if (val->estr != NULL) {
1318 if (string2ll((char*)val->estr,val->elen,&val->ell))
1319 val->flags |= OPVAL_VALID_LL;
1320 } else {
1321 /* The long long was already set, flag as valid. */
1322 val->flags |= OPVAL_VALID_LL;
1323 }
1324 }
1325 return val->flags & OPVAL_VALID_LL;
1326}
1327
1328robj *zuiObjectFromValue(zsetopval *val) {
1329 if (val->ele == NULL) {
1330 if (val->estr != NULL) {
1331 val->ele = createStringObject((char*)val->estr,val->elen);
1332 } else {
1333 val->ele = createStringObjectFromLongLong(val->ell);
1334 }
1335 val->flags |= OPVAL_DIRTY_ROBJ;
1336 }
1337 return val->ele;
1338}
1339
1340int zuiBufferFromValue(zsetopval *val) {
1341 if (val->estr == NULL) {
1342 if (val->ele != NULL) {
1343 if (val->ele->encoding == REDIS_ENCODING_INT) {
1344 val->elen = ll2string((char*)val->_buf,sizeof(val->_buf),(long)val->ele->ptr);
1345 val->estr = val->_buf;
1346 } else if (val->ele->encoding == REDIS_ENCODING_RAW) {
1347 val->elen = sdslen(val->ele->ptr);
1348 val->estr = val->ele->ptr;
1349 } else {
1350 redisPanic("Unsupported element encoding");
1351 }
1352 } else {
1353 val->elen = ll2string((char*)val->_buf,sizeof(val->_buf),val->ell);
1354 val->estr = val->_buf;
1355 }
1356 }
1357 return 1;
1358}
1359
1360/* Find value pointed to by val in the source pointer to by op. When found,
1361 * return 1 and store its score in target. Return 0 otherwise. */
1362int zuiFind(zsetopsrc *op, zsetopval *val, double *score) {
1363 if (op->subject == NULL)
1364 return 0;
1365
1366 if (op->type == REDIS_SET) {
1367 iterset *it = &op->iter.set;
1368
1369 if (op->encoding == REDIS_ENCODING_INTSET) {
1370 if (zuiLongLongFromValue(val) && intsetFind(it->is.is,val->ell)) {
1371 *score = 1.0;
1372 return 1;
1373 } else {
1374 return 0;
1375 }
1376 } else if (op->encoding == REDIS_ENCODING_HT) {
1377 zuiObjectFromValue(val);
1378 if (dictFind(it->ht.dict,val->ele) != NULL) {
1379 *score = 1.0;
1380 return 1;
1381 } else {
1382 return 0;
1383 }
1384 } else {
1385 redisPanic("Unknown set encoding");
1386 }
1387 } else if (op->type == REDIS_ZSET) {
1388 iterzset *it = &op->iter.zset;
1389 zuiObjectFromValue(val);
1390
1391 if (op->encoding == REDIS_ENCODING_ZIPLIST) {
8588bfa3 1392 if (zzlFind(it->zl.zl,val->ele,score) != NULL) {
56ce42fa
PN
1393 /* Score is already set by zzlFind. */
1394 return 1;
1395 } else {
1396 return 0;
1397 }
100ed062 1398 } else if (op->encoding == REDIS_ENCODING_SKIPLIST) {
56ce42fa
PN
1399 dictEntry *de;
1400 if ((de = dictFind(it->sl.zs->dict,val->ele)) != NULL) {
c0ba9ebe 1401 *score = *(double*)dictGetVal(de);
56ce42fa
PN
1402 return 1;
1403 } else {
1404 return 0;
1405 }
1406 } else {
1407 redisPanic("Unknown sorted set encoding");
1408 }
1409 } else {
1410 redisPanic("Unsupported type");
1411 }
1412}
1413
1414int zuiCompareByCardinality(const void *s1, const void *s2) {
1415 return zuiLength((zsetopsrc*)s1) - zuiLength((zsetopsrc*)s2);
e2641e09 1416}
1417
1418#define REDIS_AGGR_SUM 1
1419#define REDIS_AGGR_MIN 2
1420#define REDIS_AGGR_MAX 3
c0ba9ebe 1421#define zunionInterDictValue(_e) (dictGetVal(_e) == NULL ? 1.0 : *(double*)dictGetVal(_e))
e2641e09 1422
1423inline static void zunionInterAggregate(double *target, double val, int aggregate) {
1424 if (aggregate == REDIS_AGGR_SUM) {
1425 *target = *target + val;
d9e28bcf
PN
1426 /* The result of adding two doubles is NaN when one variable
1427 * is +inf and the other is -inf. When these numbers are added,
1428 * we maintain the convention of the result being 0.0. */
1429 if (isnan(*target)) *target = 0.0;
e2641e09 1430 } else if (aggregate == REDIS_AGGR_MIN) {
1431 *target = val < *target ? val : *target;
1432 } else if (aggregate == REDIS_AGGR_MAX) {
1433 *target = val > *target ? val : *target;
1434 } else {
1435 /* safety net */
1436 redisPanic("Unknown ZUNION/INTER aggregate type");
1437 }
1438}
1439
1440void zunionInterGenericCommand(redisClient *c, robj *dstkey, int op) {
1441 int i, j, setnum;
1442 int aggregate = REDIS_AGGR_SUM;
1443 zsetopsrc *src;
56ce42fa
PN
1444 zsetopval zval;
1445 robj *tmp;
255eebe2 1446 unsigned int maxelelen = 0;
e2641e09 1447 robj *dstobj;
1448 zset *dstzset;
69ef89f2 1449 zskiplistNode *znode;
8c1420ff 1450 int touched = 0;
e2641e09 1451
1452 /* expect setnum input keys to be given */
1453 setnum = atoi(c->argv[2]->ptr);
1454 if (setnum < 1) {
3ab20376
PN
1455 addReplyError(c,
1456 "at least 1 input key is needed for ZUNIONSTORE/ZINTERSTORE");
e2641e09 1457 return;
1458 }
1459
1460 /* test if the expected number of keys would overflow */
1461 if (3+setnum > c->argc) {
1462 addReply(c,shared.syntaxerr);
1463 return;
1464 }
1465
1466 /* read keys to be used for input */
56ce42fa 1467 src = zcalloc(sizeof(zsetopsrc) * setnum);
e2641e09 1468 for (i = 0, j = 3; i < setnum; i++, j++) {
1469 robj *obj = lookupKeyWrite(c->db,c->argv[j]);
56ce42fa
PN
1470 if (obj != NULL) {
1471 if (obj->type != REDIS_ZSET && obj->type != REDIS_SET) {
e2641e09 1472 zfree(src);
1473 addReply(c,shared.wrongtypeerr);
1474 return;
1475 }
56ce42fa
PN
1476
1477 src[i].subject = obj;
1478 src[i].type = obj->type;
1479 src[i].encoding = obj->encoding;
1480 } else {
1481 src[i].subject = NULL;
e2641e09 1482 }
1483
56ce42fa 1484 /* Default all weights to 1. */
e2641e09 1485 src[i].weight = 1.0;
1486 }
1487
1488 /* parse optional extra arguments */
1489 if (j < c->argc) {
1490 int remaining = c->argc - j;
1491
1492 while (remaining) {
1493 if (remaining >= (setnum + 1) && !strcasecmp(c->argv[j]->ptr,"weights")) {
1494 j++; remaining--;
1495 for (i = 0; i < setnum; i++, j++, remaining--) {
673e1fb7 1496 if (getDoubleFromObjectOrReply(c,c->argv[j],&src[i].weight,
d93f9a86 1497 "weight value is not a float") != REDIS_OK)
673e1fb7
PN
1498 {
1499 zfree(src);
e2641e09 1500 return;
673e1fb7 1501 }
e2641e09 1502 }
1503 } else if (remaining >= 2 && !strcasecmp(c->argv[j]->ptr,"aggregate")) {
1504 j++; remaining--;
1505 if (!strcasecmp(c->argv[j]->ptr,"sum")) {
1506 aggregate = REDIS_AGGR_SUM;
1507 } else if (!strcasecmp(c->argv[j]->ptr,"min")) {
1508 aggregate = REDIS_AGGR_MIN;
1509 } else if (!strcasecmp(c->argv[j]->ptr,"max")) {
1510 aggregate = REDIS_AGGR_MAX;
1511 } else {
1512 zfree(src);
1513 addReply(c,shared.syntaxerr);
1514 return;
1515 }
1516 j++; remaining--;
1517 } else {
1518 zfree(src);
1519 addReply(c,shared.syntaxerr);
1520 return;
1521 }
1522 }
1523 }
1524
56ce42fa
PN
1525 for (i = 0; i < setnum; i++)
1526 zuiInitIterator(&src[i]);
1527
e2641e09 1528 /* sort sets from the smallest to largest, this will improve our
1529 * algorithm's performance */
56ce42fa 1530 qsort(src,setnum,sizeof(zsetopsrc),zuiCompareByCardinality);
e2641e09 1531
1532 dstobj = createZsetObject();
1533 dstzset = dstobj->ptr;
02e60065 1534 memset(&zval, 0, sizeof(zval));
e2641e09 1535
1536 if (op == REDIS_OP_INTER) {
56ce42fa
PN
1537 /* Skip everything if the smallest input is empty. */
1538 if (zuiLength(&src[0]) > 0) {
1539 /* Precondition: as src[0] is non-empty and the inputs are ordered
1540 * by size, all src[i > 0] are non-empty too. */
1541 while (zuiNext(&src[0],&zval)) {
d433ebc6 1542 double score, value;
e2641e09 1543
56ce42fa 1544 score = src[0].weight * zval.score;
e2641e09 1545 for (j = 1; j < setnum; j++) {
d200342a 1546 /* It is not safe to access the zset we are
cb16b6c3 1547 * iterating, so explicitly check for equal object. */
d070abe4 1548 if (src[j].subject == src[0].subject) {
1549 value = zval.score*src[j].weight;
1550 zunionInterAggregate(&score,value,aggregate);
1551 } else if (zuiFind(&src[j],&zval,&value)) {
56ce42fa 1552 value *= src[j].weight;
d433ebc6 1553 zunionInterAggregate(&score,value,aggregate);
e2641e09 1554 } else {
1555 break;
1556 }
1557 }
1558
56ce42fa 1559 /* Only continue when present in every input. */
d433ebc6 1560 if (j == setnum) {
56ce42fa
PN
1561 tmp = zuiObjectFromValue(&zval);
1562 znode = zslInsert(dstzset->zsl,score,tmp);
1563 incrRefCount(tmp); /* added to skiplist */
1564 dictAdd(dstzset->dict,tmp,&znode->score);
1565 incrRefCount(tmp); /* added to dictionary */
255eebe2
PN
1566
1567 if (tmp->encoding == REDIS_ENCODING_RAW)
1568 if (sdslen(tmp->ptr) > maxelelen)
1569 maxelelen = sdslen(tmp->ptr);
e2641e09 1570 }
1571 }
e2641e09 1572 }
1573 } else if (op == REDIS_OP_UNION) {
1574 for (i = 0; i < setnum; i++) {
521ddcce 1575 if (zuiLength(&src[i]) == 0)
56ce42fa 1576 continue;
e2641e09 1577
56ce42fa 1578 while (zuiNext(&src[i],&zval)) {
d433ebc6
PN
1579 double score, value;
1580
56ce42fa
PN
1581 /* Skip key when already processed */
1582 if (dictFind(dstzset->dict,zuiObjectFromValue(&zval)) != NULL)
d433ebc6 1583 continue;
e2641e09 1584
56ce42fa
PN
1585 /* Initialize score */
1586 score = src[i].weight * zval.score;
e2641e09 1587
56ce42fa
PN
1588 /* Because the inputs are sorted by size, it's only possible
1589 * for sets at larger indices to hold this element. */
e2641e09 1590 for (j = (i+1); j < setnum; j++) {
d200342a 1591 /* It is not safe to access the zset we are
cb16b6c3 1592 * iterating, so explicitly check for equal object. */
1593 if(src[j].subject == src[i].subject) {
1594 value = zval.score*src[j].weight;
1595 zunionInterAggregate(&score,value,aggregate);
1596 } else if (zuiFind(&src[j],&zval,&value)) {
56ce42fa 1597 value *= src[j].weight;
d433ebc6 1598 zunionInterAggregate(&score,value,aggregate);
e2641e09 1599 }
1600 }
1601
56ce42fa
PN
1602 tmp = zuiObjectFromValue(&zval);
1603 znode = zslInsert(dstzset->zsl,score,tmp);
1604 incrRefCount(zval.ele); /* added to skiplist */
1605 dictAdd(dstzset->dict,tmp,&znode->score);
1606 incrRefCount(zval.ele); /* added to dictionary */
255eebe2
PN
1607
1608 if (tmp->encoding == REDIS_ENCODING_RAW)
1609 if (sdslen(tmp->ptr) > maxelelen)
1610 maxelelen = sdslen(tmp->ptr);
e2641e09 1611 }
e2641e09 1612 }
1613 } else {
56ce42fa 1614 redisPanic("Unknown operator");
e2641e09 1615 }
1616
56ce42fa
PN
1617 for (i = 0; i < setnum; i++)
1618 zuiClearIterator(&src[i]);
1619
8c1420ff 1620 if (dbDelete(c->db,dstkey)) {
cea8c5cd 1621 signalModifiedKey(c->db,dstkey);
8c1420ff 1622 touched = 1;
1623 server.dirty++;
1624 }
e2641e09 1625 if (dstzset->zsl->length) {
255eebe2
PN
1626 /* Convert to ziplist when in limits. */
1627 if (dstzset->zsl->length <= server.zset_max_ziplist_entries &&
1628 maxelelen <= server.zset_max_ziplist_value)
df26a0ae 1629 zsetConvert(dstobj,REDIS_ENCODING_ZIPLIST);
255eebe2 1630
e2641e09 1631 dbAdd(c->db,dstkey,dstobj);
df26a0ae 1632 addReplyLongLong(c,zsetLength(dstobj));
cea8c5cd 1633 if (!touched) signalModifiedKey(c->db,dstkey);
cbf7e107 1634 server.dirty++;
e2641e09 1635 } else {
1636 decrRefCount(dstobj);
255eebe2 1637 addReply(c,shared.czero);
e2641e09 1638 }
1639 zfree(src);
1640}
1641
1642void zunionstoreCommand(redisClient *c) {
1643 zunionInterGenericCommand(c,c->argv[1], REDIS_OP_UNION);
1644}
1645
1646void zinterstoreCommand(redisClient *c) {
1647 zunionInterGenericCommand(c,c->argv[1], REDIS_OP_INTER);
1648}
1649
1650void zrangeGenericCommand(redisClient *c, int reverse) {
5d1b4fb6
PN
1651 robj *key = c->argv[1];
1652 robj *zobj;
1653 int withscores = 0;
e2641e09 1654 long start;
1655 long end;
e2641e09 1656 int llen;
5d1b4fb6 1657 int rangelen;
e2641e09 1658
1659 if ((getLongFromObjectOrReply(c, c->argv[2], &start, NULL) != REDIS_OK) ||
1660 (getLongFromObjectOrReply(c, c->argv[3], &end, NULL) != REDIS_OK)) return;
1661
1662 if (c->argc == 5 && !strcasecmp(c->argv[4]->ptr,"withscores")) {
1663 withscores = 1;
1664 } else if (c->argc >= 5) {
1665 addReply(c,shared.syntaxerr);
1666 return;
1667 }
1668
5d1b4fb6
PN
1669 if ((zobj = lookupKeyReadOrReply(c,key,shared.emptymultibulk)) == NULL
1670 || checkType(c,zobj,REDIS_ZSET)) return;
e2641e09 1671
5d1b4fb6 1672 /* Sanitize indexes. */
df26a0ae 1673 llen = zsetLength(zobj);
e2641e09 1674 if (start < 0) start = llen+start;
1675 if (end < 0) end = llen+end;
1676 if (start < 0) start = 0;
e2641e09 1677
d0a4e24e
PN
1678 /* Invariant: start >= 0, so this test will be true when end < 0.
1679 * The range is empty when start > end or start >= length. */
e2641e09 1680 if (start > end || start >= llen) {
e2641e09 1681 addReply(c,shared.emptymultibulk);
1682 return;
1683 }
1684 if (end >= llen) end = llen-1;
1685 rangelen = (end-start)+1;
1686
e2641e09 1687 /* Return the result in form of a multi-bulk reply */
5d1b4fb6
PN
1688 addReplyMultiBulkLen(c, withscores ? (rangelen*2) : rangelen);
1689
1690 if (zobj->encoding == REDIS_ENCODING_ZIPLIST) {
1691 unsigned char *zl = zobj->ptr;
1692 unsigned char *eptr, *sptr;
1693 unsigned char *vstr;
1694 unsigned int vlen;
1695 long long vlong;
1696
1697 if (reverse)
1698 eptr = ziplistIndex(zl,-2-(2*start));
1699 else
1700 eptr = ziplistIndex(zl,2*start);
1701
eab0e26e 1702 redisAssertWithInfo(c,zobj,eptr != NULL);
4c5f0966
PN
1703 sptr = ziplistNext(zl,eptr);
1704
5d1b4fb6 1705 while (rangelen--) {
eab0e26e 1706 redisAssertWithInfo(c,zobj,eptr != NULL && sptr != NULL);
1707 redisAssertWithInfo(c,zobj,ziplistGet(eptr,&vstr,&vlen,&vlong));
5d1b4fb6
PN
1708 if (vstr == NULL)
1709 addReplyBulkLongLong(c,vlong);
1710 else
1711 addReplyBulkCBuffer(c,vstr,vlen);
1712
4c5f0966 1713 if (withscores)
5d1b4fb6 1714 addReplyDouble(c,zzlGetScore(sptr));
5d1b4fb6 1715
4c5f0966
PN
1716 if (reverse)
1717 zzlPrev(zl,&eptr,&sptr);
1718 else
1719 zzlNext(zl,&eptr,&sptr);
5d1b4fb6
PN
1720 }
1721
100ed062 1722 } else if (zobj->encoding == REDIS_ENCODING_SKIPLIST) {
5d1b4fb6
PN
1723 zset *zs = zobj->ptr;
1724 zskiplist *zsl = zs->zsl;
1725 zskiplistNode *ln;
1726 robj *ele;
1727
1728 /* Check if starting point is trivial, before doing log(N) lookup. */
1729 if (reverse) {
1730 ln = zsl->tail;
1731 if (start > 0)
1732 ln = zslGetElementByRank(zsl,llen-start);
1733 } else {
1734 ln = zsl->header->level[0].forward;
1735 if (start > 0)
1736 ln = zslGetElementByRank(zsl,start+1);
1737 }
1738
1739 while(rangelen--) {
eab0e26e 1740 redisAssertWithInfo(c,zobj,ln != NULL);
5d1b4fb6
PN
1741 ele = ln->obj;
1742 addReplyBulk(c,ele);
1743 if (withscores)
1744 addReplyDouble(c,ln->score);
1745 ln = reverse ? ln->backward : ln->level[0].forward;
1746 }
1747 } else {
1748 redisPanic("Unknown sorted set encoding");
e2641e09 1749 }
1750}
1751
1752void zrangeCommand(redisClient *c) {
1753 zrangeGenericCommand(c,0);
1754}
1755
1756void zrevrangeCommand(redisClient *c) {
1757 zrangeGenericCommand(c,1);
1758}
1759
0cfc8940
PN
1760/* This command implements ZRANGEBYSCORE, ZREVRANGEBYSCORE. */
1761void genericZrangebyscoreCommand(redisClient *c, int reverse) {
25bb8a44 1762 zrangespec range;
aff255c8 1763 robj *key = c->argv[1];
0cfc8940 1764 robj *zobj;
e2641e09 1765 int offset = 0, limit = -1;
1766 int withscores = 0;
25bb8a44
PN
1767 unsigned long rangelen = 0;
1768 void *replylen = NULL;
22b9bf15 1769 int minidx, maxidx;
e2641e09 1770
25bb8a44 1771 /* Parse the range arguments. */
22b9bf15
PN
1772 if (reverse) {
1773 /* Range is given as [max,min] */
1774 maxidx = 2; minidx = 3;
1775 } else {
1776 /* Range is given as [min,max] */
1777 minidx = 2; maxidx = 3;
1778 }
1779
1780 if (zslParseRange(c->argv[minidx],c->argv[maxidx],&range) != REDIS_OK) {
d93f9a86 1781 addReplyError(c,"min or max is not a float");
7236fdb2
PN
1782 return;
1783 }
25bb8a44
PN
1784
1785 /* Parse optional extra arguments. Note that ZCOUNT will exactly have
1786 * 4 arguments, so we'll never enter the following code path. */
1787 if (c->argc > 4) {
1788 int remaining = c->argc - 4;
1789 int pos = 4;
1790
1791 while (remaining) {
1792 if (remaining >= 1 && !strcasecmp(c->argv[pos]->ptr,"withscores")) {
1793 pos++; remaining--;
1794 withscores = 1;
1795 } else if (remaining >= 3 && !strcasecmp(c->argv[pos]->ptr,"limit")) {
1796 offset = atoi(c->argv[pos+1]->ptr);
1797 limit = atoi(c->argv[pos+2]->ptr);
1798 pos += 3; remaining -= 3;
1799 } else {
1800 addReply(c,shared.syntaxerr);
1801 return;
1802 }
1803 }
e2641e09 1804 }
25bb8a44
PN
1805
1806 /* Ok, lookup the key and get the range */
0cfc8940 1807 if ((zobj = lookupKeyReadOrReply(c,key,shared.emptymultibulk)) == NULL ||
aff255c8 1808 checkType(c,zobj,REDIS_ZSET)) return;
25bb8a44 1809
aff255c8
PN
1810 if (zobj->encoding == REDIS_ENCODING_ZIPLIST) {
1811 unsigned char *zl = zobj->ptr;
1812 unsigned char *eptr, *sptr;
1813 unsigned char *vstr;
1814 unsigned int vlen;
1815 long long vlong;
1816 double score;
e2641e09 1817
aff255c8 1818 /* If reversed, get the last node in range as starting point. */
0cfc8940 1819 if (reverse) {
8588bfa3 1820 eptr = zzlLastInRange(zl,range);
0cfc8940 1821 } else {
8588bfa3 1822 eptr = zzlFirstInRange(zl,range);
0cfc8940 1823 }
e2641e09 1824
aff255c8
PN
1825 /* No "first" element in the specified interval. */
1826 if (eptr == NULL) {
0cfc8940 1827 addReply(c, shared.emptymultibulk);
aff255c8
PN
1828 return;
1829 }
e2641e09 1830
aff255c8 1831 /* Get score pointer for the first element. */
eab0e26e 1832 redisAssertWithInfo(c,zobj,eptr != NULL);
aff255c8 1833 sptr = ziplistNext(zl,eptr);
e2641e09 1834
aff255c8
PN
1835 /* We don't know in advance how many matching elements there are in the
1836 * list, so we push this object that will represent the multi-bulk
1837 * length in the output buffer, and will "fix" it later */
0cfc8940 1838 replylen = addDeferredMultiBulkLength(c);
aff255c8
PN
1839
1840 /* If there is an offset, just traverse the number of elements without
1841 * checking the score because that is done in the next loop. */
0cfc8940
PN
1842 while (eptr && offset--) {
1843 if (reverse) {
aff255c8 1844 zzlPrev(zl,&eptr,&sptr);
0cfc8940 1845 } else {
aff255c8 1846 zzlNext(zl,&eptr,&sptr);
0cfc8940
PN
1847 }
1848 }
aff255c8
PN
1849
1850 while (eptr && limit--) {
1851 score = zzlGetScore(sptr);
1852
1853 /* Abort when the node is no longer in range. */
1854 if (reverse) {
1855 if (!zslValueGteMin(score,&range)) break;
1856 } else {
1857 if (!zslValueLteMax(score,&range)) break;
1858 }
1859
0cfc8940 1860 /* We know the element exists, so ziplistGet should always succeed */
eab0e26e 1861 redisAssertWithInfo(c,zobj,ziplistGet(eptr,&vstr,&vlen,&vlong));
0cfc8940 1862
aff255c8 1863 rangelen++;
0cfc8940
PN
1864 if (vstr == NULL) {
1865 addReplyBulkLongLong(c,vlong);
1866 } else {
1867 addReplyBulkCBuffer(c,vstr,vlen);
1868 }
1869
1870 if (withscores) {
1871 addReplyDouble(c,score);
aff255c8
PN
1872 }
1873
1874 /* Move to next node */
0cfc8940 1875 if (reverse) {
aff255c8 1876 zzlPrev(zl,&eptr,&sptr);
0cfc8940 1877 } else {
aff255c8 1878 zzlNext(zl,&eptr,&sptr);
0cfc8940 1879 }
e2641e09 1880 }
100ed062 1881 } else if (zobj->encoding == REDIS_ENCODING_SKIPLIST) {
aff255c8
PN
1882 zset *zs = zobj->ptr;
1883 zskiplist *zsl = zs->zsl;
1884 zskiplistNode *ln;
25bb8a44 1885
aff255c8 1886 /* If reversed, get the last node in range as starting point. */
0cfc8940 1887 if (reverse) {
aff255c8 1888 ln = zslLastInRange(zsl,range);
0cfc8940 1889 } else {
aff255c8 1890 ln = zslFirstInRange(zsl,range);
0cfc8940 1891 }
aff255c8
PN
1892
1893 /* No "first" element in the specified interval. */
1894 if (ln == NULL) {
0cfc8940 1895 addReply(c, shared.emptymultibulk);
aff255c8 1896 return;
25bb8a44
PN
1897 }
1898
aff255c8
PN
1899 /* We don't know in advance how many matching elements there are in the
1900 * list, so we push this object that will represent the multi-bulk
1901 * length in the output buffer, and will "fix" it later */
0cfc8940 1902 replylen = addDeferredMultiBulkLength(c);
aff255c8
PN
1903
1904 /* If there is an offset, just traverse the number of elements without
1905 * checking the score because that is done in the next loop. */
0cfc8940
PN
1906 while (ln && offset--) {
1907 if (reverse) {
1908 ln = ln->backward;
1909 } else {
1910 ln = ln->level[0].forward;
1911 }
1912 }
aff255c8
PN
1913
1914 while (ln && limit--) {
1915 /* Abort when the node is no longer in range. */
1916 if (reverse) {
1917 if (!zslValueGteMin(ln->score,&range)) break;
1918 } else {
1919 if (!zslValueLteMax(ln->score,&range)) break;
1920 }
1921
aff255c8 1922 rangelen++;
0cfc8940
PN
1923 addReplyBulk(c,ln->obj);
1924
1925 if (withscores) {
1926 addReplyDouble(c,ln->score);
aff255c8
PN
1927 }
1928
1929 /* Move to next node */
0cfc8940
PN
1930 if (reverse) {
1931 ln = ln->backward;
1932 } else {
1933 ln = ln->level[0].forward;
1934 }
aff255c8
PN
1935 }
1936 } else {
1937 redisPanic("Unknown sorted set encoding");
25bb8a44
PN
1938 }
1939
0cfc8940
PN
1940 if (withscores) {
1941 rangelen *= 2;
e2641e09 1942 }
0cfc8940
PN
1943
1944 setDeferredMultiBulkLength(c, replylen, rangelen);
e2641e09 1945}
1946
1947void zrangebyscoreCommand(redisClient *c) {
0cfc8940 1948 genericZrangebyscoreCommand(c,0);
25bb8a44
PN
1949}
1950
1951void zrevrangebyscoreCommand(redisClient *c) {
0cfc8940 1952 genericZrangebyscoreCommand(c,1);
e2641e09 1953}
1954
1955void zcountCommand(redisClient *c) {
62d774e5
PN
1956 robj *key = c->argv[1];
1957 robj *zobj;
1958 zrangespec range;
1959 int count = 0;
1960
1961 /* Parse the range arguments */
1962 if (zslParseRange(c->argv[2],c->argv[3],&range) != REDIS_OK) {
d93f9a86 1963 addReplyError(c,"min or max is not a float");
62d774e5
PN
1964 return;
1965 }
1966
1967 /* Lookup the sorted set */
1968 if ((zobj = lookupKeyReadOrReply(c, key, shared.czero)) == NULL ||
1969 checkType(c, zobj, REDIS_ZSET)) return;
1970
1971 if (zobj->encoding == REDIS_ENCODING_ZIPLIST) {
1972 unsigned char *zl = zobj->ptr;
1973 unsigned char *eptr, *sptr;
1974 double score;
1975
1976 /* Use the first element in range as the starting point */
1977 eptr = zzlFirstInRange(zl,range);
1978
1979 /* No "first" element */
1980 if (eptr == NULL) {
1981 addReply(c, shared.czero);
1982 return;
1983 }
1984
1985 /* First element is in range */
1986 sptr = ziplistNext(zl,eptr);
1987 score = zzlGetScore(sptr);
eab0e26e 1988 redisAssertWithInfo(c,zobj,zslValueLteMax(score,&range));
62d774e5
PN
1989
1990 /* Iterate over elements in range */
1991 while (eptr) {
1992 score = zzlGetScore(sptr);
1993
1994 /* Abort when the node is no longer in range. */
1995 if (!zslValueLteMax(score,&range)) {
1996 break;
1997 } else {
1998 count++;
1999 zzlNext(zl,&eptr,&sptr);
2000 }
2001 }
2002 } else if (zobj->encoding == REDIS_ENCODING_SKIPLIST) {
2003 zset *zs = zobj->ptr;
2004 zskiplist *zsl = zs->zsl;
2005 zskiplistNode *zn;
2006 unsigned long rank;
2007
2008 /* Find first element in range */
2009 zn = zslFirstInRange(zsl, range);
2010
2011 /* Use rank of first element, if any, to determine preliminary count */
2012 if (zn != NULL) {
2013 rank = zslGetRank(zsl, zn->score, zn->obj);
2014 count = (zsl->length - (rank - 1));
2015
2016 /* Find last element in range */
2017 zn = zslLastInRange(zsl, range);
2018
2019 /* Use rank of last element, if any, to determine the actual count */
2020 if (zn != NULL) {
2021 rank = zslGetRank(zsl, zn->score, zn->obj);
2022 count -= (zsl->length - rank);
2023 }
2024 }
2025 } else {
2026 redisPanic("Unknown sorted set encoding");
2027 }
2028
2029 addReplyLongLong(c, count);
e2641e09 2030}
2031
2032void zcardCommand(redisClient *c) {
d1c920c5
PN
2033 robj *key = c->argv[1];
2034 robj *zobj;
e2641e09 2035
d1c920c5
PN
2036 if ((zobj = lookupKeyReadOrReply(c,key,shared.czero)) == NULL ||
2037 checkType(c,zobj,REDIS_ZSET)) return;
e2641e09 2038
df26a0ae 2039 addReplyLongLong(c,zsetLength(zobj));
e2641e09 2040}
2041
2042void zscoreCommand(redisClient *c) {
d1c920c5
PN
2043 robj *key = c->argv[1];
2044 robj *zobj;
2045 double score;
e2641e09 2046
d1c920c5
PN
2047 if ((zobj = lookupKeyReadOrReply(c,key,shared.nullbulk)) == NULL ||
2048 checkType(c,zobj,REDIS_ZSET)) return;
e2641e09 2049
d1c920c5 2050 if (zobj->encoding == REDIS_ENCODING_ZIPLIST) {
8588bfa3 2051 if (zzlFind(zobj->ptr,c->argv[2],&score) != NULL)
d1c920c5
PN
2052 addReplyDouble(c,score);
2053 else
2054 addReply(c,shared.nullbulk);
100ed062 2055 } else if (zobj->encoding == REDIS_ENCODING_SKIPLIST) {
d1c920c5
PN
2056 zset *zs = zobj->ptr;
2057 dictEntry *de;
e2641e09 2058
d1c920c5
PN
2059 c->argv[2] = tryObjectEncoding(c->argv[2]);
2060 de = dictFind(zs->dict,c->argv[2]);
2061 if (de != NULL) {
c0ba9ebe 2062 score = *(double*)dictGetVal(de);
d1c920c5
PN
2063 addReplyDouble(c,score);
2064 } else {
2065 addReply(c,shared.nullbulk);
2066 }
2067 } else {
2068 redisPanic("Unknown sorted set encoding");
e2641e09 2069 }
2070}
2071
2072void zrankGenericCommand(redisClient *c, int reverse) {
d1c920c5
PN
2073 robj *key = c->argv[1];
2074 robj *ele = c->argv[2];
2075 robj *zobj;
2076 unsigned long llen;
e2641e09 2077 unsigned long rank;
e2641e09 2078
d1c920c5
PN
2079 if ((zobj = lookupKeyReadOrReply(c,key,shared.nullbulk)) == NULL ||
2080 checkType(c,zobj,REDIS_ZSET)) return;
df26a0ae 2081 llen = zsetLength(zobj);
e2641e09 2082
eab0e26e 2083 redisAssertWithInfo(c,ele,ele->encoding == REDIS_ENCODING_RAW);
d1c920c5
PN
2084 if (zobj->encoding == REDIS_ENCODING_ZIPLIST) {
2085 unsigned char *zl = zobj->ptr;
2086 unsigned char *eptr, *sptr;
e2641e09 2087
d1c920c5 2088 eptr = ziplistIndex(zl,0);
eab0e26e 2089 redisAssertWithInfo(c,zobj,eptr != NULL);
d1c920c5 2090 sptr = ziplistNext(zl,eptr);
eab0e26e 2091 redisAssertWithInfo(c,zobj,sptr != NULL);
d1c920c5
PN
2092
2093 rank = 1;
2094 while(eptr != NULL) {
2095 if (ziplistCompare(eptr,ele->ptr,sdslen(ele->ptr)))
2096 break;
2097 rank++;
2098 zzlNext(zl,&eptr,&sptr);
2099 }
2100
2101 if (eptr != NULL) {
2102 if (reverse)
2103 addReplyLongLong(c,llen-rank);
2104 else
2105 addReplyLongLong(c,rank-1);
e2641e09 2106 } else {
d1c920c5
PN
2107 addReply(c,shared.nullbulk);
2108 }
100ed062 2109 } else if (zobj->encoding == REDIS_ENCODING_SKIPLIST) {
d1c920c5
PN
2110 zset *zs = zobj->ptr;
2111 zskiplist *zsl = zs->zsl;
2112 dictEntry *de;
2113 double score;
2114
2115 ele = c->argv[2] = tryObjectEncoding(c->argv[2]);
2116 de = dictFind(zs->dict,ele);
2117 if (de != NULL) {
c0ba9ebe 2118 score = *(double*)dictGetVal(de);
d1c920c5 2119 rank = zslGetRank(zsl,score,ele);
eab0e26e 2120 redisAssertWithInfo(c,ele,rank); /* Existing elements always have a rank. */
d1c920c5
PN
2121 if (reverse)
2122 addReplyLongLong(c,llen-rank);
2123 else
2124 addReplyLongLong(c,rank-1);
2125 } else {
2126 addReply(c,shared.nullbulk);
e2641e09 2127 }
2128 } else {
d1c920c5 2129 redisPanic("Unknown sorted set encoding");
e2641e09 2130 }
2131}
2132
2133void zrankCommand(redisClient *c) {
2134 zrankGenericCommand(c, 0);
2135}
2136
2137void zrevrankCommand(redisClient *c) {
2138 zrankGenericCommand(c, 1);
2139}