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