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