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 26 zskiplistNode *zslCreateNode(int level, double score, robj *obj) { 27 zskiplistNode *zn = zmalloc(sizeof(*zn)+level*sizeof(struct zskiplistLevel)); 28 zn->score = score; 29 zn->obj = obj; 30 return zn; 31 } 32 33 zskiplist *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++) { 42 zsl->header->level[j].forward = NULL; 43 zsl->header->level[j].span = 0; 44 } 45 zsl->header->backward = NULL; 46 zsl->tail = NULL; 47 return zsl; 48 } 49 50 void zslFreeNode(zskiplistNode *node) { 51 decrRefCount(node->obj); 52 zfree(node); 53 } 54 55 void zslFree(zskiplist *zsl) { 56 zskiplistNode *node = zsl->header->level[0].forward, *next; 57 58 zfree(zsl->header); 59 while(node) { 60 next = node->level[0].forward; 61 zslFreeNode(node); 62 node = next; 63 } 64 zfree(zsl); 65 } 66 67 int 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 74 zskiplistNode *zslInsert(zskiplist *zsl, double score, robj *obj) { 75 zskiplistNode *update[ZSKIPLIST_MAXLEVEL], *x; 76 unsigned int rank[ZSKIPLIST_MAXLEVEL]; 77 int i, level; 78 79 redisAssert(!isnan(score)); 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]; 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; 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; 102 update[i]->level[i].span = zsl->length; 103 } 104 zsl->level = level; 105 } 106 x = zslCreateNode(level,score,obj); 107 for (i = 0; i < level; i++) { 108 x->level[i].forward = update[i]->level[i].forward; 109 update[i]->level[i].forward = x; 110 111 /* update span covered by update[i] as x is inserted here */ 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; 114 } 115 116 /* increment span for untouched levels */ 117 for (i = level; i < zsl->level; i++) { 118 update[i]->level[i].span++; 119 } 120 121 x->backward = (update[0] == zsl->header) ? NULL : update[0]; 122 if (x->level[0].forward) 123 x->level[0].forward->backward = x; 124 else 125 zsl->tail = x; 126 zsl->length++; 127 return x; 128 } 129 130 /* Internal function used by zslDelete, zslDeleteByScore and zslDeleteByRank */ 131 void zslDeleteNode(zskiplist *zsl, zskiplistNode *x, zskiplistNode **update) { 132 int i; 133 for (i = 0; i < zsl->level; i++) { 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; 137 } else { 138 update[i]->level[i].span -= 1; 139 } 140 } 141 if (x->level[0].forward) { 142 x->level[0].forward->backward = x->backward; 143 } else { 144 zsl->tail = x->backward; 145 } 146 while(zsl->level > 1 && zsl->header->level[zsl->level-1].forward == NULL) 147 zsl->level--; 148 zsl->length--; 149 } 150 151 /* Delete an element with matching score/object from the skiplist. */ 152 int 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--) { 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; 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. */ 167 x = x->level[0].forward; 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 178 static int zslValueGteMin(double value, zrangespec *spec) { 179 return spec->minex ? (value > spec->min) : (value >= spec->min); 180 } 181 182 static int zslValueLteMax(double value, zrangespec *spec) { 183 return spec->maxex ? (value < spec->max) : (value <= spec->max); 184 } 185 186 /* Returns if there is a part of the zset is in range. */ 187 int zslIsInRange(zskiplist *zsl, zrangespec *range) { 188 zskiplistNode *x; 189 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; 194 x = zsl->tail; 195 if (x == NULL || !zslValueGteMin(x->score,range)) 196 return 0; 197 x = zsl->header->level[0].forward; 198 if (x == NULL || !zslValueLteMax(x->score,range)) 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. */ 205 zskiplistNode *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 && 216 !zslValueGteMin(x->level[i].forward->score,&range)) 217 x = x->level[i].forward; 218 } 219 220 /* This is an inner range, so the next node cannot be NULL. */ 221 x = x->level[0].forward; 222 redisAssert(x != NULL); 223 224 /* Check if score <= max. */ 225 if (!zslValueLteMax(x->score,&range)) return NULL; 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. */ 231 zskiplistNode *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 && 242 zslValueLteMax(x->level[i].forward->score,&range)) 243 x = x->level[i].forward; 244 } 245 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; 251 return x; 252 } 253 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. */ 258 unsigned long zslDeleteRangeByScore(zskiplist *zsl, zrangespec range, dict *dict) { 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--) { 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; 269 update[i] = x; 270 } 271 272 /* Current node is the last with score < or <= min. */ 273 x = x->level[0].forward; 274 275 /* Delete nodes while in range. */ 276 while (x && (range.maxex ? x->score < range.max : x->score <= range.max)) { 277 zskiplistNode *next = x->level[0].forward; 278 zslDeleteNode(zsl,x,update); 279 dictDelete(dict,x->obj); 280 zslFreeNode(x); 281 removed++; 282 x = next; 283 } 284 return removed; 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 */ 289 unsigned 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--) { 296 while (x->level[i].forward && (traversed + x->level[i].span) < start) { 297 traversed += x->level[i].span; 298 x = x->level[i].forward; 299 } 300 update[i] = x; 301 } 302 303 traversed++; 304 x = x->level[0].forward; 305 while (x && traversed <= end) { 306 zskiplistNode *next = x->level[0].forward; 307 zslDeleteNode(zsl,x,update); 308 dictDelete(dict,x->obj); 309 zslFreeNode(x); 310 removed++; 311 traversed++; 312 x = next; 313 } 314 return removed; 315 } 316 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. */ 321 unsigned long zslGetRank(zskiplist *zsl, double score, robj *o) { 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--) { 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; 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. */ 345 zskiplistNode* zslGetElementByRank(zskiplist *zsl, unsigned long rank) { 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--) { 352 while (x->level[i].forward && (traversed + x->level[i].span) <= rank) 353 { 354 traversed += x->level[i].span; 355 x = x->level[i].forward; 356 } 357 if (traversed == rank) { 358 return x; 359 } 360 } 361 return NULL; 362 } 363 364 /* Populate the rangespec according to the objects min and max. */ 365 static int zslParseRange(robj *min, robj *max, zrangespec *spec) { 366 char *eptr; 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] == '(') { 377 spec->min = strtod((char*)min->ptr+1,&eptr); 378 if (eptr[0] != '\0' || isnan(spec->min)) return REDIS_ERR; 379 spec->minex = 1; 380 } else { 381 spec->min = strtod((char*)min->ptr,&eptr); 382 if (eptr[0] != '\0' || isnan(spec->min)) return REDIS_ERR; 383 } 384 } 385 if (max->encoding == REDIS_ENCODING_INT) { 386 spec->max = (long)max->ptr; 387 } else { 388 if (((char*)max->ptr)[0] == '(') { 389 spec->max = strtod((char*)max->ptr+1,&eptr); 390 if (eptr[0] != '\0' || isnan(spec->max)) return REDIS_ERR; 391 spec->maxex = 1; 392 } else { 393 spec->max = strtod((char*)max->ptr,&eptr); 394 if (eptr[0] != '\0' || isnan(spec->max)) return REDIS_ERR; 395 } 396 } 397 398 return REDIS_OK; 399 } 400 401 /*----------------------------------------------------------------------------- 402 * Ziplist-backed sorted set API 403 *----------------------------------------------------------------------------*/ 404 405 double 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. */ 427 int 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 447 unsigned int zzlLength(unsigned char *zl) { 448 return ziplistLen(zl)/2; 449 } 450 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. */ 453 void 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. */ 472 void 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 489 /* Returns if there is a part of the zset is in range. Should only be used 490 * internally by zzlFirstInRange and zzlLastInRange. */ 491 int 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. */ 517 unsigned char *zzlFirstInRange(unsigned char *zl, zrangespec range) { 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); 529 if (zslValueGteMin(score,&range)) { 530 /* Check if score <= max. */ 531 if (zslValueLteMax(score,&range)) 532 return eptr; 533 return NULL; 534 } 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. */ 545 unsigned char *zzlLastInRange(unsigned char *zl, zrangespec range) { 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); 557 if (zslValueLteMax(score,&range)) { 558 /* Check if score >= min. */ 559 if (zslValueGteMin(score,&range)) 560 return eptr; 561 return NULL; 562 } 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 576 unsigned char *zzlFind(unsigned char *zl, robj *ele, double *score) { 577 unsigned char *eptr = ziplistIndex(zl,0), *sptr; 578 579 ele = getDecodedObject(ele); 580 while (eptr != NULL) { 581 sptr = ziplistNext(zl,eptr); 582 redisAssertWithInfo(NULL,ele,sptr != NULL); 583 584 if (ziplistCompare(eptr,ele->ptr,sdslen(ele->ptr))) { 585 /* Matching element, pull out score. */ 586 if (score != NULL) *score = zzlGetScore(sptr); 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. */ 601 unsigned char *zzlDelete(unsigned char *zl, unsigned char *eptr) { 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); 607 return zl; 608 } 609 610 unsigned char *zzlInsertAt(unsigned char *zl, unsigned char *eptr, robj *ele, double score) { 611 unsigned char *sptr; 612 char scorebuf[128]; 613 int scorelen; 614 size_t offset; 615 616 redisAssertWithInfo(NULL,ele,ele->encoding == REDIS_ENCODING_RAW); 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. */ 628 redisAssertWithInfo(NULL,ele,(sptr = ziplistNext(zl,eptr)) != NULL); 629 zl = ziplistInsert(zl,sptr,(unsigned char*)scorebuf,scorelen); 630 } 631 632 return zl; 633 } 634 635 /* Insert (element,score) pair in ziplist. This function assumes the element is 636 * not yet present in the list. */ 637 unsigned char *zzlInsert(unsigned char *zl, robj *ele, double score) { 638 unsigned char *eptr = ziplistIndex(zl,0), *sptr; 639 double s; 640 641 ele = getDecodedObject(ele); 642 while (eptr != NULL) { 643 sptr = ziplistNext(zl,eptr); 644 redisAssertWithInfo(NULL,ele,sptr != NULL); 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. */ 651 zl = zzlInsertAt(zl,eptr,ele,score); 652 break; 653 } else if (s == score) { 654 /* Ensure lexicographical ordering for elements. */ 655 if (zzlCompareElements(eptr,ele->ptr,sdslen(ele->ptr)) > 0) { 656 zl = zzlInsertAt(zl,eptr,ele,score); 657 break; 658 } 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. */ 666 if (eptr == NULL) 667 zl = zzlInsertAt(zl,NULL,ele,score); 668 669 decrRefCount(ele); 670 return zl; 671 } 672 673 unsigned char *zzlDeleteRangeByScore(unsigned char *zl, zrangespec range, unsigned long *deleted) { 674 unsigned char *eptr, *sptr; 675 double score; 676 unsigned long num = 0; 677 678 if (deleted != NULL) *deleted = 0; 679 680 eptr = zzlFirstInRange(zl,range); 681 if (eptr == NULL) return zl; 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); 691 num++; 692 } else { 693 /* No longer in range. */ 694 break; 695 } 696 } 697 698 if (deleted != NULL) *deleted = num; 699 return zl; 700 } 701 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 */ 704 unsigned char *zzlDeleteRangeByRank(unsigned char *zl, unsigned int start, unsigned int end, unsigned long *deleted) { 705 unsigned int num = (end-start)+1; 706 if (deleted) *deleted = num; 707 zl = ziplistDeleteRange(zl,2*(start-1),2*num); 708 return zl; 709 } 710 711 /*----------------------------------------------------------------------------- 712 * Common sorted set API 713 *----------------------------------------------------------------------------*/ 714 715 unsigned int zsetLength(robj *zobj) { 716 int length = -1; 717 if (zobj->encoding == REDIS_ENCODING_ZIPLIST) { 718 length = zzlLength(zobj->ptr); 719 } else if (zobj->encoding == REDIS_ENCODING_SKIPLIST) { 720 length = ((zset*)zobj->ptr)->zsl->length; 721 } else { 722 redisPanic("Unknown sorted set encoding"); 723 } 724 return length; 725 } 726 727 void zsetConvert(robj *zobj, int encoding) { 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 741 if (encoding != REDIS_ENCODING_SKIPLIST) 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); 749 redisAssertWithInfo(NULL,zobj,eptr != NULL); 750 sptr = ziplistNext(zl,eptr); 751 redisAssertWithInfo(NULL,zobj,sptr != NULL); 752 753 while (eptr != NULL) { 754 score = zzlGetScore(sptr); 755 redisAssertWithInfo(NULL,zobj,ziplistGet(eptr,&vstr,&vlen,&vlong)); 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); 763 redisAssertWithInfo(NULL,zobj,dictAdd(zs->dict,ele,&node->score) == DICT_OK); 764 incrRefCount(ele); /* Added to dictionary. */ 765 zzlNext(zl,&eptr,&sptr); 766 } 767 768 zfree(zobj->ptr); 769 zobj->ptr = zs; 770 zobj->encoding = REDIS_ENCODING_SKIPLIST; 771 } else if (zobj->encoding == REDIS_ENCODING_SKIPLIST) { 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 785 while (node) { 786 ele = getDecodedObject(node->obj); 787 zl = zzlInsertAt(zl,NULL,ele,node->score); 788 decrRefCount(ele); 789 790 next = node->level[0].forward; 791 zslFreeNode(node); 792 node = next; 793 } 794 795 zfree(zs); 796 zobj->ptr = zl; 797 zobj->encoding = REDIS_ENCODING_ZIPLIST; 798 } else { 799 redisPanic("Unknown sorted set encoding"); 800 } 801 } 802 803 /*----------------------------------------------------------------------------- 804 * Sorted set commands 805 *----------------------------------------------------------------------------*/ 806 807 /* This generic command implements both ZADD and ZINCRBY. */ 808 void zaddGenericCommand(redisClient *c, int incr) { 809 static char *nanerr = "resulting score is not a number (NaN)"; 810 robj *key = c->argv[1]; 811 robj *ele; 812 robj *zobj; 813 robj *curobj; 814 double score = 0, *scores, curscore = 0.0; 815 int j, elements = (c->argc-2)/2; 816 int added = 0; 817 818 if (c->argc % 2) { 819 addReply(c,shared.syntaxerr); 820 return; 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 } 835 836 /* Lookup the key and create the sorted set if does not exist. */ 837 zobj = lookupKeyWrite(c->db,key); 838 if (zobj == NULL) { 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 } 846 dbAdd(c->db,key,zobj); 847 } else { 848 if (zobj->type != REDIS_ZSET) { 849 addReply(c,shared.wrongtypeerr); 850 zfree(scores); 851 return; 852 } 853 } 854 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 } 873 } 874 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. */ 886 zobj->ptr = zzlInsert(zobj->ptr,ele,score); 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); 891 892 signalModifiedKey(c->db,key); 893 server.dirty++; 894 if (!incr) added++; 895 } 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) { 904 curobj = dictGetKey(de); 905 curscore = *(double*)dictGetVal(de); 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 } 916 } 917 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) { 922 redisAssertWithInfo(c,curobj,zslDelete(zs->zsl,curscore,curobj)); 923 znode = zslInsert(zs->zsl,score,curobj); 924 incrRefCount(curobj); /* Re-inserted in skiplist. */ 925 dictGetVal(de) = &znode->score; /* Update score ptr. */ 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. */ 933 redisAssertWithInfo(c,NULL,dictAdd(zs->dict,ele,&znode->score) == DICT_OK); 934 incrRefCount(ele); /* Added to dictionary. */ 935 936 signalModifiedKey(c->db,key); 937 server.dirty++; 938 if (!incr) added++; 939 } 940 } else { 941 redisPanic("Unknown sorted set encoding"); 942 } 943 } 944 zfree(scores); 945 if (incr) /* ZINCRBY */ 946 addReplyDouble(c,score); 947 else /* ZADD */ 948 addReplyLongLong(c,added); 949 } 950 951 void zaddCommand(redisClient *c) { 952 zaddGenericCommand(c,0); 953 } 954 955 void zincrbyCommand(redisClient *c) { 956 zaddGenericCommand(c,1); 957 } 958 959 void zremCommand(redisClient *c) { 960 robj *key = c->argv[1]; 961 robj *zobj; 962 int deleted = 0, j; 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 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 } 979 } 980 } else if (zobj->encoding == REDIS_ENCODING_SKIPLIST) { 981 zset *zs = zobj->ptr; 982 dictEntry *de; 983 double score; 984 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 */ 991 score = *(double*)dictGetVal(de); 992 redisAssertWithInfo(c,c->argv[j],zslDelete(zs->zsl,score,c->argv[j])); 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 } 1002 } 1003 } else { 1004 redisPanic("Unknown sorted set encoding"); 1005 } 1006 1007 if (deleted) { 1008 signalModifiedKey(c->db,key); 1009 server.dirty += deleted; 1010 } 1011 addReplyLongLong(c,deleted); 1012 } 1013 1014 void zremrangebyscoreCommand(redisClient *c) { 1015 robj *key = c->argv[1]; 1016 robj *zobj; 1017 zrangespec range; 1018 unsigned long deleted; 1019 1020 /* Parse the range arguments. */ 1021 if (zslParseRange(c->argv[2],c->argv[3],&range) != REDIS_OK) { 1022 addReplyError(c,"min or max is not a float"); 1023 return; 1024 } 1025 1026 if ((zobj = lookupKeyWriteOrReply(c,key,shared.czero)) == NULL || 1027 checkType(c,zobj,REDIS_ZSET)) return; 1028 1029 if (zobj->encoding == REDIS_ENCODING_ZIPLIST) { 1030 zobj->ptr = zzlDeleteRangeByScore(zobj->ptr,range,&deleted); 1031 if (zzlLength(zobj->ptr) == 0) dbDelete(c->db,key); 1032 } else if (zobj->encoding == REDIS_ENCODING_SKIPLIST) { 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 } 1040 1041 if (deleted) signalModifiedKey(c->db,key); 1042 server.dirty += deleted; 1043 addReplyLongLong(c,deleted); 1044 } 1045 1046 void zremrangebyrankCommand(redisClient *c) { 1047 robj *key = c->argv[1]; 1048 robj *zobj; 1049 long start; 1050 long end; 1051 int llen; 1052 unsigned long deleted; 1053 1054 if ((getLongFromObjectOrReply(c, c->argv[2], &start, NULL) != REDIS_OK) || 1055 (getLongFromObjectOrReply(c, c->argv[3], &end, NULL) != REDIS_OK)) return; 1056 1057 if ((zobj = lookupKeyWriteOrReply(c,key,shared.czero)) == NULL || 1058 checkType(c,zobj,REDIS_ZSET)) return; 1059 1060 /* Sanitize indexes. */ 1061 llen = zsetLength(zobj); 1062 if (start < 0) start = llen+start; 1063 if (end < 0) end = llen+end; 1064 if (start < 0) start = 0; 1065 1066 /* Invariant: start >= 0, so this test will be true when end < 0. 1067 * The range is empty when start > end or start >= length. */ 1068 if (start > end || start >= llen) { 1069 addReply(c,shared.czero); 1070 return; 1071 } 1072 if (end >= llen) end = llen-1; 1073 1074 if (zobj->encoding == REDIS_ENCODING_ZIPLIST) { 1075 /* Correct for 1-based rank. */ 1076 zobj->ptr = zzlDeleteRangeByRank(zobj->ptr,start+1,end+1,&deleted); 1077 if (zzlLength(zobj->ptr) == 0) dbDelete(c->db,key); 1078 } else if (zobj->encoding == REDIS_ENCODING_SKIPLIST) { 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); 1090 server.dirty += deleted; 1091 addReplyLongLong(c,deleted); 1092 } 1093 1094 typedef struct { 1095 robj *subject; 1096 int type; /* Set, sorted set */ 1097 int encoding; 1098 double weight; 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; 1126 } zsetopsrc; 1127 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. */ 1140 typedef 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 1150 typedef union _iterset iterset; 1151 typedef union _iterzset iterzset; 1152 1153 void 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 } 1178 } else if (op->encoding == REDIS_ENCODING_SKIPLIST) { 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 1189 void 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 */ 1206 } else if (op->encoding == REDIS_ENCODING_SKIPLIST) { 1207 REDIS_NOTUSED(it); /* skip */ 1208 } else { 1209 redisPanic("Unknown sorted set encoding"); 1210 } 1211 } else { 1212 redisPanic("Unsupported type"); 1213 } 1214 } 1215 1216 int 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); 1233 } else if (op->encoding == REDIS_ENCODING_SKIPLIST) { 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. */ 1246 int 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) { 1258 if (!intsetGet(it->is.is,it->is.ii,(int64_t*)&val->ell)) 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; 1267 val->ele = dictGetKey(it->ht.de); 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); 1286 } else if (op->encoding == REDIS_ENCODING_SKIPLIST) { 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 1303 int 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 1328 robj *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 1340 int 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. */ 1362 int 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) { 1392 if (zzlFind(it->zl.zl,val->ele,score) != NULL) { 1393 /* Score is already set by zzlFind. */ 1394 return 1; 1395 } else { 1396 return 0; 1397 } 1398 } else if (op->encoding == REDIS_ENCODING_SKIPLIST) { 1399 dictEntry *de; 1400 if ((de = dictFind(it->sl.zs->dict,val->ele)) != NULL) { 1401 *score = *(double*)dictGetVal(de); 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 1414 int zuiCompareByCardinality(const void *s1, const void *s2) { 1415 return zuiLength((zsetopsrc*)s1) - zuiLength((zsetopsrc*)s2); 1416 } 1417 1418 #define REDIS_AGGR_SUM 1 1419 #define REDIS_AGGR_MIN 2 1420 #define REDIS_AGGR_MAX 3 1421 #define zunionInterDictValue(_e) (dictGetVal(_e) == NULL ? 1.0 : *(double*)dictGetVal(_e)) 1422 1423 inline static void zunionInterAggregate(double *target, double val, int aggregate) { 1424 if (aggregate == REDIS_AGGR_SUM) { 1425 *target = *target + val; 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; 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 1440 void zunionInterGenericCommand(redisClient *c, robj *dstkey, int op) { 1441 int i, j; 1442 long setnum; 1443 int aggregate = REDIS_AGGR_SUM; 1444 zsetopsrc *src; 1445 zsetopval zval; 1446 robj *tmp; 1447 unsigned int maxelelen = 0; 1448 robj *dstobj; 1449 zset *dstzset; 1450 zskiplistNode *znode; 1451 int touched = 0; 1452 1453 /* expect setnum input keys to be given */ 1454 if ((getLongFromObjectOrReply(c, c->argv[2], &setnum, NULL) != REDIS_OK)) 1455 return; 1456 1457 if (setnum < 1) { 1458 addReplyError(c, 1459 "at least 1 input key is needed for ZUNIONSTORE/ZINTERSTORE"); 1460 return; 1461 } 1462 1463 /* test if the expected number of keys would overflow */ 1464 if (3+setnum > c->argc) { 1465 addReply(c,shared.syntaxerr); 1466 return; 1467 } 1468 1469 /* read keys to be used for input */ 1470 src = zcalloc(sizeof(zsetopsrc) * setnum); 1471 for (i = 0, j = 3; i < setnum; i++, j++) { 1472 robj *obj = lookupKeyWrite(c->db,c->argv[j]); 1473 if (obj != NULL) { 1474 if (obj->type != REDIS_ZSET && obj->type != REDIS_SET) { 1475 zfree(src); 1476 addReply(c,shared.wrongtypeerr); 1477 return; 1478 } 1479 1480 src[i].subject = obj; 1481 src[i].type = obj->type; 1482 src[i].encoding = obj->encoding; 1483 } else { 1484 src[i].subject = NULL; 1485 } 1486 1487 /* Default all weights to 1. */ 1488 src[i].weight = 1.0; 1489 } 1490 1491 /* parse optional extra arguments */ 1492 if (j < c->argc) { 1493 int remaining = c->argc - j; 1494 1495 while (remaining) { 1496 if (remaining >= (setnum + 1) && !strcasecmp(c->argv[j]->ptr,"weights")) { 1497 j++; remaining--; 1498 for (i = 0; i < setnum; i++, j++, remaining--) { 1499 if (getDoubleFromObjectOrReply(c,c->argv[j],&src[i].weight, 1500 "weight value is not a float") != REDIS_OK) 1501 { 1502 zfree(src); 1503 return; 1504 } 1505 } 1506 } else if (remaining >= 2 && !strcasecmp(c->argv[j]->ptr,"aggregate")) { 1507 j++; remaining--; 1508 if (!strcasecmp(c->argv[j]->ptr,"sum")) { 1509 aggregate = REDIS_AGGR_SUM; 1510 } else if (!strcasecmp(c->argv[j]->ptr,"min")) { 1511 aggregate = REDIS_AGGR_MIN; 1512 } else if (!strcasecmp(c->argv[j]->ptr,"max")) { 1513 aggregate = REDIS_AGGR_MAX; 1514 } else { 1515 zfree(src); 1516 addReply(c,shared.syntaxerr); 1517 return; 1518 } 1519 j++; remaining--; 1520 } else { 1521 zfree(src); 1522 addReply(c,shared.syntaxerr); 1523 return; 1524 } 1525 } 1526 } 1527 1528 for (i = 0; i < setnum; i++) 1529 zuiInitIterator(&src[i]); 1530 1531 /* sort sets from the smallest to largest, this will improve our 1532 * algorithm's performance */ 1533 qsort(src,setnum,sizeof(zsetopsrc),zuiCompareByCardinality); 1534 1535 dstobj = createZsetObject(); 1536 dstzset = dstobj->ptr; 1537 memset(&zval, 0, sizeof(zval)); 1538 1539 if (op == REDIS_OP_INTER) { 1540 /* Skip everything if the smallest input is empty. */ 1541 if (zuiLength(&src[0]) > 0) { 1542 /* Precondition: as src[0] is non-empty and the inputs are ordered 1543 * by size, all src[i > 0] are non-empty too. */ 1544 while (zuiNext(&src[0],&zval)) { 1545 double score, value; 1546 1547 score = src[0].weight * zval.score; 1548 if (isnan(score)) score = 0; 1549 1550 for (j = 1; j < setnum; j++) { 1551 /* It is not safe to access the zset we are 1552 * iterating, so explicitly check for equal object. */ 1553 if (src[j].subject == src[0].subject) { 1554 value = zval.score*src[j].weight; 1555 zunionInterAggregate(&score,value,aggregate); 1556 } else if (zuiFind(&src[j],&zval,&value)) { 1557 value *= src[j].weight; 1558 zunionInterAggregate(&score,value,aggregate); 1559 } else { 1560 break; 1561 } 1562 } 1563 1564 /* Only continue when present in every input. */ 1565 if (j == setnum) { 1566 tmp = zuiObjectFromValue(&zval); 1567 znode = zslInsert(dstzset->zsl,score,tmp); 1568 incrRefCount(tmp); /* added to skiplist */ 1569 dictAdd(dstzset->dict,tmp,&znode->score); 1570 incrRefCount(tmp); /* added to dictionary */ 1571 1572 if (tmp->encoding == REDIS_ENCODING_RAW) 1573 if (sdslen(tmp->ptr) > maxelelen) 1574 maxelelen = sdslen(tmp->ptr); 1575 } 1576 } 1577 } 1578 } else if (op == REDIS_OP_UNION) { 1579 for (i = 0; i < setnum; i++) { 1580 if (zuiLength(&src[i]) == 0) 1581 continue; 1582 1583 while (zuiNext(&src[i],&zval)) { 1584 double score, value; 1585 1586 /* Skip key when already processed */ 1587 if (dictFind(dstzset->dict,zuiObjectFromValue(&zval)) != NULL) 1588 continue; 1589 1590 /* Initialize score */ 1591 score = src[i].weight * zval.score; 1592 if (isnan(score)) score = 0; 1593 1594 /* Because the inputs are sorted by size, it's only possible 1595 * for sets at larger indices to hold this element. */ 1596 for (j = (i+1); j < setnum; j++) { 1597 /* It is not safe to access the zset we are 1598 * iterating, so explicitly check for equal object. */ 1599 if(src[j].subject == src[i].subject) { 1600 value = zval.score*src[j].weight; 1601 zunionInterAggregate(&score,value,aggregate); 1602 } else if (zuiFind(&src[j],&zval,&value)) { 1603 value *= src[j].weight; 1604 zunionInterAggregate(&score,value,aggregate); 1605 } 1606 } 1607 1608 tmp = zuiObjectFromValue(&zval); 1609 znode = zslInsert(dstzset->zsl,score,tmp); 1610 incrRefCount(zval.ele); /* added to skiplist */ 1611 dictAdd(dstzset->dict,tmp,&znode->score); 1612 incrRefCount(zval.ele); /* added to dictionary */ 1613 1614 if (tmp->encoding == REDIS_ENCODING_RAW) 1615 if (sdslen(tmp->ptr) > maxelelen) 1616 maxelelen = sdslen(tmp->ptr); 1617 } 1618 } 1619 } else { 1620 redisPanic("Unknown operator"); 1621 } 1622 1623 for (i = 0; i < setnum; i++) 1624 zuiClearIterator(&src[i]); 1625 1626 if (dbDelete(c->db,dstkey)) { 1627 signalModifiedKey(c->db,dstkey); 1628 touched = 1; 1629 server.dirty++; 1630 } 1631 if (dstzset->zsl->length) { 1632 /* Convert to ziplist when in limits. */ 1633 if (dstzset->zsl->length <= server.zset_max_ziplist_entries && 1634 maxelelen <= server.zset_max_ziplist_value) 1635 zsetConvert(dstobj,REDIS_ENCODING_ZIPLIST); 1636 1637 dbAdd(c->db,dstkey,dstobj); 1638 addReplyLongLong(c,zsetLength(dstobj)); 1639 if (!touched) signalModifiedKey(c->db,dstkey); 1640 server.dirty++; 1641 } else { 1642 decrRefCount(dstobj); 1643 addReply(c,shared.czero); 1644 } 1645 zfree(src); 1646 } 1647 1648 void zunionstoreCommand(redisClient *c) { 1649 zunionInterGenericCommand(c,c->argv[1], REDIS_OP_UNION); 1650 } 1651 1652 void zinterstoreCommand(redisClient *c) { 1653 zunionInterGenericCommand(c,c->argv[1], REDIS_OP_INTER); 1654 } 1655 1656 void zrangeGenericCommand(redisClient *c, int reverse) { 1657 robj *key = c->argv[1]; 1658 robj *zobj; 1659 int withscores = 0; 1660 long start; 1661 long end; 1662 int llen; 1663 int rangelen; 1664 1665 if ((getLongFromObjectOrReply(c, c->argv[2], &start, NULL) != REDIS_OK) || 1666 (getLongFromObjectOrReply(c, c->argv[3], &end, NULL) != REDIS_OK)) return; 1667 1668 if (c->argc == 5 && !strcasecmp(c->argv[4]->ptr,"withscores")) { 1669 withscores = 1; 1670 } else if (c->argc >= 5) { 1671 addReply(c,shared.syntaxerr); 1672 return; 1673 } 1674 1675 if ((zobj = lookupKeyReadOrReply(c,key,shared.emptymultibulk)) == NULL 1676 || checkType(c,zobj,REDIS_ZSET)) return; 1677 1678 /* Sanitize indexes. */ 1679 llen = zsetLength(zobj); 1680 if (start < 0) start = llen+start; 1681 if (end < 0) end = llen+end; 1682 if (start < 0) start = 0; 1683 1684 /* Invariant: start >= 0, so this test will be true when end < 0. 1685 * The range is empty when start > end or start >= length. */ 1686 if (start > end || start >= llen) { 1687 addReply(c,shared.emptymultibulk); 1688 return; 1689 } 1690 if (end >= llen) end = llen-1; 1691 rangelen = (end-start)+1; 1692 1693 /* Return the result in form of a multi-bulk reply */ 1694 addReplyMultiBulkLen(c, withscores ? (rangelen*2) : rangelen); 1695 1696 if (zobj->encoding == REDIS_ENCODING_ZIPLIST) { 1697 unsigned char *zl = zobj->ptr; 1698 unsigned char *eptr, *sptr; 1699 unsigned char *vstr; 1700 unsigned int vlen; 1701 long long vlong; 1702 1703 if (reverse) 1704 eptr = ziplistIndex(zl,-2-(2*start)); 1705 else 1706 eptr = ziplistIndex(zl,2*start); 1707 1708 redisAssertWithInfo(c,zobj,eptr != NULL); 1709 sptr = ziplistNext(zl,eptr); 1710 1711 while (rangelen--) { 1712 redisAssertWithInfo(c,zobj,eptr != NULL && sptr != NULL); 1713 redisAssertWithInfo(c,zobj,ziplistGet(eptr,&vstr,&vlen,&vlong)); 1714 if (vstr == NULL) 1715 addReplyBulkLongLong(c,vlong); 1716 else 1717 addReplyBulkCBuffer(c,vstr,vlen); 1718 1719 if (withscores) 1720 addReplyDouble(c,zzlGetScore(sptr)); 1721 1722 if (reverse) 1723 zzlPrev(zl,&eptr,&sptr); 1724 else 1725 zzlNext(zl,&eptr,&sptr); 1726 } 1727 1728 } else if (zobj->encoding == REDIS_ENCODING_SKIPLIST) { 1729 zset *zs = zobj->ptr; 1730 zskiplist *zsl = zs->zsl; 1731 zskiplistNode *ln; 1732 robj *ele; 1733 1734 /* Check if starting point is trivial, before doing log(N) lookup. */ 1735 if (reverse) { 1736 ln = zsl->tail; 1737 if (start > 0) 1738 ln = zslGetElementByRank(zsl,llen-start); 1739 } else { 1740 ln = zsl->header->level[0].forward; 1741 if (start > 0) 1742 ln = zslGetElementByRank(zsl,start+1); 1743 } 1744 1745 while(rangelen--) { 1746 redisAssertWithInfo(c,zobj,ln != NULL); 1747 ele = ln->obj; 1748 addReplyBulk(c,ele); 1749 if (withscores) 1750 addReplyDouble(c,ln->score); 1751 ln = reverse ? ln->backward : ln->level[0].forward; 1752 } 1753 } else { 1754 redisPanic("Unknown sorted set encoding"); 1755 } 1756 } 1757 1758 void zrangeCommand(redisClient *c) { 1759 zrangeGenericCommand(c,0); 1760 } 1761 1762 void zrevrangeCommand(redisClient *c) { 1763 zrangeGenericCommand(c,1); 1764 } 1765 1766 /* This command implements ZRANGEBYSCORE, ZREVRANGEBYSCORE. */ 1767 void genericZrangebyscoreCommand(redisClient *c, int reverse) { 1768 zrangespec range; 1769 robj *key = c->argv[1]; 1770 robj *zobj; 1771 long offset = 0, limit = -1; 1772 int withscores = 0; 1773 unsigned long rangelen = 0; 1774 void *replylen = NULL; 1775 int minidx, maxidx; 1776 1777 /* Parse the range arguments. */ 1778 if (reverse) { 1779 /* Range is given as [max,min] */ 1780 maxidx = 2; minidx = 3; 1781 } else { 1782 /* Range is given as [min,max] */ 1783 minidx = 2; maxidx = 3; 1784 } 1785 1786 if (zslParseRange(c->argv[minidx],c->argv[maxidx],&range) != REDIS_OK) { 1787 addReplyError(c,"min or max is not a float"); 1788 return; 1789 } 1790 1791 /* Parse optional extra arguments. Note that ZCOUNT will exactly have 1792 * 4 arguments, so we'll never enter the following code path. */ 1793 if (c->argc > 4) { 1794 int remaining = c->argc - 4; 1795 int pos = 4; 1796 1797 while (remaining) { 1798 if (remaining >= 1 && !strcasecmp(c->argv[pos]->ptr,"withscores")) { 1799 pos++; remaining--; 1800 withscores = 1; 1801 } else if (remaining >= 3 && !strcasecmp(c->argv[pos]->ptr,"limit")) { 1802 if ((getLongFromObjectOrReply(c, c->argv[pos+1], &offset, NULL) != REDIS_OK) || 1803 (getLongFromObjectOrReply(c, c->argv[pos+2], &limit, NULL) != REDIS_OK)) return; 1804 pos += 3; remaining -= 3; 1805 } else { 1806 addReply(c,shared.syntaxerr); 1807 return; 1808 } 1809 } 1810 } 1811 1812 /* Ok, lookup the key and get the range */ 1813 if ((zobj = lookupKeyReadOrReply(c,key,shared.emptymultibulk)) == NULL || 1814 checkType(c,zobj,REDIS_ZSET)) return; 1815 1816 if (zobj->encoding == REDIS_ENCODING_ZIPLIST) { 1817 unsigned char *zl = zobj->ptr; 1818 unsigned char *eptr, *sptr; 1819 unsigned char *vstr; 1820 unsigned int vlen; 1821 long long vlong; 1822 double score; 1823 1824 /* If reversed, get the last node in range as starting point. */ 1825 if (reverse) { 1826 eptr = zzlLastInRange(zl,range); 1827 } else { 1828 eptr = zzlFirstInRange(zl,range); 1829 } 1830 1831 /* No "first" element in the specified interval. */ 1832 if (eptr == NULL) { 1833 addReply(c, shared.emptymultibulk); 1834 return; 1835 } 1836 1837 /* Get score pointer for the first element. */ 1838 redisAssertWithInfo(c,zobj,eptr != NULL); 1839 sptr = ziplistNext(zl,eptr); 1840 1841 /* We don't know in advance how many matching elements there are in the 1842 * list, so we push this object that will represent the multi-bulk 1843 * length in the output buffer, and will "fix" it later */ 1844 replylen = addDeferredMultiBulkLength(c); 1845 1846 /* If there is an offset, just traverse the number of elements without 1847 * checking the score because that is done in the next loop. */ 1848 while (eptr && offset--) { 1849 if (reverse) { 1850 zzlPrev(zl,&eptr,&sptr); 1851 } else { 1852 zzlNext(zl,&eptr,&sptr); 1853 } 1854 } 1855 1856 while (eptr && limit--) { 1857 score = zzlGetScore(sptr); 1858 1859 /* Abort when the node is no longer in range. */ 1860 if (reverse) { 1861 if (!zslValueGteMin(score,&range)) break; 1862 } else { 1863 if (!zslValueLteMax(score,&range)) break; 1864 } 1865 1866 /* We know the element exists, so ziplistGet should always succeed */ 1867 redisAssertWithInfo(c,zobj,ziplistGet(eptr,&vstr,&vlen,&vlong)); 1868 1869 rangelen++; 1870 if (vstr == NULL) { 1871 addReplyBulkLongLong(c,vlong); 1872 } else { 1873 addReplyBulkCBuffer(c,vstr,vlen); 1874 } 1875 1876 if (withscores) { 1877 addReplyDouble(c,score); 1878 } 1879 1880 /* Move to next node */ 1881 if (reverse) { 1882 zzlPrev(zl,&eptr,&sptr); 1883 } else { 1884 zzlNext(zl,&eptr,&sptr); 1885 } 1886 } 1887 } else if (zobj->encoding == REDIS_ENCODING_SKIPLIST) { 1888 zset *zs = zobj->ptr; 1889 zskiplist *zsl = zs->zsl; 1890 zskiplistNode *ln; 1891 1892 /* If reversed, get the last node in range as starting point. */ 1893 if (reverse) { 1894 ln = zslLastInRange(zsl,range); 1895 } else { 1896 ln = zslFirstInRange(zsl,range); 1897 } 1898 1899 /* No "first" element in the specified interval. */ 1900 if (ln == NULL) { 1901 addReply(c, shared.emptymultibulk); 1902 return; 1903 } 1904 1905 /* We don't know in advance how many matching elements there are in the 1906 * list, so we push this object that will represent the multi-bulk 1907 * length in the output buffer, and will "fix" it later */ 1908 replylen = addDeferredMultiBulkLength(c); 1909 1910 /* If there is an offset, just traverse the number of elements without 1911 * checking the score because that is done in the next loop. */ 1912 while (ln && offset--) { 1913 if (reverse) { 1914 ln = ln->backward; 1915 } else { 1916 ln = ln->level[0].forward; 1917 } 1918 } 1919 1920 while (ln && limit--) { 1921 /* Abort when the node is no longer in range. */ 1922 if (reverse) { 1923 if (!zslValueGteMin(ln->score,&range)) break; 1924 } else { 1925 if (!zslValueLteMax(ln->score,&range)) break; 1926 } 1927 1928 rangelen++; 1929 addReplyBulk(c,ln->obj); 1930 1931 if (withscores) { 1932 addReplyDouble(c,ln->score); 1933 } 1934 1935 /* Move to next node */ 1936 if (reverse) { 1937 ln = ln->backward; 1938 } else { 1939 ln = ln->level[0].forward; 1940 } 1941 } 1942 } else { 1943 redisPanic("Unknown sorted set encoding"); 1944 } 1945 1946 if (withscores) { 1947 rangelen *= 2; 1948 } 1949 1950 setDeferredMultiBulkLength(c, replylen, rangelen); 1951 } 1952 1953 void zrangebyscoreCommand(redisClient *c) { 1954 genericZrangebyscoreCommand(c,0); 1955 } 1956 1957 void zrevrangebyscoreCommand(redisClient *c) { 1958 genericZrangebyscoreCommand(c,1); 1959 } 1960 1961 void zcountCommand(redisClient *c) { 1962 robj *key = c->argv[1]; 1963 robj *zobj; 1964 zrangespec range; 1965 int count = 0; 1966 1967 /* Parse the range arguments */ 1968 if (zslParseRange(c->argv[2],c->argv[3],&range) != REDIS_OK) { 1969 addReplyError(c,"min or max is not a float"); 1970 return; 1971 } 1972 1973 /* Lookup the sorted set */ 1974 if ((zobj = lookupKeyReadOrReply(c, key, shared.czero)) == NULL || 1975 checkType(c, zobj, REDIS_ZSET)) return; 1976 1977 if (zobj->encoding == REDIS_ENCODING_ZIPLIST) { 1978 unsigned char *zl = zobj->ptr; 1979 unsigned char *eptr, *sptr; 1980 double score; 1981 1982 /* Use the first element in range as the starting point */ 1983 eptr = zzlFirstInRange(zl,range); 1984 1985 /* No "first" element */ 1986 if (eptr == NULL) { 1987 addReply(c, shared.czero); 1988 return; 1989 } 1990 1991 /* First element is in range */ 1992 sptr = ziplistNext(zl,eptr); 1993 score = zzlGetScore(sptr); 1994 redisAssertWithInfo(c,zobj,zslValueLteMax(score,&range)); 1995 1996 /* Iterate over elements in range */ 1997 while (eptr) { 1998 score = zzlGetScore(sptr); 1999 2000 /* Abort when the node is no longer in range. */ 2001 if (!zslValueLteMax(score,&range)) { 2002 break; 2003 } else { 2004 count++; 2005 zzlNext(zl,&eptr,&sptr); 2006 } 2007 } 2008 } else if (zobj->encoding == REDIS_ENCODING_SKIPLIST) { 2009 zset *zs = zobj->ptr; 2010 zskiplist *zsl = zs->zsl; 2011 zskiplistNode *zn; 2012 unsigned long rank; 2013 2014 /* Find first element in range */ 2015 zn = zslFirstInRange(zsl, range); 2016 2017 /* Use rank of first element, if any, to determine preliminary count */ 2018 if (zn != NULL) { 2019 rank = zslGetRank(zsl, zn->score, zn->obj); 2020 count = (zsl->length - (rank - 1)); 2021 2022 /* Find last element in range */ 2023 zn = zslLastInRange(zsl, range); 2024 2025 /* Use rank of last element, if any, to determine the actual count */ 2026 if (zn != NULL) { 2027 rank = zslGetRank(zsl, zn->score, zn->obj); 2028 count -= (zsl->length - rank); 2029 } 2030 } 2031 } else { 2032 redisPanic("Unknown sorted set encoding"); 2033 } 2034 2035 addReplyLongLong(c, count); 2036 } 2037 2038 void zcardCommand(redisClient *c) { 2039 robj *key = c->argv[1]; 2040 robj *zobj; 2041 2042 if ((zobj = lookupKeyReadOrReply(c,key,shared.czero)) == NULL || 2043 checkType(c,zobj,REDIS_ZSET)) return; 2044 2045 addReplyLongLong(c,zsetLength(zobj)); 2046 } 2047 2048 void zscoreCommand(redisClient *c) { 2049 robj *key = c->argv[1]; 2050 robj *zobj; 2051 double score; 2052 2053 if ((zobj = lookupKeyReadOrReply(c,key,shared.nullbulk)) == NULL || 2054 checkType(c,zobj,REDIS_ZSET)) return; 2055 2056 if (zobj->encoding == REDIS_ENCODING_ZIPLIST) { 2057 if (zzlFind(zobj->ptr,c->argv[2],&score) != NULL) 2058 addReplyDouble(c,score); 2059 else 2060 addReply(c,shared.nullbulk); 2061 } else if (zobj->encoding == REDIS_ENCODING_SKIPLIST) { 2062 zset *zs = zobj->ptr; 2063 dictEntry *de; 2064 2065 c->argv[2] = tryObjectEncoding(c->argv[2]); 2066 de = dictFind(zs->dict,c->argv[2]); 2067 if (de != NULL) { 2068 score = *(double*)dictGetVal(de); 2069 addReplyDouble(c,score); 2070 } else { 2071 addReply(c,shared.nullbulk); 2072 } 2073 } else { 2074 redisPanic("Unknown sorted set encoding"); 2075 } 2076 } 2077 2078 void zrankGenericCommand(redisClient *c, int reverse) { 2079 robj *key = c->argv[1]; 2080 robj *ele = c->argv[2]; 2081 robj *zobj; 2082 unsigned long llen; 2083 unsigned long rank; 2084 2085 if ((zobj = lookupKeyReadOrReply(c,key,shared.nullbulk)) == NULL || 2086 checkType(c,zobj,REDIS_ZSET)) return; 2087 llen = zsetLength(zobj); 2088 2089 redisAssertWithInfo(c,ele,ele->encoding == REDIS_ENCODING_RAW); 2090 if (zobj->encoding == REDIS_ENCODING_ZIPLIST) { 2091 unsigned char *zl = zobj->ptr; 2092 unsigned char *eptr, *sptr; 2093 2094 eptr = ziplistIndex(zl,0); 2095 redisAssertWithInfo(c,zobj,eptr != NULL); 2096 sptr = ziplistNext(zl,eptr); 2097 redisAssertWithInfo(c,zobj,sptr != NULL); 2098 2099 rank = 1; 2100 while(eptr != NULL) { 2101 if (ziplistCompare(eptr,ele->ptr,sdslen(ele->ptr))) 2102 break; 2103 rank++; 2104 zzlNext(zl,&eptr,&sptr); 2105 } 2106 2107 if (eptr != NULL) { 2108 if (reverse) 2109 addReplyLongLong(c,llen-rank); 2110 else 2111 addReplyLongLong(c,rank-1); 2112 } else { 2113 addReply(c,shared.nullbulk); 2114 } 2115 } else if (zobj->encoding == REDIS_ENCODING_SKIPLIST) { 2116 zset *zs = zobj->ptr; 2117 zskiplist *zsl = zs->zsl; 2118 dictEntry *de; 2119 double score; 2120 2121 ele = c->argv[2] = tryObjectEncoding(c->argv[2]); 2122 de = dictFind(zs->dict,ele); 2123 if (de != NULL) { 2124 score = *(double*)dictGetVal(de); 2125 rank = zslGetRank(zsl,score,ele); 2126 redisAssertWithInfo(c,ele,rank); /* Existing elements always have a rank. */ 2127 if (reverse) 2128 addReplyLongLong(c,llen-rank); 2129 else 2130 addReplyLongLong(c,rank-1); 2131 } else { 2132 addReply(c,shared.nullbulk); 2133 } 2134 } else { 2135 redisPanic("Unknown sorted set encoding"); 2136 } 2137 } 2138 2139 void zrankCommand(redisClient *c) { 2140 zrankGenericCommand(c, 0); 2141 } 2142 2143 void zrevrankCommand(redisClient *c) { 2144 zrankGenericCommand(c, 1); 2145 } 2146